From 9368c33b2390996d29a407f3fa0a54aded0d78f0 Mon Sep 17 00:00:00 2001 From: leavelet Date: Sat, 20 Jun 2026 21:24:54 +0000 Subject: [PATCH] L2 prereqs for CP8DP2 + L3: scope HiCache consensus to the CP group + reset pooled allocator on flush Two latent CP shared-KV L2 correctness fixes, landed BEFORE L3 (not in an L3 commit). Surfaced by a per-CP8-instance (CP8DP2EP16) scoping review. PREREQ-1 (CP-group scoping). The B1 commit/evict consensus collectives (writing_check ReduceOp.MIN, placement_digest MIN/MAX + its tp_world_size<=1 entry guard, drain_storage_control_queues, the evict/prefetch MINs, the flush barrier) AND cache_controller.prefetch_tp_group all derive from self.tp_group/self.tp_world_size, which was params.tp_cache_group. tp_cache_group equals the CP group ONLY at dp_size=1 (enable_dp_attention False -> tp_cpu_group, and attn_cp_size==tp_size -> _ATTN_CP==_TP) -- the sole reason B1 works today. Under CP8DP2 (DP attention, attn_tp_size=1) tp_cache_group is the size-1 attn-TP group, so every `tp_world_size>1` collective silently no-ops per rank and the placement assert self-disables -> divergent placement -> shared-slab corruption. Fix: for CP hicache (cp_size>1) scope self.tp_group to the CP cpu group (get_attention_cp_group().cpu_group -- already used for the slab-handle broadcast) + self.tp_world_size to its size. A no-op handle change at dp_size=1 (same group object); the intended fix at CP8DP2. The single init-point change propagates to every consensus collective + un-gates prefetch_tp_group + re-enables the placement assert. PREREQ-2 (flush reset). HiRadixCache.reset() cleared the radix tree + host pool but never reset CpSharedL2PageAllocator -> stale free list/ranges/committed after flush_cache (leak; the shared pool was never reclaimed). Added CpSharedL2PageAllocator.reset() (rebuild the per-slab free list all-free, drop ranges + committed, restore the freshly-built placement_digest) called from cache_controller.reset() after the ack queues are cleared. Safe: flush_cache is idle-gated (no in-flight backup/reserve). This is also L3's clear hookpoint. Validation: new test_reset_restores_freshly_constructed_all_free_state + 89/89 pool suite (torch-2.11 container) + import smoke. PREREQ-1 is a no-op at dp_size=1 (live no-regression confirmed on the next prefill restart); CP8DP2 correctness is by construction (CP-group membership verified) pending a 2-machine run. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../sglang/srt/managers/cache_controller.py | 8 +++++++ .../sglang/srt/mem_cache/cp_shared_l2_pool.py | 20 +++++++++++++++++ python/sglang/srt/mem_cache/hiradix_cache.py | 21 +++++++++++++++++- .../unit/mem_cache/test_cp_shared_l2_pool.py | 22 +++++++++++++++++++ 4 files changed, 70 insertions(+), 1 deletion(-) diff --git a/python/sglang/srt/managers/cache_controller.py b/python/sglang/srt/managers/cache_controller.py index 89d4c088e..46e32f878 100644 --- a/python/sglang/srt/managers/cache_controller.py +++ b/python/sglang/srt/managers/cache_controller.py @@ -867,6 +867,14 @@ class HiCacheController: self.prefetch_thread.start() self.backup_thread.start() + # PREREQ-2: reset the B1 pooled-L2 allocator so flush_cache reclaims the + # shared pool. The radix tree + host pool are cleared by the caller; without + # this the allocator free list / ranges / committed set stay stale (leak). + # Safe here: the ack queues are cleared above and flush_cache is idle-gated + # (no in-flight backup/reserve referencing dropped ranges). + if self.cp_shared_l2_page_allocator is not None: + self.cp_shared_l2_page_allocator.reset() + def clear_draft_host_pool(self) -> None: if self.draft_mem_pool_host is not None: self.draft_mem_pool_host.clear() diff --git a/python/sglang/srt/mem_cache/cp_shared_l2_pool.py b/python/sglang/srt/mem_cache/cp_shared_l2_pool.py index 1a04ae211..5bc11f40c 100644 --- a/python/sglang/srt/mem_cache/cp_shared_l2_pool.py +++ b/python/sglang/srt/mem_cache/cp_shared_l2_pool.py @@ -607,6 +607,26 @@ class CpSharedL2PageAllocator: "cp_shared_l2_objects_evicted": 0, } + def reset(self) -> None: + """Reset to the freshly-constructed all-free state (flush_cache hook). + + Rebuilds the per-slab free list to fully-free and drops all object ranges + + the committed set, so flush_cache reclaims the pooled L2 (the allocator + metadata; the slab bytes are left as-is -- unreferenced once the radix tree + and ranges are cleared). Rank-uniform: every CP rank runs it at the + idle-gated flush point, so the post-reset placement_digest is empty and + identical on every rank. Cumulative stats are preserved. + """ + self._generation = 0 + self._ranges_by_object = {} + self._committed_objects = set() + self._free_by_payload = { + payload_kind: { + slab.slab_id: [(0, slab.num_pages)] for slab in slabs + } + for payload_kind, slabs in self._slabs_by_payload.items() + } + def reserve( self, object_key: str, payload_kind: str, num_pages: int ) -> CpSharedL2ObjectRange: diff --git a/python/sglang/srt/mem_cache/hiradix_cache.py b/python/sglang/srt/mem_cache/hiradix_cache.py index 8f13381e5..ea7f88413 100644 --- a/python/sglang/srt/mem_cache/hiradix_cache.py +++ b/python/sglang/srt/mem_cache/hiradix_cache.py @@ -1015,7 +1015,26 @@ class HiRadixCache(RadixCache): slabs_by_payload=cp_shared_l2_slabs_by_payload, ) - self.tp_group = params.tp_cache_group + # PREREQ-1 (CP8DP2 scoping): the B1 commit/evict consensus collectives + # (writing_check MIN, placement_digest, drain/evict MINs, the prefetch group, + # the flush barrier) MUST reduce over the CP group -- the ranks that share the + # replicated event stream + the MAP_SHARED slab. tp_cache_group equals the CP + # group ONLY at dp_size=1 (CP==TP); under CP8DP2 (DP attention) tp_cache_group + # is the size-1 attn-TP group, so every `tp_world_size > 1` collective would + # silently no-op per rank -> divergent placement -> shared-slab corruption. + # Scope to the CP cpu group (a no-op handle change at dp_size=1). + _cp_hicache_cp_size = ( + int(getattr(params.token_to_kv_pool_allocator, "cp_size", 1) or 1) + if self._uses_cp_hicache + else 1 + ) + if self._uses_cp_hicache and _cp_hicache_cp_size > 1: + from sglang.srt.layers.dp_attention import get_attention_cp_group + + _cp_group = get_attention_cp_group() + self.tp_group = getattr(_cp_group, "cpu_group", _cp_group) + else: + self.tp_group = params.tp_cache_group self.tp_world_size = torch.distributed.get_world_size(group=self.tp_group) self.pp_rank = params.pp_rank self.pp_size = params.pp_size diff --git a/test/registered/unit/mem_cache/test_cp_shared_l2_pool.py b/test/registered/unit/mem_cache/test_cp_shared_l2_pool.py index 20345c08b..2c6d8b246 100644 --- a/test/registered/unit/mem_cache/test_cp_shared_l2_pool.py +++ b/test/registered/unit/mem_cache/test_cp_shared_l2_pool.py @@ -1019,6 +1019,28 @@ class TestCpSharedL2PageAllocator(unittest.TestCase): all(src == 0 and group == "fake-cpu-group" for src, group, _ in broadcasts) ) + def test_reset_restores_freshly_constructed_all_free_state(self): + # PREREQ-2: flush_cache must reclaim the pooled L2. reset() rebuilds the + # free list all-free + drops ranges/committed, restoring the freshly-built + # placement_digest exactly (rank-uniform at the idle flush point). + allocator = self.make_allocator(pages=8) + fresh_digest = allocator.placement_digest() + allocator.reserve("n0", PAYLOAD_TARGET_KV, 3) + allocator.reserve("n1", PAYLOAD_TARGET_KV, 2) + allocator.mark_object_committed("n0") + self.assertLess(allocator.free_pages(PAYLOAD_TARGET_KV), 8) + self.assertNotEqual(allocator.placement_digest(), fresh_digest) + + allocator.reset() + + self.assertEqual(allocator.free_pages(PAYLOAD_TARGET_KV), 8) + self.assertEqual(allocator.free_pages(PAYLOAD_DRAFT_KV), 8) + self.assertFalse(allocator.is_committed("n0")) + self.assertEqual(allocator.object_ranges("n0"), {}) + self.assertEqual(allocator.placement_digest(), fresh_digest) + # usable after reset + self.assertIsNotNone(allocator.reserve("n2", PAYLOAD_TARGET_KV, 4)) + def test_release_and_abort_return_capacity_once_without_double_free(self): allocator = self.make_allocator(pages=5) allocator.reserve("release-me", PAYLOAD_TARGET_KV, 2)