From b65aef2bd658c9a6e372c1eb342ae95af68d5ac0 Mon Sep 17 00:00:00 2001 From: Satyam Srivastava Date: Sat, 2 May 2026 20:43:42 -0700 Subject: [PATCH] [bugfix]: enable perf stage timing in spawned workers [ci] Reuse performance tracking HF sync for dashboard Add a local sync marker after successfully downloading the performance tracking snapshot from Hugging Face. The dashboard now reuses that existing snapshot when it runs after compare_baseline.py in the same CI job, avoiding a second snapshot_download call. If no prior sync marker exists, dashboard still falls back to syncing from HF, so local and standalone dashboard runs keep working. --- .../tests/wan-t2v-1.3b.json | 12 ++++---- .../tests/performance/compare_baseline.py | 8 ++--- fastvideo/tests/performance/dashboard.py | 2 +- fastvideo/tests/performance/hf_store.py | 30 ++++++++++++++++++- .../performance/test_inference_performance.py | 19 ++++++++++-- 5 files changed, 55 insertions(+), 16 deletions(-) diff --git a/.buildkite/performance-benchmarks/tests/wan-t2v-1.3b.json b/.buildkite/performance-benchmarks/tests/wan-t2v-1.3b.json index b676fe5c2..077a7cd53 100644 --- a/.buildkite/performance-benchmarks/tests/wan-t2v-1.3b.json +++ b/.buildkite/performance-benchmarks/tests/wan-t2v-1.3b.json @@ -37,16 +37,16 @@ "L40S": { "max_generation_time_s": 34.0, "max_peak_memory_mb": 11000.0, - "max_text_encoder_time_s": 2.0, - "max_dit_time_s": 25.0, - "max_vae_decode_time_s": 5.0 + "max_text_encoder_time_s": 5.0, + "max_dit_time_s": 10.0, + "max_vae_decode_time_s": 10.0 }, "default": { "max_generation_time_s": 120.0, "max_peak_memory_mb": 30000.0, - "max_text_encoder_time_s": 2.0, - "max_dit_time_s": 25.0, - "max_vae_decode_time_s": 5.0 + "max_text_encoder_time_s": 5.0, + "max_dit_time_s": 10.0, + "max_vae_decode_time_s": 10.0 } } } diff --git a/fastvideo/tests/performance/compare_baseline.py b/fastvideo/tests/performance/compare_baseline.py index 17be2b805..4897b0964 100644 --- a/fastvideo/tests/performance/compare_baseline.py +++ b/fastvideo/tests/performance/compare_baseline.py @@ -55,11 +55,9 @@ LOWER_IS_BETTER_METRICS = { def _should_persist_tracking() -> bool: - # test_scope = os.environ.get("TEST_SCOPE", "") - # branch = os.environ.get("BUILDKITE_BRANCH", "") - # return test_scope == "full" and branch == "main" - return True - + test_scope = os.environ.get("TEST_SCOPE", "") + 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") diff --git a/fastvideo/tests/performance/dashboard.py b/fastvideo/tests/performance/dashboard.py index 53254892c..4a305da50 100644 --- a/fastvideo/tests/performance/dashboard.py +++ b/fastvideo/tests/performance/dashboard.py @@ -135,7 +135,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") + local_dir = sync_from_hf("/tmp/perf-tracking", reuse_existing=True) df = load_as_dataframe(local_dir, days=days) if df.empty: diff --git a/fastvideo/tests/performance/hf_store.py b/fastvideo/tests/performance/hf_store.py index 788022bde..a21bbdef5 100644 --- a/fastvideo/tests/performance/hf_store.py +++ b/fastvideo/tests/performance/hf_store.py @@ -25,6 +25,7 @@ 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 @@ -51,7 +52,16 @@ def safe_float(value: Any) -> float | None: # --------------------------------------------------------------------------- -def sync_from_hf(local_dir: str, *, strict: bool = False) -> str: +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: """Download the HF dataset repo snapshot to *local_dir*. Returns *local_dir* so callers can chain: ``load_records(sync_from_hf(...))``. @@ -61,7 +71,19 @@ def sync_from_hf(local_dir: str, *, strict: bool = False) -> str: 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: @@ -78,6 +100,12 @@ def sync_from_hf(local_dir: str, *, strict: bool = False) -> str: 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 diff --git a/fastvideo/tests/performance/test_inference_performance.py b/fastvideo/tests/performance/test_inference_performance.py index a38f7ea36..1b269d428 100644 --- a/fastvideo/tests/performance/test_inference_performance.py +++ b/fastvideo/tests/performance/test_inference_performance.py @@ -10,6 +10,7 @@ import glob import json import os import time +from collections.abc import Mapping from datetime import datetime, timezone import torch @@ -24,13 +25,16 @@ 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", } @@ -95,7 +99,10 @@ def _extract_component_times(result: dict) -> dict[str, float | None]: logging_info = result.get("logging_info") if logging_info is None: return component_times - stages: dict = getattr(logging_info, "stages", {}) or {} + 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())) @@ -279,9 +286,15 @@ def test_inference_performance(cfg): (text encoder, DiT, VAE decode). Assert each against device-aware thresholds. """ - original = envs.environment_variables["FASTVIDEO_STAGE_LOGGING"] + 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: - envs.environment_variables["FASTVIDEO_STAGE_LOGGING"] = original + 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