From 7affb565b29447ba98a839169555bcc4ac8d46ef Mon Sep 17 00:00:00 2001 From: kzlin Date: Wed, 6 May 2026 21:29:56 +0800 Subject: [PATCH] feat(kvc): add backpressure smoke sweep + analyzer (and v6 p1 profile script) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../analysis/analyze_backpressure_smoke.py | 191 ++++++++++++++++++ scripts/sweep_backpressure_smoke.sh | 114 +++++++++++ scripts/sweep_tp1_v6_p1_profile.sh | 129 ++++++++++++ 3 files changed, 434 insertions(+) create mode 100755 scripts/analysis/analyze_backpressure_smoke.py create mode 100755 scripts/sweep_backpressure_smoke.sh create mode 100755 scripts/sweep_tp1_v6_p1_profile.sh diff --git a/scripts/analysis/analyze_backpressure_smoke.py b/scripts/analysis/analyze_backpressure_smoke.py new file mode 100755 index 0000000..6020fb8 --- /dev/null +++ b/scripts/analysis/analyze_backpressure_smoke.py @@ -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() diff --git a/scripts/sweep_backpressure_smoke.sh b/scripts/sweep_backpressure_smoke.sh new file mode 100755 index 0000000..f32fa30 --- /dev/null +++ b/scripts/sweep_backpressure_smoke.sh @@ -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" diff --git a/scripts/sweep_tp1_v6_p1_profile.sh b/scripts/sweep_tp1_v6_p1_profile.sh new file mode 100755 index 0000000..adb4582 --- /dev/null +++ b/scripts/sweep_tp1_v6_p1_profile.sh @@ -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-/ +# │ ├── 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 ==="