问题的规范:
我正在搜索日志文件的大量行,并将这些行分发到组中,以便使用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)