From ce616f46d12f7c99ffe77d1ba292d048a498a24f Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Fri, 22 May 2026 00:13:50 +0800 Subject: [PATCH] Add per-request breakdown profiling, identify KV cache memory bottleneck MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Breakdown profiling at proxy level captures: t_proxy_recv → t_prefill_sent → t_prefill_done → t_decode_sent → t_first_token Key finding: 87.7% of TTFT is spent in kv+decode phase, NOT prefill. Root cause: decode instance KV cache memory saturation (97.1% usage). With 6P+2D config, 2 decode GPUs have only ~56GB total KV cache. Large agentic requests (avg 33.6k tokens) fill this quickly. Small requests (49 tokens, prefill=0.044s) wait 114s for KV cache to be freed by large requests completing decode. vLLM log confirms: Running=0, Waiting=6, KV cache=97.1% GPU is idle but requests queue for KV cache memory, not compute. This is the fundamental bottleneck of single-machine PD separation for long-context agentic workloads: concentrating decode onto fewer GPUs creates a KV cache memory wall. Co-Authored-By: Claude Opus 4.6 (1M context) --- scripts/analyze_breakdown.py | 54 ++++++++++++++++++++++ scripts/cache_aware_proxy.py | 48 +++++++++++++++++--- scripts/profile_fnf.py | 87 ++++++++++++++++++++++++++++++++++++ 3 files changed, 182 insertions(+), 7 deletions(-) create mode 100644 scripts/analyze_breakdown.py create mode 100644 scripts/profile_fnf.py diff --git a/scripts/analyze_breakdown.py b/scripts/analyze_breakdown.py new file mode 100644 index 0000000..ee6f842 --- /dev/null +++ b/scripts/analyze_breakdown.py @@ -0,0 +1,54 @@ +"""Analyze per-request breakdown data from the proxy.""" +import json, statistics, sys + +url = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:9090/breakdown" + +if url.startswith("http"): + import urllib.request + data = json.loads(urllib.request.urlopen(url, timeout=10).read()) +else: + data = json.load(open(url)) + +print("Total records: %d" % len(data)) + +results = [] +for d in data: + keys = ["t_proxy_recv", "t_prefill_sent", "t_prefill_done", "t_decode_sent", "t_first_token"] + if not all(k in d for k in keys): + continue + results.append({ + "input": d["input_length"], + "prefill": d["t_prefill_done"] - d["t_prefill_sent"], + "proxy_gap": d["t_decode_sent"] - d["t_prefill_done"], + "kv_decode": d["t_first_token"] - d["t_decode_sent"], + "ttft": d["t_first_token"] - d["t_proxy_recv"], + }) + +results.sort(key=lambda x: x["input"]) +print("Complete breakdown: %d" % len(results)) + +if not results: + print("No complete records yet") + sys.exit(0) + +print() +print(" %8s %9s %9s %9s %9s" % ("input", "prefill", "proxy", "kv+dec", "TTFT")) +print(" %8s %9s %9s %9s %9s" % ("-----", "-------", "-----", "------", "----")) +for r in results[:25]: + print(" %8d %9.3f %9.3f %9.3f %9.3f" % ( + r["input"], r["prefill"], r["proxy_gap"], r["kv_decode"], r["ttft"])) + +print() +for key in ["prefill", "proxy_gap", "kv_decode", "ttft"]: + vals = sorted([r[key] for r in results]) + p = lambda q: vals[min(int(q * len(vals)), len(vals) - 1)] + print(" %s: p50=%.3fs p90=%.3fs mean=%.3fs" % ( + key, p(.5), p(.9), statistics.fmean(vals))) + +# Fraction of TTFT by stage +print() +print(" TTFT breakdown (fraction of total):") +for key in ["prefill", "proxy_gap", "kv_decode"]: + fracs = [r[key] / r["ttft"] * 100 for r in results if r["ttft"] > 0.01] + if fracs: + print(" %s: mean=%.1f%% of TTFT" % (key, statistics.fmean(fracs))) diff --git a/scripts/cache_aware_proxy.py b/scripts/cache_aware_proxy.py index af482e7..b78d6eb 100644 --- a/scripts/cache_aware_proxy.py +++ b/scripts/cache_aware_proxy.py @@ -196,25 +196,42 @@ async def _handle_combined(api, req_data, token_ids, input_length, session_id, h return StreamingResponse(generate(), media_type="text/event-stream") -async def _send_prefill_async(p_inst, api, prefill_data, p_headers, token_ids, input_length): +import time as _time + +# Per-request breakdown log (append-only) +_breakdown_log: list[dict] = [] + + +async def _send_prefill_async(p_inst, api, prefill_data, p_headers, token_ids, + input_length, breakdown): """Fire-and-forget prefill: send and don't block caller.""" try: resp = await p_inst.client.post(api, json=prefill_data, headers=p_headers) + breakdown["t_prefill_done"] = _time.monotonic() resp.raise_for_status() await resp.aclose() p_inst.record_prefix(token_ids) except Exception: - pass + breakdown["t_prefill_done"] = _time.monotonic() + breakdown["prefill_error"] = True finally: p_inst.ongoing_tokens -= input_length async def _handle_pd_sep(api, req_data, request_id, token_ids, input_length, session_id, headers): - """PD-Sep mode. --fire-and-forget controls prefill waiting behavior.""" + """PD-Sep mode with per-stage breakdown profiling.""" + breakdown = { + "request_id": request_id, + "input_length": input_length, + "t_proxy_recv": _time.monotonic(), + } + p_inst, _ = pick_instance(prefill_instances, token_ids, session_id, input_length, session_affinity) d_inst = min(decode_instances, key=lambda x: x.ongoing_tokens) + breakdown["p_inst"] = p_inst.url + breakdown["d_inst"] = d_inst.url prefill_data = req_data.copy() prefill_data["kv_transfer_params"] = { @@ -228,24 +245,27 @@ async def _handle_pd_sep(api, req_data, request_id, token_ids, input_length, p_headers = {**headers, "X-data-parallel-rank": "0"} p_inst.ongoing_tokens += input_length + breakdown["t_prefill_sent"] = _time.monotonic() if global_args.fire_and_forget: - # Fire-and-forget: send prefill async, immediately proceed to decode asyncio.create_task(_send_prefill_async( - p_inst, api, prefill_data, p_headers, token_ids, input_length)) + p_inst, api, prefill_data, p_headers, token_ids, input_length, breakdown)) else: - # Await: block until prefill completes, then send decode try: resp = await p_inst.client.post(api, json=prefill_data, headers=p_headers) + breakdown["t_prefill_done"] = _time.monotonic() resp.raise_for_status() await resp.aclose() p_inst.record_prefix(token_ids) except Exception as e: + breakdown["t_prefill_done"] = _time.monotonic() + breakdown["prefill_error"] = True + _breakdown_log.append(breakdown) raise HTTPException(status_code=502, detail=f"Prefill failed: {e}") finally: p_inst.ongoing_tokens -= input_length - # Stream decode + # Send decode d_inst.ongoing_tokens += input_length parsed = urllib.parse.urlparse(str(p_inst.client.base_url)) bootstrap_addr = f"http://{parsed.hostname}:{p_inst.bootstrap_port}" @@ -258,18 +278,32 @@ async def _handle_pd_sep(api, req_data, request_id, token_ids, input_length, "transfer_id": f"xfer-{request_id}", } + breakdown["t_decode_sent"] = _time.monotonic() + async def generate(): + first_token = True try: async with d_inst.client.stream("POST", api, json=decode_data, headers=headers) as resp: resp.raise_for_status() async for chunk in resp.aiter_bytes(): + if first_token: + breakdown["t_first_token"] = _time.monotonic() + first_token = False yield chunk finally: + breakdown["t_done"] = _time.monotonic() d_inst.ongoing_tokens -= input_length + _breakdown_log.append(breakdown) return StreamingResponse(generate(), media_type="application/json") +@app.get("/breakdown") +async def get_breakdown(): + """Return per-request breakdown data for analysis.""" + return _breakdown_log + + def parse_args(): p = argparse.ArgumentParser(description="Unified cache-aware global scheduler") p.add_argument("--port", type=int, default=8000) diff --git a/scripts/profile_fnf.py b/scripts/profile_fnf.py new file mode 100644 index 0000000..0ceb6e3 --- /dev/null +++ b/scripts/profile_fnf.py @@ -0,0 +1,87 @@ +"""Deep profile: why fire-and-forget TTFT is 5x worse than await.""" +import json, statistics + +await_rows = [json.loads(l) for l in open("outputs/gpu_ab_6p2d/metrics.jsonl")] +fnf_rows = [json.loads(l) for l in open("outputs/gpu_ab_6p2d_fnf/metrics.jsonl")] + +await_ok = [r for r in await_rows if not r.get("error")] +fnf_ok = [r for r in fnf_rows if not r.get("error")] + +# Match by request_id +await_by_id = {r["request_id"]: r for r in await_ok} +fnf_by_id = {r["request_id"]: r for r in fnf_ok} +common = set(await_by_id.keys()) & set(fnf_by_id.keys()) + +print("=" * 75) +print(" PROFILE: Fire-and-Forget vs Await-Prefill (same 6P+2D instances)") +print("=" * 75) +print(f" Common requests: {len(common)}") + +# Per-request comparison +diffs = [] +for rid in common: + a = await_by_id[rid] + f = fnf_by_id[rid] + if a.get("ttft_s") and f.get("ttft_s") and a["ttft_s"] > 0: + diffs.append({ + "id": rid, "input": a["input_length"], + "a_ttft": a["ttft_s"], "f_ttft": f["ttft_s"], + "ratio": f["ttft_s"] / a["ttft_s"], + "a_e2e": a["latency_s"], "f_e2e": f["latency_s"], + "a_tpot": a.get("tpot_s", 0), "f_tpot": f.get("tpot_s", 0), + "a_out": a.get("actual_output_tokens", 0) or 0, + "f_out": f.get("actual_output_tokens", 0) or 0, + }) + +diffs.sort(key=lambda x: x["input"]) + +print("\n Per-request (sorted by input_length):") +hdr = "%8s %10s %10s %7s %10s %10s %8s %8s" % ( + "input", "await_TTFT", "fnf_TTFT", "ratio", "await_E2E", "fnf_E2E", "a_TPOT", "f_TPOT") +print(" " + hdr) +print(" " + "-" * len(hdr)) +for d in diffs[:25]: + print(" %8d %10.3f %10.3f %6.1fx %10.3f %10.3f %8.4f %8.4f" % ( + d["input"], d["a_ttft"], d["f_ttft"], d["ratio"], + d["a_e2e"], d["f_e2e"], d["a_tpot"], d["f_tpot"])) + +# Statistics +if diffs: + ratios = [d["ratio"] for d in diffs] + ratios.sort() + p = lambda v, q: v[min(int(q*len(v)), len(v)-1)] + print("\n TTFT ratio (FnF / Await):") + print(" p10=%.2fx p50=%.2fx p90=%.2fx mean=%.2fx" % ( + p(ratios,.1), p(ratios,.5), p(ratios,.9), statistics.fmean(ratios))) + + faster = sum(1 for r in ratios if r < 1.0) + print(" FnF faster: %d/%d (%.0f%%)" % (faster, len(ratios), faster*100/len(ratios))) + + # Bucket by input size + print("\n TTFT ratio by input size bucket:") + buckets = [(0, 5000, "<5k"), (5000, 20000, "5-20k"), (20000, 50000, "20-50k"), (50000, 999999, ">50k")] + for lo, hi, label in buckets: + subset = [d for d in diffs if lo <= d["input"] < hi] + if subset: + rs = [d["ratio"] for d in subset] + a_ttfts = [d["a_ttft"] for d in subset] + f_ttfts = [d["f_ttft"] for d in subset] + print(" %6s: n=%3d await_TTFT=%.3f fnf_TTFT=%.3f ratio=%.2fx" % ( + label, len(subset), statistics.fmean(a_ttfts), statistics.fmean(f_ttfts), + statistics.fmean(rs))) + + # TPOT comparison + a_tpots = [d["a_tpot"] for d in diffs if d["a_tpot"] > 0] + f_tpots = [d["f_tpot"] for d in diffs if d["f_tpot"] > 0] + if a_tpots and f_tpots: + print("\n TPOT comparison:") + print(" Await: mean=%.4f p50=%.4f" % (statistics.fmean(a_tpots), sorted(a_tpots)[len(a_tpots)//2])) + print(" FnF: mean=%.4f p50=%.4f" % (statistics.fmean(f_tpots), sorted(f_tpots)[len(f_tpots)//2])) + +# Also look at non-common requests (FnF only failures) +fnf_err = [r for r in fnf_rows if r.get("error")] +await_err_ids = {r["request_id"] for r in await_rows if r.get("error")} +fnf_only_err = [r for r in fnf_err if r["request_id"] not in await_err_ids] +print("\n Errors unique to FnF: %d" % len(fnf_only_err)) +for r in fnf_only_err[:5]: + print(" input=%d err=%s" % (r["input_length"], r["error"][:60]))