feat(benchmarks): make DaseR prefetch opt-in

benchmarks/README.md

@@ -110,6 +110,14 @@ and `/infer` endpoints instead of the OpenAI completions endpoint. If
 fails fast and asks you to select only `baseline,lmcache,daser-prefix` or use
 `--load-generator internal`.
 
+Scheduler-side DaseR prefetch is disabled by default. Add
+`--daser-prefetch` to opt in; it uses two worker threads. The compatibility
+override `--daser-prefetch-max-requests N` accepts any non-negative value and
+takes precedence, so `--daser-prefetch-max-requests 0` explicitly disables
+prefetch. Both options are available on `run_bench.py` and
+`bench_start_servers.py`; the effective value and enabled state are written to
+each DaseR manifest.
+
 The vLLM-bench-specific knobs are:
 
 | Option | Meaning |

benchmarks/bench_start_servers.py

@@ -12,7 +12,10 @@
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
 from benchmarks.utils.constants import BLOCK_TOKENS
-from benchmarks.utils.servers import ServerManager
+from benchmarks.utils.servers import (
+    ServerManager,
+    resolve_daser_prefetch_max_requests,
+)
 from benchmarks.utils.sizing import parse_size_bytes
 from benchmarks.utils.system import apply_gpu_selection
 
@@ -43,7 +46,21 @@ 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(
+        "--daser-prefetch",
+        action="store_true",
+        help="Enable DaseR scheduler-side prefetch (default worker limit: 2).",
+    )
+    parser.add_argument(
+        "--daser-prefetch-max-requests",
+        type=int,
+        default=None,
+        metavar="N",
+        help=(
+            "Expert override for DaseR prefetch workers; zero explicitly "
+            "disables prefetch and takes precedence over --daser-prefetch."
+        ),
+    )
     parser.add_argument(
         "--cache-reuse-mode", choices=("chunk", "prefix"), default="chunk"
     )
@@ -63,6 +80,10 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
 
 async def main_async(args: argparse.Namespace) -> None:
     """Start services and write manifest."""
+    prefetch_max_requests = resolve_daser_prefetch_max_requests(
+        args.daser_prefetch,
+        args.daser_prefetch_max_requests,
+    )
     selected_gpu = apply_gpu_selection(args.gpu_id) or args.gpu_id
     run_id = args.run_id or time.strftime("%Y%m%d_%H%M%S")
     manager = ServerManager(
@@ -88,7 +109,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,
+        daser_prefetch_max_requests=prefetch_max_requests,
     )
     manifest = await manager.start()
     print(f"manifest={args.store_dir}/manifest.json")

benchmarks/run_bench.py

@@ -20,7 +20,11 @@
 from benchmarks.utils import vllm_bench
 from benchmarks.utils.constants import BLOCK_TOKENS
 from benchmarks.utils.datasets import add_dataset_cli_args
-from benchmarks.utils.servers import BenchmarkManifest, stop_from_pid_file
+from benchmarks.utils.servers import (
+    BenchmarkManifest,
+    resolve_daser_prefetch_max_requests,
+    stop_from_pid_file,
+)
 
 _DASER_METRICS_SETTLE_SECONDS = 2.0
 _BACKEND_CLEANUP_SETTLE_SECONDS = 2.0
@@ -81,6 +85,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: Explicitly enable scheduler-side DaseR prefetch.
         daser_prefetch_max_requests: Maximum concurrent DaseR prefetches.
         prometheus_url: Optional Prometheus base URL for scrape diagnostics.
 
@@ -118,6 +123,7 @@ class RunBenchArgs:
     bench_seed: int = 42
     bench_burstiness: float = 1.0
     evict: bool = False
+    daser_prefetch: bool = False
     daser_prefetch_max_requests: int = 0
     prometheus_url: str = "http://127.0.0.1:9090"
 
