diff --git a/app/config.py b/app/config.py index 6744090..b24464c 100644 --- a/app/config.py +++ b/app/config.py @@ -38,9 +38,10 @@ docs = false # Оптимум зависит от процессора: подберите его через POST /v1/benchmark # и впишите сюда полученное число. threads = 0 -# Сколько записей обрабатывать одновременно. Больше одного помогает редко: -# процессор и так загружен целиком, а каждый воркер держит свою копию -# моделей, это около 1 ГБ памяти. +# Сколько записей обрабатывать одновременно. Значение больше 1 включает +# отдельные процессы: библиотеки держат GIL, и в потоках задачи выстраиваются +# в очередь вместо параллельной работы. Каждый процесс держит свою копию +# моделей, это около 1 ГБ памяти на процесс. workers = 1 # Ожидаемое число говорящих в записи. 0 = определять автоматически # (на реальных звонках работает плохо, для диалога ставьте 2). diff --git a/app/main.py b/app/main.py index 908892c..bda65bc 100644 --- a/app/main.py +++ b/app/main.py @@ -42,18 +42,25 @@ _worker_stop = threading.Event() _state: dict = {"ready": False, "error": None} -def _process(job_id: str, worker: Pipeline) -> None: +def _process(job_id: str, worker: Pipeline, pool) -> None: """Задача уже помечена в работе тем, кто её забрал.""" job = store.get(job_id) if job is None: return upload = settings.data_dir / "uploads" / job_id try: - with tempfile.TemporaryDirectory() as tmp: - wav = Path(tmp) / "audio.wav" - to_wav16k(upload, wav, worker.ffmpeg) - speakers = int(job["options"].get("speakers", settings.speakers)) - result = worker.transcribe(wav, num_speakers=speakers) + speakers = int(job["options"].get("speakers", settings.speakers)) + if pool is not None: + # Считаем в отдельном процессе: библиотеки держат GIL, и в потоках + # задачи выстраиваются в очередь вместо параллельной работы. + from app.worker import run_job + + result = pool.submit(run_job, str(upload), speakers, worker.ffmpeg).result() + else: + with tempfile.TemporaryDirectory() as tmp: + wav = Path(tmp) / "audio.wav" + to_wav16k(upload, wav, worker.ffmpeg) + result = worker.transcribe(wav, num_speakers=speakers) result["filename"] = job["filename"] store.mark_done(job_id, result) log.info("задача %s готова: %.1f с аудио, x%s", job_id, @@ -65,7 +72,7 @@ def _process(job_id: str, worker: Pipeline) -> None: upload.unlink(missing_ok=True) -def _worker_loop(index: int, worker: Pipeline) -> None: +def _worker_loop(index: int, worker: Pipeline, pool=None) -> None: """Разбирает очередь. Задача захватывается атомарно, поэтому воркеров может быть много.""" last_cleanup = 0.0 while not _worker_stop.is_set(): @@ -73,7 +80,7 @@ def _worker_loop(index: int, worker: Pipeline) -> None: job_id = store.claim_next() if job_id: log.info("воркер %d взял задачу %s", index, job_id) - _process(job_id, worker) + _process(job_id, worker, pool) continue # Уборкой занимается только первый воркер, чтобы не делать её хором. if index == 0 and time.time() - last_cleanup > 3600: @@ -84,6 +91,31 @@ def _worker_loop(index: int, worker: Pipeline) -> None: _worker_stop.wait(1.0) +def _make_pool(count: int): + """Пул процессов для расчётов. При неудаче работаем в одном процессе. + + На Windows процессы поднимаются через spawn, и сбои тут возможны, поэтому + отказ не должен ронять сервис - он просто станет однопроцессным. + """ + from concurrent.futures import ProcessPoolExecutor + + from app.worker import init_worker + + try: + pool = ProcessPoolExecutor( + max_workers=count, + initializer=init_worker, + initargs=(str(settings.models_dir), settings.effective_threads(), + str(settings.replacements_path), str(settings.base_dir)), + ) + # Пустая задача проверяет, что процессы действительно поднялись + pool.submit(str, "ok").result(timeout=300) + return pool + except Exception as exc: # noqa: BLE001 - причин может быть много, важен откат + log.warning("не удалось запустить процессы (%s), работаю в одном", exc) + return None + + @asynccontextmanager async def lifespan(app: FastAPI): (settings.data_dir / "uploads").mkdir(parents=True, exist_ok=True) @@ -93,22 +125,28 @@ async def lifespan(app: FastAPI): except ModelsMissing as exc: _state["error"] = str(exc) log.error("сервис запущен без моделей: %s", exc) + count = settings.effective_workers() + pool = None + if count > 1 and _state["ready"]: + pool = _make_pool(count) + workers = [] - for i in range(settings.effective_workers()): - # Первый воркер использует уже прогретый конвейер, остальные греются сами - # при первой задаче: держать копии моделей впустую незачем. + for i in range(count): worker = pipeline if i == 0 else make_pipeline() - thread = threading.Thread(target=_worker_loop, args=(i, worker), + thread = threading.Thread(target=_worker_loop, args=(i, worker, pool), name=f"asr-worker-{i}", daemon=True) thread.start() workers.append(thread) - log.info("сервис слушает %s:%s, воркеров %d по %d потоков", settings.host, - settings.port, settings.effective_workers(), settings.effective_threads()) + log.info("сервис слушает %s:%s, воркеров %d по %d потоков, режим %s", + settings.host, settings.port, count, settings.effective_threads(), + "процессы" if pool else "один процесс") yield _worker_stop.set() for thread in workers: thread.join(timeout=5) + if pool is not None: + pool.shutdown(wait=False, cancel_futures=True) # Штатные /docs и /openapi.json отключены: они не требуют токена. Вместо них diff --git a/app/version.py b/app/version.py index 3d26edf..3d18726 100644 --- a/app/version.py +++ b/app/version.py @@ -1 +1 @@ -__version__ = "0.4.1" +__version__ = "0.5.0" diff --git a/app/worker.py b/app/worker.py new file mode 100644 index 0000000..0e68231 --- /dev/null +++ b/app/worker.py @@ -0,0 +1,45 @@ +"""Обработка задачи в отдельном процессе. + +Замеры показали, что sherpa-onnx и onnxruntime держат GIL: две задачи в двух +потоках идут ровно столько же, сколько подряд (выигрыш 1.04x). Поэтому +процессор простаивает, а воркеры-потоки бесполезны. В процессах те же две +задачи дают 1.59x даже с загрузкой моделей в каждом. + +Модуль намеренно не импортирует app.main: на Windows дочерний процесс +поднимается через spawn и заново импортирует модуль с функцией. Если бы +это был main, в каждом процессе поднимался бы ещё один веб-сервер. +""" +import logging +from pathlib import Path + +from app.pipeline import Pipeline + +log = logging.getLogger(__name__) + +# Живёт внутри процесса-воркера: модели грузятся один раз на процесс. +_pipeline: Pipeline | None = None + + +def init_worker(models_dir: str, threads: int, replacements_path: str, base_dir: str) -> None: + """Готовит процесс к работе: грузит модели один раз.""" + global _pipeline + logging.basicConfig(level=logging.WARNING) + _pipeline = Pipeline(models_dir=Path(models_dir), threads=threads, + replacements_path=Path(replacements_path), + base_dir=Path(base_dir)) + _pipeline.warmup() + + +def run_job(audio_path: str, num_speakers: int, ffmpeg: str) -> dict: + """Переводит файл в WAV и распознаёт. Выполняется в процессе-воркере.""" + import tempfile + + from app.pipeline import to_wav16k + + if _pipeline is None: + raise RuntimeError("процесс-воркер не инициализирован") + + with tempfile.TemporaryDirectory() as tmp: + wav = Path(tmp) / "audio.wav" + to_wav16k(Path(audio_path), wav, ffmpeg) + return _pipeline.transcribe(wav, num_speakers=num_speakers) diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index b37a033..4540e9c 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -208,3 +208,52 @@ class TestSeparationQuality: loud = (np.random.RandomState(0).randn(16000) * 0.3).astype(np.float32) quiet = loud * 0.1 assert segment_acoustics(loud)["loudness_db"] > segment_acoustics(quiet)["loudness_db"] + + +class TestProcessPool: + """Расчёты уходят в процессы: библиотеки держат GIL, в потоках выигрыша нет.""" + + def test_worker_module_does_not_import_main(self): + """На Windows дочерний процесс заново импортирует модуль с функцией. + + Если бы он тянул app.main, в каждом процессе поднимался бы ещё один + веб-сервер со своей очередью. Проверяем сами импорты, а не текст файла. + """ + import ast + from pathlib import Path + + src = (Path(__file__).resolve().parent.parent / "app" / "worker.py").read_text(encoding="utf-8") + imported = set() + for node in ast.walk(ast.parse(src)): + if isinstance(node, ast.Import): + imported.update(a.name for a in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + imported.add(node.module) + assert "app.main" not in imported + + def test_pool_failure_falls_back_to_single_process(self, monkeypatch): + """Сбой процессов не должен ронять сервис.""" + import sys + import types + + for name in ("sherpa_onnx", "onnx_asr", "onnxruntime"): + monkeypatch.setitem(sys.modules, name, types.ModuleType(name)) + for mod in [m for m in list(sys.modules) if m.startswith("app.")]: + del sys.modules[mod] + + import app.main as main + + monkeypatch.setattr("concurrent.futures.ProcessPoolExecutor", + lambda **kw: (_ for _ in ()).throw(OSError("нет процессов"))) + assert main._make_pool(4) is None + + def test_run_job_needs_initialised_worker(self): + import app.worker as w + + w._pipeline = None + try: + w.run_job("нет.wav", 2, "ffmpeg") + except RuntimeError as exc: + assert "не инициализирован" in str(exc) + else: + raise AssertionError("ожидалась ошибка о неинициализированном воркере")