Release 0.11.1: benchmark throughput and pipeline improvements

.github/workflows/ci.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches: [dev, main]
   pull_request:
-    branches: [dev, main]
 
 jobs:
   lint:
@@ -79,6 +78,8 @@ jobs:
                                  "--index-url", "https://download.pytorch.org/whl/cpu"])
           EOF
       - run: pip install '.[gpu]' pytest
+      - name: Install audio decoder for the worker integration test
+        run: sudo apt-get update && sudo apt-get install -y ffmpeg
       # Run all offline tests; the slow network smoke checks remain opt-in.
       - run: pytest
 

docs/Classification-and-Scoring.md

@@ -46,8 +46,10 @@ Binary scoring uses `p_not_real = 1 - p_real` and predicts non-real when that
 probability exceeds `0.5`. For example, probabilities `[0.4, 0.3, 0.3]` predict
 `real` in the three-class task but `not real` in the binary task.
 
-Metrics use successful predictions; skipped samples and inference errors are
-excluded. The fields above describe the base pass unless explicitly blended.
+Metrics use successful predictions. Unreadable source samples are recorded as
+skips; inference failures invalidate the run. Every selected sample must have a
+recorded outcome before a run can complete. The fields above describe the base
+pass unless explicitly blended.
 
 ## How SN34 is calculated
 
@@ -74,7 +76,7 @@ pass_score      = sqrt(max(1e-12, mcc_component * brier_component))
 
 A perfect predictor scores approximately `1`. Brier error at or above its
 baseline gives the numerical floor, `0.000001`. A run with no successful base
-predictions returns `sn34_score = 0`.
+predictions fails instead of publishing a score.
 
 `binary_cross_entropy` (log loss) and `per_class_recall` are additional
 diagnostics; they do not enter this formula.

docs/Running-Benchmarks.md

@@ -29,18 +29,52 @@ Filtering keeps each selected dataset's sample limit from the unfiltered run.
 For class definitions and dataset weighting, see [classification and
 scoring](Classification-and-Scoring.md).
 
+## Video sampling
+
+PyAV reads frames sequentially from the beginning, stopping after the requested
+prefix. Frame-rate sampling uses an index stride based on average FPS; short
+clips repeat the last selected frame. No whole-file copy or frame-count scan is
+required. Files with missing frame counts can be evaluated; damage beyond the
+requested prefix is not checked.
+
 ## Store inputs and results
 
 Set `--cache-dir` to a writable directory with space for the datasets; the default
 is `/.cache/gasbench`. Reuse it across runs to reuse downloaded data.
 
+For repeated audio runs, `gasbench preprocess --dataset NAME --cache-dir ./cache`
+converts cached audio to mono 16 kHz, six-second tensors. Conversion can be rerun
+and retains source files in each dataset cache's `originals/` directory.
+
 The CLI writes a JSON report, prediction parquet, and text summary to a run
 subdirectory under `results/`. Set `--results-dir` to change that location and
 `--run-name` to name the subdirectory.
 
 The Python API returns a results dictionary. Call `save_results_to_json()` to
 save it, and pass `records_parquet_path` to `run_benchmark()` to export predictions.
 
+## Diagnose runtime
+
+The JSON report's `performance` field (Python: `results["metrics"]["performance"]`)
+contains stage totals, call counts, and maximum durations for the current attempt.
+Startup, base inference, augmentation, model loading, and finalization are separate
+groups. Per-dataset timing summaries also appear in the logs, including on errors.
+Resumed predictions are counted as restored work; their original timings are not
+added to the new attempt.
+
+Use `input_wait` to identify waits for prepared batches, and compare `checkpoint`
+with its `checkpoint_encode`, `checkpoint_local_write` (write, rename, fsync), and
+`checkpoint_persist` (filesystem commit callback) components. Source reads, decode,
+transforms, and augmentation-cache hits/misses are measured in preparation workers.
+Filesystem writes also report creation, writing, file sync, rename, and directory sync separately.
+Decode includes decoder-internal I/O, including video-file and frame-image reads;
+preprocessed audio loading includes tensor deserialization.
+
+Durations are inclusive and can overlap: **do not sum worker stages into wall
+time**, or add checkpoint components to the checkpoint total. Pass `wall` is the
+sum of dataset execution and final pass checkpoint times. `inference` includes transfers, model execution,
+and output conversion; it is not a GPU-kernel-only measurement.
+
 ## Resume an interrupted run
 
 Give the run an ID and keep its storage available:
@@ -50,8 +84,11 @@ gasbench run --image-model ./my_model --full \
   --cache-dir ./cache --run-id detector-eval
 ```
 
-Rerun the same command to resume. Completed inference batches are restored;
-unfinished work runs again. Use a new run ID for a fresh evaluation.
+Rerun the same command to resume from the last durable checkpoint. Checkpoints group
+inference batches, flushing after 60 seconds or 1,024 pending records, checked between
+batches, and at pass boundaries, completion, or handled failures. Abrupt termination
+replays the uncommitted tail; a long batch can exceed the interval. Use a new run ID
+for a fresh evaluation.
 
 Checkpoints default to `<cache-dir>/runs/<run-id>/checkpoint`. Use
 `--checkpoint-dir` to place them elsewhere. Resume with the same model, benchmark
@@ -64,7 +101,8 @@ For managed workers:
 - Use a trusted filesystem that preserves input file metadata across mounts;
   pending inputs are checked for changes using file size and timestamps.
 - Pass `checkpoint_persist(directory)` to the Python API when the filesystem
-  requires an explicit commit to persist writes remotely.
+  requires an explicit commit to persist writes remotely. It must persist all
+  checkpoint files before returning; a failed commit aborts the writer.
 - Let one worker own a run directory at a time. The caller handles worker restarts.
 
 ## Evaluate robustness

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "gasbench"
-version = "0.11.0"
+version = "0.11.1"
 description = "GASBench - ML model benchmark evaluation package"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -22,6 +22,8 @@ gpu = [
     "numpy==2.4.2",
     "pillow==12.1.1",
     "opencv-python-headless==4.13.0.92",
+    "av==16.0.1",
+    # Retain Decord for existing model.py wrappers that import it.
     "decord==0.6.0; platform_system != 'Darwin'",
     "scikit-learn==1.8.0",
     "torch==2.8.0",

src/gasbench/benchmark.py

@@ -9,9 +9,10 @@
 
 from .logger import get_logger
 from .processing.media import configure_huggingface_cache
-from .benchmarks.image_bench import run_image_benchmark
-from .benchmarks.video_bench import run_video_benchmark
-from .benchmarks.audio_bench import run_audio_benchmark
+from .benchmarks.common import BenchmarkRunConfig
+from .benchmarks.runner import run_modality_benchmark
+from .benchmarks.timing import StageTimings
+from .config import DEFAULT_BATCH_SIZES
 from .benchmarks.utils import create_inference_session
 
 logger = get_logger(__name__)
@@ -97,50 +98,42 @@ async def run_benchmark(
         "gasstation_only": gasstation_only,
     }
 
-    start_time = time.time()
+    start_time = time.perf_counter()
+    model_timings = StageTimings()
 
     try:
         logger.info(
             f"BENCHMARK START: {modality.upper()} model - {model_path}"
         )
 
-        session, input_specs = await load_model_for_benchmark(
-            model_path, modality, benchmark_results
-        )
+        with model_timings.measure("wall"):
+            session, input_specs = await load_model_for_benchmark(
+                model_path, modality, benchmark_results
+            )
 
         if not session:
             return benchmark_results
 
-        # Run benchmark for the specified modality
-        benchmark_score = await execute_benchmark(
-            session,
-            input_specs,
-            modality,
-            benchmark_results,
-            mode,
-            gasstation_only,
-            cache_dir,
-            download_latest_gasstation_data,
-            seed,
-            batch_size,
-            dataset_config,
-            holdout_config,
-            records_parquet_path,
-            run_id,
-            dataset_filters,
-            skip_missing,
-            holdout_weight,
-            holdouts_only,
-            content_category,
-            score_composition,
-            multiclass_scoring,
-            checkpoint_dir=checkpoint_dir,
+        config = BenchmarkRunConfig(
+            modality=modality, mode=mode, gasstation_only=gasstation_only,
+            dataset_config_path=dataset_config, holdout_config_path=holdout_config,
+            cache_dir=cache_dir, hf_token=os.environ.get("HF_TOKEN"),
+            batch_size=DEFAULT_BATCH_SIZES[modality] if batch_size is None else batch_size,
+            augment_level=0, crop_prob=0.0, records_parquet_path=records_parquet_path,
+            run_id=run_id, dataset_filters=dataset_filters,
+            holdout_weight=holdout_weight, holdouts_only=holdouts_only,
+            content_category=content_category, score_composition=score_composition,
+            multiclass_scoring=multiclass_scoring, n_aug_per_dataset=n_aug_per_dataset,
+            aug_weight=aug_weight, aug_cache_dir=aug_cache_dir,
+            aug_cache_readonly=aug_cache_readonly, checkpoint_dir=checkpoint_dir,
             checkpoint_persist=checkpoint_persist,
-            n_aug_per_dataset=n_aug_per_dataset,
-            aug_weight=aug_weight,
-            aug_cache_dir=aug_cache_dir,
-            aug_cache_readonly=aug_cache_readonly,
         )
+        await run_modality_benchmark(
+            session, input_specs, benchmark_results, config=config, seed=seed,
+            skip_missing=skip_missing,
+            download_latest_gasstation_data=download_latest_gasstation_data,
+        )
+        benchmark_score = benchmark_results[f"{modality}_results"]["benchmark_score"]
 
         benchmark_results["benchmark_score"] = benchmark_score
 
@@ -160,12 +153,17 @@ async def run_benchmark(
         elif modality == "audio" and "audio_results" not in benchmark_results:
             benchmark_results["audio_results"] = {"error": str(e)}
 
-        raise e
+        raise
 
     finally:
         benchmark_results["metrics"]["benchmark_duration_seconds"] = (
-            time.time() - start_time
+            time.perf_counter() - start_time
         )
+        performance = benchmark_results["metrics"].setdefault(
+            "performance", {"scope": "current_attempt", "groups": {}}
+        )
+        performance["groups"]["model_load"] = model_timings.snapshot()
+        logger.info("Performance benchmark: %s", json.dumps(performance, sort_keys=True))
 
     return benchmark_results
 
@@ -215,136 +213,6 @@ async def load_model_for_benchmark(
         return None, None
 
 
-async def execute_benchmark(
-    session,
-    input_specs,
-    modality: str,
-    benchmark_results: Dict,
-    mode: str,
-    gasstation_only: bool = False,
-    cache_dir: str = "/.cache/gasbench",
-    download_latest_gasstation_data: bool = False,
-    seed: Optional[int] = None,
-    batch_size: Optional[int] = None,
-    dataset_config: Optional[str] = None,
-    holdout_config: Optional[str] = None,
-    records_parquet_path: Optional[str] = None,
-    run_id: Optional[str] = None,
-    dataset_filters: Optional[list] = None,
-    skip_missing: bool = False,
-    holdout_weight: float = 1.0,
-    holdouts_only: bool = False,
-    content_category: Optional[str] = None,
-    score_composition: Optional[Dict[str, float]] = None,
-    multiclass_scoring: bool = False,
-    n_aug_per_dataset: int = 0,
-    aug_weight: float = 0.2,
-    aug_cache_dir: Optional[str] = None,
-    aug_cache_readonly: bool = False,
-    checkpoint_dir: Optional[str] = None,
-    checkpoint_persist=None,
-) -> float:
-    """Execute the actual benchmark evaluation."""
-
-    logger.info(f"Running {modality} benchmark (mode={mode}, gasstation_only={gasstation_only}, download_latest_gasstation_data={download_latest_gasstation_data}, skip_missing={skip_missing}, holdout_weight={holdout_weight}, holdouts_only={holdouts_only}, content_category={content_category}, score_composition={score_composition}, multiclass_scoring={multiclass_scoring})")
-    if dataset_filters:
-        logger.info(f"Dataset filters: {dataset_filters}")
-    if modality == "image":
-        await run_image_benchmark(
-            session,
-            input_specs,
-            benchmark_results,
-            mode,
-            gasstation_only,
-            cache_dir,
-            download_latest_gasstation_data,
-            seed,
-            batch_size,
-            dataset_config,
-            holdout_config,
-            records_parquet_path=records_parquet_path,
-            run_id=run_id,
-            dataset_filters=dataset_filters,
-            skip_missing=skip_missing,
-            holdout_weight=holdout_weight,
-            holdouts_only=holdouts_only,
-            content_category=content_category,
-            score_composition=score_composition,
-            multiclass_scoring=multiclass_scoring,
-            checkpoint_dir=checkpoint_dir,
-            checkpoint_persist=checkpoint_persist,
-            n_aug_per_dataset=n_aug_per_dataset,
-            aug_weight=aug_weight,
-            aug_cache_dir=aug_cache_dir,
-            aug_cache_readonly=aug_cache_readonly,
-        )
-        benchmark_score = benchmark_results.get("image_results", {}).get("benchmark_score", 0.0)
-    elif modality == "video":
-        await run_video_benchmark(
-            session,
-            input_specs,
-            benchmark_results,
-            mode,
-            gasstation_only,
-            cache_dir,
-            download_latest_gasstation_data,
-            seed,
-            batch_size,
-            dataset_config,
-            holdout_config,
-            records_parquet_path=records_parquet_path,
-            run_id=run_id,
-            dataset_filters=dataset_filters,
-            skip_missing=skip_missing,
-            holdout_weight=holdout_weight,
-            holdouts_only=holdouts_only,
-            content_category=content_category,
-            score_composition=score_composition,
-            multiclass_scoring=multiclass_scoring,
-            checkpoint_dir=checkpoint_dir,
-            checkpoint_persist=checkpoint_persist,
-            n_aug_per_dataset=n_aug_per_dataset,
-            aug_weight=aug_weight,
-            aug_cache_dir=aug_cache_dir,
-            aug_cache_readonly=aug_cache_readonly,
-        )
-        benchmark_score = benchmark_results.get("video_results", {}).get("benchmark_score", 0.0)
-    elif modality == "audio":
-        await run_audio_benchmark(
-            session,
-            input_specs,
-            benchmark_results,
-            mode,
-            gasstation_only,
-            cache_dir,
-            download_latest_gasstation_data,
-            seed,
-            batch_size,
-            dataset_config,
-            holdout_config,
-            records_parquet_path=records_parquet_path,
-            run_id=run_id,
-            dataset_filters=dataset_filters,
-            skip_missing=skip_missing,
-            holdout_weight=holdout_weight,
-            holdouts_only=holdouts_only,
-            content_category=content_category,
-            score_composition=score_composition,
-            multiclass_scoring=multiclass_scoring,
-            checkpoint_dir=checkpoint_dir,
-            checkpoint_persist=checkpoint_persist,
-            n_aug_per_dataset=n_aug_per_dataset,
-            aug_weight=aug_weight,
-            aug_cache_dir=aug_cache_dir,
-            aug_cache_readonly=aug_cache_readonly,
-        )
-        benchmark_score = benchmark_results.get("audio_results", {}).get("benchmark_score", 0.0)
-    else:
-        raise ValueError(f"Invalid modality: {modality}. Must be 'image', 'video' or 'audio'")
-
-    return benchmark_score
-
-
 def format_benchmark_summary(benchmark_results: Dict) -> str:
     """Format a comprehensive summary of benchmark results as a string."""
     lines = []
@@ -507,6 +375,7 @@ def save_results_to_json(
         "overall_score": benchmark_results.get("benchmark_score", 0.0),
         "validation": benchmark_results.get("validation", {}),
         "errors": benchmark_results.get("errors", []),
+        "performance": benchmark_results.get("metrics", {}).get("performance", {}),
     }
 
     # Add modality-specific results
@@ -518,6 +387,7 @@ def save_results_to_json(
 
             output_data["results"] = {
                 "total_samples": results.get("total_samples", 0),
+                "skipped_samples": results.get("skipped_samples", 0),
                 "correct_predictions": results.get("correct_predictions", 0),
                 "accuracy": results.get("benchmark_score", 0.0),
                 "avg_inference_time_ms": results.get("avg_inference_time_ms", 0),