Compare commits

...

2 Commits

Author SHA1 Message Date
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
18 changed files with 1049 additions and 4 deletions

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

Binary file not shown.

After

Width:  |  Height:  |  Size: 114 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 99 KiB

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

@@ -100,11 +100,13 @@ def main():
continue
s = json.load(open(sp))
arm, pg, dg, ports = arm_of(run.name)
lat = s.get("latency_stats_s", {})
ttft = s.get("ttft_stats_s", {})
tpot = s.get("tpot_stats_s", {})
# `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", {})
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

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,184 @@
"""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, axes = plt.subplots(1, 3, figsize=(15, 4.2))
ax1, ax2, ax3 = axes
for arm in ["colo", *PD_ARMS]:
ax1.plot(xs, series(g, xs, arm, "e2e_mean"), **STYLE[arm])
ax1.axhline(10.0, color="red", ls=":", lw=1, label="SLO (mean E2E 10s)")
ax1.set_yscale("log"); ax1.set_xticks(xs); ax1.set_xticklabels(xs, fontsize=7)
ax1.set_xlabel("concurrent sessions N"); ax1.set_ylabel("E2E latency mean (s, log)")
ax1.set_title("(a) mean-E2E vs concurrency"); ax1.legend(fontsize=8); ax1.grid(alpha=.3, which="both")
for arm in ["colo", *PD_ARMS]:
ax2.plot(xs, series(g, xs, arm, "tps"), **STYLE[arm])
ax2.set_xticks(xs); ax2.set_xticklabels(xs, fontsize=7)
ax2.set_xlabel("concurrent sessions N"); ax2.set_ylabel("throughput (tok/s)")
ax2.set_title("(b) TPS: colo scales, PD plateaus/drops"); ax2.legend(fontsize=8); ax2.grid(alpha=.3)
for arm in ["colo", *PD_ARMS]:
ax3.plot(xs, [v * 100 if v is not None else None for v in series(g, xs, arm, "apc")], **STYLE[arm])
ax3.set_xticks(xs); ax3.set_xticklabels(xs, fontsize=7)
ax3.set_xlabel("concurrent sessions N"); ax3.set_ylabel("producer prefix-cache hit-rate (%)")
ax3.set_title("(c) APC vs concurrency"); ax3.legend(fontsize=8); ax3.grid(alpha=.3)
fig.suptitle("Fig 3 — Concurrency axis (in32768/out128, reuse~0.984): sweep N by 8 to the 10s-SLO ceiling",
fontsize=11, 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
]
}