main
py 575 lines 22.8 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 """HPC fused RoPE + QK-Norm + KV-Cache-Write (+ optional FP8 Q quant).
4
5 Decoupled from HpcAttentionImpl; extra params are passed via layer attrs.
6 """
7
8 from __future__ import annotations
9
10 import importlib.util
11 from enum import IntEnum
12 from typing import Any
13
14 import torch
15
16 from vllm.config import get_current_vllm_config_or_none
17 from vllm.forward_context import ForwardContext, get_forward_context
18 from vllm.logger import init_logger
19 from vllm.model_executor.custom_op import CustomOp
20 from vllm.model_executor.layers.hpc.hpc_module import HpcModule
21 from vllm.utils.torch_utils import direct_register_custom_op
22 from vllm.v1.attention.backends.hpc_attn import HpcAttnMetadata
23 from vllm.v1.attention.backends.registry import AttentionBackendEnum
24 from vllm.v1.attention.backends.utils import NULL_BLOCK_ID
25
26 logger = init_logger(__name__)
27
28 _hpc_rope_norm_instances: dict[str, HpcRopeNorm] = {}
29
30
31 class QkNormPolicy(IntEnum):
32 """Order of QK-RMSNorm relative to RoPE in the fused HPC rope_norm kernel.
33
34 The values are part of the HPC kernel ABI (passed through as ints), so they
35 must stay in sync with the kernel's expectations.
36 """
37
38 # No QK-Norm: apply RoPE only.
39 NONE = 0
40 # Apply RoPE first, then QK-RMSNorm.
41 ROPE_THEN_NORM = 1
42 # Apply QK-RMSNorm first, then RoPE (e.g. HunYuan V3).
43 NORM_THEN_ROPE = 2
44
45
46 def hpc_rope_norm_forward(
47 qkv: torch.Tensor,
48 output: torch.Tensor,
49 layer_name: str,
50 ) -> None:
51 """Top-level custom op: RoPE + QK-Norm + KV-Cache-Write + FP8 Q quant.
52
53 Fully opaque to torch.compile (dynamo).
54 """
55 forward_context: ForwardContext = get_forward_context()
56 attn_metadata: Any = forward_context.attn_metadata
57 if isinstance(attn_metadata, dict):
58 attn_metadata = attn_metadata[layer_name]
59
60 if attn_metadata is None:
61 output.zero_()
62 return
63
64 attn_layer = forward_context.no_compile_layers[layer_name]
65 # bind_kv_cache stores the per-layer KV cache as a single tensor
66 # (num_blocks, num_kv_heads, block_size, 2 * head_size), so use it directly.
67 kv_cache = attn_layer.kv_cache
68
69 if kv_cache.numel() == 0:
70 output.zero_()
71 return
72
73 assert kv_cache.dim() == 4, (
74 f"Expected kv_cache to have 4 dims, got {tuple(kv_cache.shape)}"
75 )
76
77 rope_norm = _hpc_rope_norm_instances[layer_name]
78 rope_norm._forward_impl(qkv, kv_cache, attn_metadata, attn_layer, output)
79
80
81 direct_register_custom_op(
82 op_name="hpc_rope_norm_forward",
83 op_func=hpc_rope_norm_forward,
84 mutates_args=["output"],
85 )
86
87
88 @CustomOp.register("hpc_rope_norm")
89 class HpcRopeNorm(CustomOp, HpcModule):
90 """HPC fused RoPE + QK-Norm + KV-Cache-Write (+ optional FP8 Q quant).
91
92 Registered as a sub-module in model layers (e.g. HunYuanAttention).
93 The QK-Norm weights are read directly from the fallback norm modules
94 (built in float32 when HPC is active), eliminating the derived copy.
95
96 forward() is dispatched by CustomOp framework:
97 - In compiled mode: forward_cuda() calls torch.ops.vllm.hpc_rope_norm_forward
98 as a splitting point — internal Python control flow is opaque
99 to torch.compile (its kernel launches can still be CUDA-graph captured).
100 - In eager/native mode: forward_native() falls back to forward_cuda().
101 """
102
103 def __init__(
104 self,
105 num_heads: int,
106 num_kv_heads: int,
107 head_dim: int,
108 cos_sin_cache: torch.Tensor,
109 use_qk_norm: bool,
110 fallback_qnorm: torch.nn.Module | None,
111 fallback_knorm: torch.nn.Module | None,
112 kv_cache_dtype: str,
113 layer_name: str,
114 qk_norm_policy: QkNormPolicy = QkNormPolicy.ROPE_THEN_NORM,
115 ) -> None:
116 super().__init__()
117 if importlib.util.find_spec("hpc") is None:
118 raise ImportError(
119 "HPCRopeNorm requires the hpc module to be installed. "
120 "Please install it from https://github.com/Tencent/hpc-ops"
121 )
122
123 self.num_heads = num_heads
124 self.num_kv_heads = num_kv_heads
125 self.head_dim = head_dim
126
127 self.q_size = num_heads * head_dim
128 self.kv_size = num_kv_heads * head_dim
129
130 # Register as a non-persistent buffer so it participates in sleep
131 # level-2 save/restore (CuMemAllocator) but is excluded from the
132 # checkpoint state_dict.
133 self.register_buffer("cos_sin_cache", cos_sin_cache.float(), persistent=False)
134
135 self.fallback_qnorm = fallback_qnorm
136 self.fallback_knorm = fallback_knorm
137
138 self.head_per_group = num_heads // num_kv_heads
139
140 self.use_fp8 = "fp8" in kv_cache_dtype
141 # The RMSNorm/RoPE ordering is model dependent (e.g. HunYuan V3 applies
142 # QK-Norm before RoPE -> NORM_THEN_ROPE), so it is supplied by the
143 # caller. When QK-Norm is disabled the policy is forced to NONE.
144 self.qk_norm_policy = qk_norm_policy if use_qk_norm else QkNormPolicy.NONE
145 if self.qk_norm_policy != QkNormPolicy.NONE and (
146 fallback_qnorm is None or fallback_knorm is None
147 ):
148 raise ValueError(
149 "QK-Norm is enabled but fallback_qnorm/fallback_knorm were not "
150 "supplied; the fused kernel reads their weights directly."
151 )
152
153 # Register layer_name + add self to the global instance registry so the
154 # module-level custom op (hpc_rope_norm_forward) can route back here.
155 self.layer_name: str | None = None
156 self.register_layer_name(layer_name)
157
158 import hpc
159
160 if self.use_fp8:
161 self._quant_type = (
162 hpc.QuantType.QPERTOKEN_PERHEAD_KPERTENSOR_VPERTENSOR.value
163 )
164 else:
165 self._quant_type = None
166
167 @classmethod
168 def support(
169 cls,
170 num_heads: int,
171 num_kv_heads: int,
172 head_dim: int,
173 kv_cache_dtype: str,
174 ) -> bool:
175 """Check whether HpcRopeNorm is supported for the given config."""
176 # HpcRopeNorm is only enabled together with the HPC attention backend.
177 vllm_config = get_current_vllm_config_or_none()
178 if (
179 vllm_config is None
180 or vllm_config.attention_config.backend != AttentionBackendEnum.HPC_ATTN
181 ):
182 return False
183
184 if kv_cache_dtype not in ("fp8_e4m3", "auto", "bfloat16"):
185 logger.warning_once(
186 f"hpc rope_norm not support kv_cache_dtype:{kv_cache_dtype}, "
187 "only support fp8_e4m3, auto, bfloat16"
188 )
189 return False
190
191 if head_dim not in (128,):
192 logger.warning_once("hpc rope_norm only support head_dim == 128.")
193 return False
194
195 head_per_group = num_heads // num_kv_heads
196 if head_per_group not in (4, 8):
197 logger.warning_once("hpc rope_norm only support head_per_group in [4, 8].")
198 return False
199
200 logger.info_once("enable hpc rope_norm")
201 return True
202
203 def register_layer_name(self, layer_name: str) -> None:
204 """Register layer_name and add self to the global registry.
205
206 The global registry is needed because the bottom-level torch op
207 (hpc_rope_norm_forward) is a module-level function and needs to
208 route back to the correct instance via layer_name.
209 """
210 self.layer_name = layer_name
211 _hpc_rope_norm_instances[layer_name] = self
212 logger.debug(
213 "[rope_norm] registered HpcRopeNorm for layer: %s",
214 layer_name,
215 )
216
217 def forward_native(
218 self,
219 qkv: torch.Tensor,
220 layer_name: str,
221 ) -> torch.Tensor:
222 """Native fallback path: delegates to forward_cuda().
223
224 For now, the default native path will use CUDA backend path.
225 Other platforms may override via OOT registration.
226 """
227 return self.forward_cuda(qkv, layer_name)
228
229 def forward_cuda(
230 self,
231 qkv: torch.Tensor,
232 layer_name: str,
233 ) -> torch.Tensor:
234 """CUDA path: invoke the torch custom op as a compile splitting point."""
235 num_tokens = qkv.shape[0]
236 output = torch.empty(
237 (num_tokens, self.num_heads, self.head_dim),
238 dtype=torch.float8_e4m3fn if self.use_fp8 else qkv.dtype,
239 device=qkv.device,
240 )
241
242 torch.ops.vllm.hpc_rope_norm_forward(qkv, output, layer_name)
243 return output
244
245 def _kv_write_scratch(self, cache: torch.Tensor) -> torch.Tensor:
246 """Throwaway dense K/V pages for the stride-aware fallback path.
247
248 ``hpc.rope_norm_store_kv[_fp8]`` always writes the paged cache itself
249 (rotated K/V *and* zero padding for the tail of every request's last
250 block), addressing it densely. When the real pages are not dense those
251 writes would land on unrelated blocks, so they are redirected here by
252 pairing this scratch cache with an all-zero block table; the real data
253 is taken out through ``out_k``/``out_v`` instead.
254
255 Returns a (2, 1, block_size, num_kv_heads, head_dim) tensor: index 0 is
256 the key scratch page, index 1 the value scratch page.
257 """
258 block_size = cache.shape[1]
259 key = (cache.dtype, block_size, cache.device)
260 if getattr(self, "_kv_scratch_key", None) != key:
261 self._kv_scratch = torch.zeros(
262 2,
263 1,
264 block_size,
265 self.num_kv_heads,
266 self.head_dim,
267 dtype=cache.dtype,
268 device=cache.device,
269 )
270 self._kv_scratch_key = key
271 return self._kv_scratch
272
273 @staticmethod
274 def _zero_pad_last_blocks(
275 key_cache: torch.Tensor,
276 value_cache: torch.Tensor,
277 seq_lens: torch.Tensor,
278 block_table: torch.Tensor,
279 ) -> None:
280 """Zero the unused tail slots of each request's last block.
281
282 The HPC attention kernels read whole blocks, so the slots past
283 ``seq_len`` must be zero. The fused op normally does this itself; on the
284 fallback path its (densely addressed) padding writes are discarded, so
285 the padding is reapplied here through the real, strided cache views.
286
287 Shapes are static and no value is inspected on the host, so this stays
288 CUDA-graph capturable. It is also self-contained: each request's last
289 block is read, masked and written back, so slots that must be preserved
290 (and requests padded in for graph capture, whose ``seq_len`` is 0) keep
291 their previous contents instead of being redirected somewhere.
292 """
293 block_size = key_cache.shape[1]
294 seq_lens = seq_lens.long()
295 # seq_len == 0 only happens for graph-capture padding; clamping keeps
296 # the gather in bounds and the all-False mask below makes the write a
297 # no-op for those rows.
298 last_block_idx = (seq_lens - 1).clamp_min(0) // block_size
299 last_block_idx = last_block_idx.clamp_max(block_table.shape[1] - 1)
300 last_block = block_table.gather(1, last_block_idx.unsqueeze(1)).squeeze(1)
301 last_block = last_block.long()
302
303 remainder = seq_lens % block_size
304 # A block that ends exactly on the boundary has no tail to clear.
305 remainder = torch.where(remainder == 0, block_size, remainder)
306
307 offsets = torch.arange(block_size, device=key_cache.device)
308 is_tail = (offsets.unsqueeze(0) >= remainder.unsqueeze(1)) & (
309 seq_lens > 0
310 ).unsqueeze(1)
311 is_tail = is_tail[..., None, None]
312
313 for cache in (key_cache, value_cache):
314 view = cache
315 if view.dtype in (torch.float8_e4m3fn, torch.float8_e5m2):
316 view = view.view(torch.uint8)
317 blocks = view.index_select(0, last_block)
318 blocks = torch.where(is_tail, blocks.new_zeros(()), blocks)
319 view.index_copy_(0, last_block, blocks)
320
321 @staticmethod
322 def _scatter_kv_cache(
323 cache: torch.Tensor,
324 src: torch.Tensor,
325 slot_mapping: torch.Tensor,
326 ) -> None:
327 """Write dense per-token K (or V) into a paged cache view.
328
329 Used on the stride-aware fallback path (see ``_forward_impl``).
330 ``cache`` is a (num_blocks, block_size, num_kv_heads, head_dim) view
331 with arbitrary block/token/head strides; ``src`` is the dense
332 (num_tokens, num_kv_heads, head_dim) buffer produced by the fused HPC
333 kernel. FP8 payloads are moved as raw bytes because ``index_copy_``
334 has no FP8 kernel.
335
336 Padded tokens carry ``PAD_SLOT_ID`` (-1); clamping sends them to the
337 reserved null block instead of tripping ``index_copy_``'s bounds check.
338 This keeps the op CUDA-graph safe (static shape, no host sync), unlike
339 masking the padded rows out.
340 """
341 # (num_blocks, block_size, ...) -> (num_slots, num_kv_heads, head_dim).
342 # Valid as a view for any layout whose block stride is a whole number
343 # of token strides, which holds for every layout the backend accepts.
344 flat = cache.flatten(0, 1)
345 if flat.dtype in (torch.float8_e4m3fn, torch.float8_e5m2):
346 flat = flat.view(torch.uint8)
347 src = src.view(torch.uint8)
348 flat.index_copy_(0, slot_mapping.clamp_min(NULL_BLOCK_ID), src)
349
350 def _forward_impl(
351 self,
352 qkv: torch.Tensor,
353 kv_cache: torch.Tensor,
354 attn_metadata: HpcAttnMetadata,
355 attn_layer: torch.nn.Module,
356 output: torch.Tensor,
357 ) -> None:
358 """Actual forward logic called by the custom op.
359
360 Writes processed q into *output* and attaches extra params
361 (e.g. FP8 scales) to *attn_layer* as attributes.
362 """
363 import hpc
364
365 # KV cache for the FP8 path is stored as uint8; view it as fp8 so the
366 # rope_norm_store_kv_fp8 kernel can write quantized K/V in-place.
367 # Must happen *before* deriving the K/V views, so that they (and any
368 # buffer whose dtype is taken from them) carry the FP8 dtype.
369 if self.use_fp8:
370 kv_cache = kv_cache.view(torch.float8_e4m3fn)
371
372 # (B, H, N, 2*hs) -> two (B, N, H, hs) K/V views, as in HpcAttentionImpl.
373 key_cache, value_cache = kv_cache.transpose(1, 2).split(self.head_dim, dim=-1)
374
375 # hpc.rope_norm_store_kv[_fp8] addresses its K/V caches as *dense*
376 # NHD tensors and ignores their strides, so it can only write the cache
377 # in place when both views are contiguous. Packed layouts (the backend's
378 # LBNHC pages interleave K and V inside the state-content dim, so each
379 # split view has a token stride of 2 * head_dim) must instead take the
380 # rotated K/V out through out_k/out_v and be scattered separately.
381 fused_kv_write = key_cache.is_contiguous() and value_cache.is_contiguous()
382 if not fused_kv_write:
383 logger.warning_once(
384 "HpcRopeNorm: KV cache pages interleave K and V, which "
385 "hpc.rope_norm_store_kv[_fp8] cannot address, so K/V are "
386 "written by a separate scatter. Update hpc-ops to a build with "
387 "strided KV cache support to re-enable the fully fused write."
388 )
389 slot_mapping = attn_metadata.slot_mapping
390
391 num_actual_tokens = attn_metadata.num_actual_tokens
392 num_prefill_reqs = attn_metadata.num_prefills
393 num_decode_reqs = attn_metadata.num_decodes
394 num_decode_tokens = attn_metadata.num_decode_tokens
395
396 qkv = qkv[:num_actual_tokens]
397
398 num_prefill_tokens = num_actual_tokens - num_decode_tokens
399
400 # Per-tensor K/V scales (shape [1]) used by the FP8 kernel.
401 k_scale = attn_layer._k_scale.reshape(1)
402 v_scale = attn_layer._v_scale.reshape(1)
403
404 if self.qk_norm_policy != QkNormPolicy.NONE:
405 assert self.fallback_qnorm is not None
406 assert self.fallback_knorm is not None
407 q_norm_weight = self.fallback_qnorm.weight
408 k_norm_weight = self.fallback_knorm.weight
409 else:
410 q_norm_weight = None
411 k_norm_weight = None
412
413 # --- Prefill ---
414 if num_prefill_reqs > 0:
415 seq_lens_prefill = attn_metadata.seq_lens[num_decode_reqs:]
416 cu_seqlens_prefill = attn_metadata.qo_indptr
417 max_seqlens = attn_metadata.max_query_len
418 block_table_prefill = attn_metadata.block_table_tensor[num_decode_reqs:]
419 qkv_prefill = qkv[num_decode_tokens:]
420 out_q_prefill = output[
421 num_decode_tokens : num_decode_tokens + num_prefill_tokens
422 ]
423
424 kv_dtype = key_cache.dtype
425 if fused_kv_write:
426 out_k = out_v = None
427 kcache_arg, vcache_arg = key_cache, value_cache
428 kv_indices_arg = block_table_prefill
429 else:
430 out_k = torch.empty(
431 num_prefill_tokens,
432 self.num_kv_heads,
433 self.head_dim,
434 dtype=kv_dtype,
435 device=qkv.device,
436 )
437 out_v = torch.empty_like(out_k)
438 # Send the kernel's own cache writes to scratch (all block ids
439 # remapped to the single scratch page).
440 scratch = self._kv_write_scratch(key_cache)
441 kcache_arg, vcache_arg = scratch[0], scratch[1]
442 kv_indices_arg = torch.zeros_like(block_table_prefill)
443
444 if self.use_fp8:
445 _, q_scale, split_k_flag = hpc.rope_norm_store_kv_fp8(
446 key_cache=kcache_arg,
447 value_cache=vcache_arg,
448 qkv=qkv_prefill,
449 cos_sin=self.cos_sin_cache,
450 num_seqlen_per_req=seq_lens_prefill,
451 q_index=cu_seqlens_prefill,
452 kvcache_indices=kv_indices_arg,
453 is_prefill=True,
454 k_scale=k_scale,
455 v_scale=v_scale,
456 quant_policy=self._quant_type,
457 max_seqlens=max_seqlens,
458 q_norm_weight=q_norm_weight,
459 k_norm_weight=k_norm_weight,
460 qk_norm_policy=self.qk_norm_policy,
461 out_q=out_q_prefill,
462 out_k=out_k,
463 out_v=out_v,
464 )
465 attn_metadata.hpc_prefill_q_scale = q_scale
466 else:
467 hpc.rope_norm_store_kv(
468 kcache_arg,
469 vcache_arg,
470 qkv_prefill,
471 self.cos_sin_cache,
472 seq_lens_prefill,
473 cu_seqlens_prefill,
474 kv_indices_arg,
475 True, # is_prefill
476 q_norm_weight=q_norm_weight,
477 k_norm_weight=k_norm_weight,
478 out_q=out_q_prefill,
479 out_k=out_k,
480 out_v=out_v,
481 qk_norm_policy=self.qk_norm_policy,
482 )
483
484 if out_k is not None:
485 self._zero_pad_last_blocks(
486 key_cache, value_cache, seq_lens_prefill, block_table_prefill
487 )
488 prefill_slots = slot_mapping[num_decode_tokens:num_actual_tokens]
489 self._scatter_kv_cache(key_cache, out_k, prefill_slots)
490 self._scatter_kv_cache(value_cache, out_v, prefill_slots)
491
492 # --- Decode ---
493 if num_decode_reqs > 0:
494 num_seq_kvcache = attn_metadata.seq_lens[:num_decode_reqs]
495 block_table_decode = attn_metadata.block_table_tensor[:num_decode_reqs]
496 qkv_decode = qkv[:num_decode_tokens]
497 # Single-token decode: q_index is the per-request prefix sum
498 # [0, 1, ..., num_decode_reqs].
499 decode_query_len = attn_metadata.decode_query_len
500 out_q_decode = output[:num_decode_tokens]
501
502 if fused_kv_write:
503 out_k = out_v = None
504 kcache_arg, vcache_arg = key_cache, value_cache
505 kv_indices_arg = block_table_decode
506 else:
507 out_k = torch.empty(
508 num_decode_tokens,
509 self.num_kv_heads,
510 self.head_dim,
511 dtype=key_cache.dtype,
512 device=qkv.device,
513 )
514 out_v = torch.empty_like(out_k)
515 scratch = self._kv_write_scratch(key_cache)
516 kcache_arg, vcache_arg = scratch[0], scratch[1]
517 kv_indices_arg = torch.zeros_like(block_table_decode)
518
519 if self.use_fp8:
520 _, q_scale, split_k_flag = hpc.rope_norm_store_kv_fp8(
521 key_cache=kcache_arg,
522 value_cache=vcache_arg,
523 qkv=qkv_decode,
524 cos_sin=self.cos_sin_cache,
525 num_seqlen_per_req=num_seq_kvcache,
526 q_index=attn_metadata.qo_indptr_decode,
527 kvcache_indices=kv_indices_arg,
528 is_prefill=False,
529 k_scale=k_scale,
530 v_scale=v_scale,
531 quant_policy=self._quant_type,
532 max_seqlens=decode_query_len,
533 q_norm_weight=q_norm_weight,
534 k_norm_weight=k_norm_weight,
535 qk_norm_policy=self.qk_norm_policy,
536 out_q=out_q_decode,
537 out_k=out_k,
538 out_v=out_v,
539 )
540 attn_metadata.hpc_decode_q_scale = q_scale
541 if split_k_flag is not None:
542 attn_metadata.hpc_split_k_flag = split_k_flag
543 else:
544 hpc.rope_norm_store_kv(
545 kcache_arg,
546 vcache_arg,
547 qkv_decode,
548 self.cos_sin_cache,
549 num_seq_kvcache,
550 attn_metadata.qo_indptr_decode,
551 kv_indices_arg,
552 False, # is_prefill
553 q_norm_weight=q_norm_weight,
554 k_norm_weight=k_norm_weight,
555 out_q=out_q_decode,
556 out_k=out_k,
557 out_v=out_v,
558 qk_norm_policy=self.qk_norm_policy,
559 )
560
561 if out_k is not None:
562 self._zero_pad_last_blocks(
563 key_cache, value_cache, num_seq_kvcache, block_table_decode
564 )
565 decode_slots = slot_mapping[:num_decode_tokens]
566 self._scatter_kv_cache(key_cache, out_k, decode_slots)
567 self._scatter_kv_cache(value_cache, out_v, decode_slots)
568
569 # Signal HpcAttentionImpl that KV cache has already been written by
570 # rope_norm_store_kv[_fp8] above, so it should skip its own
571 # reshape_and_cache_flash. Set after the fused kernels ran (either
572 # prefill or decode branch); otherwise the standard KV-write path in
573 # the attention impl would kick in (which is what non-HpcRopeNorm
574 # models rely on).
575 attn_metadata.hpc_kv_written = True