[ci] Cherrypick perf regression skills to component time commit

[ci] Add performance baseline reseed skill and failed-run artifacts

.buildkite/performance-benchmarks/tests/wan-t2v-1.3b.json

@@ -36,17 +36,11 @@
   "thresholds": {
     "L40S": {
       "max_generation_time_s": 34.0,
-      "max_peak_memory_mb": 11000.0,
-      "max_text_encoder_time_s": 5.0,
-      "max_dit_time_s": 10.0,
-      "max_vae_decode_time_s": 10.0
+      "max_peak_memory_mb": 11000.0
     },
     "default": {
       "max_generation_time_s": 120.0,
-      "max_peak_memory_mb": 30000.0,
-      "max_text_encoder_time_s": 5.0,
-      "max_dit_time_s": 10.0,
-      "max_vae_decode_time_s": 10.0
+      "max_peak_memory_mb": 30000.0
     }
   }
 }

fastvideo/tests/performance/compare_baseline.py

@@ -38,21 +38,6 @@ TRACKING_ROOT = os.environ.get(
 )
 PERF_REPORTS_DIR = os.environ.get("PERF_REPORTS_DIR", "/root/data/perf_reports")
 MAX_REGRESSION = float(os.environ.get("PERF_MAX_REGRESSION", "0.05"))
-METRICS = (
-    ("latency", "Latency", 3),
-    ("throughput", "Throughput", 3),
-    ("memory", "Memory", 1),
-    ("text_encoder_time_s", "Text Enc", 3),
-    ("dit_time_s", "DiT", 3),
-    ("vae_decode_time_s", "VAE Decode", 3),
-)
-LOWER_IS_BETTER_METRICS = {
-    "latency",
-    "memory",
-    "text_encoder_time_s",
-    "dit_time_s",
-    "vae_decode_time_s",
-}
 
 
 def _should_persist_tracking() -> bool:
@@ -60,6 +45,7 @@ def _should_persist_tracking() -> bool:
     branch = os.environ.get("BUILDKITE_BRANCH", "")
     return test_scope == "full" and branch == "main"
 
+
 def _load_current_results() -> list[dict[str, Any]]:
     pattern = os.path.join(RESULTS_DIR, "perf_*.json")
     records: list[dict[str, Any]] = []
@@ -88,9 +74,6 @@ def normalize_performance_result(result: dict[str, Any]) -> dict[str, Any]:
     latency = safe_float(result.get("avg_generation_time_s"))
     throughput = safe_float(result.get("throughput_fps"))
     memory = safe_float(result.get("max_peak_memory_mb"))
-    text_encoder_time = safe_float(result.get("text_encoder_time_s"))
-    dit_time = safe_float(result.get("dit_time_s"))
-    vae_decode_time = safe_float(result.get("vae_decode_time_s"))
 
     return {
         "model_id": model_id,
@@ -100,9 +83,6 @@ def normalize_performance_result(result: dict[str, Any]) -> dict[str, Any]:
         "latency": latency,
         "throughput": throughput,
         "memory": memory,
-        "text_encoder_time_s": text_encoder_time,
-        "dit_time_s": dit_time,
-        "vae_decode_time_s": vae_decode_time,
         "success": True,
     }
 
@@ -158,8 +138,7 @@ def _check_regressions(
 ) -> list[str]:
     failures: list[str] = []
 
-    for metric in ("latency", "memory", "text_encoder_time_s", "dit_time_s",
-                   "vae_decode_time_s"):
+    for metric in ("latency", "memory"):
         baseline = _baseline_metric(baseline_records, metric)
         curr = safe_float(current.get(metric))
         if baseline is None or curr is None or baseline <= 0:
@@ -192,7 +171,7 @@ def _metric_delta_percent(
     if curr is None or baseline is None or baseline <= 0:
         return None
 
-    if metric in LOWER_IS_BETTER_METRICS:
+    if metric in ("latency", "memory"):
         return (curr - baseline) / baseline * 100.0
     if metric == "throughput":
         return (baseline - curr) / baseline * 100.0
@@ -212,27 +191,28 @@ def _build_summary_row(
 ) -> dict[str, Any]:
     """Format a single benchmark result as a row for the Markdown table."""
 
-    metric_values: dict[str, dict[str, float | None]] = {}
-    regressions: list[float] = []
-    for metric, _label, _precision in METRICS:
-        curr = safe_float(record.get(metric))
-        baseline = _baseline_metric(baseline_records, metric)
-        regression = _metric_delta_percent(metric, record, baseline_records)
-        metric_values[metric] = {
-            "curr": curr,
-            "base": baseline,
-            "regression_pct": regression,
-        }
-        if regression is not None:
-            regressions.append(regression)
+    latency_base = _baseline_metric(baseline_records, "latency")
+    throughput_base = _baseline_metric(baseline_records, "throughput")
+    memory_base = _baseline_metric(baseline_records, "memory")
 
+    # Calculate percentages for the 'Worst Regression' column
+    latency_reg = _metric_delta_percent("latency", record, baseline_records)
+    throughput_reg = _metric_delta_percent("throughput", record, baseline_records)
+    memory_reg = _metric_delta_percent("memory", record, baseline_records)
+
+    regressions = [v for v in (latency_reg, throughput_reg, memory_reg) if v is not None]
     worst_regression_pct = max(regressions) if regressions else None
 
     return {
         "model_id": record["model_id"],
         "gpu_type": record["gpu_type"],
         "baseline_n": len(baseline_records),
-        "metrics": metric_values,
+        "latency_curr": safe_float(record.get("latency")),
+        "latency_base": latency_base,
+        "throughput_curr": safe_float(record.get("throughput")),
+        "throughput_base": throughput_base,
+        "memory_curr": safe_float(record.get("memory")),
+        "memory_base": memory_base,
         "worst_regression_pct": worst_regression_pct,
         "failed": has_failed,
     }
@@ -249,24 +229,24 @@ def _build_markdown_summary(
         "",
         ("| Model | GPU | Baseline N | Latency (curr/base) | "
          "Throughput (curr/base) | Memory (curr/base) | "
-         "Text Enc (curr/base) | DiT (curr/base) | "
-         "VAE Decode (curr/base) | Worst Regression | Status |"),
-        "|---|---|---:|---|---|---|---|---|---|---:|---|",
+         "Worst Regression | Status |"),
+        "|---|---|---:|---|---|---|---:|---|",
     ]
 
     for row in summary_rows:
-        metric_cells = []
-        for metric, _label, precision in METRICS:
-            values = row["metrics"][metric]
-            metric_cells.append(f"{_compact_value(values['curr'], precision)} / "
-                                f"{_compact_value(values['base'], precision)}")
+        latency = (f"{_compact_value(row['latency_curr'])} / "
+                   f"{_compact_value(row['latency_base'])}")
+        throughput = (f"{_compact_value(row['throughput_curr'])} / "
+                      f"{_compact_value(row['throughput_base'])}")
+        memory = (f"{_compact_value(row['memory_curr'], 1)} / "
+                  f"{_compact_value(row['memory_base'], 1)}")
 
         worst_reg = ("n/a" if row["worst_regression_pct"] is None else f"{row['worst_regression_pct']:.1f}%")
         status = "FAIL" if row["failed"] else "PASS"
 
         lines.append(f"| {row['model_id']} | {row['gpu_type']} | "
                      f"{row['baseline_n']} | "
-                     f"{' | '.join(metric_cells)} | "
+                     f"{latency} | {throughput} | {memory} | "
                      f"{worst_reg} | {status} |")
 
     return "\n".join(lines) + "\n"

fastvideo/tests/performance/dashboard.py

@@ -1,6 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
 import os
-from html import escape
 from datetime import datetime
 
 import plotly.express as px
@@ -8,15 +7,6 @@ import pandas as pd
 
 from hf_store import sync_from_hf, load_as_dataframe
 
-METRICS = (
-    "latency",
-    "throughput",
-    "memory",
-    "text_encoder_time_s",
-    "dit_time_s",
-    "vae_decode_time_s",
-)
-
 # -----------------------------
 # 1. Grouping
 # -----------------------------
