def quit_function(fn_name):
# print to stderr, unbuffered in Python 2.
print('{0} took too long'.format(fn_name), file=sys.stderr)
sys.stderr.flush() # Python 3 stderr is likely buffered.
thread.interrupt_main() # raises KeyboardInterrupt
这里是装饰师自己:
def exit_after(s):
'''
use as decorator to exit process if
function takes longer than s seconds
'''
def outer(fn):
def inner(*args, **kwargs):
timer = threading.Timer(s, quit_function, args=[fn.__name__])
timer.start()
try:
result = fn(*args, **kwargs)
finally:
timer.cancel()
return result
return inner
return outer
用法
下面的用法可以直接回答你关于5秒后退出的问题!以下内容:
@exit_after(5)
def countdown(n):
print('countdown started', flush=True)
for i in range(n, -1, -1):
print(i, end=', ', flush=True)
sleep(1)
print('countdown finished')
演示:
>>> countdown(3)
countdown started
3, 2, 1, 0, countdown finished
>>> countdown(10)
countdown started
10, 9, 8, 7, 6, countdown took too long
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<stdin>", line 11, in inner
File "<stdin>", line 6, in countdown
KeyboardInterrupt
第二个函数调用将不会完成,相反,进程应该退出并进行回溯!
KeyboardInterrupt并不总是阻止睡眠线程
请注意,在Windows的Python2上,睡眠并不总是被键盘中断所中断,例如:
@exit_after(1)
def sleep10():
sleep(10)
print('slept 10 seconds')
>>> sleep10()
sleep10 took too long # Note that it hangs here about 9 more seconds
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<stdin>", line 11, in inner
File "<stdin>", line 3, in sleep10
KeyboardInterrupt
import multiprocessing
import time
# bar
def bar():
for i in range(100):
print "Tick"
time.sleep(1)
if __name__ == '__main__':
# Start bar as a process
p = multiprocessing.Process(target=bar)
p.start()
# Wait for 10 seconds or until process finishes
p.join(10)
# If thread is still active
if p.is_alive():
print "running... let's kill it..."
# Terminate
p.terminate()
p.join()
In [1]: import signal
# Register an handler for the timeout
In [2]: def handler(signum, frame):
...: print("Forever is over!")
...: raise Exception("end of time")
...:
# This function *may* run for an indetermined time...
In [3]: def loop_forever():
...: import time
...: while 1:
...: print("sec")
...: time.sleep(1)
...:
...:
# Register the signal function handler
In [4]: signal.signal(signal.SIGALRM, handler)
Out[4]: 0
# Define a timeout for your function
In [5]: signal.alarm(10)
Out[5]: 0
In [6]: try:
...: loop_forever()
...: except Exception, exc:
...: print(exc)
....:
sec
sec
sec
sec
sec
sec
sec
sec
Forever is over!
end of time
# Cancel the timer if the function returned before timeout
# (ok, mine won't but yours maybe will :)
In [7]: signal.alarm(0)
Out[7]: 0
我发布了一个gist,它用一个decorator和一个
threading.Timer
来解决这个问题。这是一个故障。导入和设置兼容性
它是用Python 2和3测试的。它还应该在Unix/Linux和Windows下工作。
首先是进口。无论Python版本如何,这些都试图保持代码的一致性:
使用独立于版本的代码:
现在我们已经从标准库中导入了我们的功能。
exit_after
装饰器接下来,我们需要一个函数来终止子线程中的
main()
:这里是装饰师自己:
用法
下面的用法可以直接回答你关于5秒后退出的问题!以下内容:
演示:
第二个函数调用将不会完成,相反,进程应该退出并进行回溯!
KeyboardInterrupt
并不总是阻止睡眠线程请注意,在Windows的Python2上,睡眠并不总是被键盘中断所中断,例如:
除非显式检查
PyErr_CheckSignals()
,否则也不太可能中断扩展中运行的代码,请参见Cython, Python and KeyboardInterrupt ignored在任何情况下,我都不会让线程睡眠超过一秒——这是处理器时间的一个eon。
要捕捉它并执行其他操作,可以捕捉键盘中断。
您可以使用
multiprocessing.Process
来实现这一点。代码
如果在UNIX上运行,则可以使用signal包:
调用
alarm.alarm(10)
10秒后,将调用处理程序。这会引发一个异常,您可以从常规Python代码中截取它。这个模块不能很好地处理线程(但是,谁呢?)
注意由于在发生超时时引发异常,因此它可能会在函数内部被捕获并被忽略,例如某个这样的函数:
相关问题 更多 >
编程相关推荐