443 lines
16 KiB
Python
443 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Analyze Qwen235 Fixed-PD real-state proxies and Frontier component ledgers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
from pathlib import Path
|
|
from statistics import fmean
|
|
from typing import Any, Iterable
|
|
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
REPO_ROOT = HERE.parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "runs/telemetry-residual"))
|
|
|
|
from common_state import load_jsonl, numeric # noqa: E402
|
|
|
|
|
|
CONFIGS = ("tp4_ep1_mns64", "tp8_ep8_mns64")
|
|
TP_BY_CONFIG = {"tp4_ep1_mns64": 4, "tp8_ep8_mns64": 8}
|
|
REAL_TPOT_MS = {"tp4_ep1_mns64": 21.0442, "tp8_ep8_mns64": 27.9942}
|
|
LOG_PATTERN = re.compile(
|
|
r"Avg prompt throughput: (?P<prompt>[0-9.]+) tokens/s, "
|
|
r"Avg generation throughput: (?P<generation>[0-9.]+) tokens/s, "
|
|
r"Running: (?P<running>[0-9]+) reqs, Waiting: (?P<waiting>[0-9]+) reqs, "
|
|
r"GPU KV cache usage: (?P<kv>[0-9.]+)%"
|
|
)
|
|
|
|
|
|
CATEGORIES = {
|
|
"attention": {
|
|
"attention_prefill_execution_time",
|
|
"attention_decode_execution_time",
|
|
"attention_pre_proj_time",
|
|
"attention_post_proj_time",
|
|
"attention_kv_cache_save_execution_time",
|
|
"attention_rope_execution_time",
|
|
"attn_norm_time",
|
|
},
|
|
"dense_mlp_compute": {
|
|
"mlp_layer_up_proj_execution_time",
|
|
"mlp_layer_act_execution_time",
|
|
"mlp_layer_down_proj_execution_time",
|
|
"mlp_norm_time",
|
|
},
|
|
"moe_compute": {
|
|
"moe_grouped_gemm_time",
|
|
"share_expert_up_proj_time",
|
|
"share_expert_act_time",
|
|
"share_expert_down_proj_time",
|
|
},
|
|
"moe_routing": {
|
|
"moe_gating_linear_time",
|
|
"moe_gating_routing_topk_time",
|
|
"moe_shuffling_time",
|
|
},
|
|
"ep_communication": {"expert_parallel_communication_time"},
|
|
"tp_dp_communication": {
|
|
"attention_all_reduce_time",
|
|
"mlp_all_reduce_time",
|
|
"moe_tensor_parallel_allgather_time",
|
|
"share_expert_tensor_parallel_allreduce_time",
|
|
"dp_input_allreduce_time",
|
|
"dp_output_allreduce_time",
|
|
},
|
|
"pipeline_communication": {"pipeline_parallel_communication_time"},
|
|
"runtime_overhead": {
|
|
"add_attn_residual_time",
|
|
"add_ffn_residual_time",
|
|
"schedule_time",
|
|
"sampler_e2e_time",
|
|
"prepare_inputs_e2e_time",
|
|
"pp_producer_send_path_runtime_time",
|
|
"pp_receiver_head_runtime_time",
|
|
"pp_prefill_consumer_active_runtime_time",
|
|
"pp_stage_boundary_residual_runtime_time",
|
|
"process_model_outputs_time",
|
|
"ray_comm_time",
|
|
"decode_draft_proposer_time",
|
|
"mtp_terminal_overshoot_time",
|
|
},
|
|
}
|
|
|
|
|
|
def atomic_write(path: Path, text: str) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(text)
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def parse_real_log(path: Path) -> list[dict[str, float | int]]:
|
|
rows = []
|
|
for line in path.read_text(errors="replace").splitlines():
|
|
match = LOG_PATTERN.search(line)
|
|
if match is None:
|
|
continue
|
|
rows.append(
|
|
{
|
|
"prompt_tokens_per_s": float(match.group("prompt")),
|
|
"generation_tokens_per_s": float(match.group("generation")),
|
|
"running": int(match.group("running")),
|
|
"waiting": int(match.group("waiting")),
|
|
"kv_percent": float(match.group("kv")),
|
|
}
|
|
)
|
|
if not rows:
|
|
raise ValueError(f"no vLLM periodic metrics found: {path}")
|
|
return rows
|
|
|
|
|
|
def summarize_real(real_root: Path) -> dict[str, Any]:
|
|
result = {}
|
|
for config in CONFIGS:
|
|
tp = TP_BY_CONFIG[config]
|
|
expected_prompt = 4096 * 0.2 * tp
|
|
trials = []
|
|
steady_all = []
|
|
for trial in ("trial1", "trial2", "trial3"):
|
|
path = real_root / config / trial / "logs/server.log"
|
|
rows = parse_real_log(path)
|
|
steady = [
|
|
row
|
|
for row in rows
|
|
if row["prompt_tokens_per_s"] >= 0.95 * expected_prompt
|
|
and row["generation_tokens_per_s"] > 0
|
|
]
|
|
if len(steady) < 5:
|
|
raise ValueError(f"insufficient steady real proxy samples: {path}")
|
|
steady_all.extend(steady)
|
|
trials.append(
|
|
{
|
|
"trial": trial,
|
|
"path": str(path.resolve()),
|
|
"all_samples": len(rows),
|
|
"steady_samples": len(steady),
|
|
"running": numeric(row["running"] for row in steady),
|
|
"waiting": numeric(row["waiting"] for row in steady),
|
|
"kv_percent": numeric(row["kv_percent"] for row in steady),
|
|
"generation_tokens_per_s": numeric(
|
|
row["generation_tokens_per_s"] for row in steady
|
|
),
|
|
}
|
|
)
|
|
running_counts = Counter(int(row["running"]) for row in steady_all)
|
|
result[config] = {
|
|
"proxy_only": True,
|
|
"steady_rule": f"prompt throughput >= 95% of {expected_prompt:.1f} tokens/s",
|
|
"trials": trials,
|
|
"aggregate": {
|
|
"samples": len(steady_all),
|
|
"running": numeric(row["running"] for row in steady_all),
|
|
"running_histogram": {
|
|
str(key): value for key, value in sorted(running_counts.items())
|
|
},
|
|
"waiting": numeric(row["waiting"] for row in steady_all),
|
|
"kv_percent": numeric(row["kv_percent"] for row in steady_all),
|
|
"generation_tokens_per_s": numeric(
|
|
row["generation_tokens_per_s"] for row in steady_all
|
|
),
|
|
},
|
|
}
|
|
return result
|
|
|
|
|
|
def categorized_components(components: dict[str, Any]) -> dict[str, float]:
|
|
covered = set().union(*CATEGORIES.values())
|
|
unknown = set(components) - covered
|
|
missing = covered - set(components)
|
|
if unknown or missing:
|
|
raise ValueError(
|
|
f"component schema drift: unknown={sorted(unknown)}, missing={sorted(missing)}"
|
|
)
|
|
return {
|
|
category: sum(float(components[name]) for name in names)
|
|
for category, names in CATEGORIES.items()
|
|
}
|
|
|
|
|
|
def load_decode_rows(state_root: Path, config: str) -> tuple[list[dict[str, Any]], dict]:
|
|
result_path = state_root / config / "result.json"
|
|
result = json.loads(result_path.read_text())
|
|
if result.get("status") != "PASS":
|
|
raise ValueError(f"state replay did not pass: {result_path}")
|
|
if not result["scorer_equivalence"]["request_metrics_byte_identical"]:
|
|
raise ValueError(f"state replay scorer changed: {result_path}")
|
|
ledger = Path(result["state_artifacts"]["ledger"]["path"])
|
|
rows = []
|
|
for row in load_jsonl(ledger):
|
|
token_counts = [int(value) for value in row["request_num_tokens"]]
|
|
if not token_counts or any(value != 1 for value in token_counts):
|
|
continue
|
|
components = categorized_components(
|
|
row["execution_time"]["component_ledger_ms"]
|
|
)
|
|
total = float(row["execution_time"]["total_time_ms"])
|
|
if not math.isclose(sum(components.values()), total, abs_tol=1e-6):
|
|
raise ValueError(f"categorized components do not sum for {config}")
|
|
rows.append(
|
|
{
|
|
"batch_size": len(row["request_ids"]),
|
|
"total_time_ms": total,
|
|
"categories_ms": components,
|
|
"per_expert_tokens": row.get("per_expert_tokens"),
|
|
}
|
|
)
|
|
if not rows:
|
|
raise ValueError(f"no decode-only ledger rows for {config}")
|
|
return rows, result
|
|
|
|
|
|
def weighted_component_mean(rows: Iterable[dict[str, Any]]) -> dict[str, float]:
|
|
selected = list(rows)
|
|
if not selected:
|
|
raise ValueError("component mean needs rows")
|
|
weights = [row["batch_size"] for row in selected]
|
|
denominator = sum(weights)
|
|
values = {
|
|
category: sum(
|
|
row["categories_ms"][category] * weight
|
|
for row, weight in zip(selected, weights, strict=True)
|
|
)
|
|
/ denominator
|
|
for category in CATEGORIES
|
|
}
|
|
values["total"] = sum(values.values())
|
|
return values
|
|
|
|
|
|
def means_by_batch(rows: list[dict[str, Any]]) -> dict[int, dict[str, Any]]:
|
|
groups: dict[int, list[dict[str, Any]]] = defaultdict(list)
|
|
for row in rows:
|
|
groups[row["batch_size"]].append(row)
|
|
return {
|
|
batch_size: {
|
|
"n": len(group),
|
|
"components_ms": weighted_component_mean(group),
|
|
}
|
|
for batch_size, group in sorted(groups.items())
|
|
}
|
|
|
|
|
|
def reweight(
|
|
by_batch: dict[int, dict[str, Any]], histogram: dict[str, int]
|
|
) -> dict[str, Any]:
|
|
total_samples = sum(histogram.values())
|
|
supported = {
|
|
int(batch_size): count
|
|
for batch_size, count in histogram.items()
|
|
if int(batch_size) in by_batch
|
|
}
|
|
covered = sum(supported.values())
|
|
if covered == 0:
|
|
return {"coverage": 0.0, "components_ms": None, "unsupported": histogram}
|
|
values = {
|
|
category: sum(
|
|
by_batch[batch_size]["components_ms"][category] * count
|
|
for batch_size, count in supported.items()
|
|
)
|
|
/ covered
|
|
for category in (*CATEGORIES, "total")
|
|
}
|
|
return {
|
|
"coverage": covered / total_samples,
|
|
"supported_samples": covered,
|
|
"total_samples": total_samples,
|
|
"components_ms": values,
|
|
"unsupported": {
|
|
batch_size: count
|
|
for batch_size, count in histogram.items()
|
|
if int(batch_size) not in by_batch
|
|
},
|
|
}
|
|
|
|
|
|
def subtract(right: dict[str, float], left: dict[str, float]) -> dict[str, float]:
|
|
return {name: right[name] - left[name] for name in left}
|
|
|
|
|
|
def analyze(real_root: Path, state_root: Path) -> dict[str, Any]:
|
|
real = summarize_real(real_root)
|
|
sim = {}
|
|
by_batch = {}
|
|
for config in CONFIGS:
|
|
rows, replay = load_decode_rows(state_root, config)
|
|
grouped = means_by_batch(rows)
|
|
by_batch[config] = grouped
|
|
sim[config] = {
|
|
"decode_only_rows": len(rows),
|
|
"decode_batch_size": numeric(row["batch_size"] for row in rows),
|
|
"token_weighted_components_ms": weighted_component_mean(rows),
|
|
"batch_support": {
|
|
str(batch): value for batch, value in grouped.items()
|
|
},
|
|
"scorer_equivalence": replay["scorer_equivalence"],
|
|
}
|
|
|
|
real_reweighted = {
|
|
config: reweight(
|
|
by_batch[config], real[config]["aggregate"]["running_histogram"]
|
|
)
|
|
for config in CONFIGS
|
|
}
|
|
coverages = [real_reweighted[config]["coverage"] for config in CONFIGS]
|
|
proxy_contrast = None
|
|
if min(coverages) >= 0.8:
|
|
proxy_contrast = subtract(
|
|
real_reweighted[CONFIGS[1]]["components_ms"],
|
|
real_reweighted[CONFIGS[0]]["components_ms"],
|
|
)
|
|
|
|
shared = sorted(set(by_batch[CONFIGS[0]]) & set(by_batch[CONFIGS[1]]))
|
|
shared_contrasts = {
|
|
str(batch): {
|
|
"tp4_n": by_batch[CONFIGS[0]][batch]["n"],
|
|
"tp8_n": by_batch[CONFIGS[1]][batch]["n"],
|
|
"tp8_minus_tp4_ms": subtract(
|
|
by_batch[CONFIGS[1]][batch]["components_ms"],
|
|
by_batch[CONFIGS[0]][batch]["components_ms"],
|
|
),
|
|
}
|
|
for batch in shared
|
|
if min(by_batch[config][batch]["n"] for config in CONFIGS) >= 10
|
|
}
|
|
observed_real_contrast = REAL_TPOT_MS[CONFIGS[1]] - REAL_TPOT_MS[CONFIGS[0]]
|
|
if proxy_contrast is None:
|
|
verdict = "STOP: real Running proxy has insufficient exact simulator support"
|
|
elif proxy_contrast["total"] < 0:
|
|
verdict = (
|
|
"Active-batch-count mismatch alone is insufficient: after exact reweighting "
|
|
"to each config's real Running histogram, Frontier still predicts TP8 faster."
|
|
)
|
|
else:
|
|
verdict = (
|
|
"Coarse active-batch state can flip Frontier's ordering, but iteration-level "
|
|
"batch/context telemetry is required before attributing the real gap to state."
|
|
)
|
|
return {
|
|
"schema": "qwen235-fixed-pd-state-diagnosis-v1",
|
|
"status": "PASS",
|
|
"scope": {
|
|
"workload": "Fixed-PD 4096->256, 0.2 req/s/GPU, MNS64",
|
|
"real_proxy_limitation": (
|
|
"vLLM 10-second Running is active requests, not per-iteration decode batch; "
|
|
"context lengths and graph buckets are unavailable in frozen logs"
|
|
),
|
|
"graph_observability": (
|
|
"Frontier ledger exposes additive execution components but no direct graph "
|
|
"bucket/padding/launch-overhead field; graph effects remain folded into predictors"
|
|
),
|
|
},
|
|
"real": real,
|
|
"simulator": sim,
|
|
"proxy_matched": {
|
|
"method": (
|
|
"exact batch-size lookup; each simulator config is reweighted to its own "
|
|
"frozen-real steady Running histogram; no interpolation"
|
|
),
|
|
"configs": real_reweighted,
|
|
"tp8_minus_tp4_ms": proxy_contrast,
|
|
},
|
|
"same_batch_contrasts": shared_contrasts,
|
|
"reference": {
|
|
"real_tpot_ms": REAL_TPOT_MS,
|
|
"observed_real_tp8_minus_tp4_ms": observed_real_contrast,
|
|
},
|
|
"verdict": verdict,
|
|
}
|
|
|
|
|
|
def markdown(result: dict[str, Any]) -> str:
|
|
lines = [
|
|
"# Qwen235 Fixed-PD state diagnosis",
|
|
"",
|
|
f"**Verdict:** {result['verdict']}",
|
|
"",
|
|
"| Config | Real Running proxy mean | Sim decode batch mean | Proxy coverage |",
|
|
"|---|---:|---:|---:|",
|
|
]
|
|
for config in CONFIGS:
|
|
lines.append(
|
|
f"| {config} | {result['real'][config]['aggregate']['running']['mean']:.3f} "
|
|
f"| {result['simulator'][config]['decode_batch_size']['mean']:.3f} "
|
|
f"| {result['proxy_matched']['configs'][config]['coverage']:.1%} |"
|
|
)
|
|
contrast = result["proxy_matched"]["tp8_minus_tp4_ms"]
|
|
if contrast is not None:
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Frontier internal component contrast at real Running proxy",
|
|
"",
|
|
"Positive means TP8 slower; negative means Frontier gives TP8 an advantage.",
|
|
"",
|
|
"| Component | TP8 - TP4 (ms/step) |",
|
|
"|---|---:|",
|
|
]
|
|
)
|
|
for name, value in sorted(
|
|
contrast.items(), key=lambda item: abs(item[1]), reverse=True
|
|
):
|
|
lines.append(f"| {name} | {value:+.4f} |")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Interpretation boundary",
|
|
"",
|
|
f"- Real observed TPOT contrast: {result['reference']['observed_real_tp8_minus_tp4_ms']:+.4f} ms/token.",
|
|
f"- {result['scope']['real_proxy_limitation']}.",
|
|
f"- {result['scope']['graph_observability']}.",
|
|
"- Component deltas identify where Frontier creates its own TP8 advantage; without real per-stage measurements they are not yet root-cause proof.",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--real-root", type=Path, required=True)
|
|
parser.add_argument("--state-root", type=Path, required=True)
|
|
parser.add_argument("--json-output", type=Path, required=True)
|
|
parser.add_argument("--markdown-output", type=Path, required=True)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
result = analyze(args.real_root.resolve(), args.state_root.resolve())
|
|
atomic_write(args.json_output, json.dumps(result, indent=2, sort_keys=True) + "\n")
|
|
atomic_write(args.markdown_output, markdown(result))
|
|
print(json.dumps({"status": result["status"], "verdict": result["verdict"]}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|