Python Multiprocessing.Pool 惰性迭代

我想知道 python 的 Multiprocessing.Pool 类与 map、imap 和 map_async 一起使用的方式。我的特殊问题是，我想映射一个创建内存密集型对象的迭代器，并且不希望所有这些对象同时生成到内存中。我想看看各种 map() 函数是否会耗尽我的迭代器，或者仅在子进程缓慢前进时智能地调用 next() 函数，因此我编写了一些测试：

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。然而，这是最明显的例子，因为在 g2 上简单地调用一次 next() 就会打印出生成器 g() 中的所有元素，而如果 imap “懒惰”地执行此操作，我希望它只调用 go.next () 一次，因此仅打印出“1”。

有人可以澄清正在发生的事情，以及是否有某种方法可以让进程池根据需要“延迟”评估迭代器？

Thanks,

Gabe

我们先看一下程序的结尾。

多处理模块使用atexit打电话multiprocessing.util._exit_function当你的程序结束时。

如果您删除g2.next()，你的程序很快结束。

The _exit_function最终打电话Pool._terminate_pool。主线程改变了状态pool._task_handler._state from RUN to 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是允许任务处理程序线程正常运行的原因，这意味着完全消耗生成器put中的任务workers' inqueue in the Pool._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)