From 072d16fedf8d7d6ded39084075bbe34c77680444 Mon Sep 17 00:00:00 2001 From: Mahnoor Zaffar <1999mahnoor@gmail.com> Date: Sun, 16 Aug 2026 13:29:04 +0500 Subject: [PATCH 1/6] feat(benchmarks): add reproducible indexing-latency benchmark with regression detection (#1) * feat(benchmarks): add reproducible indexing-latency benchmark Adds a command that generates synthetic media via FFmpeg testsrc2 and measures per-stage indexing throughput, per-stage wall time, and peak memory across configurable modalities. Supports regression detection against a prior baseline report. - : corpus generation, run orchestrator (drives real run_index/ModelRuntime), per-stage aggregation, baseline comparison - CLI command with --modalities, --videos, --duration-seconds, --resolution, --repetitions, --input-mode (transcript/transcribe), --audio-mode, --baseline, --baseline-tolerance - documents the protocol, output schema, and limitations - 23 unit tests for validation, aggregation, clip command building, baseline comparison, and corpus spec * fix(benchmarks): address CodeRabbit review issues - Reject resolutions with extra components (e.g. 320x180x1) - Accumulate record_counts across repetitions instead of overwriting - Move corpus generation inside try block for proper failure handling - Pass reset parameter through instead of hardcoded True - Validate baseline configuration compatibility before comparison --- docs/benchmarking/README.md | 1 + docs/benchmarking/performance.md | 152 +++++++ src/vidxp/benchmarks/cli.py | 132 ++++++ src/vidxp/benchmarks/latency.py | 703 +++++++++++++++++++++++++++++++ tests/test_benchmark_latency.py | 329 +++++++++++++++ 5 files changed, 1317 insertions(+) create mode 100644 docs/benchmarking/performance.md create mode 100644 src/vidxp/benchmarks/latency.py create mode 100644 tests/test_benchmark_latency.py diff --git a/docs/benchmarking/README.md b/docs/benchmarking/README.md index cfe692fd..8e2e38bc 100644 --- a/docs/benchmarking/README.md +++ b/docs/benchmarking/README.md @@ -18,6 +18,7 @@ installation and product usage, start with the main | HiREST transcript localization | Legacy full result + current smoke | The legacy MiniLM stack scored all 193 validation pairs; current Qwen3 passed a two-video real execution smoke; 776 released test predictions remain unscored because their public bounds are placeholders | | LongVALE combined evaluation | Next | Build the visual-plus-speech adapter and validate one evaluation archive before scheduling the full run | | Actor clustering | Data-gated | The preferred BBT/Buffy evaluation still requires lawful access to the source episodes | +| Indexing latency benchmark | Ready | `vidxp benchmark index-latency` measures throughput, per-stage timings, and peak memory on synthetic FFmpeg media; supports regression detection against baselines | Read [current results](results.md) for the scores, plain-language metric definitions, honest comparisons, and the next benchmark decision. diff --git a/docs/benchmarking/performance.md b/docs/benchmarking/performance.md new file mode 100644 index 00000000..192c0bf1 --- /dev/null +++ b/docs/benchmarking/performance.md @@ -0,0 +1,152 @@ +# Latency benchmark protocol + +Status: Ready + +The latency benchmark (`vidxp benchmark index-latency`) measures indexing +throughput, per-stage timing, and peak memory using synthetic media generated +by FFmpeg on the caller's machine. It is designed for regression detection +between VidXP builds and for evaluating the latency impact of model or +architecture changes. + +## Protocol + +### Corpus generation + +The benchmark generates deterministic synthetic clips using FFmpeg's `lavfi` +source filters: + +| Parameter | Default | Notes | +|---|---|---| +| `--videos` | 1 | Number of synthetic clips | +| `--duration-seconds` | 8.0 | Wall-clock duration of each clip | +| `--fps` | 24 | Frame rate | +| `--resolution` | 320x180 | `WxH` format | +| `--audio-mode` | `none` | `none`, `sine`, or `flite` | +| `--input-mode` | `transcript` | `transcript` or `transcribe` | + +Video is generated via `testsrc2` (colour bars + timestamp). When +`--input-mode transcript` and `dialogue` is enabled, a deterministic +synthetic transcript (seeded PRNG over a fixed English vocabulary) is +supplied without real transcription. When `input-mode transcribe` is +used, `--audio-mode flite` must also be set and libflite must be +available in the ffmpeg build. + +### Indexing measurement + +Each repetition runs the full indexing pipeline via `run_index()` with +`reset=True`. The following stages are timed by the existing manifest +timing infrastructure (`core/manifest.py:record_stage`): + +| Stage | Modality | Measures | +|---|---|---| +| `frame_stream` | (all visual) | Decode throughput (frames/s) | +| `scene` | scene | SigLIP2 embedding (frames/s) | +| `actor` | actor | OpenCV detect + recognise (frames/s) | +| `visual_indexing` | all visual | Combined group wall time | +| `dialogue_indexing` | dialogue | Embedding throughput (phrases/s) | + +Peak RSS is captured via `resource.getrusage(RUSAGE_SELF).ru_maxrss` +(POSIX only; `None` on Windows, reported in bytes on macOS, KiB on +Linux). + +### Repetitions + +When `--repetitions N` > 1, each repetition runs the full cycle +(generate once, index each time after `reset`). Results are reported +as mean, min, and max across all per-video per-repetition samples. + +### Baseline comparison + +Pass `--baseline ` to compare the +current run against a prior report. For each stage present in both, +the delta ratio (`new_mean / old_mean - 1`) is computed. A stage with +a delta exceeding `--baseline-tolerance` (default 0.15 = 15%) is +flagged as a regression. The verdict is `fail` if any stage regressed, +else `pass`. + +### Output + +The benchmark writes its report to `run_directory/report.json` and +invokes `record_adapter_manifest` (embedding the corpus spec, device, +and result classification into the run's `manifest.json`). + +Report schema: + +```json +{ + "schema_version": 1, + "benchmark": "latency", + "run_id": "my-run", + "corpus": { "videos": 1, "duration_seconds": 8.0, ... }, + "modalities": ["scene", "actor"], + "device": "cpu", + "repetitions": 1, + "git": { "commit": "...", "dirty": false }, + "environment": { ... }, + "record_counts": { "scene": 8, "actor": 0 }, + "processed_frames": 8, + "stages": { + "scene": { + "runs": 1, "mean_seconds": 2.1, "min_seconds": 2.1, + "max_seconds": 2.1, "rate_per_second": 3.8 + } + }, + "summary": { + "wall_seconds": { "runs": 1, "mean_seconds": 5.0, ... }, + "peak_rss": { "unit": "bytes", "samples": 1, "value": 123456789 } + }, + "baseline": null | { "stages": {...}, "regressions": [...], "verdict": "pass" } +} +``` + +## Limitations + +- The synthetic video has no semantic scene content, so scene embeddings + are representative of throughput but not retrieval quality. +- Actors are not present in `testsrc2` video; `actor` stage measures + the per-frame face-detection overhead with zero detections. +- When `input_mode=transcript`, no real whisper transcription occurs; + dialogue embedding is measured on a synthetic transcript. +- True transcription latency (`input_mode=transcribe`) requires a + speech source (`--audio-mode flite`) and libflite in the FFmpeg + build; the generated speech is a short fixed sentence and does not + represent naturalistic conversation length or vocabulary. +- Peak RSS measures the whole-process peak, which includes Python + overhead, loaded models, and Chroma state; it is not a pure + indexing-stage measurement. + +## Usage + +```bash +# Default: single 8-second 320x180 clip, scene-only, 1 rep +vidxp benchmark index-latency --run-id my-baseline + +# Scene + actor + dialogue (synthetic transcript), 3 reps, compare with baseline +vidxp benchmark index-latency \ + --run-id v2-compare \ + --modalities scene,actor,dialogue \ + --videos 2 \ + --duration-seconds 12 \ + --repetitions 3 \ + --json \ + --baseline benchmark_runs/latency/synthetic/my-baseline/report.json + +# Real transcription (requires libflite in ffmpeg) +vidxp benchmark index-latency \ + --run-id transcribe-test \ + --modalities dialogue \ + --input-mode transcribe \ + --audio-mode flite \ + --device cpu +``` + +## Adding a new performance benchmark + +1. Define the corpus parameters and any new modality combinations in + the existing `run_latency` entry point. +2. Run the baseline and save its `report.json`. +3. Make your change (model swap, concurrency refactor, etc.). +4. Re-run with `--baseline ` and verify no + regressions. +5. Commit the baseline report to a designated location (e.g. + `docs/benchmarking/baselines/`) if it serves as a team reference. diff --git a/src/vidxp/benchmarks/cli.py b/src/vidxp/benchmarks/cli.py index 34fa5e2d..e8f069dd 100644 --- a/src/vidxp/benchmarks/cli.py +++ b/src/vidxp/benchmarks/cli.py @@ -21,6 +21,7 @@ HIREST_DEFAULT_WINDOW_FRACTION, run_hirest, ) +from vidxp.benchmarks.latency import run_latency from vidxp.benchmarks.prepare import ( PreparationPlan, execute_preparation, @@ -574,3 +575,134 @@ def hirest_command( emit_json(metrics) else: rich_print(metrics) + + +@app.command("index-latency") +def index_latency_command( + ctx: typer.Context, + run_id: Annotated[str, typer.Option(help="Arbitrary label for this run.")], + modalities: Annotated[ + str, + typer.Option( + help="Comma-separated modality names: scene,actor,dialogue." + ), + ] = "scene", + videos: Annotated[ + int, + typer.Option(min=1, help="Number of synthetic clips to generate."), + ] = 1, + duration_seconds: Annotated[ + float, + typer.Option(min=0.1, help="Duration of each synthetic clip."), + ] = 8.0, + fps: Annotated[ + int, + typer.Option(min=1, help="Frame rate of synthetic clips."), + ] = 24, + resolution: Annotated[ + str, + typer.Option( + help="Synthetic clip resolution in WxH format (e.g. 320x180)." + ), + ] = "320x180", + repetitions: Annotated[ + int, + typer.Option(min=1, help="Number of times to repeat the run."), + ] = 1, + input_mode: Annotated[ + Literal["transcript", "transcribe"], + typer.Option( + help=( + "'transcript' supplies a synthetic transcript for dialogue " + "embedding (no real transcription). 'transcribe' runs " + "real whisper on audio (requires --audio-mode flite)." + ) + ), + ] = "transcript", + audio_mode: Annotated[ + Literal["none", "sine", "flite"], + typer.Option( + help=( + "Audio track for synthetic clips: 'none' (no audio), " + "'sine' (tone), or 'flite' (speech synthesis)." + ) + ), + ] = "none", + reset: Annotated[ + bool, + typer.Option(help="Clear any existing index before running."), + ] = False, + baseline: Annotated[ + Path | None, + typer.Option( + exists=True, + dir_okay=False, + help=( + "Path to a previous latency report JSON for regression " + "comparison." + ), + ), + ] = None, + baseline_tolerance: Annotated[ + float, + typer.Option( + min=0.0, + max=5.0, + help=( + "Relative regression tolerance. A stage mean slower by " + "more than this ratio flags as regression." + ), + ), + ] = 0.15, + json_output: Annotated[ + bool, + typer.Option("--json", help="Emit machine-readable JSON."), + ] = False, +) -> None: + """Run a reproducible indexing-latency benchmark on synthetic media.""" + + selected = [item.strip() for item in modalities.split(",") if item.strip()] + if not selected: + raise typer.BadParameter( + "At least one latency modality is required.", param_hint="--modalities" + ) + for modality in selected: + _require_benchmark_dependencies(modality) + + try: + parts = resolution.lower().split("x") + if len(parts) != 2: + raise ValueError + width, height = int(parts[0]), int(parts[1]) + if width <= 0 or height <= 0: + raise ValueError + except (IndexError, ValueError, AttributeError): + raise typer.BadParameter( + f"Invalid resolution: {resolution!r}. Use WxH, e.g. 320x180.", + param_hint="--resolution", + ) + + state = state_from_context(ctx) + report = run_latency( + run_id=run_id, + output_root=state.settings.data_dir / "benchmark_runs", + ffprobe=state.settings.ffprobe_executable, + ffmpeg=state.settings.ffmpeg_executable, + modalities=tuple(selected), + videos=videos, + duration_seconds=duration_seconds, + fps=fps, + width=width, + height=height, + repetitions=repetitions, + input_mode=input_mode, + audio_mode=audio_mode, + device=state.settings.runtime_backend, + reset=reset, + baseline_path=baseline, + baseline_tolerance=baseline_tolerance, + ) + if effective_output_format(state, json_output) == OutputFormat.json: + emit_json(report) + else: + rich_print(report) diff --git a/src/vidxp/benchmarks/latency.py b/src/vidxp/benchmarks/latency.py new file mode 100644 index 00000000..87272f57 --- /dev/null +++ b/src/vidxp/benchmarks/latency.py @@ -0,0 +1,703 @@ +from __future__ import annotations + +import json +import random +import subprocess +from dataclasses import dataclass +from pathlib import Path +from statistics import mean +from time import perf_counter +from typing import Any, Literal, Mapping, Sequence + +from vidxp.benchmarks.common import ( + append_failure, + benchmark_generation_id, + benchmark_media_id, + ensure_adapter_outputs, + record_adapter_manifest, +) +from vidxp.capabilities.registry import create_capability_registry +from vidxp.core.contracts import IndexConfig, VideoSource +from vidxp.core.manifest import ManifestStore, write_json_atomic +from vidxp.core.runner import run_index +from vidxp.core.storage import IndexStorage +from vidxp.infrastructure.local_index import LOCAL_INDEX_RUNTIME_CHECKS +from vidxp.media_runtime import inspect_media_runtime +from vidxp.runtime import ModelRuntime +from vidxp.settings import VidXPSettings + + +LATENCY_BENCHMARK = "latency" +LATENCY_SPLIT = "synthetic" +LATENCY_SCHEMA_VERSION = 1 +DEFAULT_CORPUS_SEED = 2026 +SUPPORTED_MODALITIES = ("scene", "actor", "dialogue") + +_STAGE_RATES: Mapping[str, str] = { + "scene": "scene_frames", + "actor": "actor_frames", + "frame_stream": "source_frames_advanced", + "dialogue_indexing": "dialogue_phrases", +} + +_VOCABULARY = ( + "the quick brown fox jumps over the lazy dog honest sunshine light " + "morning river ocean mountain garden flower silver golden copper " + "bright shadow shadow candle lantern window door table chair book " + "letter number station market kitchen garden bakery camera video " + "music voice speech word phrase moment memory journey story world " + "quiet calm gentle peaceful vivid warm cool bright dark soft loud" +).split() + + +def _peak_rss_bytes() -> int | None: + try: + import resource + except ImportError: + return None + return int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) + + +def rss_unit() -> Literal["bytes", "KiB"]: + return "bytes" if _sys_platform() == "darwin" else "KiB" + + +def _sys_platform() -> str: + import sys + + return sys.platform + + +@dataclass(frozen=True) +class SyntheticCorpusSpec: + videos: int + duration_seconds: float + fps: int + width: int + height: int + audio_mode: Literal["none", "sine", "flite"] + seed: int + + def public_record(self) -> dict[str, Any]: + return { + "videos": self.videos, + "duration_seconds": self.duration_seconds, + "fps": self.fps, + "width": self.width, + "height": self.height, + "audio_mode": self.audio_mode, + "seed": self.seed, + } + + +def validate_latency_options( + *, + modalities: Sequence[str], + videos: int, + duration_seconds: float, + fps: int, + width: int, + height: int, + repetitions: int, + input_mode: str, + audio_mode: str, + baseline_tolerance: float, +) -> tuple[str, ...]: + selected = tuple(dict.fromkeys(modalities)) + if not selected: + raise ValueError("At least one latency modality must be selected.") + unsupported = sorted(set(selected) - set(SUPPORTED_MODALITIES)) + if unsupported: + raise ValueError( + "Latency modalities must be a subset of " + + ", ".join(SUPPORTED_MODALITIES) + + "; unsupported: " + + ", ".join(unsupported) + ) + if videos <= 0: + raise ValueError("videos must be greater than zero.") + if duration_seconds <= 0: + raise ValueError("duration_seconds must be greater than zero.") + if fps <= 0: + raise ValueError("fps must be greater than zero.") + if width <= 0 or height <= 0: + raise ValueError("width and height must be greater than zero.") + if repetitions <= 0: + raise ValueError("repetitions must be greater than zero.") + if input_mode not in {"transcript", "transcribe"}: + raise ValueError("input_mode must be 'transcript' or 'transcribe'.") + if audio_mode not in {"none", "sine", "flite"}: + raise ValueError("audio_mode must be 'none', 'sine', or 'flite'.") + if "dialogue" in selected and input_mode == "transcribe": + if audio_mode != "flite": + raise ValueError( + "Real transcription requires a speech audio source; " + "use --audio-mode flite with --input-mode transcribe." + ) + if not 0 <= baseline_tolerance <= 5: + raise ValueError("baseline_tolerance must be between zero and five.") + return selected + + +def synthetic_transcript( + *, + duration_seconds: float, + seed: int, +) -> list[dict[str, Any]]: + generator = random.Random(seed) + strides = max(1, int(duration_seconds / 0.4)) + words = [generator.choice(_VOCABULARY) for _ in range(strides)] + span = duration_seconds / len(words) + word_events = [ + { + "word": word, + "start": round(index * span, 4), + "end": round((index + 1) * span, 4), + } + for index, word in enumerate(words) + ] + return [ + { + "text": " ".join(words), + "start": 0.0, + "end": duration_seconds, + "words": word_events, + } + ] + + +def _flite_text(seed: int) -> str: + generator = random.Random(seed) + words = [generator.choice(_VOCABULARY) for _ in range(24)] + return " ".join(words) + + +def build_clip_command( + *, + spec: SyntheticCorpusSpec, + ffmpeg: str, + destination: Path, +) -> list[str]: + compact = spec.width != 0 and spec.height != 0 + if not compact: + raise ValueError("The synthetic corpus requires positive dimensions.") + command = [ + ffmpeg, + "-y", + "-v", + "error", + "-f", + "lavfi", + "-i", + f"testsrc2=size={spec.width}x{spec.height}:rate={spec.fps}", + ] + if spec.audio_mode == "sine": + command += [ + "-f", + "lavfi", + "-i", + "sine=frequency=440:sample_rate=16000", + ] + elif spec.audio_mode == "flite": + command += [ + "-f", + "lavfi", + "-i", + f"flite=text='{_flite_text(spec.seed)}',sample_rate=16000", + ] + command += [ + "-t", + f"{spec.duration_seconds:g}", + "-c:v", + "libx264", + "-pix_fmt", + "yuv420p", + ] + if spec.audio_mode == "none": + command.append("-an") + else: + command += ["-c:a", "aac"] + command.append(str(destination)) + return command + + +def _probe_duration(ffprobe: str, path: Path) -> float: + completed = subprocess.run( + [ + ffprobe, + "-v", + "error", + "-show_entries", + "format=duration", + "-of", + "json", + str(path), + ], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=60, + ) + if completed.returncode != 0: + raise ValueError( + f"ffprobe could not read a generated clip: {path}" + ) + try: + duration = float(json.loads(completed.stdout)["format"]["duration"]) + except (KeyError, ValueError, TypeError, json.JSONDecodeError) as exc: + raise ValueError( + f"ffprobe returned an invalid duration for {path}." + ) from exc + if duration <= 0: + raise ValueError(f"ffprobe reported a non-positive duration for {path}.") + return duration + + +def generate_synthetic_corpus( + *, + spec: SyntheticCorpusSpec, + directory: str | Path, + ffprobe: str, + ffmpeg: str, + audio_mode: str | None = None, +) -> list[Path]: + runtime_status = inspect_media_runtime( + ffprobe=ffprobe, + ffmpeg=ffmpeg, + ) + if not runtime_status.ready: + raise ValueError( + "The latency benchmark requires FFmpeg and ffprobe to generate " + "the synthetic corpus. Run `vidxp init`, then retry." + ) + destination = Path(directory) + destination.mkdir(parents=True, exist_ok=True) + clips = [] + for index in range(spec.videos): + path = destination / f"clip-{index:03d}.mp4" + command = build_clip_command( + spec=spec, + ffmpeg=ffmpeg, + destination=path, + ) + completed = subprocess.run( + command, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=600, + ) + if completed.returncode != 0: + stderr = (completed.stderr or "").strip() + if spec.audio_mode == "flite" and stderr: + raise ValueError( + "FFmpeg could not apply the flite speech filter " + f"(libflite likely unavailable): {stderr}" + ) + raise ValueError(f"FFmpeg could not generate {path}: {stderr}") + if not path.is_file() or path.stat().st_size == 0: + raise ValueError(f"FFmpeg did not produce {path}.") + _probe_duration(ffprobe, path) + clips.append(path) + return clips + + +def _per_video_stages( + manifest: Mapping[str, Any], + video_id: str, +) -> dict[str, float]: + video = manifest["videos"].get(video_id) + if video is None or video.get("state") == "failed": + return {} + return { + str(stage): float(entry["seconds"]) + for stage, entry in (video.get("stages") or {}).items() + if entry.get("state") != "incomplete" + and float(entry.get("seconds", 0.0)) > 0 + } + + +def _summary_ratio( + manifest: Mapping[str, Any], + video_id: str, + *, + metric: str, + stage: str, +) -> float | None: + video = manifest["videos"].get(video_id) or {} + summary = video.get("summary") or {} + seconds = _per_video_stages(manifest, video_id).get(stage) + count = summary.get(metric) + if seconds is None or count is None or seconds <= 0 or count <= 0: + return None + return float(count) / seconds + + +def aggregate_latency_runs( + manifests: Sequence[Mapping[str, Any]], + *, + wall_seconds: Sequence[float], + peak_rss_samples: Sequence[int | None], +) -> dict[str, Any]: + if len(manifests) != len(wall_seconds): + raise ValueError( + "Every latency repetition requires a wall-clock sample." + ) + stage_samples: dict[str, list[float]] = {} + rate_samples: dict[str, list[float]] = {} + per_video: list[dict[str, Any]] = [] + processed_frames = 0 + record_counts: dict[str, int] = {} + for repetition, manifest in enumerate(manifests): + processed_frames += int(manifest.get("processed_frames", 0)) + for modality, count in (manifest.get("record_counts") or {}).items(): + record_counts[modality] = record_counts.get(modality, 0) + int(count) + for video_id in sorted(manifest.get("videos", {})): + stages = _per_video_stages(manifest, video_id) + if not stages: + continue + for stage, seconds in stages.items(): + stage_samples.setdefault(stage, []).append(seconds) + rate_stages = { + stage: _summary_ratio( + manifest, + video_id, + metric=_STAGE_RATES[stage], + stage=stage, + ) + for stage in _STAGE_RATES + if stage in stages + } + for stage, rate in rate_stages.items(): + if rate is None: + continue + rate_samples.setdefault(stage, []).append(rate) + video = manifest["videos"].get(video_id, {}) + per_video.append( + { + "repetition": repetition, + "video_id": video_id, + "wall_seconds": ( + wall_seconds[repetition] + ), + "stages": dict(sorted(stages.items())), + "summary": video.get("summary", {}), + } + ) + stages: dict[str, dict[str, Any]] = {} + for stage, samples in stage_samples.items(): + values = sorted(samples) + summary: dict[str, Any] = { + "runs": len(values), + "mean_seconds": mean(values), + "min_seconds": values[0], + "max_seconds": values[-1], + } + rates = rate_samples.get(stage) + if rates: + summary["rate_per_second"] = mean(rates) + stages[stage] = summary + approximate_rss = [ + sample for sample in peak_rss_samples if sample is not None + ] + summary = { + "wall_seconds": { + "runs": len(wall_seconds), + "mean_seconds": mean(wall_seconds), + "min_seconds": min(wall_seconds), + "max_seconds": max(wall_seconds), + }, + "peak_rss": { + "unit": rss_unit(), + "samples": len(approximate_rss), + "value": int(max(approximate_rss)) if approximate_rss else None, + }, + } + return { + "per_video": per_video, + "stages": dict(sorted(stages.items())), + "summary": summary, + "processed_frames": processed_frames, + "record_counts": dict(sorted(record_counts.items())), + } + + +def _validate_baseline_compatibility( + report: Mapping[str, Any], + baseline: Mapping[str, Any], +) -> None: + baseline_corpus = baseline.get("corpus") or {} + report_corpus = report.get("corpus") or {} + corpus_keys = ("videos", "duration_seconds", "fps", "width", "height", "audio_mode", "seed") + for key in corpus_keys: + report_val = report_corpus.get(key) + baseline_val = baseline_corpus.get(key) + if report_val is not None and baseline_val is not None and report_val != baseline_val: + raise ValueError( + f"Baseline corpus mismatch: {key!r} is {baseline_val!r}, " + f"current is {report_val!r}." + ) + for key in ("input_mode", "device"): + report_val = report.get(key) + baseline_val = baseline.get(key) + if report_val is not None and baseline_val is not None and report_val != baseline_val: + raise ValueError( + f"Baseline {key!r} mismatch: {baseline_val!r}, current is {report_val!r}." + ) + report_mods = sorted(report.get("modalities") or []) + baseline_mods = sorted(baseline.get("modalities") or []) + if report_mods and baseline_mods and report_mods != baseline_mods: + raise ValueError( + f"Baseline modalities mismatch: {baseline_mods}, current is {report_mods}." + ) + + +def compare_baseline( + report: Mapping[str, Any], + baseline: Mapping[str, Any], + *, + tolerance: float, +) -> dict[str, Any]: + if tolerance < 0: + raise ValueError("Baseline tolerance must be nonnegative.") + comparisons: dict[str, dict[str, Any]] = {} + regressions: list[str] = [] + for stage, previous in (baseline.get("stages") or {}).items(): + current = (report.get("stages") or {}).get(stage) + if current is None or not previous.get("runs"): + continue + old_mean = float(previous["mean_seconds"]) + new_mean = float(current["mean_seconds"]) + if old_mean <= 0: + continue + delta = new_mean / old_mean - 1.0 + comparisons[stage] = { + "old_mean_seconds": old_mean, + "new_mean_seconds": new_mean, + "delta_ratio": delta, + "regressed": delta > tolerance, + } + if delta > tolerance: + regressions.append(stage) + return { + "schema_version": LATENCY_SCHEMA_VERSION, + "tolerance": tolerance, + "stages": dict(sorted(comparisons.items())), + "regressions": regressions, + "verdict": "fail" if regressions else "pass", + } + + +def build_latency_sources( + *, + clips: Sequence[Path], + spec: SyntheticCorpusSpec, + input_mode: Literal["transcript", "transcribe"], +) -> list[VideoSource]: + sources = [] + for index, path in enumerate(clips): + transcript = ( + synthetic_transcript( + duration_seconds=spec.duration_seconds, + seed=spec.seed + index, + ) + if input_mode == "transcript" + else None + ) + sources.append( + VideoSource( + video_id=benchmark_media_id( + LATENCY_BENCHMARK, + f"clip-{index:03d}", + ), + path=path, + source_name=f"clip-{index:03d}.mp4", + transcript=transcript, + ) + ) + return sources + + +def run_latency( + *, + run_id: str, + output_root: str | Path = "benchmark_runs", + ffprobe: str = "ffprobe", + ffmpeg: str = "ffmpeg", + modalities: Sequence[str] = ("scene",), + videos: int = 1, + duration_seconds: float = 8.0, + fps: int = 24, + width: int = 320, + height: int = 180, + repetitions: int = 1, + input_mode: Literal["transcript", "transcribe"] = "transcript", + audio_mode: Literal["none", "sine", "flite"] = "none", + device: str = "cpu", + reset: bool = False, + baseline_path: str | Path | None = None, + baseline_tolerance: float = 0.15, +) -> dict[str, Any]: + selected = validate_latency_options( + modalities=modalities, + videos=videos, + duration_seconds=duration_seconds, + fps=fps, + width=width, + height=height, + repetitions=repetitions, + input_mode=input_mode, + audio_mode=audio_mode, + baseline_tolerance=baseline_tolerance, + ) + spec = SyntheticCorpusSpec( + videos=videos, + duration_seconds=duration_seconds, + fps=fps, + width=width, + height=height, + audio_mode=audio_mode, + seed=DEFAULT_CORPUS_SEED, + ) + config = IndexConfig( + dataset=LATENCY_BENCHMARK, + split=LATENCY_SPLIT, + run_id=run_id, + enabled_modalities=selected, + device=device, + output_root=output_root, + generation_id=benchmark_generation_id( + LATENCY_BENCHMARK, + LATENCY_SPLIT, + run_id, + ), + ) + run_directory = config.run_directory + registry = create_capability_registry( + platform_runtime_checks=LOCAL_INDEX_RUNTIME_CHECKS + ) + runtime = ModelRuntime( + VidXPSettings( + repository_root=run_directory, + runtime_backend=device, + ), + allowed_specs=registry.model_specs(), + ) + ensure_adapter_outputs(run_directory) + manifests: list[dict[str, Any]] = [] + wall_samples: list[float] = [] + rss_samples: list[int | None] = [] + try: + clips = generate_synthetic_corpus( + spec=spec, + directory=run_directory / "corpus", + ffprobe=ffprobe, + ffmpeg=ffmpeg, + ) + sources = build_latency_sources( + clips=clips, + spec=spec, + input_mode=input_mode, + ) + for _ in range(repetitions): + started = perf_counter() + with IndexStorage(config) as storage: + manifest = run_index( + sources, + config, + reset=reset, + storage=storage, + manifest_store=ManifestStore( + config, + registry=registry, + runtime=runtime, + ), + registry=registry, + runtime=runtime, + ) + store_size_bytes = storage.size_bytes() + wall_samples.append(perf_counter() - started) + rss_samples.append(_peak_rss_bytes()) + manifests.append( + { + **manifest, + "store_size_bytes_at_commit": store_size_bytes, + } + ) + aggregated = aggregate_latency_runs( + manifests, + wall_seconds=wall_samples, + peak_rss_samples=rss_samples, + ) + report = { + "schema_version": LATENCY_SCHEMA_VERSION, + "benchmark": LATENCY_BENCHMARK, + "run_id": run_id, + "created_at": manifests[-1].get("completed_at"), + "corpus": spec.public_record(), + "input_mode": input_mode, + "modalities": list(selected), + "device": device, + "repetitions": repetitions, + "git": manifests[-1].get("git"), + "environment": manifests[-1].get("environment"), + "config_fingerprint": manifests[-1].get("config_fingerprint"), + "record_counts": aggregated["record_counts"], + "processed_frames": aggregated["processed_frames"], + "summary": aggregated["summary"], + "stages": aggregated["stages"], + "per_video": aggregated["per_video"], + "baseline": None, + } + if baseline_path is not None: + try: + baseline = json.loads( + Path(baseline_path).read_text(encoding="utf-8") + ) + except (OSError, json.JSONDecodeError) as exc: + raise ValueError( + f"Baseline report is not readable JSON: {baseline_path}" + ) from exc + _validate_baseline_compatibility(report, baseline) + report["baseline"] = compare_baseline( + report, + baseline, + tolerance=baseline_tolerance, + ) + write_json_atomic(run_directory / "report.json", report) + record_adapter_manifest( + run_directory, + benchmark=LATENCY_BENCHMARK, + subset={ + "label": f"latency_{run_id}", + "modalities": list(selected), + "video_count": videos, + "duration_seconds": duration_seconds, + "repetitions": repetitions, + }, + artifacts=[], + state="complete", + details={ + "device": device, + "input_mode": input_mode, + "audio_mode": audio_mode, + "corpus": spec.public_record(), + "result_classification": "performance_benchmark_not_quality_score", + }, + ) + return report + except BaseException as error: + append_failure(run_directory, stage="latency_adapter", error=error) + record_adapter_manifest( + run_directory, + benchmark=LATENCY_BENCHMARK, + subset={ + "label": f"latency_{run_id}", + "modalities": list(selected), + }, + artifacts=[], + state="failed", + ) + raise \ No newline at end of file diff --git a/tests/test_benchmark_latency.py b/tests/test_benchmark_latency.py new file mode 100644 index 00000000..6cbd7057 --- /dev/null +++ b/tests/test_benchmark_latency.py @@ -0,0 +1,329 @@ +from __future__ import annotations + +import unittest +from pathlib import Path + +from vidxp.benchmarks.latency import ( + SyntheticCorpusSpec, + aggregate_latency_runs, + build_clip_command, + build_latency_sources, + compare_baseline, + synthetic_transcript, + validate_latency_options, +) + + +class LatencyValidationTests(unittest.TestCase): + def test_validates_default_options(self): + selected = validate_latency_options( + modalities=("scene",), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=3, + input_mode="transcript", + audio_mode="none", + baseline_tolerance=0.15, + ) + self.assertEqual(selected, ("scene",)) + + def test_rejects_empty_modalities(self): + with self.assertRaises(ValueError): + validate_latency_options( + modalities=(), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcript", + audio_mode="none", + baseline_tolerance=0.15, + ) + + def test_rejects_unsupported_modality(self): + with self.assertRaises(ValueError): + validate_latency_options( + modalities=("scene", "ocr"), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcript", + audio_mode="none", + baseline_tolerance=0.15, + ) + + def test_rejects_transcribe_without_flite(self): + with self.assertRaises(ValueError): + validate_latency_options( + modalities=("scene", "dialogue"), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcribe", + audio_mode="sine", + baseline_tolerance=0.15, + ) + + def test_accepts_transcribe_with_flite(self): + selected = validate_latency_options( + modalities=("dialogue",), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcribe", + audio_mode="flite", + baseline_tolerance=0.15, + ) + self.assertEqual(selected, ("dialogue",)) + + def test_deduplicates_modalities(self): + selected = validate_latency_options( + modalities=("scene", "scene", "actor"), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcript", + audio_mode="none", + baseline_tolerance=0.15, + ) + self.assertEqual(selected, ("scene", "actor")) + + +class SyntheticTranscriptTests(unittest.TestCase): + def test_returns_one_segment_with_words(self): + transcript = synthetic_transcript(duration_seconds=10.0, seed=42) + self.assertEqual(len(transcript), 1) + segment = transcript[0] + self.assertGreater(len(segment["text"]), 0) + self.assertEqual(segment["start"], 0.0) + self.assertGreater(segment["end"], 0.0) + self.assertGreater(len(segment["words"]), 0) + for word in segment["words"]: + self.assertIn("word", word) + self.assertIsInstance(word["start"], float) + self.assertIsInstance(word["end"], float) + + def test_deterministic_across_calls(self): + first = synthetic_transcript(duration_seconds=5.0, seed=99) + second = synthetic_transcript(duration_seconds=5.0, seed=99) + self.assertEqual(first, second) + + def test_different_seeds_differ(self): + first = synthetic_transcript(duration_seconds=5.0, seed=99) + second = synthetic_transcript(duration_seconds=5.0, seed=100) + self.assertNotEqual(first, second) + + +class BuildClipCommandTests(unittest.TestCase): + def test_no_audio_default(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=8.0, fps=24, + width=320, height=180, audio_mode="none", seed=42, + ) + command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) + self.assertIn("testsrc2=size=320x180:rate=24", command) + self.assertIn("-an", command) + self.assertNotIn("-c:a", command) + + def test_sine_audio_adds_aac(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=8.0, fps=24, + width=320, height=180, audio_mode="sine", seed=42, + ) + command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) + self.assertIn("sine=frequency=440:sample_rate=16000", command) + self.assertIn("-c:a", command) + self.assertNotIn("-an", command) + + def test_flite_audio_contains_filter_ref(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=8.0, fps=24, + width=320, height=180, audio_mode="flite", seed=42, + ) + command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) + flite_args = [arg for arg in command if "flite=text=" in arg] + self.assertEqual(len(flite_args), 1) + + def test_duration_is_formatted(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=3.5, fps=30, + width=640, height=480, audio_mode="none", seed=0, + ) + command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("clip.mp4")) + idx = command.index("-t") + self.assertEqual(command[idx + 1], "3.5") + self.assertIn("testsrc2=size=640x480:rate=30", command) + + +class BuildSourcesTests(unittest.TestCase): + def test_transcript_attached_in_input_mode(self): + spec = SyntheticCorpusSpec( + videos=2, duration_seconds=4.0, fps=24, + width=320, height=180, audio_mode="none", seed=42, + ) + clips = [Path(f"{i}.mp4") for i in range(2)] + sources = build_latency_sources(clips=clips, spec=spec, input_mode="transcript") + self.assertEqual(len(sources), 2) + for index, source in enumerate(sources): + self.assertIsNotNone(source.transcript) + self.assertIsNotNone(source.path) + self.assertIsNotNone(source.video_id) + self.assertEqual(source.source_name, f"clip-{index:03d}.mp4") + + def test_no_transcript_in_transcribe_mode(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=4.0, fps=24, + width=320, height=180, audio_mode="flite", seed=42, + ) + sources = build_latency_sources(clips=[Path("0.mp4")], spec=spec, input_mode="transcribe") + for source in sources: + self.assertIsNone(source.transcript) + + +class AggregateMetricsTests(unittest.TestCase): + def _sample_manifest(self, scene_seconds, scene_frames, actor_seconds, actor_frames): + return { + "processed_frames": scene_frames, + "record_counts": {"scene": scene_frames, "actor": actor_frames}, + "git": {"commit": "abc", "dirty": False}, + "environment": {"platform": "test"}, + "config_fingerprint": "fp1", + "completed_at": "2026-01-01T00:00:00", + "videos": { + "vid-1": { + "state": "complete", + "summary": { + "scene_frames": scene_frames, + "actor_frames": actor_frames, + "source_frames_advanced": scene_frames + 100, + }, + "stages": { + "scene": {"seconds": scene_seconds, "state": ""}, + "actor": {"seconds": actor_seconds, "state": ""}, + "frame_stream": {"seconds": 0.5, "state": ""}, + }, + } + }, + } + + def test_aggregates_single_manifest(self): + result = aggregate_latency_runs( + [self._sample_manifest(2.0, 8, 1.5, 3)], + wall_seconds=[3.5], + peak_rss_samples=[100000], + ) + self.assertEqual(result["processed_frames"], 8) + self.assertEqual(result["record_counts"], {"actor": 3, "scene": 8}) + self.assertIn("scene", result["stages"]) + self.assertAlmostEqual(result["stages"]["scene"]["mean_seconds"], 2.0) + self.assertAlmostEqual(result["stages"]["scene"]["rate_per_second"], 4.0) + self.assertAlmostEqual(result["stages"]["actor"]["mean_seconds"], 1.5) + self.assertAlmostEqual(result["summary"]["wall_seconds"]["mean_seconds"], 3.5) + + def test_aggregates_multiple_manifests(self): + m1 = self._sample_manifest(2.0, 8, 1.5, 3) + m2 = self._sample_manifest(2.5, 10, 2.0, 4) + result = aggregate_latency_runs( + [m1, m2], + wall_seconds=[3.5, 4.5], + peak_rss_samples=[100000, 120000], + ) + self.assertEqual(result["processed_frames"], 18) + self.assertAlmostEqual(result["stages"]["scene"]["mean_seconds"], 2.25) + self.assertAlmostEqual(result["stages"]["scene"]["min_seconds"], 2.0) + self.assertAlmostEqual(result["stages"]["scene"]["max_seconds"], 2.5) + self.assertEqual(len(result["per_video"]), 2) + self.assertAlmostEqual( + result["summary"]["wall_seconds"]["mean_seconds"], + 4.0, + ) + + def test_skips_failed_videos(self): + manifest = { + "processed_frames": 0, + "record_counts": {}, + "git": {}, + "environment": {}, + "config_fingerprint": "fp", + "completed_at": "", + "videos": { + "vid-1": { + "state": "failed", + "summary": {}, + "stages": {}, + } + }, + } + result = aggregate_latency_runs( + [manifest], + wall_seconds=[1.0], + peak_rss_samples=[None], + ) + self.assertEqual(result["processed_frames"], 0) + self.assertEqual(result["stages"], {}) + + +class CompareBaselineTests(unittest.TestCase): + def test_no_baseline_stages_returns_empty(self): + report = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} + baseline = {"stages": {}} + result = compare_baseline(report, baseline, tolerance=0.1) + self.assertEqual(result["stages"], {}) + self.assertEqual(result["regressions"], []) + self.assertEqual(result["verdict"], "pass") + + def test_regression_detected(self): + report = {"stages": {"scene": {"mean_seconds": 3.0, "runs": 1}}} + baseline = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} + result = compare_baseline(report, baseline, tolerance=0.1) + self.assertIn("scene", result["stages"]) + self.assertAlmostEqual( + result["stages"]["scene"]["delta_ratio"], 0.5 + ) + self.assertTrue(result["stages"]["scene"]["regressed"]) + self.assertEqual(result["regressions"], ["scene"]) + self.assertEqual(result["verdict"], "fail") + + def test_improvement_not_regression(self): + report = {"stages": {"scene": {"mean_seconds": 1.5, "runs": 1}}} + baseline = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} + result = compare_baseline(report, baseline, tolerance=0.1) + self.assertFalse(result["stages"]["scene"]["regressed"]) + self.assertEqual(result["regressions"], []) + self.assertEqual(result["verdict"], "pass") + + +class CorpusSpecTests(unittest.TestCase): + def test_public_record_roundtrip(self): + spec = SyntheticCorpusSpec( + videos=2, duration_seconds=8.0, fps=24, + width=320, height=180, audio_mode="none", seed=42, + ) + record = spec.public_record() + self.assertEqual(record["videos"], 2) + self.assertEqual(record["duration_seconds"], 8.0) + self.assertEqual(record["audio_mode"], "none") + + def test_flite_mode_recorded(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=5.0, fps=30, + width=640, height=480, audio_mode="flite", seed=7, + ) + self.assertEqual(spec.public_record()["audio_mode"], "flite") From 09a2a6366ba57f8cf43428b74f66bf1f9ec0adf9 Mon Sep 17 00:00:00 2001 From: Mahnoor-Zaffar <1999mahnoor@gmail.com> Date: Sun, 16 Aug 2026 13:40:32 +0500 Subject: [PATCH 2/6] Revert "feat(benchmarks): add reproducible indexing-latency benchmark with regression detection (#1)" This reverts commit 072d16fedf8d7d6ded39084075bbe34c77680444. --- docs/benchmarking/README.md | 1 - docs/benchmarking/performance.md | 152 ------- src/vidxp/benchmarks/cli.py | 132 ------ src/vidxp/benchmarks/latency.py | 703 ------------------------------- tests/test_benchmark_latency.py | 329 --------------- 5 files changed, 1317 deletions(-) delete mode 100644 docs/benchmarking/performance.md delete mode 100644 src/vidxp/benchmarks/latency.py delete mode 100644 tests/test_benchmark_latency.py diff --git a/docs/benchmarking/README.md b/docs/benchmarking/README.md index 8e2e38bc..cfe692fd 100644 --- a/docs/benchmarking/README.md +++ b/docs/benchmarking/README.md @@ -18,7 +18,6 @@ installation and product usage, start with the main | HiREST transcript localization | Legacy full result + current smoke | The legacy MiniLM stack scored all 193 validation pairs; current Qwen3 passed a two-video real execution smoke; 776 released test predictions remain unscored because their public bounds are placeholders | | LongVALE combined evaluation | Next | Build the visual-plus-speech adapter and validate one evaluation archive before scheduling the full run | | Actor clustering | Data-gated | The preferred BBT/Buffy evaluation still requires lawful access to the source episodes | -| Indexing latency benchmark | Ready | `vidxp benchmark index-latency` measures throughput, per-stage timings, and peak memory on synthetic FFmpeg media; supports regression detection against baselines | Read [current results](results.md) for the scores, plain-language metric definitions, honest comparisons, and the next benchmark decision. diff --git a/docs/benchmarking/performance.md b/docs/benchmarking/performance.md deleted file mode 100644 index 192c0bf1..00000000 --- a/docs/benchmarking/performance.md +++ /dev/null @@ -1,152 +0,0 @@ -# Latency benchmark protocol - -Status: Ready - -The latency benchmark (`vidxp benchmark index-latency`) measures indexing -throughput, per-stage timing, and peak memory using synthetic media generated -by FFmpeg on the caller's machine. It is designed for regression detection -between VidXP builds and for evaluating the latency impact of model or -architecture changes. - -## Protocol - -### Corpus generation - -The benchmark generates deterministic synthetic clips using FFmpeg's `lavfi` -source filters: - -| Parameter | Default | Notes | -|---|---|---| -| `--videos` | 1 | Number of synthetic clips | -| `--duration-seconds` | 8.0 | Wall-clock duration of each clip | -| `--fps` | 24 | Frame rate | -| `--resolution` | 320x180 | `WxH` format | -| `--audio-mode` | `none` | `none`, `sine`, or `flite` | -| `--input-mode` | `transcript` | `transcript` or `transcribe` | - -Video is generated via `testsrc2` (colour bars + timestamp). When -`--input-mode transcript` and `dialogue` is enabled, a deterministic -synthetic transcript (seeded PRNG over a fixed English vocabulary) is -supplied without real transcription. When `input-mode transcribe` is -used, `--audio-mode flite` must also be set and libflite must be -available in the ffmpeg build. - -### Indexing measurement - -Each repetition runs the full indexing pipeline via `run_index()` with -`reset=True`. The following stages are timed by the existing manifest -timing infrastructure (`core/manifest.py:record_stage`): - -| Stage | Modality | Measures | -|---|---|---| -| `frame_stream` | (all visual) | Decode throughput (frames/s) | -| `scene` | scene | SigLIP2 embedding (frames/s) | -| `actor` | actor | OpenCV detect + recognise (frames/s) | -| `visual_indexing` | all visual | Combined group wall time | -| `dialogue_indexing` | dialogue | Embedding throughput (phrases/s) | - -Peak RSS is captured via `resource.getrusage(RUSAGE_SELF).ru_maxrss` -(POSIX only; `None` on Windows, reported in bytes on macOS, KiB on -Linux). - -### Repetitions - -When `--repetitions N` > 1, each repetition runs the full cycle -(generate once, index each time after `reset`). Results are reported -as mean, min, and max across all per-video per-repetition samples. - -### Baseline comparison - -Pass `--baseline ` to compare the -current run against a prior report. For each stage present in both, -the delta ratio (`new_mean / old_mean - 1`) is computed. A stage with -a delta exceeding `--baseline-tolerance` (default 0.15 = 15%) is -flagged as a regression. The verdict is `fail` if any stage regressed, -else `pass`. - -### Output - -The benchmark writes its report to `run_directory/report.json` and -invokes `record_adapter_manifest` (embedding the corpus spec, device, -and result classification into the run's `manifest.json`). - -Report schema: - -```json -{ - "schema_version": 1, - "benchmark": "latency", - "run_id": "my-run", - "corpus": { "videos": 1, "duration_seconds": 8.0, ... }, - "modalities": ["scene", "actor"], - "device": "cpu", - "repetitions": 1, - "git": { "commit": "...", "dirty": false }, - "environment": { ... }, - "record_counts": { "scene": 8, "actor": 0 }, - "processed_frames": 8, - "stages": { - "scene": { - "runs": 1, "mean_seconds": 2.1, "min_seconds": 2.1, - "max_seconds": 2.1, "rate_per_second": 3.8 - } - }, - "summary": { - "wall_seconds": { "runs": 1, "mean_seconds": 5.0, ... }, - "peak_rss": { "unit": "bytes", "samples": 1, "value": 123456789 } - }, - "baseline": null | { "stages": {...}, "regressions": [...], "verdict": "pass" } -} -``` - -## Limitations - -- The synthetic video has no semantic scene content, so scene embeddings - are representative of throughput but not retrieval quality. -- Actors are not present in `testsrc2` video; `actor` stage measures - the per-frame face-detection overhead with zero detections. -- When `input_mode=transcript`, no real whisper transcription occurs; - dialogue embedding is measured on a synthetic transcript. -- True transcription latency (`input_mode=transcribe`) requires a - speech source (`--audio-mode flite`) and libflite in the FFmpeg - build; the generated speech is a short fixed sentence and does not - represent naturalistic conversation length or vocabulary. -- Peak RSS measures the whole-process peak, which includes Python - overhead, loaded models, and Chroma state; it is not a pure - indexing-stage measurement. - -## Usage - -```bash -# Default: single 8-second 320x180 clip, scene-only, 1 rep -vidxp benchmark index-latency --run-id my-baseline - -# Scene + actor + dialogue (synthetic transcript), 3 reps, compare with baseline -vidxp benchmark index-latency \ - --run-id v2-compare \ - --modalities scene,actor,dialogue \ - --videos 2 \ - --duration-seconds 12 \ - --repetitions 3 \ - --json \ - --baseline benchmark_runs/latency/synthetic/my-baseline/report.json - -# Real transcription (requires libflite in ffmpeg) -vidxp benchmark index-latency \ - --run-id transcribe-test \ - --modalities dialogue \ - --input-mode transcribe \ - --audio-mode flite \ - --device cpu -``` - -## Adding a new performance benchmark - -1. Define the corpus parameters and any new modality combinations in - the existing `run_latency` entry point. -2. Run the baseline and save its `report.json`. -3. Make your change (model swap, concurrency refactor, etc.). -4. Re-run with `--baseline ` and verify no - regressions. -5. Commit the baseline report to a designated location (e.g. - `docs/benchmarking/baselines/`) if it serves as a team reference. diff --git a/src/vidxp/benchmarks/cli.py b/src/vidxp/benchmarks/cli.py index e8f069dd..34fa5e2d 100644 --- a/src/vidxp/benchmarks/cli.py +++ b/src/vidxp/benchmarks/cli.py @@ -21,7 +21,6 @@ HIREST_DEFAULT_WINDOW_FRACTION, run_hirest, ) -from vidxp.benchmarks.latency import run_latency from vidxp.benchmarks.prepare import ( PreparationPlan, execute_preparation, @@ -575,134 +574,3 @@ def hirest_command( emit_json(metrics) else: rich_print(metrics) - - -@app.command("index-latency") -def index_latency_command( - ctx: typer.Context, - run_id: Annotated[str, typer.Option(help="Arbitrary label for this run.")], - modalities: Annotated[ - str, - typer.Option( - help="Comma-separated modality names: scene,actor,dialogue." - ), - ] = "scene", - videos: Annotated[ - int, - typer.Option(min=1, help="Number of synthetic clips to generate."), - ] = 1, - duration_seconds: Annotated[ - float, - typer.Option(min=0.1, help="Duration of each synthetic clip."), - ] = 8.0, - fps: Annotated[ - int, - typer.Option(min=1, help="Frame rate of synthetic clips."), - ] = 24, - resolution: Annotated[ - str, - typer.Option( - help="Synthetic clip resolution in WxH format (e.g. 320x180)." - ), - ] = "320x180", - repetitions: Annotated[ - int, - typer.Option(min=1, help="Number of times to repeat the run."), - ] = 1, - input_mode: Annotated[ - Literal["transcript", "transcribe"], - typer.Option( - help=( - "'transcript' supplies a synthetic transcript for dialogue " - "embedding (no real transcription). 'transcribe' runs " - "real whisper on audio (requires --audio-mode flite)." - ) - ), - ] = "transcript", - audio_mode: Annotated[ - Literal["none", "sine", "flite"], - typer.Option( - help=( - "Audio track for synthetic clips: 'none' (no audio), " - "'sine' (tone), or 'flite' (speech synthesis)." - ) - ), - ] = "none", - reset: Annotated[ - bool, - typer.Option(help="Clear any existing index before running."), - ] = False, - baseline: Annotated[ - Path | None, - typer.Option( - exists=True, - dir_okay=False, - help=( - "Path to a previous latency report JSON for regression " - "comparison." - ), - ), - ] = None, - baseline_tolerance: Annotated[ - float, - typer.Option( - min=0.0, - max=5.0, - help=( - "Relative regression tolerance. A stage mean slower by " - "more than this ratio flags as regression." - ), - ), - ] = 0.15, - json_output: Annotated[ - bool, - typer.Option("--json", help="Emit machine-readable JSON."), - ] = False, -) -> None: - """Run a reproducible indexing-latency benchmark on synthetic media.""" - - selected = [item.strip() for item in modalities.split(",") if item.strip()] - if not selected: - raise typer.BadParameter( - "At least one latency modality is required.", param_hint="--modalities" - ) - for modality in selected: - _require_benchmark_dependencies(modality) - - try: - parts = resolution.lower().split("x") - if len(parts) != 2: - raise ValueError - width, height = int(parts[0]), int(parts[1]) - if width <= 0 or height <= 0: - raise ValueError - except (IndexError, ValueError, AttributeError): - raise typer.BadParameter( - f"Invalid resolution: {resolution!r}. Use WxH, e.g. 320x180.", - param_hint="--resolution", - ) - - state = state_from_context(ctx) - report = run_latency( - run_id=run_id, - output_root=state.settings.data_dir / "benchmark_runs", - ffprobe=state.settings.ffprobe_executable, - ffmpeg=state.settings.ffmpeg_executable, - modalities=tuple(selected), - videos=videos, - duration_seconds=duration_seconds, - fps=fps, - width=width, - height=height, - repetitions=repetitions, - input_mode=input_mode, - audio_mode=audio_mode, - device=state.settings.runtime_backend, - reset=reset, - baseline_path=baseline, - baseline_tolerance=baseline_tolerance, - ) - if effective_output_format(state, json_output) == OutputFormat.json: - emit_json(report) - else: - rich_print(report) diff --git a/src/vidxp/benchmarks/latency.py b/src/vidxp/benchmarks/latency.py deleted file mode 100644 index 87272f57..00000000 --- a/src/vidxp/benchmarks/latency.py +++ /dev/null @@ -1,703 +0,0 @@ -from __future__ import annotations - -import json -import random -import subprocess -from dataclasses import dataclass -from pathlib import Path -from statistics import mean -from time import perf_counter -from typing import Any, Literal, Mapping, Sequence - -from vidxp.benchmarks.common import ( - append_failure, - benchmark_generation_id, - benchmark_media_id, - ensure_adapter_outputs, - record_adapter_manifest, -) -from vidxp.capabilities.registry import create_capability_registry -from vidxp.core.contracts import IndexConfig, VideoSource -from vidxp.core.manifest import ManifestStore, write_json_atomic -from vidxp.core.runner import run_index -from vidxp.core.storage import IndexStorage -from vidxp.infrastructure.local_index import LOCAL_INDEX_RUNTIME_CHECKS -from vidxp.media_runtime import inspect_media_runtime -from vidxp.runtime import ModelRuntime -from vidxp.settings import VidXPSettings - - -LATENCY_BENCHMARK = "latency" -LATENCY_SPLIT = "synthetic" -LATENCY_SCHEMA_VERSION = 1 -DEFAULT_CORPUS_SEED = 2026 -SUPPORTED_MODALITIES = ("scene", "actor", "dialogue") - -_STAGE_RATES: Mapping[str, str] = { - "scene": "scene_frames", - "actor": "actor_frames", - "frame_stream": "source_frames_advanced", - "dialogue_indexing": "dialogue_phrases", -} - -_VOCABULARY = ( - "the quick brown fox jumps over the lazy dog honest sunshine light " - "morning river ocean mountain garden flower silver golden copper " - "bright shadow shadow candle lantern window door table chair book " - "letter number station market kitchen garden bakery camera video " - "music voice speech word phrase moment memory journey story world " - "quiet calm gentle peaceful vivid warm cool bright dark soft loud" -).split() - - -def _peak_rss_bytes() -> int | None: - try: - import resource - except ImportError: - return None - return int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) - - -def rss_unit() -> Literal["bytes", "KiB"]: - return "bytes" if _sys_platform() == "darwin" else "KiB" - - -def _sys_platform() -> str: - import sys - - return sys.platform - - -@dataclass(frozen=True) -class SyntheticCorpusSpec: - videos: int - duration_seconds: float - fps: int - width: int - height: int - audio_mode: Literal["none", "sine", "flite"] - seed: int - - def public_record(self) -> dict[str, Any]: - return { - "videos": self.videos, - "duration_seconds": self.duration_seconds, - "fps": self.fps, - "width": self.width, - "height": self.height, - "audio_mode": self.audio_mode, - "seed": self.seed, - } - - -def validate_latency_options( - *, - modalities: Sequence[str], - videos: int, - duration_seconds: float, - fps: int, - width: int, - height: int, - repetitions: int, - input_mode: str, - audio_mode: str, - baseline_tolerance: float, -) -> tuple[str, ...]: - selected = tuple(dict.fromkeys(modalities)) - if not selected: - raise ValueError("At least one latency modality must be selected.") - unsupported = sorted(set(selected) - set(SUPPORTED_MODALITIES)) - if unsupported: - raise ValueError( - "Latency modalities must be a subset of " - + ", ".join(SUPPORTED_MODALITIES) - + "; unsupported: " - + ", ".join(unsupported) - ) - if videos <= 0: - raise ValueError("videos must be greater than zero.") - if duration_seconds <= 0: - raise ValueError("duration_seconds must be greater than zero.") - if fps <= 0: - raise ValueError("fps must be greater than zero.") - if width <= 0 or height <= 0: - raise ValueError("width and height must be greater than zero.") - if repetitions <= 0: - raise ValueError("repetitions must be greater than zero.") - if input_mode not in {"transcript", "transcribe"}: - raise ValueError("input_mode must be 'transcript' or 'transcribe'.") - if audio_mode not in {"none", "sine", "flite"}: - raise ValueError("audio_mode must be 'none', 'sine', or 'flite'.") - if "dialogue" in selected and input_mode == "transcribe": - if audio_mode != "flite": - raise ValueError( - "Real transcription requires a speech audio source; " - "use --audio-mode flite with --input-mode transcribe." - ) - if not 0 <= baseline_tolerance <= 5: - raise ValueError("baseline_tolerance must be between zero and five.") - return selected - - -def synthetic_transcript( - *, - duration_seconds: float, - seed: int, -) -> list[dict[str, Any]]: - generator = random.Random(seed) - strides = max(1, int(duration_seconds / 0.4)) - words = [generator.choice(_VOCABULARY) for _ in range(strides)] - span = duration_seconds / len(words) - word_events = [ - { - "word": word, - "start": round(index * span, 4), - "end": round((index + 1) * span, 4), - } - for index, word in enumerate(words) - ] - return [ - { - "text": " ".join(words), - "start": 0.0, - "end": duration_seconds, - "words": word_events, - } - ] - - -def _flite_text(seed: int) -> str: - generator = random.Random(seed) - words = [generator.choice(_VOCABULARY) for _ in range(24)] - return " ".join(words) - - -def build_clip_command( - *, - spec: SyntheticCorpusSpec, - ffmpeg: str, - destination: Path, -) -> list[str]: - compact = spec.width != 0 and spec.height != 0 - if not compact: - raise ValueError("The synthetic corpus requires positive dimensions.") - command = [ - ffmpeg, - "-y", - "-v", - "error", - "-f", - "lavfi", - "-i", - f"testsrc2=size={spec.width}x{spec.height}:rate={spec.fps}", - ] - if spec.audio_mode == "sine": - command += [ - "-f", - "lavfi", - "-i", - "sine=frequency=440:sample_rate=16000", - ] - elif spec.audio_mode == "flite": - command += [ - "-f", - "lavfi", - "-i", - f"flite=text='{_flite_text(spec.seed)}',sample_rate=16000", - ] - command += [ - "-t", - f"{spec.duration_seconds:g}", - "-c:v", - "libx264", - "-pix_fmt", - "yuv420p", - ] - if spec.audio_mode == "none": - command.append("-an") - else: - command += ["-c:a", "aac"] - command.append(str(destination)) - return command - - -def _probe_duration(ffprobe: str, path: Path) -> float: - completed = subprocess.run( - [ - ffprobe, - "-v", - "error", - "-show_entries", - "format=duration", - "-of", - "json", - str(path), - ], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=60, - ) - if completed.returncode != 0: - raise ValueError( - f"ffprobe could not read a generated clip: {path}" - ) - try: - duration = float(json.loads(completed.stdout)["format"]["duration"]) - except (KeyError, ValueError, TypeError, json.JSONDecodeError) as exc: - raise ValueError( - f"ffprobe returned an invalid duration for {path}." - ) from exc - if duration <= 0: - raise ValueError(f"ffprobe reported a non-positive duration for {path}.") - return duration - - -def generate_synthetic_corpus( - *, - spec: SyntheticCorpusSpec, - directory: str | Path, - ffprobe: str, - ffmpeg: str, - audio_mode: str | None = None, -) -> list[Path]: - runtime_status = inspect_media_runtime( - ffprobe=ffprobe, - ffmpeg=ffmpeg, - ) - if not runtime_status.ready: - raise ValueError( - "The latency benchmark requires FFmpeg and ffprobe to generate " - "the synthetic corpus. Run `vidxp init`, then retry." - ) - destination = Path(directory) - destination.mkdir(parents=True, exist_ok=True) - clips = [] - for index in range(spec.videos): - path = destination / f"clip-{index:03d}.mp4" - command = build_clip_command( - spec=spec, - ffmpeg=ffmpeg, - destination=path, - ) - completed = subprocess.run( - command, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=600, - ) - if completed.returncode != 0: - stderr = (completed.stderr or "").strip() - if spec.audio_mode == "flite" and stderr: - raise ValueError( - "FFmpeg could not apply the flite speech filter " - f"(libflite likely unavailable): {stderr}" - ) - raise ValueError(f"FFmpeg could not generate {path}: {stderr}") - if not path.is_file() or path.stat().st_size == 0: - raise ValueError(f"FFmpeg did not produce {path}.") - _probe_duration(ffprobe, path) - clips.append(path) - return clips - - -def _per_video_stages( - manifest: Mapping[str, Any], - video_id: str, -) -> dict[str, float]: - video = manifest["videos"].get(video_id) - if video is None or video.get("state") == "failed": - return {} - return { - str(stage): float(entry["seconds"]) - for stage, entry in (video.get("stages") or {}).items() - if entry.get("state") != "incomplete" - and float(entry.get("seconds", 0.0)) > 0 - } - - -def _summary_ratio( - manifest: Mapping[str, Any], - video_id: str, - *, - metric: str, - stage: str, -) -> float | None: - video = manifest["videos"].get(video_id) or {} - summary = video.get("summary") or {} - seconds = _per_video_stages(manifest, video_id).get(stage) - count = summary.get(metric) - if seconds is None or count is None or seconds <= 0 or count <= 0: - return None - return float(count) / seconds - - -def aggregate_latency_runs( - manifests: Sequence[Mapping[str, Any]], - *, - wall_seconds: Sequence[float], - peak_rss_samples: Sequence[int | None], -) -> dict[str, Any]: - if len(manifests) != len(wall_seconds): - raise ValueError( - "Every latency repetition requires a wall-clock sample." - ) - stage_samples: dict[str, list[float]] = {} - rate_samples: dict[str, list[float]] = {} - per_video: list[dict[str, Any]] = [] - processed_frames = 0 - record_counts: dict[str, int] = {} - for repetition, manifest in enumerate(manifests): - processed_frames += int(manifest.get("processed_frames", 0)) - for modality, count in (manifest.get("record_counts") or {}).items(): - record_counts[modality] = record_counts.get(modality, 0) + int(count) - for video_id in sorted(manifest.get("videos", {})): - stages = _per_video_stages(manifest, video_id) - if not stages: - continue - for stage, seconds in stages.items(): - stage_samples.setdefault(stage, []).append(seconds) - rate_stages = { - stage: _summary_ratio( - manifest, - video_id, - metric=_STAGE_RATES[stage], - stage=stage, - ) - for stage in _STAGE_RATES - if stage in stages - } - for stage, rate in rate_stages.items(): - if rate is None: - continue - rate_samples.setdefault(stage, []).append(rate) - video = manifest["videos"].get(video_id, {}) - per_video.append( - { - "repetition": repetition, - "video_id": video_id, - "wall_seconds": ( - wall_seconds[repetition] - ), - "stages": dict(sorted(stages.items())), - "summary": video.get("summary", {}), - } - ) - stages: dict[str, dict[str, Any]] = {} - for stage, samples in stage_samples.items(): - values = sorted(samples) - summary: dict[str, Any] = { - "runs": len(values), - "mean_seconds": mean(values), - "min_seconds": values[0], - "max_seconds": values[-1], - } - rates = rate_samples.get(stage) - if rates: - summary["rate_per_second"] = mean(rates) - stages[stage] = summary - approximate_rss = [ - sample for sample in peak_rss_samples if sample is not None - ] - summary = { - "wall_seconds": { - "runs": len(wall_seconds), - "mean_seconds": mean(wall_seconds), - "min_seconds": min(wall_seconds), - "max_seconds": max(wall_seconds), - }, - "peak_rss": { - "unit": rss_unit(), - "samples": len(approximate_rss), - "value": int(max(approximate_rss)) if approximate_rss else None, - }, - } - return { - "per_video": per_video, - "stages": dict(sorted(stages.items())), - "summary": summary, - "processed_frames": processed_frames, - "record_counts": dict(sorted(record_counts.items())), - } - - -def _validate_baseline_compatibility( - report: Mapping[str, Any], - baseline: Mapping[str, Any], -) -> None: - baseline_corpus = baseline.get("corpus") or {} - report_corpus = report.get("corpus") or {} - corpus_keys = ("videos", "duration_seconds", "fps", "width", "height", "audio_mode", "seed") - for key in corpus_keys: - report_val = report_corpus.get(key) - baseline_val = baseline_corpus.get(key) - if report_val is not None and baseline_val is not None and report_val != baseline_val: - raise ValueError( - f"Baseline corpus mismatch: {key!r} is {baseline_val!r}, " - f"current is {report_val!r}." - ) - for key in ("input_mode", "device"): - report_val = report.get(key) - baseline_val = baseline.get(key) - if report_val is not None and baseline_val is not None and report_val != baseline_val: - raise ValueError( - f"Baseline {key!r} mismatch: {baseline_val!r}, current is {report_val!r}." - ) - report_mods = sorted(report.get("modalities") or []) - baseline_mods = sorted(baseline.get("modalities") or []) - if report_mods and baseline_mods and report_mods != baseline_mods: - raise ValueError( - f"Baseline modalities mismatch: {baseline_mods}, current is {report_mods}." - ) - - -def compare_baseline( - report: Mapping[str, Any], - baseline: Mapping[str, Any], - *, - tolerance: float, -) -> dict[str, Any]: - if tolerance < 0: - raise ValueError("Baseline tolerance must be nonnegative.") - comparisons: dict[str, dict[str, Any]] = {} - regressions: list[str] = [] - for stage, previous in (baseline.get("stages") or {}).items(): - current = (report.get("stages") or {}).get(stage) - if current is None or not previous.get("runs"): - continue - old_mean = float(previous["mean_seconds"]) - new_mean = float(current["mean_seconds"]) - if old_mean <= 0: - continue - delta = new_mean / old_mean - 1.0 - comparisons[stage] = { - "old_mean_seconds": old_mean, - "new_mean_seconds": new_mean, - "delta_ratio": delta, - "regressed": delta > tolerance, - } - if delta > tolerance: - regressions.append(stage) - return { - "schema_version": LATENCY_SCHEMA_VERSION, - "tolerance": tolerance, - "stages": dict(sorted(comparisons.items())), - "regressions": regressions, - "verdict": "fail" if regressions else "pass", - } - - -def build_latency_sources( - *, - clips: Sequence[Path], - spec: SyntheticCorpusSpec, - input_mode: Literal["transcript", "transcribe"], -) -> list[VideoSource]: - sources = [] - for index, path in enumerate(clips): - transcript = ( - synthetic_transcript( - duration_seconds=spec.duration_seconds, - seed=spec.seed + index, - ) - if input_mode == "transcript" - else None - ) - sources.append( - VideoSource( - video_id=benchmark_media_id( - LATENCY_BENCHMARK, - f"clip-{index:03d}", - ), - path=path, - source_name=f"clip-{index:03d}.mp4", - transcript=transcript, - ) - ) - return sources - - -def run_latency( - *, - run_id: str, - output_root: str | Path = "benchmark_runs", - ffprobe: str = "ffprobe", - ffmpeg: str = "ffmpeg", - modalities: Sequence[str] = ("scene",), - videos: int = 1, - duration_seconds: float = 8.0, - fps: int = 24, - width: int = 320, - height: int = 180, - repetitions: int = 1, - input_mode: Literal["transcript", "transcribe"] = "transcript", - audio_mode: Literal["none", "sine", "flite"] = "none", - device: str = "cpu", - reset: bool = False, - baseline_path: str | Path | None = None, - baseline_tolerance: float = 0.15, -) -> dict[str, Any]: - selected = validate_latency_options( - modalities=modalities, - videos=videos, - duration_seconds=duration_seconds, - fps=fps, - width=width, - height=height, - repetitions=repetitions, - input_mode=input_mode, - audio_mode=audio_mode, - baseline_tolerance=baseline_tolerance, - ) - spec = SyntheticCorpusSpec( - videos=videos, - duration_seconds=duration_seconds, - fps=fps, - width=width, - height=height, - audio_mode=audio_mode, - seed=DEFAULT_CORPUS_SEED, - ) - config = IndexConfig( - dataset=LATENCY_BENCHMARK, - split=LATENCY_SPLIT, - run_id=run_id, - enabled_modalities=selected, - device=device, - output_root=output_root, - generation_id=benchmark_generation_id( - LATENCY_BENCHMARK, - LATENCY_SPLIT, - run_id, - ), - ) - run_directory = config.run_directory - registry = create_capability_registry( - platform_runtime_checks=LOCAL_INDEX_RUNTIME_CHECKS - ) - runtime = ModelRuntime( - VidXPSettings( - repository_root=run_directory, - runtime_backend=device, - ), - allowed_specs=registry.model_specs(), - ) - ensure_adapter_outputs(run_directory) - manifests: list[dict[str, Any]] = [] - wall_samples: list[float] = [] - rss_samples: list[int | None] = [] - try: - clips = generate_synthetic_corpus( - spec=spec, - directory=run_directory / "corpus", - ffprobe=ffprobe, - ffmpeg=ffmpeg, - ) - sources = build_latency_sources( - clips=clips, - spec=spec, - input_mode=input_mode, - ) - for _ in range(repetitions): - started = perf_counter() - with IndexStorage(config) as storage: - manifest = run_index( - sources, - config, - reset=reset, - storage=storage, - manifest_store=ManifestStore( - config, - registry=registry, - runtime=runtime, - ), - registry=registry, - runtime=runtime, - ) - store_size_bytes = storage.size_bytes() - wall_samples.append(perf_counter() - started) - rss_samples.append(_peak_rss_bytes()) - manifests.append( - { - **manifest, - "store_size_bytes_at_commit": store_size_bytes, - } - ) - aggregated = aggregate_latency_runs( - manifests, - wall_seconds=wall_samples, - peak_rss_samples=rss_samples, - ) - report = { - "schema_version": LATENCY_SCHEMA_VERSION, - "benchmark": LATENCY_BENCHMARK, - "run_id": run_id, - "created_at": manifests[-1].get("completed_at"), - "corpus": spec.public_record(), - "input_mode": input_mode, - "modalities": list(selected), - "device": device, - "repetitions": repetitions, - "git": manifests[-1].get("git"), - "environment": manifests[-1].get("environment"), - "config_fingerprint": manifests[-1].get("config_fingerprint"), - "record_counts": aggregated["record_counts"], - "processed_frames": aggregated["processed_frames"], - "summary": aggregated["summary"], - "stages": aggregated["stages"], - "per_video": aggregated["per_video"], - "baseline": None, - } - if baseline_path is not None: - try: - baseline = json.loads( - Path(baseline_path).read_text(encoding="utf-8") - ) - except (OSError, json.JSONDecodeError) as exc: - raise ValueError( - f"Baseline report is not readable JSON: {baseline_path}" - ) from exc - _validate_baseline_compatibility(report, baseline) - report["baseline"] = compare_baseline( - report, - baseline, - tolerance=baseline_tolerance, - ) - write_json_atomic(run_directory / "report.json", report) - record_adapter_manifest( - run_directory, - benchmark=LATENCY_BENCHMARK, - subset={ - "label": f"latency_{run_id}", - "modalities": list(selected), - "video_count": videos, - "duration_seconds": duration_seconds, - "repetitions": repetitions, - }, - artifacts=[], - state="complete", - details={ - "device": device, - "input_mode": input_mode, - "audio_mode": audio_mode, - "corpus": spec.public_record(), - "result_classification": "performance_benchmark_not_quality_score", - }, - ) - return report - except BaseException as error: - append_failure(run_directory, stage="latency_adapter", error=error) - record_adapter_manifest( - run_directory, - benchmark=LATENCY_BENCHMARK, - subset={ - "label": f"latency_{run_id}", - "modalities": list(selected), - }, - artifacts=[], - state="failed", - ) - raise \ No newline at end of file diff --git a/tests/test_benchmark_latency.py b/tests/test_benchmark_latency.py deleted file mode 100644 index 6cbd7057..00000000 --- a/tests/test_benchmark_latency.py +++ /dev/null @@ -1,329 +0,0 @@ -from __future__ import annotations - -import unittest -from pathlib import Path - -from vidxp.benchmarks.latency import ( - SyntheticCorpusSpec, - aggregate_latency_runs, - build_clip_command, - build_latency_sources, - compare_baseline, - synthetic_transcript, - validate_latency_options, -) - - -class LatencyValidationTests(unittest.TestCase): - def test_validates_default_options(self): - selected = validate_latency_options( - modalities=("scene",), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=3, - input_mode="transcript", - audio_mode="none", - baseline_tolerance=0.15, - ) - self.assertEqual(selected, ("scene",)) - - def test_rejects_empty_modalities(self): - with self.assertRaises(ValueError): - validate_latency_options( - modalities=(), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcript", - audio_mode="none", - baseline_tolerance=0.15, - ) - - def test_rejects_unsupported_modality(self): - with self.assertRaises(ValueError): - validate_latency_options( - modalities=("scene", "ocr"), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcript", - audio_mode="none", - baseline_tolerance=0.15, - ) - - def test_rejects_transcribe_without_flite(self): - with self.assertRaises(ValueError): - validate_latency_options( - modalities=("scene", "dialogue"), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcribe", - audio_mode="sine", - baseline_tolerance=0.15, - ) - - def test_accepts_transcribe_with_flite(self): - selected = validate_latency_options( - modalities=("dialogue",), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcribe", - audio_mode="flite", - baseline_tolerance=0.15, - ) - self.assertEqual(selected, ("dialogue",)) - - def test_deduplicates_modalities(self): - selected = validate_latency_options( - modalities=("scene", "scene", "actor"), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcript", - audio_mode="none", - baseline_tolerance=0.15, - ) - self.assertEqual(selected, ("scene", "actor")) - - -class SyntheticTranscriptTests(unittest.TestCase): - def test_returns_one_segment_with_words(self): - transcript = synthetic_transcript(duration_seconds=10.0, seed=42) - self.assertEqual(len(transcript), 1) - segment = transcript[0] - self.assertGreater(len(segment["text"]), 0) - self.assertEqual(segment["start"], 0.0) - self.assertGreater(segment["end"], 0.0) - self.assertGreater(len(segment["words"]), 0) - for word in segment["words"]: - self.assertIn("word", word) - self.assertIsInstance(word["start"], float) - self.assertIsInstance(word["end"], float) - - def test_deterministic_across_calls(self): - first = synthetic_transcript(duration_seconds=5.0, seed=99) - second = synthetic_transcript(duration_seconds=5.0, seed=99) - self.assertEqual(first, second) - - def test_different_seeds_differ(self): - first = synthetic_transcript(duration_seconds=5.0, seed=99) - second = synthetic_transcript(duration_seconds=5.0, seed=100) - self.assertNotEqual(first, second) - - -class BuildClipCommandTests(unittest.TestCase): - def test_no_audio_default(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=8.0, fps=24, - width=320, height=180, audio_mode="none", seed=42, - ) - command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) - self.assertIn("testsrc2=size=320x180:rate=24", command) - self.assertIn("-an", command) - self.assertNotIn("-c:a", command) - - def test_sine_audio_adds_aac(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=8.0, fps=24, - width=320, height=180, audio_mode="sine", seed=42, - ) - command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) - self.assertIn("sine=frequency=440:sample_rate=16000", command) - self.assertIn("-c:a", command) - self.assertNotIn("-an", command) - - def test_flite_audio_contains_filter_ref(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=8.0, fps=24, - width=320, height=180, audio_mode="flite", seed=42, - ) - command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) - flite_args = [arg for arg in command if "flite=text=" in arg] - self.assertEqual(len(flite_args), 1) - - def test_duration_is_formatted(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=3.5, fps=30, - width=640, height=480, audio_mode="none", seed=0, - ) - command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("clip.mp4")) - idx = command.index("-t") - self.assertEqual(command[idx + 1], "3.5") - self.assertIn("testsrc2=size=640x480:rate=30", command) - - -class BuildSourcesTests(unittest.TestCase): - def test_transcript_attached_in_input_mode(self): - spec = SyntheticCorpusSpec( - videos=2, duration_seconds=4.0, fps=24, - width=320, height=180, audio_mode="none", seed=42, - ) - clips = [Path(f"{i}.mp4") for i in range(2)] - sources = build_latency_sources(clips=clips, spec=spec, input_mode="transcript") - self.assertEqual(len(sources), 2) - for index, source in enumerate(sources): - self.assertIsNotNone(source.transcript) - self.assertIsNotNone(source.path) - self.assertIsNotNone(source.video_id) - self.assertEqual(source.source_name, f"clip-{index:03d}.mp4") - - def test_no_transcript_in_transcribe_mode(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=4.0, fps=24, - width=320, height=180, audio_mode="flite", seed=42, - ) - sources = build_latency_sources(clips=[Path("0.mp4")], spec=spec, input_mode="transcribe") - for source in sources: - self.assertIsNone(source.transcript) - - -class AggregateMetricsTests(unittest.TestCase): - def _sample_manifest(self, scene_seconds, scene_frames, actor_seconds, actor_frames): - return { - "processed_frames": scene_frames, - "record_counts": {"scene": scene_frames, "actor": actor_frames}, - "git": {"commit": "abc", "dirty": False}, - "environment": {"platform": "test"}, - "config_fingerprint": "fp1", - "completed_at": "2026-01-01T00:00:00", - "videos": { - "vid-1": { - "state": "complete", - "summary": { - "scene_frames": scene_frames, - "actor_frames": actor_frames, - "source_frames_advanced": scene_frames + 100, - }, - "stages": { - "scene": {"seconds": scene_seconds, "state": ""}, - "actor": {"seconds": actor_seconds, "state": ""}, - "frame_stream": {"seconds": 0.5, "state": ""}, - }, - } - }, - } - - def test_aggregates_single_manifest(self): - result = aggregate_latency_runs( - [self._sample_manifest(2.0, 8, 1.5, 3)], - wall_seconds=[3.5], - peak_rss_samples=[100000], - ) - self.assertEqual(result["processed_frames"], 8) - self.assertEqual(result["record_counts"], {"actor": 3, "scene": 8}) - self.assertIn("scene", result["stages"]) - self.assertAlmostEqual(result["stages"]["scene"]["mean_seconds"], 2.0) - self.assertAlmostEqual(result["stages"]["scene"]["rate_per_second"], 4.0) - self.assertAlmostEqual(result["stages"]["actor"]["mean_seconds"], 1.5) - self.assertAlmostEqual(result["summary"]["wall_seconds"]["mean_seconds"], 3.5) - - def test_aggregates_multiple_manifests(self): - m1 = self._sample_manifest(2.0, 8, 1.5, 3) - m2 = self._sample_manifest(2.5, 10, 2.0, 4) - result = aggregate_latency_runs( - [m1, m2], - wall_seconds=[3.5, 4.5], - peak_rss_samples=[100000, 120000], - ) - self.assertEqual(result["processed_frames"], 18) - self.assertAlmostEqual(result["stages"]["scene"]["mean_seconds"], 2.25) - self.assertAlmostEqual(result["stages"]["scene"]["min_seconds"], 2.0) - self.assertAlmostEqual(result["stages"]["scene"]["max_seconds"], 2.5) - self.assertEqual(len(result["per_video"]), 2) - self.assertAlmostEqual( - result["summary"]["wall_seconds"]["mean_seconds"], - 4.0, - ) - - def test_skips_failed_videos(self): - manifest = { - "processed_frames": 0, - "record_counts": {}, - "git": {}, - "environment": {}, - "config_fingerprint": "fp", - "completed_at": "", - "videos": { - "vid-1": { - "state": "failed", - "summary": {}, - "stages": {}, - } - }, - } - result = aggregate_latency_runs( - [manifest], - wall_seconds=[1.0], - peak_rss_samples=[None], - ) - self.assertEqual(result["processed_frames"], 0) - self.assertEqual(result["stages"], {}) - - -class CompareBaselineTests(unittest.TestCase): - def test_no_baseline_stages_returns_empty(self): - report = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} - baseline = {"stages": {}} - result = compare_baseline(report, baseline, tolerance=0.1) - self.assertEqual(result["stages"], {}) - self.assertEqual(result["regressions"], []) - self.assertEqual(result["verdict"], "pass") - - def test_regression_detected(self): - report = {"stages": {"scene": {"mean_seconds": 3.0, "runs": 1}}} - baseline = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} - result = compare_baseline(report, baseline, tolerance=0.1) - self.assertIn("scene", result["stages"]) - self.assertAlmostEqual( - result["stages"]["scene"]["delta_ratio"], 0.5 - ) - self.assertTrue(result["stages"]["scene"]["regressed"]) - self.assertEqual(result["regressions"], ["scene"]) - self.assertEqual(result["verdict"], "fail") - - def test_improvement_not_regression(self): - report = {"stages": {"scene": {"mean_seconds": 1.5, "runs": 1}}} - baseline = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} - result = compare_baseline(report, baseline, tolerance=0.1) - self.assertFalse(result["stages"]["scene"]["regressed"]) - self.assertEqual(result["regressions"], []) - self.assertEqual(result["verdict"], "pass") - - -class CorpusSpecTests(unittest.TestCase): - def test_public_record_roundtrip(self): - spec = SyntheticCorpusSpec( - videos=2, duration_seconds=8.0, fps=24, - width=320, height=180, audio_mode="none", seed=42, - ) - record = spec.public_record() - self.assertEqual(record["videos"], 2) - self.assertEqual(record["duration_seconds"], 8.0) - self.assertEqual(record["audio_mode"], "none") - - def test_flite_mode_recorded(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=5.0, fps=30, - width=640, height=480, audio_mode="flite", seed=7, - ) - self.assertEqual(spec.public_record()["audio_mode"], "flite") From ca26095ab4b715ab2b9aae78db4488c4bb7d4f53 Mon Sep 17 00:00:00 2001 From: Mahnoor-Zaffar <1999mahnoor@gmail.com> Date: Thu, 13 Aug 2026 00:24:25 +0500 Subject: [PATCH 3/6] perf(core): run visual participants (actor, scene) in parallel within shared decode stream Refactors _consume_visual_stream so the decode loop runs on a thread and each visual participant (actor via OpenCV, scene via torch) gets its own worker thread consuming from a per-participant queue. This lets actor and scene overlap on CPU since both libraries release the GIL during computation. - Decode thread pushes RGB batches to bounded queues (maxsize=4) for backpressure - Each participant worker filters its own samples and calls process() independently - Exceptions from any thread propagate via a shared error list - Cancellation checked in both decode and participant workers with a 0.2s polling timeout on the queue get to stay responsive - The single-decode property (one pass through iter_frame_batches) is preserved; actor stays in-order (single FIFO consumer) No changes to runner.py, IndexConfig, ResourceScheduler, or any capability's indexing logic. All 5 new threading tests pass. --- src/vidxp/capabilities/visual.py | 158 ++++++++++++++------- tests/test_visual_threading.py | 236 +++++++++++++++++++++++++++++++ 2 files changed, 346 insertions(+), 48 deletions(-) create mode 100644 tests/test_visual_threading.py diff --git a/src/vidxp/capabilities/visual.py b/src/vidxp/capabilities/visual.py index a49e2dd6..d60f5d63 100644 --- a/src/vidxp/capabilities/visual.py +++ b/src/vidxp/capabilities/visual.py @@ -1,5 +1,7 @@ from __future__ import annotations +import queue +import threading from dataclasses import dataclass from time import perf_counter from typing import Any, Protocol, Sequence @@ -147,59 +149,119 @@ def _consume_visual_stream( progress: ProgressCallback | None, timings: dict[str, float], ) -> FrameStreamStats: + participant_queues = { + p.name: queue.Queue(maxsize=4) + for p in participants + } + errors: list[Exception] = [] + errors_lock = threading.Lock() stream_stats = FrameStreamStats() - stream = iter( - iter_frame_batches( - source.path, - samplings=tuple( - participant.sampling for participant in participants - ), - batch_size=max( - participant.processor.batch_size(config) - for participant in participants - ), - cancellation=cancellation, - stats=stream_stats, - ) - ) - while True: - stream_started = perf_counter() + + def _record_error(exc: Exception) -> None: + with errors_lock: + errors.append(exc) + + def _decode_worker() -> None: try: - samples = next(stream) - except StopIteration: - timings["frame_stream"] += perf_counter() - stream_started - break - rgb_samples = _rgb_samples(samples) - timings["frame_stream"] += perf_counter() - stream_started - - for participant in participants: - participant_samples = [ - sample - for sample in rgb_samples - if _is_participant_sample(sample, participant) - ] - if not participant_samples: - continue - processor_started = perf_counter() - participant.processor.process( - participant_samples, - state=participant.state, - info=info, - config=config, - storage=storage, - cancellation=cancellation, - ) - timings[participant.name] += ( - perf_counter() - processor_started + stream = iter( + iter_frame_batches( + source.path, + samplings=tuple( + p.sampling for p in participants + ), + batch_size=max( + p.processor.batch_size(config) + for p in participants + ), + cancellation=cancellation, + stats=stream_stats, + ) ) + while True: + stream_started = perf_counter() + try: + samples = next(stream) + except StopIteration: + timings["frame_stream"] += ( + perf_counter() - stream_started + ) + break + rgb_samples = _rgb_samples(samples) + timings["frame_stream"] += ( + perf_counter() - stream_started + ) + for participant_queue in participant_queues.values(): + participant_queue.put(rgb_samples) + report_progress( + progress, + "visual_indexing", + "Indexing the shared sampled-frame stream.", + stream_stats.frames_materialized, + expected, + ) + except Exception as exc: + _record_error(exc) + finally: + for q in participant_queues.values(): + q.put(None) - report_progress( - progress, - "visual_indexing", - "Indexing the shared sampled-frame stream.", - stream_stats.frames_materialized, - expected, + def _participant_worker( + participant: _Participant, + work_queue: queue.Queue, + ) -> None: + try: + while True: + cancellation.raise_if_cancelled() + try: + batch = work_queue.get(timeout=0.2) + except queue.Empty: + continue + if batch is None: + break + participant_samples = [ + sample + for sample in batch + if _is_participant_sample(sample, participant) + ] + if not participant_samples: + continue + started = perf_counter() + participant.processor.process( + participant_samples, + state=participant.state, + info=info, + config=config, + storage=storage, + cancellation=cancellation, + ) + timings[participant.name] += ( + perf_counter() - started + ) + except Exception as exc: + _record_error(exc) + + decode_thread = threading.Thread( + target=_decode_worker, name="vidxp-visual-decode" + ) + participant_threads = [ + threading.Thread( + target=_participant_worker, + args=(p, participant_queues[p.name]), + name=f"vidxp-visual-{p.name}", ) + for p in participants + ] + + decode_thread.start() + for t in participant_threads: + t.start() + decode_thread.join() + for t in participant_threads: + t.join() + + if errors: + raise errors[0] + return stream_stats diff --git a/tests/test_visual_threading.py b/tests/test_visual_threading.py new file mode 100644 index 00000000..11e58a0b --- /dev/null +++ b/tests/test_visual_threading.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import threading +import unittest +from unittest.mock import Mock, patch + +from vidxp.capabilities.visual import ( + _Participant, + _consume_visual_stream, +) +from vidxp.core.contracts import CancellationToken, IndexConfig, VideoSource +from vidxp.core.video import FrameSample, FrameSampling, FrameStreamStats + + +def _mock_participant( + name: str, + *, + batch_size: int = 4, + frame_stride: int = 1, +) -> _Participant: + processor = Mock() + processor.batch_size.return_value = batch_size + processor.sampling.return_value = FrameSampling(frame_stride=frame_stride) + processor.state = object() + return _Participant( + name=name, + processor=processor, + state=processor.state, + sampling=FrameSampling(frame_stride=frame_stride), + ) + + +def _sample(frame_index: int, timestamp: float) -> FrameSample: + return FrameSample(frame_index=frame_index, timestamp=timestamp, frame=object()) + + +class ConsumeVisualStreamTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls._rgb_patch = patch( + "vidxp.capabilities.visual._rgb_samples", + side_effect=lambda samples: samples, + ) + cls._rgb_patch.start() + + @classmethod + def tearDownClass(cls): + cls._rgb_patch.stop() + + def test_single_participant_processes_all_samples(self): + scene = _mock_participant("scene") + source = VideoSource(video_id="v1", path="unused.mp4") + config = IndexConfig(video_id="v1", enabled_modalities=("scene",)) + storage = Mock() + cancellation = CancellationToken() + timings = {"frame_stream": 0.0, "scene": 0.3} + batches = [ + [_sample(0, 0.0), _sample(1, 0.04)], + ] + + def fake_stream(path, *, stats=None, **kw): + stats.frames_advanced = 2 + stats.frames_materialized = 2 + return iter(batches) + + with patch( + "vidxp.capabilities.visual.iter_frame_batches", + side_effect=fake_stream, + ): + result = _consume_visual_stream( + source, + participants=[scene], + expected=2, + info=Mock(fps=24, frame_count=2, duration=0.08, width=2, height=2), + config=config, + storage=storage, + cancellation=cancellation, + progress=None, + timings=timings, + ) + + self.assertEqual(result.frames_advanced, 2) + self.assertEqual(result.frames_materialized, 2) + scene.processor.process.assert_called_once() + call_args = scene.processor.process.call_args[0] + self.assertEqual(len(call_args[0]), 2) + + def test_two_participants_each_receive_own_samples(self): + scene = _mock_participant("scene", frame_stride=1) + actor = _mock_participant("actor", frame_stride=2) + source = VideoSource(video_id="v1", path="unused.mp4") + config = IndexConfig(video_id="v1", enabled_modalities=("scene", "actor")) + storage = Mock() + cancellation = CancellationToken() + timings = {"frame_stream": 0.0, "scene": 0.3, "actor": 0.2} + batches = [ + [_sample(0, 0.0), _sample(1, 0.04), _sample(2, 0.08)], + [_sample(3, 0.12)], + ] + + def fake_stream(path, *, stats=None, **kw): + stats.frames_advanced = 4 + stats.frames_materialized = 4 + return iter(batches) + + with patch( + "vidxp.capabilities.visual.iter_frame_batches", + side_effect=fake_stream, + ): + _consume_visual_stream( + source, + participants=[scene, actor], + expected=4, + info=Mock(fps=24, frame_count=4, duration=0.16, width=2, height=2), + config=config, + storage=storage, + cancellation=cancellation, + progress=None, + timings=timings, + ) + + scene_sample_count = sum( + len(call[0][0]) for call in scene.processor.process.call_args_list + ) + actor_sample_count = sum( + len(call[0][0]) for call in actor.processor.process.call_args_list + ) + all_actor_samples = [ + s for call in actor.processor.process.call_args_list for s in call[0][0] + ] + self.assertEqual(scene_sample_count, 4) + self.assertEqual(actor_sample_count, 2) + for sample in all_actor_samples: + self.assertEqual(sample.frame_index % 2, 0) + + def test_cancellation_between_batches_stops_early(self): + scene = _mock_participant("scene") + source = VideoSource(video_id="v1", path="unused.mp4") + config = IndexConfig(video_id="v1", enabled_modalities=("scene",)) + storage = Mock() + cancellation = CancellationToken() + timings = {"frame_stream": 0.0, "scene": 0.3} + proceed = threading.Event() + + def blocking_batch(path, *, stats=None, cancellation=None, **kw): + yield [_sample(0, 0.0)] + proceed.wait(timeout=5) + cancellation.raise_if_cancelled() + yield [_sample(1, 0.04)] + + with patch( + "vidxp.capabilities.visual.iter_frame_batches", + side_effect=blocking_batch, + ): + from vidxp.core.contracts import IndexCancelledError + + cancel_timer = threading.Timer(0.1, lambda: (proceed.set(), cancellation.cancel())) + cancel_timer.start() + with self.assertRaises(IndexCancelledError): + _consume_visual_stream( + source, + participants=[scene], + expected=2, + info=Mock(fps=24, frame_count=2, duration=0.08, width=2, height=2), + config=config, + storage=storage, + cancellation=cancellation, + progress=None, + timings=timings, + ) + cancel_timer.cancel() + + def test_error_in_decode_propagates(self): + scene = _mock_participant("scene") + source = VideoSource(video_id="v1", path="unused.mp4") + config = IndexConfig(video_id="v1", enabled_modalities=("scene",)) + storage = Mock() + cancellation = CancellationToken() + timings = {"frame_stream": 0.0, "scene": 0.3} + + def broken_stream(path, *, stats=None, **kw): + stats = stats or FrameStreamStats() + yield [_sample(0, 0.0)] + raise RuntimeError("decode failure") + + with patch( + "vidxp.capabilities.visual.iter_frame_batches", + side_effect=broken_stream, + ): + with self.assertRaises(RuntimeError): + _consume_visual_stream( + source, + participants=[scene], + expected=2, + info=Mock(fps=24, frame_count=2, duration=0.08, width=2, height=2), + config=config, + storage=storage, + cancellation=cancellation, + progress=None, + timings=timings, + ) + + def test_error_in_participant_propagates(self): + scene = _mock_participant("scene") + source = VideoSource(video_id="v1", path="unused.mp4") + config = IndexConfig(video_id="v1", enabled_modalities=("scene",)) + storage = Mock() + cancellation = CancellationToken() + timings = {"frame_stream": 0.0, "scene": 0.3} + + def broken_process(samples, **kw): + raise ValueError("participant error") + + scene.processor.process = broken_process + + batches = [[_sample(0, 0.0)]] + + def fake_stream(path, *, stats=None, **kw): + return iter(batches) + + with patch( + "vidxp.capabilities.visual.iter_frame_batches", + side_effect=fake_stream, + ): + with self.assertRaises(ValueError): + _consume_visual_stream( + source, + participants=[scene], + expected=1, + info=Mock(fps=24, frame_count=1, duration=0.04, width=2, height=2), + config=config, + storage=storage, + cancellation=cancellation, + progress=None, + timings=timings, + ) From 7d7ed78ec7ab06c793d9e14ad3c02c6d212a6721 Mon Sep 17 00:00:00 2001 From: Mahnoor-Zaffar <1999mahnoor@gmail.com> Date: Sat, 15 Aug 2026 23:10:29 +0500 Subject: [PATCH 4/6] fix(capabilities): avoid deadlock when a visual participant fails mid-stream --- src/vidxp/capabilities/visual.py | 16 +++++++-- tests/test_visual_threading.py | 58 ++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/src/vidxp/capabilities/visual.py b/src/vidxp/capabilities/visual.py index d60f5d63..362a1c22 100644 --- a/src/vidxp/capabilities/visual.py +++ b/src/vidxp/capabilities/visual.py @@ -191,7 +191,16 @@ def _decode_worker() -> None: perf_counter() - stream_started ) for participant_queue in participant_queues.values(): - participant_queue.put(rgb_samples) + while True: + try: + participant_queue.put( + rgb_samples, timeout=0.2 + ) + break + except queue.Full: + if errors: + return + cancellation.raise_if_cancelled() report_progress( progress, "visual_indexing", @@ -203,7 +212,10 @@ def _decode_worker() -> None: _record_error(exc) finally: for q in participant_queues.values(): - q.put(None) + try: + q.put_nowait(None) + except queue.Full: + pass def _participant_worker( participant: _Participant, diff --git a/tests/test_visual_threading.py b/tests/test_visual_threading.py index 11e58a0b..937baaf2 100644 --- a/tests/test_visual_threading.py +++ b/tests/test_visual_threading.py @@ -234,3 +234,61 @@ def fake_stream(path, *, stats=None, **kw): progress=None, timings=timings, ) + + def test_error_in_participant_mid_stream_propagates_without_deadlock(self): + scene = _mock_participant("scene") + actor = _mock_participant("actor") + source = VideoSource(video_id="v1", path="unused.mp4") + config = IndexConfig(video_id="v1", enabled_modalities=("scene", "actor")) + storage = Mock() + cancellation = CancellationToken() + timings = {"frame_stream": 0.0, "scene": 0.3, "actor": 0.2} + + def broken_process(samples, **kw): + raise ValueError("actor failure mid-stream") + + actor.processor.process = broken_process + + batches = [ + [_sample(i, i / 24.0), _sample(i + 1, (i + 1) / 24.0)] + for i in range(0, 20, 2) + ] + + def fake_stream(path, *, stats=None, **kw): + stats.frames_advanced = 20 + stats.frames_materialized = 20 + return iter(batches) + + outcome = {} + + def run(): + with patch( + "vidxp.capabilities.visual.iter_frame_batches", + side_effect=fake_stream, + ): + try: + _consume_visual_stream( + source, + participants=[scene, actor], + expected=20, + info=Mock( + fps=24, frame_count=20, duration=0.8, width=2, height=2 + ), + config=config, + storage=storage, + cancellation=cancellation, + progress=None, + timings=timings, + ) + outcome["error"] = None + except ValueError as exc: + outcome["error"] = exc + + worker = threading.Thread(target=run) + worker.start() + worker.join(timeout=5) + self.assertFalse( + worker.is_alive(), + "_consume_visual_stream deadlocked on a failed participant", + ) + self.assertIsInstance(outcome.get("error"), ValueError) From f814e569f5cafa24736adf7ce128fd55e3a0d873 Mon Sep 17 00:00:00 2001 From: Mahnoor-Zaffar <1999mahnoor@gmail.com> Date: Sat, 15 Aug 2026 23:47:25 +0500 Subject: [PATCH 5/6] fix(capabilities): apply parallel visual participant review findings - Fix a critical shutdown deadlock when a participant queue is full: the sentinel drops, so workers now also exit once decoding is done and their queue is empty. - Stop workers promptly after a sibling records an error instead of processing remaining queued batches. - Normalize duplicate modality names in index_visuals before building participant/reporting structures. - Reorder the cancel timer callback so cancellation happens before unblocking the stream. - Prefer addClassCleanup over tearDownClass and drop an unused rebinding. Internal-only. --- src/vidxp/capabilities/visual.py | 10 ++++- tests/test_visual_threading.py | 68 ++++++++++++++++++++++++++++---- 2 files changed, 70 insertions(+), 8 deletions(-) diff --git a/src/vidxp/capabilities/visual.py b/src/vidxp/capabilities/visual.py index 362a1c22..5019d4ca 100644 --- a/src/vidxp/capabilities/visual.py +++ b/src/vidxp/capabilities/visual.py @@ -156,6 +156,7 @@ def _consume_visual_stream( errors: list[Exception] = [] errors_lock = threading.Lock() stream_stats = FrameStreamStats() + done = threading.Event() def _record_error(exc: Exception) -> None: with errors_lock: @@ -211,6 +212,7 @@ def _decode_worker() -> None: except Exception as exc: _record_error(exc) finally: + done.set() for q in participant_queues.values(): try: q.put_nowait(None) @@ -227,9 +229,13 @@ def _participant_worker( try: batch = work_queue.get(timeout=0.2) except queue.Empty: + if done.is_set(): + break continue if batch is None: break + if errors: + return participant_samples = [ sample for sample in batch @@ -322,7 +328,9 @@ def index_visuals( raise ValueError("Visual indexing requires a video path.") selected = tuple( - config.enabled_modalities if modalities is None else modalities + dict.fromkeys( + config.enabled_modalities if modalities is None else modalities + ) ) if not selected: raise ValueError("At least one visual capability must be selected.") diff --git a/tests/test_visual_threading.py b/tests/test_visual_threading.py index 937baaf2..d0a2d998 100644 --- a/tests/test_visual_threading.py +++ b/tests/test_visual_threading.py @@ -1,6 +1,7 @@ from __future__ import annotations import threading +import time import unittest from unittest.mock import Mock, patch @@ -9,7 +10,7 @@ _consume_visual_stream, ) from vidxp.core.contracts import CancellationToken, IndexConfig, VideoSource -from vidxp.core.video import FrameSample, FrameSampling, FrameStreamStats +from vidxp.core.video import FrameSample, FrameSampling def _mock_participant( @@ -42,10 +43,7 @@ def setUpClass(cls): side_effect=lambda samples: samples, ) cls._rgb_patch.start() - - @classmethod - def tearDownClass(cls): - cls._rgb_patch.stop() + cls.addClassCleanup(cls._rgb_patch.stop) def test_single_participant_processes_all_samples(self): scene = _mock_participant("scene") @@ -154,7 +152,9 @@ def blocking_batch(path, *, stats=None, cancellation=None, **kw): ): from vidxp.core.contracts import IndexCancelledError - cancel_timer = threading.Timer(0.1, lambda: (proceed.set(), cancellation.cancel())) + cancel_timer = threading.Timer( + 0.1, lambda: (cancellation.cancel(), proceed.set()) + ) cancel_timer.start() with self.assertRaises(IndexCancelledError): _consume_visual_stream( @@ -179,7 +179,6 @@ def test_error_in_decode_propagates(self): timings = {"frame_stream": 0.0, "scene": 0.3} def broken_stream(path, *, stats=None, **kw): - stats = stats or FrameStreamStats() yield [_sample(0, 0.0)] raise RuntimeError("decode failure") @@ -292,3 +291,58 @@ def run(): "_consume_visual_stream deadlocked on a failed participant", ) self.assertIsInstance(outcome.get("error"), ValueError) + + def test_slow_participant_at_shutdown_does_not_deadlock(self): + scene = _mock_participant("scene") + actor = _mock_participant("actor") + source = VideoSource(video_id="v1", path="unused.mp4") + config = IndexConfig(video_id="v1", enabled_modalities=("scene", "actor")) + storage = Mock() + cancellation = CancellationToken() + timings = {"frame_stream": 0.0, "scene": 0.3, "actor": 0.2} + + def slow_process(samples, **kw): + time.sleep(0.01) + + scene.processor.process = slow_process + + batches = [[_sample(i, i / 24.0)] for i in range(24)] + + def fake_stream(path, *, stats=None, **kw): + stats.frames_advanced = 24 + stats.frames_materialized = 24 + return iter(batches) + + outcome = {} + + def run(): + with patch( + "vidxp.capabilities.visual.iter_frame_batches", + side_effect=fake_stream, + ): + try: + _consume_visual_stream( + source, + participants=[scene, actor], + expected=24, + info=Mock( + fps=24, frame_count=24, duration=1.0, width=2, height=2 + ), + config=config, + storage=storage, + cancellation=cancellation, + progress=None, + timings=timings, + ) + outcome["error"] = None + except Exception as exc: + outcome["error"] = exc + + worker = threading.Thread(target=run) + worker.start() + worker.join(timeout=5) + self.assertFalse( + worker.is_alive(), + "_consume_visual_stream deadlocked at shutdown with a full queue", + ) + self.assertIsNone(outcome.get("error")) From 43d45410e9deff2c162f953ba2305962c0ccc62e Mon Sep 17 00:00:00 2001 From: Mahnoor-Zaffar <1999mahnoor@gmail.com> Date: Sat, 5 Sep 2026 19:34:03 +0500 Subject: [PATCH 6/6] fix(capabilities): carry workflow context into visual worker threads The decode and participant worker threads invoke the progress and cancellation callbacks, which in production rely on DBOS state held in ContextVars that raw threading.Threads do not inherit. Run each worker inside its own copy of the owning thread's context so progress events and cancellation checks work from the workers, and cover it with a production-style callback test. Internal-only. --- src/vidxp/capabilities/visual.py | 22 +++++++++-- tests/test_visual_threading.py | 68 ++++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 4 deletions(-) diff --git a/src/vidxp/capabilities/visual.py b/src/vidxp/capabilities/visual.py index 5019d4ca..bf45b186 100644 --- a/src/vidxp/capabilities/visual.py +++ b/src/vidxp/capabilities/visual.py @@ -1,10 +1,11 @@ from __future__ import annotations +import contextvars import queue import threading from dataclasses import dataclass from time import perf_counter -from typing import Any, Protocol, Sequence +from typing import Any, Callable, Protocol, Sequence from vidxp.capabilities.contracts import CapabilityIndexResult from vidxp.capabilities.registry import CapabilityRegistry @@ -258,13 +259,26 @@ def _participant_worker( except Exception as exc: _record_error(exc) + def _in_context( + target: Callable[..., None], + *args: Any, + ) -> Callable[[], None]: + thread_context = contextvars.copy_context() + + def runner() -> None: + thread_context.run(target, *args) + + return runner + decode_thread = threading.Thread( - target=_decode_worker, name="vidxp-visual-decode" + target=_in_context(_decode_worker), + name="vidxp-visual-decode", ) participant_threads = [ threading.Thread( - target=_participant_worker, - args=(p, participant_queues[p.name]), + target=_in_context( + _participant_worker, p, participant_queues[p.name] + ), name=f"vidxp-visual-{p.name}", ) for p in participants diff --git a/tests/test_visual_threading.py b/tests/test_visual_threading.py index d0a2d998..f563545a 100644 --- a/tests/test_visual_threading.py +++ b/tests/test_visual_threading.py @@ -3,6 +3,7 @@ import threading import time import unittest +from contextvars import ContextVar from unittest.mock import Mock, patch from vidxp.capabilities.visual import ( @@ -292,6 +293,73 @@ def run(): ) self.assertIsInstance(outcome.get("error"), ValueError) + def test_worker_callbacks_inherit_owning_thread_context(self): + workflow_id: ContextVar[str | None] = ContextVar( + "vidxp_test_workflow_id", default=None + ) + + class ContextAwareCancellationEvent: + def __init__(self) -> None: + self._cancelled = False + + def is_set(self) -> bool: + if workflow_id.get() is None: + raise RuntimeError( + "DBOS workflow context missing in cancellation" + ) + return self._cancelled + + def set(self) -> None: + self._cancelled = True + + scene = _mock_participant("scene") + actor = _mock_participant("actor") + source = VideoSource(video_id="v1", path="unused.mp4") + config = IndexConfig(video_id="v1", enabled_modalities=("scene", "actor")) + storage = Mock() + cancellation = CancellationToken(ContextAwareCancellationEvent()) + timings = {"frame_stream": 0.0, "scene": 0.3, "actor": 0.2} + batches = [ + [_sample(0, 0.0), _sample(1, 0.04)], + [_sample(2, 0.08)], + ] + progress_events = [] + + def fake_stream(path, *, stats=None, **kw): + stats.frames_advanced = 3 + stats.frames_materialized = 3 + return iter(batches) + + def publish(event): + if workflow_id.get() is None: + raise RuntimeError("DBOS workflow context missing in progress") + progress_events.append(event) + + token = workflow_id.set("test-workflow") + try: + with patch( + "vidxp.capabilities.visual.iter_frame_batches", + side_effect=fake_stream, + ): + _consume_visual_stream( + source, + participants=[scene, actor], + expected=3, + info=Mock( + fps=24, frame_count=3, duration=0.12, width=2, height=2 + ), + config=config, + storage=storage, + cancellation=cancellation, + progress=publish, + timings=timings, + ) + finally: + workflow_id.reset(token) + + self.assertTrue(progress_events) + self.assertIsNone(workflow_id.get()) + def test_slow_participant_at_shutdown_does_not_deadlock(self): scene = _mock_participant("scene") actor = _mock_participant("actor")