@@ -29,36 +19,15 @@ def group_data(df: pd.DataFrame):
 # -----------------------------
 # 2. Plot builder
 # -----------------------------
-def build_plots(df: pd.DataFrame) -> tuple[list, list[dict[str, object]]]:
+def build_plots(df: pd.DataFrame) -> list:
     figs = []
-    skipped_metrics: list[dict[str, object]] = []
 
     for (model_id, gpu_type), g in group_data(df):
         g = g.sort_values("timestamp")
 
         # One chart per metric so the y-axes aren't on wildly different scales
-        for metric in METRICS:
-            if metric not in g.columns:
-                skipped_metrics.append({
-                    "model_id": model_id,
-                    "gpu_type": gpu_type,
-                    "metric": metric,
-                    "reason": "column missing from loaded records",
-                    "records": len(g),
-                    "non_null": 0,
-                })
-                continue
-
-            non_null = int(g[metric].notna().sum())
-            if non_null == 0:
-                skipped_metrics.append({
-                    "model_id": model_id,
-                    "gpu_type": gpu_type,
-                    "metric": metric,
-                    "reason": "no non-null values in loaded records",
-                    "records": len(g),
-                    "non_null": non_null,
-                })
+        for metric in ("latency", "throughput", "memory"):
+            if g[metric].isna().all():
                 continue
 
             fig = px.line(
@@ -72,57 +41,20 @@ def build_plots(df: pd.DataFrame) -> tuple[list, list[dict[str, object]]]:
             )
             figs.append(fig)
 
-    return figs, skipped_metrics
-
-
-def render_skipped_metrics(skipped_metrics: list[dict[str, object]]) -> str:
-    if not skipped_metrics:
-        return ""
-
-    rows = [
-        "<h3>Skipped Metric Plots</h3>",
-        "<table>",
-        ("<thead><tr><th>Model</th><th>GPU</th><th>Metric</th>"
-         "<th>Records</th><th>Non-null</th><th>Reason</th></tr></thead>"),
-        "<tbody>",
-    ]
-    for item in skipped_metrics:
-        rows.append(
-            "<tr>"
-            f"<td>{escape(str(item['model_id']))}</td>"
-            f"<td>{escape(str(item['gpu_type']))}</td>"
-            f"<td>{escape(str(item['metric']))}</td>"
-            f"<td>{item['records']}</td>"
-            f"<td>{item['non_null']}</td>"
-            f"<td>{escape(str(item['reason']))}</td>"
-            "</tr>"
-        )
-    rows.extend(["</tbody>", "</table>"])
-    return "\n".join(rows)
+    return figs
 
 # -----------------------------
 # 3. Render HTML dashboard
 # -----------------------------
-def render_html(figs: list, skipped_metrics: list[dict[str, object]],
-                days: int) -> str:
+def render_html(figs: list, days: int) -> str:
     html_parts = [
         "<html>",
         "<head><meta charset='utf-8'>",
-        ("<style>"
-         "body { font-family: sans-serif; margin: 2rem; }"
-         "table { border-collapse: collapse; margin: 1rem 0 2rem; }"
-         "th, td { border: 1px solid #ddd; padding: 0.4rem 0.6rem; "
-         "text-align: left; }"
-         "th { background: #f5f5f5; }"
-         "</style>"),
+        "<style>body { font-family: sans-serif; margin: 2rem; }</style>",
         "</head><body>",
         f"<h2>Performance Dashboard (last {days} days)</h2>",
     ]
 
-    skipped_html = render_skipped_metrics(skipped_metrics)
-    if skipped_html:
-        html_parts.append(skipped_html)
-
     for fig in figs:
         html_parts.append(fig.to_html(full_html=False, include_plotlyjs="cdn"))
 
