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

L3 3.1: redesign spill — 2-phase staging pipeline + proactive FIFO (fix evicted=0 starvation)

Browse Source 
The first 3.1 (d08ee9f8d) spilled the COLDEST resident objects and held the
protect_host eviction-pin through the entire slow disk write (released only at the
durable ack). Since the coldest objects ARE eviction's victims, a B300 cachebench
wire test wedged: evict-to-fit found everything pinned (evicted=0) ->
host_reservation_failed flood -> caching broke.

Redesign (docs_internal/cp_hicache_l3_phase3_impl_design.md §R3):
- cp_l3_store: split the single spill thread into a GATHER -> WRITE pipeline. Gather
  copies each owned page off the pinned slab into a slab-independent anonymous-mmap
  staging buffer (fast RAM memcpy) and acks GATHER; a separate write thread does the
  O_DIRECT write + fdatasync + durable LMDB put off the staged copy and acks DURABLE.
  The eviction-pin releases at the GATHER ack, so the object is evictable right after
  the RAM copy; the disk write completes off the staged copy even if L2 reuses the
  pages. Slab-full frees its partial slot allocation (no orphan) + acks ok=False.
- hiradix: continuous PROACTIVE spill — enqueue each just-committed object to a FIFO
  deque at the replicated commit frontier (_commit_pending_backup) AND at radix splits
  (the freshly-committed parent half), drained in commit order (not coldest). The deque
  is capped (rank-uniform) so a disk stall can't OOM it. Spilling HOT just-committed
  objects (disjoint from eviction's COLD victims) means the cold tail is already
  L3-durable when evicted -> eviction just drops it, never contends for the pin.
- 3-element CP-group MIN drain [gather, durable, reload]: gather-MIN -> release_host;
  durable-MIN -> l3_durable, gated by an ok-AND (a second MIN over per-op ok bits) so a
  write failure on any rank means NO rank marks it durable (rank-uniform durability,
  placement_digest stays green).

opus-reviewed (FIX-THEN-SHIP) + independently re-verified; review fixes folded:
split-enqueue (HIGH), deque cap for the disk-stall OOM (MED), fail-soft-on-write-failure
documented (HIGH; deliberately NOT re-enqueued — re-gather would churn under slab-full,
recompute is correctness-safe), rate-limited error logs (MED/LOW). 54 L3 unit tests
green (new: 2-phase gather-before-durable ordering, slab-full fail-soft + no slot leak).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
孙远航
2026-06-22 22:15:48 +00:00
parent f353cf4702
commit 815da6c4e5
3 changed files with 351 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

236
python/sglang/srt/mem_cache/cp_l3_store.py
View File

@@ -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

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

@@ -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()