我正在尝试使用多进程处理一个可能会返回段错误的函数(目前我无法控制)。在子进程遇到段错误的情况下,我希望只有该子进程失败,而其他所有子任务都能够继续/返回它们的结果。
我已经从`multiprocessing.Pool`切换到`concurrent.futures.ProcessPoolExecutor`,以避免子进程挂起(或直到任意超时)的问题,正如这个bug中所记录的那样:https://bugs.python.org/issue22393。
然而,我现在面临的问题是,当第一个子任务遇到段错误时,所有正在运行的子进程都被标记为损坏(`concurrent.futures.process.BrokenProcessPool`)。
是否有一种方法只将实际上损坏的子进程标记为损坏?
我在`Python 3.7.4`中运行的代码:
我已经从`multiprocessing.Pool`切换到`concurrent.futures.ProcessPoolExecutor`,以避免子进程挂起(或直到任意超时)的问题,正如这个bug中所记录的那样:https://bugs.python.org/issue22393。
然而,我现在面临的问题是,当第一个子任务遇到段错误时,所有正在运行的子进程都被标记为损坏(`concurrent.futures.process.BrokenProcessPool`)。
是否有一种方法只将实际上损坏的子进程标记为损坏?
我在`Python 3.7.4`中运行的代码:
import concurrent.futures
import ctypes
from time import sleep
def do_something(x):
print(f"{x}; in do_something")
sleep(x*3)
if x == 2:
# raise a segmentation fault internally
return x, ctypes.string_at(0)
return x, x-1
nums = [1, 2, 3, 1.5]
executor = concurrent.futures.ProcessPoolExecutor()
result_futures = []
for num in nums:
# Using submit with a list instead of map lets you get past the first exception
# Example: https://dev59.com/m1UK5IYBdhLWcg3w9jnb#53346191
future = executor.submit(do_something, num)
result_futures.append(future)
# Wait for all results
concurrent.futures.wait(result_futures)
# After a segfault is hit for any child process (i.e. is "terminated abruptly"), the process pool becomes unusable
# and all running/pending child processes' results are set to broken
for future in result_futures:
try:
print(future.result())
except concurrent.futures.process.BrokenProcessPool:
print("broken")
结果:
(1, 0)
broken
broken
(1.5, 0.5)
期望的结果:
(1, 0)
broken
(3, 2)
(1.5, 0.5)