Compare commits

...

9 Commits

Author SHA1 Message Date
54e1f5266a MB5 PD ablation v2 results: concurrency axis + reuse 3-way writeup
- fig3_conc32k.json + fig3_concurrency_axis.png: agentic-corner concurrency
  sweep (in=32768, reuse=0.984, out=128), N 8->128, PD capped 600s / colo
  uncapped. colo completes 100% at every N (graceful, E2E 2.4s->81s); every
  static PD split collapses, earlier as N rises (viable only N<=16; <27% by N=32).
- analysis/mb5_pd_ablation/README.md: self-contained v2 writeup. Reuse axis
  3-way (A=d2048/o256, C=d2048/o128, B=d1024/o128) decomposes shape: output
  ~negligible, delta (real prefill/turn) dominant; crossover to colo at reuse
  ~90-95% robust. Run on dash2 (dash0 NICs faulted for Mooncake).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-01 09:35:25 +08:00
3f997fda14 MB5 PD ablation v2 tooling: conc completion-panel plot + gpu_monitor dep
- plot_pd_crossover.py fig_conc: lead with request-completion % (the honest
  collapse signal; latency percentiles count successes only), then mean-E2E /
  TPS; note PD-capped/colo-uncapped in the title.
- add microbench/fresh_setup/gpu_monitor.sh (referenced by the committed
  mb5_run_gpu.sh:73 for per-GPU util collection).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-01 09:35:05 +08:00
19c443e3bc paper f2a: reuse-topology decomposition + mixture-sensitivity sweep
Full-trace analysis backing figure 2a on the real 2h cluster trace:

- f2a_reuse_topology_analyze.py: infinite-KV-cache (LRU) decomposition of
  prefix-cache reuse hits into intra-session vs cross-session, by most-recent
  prior holder of each content-addressed block.
- f2a_mixture_sweep.py: sensitivity of the intra/cross split to the
  single-turn session fraction (tests whether the 93%-intra sample vs 54.6%
  full-trace gap is session-mixture selection bias) -- keep all multi-turn
  sessions, downsample single-turn to each target fraction, reclassify.

Includes the result JSONs for both.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-01 01:03:40 +08:00
9c105cf05a MB5 PD ablation: controlled-variable reuse/conc redo + campaign tooling
Reuse and concurrency axes redone with proper controlled variables, plus
the orchestration used to run them on dash0:

- run_reuse_fixed.sh: hold REAL prefill work (delta) constant, vary only
  cached prefix -> reuse = C/(C+U). Supersedes old fig1 (which held
  input=8192 and sliced prefix out, confounding "more reuse" with "less
  prefill").
- run_conc.sh: agentic-corner config (in=32768, delta=512, reuse=0.984,
  out=128) that exposes PD's structural KV-transfer tax. Supersedes old fig3.
- run_campaign{,2,3}.sh, backfill_d2048o128.sh: serial campaign drivers
  (strictly one driver at a time), out=128 sweeps, PD wall-cap for
  collapse-draining high-reuse arms, and flaked-arm backfill.
- mb5_run_gpu.sh: per-config bring-up / replay / teardown orchestrator.
- plot_pd_crossover.py: render the reuse_compare figures from fig_agg dumps.
- fig_agg.py: tolerate null stats from fully-collapsed arms (0 successes
  write the stat keys as null; `dict.get(k, {})` returns null, not {}).

Data: fig1_reuse_fixed.json, fig1_reuse_d{1024,2048}_o128.json
Figs: reuse_compare_AB.png, reuse_compare_ABC.png

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-01 01:03:27 +08:00
32f7f55990 v2: linear (default cache-aware) baseline + 2x wall-cap on first600s
Follow-up to the LMetric sweep: rerun with --policy linear (cache-aware
load + sticky session affinity, the cache_aware_proxy default) and cap
each PD-disagg arm at 2x the colo bench wall (SIGTERM bench.sh once cap
is exceeded; the cleanup trap clears vLLM and proxy; capped runs lack
metrics.summary.json so the analysis script computes from raw
metrics.jsonl).

Headline: the success-rate ceiling is policy-invariant.

  arm        linear (capped at 2x)    lmetric (uncapped)
  colo       807/807 = 100%, 964s     807/807 = 100%, 1021s
  pd6 (6:2)  472/807 =  58%, 2280s ⊗  474/807 =  59%, 3325s
  pd4 (4:4)  349/807 =  43%, 2281s ⊗  348/807 =  43%, 6850s
  pd2 (2:6)  176/807 =  22%, 2280s ⊗  180/807 =  22%, 19275s

Routing affects only how much wall is wasted timing out unreachable
requests at 600s each. Linear hits the same ceiling in 2280s as
LMetric does in 3300-19000s. This *strengthens* the §5 D-pool
capacity-ceiling thesis -- the cap is structural, not a routing
artifact.

Artifacts:
  analysis/v2/fig4r_linear.json          -- 4-arm linear summary
  analysis/v2/PD_DISAGG_LMETRIC.md       -- extended with wall-cap section
  figs/v2/fig4_linear_vs_lmetric.png     -- 3-panel side-by-side comparison
  microbench/fresh_setup/plot_fig4_linear_vs_lmetric.py
2026-06-01 00:55:40 +08:00
7529284cee v2: LMetric PD-colo vs PD-disagg on the real agentic trace
Anchor experiment for the clean-stack PD comparison using the canonical
cache-aware proxy with --policy lmetric (scripts/bench.sh harness). Two
traces x four arms = eight runs on dash1.

Headline: with the right routing baseline (LMetric), PD-colo holds 100%
completion on both traces while every static PD-disagg ratio fails
(14-65% completion), and the failure mode rotates with the split --
no static partition has a working operating point on this workload.
LMetric improves colo dramatically (TTFT p50 1.0s vs original §3 RR
7.0s; 7x) but does NOT rescue PD-disagg, confirming the bottleneck is
structural (D-pool admission + multi-turn KV accumulation), not routing.

Completion matrix:
                    first600s  full
  colo                 100%    100%
  pd6 (6:2)            58.7%   65.3%   (decode-bound)
  pd4 (4:4)            43.1%   43.9%   (both bottlenecks)
  pd2 (2:6)            22.3%   13.9%   (prefill-bound)

The original §3 RR "100% PD completion" appears to be a measurement
artifact of e13391e: producer-KV eviction acted as a relief valve,
letting more requests squeeze under the 600s timeout at the (uncosted)
price of cross-turn re-prefill. With the eviction off, PD-disagg is
worse than §3 advertised, not better.

Artifacts:
  analysis/v2/fig4l_lmetric.json     -- 8-arm summary data
  analysis/v2/PD_DISAGG_LMETRIC.md   -- writeup + reproduce recipe
  figs/v2/fig4_lmetric_pd_vs_colo.png -- 4-panel comparison figure
  microbench/fresh_setup/plot_fig4l_lmetric.py -- plot script
2026-05-31 20:15:10 +08:00
fafc44da79 MB5 PD reuse-centric ablation: tooling, data, Fig 1-3
Three-axis controlled ablation of PD-colo vs PD-disagg on synthetic regular
traces (closed-loop, controlled reuse via REPLAY_NO_REALIZED_PREFIX) on the
clean stack (e13391e gated off).

  Axis 1 (Fig 1) -- reuse 6%->94% at N=8, in8192/out256
  Axis 2 (Fig 2) -- shape in2048/out2048 -> in32768/out64 at N=8, reuse~70%
  Axis 3 (Fig 3) -- concurrency N=8/16/32/64 at reuse~71%, in8192/out256

Findings:
  * APC parity colo=PD at every reuse (5.5/22/44/66/77/82%) -- contamination
    fix validated.
  * PD edge erodes 1.57x->1.10x with reuse; prefill GPUs strand 26%->9%.
  * Shape: PD-best peaks mid-sweep (1.34x at in8192/out512); wrong PD ratio
    catastrophic at prefill extreme (in32768/out64 pd2 = 378/400, p99 432s).
  * Concurrency: PD wins N<=32 (1.23-1.29x), TIPS at N=64 -- pd2/pd4
    crater (APC 71%->1.4%, TPS -30%) while colo scales cleanly.

Infrastructure:
  * replayer: --max-inflight-sessions, --inter-turn-think, --no-realized-prefix
    (env-defaulted via REPLAY_MAX_INFLIGHT, REPLAY_INTER_TURN_THINK_S,
    REPLAY_NO_REALIZED_PREFIX).
  * mb5_run.sh: writes bench_config.json + gpu_util.csv + run_window.json +
    instance_apc.txt + metrics.jsonl for bench_report/fig_agg ingest.
  * fig_agg.py: per-arm GPU role split + producer-side APC; --json mode.
  * gpu_util_report.py: companion per-GPU util report from gpu_util.csv.
  * partial_summary.py: stats from in-flight replay_metrics.jsonl
    (works before metrics.summary.json exists).

Data: analysis/mb5_pd_ablation/fig{1,2,3}.json (24 + 20 + 16 rows).
Figures: figs/mb5_pd_ablation/fig{1_reuse,2_shape,3_concurrency}_axis.png.
2026-05-31 20:14:46 +08:00
a2111b6e18 PD-disagg docs: annotated corrections for e13391e contamination
Adds dated, non-destructive correction notes to the contaminated PD-vs-colo
artifacts after the producer-eviction bug (`evict_blocks(sent_block_ids)` on
`finished_sending`, deployed over the "fresh" pip vLLM by
`scripts/deploy_vllm_patches.sh`) was found and gated behind
`VLLM_EVICT_SENT_BLOCKS` (default off).

  PD_DISAGG_RESULTS.md  top CORRECTION banner + §6 RETRACTED marker.
                        §6 (session-affinity hot-pin) was an `e13391e`
                        artifact under controlled concurrency; §3 RR, §4
                        TPOT win, §5 D-pool ceiling, §5.1 consumer crash
                        stand.
  RESULTS_SUMMARY.md    §4 confirm+refine note: clean ablation confirms
                        the D-pool capacity thesis and adds regime-
                        dependence.
  pd_separation_analysis.md  scoped caution: thesis confirmed; flags
                        only reuse-dependent figures for cross-check
                        (this study used a different stack).
  figs/mb5/CORRECTION.md  flags mb5_producer_hotspot.png as retracted;
                        §3 RR and §5 D-pool figures stand.
2026-05-31 20:14:14 +08:00
0b180c191e v2 exp(d): expand figure to 6 panels (TTFT/E2E mean+p90, TPS, per-worker GPU util)
Per request: TTFT mean+p90, E2E mean+p90, decode TPS (output goodput; total/
prefill TPS omitted as cache-miss-inflated), and per-worker GPU-util boxplots
(8 workers/arm, tracets vs thinktime) showing utilization level + balance.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-05-30 21:10:27 +08:00
46 changed files with 2115 additions and 44 deletions

View File

@@ -97,6 +97,18 @@ dash1 GPU 0 单 instance无 kv_connectorchunked-prefill 默认开启,
- MB1 + MB2 的合计 cost-benefit phase isolation 维度上 PD-disagg 是赢的**但这件事被容量天花板压倒**。 - MB1 + MB2 的合计 cost-benefit phase isolation 维度上 PD-disagg 是赢的**但这件事被容量天花板压倒**。
- Paper §3.2 论证应该聚焦"D 池装不下"MB1/MB2 数据用作 supporting contextper-request transfer charge 量化phase isolation benefit 量化而不是 main argument - Paper §3.2 论证应该聚焦"D 池装不下"MB1/MB2 数据用作 supporting contextper-request transfer charge 量化phase isolation benefit 量化而不是 main argument
> ✅ **2026-05-30 更新 — 干净栈三轴 ablation 证实本节、并加 regime 细化。**
> 本节的容量论点D 池容量天花板 / decode 减半)在修复 `e13391e` 污染后的 clean stack
> 上**得到确认**concurrency 轴 N=64 时 PD 倾覆,**producer APC 从 71% 崩到 1.4%、TPS 30%**
> 而 colo 线性 scaleFig 3。**细化**PD 并非"在 agentic 上一律失败"——它在
> *低负载 / decode-heavy / 低复用* 区间**赢** colo真正失败的是 agentic corner高复用 +
> 短输出 + 大上下文 + 高并发)——静态 P:D split 无法同时给出复用所需的 producer 容量
> *和* decode 容量,而 colo 的弹性池两者兼得。
> **另注**:旧 MB5 文档(`PD_DISAGG_RESULTS.md` §6"session-affinity 救不了 PD / PD 复用=0%"
> 的结论是 `e13391e`producer 每次 KV 传输后 evict prefix的**污染产物,已撤回**
> 干净栈上 session 路由的 producer APC 与 colo 持平7182%)。
> 图:[`figs/mb5_pd_ablation/`](figs/mb5_pd_ablation/)。
## 5. EAR 设计的实证状态§4 ## 5. EAR 设计的实证状态§4
| Pillar | 已实证 | 待实证 | | Pillar | 已实证 | 待实证 |

View File

@@ -0,0 +1,107 @@
# PD-disagg vs colocation — controlled reuse & concurrency axes (v2)
Self-contained results for the **controlled-variable** redo of the MB5 PD-vs-colo
ablation. Supersedes the confounded first cut (held input fixed and sliced the
prefix, so "more reuse" was entangled with "less prefill"). All arms route through
the proxy at fair **APC parity** (session-routed producers reach the same prefix-cache
hit rate as colo), so PD loses on *structure*, not on broken cache.
- **Config arms:** `colo` = 8×kv_both (8C-proxy, session-affinity); PD = `6P+2D / 4P+4D / 2P+6D`.
- **Driver:** closed-loop N (`REPLAY_MAX_INFLIGHT`) + fixed think-time; `gen_synthetic_trace.py --mode regular`.
- **PD-arm wall-cap:** collapsed PD arms drain pathologically slowly, so PD arms run with a
wall-deadline (`REPLAY_MAX_DURATION`; un-run turns counted as failures → honest completion%);
**colo is uncapped** so the reference is always fully measured.
- **Hardware:** run on **dash2** (8×H20). dash0's RDMA NICs were faulted for Mooncake during
this work (could not init the transfer engine; needs an admin reset — no sudo); dash2's NICs
are healthy. cpfs/venv/data are shared across the boxes.
---
## 1. Reuse / APC axis — fixed real prefill, vary cached prefix
N=8. Hold the **real new-prefill work per turn constant** (`--delta-len`) and grow the cached
prefix → reuse = prefix/(prefix+delta). Three shapes isolate output vs delta:
| | delta (real prefill/turn) | output | role |
|---|---|---|---|
| **A** | 2048 | 256 | original |
| **C** | 2048 | 128 | A vs C = pure **output** 256→128 |
| **B** | 1024 | 128 | C vs B = pure **delta** 2048→1024 |
**PD-best advantage** = colo E2E-p90 / best-PD E2E-p90 (>1 ⇒ PD wins):
| reuse% | A d2048/o256 | C d2048/o128 | B d1024/o128 |
|---|---|---|---|
| 20 | 1.34 | 1.41 | — |
| 50 | 1.36 | 1.37 | — |
| 67 | **1.47** | **1.49** | **1.27** |
| 80 | 1.31 | 1.23 | 1.25 |
| 90 | 1.15 | 1.01 | — |
| 95 | 0.87 | 0.89 | 0.89 |
![reuse 3-way](../../figs/mb5_pd_ablation/reuse_compare_ABC.png)
**Findings:**
1. **Output length is ~negligible.** A and C (same delta) track each other across the whole
range — halving output barely moves PD's advantage.
2. **Delta (real prefill/turn) is the dominant shape factor.** B (delta=1024) sits clearly
below A/C at mid reuse (67%: 1.27 vs ~1.48). More real prefill per turn → bigger PD win,
because PD-disagg's benefit is isolating prefill from decode — more prefill to isolate.
3. **Crossover to colo at reuse ~9095% is robust** across all three shapes: PD always loses
the high-reuse / large-resident-context corner (it must KV-transfer the whole resident
context every turn for a few hundred new tokens; colo keeps it local).
*Caveat:* the clean, uncapped, 100%-completion comparison region is reuse **2080%** (carries
findings 12). At reuse 90/95% the PD arms collapse and C's points are capped-completion, while
A/B are full-drain — comparable in direction, not in exact PD completion%.
Data: `fig1_reuse_fixed.json` (A), `fig1_reuse_d2048_o128.json` (C), `fig1_reuse_d1024_o128.json` (B).
---
## 2. Concurrency axis — agentic corner, sweep N
in=32768 (prefix 32256 + delta 512, **reuse 0.984**), out=128; closed-loop N ∈ {8,16,32,48,64,96,128};
PD arms capped 600s, colo uncapped.
| N | **colo** completion · E2E-mean · TPS | best PD-arm completion |
|---|---|---|
| 8 | **256/256** · 2.4s · 326 | 6P+2D 256/256 |
| 16 | **512/512** · 3.5s · 462 | 6P+2D 439/512 (86%) |
| 32 | **1024/1024** · 13.3s · 190 | all PD **<27%** |
| 48 | **1536/1536** · 24.9s · 168 | all PD <32% |
| 64 | **2048/2048** · 38.4s · 166 | all PD <31% |
| 96 | **3072/3072** · 60.0s · 171 | PD **27%** |
| 128 | **4096/4096** · 80.8s · 181 | 4P+4D 6%, 2P+6D <1% |
![concurrency](../../figs/mb5_pd_ablation/fig3_concurrency_axis.png)
**Finding:** **colo completes 100% of requests at every concurrency level** it degrades
*gracefully* (latency rises 2.4s81s, nothing dropped). **Every static PD split collapses, and
progressively earlier as N rises**: PD is viable only at N816; by N32 it drops 7099% of
requests while its prefix-cache hit-rate craters to ~0%. colo's elastic pool absorbs the
time-varying P/D demand; the static partition + per-turn 32k KV-transfer cannot. (Latency
percentiles count successes only, so they *understate* PD read them with the completion column.)
Data: `fig3_conc32k.json`.
*Caveat:* N=128 6P+2D is missing (one transient vLLM/Mooncake startup flake at the end); does
not change the picture (all PD arms are already collapsed by N=128). The SLO auto-stop in the
driver is a no-op (a stdout-capture bug), so the full grid ran more points, not fewer.
---
## 3. Reproduce
```bash
# on a box with healthy Mooncake/RDMA NICs (dash2), cpfs mounted:
R=/home/admin/cpfs/wjh/agentic-kv-fresh
# reuse axis (three shapes): DELTA/OL pick the shape; tag carries _o${OL}
ssh dash2 "cd $R && DELTA=2048 OL=256 bash microbench/fresh_setup/run_reuse_fixed.sh"
ssh dash2 "cd $R && DELTA=2048 OL=128 bash microbench/fresh_setup/run_reuse_fixed.sh"
ssh dash2 "cd $R && DELTA=1024 OL=128 bash microbench/fresh_setup/run_reuse_fixed.sh"
# concurrency axis (capped):
ssh dash2 "cd $R && NLIST='8 16 32 48 64 96 128' CONC_PD_MAXDUR=600 bash microbench/fresh_setup/run_conc.sh"
# render (reads the *.json in this dir):
python microbench/fresh_setup/plot_pd_crossover.py
```

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@@ -23,6 +23,22 @@ Per-request breakdown shows **87.7% of TTFT** is spent waiting for KV cache memo
> Earlier cross-machine comparison (commit `1e86285`) was invalidated — baseline used warm instances. See REPORT.md §3.5. > Earlier cross-machine comparison (commit `1e86285`) was invalidated — baseline used warm instances. See REPORT.md §3.5.
| **Delta** | **-45%** | **-36%** | **-44%** | **+30pp** | | **Delta** | **-45%** | **-36%** | **-44%** | **+30pp** |
> ✅⚠️ **2026-05-30 — confirmed + refined by the clean MB5 ablation; one caveat.**
> A producer-side contamination (`e13391e`: evicts a producer's prefix-cache on every
> KV transfer) was found in the *agentic-kv-fresh* MB5 stack and gated off; everything
> was re-run clean.
> - **Confirmed:** this doc's central thesis — PD's failure is a **decode-side KV memory
> wall**, not transfer/prefill cost — is reproduced on the clean stack (concurrency
> axis: at N=64 the split collapses, APC 71%→1.4%, TPS 30%; colo scales). Fig 3.
> - **Refined:** "PD separation is net negative" is **regime-dependent**, not universal.
> Clean ablation shows PD *wins* at low load / decode-heavy / low-reuse and loses the
> **agentic corner** (high reuse + short output + large context + high concurrency).
> - **Caveat (cross-check):** if this study's runs used the fork vLLM that contains
> `e13391e`, any **producer prefix-cache / APC reuse** figures here (e.g. §5.3.1) may be
> understated. The decode-memory-wall result is *not* reuse-dependent and is unaffected.
> On the clean stack, session-routed producers reach APC parity with colo (7182%).
> Figures: [`figs/mb5_pd_ablation/`](../figs/mb5_pd_ablation/).
--- ---
## 1. Workload Characterization ## 1. Workload Characterization

