Timeout decorator
Project description
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
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')
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://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.8
2019-04-23: import multiprocess as multiprocess, not as multiprocessing - that might brake other packages
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
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for wrapt_timeout_decorator-1.2.8.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | f3e9e67ad35230e966c8b96c224231cd00808cf7370eba2b8a79b9a07eeb12f6 |
|
MD5 | e93795806e2b0d117c346dab28d5a0a5 |
|
BLAKE2b-256 | 1122f6b5dda47441cb532c62b1a2fd3b026c8bc43878b9b279830cfd6a6c3e82 |
Hashes for wrapt_timeout_decorator-1.2.8-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 23cf6b03c73f308aaa6332a03897ad6dcb4727b7cc7f7730072ae5bf8a31372a |
|
MD5 | 746300f4d8a3f1ea82b02547c2982597 |
|
BLAKE2b-256 | a1e2701f964729d6c7744f65680dd31389734f5c3e4f54d3bc20e955d229bcdc |