python之Pool.map之为什么工作进程不早点崩溃

假设我这样做:

import multiprocessing as mp 
 
def f(x): 
    raise OverflowError  # raised BEFORE the print 
    print x 
 
if __name__ == '__main__': 
    pool = mp.Pool(processes=1) 
    for _ in pool.imap_unordered(f, range(10)): 
        pass 
    pool.close() 
    pool.join() 

输出:

Traceback (most recent call last): 
  File "test0.py", line 9, in <module> 
    for _ in pool.imap_unordered(f, range(10)): 
  File "/Users/usualme/anaconda/lib/python2.7/multiprocessing/pool.py", line 659, in next 
    raise value 
OverflowError 

好的输出是有意义的。异常在 print 语句之前引发，因此没有输出。现在几乎相同的代码，但我切换了 2 行:

import multiprocessing as mp 
 
def f(x): 
    print x 
    raise OverflowError  # raised AFTER the print 
 
if __name__ == '__main__': 
    pool = mp.Pool(processes=1) 
    for _ in pool.imap_unordered(f, range(10)): 
        pass 
    pool.close() 
    pool.join() 

输出:

0 
1 
2 
3 
4 
5 
6 
7 
8 
9 
Traceback (most recent call last): 
  File "test0.py", line 9, in <module> 
    for _ in pool.imap_unordered(f, range(10)): 
  File "/Users/usualme/anaconda/lib/python2.7/multiprocessing/pool.py", line 659, in next 
    raise value 
OverflowError 

我不明白输出。我期待数字 0 后跟堆栈跟踪，或者所有 10 个数字和 10 个堆栈跟踪。为什么它打印所有数字而只打印一个堆栈跟踪？为什么工作进程一直等到最后崩溃？

请您参考如下方法:

这只是一个时间问题 - worker 进程不关心它正在运行的函数中是否引发了异常，它只是将异常返回给父进程并继续执行下一个任务.这是它正在运行的循环(稍微简化):

while maxtasks is None or (maxtasks and completed < maxtasks): 
    try: 
        task = get()  # Get task from parent 
    except (EOFError, OSError): 
        util.debug('worker got EOFError or OSError -- exiting') 
        break 
 
    if task is None: 
        util.debug('worker got sentinel -- exiting') 
        break 
 
    job, i, func, args, kwds = task 
    try: 
        result = (True, func(*args, **kwds))  # Call the function pass from the parent 
    except Exception as e:  # We end up in here if the worker raises an exception 
        if wrap_exception: 
            e = ExceptionWithTraceback(e, e.__traceback__) 
        result = (False, e)  # The exception object is stored as the result 
 
    put((job, i, result)) # Send result to parent process 

因此，即使第一个任务引发了异常，结果在两个进程之间传输以及父进程实际将结果从 Queue< 中拉出也需要一点时间 并引发 Exception。在那个时间窗口中，工作人员能够执行所有剩余的任务。如果你让 worker 函数变慢，你会发现它执行的任务更少:

import multiprocessing as mp 
import time 
 
def f(x): 
    print x 
    time.sleep(2) 
    raise OverflowError  
 
if __name__ == '__main__': 
    pool = mp.Pool(processes=1) 
    for _ in pool.imap_unordered(f, range(10)): 
        pass 
    pool.close() 
    pool.join() 

输出:

0 
1 
Traceback (most recent call last): 
  File "p.py", line 11, in <module> 
    for _ in pool.imap_unordered(f, range(10)): 
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 626, in next 
    raise value 
OverflowError 

如果您传递了一个更大的可迭代对象，您也只会看到部分结果的打印结果，因为在父对象死亡之前，工作人员没有足够的时间完成所有这些操作。

您只会看到一个异常实际上被引发，因为从父级的角度来看，一旦一个任务失败，整个 imap 调用就应该中止。父进程从单个 Queue 中按顺序从其所有子进程中提取结果，因此一旦它看到第一个异常，imap 调用就会结束，因此来自其余任务被丢弃。