Pool.map - Почему рабочий процесс не сработал раньше?

Question

Скажите, что я это сделать:

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() 

Out:

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 

Ok выход имеет смысл. Исключение возникает перед оператором 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() 

Out:

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 трасс стека. Почему он печатает все числа и только одну трассировку стека? Почему рабочий процесс ждет самого конца, чтобы потерпеть крах?

Answer 1

Это просто хронометраж - процесс 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. В это время рабочий может выполнить все остальные задачи. Если вы сделаете функцию работника медленнее, вы увидите, что он выполняет меньше заданий:

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, поэтому результаты остальных задач отбрасываются.