如何在超时后中止 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 函数即可执行此操作的方法。需要两个步骤：

使用可以传递给 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 块更改为 return 某些内容，而不是调用 raise.

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

Answer 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)

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

How can I abort a task in a multiprocessing.Pool after a timeout?

python

multiprocessing

python-multiprocessing