[Observability] Break down prefill #cached-token into device / host / storage tiers

python/sglang/srt/managers/schedule_batch.py

@@ -146,6 +146,21 @@ def _compute_pad_value(hash: int) -> int:
     return MM_PAD_SHIFT_VALUE + (hash % (1 << 30))
 
 
+def compute_cache_hit_split(
+    prefix_len: int, loaded_host_hit_length: int, storage_hit_length: int
+) -> tuple:
+    """Return (device, host, storage) token counts. Always sums to prefix_len.
+
+    loaded_host_hit_length is the number of KV tokens actually restored from
+    host in this scheduling decision. It intentionally differs from
+    host_hit_length, which can be a load-back trigger sentinel for hybrid
+    caches.
+    """
+    host_total = min(loaded_host_hit_length, prefix_len)
+    storage = min(host_total, storage_hit_length)
+    return prefix_len - host_total, host_total - storage, storage
+
+
 class BaseFinishReason:
     def to_json(self):
         raise NotImplementedError()
@@ -851,6 +866,7 @@ def __init__(
         self.host_hit_length = 0
         self.swa_host_hit_length = 0
         self.mamba_host_hit_length = 0
+        self.loaded_host_hit_length = 0
         # Total cached prefix length (on-device prefix_indices + host_hit_length),
         # capped at the max allowed prefix. Set during prefix matching at schedule
         # time and used to estimate uncached tokens / sort by longest prefix for
@@ -1195,6 +1211,7 @@ def init_next_round_input(
                 match_result.mamba_host_hit_length,
                 match_result.mamba_branching_seqlen,
             )
+            self.loaded_host_hit_length = 0
             if match_result.cache_protected_len is not None:
                 self.cache_protected_len = match_result.cache_protected_len
             else:
@@ -1445,6 +1462,9 @@ def reset_for_retract(self):
         self.last_node = None
         self.cache_protected_len = 0
         self.num_matched_prefix_tokens = 0
+        self.host_hit_length = 0
+        self.loaded_host_hit_length = 0
+        self.storage_hit_length = 0
         self.swa_uuid_for_lock = None
         self.swa_prefix_lock_released = False
         self.extend_range = None
@@ -2115,24 +2135,15 @@ def prepare_for_extend(self):
                 # Only compute once on FIRST chunk - subsequent chunks in chunked prefill
                 # would incorrectly count previously computed tokens as cache hits.
                 if not req._cache_breakdown_computed:
-                    # At this point, prefix_indices has been extended with host data
-                    # via init_load_back in schedule_policy, so:
-                    # - len(prefix_indices) = device_original + host_loaded
-                    # - host_hit_length = total tokens from host cache (including storage-prefetched)
-                    # - storage_hit_length = tokens loaded from storage backend (L3 hits)
-                    # - device_portion = len(prefix_indices) - host_hit_length
-                    #
-                    # Storage hits are now tracked via scheduler after prefetch completes.
-                    # storage_hit_length is set by scheduler.pop_prefetch_loaded_tokens()
-                    host_total = req.host_hit_length
-                    # Clamp storage to host_total to handle edge cases
-                    storage_portion = min(host_total, req.storage_hit_length)
-                    host_portion = host_total - storage_portion
-                    device_portion = max(0, len(req.prefix_indices) - host_total)
-
-                    req.cached_tokens_device = device_portion
-                    req.cached_tokens_host = host_portion
-                    req.cached_tokens_storage = storage_portion
+                    (
+                        req.cached_tokens_device,
+                        req.cached_tokens_host,
+                        req.cached_tokens_storage,
+                    ) = compute_cache_hit_split(
+                        len(req.prefix_indices),
+                        req.loaded_host_hit_length,
+                        req.storage_hit_length,
+                    )
                     req._cache_breakdown_computed = True
 
                 req.already_computed = seq_len

python/sglang/srt/managers/schedule_policy.py

@@ -37,7 +37,11 @@
 from sglang.srt.dllm.config import DllmConfig
 from sglang.srt.layers.attention.dsa.utils import is_dsa_prefill_cp_in_seq_split
 from sglang.srt.layers.utils.cp_utils import is_prefill_context_parallel_enabled
-from sglang.srt.managers.schedule_batch import Req, ScheduleBatch
+from sglang.srt.managers.schedule_batch import (
+    Req,
+    ScheduleBatch,
+    compute_cache_hit_split,
+)
 from sglang.srt.mem_cache.allocator.hisparse import (
     DeepSeekV4HiSparseTokenToKVPoolAllocator,
 )
@@ -123,6 +127,7 @@ def match_prefix_for_req(
     req.num_matched_prefix_tokens = min(
         len(req.prefix_indices) + req.host_hit_length, max_len
     )
+    req.loaded_host_hit_length = 0
     if match_result.mamba_branching_seqlen is not None:
         req.mamba_branching_seqlen = match_result.mamba_branching_seqlen
     if match_result.cache_protected_len is not None:
@@ -462,8 +467,14 @@ def __init__(
         self.can_run_list = []
         self.preempt_list = []
         self.new_chunked_req = None
+
         self.log_hit_tokens = 0
+        self.log_hit_tokens_device = 0
+        self.log_hit_tokens_host = 0
+        self.log_hit_tokens_storage = 0
+
         self.reprocessed_log_hit_tokens = 0
+
         # TODO(lsyin): report the real input tokens excluding page alignment
         self.log_input_tokens = 0
         self.reprocessed_log_input_tokens = 0
@@ -606,6 +617,23 @@ def budget_state(self):
 
         return AddReqResult.CONTINUE
 
+    def _accumulate_hit_token_split(self, req, prefix_len: int) -> None:
+        """Accumulate per-source cache-hit split alongside log_hit_tokens.
+
+        Must be called exactly once per scheduling decision (not for chunk 2+
+        of chunked prefill, where prefix_len passed to _update_prefill_budget
+        is 0). Called for both retracted and non-retracted requests so that
+        the split always sums to log_hit_tokens.
+        """
+        if prefix_len <= 0:
+            return
+        device, host, storage = compute_cache_hit_split(
+            prefix_len, req.loaded_host_hit_length, req.storage_hit_length
+        )
+        self.log_hit_tokens_device += device
+        self.log_hit_tokens_host += host
+        self.log_hit_tokens_storage += storage
+
     def _update_prefill_budget(
         self,
         prefix_len: int,
@@ -663,7 +691,13 @@ def _add_dllm_req(self, req: Req, prefix_len: int):
 
         self.can_run_list.append(req)
 
-        self._update_prefill_budget(prefix_len, trunc_len, 0, req.retracted_stain)
+        self._accumulate_hit_token_split(req, prefix_len)
+        self._update_prefill_budget(
+            prefix_len,
+            trunc_len,
+            0,
+            req.retracted_stain,
+        )
 
     def _req_inc_lock_ref(self, req: Req):
         result = self.tree_cache.inc_lock_ref(req.last_node)
@@ -947,9 +981,12 @@ def add_one_req(
                         req=req,
                     )
                 )
+                req.loaded_host_hit_length = len(new_indices)
                 req.prefix_indices = torch.cat([req.prefix_indices, new_indices])
                 prefix_len = len(req.prefix_indices)
                 req.cache_protected_len = prefix_len
+            else:
+                req.loaded_host_hit_length = 0
 
             input_tokens = self.ceil_paged_tokens(
                 len(req.full_untruncated_fill_ids) - len(req.prefix_indices)
@@ -983,6 +1020,7 @@ def add_one_req(
                 self.can_run_list.append(req)
 
                 self._req_inc_lock_ref(req)
+                self._accumulate_hit_token_split(req, prefix_len)
                 self._update_prefill_budget(
                     prefix_len,
                     input_tokens,
@@ -1026,8 +1064,12 @@ def add_one_req(
                 self.new_chunked_req = req
 
                 self._req_inc_lock_ref(req)
+                self._accumulate_hit_token_split(req, prefix_len)
                 self._update_prefill_budget(
-                    prefix_len, trunc_len, 0, req.retracted_stain
+                    prefix_len,
+                    trunc_len,
+                    0,
+                    req.retracted_stain,
                 )
 
         return self.budget_state()

python/sglang/srt/managers/scheduler_components/metrics_reporter.py

@@ -60,6 +60,15 @@ class PrefillStats:
     new_token_ratio: float
     num_running_reqs: QueueCount
     num_new_seqs: int  # len(can_run_list)
+
+    # Detailed prefix-cache hit split.
+    # log_hit_tokens should equal:
+    #   log_hit_tokens_device + log_hit_tokens_host + log_hit_tokens_storage
+    # when storage accounting is enabled. Without L3 storage, storage is 0.
+    log_hit_tokens_device: int = 0
+    log_hit_tokens_host: int = 0
+    log_hit_tokens_storage: int = 0
+
     reprocessed_log_input_tokens: int = 0
     reprocessed_log_hit_tokens: int = 0
     num_pending_tokens: int = 0
@@ -72,11 +81,18 @@ def from_adder(
         enable_priority_scheduling: bool = False,
         num_pending_tokens: int = 0,
     ):
+        log_hit_tokens_device = adder.log_hit_tokens_device
+        log_hit_tokens_host = adder.log_hit_tokens_host
+        log_hit_tokens_storage = adder.log_hit_tokens_storage
+
         return cls(
             log_input_tokens=adder.log_input_tokens,
             log_hit_tokens=adder.log_hit_tokens,
             reprocessed_log_input_tokens=adder.reprocessed_log_input_tokens,
             reprocessed_log_hit_tokens=adder.reprocessed_log_hit_tokens,
+            log_hit_tokens_device=log_hit_tokens_device,
+            log_hit_tokens_host=log_hit_tokens_host,
+            log_hit_tokens_storage=log_hit_tokens_storage,
             new_token_ratio=adder.new_token_ratio,
             num_running_reqs=QueueCount.from_reqs(
                 running_reqs, enable_priority_scheduling
@@ -538,11 +554,30 @@ def report_prefill_stats(
         )
         iter_msg = f" [{batch_iter}]" if LOG_FORWARD_ITERS else ""
 
+        cache_split_msg = (
+            (
+                f"#cached-token-device: {prefill_stats.log_hit_tokens_device}, "
+                f"#cached-token-host: {prefill_stats.log_hit_tokens_host}, "
+            )
+            if prefill_stats.log_hit_tokens_host > 0
+            or prefill_stats.log_hit_tokens_device > 0
+            else ""
+        )
+
+        if (
+            getattr(self.scheduler, "enable_hicache_storage", False)
+            or prefill_stats.log_hit_tokens_storage > 0
+        ):
+            cache_split_msg += (
+                f"#cached-token-storage: {prefill_stats.log_hit_tokens_storage}, "
+            )
+
         msg = (
             f"Prefill batch{iter_msg}, "
             f"#new-seq: {prefill_stats.num_new_seqs}, "
             f"#new-token: {prefill_stats.log_input_tokens}, "
             f"#cached-token: {prefill_stats.log_hit_tokens}, "
+            f"{cache_split_msg}"
             f"{token_usage_msg}"
             f"#running-req: {prefill_stats.num_running_reqs.total}, "
             f"#queue-req: {len(self.scheduler.waiting_queue)}, "