Maximum runtime for a function in python (bulb)

Question

I am trying to limit the time when a function is allowed to run in python (bulb). It will look something like this:

def my_function: try: long_function_time() catch TimeOutException: return "Function took too long to execute" def long_function_time: #stuff here 

At first I tried to use signals, but I was told that this is not a very good approach, since the bulb works in a streaming environment. I want the maximum runtime to be flexible, so I can easily change it.

The code I use (which sometimes does not work, I don’t know why):

 class TimedOutExc(Exception): pass def deadline(timeout, *args): def decorate(f): def handler(signum, frame): signal.alarm(0) raise TimedOutExc() def new_f(*args): signal.signal(signal.SIGALRM, handler) signal.alarm(timeout) return f(*args) new_f.__name__ = f.__name__ return new_f return decorate 

Thanks in advance!

Answer 1

This works with CPython 2 (tested since 2.7) and does not explicitly use signals, but the CPython interpreter loop. Thus, you still have to process non-Python blocks in I / O calls and similar (for example, you must set socket.timeout ). In addition, it has possible expensive runtimes (you can force it to check the timeout for every * n-th call to make it a little easier).

However, it helps for a specific class of problems (such as computing). If you want it to coexist with a profiler, you will have to invest some work.

 import sys import time class WatchdogTimeoutError(RuntimeError): """Raised in case of runtime limit violations.""" def sleeper(tick): """Endless loop.""" while True: time.sleep(tick) def watchdog(timeout, code, *args, **kwargs): "Time-limited execution." def tracer(frame, event, arg, start=time.time()): "Helper." now = time.time() if now > start + timeout: raise WatchdogTimeoutError(start, now) return tracer if event == "call" else None old_tracer = sys.gettrace() try: sys.settrace(tracer) code(*args, **kwargs) finally: sys.settrace(old_tracer) def demo(): """Show timeout executor.""" try: watchdog(5, sleeper, 0.1) except WatchdogTimeoutError, exc: start, abort = exc.args print "Aborted after %.3f secs" % (abort - start,) else: print "Ended" if __name__ == "__main__": demo() 

Answer 2

Try using streams:

 from multiprocessing import Pool, TimeoutError from time import sleep class TimedOutExc(Exception): pass def f(x): sleep(2) return x*x pool = Pool(processes=1) result = pool.apply_async(f, (1,)) try: print(result.get(timeout=1)) except TimeoutError: print 'timeout' raise TimedOutExc()