Use multiprocessing to timeout a Python function Use one process to keep time/check timeout and another process to call this Python function. from multiprocessing import Processdef inc_forever(): print('Starting function inc_forever()...') print(next(counter))def return_zero():
terminate() function will terminate foo function. p. join() is used to continue execution of main thread. If you run the above script, it will run for 10 seconds and terminate after that.
Applying timeout function using thread in PythonThe first thread is to execute the function. 2. The second thread is to measure the time taken by the function. The only second thread should whether the time is over or not.
The process for timing out an operations is described in the documentation for signal.
The basic idea is to use signal handlers to set an alarm for some time interval and raise an exception once that timer expires.
Note that this will only work on UNIX.
Here's an implementation that creates a decorator (save the following code as timeout.py
).
import errno
import os
import signal
import functools
class TimeoutError(Exception):
pass
def timeout(seconds=10, error_message=os.strerror(errno.ETIME)):
def decorator(func):
def _handle_timeout(signum, frame):
raise TimeoutError(error_message)
@functools.wraps(func)
def wrapper(*args, **kwargs):
signal.signal(signal.SIGALRM, _handle_timeout)
signal.alarm(seconds)
try:
result = func(*args, **kwargs)
finally:
signal.alarm(0)
return result
return wrapper
return decorator
This creates a decorator called @timeout
that can be applied to any long running functions.
So, in your application code, you can use the decorator like so:
from timeout import timeout
# Timeout a long running function with the default expiry of 10 seconds.
@timeout
def long_running_function1():
...
# Timeout after 5 seconds
@timeout(5)
def long_running_function2():
...
# Timeout after 30 seconds, with the error "Connection timed out"
@timeout(30, os.strerror(errno.ETIMEDOUT))
def long_running_function3():
...
I rewrote David's answer using the with
statement, it allows you do do this:
with timeout(seconds=3):
time.sleep(4)
Which will raise a TimeoutError.
The code is still using signal
and thus UNIX only:
import signal
class timeout:
def __init__(self, seconds=1, error_message='Timeout'):
self.seconds = seconds
self.error_message = error_message
def handle_timeout(self, signum, frame):
raise TimeoutError(self.error_message)
def __enter__(self):
signal.signal(signal.SIGALRM, self.handle_timeout)
signal.alarm(self.seconds)
def __exit__(self, type, value, traceback):
signal.alarm(0)
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With