我正在调用Python中的一个函数,我知道它可能会停止并迫使我重新启动脚本.
如何调用该函数或我将其包装成什么,以便如果它花费的时间超过5秒,脚本会取消它并执行其他操作？
如果在UNIX上运行,则可以使用信号包:
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
调用10秒后,调用alarm.alarm(10)处理程序.这引发了一个例外,您可以从常规Python代码中截取.
这个模块不适合线程(但那么,谁呢？)
请注意,由于我们在超时发生时引发异常,因此它可能最终被捕获并在函数内被忽略,例如一个这样的函数:
def loop_forever():
while 1:
print 'sec'
try:
time.sleep(10)
except:
continue
您可以使用multiprocessing.Process这样做.
码
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()
如何调用该函数或将其包装在哪里,以便如果脚本取消时间超过5秒？
我发布了一个要点,用装饰器来解决这个问题/问题threading.Timer.这是故障.
它使用Python 2和3进行了测试.它也应该在Unix/Linux和Windows下运行.
首先是进口.无论Python版本如何,这些都试图保持代码一致:
from __future__ import print_function
import sys
import threading
from time import sleep
try:
import thread
except ImportError:
import _thread as thread
使用版本无关的代码:
try:
range, _print = xrange, print
def print(*args, **kwargs):
flush = kwargs.pop('flush', False)
_print(*args, **kwargs)
if flush:
kwargs.get('file', sys.stdout).flush()
except NameError:
pass
现在我们从标准库中导入了我们的功能.
exit_after 装饰接下来我们需要一个函数来终止main()子线程:
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 "", line 1, in File " ", line 11, in inner File " ", line 6, in countdown KeyboardInterrupt
第二个函数调用不会完成,而是进程应该以traceback退出!
KeyboardInterrupt 并不总是停止睡眠线程请注意,Windows上的Python 2上的键盘中断并不总是会中断睡眠,例如:
@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 "", line 1, in
File "", line 11, in inner
File "", line 3, in sleep10
KeyboardInterrupt
除非明确检查PyErr_CheckSignals(),否则它是否可能会中断在扩展中运行的代码,请参阅 Cython,Python和KeyboardInterrupt忽略
在任何情况下,我都会避免睡一个线程超过一秒 - 这是处理器时间的一个因素.
如何调用该函数或我将其包装成什么,以便如果它花费的时间超过5秒,脚本会取消它并执行其他操作？
要捕获它并执行其他操作,您可以捕获KeyboardInterrupt.
>>> try:
... countdown(10)
... except KeyboardInterrupt:
... print('do something else')
...
countdown started
10, 9, 8, 7, 6, countdown took too long
do something else
我有一个不同的提议,它是一个纯函数(使用与线程建议相同的API)并且似乎工作正常(基于此线程的建议)
def timeout(func, args=(), kwargs={}, timeout_duration=1, default=None):
import signal
class TimeoutError(Exception):
pass
def handler(signum, frame):
raise TimeoutError()
# set the timeout handler
signal.signal(signal.SIGALRM, handler)
signal.alarm(timeout_duration)
try:
result = func(*args, **kwargs)
except TimeoutError as exc:
result = default
finally:
signal.alarm(0)
return result
在单元测试中搜索超时调用时,我遇到了这个线程.我没有在答案或第三方包中找到任何简单的内容,所以我在下面编写了装饰器,你可以直接进入代码:
import multiprocessing.pool
import functools
def timeout(max_timeout):
"""Timeout decorator, parameter in seconds."""
def timeout_decorator(item):
"""Wrap the original function."""
@functools.wraps(item)
def func_wrapper(*args, **kwargs):
"""Closure for function."""
pool = multiprocessing.pool.ThreadPool(processes=1)
async_result = pool.apply_async(item, args, kwargs)
# raises a TimeoutError if execution exceeds max_timeout
return async_result.get(max_timeout)
return func_wrapper
return timeout_decorator
然后就这么简单来超时测试或任何你喜欢的功能:
@timeout(5.0) # if execution takes longer than 5 seconds, raise a TimeoutError
def test_base_regression(self):
...
有很多建议,但没有使用concurrent.futures,我认为这是最清晰的处理方式.
from concurrent.futures import ProcessPoolExecutor
# Warning: this does not terminate function if timeout
def timeout_five(fnc, *args, **kwargs):
with ProcessPoolExecutor() as p:
f = p.submit(fnc, *args, **kwargs)
return f.result(timeout=5)
超级简单的阅读和维护.
我们创建一个池,提交一个进程,然后等待最多5秒钟,然后提出一个TimeoutError,你可以捕获并处理你需要的东西.
原生于python 3.2+并向后移植到2.7(pip install期货).
线程和进程之间的切换是在更换为简单ProcessPoolExecutor用ThreadPoolExecutor.
如果你想在超时时终止进程,我建议你看看Pebble.
stopit在pypi上找到的软件包似乎很好地处理了超时.
我喜欢@stopit.threading_timeoutable装饰器,它timeout为装饰函数添加一个参数,它可以满足你的期望,它会停止这个功能.
在pypi上查看:https://pypi.python.org/pypi/stopit
出色，易于使用且可靠的PyPi项目超时装饰器（https://pypi.org/project/timeout-decorator/）
安装方式：
pip install timeout-decorator
用法：
import time
import timeout_decorator
@timeout_decorator.timeout(5)
def mytest():
print "Start"
for i in range(1,10):
time.sleep(1)
print "%d seconds have passed" % i
if __name__ == '__main__':
mytest()
京公网安备 11010802040832号 | 京ICP备19059560号-6 