C裤时装表演:Python超时装饰器

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

@timeout(10)
def main_func():
    nested_func()
    while True:
        continue
@timeout(5)
def nested_func():
   print "finished doing nothing"

= & gt;这样做的结果将完全没有超时。我们将陷入无限循环。
但是,如果我从nested_func中删除@timeout注释,则会出现超时错误。
由于某种原因,我们不能在函数和嵌套函数上同时使用装饰器,任何想法为什么以及如何纠正它的工作,假设包含函数超时总是大于嵌套超时。

7

这是signal模块的计时功能的限制,您链接的装饰器使用它。

signal.alarm(time)

如果 time 为非零,则此函数请求在time秒内向进程发送SIGALRM信号。取消先前调度的任何警报(在任何时间只能调度一个警报)。则返回值是先前设置的任何警报发出前的秒数。如果time为零,则当前未调度任何警报。

所以,你看到的是,当你的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
1

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

您可以使用此装饰器,它支持嵌套装饰器: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')
0

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

假设您没有使用 UNIX。是装饰器的代码段,向您显示了可以根据需要使用的参数列表。

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

对于非 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

本站系公益性非盈利分享网址,本文来自用户投稿,不代表边看边学立场,如若转载,请注明出处

(542)
Vmware安装程序无法继续:安装 vmware工具
上一篇
Python用户注册:无法阻止用户在 Python上注册重复的用户名
下一篇

相关推荐

发表评论

登录 后才能评论

评论列表(77条)