From 04c6793ca4f0af5730788b9b4ba12c203db613d3 Mon Sep 17 00:00:00 2001 From: leavelet Date: Sat, 20 Jun 2026 23:05:55 +0000 Subject: [PATCH] L3 3.0: on-disk blob format + disk-slot free-list allocator Phase 3.0 (L3 disk durable floor) foundational primitives, pure + unit-tested: - 64B self-describing blob header (magic/version/payload-kind/content-hash/CRC), verified on every read (verify_blob) -> torn/bit-rot slots become a MISS, not garbage. - slot_bytes_for_page: header+payload rounded to the 4K O_DIRECT boundary. - CpL3SlotPool: O(1) free-list over fixed-size per-payload disk page-slots (the L3 analog of CpSharedL2PageAllocator's free list), fail-loud on double-free/OOB, with reset() for the flush_cache hook. NOT a ring buffer (eviction frees arbitrary slots). Rank-local (a rank stores only its owned pages; L3 eviction selection rides the shared LMDB replicated clock). 12/12 unit tests. Design: docs_internal/cp_hicache_l3_phase3_impl_design.md. Co-Authored-By: Claude Opus 4.8 (1M context) --- python/sglang/srt/mem_cache/cp_l3_disk.py | 218 ++++++++++++++++++ .../unit/mem_cache/test_cp_l3_disk.py | 140 +++++++++++ 2 files changed, 358 insertions(+) create mode 100644 python/sglang/srt/mem_cache/cp_l3_disk.py create mode 100644 test/registered/unit/mem_cache/test_cp_l3_disk.py diff --git a/python/sglang/srt/mem_cache/cp_l3_disk.py b/python/sglang/srt/mem_cache/cp_l3_disk.py new file mode 100644 index 000000000..b31fa8931 --- /dev/null +++ b/python/sglang/srt/mem_cache/cp_l3_disk.py @@ -0,0 +1,218 @@ +# Copyright 2023-2024 SGLang Team +# Licensed under the Apache License, Version 2.0 +"""CP HiCache L3 (disk durable floor) — on-disk primitives. + +Pure, backend-agnostic building blocks for the L3 disk tier (Phase 3.0): + * the self-describing per-page **blob header** (CRC + content hash, verified on every read), and + * the **disk-slot free-list allocator** (``CpL3SlotPool``) — O(1) alloc/free of fixed-size page slots, + the L3 analog of ``CpSharedL2PageAllocator``'s free list. + +A page is stored on disk as one *contiguous* blob (header + the gathered, layer-major page payload), even +though the host slab is ``layer_page_first`` (layer-strided) — the spill writer gathers the per-layer slices +into a contiguous buffer (measured faster than vectored I/O under O_DIRECT). Slots are fixed-size **per +payload kind** so allocation is O(1) and fragmentation-free. Each ``(payload_kind, drive)`` owns one +``CpL3SlotPool`` + one preallocated disk-slab file; pools are **local to each CP rank** (a rank stores only its +owned pages — ``owner=(page-1)%cp_size``), so slot indices need no cross-rank agreement (L3 *eviction +selection*, which must be replicated, rides the shared LMDB index's replicated clock — see the impl design). +""" + +from __future__ import annotations + +import struct +import zlib +from typing import Tuple + +CP_L3_FAILFAST_PREFIX = "[CP_L3_FAILFAST]" + +# Disk/O_DIRECT alignment: slot byte sizes + offsets are rounded up to this so a single O_DIRECT +# read/write of a whole slot uses one aligned buffer (the header lives at offset 0, payload at +# HEADER_BYTES, trailing pad to the boundary). 4 KiB covers ext4 on 4K-sector NVMe. +CP_L3_DIO_ALIGN = 4096 + +# --- blob header: 64 bytes, self-describing, CRC verified on read --- +# magic u32 | version u16 | payload_kind_id u8 | flags u8 | content_hash[32] | n_layers u16 | +# page_bytes u32 | crc32 u32 | 14x pad == 4+2+1+1+32+2+4+4 = 50, padded to 64. +CP_L3_BLOB_MAGIC = 0x4C33424C # "L3BL" +CP_L3_BLOB_VERSION = 1 +_HEADER_STRUCT = struct.Struct(" stable on-disk id (frozen; never renumber — cold-rebuild reads old blobs) +_PAYLOAD_KIND_TO_ID = {"target_kv": 0, "draft_kv": 1, "index_k": 2} +_PAYLOAD_ID_TO_KIND = {v: k for k, v in _PAYLOAD_KIND_TO_ID.items()} + +CONTENT_HASH_BYTES = 32 # sha256 digest (node.hash_value entries are 64-hex == 32 bytes) + + +def _l3_failfast(message: str) -> None: + raise ValueError(f"{CP_L3_FAILFAST_PREFIX} {message}") + + +def align_up(nbytes: int, align: int = CP_L3_DIO_ALIGN) -> int: + if nbytes < 0 or align <= 0: + _l3_failfast(f"align_up requires nbytes>=0, align>0; got nbytes={nbytes} align={align}") + return (nbytes + align - 1) // align * align + + +def slot_bytes_for_page(page_bytes: int) -> int: + """On-disk slot size = header + payload, rounded up to the O_DIRECT boundary.""" + if page_bytes <= 0: + _l3_failfast(f"page_bytes must be positive; got {page_bytes}") + return align_up(CP_L3_HEADER_BYTES + int(page_bytes)) + + +def content_hash_to_bytes(hash_str: str) -> bytes: + """node.hash_value entry (64-char sha256 hex) -> 32 raw bytes for the blob header / index key.""" + try: + raw = bytes.fromhex(hash_str) + except (ValueError, TypeError) as exc: + _l3_failfast(f"content hash is not valid hex: {hash_str!r} ({exc})") + if len(raw) != CONTENT_HASH_BYTES: + _l3_failfast( + f"content hash must be {CONTENT_HASH_BYTES} bytes ({2*CONTENT_HASH_BYTES} hex chars); " + f"got {len(raw)} bytes from {hash_str!r}" + ) + return raw + + +def compute_crc(payload) -> int: + """CRC-32 (zlib, ISO-HDLC) over the page payload bytes. Integrity check, not crypto.""" + return zlib.crc32(payload) & 0xFFFFFFFF + + +def pack_blob_header( + *, payload_kind: str, content_hash: bytes, n_layers: int, page_bytes: int, crc: int, flags: int = 0 +) -> bytes: + if payload_kind not in _PAYLOAD_KIND_TO_ID: + _l3_failfast(f"unknown payload_kind: {payload_kind!r}") + if len(content_hash) != CONTENT_HASH_BYTES: + _l3_failfast(f"content_hash must be {CONTENT_HASH_BYTES} bytes; got {len(content_hash)}") + if n_layers <= 0 or n_layers > 0xFFFF: + _l3_failfast(f"n_layers out of range: {n_layers}") + if page_bytes <= 0 or page_bytes > 0xFFFFFFFF: + _l3_failfast(f"page_bytes out of range: {page_bytes}") + return _HEADER_STRUCT.pack( + CP_L3_BLOB_MAGIC, + CP_L3_BLOB_VERSION, + _PAYLOAD_KIND_TO_ID[payload_kind], + int(flags) & 0xFF, + bytes(content_hash), + int(n_layers), + int(page_bytes), + int(crc) & 0xFFFFFFFF, + ) + + +class CpL3BlobHeader: + """Parsed blob header. Use ``unpack_blob_header`` to construct from bytes.""" + + __slots__ = ("payload_kind", "content_hash", "n_layers", "page_bytes", "crc", "flags", "version") + + def __init__(self, payload_kind, content_hash, n_layers, page_bytes, crc, flags, version): + self.payload_kind = payload_kind + self.content_hash = content_hash + self.n_layers = n_layers + self.page_bytes = page_bytes + self.crc = crc + self.flags = flags + self.version = version + + +def unpack_blob_header(raw) -> CpL3BlobHeader: + """Parse + validate a 64-byte header. Fail loud on magic/version/kind corruption.""" + if len(raw) < CP_L3_HEADER_BYTES: + _l3_failfast(f"blob header needs {CP_L3_HEADER_BYTES} bytes; got {len(raw)}") + magic, version, kind_id, flags, content_hash, n_layers, page_bytes, crc = _HEADER_STRUCT.unpack( + bytes(raw[:CP_L3_HEADER_BYTES]) + ) + if magic != CP_L3_BLOB_MAGIC: + _l3_failfast(f"bad blob magic 0x{magic:08X} (expected 0x{CP_L3_BLOB_MAGIC:08X}) — torn/foreign slot") + if version != CP_L3_BLOB_VERSION: + _l3_failfast(f"unsupported blob version {version} (expected {CP_L3_BLOB_VERSION})") + if kind_id not in _PAYLOAD_ID_TO_KIND: + _l3_failfast(f"unknown payload_kind_id {kind_id} in blob header") + return CpL3BlobHeader( + payload_kind=_PAYLOAD_ID_TO_KIND[kind_id], + content_hash=content_hash, + n_layers=n_layers, + page_bytes=page_bytes, + crc=crc, + flags=flags, + version=version, + ) + + +def verify_blob(header: CpL3BlobHeader, payload, *, expect_hash: bytes = None) -> bool: + """Verify a read-back blob: payload length, CRC, and (optionally) the expected content hash. + + Returns True iff intact. A False result means treat the slot as a MISS (recompute) + reclaim it — + this is the verify-on-read guard against torn same-length writes / bit-rot (HiCacheFile has none). + """ + if len(payload) != header.page_bytes: + return False + if compute_crc(payload) != header.crc: + return False + if expect_hash is not None and bytes(expect_hash) != bytes(header.content_hash): + return False + return True + + +class CpL3SlotPool: + """O(1) free-list over fixed-size disk page-slots for one ``(payload_kind, drive)`` (rank-local). + + Mirrors ``CpSharedL2PageAllocator``'s free-list discipline: explicit free stack + allocated set, with + double-free / out-of-range / leak fail-loud, and a ``reset()`` for the ``flush_cache`` hook. Deterministic + allocation order (lowest free index first) so tests/observability are reproducible. NOT a ring buffer (a + ring = FIFO eviction = drops hot objects); eviction frees arbitrary slots back here. + """ + + def __init__(self, num_slots: int): + if num_slots <= 0: + _l3_failfast(f"CpL3SlotPool num_slots must be positive; got {num_slots}") + self._num_slots = int(num_slots) + # stack ordered so pop() yields 0,1,2,... (deterministic, reproducible) + self._free = list(range(self._num_slots - 1, -1, -1)) + self._allocated: set[int] = set() + + @property + def num_slots(self) -> int: + return self._num_slots + + @property + def num_free(self) -> int: + return len(self._free) + + @property + def num_allocated(self) -> int: + return len(self._allocated) + + @property + def occupancy(self) -> float: + return self._num_slots and (len(self._allocated) / self._num_slots) or 0.0 + + def has_free(self) -> bool: + return bool(self._free) + + def alloc(self) -> int: + """Reserve one slot; returns its index. Caller handles a full pool (returns -1) by evicting first.""" + if not self._free: + return -1 + idx = self._free.pop() + self._allocated.add(idx) + return idx + + def free(self, idx: int) -> None: + idx = int(idx) + if idx < 0 or idx >= self._num_slots: + _l3_failfast(f"free slot index {idx} out of range [0,{self._num_slots})") + if idx not in self._allocated: + _l3_failfast(f"double-free / never-allocated L3 slot {idx}") + self._allocated.discard(idx) + self._free.append(idx) + + def reset(self) -> None: + """Return to the freshly-constructed all-free state (the ``flush_cache``/``clear`` hook).""" + self._free = list(range(self._num_slots - 1, -1, -1)) + self._allocated = set() + diff --git a/test/registered/unit/mem_cache/test_cp_l3_disk.py b/test/registered/unit/mem_cache/test_cp_l3_disk.py new file mode 100644 index 000000000..18963bbd6 --- /dev/null +++ b/test/registered/unit/mem_cache/test_cp_l3_disk.py @@ -0,0 +1,140 @@ +"""Unit tests for CP HiCache L3 on-disk primitives (cp_l3_disk): blob header + CRC + slot pool. + +Pure-Python (no torch); loaded directly so the sglang import chain isn't needed. +""" + +import importlib.util +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +_REPO_ROOT = Path(__file__).resolve().parents[4] + + +def _load_cp_l3_disk_module(): + module_name = "_test_cp_l3_disk_module" + module_path = _REPO_ROOT / "python" / "sglang" / "srt" / "mem_cache" / "cp_l3_disk.py" + spec = importlib.util.spec_from_file_location(module_name, module_path) + module = importlib.util.module_from_spec(spec) + with patch.dict(sys.modules, {module_name: module}): + spec.loader.exec_module(module) + return module + + +m = _load_cp_l3_disk_module() + + +class TestBlobHeader(unittest.TestCase): + def _hash(self, seed: int) -> bytes: + return bytes((seed + i) & 0xFF for i in range(m.CONTENT_HASH_BYTES)) + + def test_header_is_64_bytes(self): + self.assertEqual(m.CP_L3_HEADER_BYTES, 64) + + def test_pack_unpack_roundtrip(self): + h = self._hash(7) + payload = bytes(range(256)) * 1024 # 256 KiB + crc = m.compute_crc(payload) + raw = m.pack_blob_header( + payload_kind="target_kv", content_hash=h, n_layers=78, page_bytes=len(payload), crc=crc + ) + self.assertEqual(len(raw), 64) + hdr = m.unpack_blob_header(raw) + self.assertEqual(hdr.payload_kind, "target_kv") + self.assertEqual(bytes(hdr.content_hash), h) + self.assertEqual(hdr.n_layers, 78) + self.assertEqual(hdr.page_bytes, len(payload)) + self.assertEqual(hdr.crc, crc) + self.assertTrue(m.verify_blob(hdr, payload, expect_hash=h)) + + def test_all_payload_kinds_roundtrip(self): + for kind in ("target_kv", "draft_kv", "index_k"): + raw = m.pack_blob_header( + payload_kind=kind, content_hash=self._hash(1), n_layers=21, page_bytes=4096, crc=0 + ) + self.assertEqual(m.unpack_blob_header(raw).payload_kind, kind) + + def test_verify_detects_corruption(self): + h = self._hash(3) + payload = b"\xAB" * 4096 + raw = m.pack_blob_header( + payload_kind="index_k", content_hash=h, n_layers=21, page_bytes=4096, crc=m.compute_crc(payload) + ) + hdr = m.unpack_blob_header(raw) + self.assertTrue(m.verify_blob(hdr, payload, expect_hash=h)) + # torn same-length write (bit flip) -> CRC mismatch + torn = bytearray(payload) + torn[100] ^= 0x01 + self.assertFalse(m.verify_blob(hdr, bytes(torn))) + # wrong length + self.assertFalse(m.verify_blob(hdr, payload[:-1])) + # wrong expected hash + self.assertFalse(m.verify_blob(hdr, payload, expect_hash=self._hash(99))) + + def test_bad_magic_fails_loud(self): + raw = bytearray(m.pack_blob_header( + payload_kind="target_kv", content_hash=self._hash(0), n_layers=1, page_bytes=4096, crc=0)) + raw[0] ^= 0xFF + with self.assertRaises(ValueError): + m.unpack_blob_header(bytes(raw)) + + def test_content_hash_hex_conversion(self): + hexstr = "ab" * 32 + self.assertEqual(m.content_hash_to_bytes(hexstr), bytes([0xAB]) * 32) + with self.assertRaises(ValueError): + m.content_hash_to_bytes("zz" * 32) # not hex + with self.assertRaises(ValueError): + m.content_hash_to_bytes("ab" * 16) # wrong length + + def test_slot_bytes_alignment(self): + # target_kv page = 2,875,392 B (702*4096); slot = header+payload rounded to 4K + slot = m.slot_bytes_for_page(2_875_392) + self.assertEqual(slot % m.CP_L3_DIO_ALIGN, 0) + self.assertGreaterEqual(slot, 64 + 2_875_392) + self.assertEqual(slot, 703 * 4096) + + +class TestSlotPool(unittest.TestCase): + def test_alloc_order_and_full(self): + p = m.CpL3SlotPool(4) + self.assertEqual([p.alloc() for _ in range(4)], [0, 1, 2, 3]) + self.assertEqual(p.num_free, 0) + self.assertEqual(p.alloc(), -1) # full -> -1, no raise + self.assertEqual(p.num_allocated, 4) + self.assertEqual(p.occupancy, 1.0) + + def test_free_and_realloc(self): + p = m.CpL3SlotPool(4) + a, b, c = p.alloc(), p.alloc(), p.alloc() # 0,1,2 + p.free(b) # free 1 + self.assertEqual(p.num_free, 2) # {1,3} + self.assertEqual(p.alloc(), 1) # reuse the freed slot + self.assertEqual(p.alloc(), 3) + + def test_double_free_and_oob_fail_loud(self): + p = m.CpL3SlotPool(2) + i = p.alloc() + p.free(i) + with self.assertRaises(ValueError): + p.free(i) # double free + with self.assertRaises(ValueError): + p.free(5) # out of range + with self.assertRaises(ValueError): + p.free(1) # never allocated + + def test_reset(self): + p = m.CpL3SlotPool(3) + p.alloc(); p.alloc() + p.reset() + self.assertEqual(p.num_free, 3) + self.assertEqual(p.num_allocated, 0) + self.assertEqual([p.alloc() for _ in range(3)], [0, 1, 2]) + + def test_construct_invalid(self): + with self.assertRaises(ValueError): + m.CpL3SlotPool(0) + + +if __name__ == "__main__": + unittest.main()