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秒，然后终止运行。

最终，每个元素x在param，我想有与否的指示f必须被杀死，并且如果它不是-是什么返回值。即使f对于一个参数超时，我仍然希望使用下一个参数运行它。

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

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

Answer 1

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

假设每个流程可能要运行100秒钟，相比之下，为每个流程创建一个流程的开销是很小的。您可能必须制作自己的流程才能获得必要的控制。

一种选择是将f包装到另一个函数中，该函数最多执行100秒：

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)

您必须通过类似的方法来获取返回值：我可以从multiprocessing.Process中获取返回值吗？

如果进程已完成，则进程的退出代码为0；如果终止，则退出代码为非零。

来自的技术： 具有超时的一组python进程，如何将列表分成大小均匀的块？

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

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秒。需要更复杂的逻辑（例如前面的示例）来强制执行更严格的超时。