Skip to content

Instantly share code, notes, and snippets.

@a-recknagel
Last active February 27, 2024 17:17
Show Gist options
  • Save a-recknagel/6f688cb18e8b7c9fe79975a4a2e0e305 to your computer and use it in GitHub Desktop.
Save a-recknagel/6f688cb18e8b7c9fe79975a4a2e0e305 to your computer and use it in GitHub Desktop.
generic timeout construct
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