vllm.v1.attention.backends.mla.flashmla_sparse ¶

class FlashMLASparseImpl(SparseMLACommonImpl[FlashMLASparseMetadata]):
    can_return_lse_for_decode: bool = True
    supports_dcp: bool = True

    @staticmethod
    def _compute_fp8_decode_padded_heads(num_heads: int) -> int:
        # FP8 decode kernel only supports h_q = 64 or 128
        # Compute padded head count for decode
        return 64 if num_heads <= 64 else 128

    def __init__(
        self,
        num_heads: int,
        head_size: int,
        scale: float,
        num_kv_heads: int,
        alibi_slopes: list[float] | None,
        sliding_window: int | None,
        kv_cache_dtype: str,
        logits_soft_cap: float | None,
        attn_type: str,
        kv_sharing_target_layer_name: str | None,
        # MLA Specific Arguments
        topk_indices_buffer: torch.Tensor | None = None,
        indexer: "Indexer | None" = None,
        **mla_args,
    ) -> None:
        super().__init__(
            num_heads,
            head_size,
            scale,
            num_kv_heads,
            alibi_slopes,
            sliding_window,
            kv_cache_dtype,
            logits_soft_cap,
            attn_type,
            kv_sharing_target_layer_name,
            indexer=indexer,
            topk_indices_buffer=topk_indices_buffer,
            **mla_args,
        )
        self.softmax_scale = scale
        # Prefill BF16 kernel requires 64 on Hopper, 128 on Blackwell
        self.prefill_padding = (
            128 if current_platform.is_device_capability_family(100) else 64
        )
        self.fp8_decode_padded_heads = self._compute_fp8_decode_padded_heads(num_heads)

        vllm_config = get_current_vllm_config()
        max_tokens = vllm_config.scheduler_config.max_num_batched_tokens
        q_concat_heads = num_heads
        if not is_quantized_kv_cache(kv_cache_dtype):
            q_concat_heads = (
                (num_heads + self.prefill_padding - 1)
                // self.prefill_padding
                * self.prefill_padding
            )
        q_concat_shape = (max_tokens, q_concat_heads, head_size)
        if is_quantized_kv_cache(kv_cache_dtype):
            assert kv_cache_dtype in QUANTIZED_DS_MLA_CACHE_FORMATS, (
                "FlashMLA Sparse Attention backend only supports the "
                f"{sorted(QUANTIZED_DS_MLA_CACHE_FORMATS)} quantized kv-cache "
                f"dtypes, got {kv_cache_dtype}"
            )

        if self.need_to_return_lse_for_decode and not is_quantized_kv_cache(
            kv_cache_dtype
        ):
            raise NotImplementedError(
                "DCP for FlashMLA sparse requires an fp8_ds_mla kv-cache; "
                "the bf16 sparse path is not supported under DCP."
            )

        if kv_cache_dtype in QUANTIZED_DS_MLA_CACHE_FORMATS:
            # Reserve workspace during initialization
            assert vllm_config is not None and vllm_config.model_config is not None
            prefill_workspace_size = get_prefill_workspace_size(
                vllm_config.model_config.max_model_len
            )
            self.prefill_workspace_shape = (prefill_workspace_size, head_size)
            self.q_concat_buffer, self.prefill_bf16_workspace = (
                current_workspace_manager().get_simultaneous(
                    (q_concat_shape, torch.bfloat16),
                    (self.prefill_workspace_shape, torch.bfloat16),
                )
            )
        else:
            (self.q_concat_buffer,) = current_workspace_manager().get_simultaneous(
                (q_concat_shape, torch.bfloat16),
            )

    def _forward_bf16_kv(
        self,
        q: torch.Tensor,
        kv_c_and_k_pe_cache: torch.Tensor,
        topk_indices: torch.Tensor,
        attn_metadata: FlashMLASparseMetadata,
        actual_num_heads: int,
    ) -> tuple[torch.Tensor, torch.Tensor | None]:
        index_group = self.index_group
        if isinstance(index_group, HiSparseMLAIndexGroup):
            cache = index_group.cache(self.index_group_index)
        else:
            cache = None
        block_table = attn_metadata.block_table
        # req_id_per_token covers the whole batch; slice it to the MQA tokens
        # (q may exclude prefill tokens routed to dense MHA).
        req_id_per_token = attn_metadata.req_id_per_token[: topk_indices.shape[0]]
        decode_out: torch.Tensor | None = None
        if cache is not None:
            assert isinstance(index_group, HiSparseMLAIndexGroup)
            num_decode_tokens = attn_metadata.num_decode_tokens
            if num_decode_tokens > 0:
                decode_topk, decode_lengths = (
                    index_group.convert_decode_logical_to_physical_topk(
                        self.index_group_index,
                        topk_indices[:num_decode_tokens],
                        attn_metadata,
                        return_valid_counts=True,
                    )
                )
                decode_out, _ = self._bf16_flash_mla_kernel(
                    q[:num_decode_tokens],
                    index_group.physical_kv_cache(self.index_group_index),
                    decode_topk,
                    decode_lengths,
                    actual_num_heads,
                )
                if num_decode_tokens == q.shape[0]:
                    return decode_out, None
                q = q[num_decode_tokens:]
                topk_indices = topk_indices[num_decode_tokens:]
            kv_c_and_k_pe_cache, block_table, req_id_per_token = (
                index_group.stage_prefill_rows(
                    self.index_group_index, kv_c_and_k_pe_cache, attn_metadata
                )
            )
        # Convert per-request indices to global slots (decode) or workspace offsets.
        kv_rows, block_stride_rows = flat_kv_row_view(
            kv_c_and_k_pe_cache, attn_metadata.block_size
        )
        decode_only = (
            attn_metadata.num_decode_tokens
            == attn_metadata.num_actual_tokens
            == topk_indices.shape[0]
        )
        uses_host_cache = isinstance(index_group, HiSparseMLAIndexGroup)
        if not uses_host_cache and decode_only:
            topk_indices, topk_length = self._convert_logical_to_physical_topk(
                topk_indices,
                attn_metadata,
                block_stride_rows=block_stride_rows,
                return_valid_counts=True,
            )
        else:
            topk_indices, topk_length = triton_convert_req_index_to_global_index(
                req_id_per_token,
                block_table,
                topk_indices,
                BLOCK_SIZE=attn_metadata.block_size,
                BLOCK_STRIDE_ROWS=block_stride_rows,
                NUM_TOPK_TOKENS=topk_indices.shape[1],
                return_valid_counts=True,
            )

        attn_out, lse = self._bf16_flash_mla_kernel(
            q,
            kv_rows,
            topk_indices,
            topk_length,
            actual_num_heads,
        )
        if decode_out is None:
            return attn_out, lse
        return torch.cat([decode_out, attn_out], dim=0), None

    def _forward_fp8_kv_separate_prefill_decode(
        self,
        q: torch.Tensor,
        kv_c_and_k_pe_cache: torch.Tensor,
        topk_indices: torch.Tensor,
        attn_metadata: FlashMLASparseMetadata,
    ) -> torch.Tensor:
        fp8_metadata = attn_metadata.fp8_extra_metadata
        assert isinstance(fp8_metadata, FlashMLASparseMetadata.FP8SeparatePrefillDecode)
        num_decodes = fp8_metadata.num_decodes
        num_mqa_tokens = q.shape[0]
        num_decode_tokens = fp8_metadata.num_decode_tokens
        num_prefill_tokens = num_mqa_tokens - num_decode_tokens
        assert num_prefill_tokens in (0, fp8_metadata.num_prefill_tokens), (
            "FP8 sparse MLA expects either the decode subset or the full batch"
        )

        decode_topk: torch.Tensor | None = None
        index_group = self.index_group
        uses_host_cache = isinstance(index_group, HiSparseMLAIndexGroup)
        if uses_host_cache and num_decode_tokens > 0:
            decode_topk = topk_indices[:num_decode_tokens]

        prefill_ready = None
        if num_prefill_tokens > 0 and uses_host_cache:
            assert fp8_metadata.prefill is not None
            first_chunk = fp8_metadata.prefill.chunks[0]
            assert isinstance(index_group, HiSparseMLAIndexGroup)
            prefill_ready = index_group.gather_fp8_prefill(
                self.index_group_index,
                kv_c_and_k_pe_cache,
                self.prefill_bf16_workspace[: first_chunk.chunk_tot_seqlen],
                first_chunk.block_table,
                first_chunk.workspace_starts,
                len(first_chunk.block_table),
                attn_metadata,
                first_chunk.req_start_idx,
            )

        prefill_request_ids = None
        prefill_workspace_starts = None
        has_prefill_workspace = False
        if num_prefill_tokens > 0:
            assert fp8_metadata.prefill is not None
            prefill_request_ids = fp8_metadata.prefill.request_ids
            prefill_workspace_starts = fp8_metadata.prefill.workspace_starts
            has_prefill_workspace = True

        # Convert per-request indices to global slots (decode) or workspace
        # offsets (prefill).
        # For FP8 cache: prefill uses workspace mapping (upconverted to BF16)
        # For BF16 cache: always use global cache slots (no workspace)
        # prefill_workspace_starts has been adjusted in-place per chunk so
        # prefill indices automatically come out chunk-local
        topk_length = None
        if num_prefill_tokens == 0 and not uses_host_cache:
            topk_indices, topk_length = self._convert_logical_to_physical_topk(
                topk_indices,
                attn_metadata,
                block_stride_rows=None,
                return_valid_counts=True,
            )
        elif num_prefill_tokens > 0:
            topk_indices, topk_length = triton_convert_req_index_to_global_index(
                attn_metadata.req_id_per_token[: topk_indices.shape[0]],
                attn_metadata.block_table,
                topk_indices,
                BLOCK_SIZE=attn_metadata.block_size,
                NUM_TOPK_TOKENS=topk_indices.shape[1],
                HAS_PREFILL_WORKSPACE=has_prefill_workspace,
                prefill_workspace_request_ids=prefill_request_ids,
                prefill_workspace_starts=prefill_workspace_starts,
                return_valid_counts=True,
            )

        fp8_metadata = attn_metadata.fp8_extra_metadata
        assert isinstance(fp8_metadata, FlashMLASparseMetadata.FP8SeparatePrefillDecode)

        def _fp8_decode(
            q: torch.Tensor,
            topk_indices: torch.Tensor,
        ) -> torch.Tensor:
            assert fp8_metadata.decode is not None
            if uses_host_cache:
                return self._host_backed_fp8_decode(
                    q,
                    topk_indices,
                    attn_metadata,
                    fp8_metadata.decode.kernel_metadata,
                    num_decodes,
                    fp8_metadata.decode.decode_query_len,
                )
            # Reshape q: (num_decode_tokens, num_heads, head_dim)
            #         -> (num_decodes, seq_len, num_heads, head_dim)
            q = reshape_query_for_spec_decode(q, num_decodes)
            seq_len = q.shape[1]
            # Reshape topk_indices: (num_decode_tokens, topk)
            #                    -> (num_decodes, seq_len, topk)
            topk_indices = topk_indices.view(num_decodes, seq_len, -1)
            attn_out, _ = self._fp8_flash_mla_kernel(
                q=q,
                kv_c_and_k_pe_cache=kv_c_and_k_pe_cache,
                topk_indices=topk_indices,
                kernel_metadata=fp8_metadata.decode.kernel_metadata,
            )
            # Reshape output: (num_decodes, seq_len, num_heads, head_dim_v)
            #              -> (num_decode_tokens, num_heads, head_dim_v)
            return reshape_attn_output_for_spec_decode(attn_out)

        # Pure decode: direct call without allocation
        if num_decode_tokens > 0 and num_prefill_tokens == 0:
            assert fp8_metadata.decode is not None
            attn_out = _fp8_decode(
                q, decode_topk if decode_topk is not None else topk_indices
            )
        else:
            # Mixed or pure prefill: allocate output tensor
            attn_out = q.new_empty(
                (num_mqa_tokens, self.num_heads, self.kv_lora_rank),
                dtype=q.dtype,
                device=q.device,
            )

            if num_decode_tokens > 0:
                attn_out[:num_decode_tokens] = _fp8_decode(
                    q[:num_decode_tokens],
                    decode_topk
                    if decode_topk is not None
                    else topk_indices[:num_decode_tokens],
                )

            assert fp8_metadata.prefill is not None
            for chunk_index, chunk in enumerate(fp8_metadata.prefill.chunks):
                chunk_workspace = self.prefill_bf16_workspace[: chunk.chunk_tot_seqlen]
                if uses_host_cache and chunk_index > 0:
                    assert isinstance(index_group, HiSparseMLAIndexGroup)
                    prefill_ready = index_group.gather_fp8_prefill(
                        self.index_group_index,
                        kv_c_and_k_pe_cache,
                        chunk_workspace,
                        chunk.block_table,
                        chunk.workspace_starts,
                        len(chunk.block_table),
                        attn_metadata,
                        chunk.req_start_idx,
                    )
                if uses_host_cache:
                    assert prefill_ready is not None
                    current_stream().wait_event(prefill_ready)
                elif self.kv_cache_dtype == "fp8_ds_mla":
                    ops.cp_gather_and_upconvert_fp8_kv_cache(
                        kv_c_and_k_pe_cache,
                        chunk_workspace,
                        chunk.block_table,
                        chunk.workspace_starts,
                        len(chunk.block_table),
                    )
                else:
                    ops.cp_gather_and_upconvert_nvfp4_kv_cache(
                        kv_c_and_k_pe_cache.view(torch.uint8),
                        chunk_workspace,
                        chunk.block_table,
                        chunk.workspace_starts,
                        len(chunk.block_table),
                    )

                chunk_q = q[chunk.tokens_slice]
                chunk_topk_indices_workspace = topk_indices[chunk.tokens_slice]
                assert topk_length is not None
                chunk_topk_length = topk_length[chunk.tokens_slice]

                attn_out[chunk.tokens_slice], _ = self._bf16_flash_mla_kernel(
                    chunk_q,
                    chunk_workspace,
                    chunk_topk_indices_workspace,
                    chunk_topk_length,
                )

        return attn_out

    def _forward_fp8_kv_mixed_batch(
        self,
        q: torch.Tensor,
        kv_c_and_k_pe_cache: torch.Tensor,
        topk_indices: torch.Tensor,
        attn_metadata: FlashMLASparseMetadata,
    ) -> tuple[torch.Tensor, torch.Tensor | None]:
        """Mixed batch FP8 forward path that treats all tokens as one batch.

        This is equivalent to main branch's approach and avoids the BF16
        prefill kernel which has head padding overhead when num_heads is small.
        Used when use_mixed_batch is True.

        The lse is only returned when DCP needs it, otherwise None.
        """
        assert attn_metadata.fp8_extra_metadata is not None
        assert isinstance(
            attn_metadata.fp8_extra_metadata,
            FlashMLASparseMetadata.FP8KernelMetadata,
        )
        fp8_metadata = attn_metadata.fp8_extra_metadata

        block_table = attn_metadata.block_table
        # req_id_per_token covers the whole batch; slice it to the MQA tokens
        # (q may exclude prefill tokens routed to dense MHA).
        req_id_per_token = attn_metadata.req_id_per_token[: topk_indices.shape[0]]
        if self.dcp_world_size > 1:
            # The indexer emits global token ids; keep this rank's shard and
            # convert to local slots. compact_valid_to_front=False keeps the
            # scattered -1s, which the fp8 kernel masks natively and the
            # empty-row neutralization below relies on. req_id is sliced to
            # topk_indices rows (the converter grids from req_id).
            topk_indices = triton_filter_and_convert_dcp_index(
                req_id_per_token,
                block_table,
                topk_indices,
                dcp_size=self.dcp_world_size,
                dcp_rank=self.dcp_rank,
                cp_kv_cache_interleave_size=attn_metadata.cp_kv_cache_interleave_size,
                BLOCK_SIZE=attn_metadata.block_size,
                NUM_TOPK_TOKENS=topk_indices.shape[1],
                compact_valid_to_front=False,
            )
        else:
            # Convert per-request indices to global slots (decode) or workspace
            # offsets (prefill).
            decode_only = (
                attn_metadata.num_decode_tokens
                == attn_metadata.num_actual_tokens
                == topk_indices.shape[0]
            )
            if decode_only:
                topk_indices = self._convert_logical_to_physical_topk(
                    topk_indices,
                    attn_metadata,
                    block_stride_rows=None,
                    return_valid_counts=False,
                )
            else:
                topk_indices = triton_convert_req_index_to_global_index(
                    req_id_per_token,
                    block_table,
                    topk_indices,
                    BLOCK_SIZE=attn_metadata.block_size,
                    NUM_TOPK_TOKENS=topk_indices.shape[1],
                )

        _attn_out, _lse = self._fp8_flash_mla_kernel(
            q=q.unsqueeze(0),  # unsqueeze to add batch_dim: (T, H, D) -> (1, T, H, D)
            kv_c_and_k_pe_cache=kv_c_and_k_pe_cache,
            topk_indices=topk_indices.unsqueeze(0),  # (T, topk) -> (1, T, topk)
            kernel_metadata=fp8_metadata,
        )
        # Output is (1, T, H, D_v), squeeze back to (T, H, D_v)
        out = _attn_out.squeeze(0)

        if not self.need_to_return_lse_for_decode:
            return out, None

        # Kernel LSE is (1, H, T); the DCP merge consumes (T, H).
        lse = _lse.squeeze(0).transpose(0, 1)
        # Rows where this rank owns none of the selected tokens (all indices
        # -1) have undefined out/lse; (0, -inf) is the identity element of the
        # cross-rank LSE merge, so it drops this rank from those rows.
        empty_rows = (topk_indices == -1).all(dim=-1)
        out.masked_fill_(empty_rows.view(-1, 1, 1), 0.0)
        lse.masked_fill_(empty_rows.view(-1, 1), float("-inf"))
        # The head-padding slice above can leave `out` non-contiguous, and the
        # merge feeds it to reduce_scatter.
        return out.contiguous(), lse

    def _fp8_flash_mla_kernel(
        self,
        q: torch.Tensor,
        kv_c_and_k_pe_cache: torch.Tensor,
        topk_indices: torch.Tensor,
        kernel_metadata: FlashMLASparseMetadata.FP8KernelMetadata,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        # q shape: (batch, seq_len, num_heads, head_dim)
        actual_num_heads = q.size(2)
        padded_num_heads = self.fp8_decode_padded_heads

        # Pad query if needed (kernel only supports h_q = 64 or 128)
        if actual_num_heads < padded_num_heads:
            logger.warning_once(
                f"Padding num_heads from {actual_num_heads} to "
                f"{padded_num_heads} for FP8 sparse decode kernel"
            )
            q_padded = q.new_zeros((q.size(0), q.size(1), padded_num_heads, q.size(3)))
            q_padded[:, :, :actual_num_heads, :] = q
            q = q_padded

        out, lse = flash_mla_with_kvcache(
            q=q,
            k_cache=kv_c_and_k_pe_cache.view(torch.uint8).unsqueeze(-2),
            block_table=kernel_metadata.dummy_block_table,
            head_dim_v=512,
            cache_seqlens=kernel_metadata.cache_lens,
            tile_scheduler_metadata=kernel_metadata.scheduler_metadata,
            is_fp8_kvcache=True,
            indices=topk_indices,
            softmax_scale=self.softmax_scale,
        )

        # Slice output and lse back to actual head count if we padded
        if actual_num_heads < padded_num_heads:
            out = out[:, :, :actual_num_heads, :]
            lse = lse[:, :actual_num_heads, :]

        return out, lse

    def _host_backed_fp8_decode(
        self,
        q: torch.Tensor,
        topk_indices: torch.Tensor,
        attn_metadata: FlashMLASparseMetadata,
        kernel_metadata: FlashMLASparseMetadata.FP8KernelMetadata,
        num_decodes: int,
        decode_query_len: int,
    ) -> torch.Tensor:
        assert isinstance(self.index_group, HiSparseMLAIndexGroup)
        physical_topk = self.index_group.convert_decode_logical_to_physical_topk(
            self.index_group_index,
            topk_indices,
            attn_metadata,
            return_valid_counts=False,
            num_decodes=num_decodes,
            decode_query_len=decode_query_len,
        )
        assert isinstance(physical_topk, torch.Tensor)
        q = reshape_query_for_spec_decode(q, num_decodes)
        physical_topk = physical_topk.view(num_decodes, q.shape[1], -1)
        output, _ = self._fp8_flash_mla_kernel(
            q=q,
            kv_c_and_k_pe_cache=self.index_group.physical_kv_cache(
                self.index_group_index
            ),
            topk_indices=physical_topk,
            kernel_metadata=kernel_metadata,
        )
        return reshape_attn_output_for_spec_decode(output)

    def _bf16_flash_mla_kernel(
        self,
        q: torch.Tensor,
        kv_c_and_k_pe_cache: torch.Tensor,
        topk_indices: torch.Tensor,
        topk_length: torch.Tensor | None = None,
        actual_num_heads: int | None = None,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        num_tokens = q.shape[0]
        kv_c_and_k_pe_cache = kv_c_and_k_pe_cache.view(
            -1, 1, kv_c_and_k_pe_cache.shape[-1]
        )

        # NOTE(Chen): kernel requires num_local_head to be a multiple of
        # 64 on hopper and 128 on blackwell. Pad from q's head count, not
        # self.num_heads: under DCP the heads are all-gathered before this.
        if actual_num_heads is None:
            actual_num_heads = q.shape[1]
        padded_num_heads = (
            (actual_num_heads + self.prefill_padding - 1)
            // self.prefill_padding
            * self.prefill_padding
        )
        if q.shape[1] < padded_num_heads:
            logger.warning_once(
                f"Padding num_heads from {actual_num_heads} to "
                f"{padded_num_heads} for BF16 sparse prefill kernel"
            )
            q_padded = q.new_empty((q.shape[0], padded_num_heads, q.shape[2]))
            q_padded[:, :actual_num_heads, :] = q
            q = q_padded

        topk_indices = topk_indices.view(num_tokens, 1, -1)
        output, _, lse = flash_mla_sparse_fwd(
            q,
            kv_c_and_k_pe_cache,
            topk_indices,
            self.softmax_scale,
            topk_length=topk_length,
        )

        output = output[:, :actual_num_heads, :]
        lse = lse[:, :actual_num_heads]
        return output, lse

    def forward_mqa(
        self,
        q: torch.Tensor | tuple[torch.Tensor, torch.Tensor],
        kv_c_and_k_pe_cache: torch.Tensor,
        attn_metadata: FlashMLASparseMetadata,
        layer: AttentionLayer,
    ) -> tuple[torch.Tensor, torch.Tensor | None]:
        # NOTE(lucas): for the sparse FlashMLA kernels the kernels want to use
        # MQA 576/512 approach for both prefill and decode

        # Concatenate q if it's a tuple (ql_nope, q_pe)
        actual_num_heads = self.num_heads
        if isinstance(q, tuple):
            ql_nope, q_pe = q
            q = self.q_concat_buffer[: ql_nope.shape[0]]
            ops.concat_mla_q(ql_nope, q_pe, q)
        else:
            actual_num_heads = q.shape[1]

        num_actual_toks = q.shape[0]

        # Get topk indices
        assert self.topk_indices_buffer is not None
        topk_indices = self.topk_indices_buffer[:num_actual_toks]

        use_fp8_cache = self.kv_cache_dtype in QUANTIZED_DS_MLA_CACHE_FORMATS

        lse: torch.Tensor | None = None

        if not use_fp8_cache:
            attn_out, bf16_lse = self._forward_bf16_kv(
                q,
                kv_c_and_k_pe_cache,
                topk_indices,
                attn_metadata,
                actual_num_heads,
            )
            if self.need_to_return_lse_for_decode:
                lse = bf16_lse
        elif attn_metadata.fp8_use_mixed_batch:
            attn_out, lse = self._forward_fp8_kv_mixed_batch(
                q, kv_c_and_k_pe_cache, topk_indices, attn_metadata
            )
        else:
            attn_out = self._forward_fp8_kv_separate_prefill_decode(
                q, kv_c_and_k_pe_cache, topk_indices, attn_metadata
            )

        return attn_out, lse

@dataclass
class FlashMLASparseMetadata(SparseMLACommonMetadata):
    @dataclass
    class FP8KernelMetadata:
        scheduler_metadata: FlashMLASchedMeta
        dummy_block_table: torch.Tensor
        cache_lens: torch.Tensor

    @dataclass
    class FP8SeparatePrefillDecode:
        @dataclass
        class Decode:
            seq_lens: torch.Tensor
            kernel_metadata: "FlashMLASparseMetadata.FP8KernelMetadata"
            decode_query_len: int  # needed for reshape in spec decode

        @dataclass
        class Prefill:
            # Request ID for each token: -1 for decode tokens, request index
            # (0, 1, 2, ...) for prefill tokens.
            # Shape: [num_actual_tokens]
            request_ids: torch.Tensor

            # Workspace start offsets for all prefill requests
            # Shape: [num_prefill_reqs], adjusted in-place per chunk to be
            # 0-indexed within each chunk. Used to map prefill tokens to workspace
            # offsets in convert_logical_index_to_physical_index
            workspace_starts: torch.Tensor

            @dataclass
            class Chunk:
                """Metadata for a chunk of prefill requests.

                Prefill requests may be chunked to fit within the fixed workspace size.
                """

                tokens_slice: slice
                block_table: torch.Tensor
                req_start_idx: int
                workspace_starts: torch.Tensor
                chunk_tot_seqlen: int
                seq_lens: torch.Tensor | None = None

            chunks: list[Chunk]

        num_prefills: int = 0
        num_decodes: int = 0
        num_prefill_tokens: int = 0
        num_decode_tokens: int = 0

        decode: Decode | None = None
        prefill: Prefill | None = None

    fp8_extra_metadata: FP8SeparatePrefillDecode | FP8KernelMetadata | None = None
    fp8_use_mixed_batch: bool = False

class FlashMLASparseMetadataBuilder(
    SparseMLACommonMetadataBuilder[FlashMLASparseMetadata]
):
    _cudagraph_support: ClassVar[AttentionCGSupport] = AttentionCGSupport.UNIFORM_BATCH
    require_uniform_decodes: ClassVar[bool] = True
    hisparse_supports_multi_token_decode: ClassVar[bool] = True
    metadata_cls = FlashMLASparseMetadata

    def __init__(
        self,
        kv_cache_spec: AttentionSpec,
        layer_names: list[str],
        vllm_config: VllmConfig,
        device: torch.device,
    ) -> None:
        super().__init__(kv_cache_spec, layer_names, vllm_config, device)
        cache_config = vllm_config.cache_config
        parallel_config = vllm_config.parallel_config

        num_q_heads = self.model_config.get_num_attention_heads(parallel_config)
        if current_platform.is_device_capability_family(100):
            threshold = {8: 128, 16: 128, 32: 128, 64: 256, 128: 1024}.get(
                num_q_heads, 1024
            )
        else:
            threshold = {16: 128, 32: 128, 64: 256, 128: 256}.get(num_q_heads, 256)
        self.use_hisparse = vllm_config.attention_config.hisparse_config is not None
        if self.use_hisparse:
            threshold = 1
        # Varlen decodes are safe under DCP: causality comes from the
        # indexer's top-k indices, not from the kernel metadata.
        self._init_reorder_batch_threshold(
            threshold,
            supports_spec_as_decode=True,
            supports_dcp_with_varlen=(parallel_config.cp_kv_cache_interleave_size == 1),
        )

        sm_count = num_compute_units(device.index)

        self.num_heads = self.model_config.get_num_attention_heads(parallel_config)
        # FP8 decode kernel only supports h_q = 64 or 128, so we need to pad
        self.fp8_decode_padded_heads = (
            FlashMLASparseImpl._compute_fp8_decode_padded_heads(self.num_heads)
        )

        self.use_fp8_kv_cache = (
            cache_config.cache_dtype in QUANTIZED_DS_MLA_CACHE_FORMATS
        )
        max_num_seqs = vllm_config.scheduler_config.max_num_seqs
        # Shape: [max_num_seqs], all elements = topk_tokens (constant for full-CG)
        self.topk_tokens_tensor = torch.full(
            (max_num_seqs,), self.topk_tokens, device=device, dtype=torch.int32
        )
        # Shape: [max_num_seqs], all elements = max_model_len
        self.max_model_len_tensor = torch.full(
            (max_num_seqs,),
            self.model_config.max_model_len,
            device=device,
            dtype=torch.int32,
        )
        # this is ignored by `flash_mla_with_kvcache` if indices not None
        self.dummy_block_table = torch.empty(
            (max_num_seqs, 1), dtype=torch.int32, device=self.device
        )

        # Equation taken from FlashMLA/csrc/api/sparse_decode.h
        # For sparse FP8 decode, the formula depends on architecture:
        # - SM90 (Hopper): num_sm_parts = num_sms / s_q / (h_q/64)
        # - SM100 (Blackwell head64/head64x2): num_sm_parts = num_sms / s_q
        # - SM100 (Blackwell head128): num_sm_parts = num_sms / s_q / 2
        # For max buffer size, use s_q = 1 (the case that produces largest output)
        # Use padded head count since that's what will be passed to the kernel
        h_q = self.fp8_decode_padded_heads
        if current_platform.is_device_capability_family(100):
            # SM100 head64 or head64x2 uses full SM count
            max_num_sm_parts = sm_count
        else:
            # SM90 uses h_q/64 divisor
            max_num_sm_parts = sm_count // max(1, h_q // 64)
        self.tile_scheduler_metadata_buffer = torch.empty(
            # TileSchedulerMetaDataSize = 8
            # see: FlashMLA/csrc/params.h
            (max_num_sm_parts, 8),
            dtype=torch.int32,
            device=device,
        )
        # Sized for per-request batching (num_decodes + 1)
        self.num_splits_buffer = torch.empty(
            (max_num_seqs + 1,),
            dtype=torch.int32,
            device=device,
        )

        self.fp8_use_mixed_batch = (
            self.num_heads < MIN_HEADS_FOR_BF16_PREFILL and not self.use_hisparse
        )

        if parallel_config.decode_context_parallel_size > 1:
            if parallel_config.dcp_comm_backend != "ag_rs":
                raise NotImplementedError(
                    "DCP for FlashMLA sparse is only validated with the "
                    "default 'ag_rs' DCP comm backend; got "
                    f"'{parallel_config.dcp_comm_backend}'"
                )
            if not self.fp8_use_mixed_batch:
                raise NotImplementedError(
                    "DCP for FlashMLA sparse is only supported on the "
                    "mixed-batch fp8 path (num_heads < "
                    f"{MIN_HEADS_FOR_BF16_PREFILL}); the separate "
                    "prefill/decode path returns the LSE for decode tokens "
                    "only, while the DCP merge needs it for every token"
                )
            # Head padding (and the tile-scheduler metadata sized from it) is
            # computed from the local head count, but the kernel runs on the
            # DCP-gathered heads.
            gathered_num_heads = (
                self.num_heads * parallel_config.decode_context_parallel_size
            )
            gathered_padded_heads = FlashMLASparseImpl._compute_fp8_decode_padded_heads(
                gathered_num_heads
            )
            if self.fp8_decode_padded_heads != gathered_padded_heads:
                raise NotImplementedError(
                    "DCP for FlashMLA sparse requires the local and "
                    "DCP-gathered head counts to pad to the same fp8 decode "
                    f"kernel envelope; got {self.num_heads} local heads "
                    f"(pad to {self.fp8_decode_padded_heads}) vs "
                    f"{gathered_num_heads} gathered heads (pad to "
                    f"{gathered_padded_heads})"
                )

    def _build_fp8_mixed_decode_prefill(
        self,
    ) -> FlashMLASparseMetadata.FP8KernelMetadata:
        """Build FP8 metadata treating MQA tokens as one batch.

        The scheduler initializes lazily from the runtime query shape, which may
        be the full batch or only decodes when prefills use dense MHA. This avoids
        the BF16 prefill kernel's head-padding overhead at high TP.
        """

        scheduler_metadata, _ = get_mla_metadata()
        return FlashMLASparseMetadata.FP8KernelMetadata(
            scheduler_metadata=scheduler_metadata,
            cache_lens=self.max_model_len_tensor[:1],
            dummy_block_table=self.dummy_block_table[:1],
        )

    def _build_fp8_separate_prefill_decode(
        self,
        common_attn_metadata: CommonAttentionMetadata,
        metadata: FlashMLASparseMetadata,
    ) -> "FlashMLASparseMetadata.FP8SeparatePrefillDecode":
        num_tokens = common_attn_metadata.num_actual_tokens

        (num_decodes, num_prefills, num_decode_tokens, num_prefill_tokens) = (
            metadata.num_decodes,
            metadata.num_prefills,
            metadata.num_decode_tokens,
            num_tokens - metadata.num_decode_tokens,
        )

        decode_query_len = 0
        active_num_decodes = num_decodes
        if num_decodes > 0:
            query_start_loc_cpu = common_attn_metadata.query_start_loc_cpu
            decode_query_len = (query_start_loc_cpu[1] - query_start_loc_cpu[0]).item()
            assert decode_query_len > 0
            active_num_decodes = num_decode_tokens // decode_query_len
            assert active_num_decodes * decode_query_len == num_decode_tokens

        FP8Meta = FlashMLASparseMetadata.FP8SeparatePrefillDecode
        fp8_metadata = FP8Meta(
            num_decodes=active_num_decodes,
            num_prefills=num_prefills,
            num_decode_tokens=num_decode_tokens,
            num_prefill_tokens=num_prefill_tokens,
        )

        # Extract prefill sequence lengths (context + query, not just query)
        # Decode requests come first in the batch, prefill requests follow
        prefill_request_id = None
        prefill_workspace_starts = None
        prefill_chunks = None

        # For pure decode batches, prefill_request_id will be None
        # For mixed batches, it will have -1 for decode and request_id for prefill
        if num_prefills > 0:
            # Upper bound is exact for prefill rows (the `[num_decodes:]`
            # slice below), so no D2H sync is needed.
            seq_lens_cpu = common_attn_metadata.seq_lens_cpu_upper_bound
            assert seq_lens_cpu is not None
            query_start_loc_cpu = common_attn_metadata.query_start_loc_cpu

            prefill_seq_lens_cpu = seq_lens_cpu[num_decodes:]

            # Build prefill_request_id: -1 for decode, request index for
            # prefill. This enables a single
            # convert_logical_index_to_physical_index call for all tokens
            prefill_request_id = torch.full(
                (num_tokens,), -1, dtype=torch.int32, device=self.device
            )
            # Map prefill tokens to their request IDs (0, 1, 2, ...)
            for req_idx in range(num_prefills):
                # Get query token range for this prefill request
                global_req_idx = num_decodes + req_idx
                req_query_start = query_start_loc_cpu[global_req_idx]
                req_query_end = query_start_loc_cpu[global_req_idx + 1]
                prefill_request_id[req_query_start:req_query_end] = req_idx

            # will be adjusted by chunk loop
            prefill_workspace_starts_cpu = torch.zeros(
                num_prefills, dtype=torch.int32, pin_memory=True
            )
            prefill_workspace_starts_cpu[1:] = torch.cumsum(
                prefill_seq_lens_cpu[:-1], dim=0
            )
            # populated by non-blocking copy after prefill_workspace_starts_cpu is
            # updated by each chunk
            prefill_workspace_starts = torch.empty(
                num_prefills, dtype=torch.int32, device=self.device
            )

            # Chunk prefill requests to fit within workspace size
            max_prefill_buffer_size = get_prefill_workspace_size(
                self.vllm_config.model_config.max_model_len
            )
            chunk_bounds = split_prefill_chunks(
                prefill_seq_lens_cpu, max_prefill_buffer_size
            )

            prefill_chunks = []
            for chunk_start, chunk_end in chunk_bounds:
                # Adjust workspace_starts in-place per chunk to be
                # 0-indexed within each chunk
                # Example: seq_lens=[10,15,20,5], chunks=[[0,2],[2,4]]
                #   Initial: workspace_starts=[0,10,25,45]
                #   After:   workspace_starts=[0,10,0,20]
                #           (chunk 0 starts at 0, chunk 1 starts at 0)
                offset = prefill_workspace_starts_cpu[chunk_start].item()
                prefill_workspace_starts_cpu[chunk_start:chunk_end] -= offset

                chunk_tot_seqlen = prefill_seq_lens_cpu[chunk_start:chunk_end].sum()
                token_start = query_start_loc_cpu[num_decodes + chunk_start].item()
                token_end = query_start_loc_cpu[num_decodes + chunk_end].item()
                tokens_slice = slice(token_start, token_end)

                # Create chunk view of gpu tensor
                chunk_workspace_starts = prefill_workspace_starts[chunk_start:chunk_end]
                chunk_block_table = common_attn_metadata.block_table_tensor[
                    num_decodes + chunk_start : num_decodes + chunk_end
                ]
                chunk_seq_lens = common_attn_metadata.seq_lens[
                    num_decodes + chunk_start : num_decodes + chunk_end
                ]

                prefill_chunks.append(
                    FP8Meta.Prefill.Chunk(
                        tokens_slice=tokens_slice,
                        block_table=chunk_block_table,
                        req_start_idx=chunk_start,
                        workspace_starts=chunk_workspace_starts,
                        chunk_tot_seqlen=chunk_tot_seqlen,
                        seq_lens=chunk_seq_lens,
                    )
                )

            prefill_workspace_starts.copy_(
                prefill_workspace_starts_cpu, non_blocking=True
            )

            fp8_metadata.prefill = FP8Meta.Prefill(
                request_ids=prefill_request_id,
                workspace_starts=prefill_workspace_starts,
                chunks=prefill_chunks,
            )

        if num_decodes > 0:
            # Use padded head count since that's what the kernel will see
            scheduler_metadata, _ = get_mla_metadata()

            kernel_meta = FlashMLASparseMetadata.FP8KernelMetadata(
                scheduler_metadata=scheduler_metadata,
                dummy_block_table=self.dummy_block_table[:active_num_decodes],
                cache_lens=self.max_model_len_tensor[:active_num_decodes],
            )
            fp8_metadata.decode = FP8Meta.Decode(
                seq_lens=common_attn_metadata.seq_lens[:active_num_decodes],
                kernel_metadata=kernel_meta,
                decode_query_len=decode_query_len,
            )

        return fp8_metadata

    def build(
        self,
        common_prefix_len: int,
        common_attn_metadata: CommonAttentionMetadata,
        fast_build: bool = False,
    ) -> FlashMLASparseMetadata:
        metadata = super().build(common_prefix_len, common_attn_metadata, fast_build)

        metadata.fp8_use_mixed_batch = self.fp8_use_mixed_batch
        if self.use_fp8_kv_cache:
            if self.fp8_use_mixed_batch:
                metadata.fp8_extra_metadata = self._build_fp8_mixed_decode_prefill()
            else:
                metadata.fp8_extra_metadata = self._build_fp8_separate_prefill_decode(
                    common_attn_metadata, metadata
                )

        return metadata