mirror of
https://github.com/debpalash/VoiceStudio.git
synced 2026-10-02 01:26:35 +08:00
The liveness probe was a sync route that imported torch and queried the CUDA driver on every probe from the shared worker pool, so a busy but healthy backend missed its 1.5 s deadline and was reported as unresponsive. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 5.5
parent
8768f7803c
commit
fe71ce5c56
+53
-10
@@ -1847,8 +1847,59 @@ _mimetypes.add_type("audio/flac", ".flac")
|
||||
# not-ready, exactly like the connection-refused it replaces), the full body
|
||||
# once ready. No torch import pre-ready — it would block 10-20s on the very
|
||||
# import whose progress this endpoint exists to report.
|
||||
_health_device: str | None = None
|
||||
_health_device_lock = threading.Lock()
|
||||
_health_device_thread: threading.Thread | None = None
|
||||
|
||||
|
||||
def _probe_health_device() -> str:
|
||||
"""Name the compute device. Blocking: imports torch and may initialise the
|
||||
CUDA driver, which can take seconds and stall behind a busy GPU."""
|
||||
import torch
|
||||
|
||||
if torch.cuda.is_available():
|
||||
return f"cuda ({torch.cuda.get_device_name(0)})"
|
||||
if hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
|
||||
return "mps"
|
||||
return "cpu"
|
||||
|
||||
|
||||
def _resolve_health_device() -> None:
|
||||
global _health_device
|
||||
try:
|
||||
_health_device = _probe_health_device()
|
||||
except Exception: # noqa: BLE001 - a liveness label must never fail a probe
|
||||
_health_device = "unknown"
|
||||
|
||||
|
||||
def _cached_health_device() -> str:
|
||||
"""The device label without ever blocking the caller.
|
||||
|
||||
The first call starts one background resolver and answers "unknown"; every
|
||||
later call returns the cached label. /health is the shell's liveness probe,
|
||||
polled every 2 s for the life of the app with a 1.5 s deadline, so it must
|
||||
cost O(1): re-asking torch/the CUDA driver on each probe, from a worker
|
||||
thread competing with a generation for the GIL and the driver, is how a
|
||||
healthy, busy backend got reported as "not responding" (#2490, #2491).
|
||||
"""
|
||||
global _health_device_thread
|
||||
if _health_device is not None:
|
||||
return _health_device
|
||||
with _health_device_lock:
|
||||
if _health_device is None and _health_device_thread is None:
|
||||
_health_device_thread = threading.Thread(
|
||||
target=_resolve_health_device, name="health-device", daemon=True
|
||||
)
|
||||
_health_device_thread.start()
|
||||
return _health_device or "unknown"
|
||||
|
||||
|
||||
# `async def`, deliberately: a sync route runs in the shared 40-thread worker
|
||||
# pool, where it queues behind every blocked sync route (model/status polls
|
||||
# waiting on a load lock, long synchronous handlers). The liveness probe must
|
||||
# depend on the event loop alone, because that is what it is reporting on.
|
||||
@app.get("/health")
|
||||
def health():
|
||||
async def health():
|
||||
if not _startup_progress.is_ready():
|
||||
_step, _label = _startup_progress.current_step()
|
||||
return JSONResponse(
|
||||
@@ -1861,15 +1912,7 @@ def health():
|
||||
},
|
||||
headers={"Retry-After": "2"},
|
||||
)
|
||||
import torch
|
||||
|
||||
device = "cpu"
|
||||
if torch.cuda.is_available():
|
||||
device = f"cuda ({torch.cuda.get_device_name(0)})"
|
||||
elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
|
||||
device = "mps"
|
||||
|
||||
return {"status": "ok", "device": device, "version": APP_VERSION}
|
||||
return {"status": "ok", "device": _cached_health_device(), "version": APP_VERSION}
|
||||
|
||||
|
||||
# ── Startup progress ────────────────────────────────────────────────────
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
"""#2490 / #2491: the liveness probe must depend on the event loop alone.
|
||||
|
||||
The desktop shell polls ``/health`` every 2 s with a 1.5 s deadline for the life
|
||||
of the app, and reports "Backend is running but temporarily not responding"
|
||||
when three probes in a row miss. ``/health`` used to be a sync route that
|
||||
imported torch and asked the CUDA driver for the device name on EVERY probe,
|
||||
from the shared 40-thread worker pool. Under a generation (GIL + driver busy)
|
||||
or with the pool full of blocked sync routes, a perfectly healthy backend missed
|
||||
its probes and was announced as unresponsive.
|
||||
|
||||
Fail-before: ``health`` is a plain function and a blocked device probe blocks it.
|
||||
Pass-after: it is a coroutine that answers immediately and never waits on torch.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import importlib
|
||||
import inspect
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(
|
||||
0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "backend")
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def main(monkeypatch, tmp_path):
|
||||
monkeypatch.setenv("OMNIVOICE_DATA_DIR", str(tmp_path))
|
||||
module = importlib.import_module("main")
|
||||
monkeypatch.setattr(module, "_health_device", None, raising=False)
|
||||
monkeypatch.setattr(module, "_health_device_thread", None, raising=False)
|
||||
return module
|
||||
|
||||
|
||||
def test_health_runs_on_the_event_loop_not_the_worker_pool(main):
|
||||
assert inspect.iscoroutinefunction(main.health), (
|
||||
"a sync /health queues behind every blocked sync route in the 40-thread "
|
||||
"pool, so liveness would depend on unrelated handlers"
|
||||
)
|
||||
|
||||
|
||||
def test_a_blocked_device_probe_never_delays_the_liveness_answer(main, monkeypatch):
|
||||
release = threading.Event()
|
||||
entered = threading.Event()
|
||||
|
||||
def _blocked_probe() -> str:
|
||||
entered.set()
|
||||
release.wait(30)
|
||||
return "cuda (Test GPU)"
|
||||
|
||||
monkeypatch.setattr(main, "_probe_health_device", _blocked_probe)
|
||||
try:
|
||||
started = time.monotonic()
|
||||
first = asyncio.run(main.health())
|
||||
assert time.monotonic() - started < 1.0, "the probe waited on the device lookup"
|
||||
assert first["status"] == "ok"
|
||||
assert first["device"] == "unknown"
|
||||
assert entered.wait(10), "the device label was never resolved in the background"
|
||||
|
||||
# Repeated probes while the lookup is still wedged stay instant and do
|
||||
# not stack up more resolver threads.
|
||||
for _ in range(3):
|
||||
assert asyncio.run(main.health())["status"] == "ok"
|
||||
assert (
|
||||
sum(t.name == "health-device" and t.is_alive() for t in threading.enumerate()) == 1
|
||||
)
|
||||
finally:
|
||||
release.set()
|
||||
if main._health_device_thread is not None:
|
||||
main._health_device_thread.join(timeout=10)
|
||||
|
||||
assert asyncio.run(main.health())["device"] == "cuda (Test GPU)"
|
||||
|
||||
|
||||
def test_a_failing_device_lookup_does_not_fail_the_probe(main, monkeypatch):
|
||||
def _boom() -> str:
|
||||
raise RuntimeError("driver exploded")
|
||||
|
||||
monkeypatch.setattr(main, "_probe_health_device", _boom)
|
||||
asyncio.run(main.health())
|
||||
main._health_device_thread.join(timeout=10)
|
||||
body = asyncio.run(main.health())
|
||||
assert body["status"] == "ok"
|
||||
assert body["device"] == "unknown"
|
||||
Reference in New Issue
Block a user