如何获得Python多处理池剩余的“工作”量？

Question

到目前为止，每当我需要使用^{}时，我都是通过手动创建一个“进程池”并与所有子进程共享一个工作队列来完成的。

例如：

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)

现在我认为^{}可以简化很多代码。

我不知道的是，我怎样才能监控还有多少“工作”要做。

举个例子：

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

我想出了下面的异步调用解决方案。

小玩具脚本的例子，但应该广泛应用我认为。

基本上，在无限循环中，在列表生成器中轮询结果对象的ready值，然后求和以计算剩余的已调度池任务数。

一旦没有剩余的break和join（）&close（）。

根据需要添加睡眠循环。

与上述解决方案相同的原理，但没有队列。如果还跟踪最初发送池的任务数，则可以计算完成百分比等。。。

import multiprocessing
import os
import time
from random import randrange


def worker():
    print os.getpid()

    #simulate work
    time.sleep(randrange(5))

if __name__ == '__main__':

    pool = multiprocessing.Pool(processes=8)
    result_objs = []

    print "Begin dispatching work"

    task_count = 10
    for x in range(task_count):
        result_objs.append(pool.apply_async(func=worker))

    print "Done dispatching work"

    while True:
        incomplete_count = sum(1 for x in result_objs if not x.ready())

        if incomplete_count == 0:
            print "All done"
            break

        print str(incomplete_count) + " Tasks Remaining"
        print str(float(task_count - incomplete_count) / task_count * 100) + "% Complete"
        time.sleep(.25)

    pool.close()
    pool.join()