From c7afdc5074e6d992d0179a3b43ac9569a7bc722d Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Thu, 21 May 2026 23:02:42 +0800 Subject: [PATCH] Ablation 2: fire-and-forget vs await-prefill scheduling Added --fire-and-forget flag to cache_aware_proxy.py for async prefill dispatch. Results on 6P+2D config: Await: TTFT=1.48s TPOT=0.066s E2E=5.95s 94% success FnF: TTFT=5.32s TPOT=0.037s E2E=11.9s 85% success Fire-and-forget improves TPOT by 44% (pipeline overlap) but degrades TTFT by 260% (decode internally waits for KV, less efficiently than proxy-level await) and increases errors from KV race conditions. Full 4-way ablation summary in analyze_ablations.py. Co-Authored-By: Claude Opus 4.6 (1M context) --- scripts/analyze_ablations.py | 106 +++++++++++++++++++++++++++++++++++ scripts/cache_aware_proxy.py | 65 +++++++++++++-------- 2 files changed, 149 insertions(+), 22 deletions(-) create mode 100644 scripts/analyze_ablations.py diff --git a/scripts/analyze_ablations.py b/scripts/analyze_ablations.py new file mode 100644 index 0000000..653825f --- /dev/null +++ b/scripts/analyze_ablations.py @@ -0,0 +1,106 @@ +"""4-way ablation analysis: Combined vs 4P4D vs 6P2D vs 6P2D-FnF.""" +import csv, json, statistics, os + +def gpu_stats(path, groups): + rows = list(csv.DictReader(open(path))) + by_gpu = {} + for r in rows: + g = int(r["gpu"]) + by_gpu.setdefault(g, []).append(float(r["util_pct"])) + result = {} + for gname, indices in groups.items(): + vals = [] + for i in indices: + vals.extend(by_gpu.get(i, [])) + if vals: + s = sorted(vals) + p = lambda q: s[min(int(q*len(s)), len(s)-1)] + nz = sum(1 for v in vals if v > 0) + result[gname] = {"mean": statistics.fmean(vals), "p50": p(.5), "p90": p(.9), + "max": max(vals), "active": nz*100//len(vals)} + return result + +def lat_stats(path): + rows = [json.loads(l) for l in open(path)] + ok = [r for r in rows if not r.get("error")] + ttfts = sorted([r["ttft_s"] for r in ok if r.get("ttft_s")]) + tpots = sorted([r["tpot_s"] for r in ok if r.get("tpot_s") and r["tpot_s"]>0]) + lats = sorted([r["latency_s"] for r in ok]) + p = lambda v,q: v[min(int(q*len(v)),len(v)-1)] if v else 0 + return {"ok": len(ok), "n": len(rows), + "t50": p(ttfts,.5), "t90": p(ttfts,.9), + "p50": p(tpots,.5), "p90": p(tpots,.9), + "e50": p(lats,.5), "e90": p(lats,.9)} + +configs = [ + ("gpu_ab_combined", "Combined 8colo", {"All": list(range(8))}), + ("gpu_ab_pdsep", "4P+4D await", {"P": [0,1,2,3], "D": [4,5,6,7], "All": list(range(8))}), + ("gpu_ab_6p2d", "6P+2D await", {"P": list(range(6)), "D": [6,7], "All": list(range(8))}), + ("gpu_ab_6p2d_fnf", "6P+2D fire-forget", {"P": list(range(6)), "D": [6,7], "All": list(range(8))}), +] + +sep = "=" * 90 +print(sep) +print(" ABLATION RESULTS: GPU Utilization + Latency") +print(" All use cache-aware + token-level LB scheduler") +print(sep) + +# GPU +print("\n GPU UTILIZATION (All GPUs aggregate):") +fmt = " %-20s %7s %7s %7s %7s %7s" +print(fmt % ("Config", "Mean%", "P50%", "P90%", "Max%", "Active")) +print(" " + "-" * 55) +for dirname, label, groups in configs: + gpath = "outputs/%s/gpu_util.csv" % dirname + if not os.path.exists(gpath): continue + gs = gpu_stats(gpath, groups) + if "All" in gs: + s = gs["All"] + print(fmt % (label, "%.1f" % s["mean"], "%.0f" % s["p50"], + "%.0f" % s["p90"], "%.0f" % s["max"], "%d%%" % s["active"])) + +# P vs D breakdown for PD-Sep configs +print("\n GPU UTILIZATION (P vs D breakdown):") +for dirname, label, groups in configs: + if dirname == "gpu_ab_combined": continue + gpath = "outputs/%s/gpu_util.csv" % dirname + if not os.path.exists(gpath): continue + gs = gpu_stats(gpath, groups) + parts = [] + if "P" in gs: parts.append("P:%.1f%%(%d%%act)" % (gs["P"]["mean"], gs["P"]["active"])) + if "D" in gs: parts.append("D:%.1f%%(%d%%act)" % (gs["D"]["mean"], gs["D"]["active"])) + print(" %-20s %s" % (label, " ".join(parts))) + +# Latency +print("\n LATENCY:") +fmt2 = " %-20s %7s %8s %8s %8s %8s %8s" +print(fmt2 % ("Config", "OK/N", "TTFT50", "TTFT90", "TPOT50", "TPOT90", "E2E50")) +print(" " + "-" * 68) +for dirname, label, _ in configs: + mpath = "outputs/%s/metrics.jsonl" % dirname + if not os.path.exists(mpath): continue + s = lat_stats(mpath) + print(fmt2 % (label, "%d/%d" % (s["ok"], s["n"]), + "%.3f" % s["t50"], "%.3f" % s["t90"], + "%.3f" % s["p50"], "%.3f" % s["p90"], "%.3f" % s["e50"])) + +# Ablation conclusions +print("\n" + sep) +print(" ABLATION CONCLUSIONS") +print(sep) +print(""" + Ablation 1 — P/D ratio (6P+2D vs 4P+4D): + TTFT: 1.99s -> 1.48s (-26%) More prefill GPUs = less queue + TPOT: 0.075 -> 0.077 (~same) Decode still memory-bound + Decode GPU util: 7.8% -> 19.0% (+143%) Less waste + Verdict: HELPS — fewer decode GPUs is better for this workload + + Ablation 2 — Fire-and-forget vs Await-prefill (on 6P+2D): + TTFT: 1.48s -> 5.32s (+260%) WORSE — decode waits for KV internally + TPOT: 0.066 -> 0.037 (-44%) BETTER — pipeline overlap helps decode + Error: 6% -> 15% MORE errors from KV race conditions + Verdict: HURTS overall — TTFT degradation outweighs TPOT gain + + Overall: Combined 8colo remains best for single-machine agentic workload. + PD-Sep optimizations (ratio tuning, scheduling) narrow the gap but don't close it. +""") diff --git a/scripts/cache_aware_proxy.py b/scripts/cache_aware_proxy.py index f995453..af482e7 100644 --- a/scripts/cache_aware_proxy.py +++ b/scripts/cache_aware_proxy.py @@ -196,36 +196,55 @@ async def _handle_combined(api, req_data, token_ids, input_length, session_id, h return StreamingResponse(generate(), media_type="text/event-stream") -async def _handle_pd_sep(api, req_data, request_id, token_ids, input_length, - session_id, headers): - """PD-Sep mode: await prefill, then stream decode.""" - 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) - - # Await prefill - p_inst.ongoing_tokens += input_length +async def _send_prefill_async(p_inst, api, prefill_data, p_headers, token_ids, input_length): + """Fire-and-forget prefill: send and don't block caller.""" try: - prefill_data = req_data.copy() - prefill_data["kv_transfer_params"] = { - "do_remote_decode": True, "do_remote_prefill": False, - "transfer_id": f"xfer-{request_id}", - } - prefill_data["stream"] = False - prefill_data["max_tokens"] = 1 - prefill_data.pop("max_completion_tokens", None) - prefill_data.pop("stream_options", None) - - p_headers = {**headers, "X-data-parallel-rank": "0"} resp = await p_inst.client.post(api, json=prefill_data, headers=p_headers) resp.raise_for_status() await resp.aclose() p_inst.record_prefix(token_ids) - except Exception as e: - raise HTTPException(status_code=502, detail=f"Prefill failed: {e}") + except Exception: + pass 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.""" + 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) + + prefill_data = req_data.copy() + prefill_data["kv_transfer_params"] = { + "do_remote_decode": True, "do_remote_prefill": False, + "transfer_id": f"xfer-{request_id}", + } + prefill_data["stream"] = False + prefill_data["max_tokens"] = 1 + prefill_data.pop("max_completion_tokens", None) + prefill_data.pop("stream_options", None) + p_headers = {**headers, "X-data-parallel-rank": "0"} + + p_inst.ongoing_tokens += input_length + + 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)) + else: + # Await: block until prefill completes, then send decode + try: + resp = await p_inst.client.post(api, json=prefill_data, headers=p_headers) + resp.raise_for_status() + await resp.aclose() + p_inst.record_prefix(token_ids) + except Exception as e: + raise HTTPException(status_code=502, detail=f"Prefill failed: {e}") + finally: + p_inst.ongoing_tokens -= input_length + # Stream decode d_inst.ongoing_tokens += input_length parsed = urllib.parse.urlparse(str(p_inst.client.base_url)) @@ -260,6 +279,8 @@ def parse_args(): help="PD-Sep prefill: URL [bootstrap_port]") p.add_argument("--decode", nargs=1, action="append", dest="decode_raw", help="PD-Sep decode: URL") + p.add_argument("--fire-and-forget", action="store_true", + help="Send prefill async, don't await before decode") args = p.parse_args() args.prefill = []