我在 for 循环中运行一个函数,如下所示:
for element in my_list:
my_function(element)
出于某种原因,某些元素可能会导致函数处理时间很长(甚至可能是一些我无法真正追踪它来自何处的无限循环)。因此,如果当前元素的处理时间超过 2 秒,我想添加一些循环控制来跳过当前元素。如何才能做到这一点?
我在 for 循环中运行一个函数,如下所示:
for element in my_list:
my_function(element)
出于某种原因,某些元素可能会导致函数处理时间很长(甚至可能是一些我无法真正追踪它来自何处的无限循环)。因此,如果当前元素的处理时间超过 2 秒,我想添加一些循环控制来跳过当前元素。如何才能做到这一点?
我不鼓励最明显的答案 - 使用 signal.alarm() 和异步引发异常以跳出任务执行的警报信号处理程序。理论上它应该工作得很好,但实际上 cPython 解释器代码并不能保证处理程序在您想要的时间范围内执行。信号处理可能会延迟 x 个字节码指令,因此在显式取消警报后(在 try 块的上下文之外)仍可能引发异常。
我们经常遇到的一个问题是警报处理程序的异常会在可超时代码完成后引发。
由于没有太多可用的线程控制方式,我依靠进程控制来处理必须超时的任务。基本上,要点是将任务交给子进程并在任务花费太长时间时终止子进程。 multiprocessing.pool并没有那么复杂——所以我有一个自制的池来进行这种级别的控制。
像这样的东西:
import signal
import time
class Timeout(Exception):
pass
def try_one(func,t):
def timeout_handler(signum, frame):
raise Timeout()
old_handler = signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(t) # triger alarm in 3 seconds
try:
t1=time.clock()
func()
t2=time.clock()
except Timeout:
print('{} timed out after {} seconds'.format(func.__name__,t))
return None
finally:
signal.signal(signal.SIGALRM, old_handler)
signal.alarm(0)
return t2-t1
def troublesome():
while True:
pass
try_one(troublesome,2)
该函数troublsome
永远不会自行返回。如果使用try_one(troublesome,2)
成功2秒后超时。