如何在超时后中止multiprocessing.Pool中的任务?
我正在尝试以这种方式使用python的多处理程序包:
featureClass = [[1000,k,1] for k in drange(start,end,step)] #list of argumentsfor f in featureClass:
pool .apply_async(worker, args=f,callback=collectMyResult)
pool.close()
pool.join
从池的进程中,我要避免等待等待60多个返回结果的进程。那可能吗?
回答:
这是一种无需更改worker
功能即可执行此操作的方法。需要两个步骤:
- 使用
maxtasksperchild
您可以传递的选项multiprocessing.Pool
来确保每次执行任务后重新启动池中的工作进程。 - 将现有的辅助函数包装在另一个函数中,该函数将调用
worker
守护程序线程,然后等待该线程的结果timeout
数秒钟。使用守护程序线程很重要,因为进程在退出之前不会等待守护程序线程完成。
如果超时到期,则退出(或中止-
由您决定)包装函数,该函数将终止任务,并且由于已设置maxtasksperchild=1
,导致Pool
终止工作进程并启动一个新的进程。这将意味着从事实际工作的后台线程也被中止,因为它是一个守护程序线程,并且正在运行的进程已关闭。
import multiprocessingfrom 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
。请注意,这意味着超时时不会执行您的回调。如果这是不可接受的,则只需更改的except
块abortable_worker
以返回某些内容,而不是调用raise
。
还请记住Pool
,由于增加的开销,在每次执行任务后重新启动工作进程将对的性能产生负面影响。您应该针对您的用例进行衡量,并考虑是否值得进行折中以具有中止工作的能力。如果有问题,则可能需要尝试另一种方法,例如,worker
如果运行时间过长,请合作中断,而不是尝试从外部终止它。关于SO的很多问题都涉及此主题。
以上是 如何在超时后中止multiprocessing.Pool中的任务? 的全部内容, 来源链接: utcz.com/qa/409363.html