wrapt-timeout-decorator 1.2.6

Timeout decorator

wrapt_timeout_decorator

there are many timeout decorators out there - that one focuses on correctness if using with Classes, methods,

class methods, static methods and so on, preserving also the traceback information for Pycharm debugging.

There is also a powerful eval function, it allows to read the desired timeout value even from Class attributes.

It is very flexible and can be used from python 2.7 to python 3.x, pypy, pypy3 and probably other dialects.

Since it is using multiprocess and dill, this decorator can be used on more sophisticated objects

when not using signals (under Windows for instance). In that case multiprocess and multiprocess.pipe is used

to communicate with the child process (instead of multiprocessing.queue) which is faster and might work on Amazon AWS.

100% code coverage, tested under Linux, OsX, Windows and Wine

---

Report Issues

Contribute

Pull Request

Code of Conduct

---

Try it in Jupyter Notebook

You might try it right away in Jupyter Notebook by using the “launch binder” badge, or click here

Installation and Upgrade

From source code:

# normal install
python setup.py install
# test without installing
python setup.py test

via pip latest Release:

# latest Release from pypi
pip install wrapt_timeout_decorator

via pip latest Development Version:

# upgrade all dependencies regardless of version number (PREFERRED)
pip install --upgrade https://github.com/bitranox/wrapt_timeout_decorator/archive/master.zip  --upgrade-strategy eager
# normal install
pip install --upgrade https://github.com/bitranox/wrapt_timeout_decorator/archive/master.zip
# test without installing
pip install --upgrade https://github.com/bitranox/wrapt_timeout_decorator/archive/master.zip --install-option test

via requirements.txt:

# Insert following line in Your requirements.txt:
# for the latest Release:
# wrapt_timeout_decorator
# for the latest Development Version :
# https://github.com/bitranox/wrapt_timeout_decorator/archive/master.zip

# to install and upgrade all modules mentioned in requirements.txt:
pip install --upgrade -r /<path>/requirements.txt

via python:

# latest Release:
python -m pip install wrapt_timeout_decorator

# latest Development Version:
python -m pip install --upgrade https://github.com/bitranox/wrapt_timeout_decorator/archive/master.zip

Basic Usage

import time
from wrapt_timeout_decorator import *

@timeout(5)
def mytest(message):
    print(message)
    for i in range(1,10):
        time.sleep(1)
        print('{} seconds have passed'.format(i))

if __name__ == '__main__':
    mytest('starting')

Specify an alternate exception to raise on timeout:

import time
from wrapt_timeout_decorator import *

@timeout(5, timeout_exception=StopIteration)
def mytest(message):
    print(message)
    for i in range(1,10):
        time.sleep(1)
        print('{} seconds have passed'.format(i))

if __name__ == '__main__':
    mytest('starting')

Parameters

@timeout(dec_timeout, use_signals, timeout_exception, exception_message, dec_allow_eval, dec_hard_timeout)
def decorated_function(*args, **kwargs):
    # interesting things happens here ...
    ...

