多進程套件 - torch.multiprocessing

torch.multiprocessing 是原生 multiprocessing 模組的封裝。

它註冊自訂的 reducer，這些 reducer 使用共享記憶體，在不同的進程中提供對相同資料的共享檢視。一旦張量/儲存體被移動到 shared_memory（請參閱 share_memory_()），就可以將其傳送到其他進程，而無需進行任何複製。

這個 API 與原始模組 100% 相容 - 只需要將 import multiprocessing 更改為 import torch.multiprocessing，就可以將所有透過佇列傳送或透過其他機制共享的張量，移動到共享記憶體中。

由於 API 的相似性，我們不會詳細說明此套件的大部分內容，建議您參考原始模組的優秀文檔。

警告

如果主程序突然退出（例如，由於收到信號），Python 的 multiprocessing 有時無法清除其子進程。這是一個已知的問題，因此如果您在終止直譯器後看到任何資源洩漏，則可能表示發生了這種情況。

策略管理

torch.multiprocessing.get_all_sharing_strategies()

傳回目前系統上支援的一組共享策略。

torch.multiprocessing.get_sharing_strategy()

傳回用於共享 CPU 張量的當前策略。

torch.multiprocessing.set_sharing_strategy(new_strategy)

設定用於共享 CPU 張量的策略。

參數

new_strategy (str) – 選定策略的名稱。應該是 get_all_sharing_strategies() 傳回的值之一。

共享 CUDA 張量

只有在使用 spawn 或 forkserver 啟動方法的 Python 3 中才支援在程序之間共享 CUDA 張量。

與 CPU 張量不同，只要接收程序保留張量的副本，就需要傳送程序保留原始張量。 引用計數是在底層實作的，但需要使用者遵循以下最佳實踐。

警告

如果消費者進程異常地因為致命信號而死亡，只要傳送進程正在運行，共享張量可能會永遠保存在記憶體中。

1. 盡快釋放消費者中的記憶體。

## Good
x = queue.get()
# do somethings with x
del x

## Bad
x = queue.get()
# do somethings with x
# do everything else (producer have to keep x in memory)

2. 保持生產者進程運行直到所有消費者退出。這將防止生產者進程釋放消費者仍在使用的記憶體的情況。

## producer
# send tensors, do something
event.wait()

## consumer
# receive tensors and use them
event.set()

3. 不要傳遞接收到的張量。

# not going to work
x = queue.get()
queue_2.put(x)

# you need to create a process-local copy
x = queue.get()
x_clone = x.clone()
queue_2.put(x_clone)

# putting and getting from the same queue in the same process will likely end up with segfault
queue.put(tensor)
x = queue.get()

共享策略

本節簡要概述了不同的共享策略如何運作。 請注意，它僅適用於 CPU 張量 - CUDA 張量將始終使用 CUDA API，因為這是它們可以共享的唯一方式。

文件描述符 - file_descriptor

注意

這是預設策略（macOS 和 OS X 除外，因為它們不受支援）。

此策略將使用文件描述符作為共享記憶體控制代碼。 每當儲存移動到共享記憶體時，從 shm_open 取得的文件描述符會與物件一起快取，並且當它要傳送到其他進程時，文件描述符將（例如，透過 UNIX socket）傳輸到它。 接收者也會快取文件描述符並 mmap 它，以取得對儲存資料的共享檢視。

請注意，如果共享的張量很多，此策略會長時間保持大量開啟的文件描述符。 如果您的系統對開啟的文件描述符數量有低限制，並且您無法提高它們，則應使用 file_system 策略。

文件系統 - file_system

此策略將使用給 shm_open 的文件名來識別共享記憶體區域。 這有一個好處是不需要實作快取從中取得的文件描述符，但同時也容易導致共享記憶體洩漏。 檔案無法在其建立後立即刪除，因為其他進程需要存取它才能開啟它們的檢視。 如果進程嚴重崩潰或被終止，並且沒有呼叫儲存解構函數，則檔案將保留在系統中。 這是非常嚴重的，因為它們會一直消耗記憶體，直到系統重新啟動或手動釋放它們。

為了應對共享記憶體檔案洩漏的問題，torch.multiprocessing 將產生一個名為 torch_shm_manager 的守護程序，它將自己與當前進程組隔離，並追蹤所有共享記憶體分配。 一旦所有連接到它的進程退出，它將等待一會兒以確保不會有新的連接，並將迭代該組分配的所有共享記憶體檔案。 如果它發現它們中的任何一個仍然存在，它們將被釋放。 我們已經測試了這種方法，並且證明它對於各種故障都很穩健。 儘管如此，如果您的系統具有足夠高的限制，並且 file_descriptor 是一種支援的策略，我們不建議切換到這種策略。

產生子進程

注意

適用於 Python >= 3.4。

這取決於 Python 的 multiprocessing 套件中的 spawn 啟動方法。

產生多個子進程來執行某些功能可以透過建立 Process 實例並呼叫 join 來等待它們完成。 這種方法在處理單個子進程時效果很好，但在處理多個進程時會出現潛在問題。

也就是說，依序加入進程意味著它們將依序終止。 如果它們沒有終止，並且第一個進程沒有終止，則進程終止將不會被注意到。 此外，沒有用於錯誤傳播的本機工具。

下面的 spawn 函數解決了這些問題，並負責錯誤傳播、亂序終止，並會在檢測到其中一個進程發生錯誤時主動終止進程。

torch.multiprocessing.spawn.spawn(fn, args=(), nprocs=1, join=True, daemon=False, start_method='spawn')

產生 nprocs 個執行 fn 函式，並帶有 args 參數的行程。

如果其中一個行程以非零的退出狀態退出，則其餘行程將被終止，並引發一個例外，指出終止的原因。如果子行程中捕獲到例外，則會將其轉發，並且其追蹤資訊會包含在父行程中引發的例外中。

參數

• fn (function) –

  函式被呼叫作為產生行程的進入點。此函式必須在模組的頂層定義，以便可以被封裝 (pickled) 和產生 (spawned)。這是 multiprocessing 強制的要求。

  此函式會以 fn(i, *args) 的形式呼叫，其中 i 是行程索引，而 args 是傳遞的參數元組。

• args (tuple) – 傳遞給 fn 的參數。

• nprocs (int) – 要產生的行程數量。

• join (bool) – 是否對所有行程執行阻塞式的 join。

• daemon (bool) – 產生行程的守護行程 (daemon) 標誌。如果設定為 True，將會建立守護行程。

• start_method (str) – (已棄用) 此方法將始終使用 spawn 作為啟動方法。 若要使用不同的啟動方法，請使用 start_processes()。

Returns

如果 join 為 True，則回傳 None。如果 join 為 False，則回傳 ProcessContext

class torch.multiprocessing.SpawnContext

當呼叫 spawn() 且 join=False 時，會回傳此物件。

join(timeout=None, grace_period=None)

加入產生上下文中的一個或多個行程。

嘗試加入此產生上下文中一個或多個行程。如果其中一個行程以非零的退出狀態退出，此函式會終止其餘行程（可選擇提供寬限期），並引發一個例外，指出第一個行程退出的原因。

如果所有行程都已成功加入，則回傳 True；如果還有更多行程需要加入，則回傳 False。

參數

• timeout (float) – 放棄等待之前等待的時間長度（以秒為單位）。

• grace_period (float) – 當任何行程失敗時，在終止其他行程之前，等待其他行程優雅關閉的時間長度（以秒為單位）。如果它們仍然沒有退出，則再次等待寬限期後再終止它們。