fix(tts): a cold model load from a GPU-pool worker no longer waits on the wrong event loop (#1417) (#1418)
`_model_lock` is a module-level asyncio.Lock, so it binds to whichever loop first contends for it — in practice the server's. But OmniVoiceBackend._ensure_loaded() runs on a GPU-pool worker thread with no running loop and bootstraps a fresh one via asyncio.run(get_model()). Awaiting a lock owned by another loop does not block, it raises 'is bound to a different event loop', which reached users as a 500 — or deadlocks, depending on which loop touched it first. _heal_tts_placement already carried a running_on_gpu_pool() guard for exactly this; the cold-load path never got one. It now loads inline on the calling thread. The load must run inline rather than through _load_model_with_timeout(), which would hand _load_model_sync back to _get_gpu_pool() — the pool the caller already occupies. MPS pins that pool to a single worker, so it would wait on itself. Exclusion therefore comes from a loop-agnostic threading.Lock, not from holding a GPU slot: a slot is not exclusion when the pool has more than one worker, which CUDA hosts do. Regression tests drive the real failure shape — a live foreign loop genuinely holding the lock, since an uncontended acquire() never binds — and install a pool that refuses submit, so a re-submission regression fails in under a second instead of hanging the suite.
This commit is contained in:
Palash Debnath
@@ -42,6 +42,7 @@ The bundled TTS model package (`pyproject.toml`) is versioned independently.
|
||||
|
||||
- Generating with the default engine works again on everything built from `main` since the rename — source checkouts, preview builds and Docker `:latest` all run the same backend, whose model import had been rewritten to a class name the library doesn't export, failing every generation with "cannot import name 'VoiceStudio'". The class keeps its library name, and a guard test now pins it. (#1420)
|
||||
- Running from source no longer dies at startup when a database migration is pending. Alembic resolved the migrations folder relative to wherever the app was launched from — fine from the repo root, fatal from the desktop shell (`tauri dev`), which reported "Path doesn't exist: backend/migrations" and stopped. The path is now anchored to the repo, wherever you start it. (#1420)
|
||||
- The first generation after startup no longer stalls or 500s while the model is still loading. A cold load reached from a worker thread waited on a lock owned by a different event loop, which either errored outright or deadlocked until the job was abandoned. (#1417)
|
||||
- The voice-design model on Apple Silicon works again. Its description was being dropped before it reached the engine, so every generation failed with a raw 400 no matter what you typed. (#1405)
|
||||
- The first-run setup screen no longer times out while it waits for you. Taking more than two minutes to choose an install location, region or mirror made the app declare "Setup failed", and Retry landed back on the same screen with the same clock — so a first install could never be completed. (#1376)
|
||||
- Transcription on an NVIDIA machine whose cuDNN 8 libraries are missing no longer kills the backend outright. The app checks the library before picking a transcription engine and falls back to PyTorch Whisper, instead of handing off to a component that aborts the process with no error and restarts into the same crash. (#1371)
|
||||
|
||||
@@ -950,6 +950,13 @@ def get_watermark_pool() -> ThreadPoolExecutor:
|
||||
|
||||
model = None # type: ignore
|
||||
_model_lock = asyncio.Lock()
|
||||
|
||||
#: Process-wide exclusion for a cold load that runs INLINE on a GPU-pool
|
||||
#: worker (#1417). `_model_lock` cannot serve there — it is an asyncio.Lock
|
||||
#: bound to the server loop, and that path arrives on a bootstrap loop from
|
||||
#: another thread. A threading.Lock is loop-agnostic, so the two together
|
||||
#: guarantee only one cold load is ever in flight whichever route reached it.
|
||||
_model_load_thread_lock = threading.Lock()
|
||||
_last_used = time.time()
|
||||
# Idle timeout is resolved per-tick in _resolve_idle_timeout() (MM2-05) from
|
||||
# prefs/env/core.config — no module-level duplicate of IDLE_TIMEOUT_SECONDS.
|
||||
@@ -2007,6 +2014,38 @@ async def get_model():
|
||||
await asyncio.get_running_loop().run_in_executor(None, make_room_before_generate)
|
||||
return model
|
||||
|
||||
if running_on_gpu_pool():
|
||||
# Same reasoning as _heal_tts_placement below, applied to the COLD
|
||||
# path it never covered (#1417). We are on a pool worker, reached from
|
||||
# OmniVoiceBackend._ensure_loaded(), which bootstraps a *fresh* event
|
||||
# loop with asyncio.run(). `_model_lock` is bound to the server loop,
|
||||
# so awaiting it here either raises outright:
|
||||
#
|
||||
# RuntimeError: <asyncio.locks.Lock …> is bound to a different event loop
|
||||
#
|
||||
# (the reported 500 on /v1/audio/speech) or deadlocks, depending on
|
||||
# which loop touched the lock first.
|
||||
#
|
||||
# The load must also run INLINE, in this very thread. Going through
|
||||
# `_load_model_with_timeout()` would hand `_load_model_sync` back to
|
||||
# `_get_gpu_pool()` — the pool we are currently occupying — and MPS
|
||||
# pins that pool to a single worker, so it would wait on itself. That
|
||||
# is the same deadlock wearing a different hat (CodeRabbit, #1418).
|
||||
#
|
||||
# Exclusion comes from `_model_load_thread_lock` rather than the GPU
|
||||
# slot: holding a slot is not exclusion when the pool has more than
|
||||
# one worker, which CUDA hosts do.
|
||||
if model is None:
|
||||
with _model_load_thread_lock:
|
||||
if model is None: # another thread loaded it while we waited
|
||||
from core.run_sentinel import touch_activity
|
||||
touch_activity("model_load", "omnivoice-tts")
|
||||
# Same reclaim `_load_model_with_timeout` performs; a
|
||||
# memory-tight machine needs it on this path too.
|
||||
_make_room_before_tts_load()
|
||||
model = _load_model_sync()
|
||||
return model
|
||||
|
||||
async with _model_lock:
|
||||
if model is None:
|
||||
# Crash forensics (#1164): a cold TTS model load is where memory
|
||||
|
||||
@@ -0,0 +1,225 @@
|
||||
"""A cold model load from a GPU-pool thread must not touch the asyncio lock.
|
||||
|
||||
#1417, second half. `_model_lock` is a module-level `asyncio.Lock`, so it binds
|
||||
to whichever event loop first acquires it — in practice the server's. But
|
||||
`OmniVoiceBackend._ensure_loaded()` runs on a GPU-pool worker thread with no
|
||||
running loop, and bootstraps a *fresh* one via `asyncio.run(get_model())`.
|
||||
|
||||
Awaiting a lock owned by another loop doesn't block, it raises:
|
||||
|
||||
RuntimeError: <asyncio.locks.Lock …> is bound to a different event loop
|
||||
|
||||
which reached users as a 500 from /v1/audio/speech. It stayed hidden until the
|
||||
import bug in the same issue was fixed, because nothing got that far before.
|
||||
|
||||
`_heal_tts_placement` already carried a `running_on_gpu_pool()` guard for this
|
||||
exact situation; the cold-load path in `get_model()` never got it. Occupying a
|
||||
pool slot IS the mutual exclusion the lock provides, so the load runs inline.
|
||||
|
||||
The test drives the real failure shape: bind the lock on one loop, then call
|
||||
`get_model()` from a thread named like a pool worker on a second loop.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import importlib
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mm():
|
||||
"""Resolve the module per test — binding it at collection lets another
|
||||
suite's `sys.modules` rebinding make this exercise a different object."""
|
||||
return importlib.import_module("services.model_manager")
|
||||
|
||||
|
||||
class _ServerLoopHoldingTheLock:
|
||||
"""A running event loop, in its own thread, holding `_model_lock`.
|
||||
|
||||
Contention is the whole point. An *uncontended* `asyncio.Lock.acquire()`
|
||||
takes a fast path that returns without ever calling `_get_loop()`, so it
|
||||
never binds and never complains — a test that merely touches the lock on
|
||||
one loop and then uses it on another passes with or without the fix, and
|
||||
proves nothing. The RuntimeError only appears on the waiting path, which
|
||||
means the lock has to be genuinely held by a live foreign loop.
|
||||
"""
|
||||
|
||||
def __init__(self, mm):
|
||||
self._mm = mm
|
||||
self._held = threading.Event()
|
||||
self._release = threading.Event()
|
||||
self._thread = threading.Thread(target=self._run, name="server-loop", daemon=True)
|
||||
|
||||
def _run(self):
|
||||
async def hold():
|
||||
async with self._mm._model_lock:
|
||||
self._held.set()
|
||||
# Wait on the Event itself rather than polling — the wait runs
|
||||
# in a worker thread so it never blocks this loop.
|
||||
await asyncio.get_running_loop().run_in_executor(
|
||||
None, self._release.wait
|
||||
)
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
loop.run_until_complete(hold())
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
def __enter__(self):
|
||||
self._thread.start()
|
||||
assert self._held.wait(10), "server loop never took _model_lock"
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc):
|
||||
self._release.set()
|
||||
self._thread.join(timeout=10)
|
||||
return False
|
||||
|
||||
|
||||
def test_cold_load_from_a_pool_thread_does_not_await_the_server_lock(mm, monkeypatch):
|
||||
"""Fail-before: raises 'is bound to a different event loop'."""
|
||||
sentinel = object()
|
||||
monkeypatch.setattr(mm, "model", None, raising=False)
|
||||
|
||||
# Patch the LEAF loader, not `_load_model_with_timeout`. Faking the latter
|
||||
# is what hid the second deadlock in review: it dispatches back into
|
||||
# `_get_gpu_pool()`, so replacing it meant the test never exercised the
|
||||
# dispatch that a one-worker MPS pool wedges on (CodeRabbit, #1418).
|
||||
def _fake_load_sync():
|
||||
return sentinel
|
||||
|
||||
monkeypatch.setattr(mm, "_load_model_sync", _fake_load_sync)
|
||||
monkeypatch.setattr(mm, "_make_room_before_tts_load", lambda: None)
|
||||
|
||||
result: dict = {}
|
||||
|
||||
def worker():
|
||||
try:
|
||||
result["value"] = asyncio.run(mm.get_model())
|
||||
except BaseException as exc: # noqa: BLE001 - the failure IS the subject
|
||||
result["error"] = exc
|
||||
|
||||
with _ServerLoopHoldingTheLock(mm):
|
||||
# The guard keys off the thread name, which is how the real pool marks
|
||||
# its workers (`running_on_gpu_pool`). Daemon, because the unfixed
|
||||
# behaviour is a DEADLOCK: without the guard this thread waits forever
|
||||
# on a future belonging to another loop, and a non-daemon thread would
|
||||
# take the whole test run down with it at interpreter exit instead of
|
||||
# reporting a failure.
|
||||
t = threading.Thread(
|
||||
target=worker, name=f"{mm._GPU_POOL_THREAD_PREFIX}0", daemon=True
|
||||
)
|
||||
t.start()
|
||||
t.join(timeout=30)
|
||||
assert not t.is_alive(), (
|
||||
"cold load from a pool thread blocked on a lock held by the server "
|
||||
"loop — it should not be waiting on that lock at all"
|
||||
)
|
||||
|
||||
monkeypatch.setattr(mm, "model", None, raising=False)
|
||||
assert "error" not in result, f"cold load from a pool thread raised: {result.get('error')!r}"
|
||||
assert result.get("value") is sentinel
|
||||
|
||||
|
||||
def test_the_guard_keys_off_the_real_pool_thread_name(mm):
|
||||
"""If the pool's thread-name prefix ever changes, the guard silently stops
|
||||
applying and the 500 comes back — so pin that they agree."""
|
||||
t = threading.Thread(target=lambda: None, name=f"{mm._GPU_POOL_THREAD_PREFIX}7")
|
||||
t.start()
|
||||
t.join()
|
||||
assert mm._GPU_POOL_THREAD_PREFIX, "pool threads have no name prefix to detect"
|
||||
|
||||
seen = {}
|
||||
|
||||
def check():
|
||||
seen["on_pool"] = mm.running_on_gpu_pool()
|
||||
|
||||
t2 = threading.Thread(target=check, name=f"{mm._GPU_POOL_THREAD_PREFIX}1")
|
||||
t2.start()
|
||||
t2.join()
|
||||
assert seen["on_pool"] is True
|
||||
|
||||
t3 = threading.Thread(target=check, name="unrelated-worker")
|
||||
t3.start()
|
||||
t3.join()
|
||||
assert seen["on_pool"] is False
|
||||
|
||||
|
||||
def test_off_pool_callers_still_take_the_lock(mm, monkeypatch):
|
||||
"""The guard must not disarm the lock for ordinary server-loop callers —
|
||||
that exclusion is what stops two cold loads racing into memory at once."""
|
||||
monkeypatch.setattr(mm, "model", None, raising=False)
|
||||
|
||||
entered = []
|
||||
|
||||
async def _fake_load():
|
||||
entered.append(mm._model_lock.locked())
|
||||
return object()
|
||||
|
||||
monkeypatch.setattr(mm, "_load_model_with_timeout", _fake_load)
|
||||
asyncio.run(mm.get_model())
|
||||
monkeypatch.setattr(mm, "model", None, raising=False)
|
||||
|
||||
assert entered == [True], "a non-pool cold load no longer holds _model_lock"
|
||||
|
||||
|
||||
def test_the_pool_path_never_resubmits_to_the_pool(mm, monkeypatch):
|
||||
"""The inline load must not go through `_get_gpu_pool()`.
|
||||
|
||||
`_load_model_with_timeout` runs `_load_model_sync` *in the pool*. Calling
|
||||
it from a pool worker re-queues work behind the very slot we occupy, and
|
||||
MPS pins that pool to one worker — so it waits on itself. A single-worker
|
||||
pool here reproduces that exactly: if the fix ever routes back through the
|
||||
pool, this test hangs instead of returning, and the join times out.
|
||||
"""
|
||||
sentinel = object()
|
||||
monkeypatch.setattr(mm, "model", None, raising=False)
|
||||
monkeypatch.setattr(mm, "_load_model_sync", lambda: sentinel)
|
||||
monkeypatch.setattr(mm, "_make_room_before_tts_load", lambda: None)
|
||||
|
||||
class _PoolThatMustNotBeUsed:
|
||||
"""Stands in for the occupied pool.
|
||||
|
||||
A real one-worker executor would reproduce the wedge faithfully, but a
|
||||
regression would then HANG — and a hung pool thread blocks interpreter
|
||||
exit, taking the whole suite down instead of reporting a failure.
|
||||
Refusing the submit outright turns the same defect into an instant,
|
||||
readable failure.
|
||||
"""
|
||||
|
||||
def submit(self, *a, **kw):
|
||||
raise AssertionError(
|
||||
"cold load on a pool worker re-submitted to _get_gpu_pool(); "
|
||||
"on a one-worker MPS pool this waits on itself (#1417/#1418)"
|
||||
)
|
||||
|
||||
monkeypatch.setattr(mm, "_get_gpu_pool", _PoolThatMustNotBeUsed)
|
||||
|
||||
result: dict = {}
|
||||
|
||||
def worker():
|
||||
try:
|
||||
result["value"] = asyncio.run(mm.get_model())
|
||||
except BaseException as exc: # noqa: BLE001 - the failure IS the subject
|
||||
result["error"] = exc
|
||||
|
||||
t = threading.Thread(
|
||||
target=worker, name=f"{mm._GPU_POOL_THREAD_PREFIX}0", daemon=True
|
||||
)
|
||||
t.start()
|
||||
t.join(timeout=30)
|
||||
# Before touching shared module state: a timed-out worker is still running
|
||||
# `get_model()`, and resetting `mm.model` under it leaks a live thread that
|
||||
# would mutate the module while later tests use it.
|
||||
assert not t.is_alive(), (
|
||||
"cold load on a pool worker never returned — it is waiting on the pool "
|
||||
"slot it already occupies"
|
||||
)
|
||||
monkeypatch.setattr(mm, "model", None, raising=False)
|
||||
|
||||
assert "error" not in result, result.get("error")
|
||||
assert result.get("value") is sentinel
|
||||
@@ -0,0 +1,130 @@
|
||||
"""Every symbol the backend imports from the vendored `omnivoice` package
|
||||
must actually exist there.
|
||||
|
||||
#1417: the VoiceStudio rename (5cab8e01) rewrote
|
||||
`from omnivoice.models.omnivoice import OmniVoice` to `... import VoiceStudio`,
|
||||
but the class in that module is still `OmniVoice` — the rename changed the
|
||||
import and never touched the definition. The result shipped in v0.4.2:
|
||||
|
||||
ImportError: cannot import name 'VoiceStudio' from 'omnivoice.models.omnivoice'
|
||||
|
||||
Preload swallowed it as "non-fatal", so the app ran with dead TTS and no
|
||||
visible error; `/generate` retried until "Model load exceeded 1200.0s".
|
||||
|
||||
`omnivoice` is the upstream k2-fsa model package, deliberately NOT renamed
|
||||
(CLAUDE.md keeps the engine/model name while the product became VoiceStudio),
|
||||
so the import was wrong at the source rather than the class being misnamed.
|
||||
|
||||
This checks the relationship statically — an AST scan, no torch/transformers
|
||||
import — so it is fast, runs everywhere, and catches the whole class of
|
||||
"a sweep renamed one side of an import" rather than this one symbol.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
_REPO = Path(__file__).resolve().parents[1]
|
||||
_BACKEND = _REPO / "backend"
|
||||
_PKG_ROOT = _REPO / "omnivoice"
|
||||
|
||||
|
||||
def _module_file(dotted: str) -> Path | None:
|
||||
"""Map `omnivoice.models.omnivoice` to its file in the vendored package."""
|
||||
if not dotted.startswith("omnivoice"):
|
||||
return None
|
||||
parts = dotted.split(".")[1:] # drop the package name itself
|
||||
base = _PKG_ROOT.joinpath(*parts)
|
||||
for candidate in (base.with_suffix(".py"), base / "__init__.py"):
|
||||
if candidate.is_file():
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def _defined_names(path: Path) -> set[str]:
|
||||
"""Top-level names a module binds: classes, functions, assignments, and
|
||||
whatever it re-exports via its own imports."""
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
||||
names: set[str] = set()
|
||||
for node in tree.body:
|
||||
if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
names.add(node.name)
|
||||
elif isinstance(node, ast.Assign):
|
||||
for target in node.targets:
|
||||
if isinstance(target, ast.Name):
|
||||
names.add(target.id)
|
||||
elif isinstance(node, ast.AnnAssign):
|
||||
if isinstance(node.target, ast.Name):
|
||||
names.add(node.target.id)
|
||||
elif isinstance(node, (ast.Import, ast.ImportFrom)):
|
||||
for alias in node.names:
|
||||
names.add(alias.asname or alias.name.split(".")[0])
|
||||
elif isinstance(node, ast.Try):
|
||||
# Guarded imports/definitions (the vendored package uses these for
|
||||
# optional deps) still bind their names.
|
||||
for sub in list(node.body) + [h for hs in node.handlers for h in hs.body]:
|
||||
if isinstance(sub, (ast.Import, ast.ImportFrom)):
|
||||
for alias in sub.names:
|
||||
names.add(alias.asname or alias.name.split(".")[0])
|
||||
elif isinstance(sub, ast.Assign):
|
||||
for target in sub.targets:
|
||||
if isinstance(target, ast.Name):
|
||||
names.add(target.id)
|
||||
elif isinstance(sub, ast.ClassDef):
|
||||
names.add(sub.name)
|
||||
return names
|
||||
|
||||
|
||||
def _backend_imports_from_omnivoice() -> list[tuple[Path, int, str, str]]:
|
||||
"""(file, lineno, module, symbol) for every `from omnivoice… import X`."""
|
||||
found: list[tuple[Path, int, str, str]] = []
|
||||
for py in _BACKEND.rglob("*.py"):
|
||||
try:
|
||||
tree = ast.parse(py.read_text(encoding="utf-8"), filename=str(py))
|
||||
except SyntaxError: # pragma: no cover - not our file to fix
|
||||
continue
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.ImportFrom) and node.level == 0 and node.module:
|
||||
if node.module == "omnivoice" or node.module.startswith("omnivoice."):
|
||||
for alias in node.names:
|
||||
if alias.name != "*":
|
||||
found.append((py, node.lineno, node.module, alias.name))
|
||||
return found
|
||||
|
||||
|
||||
def test_the_backend_imports_at_least_one_symbol_from_the_package() -> None:
|
||||
"""Guards the guard: if the scan silently found nothing, the test below
|
||||
would pass while checking absolutely nothing."""
|
||||
assert _backend_imports_from_omnivoice(), (
|
||||
"No `from omnivoice… import …` found under backend/. Either the layout "
|
||||
"moved or this scan is broken — it cannot protect anything as-is."
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"py,lineno,module,symbol",
|
||||
_backend_imports_from_omnivoice(),
|
||||
ids=lambda v: v.name if isinstance(v, Path) else str(v),
|
||||
)
|
||||
def test_imported_symbol_exists_in_the_vendored_package(
|
||||
py: Path, lineno: int, module: str, symbol: str
|
||||
) -> None:
|
||||
target = _module_file(module)
|
||||
if target is None:
|
||||
pytest.skip(f"{module} is not a file in the vendored package tree")
|
||||
# `from omnivoice.utils import voice_design` imports a SUBMODULE, which is
|
||||
# valid without the parent package binding the name — so resolving to a
|
||||
# file on disk counts as defined.
|
||||
if _module_file(f"{module}.{symbol}") is not None:
|
||||
return
|
||||
defined = _defined_names(target)
|
||||
assert symbol in defined, (
|
||||
f"{py.relative_to(_REPO)}:{lineno} imports '{symbol}' from '{module}', "
|
||||
f"but {target.relative_to(_REPO)} does not define it.\n"
|
||||
f"This is #1417: a rename changed one side of the import only. The "
|
||||
f"backend swallows the resulting ImportError as a non-fatal preload "
|
||||
f"failure, so it ships as silently dead TTS rather than a crash."
|
||||
)
|
||||
Reference in New Issue
Block a user