如何限制函数可以运行的时间量(添加超时)

Question

如何设置函数可以运行的最长时间的限制？例如，使用 time.sleep 作为占位符函数，如何将time.sleep可以运行的时间限制为最多 5 分钟(300 秒(？

import time
try:
    # As noted above `time.sleep` is a placeholder for a function 
    # which takes 10 minutes to complete.
    time.sleep(600)
except:
    print('took too long')

也就是说，如何在 300 秒后限制和中断上述time.sleep(600)？

Answer 1

在 POSIX 上，您可以在signal模块中提供简单干净的解决方案。

import signal
import time
class Timeout(Exception):
    pass
def handler(sig, frame):
    raise Timeout
signal.signal(signal.SIGALRM, handler)  # register interest in SIGALRM events
signal.alarm(2)  # timeout in 2 seconds
try:
    time.sleep(60)
except Timeout:
    print('took too long')

警告：

• 不适用于所有平台，例如Windows。
在线程
• 应用程序中不起作用，只在主线程中工作。

对于其他读者来说，上述警告是一个交易破坏者，你将需要一个更重量级的方法。 最好的选择通常是在单独的进程(或可能是线程(中运行代码，如果时间过长，则终止该进程。 例如，请参阅multiprocessing模块。

Answer 2

目前可能的首选选项之一是使用 Python。

多处理(特别是它的proc.join(timeoutTime(方法(

模块 ( 见教程 (

只需复制/粘贴下面的代码示例并运行它。这是你所追求的吗？

def beBusyFor(noOfSeconds):
    import time
    print("    beBusyFor() message: going to rest for", noOfSeconds, "seconds")
    time.sleep(noOfSeconds)
    print("    beBusyFor() message: was resting", noOfSeconds, "seconds, now AWAKE")
import multiprocessing
noOfSecondsBusy = 5; timeoutTime  = 3
print("--- noOfSecondsBusy = 5; timeoutTime  = 3 ---")
proc = multiprocessing.Process(target=beBusyFor, args=(noOfSecondsBusy, ))
print("Start beBusyFor()")
proc.start()
print("beBusyFor() is running")
proc.join(timeoutTime)
if proc.is_alive():
    print(timeoutTime, "seconds passed, beBusyFor() still running, terminate()" )
    proc.terminate()
else:
    print("OK, beBusyFor() has finished its work in time.")
#:if    
print()
noOfSecondsBusy = 2; timeoutTime  = 3
print("--- noOfSecondsBusy = 2; timeoutTime  = 3 ---")
proc = multiprocessing.Process(target=beBusyFor, args=(noOfSecondsBusy, ))
print("Start beBusyFor()")
proc.start()
print("beBusyFor() started")
proc.join(timeoutTime)
if proc.is_alive():
    print(timeoutTime, "seconds passed, beBusyFor() still running, terminate()" )
    proc.terminate()
else:
    print("OK, beBusyFor() has finished its work in time.")
#:if    

它输出：

--- noOfSecondsBusy = 5; timeoutTime  = 3 ---
Start beBusyFor()
beBusyFor() is running
    beBusyFor() message: going to rest for 5 seconds
3 seconds passed, beBusyFor() still running, terminate()
--- noOfSecondsBusy = 2; timeoutTime  = 3 ---
Start beBusyFor()
beBusyFor() started
    beBusyFor() message: going to rest for 2 seconds
    beBusyFor() message: was resting 2 seconds, now AWAKE
OK, beBusyFor() has finished its work in time.

我知道的另一个选项是使用

装饰器功能和信号模块

查看我在这里提供的代码来源的网页(只需进行一个小的调整即可使其在Python 3.6上运行(：

import signal
class TimeoutError(Exception):
    def __init__(self, value = "Timed Out"):
        self.value = value
    def __str__(self):
        return repr(self.value)
def timeout(seconds_before_timeout):
    def decorate(f):
        def handler(signum, frame):
            raise TimeoutError()
        def new_f(*args, **kwargs):
            old = signal.signal(signal.SIGALRM, handler)
            signal.alarm(seconds_before_timeout)
            try:
                result = f(*args, **kwargs)
            finally:
                signal.signal(signal.SIGALRM, old)
            signal.alarm(0)
            return result
        # new_f.func_name = f.func_name
        new_f.__name__ = f.__name__
        return new_f
    return decorate
# Try it out:
import time
@timeout(5)
def mytest():
    print( "mytest() message:  Started" )
    for i in range(1,10):
        time.sleep(1)
        print( "mytest() message:  %d seconds have passed" % i )
try:
    mytest()
except TimeoutError as e:
    print("stopped executing mytest() because it", e)
print("continuing script execution past call of mytest()")

上面的代码输出：

mytest() message:  Started
mytest() message:  1 seconds have passed
mytest() message:  2 seconds have passed
mytest() message:  3 seconds have passed
mytest() message:  4 seconds have passed
stopped executing mytest() because it 'Timed Out'
continuing script execution past call of mytest()