内斯特布尔超时装饰器?(即超时装饰函数调用超时装饰函数)

Question

我需要一个装饰器（或功能上等效的东西），它允许下面的代码按预期工作：

@timeout(1)
def outer():
    inner()
@timeout(5)
def inner():
    time.sleep(3)
    print("Should never be printed if you call outer()")
outer()
# The outer timeout is ignored and "property" finishes

代码似乎毫无意义，但实际上，outer调用多个需要不确定时间的函数，其中一些函数有自己的超时。

我在这里尝试了超时装饰器和两个 SO 答案，但没有一个有效。

Answer 1

像这样：

def timeout(timeout, raise_exc=True):
    """
    raise_exc - if exception should be raised on timeout 
                or exception inside decorated func.
                Otherwise None will be returned.
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            res = None
            exc = None
            def _run():
                nonlocal res
                nonlocal exc
                try:
                    res = func(*args, **kwargs)
                except Exception as e:
                    exc = e
            t = threading.Thread(target=_run)
            t.daemon = True
            t.start()
            t.join(timeout=timeout)
            if raise_exc and t.is_alive():
                raise TimeoutError()
            elif raise_exc and (exc is not None):
                raise exc
            else:
                return res
        return wrapper
    return decorator

例子：

@timeout(0.5, raise_exc=False)
def outer():
    return inner()
@timeout(2)
def inner():
    time.sleep(1)
    return "Shouldn't be printed"
print(outer())  # None

和

@timeout(2, raise_exc=False)
def outer():
    return inner()
@timeout(2)
def inner():
    time.sleep(1)
    return "Should be printed"
print(outer())  # Should be printed

请注意，您的任务只能通过线程或进程来解决，但这可能会导致一些不明显的问题。我建议你考虑一下你的任务是否可以在没有它的情况下解决。在大多数情况下，您可以将代码拆分为多个部分，并在每个部分之后检查超时。像这样：

def outer(arg, timeout=None):
    t = Timeout(timeout)
    # some operation:
    time.sleep(1)
    if t.is_timeout: return None
    # use time left as subfunction's timeout:
    return inner(arg, timeout=t.time_left)

Answer 2

timeout函数使用 threading.Timer 设置计时器，thread.interrupt_main中断主线程。

from thread import interrupt_main
from threading import Timer
from time import time, sleep
def timeout(secs):
    def wrapper(func):
        timer = Timer(secs, interrupt_main)
        def decorated(*args, **kwargs):
            timer.start()
            return func(*args, **kwargs)
        return decorated
    return wrapper            
@timeout(1)
def outer():
    inner()
@timeout(5)
def inner():
    sleep(3)
    print("Should never be printed if you call outer()")
try:
    outer()
except:
    print('timed out')