"""
dec_timeout         the timeout period in seconds, or a string that can be evaluated when dec_allow_eval = True
                    type: float, integer or string
                    default: None (no Timeout set)
                    can be overridden by passing the kwarg dec_timeout to the decorated function*

use_signals         if to use signals (linux, osx) to realize the timeout. The most accurate and preferred method.
                    Please note that signals can be used only in the main thread and only on linux. In all other cases
                    (not the main thread, or under Windows) signals will not be used, no matter what You set here,
                    in that cases use_signals will be disabled automatically.
                    type: boolean
                    default: True
                    can be overridden by passing the kwarg use_signals to the decorated function*

timeout_exception   the Exception that will be raised if a timeout occurs.
                    type: exception
                    default: TimeoutError, on Python < 3.3: Assertion Error (since TimeoutError does not exist on that Python Versions)

exception_message   custom Exception message.
                    type: str
                    default : 'Function {function_name} timed out after {dec_timeout} seconds' (will be formatted)

dec_allow_eval      will allow to evaluate the parameter dec_timeout.
                    If enabled, the parameter of the function dec_timeout, or the parameter passed
                    by kwarg dec_timeout will be evaluated if its type is string. You can access :
                    wrapped (the decorated function object and all the exposed objects below)
                    instance    Example: 'instance.x' - see example above or doku
                    args        Example: 'args[0]' - the timeout is the first argument in args
                    kwargs      Example: 'kwargs["max_time"] * 2'
                    type: bool
                    default: false
                    can be overridden by passing the kwarg dec_allow_eval to the decorated function*

dec_hard_timeout    only relevant when signals can not be used. In that case a new process needs to be created.
                    The creation of the process on windows might take 0.5 seconds and more, depending on the size
                    of the main module and modules to be imported. Especially useful for small timeout periods.

                    dec_hard_timeout = True : the decorated function will time out after dec_timeout, no matter what -
                    that means if You set 0.1 seconds here, the subprocess can not be created in that time and the
                    function will always time out and never run.

                    dec_hard_timeout = False : the decorated function will time out after the called function
                    is allowed to run for dec_timeout seconds. The time needed to create that process is not considered.
                    That means if You set 0.1 seconds here, and the time to create the subprocess is 0.5 seconds,
                    the decorated function will time out after 0.6 seconds in total, allowing the decorated function to run
                    for 0.1 seconds.

                    type: bool
                    default: false
                    can be overridden by passing the kwarg dec_hard_timeout to the decorated function*

* that means the decorated_function must not use that kwarg itself, since this kwarg will be popped from the kwargs
"""

Multithreading

By default, timeout-decorator uses signals to limit the execution time of the given function. This approach does not work if your function is executed not in the main thread (for example if it’s a worker thread of the web application) or when the operating system does not support signals (aka Windows). There is an alternative timeout strategy for this case - by using multiprocessing. This is done automatically, so you dont need to set use_signals=False. You can force not to use signals on Linux by passing the parameter use_signals=False to the timeout decorator function for testing. If Your program should (also) run on Windows, I recommend to test under Windows, since Windows does not support forking (read more under Section use with Windows). The following Code will run on Linux but NOT on Windows :

import time
from wrapt_timeout_decorator import *

@timeout(5, use_signals=False)
def mytest(message):
    print(message)
    for i in range(1,10):
        time.sleep(1)
        print('{} seconds have passed'.format(i))

if __name__ == '__main__':
    mytest('starting')

Warning

Make sure that in case of multiprocessing strategy for timeout, your function does not return objects which cannot be pickled, otherwise it will fail at marshalling it between master and child processes. To cover more cases, we use multiprocess and dill instead of multiprocessing and pickle.

Since Signals will not work on Windows, it is disabled by default, whatever You set.

Override with kwargs

decorator parameters starting with dec_* and use_signals can be overridden by kwargs with the same name :

import time
from wrapt_timeout_decorator import *

@timeout(dec_timeout=5, use_signals=False)
def mytest(message):
    print(message)
    for i in range(1,10):
        time.sleep(1)
        print('{} seconds have passed'.format(i))

if __name__ == '__main__':
    mytest('starting',dec_timeout=12)   # override the decorators setting. The kwarg dec_timeout will be not
                                        # passed to the decorated function.

Using the decorator without actually decorating the function

import time
from wrapt_timeout_decorator import *

def mytest(message):
    print(message)
    for i in range(1,10):
        time.sleep(1)
        print('{} seconds have passed'.format(i))

if __name__ == '__main__':
    timeout(dec_timeout=5)(mytest)('starting')

Using allow_eval

This is very powerful, but can be also very dangerous if you accept strings to evaluate from UNTRUSTED input.

read: https://nedbatchelder.com/blog/201206/eval_really_is_dangerous.html

