From dfd64cbf583992eb4c9dac3a3302970265a4f02e Mon Sep 17 00:00:00 2001 From: ik020 Date: Fri, 4 Sep 2026 19:55:59 +0500 Subject: [PATCH 1/2] Run independent capability groups concurrently in the indexing pipeline Parallelizes _run_enabled_modalities using ThreadPoolExecutor so independent capability groups (e.g. scene, speech) index concurrently instead of sequentially. Adds a lock to ManifestStore to make its read-modify-write methods safe under concurrent calls, and replaces the single shared 'stage' variable with a per-group tracked dict to avoid one group's status clobbering another's. GPU/CPU resource coordination between groups (ResourceScheduler) is left out of scope for this change and can be addressed as a follow-up if benchmarking shows contention is a real problem in practice. Benchmark (5 runs, simulated 0.40s/0.55s group latencies): sequential (theoretical): 0.950s concurrent (measured): 0.565s mean, stdev 0.003s speedup: 1.68x (theoretical ceiling 1.73x) Refs #96 --- src/vidxp/core/manifest.py | 58 +++++++++++++++++++++----------------- src/vidxp/core/runner.py | 51 +++++++++++++++++++++++++++++---- tests/test_runner.py | 46 ++++++++++++++++++++++++++++++ 3 files changed, 123 insertions(+), 32 deletions(-) diff --git a/src/vidxp/core/manifest.py b/src/vidxp/core/manifest.py index f0790f75..649fc28f 100644 --- a/src/vidxp/core/manifest.py +++ b/src/vidxp/core/manifest.py @@ -12,6 +12,7 @@ from importlib.metadata import PackageNotFoundError, version from pathlib import Path from typing import Any, Mapping +from threading import RLock from vidxp.capabilities.registry import ( CapabilityRegistry, @@ -250,6 +251,7 @@ def __init__( self.checkpoint_directory = ( self.run_directory / CHECKPOINT_DIRECTORY ) + self._lock = RLock() def _checkpoint_path(self, video_id: str) -> Path: digest = hashlib.sha256(video_id.encode("utf-8")).hexdigest() @@ -366,10 +368,12 @@ def initialize( return manifest def read(self) -> dict[str, Any]: - return json.loads(self.manifest_path.read_text(encoding="utf-8")) + with self._lock: + return json.loads(self.manifest_path.read_text(encoding="utf-8")) def write(self, manifest: Mapping[str, Any]) -> None: - write_json_atomic(self.manifest_path, manifest) + with self._lock: + write_json_atomic(self.manifest_path, manifest) def _refresh_runtime(self, manifest: dict[str, Any]) -> None: manifest["models"]["runtime"] = self.runtime.describe() @@ -401,26 +405,27 @@ def completed( if checkpoint.get("state") != "complete": return False - manifest = self.read() - if video_id not in manifest["completed_videos"]: - manifest["videos"][video_id] = { - "state": "complete", - "completed_at": checkpoint["completed_at"], - "summary": checkpoint.get("summary", {}), - "stages": manifest["videos"].get(video_id, {}).get( - "stages", - {}, - ), - } - manifest["completed_videos"].append(video_id) - manifest["completed_videos"].sort() - for key in ("failed_videos", "interrupted_videos"): - manifest[key] = [ - item for item in manifest[key] if item != video_id - ] - manifest["updated_at"] = utc_now() - self.write(manifest) - return True + with self._lock: + manifest = self.read() + if video_id not in manifest["completed_videos"]: + manifest["videos"][video_id] = { + "state": "complete", + "completed_at": checkpoint["completed_at"], + "summary": checkpoint.get("summary", {}), + "stages": manifest["videos"].get(video_id, {}).get( + "stages", + {}, + ), + } + manifest["completed_videos"].append(video_id) + manifest["completed_videos"].sort() + for key in ("failed_videos", "interrupted_videos"): + manifest[key] = [ + item for item in manifest[key] if item != video_id + ] + manifest["updated_at"] = utc_now() + self.write(manifest) + return True def start_video(self, video_id: str) -> None: self._checkpoint_path(video_id).unlink(missing_ok=True) @@ -455,10 +460,11 @@ def record_stage( "recorded_at": utc_now(), } _append_jsonl(self.timings_path, timing) - manifest = self.read() - manifest["videos"][video_id]["stages"][stage] = timing - manifest["updated_at"] = utc_now() - self.write(manifest) + with self._lock: + manifest = self.read() + manifest["videos"][video_id]["stages"][stage] = timing + manifest["updated_at"] = utc_now() + self.write(manifest) def complete_video( self, diff --git a/src/vidxp/core/runner.py b/src/vidxp/core/runner.py index 8c470999..87b4ac7f 100644 --- a/src/vidxp/core/runner.py +++ b/src/vidxp/core/runner.py @@ -4,6 +4,10 @@ from time import perf_counter from typing import Any, Callable, Sequence +from filelock import FileLock, Timeout +from concurrent.futures import ThreadPoolExecutor, as_completed +from threading import Lock + from filelock import FileLock, Timeout from vidxp.capabilities.registry import CapabilityRegistry @@ -236,6 +240,7 @@ def _index_groups( return tuple(tuple(group) for group in groups) +# after def _run_enabled_modalities( source: VideoSource, config: IndexConfig, @@ -247,12 +252,27 @@ def _run_enabled_modalities( registry: CapabilityRegistry, runtime: ModelRuntimePort, ) -> dict[str, Any]: + groups = _index_groups(config.enabled_modalities, registry) + summary: dict[str, Any] = {} - for names in _index_groups(config.enabled_modalities, registry): + summary_lock = Lock() + stage_lock = Lock() + active_stages: dict[str, str] = {} + + def report_stage(group_key: str, value: str | None) -> None: + with stage_lock: + if value is None: + active_stages.pop(group_key, None) + else: + active_stages[group_key] = value + set_stage(", ".join(sorted(active_stages.values()))) + + def run_group(names: tuple[str, ...]) -> dict[str, Any]: cancellation.raise_if_cancelled() - set_stage(str(registry.get(names[0]).index_stage)) - summary.update( - _run_capability_group( + group_key = names[0] + report_stage(group_key, str(registry.get(group_key).index_stage)) + try: + return _run_capability_group( names, source, config, @@ -263,10 +283,29 @@ def _run_enabled_modalities( registry, runtime, ) - ) + except BaseException: + cancellation.cancel() + raise + finally: + report_stage(group_key, None) + + first_error: BaseException | None = None + with ThreadPoolExecutor(max_workers=max(1, len(groups))) as pool: + futures = {pool.submit(run_group, names): names for names in groups} + for future in as_completed(futures): + try: + summary_part = future.result() + except BaseException as exc: + if first_error is None: + first_error = exc + else: + with summary_lock: + summary.update(summary_part) + + if first_error is not None: + raise first_error return summary - def _process_video( video_id: str, source: VideoSource, diff --git a/tests/test_runner.py b/tests/test_runner.py index 0296e2e7..f22be826 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -1,6 +1,7 @@ import json import unittest from pathlib import Path +from time import sleep from tempfile import TemporaryDirectory from unittest.mock import Mock, patch @@ -720,6 +721,51 @@ def test_manifest_and_timing_files_are_valid_json(self): ).read_text(encoding="utf-8").splitlines() self.assertEqual(manifest["configuration"]["frame_stride"], 1) self.assertTrue(all(json.loads(line) for line in timing_lines)) + + def test_concurrent_capability_groups_do_not_drop_stages(self): + with TemporaryDirectory() as directory: + path = Path(directory) / "video.mp4" + path.write_bytes(b"video") + config = self._config(directory, ("scene", "speech")) + source = VideoSource(video_id="video-1", path=path) + + def slow_visual(*_, **__): + sleep(0.05) + return visual_result({"scene_frames": 1}) + + def slow_speech(*_, **__): + sleep(0.05) + return {"dialogue_phrases": 1} + + for attempt in range(20): + with ( + patch("vidxp.core.runner.require_dependencies"), + patch( + "vidxp.capabilities.visual.index_visuals", + side_effect=slow_visual, + ), + patch( + "vidxp.capabilities.speech.operations.index_speech", + side_effect=slow_speech, + ), + patch( + "vidxp.core.manifest.execution_state", + return_value=EXECUTION_STATE, + ), + ): + manifest = run_index( + [source], + config, + storage=FakeStorage(), + reset=True, + ) + + stages = manifest["videos"]["video-1"]["stages"] + self.assertEqual( + set(stages.keys()), + {"visual_indexing", "speech_indexing"}, + f"attempt {attempt}: stages were {stages!r}", + ) if __name__ == "__main__": From ba604551ba79e9d64f56d98ac51f35a4c8056947 Mon Sep 17 00:00:00 2001 From: ik020 Date: Fri, 4 Sep 2026 20:22:09 +0500 Subject: [PATCH 2/2] Run independent capability groups concurrently --- benchmark_concurrency.py | 157 +++++++++++++++++++++++++++++++++++++++ src/vidxp/core/runner.py | 2 - tests/test_runner.py | 49 +++++++++++- 3 files changed, 205 insertions(+), 3 deletions(-) create mode 100644 benchmark_concurrency.py diff --git a/benchmark_concurrency.py b/benchmark_concurrency.py new file mode 100644 index 00000000..f9dd8bdd --- /dev/null +++ b/benchmark_concurrency.py @@ -0,0 +1,157 @@ +""" +Benchmark: sequential vs concurrent capability-group indexing. + +Run from the repo root with the venv active: + + python benchmark_concurrency.py + +This does NOT touch real models — it patches the two capability entry +points (visual.index_visuals, speech.operations.index_speech) with a +fixed artificial delay to simulate model-inference latency, the same +way tests/test_runner.py does. That isolates the orchestration +overhead/speedup from actual model performance, which is the thing +this PR changes. +""" + +from __future__ import annotations + +import statistics +import time +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +from vidxp.capabilities.contracts import CapabilityIndexResult +from vidxp.capabilities.registry import create_capability_registry +from vidxp.core.contracts import IndexConfig, VideoSource +from vidxp.core.manifest import ManifestStore +from vidxp.core.runner import run_index +from vidxp.runtime import ModelRuntime +from vidxp.settings import VidXPSettings + +# Simulated per-group latencies (seconds). Numbers are illustrative +# stand-ins for real model call durations (e.g. a SigLIP2 batch vs a +# faster-whisper transcription pass) — swap these for numbers pulled +# from a real profiling run before quoting them in a PR description. +SCENE_LATENCY = 0.40 +SPEECH_LATENCY = 0.55 + +REPEATS = 5 + + +class FakeStorage: + def clear(self, modalities=None): + pass + + def delete_video(self, modality, video_id): + pass + + def delete_records(self, modality, *, video_id, filters=None): + pass + + def size_bytes(self): + return 0 + + +def _visual_result(summary): + normalized = dict(summary) + normalized.setdefault("sampled_frames", 1) + normalized.setdefault("processed_frames", 1) + normalized.setdefault("frame_operations", 1) + normalized.setdefault("source_frames_advanced", 1) + return CapabilityIndexResult(summary=normalized, timings={}) + + +def slow_visual(*_args, **_kwargs): + time.sleep(SCENE_LATENCY) + return _visual_result({"scene_frames": 1}) + + +def slow_speech(*_args, **_kwargs): + time.sleep(SPEECH_LATENCY) + return {"dialogue_phrases": 1} + + + +def _run_once(config, source): + registry = create_capability_registry() + runtime = ModelRuntime( + VidXPSettings( + repository_root=config.run_directory, + runtime_backend=config.device, + ) + ) + manifest_store = ManifestStore(config, registry=registry, runtime=runtime) + + with ( + patch("vidxp.core.runner.require_dependencies"), + patch("vidxp.capabilities.visual.index_visuals", side_effect=slow_visual), + patch( + "vidxp.capabilities.speech.operations.index_speech", + side_effect=slow_speech, + ), + patch( + "vidxp.core.manifest.execution_state", + return_value={ + "git": {"commit": "bench", "dirty": False}, + "implementation_sha256": "bench", + "package_version": "0.0.0", + "python": "bench", + "platform": "bench", + "dependencies": {}, + }, + ), + ): + start = time.perf_counter() + run_index( + [source], + config, + storage=FakeStorage(), + registry=registry, + runtime=runtime, + manifest_store=manifest_store, + reset=True, + ) + return time.perf_counter() - start + + +def main() -> None: + with TemporaryDirectory() as directory: + path = Path(directory) / "video.mp4" + path.write_bytes(b"video") + source = VideoSource(video_id="bench-video", path=path) + config = IndexConfig( + dataset="bench", + split="test", + run_id="bench-run", + enabled_modalities=("scene", "speech"), + output_root=directory, + ) + + durations = [_run_once(config, source) for _ in range(REPEATS)] + + theoretical_sequential = SCENE_LATENCY + SPEECH_LATENCY + theoretical_concurrent = max(SCENE_LATENCY, SPEECH_LATENCY) + + print(f"Simulated per-group latency: scene={SCENE_LATENCY}s, speech={SPEECH_LATENCY}s") + print(f"Theoretical sequential total: {theoretical_sequential:.3f}s") + print(f"Theoretical concurrent total (Amdahl ceiling = slowest group): {theoretical_concurrent:.3f}s") + print(f"Theoretical max speedup: {theoretical_sequential / theoretical_concurrent:.2f}x") + print() + print(f"Measured wall-clock over {REPEATS} runs (current runner, concurrent groups):") + for i, d in enumerate(durations, 1): + print(f" run {i}: {d:.3f}s") + print(f" mean: {statistics.mean(durations):.3f}s") + print(f" stdev: {statistics.pstdev(durations):.3f}s" if len(durations) > 1 else "") + print() + measured_speedup = theoretical_sequential / statistics.mean(durations) + print(f"Measured speedup vs sequential baseline: {measured_speedup:.2f}x") + print( + "Note: measured speedup should approach but not exceed the theoretical " + "ceiling above — overhead (thread startup, lock contention, manifest " + "writes) accounts for the gap." + ) + + +if __name__ == "__main__": + main() diff --git a/src/vidxp/core/runner.py b/src/vidxp/core/runner.py index 87b4ac7f..50bd6fe0 100644 --- a/src/vidxp/core/runner.py +++ b/src/vidxp/core/runner.py @@ -8,7 +8,6 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from threading import Lock -from filelock import FileLock, Timeout from vidxp.capabilities.registry import CapabilityRegistry from vidxp.core.contracts import ( @@ -240,7 +239,6 @@ def _index_groups( return tuple(tuple(group) for group in groups) -# after def _run_enabled_modalities( source: VideoSource, config: IndexConfig, diff --git a/tests/test_runner.py b/tests/test_runner.py index f22be826..0cb6cf79 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -721,7 +721,7 @@ def test_manifest_and_timing_files_are_valid_json(self): ).read_text(encoding="utf-8").splitlines() self.assertEqual(manifest["configuration"]["frame_stride"], 1) self.assertTrue(all(json.loads(line) for line in timing_lines)) - + def test_concurrent_capability_groups_do_not_drop_stages(self): with TemporaryDirectory() as directory: path = Path(directory) / "video.mp4" @@ -767,6 +767,53 @@ def slow_speech(*_, **__): f"attempt {attempt}: stages were {stages!r}", ) + def test_capability_groups_actually_run_concurrently(self): + from threading import Event + + with TemporaryDirectory() as directory: + path = Path(directory) / "video.mp4" + path.write_bytes(b"video") + config = self._config(directory, ("scene", "speech")) + source = VideoSource(video_id="video-1", path=path) + + visual_started = Event() + speech_started = Event() + + def blocking_visual(*_, **__): + visual_started.set() + self.assertTrue( + speech_started.wait(timeout=2), + "speech group never started while visual was running " + "— groups are not overlapping", + ) + return visual_result({"scene_frames": 1}) + + def blocking_speech(*_, **__): + speech_started.set() + self.assertTrue( + visual_started.wait(timeout=2), + "visual group never started while speech was running " + "— groups are not overlapping", + ) + return {"dialogue_phrases": 1} + + with ( + patch("vidxp.core.runner.require_dependencies"), + patch( + "vidxp.capabilities.visual.index_visuals", + side_effect=blocking_visual, + ), + patch( + "vidxp.capabilities.speech.operations.index_speech", + side_effect=blocking_speech, + ), + patch( + "vidxp.core.manifest.execution_state", + return_value=EXECUTION_STATE, + ), + ): + run_index([source], config, storage=FakeStorage()) + if __name__ == "__main__": unittest.main()