Python中ThreadPool中每个线程的超时

Question

我正在使用Python 2.7。

我目前使用ThreadPoolExecuter这样的：

params = [1,2,3,4,5,6,7,8,9,10] 
with concurrent.futures.ThreadPoolExecutor(5) as executor: 
    result = list(executor.map(f, params)) 

问题是f有时运行时间过长。每当我运行f，我想限制它的运行到100秒，然后杀死它。

最终，对于xparam中的每个元素，我想指出是否必须杀死f，如果不是 - 返回值是多少。 即使f超时一个参数，我仍然想运行它与下一个参数。

executer.map方法确实有一个timeout参数，但它为整个运行设置了一个超时时间，从调用时间到executer.map，而不是分别为每个线程。

什么是最简单的方法来获得我想要的行为？

Answer 1

这个答案是关于python的多处理库，它通常比线程库更可取，除非你的函数正在等待网络调用。请注意，多处理和线程库具有相同的接口。

鉴于您每个进程运行潜在的100秒，相比之下，创建每个进程的开销相当小。您可能必须制定自己的流程才能获得必要的控制权。

一种选择是包装在另一个函数f将为是100秒exectue：

from multiprocessing import Pool 

def timeout_f(arg): 
    pool = Pool(processes=1) 
    return pool.apply_async(f, [arg]).get(timeout=100) 

然后你的代码更改为：

result = list(executor.map(timeout_f, params)) 

---

或者，你可以写您自己的线程/过程控制：

from multiprocessing import Process 
from time import time 

def chunks(l, n): 
    """ Yield successive n-sized chunks from l. """ 
    for i in xrange(0, len(l), n): 
     yield l[i:i+n] 

processes = [Process(target=f, args=(i,)) for i in params] 
exit_codes = [] 
for five_processes = chunks(processes, 5): 
    for p in five_processes: 
     p.start() 
    time_waited = 0 
    start = time() 
    for p in five_processes: 
     if time_waited >= 100: 
      p.join(0) 
      p.terminate() 
     p.join(100 - time_waited) 
     p.terminate() 
     time_waited = time() - start 
    for p in five_processes: 
     exit_codes.append(p.exit_code) 

您需要通过类似Can I get a return value from multiprocessing.Process?

得到返回值如果进程完成，进程的退出代码为0，如果它们已终止，进程的退出代码为非零。

技术来自： Join a group of python processes with a timeout，How do you split a list into evenly sized chunks?

---

作为另一种选择，你可以只尝试使用apply_async上multiprocessing.Pool

from multiprocessing import Pool, TimeoutError 
from time import sleep  

if __name__ == "__main__": 
    pool = Pool(processes=5) 
    processes = [pool.apply_async(f, [i]) for i in params] 
    results = [] 
    for process in processes: 
     try: 
      result.append(process.get(timeout=100)) 
     except TimeoutError as e: 
      results.append(e) 

注意上面可能等待超过100秒，每过程，就好像第一个过程需要50秒完成，第二个过程在其运行时间内将有50秒多余的时间。更复杂的逻辑（如前面的例子）需要执行更严格的超时。