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
831 lines
30 KiB
Python
831 lines
30 KiB
Python
from __future__ import annotations
|
|
|
|
from sglang.srt.mem_cache.cache_init_params import CacheInitParams
|
|
|
|
"""
|
|
Copyright 2023-2024 SGLang Team
|
|
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.
|
|
"""
|
|
|
|
"""
|
|
The radix tree data structure for managing the KV cache.
|
|
"""
|
|
|
|
import heapq
|
|
import logging
|
|
import sys
|
|
import time
|
|
from array import array
|
|
from collections import defaultdict
|
|
from typing import TYPE_CHECKING, Any, Iterator, List, Optional, Tuple, Union
|
|
|
|
import torch
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from sglang.srt.mem_cache.base_prefix_cache import (
|
|
BasePrefixCache,
|
|
DecLockRefParams,
|
|
DecLockRefResult,
|
|
EvictParams,
|
|
EvictResult,
|
|
IncLockRefResult,
|
|
InsertParams,
|
|
InsertResult,
|
|
MatchPrefixParams,
|
|
MatchResult,
|
|
)
|
|
from sglang.srt.mem_cache.events import KVCacheEventMixin
|
|
from sglang.srt.mem_cache.session_radix_cache import SessionRadixCacheMixin
|
|
from sglang.srt.mem_cache.utils import (
|
|
get_eviction_strategy,
|
|
get_hash_str,
|
|
split_node_hash_value,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from sglang.srt.managers.schedule_batch import Req
|
|
|
|
|
|
class RadixKey:
|
|
"""is_bigram=True: token_ids holds raw tokens (N+1 for N bigrams); slices share one boundary token."""
|
|
|
|
__slots__ = ("token_ids", "extra_key", "is_bigram", "limit")
|
|
|
|
def __init__(
|
|
self,
|
|
token_ids: array[int],
|
|
extra_key: Optional[str] = None,
|
|
is_bigram: bool = False,
|
|
limit: Optional[int] = None,
|
|
):
|
|
# token ids sequence (raw ints in both modes)
|
|
self.token_ids = token_ids
|
|
# extra key (e.g. lora_id, cache_salt)
|
|
self.extra_key = extra_key
|
|
# bigram view over token_ids: length = max(0, len(token_ids) - 1)
|
|
self.is_bigram = is_bigram
|
|
# Optional cap on raw tokens: behave as if token_ids were sliced to
|
|
# token_ids[:limit], without the O(n) copy. None = use all tokens.
|
|
self.limit = limit
|
|
|
|
def _raw_len(self) -> int:
|
|
n = len(self.token_ids)
|
|
if self.limit is not None and self.limit < n:
|
|
return self.limit
|
|
return n
|
|
|
|
def raw_token_ids(self) -> array:
|
|
"""token_ids honoring `limit` (copies only when capped)."""
|
|
n = self._raw_len()
|
|
t = self.token_ids
|
|
return t if n == len(t) else t[:n]
|
|
|
|
def __len__(self) -> int:
|
|
n = self._raw_len()
|
|
if self.is_bigram:
|
|
return n - 1 if n > 0 else 0
|
|
return n
|
|
|
|
# TODO(Jialin): vectorize with numpy without PyLong boxing
|
|
def __iter__(self) -> Iterator:
|
|
t = self.token_ids
|
|
n = self._raw_len()
|
|
if self.is_bigram:
|
|
for i in range(n - 1 if n > 0 else 0):
|
|
yield (t[i], t[i + 1])
|
|
elif n == len(t):
|
|
yield from t
|
|
else:
|
|
for i in range(n):
|
|
yield t[i]
|
|
|
|
def __getitem__(self, idx: Union[int, slice]) -> RadixKey:
|
|
# Normalize int -> 1-element slice so the rest handles one shape.
|
|
if isinstance(idx, int):
|
|
if idx < 0:
|
|
idx += len(self)
|
|
if idx < 0 or idx >= len(self):
|
|
raise IndexError(f"RadixKey index out of range: {idx}")
|
|
idx = slice(idx, idx + 1)
|
|
start, stop, step = idx.indices(len(self))
|
|
if step != 1:
|
|
raise ValueError("RadixKey slice step must be 1")
|
|
|
|
if self.is_bigram:
|
|
# bigrams [start, stop) span raw tokens [start, stop + 1);
|
|
# empty slice -> empty raw tokens (not a dangling boundary token).
|
|
raw = self.token_ids[start : stop + 1] if stop > start else array("q")
|
|
return RadixKey(raw, self.extra_key, is_bigram=True)
|
|
return RadixKey(self.token_ids[start:stop], self.extra_key)
|
|
|
|
def __repr__(self) -> str:
|
|
preview = self.token_ids[:10]
|
|
return f"RadixKey(extra_key={self.extra_key!r}, token_ids={preview}{'...' if len(self.token_ids) > 10 else ''}, is_bigram={self.is_bigram})"
|
|
|
|
def page_aligned(self, page_size: int) -> RadixKey:
|
|
if page_size == 1:
|
|
return self
|
|
aligned_len = len(self) // page_size * page_size
|
|
return self[:aligned_len]
|
|
|
|
def maybe_to_bigram_view(
|
|
self,
|
|
is_eagle: bool,
|
|
value: Optional[torch.Tensor] = None,
|
|
) -> Tuple[RadixKey, Optional[torch.Tensor]]:
|
|
# O(1): flip the bigram flag instead of materializing a tuple list.
|
|
# value is paired with raw tokens and gets truncated to the bigram count.
|
|
if is_eagle and not self.is_bigram:
|
|
self.is_bigram = True
|
|
if value is not None:
|
|
value = value[: len(self)]
|
|
return self, value
|
|
|
|
def _check_compatible(self, other: RadixKey) -> None:
|
|
if self.extra_key != other.extra_key:
|
|
raise ValueError(
|
|
f"RadixKey operations require matching extra_key, but got "
|
|
f"{self.extra_key=} != {other.extra_key=}"
|
|
)
|
|
|
|
def match(self, other: RadixKey, page_size: int = 1) -> int:
|
|
"""Logical-unit prefix length shared with ``other``. Result is rounded down to ``page_size``."""
|
|
self._check_compatible(other)
|
|
t0, t1 = self.token_ids, other.token_ids
|
|
assert type(t0) is type(t1), (type(t0), type(t1))
|
|
n = min(len(t0), len(t1))
|
|
|
|
# Exponential search for the first diverging token: gallop in doubling
|
|
# windows (one C-level slice compare each), then binary-search the window
|
|
# holding the divergence -- no per-token Python loop on long shared prefixes.
|
|
matched_tokens = n
|
|
lo = 0
|
|
step = 1
|
|
while lo < n:
|
|
hi = lo + step if lo + step < n else n
|
|
if t0[lo:hi] != t1[lo:hi]:
|
|
while hi - lo > 1:
|
|
mid = (lo + hi) // 2
|
|
if t0[lo:mid] == t1[lo:mid]:
|
|
lo = mid
|
|
else:
|
|
hi = mid
|
|
matched_tokens = lo
|
|
break
|
|
lo = hi
|
|
step *= 2
|
|
|
|
if self.is_bigram:
|
|
matched = max(0, min(matched_tokens - 1, len(self), len(other)))
|
|
return (matched // page_size) * page_size if page_size > 1 else matched
|
|
|
|
matched_tokens = min(matched_tokens, len(self), len(other))
|
|
if page_size == 1:
|
|
return matched_tokens
|
|
return (matched_tokens // page_size) * page_size
|
|
|
|
def child_key(self, page_size: int = 1):
|
|
"""Hashable dict-key for the first ``page_size`` logical units, namespaced by ``extra_key``."""
|
|
t = self.token_ids
|
|
if self.is_bigram:
|
|
if page_size == 1:
|
|
plain = (t[0], t[1])
|
|
else:
|
|
plain = tuple((t[j], t[j + 1]) for j in range(page_size))
|
|
else:
|
|
plain = t[0] if page_size == 1 else tuple(t[:page_size])
|
|
return plain if self.extra_key is None else (self.extra_key, plain)
|
|
|
|
def hash_page(self, start: int, end: int, prior_hash: Optional[str] = None) -> str:
|
|
"""SHA256 for logical units [start, end); bigram mode feeds overlapping (t_i, t_{i+1}) byte pairs."""
|
|
hash_value = get_hash_str(self[start:end], prior_hash)
|
|
assert isinstance(hash_value, str)
|
|
return hash_value
|
|
|
|
|
|
class TreeNode:
|
|
|
|
counter = 0
|
|
|
|
def __init__(self, id: Optional[int] = None, priority: int = 0):
|
|
self.children = defaultdict(TreeNode)
|
|
self.parent: TreeNode = None
|
|
self.key: RadixKey = None
|
|
self.value: Optional[torch.Tensor] = None
|
|
self.lock_ref = 0
|
|
self.last_access_time = time.monotonic()
|
|
self.creation_time = time.monotonic()
|
|
|
|
self.hit_count = 0
|
|
# indicating the node is locked to protect from eviction
|
|
# incremented when the node is referenced by a storage operation
|
|
self.host_ref_counter = 0
|
|
# store the host indices of KV cache
|
|
self.host_value: Optional[torch.Tensor] = None
|
|
self.write_through_pending_id: Optional[int] = None
|
|
# store hash values of each pages
|
|
self.hash_value: Optional[List[str]] = None
|
|
# priority for priority-aware eviction
|
|
self.priority = priority
|
|
|
|
self.id = TreeNode.counter if id is None else id
|
|
TreeNode.counter += 1
|
|
|
|
@property
|
|
def evicted(self):
|
|
return self.value is None
|
|
|
|
@property
|
|
def backuped(self):
|
|
return self.host_value is not None
|
|
|
|
def protect_host(self):
|
|
"""Protect the host value from eviction."""
|
|
self.host_ref_counter += 1
|
|
|
|
def release_host(self):
|
|
"""Release the host value, allowing it to be evicted."""
|
|
if self.host_ref_counter > 0:
|
|
self.host_ref_counter -= 1
|
|
else:
|
|
raise RuntimeError("Host reference counter is already zero.")
|
|
|
|
def get_last_hash_value(self) -> Optional[str]:
|
|
"""Returns the hash value of the last page in this node."""
|
|
if self.hash_value is None or len(self.hash_value) == 0:
|
|
return None
|
|
return self.hash_value[-1]
|
|
|
|
def get_prefix_hash_values(self, node: TreeNode) -> List[str]:
|
|
if node is None or node.hash_value is None:
|
|
return []
|
|
|
|
return node.get_prefix_hash_values(node.parent) + node.hash_value
|
|
|
|
def __lt__(self, other: TreeNode):
|
|
return self.last_access_time < other.last_access_time
|
|
|
|
|
|
class RadixCache(SessionRadixCacheMixin, KVCacheEventMixin, BasePrefixCache):
|
|
def __init__(self, params: CacheInitParams):
|
|
self.disable = params.disable
|
|
self.req_to_token_pool = params.req_to_token_pool
|
|
self.token_to_kv_pool_allocator = params.token_to_kv_pool_allocator
|
|
self.page_size = params.page_size
|
|
self.enable_kv_cache_events = params.enable_kv_cache_events
|
|
self.enable_session_radix_cache = params.enable_session_radix_cache
|
|
self.is_eagle = params.is_eagle
|
|
self.disable_finished_insert = params.disable_finished_insert
|
|
self.eviction_policy = params.eviction_policy.lower()
|
|
|
|
self.kv_event_queue = []
|
|
|
|
if params.enable_metrics:
|
|
self.init_metrics_collector()
|
|
|
|
if self.token_to_kv_pool_allocator:
|
|
dev = self.token_to_kv_pool_allocator.device
|
|
if isinstance(dev, (str, torch.device)):
|
|
self.device = torch.device(dev)
|
|
else:
|
|
self.device = torch.device("cpu")
|
|
else:
|
|
self.device = torch.device("cpu")
|
|
|
|
self.eviction_strategy = get_eviction_strategy(self.eviction_policy)
|
|
|
|
self.evictable_leaves = set()
|
|
self.reset()
|
|
|
|
@classmethod
|
|
def create_simulated(
|
|
self,
|
|
disable: bool = False,
|
|
mock_allocator: Optional[Any] = None,
|
|
page_size: int = 1,
|
|
enable_kv_cache_events: bool = False,
|
|
) -> RadixCache:
|
|
"""Init a radix cache without memory pools for simulation purpose."""
|
|
params = CacheInitParams(
|
|
disable=disable,
|
|
req_to_token_pool=None,
|
|
token_to_kv_pool_allocator=mock_allocator,
|
|
page_size=page_size,
|
|
enable_kv_cache_events=enable_kv_cache_events,
|
|
)
|
|
return RadixCache(params)
|
|
|
|
##### Public API #####
|
|
|
|
def reset(self):
|
|
# Initialize root with minimum priority so any real priority overrides it
|
|
self.root_node = TreeNode(priority=-sys.maxsize)
|
|
self.root_node.key = RadixKey(token_ids=array("q"), extra_key=None)
|
|
self.root_node.value = []
|
|
self.root_node.host_value = []
|
|
self.root_node.lock_ref = 1
|
|
self.root_node.hash_value = []
|
|
self.evictable_size_ = 0
|
|
self.protected_size_ = 0
|
|
self.evictable_leaves.clear()
|
|
self._reset_session_radix_state()
|
|
self._empty_match_result = MatchResult(
|
|
device_indices=torch.empty(
|
|
(0,),
|
|
dtype=torch.int64,
|
|
device=self.device,
|
|
),
|
|
last_device_node=self.root_node,
|
|
last_host_node=self.root_node,
|
|
best_match_node=self.root_node,
|
|
)
|
|
self._record_all_cleared_event()
|
|
|
|
def match_prefix(self, params: MatchPrefixParams) -> MatchResult:
|
|
"""Find the longest cached prefix of ``key`` in the radix tree.
|
|
|
|
The logical namespace for prefix matching is determined by both the
|
|
token id sequence and the optional ``extra_key`` carried by ``RadixKey``.
|
|
Entries that share identical leading token ids but have *different*
|
|
``extra_key`` values are intentionally kept disjoint and never share
|
|
prefix nodes. This is useful to:
|
|
|
|
* Isolate KV cache lines for different LoRA / adapter IDs.
|
|
* Separate requests that intentionally should not share state (e.g.,
|
|
different sampling salt, cache version, or retrieval augmentation
|
|
context) by supplying a distinct ``extra_key``.
|
|
|
|
Args:
|
|
params (MatchPrefixParams): Parameters containing the lookup key
|
|
with a list of token ids and an optional ``extra_key`` namespace tag.
|
|
If ``page_size > 1`` the length is internally truncated to a multiple
|
|
of ``page_size`` before matching. Passing an empty key returns an
|
|
empty result with the root as the last node.
|
|
|
|
Returns:
|
|
MatchResult: ``device_indices`` is a 1-D ``torch.int64`` tensor of
|
|
the concatenated KV cache indices corresponding to the longest
|
|
cached prefix (may be length 0).
|
|
``last_device_node`` and ``last_host_node`` (currently the same) are the tree node objects
|
|
representing the terminal node of the matched prefix. This method
|
|
may mutate internal structure by splitting an existing node if the
|
|
match ends inside a stored segment.
|
|
|
|
Internal updates:
|
|
* Refreshes access metadata (timestamps) used by the
|
|
configured eviction strategy.
|
|
* If the lookup ends inside a stored segment the node is split once
|
|
to expose a precise boundary; this structural refinement improves
|
|
subsequent match efficiency and does not duplicate data.
|
|
"""
|
|
key = params.key
|
|
key, _ = key.maybe_to_bigram_view(self.is_eagle)
|
|
|
|
if self.disable or len(key) == 0:
|
|
return self._empty_match_result
|
|
|
|
key = key.page_aligned(self.page_size)
|
|
|
|
if len(key) == 0:
|
|
return self._empty_match_result
|
|
|
|
value, last_node = self._match_prefix_helper(self.root_node, key)
|
|
if value:
|
|
value = torch.cat(value)
|
|
else:
|
|
value = self._empty_match_result.device_indices
|
|
return MatchResult(
|
|
device_indices=value,
|
|
last_device_node=last_node,
|
|
last_host_node=last_node,
|
|
best_match_node=last_node,
|
|
)
|
|
|
|
def insert(self, params: InsertParams) -> InsertResult:
|
|
if self.disable:
|
|
return InsertResult(prefix_len=0)
|
|
|
|
key = params.key
|
|
value = params.value
|
|
priority = params.priority
|
|
chunked = params.chunked
|
|
|
|
key, value = key.maybe_to_bigram_view(self.is_eagle, value)
|
|
key = key.page_aligned(self.page_size)
|
|
if value is not None:
|
|
value = value[: len(key)]
|
|
else:
|
|
# Debug/test fallback: use token ids themselves as values.
|
|
value = torch.tensor(key.token_ids[: len(key)], dtype=torch.int64)
|
|
|
|
prefix_len, last_node = self._insert_helper(
|
|
self.root_node, key, value, priority, chunked
|
|
)
|
|
return InsertResult(prefix_len=prefix_len, last_device_node=last_node)
|
|
|
|
def cache_finished_req(self, req: Req, is_insert: bool = True):
|
|
"""Cache request when it finishes."""
|
|
# In deterministic mode, disable finished request insertion to radix cache
|
|
if self.disable_finished_insert:
|
|
is_insert = False
|
|
|
|
kv_committed_len = req.pop_committed_kv_cache()
|
|
if self.disable:
|
|
kv_indices = self.req_to_token_pool.req_to_token[
|
|
req.req_pool_idx, :kv_committed_len
|
|
]
|
|
self.token_to_kv_pool_allocator.free(kv_indices)
|
|
return
|
|
|
|
token_ids = (req.origin_input_ids + req.output_ids)[:kv_committed_len]
|
|
kv_indices = self.req_to_token_pool.req_to_token[
|
|
req.req_pool_idx, : len(token_ids)
|
|
]
|
|
|
|
radix_key = RadixKey(
|
|
token_ids, req.extra_key, is_bigram=self.is_eagle
|
|
).page_aligned(self.page_size)
|
|
key_len = len(radix_key)
|
|
values = kv_indices[:key_len].to(dtype=torch.int64, copy=True)
|
|
|
|
# Radix Cache takes one ref in memory pool
|
|
if is_insert:
|
|
priority = getattr(req, "priority", 0) or 0
|
|
result = self.insert(
|
|
InsertParams(key=radix_key, value=values, priority=priority)
|
|
)
|
|
session_leaf = result.last_device_node
|
|
# Free the duplicates that were already in the tree
|
|
self.token_to_kv_pool_allocator.free(
|
|
kv_indices[req.cache_protected_len : result.prefix_len]
|
|
)
|
|
else:
|
|
session_leaf = None
|
|
self.token_to_kv_pool_allocator.free(
|
|
kv_indices[req.cache_protected_len : key_len]
|
|
)
|
|
|
|
# free the unaligned tail
|
|
self.token_to_kv_pool_allocator.free(kv_indices[key_len:])
|
|
|
|
self._tag_session_leaf(req, radix_key, node=session_leaf)
|
|
|
|
# Remove req slot release the cache lock
|
|
if req.last_node is not None:
|
|
self.dec_lock_ref(req.last_node)
|
|
|
|
def cache_unfinished_req(self, req: Req, chunked=False):
|
|
"""Cache request when it is unfinished."""
|
|
if self.disable:
|
|
return
|
|
|
|
token_ids = req.get_fill_ids()
|
|
kv_indices = self.req_to_token_pool.req_to_token[
|
|
req.req_pool_idx, : len(token_ids)
|
|
]
|
|
|
|
radix_key = RadixKey(
|
|
token_ids, req.extra_key, is_bigram=self.is_eagle
|
|
).page_aligned(self.page_size)
|
|
values = kv_indices[: len(radix_key)].to(dtype=torch.int64, copy=True)
|
|
|
|
# Radix Cache takes one ref in memory pool
|
|
result = self.insert(
|
|
InsertParams(
|
|
key=radix_key,
|
|
value=values,
|
|
chunked=chunked,
|
|
priority=getattr(req, "priority", 0) or 0,
|
|
)
|
|
)
|
|
new_prefix_len = result.prefix_len
|
|
|
|
self.token_to_kv_pool_allocator.free(
|
|
kv_indices[req.cache_protected_len : new_prefix_len]
|
|
)
|
|
|
|
# The prefix indices could be updated, reuse it
|
|
match_result = self.match_prefix(MatchPrefixParams(key=radix_key))
|
|
new_indices, new_last_node = (
|
|
match_result.device_indices,
|
|
match_result.last_device_node,
|
|
)
|
|
assert len(new_indices) == len(
|
|
radix_key
|
|
), f"{len(new_indices)=}, {len(radix_key)=}"
|
|
|
|
self.req_to_token_pool.write(
|
|
(req.req_pool_idx, slice(req.cache_protected_len, len(new_indices))),
|
|
new_indices[req.cache_protected_len :],
|
|
)
|
|
|
|
# The cache_protected_len is not always equal to len(req.prefix_indices)
|
|
# since for page_size > 1, the partial part is added to req.prefix_indices, but that part of kv indices is not added to the tree.
|
|
# It should be freed in the next cache_unfinished_req and final cache_finished_req to avoid memory leak.
|
|
# So we introduce this `cache_protected_len` field to make sure the partial part can be freed correctly.
|
|
req.cache_protected_len = len(new_indices)
|
|
|
|
self.dec_lock_ref(req.last_node)
|
|
self.inc_lock_ref(new_last_node)
|
|
|
|
# `req.prefix_indices` will be used in `PrefillAdder::add_chunked_req` later
|
|
# - page_size != 1: there is a partial page at the end, keep the full kv_indices
|
|
# - eagle case: bigram keys will only cache len - 1 kv indices
|
|
if len(new_indices) < len(kv_indices):
|
|
req.prefix_indices = torch.cat(
|
|
[new_indices, kv_indices[len(new_indices) :]]
|
|
)
|
|
else:
|
|
req.prefix_indices = new_indices
|
|
|
|
req.last_node = new_last_node
|
|
|
|
self._tag_session_leaf(req, radix_key, node=new_last_node)
|
|
|
|
def pretty_print(self):
|
|
self._print_helper(self.root_node, 0)
|
|
print(f"#tokens: {self.total_size()}")
|
|
|
|
def total_size(self):
|
|
return self._total_size_helper()
|
|
|
|
def evict(self, params: EvictParams) -> EvictResult:
|
|
if self.disable:
|
|
return EvictResult()
|
|
|
|
start_time = time.perf_counter()
|
|
num_tokens = params.num_tokens
|
|
leaves = list(self.evictable_leaves)
|
|
eviction_heap = [
|
|
(self.eviction_strategy.get_priority(node), node) for node in leaves
|
|
]
|
|
heapq.heapify(eviction_heap)
|
|
|
|
num_evicted = 0
|
|
while num_evicted < num_tokens and len(eviction_heap):
|
|
_priority, x = heapq.heappop(eviction_heap)
|
|
|
|
self.token_to_kv_pool_allocator.free(x.value)
|
|
num_evicted += len(x.value)
|
|
self._delete_leaf(x)
|
|
|
|
if len(x.parent.children) == 0 and x.parent.lock_ref == 0:
|
|
new_priority = self.eviction_strategy.get_priority(x.parent)
|
|
heapq.heappush(eviction_heap, (new_priority, x.parent))
|
|
|
|
self._record_remove_event(x)
|
|
|
|
self.update_eviction_metrics(num_evicted, start_time)
|
|
return EvictResult(num_tokens_evicted=num_evicted)
|
|
|
|
def inc_lock_ref(self, node: TreeNode) -> IncLockRefResult:
|
|
if self.disable:
|
|
return IncLockRefResult(delta=0)
|
|
|
|
delta = 0
|
|
while node != self.root_node:
|
|
if node.lock_ref == 0:
|
|
self.evictable_size_ -= len(node.key)
|
|
self.protected_size_ += len(node.key)
|
|
delta -= len(node.key)
|
|
node.lock_ref += 1
|
|
self._update_leaf_status(node)
|
|
node = node.parent
|
|
return IncLockRefResult(delta=delta)
|
|
|
|
def dec_lock_ref(
|
|
self, node: TreeNode, params: Optional[DecLockRefParams] = None
|
|
) -> DecLockRefResult:
|
|
if self.disable:
|
|
return DecLockRefResult(delta=0)
|
|
|
|
delta = 0
|
|
while node != self.root_node:
|
|
if node.lock_ref == 1:
|
|
self.evictable_size_ += len(node.key)
|
|
self.protected_size_ -= len(node.key)
|
|
delta += len(node.key)
|
|
node.lock_ref -= 1
|
|
self._update_leaf_status(node)
|
|
if node.parent is None:
|
|
assert (
|
|
node is self.root_node
|
|
), "This request holds the node from another tree"
|
|
node = node.parent
|
|
return DecLockRefResult(delta=delta)
|
|
|
|
def evictable_size(self):
|
|
return self.evictable_size_
|
|
|
|
def protected_size(self):
|
|
# protected size refers to the size of the cache that is locked
|
|
return self.protected_size_
|
|
|
|
def all_values_flatten(self):
|
|
values = []
|
|
|
|
def _dfs_helper(node: TreeNode):
|
|
for _, child in node.children.items():
|
|
values.append(child.value)
|
|
_dfs_helper(child)
|
|
|
|
_dfs_helper(self.root_node)
|
|
return torch.cat(values)
|
|
|
|
##### Internal Helper Functions #####
|
|
|
|
def _match_prefix_helper(self, node: TreeNode, key: RadixKey):
|
|
access_time = time.monotonic()
|
|
node.last_access_time = access_time
|
|
|
|
child_key = key.child_key(self.page_size)
|
|
|
|
value = []
|
|
while len(key) > 0 and child_key in node.children.keys():
|
|
child = node.children[child_key]
|
|
child.last_access_time = access_time
|
|
prefix_len = child.key.match(key, page_size=self.page_size)
|
|
if prefix_len < len(child.key):
|
|
new_node = self._split_node(child.key, child, prefix_len)
|
|
value.append(new_node.value)
|
|
node = new_node
|
|
break
|
|
else:
|
|
value.append(child.value)
|
|
node = child
|
|
key = key[prefix_len:]
|
|
|
|
if len(key):
|
|
child_key = key.child_key(self.page_size)
|
|
|
|
return value, node
|
|
|
|
def _split_node(self, key: RadixKey, child: TreeNode, split_len: int):
|
|
# new_node -> child
|
|
# New node inherits child's priority (represents shared prefix)
|
|
new_node = TreeNode(priority=child.priority)
|
|
new_node.hit_count = child.hit_count
|
|
new_node.children = {key[split_len:].child_key(self.page_size): child}
|
|
new_node.parent = child.parent
|
|
new_node.lock_ref = child.lock_ref
|
|
new_node.key = child.key[:split_len]
|
|
new_node.value = child.value[:split_len].clone()
|
|
child.parent = new_node
|
|
child.key = child.key[split_len:]
|
|
child.value = child.value[split_len:].clone()
|
|
new_node.parent.children[key.child_key(self.page_size)] = new_node
|
|
|
|
# Split hash_value if it was already computed, otherwise leave as None
|
|
new_node.hash_value, child.hash_value = split_node_hash_value(
|
|
child.hash_value, split_len, self.page_size
|
|
)
|
|
|
|
return new_node
|
|
|
|
def _inc_hit_count(self, node: TreeNode, chunked: bool = False):
|
|
# Skip the hit count update for chunked requests to avoid self-referencing
|
|
# inflation where a chunked request increments hit_count on nodes it created
|
|
# in previous chunks.
|
|
if chunked:
|
|
return
|
|
node.hit_count += 1
|
|
|
|
def _insert_helper(
|
|
self,
|
|
node: TreeNode,
|
|
key: RadixKey,
|
|
value,
|
|
priority: int = 0,
|
|
chunked: bool = False,
|
|
):
|
|
# Convert None priority to 0
|
|
if priority is None:
|
|
priority = 0
|
|
access_time = time.monotonic()
|
|
node.last_access_time = access_time
|
|
# Update priority along the path (take max to propagate higher priority)
|
|
node.priority = max(node.priority, priority)
|
|
if len(key) == 0:
|
|
return 0, node
|
|
|
|
child_key = key.child_key(self.page_size)
|
|
|
|
total_prefix_length = 0
|
|
while len(key) > 0 and child_key in node.children.keys():
|
|
node = node.children[child_key]
|
|
node.last_access_time = access_time
|
|
prefix_len = node.key.match(key, page_size=self.page_size)
|
|
total_prefix_length += prefix_len
|
|
key = key[prefix_len:]
|
|
value = value[prefix_len:]
|
|
|
|
if prefix_len < len(node.key):
|
|
new_node = self._split_node(node.key, node, prefix_len)
|
|
new_node.priority = max(new_node.priority, priority)
|
|
self._inc_hit_count(new_node, chunked)
|
|
node = new_node
|
|
else:
|
|
node.priority = max(node.priority, priority)
|
|
self._inc_hit_count(node, chunked)
|
|
if len(key):
|
|
child_key = key.child_key(self.page_size)
|
|
|
|
if len(key):
|
|
new_node = TreeNode(priority=priority)
|
|
new_node.parent = node
|
|
new_node.key = key
|
|
new_node.value = value.clone()
|
|
self._inc_hit_count(new_node, chunked)
|
|
node.children[child_key] = new_node
|
|
self.evictable_size_ += len(key)
|
|
self._update_leaf_status(node)
|
|
self._update_leaf_status(new_node)
|
|
# Hash will be computed lazily during event emission
|
|
self._record_store_event(new_node)
|
|
node = new_node
|
|
return total_prefix_length, node
|
|
|
|
def _print_helper(self, node: TreeNode, indent: int):
|
|
"""Prints the radix tree in a human-readable format."""
|
|
stack = [(node, indent)]
|
|
while stack:
|
|
current_node, current_indent = stack.pop()
|
|
print(
|
|
" " * current_indent,
|
|
len(current_node.key),
|
|
current_node.key.token_ids[:10],
|
|
f"r={current_node.lock_ref}",
|
|
)
|
|
for key, child in current_node.children.items():
|
|
stack.append((child, current_indent + 2))
|
|
|
|
assert key == child.key.child_key(
|
|
self.page_size
|
|
), f"{key=}, {child.key.child_key(self.page_size)=}"
|
|
|
|
def _delete_leaf(self, node):
|
|
key = node.key.child_key(self.page_size)
|
|
v = node.parent.children.pop(key, None)
|
|
assert v == node, f"parent does not have child key, {key}"
|
|
|
|
self._discard_session_leaf(node)
|
|
self.evictable_size_ -= len(node.key)
|
|
if node in self.evictable_leaves:
|
|
self.evictable_leaves.remove(node)
|
|
self._update_leaf_status(node.parent)
|
|
|
|
def _update_leaf_status(self, node: TreeNode):
|
|
if node.evicted or node.lock_ref > 0:
|
|
if node in self.evictable_leaves:
|
|
self.evictable_leaves.remove(node)
|
|
return
|
|
|
|
for child in node.children.values():
|
|
if not child.evicted:
|
|
if node in self.evictable_leaves:
|
|
self.evictable_leaves.remove(node)
|
|
return
|
|
|
|
if node not in self.evictable_leaves:
|
|
self.evictable_leaves.add(node)
|
|
|
|
def _total_size_helper(self):
|
|
total_size = 0
|
|
stack = [self.root_node]
|
|
while stack:
|
|
current_node = stack.pop()
|
|
total_size += len(current_node.value)
|
|
for child in current_node.children.values():
|
|
if child.evicted:
|
|
continue
|
|
stack.append(child)
|
|
return total_size
|
|
|
|
|
|
if __name__ == "__main__":
|
|
tree = RadixCache.create_simulated()
|
|
|
|
tree.insert(InsertParams(key=RadixKey(token_ids=array("q", [1, 2, 3]))))
|
|
tree.insert(InsertParams(key=RadixKey(token_ids=array("q", [1, 2, 3]))))
|
|
tree.insert(InsertParams(key=RadixKey(token_ids=array("q", [1, 2, 4, 5]))))
|
|
tree.insert(InsertParams(key=RadixKey(token_ids=array("q", [1, 2, 4, 5, 6, 7]))))
|
|
tree.insert(InsertParams(key=RadixKey(token_ids=array("q", [8, 9, 10, 11, 12]))))
|
|
tree.pretty_print()
|
|
|
|
print(
|
|
tree.match_prefix(
|
|
MatchPrefixParams(key=RadixKey(token_ids=array("q", [1, 2, 3, 13, 14])))
|
|
)
|
|
)
|