@@ -190,7 +196,21 @@ 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(
+        "--daser-prefetch",
+        action="store_true",
+        help="Enable DaseR scheduler-side prefetch (default worker limit: 2).",
+    )
+    parser.add_argument(
+        "--daser-prefetch-max-requests",
+        type=int,
+        default=None,
+        metavar="N",
+        help=(
+            "Expert override for DaseR prefetch workers; accepts zero to "
+            "explicitly disable and takes precedence over --daser-prefetch."
+        ),
+    )
     parser.add_argument(
         "--prometheus-url",
         default="http://127.0.0.1:9090",
@@ -200,6 +220,13 @@ def parse_args(argv: list[str] | None = None) -> RunBenchArgs:
         ),
     )
     args = parser.parse_args(argv)
+    try:
+        effective_prefetch_max_requests = resolve_daser_prefetch_max_requests(
+            args.daser_prefetch,
+            args.daser_prefetch_max_requests,
+        )
+    except ValueError as exc:
+        parser.error(str(exc))
     parsed = RunBenchArgs(
         backend=args.backend,
         load_generator=args.load_generator,
@@ -231,7 +258,8 @@ 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,
+        daser_prefetch=args.daser_prefetch,
+        daser_prefetch_max_requests=effective_prefetch_max_requests,
         prometheus_url=args.prometheus_url,
     )
     try:
@@ -275,6 +303,8 @@ def run_benchmark(args: RunBenchArgs) -> Path:
         _print_kv("dataset", args.dataset)
         _print_kv("max_samples", args.max_samples)
     _print_kv("block_size", args.block_size)
+    _print_kv("daser_prefetch_enabled", args.daser_prefetch_max_requests > 0)
+    _print_kv("daser_prefetch_max_requests", args.daser_prefetch_max_requests)
     _print_kv("output", prepare_path)
     if args.load_generator in ("vllm-bench", "vllm-bench-prefix"):
         prepare = {"config": vllm_bench.prepare_config(args, run_root)}
@@ -394,6 +424,7 @@ def _validate_run_args(args: RunBenchArgs) -> None:
     non_negative_ints = {
         "max_num_batched_tokens": args.max_num_batched_tokens,
         "max_context_tokens": args.max_context_tokens,
+        "daser_prefetch_max_requests": args.daser_prefetch_max_requests,
     }
     for name, value in non_negative_ints.items():
         if value < 0:
@@ -542,7 +573,7 @@ def _start_command(
         "--l2-size",
         str(derived_l2),
     ]
