任何实际有效的并发期货超时

Question

尝试以便宜的价格编写基于进程的超时（同步），如下所示：

from concurrent.futures import ProcessPoolExecutor
def call_with_timeout(func, *args, timeout=3):
    with ProcessPoolExecutor(max_workers=1) as pool:
        future = pool.submit(func, *args)
        result = future.result(timeout=timeout)

但似乎传递给future.result的timeout论点并没有像宣传的那样真正起作用。

>>> t0 = time.time()
... call_with_timeout(time.sleep, 2, timeout=3)
... delta = time.time() - t0
... print('wall time:', delta)
wall time: 2.016767978668213

还行。

>>> t0 = time.time()
... call_with_timeout(time.sleep, 5, timeout=3)
... delta = time.time() - t0
... print('wall time:', delta)
# TimeoutError

不正常 - 5 秒后解锁，而不是 3 秒。

相关问题显示了如何使用线程池或信号来执行此操作。如何在 n 秒后将提交到池的进程超时，而不使用任何多处理_private API？硬杀就可以了，无需请求干净关机。

Answer 1

你可能想看看pebble .

它的ProcessPool旨在解决这个确切的问题：启用超时和取消正在运行的任务，而无需关闭整个池。

当将来超时或被取消时，工作线程实际上被终止，有效地停止了计划函数的执行。

超时：

pool = pebble.ProcessPool(max_workers=1)
future = pool.schedule(func, args=args, timeout=1)
try:
    future.result()
except TimeoutError:
    print("Timeout")

例：

def call_with_timeout(func, *args, timeout=3):
    pool = pebble.ProcessPool(max_workers=1)
    with pool:
        future = pool.schedule(func, args=args, timeout=timeout)
        return future.result()

Answer 2

超时的行为

正常。 future.result(timeout=timeout)在给定超时后停止。关闭池仍会等待所有待处理的期货完成执行，这会导致意外延迟。

您可以通过调用 shutdown(wait=False) 在后台关闭，但整个 Python 程序不会结束，直到所有待处理的期货完成执行：

def call_with_timeout(func, *args, timeout=3):
    pool = ProcessPoolExecutor(max_workers=1)
    try:
        future = pool.submit(func, *args)
        result = future.result(timeout=timeout)
    finally:
        pool.shutdown(wait=False)

执行程序 API 无法取消已在执行的调用。 future.cancel()只能取消尚未开始的呼叫。如果你想要突然中止的功能，你可能应该使用除concurrent.futures.ProcessPoolExecutor以外的其他东西。