Add validated Frontier profile assembly
This commit is contained in:
@@ -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()
|
||||
Reference in New Issue
Block a user