Python 超时装饰器

Question

我正在使用here提到的代码解决方案。
我是装饰器的新手，如果我想写如下内容，我不明白为什么这个解决方案不起作用：

@timeout(10)
def main_func():
    nested_func()
    while True:
        continue

@timeout(5)
def nested_func():
   print "finished doing nothing"

=> 这样的结果根本不会超时。我们将陷入无限循环。
但是，如果我从 nested_func 中删除 @timeout 注释，则会出现超时错误。
出于某种原因，我们不能同时在函数和嵌套函数上使用装饰器，知道为什么以及如何纠正它以使其正常工作，假设包含函数超时始终必须大于嵌套超时。

Answer 1

这是signal 模块计时功能的限制，您链接的装饰器使用该功能。这是相关的piece of the documentation（重点由我添加）：

signal.alarm(time)

如果时间不为零，则此函数请求在time 秒内将SIGALRM 信号发送到进程。 任何以前安排的警报都被取消（任何时候只能安排一个警报）。返回的值是在任何以前设置的警报被传递之前的秒数。如果time 为零，则不安排警报，并取消任何已安排的警报。如果返回值为零，则当前没有安排警报。 （参见 Unix 手册页警报 (2)。）可用性：Unix。

所以，您看到的是，当您的 nested_func 被调用时，它的计时器取消了外部函数的计时器。

您可以更新装饰器以注意alarm 调用的返回值（这将是上一个警报（如果有）到期之前的时间）。正确获取细节有点复杂，因为内部计时器需要跟踪其函数运行了多长时间，因此它可以修改前一个计时器的剩余时间。这是一个未经测试的装饰器版本，我认为它基本上是正确的（但我不完全确定它在所有异常情况下都能正常工作）：

import time
import signal

class TimeoutError(Exception):
    def __init__(self, value = "Timed Out"):
        self.value = value
    def __str__(self):
        return repr(self.value)

def timeout(seconds_before_timeout):
    def decorate(f):
        def handler(signum, frame):
            raise TimeoutError()
        def new_f(*args, **kwargs):
            old = signal.signal(signal.SIGALRM, handler)
            old_time_left = signal.alarm(seconds_before_timeout)
            if 0 < old_time_left < second_before_timeout: # never lengthen existing timer
                signal.alarm(old_time_left)
            start_time = time.time()
            try:
                result = f(*args, **kwargs)
            finally:
                if old_time_left > 0: # deduct f's run time from the saved timer
                    old_time_left -= time.time() - start_time
                signal.signal(signal.SIGALRM, old)
                signal.alarm(old_time_left)
            return result
        new_f.func_name = f.func_name
        return new_f
    return decorate

Answer 2

目前 Python 的 PyPI 库中有一个更好的 timeout decorator 版本。它支持基于 UNIX 和非 UNIX 的操作系统。提到信号的部分 - 专门用于 UNIX。

假设您没有使用 UNIX。 下面是来自装饰器的代码 sn-p，它显示了您可以根据需要使用的参数列表。

def timeout(seconds=None, use_signals=True, timeout_exception=TimeoutError, exception_message=None)

用于在 NON-UNIX 基本操作系统上实施。这就是我会做的：

import time
import timeout_decorator

@timeout_decorator.timeout(10, use_signals=False)
def main_func():
    nested_func()
    while True:
        continue

@timeout_decorator.timeout(5, use_signals=False)
def nested_func():
    print "finished doing nothing"

如果你注意到了，我在做 use_signals=False。就是这样，你应该很高兴。

Answer 3

正如 Blckknght 指出的那样，您不能将信号用于嵌套装饰器 - 但您可以使用多处理来实现这一点。

你可以使用这个装饰器，它支持嵌套装饰器：https://github.com/bitranox/wrapt_timeout_decorator

正如 ABADGER1999 在他的博客中指出的那样 https://anonbadger.wordpress.com/2018/12/15/python-signal-handlers-and-exceptions/ 使用信号和 TimeoutException 可能不是最好的主意 - 因为它可以被装饰函数捕获。

当然，您可以使用自己的异常，从基本异常类派生，但代码可能仍然无法按预期工作 - 看下一个例子 - 你可以在 jupyter 中尝试一下：https://mybinder.org/v2/gh/bitranox/wrapt_timeout_decorator/master?filepath=jupyter_test_wrapt_timeout_decorator.ipynb

import time
from wrapt_timeout_decorator import *

# caveats when using signals - the TimeoutError raised by the signal may be caught
# inside the decorated function.
# So You might use Your own Exception, derived from the base Exception Class.
# In Python-3.7.1 stdlib there are over 300 pieces of code that will catch your timeout
# if you were to base an exception on Exception. If you base your exception on BaseException,
# there are still 231 places that can potentially catch your exception.
# You should use use_signals=False if You want to make sure that the timeout is handled correctly !
# therefore the default value for use_signals = False on this decorator !

@timeout(5, use_signals=True)
def mytest(message):
    try:
        print(message)
        for i in range(1,10):
            time.sleep(1)
            print('{} seconds have passed - lets assume we read a big file here'.format(i))
    # TimeoutError is a Subclass of OSError - therefore it is caught here !
    except OSError:
        for i in range(1,10):
            time.sleep(1)
            print('Whats going on here ? - Ooops the Timeout Exception is catched by the OSError ! {}'.format(i))
    except Exception:
        # even worse !
        pass
    except:
        # the worst - and exists more then 300x in actual Python 3.7 stdlib Code !
        # so You never really can rely that You catch the TimeoutError when using Signals !
        pass


if __name__ == '__main__':
    try:
        mytest('starting')
        print('no Timeout Occured')
    except TimeoutError():
        # this will never be printed because the decorated function catches implicitly the TimeoutError !
        print('Timeout Occured')