If enabled, the parameter of the function dec_timeout, or the parameter passed by kwarg dec_timeout will be evaluated if its type is string.

You can access :

wrapped (the function object)

instance Example: ‘instance.x’ - an attribute of the instance of the class instance

args Example: ‘args[0]’ - the timeout is the first argument in args

kwargs Example: ‘kwargs[“max_time”] * 2’

and of course all attributes You can think of - that makes it powerful but dangerous.

by default allow_eval is disabled - but You can enable it in order to cover some edge cases without

modifying the timeout decorator.

def class ClassTest4(object):
    def __init__(self,x):
        self.x=x

    @timeout('instance.x', dec_allow_eval=True)
    def test_method(self):
        print('swallow')

    @timeout(1)
    def foo3(self):
        print('parrot')

    @timeout(dec_timeout='args[0] + kwargs.pop("more_time",0)', dec_allow_eval=True)
    def foo4(self,base_delay):
        time.sleep(base_delay)
        print('knight')


if __name__ == '__main__':
    # or override via kwarg :
    my_foo = ClassTest4(3)
    my_foo.test_method(dec_timeout='instance.x * 2.5 +1')
    my_foo.foo3(dec_timeout='instance.x * 2.5 +1', dec_allow_eval=True)
    my_foo.foo4(1,more_time=3)  # this will time out in 4 seconds

Logging

when signals=False (on Windows), logging in the wrapped function can be tricky. Since a new process is created, we can not use the logger object of the main process. Further development is needed to connect to the main process logger via a socket or queue.

When the wrapped function is using logger=logging.getLogger(), a new Logger Object is created. Setting up that Logger can be tricky (File Logging from two Processes is not supported …) I think I will use a socket to implement that (SocketHandler and some Receiver Thread)

Until then, You need to set up Your own new logger in the decorated function, if logging is needed. Again - keep in mind that You can not write to the same logfile from different processes ! (although there are logging modules which can do that)

use with Windows

On Windows the main module is imported again (but with name != ‘main’) because Windows is trying to simulate a forking-like behavior on a system that doesn’t have forking. multiprocessing has no way to know that you didn’t do anything important in you main module, so the import is done “just in case” to create an environment similar to the one in your main process.

It is more a problem of Windows, because the Windows Operating System does neither support “fork”, nor “signals” You can find more information on that here:

https://stackoverflow.com/questions/45110287/workaround-for-using-name-main-in-python-multiprocessing

https://docs.python.org/2/library/multiprocessing.html#windows

under Windows classes and functions in the __main__ context can not be pickled, You need to put the decorated Classes and functions into another module. In general (especially for windows) , the main() program should not have anything but the main function, the real thing should happen in the modules. I am also used to put all settings or configurations in a different file - so all processes or threads can access them (and also to keep them in one place together, not to forget typing hints and name completion in Your favorite editor)

You can find more information on that here: https://stackoverflow.com/questions/45616584/serializing-an-object-in-main-with-pickle-or-dill

Please note that for some unknown reason, probably in multiprocess, Class methods can not be decorated at all under Windows with Python 2.7

