Compare commits

..
Author SHA1 Message Date
hsliuustc0106 b188874539 [Perf][Diffusion] Add configurable H3 VAE stacked tiling
Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-09-03 17:17:59 +00:00
16 changed files with 215 additions and 76 deletions

No files matched your search

@@ -120,10 +120,18 @@ Additional requirements:
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `vae_use_tiling` | bool | False | Must be set to `True` when using VAE patch parallelism. |
| `vae_stack_tiling` | `"auto"` \| `"true"` \| `"false"` | `"false"` | For VAEs with native support (currently MiniMax H3), batch local decoder tiles into fewer model calls. `"auto"` enables it when the request has at least two tiles per VAE-parallel rank. |
!!! note "Automatic VAE Tiling"
When `vae_patch_parallel_size > 1` and the model has a distributed VAE (`DistributedVaeMixin`), the system automatically sets `vae_use_tiling=True` if not already enabled.
!!! tip "MiniMax H3 stacked tiling"
MiniMax H3 supports lossless native stacked tiling. Enable it with
`--vae-stack-tiling auto` (or `true`) to batch each rank's spatial tiles
into one decoder call per temporal window. `auto` keeps the sequential
path for shapes that do not provide multiple tiles per rank, while
`false` remains the default to avoid increasing the VAE activation peak.
---
## Spatially-Sharded Decode (Wan)
@@ -21,7 +21,6 @@ from vllm.multimodal.media.audio import load_audio
from vllm.sampling_params import SamplingParams
from vllm_omni.entrypoints.omni import Omni
from vllm_omni.utils.profiler import add_profiler_config_arg
from vllm_omni.utils.tracking_parser import TrackingArgumentParser
SEED = 42
@@ -326,7 +325,6 @@ def main(args):
else:
query_result = query_func()
args.quantization_config = quantization_config
# ``vars(args)`` below forwards profiler_config to Omni.
omni_kwargs = vars(args).copy()
# Override CLI --model with the derived model_name.
omni_kwargs["model"] = model_name
@@ -380,11 +378,11 @@ def main(args):
for i, prompt in enumerate(prompts):
prompt["modalities"] = output_modalities
profiler_enabled = args.profiler_config is not None
profiler_enabled = bool(os.getenv("VLLM_TORCH_PROFILER_DIR"))
if profiler_enabled and hasattr(omni, "start_profile"):
omni.start_profile(stages=[0])
elif profiler_enabled:
print("[Warn] --profiler-config was provided, but current engine does not support profiler controls.")
print("[Warn] VLLM_TORCH_PROFILER_DIR is set, but current engine does not support profiler controls.")
omni_generator = omni.generate(prompts, sampling_params_list, py_generator=args.py_generator)
# Determine output directory: prefer --output-dir; fallback to --output-wav
@@ -460,7 +458,6 @@ def parse_args():
default=False,
help="Enable writing detailed statistics (default: disabled)",
)
add_profiler_config_arg(parser)
parser.add_argument(
"--stage-init-timeout",
type=int,
@@ -14,7 +14,6 @@ from vllm_omni.entrypoints.omni import Omni
from vllm_omni.model_executor.models.cosyvoice3.tokenizer import get_qwen_tokenizer
from vllm_omni.model_executor.models.cosyvoice3.utils import extract_text_token
from vllm_omni.transformers_utils.configs.cosyvoice3 import CosyVoice3Config
from vllm_omni.utils.profiler import add_profiler_config_arg
# Upstream zero-shot reference clip
ZERO_SHOT_PROMPT_URL = "https://raw.githubusercontent.com/FunAudioLLM/CosyVoice/main/asset/zero_shot_prompt.wav"
@@ -46,7 +45,6 @@ def run_e2e():
help="Override the deploy config path. If unset, auto-loads "
"vllm_omni/deploy/cosyvoice3.yaml based on the HF model_type.",
)
add_profiler_config_arg(parser)
parser.add_argument("--text", type=str, default="Hello, this is a test of the CosyVoice system capability.")
parser.add_argument(
"--prompt-text",
@@ -81,7 +79,6 @@ def run_e2e():
deploy_config=args.deploy_config,
tokenizer=args.tokenizer,
log_stats=True,
profiler_config=args.profiler_config,
)
sampling_cfg = {"top_p": 0.8, "top_k": 25, "eos_token_id": 6561 + 1}
@@ -152,8 +149,8 @@ def run_e2e():
sampling_params_list = [gpt_sampling, s2mel_sampling]
profiler_enabled = args.profiler_config is not None
if profiler_enabled:
# Start profiling (requires VLLM_TORCH_PROFILER_DIR env var)
if os.environ.get("VLLM_TORCH_PROFILER_DIR"):
print("Starting profiler...")
omni.start_profile()
@@ -161,7 +158,7 @@ def run_e2e():
outputs = list(omni.generate(prompts, sampling_params_list=sampling_params_list[:2]))
# Stop profiling and get results
if profiler_enabled:
if os.environ.get("VLLM_TORCH_PROFILER_DIR"):
print("Stopping profiler...")
profile_results = omni.stop_profile()
print(f"Profile traces saved to: {profile_results}")
@@ -614,6 +614,54 @@ def test_initialize_model_sets_current_diffusion_config_during_model_constructio
assert get_current_diffusion_config_or_none() is None
def test_configure_vae_stack_tiling_enables_tiling_and_native_capability():
from vllm_omni.diffusion.registry import _configure_vae_stack_tiling
calls = []
vae = SimpleNamespace(
supports_stack_tiling=True,
set_stack_tiling=lambda mode: calls.append(mode),
)
model = SimpleNamespace(vae=vae)
od_config = SimpleNamespace(
model_class_name="MiniMaxH3Pipeline",
vae_stack_tiling="auto",
vae_use_tiling=False,
)
_configure_vae_stack_tiling(model, od_config)
assert od_config.vae_use_tiling is True
assert calls == ["auto"]
def test_configure_vae_stack_tiling_rejects_unsupported_vae():
from vllm_omni.diffusion.registry import _configure_vae_stack_tiling
od_config = SimpleNamespace(
model_class_name="UnsupportedPipeline",
vae_stack_tiling="true",
vae_use_tiling=False,
)
with pytest.raises(ValueError, match="does not declare stacked-tiling support"):
_configure_vae_stack_tiling(SimpleNamespace(vae=SimpleNamespace()), od_config)
def test_configure_vae_stack_tiling_auto_falls_back_for_unsupported_vae():
from vllm_omni.diffusion.registry import _configure_vae_stack_tiling
od_config = SimpleNamespace(
model_class_name="UnsupportedPipeline",
vae_stack_tiling="auto",
vae_use_tiling=False,
)
_configure_vae_stack_tiling(SimpleNamespace(vae=SimpleNamespace()), od_config)
assert od_config.vae_use_tiling is False
def test_load_model_custom_pipeline_sets_current_diffusion_config(monkeypatch):
import vllm_omni.diffusion.model_loader.diffusers_loader as loader_mod
@@ -30,6 +30,7 @@ class _FakeCheckpointModel:
def __init__(self):
self.vae_ratio = RATIO
self.parallel_tiling = True
self.stack_tiling = False
self.decoded_with = []
def split_tiles(self, input_len, is_decoder=False):
@@ -57,6 +58,7 @@ def _vae(parallel_size, state):
vae.model = _FakeCheckpointModel()
vae.remote = remote
vae.parallel_size = parallel_size
vae.stack_tiling_mode = "false"
return vae
@@ -89,6 +91,33 @@ def test_tile_count_uses_the_decoder_grid_not_the_encoder_one(h, w, decoder_tile
assert as_encoder == encoder_tiles != decoder_tiles
@pytest.mark.parametrize(
("parallel_size", "h", "w", "expected"),
[
(1, 48, 84, True),
(2, 48, 84, True),
(4, 48, 84, True),
(8, 48, 84, True),
(16, 48, 84, False),
(4, 16, 16, False),
],
)
def test_auto_stack_tiling_requires_multiple_tiles_per_rank(parallel_size, h, w, expected):
vae = _vae(parallel_size, {})
vae.set_stack_tiling("auto")
assert MiniMaxH3VideoVAE._should_stack_decode_tiles(vae, _latent(h, w)) is expected
@pytest.mark.parametrize(("mode", "expected"), [(False, False), (True, True), ("false", False), ("true", True)])
def test_stack_tiling_configuration_accepts_bool_and_string_modes(mode, expected):
vae = _vae(1, {})
vae.set_stack_tiling(mode)
assert vae.model.stack_tiling is expected
def test_rank_local_tiling_restores_the_group_state():
state = {"sp_size": 4, "sp_rank": 2, "sp_enabled": True, "sp_process_group": "pg"}
vae = _vae(4, state)
@@ -156,6 +185,7 @@ def _dispatch_vae(parallel_size):
seen["sp_enabled"] = state["sp_enabled"]
seen["sp_process_group"] = state["sp_process_group"]
seen["parallel_tiling"] = vae.model.parallel_tiling
seen["stack_tiling"] = vae.model.stack_tiling
# the guard never inspects the decoded shape, so keep it small
return torch.zeros(1, 3, 2, 4, 4)
@@ -212,3 +242,15 @@ def test_decode_latent_leaves_the_group_state_alone_when_the_grid_is_large_enoug
assert seen["sp_enabled"] is True
assert seen["sp_process_group"] is state["sp_process_group"]
assert seen["parallel_tiling"] is True
def test_decode_latent_restores_native_stack_tiling_state():
vae, _, seen = _dispatch_vae(4)
vae.set_stack_tiling("auto")
# The adapter must not leak its per-request decision into the remote VAE.
vae.model.stack_tiling = True
MiniMaxH3VideoVAE.decode_latent(vae, _latent(16, 16))
assert seen["stack_tiling"] is False
assert vae.model.stack_tiling is True
@@ -1,23 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from pathlib import Path
import pytest
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
_REPO_ROOT = Path(__file__).resolve().parents[2]
_EXAMPLES = (
"examples/offline_inference/text_to_speech/cosyvoice3/end2end.py",
"examples/offline_inference/qwen2_5_omni/end2end.py",
)
@pytest.mark.parametrize("relative_path", _EXAMPLES)
def test_offline_examples_use_typed_profiler_config(relative_path: str):
source = (_REPO_ROOT / relative_path).read_text(encoding="utf-8")
assert "add_profiler_config_arg(parser)" in source
assert "profiler_enabled = args.profiler_config is not None" in source
assert "VLLM_TORCH_PROFILER_DIR" not in source
@@ -395,6 +395,28 @@ def test_serve_cli_accepts_diffusion_pipeline_profiler_flag():
assert stage_cfg["engine_args"]["enable_diffusion_pipeline_profiler"] is True
def test_serve_cli_forwards_vae_stack_tiling():
parser = TrackingArgumentParser()
subparsers = parser.add_subparsers(dest="command")
OmniServeCommand().subparser_init(subparsers)
args = parser.parse_args(
[
"serve",
"MiniMaxAI/MiniMax-H3",
"--omni",
"--vae-stack-tiling",
"auto",
]
)
explicit_kwargs = args.get_explicit_kwargs_dict()
stage_cfg = AsyncOmniEngine._create_default_diffusion_stage_cfg(explicit_kwargs)[0]
assert args.vae_stack_tiling == "auto"
assert stage_cfg["engine_args"]["vae_stack_tiling"] == "auto"
def test_serve_cli_forwards_distilled_lora_to_diffusion_stage():
"""Ensure startup distilled LoRA options reach the online diffusion stage."""
parser = TrackingArgumentParser()
+1
View File
@@ -711,6 +711,7 @@ class _DiffusionConfigProjection:
fa_deterministic: bool = False
vae_use_slicing: bool = False
vae_use_tiling: bool = False
vae_stack_tiling: Literal["auto", "true", "false"] = "false"
mask_strategy_file_path: str | None = None
skip_time_steps: int = 15
VSA_sparsity: float = 0.0
+2 -1
View File
@@ -11,7 +11,7 @@ from collections.abc import Callable, Mapping
from dataclasses import asdict, dataclass, field, fields
from enum import Enum
from pathlib import Path
from typing import Any, NamedTuple
from typing import Any, Literal, NamedTuple
from transformers import PretrainedConfig
from vllm.logger import init_logger
@@ -462,6 +462,7 @@ class StageDeployConfig:
step_execution: bool | None = None
vae_use_slicing: bool | None = None
vae_use_tiling: bool | None = None
vae_stack_tiling: Literal["auto", "true", "false"] | None = None
boundary_ratio: float | None = None
flow_shift: float | None = None
diffusion_kv_cache_dtype: str | None = None
+2 -1
View File
@@ -8,7 +8,7 @@ import random
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field, fields
from enum import Enum
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Literal
import diffusers
import huggingface_hub
@@ -831,6 +831,7 @@ class OmniDiffusionConfig:
# VAE memory optimization parameters
vae_use_slicing: bool = False
vae_use_tiling: bool = False
vae_stack_tiling: Literal["auto", "true", "false"] = "false"
# STA (Sliding Tile Attention) parameters
mask_strategy_file_path: str | None = None
+39 -2
View File
@@ -9,7 +9,7 @@ import json
from collections.abc import Iterator
from contextlib import AbstractContextManager, contextmanager, nullcontext
from pathlib import Path
from typing import Any
from typing import Any, ClassVar
import torch
import torch.distributed as dist
@@ -115,6 +115,8 @@ class _AudioVAEDeterminismContext(AbstractContextManager):
class MiniMaxH3VideoVAE(nn.Module, DistributedVaeMixin):
"""Adapter around the checkpoint's native parallel-tiled video VAE."""
supports_stack_tiling: ClassVar[bool] = True
def __init__(
self,
component_path: str,
@@ -153,6 +155,30 @@ class MiniMaxH3VideoVAE(nn.Module, DistributedVaeMixin):
self.use_slicing = False
self.parallel_size = 1
self.device_module = torch.get_device_module()
self.stack_tiling_mode = "false"
def set_stack_tiling(self, mode: str | bool) -> None:
"""Configure native decoder tile batching.
``auto`` is resolved per decode because the useful batch size depends
on both the request's tile grid and the VAE parallel group size.
"""
if not hasattr(self.model, "stack_tiling"):
raise RuntimeError("MiniMax H3 video VAE remote model does not expose stack_tiling")
if isinstance(mode, bool):
mode = "true" if mode else "false"
if mode not in ("auto", "true", "false"):
raise ValueError(f"invalid VAE stacked-tiling mode: {mode!r}")
self.stack_tiling_mode = mode
self.model.stack_tiling = mode == "true"
def _should_stack_decode_tiles(self, latent: torch.Tensor) -> bool:
if self.stack_tiling_mode != "auto":
return self.stack_tiling_mode == "true"
# Stacking is worthwhile when every rank has more than one tile in
# the common round-robin distribution. The tile grid is global and
# therefore identical on all ranks.
return self._decoder_tile_count(latent) >= 2 * self.parallel_size
def load_to_device(self) -> None:
if self._stager is not None:
@@ -388,7 +414,18 @@ class MiniMaxH3VideoVAE(nn.Module, DistributedVaeMixin):
tiling_context = nullcontext()
with tiling_context:
decoded = self.model.decode_base(latent * std + mean)
stack_tiling_mode = getattr(self, "stack_tiling_mode", "false")
if stack_tiling_mode == "false" and not hasattr(self.model, "stack_tiling"):
# Keep the default path compatible with older H3 snapshots
# that predate the native stacked-tiling attribute.
decoded = self.model.decode_base(latent * std + mean)
else:
previous_stack_tiling = self.model.stack_tiling
self.model.stack_tiling = self._should_stack_decode_tiles(latent)
try:
decoded = self.model.decode_base(latent * std + mean)
finally:
self.model.stack_tiling = previous_stack_tiling
frames = self.model.processor.revert_tensor(decoded)
if frames.ndim == 4:
frames = frames.unsqueeze(0).transpose(1, 2)
+34
View File
@@ -387,6 +387,38 @@ def _prepare_diffusion_quant_config(
configure_quant_config(quant_config, model_class)
def _configure_vae_stack_tiling(model: nn.Module, od_config: OmniDiffusionConfig) -> None:
"""Configure optional native batching of VAE decoder tiles."""
mode = getattr(od_config, "vae_stack_tiling", "false")
if isinstance(mode, bool):
mode = "true" if mode else "false"
if mode not in ("auto", "true", "false"):
raise ValueError(f"invalid vae_stack_tiling mode: {mode!r}")
if mode == "false":
return
vae = getattr(model, "vae", None)
if vae is None or not getattr(vae, "supports_stack_tiling", False):
if mode == "auto":
logger.info(
"VAE stacked tiling auto mode is unavailable for %s; using sequential tiles.",
od_config.model_class_name,
)
return
raise ValueError(
f"vae_stack_tiling was requested, but {od_config.model_class_name} does not declare stacked-tiling support"
)
setter = getattr(vae, "set_stack_tiling", None)
if not callable(setter):
raise ValueError("vae_stack_tiling was requested, but the VAE does not provide set_stack_tiling()")
if not od_config.vae_use_tiling:
logger.info("vae_stack_tiling requires vae_use_tiling; automatically enabling it.")
od_config.vae_use_tiling = True
setter(mode)
logger.info("VAE stacked tiling mode set to %s for %s.", mode, od_config.model_class_name)
def initialize_model(
od_config: OmniDiffusionConfig,
) -> nn.Module:
@@ -428,6 +460,8 @@ def initialize_model(
)
od_config.vae_use_tiling = True
_configure_vae_stack_tiling(model, od_config)
# Configure VAE memory optimization settings from config
if hasattr(model, "vae") and hasattr(model.vae, "use_slicing"):
model.vae.use_slicing = od_config.vae_use_slicing
+2 -1
View File
@@ -7,7 +7,7 @@ import os
import tempfile
from collections.abc import Callable
from dataclasses import dataclass, field, fields
from typing import Any, cast
from typing import Any, Literal, cast
from vllm.engine.arg_utils import AsyncEngineArgs, EngineArgs
from vllm.logger import init_logger
@@ -566,6 +566,7 @@ class OrchestratorArgs:
step_execution: bool = False
vae_use_slicing: bool = False
vae_use_tiling: bool = False
vae_stack_tiling: Literal["auto", "true", "false"] = "false"
enable_multithread_weight_load: bool = True
num_weight_load_threads: int = 4
enable_cpu_offload: bool = False
+1
View File
@@ -1132,6 +1132,7 @@ class AsyncOmniEngine:
"request_batch_max_wait_ms": kwargs.get("request_batch_max_wait_ms", 0.0),
"vae_use_slicing": kwargs.get("vae_use_slicing", False),
"vae_use_tiling": kwargs.get("vae_use_tiling", False),
"vae_stack_tiling": kwargs.get("vae_stack_tiling", "false"),
"cache_backend": cache_backend,
"cache_config": cache_config,
"enable_cache_dit_summary": kwargs.get("enable_cache_dit_summary", False),
+9
View File
@@ -686,6 +686,15 @@ class OmniServeCommand(CLISubcommand):
action="store_true",
help="Enable VAE tiling for memory optimization (useful for mitigating OOM issues).",
)
omni_config_group.add_argument(
"--vae-stack-tiling",
choices=("auto", "true", "false"),
default="false",
help=(
"Batch native VAE tiles into fewer model calls. 'auto' enables it "
"when each VAE-parallel rank has multiple tiles (default: false)."
),
)
# Parallel weight loading (faster diffusion startup)
omni_config_group.add_argument(
-37
View File
@@ -1,37 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from __future__ import annotations
import argparse
import json
from typing import Any
def parse_json_object(value: str, flag_name: str = "argument") -> dict[str, Any]:
"""Parse a CLI value as a JSON object, attributing errors to ``flag_name``."""
try:
config = json.loads(value)
except json.JSONDecodeError as e:
raise argparse.ArgumentTypeError(f"{flag_name} must be valid JSON: {e}") from e
if not isinstance(config, dict):
raise argparse.ArgumentTypeError(f"{flag_name} must be a JSON object")
return config
def parse_profiler_config(value: str) -> dict[str, Any]:
"""Parse the JSON object passed to ``--profiler-config``."""
return parse_json_object(value, flag_name="--profiler-config")
def add_profiler_config_arg(parser: argparse.ArgumentParser) -> None:
"""Add the shared JSON profiler configuration option to ``parser``."""
parser.add_argument(
"--profiler-config",
type=parse_profiler_config,
default=None,
help=(
"JSON profiler config for torch/cuda profiling, e.g. "
'\'{"profiler":"torch","torch_profiler_dir":"./perf"}\'.'
),
)