-    if backend_run.backend == "daser" and args.daser_prefetch_max_requests:
+    if backend_run.backend == "daser":
         command.extend(
             [
                 "--daser-prefetch-max-requests",

benchmarks/utils/servers.py

@@ -27,10 +27,38 @@
 LMCACHE_HTTP_PORT = 8080
 LMCACHE_MP_CONNECTOR_NAME = "DaseRBenchLMCacheMPConnector"
 LMCACHE_MP_CONNECTOR_MODULE = "benchmarks.utils.lmcache_connector_shim"
+DEFAULT_DASER_PREFETCH_MAX_REQUESTS = 2
 REPO_ROOT = Path(__file__).resolve().parents[2]
 LMCACHE_REPO_ROOT = REPO_ROOT.parent / "LMCache"
 
 
+def resolve_daser_prefetch_max_requests(
+    prefetch_enabled: bool,
+    override: int | None,
+) -> int:
+    """Resolve the effective scheduler prefetch worker limit.
+
+    Args:
+        prefetch_enabled: Whether the explicit boolean opt-in was supplied.
+        override: Optional numeric compatibility override. Zero explicitly
+            disables prefetch even when the boolean flag is present.
+
+    Returns:
+        Effective non-negative worker limit. The opt-in default is two.
+
+    Raises:
+        ValueError: If ``override`` is negative.
+
+    Thread-safety:
+        Pure function; safe to call from CLI parsing or service setup.
+    """
+    if override is not None:
+        if override < 0:
+            raise ValueError("daser prefetch worker limit must be non-negative")
+        return override
+    return DEFAULT_DASER_PREFETCH_MAX_REQUESTS if prefetch_enabled else 0
+
+
 @dataclass(frozen=True)
 class ServiceEndpoint:
     """HTTP endpoint for a benchmark service.
@@ -62,6 +90,8 @@ class BenchmarkManifest:
         log_dir: Log directory.
         pid_file: JSON file containing subprocess PIDs.
         block_size: vLLM KV block size in tokens.
+        prefetch_enabled: Whether scheduler prefetch is effectively enabled.
+        prefetch_max_requests: Effective scheduler prefetch worker limit.
 
     Thread-safety:
         Immutable value object.
@@ -79,6 +109,8 @@ class BenchmarkManifest:
     log_dir: str
     pid_file: str
     block_size: int = BLOCK_TOKENS
+    prefetch_enabled: bool = False
+    prefetch_max_requests: int = 0
 
     def write(self, path: str | Path) -> None:
         """Write manifest JSON atomically enough for local benchmark use."""
@@ -104,6 +136,10 @@ def read(cls, path: str | Path) -> "BenchmarkManifest":
         }
         payload["endpoints"] = endpoints
         payload.setdefault("block_size", 16)  # Legacy manifests predate this field.
+        payload.setdefault(
+            "prefetch_enabled", bool(payload.get("prefetch_max_requests", 0))
+        )
+        payload.setdefault("prefetch_max_requests", 0)
         return cls(**payload)
 
 
@@ -161,6 +197,8 @@ def __init__(
         """
         if tensor_parallel_size <= 0:
             raise ValueError("tensor_parallel_size must be positive")
+        if daser_prefetch_max_requests < 0:
+            raise ValueError("daser_prefetch_max_requests must be non-negative")
         self.run_id = run_id
         self.backend = backend
         self.model = model
@@ -234,6 +272,8 @@ def manifest(self) -> BenchmarkManifest:
             log_dir=str(self.log_dir),
             pid_file=str(self.pid_file),
             block_size=self.block_size,
+            prefetch_enabled=self.daser_prefetch_max_requests > 0,
+            prefetch_max_requests=self.daser_prefetch_max_requests,
         )
 
     async def start_lmcache_mp_server(self) -> None:
@@ -347,17 +387,29 @@ def lmcache_process_env(self) -> dict[str, str]:
 
     async def start_vllm_daser(self) -> None:
         """Start vLLM with DaseR connector."""
-        kv_config = {
+        await self._start_vllm("vllm_daser.log", self.daser_kv_transfer_config())
+
+    def daser_kv_transfer_config(self) -> dict[str, Any]:
+        """Return the vLLM KV transfer payload for a DaseR service.
+
+        Returns:
+            JSON-serialisable vLLM connector configuration, including the
+            effective prefetch state used by the scheduler role.
+
+        Thread-safety:
+            Pure calculation over immutable service configuration.
+        """
+        return {
             "kv_connector": "DaserConnector",
             "kv_connector_module_path": "daser.connector.daser_connector",
             "kv_role": "kv_both",
             "kv_connector_extra_config": {
                 "socket_path": str(self.socket_path),
                 "cache_reuse_mode": self.reuse_mode,
                 "prefetch_max_requests": self.daser_prefetch_max_requests,
+                "prefetch_enabled": self.daser_prefetch_max_requests > 0,
             },
         }
-        await self._start_vllm("vllm_daser.log", kv_config)
 
     async def start_daser_server(self) -> None:
         """Start DaseR HTTP + IPC server."""

docs/development.md

@@ -187,6 +187,11 @@ under one run directory. Use a specific backend name to run only one row of the
 matrix; `--backend daser` still honors `--cache-reuse-mode` for compatibility.
 `--cache-reuse-mode` is only passed to DaseR rows; baseline and LMCache use the
 same prepared prompts and record `reuse_mode: none`.
+Scheduler-side DaseR prefetch is off by default. Pass `--daser-prefetch` to
+enable it with two workers, or use the compatibility override
+`--daser-prefetch-max-requests N` (including `0` to force it off). The numeric
+override takes precedence over the boolean flag, and the effective state is
+recorded in the DaseR manifest.
 The Python entry point prints stage separators for prepare, backend start,
 cold/warm load, a final `== COMPARISON SUMMARY ==` with cold, warm,
 correctness, elapsed-time, and throughput fields, and the final `run_root`.