ActiveState Code

Recipe 483752: timelimit: tell a function to time out after a time limit

Have a function that you want to put a time limit on? Just use this decorator like so:

@timelimit(10) def myfunction(...): ...

will limit myfunction to 10 seconds, raising TimeoutError if it takes longer.

Python 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import threading
class TimeoutError(Exception): pass

def timelimit(timeout):
    def internal(function):
        def internal2(*args, **kw):
            class Calculator(threading.Thread):
                def __init__(self):
                    threading.Thread.__init__(self)
                    self.result = None
                    self.error = None
                
                def run(self):
                    try:
                        self.result = function(*args, **kw)
                    except:
                        self.error = sys.exc_info()[0]
            
            c = Calculator()
            c.start()
            c.join(timeout)
            if c.isAlive():
                raise TimeoutError
            if c.error:
                raise c.error
            return c.result
        return internal2
    return internal

Discussion

This is better than other implementations using signal.alarm because alarms don't work inside threads and aren't supported on Windows. Some of the code was inspired by http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/473878

Comments

  1. 1. At 7:04 p.m. on 10 apr 2006, Aaron Swartz (the author) said:

    Don't forget to import sys. Oops, there should be an import sys at the top.

  2. 2. At 7:49 a.m. on 11 apr 2006, Andrew Bennetts said:

    Two obvious problems. Problem 1: this doesn't actually stop the function after the timeout. It leaves it running (in a seperate thread), still consuming resources.

    Problem 2: this isn't reliable. It can't interrupt functions that don't release the GIL. In fact, it can't interrupt functions that never return to the ceval loop. Here's an example:

    @timelimit(3)
def unstoppable():
    import itertools
    print list(itertools.ifilter(None, itertools.repeat(0)))

unstoppable()

Sign in to comment