layer.py 26.6 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
"""Attention layer."""
4

5
from typing import Callable, Optional
6
7
8

import torch
import torch.nn as nn
9
import torch.nn.functional as F
10

11
import vllm.envs as envs
12
from vllm.attention import AttentionType
13
from vllm.attention.backends.abstract import AttentionBackend
14
15
from vllm.attention.backends.registry import _Backend, backend_name_to_enum
from vllm.attention.selector import get_attn_backend
16
from vllm.attention.utils.kv_sharing_utils import validate_kv_sharing_target
17
from vllm.config import CacheConfig, get_current_vllm_config
18
19
20
21
22
from vllm.distributed.kv_transfer import (
    get_kv_transfer_group,
    has_kv_transfer_group,
    is_v1_kv_transfer_group,
)
23
from vllm.forward_context import ForwardContext, get_forward_context
24
from vllm.logger import init_logger
25
from vllm.model_executor.layers.attention_layer_base import AttentionLayerBase
26
from vllm.model_executor.layers.linear import UnquantizedLinearMethod
27
from vllm.model_executor.layers.quantization import QuantizationConfig
28
from vllm.model_executor.layers.quantization.input_quant_fp8 import QuantFP8
29
from vllm.model_executor.layers.quantization.kv_cache import BaseKVCacheMethod
30
from vllm.model_executor.layers.quantization.utils.quant_utils import GroupShape
31
from vllm.model_executor.models.vision import get_vit_attn_backend
32
from vllm.platforms import current_platform
33
from vllm.utils import GiB_bytes, direct_register_custom_op
34

35
36
logger = init_logger(__name__)
USE_XFORMERS_OPS = None
37
try:
38
    tag_cudagraph_unsafe = (torch._C.Tag.cudagraph_unsafe,)
39
40
except AttributeError:
    tag_cudagraph_unsafe = ()  # type: ignore[assignment]
41
42
43
44
45
46
47


def check_xformers_availability():
    global USE_XFORMERS_OPS
    if USE_XFORMERS_OPS is not None:
        return USE_XFORMERS_OPS

48
    if current_platform.is_cuda() and current_platform.has_device_capability(100):
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
        # Xformers FA is not compatible with B200
        USE_XFORMERS_OPS = False
    else:
        try:
            from importlib.util import find_spec

            find_spec("xformers.ops")
            USE_XFORMERS_OPS = True
        except ImportError:
            USE_XFORMERS_OPS = False

    # the warning only needs to be shown once
    if not USE_XFORMERS_OPS:
        logger.warning("Xformers is not available, falling back.")

    return USE_XFORMERS_OPS

66

67
def check_upstream_fa_availability(dtype: torch.dtype):
68
69
70
71
72
    if (
        dtype in (torch.float16, torch.bfloat16)
        and current_platform.is_cuda()
        and current_platform.has_device_capability(80)
    ):
73
        from transformers.utils import is_flash_attn_2_available
74

75
        return is_flash_attn_2_available()
76
77
    if current_platform.is_rocm():
        from importlib.util import find_spec
78

79
        return find_spec("flash_attn") is not None
80
81
82
    return False


83
def maybe_get_vit_flash_attn_backend(
84
85
86
87
88
89
90
    attn_backend: _Backend, use_upstream_fa: bool
) -> tuple[_Backend, Callable]:
    if (
        attn_backend != _Backend.FLASH_ATTN
        and attn_backend != _Backend.ROCM_AITER_FA
        and check_upstream_fa_availability(torch.get_default_dtype())
    ):
91
92
93
        attn_backend = _Backend.FLASH_ATTN
        use_upstream_fa = True

94
    if current_platform.is_rocm() and attn_backend == _Backend.FLASH_ATTN:
95
96
        use_upstream_fa = True

97
    if attn_backend in {_Backend.FLASH_ATTN, _Backend.ROCM_AITER_FA}:
98
99
100
101
102
103
104
105
106
107
108
109
110
        if attn_backend == _Backend.ROCM_AITER_FA:
            from aiter import flash_attn_varlen_func
        else:
            if use_upstream_fa:
                from flash_attn import flash_attn_varlen_func
            else:
                from vllm.vllm_flash_attn import flash_attn_varlen_func
    else:
        flash_attn_varlen_func = None

    return attn_backend, flash_attn_varlen_func


