From b6dfbfcad7499fea8c31da51795ac721ce59af51 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Sun, 19 Jul 2026 18:37:18 +0800 Subject: [PATCH] Analyze Qwen235 fixed PD simulator state --- .../analyze_qwen235_fixed_pd_state.py | 442 ++++++++++++++++++ .../test_fidelity_envelope.py | 36 ++ 2 files changed, 478 insertions(+) create mode 100644 runs/frontier-fidelity-envelope-v1/analyze_qwen235_fixed_pd_state.py diff --git a/runs/frontier-fidelity-envelope-v1/analyze_qwen235_fixed_pd_state.py b/runs/frontier-fidelity-envelope-v1/analyze_qwen235_fixed_pd_state.py new file mode 100644 index 0000000..35ac477 --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/analyze_qwen235_fixed_pd_state.py @@ -0,0 +1,442 @@ +#!/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[0-9.]+) tokens/s, " + r"Avg generation throughput: (?P[0-9.]+) tokens/s, " + r"Running: (?P[0-9]+) reqs, Waiting: (?P[0-9]+) reqs, " + r"GPU KV cache usage: (?P[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() diff --git a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py index d045f95..8a1de8c 100644 --- a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py +++ b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py @@ -187,6 +187,42 @@ class FidelityEnvelopeTest(unittest.TestCase): base, metrics_root=Path("/new/metrics"), run_id="state-run" ) + def test_qwen235_real_state_proxy_parser(self) -> None: + module = load("analyze_qwen235_fixed_pd_state.py") + line = ( + "Engine 000: Avg prompt throughput: 3276.8 tokens/s, " + "Avg generation throughput: 204.8 tokens/s, Running: 5 reqs, " + "Waiting: 0 reqs, GPU KV cache usage: 5.6%, Prefix cache hit rate: 0.0%\n" + ) + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary) / "server.log" + path.write_text(line) + self.assertEqual( + module.parse_real_log(path), + [ + { + "prompt_tokens_per_s": 3276.8, + "generation_tokens_per_s": 204.8, + "running": 5, + "waiting": 0, + "kv_percent": 5.6, + } + ], + ) + + def test_qwen235_component_categories_are_additive(self) -> None: + module = load("analyze_qwen235_fixed_pd_state.py") + components = { + name: 1.0 + for names in module.CATEGORIES.values() + for name in names + } + categorized = module.categorized_components(components) + self.assertEqual(sum(categorized.values()), float(len(components))) + components["unknown_graph_overhead"] = 1.0 + with self.assertRaisesRegex(ValueError, "component schema drift"): + module.categorized_components(components) + def test_materialize_qwen235_allreduce_requires_serving_contract(self) -> None: module = load("materialize_qwen235_v020_allreduce.py") with tempfile.TemporaryDirectory() as temporary: