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

Posted 2023-02-16

【中文标题】超时后如何中止 multiprocessing.Pool 中的任务？

【英文标题】：How can I abort a task in a multiprocessing.Pool after a timeout?

【发布时间】：2015-06-12 04:36:53

【问题描述】：

我正在尝试以这种方式使用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 秒才能返回结果的进程。这可能吗？

【问题讨论】：

worker 是什么样的？使用multiprocessing.Pool 执行此操作的最简单方法是使worker 可中断，但这可能是不可能的，具体取决于它在做什么。

worker 是一个简单的函数，有一个列表输入和一个列表输出

它实际上在做什么呢？我假设它正在迭代列表，但是它对每个项目执行什么样的操作？每次手术需要多长时间？

好的！ worker 的任务是训练一个 svm 模型（#machine_learning, #classification）并返回经过的时间

我对同样的问题有一个可行的解决方案，发布在这里：***.com/a/40724036/2512195

【参考方案1】：

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

使用您可以传递给multiprocessing.Pool 的maxtasksperchild 选项确保池中的工作进程在每次任务执行后重新启动。 将现有的工作函数包装在另一个函数中，该函数将在守护线程中调用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。请注意，这意味着当发生超时时您的回调将不会执行。如果这不可接受，只需更改 abortable_worker 的 except 块以返回某些内容，而不是调用 raise。

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

【讨论】：

再问一个问题，你的代码中有多少并行进程？

@wajdi 它使用默认的进程数，它始终等于执行脚本的机器上的 CPU 数。如果要指定数字，请将其传递给multiprocessing.Pool 构造函数：pool = multiprocessing.Pool(4)。

完美，所以pool = multiprocessing.Pool(N) 将并行启动 N 个工作人员，但如果我有超过 N 个工作人员，例如 Nx2 怎么办？工人将按什么顺序启动.. 感谢您抽出宝贵时间@dano

有效吗？我尝试运行上面的示例代码，将worker func更改为def worker(x, y, z): \n while 1:\n pass，过程不会结束。 ***.com/a/24634225/3291799 确实有效，但方式很危险。

@jjjjjj worker 是包含您要在子进程中执行的任何代码的函数。

【参考方案2】：

我们可以使用 gevent.Timeout 来设置 worker 运行的时间。 gevent tutorial

from multiprocessing.dummy import Pool 
#you should install gevent.
from gevent import Timeout
from gevent import monkey
monkey.patch_all()
import time

def worker(sleep_time):
    try:

        seconds = 5  # max time the worker may run
        timeout = Timeout(seconds) 
        timeout.start()
        time.sleep(sleep_time)
        print "%s is a early bird"%sleep_time
    except:
        print "%s is late(time out)"%sleep_time

pool = Pool(4)

pool.map(worker, range(10))


output:
0 is a early bird
1 is a early bird
2 is a early bird
3 is a early bird
4 is a early bird
8 is late(time out)
5 is late(time out)
6 is late(time out)
7 is late(time out)
9 is late(time out)

【讨论】：

请提供一些文字cmets。

为什么需要猴子补丁？这只是 gevent 必须做的事情才能使事情变得非阻塞吗？

这对我不起作用。猴子补丁抛出错误

这将无限期挂起，除非使用monkey.patch_all(thread=False)