Signal Handling and Process Lifecycle Management

Updated

September 7, 2026

Signal Handling and Process Lifecycle Management

After reading this chapter, you will master POSIX signal management using Python’s signal module, design graceful shutdown controllers for Kubernetes and Docker containers (SIGTERM and SIGINT), implement dynamic configuration reloads without process restarts via SIGHUP, apply execution watchdog alarms (SIGALRM), and resolve thread-affinity constraints.

Mental model

Operating systems communicate lifecycle events to running processes using signals (asynchronous software interrupts). When an orchestrator (such as systemd, Docker, or Kubernetes) stops a container, it sends SIGTERM to request a graceful shutdown, giving the process a grace period (e.g. 30 seconds) before escalating to uncatchable SIGKILL.

In CPython, the runtime intercepts POSIX signals in the C runtime layer, sets an internal flag, and executes your Python handler on the main thread between bytecode evaluation instructions:

Kernel / Orchestrator (Docker / K8s / systemd)
  │
  ▼ Sends POSIX Signal (e.g. SIGTERM / SIGINT / SIGHUP)
┌────────────────────────────────────────────────────────┐
│ CPython Runtime (C Signal Trampoline)                  │
│   Sets pending signal flag                             │
└────────────────────────┬───────────────────────────────┘
                         │
                         ▼ Checked at bytecode loop boundary (ceval.c)
┌────────────────────────────────────────────────────────┐
│ Python Main Thread Signal Handler                      │
│   Executes registered callback: handler(signum, frame) │
│                                                        │
│   Action:                                              │
│   1. Set is_shutting_down = True                       │
│   2. Finish current transaction / drain in-flight jobs │
│   3. Flush logs and close database connection pools    │
│   4. Exit cleanly with sys.exit(0)                     │
└────────────────────────────────────────────────────────┘

Minimal example

Save as graceful_shutdown.py:

# graceful_shutdown.py
import signal
import sys
import time
from types import FrameType

class GracefulWorker:
    def __init__(self) -> None:
        self.running = True
        # Register signal handlers for both Ctrl+C (SIGINT) and container termination (SIGTERM)
        signal.signal(signal.SIGINT, self._handle_exit_signal)
        signal.signal(signal.SIGTERM, self._handle_exit_signal)

    def _handle_exit_signal(self, signum: int, frame: FrameType | None) -> None:
        sig_name = signal.Signals(signum).name
        print(f"\n[SIGNAL] Received {sig_name} ({signum}). Initiating graceful termination...")
        self.running = False

    def run(self, max_cycles: int = 4) -> None:
        print(f"Worker PID: {sys.argv[0]} active. Running up to {max_cycles} cycles.")
        for cycle in range(1, max_cycles + 1):
            if not self.running:
                break
            print(f"  [Cycle #{cycle}] Processing in-flight transactions...")
            time.sleep(0.05)

        print("[CLEANUP] Closing connection pools and flushing buffers.")
        print("[EXIT] Process shutdown completed cleanly.")

def main() -> None:
    worker = GracefulWorker()
    worker.run()

if __name__ == "__main__":
    main()

Run via uv run python graceful_shutdown.py:

