ошибки отладки в многопроцессорности python

Question

Я использую функцию Pool функции модуля multiprocessing, чтобы параллельно работать с одним и тем же кодом на разных данных.

Оказывается, что на некоторых данных мой код вызывает исключение, но четкая линия, в которой это происходит не дано:

Traceback (most recent call last): 
    File "my_wrapper_script.py", line 366, in <module> 
    main() 
    File "my_wrapper_script.py", line 343, in main 
    results = pool.map(process_function, folders) 
    File "/usr/lib64/python2.6/multiprocessing/pool.py", line 148, in map 
    return self.map_async(func, iterable, chunksize).get() 
    File "/usr/lib64/python2.6/multiprocessing/pool.py", line 422, in get 
    raise self._value 
KeyError: 'some_key' 

Я знаю multiprocessing.log_to_stderr(), но мне кажется, что это полезно, когда возникают проблемы параллелизма, что не является моим делом.

Любые идеи?

Answer 1

Если вы используете новую достаточно новую версию Python, вы увидите, что реальное исключение печатается до этого. Например, вот пример, который терпит неудачу:

import multiprocessing 

def inner(): 
    raise Exception("FAIL") 

def f(): 
    print("HI") 
    inner() 

p = multiprocessing.Pool() 
p.apply(f) 
p.close() 
p.join() 

Вот исключение при выполнении этого с питоном 3.4:

multiprocessing.pool.RemoteTraceback: 
""" 
Traceback (most recent call last): 
    File "/usr/local/lib/python3.4/multiprocessing/pool.py", line 119, in worker 
    result = (True, func(*args, **kwds)) 
    File "test.py", line 9, in f 
    inner() 
    File "test.py", line 4, in inner 
    raise Exception("FAIL") 
Exception: FAIL 
""" 

The above exception was the direct cause of the following exception: 

Traceback (most recent call last): 
    File "test.py", line 13, in <module> 
    p.apply(f) 
    File "/usr/local/lib/python3.4/multiprocessing/pool.py", line 253, in apply 
    return self.apply_async(func, args, kwds).get() 
    File "/usr/local/lib/python3.4/multiprocessing/pool.py", line 599, in get 
    raise self._value 
Exception: FAIL 

При использовании более новой версии не вариант, проще всего сделать, это чтобы обернуть функцию уборщица в попытке/за исключением блока, который будет печатать исключение до повторного повышения его:

import multiprocessing 
import traceback 

def inner(): 
    raise Exception("FAIL") 

def f(): 
    try: 
     print("HI") 
     inner() 
    except Exception: 
     print("Exception in worker:") 
     traceback.print_exc() 
     raise 

p = multiprocessing.Pool() 
p.apply(f) 
p.close() 
p.join() 

Выход:

HI 
Exception in worker: 
Traceback (most recent call last): 
    File "test.py", line 11, in f 
    inner() 
    File "test.py", line 5, in inner 
    raise Exception("FAIL") 
Exception: FAIL 
Traceback (most recent call last): 
    File "test.py", line 18, in <module> 
    p.apply(f) 
    File "/usr/local/lib/python2.7/multiprocessing/pool.py", line 244, in apply 
    return self.apply_async(func, args, kwds).get() 
    File "/usr/local/lib/python2.7/multiprocessing/pool.py", line 558, in get 
    raise self._value 
Exception: FAIL 

Answer 2

Вам необходимо реализовать свой собственный блок try/except в рабочем состоянии. В зависимости от того, как вы хотите организовать свой код, вы можете войти в stderr, как вы упомянули выше, зайти в другое место, например файл, вернуть какой-то код ошибки или даже пометить исключение с текущей трассировкой и повторным рейзом. Вот пример последней техники:

import traceback 
import multiprocessing as mp 

class MyError(Exception): 
    pass 

def worker(): 
    try: 
     # your real code here 
     raise MyError("boom") 
    except Exception, e: 
     e.traceback = traceback.format_exc() 
     raise 

def main(): 
    pool = mp.Pool() 
    try: 
     print "run worker" 
     result = pool.apply_async(worker) 
     result.get() 
    # handle exceptions you expect 
    except MyError, e: 
     print e.traceback 
    # re-raise the rest 
    except Exception, e: 
     print e.traceback 
     raise 


if __name__=="__main__": 
    main() 

возвращает

run worker 
Traceback (most recent call last): 
    File "doit.py", line 10, in worker 
    raise MyError("boom") 
MyError: boom