您好, 欢迎来到 !    登录 | 注册 | | 设为首页 | 收藏本站

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

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

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

如果超时到期,则退出(或中止- 由您决定)包装函数,该函数将终止任务,并且由于已设置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。请注意,这意味着超时时不会执行您的回调。如果这是不可接受的,则只需更改的exceptabortable_worker以返回某些内容,而不是调用raise

还请记住Pool,由于增加的开销,在每次执行任务后重新启动工作进程将对的性能产生负面影响。您应该针对您的用例进行衡量,并考虑是否值得进行折中以具有中止工作的能力。如果有问题,则可能需要尝试另一种方法,例如,worker如果运行时间过长,请合作中断,而不是尝试从外部终止它。关于SO的很多问题都涉及此主题

其他 2022/1/1 18:26:34 有638人围观

撰写回答


你尚未登录,登录后可以

和开发者交流问题的细节

关注并接收问题和回答的更新提醒

参与内容的编辑和改进,让解决方法与时俱进

请先登录

推荐问题


联系我
置顶