L3 3.0: POSIX disk backend (CpL3DiskSlab, O_DIRECT page slots)

Preallocated disk-slab file per (payload_kind, drive); a page = one contiguous blob
(64B header + gathered payload + pad) in a 4K-aligned slot. O_DIRECT single-transfer
write/read (measured best vs vectored), buffered+fdatasync fallback for FS without O_DIRECT.
write_slot/read_slot convenience + write_aligned/read_into zero-copy (the QD>1 spill/reload
path gathers/scatters straight into the aligned slot buffer). verify-on-read: absent/torn/
foreign slot -> None (a MISS to recompute + reclaim), never a crash. posix_fallocate preallocate.
9/9 unit tests (real temp FS).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

python/sglang/srt/mem_cache/cp_l3_posix.py

@@ -0,0 +1,154 @@
+# Copyright 2023-2024 SGLang Team
+# Licensed under the Apache License, Version 2.0
+"""CP HiCache L3 — POSIX disk backend (preallocated disk-slab files, O_DIRECT page I/O).
+
+One ``CpL3DiskSlab`` per ``(payload_kind, drive)``: a preallocated file of fixed-size page slots. A page is
+stored as one contiguous blob = 64B header + the (already-gathered, layer-major) page payload + zero pad to the
+4K O_DIRECT boundary. Writes/reads use a single aligned O_DIRECT transfer of the whole slot (measured faster
+than vectored I/O — research). On a filesystem without O_DIRECT (e.g. tmpfs in CI), it falls back to buffered
+I/O + fdatasync (correctness preserved; durability still relies on the PLP gate at config time).
+
+read_slot returns None (a MISS, to be recomputed + the slot reclaimed) for any absent/torn/foreign slot — the
+verify-on-read guard (CRC + content-hash in the header). Never raises on a bad slot; fails loud only on real
+I/O / programming errors.
+"""
+
+from __future__ import annotations
+
+import mmap
+import os
+from typing import Optional
+
+from sglang.srt.mem_cache.cp_l3_disk import (
+    CP_L3_HEADER_BYTES,
+    compute_crc,
+    pack_blob_header,
+    slot_bytes_for_page,
+    unpack_blob_header,
+    verify_blob,
+    _l3_failfast,
+)
+
+
+class CpL3DiskSlab:
+    """Preallocated disk-slab file of fixed-size page slots for one (payload_kind, drive)."""
+
+    def __init__(self, path: str, payload_kind: str, page_bytes: int, num_slots: int):
+        if num_slots <= 0:
+            _l3_failfast(f"CpL3DiskSlab num_slots must be positive; got {num_slots}")
+        self.path = str(path)
+        self.payload_kind = payload_kind
+        self.page_bytes = int(page_bytes)
+        self.slot_bytes = slot_bytes_for_page(page_bytes)  # 4K-aligned (header+payload+pad)
+        self.num_slots = int(num_slots)
+        self._fd = -1
+        self.direct = False
+
+    @property
+    def capacity_bytes(self) -> int:
+        return self.num_slots * self.slot_bytes
+
+    def open(self, *, create: bool = True) -> "CpL3DiskSlab":
+        flags = os.O_RDWR | (os.O_CREAT if create else 0)
+        try:
+            self._fd = os.open(self.path, flags | os.O_DIRECT, 0o600)
+            self.direct = True
+        except OSError:
+            self._fd = os.open(self.path, flags, 0o600)  # tmpfs / FS without O_DIRECT
+            self.direct = False
+        if create:
+            try:
+                os.posix_fallocate(self._fd, 0, self.capacity_bytes)
+            except (OSError, AttributeError):
+                os.ftruncate(self._fd, self.capacity_bytes)
+        return self
+
+    def _check_slot(self, slot_idx: int) -> None:
+        if slot_idx < 0 or slot_idx >= self.num_slots:
+            _l3_failfast(f"slot index {slot_idx} out of range [0,{self.num_slots}) for {self.path}")
+        if self._fd < 0:
+            _l3_failfast(f"disk slab {self.path} is not open")
+
+    def new_aligned_buffer(self) -> mmap.mmap:
+        """A page-aligned slot-sized buffer (mmap) suitable for O_DIRECT. The spill writer gathers the
+        layer slices straight into ``buf[CP_L3_HEADER_BYTES:CP_L3_HEADER_BYTES+page_bytes]`` (zero-copy)."""
+        return mmap.mmap(-1, self.slot_bytes)
+
+    def fill_header(self, buf: mmap.mmap, content_hash: bytes, n_layers: int, crc: int) -> None:
+        buf[0:CP_L3_HEADER_BYTES] = pack_blob_header(
+            payload_kind=self.payload_kind, content_hash=content_hash,
+            n_layers=n_layers, page_bytes=self.page_bytes, crc=crc,
+        )
+
+    def write_aligned(self, slot_idx: int, buf: mmap.mmap) -> None:
+        """Write a pre-built aligned slot buffer (header already filled). One O_DIRECT transfer."""
+        self._check_slot(slot_idx)
+        if len(buf) != self.slot_bytes:
+            _l3_failfast(f"aligned buffer is {len(buf)} bytes; slot is {self.slot_bytes}")
+        n = os.pwrite(self._fd, buf, slot_idx * self.slot_bytes)
+        if n != self.slot_bytes:
+            _l3_failfast(f"short write {n}/{self.slot_bytes} at slot {slot_idx}")
+
+    def write_slot(self, slot_idx: int, content_hash: bytes, n_layers: int, payload) -> int:
+        """Convenience: build the aligned blob (header+payload+pad) and write it. Returns the payload CRC."""
+        if len(payload) != self.page_bytes:
+            _l3_failfast(f"payload is {len(payload)} bytes; page is {self.page_bytes}")
+        crc = compute_crc(payload)
+        buf = self.new_aligned_buffer()
+        try:
+            self.fill_header(buf, content_hash, n_layers, crc)
+            buf[CP_L3_HEADER_BYTES:CP_L3_HEADER_BYTES + self.page_bytes] = bytes(payload)
+            # trailing pad already zero (mmap is zero-filled)
+            self.write_aligned(slot_idx, buf)
+        finally:
+            buf.close()
+        return crc
+
+    def read_slot(self, slot_idx: int, *, expect_hash: Optional[bytes] = None) -> Optional[bytes]:
+        """Read + verify a slot. Returns the payload bytes, or None if the slot is absent/torn/foreign
+        (verify-on-read: CRC + optional content-hash). Never raises on a bad slot."""
+        self._check_slot(slot_idx)
+        buf = self.new_aligned_buffer()
+        try:
+            n = os.preadv(self._fd, [buf], slot_idx * self.slot_bytes)
+            if n < CP_L3_HEADER_BYTES + self.page_bytes:
+                return None
+            try:
+                hdr = unpack_blob_header(buf[:CP_L3_HEADER_BYTES])
+            except ValueError:
+                return None  # bad magic/version = unwritten or foreign slot -> miss
+            if hdr.page_bytes != self.page_bytes:
+                return None
+            payload = bytes(buf[CP_L3_HEADER_BYTES:CP_L3_HEADER_BYTES + hdr.page_bytes])
+        finally:
+            buf.close()
+        if not verify_blob(hdr, payload, expect_hash=expect_hash):
+            return None
+        return payload
+
+    def read_into(self, slot_idx: int, dst_buf: mmap.mmap, *, expect_hash: Optional[bytes] = None) -> bool:
+        """Read a slot into a caller-provided aligned buffer (zero-copy reload). Returns True iff intact;
+        on True the payload is at dst_buf[CP_L3_HEADER_BYTES:CP_L3_HEADER_BYTES+page_bytes]."""
+        self._check_slot(slot_idx)
+        if len(dst_buf) != self.slot_bytes:
+            _l3_failfast(f"dst buffer is {len(dst_buf)} bytes; slot is {self.slot_bytes}")
+        n = os.preadv(self._fd, [dst_buf], slot_idx * self.slot_bytes)
+        if n < CP_L3_HEADER_BYTES + self.page_bytes:
+            return False
+        try:
+            hdr = unpack_blob_header(dst_buf[:CP_L3_HEADER_BYTES])
+        except ValueError:
+            return False
+        if hdr.page_bytes != self.page_bytes:
+            return False
+        payload = memoryview(dst_buf)[CP_L3_HEADER_BYTES:CP_L3_HEADER_BYTES + hdr.page_bytes]
+        return verify_blob(hdr, payload, expect_hash=expect_hash)
+
+    def fdatasync(self) -> None:
+        if self._fd >= 0:
+            os.fdatasync(self._fd)
+
+    def close(self) -> None:
+        if self._fd >= 0:
+            os.close(self._fd)
+            self._fd = -1

