Gahow Wang
1b9268ba4c
Fixed race condition in P instance selection (all going to inst_0). P2P design: HEAVY requests prefill on least-loaded OTHER instance, KV transfer via Mooncake, decode on session-sticky instance. Result (200 req, fresh restart, vs baseline): TTFT p50: 1.080 -> 0.939 (-13%) <- median improves (decode not disrupted) TTFT p90: 9.410 -> 14.987 (+59%) <- tail worsens (KV transfer on large req) TPOT p90: 0.076 -> 0.075 (-1%) <- unchanged (not the bottleneck) E2E p50: 5.306 -> 5.565 (+5%) <- slightly worse overall The P2P offload helps the common case (WARM/MEDIUM get lower TTFT because their instance isn't blocked by a heavy prefill) but hurts HEAVY requests (extra KV transfer latency). This is a median-vs-tail tradeoff. For SLOs targeting p50: P2P offload helps. For SLOs targeting p90/p99: baseline combined is better. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
143 lines
4.9 KiB
Python
143 lines
4.9 KiB
Python
"""Analyze prefill aggregation strategy: 1 aggregator GPU + 7 combined GPUs."""
|
|
import json
|
|
from collections import defaultdict
|
|
|
|
rows = [json.loads(l) for l in open("traces/sampled_1000req_seed42.jsonl")]
|
|
rows.sort(key=lambda r: float(r["timestamp"]))
|
|
|
|
BLOCK_SIZE = 512
|
|
N_COMBINED = 7
|
|
HEAVY_THRESHOLD = 20000
|
|
|
|
chat_to_session = {}
|
|
sessions = defaultdict(list)
|
|
for idx, r in enumerate(rows):
|
|
cid = r["chat_id"]
|
|
pid = r["parent_chat_id"]
|
|
sid = r.get("session_id", str(cid) if pid < 0 else chat_to_session.get(pid, str(pid)))
|
|
chat_to_session[cid] = str(sid)
|
|
sessions[str(sid)].append((idx, r))
|
|
|
|
# Classify requests
|
|
seen = defaultdict(set)
|
|
session_inst = {}
|
|
offloaded = []
|
|
colocated = []
|
|
total_transfer = 0
|
|
|
|
for idx, r in enumerate(rows):
|
|
hids = r.get("hash_ids", [])
|
|
il = r["input_length"]
|
|
sid = str(r.get("session_id", chat_to_session.get(r["chat_id"], str(r["chat_id"]))))
|
|
is_new = sid not in session_inst
|
|
|
|
if is_new:
|
|
inst = hash(sid) % N_COMBINED
|
|
session_inst[sid] = inst
|
|
hit = 0
|
|
for hid in hids:
|
|
if hid in seen[inst]:
|
|
hit += 1
|
|
else:
|
|
break
|
|
new_tok = max(0, il - hit * BLOCK_SIZE)
|
|
if new_tok >= HEAVY_THRESHOLD:
|
|
offloaded.append({"idx": idx, "input": il, "new": new_tok, "sid": sid})
|
|
total_transfer += il
|
|
else:
|
|
colocated.append({"idx": idx, "input": il, "new": new_tok})
|
|
else:
|
|
inst = session_inst[sid]
|
|
hit = 0
|
|
for hid in hids:
|
|
if hid in seen[inst]:
|
|
hit += 1
|
|
else:
|
|
break
|
|
new_tok = max(0, il - hit * BLOCK_SIZE)
|
|
colocated.append({"idx": idx, "input": il, "new": new_tok})
|
|
|
|
target = session_inst.get(sid, 0)
|
|
for hid in hids:
|
|
seen[target].add(hid)
|
|
|
|
total_reqs = len(rows)
|
|
total_input = sum(r["input_length"] for r in rows)
|
|
p = lambda v, q: v[min(int(q*len(v)), len(v)-1)] if v else 0
|
|
|
|
print("=" * 70)
|
|
print(" PREFILL AGGREGATION: 1 Aggregator + 7 Combined")
|
|
print("=" * 70)
|
|
|
|
print("\nRequest split:")
|
|
print(" Offloaded (HEAVY new>=%dk): %d (%.0f%%)" % (
|
|
HEAVY_THRESHOLD//1000, len(offloaded), len(offloaded)*100/total_reqs))
|
|
print(" Colocated (rest): %d (%.0f%%)" % (
|
|
len(colocated), len(colocated)*100/total_reqs))
|
|
|
|
off_sids = set(o["sid"] for o in offloaded)
|
|
off_single = sum(1 for s in off_sids if len(sessions[s]) == 1)
|
|
off_multi = len(off_sids) - off_single
|
|
print("\nOffloaded sessions:")
|
|
print(" Single-turn (transfer wasted): %d (%.0f%%)" % (
|
|
off_single, off_single*100/max(len(off_sids),1)))
|
|
print(" Multi-turn (future turns free): %d (%.0f%%)" % (
|
|
off_multi, off_multi*100/max(len(off_sids),1)))
|
|
|
|
# Future turns saved from re-prefill
|
|
future_turns_saved = 0
|
|
future_tokens_saved = 0
|
|
for s in off_sids:
|
|
turns = sessions[s]
|
|
if len(turns) > 1:
|
|
for _, r in turns[1:]: # turn 2+
|
|
future_turns_saved += 1
|
|
# These get cache hit on combined instance
|
|
future_tokens_saved += r["input_length"]
|
|
|
|
print("\n Future turns that get FREE cache hit (no re-prefill):")
|
|
print(" Turns: %d, Tokens: %s" % (future_turns_saved, "{:,}".format(future_tokens_saved)))
|
|
|
|
print("\nKV transfer:")
|
|
print(" Volume: %s tokens (%.1f%% of total input)" % (
|
|
"{:,}".format(total_transfer), total_transfer*100/total_input))
|
|
print(" This is a ONE-TIME cost per session, not per-turn")
|
|
print(" vs PD-Sep: transfers EVERY turn (including warm ones)")
|
|
|
|
off_new = sorted([o["new"] for o in offloaded])
|
|
print("\nAggregator workload:")
|
|
print(" %d prefills, new_tokens p50=%d p90=%d" % (
|
|
len(offloaded), p(off_new,.5), p(off_new,.9)))
|
|
print(" Total new tokens: %s" % "{:,}".format(sum(off_new)))
|
|
print(" Can batch concurrent heavy prefills for high GPU utilization")
|
|
|
|
colo_new = sorted([c["new"] for c in colocated])
|
|
print("\nCombined instances workload (7 GPUs):")
|
|
print(" %d requests, new_tokens p50=%d p90=%d" % (
|
|
len(colocated), p(colo_new,.5), p(colo_new,.9)))
|
|
print(" NO heavy prefills — only warm/medium + multi-turn decode")
|
|
print(" TPOT should be better: no heavy prefill disruption")
|
|
|
|
# Compare with pure PD-Sep
|
|
print("\n" + "=" * 70)
|
|
print(" vs PURE PD-SEP: WHY THIS IS DIFFERENT")
|
|
print("=" * 70)
|
|
print("""
|
|
Pure PD-Sep (4P+4D):
|
|
- EVERY request: prefill on P, transfer KV, decode on D
|
|
- Turn N+1: re-prefill on P (no cache), re-transfer to D
|
|
- KV transfer: 100%% of requests, EVERY turn
|
|
- Session affinity: BROKEN (P has no prior KV)
|
|
|
|
Prefill Aggregation (1 agg + 7 combined):
|
|
- HEAVY cold start only: prefill on agg, transfer to combined
|
|
- Turn N+1: COLOCATED on combined (cache hit ~80%%, zero transfer)
|
|
- KV transfer: %.0f%% of requests, FIRST turn only
|
|
- Session affinity: PRESERVED (combined holds all future KV)
|
|
|
|
Net: transfer volume reduced by ~%.0fx vs PD-Sep
|
|
""" % (
|
|
len(offloaded)*100/total_reqs,
|
|
total_input / max(total_transfer, 1),
|
|
))
|