Compare commits

...
Author SHA1 Message Date
Hongsheng Liu 1cdaddd79b [Perf][Diffusion] Compile and prewarm VAE decode
Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>
2026-08-11 07:14:35 +08:00
fan2956andFan c27623c29c [Bugfix] rainfusion support all shape video (#6000)
Signed-off-by: Fan <fan@FandeMacBook-Air.local>
Co-authored-by: Fan <fan@FandeMacBook-Air.local>
2026-08-11 05:10:50 +08:00
Yueqian Lin f8fa86fb1f [BugFix] Don't fail the TTS ratchet when the branch count goes down (#6008)
Signed-off-by: Yueqian Lin <linyueqian@outlook.com>
2026-08-11 05:08:31 +08:00
ooooooyeandHongsheng Liu 422d8fd8f9 [Bugfix][Diffusion] Restore supports_packed_mask_free on teacache FakeBackend (#5997)
Signed-off-by: brandneway <gyuan4892@gmail.com>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
2026-08-10 22:20:37 +08:00
Hongsheng Liu 62e239bf1e [Docs] Fix broken attention backend link (#5998)
Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>
2026-08-10 21:55:01 +08:00
22 changed files with 400 additions and 75 deletions

No files matched your search

+4 -1
View File
@@ -66,7 +66,10 @@ repos:
entry: python tools/pre_commit/check_tts_adapter.py entry: python tools/pre_commit/check_tts_adapter.py
language: python language: python
pass_filenames: false pass_filenames: false
files: ^vllm_omni/entrypoints/openai/(serving_speech\.py|tts_adapters/__init__\.py)$ files: ^(vllm_omni/entrypoints/openai/(serving_speech\.py|tts_adapters/__init__\.py)|tools/pre_commit/check_tts_adapter\.py)$
# The budget-is-now-slack reminder is printed on a PASSING run, and
# pre-commit hides output from passing hooks unless the hook is verbose.
verbose: true
# Keep `suggestion` last # Keep `suggestion` last
- id: suggestion - id: suggestion
+12 -15
View File
@@ -283,23 +283,20 @@ blocks — use Ulysses SP (`ring_degree=1`).
### Which geometries run sparse ### Which geometries run sparse
Sparsity is only applied when the video segment is a **multiple of 128 rows**, where the row count MindIE-SD rf_v2 handles arbitrary video grids by spatially rearranging video first and, when
is `latent_t × (height / 32) × (width / 32)`. Otherwise `rf_v2`'s block mask and the kernel's own necessary, promoting a real-video suffix to its always-kept prefix segment. It never pads: rf_v2
tiling disagree on the block count, and the block straddling the seam mixes video and prefix rows does not consume an attention mask for padded keys. The remaining video rows stay sparse, while the
that selection may then drop. Any resolution still runs — an unaligned geometry falls back to dense promoted rows remain visible to every query and key.
attention and logs `RAINFUSION_ATTN staying dense` with the row count it computed — but it gets no
speedup, so pick an aligned geometry when you want one.
Alignment is necessary, not sufficient, for good quality. `rf_v2` groups video positions into 8x8 For example, MiniMax-H3 at 1344x768 has grid `(62, 24, 42)`: its 62496 video rows leave a 32-row
spatial tiles (two tiles fill one 128-row block), which is what makes a selected block a compact 128-block residue, and its two non-8-wide latent columns form a 2928-row rearranged suffix.
patch of the frame. When the latent `h` or `w` is not a multiple of 8, the leftover rows or columns MindIE-SD promotes 2976 rows, leaving 59520 sparse rows (`465 × 128`). This keeps the block mask
are peeled off and appended as a flat run instead of being tiled, so they get pooled with spatially and kernel tiling aligned, at the cost of retaining more blocks.
distant positions and selection can no longer rank them meaningfully. This is silent, and invisible
to a `sparsity=0` check, because a fully populated mask does not care how blocks are grouped.
"Multiple of 8" on the latent grid means **width and height that are multiples of 256**. Aligned 8x8-aligned spatial grids remain preferable for performance. A latent `h` or `w` not divisible by 8
resolutions off that grid still run sparse; they just lose more fidelity at the same `sparsity`, creates an always-kept residual suffix and lowers realized sparsity. "Multiple of 8" on the latent
which you can buy back with `start_step`. grid means **width and height that are multiples of 256**. This path requires a MindIE-SD release
that implements protected video tails for rf_v2.
## End-to-End Benchmark (BF16, sm_120 RTX Pro 6000 Blackwell) ## End-to-End Benchmark (BF16, sm_120 RTX Pro 6000 Blackwell)
@@ -99,7 +99,7 @@ a native vLLM-Omni pipeline for those cases.
### Attention Backends ### Attention Backends
The diffusers backend converts The diffusers backend converts
[vLLM-Omni standard of attention backend setting](../../../docs/user_guide/diffusion/attention_backends.md) [vLLM-Omni standard of attention backend setting](../../diffusion/attention_backends.md)
to [diffusers standard](https://huggingface.co/docs/diffusers/optimization/attention_backends#available-backends). to [diffusers standard](https://huggingface.co/docs/diffusers/optimization/attention_backends#available-backends).
Specifically for `FLASH_ATTN`, it will first attempt to use FlashAttention-3 and then FlashAttention-2. Specifically for `FLASH_ATTN`, it will first attempt to use FlashAttention-3 and then FlashAttention-2.
+9
View File
@@ -751,6 +751,7 @@ def test_from_pipeline_config_routes_regional_compile_dynamic(tmp_path):
" - stage_id: 0", " - stage_id: 0",
" diffusion_compile_granularity: regional", " diffusion_compile_granularity: regional",
" diffusion_compile_dynamic: false", " diffusion_compile_dynamic: false",
" diffusion_compile_vae: false",
] ]
) )
) )
@@ -762,13 +763,16 @@ def test_from_pipeline_config_routes_regional_compile_dynamic(tmp_path):
cli_overrides={ cli_overrides={
"diffusion_compile_granularity": "full", "diffusion_compile_granularity": "full",
"diffusion_compile_dynamic": True, "diffusion_compile_dynamic": True,
"diffusion_compile_vae": True,
}, },
).stage_by_id(0) ).stage_by_id(0)
assert configured_stage.diffusion_config.diffusion_compile_granularity == "regional" assert configured_stage.diffusion_config.diffusion_compile_granularity == "regional"
assert configured_stage.diffusion_config.diffusion_compile_dynamic is False assert configured_stage.diffusion_config.diffusion_compile_dynamic is False
assert configured_stage.diffusion_config.diffusion_compile_vae is False
assert overridden_stage.diffusion_config.diffusion_compile_granularity == "full" assert overridden_stage.diffusion_config.diffusion_compile_granularity == "full"
assert overridden_stage.diffusion_config.diffusion_compile_dynamic is True assert overridden_stage.diffusion_config.diffusion_compile_dynamic is True
assert overridden_stage.diffusion_config.diffusion_compile_vae is True
def test_structured_diffusion_config_rejects_non_boolean_compile_dynamic(): def test_structured_diffusion_config_rejects_non_boolean_compile_dynamic():
@@ -776,6 +780,11 @@ def test_structured_diffusion_config_rejects_non_boolean_compile_dynamic():
omni_config_module._DiffusionConfigProjection(diffusion_compile_dynamic="false") omni_config_module._DiffusionConfigProjection(diffusion_compile_dynamic="false")
def test_structured_diffusion_config_rejects_non_boolean_compile_vae():
with pytest.raises(ValidationError, match="diffusion_compile_vae"):
omni_config_module._DiffusionConfigProjection(diffusion_compile_vae="false")
def test_structured_diffusion_config_rejects_invalid_compile_granularity(): def test_structured_diffusion_config_rejects_invalid_compile_granularity():
with pytest.raises(ValidationError, match="diffusion_compile_granularity"): with pytest.raises(ValidationError, match="diffusion_compile_granularity"):
omni_config_module._DiffusionConfigProjection(diffusion_compile_granularity="block") omni_config_module._DiffusionConfigProjection(diffusion_compile_granularity="block")
@@ -5,10 +5,9 @@
The grids here are the ones MiniMax-H3 FL2VA actually produces for an 8.7s clip: The grids here are the ones MiniMax-H3 FL2VA actually produces for an 8.7s clip:
1280x768 gives a 62x24x40 latent grid whose 59520 video rows land on the 128-row 1280x768 gives a 62x24x40 latent grid whose 59520 video rows land on the 128-row
kernel block, and 1344x768 gives 62x24x42 whose 62496 rows straddle it. Only the kernel block, and 1344x768 gives 62x24x42 whose 62496 rows do not. Both are
first can run sparse: rf_v2 tiles the sequence as a whole but pools its block handed to MindIE-SD: its rf_v2 preprocessing promotes an irregular video tail
mask from the video and prefix segments separately, so the two only agree on the to the always-kept prefix segment before generating the block mask.
block count when the video segment ends on a boundary.
""" """
import dataclasses import dataclasses
@@ -57,10 +56,13 @@ def test_block_aligned_video_segment_runs_sparse():
assert plan.latent_shape == list(ALIGNED_GRID) assert plan.latent_shape == list(ALIGNED_GRID)
def test_misaligned_video_segment_stays_dense(): def test_misaligned_video_segment_runs_sparse_with_updated_mindiesd():
# Realigning the sequence would mean padding it, and rf_v2 ignores attn_mask: plan = make_impl()._resolve_plan(make_metadata(MISALIGNED_GRID))
# pad keys would take a share of every real query's softmax denominator.
assert make_impl()._resolve_plan(make_metadata(MISALIGNED_GRID)) is None assert plan is not None
assert plan.prefix_len == PREFIX_ROWS
assert plan.used_len == PREFIX_ROWS + 62496
assert plan.latent_shape == list(MISALIGNED_GRID)
@pytest.mark.parametrize("prefix_len", [1, 127, _BLOCK_SIZE, 710, 900]) @pytest.mark.parametrize("prefix_len", [1, 127, _BLOCK_SIZE, 710, 900])
@@ -155,3 +157,28 @@ def test_fully_populated_mask_reproduces_dense_attention():
).transpose(1, 2) ).transpose(1, 2)
error = (out.float() - reference.float()).abs().mean() / reference.float().abs().mean() error = (out.float() - reference.float()).abs().mean() / reference.float().abs().mean()
assert error < 2e-3, f"mean relative error {error:.4%} against dense attention" assert error < 2e-3, f"mean relative error {error:.4%} against dense attention"
@pytest.mark.skipif(not current_omni_platform.is_npu(), reason="rf_v2 runs on Ascend NPU only.")
def test_irregular_video_tail_reproduces_dense_attention_with_updated_mindiesd():
"""vLLM forwards the full irregular grid; MindIE-SD owns its tail handling."""
torch.manual_seed(1)
prefix_len, grid = 384, (4, 16, 10)
video_len = grid[0] * grid[1] * grid[2]
used = prefix_len + video_len
heads, head_dim = 4, 128
q, k, v = (torch.randn(1, used, heads, head_dim, dtype=torch.bfloat16, device="npu") for _ in range(3))
impl = make_impl()
impl.rainfusion = dataclasses.replace(impl.rainfusion, sparsity=0.0)
plan = RainFusionPlan(prefix_len=prefix_len, used_len=used, latent_shape=list(grid))
out = impl._forward_sparse_npu(q, k, v, plan)
reference = torch.nn.functional.scaled_dot_product_attention(
q.transpose(1, 2),
k.transpose(1, 2),
v.transpose(1, 2),
scale=impl.softmax_scale,
).transpose(1, 2)
error = (out.float() - reference.float()).abs().mean() / reference.float().abs().mean()
assert error < 2e-3, f"mean relative error {error:.4%} against dense attention"
+8
View File
@@ -488,6 +488,14 @@ class _MiniMaxH3FakeAttention(nn.Module):
class FakeBackend: class FakeBackend:
supports_prefix_kv_slicing = False supports_prefix_kv_slicing = False
# Mirror AttentionBackend.supports_packed_mask_free: H3's packed
# attention path calls attn_backend.supports_packed_mask_free() to decide
# whether to build the padding mask. Return False so the fake exercises
# the masked path (the abstract backend default).
@classmethod
def supports_packed_mask_free(cls) -> bool:
return False
def __init__(self, **kwargs): def __init__(self, **kwargs):
del kwargs del kwargs
super().__init__() super().__init__()
+48 -1
View File
@@ -5,7 +5,7 @@ import pytest
import torch.nn as nn import torch.nn as nn
import vllm_omni.diffusion.compile as compile_module import vllm_omni.diffusion.compile as compile_module
from vllm_omni.diffusion.compile import regionally_compile from vllm_omni.diffusion.compile import compile_callable_with_fallback, regionally_compile
pytestmark = [pytest.mark.core_model, pytest.mark.diffusion, pytest.mark.cpu] pytestmark = [pytest.mark.core_model, pytest.mark.diffusion, pytest.mark.cpu]
@@ -74,3 +74,50 @@ def test_regionally_compile_does_not_partially_mutate_on_setup_failure(monkeypat
regionally_compile(model, dynamic=True) regionally_compile(model, dynamic=True)
assert [block.forward.__func__ for block in model.transformer_blocks] == original_forwards assert [block.forward.__func__ for block in model.transformer_blocks] == original_forwards
def test_compile_callable_falls_back_once_after_lazy_failure(monkeypatch, caplog):
compiled_calls = 0
eager_calls = 0
def eager(value):
nonlocal eager_calls
eager_calls += 1
return value + 1
def compile_fn(fn, **kwargs):
assert fn is eager
assert kwargs == {"mode": "default"}
def compiled(value):
nonlocal compiled_calls
compiled_calls += 1
raise RuntimeError("lazy failure")
return compiled
monkeypatch.setattr(compile_module.torch, "compile", compile_fn)
callable_fn, activated = compile_callable_with_fallback(eager, label="test callable", mode="default")
assert activated
assert callable_fn(1) == 2
assert callable_fn(2) == 3
assert compiled_calls == 1
assert eager_calls == 2
assert "disabling compilation and retrying eagerly" in caplog.text
def test_compile_callable_keeps_eager_on_setup_failure(monkeypatch, caplog):
def eager(value):
return value + 1
def compile_fn(*args, **kwargs):
raise RuntimeError("setup failure")
monkeypatch.setattr(compile_module.torch, "compile", compile_fn)
callable_fn, activated = compile_callable_with_fallback(eager, label="test callable")
assert not activated
assert callable_fn is eager
assert callable_fn(1) == 2
assert "continuing eagerly" in caplog.text
+7
View File
@@ -515,16 +515,19 @@ class TestDiffusionCompileConfig:
assert config.diffusion_compile_granularity == "regional" assert config.diffusion_compile_granularity == "regional"
assert config.diffusion_compile_dynamic is True assert config.diffusion_compile_dynamic is True
assert config.diffusion_compile_vae is True
def test_from_kwargs_preserves_compile_controls(self) -> None: def test_from_kwargs_preserves_compile_controls(self) -> None:
config = OmniDiffusionConfig.from_kwargs( config = OmniDiffusionConfig.from_kwargs(
model="test", model="test",
diffusion_compile_granularity="full", diffusion_compile_granularity="full",
diffusion_compile_dynamic=False, diffusion_compile_dynamic=False,
diffusion_compile_vae=False,
) )
assert config.diffusion_compile_granularity == "full" assert config.diffusion_compile_granularity == "full"
assert config.diffusion_compile_dynamic is False assert config.diffusion_compile_dynamic is False
assert config.diffusion_compile_vae is False
def test_config_rejects_invalid_compile_granularity(self) -> None: def test_config_rejects_invalid_compile_granularity(self) -> None:
with pytest.raises(ValueError, match="diffusion_compile_granularity"): with pytest.raises(ValueError, match="diffusion_compile_granularity"):
@@ -534,6 +537,10 @@ class TestDiffusionCompileConfig:
with pytest.raises(TypeError, match="diffusion_compile_dynamic"): with pytest.raises(TypeError, match="diffusion_compile_dynamic"):
OmniDiffusionConfig(model="test", diffusion_compile_dynamic="false") OmniDiffusionConfig(model="test", diffusion_compile_dynamic="false")
def test_config_rejects_non_boolean_compile_vae(self) -> None:
with pytest.raises(TypeError, match="diffusion_compile_vae"):
OmniDiffusionConfig(model="test", diffusion_compile_vae="false")
@pytest.mark.parametrize( @pytest.mark.parametrize(
"kwargs, feature", "kwargs, feature",
[ [
@@ -130,6 +130,23 @@ class _CompileTrackingModel:
return self return self
class _DecodeVAE(torch.nn.Module):
def decode(self, value):
return value + 1
class _PipelineWithVAE(torch.nn.Module):
_dit_modules = []
_encoder_modules = []
_vae_modules = ["vae", "vae_alias"]
_resident_modules = []
def __init__(self):
super().__init__()
self.vae = _DecodeVAE()
self.vae_alias = self.vae
def _make_request(): def _make_request():
sampling_params = SimpleNamespace( sampling_params = SimpleNamespace(
generator=None, generator=None,
@@ -320,6 +337,31 @@ def test_compile_transformer_falls_back_after_synchronous_setup_failure(monkeypa
assert "lazy compilation errors" in caplog.text assert "lazy compilation errors" in caplog.text
@pytest.mark.core_model
@pytest.mark.cpu
def test_compile_vae_decodes_uses_discovery_and_deduplicates_aliases(monkeypatch):
runner = object.__new__(DiffusionModelRunner)
runner.pipeline = _PipelineWithVAE()
compile_calls = []
def compile_fn(fn, **kwargs):
compile_calls.append((fn, kwargs))
return lambda *args, **call_kwargs: fn(*args, **call_kwargs) + 10
monkeypatch.setattr(model_runner_module.torch, "compile", compile_fn)
DiffusionModelRunner._compile_vae_decodes(runner)
assert len(compile_calls) == 1
assert compile_calls[0][1] == {
"mode": "default",
"fullgraph": False,
"dynamic": None,
}
assert runner.pipeline.vae.decode(1) == 12
assert runner.pipeline.vae_alias.decode is runner.pipeline.vae.decode
@pytest.mark.core_model @pytest.mark.core_model
@pytest.mark.cpu @pytest.mark.cpu
def test_execute_stepwise_streaming_returns_chunks_at_boundaries(monkeypatch): def test_execute_stepwise_streaming_returns_chunks_at_boundaries(monkeypatch):
@@ -971,6 +971,44 @@ class TestDiffusionEngine:
admitted_request = engine.add_req_and_wait_for_response.call_args.args[0] admitted_request = engine.add_req_and_wait_for_response.call_args.args[0]
assert admitted_request.request_id == "dummy_req_id" assert admitted_request.request_id == "dummy_req_id"
def test_compile_warmup_repeats_through_normal_request_path(self, mocker: MockerFixture) -> None:
engine = DiffusionEngine.__new__(DiffusionEngine)
engine.od_config = SimpleNamespace(
model_class_name="mock_model",
diffusion_load_format="default",
diffusion_compile_vae=True,
enforce_eager=False,
)
engine.add_req_and_wait_for_response = mocker.Mock(
side_effect=[DiffusionOutput(output=None), DiffusionOutput(output=None)]
)
engine._dummy_run()
requests = [call.args[0] for call in engine.add_req_and_wait_for_response.call_args_list]
assert len(requests) == 2
assert requests[0].is_dummy_run()
assert not requests[1].is_dummy_run()
assert requests[1].prompt == {"prompt": "A detailed image."}
assert requests[1].sampling_params.need_kv_receive is False
assert requests[1].sampling_params.num_inference_steps == 1
def test_compile_real_path_warmup_failure_is_non_fatal(self, mocker: MockerFixture, caplog) -> None:
engine = DiffusionEngine.__new__(DiffusionEngine)
engine.od_config = SimpleNamespace(
model_class_name="mock_model",
diffusion_load_format="default",
diffusion_compile_vae=True,
enforce_eager=False,
)
engine.add_req_and_wait_for_response = mocker.Mock(
side_effect=[DiffusionOutput(output=None), RuntimeError("prewarm failure")]
)
engine._dummy_run()
assert "serving will continue" in caplog.text
class TestStepScheduler: class TestStepScheduler:
def setup_method(self) -> None: def setup_method(self) -> None:
@@ -87,6 +87,7 @@ def test_default_stage_config_defaults_nullified_parallel_size_kwargs():
"enforce_eager": None, "enforce_eager": None,
"diffusion_compile_granularity": None, "diffusion_compile_granularity": None,
"diffusion_compile_dynamic": None, "diffusion_compile_dynamic": None,
"diffusion_compile_vae": None,
} }
)[0] )[0]
@@ -98,6 +99,7 @@ def test_default_stage_config_defaults_nullified_parallel_size_kwargs():
assert stage_cfg["engine_args"]["enforce_eager"] is False assert stage_cfg["engine_args"]["enforce_eager"] is False
assert stage_cfg["engine_args"]["diffusion_compile_granularity"] == "regional" assert stage_cfg["engine_args"]["diffusion_compile_granularity"] == "regional"
assert stage_cfg["engine_args"]["diffusion_compile_dynamic"] is True assert stage_cfg["engine_args"]["diffusion_compile_dynamic"] is True
assert stage_cfg["engine_args"]["diffusion_compile_vae"] is True
def test_default_stage_config_propagates_ulysses_mode(): def test_default_stage_config_propagates_ulysses_mode():
@@ -315,7 +317,7 @@ def test_serve_cli_forwards_distributed_offload_residency():
def test_serve_cli_accepts_diffusion_compile_controls(): def test_serve_cli_accepts_diffusion_compile_controls():
"""Ensure both compile controls reach the diffusion stage.""" """Ensure all compile controls reach the diffusion stage."""
parser = TrackingArgumentParser() parser = TrackingArgumentParser()
subparsers = parser.add_subparsers(dest="command") subparsers = parser.add_subparsers(dest="command")
OmniServeCommand().subparser_init(subparsers) OmniServeCommand().subparser_init(subparsers)
@@ -328,6 +330,7 @@ def test_serve_cli_accepts_diffusion_compile_controls():
"--diffusion-compile-granularity", "--diffusion-compile-granularity",
"full", "full",
"--no-diffusion-compile-dynamic", "--no-diffusion-compile-dynamic",
"--no-diffusion-compile-vae",
] ]
) )
@@ -336,8 +339,10 @@ def test_serve_cli_accepts_diffusion_compile_controls():
assert args.diffusion_compile_granularity == "full" assert args.diffusion_compile_granularity == "full"
assert args.diffusion_compile_dynamic is False assert args.diffusion_compile_dynamic is False
assert args.diffusion_compile_vae is False
assert stage_cfg["engine_args"]["diffusion_compile_granularity"] == "full" assert stage_cfg["engine_args"]["diffusion_compile_granularity"] == "full"
assert stage_cfg["engine_args"]["diffusion_compile_dynamic"] is False assert stage_cfg["engine_args"]["diffusion_compile_dynamic"] is False
assert stage_cfg["engine_args"]["diffusion_compile_vae"] is False
def test_serve_cli_accepts_diffusion_attention_backend(): def test_serve_cli_accepts_diffusion_attention_backend():
+22 -6
View File
@@ -123,18 +123,29 @@ def _legacy_detectors(tree: ast.AST) -> list[tuple[int, str]]:
return found return found
def _check(label: str, path: Path, actual: int, budget: int, sample: list[str], budget_name: str) -> list[str]: def _check(
label: str,
path: Path,
actual: int,
budget: int,
sample: list[str],
budget_name: str,
notices: list[str],
) -> list[str]:
if actual > budget: if actual > budget:
listing = "\n".join(f" {line}" for line in sample) listing = "\n".join(f" {line}" for line in sample)
return [ return [
f"{path}: {actual} {label} exceeds the budget of {budget}.\n{listing}\n\n{_GUIDANCE}\n", f"{path}: {actual} {label} exceeds the budget of {budget}.\n{listing}\n\n{_GUIDANCE}\n",
] ]
if actual < budget: if actual < budget:
return [ # Progress is never a build failure. Removing branches is the goal, so a
f"{path}: {actual} {label}, down from a budget of {budget}. " # PR that does it must not hand-edit this file to stay green -- nor break
f"Lower {budget_name} to {actual} in {Path(__file__).relative_to(REPO_ROOT)} " # `main` for everyone else when it merges from a stale base.
f"so the ratchet holds the new ground.", notices.append(
] f"{path}: {actual} {label}, below the budget of {budget}. "
f"Lower {budget_name} to {actual} in "
f"{Path(__file__).relative_to(REPO_ROOT)} to hold the new ground."
)
return [] return []
@@ -152,6 +163,7 @@ def main(argv: list[str] | None = None) -> int:
print(f"check_tts_adapter: expected file not found: {path}", file=sys.stderr) print(f"check_tts_adapter: expected file not found: {path}", file=sys.stderr)
return 1 return 1
notices: list[str] = []
branches = _model_type_branches(ast.parse(serving_path.read_text())) branches = _model_type_branches(ast.parse(serving_path.read_text()))
detectors = _legacy_detectors(ast.parse(adapters_path.read_text())) detectors = _legacy_detectors(ast.parse(adapters_path.read_text()))
@@ -162,6 +174,7 @@ def main(argv: list[str] | None = None) -> int:
MAX_MODEL_TYPE_BRANCHES, MAX_MODEL_TYPE_BRANCHES,
[f"line {lineno}: {src}" for lineno, src in branches], [f"line {lineno}: {src}" for lineno, src in branches],
"MAX_MODEL_TYPE_BRANCHES", "MAX_MODEL_TYPE_BRANCHES",
notices,
) )
errors += _check( errors += _check(
"unmigrated model types in LEGACY_TTS_DETECTORS", "unmigrated model types in LEGACY_TTS_DETECTORS",
@@ -170,8 +183,11 @@ def main(argv: list[str] | None = None) -> int:
MAX_LEGACY_DETECTORS, MAX_LEGACY_DETECTORS,
[f"line {lineno}: {name}" for lineno, name in detectors], [f"line {lineno}: {name}" for lineno, name in detectors],
"MAX_LEGACY_DETECTORS", "MAX_LEGACY_DETECTORS",
notices,
) )
for notice in notices:
print(f"check_tts_adapter: {notice}", file=sys.stderr)
for error in errors: for error in errors:
print(f"check_tts_adapter: {error}", file=sys.stderr) print(f"check_tts_adapter: {error}", file=sys.stderr)
return 1 if errors else 0 return 1 if errors else 0
+1
View File
@@ -570,6 +570,7 @@ class _DiffusionConfigProjection:
pin_cpu_memory: bool = True pin_cpu_memory: bool = True
diffusion_compile_granularity: Literal["regional", "full"] = "regional" diffusion_compile_granularity: Literal["regional", "full"] = "regional"
diffusion_compile_dynamic: bool = Field(default=True, strict=True) diffusion_compile_dynamic: bool = Field(default=True, strict=True)
diffusion_compile_vae: bool = Field(default=True, strict=True)
fa_deterministic: bool = False fa_deterministic: bool = False
vae_use_slicing: bool = False vae_use_slicing: bool = False
vae_use_tiling: bool = False vae_use_tiling: bool = False
+1
View File
@@ -394,6 +394,7 @@ class StageDeployConfig:
# Diffusion execution, cache, and VAE behavior. # Diffusion execution, cache, and VAE behavior.
diffusion_compile_granularity: str | None = None diffusion_compile_granularity: str | None = None
diffusion_compile_dynamic: bool | None = None diffusion_compile_dynamic: bool | None = None
diffusion_compile_vae: bool | None = None
fa_deterministic: bool | None = None fa_deterministic: bool | None = None
cache_backend: str | None = None cache_backend: str | None = None
cache_config: dict[str, Any] | None = None cache_config: dict[str, Any] | None = None
@@ -128,9 +128,9 @@ class RainFusionAttentionImpl(AttentionImpl):
whose extent the model publishes as ``AttentionMetadata.video_layout``. Every whose extent the model publishes as ``AttentionMetadata.video_layout``. Every
other case — warmup denoise steps, exempt layers, a layer that does not declare other case — warmup denoise steps, exempt layers, a layer that does not declare
``qkv_layout="BSND"``, sequences without a published video segment, video ``qkv_layout="BSND"``, sequences without a published video segment, video
segments too short to pay for block selection or not aligned to the kernel's segments too short to pay for block selection — delegates to FlashAttention,
block size — delegates to FlashAttention, so a model can select this backend so a model can select this backend unconditionally. MindIE-SD handles an
unconditionally. irregular video tail internally, retaining it outside the sparse blocks.
""" """
def __init__( def __init__(
@@ -292,25 +292,6 @@ class RainFusionAttentionImpl(AttentionImpl):
_MIN_VIDEO_BLOCKS, _MIN_VIDEO_BLOCKS,
) )
return None return None
if video_len % _BLOCK_SIZE != 0:
# rf_v2 tiles the sequence into ceil((video + prefix) / 128) blocks but
# builds its block mask from the two segments pooled apart. The counts
# only agree when the video segment ends on a block boundary; otherwise
# a prefix block is forced on out of range, and the block straddling the
# seam holds real prefix rows that selection may drop. Padding the
# sequence to realign it is not an option because rf_v2 ignores
# attn_mask, so pad keys would take a share of every softmax denominator.
logger.warning_once(
"RAINFUSION_ATTN staying dense: latent grid %s gives %d video rows, which is not a "
"multiple of the %d-row kernel block. Choose a geometry whose latent_t * h * w is a "
"multiple of %d to enable sparsity.",
tuple(latent_shape),
video_len,
_BLOCK_SIZE,
_BLOCK_SIZE,
)
return None
logger.info_once( logger.info_once(
"RAINFUSION_ATTN active: sparsity=%.2f, start_step=%d, exempt_layers=%d, " "RAINFUSION_ATTN active: sparsity=%.2f, start_step=%d, exempt_layers=%d, "
"latent_grid=%s, prefix_rows=%d, video_rows=%d. Realized sparsity is lower than nominal " "latent_grid=%s, prefix_rows=%d, video_rows=%d. Realized sparsity is lower than nominal "
+54
View File
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: Apache-2.0 # SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from collections.abc import Callable
from typing import Any from typing import Any
import torch import torch
@@ -9,6 +10,59 @@ from vllm.logger import init_logger
logger = init_logger(__name__) logger = init_logger(__name__)
class _FallbackCompiledCallable:
"""Run a compiled callable until it fails, then permanently use eager."""
def __init__(
self,
eager: Callable[..., Any],
compiled: Callable[..., Any],
*,
label: str,
) -> None:
self._eager = eager
self._compiled: Callable[..., Any] | None = compiled
self._label = label
def __call__(self, *args: Any, **kwargs: Any) -> Any:
compiled = self._compiled
if compiled is None:
return self._eager(*args, **kwargs)
try:
return compiled(*args, **kwargs)
except Exception as exc:
self._compiled = None
logger.warning(
"%s failed during lazy torch.compile execution (%s); disabling compilation and retrying eagerly.",
self._label,
exc,
)
return self._eager(*args, **kwargs)
def compile_callable_with_fallback(
fn: Callable[..., Any],
*,
label: str,
**compile_kwargs: Any,
) -> tuple[Callable[..., Any], bool]:
"""Compile ``fn`` with a fail-closed eager fallback.
The boolean return value reports whether lazy compilation was activated.
Synchronous setup failures return the original callable unchanged.
"""
try:
compiled = torch.compile(fn, **compile_kwargs)
except Exception as exc:
logger.warning(
"%s torch.compile setup failed (%s); continuing eagerly.",
label,
exc,
)
return fn, False
return _FallbackCompiledCallable(fn, compiled, label=label), True
def _matches_repeated_block( def _matches_repeated_block(
name: str, name: str,
module: nn.Module, module: nn.Module,
+5
View File
@@ -726,6 +726,9 @@ class OmniDiffusionConfig:
# provide its own setup_compile() implementation. # provide its own setup_compile() implementation.
diffusion_compile_granularity: str = "regional" diffusion_compile_granularity: str = "regional"
diffusion_compile_dynamic: bool = True diffusion_compile_dynamic: bool = True
# Compile discovered VAE ``decode`` callables whenever generic diffusion
# torch.compile is enabled. Runtime failures permanently fall back to eager.
diffusion_compile_vae: bool = True
# Parallel weight loading (for faster diffusion model startup) # Parallel weight loading (for faster diffusion model startup)
enable_multithread_weight_load: bool = True enable_multithread_weight_load: bool = True
@@ -930,6 +933,8 @@ class OmniDiffusionConfig:
) )
if not isinstance(self.diffusion_compile_dynamic, bool): if not isinstance(self.diffusion_compile_dynamic, bool):
raise TypeError(f"diffusion_compile_dynamic must be a bool, got {type(self.diffusion_compile_dynamic)!r}") raise TypeError(f"diffusion_compile_dynamic must be a bool, got {type(self.diffusion_compile_dynamic)!r}")
if not isinstance(self.diffusion_compile_vae, bool):
raise TypeError(f"diffusion_compile_vae must be a bool, got {type(self.diffusion_compile_vae)!r}")
self.diffusion_kv_mode = parse_diffusion_kv_cache_mode(self.diffusion_kv_mode) self.diffusion_kv_mode = parse_diffusion_kv_cache_mode(self.diffusion_kv_mode)
if self.omni_kv_config is None: if self.omni_kv_config is None:
+51 -18
View File
@@ -56,6 +56,7 @@ if TYPE_CHECKING:
logger = init_logger(__name__) logger = init_logger(__name__)
_ASYNC_OUTPUT_TIMEOUT = 30.0 # seconds _ASYNC_OUTPUT_TIMEOUT = 30.0 # seconds
_VAE_COMPILE_WARMUP_REQUEST_ID = "vae_compile_warmup_req_id"
__all__ = [ __all__ = [
"DiffusionEngine", "DiffusionEngine",
@@ -851,28 +852,26 @@ class DiffusionEngine:
num_inference_steps = 1 num_inference_steps = 1
height = 512 height = 512
width = 512 width = 512
prompt: OmniTextPrompt = {"prompt": "dummy run"}
supports_image_input, supports_audio_input = supports_multimodal_input(self.od_config) supports_image_input, supports_audio_input = supports_multimodal_input(self.od_config)
if supports_image_input:
# Provide a dummy image input if the model supports it
color_format = image_color_format(self.od_config.model_class_name)
dummy_image = PIL.Image.new(color_format, (width, height))
prompt.setdefault("multi_modal_data", {})["image"] = dummy_image
if supports_audio_input:
audio_sr = 16000
dummy_audio = np.random.randn(audio_sr * 2).astype(np.float32)
prompt.setdefault("multi_modal_data", {})["audio"] = dummy_audio
num_frames = get_dummy_run_num_frames(self.od_config.model_class_name, supports_audio_input) num_frames = get_dummy_run_num_frames(self.od_config.model_class_name, supports_audio_input)
if num_frames <= 0: if num_frames <= 0:
logger.info("Skipping dummy warmup run (num_frames=0)") logger.info("Skipping dummy warmup run (num_frames=0)")
return return
req = OmniDiffusionRequest(
prompt=prompt, def build_request(request_id: str, prompt_text: str, *, real_path: bool) -> OmniDiffusionRequest:
request_id=DUMMY_DIFFUSION_REQUEST_ID, prompt: OmniTextPrompt = {"prompt": prompt_text}
sampling_params=OmniDiffusionSamplingParams( if supports_image_input:
color_format = image_color_format(self.od_config.model_class_name)
prompt.setdefault("multi_modal_data", {})["image"] = PIL.Image.new(
color_format,
(width, height),
)
if supports_audio_input:
audio_sr = 16000
prompt.setdefault("multi_modal_data", {})["audio"] = np.random.randn(audio_sr * 2).astype(np.float32)
sampling_params = OmniDiffusionSamplingParams(
height=height, height=height,
width=width, width=width,
num_inference_steps=num_inference_steps, num_inference_steps=num_inference_steps,
@@ -885,13 +884,47 @@ class DiffusionEngine:
# Disable CFG for warmup to avoid triggering CFG parallel # Disable CFG for warmup to avoid triggering CFG parallel
# validation when cfg_parallel_size > 1. # validation when cfg_parallel_size > 1.
extra_args={"cfg_text_scale": 1.0, "cfg_img_scale": 1.0}, extra_args={"cfg_text_scale": 1.0, "cfg_img_scale": 1.0},
), )
) if real_path:
# Internal pretraffic must exercise ordinary pipeline branches
# without waiting for KV produced by an upstream stage.
sampling_params.need_kv_receive = False
return OmniDiffusionRequest(
prompt=prompt,
request_id=request_id,
sampling_params=sampling_params,
)
req = build_request(DUMMY_DIFFUSION_REQUEST_ID, "dummy run", real_path=False)
logger.info("dummy run to warm up the model") logger.info("dummy run to warm up the model")
output = self.add_req_and_wait_for_response(req) output = self.add_req_and_wait_for_response(req)
if output.error: if output.error:
raise RuntimeError(f"Dummy run failed: {output.error}") raise RuntimeError(f"Dummy run failed: {output.error}")
compile_real_path = bool(
getattr(self.od_config, "diffusion_compile_vae", False)
and not getattr(self.od_config, "enforce_eager", True)
)
if not compile_real_path:
return
logger.info("real request-path warmup for compiled VAE decode")
real_path_req = build_request(
_VAE_COMPILE_WARMUP_REQUEST_ID,
"A detailed image.",
real_path=True,
)
try:
real_path_output = self.add_req_and_wait_for_response(real_path_req)
except Exception as exc:
logger.warning("Real request-path compile warmup failed (%s); serving will continue.", exc)
return
if real_path_output.error:
logger.warning(
"Real request-path compile warmup returned an error (%s); serving will continue.",
real_path_output.error,
)
def _submit_rpc( def _submit_rpc(
self, self,
method: str, method: str,
@@ -28,7 +28,7 @@ from vllm_omni.diffusion.cache.prompt_embed_cache import (
resolve_prompt_embed_cache_config, resolve_prompt_embed_cache_config,
) )
from vllm_omni.diffusion.cache.selector import get_cache_backend from vllm_omni.diffusion.cache.selector import get_cache_backend
from vllm_omni.diffusion.compile import regionally_compile from vllm_omni.diffusion.compile import compile_callable_with_fallback, regionally_compile
from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig
from vllm_omni.diffusion.forward_context import set_forward_context from vllm_omni.diffusion.forward_context import set_forward_context
from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader
@@ -40,6 +40,7 @@ from vllm_omni.diffusion.models.interface import (
supports_step_execution, supports_step_execution,
) )
from vllm_omni.diffusion.offloader import get_offload_backend from vllm_omni.diffusion.offloader import get_offload_backend
from vllm_omni.diffusion.offloader.module_collector import ModuleDiscovery
from vllm_omni.diffusion.registry import _NO_CACHE_ACCELERATION from vllm_omni.diffusion.registry import _NO_CACHE_ACCELERATION
from vllm_omni.diffusion.request import OmniDiffusionRequest from vllm_omni.diffusion.request import OmniDiffusionRequest
from vllm_omni.diffusion.sched.interface import DiffusionSchedulerOutput, KVPrefetchJob from vllm_omni.diffusion.sched.interface import DiffusionSchedulerOutput, KVPrefetchJob
@@ -188,6 +189,39 @@ class DiffusionModelRunner(OmniConnectorModelRunnerMixin):
compile_dynamic, compile_dynamic,
) )
def _compile_vae_decodes(self) -> None:
"""Compile declared VAE decode callables with eager fallback."""
assert self.pipeline is not None
modules = ModuleDiscovery.discover(self.pipeline)
compiled_count = 0
for name, vae in zip(modules.vae_names, modules.vaes, strict=True):
decode = getattr(vae, "decode", None)
if not callable(decode):
logger.debug("Model runner: VAE component %s has no decode callable; skipping compile.", name)
continue
label = f"VAE component {name}.decode"
compiled_decode, activated = compile_callable_with_fallback(
decode,
label=label,
mode="default",
fullgraph=False,
dynamic=None,
)
if not activated:
continue
setattr(vae, "decode", compiled_decode)
compiled_count += 1
if compiled_count:
logger.info(
"Model runner: configured lazy torch.compile for %d VAE decode callable(s); "
"startup warmup will compile the real request path.",
compiled_count,
)
else:
logger.info("Model runner: no compile-compatible VAE decode callable was discovered.")
def load_model( def load_model(
self, self,
memory_pool_context_fn: Callable[[str], AbstractContextManager[Any]] | None = None, memory_pool_context_fn: Callable[[str], AbstractContextManager[Any]] | None = None,
@@ -277,7 +311,8 @@ class DiffusionModelRunner(OmniConnectorModelRunnerMixin):
# Apply torch.compile if not in eager mode # Apply torch.compile if not in eager mode
if not self.od_config.enforce_eager: if not self.od_config.enforce_eager:
if current_omni_platform.supports_torch_inductor(): if current_omni_platform.supports_torch_inductor():
if hasattr(self.pipeline, "setup_compile"): pipeline_owns_compile = hasattr(self.pipeline, "setup_compile")
if pipeline_owns_compile:
try: try:
self.pipeline.setup_compile() self.pipeline.setup_compile()
except Exception as exc: except Exception as exc:
@@ -291,6 +326,9 @@ class DiffusionModelRunner(OmniConnectorModelRunnerMixin):
transformer_attrs = ("transformer", "transformer_2") transformer_attrs = ("transformer", "transformer_2")
for attr_name in transformer_attrs: for attr_name in transformer_attrs:
self._compile_transformer(attr_name) self._compile_transformer(attr_name)
if self.od_config.diffusion_compile_vae and not pipeline_owns_compile:
self._compile_vae_decodes()
else: else:
logger.warning( logger.warning(
"Model runner: Platform %s does not support torch inductor, skipping torch.compile.", "Model runner: Platform %s does not support torch inductor, skipping torch.compile.",
+1
View File
@@ -485,6 +485,7 @@ class OrchestratorArgs:
diffusion_attention_config: str | None = None diffusion_attention_config: str | None = None
diffusion_compile_granularity: str | None = None diffusion_compile_granularity: str | None = None
diffusion_compile_dynamic: bool | None = None diffusion_compile_dynamic: bool | None = None
diffusion_compile_vae: bool | None = None
cache_backend: str = "none" cache_backend: str = "none"
cache_config: str | None = None cache_config: str | None = None
enable_cache_dit_summary: bool = False enable_cache_dit_summary: bool = False
+3
View File
@@ -1043,6 +1043,9 @@ class AsyncOmniEngine:
"diffusion_compile_dynamic": ( "diffusion_compile_dynamic": (
True if kwargs.get("diffusion_compile_dynamic") is None else kwargs["diffusion_compile_dynamic"] True if kwargs.get("diffusion_compile_dynamic") is None else kwargs["diffusion_compile_dynamic"]
), ),
"diffusion_compile_vae": (
True if kwargs.get("diffusion_compile_vae") is None else kwargs["diffusion_compile_vae"]
),
"fa_deterministic": bool(kwargs.get("fa_deterministic", False)), "fa_deterministic": bool(kwargs.get("fa_deterministic", False)),
"boundary_ratio": kwargs.get("boundary_ratio", None), "boundary_ratio": kwargs.get("boundary_ratio", None),
"flow_shift": kwargs.get("flow_shift", None), "flow_shift": kwargs.get("flow_shift", None),
+9
View File
@@ -446,6 +446,15 @@ class OmniServeCommand(CLISubcommand):
"Disable for fixed-shape workloads with --no-diffusion-compile-dynamic." "Disable for fixed-shape workloads with --no-diffusion-compile-dynamic."
), ),
) )
omni_config_group.add_argument(
"--diffusion-compile-vae",
action=argparse.BooleanOptionalAction,
default=None,
help=(
"Compile discovered VAE decode callables when diffusion torch.compile is enabled. "
"Use --no-diffusion-compile-vae to keep VAE decode eager."
),
)
omni_config_group.add_argument( omni_config_group.add_argument(
"--fa-deterministic", "--fa-deterministic",
dest="fa_deterministic", dest="fa_deterministic",