View File

@@ -0,0 +1,165 @@
# PD-colo vs PD-disagg on the real agentic trace — LMetric (cache-aware) clean-stack anchor
**Figure:** [`figs/v2/fig4_lmetric_pd_vs_colo.png`](../../figs/v2/fig4_lmetric_pd_vs_colo.png)
**Data:** [`analysis/v2/fig4l_lmetric.json`](fig4l_lmetric.json)
**Date:** 2026-05-31 · Hardware: dash1, 8×H20 · Model: Qwen3-Coder-30B-A3B-Instruct
· vLLM 0.18.1 (V1, chunked-prefill on, `e13391e` eviction gated **off**)
· Mooncake 0.3.11 · Routing: cache-aware proxy with **`--policy lmetric`**
· Replayer per-request timeout 600 s.
## TL;DR
On the production agentic trace with the *right* routing baseline (LMetric, cache-aware),
**PD-colo (8× kv_both) keeps 100 % completion on both traces** and matches the daily-bench
expectation (~17 min for the high-load first600s, ~50 min for the full trace, with E2E p50
~3 s and TTFT p50 ~1 s — **3.57× better than the original §3 round-robin baseline at the
same wall-clock**). Every static **PD-disagg ratio fails** (1465 % completion), and the
failure mode rotates predictably with the split — **no static partition has a working
operating point on this workload**. LMetric improves colo dramatically; it does *not*
rescue PD-disagg, confirming the bottleneck is structural (D-pool admission capacity +
multi-turn KV accumulation), not routing. A follow-up linear-policy run with PD-disagg
**wall-capped at 2× the colo wall** (see end of doc) hits the **identical** success-rate
ceiling — confirming the cap is structural, not policy-driven.
## Setup
- Trace: `w600_r0.0015_st30.jsonl` (1214 reqs, 274 sessions, agentic multi-turn,
contexts up to ~112 k tokens; "first600s" variant = same heavy sessions compressed
into 600 s → 807 reqs at 3.2× higher arrival rate).
- 8 instances on 8 GPUs.
- `--mode baseline` for colo (plain vLLM); `--mode pdsep --pd-ratio P:D` for the three PD
splits, all with Mooncake KV transfer.
- Cache-aware proxy with LMetric scoring (`P_tokens × num_requests`) + session affinity
for multi-turn (the colleague's canonical baseline).
## Results
### first600s (1.35 req/s, high-load stress)
| arm | success | E2E mean / p50 / p90 / p99 | TTFT p90 | TPOT p99 | TPS | wall |
|---|---|---|---|---|---|---|
| **colo (8C)** | **807/807 = 100 %** | 11.1 / 3.27 / 28.6 / 95.9 s | 14.5 s | 388 ms | 226 | 17.0 min |
| pd6 (6:2) | 474/807 = **58.7 %** | 83.2 / 6.75 / 382 / 542 s | 380 s | 19 ms | 40 | 55 min |
| pd4 (4:4) | 348/807 = **43.1 %** | 203 / 215 / 477 / 575 s | 475 s | 25 ms | 15 | 114 min |
| pd2 (2:6) | 180/807 = **22.3 %** | 380 / 536 / 579 / 602 s | 577 s | 18 ms | 34 | 321 min* |
### Full trace (0.42 req/s, original §3 anchor load)
| arm | success | E2E mean / p50 / p90 / p99 | TTFT p90 | TPOT p99 | TPS | wall |
|---|---|---|---|---|---|---|
| **colo (8C)** | **1214/1214 = 100 %** | 10.9 / 3.13 / 29.6 / 93.7 s | 16.9 s | 254 ms | 125 | 49.9 min |
| pd6 (6:2) | 793/1214 = **65.3 %** | 61.9 / 3.70 / 307 / 477 s | 300 s | 18 ms | 46 | 94 min |
| pd4 (4:4) | 533/1214 = **43.9 %** | 131 / 8.22 / 468 / 531 s | 467 s | 21 ms | 13 | 231 min |
| pd2 (2:6) | 169/1214 = **13.9 %** | 195 / 6.82 / 552 / 593 s | 549 s | 13 ms | 1 | 563 min |
\* The pd2 wall-clock is dominated by per-request timeouts (`request_timeout=600 s`)
draining concurrently behind the multi-turn session causality.
## Five clean findings
1. **LMetric+colo is the right baseline.** Full-trace colo wall **49.9 min ≈ the original
§3 RR's 49.9 min**, but E2E p50 **3.13 s vs §3's 10.8 s (3.5×)** and TTFT p50
**1.02 s vs §3's 7.0 s (7×)**. Same throughput envelope, far better latency — by virtue
of cache-aware routing concentrating each session's turns onto one instance for
prefix-cache reuse. The original §3 RR was an *unfairly weak* colo baseline.
2. **Every static PD-disagg ratio fails on the agentic workload.** Completion drops to
1465 %, on *both* traces. The drop is not a high-load artifact (it holds at the
original §3 arrival rate of 0.42 req/s); it is structural.
3. **Failure mode rotates predictably with the P:D split:**
- **pd2 (2 producers)** → prefill-bound → 7886 % TTFT timeouts.
- **pd6 (2 decode)** → decode-admission-bound → 3541 % TTFT timeouts.
- **pd4 (4P+4D)** → both bottlenecks hit → 57 % TTFT timeouts.
- **No static ratio works.** Colo's elastic 8-GPU pool absorbs whichever phase is
hot at the moment.
4. **Decode isolation works, but doesn't matter under failure.** TPOT p99 on every PD
arm is **1325 ms** — an order of magnitude better than colo's 254388 ms — but the
win applies only to the 1465 % of requests that get admitted. The other 3586 %
time out before ever seeing a first token, so the TPOT win is invisible to the end user.
5. **The §3 RR "100 % PD completion" was a measurement artifact.** Original §3
(contaminated stack, RR routing) reported 100 % completion for pd6/pd4. LMetric on
the clean stack shows 4465 %. Most plausible mechanism: `e13391e`'s eviction of
producer KV on every transfer acted as a **relief valve**, reducing producer-pool
pressure and letting more requests squeeze under the 600 s timeout — at the (uncosted)
price of cross-turn re-prefill. With the eviction off, producers retain prefix
correctly → cache works on PD too → but the cache itself contends for producer
pool capacity, and the decode-pool admission ceiling tips earlier. **PD-disagg is
worse on agentic than §3 advertised, not better.**
## Linear-policy + wall-cap follow-up (2026-06-01) — the success ceiling is policy-invariant
To check whether the LMetric routing was secretly handicapping PD-disagg, we re-ran
first600s with the **default `--policy linear`** (cache-aware load score + sticky
session affinity — the original baseline of the cache_aware_proxy stack) and
**wall-capped each PD-disagg arm at 2 × the colo wall** (kill bench.sh + cleanup
GPUs once cap is exceeded, record `records_at_cap`).
| arm | linear success | linear wall | linear @-cap? | lmetric success | lmetric wall |
|---|---|---|---|---|---|
| **colo** | 807/807 = **100 %** | 964 s | — | 807/807 = **100 %** | 1021 s |
| **pd6 (6:2)** | **472/807 = 58 %** | 2280 s | ⊗ cap (706 dispatched) | 474/807 = 59 % | 3325 s |
| **pd4 (4:4)** | **349/807 = 43 %** | 2281 s | ⊗ cap (577 dispatched) | 348/807 = 43 % | 6850 s |
| **pd2 (2:6)** | **176/807 = 22 %** | 2280 s | ⊗ cap (521 dispatched) | 180/807 = 22 % | 19275 s |
→ Figure: [`figs/v2/fig4_linear_vs_lmetric.png`](../../figs/v2/fig4_linear_vs_lmetric.png) ·
data: [`fig4r_linear.json`](fig4r_linear.json)
**Three clean conclusions from the wall-cap experiment:**
1. **The success-rate ceiling is structural, not a routing artifact.** Linear and
LMetric — two very different scoring policies (one session-sticky cache-aware,
the other non-sticky pure load) — converge on **identical success rates**
(58 / 43 / 22 %) for every PD-disagg ratio. Routing has *zero* effect on the
completion ceiling. The bottleneck is the static P:D split itself.
2. **LMetric's longer wall was wall *wasted on requests that will never succeed*.**
When the cap is enforced, linear hits the same ceiling in 2280 s as LMetric did
in 330019000 s — the extra wall just slowly times out the unreachable
requests at 600 s each.
3. **The wall-cap is the right way to bench PD-disagg.** Reporting "completion %"
without a wall budget is misleading (the bench eventually completes if you wait
forever, but only by counting timeouts as failures over hours). The honest
metric is **success-in-2×-colo-wall**, which gives the same answer for both
routings and matches what an end user would observe on a real SLO.
This **strengthens** the §5 D-pool capacity-ceiling thesis: even with
session-affinity routing serving every request to a warm prefix cache (which
*should* maximise PD's throughput), the static D-pool can't admit more than
~22 / 43 / 58 % of the agentic trace within 2× the colo budget. Colo wins not
because routing is smarter, but because its **elastic pool** absorbs whichever
phase is hot — there's no cap to hit.
---
## Reproduce
```bash
# On dash1, from the main repo /home/admin/cpfs/wjh/agentic-kv:
for TR in w600_r0.0015_st30.jsonl w600_r0.0015_st30_first600s.jsonl; do
TRACE=traces/$TR bash scripts/bench.sh --tag fig4l_lmetric_colo_${TR%.*} \
--mode baseline --policy lmetric
for r in 6:2 4:4 2:6; do
TRACE=traces/$TR bash scripts/bench.sh --tag fig4l_lmetric_${r/:/p}_${TR%.*} \
--mode pdsep --pd-ratio $r --policy lmetric
done
done
python microbench/fresh_setup/plot_fig4l_lmetric.py
python microbench/fresh_setup/plot_fig4_linear_vs_lmetric.py
```
For the linear + 2× wall-cap variant, run colo first to get `wall_clock_s`,
compute `CAP=2*wall`, then launch each PD-disagg arm in the background and
`SIGTERM` it (so bench.sh's cleanup trap fires) once `date +%s` minus the
arm's start time exceeds `CAP`. The capped runs lack `metrics.summary.json`
(replayer was killed before it could write); compute the summary directly from
`metrics.jsonl` (see the inline collector used to build
`analysis/v2/fig4r_linear.json`).
Source `bench.sh` cleans GPUs before each arm and writes `metrics.jsonl` +
`metrics.summary.json` per tag. Aggregation script: see the inline JSON dump used
to build `analysis/v2/fig4l_lmetric.json`.

View File

@@ -0,0 +1 @@
[{"tag": "fig4l_lmetric_colo_first600s", "arm": "colo", "trace": "first600s", "n": 807, "req": 807, "e2e": {"count": 807.0, "mean": 11.066699584425269, "p50": 3.27055042097345, "p90": 28.745733462180937, "p99": 97.40008939541167}, "ttft": {"count": 807.0, "mean": 5.119651803458883, "p50": 1.2114678020589054, "p90": 14.777630288852365, "p99": 50.68302261995841}, "tpot": {"count": 807.0, "mean": 0.03004899278845205, "p50": 0.009643197803618922, "p90": 0.042092699501536976, "p99": 0.3919741264067197}, "wall": 1020.5351374909515, "tps": 226.12940164644368}, {"tag": "fig4l_lmetric_colo_full", "arm": "colo", "trace": "full", "n": 1214, "req": 1214, "e2e": {"count": 1214.0, "mean": 10.928977524270508, "p50": 3.1279119075043127, "p90": 30.011970606888667, "p99": 94.77313101590481}, "ttft": {"count": 1214.0, "mean": 5.533819193267678, "p50": 1.017395684029907, "p90": 17.36427243486981, "p99": 51.49416554694993}, "tpot": {"count": 1214.0, "mean": 0.02049970290344434, "p50": 0.009544484575988867, "p90": 0.032480608771520716, "p99": 0.26057810739537074}, "wall": 2993.276069591986, "tps": 125.38402448497122}, {"tag": "fig4l_lmetric_pd2_first600s", "arm": "2P+6D", "trace": "first600s", "n": 180, "req": 807, "e2e": {"count": 180.0, "mean": 380.2505690135715, "p50": 535.6594606440049, "p90": 579.5011055286858, "p99": 601.5567972306756}, "ttft": {"count": 180.0, "mean": 378.7133691522933, "p50": 534.4269686369807, "p90": 577.3534130641376, "p99": 596.404559875431}, "tpot": {"count": 180.0, "mean": 0.007975266077679418, "p50": 0.007166497974743372, "p90": 0.012511071875514153, "p99": 0.017508981961061446}, "wall": 19275.367093455978, "tps": 1.8895100582735462}, {"tag": "fig4l_lmetric_pd2_full", "arm": "2P+6D", "trace": "full", "n": 169, "req": 1214, "e2e": {"count": 169.0, "mean": 194.88523891245458, "p50": 6.817620265996084, "p90": 552.1569225640735, "p99": 595.3934216396092}, "ttft": {"count": 169.0, "mean": 193.4153314989016, "p50": 5.60239192598965, "p90": 549.3611521873856, "p99": 582.4436428000824}, "tpot": {"count": 169.0, "mean": 0.007747395842651413, "p50": 0.007691574401794991, "p90": 0.011201243427351017, "p99": 0.013311375577245894}, "wall": 33770.57413210906, "tps": 0.9869539045920406}, {"tag": "fig4l_lmetric_pd4_first600s", "arm": "4P+4D", "trace": "first600s", "n": 348, "req": 807, "e2e": {"count": 348.0, "mean": 202.63302869595395, "p50": 214.03008900902933, "p90": 477.40967412578175, "p99": 576.6393926549597}, "ttft": {"count": 348.0, "mean": 199.96385804087797, "p50": 213.50966987549327, "p90": 475.7766476540827, "p99": 559.6153268160638}, "tpot": {"count": 348.0, "mean": 0.008873619369764751, "p50": 0.007645836479973812, "p90": 0.013845969236959285, "p99": 0.02567216653158788}, "wall": 6850.181333696004, "tps": 15.00296050477674}, {"tag": "fig4l_lmetric_pd4_full", "arm": "4P+4D", "trace": "full", "n": 533, "req": 1214, "e2e": {"count": 533.0, "mean": 130.94711188977982, "p50": 8.219856544979848, "p90": 473.44134307731883, "p99": 533.2597587251009}, "ttft": {"count": 533.0, "mean": 127.83193208824007, "p50": 4.8246813879814, "p90": 467.54664219671395, "p99": 528.8304683346115}, "tpot": {"count": 533.0, "mean": 0.008886429490232585, "p50": 0.007981476340708988, "p90": 0.013570741891233497, "p99": 0.023050950961825044}, "wall": 13884.384965199977, "tps": 12.621372890425038}, {"tag": "fig4l_lmetric_pd6_first600s", "arm": "6P+2D", "trace": "first600s", "n": 474, "req": 807, "e2e": {"count": 474.0, "mean": 83.15809065495806, "p50": 6.7270191764691845, "p90": 391.6558471220078, "p99": 544.7372293809171}, "ttft": {"count": 474.0, "mean": 80.70155321074382, "p50": 4.1273433425230905, "p90": 390.00296151017517, "p99": 539.0574236416071}, "tpot": {"count": 474.0, "mean": 0.008519881756330928, "p50": 0.00803907146806204, "p90": 0.012583933303093976, "p99": 0.018606097790947705}, "wall": 3325.2749515309697, "tps": 39.705588838364164}, {"tag": "fig4l_lmetric_pd6_full", "arm": "6P+2D", "trace": "full", "n": 793, "req": 1214, "e2e": {"count": 793.0, "mean": 61.907526705667, "p50": 3.69814173609484, "p90": 308.2633092067672, "p99": 477.48038318102715}, "ttft": {"count": 793.0, "mean": 59.25069201986225, "p50": 1.402295546955429, "p90": 302.5604081378088, "p99": 475.3738951798529}, "tpot": {"count": 793.0, "mean": 0.009137289999448822, "p50": 0.008635683270933276, "p90": 0.013065757584108427, "p99": 0.01816783740464599}, "wall": 5662.029295974993, "tps": 39.24494000021532}]

View File

@@ -0,0 +1 @@
[{"tag": "fig4r_linear_colo_first600s", "arm": "colo", "trace": "first600s", "policy": "linear", "n": 807, "req": 807, "dispatched": 807, "e2e": {"count": 807.0, "mean": 8.436370009274967, "p50": 2.5224755640374497, "p90": 22.65510415879542, "p99": 75.54369598095519}, "ttft": {"count": 807.0, "mean": 4.2332503390957195, "p50": 0.8872958200518042, "p90": 11.684667797433207, "p99": 44.98891795879462}, "tpot": {"count": 807.0, "mean": 0.020958194728517718, "p50": 0.00851320761584622, "p90": 0.026440129078245465, "p99": 0.30344440533287176}, "wall": 963.6191155100241, "tps": 239.4857016486815, "capped": false}, {"tag": "fig4r_linear_pd2_first600s", "arm": "2P+6D", "trace": "first600s", "policy": "linear", "n": 176, "req": 807, "dispatched": 521, "e2e": {"count": 176, "mean": 378.5561210460834, "p50": 536.7719694490079, "p90": 583.832092280034, "p99": 601.3415494390065}, "ttft": {"count": 176, "mean": 377.12570991374446, "p50": 536.1157373189926, "p90": 580.3465002350276, "p99": 598.0943597999867}, "tpot": {"count": 176, "mean": 0.007864906140929698, "p50": 0.007212154543958604, "p90": 0.011962352272927423, "p99": 0.017870794738764347}, "wall": 2280, "tps": 14.419736842105262, "capped": true}, {"tag": "fig4r_linear_pd4_first600s", "arm": "4P+4D", "trace": "first600s", "policy": "linear", "n": 349, "req": 807, "dispatched": 577, "e2e": {"count": 349, "mean": 264.8537863784421, "p50": 306.6853819829412, "p90": 488.64622142596636, "p99": 596.5830293919425}, "ttft": {"count": 349, "mean": 262.3163347712099, "p50": 299.75751709297765, "p90": 485.475125996978, "p99": 596.4081599479541}, "tpot": {"count": 349, "mean": 0.010442244895290958, "p50": 0.008213572105774598, "p90": 0.019443845545703716, "p99": 0.028178529054794}, "wall": 2281, "tps": 38.306882946076286, "capped": true}, {"tag": "fig4r_linear_pd6_first600s", "arm": "6P+2D", "trace": "first600s", "policy": "linear", "n": 472, "req": 807, "dispatched": 706, "e2e": {"count": 472, "mean": 118.632779156234, "p50": 12.702161715948023, "p90": 458.1609142010566, "p99": 526.5488834320568}, "ttft": {"count": 472, "mean": 115.80202843308507, "p50": 9.745031949947588, "p90": 455.81679951993283, "p99": 516.5850186559837}, "tpot": {"count": 472, "mean": 0.00950947083585719, "p50": 0.008435572332624966, "p90": 0.015233499645638644, "p99": 0.023447183093280886}, "wall": 2280, "tps": 61.69210526315789, "capped": true}]

19
figs/mb5/CORRECTION.md Normal file
View File

@@ -0,0 +1,19 @@
# ⚠️ Correction notice for figs/mb5/ (2026-05-30)
These figures back `microbench/fresh_setup/PD_DISAGG_RESULTS.md`. A producer-side
contamination was later found in the stack that produced them: commit **`e13391e`**
(deployed over the "fresh" pip vLLM by `scripts/deploy_vllm_patches.sh`) evicts a
producer's prefix-cache blocks on every KV transfer, so a disaggregated producer
could never keep a session's prefix warm. It is now gated behind
`VLLM_EVICT_SENT_BLOCKS` (default off) and everything was re-run clean.
| figure | section | status |
|---|---|---|
| `mb5_producer_hotspot.png` | §6.3 session-affinity hot-pinning | 🛑 **RETRACTED** — pure `e13391e` artifact. On the clean stack, session-routed producers reach APC parity with colo (7182%); there is no 0%-util stall / hot-pin pathology. |
| `mb5_latency_compare.png` | §3 round-robin headline | ✅ stands — RR's ~0% prefix-hit is a *routing* artifact (consecutive turns → different producers), not the eviction bug; reproduced clean. |
| `mb5_kv_timeline.png`, `mb5_role_split.png`, `mb5_peak_utilization.png` | §5 per-role KV pool occupancy | ✅ D-pool capacity-ceiling mechanism stands (decode pegs while prefill strands). P-pool occupancy may read slightly low under eviction; the qualitative split is unaffected. |
| `mb5_summary.csv` | aggregate | mixed — §3/§5 rows valid; any session-affinity rows superseded. |
**Superseded by the corrected three-axis ablation:** [`../mb5_pd_ablation/`](../mb5_pd_ablation/)
(reuse / shape / concurrency), data in [`../../analysis/mb5_pd_ablation/`](../../analysis/mb5_pd_ablation/).
Raw §6 data `analysis/mb5/session_prod.json` is contaminated; `analysis/mb5/rr_prod.json` (round-robin) stands.

Binary file not shown.

After

Width:  |  Height:  |  Size: 122 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 146 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 171 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 114 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 99 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 109 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 149 KiB

View File

@@ -10,6 +10,51 @@ Date: 2026-05-28 · Hardware: dash1, 8×GPU · Model: Qwen3-Coder-30B-A3B-Instru
--- ---
## ⚠️ CORRECTION (2026-05-30) — read before §6
A contamination was found in the "fresh" vLLM used for the runs below.
`scripts/deploy_vllm_patches.sh` had copied our fork commit **`e13391e`** over the
pip-installed release; that commit calls `evict_blocks(sent_block_ids)` on
`finished_sending`, i.e. it **evicts a producer's prefix-cache blocks on every KV
transfer**. So a disaggregated producer could never keep a session's prefix warm,
*regardless of routing*. We have since gated that behind `VLLM_EVICT_SENT_BLOCKS`
(default off) and re-run everything on the corrected stack.
**Retracted (was a pure artifact of `e13391e`):**
- **All of §6** ("smarter routing does not save PD" / "session-affinity is
*strictly worse*" / "GPUs at ~0%" / "producer hot-pinning" / "producer prefix-hit
~0.2%"). On the corrected stack, **session-affinity recovers producer reuse to
full parity with colo (APC 7182%)** — the collapse was the eviction bug starving
the very cache session-affinity exists to fill, not a routing pathology.
- The framing that PD reuse is "0% / fundamentally broken." PD reuses prefix
*exactly as well as colo* once routing is session-sticky.
**Still stands (independent of `e13391e`):**
- **§3 round-robin** numbers — RR sends consecutive turns to *different* producers,
so its ~0% prefix-hit is a **routing** artifact (not the eviction bug) and is
reproduced on the clean stack; RR PD still loses to 8C.
- **§4** PD wins TPOT (decode isolation) — robust.
- **§5.1** the consumer counter-underflow crash — a real, separate vLLM 0.18.1 bug.
- **§5** the D-pool capacity-ceiling mechanism (decode side pegs while prefill
strands) — real.
**Corrected verdict (the real reason PD loses on agentic).** It is *not* "routing
can't help." On the clean stack PD is **regime-dependent**: it *wins* at low
load / decode-heavy / low-reuse, and loses the **agentic corner** (high reuse +
short output + large context + high concurrency) through a structural crossover —
its static P:D split cannot simultaneously provide the prefix-cache capacity
(needs many producers) *and* the decode capacity (needs many decoders) that
agentic demands at once, while colo's elastic pool provides both. See the
three-axis ablation: **reuse** erodes the edge (1.57×→1.10×), **shape** rotates the
best ratio and is catastrophic at the prefill extreme, and **concurrency** tips PD
at N=64 (APC craters 71%→1.4%, TPS 30%) while colo scales cleanly.
→ Figures: [`figs/mb5_pd_ablation/`](../../figs/mb5_pd_ablation/) ·
data: [`analysis/mb5_pd_ablation/`](../../analysis/mb5_pd_ablation/) ·
the clean re-run of *this exact* w600 experiment (ratio-swept) is the Fig 4 anchor.
---
## TL;DR (verdict) ## TL;DR (verdict)
**No static prefill/decode split beats 8-way colocation (8C) on this agentic **No static prefill/decode split beats 8-way colocation (8C) on this agentic
@@ -205,6 +250,15 @@ single failed request, which is required to compare routing arms fairly in §6.
## 6. The routing handicap — and whether smarter routing rescues PD ## 6. The routing handicap — and whether smarter routing rescues PD
> 🛑 **RETRACTED (2026-05-30) — this entire section is an artifact of `e13391e`.**
> The session-affinity runs below were starved by the producer-eviction bug, so
> they could never collect prefix-cache reuse. On the corrected stack
> session-affinity reaches **APC parity with colo (7182%)** and does *not* stall
> at 0% GPU util. The real mechanism is the capacity/concurrency crossover, not a
> routing pathology — see the CORRECTION banner at the top and
> [`figs/mb5_pd_ablation/`](../../figs/mb5_pd_ablation/). Text kept below for the
> record only.
Every PD config above shows **prefix-cache hit = 0%**, versus 8C's 19%. That Every PD config above shows **prefix-cache hit = 0%**, versus 8C's 19%. That
is not fundamental to disaggregation — it is the stock proxy round-robining is not fundamental to disaggregation — it is the stock proxy round-robining
the **prefill** side: consecutive turns of one agentic session land on the **prefill** side: consecutive turns of one agentic session land on

View File

@@ -0,0 +1,35 @@
#!/usr/bin/env bash
# Backfill the d2048/o128 reuse arms that vLLM startup-flaked out (transient
# "Engine core initialization failed", intermittent). Retry up to 4x each with a
# clean teardown between attempts; HEALTH_MAX_TRIES=180 so a crashed launch fails
# in ~6min (not 10) before retrying. Then re-aggregate the figure JSON.
cd /home/admin/cpfs/wjh/agentic-kv-fresh
export MB5_VENV=$PWD/.venv_dash0
export HEALTH_MAX_TRIES=180
VPY=$MB5_VENV/bin/python
DELTA=2048; OL=128; N=8; THINK=0.5; TURNS=8; NSESS=48
MISS="${MISS:-4096:6P+2D 18432:6P+2D 38912:8C-proxy 38912:6P+2D}"
echo "=== BACKFILL START $(date) miss='$MISS' ==="
for pc in $MISS; do
pfx=${pc%%:*}; cfg=${pc##*:}
tag="reuse_p${pfx}_d${DELTA}_o${OL}"; trace="traces_synth/${tag}.jsonl"
$VPY scripts/gen_synthetic_trace.py --out "$trace" --mode regular --qps "$NSESS" --duration-s 1 \
--turns "$TURNS" --prefix-len "$pfx" --delta-len "$DELTA" --output-len "$OL" --seed 42 >/dev/null 2>&1
dur=""; [ "$cfg" != "8C-proxy" ] && dur=500
ok=0
for attempt in 1 2 3 4; do
echo "[backfill] $tag $cfg attempt=$attempt $(date +%T)"
MB5_P_ROUTING=session MB5_COLO_ROUTING=session \
REPLAY_MAX_INFLIGHT=$N REPLAY_INTER_TURN_THINK_S=$THINK REPLAY_NO_REALIZED_PREFIX=1 REPLAY_MAX_DURATION="$dur" \
CONFIGS="$cfg" REPS=1 TRACE="$trace" RUN_TAG="$tag" \
bash scripts/mb5_run_gpu.sh >/dev/null 2>&1
if [ -f "mb5_runs/${tag}_${cfg}_rep1/replay_metrics.summary.json" ]; then
echo " OK $cfg pfx=$pfx attempt=$attempt"; ok=1; break; fi
echo " FAILED attempt=$attempt; cleanup+retry"
MB5_VENV=$PWD/.venv_dash0 bash scripts/mb5_launch.sh stop >/dev/null 2>&1; sleep 5
done
[ $ok = 0 ] && echo "[backfill] GAVE UP $tag $cfg"
done
dirs=(); for d in mb5_runs/reuse_*_d2048_o128_*_rep1; do [ -f "$d/replay_metrics.summary.json" ] && dirs+=("$d"); done
$VPY scripts/fig_agg.py --json "${dirs[@]}" > analysis/mb5_pd_ablation/fig1_reuse_d2048_o128.json
echo "=== BACKFILL DONE dirs=${#dirs[@]}/24 $(date) ==="

View File

@@ -0,0 +1,142 @@
"""Aggregate a set of MB5 run dirs into one comparison table.
Pulls the three core metrics the analysis cares about, per run:
- E2E latency (from replay_metrics.summary.json: latency_stats_s)
- TPS (output tokens / wall_clock_s)
- GPU util by workers (gpu_util.csv over run_window, split prefill/decode by role)
plus honest reuse (producer-side APC from instance_apc.txt) and TTFT/TPOT for logs.
Arm + GPU role split + producer APC ports are inferred from the dir name:
*_colo_* -> 8 kv_both ; apc ports 8000-8007 (all keep prefix)
*_pd6_* -> 6P+2D P0-5/D6-7 ; apc 8000-8005
*_pd_* -> 4P+4D P0-3/D4-7 ; apc 8000-8003 (note: "pd" not "pd4")
*_pd2_* -> 2P+6D P0-1/D2-7 ; apc 8000-8001
Usage: fig_agg.py <run_dir> [<run_dir> ...]
"""
from __future__ import annotations
import csv
import json
import re
import statistics
import sys
from pathlib import Path
def arm_of(name: str):
# New driver naming (run_conc.sh / run_reuse_fixed.sh): "..._<CONFIG>_rep<r>".
if "8C-proxy" in name:
return "colo", list(range(8)), [], list(range(8000, 8008))
if "6P+2D" in name:
return "6P+2D", [0, 1, 2, 3, 4, 5], [6, 7], list(range(8000, 8006))
if "2P+6D" in name:
return "2P+6D", [0, 1], [2, 3, 4, 5, 6, 7], list(range(8000, 8002))
if "4P+4D" in name:
return "4P+4D", [0, 1, 2, 3], [4, 5, 6, 7], list(range(8000, 8004))
# Legacy naming (original May-30 corrected runs).
if "_colo_" in name or name.endswith("_colo"):
return "colo", list(range(8)), [], list(range(8000, 8008))
if "_pd6_" in name:
return "6P+2D", [0, 1, 2, 3, 4, 5], [6, 7], list(range(8000, 8006))
if "_pd2_" in name:
return "2P+6D", [0, 1], [2, 3, 4, 5, 6, 7], list(range(8000, 8002))
if "_pd4_" in name or "_pd_" in name:
return "4P+4D", [0, 1, 2, 3], [4, 5, 6, 7], list(range(8000, 8004))
return "?", list(range(8)), [], list(range(8000, 8008))
def util_split(run: Path, pgpus, dgpus):
win = {}
wp = run / "run_window.json"
if wp.exists():
win = json.load(open(wp))
t0, t1 = win.get("t_start_unix"), win.get("t_end_unix")
csvp = run / "gpu_util.csv"
if not csvp.exists():
return None, None
by = {}
for row in csv.DictReader(open(csvp)):
try:
ts = float(row["timestamp"]); g = int(row["gpu"]); u = float(row["util_pct"])
except (ValueError, KeyError):
continue
if t0 and not (t0 <= ts <= t1):
continue
by.setdefault(g, []).append(u)
pm = [v for g in pgpus for v in by.get(g, [])]
dm = [v for g in dgpus for v in by.get(g, [])]
return (statistics.fmean(pm) if pm else None,
statistics.fmean(dm) if dm else None)
def apc(run: Path, ports):
f = run / "instance_apc.txt"
if not f.exists():
return None
q = h = 0
for line in open(f):
m = dict(re.findall(r"(\w+)=(\S+)", line))
try:
p = int(m.get("port", -1))
except ValueError:
continue
if p in ports:
q += float(m.get("queries", 0)); h += float(m.get("hits", 0))
return (h / q) if q else None
def main():
args = sys.argv[1:]
as_json = False
if "--json" in args:
as_json = True
args = [a for a in args if a != "--json"]
rows = []
for d in args:
run = Path(d)
sp = run / "replay_metrics.summary.json"
if not sp.exists():
continue
s = json.load(open(sp))
arm, pg, dg, ports = arm_of(run.name)
# `or {}` because a fully-collapsed arm (0 successes) writes these as null,
# and dict.get(k, {}) returns null (not {}) when the key exists with value null.
lat = s.get("latency_stats_s") or {}
ttft = s.get("ttft_stats_s") or {}
tpot = s.get("tpot_stats_s") or {}
wall = s.get("wall_clock_s") or 1.0
out = s.get("actual_output_tokens_stats") or {}
n = s.get("success_count", 0); req = s.get("request_count", 0)
tot_out = out.get("count", 0) * out.get("mean", 0)
tps = tot_out / wall
pu, du = util_split(run, pg, dg)
a = apc(run, ports)
rows.append({
"name": run.name, "arm": arm, "n": n, "req": req,
"e2e_p50": lat.get("p50"), "e2e_p90": lat.get("p90"), "e2e_p99": lat.get("p99"),
"e2e_mean": lat.get("mean"),
"ttft_p90": ttft.get("p90"), "tpot_p99": tpot.get("p99"),
"tps": tps, "wall": wall, "pu": pu, "du": du, "apc": a,
})
if as_json:
print(json.dumps(rows))
return
def f(x, w=7, p=1):
return f"{x:>{w}.{p}f}" if isinstance(x, (int, float)) else f"{'-':>{w}}"
hdr = (f"{'run':<34}{'arm':>7}{'ok/req':>9}{'E2Ep50':>8}{'E2Ep90':>8}{'E2Ep99':>8}"
f"{'TPS':>8}{'Putil':>7}{'Dutil':>7}{'APC%':>7}{'TTFTp90':>9}{'TPOTp99ms':>10}")
print(hdr); print("-" * len(hdr))
for r in sorted(rows, key=lambda r: r["name"]):
print(f"{r['name']:<34}{r['arm']:>7}{str(r['n'])+'/'+str(r['req']):>9}"
f"{f(r['e2e_p50'])}{f(r['e2e_p90'])}{f(r['e2e_p99'])}"
f"{f(r['tps'],8,1)}{f(r['pu'])}{f(r['du'])}"
f"{f((r['apc'] or 0)*100)}{f(r['ttft_p90'],9,2)}"
f"{f((r['tpot_p99'] or 0)*1000,10,1)}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,18 @@
#!/bin/bash
# Sample GPU utilization every 5s, output CSV
# Usage: bash gpu_monitor.sh <output_file> [interval_s]
# Runs until killed (Ctrl+C or kill)
OUT="${1:-/tmp/gpu_util.csv}"
INTERVAL="${2:-5}"
echo "timestamp,gpu,util_pct,mem_used_mb,mem_total_mb,power_w" > "$OUT"
while true; do
TS=$(date +%s.%N)
nvidia-smi --query-gpu=index,utilization.gpu,memory.used,memory.total,power.draw \
--format=csv,noheader,nounits 2>/dev/null | while IFS=', ' read -r idx util mem_used mem_total power; do
echo "$TS,$idx,$util,$mem_used,$mem_total,$power"
done >> "$OUT"
sleep "$INTERVAL"
done

View File

@@ -0,0 +1,71 @@
"""Per-GPU utilization report from gpu_util.csv (companion to bench_report.py).
bench_report's per-worker GPU util needs request routing (breakdown.json), which
the MB5 proxy doesn't log. But worker == GPU by index, and the prefill/decode role
split is fixed by config, so per-GPU util from gpu_util.csv directly answers
"GPU utils by workers" — and for PD it exposes the key signal: are the prefill-side
GPUs saturated while the decode-side idles (or vice versa, or stalled at ~0)?
Usage:
gpu_util_report.py <run_dir> [--prefill-gpus 0,1,2,3 --decode-gpus 4,5,6,7]
"""
from __future__ import annotations
import argparse
import csv
import json
import statistics
from pathlib import Path
def pct(xs, p):
xs = sorted(xs)
return xs[max(0, min(len(xs) - 1, int(round(p / 100 * (len(xs) - 1)))))] if xs else None
def main():
ap = argparse.ArgumentParser()
ap.add_argument("run_dir", type=Path)
ap.add_argument("--prefill-gpus", default="")
ap.add_argument("--decode-gpus", default="")
a = ap.parse_args()
win = {}
wp = a.run_dir / "run_window.json"
if wp.exists():
win = json.load(open(wp))
t0, t1 = win.get("t_start_unix"), win.get("t_end_unix")
csvp = a.run_dir / "gpu_util.csv"
if not csvp.exists():
print(f"{a.run_dir.name}: gpu_util.csv absent"); return
by_gpu = {}
for row in csv.DictReader(open(csvp)):
try:
ts = float(row["timestamp"]); g = int(row["gpu"]); u = float(row["util_pct"]); m = float(row["mem_used_mb"])
except (ValueError, KeyError):
continue
if t0 and not (t0 <= ts <= t1):
continue
by_gpu.setdefault(g, {"u": [], "m": []})
by_gpu[g]["u"].append(u); by_gpu[g]["m"].append(m)
print(f"=== {a.run_dir.name}: per-GPU util over replay window ({sum(len(d['u']) for d in by_gpu.values())} samples) ===")
print(f"{'gpu':>4}{'util_mean':>11}{'util_p90':>10}{'util_max':>10}{'mem_max_GB':>12}")
for g in sorted(by_gpu):
u, m = by_gpu[g]["u"], by_gpu[g]["m"]
print(f"{g:>4}{statistics.fmean(u):>11.1f}{pct(u,90):>10.1f}{max(u):>10.1f}{max(m)/1024:>12.1f}")
def agg(gpus, label):
gpus = [int(x) for x in gpus.split(",") if x != ""]
us = [v for g in gpus for v in by_gpu.get(g, {}).get("u", [])]
if us:
print(f" {label:<14} gpus={gpus} util mean={statistics.fmean(us):.1f}% p90={pct(us,90):.1f}% max={max(us):.1f}%")
if a.prefill_gpus:
agg(a.prefill_gpus, "prefill-side")
if a.decode_gpus:
agg(a.decode_gpus, "decode-side")
if __name__ == "__main__":
main()

View File

@@ -69,6 +69,13 @@ run_one() {
source "${VENV}/bin/activate" source "${VENV}/bin/activate"
local replay_out="${rundir}/replay_metrics.jsonl" local replay_out="${rundir}/replay_metrics.jsonl"
mkdir -p "$(dirname "${replay_out}")" mkdir -p "$(dirname "${replay_out}")"
# bench_report.py inputs: worker->gpu map (worker i == gpu i for every config;
# for PD, workers 0-3 are producers on gpu0-3, 4-7 consumers on gpu4-7).
printf '{"base_port":8000,"n_instances":8,"gpu_indices":[0,1,2,3,4,5,6,7]}\n' \
> "${rundir}/bench_config.json"
# per-GPU utilization timeseries over the replay window (2s sampling)
bash "${SCRIPT_DIR}/gpu_monitor.sh" "${rundir}/gpu_util.csv" 2 >/dev/null 2>&1 &
local GPU_MON=$!
local t0 local t0
t0=$(date +%s.%N) t0=$(date +%s.%N)
if ! PYTHONPATH="${FRESH_ROOT}" python -m replayer \ if ! PYTHONPATH="${FRESH_ROOT}" python -m replayer \
@@ -82,6 +89,7 @@ run_one() {
t1=$(date +%s.%N) t1=$(date +%s.%N)
local wall=$(python -c "print(${t1} - ${t0})") local wall=$(python -c "print(${t1} - ${t0})")
echo "[mb5-run] REPLAY FAILED after ${wall} s; see ${OUT_ROOT}/${config}_rep${rep}_replay.log" echo "[mb5-run] REPLAY FAILED after ${wall} s; see ${OUT_ROOT}/${config}_rep${rep}_replay.log"
kill "${GPU_MON}" 2>/dev/null || true
bash "${LAUNCH}" stop > /dev/null 2>&1 || true bash "${LAUNCH}" stop > /dev/null 2>&1 || true
return 1 return 1
fi fi
@@ -91,6 +99,9 @@ run_one() {
wall_clock_s=$(python -c "print(${t1} - ${t0})") wall_clock_s=$(python -c "print(${t1} - ${t0})")
echo "[mb5-run] replay done in ${wall_clock_s}s" echo "[mb5-run] replay done in ${wall_clock_s}s"
echo "${wall_clock_s}" > "${rundir}/wall_clock_s.txt" echo "${wall_clock_s}" > "${rundir}/wall_clock_s.txt"
kill "${GPU_MON}" 2>/dev/null || true
printf '{"t_start_unix":%s,"t_end_unix":%s}\n' "${t0}" "${t1}" > "${rundir}/run_window.json"
cp -f "${replay_out}" "${rundir}/metrics.jsonl" # bench_report.py expects metrics.jsonl
# Per-instance prefix-cache counters, scraped from each backend BEFORE # Per-instance prefix-cache counters, scraped from each backend BEFORE
# teardown. For PD this is the only honest reuse signal: producer ports # teardown. For PD this is the only honest reuse signal: producer ports

View File

@@ -0,0 +1,144 @@
#!/usr/bin/env bash
# Orchestrator for MB5: for each CONFIG × rep, bring up the stack, run a
# trace replay against it, collect KV snapshots and replayer metrics,
# tear down.
#
# Designed to be run on dash1 (or any host with cpfs mounted at
# /home/admin/cpfs/wjh/).
#
# Env vars (with defaults):
# CONFIGS : space-separated MB5 configs (default: "8C 6P+2D 4P+4D 2P+6D")
# REPS : reps per config (default: 3)
# TRACE : trace JSONL path
# (default: /home/admin/cpfs/wjh/agentic-kv/traces/w600_r0.0015_st30.jsonl)
# RUN_TAG : output root tag (default: $(date +%Y%m%d_%H%M%S))
# REQUEST_LIMIT : optional, cap replay requests (default: none)
set -eo pipefail
FRESH_ROOT="/home/admin/cpfs/wjh/agentic-kv-fresh"
# MB5_VENV lets a second host use an isolated venv clone (see mb5_launch.sh).
VENV="${MB5_VENV:-${FRESH_ROOT}/.venv}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
LAUNCH="${SCRIPT_DIR}/mb5_launch.sh"
REPLAYER_DIR="${FRESH_ROOT}/replayer"
CONFIGS="${CONFIGS:-8C 6P+2D 4P+4D 2P+6D}"
REPS="${REPS:-3}"
TRACE="${TRACE:-/home/admin/cpfs/wjh/agentic-kv/traces/w600_r0.0015_st30.jsonl}"
RUN_TAG="${RUN_TAG:-$(date +%Y%m%d_%H%M%S)}"
MODEL_NAME="${MODEL_NAME:-/home/admin/cpfs/wjh/models/Qwen/Qwen3-Coder-30B-A3B-Instruct}"
REQUEST_LIMIT_ARG=""
if [ -n "${REQUEST_LIMIT:-}" ]; then
REQUEST_LIMIT_ARG="--request-limit ${REQUEST_LIMIT}"
fi
OUT_ROOT="${FRESH_ROOT}/mb5_runs/${RUN_TAG}"
mkdir -p "${OUT_ROOT}"
echo "[mb5-run] RUN_TAG=${RUN_TAG}"
echo "[mb5-run] OUT_ROOT=${OUT_ROOT}"
echo "[mb5-run] CONFIGS=${CONFIGS}"
echo "[mb5-run] REPS=${REPS}"
echo "[mb5-run] TRACE=${TRACE}"
run_one() {
local config="$1" rep="$2"
local label="${RUN_TAG}_${config}_rep${rep}"
local rundir="${FRESH_ROOT}/mb5_runs/${label}"
echo ""
echo "======== ${config} rep${rep} ========"
# Launch
if ! CONFIG="${config}" RUN_LABEL="${RUN_TAG}_${config}_rep${rep}" \
bash "${LAUNCH}" start > "${OUT_ROOT}/${config}_rep${rep}_launch.log" 2>&1; then
echo "[mb5-run] LAUNCH FAILED for ${config} rep${rep}; see ${OUT_ROOT}/${config}_rep${rep}_launch.log"
return 1
fi
# Extract ENDPOINTS line emitted by mb5_launch.sh
local endpoints
endpoints=$(grep "^ENDPOINTS=" "${OUT_ROOT}/${config}_rep${rep}_launch.log" | tail -1 | cut -d= -f2-)
if [ -z "${endpoints}" ]; then
echo "[mb5-run] ERROR: no ENDPOINTS in launch log"
bash "${LAUNCH}" stop > /dev/null 2>&1 || true
return 1
fi
echo "[mb5-run] endpoints: ${endpoints}"
# Replay
source "${VENV}/bin/activate"
local replay_out="${rundir}/replay_metrics.jsonl"
mkdir -p "$(dirname "${replay_out}")"
# per-GPU utilization timeseries over the replay window (2s sampling)
bash "${FRESH_ROOT}/microbench/fresh_setup/gpu_monitor.sh" "${rundir}/gpu_util.csv" 2 >/dev/null 2>&1 &
local GPU_MON=$!
local t0
t0=$(date +%s.%N)
if ! PYTHONPATH="${FRESH_ROOT}" python -m replayer \
--endpoint "${endpoints}" \
--trace "${TRACE}" \
--output "${replay_out}" \
--model "${MODEL_NAME}" \
${REQUEST_LIMIT_ARG} \
> "${OUT_ROOT}/${config}_rep${rep}_replay.log" 2>&1; then
local t1
t1=$(date +%s.%N)
local wall=$(python -c "print(${t1} - ${t0})")
echo "[mb5-run] REPLAY FAILED after ${wall} s; see ${OUT_ROOT}/${config}_rep${rep}_replay.log"
kill "${GPU_MON}" 2>/dev/null || true
bash "${LAUNCH}" stop > /dev/null 2>&1 || true
return 1
fi
local t1
t1=$(date +%s.%N)
local wall_clock_s
wall_clock_s=$(python -c "print(${t1} - ${t0})")
echo "[mb5-run] replay done in ${wall_clock_s}s"
echo "${wall_clock_s}" > "${rundir}/wall_clock_s.txt"
kill "${GPU_MON}" 2>/dev/null || true
printf '{"t_start_unix":%s,"t_end_unix":%s}\n' "${t0}" "${t1}" > "${rundir}/run_window.json"
# Per-instance prefix-cache counters, scraped from each backend BEFORE
# teardown. For PD this is the only honest reuse signal: producer ports
# (the low ones) show cross-turn prefix-cache hits; the consumer's
# per-request cached_tokens is meaningless (it counts transferred KV).
{
for p in 8000 8001 8002 8003 8004 8005 8006 8007; do
m=$(curl -s --noproxy '*' "http://127.0.0.1:${p}/metrics" 2>/dev/null) || continue
q=$(printf '%s' "$m" | awk '/^vllm:prefix_cache_queries_total/{print $2; exit}')
h=$(printf '%s' "$m" | awk '/^vllm:prefix_cache_hits_total/{print $2; exit}')
[ -n "${q}" ] && echo "port=${p} queries=${q} hits=${h}"
done
} > "${rundir}/instance_apc.txt" 2>/dev/null || true
# Stop launch (cleans up vllm + proxy; reverts patch on last call)
CONFIG="${config}" RUN_LABEL="${RUN_TAG}_${config}_rep${rep}" \
bash "${LAUNCH}" stop > "${OUT_ROOT}/${config}_rep${rep}_stop.log" 2>&1 || true
sleep 10 # cooldown so GPUs settle before next config
echo "[mb5-run] DONE ${config} rep${rep}"
}
# Quick check that the launch script and replayer are reachable
if [ ! -f "${LAUNCH}" ]; then echo "missing ${LAUNCH}"; exit 1; fi
if [ ! -d "${REPLAYER_DIR}" ]; then echo "missing ${REPLAYER_DIR}"; exit 1; fi
if [ ! -f "${TRACE}" ]; then echo "missing trace ${TRACE}"; exit 1; fi
# Iterate
failures=0
for config in ${CONFIGS}; do
for ((rep=1; rep<=REPS; rep++)); do
if ! run_one "${config}" "${rep}"; then
failures=$((failures+1))
fi
done
done
# Final patch revert (defensive — mb5_launch.sh stop also reverts)
python "${SCRIPT_DIR}/instrument_kv_snapshot.py" --revert --venv "${VENV}" 2>/dev/null || true
echo ""
echo "======== ALL CONFIGS DONE ========"
echo "failures: ${failures}"
echo "results under: ${FRESH_ROOT}/mb5_runs/${RUN_TAG}_*"
echo "to plot: python plot_kv_pool_timeline.py --run-tag ${RUN_TAG}"

View File

@@ -0,0 +1,98 @@
"""Compute a per-run summary directly from replay_metrics.jsonl (for partial / in-flight runs).
Used when the replayer hasn't completed (so replay_metrics.summary.json doesn't exist
yet) but enough records have streamed to disk to read out the per-arm result.
Also accepts a finished run's directory and prints the same one-line summary for
apples-to-apples comparison.
"""
from __future__ import annotations
import json
import re
import statistics
import sys
from pathlib import Path
def stats(xs):
xs = sorted(xs)
n = len(xs)
if n == 0:
return None
return {
"n": n,
"mean": statistics.fmean(xs),
"p50": xs[n // 2],
"p90": xs[int(0.9 * (n - 1))],
"p99": xs[int(0.99 * (n - 1))],
}
def apc(run: Path, producer_ports):
f = run / "instance_apc.txt"
if not f.exists():
return None
q = h = 0.0
for line in open(f):
m = dict(re.findall(r"(\w+)=(\S+)", line))
try:
p = int(m.get("port", -1))
except ValueError:
continue
if p in producer_ports:
q += float(m.get("queries", 0))
h += float(m.get("hits", 0))
return (h / q) if q else None
def main():
for d in sys.argv[1:]:
run = Path(d)
# prefer the live replay_metrics.jsonl (so partials work); fall back to metrics.jsonl
for fn in ("replay_metrics.partial.jsonl", "replay_metrics.jsonl", "metrics.jsonl"):
p = run / fn
if p.exists():
rec_path = p
break
else:
print(f"{run.name}: no records"); continue
recs = [json.loads(l) for l in open(rec_path)]
oks = [r for r in recs if r.get("error") is None]
lat = stats([r["latency_s"] for r in oks if "latency_s" in r])
ttft = stats([r["ttft_s"] for r in oks if "ttft_s" in r])
tpot = stats([r["tpot_s"] for r in oks if "tpot_s" in r])
out = sum(r.get("actual_output_tokens", r.get("output_length", 0)) for r in oks)
ts = [r["t_dispatch_unix"] for r in oks if "t_dispatch_unix" in r]
tf = [r["t_finish_unix"] for r in oks if "t_finish_unix" in r]
span = max(tf) - min(ts) if ts and tf else 0
tps = out / span if span else 0
# producer ports by arm tag in dirname
n = run.name
if "_colo_" in n:
ports = list(range(8000, 8008))
elif "_pd6_" in n:
ports = list(range(8000, 8006))
elif "_pd2_" in n:
ports = list(range(8000, 8002))
else:
ports = list(range(8000, 8004))
a = apc(run, ports)
print(f"{run.name}")
print(f" n_ok={len(oks)}/{len(recs)}"
+ (f" (target=1214 -> {len(oks)*100/1214:.1f}%)" if len(recs) < 1214 else ""))
if lat:
print(f" E2E mean={lat['mean']:.2f} p50={lat['p50']:.2f} p90={lat['p90']:.2f} p99={lat['p99']:.2f}")
if ttft:
print(f" TTFT mean={ttft['mean']:.2f} p50={ttft['p50']:.2f} p90={ttft['p90']:.2f} p99={ttft['p99']:.2f}")
if tpot:
print(f" TPOT mean={tpot['mean']*1000:.1f}ms p90={tpot['p90']*1000:.1f}ms p99={tpot['p99']*1000:.1f}ms")
print(f" output_tokens={out:.0f} span={span:.0f}s TPS={tps:.0f}")
if a is not None:
print(f" producer APC={a*100:.1f}%")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,104 @@
"""Linear vs LMetric routing on the real agentic trace (first600s).
Visualizes the wall-cap finding: with the 2x-colo-wall cap on PD-disagg arms,
linear and LMetric reach the *same* success-rate ceiling -- the static P:D
split has a structural completion ceiling that does not depend on the routing
policy or on how long you keep retrying. Routing affects only how much wall
time is wasted on requests that will never succeed.
Inputs : analysis/v2/fig4l_lmetric.json (8 arms, both traces; we use first600s)
analysis/v2/fig4r_linear.json (4 arms, first600s, PD wall-capped)
Output : figs/v2/fig4_linear_vs_lmetric.png
"""
from __future__ import annotations
import json
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
ROOT = Path(__file__).resolve().parents[2]
DATA = ROOT / "analysis" / "v2"
OUT = ROOT / "figs" / "v2" / "fig4_linear_vs_lmetric.png"
ARMS = ["colo", "6P+2D", "4P+4D", "2P+6D"]
POLICY_COLOR = {"linear": "#9467bd", "lmetric": "#2ca02c"}
POLICY_LABEL = {"linear": "linear (cache-aware + session-affinity)",
"lmetric": "LMetric (P_tokens × BS)"}
def pick(rows, arm, trace="first600s"):
for r in rows:
if r["arm"] == arm and r["trace"] == trace:
return r
return None
def main():
lin = json.load(open(DATA / "fig4r_linear.json"))
lme = json.load(open(DATA / "fig4l_lmetric.json"))
# colo wall (linear) sets the 2x cap reference
colo_lin_wall = pick(lin, "colo")["wall"]
cap = 2 * colo_lin_wall
fig, axes = plt.subplots(1, 3, figsize=(15, 4.5))
x = np.arange(len(ARMS))
w = 0.38
# (a) success rate
ax = axes[0]
for i, (pol, rows) in enumerate([("linear", lin), ("lmetric", lme)]):
vals = [pick(rows, a)["n"] / pick(rows, a)["req"] * 100 for a in ARMS]
bars = ax.bar(x + (i - 0.5) * w, vals, w, color=POLICY_COLOR[pol], label=POLICY_LABEL[pol])
for bx, bv in zip(x + (i - 0.5) * w, vals):
ax.annotate(f"{bv:.0f}%", (bx, bv + 1.5), ha="center", fontsize=8)
ax.axhline(100, color="grey", ls=":", lw=1)
ax.set_xticks(x); ax.set_xticklabels(ARMS)
ax.set_ylabel("success rate (% of trace)"); ax.set_ylim(0, 115)
ax.set_title("(a) success ceiling is policy-invariant")
ax.legend(fontsize=8, loc="upper right"); ax.grid(alpha=.3, axis="y")
# (b) wall (log y) with cap line
ax = axes[1]
for i, (pol, rows) in enumerate([("linear", lin), ("lmetric", lme)]):
vals = [pick(rows, a)["wall"] for a in ARMS]
bars = ax.bar(x + (i - 0.5) * w, vals, w, color=POLICY_COLOR[pol],
label=POLICY_LABEL[pol])
for bx, bv, r in zip(x + (i - 0.5) * w, vals,
[pick(rows, a) for a in ARMS]):
mark = "" if r.get("capped") else ""
ax.annotate(f"{bv:.0f}s{mark}", (bx, bv * 1.05), ha="center", fontsize=7)
ax.axhline(cap, color="red", ls="--", lw=1.5,
label=f"2× colo wall cap = {cap:.0f}s")
ax.set_xticks(x); ax.set_xticklabels(ARMS)
ax.set_ylabel("wall-clock (s, log)"); ax.set_yscale("log")
ax.set_title("(b) linear w/ cap vs lmetric w/o cap — ⊗ = cap-killed")
ax.legend(fontsize=8, loc="upper left"); ax.grid(alpha=.3, which="both", axis="y")
# (c) goodput per minute of wall (success rate / wall × 60)
ax = axes[2]
for i, (pol, rows) in enumerate([("linear", lin), ("lmetric", lme)]):
vals = [pick(rows, a)["n"] / pick(rows, a)["wall"] * 60 for a in ARMS]
bars = ax.bar(x + (i - 0.5) * w, vals, w, color=POLICY_COLOR[pol], label=POLICY_LABEL[pol])
for bx, bv in zip(x + (i - 0.5) * w, vals):
ax.annotate(f"{bv:.1f}", (bx, bv + max(vals) * 0.02),
ha="center", fontsize=8)
ax.set_xticks(x); ax.set_xticklabels(ARMS)
ax.set_ylabel("goodput (successful req / min)")
ax.set_title("(c) linear+cap is 1.517× more wall-efficient on PD")
ax.legend(fontsize=8, loc="upper right"); ax.grid(alpha=.3, axis="y")
fig.suptitle("Fig 4r — Linear vs LMetric on the real agentic trace (first600s, "
"PD-disagg wall-capped at 2× colo)",
fontsize=12, y=1.0)
fig.tight_layout()
fig.savefig(OUT, dpi=130, bbox_inches="tight")
print(f"wrote {OUT}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,113 @@
"""Render the LMetric PD-colo vs PD-disagg figure on the real agentic trace.
Input : analysis/v2/fig4l_lmetric.json (8 arms = 4 ratios x 2 traces)
Output : figs/v2/fig4_lmetric_pd_vs_colo.png
Four panels x four ratios x two traces:
(a) completion rate %
(b) E2E latency (mean / p50 / p90)
(c) throughput (output tokens / second)
(d) bench wall-clock seconds
The thesis the figure visualizes: with LMetric routing,
- colo (elastic 8-GPU pool) holds 100% completion on both traces
- every PD-disagg ratio fails (completion 14-65%), and the failure mode
rotates with the split (pd2 = prefill-bound, pd6 = decode-bound)
- routing policy does not rescue PD-disagg; the bottleneck is structural.
"""
from __future__ import annotations
import json
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
ROOT = Path(__file__).resolve().parents[2]
DATA = ROOT / "analysis" / "v2" / "fig4l_lmetric.json"
OUT = ROOT / "figs" / "v2" / "fig4_lmetric_pd_vs_colo.png"
OUT.parent.mkdir(parents=True, exist_ok=True)
ARMS = ["colo", "6P+2D", "4P+4D", "2P+6D"] # decode-rich -> prefill-rich
TRACES = ["first600s", "full"]
TRACE_LABEL = {"first600s": "first600s (1.35 req/s, high load)",
"full": "full w600 (0.42 req/s, original §3)"}
COLOR = {"first600s": "#1f77b4", "full": "#ff7f0e"}
def pick(rows, trace, arm):
for r in rows:
if r["trace"] == trace and r["arm"] == arm:
return r
return None
def main():
rows = json.load(open(DATA))
fig, axes = plt.subplots(2, 2, figsize=(12, 8))
width = 0.38
x = np.arange(len(ARMS))
# (a) completion %
ax = axes[0, 0]
for i, tr in enumerate(TRACES):
vals = [pick(rows, tr, a)["n"] / pick(rows, tr, a)["req"] * 100 for a in ARMS]
bars = ax.bar(x + (i - 0.5) * width, vals, width, color=COLOR[tr], label=TRACE_LABEL[tr])
for bx, bv in zip(x + (i - 0.5) * width, vals):
ax.annotate(f"{bv:.0f}%", (bx, bv + 1.5), ha="center", fontsize=8)
ax.axhline(100, color="grey", ls=":", lw=1)
ax.set_xticks(x); ax.set_xticklabels(ARMS)
ax.set_ylabel("completion (%)"); ax.set_ylim(0, 115)
ax.set_title("(a) request completion — colo holds 100%, all PD ratios fail")
ax.legend(fontsize=8); ax.grid(alpha=.3, axis="y")
# (b) E2E percentiles
ax = axes[0, 1]
for i, tr in enumerate(TRACES):
p50 = [pick(rows, tr, a)["e2e"]["p50"] for a in ARMS]
p90 = [pick(rows, tr, a)["e2e"]["p90"] for a in ARMS]
off = (i - 0.5) * width
ax.bar(x + off, p90, width, color=COLOR[tr], alpha=0.55, label=f"{tr} p90")
ax.bar(x + off, p50, width, color=COLOR[tr], alpha=1.0, label=f"{tr} p50")
ax.axhline(600, color="red", ls=":", lw=1, label="600 s request timeout")
ax.set_xticks(x); ax.set_xticklabels(ARMS)
ax.set_ylabel("E2E latency (s, log)"); ax.set_yscale("log")
ax.set_title("(b) E2E p50 (solid) + p90 (faded) — PD pegs at the timeout")
ax.legend(fontsize=7, ncol=2); ax.grid(alpha=.3, which="both", axis="y")
# (c) TPS
ax = axes[1, 0]
for i, tr in enumerate(TRACES):
vals = [pick(rows, tr, a)["tps"] for a in ARMS]
ax.bar(x + (i - 0.5) * width, vals, width, color=COLOR[tr], label=TRACE_LABEL[tr])
for bx, bv in zip(x + (i - 0.5) * width, vals):
ax.annotate(f"{bv:.0f}", (bx, bv + 4), ha="center", fontsize=8)
ax.set_xticks(x); ax.set_xticklabels(ARMS)
ax.set_ylabel("throughput (output tokens/s)")
ax.set_title("(c) throughput — PD throughput crashes 5100×")
ax.legend(fontsize=8); ax.grid(alpha=.3, axis="y")
# (d) wall (min)
ax = axes[1, 1]
for i, tr in enumerate(TRACES):
vals = [pick(rows, tr, a)["wall"] / 60 for a in ARMS]
ax.bar(x + (i - 0.5) * width, vals, width, color=COLOR[tr], label=TRACE_LABEL[tr])
for bx, bv in zip(x + (i - 0.5) * width, vals):
ax.annotate(f"{bv:.0f}m", (bx, bv * 1.05), ha="center", fontsize=8)
ax.set_xticks(x); ax.set_xticklabels(ARMS)
ax.set_ylabel("bench wall-clock (min, log)"); ax.set_yscale("log")
ax.set_title("(d) wall-clock — PD drain dilates the run")
ax.legend(fontsize=8); ax.grid(alpha=.3, which="both", axis="y")
fig.suptitle("Fig 4 (LMetric) — PD-colo vs PD-disagg on the real agentic trace "
"(`w600_r0.0015_st30`), clean stack, cache-aware LMetric routing",
fontsize=12, y=1.0)
fig.tight_layout()
fig.savefig(OUT, dpi=130, bbox_inches="tight")
print(f"wrote {OUT}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,189 @@
"""Render the three PD-vs-colo crossover figures from fig_agg JSON dumps.
Inputs (produced by `fig_agg.py --json`):
analysis/mb5_pd_ablation/fig1_reuse_fixed.json reuse axis (N=8, FIXED real
prefill delta=2048; vary cached prefix -> reuse = pfx/(pfx+delta).
Controlled-variable: real new-prefill work is constant across the sweep,
only the cached fraction (and total context) grows. Supersedes the old
fig1.json, which held input=8192 and sliced prefix out of it so delta
shrank 15x as reuse rose — a confound, not a pure reuse axis.)
analysis/mb5_pd_ablation/fig2.json shape axis (N=8, reuse~70%)
analysis/mb5_pd_ablation/fig3_conc32k.json concurrency (in32768/out128,
reuse~0.984 = 32256 resident + 512 real new-prefill per turn; retuned
2026-05-31 to the agentic corner so PD pays the full-context per-turn
KV-transfer tax while colo keeps it resident; vary N by step 8 up to the
mean-E2E<=10s SLO ceiling)
Each figure overlays colo + the three PD ratios and marks the PD-best advantage.
All three share the corrected (uncontaminated, e13391e-gated-off) stack.
"""
from __future__ import annotations
import json
import re
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
ROOT = Path(__file__).resolve().parents[2]
DATA = ROOT / "analysis" / "mb5_pd_ablation"
OUT = ROOT / "figs" / "mb5_pd_ablation"
OUT.mkdir(parents=True, exist_ok=True)
PD_ARMS = ["2P+6D", "4P+4D", "6P+2D"]
STYLE = {
"colo": dict(color="k", marker="o", lw=2.4, ls="-", label="colo (8×kv_both)"),
"2P+6D": dict(color="#1f77b4", marker="s", lw=1.6, ls="--", label="PD 2P+6D"),
"4P+4D": dict(color="#2ca02c", marker="^", lw=1.6, ls="--", label="PD 4P+4D"),
"6P+2D": dict(color="#ff7f0e", marker="v", lw=1.6, ls="--", label="PD 6P+2D"),
}
def load(name):
return json.load(open(DATA / name))
def by_axis(rows, keyfn):
"""Group rows -> {axis_val: {arm: row}}."""
out = {}
for r in rows:
k = keyfn(r["name"])
if k is None:
continue
out.setdefault(k, {})[r["arm"]] = r
return out
def pd_best(armmap, metric="e2e_p90"):
vals = [(a, armmap[a][metric]) for a in PD_ARMS
if a in armmap and armmap[a].get(metric) is not None]
return min(vals, key=lambda t: t[1]) if vals else (None, None)
def series(grp, xs, arm, metric):
return [grp[x][arm].get(metric) if arm in grp[x] else None for x in xs]
# ---------- Fig 1: reuse axis ----------
def _reuse_pct(name):
"""Reuse % from a `reuse_p{pfx}_d{delta}_{arm}` run name: pfx/(pfx+delta)."""
m = re.search(r"_p(\d+)_d(\d+)", name)
if not m:
return None
pfx, delta = int(m.group(1)), int(m.group(2))
return round(pfx / (pfx + delta) * 100)
def fig_reuse():
g = by_axis(load("fig1_reuse_fixed.json"), _reuse_pct)
xs = sorted(g)
reuse = xs
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(11, 4.2))
for arm in ["colo", *PD_ARMS]:
ax1.plot(reuse, series(g, xs, arm, "e2e_p90"), **STYLE[arm])
ax1.set_xlabel("intra-session KV reuse (%) [fixed real prefill, delta=2048]")
ax1.set_ylabel("E2E latency p90 (s)")
ax1.set_title("(a) E2E-p90 vs reuse (N=8, delta=2048/out256)")
ax1.legend(fontsize=8); ax1.grid(alpha=.3)
adv, putil = [], []
for x in xs:
co = g[x]["colo"]["e2e_p90"]; _, b = pd_best(g[x])
adv.append(co / b if b else None)
a = pd_best(g[x])[0]
putil.append(g[x][a].get("pu") if a else None)
ax2.plot(reuse, adv, color="purple", marker="D", lw=2, label="PD-best advantage (colo/PD)")
ax2.axhline(1.0, color="grey", ls=":", lw=1)
ax2.set_xlabel("intra-session KV reuse (%)"); ax2.set_ylabel("advantage (>1 = PD wins)")
ax2b = ax2.twinx()
ax2b.plot(reuse, putil, color="brown", marker="x", lw=1.4, ls="-.", label="PD-best prefill-GPU util")
ax2b.set_ylabel("prefill-GPU util (%)", color="brown"); ax2b.tick_params(axis="y", colors="brown")
ax2.set_title("(b) advantage erodes; prefill GPUs go idle")
l1, la1 = ax2.get_legend_handles_labels(); l2, la2 = ax2b.get_legend_handles_labels()
ax2.legend(l1 + l2, la1 + la2, fontsize=8, loc="center right"); ax2.grid(alpha=.3)
fig.suptitle("Fig 1 — Reuse axis (fixed real prefill delta=2048): PD's edge vs rising cache reuse",
fontsize=11, y=1.02)
fig.tight_layout(); p = OUT / "fig1_reuse_axis.png"; fig.savefig(p, dpi=130, bbox_inches="tight")
print("wrote", p)
# ---------- Fig 2: shape axis ----------
def fig_shape():
g = by_axis(load("fig2.json"),
lambda n: ((int(m.group(1)), int(m.group(2)))
if (m := re.search(r"_in(\d+)_out(\d+)_", n)) else None))
xs = sorted(g, key=lambda t: t[0]) # ascending input
labels = [f"in{i}\nout{o}" for i, o in xs]
xi = list(range(len(xs)))
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(11, 4.2))
for arm in ["colo", *PD_ARMS]:
ax1.plot(xi, series(g, xs, arm, "e2e_p90"), **STYLE[arm])
ax1.set_xticks(xi); ax1.set_xticklabels(labels, fontsize=7)
ax1.set_xlabel("shape (decode-heavy → prefill-heavy)"); ax1.set_ylabel("E2E latency p90 (s)")
ax1.set_title("(a) E2E-p90 vs shape (N=8, reuse~70%)")
ax1.legend(fontsize=8); ax1.grid(alpha=.3)
adv, comp = [], []
for x in xs:
co = g[x]["colo"]["e2e_p90"]; a, b = pd_best(g[x])
adv.append(co / b if b else None)
# completion of the worst PD arm (exposes catastrophic ratio)
worst = min((g[x][arm]["n"] / g[x][arm]["req"]) for arm in PD_ARMS if arm in g[x])
comp.append(worst * 100)
ax2.plot(xi, adv, color="purple", marker="D", lw=2, label="PD-best advantage (colo/PD)")
ax2.axhline(1.0, color="grey", ls=":", lw=1)
ax2.set_xticks(xi); ax2.set_xticklabels(labels, fontsize=7)
ax2.set_xlabel("shape"); ax2.set_ylabel("advantage (>1 = PD wins)")
ax2b = ax2.twinx()
ax2b.plot(xi, comp, color="red", marker="x", lw=1.4, ls="-.", label="worst-PD-arm completion %")
ax2b.set_ylabel("worst PD completion (%)", color="red"); ax2b.tick_params(axis="y", colors="red")
ax2b.set_ylim(80, 101)
ax2.set_title("(b) advantage peaks mid-sweep; wrong ratio catastrophic at prefill extreme")
l1, la1 = ax2.get_legend_handles_labels(); l2, la2 = ax2b.get_legend_handles_labels()
ax2.legend(l1 + l2, la1 + la2, fontsize=8, loc="lower left"); ax2.grid(alpha=.3)
fig.suptitle("Fig 2 — Shape axis: PD wins decode-heavy, ties prefill-heavy; optimal ratio rotates",
fontsize=11, y=1.02)
fig.tight_layout(); p = OUT / "fig2_shape_axis.png"; fig.savefig(p, dpi=130, bbox_inches="tight")
print("wrote", p)
# ---------- Fig 3: concurrency axis ----------
def fig_conc():
g = by_axis(load("fig3_conc32k.json"),
lambda n: (int(m.group(1)) if (m := re.search(r"_N(\d+)_", n)) else None))
xs = sorted(g)
fig, (ax1, ax2, ax3) = plt.subplots(1, 3, figsize=(15, 4.2))
# (a) request completion % — the headline (latency percentiles count successes
# only, so they understate PD; completion is the honest collapse signal).
for arm in ["colo", *PD_ARMS]:
comp = [(g[x][arm]["n"] / g[x][arm]["req"] * 100) if arm in g[x] else None for x in xs]
ax1.plot(xs, comp, **STYLE[arm])
ax1.axhline(100, color="grey", ls=":", lw=1)
ax1.set_xticks(xs); ax1.set_xticklabels(xs, fontsize=7)
ax1.set_xlabel("concurrent sessions N"); ax1.set_ylabel("request completion (%)")
ax1.set_title("(a) completion: colo 100%, PD collapses"); ax1.legend(fontsize=8); ax1.grid(alpha=.3)
for arm in ["colo", *PD_ARMS]:
ax2.plot(xs, series(g, xs, arm, "e2e_mean"), **STYLE[arm])
ax2.axhline(10.0, color="red", ls=":", lw=1, label="SLO 10s")
ax2.set_yscale("log"); ax2.set_xticks(xs); ax2.set_xticklabels(xs, fontsize=7)
ax2.set_xlabel("concurrent sessions N"); ax2.set_ylabel("E2E latency mean (s, log)")
ax2.set_title("(b) mean-E2E (successes only)"); ax2.legend(fontsize=8); ax2.grid(alpha=.3, which="both")
for arm in ["colo", *PD_ARMS]:
ax3.plot(xs, series(g, xs, arm, "tps"), **STYLE[arm])
ax3.set_xticks(xs); ax3.set_xticklabels(xs, fontsize=7)
ax3.set_xlabel("concurrent sessions N"); ax3.set_ylabel("throughput (tok/s)")
ax3.set_title("(c) TPS"); ax3.legend(fontsize=8); ax3.grid(alpha=.3)
fig.suptitle("Fig 3 — Concurrency axis (in32768/out128, reuse~0.984, PD capped 600s / colo uncapped): "
"colo degrades gracefully (100% completion), PD collapses earlier as N rises",
fontsize=10, y=1.02)
fig.tight_layout(); p = OUT / "fig3_concurrency_axis.png"; fig.savefig(p, dpi=130, bbox_inches="tight")
print("wrote", p)
if __name__ == "__main__":
fig_reuse(); fig_shape(); fig_conc()

View File

@@ -0,0 +1,26 @@
#!/usr/bin/env bash
# Unattended serial PD-ablation campaign: reuse sweep -> conc sweep.
# STRICTLY one driver at a time (the hard lesson): each inner driver brings up and
# tears down its own vLLM per config via scripts/mb5_run_gpu.sh, and the two sweeps
# run sequentially (reuse fully finishes + tears down before conc starts). We verify
# GPUs are clear between sweeps. NO set -e here: a sub-sweep nonzero must NOT skip the
# other sweep; rc is captured and reported. Detached launch writes a DONE marker.
cd /home/admin/cpfs/wjh/agentic-kv-fresh
export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
FS=microbench/fresh_setup
echo "=== CAMPAIGN START $(date) ==="
echo "=== [1/2] REUSE SWEEP (fixed real prefill delta=2048, out=256, reuse 20-95%, N=8) $(date) ==="
bash "$FS/run_reuse_fixed.sh"; rc_reuse=$?
echo "=== reuse sweep rc=$rc_reuse $(date) ==="
sleep 15
echo "--- GPU mem after reuse sweep (expect ~0 before conc) ---"
nvidia-smi --query-gpu=index,memory.used --format=csv,noheader | head -8
echo "=== [2/2] CONC SWEEP (in=32768 reuse=0.984, balanced N grid 8 16 32 48 64 96 128) $(date) ==="
NLIST="8 16 32 48 64 96 128" bash "$FS/run_conc.sh"; rc_conc=$?
echo "=== conc sweep rc=$rc_conc $(date) ==="
echo "=== CAMPAIGN DONE reuse_rc=$rc_reuse conc_rc=$rc_conc $(date) ==="

View File

@@ -0,0 +1,26 @@
#!/usr/bin/env bash
# Campaign 2 (2026-05-31): two extra reuse sweeps at out=128 (user request:
# delta=1024/out=128 and delta=2048/out=128), then the capped conc restart.
# STRICTLY one driver at a time; reuse sweeps run uncapped (mild collapse, matches
# the existing d2048/o256 sweep), conc runs with the PD-arm wall-cap. NO set -e.
cd /home/admin/cpfs/wjh/agentic-kv-fresh
export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
FS=microbench/fresh_setup
echo "=== CAMPAIGN2 START $(date) ==="
echo "=== [1/3] REUSE delta=1024 out=128 (reuse 0.33-0.97) $(date) ==="
DELTA=1024 OL=128 bash "$FS/run_reuse_fixed.sh"; rc1=$?
echo "=== reuse d1024 o128 rc=$rc1 $(date) ==="
sleep 12; nvidia-smi --query-gpu=index,memory.used --format=csv,noheader | head -8
echo "=== [2/3] REUSE delta=2048 out=128 (reuse 0.20-0.95) $(date) ==="
DELTA=2048 OL=128 bash "$FS/run_reuse_fixed.sh"; rc2=$?
echo "=== reuse d2048 o128 rc=$rc2 $(date) ==="
sleep 12; nvidia-smi --query-gpu=index,memory.used --format=csv,noheader | head -8
echo "=== [3/3] CONC capped (PD wall=${CONC_PD_MAXDUR:-600}s, colo uncapped), N 8..128 $(date) ==="
NLIST="8 16 32 48 64 96 128" bash "$FS/run_conc.sh"; rc3=$?
echo "=== conc rc=$rc3 $(date) ==="
echo "=== CAMPAIGN2 DONE reuse_d1024_o128=$rc1 reuse_d2048_o128=$rc2 conc=$rc3 $(date) ==="

View File

@@ -0,0 +1,20 @@
#!/usr/bin/env bash
# Campaign 3 (2026-05-31): the uncapped d2048/o128 reuse sweep stalled on a
# collapse-draining high-reuse PD arm (4P+4D @ reuse 0.90, ~1 req/several-min).
# Finish it by re-running ONLY the high-reuse points (0.90, 0.95) WITH the PD
# wall-cap (low-reuse arms already completed and are cap-insensitive). Then run
# the capped conc sweep. STRICTLY serial. NO set -e.
cd /home/admin/cpfs/wjh/agentic-kv-fresh
export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
FS=microbench/fresh_setup
echo "=== CAMPAIGN3 START $(date) ==="
echo "=== [1/2] finish reuse d2048/o128: re-run pts pfx=18432,38912 (PD capped 500s) $(date) ==="
DELTA=2048 OL=128 PFXS="18432 38912" REUSE_PD_MAXDUR=500 bash "$FS/run_reuse_fixed.sh"; rc1=$?
echo "=== reuse d2048 o128 finish rc=$rc1 $(date) ==="
sleep 12; nvidia-smi --query-gpu=index,memory.used --format=csv,noheader | head -8
echo "=== [2/2] CONC capped (PD wall=600s, colo uncapped), N 8..128 $(date) ==="
NLIST="8 16 32 48 64 96 128" CONC_PD_MAXDUR=600 bash "$FS/run_conc.sh"; rc2=$?
echo "=== conc rc=$rc2 $(date) ==="
echo "=== CAMPAIGN3 DONE reuse_finish=$rc1 conc=$rc2 $(date) ==="

View File

@@ -0,0 +1,70 @@
#!/usr/bin/env bash
# Concurrency axis, agentic-corner config. Supersedes old fig3 (in~8192/out256).
# RETUNED 2026-05-31 for realism (C2): hold total context in=32768 but shrink the
# real per-turn new-prefill to delta=512 and push reuse to 0.984 (real agentic
# reuse ->99.6%). prefix 32256 + delta 512. out=128. This is the corner that
# exposes PD's structural tax: colo keeps the 32k resident KV local, but PD must
# KV-transfer the whole 32k context every turn even though only 512 tokens are new
# (C2 PD-tax ~250-450x). Sweep closed-loop N by step 8 up to mean-E2E<=SLO ceiling.
# Wiring per memory project-mb5-pd-ablation-wiring: .venv_dash0, traces_synth/,
# CONFIG 8C-proxy + PD, MB5_P_ROUTING=session + MB5_COLO_ROUTING=session,
# N=REPLAY_MAX_INFLIGHT closed loop + REPLAY_INTER_TURN_THINK_S,
# REPLAY_NO_REALIZED_PREFIX=1. RUN ONLY ONE DRIVER AT A TIME (shared GPUs/ports).
set -eo pipefail
cd /home/admin/cpfs/wjh/agentic-kv-fresh
export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
VPY="$MB5_VENV/bin/python"
PFX="${PFX:-32256}"; DELTA="${DELTA:-512}"; OL="${OL:-128}" # reuse=0.984, in=32768
THINK="${THINK:-0.5}"; TURNS="${TURNS:-8}"
NSTART="${NSTART:-8}"; NSTEP="${NSTEP:-8}"; NMAX="${NMAX:-128}"
NLIST="${NLIST:-}" # explicit N grid (overrides NSTART/STEP/MAX), e.g. "8 16 32 48 64 96 128"
CONC_PD_MAXDUR="${CONC_PD_MAXDUR:-600}" # wall-deadline (s) for PD arms only; bounds collapsed-arm
# drain (un-run turns = failures). colo (8C-proxy) runs UNCAPPED
# so the headline reference is always fully measured.
SLO="${SLO:-10.0}"
SESS_PER_N="${SESS_PER_N:-4}"
CFGS="${CFGS:-8C-proxy 2P+6D 4P+4D 6P+2D}"
ONLY_N="${ONLY_N:-}"
run_N() {
local N="$1"; local sess=$(( SESS_PER_N * N ))
local tag="conc32k_N${N}"; local trace="traces_synth/${tag}.jsonl"
"$VPY" scripts/gen_synthetic_trace.py --out "$trace" --mode regular \
--qps "$sess" --duration-s 1 --turns "$TURNS" \
--prefix-len "$PFX" --delta-len "$DELTA" --output-len "$OL" --seed 42 >/dev/null
echo "[conc32k] N=$N sess=$sess in=$((PFX+DELTA)) out=$OL -> $trace"
for cfg in $CFGS; do
echo " -> $cfg"
local dur=""; [ "$cfg" != "8C-proxy" ] && dur="$CONC_PD_MAXDUR" # colo uncapped
MB5_P_ROUTING=session MB5_COLO_ROUTING=session \
REPLAY_MAX_INFLIGHT="$N" REPLAY_INTER_TURN_THINK_S="$THINK" REPLAY_NO_REALIZED_PREFIX=1 \
REPLAY_MAX_DURATION="$dur" \
CONFIGS="$cfg" REPS=1 TRACE="$trace" RUN_TAG="$tag" \
bash scripts/mb5_run_gpu.sh >/dev/null 2>&1 || echo " [warn] ${tag}_${cfg} failed" >&2
done
local d="mb5_runs/${tag}_8C-proxy_rep1"
if [ -f "$d/replay_metrics.summary.json" ]; then
"$VPY" scripts/fig_agg.py --json "$d" 2>/dev/null \
| "$VPY" -c "import sys,json;r=json.load(sys.stdin);print(r[0].get('e2e_mean') if r else 'nan')"
else echo nan; fi
}
if [ -n "$ONLY_N" ]; then
echo "[conc32k] SMOKE N=$ONLY_N cfgs='$CFGS'"
t0=$(date +%s); m=$(run_N "$ONLY_N"); t1=$(date +%s)
echo "[conc32k] SMOKE N=$ONLY_N colo mean-E2E=${m}s wall=$(( t1 - t0 ))s; compare:"
"$VPY" scripts/fig_agg.py mb5_runs/conc32k_N${ONLY_N}_*_rep1 2>&1
exit 0
fi
if [ -n "$NLIST" ]; then NSEQ="$NLIST"; else NSEQ=$(seq "$NSTART" "$NSTEP" "$NMAX"); fi
for N in $NSEQ; do
echo "[conc32k] === N=$N ==="
m=$(run_N "$N"); echo "[conc32k] N=$N colo mean-E2E=${m}s"
over=$("$VPY" -c "print(1 if float('${m}')>${SLO} else 0)" 2>/dev/null || echo 0)
[ "$over" = "1" ] && { echo "[conc32k] colo crossed SLO ${SLO}s at N=$N -> stop"; break; }
done
dirs=(); for d in mb5_runs/conc32k_N*_rep1; do [ -d "$d" ] && dirs+=("$d"); done
"$VPY" scripts/fig_agg.py --json "${dirs[@]}" > analysis/mb5_pd_ablation/fig3_conc32k.json
echo "[conc32k] done -> analysis/mb5_pd_ablation/fig3_conc32k.json (${#dirs[@]} dirs)"

View File

@@ -0,0 +1,72 @@
#!/usr/bin/env bash
# Reuse axis, DONE RIGHT (controlled variable). Supersedes old fig1.
# Hold REAL (uncached) prefill work constant: --delta-len = U fixed.
# Vary only --prefix-len = C -> reuse = C/(C+U). Context grows with reuse but
# the tokens that must actually be prefilled each turn stays = U.
# Old fig1 held input=8192 and sliced prefix out of it, so delta shrank 15x as
# reuse rose -> confounded "more reuse" with "less prefill". This fixes that.
#
# Wiring matches the corrected MB5 stack (see memory project-mb5-pd-ablation-wiring):
# .venv_dash0, traces_synth/, CONFIG 8C-proxy + PD, MB5_P_ROUTING=session,
# N injected via REPLAY_MAX_INFLIGHT (closed loop) + REPLAY_INTER_TURN_THINK_S,
# REPLAY_NO_REALIZED_PREFIX=1 (reuse governed by hash_ids, required for this sweep).
set -eo pipefail
cd /home/admin/cpfs/wjh/agentic-kv-fresh
export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
VPY="$MB5_VENV/bin/python"
DELTA="${DELTA:-2048}" # fixed real prefill per turn (USER-CHOSEN)
OL="${OL:-256}"
N="${N:-8}"
THINK="${THINK:-0.5}"
TURNS="${TURNS:-8}"
NSESS="${NSESS:-48}" # number of sessions (closed-loop: arrival rate is
# irrelevant, only the count matters; ~6 waves at N=8)
PFXS="${PFXS:-512 2048 4096 8192 18432 38912}" # reuse .20 .50 .67 .80 .90 .95
CFGS="${CFGS:-8C-proxy 2P+6D 4P+4D 6P+2D}"
REUSE_PD_MAXDUR="${REUSE_PD_MAXDUR:-500}" # wall-deadline (s) for PD arms only (colo uncapped):
# bounds the collapse-drain that stalls high-reuse PD arms
# (un-run turns = failures, honest completion%). 0/empty = off.
ONLY_PFX="${ONLY_PFX:-}" # smoke a single prefix then exit
run_point() { # <pfx>
local pfx="$1"
local reuse; reuse=$(python3 -c "print(f'{$pfx/($pfx+$DELTA):.3f}')")
local tag="reuse_p${pfx}_d${DELTA}_o${OL}" # _o${OL} so different output lens don't collide
local trace="traces_synth/${tag}.jsonl"
# Closed-loop: pass NSESS as qps with duration 1 so n_sessions = NSESS
# exactly (gen_regular: n_sessions = int(duration_s * session_qps)).
"$VPY" scripts/gen_synthetic_trace.py --out "$trace" --mode regular \
--qps "$NSESS" --duration-s 1 --turns "$TURNS" \
--prefix-len "$pfx" --delta-len "$DELTA" --output-len "$OL" --seed 42 >/dev/null
echo "[reuse] pfx=$pfx delta=$DELTA reuse=$reuse in=$((pfx+DELTA)) -> $trace"
for cfg in $CFGS; do
echo " -> $cfg"
# Both routings set to session so BOTH colo (kv_both) and PD producers
# pin a session's turns to one instance and reuse its prefix cache — the
# fair cache-aware comparison. P_ROUTING is ignored by colo, COLO_ROUTING
# by PD, so setting both is harmless and symmetric.
local dur=""; [ "$cfg" != "8C-proxy" ] && dur="$REUSE_PD_MAXDUR" # colo uncapped
MB5_P_ROUTING=session MB5_COLO_ROUTING=session \
REPLAY_MAX_INFLIGHT="$N" REPLAY_INTER_TURN_THINK_S="$THINK" \
REPLAY_NO_REALIZED_PREFIX=1 REPLAY_MAX_DURATION="$dur" \
CONFIGS="$cfg" REPS=1 TRACE="$trace" RUN_TAG="$tag" \
bash scripts/mb5_run_gpu.sh >/dev/null 2>&1 || echo " [warn] $cfg failed" >&2
done
}
if [ -n "$ONLY_PFX" ]; then
echo "[reuse] SMOKE pfx=$ONLY_PFX cfgs='$CFGS'"
t0=$(date +%s); run_point "$ONLY_PFX"; t1=$(date +%s)
echo "[reuse] SMOKE done wall=$(( t1 - t0 ))s; compare:"
"$VPY" scripts/fig_agg.py mb5_runs/reuse_p${ONLY_PFX}_d${DELTA}_o${OL}_*_rep1
exit 0
fi
for pfx in $PFXS; do run_point "$pfx"; done
# Aggregate ONLY this sweep's dirs (matched by delta+output) so the three
# reuse figures (d2048/o256, d1024/o128, d2048/o128) never cross-contaminate.
dirs=(); for d in mb5_runs/reuse_*_d${DELTA}_o${OL}_*_rep1; do [ -d "$d" ] && dirs+=("$d"); done
OUTJSON="analysis/mb5_pd_ablation/fig1_reuse_d${DELTA}_o${OL}.json"
"$VPY" scripts/fig_agg.py --json "${dirs[@]}" > "$OUTJSON"
echo "[reuse] done -> $OUTJSON (${#dirs[@]} dirs)"

View File

@@ -0,0 +1,120 @@
#!/usr/bin/env python3
"""
f2a sensitivity: how does the intra/cross reuse split move as we change the
single-turn session fraction? (Tests whether the old 93%-intra sample vs 54.6%
full-trace gap is just session-mixture selection bias.)
Keep ALL multi-turn sessions; downsample single-turn sessions to hit each target
single-turn fraction f. Re-run the LRU (last-touched), reuse-hits-only
classification on the filtered request stream.
python3 f2a_mixture_sweep.py ~/ali-trace/.../051315-051317.jsonl /tmp/f2a_sweep.json
"""
import sys, json, time, random
from collections import Counter, defaultdict
PATH = sys.argv[1]
OUT = sys.argv[2] if len(sys.argv) > 2 else "/tmp/f2a_sweep.json"
random.seed(0)
t0 = time.time()
chat_parent = {}
records = []
with open(PATH) as f:
for line in f:
d = json.loads(line)
cid = d["chat_id"]; pc = d.get("parent_chat_id")
chat_parent[cid] = 0 if pc is None else pc
records.append((d.get("timestamp", 0.0), cid, d.get("hash_ids") or []))
sys.stderr.write(f"[{time.time()-t0:.0f}s] loaded {len(records)}\n")
root_cache = {}
def resolve_root(cid):
chain = []; cur = cid
while True:
if cur in root_cache:
r = root_cache[cur]; break
p = chat_parent.get(cur, 0)
if p == 0 or p not in chat_parent:
r = cur; break
chain.append(cur); cur = p
if len(chain) > 100000:
r = cur; break
for nd in chain:
root_cache[nd] = r
root_cache[cid] = r
return r
records.sort(key=lambda x: x[0])
roots = [resolve_root(cid) for _, cid, _ in records]
req_per_root = Counter(roots)
single_roots = [r for r, c in req_per_root.items() if c == 1]
multi_roots = [r for r, c in req_per_root.items() if c >= 2]
M = len(multi_roots)
sys.stderr.write(f"[{time.time()-t0:.0f}s] roots: single={len(single_roots)} multi={M}\n")
GAP_EDGES = [1, 10, 60, 300, 1800, 3600, float("inf")]
def gbucket(g):
for i, e in enumerate(GAP_EDGES):
if g < e:
return i
return len(GAP_EDGES) - 1
def classify(kept): # kept=None -> keep all
last_root = {}; last_ts = {}
intra = cross = new = 0
rec_i = [0] * len(GAP_EDGES); rec_c = [0] * len(GAP_EDGES)
for (ts, cid, hs), r in zip(records, roots):
if kept is not None and r not in kept:
continue
for h in hs:
lr = last_root.get(h)
if lr is None:
new += 1
else:
gb = gbucket(max(0.0, ts - last_ts[h]))
if lr == r:
intra += 1; rec_i[gb] += 1
else:
cross += 1; rec_c[gb] += 1
last_root[h] = r; last_ts[h] = ts
return intra, cross, new, rec_i, rec_c
def cum_le(rec, idx): # cumulative fraction with gap-bucket <= idx
tot = sum(rec) or 1
return sum(rec[: idx + 1]) / tot
targets = [("full", None), (0.75, None), (0.50, None),
(0.25, None), (0.10, None), (0.00, None)]
rows = []
for label, _ in targets:
if label == "full":
kept = None
f_actual = len(single_roots) / (len(single_roots) + M)
else:
f = float(label)
S = min(len(single_roots), int(round(M * f / (1 - f)))) if f < 1 else len(single_roots)
keep_single = set(random.sample(single_roots, S)) if S < len(single_roots) else set(single_roots)
kept = set(multi_roots) | keep_single
f_actual = S / (S + M)
intra, cross, new, rec_i, rec_c = classify(kept)
reuse = intra + cross
n_sess = (len(single_roots) + M) if kept is None else len(kept)
row = {
"target": label, "single_turn_frac": round(f_actual, 4), "n_sessions": n_sess,
"new": new, "intra": intra, "cross": cross, "reuse": reuse,
"intra_frac_of_reuse": round(intra / reuse, 4),
"cross_frac_of_reuse": round(cross / reuse, 4),
"intra_le60s": round(cum_le(rec_i, 2), 4),
"cross_le60s": round(cum_le(rec_c, 2), 4),
}
rows.append(row)
sys.stderr.write(f"[{time.time()-t0:.0f}s] f={row['single_turn_frac']}: "
f"intra={row['intra_frac_of_reuse']} cross={row['cross_frac_of_reuse']}\n")
json.dump({"rows": rows, "n_single": len(single_roots), "n_multi": M}, open(OUT, "w"), indent=2)
print(f"{'single-turn%':>12} {'sessions':>10} {'intra%':>8} {'cross%':>8} {'intra<=60s':>11} {'cross<=60s':>11}")
for r in rows:
print(f"{r['single_turn_frac']*100:>11.1f}% {r['n_sessions']:>10} "
f"{r['intra_frac_of_reuse']*100:>7.1f}% {r['cross_frac_of_reuse']*100:>7.1f}% "
f"{r['intra_le60s']*100:>10.1f}% {r['cross_le60s']*100:>10.1f}%")

View File

@@ -0,0 +1,84 @@
{
"rows": [
{
"target": "full",
"single_turn_frac": 0.9026,
"n_sessions": 1307276,
"new": 20650883,
"intra": 65166144,
"cross": 54134925,
"reuse": 119301069,
"intra_frac_of_reuse": 0.5462,
"cross_frac_of_reuse": 0.4538,
"intra_le60s": 0.8865,
"cross_le60s": 0.8706
},
{
"target": 0.75,
"single_turn_frac": 0.75,
"n_sessions": 509144,
"new": 15446415,
"intra": 66081759,
"cross": 26932604,
"reuse": 93014363,
"intra_frac_of_reuse": 0.7104,
"cross_frac_of_reuse": 0.2896,
"intra_le60s": 0.8844,
"cross_le60s": 0.8568
},
{
"target": 0.5,
"single_turn_frac": 0.5,
"n_sessions": 254572,
"new": 12843712,
"intra": 66548474,
"cross": 18990485,
"reuse": 85538959,
"intra_frac_of_reuse": 0.778,
"cross_frac_of_reuse": 0.222,
"intra_le60s": 0.8832,
"cross_le60s": 0.8881
},
{
"target": 0.25,
"single_turn_frac": 0.25,
"n_sessions": 169715,
"new": 11553493,
"intra": 66732961,
"cross": 16726772,
"reuse": 83459733,
"intra_frac_of_reuse": 0.7996,
"cross_frac_of_reuse": 0.2004,
"intra_le60s": 0.8827,
"cross_le60s": 0.9087
},
{
"target": 0.1,
"single_turn_frac": 0.1,
"n_sessions": 141429,
"new": 11036894,
"intra": 66798704,
"cross": 16084035,
"reuse": 82882739,
"intra_frac_of_reuse": 0.8059,
"cross_frac_of_reuse": 0.1941,
"intra_le60s": 0.8826,
"cross_le60s": 0.9152
},
{
"target": 0.0,
"single_turn_frac": 0.0,
"n_sessions": 127286,
"new": 10724167,
"intra": 66834552,
"cross": 15799085,
"reuse": 82633637,
"intra_frac_of_reuse": 0.8088,
"cross_frac_of_reuse": 0.1912,
"intra_le60s": 0.8825,
"cross_le60s": 0.9184
}
],
"n_single": 1179990,
"n_multi": 127286
}

View File

@@ -0,0 +1,182 @@
#!/usr/bin/env python3
"""
f2a reuse topology — full-trace, infinite-KV-cache decomposition (LRU semantics).
Question: on the real 2h cluster trace, assuming an *infinite* KV cache (nothing
ever evicted), where do prefix-cache REUSE HITS come from?
We classify only reuse hits (the 1st occurrence of a block is `new` = irreducible
prefill; it is reported only as context for the APC ceiling, not in the split).
A block (content-addressed `hash_id`) processed in timestamp order. For each hit we
look at the block's **most recent prior holder** (last computed OR used = LRU):
intra : last touch was the SAME session (parent_chat_id chain)
cross : last touch was a DIFFERENT session
After classifying, the block's last-holder / last-time are updated to the current
request (LRU refresh). The reuse "recency" is the **LRU reuse distance** = time since
the block was last touched (what a finite TTL/LRU cache would need to retain).
`cross` is further resolved by *block popularity* = number of distinct sessions that
ever touch the block: a handful of hugely-popular blocks are the shared system/tool
prefix; low-popularity cross blocks are genuine cross-session content.
Run on dash2 (trace lives there):
python3 f2a_reuse_topology_analyze.py \
~/ali-trace/trace-glm5.1-formatted/051315-051317.jsonl /tmp/f2a_result.json
"""
import sys, json, time
from collections import defaultdict
PATH = sys.argv[1]
OUT = sys.argv[2] if len(sys.argv) > 2 else "/tmp/f2a_result.json"
POP_CAP = 4096 # cap per-block root set; >= this is "very shared", buckets unaffected
t0 = time.time()
chat_parent = {}
records = [] # (ts, chat_id, hash_ids)
total_input_tokens = 0
total_blocks = 0
turn1 = 0
n = 0
with open(PATH) as f:
for line in f:
d = json.loads(line)
cid = d["chat_id"]
pc = d.get("parent_chat_id")
chat_parent[cid] = 0 if pc is None else pc
hs = d.get("hash_ids") or []
records.append((d.get("timestamp", 0.0), cid, hs))
total_input_tokens += d.get("input_length", 0) or 0
total_blocks += len(hs)
if (d.get("turn", 1) or 1) == 1:
turn1 += 1
n += 1
sys.stderr.write(f"[{time.time()-t0:.0f}s] loaded {n} reqs, {total_blocks} block-occ\n")
# resolve session root by following parent_chat_id to turn-1 / out-of-window head
root_cache = {}
def resolve_root(cid):
chain = []
cur = cid
while True:
if cur in root_cache:
r = root_cache[cur]; break
p = chat_parent.get(cur, 0)
if p == 0 or p not in chat_parent:
r = cur; break
chain.append(cur); cur = p
if len(chain) > 100000:
r = cur; break
for nd in chain:
root_cache[nd] = r
root_cache[cid] = r
return r
records.sort(key=lambda r: r[0])
sys.stderr.write(f"[{time.time()-t0:.0f}s] sorted by ts\n")
last_root = {} # block -> root of MOST RECENT holder (LRU)
last_ts = {} # block -> ts of most recent touch (LRU)
roots_of = defaultdict(set) # block -> set of distinct roots (capped) = popularity
intra_cnt = defaultdict(int) # block -> intra reuse hits
cross_cnt = defaultdict(int) # block -> cross reuse hits
new = intra = cross = 0
# LRU reuse distance of each hit: gap = consumer_ts - last_touch_ts
GAP_EDGES = [1, 10, 60, 300, 1800, 3600, float("inf")] # seconds
GAP_LABELS = ["<1s", "1-10s", "10-60s", "1-5min", "5-30min", "30-60min", ">60min"]
rec_intra = [0] * len(GAP_EDGES)
rec_cross = [0] * len(GAP_EDGES)
def gap_bucket(g):
for i, e in enumerate(GAP_EDGES):
if g < e:
return i
return len(GAP_EDGES) - 1
for ts, cid, hs in records:
if not hs:
continue
r = resolve_root(cid)
for h in hs:
lr = last_root.get(h)
if lr is None:
new += 1 # first compute: not a hit
else:
gb = gap_bucket(max(0.0, ts - last_ts[h]))
if lr == r:
intra += 1; intra_cnt[h] += 1; rec_intra[gb] += 1
else:
cross += 1; cross_cnt[h] += 1; rec_cross[gb] += 1
last_root[h] = r # LRU refresh: now held by current session
last_ts[h] = ts
s = roots_of[h]
if len(s) < POP_CAP:
s.add(r)
sys.stderr.write(f"[{time.time()-t0:.0f}s] classified: new={new} intra={intra} cross={cross}\n")
# popularity buckets: distinct sessions touching a block
POP_EDGES = [2, 10, 100, 1000, float("inf")]
POP_LABELS = ["1 (private)", "2-9", "10-99", "100-999", ">=1000"]
def pop_bucket(p):
if p <= 1:
return 0
for i, e in enumerate(POP_EDGES[1:], start=1):
if p < e:
return i
return len(POP_LABELS) - 1
pop_blocks = [0] * len(POP_LABELS)
pop_intra = [0] * len(POP_LABELS)
pop_cross = [0] * len(POP_LABELS)
for h in last_root:
p = len(roots_of[h])
b = pop_bucket(p)
pop_blocks[b] += 1
pop_intra[b] += intra_cnt.get(h, 0)
pop_cross[b] += cross_cnt.get(h, 0)
eff_blk = total_input_tokens / total_blocks if total_blocks else 0.0
total_occ = new + intra + cross
reuse = intra + cross
result = {
"trace": PATH,
"semantics": "LRU last-touched; reuse-hits only (new excluded from split)",
"n_requests": n,
"n_sessions": len(set(resolve_root(c) for c in chat_parent)),
"turn1_frac": turn1 / n,
"block_size_tokens_eff": eff_blk,
"total_input_tokens": total_input_tokens,
"total_block_occ": total_occ,
"distinct_blocks": len(last_root),
"new_occ": new, # context only
"apc_ceiling": reuse / total_occ, # context only
# REUSE-ONLY decomposition (the headline)
"reuse_total": reuse,
"reuse": {"intra": intra, "cross": cross},
"reuse_frac": {"intra": intra / reuse, "cross": cross / reuse},
# cross resolved by popularity (over reuse hits)
"pop_labels": POP_LABELS,
"pop_blocks": pop_blocks,
"pop_intra": pop_intra,
"pop_cross": pop_cross,
# LRU reuse-distance recency (over reuse hits)
"gap_labels": GAP_LABELS,
"rec_intra": rec_intra,
"rec_cross": rec_cross,
}
with open(OUT, "w") as f:
json.dump(result, f, indent=2)
sys.stderr.write(f"[{time.time()-t0:.0f}s] wrote {OUT}\n")
# human summary
print(json.dumps({k: result[k] for k in
("n_requests","n_sessions","distinct_blocks","reuse_total",
"reuse_frac","apc_ceiling")}, indent=2))
print(f"new(context)={new} intra={intra} cross={cross}")
print("popularity blocks / intra-hits / cross-hits:")
for i, lab in enumerate(POP_LABELS):
print(f" {lab:>12}: {pop_blocks[i]:>10} | {pop_intra[i]:>11} | {pop_cross[i]:>11}")
print("LRU reuse-distance intra / cross:")
for i, lab in enumerate(GAP_LABELS):
print(f" {lab:>8}: {rec_intra[i]:>11} | {rec_cross[i]:>11}")

View File

@@ -0,0 +1,77 @@
{
"trace": "051315-051317.jsonl",
"semantics": "LRU last-touched; reuse-hits only (new excluded from split)",
"n_requests": 2114220,
"n_sessions": 1307276,
"turn1_frac": 0.6183254344391785,
"block_size_tokens_eff": 508.1517503092776,
"total_input_tokens": 71116829368,
"total_block_occ": 139951952,
"distinct_blocks": 20650883,
"new_occ": 20650883,
"apc_ceiling": 0.8524430513123532,
"reuse_total": 119301069,
"reuse": {
"intra": 65166144,
"cross": 54134925
},
"reuse_frac": {
"intra": 0.5462326913432771,
"cross": 0.45376730865672293
},
"pop_labels": [
"1 (private)",
"2-9",
"10-99",
"100-999",
">=1000"
],
"pop_blocks": [
14581108,
5535433,
517069,
16153,
1120
],
"pop_intra": [
44515497,
14288480,
5421050,
924419,
16698
],
"pop_cross": [
0,
20230912,
13750153,
7689338,
12464522
],
"gap_labels": [
"<1s",
"1-10s",
"10-60s",
"1-5min",
"5-30min",
"30-60min",
">60min"
],
"rec_intra": [
390952,
26060293,
31317556,
5877221,
1384772,
109673,
25677
],
"rec_cross": [
13222875,
22254795,
11653445,
4965765,
1747487,
220816,
69742
]
}

View File

@@ -30,12 +30,23 @@ def main() -> None:
default=float(_env_think) if _env_think else None, default=float(_env_think) if _env_think else None,
help="Closed-loop think-time (s) after each turn completes; " help="Closed-loop think-time (s) after each turn completes; "
"ignore absolute trace schedule. Env: REPLAY_INTER_TURN_THINK_S") "ignore absolute trace schedule. Env: REPLAY_INTER_TURN_THINK_S")
p.add_argument("--no-realized-prefix",
action="store_true",
default=bool(os.environ.get("REPLAY_NO_REALIZED_PREFIX")),
help="Controlled-reuse mode: prompt = hash-built tokens only "
"(reuse set by hash_ids). Env: REPLAY_NO_REALIZED_PREFIX")
p.add_argument("--dispatch-mode", choices=["tracets", "thinktime"], p.add_argument("--dispatch-mode", choices=["tracets", "thinktime"],
default=os.environ.get("REPLAY_DISPATCH_MODE", "tracets"), default=os.environ.get("REPLAY_DISPATCH_MODE", "tracets"),
help="tracets (Mode 1): absolute trace ts = max(prev_finished, ts). " help="tracets (Mode 1): absolute trace ts = max(prev_finished, ts). "
"thinktime (Mode 2): turn-k at prev_finished + " "thinktime (Mode 2): turn-k at prev_finished + "
"time_to_parent_chat. Env: REPLAY_DISPATCH_MODE") "time_to_parent_chat. Env: REPLAY_DISPATCH_MODE")
p.add_argument("--request-timeout", type=float, default=600.0) p.add_argument("--request-timeout", type=float, default=600.0)
_env_maxdur = os.environ.get("REPLAY_MAX_DURATION")
p.add_argument("--max-duration", type=float,
default=float(_env_maxdur) if _env_maxdur else None,
help="Overall wall-clock deadline (s): cancel in-flight + write "
"summary (un-run turns counted as failures) to bound a "
"collapsed config's drain. Env: REPLAY_MAX_DURATION")
p.add_argument("--request-limit", type=int, default=None, p.add_argument("--request-limit", type=int, default=None,
help="Limit number of requests to replay") help="Limit number of requests to replay")
p.add_argument("-v", "--verbose", action="store_true") p.add_argument("-v", "--verbose", action="store_true")
@@ -56,7 +67,9 @@ def main() -> None:
request_limit=args.request_limit, request_limit=args.request_limit,
max_inflight_sessions=args.max_inflight_sessions, max_inflight_sessions=args.max_inflight_sessions,
inter_turn_think_s=args.inter_turn_think, inter_turn_think_s=args.inter_turn_think,
no_realized_prefix=args.no_realized_prefix,
dispatch_mode=args.dispatch_mode, dispatch_mode=args.dispatch_mode,
max_duration_s=args.max_duration,
) )
results = asyncio.run(replay_trace(config)) results = asyncio.run(replay_trace(config))

View File

@@ -66,6 +66,13 @@ class ReplayConfig:
# max_inflight_sessions=N this is a stable N-user closed-loop (no open-loop # max_inflight_sessions=N this is a stable N-user closed-loop (no open-loop
# runaway), so it removes the "immediate retrigger under load" artifact. # runaway), so it removes the "immediate retrigger under load" artifact.
inter_turn_think_s: float | None = None inter_turn_think_s: float | None = None
# Controlled-reuse mode: skip _apply_realized_prefix so each turn's prompt is
# exactly the hash-built tokens. Then prefix-cache reuse is governed solely by
# the generated hash_ids (shared prefix blocks hit, fresh delta blocks miss) —
# required for the reuse-fraction sweep, where realized-prefix would otherwise
# force every fixed-length turn to ≈ the prior turn (≈100% reuse regardless).
# Keep OFF (realized-prefix ON) for the real agentic trace.
no_realized_prefix: bool = False
# Dispatch timing for intra-session turns: # Dispatch timing for intra-session turns:
# "tracets" (Mode 1): fire at absolute trace timestamp -> effectively # "tracets" (Mode 1): fire at absolute trace timestamp -> effectively
# max(prev_finished, trace_ts); collapses think-time to 0 when # max(prev_finished, trace_ts); collapses think-time to 0 when
@@ -73,6 +80,25 @@ class ReplayConfig:
# "thinktime" (Mode 2): turn-1 at trace arrival; turn-k at # "thinktime" (Mode 2): turn-1 at trace arrival; turn-k at
# prev_finished + time_to_parent_chat (real production gap). # prev_finished + time_to_parent_chat (real production gap).
dispatch_mode: str = "tracets" dispatch_mode: str = "tracets"
# Overall wall-clock deadline for the whole replay (seconds). When exceeded,
# stop awaiting in-flight sessions, cancel them, and write the summary over
# whatever completed — un-run turns are counted as failures so completion%
# stays honest (request_count == full trace). None = no deadline (default,
# original behavior unchanged). Used to bound the slow drain of a collapsed
# config in a sweep. Env: REPLAY_MAX_DURATION.
max_duration_s: float | None = None
def _skipped_metric() -> "RequestMetrics":
"""Placeholder failure row for a turn never run due to a max_duration cutoff.
Only its error (non-None) matters: it counts toward request/error totals but
is excluded from latency/ttft/tpot percentiles (successes only)."""
return RequestMetrics(
request_id="deadline_skipped", session_id="", turn_id=-1,
trace_timestamp_s=0.0, input_length=0, output_length=0,
request_type="skipped", effective_input_length=None, cached_tokens=0,
latency_s=None, ttft_s=None, tpot_s=None, error="deadline_skipped",
)
def _build_prompt_token_ids(req: TraceRequest) -> list[int]: def _build_prompt_token_ids(req: TraceRequest) -> list[int]:
@@ -318,10 +344,9 @@ async def _run_session(
if elapsed < target_wall: if elapsed < target_wall:
await asyncio.sleep(target_wall - elapsed) await asyncio.sleep(target_wall - elapsed)
token_ids = _apply_realized_prefix( token_ids = _build_prompt_token_ids(req)
_build_prompt_token_ids(req), if not config.no_realized_prefix:
realized_context, token_ids = _apply_realized_prefix(token_ids, realized_context)
)
result = await _dispatch_request( result = await _dispatch_request(
client=client, config=config, req=req, client=client, config=config, req=req,
prompt_token_ids=token_ids, sem=request_sem, prompt_token_ids=token_ids, sem=request_sem,
@@ -410,25 +435,44 @@ async def replay_trace(config: ReplayConfig) -> list[RequestMetrics]:
trust_env=False, trust_env=False,
limits=limits, limits=limits,
) as client: ) as client:
states = [_SessionState(session_id=sid, turns=turns)
for sid, turns in sessions]
tasks = [ tasks = [
asyncio.create_task(_run_session( asyncio.create_task(_run_session(
state=_SessionState(session_id=sid, turns=turns), state=st, config=config, client=client,
config=config, client=client,
request_sem=request_sem, request_sem=request_sem,
earliest_ts=earliest_ts, sweep_start=sweep_start, earliest_ts=earliest_ts, sweep_start=sweep_start,
sink=sink, sink=sink,
session_sem=session_sem, session_sem=session_sem,
)) ))
for sid, turns in sessions for st in states
] ]
all_results = await asyncio.gather(*tasks) if config.max_duration_s and config.max_duration_s > 0:
_done, pending = await asyncio.wait(
tasks, timeout=config.max_duration_s)
if pending:
logger.warning(
"max_duration %.0fs reached: cancelling %d in-flight "
"session(s); un-run turns counted as failures",
config.max_duration_s, len(pending))
for t in pending:
t.cancel()
await asyncio.gather(*pending, return_exceptions=True)
else:
await asyncio.gather(*tasks)
finally: finally:
sink.close() sink.close()
sweep_elapsed = time.perf_counter() - sweep_start sweep_elapsed = time.perf_counter() - sweep_start
post_metrics = await _snapshot_prefix_cache_metrics(config.endpoint_url) post_metrics = await _snapshot_prefix_cache_metrics(config.endpoint_url)
flat = [m for group in all_results for m in group] # Build from the session states (identical to the gather return in the
# uncapped path) so partially-completed (cancelled) sessions still contribute
# their finished turns; pad un-run turns as failures so request_count == trace.
flat = [m for st in states for m in st.metrics]
missing = n_requests - len(flat)
if missing > 0:
flat.extend(_skipped_metric() for _ in range(missing))
summary_path = config.output_path.with_suffix(".summary.json") summary_path = config.output_path.with_suffix(".summary.json")
write_summary_json(summary_path, flat) write_summary_json(summary_path, flat)

View File

@@ -27,6 +27,13 @@ Analyzer: `scripts/bench_report.py` (summaries in `results/`).
## Result (ms; `figs/exp_d_policy_dispatch.png`) ## Result (ms; `figs/exp_d_policy_dispatch.png`)
The figure has 6 panels — TTFT mean/p90, E2E mean/p90, decode TPS (output
goodput), and the per-worker GPU-util box (8 workers/arm). Decode TPS is the
honest throughput metric (total/prefill TPS is inflated by cache-miss recompute,
e.g. LMetric); thinktime ≥ tracets on it everywhere (the system drains faster
with real think-time). The GPU-util box shows LPWL also keeps the tightest
worker balance.
| policy | mode | TTFT p90 | E2E mean | E2E p90 | E2E p99 | TPOT p90 | APC | req-bal | | policy | mode | TTFT p90 | E2E mean | E2E p90 | E2E p99 | TPOT p90 | APC | req-bal |
|---|---|---:|---:|---:|---:|---:|---:|---:| |---|---|---:|---:|---:|---:|---:|---:|---:|
| **LPWL** | tracets | 11099 | 9827 | 25366 | 93929 | 33 | 0.650 | **1.49×** | | **LPWL** | tracets | 11099 | 9827 | 25366 | 93929 | 33 | 0.650 | **1.49×** |

View File

@@ -1,8 +1,9 @@
"""exp (d): 5-policy routing under tracets vs thinktime dispatch. """exp (d): 5-policy routing under tracets vs thinktime dispatch.
Shows the ranking FLIP: under the faithful `thinktime` load the parameter-free Six panels: TTFT mean/p90, E2E mean/p90, decode-TPS (output goodput), and the
LPWL (leastwork) is the clear winner, but under `tracets` (think-collapse bursts) per-worker GPU-util distribution. Shows the ranking FLIP — under faithful
its advantage disappears (it ties unified_ab on TTFT p90 and *loses* on E2E mean). `thinktime` the parameter-free LPWL (leastwork) is the clear winner; under
`tracets` (think-collapse bursts) its advantage disappears.
Reads the two bench_report summaries; writes v2/figs/exp_d_policy_dispatch.png. Reads the two bench_report summaries; writes v2/figs/exp_d_policy_dispatch.png.
Usage: python v2/exp_d_policy_dispatch/plot.py Usage: python v2/exp_d_policy_dispatch/plot.py
@@ -13,56 +14,78 @@ import os
import matplotlib import matplotlib
matplotlib.use("Agg") matplotlib.use("Agg")
import matplotlib.pyplot as plt import matplotlib.pyplot as plt
from matplotlib.patches import Patch
HERE = os.path.dirname(__file__) HERE = os.path.dirname(__file__)
TC = json.load(open(os.path.join(HERE, "results/tracets.json"))) TC = json.load(open(os.path.join(HERE, "results/tracets.json")))
TT = json.load(open(os.path.join(HERE, "results/thinktime.json"))) TT = json.load(open(os.path.join(HERE, "results/thinktime.json")))
# canonical order: LPWL first; pretty labels
ARMS = ["leastwork", "unified_ab", "unified_def", "lmetric", "sticky"] ARMS = ["leastwork", "unified_ab", "unified_def", "lmetric", "sticky"]
LABEL = {"leastwork": "LPWL\n(leastwork)", "unified_ab": "unified\n+A+B", LABEL = {"leastwork": "LPWL\n(leastwork)", "unified_ab": "unified\n+A+B",
"unified_def": "unified\ndefault", "lmetric": "LMetric", "sticky": "sticky"} "unified_def": "unified\ndefault", "lmetric": "LMetric", "sticky": "sticky"}
C_TC, C_TT = "#d62728", "#2ca02c" # tracets red / thinktime green (match exp_c) C_TC, C_TT = "#d62728", "#2ca02c" # tracets red / thinktime green (match exp_c)
W = 0.38
def panel(ax, key, sub, title, ylab): def bar_panel(ax, tc, tt, title, ylab, fmt="{:.1f}", higher_better=False):
tc = [TC[a][key][sub] / 1000.0 for a in ARMS] # ms -> s
tt = [TT[a][key][sub] / 1000.0 for a in ARMS]
x = range(len(ARMS)) x = range(len(ARMS))
w = 0.38 b1 = ax.bar([i - W / 2 for i in x], tc, W, color=C_TC)
b1 = ax.bar([i - w / 2 for i in x], tc, w, label="tracets (burst)", color=C_TC) b2 = ax.bar([i + W / 2 for i in x], tt, W, color=C_TT)
b2 = ax.bar([i + w / 2 for i in x], tt, w, label="thinktime (faithful)", color=C_TT)
for bars in (b1, b2): for bars in (b1, b2):
for r in bars: for r in bars:
ax.text(r.get_x() + r.get_width() / 2, r.get_height(), ax.text(r.get_x() + r.get_width() / 2, r.get_height(),
f"{r.get_height():.1f}", ha="center", va="bottom", fontsize=8) fmt.format(r.get_height()), ha="center", va="bottom", fontsize=7.5)
ax.set_xticks(list(x)); ax.set_xticklabels([LABEL[a] for a in ARMS], fontsize=9) ax.set_xticks(list(x)); ax.set_xticklabels([LABEL[a] for a in ARMS], fontsize=8.5)
ax.set_ylabel(ylab); ax.set_title(title, fontsize=11) arrow = "higher = better" if higher_better else "lower = better"
ax.set_ylabel(ylab); ax.set_title(f"{title} ({arrow})", fontsize=10.5)
ax.grid(axis="y", alpha=.3); ax.set_ylim(0, max(tc + tt) * 1.18)
def gpu_panel(ax):
"""Per-worker gpu_util_mean distribution: tracets vs thinktime box per policy."""
def utils(D, a):
pw = D[a]["per_worker"]
return [pw[w]["gpu_util_mean"] for w in sorted(pw, key=int)
if pw[w].get("gpu_util_mean") is not None]
for i, a in enumerate(ARMS):
for D, off, c in [(TC, -W / 2, C_TC), (TT, +W / 2, C_TT)]:
bp = ax.boxplot([utils(D, a)], positions=[i + off], widths=0.30,
patch_artist=True, showfliers=False,
medianprops=dict(color="black"))
bp["boxes"][0].set(facecolor=c, alpha=.65)
ax.set_xticks(range(len(ARMS)))
ax.set_xticklabels([LABEL[a] for a in ARMS], fontsize=8.5)
ax.set_ylabel("per-worker GPU util %"); ax.set_ylim(0, 100)
ax.set_title("Per-worker GPU util (box = 8 workers; tighter = balanced)", fontsize=10.5)
ax.grid(axis="y", alpha=.3) ax.grid(axis="y", alpha=.3)
ax.set_ylim(0, max(tc + tt) * 1.18)
# mark LPWL-thinktime as the winner (lowest green) in each panel
ax.annotate("LPWL wins\nunder thinktime", xy=(0 + w / 2, tt[0]),
xytext=(0.9, max(tc + tt) * 0.86), fontsize=8.5, color=C_TT,
ha="left", arrowprops=dict(arrowstyle="->", color=C_TT, lw=1.3))
return b1, b2
fig, (axL, axR) = plt.subplots(1, 2, figsize=(11.2, 4.6)) def col(D, key, sub, scale=1.0):
panel(axL, "ttft_ms", "p90", "TTFT p90 (lower = better)", "TTFT p90 (s)") return [D[a][key][sub] * scale for a in ARMS]
panel(axR, "e2e_ms", "mean", "E2E mean (lower = better)", "E2E mean (s)")
axL.legend(loc="upper left", fontsize=9)
fig.suptitle("5-policy routing: dispatch mode flips the ranking — " fig, ax = plt.subplots(2, 3, figsize=(15.5, 8.6))
"LPWL is best under faithful thinktime, only ties/loses under tracets bursts", bar_panel(ax[0, 0], col(TC, "ttft_ms", "mean", 1e-3), col(TT, "ttft_ms", "mean", 1e-3),
fontsize=11.5) "TTFT mean", "s")
fig.tight_layout(rect=(0, 0, 1, 0.95)) bar_panel(ax[0, 1], col(TC, "ttft_ms", "p90", 1e-3), col(TT, "ttft_ms", "p90", 1e-3),
"TTFT p90", "s")
bar_panel(ax[0, 2],
[TC[a]["throughput"]["decode_tps"] for a in ARMS],
[TT[a]["throughput"]["decode_tps"] for a in ARMS],
"Decode TPS (output goodput)", "tok/s", fmt="{:.0f}", higher_better=True)
bar_panel(ax[1, 0], col(TC, "e2e_ms", "mean", 1e-3), col(TT, "e2e_ms", "mean", 1e-3),
"E2E mean", "s")
bar_panel(ax[1, 1], col(TC, "e2e_ms", "p90", 1e-3), col(TT, "e2e_ms", "p90", 1e-3),
"E2E p90", "s")
gpu_panel(ax[1, 2])
fig.legend(handles=[Patch(facecolor=C_TC, label="tracets (burst artifact)"),
Patch(facecolor=C_TT, label="thinktime (faithful load)")],
loc="lower center", ncol=2, fontsize=10.5, bbox_to_anchor=(0.5, 0.0))
fig.suptitle("5-policy routing: tracets vs thinktime (807 reqs, dash0 8xH20) — "
"LPWL wins across the board under faithful thinktime",
fontsize=12.5)
fig.tight_layout(rect=(0, 0.035, 1, 0.96))
out = os.path.join(HERE, "..", "figs", "exp_d_policy_dispatch.png") out = os.path.join(HERE, "..", "figs", "exp_d_policy_dispatch.png")
fig.savefig(out, dpi=140) fig.savefig(out, dpi=140)
print("wrote", os.path.normpath(out)) print("wrote", os.path.normpath(out))
# also print the deltas the README cites
print("\npolicy TTFTp90 tc->tt E2Emean tc->tt")
for a in ARMS:
t1, t2 = TC[a]["ttft_ms"]["p90"], TT[a]["ttft_ms"]["p90"]
e1, e2 = TC[a]["e2e_ms"]["mean"], TT[a]["e2e_ms"]["mean"]
print(f"{a:<13} {t1/1000:5.1f}->{t2/1000:4.1f}s ({(t2-t1)/t1:+.0%}) "
f"{e1/1000:5.1f}->{e2/1000:4.1f}s ({(e2-e1)/e1:+.0%})")

Binary file not shown.

Before

Width:  |  Height:  |  Size: 81 KiB

After

Width:  |  Height:  |  Size: 152 KiB