diff --git a/python/sglang/srt/mem_cache/cp_l3_config.py b/python/sglang/srt/mem_cache/cp_l3_config.py new file mode 100644 index 000000000..fc08802ab --- /dev/null +++ b/python/sglang/srt/mem_cache/cp_l3_config.py @@ -0,0 +1,161 @@ +# Copyright 2023-2024 SGLang Team +# Licensed under the Apache License, Version 2.0 +"""CP HiCache L3 — configuration (config-driven, no hardcoded disk paths). + +Per-machine config consumed by ``CpL3Store.from_config``: the L3 disks (paths + per-disk byte budgets), the +**rank -> disk mapping** (M CP ranks >= N disks; ranks may share a disk — throughput is drive-count-bound, not +rank-count-bound), the backend, the shared metadata-index location, and the **PLP durability gate** (L3 +durability rides on power-loss-protected NVMe + O_DIRECT, NOT on the fsync barrier doing a NAND flush — so a +non-PLP disk is unsafe and must fail loud when ``require_plp`` is set). + +Loadable from TOML or JSON. Pure parse + mapping + validation here; the PLP probe touches the disk at runtime. +""" + +from __future__ import annotations + +import json +import os +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Optional, Tuple + +CP_L3_CONFIG_FAILFAST_PREFIX = "[CP_L3_CONFIG_FAILFAST]" +CP_L3_VALID_BACKENDS = frozenset(("posix", "3fs")) +_GIB = 1024 ** 3 + + +def _cfg_failfast(message: str) -> None: + raise ValueError(f"{CP_L3_CONFIG_FAILFAST_PREFIX} {message}") + + +@dataclass(frozen=True) +class CpL3DiskConfig: + path: str + budget_bytes: int + + def __post_init__(self): + if not self.path: + _cfg_failfast("disk path must be non-empty") + if self.budget_bytes <= 0: + _cfg_failfast(f"disk budget must be positive; got {self.budget_bytes} for {self.path!r}") + + +@dataclass(frozen=True) +class CpL3Config: + backend: str + disks: Tuple[CpL3DiskConfig, ...] + require_plp: bool + index_map_bytes: int + index_dir: str + rank_disk_map: Optional[Tuple[int, ...]] # explicit per-rank disk index, or None == balanced round-robin + + def __post_init__(self): + if self.backend not in CP_L3_VALID_BACKENDS: + _cfg_failfast(f"unknown backend {self.backend!r}; valid: {sorted(CP_L3_VALID_BACKENDS)}") + if not self.disks: + _cfg_failfast("at least one L3 disk is required") + if self.index_map_bytes <= 0: + _cfg_failfast(f"index_map_bytes must be positive; got {self.index_map_bytes}") + if not self.index_dir: + _cfg_failfast("index_dir must be non-empty") + + # ---- rank -> disk mapping (balanced round-robin unless explicit) ---- + def disk_for_rank(self, cp_rank: int, cp_size: int) -> int: + self.validate(cp_size) + if cp_rank < 0 or cp_rank >= cp_size: + _cfg_failfast(f"cp_rank {cp_rank} out of range [0,{cp_size})") + if self.rank_disk_map is not None: + return int(self.rank_disk_map[cp_rank]) + # balanced: rank r -> disk r % N; each disk gets floor/ceil(cp_size/N) ranks (graceful sharing) + return cp_rank % len(self.disks) + + def validate(self, cp_size: int) -> None: + if cp_size <= 0: + _cfg_failfast(f"cp_size must be positive; got {cp_size}") + if self.rank_disk_map is not None: + if len(self.rank_disk_map) != cp_size: + _cfg_failfast( + f"rank_disk_map length {len(self.rank_disk_map)} != cp_size {cp_size}" + ) + for r, d in enumerate(self.rank_disk_map): + if d < 0 or d >= len(self.disks): + _cfg_failfast(f"rank_disk_map[{r}]={d} out of range [0,{len(self.disks)})") + + @classmethod + def from_dict(cls, d: dict) -> "CpL3Config": + cfg = dict(d.get("cp_l3", d)) # accept either {cp_l3:{...}} or the inner dict + disks_raw = cfg.get("disks") + if not disks_raw: + _cfg_failfast("config has no 'disks'") + disks = tuple( + CpL3DiskConfig(path=str(x["path"]), budget_bytes=int(round(float(x["budget_gb"]) * _GIB))) + for x in disks_raw + ) + rdm = cfg.get("rank_disk_map", "balanced") + if isinstance(rdm, str): + if rdm != "balanced": + _cfg_failfast(f"rank_disk_map string must be 'balanced'; got {rdm!r}") + rank_disk_map = None + else: + rank_disk_map = tuple(int(x) for x in rdm) + index_dir = str(cfg.get("index_dir") or os.path.join(disks[0].path, "l3_index")) + return cls( + backend=str(cfg.get("backend", "posix")), + disks=disks, + require_plp=bool(cfg.get("require_plp", True)), + index_map_bytes=int(round(float(cfg.get("index_map_gb", 16)) * _GIB)), + index_dir=index_dir, + rank_disk_map=rank_disk_map, + ) + + @classmethod + def from_file(cls, path: str) -> "CpL3Config": + p = Path(path) + if not p.exists(): + _cfg_failfast(f"L3 config file not found: {path}") + text = p.read_bytes() + if p.suffix.lower() == ".json": + return cls.from_dict(json.loads(text)) + import tomllib # py3.11+ + + return cls.from_dict(tomllib.loads(text.decode("utf-8"))) + + +def probe_disk_plp(path: str, *, samples: int = 16, threshold_us: float = 200.0) -> Tuple[bool, float]: + """Best-effort PLP probe: O_DIRECT 4K write + fdatasync latency. PLP/datacenter NVMe acks the flush from + non-volatile cache (~sub-100us; we measured ~89 ns on the Dell DC NVMe 7500); a consumer drive without PLP + pays a real NAND flush (ms). Returns (is_plp_like, median_fdatasync_us). Heuristic — the caller fails loud + when ``require_plp`` and this is False. + """ + os.makedirs(path, exist_ok=True) + probe = os.path.join(path, f".cp_l3_plp_probe.{os.getpid()}") + buf = None + fd = -1 + lat = [] + try: + try: + fd = os.open(probe, os.O_RDWR | os.O_CREAT | os.O_DIRECT, 0o600) + except OSError: + fd = os.open(probe, os.O_RDWR | os.O_CREAT, 0o600) # FS without O_DIRECT (e.g. tmpfs) -> buffered + import mmap as _mmap + + buf = _mmap.mmap(-1, 4096) # page-aligned for O_DIRECT + buf[:] = b"\xA5" * 4096 + for i in range(max(2, samples)): + os.pwrite(fd, buf, (i % 64) * 4096) + t0 = time.perf_counter() + os.fdatasync(fd) + lat.append((time.perf_counter() - t0) * 1e6) + lat.sort() + median = lat[len(lat) // 2] + return (median <= threshold_us, median) + finally: + if buf is not None: + buf.close() + if fd >= 0: + os.close(fd) + try: + os.unlink(probe) + except OSError: + pass diff --git a/test/registered/unit/mem_cache/test_cp_l3_config.py b/test/registered/unit/mem_cache/test_cp_l3_config.py new file mode 100644 index 000000000..38746d72d --- /dev/null +++ b/test/registered/unit/mem_cache/test_cp_l3_config.py @@ -0,0 +1,103 @@ +"""Unit tests for CP HiCache L3 config (cp_l3_config): parse, rank->disk mapping, validation, PLP probe.""" + +import importlib.util +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +_REPO_ROOT = Path(__file__).resolve().parents[4] + + +def _load(name, rel): + module_name = f"_test_{name}_module" + spec = importlib.util.spec_from_file_location(module_name, _REPO_ROOT / rel) + 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_config", "python/sglang/srt/mem_cache/cp_l3_config.py") +_GIB = 1024 ** 3 + + +def _cfg(**over): + d = { + "backend": "posix", + "disks": [{"path": "/cache/l3", "budget_gb": 2000}, {"path": "/mnt/nvme3/l3", "budget_gb": 1000}], + "require_plp": True, + "index_map_gb": 16, + } + d.update(over) + return m.CpL3Config.from_dict(d) + + +class TestConfigParse(unittest.TestCase): + def test_from_dict_basic(self): + c = _cfg() + self.assertEqual(c.backend, "posix") + self.assertEqual(len(c.disks), 2) + self.assertEqual(c.disks[0].budget_bytes, 2000 * _GIB) + self.assertTrue(c.require_plp) + self.assertEqual(c.index_map_bytes, 16 * _GIB) + self.assertEqual(c.index_dir, "/cache/l3/l3_index") # default = first disk + self.assertIsNone(c.rank_disk_map) + + def test_accepts_cp_l3_wrapper(self): + c = m.CpL3Config.from_dict({"cp_l3": {"disks": [{"path": "/d", "budget_gb": 1}]}}) + self.assertEqual(len(c.disks), 1) + + def test_balanced_mapping_8ranks_4disks(self): + c = _cfg(disks=[{"path": f"/d{i}", "budget_gb": 1} for i in range(4)]) + got = [c.disk_for_rank(r, 8) for r in range(8)] + self.assertEqual(got, [0, 1, 2, 3, 0, 1, 2, 3]) # 2 ranks/disk, balanced + + def test_balanced_8ranks_3disks_graceful(self): + c = _cfg(disks=[{"path": f"/d{i}", "budget_gb": 1} for i in range(3)]) + got = [c.disk_for_rank(r, 8) for r in range(8)] + self.assertEqual(got, [0, 1, 2, 0, 1, 2, 0, 1]) # ranks share disks gracefully + + def test_explicit_rank_disk_map(self): + c = _cfg(disks=[{"path": "/a", "budget_gb": 1}, {"path": "/b", "budget_gb": 1}], + rank_disk_map=[0, 0, 1, 1]) + self.assertEqual([c.disk_for_rank(r, 4) for r in range(4)], [0, 0, 1, 1]) + + def test_explicit_map_wrong_length_fails(self): + c = _cfg(rank_disk_map=[0, 1]) + with self.assertRaises(ValueError): + c.disk_for_rank(0, 8) # map len 2 != cp_size 8 + + def test_explicit_map_oob_fails(self): + with self.assertRaises(ValueError): + _cfg(disks=[{"path": "/a", "budget_gb": 1}], rank_disk_map=[0, 5]).disk_for_rank(0, 2) + + def test_empty_disks_fails(self): + with self.assertRaises(ValueError): + m.CpL3Config.from_dict({"disks": []}) + + def test_bad_backend_fails(self): + with self.assertRaises(ValueError): + _cfg(backend="leveldb") + + def test_from_file_json(self): + with tempfile.TemporaryDirectory() as td: + p = os.path.join(td, "l3.json") + Path(p).write_text('{"cp_l3": {"disks": [{"path": "/x", "budget_gb": 5}], "index_map_gb": 8}}') + c = m.CpL3Config.from_file(p) + self.assertEqual(c.disks[0].budget_bytes, 5 * _GIB) + self.assertEqual(c.index_map_bytes, 8 * _GIB) + + +class TestPlpProbe(unittest.TestCase): + def test_probe_returns_bool_and_latency(self): + with tempfile.TemporaryDirectory() as td: + is_plp, median_us = m.probe_disk_plp(td, samples=4) + self.assertIsInstance(is_plp, bool) + self.assertGreater(median_us, 0.0) + + +if __name__ == "__main__": + unittest.main()