python - 如何获得 "work"剩余的数量由 Python 多处理池完成？

Question

到目前为止，只要我需要使用 multiprocessing我通过手动创建一个“进程池”并与所有子进程共享一个工作队列来做到这一点。

例如:

from multiprocessing import Process, Queue


class MyClass:

    def __init__(self, num_processes):
        self._log         = logging.getLogger()
        self.process_list = []
        self.work_queue   = Queue()
        for i in range(num_processes):
            p_name = 'CPU_%02d' % (i+1)
            self._log.info('Initializing process %s', p_name)
            p = Process(target = do_stuff,
                        args   = (self.work_queue, 'arg1'),
                        name   = p_name)

这样我就可以将内容添加到队列中，这些内容将被子进程使用。然后我可以通过检查 Queue.qsize() 来监控处理的进度:

    while True:
        qsize = self.work_queue.qsize()
        if qsize == 0:
            self._log.info('Processing finished')
            break
        else:
            self._log.info('%d simulations still need to be calculated', qsize)

现在我认为 multiprocessing.Pool可以大大简化这段代码。

我找不到的是如何监控仍有待完成的“工作”量。

举个例子:

from multiprocessing import Pool


class MyClass:

    def __init__(self, num_processes):
        self.process_pool = Pool(num_processes)
        # ...
        result_list = []
        for i in range(1000):            
            result = self.process_pool.apply_async(do_stuff, ('arg1',))
            result_list.append(result)
        # ---> here: how do I monitor the Pool's processing progress?
        # ...?

有什么想法吗？

Answer 1

使用Manager 队列。这是一个在工作进程之间共享的队列。如果您使用普通队列，它会被每个工作人员 pickle 和 unpickled 并因此被复制，这样队列就不能被每个工作人员更新。

然后，您可以让您的工作人员向队列中添加内容，并在工作人员工作时监视队列的状态。您需要使用 map_async 来执行此操作，因为这可以让您看到整个结果何时准备就绪，从而可以打破监控循环。

例子:

import time
from multiprocessing import Pool, Manager


def play_function(args):
    """Mock function, that takes a single argument consisting
    of (input, queue). Alternately, you could use another function
    as a wrapper.
    """
    i, q = args
    time.sleep(0.1)  # mock work
    q.put(i)
    return i

p = Pool()
m = Manager()
q = m.Queue()

inputs = range(20)
args = [(i, q) for i in inputs]
result = p.map_async(play_function, args)

# monitor loop
while True:
    if result.ready():
        break
    else:
        size = q.qsize()
        print(size)
        time.sleep(0.1)

outputs = result.get()