main
py 554 lines 21 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 """NVIDIA QSA owner with Triton kernels."""
4
5 from __future__ import annotations
6
7 from typing import ClassVar, cast
8
9 import torch
10 from torch import nn
11 from transformers import Qwen4ExpTextConfig
12
13 from vllm.compilation.breakable_cudagraph import eager_break_during_capture
14 from vllm.config import VllmConfig
15 from vllm.config.cache import CacheDType
16 from vllm.distributed import get_tensor_model_parallel_world_size
17 from vllm.forward_context import get_forward_context
18 from vllm.model_executor.layers.attention.attention import (
19 set_default_quant_scales,
20 )
21 from vllm.model_executor.layers.attention_layer_base import AttentionLayerBase
22 from vllm.model_executor.layers.layernorm import GemmaRMSNorm
23 from vllm.model_executor.layers.linear import QKVParallelLinear, RowParallelLinear
24 from vllm.model_executor.layers.quantization import QuantizationConfig
25 from vllm.model_executor.layers.rotary_embedding import MRotaryEmbedding, get_rope
26 from vllm.model_executor.models.qwen3_next import Qwen3NextAttention
27 from vllm.platforms import current_platform
28 from vllm.platforms.interface import DeviceCapability
29 from vllm.utils.torch_utils import (
30 kv_cache_dtype_str_to_dtype,
31 )
32 from vllm.v1.attention.backend import (
33 AttentionBackend,
34 AttentionCGSupport,
35 AttentionType,
36 MultipleOf,
37 )
38 from vllm.v1.attention.backends.fa_utils import is_flash_attn_varlen_func_available
39 from vllm.v1.attention.backends.flash_attn import (
40 FlashAttentionBackend,
41 FlashAttentionImpl,
42 FlashAttentionMetadata,
43 FlashAttentionMetadataBuilder,
44 )
45 from vllm.v1.kv_cache_interface import (
46 FullAttentionSpec,
47 KVCacheSpec,
48 get_kv_quant_mode,
49 )
50
51 from ..common.qsa_cache import QSAForwardMetadata
52 from . import model
53 from .indexer_qsa import QSAIndexer
54
55
56 class Qwen4ExpQSAMetadataBuilder(FlashAttentionMetadataBuilder):
57 """Flash metadata supporting uniform decode and target-verify graphs."""
58
59 _cudagraph_support: ClassVar[AttentionCGSupport] = AttentionCGSupport.UNIFORM_BATCH
60
61
62 class Qwen4ExpQSAFlashAttentionBackend(FlashAttentionBackend):
63 """FullAttentionSpec backend used by the merged QSA owner."""
64
65 supported_dtypes: ClassVar[list[torch.dtype]] = [torch.bfloat16]
66 # fp8/fp8_e4m3: e4m3 bytes in a uint8 cache, written by reshape_and_cache
67 # with the layer's per-tensor scales and dequantized on load inside the QSA
68 # Triton kernel. flash-attn never runs over this cache, so its fp8 probe
69 # does not apply (see supports_kv_cache_dtype and the impl constructor).
70 supported_kv_cache_dtypes: ClassVar[list[CacheDType]] = [
71 "auto",
72 "bfloat16",
73 "fp8",
74 "fp8_e4m3",
75 ]
76
77 @classmethod
78 def supports_kv_cache_dtype(cls, kv_cache_dtype: CacheDType | None) -> bool:
79 return kv_cache_dtype is None or kv_cache_dtype in cls.supported_kv_cache_dtypes
80
81 @classmethod
82 def supports_combination(
83 cls,
84 head_size: int,
85 dtype: torch.dtype,
86 kv_cache_dtype: CacheDType | None,
87 block_size: int | None,
88 use_mla: bool,
89 has_sink: bool,
90 use_sparse: bool,
91 use_mm_prefix: bool,
92 device_capability: DeviceCapability,
93 ) -> str | None:
94 # QSA dequantizes the fp8 KV in its own Triton kernel and never runs
95 # flash-attn over the quantized cache, so the parent's fp8-KV rejection
96 # does not apply and every combination it is handed is accepted here.
97 return None
98
99 @staticmethod
100 def get_name() -> str:
101 return "QWEN4_EXP_QSA_TRITON"
102
103 @staticmethod
104 def get_supported_kernel_block_sizes(kv_cache_spec=None) -> list[int | MultipleOf]:
105 # QSA consumes manager pages directly and does not use FA4 paged attention.
106 return [MultipleOf(16)]
107
108 @staticmethod
109 def get_impl_cls() -> type[Qwen4ExpQSAFlashAttentionImpl]:
110 return Qwen4ExpQSAFlashAttentionImpl
111
112 @staticmethod
113 def get_builder_cls() -> type[Qwen4ExpQSAMetadataBuilder]:
114 return Qwen4ExpQSAMetadataBuilder
115
116 @classmethod
117 def is_sparse(cls) -> bool:
118 return True
119
120 @classmethod
121 def supports_kv_connector(cls) -> bool:
122 return False
123
124
125 class Qwen4ExpQSAFlashAttentionImpl(FlashAttentionImpl):
126 """Run paged sparse GQA with the QSA Triton kernel."""
127
128 supports_dcp: bool = False
129 supports_pcp: bool = False
130
131 def __init__(
132 self,
133 num_heads: int,
134 head_size: int,
135 scale: float,
136 num_kv_heads: int,
137 alibi_slopes: list[float] | None,
138 sliding_window: int | None,
139 kv_cache_dtype: str,
140 logits_soft_cap: float | None = None,
141 attn_type: AttentionType = AttentionType.DECODER,
142 kv_sharing_target_layer_name: str | None = None,
143 sinks: torch.Tensor | None = None,
144 ) -> None:
145 # The parent constructor probes flash-attn for quantized-KV support and
146 # raises where it is unavailable (sm120), but QSA dequantizes fp8 inside
147 # its own Triton kernel and never runs flash-attn over the cache. Hand
148 # the parent "auto" for that probe and restore the real dtype afterwards:
149 # the parent only uses it there, and do_kv_cache_update reads the
150 # attribute at call time.
151 real_kv_cache_dtype = kv_cache_dtype
152 if kv_cache_dtype in ("fp8", "fp8_e4m3"):
153 kv_cache_dtype = "auto"
154 super().__init__(
155 num_heads,
156 head_size,
157 scale,
158 num_kv_heads,
159 alibi_slopes,
160 sliding_window,
161 kv_cache_dtype,
162 logits_soft_cap,
163 attn_type,
164 kv_sharing_target_layer_name,
165 sinks,
166 )
167 self.kv_cache_dtype = real_kv_cache_dtype
168 if not is_flash_attn_varlen_func_available():
169 raise NotImplementedError("Qwen4Exp QSA requires FlashAttention")
170 if self.dcp_world_size != 1:
171 raise NotImplementedError(
172 "Qwen4Exp QSA does not support decode context parallelism"
173 )
174 if self.kv_cache_dtype not in ("auto", "bfloat16", "fp8", "fp8_e4m3"):
175 raise NotImplementedError(
176 "Qwen4Exp QSA requires a BF16 or FP8-e4m3 main KV cache"
177 )
178 self.supports_quant_query_input = False
179
180 def forward_qsa(
181 self,
182 layer: torch.nn.Module,
183 query: torch.Tensor,
184 key: torch.Tensor,
185 value: torch.Tensor,
186 kv_cache: torch.Tensor,
187 attn_metadata: FlashAttentionMetadata,
188 output: torch.Tensor,
189 token_to_req: torch.Tensor,
190 use_prefill_config: bool,
191 output_gate: torch.Tensor,
192 output_scale: torch.Tensor | None = None,
193 output_block_scale: torch.Tensor | None = None,
194 ) -> torch.Tensor:
195 del key, value
196 if output_scale is not None or output_block_scale is not None:
197 raise NotImplementedError("QSA does not support fused output quantization")
198 if self.alibi_slopes is not None or self.sinks is not None:
199 raise NotImplementedError("QSA does not support ALiBi or attention sinks")
200 if self.sliding_window != (-1, -1):
201 raise NotImplementedError("QSA does not support sliding-window attention")
202
203 num_tokens = attn_metadata.num_actual_tokens
204 output.zero_()
205 if num_tokens == 0:
206 return output
207
208 topk_buffer = getattr(layer, "topk_indices_buffer", None)
209 if topk_buffer is None:
210 raise RuntimeError("QSA owner did not provide its top-k buffer")
211 logical_indices = topk_buffer[:num_tokens]
212 token_to_req = token_to_req[:num_tokens]
213 key_cache, value_cache = kv_cache.transpose(1, 2).split(self.head_size, dim=-1)
214 k_scale = v_scale = None
215 if self.kv_cache_dtype in ("fp8", "fp8_e4m3"):
216 # The cache is allocated as uint8; reinterpret the e4m3 bytes
217 # (same itemsize, so shape and strides are preserved).
218 key_cache = key_cache.view(torch.float8_e4m3fn)
219 value_cache = value_cache.view(torch.float8_e4m3fn)
220 # Host-side per-tensor dequant scales (Python floats), as used by
221 # other host-scale backends; folded into the kernel's scales.
222 k_scale = layer._k_scale_float
223 v_scale = layer._v_scale_float
224 if query.dtype != torch.bfloat16 or key_cache.dtype not in (
225 torch.bfloat16,
226 torch.float8_e4m3fn,
227 ):
228 raise NotImplementedError(
229 "Qwen4Exp QSA requires BF16 Q and BF16 or FP8-e4m3 K/V"
230 )
231
232 from .ops.qsa import qsa_sparse_paged_attention
233
234 qsa_sparse_paged_attention(
235 query[:num_tokens],
236 key_cache,
237 value_cache,
238 logical_indices,
239 attn_metadata.block_table,
240 token_to_req,
241 use_prefill_config,
242 output[:num_tokens],
243 k_scale=k_scale,
244 v_scale=v_scale,
245 output_gate=output_gate[:num_tokens],
246 )
247 return output
248
249
250 class Qwen4ExpQSAAttention(Qwen3NextAttention, AttentionLayerBase):
251 """Merged Qwen full-attention owner with a QSA index side branch."""
252
253 supports_dcp = False
254
255 def __init__(
256 self,
257 *,
258 vllm_config: VllmConfig,
259 config: Qwen4ExpTextConfig,
260 layer_id: int,
261 quant_config: QuantizationConfig | None = None,
262 reduce_results: bool = True,
263 prefix: str = "",
264 ) -> None:
265 nn.Module.__init__(self)
266 cache_config = vllm_config.cache_config
267 model_config = vllm_config.model_config
268 if cache_config is None:
269 raise ValueError("Qwen4Exp QSA requires a paged KV cache")
270 if model_config.dtype != torch.bfloat16:
271 raise NotImplementedError("Qwen4Exp QSA currently requires BF16")
272 if cache_config.cache_dtype not in (
273 "auto",
274 "bfloat16",
275 "fp8",
276 "fp8_e4m3",
277 ):
278 raise NotImplementedError(
279 "Qwen4Exp QSA requires a BF16 or FP8-e4m3 main KV cache"
280 )
281 if getattr(quant_config, "kv_cache_scheme", None) is not None:
282 raise NotImplementedError("Qwen4Exp QSA does not support KV quantization")
283 parallel_config = vllm_config.parallel_config
284 if (
285 parallel_config.prefill_context_parallel_size > 1
286 or parallel_config.decode_context_parallel_size > 1
287 ):
288 raise NotImplementedError(
289 "Qwen4Exp QSA does not support context parallelism"
290 )
291 if not getattr(config, "is_causal", True):
292 raise NotImplementedError("Qwen4Exp QSA requires causal decoder attention")
293
294 self.config = config
295 self.hidden_size = int(config.hidden_size)
296 tp_size = get_tensor_model_parallel_world_size()
297 self.total_num_heads = int(config.num_attention_heads)
298 if self.total_num_heads % tp_size:
299 raise ValueError("QSA attention heads must be divisible by TP size")
300 self.num_heads = self.total_num_heads // tp_size
301 # Decode/verify batches have at most 1 + num_spec query tokens per
302 # request; use_prefill_config (max_query_len > this) steers the
303 # config table. Shorter batches take the decode profile — harmless,
304 # the difference is tile-shape tuning, not correctness.
305 self._max_decode_query_len = 1 + vllm_config.num_speculative_tokens
306 self.total_num_kv_heads = int(config.num_key_value_heads)
307 if self.total_num_kv_heads >= tp_size:
308 if self.total_num_kv_heads % tp_size:
309 raise ValueError("QSA KV heads must be divisible by TP size")
310 elif tp_size % self.total_num_kv_heads:
311 raise ValueError("TP size must be divisible by replicated QSA KV heads")
312 self.num_kv_heads = max(1, self.total_num_kv_heads // tp_size)
313 self.head_dim = int(config.head_dim or self.hidden_size // self.num_heads)
314 self.q_size = self.num_heads * self.head_dim
315 self.kv_size = self.num_kv_heads * self.head_dim
316 self.scaling = self.head_dim**-0.5
317 self.dual_chunk_attention_config = getattr(
318 config, "dual_chunk_attention_config", None
319 )
320 if self.dual_chunk_attention_config is not None:
321 raise NotImplementedError("Qwen4Exp QSA does not support dual-chunk RoPE")
322 # Qwen4Exp full-attention checkpoints always pack a sigmoid output
323 # gate next to Q, even when an inherited config default says otherwise.
324 self.attn_output_gate = True
325
326 self.qkv_proj = QKVParallelLinear(
327 self.hidden_size,
328 self.head_dim,
329 self.total_num_heads * (1 + self.attn_output_gate),
330 self.total_num_kv_heads,
331 bias=False,
332 quant_config=model.without_modelopt_fp4(quant_config),
333 prefix=f"{prefix}.qkv_proj",
334 )
335 self.o_proj = RowParallelLinear(
336 self.total_num_heads * self.head_dim,
337 self.hidden_size,
338 bias=False,
339 reduce_results=reduce_results,
340 quant_config=quant_config,
341 prefix=f"{prefix}.o_proj",
342 )
343 self.rotary_emb = get_rope(
344 head_size=self.head_dim,
345 max_position=config.max_position_embeddings,
346 rope_parameters=config.rope_parameters,
347 )
348 self.q_norm = GemmaRMSNorm(self.head_dim, eps=config.rms_norm_eps)
349 self.k_norm = GemmaRMSNorm(self.head_dim, eps=config.rms_norm_eps)
350
351 mm_config = model_config.multimodal_config
352 text_only = mm_config is None or mm_config.language_model_only
353 mrope_section = getattr(self.rotary_emb, "mrope_section", None)
354 supports_mrope = bool(
355 type(self.rotary_emb) is MRotaryEmbedding
356 and mrope_section
357 and len(mrope_section) == 3
358 and sum(mrope_section) == self.rotary_emb.rotary_dim // 2
359 and getattr(self.rotary_emb, "mrope_interleaved", False)
360 )
361 supports_dtype = getattr(self.rotary_emb, "dtype", None) in (
362 torch.float16,
363 torch.bfloat16,
364 )
365 self.use_fused_qk_norm_rope_gate = (
366 self.attn_output_gate
367 and getattr(self.rotary_emb, "is_neox_style", False)
368 and current_platform.is_cuda()
369 and supports_dtype
370 and (text_only or supports_mrope)
371 )
372
373 self.layer_name = f"{prefix}.attn"
374 self.attn_type = AttentionType.DECODER
375 self.kv_cache_dtype = cache_config.cache_dtype
376 self.kv_cache_torch_dtype = kv_cache_dtype_str_to_dtype(
377 self.kv_cache_dtype, model_config
378 )
379 if self.kv_cache_torch_dtype not in (torch.bfloat16, torch.uint8):
380 raise NotImplementedError(
381 "Qwen4Exp QSA requires BF16 or FP8-e4m3 (uint8) cache storage"
382 )
383 self.kv_sharing_target_layer_name = None
384 self.kv_cache = torch.tensor([])
385 set_default_quant_scales(self, register_buffer=True)
386
387 self.attn_backend = Qwen4ExpQSAFlashAttentionBackend
388 self.impl = Qwen4ExpQSAFlashAttentionImpl(
389 self.num_heads,
390 self.head_dim,
391 self.scaling,
392 self.num_kv_heads,
393 None,
394 None,
395 self.kv_cache_dtype,
396 None,
397 AttentionType.DECODER,
398 None,
399 )
400 self.indexer = QSAIndexer(
401 vllm_config=vllm_config,
402 config=config,
403 layer_id=layer_id,
404 rotary_emb=self.rotary_emb,
405 quant_config=quant_config,
406 prefix=f"{prefix}.indexer",
407 )
408 # One launch does the indexer prepare, the main QK-norm/RoPE/gate and
409 # the main K/V cache write (see QSAIndexer.forward); otherwise all of
410 # them take the separate kernels.
411 self.use_fused_qsa_prepare = (
412 self.use_fused_qk_norm_rope_gate and self.indexer.use_fused_pre_indexer
413 )
414 max_tokens = vllm_config.scheduler_config.max_num_batched_tokens
415 # PACKED selection buffer: the trailing column holds each row's
416 # valid-entry count (written by the expand kernel) — never a token
417 # index; the sparse attention kernel reads it as its loop bound.
418 # MTP skip_topk steps reuse rows frozen from step 0; the count is
419 # a row column, so compaction/reuse keep it paired with the content.
420 self.register_buffer(
421 "topk_indices_buffer",
422 torch.empty(
423 max_tokens,
424 self.indexer.packed_output_width,
425 dtype=torch.int32,
426 ),
427 persistent=False,
428 )
429
430 static_context = vllm_config.compilation_config.static_forward_context
431 if self.layer_name in static_context:
432 raise ValueError(f"Duplicate layer name: {self.layer_name}")
433 static_context[self.layer_name] = self
434
435 def get_attn_backend(self) -> type[AttentionBackend]:
436 return self.attn_backend
437
438 def get_kv_cache_spec(self, vllm_config: VllmConfig) -> KVCacheSpec:
439 return FullAttentionSpec(
440 block_size=vllm_config.cache_config.block_size,
441 num_kv_heads=self.num_kv_heads,
442 head_size=self.head_dim,
443 head_size_v=self.head_dim,
444 dtype=self.kv_cache_torch_dtype,
445 kv_quant_mode=get_kv_quant_mode(self.kv_cache_dtype),
446 )
447
448 @eager_break_during_capture
449 def _run_qsa(
450 self,
451 projected_qk: torch.Tensor,
452 positions: torch.Tensor,
453 query: torch.Tensor | None,
454 key: torch.Tensor | None,
455 value: torch.Tensor | None,
456 output: torch.Tensor,
457 output_gate: torch.Tensor | None,
458 qkv: torch.Tensor,
459 ) -> None:
460 # query/key/value/output_gate are None when the fused prepare runs
461 # inside the indexer launch.
462 metadata = get_forward_context().attn_metadata
463 if isinstance(metadata, list):
464 metadata = metadata[0]
465 if not isinstance(metadata, dict):
466 output.zero_()
467 return
468 main_metadata = cast(FlashAttentionMetadata, metadata[self.layer_name])
469 if self.kv_cache.numel() == 0:
470 raise RuntimeError("QSA main K/V cache is not bound")
471
472 num_tokens = main_metadata.num_actual_tokens
473 side_metadata = cast(
474 QSAForwardMetadata,
475 metadata[self.indexer.raw_key_cache.prefix],
476 )
477 if side_metadata.num_actual_tokens != num_tokens:
478 raise RuntimeError("QSA main and side metadata token counts disagree")
479 selected, main_outputs = self.indexer(
480 projected_qk,
481 positions,
482 self.topk_indices_buffer[:num_tokens],
483 attn=self,
484 qkv=qkv,
485 slot_mapping=main_metadata.slot_mapping,
486 )
487 if selected.shape != (num_tokens, self.indexer.packed_output_width):
488 raise RuntimeError("QSA indexer returned an invalid selection shape")
489 impl = cast(Qwen4ExpQSAFlashAttentionImpl, self.impl)
490 if main_outputs is None:
491 assert key is not None and value is not None
492 impl.do_kv_cache_update(
493 self,
494 key,
495 value,
496 self.kv_cache,
497 main_metadata.slot_mapping,
498 )
499 else:
500 query, output_gate = main_outputs
501 assert query is not None and output_gate is not None
502 impl.forward_qsa(
503 self,
504 query,
505 key,
506 value,
507 self.kv_cache,
508 main_metadata,
509 output,
510 token_to_req=side_metadata.token_to_req,
511 use_prefill_config=main_metadata.max_query_len > self._max_decode_query_len,
512 output_gate=output_gate,
513 )
514
515 def forward(
516 self,
517 positions: torch.Tensor,
518 hidden_states: torch.Tensor,
519 ) -> torch.Tensor:
520 qkv, _ = self.qkv_proj(hidden_states)
521 num_tokens = hidden_states.shape[0]
522 if not self.use_fused_qsa_prepare:
523 q, k, v, gate = self._project_qkv_gate(qkv, positions)
524 assert gate is not None
525 query = q.view(num_tokens, self.num_heads, self.head_dim)
526 key = k.view(num_tokens, self.num_kv_heads, self.head_dim)
527 value = v.view(num_tokens, self.num_kv_heads, self.head_dim)
528 else:
529 # Norm/RoPE/gate and the K/V cache write happen inside _run_qsa.
530 query = key = value = gate = None
531 attn_output = qkv.new_empty(num_tokens, self.num_heads, self.head_dim)
532 # Keep the index projection outside the eager break.
533 projected_qk, _ = self.indexer.index_qk_proj(hidden_states)
534 self._run_qsa(
535 projected_qk,
536 positions,
537 query,
538 key,
539 value,
540 attn_output,
541 gate,
542 qkv,
543 )
544 flat_output = attn_output.view(num_tokens, -1)
545 output, _ = self.o_proj(flat_output)
546 return output
547
548
549 __all__ = [
550 "QSAIndexer",
551 "Qwen4ExpQSAAttention",
552 "Qwen4ExpQSAFlashAttentionBackend",
553 "Qwen4ExpQSAFlashAttentionImpl",
554 ]