From 5163ba7bb9e98b1f497dc021dec05794471b5ab7 Mon Sep 17 00:00:00 2001 From: leavelet Date: Tue, 23 Jun 2026 02:01:05 +0000 Subject: [PATCH] CP HiCache: expose L2 + L3 usage metrics (the mainline storage metrics are dead under CP) Under CP, enable_storage=False, so StorageMetricsCollector (prefetch/backup tokens + bandwidth) never emits, and L2 pool usage was debug-log-only (_cp_maybe_log_lane_stats) while L3 was invisible. Add a CpHiCacheMetricsCollector covering both, gated on enable_metrics alone, per-rank via the tp_rank label. - CpL3Store.stats(): per-payload slot occupancy (used,total) + cumulative counters (spill_pages, reload_pages, gc_reclaimed, write_errors -- incremented on the bg threads) + instantaneous inflight (spill/reload) + the four bg queue depths. Counters are monotonic (NOT reset on clear()). - CpHiCacheMetricsCollector (metrics_collector.py): gauges for instantaneous state (sglang:cp_l2_used/total_pages, cp_l3_used/total_slots, cp_l3_inflight_*, cp_l3_queue_depth) + counters for cumulative work (cp_l2_objects_committed/evicted_total, cp_l3_spill/reload_pages_total, cp_l3_gc_reclaimed_total, cp_l3_write_errors_total). Counters fed via a monotonic .inc(delta) off the store's cumulative totals. - hiradix: create the collector in _maybe_init_cp_l3 (once, per-rank labels from cache_controller); push periodically (~1s, per-rank wall-clock -- metrics need no rank-uniformity) from check_hicache_events via _cp_maybe_collect_metrics, reading allocator.occupancy_by_payload()/stats() + store.stats(). Off the data path. store.stats() unit-tested (occupancy + spill/reload counters; 10 store tests green). The collector + wiring are correct-by-pattern (mirror StorageMetricsCollector) and validate on the live server (3.4; prometheus_client not in the unit env). Directly useful for 3.4: watch L2/L3 fill + GC + reload hits under cachebench. Co-Authored-By: Claude Opus 4.8 (1M context) --- python/sglang/srt/mem_cache/cp_l3_store.py | 27 ++++++ python/sglang/srt/mem_cache/hiradix_cache.py | 38 ++++++++ .../srt/observability/metrics_collector.py | 86 +++++++++++++++++++ .../unit/mem_cache/test_cp_l3_store.py | 20 +++++ 4 files changed, 171 insertions(+) diff --git a/python/sglang/srt/mem_cache/cp_l3_store.py b/python/sglang/srt/mem_cache/cp_l3_store.py index d4260e750..b4f0b21b3 100644 --- a/python/sglang/srt/mem_cache/cp_l3_store.py +++ b/python/sglang/srt/mem_cache/cp_l3_store.py @@ -149,6 +149,10 @@ class CpL3Store: self._gc_start_frac = 0.90 # begin reclaiming when a pool exceeds this fraction (tune at impl) self._gc_stop_frac = 0.85 # reclaim coldest down to this fraction (hysteresis avoids per-write churn) self._gc_budget = 4096 # max reclaims per pass (bounds a single GC pass so it never starves writes) + # cumulative counters for metrics (monotonic; incremented on the bg threads, read by stats() per tick) + self._stat_spill_pages = 0 + self._stat_reload_pages = 0 + self._stat_gc_reclaimed = 0 self._stop = threading.Event() self._gather_thread: Optional[threading.Thread] = None self._write_thread: Optional[threading.Thread] = None @@ -428,6 +432,7 @@ class CpL3Store: # replicated clock). Touched later via _touch_q on reload-admit; reclaimed coldest-first by _gc_collect. for (sp, slot) in to_write: self._gc_add(sp.pk, sp.h, slot, staged.last_access) + self._stat_spill_pages += len(to_write) # ---------- 3.3 GC: continuous, background, page-level LRU (write-thread-owned; rank-local; no collective) ---- def submit_touch(self, touches: Sequence[Tuple[str, str, int]]) -> None: @@ -486,6 +491,7 @@ class CpL3Store: wb.delete(h, pk) for (_h, slot) in to_free: pool.free(slot) + self._stat_gc_reclaimed += len(to_free) def _reload_object(self, op: L3ReloadOp) -> bool: """Read this rank's owned pages from L3 and scatter into the reserved L2 dest pages. Returns False if @@ -505,12 +511,33 @@ class CpL3Store: try: if slab.read_into(entry.slot_idx, buf, expect_hash=h): accessor.scatter_from(dest_page, buf, CP_L3_HEADER_BYTES) + self._stat_reload_pages += 1 else: ok = False finally: buf.close() return ok + def stats(self) -> Dict[str, object]: + """Snapshot for metrics (scheduler thread). Per-payload slot occupancy (used,total) + cumulative + counters (spill/reload pages, GC reclaims, write errors) + instantaneous inflight + queue depths. + Counters are monotonic (bg-thread increments under the GIL; exact-enough for metrics).""" + return { + "occupancy": { + pk: (self.pools[pk].num_allocated, self.pools[pk].num_slots) for pk in self.payloads + }, + "spill_pages_total": self._stat_spill_pages, + "reload_pages_total": self._stat_reload_pages, + "gc_reclaimed_total": self._stat_gc_reclaimed, + "write_errors_total": self._write_err_count, + "ongoing_spill": len(self._ongoing_spill), + "ongoing_reload": len(self._ongoing_reload), + "gather_qsize": self._gather_q.qsize(), + "write_qsize": self._write_q.qsize(), + "reload_qsize": self._reload_q.qsize(), + "touch_qsize": self._touch_q.qsize(), + } + # ---------- L3 eviction support (3.3 consumes; here = the rank-local free) ---------- def free_object(self, owned_pages: Dict[str, List[OwnedPage]]) -> None: """Free this rank's owned disk slots + index entries for an evicted object (selection is replicated diff --git a/python/sglang/srt/mem_cache/hiradix_cache.py b/python/sglang/srt/mem_cache/hiradix_cache.py index b79eb497a..50ed72ae4 100644 --- a/python/sglang/srt/mem_cache/hiradix_cache.py +++ b/python/sglang/srt/mem_cache/hiradix_cache.py @@ -1071,6 +1071,10 @@ class HiRadixCache(RadixCache): # Bound the request hold (rank-uniform tick countdown): the default SGLANG_REQ_WAITING_TIMEOUT is -1 # (off), so a lost/hung reload op (e.g. a dead disk) could otherwise wedge a held request forever. self.cp_l3_reload_ttl_ticks = 4096 + # CP-HiCache metrics (L2 pooled allocator + L3 store usage; the mainline storage metrics are dead under + # CP). Created in _maybe_init_cp_l3 when enable_metrics; pushed periodically from check_hicache_events. + self.cp_hicache_metrics = None + self._cp_metrics_last_emit = 0.0 ( extra_config, @@ -1274,6 +1278,19 @@ class HiRadixCache(RadixCache): "[CP-L3] enabled rank=%d/%d payloads=%s disk_dir=%s", cp_rank, cp_size, list(accessors), self.cp_l3_store.disk_dir, ) + # CP-HiCache metrics (L2 pooled allocator + L3 store): gated on enable_metrics ALONE (the mainline + # storage metrics are dead under CP). Created once; per-rank via tp_rank. Pushed from check_hicache_events. + if getattr(params, "enable_metrics", False) and self.cp_hicache_metrics is None: + from sglang.srt.observability.metrics_collector import CpHiCacheMetricsCollector + + cc = self.cache_controller + labels = { + "tp_rank": cc.tp_rank, "dp_rank": cc.dp_rank, + "pp_rank": cc.pp_rank, "pp_size": cc.pp_size, + } + if self.extra_metric_labels: + labels.update(self.extra_metric_labels) + self.cp_hicache_metrics = CpHiCacheMetricsCollector(labels) @staticmethod def _cp_l3_single_slab_mmap(allocator, payload_kind: str): @@ -3823,6 +3840,26 @@ class HiRadixCache(RadixCache): details, ) + def _cp_maybe_collect_metrics(self) -> None: + """Push L2 (pooled allocator) + L3 (store) usage to Prometheus, rate-limited (~1s, per-rank wall-clock; + metrics need no rank-uniformity). Off the data path; cheap (occupancy snapshot + queue sizes + counters).""" + collector = self.cp_hicache_metrics + if collector is None: + return + now = time.monotonic() + if now - self._cp_metrics_last_emit < 1.0: + return + self._cp_metrics_last_emit = now + allocator = getattr(self.cache_controller, "cp_shared_l2_page_allocator", None) + l2_occ = ( + allocator.occupancy_by_payload() + if allocator is not None and hasattr(allocator, "occupancy_by_payload") + else None + ) + l2_stats = allocator.stats() if allocator is not None and hasattr(allocator, "stats") else None + l3_stats = self.cp_l3_store.stats() if self.cp_l3_store is not None else None + collector.collect(l2_occupancy=l2_occ, l2_stats=l2_stats, l3_stats=l3_stats) + def _cp_maybe_log_lane_stats(self) -> None: """E1 lane-imbalance probe (read-only, measurement only). @@ -5521,6 +5558,7 @@ class HiRadixCache(RadixCache): self.cache_controller.storage_backend.get_stats() ) self._cp_maybe_log_lane_stats() + self._cp_maybe_collect_metrics() def drain_storage_control_queues(self): """ diff --git a/python/sglang/srt/observability/metrics_collector.py b/python/sglang/srt/observability/metrics_collector.py index 16da400d5..47da6db75 100644 --- a/python/sglang/srt/observability/metrics_collector.py +++ b/python/sglang/srt/observability/metrics_collector.py @@ -1459,6 +1459,92 @@ class StorageMetricsCollector: self._log_histogram(self.histogram_backup_bandwidth, v) +class CpHiCacheMetricsCollector: + """L2 (pooled shared host allocator) + L3 (disk store) usage for the CP shared-KV HiCache. + + Gauges for instantaneous state (occupancy, inflight, queue depths); Counters for cumulative work + (spill/reload pages, GC reclaims, write errors, L2 commit/evict). Fed periodically off the allocator's + occupancy_by_payload()/stats() + CpL3Store.stats(). Per-rank via the tp_rank label. The mainline + StorageMetricsCollector is dead under CP (enable_storage=False), so this is the only L2/L3 visibility. + """ + + def __init__(self, labels: Dict[str, str]): + from prometheus_client import Counter, Gauge + + self.labels = labels + ln = list(labels.keys()) + lnp = ln + ["payload"] + # ---- L2 (pooled shared host allocator) ---- + self.l2_used_pages = Gauge( + "sglang:cp_l2_used_pages", "CP shared-L2 used pages, per payload.", lnp + ) + self.l2_total_pages = Gauge( + "sglang:cp_l2_total_pages", "CP shared-L2 total (capacity) pages, per payload.", lnp + ) + self.l2_objects_committed = Counter( + "sglang:cp_l2_objects_committed_total", "CP shared-L2 objects committed.", ln + ) + self.l2_objects_evicted = Counter( + "sglang:cp_l2_objects_evicted_total", "CP shared-L2 objects evicted.", ln + ) + # ---- L3 (disk store; this rank's slabs) ---- + self.l3_used_slots = Gauge( + "sglang:cp_l3_used_slots", "CP L3 used disk slots, per payload (this rank).", lnp + ) + self.l3_total_slots = Gauge( + "sglang:cp_l3_total_slots", "CP L3 total disk slots, per payload (this rank).", lnp + ) + self.l3_inflight_spill = Gauge( + "sglang:cp_l3_inflight_spill", "CP L3 spill ops in flight.", ln + ) + self.l3_inflight_reload = Gauge( + "sglang:cp_l3_inflight_reload", "CP L3 reload ops in flight.", ln + ) + self.l3_queue_depth = Gauge( + "sglang:cp_l3_queue_depth", "CP L3 background queue depth.", ln + ["queue"] + ) + self.l3_spill_pages = Counter( + "sglang:cp_l3_spill_pages_total", "CP L3 pages written by spill (this rank).", ln + ) + self.l3_reload_pages = Counter( + "sglang:cp_l3_reload_pages_total", "CP L3 pages read by reload (this rank).", ln + ) + self.l3_gc_reclaimed = Counter( + "sglang:cp_l3_gc_reclaimed_total", "CP L3 pages reclaimed by GC (this rank).", ln + ) + self.l3_write_errors = Counter( + "sglang:cp_l3_write_errors_total", "CP L3 write failures, fail-soft (this rank).", ln + ) + self._last: Dict[str, int] = {} # last cumulative value per counter, for a monotonic .inc(delta) + + def _inc_to(self, counter, key: str, current: int) -> None: + delta = int(current) - self._last.get(key, 0) + if delta > 0: + counter.labels(**self.labels).inc(delta) + self._last[key] = int(current) + + def collect(self, *, l2_occupancy=None, l2_stats=None, l3_stats=None) -> None: + if l2_occupancy: # {payload: (used, total)} + for pk, (used, total) in l2_occupancy.items(): + self.l2_used_pages.labels(**self.labels, payload=pk).set(used) + self.l2_total_pages.labels(**self.labels, payload=pk).set(total) + if l2_stats: + self._inc_to(self.l2_objects_committed, "l2_committed", l2_stats.get("cp_shared_l2_objects_committed", 0)) + self._inc_to(self.l2_objects_evicted, "l2_evicted", l2_stats.get("cp_shared_l2_objects_evicted", 0)) + if l3_stats: + for pk, (used, total) in l3_stats["occupancy"].items(): + self.l3_used_slots.labels(**self.labels, payload=pk).set(used) + self.l3_total_slots.labels(**self.labels, payload=pk).set(total) + self.l3_inflight_spill.labels(**self.labels).set(l3_stats["ongoing_spill"]) + self.l3_inflight_reload.labels(**self.labels).set(l3_stats["ongoing_reload"]) + for q in ("gather", "write", "reload", "touch"): + self.l3_queue_depth.labels(**self.labels, queue=q).set(l3_stats[f"{q}_qsize"]) + self._inc_to(self.l3_spill_pages, "l3_spill", l3_stats["spill_pages_total"]) + self._inc_to(self.l3_reload_pages, "l3_reload", l3_stats["reload_pages_total"]) + self._inc_to(self.l3_gc_reclaimed, "l3_gc", l3_stats["gc_reclaimed_total"]) + self._inc_to(self.l3_write_errors, "l3_werr", l3_stats["write_errors_total"]) + + class ExpertDispatchCollector: def __init__(self, ep_size: int) -> None: from prometheus_client import Histogram diff --git a/test/registered/unit/mem_cache/test_cp_l3_store.py b/test/registered/unit/mem_cache/test_cp_l3_store.py index 93c063e68..c821d7354 100644 --- a/test/registered/unit/mem_cache/test_cp_l3_store.py +++ b/test/registered/unit/mem_cache/test_cp_l3_store.py @@ -272,6 +272,26 @@ class TestCpL3Store(unittest.TestCase): self.assertEqual(store.exists_prefix([_h(4000 + 0)], ["target_kv"]), 1) # touched -> warm -> survived self.assertEqual(store.exists_prefix([_h(4000 + 1)], ["target_kv"]), 0) # next-coldest -> reclaimed + def test_stats_reports_occupancy_and_counters(self): + # stats() snapshot: per-payload slot occupancy + cumulative spill/reload/GC counters + inflight/queues. + pages = {"target_kv": [(2, _h(2)), (5, _h(5))]} + self.store.submit_spill("o", pages, last_access=10) + self.assertEqual(self._finish_spill(), [(1, True)]) + s = self.store.stats() + self.assertEqual(s["occupancy"]["target_kv"], (2, self.store.pools["target_kv"].num_slots)) + self.assertEqual(s["spill_pages_total"], 2) + self.assertEqual(s["gc_reclaimed_total"], 0) + self.assertEqual(s["ongoing_spill"], 0) + self.assertEqual(s["ongoing_reload"], 0) + for k in ("reload_pages_total", "write_errors_total", "gather_qsize", "write_qsize", + "reload_qsize", "touch_qsize"): + self.assertIn(k, s) + # reload the 2 pages -> reload_pages_total advances + self.store.submit_reload("o", pages) + self.assertTrue(_wait_ack(self.store.ack_reload_qsize, 1)) + self.store.drain_reload_acks(self.store.ack_reload_qsize()) + self.assertEqual(self.store.stats()["reload_pages_total"], 2) + def test_clear_resets(self): self.store.submit_spill("o", {"target_kv": [(0, _h(0))]}, last_access=1) self._finish_spill()