# 多進程最佳實踐 # 多進程最佳實踐 `torch.multiprocessing`是Python`multiprocessing`的替代品。它支持完全相同的操作，但擴展了它以便通過`multiprocessing.Queue`發送的所有張量將其數據移動到共享內存中，并且只會向其他進程發送一個句柄。 > **Note** > > 當`Variable`發送到另一個進程時，`Variable.data`和`Variable.grad.data`都將被共享。 這允許實現各種訓練方法，如Hogwild，A3C或需要異步操作的任何其他方法。 ## 共享CUDA張量 僅在Python 3中使用`spawn`或`forkserver`啟動方法才支持在進程之間共享CUDA張量。Python 2中的`multiprocessing`只能使用`fork`創建子進程，并且不被CUDA運行時所支持。 > **Warning** > > CUDA API要求導出到其他進程的分配，只要它們被使用就要一直保持有效。您應該小心，確保您共享的CUDA張量只要有必要就不要超出范圍。這不是共享模型參數的問題，但傳遞其他類型的數據應該小心。注意，此限制不適用于共享CPU內存。 參考：[使用 nn.DataParallel 替代 multiprocessing](cuda.html) ## 最佳實踐和提示 ### 避免和抵制死鎖 當一個新進程被產生時，有很多事情可能會出錯，最常見的死鎖原因是后臺線程。如果有任何線程持有鎖或導入模塊，并且`fork`被調用，則子進程很可能處于損壞的狀態，并以不同的方式死鎖或失敗。注意，即使您沒有，Python內置的庫也可能會這樣做 —— 不需要看得比`multiprocessing`更遠。`multiprocessing.Queue`實際上是一個非常復雜的類，它產生用于序列化，發送和接收對象的多個線程，它們也可能引起上述問題。如果您發現自己處于這種情況，請嘗試使用`multiprocessing.queues.SimpleQueue`，這不會使用任何其他線程。 我們正在竭盡全力把它設計得更簡單，并確保這些死鎖不會發生，但有些事情無法控制。如果有任何問題您無法一時無法解決，請嘗試在論壇上提出，我們將看看是否可以解決問題。 ### 重用經過隊列的緩沖區 記住每次將`Tensor`放入`multiprocessing.Queue`時，必須將其移動到共享內存中。如果它已經被共享，它是一個無效的操作，否則會產生一個額外的內存副本，這會減緩整個進程。即使你有一個進程池來發送數據到一個進程，使它返回緩沖區 —— 這幾乎是免費的，并且允許你在發送下一個batch時避免產生副本。 ### 異步多進程訓練（例如Hogwild） 使用`torch.multiprocessing`，可以異步地訓練模型，參數可以一直共享，也可以定期同步。在第一種情況下，我們建議發送整個模型對象，而在后者中，我們建議只發送`state_dict()`。 我們建議使用`multiprocessing.Queue`來在進程之間傳遞各種PyTorch對象。例如， 當使用fork啟動方法時，可能會繼承共享內存中的張量和存儲器，但這是非常容易出錯的，應謹慎使用，而且只能由高級用戶使用。隊列雖然有時是一個較不優雅的解決方案，但基本上能在所有情況下正常工作。 > **Warning**你應該注意有關全局語句，它們沒有被`if __name__ == '__main__'`保護。如果使用與`fork`不同的啟動方法，則它們將在所有子進程中執行。 #### Hogwild 在[examples repository](https://github.com/pytorch/examples/tree/master/mnist_hogwild)中可以找到具體的Hogwild實現，可以展示代碼的整體結構。下面也有一個小例子： ``` import torch.multiprocessing as mp from model import MyModel def train(model): # Construct data_loader, optimizer, etc. for data, labels in data_loader: optimizer.zero_grad() loss_fn(model(data), labels).backward() optimizer.step() # This will update the shared parameters if __name__ == '__main__': num_processes = 4 model = MyModel() # NOTE: this is required for the ``fork`` method to work model.share_memory() processes = [] for rank in range(num_processes): p = mp.Process(target=train, args=(model,)) p.start() processes.append(p) for p in processes: p.join() ```