[Perf][Diffusion] MiniMax-H3: opt into packed varlen attention on NPU to eliminate quadratic mask materialization (#5891)

Signed-off-by: brandneway <gyuan4892@gmail.com>

tests/diffusion/attention/test_flash_attn.py

@@ -9,6 +9,10 @@ This script tests two main scenarios:
 2. Case 2: Comparing FlashAttention and SDPA backends for batch_size=2 with padding
 """
 
+import sys
+from types import SimpleNamespace
+from unittest.mock import Mock
+
 import pytest
 import torch
 
@@ -371,6 +375,264 @@ def test_packed_varlen_metadata_must_be_complete(monkeypatch):
         impl.forward_cuda(query, query, query, metadata)
 
 
+# ---------------------------------------------------------------------------
+# NPU mask-free packed attention (PR #5891).
+#
+# These tests cover the backend-side NPU branches that ``forward_fa_npu`` adds:
+#   - boundary resolution in ``_resolve_packed_seq_npu`` (pure Python, no device
+#     sync, no MindIE-SD import);
+#   - env dispatch in ``forward_fa_npu`` (which op a packed forward takes);
+#   - the laser input pre-scaling math in ``_forward_prefix_kv_slice_npu``.
+#
+# They run on CPU without a real NPU by injecting a fake ``mindiesd`` module
+# into ``sys.modules`` (same pattern as ``test_paged_attention`` /
+# ``benchmarks/conftest``). No production code is exercised for real kernels.
+
+
+def _npu_impl(causal: bool = False) -> FlashAttentionImpl:
+    return FlashAttentionImpl(num_heads=2, head_size=4, softmax_scale=0.5, causal=causal)
+
+
+def _fake_mindiesd(monkeypatch, *, attention_forward=None, attention_forward_varlen=None):
+    """Install a stub ``mindiesd`` so local ``from mindiesd import ...`` works.
+
+    Restored automatically by monkeypatch. Defaults are no-op callables so the
+    top-level import in ``forward_fa_npu`` succeeds even when the test mocks the
+    instance methods that would otherwise call them.
+    """
+    fake = SimpleNamespace(
+        attention_forward=attention_forward or Mock(return_value=torch.zeros(1)),
+        attention_forward_varlen=attention_forward_varlen or Mock(return_value=torch.zeros(1)),
+    )
+    monkeypatch.setitem(sys.modules, "mindiesd", fake)
+    return fake
+
+
+# --- Test group A: boundary resolution (_resolve_packed_seq_npu) -------------
+
+
+def test_resolve_packed_seq_accepts_real_plus_pad_contract():
+    impl = _npu_impl()
+    q = torch.randn(1, 8, 2, 4)
+    cu = torch.tensor([0, 5, 8], dtype=torch.int32)
+    extra = {"cu_seqlens_q": cu, "cu_seqlens_k": cu, "max_seqlen_q": 5, "max_seqlen_k": 5}
+
+    assert impl._resolve_packed_seq_npu(q, q, extra) == ([5, 8], [5, 8])
+
+
+def test_resolve_packed_seq_single_document_no_padding():
+    impl = _npu_impl()
+    q = torch.randn(1, 8, 2, 4)
+    cu = torch.tensor([0, 8], dtype=torch.int32)
+    extra = {"cu_seqlens_q": cu, "cu_seqlens_k": cu, "max_seqlen_q": 8, "max_seqlen_k": 8}
+
+    assert impl._resolve_packed_seq_npu(q, q, extra) == ([8], [8])
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        "causal",
+        "missing_key",
+        "batch_ne_1",
+        "cu_too_long",
+        "cu_qk_mismatch",
+        "max_seqlen_not_int",
+        "used_gt_total",
+        "used_zero",
+        "pad_longer_than_real",
+    ],
+)
+def test_resolve_packed_seq_rejects_malformed_metadata(case):
+    """Any metadata outside the exact [real, pad] contract returns None so the
+    caller falls back to the masked path."""
+    cu3 = torch.tensor([0, 5, 8], dtype=torch.int32)
+    cu2 = torch.tensor([0, 8], dtype=torch.int32)
+    cu4 = torch.tensor([0, 3, 5, 8], dtype=torch.int32)
+    cu_short = torch.tensor([0, 3, 8], dtype=torch.int32)
+    base = {"cu_seqlens_q": cu3, "cu_seqlens_k": cu3, "max_seqlen_q": 5, "max_seqlen_k": 5}
+    q = torch.randn(1, 8, 2, 4)
+
+    impl = _npu_impl(causal=(case == "causal"))
+    if case == "missing_key":
+        extra = {"cu_seqlens_q": cu3, "max_seqlen_q": 5}  # 3 of 4 keys absent
+    elif case == "batch_ne_1":
+        q = torch.randn(2, 8, 2, 4)
+        extra = base
+    elif case == "cu_too_long":
+        extra = {**base, "cu_seqlens_q": cu4, "cu_seqlens_k": cu4}
+    elif case == "cu_qk_mismatch":
+        extra = {**base, "cu_seqlens_q": cu3, "cu_seqlens_k": cu2}
+    elif case == "max_seqlen_not_int":
+        extra = {**base, "max_seqlen_q": 5.0}
+    elif case == "used_gt_total":
+        extra = {**base, "max_seqlen_q": 9}  # 9 > total length 8
+    elif case == "used_zero":
+        extra = {**base, "max_seqlen_q": 0, "max_seqlen_k": 0}
+    elif case == "pad_longer_than_real":
+        # real=3, pad=5: the real document must be the longer one.
+        extra = {
+            "cu_seqlens_q": cu_short,
+            "cu_seqlens_k": cu_short,
+            "max_seqlen_q": 3,
+            "max_seqlen_k": 3,
+        }
+    else:
+        extra = base
+
+    assert impl._resolve_packed_seq_npu(q, q, extra) is None
+
+
+# --- Test group B: env dispatch in forward_fa_npu (current behavior) --------
+
+
+def _npu_impl_with_mocked_paths() -> tuple[FlashAttentionImpl, torch.Tensor]:
+    """FlashAttentionImpl whose two packed paths are Mocks returning a sentinel.
+
+    Lets a test assert which branch ``forward_fa_npu`` dispatched to without
+    running any real kernel.
+    """
+    impl = _npu_impl()
+    sentinel = torch.tensor([1.0])
+    impl._forward_varlen_packed_npu = Mock(return_value=sentinel)
+    impl._forward_prefix_kv_slice_npu = Mock(return_value=sentinel)
+    return impl, sentinel
+
+
+def test_npu_env_unset_uses_varlen(monkeypatch):
+    monkeypatch.delenv("MINDIE_SD_FA_TYPE", raising=False)
+    _fake_mindiesd(monkeypatch)
+    impl, _ = _npu_impl_with_mocked_paths()
+    metadata = AttentionMetadata(extra={"npu_attn_varlen": True})
+
+    impl.forward_fa_npu(torch.randn(1, 8, 2, 4), *[torch.randn(1, 8, 2, 4)] * 2, metadata)
+
+    impl._forward_varlen_packed_npu.assert_called_once()
+    impl._forward_prefix_kv_slice_npu.assert_not_called()
+
+
+def test_npu_env_laser_uses_slice(monkeypatch):
+    monkeypatch.setenv("MINDIE_SD_FA_TYPE", "ascend_laser_attention")
+    _fake_mindiesd(monkeypatch)
+    impl, _ = _npu_impl_with_mocked_paths()
+    metadata = AttentionMetadata(extra={"npu_attn_varlen": True})
+
+    impl.forward_fa_npu(torch.randn(1, 8, 2, 4), *[torch.randn(1, 8, 2, 4)] * 2, metadata)
+
+    impl._forward_prefix_kv_slice_npu.assert_called_once()
+    impl._forward_varlen_packed_npu.assert_not_called()
+
+
+@pytest.mark.parametrize("fa_type", ["prompt_flash_attn", "fused_attn_score", "ascend_laser_attenton", "garbage"])
+def test_npu_env_non_laser_value_falls_back_to_varlen(monkeypatch, fa_type):
+    """Locks in the CURRENT dispatch behavior (PR #5891).
+
+    Any value other than ``ascend_laser_attention`` — including the valid
+    ``prompt_flash_attn``/``fused_attn_score`` and typos — is silently routed to
+    the TND varlen op, which cannot honor ``MINDIE_SD_FA_TYPE``. This is review
+    comment #1 on the PR and is intentionally captured here as a regression
+    net: update this test when the dispatch starts honoring/rejecting these.
+    """
+    monkeypatch.setenv("MINDIE_SD_FA_TYPE", fa_type)
+    _fake_mindiesd(monkeypatch)
+    impl, _ = _npu_impl_with_mocked_paths()
+    metadata = AttentionMetadata(extra={"npu_attn_varlen": True})
+
+    impl.forward_fa_npu(torch.randn(1, 8, 2, 4), *[torch.randn(1, 8, 2, 4)] * 2, metadata)
+
+    impl._forward_varlen_packed_npu.assert_called_once()
+    impl._forward_prefix_kv_slice_npu.assert_not_called()
+
+
+def test_npu_varlen_opt_in_unset_takes_mask_path(monkeypatch):
+    """Models that do not set ``npu_attn_varlen`` (e.g. Wan/Cosmos) keep using
+    the existing masked attention_forward path; the new branches are skipped."""
+    fake_forward = Mock(return_value=torch.zeros(1, 8, 2, 4))
+    _fake_mindiesd(monkeypatch, attention_forward=fake_forward)
+    impl, _ = _npu_impl_with_mocked_paths()
+    metadata = AttentionMetadata(
+        attn_mask=torch.tensor([[True, True, True, True, True, False, False, False]]),
+        extra={},  # no npu_attn_varlen opt-in
+    )
+    q = torch.randn(1, 8, 2, 4)
+
+    out = impl.forward_fa_npu(q, q, q, metadata)
+
+    impl._forward_varlen_packed_npu.assert_not_called()
+    impl._forward_prefix_kv_slice_npu.assert_not_called()
+    fake_forward.assert_called_once()
+    assert out is fake_forward.return_value
+
+
+# --- Test group C: laser input pre-scaling in _forward_prefix_kv_slice_npu --
+
+
+def test_prefix_kv_slice_applies_laser_input_scaling(monkeypatch):
+    monkeypatch.setenv("MINDIE_SD_FA_TYPE", "ascend_laser_attention")
+    captured: dict = {}
+
+    def fake_attention_forward(query, key, value, **kwargs):
+        captured["query"] = query
+        captured["key"] = key
+        captured["value"] = value
+        captured.update(kwargs)
+        return torch.ones_like(query)  # deterministic stand-in kernel output
+
+    _fake_mindiesd(monkeypatch, attention_forward=fake_attention_forward)
+    impl = _npu_impl()  # softmax_scale == 0.5
+    q = torch.randn(1, 8, 2, 4)
+    cu = torch.tensor([0, 5, 8], dtype=torch.int32)
+    extra = {
+        "cu_seqlens_q": cu,
+        "cu_seqlens_k": cu,
+        "max_seqlen_q": 5,
+        "max_seqlen_k": 5,
+        "laser_input_scale": 256.0,
+    }
+
+    out = impl._forward_prefix_kv_slice_npu(q, q, q, extra)
+
+    # K/V sliced to the valid prefix (used_k == 5); q keeps full length.
+    assert captured["key"].shape == (1, 5, 2, 4)
+    assert captured["value"].shape == (1, 5, 2, 4)
+    assert captured["query"].shape == (1, 8, 2, 4)
+    assert captured["attn_mask"] is None
+    # Pre-divide q/k/v by the factor and compensate the kernel scale by its
+    # square (exact for power-of-two); output scaled back by the factor.
+    torch.testing.assert_close(captured["query"], q / 256.0)
+    torch.testing.assert_close(captured["key"], q[:, :5] / 256.0)
+    assert captured["scale"] == pytest.approx(0.5 * 256.0**2)
+    torch.testing.assert_close(out, torch.ones(1, 8, 2, 4) * 256.0)
+
+
+def test_prefix_kv_slice_no_scaling_without_factor(monkeypatch):
+    monkeypatch.setenv("MINDIE_SD_FA_TYPE", "ascend_laser_attention")
+    captured: dict = {}
+
+    def fake_attention_forward(query, key, value, **kwargs):
+        captured["query"] = query
+        captured.update(kwargs)
+        return torch.ones_like(query)
+
+    _fake_mindiesd(monkeypatch, attention_forward=fake_attention_forward)
+    impl = _npu_impl()
+    q = torch.randn(1, 8, 2, 4)
+    cu = torch.tensor([0, 5, 8], dtype=torch.int32)
+    extra = {
+        "cu_seqlens_q": cu,
+        "cu_seqlens_k": cu,
+        "max_seqlen_q": 5,
+        "max_seqlen_k": 5,
+        # no laser_input_scale → no pre-scaling
+    }
+
+    out = impl._forward_prefix_kv_slice_npu(q, q, q, extra)
+
+    torch.testing.assert_close(captured["query"], q)
+    assert captured["scale"] == pytest.approx(0.5)
+    torch.testing.assert_close(out, torch.ones(1, 8, 2, 4))
+
+
 if __name__ == "__main__":
     print("Running FlashAttention Padding Tests...")
     print("=" * 60)

tests/diffusion/models/minimax_h3/test_minimax_h3_contract.py

@@ -492,6 +492,10 @@ def test_cudnn_packed_attention_uses_python_length_without_padding_mask():
     class FakeBackend:
         supports_prefix_kv_slicing = True
 
+        @classmethod
+        def supports_packed_mask_free(cls) -> bool:
+            return False
+
     class FakeAttention(torch.nn.Module):
         attn_backend = FakeBackend
 
@@ -526,6 +530,48 @@ def test_cudnn_packed_attention_uses_python_length_without_padding_mask():
     assert attention.attention.metadata.extra["valid_kv_length"] == 5
 
 
+def test_packed_attention_skips_mask_for_packed_mask_free_backend():
+    from vllm_omni.diffusion.models.minimax_h3.minimax_h3_transformer import (
+        MiniMaxH3Attention,
+    )
+
+    class FakeBackend:
+        supports_prefix_kv_slicing = False
+
+        @classmethod
+        def supports_packed_mask_free(cls) -> bool:
+            return True
+
+    class FakeAttention(torch.nn.Module):
+        attn_backend = FakeBackend
+
+        def __init__(self):
+            super().__init__()
+            self.metadata = None
+
+        def forward(self, query, key, value, metadata):
+            self.metadata = metadata
+            return query
+
+    attention = object.__new__(MiniMaxH3Attention)
+    torch.nn.Module.__init__(attention)
+    attention.attention = FakeAttention()
+    q = torch.randn(8, 2, 4)
+
+    attention._run_packed_attention(
+        q,
+        q,
+        q,
+        cu_seqlens=torch.tensor([0, 5, 8], dtype=torch.int32),
+        max_seqlen=5,
+        packed_total=8,
+    )
+
+    assert attention.attention.metadata.attn_mask is None
+    assert attention.attention.metadata.extra["valid_kv_length"] == 5
+    assert attention.attention.metadata.extra["npu_attn_varlen"] is True
+
+
 def test_packed_attention_keeps_padding_mask_for_other_backends():
     from vllm_omni.diffusion.models.minimax_h3.minimax_h3_transformer import (
         MiniMaxH3Attention,
@@ -534,6 +580,10 @@ def test_packed_attention_keeps_padding_mask_for_other_backends():
     class FakeBackend:
         supports_prefix_kv_slicing = False
 
+        @classmethod
+        def supports_packed_mask_free(cls) -> bool:
+            return False
+
     class FakeAttention(torch.nn.Module):
         attn_backend = FakeBackend