dlt.common.runtime.signals
signal_receiver
def signal_receiver(sig: int, frame: FrameType) -> None
Handle POSIX signals with two-stage escalation.
This handler is installed by delayed_signals(). On the first occurrence of a supported signal (eg. SIGINT, SIGTERM) it requests a graceful shutdown by setting a process-wide flag and waking sleeping threads via exit_event. A second occurrence of the same signal escalates by delegating to the original handler or the system default, which typically results in an immediate process termination (eg. KeyboardInterrupt for SIGINT).
Arguments:
sig- Signal number (for example, signal.SIGINT or signal.SIGTERM).frame- The current stack frame when the signal was received.
Notes:
- The CPython runtime delivers signal handlers in the main thread only. Worker threads must cooperatively observe shutdown via raise_if_signalled() or the signal-aware sleep().
signal_received
def signal_received() -> bool
check if a signal was received
sleep
def sleep(sleep_seconds: float) -> None
A signal-aware version of sleep function. Will wake up if signal is received but will not raise exception.
wake_all
def wake_all() -> None
Wakes all threads sleeping on event
delayed_signals
@contextmanager
def delayed_signals() -> Iterator[None]
Will delay signalling until raise_if_signalled is explicitly used or when
a second signal with the same int value arrives.
A no-op when not called on main thread.
Can be nested - nested calls are no-ops.