Worker PID: graceful_shutdown.py active. Running up to 4 cycles.
  [Cycle #1] Processing in-flight transactions...
  [Cycle #2] Processing in-flight transactions...
  [Cycle #3] Processing in-flight transactions...
  [Cycle #4] Processing in-flight transactions...
[CLEANUP] Closing connection pools and flushing buffers.
[EXIT] Process shutdown completed cleanly.

Worked examples

Case 1: Container Termination Controller with In-Flight Drain

In production container environments, terminating pods must stop accepting new traffic while completing tasks already underway:

# container_drain.py
import signal
import sys
import time
from types import FrameType

class TaskQueueConsumer:
    def __init__(self) -> None:
        self.shutdown_requested = False
        self.active_jobs = 0
        signal.signal(signal.SIGTERM, self._on_sigterm)

    def _on_sigterm(self, signum: int, frame: FrameType | None) -> None:
        print("\n[SIGTERM] Orchestrator requested pod eviction.")
        print("  * Halting queue consumption (no new jobs).")
        self.shutdown_requested = True

    def process_queue(self, queue: list[str]) -> None:
        while queue:
            if self.shutdown_requested:
                print(f"[DRAIN] Abandoning {len(queue)} pending jobs for sibling pods.")
                break

            job = queue.pop(0)
            self.active_jobs += 1
            print(f"  [Executing] Processing job '{job}'...")
            time.sleep(0.02)
            self.active_jobs -= 1

        print(f"[STATUS] Active jobs remaining: {self.active_jobs}. Ready to exit.")

def main() -> None:
    consumer = TaskQueueConsumer()
    # Simulate pending work queue
    work = ["job-101", "job-102", "job-103", "job-104"]
    consumer.process_queue(work)

if __name__ == "__main__":
    main()

Run:

uv run python container_drain.py

Output:

  [Executing] Processing job 'job-101'...
  [Executing] Processing job 'job-102'...
  [Executing] Processing job 'job-103'...
  [Executing] Processing job 'job-104'...
[STATUS] Active jobs remaining: 0. Ready to exit.

Case 2: Zero-Downtime Configuration Reload via `SIGHUP`

UNIX daemons (like NGINX and PostgreSQL) listen for SIGHUP (Hangup Signal) to reload configuration files from disk without terminating client connections:

# config_reloader.py
import signal
import sys
import time
from types import FrameType

class HotReloadService:
    def __init__(self) -> None:
        self.config: dict[str, str | int] = {"rate_limit": 100, "log_level": "INFO"}
        # Register SIGHUP for dynamic configuration reload
        if hasattr(signal, "SIGHUP"):
            signal.signal(signal.SIGHUP, self._on_sighup)

    def _on_sighup(self, signum: int, frame: FrameType | None) -> None:
        print("\n[SIGHUP] Received reload signal! Re-reading configuration from disk...")
        # Simulate loading new configuration
        self.config["rate_limit"] = 500
        self.config["log_level"] = "DEBUG"
        print(f"  [RELOADED] Updated config in-flight: {self.config}")

    def run_tick(self) -> None:
        print(f"Service running with: rate_limit={self.config['rate_limit']}, level={self.config['log_level']}")

def main() -> None:
    service = HotReloadService()
    service.run_tick()

    # Simulate programmatic SIGHUP delivery to our own process
    if hasattr(signal, "SIGHUP"):
        signal.raise_signal(signal.SIGHUP)

    service.run_tick()

if __name__ == "__main__":
    main()

Run:

uv run python config_reloader.py

Output:

Service running with: rate_limit=100, level=INFO

[SIGHUP] Received reload signal! Re-reading configuration from disk...
  [RELOADED] Updated config in-flight: {'rate_limit': 500, 'log_level': 'DEBUG'}
Service running with: rate_limit=500, level=DEBUG

Case 3: Watchdog Execution Timers with `SIGALRM`

When calling legacy C libraries, blocking filesystem calls, or operations that do not accept a timeout argument, signal.alarm() schedules a SIGALRM after a specified number of seconds:

# watchdog_timer.py
import signal
import time
from types import FrameType

class ExecutionTimeoutError(TimeoutError):
    """Raised when an operation exceeds the scheduled SIGALRM threshold."""

def alarm_handler(signum: int, frame: FrameType | None) -> None:
    raise ExecutionTimeoutError("Operation timed out: watchdog timer expired.")

def perform_bounded_operation(duration_seconds: float, timeout_seconds: int) -> str:
    # Set handler and start countdown
    old_handler = signal.signal(signal.SIGALRM, alarm_handler)
    signal.alarm(timeout_seconds)

    try:
        time.sleep(duration_seconds)
        return "SUCCESS"
    finally:
        # ALWAYS cancel alarm in finally block (0 seconds disables the alarm)
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)

def main() -> None:
    if not hasattr(signal, "SIGALRM"):
        print("SIGALRM is only available on UNIX/POSIX platforms.")
        return

    # 1. Operation within threshold
    print("Testing operation completing within timeout (0.01s < 1s):")
    status = perform_bounded_operation(duration_seconds=0.01, timeout_seconds=1)
    print(f"  Result: {status}")

    # 2. Operation exceeding threshold
    print("\nTesting operation exceeding timeout (2s > 1s):")
    try:
        perform_bounded_operation(duration_seconds=2.0, timeout_seconds=1)
    except ExecutionTimeoutError as err:
        print(f"  Caught expected timeout: {err}")

if __name__ == "__main__":
    main()

Run:

uv run python watchdog_timer.py

Output:

Testing operation completing within timeout (0.01s < 1s):
  Result: SUCCESS

Testing operation exceeding timeout (2s > 1s):
  Caught expected timeout: Operation timed out: watchdog timer expired.

Pitfalls

Pitfall 1: Registering Signals Outside the Main Thread

Python requires that all signal handlers be registered and executed exclusively on the main thread. Attempting to call signal.signal() from a worker thread raises ValueError:

import threading
import signal

def worker_thread():
    try:
        signal.signal(signal.SIGINT, lambda s, f: None)
    except ValueError as err:
        print(f"Thread error: {err}")

t = threading.Thread(target=worker_thread)
t.start()
t.join()

Output:

Thread error: signal only works in main thread of the main interpreter

Pitfall 2: Deadlocks from Non-Reentrant Code in Signal Handlers

A signal handler can interrupt execution at any arbitrary bytecode instruction, including while a lock is currently acquired. If the handler attempts to acquire the same lock, the process permanently deadlocks:

# THE TRAP: Deadlock!
import threading
lock = threading.Lock()

def unsafe_handler(signum, frame):
    lock.acquire()  # DEADLOCK if main thread was interrupted while holding 'lock'!
    # do cleanup...
    lock.release()

# THE FIX:
# In signal handlers, only set simple atomic flags (booleans, events)
# and let the main execution loop inspect the flag and release locks cleanly.
shutdown_event = threading.Event()
def safe_handler(signum, frame):
    shutdown_event.set()

Pitfall 3: PID 1 in Docker Containers Ignoring Signals

When a Python script runs as PID 1 inside a Docker container (e.g. CMD ["python", "app.py"]), the Linux kernel treats PID 1 specially: default signal handlers are disabled. If your script does not explicitly register a handler for SIGTERM, docker stop will hang for 10 seconds before forcibly terminating with SIGKILL!

# THE FIX: Always register an explicit handler for SIGTERM in container entrypoints:
signal.signal(signal.SIGTERM, lambda s, f: sys.exit(0))

Exercises

Write a script that tracks how many times SIGINT (Ctrl+C) is pressed. On the first press, print a warning; on the second press within 3 seconds, terminate the application.
Implement a context manager time_limit(seconds) using signal.alarm() that raises a custom TimeoutException if the code block takes longer than seconds.
Create a daemon simulator that catches SIGHUP to toggle a debug logging boolean flag while continuing to loop.
Demonstrate how signal.pause() suspends the current process until any POSIX signal is received.
Write a script that sends SIGUSR1 to another Python process using os.kill(target_pid, signal.SIGUSR1) to trigger a diagnostic memory dump.

Signal Handling and Process Lifecycle Management

Mental model

Minimal example

Worked examples

Case 1: Container Termination Controller with In-Flight Drain

Case 2: Zero-Downtime Configuration Reload via SIGHUP

Case 3: Watchdog Execution Timers with SIGALRM

Pitfalls

Pitfall 1: Registering Signals Outside the Main Thread

Pitfall 2: Deadlocks from Non-Reentrant Code in Signal Handlers

Pitfall 3: PID 1 in Docker Containers Ignoring Signals

Exercises

Further reading

Case 2: Zero-Downtime Configuration Reload via `SIGHUP`

Case 3: Watchdog Execution Timers with `SIGALRM`