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

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和两个 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)