Python 多处理:在第一个子错误时中止映射

2022-01-12 00:00:00 python python-multiprocessing multiprocessing process-pool

问题描述

当其中一个孩子中止和/或抛出异常时，中止多处理的正确方法是什么?

What's the proper way of aborting multiprocessing when one of the child aborts and/or throw an Exception?

我发现了各种各样的问题(通用多处理错误处理, 如何在出现异常时关闭多处理池但没有答案，...)，但没有关于如何停止对子异常进行多处理的明确答案.

I found various questions around that (generic multiprocessing error handling, how to close multiprocessing pool on exception but without answer, ...), but no clear answer on how to stop multiprocessing on child exception.

例如，我期望以下代码:

For instance, I expect the following code:

def f(x):
    sleep(x)
    print(f"f({x})")
    return 1.0 / (x - 2)


def main():
    with Pool(4) as p:
        try:
            r = p.map(f, range(7))
        except Exception as e:
            print(f"oops: {e}")
            p.close()
            p.terminate()
    print("end")


if __name__ == '__main__':
    main()

输出:

f(0)
f(1)
f(2)
oops: float division by zero
end

相反，它在检测/处理异常之前对所有项目应用 f 函数:

Instead, it applies f function on all items before detecting/handling the exception:

f(0)
f(1)
f(2)
f(4)
f(3)
f(5)
f(6)
oops: float division by zero
end

有没有办法直接捕获异常?

Isn't there any way to catch the exception directly?

解决方案

我认为你需要 apply_async 来解决这个问题，这样你就可以对每一个结果而不是累积结果采取行动.pool.apply_async 提供了一个 error_callback 参数，您可以使用它来注册您的错误处理程序.apply_async 没有阻塞，因此您需要 join() 池.我还使用了一个标志 terminated 来知道在没有异常发生的情况下何时可以正常处理结果.

I think you're going to need apply_async for this, so you can act upon every single result instead of the cumulative result. pool.apply_async offers an error_callback parameter you can use to register your error-handler. apply_async is not blocking, so you'll need to join() the pool. I'm also using a flag terminated to know when results can be processed normally in case no exception occured.

from time import sleep
from multiprocessing import Pool

def f(x):
    sleep(x)
    print(f"f({x})")
    return 1.0 / (x - 2)

def on_error(e):
    global terminated
    terminated = True
    pool.terminate()
    print(f"oops:{e}")


def main():
    global pool
    global terminated

    terminated = False

    pool = Pool(4)
    results = [pool.apply_async(f, (x,), error_callback=on_error)
               for x in range(7)]
    pool.close()
    pool.join()

    if not terminated:
        for r in results:
            print(r.get())

    print("end")


if __name__ == '__main__':
    main()

相关文章