111
class Attention(nn.Module, AttentionLayerBase):
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
    """Attention layer.

    This class takes query, key, and value tensors as input. The input tensors
    can either contain prompt tokens or generation tokens.
    The class does the following:

    1. Store the input key and value tensors in the KV cache.
    2. Perform (multi-head/multi-query/grouped-query) attention.
    3. Return the output tensor.
    """

    def __init__(
        self,
        num_heads: int,
        head_size: int,
        scale: float,
        num_kv_heads: Optional[int] = None,
129
        alibi_slopes: Optional[list[float]] = None,
130
        cache_config: Optional[CacheConfig] = None,
131
        quant_config: Optional[QuantizationConfig] = None,
132
        logits_soft_cap: Optional[float] = None,
133
        per_layer_sliding_window: Optional[int] = None,
134
        use_mla: bool = False,
135
        use_sparse: bool = False,
136
        prefix: str = "",
137
        attn_type: str = AttentionType.DECODER,
138
        kv_sharing_target_layer_name: Optional[str] = None,
139
        attn_backend: Optional[type[AttentionBackend]] = None,
140
        **extra_impl_args,
141
    ) -> None:
142
143
144
145
        """
        The KV cache is stored inside this class and is accessed via
        `self.kv_cache`.
        """
146
        super().__init__()
147
148
149
150
151
152
153
154
155
        if per_layer_sliding_window is not None:
            # per-layer sliding window
            sliding_window = per_layer_sliding_window
        elif cache_config is not None:
            # model-level sliding window
            sliding_window = cache_config.sliding_window
        else:
            sliding_window = None

156
157
158
        if cache_config is not None:
            kv_cache_dtype = cache_config.cache_dtype
            block_size = cache_config.block_size
159
            calculate_kv_scales = cache_config.calculate_kv_scales
160
161
162
        else:
            kv_cache_dtype = "auto"
            block_size = 16
163
            calculate_kv_scales = False
164
165
        if num_kv_heads is None:
            num_kv_heads = num_heads
166
167
168
        assert num_heads % num_kv_heads == 0, (
            f"num_heads ({num_heads}) is not divisible by num_kv_heads ({num_kv_heads})"
        )
169

170
        # The default k/v_scale is set to 1.0. This is ignored
171
172
        # when kv-cache is not fp8, and should be used with
        # kv-cache in fp8_e5m2. For kv-cache in fp8_e4m3, we
173
        # expect the pre-quantized k/v_scale to be loaded along
174
175
        # with the model weights.
        self.kv_cache_dtype = kv_cache_dtype
176
177
178
        self.calculate_kv_scales = calculate_kv_scales
        self._k_scale = torch.tensor(1.0, dtype=torch.float32)
        self._v_scale = torch.tensor(1.0, dtype=torch.float32)
179
180
181
        # FlashAttn doesn't support quantizing the kv-cache only
        # but requires q to be quantized as well.
        self._q_scale = torch.tensor(1.0, dtype=torch.float32)
182
        self._prob_scale = torch.tensor(1.0, dtype=torch.float32)
183

184
185
186
187
        # We also keep q/k/v_scale on host (cpu) memory for attention
        # backends that require the scales to be on host instead of on device.
        # e.g. Flashinfer
        self._q_scale_float = 1.0
188
189
190
        self._k_scale_float = 1.0
        self._v_scale_float = 1.0

191
192
193
194
        # The output scale on host memory. This should be the input scale of
        # the quant op after this attention layer.
        self._o_scale_float: Optional[float] = None

195
        self.use_mla = use_mla
196
        self.use_sparse = use_sparse
197
198
199
200
        self.num_heads = num_heads
        self.head_size = head_size
        self.num_kv_heads = num_kv_heads
        self.sliding_window = sliding_window
201
        self.has_sink = extra_impl_args.get("sinks") is not None
202

203
204
205
        quant_method = (
            quant_config.get_quant_method(self, prefix=prefix) if quant_config else None
        )
206
        if quant_method is not None and not isinstance(
207
208
            quant_method, UnquantizedLinearMethod
        ):
