File size: 1,882 Bytes
6172a47 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | """Track and clean up spawned CLI subprocesses.
This is a safety net for cases where the server is interrupted (Ctrl+C) and the
FastAPI lifespan cleanup doesn't run to completion. We only track processes we
spawn so we don't accidentally kill unrelated system processes.
"""
from __future__ import annotations
import atexit
import os
import subprocess
import threading
from loguru import logger
_lock = threading.Lock()
_pids: set[int] = set()
_atexit_registered = False
def ensure_atexit_registered() -> None:
global _atexit_registered
with _lock:
if _atexit_registered:
return
atexit.register(kill_all_best_effort)
_atexit_registered = True
def register_pid(pid: int) -> None:
if not pid:
return
ensure_atexit_registered()
with _lock:
_pids.add(int(pid))
def unregister_pid(pid: int) -> None:
if not pid:
return
with _lock:
_pids.discard(int(pid))
def kill_all_best_effort() -> None:
"""Kill any still-running registered pids (best-effort)."""
with _lock:
pids = list(_pids)
_pids.clear()
if not pids:
return
if os.name == "nt":
for pid in pids:
try:
# /T kills child processes, /F forces termination.
subprocess.run(
["taskkill", "/PID", str(pid), "/T", "/F"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
except Exception as e:
logger.debug("process_registry: taskkill failed pid=%s: %s", pid, e)
return
# Best-effort fallback for non-Windows.
for pid in pids:
try:
os.kill(pid, 9)
except Exception as e:
logger.debug("process_registry: kill failed pid=%s: %s", pid, e)
|