时间戳

首页 > 解决方案 > 在 python 中使用上下文管理器超时进程

python - 在 python 中使用上下文管理器超时进程

问题描述

我正在尝试编写一个上下文管理器来使进程超时，并且我有以下内容：

import threading
import _thread
import time
from contextlib import contextmanager

@contextmanager
def raise_timeout(timeout):
    timer = threading.Timer(timeout, _thread.interrupt_main)
    timer.start()

    try:
        yield
    except:
        print(f"timeout after {timeout} seconds")
        raise
    finally:
        timer.cancel()

with raise_timeout(1):
    time.sleep(5)

问题是当达到超时时，它仍然等待在time.sleep(5)引发异常之前完成。当我运行脚本时，我等待了 5 秒钟，然后在终端中获得以下输出：

> timeout after 1 seconds Traceback (most recent call last):   File
> "scratch1.py",
> line xx, in <module>
>     time.sleep(5) KeyboardInterrupt

但是如果我如下所示直接运行计时器，我会在 1 秒超时后立即出现异常。

timer = threading.Timer(1, _thread.interrupt_main)
timer.start()

我只是想知道 with 语句中的进程如何阻止异常直到它完成？

标签： pythonpython-multithreadingcontextmanager

解决方案

我认为您的Timer解决方案无法正常运行，因为time.sleep调用时不会被杀死_thread.interrupt_main()。

_thread文档对它的功能提出了建议：

线程与中断的交互很奇怪：KeyboardInterrupt异常将被任意线程接收。（当信号模块可用时，中断总是转到主线程。）

尽管Python 的错误跟踪器上的一个问题表明文档_thread.interrupt_main()不精确。

应该注意的是，替换一个长时间运行的进程而不是time.sleep您的上下文管理器可以按预期工作。

尝试：

with raise_timeout(1):
    for _ in range(1_000_000_000):
        pass

它会引发一个KeyboardInterrupt

如果可移植性不是问题，这里有一个解决方案signal，它只在 Unix 上可用，因为它使用SIGALARM.

import signal
import time
from contextlib import contextmanager

class TimeOutException(Exception):
    pass

@contextmanager
def raise_timeout(timeout):
    def _handler(signum, frame):
        raise TimeOutException()
    signal.signal(signal.SIGALRM, _handler)
    signal.alarm(timeout)

    try:
        yield
    except TimeOutException:
        print(f"Timeout after {timeout} seconds")
        raise
    finally:
        signal.alarm(0)

with raise_timeout(1):
    time.sleep(2)

这将引发TimeOutException

推荐阅读