209
            assert isinstance(quant_method, BaseKVCacheMethod)
210
211
            # TODO (mgoin): kv cache dtype should be specified in the FP8
            # checkpoint config and become the "auto" behavior
212
            if self.kv_cache_dtype == "fp8_e5m2":
213
214
215
                raise ValueError(
                    "fp8_e5m2 kv-cache is not supported with fp8 checkpoints."
                )
216
217
218
219
220
221
            # If quantization is enabled, we make "k_scale" and "v_scale"
            # parameters so that it can be loaded from the model checkpoint.
            # The k/v_scale will then be converted back to native float32
            # values after weight loading.
            self.quant_method = quant_method
            self.quant_method.create_weights(self)
222

223
224
225
        # During model initialization, the default dtype is set as the model
        # weight and activation dtype.
        dtype = torch.get_default_dtype()
226
        if attn_backend is None:
227
228
229
230
231
232
233
234
235
            self.attn_backend = get_attn_backend(
                head_size,
                dtype,
                kv_cache_dtype,
                block_size,
                use_mla=use_mla,
                has_sink=self.has_sink,
                use_sparse=use_sparse,
            )
236
237
238
239
        else:
            self.attn_backend = attn_backend

        impl_cls = self.attn_backend.get_impl_cls()
240
241
242
243
244
245
246
247
248
249
250
251
252
        self.impl = impl_cls(
            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,
            **extra_impl_args,
        )
253
        self.backend = backend_name_to_enum(self.attn_backend.get_name())
254
        self.dtype = dtype
255

256
257
258
259
        # For cuda-alike (CUDA and ROCM) and cpu platforms, we control how
        # torch.compile works by registering the attention as one giant
        # opaque custom op. For other platforms, we directly call them
        # and let torch.compile handle them.
260
        self.use_direct_call = not current_platform.opaque_attention_op()
261

262
        self.use_output = self.attn_backend.accept_output_buffer
263
264
265
266
267
        compilation_config = get_current_vllm_config().compilation_config
        if prefix in compilation_config.static_forward_context:
            raise ValueError(f"Duplicate layer name: {prefix}")
        compilation_config.static_forward_context[prefix] = self
        self.layer_name = prefix
268
        self.attn_type = attn_type
269
270
271
272
273
274
275
276
277

        if kv_sharing_target_layer_name is not None:
            validate_kv_sharing_target(
                prefix,
                kv_sharing_target_layer_name,
                compilation_config.static_forward_context,
            )
        self.kv_sharing_target_layer_name = kv_sharing_target_layer_name

278
279
280
281
        # use a placeholder kv cache tensor during init, which will be replaced
        # by bind_kv_cache
        # this variable will not be accessed if use_direct_call is True
        self.kv_cache = [
282
283
284
285
            torch.tensor([])
            for _ in range(
                get_current_vllm_config().parallel_config.pipeline_parallel_size
            )
286
        ]
287

288
        try:
289
290
291
            self.q_range = torch.tensor(envs.Q_SCALE_CONSTANT, dtype=torch.float32)
            self.k_range = torch.tensor(envs.K_SCALE_CONSTANT, dtype=torch.float32)
            self.v_range = torch.tensor(envs.V_SCALE_CONSTANT, dtype=torch.float32)
292
        except torch.cuda.OutOfMemoryError as e:
293
            logger.error("Failed to initialize attention q/k/v range constants: %s", e)
294
295
            if torch.cuda.is_available():
                logger.debug("CUDA device: %s", torch.cuda.current_device())
296
297
298
299
300
301
                logger.debug(
                    "Allocated: %.2f GiB", torch.cuda.memory_allocated() / GiB_bytes
                )
                logger.debug(
                    "Reserved: %.2f GiB", torch.cuda.memory_reserved() / GiB_bytes
                )
302
303
304
            raise RuntimeError(
                "Failed to initialize q/k/v range constants. "
                "This may be caused by insufficient memory to allocate "
305
306
                "kv cache."
            ) from e
307

308
309
        # for attn backends supporting query quantization
        self.query_quant = None
310
311
312
313
314
        if (
            self.kv_cache_dtype.startswith("fp8")
            and self.attn_backend.supports_quant_query_input
        ):
            self.query_quant = QuantFP8(static=True, group_shape=GroupShape.PER_TENSOR)
