feat(connector): add tiered io_uring KV prefetch

benchmarks/bench_start_servers.py

@@ -43,6 +43,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--block-size", type=int, default=BLOCK_TOKENS)
     parser.add_argument("--l1-size", type=parse_size_bytes, default="256gib")
     parser.add_argument("--l2-size", type=parse_size_bytes, default="300gib")
+    parser.add_argument("--daser-prefetch-max-requests", type=int, default=0)
     parser.add_argument(
         "--cache-reuse-mode", choices=("chunk", "prefix"), default="chunk"
     )
@@ -87,6 +88,7 @@ async def main_async(args: argparse.Namespace) -> None:
         skip_l2=args.skip_l2,
         tensor_parallel_size=args.tensor_parallel_size,
         trust_remote_code=args.trust_remote_code,
+        daser_prefetch_max_requests=args.daser_prefetch_max_requests,
     )
     manifest = await manager.start()
     print(f"manifest={args.store_dir}/manifest.json")

benchmarks/run_bench.py

@@ -81,6 +81,7 @@ class RunBenchArgs:
         bench_seed: vLLM bench random seed.
         bench_burstiness: vLLM bench burstiness factor.
         evict: Whether to enable L2 and eviction sizing.
+        daser_prefetch_max_requests: Maximum concurrent DaseR prefetches.
         prometheus_url: Optional Prometheus base URL for scrape diagnostics.
 
     Thread-safety:
@@ -117,6 +118,7 @@ class RunBenchArgs:
     bench_seed: int = 42
     bench_burstiness: float = 1.0
     evict: bool = False
+    daser_prefetch_max_requests: int = 0
     prometheus_url: str = "http://127.0.0.1:9090"
 
 
@@ -188,6 +190,7 @@ def parse_args(argv: list[str] | None = None) -> RunBenchArgs:
     parser.add_argument("--bench-seed", type=int, default=42)
     parser.add_argument("--bench-burstiness", type=float, default=1.0)
     parser.add_argument("--evict", action="store_true")
+    parser.add_argument("--daser-prefetch-max-requests", type=int, default=0)
     parser.add_argument(
         "--prometheus-url",
         default="http://127.0.0.1:9090",
@@ -228,6 +231,7 @@ def parse_args(argv: list[str] | None = None) -> RunBenchArgs:
         bench_seed=args.bench_seed,
         bench_burstiness=args.bench_burstiness,
         evict=args.evict,
+        daser_prefetch_max_requests=args.daser_prefetch_max_requests,
         prometheus_url=args.prometheus_url,
     )
     try:
@@ -538,6 +542,13 @@ def _start_command(
         "--l2-size",
         str(derived_l2),
     ]
+    if backend_run.backend == "daser" and args.daser_prefetch_max_requests:
+        command.extend(
+            [
+                "--daser-prefetch-max-requests",
+                str(args.daser_prefetch_max_requests),
+            ]
+        )
     if backend_run.backend == "daser":
         command.extend(["--cache-reuse-mode", backend_run.reuse_mode])
     if args.trust_remote_code:

benchmarks/utils/loadgen.py

@@ -25,6 +25,7 @@
 from benchmarks.utils.servers import LMCACHE_HTTP_PORT, BenchmarkManifest
 
 _LMCACHE_QUIESCENCE_TIMEOUT_SECONDS = 600.0
+_DASER_DRAIN_TIMEOUT_SECONDS = 360.0
 
 
 @dataclass
@@ -211,6 +212,7 @@ async def run_daser_prefix(
         metrics=await collect_phase_metrics(manifest, before_warm),
         elapsed_ms=warm_elapsed_ms,
     )
+    await _wait_daser_drained(manifest)
     return {"cold": cold_phase, "warm": warm_phase}
 
 
@@ -264,6 +266,7 @@ async def run_lmcache(
         metrics=await collect_phase_metrics(manifest, before_warm),
         elapsed_ms=warm_elapsed_ms,
     )
+    await _wait_daser_drained(manifest)
     return {"cold": cold_phase, "warm": warm_phase}
 
 
@@ -515,7 +518,9 @@ async def _wait_daser_drained(manifest: BenchmarkManifest) -> None:
     daser = manifest.endpoints.get("daser")
     if daser is None:
         return
-    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+    async with httpx.AsyncClient(
+        timeout=httpx.Timeout(_DASER_DRAIN_TIMEOUT_SECONDS)
+    ) as client:
         response = await client.post(f"{daser.url}/drain")
         response.raise_for_status()
 

benchmarks/utils/servers.py

