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

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 方法返回的内容吗？请参阅进程池。

类 multiprocessing.pool.AsyncResult

Pool.apply_async() 和 Pool.map_async().get([timeout]) 返回结果的类
当结果到达时返回。如果 timeout 不是 None 并且结果没有在
超时秒然后 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

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

更新：此错误已在 Python 3.4 中修复，由 Richard Oudkerk 提供。请参阅问题multiprocessing.pool.Async 的 get 方法应返回完整的回溯。

Answer 2

我对这个问题有一个合理的解决方案，至少出于调试目的。我目前没有解决方案可以在主流程中引发异常。我的第一个想法是使用装饰器，但你只能 pickle 在模块顶层定义的函数，所以这是正确的。

相反，一个简单的包装类和一个 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 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()

优点：辅助函数是异步运行的，因此如果例如，您在多个内核上运行许多任务，它将比原始解决方案效率更高。

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

Answer 4

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

此外，如果您要提交一个大型任务数，并且您希望主进程在其中一个任务失败时立即失败，您可以使用以下代码片段：

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

由于您已经使用了 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，它显示进程中异常的完整回溯。 Python2。 下载并将其添加到您的代码中：

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