315

316
317
318
319
320
    def forward(
        self,
        query: torch.Tensor,
        key: torch.Tensor,
        value: torch.Tensor,
321
322
323
324
        # For some alternate attention backends like MLA the attention output
        # shape does not match the query shape, so we optionally let the model
        # definition specify the output tensor shape.
        output_shape: Optional[torch.Size] = None,
325
    ) -> torch.Tensor:
326
327
328
329
330
331
332
333
334
        """
        The KV cache is stored inside this class and is accessed via
        `self.kv_cache`.

        Attention metadata (`attn_metadata`) is set using a context manager in
        the model runner's `execute_model` method. It is accessed via forward
        context using
        `vllm.forward_context.get_forward_context().attn_metadata`.
        """
Chen Zhang's avatar
Chen Zhang committed
335
        if self.calculate_kv_scales:
336
            torch.ops.vllm.maybe_calc_kv_scales(query, key, value, self.layer_name)
337
338
339
340
341
342
343
344
345
346
347

        output_dtype = query.dtype
        if self.query_quant is not None:
            # quantizing with a simple torch operation enables
            # torch.compile to fuse this into previous ops
            # which reduces overheads during decoding.
            # Otherwise queries are quantized using custom ops
            # which causes decoding overheads
            assert self.kv_cache_dtype in {"fp8", "fp8_e4m3"}
            query, _ = self.query_quant(query, self._q_scale)

348
        if self.use_output:
349
350
            output_shape = output_shape if output_shape is not None else query.shape
            output = torch.zeros(output_shape, dtype=output_dtype, device=query.device)
351
352
353
354
355
356
357
358
359
360
361
362
363
364
            hidden_size = output_shape[-1]
            # We skip reshaping query, key and value tensors for the MLA
            # backend since these tensors have different semantics and are
            # processed differently.
            if not self.use_mla:
                # Reshape the query, key, and value tensors.
                # NOTE(woosuk): We do this outside the custom op to minimize the
                # CPU overheads from the non-CUDA-graph regions.
                query = query.view(-1, self.num_heads, self.head_size)
                output = output.view(-1, self.num_heads, self.head_size)
                if key is not None:
                    key = key.view(-1, self.num_kv_heads, self.head_size)
                if value is not None:
                    value = value.view(-1, self.num_kv_heads, self.head_size)
365
            if self.use_direct_call:
Chen Zhang's avatar
Chen Zhang committed
366
                forward_context: ForwardContext = get_forward_context()
367
                attn_metadata = forward_context.attn_metadata
368
369
                if isinstance(attn_metadata, dict):
                    attn_metadata = attn_metadata[self.layer_name]
Chen Zhang's avatar
Chen Zhang committed
370
                self_kv_cache = self.kv_cache[forward_context.virtual_engine]
371
372
373
                self.impl.forward(
                    self, query, key, value, self_kv_cache, attn_metadata, output=output
                )
374
375
            else:
                torch.ops.vllm.unified_attention_with_output(
376
377
                    query, key, value, output, self.layer_name
                )
378
            return output.view(-1, hidden_size)
379
        else:
380
            if self.use_direct_call:
Chen Zhang's avatar
Chen Zhang committed
381
                forward_context = get_forward_context()
382
                attn_metadata = forward_context.attn_metadata
383
384
                if isinstance(attn_metadata, dict):
                    attn_metadata = attn_metadata[self.layer_name]
Chen Zhang's avatar
Chen Zhang committed
385
                self_kv_cache = self.kv_cache[forward_context.virtual_engine]
386
387
388
                return self.impl.forward(
                    self, query, key, value, self_kv_cache, attn_metadata
                )
389
390
            else:
                return torch.ops.vllm.unified_attention(
391
392
                    query, key, value, self.layer_name
                )
393

394
395
    def calc_kv_scales(self, query, key, value):
        self._q_scale.copy_(torch.abs(query).max() / self.q_range)
396
397
        self._k_scale.copy_(torch.abs(key).max() / self.k_range)
        self._v_scale.copy_(torch.abs(value).max() / self.v_range)