test/registered/unit/mem_cache/test_cp_l3_posix.py

@@ -0,0 +1,115 @@
+"""Unit tests for the CP HiCache L3 POSIX disk backend (cp_l3_posix.CpL3DiskSlab).
+
+Stubs the sglang package chain (cross-module import of cp_l3_disk). Uses a real temp FS (O_DIRECT if the
+FS supports it, else buffered fallback — both exercised correctly).
+"""
+
+import importlib.util
+import os
+import sys
+import tempfile
+import types
+import unittest
+from pathlib import Path
+
+_REPO_ROOT = Path(__file__).resolve().parents[4]
+_MEM = _REPO_ROOT / "python" / "sglang" / "srt" / "mem_cache"
+
+
+def _load_file(module_name, path):
+    spec = importlib.util.spec_from_file_location(module_name, path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def _load_posix():
+    for name, pkg_path in (
+        ("sglang", _REPO_ROOT / "python" / "sglang"),
+        ("sglang.srt", _REPO_ROOT / "python" / "sglang" / "srt"),
+        ("sglang.srt.mem_cache", _MEM),
+    ):
+        if name not in sys.modules:
+            mod = types.ModuleType(name)
+            mod.__path__ = [str(pkg_path)]
+            sys.modules[name] = mod
+    _load_file("sglang.srt.mem_cache.cp_l3_disk", _MEM / "cp_l3_disk.py")
+    return _load_file("sglang.srt.mem_cache.cp_l3_posix", _MEM / "cp_l3_posix.py")
+
+
+px = _load_posix()
+PAGE = 8192          # small page for fast tests
+NSLOTS = 8
+
+
+def _h(seed):
+    return bytes((seed + i) & 0xFF for i in range(32))
+
+
+class TestDiskSlab(unittest.TestCase):
+    def setUp(self):
+        self._td = tempfile.TemporaryDirectory()
+        self.slab = px.CpL3DiskSlab(
+            os.path.join(self._td.name, "target_kv.slab"), "target_kv", PAGE, NSLOTS
+        ).open()
+
+    def tearDown(self):
+        self.slab.close()
+        self._td.cleanup()
+
+    def test_slot_bytes_aligned(self):
+        self.assertEqual(self.slab.slot_bytes % 4096, 0)
+        self.assertEqual(self.slab.capacity_bytes, NSLOTS * self.slab.slot_bytes)
+
+    def test_write_read_roundtrip(self):
+        payload = bytes((i * 7) & 0xFF for i in range(PAGE))
+        h = _h(5)
+        crc = self.slab.write_slot(3, h, 78, payload)
+        got = self.slab.read_slot(3, expect_hash=h)
+        self.assertEqual(got, payload)
+        # without expect_hash also intact
+        self.assertEqual(self.slab.read_slot(3), payload)
+
+    def test_unwritten_slot_is_miss_not_crash(self):
+        self.assertIsNone(self.slab.read_slot(7))  # fresh fallocate = zeros -> bad magic -> None
+
+    def test_wrong_hash_is_miss(self):
+        payload = b"\x11" * PAGE
+        self.slab.write_slot(2, _h(1), 78, payload)
+        self.assertIsNone(self.slab.read_slot(2, expect_hash=_h(99)))
+
+    def test_corruption_detected(self):
+        payload = b"\x22" * PAGE
+        self.slab.write_slot(1, _h(2), 78, payload)
+        # corrupt a payload byte on disk directly
+        with open(self.slab.path, "r+b") as f:
+            f.seek(1 * self.slab.slot_bytes + 64 + 100)  # into payload
+            f.write(b"\xFF")
+        self.assertIsNone(self.slab.read_slot(1, expect_hash=_h(2)))  # CRC mismatch -> miss
+
+    def test_read_into_zero_copy(self):
+        payload = bytes((i * 3) & 0xFF for i in range(PAGE))
+        h = _h(8)
+        self.slab.write_slot(4, h, 78, payload)
+        buf = self.slab.new_aligned_buffer()
+        ok = self.slab.read_into(4, buf, expect_hash=h)
+        self.assertTrue(ok)
+        self.assertEqual(bytes(buf[64:64 + PAGE]), payload)
+        buf.close()
+
+    def test_oob_slot_fails_loud(self):
+        with self.assertRaises(ValueError):
+            self.slab.write_slot(NSLOTS, _h(0), 1, b"\x00" * PAGE)
+
+    def test_payload_size_mismatch_fails(self):
+        with self.assertRaises(ValueError):
+            self.slab.write_slot(0, _h(0), 1, b"\x00" * (PAGE - 1))
+
+    def test_fdatasync(self):
+        self.slab.write_slot(0, _h(0), 1, b"\x33" * PAGE)
+        self.slab.fdatasync()  # must not raise
+
+
+if __name__ == "__main__":
+    unittest.main()