fix(#4973): replace per-request diagnostics Timer with a process-global watchdog thread
A threading.Timer per RequestDiagnostics spawns one OS thread per request, held alive until finish() cancels it. Under sustained /api/sessions poll load this exhausts the per-process thread cap (RuntimeError: can't start new thread) and the server stops accepting connections. Replace with a single lazy-init watchdog daemon thread that scans pending diags every ~1s; public API unchanged. Caps timeout-tracking at 1 thread/process regardless of request rate.
This commit is contained in:
1 parent
6dd1c5e0fa
commit
2f7c5f01ee
2 files changed
+166
-8
No files matched your search
@@ -16,6 +16,73 @@ from typing import Any
|
|||||||
DEFAULT_SLOW_REQUEST_SECONDS = 5.0
|
DEFAULT_SLOW_REQUEST_SECONDS = 5.0
|
||||||
MAX_STACK_FRAMES_PER_THREAD = 40
|
MAX_STACK_FRAMES_PER_THREAD = 40
|
||||||
|
|
||||||
|
# Process-global watchdog: a single daemon thread scans all in-flight
|
||||||
|
# RequestDiagnostics instances instead of each request spawning its own
|
||||||
|
# threading.Timer. A Timer-per-request spawns one OS thread per request, held
|
||||||
|
# alive until finish() cancels it; under sustained /api/sessions poll load that
|
||||||
|
# exhausts the per-process thread cap ("RuntimeError: can't start new thread")
|
||||||
|
# and the server stops accepting connections (#4973). One watchdog thread per
|
||||||
|
# process caps the timeout-tracking cost regardless of request rate.
|
||||||
|
_WATCHDOG_TICK_SECONDS = 1.0
|
||||||
|
_watchdog_lock = threading.Lock()
|
||||||
|
_watchdog_pending: "dict[str, tuple[float, RequestDiagnostics]]" = {}
|
||||||
|
_watchdog_cv = threading.Condition(_watchdog_lock)
|
||||||
|
_watchdog_thread: "threading.Thread | None" = None
|
||||||
|
|
||||||
|
|
||||||
|
def _ensure_watchdog_running() -> None:
|
||||||
|
global _watchdog_thread
|
||||||
|
# Caller already holds _watchdog_lock.
|
||||||
|
if _watchdog_thread is not None and _watchdog_thread.is_alive():
|
||||||
|
return
|
||||||
|
t = threading.Thread(
|
||||||
|
target=_watchdog_loop,
|
||||||
|
name="request-diagnostics-watchdog",
|
||||||
|
daemon=True,
|
||||||
|
)
|
||||||
|
_watchdog_thread = t
|
||||||
|
t.start()
|
||||||
|
|
||||||
|
|
||||||
|
def _watchdog_loop() -> None:
|
||||||
|
# The loop body cannot raise while holding _watchdog_cv (the scan is pure
|
||||||
|
# dict ops; _on_timeout is fired outside the lock inside a try/except), so
|
||||||
|
# the watchdog thread never dies with entries pending. _ensure_watchdog_running
|
||||||
|
# therefore only needs to (re)spawn lazily on register, not supervise a crash.
|
||||||
|
while True:
|
||||||
|
fired: list[RequestDiagnostics] = []
|
||||||
|
with _watchdog_cv:
|
||||||
|
# Sleep until the next tick, but wake early if work arrives/changes.
|
||||||
|
_watchdog_cv.wait(_WATCHDOG_TICK_SECONDS)
|
||||||
|
if _watchdog_pending:
|
||||||
|
now = time.monotonic()
|
||||||
|
expired = [
|
||||||
|
rid for rid, (deadline, _diag) in _watchdog_pending.items()
|
||||||
|
if now >= deadline
|
||||||
|
]
|
||||||
|
for rid in expired:
|
||||||
|
_deadline, diag = _watchdog_pending.pop(rid)
|
||||||
|
fired.append(diag)
|
||||||
|
# Fire _on_timeout outside the watchdog lock so a slow logger / stack
|
||||||
|
# snapshot can't stall the scan or block registering new requests.
|
||||||
|
for diag in fired:
|
||||||
|
try:
|
||||||
|
diag._on_timeout()
|
||||||
|
except Exception: # never let one bad record kill the watchdog
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _watchdog_register(request_id: str, deadline: float, diag: "RequestDiagnostics") -> None:
|
||||||
|
with _watchdog_cv:
|
||||||
|
_watchdog_pending[request_id] = (deadline, diag)
|
||||||
|
_ensure_watchdog_running()
|
||||||
|
_watchdog_cv.notify()
|
||||||
|
|
||||||
|
|
||||||
|
def _watchdog_unregister(request_id: str) -> None:
|
||||||
|
with _watchdog_cv:
|
||||||
|
_watchdog_pending.pop(request_id, None)
|
||||||
|
|
||||||
|
|
||||||
def _slow_request_seconds() -> float:
|
def _slow_request_seconds() -> float:
|
||||||
raw = os.getenv("HERMES_WEBUI_SLOW_REQUEST_SECONDS", "").strip()
|
raw = os.getenv("HERMES_WEBUI_SLOW_REQUEST_SECONDS", "").strip()
|
||||||
@@ -53,11 +120,12 @@ class RequestDiagnostics:
|
|||||||
self._current_stage_started = self.started_monotonic
|
self._current_stage_started = self.started_monotonic
|
||||||
self._finished = False
|
self._finished = False
|
||||||
self._watchdog_logged = False
|
self._watchdog_logged = False
|
||||||
self._timer: threading.Timer | None = None
|
|
||||||
if auto_start and self.timeout_seconds > 0:
|
if auto_start and self.timeout_seconds > 0:
|
||||||
self._timer = threading.Timer(self.timeout_seconds, self._on_timeout)
|
_watchdog_register(
|
||||||
self._timer.daemon = True
|
self.request_id,
|
||||||
self._timer.start()
|
self.started_monotonic + self.timeout_seconds,
|
||||||
|
self,
|
||||||
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def maybe_start(
|
def maybe_start(
|
||||||
@@ -91,16 +159,16 @@ class RequestDiagnostics:
|
|||||||
self._current_stage_started = now
|
self._current_stage_started = now
|
||||||
|
|
||||||
def finish(self) -> None:
|
def finish(self) -> None:
|
||||||
timer = None
|
|
||||||
record = None
|
record = None
|
||||||
with self._lock:
|
with self._lock:
|
||||||
if self._finished:
|
if self._finished:
|
||||||
return
|
return
|
||||||
self._finished = True
|
self._finished = True
|
||||||
timer = self._timer
|
|
||||||
record = self._build_record_locked(include_stacks=False)
|
record = self._build_record_locked(include_stacks=False)
|
||||||
if timer is not None:
|
# Drop ourselves from the watchdog so the process-global scan never
|
||||||
timer.cancel()
|
# fires _on_timeout for a completed request (and the pending dict stays
|
||||||
|
# bounded by the number of in-flight requests).
|
||||||
|
_watchdog_unregister(self.request_id)
|
||||||
if record and self.timeout_seconds > 0 and record["elapsed_ms"] >= self.timeout_seconds * 1000:
|
if record and self.timeout_seconds > 0 and record["elapsed_ms"] >= self.timeout_seconds * 1000:
|
||||||
self.logger.warning(
|
self.logger.warning(
|
||||||
"Slow WebUI request completed: %s",
|
"Slow WebUI request completed: %s",
|
||||||
|
|||||||
@@ -0,0 +1,90 @@
|
|||||||
|
"""Regression coverage for #4973 — RequestDiagnostics must not spawn one OS
|
||||||
|
thread per request.
|
||||||
|
|
||||||
|
Before the fix, every ``RequestDiagnostics(auto_start=True)`` started its own
|
||||||
|
``threading.Timer`` (one OS thread per request, held alive until ``finish()``).
|
||||||
|
Under sustained ``/api/sessions`` poll load that exhausts the per-process thread
|
||||||
|
cap (``RuntimeError: can't start new thread``) and the server stops accepting
|
||||||
|
connections. The fix replaces the per-instance timer with a single
|
||||||
|
process-global watchdog daemon thread.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
|
||||||
|
import api.request_diagnostics as rd
|
||||||
|
from api.request_diagnostics import RequestDiagnostics
|
||||||
|
|
||||||
|
|
||||||
|
def _watchdog_thread_count() -> int:
|
||||||
|
return sum(
|
||||||
|
1 for t in threading.enumerate()
|
||||||
|
if t.name == "request-diagnostics-watchdog"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_many_concurrent_diags_do_not_spawn_a_thread_each():
|
||||||
|
"""N in-flight auto_start diags must add at most ONE watchdog thread."""
|
||||||
|
logger = logging.getLogger("test.issue4973.bounded")
|
||||||
|
baseline = _watchdog_thread_count()
|
||||||
|
|
||||||
|
diags = [
|
||||||
|
RequestDiagnostics(
|
||||||
|
"GET", "/api/sessions", logger=logger, timeout_seconds=30
|
||||||
|
)
|
||||||
|
for _ in range(200)
|
||||||
|
]
|
||||||
|
try:
|
||||||
|
# The whole point: 200 concurrent diags, not 200 new threads.
|
||||||
|
added = _watchdog_thread_count() - baseline
|
||||||
|
assert added <= 1, f"expected <=1 watchdog thread, got {added} added"
|
||||||
|
# All 200 are registered as pending (none lost).
|
||||||
|
with rd._watchdog_cv:
|
||||||
|
pending = len(rd._watchdog_pending)
|
||||||
|
assert pending >= 200
|
||||||
|
finally:
|
||||||
|
for d in diags:
|
||||||
|
d.finish()
|
||||||
|
|
||||||
|
# finish() unregisters every diag so the pending dict stays bounded.
|
||||||
|
with rd._watchdog_cv:
|
||||||
|
remaining = sum(
|
||||||
|
1 for _rid, (_dl, diag) in rd._watchdog_pending.items()
|
||||||
|
if diag in diags
|
||||||
|
)
|
||||||
|
assert remaining == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_watchdog_fires_on_timeout_for_a_slow_request(caplog):
|
||||||
|
"""A request that never finishes before its deadline still gets logged by
|
||||||
|
the process-global watchdog (behavioral, not just structural)."""
|
||||||
|
logger = logging.getLogger("test.issue4973.fires")
|
||||||
|
# Tiny timeout so the watchdog (1s tick) fires quickly.
|
||||||
|
diag = RequestDiagnostics(
|
||||||
|
"GET", "/api/sessions", logger=logger, timeout_seconds=0.05
|
||||||
|
)
|
||||||
|
diag.stage("slow_stage")
|
||||||
|
with caplog.at_level(logging.WARNING, logger=logger.name):
|
||||||
|
deadline = time.monotonic() + 4.0
|
||||||
|
while time.monotonic() < deadline:
|
||||||
|
if any("still running" in (r.getMessage() or "") for r in caplog.records):
|
||||||
|
break
|
||||||
|
time.sleep(0.1)
|
||||||
|
assert any("still running" in (r.getMessage() or "") for r in caplog.records), \
|
||||||
|
"watchdog did not log the slow request within the window"
|
||||||
|
diag.finish()
|
||||||
|
|
||||||
|
|
||||||
|
def test_finish_before_deadline_prevents_watchdog_log(caplog):
|
||||||
|
"""A request that finishes fast must NOT be logged by the watchdog."""
|
||||||
|
logger = logging.getLogger("test.issue4973.fast")
|
||||||
|
diag = RequestDiagnostics(
|
||||||
|
"GET", "/api/sessions", logger=logger, timeout_seconds=0.05
|
||||||
|
)
|
||||||
|
diag.finish() # completes immediately, well under the deadline
|
||||||
|
with caplog.at_level(logging.WARNING, logger=logger.name):
|
||||||
|
time.sleep(1.5) # give the watchdog a couple of ticks
|
||||||
|
assert not any(
|
||||||
|
"still running" in (r.getMessage() or "") for r in caplog.records
|
||||||
|
), "watchdog logged a request that already finished"
|
||||||
Reference in new issue
Block a user