Files
wehub-resource-sync 94057c3d3e
PR Test (NPU) / check-changes (push) Has been cancelled
PR Test (NPU) / pr-gate (push) Has been cancelled
PR Test (NPU) / set-image-config (push) Has been cancelled
PR Test (NPU) / stage-b-test-1-npu-a2 (0) (push) Has been cancelled
PR Test (NPU) / stage-b-test-1-npu-a2 (1) (push) Has been cancelled
PR Test (NPU) / stage-b-test-2-npu-a2 (0) (push) Has been cancelled
PR Test (NPU) / stage-b-test-2-npu-a2 (1) (push) Has been cancelled
PR Test (NPU) / stage-b-test-4-npu-a3 (push) Has been cancelled
PR Test (NPU) / stage-b-test-16-npu-a3 (push) Has been cancelled
PR Test (NPU) / multimodal-gen-test-1-npu-a3 (push) Has been cancelled
PR Test (NPU) / multimodal-gen-test-2-npu-a3 (push) Has been cancelled
PR Test (Arm64) / pr-gate (push) Has been cancelled
PR Test (Arm64) / check-changes (push) Has been cancelled
PR Test (Arm64) / build-test (push) Has been cancelled
PR Test (sgl-router) / gate (push) Has been cancelled
PR Test (sgl-router) / tier-1 — lint (push) Has been cancelled
PR Test (sgl-router) / tier-2 — build + test (push) Has been cancelled
PR Test (sgl-router) / tier-3 — docker (placeholder) (push) Has been cancelled
PR Test (sgl-router) / tier-3 — k8s integration (push) Has been cancelled
PR Test (sgl-router) / tier-3 — e2e (push) Has been cancelled
PR Test (sgl-router) / finish (push) Has been cancelled
PR Test (NPU) / single-node-poc (map[name:qwen3_6_27b_w8a8_1p_in64k_out1k_50ms runner:linux-aarch64-a3-2 test_case:test/registered/ascend/performance/qwen3_6_27b/test_npu_qwen3_6_27b_w8a8_1p_in64k_out1k_50ms.py test_type:perf]) (push) Has been cancelled
PR Test (NPU) / pr-test-npu-finish (push) Has been cancelled
PR Test (Xeon) / pr-gate (push) Has been cancelled
PR Test (Xeon) / check-changes (push) Has been cancelled
PR Test (Xeon) / build-test (, xeon-gnr, base-b-test-cpu) (push) Has been cancelled
PR Test (XPU) / check-changes (push) Has been cancelled
PR Test (XPU) / pr-gate (push) Has been cancelled
PR Test (XPU) / stage-a-test-1-gpu-xpu (push) Has been cancelled
PR Test (XPU) / wait-for-stage-a (push) Has been cancelled
PR Test (XPU) / stage-b-test-1-gpu-xpu (push) Has been cancelled
PR Test (XPU) / finish (push) Has been cancelled
CI Model Inventory / build-inventory (push) Has been cancelled
Lint / lint (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark Compilation Check (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark - Manual Policy (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark - Request Processing (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark Summary (push) Has been cancelled
PR Test (SMG) / build-wheel (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on windows (x86_64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on macos (x86_64 - auto) (push) Has been cancelled
PR Test (SMG) / python-unit-tests (push) Has been cancelled
PR Test (SMG) / unit-tests (push) Has been cancelled
PR Test (SMG) / benchmarks (push) Has been cancelled
PR Test (SMG) / chat-completions (push) Has been cancelled
PR Test (SMG) / chat-completions-4gpu (push) Has been cancelled
PR Test (SMG) / e2e (push) Has been cancelled
PR Test (SMG) / docker-build-test (push) Has been cancelled
PR Test (SMG) / k8s-integration (push) Has been cancelled
PR Test (SMG) / finish (push) Has been cancelled
PR Test (SMG) / summarize-benchmarks (push) Has been cancelled
Release SGLang Model Gateway Docker Image / publish (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on macos (aarch64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (aarch64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (x86_64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (aarch64 - musllinux_1_1) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (x86_64 - musllinux_1_1) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / Build SDist (push) Has been cancelled
Release SGLang Model Gateway to PyPI / Upload to PyPI (push) Has been cancelled
Release SGLang Kernels / build-cu129-matrix (aarch64, 12.9, 3.10, arm-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / build-cu129-matrix (x86_64, 12.9, 3.10, x64-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / release-cu129 (push) Has been cancelled
Release SGLang Kernels / build-cu130-matrix (aarch64, 13.0, 3.10, arm-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / build-cu130-matrix (x86_64, 13.0, 3.10, x64-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / release-cu130 (push) Has been cancelled
Release SGLang Kernels / build-rocm-matrix (3.10, 700) (push) Has been cancelled
Release SGLang Kernels / build-rocm-matrix (3.10, 720) (push) Has been cancelled
Release SGLang Kernels / release-rocm700 (push) Has been cancelled
Release SGLang Kernels / release-rocm720 (push) Has been cancelled
Release SGLang Kernels / build-musa43 (43, 3.10) (push) Has been cancelled
Release SGLang Kernels / release-musa43 (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:38:16 +08:00

692 lines
25 KiB
Python

from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Optional
import numpy as np
import torch
from sglang.kernels.ops.memory.common import (
_get_last_loc_safe_kernel as _get_last_loc_safe_kernel,
)
from sglang.kernels.ops.memory.common import get_last_loc_kernel as get_last_loc_kernel
from sglang.kernels.ops.memory.common import (
get_last_loc_triton,
get_last_loc_triton_safe,
write_req_to_token_pool_triton,
)
from sglang.srt.hardware_backend.npu.dsv4.dsv4_common_hooks import (
maybe_evict_dsv4_state_on_swa,
maybe_write_dsv4_decode,
maybe_write_dsv4_extend,
)
from sglang.srt.mem_cache.allocator.swa import SWATokenToKVPoolAllocator
from sglang.srt.mem_cache.base_prefix_cache import BasePrefixCache, EvictParams
from sglang.srt.mem_cache.memory_pool import HybridReqToTokenPool, ReqToTokenPool
from sglang.srt.runtime_context import get_server_args
from sglang.srt.server_args import ServerArgs
from sglang.srt.utils import is_cuda, is_hip, is_npu, support_triton
from sglang.srt.utils.common import ceil_align, is_pin_memory_available
_is_npu = is_npu()
_is_hip = is_hip()
_is_cuda = is_cuda()
if TYPE_CHECKING:
from sglang.srt.managers.schedule_batch import Req, ScheduleBatch
from sglang.srt.mem_cache.allocator import BaseTokenToKVPoolAllocator
from sglang.srt.model_executor.forward_batch_info import DSV4StateLens
# Needs 2 + 1 slots for mamba request with prefix cache. 2 for ping pong cache, 1 for running mamba state.
MAMBA_STATE_PER_REQ_PREFIX_CACHE = 3
# Lazy mode: 1 + 1 slots (1 ping-pong + 1 running), second ping-pong allocated on demand at boundary.
MAMBA_STATE_PER_REQ_PREFIX_CACHE_LAZY = 2
MAMBA_STATE_PER_REQ_NO_CACHE = 1
logger = logging.getLogger(__name__)
def kv_to_page_indices(kv_indices: np.ndarray, page_size: int):
# The page is guaranteed to be full except the last page.
if page_size == 1:
return kv_indices
return kv_indices[::page_size] // page_size
def kv_to_page_num(num_kv_indices: int, page_size: int):
return (num_kv_indices + page_size - 1) // page_size
def page_align_floor(length: int, page_size: int) -> int:
return (length // page_size) * page_size
def free_swa_out_of_window_slots(
req: Req,
pre_len: int,
*,
sliding_window_size: int,
page_size: int,
req_to_token_pool: ReqToTokenPool,
token_to_kv_pool_allocator: BaseTokenToKVPoolAllocator,
is_chunk_cache: bool = False,
) -> None:
# For swa radix cache, we need to evict the tokens that are not in the tree cache and also not in the sliding window
assert (
req.cache_protected_len % page_size == 0
), "cache_protected_len must be page aligned"
evict_floor = max(req.cache_protected_len, getattr(req, "swa_evict_floor", 0))
if page_size > 1 and evict_floor > req.cache_protected_len:
evict_floor = -(-evict_floor // page_size) * page_size
req.swa_evicted_seqlen = max(req.swa_evicted_seqlen, evict_floor)
if is_chunk_cache:
# Chunk cache builds no radix tree, so no tombstone-leaf concern; evict
# up to the window boundary (the trailing floor keeps it page-aligned).
evict_threshold = pre_len - sliding_window_size
else:
# Radix cache: keep max(window, page). The trailing floor page-aligns the
# frontier, and subtracting at least one page keeps it below the insert
# boundary (page_floor(seq_len)) so the last leaf is never all-tombstone.
# No extra page margin is needed.
evict_threshold = pre_len - max(sliding_window_size, page_size)
new_swa_evicted_seqlen = max(
req.swa_evicted_seqlen,
evict_threshold,
)
if page_size > 1:
new_swa_evicted_seqlen = (new_swa_evicted_seqlen // page_size) * page_size
if new_swa_evicted_seqlen > req.swa_evicted_seqlen:
free_slots = req_to_token_pool.req_to_token[
req.req_pool_idx, req.swa_evicted_seqlen : new_swa_evicted_seqlen
]
token_to_kv_pool_allocator.free_swa(free_slots)
maybe_evict_dsv4_state_on_swa(
token_to_kv_pool_allocator, req_to_token_pool, req, new_swa_evicted_seqlen
)
req.swa_evicted_seqlen = new_swa_evicted_seqlen
def maybe_cache_unfinished_req(req: Req, tree_cache: BasePrefixCache, **kwargs):
if getattr(req, "skip_radix_cache_insert", False):
return
tree_cache.cache_unfinished_req(req, **kwargs)
def write_cache_indices(
out_cache_loc: torch.Tensor,
req_pool_indices_tensor: torch.Tensor,
req_pool_indices_cpu: torch.Tensor,
prefix_lens_tensor: torch.Tensor,
prefix_lens_cpu: torch.Tensor,
seq_lens_tensor: torch.Tensor,
seq_lens_cpu: torch.Tensor,
extend_lens_tensor: torch.Tensor,
extend_lens_cpu: torch.Tensor,
prefix_tensors: list[torch.Tensor],
req_to_token_pool: ReqToTokenPool,
):
if support_triton(get_server_args().attention_backend):
prefix_pointers = torch.tensor(
[t.data_ptr() for t in prefix_tensors],
dtype=torch.uint64,
pin_memory=is_pin_memory_available(req_to_token_pool.device),
).to(req_to_token_pool.device, non_blocking=True)
# TODO: some tensors can be reused for ForwardBatchInfo (e.g., extend_lens, cumsum_start)
write_req_to_token_pool_triton[(req_pool_indices_tensor.shape[0],)](
req_to_token_pool.req_to_token,
req_pool_indices_tensor,
prefix_pointers,
prefix_lens_tensor,
seq_lens_tensor,
extend_lens_tensor,
out_cache_loc,
req_to_token_pool.req_to_token.shape[1],
)
else:
pt = 0
for i in range(req_pool_indices_cpu.shape[0]):
req_idx = req_pool_indices_cpu[i].item()
prefix_len = prefix_lens_cpu[i].item()
seq_len = seq_lens_cpu[i].item()
extend_len = extend_lens_cpu[i].item()
req_to_token_pool.write(
(req_idx, slice(0, prefix_len)),
prefix_tensors[i],
)
req_to_token_pool.write(
(req_idx, slice(prefix_len, seq_len)),
out_cache_loc[pt : pt + extend_len],
)
pt += extend_len
def get_last_loc(
req_to_token: torch.Tensor,
req_pool_indices_tensor: torch.Tensor,
prefix_lens_tensor: torch.Tensor,
) -> torch.Tensor:
attn_backend = get_server_args().attention_backend
uses_triton_dispatch = attn_backend not in ("ascend", "torch_native")
if _is_hip and uses_triton_dispatch:
# HIP-only: the legacy get_last_loc_triton kernel emits a
# mixed-width int32->int64 store that Triton mis-compiles on HIP,
# producing out-of-range last_loc values under EAGLE +
# page_size>1 (e.g. with aiter unified attention or the triton
# attention backend). The bug is in the Triton HIP codegen, not
# in any particular attention backend, so route every HIP path
# that would otherwise use get_last_loc_triton through the
# int32-safe variant. Non-HIP hardware keeps the original
# dispatcher below.
return get_last_loc_triton_safe(
req_to_token, req_pool_indices_tensor, prefix_lens_tensor
)
if uses_triton_dispatch:
impl = get_last_loc_triton
else:
impl = get_last_loc_torch
return impl(req_to_token, req_pool_indices_tensor, prefix_lens_tensor)
def get_last_loc_torch(
req_to_token: torch.Tensor,
req_pool_indices_tensor: torch.Tensor,
prefix_lens_tensor: torch.Tensor,
) -> torch.Tensor:
return torch.where(
prefix_lens_tensor > 0,
req_to_token[req_pool_indices_tensor, prefix_lens_tensor - 1],
torch.full_like(prefix_lens_tensor, -1),
)
def get_alloc_len_per_decode(server_args: Optional[ServerArgs] = None) -> int:
if server_args is None:
server_args = get_server_args()
if server_args.speculative_algorithm is None:
return 1
# Spec decoding allocates max(topk * num_steps, num_draft_tokens) per decode step.
spec_steps = server_args.speculative_num_steps or 1
spec_topk = server_args.speculative_eagle_topk or 1
spec_tokens = server_args.max_speculative_num_draft_tokens
page_size = server_args.page_size
from sglang.srt.speculative.spec_info import SpeculativeAlgorithm
spec_algo = SpeculativeAlgorithm.from_string(server_args.speculative_algorithm)
if page_size == 1 or spec_topk == 1 or not spec_algo.has_draft_kv():
return max(spec_steps * spec_topk, spec_tokens)
else:
# spec v2 tree (page>1, topk>1): worst-case page-aligned footprint per
# topk branch is ceil((page_size-1 + num_steps) / page) pages, each branch
# duplicated -- reserve for all topk branches.
num_new_pages_per_topk = (
(page_size - 1) + spec_steps + page_size - 1
) // page_size
return max(num_new_pages_per_topk * page_size * spec_topk, spec_tokens)
def get_alloc_reserve_per_decode(server_args: Optional[ServerArgs] = None) -> int:
"""KV length reserved per request at each decode step.
The 2x is a double-buffer that absorbs the kv_committed_len lag in overlap
mode; see eagle_utils.eagle_prepare_for_decode.
"""
return 2 * get_alloc_len_per_decode(server_args)
def get_req_to_token_extra_context_len(server_args: ServerArgs) -> int:
"""req_to_token row headroom beyond the model context length.
Sized to hold the decode over-allocation; the spec v2 page>1 topk>1 holey
draft footprint can outgrow the default num_draft_tokens headroom.
"""
# FIXME(lsyin): temporary fix for the context length issue under spec decoding
extra = 4 + (server_args.max_speculative_num_draft_tokens or 0)
if (
server_args.speculative_algorithm is not None
and server_args.page_size > 1
and (server_args.speculative_eagle_topk or 1) > 1
):
extra = max(extra, get_alloc_reserve_per_decode(server_args))
return extra
def alloc_token_slots(
tree_cache: BasePrefixCache,
num_tokens: int,
backup_state: bool = False,
):
allocator = tree_cache.token_to_kv_pool_allocator
evict_from_tree_cache(tree_cache, num_tokens)
state = None
if backup_state:
state = allocator.backup_state()
out_cache_loc = allocator.alloc(num_tokens)
if out_cache_loc is None:
error_msg = (
f"Out of memory. Try to lower your batch size.\n"
f"Try to allocate {num_tokens} tokens.\n"
f"{available_and_evictable_str(tree_cache)}"
)
logger.error(error_msg)
if tree_cache is not None:
tree_cache.pretty_print()
raise RuntimeError(error_msg)
return (out_cache_loc, state) if backup_state else out_cache_loc
def evict_from_tree_cache(tree_cache: BasePrefixCache | None, num_tokens: int):
if tree_cache is None:
return
if tree_cache.is_chunk_cache():
return
allocator = tree_cache.token_to_kv_pool_allocator
if isinstance(allocator, SWATokenToKVPoolAllocator):
# Hybrid allocator
full_available_size = allocator.full_available_size()
swa_available_size = allocator.swa_available_size()
if full_available_size < num_tokens or swa_available_size < num_tokens:
full_num_tokens = max(0, num_tokens - full_available_size)
swa_num_tokens = max(0, num_tokens - swa_available_size)
tree_cache.evict(
EvictParams(num_tokens=full_num_tokens, swa_num_tokens=swa_num_tokens)
)
else:
# Standard allocator
if allocator.available_size() < num_tokens:
tree_cache.evict(EvictParams(num_tokens=num_tokens))
def _compute_dsv4_state_lens(batch, *, is_decode: bool):
"""Per-req c{4,128}_state pool alloc lens (``DSV4StateLens``) for this step.
None on CUDA / non-V4 paths (allocator has no ``compute_dsv4_state_lens_*``).
"""
allocator = batch.token_to_kv_pool_allocator
if not hasattr(allocator, "compute_dsv4_state_lens_extend"):
return None
if is_decode:
return allocator.compute_dsv4_state_lens_decode(batch.reqs)
return allocator.compute_dsv4_state_lens_extend(
batch.reqs, batch.seq_lens_cpu.tolist()
)
def alloc_paged_token_slots_extend(
tree_cache: BasePrefixCache,
prefix_lens: torch.Tensor,
prefix_lens_cpu: torch.Tensor,
seq_lens: torch.Tensor,
seq_lens_cpu: torch.Tensor,
last_loc: torch.Tensor,
extend_num_tokens: int,
backup_state: bool = False,
req_pool_indices: Optional[torch.Tensor] = None,
dsv4_state_lens: Optional[DSV4StateLens] = None,
batch=None,
):
# Over estimate the number of tokens: assume each request needs a new page.
allocator = tree_cache.token_to_kv_pool_allocator
num_tokens = extend_num_tokens + len(seq_lens_cpu) * allocator.page_size
evict_from_tree_cache(tree_cache, num_tokens)
state = None
if backup_state:
state = allocator.backup_state()
is_dsv4 = req_pool_indices is not None and hasattr(allocator, "c4_attn_allocator")
extra_alloc_kwargs = {}
if is_dsv4:
extra_alloc_kwargs["req_pool_indices"] = req_pool_indices
# Per-call per-req tables for the c-pool / state last_loc lookup.
if batch is not None:
extra_alloc_kwargs["req_to_token_pool"] = batch.req_to_token_pool
if dsv4_state_lens is not None:
extra_alloc_kwargs["dsv4_state_lens"] = dsv4_state_lens
out = allocator.alloc_extend(
prefix_lens,
prefix_lens_cpu,
seq_lens,
seq_lens_cpu,
last_loc,
extend_num_tokens,
**extra_alloc_kwargs,
)
if is_dsv4:
bundle = out
out_cache_loc = None if bundle is None else bundle.out_full_loc
if batch is not None:
batch.out_cache_loc_dsv4 = bundle
else:
out_cache_loc = out
if out_cache_loc is None:
error_msg = (
f"Prefill out of memory. Try to lower your batch size.\n"
f"Try to allocate {extend_num_tokens} tokens.\n"
f"{available_and_evictable_str(tree_cache)}"
)
logger.error(error_msg)
if tree_cache is not None:
tree_cache.pretty_print()
raise RuntimeError(error_msg)
return (out_cache_loc, state) if backup_state else out_cache_loc
def alloc_req_slots(
req_to_token_pool: ReqToTokenPool,
reqs: list[Req],
tree_cache: BasePrefixCache | None,
) -> list[int]:
"""Allocate request slots from the pool.
Fail-loud: raises ``RuntimeError`` if the pool can't satisfy the batch. An
alloc failure here means the admission budget (``PrefillAdder``) was wrong
and should surface rather than be masked.
"""
num_reqs = len(reqs)
if isinstance(req_to_token_pool, HybridReqToTokenPool):
# Byte-coordinated for the shared allocator (accounts for the peer full
# sub-pool's bytes); plain slot free count for the non-shared one.
mamba_available_size = (
req_to_token_pool.mamba_allocator.schedulable_available_size()
)
# Eviction headroom factor: 3x (or lazy variant) for radix COW, 1x for chunk.
if tree_cache.supports_mamba():
factor = (
MAMBA_STATE_PER_REQ_PREFIX_CACHE_LAZY
if req_to_token_pool.enable_mamba_extra_buffer_lazy
else MAMBA_STATE_PER_REQ_PREFIX_CACHE
)
else:
factor = MAMBA_STATE_PER_REQ_NO_CACHE
mamba_state_needed = num_reqs * factor
if mamba_available_size < mamba_state_needed:
if tree_cache is not None and tree_cache.supports_mamba():
mamba_num = max(0, mamba_state_needed - mamba_available_size)
tree_cache.evict(EvictParams(num_tokens=0, mamba_num=mamba_num))
req_pool_indices = req_to_token_pool.alloc(reqs)
if req_pool_indices is None:
raise RuntimeError(
"alloc_req_slots runs out of memory. "
"Please set a smaller number for `--max-running-requests`. "
f"{req_to_token_pool.available_size()=}, {num_reqs=}, "
)
return req_pool_indices
def _alloc_page_size(batch: ScheduleBatch) -> int:
# DCP swaps in an allocator whose page_size is server_args.page_size *
# dcp_size, so it can be > 1 even when tree_cache.page_size is 1; branch on
# the real allocator's page_size there. Elsewhere the two are equal.
if (_is_hip or _is_cuda) and get_server_args().dcp_size > 1:
return batch.tree_cache.token_to_kv_pool_allocator.page_size
return batch.tree_cache.page_size
def alloc_for_extend(
batch: ScheduleBatch,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
"""
Allocate KV cache for extend batch and write to req_to_token_pool.
Returns ``(out_cache_loc, req_pool_indices_device, req_pool_indices_cpu)``
(the last is the host/CPU mirror). ``alloc_req_slots`` raises ``RuntimeError``
if the pool can't satisfy the batch (fail-loud — see its docstring).
"""
# free out-of-window swa tokens
batch.maybe_evict_swa()
prefix_tensors = [r.prefix_indices for r in batch.reqs]
# Create tensors for allocation
prefix_lens_cpu = torch.tensor(batch.prefix_lens, dtype=torch.int64)
extend_lens_cpu = torch.tensor(batch.extend_lens, dtype=torch.int64)
prefix_lens_device = prefix_lens_cpu.to(batch.device, non_blocking=True)
extend_lens_device = extend_lens_cpu.to(batch.device, non_blocking=True)
# Allocate req slots (raises RuntimeError if the pool is exhausted)
req_pool_indices = alloc_req_slots(
batch.req_to_token_pool, batch.reqs, batch.tree_cache
)
req_pool_indices_cpu = torch.tensor(req_pool_indices, dtype=torch.int64)
req_pool_indices_device = req_pool_indices_cpu.to(batch.device, non_blocking=True)
# Allocate KV cache (throws exception on failure)
if _alloc_page_size(batch) == 1:
out_cache_loc = alloc_token_slots(batch.tree_cache, batch.extend_num_tokens)
else:
# Paged allocation - build last_loc
last_loc = [
(t[-1:] if len(t) > 0 else torch.tensor([-1], device=batch.device))
for t in prefix_tensors
]
out_cache_loc = alloc_paged_token_slots_extend(
tree_cache=batch.tree_cache,
prefix_lens=prefix_lens_device,
prefix_lens_cpu=prefix_lens_cpu,
seq_lens=batch.seq_lens,
seq_lens_cpu=batch.seq_lens_cpu,
last_loc=torch.cat(last_loc),
extend_num_tokens=batch.extend_num_tokens,
req_pool_indices=req_pool_indices_device,
dsv4_state_lens=_compute_dsv4_state_lens(batch, is_decode=False),
batch=batch,
)
# Write to req_to_token_pool
write_cache_indices(
out_cache_loc,
req_pool_indices_device,
req_pool_indices_cpu,
prefix_lens_device,
prefix_lens_cpu,
batch.seq_lens,
batch.seq_lens_cpu,
extend_lens_device,
extend_lens_cpu,
prefix_tensors,
batch.req_to_token_pool,
)
# DSV4-NPU hook: no-op on non-DSV4 paths.
if _is_npu:
maybe_write_dsv4_extend(
batch,
req_pool_indices_cpu,
prefix_lens_cpu,
batch.seq_lens_cpu,
)
return out_cache_loc, req_pool_indices_device, req_pool_indices_cpu
def alloc_paged_token_slots_decode(
tree_cache: BasePrefixCache,
seq_lens: torch.Tensor,
seq_lens_cpu: torch.Tensor,
last_loc: torch.Tensor,
token_per_req: int = 1,
req_pool_indices: Optional[torch.Tensor] = None,
dsv4_state_lens: Optional[DSV4StateLens] = None,
batch=None,
) -> torch.Tensor:
"""Allocate paged KV cache for decode batch."""
allocator = tree_cache.token_to_kv_pool_allocator
# Over estimate the number of tokens: assume each request needs a new page.
num_tokens = len(seq_lens) * allocator.page_size
evict_from_tree_cache(tree_cache, num_tokens)
# DSV4-NPU allocator also needs req_pool_indices + per-req state lens and
# returns a DSV4OutCacheLoc bundle; hasattr-gated so others stay unchanged.
is_dsv4 = req_pool_indices is not None and hasattr(allocator, "c4_attn_allocator")
extra_alloc_kwargs = {}
if is_dsv4:
extra_alloc_kwargs["req_pool_indices"] = req_pool_indices
# Per-call per-req tables for the last_loc lookup.
if batch is not None:
extra_alloc_kwargs["req_to_token_pool"] = batch.req_to_token_pool
if dsv4_state_lens is not None:
extra_alloc_kwargs["dsv4_state_lens"] = dsv4_state_lens
out = allocator.alloc_decode(seq_lens, seq_lens_cpu, last_loc, **extra_alloc_kwargs)
if is_dsv4:
bundle = out
out_cache_loc = None if bundle is None else bundle.out_full_loc
if batch is not None:
batch.out_cache_loc_dsv4 = bundle
else:
out_cache_loc = out
if out_cache_loc is None:
error_msg = (
f"Decode out of memory. Try to lower your batch size.\n"
f"Try to allocate {len(seq_lens) * token_per_req} tokens.\n"
f"{available_and_evictable_str(tree_cache)}"
)
logger.error(error_msg)
if tree_cache is not None:
tree_cache.pretty_print()
raise RuntimeError(error_msg)
return out_cache_loc
def alloc_for_decode(batch: ScheduleBatch, token_per_req: int) -> torch.Tensor:
"""
Allocate KV cache for decode batch and write to req_to_token_pool.
Returns:
out_cache_loc: allocated cache locations
"""
batch.maybe_evict_swa()
seq_lens_gpu = batch.seq_lens
bs = seq_lens_gpu.shape[0]
if _alloc_page_size(batch) == 1:
# Non-paged allocation
out_cache_loc = alloc_token_slots(batch.tree_cache, bs * token_per_req)
else:
# Paged allocation
last_loc = batch.req_to_token_pool.req_to_token[
batch.req_pool_indices, seq_lens_gpu - 1
]
seq_lens_next = seq_lens_gpu + token_per_req
out_cache_loc = alloc_paged_token_slots_decode(
tree_cache=batch.tree_cache,
seq_lens=seq_lens_next,
seq_lens_cpu=batch.seq_lens_cpu + token_per_req,
last_loc=last_loc,
token_per_req=token_per_req,
req_pool_indices=batch.req_pool_indices,
dsv4_state_lens=_compute_dsv4_state_lens(batch, is_decode=True),
batch=batch,
)
# Write to req_to_token_pool
if batch.model_config.is_encoder_decoder:
locs = batch.encoder_lens + seq_lens_gpu
else:
locs = seq_lens_gpu.clone()
batch.req_to_token_pool.write(
(batch.req_pool_indices, locs), out_cache_loc.to(torch.int32)
)
# DSV4-NPU hook: no-op on non-DSV4 paths.
if _is_npu:
maybe_write_dsv4_decode(
batch,
batch.seq_lens_cpu + token_per_req,
token_per_req,
)
return out_cache_loc
def release_kv_cache(req: Req, tree_cache: BasePrefixCache, is_insert: bool = True):
# MambaRadixCache may alloc mamba state before alloc KV cache
if req.req_pool_idx is None:
assert (
tree_cache.supports_mamba()
), "Only MambaRadixCache allow freeing before alloc"
# TODO (csy, hanming): clean up this early allocation logic
if req.mamba_pool_idx is not None:
tree_cache.req_to_token_pool.mamba_allocator.free(
req.mamba_pool_idx.unsqueeze(-1)
)
req.mamba_pool_idx = None
return
tree_cache.cache_finished_req(
req,
is_insert=is_insert and not getattr(req, "skip_radix_cache_insert", False),
)
# StreamingSession.cache_finished_req handles speculative tail trim
# and bookkeeping flag sync internally, then sets req_pool_idx = None.
if req.req_pool_idx is None:
return
start_p, end_p = req.pop_overallocated_kv_cache()
global_server_args = get_server_args()
page_size = global_server_args.page_size
spec_algo = global_server_args.speculative_algorithm
# strip_thinking_cache intentionally reports output tokens as overallocated
# so they fall into the free path below (#22373).
if spec_algo is None and not global_server_args.strip_thinking_cache:
assert (
start_p == end_p
), f"Unexpected overallocated KV cache, {req.kv_committed_len=}, {req.kv_allocated_len=}"
if page_size > 1:
start_p = ceil_align(start_p, page_size)
if start_p < end_p:
indices_to_free = tree_cache.req_to_token_pool.req_to_token[req.req_pool_idx][
start_p:end_p
]
tree_cache.token_to_kv_pool_allocator.free(indices_to_free)
# If the prefix cache doesn't manage mamba states, we must free them here.
if isinstance(tree_cache.req_to_token_pool, HybridReqToTokenPool) and (
not tree_cache.supports_mamba()
):
assert (
req.mamba_pool_idx is not None
), "mamba state is freed while the tree cache does not manage mamba states"
tree_cache.req_to_token_pool.free_mamba_cache(req)
# DSV4-NPU's free() also releases c4/c128 state pages; no-op for others.
tree_cache.req_to_token_pool.free(req)
def available_and_evictable_str(tree_cache: BasePrefixCache) -> str:
return tree_cache.available_and_evictable_str()