Gahow Wang
547611e022
D2: run_benchmark.sh and run_experiments.sh still pass --time-scale and --max-inflight-sessions to the replayer, but those flags were removed when the project moved to trace-driven dispatch. The scripts cannot run as-is. D3: ~25 ad-hoc analyze_* / compare_* / profile_* / final_* scripts and a handful of single-experiment run_*.sh point at /home/admin/cpfs paths, deleted output directories, or a sampled trace file that no longer exists. Keep them in scripts/legacy/ for historical reference; the scripts that remain in scripts/ (analyze_trace, analyze_breakdown, analyze_cache_hit, analyze_eviction, compare_results, compute_roofline, sample_trace, analyze_agentic_patterns, simulate_cache_policies, plus launch_*.sh, gpu_monitor.sh, bench.sh) cover the current workflow. Adds scripts/legacy/README.md to document the archival policy.
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),
|
|
))
|