main
py 398 lines 16.1 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4 from typing import TYPE_CHECKING, ClassVar, cast
5
6 import torch
7
8 from vllm.forward_context import get_forward_context
9 from vllm.models.deepseek_v4.attention import DeepseekV4Attention
10 from vllm.models.deepseek_v4.common.ops import (
11 combine_topk_swa_indices,
12 compute_global_topk_indices_and_lens,
13 dequantize_and_gather_k_cache,
14 )
15 from vllm.models.deepseek_v4.nvidia.ops.o_proj import (
16 compute_fp8_einsum_recipe,
17 deep_gemm_fp8_o_proj,
18 )
19 from vllm.models.deepseek_v4.sparse_mla import (
20 DeepseekV4FlashMLABackend,
21 DeepseekV4FlashMLAMetadata,
22 )
23 from vllm.utils.math_utils import round_up
24 from vllm.v1.attention.backend import AttentionCGSupport
25 from vllm.v1.attention.backends.mla.sparse_swa import (
26 DeepseekSparseSWABackend,
27 DeepseekSparseSWAMetadataBuilder,
28 )
29 from vllm.v1.attention.ops.flashmla import (
30 flash_mla_sparse_fwd,
31 flash_mla_with_kvcache,
32 )
33 from vllm.v1.worker.workspace import current_workspace_manager
34
35 if TYPE_CHECKING:
36 from vllm.v1.attention.backends.mla.sparse_swa import DeepseekSparseSWAMetadata
37
38
39 class DeepseekSparseSWAFlashMLAMetadataBuilder(DeepseekSparseSWAMetadataBuilder):
40 """SWA metadata for the FlashMLA decode path, which allows varlen decode."""
41
42 _cudagraph_support: ClassVar[AttentionCGSupport] = AttentionCGSupport.ALWAYS
43
44
45 class DeepseekSparseSWAFlashMLABackend(DeepseekSparseSWABackend):
46 @staticmethod
47 def get_builder_cls() -> type[DeepseekSparseSWAFlashMLAMetadataBuilder]:
48 return DeepseekSparseSWAFlashMLAMetadataBuilder
49
50
51 class DeepseekV4FlashMLAAttention(DeepseekV4Attention):
52 """FlashMLA sparse MLA attention layer for DeepSeek V4 (CUDA)."""
53
54 backend_cls = DeepseekV4FlashMLABackend
55 swa_backend_cls = DeepseekSparseSWAFlashMLABackend
56
57 def __init__(self, *args, **kwargs) -> None:
58 super().__init__(*args, **kwargs)
59 self._einsum_recipe, self._tma_aligned_scales = compute_fp8_einsum_recipe()
60
61 def _o_proj(self, o: torch.Tensor, positions: torch.Tensor) -> torch.Tensor:
62 return deep_gemm_fp8_o_proj(
63 o,
64 positions,
65 self.rotary_emb.cos_sin_cache,
66 self.wo_a,
67 self.wo_b,
68 n_groups=self.n_local_groups,
69 heads_per_group=self.n_local_heads // self.n_local_groups,
70 nope_dim=self.nope_head_dim,
71 rope_dim=self.rope_head_dim,
72 o_lora_rank=self.o_lora_rank,
73 einsum_recipe=self._einsum_recipe,
74 tma_aligned_scales=self._tma_aligned_scales,
75 )
76
77 @classmethod
78 def get_padded_num_q_heads(cls, num_heads: int) -> int:
79 # FP8 decode kernel only supports h_q = 64 or 128.
80 if num_heads > 128:
81 raise ValueError(
82 f"DeepseekV4 FlashMLA does not support {num_heads} heads "
83 "(FP8 decode kernel requires h_q in {64, 128})."
84 )
85 return 64 if num_heads <= 64 else 128
86
87 def forward_mqa(
88 self,
89 q: torch.Tensor,
90 kv: torch.Tensor,
91 positions: torch.Tensor,
92 output: torch.Tensor,
93 ) -> None:
94 assert output.shape == q.shape, (
95 f"output buffer shape {output.shape} must match q shape {q.shape}"
96 )
97 assert output.dtype == q.dtype, (
98 f"output buffer dtype {output.dtype} must match q dtype {q.dtype}"
99 )
100
101 # Get SWA and indexer metadata from forward context
102 forward_context = get_forward_context()
103 attn_metadata = forward_context.attn_metadata
104
105 if attn_metadata is None:
106 # Warmup dummy run: no real metadata. Reserve the same bf16
107 # gather workspace _forward_prefill would; the dequantize / topk
108 # / sparse_fwd kernels are skipped this step.
109 swa_only = self.compress_ratio <= 1
110 N = (
111 0
112 if swa_only
113 else (self.max_model_len + self.compress_ratio - 1)
114 // self.compress_ratio
115 )
116 M = N + self.window_size + self.max_num_batched_tokens
117 if swa_only:
118 top_k = 0
119 else:
120 assert self.topk_indices_buffer is not None
121 top_k = self.topk_indices_buffer.shape[-1]
122 combined_topk = round_up(
123 top_k + self.window_size + self.max_image_tokens, 128
124 )
125 current_workspace_manager().get_simultaneous(
126 ((self.PREFILL_CHUNK_SIZE, M, q.shape[-1]), torch.bfloat16),
127 ((self.max_num_batched_tokens, combined_topk), torch.int32),
128 ((self.max_num_batched_tokens,), torch.int32),
129 )
130 output.zero_()
131 return
132
133 assert isinstance(attn_metadata, dict)
134 flashmla_metadata = cast(
135 DeepseekV4FlashMLAMetadata | None, attn_metadata.get(self.prefix)
136 )
137 swa_metadata = cast(
138 "DeepseekSparseSWAMetadata | None",
139 attn_metadata.get(self.swa_cache_layer.prefix),
140 )
141 assert swa_metadata is not None
142
143 swa_only = self.compress_ratio <= 1
144 # SWA-only layers (compress_ratio <= 1) don't have their own KV cache
145 # allocation, so self.kv_cache may be empty after profiling cleanup.
146 self_kv_cache = self.kv_cache if not swa_only else None
147 swa_kv_cache = self.swa_cache_layer.kv_cache
148
149 # Split prefill and decode
150 num_decodes = swa_metadata.num_decodes
151 num_prefills = swa_metadata.num_prefills
152 num_decode_tokens = swa_metadata.num_decode_tokens
153
154 if num_prefills > 0:
155 self._forward_prefill(
156 q=q[num_decode_tokens:],
157 positions=positions[num_decode_tokens:],
158 compressed_k_cache=self_kv_cache,
159 swa_k_cache=swa_kv_cache,
160 output=output[num_decode_tokens:],
161 attn_metadata=flashmla_metadata,
162 swa_metadata=swa_metadata,
163 )
164 if num_decodes > 0:
165 self._forward_decode(
166 q=q[:num_decode_tokens],
167 kv_cache=self_kv_cache,
168 swa_metadata=swa_metadata,
169 attn_metadata=flashmla_metadata,
170 swa_only=swa_only,
171 output=output[:num_decode_tokens],
172 )
173
174 def _forward_decode(
175 self,
176 q: torch.Tensor,
177 kv_cache: torch.Tensor | None, # Only used when compress_ratio > 1
178 swa_metadata: "DeepseekSparseSWAMetadata",
179 attn_metadata: DeepseekV4FlashMLAMetadata | None,
180 swa_only: bool,
181 output: torch.Tensor,
182 ) -> None:
183 num_decodes = swa_metadata.num_decodes
184 num_decode_tokens = swa_metadata.num_decode_tokens
185
186 topk_indices = None
187 topk_lens = None
188 if not swa_only:
189 assert attn_metadata is not None
190 assert swa_metadata.is_valid_token is not None
191 block_size = attn_metadata.block_size // self.compress_ratio
192 is_valid = swa_metadata.is_valid_token[:num_decode_tokens]
193 if self.compress_ratio == 4:
194 # C4A: local indices differ per layer (filled by Indexer).
195 assert self.topk_indices_buffer is not None
196 global_indices, topk_lens = compute_global_topk_indices_and_lens(
197 self.topk_indices_buffer[:num_decode_tokens],
198 swa_metadata.token_to_req_indices,
199 attn_metadata.block_table[:num_decodes],
200 block_size,
201 is_valid,
202 )
203 topk_indices = global_indices.view(num_decode_tokens, 1, -1)
204 else:
205 # C128A: pre-computed during metadata build.
206 topk_indices = attn_metadata.c128a_global_decode_topk_indices
207 topk_lens = attn_metadata.c128a_decode_topk_lens
208
209 swa_indices = swa_metadata.decode_swa_indices
210 swa_lens = swa_metadata.decode_swa_lens
211
212 # We treat queries in the same seq as different queries
213 # and later we only attend by generated indices.
214 # q arrives pre-padded to self.padded_heads by the outer wrapper.
215 q = q.unsqueeze(1)
216
217 # Prepare SWA cache (num_blocks, swa_block_size, 1, head_bytes)
218 # Use unsqueeze to preserve strides (handles padded blocks correctly)
219 swa_cache = self.swa_cache_layer.kv_cache.unsqueeze(-2)
220 # Reshape KV cache to (num_blocks, block_size, 1, head_bytes)
221 if kv_cache is not None:
222 kv_cache = kv_cache.unsqueeze(-2)
223
224 # One FlashMLASchedMeta per layer type, shared across all same-type
225 # layers within this decode step. The first forward call per type
226 # triggers the in-kernel planner (allocating tile_scheduler_metadata
227 # and num_splits via PyTorch's graph-aware allocator so CUDA graph
228 # capture reuses the same addresses on replay); subsequent same-type
229 # layers see have_initialized=True and skip the planner.
230 if self.compress_ratio <= 1:
231 tile_metadata = swa_metadata.tile_sched_swaonly
232 elif self.compress_ratio == 4:
233 tile_metadata = swa_metadata.tile_sched_c4a
234 elif self.compress_ratio == 128:
235 tile_metadata = swa_metadata.tile_sched_c128a
236 else:
237 raise ValueError(
238 f"Unsupported compress_ratio={self.compress_ratio}; "
239 "expected 1, 4, or 128."
240 )
241 assert tile_metadata is not None, (
242 "swa_metadata missing tile_sched entry for "
243 f"compress_ratio={self.compress_ratio}; "
244 "DeepseekSparseSWAMetadataBuilder.build_tile_scheduler did not "
245 "allocate one for this layer type."
246 )
247
248 out, _ = flash_mla_with_kvcache(
249 q=q,
250 k_cache=swa_cache,
251 block_table=None,
252 head_dim_v=512,
253 tile_scheduler_metadata=tile_metadata,
254 cache_seqlens=None,
255 is_fp8_kvcache=True,
256 indices=swa_indices,
257 topk_length=swa_lens,
258 softmax_scale=self.scale,
259 attn_sink=self.attn_sink,
260 extra_k_cache=kv_cache if not swa_only else None,
261 extra_indices_in_kvcache=topk_indices,
262 extra_topk_length=topk_lens,
263 out=output.unsqueeze(1),
264 )
265
266 def _forward_prefill(
267 self,
268 q: torch.Tensor,
269 positions: torch.Tensor,
270 compressed_k_cache: torch.Tensor | None, # Only used when compress_ratio > 1
271 swa_k_cache: torch.Tensor,
272 output: torch.Tensor,
273 attn_metadata: DeepseekV4FlashMLAMetadata | None,
274 swa_metadata: "DeepseekSparseSWAMetadata",
275 ) -> None:
276 swa_only = attn_metadata is None
277
278 num_prefill_tokens = swa_metadata.num_prefill_tokens
279 num_decodes = swa_metadata.num_decodes
280 num_decode_tokens = swa_metadata.num_decode_tokens
281
282 # Use pre-computed prefill metadata.
283 seq_lens = swa_metadata.prefill_seq_lens
284 gather_lens = swa_metadata.prefill_gather_lens
285 assert seq_lens is not None
286 assert gather_lens is not None
287
288 # Derive prefill-local token offsets from the full query_start_loc_cpu.
289 query_start_loc_cpu = swa_metadata.query_start_loc_cpu
290 query_start_loc = swa_metadata.query_start_loc
291 assert query_start_loc_cpu is not None
292 assert query_start_loc is not None
293 prefill_token_base = query_start_loc_cpu[num_decodes]
294
295 if not swa_only:
296 if self.compress_ratio == 4:
297 assert self.topk_indices_buffer is not None
298 topk_indices = self.topk_indices_buffer[num_decode_tokens:]
299 topk_indices = topk_indices[:num_prefill_tokens]
300 else:
301 # C128A: pre-computed during metadata build.
302 assert attn_metadata is not None
303 topk_indices = attn_metadata.c128a_prefill_topk_indices
304 top_k = topk_indices.shape[-1]
305 else:
306 # NOTE(woosuk): topk_indices will not be used for SWA-only layers.
307 assert self.topk_indices_buffer is not None
308 topk_indices = self.topk_indices_buffer[num_decode_tokens:]
309 top_k = 0
310 chunk_plan = swa_metadata.get_prefill_chunk_plan(
311 compress_ratio=self.compress_ratio,
312 prefill_chunk_size=self.PREFILL_CHUNK_SIZE,
313 )
314 assert chunk_plan, "prefill chunk plan must be non-empty when num_prefills > 0"
315 workspace_manager = current_workspace_manager()
316 combined_topk = round_up(top_k + self.window_size + self.max_image_tokens, 128)
317 for chunk_start, chunk_end, chunk_N, chunk_M in chunk_plan:
318 chunk_size = chunk_end - chunk_start
319 workspace = workspace_manager.get_simultaneous(
320 ((chunk_size, chunk_M, q.shape[-1]), torch.bfloat16),
321 ((self.max_num_batched_tokens, combined_topk), torch.int32),
322 ((self.max_num_batched_tokens,), torch.int32),
323 )
324 kv, combined_indices_out, combined_lens_out = workspace
325 if not swa_only:
326 # Gather compressed KV
327 assert attn_metadata is not None
328 block_table = attn_metadata.block_table[num_decodes:]
329 dequantize_and_gather_k_cache(
330 kv[:chunk_size],
331 compressed_k_cache,
332 seq_lens=seq_lens[chunk_start:chunk_end] // self.compress_ratio,
333 gather_lens=None,
334 block_table=block_table[chunk_start:chunk_end],
335 block_size=attn_metadata.block_size // self.compress_ratio,
336 offset=0,
337 )
338
339 # Gather SWA KV
340 swa_block_table = swa_metadata.block_table[num_decodes:]
341 dequantize_and_gather_k_cache(
342 kv[:chunk_size],
343 swa_k_cache,
344 seq_lens=seq_lens[chunk_start:chunk_end],
345 gather_lens=gather_lens[chunk_start:chunk_end],
346 block_table=swa_block_table[chunk_start:chunk_end],
347 block_size=swa_metadata.block_size,
348 offset=chunk_N,
349 )
350
351 # Combine the topk indices and SWA indices for gathered KV cache
352 query_start = (
353 query_start_loc_cpu[num_decodes + chunk_start] - prefill_token_base
354 )
355 query_end = (
356 query_start_loc_cpu[num_decodes + chunk_end] - prefill_token_base
357 )
358 combined_indices_out = combined_indices_out[: query_end - query_start]
359 combined_lens_out = combined_lens_out[: query_end - query_start]
360
361 combined_indices, combined_lens = combine_topk_swa_indices(
362 topk_indices[query_start:query_end],
363 query_start_loc[
364 num_decodes + chunk_start : num_decodes + chunk_end + 1
365 ],
366 seq_lens[chunk_start:chunk_end],
367 gather_lens[chunk_start:chunk_end],
368 self.window_size,
369 self.compress_ratio,
370 top_k,
371 chunk_M,
372 chunk_N,
373 out=(combined_indices_out, combined_lens_out),
374 left_visible=(
375 swa_metadata.prefill_left_visible[
376 num_decode_tokens + query_start : num_decode_tokens + query_end
377 ]
378 if swa_metadata.prefill_left_visible is not None
379 else None
380 ),
381 right_visible=(
382 swa_metadata.prefill_right_visible[
383 num_decode_tokens + query_start : num_decode_tokens + query_end
384 ]
385 if swa_metadata.prefill_right_visible is not None
386 else None
387 ),
388 max_image_tokens=self.max_image_tokens,
389 )
390 flash_mla_sparse_fwd(
391 q=q[query_start:query_end],
392 kv=kv.view(-1, 1, q.shape[-1]),
393 indices=combined_indices.unsqueeze(1),
394 sm_scale=self.scale,
395 attn_sink=self.attn_sink,
396 topk_length=combined_lens,
397 out=output[query_start:query_end],
398 )