我正在尝试为特定功能实施超时。我检查了SE中的许多问题,找不到适合我问题的解决方案,因为:
我已经为特定任务开发了已经设计好的自定义模块(例如MyModule
),并且其中定义了功能。由于外部因素,其中一个功能(例如MyFunc
)有可能永远运行,而我只是不希望python脚本挂起。
我打算添加超时功能,如下面的伪代码所述:
import MyModule
set_timeout(T)
MyResult=MyModule.MyFunc()
#Come to this part of script after execution of MyFunc() or after T seconds (the latter on priority)
if Timeout occurred:
print 'MyFunc did not execute completely'
else:
print 'MyFunc completed'
但是我不确定哪个模块可以在python上实现。请注意,我是新手,并且我编写的所有脚本都直接基于SE Answers或Python文档。
我认为解决此问题的一种好方法是创建一个装饰器并使用Thread.join(timeout)方法。请记住,没有杀死线程的好方法,因此只要您的程序正在运行,它就会或多或少地继续在后台运行。
首先,创建一个这样的装饰器:
from threading import Thread
import functools
def timeout(timeout):
def deco(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
res = [Exception('function [%s] timeout [%s seconds] exceeded!' % (func.__name__, timeout))]
def newFunc():
try:
res[0] = func(*args, **kwargs)
except Exception, e:
res[0] = e
t = Thread(target=newFunc)
t.daemon = True
try:
t.start()
t.join(timeout)
except Exception, je:
print 'error starting thread'
raise je
ret = res[0]
if isinstance(ret, BaseException):
raise ret
return ret
return wrapper
return deco
然后,执行以下操作:
func = timeout(timeout=16)(MyModule.MyFunc)
try:
func()
except:
pass #handle errors here
您可以在需要的任何地方使用此装饰器,例如:
@timeout(60)
def f():
...
@ acushner的答案适用于python 3.5:
from threading import Thread
import functools
def timeout(seconds_before_timeout):
def deco(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
res = [Exception('function [%s] timeout [%s seconds] exceeded!' % (func.__name__, seconds_before_timeout))]
def newFunc():
try:
res[0] = func(*args, **kwargs)
except Exception as e:
res[0] = e
t = Thread(target=newFunc)
t.daemon = True
try:
t.start()
t.join(seconds_before_timeout)
except Exception as e:
print('error starting thread')
raise e
ret = res[0]
if isinstance(ret, BaseException):
raise ret
return ret
return wrapper
return deco