让我们首先看一下程序的结尾。
multiprocessing模块使用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
让任务处理线程正常运行,这意味着它会在Pool._handle_tasks
函数中将任务放入worker
的inqueue
中并完全消耗生成器。
总之,所有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)
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)
time.sleep
调用并在f
中添加print os.getpid(), x
后,行为看起来更加奇怪,有时只打印2或3个不同的PID,并且总是执行不同数量的迭代... 顺便问一下,您使用的是哪个Python版本? - Terseus