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
1418 lines
58 KiB
Python
1418 lines
58 KiB
Python
from __future__ import annotations
|
|
|
|
"""
|
|
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 hybrid (full and SWA) KV cache.
|
|
"""
|
|
|
|
import heapq
|
|
import time
|
|
from collections import defaultdict
|
|
from typing import TYPE_CHECKING, List, Optional, Tuple
|
|
|
|
import torch
|
|
from numpy import float64
|
|
|
|
from sglang.srt.environ import envs
|
|
from sglang.srt.mem_cache.allocator.swa import SWATokenToKVPoolAllocator
|
|
from sglang.srt.mem_cache.base_prefix_cache import (
|
|
BasePrefixCache,
|
|
DecLockRefParams,
|
|
DecLockRefResult,
|
|
EvictParams,
|
|
EvictResult,
|
|
IncLockRefResult,
|
|
InsertParams,
|
|
InsertResult,
|
|
MatchPrefixParams,
|
|
MatchResult,
|
|
)
|
|
from sglang.srt.mem_cache.cache_init_params import CacheInitParams
|
|
from sglang.srt.mem_cache.events import KVCacheEventMixin
|
|
from sglang.srt.mem_cache.radix_cache import RadixKey
|
|
from sglang.srt.mem_cache.utils import split_node_hash_value
|
|
|
|
if TYPE_CHECKING:
|
|
from sglang.srt.managers.schedule_batch import Req
|
|
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TreeNode:
|
|
|
|
counter = 0
|
|
swa_uuid_counter = 1
|
|
last_access_time_counter_float = float64(1.0)
|
|
|
|
def __init__(self, id: Optional[int] = None):
|
|
self.children = defaultdict(TreeNode)
|
|
self.parent: TreeNode = None
|
|
self.key: RadixKey = None
|
|
self.value: Optional[torch.Tensor] = None
|
|
# swa_tombstone is used to indicate the kv indices have been freed for swa layers
|
|
self.swa_tombstone = False
|
|
# invariant: for any node, if swa_lock_ref is locked, full_lock_ref must be locked;
|
|
# if full_lock_ref is locked, swa_lock_ref doesn't need to be locked. So,
|
|
# full_lock_ref is always >= swa_lock_ref.
|
|
self.full_lock_ref = 0
|
|
self.swa_lock_ref = 0
|
|
# last access time is only used for sanity check. LRU is maintained by the lru list.
|
|
self.last_access_time = get_last_access_time()
|
|
|
|
self.hit_count = 0
|
|
# store the host indices of KV cache
|
|
self.host_value = None
|
|
# store hash values of each page
|
|
self.hash_value: Optional[List[str]] = None
|
|
|
|
# for lru list, invariant:
|
|
# 1. prev has greater last_access_time
|
|
# 2. next has smaller last_access_time
|
|
self.prev = None
|
|
self.next = None
|
|
self.swa_prev = None
|
|
self.swa_next = None
|
|
|
|
self.id = TreeNode.counter if id is None else id
|
|
TreeNode.counter += 1
|
|
self.swa_uuid = None
|
|
|
|
@property
|
|
def evicted(self):
|
|
return self.value is None
|
|
|
|
@property
|
|
def backuped(self):
|
|
return self.host_value is not None
|
|
|
|
def __lt__(self, other: TreeNode):
|
|
return self.last_access_time < other.last_access_time
|
|
|
|
|
|
def gen_swa_uuid() -> int:
|
|
TreeNode.swa_uuid_counter += 1
|
|
return TreeNode.swa_uuid_counter
|
|
|
|
|
|
def get_last_access_time() -> float64:
|
|
ret = TreeNode.last_access_time_counter_float
|
|
TreeNode.last_access_time_counter_float += 1.0
|
|
return ret
|
|
|
|
|
|
class LRUList:
|
|
def __init__(self, is_swa_list: bool = False):
|
|
self.is_swa_list = is_swa_list
|
|
if self.is_swa_list:
|
|
self.prv = "swa_prev"
|
|
self.nxt = "swa_next"
|
|
self.lock_ref = "swa_lock_ref"
|
|
else:
|
|
self.prv = "prev"
|
|
self.nxt = "next"
|
|
self.lock_ref = "full_lock_ref"
|
|
# Initialize dummy head and tail nodes
|
|
self.head = TreeNode() # Most recently used side
|
|
self.tail = TreeNode() # Least recently used side
|
|
setattr(self.head, self.nxt, self.tail) # self.head.next = self.tail
|
|
setattr(self.tail, self.prv, self.head) # self.tail.prev = self.head
|
|
self.cache = {}
|
|
|
|
def _add_node(self, node):
|
|
"""Helper to add node right after head (most recently used)"""
|
|
self._add_node_after(self.head, node)
|
|
|
|
def _add_node_after(self, old_node, new_node):
|
|
"""Helper to add node right after old_node"""
|
|
setattr(new_node, self.prv, old_node) # new_node.prev = old_node
|
|
setattr(
|
|
new_node, self.nxt, getattr(old_node, self.nxt)
|
|
) # new_node.next = old_node.next
|
|
setattr(
|
|
getattr(old_node, self.nxt), self.prv, new_node
|
|
) # old_node.next.prev = new_node
|
|
setattr(old_node, self.nxt, new_node) # old_node.next = new_node
|
|
|
|
def _remove_node(self, node):
|
|
"""Helper to remove node from linked list"""
|
|
setattr(
|
|
getattr(node, self.prv), self.nxt, getattr(node, self.nxt)
|
|
) # node.prev.next = node.next
|
|
setattr(
|
|
getattr(node, self.nxt), self.prv, getattr(node, self.prv)
|
|
) # node.next.prev = node.prev
|
|
# Clear self pointers to break reference cycles among evicted nodes.
|
|
setattr(node, self.prv, None)
|
|
setattr(node, self.nxt, None)
|
|
|
|
def _get_lru(self) -> Optional[TreeNode]:
|
|
"""
|
|
Get the least recently used node
|
|
"""
|
|
if len(self.cache) == 0:
|
|
return None
|
|
return getattr(self.tail, self.prv)
|
|
|
|
def reset_node_mru(self, node):
|
|
"""
|
|
Move a (existing) node to most recently used position
|
|
"""
|
|
assert node.id in self.cache, f"Resetting node {node.id=} not in lru list"
|
|
assert (
|
|
not self.is_swa_list or not node.swa_tombstone
|
|
), f"Resetting swa tombstone node in swa lru list: {node.id=}"
|
|
self._remove_node(node)
|
|
self._add_node(node)
|
|
|
|
def reset_node_and_parents_mru(self, node, root_node):
|
|
"""
|
|
Move an (existing) node and its parents to most recently used position. Child node is
|
|
more recently used than parent node.
|
|
"""
|
|
prev_node = self.head
|
|
while node != root_node:
|
|
# for swa lru list, only reset non-tombstone nodes
|
|
if not self.is_swa_list or not node.swa_tombstone:
|
|
assert (
|
|
node.id in self.cache
|
|
), f"Resetting node {node.id=} not in lru list when resetting node and parents mru"
|
|
self._remove_node(node)
|
|
self._add_node_after(prev_node, node)
|
|
prev_node = node
|
|
node = node.parent
|
|
|
|
def insert_mru(self, node):
|
|
"""
|
|
Insert a (new) node as most recently used
|
|
"""
|
|
assert (
|
|
not self.is_swa_list or not node.swa_tombstone
|
|
), f"Inserting swa tombstone node in swa lru list: {node.id=}"
|
|
assert (
|
|
node.id not in self.cache
|
|
), f"Inserting node {node.id=} already in lru list, existing node: {self.cache[node.id].id=}"
|
|
self.cache[node.id] = node
|
|
self._add_node(node)
|
|
|
|
def remove_node(self, node: TreeNode):
|
|
"""
|
|
Remove node from lru list
|
|
"""
|
|
assert node.id in self.cache, f"Removing node {node.id=} not in lru list"
|
|
assert (
|
|
not self.is_swa_list or not node.swa_tombstone
|
|
), f"Removing swa tombstone node from swa lru list: {node.id=}"
|
|
del self.cache[node.id]
|
|
self._remove_node(node)
|
|
|
|
def get_lru_no_lock(self) -> Optional[TreeNode]:
|
|
"""
|
|
Get the least recently used node that is not locked
|
|
"""
|
|
return self.get_prev_no_lock(self.tail, check_id=False)
|
|
|
|
def get_leaf_lru_no_lock(self) -> Optional[TreeNode]:
|
|
"""
|
|
Get the least recently used leaf node that is not locked
|
|
"""
|
|
return self.get_prev_leaf_no_lock(self.tail, check_id=False)
|
|
|
|
def get_prev_no_lock(
|
|
self, node: TreeNode, check_id: bool = True
|
|
) -> Optional[TreeNode]:
|
|
"""
|
|
Get the previous (i.e. more recently used) node that is not locked
|
|
"""
|
|
if check_id:
|
|
assert (
|
|
node.id in self.cache
|
|
), f"Getting prev of node {node.id=} not in lru list"
|
|
x = getattr(node, self.prv) # x = node.prev
|
|
while getattr(x, self.lock_ref) > 0:
|
|
x = getattr(x, self.prv) # x = x.prev
|
|
# if x is the head, it means there is no node in the lru list without lock
|
|
if x == self.head:
|
|
return None
|
|
return x
|
|
|
|
def get_prev_leaf_no_lock(self, node: TreeNode, check_id: bool = True):
|
|
"""
|
|
Get the previous (i.e. more recently used) leaf node that is not locked
|
|
"""
|
|
if check_id:
|
|
assert (
|
|
node.id in self.cache
|
|
), f"Getting prev of node {node.id=} not in lru list"
|
|
x = getattr(node, self.prv) # x = node.prev
|
|
while getattr(x, self.lock_ref) > 0 or len(x.children) > 0:
|
|
x = getattr(x, self.prv) # x = x.prev
|
|
# if x is the head, it means there is no leaf node in the lru list without lock
|
|
if x == self.head:
|
|
return None
|
|
return x
|
|
|
|
def in_list(self, node: Optional[TreeNode]):
|
|
"""
|
|
Check if the node is in the lru list
|
|
"""
|
|
if not node:
|
|
return False
|
|
return node.id in self.cache
|
|
|
|
# Note: this is expensive, only use for debug
|
|
def sanity_check_evictable_size(self):
|
|
"""
|
|
Check the evictable size (i.e. the size of the nodes that are not locked)
|
|
"""
|
|
node = self.get_lru_no_lock()
|
|
evictable_size = 0
|
|
while self.in_list(node):
|
|
evictable_size += len(node.value)
|
|
node = self.get_prev_no_lock(node)
|
|
return evictable_size
|
|
|
|
# Note: this is expensive, only use for debug or idle check
|
|
def sanity_check(self, tree_cache: SWARadixCache):
|
|
"""
|
|
Check if the lru list is valid by rebuilding the lru list from the tree, heapifying it, and
|
|
checking if the lru list is valid.
|
|
"""
|
|
try:
|
|
if self.is_swa_list:
|
|
nodes = tree_cache._collect_nontombstone_nodes()
|
|
else:
|
|
nodes = tree_cache._collect_all_nodes()
|
|
total_nodes = len(nodes)
|
|
total_lru_plus_1 = len(self.cache) + 1
|
|
# heapify based on last_access_time
|
|
heapq.heapify(nodes)
|
|
# the root node is not in the lru list
|
|
assert (
|
|
len(nodes) == len(self.cache) + 1
|
|
), f"len(nodes): {len(nodes)} != len(self.cache) + 1: {len(self.cache) + 1}"
|
|
|
|
x_lru = self._get_lru()
|
|
while len(nodes):
|
|
x = heapq.heappop(nodes)
|
|
if x == tree_cache.root_node:
|
|
# root node is not in the lru list
|
|
continue
|
|
assert (
|
|
x == x_lru
|
|
), f"Incorrect LRU list, {self.is_swa_list=}, x: {x.id=} != x_lru: {x_lru.id=}"
|
|
assert (
|
|
x_lru.full_lock_ref == 0
|
|
), f"x_lru should not be locked when idle, {x_lru.full_lock_ref=}, {x_lru.swa_uuid=}, {x_lru.id=}"
|
|
assert (
|
|
x_lru.swa_lock_ref == 0
|
|
), f"x_lru should not be locked when idle, {x_lru.swa_lock_ref=}, {x_lru.swa_uuid=}, {x_lru.id=}"
|
|
x_lru = getattr(x, self.prv)
|
|
|
|
if self.is_swa_list:
|
|
evictable_size = tree_cache.swa_evictable_size()
|
|
lru_list_evictable_size = self.sanity_check_evictable_size()
|
|
else:
|
|
evictable_size = tree_cache.full_evictable_size()
|
|
lru_list_evictable_size = self.sanity_check_evictable_size()
|
|
|
|
assert (
|
|
evictable_size == lru_list_evictable_size
|
|
), f"{self.is_swa_list=}, total nodes: {total_nodes}, total lru plus 1: {total_lru_plus_1}, evictable size: {evictable_size} != lru list evictable size: {lru_list_evictable_size}"
|
|
except Exception as e:
|
|
msg = f"SWA Radix tree sanity check failed, ping @hanming-lu: {e}"
|
|
logger.error(msg)
|
|
raise Exception(msg)
|
|
|
|
|
|
class SWARadixCache(KVCacheEventMixin, BasePrefixCache):
|
|
def __init__(self, params: CacheInitParams):
|
|
assert isinstance(params.token_to_kv_pool_allocator, SWATokenToKVPoolAllocator)
|
|
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.disable = params.disable
|
|
self.is_eagle = params.is_eagle
|
|
self.enable_kv_cache_events = params.enable_kv_cache_events
|
|
self.kv_event_queue = []
|
|
|
|
if self.token_to_kv_pool_allocator:
|
|
self.device = self.token_to_kv_pool_allocator.device
|
|
else:
|
|
self.device = torch.device("cpu")
|
|
|
|
if params.enable_metrics:
|
|
self.init_metrics_collector()
|
|
|
|
self.sliding_window_size = params.sliding_window_size
|
|
self.reset()
|
|
|
|
##### Public API #####
|
|
|
|
def supports_swa(self) -> bool:
|
|
assert (
|
|
self.sliding_window_size is not None
|
|
), "sliding_window_size must be set for SWARadixCache"
|
|
return True
|
|
|
|
def swa_reprefill_tail_tokens(self) -> int:
|
|
"""Tokens at the tail of a matched prefix that must NOT be reused.
|
|
|
|
The DeepSeek-V4 unified_kv layout keeps SWA in a per-request ring
|
|
(addressed by ``req_pool_idx * window + pos % window``), which is NOT
|
|
content-stable and is never stored in the radix tree. A reused prefix
|
|
therefore carries another request's stale SWA in the ring. Hold back the
|
|
trailing sliding window from the match so it gets re-prefilled into THIS
|
|
request's ring, making the decode window read freshly-written data.
|
|
|
|
No-op (0) for the index-addressed SWA pool, whose slots are
|
|
content-stable and safe to reuse.
|
|
"""
|
|
from sglang.srt.layers.attention.dsv4.unified_kv_kernels.env_gate import (
|
|
is_unified_kv_triton,
|
|
)
|
|
|
|
if self.sliding_window_size and is_unified_kv_triton():
|
|
return self.sliding_window_size
|
|
return 0
|
|
|
|
def reset(self) -> None:
|
|
self.root_node = TreeNode()
|
|
self.root_node.key = []
|
|
self.root_node.value = []
|
|
self.root_node.hash_value = []
|
|
self.root_node.full_lock_ref = 1
|
|
self.root_node.swa_lock_ref = 1
|
|
self.full_evictable_size_ = 0
|
|
self.swa_evictable_size_ = 0
|
|
self.full_protected_size_ = 0
|
|
self.swa_protected_size_ = 0
|
|
# LRU lists are used to maintain the order of eviction of the nodes in the tree
|
|
self.full_lru_list = LRUList(is_swa_list=False)
|
|
self.swa_lru_list = LRUList(is_swa_list=True)
|
|
self._record_all_cleared_event()
|
|
|
|
def match_prefix(self, params: MatchPrefixParams) -> MatchResult:
|
|
"""Find the matching prefix from the radix tree.
|
|
Args:
|
|
params: MatchPrefixParams containing key.
|
|
Returns:
|
|
A tuple of a tensor of matching prefix token IDs and
|
|
the last node that contains the prefix values. Note that
|
|
this API can modify the internal state of the Radix tree.
|
|
The last node create a new child if the prefix is shorter
|
|
than the last node's value.
|
|
"""
|
|
|
|
key = self._match_pre_processor(params)
|
|
if key is None:
|
|
return 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,
|
|
)
|
|
|
|
value, last_node, best_value_len = self._match_prefix_helper(key)
|
|
return self._match_post_processor(params, value, last_node, best_value_len)
|
|
|
|
def insert(self, params: InsertParams) -> InsertResult:
|
|
if self.disable:
|
|
return InsertResult(prefix_len=0)
|
|
|
|
key = params.key
|
|
value = params.value
|
|
prev_prefix_len = params.prev_prefix_len
|
|
swa_evicted_seqlen = params.swa_evicted_seqlen
|
|
|
|
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:
|
|
value = torch.tensor(key.token_ids[: len(key)], dtype=torch.int64)
|
|
|
|
prefix_len = self._insert_helper(
|
|
self.root_node, key, value, prev_prefix_len, swa_evicted_seqlen
|
|
)
|
|
return InsertResult(prefix_len=prefix_len)
|
|
|
|
def cache_finished_req(self, req: Req, is_insert: bool = True) -> None:
|
|
"""Cache request when it finishes."""
|
|
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, :kv_committed_len
|
|
]
|
|
|
|
radix_key = RadixKey(
|
|
token_ids, req.extra_key, is_bigram=self.is_eagle
|
|
).page_aligned(self.page_size)
|
|
page_aligned_len = len(radix_key)
|
|
values = kv_indices[:page_aligned_len].to(dtype=torch.int64, copy=True)
|
|
old_prefix_len = req.cache_protected_len
|
|
|
|
# Radix Cache takes one ref in memory pool
|
|
# Note: the insert function already frees the overlapped kv_indices
|
|
if is_insert:
|
|
self.insert(
|
|
InsertParams(
|
|
key=radix_key,
|
|
value=values,
|
|
prev_prefix_len=old_prefix_len,
|
|
swa_evicted_seqlen=req.swa_evicted_seqlen,
|
|
)
|
|
)
|
|
else:
|
|
self.token_to_kv_pool_allocator.free(
|
|
kv_indices[old_prefix_len:page_aligned_len]
|
|
)
|
|
|
|
# free the unaligned tail
|
|
self.token_to_kv_pool_allocator.free(kv_indices[page_aligned_len:])
|
|
|
|
# Remove req slot release the cache lock
|
|
self.dec_lock_ref(
|
|
req.last_node,
|
|
DecLockRefParams(swa_uuid_for_lock=req.swa_uuid_for_lock),
|
|
skip_swa=req.swa_prefix_lock_released,
|
|
)
|
|
req.swa_prefix_lock_released = False
|
|
|
|
def cache_unfinished_req(self, req: Req, chunked=False) -> None:
|
|
"""Cache request when it is unfinished."""
|
|
if self.disable:
|
|
kv_indices = self.req_to_token_pool.req_to_token[
|
|
req.req_pool_idx, : req.extend_range.end
|
|
]
|
|
|
|
# `req.prefix_indices` will be used in `PrefillAdder::add_chunked_req` later
|
|
req.prefix_indices = kv_indices
|
|
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)
|
|
old_prefix_len = req.cache_protected_len
|
|
|
|
# Radix Cache takes one ref in memory pool
|
|
# Note: the insert function already frees the overlapped kv_indices
|
|
result = self.insert(
|
|
InsertParams(
|
|
key=radix_key,
|
|
value=values,
|
|
prev_prefix_len=old_prefix_len,
|
|
)
|
|
)
|
|
new_prefix_len = result.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 old_prefix_len <= len(new_indices), f"{old_prefix_len=}, {new_indices=}"
|
|
assert new_prefix_len <= len(new_indices), f"{new_prefix_len=}, {new_indices=}"
|
|
self.req_to_token_pool.write(
|
|
(req.req_pool_idx, slice(old_prefix_len, len(new_indices))),
|
|
new_indices[old_prefix_len:],
|
|
)
|
|
|
|
req.cache_protected_len = len(new_indices)
|
|
|
|
self.dec_lock_ref(
|
|
req.last_node,
|
|
DecLockRefParams(swa_uuid_for_lock=req.swa_uuid_for_lock),
|
|
skip_swa=req.swa_prefix_lock_released,
|
|
)
|
|
req.swa_prefix_lock_released = False
|
|
result = self.inc_lock_ref(new_last_node)
|
|
swa_uuid_for_lock = result.swa_uuid_for_lock
|
|
|
|
# `req.prefix_indices` will be used in `PrefillAdder::add_chunked_req` later
|
|
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
|
|
req.swa_uuid_for_lock = swa_uuid_for_lock
|
|
|
|
def pretty_print(self) -> None:
|
|
self._print_helper(self.root_node, 0)
|
|
total_size, total_swa_size = self._total_size_helper()
|
|
print(f"#full_tokens: {total_size}, #swa_tokens: {total_swa_size}")
|
|
|
|
def total_size(self) -> Tuple[int, int]:
|
|
return self._total_size_helper()
|
|
|
|
def evict(self, params: EvictParams) -> EvictResult:
|
|
if self.disable:
|
|
return EvictResult()
|
|
start_time = time.perf_counter()
|
|
full_num_tokens = params.num_tokens
|
|
swa_num_tokens = params.swa_num_tokens
|
|
full_num_evicted = 0
|
|
swa_num_evicted = 0
|
|
if full_num_tokens > 0:
|
|
# get the least recently used leaf node that is not locked
|
|
x = self.full_lru_list.get_leaf_lru_no_lock()
|
|
|
|
while full_num_evicted < full_num_tokens and self.full_lru_list.in_list(x):
|
|
assert (
|
|
x != self.root_node
|
|
), f"root node should not exist in full lru list, {x.id=}"
|
|
assert x.full_lock_ref == 0, f"node is in use, {x.id=}"
|
|
|
|
# 1. free node kv indices, evict full and swa tokens
|
|
self._record_remove_event(x)
|
|
self.token_to_kv_pool_allocator.free(x.value)
|
|
full_num_evicted += len(x.value)
|
|
# Tombstoned leaves had their SWA freed earlier in `dec_swa_lock_only`
|
|
if not x.swa_tombstone:
|
|
swa_num_evicted += len(x.value)
|
|
|
|
# 2. get the next leaf, update the lru lists
|
|
x_next = self.full_lru_list.get_prev_leaf_no_lock(x)
|
|
self.full_lru_list.remove_node(x)
|
|
if not x.swa_tombstone:
|
|
self.swa_lru_list.remove_node(x)
|
|
|
|
# 3. delete the leaf node
|
|
self._delete_leaf(x)
|
|
|
|
# 4. Iteratively delete tombstone leaves to maintain invariant that leaf nodes are not tombstone
|
|
x, leaf_full_num_evicted = self._iteratively_delete_tombstone_leaf(x)
|
|
full_num_evicted += leaf_full_num_evicted
|
|
|
|
# 5. if parent has no more children, it is a leaf. It is possible that this node is lru, so
|
|
# we need to get the first leaf node in the lru list
|
|
if len(x.parent.children) == 0:
|
|
x_next = self.full_lru_list.get_leaf_lru_no_lock()
|
|
|
|
x = x_next
|
|
|
|
if swa_num_evicted < swa_num_tokens:
|
|
# get the least recently used node that is not locked, doesn't have to be a leaf
|
|
x = self.swa_lru_list.get_lru_no_lock()
|
|
|
|
# evict lru leaf nodes until swa_num_tokens is reached
|
|
while swa_num_evicted < swa_num_tokens and (self.swa_lru_list.in_list(x)):
|
|
assert not x.swa_tombstone, f"duplicate swa tombstone node, {x.id=}"
|
|
assert x != self.root_node, f"root node is not evictable, {x.id=}"
|
|
assert x.swa_lock_ref == 0, f"node is in use by swa kv indices, {x.id=}"
|
|
|
|
if len(x.children) > 0:
|
|
# 1. an internal node, free swa tokens.
|
|
self.token_to_kv_pool_allocator.free_swa(x.value)
|
|
swa_num_evicted += len(x.value)
|
|
|
|
# 2. get the next node, update the lru lists
|
|
x_next = self.swa_lru_list.get_prev_no_lock(x)
|
|
self.swa_lru_list.remove_node(x)
|
|
|
|
# 3. tombstone the node
|
|
self._tombstone_internal_node(x)
|
|
elif x.full_lock_ref > 0:
|
|
# Leaf still holds a full-side lock (can happen when the
|
|
# SWA leaf-lock early-release optimization revived a
|
|
# tombstoned leaf. Treat it like an internal tombstone.
|
|
self.token_to_kv_pool_allocator.free_swa(x.value)
|
|
swa_num_evicted += len(x.value)
|
|
|
|
x_next = self.swa_lru_list.get_prev_no_lock(x)
|
|
self.swa_lru_list.remove_node(x)
|
|
|
|
self.swa_evictable_size_ -= len(x.value)
|
|
x.swa_tombstone = True
|
|
else:
|
|
assert (
|
|
x.full_lock_ref == 0
|
|
), f"leaf node with full lock must also have swa lock, {x.id=}"
|
|
# 1. a leaf node, free full and swa tokens
|
|
self._record_remove_event(x)
|
|
self.token_to_kv_pool_allocator.free(x.value)
|
|
full_num_evicted += len(x.value)
|
|
swa_num_evicted += len(x.value)
|
|
|
|
# 2. get the next node, update the lru lists
|
|
x_next = self.swa_lru_list.get_prev_no_lock(x)
|
|
self.full_lru_list.remove_node(x)
|
|
self.swa_lru_list.remove_node(x)
|
|
|
|
# 3. delete the leaf node
|
|
self._delete_leaf(x)
|
|
|
|
# 4. Iteratively delete tombstone leaves to maintain invariant that leaf nodes are not tombstone
|
|
self._iteratively_delete_tombstone_leaf(x)
|
|
|
|
x = x_next
|
|
|
|
self.update_eviction_metrics(full_num_evicted + swa_num_evicted, start_time)
|
|
return EvictResult(
|
|
num_tokens_evicted=full_num_evicted, swa_num_tokens_evicted=swa_num_evicted
|
|
)
|
|
|
|
def inc_lock_ref(self, node: TreeNode) -> IncLockRefResult:
|
|
"""
|
|
Increment the lock reference count for the node. Returns the swa_uuid_for_lock, which needs
|
|
to be passed to dec_lock_ref.
|
|
It locks the full_lock_ref for nodes between the [last node, root), exclusive.
|
|
It locks the swa_lock_ref for nodes between the [last node, swa_uuid_for_lock], inclusive.
|
|
"""
|
|
if self.disable:
|
|
return IncLockRefResult()
|
|
|
|
swa_lock_size = 0
|
|
swa_uuid_for_lock = None
|
|
while node != self.root_node:
|
|
# lock full from node to root
|
|
assert (
|
|
node.full_lock_ref >= 0
|
|
), f"inc_lock_ref on node with {node.full_lock_ref=}, {node.id=}"
|
|
if node.full_lock_ref == 0:
|
|
self.full_evictable_size_ -= len(node.value)
|
|
self.full_protected_size_ += len(node.value)
|
|
node.full_lock_ref += 1
|
|
|
|
# lock swa if we have not reached the sliding window size.
|
|
# When we reach the sliding window size, we will set the swa_uuid_for_lock.
|
|
# caller needs to pass the swa_uuid_for_lock to dec_lock_ref
|
|
if swa_lock_size < self.sliding_window_size:
|
|
assert (
|
|
not node.swa_tombstone
|
|
), f"inc_lock_swa on swa_tombstone node, {node.id=}"
|
|
if node.swa_lock_ref == 0:
|
|
self.swa_evictable_size_ -= len(node.value)
|
|
self.swa_protected_size_ += len(node.value)
|
|
node.swa_lock_ref += 1
|
|
swa_lock_size += len(node.value)
|
|
if swa_lock_size >= self.sliding_window_size:
|
|
if node.swa_uuid is None:
|
|
node.swa_uuid = gen_swa_uuid()
|
|
swa_uuid_for_lock = node.swa_uuid
|
|
node = node.parent
|
|
return IncLockRefResult(swa_uuid_for_lock=swa_uuid_for_lock)
|
|
|
|
def dec_lock_ref(
|
|
self,
|
|
node: TreeNode,
|
|
params: Optional[DecLockRefParams] = None,
|
|
skip_swa: bool = False,
|
|
) -> DecLockRefResult:
|
|
"""
|
|
Decrement the lock reference count for the node.
|
|
It unlocks the full_lock_ref for nodes between the [last node, root), exclusive.
|
|
It unlocks the swa_lock_ref for nodes between the [last node, swa_uuid_for_lock], inclusive.
|
|
If swa_uuid_for_lock is None, it unlocks to the root, exclusive.
|
|
|
|
If skip_swa is True, only the full_lock_ref is decremented; the SWA lock is
|
|
assumed to have been released already (e.g. via `dec_swa_lock_only`).
|
|
"""
|
|
swa_uuid_for_lock = params.swa_uuid_for_lock if params is not None else None
|
|
|
|
if self.disable:
|
|
return DecLockRefResult()
|
|
|
|
dec_lock_swa = not skip_swa
|
|
while node != self.root_node:
|
|
assert (
|
|
node.full_lock_ref > 0
|
|
), f"dec_lock_ref on node with {node.full_lock_ref=}, {node.id=}"
|
|
if node.full_lock_ref == 1:
|
|
self.full_evictable_size_ += len(node.value)
|
|
self.full_protected_size_ -= len(node.value)
|
|
node.full_lock_ref -= 1
|
|
|
|
if dec_lock_swa:
|
|
assert (
|
|
not node.swa_tombstone
|
|
), f"dec_lock_ref on swa_tombstone node, {node.id=}"
|
|
assert (
|
|
node.swa_lock_ref > 0
|
|
), f"dec_lock_ref on node with {node.swa_lock_ref=}, {node.id=}"
|
|
|
|
if node.swa_lock_ref == 1:
|
|
self.swa_evictable_size_ += len(node.value)
|
|
self.swa_protected_size_ -= len(node.value)
|
|
node.swa_lock_ref -= 1
|
|
if swa_uuid_for_lock and node.swa_uuid == swa_uuid_for_lock:
|
|
dec_lock_swa = False
|
|
|
|
node = node.parent
|
|
|
|
return DecLockRefResult()
|
|
|
|
def dec_swa_lock_only(
|
|
self, node: TreeNode, swa_uuid_for_lock: Optional[int] = None
|
|
):
|
|
"""
|
|
Decrement only the swa_lock_ref (and swa_protected_size_) along the chain
|
|
[node, swa_uuid_for_lock], inclusive. The full_lock_ref is left untouched
|
|
so the caller's full-cache protection is preserved.
|
|
|
|
Used to early-release the SWA portion of a request's tree lock once the
|
|
request's decode position has advanced past the sliding window, so the
|
|
protected window can be reclaimed.
|
|
|
|
For internal nodes, the standard protected -> evictable transition is
|
|
applied (node stays in swa_lru_list and may be evicted by SWA LRU later).
|
|
For leaf nodes, since `swa_lru_list` cannot contain a leaf with
|
|
`full_lock_ref > 0` (SWA-eviction would also delete the still-referenced
|
|
leaf), we instead free the SWA pool slots immediately and mark the leaf
|
|
as `swa_tombstone=True`. The full kv stays alive until the full-side
|
|
lock drops; future prefix-matches stop before this tombstoned leaf.
|
|
|
|
Caller must ensure this is invoked at most once per (node, swa_uuid_for_lock)
|
|
pair (track via e.g. `Req.swa_prefix_lock_released`). When the request
|
|
finally releases its full lock via `dec_lock_ref`, pass `skip_swa=True`
|
|
to avoid touching SWA state again.
|
|
"""
|
|
if self.disable:
|
|
return
|
|
|
|
while node != self.root_node:
|
|
assert (
|
|
not node.swa_tombstone
|
|
), f"dec_swa_lock_only on swa_tombstone node, {node.id=}"
|
|
assert (
|
|
node.swa_lock_ref > 0
|
|
), f"dec_swa_lock_only on node with {node.swa_lock_ref=}, {node.id=}"
|
|
|
|
if node.swa_lock_ref == 1:
|
|
self.swa_protected_size_ -= len(node.value)
|
|
if len(node.children) == 0:
|
|
# Leaf: free SWA pool slots and tombstone, and remove from
|
|
# swa_lru_list so SWA-eviction won't pick this tombstoned
|
|
# leaf (which still holds full_lock_ref > 0). The full kv
|
|
# stays alive until the request releases its full lock.
|
|
self.token_to_kv_pool_allocator.free_swa(node.value)
|
|
self.swa_lru_list.remove_node(node)
|
|
node.swa_tombstone = True
|
|
else:
|
|
# Internal: standard protected -> evictable.
|
|
self.swa_evictable_size_ += len(node.value)
|
|
node.swa_lock_ref -= 1
|
|
|
|
if swa_uuid_for_lock and node.swa_uuid == swa_uuid_for_lock:
|
|
break
|
|
node = node.parent
|
|
|
|
def sanity_check(self):
|
|
self.full_lru_list.sanity_check(self)
|
|
self.swa_lru_list.sanity_check(self)
|
|
|
|
def evictable_size(self) -> Tuple[int, int]:
|
|
# Note: use full_evictable_size() and swa_evictable_size() instead.
|
|
raise NotImplementedError
|
|
|
|
def full_evictable_size(self) -> int:
|
|
return self.full_evictable_size_
|
|
|
|
def swa_evictable_size(self) -> int:
|
|
return self.swa_evictable_size_
|
|
|
|
def protected_size(self) -> Tuple[int, int]:
|
|
# Note: use full_protected_size() and swa_protected_size() instead.
|
|
raise NotImplementedError
|
|
|
|
def full_protected_size(self) -> int:
|
|
# protected size refers to the size of the full cache that is locked
|
|
return self.full_protected_size_
|
|
|
|
def swa_protected_size(self) -> int:
|
|
# protected size refers to the size of the swa cache that is locked
|
|
return self.swa_protected_size_
|
|
|
|
def all_values_flatten(self) -> torch.Tensor:
|
|
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)
|
|
|
|
def available_and_evictable_str(self) -> str:
|
|
full_available_size = self.token_to_kv_pool_allocator.full_available_size()
|
|
swa_available_size = self.token_to_kv_pool_allocator.swa_available_size()
|
|
full_evictable_size = self.full_evictable_size()
|
|
swa_evictable_size = self.swa_evictable_size()
|
|
return (
|
|
f"Available full tokens: {full_available_size + full_evictable_size} ({full_available_size=} + {full_evictable_size=})\n"
|
|
f"Available swa tokens: {swa_available_size + swa_evictable_size} ({swa_available_size=} + {swa_evictable_size=})\n"
|
|
f"Full LRU list evictable size: {self.full_lru_list.sanity_check_evictable_size()}\n"
|
|
f"SWA LRU list evictable size: {self.swa_lru_list.sanity_check_evictable_size()}\n"
|
|
)
|
|
|
|
##### Internal Helper Functions #####
|
|
|
|
def _match_prefix_helper(
|
|
self, key: RadixKey
|
|
) -> Tuple[List[torch.Tensor], TreeNode, int]:
|
|
"""
|
|
SWA prefix matching helper. It factors in the sliding window size such that
|
|
the matched node is guaranteed to either 1. connected to root without swa tombstone,
|
|
or 2. the number of matching tokens from the matched node to the last swa tombstone
|
|
node is greater than or equal to the sliding window size.
|
|
"""
|
|
node = self.root_node
|
|
child_key = key.child_key(self.page_size)
|
|
|
|
value = []
|
|
# for path connected to root without tombstone, always match, so set to inf
|
|
match_len_since_tombstone = float("inf")
|
|
best_value_len = 0
|
|
best_last_node = node
|
|
enable_compact = envs.SGLANG_OPT_SWA_RADIX_CACHE_COMPACT.get()
|
|
while len(key) > 0 and child_key in node.children.keys():
|
|
child = node.children[child_key]
|
|
|
|
if enable_compact:
|
|
self._compact_single_child_chain(child)
|
|
|
|
if child.swa_tombstone:
|
|
# update best_value_len and best_last_node if needed
|
|
if match_len_since_tombstone >= self.sliding_window_size:
|
|
best_value_len = len(value)
|
|
best_last_node = node
|
|
# reset match_len_since_tombstone if we hit a tombstone node
|
|
match_len_since_tombstone = 0
|
|
|
|
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)
|
|
if not new_node.swa_tombstone:
|
|
match_len_since_tombstone += len(new_node.value)
|
|
node = new_node
|
|
break
|
|
else:
|
|
value.append(child.value)
|
|
if not child.swa_tombstone:
|
|
match_len_since_tombstone += len(child.value)
|
|
node = child
|
|
key = key[prefix_len:]
|
|
|
|
if len(key):
|
|
child_key = key.child_key(self.page_size)
|
|
|
|
# handle best_value_len and best_last_node, for the case that last node is fully matched
|
|
if match_len_since_tombstone >= self.sliding_window_size:
|
|
best_value_len = len(value)
|
|
best_last_node = node
|
|
|
|
return value, best_last_node, best_value_len
|
|
|
|
def _match_pre_processor(self, params: MatchPrefixParams) -> Optional[RadixKey]:
|
|
"""Preprocess the key before matching."""
|
|
key = params.key
|
|
key, _ = key.maybe_to_bigram_view(self.is_eagle)
|
|
if self.disable or len(key) == 0:
|
|
return None
|
|
key = key.page_aligned(self.page_size)
|
|
if len(key) == 0:
|
|
return None
|
|
return key
|
|
|
|
def _match_post_processor(
|
|
self,
|
|
params: MatchPrefixParams,
|
|
value: List[torch.Tensor],
|
|
last_node: TreeNode,
|
|
best_value_len: int,
|
|
) -> MatchResult:
|
|
"""Post-process the matched result."""
|
|
node_update = last_node
|
|
# update time for matched nodes, and make nodes closer to root to be least recently used
|
|
# this allows swa to evict nodes closer to root first
|
|
self.full_lru_list.reset_node_and_parents_mru(node_update, self.root_node)
|
|
self.swa_lru_list.reset_node_and_parents_mru(node_update, self.root_node)
|
|
|
|
# This last_access_time is for sanity check, can be deleted after validation in production
|
|
cur_time = get_last_access_time()
|
|
while node_update:
|
|
node_update.last_access_time = cur_time
|
|
cur_time -= (
|
|
0.00001 # assuming less than 100000 nodes in a branch of the tree
|
|
)
|
|
node_update = node_update.parent
|
|
|
|
value = value[:best_value_len]
|
|
if value:
|
|
value = torch.cat(value)
|
|
else:
|
|
value = torch.empty((0,), dtype=torch.int64, device=self.device)
|
|
|
|
return MatchResult(
|
|
device_indices=value,
|
|
last_device_node=last_node,
|
|
last_host_node=last_node,
|
|
best_match_node=last_node,
|
|
)
|
|
|
|
def _compact_single_child_chain(self, node: TreeNode) -> None:
|
|
# FIXME(ispobock): drifts retract pool accounting (commit 6348cb506);
|
|
# also overwrites active swa_uuid when window > page_size. Off by
|
|
# default via SGLANG_OPT_SWA_RADIX_CACHE_COMPACT.
|
|
while len(node.children) == 1:
|
|
child = next(iter(node.children.values()))
|
|
if len(child.children) == 0:
|
|
break
|
|
sum_gc_full_lock_ref = sum(
|
|
gc.full_lock_ref for gc in child.children.values()
|
|
)
|
|
if child.full_lock_ref > sum_gc_full_lock_ref:
|
|
break
|
|
if (
|
|
child.swa_tombstone != node.swa_tombstone
|
|
or child.full_lock_ref != node.full_lock_ref
|
|
or child.swa_lock_ref != node.swa_lock_ref
|
|
):
|
|
break
|
|
|
|
# Preserve is_bigram: main #23106 made bigram an O(1) flag on RadixKey;
|
|
# the constructor defaults to False, so concat without explicit flag
|
|
# silently demotes EAGLE/MTP bigram keys → match() returns 0 →
|
|
# _split_node assert.
|
|
node.key = RadixKey(
|
|
node.key.token_ids + child.key.token_ids,
|
|
node.key.extra_key,
|
|
is_bigram=node.key.is_bigram,
|
|
)
|
|
node.value = torch.cat([node.value, child.value])
|
|
node.children = child.children
|
|
for grandchild in node.children.values():
|
|
grandchild.parent = node
|
|
|
|
if child.swa_uuid is not None:
|
|
node.swa_uuid = child.swa_uuid
|
|
|
|
if node.hash_value is not None and child.hash_value is not None:
|
|
node.hash_value = list(node.hash_value) + list(child.hash_value)
|
|
else:
|
|
node.hash_value = None
|
|
|
|
self.full_lru_list.remove_node(child)
|
|
if not child.swa_tombstone:
|
|
self.swa_lru_list.remove_node(child)
|
|
|
|
def _maybe_split_leaf_for_swa_lock(self, leaf: TreeNode) -> TreeNode:
|
|
"""``inc_lock_ref`` protects ``len(leaf.value)`` SWA tokens for the
|
|
leaf even though SWA only actually needs the last
|
|
``sliding_window_size`` tokens. With chunked prefill, leaves can be
|
|
thousands of tokens long, which inflates ``swa_protected_size_`` by
|
|
~``chunked_prefill_size / sliding_window_size`` and causes premature
|
|
SWA pool exhaustion / retract thrashing.
|
|
"""
|
|
if (
|
|
leaf is self.root_node
|
|
or leaf.swa_lock_ref > 0
|
|
or leaf.swa_tombstone
|
|
or len(leaf.value) == 0
|
|
):
|
|
return leaf
|
|
|
|
# Smallest page-aligned size that still covers the sliding window.
|
|
tail_size = (
|
|
(self.sliding_window_size + self.page_size - 1)
|
|
// self.page_size
|
|
* self.page_size
|
|
)
|
|
if len(leaf.value) <= tail_size:
|
|
return leaf
|
|
|
|
split_at = len(leaf.value) - tail_size
|
|
|
|
if split_at <= 0 or split_at >= len(leaf.value):
|
|
return leaf
|
|
if self.page_size > 1 and (
|
|
split_at % self.page_size != 0 or len(leaf.value) % self.page_size != 0
|
|
):
|
|
return leaf
|
|
|
|
self._split_node(leaf.key, leaf, split_at)
|
|
return leaf
|
|
|
|
def _split_node(self, key: RadixKey, child: TreeNode, split_len: int) -> TreeNode:
|
|
# new_node -> child
|
|
new_node = TreeNode()
|
|
new_node.children = {key[split_len:].child_key(self.page_size): child}
|
|
new_node.parent = child.parent
|
|
new_node.swa_tombstone = child.swa_tombstone
|
|
new_node.full_lock_ref = child.full_lock_ref
|
|
new_node.swa_lock_ref = child.swa_lock_ref
|
|
new_node.key = child.key[:split_len]
|
|
assert len(new_node.key) > 0, f"new_node.key should not be empty"
|
|
new_node.value = child.value[:split_len].clone()
|
|
# parent inherits the swa_uuid from child for swa lock ref
|
|
new_node.swa_uuid = child.swa_uuid
|
|
child.swa_uuid = None
|
|
# child time should be later than parent's time for swa tombstone
|
|
child.last_access_time = get_last_access_time()
|
|
|
|
# remove the child from the lru lists because it is being split
|
|
self.full_lru_list.remove_node(child)
|
|
if not new_node.swa_tombstone:
|
|
self.swa_lru_list.remove_node(child)
|
|
child.parent = new_node
|
|
child.key = child.key[split_len:]
|
|
assert len(child.key) > 0, f"child.key should not be empty"
|
|
child.value = child.value[split_len:].clone()
|
|
new_node.parent.children[key.child_key(self.page_size)] = new_node
|
|
new_node.hash_value, child.hash_value = split_node_hash_value(
|
|
child.hash_value, split_len, self.page_size
|
|
)
|
|
|
|
# insert the new node and child into the lru lists, insert
|
|
# parent first so that parent is after child in the lru list
|
|
self.full_lru_list.insert_mru(new_node)
|
|
self.full_lru_list.insert_mru(child)
|
|
if not new_node.swa_tombstone:
|
|
self.swa_lru_list.insert_mru(new_node)
|
|
self.swa_lru_list.insert_mru(child)
|
|
return new_node
|
|
|
|
def _insert_helper(
|
|
self,
|
|
node: TreeNode,
|
|
key: RadixKey,
|
|
value,
|
|
update_kv_after_len: int,
|
|
swa_evicted_seqlen: int = 0,
|
|
) -> int:
|
|
# Update the last access time from root to leaf, so that
|
|
# swa will tombstone the node closer to root first
|
|
node.last_access_time = get_last_access_time()
|
|
if node != self.root_node:
|
|
self.full_lru_list.reset_node_mru(node)
|
|
if not node.swa_tombstone:
|
|
self.swa_lru_list.reset_node_mru(node)
|
|
if len(key) == 0:
|
|
return 0
|
|
|
|
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 = get_last_access_time()
|
|
self.full_lru_list.reset_node_mru(node)
|
|
if not node.swa_tombstone:
|
|
self.swa_lru_list.reset_node_mru(node)
|
|
prefix_len = node.key.match(key, page_size=self.page_size)
|
|
|
|
if prefix_len < len(node.key):
|
|
new_node = self._split_node(node.key, node, prefix_len)
|
|
node = new_node
|
|
|
|
# if tombstone after update_kv_after_len, update node.value to be the input value.
|
|
# This is needed because it is possible that the last sliding window size tokens
|
|
# contains tombstone. If this is the case and we don't update the kv value, then
|
|
# the prefill prefix matching will stuck.
|
|
if update_kv_after_len < total_prefix_length + prefix_len:
|
|
# For page_size > 1 and chunked prefill case, update_kv_after_len may be not page-aligned due to a trailing partial page
|
|
# (kept in the request but not inserted into the radix tree) appended to prefix_indices.
|
|
if node.swa_tombstone:
|
|
assert (
|
|
node.swa_lock_ref == 0
|
|
), f"tombstone swa_lock_ref should always be 0, {node.full_lock_ref=}, {node.swa_lock_ref=}, {node.id=}"
|
|
assert (
|
|
swa_evicted_seqlen % self.page_size == 0
|
|
), f"swa_evicted_seqlen must be page aligned, {swa_evicted_seqlen=}, {self.page_size=}"
|
|
if swa_evicted_seqlen <= total_prefix_length:
|
|
# Branch 1: all swa tokens of value[:prefix_len] are not evicted, so we can insert it to the tree directly.
|
|
if node.full_lock_ref > 0:
|
|
# Full KV is still locked by a running request. Keep it
|
|
# and adopt the incoming SWA instead of freeing in-flight
|
|
# Full slots.
|
|
self._recover_tombstone_keeping_locked_full(
|
|
node, value[:prefix_len]
|
|
)
|
|
else:
|
|
# Free full tokens in the original tree node.
|
|
self.token_to_kv_pool_allocator.free(
|
|
node.value[:prefix_len]
|
|
)
|
|
# Overwrite the new value in request to the tree node.
|
|
node.value = value[:prefix_len].clone()
|
|
node.swa_tombstone = False
|
|
self.swa_lru_list.insert_mru(node)
|
|
self.swa_evictable_size_ += len(node.value)
|
|
elif swa_evicted_seqlen < total_prefix_length + prefix_len:
|
|
# Branch 2: part of swa tokens of value[:prefix_len] are evicted, so we need to split the node and insert the value to new node.
|
|
start_update_idx = swa_evicted_seqlen - total_prefix_length
|
|
if node.full_lock_ref > 0:
|
|
# Split first so the recovered suffix keeps the locked
|
|
# Full slots, then adopt the incoming SWA for that suffix.
|
|
self._split_node(node.key, node, start_update_idx)
|
|
self._recover_tombstone_keeping_locked_full(
|
|
node, value[start_update_idx:prefix_len]
|
|
)
|
|
self.token_to_kv_pool_allocator.free(
|
|
value[:start_update_idx]
|
|
)
|
|
else:
|
|
self.token_to_kv_pool_allocator.free(
|
|
node.value[start_update_idx:prefix_len]
|
|
)
|
|
self._split_node(node.key, node, start_update_idx)
|
|
# Here node is the new node after split, so we can overwrite the value to the new node.
|
|
# The old node is still swa tombstone and the full token is not freed.
|
|
node.value = value[start_update_idx:prefix_len].clone()
|
|
self.token_to_kv_pool_allocator.free(
|
|
value[:start_update_idx]
|
|
)
|
|
node.swa_tombstone = False
|
|
self.swa_lru_list.insert_mru(node)
|
|
self.swa_evictable_size_ += len(node.value)
|
|
else:
|
|
# Branch 3: all swa tokens of value[:prefix_len] are evicted, so we don't need to update the node.
|
|
self.token_to_kv_pool_allocator.free(value[:prefix_len])
|
|
else:
|
|
# The node is not tombstone, so we don't need to update the node.
|
|
self.token_to_kv_pool_allocator.free(value[:prefix_len])
|
|
|
|
total_prefix_length += prefix_len
|
|
key = key[prefix_len:]
|
|
value = value[prefix_len:]
|
|
|
|
if len(key):
|
|
child_key = key.child_key(self.page_size)
|
|
|
|
if len(key):
|
|
# Layout: |--- total_prefix_length ---|--- len(key) ---|
|
|
# ^ ^ ^
|
|
# 0 total_prefix_length total_length
|
|
#
|
|
# Cases based on swa_evicted_seqlen position:
|
|
# 1. swa_evicted_seqlen <= total_prefix_length:
|
|
# Already handled in the while loop above. All of len(key) is non-tombstone.
|
|
# 2. total_prefix_length < swa_evicted_seqlen < total_length:
|
|
# Split: [total_prefix_length, swa_evicted_seqlen) as tombstone,
|
|
# [swa_evicted_seqlen, total_length) as non-tombstone.
|
|
# 3. swa_evicted_seqlen == total_length:
|
|
# All remaining tokens are evicted. Free value and return without
|
|
# creating a node (leaf nodes must not be tombstone).
|
|
# Note: the -page_size fix in _evict_swa prevents this case from
|
|
# occurring in normal operation. This check is a defensive guard
|
|
# against unexpected eviction states from other code paths.
|
|
if swa_evicted_seqlen == total_prefix_length + len(key):
|
|
self.token_to_kv_pool_allocator.free(value)
|
|
return total_prefix_length
|
|
|
|
if (
|
|
swa_evicted_seqlen > total_prefix_length
|
|
and swa_evicted_seqlen < total_prefix_length + len(key)
|
|
):
|
|
swa_tombstone_len = swa_evicted_seqlen - total_prefix_length
|
|
node = self._add_new_node(
|
|
node,
|
|
key[:swa_tombstone_len],
|
|
value[:swa_tombstone_len],
|
|
swa_tombstone=True,
|
|
)
|
|
key = key[swa_tombstone_len:]
|
|
value = value[swa_tombstone_len:]
|
|
|
|
new_leaf = self._add_new_node(node, key, value, swa_tombstone=False)
|
|
|
|
if envs.SGLANG_OPT_SWA_SPLIT_LEAF_ON_INSERT.get():
|
|
# Cap the leaf at one (page-aligned) sliding window so a future
|
|
# inc_lock_ref only protects `sliding_window_size` tokens of SWA pool.
|
|
self._maybe_split_leaf_for_swa_lock(new_leaf)
|
|
|
|
return total_prefix_length
|
|
|
|
def _recover_tombstone_keeping_locked_full(
|
|
self, node: TreeNode, incoming_full: torch.Tensor
|
|
) -> None:
|
|
"""Recover a tombstoned node whose Full KV is locked by a running request.
|
|
|
|
Keep node.value, the locked Full slots, and re-point its full->SWA
|
|
mapping at the incoming request's fresh SWA. Free only the incoming
|
|
redundant Full slots, not their SWA slots.
|
|
"""
|
|
assert len(node.value) == len(incoming_full), (
|
|
f"locked-full recover size mismatch: {len(node.value)=}, "
|
|
f"{len(incoming_full)=}"
|
|
)
|
|
|
|
allocator = self.token_to_kv_pool_allocator
|
|
swa_value = allocator.translate_loc_from_full_to_swa(incoming_full)
|
|
allocator.set_full_to_swa_mapping(node.value, swa_value)
|
|
allocator.full_to_swa_index_mapping[incoming_full.to(torch.int64)] = 0
|
|
allocator.full_attn_allocator.free(incoming_full)
|
|
|
|
node.swa_tombstone = False
|
|
self.swa_lru_list.insert_mru(node)
|
|
self.swa_evictable_size_ += len(node.value)
|
|
|
|
def _add_new_node(
|
|
self,
|
|
parent: TreeNode,
|
|
key: RadixKey,
|
|
value: torch.Tensor,
|
|
swa_tombstone: bool = False,
|
|
) -> TreeNode:
|
|
assert len(key) > 0, f"key should not be empty"
|
|
new_node = TreeNode()
|
|
new_node.parent = parent
|
|
new_node.key = key
|
|
new_node.value = value.clone()
|
|
new_node.swa_tombstone = swa_tombstone
|
|
parent.children[key.child_key(self.page_size)] = new_node
|
|
self.full_lru_list.insert_mru(new_node)
|
|
self.full_evictable_size_ += len(value)
|
|
if not swa_tombstone:
|
|
self.swa_lru_list.insert_mru(new_node)
|
|
self.swa_evictable_size_ += len(value)
|
|
self._record_store_event(new_node)
|
|
return new_node
|
|
|
|
def _iteratively_delete_tombstone_leaf(
|
|
self, node: TreeNode
|
|
) -> Tuple[TreeNode, int]:
|
|
full_num_evicted = 0
|
|
while node.parent.swa_tombstone and len(node.parent.children) == 0:
|
|
# root node is not evictable
|
|
if node.parent == self.root_node:
|
|
break
|
|
# if locked, means node is in use, skip
|
|
if node.parent.full_lock_ref > 0:
|
|
break
|
|
assert (
|
|
node.parent.swa_lock_ref == 0
|
|
), f"tombstone swa_lock_ref should always be 0, {node.parent.full_lock_ref=}, {node.parent.swa_lock_ref=}, {node.parent.id=}"
|
|
# delete tombstone node evicts full tokens
|
|
self._record_remove_event(node.parent)
|
|
self.token_to_kv_pool_allocator.free(node.parent.value)
|
|
full_num_evicted += len(node.parent.value)
|
|
self.full_lru_list.remove_node(node.parent)
|
|
self._delete_tombstone_leaf(node.parent)
|
|
node = node.parent
|
|
|
|
return node, full_num_evicted
|
|
|
|
def _delete_leaf(self, node: TreeNode) -> None:
|
|
assert len(node.children) == 0, f"leaf node has children, {node.id=}"
|
|
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.full_evictable_size_ -= len(node.key)
|
|
# Tombstoned leaves were never (re-)added to swa_lru_list and were
|
|
# already removed from swa_evictable_size_ when they were tombstoned.
|
|
if not node.swa_tombstone:
|
|
self.swa_evictable_size_ -= len(node.key)
|
|
|
|
def _tombstone_internal_node(self, node: TreeNode) -> None:
|
|
assert len(node.children) != 0, f"Cannot tombstone a leaf node, {node.id=}"
|
|
node.swa_tombstone = True
|
|
self.swa_evictable_size_ -= len(node.key)
|
|
|
|
def _delete_tombstone_leaf(self, node: TreeNode) -> None:
|
|
assert (
|
|
node.swa_tombstone
|
|
), f"Deleting a unexpected non-tombstone leaf node, {node.id=}"
|
|
assert len(node.children) == 0, f"leaf node has children, {node.id=}"
|
|
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.full_evictable_size_ -= len(node.key)
|
|
|
|
def _collect_nontombstone_nodes(self) -> List[TreeNode]:
|
|
ret_list = []
|
|
stack = [self.root_node]
|
|
|
|
while stack:
|
|
cur_node = stack.pop()
|
|
if not cur_node.swa_tombstone:
|
|
ret_list.append(cur_node)
|
|
stack.extend(cur_node.children.values())
|
|
|
|
return ret_list
|
|
|
|
def _collect_all_nodes(self) -> List[TreeNode]:
|
|
ret_list = []
|
|
stack = [self.root_node]
|
|
while stack:
|
|
cur_node = stack.pop()
|
|
ret_list.append(cur_node)
|
|
stack.extend(cur_node.children.values())
|
|
return ret_list
|
|
|
|
def _print_helper(self, node: TreeNode, indent: int) -> None:
|
|
"""Prints the radix tree in a human-readable format."""
|
|
stack = [(node, indent)]
|
|
while stack:
|
|
current_node, current_indent = stack.pop()
|
|
print(
|
|
" " * current_indent,
|
|
current_node.id,
|
|
len(current_node.key),
|
|
f"fr={current_node.full_lock_ref}",
|
|
f"sr={current_node.swa_lock_ref}",
|
|
f"fll={self.full_lru_list.in_list(current_node)}",
|
|
f"sll={self.swa_lru_list.in_list(current_node)}",
|
|
f"ts={current_node.swa_tombstone}",
|
|
)
|
|
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 _total_size_helper(self) -> Tuple[int, int]:
|
|
total_size = 0
|
|
total_swa_size = 0
|
|
stack = [self.root_node]
|
|
while stack:
|
|
current_node = stack.pop()
|
|
total_size += len(current_node.value)
|
|
if not current_node.swa_tombstone:
|
|
total_swa_size += len(current_node.value)
|
|
for child in current_node.children.values():
|
|
if child.evicted:
|
|
continue
|
|
stack.append(child)
|
|
return total_size, total_swa_size
|