feat(kvc): add backpressure smoke sweep + analyzer (and v6 p1 profile script)
scripts/sweep_backpressure_smoke.sh: 4-run smoke matrix (KVC baseline / KVC + backpressure / KVC + backpressure @ time-scale=1 / DP @ time-scale=1) designed to fit ~3-4h GPU budget. Validates §3 backpressure implementation and partially probes §7 time-scale distortion. scripts/analysis/analyze_backpressure_smoke.py: consumes the new structural/* jsonl files plus request-metrics; emits headline metrics, backpressure histograms, admission probe stats, and per-session pinning distribution. scripts/sweep_tp1_v6_p1_profile.sh: pre-existing v6 P1 profile sweep script (was untracked; included for completeness). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
kzlin
191
scripts/analysis/analyze_backpressure_smoke.py
Executable file
191
scripts/analysis/analyze_backpressure_smoke.py
Executable file
@@ -0,0 +1,191 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Analyze backpressure smoke sweep outputs.
|
||||
|
||||
For each run dir with a `request-metrics.jsonl` and the new `structural/`
|
||||
subdir (admission-events.jsonl, backpressure-events.jsonl,
|
||||
session-d-binding.jsonl), report:
|
||||
|
||||
- Headline (errors, latency, ttft, direct-to-D rate)
|
||||
- Backpressure pause histogram (count, p50/p90 sleep, total pause time per D)
|
||||
- Admission probe stats (RPC count, mean RTT, queue_depth distribution,
|
||||
pause_ms distribution)
|
||||
- Session pinning (distinct D per session, bimodal direct-to-D rate)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import statistics
|
||||
from collections import Counter, defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def load_jsonl(path: Path) -> list[dict]:
|
||||
if not path.exists():
|
||||
return []
|
||||
return [json.loads(l) for l in path.open("r", encoding="utf-8") if l.strip()]
|
||||
|
||||
|
||||
def summarize_run(run_dir: Path) -> dict:
|
||||
metrics_path = next(run_dir.rglob("request-metrics.jsonl"), None)
|
||||
if metrics_path is None:
|
||||
return {"run_dir": str(run_dir), "error": "no request-metrics.jsonl"}
|
||||
|
||||
summary_path = metrics_path.with_suffix(metrics_path.suffix + ".summary.json")
|
||||
summary = (
|
||||
json.load(summary_path.open()) if summary_path.exists() else {}
|
||||
)
|
||||
|
||||
structural_dir = run_dir / "structural"
|
||||
if not structural_dir.exists():
|
||||
# try metrics dir's parent / structural
|
||||
structural_dir = metrics_path.parent / "structural"
|
||||
|
||||
admission_events = load_jsonl(structural_dir / "admission-events.jsonl")
|
||||
backpressure_events = load_jsonl(structural_dir / "backpressure-events.jsonl")
|
||||
binding_events = load_jsonl(structural_dir / "session-d-binding.jsonl")
|
||||
|
||||
out: dict = {"run_dir": str(run_dir)}
|
||||
|
||||
# Headline metrics from summary.json
|
||||
out["request_count"] = summary.get("request_count")
|
||||
out["error_count"] = summary.get("error_count")
|
||||
out["latency"] = summary.get("latency_stats_s")
|
||||
out["ttft"] = summary.get("ttft_stats_s")
|
||||
out["execution_modes"] = summary.get("execution_modes")
|
||||
out["per_decode_load"] = summary.get("per_decode_load")
|
||||
out["per_prefill_load"] = summary.get("per_prefill_load")
|
||||
|
||||
# Direct-to-D rate from execution_modes
|
||||
em = summary.get("execution_modes", {}) or {}
|
||||
direct = em.get("kvcache-direct-to-d-session", 0)
|
||||
total = sum(em.values()) or 1
|
||||
out["direct_to_d_rate"] = direct / total
|
||||
|
||||
# Session pinning
|
||||
bind_per_session: dict[str, set[int]] = defaultdict(set)
|
||||
for ev in binding_events:
|
||||
bind_per_session[ev["session_id"]].add(ev["decode_worker_index"])
|
||||
if bind_per_session:
|
||||
out["session_count"] = len(bind_per_session)
|
||||
out["avg_distinct_d_per_session"] = (
|
||||
sum(len(v) for v in bind_per_session.values()) / len(bind_per_session)
|
||||
)
|
||||
else:
|
||||
out["session_count"] = 0
|
||||
out["avg_distinct_d_per_session"] = None
|
||||
|
||||
# Direct-to-D rate per session (bimodal check)
|
||||
records = load_jsonl(metrics_path)
|
||||
sess_records: dict[str, list[dict]] = defaultdict(list)
|
||||
for r in records:
|
||||
sess_records[r["session_id"]].append(r)
|
||||
rates = []
|
||||
for sid, turns in sess_records.items():
|
||||
ndir = sum(
|
||||
1 for t in turns if t.get("execution_mode") == "kvcache-direct-to-d-session"
|
||||
)
|
||||
rates.append(ndir / len(turns))
|
||||
if rates:
|
||||
buckets = [0, 0, 0, 0, 0]
|
||||
for r in rates:
|
||||
buckets[min(4, int(r * 5))] += 1
|
||||
out["direct_to_d_rate_buckets"] = {
|
||||
"0-20%": buckets[0],
|
||||
"20-40%": buckets[1],
|
||||
"40-60%": buckets[2],
|
||||
"60-80%": buckets[3],
|
||||
"80-100%": buckets[4],
|
||||
}
|
||||
|
||||
# Backpressure events
|
||||
if backpressure_events:
|
||||
sleeps = [ev["sleep_s"] for ev in backpressure_events]
|
||||
out["backpressure"] = {
|
||||
"event_count": len(backpressure_events),
|
||||
"total_sleep_s": round(sum(sleeps), 2),
|
||||
"sleep_p50_s": round(statistics.median(sleeps), 4),
|
||||
"sleep_p90_s": round(
|
||||
sorted(sleeps)[int(len(sleeps) * 0.9)] if sleeps else 0, 4
|
||||
),
|
||||
"events_per_d": dict(
|
||||
Counter(ev["server_url"] for ev in backpressure_events).most_common()
|
||||
),
|
||||
}
|
||||
else:
|
||||
out["backpressure"] = {"event_count": 0, "note": "no backpressure events"}
|
||||
|
||||
# Admission probe stats
|
||||
if admission_events:
|
||||
rtts = [ev["rtt_s"] for ev in admission_events]
|
||||
depths = [ev.get("queue_depth", 0) for ev in admission_events]
|
||||
pauses = [ev.get("recommended_pause_ms", 0) for ev in admission_events]
|
||||
out["admission_probes"] = {
|
||||
"count": len(admission_events),
|
||||
"mean_rtt_s": round(sum(rtts) / len(rtts), 4),
|
||||
"p99_rtt_s": round(sorted(rtts)[int(len(rtts) * 0.99)], 4),
|
||||
"queue_depth_p50": int(statistics.median(depths)),
|
||||
"queue_depth_p90": int(sorted(depths)[int(len(depths) * 0.9)]),
|
||||
"queue_depth_max": max(depths),
|
||||
"pause_ms_p50": int(statistics.median(pauses)),
|
||||
"pause_ms_p90": int(sorted(pauses)[int(len(pauses) * 0.9)]),
|
||||
"pause_ms_max": max(pauses),
|
||||
"nonzero_pause_count": sum(1 for p in pauses if p > 0),
|
||||
"by_reason": dict(
|
||||
Counter(ev.get("reason") or "ok" for ev in admission_events).most_common()
|
||||
),
|
||||
}
|
||||
|
||||
return out
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("sweep_root", type=Path)
|
||||
ap.add_argument("--json", action="store_true", help="emit JSON only")
|
||||
args = ap.parse_args()
|
||||
|
||||
summaries = []
|
||||
for run_dir in sorted(args.sweep_root.iterdir()):
|
||||
if not run_dir.is_dir():
|
||||
continue
|
||||
summary = summarize_run(run_dir)
|
||||
summaries.append(summary)
|
||||
|
||||
if args.json:
|
||||
print(json.dumps(summaries, indent=2))
|
||||
return
|
||||
|
||||
for s in summaries:
|
||||
print(f"\n{'=' * 70}")
|
||||
print(f" {s['run_dir']}")
|
||||
print(f"{'=' * 70}")
|
||||
if "error" in s:
|
||||
print(f" ERROR: {s['error']}")
|
||||
continue
|
||||
print(f" reqs={s.get('request_count')} errors={s.get('error_count')}")
|
||||
if s.get("latency"):
|
||||
lt = s["latency"]
|
||||
print(
|
||||
f" latency: mean={lt.get('mean'):.3f} "
|
||||
f"p50={lt.get('p50'):.3f} p90={lt.get('p90'):.3f} p99={lt.get('p99'):.3f}"
|
||||
)
|
||||
if s.get("ttft"):
|
||||
tt = s["ttft"]
|
||||
print(
|
||||
f" ttft: mean={tt.get('mean'):.3f} "
|
||||
f"p50={tt.get('p50'):.3f} p90={tt.get('p90'):.3f}"
|
||||
)
|
||||
print(f" direct_to_d_rate: {s.get('direct_to_d_rate', 0) * 100:.1f}%")
|
||||
print(f" sessions: {s.get('session_count')} | "
|
||||
f"avg distinct-D-per-session: {s.get('avg_distinct_d_per_session')}")
|
||||
if s.get("direct_to_d_rate_buckets"):
|
||||
print(f" direct-to-D distribution by session: {s['direct_to_d_rate_buckets']}")
|
||||
if s.get("backpressure"):
|
||||
print(f" backpressure: {s['backpressure']}")
|
||||
if s.get("admission_probes"):
|
||||
print(f" admission probes: {s['admission_probes']}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
114
scripts/sweep_backpressure_smoke.sh
Executable file
114
scripts/sweep_backpressure_smoke.sh
Executable file
@@ -0,0 +1,114 @@
|
||||
#!/usr/bin/env bash
|
||||
# Smoke sweep: validate backpressure code change on top of v5 Option D config.
|
||||
# Designed to fit in ~3-4h GPU budget (4 runs × ~30-60 min).
|
||||
#
|
||||
# Usage:
|
||||
# bash scripts/sweep_backpressure_smoke.sh
|
||||
#
|
||||
# Prerequisites: GPUs available; trace at outputs/qwen35-swebench-50sess.jsonl;
|
||||
# model at $MODEL_PATH (default Qwen3-30B-A3B-Instruct-2507).
|
||||
set -euo pipefail
|
||||
|
||||
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
cd "$REPO_ROOT"
|
||||
|
||||
OUT_ROOT=${OUT_ROOT:-outputs/sweep_backpressure_smoke}
|
||||
TRACE=${TRACE:-outputs/qwen35-swebench-50sess.jsonl}
|
||||
MODEL=${MODEL:-/mnt/kzlin/workflow/pd-hybrid/simm-swe-bench/models/Qwen3-30B-A3B-Instruct-2507}
|
||||
|
||||
mkdir -p "$OUT_ROOT"
|
||||
LOG="$OUT_ROOT/sweep.log"
|
||||
echo "[$(date '+%F %T')] Starting backpressure smoke sweep" | tee -a "$LOG"
|
||||
echo " Trace: $TRACE" | tee -a "$LOG"
|
||||
echo " Model: $MODEL" | tee -a "$LOG"
|
||||
echo " Output root: $OUT_ROOT" | tee -a "$LOG"
|
||||
|
||||
KVC_COMMON_ARGS=(
|
||||
--trace "$TRACE"
|
||||
--model "$MODEL"
|
||||
--mechanism kvcache-centric
|
||||
--policy kv-aware
|
||||
--kvcache-admission-mode worker
|
||||
--kvcache-seed-min-turn-id 1
|
||||
--kvcache-seed-max-inflight-decode -1
|
||||
--kvcache-prefill-backup-policy release-after-transfer
|
||||
--kvcache-prefill-priority-eviction
|
||||
--prefill-workers 2
|
||||
--decode-workers 6
|
||||
--prefill-gpu-ids 0,1
|
||||
--decode-gpu-ids 2,3,4,5,6,7
|
||||
--transfer-backend mooncake
|
||||
--target-duration-s 2000
|
||||
--session-sample-rate 1.0
|
||||
--min-turns 2
|
||||
--concurrency-limit 32
|
||||
)
|
||||
|
||||
DP_COMMON_ARGS=(
|
||||
--trace "$TRACE"
|
||||
--model "$MODEL"
|
||||
--mechanism pd-colo
|
||||
--policy kv-aware
|
||||
--direct-workers 8
|
||||
--direct-gpu-ids 0,1,2,3,4,5,6,7
|
||||
--transfer-backend mooncake
|
||||
--target-duration-s 2000
|
||||
--session-sample-rate 1.0
|
||||
--min-turns 2
|
||||
--concurrency-limit 32
|
||||
)
|
||||
|
||||
run_kvc_baseline_ts10() {
|
||||
local out="$OUT_ROOT/E1_kvc_baseline_ts10"
|
||||
echo "[$(date '+%F %T')] === E1: KVC baseline (no backpressure) time-scale=10 ===" | tee -a "$LOG"
|
||||
python -m agentic_pd_hybrid.cli benchmark-live \
|
||||
"${KVC_COMMON_ARGS[@]}" \
|
||||
--output-root "$out" \
|
||||
--time-scale 10 \
|
||||
2>&1 | tee -a "$LOG"
|
||||
}
|
||||
|
||||
run_kvc_backpressure_ts10() {
|
||||
local out="$OUT_ROOT/E2_kvc_backpressure_ts10"
|
||||
echo "[$(date '+%F %T')] === E2: KVC + backpressure ON, time-scale=10 ===" | tee -a "$LOG"
|
||||
python -m agentic_pd_hybrid.cli benchmark-live \
|
||||
"${KVC_COMMON_ARGS[@]}" \
|
||||
--output-root "$out" \
|
||||
--time-scale 10 \
|
||||
--enable-backpressure \
|
||||
--backpressure-max-pause-s 2.0 \
|
||||
2>&1 | tee -a "$LOG"
|
||||
}
|
||||
|
||||
run_kvc_backpressure_ts1() {
|
||||
local out="$OUT_ROOT/E3_kvc_backpressure_ts1_short"
|
||||
echo "[$(date '+%F %T')] === E3: KVC + backpressure ON, time-scale=1, FIRST 1000 reqs ===" | tee -a "$LOG"
|
||||
python -m agentic_pd_hybrid.cli benchmark-live \
|
||||
"${KVC_COMMON_ARGS[@]}" \
|
||||
--output-root "$out" \
|
||||
--time-scale 1 \
|
||||
--enable-backpressure \
|
||||
--backpressure-max-pause-s 2.0 \
|
||||
--target-duration-s 1800 \
|
||||
2>&1 | tee -a "$LOG"
|
||||
}
|
||||
|
||||
run_dp_baseline_ts1() {
|
||||
local out="$OUT_ROOT/E4_dp_ts1_short"
|
||||
echo "[$(date '+%F %T')] === E4: 8-way DP cache-aware, time-scale=1, FIRST 1000 reqs ===" | tee -a "$LOG"
|
||||
python -m agentic_pd_hybrid.cli benchmark-live \
|
||||
"${DP_COMMON_ARGS[@]}" \
|
||||
--output-root "$out" \
|
||||
--time-scale 1 \
|
||||
--target-duration-s 1800 \
|
||||
2>&1 | tee -a "$LOG"
|
||||
}
|
||||
|
||||
# Sequence — add/remove as fits the budget.
|
||||
run_kvc_baseline_ts10
|
||||
run_kvc_backpressure_ts10
|
||||
run_kvc_backpressure_ts1
|
||||
run_dp_baseline_ts1
|
||||
|
||||
echo "[$(date '+%F %T')] === sweep DONE ===" | tee -a "$LOG"
|
||||
echo "Run analysis with: python scripts/analysis/analyze_backpressure_smoke.py $OUT_ROOT" | tee -a "$LOG"
|
||||
129
scripts/sweep_tp1_v6_p1_profile.sh
Executable file
129
scripts/sweep_tp1_v6_p1_profile.sh
Executable file
@@ -0,0 +1,129 @@
|
||||
#!/bin/bash
|
||||
# v6 P1: re-run the v5 (Option D) config with the pool_breakdown instrument
|
||||
# (commit 4978c0d) so d-pool-timeseries.jsonl carries radix_protected /
|
||||
# slot_private / running_batch / {transfer,prealloc,retracted}_queue tokens.
|
||||
#
|
||||
# This is the same config as scripts/sweep_tp1_v5_optD_profile.sh but writes
|
||||
# to a separate output dir, leaving the pre-instrument v5+profile run intact
|
||||
# for before/after comparison.
|
||||
#
|
||||
# Output:
|
||||
# outputs/qwen3-30b-tp1-v6-p1-profile/
|
||||
# ├── kvcache-centric-kv-aware-worker-admission-<ts>/
|
||||
# │ ├── request-metrics.jsonl
|
||||
# │ ├── request-metrics.jsonl.summary.json
|
||||
# │ └── d-pool-timeseries.jsonl ← now with pool_breakdown fields
|
||||
# ├── exp{1,2}_*_metrics.jsonl
|
||||
# └── exp{1,2}_*_pool_timeseries.jsonl
|
||||
set -euo pipefail
|
||||
cd "$(dirname "$0")/.."
|
||||
|
||||
MODEL=/mnt/kzlin/workflow/pd-hybrid/simm-swe-bench/models/Qwen3-30B-A3B-Instruct-2507
|
||||
TRACE=outputs/qwen35-swebench-50sess.jsonl
|
||||
OUTPUT=outputs/qwen3-30b-tp1-v6-p1-profile
|
||||
VENV_PYTHON=.venv/bin/python
|
||||
RESULTS_FILE=$OUTPUT/sweep_results.txt
|
||||
POLL_INTERVAL=1.0
|
||||
|
||||
mkdir -p $OUTPUT
|
||||
|
||||
log() {
|
||||
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" | tee -a $RESULTS_FILE
|
||||
}
|
||||
|
||||
save_result() {
|
||||
local label=$1
|
||||
local run_dir=$2
|
||||
log "=== $label COMPLETED ==="
|
||||
if [ -f "$run_dir/request-metrics.jsonl.summary.json" ]; then
|
||||
log "Summary:"
|
||||
cat "$run_dir/request-metrics.jsonl.summary.json" >> $RESULTS_FILE
|
||||
echo "" >> $RESULTS_FILE
|
||||
cp "$run_dir/request-metrics.jsonl.summary.json" "$OUTPUT/${label}_summary.json"
|
||||
cp "$run_dir/request-metrics.jsonl" "$OUTPUT/${label}_metrics.jsonl"
|
||||
if [ -f "$run_dir/d-pool-timeseries.jsonl" ]; then
|
||||
cp "$run_dir/d-pool-timeseries.jsonl" "$OUTPUT/${label}_pool_timeseries.jsonl"
|
||||
log "Pool timeseries: $(wc -l < $OUTPUT/${label}_pool_timeseries.jsonl) rows"
|
||||
else
|
||||
log "WARNING: no d-pool-timeseries.jsonl produced"
|
||||
fi
|
||||
log "Saved to $OUTPUT/${label}_summary.json + ${label}_metrics.jsonl + ${label}_pool_timeseries.jsonl"
|
||||
else
|
||||
log "WARNING: No summary file found in $run_dir"
|
||||
fi
|
||||
}
|
||||
|
||||
log "Starting v6 P1 sweep (v5 Option D config + ${POLL_INTERVAL}s pool polling + pool_breakdown)"
|
||||
log "Model: $MODEL"
|
||||
log "Trace: $TRACE (4449 requests, 52 sessions)"
|
||||
log "Goal: capture pool_breakdown fields (radix_protected / slot_private / running_batch / queues)"
|
||||
log " to decompose 'other' on the v5 baseline workload"
|
||||
|
||||
########################################
|
||||
# Experiment 1: 1P + 7D KVC kv-aware Option D + profile
|
||||
########################################
|
||||
log ""
|
||||
log "=== [EXP1] 1P7D KVC kv-aware Option D + profile ==="
|
||||
PYTHONPATH=src:third_party/sglang/python \
|
||||
$VENV_PYTHON -m agentic_pd_hybrid.cli benchmark-live \
|
||||
--trace $TRACE \
|
||||
--output-root $OUTPUT \
|
||||
--mechanism kvcache-centric \
|
||||
--policy kv-aware \
|
||||
--model-path $MODEL \
|
||||
--prefill-workers 1 --decode-workers 7 \
|
||||
--prefill-tp-size 1 --decode-tp-size 1 \
|
||||
--prefill-gpu-ids 0 --decode-gpu-ids 1,2,3,4,5,6,7 \
|
||||
--transfer-backend mooncake \
|
||||
--gpu-budget 8 \
|
||||
--time-scale 10 \
|
||||
--session-sample-rate 1.0 \
|
||||
--target-duration-s 100000 \
|
||||
--concurrency-limit 32 \
|
||||
--timeout-s 900 \
|
||||
--request-timeout-s 300 \
|
||||
--kvcache-admission-mode worker \
|
||||
--kvcache-seed-min-turn-id 1 \
|
||||
--kvcache-seed-max-inflight-decode -1 \
|
||||
--kvcache-prefill-backup-policy release-after-transfer \
|
||||
--kvcache-prefill-priority-eviction \
|
||||
--pool-poll-interval-s $POLL_INTERVAL
|
||||
|
||||
EXP1_DIR=$(ls -td $OUTPUT/kvcache-centric-*/ 2>/dev/null | head -1)
|
||||
save_result "exp1_1p7d_kvc_v6_p1" "$EXP1_DIR"
|
||||
|
||||
########################################
|
||||
# Experiment 2: 2P + 6D KVC kv-aware Option D + profile
|
||||
########################################
|
||||
log ""
|
||||
log "=== [EXP2] 2P6D KVC kv-aware Option D + profile ==="
|
||||
PYTHONPATH=src:third_party/sglang/python \
|
||||
$VENV_PYTHON -m agentic_pd_hybrid.cli benchmark-live \
|
||||
--trace $TRACE \
|
||||
--output-root $OUTPUT \
|
||||
--mechanism kvcache-centric \
|
||||
--policy kv-aware \
|
||||
--model-path $MODEL \
|
||||
--prefill-workers 2 --decode-workers 6 \
|
||||
--prefill-tp-size 1 --decode-tp-size 1 \
|
||||
--prefill-gpu-ids 0,1 --decode-gpu-ids 2,3,4,5,6,7 \
|
||||
--transfer-backend mooncake \
|
||||
--gpu-budget 8 \
|
||||
--time-scale 10 \
|
||||
--session-sample-rate 1.0 \
|
||||
--target-duration-s 100000 \
|
||||
--concurrency-limit 32 \
|
||||
--timeout-s 900 \
|
||||
--request-timeout-s 300 \
|
||||
--kvcache-admission-mode worker \
|
||||
--kvcache-seed-min-turn-id 1 \
|
||||
--kvcache-seed-max-inflight-decode -1 \
|
||||
--kvcache-prefill-backup-policy release-after-transfer \
|
||||
--kvcache-prefill-priority-eviction \
|
||||
--pool-poll-interval-s $POLL_INTERVAL
|
||||
|
||||
EXP2_DIR=$(ls -td $OUTPUT/kvcache-centric-*/ 2>/dev/null | head -1)
|
||||
save_result "exp2_2p6d_kvc_v6_p1" "$EXP2_DIR"
|
||||
|
||||
log ""
|
||||
log "=== ALL v6 P1 EXPERIMENTS DONE ==="
|
||||
Reference in New Issue
Block a user