kv-router: conditional disagg + overlap-driven KV migration on v1.4.1

benchmarks/di-dsv4/README.md

@@ -0,0 +1,30 @@
+# DSv4-Flash benchmark harness
+
+Scripts used to evaluate dynamo disagg / KV migration / engine configs against
+the production fleet.
+
+| script | purpose |
+| --- | --- |
+| `migtest.sh` | aiperf sweep vs a second fleet; warms both endpoints first |
+| `mirror_h2h.sh` | mirror the same prod shards to two fleets, compare N windows |
+| `compare.sh` | one fleet vs prod, N windows |
+| `h2h.sh` | synthetic 7v7 aiperf |
+| `kvxfer_force.py` | force a KV migration and check output determinism at temp=0 |
+| `kvxfer_cross.py` | migration-vs-cold-prefill crossover sweep |
+| `summarize.py` | median-per-window summary of compare/mirror logs |
+
+## Hard-won rules
+
+1. **Warm before the first measured point.** The first aiperf concurrency level
+   after a fleet boot reads ~5.8x slow (1.53 vs 8.90 req/s on identical config).
+   This produced three false "engine regression" verdicts.
+2. **Never trust pod logs for rates.** The container log holds only ~900 lines,
+   so `--since=1h` and `--since=1m` return nearly the same count. Use
+   `dynamo_frontend_requests_total{status,error_type}` instead.
+3. **Preflight both endpoints.** Port-forwards die when a pod rolls, and the
+   harness silently records `req/s=0.00` rather than failing.
+4. **Give each synthetic prompt a distinct first 256 tokens**, or they collide
+   on one session id and the seed request 400s.
+5. **Mirror the same shards to both candidates.** Synthetic traffic that is too
+   uniform/cacheable flatters KV-aware routing; a shard-sliced mirror is a
+   load-based sample, so both sides must get the identical slice.

benchmarks/di-dsv4/compare.sh

