多处理池map_async的意外行为

Question

我有一些代码对python 3应用程序中的多个文件执行相同的操作，因此似乎是multiprocessing的绝佳候选者。我正在尝试使用Pool将工作分配给一定数量的流程。我希望代码在这些计算进行时继续做其他事情(主要是为用户显示东西)，所以我想为此使用multiprocessing.Pool类的map_async函数。我希望在调用这个之后，代码将继续，结果将由我指定的回调处理，但这似乎没有发生。以下代码显示了我尝试调用map_async的三种方式以及我看到的结果：

import multiprocessing
NUM_PROCS = 4
def func(arg_list):
arg1 = arg_list[0]
arg2 = arg_list[1]
print('start func')
print ('arg1 = {0}'.format(arg1))
print ('arg2 = {0}'.format(arg2))
time.sleep(1)
result1 = arg1 * arg2
print('end func')
return result1
def callback(result):
print('result is {0}'.format(result))

def error_handler(error1):
print('error in calln {0}'.format(error1))

def async1(arg_list1):
# This is how my understanding of map_async suggests i should
# call it. When I execute this, the target function func() is not called
with multiprocessing.Pool(NUM_PROCS) as p1:
r1 = p1.map_async(func,
arg_list1,
callback=callback,
error_callback=error_handler)

def async2(arg_list1):
with multiprocessing.Pool(NUM_PROCS) as p1:
# If I call the wait function on the result for a small
# amount of time, then the target function func() is called
# and executes sucessfully in 2 processes, but the callback
# function is never called so the results are not processed
r1 = p1.map_async(func,
arg_list1,
callback=callback,
error_callback=error_handler)
r1.wait(0.1)

def async3(arg_list1):
# if I explicitly call join on the pool, then the target function func()
# successfully executes in 2 processes and the callback function is also
# called, but by calling join the processing is not asynchronous any more
# as join blocks the main process until the other processes are finished.
with multiprocessing.Pool(NUM_PROCS) as p1:
r1 = p1.map_async(func,
arg_list1,
callback=callback,
error_callback=error_handler)
p1.close()
p1.join()

def main():
arg_list1 = [(5, 3), (7, 4), (-8, 10), (4, 12)]
async3(arg_list1)
print('pool executed successfully')

if __name__ == '__main__':
main()

当在 main 中调用async1、async2或async3时，结果在每个函数的注释中描述。谁能解释为什么不同的呼叫会以这种方式运行？最终，我想像在async1中一样调用map_async，这样我就可以在工作进程繁忙时在主进程的其他进程中做一些事情。我已经在较旧的 RH6 linux 盒和较新的 ubuntu VM 上用 python 2.7 和 3.6 测试了这段代码，结果相同。

Answer 1

发生这种情况是因为当您将multiprocessing.Pool用作上下文管理器时，当您离开with块时会调用pool.terminate()，该块会立即退出所有工作线程，而无需等待正在进行的任务完成。

版本 3.3 中的新功能：Pool对象现在支持上下文管理协议 – 请参阅上下文管理器Types. __enter__()返回池对象，__exit__()调用terminate()。

IMO使用terminate()作为上下文管理器的__exit__方法不是一个很好的设计选择，因为似乎大多数人直觉地期望调用close()，这将等待正在进行的任务完成，然后再退出。不幸的是，您所能做的就是重构代码，不再使用上下文管理器，或者重构代码，以确保在Pool完成其工作之前不会离开with块。