Параллельные задачи считаются в отдельных процессах

Замер подтвердил догадку: sherpa-onnx и onnxruntime держат GIL. Две задачи
в двух потоках идут ровно столько же, сколько подряд (1.04x у диаризации,
1.17x у распознавания) - отсюда и незагруженный процессор при работе.
В двух процессах те же задачи дают 1.59x даже с загрузкой моделей в каждом.

Пул процессов включается при workers > 1. Функции воркера вынесены в модуль,
не тянущий app.main: на Windows дочерний процесс поднимается через spawn и
иначе стартовал бы ещё один веб-сервер. Если процессы не запустятся, сервис
откатывается на однопроцессный режим.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Vladimir Bryzgalov
2026-08-15 23:27:07 +05:00
co-authored by Claude Opus 5
parent e2db164f0b
commit 41f3651cc8
5 changed files with 151 additions and 18 deletions
+49
View File
@@ -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("ожидалась ошибка о неинициализированном воркере")