Timeout on a function call Timeout on a function call python python

Timeout on a function call


You may use the signal package if you are running on UNIX:

In [1]: import signal# Register an handler for the timeoutIn [2]: def handler(signum, frame):   ...:     print("Forever is over!")   ...:     raise Exception("end of time")   ...: # This function *may* run for an indetermined time...In [3]: def loop_forever():   ...:     import time   ...:     while 1:   ...:         print("sec")   ...:         time.sleep(1)   ...:            ...:         # Register the signal function handlerIn [4]: signal.signal(signal.SIGALRM, handler)Out[4]: 0# Define a timeout for your functionIn [5]: signal.alarm(10)Out[5]: 0In [6]: try:   ...:     loop_forever()   ...: except Exception, exc:    ...:     print(exc)   ....: secsecsecsecsecsecsecsecForever is over!end of time# Cancel the timer if the function returned before timeout# (ok, mine won't but yours maybe will :)In [7]: signal.alarm(0)Out[7]: 0

10 seconds after the call signal.alarm(10), the handler is called. This raises an exception that you can intercept from the regular Python code.

This module doesn't play well with threads (but then, who does?)

Note that since we raise an exception when timeout happens, it may end up caught and ignored inside the function, for example of one such function:

def loop_forever():    while 1:        print('sec')        try:            time.sleep(10)        except:            continue


You can use multiprocessing.Process to do exactly that.

Code

import multiprocessingimport time# bardef bar():    for i in range(100):        print "Tick"        time.sleep(1)if __name__ == '__main__':    # Start bar as a process    p = multiprocessing.Process(target=bar)    p.start()    # Wait for 10 seconds or until process finishes    p.join(10)    # If thread is still active    if p.is_alive():        print "running... let's kill it..."        # Terminate - may not work if process is stuck for good        p.terminate()        # OR Kill - will work for sure, no chance for process to finish nicely however        # p.kill()        p.join()


How do I call the function or what do I wrap it in so that if it takes longer than 5 seconds the script cancels it?

I posted a gist that solves this question/problem with a decorator and a threading.Timer. Here it is with a breakdown.

Imports and setups for compatibility

It was tested with Python 2 and 3. It should also work under Unix/Linux and Windows.

First the imports. These attempt to keep the code consistent regardless of the Python version:

from __future__ import print_functionimport sysimport threadingfrom time import sleeptry:    import threadexcept ImportError:    import _thread as thread

Use version independent code:

try:    range, _print = xrange, print    def print(*args, **kwargs):         flush = kwargs.pop('flush', False)        _print(*args, **kwargs)        if flush:            kwargs.get('file', sys.stdout).flush()            except NameError:    pass

Now we have imported our functionality from the standard library.

exit_after decorator

Next we need a function to terminate the main() from the child thread:

def quit_function(fn_name):    # print to stderr, unbuffered in Python 2.    print('{0} took too long'.format(fn_name), file=sys.stderr)    sys.stderr.flush() # Python 3 stderr is likely buffered.    thread.interrupt_main() # raises KeyboardInterrupt

And here is the decorator itself:

def exit_after(s):    '''    use as decorator to exit process if     function takes longer than s seconds    '''    def outer(fn):        def inner(*args, **kwargs):            timer = threading.Timer(s, quit_function, args=[fn.__name__])            timer.start()            try:                result = fn(*args, **kwargs)            finally:                timer.cancel()            return result        return inner    return outer

Usage

And here's the usage that directly answers your question about exiting after 5 seconds!:

@exit_after(5)def countdown(n):    print('countdown started', flush=True)    for i in range(n, -1, -1):        print(i, end=', ', flush=True)        sleep(1)    print('countdown finished')

Demo:

>>> countdown(3)countdown started3, 2, 1, 0, countdown finished>>> countdown(10)countdown started10, 9, 8, 7, 6, countdown took too longTraceback (most recent call last):  File "<stdin>", line 1, in <module>  File "<stdin>", line 11, in inner  File "<stdin>", line 6, in countdownKeyboardInterrupt

The second function call will not finish, instead the process should exit with a traceback!

KeyboardInterrupt does not always stop a sleeping thread

Note that sleep will not always be interrupted by a keyboard interrupt, on Python 2 on Windows, e.g.:

@exit_after(1)def sleep10():    sleep(10)    print('slept 10 seconds')>>> sleep10()sleep10 took too long         # Note that it hangs here about 9 more secondsTraceback (most recent call last):  File "<stdin>", line 1, in <module>  File "<stdin>", line 11, in inner  File "<stdin>", line 3, in sleep10KeyboardInterrupt

nor is it likely to interrupt code running in extensions unless it explicitly checks for PyErr_CheckSignals(), see Cython, Python and KeyboardInterrupt ignored

I would avoid sleeping a thread more than a second, in any case - that's an eon in processor time.

How do I call the function or what do I wrap it in so that if it takes longer than 5 seconds the script cancels it and does something else?

To catch it and do something else, you can catch the KeyboardInterrupt.

>>> try:...     countdown(10)... except KeyboardInterrupt:...     print('do something else')... countdown started10, 9, 8, 7, 6, countdown took too longdo something else


matomo