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

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

这是一种无需更改

worker

功能即可执行此 *** 作的方法。需要两个步骤：

1. 使用

   maxtasksperchild

   您可以传递的选项

   multiprocessing.Pool

   来确保每次执行任务后重新启动池中的工作进程。
2. 将现有的辅助函数包装在另一个函数中，该函数将调用

   worker

   守护程序线程，然后等待该线程的结果

   timeout

   数秒钟。使用守护程序线程很重要，因为进程在退出之前不会等待守护程序线程完成。

如果超时到期，则退出（或中止-
由您决定）包装函数，该函数将终止任务，并且由于已设置

maxtasksperchild=1

，导致

Pool

终止工作进程并启动一个新的进程。这将意味着从事实际工作的后台线程也被中止，因为它是一个守护程序线程，并且正在运行的进程已关闭。

import multiprocessingfrom multiprocessing.dummy import Pool as ThreadPoolfrom functools import partialdef worker(x, y, z):    pass # Do whatever heredef 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")        raiseif __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

。请注意，这意味着超时时不会执行您的回调。如果这是不可接受的，则只需更改的

except

块

abortable_worker

以返回某些内容，而不是调用

raise

。

还请记住

Pool

，由于增加的开销，在每次执行任务后重新启动工作进程将对的性能产生负面影响。您应该针对您的用例进行衡量，并考虑是否值得进行折中以具有中止工作的能力。如果有问题，则可能需要尝试另一种方法，例如，

worker

如果运行时间过长，请合作中断，而不是尝试从外部终止它。关于SO的很多问题都涉及此主题。