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.
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
|
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
Don't forget to import sys. Oops, there should be an import sys at the top.
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: