Merge PR #1603: dispatch foreign loops through serving loop
This commit is contained in:
debpalash
+2
-2
@@ -18,7 +18,7 @@ the frozen-backend fallback mirror it for their toolchains.
|
||||
- The backend binds its port immediately and reports startup progress live — `/health` answers 503-with-step and a new `/startup/progress` endpoint lists every step while PyTorch, API routes, and database migrations load in the background, so "starting at step X" is never mistakable for "dead"; the desktop splash narrates each step (#1550)
|
||||
|
||||
### Added
|
||||
- The locally cached AudioSeal watermark generator warms on a background thread ~35s after boot (`OMNIVOICE_PRELOAD_WATERMARK=0` opts out; explicitly setting `=1` may download it), so the first synthesis no longer serializes the audioseal import + model load inline — measured at ~42s on a cold filesystem, 3s short of a 90s client timeout (#1576) — thanks @paoloantinoro!
|
||||
- The locally cached AudioSeal watermark generator warms on a background thread ~35s after boot (`OMNIVOICE_PRELOAD_WATERMARK=0` opts out; explicitly setting `=1` may download it), so the first synthesis no longer serializes the audioseal import + model load inline — measured at ~42s on a cold filesystem, 3s short of a 90s client timeout (#1576) — thanks @paoloantinori!
|
||||
- Voices you've cloned stay "warm" across restarts — encoded references now persist to disk (~10 KB each), so the first generation of a session skips the re-encode and any transcription pass; `OMNIVOICE_PROMPT_DISK_CACHE=0` opts out (#1565)
|
||||
- Optional FlashInfer acceleration for the default engine on CUDA (`OMNIVOICE_FLASHINFER=1`, ~2.2x measured) — needs the optional `flashinfer-python` package; missing package or kernel failure logs why and falls back to the standard path (#1565)
|
||||
- The bug reporter notices when you're on an outdated build and offers the latest release before filing — with a "File anyway" escape hatch — and stamps a `Build status` line into every report so up-to-date reports are tellable from stale ones (#1547)
|
||||
@@ -32,7 +32,7 @@ the frozen-backend fallback mirror it for their toolchains.
|
||||
- The OmniVoice guide now covers combining style attributes with a reference clip (consistent instruct stabilizes cloning; the reference wins conflicts), inline pronunciation control (pinyin / CMU phonemes), and corrects the claim that the default engine can't do voice design — it can, from attributes (#1565)
|
||||
|
||||
### Fixed
|
||||
- Watermark embedding failures now log the full traceback instead of just the exception message, so a silently-unmarked-audio incident (audio passes through unmarked by design) is diagnosable from the log alone (#1576) — thanks @paoloantinoro!
|
||||
- Watermark embedding failures now log the full traceback instead of just the exception message, so a silently-unmarked-audio incident (audio passes through unmarked by design) is diagnosable from the log alone (#1576) — thanks @paoloantinori!
|
||||
- Dubbing now recovers rapid two-speaker exchanges when diarization collapses them, defaults new projects to lip sync without overwriting saved timing choices, and keeps the editor usable on narrow screens (#1584) — thanks @victordonat0!
|
||||
- `OMNIVOICE_ASR_BACKEND=omnivoice` now selects the PyTorch-native Whisper path, so the documented ROCm escape hatch no longer fails as an unknown engine (#1582) — thanks @patmansk!
|
||||
- Network Sharing from Windows MSI/portable installs now serves the bundled web interface to LAN devices instead of redirecting them to their own `localhost` (#1589) — thanks @TWIISTED-STUDIOS!
|
||||
|
||||
@@ -357,15 +357,11 @@ async def _render_archetype_wav(a: dict, out_path: Path) -> None:
|
||||
# Runs on the dedicated watermark pool (#1190): AudioSeal embedding is CPU
|
||||
# work that holds no VRAM, so it must not occupy a GPU worker ahead of the
|
||||
# next generate on 1-worker hosts.
|
||||
from services.watermark import mark_synthetic
|
||||
from services.model_manager import get_watermark_pool
|
||||
import functools
|
||||
audio_tensor = await run_on_gpu_pool_guarded(
|
||||
functools.partial(mark_synthetic, audio_tensor, model.sampling_rate,
|
||||
context="archetypes.render"),
|
||||
what="Archetype watermark",
|
||||
from services.watermark import mark_synthetic_async
|
||||
audio_tensor = await mark_synthetic_async(
|
||||
audio_tensor, model.sampling_rate,
|
||||
context="archetypes.render",
|
||||
timeout=generate_timeout_s(""),
|
||||
executor=get_watermark_pool(),
|
||||
)
|
||||
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@@ -413,19 +413,15 @@ async def _run_batch_pipeline(job_id: str, job: dict):
|
||||
# unmarked while the interactive dub pipeline marked every segment.
|
||||
# One whole-track embed (chunked internally, #1045) is equivalent to
|
||||
# dub_generate's per-segment marks: the 16-bit message repeats
|
||||
# throughout. Runs in the GPU pool like generate's finalize; never
|
||||
# raises (degrades to unmarked on failure, same as every producer).
|
||||
# throughout. Never raises (degrades to unmarked on failure, same as
|
||||
# every producer).
|
||||
# Dispatched to the dedicated watermark pool, not the GPU pool (#1190):
|
||||
# AudioSeal embedding is CPU work that holds no VRAM, and a whole-track
|
||||
# embed is long enough that occupying a GPU worker with it stalled the
|
||||
# next language's segments on 1-worker hosts.
|
||||
from services.watermark import mark_synthetic
|
||||
from services.model_manager import get_watermark_pool
|
||||
import functools
|
||||
full_audio = await loop.run_in_executor(
|
||||
get_watermark_pool(),
|
||||
functools.partial(mark_synthetic, full_audio, sr,
|
||||
context="batch.dub_track"),
|
||||
from services.watermark import mark_synthetic_async
|
||||
full_audio = await mark_synthetic_async(
|
||||
full_audio, sr, context="batch.dub_track",
|
||||
)
|
||||
|
||||
# Same assembly pattern as dub_generate.py:390 — `full_audio` is a
|
||||
|
||||
@@ -870,7 +870,6 @@ async def _finalize_generation(
|
||||
Returns ``(watermarked_tensor, meta)`` where ``meta`` carries
|
||||
``id`` / ``filename`` / ``duration`` / ``gen_time``.
|
||||
"""
|
||||
loop = asyncio.get_running_loop()
|
||||
# Invisible AudioSeal provenance watermark on the final audio. Embedding
|
||||
# was previously only wired into the dub pipeline (dub_generate.py), so
|
||||
# plain TTS came out unmarked despite the setting being on — and the same
|
||||
@@ -882,12 +881,9 @@ async def _finalize_generation(
|
||||
# AudioSeal embedding is CPU work that holds no VRAM, so occupying a GPU
|
||||
# worker with it only delays the next generate on 1-worker hosts.
|
||||
if not already_marked:
|
||||
from services.watermark import mark_synthetic
|
||||
from services.model_manager import get_watermark_pool
|
||||
audio_tensor = await loop.run_in_executor(
|
||||
get_watermark_pool(),
|
||||
functools.partial(mark_synthetic, audio_tensor, sample_rate,
|
||||
context="generate.finalize"),
|
||||
from services.watermark import mark_synthetic_async
|
||||
audio_tensor = await mark_synthetic_async(
|
||||
audio_tensor, sample_rate, context="generate.finalize",
|
||||
)
|
||||
gen_time = round(time.time() - start_time, 2)
|
||||
|
||||
@@ -1822,12 +1818,10 @@ async def generate_speech(
|
||||
# (#1190): AudioSeal embedding is CPU work that owns no
|
||||
# VRAM, and on a 1-worker host it used to serialize
|
||||
# directly ahead of the next generate.
|
||||
from services.watermark import mark_synthetic
|
||||
from services.model_manager import get_watermark_pool
|
||||
_preview = await asyncio.get_running_loop().run_in_executor(
|
||||
get_watermark_pool(),
|
||||
functools.partial(mark_synthetic, audio_tensor, sample_rate,
|
||||
context="generate.stream_preview"),
|
||||
from services.watermark import mark_synthetic_async
|
||||
_preview = await mark_synthetic_async(
|
||||
audio_tensor, sample_rate,
|
||||
context="generate.stream_preview",
|
||||
)
|
||||
yield _line({"type": "chunk", "seq": 0, "pcm": _pcm16_b64(_preview)})
|
||||
else:
|
||||
@@ -1849,12 +1843,10 @@ async def generate_speech(
|
||||
# Provenance-mark the streamed copy off the GPU pool
|
||||
# (#1169 mark, #1190 placement): CPU-only AudioSeal
|
||||
# work must not occupy a GPU worker between chunks.
|
||||
from services.watermark import mark_synthetic
|
||||
from services.model_manager import get_watermark_pool
|
||||
preview = await asyncio.get_running_loop().run_in_executor(
|
||||
get_watermark_pool(),
|
||||
functools.partial(mark_synthetic, preview, sample_rate,
|
||||
context="generate.stream_preview"),
|
||||
from services.watermark import mark_synthetic_async
|
||||
preview = await mark_synthetic_async(
|
||||
preview, sample_rate,
|
||||
context="generate.stream_preview",
|
||||
)
|
||||
if i == 0:
|
||||
# After the first render so lazy-loading engines
|
||||
|
||||
+11
-15
@@ -64,31 +64,27 @@ def emit(kind: str, payload: dict[str, Any] | None = None) -> None:
|
||||
**(payload or {}),
|
||||
}
|
||||
event_str = json.dumps(event)
|
||||
loop = asyncio.get_running_loop() if _on_event_loop() else _serving_loop
|
||||
if loop is None:
|
||||
try:
|
||||
caller_loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
caller_loop = None
|
||||
target_loop = _serving_loop or caller_loop
|
||||
if target_loop is None:
|
||||
# No serving loop yet — nobody to notify; dropping is correct.
|
||||
logger.debug("No event loop — event dropped: %s", kind)
|
||||
return
|
||||
try:
|
||||
if _on_event_loop():
|
||||
loop.create_task(_broadcast(event_str))
|
||||
if caller_loop is target_loop:
|
||||
target_loop.create_task(_broadcast(event_str))
|
||||
else:
|
||||
# Threadpool worker (sync endpoint body): the only thread-safe way in.
|
||||
loop.call_soon_threadsafe(_schedule_broadcast, event_str)
|
||||
# Sync endpoints and async producers on a foreign loop must both
|
||||
# hand off: the lock and listener queues belong to serving_loop.
|
||||
target_loop.call_soon_threadsafe(_schedule_broadcast, event_str)
|
||||
except RuntimeError:
|
||||
# The serving loop closed between capture and use (app shutdown).
|
||||
logger.debug("Event loop closed — event dropped: %s", kind)
|
||||
|
||||
|
||||
def _on_event_loop() -> bool:
|
||||
"""True when called from the running event loop (async-context emit)."""
|
||||
try:
|
||||
asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _schedule_broadcast(event_str: str) -> None:
|
||||
"""Run `_broadcast` on the serving loop; called via call_soon_threadsafe."""
|
||||
asyncio.get_running_loop().create_task(_broadcast(event_str))
|
||||
|
||||
@@ -1112,6 +1112,10 @@ class _WatermarkExecutor(Executor):
|
||||
self._thread is None or not self._thread.is_alive()
|
||||
)
|
||||
|
||||
def is_shutdown(self) -> bool:
|
||||
with self._lock:
|
||||
return self._shutdown
|
||||
|
||||
def shutdown(
|
||||
self,
|
||||
wait: bool = True,
|
||||
@@ -1150,7 +1154,10 @@ def begin_watermark_pool_lifecycle() -> None:
|
||||
and _watermark_pool_singleton.is_stopped()
|
||||
):
|
||||
_watermark_pool_singleton = None
|
||||
_watermark_pool_accepting = _watermark_pool_singleton is None
|
||||
_watermark_pool_accepting = (
|
||||
_watermark_pool_singleton is None
|
||||
or not _watermark_pool_singleton.is_shutdown()
|
||||
)
|
||||
|
||||
|
||||
def get_watermark_pool() -> _WatermarkExecutor:
|
||||
|
||||
@@ -286,6 +286,36 @@ def mark_synthetic(
|
||||
return marked
|
||||
|
||||
|
||||
async def mark_synthetic_async(
|
||||
waveform: torch.Tensor,
|
||||
sample_rate: int,
|
||||
*,
|
||||
context: str,
|
||||
force: bool = False,
|
||||
timeout: float | None = None,
|
||||
) -> torch.Tensor:
|
||||
"""Dispatch marking without letting a draining pool lose finished audio."""
|
||||
import asyncio
|
||||
import functools
|
||||
|
||||
from services.model_manager import get_watermark_pool, run_on_gpu_pool_guarded
|
||||
|
||||
try:
|
||||
pool = get_watermark_pool()
|
||||
except RuntimeError:
|
||||
logger.warning("Watermark skipped while the prior worker is shutting down")
|
||||
return waveform
|
||||
|
||||
job = functools.partial(
|
||||
mark_synthetic, waveform, sample_rate, context=context, force=force
|
||||
)
|
||||
if timeout is not None:
|
||||
return await run_on_gpu_pool_guarded(
|
||||
job, what="Audio watermark", timeout=timeout, executor=pool
|
||||
)
|
||||
return await asyncio.get_running_loop().run_in_executor(pool, job)
|
||||
|
||||
|
||||
@torch.no_grad()
|
||||
def embed_watermark(
|
||||
waveform: torch.Tensor,
|
||||
|
||||
@@ -62,6 +62,21 @@ import pytest
|
||||
import warnings as _warnings
|
||||
|
||||
|
||||
# App lifespans deliberately close watermark admission during shutdown. The
|
||||
# test process then keeps running and many route tests call handlers without
|
||||
# starting another lifespan, so restore the equivalent of a fresh lifespan at
|
||||
# every test boundary. ``begin_watermark_pool_lifecycle`` still refuses to
|
||||
# reopen while a timed-out worker is genuinely alive, preserving that race
|
||||
# signal instead of hiding it.
|
||||
@pytest.fixture(autouse=True)
|
||||
def _watermark_pool_lifecycle_baseline():
|
||||
model_manager = sys.modules.get("services.model_manager")
|
||||
begin = getattr(model_manager, "begin_watermark_pool_lifecycle", None)
|
||||
if callable(begin):
|
||||
begin()
|
||||
yield
|
||||
|
||||
|
||||
# ── torch default-dtype isolation (CI flaky trio) ───────────────────────────
|
||||
# Three tests (test_effects_chain / test_generation_audio_guard /
|
||||
# test_persona_bundle) fail intermittently on CI — never locally — with
|
||||
|
||||
@@ -66,6 +66,58 @@ def test_emit_from_thread_reaches_serving_loop(bus, tmp_path):
|
||||
loop.close()
|
||||
|
||||
|
||||
def test_emit_from_foreign_running_loop_reaches_serving_loop(bus):
|
||||
"""An async producer may run on a worker loop, but listener state belongs
|
||||
to the WebSocket serving loop and must only be touched there."""
|
||||
serving_loop = asyncio.new_event_loop()
|
||||
serving_loop.set_debug(True)
|
||||
received: list[str] = []
|
||||
started = threading.Event()
|
||||
done = threading.Event()
|
||||
|
||||
async def serve_with_waiter_ready():
|
||||
q = await bus.subscribe()
|
||||
waiter = asyncio.create_task(q.get())
|
||||
await asyncio.sleep(0) # q.get() has installed its serving-loop Future
|
||||
started.set()
|
||||
try:
|
||||
received.append(await asyncio.wait_for(waiter, 0.5))
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
finally:
|
||||
done.set()
|
||||
await bus.unsubscribe(q)
|
||||
|
||||
def run_serving_loop():
|
||||
asyncio.set_event_loop(serving_loop)
|
||||
serving_loop.run_until_complete(serve_with_waiter_ready())
|
||||
|
||||
serving_thread = threading.Thread(
|
||||
target=run_serving_loop, name="test-serving-loop"
|
||||
)
|
||||
serving_thread.start()
|
||||
try:
|
||||
assert started.wait(2.0), "serving loop never subscribed"
|
||||
|
||||
async def foreign_async_caller():
|
||||
assert asyncio.get_running_loop() is not serving_loop
|
||||
bus.emit("profiles", {"action": "updated", "id": "foreign-loop"})
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
asyncio.run(foreign_async_caller())
|
||||
|
||||
assert done.wait(2.0), (
|
||||
"emit() ran listener delivery on the caller's foreign loop"
|
||||
)
|
||||
assert received, "foreign-loop event never reached the serving loop"
|
||||
payload = json.loads(received[0])
|
||||
assert payload["id"] == "foreign-loop"
|
||||
finally:
|
||||
serving_loop.call_soon_threadsafe(done.set)
|
||||
serving_thread.join(2.0)
|
||||
serving_loop.close()
|
||||
|
||||
|
||||
async def _serve(bus, received: list[str], started: threading.Event, done: threading.Event):
|
||||
q = await bus.subscribe()
|
||||
started.set()
|
||||
|
||||
@@ -210,7 +210,6 @@ def test_watermark_pool_rebuilds_after_shutdown_drain():
|
||||
assert get_watermark_pool().submit(lambda: "ok").result(timeout=5) == "ok"
|
||||
# Clean up the replacement so later tests start from a fresh pool too.
|
||||
shutdown_watermark_pool()
|
||||
begin_watermark_pool_lifecycle()
|
||||
|
||||
|
||||
def test_watermark_pool_shutdown_waits_for_active_worker():
|
||||
@@ -241,7 +240,6 @@ def test_watermark_pool_shutdown_waits_for_active_worker():
|
||||
release.set()
|
||||
shutdown_thread.join(timeout=2)
|
||||
assert shutdown_done.is_set()
|
||||
begin_watermark_pool_lifecycle()
|
||||
|
||||
|
||||
def test_watermark_pool_shutdown_deadline_bounds_stuck_worker():
|
||||
@@ -274,10 +272,9 @@ def test_watermark_pool_shutdown_deadline_bounds_stuck_worker():
|
||||
assert pool._thread is not None and pool._thread.daemon
|
||||
release.set()
|
||||
pool._thread.join(timeout=1)
|
||||
begin_watermark_pool_lifecycle()
|
||||
|
||||
|
||||
def test_watermark_pool_cannot_be_replaced_while_timed_out_worker_is_alive():
|
||||
def test_watermark_pool_cannot_be_replaced_while_timed_out_worker_is_alive(watermark):
|
||||
"""A producer racing bounded shutdown cannot create an undrained pool."""
|
||||
from services.model_manager import (
|
||||
begin_watermark_pool_lifecycle,
|
||||
@@ -306,6 +303,18 @@ def test_watermark_pool_cannot_be_replaced_while_timed_out_worker_is_alive():
|
||||
begin_watermark_pool_lifecycle()
|
||||
with pytest.raises(RuntimeError, match="shutting down"):
|
||||
get_watermark_pool()
|
||||
# Finished synthesis must still be returned unchanged: pool admission
|
||||
# is outside mark_synthetic's synchronous fail-open boundary.
|
||||
import asyncio
|
||||
import torch
|
||||
|
||||
audio = torch.zeros(1, 240)
|
||||
marked = asyncio.run(
|
||||
watermark.mark_synthetic_async(
|
||||
audio, 24000, context="test.restart_during_shutdown"
|
||||
)
|
||||
)
|
||||
assert marked is audio
|
||||
finally:
|
||||
release.set()
|
||||
pool._thread.join(timeout=1)
|
||||
@@ -316,7 +325,6 @@ def test_watermark_pool_cannot_be_replaced_while_timed_out_worker_is_alive():
|
||||
assert replacement is not pool
|
||||
assert replacement.submit(lambda: "ok").result(timeout=1) == "ok"
|
||||
shutdown_watermark_pool()
|
||||
begin_watermark_pool_lifecycle()
|
||||
|
||||
|
||||
def test_prefetched_model_gets_one_extra_idle_window(monkeypatch, watermark):
|
||||
|
||||
Reference in New Issue
Block a user