@@ -135,7 +67,7 @@ def render_html(figs: list, skipped_metrics: list[dict[str, object]],
 def main() -> None:
     days = int(os.environ.get("DASHBOARD_DAYS", "30"))
 
-    local_dir = sync_from_hf("/tmp/perf-tracking", reuse_existing=True)
+    local_dir = sync_from_hf("/tmp/perf-tracking")
     df = load_as_dataframe(local_dir, days=days)
 
     if df.empty:
@@ -147,15 +79,8 @@ def main() -> None:
           f"{df['gpu_type'].nunique()} GPU type(s), "
           f"date range: {df['timestamp'].min()} → {df['timestamp'].max()}")
 
-    figs, skipped_metrics = build_plots(df)
-    if skipped_metrics:
-        print("Skipped metric plots:")
-        for item in skipped_metrics:
-            print(f"  - {item['model_id']} | {item['gpu_type']} | "
-                  f"{item['metric']}: {item['reason']} "
-                  f"({item['non_null']}/{item['records']} non-null)")
-    print(f"Generated {len(figs)} metric plot(s)")
-    html = render_html(figs, skipped_metrics, days)
+    figs = build_plots(df)
+    html = render_html(figs, days)
 
     commit_sha = os.environ.get("BUILDKITE_COMMIT", "unknown")[:7]
     timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -172,4 +97,4 @@ def main() -> None:
     print(f"Dashboard generated: {output_file}")
 
 if __name__ == "__main__":
-    main()
+    main()

fastvideo/tests/performance/hf_store.py

@@ -25,7 +25,6 @@ from huggingface_hub import HfApi, snapshot_download
 
 HF_REPO_ID: str = os.environ.get("HF_REPO_ID", "FastVideo/performance-tracking")
 HF_TOKEN: str | None = os.environ.get("HF_API_KEY")
-SYNC_MARKER = ".hf_sync_complete"
 
 # ---------------------------------------------------------------------------
 # Low-level helpers
@@ -52,16 +51,7 @@ def safe_float(value: Any) -> float | None:
 # ---------------------------------------------------------------------------
 
 
-def _sync_marker_path(local_dir: str) -> str:
-    return os.path.join(local_dir, SYNC_MARKER)
-
-
-def sync_from_hf(
-    local_dir: str,
-    *,
-    strict: bool = False,
-    reuse_existing: bool = False,
-) -> str:
+def sync_from_hf(local_dir: str, *, strict: bool = False) -> str:
     """Download the HF dataset repo snapshot to *local_dir*.
 
     Returns *local_dir* so callers can chain: ``load_records(sync_from_hf(...))``.
@@ -71,19 +61,7 @@ def sync_from_hf(
     unavailable. Callers that depend on the sync for correctness (e.g. the
     main-branch baseline writer) must pass ``strict=True`` so that misconfig
     or transient HF errors fail loud rather than silently reset the baseline.
-
-    When ``reuse_existing=True``, a previous successful sync in ``local_dir``
-    is reused. This avoids duplicate HF snapshot checks when compare and
-    dashboard scripts run sequentially in the same CI job.
     """
-    marker_path = _sync_marker_path(local_dir)
-    if reuse_existing and os.path.exists(marker_path):
-        print(f"hf_store: reusing existing sync at {local_dir}")
-        return local_dir
-
-    if not reuse_existing and os.path.exists(marker_path):
-        os.remove(marker_path)
-
     if not HF_REPO_ID:
         msg = "hf_store: HF_REPO_ID not set"
         if strict:
@@ -100,12 +78,6 @@ def sync_from_hf(
             token=HF_TOKEN,
             allow_patterns="*.json",
         )
-        os.makedirs(local_dir, exist_ok=True)
-        with open(marker_path, "w", encoding="utf-8") as marker:
-            json.dump({
-                "repo_id": HF_REPO_ID,
-                "synced_at": datetime.now(timezone.utc).isoformat(),
-            }, marker)
     except Exception as exc:
         if strict:
             raise
