Cache-size sweep: build_meta is O(|cache|), +85.6 μs / 1k blocks

Follow-up to Microbench 3 that finally tests H5 (cache-size
dependence) and instruments worker-side connector callbacks the
original patch missed.

Patch v2 (apply_step_timing_v2.py) adds:
  scheduler: `cache_size` field in engine_step.jsonl
  worker:    `get_finished_us` + `start_load_kv_us` in worker_step.r0.jsonl
  uses BLOCK_BEGIN/END sentinels for safe multi-line revert
  (the original v1 patch survives this v2's apply/revert cycle)

Driver: continuous open-loop (1.5 req/s, 4096x256 random per req)
that lets APC fill from 0 → ceiling within one vLLM lifetime so a
single run produces the full cache_size sweep. Decode-only steps
are filtered post-hoc to remove prefill-mix variance.

Findings (H20 96GB, ceiling reached ~17.5k blocks; n=15-18k decode
steps per config):

  config         | slope (μs / 1k blocks) | step_dur p50 @ |cache|=16.6k
  ---------------|------------------------|-----------------------------
  mooncake_both  | +85.6                  | 1528 μs (build_meta=1442, 94%)
  noop_connector | -0.8 (≈0)              |  79 μs
  plain          | +1.0 (≈0)              |  84 μs

  Worker-side get_finished p50/p90/p99 (μs/step):
    mooncake_both:  180 / 257 / 333
    noop_connector:   0 /   0 /   2

H5 PASSES. mooncake_both step_duration scales linearly with |cache|
because build_connector_meta walks set(cache.keys()) every step
(`mooncake_connector.py:434-450`). plain and noop are flat.

The previously-uninstrumented get_finished() adds a constant
180 μs/step on top — two `run_coroutine_threadsafe(...).result()`
blocking waits in kv_both mode (`mooncake_connector.py:1107-1137`)
fire every step even when no transfer is pending.

Trace-replay reconciliation (APC ≈ 79% → |cache| ≈ 13k blocks):
  build_meta @ 13k ≈ 1060 μs + get_finished ≈ 180 μs = 1.24 ms/step
  On ~7 ms decode forward → +15-20% TPOT per step.
  This explains most of the trace-replay +25% TPOT p90 gap from
  single-instance per-step cost alone, leaving a smaller residual
  for multi-instance coupling than originally assumed.

Two clear fixes pointed out in REPORT.md:
  1. replace O(|cache|) per-step walk with incremental delta
     listener using block_pool's add/remove callbacks
  2. short-circuit get_finished() when both producer/consumer
     queues are empty in kv_both

Heavy raw artifacts (engine_step.jsonl, vllm_stdout/stderr,
.vllm.pid) are .gitignored — they re-derive from `bash run_all.sh`
and SUMMARY.md / per_config.json fully capture the conclusions.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

microbench/connector_tax/cache_sweep/run_cache_sweep.py

@@ -0,0 +1,220 @@
+#!/usr/bin/env python3
+"""Continuous open-loop driver for the cache-size sweep.
+
+The intent is to fill the prefix-cache from 0 up to GPU ceiling within a
+single vLLM lifetime, so the per-step `cache_size` field (added by
+`apply_step_timing_v2.py`) sweeps through every value the engine can
+hold. Offline analysis bins by `cache_size` to recover the per-step
+overhead curve.
+
+Workload:
+  - Open-loop Poisson at fixed rate (default 1.5 req/s).
+  - Random per-request content (UUID + hash, calibrated to ~4096
+    tokens). Zero prefix-cache hits ⇒ cache strictly grows until
+    LRU eviction kicks in.
+  - max_tokens / min_tokens = 256, temperature=0, ignore_eos=True.
+  - Duration default 8 min.
+
+Writes per-request metrics to `requests.jsonl`. The patch emits
+per-step rows to `engine_step.jsonl` (set via AGENTIC_STEP_LOG_PATH)
+and worker rows to `worker_step.jsonl.r0` (set via
+CT_WORKER_STEP_LOG_PATH).
+
+Usage:
+  run_cache_sweep.py \\
+      --url http://127.0.0.1:8000/v1/chat/completions \\
+      --model /path/to/qwen \\
+      --rate 1.5 --duration 480 \\
+      --output-dir results/<date>/<config>
+"""
+
+import argparse
+import asyncio
+import hashlib
+import json
+import random
+import time
+import uuid
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+import httpx
+
+
+@dataclass
+class ReqMetric:
+    req_id: str
+    rate_target: float
+    input_tokens_target: int
+    output_tokens_target: int
+    t_send_ns: int
+    t_first_token_ns: int | None = None
+    t_last_token_ns: int | None = None
+    prompt_tokens: int = 0
+    completion_tokens: int = 0
+    inflight_at_send: int = 0
+    error: str | None = None
+
+
+def make_random_prompt(target_tokens: int) -> str:
+    """Same calibration as the bench_loop.py used elsewhere:
+    'Block N: <32-hex>' tokenizes to ~35 tokens on Qwen3-Coder."""
+    n_parts = max(1, target_tokens // 35)
+    seed = uuid.uuid4().hex
+    parts = []
+    for i in range(n_parts):
+        h = hashlib.md5(f"{seed}_{i}_{time.time_ns()}".encode()).hexdigest()
+        parts.append(f"Block {i}: {h}")
+    return " ".join(parts)
+
+
+async def send_one(client, url, model, inp_tokens, out_tokens,
+                   rate, inflight, inflight_cap, fh):
+    rid = uuid.uuid4().hex[:16]
+    if inflight[0] >= inflight_cap:
+        m = ReqMetric(req_id=rid, rate_target=rate,
+                      input_tokens_target=inp_tokens,
+                      output_tokens_target=out_tokens,
+                      t_send_ns=time.perf_counter_ns(),
+                      inflight_at_send=inflight[0],
+                      error="dropped_inflight_cap")
+        fh.write(json.dumps(asdict(m)) + "\n")
+        return
+
+    inflight[0] += 1
+    m = ReqMetric(req_id=rid, rate_target=rate,
+                  input_tokens_target=inp_tokens,
+                  output_tokens_target=out_tokens,
+                  t_send_ns=time.perf_counter_ns(),
+                  inflight_at_send=inflight[0])
+    try:
+        prompt = make_random_prompt(inp_tokens)
+        payload = {
+            "model": model,
+            "messages": [{"role": "user", "content": prompt}],
+            "max_tokens": out_tokens,
+            "min_tokens": out_tokens,
+            "temperature": 0,
+            "ignore_eos": True,
+            "stream": True,
+            "stream_options": {"include_usage": True},
+        }
+        async with client.stream("POST", url, json=payload, timeout=600.0) as resp:
+            resp.raise_for_status()
+            async for line in resp.aiter_lines():
+                if not line.startswith("data: "):
+                    continue
+                data = line[6:]
+                if data.strip() == "[DONE]":
+                    break
+                try:
+                    chunk = json.loads(data)
+                except json.JSONDecodeError:
+                    continue
+                usage = chunk.get("usage")
+                if usage:
+                    m.prompt_tokens = usage.get("prompt_tokens", m.prompt_tokens)
+                    m.completion_tokens = usage.get(
+                        "completion_tokens", m.completion_tokens)
+                choices = chunk.get("choices") or []
+                if not choices:
+                    continue
+                delta = choices[0].get("delta", {})
+                if "role" in delta:
+                    continue
+                now = time.perf_counter_ns()
+                if m.t_first_token_ns is None:
+                    m.t_first_token_ns = now
+                m.t_last_token_ns = now
+    except Exception as e:
+        m.error = f"{type(e).__name__}: {e}"
+    finally:
+        inflight[0] -= 1
+    fh.write(json.dumps(asdict(m)) + "\n")
+
+
+async def main_async(args):
+    out_dir = Path(args.output_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    req_path = out_dir / "requests.jsonl"
+
+    inflight = [0]
+    pending: list[asyncio.Task] = []
+    interval_mean = 1.0 / args.rate
+    rng = random.Random(int(time.time_ns()) & 0xFFFFFFFF)
+
+    print(f"[bench] rate={args.rate} shape=({args.input_tokens},{args.output_tokens}) "
+          f"duration={args.duration}s output={out_dir}")
+
+    fh = open(req_path, "a", buffering=1)
+    t0 = time.perf_counter()
+    last_print = t0
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(600.0)) as client:
+        # producer
+        async def producer():
+            while time.perf_counter() - t0 < args.duration:
+                pending.append(asyncio.create_task(
+                    send_one(client, args.url, args.model,
+                             args.input_tokens, args.output_tokens,
+                             args.rate, inflight, args.inflight_cap, fh)
+                ))
+                await asyncio.sleep(rng.expovariate(1.0 / interval_mean))
+
+        # heartbeat
+        async def heartbeat():
+            nonlocal last_print
+            while time.perf_counter() - t0 < args.duration + 1:
+                now = time.perf_counter()
+                if now - last_print >= 30:
+                    print(f"  t+{int(now - t0):4d}s inflight={inflight[0]} "
+                          f"pending={len(pending)}")
+                    last_print = now
+                await asyncio.sleep(2.0)
+
+        prod = asyncio.create_task(producer())
+        hb = asyncio.create_task(heartbeat())
+        await prod
+        hb.cancel()
+        try:
+            await hb
+        except asyncio.CancelledError:
+            pass
+
+        # final drain
+        if pending:
+            await asyncio.gather(*pending, return_exceptions=True)
+
+    fh.close()
+
+    # Quick summary so the orchestrator can sanity-check the cell ran.
+    n_lines = sum(1 for _ in open(req_path))
+    with open(out_dir / "run_summary.json", "w") as f:
+        json.dump({
+            "rate": args.rate,
+            "input_tokens": args.input_tokens,
+            "output_tokens": args.output_tokens,
+            "duration_target_s": args.duration,
+            "duration_actual_s": time.perf_counter() - t0,
+            "n_requests": n_lines,
+        }, f, indent=2)
+    print(f"[done] {n_lines} requests in {time.perf_counter() - t0:.1f}s")
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--url", required=True)
+    ap.add_argument("--model", required=True)
+    ap.add_argument("--rate", type=float, default=1.5)
+    ap.add_argument("--input-tokens", type=int, default=4096)
+    ap.add_argument("--output-tokens", type=int, default=256)
+    ap.add_argument("--duration", type=float, default=480.0,
+                    help="Total run duration in seconds (default 8 min)")
+    ap.add_argument("--inflight-cap", type=int, default=256)
+    ap.add_argument("--output-dir", required=True)
+    args = ap.parse_args()
+    asyncio.run(main_async(args))
+
+
+if __name__ == "__main__":
+    main()