python之multiprocessing.Queue 中的大对象死锁

当您向 multiprocessing.Queue 提供足够大的对象时，程序似乎会在奇怪的地方挂起。考虑这个最小的例子:

import multiprocessing 
 
def dump_dict(queue, size): 
  queue.put({x: x for x in range(size)}) 
  print("Dump finished") 
 
if __name__ == '__main__': 
  SIZE = int(1e5) 
  queue = multiprocessing.Queue() 
  process = multiprocessing.Process(target=dump_dict, args=(queue, SIZE)) 
  print("Starting...") 
  process.start() 
  print("Joining...") 
  process.join() 
  print("Done") 
  print(len(queue.get())) 

如果 SIZE 参数足够小(<= 1e4 至少在我的例子中)，整个程序运行顺利没有问题，但是一旦 SIZE 是足够大，程序卡在奇怪的地方。现在，在搜索解释时，即 python multiprocessing - process hangs on join for large queue ，我一直看到“你需要从队列中消费”的笼统答案。但奇怪的是，程序实际上打印了 Dump finished，即在将对象放入 queue 后到达了代码行。此外，使用 Queue.put_nowait 而不是 Queue.put 并没有什么不同。

最后，如果您使用 Process.join(1) 而不是 Process.join()，则整个过程结束时队列中有完整的字典(即 print(len(..)) 行将打印 10000)。

有人可以让我对此有更深入的了解吗？

请您参考如下方法:

您需要在 process.join() 之前在父级中执行 queue.get() 以防止死锁。队列已经用它的第一个 queue.put() 生成了一个供给线程，并且您的工作进程中的 MainThread 在退出之前加入了这个供给线程。因此，在结果完全刷新到 (OS-pipe-) 缓冲区之前，工作进程不会退出，但是您的结果太大而无法放入缓冲区，并且您的 parent 在工作人员退出之前不会从队列中读取，导致死锁。

您会看到 print("Dump finished") 的输出，因为实际的发送是从馈线线程发生的，queue.put() 本身只是附加到一个collections.deque 在工作进程中作为中间步骤。