@@ -251,18 +223,14 @@ def load_records_for_model(
 # DataFrame helpers (dashboard / analytics consumers)
 # ---------------------------------------------------------------------------
 
-_NUMERIC_COLS = (
-    "latency", "throughput", "memory",
-    "text_encoder_time_s", "dit_time_s", "vae_decode_time_s",
-)
+_NUMERIC_COLS = ("latency", "throughput", "memory")
 
 
 def normalize_dataframe(df: pd.DataFrame) -> pd.DataFrame:
     """Apply standard type coercions to a raw records DataFrame.
 
     - Parses ``timestamp`` to UTC-aware datetime.
-    - Coerces ``latency``, ``throughput``, ``memory``, ``text_encoder_time_s``,
-      ``dit_time_s``, ``vae_decode_time_s`` to float.
+    - Coerces ``latency``, ``throughput``, ``memory`` to float.
     - Adds a ``config_id`` column (first 7 chars of ``commit_sha``).
 
     Returns the mutated DataFrame (also modifies in place for efficiency).

fastvideo/tests/performance/test_inference_performance.py

@@ -10,34 +10,17 @@ import glob
 import json
 import os
 import time
-from collections.abc import Mapping
 from datetime import datetime, timezone
 
 import torch
 import pytest
 
-import fastvideo.envs as envs
 from fastvideo import VideoGenerator
 from fastvideo.logger import init_logger
 from fastvideo.worker.multiproc_executor import MultiprocExecutor
 
 logger = init_logger(__name__)
 
-STAGE_METRIC_MAP: dict[str, str] = {
-    "TextEncoderStage": "text_encoder_time_s",
-    "TextEncodingStage": "text_encoder_time_s",
-    "CLIPTextEncoderStage": "text_encoder_time_s",
-    "T5TextEncoderStage": "text_encoder_time_s",
-    "DiTStage": "dit_time_s",
-    "DenoisingStage": "dit_time_s",
-    "DmdDenoisingStage": "dit_time_s",
-    "TransformerStage": "dit_time_s",
-    "VAEDecodeStage": "vae_decode_time_s",
-    "VAEDecoderStage": "vae_decode_time_s",
-    "DecodingStage": "vae_decode_time_s",
-    "DecodeStage": "vae_decode_time_s",
-}
-
 # -- Config discovery -------------------------------------------------------
 
 _BENCHMARKS_DIR = os.path.join(
@@ -90,53 +73,16 @@ def _shutdown_executor(generator):
         generator.executor.shutdown()
 
 
-def _extract_component_times(result: dict) -> dict[str, float | None]:
-    component_times: dict[str, float | None] = {
-        "text_encoder_time_s": None,
-        "dit_time_s": None,
-        "vae_decode_time_s": None,
-    }
-    logging_info = result.get("logging_info")
-    if logging_info is None:
-        return component_times
-    if isinstance(logging_info, Mapping):
-        stages: dict = logging_info.get("stages", {}) or {}
-    else:
-        stages: dict = getattr(logging_info, "stages", {}) or {}
-    if not stages:
-        return component_times
-    logger.info("Discovered pipeline stages: %s", list(stages.keys()))
-    for stage_name, stage_data in stages.items():
-        metric_key = STAGE_METRIC_MAP.get(stage_name)
-        if metric_key is None:
-            logger.debug("Unmapped stage '%s' (%.3fs)",
-                         stage_name,
-                         stage_data.get("execution_time", 0))
-            continue
-        elapsed = stage_data.get("execution_time")
-        if elapsed is None:
-            continue
-        existing = component_times[metric_key]
-        component_times[metric_key] = (
-            elapsed if existing is None else existing + elapsed)
-    return component_times
-
-
-def _avg_component(all_component_times: list[dict], key: str) -> float | None:
-    vals = [r[key] for r in all_component_times if r.get(key) is not None]
-    return round(sum(vals) / len(vals), 3) if vals else None
-
-
 def _run_generation(generator, prompt, generation_kwargs):
-    """Run a single generation, return (elapsed_s, peak_memory_mb, component_times)."""
+    """Run a single generation, return (elapsed_s, peak_memory_mb)."""
     torch.cuda.synchronize()
     start = time.perf_counter()
     result = generator.generate_video(prompt, **generation_kwargs)
     torch.cuda.synchronize()
     elapsed = time.perf_counter() - start
     peak_memory_mb = result.get("peak_memory_mb", 0.0) or 0.0
-    component_times = _extract_component_times(result)
-    return elapsed, peak_memory_mb, component_times
+    return elapsed, peak_memory_mb
+
 
 def _write_results(results):
     """Write JSON results to the results directory."""
@@ -156,7 +102,15 @@ def _write_results(results):
 
 # -- Test -------------------------------------------------------------------
 
-def _run_benchmark(cfg):
+
+@pytest.mark.parametrize(
+    "cfg",
+    _BENCHMARK_CONFIGS,
+    ids=[c["benchmark_id"] for c in _BENCHMARK_CONFIGS],
+)
+def test_inference_performance(cfg):
+    """Measure generation latency and peak GPU memory,
+    assert against device-aware thresholds."""
     run_config = cfg.get("run_config", {})
     required_gpus = run_config.get("required_gpus", 1)
     available = torch.cuda.device_count()
@@ -198,18 +152,12 @@ def _run_benchmark(cfg):
 
         times = []
         peak_memories = []
-        all_component_times = []
         for i in range(num_measure):
             logger.info("Measurement run %d/%d", i + 1, num_measure)
-            elapsed, peak_mb, component_times = _run_generation(
-                generator,
-                prompt,
-                gen_kwargs,
-            )
+            elapsed, peak_mb = _run_generation(generator, prompt, gen_kwargs)
             logger.info("  Time: %.2fs, Peak memory: %.0fMB", elapsed, peak_mb)
             times.append(elapsed)
             peak_memories.append(peak_mb)
-            all_component_times.append(component_times)
     finally:
         _shutdown_executor(generator)
 
@@ -238,11 +186,6 @@ def _run_benchmark(cfg):
         "commit": os.environ.get("BUILDKITE_COMMIT", ""),
         "pr_number": os.environ.get("BUILDKITE_PULL_REQUEST", ""),
         "timestamp": datetime.now(timezone.utc).isoformat(),
-        "text_encoder_time_s": _avg_component(all_component_times,
-                                              "text_encoder_time_s"),
-        "dit_time_s": _avg_component(all_component_times, "dit_time_s"),
-        "vae_decode_time_s": _avg_component(all_component_times,
-                                            "vae_decode_time_s"),
     }
 
     logger.info(
@@ -260,41 +203,3 @@ def _run_benchmark(cfg):
     assert max_peak_memory <= max_mem, (
         f"Peak memory {max_peak_memory:.0f}MB exceeds "
         f"threshold {max_mem:.0f}MB for {device_name}")
-
-    component_thresholds = {
-        "text_encoder_time_s": thresholds.get("max_text_encoder_time_s"),
-        "dit_time_s": thresholds.get("max_dit_time_s"),
-        "vae_decode_time_s": thresholds.get("max_vae_decode_time_s"),
-    }
-    for metric, max_val in component_thresholds.items():
-        if max_val is None:
-            continue
-        actual = results[metric]
-        if actual is not None:
-            assert actual <= max_val, (
-                f"{metric} {actual:.3f}s exceeds threshold {max_val:.3f}s "
-                f"for {device_name}")
-
-
-@pytest.mark.parametrize(
-    "cfg",
-    _BENCHMARK_CONFIGS,
-    ids=[c["benchmark_id"] for c in _BENCHMARK_CONFIGS],
-)
-def test_inference_performance(cfg):
-    """Measure generation latency, peak GPU memory, and component-level timings
-    (text encoder, DiT, VAE decode). Assert each against device-aware thresholds.
-    """
-
-    original_env = os.environ.get("FASTVIDEO_STAGE_LOGGING")
-    original_getter = envs.environment_variables["FASTVIDEO_STAGE_LOGGING"]
-    os.environ["FASTVIDEO_STAGE_LOGGING"] = "1"
-    envs.environment_variables["FASTVIDEO_STAGE_LOGGING"] = lambda: True
-    try:
-        _run_benchmark(cfg)
-    finally:
-        if original_env is None:
-            os.environ.pop("FASTVIDEO_STAGE_LOGGING", None)
-        else:
-            os.environ["FASTVIDEO_STAGE_LOGGING"] = original_env
-        envs.environment_variables["FASTVIDEO_STAGE_LOGGING"] = original_getter