可嵌套的超时装饰器？（即超时装饰函数调用超时装饰函数）

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 调用了多个函数，这些函数花费的时间不确定，其中一些函数有自己的超时时间。

我尝试了 timeout-decorator and two SO answers here，但 none 有效。

Answer 1

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')

Answer 2

像这样：

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

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

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)

可嵌套的超时装饰器？（即超时装饰函数调用超时装饰函数）

Nestable timeout decorator? (i.e. timeout decorated func calling timeout decorated func)

python

timeout

decorator

可嵌套的超时装饰器？ （即超时装饰函数调用超时装饰函数）

Nestable timeout decorator? (i.e. timeout decorated func calling timeout decorated func)

python

timeout

decorator

可嵌套的超时装饰器？（即超时装饰函数调用超时装饰函数）