How can I abort a task in a multiprocessing.Pool a

问题:

I am trying to use the multiprocessing package of python in this way:

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

From processes of the pool I want to avoid waiting those which take more than 60s to return its result. Is that possible?

回答1:

Here's a way you can do this without needing to change your worker function. The idea is to wrap the worker in another function, which will call worker in a background thread, and then wait for a result for for timeout seconds. If the timeout expires, it raises an exception, which will abruptly terminate the thread worker is executing in:

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")
        p.terminate()
        raise

if __name__ == "__main__":
    pool = multiprocessing.Pool()
    featureClass = [[1000,k,1] for k in drange(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()

Any function that timeouts will raise multiprocessing.TimeoutError. Note that this means your callback won't execute when a timeout occurs. If this isn't acceptable, just change the except block of abortable_worker to return something instead of calling raise.

回答2:

we can use gevent.Timeout to set time of worker running . 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)