docs: v4 final results, error analysis, and updated journey
Add v4 sweep results and post-mortem analysis showing:
- direct-to-D path: 54.3% (1P7D) / 58.0% (2P6D) of requests now use
KVC cleanly. P50=0.5s and TTFT P50=0.043s; this path beats baseline
8DP across the board (P50 -24%, TTFT P50 -54%, TTFT P90 -79%).
- Overall vs baseline (errors+truncated excluded):
v4 2P6D P50=0.85s vs baseline 0.66s (28% slower).
Reason is not errors -- 35% of requests still hit
fallback-large-append-session-cap, where capacity-based
cap = usable_tokens / target_tokens evaluates to 1-2 (not 16)
for large agentic inputs.
- 9-10% errors on KVC variants are mooncake TCP transfer timeouts,
not SGLang logic bugs. Prefill log shows
"Failed to send kv chunk ... 32s timeout ... session not alive".
Errors concentrate in turn>=31 (large inputs) after run >44.8%.
Track:
- docs/KVC_DEBUG_JOURNEY_V1_TO_V4.md: append v4 results table,
per-mode breakdown, and error root cause.
- scripts/analysis/{analyze_v3,analyze_v4,analyze_errors,compare_no_error}.py
- outputs/qwen3-30b-tp1-v{3,4}*/exp*_summary.json (force-added,
small JSON; metrics.jsonl excluded due to size).
- outputs/qwen3-30b-tp1-v{3,4}*/sweep_results.txt
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
kzlin
83
scripts/analysis/analyze_errors.py
Normal file
83
scripts/analysis/analyze_errors.py
Normal file
@@ -0,0 +1,83 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Deep dive into v4 errors: which path, which D, which session, which turn."""
|
||||
import json
|
||||
import numpy as np
|
||||
from pathlib import Path
|
||||
from collections import Counter, defaultdict
|
||||
|
||||
BASE = Path(__file__).parent
|
||||
|
||||
def load_rows(jsonl_path):
|
||||
rows = []
|
||||
with open(jsonl_path) as f:
|
||||
for line in f:
|
||||
rows.append(json.loads(line))
|
||||
return rows
|
||||
|
||||
# Compare v3 and v4 errors
|
||||
for label, path in [
|
||||
("v3 1P7D", BASE.parent / "qwen3-30b-tp1-v3-kvaware/exp1_1p7d_kvc_kvaware_metrics.jsonl"),
|
||||
("v4 1P7D", BASE / "exp1_1p7d_kvc_cap16_metrics.jsonl"),
|
||||
("v3 2P6D", BASE.parent / "qwen3-30b-tp1-v3-kvaware/exp2_2p6d_kvc_kvaware_metrics.jsonl"),
|
||||
("v4 2P6D", BASE / "exp2_2p6d_kvc_cap16_metrics.jsonl"),
|
||||
]:
|
||||
if not path.exists():
|
||||
print(f"\nSKIP {label}: {path} not found")
|
||||
continue
|
||||
rows = load_rows(path)
|
||||
err = [r for r in rows if r.get("error") is not None]
|
||||
print(f"\n========== {label} ({len(err)} errors / {len(rows)} total = {len(err)/len(rows)*100:.1f}%) ==========")
|
||||
|
||||
# Error finish_reason distribution
|
||||
fr_counter = Counter()
|
||||
for r in err:
|
||||
fr = str(r.get("finish_reason") or r.get("error") or "?")
|
||||
fr_counter[fr[:80]] += 1
|
||||
print(f"finish_reason distribution:")
|
||||
for fr, cnt in fr_counter.most_common():
|
||||
print(f" {cnt:>4}x {fr}")
|
||||
|
||||
# Errors by execution mode (these are aborted before mode assignment usually)
|
||||
mode_counter = Counter(r.get("execution_mode", "?") for r in err)
|
||||
print(f"\nerror by execution_mode:")
|
||||
for mode, cnt in mode_counter.most_common():
|
||||
print(f" {cnt:>4}x {mode}")
|
||||
|
||||
# Errors per D worker
|
||||
dw_counter = Counter(r.get("assigned_decode_node", "?") for r in err)
|
||||
print(f"\nerror per assigned_decode_node:")
|
||||
for dw, cnt in dw_counter.most_common():
|
||||
print(f" {cnt:>4}x {dw}")
|
||||
|
||||
# Errors by turn distribution
|
||||
turn_counter = Counter(r.get("turn_id", -1) for r in err)
|
||||
early = sum(c for t, c in turn_counter.items() if t <= 5)
|
||||
mid = sum(c for t, c in turn_counter.items() if 5 < t <= 30)
|
||||
late = sum(c for t, c in turn_counter.items() if t > 30)
|
||||
print(f"\nerror by turn: early(0-5)={early} mid(6-30)={mid} late(31+)={late}")
|
||||
|
||||
# Per-session error rate
|
||||
per_sess_err = defaultdict(int)
|
||||
per_sess_total = defaultdict(int)
|
||||
for r in rows:
|
||||
per_sess_total[r["session_id"]] += 1
|
||||
if r.get("error") is not None:
|
||||
per_sess_err[r["session_id"]] += 1
|
||||
sess_with_err = [(sid, per_sess_err[sid], per_sess_total[sid]) for sid in per_sess_err]
|
||||
sess_with_err.sort(key=lambda x: -x[1])
|
||||
print(f"\ntop 5 sessions by error count:")
|
||||
for sid, e, t in sess_with_err[:5]:
|
||||
print(f" session {sid}: {e}/{t} errors ({e/t*100:.0f}%)")
|
||||
|
||||
# Errors timeline: are they bursty?
|
||||
err_ts = sorted([r.get("trace_timestamp_s", 0) for r in err])
|
||||
if err_ts:
|
||||
first_ts = err_ts[0]
|
||||
last_ts = err_ts[-1]
|
||||
all_ts = sorted([r.get("trace_timestamp_s", 0) for r in rows])
|
||||
first_all = all_ts[0]
|
||||
last_all = all_ts[-1]
|
||||
run_duration = last_all - first_all
|
||||
err_first_pct = (err_ts[0] - first_all) / run_duration * 100 if run_duration > 0 else 0
|
||||
err_last_pct = (err_ts[-1] - first_all) / run_duration * 100 if run_duration > 0 else 0
|
||||
print(f"\nerror time range (% of run): {err_first_pct:.1f}% - {err_last_pct:.1f}%")
|
||||
89
scripts/analysis/analyze_v3.py
Normal file
89
scripts/analysis/analyze_v3.py
Normal file
@@ -0,0 +1,89 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Analyze v3 (kv-aware) results — find why fallback-large-append-session-cap dominates."""
|
||||
import json
|
||||
import numpy as np
|
||||
from pathlib import Path
|
||||
from collections import Counter, defaultdict
|
||||
|
||||
BASE = Path(__file__).parent
|
||||
|
||||
def load_rows(jsonl_path):
|
||||
rows = []
|
||||
with open(jsonl_path) as f:
|
||||
for line in f:
|
||||
rows.append(json.loads(line))
|
||||
return rows
|
||||
|
||||
exp1 = load_rows(BASE / "exp1_1p7d_kvc_kvaware_metrics.jsonl")
|
||||
exp2 = load_rows(BASE / "exp2_2p6d_kvc_kvaware_metrics.jsonl")
|
||||
|
||||
for name, rows in [("Exp1 1P7D", exp1), ("Exp2 2P6D", exp2)]:
|
||||
print(f"\n========== {name} ==========")
|
||||
ok = [r for r in rows if r.get("error") is None]
|
||||
|
||||
# Execution mode breakdown by latency
|
||||
modes = Counter(r["execution_mode"] for r in ok)
|
||||
print(f"\nExecution modes (n={len(ok)}):")
|
||||
for mode, count in modes.most_common():
|
||||
mode_rows = [r for r in ok if r["execution_mode"] == mode]
|
||||
lats = [r["latency_s"] for r in mode_rows]
|
||||
ttfts = [r["ttft_s"] for r in mode_rows]
|
||||
print(f" {mode}: n={count} ({count/len(ok)*100:.1f}%) "
|
||||
f"lat P50={np.percentile(lats,50):.3f}s P90={np.percentile(lats,90):.3f}s | "
|
||||
f"ttft P50={np.percentile(ttfts,50):.3f}s P90={np.percentile(ttfts,90):.3f}s")
|
||||
|
||||
# Per-D session distribution
|
||||
per_d_sessions = defaultdict(set)
|
||||
for r in ok:
|
||||
d = r.get("assigned_decode_node", "?")
|
||||
per_d_sessions[d].add(r["session_id"])
|
||||
print(f"\nSessions per D worker:")
|
||||
for d in sorted(per_d_sessions.keys()):
|
||||
print(f" {d}: {len(per_d_sessions[d])} unique sessions")
|
||||
|
||||
# session-cap fallback analysis
|
||||
sc_rows = [r for r in ok if r["execution_mode"] == "pd-router-fallback-large-append-session-cap"]
|
||||
if sc_rows:
|
||||
print(f"\nSession-cap fallback details (n={len(sc_rows)}):")
|
||||
# Which sessions hit this most?
|
||||
sc_per_sess = Counter(r["session_id"] for r in sc_rows)
|
||||
print(f" Sessions hitting session-cap (top 5):")
|
||||
for sid, cnt in sc_per_sess.most_common(5):
|
||||
print(f" session {sid}: {cnt} times")
|
||||
# Per-D distribution
|
||||
sc_per_d = Counter(r.get("assigned_decode_node", "?") for r in sc_rows)
|
||||
print(f" Per-D distribution: {dict(sc_per_d.most_common())}")
|
||||
# Input length distribution
|
||||
inp = [r.get("input_length", 0) for r in sc_rows]
|
||||
print(f" Input length: P50={np.percentile(inp,50):.0f} P90={np.percentile(inp,90):.0f}")
|
||||
# Turn distribution
|
||||
turns = Counter(r.get("turn_id", -1) for r in sc_rows)
|
||||
print(f" Turn distribution (top 5): {dict(turns.most_common(5))}")
|
||||
|
||||
# Direct-to-D analysis (ideal path)
|
||||
dd_rows = [r for r in ok if r["execution_mode"] == "kvcache-direct-to-d-session"]
|
||||
if dd_rows:
|
||||
lats = [r["latency_s"] for r in dd_rows]
|
||||
ttfts = [r["ttft_s"] for r in dd_rows]
|
||||
kv_blocks = [r.get("actual_kv_transfer_blocks", 0) for r in dd_rows]
|
||||
cached = [r.get("cached_tokens", 0) for r in dd_rows]
|
||||
print(f"\nDirect-to-D details (n={len(dd_rows)}):")
|
||||
print(f" lat P50={np.percentile(lats,50):.3f}s P90={np.percentile(lats,90):.3f}s P99={np.percentile(lats,99):.3f}s")
|
||||
print(f" ttft P50={np.percentile(ttfts,50):.3f}s P90={np.percentile(ttfts,90):.3f}s")
|
||||
print(f" KV transfer: P50={np.percentile(kv_blocks,50):.0f} (should be 0 — no P involved)")
|
||||
print(f" cached_tokens P50={np.percentile(cached,50):.0f}")
|
||||
|
||||
# Sessions: how many turns each, how many used direct-to-d
|
||||
print(f"\nPer-session direct-to-D rate (top 10 by total turns):")
|
||||
per_sess = defaultdict(list)
|
||||
for r in ok:
|
||||
per_sess[r["session_id"]].append(r)
|
||||
sess_stats = []
|
||||
for sid, sreqs in per_sess.items():
|
||||
total = len(sreqs)
|
||||
dd = sum(1 for r in sreqs if r["execution_mode"] == "kvcache-direct-to-d-session")
|
||||
sc = sum(1 for r in sreqs if "session-cap" in r["execution_mode"])
|
||||
sess_stats.append((sid, total, dd, sc))
|
||||
sess_stats.sort(key=lambda x: -x[1])
|
||||
for sid, total, dd, sc in sess_stats[:10]:
|
||||
print(f" session {sid}: {total} turns, {dd} direct-to-D ({dd/total*100:.0f}%), {sc} session-cap fallback ({sc/total*100:.0f}%)")
|
||||
52
scripts/analysis/analyze_v4.py
Normal file
52
scripts/analysis/analyze_v4.py
Normal file
@@ -0,0 +1,52 @@
|
||||
#!/usr/bin/env python3
|
||||
"""V4 results analysis: errors, execution modes, latency by mode."""
|
||||
import json
|
||||
import numpy as np
|
||||
from pathlib import Path
|
||||
from collections import Counter
|
||||
|
||||
BASE = Path(__file__).parent
|
||||
|
||||
def load_rows(jsonl_path):
|
||||
rows = []
|
||||
with open(jsonl_path) as f:
|
||||
for line in f:
|
||||
rows.append(json.loads(line))
|
||||
return rows
|
||||
|
||||
for name, path in [
|
||||
("Exp1 1P7D cap=16", BASE / "exp1_1p7d_kvc_cap16_metrics.jsonl"),
|
||||
("Exp2 2P6D cap=16", BASE / "exp2_2p6d_kvc_cap16_metrics.jsonl"),
|
||||
]:
|
||||
rows = load_rows(path)
|
||||
print(f"\n========== {name} ==========")
|
||||
ok = [r for r in rows if r.get("error") is None]
|
||||
err = [r for r in rows if r.get("error") is not None]
|
||||
print(f"Total: {len(rows)}, OK: {len(ok)}, Errors: {len(err)}")
|
||||
|
||||
# Errors finish_reason
|
||||
if err:
|
||||
finish_reasons = Counter()
|
||||
for r in err:
|
||||
fr = str(r.get("finish_reason") or r.get("error") or "?")
|
||||
# Truncate long messages
|
||||
short = fr[:120]
|
||||
finish_reasons[short] += 1
|
||||
print(f"\nError finish_reasons (top 5):")
|
||||
for fr, cnt in finish_reasons.most_common(5):
|
||||
print(f" {cnt}x: {fr}")
|
||||
|
||||
# Execution mode latency breakdown
|
||||
modes = Counter(r["execution_mode"] for r in ok)
|
||||
print(f"\nTop execution modes by latency:")
|
||||
print(f"{'mode':<55}{'n':<8}{'%':<8}{'P50 lat':<10}{'P90 lat':<10}{'TTFT P50':<10}")
|
||||
for mode, count in modes.most_common(8):
|
||||
mode_rows = [r for r in ok if r["execution_mode"] == mode]
|
||||
lats = [r["latency_s"] for r in mode_rows]
|
||||
ttfts = [r["ttft_s"] for r in mode_rows]
|
||||
print(f" {mode:<53}{count:<8}{count/len(ok)*100:>5.1f}% {np.percentile(lats,50):>7.3f}s {np.percentile(lats,90):>7.3f}s {np.percentile(ttfts,50):>7.3f}s")
|
||||
|
||||
# Per-D load
|
||||
per_d = Counter(r.get("assigned_decode_node", "?") for r in ok)
|
||||
print(f"\nPer-D load: max/min ratio = {max(per_d.values())/max(min(per_d.values()),1):.2f}x")
|
||||
print(f" {dict(per_d.most_common())}")
|
||||
136
scripts/analysis/compare_no_error.py
Normal file
136
scripts/analysis/compare_no_error.py
Normal file
@@ -0,0 +1,136 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Compare KVC variants vs baseline, EXCLUDING errors and truncated requests."""
|
||||
import json
|
||||
import numpy as np
|
||||
from pathlib import Path
|
||||
|
||||
OUT = Path("/mnt/kzlin/workflow/pd-hybrid/agentic-pd-hybrid/outputs")
|
||||
|
||||
DATASETS = [
|
||||
("baseline 8DP", OUT / "qwen3-30b-tp1-v2-fixed/exp1_8way_dp_cache_aware_metrics.jsonl"),
|
||||
("v3 1P7D", OUT / "qwen3-30b-tp1-v3-kvaware/exp1_1p7d_kvc_kvaware_metrics.jsonl"),
|
||||
("v3 2P6D", OUT / "qwen3-30b-tp1-v3-kvaware/exp2_2p6d_kvc_kvaware_metrics.jsonl"),
|
||||
("v4 1P7D", OUT / "qwen3-30b-tp1-v4-cap16/exp1_1p7d_kvc_cap16_metrics.jsonl"),
|
||||
("v4 2P6D", OUT / "qwen3-30b-tp1-v4-cap16/exp2_2p6d_kvc_cap16_metrics.jsonl"),
|
||||
]
|
||||
|
||||
def load_rows(path):
|
||||
rows = []
|
||||
with open(path) as f:
|
||||
for line in f:
|
||||
rows.append(json.loads(line))
|
||||
return rows
|
||||
|
||||
def is_truncated(row):
|
||||
a = row.get("actual_output_tokens")
|
||||
r = row.get("requested_output_tokens")
|
||||
if a is not None and r is not None and r > 1:
|
||||
return a < r * 0.5
|
||||
return False
|
||||
|
||||
def stats(values):
|
||||
if not values:
|
||||
return {"n": 0}
|
||||
a = np.array(values)
|
||||
return {
|
||||
"n": len(a),
|
||||
"mean": float(np.mean(a)),
|
||||
"p50": float(np.percentile(a, 50)),
|
||||
"p90": float(np.percentile(a, 90)),
|
||||
"p99": float(np.percentile(a, 99)),
|
||||
}
|
||||
|
||||
def fmt(s, key):
|
||||
if s["n"] == 0:
|
||||
return "N/A"
|
||||
v = s[key]
|
||||
return f"{v:.3f}s" if v < 100 else f"{v:.1f}s"
|
||||
|
||||
results = []
|
||||
for label, path in DATASETS:
|
||||
if not path.exists():
|
||||
print(f"SKIP {label}")
|
||||
continue
|
||||
rows = load_rows(path)
|
||||
total = len(rows)
|
||||
err_n = sum(1 for r in rows if r.get("error") is not None)
|
||||
trunc_n = sum(1 for r in rows if r.get("error") is None and is_truncated(r))
|
||||
|
||||
# Filter: error=None AND not truncated AND latency present
|
||||
clean = [r for r in rows
|
||||
if r.get("error") is None
|
||||
and not is_truncated(r)
|
||||
and r.get("latency_s") is not None]
|
||||
|
||||
lats = [r["latency_s"] for r in clean]
|
||||
ttfts = [r["ttft_s"] for r in clean if r.get("ttft_s") is not None]
|
||||
|
||||
results.append({
|
||||
"label": label,
|
||||
"total": total,
|
||||
"err": err_n,
|
||||
"trunc": trunc_n,
|
||||
"clean_n": len(clean),
|
||||
"lat": stats(lats),
|
||||
"ttft": stats(ttfts),
|
||||
})
|
||||
|
||||
# Print comparison table
|
||||
print(f"\n{'='*100}")
|
||||
print("LATENCY (excluding errors AND truncated)")
|
||||
print(f"{'='*100}")
|
||||
print(f"{'config':<16}{'total':>7}{'err':>6}{'trunc':>7}{'clean':>7} {'mean':>9}{'P50':>9}{'P90':>9}{'P99':>9}")
|
||||
for r in results:
|
||||
print(f"{r['label']:<16}{r['total']:>7}{r['err']:>6}{r['trunc']:>7}{r['clean_n']:>7} "
|
||||
f"{fmt(r['lat'],'mean'):>9}{fmt(r['lat'],'p50'):>9}{fmt(r['lat'],'p90'):>9}{fmt(r['lat'],'p99'):>9}")
|
||||
|
||||
print(f"\n{'='*100}")
|
||||
print("TTFT (excluding errors AND truncated)")
|
||||
print(f"{'='*100}")
|
||||
print(f"{'config':<16}{'clean':>7} {'mean':>9}{'P50':>9}{'P90':>9}{'P99':>9}")
|
||||
for r in results:
|
||||
print(f"{r['label']:<16}{r['clean_n']:>7} "
|
||||
f"{fmt(r['ttft'],'mean'):>9}{fmt(r['ttft'],'p50'):>9}{fmt(r['ttft'],'p90'):>9}{fmt(r['ttft'],'p99'):>9}")
|
||||
|
||||
# Also: per-execution-mode breakdown for v4 only (the most interesting)
|
||||
print(f"\n{'='*100}")
|
||||
print("V4 2P6D: per-execution-mode (excluding errors and truncated)")
|
||||
print(f"{'='*100}")
|
||||
v4_2p6d = next((p for l, p in DATASETS if l == "v4 2P6D"), None)
|
||||
if v4_2p6d:
|
||||
rows = load_rows(v4_2p6d)
|
||||
clean = [r for r in rows if r.get("error") is None and not is_truncated(r)]
|
||||
from collections import Counter
|
||||
modes = Counter(r["execution_mode"] for r in clean)
|
||||
print(f"{'mode':<55}{'n':>7}{'%':>7} {'mean':>9}{'P50':>9}{'P90':>9}{'P99':>9}")
|
||||
for mode, count in modes.most_common(10):
|
||||
m_rows = [r for r in clean if r["execution_mode"] == mode]
|
||||
s = stats([r["latency_s"] for r in m_rows])
|
||||
pct = count/len(clean)*100
|
||||
print(f" {mode:<53}{count:>7}{pct:>6.1f}% {fmt(s,'mean'):>9}{fmt(s,'p50'):>9}{fmt(s,'p90'):>9}{fmt(s,'p99'):>9}")
|
||||
|
||||
# Also: WHAT IF we only count direct-to-D? (Pure KVC performance)
|
||||
print(f"\n{'='*100}")
|
||||
print("Pure KVC (kvcache-direct-to-d-session ONLY) vs Baseline")
|
||||
print(f"{'='*100}")
|
||||
for label, path in DATASETS:
|
||||
if not path.exists() or "1P7D" not in label and "2P6D" not in label:
|
||||
continue
|
||||
rows = load_rows(path)
|
||||
direct = [r for r in rows
|
||||
if r.get("error") is None and not is_truncated(r)
|
||||
and r.get("execution_mode") == "kvcache-direct-to-d-session"]
|
||||
if not direct:
|
||||
continue
|
||||
s_lat = stats([r["latency_s"] for r in direct])
|
||||
s_ttft = stats([r["ttft_s"] for r in direct if r.get("ttft_s") is not None])
|
||||
print(f"{label:<16}n={s_lat['n']:>5} lat: P50={fmt(s_lat,'p50')} P90={fmt(s_lat,'p90')} ttft: P50={fmt(s_ttft,'p50')} P90={fmt(s_ttft,'p90')}")
|
||||
|
||||
# Baseline for reference (already non-fallback by definition)
|
||||
print()
|
||||
baseline_path = OUT / "qwen3-30b-tp1-v2-fixed/exp1_8way_dp_cache_aware_metrics.jsonl"
|
||||
baseline_rows = load_rows(baseline_path)
|
||||
clean = [r for r in baseline_rows if r.get("error") is None and not is_truncated(r)]
|
||||
s_lat = stats([r["latency_s"] for r in clean])
|
||||
s_ttft = stats([r["ttft_s"] for r in clean if r.get("ttft_s") is not None])
|
||||
print(f"{'baseline 8DP':<16}n={s_lat['n']:>5} lat: P50={fmt(s_lat,'p50')} P90={fmt(s_lat,'p90')} ttft: P50={fmt(s_ttft,'p50')} P90={fmt(s_ttft,'p90')}")
|
||||
Reference in New Issue
Block a user