Decorating a timeout function

Question

I've picked up the timeout-function below from ActiveState's recipes for Python2 and polished it for python3.4.

Is there any leaner, less clunky way to write it?

import signal, time
class TimedOutExc(Exception):
 def __init__(self, value = "Timed Out"):
 self.value = value
 def __str__(self):
 return repr(self.value)
def TimedOutFn(f, timeout, *args, **kwargs):
 def handler(signum, frame):
 raise TimedOutExc()
 old = signal.signal(signal.SIGALRM, handler)
 signal.alarm(timeout)
 try:
 result = f(*args, **kwargs)
 finally:
 signal.signal(signal.SIGALRM, old)
 signal.alarm(0)
 return result
def timed_out(timeout):
 def decorate(f):
 def handler(signum, frame):
 raise TimedOutExc()
 def new_f(*args, **kwargs):
 old = signal.signal(signal.SIGALRM, handler)
 signal.alarm(timeout)
 try:
 result = f(*args, **kwargs)
 finally:
 signal.signal(signal.SIGALRM, old)
 signal.alarm(0)
 return result
 new_f.__name__ = f.__name__
 return new_f
 return decorate
def fn_1(secs):
 time.sleep(secs)
 return "Finished"
@timed_out(4)
def fn_2(secs):
 time.sleep(secs)
 return "Finished"
@timed_out(2)
def fn_3(secs):
 time.sleep(secs)
 return "Finished"
@timed_out(2)
def fn_4(secs):
 try:
 time.sleep(secs)
 return "Finished"
 except TimedOutExc:
 print( "(Caught TimedOutExc, so cleaining up, and re-raising it) - ")
 raise TimedOutExc
if __name__ == '__main__':
 try:
 print( "fn_1 (sleep 2, timeout 4): ")
 print( TimedOutFn(fn_1, 4, 2) )
 except TimedOutExc:
 print( "took too long")
 try:
 print( "fn_2 (sleep 2, timeout 4): ")
 print( fn_2(2) )
 except TimedOutExc:
 print( "took too long")
 try:
 print( "fn_1 (sleep 4, timeout 2): ")
 print( TimedOutFn(fn_1, 2, 4) )
 except TimedOutExc:
 print( "took too long")
 try:
 print( "fn_3 (sleep 4, timeout 2): ")
 print( fn_3(4) )
 except TimedOutExc:
 print( "took too long")
 try:
 print( "fn_4 (sleep 4, timeout 2): ")
 print( fn_4(4) )
 except TimedOutExc:
 print( "took too long")

Answer 1

1. Review

1. There are no docstrings. What do these functions do? How am I supposed to use them?

2. The use of the alarm mechanism means that the timeout occurs asynchronously with respect to the timed out function, and so data structures may be left in an inconsistent state. It's important to warn the user about this: it's hard to write functions that are safe against timeouts of this kind.

3. The functions and classes could have better names: I suggest TimedOutExc → TimedOut (it's clear that it's an exception because of its superclass); TimedOutFn → call_with_timeout (following PEP8); timed_out → with_timeout.

4. There's no need to give the exception class its own __init__ and __str__ method: the Exception class already has these methods.

5. The implementation of the timed_out decorator is almost identical to TimedOutFn. It would be better for new_f to call TimedOutFn instead of repeating the code.

6. The decorator copies __name__ from the original function, but what about __doc__ and __module__ and __qualname__? It would be better to use the built-in functools.wraps.

7. I think it's slightly better for call_with_timeout to take the timeout argument first. That way the function and its arguments are adjacent.

8. In call_with_timeout, there are two state changes that need to be reverted: the updating of the signal handler, and the setting of the alarm. But only one of these is protected by a try: ... finally: ....

9. There's a lot of duplicated code in the test cases. This could be refactored into methods.

10. The test cases don't actually tell you whether the tests passed or failed (you have to know what the output is supposed to be in each case). It would be a good idea to use the facilities in the unittest module.

2. Revised code

import functools
import signal
class TimedOut(Exception):
 pass
def call_with_timeout(timeout, f, *args, **kwargs):
 """Call f with the given arguments, but if timeout seconds pass before
 f returns, raise TimedOut. The exception is raised asynchronously,
 so data structures being updated by f may be in an inconsistent state.
 """
 def handler(signum, frame):
 raise TimedOut("Timed out after {} seconds.".format(timeout))
 old = signal.signal(signal.SIGALRM, handler)
 try:
 signal.alarm(timeout)
 try:
 return f(*args, **kwargs)
 finally:
 signal.alarm(0)
 finally:
 signal.signal(signal.SIGALRM, old)
def with_timeout(timeout):
 """Decorator for a function that causes it to timeout after the given
 number of seconds.
 """
 def decorator(f):
 @functools.wraps(f)
 def wrapped(*args, **kwargs):
 return call_with_timeout(timeout, f, *args, **kwargs)
 return wrapped
 return decorator