@@ -132,6 +132,7 @@ def __init__(
         skip_l2: bool = False,
         tensor_parallel_size: int = 1,
         trust_remote_code: bool = False,
+        daser_prefetch_max_requests: int = 0,
     ) -> None:
         """Initialize the service manager.
 
@@ -156,6 +157,7 @@ def __init__(
             skip_l2: Disable L2 persistence/adapters for L1-only no-evict runs.
             tensor_parallel_size: vLLM tensor-parallel rank count.
             trust_remote_code: allow model/tokenizer repository Python code.
+            daser_prefetch_max_requests: maximum concurrent scheduler prefetches.
         """
         if tensor_parallel_size <= 0:
             raise ValueError("tensor_parallel_size must be positive")
@@ -179,6 +181,7 @@ def __init__(
         self.skip_l2 = skip_l2
         self.tensor_parallel_size = tensor_parallel_size
         self.trust_remote_code = trust_remote_code
+        self.daser_prefetch_max_requests = daser_prefetch_max_requests
         self.log_dir = self.store_dir / "logs"
         self.pid_file = self.store_dir / "pids.json"
         self.socket_path = self.store_dir / "daser.sock"
@@ -351,6 +354,7 @@ async def start_vllm_daser(self) -> None:
             "kv_connector_extra_config": {
                 "socket_path": str(self.socket_path),
                 "cache_reuse_mode": self.reuse_mode,
+                "prefetch_max_requests": self.daser_prefetch_max_requests,
             },
         }
         await self._start_vllm("vllm_daser.log", kv_config)

benchmarks/utils/vllm_bench.py

@@ -24,6 +24,7 @@
     slot_size_for_block_tokens,
 )
 from benchmarks.utils.loadgen import (
+    _DASER_DRAIN_TIMEOUT_SECONDS,
     _wait_lmcache_quiescent,
     backend_server_hit_rate,
     collect_phase_metrics,
@@ -268,6 +269,8 @@ def run_load(
             backend_dir,
             run_command=run_command,
         )
+        if backend_run.backend == "daser":
+            _drain_daser(manifest, print_kv=print_kv)
         result_path.write_text(json.dumps(result, indent=2), encoding="utf-8")
         return
     if backend_run.backend == "vllm":
@@ -308,6 +311,8 @@ def run_load(
             warm_metrics, warm_hit_rate = _run_phase(
                 args, manifest, warm_raw, run_command=run_command
             )
+        if backend_run.backend == "daser":
+            _drain_daser(manifest, print_kv=print_kv)
         if backend_run.backend == "daser" and args.evict:
             _require_daser_evict_tier_activity(warm_metrics)
         cold_summary = _normalise_result(cold_raw)
@@ -451,7 +456,7 @@ def _drain_daser(
     endpoint = manifest.endpoints.get("daser")
     if endpoint is None:
         return
-    response = httpx.post(f"{endpoint.url}/drain", timeout=30.0)
+    response = httpx.post(f"{endpoint.url}/drain", timeout=_DASER_DRAIN_TIMEOUT_SECONDS)
     response.raise_for_status()
     print_kv("daser_drain_status", "ok")
 

daser/connector/daser_connector.py

@@ -96,13 +96,18 @@ def __init__(
 
         socket_path = str(extra.get("socket_path", "/tmp/daser.sock"))
         if role == KVConnectorRole.SCHEDULER:
+            prefetch_max_requests = int(extra.get("prefetch_max_requests", 0))
+            if prefetch_max_requests < 0:
+                raise ValueError("prefetch_max_requests must be non-negative")
             self._request_lifecycle = RequestLifecycle(
                 ipc_client=IPCClientSync(socket_path),
                 block_tokens=16,
                 slot_size=0,
                 model_id="default",
                 cache_reuse_mode=str(extra.get("cache_reuse_mode", "chunk")),
                 runtime_config_ready=False,
+                socket_path=socket_path,
+                prefetch_max_requests=prefetch_max_requests,
             )
             self._request_lifecycle.refresh_runtime_config()
         else:
@@ -135,6 +140,22 @@ def __init__(
 
         logger.info("[CONNECTOR] role=%s socket=%s", role.name, socket_path)
 
+    def shutdown(self) -> None:
+        """Stop connector-side scheduler or worker resources.
+
+        Async/thread-safety:
+            Called by vLLM during shutdown. Scheduler prefetch workers are
+            joined and worker transfer pipelines use their existing shutdown
+            path.
+        """
+        lifecycle = getattr(self, "_request_lifecycle", None)
+        if lifecycle is not None:
+            lifecycle.shutdown()
+            return
+        runtime = getattr(self, "_worker_runtime", None)
+        if runtime is not None:
+            runtime.shutdown()
+
     @property
     def prefer_cross_layer_blocks(self) -> bool:
         """Request vLLM cross-layer KV cache blocks for bulk chunk transfers.