如何限制函数可以运行的时间量（添加超时）？

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')

即上面的time.sleep(600)怎么能在300秒后被限制中断呢？

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')

注意事项：

• 不适用于所有平台，例如视窗。
• 在线程应用程序中不起作用，仅在主线程中起作用。

对于上面的警告会破坏交易的其他读者，您将需要一种更重量级的方法。最好的选择通常是在一个单独的进程（或者可能是一个线程）中运行代码，如果花费的时间太长，则终止该进程。例如，参见multiprocessing 模块。

Answer 2

目前可能首选的实现您想要的选项之一是使用 Python

多处理（尤其是它的 proc.join(timeoutTime) 方法）

模块（见tutorial）

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

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.

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

装饰器函数和信号模块

查看我在此处提供的web page with origin of the code（只需稍作调整即可使其在 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()