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

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

我有一个合理的解决方案，至少用于调试目的。我目前没有一个可以在主要流程中引发异常的解决方案。我的第一个想法是使用装饰器，但是您只能腌制在模块顶层定义的函数，所以没错。

取而代之的是一个简单的包装类和一个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