Created
May 2, 2023 15:34
-
-
Save colematt/4009bc1fe687fbe180db44abdeadb188 to your computer and use it in GitHub Desktop.
[Timeout a Long Running Function] #python3
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import signal | |
import time | |
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 __name__ == "__main__": | |
with timeout(seconds=3): | |
time.sleep(4) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import errno | |
import os | |
import signal | |
import functools | |
import time | |
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 | |
# Timeout a long running function with the default expiry of 10 seconds. | |
@timeout | |
def long_running_function1(secs:int) -> None: | |
time.sleep(secs) | |
# Timeout after 5 seconds | |
@timeout(5) | |
def long_running_function2(secs:int) -> None: | |
time.sleep(secs) | |
# Timeout after 30 seconds, with the error "Connection timed out" | |
@timeout(30, os.strerror(errno.ETIMEDOUT)) | |
def long_running_function3(secs:int) -> None: | |
time.sleep(secs) | |
if __name__ == "__main__": | |
long_running_function1(11) | |
long_running_function2(6) | |
long_running_function3(31) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment