Python Multiprocessing.Pool惰性迭代

我想知道python的Multiprocessing.Pool类与地图,imap和map_async工作的方式。 我特别的问题是,我想映射一个迭代器,创build内存重的对象,并不希望所有这些对象同时生成内存。 我想看看各种map()函数是否会干扰我的迭代器,或者只是在subprocess缓慢前进时智能地调用next()函数,所以我篡改了一些testing:

def g(): for el in xrange(100): print el yield el def f(x): time.sleep(1) return x*x if __name__ == '__main__': pool = Pool(processes=4) # start 4 worker processes go = g() g2 = pool.imap(f, go) g2.next() 

依此类推,使用map,imap和map_async。 然而,这是最公然的例子,因为简单地调用next()在g2上的一次打印出我的生成器g()中的所有元素,而如果imap是这样做的“懒惰”,我希望它只能调用go.next ()一次,因此只打印出'1'。

有人可以清理发生的事情吗?如果有什么办法让stream程池“懒惰地”根据需要评估迭代器?

谢谢,

加布

我们先看看程序的结尾。

程序结束时,多处理模块使用atexit来调用multiprocessing.util._exit_function

如果删除g2.next() ,程序将很快结束。

_exit_function最终调用Pool._terminate_pool 。 主线程将pool._task_handler._state的状态从RUN更改为TERMINATE 。 同时, pool._task_handler线程在Pool._handle_tasks循环,并在达到条件时退出

  if thread._state: debug('task handler found thread._state != RUN') break 

(请参阅/usr/lib/python2.6/multiprocessing/pool.py)

这是什么阻止任务处理程序完全消耗您的生成器, g() 。 如果你看看Pool._handle_tasks你会看到

  for i, task in enumerate(taskseq): ... try: put(task) except IOError: debug('could not put task on queue') break 

这是消耗你的发电机的代码。 ( taskseq不完全是你的生成器,但是由于taskseq被消耗,你的生成器也是如此)。

相反,当你调用g2.next() ,主线程会调用IMapIterator.next ,并等待它到达self._cond.wait(timeout)

主线程正在等待,而不是调用_exit_function是什么让任务处理程序线程正常运行,这意味着完全消耗生成器,因为它将任务放在workerinqueuePool._handle_tasks函数中。

底线是所有的Pool映射函数都使用它给出的整个迭代。 如果你想要大块消耗发电机,你可以这样做:

 import multiprocessing as mp import itertools import time def g(): for el in xrange(50): print el yield el def f(x): time.sleep(1) return x * x if __name__ == '__main__': pool = mp.Pool(processes=4) # start 4 worker processes go = g() result = [] N = 11 while True: g2 = pool.map(f, itertools.islice(go, N)) if g2: result.extend(g2) time.sleep(1) else: break print(result) 

我也有这个问题,并且对地图消耗了所有元素感到失望。 我编写了一个函数,它在多处理中使用Queue数据types来延迟地使用迭代器。 这与@unutbu在回答评论时所描述的相似,但是正如他指出的那样,由于没有用于重新加载队列的callback机制。 Queue数据types暴露了一个超时参数,我已经用了100毫秒来达到很好的效果。

 from multiprocessing import Process, Queue, cpu_count from Queue import Full as QueueFull from Queue import Empty as QueueEmpty def worker(recvq, sendq): for func, args in iter(recvq.get, None): result = func(*args) sendq.put(result) def pool_imap_unordered(function, iterable, procs=cpu_count()): # Create queues for sending/receiving items from iterable. sendq = Queue(procs) recvq = Queue() # Start worker processes. for rpt in xrange(procs): Process(target=worker, args=(sendq, recvq)).start() # Iterate iterable and communicate with worker processes. send_len = 0 recv_len = 0 itr = iter(iterable) try: value = itr.next() while True: try: sendq.put((function, value), True, 0.1) send_len += 1 value = itr.next() except QueueFull: while True: try: result = recvq.get(False) recv_len += 1 yield result except QueueEmpty: break except StopIteration: pass # Collect all remaining results. while recv_len < send_len: result = recvq.get() recv_len += 1 yield result # Terminate worker processes. for rpt in xrange(procs): sendq.put(None) 

该解决scheme的优点是不会将请求分配到Pool.map。 一个人不能阻止别人进步。 因人而异。 请注意,您可能想要使用不同的对象来为工作人员发送终止信号。 在这个例子中,我使用了None。

testing“的Python 2.7(r27:82525,2010年7月4日,09:01:59)[MSC v.1500 32位(英特尔)]在win32”

你想要的是在NuMap包中实现的,从网站:

NuMap是一个并行(基于线程或进程,本地或远程),缓冲,多任务,itertools.imap或multiprocessing.Pool.imap函数的替代品。 就像imap一样,它对一个序列的元素或者可迭代的元素进行评估,而且这样做是懒惰的。 懒惰可以通过“步幅”和“缓冲”参数进行调整。