我应该使用什么数据结构来存储并行运行的进程返回的结果？

Question

我正在使用multiprocessing 模块并行运行一个函数。我并行运行的进程本身具有并行运行的子进程，因此我不能使用 pool 类，除了最低级别的子进程（即那些不再创建子进程的子进程），因为守护进程可能不会创造孩子。

因此，我使用Process，并手动管理运行和加入进程。

今天，我花了很长时间尝试调试我的代码，因为它似乎挂了，但我不明白在哪里。经过一些调试，我最终发现当我的一个函数试图将数据放入其中时，我用来存储结果的 multiprorcessing.Queue 对象无限期地阻塞。我还不确定为什么它会无限期地阻塞，但我已经确认这是问题，因为删除 put 命令允许继续执行（虽然，我没有得到任何数据）。

multiprorcessing.Queue 是用于存储并行运行的函数返回的信息的正确对象吗？

Answer 1

多进程系统难以调试。我建议彻底调试所有底层函数，然后添加一点 multiproc 来品尝。

因为 multiproc 可能会令人困惑，我建议尽早并经常记录。如果你记录太多，很容易去掉绒毛。但是，如果您在日志中看不到奇怪的极端情况，那么事情可能会......困难:)

worker 用它的名字记录，而 parent 记录为“MainProcess”。正常的东西被记录为 INFO，可怕的问题被记录为 ERROR。

为了模拟“不能将东西放入我的队列”错误，我创建了输出 Queue 以仅容纳一项。还有其他代码会监视它，并专门记录它。 （将 if 0 更改为 if 1 以获取完整的代码以运行。）

玩得开心！

import logging, multiprocessing, Queue

def myproc(arg):
    return arg*2

def worker(inqueue, outqueue):
    mylog = multiprocessing.get_logger()
    mylog.info('start')
    for job in iter(inqueue.get, 'STOP'):
        mylog.info('got %s', job)
        try:
            outqueue.put( myproc(job), timeout=1 )
        except Queue.Full:
            mylog.error('queue full!')

    mylog.info('done')


logger = multiprocessing.log_to_stderr(
    level=logging.INFO,
)
logger.info('setup')

inqueue, outqueue = multiprocessing.Queue(), multiprocessing.Queue()
if 1:                           # debug 'queue full!' issues
    outqueue = multiprocessing.Queue(maxsize=1)
# prefill with 3 jobs
for num in range(3):
    inqueue.put(num)
# signal end of jobs
inqueue.put('STOP')

worker_p = multiprocessing.Process(
    target=worker, args=(inqueue, outqueue),
    name='worker',
)
worker_p.start()

worker_p.join()

logger.info('done')

示例运行：

[INFO/MainProcess] setup
[INFO/worker] child process calling self.run()
[INFO/worker] start
[INFO/worker] got 0
[INFO/worker] got 1
[ERROR/worker] queue full!
[INFO/worker] got 2
[ERROR/worker] queue full!
[INFO/worker] done
[INFO/worker] process shutting down
[INFO/worker] process exiting with exitcode 0
[INFO/MainProcess] done
[INFO/MainProcess] process shutting down