eventlet
import time
import eventlet
eventlet.monkey_patch()
with eventlet.Timeout(2,False):
print '这条语句正常执行'
time.sleep(4)
print '没有跳过这条输出'
print '跳过了输出'
问题: 1、针对子进程无法跳出
timeout_decorator
from tqdm import trange
import sys
import time
import timeout_decorator
@timeout_decorator.timeout(int(sys.argv[2]))
def test():
if sys.argv[1] == '--timeout':
for i in trange(3):
time.sleep(1)
print ('>>> {} seconds passed.'.format(i+1))
return 0
if __name__ == '__main__':
try:
test()
except Exception as e:
print ('Timeout Error Catched!')
print (e)
print ("Timeout Task Ended!")
问题: 1、报错signal only works in main thread 在默认情况下,timeout-decorator使用信号来限制给定函数的执行时间。
如果您的函数不在主线程中执行(例如,如果它是Web应用程序的工作线程),则此配置不起作用。在这种情况下使用多进程来作为超时替代策略。要使用它,只需将use_signals = False传递给超时装饰器函数:
import time
import timeout_decorator
@timeout_decorator.timeout(5, use_signals=False)
def mytest():
print "Start"
for i in range(1,10):
time.sleep(1)
print("{} seconds have passed".format(i))
if __name__ == '__main__':
mytest()
2、多线程使用会有问题,在多处理策略超时的情况下,您的函数不会返回无法被pickle的对象,否则它将在主进程和子进程之间进行编组时失败。
signal
import signal
class TimeOutException(Exception):
pass
def handle(signum, frame):
raise TimeOutException("运行超时!")
def set_timeout(timeout, callback):
def wrapper(func):
def inner(*args, **kwargs):
try:
signal.signal(signal.SIGALRM, handle)
signal.alarm(timeout)
rs = func(*args, **kwargs)
signal.alarm(0)
return rs
except TimeOutException as e:
callback()
return inner
return wrapper
def process_time():
print("超时了")
@set_timeout(40, process_time)
def function():
pass
超时信号只能设置在主线程函数,不能设置在子线程函数
|