Here an example that will work on Linux but wont work on Windows (the variable “name” and the function “sleep” wont be found in the spawned process :

main.py:

from time import sleep
from wrapt_timeout_decorator import *

name="my_var_name"


@timeout(5, use_signals=False)
def mytest():
    print("Start ", name)
    for i in range(1,10):
        sleep(1)
        print("{} seconds have passed".format(i))
    return i


if __name__ == '__main__':
    mytest()

here the same example, which will work on Windows:

# my_program_main.py:

from multiprocessing import freeze_support
import lib_test

def main():
    lib_test.mytest()


if __name__ == '__main__':
    freeze_support()
    main()

# conf_my_program.py:

class ConfMyProgram(object):
    def __init__(self):
        self.name:str = 'my_var_name'

conf_my_program = ConfMyProgram()

# lib_test.py:

from wrapt_timeout_decorator import *
from time import sleep
from conf_my_program import conf_my_program

@timeout(5, use_signals=False)
def mytest():
    print("Start ", conf_my_program.name)
    for i in range(1,10):
        sleep(1)
        print("{} seconds have passed".format(i))
    return i

convenience function to detect pickle errors

remember that decorated functions in Windows needs to be pickable. In order to detect pickle problems You can use :

from wrapt_timeout_decorator import *
# always remember that the "object_to_pickle" should not be defined within the main context
detect_unpickable_objects(object_to_pickle, dill_trace=True)  # type: (Any, bool) -> Dict

use_signals = False (Windows) gives different total time

when use_signals = False (this is the only method available on Windows), the timeout function is realized by starting another process and terminate that process after the given timeout. Under Linux fork() of a new process is very fast, under Windows it might take some considerable time, because the main context needs to be reloaded on spawn() since fork() is not available on Windows. Spawning of a small module might take something like 0.5 seconds and more.

Since it is not predictable how long the spawn() will take on windows, the timeout will start AFTER spawning the new process.

This means that the timeout given, is the time the process is allowed to run, excluding the time to setup the process itself. This is especially important if You use small timeout periods :

for Instance:

@timeout(0.1)
def test():
    time.sleep(0.2)

the total time to timeout on linux with use_signals = False will be around 0.1 seconds, but on windows this will take about 0.6 seconds. 0.5 seconds to set up the new process, and giving the function test() 0.1 seconds to run !

If You need that a decorated function should time out exactly after the given timeout, You can pass the parameter dec_hard_timeout=True. in this case the function will time out exactly after the given time, no matter how long it took to spawn the process itself. In that case, if You set up the time out too short, the process might never run and will always timeout.

Requirements

following Packets will be installed / needed :

dill, see : https://github.com/uqfoundation/dill

multiprocess, see: https://github.com/uqfoundation/multiprocess

wrapt, see : https://github.com/GrahamDumpleton/wrapt

pytest, see : https://github.com/pytest-dev/pytest

typing, see : https://pypi.org/project/typing/

Acknowledgement

Derived from

https://github.com/pnpnpn/timeout-decorator

http://www.saltycrane.com/blog/2010/04/using-python-timeout-decorator-uploading-s3/

and special thanks to “uncle bob” Robert C. Martin, especially for his books on “clean code” and “clean architecture”

Contribute

I would love for you to fork and send me pull request for this project. Please contribute.

Future Enhancements:

better logging for signals=false. Since a new process is created, we can not log to the logger of the main process. logger=logging.getLogger() will crate a new Logger in the wrapped function.

License

This software is licensed under the MIT license

See License file

Changelog

1.2.0

2019-04-09: initial PyPi release

1.1.0

2019-04-03: added pickle analyze convenience function

1.0.9

2019-03-27: added OsX and Windows tests, added parameter dec_hard_timeout for Windows, 100% Code Coverage

1.0.8

2019-02-26: complete refractoring and code cleaning

1.0.7

2019-02-25: fix pickle detection, added some tests, codecov now correctly combining the coverage of all tests

1.0.6

2019-02-24: fix pickle detection when use_signals = False, drop Python2.6 support since wrapt dropped it.

1.0.5

2018-09-13: use multiprocessing.pipe instead of queue If we are not able to use signals, we need to spawn a new process. This was done in the past by pickling the target function and put it on a queue - now this is done with a half-duplex pipe.

• it is faster

• it probably can work on Amazon AWS, since there You must not use queues

1.0.4

2017-12-02: automatic detection if we are in the main thread. Signals can only be used in the main thread. If the decorator is running in a subthread, we automatically disable signals.

1.0.3

2017-11-30: using dill and multiprocess to enhance windows functionality

1.0.0

2017-11-10: Initial public release