398
        self._q_scale_float = self._q_scale.item()
399
400
401
402
403
        self._k_scale_float = self._k_scale.item()
        self._v_scale_float = self._v_scale.item()
        # We only calculate the scales once
        self.calculate_kv_scales = False

404
405
406
407
408
    def extra_repr(self) -> str:
        s = f"head_size={self.impl.head_size}"  # type: ignore
        s += f", num_heads={self.impl.num_heads}"  # type: ignore
        s += f", num_kv_heads={self.impl.num_kv_heads}"  # type: ignore
        s += f", scale={self.impl.scale}"  # type: ignore
409
        s += f", backend={self.impl.__class__.__name__}"
410
        return s
411

412
    def process_weights_after_loading(self, act_dtype: torch.dtype):
413
        if hasattr(self.impl, "process_weights_after_loading"):
414
            self.impl.process_weights_after_loading(act_dtype)
415

416
        # FlashInfer requires attention sinks to be float32
417
        if self.backend == _Backend.FLASHINFER and hasattr(self.impl, "sinks"):
418
            from vllm.v1.attention.backends.flashinfer import FlashInferImpl
419

420
            assert isinstance(self.impl, FlashInferImpl)
421
            if self.impl.sinks is not None and self.impl.sinks.dtype != torch.float32:
422
423
                self.impl.sinks = self.impl.sinks.to(torch.float32)

424
425
426
    def get_attn_backend(self) -> type[AttentionBackend]:
        return self.attn_backend

427

428
429
430
431
432
433
434
435
436
class MultiHeadAttention(nn.Module):
    """Multi-headed attention without any cache, used for ViT."""

    def __init__(
        self,
        num_heads: int,
        head_size: int,
        scale: float,
        num_kv_heads: Optional[int] = None,
437
438
439
440
        # This has no effect, it is only here to make it easier to swap
        # between Attention and MultiHeadAttention
        prefix: str = "",
    ) -> None:
441
442
443
444
445
        super().__init__()
        self.num_heads = num_heads
        self.head_size = head_size
        self.scale = scale
        self.num_kv_heads = num_heads if num_kv_heads is None else num_kv_heads
446
        self.layer_name = prefix
447

448
449
        assert self.num_heads % self.num_kv_heads == 0, (
            f"num_heads ({self.num_heads}) is not "
450
            f"divisible by num_kv_heads ({self.num_kv_heads})"
451
        )
452
453
        self.num_queries_per_kv = self.num_heads // self.num_kv_heads

454
455
        # During model initialization, the default dtype is set as the model
        # weight and activation dtype.
456
        dtype = torch.get_default_dtype()
457
458
459
460
461
462
463
464
465

        # Determine the attention backend
        backend = get_vit_attn_backend(head_size=head_size, dtype=dtype)

        # Some auto-selected backends can be upgraded
        # to upstream flash attention if available.
        # If vllm native fa is selected, we use it directly.
        use_upstream_fa = False

466
467
        if current_platform.is_xpu():
            # currently, only torch_sdpa is supported on xpu
468
469
            self.attn_backend = _Backend.TORCH_SDPA
        else:
470
471
472
473
474
475
476
477
478
479
480
481
            self.attn_backend = (
                backend
                if backend
                in {
                    _Backend.TORCH_SDPA,
                    _Backend.XFORMERS,
                    _Backend.PALLAS,
                    _Backend.ROCM_AITER_FA,
                    _Backend.FLASH_ATTN,
                }
                else _Backend.TORCH_SDPA
            )
482

483
484
        self.attn_backend, self._flash_attn_varlen_func = (
            maybe_get_vit_flash_attn_backend(
485
486
487
                self.attn_backend,
                use_upstream_fa,
            )
488
        )
489

490
        if self.attn_backend == _Backend.XFORMERS and not check_xformers_availability():
491
492
            self.attn_backend = _Backend.TORCH_SDPA

493
        self.is_flash_attn_backend = self.attn_backend in {
494
495
            _Backend.FLASH_ATTN,
            _Backend.ROCM_AITER_FA,
496
497
498
499
        }

        # this condition is just to make sure that the
        # use_upstream_fa in the log is correct
