1
\$\begingroup\$

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")
\$\endgroup\$
2
\$\begingroup\$

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 TimedOutExcTimedOut (it's clear that it's an exception because of its superclass); TimedOutFncall_with_timeout (following PEP8); timed_outwith_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
\$\endgroup\$
0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.