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

Question

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

我发现了围绕这个问题的各种问题(通用多处理错误处理，如何在异常时关闭多处理池但没有答案，...(，但没有关于如何在子异常时停止多处理的明确答案。

例如，我希望以下代码：

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函数：

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

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

Answer 1

我认为您将需要apply_async，因此您可以对每个结果而不是累积结果进行操作。pool.apply_async提供了一个可用于注册错误处理程序的error_callback参数。apply_async不会阻塞，因此您需要join()池。我还使用标志terminated来了解在没有发生异常的情况下何时可以正常处理结果。

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