500
        if current_platform.is_rocm() and self.attn_backend == _Backend.FLASH_ATTN:
501
            use_upstream_fa = True
502
503
504

        logger.info_once(
            f"MultiHeadAttention attn_backend: {self.attn_backend}, "
505
506
            f"use_upstream_fa: {use_upstream_fa}"
        )
507

508
509
510
511
512
513
    def forward(
        self,
        query: torch.Tensor,
        key: torch.Tensor,
        value: torch.Tensor,
    ) -> torch.Tensor:
514
        """Input shape:
515
516
517
518
        (batch_size x seq_len x hidden_size) or
        (batch_size x seq_len x num_heads x head_size)
        """
        bsz, q_len = query.size()[:2]
519
520
521
522
523
524
        kv_len = key.size(1)

        query = query.view(bsz, q_len, self.num_heads, self.head_size)
        key = key.view(bsz, kv_len, self.num_kv_heads, self.head_size)
        value = value.view(bsz, kv_len, self.num_kv_heads, self.head_size)

525
526
527
528
529
        if (num_repeat := self.num_queries_per_kv) > 1:
            # Handle MQA and GQA
            key = torch.repeat_interleave(key, num_repeat, dim=2)
            value = torch.repeat_interleave(value, num_repeat, dim=2)

530
        if self.is_flash_attn_backend:
531
532
533
534
535
536
            cu_seqlens_q = torch.arange(
                0, (bsz + 1) * q_len, step=q_len, dtype=torch.int32, device=query.device
            )
            cu_seqlens_k = torch.arange(
                0, (bsz + 1) * kv_len, step=kv_len, dtype=torch.int32, device=key.device
            )
537
538
539
540
541
542
543
544
545
546
547
548

            out = self._flash_attn_varlen_func(
                query.flatten(0, 1),
                key.flatten(0, 1),
                value.flatten(0, 1),
                cu_seqlens_q=cu_seqlens_q,
                cu_seqlens_k=cu_seqlens_k,
                max_seqlen_q=q_len,
                max_seqlen_k=kv_len,
                softmax_scale=self.scale,
            )
        elif self.attn_backend == _Backend.XFORMERS:
549
550
            from xformers import ops as xops

551
552
553
            out = xops.memory_efficient_attention_forward(
                query, key, value, scale=self.scale
            )
554
        elif self.attn_backend == _Backend.TORCH_SDPA:
555
556
            query, key, value = (x.transpose(1, 2) for x in (query, key, value))
            out = F.scaled_dot_product_attention(query, key, value, scale=self.scale)
557
            out = out.transpose(1, 2)
558
        elif self.attn_backend == _Backend.PALLAS:
559
            query, key, value = (x.transpose(1, 2) for x in (query, key, value))
560
            from torch_xla.experimental.custom_kernel import flash_attention
561

562
563
            out = flash_attention(query, key, value, sm_scale=self.scale)
            out = out.transpose(1, 2)
564
565
566
        else:
            # ViT attention hasn't supported this backend yet
            raise NotImplementedError(
567
568
                f"ViT attention hasn't supported {self.attn_backend} backend yet."
            )
569

570
        return out.reshape(bsz, q_len, -1)
571
572


573
574
575
576
577
578
579
580
581
582
def wait_for_kv_layer_from_connector(layer_name: str):
    if not has_kv_transfer_group() or not is_v1_kv_transfer_group():
        return

    connector = get_kv_transfer_group()

    forward_context: ForwardContext = get_forward_context()
    attn_metadata = forward_context.attn_metadata
    if attn_metadata is None:
        return
583
    assert isinstance(attn_metadata, dict)
584
585
586
587
588
    connector.wait_for_layer_load(layer_name)


def maybe_save_kv_layer_to_connector(
    layer_name: str,
589
    kv_cache_layer: list[torch.Tensor],
590
591
592
593
594
595
596
597
598
599
):
    if not has_kv_transfer_group() or not is_v1_kv_transfer_group():
        return

    connector = get_kv_transfer_group()

    forward_context: ForwardContext = get_forward_context()
    attn_metadata = forward_context.attn_metadata
    if attn_metadata is None:
        return
600
    assert isinstance(attn_metadata, dict)
