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

Question

How to set a limit on the maximum amount of time a function can run for? For example, using time.sleep as a placeholder function, how to limit the amount of time time.sleep can run to a maximum of 5 mins (300 seconds)?

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

That is, how can time.sleep(600) above be limited and interrupted after 300 seconds?

Answer 1

On POSIX you have a simple and clean solution available in signal module.

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

Caveats:

• Does not work on all platforms, e.g. Windows.
• Does not work in threaded applications, only the main thread.

For other readers where the caveats above are a deal breaker, you will need a more heavyweight approach. The best option is usually to run the code in a separate process (or possibly a thread), and terminate that process if it takes too long. See multiprocessing module for example.

Answer 2

One of currently probably preferred options to accomplish what you want is usage of Pythons

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

module ( see tutorial )

Just copy/paste the code example below and run it. Does it what you are after?

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    

it outputs:

--- 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.

Another known to me option is using a

decorator function and the signal module

Checkout the web page with origin of the code I have provided here (only one small adjustment was necessary to make it run on 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()")

The code above outputs:

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