* stash for now * initial commit * small updated * up * up * works! * nits and fixes * don't loop too much * finish working example * update * fix the small freeblocks issue * feat: stream inputs to continuous batch * fix: update attn from `eager` to `sdpa` * refactor: fmt * refactor: cleanup unnecessary code * feat: add `update` fn to `PagedAttentionCache` * feat: broken optimal block size computation * fix: debugging invalid cache logic * fix: attention mask * refactor: use custom prompts for example * feat: add streaming output * fix: prefill split refactor: add doc strings and unsound/redundant logic fix: compute optimal blocks logic * fix: send decoded tokens when `prefilling_split` -> `decoding` * refactor: move logic to appropriate parent class * fix: remove truncation as we split prefilling anyways refactor: early return when we have enough selected requests * feat: add paged attention forward * push Ggraoh> * add paged sdpa * update * btter mps defaults * feat: add progress bar for `generate_batch` * feat: add opentelemetry metrics (ttft + batch fill %age) * feat: add tracing * Add cuda graphs (#38059) * draft cudagraphs addition * nits * styling * update * fix * kinda draft of what it should look like * fixes * lol * not sure why inf everywhere * can generate but output is shit * some fixes * we should have a single device synch * broken outputs but it does run * refactor * updates * updates with some fixes * fix mask causality * another commit that casts after * add error * simplify example * update * updates * revert llama changes * fix merge conflicts * fix: tracing and metrics * my updates * update script default values * fix block allocation issue * fix prefill split attnetion mask * no bugs * add paged eager * fix * update * style * feat: add pytorch traces * fix * fix * refactor: remove pytorch profiler data * style * nits * cleanup * draft test file * fix * fix * fix paged and graphs * small renamings * cleanups and push * refactor: move tracing and metrics logic to utils * refactor: trace more blocks of code * nits * nits * update * to profile or not to profile * refactor: create new output object * causal by default * cleanup but generations are still off for IDK what reason * simplifications but not running still * this does work. * small quality of life updates * nits * updaet * fix the scheduler * fix warning * ol * fully fixed * nits * different generation parameters * nice * just style * feat: add cache memory usage * feat: add kv cache free memory * feat: add active/waiting count & req latency * do the sampling * fix: synchronize CUDA only if available and improve error handling in ContinuousBatchingManager * fix on mps * feat: add dashboard & histogram buckets * perf: improve waiting reqs data structures * attempt to compile, but we should only do it on mps AFAIK * feat: decouple scheduling logic * just a draft * c;eanup and fixup * optional * style * update * update * remove the draft documentation * fix import as well * update * fix the test * style doomed --------- Co-authored-by: Luc Georges <luc.sydney.georges@gmail.com>
444 lines
19 KiB
Python
444 lines
19 KiB
Python
# Copyright 2024 The Fairseq Authors and the HuggingFace Inc. team. All rights reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import inspect
|
|
import os
|
|
from typing import Optional, TypedDict
|
|
|
|
import torch
|
|
import torch.nn.functional as F
|
|
|
|
from .utils import (
|
|
is_flash_attn_2_available,
|
|
is_flash_attn_greater_or_equal,
|
|
is_flash_attn_greater_or_equal_2_10,
|
|
is_torch_npu_available,
|
|
logging,
|
|
)
|
|
|
|
|
|
logger = logging.get_logger(__name__)
|
|
flash_attn_func = None
|
|
|
|
|
|
if is_flash_attn_2_available():
|
|
from flash_attn.bert_padding import index_first_axis, pad_input, unpad_input # noqa
|
|
from flash_attn import flash_attn_func, flash_attn_varlen_func
|
|
from flash_attn.layers.rotary import apply_rotary_emb # noqa
|
|
|
|
|
|
# patch functions in package `flash-attn` when using flash-attention on Ascend NPU.
|
|
if is_torch_npu_available():
|
|
from torch_npu import npu_rotary_mul as apply_rotary_emb # noqa
|
|
|
|
from .integrations.npu_flash_attention import index_first_axis, pad_input, unpad_input
|
|
from .integrations.npu_flash_attention import npu_flash_attn_func as flash_attn_func
|
|
from .integrations.npu_flash_attention import npu_flash_attn_varlen_func as flash_attn_varlen_func
|
|
|
|
|
|
if flash_attn_func:
|
|
_flash_supports_window_size = "window_size" in list(inspect.signature(flash_attn_func).parameters)
|
|
|
|
|
|
def is_flash_attn_available():
|
|
"""Determine whether flash-attention can be used or not."""
|
|
|
|
# if package `flash-attn` is available, flash-attention can be used natively.
|
|
if is_flash_attn_2_available():
|
|
return True
|
|
|
|
# flash-attention can be used on Ascend NPU without package `flash-attn`
|
|
if is_torch_npu_available():
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def flash_attn_supports_top_left_mask():
|
|
"""Determine whether flash-attention uses top-left or down-right mask"""
|
|
|
|
if is_flash_attn_2_available():
|
|
# top-left mask is used in package `flash-attn` with version lower than 2.1.0
|
|
return not is_flash_attn_greater_or_equal_2_10()
|
|
|
|
if is_torch_npu_available():
|
|
# down-right mask is used on Ascend NPU by default, set env `NPU_FA2_SPARSE_MODE=2` to activate top-left mask.
|
|
from .integrations.npu_flash_attention import is_npu_fa2_top_left_aligned_causal_mask
|
|
|
|
return is_npu_fa2_top_left_aligned_causal_mask()
|
|
|
|
return False
|
|
|
|
|
|
def _get_unpad_data(attention_mask: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor, int]:
|
|
"""
|
|
Retrieves indexing data required to repad unpadded (ragged) tensors.
|
|
|
|
Arguments:
|
|
attention_mask (`torch.Tensor`):
|
|
Boolean or int tensor of shape (batch_size, sequence_length), 1 means valid and 0 means not valid.
|
|
|
|
Return:
|
|
indices (`torch.Tensor`):
|
|
The indices of non-masked tokens from the flattened input sequence.
|
|
cu_seqlens (`torch.Tensor`):
|
|
The cumulative sequence lengths, used to index into ragged (unpadded) tensors. `cu_seqlens` shape is (batch_size + 1,).
|
|
max_seqlen_in_batch (`int`):
|
|
Maximum sequence length in batch.
|
|
"""
|
|
seqlens_in_batch = attention_mask.sum(dim=-1, dtype=torch.int32)
|
|
indices = torch.nonzero(attention_mask.flatten(), as_tuple=False).flatten()
|
|
max_seqlen_in_batch = seqlens_in_batch.max().item()
|
|
cu_seqlens = F.pad(torch.cumsum(seqlens_in_batch, dim=0, dtype=torch.int32), (1, 0))
|
|
return (
|
|
indices,
|
|
cu_seqlens,
|
|
max_seqlen_in_batch,
|
|
)
|
|
|
|
|
|
def _upad_input(
|
|
query_layer: torch.Tensor,
|
|
key_layer: torch.Tensor,
|
|
value_layer: torch.Tensor,
|
|
attention_mask: torch.Tensor,
|
|
query_length: int,
|
|
):
|
|
"""
|
|
Unpads query, key, and values tensors, using a single dimension for all tokens even though they belong to different batches.
|
|
|
|
This function is used instead of `flash_attn.bert_padding.unpad_input` in order to avoid the recomputation of the same intermediary
|
|
tensors for query, key, value tensors.
|
|
|
|
Arguments:
|
|
query_layer (`torch.Tensor`):
|
|
Query state with padding. Shape: (batch_size, query_length, num_heads, head_dim).
|
|
key_layer (`torch.Tensor`):
|
|
Key state with padding. Shape: (batch_size, kv_seq_len, num_key_value_heads, head_dim).
|
|
value_layer (`torch.Tensor`):
|
|
Value state with padding. Shape: (batch_size, kv_seq_len, num_key_value_heads, head_dim).
|
|
attention_mask (`torch.Tensor`):
|
|
Boolean or int tensor of shape (batch_size, sequence_length), 1 means valid and 0 means not valid.
|
|
query_length (`int`):
|
|
Target length.
|
|
|
|
Return:
|
|
query_layer (`torch.Tensor`):
|
|
Query state without padding. Shape: (total_target_length, num_heads, head_dim).
|
|
key_layer (`torch.Tensor`):
|
|
Key state with padding. Shape: (total_source_length, num_key_value_heads, head_dim).
|
|
value_layer (`torch.Tensor`):
|
|
Value state with padding. Shape: (total_source_length, num_key_value_heads, head_dim).
|
|
indices_q (`torch.Tensor`):
|
|
The indices of non-masked tokens from the flattened input target sequence.
|
|
(cu_seqlens_q, cu_seqlens_k) (`Tuple[int]`):
|
|
The cumulative sequence lengths for the target (query) and source (key, value), used to index into ragged (unpadded) tensors. `cu_seqlens` shape is (batch_size + 1,).
|
|
(max_seqlen_in_batch_q, max_seqlen_in_batch_k) (`Tuple[int]`):
|
|
Maximum sequence length in batch (`max_seqlen_in_batch_q` for the target sequence i.e. query, `max_seqlen_in_batch_k` for the source sequence i.e. key/value).
|
|
"""
|
|
indices_k, cu_seqlens_k, max_seqlen_in_batch_k = _get_unpad_data(attention_mask)
|
|
|
|
# With static caches, the k/v states may be larger than the mask -> we need to slice them to avoid generating garbage
|
|
# It's a bit of an anti-pattern, but otherwise we silently compute wrong attentions scores
|
|
if key_layer.shape[1] > (seq_len := attention_mask.shape[-1]):
|
|
key_layer, value_layer = key_layer[:, :seq_len, :, :], value_layer[:, :seq_len, :, :]
|
|
|
|
batch_size, kv_seq_len, num_key_value_heads, head_dim = key_layer.shape
|
|
|
|
key_layer = index_first_axis(key_layer.reshape(batch_size * kv_seq_len, num_key_value_heads, head_dim), indices_k)
|
|
value_layer = index_first_axis(
|
|
value_layer.reshape(batch_size * kv_seq_len, num_key_value_heads, head_dim), indices_k
|
|
)
|
|
if query_length == kv_seq_len:
|
|
query_layer = index_first_axis(query_layer.reshape(batch_size * kv_seq_len, -1, head_dim), indices_k)
|
|
cu_seqlens_q = cu_seqlens_k
|
|
max_seqlen_in_batch_q = max_seqlen_in_batch_k
|
|
indices_q = indices_k
|
|
elif query_length == 1:
|
|
max_seqlen_in_batch_q = 1
|
|
cu_seqlens_q = torch.arange(
|
|
batch_size + 1, dtype=torch.int32, device=query_layer.device
|
|
) # There is a memcpy here, that is very bad.
|
|
indices_q = cu_seqlens_q[:-1]
|
|
query_layer = query_layer.squeeze(1)
|
|
else:
|
|
# The -q_len: slice assumes left padding.
|
|
attention_mask = attention_mask[:, -query_length:]
|
|
query_layer, indices_q, cu_seqlens_q, max_seqlen_in_batch_q, *_ = unpad_input(query_layer, attention_mask)
|
|
|
|
return (
|
|
query_layer,
|
|
key_layer,
|
|
value_layer,
|
|
indices_q,
|
|
(cu_seqlens_q, cu_seqlens_k),
|
|
(max_seqlen_in_batch_q, max_seqlen_in_batch_k),
|
|
)
|
|
|
|
|
|
def prepare_fa2_from_position_ids(query, key, value, position_ids):
|
|
"""
|
|
This function returns necessary arguments to call `flash_attn_varlen_func`.
|
|
All three query, key, value states will be flattened.
|
|
Cumulative lengths of each examples in the batch will be extracted from position_ids.
|
|
|
|
NOTE: ideally cumulative lengths should be prepared at the data collator stage
|
|
|
|
Arguments:
|
|
query (`torch.Tensor`):
|
|
Query state with padding. Shape: (batch_size, query_length, num_heads, head_dim).
|
|
key (`torch.Tensor`):
|
|
Key state with padding. Shape: (batch_size, kv_seq_len, num_key_value_heads, head_dim).
|
|
value (`torch.Tensor`):
|
|
Value state with padding. Shape: (batch_size, kv_seq_len, num_key_value_heads, head_dim).
|
|
position_ids (`torch.Tensor`):
|
|
Boolean or int tensor of shape (batch_size, sequence_length), 1 means valid and 0 means not valid.
|
|
|
|
Return:
|
|
query (`torch.Tensor`):
|
|
Query state without padding. Shape: (total_target_length, num_heads, head_dim).
|
|
key (`torch.Tensor`):
|
|
Key state with padding. Shape: (total_source_length, num_key_value_heads, head_dim).
|
|
value (`torch.Tensor`):
|
|
Value state with padding. Shape: (total_source_length, num_key_value_heads, head_dim).
|
|
indices_q (`torch.Tensor`):
|
|
The indices of non-masked tokens from the flattened input target sequence.
|
|
(cu_seqlens_q, cu_seqlens_k) (`Tuple[int]`):
|
|
The cumulative sequence lengths for the target (query) and source (key, value), used to index into ragged (unpadded) tensors. `cu_seqlens` shape is (batch_size + 1,).
|
|
(max_seqlen_in_batch_q, max_seqlen_in_batch_k) (`Tuple[int]`):
|
|
Maximum sequence length in batch (`max_seqlen_in_batch_q` for the target sequence i.e. query, `max_seqlen_in_batch_k` for the source sequence i.e. key/value).
|
|
"""
|
|
query = query.view(-1, query.size(-2), query.size(-1))
|
|
key = key.contiguous().view(-1, key.size(-2), key.size(-1))
|
|
value = value.contiguous().view(-1, value.size(-2), value.size(-1))
|
|
position_ids = position_ids.flatten()
|
|
indices_q = torch.arange(position_ids.size(0), device=position_ids.device, dtype=torch.int32)
|
|
|
|
cu_seq_lens = torch.cat(
|
|
(
|
|
indices_q[position_ids == 0],
|
|
torch.tensor(position_ids.size(), device=position_ids.device, dtype=torch.int32),
|
|
)
|
|
)
|
|
|
|
max_length = position_ids.max() + 1
|
|
|
|
return (query, key, value, indices_q, (cu_seq_lens, cu_seq_lens), (max_length, max_length))
|
|
|
|
|
|
def fa_peft_integration_check(
|
|
query: torch.Tensor,
|
|
key: torch.Tensor,
|
|
value: torch.Tensor,
|
|
target_dtype: Optional[torch.dtype] = None,
|
|
):
|
|
"""
|
|
PEFT usually casts the layer norms in float32 for training stability reasons
|
|
therefore the input hidden states gets silently casted in float32. Hence, we need
|
|
cast them back in float16 / bfloat16 just to be sure everything works as expected.
|
|
This might slowdown training & inference so it is recommended to not cast the LayerNorms!
|
|
|
|
Args:
|
|
query (`torch.Tensor`):
|
|
Input query states to be passed to Flash Attention API
|
|
key (`torch.Tensor`):
|
|
Input key states to be passed to Flash Attention API
|
|
value (`torch.Tensor`):
|
|
Input value states to be passed to Flash Attention API
|
|
target_dtype (`torch.dtype`, *optional*):
|
|
The dtype to convert the attention tensors to. Conversion can be ignored by
|
|
not providing the target dtype.
|
|
"""
|
|
if target_dtype is None:
|
|
return query, key, value
|
|
|
|
input_dtype = query.dtype
|
|
if input_dtype == torch.float32:
|
|
logger.warning_once(
|
|
f"The input hidden states seems to be silently casted in float32, this might be related to"
|
|
f" the fact you have upcasted embedding or layer norm layers in float32. We will cast back the input in"
|
|
f" {target_dtype}."
|
|
)
|
|
|
|
query = query.to(target_dtype)
|
|
key = key.to(target_dtype)
|
|
value = value.to(target_dtype)
|
|
|
|
return query, key, value
|
|
|
|
|
|
flash_241 = is_flash_attn_greater_or_equal("2.4.1")
|
|
deterministic_g = os.environ.get("FLASH_ATTENTION_DETERMINISTIC", "0") == "1"
|
|
|
|
|
|
def _flash_attention_forward(
|
|
query_states: torch.Tensor,
|
|
key_states: torch.Tensor,
|
|
value_states: torch.Tensor,
|
|
attention_mask: Optional[torch.Tensor],
|
|
query_length: int,
|
|
is_causal: bool,
|
|
dropout: float = 0.0,
|
|
position_ids: Optional[torch.Tensor] = None,
|
|
softmax_scale: Optional[float] = None,
|
|
sliding_window: Optional[int] = None,
|
|
use_top_left_mask: bool = False,
|
|
softcap: Optional[float] = None,
|
|
deterministic: Optional[bool] = None,
|
|
cu_seq_lens_q: Optional[torch.LongTensor] = None,
|
|
cu_seq_lens_k: Optional[torch.LongTensor] = None,
|
|
max_length_q: Optional[int] = None,
|
|
max_length_k: Optional[int] = None,
|
|
target_dtype: Optional[torch.dtype] = None,
|
|
**kwargs,
|
|
):
|
|
"""
|
|
Calls the forward method of Flash Attention - if the input hidden states contain at least one padding token
|
|
first unpad the input, then computes the attention scores and pad the final attention scores.
|
|
|
|
Args:
|
|
query_states (`torch.Tensor`):
|
|
Input query states to be passed to Flash Attention API
|
|
key_states (`torch.Tensor`):
|
|
Input key states to be passed to Flash Attention API
|
|
value_states (`torch.Tensor`):
|
|
Input value states to be passed to Flash Attention API
|
|
attention_mask (`torch.Tensor`, *optional*):
|
|
The padding mask - corresponds to a tensor of size `(batch_size, seq_len)` where 0 stands for the
|
|
position of padding tokens and 1 for the position of non-padding tokens.
|
|
dropout (`float`):
|
|
Attention dropout
|
|
softmax_scale (`float`, *optional*):
|
|
The scaling of QK^T before applying softmax. Default to 1 / sqrt(head_dim)
|
|
use_top_left_mask (`bool`, defaults to `False`):
|
|
flash_attn<2.1 generates top-left aligned causal mask, while what is needed here is bottom-right alignment, that was made default for flash_attn>=2.1. This attribute is used to handle this difference.
|
|
softcap (`float`, *optional*):
|
|
Softcap for the attention logits, used e.g. in gemma2.
|
|
deterministic (`bool`, *optional*):
|
|
Determines if the deterministic option introduced in flash_attn>=2.4.1 is enabled.
|
|
"""
|
|
if not use_top_left_mask:
|
|
causal = is_causal
|
|
else:
|
|
# TODO: Remove the `query_length != 1` check once Flash Attention for RoCm is bumped to 2.1.
|
|
causal = is_causal and query_length != 1
|
|
|
|
# Assuming 4D tensors, key_states.shape[1] is the key/value sequence length (source length).
|
|
use_sliding_windows = (
|
|
_flash_supports_window_size and sliding_window is not None and key_states.shape[1] > sliding_window
|
|
)
|
|
flash_kwargs = {"window_size": (sliding_window, sliding_window)} if use_sliding_windows else {}
|
|
|
|
if flash_241:
|
|
if deterministic is None:
|
|
deterministic = deterministic_g
|
|
flash_kwargs["deterministic"] = deterministic
|
|
|
|
if softcap is not None:
|
|
flash_kwargs["softcap"] = softcap
|
|
|
|
# PEFT possibly silently casts tensors to fp32, this potentially reconverts to correct dtype or is a no op
|
|
query_states, key_states, value_states = fa_peft_integration_check(
|
|
query_states, key_states, value_states, target_dtype
|
|
)
|
|
|
|
# Contains at least one padding token in the sequence
|
|
if attention_mask is not None:
|
|
batch_size = query_states.shape[0]
|
|
query_states, key_states, value_states, indices_q, cu_seq_lens, max_seq_lens = _upad_input(
|
|
query_states, key_states, value_states, attention_mask, query_length
|
|
)
|
|
cu_seqlens_q, cu_seqlens_k = cu_seq_lens
|
|
max_seqlen_in_batch_q, max_seqlen_in_batch_k = max_seq_lens
|
|
|
|
attn_output_unpad = flash_attn_varlen_func(
|
|
query_states,
|
|
key_states,
|
|
value_states,
|
|
cu_seqlens_q=cu_seqlens_q,
|
|
cu_seqlens_k=cu_seqlens_k,
|
|
max_seqlen_q=max_seqlen_in_batch_q,
|
|
max_seqlen_k=max_seqlen_in_batch_k,
|
|
dropout_p=dropout,
|
|
softmax_scale=softmax_scale,
|
|
causal=causal,
|
|
**flash_kwargs,
|
|
)
|
|
attn_output = pad_input(attn_output_unpad, indices_q, batch_size, query_length)
|
|
|
|
# If position_ids is provided and check all examples do not contain only 1 sequence, If tensor in increasing
|
|
# then we probably have one sequence, otherwise it is packed. Additionally check we are in pre-fill/training stage.
|
|
# Use `flash_attn_varlen_func` to prevent cross-example attention and also allow padding free approach
|
|
elif position_ids is not None and (
|
|
max_length_q is not None or (query_length != 1 and not (torch.diff(position_ids, dim=-1) >= 0).all())
|
|
):
|
|
batch_size = query_states.size(0)
|
|
|
|
if cu_seq_lens_q is None or cu_seq_lens_k is None:
|
|
query_states, key_states, value_states, indices_q, cu_seq_lens, max_seq_lens = (
|
|
prepare_fa2_from_position_ids(query_states, key_states, value_states, position_ids)
|
|
)
|
|
|
|
cu_seq_lens_q, cu_seq_lens_k = cu_seq_lens
|
|
max_length_q, max_length_k = max_seq_lens
|
|
|
|
else:
|
|
query_states = query_states.reshape(-1, query_states.size(-2), query_states.size(-1))
|
|
key_states = key_states.reshape(-1, key_states.size(-2), key_states.size(-1))
|
|
value_states = value_states.reshape(-1, value_states.size(-2), value_states.size(-1))
|
|
|
|
attn_output = flash_attn_varlen_func(
|
|
query_states,
|
|
key_states,
|
|
value_states,
|
|
cu_seqlens_q=cu_seq_lens_q,
|
|
cu_seqlens_k=cu_seq_lens_k,
|
|
max_seqlen_q=max_length_q,
|
|
max_seqlen_k=max_length_k,
|
|
dropout_p=dropout,
|
|
softmax_scale=softmax_scale,
|
|
causal=causal,
|
|
**flash_kwargs,
|
|
)
|
|
|
|
attn_output = attn_output.view(batch_size, -1, attn_output.size(-2), attn_output.size(-1))
|
|
|
|
else:
|
|
attn_output = flash_attn_func(
|
|
query_states, key_states, value_states, dropout, softmax_scale=softmax_scale, causal=causal, **flash_kwargs
|
|
)
|
|
|
|
return attn_output
|
|
|
|
|
|
class FlashAttentionKwargs(TypedDict, total=False):
|
|
"""
|
|
Keyword arguments for Flash Attention with Compile.
|
|
|
|
Attributes:
|
|
cumulative_seqlens_q (`torch.LongTensor`, *optional*)
|
|
Gets cumulative sequence length for query state.
|
|
cumulative_seqlens_k (`torch.LongTensor`, *optional*)
|
|
Gets cumulative sequence length for key state.
|
|
max_length_q (`int`, *optional*):
|
|
Maximum sequence length for query state.
|
|
max_length_k (`int`, *optional*):
|
|
Maximum sequence length for key state.
|
|
"""
|
|
|
|
cumulative_seqlens_q: Optional[torch.LongTensor]
|
|
cumulative_seqlens_k: Optional[torch.LongTensor]
|
|
max_length_q: Optional[int]
|
|
max_length_k: Optional[int]
|