将复杂的字典放入返回队列时，多处理过程未加入 [英] Multiprocessing process does not join when putting complex dictionary in return queue

问题描述

给出一个具有读取队列和写入队列的相当标准的读取/写入多线程进程:

Given a pretty standard read/write multithreaded process with a read Queue and a write Queue:

打印8次worker done，但是从不传递join()语句.但是，如果我用`queue_out.put(1)替换queue_out.put(r)，它将起作用.

8 times worker done is printed, but the join() statement is never passed. But if I replace queue_out.put(r) by `queue_out.put(1) it works.

这让我的大脑融化了，可能真的很愚蠢.我是否应该复制字典并将其放入return Queue中?我在某个地方犯了一个愚蠢的错误吗?

This is melting my brain, probably something really stupid. Should I make a copy of my dictionary and put that in the return Queue? Did I make a stupid mistake somewhere?

处理功能

def reader(queue_in, queue_out, funktion):
    # Read from the queue
    while True:
        r = queue_in.get()
        if r == 'DONE':
            return
        funktion(r) # funktion adds additional keys to the dictionary
        queue_out.put(r) # <---- replacing r by 1 does let me join()
    print "worker done" # <----- this happens

填充输入队列

def writer(generator, queue):
    # Write to the queue
    for r in enumerate(generator):
        # r is a complex dictionary
        queue.put(r)    
    print "writer done"
    for _ in range(0, WORKERS):
        queue.put((-1, "DONE"))

其余

WORKERS = 8

# init Queues
queue_in = Queue()
queue_out = Queue()

# Start processes, with input and output quests
readers = []
for _ in range(0, WORKERS):
    p = Process(target=reader, args=(queue_in, queue_out, funktion))
    p.daemon = True
    p.start()
    readers.append(p)

writer(generator, queue_in)

for p in readers:
    p.join()

print "joined"  # <---- this never happens

queue_in.close()

while not queue_out.empty():
    print queue_out.get()
queue_out.close()

推荐答案

我认为我一直从两个来源将其拼凑在一起，因为我总是遇到相同的问题.我认为重要的是，这是在Windows中.

I think I have pieced this together from two sources as I always have the same problem. I think the important thing is that this is in Windows.

文档中的注释

由于Windows缺少os.fork()，因此它具有一些额外的限制:

Since Windows lacks os.fork() it has a few extra restrictions:

然后在此处中阅读答案 join()用于分叉处理.

Then read the answers here that join() is for forked processed.

我一直设法以与您类似的方式运行multiprocessing，而不使用join()并且没有看到任何错误-我很高兴有一个反例来解释为什么需要它.确实，删除它可以解决您的问题.

I have always managed to run multiprocessing in a similar fashion to you without using join() and not seen any errors - I'm more than happy for a counterexample to explain why it's needed. Indeed, removing it has corrected your issue.

本文深入探讨了这些差异与子进程在操作系统之间进行多处理.我确实认为join()的问题应该在文档中更加明确.

And this article goes into more depth about the differences with child processes in multiprocessing between operating systems. I do think that the issue with join(), specifically, should be more explicit in the documentation.

这篇关于将复杂的字典放入返回队列时，多处理过程未加入的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！