多處理引發的異常未檢測到池

Question

似乎是從multiprocessing.Pool進程引發異常時，沒有堆棧跟蹤或任何其他指示失敗。例如：

from multiprocessing import Pool 

def go(): 
    print(1) 
    raise Exception() 
    print(2) 

p = Pool() 
p.apply_async(go) 
p.close() 
p.join() 

打印1並停止靜音。有趣的是，引發一個BaseException反而起作用。有什麼辦法讓所有異常的行爲與BaseException相同嗎？

Answer 1

我對這個問題的合理解決，至少在調試的目的。目前我還沒有一個解決方案可以在主流程中引發異常。我的第一個想法是使用裝飾，但你只能醃functions defined at the top level of a module，所以這是正確的。

取而代之的是一個簡單的包裝類和一個Pool子類，它使用apply_async（因此爲apply）。我將離開map_async作爲讀者的練習。

import traceback 
from multiprocessing.pool import Pool 
import multiprocessing 

# Shortcut to multiprocessing's logger 
def error(msg, *args): 
    return multiprocessing.get_logger().error(msg, *args) 

class LogExceptions(object): 
    def __init__(self, callable): 
     self.__callable = callable 

    def __call__(self, *args, **kwargs): 
     try: 
      result = self.__callable(*args, **kwargs) 

     except Exception as e: 
      # Here we add some debugging help. If multiprocessing's 
      # debugging is on, it will arrange to log the traceback 
      error(traceback.format_exc()) 
      # Re-raise the original exception so the Pool worker can 
      # clean up 
      raise 

     # It was fine, give a normal answer 
     return result 

class LoggingPool(Pool): 
    def apply_async(self, func, args=(), kwds={}, callback=None): 
     return Pool.apply_async(self, LogExceptions(func), args, kwds, callback) 

def go(): 
    print(1) 
    raise Exception() 
    print(2) 

multiprocessing.log_to_stderr() 
p = LoggingPool(processes=1) 

p.apply_async(go) 
p.close() 
p.join() 

這給了我：

1 
[ERROR/PoolWorker-1] Traceback (most recent call last): 
    File "mpdebug.py", line 24, in __call__ 
    result = self.__callable(*args, **kwargs) 
    File "mpdebug.py", line 44, in go 
    raise Exception() 
Exception 

Answer 2

我會嘗試使用PDB：

import pdb 
import sys 
def handler(type, value, tb): 
    pdb.pm() 
sys.excepthook = handler 

Answer 3

也許我失去了一些東西，但是，這不是什麼結果對象返回的get方法？見Process Pools。

類multiprocessing.pool.AsyncResult

類由Pool.apply_async（）和Pool.map_async（）返回的結果。獲得（[超時]）
返回結果當它到達。如果超時時間不爲「無」，並且結果未超過 超時秒數，則會引發multiprocessing.TimeoutError。如果遠程調用異常，則該異常將由get（）重新調整。

所以，稍微修改你的榜樣，一個可以做

from multiprocessing import Pool 

def go(): 
    print(1) 
    raise Exception("foobar") 
    print(2) 

p = Pool() 
x = p.apply_async(go) 
x.get() 
p.close() 
p.join() 

其中給出的結果

1 
Traceback (most recent call last): 
    File "rob.py", line 10, in <module> 
    x.get() 
    File "/usr/lib/python2.6/multiprocessing/pool.py", line 422, in get 
    raise self._value 
Exception: foobar 

這並不完全令人滿意，因爲它不打印回溯，但有總比沒有好。

UPDATE：此bug已被固定在Python 3.4，理查德奧德柯克的禮貌。請參閱問題get method of multiprocessing.pool.Async should return full traceback。

Answer 4

我創建了一個模塊RemoteException.py，顯示在一個過程中異常的全面回溯。 Python2。Download it並添加到您的代碼：

import RemoteException 

@RemoteException.showError 
def go(): 
    raise Exception('Error!') 

if __name__ == '__main__': 
    import multiprocessing 
    p = multiprocessing.Pool(processes = 1) 
    r = p.apply(go) # full traceback is shown here 

Answer 5

我已經受夠了這個裝飾的成功記錄例外：

import traceback, functools, multiprocessing 

