chenchenghao/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Remove temporary CP cache-hit trace instrumentation

Browse Source 
The GSM8K/cache-hit debugging pass added request-correlation logs across scheduler, radix, HiCache, Mooncake, and prefill handoff. The root cause work has moved to request-slot remap semantics, so those high-cardinality traces are no longer needed in the runtime diff.

This removes the temporary tracing helpers and call sites while leaving existing fail-fast checks, fallback warnings, and explicit debug/timing infrastructure that is still part of normal CP shared-KV diagnostics.

Constraint: Production CP hot paths should not carry investigation-only request signatures or transfer summaries
Rejected: Keep all debug logs gated by env | even gated logs increase maintenance surface and encourage stale diagnosis paths
Confidence: high
Scope-risk: moderate
Directive: Reintroduce request-correlation logs only as a narrow opt-in probe with a planned removal point
Tested: Local py_compile for touched runtime files; remote py_compile; remote pytest test_cp_shared_kv_runtime.py, test_nsa_cp_utils.py, test_cp_shared_kv_layout.py => 263 passed, 5 warnings, 2 subtests passed
Not-tested: Long-running ETE log-volume comparison after removal
This commit is contained in:
laoyao0822
2026-06-08 20:57:08 +08:00
parent 3698b0c22c
commit a32c639147
7 changed files with 5 additions and 370 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
python/sglang/srt/disaggregation/mooncake/conn.py
View File

