Python多处理进程以静默方式崩溃

Question

我正在使用Python 2.7.3.我使用子类multiprocessing.Process对象并行化了一些代码.如果我的子类Process对象中的代码没有错误,那么一切运行正常.但是,如果有在我的子类的过程对象代码中的错误,他们显然会崩溃默默(没有堆栈跟踪打印到父shell)和CPU使用率将下降到零.父代码永远不会崩溃,给人的印象是执行只是挂起.同时,很难追踪代码中的错误,因为没有给出关于错误位置的指示.

我在stackoverflow上找不到任何其他问题来处理同样的问题.

我想子类化的Process对象似乎是静默崩溃的,因为它们无法向父shell发送错误消息,但我想知道我能做些什么,这样我至少可以更高效地调试(以及其他我的代码的用户可以告诉我他们何时遇到问题).

编辑:我的实际代码过于复杂,但在它的错误的子类的处理对象的一个简单的例子是这样的:

from multiprocessing import Process, Queue

class Worker(Process):

    def __init__(self, inputQueue, outputQueue):

        super(Worker, self).__init__()

        self.inputQueue = inputQueue
        self.outputQueue = outputQueue

    def run(self):

        for i in iter(self.inputQueue.get, 'STOP'):

            # (code that does stuff)

            1 / 0 # Dumb error

            # (more code that does stuff)

            self.outputQueue.put(result)

Answer 1

你真正想要的是将异常传递给父进程的一些方法,对吧？然后你可以随意处理它们.

如果使用concurrent.futures.ProcessPoolExecutor,这是自动的.如果你使用multiprocessing.Pool,它是微不足道的.如果使用显式的Process和Queue,你必须做一些工作,但它不是说太多.

例如:

def run(self):
    try:
        for i in iter(self.inputQueue.get, 'STOP'):
            # (code that does stuff)
            1 / 0 # Dumb error
            # (more code that does stuff)
            self.outputQueue.put(result)
    except Exception as e:
        self.outputQueue.put(e)

然后,您的调用代码可以Exception像其他任何东西一样从队列中读取s.而不是这个:

yield outq.pop()

做这个:

result = outq.pop()
if isinstance(result, Exception):
    raise result
yield result

(我不知道你的实际父进程队列读取代码是做什么的,因为你的最小样本只是忽略了队列.但希望这解释了这个想法,即使你的真实代码实际上并没有像这样工作.)

这假定您要中止任何未处理的异常run.如果你想传回异常并继续下一个异常i in iter,只需将其try移入for,而不是围绕它.

这也假定Exceptions不是有效值.如果这是一个问题,最简单的解决方案就是推送(result, exception)元组:

def run(self):
    try:
        for i in iter(self.inputQueue.get, 'STOP'):
            # (code that does stuff)
            1 / 0 # Dumb error
            # (more code that does stuff)
            self.outputQueue.put((result, None))
    except Exception as e:
        self.outputQueue.put((None, e))

然后,您的弹出代码执行此操作:

result, exception = outq.pop()
if exception:
    raise exception
yield result

您可能会注意到这与node.js回调样式类似,您将传递(err, result)给每个回调.是的,这很烦人,而且你会以这种方式弄乱代码.但除了包装器外,你实际上并没有使用它; 从队列中获取值或在内部调用的所有"应用程序级"代码run只看到正常的返回/收益和引发的异常.

您甚至可能想要考虑构建一个Future规范concurrent.futures(或按原样使用该类),即使您正在排队并手动执行.它并不难,它为您提供了一个非常好的API,特别是对于调试.

最后,值得注意的是,使用执行程序/池设计可以使围绕工作程序和队列构建的大多数代码变得更加简单,即使您完全确定每个队列只需要一个工作程序.只需废弃所有样板,然后将Worker.run方法中的循环转换为函数(正常情况下returns或raises,而不是附加到队列中).在呼叫方,再次废弃所有样板和刚刚submit或map作业功能及其参数.

您的整个示例可以简化为:

def job(i):
    # (code that does stuff)
    1 / 0 # Dumb error
    # (more code that does stuff)
    return result

with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
    results = executor.map(job, range(10))

并且它会自动正确处理异常.

正如您在评论中提到的,异常的回溯不会追溯到子进程; 它只能进行手动raise result调用(或者,如果您使用的是池或执行程序,那么池或执行程序的内容).

原因是它multiprocessing.Queue建立在最基础之上pickle,并且酸洗异常并没有腌制它们的追溯.原因是你不能腌制回溯.原因是回溯充满了对本地执行上下文的引用,因此使它们在另一个进程中工作将非常困难.

那么......你能做些什么呢？不要去寻找一个完全通用的解决方案.相反,想想你真正需要什么.90%的情况下,你想要的是"记录异常,使用traceback,并继续"或"打印异常,使用traceback,stderr并且exit(1)像默认的未处理异常处理程序一样".对于其中任何一个,您根本不需要传递异常; 只需在子端进行格式化并传递一个字符串.如果你确实需要一些更加花哨的东西,请确切地计算出你需要的东西,并传递足够的信息来手动将它们组合在一起.如果您不知道如何格式化回溯和异常,请参阅该traceback模块.这很简单.这意味着你根本不需要进入泡菜机械.(并不是说copyreg挑选者很难或用__reduce__方法或任何东西写一个持有人类,但如果你不需要,为什么要学习这一切？)