1023 lines
39 KiB
Python
1023 lines
39 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
|
|
|
"""Speed + memory lever benchmark for the VIDEO diffusion backend (B200).
|
|
|
|
Drives the SAME lever functions the loader calls (``quantize_text_encoders`` / ``quantize_vae``
|
|
/ ``quantize_transformer`` / ``apply_speed_optims`` / ``apply_attention_backend`` /
|
|
``apply_step_cache`` + ``maybe_toggle_step_cache``) with the loader's own default arguments, so
|
|
each configuration reflects a real load.
|
|
|
|
Per config it loads the pipeline fresh (quant/compile mutate irreversibly), warms up, then
|
|
measures a short clip: total latency, median per-step ms, peak resident GB, steady weight GB, and
|
|
per-frame LPIPS(AlexNet) vs the bit-exact reference (everything off/dense/native). Isolates each
|
|
lever's contribution and whether ``max`` compile / flash4 leave speed on the table.
|
|
|
|
Memory/timing idiom lifted from scripts/quant_speedmem_bench.py.
|
|
|
|
Example:
|
|
CUDA_VISIBLE_DEVICES=0 python scripts/video_speedmem_bench.py --family wan2.2-ti2v-5b \\
|
|
--configs reference,compile,cudnn,fbcache,ditquant,shipped,speedmax,flash4 \\
|
|
--steps 30 --num-frames 25 --width 512 --height 320 --iters 3
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import gc
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import types
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
os.environ.setdefault("BITSANDBYTES_NOWELCOME", "1")
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
_BACKEND_ROOT = _REPO_ROOT / "studio" / "backend"
|
|
for _p in (str(_BACKEND_ROOT), str(_REPO_ROOT / "scripts")):
|
|
if _p not in sys.path:
|
|
sys.path.insert(0, _p)
|
|
|
|
PROMPT = (
|
|
"A cinematic drone shot flying over a misty mountain valley at sunrise, "
|
|
"golden light, volumetric fog, highly detailed, smooth camera motion"
|
|
)
|
|
|
|
_FAMILIES: dict[str, dict[str, Any]] = {
|
|
"wan2.2-ti2v-5b": {
|
|
"repo": "Wan-AI/Wan2.2-TI2V-5B-Diffusers",
|
|
"vae_force_fp32": True,
|
|
"guidance": 5.0,
|
|
},
|
|
"ltx-2": {"repo": "Lightricks/LTX-2", "vae_force_fp32": False, "guidance": 4.0},
|
|
"hunyuanvideo-1.5": {
|
|
"repo": "hunyuanvideo-community/HunyuanVideo-1.5-Diffusers-480p_t2v",
|
|
"vae_force_fp32": False,
|
|
"guidance": 6.0,
|
|
},
|
|
"hunyuanvideo-1.5-720p": {
|
|
"repo": "hunyuanvideo-community/HunyuanVideo-1.5-Diffusers-720p_t2v",
|
|
"vae_force_fp32": False,
|
|
"guidance": 6.0,
|
|
},
|
|
# Wan2.2-A14B is a dual-expert MoE (transformer + transformer_2); _apply_levers quantizes both.
|
|
"wan2.2-t2v-a14b": {
|
|
"repo": "Wan-AI/Wan2.2-T2V-A14B-Diffusers",
|
|
"vae_force_fp32": True,
|
|
"guidance": 5.0,
|
|
},
|
|
}
|
|
|
|
|
|
# ── cuda memory / timing helpers ───────────────────────────────────────────────
|
|
def _sync() -> None:
|
|
import torch
|
|
if torch.cuda.is_available():
|
|
torch.cuda.synchronize()
|
|
|
|
|
|
def _reset_peak() -> None:
|
|
import torch
|
|
if torch.cuda.is_available():
|
|
torch.cuda.reset_peak_memory_stats()
|
|
|
|
|
|
def _alloc_gb() -> float:
|
|
import torch
|
|
return torch.cuda.memory_allocated() / 1e9 if torch.cuda.is_available() else 0.0
|
|
|
|
|
|
def _peak_gb() -> float:
|
|
import torch
|
|
return torch.cuda.max_memory_allocated() / 1e9 if torch.cuda.is_available() else 0.0
|
|
|
|
|
|
def _empty() -> None:
|
|
import torch
|
|
gc.collect()
|
|
if torch.cuda.is_available():
|
|
torch.cuda.empty_cache()
|
|
|
|
|
|
def _median(xs: list[float]) -> float:
|
|
return sorted(xs)[len(xs) // 2] if xs else 0.0
|
|
|
|
|
|
_LP: dict = {}
|
|
|
|
|
|
def _lpips_alex(ref_arr, arr):
|
|
"""LPIPS(AlexNet) between two HxWx3 uint8 frames (net on CPU). None if lpips missing."""
|
|
try:
|
|
import lpips
|
|
import torch
|
|
|
|
fn = _LP.get("fn")
|
|
if fn is None:
|
|
fn = lpips.LPIPS(net = "alex", verbose = False).eval()
|
|
_LP["fn"] = fn
|
|
|
|
def _t(a):
|
|
import torch as _torch
|
|
return _torch.from_numpy(a).float().permute(2, 0, 1).unsqueeze(0) / 127.5 - 1.0
|
|
|
|
with torch.no_grad():
|
|
return float(fn(_t(ref_arr), _t(arr)).item())
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _frames_to_arrays(output) -> list:
|
|
"""Normalize a video pipeline output to a list of HxWx3 uint8 numpy frames."""
|
|
import numpy as np
|
|
|
|
frames = getattr(output, "frames", None)
|
|
if frames is None:
|
|
return []
|
|
batch0 = frames[0]
|
|
arrs = []
|
|
for fr in batch0:
|
|
if hasattr(fr, "convert"): # PIL image
|
|
arrs.append(np.array(fr.convert("RGB")))
|
|
else:
|
|
a = np.asarray(fr)
|
|
if a.dtype != np.uint8:
|
|
a = np.clip(a * (255.0 if a.max() <= 1.0 else 1.0), 0, 255).astype(np.uint8)
|
|
arrs.append(a)
|
|
return arrs
|
|
|
|
|
|
def _mean_luma(arrs: list) -> Optional[float]:
|
|
"""Mean Rec.601 luma over all frames (0-255). ~0 == black frames (the fp8 failure signal)."""
|
|
import numpy as np
|
|
|
|
if not arrs:
|
|
return None
|
|
vals = []
|
|
for a in arrs:
|
|
a = np.asarray(a).astype(np.float32)
|
|
if a.ndim == 3 and a.shape[-1] >= 3:
|
|
luma = 0.299 * a[..., 0] + 0.587 * a[..., 1] + 0.114 * a[..., 2]
|
|
else:
|
|
luma = a
|
|
vals.append(float(luma.mean()))
|
|
return round(sum(vals) / len(vals), 3) if vals else None
|
|
|
|
|
|
def _mean_lpips(ref_arrs: list, arrs: list) -> Optional[float]:
|
|
"""Mean per-frame LPIPS over the min common frame count."""
|
|
if not ref_arrs or not arrs:
|
|
return None
|
|
n = min(len(ref_arrs), len(arrs))
|
|
vals = []
|
|
for i in range(n):
|
|
v = _lpips_alex(ref_arrs[i], arrs[i])
|
|
if v is not None:
|
|
vals.append(v)
|
|
return round(sum(vals) / len(vals), 4) if vals else None
|
|
|
|
|
|
def _import_diffusers():
|
|
import torch # noqa: F401
|
|
import torchao # noqa: F401
|
|
import diffusers.utils.import_utils as iu
|
|
|
|
iu._bitsandbytes_available = False
|
|
import diffusers
|
|
|
|
return diffusers
|
|
|
|
|
|
def _target():
|
|
"""The object the real casters/optimisers read: a stand-in for DiffusionDeviceTarget.
|
|
supports_default_torch_compile must be True or compile_eligible() bails."""
|
|
import torch
|
|
return types.SimpleNamespace(
|
|
device = "cuda",
|
|
dtype = torch.bfloat16,
|
|
supports_default_torch_compile = True,
|
|
)
|
|
|
|
|
|
# ── config matrix ──────────────────────────────────────────────────────────────
|
|
# Each config names the lever settings for a fresh pipe, built UP from the bit-exact
|
|
# reference so each isolates one lever. `shipped` is the default; `speedmax`/`flash4`
|
|
# probe untapped headroom.
|
|
_CONFIGS: dict[str, dict[str, Any]] = {
|
|
# te vae dit speed attn cache
|
|
"reference": dict(te = "none", vae = "none", dit = "none", speed = "off", attn = "native", cache = "off"),
|
|
"compile": dict(te = "none", vae = "none", dit = "none", speed = "default", attn = "native", cache = "off"),
|
|
"cudnn": dict(te = "none", vae = "none", dit = "none", speed = "default", attn = "auto", cache = "off"),
|
|
"fbcache": dict(te = "none", vae = "none", dit = "none", speed = "default", attn = "auto", cache = "auto"),
|
|
"ditquant": dict(te = "none", vae = "none", dit = "auto", speed = "default", attn = "auto", cache = "auto"),
|
|
"shipped": dict(te = "auto", vae = "auto", dit = "auto", speed = "default", attn = "auto", cache = "auto"),
|
|
"speedmax": dict(te = "auto", vae = "auto", dit = "auto", speed = "max", attn = "auto", cache = "auto"),
|
|
"flash4": dict(te = "auto", vae = "auto", dit = "auto", speed = "default", attn = "flash4", cache = "auto"),
|
|
# diagnostics: isolate whether the DiT-quant + compile crash needs FBCache.
|
|
"diag_ditq_default_nocache": dict(
|
|
te = "none", vae = "none", dit = "auto", speed = "default", attn = "native", cache = "off"
|
|
),
|
|
"diag_ditq_max_nocache": dict(
|
|
te = "none", vae = "none", dit = "auto", speed = "max", attn = "native", cache = "off"
|
|
),
|
|
"diag_ditq_nocompile": dict(
|
|
te = "none", vae = "none", dit = "auto", speed = "eager", attn = "native", cache = "off"
|
|
),
|
|
# which quant scheme survives torch.compile? (fp8/mslk fails "fake tensors"; test int8/mxfp8)
|
|
"diag_ditint8_compile": dict(
|
|
te = "none", vae = "none", dit = "int8", speed = "default", attn = "native", cache = "off"
|
|
),
|
|
"diag_ditmxfp8_compile": dict(
|
|
te = "none", vae = "none", dit = "mxfp8", speed = "default", attn = "native", cache = "off"
|
|
),
|
|
"diag_ditint8_fbcache": dict(
|
|
te = "none", vae = "none", dit = "int8", speed = "default", attn = "native", cache = "auto"
|
|
),
|
|
# isolate the quant x FBCache over-caching interaction at production size:
|
|
"ditfp8_nocache": dict(
|
|
te = "none", vae = "none", dit = "auto", speed = "default", attn = "auto", cache = "off"
|
|
),
|
|
"te_fbcache": dict(
|
|
te = "auto", vae = "none", dit = "none", speed = "default", attn = "auto", cache = "auto"
|
|
),
|
|
# Companion-quant accuracy isolation vs the reference (uncached): TE-only and VAE-only on
|
|
# the trim+cudnn+compile stack. With compile rounding fixed, the companions are the
|
|
# next-largest divergence source.
|
|
"diag_te_nocache": dict(
|
|
te = "auto", vae = "none", dit = "none", speed = "default", attn = "auto", cache = "off"
|
|
),
|
|
"diag_vae_nocache": dict(
|
|
te = "none", vae = "auto", dit = "none", speed = "default", attn = "auto", cache = "off"
|
|
),
|
|
"ditfp8_fbcache": dict(
|
|
te = "none", vae = "none", dit = "auto", speed = "default", attn = "auto", cache = "auto"
|
|
),
|
|
"ditint8_fbcache_prod": dict(
|
|
te = "none", vae = "none", dit = "int8", speed = "default", attn = "auto", cache = "auto"
|
|
),
|
|
# mixed-fp8 vs int8 head-to-head on Wan/Hunyuan. fp8 goes through the production family
|
|
# exclude (input embedders kept bf16), so it is only non-black if that wiring is live.
|
|
"ditfp8mixed_nocache": dict(
|
|
te = "none", vae = "none", dit = "fp8", speed = "default", attn = "native", cache = "off"
|
|
),
|
|
"ditfp8mixed_fbcache": dict(
|
|
te = "none", vae = "none", dit = "fp8", speed = "default", attn = "auto", cache = "auto"
|
|
),
|
|
"ditint8_nocache": dict(
|
|
te = "none", vae = "none", dit = "int8", speed = "default", attn = "native", cache = "off"
|
|
),
|
|
# Hunyuan trim isolation: "cudnn" above is the same stack WITH the trim (auto-engaged
|
|
# under any speed tier), so trim_off isolates its win. Only this row forces trim off.
|
|
"trim_off": dict(
|
|
te = "none",
|
|
vae = "none",
|
|
dit = "none",
|
|
speed = "default",
|
|
attn = "auto",
|
|
cache = "off",
|
|
trim = False,
|
|
),
|
|
# Compile isolation at matched attention/trim: eager tier (no compile) vs "cudnn"
|
|
# (default tier = regional compile).
|
|
"eager_trim": dict(te = "none", vae = "none", dit = "none", speed = "eager", attn = "auto", cache = "off"),
|
|
# int8 DiT baseline at the production attention stack (cudnn + trim), cache off,
|
|
# comparable to the "cudnn" dense row.
|
|
"int8_cudnn": dict(
|
|
te = "none", vae = "none", dit = "int8", speed = "default", attn = "auto", cache = "off"
|
|
),
|
|
# The full companion-quant stack WITHOUT step caching (te/vae auto + dit auto + compile
|
|
# + cudnn + trim): the stacked-best default when the family's auto cache stays off.
|
|
"shipped_nocache": dict(
|
|
te = "auto", vae = "auto", dit = "auto", speed = "default", attn = "auto", cache = "off"
|
|
),
|
|
# Compile-parity isolation: the "cudnn" compiled stack but with inductor's
|
|
# emulate_precision_casts turned back OFF, to measure that parity flag's effect.
|
|
"epc_off": dict(
|
|
te = "none",
|
|
vae = "none",
|
|
dit = "none",
|
|
speed = "default",
|
|
attn = "auto",
|
|
cache = "off",
|
|
epc = False,
|
|
),
|
|
# Explicit cache modes at the dense compiled stack (bypass AUTO), for FBCache-vs-MagCache
|
|
# head-to-head rows at identical settings.
|
|
"fbcache_explicit": dict(
|
|
te = "none", vae = "none", dit = "none", speed = "default", attn = "auto", cache = "fbcache"
|
|
),
|
|
"magcache_explicit": dict(
|
|
te = "none", vae = "none", dit = "none", speed = "default", attn = "auto", cache = "magcache"
|
|
),
|
|
}
|
|
|
|
|
|
def _ref_cache_path(out, *, family, seed, steps, num_frames, width, height):
|
|
"""Reference-frame cache keyed by every parameter that changes the reference clip.
|
|
|
|
The persisted reference (for the "run reference once, score other configs in parallel
|
|
processes" workflow) shares the default --out dir across runs, so a single unkeyed
|
|
ref_frames.npz would let a later reference-less run of a different family / seed / steps /
|
|
frames / resolution score LPIPS against the wrong baseline. Key by all of them so a
|
|
reference-less run only reuses a reference computed for the same parameters."""
|
|
from pathlib import Path
|
|
return Path(out) / (
|
|
f"ref_frames_{family}_seed{seed}_st{steps}_f{num_frames}_{width}x{height}.npz"
|
|
)
|
|
|
|
|
|
def _build_pipe(repo: str, force_fp32_vae: bool):
|
|
import torch
|
|
|
|
diffusers = _import_diffusers()
|
|
# Wan-style VAEs decode in fp32 (the loader pins this via vae_force_fp32). A scalar bf16
|
|
# dtype truncates the fp32 VAE weights at load and a later .to(float32) only widens the
|
|
# lossy values (banding), so pin the VAE fp32 per-component like the loader.
|
|
torch_dtype = torch.bfloat16
|
|
if force_fp32_vae:
|
|
torch_dtype = {"vae": torch.float32, "default": torch.bfloat16}
|
|
pipe = diffusers.DiffusionPipeline.from_pretrained(repo, torch_dtype = torch_dtype)
|
|
# Stays on CPU: the caller applies levers FIRST, then places on CUDA (like the loader).
|
|
# Placing the dense pipeline first would OOM configs whose quantized form fits, and
|
|
# record a dense load peak for a quantized row.
|
|
if force_fp32_vae and getattr(pipe, "vae", None) is not None:
|
|
pipe.vae.to(torch.float32) # belt-and-suspenders; a no-op on the primary path above
|
|
return pipe
|
|
|
|
|
|
class _SecondExpertView:
|
|
"""Present ``pipe.transformer_2`` as ``.transformer`` so single-DiT lever functions run on an
|
|
MoE second expert (Wan2.2-A14B), like the loader's _SecondDiTView. Reads delegate to the pipe
|
|
except ``transformer``; a ``transformer`` write is routed to ``transformer_2``."""
|
|
|
|
def __init__(self, pipe):
|
|
object.__setattr__(self, "_pipe", pipe)
|
|
|
|
@property
|
|
def transformer(self):
|
|
return self._pipe.transformer_2
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._pipe, name)
|
|
|
|
def __setattr__(self, name, value):
|
|
setattr(self._pipe, "transformer_2" if name == "transformer" else name, value)
|
|
|
|
|
|
def _apply_levers(
|
|
pipe,
|
|
cfg: dict,
|
|
*,
|
|
fam_name: str,
|
|
fam_obj,
|
|
force_fp32_vae: bool,
|
|
default_steps: int,
|
|
cache_threshold: Optional[float] = None,
|
|
cache_quality: Optional[str] = None,
|
|
logger = None,
|
|
) -> dict:
|
|
"""Apply the configured levers with the loader's own argument values, in the loader's order:
|
|
quant (dit -> te -> vae) THEN optimisation layers (cache -> attention -> speed). For a
|
|
dual-expert MoE (pipe.transformer_2 present) every DiT-touching lever is applied to BOTH experts
|
|
via _SecondExpertView, exactly like the loader, so A14B latency + accuracy are real."""
|
|
from core.inference.diffusion_precision import quantize_text_encoders
|
|
from core.inference.diffusion_vae_quant import quantize_vae
|
|
from core.inference.diffusion_transformer_quant import (
|
|
quantize_transformer,
|
|
is_int8_memory_fallback,
|
|
)
|
|
from core.inference.diffusion_speed import apply_speed_optims, snapshot_backend_flags
|
|
from core.inference.diffusion_attention import (
|
|
select_attention_backend,
|
|
apply_attention_backend,
|
|
install_hunyuan_attention_trim,
|
|
)
|
|
from core.inference.diffusion_cache import (
|
|
apply_step_cache,
|
|
auto_cache_mode,
|
|
auto_cache_quality,
|
|
normalize_cache_quality,
|
|
FBCACHE_MIN_STEPS,
|
|
)
|
|
from core.inference.video import _step_cache_all_or_none
|
|
|
|
tgt = _target()
|
|
engaged = {
|
|
"dit": None,
|
|
"te": None,
|
|
"vae": None,
|
|
"attn": None,
|
|
"cache": None,
|
|
"speed_optims": {},
|
|
}
|
|
|
|
# DiT-touching levers run per expert: [pipe] for a single-DiT family, plus a second-expert view
|
|
# for a dual-expert MoE. Each view exposes the expert as ``.transformer``.
|
|
views = [pipe]
|
|
if getattr(pipe, "transformer_2", None) is not None:
|
|
views.append(_SecondExpertView(pipe))
|
|
|
|
# DiT quant (resident): mutates each expert's transformer in place. Mirror the loader's
|
|
# dense-fit skip: for an AUTO request on an int8-fallback family (HunyuanVideo-1.5), run dense
|
|
# when it fits resident (the bench always loads resident). Explicit int8/fp8 configs are
|
|
# honored. Without this the auto rows would measure int8 where the loader runs dense.
|
|
dense_fit_skip = cfg["dit"] == "auto" and is_int8_memory_fallback(tgt, fam_name)
|
|
if cfg["dit"] not in ("none", "off") and not dense_fit_skip:
|
|
schemes = [
|
|
quantize_transformer(v, tgt, mode = cfg["dit"], family = fam_name, logger = logger)
|
|
for v in views
|
|
]
|
|
# All-or-none across experts, like the loader: a second-expert miss can't fall back to
|
|
# dense, so reject here rather than publish timings for a pipeline users can't load.
|
|
n_engaged = sum(1 for s in schemes if s is not None)
|
|
if 0 < n_engaged < len(views):
|
|
raise RuntimeError(
|
|
f"dit quant '{cfg['dit']}' engaged on only {n_engaged}/{len(views)} experts; "
|
|
"production rejects this partial state, so the row would be unloadable"
|
|
)
|
|
engaged["dit"] = schemes[0]
|
|
engaged["dit_experts"] = schemes
|
|
_empty()
|
|
dit_quant_active = engaged["dit"] is not None
|
|
|
|
# TE quant (once; text encoders are shared, not per-expert).
|
|
if cfg["te"] not in ("none", "off"):
|
|
engaged["te"] = quantize_text_encoders(
|
|
pipe, tgt, mode = cfg["te"], family = fam_name, offload_active = False, logger = logger
|
|
)
|
|
_empty()
|
|
|
|
# VAE quant (once; Wan force_fp32 pins dense inside quantize_vae regardless).
|
|
if cfg["vae"] not in ("none", "off"):
|
|
engaged["vae"] = quantize_vae(
|
|
pipe,
|
|
tgt,
|
|
mode = cfg["vae"],
|
|
family = fam_name,
|
|
offload_active = False,
|
|
force_fp32 = force_fp32_vae,
|
|
logger = logger,
|
|
)
|
|
_empty()
|
|
|
|
# ── optimisation layers ──
|
|
# Process-wide flags (cudnn.benchmark, TF32/fp16 accumulation, emulate_precision_casts):
|
|
# snapshotted here and restored by _run_config after the row, like the unload path -- else
|
|
# a speed-enabled row poisons a later reference row in the same process.
|
|
engaged["_flags_snapshot"] = snapshot_backend_flags()
|
|
speed = cfg["speed"]
|
|
speed_active = speed != "off"
|
|
|
|
# A quantized DiT must be compiled (eager dynamic quant ~30x slower), matching the loader.
|
|
if dit_quant_active and speed == "off":
|
|
speed = "default"
|
|
speed_active = True
|
|
|
|
# Step cache FIRST (compile keys fullgraph off an active cache); per expert.
|
|
cache_active = False
|
|
if cfg["cache"] in ("auto", "fbcache", "magcache"):
|
|
# Per-family auto mode like the loader: MagCache for HunyuanVideo-1.5, FBCache
|
|
# elsewhere. An explicit "fbcache"/"magcache" config bypasses the auto policy.
|
|
if cfg["cache"] == "auto":
|
|
cache_request = (
|
|
auto_cache_mode(fam_name) if default_steps >= FBCACHE_MIN_STEPS else None
|
|
)
|
|
else:
|
|
cache_request = cfg["cache"]
|
|
if cache_request is not None:
|
|
# Quality preset like the loader: an unset request takes the family's auto default.
|
|
quality = normalize_cache_quality(cache_quality) or auto_cache_quality(fam_name)
|
|
|
|
# All-or-none across MoE experts, exactly like the loader: overwriting engaged["cache"]
|
|
# per expert would leave one expert cached and one dense on a partial engage while the
|
|
# row reports the cache off -- a config that never runs in production. The shared helper
|
|
# rolls the engaged expert(s) back so the row measures a real configuration.
|
|
def _engage_cache(v: Any, expert: str) -> Optional[str]:
|
|
return apply_step_cache(
|
|
v,
|
|
mode = cache_request,
|
|
threshold = cache_threshold,
|
|
quant_active = dit_quant_active,
|
|
family = fam_name,
|
|
steps = default_steps,
|
|
quality = quality,
|
|
expert = expert,
|
|
logger = logger,
|
|
)
|
|
|
|
engaged["cache"], cache_partial_reason = _step_cache_all_or_none(
|
|
pipe, fam_obj, _engage_cache, logger = logger
|
|
)
|
|
if cache_partial_reason and logger is not None:
|
|
logger.warning("benchmark cache disabled: %s", cache_partial_reason)
|
|
cache_active = engaged["cache"] not in (None, "off")
|
|
|
|
# HunyuanVideo-1.5 joint-attention trim (per expert), BEFORE the backend set like the loader.
|
|
# Drops the ~99% zero-padded text tokens so the fused SDPA runs (~18x/DiT-forward, cosine ~1.0).
|
|
# Gated on an active tier; no-op for non-Hunyuan families.
|
|
trim_engaged = False
|
|
if speed_active and cfg.get("trim", True):
|
|
for v in views:
|
|
trim_engaged = install_hunyuan_attention_trim(v, fam_obj, logger = logger) or trim_engaged
|
|
engaged["attn_trim"] = trim_engaged
|
|
|
|
# Attention (per expert).
|
|
backend = select_attention_backend(tgt, cfg["attn"], speed_active = speed_active)
|
|
for v in views:
|
|
engaged["attn"] = apply_attention_backend(v, backend, logger = logger)
|
|
|
|
# Speed profile: apply_speed_optims fans out over EVERY denoiser DiT internally
|
|
# (transformer + transformer_2), so ONE pipe-level call covers a dual-expert MoE --
|
|
# exactly like the production loader. Looping over the second-expert view would compile
|
|
# and fuse transformer_2 a second time, so the warmup/compiled state these rows measure
|
|
# would no longer match the loader this script mirrors.
|
|
if speed != "off":
|
|
engaged["speed_optims"] = apply_speed_optims(
|
|
pipe,
|
|
tgt,
|
|
is_gguf = False,
|
|
family = fam_obj,
|
|
speed_mode = speed,
|
|
cache_active = cache_active,
|
|
offload_active = False,
|
|
logger = logger,
|
|
)
|
|
engaged["_effective_speed"] = speed
|
|
# emulate_precision_casts A/B: the speed layer sets it True; compile is lazy, so flipping
|
|
# it back off here gives the pre-round-2 inductor numerics for the run ("epc_off" config).
|
|
if not cfg.get("epc", True):
|
|
try:
|
|
import torch
|
|
torch._inductor.config.emulate_precision_casts = False
|
|
engaged["epc"] = False
|
|
except Exception:
|
|
pass
|
|
return engaged
|
|
|
|
|
|
def _timed_video(
|
|
pipe,
|
|
*,
|
|
steps,
|
|
width,
|
|
height,
|
|
num_frames,
|
|
guidance,
|
|
seed,
|
|
cache_mode,
|
|
dit_quant_active,
|
|
default_steps,
|
|
guidance_via_guider = False,
|
|
cache_threshold = None,
|
|
cache_quality = None,
|
|
family = None,
|
|
fam_obj = None,
|
|
logger = None,
|
|
):
|
|
"""One clip generation. Re-checks the step cache per generation (maybe_toggle_step_cache)
|
|
exactly like the loader, then times total + per-step. Returns (output, total_s, [per_step_ms])."""
|
|
import torch
|
|
|
|
from core.inference.diffusion_cache import (
|
|
_disengage_step_cache,
|
|
apply_step_cache,
|
|
auto_cache_mode,
|
|
auto_cache_quality,
|
|
maybe_toggle_step_cache,
|
|
normalize_cache_quality,
|
|
)
|
|
from core.inference.video import _step_cache_all_or_none
|
|
|
|
# Re-check the step cache per generation exactly like the loader, and route it through the
|
|
# SAME transactional helper (_step_cache_all_or_none) the load path uses. Toggling each expert
|
|
# view independently (swallowing per-expert results/exceptions) can engage one expert while
|
|
# transformer_2 fails, so the row would time a mixed cached/uncached MoE state -- a config
|
|
# production never runs, and one whose stale engagement would poison later rows in the matrix.
|
|
# The helper rolls back a partial engage so every timed row measures a real configuration.
|
|
if cache_mode == "auto":
|
|
|
|
def _toggle_cache(view: Any, expert: str) -> Optional[str]:
|
|
return maybe_toggle_step_cache(
|
|
view,
|
|
steps = steps,
|
|
quant_active = dit_quant_active,
|
|
threshold = cache_threshold,
|
|
mode = auto_cache_mode(family),
|
|
family = family,
|
|
quality = normalize_cache_quality(cache_quality) or auto_cache_quality(family),
|
|
expert = expert,
|
|
logger = logger,
|
|
)
|
|
|
|
_step_cache_all_or_none(pipe, fam_obj, _toggle_cache, logger = logger)
|
|
elif cache_mode == "magcache":
|
|
# An explicit magcache never toggles off, but production re-engages it when the step
|
|
# count differs (marker carries "#s{steps}"; endswith, not substring). _apply_levers
|
|
# installed at default_steps, so a --steps override would else time a stale curve. Wrapped
|
|
# in the all-or-none helper so a mixed resize rolls back instead of timing a split pair.
|
|
def _resize_magcache(view: Any, expert: str) -> Optional[str]:
|
|
transformer = getattr(view, "transformer", None)
|
|
marker = getattr(transformer, "_unsloth_step_cache", None)
|
|
if not marker or str(marker).endswith(f"#s{int(steps)}"):
|
|
return "magcache" # already sized for these steps
|
|
# Fail closed: reapplying over a cache that would not disengage double-hooks the
|
|
# transformer and times a stale curve as if it were fresh.
|
|
if not _disengage_step_cache(
|
|
transformer,
|
|
reason = f"explicit magcache re-interpolating for {steps} steps",
|
|
logger = logger,
|
|
):
|
|
raise RuntimeError(
|
|
"could not disable the existing MagCache before resizing it for "
|
|
f"{steps} steps; restart the benchmark with a fresh pipeline"
|
|
)
|
|
return apply_step_cache(
|
|
view,
|
|
mode = "magcache",
|
|
threshold = cache_threshold,
|
|
quant_active = dit_quant_active,
|
|
family = family,
|
|
steps = steps,
|
|
quality = normalize_cache_quality(cache_quality) or auto_cache_quality(family),
|
|
expert = expert,
|
|
logger = logger,
|
|
)
|
|
|
|
_step_cache_all_or_none(pipe, fam_obj, _resize_magcache, logger = logger)
|
|
|
|
# Clear step-cache residuals before EVERY generation, like production: diffusers keys
|
|
# them on the long-lived transformer, so measured iterations would otherwise start
|
|
# against the previous clip's cache state. Best-effort, uncached is a no-op.
|
|
for name in ("transformer", "transformer_2"):
|
|
module = getattr(pipe, name, None)
|
|
reset = getattr(module, "_reset_stateful_cache", None) or getattr(
|
|
module, "reset_stateful_hooks", None
|
|
)
|
|
if callable(reset):
|
|
try:
|
|
reset()
|
|
except Exception:
|
|
pass
|
|
|
|
g = torch.Generator(device = "cuda").manual_seed(seed)
|
|
step_ts: list[float] = []
|
|
last = [0.0]
|
|
|
|
def _cb(pp, i, t, kw):
|
|
torch.cuda.synchronize()
|
|
now = time.perf_counter()
|
|
if last[0]:
|
|
step_ts.append((now - last[0]) * 1000.0)
|
|
last[0] = now
|
|
return kw
|
|
|
|
kwargs = dict(
|
|
prompt = PROMPT,
|
|
width = width,
|
|
height = height,
|
|
num_frames = num_frames,
|
|
num_inference_steps = steps,
|
|
generator = g,
|
|
)
|
|
restore_step = None
|
|
if guidance_via_guider:
|
|
# HunyuanVideo-1.5: CFG lives on a guider component; __call__ takes no guidance_scale
|
|
# (the loader writes the scale onto pipe.guider).
|
|
guider = getattr(pipe, "guider", None)
|
|
if guider is not None and hasattr(guider, "guidance_scale"):
|
|
try:
|
|
guider.guidance_scale = guidance
|
|
except Exception:
|
|
pass
|
|
# __call__ ignores callback_on_step_end here, so the _cb step timer never fires and
|
|
# per_step_ms would publish 0.0 for every Hunyuan row. Time the denoise via a
|
|
# scheduler.step wrapper instead, restored after generation.
|
|
sched = getattr(pipe, "scheduler", None)
|
|
orig_step = getattr(sched, "step", None)
|
|
if callable(orig_step):
|
|
|
|
def _timed_step(*a, **k):
|
|
torch.cuda.synchronize()
|
|
now = time.perf_counter()
|
|
if last[0]:
|
|
step_ts.append((now - last[0]) * 1000.0)
|
|
last[0] = now
|
|
return orig_step(*a, **k)
|
|
|
|
sched.step = _timed_step
|
|
restore_step = lambda: setattr(sched, "step", orig_step) # noqa: E731
|
|
else:
|
|
kwargs["guidance_scale"] = guidance
|
|
kwargs["callback_on_step_end"] = _cb
|
|
|
|
_sync()
|
|
t0 = time.perf_counter()
|
|
try:
|
|
out = pipe(**kwargs)
|
|
finally:
|
|
if restore_step is not None:
|
|
restore_step()
|
|
_sync()
|
|
return out, (time.perf_counter() - t0), step_ts
|
|
|
|
|
|
def _run_config(
|
|
name: str,
|
|
cfg: dict,
|
|
*,
|
|
family: str,
|
|
steps: int,
|
|
width: int,
|
|
height: int,
|
|
num_frames: int,
|
|
seed: int,
|
|
iters: int,
|
|
out: Path,
|
|
cache_threshold: Optional[float] = None,
|
|
cache_quality: Optional[str] = None,
|
|
logger = None,
|
|
):
|
|
import numpy as np
|
|
|
|
from core.inference.video_families import detect_video_family
|
|
|
|
spec = _FAMILIES[family]
|
|
repo = spec["repo"]
|
|
force_fp32 = spec.get("vae_force_fp32", False)
|
|
guidance = spec.get("guidance", 5.0)
|
|
fam_obj = detect_video_family(repo)
|
|
default_steps = getattr(fam_obj, "default_steps", 50)
|
|
gvg = bool(getattr(fam_obj, "guidance_via_guider", False))
|
|
|
|
_empty()
|
|
_reset_peak()
|
|
# Fresh dynamo state per config so a prior config's compiled graphs can't leak in.
|
|
try:
|
|
import torch
|
|
torch._dynamo.reset()
|
|
except Exception:
|
|
pass
|
|
# Loader order: build on CPU, apply levers, THEN place on CUDA. Placing dense first would
|
|
# OOM configs whose quantized form fits, and misrecord load_peak_gb.
|
|
pipe = _build_pipe(repo, force_fp32)
|
|
engaged = _apply_levers(
|
|
pipe,
|
|
cfg,
|
|
fam_name = family,
|
|
fam_obj = fam_obj,
|
|
force_fp32_vae = force_fp32,
|
|
default_steps = default_steps,
|
|
cache_threshold = cache_threshold,
|
|
cache_quality = cache_quality,
|
|
logger = logger,
|
|
)
|
|
pipe = pipe.to("cuda")
|
|
_sync()
|
|
load_peak = _peak_gb()
|
|
_empty()
|
|
weights_gb = _alloc_gb()
|
|
dit_active = engaged["dit"] is not None
|
|
cache_mode = cfg["cache"]
|
|
|
|
# warmup (pays the one-time compile / autotune)
|
|
warmup_t0 = time.perf_counter()
|
|
_timed_video(
|
|
pipe,
|
|
steps = steps,
|
|
width = width,
|
|
height = height,
|
|
num_frames = num_frames,
|
|
guidance = guidance,
|
|
seed = seed,
|
|
cache_mode = cache_mode,
|
|
dit_quant_active = dit_active,
|
|
default_steps = default_steps,
|
|
guidance_via_guider = gvg,
|
|
cache_threshold = cache_threshold,
|
|
cache_quality = cache_quality,
|
|
family = family,
|
|
fam_obj = fam_obj,
|
|
logger = logger,
|
|
)
|
|
warmup_s = time.perf_counter() - warmup_t0
|
|
_reset_peak()
|
|
dts, steps_ms = [], []
|
|
last_out = None
|
|
for _ in range(iters):
|
|
last_out, dt, st = _timed_video(
|
|
pipe,
|
|
steps = steps,
|
|
width = width,
|
|
height = height,
|
|
num_frames = num_frames,
|
|
guidance = guidance,
|
|
seed = seed,
|
|
cache_mode = cache_mode,
|
|
dit_quant_active = dit_active,
|
|
default_steps = default_steps,
|
|
guidance_via_guider = gvg,
|
|
cache_threshold = cache_threshold,
|
|
cache_quality = cache_quality,
|
|
family = family,
|
|
fam_obj = fam_obj,
|
|
logger = logger,
|
|
)
|
|
dts.append(dt)
|
|
steps_ms.append(_median(st) if st else 0.0)
|
|
gen_peak = _peak_gb()
|
|
arrs = _frames_to_arrays(last_out)
|
|
# save a mid frame for eyeballing
|
|
try:
|
|
if arrs:
|
|
from PIL import Image
|
|
Image.fromarray(arrs[len(arrs) // 2]).save(out / f"vid_{family}_{name}.png")
|
|
except Exception:
|
|
pass
|
|
|
|
# Persist reference frames so parallel per-config processes can score LPIPS against them.
|
|
if name == "reference" and arrs:
|
|
try:
|
|
import numpy as _np
|
|
_np.savez_compressed(
|
|
_ref_cache_path(
|
|
out,
|
|
family = family,
|
|
seed = seed,
|
|
steps = steps,
|
|
num_frames = num_frames,
|
|
width = width,
|
|
height = height,
|
|
),
|
|
*arrs,
|
|
)
|
|
except Exception:
|
|
pass
|
|
# Persist EVERY config's frames too, so a row generated before the reference exists can
|
|
# be LPIPS-rescored offline instead of publishing null.
|
|
if arrs:
|
|
try:
|
|
import numpy as _np
|
|
_np.savez_compressed(out / f"frames_{family}_{name}.npz", *arrs)
|
|
except Exception:
|
|
pass
|
|
|
|
# Report the GENERATION-time cache state, not the load-time one: the per-generation recheck can
|
|
# toggle an auto cache off or re-size a magcache. The marker's mode prefix IS the live state.
|
|
cache_marker = getattr(getattr(pipe, "transformer", None), "_unsloth_step_cache", None)
|
|
row = {
|
|
"config": name,
|
|
"family": family,
|
|
"levers": cfg,
|
|
"dit_scheme": engaged["dit"] or "dense",
|
|
"dit_experts": engaged.get("dit_experts"),
|
|
"te_scheme": engaged["te"] or "dense",
|
|
"vae_scheme": engaged["vae"] or "dense",
|
|
"attn": engaged["attn"] or "native",
|
|
"cache": str(cache_marker).split("@")[0] if cache_marker else "off",
|
|
"cache_at_load": engaged["cache"] or "off",
|
|
"effective_speed": engaged.get("_effective_speed"),
|
|
"speed_optims": engaged["speed_optims"],
|
|
"attn_trim": engaged.get("attn_trim", False),
|
|
"cache_threshold": cache_threshold,
|
|
"cache_quality": cache_quality,
|
|
"cache_marker": cache_marker,
|
|
"load_peak_gb": round(load_peak, 2),
|
|
"weights_gb": round(weights_gb, 2),
|
|
"gen_peak_gb": round(gen_peak, 2),
|
|
"warmup_s": round(warmup_s, 3),
|
|
"gen_latency_s": round(_median(dts), 3),
|
|
"per_step_ms": round(_median(steps_ms), 1),
|
|
"n_frames": len(arrs),
|
|
"mean_luma": _mean_luma(arrs),
|
|
}
|
|
del pipe
|
|
_empty()
|
|
# Restore the process-wide backend flags this config's speed layer mutated (snapshot from
|
|
# _apply_levers), like the unload path, so a later config starts from clean flags.
|
|
try:
|
|
from core.inference.diffusion_speed import restore_backend_flags
|
|
restore_backend_flags(engaged.get("_flags_snapshot"))
|
|
except Exception:
|
|
pass
|
|
return row, arrs
|
|
|
|
|
|
def main(argv = None) -> int:
|
|
ap = argparse.ArgumentParser(description = __doc__)
|
|
ap.add_argument("--family", default = "wan2.2-ti2v-5b", choices = sorted(_FAMILIES))
|
|
ap.add_argument(
|
|
"--configs", default = ",".join(_CONFIGS), help = "comma list from: " + ",".join(_CONFIGS)
|
|
)
|
|
ap.add_argument("--steps", type = int, default = 30)
|
|
ap.add_argument("--num-frames", type = int, default = 25)
|
|
ap.add_argument("--width", type = int, default = 512)
|
|
ap.add_argument("--height", type = int, default = 320)
|
|
ap.add_argument("--seed", type = int, default = 42)
|
|
ap.add_argument("--iters", type = int, default = 3)
|
|
ap.add_argument("--out", default = "outputs/video_speedmem")
|
|
ap.add_argument(
|
|
"--cache-threshold",
|
|
type = float,
|
|
default = None,
|
|
help = "FBCache residual-diff threshold override (None -> the production default)",
|
|
)
|
|
ap.add_argument(
|
|
"--cache-quality",
|
|
default = None,
|
|
choices = ("quality", "balanced", "fast"),
|
|
help = "Step-cache quality preset (None -> the family's production auto default)",
|
|
)
|
|
args = ap.parse_args(argv)
|
|
|
|
import logging
|
|
|
|
logging.basicConfig(level = logging.INFO, format = "%(message)s")
|
|
logger = logging.getLogger("videobench")
|
|
|
|
out = Path(args.out)
|
|
out.mkdir(parents = True, exist_ok = True)
|
|
|
|
names = [c.strip() for c in args.configs.split(",") if c.strip()]
|
|
for n in names:
|
|
if n not in _CONFIGS:
|
|
raise SystemExit(f"unknown config '{n}'; choose from {list(_CONFIGS)}")
|
|
|
|
print(
|
|
f"== video speed+mem bench: family={args.family} configs={names} "
|
|
f"steps={args.steps} frames={args.num_frames} {args.width}x{args.height} ==",
|
|
flush = True,
|
|
)
|
|
|
|
rows = []
|
|
ref_arrs = None
|
|
# If not (re)computing the reference in this run, load persisted reference frames for LPIPS.
|
|
if "reference" not in names:
|
|
ref_npz = _ref_cache_path(
|
|
out,
|
|
family = args.family,
|
|
seed = args.seed,
|
|
steps = args.steps,
|
|
num_frames = args.num_frames,
|
|
width = args.width,
|
|
height = args.height,
|
|
)
|
|
if ref_npz.exists():
|
|
try:
|
|
import numpy as _np
|
|
with _np.load(ref_npz) as z:
|
|
ref_arrs = [z[k] for k in z.files]
|
|
except Exception:
|
|
ref_arrs = None
|
|
row_arrs: list = []
|
|
for n in names:
|
|
row, arrs = _run_config(
|
|
n,
|
|
_CONFIGS[n],
|
|
family = args.family,
|
|
steps = args.steps,
|
|
width = args.width,
|
|
height = args.height,
|
|
num_frames = args.num_frames,
|
|
seed = args.seed,
|
|
iters = args.iters,
|
|
out = out,
|
|
cache_threshold = args.cache_threshold,
|
|
cache_quality = args.cache_quality,
|
|
logger = logger,
|
|
)
|
|
if n == "reference":
|
|
ref_arrs = arrs
|
|
row["lpips_vs_reference"] = _mean_lpips(ref_arrs, arrs) if ref_arrs is not None else None
|
|
rows.append(row)
|
|
row_arrs.append(arrs)
|
|
print(
|
|
f" [{n}] {json.dumps({k: row[k] for k in ('dit_scheme','te_scheme','attn','cache','effective_speed','weights_gb','gen_peak_gb','gen_latency_s','per_step_ms','lpips_vs_reference')})}",
|
|
flush = True,
|
|
)
|
|
|
|
# Rows finalized BEFORE the reference clip was generated scored lpips_vs_reference as None;
|
|
# rescore them now that the reference frames exist.
|
|
if ref_arrs is not None:
|
|
for r, arrs in zip(rows, row_arrs):
|
|
if r["lpips_vs_reference"] is None:
|
|
r["lpips_vs_reference"] = _mean_lpips(ref_arrs, arrs)
|
|
|
|
# speedups relative to reference (if present)
|
|
ref_lat = next((r["gen_latency_s"] for r in rows if r["config"] == "reference"), None)
|
|
for r in rows:
|
|
r["speedup_vs_reference"] = (
|
|
round(ref_lat / r["gen_latency_s"], 3) if ref_lat and r["gen_latency_s"] else None
|
|
)
|
|
|
|
dest = out / f"video_{args.family}_{'-'.join(names)}.json"
|
|
with open(dest, "w", encoding = "utf-8") as fh:
|
|
json.dump(rows, fh, indent = 2)
|
|
print(f"wrote {dest}", flush = True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|