如何使用 concurrent.futures 超时？

Question

我正在尝试使用 concurrent.futures 模块让超时以在 python3.2 中工作。但是，当它超时时，它并没有真正停止执行。我尝试使用线程和进程池执行器，它们都没有停止任务，只有在完成之前才会引发超时。那么有谁知道它是否有可能让它工作？

import concurrent.futures
import time
import datetime

max_numbers = [10000000, 10000000, 10000000, 10000000, 10000000]

def run_loop(max_number):
    print("Started:", datetime.datetime.now(), max_number)
    last_number = 0;
    for i in range(1, max_number + 1):
        last_number = i * i
    return last_number

def main():
    with concurrent.futures.ProcessPoolExecutor(max_workers=len(max_numbers)) as executor:
        try:
            for future in concurrent.futures.as_completed(executor.map(run_loop, max_numbers, timeout=1), timeout=1):
                print(future.result(timeout=1))
        except concurrent.futures._base.TimeoutError:
            print("This took to long...")

if __name__ == '__main__':
    main()

Answer 1

据我所知，TimeoutError 实际上是在您期望的时候引发的，而不是在任务完成之后引发的。

但是，您的程序本身将继续运行，直到所有正在运行的任务完成。这是因为当前正在执行的任务（在您的情况下，可能是您提交的所有任务，因为您的池大小等于任务数）实际上并未“杀死”。

引发 TimeoutError，因此您可以选择不等到任务完成（而是执行其他操作），但任务将继续运行直到完成。只要你的Executor的线程/子进程中有未完成的任务，python就不会退出。

据我所知，仅仅“停止”当前正在执行的 Futures 是不可能的，您只能“取消”尚未启动的计划任务。在您的情况下，不会有任何，但假设您有 5 个线程/进程池，并且您想要处理 100 个项目。在某个时候，可能有 20 个已完成的任务、5 个正在运行的任务和 75 个计划的任务。在这种情况下，您可以取消这 76 个计划任务，但无论您是否等待结果，正在运行的 4 个任务将继续执行直到完成。

即使不能那样做，我想应该有办法达到你想要的最终结果。也许这个版本可以帮助你（不确定它是否完全符合你的要求，但它可能会有一些用处）：

import concurrent.futures
import time
import datetime

max_numbers = [10000000, 10000000, 10000000, 10000000, 10000000]

class Task:
    def __init__(self, max_number):
        self.max_number = max_number
        self.interrupt_requested = False

    def __call__(self):
        print("Started:", datetime.datetime.now(), self.max_number)
        last_number = 0;
        for i in xrange(1, self.max_number + 1):
            if self.interrupt_requested:
                print("Interrupted at", i)
                break
            last_number = i * i
        print("Reached the end")
        return last_number

    def interrupt(self):
        self.interrupt_requested = True

def main():
    with concurrent.futures.ThreadPoolExecutor(max_workers=len(max_numbers)) as executor:
        tasks = [Task(num) for num in max_numbers]
        for task, future in [(i, executor.submit(i)) for i in tasks]:
            try:
                print(future.result(timeout=1))
            except concurrent.futures.TimeoutError:
                print("this took too long...")
                task.interrupt()


if __name__ == '__main__':
    main()

通过为每个“任务”创建一个可调用对象，并将其提供给执行程序而不是简单的函数，您可以提供一种“中断”任务的方法。 提示：删除task.interrupt() 行，看看会发生什么，这可能会让我更容易理解我上面的冗长解释；-)

Answer 2

最近我也遇到了这个问题，最后我使用ProcessPoolExecutor提出了以下解决方案：

---

def main():
    with concurrent.futures.ProcessPoolExecutor(max_workers=len(max_numbers)) as executor:
        try:
            for future in concurrent.futures.as_completed(executor.map(run_loop, max_numbers, timeout=1), timeout=1):
                print(future.result(timeout=1))
        except concurrent.futures._base.TimeoutError:
            print("This took to long...")
            stop_process_pool(executor)

def stop_process_pool(executor):
    for pid, process in executor._processes.items():
        process.terminate()
    executor.shutdown()