未检测到多处理池中引发的异常

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

也许我遗漏了一些东西，但这不是 Result 对象的 get 方法返回的内容吗？见Process Pools。

类 multiprocessing.pool.AsyncResult

Pool.apply_async() 和 Pool.map_async().get([timeout]) 返回结果的类
到达时返回结果。如果 timeout 不是 None 并且结果没有在 timeout 秒然后 multiprocessing.TimeoutError 被提出。如果遥控器 call 引发异常，然后 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

这并不完全令人满意，因为它不打印回溯，但总比没有好。

更新：此错误已在 Python 3.4 中得到修复，由 Richard Oudkerk 提供。请参阅问题get method of multiprocessing.pool.Async should return full traceback。

Answer 2

我有一个合理的解决方案，至少用于调试目的。我目前没有可以在主进程中引发异常的解决方案。我的第一个想法是使用装饰器，但你只能腌制functions defined at the top level of a module，所以就这样了。

取而代之的是一个简单的包装类和一个将其用于apply_async（以及因此apply）的Pool 子类。我会把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 3

撰写本文时得票最多的解决方案有问题：

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

优点：worker函数是异步运行的，所以如果你在多个内核上运行许多任务，它会比原来的解决方案效率高很多。

缺点：如果worker函数中出现异常，只有在池完成所有任务后才会引发。这可能是也可能不是理想的行为。 根据@colinfang 的评论编辑，修复了这个问题。

Answer 4

我已经用这个装饰器成功记录了异常：

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 5

由于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 的更多信息，请查看this

此外，如果您要提交大量任务，并且希望您的主进程在您的一项任务失败后立即失败，您可以使用以下 sn-p：

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]

只有在所有任务都执行完毕后，所有其他答案才会失败。

Answer 6

既然你用过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 7

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 8

我创建了一个模块RemoteException.py，它显示了进程中异常的完整回溯。蟒蛇2。 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 9

我会尝试使用 pdb：

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