601
    connector.save_kv_layer(layer_name, kv_cache_layer, attn_metadata[layer_name])
602
603


604
605
606
607
608
609
610
611
612
613
614
615
616
def maybe_calc_kv_scales(
    query: torch.Tensor,
    key: torch.Tensor,
    value: torch.Tensor,
    layer_name: str,
) -> None:
    forward_context: ForwardContext = get_forward_context()
    attn_metadata = forward_context.attn_metadata

    if isinstance(attn_metadata, dict):
        attn_metadata = attn_metadata[layer_name]

    if attn_metadata is None or not getattr(
617
618
        attn_metadata, "enable_kv_scales_calculation", False
    ):
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
        return

    self = forward_context.no_compile_layers[layer_name]
    self.calc_kv_scales(query, key, value)


def maybe_calc_kv_scales_fake(
    query: torch.Tensor,
    key: torch.Tensor,
    value: torch.Tensor,
    layer_name: str,
) -> None:
    return


direct_register_custom_op(
    op_name="maybe_calc_kv_scales",
    op_func=maybe_calc_kv_scales,
    mutates_args=["query", "key", "value"],
    fake_impl=maybe_calc_kv_scales_fake,
)


642
643
644
645
646
647
def unified_attention(
    query: torch.Tensor,
    key: torch.Tensor,
    value: torch.Tensor,
    layer_name: str,
) -> torch.Tensor:
648
649
    wait_for_kv_layer_from_connector(layer_name)

650
    forward_context: ForwardContext = get_forward_context()
651
    attn_metadata = forward_context.attn_metadata
652
653
    if isinstance(attn_metadata, dict):
        attn_metadata = attn_metadata[layer_name]
654
    self = forward_context.no_compile_layers[layer_name]
655
    kv_cache = self.kv_cache[forward_context.virtual_engine]
656
    output = self.impl.forward(self, query, key, value, kv_cache, attn_metadata)
657
658
659

    maybe_save_kv_layer_to_connector(layer_name, kv_cache)
    return output
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674


def unified_attention_fake(
    query: torch.Tensor,
    key: torch.Tensor,
    value: torch.Tensor,
    layer_name: str,
) -> torch.Tensor:
    return torch.empty_like(query).contiguous()


direct_register_custom_op(
    op_name="unified_attention",
    op_func=unified_attention,
    fake_impl=unified_attention_fake,
675
    tags=tag_cudagraph_unsafe,
676
)
677
678
679
680
681
682
683
684


def unified_attention_with_output(
    query: torch.Tensor,
    key: torch.Tensor,
    value: torch.Tensor,
    output: torch.Tensor,
    layer_name: str,
685
    output_scale: Optional[torch.Tensor] = None,
686
    output_block_scale: Optional[torch.Tensor] = None,
687
) -> None:
688
    wait_for_kv_layer_from_connector(layer_name)
689
    forward_context: ForwardContext = get_forward_context()
690
    attn_metadata = forward_context.attn_metadata
691
692
    if isinstance(attn_metadata, dict):
        attn_metadata = attn_metadata[layer_name]
693
    self = forward_context.no_compile_layers[layer_name]
694
    kv_cache = self.kv_cache[forward_context.virtual_engine]
695
696
697
698
699
700
701
702
703
704
705
    self.impl.forward(
        self,
        query,
        key,
        value,
        kv_cache,
        attn_metadata,
        output=output,
        output_scale=output_scale,
        output_block_scale=output_block_scale,
    )
706

707
708
    maybe_save_kv_layer_to_connector(layer_name, kv_cache)

709
710
711
712
713
714
715

def unified_attention_with_output_fake(
    query: torch.Tensor,
    key: torch.Tensor,
    value: torch.Tensor,
    output: torch.Tensor,
    layer_name: str,
716
    output_scale: Optional[torch.Tensor] = None,
717
    output_block_scale: Optional[torch.Tensor] = None,
718
719
720
721
722
723
724
) -> None:
    return


direct_register_custom_op(
    op_name="unified_attention_with_output",
    op_func=unified_attention_with_output,
725
    mutates_args=["output", "output_block_scale"],
726
    fake_impl=unified_attention_with_output_fake,
727
    tags=tag_cudagraph_unsafe,
728
)