Compare commits

..
Author SHA1 Message Date
SolitaryThinker 20581fe9c6 [misc]: double CI lane CPU/memory reservations (8->16 vCPU, 32->64GB)
The cpu=8 floor from #1566 did not stop the packed-host slowdown: two
unrelated PRs hit the exact TE=3.6s signature post-#1566 with healthy
GPU telemetry. A reservation is a floor, not isolation — doubling it
halves how much the scheduler can pack next to the benchmark.
2026-07-07 00:22:08 -07:00
595f0ea60e [feat] Add DreamX-World 5B Cam and AR pipelines (#1538)
Co-authored-by: Suckl <Suckl@users.noreply.github.com>
Co-authored-by: SolitaryThinker <wlsaidhi@gmail.com>
2026-07-07 06:16:18 +00:00
William Lin d921832cd2 [misc]: reserve CPU/memory for timing-sensitive Modal CI lanes (#1566) 2026-07-06 22:13:53 -07:00
William Lin 629697629a [bugfix]: free CUDA memory between train-framework model tests (LongCat OOM on L40S) (#1565) 2026-07-06 21:27:30 -07:00
William Lin a25313beec [ci]: wire fastvideo/tests/ops/ into the unit-test lane (#1559) 2026-07-06 12:07:26 -07:00
William Lin dbde64385b [bugfix]: bump FA4 pin to the CuTe DSL 4.6 compatible rev (#1564) 2026-07-06 12:06:59 -07:00
William Lin 9d909f5f04 [test]: remove dead and duplicate tests (-489 lines) (#1556) 2026-07-05 15:53:40 -07:00
William Lin 76b0550c15 [ci]: run pre-commit on fork PRs without manual approval (#1555) 2026-07-05 14:18:16 -07:00
William Lin 384c1e9493 [misc]: update reseed-performance-baseline skill for the hf_store move (#1545 follow-up) (#1553) 2026-07-05 14:16:55 -07:00
William Lin b1dbcc93f6 [misc]: reformat fastvideo/performance to the repo yapf config (#1554) 2026-07-05 14:16:20 -07:00
William Lin b93833772e [ci]: guard against test directories no CI lane collects (#1552) 2026-07-05 14:07:38 -07:00
Mac Lee 30b523edd6 [ci] Normalize performance stage component metrics (#1475) (#1550) 2026-07-05 14:05:26 -07:00
Mac LeeandSolitaryThinker 6aab7f3832 [ci] cover Hunyuan 1.5 chat-list text preprocessing (#1518)
Co-authored-by: SolitaryThinker <wlsaidhi@gmail.com>
2026-07-05 12:05:25 -07:00
Mac Lee 9cd53fe5f8 [ci] Add metric-specific performance thresholds (#1545) 2026-07-05 12:04:33 -07:00
40 changed files with 849 additions and 720 deletions
@@ -69,7 +69,7 @@ approval, then upload reviewed accepted baseline records.
| `model_id` | Yes | Benchmark id, e.g. `wan-t2v-1.3b-2gpu`. This maps to the HF subdirectory after `sanitize(model_id)`. |
| `gpu_type` | Yes | Exact GPU device string from the performance record, e.g. the L40S device name emitted by CI. Baselines are GPU-specific. |
| `source_results` | Yes | One or more local paths or Buildkite artifact URLs for accepted shifted performance JSONs. Prefer normalized `normalized_perf_*.json` artifacts emitted by `compare_baseline.py`. Accept `source_result` as an alias only for a single JSON. |
| `max_intra_batch_regression` | No | Maximum allowed regression of any source JSON against the source batch median. Default: `PERF_MAX_REGRESSION` if set, otherwise `0.05` (5%). |
| `max_intra_batch_regression` | No | Maximum allowed regression of any source JSON against the source batch median. Default: `0.05` (5%). |
| `intent_rationale` | Yes | One-line explanation for why the baseline shift is legitimate. This is written into provenance and should be reused in the PR. |
Hardcoded defaults:
@@ -148,8 +148,7 @@ For each metric with at least two non-null source values:
4. Stop if any source record regresses against the batch median by more than
`max_intra_batch_regression`.
Default `max_intra_batch_regression` to `PERF_MAX_REGRESSION` when set,
otherwise `0.05`. Print a table with per-source values, batch median, and
Default `max_intra_batch_regression` to `0.05`. Print a table with per-source values, batch median, and
worst intra-batch regression.
This check prevents uploading a mixed batch where one JSON is materially
@@ -183,7 +182,7 @@ present, that run is not a valid source for baseline reseeding.
### 2. Sync and back up existing HF records under /tmp
Use `fastvideo/tests/performance/hf_store.py` helpers directly. Do **not** use
Use `fastvideo/performance/hf_store.py` helpers directly. Do **not** use
`compare_baseline.py` as a sync shortcut; on full main runs it can persist
records, while this step must only fetch and back up existing history.
@@ -192,7 +191,7 @@ The sync command pattern is:
```bash
export PERFORMANCE_TRACKING_ROOT="${PERFORMANCE_TRACKING_ROOT:-/tmp/perf-tracking}"
export HF_REPO_ID="${HF_REPO_ID:-FastVideo/performance-tracking}"
PYTHONPATH=fastvideo/tests/performance python -c 'from hf_store import sync_from_hf; import os; sync_from_hf(os.environ["PERFORMANCE_TRACKING_ROOT"], strict=True)'
python -c 'from fastvideo.performance.hf_store import sync_from_hf; import os; sync_from_hf(os.environ["PERFORMANCE_TRACKING_ROOT"], strict=True)'
```
Then back up only the sanitized model directory under `/tmp`:
@@ -200,8 +199,8 @@ Then back up only the sanitized model directory under `/tmp`:
```bash
SHORT_COMMIT=$(git rev-parse --short=12 HEAD)
TIMESTAMP=$(date -u +%Y%m%d_%H%M%S)
MODEL_SAFE=$(PYTHONPATH=fastvideo/tests/performance python - <<'PY'
from hf_store import sanitize
MODEL_SAFE=$(python - <<'PY'
from fastvideo.performance.hf_store import sanitize
print(sanitize("<model_id>"))
PY
)
@@ -235,7 +234,7 @@ first baseline seed. Continue, but report that baseline history was empty.
Load the last 5 successful records for the target:
```python
from hf_store import load_records_for_model
from fastvideo.performance.hf_store import load_records_for_model
records = load_records_for_model(
"/tmp/perf-tracking",
@@ -372,7 +371,7 @@ prepared records plus backup on disk.
Use the shared storage helper so the path and repo type match CI:
```python
from hf_store import upload_record
from fastvideo.performance.hf_store import upload_record
upload_record("<local_record_path>", record, strict=True)
```
@@ -460,7 +459,7 @@ directories created for this reseed. Never remove unrelated `/tmp` contents.
intentional baseline replacement.
- `fastvideo/tests/performance/compare_baseline.py` — normalization, rolling
median comparison, and persistence rules.
- `fastvideo/tests/performance/hf_store.py` — HF sync, record loading,
- `fastvideo/performance/hf_store.py` — HF sync, record loading,
`sanitize()`, and `upload_record()`.
- `fastvideo/tests/performance/test_inference_performance.py` — source result
JSON schema.
+19 -2
View File
@@ -1,7 +1,11 @@
name: pre-commit
on:
pull_request:
# pull_request_target instead of pull_request: the workflow definition and
# the hook config are always taken from the BASE branch, so fork /
# first-time-contributor PRs run immediately without a maintainer clicking
# "Approve and run". The PR head is checked out as data only.
pull_request_target:
branches: [main]
workflow_call:
inputs:
@@ -15,12 +19,25 @@ permissions:
jobs:
pre-commit:
if: github.event_name == 'workflow_call' || github.event.pull_request.draft != true
if: github.event.pull_request.draft != true
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
ref: ${{ inputs.ref || '' }}
# For PR events, lint the PR head — but keep the hook definitions from
# the base branch so an untrusted PR cannot alter what gets executed.
- name: Save trusted hook config
if: github.event_name == 'pull_request_target'
run: cp .pre-commit-config.yaml "$RUNNER_TEMP/trusted-pre-commit-config.yaml"
- uses: actions/checkout@v4
if: github.event_name == 'pull_request_target'
with:
ref: ${{ github.event.pull_request.head.sha }}
persist-credentials: false
- name: Restore trusted hook config
if: github.event_name == 'pull_request_target'
run: cp "$RUNNER_TEMP/trusted-pre-commit-config.yaml" .pre-commit-config.yaml
- uses: actions/setup-python@v5
with:
python-version: "3.12"
+8 -3
View File
@@ -12,13 +12,17 @@ Defaults:
- `HF_REPO_ID=FastVideo/performance-tracking`
- `PERFORMANCE_TRACKING_ROOT=/tmp/fastvideo-perf-dashboard`
- `PERF_MAX_REGRESSION=0.05`
Records can include source metadata:
Records can include source metadata and rolling-baseline policy context:
- `run_source`: `pr`, `local`, `scheduled_main`, or `unknown`
- `baseline_eligible`: only successful scheduled-main records should be true
- Buildkite metadata such as branch, PR number, build URL, build ID, and job ID
- `regression_thresholds`: per-metric rolling-baseline percent and absolute
floors used for recomputed status context
Dashboard/API metric payloads expose `threshold_exceeded` for raw threshold
crossings; `regressed` remains the gated CI-failure signal.
Set one of `HF_API_KEY`, `HUGGINGFACE_HUB_TOKEN`, or `HF_TOKEN` if the
configured dataset repo requires authenticated access:
@@ -89,7 +93,8 @@ Trend charts show metric-specific axes and exact point details on hover/focus:
- PR number, branch, and Buildkite URL when present
The latest status table uses the stored JSON `success` value. Recomputed
baseline context is shown separately and does not override stored status.
baseline context applies each metric's percent and absolute regression floors
and does not override stored status.
## API
@@ -440,6 +440,8 @@ export default function App() {
<th>Throughput</th>
<th>Memory</th>
<th>Worst</th>
<th>Exceeded</th>
<th>Failing</th>
</tr>
</thead>
<tbody>
@@ -465,6 +467,12 @@ export default function App() {
<td>{formatNumber(row.metrics.throughput?.current, 3)}</td>
<td>{formatNumber(row.metrics.memory?.current, 1)}</td>
<td>{formatNumber(row.worst_regression_pct, 1)}%</td>
<td>
{row.threshold_exceeded_metrics.length
? row.threshold_exceeded_metrics.join(", ")
: "none"}
</td>
<td>{row.failing_metrics.length ? row.failing_metrics.join(", ") : "none"}</td>
</tr>
))}
</tbody>
@@ -2,6 +2,12 @@ export type MetricValue = {
current: number | null;
baseline: number | null;
regression_pct: number | null;
absolute_delta: number | null;
threshold_percent: number;
threshold_absolute: number;
gated: boolean;
threshold_exceeded: boolean;
regressed: boolean;
label: string;
lower_is_better: boolean;
precision: number;
@@ -15,7 +21,8 @@ export type SummaryRow = {
success: boolean;
baseline_n: number;
worst_regression_pct: number | null;
regression_threshold_pct: number;
threshold_exceeded_metrics: string[];
failing_metrics: string[];
computed_regression_status: "pass" | "fail";
status: "pass" | "fail";
run_source: RunSource;
+2 -2
View File
@@ -68,7 +68,7 @@ ARG FLASH_ATTN_WHEEL_RELEASE_ARM64=https://github.com/mjun0812/flash-attention-p
# cutlass-4.4 `cute.core.ThrMma` API, which crashes on the cutlass-dsl 4.5 that
# flashinfer/quack pull in. After the wheel install we overlay this cutlass-4.5-safe
# upstream cute (flash-attn-4) so the image runs FA4 instead of the FA2 fallback.
ARG FA4_CUTE_REF=940cd9680f3315f2f06b43ab5bea2c2cf2d96806
ARG FA4_CUTE_REF=82d6441eec5d4dfec120153db2c0145ae855a083
# Provided automatically by BuildKit/buildx (e.g. "amd64" / "arm64") and used to
# select the prebuilt flash-attn wheel. Empty under a plain `docker build` without
@@ -161,7 +161,7 @@ RUN --mount=type=cache,target=/opt/uv/cache \
uv pip install flash-attn==${FLASH_ATTN_VERSION} --no-build-isolation; \
fi
# Overlay the cutlass-4.5-safe upstream FA4 cute (FA4_CUTE_REF) over the
# Overlay the CuTe-DSL-4.6-compatible upstream FA4 cute (FA4_CUTE_REF) over the
# wheel/source one so the image runs FA4, not the FA2 fallback. This pulls the FA4
# runtime stack (cutlass-dsl, quack-kernels, apache-tvm-ffi, torch-c-dlpack-ext) --
# the same deps the [dreamverse] extra already installs in CI; the installed torch
+73 -28
View File
@@ -72,7 +72,10 @@ fastvideo/tests/performance/
│ writes Markdown summary + (optionally) uploads new records
├── dashboard.py
│ └── builds time-series Plotly HTML from HF history
└── hf_store.py # shared HF I/O + DataFrame helpers
fastvideo/performance/
├── hf_store.py # shared HF I/O + DataFrame helpers
└── metric_policy.py # shared rolling-baseline threshold policy
```
The HF dataset (`FastVideo/performance-tracking` by default) holds one
@@ -92,25 +95,28 @@ and recipe changes instead of treating all records for a model as equivalent.
## Metrics
Each benchmark records six metrics:
Each benchmark records six metrics. The rolling-baseline comparator also has a
per-metric policy with direction, percent threshold, absolute threshold, and a
`gated` flag.
| Metric | Raw key | Normalized key | Direction |
|---|---|---|---|
| End-to-end generation latency | `avg_generation_time_s` | `latency` | Lower is better |
| Video throughput | `throughput_fps` | `throughput` | Higher is better |
| Peak GPU memory | `max_peak_memory_mb` | `memory` | Lower is better |
| Text encoder time | `text_encoder_time_s` | `text_encoder_time_s` | Lower is better |
| DiT denoising time | `dit_time_s` | `dit_time_s` | Lower is better |
| VAE decode time | `vae_decode_time_s` | `vae_decode_time_s` | Lower is better |
| Metric | Raw key | Normalized key | Direction | Default rolling policy |
|---|---|---|---|---|
| End-to-end generation latency | `avg_generation_time_s` | `latency` | Lower is better | 8% and 0.5 s |
| Video throughput | `throughput_fps` | `throughput` | Higher is better | 8% and 0.05 FPS |
| Peak GPU memory | `max_peak_memory_mb` | `memory` | Lower is better | 5% and 256 MB |
| Text encoder time | `text_encoder_time_s` | `text_encoder_time_s` | Lower is better | 5% and 0.25 s |
| DiT denoising time | `dit_time_s` | `dit_time_s` | Lower is better | 5% and 0.25 s |
| VAE decode time | `vae_decode_time_s` | `vae_decode_time_s` | Lower is better | 5% and 0.25 s |
`test_inference_performance.py` temporarily sets `FASTVIDEO_STAGE_LOGGING=1`
while it runs so pipeline stage execution times are available in
`generate_video(...).logging_info`. Stage logs use pipeline-unique keys such as
`prompt_encoding_stage` so duplicate stage classes do not collide. For
`PipelineStage` entries, the extractor maps the `stage_class` field:
`TextEncodingStage` maps to `text_encoder_time_s`, `DenoisingStage` and
`DmdDenoisingStage` map to `dit_time_s`, and `DecodingStage` maps to
`vae_decode_time_s`, with a fallback for older logs that used the class name as
`PipelineStage` entries, shared component stage bases emit a stable
`component_metric`: text encoding stages map to `text_encoder_time_s`,
denoising stages and subclasses map to `dit_time_s`, and decoding stages map to
`vae_decode_time_s`. The extractor falls back to known `stage_class` names for
older logs that do not include `component_metric` or that used the class name as
the stage key. Generator-side timings such as `PostDecodeFrameProcessStage`,
`VideoSaveStage`, and `AudioMuxStage` are intentionally ignored. If a pipeline
does not report one of the mapped stages, that component metric is stored as
@@ -156,9 +162,22 @@ headroom and almost never need touching.
`compare_baseline.py` loads the last 5 successful, baseline-eligible records
for the same `(model_id, gpu_type)` from the HF dataset, computes the median
for each available metric, and fails if the current run regresses by more than
`PERF_MAX_REGRESSION` (default 5%). For latency, memory, and component times,
higher values are regressions. For throughput, lower values are regressions.
for each available metric, and evaluates the current run with the metric's
rolling regression policy. For latency, memory, and component times, higher
values are regressions. For throughput, lower values are regressions.
A metric exceeds its rolling threshold when both of these are true:
```text
percent_delta > threshold_percent
absolute_delta > threshold_absolute
```
Gated metrics fail CI when that threshold crossing happens. Set `gated: false`
for metrics that should remain visible in reports and the dashboard without
failing CI. Dashboard/API payloads expose `threshold_exceeded` separately from
`regressed`, where `regressed` means a gated CI failure. Missing or `null`
metrics are skipped.
This is the **drift detector** — it catches sub-threshold regressions that
slowly add up. Only scheduled-main successful records are baseline eligible.
@@ -239,6 +258,13 @@ Written by `test_inference_performance.py`. One file per benchmark run.
"max_dit_time_s": 10.0,
"max_vae_decode_time_s": 10.0
},
"regression_thresholds": {
"latency": {
"threshold_percent": 0.10,
"threshold_absolute": 1.0,
"gated": true
}
},
"commit": "<full sha>",
"pr_number": "1234",
"timestamp": "2026-05-08T22:00:00+00:00",
@@ -265,6 +291,13 @@ result, used as the rolling-baseline source of truth.
"text_encoder_time_s": 2.141,
"dit_time_s": 8.437,
"vae_decode_time_s": 3.208,
"regression_thresholds": {
"latency": {
"threshold_percent": 0.08,
"threshold_absolute": 0.5,
"gated": true
}
},
"success": true
}
```
@@ -281,18 +314,17 @@ successful main/full-suite uploads and remain eligible for rolling baselines.
| Variable | Default | Used by | Purpose |
|---|---|---|---|
| `PERF_MAX_REGRESSION` | `0.05` | `compare_baseline.py` | Per-metric regression fraction that fails the build. |
| `PERFORMANCE_TRACKING_ROOT` | `/tmp/perf-tracking` | `compare_baseline.py`, `dashboard.py` | Local directory the HF dataset is synced to. |
| `PERF_REPORTS_DIR` | `/root/data/perf_reports` | `compare_baseline.py`, `dashboard.py` | Where the Markdown summary and Plotly HTML get written for Buildkite to pick up. |
| `HF_REPO_ID` | `FastVideo/performance-tracking` | `hf_store.py` | HF dataset repo holding rolling-baseline records. |
| `HF_API_KEY`, `HUGGINGFACE_HUB_TOKEN`, `HF_TOKEN` | unset | `hf_store.py` | Required for upload or private dataset reads. |
| `HF_REPO_ID` | `FastVideo/performance-tracking` | `fastvideo/performance/hf_store.py` | HF dataset repo holding rolling-baseline records. |
| `HF_API_KEY`, `HUGGINGFACE_HUB_TOKEN`, `HF_TOKEN` | unset | `fastvideo/performance/hf_store.py` | Required for upload or private dataset reads. |
| `PERF_RUN_SOURCE` | inferred | `compare_baseline.py` | Source metadata for uploaded records: `pr`, `local`, `scheduled_main`, or `unknown`. |
| `PERF_UPLOAD_POLICY` | `never` | `compare_baseline.py` | Upload policy: `never`, `pass`, or `always`. |
| `PERF_PYTEST_RC` | unset | `compare_baseline.py` | Static-threshold pytest exit code, used so scheduled-main failures can be uploaded with `success=false`. |
| `TEST_SCOPE` | unset | `compare_baseline.py` | CI context used to infer scheduled-main runs together with `BUILDKITE_BRANCH=main`. |
| `BUILDKITE_BRANCH`, `BUILDKITE_COMMIT`, `BUILDKITE_PULL_REQUEST` | unset | `compare_baseline.py`, `test_inference_performance.py` | CI metadata stamped into records. |
| `DASHBOARD_DAYS` | `30` | `dashboard.py` | Lookback window for the Plotly trend pages. |
| `PERFORMANCE_TRACKING_SYNC_REUSE_TTL_SECONDS` | `3600` | `hf_store.py` | Freshness window for reusing an existing HF sync when requested by dashboard consumers. |
| `PERFORMANCE_TRACKING_SYNC_REUSE_TTL_SECONDS` | `3600` | `fastvideo/performance/hf_store.py` | Freshness window for reusing an existing HF sync when requested by dashboard consumers. |
| `FASTVIDEO_STAGE_LOGGING` | set by the pytest test | `test_inference_performance.py` | Enables pipeline stage timing capture for component metrics during benchmark runs. |
## CI integration
@@ -304,9 +336,11 @@ Buildkite artifact upload is in
`.buildkite/scripts/pr_test.sh:upload_performance_artifacts`.
Each performance build runs pytest first. If that fixed-threshold phase fails,
`compare_baseline.py` is skipped, so Markdown summaries and normalized JSON
artifacts are not emitted. The dashboard still runs best-effort for
observability. When pytest passes, the rolling-baseline phase emits:
PR/direct runs skip `compare_baseline.py` because they only upload passing
records. Scheduled-main runs still execute `compare_baseline.py` with
`PERF_PYTEST_RC` set so the failed canonical attempt is visible in normalized
JSON and dashboard history. The dashboard runs best-effort for observability.
When the rolling-baseline phase runs, it emits:
* **Markdown summary** — appended to `$GITHUB_STEP_SUMMARY` when that variable
is set, and written as `perf_<sha>_<ts>.md` for Buildkite upload. Contains a
@@ -347,9 +381,13 @@ observability. When pytest passes, the rolling-baseline phase emits:
"max_vae_decode_time_s": 10.0
},
"default": { "max_generation_time_s": 120.0, "max_peak_memory_mb": 30000.0 }
},
"regression_thresholds": {
"latency": { "threshold_percent": 0.10, "threshold_absolute": 1.0, "gated": true }
}
}
```
}
```
Legacy v1 configs without `config_schema_version` still load, but should not
gain v2 identity or metadata fields until they are migrated to
@@ -372,6 +410,13 @@ observability. When pytest passes, the rolling-baseline phase emits:
a useful fixed gate. The rolling baseline will still track component times
when static component thresholds are omitted.
6. Omit `regression_thresholds` to use the default rolling-baseline policy, or
include only benchmark-specific deviations. Tune these independently from
the fixed thresholds when a metric is noisy or should be informational. The
fixed `thresholds` block is an absolute pytest ceiling. The
`regression_thresholds` block controls rolling-baseline comparisons against
recent scheduled-main records.
## Troubleshooting
**"No baseline for ... Initializing"** — first run for this `(model_id,
@@ -388,5 +433,5 @@ pipelines that did not report a mapped component stage.
**Component timing is `null`** — the generated result did not include a mapped
stage in `logging_info.stages`. Check that the pipeline emits stage logging
and that the stage name is listed in `STAGE_METRIC_MAP` in
`test_inference_performance.py`.
and that the stage emits `component_metric` or is covered by the legacy
`STAGE_METRIC_MAP` fallback in `test_inference_performance.py`.
+2 -2
View File
@@ -58,8 +58,8 @@ pipeline initialization and sampling.
| FastWan2.1 T2V 1.3B | `FastVideo/FastWan2.1-T2V-1.3B-Diffusers` | 480P | ⭕ | ⭕ | ⭕ | ✅ | ⭕ |
| FastWan2.2 TI2V 5B Full Attn* | `FastVideo/FastWan2.2-TI2V-5B-FullAttn-Diffusers` | 720P | ⭕ | ⭕ | ⭕ | ✅ | ⭕ |
| Wan2.2 TI2V 5B | `Wan-AI/Wan2.2-TI2V-5B-Diffusers` | 720P | ⭕ | ⭕ | ✅ | ⭕ | ⭕ |
| DreamX-World 5B Cam | `GD-ML/DreamX-World-5B-Cam` | 480P | ⭕ | ⭕ | ⭕ | ⭕ | ⭕ |
| DreamX-World 5B AR | `GD-ML/DreamX-World-5B` | 704px1280p | ⭕ | ⭕ | ⭕ | ⭕ | ⭕ |
| DreamX-World 5B Cam | `FastVideo/DreamX-World-5B-Cam-Diffusers` | 480P | ⭕ | ⭕ | ⭕ | ⭕ | ⭕ |
| DreamX-World 5B AR | `FastVideo/DreamX-World-5B-Diffusers` | 704px1280p | ⭕ | ⭕ | ⭕ | ⭕ | ⭕ |
| Lucy Edit Dev 5B*** | `decart-ai/Lucy-Edit-Dev` | 480P | ⭕ | ⭕ | ⭕ | ⭕ | ⭕ |
| Wan2.2 T2V A14B | `Wan-AI/Wan2.2-T2V-A14B-Diffusers` | 480P<br>720P | ❌ | ❌ | ✅ | ⭕ | ⭕ |
| Wan2.2 I2V A14B | `Wan-AI/Wan2.2-I2V-A14B-Diffusers` | 480P<br>720P | ❌ | ❌ | ✅ | ⭕ | ⭕ |
@@ -15,7 +15,7 @@ def _env_float(name: str, default: float) -> float:
def main():
model_name = os.getenv("DREAMX_WORLD_MODEL_DIR", "GD-ML/DreamX-World-5B-Cam")
model_name = os.getenv("DREAMX_WORLD_MODEL_DIR", "FastVideo/DreamX-World-5B-Cam-Diffusers")
generator = VideoGenerator.from_pretrained(
model_name,
num_gpus=1,
+8
View File
@@ -23,6 +23,7 @@ If you only need to use the distributed environment without model parallelism,
you can skip the model parallel initialization and destruction steps.
"""
import contextlib
import gc
import os
import pickle
import weakref
@@ -1006,6 +1007,13 @@ def cleanup_dist_env_and_memory(shutdown_ray: bool = False):
if shutdown_ray:
import ray # Lazy import Ray
ray.shutdown()
# Actually free GPU memory, as the name promises. FSDP-wrapped modules sit
# in reference cycles (module <-> FSDP state <-> hooks), so dropping the
# last user reference does not free their parameters until a gc pass runs.
# Without this, back-to-back model-load tests in one pytest session OOM.
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
def get_same_node_ranks(pg: ProcessGroup | StatelessProcessGroup, source_rank: int = 0) -> list[int]:
+2
View File
@@ -0,0 +1,2 @@
# SPDX-License-Identifier: Apache-2.0
"""Performance benchmark and dashboard utilities."""
@@ -57,9 +57,7 @@ def is_baseline_eligible_record(record: dict[str, Any]) -> bool:
"""
if record.get("baseline_eligible") is True:
return True
if "baseline_eligible" not in record and "run_source" not in record:
return True
return False
return "baseline_eligible" not in record and "run_source" not in record
def resolve_hf_token() -> str | None:
@@ -312,8 +310,12 @@ def load_records_for_model(
# ---------------------------------------------------------------------------
_NUMERIC_COLS = (
"latency", "throughput", "memory",
"text_encoder_time_s", "dit_time_s", "vae_decode_time_s",
"latency",
"throughput",
"memory",
"text_encoder_time_s",
"dit_time_s",
"vae_decode_time_s",
)
+116
View File
@@ -0,0 +1,116 @@
# SPDX-License-Identifier: Apache-2.0
"""Metric policy for rolling performance baseline comparisons."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class MetricPolicy:
key: str
label: str
precision: int
lower_is_better: bool
threshold_percent: float
threshold_absolute: float
gated: bool = True
@dataclass(frozen=True)
class MetricDelta:
absolute: float
percent: float
threshold_exceeded: bool
regressed: bool
DEFAULT_METRIC_POLICIES: tuple[MetricPolicy, ...] = (
MetricPolicy("latency", "Latency", 3, True, 0.08, 0.5),
MetricPolicy("throughput", "Throughput", 3, False, 0.08, 0.05),
MetricPolicy("memory", "Memory", 1, True, 0.05, 256.0),
MetricPolicy("text_encoder_time_s", "Text Enc", 3, True, 0.05, 0.25),
MetricPolicy("dit_time_s", "DiT", 3, True, 0.05, 0.25),
MetricPolicy("vae_decode_time_s", "VAE Decode", 3, True, 0.05, 0.25),
)
def _optional_float(value: Any) -> float | None:
if value is None or isinstance(value, bool):
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def _optional_bool(value: Any) -> bool | None:
if isinstance(value, bool):
return value
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in {"1", "true", "yes", "on"}:
return True
if normalized in {"0", "false", "no", "off"}:
return False
return None
def resolve_metric_policies(threshold_overrides: Mapping[str, Any] | None, ) -> tuple[MetricPolicy, ...]:
"""Return default metric policies with optional per-metric overrides."""
if not isinstance(threshold_overrides, Mapping):
threshold_overrides = {}
policies: list[MetricPolicy] = []
for base_policy in DEFAULT_METRIC_POLICIES:
raw_override = threshold_overrides.get(base_policy.key, {})
if not isinstance(raw_override, Mapping):
raw_override = {}
threshold_percent = _optional_float(raw_override.get("threshold_percent"))
threshold_absolute = _optional_float(raw_override.get("threshold_absolute"))
gated = _optional_bool(raw_override.get("gated"))
policies.append(
MetricPolicy(
key=base_policy.key,
label=base_policy.label,
precision=base_policy.precision,
lower_is_better=base_policy.lower_is_better,
threshold_percent=(base_policy.threshold_percent if threshold_percent is None else threshold_percent),
threshold_absolute=(base_policy.threshold_absolute
if threshold_absolute is None else threshold_absolute),
gated=base_policy.gated if gated is None else gated,
))
return tuple(policies)
def serialize_metric_thresholds(policies: tuple[MetricPolicy, ...], ) -> dict[str, dict[str, float | bool]]:
return {
policy.key: {
"threshold_percent": policy.threshold_percent,
"threshold_absolute": policy.threshold_absolute,
"gated": policy.gated,
}
for policy in policies
}
def regression_delta(
policy: MetricPolicy,
current: float,
baseline: float,
) -> MetricDelta | None:
if baseline <= 0:
return None
absolute_delta = current - baseline if policy.lower_is_better else baseline - current
percent_delta = absolute_delta / baseline
threshold_exceeded = (percent_delta > policy.threshold_percent and absolute_delta > policy.threshold_absolute)
return MetricDelta(
absolute=absolute_delta,
percent=percent_delta,
threshold_exceeded=threshold_exceeded,
regressed=policy.gated and threshold_exceeded,
)
+1 -2
View File
@@ -13,7 +13,7 @@ from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from fastvideo.tests.performance import hf_store
from fastvideo.performance import hf_store
from .service import build_latest_summary, build_trends, filter_records
@@ -150,7 +150,6 @@ def create_app(store: PerformanceDataStore | None = None) -> FastAPI:
filtered = filter_records(loaded, model_id=model_id, gpu_type=gpu_type)
rows = build_latest_summary(
filtered,
max_regression=float(os.environ.get("PERF_MAX_REGRESSION", "0.05")),
run_source=run_source,
)
return {
+2 -20
View File
@@ -1,26 +1,8 @@
# SPDX-License-Identifier: Apache-2.0
"""Metric definitions shared by the performance dashboard backend."""
from __future__ import annotations
from fastvideo.performance.metric_policy import DEFAULT_METRIC_POLICIES
from dataclasses import dataclass
@dataclass(frozen=True)
class MetricDefinition:
key: str
label: str
precision: int
lower_is_better: bool
METRICS: tuple[MetricDefinition, ...] = (
MetricDefinition("latency", "Latency", 3, True),
MetricDefinition("throughput", "Throughput", 3, False),
MetricDefinition("memory", "Memory", 1, True),
MetricDefinition("text_encoder_time_s", "Text Encoder", 3, True),
MetricDefinition("dit_time_s", "DiT", 3, True),
MetricDefinition("vae_decode_time_s", "VAE Decode", 3, True),
)
METRICS = DEFAULT_METRIC_POLICIES
METRIC_BY_KEY = {metric.key: metric for metric in METRICS}
+41 -45
View File
@@ -13,9 +13,8 @@ from collections import defaultdict
from datetime import datetime, timezone
from typing import Any
from fastvideo.tests.performance.hf_store import is_baseline_eligible_record, safe_float
from .metrics import METRICS
from fastvideo.performance.hf_store import is_baseline_eligible_record, safe_float
from fastvideo.performance.metric_policy import regression_delta, resolve_metric_policies
Record = dict[str, Any]
@@ -95,19 +94,9 @@ def baseline_value(records: list[Record], metric_key: str) -> float | None:
return float(statistics.median(values))
def regression_percent(metric_key: str, current: float | None, baseline: float | None) -> float | None:
if current is None or baseline is None or baseline <= 0:
return None
metric = next(metric for metric in METRICS if metric.key == metric_key)
if metric.lower_is_better:
return (current - baseline) / baseline * 100.0
return (baseline - current) / baseline * 100.0
def build_latest_summary(records: list[Record],
*,
baseline_window: int = 5,
max_regression: float = 0.05,
run_source: str | None = None) -> list[Record]:
rows: list[Record] = []
for (model_id, gpu_type), group in group_by_model_gpu(records).items():
@@ -123,52 +112,58 @@ def build_latest_summary(records: list[Record],
if record is not latest and record.get("success", True) and is_baseline_eligible_record(record)
]
baseline_records = baseline_pool[-baseline_window:]
metric_policies = resolve_metric_policies(latest.get("regression_thresholds"))
metrics: dict[str, Record] = {}
regressions: list[float] = []
for metric in METRICS:
current = safe_float(latest.get(metric.key))
baseline = baseline_value(baseline_records, metric.key)
regression = regression_percent(metric.key, current, baseline)
metrics[metric.key] = {
failing_metrics: list[str] = []
threshold_exceeded_metrics: list[str] = []
for policy in metric_policies:
current = safe_float(latest.get(policy.key))
baseline = baseline_value(baseline_records, policy.key)
delta = None
if current is not None and baseline is not None:
delta = regression_delta(policy, current, baseline)
regression = None if delta is None else delta.percent * 100.0
metrics[policy.key] = {
"current": current,
"baseline": baseline,
"regression_pct": regression,
"label": metric.label,
"lower_is_better": metric.lower_is_better,
"precision": metric.precision,
"absolute_delta": None if delta is None else delta.absolute,
"threshold_percent": policy.threshold_percent * 100.0,
"threshold_absolute": policy.threshold_absolute,
"gated": policy.gated,
"threshold_exceeded": False if delta is None else delta.threshold_exceeded,
"regressed": False if delta is None else delta.regressed,
"label": policy.label,
"lower_is_better": policy.lower_is_better,
"precision": policy.precision,
}
if regression is not None:
regressions.append(regression)
if delta is not None and delta.threshold_exceeded:
threshold_exceeded_metrics.append(policy.key)
if delta is not None and delta.regressed:
failing_metrics.append(policy.key)
worst_regression = max(regressions) if regressions else None
success = bool(latest.get("success", True))
status = "pass" if success else "fail"
rows.append({
"model_id":
model_id,
"gpu_type":
gpu_type,
"timestamp":
latest.get("timestamp"),
"commit_sha":
latest.get("commit_sha"),
"model_id": model_id,
"gpu_type": gpu_type,
"timestamp": latest.get("timestamp"),
"commit_sha": latest.get("commit_sha"),
**record_metadata(latest),
"success":
success,
"baseline_n":
len(baseline_records),
"worst_regression_pct":
worst_regression,
"regression_threshold_pct":
max_regression * 100.0,
"computed_regression_status":
"fail" if worst_regression is not None and worst_regression > max_regression * 100.0 else "pass",
"status":
status,
"metrics":
metrics,
"success": success,
"baseline_n": len(baseline_records),
"worst_regression_pct": worst_regression,
"threshold_exceeded_metrics": threshold_exceeded_metrics,
"failing_metrics": failing_metrics,
"computed_regression_status": "fail" if failing_metrics else "pass",
"status": status,
"metrics": metrics,
})
return sorted(rows, key=lambda row: (row["status"] != "fail", row["model_id"], row["gpu_type"]))
@@ -179,14 +174,15 @@ def build_trends(records: list[Record]) -> list[Record]:
for (model_id, gpu_type), group in group_by_model_gpu(records).items():
points = []
for record in group:
metric_policies = resolve_metric_policies(record.get("regression_thresholds"))
point = {
"timestamp": record.get("timestamp"),
"commit_sha": record.get("commit_sha"),
**record_metadata(record),
"success": bool(record.get("success", True)),
"metrics": {
metric.key: safe_float(record.get(metric.key))
for metric in METRICS
policy.key: safe_float(record.get(policy.key))
for policy in metric_policies
},
}
points.append(point)
+4
View File
@@ -34,6 +34,7 @@ class PipelineStage(ABC):
composed with other stages to create a complete pipeline. Each stage is responsible
for a specific part of the process, such as prompt encoding, latent preparation, etc.
"""
performance_component_metric: str | None = None
def verify_input(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult:
"""
@@ -155,6 +156,9 @@ class PipelineStage(ABC):
logger.info("[%s] Execution completed in %s ms", stage_name, execution_time * 1000)
batch.logging_info.add_stage_execution_time(stage_key, execution_time)
batch.logging_info.add_stage_metric(stage_key, "stage_class", stage_class_name)
component_metric = self.performance_component_metric
if component_metric is not None:
batch.logging_info.add_stage_metric(stage_key, "component_metric", component_metric)
except Exception as e:
torch.cuda.synchronize()
execution_time = time.perf_counter() - start_time
+1
View File
@@ -28,6 +28,7 @@ class DecodingStage(PipelineStage):
This stage handles the decoding of latent representations into the final
output format (e.g., pixel values).
"""
performance_component_metric = "vae_decode_time_s"
def __init__(self, vae, pipeline=None) -> None:
self.vae: ParallelTiledVAE = vae
+2
View File
@@ -51,6 +51,7 @@ class DenoisingStage(PipelineStage):
This stage handles the iterative denoising process that transforms
the initial noise into the final output.
"""
performance_component_metric = "dit_time_s"
def __init__(self, transformer, scheduler, pipeline=None, transformer_2=None, vae=None) -> None:
super().__init__()
@@ -1206,6 +1207,7 @@ class Cosmos25V2WDenoisingStage(Cosmos25DenoisingStage):
class Cosmos25AutoDenoisingStage(PipelineStage):
"""Route Cosmos 2.5 denoising to T2W vs V2W/I2W."""
performance_component_metric = "dit_time_s"
def __init__(self, transformer, scheduler) -> None:
super().__init__()
@@ -24,6 +24,7 @@ class TextEncodingStage(PipelineStage):
This stage handles the encoding of text prompts into the embedding space
expected by the diffusion model.
"""
performance_component_metric = "text_encoder_time_s"
def __init__(self, text_encoders, tokenizers) -> None:
"""
@@ -350,6 +351,7 @@ class Cosmos25TextEncodingStage(PipelineStage):
Cosmos 2.5 uses Reason1 (Qwen2.5-VL) and relies on the encoder's
`compute_text_embeddings_online()`.
"""
performance_component_metric = "text_encoder_time_s"
def __init__(self, text_encoder) -> None:
super().__init__()
+2 -2
View File
@@ -780,7 +780,7 @@ def _register_configs() -> None:
pipeline_config_cls=DreamXWorld5BCamPipelineConfig,
workload_types=(WorkloadType.I2V, ),
hf_model_paths=[
"GD-ML/DreamX-World-5B-Cam",
"FastVideo/DreamX-World-5B-Cam-Diffusers",
],
model_detectors=[
# Mutually exclusive with the AR detector below: Cam requires an
@@ -798,7 +798,7 @@ def _register_configs() -> None:
pipeline_config_cls=DreamXWorld5BARPipelineConfig,
workload_types=(WorkloadType.I2V, ),
hf_model_paths=[
"GD-ML/DreamX-World-5B",
"FastVideo/DreamX-World-5B-Diffusers",
],
model_detectors=[
lambda path:
@@ -17,46 +17,6 @@ class TestLegacyTorchCompileKwargsTranslation:
first-class :class:`CompileConfig` fields and anything unknown falls
into ``extras``."""
def test_all_typed_keys_promoted(self) -> None:
config = legacy_from_pretrained_to_config(
"/models/ltx2",
{
"enable_torch_compile": True,
"torch_compile_kwargs": {
"backend": "inductor",
"fullgraph": True,
"mode": "max-autotune-no-cudagraphs",
"dynamic": False,
},
},
)
compile_config = config.engine.compile
assert compile_config.enabled is True
assert compile_config.backend == "inductor"
assert compile_config.fullgraph is True
assert compile_config.mode == "max-autotune-no-cudagraphs"
assert compile_config.dynamic is False
assert compile_config.extras == {}
def test_unknown_keys_land_in_extras(self) -> None:
config = legacy_from_pretrained_to_config(
"/models/ltx2",
{
"enable_torch_compile": True,
"torch_compile_kwargs": {
"backend": "inductor",
"options": {"triton.cudagraphs": False},
"disable": False,
},
},
)
compile_config = config.engine.compile
assert compile_config.backend == "inductor"
assert compile_config.extras == {
"options": {"triton.cudagraphs": False},
"disable": False,
}
def test_empty_kwargs_produces_empty_extras(self) -> None:
config = legacy_from_pretrained_to_config(
"/models/ltx2",
@@ -117,13 +77,6 @@ class TestLegacyLtx2VaeTilingTranslation:
``generator.pipeline.vae_tiling``; reverse direction emits the
legacy name back to FastVideoArgs."""
def test_forward_routes_to_pipeline_vae_tiling(self) -> None:
config = legacy_from_pretrained_to_config(
"/models/ltx2",
{"ltx2_vae_tiling": False},
)
assert config.pipeline.vae_tiling is False
def test_true_round_trips(self) -> None:
config = legacy_from_pretrained_to_config(
"/models/ltx2",
@@ -0,0 +1,84 @@
# SPDX-License-Identifier: Apache-2.0
"""Guard: every test directory must be collected by some CI lane or be on
the explicit allowlist below.
Three separate incidents on 2026-07-05 found test files that no CI lane
ever collects (fastvideo/tests/stages/, tests/local_tests/ additions in
PR #1509, and this sweep found seven dark directories in total): the tests
pass review, merge, and then silently never run. This test makes going
dark an explicit, reviewed decision instead of an accident: adding a new
test directory fails CI until it is either wired into a lane or
allowlisted here with a reason.
Pure text analysis — no fastvideo imports, no GPU, no torch.
"""
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[3]
TESTS_ROOT = REPO_ROOT / "fastvideo" / "tests"
# Files whose text constitutes "a CI lane references this directory".
CI_SOURCES = [
TESTS_ROOT / "modal" / "pr_test.py",
TESTS_ROOT / "modal" / "ssim_test.py",
*sorted((REPO_ROOT / ".buildkite").rglob("*.yml")),
*sorted((REPO_ROOT / ".buildkite").rglob("*.sh")),
]
# Directories that intentionally have no CI lane today. Every entry needs a
# reason; remove the entry when the directory gets wired into a lane.
# State as found on 2026-07-05 — these SHOULD shrink over time, not grow.
ALLOWLIST = {
"attention": "no lane yet — GPU attention-backend tests, run manually",
"audio": "no lane yet — audio encoder tests, run manually",
"distributed": "no lane yet — multi-GPU torchrun tests, run manually",
"hooks": "no lane yet — run manually",
"layers": "no lane yet — torchrun FSDP dispatch tests, run manually",
"nightly": "by design: nightly cadence, not per-PR",
"modal": "CI infrastructure itself, not a test suite",
}
def _dirs_with_tests() -> list[str]:
dirs = []
for child in sorted(TESTS_ROOT.iterdir()):
if child.is_dir() and any(child.rglob("test_*.py")):
dirs.append(child.name)
return dirs
def _ci_text() -> str:
return "\n".join(
src.read_text(errors="replace") for src in CI_SOURCES if src.exists())
def test_every_test_directory_is_collected_or_allowlisted():
ci_text = _ci_text()
dark = [
name for name in _dirs_with_tests()
if f"tests/{name}" not in ci_text and name not in ALLOWLIST
]
assert not dark, (
f"Test directories not referenced by any CI lane and not "
f"allowlisted: {dark}. Wire them into a lane in "
f"fastvideo/tests/modal/pr_test.py (or a Buildkite step), or add an "
f"allowlist entry with a reason in {__file__}.")
def test_local_tests_stays_out_of_ci():
# tests/local_tests/ (repo root) is developer-local by design (author
# decision, 2026-07-05): parity scaffolds and machine-specific checks
# that must never gate CI. Fail if any CI source starts collecting it.
assert "tests/local_tests" not in _ci_text(), (
"tests/local_tests/ is local-only by design; remove the CI "
"reference or move the tests into a fastvideo/tests/ lane.")
def test_allowlist_entries_are_still_real_directories():
# A stale allowlist hides regressions; entries must track reality.
missing = [
name for name in ALLOWLIST
if name != "modal" and not (TESTS_ROOT / name).is_dir()
]
assert not missing, (
f"Allowlisted directories no longer exist — remove them: {missing}")
+13 -1
View File
@@ -187,6 +187,8 @@ def run_transformer_tests():
@app.function(gpu="L40S:4",
cpu=16.0,
memory=65536,
image=image,
timeout=900,
secrets=[
@@ -201,6 +203,8 @@ def run_training_tests():
@app.function(gpu="L40S:2",
cpu=16.0,
memory=65536,
image=image,
timeout=900,
secrets=[
@@ -276,7 +280,7 @@ def run_self_forcing_tests():
@app.function(gpu="L40S:1", image=image, timeout=900)
def run_unit_test():
run_test(
"pytest ./fastvideo/tests/api/ ./fastvideo/tests/contract/ ./fastvideo/tests/dataset/ ./fastvideo/tests/workflow/ ./fastvideo/tests/entrypoints/ ./fastvideo/tests/train/ --ignore=./fastvideo/tests/entrypoints/test_openai_api_integration.py --ignore=./fastvideo/tests/train/models --ignore=./fastvideo/tests/train/methods -vs"
"pytest ./fastvideo/tests/api/ ./fastvideo/tests/contract/ ./fastvideo/tests/dataset/ ./fastvideo/tests/workflow/ ./fastvideo/tests/entrypoints/ ./fastvideo/tests/train/ ./fastvideo/tests/stages/ ./fastvideo/tests/ops/ --ignore=./fastvideo/tests/entrypoints/test_openai_api_integration.py --ignore=./fastvideo/tests/train/models --ignore=./fastvideo/tests/train/methods -vs"
)
@@ -318,6 +322,8 @@ def run_dreamverse_app_tests():
@app.function(gpu="L40S:1",
cpu=16.0,
memory=65536,
image=image,
timeout=1800,
secrets=[
@@ -400,6 +406,8 @@ def run_lora_extraction_tests():
@app.function(gpu="L40S:2",
cpu=16.0,
memory=65536,
image=image,
timeout=1800,
secrets=[
@@ -427,6 +435,8 @@ def run_performance_tests():
"export PERF_RUN_SOURCE='unknown'; "
"export PERF_UPLOAD_POLICY='never'; "
"fi; "
"(nvidia-smi --query-gpu=index,timestamp,clocks.sm,clocks.max.sm,power.draw,power.limit,temperature.gpu "
"--format=csv -l 10 > /tmp/gpu_telemetry.csv 2>/dev/null &); "
"pytest ./fastvideo/tests/performance -vs; "
"PYTEST_RC=$?; "
"PERF_RC=0; "
@@ -435,6 +445,8 @@ def run_performance_tests():
"PERF_RC=$?; "
"fi; "
"python ./fastvideo/tests/performance/dashboard.py || true; "
"echo '--- GPU telemetry (clocks.sm vs clocks.max.sm reveals capped hosts) ---'; "
"cat /tmp/gpu_telemetry.csv || true; "
"FINAL_RC=$PYTEST_RC; "
"if [ $FINAL_RC -eq 0 ]; then FINAL_RC=$PERF_RC; fi; "
"exit $FINAL_RC")
@@ -34,6 +34,9 @@ class TestAbsMaxFP8LinearMethod(unittest.TestCase):
method._convert_scale(scale)
def test_create_weights_rejects_invalid_dtype(self):
# float64 is outside the supported set (bf16/fp16/fp32). The test
# originally passed float32, which create_weights has accepted since
# PR #981 — it never ran in CI, so the mismatch went unnoticed.
method = AbsMaxFP8LinearMethod()
layer = nn.Module()
with self.assertRaisesRegex(AssertionError, "only supports"):
@@ -43,7 +46,7 @@ class TestAbsMaxFP8LinearMethod(unittest.TestCase):
output_partition_sizes=[3],
input_size=2,
output_size=3,
params_dtype=torch.float32,
params_dtype=torch.float64,
)
def test_absmax_fp8_parameter_weight_loader(self):
+92 -75
View File
@@ -8,8 +8,8 @@ This script:
baseline-eligible successful records (filtered by gpu_type),
4) writes normalized records back to the HF dataset repo according to
PERF_UPLOAD_POLICY,
5) exits non-zero if any metric regresses by more than PERF_MAX_REGRESSION
(default 5%).
5) exits non-zero if any gated metric exceeds both its percent and absolute
regression floors.
"""
import glob
@@ -21,21 +21,36 @@ from datetime import datetime, timezone
from typing import Any
try:
from .hf_store import (
from fastvideo.performance.hf_store import (
load_records_for_model,
safe_float,
sanitize,
sync_from_hf,
upload_record,
)
from fastvideo.performance.metric_policy import (
MetricPolicy,
regression_delta,
resolve_metric_policies,
serialize_metric_thresholds,
)
except ImportError:
from hf_store import (
repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
if repo_root not in sys.path:
sys.path.insert(0, repo_root)
from fastvideo.performance.hf_store import (
load_records_for_model,
safe_float,
sanitize,
sync_from_hf,
upload_record,
)
from fastvideo.performance.metric_policy import (
MetricPolicy,
regression_delta,
resolve_metric_policies,
serialize_metric_thresholds,
)
RESULTS_DIR = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
@@ -46,25 +61,9 @@ TRACKING_ROOT = os.environ.get(
"/tmp/perf-tracking",
)
PERF_REPORTS_DIR = os.environ.get("PERF_REPORTS_DIR", "/root/data/perf_reports")
MAX_REGRESSION = float(os.environ.get("PERF_MAX_REGRESSION", "0.05"))
UPLOAD_POLICY = os.environ.get("PERF_UPLOAD_POLICY", "never").strip().lower()
VALID_UPLOAD_POLICIES = {"never", "pass", "always"}
VALID_RUN_SOURCES = {"pr", "local", "scheduled_main", "unknown"}
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:
@@ -168,6 +167,7 @@ def normalize_performance_result(result: dict[str, Any]) -> dict[str, Any]:
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"))
metric_policies = resolve_metric_policies(result.get("regression_thresholds"))
return {
"model_id": model_id,
@@ -180,6 +180,7 @@ def normalize_performance_result(result: dict[str, Any]) -> dict[str, Any]:
"text_encoder_time_s": text_encoder_time,
"dit_time_s": dit_time,
"vae_decode_time_s": vae_decode_time,
"regression_thresholds": serialize_metric_thresholds(metric_policies),
"success": True,
**_record_metadata(_detect_run_source(), result),
}
@@ -229,55 +230,41 @@ def _baseline_metric(records: list[dict[str, Any]], key: str) -> float | None:
return statistics.median(values)
def _metric_policy_summary(policy: MetricPolicy) -> str:
gated = "gated" if policy.gated else "info"
return (
f"{gated}, >{policy.threshold_percent * 100:.1f}% "
f"and >{policy.threshold_absolute:.{policy.precision}f}"
)
def _check_regressions(
current: dict[str, Any],
baseline_records: list[dict[str, Any]],
max_regression: float,
metric_policies: tuple[MetricPolicy, ...],
) -> list[str]:
failures: list[str] = []
for metric, _label, _precision in METRICS:
if metric not in LOWER_IS_BETTER_METRICS:
for policy in metric_policies:
baseline = _baseline_metric(baseline_records, policy.key)
curr = safe_float(current.get(policy.key))
if baseline is None or curr is None:
continue
baseline = _baseline_metric(baseline_records, metric)
curr = safe_float(current.get(metric))
if baseline is None or curr is None or baseline <= 0:
delta = regression_delta(policy, curr, baseline)
if delta is None or not delta.regressed:
continue
regression = (curr - baseline) / baseline
if regression > max_regression:
failures.append(f"{current['model_id']} {metric} regressed by "
f"{regression * 100:.1f}% "
f"(current={curr:.3f}, baseline_median={baseline:.3f})")
baseline_tp = _baseline_metric(baseline_records, "throughput")
curr_tp = safe_float(current.get("throughput"))
if baseline_tp is not None and curr_tp is not None and baseline_tp > 0:
regression = (baseline_tp - curr_tp) / baseline_tp
if regression > max_regression:
failures.append(f"{current['model_id']} throughput regressed by "
f"{regression * 100:.1f}% "
f"(current={curr_tp:.3f}, baseline_median={baseline_tp:.3f})")
failures.append(
f"{current['model_id']} {policy.key} regressed by "
f"{delta.percent * 100:.1f}% and "
f"{delta.absolute:.{policy.precision}f} "
f"(current={curr:.{policy.precision}f}, "
f"baseline_median={baseline:.{policy.precision}f}, "
f"threshold={_metric_policy_summary(policy)})"
)
return failures
def _metric_delta_percent(
metric: str,
current: dict[str, Any],
baseline_records: list[dict[str, Any]],
) -> float | None:
curr = safe_float(current.get(metric))
baseline = _baseline_metric(baseline_records, metric)
if curr is None or baseline is None or baseline <= 0:
return None
if metric in LOWER_IS_BETTER_METRICS:
return (curr - baseline) / baseline * 100.0
if metric == "throughput":
return (baseline - curr) / baseline * 100.0
return None
def _compact_value(value: float | None, precision: int = 3) -> str:
if value is None:
return "n/a"
@@ -287,23 +274,42 @@ def _compact_value(value: float | None, precision: int = 3) -> str:
def _build_summary_row(
record: dict[str, Any],
baseline_records: list[dict[str, Any]],
metric_policies: tuple[MetricPolicy, ...],
has_failed: bool,
) -> dict[str, Any]:
"""Format a single benchmark result as a row for the Markdown table."""
metric_values: dict[str, dict[str, float | None]] = {}
metric_values: dict[str, dict[str, Any]] = {}
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] = {
failing_metrics: list[str] = []
threshold_exceeded_metrics: list[str] = []
for policy in metric_policies:
curr = safe_float(record.get(policy.key))
baseline = _baseline_metric(baseline_records, policy.key)
delta = (
regression_delta(policy, curr, baseline)
if curr is not None and baseline is not None
else None
)
regression = None if delta is None else delta.percent * 100.0
absolute_delta = None if delta is None else delta.absolute
metric_values[policy.key] = {
"curr": curr,
"base": baseline,
"regression_pct": regression,
"absolute_delta": absolute_delta,
"threshold_percent": policy.threshold_percent * 100.0,
"threshold_absolute": policy.threshold_absolute,
"gated": policy.gated,
"threshold_exceeded": False if delta is None else delta.threshold_exceeded,
"regressed": False if delta is None else delta.regressed,
}
if regression is not None:
regressions.append(regression)
if delta is not None and delta.threshold_exceeded:
threshold_exceeded_metrics.append(policy.key)
if delta is not None and delta.regressed:
failing_metrics.append(policy.key)
worst_regression_pct = max(regressions) if regressions else None
@@ -313,40 +319,50 @@ def _build_summary_row(
"baseline_n": len(baseline_records),
"metrics": metric_values,
"worst_regression_pct": worst_regression_pct,
"threshold_exceeded_metrics": threshold_exceeded_metrics,
"failing_metrics": failing_metrics,
"failed": has_failed,
}
def _build_markdown_summary(
summary_rows: list[dict[str, Any]],
max_regression: float,
metric_policies: tuple[MetricPolicy, ...],
) -> str:
lines = [
"## Performance Baseline Comparison",
"",
f"Threshold: regressions greater than {max_regression * 100:.1f}% fail",
"Threshold: gated metrics fail only when both percent and absolute "
"regression floors are exceeded.",
"",
("| 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 |"),
"|---|---|---:|---|---|---|---|---|---|---:|---|",
"VAE Decode (curr/base) | Worst Regression | Exceeded Metrics | "
"Failing Metrics | 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)}")
for policy in metric_policies:
values = row["metrics"][policy.key]
metric_cells.append(f"{_compact_value(values['curr'], policy.precision)} / "
f"{_compact_value(values['base'], policy.precision)}")
worst_reg = ("n/a" if row["worst_regression_pct"] is None else f"{row['worst_regression_pct']:.1f}%")
exceeded_metrics = (
", ".join(row["threshold_exceeded_metrics"])
if row["threshold_exceeded_metrics"]
else "none"
)
failing_metrics = ", ".join(row["failing_metrics"]) if row["failing_metrics"] else "none"
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"{worst_reg} | {status} |")
f"{worst_reg} | {exceeded_metrics} | {failing_metrics} | {status} |")
return "\n".join(lines) + "\n"
@@ -400,6 +416,7 @@ def main() -> int:
for raw in current_results:
record = _normalize_record(raw)
metric_policies = resolve_metric_policies(record.get("regression_thresholds"))
baseline_records = load_records_for_model(
TRACKING_ROOT,
@@ -416,7 +433,7 @@ def main() -> int:
failures: list[str] = []
record["success"] = True
else:
failures = _check_regressions(record, baseline_records, MAX_REGRESSION)
failures = _check_regressions(record, baseline_records, metric_policies)
if static_threshold_failed:
failures.append(f"{record['model_id']} fixed-threshold phase failed "
f"(PERF_PYTEST_RC={os.environ.get('PERF_PYTEST_RC')})")
@@ -434,10 +451,10 @@ def main() -> int:
print("Tracking upload skipped for "
f"{record['model_id']} ({record['run_source']}, success={record['success']})")
summary_rows.append(_build_summary_row(record, baseline_records, bool(failures)))
summary_rows.append(_build_summary_row(record, baseline_records, metric_policies, bool(failures)))
commit_sha = os.environ.get("BUILDKITE_COMMIT", "unknown")[:7]
markdown = _build_markdown_summary(summary_rows, MAX_REGRESSION)
markdown = _build_markdown_summary(summary_rows, resolve_metric_policies(None))
_emit_markdown_summary(markdown, commit_sha)
if all_failures:
+8 -1
View File
@@ -1,12 +1,19 @@
# SPDX-License-Identifier: Apache-2.0
import os
import sys
from html import escape
from datetime import datetime
import plotly.express as px
import pandas as pd
from hf_store import sync_from_hf, load_as_dataframe
try:
from fastvideo.performance.hf_store import load_as_dataframe, sync_from_hf
except ImportError:
repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
if repo_root not in sys.path:
sys.path.insert(0, repo_root)
from fastvideo.performance.hf_store import load_as_dataframe, sync_from_hf
TRACKING_ROOT = os.environ.get(
"PERFORMANCE_TRACKING_ROOT",
@@ -1,6 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
from fastvideo.tests.performance import compare_baseline
from fastvideo.performance.metric_policy import resolve_metric_policies
def _raw_result():
@@ -72,9 +73,157 @@ def test_normalized_record_includes_source_metadata(monkeypatch):
assert record["job_id"] == "job-1"
def test_normalized_record_includes_effective_regression_thresholds(monkeypatch):
monkeypatch.setenv("PERF_RUN_SOURCE", "pr")
raw = _raw_result()
raw["regression_thresholds"] = {
"latency": {
"threshold_percent": 0.09,
"threshold_absolute": 0.75,
"gated": True,
},
"throughput": {
"gated": False,
},
}
record = compare_baseline.normalize_performance_result(raw)
assert record["regression_thresholds"]["latency"] == {
"threshold_percent": 0.09,
"threshold_absolute": 0.75,
"gated": True,
}
assert record["regression_thresholds"]["throughput"]["gated"] is False
def test_invalid_regression_threshold_container_uses_defaults():
policies = resolve_metric_policies(["not", "a", "mapping"])
latency = next(policy for policy in policies if policy.key == "latency")
assert latency.threshold_percent == 0.08
assert latency.threshold_absolute == 0.5
assert latency.gated is True
def test_boolean_regression_threshold_values_are_ignored():
policies = resolve_metric_policies({
"latency": {
"threshold_percent": True,
"threshold_absolute": False,
"gated": "false",
}
})
latency = next(policy for policy in policies if policy.key == "latency")
assert latency.threshold_percent == 0.08
assert latency.threshold_absolute == 0.5
assert latency.gated is False
def test_baseline_eligibility_only_for_successful_scheduled_main():
assert compare_baseline._is_baseline_eligible("scheduled_main", True) is True
assert compare_baseline._is_baseline_eligible("scheduled_main", False) is False
assert compare_baseline._is_baseline_eligible("pr", True) is False
assert compare_baseline._is_baseline_eligible("local", True) is False
def test_latency_regression_requires_percent_and_absolute_floors():
baseline = [{"latency": 10.0}]
current = {"model_id": "wan", "latency": 10.6}
percent_only = resolve_metric_policies({
"latency": {
"threshold_percent": 0.05,
"threshold_absolute": 0.75,
}
})
absolute_only = resolve_metric_policies({
"latency": {
"threshold_percent": 0.10,
"threshold_absolute": 0.5,
}
})
both = resolve_metric_policies({
"latency": {
"threshold_percent": 0.05,
"threshold_absolute": 0.5,
}
})
assert compare_baseline._check_regressions(current, baseline, percent_only) == []
assert compare_baseline._check_regressions(current, baseline, absolute_only) == []
failures = compare_baseline._check_regressions(current, baseline, both)
assert len(failures) == 1
assert "latency regressed by 6.0% and 0.600" in failures[0]
def test_throughput_regression_uses_higher_is_better_direction():
baseline = [{"throughput": 10.0}]
current = {"model_id": "wan", "throughput": 9.0}
policies = resolve_metric_policies({
"throughput": {
"threshold_percent": 0.05,
"threshold_absolute": 0.5,
}
})
failures = compare_baseline._check_regressions(current, baseline, policies)
assert len(failures) == 1
assert "throughput regressed by 10.0% and 1.000" in failures[0]
def test_memory_regression_uses_metric_specific_absolute_floor():
baseline = [{"memory": 10000.0}]
current = {"model_id": "wan", "memory": 10600.0}
policies = resolve_metric_policies({
"memory": {
"threshold_percent": 0.05,
"threshold_absolute": 256.0,
}
})
failures = compare_baseline._check_regressions(current, baseline, policies)
assert len(failures) == 1
assert "memory regressed by 6.0% and 600.0" in failures[0]
def test_component_metric_can_gate_independently():
baseline = [{"dit_time_s": 8.0}]
current = {"model_id": "wan", "dit_time_s": 8.6}
policies = resolve_metric_policies({
"dit_time_s": {
"threshold_percent": 0.05,
"threshold_absolute": 0.25,
}
})
failures = compare_baseline._check_regressions(current, baseline, policies)
assert len(failures) == 1
assert "dit_time_s regressed by 7.5% and 0.600" in failures[0]
def test_informational_metric_remains_visible_without_failing():
baseline = [{"throughput": 10.0}]
current = {"model_id": "wan", "gpu_type": "NVIDIA L40S", "throughput": 8.0}
policies = resolve_metric_policies({
"throughput": {
"threshold_percent": 0.01,
"threshold_absolute": 0.01,
"gated": False,
}
})
row = compare_baseline._build_summary_row(current, baseline, policies, False)
assert compare_baseline._check_regressions(current, baseline, policies) == []
assert row["metrics"]["throughput"]["regression_pct"] == 20.0
assert row["metrics"]["throughput"]["gated"] is False
assert row["metrics"]["throughput"]["threshold_exceeded"] is True
assert row["metrics"]["throughput"]["regressed"] is False
assert row["threshold_exceeded_metrics"] == ["throughput"]
assert row["failing_metrics"] == []
@@ -68,6 +68,8 @@ def test_summary_endpoint_returns_latest_group_status():
assert body["count"] == 1
assert body["status_counts"] == {"pass": 1, "fail": 0}
assert body["rows"][0]["metrics"]["latency"]["baseline"] == 10.0
assert body["rows"][0]["metrics"]["latency"]["threshold_exceeded"] is True
assert body["rows"][0]["threshold_exceeded_metrics"] == ["latency", "throughput"]
assert body["rows"][0]["computed_regression_status"] == "fail"
@@ -1,6 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
from fastvideo.performance import hf_store
from fastvideo.performance_dashboard.service import build_latest_summary, build_trends, filter_records
from fastvideo.tests.performance import hf_store
def _record(ts, commit, latency, throughput, success=True, **metadata):
@@ -28,16 +28,23 @@ def test_build_latest_summary_uses_previous_successful_records_for_baseline():
_record("2026-01-03T00:00:00+00:00", "c" * 40, 11.0, 9.0),
]
rows = build_latest_summary(records, max_regression=0.05)
rows = build_latest_summary(records)
assert len(rows) == 1
row = rows[0]
assert row["baseline_n"] == 1
assert row["metrics"]["latency"]["baseline"] == 10.0
assert row["metrics"]["latency"]["regression_pct"] == 10.0
assert row["metrics"]["latency"]["absolute_delta"] == 1.0
assert row["metrics"]["latency"]["threshold_percent"] == 8.0
assert row["metrics"]["latency"]["threshold_absolute"] == 0.5
assert row["metrics"]["latency"]["threshold_exceeded"] is True
assert row["metrics"]["latency"]["regressed"] is True
assert row["metrics"]["throughput"]["regression_pct"] == 10.0
assert row["status"] == "pass"
assert row["computed_regression_status"] == "fail"
assert row["threshold_exceeded_metrics"] == ["latency", "throughput"]
assert row["failing_metrics"] == ["latency", "throughput"]
def test_build_latest_summary_status_uses_latest_record_success_field():
@@ -73,7 +80,7 @@ def test_build_latest_summary_run_source_filter_keeps_canonical_baseline():
),
]
rows = build_latest_summary(records, max_regression=0.05, run_source="pr")
rows = build_latest_summary(records, run_source="pr")
assert len(rows) == 1
assert rows[0]["run_source"] == "pr"
@@ -83,6 +90,60 @@ def test_build_latest_summary_run_source_filter_keeps_canonical_baseline():
assert rows[0]["computed_regression_status"] == "fail"
def test_build_latest_summary_requires_absolute_floor_for_computed_regression():
records = [
_record("2026-01-01T00:00:00+00:00", "a" * 40, 10.0, 10.0),
_record(
"2026-01-02T00:00:00+00:00",
"b" * 40,
10.6,
10.0,
regression_thresholds={
"latency": {
"threshold_percent": 0.05,
"threshold_absolute": 0.75,
"gated": True,
}
},
),
]
rows = build_latest_summary(records)
assert round(rows[0]["metrics"]["latency"]["regression_pct"], 1) == 6.0
assert round(rows[0]["metrics"]["latency"]["absolute_delta"], 3) == 0.6
assert rows[0]["metrics"]["latency"]["threshold_exceeded"] is False
assert rows[0]["metrics"]["latency"]["regressed"] is False
assert rows[0]["computed_regression_status"] == "pass"
def test_build_latest_summary_separates_informational_threshold_crossing():
records = [
_record("2026-01-01T00:00:00+00:00", "a" * 40, 10.0, 10.0),
_record(
"2026-01-02T00:00:00+00:00",
"b" * 40,
10.6,
10.0,
regression_thresholds={
"latency": {
"threshold_percent": 0.05,
"threshold_absolute": 0.5,
"gated": False,
}
},
),
]
rows = build_latest_summary(records)
assert rows[0]["metrics"]["latency"]["threshold_exceeded"] is True
assert rows[0]["metrics"]["latency"]["regressed"] is False
assert rows[0]["threshold_exceeded_metrics"] == ["latency"]
assert rows[0]["failing_metrics"] == []
assert rows[0]["computed_regression_status"] == "pass"
def test_filter_records_and_trends_preserve_metric_points():
records = [
_record("2026-01-01T00:00:00+00:00", "a" * 40, 10.0, 10.0),
@@ -179,7 +179,11 @@ def _extract_component_times(result: dict) -> dict[str, float | None]:
logger.debug("Skipping malformed stage '%s' data: %r", stage_name, stage_data)
continue
stage_class = stage_data.get("stage_class", stage_name)
metric_key = STAGE_METRIC_MAP.get(stage_class)
component_metric = stage_data.get("component_metric")
if isinstance(component_metric, str) and component_metric in component_times:
metric_key = component_metric
else:
metric_key = STAGE_METRIC_MAP.get(stage_class)
if metric_key is None:
logger.debug("Unmapped stage '%s' class '%s' (%.3fs)",
stage_name,
@@ -309,6 +313,7 @@ def _run_benchmark(cfg):
"max_peak_memory_mb": round(max_peak_memory, 1),
"individual_peak_memories_mb": [round(m, 1) for m in peak_memories],
"thresholds": thresholds,
"regression_thresholds": cfg.get("regression_thresholds", {}),
"commit": os.environ.get("BUILDKITE_COMMIT", ""),
"pr_number": os.environ.get("BUILDKITE_PULL_REQUEST", ""),
"timestamp": datetime.now(timezone.utc).isoformat(),
@@ -1,13 +1,20 @@
# SPDX-License-Identifier: Apache-2.0
from fastvideo.pipelines.pipeline_batch_info import PipelineLoggingInfo
from fastvideo.pipelines.stages.denoising import Cosmos25AutoDenoisingStage, DenoisingStage
from fastvideo.pipelines.stages.text_encoding import Cosmos25TextEncodingStage
from fastvideo.tests.performance.test_inference_performance import _extract_component_times
class SubclassStyleDenoisingStage(DenoisingStage):
pass
def test_extract_component_times_handles_pipeline_logging_info_object():
logging_info = PipelineLoggingInfo()
logging_info.add_stage_execution_time("prompt_encoding_stage", 1.25)
logging_info.add_stage_metric("prompt_encoding_stage", "stage_class", "TextEncodingStage")
logging_info.add_stage_metric("prompt_encoding_stage", "component_metric", "text_encoder_time_s")
assert _extract_component_times({"logging_info": logging_info}) == {
"text_encoder_time_s": 1.25,
@@ -45,6 +52,35 @@ def test_extract_component_times_uses_stage_class_for_pipeline_stage_keys():
}
def test_denoising_stage_subclasses_inherit_component_metric():
assert SubclassStyleDenoisingStage.performance_component_metric == "dit_time_s"
def test_cosmos25_direct_pipeline_stages_define_component_metrics():
assert Cosmos25TextEncodingStage.performance_component_metric == "text_encoder_time_s"
assert Cosmos25AutoDenoisingStage.performance_component_metric == "dit_time_s"
def test_extract_component_times_uses_component_metric_for_stage_subclasses():
result = {
"logging_info": {
"stages": {
"denoising_stage": {
"execution_time": 4.2,
"stage_class": "CosmosDenoisingStage",
"component_metric": "dit_time_s",
},
},
},
}
assert _extract_component_times(result) == {
"text_encoder_time_s": None,
"dit_time_s": 4.2,
"vae_decode_time_s": None,
}
def test_extract_component_times_keeps_legacy_class_name_keys():
# Backward-compatibility check for logs produced before pipeline-unique
# stage keys carried a separate stage_class field.
@@ -20,22 +20,13 @@ REQUIRED_GPUS = 1
device_reference_folder = resolve_inference_device_reference_folder(logger)
_LOCAL_CONVERTED_MODEL = Path("converted_weights/dreamx_world")
_MODEL_PATH = os.getenv(
"DREAMX_WORLD_SSIM_MODEL_PATH",
str(_LOCAL_CONVERTED_MODEL),
)
_LOCAL_AR_CANDIDATES = (
Path("/tmp/converted_dreamx_world_ar"),
Path("/root/data/dreamx_world_ar_converted"),
)
_DEFAULT_AR_MODEL_PATH = next(
(str(path) for path in _LOCAL_AR_CANDIDATES if path.exists()),
str(_LOCAL_AR_CANDIDATES[0]),
"FastVideo/DreamX-World-5B-Cam-Diffusers",
)
_AR_MODEL_PATH = os.getenv(
"DREAMX_WORLD_AR_SSIM_MODEL_PATH",
_DEFAULT_AR_MODEL_PATH,
"FastVideo/DreamX-World-5B-Diffusers",
)
DREAMX_WORLD_PARAMS = {
@@ -137,13 +128,6 @@ def test_dreamx_world_inference_similarity(
model_id: str,
tmp_path: Path,
) -> None:
model_path = Path(str(DREAMX_WORLD_MODEL_TO_PARAMS[model_id]["model_path"]))
if not model_path.exists():
pytest.skip(
f"DreamX-World converted model path is missing: {model_path}. "
"Set DREAMX_WORLD_SSIM_MODEL_PATH to a FastVideo-loadable converted root."
)
image_path = tmp_path / "dreamx_world_ssim_input.png"
_write_deterministic_reference_image(image_path)
@@ -182,13 +166,6 @@ def test_dreamx_world_ar_inference_similarity(
attention_backend_name: str,
model_id: str,
) -> None:
model_path = Path(str(DREAMX_WORLD_AR_MODEL_TO_PARAMS[model_id]["model_path"]))
if not model_path.exists():
pytest.skip(
f"DreamX-World AR converted model path is missing: {model_path}. "
"Set DREAMX_WORLD_AR_SSIM_MODEL_PATH to a FastVideo-loadable converted root."
)
run_text_to_video_similarity_test(
logger=logger,
script_dir=os.path.dirname(os.path.abspath(__file__)),
@@ -21,6 +21,26 @@ class FakeTokenizer:
"attention_mask": torch.ones(B, seq_len, dtype=torch.long),
})
class FakeChatTokenizer:
def __init__(self):
self.last_messages = None
self.last_kwargs = None
def apply_chat_template(self, messages, **kwargs):
self.last_messages = messages
self.last_kwargs = kwargs
assert isinstance(messages[0], list)
assert messages[0][0]["role"] == "system"
assert messages[0][1]["role"] == "user"
B = len(messages)
seq_len = int(kwargs.get("max_length", 4))
return TensorDict({
"input_ids": torch.arange(B * seq_len).view(B, seq_len),
"attention_mask": torch.ones(B, seq_len, dtype=torch.long),
})
class FakeTextEncoder(torch.nn.Module):
def __init__(self, hidden_size=8):
super().__init__()
@@ -38,6 +58,14 @@ class FakeTextEncoder(torch.nn.Module):
def id_preprocess(x: str) -> str:
return x
def chat_list_preprocess(x: str):
return [
{"role": "system", "content": "Describe the video."},
{"role": "user", "content": x if x else " "},
]
def take_mean_postprocess(outputs: BaseEncoderOutput) -> torch.Tensor:
# [B, T, H] -> [B, H]
return outputs.last_hidden_state.mean(dim=1)
@@ -156,3 +184,32 @@ def test_encode_text_does_not_force_hidden_states_for_ltx2_prefix():
stage.encode_text("a", fastvideo_args, encoder_index=[0])
assert stage.text_encoders[0].last_output_hidden_states is False
def test_chat_list_preprocess_output_is_not_stripped():
fastvideo_args, hidden = make_args(num_encoders=1, text_len=5, hidden_size=8)
encoder_config = fastvideo_args.pipeline_config.text_encoder_configs[0]
encoder_config.is_chat_model = True
encoder_config.treat_empty_as_dot = True
fastvideo_args.pipeline_config.preprocess_text_funcs = (chat_list_preprocess, )
tokenizer = FakeChatTokenizer()
stage = TextEncodingStage(
text_encoders=[FakeTextEncoder(hidden_size=hidden)],
tokenizers=[tokenizer],
)
embeds, masks = stage.encode_text(
"a robotic arm welding a metal structure",
fastvideo_args,
encoder_index=[0],
return_attention_mask=True,
)
assert embeds[0].shape == (1, hidden)
assert masks[0].shape == (1, 5)
assert tokenizer.last_messages == [[
{"role": "system", "content": "Describe the video."},
{"role": "user", "content": "a robotic arm welding a metal structure"},
]]
assert tokenizer.last_kwargs["return_tensors"] == "pt"
@@ -1,104 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
import glob
import json
import os
import pytest
import torch
from fastvideo.configs.pipelines.base import PipelineConfig
from fastvideo.distributed.parallel_state import get_sp_parallel_rank, get_sp_world_size
from fastvideo.logger import init_logger
from fastvideo.models.loader.component_loader import TransformerLoader
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.utils import maybe_download_model
from fastvideo.forward_context import set_forward_context
from fastvideo.configs.models.dits import HunyuanVideoConfig
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
logger = init_logger(__name__)
os.environ["MASTER_ADDR"] = "localhost"
os.environ["MASTER_PORT"] = "29503"
BASE_MODEL_PATH = "hunyuanvideo-community/HunyuanVideo"
MODEL_PATH = maybe_download_model(BASE_MODEL_PATH, local_dir=os.path.join("data", BASE_MODEL_PATH))
TRANSFORMER_PATH = os.path.join(MODEL_PATH, "transformer")
CONFIG_PATH = os.path.join(TRANSFORMER_PATH, "config.json")
LOCAL_RANK = 0
RANK = 0
WORLD_SIZE = 1
# Latent generated on commit c021e8a27cf437ac22827f2bc58b7f006561317f with 1 x L40S
REFERENCE_LATENT = 93.23263549804688
@pytest.mark.usefixtures("distributed_setup")
@pytest.mark.skip(reason="Skipping test_hunyuanvideo_distributed")
def test_hunyuanvideo_distributed():
logger.info(f"Initializing process: rank={RANK}, local_rank={LOCAL_RANK}, world_size={WORLD_SIZE}")
torch.cuda.set_device(f"cuda:{LOCAL_RANK}")
# Get tensor parallel info
sp_rank = get_sp_parallel_rank()
sp_world_size = get_sp_world_size()
logger.info(f"Process rank {RANK} initialized with SP rank {sp_rank} in SP world size {sp_world_size}")
config = json.load(open(CONFIG_PATH))
# remove "_class_name": "HunyuanVideoTransformer3DModel", "_diffusers_version": "0.32.0.dev0",
# TODO: write normalize config function
config.pop("_class_name")
config.pop("_diffusers_version")
precision_str = "bf16"
args = FastVideoArgs(
model_path=TRANSFORMER_PATH,
dit_cpu_offload=True,
pipeline_config=PipelineConfig(dit_config=HunyuanVideoConfig(), dit_precision=precision_str),
)
args.device = torch.device(f"cuda:{LOCAL_RANK}")
loader = TransformerLoader()
model = loader.load(TRANSFORMER_PATH, args)
model.eval()
# Create random inputs for testing
batch_size = 1
seq_len = 3
device = torch.device(f"cuda:{LOCAL_RANK}")
# Video latents [B, C, T, H, W]
hidden_states = torch.randn(batch_size, 16, 8, 16, 16, device=device, dtype=torch.bfloat16)
chunk_per_rank = hidden_states.shape[2] // sp_world_size
hidden_states = hidden_states[:, :, sp_rank * chunk_per_rank : (sp_rank + 1) * chunk_per_rank]
# Text embeddings [B, L, D] (including global token)
encoder_hidden_states = torch.randn(batch_size, seq_len + 1, 4096, device=device, dtype=torch.bfloat16)
# Timestep
timestep = torch.tensor([500], device=device, dtype=torch.bfloat16)
forward_batch = ForwardBatch(
data_type="dummy",
)
# Disable gradients for inference
with torch.no_grad():
# Run inference on model
with torch.amp.autocast(device_type="cuda", dtype=torch.bfloat16):
with set_forward_context(current_timestep=0, attn_metadata=None, forward_batch=forward_batch):
output = model(
hidden_states=hidden_states,
encoder_hidden_states=encoder_hidden_states,
timestep=timestep,
)
latent = output.double().sum().item()
# Check if latents are similar
diff_output_latents = abs(REFERENCE_LATENT - latent)
logger.info(f"Reference latent: {REFERENCE_LATENT}, Current latent: {latent}")
assert diff_output_latents < 1e-4, f"Output latents differ significantly: max diff = {diff_output_latents}"
@@ -1,133 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
import os
import pytest
import torch
from fastvideo.configs.models.dits.matrixgame2 import MatrixGame2WanVideoConfig
from fastvideo.configs.pipelines import PipelineConfig
from fastvideo.distributed.parallel_state import (
get_sp_parallel_rank,
get_sp_world_size,
)
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.forward_context import set_forward_context
from fastvideo.logger import init_logger
from fastvideo.models.loader.component_loader import TransformerLoader
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.utils import maybe_download_model
logger = init_logger(__name__)
os.environ["MASTER_ADDR"] = "localhost"
os.environ["MASTER_PORT"] = "29505"
os.environ.setdefault("TORCHDYNAMO_DISABLE", "1")
MODEL_PATH = maybe_download_model("FastVideo/Matrix-Game-2.0-Base-Distilled-Diffusers")
TRANSFORMER_PATH = os.path.join(MODEL_PATH, "transformer")
REFERENCE_LATENT = -111835.58895772696
@pytest.mark.skip(reason="Not reliably reproducible")
@pytest.mark.usefixtures("distributed_setup")
def test_matrixgame2_transformer():
transformer_path = TRANSFORMER_PATH
sp_rank = get_sp_parallel_rank()
sp_world_size = get_sp_world_size()
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
precision = torch.bfloat16
precision_str = "bf16"
args = FastVideoArgs(
model_path=transformer_path,
dit_cpu_offload=False,
use_fsdp_inference=False,
pipeline_config=PipelineConfig(
dit_config=MatrixGame2WanVideoConfig(), dit_precision=precision_str
),
)
args.device = device
loader = TransformerLoader()
model = loader.load(transformer_path, args).to(device, dtype=precision)
model.eval()
total_params = sum(p.numel() for p in model.parameters())
weight_sum = sum(p.to(torch.float64).sum().item() for p in model.parameters())
weight_mean = weight_sum / total_params
logger.info("Total parameters: %s", total_params)
logger.info("Weight sum: %s", weight_sum)
logger.info("Weight mean: %s", weight_mean)
torch.manual_seed(42)
batch_size = 1
latent_frames = 15
latent_height = 44
latent_width = 80
image_dim = 1280
image_seq_len = 257
# Set num_frame_per_block to match actual latent_frames for action_model
model.num_frame_per_block = latent_frames
# Force full attention mode
model.local_attn_size = -1
# Clear cached block masks so they regenerate with correct settings
model.block_mask = None
model.block_mask_mouse = None
model.block_mask_keyboard = None
# Video latents [B, C, T, H, W]
x = torch.randn(batch_size, 16, latent_frames, latent_height, latent_width,
device=device, dtype=precision)
cond_concat = torch.randn(batch_size, 20, latent_frames, latent_height, latent_width,
device=device, dtype=precision)
hidden_states = torch.cat([x, cond_concat], dim=1)
if sp_world_size > 1:
chunk_per_rank = hidden_states.shape[2] // sp_world_size
hidden_states = hidden_states[:, :, sp_rank * chunk_per_rank:(sp_rank + 1) * chunk_per_rank]
# Image embeddings
encoder_hidden_states_image = [
torch.randn(batch_size, image_seq_len, image_dim, device=device, dtype=precision)
]
# Timestep
timestep = torch.full((batch_size, latent_frames), 500, device=device, dtype=precision)
# Action conditions
num_pixel_frames = (latent_frames - 1) * 4 + 1
mouse_cond = torch.randn(batch_size, num_pixel_frames, 2, device=device, dtype=precision)
keyboard_cond = torch.randn(batch_size, num_pixel_frames, 4, device=device, dtype=precision)
# Text embeddings (empty for i2v)
encoder_hidden_states = torch.zeros(batch_size, 0, model.config.hidden_size,
device=device, dtype=precision)
forward_batch = ForwardBatch(data_type="dummy")
model.num_frame_per_block = latent_frames
with torch.no_grad():
with torch.amp.autocast('cuda', dtype=precision):
with set_forward_context(current_timestep=0, attn_metadata=None, forward_batch=forward_batch):
output = model(
hidden_states=hidden_states,
encoder_hidden_states=encoder_hidden_states,
encoder_hidden_states_image=encoder_hidden_states_image,
timestep=timestep,
mouse_cond=mouse_cond,
keyboard_cond=keyboard_cond,
)
latent = output.double().sum().item()
diff = abs(REFERENCE_LATENT - latent)
relative_diff = diff / abs(REFERENCE_LATENT)
logger.info(f"Reference latent: {REFERENCE_LATENT}, Current latent: {latent}")
logger.info(f"Absolute diff: {diff}, Relative diff: {relative_diff * 100:.4f}%")
# The 0.19% diff now
assert relative_diff < 0.005, f"Output latents differ significantly: relative diff = {relative_diff * 100:.4f}% (max allowed: 0.5%)"
-82
View File
@@ -1,82 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
import json
import os
import pytest
import torch
from safetensors.torch import load_file
from fastvideo.configs.pipelines import PipelineConfig
from fastvideo.logger import init_logger
# from fastvideo.models.vaes.hunyuanvae import (
# AutoencoderKLHunyuanVideo as MyHunyuanVAE)
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.models.loader.component_loader import VAELoader
from fastvideo.configs.models.vaes import HunyuanVAEConfig
from fastvideo.utils import maybe_download_model
logger = init_logger(__name__)
os.environ["MASTER_ADDR"] = "localhost"
os.environ["MASTER_PORT"] = "29503"
BASE_MODEL_PATH = "hunyuanvideo-community/HunyuanVideo"
MODEL_PATH = maybe_download_model(BASE_MODEL_PATH, local_dir=os.path.join("data", BASE_MODEL_PATH))
VAE_PATH = os.path.join(MODEL_PATH, "vae")
CONFIG_PATH = os.path.join(VAE_PATH, "config.json")
# Latent generated on commit d71a4ebffc2034922fc379568b6a6aa722f3744c with 1 x A40
# torch 2.7.1
A40_REFERENCE_LATENT = -106.22467041015625
# Latent generated on commit 2b54068960c41d42221e8b8719a374b499855029 with 1 x L40S
L40S_REFERENCE_LATENT = -157.09130859375
@pytest.mark.usefixtures("distributed_setup")
@pytest.mark.skip(reason="Skipping test_hunyuan_vae")
def test_hunyuan_vae():
device = torch.device("cuda:0")
precision = torch.bfloat16
precision_str = "bf16"
args = FastVideoArgs(
model_path=VAE_PATH, pipeline_config=PipelineConfig(vae_config=HunyuanVAEConfig(), vae_precision=precision_str)
)
args.device = device
args.vae_cpu_offload = False
loader = VAELoader()
model = loader.load(VAE_PATH, args)
model.enable_tiling(
tile_sample_min_height=32,
tile_sample_min_width=32,
tile_sample_min_num_frames=8,
tile_sample_stride_height=16,
tile_sample_stride_width=16,
tile_sample_stride_num_frames=4,
)
batch_size = 1
# Video input [B, C, T, H, W]
input_tensor = torch.randn(batch_size, 3, 21, 64, 64, device=device, dtype=torch.bfloat16)
# Disable gradients for inference
with torch.no_grad():
latent = model.encode(input_tensor).mean.double().sum().item()
# Check if latents are similar
device_name = torch.cuda.get_device_name()
if "A40" in device_name:
REFERENCE_LATENT = A40_REFERENCE_LATENT
elif "L40S" in device_name:
REFERENCE_LATENT = L40S_REFERENCE_LATENT
else:
raise ValueError(f"Unknown device: {device_name}")
diff_encoded_latents = abs(REFERENCE_LATENT - latent)
logger.info(f"Reference latent: {REFERENCE_LATENT}, Current latent: {latent}")
assert diff_encoded_latents < 1e-4, f"Encoded latents differ significantly: max diff = {diff_encoded_latents}"
@@ -1,113 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
import os
import numpy as np
import pytest
import torch
from diffusers import AutoencoderKLWan
from fastvideo.configs.pipelines import PipelineConfig
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.logger import init_logger
from fastvideo.models.loader.component_loader import VAELoader
from fastvideo.configs.models.vaes import WanVAEConfig
from fastvideo.utils import maybe_download_model
try:
from wan.modules.vae2_2 import Wan2_2_VAE
except ImportError:
Wan2_2_VAE = None
logger = init_logger(__name__)
os.environ["MASTER_ADDR"] = "localhost"
os.environ["MASTER_PORT"] = "29503"
BASE_MODEL_PATH = "Wan-AI/Wan2.2-TI2V-5B-Diffusers"
MODEL_PATH = maybe_download_model(BASE_MODEL_PATH)
VAE_PATH = os.path.join(MODEL_PATH, "vae")
@pytest.mark.skip(reason="disable test")
@pytest.mark.usefixtures("distributed_setup")
def test_official_wan_vae():
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
precision = torch.bfloat16
precision_str = "bf16"
args = FastVideoArgs(model_path=VAE_PATH, pipeline_config=PipelineConfig(vae_config=WanVAEConfig(), vae_precision=precision_str))
args.device = device
args.vae_cpu_offload = False
loader = VAELoader()
model2 = loader.load(VAE_PATH, args)
assert model2.use_feature_cache # Default to use the original WanVAE algorithm
model1 = Wan2_2_VAE(vae_pth="/mnt/weka/home/hao.zhang/wei/Wan2.2_VAE.pth", device=device, dtype=precision)
# Create identical inputs for both models
batch_size = 1
# Video input [B, C, T, H, W]
input_tensor = torch.randn(batch_size,
3,
81,
32,
32,
device=device,
dtype=precision)
# latent_tensor = torch.randn(batch_size,
# 16,
# 21,
# 32,
# 32,
# device=device,
# dtype=precision)
# Disable gradients for inference
with torch.no_grad():
# Test encoding
logger.info("Testing encoding...")
latent1 = model1.encode([input_tensor.squeeze(0)])[0].unsqueeze(0)
print("--------------------------------")
latent2 = model2.encode(input_tensor)
# Check if latents have the same shape
assert latent1.shape == latent2.mean.shape, f"Latent shapes don't match: {latent1.mean.shape} vs {latent2.mean.shape}"
# Check if latents are similar
max_diff_encode = torch.max(torch.abs(latent1 - latent2.mean))
mean_diff_encode = torch.mean(torch.abs(latent1 - latent2.mean))
logger.info("Maximum difference between encoded latents: %s",
max_diff_encode.item())
logger.info("Mean difference between encoded latents: %s",
mean_diff_encode.item())
assert max_diff_encode < 1e-5, f"Encoded latents differ significantly: max diff = {mean_diff_encode.item()}"
# Test decoding
logger.info("Testing decoding...")
# latent1_tensor = latent1.mode()
# mean1 = (torch.tensor(model1.config.latents_mean).view(
# 1, model1.config.z_dim, 1, 1, 1).to(input_tensor.device,
# input_tensor.dtype))
# std1 = (1.0 / torch.tensor(model1.config.latents_std).view(
# 1, model1.config.z_dim, 1, 1, 1)).to(input_tensor.device,
# input_tensor.dtype)
# latent1_tensor = latent1_tensor / std1 + mean1
# output1 = model1.decode(latent1_tensor).sample
output1 = model1.decode([latent1.squeeze(0)])
output1 = output1[0].unsqueeze(0)
mean2 = model2.config.arch_config.shift_factor.to(input_tensor.device, input_tensor.dtype)
std2 = model2.config.arch_config.scaling_factor.to(input_tensor.device, input_tensor.dtype)
latent2_tensor = latent2.mode()
latent2_tensor = latent2_tensor / std2 + mean2
output2 = model2.decode(latent2_tensor)
# Check if outputs have the same shape
assert output1.shape == output2.shape, f"Output shapes don't match: {output1.shape} vs {output2.shape}"
# Check if outputs are similar
max_diff_decode = torch.max(torch.abs(output1.float() - output2.float()))
mean_diff_decode = torch.mean(torch.abs(output1.float() - output2.float()))
logger.info("Maximum difference between decoded outputs: %s",
max_diff_decode.item())
logger.info("Mean difference between decoded outputs: %s",
mean_diff_decode.item())
assert max_diff_decode < 1e-5, f"Decoded outputs differ significantly: max diff = {max_diff_decode.item()}"
+1 -1
View File
@@ -114,7 +114,7 @@ fastvideo-kernel = [
imagebind = { git = "https://github.com/facebookresearch/ImageBind.git", rev = "53680b02d7e37b19b124fa37bae4b6c98c38f5be" }
# FA4 cute, pinned to a cutlass-4.5-compatible revision. torch.compile support
# comes from FastVideo's own custom_op wrappers.
flash-attn-4 = { git = "https://github.com/Dao-AILab/flash-attention.git", rev = "940cd9680f3315f2f06b43ab5bea2c2cf2d96806", subdirectory = "flash_attn/cute" }
flash-attn-4 = { git = "https://github.com/Dao-AILab/flash-attention.git", rev = "82d6441eec5d4dfec120153db2c0145ae855a083", subdirectory = "flash_attn/cute" }
[project.optional-dependencies]
+13 -10
View File
@@ -10,15 +10,6 @@ import pytest
os.environ["CUTE_DSL_ENABLE_TVM_FFI"] = "1"
# Warm up flashinfer JIT before any FA4 imports
from flashinfer.quantization import nvfp4_quantize, SfLayout
_warmup = nvfp4_quantize(
torch.randn(4, 128, device="cuda", dtype=torch.bfloat16),
torch.ones(1, device="cuda", dtype=torch.float32),
sfLayout=SfLayout.layout_128x4, do_shuffle=False,
)
del _warmup
def _make_impl(nheads, headdim, nvfp4=False):
from fastvideo.attention.backends.flash_attn import FlashAttentionImpl
@@ -45,7 +36,8 @@ def _cuda_timer(fn, warmup=5, iters=20):
@pytest.mark.skipif(
torch.cuda.get_device_capability() not in [(10, 0), (10, 3)],
not torch.cuda.is_available()
or torch.cuda.get_device_capability() not in [(10, 0), (10, 3)],
reason="Requires Blackwell GPU (sm100a or sm103a)"
)
class TestNVFP4FA4:
@@ -55,6 +47,17 @@ class TestNVFP4FA4:
MODEL_HEADDIM = 128
MODEL_SEQLEN = 32760 # 480x832 video, 81 frames
@pytest.fixture(scope="class", autouse=True)
def _flashinfer_warmup(self):
# Warm up flashinfer JIT before any FA4 imports
from flashinfer.quantization import SfLayout, nvfp4_quantize
_warmup = nvfp4_quantize(
torch.randn(4, 128, device="cuda", dtype=torch.bfloat16),
torch.ones(1, device="cuda", dtype=torch.float32),
sfLayout=SfLayout.layout_128x4, do_shuffle=False,
)
del _warmup
def test_quantize_fn_shapes(self):
"""_nvfp4_quantize_for_fa4 produces correct shapes and strides."""
from fastvideo.attention.backends.flash_attn import _nvfp4_quantize_for_fa4