diff --git a/backend/api/routers/dub_generate.py b/backend/api/routers/dub_generate.py index a9d9e9ba..1dbfa693 100644 --- a/backend/api/routers/dub_generate.py +++ b/backend/api/routers/dub_generate.py @@ -16,7 +16,7 @@ from schemas.requests import DubRequest from services.model_manager import get_model, _gpu_pool from services.audio_dsp import apply_mastering, normalize_audio, apply_effects_chain, get_effect_chain from services.audio_io import atomic_save_wav, _safe_torchaudio_save -from services.ffmpeg_utils import find_ffmpeg +from services.ffmpeg_utils import find_ffmpeg, spawn_subprocess from services.rvc import apply_rvc, is_enabled as rvc_is_enabled from services.incremental import segment_fingerprint from services.watermark import embed_watermark @@ -86,7 +86,7 @@ async def _pitch_preserving_stretch( # Mono float32 via stdin → ffmpeg → stdout. One subprocess per segment, # run off the event loop so concurrent requests stay responsive. arr = wav.detach().cpu().to(torch.float32).numpy().reshape(-1).astype(np.float32, copy=False) - proc = await asyncio.create_subprocess_exec( + proc = await spawn_subprocess( find_ffmpeg(), "-hide_banner", "-loglevel", "error", "-y", "-f", "f32le", "-ar", str(sr), "-ac", "1", "-i", "pipe:0", "-af", filter_str, diff --git a/backend/api/routers/gallery.py b/backend/api/routers/gallery.py index c235796d..a2c3a203 100644 --- a/backend/api/routers/gallery.py +++ b/backend/api/routers/gallery.py @@ -14,6 +14,7 @@ from pydantic import BaseModel from core.db import db_conn from core.config import VOICES_DIR, OUTPUTS_DIR from core import event_bus +from services.ffmpeg_utils import spawn_subprocess logger = logging.getLogger("omnivoice.gallery") @@ -199,7 +200,7 @@ async def search_youtube( ): """Search YouTube for character/celebrity clips using yt-dlp.""" try: - result = await asyncio.create_subprocess_exec( + result = await spawn_subprocess( "yt-dlp", "--dump-json", "--remote-components", "ejs:github", @@ -273,7 +274,7 @@ async def download_youtube_clip( video_url, ] - result = await asyncio.create_subprocess_exec( + result = await spawn_subprocess( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, diff --git a/backend/api/routers/tools.py b/backend/api/routers/tools.py index 7e95d02c..7af6017f 100644 --- a/backend/api/routers/tools.py +++ b/backend/api/routers/tools.py @@ -27,7 +27,7 @@ from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field from services import director, speech_rate, incremental -from services.ffmpeg_utils import find_ffmpeg, find_ffprobe +from services.ffmpeg_utils import find_ffmpeg, find_ffprobe, spawn_subprocess logger = logging.getLogger("omnivoice.tools") router = APIRouter() @@ -54,7 +54,7 @@ async def probe(req: ProbeReq): status_code=501, detail="ffprobe binary not available. Install system ffmpeg or re-run the setup.", ) - proc = await asyncio.create_subprocess_exec( + proc = await spawn_subprocess( ffprobe, "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", target, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, diff --git a/backend/services/ffmpeg_utils.py b/backend/services/ffmpeg_utils.py index 6dee08d1..d4782104 100644 --- a/backend/services/ffmpeg_utils.py +++ b/backend/services/ffmpeg_utils.py @@ -136,6 +136,7 @@ async def _spawn_thread_fallback(cmd, **kwargs): stdout=subprocess.PIPE if stdout == asyncio.subprocess.PIPE else stdout, stderr=subprocess.PIPE if stderr == asyncio.subprocess.PIPE else stderr, stdin=subprocess.PIPE if stdin == asyncio.subprocess.PIPE else stdin, + **kwargs, # forward cwd / env / etc. so the fallback matches the async call ) proc = await loop.run_in_executor(None, _run) @@ -166,6 +167,20 @@ async def _spawn_thread_fallback(cmd, **kwargs): return _AsyncCompatProc(proc) +async def spawn_subprocess(*args, **kwargs): + """Drop-in replacement for ``asyncio.create_subprocess_exec``. + + Falls back to a thread-based ``subprocess.Popen`` (wrapped to match the + asyncio Process interface) on event loops without subprocess support — + notably the Windows ``SelectorEventLoop`` that uvicorn forces under + ``--reload``/multi-worker (``use_subprocess=True``), where the native call + raises ``NotImplementedError`` (GH #122). Also inherits the EAGAIN retry. + On loops that DO support subprocesses (Proactor, posix) the native path is + used unchanged, so there is no behavior change off the broken loop. + """ + return await _spawn_with_retry(list(args), **kwargs) + + async def _spawn_with_retry(cmd, **kwargs): """Spawn a subprocess, retrying briefly on EAGAIN (posix_spawn resource pressure).""" delay = 0.1 diff --git a/backend/services/sonitranslate.py b/backend/services/sonitranslate.py index 0453596a..d80adaef 100644 --- a/backend/services/sonitranslate.py +++ b/backend/services/sonitranslate.py @@ -15,6 +15,8 @@ import sys from pathlib import Path from typing import Optional +from services.ffmpeg_utils import spawn_subprocess + logger = logging.getLogger("omnivoice.sonitranslate") # Default install location — inside the OmniVoice project tree @@ -73,7 +75,7 @@ async def install(progress_callback=None) -> dict: logger.info("Cloning SoniTranslate...") if progress_callback: progress_callback("Cloning SoniTranslate repository...") - proc = await asyncio.create_subprocess_exec( + proc = await spawn_subprocess( "git", "clone", "--depth", "1", "https://github.com/R3gm/SoniTranslate.git", str(SONI_DIR), @@ -91,7 +93,7 @@ async def install(progress_callback=None) -> dict: progress_callback("Creating virtualenv...") python = sys.executable - proc = await asyncio.create_subprocess_exec( + proc = await spawn_subprocess( python, "-m", "venv", str(SONI_VENV), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, @@ -103,7 +105,7 @@ async def install(progress_callback=None) -> dict: if progress_callback: progress_callback("Installing base requirements (this may take a while)...") - proc = await asyncio.create_subprocess_exec( + proc = await spawn_subprocess( pip, "install", "-r", str(SONI_DIR / "requirements_base.txt"), cwd=str(SONI_DIR), stdout=asyncio.subprocess.PIPE, @@ -116,7 +118,7 @@ async def install(progress_callback=None) -> dict: # Install extra requirements if progress_callback: progress_callback("Installing extra requirements...") - proc = await asyncio.create_subprocess_exec( + proc = await spawn_subprocess( pip, "install", "-r", str(SONI_DIR / "requirements_extra.txt"), cwd=str(SONI_DIR), stdout=asyncio.subprocess.PIPE, diff --git a/tests/test_subprocess_fallback.py b/tests/test_subprocess_fallback.py new file mode 100644 index 00000000..0791b635 --- /dev/null +++ b/tests/test_subprocess_fallback.py @@ -0,0 +1,87 @@ +"""Windows --reload subprocess regression (issue #122 "Extract: Unknown Error"). + +uvicorn forces the SelectorEventLoop on Windows when use_subprocess=True +(`--reload` / multi-worker), where `asyncio.create_subprocess_exec` raises +NotImplementedError. `spawn_subprocess` must transparently fall back to a +thread-based subprocess so ffmpeg/yt-dlp/ffprobe spawns still work. +""" +from __future__ import annotations + +import asyncio +import os +import sys + +from services import ffmpeg_utils + + +def _force_notimplemented(monkeypatch): + async def _boom(*a, **k): + raise NotImplementedError("no subprocess on this loop (simulated Windows SelectorEventLoop)") + monkeypatch.setattr(ffmpeg_utils.asyncio, "create_subprocess_exec", _boom) + + +def test_spawn_subprocess_falls_back_to_thread_on_notimplemented(monkeypatch): + _force_notimplemented(monkeypatch) + + async def run(): + proc = await ffmpeg_utils.spawn_subprocess( + sys.executable, "-c", "import sys; sys.stdout.write('ok')", + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + out, _ = await proc.communicate() + return proc.returncode, out + + rc, out = asyncio.run(run()) + assert rc == 0 + assert out == b"ok" + + +def test_spawn_subprocess_fallback_forwards_cwd(monkeypatch, tmp_path): + # sonitranslate's pip install passes cwd= — the thread fallback must honor it. + _force_notimplemented(monkeypatch) + + async def run(): + proc = await ffmpeg_utils.spawn_subprocess( + sys.executable, "-c", "import os,sys; sys.stdout.write(os.getcwd())", + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + cwd=str(tmp_path), + ) + out, _ = await proc.communicate() + return out + + out = asyncio.run(run()) + assert os.path.realpath(out.decode()) == os.path.realpath(str(tmp_path)) + + +def test_spawn_subprocess_fallback_passes_stdin_input(monkeypatch): + # dub_generate's atempo pipes audio bytes via stdin → communicate(input=...). + _force_notimplemented(monkeypatch) + + async def run(): + proc = await ffmpeg_utils.spawn_subprocess( + sys.executable, "-c", "import sys; sys.stdout.buffer.write(sys.stdin.buffer.read())", + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + out, _ = await proc.communicate(input=b"abc123") + return proc.returncode, out + + rc, out = asyncio.run(run()) + assert rc == 0 + assert out == b"abc123" + + +def test_spawn_subprocess_native_path_when_loop_supports_it(): + # On a loop WITH subprocess support (posix / Windows Proactor) the native + # asyncio path is used unchanged — no behavior change off the broken loop. + async def run(): + proc = await ffmpeg_utils.spawn_subprocess( + sys.executable, "-c", "print('hi')", + stdout=asyncio.subprocess.PIPE, + ) + out, _ = await proc.communicate() + return proc.returncode, out + + rc, out = asyncio.run(run()) + assert rc == 0 + assert b"hi" in out