From 97e66ae2769d43575fdacb300974b6b5eccc89da Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Wed, 15 Jul 2026 19:04:40 +0800 Subject: [PATCH] Add validated Frontier profile assembly --- .../best_effort/assemble_profiles.py | 323 ++++++++++++++++++ 1 file changed, 323 insertions(+) create mode 100644 runs/frontier-multicase-sufficiency-v0/best_effort/assemble_profiles.py diff --git a/runs/frontier-multicase-sufficiency-v0/best_effort/assemble_profiles.py b/runs/frontier-multicase-sufficiency-v0/best_effort/assemble_profiles.py new file mode 100644 index 0000000..a9e4053 --- /dev/null +++ b/runs/frontier-multicase-sufficiency-v0/best_effort/assemble_profiles.py @@ -0,0 +1,323 @@ +#!/usr/bin/env python3 +"""Assemble and validate the Qwen3-235B Frontier best-effort profiles.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path + +import pandas as pd + + +MODEL = "Qwen3-235B-A22B-FP8" +HARDWARE = "h20" +TOKENS = [1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384] +QUANT_SIGNATURE = "method=fp8|act=dynamic|serialized=True|block=128x128" +MEASUREMENT_TYPE = "CUDA_EVENT" +PROFILING_PRECISION = "BF16" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--linear-tp4", type=Path, required=True) + parser.add_argument("--linear-tp8", type=Path, required=True) + parser.add_argument("--attention-standard", type=Path, required=True) + parser.add_argument("--attention-mixed", type=Path, required=True) + parser.add_argument("--moe-tp4-ep1", type=Path, required=True) + parser.add_argument("--moe-tp1-ep8", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + parser.add_argument( + "--frontier-commit", + default="d9cfeb6d8791fbf2f295dd9744c56a666171776e", + ) + return parser.parse_args() + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def read_csv(path: Path, label: str) -> pd.DataFrame: + if not path.is_file(): + raise FileNotFoundError(f"{label}: missing input CSV: {path}") + data = pd.read_csv(path) + if data.empty: + raise ValueError(f"{label}: input CSV is empty: {path}") + return data + + +def require_columns(data: pd.DataFrame, columns: list[str], label: str) -> None: + missing = [column for column in columns if column not in data.columns] + if missing: + raise ValueError(f"{label}: missing columns: {missing}") + + +def require_exact_values( + data: pd.DataFrame, column: str, expected: set[object], label: str +) -> None: + actual = set(data[column].dropna().unique().tolist()) + if actual != expected: + raise ValueError( + f"{label}: {column} values mismatch: expected={sorted(expected)!r}, " + f"actual={sorted(actual)!r}" + ) + + +def validate_common(data: pd.DataFrame, label: str) -> None: + columns = ["quant_signature", "profiling_precision", "measurement_type"] + require_columns(data, columns, label) + require_exact_values(data, "quant_signature", {QUANT_SIGNATURE}, label) + require_exact_values(data, "profiling_precision", {PROFILING_PRECISION}, label) + require_exact_values(data, "measurement_type", {MEASUREMENT_TYPE}, label) + + +def validate_time_columns( + data: pd.DataFrame, columns: list[str], label: str +) -> None: + require_columns(data, columns, label) + if data[columns].isna().any(axis=None): + counts = data[columns].isna().sum() + raise ValueError(f"{label}: NaN timing values: {counts[counts > 0].to_dict()}") + if (data[columns] < 0).any(axis=None): + raise ValueError(f"{label}: negative timing value") + + +def validate_token_grid( + data: pd.DataFrame, + group_columns: list[str], + expected_rows_per_point: int, + label: str, +) -> None: + for group, rows in data.groupby(group_columns, dropna=False): + counts = rows["num_tokens"].value_counts().to_dict() + expected = {token: expected_rows_per_point for token in TOKENS} + if counts != expected: + raise ValueError( + f"{label}: token grid mismatch for group={group}: " + f"expected={expected}, actual={counts}" + ) + + +def assemble_linear(tp4: pd.DataFrame, tp8: pd.DataFrame) -> pd.DataFrame: + label = "linear" + for source_label, data, expected_tp in ( + ("linear-tp4", tp4, {1, 4}), + ("linear-tp8", tp8, {1, 8}), + ): + validate_common(data, source_label) + require_columns(data, ["num_tensor_parallel_workers", "num_tokens"], source_label) + require_exact_values(data, "num_tensor_parallel_workers", expected_tp, source_label) + validate_token_grid(data, ["num_tensor_parallel_workers"], 1, source_label) + + # The TP=1 rows are replicated operators. Keep the rows from the TP4 run and + # add only the TP=8 sharded operators from the second run. + combined = pd.concat( + [tp4, tp8[tp8["num_tensor_parallel_workers"] == 8]], ignore_index=True + ) + require_exact_values(combined, "num_tensor_parallel_workers", {1, 4, 8}, label) + validate_token_grid(combined, ["num_tensor_parallel_workers"], 1, label) + + replicated = combined[combined["num_tensor_parallel_workers"] == 1] + sharded = combined[combined["num_tensor_parallel_workers"] > 1] + validate_time_columns( + replicated, + [ + "time_stats.emb.median", + "time_stats.input_layernorm.median", + "time_stats.post_attention_layernorm.median", + ], + "linear replicated operators", + ) + validate_time_columns( + sharded, + [ + "time_stats.attn_pre_proj.median", + "time_stats.attn_rope.median", + "time_stats.attn_post_proj.median", + ], + "linear sharded attention operators", + ) + return combined.sort_values( + ["num_tensor_parallel_workers", "num_tokens"], kind="stable" + ).reset_index(drop=True) + + +def assemble_attention(standard: pd.DataFrame, mixed: pd.DataFrame) -> pd.DataFrame: + timing_columns = [ + "time_stats.attn_input_reshape.median", + "time_stats.attn_kv_cache_save.median", + "time_stats.attn_prefill.median", + "time_stats.attn_decode.median", + "time_stats.attn_output_reshape.median", + ] + for label, data, expected_mixed, expected_rows in ( + ("attention-standard", standard, {False}, 390), + ("attention-mixed", mixed, {True}, 336), + ): + validate_common(data, label) + require_columns( + data, + ["num_tensor_parallel_workers", "is_prefill", "is_mixed_batch"], + label, + ) + require_exact_values(data, "num_tensor_parallel_workers", {4, 8}, label) + require_exact_values(data, "is_prefill", {True}, label) + require_exact_values(data, "is_mixed_batch", expected_mixed, label) + validate_time_columns(data, timing_columns, label) + if len(data) != expected_rows: + raise ValueError( + f"{label}: row count mismatch: expected={expected_rows}, actual={len(data)}" + ) + + combined = pd.concat([standard, mixed], ignore_index=True) + if len(combined) != 726: + raise ValueError(f"attention: expected 726 rows, got {len(combined)}") + sort_columns = [ + "num_tensor_parallel_workers", + "is_mixed_batch", + "total_tokens", + "total_prefill_tokens", + "kv_cache_size", + "batch_size", + ] + return combined.sort_values(sort_columns, kind="stable").reset_index(drop=True) + + +def assemble_moe(tp4_ep1: pd.DataFrame, tp1_ep8: pd.DataFrame) -> pd.DataFrame: + timing_columns = [ + "time_stats.moe_gating_linear.median", + "time_stats.moe_gating_routing_topk.median", + "time_stats.moe_shuffling.median", + "time_stats.moe_grouped_gemm.median", + ] + cases = ( + ("moe-tp4-ep1", tp4_ep1, 4, 1, 128), + ("moe-tp1-ep8", tp1_ep8, 1, 8, 16), + ) + for label, data, expected_tp, expected_ep, expected_local_experts in cases: + validate_common(data, label) + require_columns( + data, + [ + "num_tensor_parallel_workers", + "expert_parallel_size", + "num_experts_per_device", + "num_tokens", + "load_distribution", + "seed", + "routing_runtime_path", + "gating_runtime_context", + ], + label, + ) + require_exact_values(data, "num_tensor_parallel_workers", {expected_tp}, label) + require_exact_values(data, "expert_parallel_size", {expected_ep}, label) + require_exact_values(data, "num_experts_per_device", {expected_local_experts}, label) + require_exact_values( + data, + "load_distribution", + {"uniform", "skewed", "extremely_skewed"}, + label, + ) + require_exact_values(data, "seed", {0, 1}, label) + require_exact_values(data, "routing_runtime_path", {"standard_fused_topk"}, label) + require_exact_values(data, "gating_runtime_context", {"prefill_hot"}, label) + validate_token_grid( + data, + ["num_tensor_parallel_workers", "expert_parallel_size"], + 6, + label, + ) + validate_time_columns(data, timing_columns, label) + if len(data) != 90: + raise ValueError(f"{label}: expected 90 rows, got {len(data)}") + + combined = pd.concat([tp4_ep1, tp1_ep8], ignore_index=True) + return combined.sort_values( + [ + "num_tensor_parallel_workers", + "expert_parallel_size", + "num_tokens", + "load_distribution", + "seed", + ], + kind="stable", + ).reset_index(drop=True) + + +def main() -> None: + args = parse_args() + input_paths = { + "linear_tp4": args.linear_tp4, + "linear_tp8": args.linear_tp8, + "attention_standard": args.attention_standard, + "attention_mixed": args.attention_mixed, + "moe_tp4_ep1": args.moe_tp4_ep1, + "moe_tp1_ep8": args.moe_tp1_ep8, + } + inputs = {name: read_csv(path, name) for name, path in input_paths.items()} + + outputs = { + "linear_op.csv": assemble_linear(inputs["linear_tp4"], inputs["linear_tp8"]), + "attention.csv": assemble_attention( + inputs["attention_standard"], inputs["attention_mixed"] + ), + "moe.csv": assemble_moe(inputs["moe_tp4_ep1"], inputs["moe_tp1_ep8"]), + } + + output_dir = args.output_root / "compute" / HARDWARE / MODEL + output_dir.mkdir(parents=True, exist_ok=True) + output_paths: dict[str, Path] = {} + for name, data in outputs.items(): + path = output_dir / name + data.to_csv(path, index=False) + output_paths[name] = path + + manifest = { + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "frontier_commit": args.frontier_commit, + "model": MODEL, + "hardware": HARDWARE, + "contract": { + "quant_signature": QUANT_SIGNATURE, + "profiling_precision": PROFILING_PRECISION, + "profiling_precision_semantics": "BF16 kernel output/accumulation; weights and activations are block FP8 W8A8", + "measurement_type": MEASUREMENT_TYPE, + "linear_tensor_parallel_sizes": [1, 4, 8], + "attention_tensor_parallel_sizes": [4, 8], + "moe_layouts": [ + {"tensor_parallel_size": 4, "expert_parallel_size": 1}, + {"tensor_parallel_size": 1, "expert_parallel_size": 8}, + ], + }, + "inputs": { + name: { + "path": str(path.resolve()), + "sha256": sha256(path), + "rows": len(inputs[name]), + } + for name, path in input_paths.items() + }, + "outputs": { + name: { + "path": str(path.resolve()), + "sha256": sha256(path), + "rows": len(outputs[name]), + } + for name, path in output_paths.items() + }, + } + manifest_path = args.output_root / "profile_manifest.json" + manifest_path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") + print(json.dumps(manifest, indent=2)) + + +if __name__ == "__main__": + main()