python - 超时后如何中止 multiprocessing.Pool 中的任务？

Question

我正在尝试以这种方式使用 python 的多处理包:

featureClass = [[1000, k, 1] for k in drange(start, end, step)] #list of arguments
for f in featureClass:
  pool.apply_async(worker, args=f, callback=collectMyResult)
pool.close()
pool.join

在池的进程中，我想避免等待那些需要超过 60 秒才能返回结果的进程。这可能吗？

Answer 1

这是一种无需更改 worker 函数即可执行此操作的方法。需要两个步骤:

1. 使用可以传递给 multiprocessing.Pool 的 maxtasksperchild 选项来确保池中的工作进程在每次任务执行后重新启动。
2. 将您现有的工作函数包装在另一个函数中，该函数将在守护线程中调用 worker，然后等待该线程的结果 timeout 秒。使用守护线程很重要，因为进程在退出前不会等待守护线程完成。

如果超时到期，您退出(或中止 - 这取决于您)包装函数，这将结束任务，并且因为您设置了 maxtasksperchild=1，导致 Pool 终止工作进程并启动一个新进程。这意味着执行您实际工作的后台线程也会中止，因为它是守护线程，并且它所在的进程已关闭。

import multiprocessing
from multiprocessing.dummy import Pool as ThreadPool
from functools import partial

def worker(x, y, z):
    pass # Do whatever here

def collectMyResult(result):
    print("Got result {}".format(result))

def abortable_worker(func, *args, **kwargs):
    timeout = kwargs.get('timeout', None)
    p = ThreadPool(1)
    res = p.apply_async(func, args=args)
    try:
        out = res.get(timeout)  # Wait timeout seconds for func to complete.
        return out
    except multiprocessing.TimeoutError:
        print("Aborting due to timeout")
        raise

if __name__ == "__main__":
    pool = multiprocessing.Pool(maxtasksperchild=1)
    featureClass = [[1000,k,1] for k in range(start,end,step)] #list of arguments
    for f in featureClass:
      abortable_func = partial(abortable_worker, worker, timeout=3)
      pool.apply_async(abortable_func, args=f,callback=collectMyResult)
    pool.close()
    pool.join()

任何超时的函数都会引发multiprocessing.TimeoutError。请注意，这意味着您的回调不会在超时发生时执行。如果这是 Not Acceptable ，只需更改 abortable_worker 的 except block 以返回某些内容，而不是调用 raise。

另请记住，在每次任务执行后重新启动工作进程会对 Pool 的性能产生负面影响，因为会增加开销。您应该针对您的用例衡量这一点，看看是否值得权衡是否有能力中止工作。如果这是一个问题，您可能需要尝试另一种方法，比如在 worker 运行时间过长时合作中断它，而不是试图从外部杀死它。有很多关于 SO 的问题都涵盖了这个主题。