Spaces:
Sleeping
Sleeping
File size: 1,884 Bytes
f81adc4 | 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 76 77 | from __future__ import annotations
"""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.
"""
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)
|