From 06e39e49ab1dfe267ae4da98329de9f0dd497fe6 Mon Sep 17 00:00:00 2001 From: GentleCold Date: Sat, 13 Jun 2026 19:14:56 +0800 Subject: [PATCH 1/6] chore(bench): lmcache drain --- benchmarks/run_bench.py | 9 +- benchmarks/utils/loadgen.py | 6 +- tests/unit/test_benchmark_unified_utils.py | 112 +++++++++++++++++++++ 3 files changed, 122 insertions(+), 5 deletions(-) diff --git a/benchmarks/run_bench.py b/benchmarks/run_bench.py index 629859b..25777bc 100644 --- a/benchmarks/run_bench.py +++ b/benchmarks/run_bench.py @@ -23,7 +23,11 @@ COMPARISON_IOURING_MEM, slot_size_for_block_tokens, ) -from benchmarks.utils.loadgen import backend_server_hit_rate, collect_phase_metrics +from benchmarks.utils.loadgen import ( + _wait_lmcache_quiescent, + backend_server_hit_rate, + collect_phase_metrics, +) from benchmarks.utils.servers import BenchmarkManifest, stop_from_pid_file from benchmarks.utils.sizing import ( BenchmarkCapacityLimits, @@ -597,7 +601,8 @@ def _run_vllm_bench_load( cold_raw, ) if backend_run.backend == "lmcache": - _wait_with_message("lmcache_warm_settle_s", 10.0) + _print_kv("lmcache_warm_wait", "quiescent") + asyncio.run(_wait_lmcache_quiescent(manifest, settle_seconds=0.0)) elif backend_run.backend == "daser": _drain_daser(manifest) warm_metrics, warm_hit_rate = _run_vllm_bench_phase( diff --git a/benchmarks/utils/loadgen.py b/benchmarks/utils/loadgen.py index cad1f38..689fc97 100644 --- a/benchmarks/utils/loadgen.py +++ b/benchmarks/utils/loadgen.py @@ -220,7 +220,7 @@ async def run_lmcache( max_inflight: int, gen_params: dict[str, Any], timeout: float, - settle_seconds: float = 10.0, + settle_seconds: float = 0.0, chunk_aligned_prompts: bool = False, ) -> dict[str, Any]: """Run LMCache cold and warm full-prompt phases.""" @@ -469,10 +469,10 @@ async def _wait_lmcache_quiescent( manifest: BenchmarkManifest, settle_seconds: float ) -> None: del manifest - deadline = time.monotonic() + max(1.0, settle_seconds) + del settle_seconds stable = 0 async with httpx.AsyncClient(timeout=httpx.Timeout(5.0)) as client: - while time.monotonic() < deadline: + while True: status = await _get_json( client, f"http://127.0.0.1:{LMCACHE_HTTP_PORT}/status" ) diff --git a/tests/unit/test_benchmark_unified_utils.py b/tests/unit/test_benchmark_unified_utils.py index 29e5232..48a57f2 100644 --- a/tests/unit/test_benchmark_unified_utils.py +++ b/tests/unit/test_benchmark_unified_utils.py @@ -45,6 +45,7 @@ PhaseResult, RequestResult, _metric_hit_ratios, + _wait_lmcache_quiescent, lmcache_metrics_url, run_daser_chunk, run_daser_prefix, @@ -1272,6 +1273,97 @@ async def fake_vllm_completion_stream( assert calls == ["request", "drain", "request"] +async def test_lmcache_quiescence_wait_has_no_settle_deadline(monkeypatch) -> None: + """LMCache warm-up waits for quiescence instead of returning after 10 seconds.""" + polls = 0 + sleeps: list[float] = [] + + busy_status = { + "storage_manager": { + "store_controller": { + "pending_keys_count": 1, + "in_flight_task_count": 0, + }, + "prefetch_controller": { + "submission_queue_size": 0, + "pending_queue_size": 0, + "in_flight_request_count": 0, + "lookup_phase_count": 0, + "load_phase_count": 0, + }, + } + } + quiescent_status = { + "storage_manager": { + "store_controller": { + "pending_keys_count": 0, + "in_flight_task_count": 0, + }, + "prefetch_controller": { + "submission_queue_size": 0, + "pending_queue_size": 0, + "in_flight_request_count": 0, + "lookup_phase_count": 0, + "load_phase_count": 0, + }, + } + } + + async def fake_get_json(_client, _url): + nonlocal polls + polls += 1 + if polls <= 12: + return busy_status + return quiescent_status + + async def fake_sleep(seconds): + sleeps.append(seconds) + + now = -1.0 + + def fake_monotonic() -> float: + nonlocal now + now += 1.0 + return now + + class FakeClient: + def __init__(self, *_args, **_kwargs) -> None: + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args) -> None: + return None + + import benchmarks.utils.loadgen as loadgen + + monkeypatch.setattr(loadgen, "_get_json", fake_get_json) + monkeypatch.setattr(loadgen.asyncio, "sleep", fake_sleep) + monkeypatch.setattr(loadgen.httpx, "AsyncClient", FakeClient) + monkeypatch.setattr(loadgen.time, "monotonic", fake_monotonic) + + await _wait_lmcache_quiescent( + BenchmarkManifest( + run_id="test", + backend="lmcache", + reuse_mode="none", + model="model", + store_dir="/store", + l1_size_bytes=1, + l2_size_bytes=1, + skip_l2=True, + endpoints={"vllm": ServiceEndpoint("http://127.0.0.1:8001")}, + log_dir="/logs", + pid_file="/pids.json", + ), + settle_seconds=10.0, + ) + + assert polls == 15 + assert len(sleeps) == 14 + + def test_add_phase_comparison_records_cold_warm_correctness() -> None: """IMDB-style service results include cold/warm exact-match correctness.""" result = { @@ -2392,6 +2484,7 @@ def test_run_bench_vllm_bench_entrypoint_runs_openai_rows( """vLLM bench mode starts OpenAI-compatible rows and writes a summary.""" run_root = tmp_path / "run_20260102_030405" commands: list[list[str]] = [] + lmcache_waits: list[str] = [] def fake_run_command(command: list[str]) -> None: commands.append(command) @@ -2453,6 +2546,12 @@ def fake_run_command(command: list[str]) -> None: monkeypatch.setattr("benchmarks.run_bench._run_command", fake_run_command) monkeypatch.setattr("benchmarks.run_bench.stop_from_pid_file", lambda _path: None) + monkeypatch.setattr( + "benchmarks.run_bench._wait_with_message", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("vLLM bench LMCache warm wait must not use fixed sleep") + ), + ) monkeypatch.setattr( "benchmarks.run_bench.time.strftime", lambda _fmt: "20260102_030405", @@ -2463,6 +2562,17 @@ def fake_run_command(command: list[str]) -> None: lambda *_args, **_kwargs: None, ) + async def fake_wait_lmcache_quiescent( + manifest: BenchmarkManifest, + settle_seconds: float, + ) -> None: + lmcache_waits.append(f"{manifest.backend}:{settle_seconds}") + + monkeypatch.setattr( + "benchmarks.run_bench._wait_lmcache_quiescent", + fake_wait_lmcache_quiescent, + ) + async def fake_collect_phase_metrics( manifest: BenchmarkManifest, before_metrics: dict[str, Any] | None = None, @@ -2522,8 +2632,10 @@ async def fake_collect_phase_metrics( assert "warm_ttft_ms_mean: 11.0" in captured assert "warm_backend_cache_hit_rate: 0.8" in captured assert "warm_backend_cache_hit_rate: 0.85" in captured + assert "lmcache_warm_wait: quiescent" in captured assert "cold_warm_exact_match_accuracy: 1.0" in captured assert "cold_warm_exact_match_accuracy: 0.6666666666666666" in captured + assert lmcache_waits == ["lmcache:0.0"] assert len([cmd for cmd in commands if cmd[:3] == ["vllm", "bench", "serve"]]) == 5 for command in commands: if command[:3] == ["vllm", "bench", "serve"]: From b29535c403e2a315aba7cb939087e4445301a7e6 Mon Sep 17 00:00:00 2001 From: GentleCold Date: Sat, 13 Jun 2026 19:34:12 +0800 Subject: [PATCH 2/6] refactor(transfer): fold skip-l2 into iouring - Route skip_l2 through TieredIOUringTransferLayer instead of a separate memory backend. - Disable SSD, io_uring, and L2 persistence resources inside iouring skip_l2 mode. - Migrate L1-only behavior tests and documentation to the iouring skip_l2 path. --- daser/server/ipc/server.py | 11 +- daser/transfer/__init__.py | 2 - daser/transfer/iouring/layer.py | 195 +++++- daser/transfer/memory/__init__.py | 5 - daser/transfer/memory/layer.py | 571 ------------------ docs/design/architecture.md | 10 +- docs/design/components.md | 17 +- .../3_server_managed_transfer.md | 11 +- tests/server/test_ipc_server.py | 54 +- ...er.py => test_iouring_skip_l2_transfer.py} | 48 +- 10 files changed, 265 insertions(+), 659 deletions(-) delete mode 100644 daser/transfer/memory/__init__.py delete mode 100644 daser/transfer/memory/layer.py rename tests/transfer/{test_l1_only_transfer.py => test_iouring_skip_l2_transfer.py} (83%) diff --git a/daser/server/ipc/server.py b/daser/server/ipc/server.py index a7cb70f..d702433 100644 --- a/daser/server/ipc/server.py +++ b/daser/server/ipc/server.py @@ -19,7 +19,6 @@ from daser.transfer import TransferLayer from daser.transfer.cuda_ipc import open_cuda_ipc_buffer from daser.transfer.iouring import TieredIOUringTransferLayer -from daser.transfer.memory import L1OnlyTransferLayer logger = init_logger(__name__) @@ -638,11 +637,10 @@ def _ensure_transfer(self) -> TransferLayer: return self._transfer mode = str(self._runtime_config.get("transfer_mode", "gds")) path = str(self._runtime_config.get("store_path", "")) - if bool(self._runtime_config.get("skip_l2", False)): - self._transfer = L1OnlyTransferLayer( - l1_bytes=int(self._runtime_config.get("l1_size_bytes", 0)), - ) - elif mode == "gds": + skip_l2 = bool(self._runtime_config.get("skip_l2", False)) + if mode == "gds": + if skip_l2: + raise ValueError("skip_l2 is incompatible with gds transfer") from daser.transfer.gds import GDSTransferLayer self._transfer = GDSTransferLayer(path) @@ -661,6 +659,7 @@ def _ensure_transfer(self) -> TransferLayer: path=path, l1_bytes=int(self._runtime_config.get("l1_size_bytes", l2_bytes)), l2_bytes=l2_bytes, + skip_l2=skip_l2, ) else: raise ValueError(f"unknown transfer_mode: {mode}") diff --git a/daser/transfer/__init__.py b/daser/transfer/__init__.py index 158879b..975fb84 100644 --- a/daser/transfer/__init__.py +++ b/daser/transfer/__init__.py @@ -2,11 +2,9 @@ from daser.transfer.base import TransferLayer, TransferMode, TransferStats from daser.transfer.iouring import TieredIOUringTransferLayer -from daser.transfer.memory import L1OnlyTransferLayer __all__ = [ "GDSTransferLayer", - "L1OnlyTransferLayer", "TieredIOUringTransferLayer", "TransferBackend", "TransferLayer", diff --git a/daser/transfer/iouring/layer.py b/daser/transfer/iouring/layer.py index e27286a..7ed028d 100644 --- a/daser/transfer/iouring/layer.py +++ b/daser/transfer/iouring/layer.py @@ -60,30 +60,35 @@ def __init__( l1_bytes: int, l2_bytes: int, io_workers: int = 8, + skip_l2: bool = False, ) -> None: if l1_bytes <= 0: raise ValueError("l1_bytes must be positive") - if l2_bytes <= 0: + if not skip_l2 and l2_bytes <= 0: raise ValueError("l2_bytes must be positive") - if l1_bytes > l2_bytes: + if not skip_l2 and l1_bytes > l2_bytes: raise ValueError("l1_bytes must not exceed l2_bytes") if io_workers <= 0: raise ValueError("io_workers must be positive") - parent = os.path.dirname(path) - if parent: - os.makedirs(parent, exist_ok=True) - with open(path, "a+b") as f: - f.truncate(l2_bytes) - + self._skip_l2 = skip_l2 self._path = path - self._fd = os.open(path, os.O_RDWR | os.O_DIRECT) - self._urings = [NativeIOUring(entries=64) for _ in range(io_workers)] + self._fd: int | None = None + self._urings: list[NativeIOUring] = [] + self._io_executor: concurrent.futures.ThreadPoolExecutor | None = None self._uring_lock = threading.Lock() self._next_uring_index = 0 - self._io_executor = concurrent.futures.ThreadPoolExecutor( - max_workers=io_workers, - thread_name_prefix="daser-iouring", - ) + if not skip_l2: + parent = os.path.dirname(path) + if parent: + os.makedirs(parent, exist_ok=True) + with open(path, "a+b") as f: + f.truncate(l2_bytes) + self._fd = os.open(path, os.O_RDWR | os.O_DIRECT) + self._urings = [NativeIOUring(entries=64) for _ in range(io_workers)] + self._io_executor = concurrent.futures.ThreadPoolExecutor( + max_workers=io_workers, + thread_name_prefix="daser-iouring", + ) self._l1_bytes = l1_bytes self._l2_bytes = l2_bytes self._pool = PinnedMemoryPool( @@ -102,11 +107,14 @@ def __init__( self._lock = asyncio.Lock() self.stats = TransferStats() logger.info( - "[TRANSFER:iouring] path=%s l1=%d l2=%d direct_io=True io_workers=%d", + "[TRANSFER:iouring] path=%s l1=%d l2=%d direct_io=%s " + "io_workers=%d skip_l2=%s", path, l1_bytes, l2_bytes, + not skip_l2, io_workers, + skip_l2, ) async def load_bytes(self, dst: Any, file_offset: int, nbytes: int) -> int: @@ -131,8 +139,17 @@ async def load_bytes(self, dst: Any, file_offset: int, nbytes: int) -> int: file_offset=file_offset, nbytes=nbytes, ) + if self._skip_l2 and misses: + self.stats.l1_misses += len(misses) + raise KeyError( + "skip_l2 cache miss for range " + f"[{file_offset}, {file_offset + nbytes})" + ) if l1_hits: - self._record_l1_hits_locked(l1_hits) + self._record_l1_hits_locked( + l1_hits, + hit_count=1 if self._skip_l2 else None, + ) self._copy_grouped_to_dst( dst, [ @@ -195,8 +212,17 @@ async def load_bytes_grouped( file_offset=file_offset, nbytes=nbytes, ) + if self._skip_l2 and span_misses: + self.stats.l1_misses += 1 + raise KeyError( + "skip_l2 cache miss for range " + f"[{file_offset}, {file_offset + nbytes})" + ) if l1_hits: - self._record_l1_hits_locked(l1_hits) + self._record_l1_hits_locked( + l1_hits, + hit_count=1 if self._skip_l2 else None, + ) merged_l1.extend( ( hit.target_offset, @@ -243,6 +269,24 @@ async def store_bytes(self, src: Any, file_offset: int, nbytes: int) -> int: """ self._check_range(file_offset, nbytes) key = (file_offset, nbytes) + if self._skip_l2: + async with self._lock: + hit = self._find_l1_locked(file_offset, nbytes) + if hit is not None: + hit_key, cached, target_offset = hit + self._copy_src_to_pinned_at(src, cached, target_offset, nbytes) + self._policy.access(hit_key) + self._l1.move_to_end(hit_key) + return nbytes + data = self._reserve_l1_buffer_locked_or_raise(key, nbytes) + try: + self._copy_src_to_pinned_at(src, data, 0, nbytes) + except BaseException: + data.close() + raise + self._put_l1_locked(key, data) + return nbytes + data = await self._reserve_l1_buffer(key, nbytes) try: self._copy_src_to_pinned(src, data, nbytes) @@ -278,10 +322,15 @@ async def store_bytes_grouped( Total number of bytes stored. Async/thread-safety: - Uses ``store_bytes`` for each span so L1 visibility and asynchronous - L2 scheduling keep the same ordering guarantees as single-span - stores. + In normal tiered mode this uses ``store_bytes`` for each span so L1 + visibility and asynchronous L2 scheduling keep the same ordering + guarantees as single-span stores. In ``skip_l2`` mode it serializes + L1 metadata once for the whole group because no pending L2 writer + can retain evicted pool slices. """ + if self._skip_l2: + return await self._store_bytes_grouped_l1_only(src, spans) + total = 0 for span in spans: source_offset = int(span.get("source_offset", 0)) @@ -298,6 +347,8 @@ async def drain(self) -> None: Must be called from the owning asyncio event loop before shutdown when durable L2 contents are required. """ + if self._skip_l2: + return None while True: async with self._lock: pending = list(self._pending_l2.values()) @@ -315,10 +366,12 @@ def close(self) -> None: if pending_buffer not in self._l1.values(): pending_buffer.close() self._pending_l2_buffers.clear() - self._io_executor.shutdown(wait=True) + if self._io_executor is not None: + self._io_executor.shutdown(wait=True) for uring in self._urings: uring.close() - os.close(self._fd) + if self._fd is not None: + os.close(self._fd) self._pool.close() def _check_range(self, file_offset: int, nbytes: int) -> None: @@ -333,6 +386,12 @@ def _check_range(self, file_offset: int, nbytes: int) -> None: """ if file_offset < 0 or nbytes < 0: raise ValueError("file_offset and nbytes must be non-negative") + if self._skip_l2: + if nbytes > self._l1_bytes: + raise ValueError( + f"range {nbytes} bytes exceeds L1 capacity {self._l1_bytes}" + ) + return if file_offset + nbytes > self._l2_bytes: raise ValueError( f"range [{file_offset}, {file_offset + nbytes}) exceeds " @@ -463,7 +522,11 @@ def _resolve_l1_subranges_locked( cursor = gap_end return hits, misses - def _record_l1_hits_locked(self, hits: list[_L1RangeHit]) -> None: + def _record_l1_hits_locked( + self, + hits: list[_L1RangeHit], + hit_count: int | None = None, + ) -> None: """Update replacement state and stats for L1 hit slices. Args: @@ -478,7 +541,7 @@ def _record_l1_hits_locked(self, hits: list[_L1RangeHit]) -> None: for hit in hits: self._policy.access(hit.key) self._l1.move_to_end(hit.key) - self.stats.l1_hits += 1 + self.stats.l1_hits += len(hits) if hit_count is None else hit_count def _find_pending_l2_locked( self, @@ -500,6 +563,8 @@ def _read_l2_into( uring: NativeIOUring, ) -> int: """Blocking io_uring L2 read into pinned memory.""" + if self._fd is None: + raise RuntimeError("L2 reads are disabled when skip_l2 is true") return uring.read_into(self._fd, file_offset, dst.view()) def _write_l2( @@ -509,6 +574,8 @@ def _write_l2( uring: NativeIOUring, ) -> None: """Blocking io_uring L2 write.""" + if self._fd is None: + raise RuntimeError("L2 writes are disabled when skip_l2 is true") written = uring.write(self._fd, file_offset, data.view()) if written != len(data): raise IOError(f"short io_uring write: {written} != {len(data)}") @@ -521,6 +588,8 @@ def _schedule_l2_write_locked( previous: list[asyncio.Task[None]], ) -> asyncio.Task[None]: """Schedule one L2 write and start independent IO immediately.""" + if self._io_executor is None: + raise RuntimeError("L2 writes are disabled when skip_l2 is true") if previous: return asyncio.create_task( self._write_l2_async(key, file_offset, data, previous) @@ -576,6 +645,8 @@ async def _write_l2_async( if previous: await asyncio.gather(*previous) loop = asyncio.get_event_loop() + if self._io_executor is None: + raise RuntimeError("L2 writes are disabled when skip_l2 is true") await loop.run_in_executor( self._io_executor, self._write_l2, @@ -626,6 +697,8 @@ async def _load_l2_miss_batch( ) -> None: """Read one bounded L2 miss batch and promote it to L1.""" loop = asyncio.get_event_loop() + if self._io_executor is None: + raise RuntimeError("L2 reads are disabled when skip_l2 is true") reads: list[tuple[dict[str, int], PinnedMemorySlice]] = [] try: for span in misses: @@ -695,6 +768,8 @@ def _next_l2_miss_batch( def _next_uring(self) -> NativeIOUring: """Return the next native io_uring instance for one L2 operation.""" + if not self._urings: + raise RuntimeError("io_uring rings are disabled when skip_l2 is true") with self._uring_lock: uring = self._urings[self._next_uring_index] self._next_uring_index = (self._next_uring_index + 1) % len(self._urings) @@ -891,6 +966,62 @@ def _copy_src_to_pinned( return pinned.view()[:nbytes] = memoryview(src).cast("B")[:nbytes] + def _copy_src_to_pinned_at( + self, + src: Any, + pinned: PinnedMemorySlice, + target_offset: int, + nbytes: int, + ) -> None: + """Copy bytes from a CPU or CUDA source into pinned host memory.""" + if hasattr(src, "data") and getattr(src.data, "ptr", None) is not None: + from cupy.cuda import runtime + + runtime.memcpy( + pinned.ptr_at(target_offset), + int(src.data.ptr), + nbytes, + runtime.memcpyDeviceToHost, + ) + return + pinned.view()[target_offset : target_offset + nbytes] = memoryview(src).cast( + "B" + )[:nbytes] + + async def _store_bytes_grouped_l1_only( + self, + src: Any, + spans: list[dict[str, Any]], + ) -> int: + """Store grouped spans in L1 without scheduling L2 persistence.""" + total = 0 + async with self._lock: + self._raise_l2_error_locked() + for span in spans: + source_offset = int(span.get("source_offset", 0)) + nbytes = int(span["nbytes"]) + file_offset = int(span["file_offset"]) + self._check_range(file_offset, nbytes) + key = (file_offset, nbytes) + hit = self._find_l1_locked(file_offset, nbytes) + source = self._slice_src(src, source_offset, nbytes) + if hit is not None: + hit_key, cached, target_offset = hit + self._copy_src_to_pinned_at(source, cached, target_offset, nbytes) + self._policy.access(hit_key) + self._l1.move_to_end(hit_key) + total += nbytes + continue + data = self._reserve_l1_buffer_locked_or_raise(key, nbytes) + try: + self._copy_src_to_pinned_at(source, data, 0, nbytes) + except BaseException: + data.close() + raise + self._put_l1_locked(key, data) + total += nbytes + return total + async def _reserve_l1_buffer( self, key: tuple[int, int], @@ -951,6 +1082,22 @@ def _reserve_l1_buffer_locked( data = self._pool.allocate(nbytes) return data, None + def _reserve_l1_buffer_locked_or_raise( + self, + key: tuple[int, int], + nbytes: int, + ) -> PinnedMemorySlice: + """Reserve pinned L1 space when no pending L2 writer can block reuse.""" + data, wait_for = self._reserve_l1_buffer_locked(key, nbytes) + if data is not None: + return data + if wait_for is not None: + raise RuntimeError("unexpected pending L2 write in skip_l2 mode") + raise MemoryError( + f"could not reserve {nbytes} pinned L1 bytes from " + f"{self._l1_bytes} byte pool" + ) + def _release_l1_buffer_locked( self, key: tuple[int, int], diff --git a/daser/transfer/memory/__init__.py b/daser/transfer/memory/__init__.py deleted file mode 100644 index 7bcf371..0000000 --- a/daser/transfer/memory/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 - -from daser.transfer.memory.layer import L1OnlyTransferLayer - -__all__ = ["L1OnlyTransferLayer"] diff --git a/daser/transfer/memory/layer.py b/daser/transfer/memory/layer.py deleted file mode 100644 index a30ff00..0000000 --- a/daser/transfer/memory/layer.py +++ /dev/null @@ -1,571 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 - -# Standard -import asyncio -import bisect -from collections import OrderedDict -from typing import Any - -# First Party -from daser.logging import init_logger -from daser.replacement import LRUReplacementPolicy -from daser.transfer.base import TransferLayer, TransferStats -from daser.transfer.iouring.pinned_pool import PinnedMemoryPool, PinnedMemorySlice - -logger = init_logger(__name__) - -_L1_ALIGNMENT = 4096 - - -class L1OnlyTransferLayer(TransferLayer): - """Async pinned-memory transfer layer with no L2 persistence. - - Args: - l1_bytes: maximum memory-tier bytes. - - Async/thread-safety: - Public async methods serialize L1 metadata through an asyncio lock. - No file descriptors or background L2 writes are created. - """ - - coalesce_store_spans = True - - def __init__(self, l1_bytes: int) -> None: - if l1_bytes <= 0: - raise ValueError("l1_bytes must be positive") - self._l1_bytes = l1_bytes - self._pool = PinnedMemoryPool(l1_bytes, alignment=_L1_ALIGNMENT) - self._l1: OrderedDict[tuple[int, int], PinnedMemorySlice] = OrderedDict() - self._l1_starts: list[int] = [] - self._l1_by_start: dict[int, tuple[int, int]] = {} - self._l1_used = 0 - self._policy = LRUReplacementPolicy[tuple[int, int]]() - self._lock = asyncio.Lock() - self.stats = TransferStats() - logger.info("[TRANSFER:l1-only] l1=%d", l1_bytes) - - async def load_bytes(self, dst: Any, file_offset: int, nbytes: int) -> int: - """Load bytes from L1 into a writable destination buffer. - - Args: - dst: writable byte buffer or CUDA-like destination. - file_offset: logical byte offset used as the L1 key. - nbytes: number of bytes to load. - - Returns: - Number of bytes loaded. - - Raises: - KeyError: when the requested range is not resident in L1. - """ - self._check_range(file_offset, nbytes) - async with self._lock: - chunks = self._find_l1_chunks_locked(file_offset, nbytes) - if chunks is None: - self.stats.l1_misses += 1 - raise KeyError( - "L1-only cache miss for range " - f"[{file_offset}, {file_offset + nbytes})" - ) - for _target_offset, hit_key, _cached, _source_offset, _nbytes in chunks: - self._policy.access(hit_key) - self._l1.move_to_end(hit_key) - self.stats.l1_hits += 1 - if len(chunks) == 1: - _target_offset, _hit_key, cached, source_offset, _chunk_nbytes = chunks[ - 0 - ] - self._copy_pinned_to_dst(dst, cached, source_offset, nbytes) - else: - copy_chunks = [ - (target_offset, cached, source_offset, chunk_nbytes) - for ( - target_offset, - _key, - cached, - source_offset, - chunk_nbytes, - ) in chunks - ] - self._copy_grouped_to_dst(dst, copy_chunks) - return nbytes - - def _find_l1_chunks_locked( - self, - file_offset: int, - nbytes: int, - ) -> list[tuple[int, tuple[int, int], PinnedMemorySlice, int, int]] | None: - """Return adjacent resident L1 chunks covering a byte range.""" - end = file_offset + nbytes - cursor = file_offset - chunks: list[tuple[int, tuple[int, int], PinnedMemorySlice, int, int]] = [] - while cursor < end: - hit = self._find_l1_locked(cursor, 1) - if hit is None: - return None - key, cached, source_offset = hit - key_end = key[0] + key[1] - chunk_nbytes = min(end, key_end) - cursor - if chunk_nbytes <= 0: - return None - chunks.append( - ( - cursor - file_offset, - key, - cached, - source_offset, - chunk_nbytes, - ) - ) - cursor += chunk_nbytes - return chunks - - async def load_bytes_grouped( - self, - dst: Any, - spans: list[dict[str, int]], - ) -> int: - """Load multiple spans from L1 into a destination buffer. - - Args: - dst: writable byte buffer or CUDA-like destination. - spans: span dicts with target_offset, file_offset, and nbytes. - - Returns: - Total number of bytes loaded. - - Raises: - KeyError: when any requested span is not resident in L1. - """ - total = 0 - chunks: list[tuple[int, PinnedMemorySlice, int, int]] = [] - accessed: list[tuple[int, int]] = [] - async with self._lock: - for span in spans: - target_offset = int(span.get("target_offset", 0)) - file_offset = int(span["file_offset"]) - nbytes = int(span["nbytes"]) - self._check_range(file_offset, nbytes) - total += nbytes - span_chunks = self._find_l1_chunks_locked(file_offset, nbytes) - if span_chunks is None: - self.stats.l1_misses += 1 - raise KeyError( - "L1-only cache miss for range " - f"[{file_offset}, {file_offset + nbytes})" - ) - for ( - relative_target, - hit_key, - cached, - source_offset, - chunk_nbytes, - ) in span_chunks: - accessed.append(hit_key) - chunks.append( - ( - target_offset + relative_target, - cached, - source_offset, - chunk_nbytes, - ) - ) - for hit_key in accessed: - self._policy.access(hit_key) - self._l1.move_to_end(hit_key) - self.stats.l1_hits += len(spans) - self._copy_grouped_to_dst(dst, chunks) - return total - - async def store_bytes(self, src: Any, file_offset: int, nbytes: int) -> int: - """Store bytes in L1 memory without scheduling L2 persistence. - - Args: - src: readable byte buffer or CUDA-like source. - file_offset: logical byte offset used as the L1 key. - nbytes: number of bytes to store. - - Returns: - Number of bytes stored. - """ - self._check_range(file_offset, nbytes) - key = (file_offset, nbytes) - async with self._lock: - hit = self._find_l1_locked(file_offset, nbytes) - if hit is not None: - hit_key, cached, target_offset = hit - self._copy_src_to_pinned_at(src, cached, target_offset, nbytes) - self._policy.access(hit_key) - self._l1.move_to_end(hit_key) - return nbytes - data = self._reserve_l1_buffer_locked(key, nbytes) - try: - self._copy_src_to_pinned_at(src, data, 0, nbytes) - except BaseException: - data.close() - raise - self._put_l1_locked(key, data) - return nbytes - - async def store_bytes_grouped( - self, - src: Any, - spans: list[dict[str, Any]], - ) -> int: - """Store multiple spans in L1 memory. - - Args: - src: readable byte buffer or CUDA-like source. - spans: span dicts with source_offset, file_offset, and nbytes. - - Returns: - Total number of bytes stored. - - Async/thread-safety: - Serializes L1 metadata once for the whole group while preserving - per-span replacement ordering. - """ - total = 0 - async with self._lock: - for span in spans: - source_offset = int(span.get("source_offset", 0)) - nbytes = int(span["nbytes"]) - file_offset = int(span["file_offset"]) - self._check_range(file_offset, nbytes) - key = (file_offset, nbytes) - hit = self._find_l1_locked(file_offset, nbytes) - source = self._slice_src(src, source_offset, nbytes) - if hit is not None: - hit_key, cached, target_offset = hit - self._copy_src_to_pinned_at(source, cached, target_offset, nbytes) - self._policy.access(hit_key) - self._l1.move_to_end(hit_key) - total += nbytes - continue - data = self._reserve_l1_buffer_locked(key, nbytes) - try: - self._copy_src_to_pinned_at(source, data, 0, nbytes) - except BaseException: - data.close() - raise - self._put_l1_locked(key, data) - total += nbytes - return total - - async def drain(self) -> None: - """Return immediately because L1-only mode has no background writes.""" - return None - - def close(self) -> None: - """Release all L1 pinned memory resources.""" - for data in self._l1.values(): - data.close() - self._l1.clear() - self._l1_starts.clear() - self._l1_by_start.clear() - self._pool.close() - - def _check_range(self, file_offset: int, nbytes: int) -> None: - """Validate a logical byte range. - - Args: - file_offset: logical byte offset. - nbytes: byte count. - - Raises: - ValueError: when the range is negative or too large for L1. - """ - if file_offset < 0 or nbytes < 0: - raise ValueError("file_offset and nbytes must be non-negative") - if nbytes > self._l1_bytes: - raise ValueError( - f"range {nbytes} bytes exceeds L1 capacity {self._l1_bytes}" - ) - - def _put_l1_locked(self, key: tuple[int, int], data: PinnedMemorySlice) -> None: - """Insert a range into L1 and evict old ranges until within capacity.""" - self._l1[key] = data - self._insert_l1_index_locked(key) - self._l1.move_to_end(key) - self._policy.insert(key) - self._l1_used += len(data) - while self._l1_used > self._l1_bytes: - victim = self._policy.evict() - if victim is None: - break - removed = self._l1.pop(victim, None) - self._remove_l1_index_locked(victim) - if removed is not None: - self._l1_used -= len(removed) - removed.close() - - def _drop_overlapping_l1_locked(self, file_offset: int, nbytes: int) -> None: - """Remove L1 entries overlapping a new store range.""" - end = file_offset + nbytes - victims = [ - key for key in self._l1 if key[0] < end and file_offset < key[0] + key[1] - ] - for victim in victims: - removed = self._l1.pop(victim, None) - self._remove_l1_index_locked(victim) - self._policy.remove(victim) - if removed is not None: - self._l1_used -= len(removed) - removed.close() - - def _find_l1_locked( - self, - file_offset: int, - nbytes: int, - ) -> tuple[tuple[int, int], PinnedMemorySlice, int] | None: - """Return an L1 range covering the requested byte span.""" - end = file_offset + nbytes - idx = bisect.bisect_right(self._l1_starts, file_offset) - 1 - if idx < 0: - return None - start = self._l1_starts[idx] - key = self._l1_by_start.get(start) - if key is None: - return None - data = self._l1.get(key) - if data is None: - return None - if end <= key[0] + key[1]: - return key, data, file_offset - key[0] - return None - - def _reserve_l1_buffer_locked( - self, - key: tuple[int, int], - nbytes: int, - ) -> PinnedMemorySlice: - """Reserve pinned memory for one L1 range.""" - if nbytes > self._l1_bytes: - raise ValueError( - f"range {nbytes} bytes exceeds L1 capacity {self._l1_bytes}" - ) - self._drop_overlapping_l1_locked(key[0], key[1]) - data = self._pool.allocate(nbytes) - while data is None: - victim = self._policy.evict() - if victim is None: - raise MemoryError( - f"could not reserve {nbytes} pinned L1 bytes from " - f"{self._l1_bytes} byte pool" - ) - removed = self._l1.pop(victim, None) - self._remove_l1_index_locked(victim) - if removed is not None: - self._l1_used -= len(removed) - removed.close() - data = self._pool.allocate(nbytes) - return data - - def _insert_l1_index_locked(self, key: tuple[int, int]) -> None: - """Add one L1 range to the start-offset lookup index.""" - start = key[0] - existing = self._l1_by_start.get(start) - if existing == key: - return - if existing is not None: - self._remove_l1_index_locked(existing) - bisect.insort(self._l1_starts, start) - self._l1_by_start[start] = key - - def _remove_l1_index_locked(self, key: tuple[int, int]) -> None: - """Remove one L1 range from the start-offset lookup index.""" - start = key[0] - if self._l1_by_start.get(start) != key: - return - del self._l1_by_start[start] - idx = bisect.bisect_left(self._l1_starts, start) - if idx < len(self._l1_starts) and self._l1_starts[idx] == start: - self._l1_starts.pop(idx) - - def _copy_pinned_to_dst( - self, - dst: Any, - data: PinnedMemorySlice, - source_offset: int, - nbytes: int, - ) -> None: - """Copy pinned host bytes into a CPU or CUDA destination.""" - target = self._slice_dst(dst, 0, nbytes) - dst_ptr = self._cuda_array_ptr(target) - if dst_ptr is not None: - from cupy.cuda import runtime - - runtime.memcpyAsync( - dst_ptr, - data.ptr_at(source_offset), - nbytes, - runtime.memcpyHostToDevice, - 0, - ) - return - if hasattr(target, "set"): - import numpy - - host = numpy.frombuffer( - data.view()[source_offset : source_offset + nbytes], - dtype=numpy.uint8, - count=nbytes, - ) - target.set(host) - return - memoryview(dst).cast("B")[:nbytes] = data.view()[ - source_offset : source_offset + nbytes - ] - - def _copy_grouped_to_dst( - self, - dst: Any, - chunks: list[tuple[int, PinnedMemorySlice, int, int]], - ) -> None: - """Copy source chunks into the destination without staging repacks.""" - if not chunks: - return - first_target = self._slice_dst(dst, chunks[0][0], chunks[0][3]) - if self._cuda_array_ptr(first_target) is not None: - self._copy_grouped_to_cuda_dst(dst, chunks) - return - - for target_offset, data, source_offset, nbytes in self._coalesce_copy_chunks( - chunks - ): - target = self._slice_dst(dst, target_offset, nbytes) - if hasattr(target, "set"): - import numpy - - host = numpy.frombuffer( - data.view()[source_offset : source_offset + nbytes], - dtype=numpy.uint8, - count=nbytes, - ) - target.set(host) - continue - dst_view = memoryview(dst).cast("B") - dst_view[target_offset : target_offset + nbytes] = data.view()[ - source_offset : source_offset + nbytes - ] - - def _copy_grouped_to_cuda_dst( - self, - dst: Any, - chunks: list[tuple[int, PinnedMemorySlice, int, int]], - ) -> None: - """Copy grouped pinned ranges into a CUDA destination.""" - from cupy.cuda import runtime - - ordered = sorted(chunks, key=lambda item: item[0]) - merged: list[tuple[int, int, int]] = [] - for target_offset, data, source_offset, nbytes in ordered: - source_ptr = data.ptr_at(source_offset) - if not merged: - merged.append((target_offset, source_ptr, nbytes)) - continue - prev_target, prev_source, prev_nbytes = merged[-1] - if ( - target_offset == prev_target + prev_nbytes - and source_ptr == prev_source + prev_nbytes - ): - merged[-1] = (prev_target, prev_source, prev_nbytes + nbytes) - continue - merged.append((target_offset, source_ptr, nbytes)) - - for target_offset, source_ptr, nbytes in merged: - target = self._slice_dst(dst, target_offset, nbytes) - dst_ptr = self._cuda_array_ptr(target) - if dst_ptr is None: - raise TypeError("grouped CUDA copy target lost CUDA array interface") - runtime.memcpyAsync( - dst_ptr, - source_ptr, - nbytes, - runtime.memcpyHostToDevice, - 0, - ) - - def _coalesce_copy_chunks( - self, - chunks: list[tuple[int, PinnedMemorySlice, int, int]], - ) -> list[tuple[int, PinnedMemorySlice, int, int]]: - """Merge adjacent L1-hit copies with contiguous source and target.""" - ordered = sorted(chunks, key=lambda item: item[0]) - merged: list[tuple[int, PinnedMemorySlice, int, int]] = [] - for target_offset, data, source_offset, nbytes in ordered: - if not merged: - merged.append((target_offset, data, source_offset, nbytes)) - continue - prev_target, prev_data, prev_source, prev_nbytes = merged[-1] - if ( - prev_data is data - and target_offset == prev_target + prev_nbytes - and source_offset == prev_source + prev_nbytes - ): - merged[-1] = ( - prev_target, - prev_data, - prev_source, - prev_nbytes + nbytes, - ) - continue - merged.append((target_offset, data, source_offset, nbytes)) - return merged - - def _slice_dst(self, dst: Any, offset: int, nbytes: int) -> Any: - """Return a writable destination slice.""" - if hasattr(dst, "set"): - try: - return dst[offset : offset + nbytes] - except (TypeError, KeyError, IndexError): - if offset == 0: - return dst - raise - if isinstance(dst, bytearray | memoryview): - return memoryview(dst).cast("B")[offset : offset + nbytes] - try: - return dst[offset : offset + nbytes] - except (TypeError, KeyError, IndexError): - pass - return memoryview(dst).cast("B")[offset : offset + nbytes] - - def _slice_src(self, src: Any, offset: int, nbytes: int) -> Any: - """Return a readable source slice.""" - if hasattr(src, "get"): - return src[offset : offset + nbytes] - try: - return src[offset : offset + nbytes] - except (TypeError, KeyError, IndexError): - pass - return memoryview(src).cast("B")[offset : offset + nbytes] - - def _cuda_array_ptr(self, dst: Any) -> int | None: - """Return a CUDA device pointer for a CuPy-like array destination.""" - data = getattr(dst, "data", None) - ptr = getattr(data, "ptr", None) - if ptr is None: - return None - return int(ptr) - - def _copy_src_to_pinned_at( - self, - src: Any, - pinned: PinnedMemorySlice, - target_offset: int, - nbytes: int, - ) -> None: - """Copy bytes from a CPU or CUDA source into pinned host memory.""" - if hasattr(src, "data") and getattr(src.data, "ptr", None) is not None: - from cupy.cuda import runtime - - runtime.memcpy( - pinned.ptr_at(target_offset), - int(src.data.ptr), - nbytes, - runtime.memcpyDeviceToHost, - ) - return - pinned.view()[target_offset : target_offset + nbytes] = memoryview(src).cast( - "B" - )[:nbytes] diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 5a4a645..fce7760 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -42,15 +42,14 @@ graph TB TL["TransferLayer
server-owned data plane"] GDS["GDS backend
kvikio/cuFile"] IOR["iouring backend
O_DIRECT"] - MEM["L1-only backend
skip_l2"] + SKIP["skip_l2
disable L2"] L1["Pinned host memory
L1 LRU pool"] RP["ReplacementPolicy
LRU"] TL --> GDS TL --> IOR - TL --> MEM IOR --> L1 - MEM --> L1 + IOR --> SKIP RP --> L1 end @@ -250,8 +249,9 @@ system，之后不做运行时切换： | `gds` | server 打开 worker CUDA IPC staging buffer，使用 kvikio/cuFile 做 GPU ↔ NVMe 直接 DMA | | `iouring` | server 打开 worker CUDA IPC staging buffer，SSD 作为 L2，pinned host memory 作为 L1，L1 使用 LRU；L2 使用 `O_DIRECT` io_uring，范围必须 4096-byte 对齐 | -`--skip-l2` 是 `iouring` 兼容的 memory-only 开关：server 初始化 -`L1OnlyTransferLayer`，不打开 SSD 文件，store 只写 L1，load 只查 L1。 +`--skip-l2` 是 `iouring` 的 memory-only 开关：server 仍初始化 +`TieredIOUringTransferLayer`，但禁用 SSD 文件、io_uring rings 和 L2 write/read +路径；store 只写 L1，load 只查 L1。 ### Cache reuse mode diff --git a/docs/design/components.md b/docs/design/components.md index 52a2e67..4163d72 100644 --- a/docs/design/components.md +++ b/docs/design/components.md @@ -12,8 +12,7 @@ | `IPCClientAsync` | vLLM worker | asyncio Unix socket 客户端，用于 `transfer_store`、`transfer_load`、`commit_chunk` | | `TransferLayer` | DaseR | `daser/transfer/base.py`；server-owned KV 数据传输抽象，由 `IPCServer` 按 runtime config 初始化 | | `GDSTransferLayer` | DaseR | `daser/transfer/gds/`；封装 kvikio cuFile / compat IO；backend 在初始化时选定，运行期不可切换 | -| `TieredIOUringTransferLayer` | DaseR | `daser/transfer/iouring/`；L1 pinned-memory + L2 SSD transfer，L1 使用 LRU replacement | -| `L1OnlyTransferLayer` | DaseR | `daser/transfer/memory/`；`--skip-l2` 的 volatile L1-only transfer，不创建 store file，不做 L2 fallback | +| `TieredIOUringTransferLayer` | DaseR | `daser/transfer/iouring/`；L1 pinned-memory + L2 SSD transfer，L1 使用 LRU replacement；`--skip-l2` 时禁用 L2 store/io_uring 路径 | | `ReplacementPolicy` | DaseR | `daser/replacement/`；通用替换策略抽象，当前实现为 `LRUReplacementPolicy` | | `python -m daser.server` | DaseR | CLI 入口；解析配置，构造 `ServerCore`，启动 HTTP server 和 IPC server，关机保存 index | | `HTTP server` | DaseR | `daser/server/http/`；FastAPI routes、tokenize/chunk、vLLM HTTP 调用、文档 API 和 `/infer` | @@ -52,9 +51,10 @@ iouring transfer 的 L1 状态不进入 `MetadataStore`。L1 是 命中表和 LRU `ReplacementPolicy`；它是可丢失的加速层，重启后可以从 L2 `daser.store` 重新填充。L2/ring-buffer metadata 才需要随 `daser.index` 持久化。 -`--skip-l2` 下，`L1OnlyTransferLayer` 复用同样的 logical slot/file_offset -控制面，但不分配 `daser.store`，也不保存或恢复 `daser.index`。因此 lookup 和 -store 都只对当前进程内的 L1 bytes 有意义；L1 淘汰或进程重启后没有 L2 可恢复。 +`--skip-l2` 下，`TieredIOUringTransferLayer` 复用同样的 logical +slot/file_offset 控制面，但不分配 `daser.store`，不创建 io_uring rings，也不保存 +或恢复 `daser.index`。因此 lookup 和 store 都只对当前进程内的 L1 bytes 有意义； +L1 淘汰或进程重启后没有 L2 可恢复。 --- @@ -164,10 +164,9 @@ class TransferLayer(ABC): load 时先查 L1，miss 再从 L2 读入并 promote 到 L1。L2 文件使用 `O_DIRECT` 打开，所有 L2 offset 和 byte count 都要求 4096-byte 对齐。 Pinned L1 pool 在 transfer 初始化时一次性分配，后续 store/load 热路径只 - lease pool slice。 -- `L1OnlyTransferLayer`：`--skip-l2` 时使用；server 仍通过 CUDA IPC 访问 - worker staging buffer，但 bytes 只进入 pinned host L1。load miss 直接报错， - 因为没有 L2 文件可读回。该模式和 GDS 冲突。 + lease pool slice。`--skip-l2` 时使用同一个 transfer 实现和同一套 + pinned-memory copy path，但不打开 SSD 文件，load miss 直接报错，因为没有 L2 + 文件可读回。该模式和 GDS 冲突。 connector 不感知具体 transfer 实现，只发送 `transfer_store` / `transfer_load` IPC 请求和 CUDA IPC handle。 diff --git a/docs/optimizations/3_server_managed_transfer.md b/docs/optimizations/3_server_managed_transfer.md index 10ced39..3f5acac 100644 --- a/docs/optimizations/3_server_managed_transfer.md +++ b/docs/optimizations/3_server_managed_transfer.md @@ -19,11 +19,12 @@ The server selects one transfer mode at startup: L1 first, schedules L2 writes asynchronously through native io_uring syscalls, serves loads from L1 spans when present, and reads L2 plus promotes into L1 on misses. The L1 replacement policy is pluggable and currently backed by LRU. -- `iouring --skip-l2`: a volatile L1-only transfer path using - `L1OnlyTransferLayer`. It keeps the same IPC lookup/store flow and logical - slot offsets, but does not create `daser.store`, does not write L2, and does - not persist `daser.index`. Loads only succeed for ranges still resident in L1. - This mode is rejected with `gds` because GDS requires an L2 store file. +- `iouring --skip-l2`: the same `TieredIOUringTransferLayer` with the L2 + store/io_uring path disabled. It keeps the same IPC lookup/store flow and + logical slot offsets, but does not create `daser.store`, does not write L2, + and does not persist `daser.index`. Loads only succeed for ranges still + resident in L1. This mode is rejected with `gds` because GDS requires an L2 + store file. For both modes, the server performs transfer operations against CUDA IPC handles provided by the worker. The connector no longer chooses a transfer implementation diff --git a/tests/server/test_ipc_server.py b/tests/server/test_ipc_server.py index 2d5d5ac..7c8380b 100644 --- a/tests/server/test_ipc_server.py +++ b/tests/server/test_ipc_server.py @@ -650,33 +650,57 @@ def close(self) -> None: @pytest.mark.asyncio -async def test_skip_l2_selects_l1_only_transfer_without_store_path( +async def test_skip_l2_selects_iouring_transfer_without_store_path( tmp_path, monkeypatch: pytest.MonkeyPatch, ) -> None: - """skip_l2 should wire IPC transfer ops to the L1-only implementation.""" + """skip_l2 should wire IPC transfer ops to iouring with its L2 tier disabled.""" init_kwargs: list[dict[str, Any]] = [] - class FakeL1OnlyTransfer: + class FakeTieredIOUringTransfer: + coalesce_store_spans = True + def __init__(self, **kwargs: Any) -> None: init_kwargs.append(kwargs) - async def store_bytes(self, src: Any, file_offset: int, nbytes: int) -> int: - assert file_offset == 0 - assert bytes(src) == b"a" * nbytes - return nbytes + async def store_bytes_grouped( + self, + src: Any, + spans: list[dict[str, Any]], + ) -> int: + total = 0 + for span in spans: + source_offset = int(span.get("source_offset", 0)) + nbytes = int(span["nbytes"]) + assert int(span["file_offset"]) == 0 + assert bytes(src[source_offset : source_offset + nbytes]) == ( + b"a" * nbytes + ) + total += nbytes + return total - async def load_bytes(self, dst: Any, file_offset: int, nbytes: int) -> int: - assert file_offset == 0 - memoryview(dst).cast("B")[:nbytes] = b"a" * nbytes - return nbytes + async def load_bytes_grouped( + self, + dst: Any, + spans: list[dict[str, Any]], + ) -> int: + total = 0 + for span in spans: + target_offset = int(span.get("target_offset", 0)) + nbytes = int(span["nbytes"]) + assert int(span["file_offset"]) == 0 + memoryview(dst).cast("B")[target_offset : target_offset + nbytes] = ( + b"a" * nbytes + ) + total += nbytes + return total def close(self) -> None: pass monkeypatch.setattr( - "daser.server.ipc.server.L1OnlyTransferLayer", - FakeL1OnlyTransfer, + "daser.server.ipc.server.TieredIOUringTransferLayer", + FakeTieredIOUringTransfer, ) runtime_config = make_runtime_config(tmp_path) @@ -706,6 +730,8 @@ def close(self) -> None: finally: await server.stop() - assert init_kwargs == [{"l1_bytes": 8192}] + assert init_kwargs == [ + {"path": "", "l1_bytes": 8192, "l2_bytes": 8192, "skip_l2": True} + ] assert store == {"ok": True, "bytes": SLOT_SIZE, "chunk_keys": []} assert load == {"ok": True, "bytes": SLOT_SIZE, "data": b"a" * SLOT_SIZE} diff --git a/tests/transfer/test_l1_only_transfer.py b/tests/transfer/test_iouring_skip_l2_transfer.py similarity index 83% rename from tests/transfer/test_l1_only_transfer.py rename to tests/transfer/test_iouring_skip_l2_transfer.py index be984af..973e9ef 100644 --- a/tests/transfer/test_l1_only_transfer.py +++ b/tests/transfer/test_iouring_skip_l2_transfer.py @@ -7,7 +7,7 @@ import pytest # First Party -from daser.transfer.memory import L1OnlyTransferLayer +from daser.transfer.iouring import TieredIOUringTransferLayer ALIGNMENT = 4096 @@ -22,9 +22,19 @@ def _block(byte: bytes, size: int = ALIGNMENT) -> bytearray: return bytearray(byte * size) -def test_l1_only_transfer_stores_and_loads_without_store_file(tmp_path) -> None: - """Memory-only transfer should never create or require daser.store.""" - layer = L1OnlyTransferLayer(l1_bytes=ALIGNMENT * 2) +def _layer(tmp_path, l1_bytes: int) -> TieredIOUringTransferLayer: + """Return an iouring transfer layer with its L2 tier disabled.""" + return TieredIOUringTransferLayer( + path=str(tmp_path / "daser.store"), + l1_bytes=l1_bytes, + l2_bytes=l1_bytes, + skip_l2=True, + ) + + +def test_iouring_skip_l2_stores_and_loads_without_store_file(tmp_path) -> None: + """iouring skip_l2 should never create or require daser.store.""" + layer = _layer(tmp_path, ALIGNMENT * 2) store_path = tmp_path / "daser.store" dst = bytearray(ALIGNMENT) @@ -44,9 +54,9 @@ def test_l1_only_transfer_stores_and_loads_without_store_file(tmp_path) -> None: assert layer.stats.l2_writes == 0 -def test_l1_only_transfer_raises_on_l1_miss_after_eviction() -> None: +def test_iouring_skip_l2_raises_on_l1_miss_after_eviction(tmp_path) -> None: """With no L2 fallback, evicted byte ranges are no longer loadable.""" - layer = L1OnlyTransferLayer(l1_bytes=ALIGNMENT) + layer = _layer(tmp_path, ALIGNMENT) try: _run(layer.store_bytes(_block(b"a"), file_offset=0, nbytes=ALIGNMENT)) _run( @@ -58,7 +68,7 @@ def test_l1_only_transfer_raises_on_l1_miss_after_eviction() -> None: ) dst = bytearray(ALIGNMENT) - with pytest.raises(KeyError, match="L1-only cache miss"): + with pytest.raises(KeyError, match="skip_l2 cache miss"): _run(layer.load_bytes(dst, file_offset=0, nbytes=ALIGNMENT)) finally: layer.close() @@ -67,9 +77,9 @@ def test_l1_only_transfer_raises_on_l1_miss_after_eviction() -> None: assert layer.stats.l2_reads == 0 -def test_l1_only_transfer_grouped_loads_l1_ranges() -> None: +def test_iouring_skip_l2_grouped_loads_l1_ranges(tmp_path) -> None: """Grouped operations should serve all spans from L1 memory.""" - layer = L1OnlyTransferLayer(l1_bytes=ALIGNMENT * 2) + layer = _layer(tmp_path, ALIGNMENT * 2) try: _run( layer.store_bytes_grouped( @@ -109,9 +119,9 @@ def test_l1_only_transfer_grouped_loads_l1_ranges() -> None: assert layer.stats.l2_writes == 0 -def test_l1_only_grouped_store_uses_single_lock_pass(monkeypatch) -> None: - """Grouped L1 stores should avoid per-span store_bytes overhead.""" - layer = L1OnlyTransferLayer(l1_bytes=ALIGNMENT * 3) +def test_iouring_grouped_store_uses_single_lock_pass(tmp_path, monkeypatch) -> None: + """Grouped stores should avoid per-span store_bytes overhead.""" + layer = _layer(tmp_path, ALIGNMENT * 3) store_bytes_calls = 0 lock_entries = 0 original_store_bytes = layer.store_bytes @@ -167,9 +177,11 @@ async def __aexit__(self, exc_type, exc, tb): assert lock_entries == 2 -def test_l1_only_transfer_overwrite_preserves_adjacent_coalesced_ranges() -> None: +def test_iouring_skip_l2_overwrite_preserves_adjacent_coalesced_ranges( + tmp_path, +) -> None: """Overwriting part of a coalesced L1 range should keep neighbors loadable.""" - layer = L1OnlyTransferLayer(l1_bytes=ALIGNMENT * 4) + layer = _layer(tmp_path, ALIGNMENT * 4) try: _run( layer.store_bytes( @@ -214,9 +226,9 @@ def test_l1_only_transfer_overwrite_preserves_adjacent_coalesced_ranges() -> Non assert layer.stats.l2_reads == 0 -def test_l1_only_transfer_loads_across_adjacent_l1_ranges() -> None: +def test_iouring_skip_l2_loads_across_adjacent_l1_ranges(tmp_path) -> None: """A logical load range may span multiple adjacent resident L1 entries.""" - layer = L1OnlyTransferLayer(l1_bytes=ALIGNMENT * 4) + layer = _layer(tmp_path, ALIGNMENT * 4) try: _run( layer.store_bytes( @@ -247,9 +259,9 @@ def test_l1_only_transfer_loads_across_adjacent_l1_ranges() -> None: assert layer.stats.l2_reads == 0 -def test_l1_only_transfer_rejects_ranges_larger_than_l1() -> None: +def test_iouring_skip_l2_rejects_ranges_larger_than_l1(tmp_path) -> None: """A single store span must fit in the configured L1 capacity.""" - layer = L1OnlyTransferLayer(l1_bytes=ALIGNMENT) + layer = _layer(tmp_path, ALIGNMENT) try: with pytest.raises(ValueError, match="exceeds L1 capacity"): _run( From 67a833492c5b5a306d7d732768c90dd3159cf5ee Mon Sep 17 00:00:00 2001 From: GentleCold Date: Sun, 14 Jun 2026 00:35:27 +0800 Subject: [PATCH 3/6] chore(tests): align evict lmcache sizing - Stop expanding LMCache evict L1 sizing by the eviction watermark - Use the same derived L1 byte target for DaseR and LMCache benchmark runs - Update benchmark docs and unit tests for the equalized sizing policy --- benchmarks/README.md | 12 ++++----- benchmarks/utils/servers.py | 7 +----- benchmarks/utils/sizing.py | 29 +--------------------- tests/unit/test_benchmark_unified_utils.py | 2 +- tests/unit/test_benchmark_utils.py | 12 ++++----- 5 files changed, 15 insertions(+), 47 deletions(-) diff --git a/benchmarks/README.md b/benchmarks/README.md index 2e9dc2b..ff331e8 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -236,12 +236,12 @@ services: Human-readable sizes use MiB below 1 GiB and GiB otherwise. DaseR receives the derived L1 byte size directly. In evict runs, DaseR also receives the derived -L2 byte size. LMCache's L1 CLI accepts only integer GiB and starts eviction at -an 80% trigger watermark, so evict runs configure `lmcache_l1_gb` as -`ceil(derived_l1 / 0.8)` in GiB and pass `--eviction-trigger-watermark 0.8`. -For small workloads, LMCache's integer-GiB granularity can still make its -effective L1 slightly larger than DaseR's byte-exact L1. In no-evict runs, -DaseR starts with `--skip-l2` and no `--l2-size`, and `lmcache_l2_gb` is +L2 byte size. LMCache's L1 CLI accepts only integer GiB, so runs configure +`lmcache_l1_gb` as `ceil(derived_l1)` in GiB and still pass +`--eviction-trigger-watermark 0.8`. For small workloads, LMCache's integer-GiB +granularity can still make its configured L1 slightly larger than DaseR's +byte-exact L1. In no-evict runs, DaseR starts with `--skip-l2` and no +`--l2-size`, and `lmcache_l2_gb` is `null` because LMCache has no L2 adapter. In evict runs, LMCache's current FS L2 adapter CLI does not expose an L2 capacity limit, so the report should treat LMCache L2 as bounded by the filesystem free space rather than by the derived diff --git a/benchmarks/utils/servers.py b/benchmarks/utils/servers.py index 2d53cc9..f42d4bc 100644 --- a/benchmarks/utils/servers.py +++ b/benchmarks/utils/servers.py @@ -20,7 +20,6 @@ from benchmarks.utils.sizing import ( LMCACHE_EVICTION_TRIGGER_WATERMARK, bytes_to_lmcache_gb, - bytes_to_lmcache_gb_for_effective_l1, ) LMCACHE_MP_HOST = "tcp://localhost" @@ -245,11 +244,7 @@ def _lmcache_mp_server_command(self) -> list[str]: Pure helper except for creating the L2 scratch directory when the adapter is enabled. """ - l1_gb = ( - bytes_to_lmcache_gb(self.l1_size_bytes) - if self.skip_l2 - else bytes_to_lmcache_gb_for_effective_l1(self.l1_size_bytes) - ) + l1_gb = bytes_to_lmcache_gb(self.l1_size_bytes) cmd = [ "lmcache", "server", diff --git a/benchmarks/utils/sizing.py b/benchmarks/utils/sizing.py index a4036c0..76e6cc6 100644 --- a/benchmarks/utils/sizing.py +++ b/benchmarks/utils/sizing.py @@ -238,9 +238,7 @@ def derive_benchmark_sizing( daser_l2_bytes=daser_l2_bytes, daser_l1_bytes=daser_l1_bytes, lmcache_disk_gb=None, - lmcache_cpu_gb=bytes_to_lmcache_gb_for_effective_l1(daser_l1_bytes) - if evict - else bytes_to_lmcache_gb(daser_l1_bytes), + lmcache_cpu_gb=bytes_to_lmcache_gb(daser_l1_bytes), capacity_capped=capacity_capped, ) @@ -310,31 +308,6 @@ def bytes_to_lmcache_gb(nbytes: int) -> int: return math.ceil(nbytes / BYTES_PER_GIB) -def bytes_to_lmcache_gb_for_effective_l1( - nbytes: int, - watermark: float = LMCACHE_EVICTION_TRIGGER_WATERMARK, -) -> int: - """Convert effective L1 bytes to LMCache's configured GiB size. - - Args: - nbytes: Desired effective L1 capacity in bytes. - watermark: LMCache eviction trigger watermark. LMCache starts evicting - at this fraction of ``--l1-size-gb``. - - Returns: - LMCache ``--l1-size-gb`` value whose watermark capacity is at least - ``nbytes``. - - Thread-safety: - Pure function. - """ - if nbytes <= 0: - return 0 - if watermark <= 0: - raise ValueError("watermark must be positive") - return bytes_to_lmcache_gb(math.ceil(nbytes / watermark)) - - def format_capacity(nbytes: int) -> str: """Format a byte capacity for human-readable benchmark output. diff --git a/tests/unit/test_benchmark_unified_utils.py b/tests/unit/test_benchmark_unified_utils.py index 48a57f2..a402c24 100644 --- a/tests/unit/test_benchmark_unified_utils.py +++ b/tests/unit/test_benchmark_unified_utils.py @@ -748,7 +748,7 @@ def test_lmcache_evict_start_keeps_l2_adapter(tmp_path: Path) -> None: cmd = manager._lmcache_mp_server_command() # noqa: SLF001 l2_spec = json.loads(cmd[cmd.index("--l2-adapter") + 1]) - assert cmd[cmd.index("--l1-size-gb") + 1] == "2" + assert cmd[cmd.index("--l1-size-gb") + 1] == "1" assert cmd[cmd.index("--eviction-trigger-watermark") + 1] == "0.8" assert l2_spec["type"] == "fs" assert l2_spec["base_path"].endswith("lmcache_mp_disk") diff --git a/tests/unit/test_benchmark_utils.py b/tests/unit/test_benchmark_utils.py index 3084cb1..61cf0a8 100644 --- a/tests/unit/test_benchmark_utils.py +++ b/tests/unit/test_benchmark_utils.py @@ -9,7 +9,6 @@ BenchmarkCapacityLimits, align_down_gib, bytes_to_lmcache_gb, - bytes_to_lmcache_gb_for_effective_l1, derive_benchmark_sizing, derive_capacity_limits, format_capacity, @@ -165,6 +164,7 @@ def test_derive_benchmark_sizing_evict_keeps_l2_full_and_l1_partial() -> None: assert sizing.daser_l2_bytes // slot_size >= total_blocks assert sizing.daser_l1_bytes // slot_size == int(total_blocks * 0.8) assert sizing.daser_l1_bytes < sizing.daser_l2_bytes + assert sizing.lmcache_cpu_gb == bytes_to_lmcache_gb(sizing.daser_l1_bytes) def test_derive_benchmark_sizing_evict_uses_80_percent_mem_free_cap() -> None: @@ -229,13 +229,13 @@ def test_bytes_to_lmcache_gb_rounds_nonzero_capacity_up() -> None: assert bytes_to_lmcache_gb(1_073_479_680) == 1 -def test_bytes_to_lmcache_gb_for_effective_l1_accounts_for_watermark() -> None: - """Evict runs configure LMCache so its 80% watermark matches DaseR L1.""" +def test_bytes_to_lmcache_gb_does_not_expand_for_lmcache_watermark() -> None: + """Evict runs pass the same L1 capacity target to DaseR and LMCache.""" gib = 1024**3 - assert bytes_to_lmcache_gb_for_effective_l1(0) == 0 - assert bytes_to_lmcache_gb_for_effective_l1(1) == 1 - assert bytes_to_lmcache_gb_for_effective_l1(2 * gib) == 3 + assert bytes_to_lmcache_gb(0) == 0 + assert bytes_to_lmcache_gb(1) == 1 + assert bytes_to_lmcache_gb(2 * gib) == 2 def test_derive_benchmark_sizing_rejects_impossible_capacity() -> None: From a3bb816b8895a3bda07515387b8c959728835205 Mon Sep 17 00:00:00 2001 From: GentleCold Date: Sun, 14 Jun 2026 01:01:36 +0800 Subject: [PATCH 4/6] fix(transfer): preserve skip-l2 overwrite fragments - Preserve non-overlapping L1 fragments during skip-l2 overwrites - Avoid partial-tail in-place writes past the cached L1 slice - Add an LMCache quiescence timeout to prevent benchmark hangs - Cover skip-l2 overwrite and LMCache wait edge cases --- benchmarks/utils/loadgen.py | 9 +- daser/transfer/iouring/layer.py | 134 +++++++++++++++-- .../transfer/test_iouring_skip_l2_transfer.py | 138 ++++++++++++++++++ tests/unit/test_benchmark_unified_utils.py | 77 +++++++++- 4 files changed, 340 insertions(+), 18 deletions(-) diff --git a/benchmarks/utils/loadgen.py b/benchmarks/utils/loadgen.py index 689fc97..16e5b08 100644 --- a/benchmarks/utils/loadgen.py +++ b/benchmarks/utils/loadgen.py @@ -24,6 +24,8 @@ from benchmarks.utils.prompts import build_prompt_payloads from benchmarks.utils.servers import LMCACHE_HTTP_PORT, BenchmarkManifest +_LMCACHE_QUIESCENCE_TIMEOUT_SECONDS = 600.0 + @dataclass class RequestResult: @@ -469,10 +471,15 @@ async def _wait_lmcache_quiescent( manifest: BenchmarkManifest, settle_seconds: float ) -> None: del manifest - del settle_seconds + timeout_seconds = max(_LMCACHE_QUIESCENCE_TIMEOUT_SECONDS, float(settle_seconds)) + deadline = time.monotonic() + timeout_seconds stable = 0 async with httpx.AsyncClient(timeout=httpx.Timeout(5.0)) as client: while True: + if time.monotonic() >= deadline: + raise TimeoutError( + f"LMCache did not become quiescent within {timeout_seconds:.1f}s" + ) status = await _get_json( client, f"http://127.0.0.1:{LMCACHE_HTTP_PORT}/status" ) diff --git a/daser/transfer/iouring/layer.py b/daser/transfer/iouring/layer.py index 7ed028d..119f8a4 100644 --- a/daser/transfer/iouring/layer.py +++ b/daser/transfer/iouring/layer.py @@ -274,11 +274,16 @@ async def store_bytes(self, src: Any, file_offset: int, nbytes: int) -> int: hit = self._find_l1_locked(file_offset, nbytes) if hit is not None: hit_key, cached, target_offset = hit - self._copy_src_to_pinned_at(src, cached, target_offset, nbytes) - self._policy.access(hit_key) - self._l1.move_to_end(hit_key) - return nbytes - data = self._reserve_l1_buffer_locked_or_raise(key, nbytes) + if target_offset + nbytes <= len(cached): + self._copy_src_to_pinned_at(src, cached, target_offset, nbytes) + self._policy.access(hit_key) + self._l1.move_to_end(hit_key) + return nbytes + data = self._reserve_l1_buffer_locked_or_raise( + key, + nbytes, + preserve_overlaps=True, + ) try: self._copy_src_to_pinned_at(src, data, 0, nbytes) except BaseException: @@ -407,8 +412,16 @@ def _check_range(self, file_offset: int, nbytes: int) -> None: ) def _put_l1_locked(self, key: tuple[int, int], data: PinnedMemorySlice) -> None: - """Insert bytes into L1 and evict until capacity is respected.""" + """Insert bytes into L1 after dropping overlapping ranges.""" self._drop_overlapping_l1_locked(key[0], key[1]) + self._insert_l1_entry_locked(key, data) + + def _insert_l1_entry_locked( + self, + key: tuple[int, int], + data: PinnedMemorySlice, + ) -> None: + """Insert one non-overlapping L1 entry and enforce capacity.""" if len(data) > self._l1_bytes: return self._l1[key] = data @@ -427,7 +440,13 @@ def _put_l1_locked(self, key: tuple[int, int], data: PinnedMemorySlice) -> None: self._l1_used -= len(removed) self._release_l1_buffer_locked(victim, removed) - def _drop_overlapping_l1_locked(self, file_offset: int, nbytes: int) -> None: + def _drop_overlapping_l1_locked( + self, + file_offset: int, + nbytes: int, + *, + preserve_remainder: bool = False, + ) -> None: """Remove L1 entries that overlap a newly written byte range.""" end = file_offset + nbytes victims = [ @@ -437,9 +456,73 @@ def _drop_overlapping_l1_locked(self, file_offset: int, nbytes: int) -> None: removed = self._l1.pop(victim, None) self._remove_l1_index_locked(victim) self._policy.remove(victim) + preserved = ( + self._preserve_non_overlapping_l1_bytes( + victim, + removed, + file_offset, + end, + ) + if preserve_remainder and removed is not None + else [] + ) if removed is not None: self._l1_used -= len(removed) self._release_l1_buffer_locked(victim, removed) + for preserved_key, payload in preserved: + self._put_preserved_l1_fragment_locked(preserved_key, payload) + + def _preserve_non_overlapping_l1_bytes( + self, + key: tuple[int, int], + data: PinnedMemorySlice, + overlap_start: int, + overlap_end: int, + ) -> list[tuple[tuple[int, int], bytes]]: + """Return old L1 fragments that fall outside an overwrite range.""" + key_start, key_size = key + key_end = key_start + key_size + fragments: list[tuple[tuple[int, int], bytes]] = [] + view = data.view() + if key_start < overlap_start: + keep = overlap_start - key_start + fragments.append(((key_start, keep), bytes(view[:keep]))) + if overlap_end < key_end: + source_offset = overlap_end - key_start + keep = key_end - overlap_end + fragments.append( + ( + (overlap_end, keep), + bytes(view[source_offset : source_offset + keep]), + ) + ) + return fragments + + def _put_preserved_l1_fragment_locked( + self, + key: tuple[int, int], + payload: bytes, + ) -> None: + """Insert one preserved fragment copied out of an overwritten L1 range.""" + if not payload: + return + data, wait_for = self._reserve_l1_buffer_locked( + key, + len(payload), + drop_overlaps=False, + ) + if data is None: + if wait_for is not None: + raise RuntimeError("unexpected pending L2 write preserving L1 range") + raise MemoryError( + f"could not preserve {len(payload)} L1 bytes from overwritten range" + ) + try: + data.view()[: len(payload)] = payload + except BaseException: + data.close() + raise + self._insert_l1_entry_locked(key, data) def _find_l1_locked( self, @@ -1007,12 +1090,19 @@ async def _store_bytes_grouped_l1_only( source = self._slice_src(src, source_offset, nbytes) if hit is not None: hit_key, cached, target_offset = hit - self._copy_src_to_pinned_at(source, cached, target_offset, nbytes) - self._policy.access(hit_key) - self._l1.move_to_end(hit_key) - total += nbytes - continue - data = self._reserve_l1_buffer_locked_or_raise(key, nbytes) + if target_offset + nbytes <= len(cached): + self._copy_src_to_pinned_at( + source, cached, target_offset, nbytes + ) + self._policy.access(hit_key) + self._l1.move_to_end(hit_key) + total += nbytes + continue + data = self._reserve_l1_buffer_locked_or_raise( + key, + nbytes, + preserve_overlaps=True, + ) try: self._copy_src_to_pinned_at(source, data, 0, nbytes) except BaseException: @@ -1061,13 +1151,21 @@ def _reserve_l1_buffer_locked( self, key: tuple[int, int], nbytes: int, + *, + drop_overlaps: bool = True, + preserve_overlaps: bool = False, ) -> tuple[PinnedMemorySlice | None, asyncio.Task[None] | None]: """Try to reserve pinned L1 space for a new store or promoted load.""" if nbytes > self._l1_bytes: raise ValueError( f"range {nbytes} bytes exceeds L1 capacity {self._l1_bytes}" ) - self._drop_overlapping_l1_locked(key[0], key[1]) + if drop_overlaps: + self._drop_overlapping_l1_locked( + key[0], + key[1], + preserve_remainder=preserve_overlaps, + ) data = self._pool.allocate(nbytes) while data is None: victim = self._policy.evict() @@ -1086,9 +1184,15 @@ def _reserve_l1_buffer_locked_or_raise( self, key: tuple[int, int], nbytes: int, + *, + preserve_overlaps: bool = False, ) -> PinnedMemorySlice: """Reserve pinned L1 space when no pending L2 writer can block reuse.""" - data, wait_for = self._reserve_l1_buffer_locked(key, nbytes) + data, wait_for = self._reserve_l1_buffer_locked( + key, + nbytes, + preserve_overlaps=preserve_overlaps, + ) if data is not None: return data if wait_for is not None: diff --git a/tests/transfer/test_iouring_skip_l2_transfer.py b/tests/transfer/test_iouring_skip_l2_transfer.py index 973e9ef..2dec913 100644 --- a/tests/transfer/test_iouring_skip_l2_transfer.py +++ b/tests/transfer/test_iouring_skip_l2_transfer.py @@ -226,6 +226,144 @@ def test_iouring_skip_l2_overwrite_preserves_adjacent_coalesced_ranges( assert layer.stats.l2_reads == 0 +def test_iouring_skip_l2_partial_tail_overwrite_replaces_full_new_range( + tmp_path, +) -> None: + """A write that extends past an existing L1 entry should replace the range.""" + layer = _layer(tmp_path, ALIGNMENT * 4) + try: + _run( + layer.store_bytes( + _block(b"a") + _block(b"b"), + file_offset=0, + nbytes=ALIGNMENT * 2, + ) + ) + _run( + layer.store_bytes( + _block(b"x") + _block(b"y"), + file_offset=ALIGNMENT, + nbytes=ALIGNMENT * 2, + ) + ) + + dst = bytearray(ALIGNMENT * 3) + loaded = _run( + layer.load_bytes_grouped( + dst, + [ + {"target_offset": 0, "file_offset": 0, "nbytes": ALIGNMENT}, + { + "target_offset": ALIGNMENT, + "file_offset": ALIGNMENT, + "nbytes": ALIGNMENT * 2, + }, + ], + ) + ) + finally: + layer.close() + + assert loaded == ALIGNMENT * 3 + assert bytes(dst) == bytes(_block(b"a") + _block(b"x") + _block(b"y")) + assert layer.stats.l1_misses == 0 + assert layer.stats.l2_reads == 0 + + +def test_iouring_skip_l2_grouped_partial_tail_overwrite_replaces_full_new_range( + tmp_path, +) -> None: + """Grouped writes should preserve old L1 fragments around overwrite tails.""" + layer = _layer(tmp_path, ALIGNMENT * 4) + try: + _run( + layer.store_bytes( + _block(b"a") + _block(b"b"), + file_offset=0, + nbytes=ALIGNMENT * 2, + ) + ) + _run( + layer.store_bytes_grouped( + _block(b"x") + _block(b"y"), + [ + { + "source_offset": 0, + "file_offset": ALIGNMENT, + "nbytes": ALIGNMENT * 2, + } + ], + ) + ) + + dst = bytearray(ALIGNMENT * 3) + loaded = _run( + layer.load_bytes_grouped( + dst, + [ + {"target_offset": 0, "file_offset": 0, "nbytes": ALIGNMENT}, + { + "target_offset": ALIGNMENT, + "file_offset": ALIGNMENT, + "nbytes": ALIGNMENT * 2, + }, + ], + ) + ) + finally: + layer.close() + + assert loaded == ALIGNMENT * 3 + assert bytes(dst) == bytes(_block(b"a") + _block(b"x") + _block(b"y")) + assert layer.stats.l1_misses == 0 + assert layer.stats.l2_reads == 0 + + +def test_iouring_skip_l2_overwrite_evicts_preserved_fragments_when_l1_is_full( + tmp_path, +) -> None: + """Preserved fragments are best-effort when skip_l2 L1 has no spare room.""" + layer = _layer(tmp_path, ALIGNMENT * 2) + try: + _run( + layer.store_bytes( + _block(b"a") + _block(b"b"), + file_offset=0, + nbytes=ALIGNMENT * 2, + ) + ) + _run( + layer.store_bytes( + _block(b"x") + _block(b"y"), + file_offset=ALIGNMENT, + nbytes=ALIGNMENT * 2, + ) + ) + + dst = bytearray(ALIGNMENT * 2) + loaded = _run( + layer.load_bytes( + dst, + file_offset=ALIGNMENT, + nbytes=ALIGNMENT * 2, + ) + ) + with pytest.raises(KeyError, match="skip_l2 cache miss"): + _run( + layer.load_bytes( + bytearray(ALIGNMENT), + file_offset=0, + nbytes=ALIGNMENT, + ) + ) + finally: + layer.close() + + assert loaded == ALIGNMENT * 2 + assert bytes(dst) == bytes(_block(b"x") + _block(b"y")) + assert layer.stats.l2_reads == 0 + + def test_iouring_skip_l2_loads_across_adjacent_l1_ranges(tmp_path) -> None: """A logical load range may span multiple adjacent resident L1 entries.""" layer = _layer(tmp_path, ALIGNMENT * 4) diff --git a/tests/unit/test_benchmark_unified_utils.py b/tests/unit/test_benchmark_unified_utils.py index a402c24..5d0d9d2 100644 --- a/tests/unit/test_benchmark_unified_utils.py +++ b/tests/unit/test_benchmark_unified_utils.py @@ -1273,8 +1273,10 @@ async def fake_vllm_completion_stream( assert calls == ["request", "drain", "request"] -async def test_lmcache_quiescence_wait_has_no_settle_deadline(monkeypatch) -> None: - """LMCache warm-up waits for quiescence instead of returning after 10 seconds.""" +async def test_lmcache_quiescence_wait_can_exceed_old_fixed_sleep( + monkeypatch, +) -> None: + """LMCache warm-up waits for quiescence instead of a fixed 10s sleep.""" polls = 0 sleeps: list[float] = [] @@ -1364,6 +1366,77 @@ async def __aexit__(self, *_args) -> None: assert len(sleeps) == 14 +async def test_lmcache_quiescence_wait_times_out(monkeypatch) -> None: + """LMCache warm-up should fail clearly instead of waiting forever.""" + sleeps: list[float] = [] + busy_status = { + "storage_manager": { + "store_controller": { + "pending_keys_count": 1, + "in_flight_task_count": 0, + }, + "prefetch_controller": { + "submission_queue_size": 0, + "pending_queue_size": 0, + "in_flight_request_count": 0, + "lookup_phase_count": 0, + "load_phase_count": 0, + }, + } + } + + async def fake_get_json(_client, _url): + return busy_status + + async def fake_sleep(seconds): + sleeps.append(seconds) + + now = -1.0 + + def fake_monotonic() -> float: + nonlocal now + now += 1.0 + return now + + class FakeClient: + def __init__(self, *_args, **_kwargs) -> None: + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args) -> None: + return None + + import benchmarks.utils.loadgen as loadgen + + monkeypatch.setattr(loadgen, "_get_json", fake_get_json) + monkeypatch.setattr(loadgen.asyncio, "sleep", fake_sleep) + monkeypatch.setattr(loadgen.httpx, "AsyncClient", FakeClient) + monkeypatch.setattr(loadgen.time, "monotonic", fake_monotonic) + monkeypatch.setattr(loadgen, "_LMCACHE_QUIESCENCE_TIMEOUT_SECONDS", 3.0) + + with pytest.raises(TimeoutError, match="LMCache did not become quiescent"): + await _wait_lmcache_quiescent( + BenchmarkManifest( + run_id="test", + backend="lmcache", + reuse_mode="none", + model="model", + store_dir="/store", + l1_size_bytes=1, + l2_size_bytes=1, + skip_l2=True, + endpoints={"vllm": ServiceEndpoint("http://127.0.0.1:8001")}, + log_dir="/logs", + pid_file="/pids.json", + ), + settle_seconds=0.0, + ) + + assert len(sleeps) == 2 + + def test_add_phase_comparison_records_cold_warm_correctness() -> None: """IMDB-style service results include cold/warm exact-match correctness.""" result = { From dc27467a134a413391fe96ca22bc7d552ff329ea Mon Sep 17 00:00:00 2001 From: GentleCold Date: Sun, 14 Jun 2026 01:51:04 +0800 Subject: [PATCH 5/6] fix(bench): use external lmcache mp connector - Add the LMCache checkout ahead of DaseR in LMCache benchmark subprocess PYTHONPATH - Pass kv_connector_module_path for the LMCache MP connector - Cover LMCache benchmark connector and import path selection --- benchmarks/utils/servers.py | 49 +++++++++++++++--- tests/unit/test_benchmark_unified_utils.py | 59 ++++++++++++++++++++++ 2 files changed, 102 insertions(+), 6 deletions(-) diff --git a/benchmarks/utils/servers.py b/benchmarks/utils/servers.py index f42d4bc..c023774 100644 --- a/benchmarks/utils/servers.py +++ b/benchmarks/utils/servers.py @@ -25,7 +25,9 @@ LMCACHE_MP_HOST = "tcp://localhost" LMCACHE_MP_PORT = 5555 LMCACHE_HTTP_PORT = 8080 +LMCACHE_MP_CONNECTOR_MODULE = "lmcache.integration.vllm.lmcache_mp_connector" REPO_ROOT = Path(__file__).resolve().parents[2] +LMCACHE_REPO_ROOT = REPO_ROOT.parent / "LMCache" @dataclass(frozen=True) @@ -225,7 +227,11 @@ def manifest(self) -> BenchmarkManifest: async def start_lmcache_mp_server(self) -> None: """Start the LMCache MP server.""" cmd = self._lmcache_mp_server_command() - proc = self._start(cmd, "lmcache_mp_server.log") + proc = self._start( + cmd, + "lmcache_mp_server.log", + extra_env=self.lmcache_process_env(), + ) await self._wait_healthy( f"http://127.0.0.1:{LMCACHE_HTTP_PORT}", "/healthcheck", @@ -282,19 +288,50 @@ async def start_vllm_only(self) -> None: async def start_vllm_lmcache(self) -> None: """Start vLLM with LMCache MP connector.""" + await self._start_vllm( + "vllm_lmcache.log", + self.lmcache_kv_transfer_config(), + extra_env={ + **self.lmcache_process_env(), + "PYTHONHASHSEED": "42", + }, + ) + + def lmcache_kv_transfer_config(self) -> dict[str, Any]: + """Return vLLM KV transfer config for the sibling LMCache checkout. + + Returns: + JSON-serialisable vLLM ``--kv-transfer-config`` payload. + + Thread-safety: + Pure calculation over module constants and port configuration. + """ kv_config = { "kv_connector": "LMCacheMPConnector", + "kv_connector_module_path": LMCACHE_MP_CONNECTOR_MODULE, "kv_role": "kv_both", "kv_connector_extra_config": { "lmcache.mp.host": LMCACHE_MP_HOST, "lmcache.mp.port": LMCACHE_MP_PORT, }, } - await self._start_vllm( - "vllm_lmcache.log", - kv_config, - extra_env={"PYTHONHASHSEED": "42"}, - ) + return kv_config + + def lmcache_process_env(self) -> dict[str, str]: + """Return environment overrides for LMCache benchmark subprocesses. + + Returns: + Environment entries that make the sibling ``../LMCache`` checkout + take precedence over installed LMCache packages and vLLM built-ins. + + Thread-safety: + Reads the current process environment without mutating it. + """ + pythonpath = os.environ.get("PYTHONPATH") + entries = [str(LMCACHE_REPO_ROOT), str(REPO_ROOT)] + if pythonpath: + entries.append(pythonpath) + return {"PYTHONPATH": os.pathsep.join(entries)} async def start_vllm_daser(self) -> None: """Start vLLM with DaseR connector.""" diff --git a/tests/unit/test_benchmark_unified_utils.py b/tests/unit/test_benchmark_unified_utils.py index 5d0d9d2..48a3621 100644 --- a/tests/unit/test_benchmark_unified_utils.py +++ b/tests/unit/test_benchmark_unified_utils.py @@ -4,6 +4,7 @@ # Standard import asyncio import json +import os from pathlib import Path import subprocess import sys @@ -67,6 +68,8 @@ build_full_prompt, ) from benchmarks.utils.servers import ( + LMCACHE_MP_CONNECTOR_MODULE, + LMCACHE_REPO_ROOT, REPO_ROOT, BenchmarkManifest, ServerManager, @@ -2786,6 +2789,29 @@ def test_vllm_start_uses_vllm_generation_config(tmp_path: Path) -> None: assert command[command.index("--generation-config") + 1] == "vllm" +def test_lmcache_vllm_start_uses_external_mp_connector_module( + tmp_path: Path, +) -> None: + """LMCache benchmark rows should not use vLLM's built-in MP connector.""" + manager = ServerManager( + run_id="run1", + backend="lmcache", + model="/models/qwen", + store_dir=tmp_path, + gpu_id="2", + gpu_util=0.85, + max_num_seqs=32, + l1_size_bytes=1024, + l2_size_bytes=2048, + ) + + command = manager.vllm_command(manager.lmcache_kv_transfer_config()) + payload = json.loads(command[command.index("--kv-transfer-config") + 1]) + + assert payload["kv_connector"] == "LMCacheMPConnector" + assert payload["kv_connector_module_path"] == LMCACHE_MP_CONNECTOR_MODULE + + def test_vllm_start_can_override_max_num_batched_tokens(tmp_path: Path) -> None: """Benchmark vLLM servers can override the scheduler token budget.""" manager = ServerManager( @@ -2894,6 +2920,39 @@ def test_start_process_prefers_current_repo_on_pythonpath(tmp_path: Path) -> Non assert (tmp_path / "logs" / "pythonpath.log").read_text().strip() == str(REPO_ROOT) +def test_lmcache_start_process_prefers_lmcache_checkout_then_current_repo( + tmp_path: Path, +) -> None: + """LMCache benchmark rows import LMCache from the sibling checkout.""" + manager = ServerManager( + run_id="run1", + backend="lmcache", + model="/models/qwen", + store_dir=tmp_path, + gpu_id="2", + gpu_util=0.85, + max_num_seqs=32, + l1_size_bytes=1024, + l2_size_bytes=2048, + ) + + proc = manager._start( # noqa: SLF001 + [ + sys.executable, + "-c", + "import os; print(os.environ['PYTHONPATH'])", + ], + "pythonpath.log", + extra_env=manager.lmcache_process_env(), + ) + proc.wait(timeout=5) + + entries = ( + (tmp_path / "logs" / "pythonpath.log").read_text().strip().split(os.pathsep) + ) + assert entries[:2] == [str(LMCACHE_REPO_ROOT), str(REPO_ROOT)] + + async def test_wait_healthy_fails_when_startup_process_exits( tmp_path: Path, ) -> None: From e79e2193d38645a1950c10c89f8eae08ed9a99bf Mon Sep 17 00:00:00 2001 From: GentleCold Date: Sun, 14 Jun 2026 02:26:11 +0800 Subject: [PATCH 6/6] fix(bench): bypass lmcache connector registry - Add a benchmark shim connector name that avoids vLLM's built-in registry entry - Route the shim to the sibling LMCache checkout connector - Cover the non-conflicting connector name in benchmark tests --- benchmarks/utils/lmcache_connector_shim.py | 16 ++++++++++++++++ benchmarks/utils/servers.py | 5 +++-- tests/unit/test_benchmark_unified_utils.py | 5 +++-- 3 files changed, 22 insertions(+), 4 deletions(-) create mode 100644 benchmarks/utils/lmcache_connector_shim.py diff --git a/benchmarks/utils/lmcache_connector_shim.py b/benchmarks/utils/lmcache_connector_shim.py new file mode 100644 index 0000000..13cdfe1 --- /dev/null +++ b/benchmarks/utils/lmcache_connector_shim.py @@ -0,0 +1,16 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Benchmark-only alias for LMCache's external MP connector.""" + +from __future__ import annotations + +from lmcache.integration.vllm.lmcache_mp_connector import ( + LMCacheMPConnector as _ExternalLMCacheMPConnector, +) + + +class DaseRBenchLMCacheMPConnector(_ExternalLMCacheMPConnector): + """Alias that bypasses vLLM's built-in ``LMCacheMPConnector`` registry entry. + + Thread-safety: + Same as ``lmcache.integration.vllm.lmcache_mp_connector.LMCacheMPConnector``. + """ diff --git a/benchmarks/utils/servers.py b/benchmarks/utils/servers.py index c023774..5baaa53 100644 --- a/benchmarks/utils/servers.py +++ b/benchmarks/utils/servers.py @@ -25,7 +25,8 @@ LMCACHE_MP_HOST = "tcp://localhost" LMCACHE_MP_PORT = 5555 LMCACHE_HTTP_PORT = 8080 -LMCACHE_MP_CONNECTOR_MODULE = "lmcache.integration.vllm.lmcache_mp_connector" +LMCACHE_MP_CONNECTOR_NAME = "DaseRBenchLMCacheMPConnector" +LMCACHE_MP_CONNECTOR_MODULE = "benchmarks.utils.lmcache_connector_shim" REPO_ROOT = Path(__file__).resolve().parents[2] LMCACHE_REPO_ROOT = REPO_ROOT.parent / "LMCache" @@ -307,7 +308,7 @@ def lmcache_kv_transfer_config(self) -> dict[str, Any]: Pure calculation over module constants and port configuration. """ kv_config = { - "kv_connector": "LMCacheMPConnector", + "kv_connector": LMCACHE_MP_CONNECTOR_NAME, "kv_connector_module_path": LMCACHE_MP_CONNECTOR_MODULE, "kv_role": "kv_both", "kv_connector_extra_config": { diff --git a/tests/unit/test_benchmark_unified_utils.py b/tests/unit/test_benchmark_unified_utils.py index 48a3621..b4817ae 100644 --- a/tests/unit/test_benchmark_unified_utils.py +++ b/tests/unit/test_benchmark_unified_utils.py @@ -69,6 +69,7 @@ ) from benchmarks.utils.servers import ( LMCACHE_MP_CONNECTOR_MODULE, + LMCACHE_MP_CONNECTOR_NAME, LMCACHE_REPO_ROOT, REPO_ROOT, BenchmarkManifest, @@ -2792,7 +2793,7 @@ def test_vllm_start_uses_vllm_generation_config(tmp_path: Path) -> None: def test_lmcache_vllm_start_uses_external_mp_connector_module( tmp_path: Path, ) -> None: - """LMCache benchmark rows should not use vLLM's built-in MP connector.""" + """LMCache benchmark rows should bypass vLLM's built-in MP connector.""" manager = ServerManager( run_id="run1", backend="lmcache", @@ -2808,7 +2809,7 @@ def test_lmcache_vllm_start_uses_external_mp_connector_module( command = manager.vllm_command(manager.lmcache_kv_transfer_config()) payload = json.loads(command[command.index("--kv-transfer-config") + 1]) - assert payload["kv_connector"] == "LMCacheMPConnector" + assert payload["kv_connector"] == LMCACHE_MP_CONNECTOR_NAME assert payload["kv_connector_module_path"] == LMCACHE_MP_CONNECTOR_MODULE