@@ -0,0 +1,27 @@
+set -u
+LABEL="$1"; WINDOWS="${2:-7}"; GAP="${3:-300}"
+cd /data/home/pernekhan/backend
+export PATH=$HOME/miniconda3/bin:$PATH
+PY=$HOME/miniconda3/envs/di-main/bin/python
+D='deepseek-ai/DeepSeek-V4-Flash-0731-roce-disagg'
+P='deepseek-ai/DeepSeek-V4-Flash-0731'
+
+q() { $PY -m scripts.cli vm-query --instant --label='' --query "$1" 2>/dev/null | tail -1 | awk -F, '{print $NF+0}'; }
+tq() { q "histogram_quantile($1, sum(rate(vllm:time_to_first_token_seconds_bucket{model_name=\"$2\"}[5m])) by (le))"; }
+hit() { q "sum(rate(vllm:prefix_cache_hits_total{model_name=\"$1\"$2}[5m]))/sum(rate(vllm:prefix_cache_queries_total{model_name=\"$1\"$2}[5m]))"; }
+gen() { q "sum(rate(vllm:generation_tokens_total{model_name=\"$1\"}[5m]))"; }
+req() { q "sum(rate(vllm:request_success_total{model_name=\"$1\"}[5m]))"; }
+
+echo "CONFIG=$LABEL windows=$WINDOWS gap=${GAP}s"
+for i in $(seq 1 "$WINDOWS"); do
+  echo "W$i t=$(date -u +%H:%M)" \
+    "hitD=$(hit "$D" ',dynamo_component="backend"')" \
+    "hitP=$(hit "$P" '')" \
+    "t50D=$(tq 0.50 "$D") t90D=$(tq 0.90 "$D") t99D=$(tq 0.99 "$D")" \
+    "t50P=$(tq 0.50 "$P") t90P=$(tq 0.90 "$P") t99P=$(tq 0.99 "$P")" \
+    "genD=$(gen "$D") genP=$(gen "$P")" \
+    "reqD=$(req "$D") reqP=$(req "$P")" \
+    "eng=$(q "count(vllm:num_requests_running{model_name=\"$P\"})")"
+  [ "$i" -lt "$WINDOWS" ] && sleep "$GAP"
+done
+echo "DONE_$LABEL"

benchmarks/di-dsv4/h2h.sh

@@ -0,0 +1,48 @@
+set -u
+T=/home/pernekhan/.claude/jobs/3aed5ed4/tmp
+export PATH=$HOME/miniconda3/bin:$PATH
+TOKDIR=$T/ds4tok/tokenizers/di--deepseek-ai--DeepSeek-V4-Flash-0731--D9jmKKxS
+KEY=$(cat $T/.tok)
+OUT=$T/h2h; mkdir -p $OUT
+
+# prod-shaped: long shared prefixes (cacheable) + unique tail, ~300 output tokens
+PREFIX_N=64
+PREFIX_LEN=6000
+IN_MEAN=3000
+IN_STD=2000
+OSL=300
+
+run() {
+  NAME=$1; MODEL=$2; URL=$3; C=$4; N=$5; shift 5
+  D=$OUT/${NAME}_c${C}
+  rm -rf $D
+  timeout 2400 conda run -n di-main --no-capture-output aiperf profile \
+    -m "$MODEL" --url "$URL" --endpoint-type chat --streaming \
+    --tokenizer "$TOKDIR" --tokenizer-trust-remote-code \
+    --num-prefix-prompts $PREFIX_N --prefix-prompt-length $PREFIX_LEN \
+    --synthetic-input-tokens-mean $IN_MEAN --synthetic-input-tokens-stddev $IN_STD \
+    --output-tokens-mean $OSL --output-tokens-stddev 0 \
+    --concurrency "$C" --request-count "$N" --num-warmup-requests 8 \
+    --random-seed 8800 --output-artifact-dir "$D" "$@" >/dev/null 2>&1
+  python3 - "$D/profile_export_aiperf.json" "$NAME" "$C" <<'PY'
+import json,sys
+try: d=json.load(open(sys.argv[1]))
+except Exception:
+    print(f"  {sys.argv[2]:<8} c={sys.argv[3]:<4} FAILED"); raise SystemExit
+g=lambda k,f='avg': (d.get(k) or {}).get(f) or 0
+print(f"  {sys.argv[2]:<8} c={sys.argv[3]:<4} req/s={g('request_throughput'):6.2f} "
+      f"TTFT p50={g('time_to_first_token','p50'):7.0f} p90={g('time_to_first_token','p90'):7.0f} p99={g('time_to_first_token','p99'):8.0f}ms "
+      f"ITL p50={g('inter_token_latency','p50'):5.1f}ms out_tok/s={g('output_token_throughput'):7.0f}")
+PY
+}
+
+echo "=== 7 GPU vs 7 GPU, identical synthetic traffic (seed 8800) ==="
+echo "    prefix pool=$PREFIX_N x ${PREFIX_LEN}tok, unique tail mean=$IN_MEAN, out=$OSL"
+for C in 8 24 48 96; do
+  N=$(( C * 12 )); [ $N -lt 200 ] && N=200
+  run vllm7 "Pernekhan/DeepSeek-V4-Flash-0731-test" "http://localhost:18002" $C $N \
+      --custom-endpoint /v1/openai/chat/completions --api-key "$KEY"
+  run dynamo7 "deepseek-ai/DeepSeek-V4-Flash-0731-roce-disagg" "http://localhost:18001" $C $N
+  echo ""
+done
+echo H2H_COMPLETE

benchmarks/di-dsv4/kvxfer_cross.py

@@ -0,0 +1,33 @@
+import json, urllib.request, time, sys
+URL="http://localhost:80/v1/chat/completions"; MODEL="deepseek-ai/DeepSeek-V4-Flash-0731-roce-disagg"
+A,B,REPS = sys.argv[1], sys.argv[2], int(sys.argv[3])
+UNIT="Timing probe segment for decode to decode key value transfer measurement. "
+def body_of(prompt):
+    return {"model":MODEL,"temperature":0.0,"max_tokens":16,
+            "messages":[{"role":"user","content":prompt}]}
+def post(prompt, worker):
+    r=urllib.request.Request(URL,data=json.dumps(body_of(prompt)).encode(),
+                             headers={"Content-Type":"application/json"})
+    if worker: r.add_header("x-dynamo-worker-instance-id",worker)
+    t=time.time()
+    try:
+        with urllib.request.urlopen(r,timeout=600) as resp: d=json.load(resp)
+    except Exception as e: return time.time()-t,{"error":str(e)}
+    return time.time()-t,d
+def mk(tag,mult): return (f"Probe {tag} segment for decode to decode transfer measurement. "*mult) + "\n\nQuestion: say ok."
+for mult,label in ((1200,"21k"),(3200,"67k")):
+    cold=[]; mig=[]
+    for i in range(REPS):
+        stamp=f"{label}-{i}-{int(time.time())}"
+        el,d=post(mk("cold-"+stamp,mult),B)
+        pt=(d.get("usage") or {}).get("prompt_tokens")
+        cold.append(el); print(f"COLD  {label} rep{i} sec={el:.3f} tokens={pt} err={d.get('error')}"); sys.stdout.flush()
+        p=mk("mig-"+stamp,mult)
+        el2,d2=post(p,A); print(f"SEED  {label} rep{i} sec={el2:.3f} tokens={(d2.get('usage') or {}).get('prompt_tokens')} err={d2.get('error')}"); sys.stdout.flush()
+        time.sleep(75)
+        el3,d3=post(p,B)
+        mig.append(el3); print(f"MIGR  {label} rep{i} sec={el3:.3f} tokens={(d3.get('usage') or {}).get('prompt_tokens')} err={d3.get('error')}"); sys.stdout.flush()
+    if cold and mig:
+        import statistics as st
+        c=st.median(cold); m=st.median(mig)
+        print(f"RESULT {label} cold med={c:.3f} min={min(cold):.3f} | migrated med={m:.3f} min={min(mig):.3f} | speedup_med={c/m:.2f}x speedup_min={min(cold)/min(mig):.2f}x")

benchmarks/di-dsv4/kvxfer_force.py

@@ -0,0 +1,39 @@
+import json, urllib.request, time, sys
+
+URL = "http://localhost:80/v1/chat/completions"
+MODEL = "deepseek-ai/DeepSeek-V4-Flash-0731-roce-disagg"
+A, B, WAIT = sys.argv[1], sys.argv[2], int(sys.argv[3])
+MULT = int(sys.argv[4]) if len(sys.argv) > 4 else 260
+FILLER = ("The migration harness verifies decode-to-decode key value transfer. " * MULT)
+PROMPT = FILLER + "\n\nQuestion: Reply with exactly the five words: alpha bravo charlie delta echo."
+
+def post(prompt, worker, max_tokens=64):
+    body = {"model": MODEL, "temperature": 0.0, "max_tokens": max_tokens,
+            "messages": [{"role": "user", "content": prompt}]}
+    req = urllib.request.Request(URL, data=json.dumps(body).encode(),
+                                 headers={"Content-Type": "application/json"})
+    if worker:
+        req.add_header("x-dynamo-worker-instance-id", worker)
+    try:
+        with urllib.request.urlopen(req, timeout=180) as r:
+            return json.load(r)
+    except Exception as e:
+        return {"error": str(e)}
+
+def summarize(tag, d):
+    ch = d.get("choices", [{}])[0].get("message", {}).get("content")
+    u = d.get("usage", {}) or {}
+    print(tag, json.dumps({
+        "out": ch,
+        "prompt_tokens": u.get("prompt_tokens"),
+        "cached": (u.get("prompt_tokens_details") or {}).get("cached_tokens"),
+        "completion_tokens": u.get("completion_tokens"),
+        "err": d.get("error"),
+    }))
+    return ch
+
+r1 = post(PROMPT, A); o1 = summarize("SEED_A", r1)
+print("WAIT", WAIT); sys.stdout.flush()
+time.sleep(WAIT)
+r2 = post(PROMPT, B); o2 = summarize("FORCED_B", r2)
+print("IDENTICAL", json.dumps(o1 is not None and o1 == o2))

benchmarks/di-dsv4/migtest.sh

@@ -0,0 +1,90 @@
+set -u
+T=/home/pernekhan/.claude/jobs/3aed5ed4/tmp
+export PATH=$HOME/miniconda3/bin:$PATH
+PY=$HOME/miniconda3/envs/di-main/bin/python
+TOKDIR=$T/ds4tok/tokenizers/di--deepseek-ai--DeepSeek-V4-Flash-0731--D9jmKKxS
+KEY=$(cat $T/.tok)
+OUT=$T/migtest; mkdir -p $OUT
+NS=deepinfra
+D='deepseek-ai/DeepSeek-V4-Flash-0731-roce-disagg'
+V='Pernekhan/DeepSeek-V4-Flash-0731-test'
+
+# Prompts must clear the migration gate: overlap_blocks*256 >= 16384 tokens.
+# 12 shared prefixes x 24k tokens, ~2k unique tail => ~26k prompts, ~24k cacheable.
+PREFIX_N=12
+PREFIX_LEN=24000
+IN_MEAN=2000
+OSL=300
+
+fe() { kubectl -n $NS get pods --no-headers | grep 'roce-disagg-fron' | grep -v Terminating | head -1 | awk '{print $1}'; }
+
+run() {
+  NAME=$1; MODEL=$2; URL=$3; C=$4; N=$5; shift 5
+  DIR=$OUT/${NAME}_c${C}; rm -rf $DIR
+  timeout 3600 conda run -n di-main --no-capture-output aiperf profile \
+    -m "$MODEL" --url "$URL" --endpoint-type chat --streaming \
+    --tokenizer "$TOKDIR" --tokenizer-trust-remote-code \
+    --num-prefix-prompts $PREFIX_N --prefix-prompt-length $PREFIX_LEN \
+    --synthetic-input-tokens-mean $IN_MEAN --synthetic-input-tokens-stddev 500 \
+    --output-tokens-mean $OSL --output-tokens-stddev 0 \
+    --concurrency "$C" --request-count "$N" --num-warmup-requests 6 \
+    --random-seed 4242 --output-artifact-dir "$DIR" "$@" >/dev/null 2>&1
+  $PY - "$DIR/profile_export_aiperf.json" "$NAME" "$C" <<'PY'
+import json,sys
+try: d=json.load(open(sys.argv[1]))
+except Exception:
+    print(f"  {sys.argv[2]:<9} c={sys.argv[3]:<4} FAILED"); raise SystemExit
+g=lambda k,f='avg': (d.get(k) or {}).get(f) or 0
+print(f"  {sys.argv[2]:<9} c={sys.argv[3]:<4} req/s={g('request_throughput'):6.2f} "
+      f"TTFT p50={g('time_to_first_token','p50'):7.0f} p90={g('time_to_first_token','p90'):7.0f} p99={g('time_to_first_token','p99'):8.0f}ms "
+      f"ITL={g('inter_token_latency','p50'):5.1f}ms tok/s={g('output_token_throughput'):7.0f}")
+PY
+}
+
+# Warm both fleets before the first measured point. Without this the first
+# concurrency level absorbs all CUDA-graph/JIT warmup and reads ~5x slow --
+# it produced three bogus "engine regression" verdicts before it was caught.
+warmup() {
+  timeout 900 conda run -n di-main --no-capture-output aiperf profile \
+    -m "$2" --url "$3" --endpoint-type chat --streaming \
+    --tokenizer "$TOKDIR" --tokenizer-trust-remote-code \
+    --num-prefix-prompts $PREFIX_N --prefix-prompt-length $PREFIX_LEN \
+    --synthetic-input-tokens-mean $IN_MEAN --output-tokens-mean $OSL \
+    --concurrency 16 --request-count 64 --num-warmup-requests 4 \
+    --random-seed 4242 --output-artifact-dir "$OUT/warm_$1" "${@:4}" >/dev/null 2>&1
+  echo "    warmed $1"
+}
+warmup dynamo7 "$D" "http://localhost:18041"
+warmup vllm7 "$V" "http://localhost:18042" --custom-endpoint /v1/openai/chat/completions --api-key "$KEY"
+echo "WARMUP_DONE"
+
+echo "=== 7 dynamo GPU vs 7 standalone vLLM GPU, no mirror, seed 4242 ==="
+echo "    prompts: ${PREFIX_N} shared prefixes x ${PREFIX_LEN} tok + ~${IN_MEAN} tail (clears the 16384 gate)"
+echo ""
+
+for C in 24 48 128; do
+  N=$(( C * 8 )); [ $N -lt 150 ] && N=150
+
+  # capture the dynamo frontend log for the whole dynamo run so migration
+  # counts are complete (container log holds only ~900 lines)
+  FP=$(fe)
+  LOG=$OUT/fe_c${C}.log
+  ( kubectl -n $NS logs -f "$FP" --since=5s > "$LOG" 2>/dev/null ) &
+  TAILPID=$!
+  sleep 3
+  run dynamo7 "$D" "http://localhost:18041" $C $N
+  sleep 5
+  kill $TAILPID 2>/dev/null; wait $TAILPID 2>/dev/null
+
+  SP=$(grep -c 'KVMIGRATE_SPILL' "$LOG" 2>/dev/null || echo 0)
+  HS=$(grep -c 'handshake obtained' "$LOG" 2>/dev/null || echo 0)
+  MG=$(grep -c 'KVMIGRATE: decoding on target' "$LOG" 2>/dev/null || echo 0)
+  PF=$(grep -c 'producer dispatch failed' "$LOG" 2>/dev/null || echo 0)
+  echo "    migration@c${C}: spills=$SP handshakes=$HS migrated=$MG producer_failed=$PF (of $N requests)"
+  grep -oE 'migratable_tokens=[0-9]+' "$LOG" 2>/dev/null | head -3 | sed 's/^/      sample /'
+
+  run vllm7 "$V" "http://localhost:18042" $C $N \
+      --custom-endpoint /v1/openai/chat/completions --api-key "$KEY"
+  echo ""
+done
+echo MIGTEST_COMPLETE

benchmarks/di-dsv4/mirror_h2h.sh

@@ -0,0 +1,29 @@
+set -u
+LABEL="${1:-run}"; WINDOWS="${2:-7}"; GAP="${3:-300}"
+cd /data/home/pernekhan/backend
+export PATH=$HOME/miniconda3/bin:$PATH
+PY=$HOME/miniconda3/envs/di-main/bin/python
+D='deepseek-ai/DeepSeek-V4-Flash-0731-roce-disagg'
+V='Pernekhan/DeepSeek-V4-Flash-0731-test'
+
+q() { $PY -m scripts.cli vm-query --instant --label='' --query "$1" 2>/dev/null | tail -1 | awk -F, '{print $NF+0}'; }
+tq() { q "histogram_quantile($1, sum(rate(vllm:time_to_first_token_seconds_bucket{model_name=\"$2\"}[5m])) by (le))"; }
+hit() { q "sum(rate(vllm:prefix_cache_hits_total{model_name=\"$1\"}[5m]))/sum(rate(vllm:prefix_cache_queries_total{model_name=\"$1\"}[5m]))"; }
+gen() { q "sum(rate(vllm:generation_tokens_total{model_name=\"$1\"}[5m]))"; }
+req() { q "sum(rate(vllm:request_success_total{model_name=\"$1\"}[5m]))"; }
+eng() { q "count(vllm:num_requests_running{model_name=\"$1\"})"; }
+kv()  { q "avg(vllm:kv_cache_usage_perc{model_name=\"$1\"})"; }
+
+echo "CONFIG=$LABEL windows=$WINDOWS (same 7 prod shards mirrored to both)"
+for i in $(seq 1 "$WINDOWS"); do
+  echo "W$i t=$(date -u +%H:%M)" \
+    "hitD=$(hit "$D") hitV=$(hit "$V")" \
+    "t50D=$(tq 0.50 "$D") t90D=$(tq 0.90 "$D") t99D=$(tq 0.99 "$D")" \
+    "t50V=$(tq 0.50 "$V") t90V=$(tq 0.90 "$V") t99V=$(tq 0.99 "$V")" \
+    "genD=$(gen "$D") genV=$(gen "$V")" \
+    "reqD=$(req "$D") reqV=$(req "$V")" \
+    "engD=$(eng "$D") engV=$(eng "$V")" \
+    "kvD=$(kv "$D") kvV=$(kv "$V")"
+  [ "$i" -lt "$WINDOWS" ] && sleep "$GAP"
+done
+echo "DONE_$LABEL"

benchmarks/di-dsv4/summarize.py

@@ -0,0 +1,61 @@
+import sys, statistics as st
+
+GPUS_DISAGG = 13
+
+def load(path):
+    rows = []
+    for line in open(path):
+        if not line.startswith('W'):
+            continue
+        d = {}
+        for kv in line.split():
+            if '=' in kv:
+                k, v = kv.split('=', 1)
+                d[k] = v
+        try:
+            rows.append({k: float(v) for k, v in d.items() if k != 't'})
+        except ValueError:
+            pass
+    return rows
+
+def med(rows, key):
+    vals = [r[key] for r in rows if key in r and r[key] > 0]
+    return st.median(vals) if vals else float('nan')
+
+def report(label, path):
+    rows = load(path)
+    if not rows:
+        print(f"  {label}: no windows"); return None
+    eng = med(rows, 'eng')
+    out = {
+        'n': len(rows),
+        'hitD': med(rows, 'hitD'), 'hitP': med(rows, 'hitP'),
+        't50D': med(rows, 't50D') * 1000, 't90D': med(rows, 't90D') * 1000, 't99D': med(rows, 't99D') * 1000,
+        't50P': med(rows, 't50P') * 1000, 't90P': med(rows, 't90P') * 1000, 't99P': med(rows, 't99P') * 1000,
+        'genD_gpu': med(rows, 'genD') / GPUS_DISAGG, 'genP_gpu': med(rows, 'genP') / eng,
+        'reqD_gpu': med(rows, 'reqD') / GPUS_DISAGG, 'reqP_gpu': med(rows, 'reqP') / eng,
+        'eng': eng,
+    }
+    print(f"\n  === {label} (n={out['n']} windows, prod engines={eng:.0f}) ===")
+    print(f"    {'metric':<22} {'disagg':>10} {'prod':>10} {'ratio':>9}")
+    def row(name, d, p, better_low=False, fmt="{:.0f}"):
+        r = (p / d) if better_low else (d / p)
+        mark = "  <-- win" if r > 1.0 else ""
+        print(f"    {name:<22} {fmt.format(d):>10} {fmt.format(p):>10} {r:>8.2f}x{mark}")
+    row('cache hit', out['hitD'], out['hitP'], fmt="{:.3f}")
+    row('TTFT p50 (ms)', out['t50D'], out['t50P'], better_low=True)
+    row('TTFT p90 (ms)', out['t90D'], out['t90P'], better_low=True)
+    row('TTFT p99 (ms)', out['t99D'], out['t99P'], better_low=True)
+    row('gen tok/s per GPU', out['genD_gpu'], out['genP_gpu'])
+    row('req/s per GPU', out['reqD_gpu'], out['reqP_gpu'])
+    return out
+
+a = report('A: queue gate ON', sys.argv[1])
+b = report('B: queue gate OFF', sys.argv[2]) if len(sys.argv) > 2 else None
+
+if a and b:
+    print("\n  === A vs B (disagg only, median of windows) ===")
+    for k, name, low in (('t50D','TTFT p50',True), ('t90D','TTFT p90',True), ('t99D','TTFT p99',True),
+                         ('hitD','cache hit',False), ('genD_gpu','gen tok/s/GPU',False), ('reqD_gpu','req/s/GPU',False)):
+        delta = (a[k] / b[k]) if low else (b[k] / a[k])
+        print(f"    {name:<16} A={a[k]:>10.3f}  B={b[k]:>10.3f}   B is {delta:.2f}x {'better' if delta>1 else 'worse'}")