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

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

注意事项：

并非适用于所有平台，例如Windows。
在线程应用程序中不起作用，仅在主线程中起作用。

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

Answer 2

目前可能首选的方法之一是使用 Pythons

multiprocessing ( especially its proc.join(timeoutTime) method )

模块（参见tutorial）

只是 copy/paste 下面的代码示例和运行它。它是你追求的吗？

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.

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

decorator function and the signal module

检查我在这里提供的 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()

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

How to limit the amount of time a function can run for (add a timeout)?

python

linux

timeout

scheduling

timer

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

How to limit the amount of time a function can run for (add a timeout)?

python

linux

timeout

scheduling

timer

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