chore: add standalone qwenapp and session-routing analysis scripts
Three one-off scripts kept outside the two packages: a formatter for the qwenapp Aquila access-log shape, its characterization data generator, and a session-sticky routing imbalance simulation over a formatted trace.
This commit is contained in:
339
analyze_qwenapp_trace.py
Normal file
339
analyze_qwenapp_trace.py
Normal file
@@ -0,0 +1,339 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate all characterization CSV/JSON data for the qwenapp trace.
|
||||
|
||||
Reads the formatted qwenapp trace (release format with hash_ids) and produces:
|
||||
1. token_length_cdf_qwenapp.csv + .stats.json
|
||||
2. f2a_block_lifetime_cdf_qwenapp.csv + .stats.json
|
||||
3. f2a_reuse_gap_cdf_qwenapp.csv
|
||||
4. f2a_reuse_timeline_qwenapp.csv
|
||||
5. f2b_session_skew_cdf_qwenapp.json
|
||||
|
||||
Run on dash1:
|
||||
python paper/data/analyze_qwenapp_trace.py \
|
||||
~/ali-trace/trace-qwenapp-formatted/0529-1000-1200-ttp.jsonl
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import math
|
||||
import sys
|
||||
from collections import Counter, defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
def _percentile(sorted_arr, p):
|
||||
idx = int(len(sorted_arr) * p / 100.0)
|
||||
return sorted_arr[min(idx, len(sorted_arr) - 1)]
|
||||
|
||||
|
||||
def _log_bins(lo, hi, n=200):
|
||||
return np.logspace(np.log10(max(lo, 1e-4)), np.log10(hi), n)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. Token length CDF
|
||||
# ---------------------------------------------------------------------------
|
||||
def build_token_length(rows, sessions, outdir):
|
||||
input_toks = sorted(r["input_length"] for r in rows)
|
||||
output_toks = sorted(r["output_length"] for r in rows)
|
||||
|
||||
append_toks = []
|
||||
for sid, sess_rows in sessions.items():
|
||||
if len(sess_rows) < 2:
|
||||
continue
|
||||
for i in range(1, len(sess_rows)):
|
||||
delta = sess_rows[i]["input_length"] - sess_rows[i - 1]["input_length"]
|
||||
append_toks.append(delta)
|
||||
append_toks.sort()
|
||||
|
||||
n = max(len(input_toks), len(output_toks), len(append_toks))
|
||||
cdf_points = np.linspace(0, 1, min(n, 2000))
|
||||
|
||||
with (outdir / "token_length_cdf_qwenapp.csv").open("w", newline="") as f:
|
||||
w = csv.writer(f)
|
||||
w.writerow(["cdf", "input_tokens", "append_tokens", "output_tokens"])
|
||||
for c in cdf_points:
|
||||
ii = min(int(c * len(input_toks)), len(input_toks) - 1)
|
||||
ai = min(int(c * len(append_toks)), len(append_toks) - 1) if append_toks else 0
|
||||
oi = min(int(c * len(output_toks)), len(output_toks) - 1)
|
||||
w.writerow([
|
||||
f"{c:.3f}",
|
||||
input_toks[ii],
|
||||
append_toks[ai] if append_toks else 0,
|
||||
output_toks[oi],
|
||||
])
|
||||
|
||||
n_root = sum(1 for sid, sr in sessions.items() for i, r in enumerate(sr) if i == 0)
|
||||
n_neg = sum(1 for a in append_toks if a < 0)
|
||||
stats = {
|
||||
"trace": "trace-qwenapp-formatted/0529-1000-1200.jsonl",
|
||||
"n_records": len(rows),
|
||||
"input": {
|
||||
"n": len(input_toks), "p50": _percentile(input_toks, 50),
|
||||
"p90": _percentile(input_toks, 90), "p99": _percentile(input_toks, 99),
|
||||
"mean": sum(input_toks) / len(input_toks),
|
||||
"min": input_toks[0], "max": input_toks[-1],
|
||||
},
|
||||
"append": {
|
||||
"n": len(append_toks),
|
||||
"p50": _percentile(append_toks, 50) if append_toks else 0,
|
||||
"p90": _percentile(append_toks, 90) if append_toks else 0,
|
||||
"p99": _percentile(append_toks, 99) if append_toks else 0,
|
||||
"mean": sum(append_toks) / len(append_toks) if append_toks else 0,
|
||||
"min": append_toks[0] if append_toks else 0,
|
||||
"max": append_toks[-1] if append_toks else 0,
|
||||
"n_root": n_root, "n_negative": n_neg,
|
||||
},
|
||||
"output": {
|
||||
"n": len(output_toks), "p50": _percentile(output_toks, 50),
|
||||
"p90": _percentile(output_toks, 90), "p99": _percentile(output_toks, 99),
|
||||
"mean": sum(output_toks) / len(output_toks),
|
||||
"min": output_toks[0], "max": output_toks[-1],
|
||||
},
|
||||
}
|
||||
(outdir / "token_length_cdf_qwenapp.stats.json").write_text(
|
||||
json.dumps(stats, indent=4, ensure_ascii=False))
|
||||
print(f" token_length: {len(rows)} rows, append={len(append_toks)}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. Block lifetime CDF
|
||||
# ---------------------------------------------------------------------------
|
||||
def build_block_lifetime(rows, outdir):
|
||||
block_first: dict[int, float] = {}
|
||||
block_last: dict[int, float] = {}
|
||||
for r in rows:
|
||||
ts = r["timestamp"]
|
||||
for h in r["hash_ids"]:
|
||||
if h not in block_first:
|
||||
block_first[h] = ts
|
||||
block_last[h] = ts
|
||||
|
||||
lifetimes = []
|
||||
for h in block_first:
|
||||
lt = block_last[h] - block_first[h]
|
||||
if lt > 0:
|
||||
lifetimes.append(lt)
|
||||
lifetimes.sort()
|
||||
|
||||
if not lifetimes:
|
||||
print(" block_lifetime: no reused blocks!")
|
||||
return
|
||||
|
||||
bins = np.concatenate([
|
||||
_log_bins(0.01, lifetimes[-1] * 1.01, 300),
|
||||
[float("inf")],
|
||||
])
|
||||
counts, _ = np.histogram(lifetimes, bins=bins)
|
||||
|
||||
with (outdir / "f2a_block_lifetime_cdf_qwenapp.csv").open("w", newline="") as f:
|
||||
w = csv.writer(f)
|
||||
w.writerow(["bucket_upper_s", "count"])
|
||||
for edge, cnt in zip(bins[1:], counts):
|
||||
w.writerow([edge, int(cnt)])
|
||||
|
||||
stats = {
|
||||
"n_records": len(rows),
|
||||
"n_distinct_blocks": len(block_first),
|
||||
"n_reused_blocks": len(lifetimes),
|
||||
"reused_fraction": len(lifetimes) / len(block_first) if block_first else 0,
|
||||
"p50_s": _percentile(lifetimes, 50),
|
||||
"p90_s": _percentile(lifetimes, 90),
|
||||
"p99_s": _percentile(lifetimes, 99),
|
||||
"mean_s": sum(lifetimes) / len(lifetimes),
|
||||
"max_s": lifetimes[-1],
|
||||
}
|
||||
(outdir / "f2a_block_lifetime_cdf_qwenapp.stats.json").write_text(
|
||||
json.dumps(stats, indent=4, ensure_ascii=False))
|
||||
print(f" block_lifetime: {len(lifetimes)} reused blocks / {len(block_first)} total")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. Reuse gap CDF
|
||||
# ---------------------------------------------------------------------------
|
||||
def build_reuse_gap(rows, sessions, outdir):
|
||||
block_last_touch: dict[int, float] = {}
|
||||
block_session: dict[int, str] = {}
|
||||
|
||||
# For qwenapp: no system/content split via tools. Instead:
|
||||
# intra-session = reuse within the same session
|
||||
# cross-session = reuse from a different session
|
||||
# We don't have a system-prompt block identifier, so approximate:
|
||||
# system-prompt blocks = the first N blocks of each request that match
|
||||
# across sessions (shared system prompt prefix)
|
||||
gaps_intra = []
|
||||
gaps_cross = []
|
||||
|
||||
for r in rows:
|
||||
ts = r["timestamp"]
|
||||
sid = r["session_id"]
|
||||
for h in r["hash_ids"]:
|
||||
if h in block_last_touch:
|
||||
gap = ts - block_last_touch[h]
|
||||
if gap > 0:
|
||||
prev_sid = block_session[h]
|
||||
if prev_sid == sid:
|
||||
gaps_intra.append(gap)
|
||||
else:
|
||||
gaps_cross.append(gap)
|
||||
block_last_touch[h] = ts
|
||||
block_session[h] = sid
|
||||
|
||||
max_gap = max(
|
||||
max(gaps_intra) if gaps_intra else 1,
|
||||
max(gaps_cross) if gaps_cross else 1,
|
||||
)
|
||||
bins = np.concatenate([
|
||||
_log_bins(0.01, max_gap * 1.01, 300),
|
||||
[float("inf")],
|
||||
])
|
||||
h_intra, _ = np.histogram(gaps_intra, bins=bins) if gaps_intra else (np.zeros(len(bins) - 1), None)
|
||||
h_cross, _ = np.histogram(gaps_cross, bins=bins) if gaps_cross else (np.zeros(len(bins) - 1), None)
|
||||
|
||||
with (outdir / "f2a_reuse_gap_cdf_qwenapp.csv").open("w", newline="") as f:
|
||||
w = csv.writer(f)
|
||||
w.writerow(["bucket_upper_s", "intra", "cross_session"])
|
||||
for edge, ci, cc in zip(bins[1:], h_intra, h_cross):
|
||||
w.writerow([edge, int(ci), int(cc)])
|
||||
|
||||
print(f" reuse_gap: intra={len(gaps_intra)}, cross={len(gaps_cross)}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. Reuse timeline
|
||||
# ---------------------------------------------------------------------------
|
||||
def build_reuse_timeline(rows, sessions, outdir, bin_s=30.0):
|
||||
max_ts = max(r["timestamp"] for r in rows)
|
||||
n_bins = int(math.ceil(max_ts / bin_s))
|
||||
|
||||
bin_data = [{"n_req": 0, "intra_tok": 0, "cross_tok": 0,
|
||||
"new_tok": 0, "total_tok": 0} for _ in range(n_bins)]
|
||||
|
||||
block_last_touch: dict[int, float] = {}
|
||||
block_session: dict[int, str] = {}
|
||||
|
||||
for r in rows:
|
||||
ts = r["timestamp"]
|
||||
b = min(int(ts / bin_s), n_bins - 1)
|
||||
bin_data[b]["n_req"] += 1
|
||||
|
||||
n_blocks = len(r["hash_ids"])
|
||||
reuse_intra = 0
|
||||
reuse_cross = 0
|
||||
new_blocks = 0
|
||||
sid = r["session_id"]
|
||||
|
||||
for h in r["hash_ids"]:
|
||||
if h in block_last_touch:
|
||||
prev_sid = block_session[h]
|
||||
if prev_sid == sid:
|
||||
reuse_intra += 1
|
||||
else:
|
||||
reuse_cross += 1
|
||||
else:
|
||||
new_blocks += 1
|
||||
block_last_touch[h] = ts
|
||||
block_session[h] = sid
|
||||
|
||||
tok_per_block = r["input_length"] / n_blocks if n_blocks > 0 else 0
|
||||
bin_data[b]["intra_tok"] += reuse_intra * tok_per_block
|
||||
bin_data[b]["cross_tok"] += reuse_cross * tok_per_block
|
||||
bin_data[b]["new_tok"] += new_blocks * tok_per_block
|
||||
bin_data[b]["total_tok"] += r["input_length"]
|
||||
|
||||
with (outdir / "f2a_reuse_timeline_qwenapp.csv").open("w", newline="") as f:
|
||||
w = csv.writer(f)
|
||||
w.writerow(["bin_start_s", "n_req", "qps", "intra_tok_s",
|
||||
"cross_tok_s", "new_tok_s", "reuse_tok_s", "total_tok_s", "reuse_frac"])
|
||||
for i, d in enumerate(bin_data):
|
||||
t0 = i * bin_s
|
||||
qps = d["n_req"] / bin_s
|
||||
reuse = d["intra_tok"] + d["cross_tok"]
|
||||
total = d["total_tok"]
|
||||
frac = reuse / total if total > 0 else 0
|
||||
w.writerow([
|
||||
f"{t0:.1f}", d["n_req"], f"{qps:.3f}",
|
||||
f"{d['intra_tok']:.1f}", f"{d['cross_tok']:.1f}",
|
||||
f"{d['new_tok']:.1f}", f"{reuse:.1f}", f"{total:.1f}",
|
||||
f"{frac:.4f}",
|
||||
])
|
||||
|
||||
print(f" reuse_timeline: {n_bins} bins")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. Session skew
|
||||
# ---------------------------------------------------------------------------
|
||||
def build_session_skew(rows, sessions, outdir):
|
||||
session_tokens = {}
|
||||
for sid, sess_rows in sessions.items():
|
||||
session_tokens[sid] = sum(r["input_length"] for r in sess_rows)
|
||||
|
||||
total = sum(session_tokens.values())
|
||||
ranked = sorted(session_tokens.values(), reverse=True)
|
||||
cumsum = np.cumsum(ranked) / total * 100.0
|
||||
n = len(ranked)
|
||||
rank_pct = np.arange(1, n + 1) / n * 100.0
|
||||
|
||||
step = max(1, n // 500)
|
||||
samples = [{"rank_pct": float(rank_pct[i]), "cum_pct": float(cumsum[i])}
|
||||
for i in range(0, n, step)]
|
||||
if samples[-1]["rank_pct"] < 100.0:
|
||||
samples.append({"rank_pct": 100.0, "cum_pct": 100.0})
|
||||
|
||||
anchors = {}
|
||||
for pct in [1, 5, 10, 25, 50]:
|
||||
idx = max(0, int(n * pct / 100.0) - 1)
|
||||
anchors[f"top{pct}pct"] = float(cumsum[idx])
|
||||
|
||||
data = {
|
||||
"n_sessions": n,
|
||||
"total_tokens": int(total),
|
||||
"anchors_check": anchors,
|
||||
"source_trace": "trace-qwenapp-formatted/0529-1000-1200.jsonl",
|
||||
"samples": samples,
|
||||
}
|
||||
(outdir / "f2b_session_skew_cdf_qwenapp.json").write_text(
|
||||
json.dumps(data, indent=2, ensure_ascii=False))
|
||||
print(f" session_skew: {n} sessions, top1%={anchors['top1pct']:.1f}%")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
def main():
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument("trace", type=Path)
|
||||
p.add_argument("--outdir", type=Path, default=None)
|
||||
args = p.parse_args()
|
||||
|
||||
outdir = args.outdir or Path(__file__).resolve().parent
|
||||
outdir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
print(f"Reading {args.trace} ...")
|
||||
rows = []
|
||||
sessions: dict[str, list[dict]] = defaultdict(list)
|
||||
with args.trace.open() as f:
|
||||
for line in f:
|
||||
r = json.loads(line)
|
||||
rows.append(r)
|
||||
sessions[r["session_id"]].append(r)
|
||||
|
||||
for sess_rows in sessions.values():
|
||||
sess_rows.sort(key=lambda r: r["turn"])
|
||||
|
||||
print(f"Loaded {len(rows)} rows, {len(sessions)} sessions")
|
||||
|
||||
build_token_length(rows, sessions, outdir)
|
||||
build_block_lifetime(rows, outdir)
|
||||
build_reuse_gap(rows, sessions, outdir)
|
||||
build_reuse_timeline(rows, sessions, outdir)
|
||||
build_session_skew(rows, sessions, outdir)
|
||||
|
||||
print("Done.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
269
format_qwenapp_trace.py
Normal file
269
format_qwenapp_trace.py
Normal file
@@ -0,0 +1,269 @@
|
||||
"""Format raw qwenapp Aquila access-log trace into the standard release format.
|
||||
|
||||
qwenapp traces differ from coding-agent traces in two ways:
|
||||
1. Prompts are pre-rendered strings (not a `messages` list).
|
||||
2. No tool calls — this is a general chat + RAG workload.
|
||||
|
||||
The script reads the raw JSONL, groups rows by session_id, assigns
|
||||
chat_id / parent_chat_id / turn, tokenizes the prompt to build hash_ids
|
||||
for prefix-cache replay, and writes the standard format understood by
|
||||
the replayer and sample_trace.py.
|
||||
|
||||
Usage (run on dash1 where the trace and tokenizer live):
|
||||
python scripts/format_qwenapp_trace.py \
|
||||
--input ~/ali-trace/trace-qwenapp/0529-1000-1200.jsonl \
|
||||
--output ~/ali-trace/trace-qwenapp-formatted/0529-1000-1200.jsonl \
|
||||
--tokenizer ~/ali-trace/trace_model_meta/Qwen/Qwen3-Coder-480B-A35B-Instruct/tokenizer.json
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import sys
|
||||
from array import array
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
from tokenizers import Tokenizer
|
||||
from tqdm.auto import tqdm
|
||||
|
||||
|
||||
def _block_digest(block: list[int]) -> bytes:
|
||||
digest = hashlib.blake2b(digest_size=16)
|
||||
digest.update(len(block).to_bytes(4, "little", signed=False))
|
||||
digest.update(array("I", block).tobytes())
|
||||
return digest.digest()
|
||||
|
||||
|
||||
def _parse_row(raw: dict) -> dict | None:
|
||||
"""Extract the fields we need from one raw Aquila access-log row."""
|
||||
request_params = raw.get("request_params")
|
||||
if isinstance(request_params, str):
|
||||
request_params = json.loads(request_params)
|
||||
response_params = raw.get("response_params")
|
||||
if isinstance(response_params, str):
|
||||
response_params = json.loads(response_params)
|
||||
|
||||
if not isinstance(request_params, dict) or not isinstance(response_params, dict):
|
||||
return None
|
||||
|
||||
inp = request_params.get("payload", {}).get("input", {})
|
||||
resp_payload = response_params.get("payload", {})
|
||||
usage = resp_payload.get("usage", {})
|
||||
resp_header = response_params.get("header", {})
|
||||
|
||||
prompt = inp.get("prompt", "")
|
||||
if not prompt:
|
||||
return None
|
||||
|
||||
input_tokens = int(usage.get("input_tokens", 0))
|
||||
output_tokens = int(usage.get("output_tokens", 0))
|
||||
if input_tokens <= 0:
|
||||
return None
|
||||
|
||||
status_code = str(resp_header.get("status_code", raw.get("status_code", "")))
|
||||
if status_code and status_code != "200" and status_code != "1000":
|
||||
return None
|
||||
|
||||
# Timestamp: prefer the `time` field (ms precision), fall back to __time__
|
||||
time_str = raw.get("time", "")
|
||||
wall_clock_s = _parse_time_to_epoch(time_str) if time_str else None
|
||||
if wall_clock_s is None:
|
||||
epoch_s = raw.get("__time__")
|
||||
wall_clock_s = float(epoch_s) if epoch_s else 0.0
|
||||
|
||||
session_id = str(inp.get("session_id", ""))
|
||||
user_id = str((inp.get("source") or {}).get("user_id", ""))
|
||||
request_id = str(raw.get("request_id", inp.get("request_id", "")))
|
||||
|
||||
return {
|
||||
"session_id": session_id,
|
||||
"user_id": user_id,
|
||||
"request_id": request_id,
|
||||
"wall_clock_s": wall_clock_s,
|
||||
"input_tokens": input_tokens,
|
||||
"output_tokens": output_tokens,
|
||||
"prompt": prompt,
|
||||
"total_cost_time_ms": int(raw.get("total_cost_time", 0)),
|
||||
}
|
||||
|
||||
|
||||
def _parse_time_to_epoch(time_str: str) -> float | None:
|
||||
"""Parse '2026-05-29 09:59:59.505' (UTC+8) to epoch seconds."""
|
||||
import datetime
|
||||
try:
|
||||
dt = datetime.datetime.strptime(time_str.strip(), "%Y-%m-%d %H:%M:%S.%f")
|
||||
except ValueError:
|
||||
try:
|
||||
dt = datetime.datetime.strptime(time_str.strip(), "%Y-%m-%d %H:%M:%S")
|
||||
except ValueError:
|
||||
return None
|
||||
tz_utc8 = datetime.timezone(datetime.timedelta(hours=8))
|
||||
dt = dt.replace(tzinfo=tz_utc8)
|
||||
return dt.timestamp()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
p = argparse.ArgumentParser(description=__doc__,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
p.add_argument("--input", type=Path, required=True)
|
||||
p.add_argument("--output", type=Path, required=True)
|
||||
p.add_argument("--tokenizer", type=Path, required=True,
|
||||
help="Path to tokenizer.json (Qwen3 family)")
|
||||
p.add_argument("--block-size", type=int, default=512)
|
||||
p.add_argument("--ttp", action="store_true",
|
||||
help="Also emit time_to_parent_chat field")
|
||||
args = p.parse_args()
|
||||
|
||||
print(f"Loading tokenizer from {args.tokenizer} ...")
|
||||
tokenizer = Tokenizer.from_file(str(args.tokenizer))
|
||||
|
||||
# --- Pass 1: read all rows, group by session ---
|
||||
print(f"Reading {args.input} ...")
|
||||
sessions: dict[str, list[dict]] = defaultdict(list)
|
||||
skipped = 0
|
||||
total = 0
|
||||
|
||||
with args.input.open("r", encoding="utf-8") as fh:
|
||||
for line in tqdm(fh, desc="Read", unit="row"):
|
||||
total += 1
|
||||
raw = json.loads(line)
|
||||
parsed = _parse_row(raw)
|
||||
if parsed is None:
|
||||
skipped += 1
|
||||
continue
|
||||
sessions[parsed["session_id"]].append(parsed)
|
||||
|
||||
print(f" Read {total} rows, kept {total - skipped}, skipped {skipped}")
|
||||
print(f" Sessions: {len(sessions)}")
|
||||
|
||||
# --- Sort each session by timestamp and assign turn/chat_id ---
|
||||
for rows in sessions.values():
|
||||
rows.sort(key=lambda r: r["wall_clock_s"])
|
||||
|
||||
# Assign globally-unique chat_ids
|
||||
next_chat_id = 1
|
||||
chat_id_map: dict[str, int] = {} # request_id -> chat_id
|
||||
all_rows: list[dict] = []
|
||||
|
||||
for sid, rows in sessions.items():
|
||||
prev_chat_id = -1
|
||||
prev_end_s = None
|
||||
for turn_idx, row in enumerate(rows, start=1):
|
||||
cid = next_chat_id
|
||||
next_chat_id += 1
|
||||
chat_id_map[row["request_id"]] = cid
|
||||
|
||||
ttp = None
|
||||
if prev_end_s is not None:
|
||||
ttp = row["wall_clock_s"] - prev_end_s
|
||||
if ttp < 0:
|
||||
ttp = 0.0
|
||||
|
||||
all_rows.append({
|
||||
"session_id": sid,
|
||||
"chat_id": cid,
|
||||
"parent_chat_id": prev_chat_id,
|
||||
"turn": turn_idx,
|
||||
"wall_clock_s": row["wall_clock_s"],
|
||||
"input_tokens": row["input_tokens"],
|
||||
"output_tokens": row["output_tokens"],
|
||||
"prompt": row["prompt"],
|
||||
"time_to_parent_chat": ttp,
|
||||
"total_cost_time_ms": row["total_cost_time_ms"],
|
||||
})
|
||||
|
||||
prev_chat_id = cid
|
||||
prev_end_s = row["wall_clock_s"] + row["total_cost_time_ms"] / 1000.0
|
||||
|
||||
# Sort globally by timestamp
|
||||
all_rows.sort(key=lambda r: r["wall_clock_s"])
|
||||
|
||||
# Re-zero timestamps
|
||||
t0 = all_rows[0]["wall_clock_s"] if all_rows else 0.0
|
||||
|
||||
# --- Pass 2: tokenize and hash ---
|
||||
print(f"Tokenizing {len(all_rows)} rows (block_size={args.block_size}) ...")
|
||||
block_ids_by_digest: dict[str, int] = {}
|
||||
next_block_id = 0
|
||||
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
with args.output.open("w", encoding="utf-8") as out:
|
||||
for row in tqdm(all_rows, desc="Tokenize+write", unit="row"):
|
||||
token_ids = tokenizer.encode(row["prompt"]).ids
|
||||
|
||||
# Build block hashes
|
||||
hash_ids = []
|
||||
for i in range(0, len(token_ids), args.block_size):
|
||||
block = token_ids[i:i + args.block_size]
|
||||
digest_hex = _block_digest(block).hex()
|
||||
bid = block_ids_by_digest.get(digest_hex)
|
||||
if bid is None:
|
||||
bid = next_block_id
|
||||
next_block_id += 1
|
||||
block_ids_by_digest[digest_hex] = bid
|
||||
hash_ids.append(bid)
|
||||
|
||||
# Pad hash_ids to match provider input_tokens if needed
|
||||
target_blocks = (row["input_tokens"] + args.block_size - 1) // args.block_size
|
||||
while len(hash_ids) < target_blocks:
|
||||
pad_key = f"pad:{row['chat_id']}:{len(hash_ids)}"
|
||||
pad_hex = hashlib.blake2b(pad_key.encode(), digest_size=16).hexdigest()
|
||||
bid = block_ids_by_digest.get(pad_hex)
|
||||
if bid is None:
|
||||
bid = next_block_id
|
||||
next_block_id += 1
|
||||
block_ids_by_digest[pad_hex] = bid
|
||||
hash_ids.append(bid)
|
||||
hash_ids = hash_ids[:target_blocks]
|
||||
|
||||
out_row = {
|
||||
"chat_id": row["chat_id"],
|
||||
"parent_chat_id": row["parent_chat_id"],
|
||||
"timestamp": round(row["wall_clock_s"] - t0, 3),
|
||||
"input_length": row["input_tokens"],
|
||||
"output_length": row["output_tokens"],
|
||||
"type": "app",
|
||||
"turn": row["turn"],
|
||||
"hash_ids": hash_ids,
|
||||
"session_id": row["session_id"],
|
||||
}
|
||||
if args.ttp and row["time_to_parent_chat"] is not None:
|
||||
out_row["time_to_parent_chat"] = round(row["time_to_parent_chat"], 3)
|
||||
|
||||
out.write(json.dumps(out_row, ensure_ascii=False) + "\n")
|
||||
|
||||
# --- Summary ---
|
||||
n_sessions = len(sessions)
|
||||
n_requests = len(all_rows)
|
||||
turns_per = [len(v) for v in sessions.values()]
|
||||
multi = sum(1 for t in turns_per if t > 1)
|
||||
span_s = all_rows[-1]["wall_clock_s"] - t0 if all_rows else 0
|
||||
qps = n_requests / span_s if span_s > 0 else 0
|
||||
inp_lens = [r["input_tokens"] for r in all_rows]
|
||||
out_lens = [r["output_tokens"] for r in all_rows]
|
||||
|
||||
unique_blocks = next_block_id
|
||||
shared_blocks = sum(1 for d, bid in block_ids_by_digest.items()
|
||||
if not d.startswith("pad:")) # rough
|
||||
# Count actually shared (appear in >1 request)
|
||||
block_ref_count: dict[int, int] = defaultdict(int)
|
||||
# Re-scan is expensive; use the hash_ids we just wrote
|
||||
# Instead, just report unique count
|
||||
print(f"\n=== Summary ===")
|
||||
print(f" Requests: {n_requests}")
|
||||
print(f" Sessions: {n_sessions}")
|
||||
print(f" Multi-turn: {multi} ({multi*100/n_sessions:.1f}%)")
|
||||
print(f" Turns/session: min={min(turns_per)} max={max(turns_per)} avg={sum(turns_per)/len(turns_per):.1f}")
|
||||
print(f" Input tokens: min={min(inp_lens)} max={max(inp_lens)} avg={sum(inp_lens)/len(inp_lens):.0f}")
|
||||
print(f" Output tokens: min={min(out_lens)} max={max(out_lens)} avg={sum(out_lens)/len(out_lens):.0f}")
|
||||
print(f" Trace span: {span_s:.1f}s ({span_s/60:.1f} min)")
|
||||
print(f" QPS: {qps:.2f}")
|
||||
print(f" Unique hash blocks: {unique_blocks}")
|
||||
print(f"\nWrote {n_requests} rows to {args.output}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
144
session_sticky_imbalance.py
Normal file
144
session_sticky_imbalance.py
Normal file
@@ -0,0 +1,144 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Session-sticky routing imbalance simulation over a formatted ali-trace.
|
||||
|
||||
Routing policy:
|
||||
- turn==1 (parent_chat_id == -1): dispatch to the instance with the FEWEST
|
||||
total requests received in the trailing LOAD_WIN seconds (least-loaded).
|
||||
- turn>1: pinned to the same instance as the request's parent turn (sticky).
|
||||
|
||||
For every tumbling WINDOW-second time window we count the total requests landed
|
||||
on each of the N instances and report imbalance = max / mean over all N
|
||||
instances (idle instances counted in the mean). A CDF of the per-window
|
||||
imbalance values is plotted.
|
||||
|
||||
Run on dash0 (trace lives there). Trace is timestamp-sorted and every parent
|
||||
precedes its child, so a single in-order pass suffices.
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
from collections import deque
|
||||
|
||||
import numpy as np
|
||||
import matplotlib
|
||||
|
||||
matplotlib.use("Agg")
|
||||
import matplotlib.pyplot as plt
|
||||
|
||||
|
||||
def simulate(trace, n_inst, window, load_win, weight="req", route_by="req"):
|
||||
# `route_by` is the trailing-load_win quantity the turn-1 least-loaded
|
||||
# routing balances on; `weight` is what each window accumulates for the
|
||||
# imbalance statistic. Both take values:
|
||||
# "req" -> 1 per request (request count)
|
||||
# "tokens" -> input_length + output_length per request (total tokens)
|
||||
assign = {} # chat_id -> instance
|
||||
cnt = np.zeros(n_inst, dtype=np.int64) # trailing-load_win route load per inst
|
||||
dq = deque() # (ts, inst, rw) pending expiry from the window
|
||||
wincounts = {} # window_idx -> np.array(n_inst)
|
||||
|
||||
with open(trace) as fh:
|
||||
for line in fh:
|
||||
d = json.loads(line)
|
||||
t = d["timestamp"]
|
||||
tok = d["input_length"] + d["output_length"]
|
||||
|
||||
# expire dispatches older than the trailing load window
|
||||
thr = t - load_win
|
||||
while dq and dq[0][0] < thr:
|
||||
_, old, ow = dq.popleft()
|
||||
cnt[old] -= ow
|
||||
|
||||
if d["parent_chat_id"] == -1: # turn 1: least-loaded
|
||||
inst = int(cnt.argmin())
|
||||
else: # turn>1: sticky to parent
|
||||
inst = assign[d["parent_chat_id"]]
|
||||
assign[d["chat_id"]] = inst # chain so deeper turns follow
|
||||
|
||||
rw = (tok if route_by == "tokens"
|
||||
else d["input_length"] if route_by == "input"
|
||||
else 1)
|
||||
dq.append((t, inst, rw))
|
||||
cnt[inst] += rw
|
||||
|
||||
w = int(t // window)
|
||||
arr = wincounts.get(w)
|
||||
if arr is None:
|
||||
arr = np.zeros(n_inst, dtype=np.int64)
|
||||
wincounts[w] = arr
|
||||
arr[inst] += tok if weight == "tokens" else 1
|
||||
|
||||
ws = sorted(wincounts)
|
||||
imb = np.empty(len(ws))
|
||||
for i, w in enumerate(ws):
|
||||
arr = wincounts[w]
|
||||
mean = arr.sum() / n_inst
|
||||
imb[i] = arr.max() / mean if mean > 0 else 0.0
|
||||
return imb, [wincounts[w] for w in ws]
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--trace", required=True)
|
||||
ap.add_argument("--n-inst", type=int, default=380)
|
||||
ap.add_argument("--window", type=float, default=60.0)
|
||||
ap.add_argument("--load-win", type=float, default=60.0)
|
||||
ap.add_argument("--weight", choices=["req", "tokens"], default="req",
|
||||
help="per-window quantity: request count or total tokens")
|
||||
ap.add_argument("--route-by", choices=["req", "tokens", "input"],
|
||||
default="req",
|
||||
help="turn-1 least-loaded routing balances on this quantity "
|
||||
"(input = arrival-known input_length only)")
|
||||
ap.add_argument("--out-prefix", required=True)
|
||||
args = ap.parse_args()
|
||||
|
||||
imb, _ = simulate(args.trace, args.n_inst, args.window, args.load_win,
|
||||
args.weight, args.route_by)
|
||||
|
||||
pcts = {p: float(np.percentile(imb, p)) for p in (50, 90, 95, 99)}
|
||||
stats = {
|
||||
"trace": args.trace,
|
||||
"n_inst": args.n_inst,
|
||||
"window_s": args.window,
|
||||
"load_win_s": args.load_win,
|
||||
"weight": args.weight,
|
||||
"route_by": args.route_by,
|
||||
"n_windows": len(imb),
|
||||
"imbalance_min": float(imb.min()),
|
||||
"imbalance_mean": float(imb.mean()),
|
||||
"imbalance_p50": pcts[50],
|
||||
"imbalance_p90": pcts[90],
|
||||
"imbalance_p95": pcts[95],
|
||||
"imbalance_p99": pcts[99],
|
||||
"imbalance_max": float(imb.max()),
|
||||
}
|
||||
with open(args.out_prefix + "_stats.json", "w") as f:
|
||||
json.dump(stats, f, indent=2)
|
||||
np.save(args.out_prefix + "_imbalance.npy", imb)
|
||||
|
||||
# CDF
|
||||
x = np.sort(imb)
|
||||
y = np.arange(1, len(x) + 1) / len(x)
|
||||
fig, ax = plt.subplots(figsize=(6, 4))
|
||||
ax.plot(x, y, marker=".", ms=4, lw=1.4, color="#1f77b4")
|
||||
ax.axvline(1.0, color="gray", ls="--", lw=1, label="perfect balance (1.0)")
|
||||
ax.axvline(pcts[50], color="#2ca02c", ls=":", lw=1,
|
||||
label=f"p50={pcts[50]:.2f}")
|
||||
ax.axvline(pcts[90], color="#d62728", ls=":", lw=1,
|
||||
label=f"p90={pcts[90]:.2f}")
|
||||
qty = "total tokens" if args.weight == "tokens" else "num_req"
|
||||
ax.set_xlabel("per-window %s imbalance (max / mean over %d instances)"
|
||||
% (qty, args.n_inst))
|
||||
ax.set_ylabel("CDF over %d one-minute windows" % len(imb))
|
||||
routeq = {"tokens": "tokens", "input": "input-only"}.get(
|
||||
args.route_by, "num_req")
|
||||
ax.set_title("Session-sticky imbalance: measure=%s, route-by=%s"
|
||||
% (qty, routeq))
|
||||
ax.grid(True, alpha=0.3)
|
||||
ax.legend(fontsize=8)
|
||||
fig.tight_layout()
|
||||
fig.savefig(args.out_prefix + "_cdf.png", dpi=140)
|
||||
print(json.dumps(stats, indent=2))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user