From 584dfb58c7e6b611cb40b49fa220da8c1eff09f9 Mon Sep 17 00:00:00 2001 From: Ark Deliev Date: Fri, 28 Aug 2026 00:46:52 +0000 Subject: [PATCH 1/2] perf: budget transcribe admission in audio-minutes, not video count The long-video gate serialised every video over 45 minutes so that only one decoded at a time. The current backlog is entirely ~51-minute videos, so all four workers queue behind a single slot and throughput collapses to 1x: 1575 translations at ~9 min each projects to ~10 days. Measured peak RSS of a translate pass (large-v3, float16, 16 kHz mono) by transcribing one source at five lengths and reading ru_maxrss: 15 min 3.21 GB 30 min 3.21 GB 51 min 3.25 GB 103 min 6.06 GB 205 min 11.68 GB That fits peak = 0.44 GB + 0.0548 GB per audio-minute, and predicts the 51 min point at 3.25 GB against 3.25 GB measured. Below ~50 minutes the one-off model-load spike (~3.2 GB) sits above the line and masks it, which is why short videos all look equally cheap. So the constraint is total audio-minutes decoding at once, not how long any one video is. A count-based gate cannot express that: it either serialises videos that would fit together comfortably, or lets a multi-hour video run alongside three others. The kernel OOM on 2026-08-01 was the latter -- 39:06, 1:12:35, 1:20:21 and 6:28:57 in flight, anon-rss 31.4 GB. Replace the gate with AudioMinutesBudget: workers charge a job's duration before decoding and refund it after. Admission is strictly FIFO, because a long video would otherwise starve -- it waits for most of the budget while short videos behind it keep fitting into the slack it is accumulating. A video longer than the whole budget is charged the full budget rather than refused, so it decodes alone instead of deadlocking. Default budget of 220 minutes is sized against the ~13.5 GB service floor (steady RSS with all workers warm, 473 samples): four ~51-minute videos at once, projected peak ~23 GB, ~8 GB headroom. Restores 4x on the current backlog. Tunable with --transcribe-minutes-budget, replacing --long-video-minutes. This does not help the 6.5-hour outlier, which projects to ~21.7 GB on top of the floor and exceeds the box running completely alone -- under the old gate equally. That needs chunked decoding or a permanent skip, tracked separately. --- src/python/tools/archive_transcriber.py | 176 ++++++++++++++++++------ tests/test_archive_transcriber.py | 146 ++++++++++++++------ 2 files changed, 239 insertions(+), 83 deletions(-) diff --git a/src/python/tools/archive_transcriber.py b/src/python/tools/archive_transcriber.py index 6e928d9..0cd50da 100644 --- a/src/python/tools/archive_transcriber.py +++ b/src/python/tools/archive_transcriber.py @@ -135,14 +135,32 @@ def signal_handler(signum: int, frame: Optional[FrameType]) -> None: # CPU-bound; a wide thread pool turns hours of sequential stats into minutes. STARTUP_STAT_THREADS = 64 -# Peak RSS during model.transcribe() grows with audio duration (~3 GB for a -# 27-minute video on faster-whisper 1.2.1); four workers on multi-hour videos -# exceeded the 31 GB of this box (no swap) and got the service OOM-killed -# daily. Videos above the --long-video-minutes threshold therefore hold this -# semaphore so only one of them transcribes at a time. -# Bounded so an unbalanced release() raises instead of silently widening the -# gate and letting two long videos transcribe at once. -LONG_VIDEO_GATE = threading.BoundedSemaphore(1) +# Peak RSS of a transcribe is dominated by audio decoding, not by the model, +# and grows linearly with duration. Measured on this box (faster-whisper 1.2.1, +# large-v3, float16, 16 kHz mono) by transcribing the same source at five +# lengths and reading ru_maxrss: +# +# 15 min 3.21 GB 103 min 6.06 GB +# 30 min 3.21 GB 205 min 11.68 GB +# 51 min 3.25 GB +# +# which fits peak = 0.44 GB + 0.0548 GB per audio-minute. Below ~50 minutes the +# one-off model-load spike (~3.2 GB) sits above that line and masks it, which is +# why short videos all cost the same. +# +# Admission is therefore budgeted in audio-minutes in flight rather than in +# video count: several ordinary videos decode in parallel, while one multi-hour +# video consumes the whole budget and runs alone. A count-based gate cannot +# express that -- it either serialises videos that would comfortably fit +# together, or lets a multi-hour video run alongside three others and blows the +# 31 GB of this box (no swap), which is what OOM-killed the service daily. +# +# Sizing the default: the service floor is ~13.5 GB (steady RSS with all workers +# warm, sampled 473 times), and each extra concurrent decode adds +# 0.44 + 0.0548 * minutes. 220 minutes admits four ~51-minute videos at once -- +# what the current backlog looks like -- for a projected peak near 23 GB, which +# leaves ~8 GB of headroom. Raise it only against a re-measured floor. +TRANSCRIBE_MINUTES_BUDGET = 220.0 # Error types that will not succeed on retry while the input is unchanged # (e.g. FFmpeg cannot extract audio from a corrupt container). Jobs whose @@ -151,6 +169,76 @@ def signal_handler(signum: int, frame: Optional[FrameType]) -> None: PERMANENT_ERROR_TYPES = {"audio_extraction"} +class AudioMinutesBudget: + """Admission control on the total audio-minutes being decoded at once. + + Workers charge a job's duration against a fixed budget before transcribing + and refund it afterwards, so concurrency adapts to how expensive the jobs + actually are instead of being a fixed worker count. + + Admission is strictly FIFO. A long video would otherwise starve + indefinitely: it needs most of the budget, while short videos behind it + keep fitting into the slack it is waiting to accumulate. Head-of-line + blocking is the point here, not a side effect. + """ + + def __init__(self, budget_minutes: float) -> None: + if budget_minutes <= 0: + raise ValueError("budget_minutes must be positive") + self._budget = budget_minutes + self._available = budget_minutes + self._cond = threading.Condition() + self._next_ticket = 0 + self._now_serving = 0 + + @property + def budget_minutes(self) -> float: + return self._budget + + def acquire(self, minutes: Optional[float]) -> float: + """Block until `minutes` of budget is free; return the amount charged. + + A video longer than the entire budget is charged the whole budget + rather than being refused, so it runs alone instead of deadlocking. + An unknown duration is charged nothing -- the caller could not have + probed it, and refusing to run it would strand the job forever. + """ + cost = min(max(minutes or 0.0, 0.0), self._budget) + with self._cond: + ticket = self._next_ticket + self._next_ticket += 1 + while ticket != self._now_serving or cost > self._available: + self._cond.wait() + self._now_serving += 1 + self._available -= cost + self._cond.notify_all() + return cost + + def release(self, cost: float) -> None: + with self._cond: + self._available += cost + if self._available > self._budget: + # Mirrors BoundedSemaphore: an unbalanced release must fail + # loudly rather than silently widen the budget. + self._available -= cost + raise ValueError("released more budget than was acquired") + self._cond.notify_all() + + +_BUDGET_LOCK = threading.Lock() +_BUDGET: Optional[AudioMinutesBudget] = None + + +def get_transcribe_budget(args: Any) -> AudioMinutesBudget: + """Return the process-wide admission budget, built once from args.""" + global _BUDGET + with _BUDGET_LOCK: + if _BUDGET is None: + configured = float(getattr(args, "transcribe_minutes_budget", 0) or 0) + _BUDGET = AudioMinutesBudget(configured or TRANSCRIBE_MINUTES_BUDGET) + return _BUDGET + + class SegmentLike(Protocol): """Protocol for transcription segment objects.""" @@ -1226,9 +1314,9 @@ def sort_jobs_by_size(jobs: List[VideoJob]) -> List[VideoJob]: """Order jobs smallest video first (size is a cheap NFS-side proxy for duration). Concurrent workers then always hold similarly-sized jobs, which bounds - their combined transcribe peak RSS, and the multi-hour videos that need - the LONG_VIDEO_GATE run last — after the bulk of the queue has completed — - instead of monopolizing the head of every cycle. + their combined transcribe peak RSS, and the multi-hour videos that consume + the whole admission budget run last — after the bulk of the queue has + completed — instead of monopolizing the head of every cycle. """ def _size(job: VideoJob) -> int: @@ -1247,22 +1335,22 @@ def _size(job: VideoJob) -> int: return [jobs[i] for i in order] -def acquire_long_video_slot(duration: Optional[float], threshold_minutes: float, video_path: Path) -> bool: - """Block until this video may transcribe; True if the exclusive long-video slot was taken. +def acquire_transcribe_budget(duration: Optional[float], budget: AudioMinutesBudget, video_path: Path) -> float: + """Block until this video may decode; return the audio-minutes charged. - The caller must release LONG_VIDEO_GATE when done iff this returns True. - Videos shorter than the threshold (or an unknown duration, or a disabled - threshold <= 0) never wait and never take the slot. + The caller must release the same amount when done. A duration of None + (unprobeable container) is charged nothing so the job still runs and + reports its real error. """ - if threshold_minutes <= 0 or not duration or duration < threshold_minutes * 60: - return False - LOGGER.info( - "[LongVideo] %s is %.0f min long; waiting for exclusive long-video slot", - video_path, - duration / 60, - ) - LONG_VIDEO_GATE.acquire() - return True + minutes = (duration or 0.0) / 60 + if minutes >= budget.budget_minutes: + LOGGER.info( + "[Budget] %s is %.0f min long; it exceeds the %.0f min budget and will decode alone", + video_path, + minutes, + budget.budget_minutes, + ) + return budget.acquire(minutes) def extract_audio(video_path: Path, sample_rate: int) -> Path: @@ -1641,16 +1729,15 @@ def process_transcription_only(job: VideoJob, args: argparse.Namespace, quiet: b filter_words: List[str] = load_filter_words() audio_path: Optional[Path] = None - holds_long_slot = False - long_video_minutes = float(getattr(args, "long_video_minutes", 0) or 0) + budget = get_transcribe_budget(args) + charged = 0.0 try: - if long_video_minutes > 0: - try: - duration_hint = probe_video_metadata(job.video_path).duration - except Exception: - duration_hint = None # broken container: extract_audio below reports the real error - holds_long_slot = acquire_long_video_slot(duration_hint, long_video_minutes, job.video_path) + try: + duration_hint = probe_video_metadata(job.video_path).duration + except Exception: + duration_hint = None # broken container: extract_audio below reports the real error + charged = acquire_transcribe_budget(duration_hint, budget, job.video_path) audio_path = extract_audio(job.video_path, args.sample_rate) LOGGER.debug("[Transcription] Loading model %s...", args.model) @@ -1719,8 +1806,8 @@ def process_transcription_only(job: VideoJob, args: argparse.Namespace, quiet: b } finally: - if holds_long_slot: - LONG_VIDEO_GATE.release() + if charged: + budget.release(charged) if audio_path and audio_path.exists(): try: audio_path.unlink() @@ -1744,14 +1831,14 @@ def process_translation_only( filter_words: List[str] = load_filter_words() audio_path: Optional[Path] = None - holds_long_slot = False - long_video_minutes = float(getattr(args, "long_video_minutes", 0) or 0) + budget = get_transcribe_budget(args) + charged = 0.0 try: # Probe inside the try: a corrupt container must produce an error # record, not an exception that escapes the worker thread. metadata = probe_video_metadata(job.video_path) - holds_long_slot = acquire_long_video_slot(metadata.duration, long_video_minutes, job.video_path) + charged = acquire_transcribe_budget(metadata.duration, budget, job.video_path) # Read existing Russian VTT for segment count comparison ru_content = job.ru_vtt.read_text(encoding="utf-8") if job.ru_vtt.exists() else "" @@ -1874,8 +1961,8 @@ def __init__(self, text: str) -> None: return error_record finally: - if holds_long_slot: - LONG_VIDEO_GATE.release() + if charged: + budget.release(charged) if audio_path and audio_path.exists(): try: audio_path.unlink() @@ -1980,12 +2067,13 @@ def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: ) parser.add_argument("--workers", type=int, default=1, help="Number of worker threads for processing") parser.add_argument( - "--long-video-minutes", + "--transcribe-minutes-budget", type=float, - default=45.0, + default=TRANSCRIBE_MINUTES_BUDGET, help=( - "Videos longer than this many minutes are transcribed one at a time " - "to bound peak RAM (transcribe RSS grows with duration); 0 disables the gate" + "Total audio-minutes allowed to decode concurrently across all workers. " + "Peak RAM is roughly 0.44 GB + 0.055 GB per audio-minute in flight, so this " + "bounds peak RSS; a video longer than the budget decodes alone" ), ) parser.add_argument( diff --git a/tests/test_archive_transcriber.py b/tests/test_archive_transcriber.py index 24d5084..f714ce6 100644 --- a/tests/test_archive_transcriber.py +++ b/tests/test_archive_transcriber.py @@ -6,6 +6,8 @@ import re import sys import tempfile +import threading +import time import typing as _typing from pathlib import Path from typing import Callable, Optional @@ -691,46 +693,112 @@ def test_empty_list(self): assert archive_transcriber.sort_jobs_by_size([]) == [] -class TestAcquireLongVideoSlot: - """Only videos over the threshold take the exclusive slot; others never wait.""" - - def _release_if(self, acquired: bool) -> None: - if acquired: - archive_transcriber.LONG_VIDEO_GATE.release() - - def test_short_video_does_not_take_slot(self): - acquired = archive_transcriber.acquire_long_video_slot(10 * 60, 45.0, Path("v.mp4")) - self._release_if(acquired) - assert acquired is False - - def test_long_video_takes_and_holds_slot(self): - acquired = archive_transcriber.acquire_long_video_slot(120 * 60, 45.0, Path("v.mp4")) - try: - assert acquired is True - # Slot is exclusive while held - assert archive_transcriber.LONG_VIDEO_GATE.acquire(blocking=False) is False - finally: - self._release_if(acquired) - # And free again after release - assert archive_transcriber.LONG_VIDEO_GATE.acquire(blocking=False) is True - archive_transcriber.LONG_VIDEO_GATE.release() - - def test_disabled_threshold_never_gates(self): - acquired = archive_transcriber.acquire_long_video_slot(999 * 60, 0, Path("v.mp4")) - self._release_if(acquired) - assert acquired is False - - def test_unknown_duration_never_gates(self): - for duration in (None, 0.0): - acquired = archive_transcriber.acquire_long_video_slot(duration, 45.0, Path("v.mp4")) - self._release_if(acquired) - assert acquired is False +class TestAudioMinutesBudget: + """Admission is bounded by audio-minutes in flight, not by video count.""" + + def test_videos_fitting_the_budget_run_concurrently(self): + budget = archive_transcriber.AudioMinutesBudget(200.0) + charged = [budget.acquire(50.0) for _ in range(4)] + # Four 50-minute videos fit; none of the acquires blocked. + assert charged == [50.0, 50.0, 50.0, 50.0] + + def test_acquire_blocks_once_the_budget_is_spent(self): + budget = archive_transcriber.AudioMinutesBudget(100.0) + budget.acquire(80.0) + admitted = threading.Event() + + def _second() -> None: + budget.acquire(40.0) + admitted.set() + + t = threading.Thread(target=_second, daemon=True) + t.start() + # 80 + 40 > 100, so the second video waits rather than overcommitting. + assert admitted.wait(timeout=0.2) is False + budget.release(80.0) + assert admitted.wait(timeout=2.0) is True + t.join(timeout=2.0) + + def test_video_longer_than_budget_is_capped_and_runs_alone(self): + budget = archive_transcriber.AudioMinutesBudget(200.0) + # A 6.5-hour video must not deadlock waiting for budget it can never get. + charged = budget.acquire(389.0) + assert charged == 200.0 + + admitted = threading.Event() + + def _other() -> None: + budget.acquire(1.0) + admitted.set() + + t = threading.Thread(target=_other, daemon=True) + t.start() + # ...but while it holds the whole budget, nothing else decodes. + assert admitted.wait(timeout=0.2) is False + budget.release(charged) + assert admitted.wait(timeout=2.0) is True + t.join(timeout=2.0) + + def test_unknown_duration_is_charged_nothing(self): + budget = archive_transcriber.AudioMinutesBudget(200.0) + assert budget.acquire(None) == 0.0 + assert budget.acquire(0.0) == 0.0 + + def test_admission_is_fifo_so_long_videos_cannot_starve(self): + budget = archive_transcriber.AudioMinutesBudget(100.0) + budget.acquire(60.0) + + order: list[str] = [] + big_waiting = threading.Event() + big_done = threading.Event() + small_done = threading.Event() + + def _big() -> None: + big_waiting.set() + budget.acquire(100.0) + order.append("big") + big_done.set() + + def _small() -> None: + budget.acquire(10.0) + order.append("small") + small_done.set() + + big = threading.Thread(target=_big, daemon=True) + big.start() + big_waiting.wait(timeout=1.0) + time.sleep(0.05) # let the big video take its ticket first + + small = threading.Thread(target=_small, daemon=True) + small.start() + # 10 minutes would fit in the 40 still free, but the big video queued + # first; letting the small one skip ahead is how long videos starve. + assert small_done.wait(timeout=0.2) is False + + budget.release(60.0) + assert big_done.wait(timeout=2.0) is True + budget.release(100.0) + assert small_done.wait(timeout=2.0) is True + big.join(timeout=2.0) + small.join(timeout=2.0) + assert order == ["big", "small"] def test_unbalanced_release_fails_loudly(self): - # BoundedSemaphore: an over-release must raise rather than silently - # widening the gate to allow two concurrent long videos. + # An over-release must raise rather than silently widening the budget. + budget = archive_transcriber.AudioMinutesBudget(200.0) with pytest.raises(ValueError): - archive_transcriber.LONG_VIDEO_GATE.release() + budget.release(1.0) + + def test_failed_release_leaves_the_budget_intact(self): + budget = archive_transcriber.AudioMinutesBudget(200.0) + with pytest.raises(ValueError): + budget.release(1.0) + # The rejected refund must not have been applied. + assert budget.acquire(200.0) == 200.0 + + def test_rejects_non_positive_budget(self): + with pytest.raises(ValueError): + archive_transcriber.AudioMinutesBudget(0) class TestTwoPhaseQueueing: @@ -746,7 +814,7 @@ def _args(self, tmp: Path): workers=1, progress=False, verbose=False, - long_video_minutes=0, + transcribe_minutes_budget=1000.0, ) def _make_video(self, tmp: Path, stem: str, *, ru=False, en=False, smil=False, stale_ru=False): @@ -836,7 +904,7 @@ def _args(self): vad_filter=True, trim_silence=False, verbose=False, - long_video_minutes=0, + transcribe_minutes_budget=1000.0, ) def test_ffmpeg_failure_marked_permanent(self, monkeypatch): From 0affd67c92867a08c3333e12aa5cfe18dadb8d8f Mon Sep 17 00:00:00 2001 From: Ark Deliev Date: Fri, 28 Aug 2026 00:56:06 +0000 Subject: [PATCH 2/2] review: tolerate float drift when refunding budget Charges are duration/60, and workers finish in a different order than they started, so refunding out of order leaves _available a few ulps above the budget -- 220.00000000000003 for three charges. A strict > comparison rejects that: release() raises from the worker's finally block, the manifest record is lost, and future.result() fails the whole phase. It is not rare. Over 200k random 2-4 charge sequences with shuffled refund order, 13756 (~7%) overshoot. Compare against a 1e-6 tolerance and clamp to the budget instead of reverting. Clamping on the last outstanding refund snaps back to exact, so rounding cannot accumulate across a long run, and 1e-6 stays far below a genuine double-release (off by a whole video's worth) -- covered by a test. --- src/python/tools/archive_transcriber.py | 21 ++++++++++++++++----- tests/test_archive_transcriber.py | 24 ++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/src/python/tools/archive_transcriber.py b/src/python/tools/archive_transcriber.py index 0cd50da..c259597 100644 --- a/src/python/tools/archive_transcriber.py +++ b/src/python/tools/archive_transcriber.py @@ -168,6 +168,10 @@ def signal_handler(signum: int, frame: Optional[FrameType]) -> None: # excluded from Phase 1 instead of failing again every cycle. PERMANENT_ERROR_TYPES = {"audio_extraction"} +# Slack allowed when refunding budget, to absorb float rounding without letting +# a genuine double-release (which is off by a whole video's worth) slip through. +_BUDGET_EPSILON = 1e-6 + class AudioMinutesBudget: """Admission control on the total audio-minutes being decoded at once. @@ -216,12 +220,19 @@ def acquire(self, minutes: Optional[float]) -> float: def release(self, cost: float) -> None: with self._cond: - self._available += cost - if self._available > self._budget: - # Mirrors BoundedSemaphore: an unbalanced release must fail - # loudly rather than silently widen the budget. - self._available -= cost + # Mirrors BoundedSemaphore: an unbalanced release must fail loudly + # rather than silently widen the budget. The tolerance absorbs float + # rounding -- charges are duration/60, and refunding them in a + # different order than they were charged (workers finish out of + # order) lands a few ulps above the budget -- 220.00000000000003 + # for three charges, in ~7% of orderings. Without it, release() + # raises from the worker's finally block, loses the manifest record + # and fails the whole phase through future.result(). + if self._available + cost > self._budget + _BUDGET_EPSILON: raise ValueError("released more budget than was acquired") + # Clamping on the last outstanding refund snaps the budget back to + # exact, so rounding cannot accumulate across a long run. + self._available = min(self._available + cost, self._budget) self._cond.notify_all() diff --git a/tests/test_archive_transcriber.py b/tests/test_archive_transcriber.py index f714ce6..30f1516 100644 --- a/tests/test_archive_transcriber.py +++ b/tests/test_archive_transcriber.py @@ -783,12 +783,36 @@ def _small() -> None: small.join(timeout=2.0) assert order == ["big", "small"] + def test_out_of_order_refunds_do_not_raise_on_float_drift(self): + # Charges are duration/60, and workers finish in a different order than + # they started. Refunding out of order leaves _available a few ulps + # above the budget, which a strict > comparison rejects -- raising from + # the worker's finally block and failing the whole phase. These three + # charges reproduce it exactly: they refund to 220.00000000000003. + budget = archive_transcriber.AudioMinutesBudget(220.0) + charges = [66.40270988202536, 28.242821253198972, 4.331763230250052] + for charge in charges: + budget.acquire(charge) + for index in (1, 0, 2): + budget.release(charges[index]) + # And the drift must not be left behind to accumulate. + assert budget.acquire(220.0) == 220.0 + def test_unbalanced_release_fails_loudly(self): # An over-release must raise rather than silently widening the budget. budget = archive_transcriber.AudioMinutesBudget(200.0) with pytest.raises(ValueError): budget.release(1.0) + def test_double_release_of_a_real_charge_still_raises(self): + # The float tolerance must not be wide enough to hide a genuine + # double-refund, which is off by a whole video's worth of budget. + budget = archive_transcriber.AudioMinutesBudget(200.0) + charged = budget.acquire(51.0) + budget.release(charged) + with pytest.raises(ValueError): + budget.release(charged) + def test_failed_release_leaves_the_budget_intact(self): budget = archive_transcriber.AudioMinutesBudget(200.0) with pytest.raises(ValueError):