Last active
February 27, 2024 17:17
-
-
Save a-recknagel/6f688cb18e8b7c9fe79975a4a2e0e305 to your computer and use it in GitHub Desktop.
generic timeout construct
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 contextlib | |
import os | |
from threading import Thread | |
from time import sleep | |
from datetime import datetime | |
class CustomInterrupt(Exception): | |
SIG = 63 | |
def handle_custom_interrupt(signum, _): | |
raise CustomInterrupt | |
signal.signal(CustomInterrupt.SIG, handle_custom_interrupt) | |
@contextlib.contextmanager | |
def timeout(seconds: float = 3.0): | |
class TimedOut: | |
def __init__(self, seconds_: float): | |
self.seconds: float = seconds_ | |
self.state: bool | None = None | |
self.thread = Thread(target=self._task) | |
def _task(self): | |
sleep(self.seconds) | |
if self.state is None: | |
os.kill(os.getpid(), CustomInterrupt.SIG) | |
def start(self): | |
self.thread.start() | |
def __bool__(self): | |
if self.state is None: | |
raise ValueError( | |
"Can't decide yet, value only becomes meaningful on exit.") | |
return self.state | |
timed_out = TimedOut(seconds) | |
try: | |
timed_out.start() | |
yield timed_out | |
timed_out.state = False | |
except CustomInterrupt: | |
timed_out.state = True | |
finally: | |
timed_out.thread.join() # tbh, no idea why I'm doing this | |
start = datetime.now() | |
print(1, (datetime.now()-start).seconds) | |
with timeout(1) as timed_out: | |
while True: | |
pass | |
print(2, (datetime.now()-start).seconds) | |
with timeout(1) as timed_out: | |
sleep(5) | |
print(3, (datetime.now()-start).seconds) | |
# output: | |
# 1 0 | |
# 2 1 | |
# 3 2 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment