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秒多餘的時間。更復雜的邏輯（如前面的例子）需要執行更嚴格的超時。