diff --git a/python/sglang/srt/mem_cache/cp_l3_store.py b/python/sglang/srt/mem_cache/cp_l3_store.py index 6e70e9a9b..790328e12 100644 --- a/python/sglang/srt/mem_cache/cp_l3_store.py +++ b/python/sglang/srt/mem_cache/cp_l3_store.py @@ -17,12 +17,15 @@ index's replicated clock. The store is rank-local for I/O (own disk); only the L from __future__ import annotations +import logging import os import queue import threading from dataclasses import dataclass, field from typing import Dict, List, Optional, Sequence, Tuple +logger = logging.getLogger(__name__) + from sglang.srt.mem_cache.cp_l3_config import CpL3Config, probe_disk_plp from sglang.srt.mem_cache.cp_l3_disk import ( CP_L3_HEADER_BYTES, @@ -65,6 +68,37 @@ class _Ack: ok: bool +@dataclass +class _StagedPage: + """One owned page gathered off the (pinned) L2 slab into a slab-independent, write-ready buffer. + + The buffer is an anonymous mmap (its own RAM, not a view into the slab) with the 64B header already + filled, so the disk write can run AFTER the eviction pin is released and the L2 slab is reused.""" + + pk: str + h: bytes # content hash (32B) + crc: int + buf: "object" # mmap.mmap, slot-sized + aligned (CpL3DiskSlab.new_aligned_buffer) + + +@dataclass +class _StagedObject: + """Phase-1 output: an object's owned pages staged in RAM, awaiting the phase-2 disk write.""" + + op_id: int + pages: List[_StagedPage] + last_access: int + failed: bool = False # gather raised (a bug -- the slab is pinned) -> write skips, durable-ack ok=False + + def close(self) -> None: + for p in self.pages: + try: + p.buf.close() + except Exception: + pass + self.pages = [] + + class CpL3Store: def __init__( self, @@ -89,14 +123,23 @@ class CpL3Store: self.payloads: Tuple[str, ...] = tuple(slabs.keys()) self._op_counter = 0 - self._spill_q: "queue.Queue[L3SpillOp]" = queue.Queue() + # Spill is a 2-phase pipeline so the eviction pin releases after the (fast) RAM gather, NOT after the + # (slow) disk write: gather thread copies the pinned slab -> staged buffer + acks GATHER; the write + # thread drains the staged objects (O_DIRECT + fdatasync + durable index put) + acks DURABLE. Both are + # single FIFO threads, so each ack stream is in submit order -> the caller's cross-rank MIN-drain frees + # the same op_ids on every rank (the placement-replication invariant). + self._gather_q: "queue.Queue[L3SpillOp]" = queue.Queue() + self._write_q: "queue.Queue[_StagedObject]" = queue.Queue() self._reload_q: "queue.Queue[L3ReloadOp]" = queue.Queue() - self._ack_spill_q: "queue.Queue[_Ack]" = queue.Queue() + self._ack_gather_q: "queue.Queue[int]" = queue.Queue() # op_id; gather done -> caller release_host + self._ack_durable_q: "queue.Queue[_Ack]" = queue.Queue() # (op_id, ok); durable -> caller l3_durable self._ack_reload_q: "queue.Queue[_Ack]" = queue.Queue() - self._ongoing_spill: set = set() + self._ongoing_spill: set = set() # submitted .. durable-acked (the inflight-budget basis + idle gate) self._ongoing_reload: set = set() + self._write_err_count = 0 # rate-limit the write-error log (slab-full pre-3.3 streams one/commit) self._stop = threading.Event() - self._spill_thread: Optional[threading.Thread] = None + self._gather_thread: Optional[threading.Thread] = None + self._write_thread: Optional[threading.Thread] = None self._reload_thread: Optional[threading.Thread] = None self._started = False @@ -153,9 +196,11 @@ class CpL3Store: if self._started: return self._stop.clear() - self._spill_thread = threading.Thread(target=self._spill_loop, daemon=True, name="cp-l3-spill") + self._gather_thread = threading.Thread(target=self._gather_loop, daemon=True, name="cp-l3-gather") + self._write_thread = threading.Thread(target=self._write_loop, daemon=True, name="cp-l3-write") self._reload_thread = threading.Thread(target=self._reload_loop, daemon=True, name="cp-l3-reload") - self._spill_thread.start() + self._gather_thread.start() + self._write_thread.start() self._reload_thread.start() self._started = True @@ -171,8 +216,8 @@ class CpL3Store: def submit_spill(self, object_key: str, owned_pages: Dict[str, List[OwnedPage]], last_access: int) -> int: op = L3SpillOp(self._next_op_id(), object_key, owned_pages, last_access) - self._ongoing_spill.add(op.op_id) - self._spill_q.put(op) + self._ongoing_spill.add(op.op_id) # cleared only at the durable-ack (gather-ack just releases the pin) + self._gather_q.put(op) return op.op_id def submit_reload(self, object_key: str, owned_pages: Dict[str, List[OwnedPage]]) -> int: @@ -185,17 +230,35 @@ class CpL3Store: return bool(self._ongoing_spill) or bool(self._ongoing_reload) # ---------- completion harvest (scheduler thread; caller does the CP-group MIN) ---------- - def ack_spill_qsize(self) -> int: - return self._ack_spill_q.qsize() + def ack_gather_qsize(self) -> int: + return self._ack_gather_q.qsize() + + def ack_durable_qsize(self) -> int: + return self._ack_durable_q.qsize() def ack_reload_qsize(self) -> int: return self._ack_reload_q.qsize() - def drain_spill_acks(self, n: int) -> List[Tuple[int, bool]]: + def drain_gather_acks(self, n: int) -> List[int]: + """Pop n gather-done op_ids FIFO. The caller releases the eviction pin for each: the staged copy is + slab-independent, so the object may be evicted from L2 now; the disk write still completes off the + staged copy. Does NOT clear _ongoing_spill -- the op stays inflight (budget + idle gate) until durable.""" out = [] for _ in range(n): try: - ack = self._ack_spill_q.get_nowait() + out.append(self._ack_gather_q.get_nowait()) + except queue.Empty: + break + return out + + def drain_durable_acks(self, n: int) -> List[Tuple[int, bool]]: + """Pop n durable op_ids FIFO as (op_id, ok). ok=False => the disk write / index put failed (e.g. the + slab filled before 3.3 evict-to-fit lands); the caller must NOT mark it L3-durable (fail-soft: the + object is recomputed on a future miss). Clears _ongoing_spill (the op is done, success or not).""" + out = [] + for _ in range(n): + try: + ack = self._ack_durable_q.get_nowait() except queue.Empty: break self._ongoing_spill.discard(ack.op_id) @@ -214,21 +277,53 @@ class CpL3Store: return out # ---------- background workers ---------- - def _spill_loop(self) -> None: + def _gather_loop(self) -> None: + """Phase 1 (fast): copy each owned page off the PINNED L2 slab into a slab-independent staged buffer, + ack GATHER (-> the caller releases the eviction pin), and hand the staged object to the write thread. + A pure RAM memcpy off a pinned slab: a failure here is a bug -> log loud, ack ok=False at durable, but + STILL release the pin (push the gather-ack) + survive the thread (never wedge the spill pipeline).""" while not self._stop.is_set(): try: - op = self._spill_q.get(block=True, timeout=0.2) + op = self._gather_q.get(block=True, timeout=0.2) except queue.Empty: continue - ok = True try: - self._spill_object(op) - except Exception as exc: # never kill the thread; report failure -> caller keeps L2 (no defer-free) - ok = False - import logging + staged = self._gather_object(op) + except Exception as exc: + # gather is a pure RAM memcpy off a PINNED slab -> a failure is a real bug, not backpressure + # (so it won't flood): log loud, fail the op soft (no wedge). + logger.error( + "L3 gather op %s failed (BUG: the slab is pinned during gather): %r", op.op_id, exc + ) + staged = _StagedObject(op.op_id, [], op.last_access, failed=True) + self._write_q.put(staged) + self._ack_gather_q.put(op.op_id) # release the pin regardless: the slab read is finished/aborted - logging.getLogger(__name__).error("L3 spill op %s failed: %r", op.op_id, exc) - self._ack_spill_q.put(_Ack(op.op_id, ok)) + def _write_loop(self) -> None: + """Phase 2 (slow, off the pin): write each staged page (O_DIRECT) + fdatasync + durable index put, + then ack DURABLE. A failure (slab full pre-3.3, or an I/O error) acks ok=False -> the caller keeps it + non-durable (fail-soft: recompute on a future miss); the thread survives. The staged buffers are freed + here (success or fail).""" + while not self._stop.is_set(): + try: + staged = self._write_q.get(block=True, timeout=0.2) + except queue.Empty: + continue + ok = not staged.failed + if not staged.failed: + try: + self._write_object(staged) + except Exception as exc: # never kill the thread; report failure -> caller keeps it non-durable + ok = False + # slab-full (pre-3.3) streams one failure per new commit -> rate-limit to avoid flooding. + self._write_err_count += 1 + if self._write_err_count == 1 or self._write_err_count % 256 == 0: + logger.error( + "L3 write op %s failed (%d total; slab full? 3.3 disk-eviction reduces this): %r", + staged.op_id, self._write_err_count, exc, + ) + staged.close() + self._ack_durable_q.put(_Ack(staged.op_id, ok)) def _reload_loop(self) -> None: while not self._stop.is_set(): @@ -246,38 +341,70 @@ class CpL3Store: logging.getLogger(__name__).error("L3 reload op %s failed: %r", op.op_id, exc) self._ack_reload_q.put(_Ack(op.op_id, ok)) - def _spill_object(self, op: L3SpillOp) -> None: - """Gather+write this rank's owned pages of the object, then durably commit their index entries. - Zero-owned ranks no-op (the ack alone keeps the MIN-drain in lockstep).""" - touched: set = set() - with self.index.write_batch() as wb: - for pk, pages in op.owned_pages.items(): - if not pages: + def _gather_object(self, op: L3SpillOp) -> "_StagedObject": + """Phase 1: gather this rank's owned, not-yet-durable pages off the pinned slab into write-ready staged + buffers (header filled, CRC computed). Content-hash dedup skips pages already in L3. A zero-owned rank + (or a fully-deduped object) stages zero pages -- the durable-ack alone keeps the MIN-drain in lockstep. + On any error the partial buffers are closed before re-raising (no mmap leak).""" + pages: List[_StagedPage] = [] + try: + for pk, owned in op.owned_pages.items(): + if not owned: continue - slab, pool, accessor = self.slabs[pk], self.pools[pk], self.accessors[pk] - for (slab_page, h_hex) in pages: + slab, accessor = self.slabs[pk], self.accessors[pk] + for (slab_page, h_hex) in owned: h = content_hash_to_bytes(h_hex) if self.index.exists(h, pk): continue # content-hash dedup: already durable in L3 - slot = pool.alloc() - if slot < 0: - _l3_failfast( - f"L3 {pk} slab full on rank {self.cp_rank} (evict-to-fit must run before spill)" - ) buf = slab.new_aligned_buffer() - try: - accessor.gather_into(slab_page, buf, CP_L3_HEADER_BYTES) - crc = compute_crc(memoryview(buf)[CP_L3_HEADER_BYTES:CP_L3_HEADER_BYTES + slab.page_bytes]) - slab.fill_header(buf, h, accessor.n_layers, crc) - slab.write_aligned(slot, buf) - finally: - buf.close() - wb.put(h, pk, CpL3IndexEntry( - disk_id=self.disk_id, file_id=0, slot_idx=slot, page_bytes=slab.page_bytes, - crc=crc, last_access=op.last_access, hit_count=0, - )) - touched.add(pk) - # data -> fdatasync (durable) -> index commit (on __exit__, durable) : the crash-safe ordering + accessor.gather_into(slab_page, buf, CP_L3_HEADER_BYTES) + crc = compute_crc( + memoryview(buf)[CP_L3_HEADER_BYTES:CP_L3_HEADER_BYTES + slab.page_bytes] + ) + slab.fill_header(buf, h, accessor.n_layers, crc) + pages.append(_StagedPage(pk=pk, h=h, crc=crc, buf=buf)) + except Exception: + for p in pages: + try: + p.buf.close() + except Exception: + pass + raise + return _StagedObject(op.op_id, pages, op.last_access) + + def _write_object(self, staged: "_StagedObject") -> None: + """Phase 2: write the staged pages durably (off the released pin). Allocates ALL disk slots first so a + slab-full leaves no orphan blob (frees the slots it took + raises -> fail-soft until 3.3 evict-to-fit); + then writes, fdatasyncs the data, and commits the index -- the crash-safe data->fsync->index ordering.""" + if not staged.pages: + return # zero-owned / fully-deduped: durable via the other ranks; nothing to write here + to_write: List[Tuple[_StagedPage, int]] = [] + try: + for sp in staged.pages: + if self.index.exists(sp.h, sp.pk): + continue # raced durable since gather (another op/rank) -> dedup + slot = self.pools[sp.pk].alloc() + if slot < 0: + _l3_failfast( + f"L3 {sp.pk} slab full on rank {self.cp_rank} (3.3 evict-to-fit must run before spill)" + ) + to_write.append((sp, slot)) + except Exception: + for (sp, slot) in to_write: + self.pools[sp.pk].free(slot) # no orphan disk blob on slab-full + raise + if not to_write: + return + touched: set = set() + with self.index.write_batch() as wb: + for (sp, slot) in to_write: + self.slabs[sp.pk].write_aligned(slot, sp.buf) + wb.put(sp.h, sp.pk, CpL3IndexEntry( + disk_id=self.disk_id, file_id=0, slot_idx=slot, page_bytes=self.slabs[sp.pk].page_bytes, + crc=sp.crc, last_access=staged.last_access, hit_count=0, + )) + touched.add(sp.pk) + # data -> fdatasync (durable) -> index commit (write_batch __exit__, durable): crash-safe ordering for pk in touched: self.slabs[pk].fdatasync() # (write_batch __exit__ committed the index durably here) @@ -323,7 +450,16 @@ class CpL3Store: def clear(self) -> None: """flush_cache hook: stop+drain threads, reset slot pools + index, restart. Idle-gated by the caller.""" self._stop_threads() - for q in (self._spill_q, self._reload_q, self._ack_spill_q, self._ack_reload_q): + # close any staged buffers still queued for the write thread (avoid mmap leak) before clearing queues + while True: + try: + self._write_q.get_nowait().close() + except queue.Empty: + break + for q in ( + self._gather_q, self._write_q, self._ack_gather_q, self._ack_durable_q, + self._reload_q, self._ack_reload_q, + ): with q.mutex: q.queue.clear() self._ongoing_spill.clear() @@ -337,7 +473,7 @@ class CpL3Store: if not self._started: return self._stop.set() - for t in (self._spill_thread, self._reload_thread): + for t in (self._gather_thread, self._write_thread, self._reload_thread): if t is not None: t.join(timeout=3) self._started = False diff --git a/python/sglang/srt/mem_cache/hiradix_cache.py b/python/sglang/srt/mem_cache/hiradix_cache.py index a199edcd0..4270dd58f 100644 --- a/python/sglang/srt/mem_cache/hiradix_cache.py +++ b/python/sglang/srt/mem_cache/hiradix_cache.py @@ -1,6 +1,7 @@ from __future__ import annotations import atexit +import collections import heapq import json import logging @@ -1045,7 +1046,17 @@ class HiRadixCache(RadixCache): # pooled shared-L2 (server_args validation already enforced the prerequisites + mutual exclusion). self.enable_cp_l3 = bool(getattr(server_args, "enable_cp_l3", False)) and self._uses_cp_hicache self.cp_l3_store = None # constructed after the host pool + allocator (slab accessors need them) - self.ongoing_l3_spill = {} # op_id -> pinned (protect_host'd) resident TreeNode being copied to L3 + self.ongoing_l3_spill = {} # op_id -> pinned (protect_host'd) resident TreeNode; submitted..durable-acked + # PROACTIVE spill: FIFO of (object_key, node) enqueued at the replicated commit frontier + # (_commit_pending_backup) + at radix splits, drained by the maintainer in commit order -> L3 stays + # caught up with L2. Capped: if the disk stalls, the bg write thread blocks -> ongoing pins at + # max_inflight -> budget hits 0 -> the maintainer stops popping; without a cap the deque would then grow + # unbounded with every new commit (OOM). The cap is a replicated constant (rank-uniform drop) sized well + # above the max committed-object backlog, so it only fires under a genuine disk stall (fail-soft: the + # dropped objects recompute on a miss). drop count is for a rate-limited warn. + self.cp_l3_spill_pending = collections.deque() + self.cp_l3_spill_pending_cap = 1 << 18 # 262144 entries (~tens of MB of (str,node) refs) + self._cp_l3_spill_dropped = 0 # backpressure: cap concurrent background spills (bounds the bg queue + staging RAM, rank-uniform) self.cp_l3_spill_max_inflight = 32 @@ -1265,39 +1276,79 @@ class HiRadixCache(RadixCache): return mapping.mmap def _drain_l3_control_queues(self) -> None: - """Per-tick CP-cpu-group MIN over the L3 ack-queue sizes, then drain MIN of each (R1 CRIT-3/4). - Idle early-return: ongoing_spill/reload is a replicated quantity (3.1+ submits are lockstep on the - replicated plan), so an idle tick issues no collective. 3.0: always idle -> inert (3.1/3.2 act on acks).""" + """Per-tick CP-cpu-group MIN over the three L3 ack-queue sizes [gather, durable, reload], then drain + MIN of each: gather-ack -> release the eviction pin (phase 1), durable-ack -> mark l3_durable under an + ok-AND (phase 2), reload-ack -> 3.2. Idle early-return: has_inflight() is a replicated quantity (submits + are lockstep on the replicated plan), so an idle tick issues no collective.""" store = self.cp_l3_store if store is None or not store.has_inflight(): return qsizes = torch.tensor( - [store.ack_spill_qsize(), store.ack_reload_qsize()], dtype=torch.int + [store.ack_gather_qsize(), store.ack_durable_qsize(), store.ack_reload_qsize()], dtype=torch.int ) if self.tp_world_size > 1: self._cp_hicache_all_reduce( qsizes, op=torch.distributed.ReduceOp.MIN, group=self.tp_group, tag="l3_queue_min" ) - n_spill, n_reload = map(int, qsizes.tolist()) - for op_id, _ok in store.drain_spill_acks(n_spill): - # spill done (ok or not): unpin the resident node. ok=True -> object now L3-durable + L2-resident; - # ok=False -> the copy failed, just unpin (the maintainer retries it next pass). No allocator - # release here (approach D: the object was never freed -- it's a copy, not a move). - node = self.ongoing_l3_spill.pop(op_id, None) - if node is not None: - if _ok: - node.l3_durable = True # now reloadable from L3; maintainer skips it (O(1)) + n_gather, n_durable, n_reload = map(int, qsizes.tolist()) + # Phase 1 -- GATHER done: release the eviction pin. The staged copy is slab-independent, so the object + # may be evicted from L2 now; the disk write still completes off the staged copy. The node stays in + # ongoing_l3_spill (inflight, counts toward budget) until the durable ack. Pin release is MIN-gated, so + # it is rank-uniform -- a divergent unpin would drift the shared-slab eviction order (placement_digest). + for op_id in store.drain_gather_acks(n_gather): + node = self.ongoing_l3_spill.get(op_id) + if node is not None and int(getattr(node, "host_ref_counter", 0) or 0) > 0: node.release_host() + # Phase 2 -- DURABLE: mark L3-durable, but ONLY if EVERY rank's write succeeded (rank-uniform durability + # via an ok-AND, i.e. ReduceOp.MIN over the per-op ok bits). A write failure on any rank (e.g. slab full + # pre-3.3) => no rank marks it durable => fail-soft recompute, never a divergent l3_durable that would + # split eviction eligibility. The acks are FIFO in replicated submit order, so the ok tensor aligns + # per-op across ranks. No allocator release here (approach D: it was a copy, not a move). + # A failed object is deliberately NOT re-enqueued: ok=False means slab-full (until 3.3 disk-eviction) + # or a disk I/O error -- an immediate retry would just re-gather + re-fail every tick (churn strictly + # worse than the benign abandon). It is retried on its next natural re-commit; 3.3 removes the slab-full + # cause. Correctness is unaffected (a non-durable object simply recomputes on a future miss). + durable = store.drain_durable_acks(n_durable) + if durable: + oks = torch.tensor([1 if ok else 0 for (_id, ok) in durable], dtype=torch.int) + if self.tp_world_size > 1: + self._cp_hicache_all_reduce( + oks, op=torch.distributed.ReduceOp.MIN, group=self.tp_group, tag="l3_durable_ok" + ) + for (op_id, _ok), agreed in zip(durable, oks.tolist()): + node = self.ongoing_l3_spill.pop(op_id, None) + if node is not None and int(agreed) == 1: + node.l3_durable = True # now reloadable from L3; maintainer skips it (O(1)) store.drain_reload_acks(n_reload) # 3.2 wires reload-ack admission + pin release + def _cp_l3_enqueue_spill(self, object_key: str, node: TreeNode) -> None: + """Enqueue a committed object for proactive spill (rank-uniform: called only from the replicated + commit frontier + radix split). Drops (fail-soft) past the cap -- only reachable under a disk stall + (the deque can't drain), so the drop bounds RAM; the object recomputes on a future miss. len() is O(1) + and rank-uniform (the deque fills + drains identically on every rank), so the cap decision can't diverge.""" + if len(self.cp_l3_spill_pending) < self.cp_l3_spill_pending_cap: + self.cp_l3_spill_pending.append((object_key, node)) + return + self._cp_l3_spill_dropped += 1 + if self._cp_l3_spill_dropped == 1 or self._cp_l3_spill_dropped % 1024 == 0: + logger.warning( + "[CP-L3] spill backlog at cap %d (disk falling behind); dropped %d spill candidates " + "(fail-soft: they recompute on miss). 3.3 disk-eviction reduces this.", + self.cp_l3_spill_pending_cap, self._cp_l3_spill_dropped, + ) + def _cp_l3_spill_maintainer(self) -> None: - """Background spill (approach D, 3.1): COPY the coldest resident committed objects (replicated SLRU - order) that are not yet in L3 to the disk tier, off the critical path. The object stays L2-resident; - protect_host pins the LIVE node during the async copy (so the slab can't be evicted underneath the bg - gather). Rank-synced by construction: selection rides replicated state, so all ranks submit the SAME - objects in lockstep -> object-granular acks + the MIN-drain stay aligned. Bounded per tick + backpressured.""" + """Continuous PROACTIVE spill (3.1): drain the FIFO of just-committed objects (filled at the replicated + commit frontier, _commit_pending_backup) -- NOT a coldest-scan. Each still-resident, not-yet-durable, + unpinned object is submitted as a background COPY to L3; protect_host pins the LIVE node only until the + RAM gather completes (released at the gather-ack), the slow disk write runs off the slab-independent + staged copy. Because we spill HOT just-committed objects (disjoint from eviction's COLD victims), the + cold tail is ALREADY L3-durable when evicted -> eviction just drops it, never contends for the pin. + Rank-synced by construction: the FIFO is filled in the rank-uniform MIN-frontier order, so all ranks + submit the SAME objects in lockstep -> object-granular acks + the MIN-drain stay aligned. Bounded per + tick by the inflight budget; budget-limited entries stay queued (drained next tick).""" store = self.cp_l3_store - if store is None: + if store is None or not self.cp_l3_spill_pending: return budget = self.cp_l3_spill_max_inflight - len(self.ongoing_l3_spill) if budget <= 0: @@ -1305,31 +1356,33 @@ class HiRadixCache(RadixCache): allocator = getattr(self.cache_controller, "cp_shared_l2_page_allocator", None) if allocator is None: return - # coldest-first = the eviction order, so we make objects durable just before they'd be evicted. - candidates = sorted(self.evictable_host_leaves, key=self._cp_host_evict_key) + pending = self.cp_l3_spill_pending submitted = 0 - for node in candidates: - if submitted >= budget: - break + while pending and submitted < budget: + object_key, node = pending.popleft() + # Every non-submittable state below is TERMINAL -> drop the entry (popped). Only the budget limit + # (loop guard) leaves entries queued. Under the current CP+L3 config host_ref_counter is set ONLY by + # our L3 pin (radix_cache:248) -- storage prefetch/backup is mutually-exclusive (server_args) and + # pin_prefix is a CP no-op (pin_expiry stays 0) -- so >0 here means an in-flight spill of this same + # node -> it will become durable; drop. (Revisit this drop rule if either exclusion is relaxed.) if getattr(node, "l3_durable", False): - continue # O(1): already copied to L3 (set on ack / memoized below) + continue # already durable in L3 meta = getattr(node, "cp_hicache", None) if meta is None or not self._is_cp_shared_l2_metadata(meta): - continue + continue # evicted / rolled back since commit if int(getattr(node, "host_ref_counter", 0) or 0) > 0: - continue # pinned (already spilling / protected) -> skip (excludes in-flight from re-select) - object_key = getattr(meta, "object_key", "") + continue # already being spilled (our pin) -> the in-flight op makes it durable if not object_key or not allocator.is_committed(object_key): - continue # M-1: only committed objects (owned-page D2H complete) + continue # rolled back / no longer a committed pool object page_keys = getattr(node, "hash_value", None) if not page_keys: - continue + continue # no hash chain (shouldn't happen under enable_cp_l3 hash-from-init) required = tuple(getattr(meta, "required_payloads", ())) if store.exists_prefix(page_keys, required) >= len(page_keys): - node.l3_durable = True # memoize (e.g. a reloaded-from-L3 node) -> O(1) skip next tick - continue # already fully durable in L3 (dedup) + node.l3_durable = True # already fully durable (e.g. a reloaded-from-L3 node) -> memoize O(1) + continue owned_pages = self._cp_l3_build_owned_pages(meta, page_keys, required, store) - last_access = int(getattr(node, "last_access_time", 0) or 0) + last_access = int(getattr(node, "last_access_time", 0) or 0) # rank-replicated logical clock node.protect_host() op_id = store.submit_spill(object_key, owned_pages, last_access) self.ongoing_l3_spill[op_id] = node @@ -2880,6 +2933,8 @@ class HiRadixCache(RadixCache): if self.cp_l3_store is not None: # flush_cache is idle-gated -> no in-flight L3 ops; clear() stops/drains/resets/restarts. self.cp_l3_store.clear() + self.cp_l3_spill_pending.clear() # drop queued (now-flushed) spill candidates + self.ongoing_l3_spill.clear() self.token_to_kv_pool_host.clear() self.cache_controller.clear_draft_host_pool() # Clear per-request tracking dicts @@ -3009,6 +3064,11 @@ class HiRadixCache(RadixCache): object_key = self.cache_controller._cp_shared_l2_object_key(node_id) if allocator.object_ranges(object_key): allocator.mark_object_committed(object_key) + if self.enable_cp_l3: + # 3.1 proactive spill: enqueue at the replicated commit frontier (writing_check's MIN + # gates this transition rank-uniformly), so all ranks queue the SAME objects in the SAME + # order. The maintainer drains FIFO + validates each entry (a since-evicted node is dropped). + self._cp_l3_enqueue_spill(object_key, node) return node def _remove_undrained_write_acks(self, node_id: int) -> bool: @@ -5591,6 +5651,16 @@ class HiRadixCache(RadixCache): # evict-subtraction uses the correct per-half counts. self._cp_account_enter_committed(new_node.cp_hicache) self._cp_account_enter_committed(child.cp_hicache) + if self.enable_cp_l3: + # 3.1 proactive spill: the parent half (new_node) is a FRESH committed object + # (parent_object_key) that the commit frontier never sees -> enqueue it here, else its + # pages would never become L3-durable (the durable prefix would be holed after a split). + # The child half KEEPS the original object_key (split: child_key=_old_cp_meta.object_key) + # so its existing FIFO entry stays valid -- don't double-enqueue it. Rank-uniform: splits + # replay identically on every rank (replicated radix), so new_node.id/parent_object_key + # and the enqueue order match across ranks. split_committed_object marked both keys + # committed, so the maintainer's is_committed(parent_object_key) check passes. + self._cp_l3_enqueue_spill(parent_object_key, new_node) elif child.backuped: new_node.host_value = child.host_value[:split_len].clone() child.host_value = child.host_value[split_len:].clone() 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 e5179c33a..e09bfe13e 100644 --- a/test/registered/unit/mem_cache/test_cp_l3_store.py +++ b/test/registered/unit/mem_cache/test_cp_l3_store.py @@ -93,15 +93,30 @@ class TestCpL3Store(unittest.TestCase): self.store.close() self._td.cleanup() + def _finish_spill(self): + """Drive the 2-phase spill to completion (store-level; no cross-rank MIN). Returns durable acks + as [(op_id, ok), ...]. Draining the gather-ack is what releases the eviction pin in production.""" + self.assertTrue(_wait_ack(self.store.ack_durable_qsize, 1)) + self.store.drain_gather_acks(self.store.ack_gather_qsize()) + return self.store.drain_durable_acks(self.store.ack_durable_qsize()) + + def test_two_phase_gather_precedes_durable(self): + # the core 3.1 fix: gather (pin release) acks BEFORE the slow durable write; inflight until durable. + op_id = self.store.submit_spill("obj-2p", {"target_kv": [(0, _h(0))]}, last_access=1) + self.assertTrue(_wait_ack(self.store.ack_durable_qsize, 1)) # durable implies gather already acked + self.assertGreaterEqual(self.store.ack_gather_qsize(), 1) + self.assertEqual(self.store.drain_gather_acks(self.store.ack_gather_qsize()), [op_id]) + self.assertTrue(self.store.has_inflight()) # gather drained, but still inflight until durable + self.assertEqual(self.store.drain_durable_acks(self.store.ack_durable_qsize()), [(op_id, True)]) + self.assertFalse(self.store.has_inflight()) + def test_spill_exists_reload_roundtrip(self): # spill pages 2 and 5 pages = {"target_kv": [(2, _h(2)), (5, _h(5))]} orig2 = self.acc.gather(2) orig5 = self.acc.gather(5) - self.store.submit_spill("obj-A", pages, last_access=10) - self.assertTrue(_wait_ack(self.store.ack_spill_qsize, 1)) - acks = self.store.drain_spill_acks(self.store.ack_spill_qsize()) - self.assertEqual(acks, [(1, True)]) + op_id = self.store.submit_spill("obj-A", pages, last_access=10) + self.assertEqual(self._finish_spill(), [(op_id, True)]) self.assertFalse(self.store.has_inflight()) # exists_prefix: consecutive present count @@ -125,38 +140,69 @@ class TestCpL3Store(unittest.TestCase): self.assertEqual(self.acc.gather(5), orig5) def test_zero_owned_spill_acks_in_lockstep(self): - # a rank that owns no pages of the object still must ack (empty op) + # a rank that owns no pages of the object still must ack (empty op) -- both phases op_id = self.store.submit_spill("obj-empty", {"target_kv": []}, last_access=1) - self.assertTrue(_wait_ack(self.store.ack_spill_qsize, 1)) - self.assertEqual(self.store.drain_spill_acks(1), [(op_id, True)]) + self.assertEqual(self._finish_spill(), [(op_id, True)]) + self.assertFalse(self.store.has_inflight()) def test_dedup_skip_existing(self): pages = {"target_kv": [(3, _h(3))]} self.store.submit_spill("o1", pages, last_access=1) - self.assertTrue(_wait_ack(self.store.ack_spill_qsize, 1)) - self.store.drain_spill_acks(1) + self._finish_spill() free_after_first = self.store.pools["target_kv"].num_free # spill the SAME hash again -> dedup, no new slot consumed self.store.submit_spill("o1b", pages, last_access=2) - self.assertTrue(_wait_ack(self.store.ack_spill_qsize, 1)) - self.store.drain_spill_acks(1) + self._finish_spill() self.assertEqual(self.store.pools["target_kv"].num_free, free_after_first) def test_free_object_releases_slots_and_index(self): pages = {"target_kv": [(1, _h(1)), (4, _h(4))]} before = self.store.pools["target_kv"].num_free self.store.submit_spill("o", pages, last_access=1) - self.assertTrue(_wait_ack(self.store.ack_spill_qsize, 1)) - self.store.drain_spill_acks(1) + self._finish_spill() self.assertEqual(self.store.pools["target_kv"].num_free, before - 2) self.store.free_object(pages) self.assertEqual(self.store.pools["target_kv"].num_free, before) self.assertEqual(self.store.exists_prefix([_h(1)], ["target_kv"]), 0) + def test_slab_full_fail_soft_frees_partial_alloc(self): + # _write_object allocs ALL of an object's slots before writing; a slab-full mid-object must free the + # slots it already took (no orphan) and ack durable ok=False (fail-soft, not a crash). Tiny budget. + td = tempfile.TemporaryDirectory() + self.addCleanup(td.cleanup) + cfg = cfg_mod.CpL3Config.from_dict({ + "backend": "posix", "require_plp": False, "index_map_gb": 0.05, + "disks": [{"path": os.path.join(td.name, "d"), "budget_gb": 0.00002}], + }) + store = store_mod.CpL3Store.from_config( + cfg, cp_rank=0, cp_size=1, accessors={"target_kv": self.acc}) + store.connect(cfg) + self.addCleanup(store.close) + pool = store.pools["target_kv"] + n = pool.num_slots + self.assertGreaterEqual(n, 2) + op = 0 + for i in range(n - 1): # fill all but one slot (single-page objects, distinct hashes) + op += 1 + store.submit_spill(f"o{i}", {"target_kv": [(i % PAGE_NUM, _h(2000 + i))]}, last_access=i) + self.assertTrue(_wait_ack(store.ack_durable_qsize, 1)) + store.drain_gather_acks(store.ack_gather_qsize()) + self.assertEqual(store.drain_durable_acks(store.ack_durable_qsize()), [(op, True)]) + self.assertEqual(pool.num_free, 1) + # 2-distinct-page object with only 1 free slot: page-1 allocs, page-2 fails -> whole op ok=False, + # and the page-1 slot is freed (num_free back to 1, no leak). + op += 1 + store.submit_spill( + "overflow", {"target_kv": [(0, _h(8001)), (1, _h(8002))]}, last_access=op) + self.assertTrue(_wait_ack(store.ack_durable_qsize, 1)) + store.drain_gather_acks(store.ack_gather_qsize()) + self.assertEqual(store.drain_durable_acks(store.ack_durable_qsize()), [(op, False)]) + self.assertEqual(pool.num_free, 1) # no orphan: the partial alloc was rolled back + self.assertFalse(store.has_inflight()) + def test_clear_resets(self): self.store.submit_spill("o", {"target_kv": [(0, _h(0))]}, last_access=1) - self.assertTrue(_wait_ack(self.store.ack_spill_qsize, 1)) - self.store.drain_spill_acks(1) + self._finish_spill() self.store.clear() self.assertEqual(self.store.exists_prefix([_h(0)], ["target_kv"]), 0) self.assertEqual(self.store.pools["target_kv"].num_free, self.store.pools["target_kv"].num_slots)