如何在python中超时函数,超时小于一秒



问题的规范:

我正在搜索日志文件的大量行,并将这些行分发到组中,以便使用re.match()函数存储的正则表达式(RegExses)。不幸的是,我的一些正则表达式太复杂了,Python有时会让自己回溯地狱。因此,我需要通过某种超时来保护它。

问题:

  • re.match,我正在使用,是Python的功能,正如我在StackOverflow上的某个地方发现的那样(我真的很抱歉,我现在找不到链接:-( )。运行 Python 库时中断线程是非常困难的。因此,线程不在游戏中。
  • 因为re.match函数的评估需要相对较短的时间,我想用这个函数分析大量的行,我需要一些不会花费太长时间执行的超时函数(这使得线程更不合适,初始化新线程需要很长时间)并且可以设置为小于一秒><。出于这些原因,此处的答案 - 函数调用超时在这里 - 如果装饰器需要太长时间才能完成超时功能(警报 - 1 秒及以上)不在桌面上。

我今天早上一直在寻找这个问题的解决方案,但我没有找到任何令人满意的答案。

解决方案:

我刚刚修改了这里发布的脚本:如果完成时间太长,则超时功能。

这是代码:

from functools import wraps
import errno
import os
import signal
class TimeoutError(Exception):
    pass
def timeout(seconds=10, error_message=os.strerror(errno.ETIME)):
    def decorator(func):
        def _handle_timeout(signum, frame):
            raise TimeoutError(error_message)
        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, _handle_timeout)
            signal.setitimer(signal.ITIMER_REAL,seconds) #used timer instead of alarm
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)
            return result
        return wraps(func)(wrapper)
    return decorator

然后你可以像这样使用它:

from timeout import timeout 
from time import time
@timeout(0.01)
def loop():
    while True:
       pass
try:
    begin = time.time()
    loop()
except TimeoutError, e:
    print "Time elapsed: {:.3f}s".format(time.time() - begin)

哪些打印

Time elapsed: 0.010s

这就是我们如何定义超时slow_function函数,但即使在引发异常后也不会停止slow_function:

import threading, time
class TimeoutError(Exception):
    pass
def slow_function():
    time.sleep(1000)
    return "Done"
def run_with_timeout(func, timeout):
    def target():
        nonlocal result
        result = func()
    result = None
    thread = threading.Thread(target=target)
    thread.start()
    thread.join(timeout)
    if thread.is_alive():
        raise TimeoutError("Function timed out")
    return result
try:
    result = run_with_timeout(slow_function, 3)
except TimeoutError:
    print("Function timed out")
else:
    print("Function returned:", result)

最新更新