@@ -64,21 +64,6 @@ def _cp_draft_shared_kv_debug(message: str, *args, limit: int = 64) -> None:
logger.info("[CP_DRAFT_SHARED_KV] " + message, *args)
def _cp_shared_kv_bs_gt1_mooncake_debug(
key: str, message: str, *args, limit: Optional[int] = None
) -> None:
if not envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG.get():
return
if limit is None:
limit = int(envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG_LIMIT.get())
count_key = "bs_gt1:" + key
count = _CP_SHARED_DEBUG_COUNTS.get(count_key, 0)
if limit > 0 and count >= limit:
return
_CP_SHARED_DEBUG_COUNTS[count_key] = count + 1
logger.info("[CP_SHARED_KV_BS_GT1_DEBUG] event=%s " + message, key, *args)
def _mooncake_transfer_stats_enabled() -> bool:
return envs.SGLANG_DISAGGREGATION_TRANSFER_STATS.get()
@@ -1226,19 +1211,6 @@ class MooncakeKVManager(CommonKVManager):
_np_summary(chunked_dst_kv_indice),
kv_chunk.is_last_chunk,
)
_cp_shared_kv_bs_gt1_mooncake_debug(
"transfer_worker_kv",
"cp_rank=%s room=%s prefill_pages=%s "
"logical_positions=%s dst_pages=%s is_last=%s "
"session=%s",
self.attn_cp_rank,
kv_chunk.room,
_np_summary(kv_chunk.prefill_kv_indices),
_np_summary(kv_chunk.logical_page_positions),
_np_summary(chunked_dst_kv_indice),
kv_chunk.is_last_chunk,
req.mooncake_session_id,
)
if envs.SGLANG_DEBUG_CP_SHARED_KV.get():
_cp_shared_debug_log(
"transfer_worker_kv",
@@ -1354,18 +1326,6 @@ class MooncakeKVManager(CommonKVManager):
),
_np_summary(dst_state_indices),
)
_cp_shared_kv_bs_gt1_mooncake_debug(
"transfer_worker_state",
"cp_rank=%s room=%s prefill_state_pages=%s "
"state_positions=%s dst_state_pages=%s "
"session=%s",
self.attn_cp_rank,
kv_chunk.room,
_np_summary(kv_chunk.state_indices),
_np_summary(kv_chunk.state_logical_page_positions),
_np_summary(dst_state_indices),
req.mooncake_session_id,
)
self.maybe_send_extra(
req,
kv_chunk.state_indices,
@@ -1746,22 +1706,6 @@ class MooncakeKVSender(CommonKVSender):
is_last_chunk,
getattr(self.kv_mgr.kv_args, "draft_kv_buffer_count", None),
)
_cp_shared_kv_bs_gt1_mooncake_debug(
"sender_filter",
"cp_rank=%s room=%s page_start=%s orig_kv_pages=%s "
"filtered_kv_pages=%s kv_positions=%s orig_state_pages=%s "
"filtered_state_pages=%s state_positions=%s is_last=%s",
self.kv_mgr.attn_cp_rank,
self.bootstrap_room,
chunk_page_start,
_np_summary(orig_kv_indices),
_np_summary(kv_indices),
_np_summary(logical_page_positions),
_np_summary(orig_state_indices),
_np_summary(state_indices),
_np_summary(state_logical_page_positions),
is_last_chunk,
)
# Special handling for cp
elif self.kv_mgr.enable_all_cp_ranks_for_transfer:
kv_indices, index_slice = filter_kv_indices_for_cp_rank(

5
python/sglang/srt/disaggregation/prefill.py
View File

@@ -43,7 +43,6 @@ from sglang.srt.disaggregation.utils import (
kv_to_page_num,
poll_and_all_reduce_attn_cp_tp_group,
prepare_abort,
token_ids_debug_signature,
)
from sglang.srt.managers.schedule_batch import (
FINISH_ABORT,
@@ -1305,7 +1304,7 @@ class SchedulerDisaggregationPrefillMixin:
"rid=%s room=%s start_idx=%s end_idx=%s last_chunk=%s "
"page_size=%s pages=%s state_pages=%s prefix_len=%s "
"host_hit_length=%s extend_input_len=%s fill_len=%s "
"origin_input_len=%s has_draft_pool=%s origin_sig=%s fill_sig=%s",
"origin_input_len=%s has_draft_pool=%s",
req.rid,
req.bootstrap_room,
start_idx,
@@ -1320,8 +1319,6 @@ class SchedulerDisaggregationPrefillMixin:
len(req.fill_ids),
len(req.origin_input_ids),
has_draft_pool,
token_ids_debug_signature(getattr(req, "origin_input_ids", None)),
token_ids_debug_signature(getattr(req, "fill_ids", None)),
)
if has_draft_pool and draft_prefix_overlap > 0:
_cp_draft_shared_kv_debug(

25
python/sglang/srt/disaggregation/utils.py
View File

@@ -67,31 +67,6 @@ def _cp_shared_kv_poll_debug_enabled() -> bool:
return envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG.get()
def token_ids_debug_signature(token_ids: Any, *, sample: int = 6) -> str:
"""Return a stable, bounded signature for request-token correlation logs."""
if token_ids is None:
return "None"
if isinstance(token_ids, torch.Tensor):
values = token_ids.detach().reshape(-1).cpu().tolist()
elif isinstance(token_ids, np.ndarray):
values = token_ids.reshape(-1).tolist()
else:
values = list(token_ids)
mask = (1 << 63) - 1
digest = 1469598103934665603
for value in values:
item = int(value) & 0xFFFFFFFFFFFFFFFF
for shift in (0, 8, 16, 24, 32, 40, 48, 56):
digest ^= (item >> shift) & 0xFF
digest = (digest * 1099511628211) & mask
head = values[:sample]
tail = values[-sample:] if len(values) > sample else values[:]
return f"len={len(values)} hash={digest} head={head} tail={tail}"
def _poll_queue_debug_hash(debug_ids: Optional[list[str]]) -> int:
if not debug_ids:
return 0

4
python/sglang/srt/managers/schedule_policy.py
View File

@@ -198,8 +198,7 @@ class SchedulePolicy:
# NOTE: the prefix_indices must always be aligned with last_node
match_result = self.tree_cache.match_prefix(
MatchPrefixParams(
key=RadixKey(token_ids=prefix_ids, extra_key=extra_key),
req=r,
key=RadixKey(token_ids=prefix_ids, extra_key=extra_key)
)
)
(
@@ -858,7 +857,6 @@ class PrefillAdder:
last_host_node=req.last_host_node,
host_hit_length=req.host_hit_length,
mem_quota=self._get_load_back_mem_quota(real_input_tokens),
req=req,
)
)
req.prefix_indices = torch.cat([req.prefix_indices, new_indices])

8
python/sglang/srt/managers/scheduler.py
View File

@@ -60,7 +60,6 @@ from sglang.srt.disaggregation.utils import (
ReqToMetadataIdxAllocator,
TransferBackend,
prepare_abort,
token_ids_debug_signature,
)
from sglang.srt.distributed import get_pp_group, get_world_group
from sglang.srt.distributed.parallel_state import get_tp_group
@@ -2585,7 +2584,7 @@ class Scheduler(
"scheduler_prefill_batch",
"bs=%s extend_lens=%s prefix_lens=%s seq_lens=%s "
"out_cache_tokens=%s chunked_req=%s enable_bs_gt1=%s "
"max_batch_reqs=%s max_total_extend=%s rids=%s origin_sigs=%s",
"max_batch_reqs=%s max_total_extend=%s",
len(can_run_list),
list(getattr(new_batch, "extend_lens", []) or []),
list(getattr(new_batch, "prefix_lens", []) or []),
@@ -2604,11 +2603,6 @@ class Scheduler(
"cp_shared_kv_prefill_max_total_extend_tokens",
None,
),
[getattr(req, "rid", None) for req in can_run_list],
[
token_ids_debug_signature(getattr(req, "origin_input_ids", None))
for req in can_run_list
],
)
# Record prefill stats for logging after forward

187
python/sglang/srt/mem_cache/hiradix_cache.py
View File

@@ -65,33 +65,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
_CP_HICACHE_BS_GT1_DEBUG_COUNTS: Dict[str, int] = {}
def _cp_hicache_bs_gt1_debug(
key: str,
message: str,
*args,
limit: Optional[int] = None,
) -> None:
"""Env-gated, rate-limited CP HiCache debug logging.
Keep this local to mem_cache to avoid importing NSA attention helpers from
the radix cache path. It is intentionally default-off because these paths
run on scheduler/control hot paths.
"""
if not envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG.get():
return
if limit is None:
limit = envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG_LIMIT.get()
limit = int(limit)
count = _CP_HICACHE_BS_GT1_DEBUG_COUNTS.get(key, 0)
if limit > 0 and count >= limit:
return
_CP_HICACHE_BS_GT1_DEBUG_COUNTS[key] = count + 1
logger.info("[CP_SHARED_KV_BS_GT1_DEBUG][hicache] event=%s " + message, key, *args)
def _estimate_hicache_size_per_token(kv_cache) -> int:
dtype_size = kv_cache.store_dtype.itemsize
@@ -2019,20 +1992,6 @@ class HiRadixCache(RadixCache):
node.host_value = None
if pending.locked:
self.dec_node_lock_ref(node)
_cp_hicache_bs_gt1_debug(
"commit_pending_backup",
"node_id=%d key_len=%d host_len=%d padded_len=%s "
"owned_positions=%d host_indices=%d page_owner_pages=%d "
"remaining_pending=%d",
node.id,
len(node.key),
node.host_len,
getattr(pending.metadata, "padded_len", None),
pending.metadata.owned_positions.numel(),
pending.metadata.host_indices.numel(),
pending.metadata.page_owners.numel(),
len(self.pending_host_backups),
)
return node
def _rollback_pending_backup(self, node_id: int) -> TreeNode:
@@ -2197,21 +2156,6 @@ class HiRadixCache(RadixCache):
locked=True,
)
prepared.attached = True
_cp_hicache_bs_gt1_debug(
"attach_prepared_backup",
"node_id=%d key_len=%d value_len=%d logical_len=%d padded_len=%s "
"owned_positions=%d host_indices=%d page_owner_pages=%d "
"pending_backups=%d",
node.id,
len(node.key),
len(node.value),
prepared.logical_len,
getattr(prepared.metadata, "padded_len", None),
prepared.metadata.owned_positions.numel(),
prepared.metadata.host_indices.numel(),
prepared.metadata.page_owners.numel(),
len(self.pending_host_backups),
)
logger.info(
"[HiCache-write] attached prepared CP backup: node_id=%d logical_len=%d owned_positions=%d pending_backups=%d",
node.id,
@@ -2336,7 +2280,6 @@ class HiRadixCache(RadixCache):
request_len: int,
*,
floor_exact_key: bool = False,
debug_req_id: Optional[str] = None,
) -> int:
"""Floor exact CP valid-tail hits to the previous physical page.
@@ -2363,29 +2306,7 @@ class HiRadixCache(RadixCache):
or prefix_len % self.page_size == 0
):
return prefix_len
floored_len = floor_to_page_len(prefix_len, self.page_size)
metadata = getattr(child, "cp_hicache", None)
_cp_hicache_bs_gt1_debug(
"match_floor_valid_tail",
"rid=%s node_id=%s prefix_len=%d floored_len=%d request_len=%d "
"child_key_len=%d page_size=%d evicted=%s host_len=%d "
"has_cp_hicache=%s padded_len=%s pending_write=%s floor_exact_key=%s",
debug_req_id,
getattr(child, "id", None),
prefix_len,
floored_len,
request_len,
len(child.key),
self.page_size,
child.evicted,
getattr(child, "host_len", 0),
metadata is not None,
getattr(metadata, "padded_len", None),
self._node_host_write_pending(child),
floor_exact_key,
limit=64,
)
return floored_len
return floor_to_page_len(prefix_len, self.page_size)
def _cp_subtree_has_unprunable_state(self, node: TreeNode) -> bool:
stack = [node]
@@ -2637,25 +2558,6 @@ class HiRadixCache(RadixCache):
metadata=result.metadata,
logical_len=len(kv_indices),
)
_cp_hicache_bs_gt1_debug(
"prepare_write_backup",
"node_id=%d rid=%s logical_len=%d padded_len=%s "
"owned_positions=%d host_indices=%d draft_host_indices=%s "
"page_owner_pages=%d admission_checked=%s",
node_id,
getattr(req, "rid", "<unknown>"),
len(kv_indices),
getattr(result.metadata, "padded_len", None),
result.metadata.owned_positions.numel(),
result.metadata.host_indices.numel(),
(
None
if getattr(result.metadata, "draft_host_indices", None) is None
else result.metadata.draft_host_indices.numel()
),
result.metadata.page_owners.numel(),
admission_checked,
)
logger.info(
"[HiCache-write] prepared CP per-layer backup before forward: node_id=%d rid=%s logical_len=%d owned_positions=%d",
node_id,
@@ -3504,30 +3406,6 @@ class HiRadixCache(RadixCache):
self.dec_lock_ref(ancester_node)
raise
host_hit_len = load_back_plan.host_hit_len
node_debug_summary = [
(
getattr(x, "id", None),
len(x.key) if getattr(x, "key", None) is not None else 0,
getattr(x, "host_len", 0),
getattr(getattr(x, "cp_hicache", None), "padded_len", None),
)
for x in nodes_to_load
]
_cp_hicache_bs_gt1_debug(
"load_back_plan",
"node_id=%d nodes=%s host_hit_len=%d threshold=%d "
"required_by_owner=%s available_by_owner=%s deficit_by_owner=%s "
"free_room_deficit_by_owner=%s",
last_hit_node.id,
node_debug_summary,
host_hit_len,
self.load_back_threshold,
load_back_plan.required_by_owner,
load_back_plan.available_by_owner,
load_back_plan.deficit_by_owner,
load_back_plan.free_room_deficit_by_owner,
limit=64,
)
logger.info(
"[HiCache-load] load_back CP: node_id=%d nodes_to_load=%d "
"host_hit_len=%d threshold=%d required_by_owner=%s "
@@ -3760,46 +3638,16 @@ class HiRadixCache(RadixCache):
):
last_node = params.last_host_node
mem_quota = params.mem_quota
debug_req_id = getattr(getattr(params, "req", None), "rid", None)
if last_node.evicted:
if self._uses_cp_hicache:
_cp_hicache_bs_gt1_debug(
"init_load_back_start",
"rid=%s last_host_node=%s host_hit_length=%d mem_quota=%s "
"node_host_len=%d",
debug_req_id,
getattr(last_node, "id", None),
int(params.host_hit_length),
mem_quota,
getattr(last_node, "host_len", 0),
)
loading_values = self.load_back(last_node, mem_quota)
if loading_values is not None:
logger.info(
f"loading back {len(loading_values)} tokens for node {last_node.id}"
)
if self._uses_cp_hicache:
_cp_hicache_bs_gt1_debug(
"init_load_back_loaded",
"rid=%s loaded_tokens=%d loaded_node=%s host_hit_length=%d",
debug_req_id,
len(loading_values),
getattr(last_node, "id", None),
int(params.host_hit_length),
)
return loading_values, last_node
while last_node.evicted:
last_node = last_node.parent
if self._uses_cp_hicache:
_cp_hicache_bs_gt1_debug(
"init_load_back_unloaded",
"rid=%s fallback_device_node=%s host_hit_length=%d mem_quota=%s",
debug_req_id,
getattr(last_node, "id", None),
int(params.host_hit_length),
mem_quota,
)
return (
torch.empty((0,), dtype=torch.int64, device=self.device),
@@ -4022,13 +3870,10 @@ class HiRadixCache(RadixCache):
deferred_node = None
try:
debug_req = getattr(params, "req", None)
debug_req_id = getattr(debug_req, "rid", None)
value, last_node = self._match_prefix_helper(
self.root_node,
key,
floor_exact_key=getattr(params, "cp_floor_exact", True),
debug_req_id=debug_req_id,
)
except HiCachePendingBackupSplit as exc:
value = []
@@ -4061,28 +3906,6 @@ class HiRadixCache(RadixCache):
if not last_host_node.backuped:
last_host_node = self.root_node
if self._uses_cp_hicache:
debug_req = getattr(params, "req", None)
_cp_hicache_bs_gt1_debug(
"match_prefix_result",
"rid=%s key_len=%d device_tokens=%d host_hit_length=%d "
"last_device_node=%s last_device_evicted=%s last_device_host_len=%d "
"last_host_node=%s last_host_evicted=%s last_host_host_len=%d "
"deferred_node=%s cp_floor_exact=%s",
getattr(debug_req, "rid", None),
len(key),
len(value),
host_hit_length,
getattr(last_node, "id", None),
getattr(last_node, "evicted", None),
getattr(last_node, "host_len", 0),
getattr(last_host_node, "id", None),
getattr(last_host_node, "evicted", None),
getattr(last_host_node, "host_len", 0),
getattr(deferred_node, "id", None),
getattr(params, "cp_floor_exact", True),
)
return MatchResult(
device_indices=value,
last_device_node=last_node,
@@ -4180,12 +4003,7 @@ class HiRadixCache(RadixCache):
return matched_length
def _match_prefix_helper(
self,
node: TreeNode,
key: RadixKey,
*,
floor_exact_key: bool = True,
debug_req_id: Optional[str] = None,
self, node: TreeNode, key: RadixKey, *, floor_exact_key: bool = True
):
node.last_access_time = time.monotonic()
child_key = self.get_child_key_fn(key)
@@ -4203,7 +4021,6 @@ class HiRadixCache(RadixCache):
raw_prefix_len,
len(key),
floor_exact_key=floor_exact_key,
debug_req_id=debug_req_id,
)
stop_after_page_floor = prefix_len != raw_prefix_len
prune_stale_tail_after_split = stop_after_page_floor

90
python/sglang/srt/mem_cache/radix_cache.py
View File

@@ -31,8 +31,6 @@ from typing import TYPE_CHECKING, Any, Iterator, List, Optional, Tuple, Union
import torch
from sglang.srt.environ import envs
logger = logging.getLogger(__name__)
from sglang.srt.disaggregation.kv_events import (
@@ -68,41 +66,6 @@ from sglang.srt.mem_cache.hicache_storage import get_hash_str, hash_str_to_int64
if TYPE_CHECKING:
from sglang.srt.managers.schedule_batch import Req
_CP_SHARED_KV_RADIX_DEBUG_COUNTS: dict[str, int] = {}
def _cp_shared_kv_radix_debug(
key: str,
message: str,
*args,
limit: int | None = None,
) -> None:
if not envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG.get():
return
if limit is None:
limit = int(envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG_LIMIT.get())
count = _CP_SHARED_KV_RADIX_DEBUG_COUNTS.get(key, 0)
if limit > 0 and count >= limit:
return
_CP_SHARED_KV_RADIX_DEBUG_COUNTS[key] = count + 1
logger.info("[CP_SHARED_KV_BS_GT1_DEBUG][radix] event=%s " + message, key, *args)
def _cp_shared_kv_page_head_for_debug(
locs: torch.Tensor | None,
*,
page_size: int,
max_items: int = 8,
) -> list[int] | None:
if locs is None:
return None
if page_size <= 0 or locs.numel() == 0:
return []
pages = locs[::page_size]
if pages.numel() == 0:
return []
return (pages[:max_items].detach().to("cpu", non_blocking=False) // page_size).tolist()
class RadixKey:
def __init__(
@@ -687,24 +650,6 @@ class RadixCache(BasePrefixCache):
values = kv_indices[: len(keys)].to(dtype=torch.int64, copy=True)
radix_key = RadixKey(keys, req.extra_key, is_bigram=self.is_eagle)
prepared_cp_backup = getattr(req, "cp_hicache_prepared_backup", None)
if getattr(self, "_uses_cp_hicache", False):
_cp_shared_kv_radix_debug(
"unfinished_pre_insert",
"rid=%s chunked=%s key_len=%d kv_len=%d cache_protected_len=%d "
"req_pool_idx=%s value_pages=%s kv_pages=%s prefix_pages=%s",
getattr(req, "rid", None),
chunked,
len(keys),
int(kv_indices.numel()),
int(getattr(req, "cache_protected_len", 0) or 0),
getattr(req, "req_pool_idx", None),
_cp_shared_kv_page_head_for_debug(values, page_size=self.page_size),
_cp_shared_kv_page_head_for_debug(kv_indices, page_size=self.page_size),
_cp_shared_kv_page_head_for_debug(
getattr(req, "prefix_indices", None), page_size=self.page_size
),
limit=160,
)
# Radix Cache takes one ref in memory pool
result = self.insert(
@@ -761,46 +706,11 @@ class RadixCache(BasePrefixCache):
match_result.last_device_node,
)
assert len(new_indices) == len(keys), f"{len(new_indices)=}, {len(keys)=}"
if getattr(self, "_uses_cp_hicache", False):
_cp_shared_kv_radix_debug(
"unfinished_post_match",
"rid=%s chunked=%s key_len=%d old_cache_protected_len=%d "
"new_indices_len=%d new_node=%s new_pages=%s kv_pages_before=%s",
getattr(req, "rid", None),
chunked,
len(keys),
int(getattr(req, "cache_protected_len", 0) or 0),
int(new_indices.numel()),
getattr(new_last_node, "id", None),
_cp_shared_kv_page_head_for_debug(
new_indices, page_size=self.page_size
),
_cp_shared_kv_page_head_for_debug(kv_indices, page_size=self.page_size),
limit=160,
)
self.req_to_token_pool.write(
(req.req_pool_idx, slice(req.cache_protected_len, len(new_indices))),
new_indices[req.cache_protected_len :],
)
if getattr(self, "_uses_cp_hicache", False):
final_kv_indices = self.req_to_token_pool.req_to_token[
req.req_pool_idx, : len(token_ids)
]
_cp_shared_kv_radix_debug(
"unfinished_post_write",
"rid=%s chunked=%s key_len=%d cache_protected_len=%d "
"final_kv_len=%d final_pages=%s",
getattr(req, "rid", None),
chunked,
len(keys),
int(getattr(req, "cache_protected_len", 0) or 0),
int(final_kv_indices.numel()),
_cp_shared_kv_page_head_for_debug(
final_kv_indices, page_size=self.page_size
),
limit=160,
)
# 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.