def trace_unhandled_exceptions(func): 
    @functools.wraps(func) 
    def wrapped_func(*args, **kwargs): 
     try: 
      func(*args, **kwargs) 
     except: 
      print 'Exception in '+func.__name__ 
      traceback.print_exc() 
    return wrapped_func 

與問題的代碼，這是

@trace_unhandled_exceptions 
def go(): 
    print(1) 
    raise Exception() 
    print(2) 

p = multiprocessing.Pool(1) 

p.apply_async(go) 
p.close() 
p.join() 

簡單裝飾你傳遞給你的進程池的函數。這項工作的關鍵是@functools.wraps(func)，否則多處理將拋出PicklingError。上述

代碼給出

1 
Exception in go 
Traceback (most recent call last): 
    File "<stdin>", line 5, in wrapped_func 
    File "<stdin>", line 4, in go 
Exception 

Answer 6

與寫作的時候票數最多的解決方案的一個問題：

from multiprocessing import Pool 

def go(): 
    print(1) 
    raise Exception("foobar") 
    print(2) 

p = Pool() 
x = p.apply_async(go) 
x.get() ## waiting here for go() to complete... 
p.close() 
p.join() 

由於@dfrankow指出，它會等待x.get()，這廢墟異步運行任務的點。因此，爲了提高效率（特別是如果你的工人功能go需要很長的時間）我將其更改爲：

from multiprocessing import Pool 

def go(x): 
    print(1) 
    # task_that_takes_a_long_time() 
    raise Exception("Can't go anywhere.") 
    print(2) 
    return x**2 

p = Pool() 
results = [] 
for x in range(1000): 
    results.append(p.apply_async(go, [x])) 

p.close() 

for r in results: 
    r.get() 

優勢：工人功能異步運行，所以例如，如果你正在運行多在幾個核心上的任務，它將比原來的解決方案高效得多。

缺點：如果在工人功能的異常，它只會提高池後已經完成了所有的任務。這可能是也可能不是理想的行爲。編輯根據@ colinfang的評論，這解決了這個問題。

Answer 7

既然你已經使用apply_sync，我猜這個用例是想做一些同步任務。使用回調進行處理是另一種選擇。請注意，此選項僅適用於python3.2及更高版本，python2.7不適用。

from multiprocessing import Pool 

def callback(result): 
    print('success', result) 

def callback_error(result): 
    print('error', result) 

def go(): 
    print(1) 
    raise Exception() 
    print(2) 

p = Pool() 
p.apply_async(go, callback=callback, error_callback=callback_error) 

# You can do another things 

p.close() 
p.join() 

Answer 8

import logging 
from multiprocessing import Pool 

def proc_wrapper(func, *args, **kwargs): 
    """Print exception because multiprocessing lib doesn't return them right.""" 
    try: 
     return func(*args, **kwargs) 
    except Exception as e: 
     logging.exception(e) 
     raise 

def go(x): 
    print x 
    raise Exception("foobar") 

p = Pool() 
p.apply_async(proc_wrapper, (go, 5)) 
p.join() 
p.close() 

Answer 9

由於已經爲multiprocessing.Pool提供像樣的答案，我將提供使用的完整性不同的方法解決。

對於python >= 3.2以下解決方案似乎是最簡單的：

from concurrent.futures import ProcessPoolExecutor, wait 

def go(): 
    print(1) 
    raise Exception() 
    print(2) 


futures = [] 
with ProcessPoolExecutor() as p: 
    for i in range(10): 
     futures.append(p.submit(go)) 

results = [f.result() for f in futures] 

優點：

• 很少的代碼
• 提出的主要工序異常
• 提供一個堆棧跟蹤
• 無外部依存關係

有關API的更多信息請查看：https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor

此外，如果您要提交大量的任務，你想你的主要過程，儘快失敗，因爲你的任務之一失敗了，你可以請使用以下代碼片段：

from concurrent.futures import ProcessPoolExecutor, wait, FIRST_EXCEPTION, as_completed 
import time 


def go(): 
    print(1) 
    time.sleep(0.3) 
    raise Exception() 
    print(2) 


futures = [] 
with ProcessPoolExecutor(1) as p: 
    for i in range(10): 
     futures.append(p.submit(go)) 

    for f in as_completed(futures): 
     if f.exception() is not None: 
      for f in futures: 
       f.cancel() 
      break 

[f.result() for f in futures] 

所有其他答案只在所有任務執行完成後纔會失敗。