#!/usr/bin/env python3 """Freeze Frontier's Qwen30 exact production-trace response surface.""" from __future__ import annotations import argparse import csv import importlib.util import json import math import os import subprocess import sys import time from pathlib import Path from typing import Any TARGET_PASS_RATE = 0.95 TPOT_SLOS_MS = (50.0, 100.0, 150.0, 180.0) WINDOW_SECONDS = 600.0 GRAPH_CAPTURE_SIZES_BY_MNS = { 8: (1, 2, 4, 8, 16), 16: (1, 2, 4, 8, 16, 24, 32), 32: (1, 2, 4, 8, 16, 24, 32, 40, 48, 56, 64), 64: (1, 2, 4, 8, 16, 24, 32, 40, 48, 56, 64, 72, 80, 88, 96, 104, 112, 120, 128), } REAL_NUM_BLOCKS_BY_CONFIG = { (1, 8): 20137, (1, 16): 20128, (1, 32): 20108, (1, 64): 20069, (2, 8): 76639, (2, 16): 76620, (2, 32): 76583, (2, 64): 76505, (4, 8): 191930, (4, 16): 191882, (4, 32): 191786, (4, 64): 191589, } KERNEL_DECODE_KV_CONTEXTS = (128, 1024, 2048, 4096, 8192, 16384, 32768, 40960) BASE_RUNNER = ( Path(__file__).resolve().parents[1] / "frontier-phase-factorial-v0/run_frontier_qwen30_prefill_surface.py" ) def load_base(): spec = importlib.util.spec_from_file_location("qwen30_prefill_surface_base", BASE_RUNNER) if spec is None or spec.loader is None: raise ImportError(BASE_RUNNER) module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module BASE = load_base() def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--frontier-source", type=Path, required=True) parser.add_argument("--replayserve-root", type=Path, required=True) parser.add_argument("--profile-root", type=Path, required=True) parser.add_argument("--kernel-profile-root", type=Path) parser.add_argument("--python-deps", type=Path, required=True) parser.add_argument("--output-root", type=Path, required=True) parser.add_argument( "--trace", action="append", required=True, help="Frozen trace anchor as LABEL=PATH; repeat in increasing load order.", ) parser.add_argument("--config", action="append") parser.add_argument( "--rate-contract", choices=("trace-window", "uniform-spacing"), default="trace-window", ) parser.add_argument( "--prefix-caching", action=argparse.BooleanOptionalAction, default=True, ) parser.add_argument( "--cc-backend", choices=("analytical", "vidur"), default="vidur" ) parser.add_argument("--allreduce-csv", type=Path) parser.add_argument("--timeout-seconds", type=float, default=1800.0) parser.add_argument("--predictor-training-job-threads", type=int, default=1) parser.add_argument( "--decode-cuda-graph-mode", choices=("none", "full_decode_only", "piecewise"), default="none", ) parser.add_argument( "--align-real-graph-runtime", action="store_true", help="Use real observed capture lists and per-(TP,MNS) KV blocks.", ) parser.add_argument( "--fresh-predictor-cache", action="store_true", help="Disable Frontier predictor cache reuse for this profile family.", ) parser.add_argument("--resume", action="store_true") parser.add_argument("--continue-on-failure", action="store_true") return parser.parse_args() def ttft_limit_ms(input_tokens: int) -> float: return 1000.0 + 1000.0 * input_tokens / 8000.0 def percentile(values: list[float], fraction: float) -> float | None: if not values: return None ordered = sorted(values) return ordered[math.ceil(fraction * len(ordered)) - 1] def manifest_offered_rate(path: Path) -> tuple[float | None, str | None]: manifest_path = path.with_name("manifest.json") if not manifest_path.is_file(): return None, None manifest = json.loads(manifest_path.read_text()) if manifest.get("public_csv") != str(path): return None, None rate = manifest.get("global_offered_request_rate") if not isinstance(rate, (int, float)) or not math.isfinite(rate) or rate <= 0: raise ValueError(f"invalid global_offered_request_rate in {manifest_path}") return float(rate), str(manifest_path) def parse_trace( specification: str, *, rate_contract: str = "trace-window", prefix_caching: bool ) -> dict[str, Any]: if "=" not in specification: raise ValueError(f"trace must be LABEL=PATH: {specification}") label, raw_path = specification.split("=", 1) if not label or "/" in label: raise ValueError(f"invalid trace label: {label!r}") path = Path(raw_path).resolve() if not path.is_file(): raise FileNotFoundError(path) with path.open(newline="") as source: rows = list(csv.DictReader(source)) if not rows: raise ValueError(f"empty trace: {path}") required = { "arrived_at", "num_prefill_tokens", "num_decode_tokens", "session_id", "block_hash_ids", } if not required.issubset(rows[0]): raise ValueError(f"trace columns missing: {required - set(rows[0])}") arrivals = [float(row["arrived_at"]) for row in rows] if any(not math.isfinite(value) or value < 0 for value in arrivals): raise ValueError(f"invalid arrival in {path}") if any(right < left for left, right in zip(arrivals, arrivals[1:])): raise ValueError(f"arrival order drift in {path}") if rate_contract == "trace-window": offered_request_rate, rate_manifest = manifest_offered_rate(path) if offered_request_rate is None: offered_request_rate = len(rows) / WINDOW_SECONDS rate_source = "legacy_fixed_600_second_window" else: rate_source = f"manifest:{rate_manifest}" elif rate_contract == "uniform-spacing": if len(rows) < 2 or arrivals[-1] <= arrivals[0]: raise ValueError("uniform-spacing traces require at least two arrivals") intervals = [right - left for left, right in zip(arrivals, arrivals[1:])] expected_interval = (arrivals[-1] - arrivals[0]) / (len(arrivals) - 1) if any(abs(value - expected_interval) > 1e-9 for value in intervals): raise ValueError(f"non-uniform fixed trace: {path}") offered_request_rate = 1.0 / expected_interval rate_source = "uniform_spacing" else: raise ValueError(f"unknown rate contract: {rate_contract}") shapes = [ (int(row["num_prefill_tokens"]), int(row["num_decode_tokens"])) for row in rows ] if any(prefill <= 0 or decode <= 0 or prefill + decode > 40960 for prefill, decode in shapes): raise ValueError(f"out-of-contract shape in {path}") if prefix_caching: for index, (row, (prefill, _)) in enumerate(zip(rows, shapes, strict=True)): block_ids = [value for value in row["block_hash_ids"].split("|") if value] if len(block_ids) != prefill // 16: raise ValueError( "Frontier prefix-cache trace must expose only complete " f"16-token blocks: row={index}, ISL={prefill}, " f"ids={len(block_ids)}, expected={prefill // 16}" ) return { "label": label, "path": path, "sha256": BASE.sha256(path), "requests": len(rows), "offered_request_rate": offered_request_rate, "offered_request_rate_source": rate_source, "first_arrival_s": arrivals[0], "last_arrival_s": arrivals[-1], "shapes": shapes, } def find_request_metrics(run_dir: Path) -> Path: matches = list((run_dir / "metrics").rglob("request_metrics.csv")) if len(matches) != 1: raise RuntimeError(f"expected one request_metrics.csv, got {matches}") return matches[0] def trace_manifest_entry(trace: dict[str, Any]) -> dict[str, Any]: return { key: str(value) if isinstance(value, Path) else value for key, value in trace.items() if key != "shapes" } def classify_frontier_failure(stderr: str) -> str: if "Sequential simulation ended with non-empty scheduler state" in stderr: return "scheduler_stall" return "frontier_error" def score(path: Path, expected_shapes: list[tuple[int, int]]) -> dict[str, Any]: with path.open(newline="") as source: rows = list(csv.DictReader(source)) if len(rows) != len(expected_shapes): raise ValueError(f"request count mismatch: {len(rows)} != {len(expected_shapes)}") rows.sort(key=lambda row: int(row["Request Id"])) request_metrics = [] for index, (row, expected) in enumerate(zip(rows, expected_shapes, strict=True)): if int(row["Request Id"]) != index: raise ValueError("Frontier request ID/order drift") shape = ( int(float(row["request_num_prefill_tokens"])), int(float(row["request_num_decode_tokens"])), ) if shape != expected: raise ValueError(f"request shape drift at {index}: {shape} != {expected}") ttft = float(row["ttft"]) e2e = float(row["request_e2e_time"]) tpot = (e2e - ttft) / (shape[1] - 1) if shape[1] > 1 else None values = [ttft, e2e] + ([] if tpot is None else [tpot]) if not all(math.isfinite(value) and value >= 0 for value in values): raise ValueError(f"invalid latency at request {index}") request_metrics.append( { "request_id": index, "input_tokens": shape[0], "output_tokens": shape[1], "ttft_ms": ttft, "ttft_limit_ms": ttft_limit_ms(shape[0]), "tpot_ms": tpot, "e2e_ms": e2e, } ) slos = {} for limit in TPOT_SLOS_MS: passed = sum( row["ttft_ms"] <= row["ttft_limit_ms"] and (row["tpot_ms"] is None or row["tpot_ms"] <= limit) for row in request_metrics ) pass_rate = passed / len(request_metrics) slos[f"tpot_{int(limit)}ms"] = { "passed": passed, "pass_rate": pass_rate, "feasible": pass_rate >= TARGET_PASS_RATE, } ttfts = [float(row["ttft_ms"]) for row in request_metrics] tpots = [float(row["tpot_ms"]) for row in request_metrics if row["tpot_ms"] is not None] e2es = [float(row["e2e_ms"]) for row in request_metrics] return { "ttft_mean_ms": sum(ttfts) / len(ttfts), "ttft_p50_ms": percentile(ttfts, 0.50), "ttft_p90_ms": percentile(ttfts, 0.90), "ttft_p95_ms": percentile(ttfts, 0.95), "tpot_mean_ms": sum(tpots) / len(tpots), "tpot_p50_ms": percentile(tpots, 0.50), "tpot_p90_ms": percentile(tpots, 0.90), "tpot_p95_ms": percentile(tpots, 0.95), "e2e_mean_ms": sum(e2es) / len(e2es), "e2e_p50_ms": percentile(e2es, 0.50), "e2e_p90_ms": percentile(e2es, 0.90), "e2e_p95_ms": percentile(e2es, 0.95), "slos": slos, } def kernel_profile_paths(root: Path) -> dict[str, Path]: paths = { "linear": root / "linear_op.csv", "attention": root / "attention.csv", "moe": root / "moe.csv", "manifest": root / "manifest.json", } missing = [str(path) for path in paths.values() if not path.is_file()] if missing: raise FileNotFoundError(missing) return paths def validate_kernel_profile(paths: dict[str, Path]) -> dict[str, Any]: manifest = json.loads(paths["manifest"].read_text()) outputs = manifest.get("outputs", {}) for filename, name in ( ("linear_op.csv", "linear"), ("attention.csv", "attention"), ("moe.csv", "moe"), ): if outputs.get(filename) != BASE.sha256(paths[name]): raise ValueError(f"kernel-only profile hash mismatch for {filename}") with paths["linear"].open(newline="") as source: linear_rows = list(csv.DictReader(source)) with paths["attention"].open(newline="") as source: attention_rows = list(csv.DictReader(source)) with paths["moe"].open(newline="") as source: moe_rows = list(csv.DictReader(source)) for label, rows in (("linear", linear_rows), ("attention", attention_rows), ("moe", moe_rows)): if not rows or {row.get("measurement_type") for row in rows} != {"KERNEL_ONLY"}: raise ValueError(f"{label} lacks an exclusive KERNEL_ONLY measurement family") required_buckets = set(GRAPH_CAPTURE_SIZES_BY_MNS[64]) coverage: dict[str, Any] = {} for tp in (1, 2, 4): linear_tokens = { int(float(row["num_tokens"])) for row in linear_rows if int(float(row["num_tensor_parallel_workers"])) == tp } moe_tokens = { int(float(row["num_tokens"])) for row in moe_rows if int(float(row["num_tensor_parallel_workers"])) == tp } attention_pairs = { (int(float(row["batch_size"])), int(float(row["kv_cache_size"]))) for row in attention_rows if int(float(row["num_tensor_parallel_workers"])) == tp and row["is_prefill"].lower() == "false" and row.get("is_true_mixed_batch", "").lower() != "true" } missing_linear = required_buckets - linear_tokens missing_moe = required_buckets - moe_tokens missing_attention = { (bucket, kv) for bucket in required_buckets for kv in KERNEL_DECODE_KV_CONTEXTS if (bucket, kv) not in attention_pairs } if missing_linear or missing_moe or missing_attention: raise ValueError( f"kernel-only profile coverage TP{tp}: linear={sorted(missing_linear)}, " f"moe={sorted(missing_moe)}, attention={sorted(missing_attention)}" ) coverage[str(tp)] = { "linear_tokens": sorted(linear_tokens), "moe_tokens": sorted(moe_tokens), "attention_decode_pairs": len(attention_pairs), } return {"manifest": manifest, "coverage": coverage} def main() -> None: args = parse_args() if args.predictor_training_job_threads <= 0: raise ValueError("predictor training job threads must be positive") for name in ( "frontier_source", "replayserve_root", "profile_root", "python_deps", "output_root", ): setattr(args, name, getattr(args, name).resolve()) if args.allreduce_csv is not None: args.allreduce_csv = args.allreduce_csv.resolve() if args.kernel_profile_root is not None: args.kernel_profile_root = args.kernel_profile_root.resolve() if args.decode_cuda_graph_mode == "none": raise ValueError("--kernel-profile-root requires a non-none graph mode") traces = [ parse_trace( specification, rate_contract=args.rate_contract, prefix_caching=args.prefix_caching, ) for specification in args.trace ] if len({trace["label"] for trace in traces}) != len(traces): raise ValueError("trace labels must be unique") if any( right["offered_request_rate"] <= left["offered_request_rate"] for left, right in zip(traces, traces[1:]) ): raise ValueError("trace anchors must be supplied in increasing load order") selected = list(BASE.GRID) if args.config: wanted = set(args.config) selected = [config for config in BASE.GRID if config.name in wanted] if {config.name for config in selected} != wanted: raise ValueError(f"unknown configs: {wanted - {config.name for config in selected}}") paths = BASE.profile_paths(args.profile_root) coverage = BASE.validate_profile(paths) kernel_paths = None kernel_coverage = None if args.kernel_profile_root is not None: kernel_paths = kernel_profile_paths(args.kernel_profile_root) kernel_coverage = validate_kernel_profile(kernel_paths) builder = BASE.load_module( "qwen30_exact_trace_frontier_builder", args.replayserve_root / "tools/run_frontier_sweep.py", ) frontier_head = subprocess.run( ["git", "-C", str(args.frontier_source), "rev-parse", "HEAD"], check=True, text=True, stdout=subprocess.PIPE, ).stdout.strip() environment = os.environ.copy() pythonpath = [str(args.python_deps), str(args.frontier_source)] if environment.get("PYTHONPATH"): pythonpath.append(environment["PYTHONPATH"]) environment.update( { "PYTHONPATH": ":".join(pythonpath), "CUDA_VISIBLE_DEVICES": "", "NVIDIA_VISIBLE_DEVICES": "void", "WANDB_DISABLED": "true", "VIDUR_DISABLE_WANDB": "1", "FRONTIER_LOG_LEVEL": "WARNING", "PYTHONDONTWRITEBYTECODE": "1", } ) config_results = [] for config in selected: loads = [] config_knobs = BASE.knobs(config, paths, args.output_root / "cache") config_knobs["enable_prefix_caching"] = args.prefix_caching config_knobs["prediction_max_tokens_per_request"] = 40960 config_knobs["decode_cuda_graph_mode"] = args.decode_cuda_graph_mode config_knobs["no_cache"] = args.fresh_predictor_cache if args.align_real_graph_runtime: config_knobs["num_blocks"] = REAL_NUM_BLOCKS_BY_CONFIG[(config.tp, config.mns)] if kernel_paths is not None: config_knobs.update( { "linear_op_kernel_only_input_file": str(kernel_paths["linear"]), "atten_kernel_only_input_file": str(kernel_paths["attention"]), "moe_kernel_only_input_file": str(kernel_paths["moe"]), } ) for trace in traces: run_dir = args.output_root / "runs" / config.name / trace["label"] result_path = run_dir / "result.json" if args.resume and result_path.is_file(): loads.append(json.loads(result_path.read_text())) continue run_dir.mkdir(parents=True, exist_ok=True) command = builder.build_frontier_command( python_bin="/usr/bin/python3", trace_file=trace["path"], metrics_root=run_dir / "metrics", run_id=f"qwen30_trace_{config.name}_{trace['label']}", knobs=config_knobs, ) command.extend( [ "--random_forrest_execution_time_predictor_config_num_training_job_threads", str(args.predictor_training_job_threads), ] ) if args.align_real_graph_runtime: command.extend( [ "--cudagraph_capture_sizes", *(str(size) for size in GRAPH_CAPTURE_SIZES_BY_MNS[config.mns]), ] ) command = BASE.configure_cc_command( command, backend=args.cc_backend, allreduce_csv=args.allreduce_csv, cache=args.output_root / "cc-cache", ) BASE.write_json(run_dir / "command.json", command) started = time.time() with (run_dir / "stdout.log").open("w") as stdout, ( run_dir / "stderr.log" ).open("w") as stderr: completed = subprocess.run( command, cwd=args.frontier_source, env=environment, stdout=stdout, stderr=stderr, timeout=args.timeout_seconds, check=False, ) if completed.returncode != 0: stderr_path = run_dir / "stderr.log" result = { "status": "frontier_failed", "failure_kind": classify_frontier_failure( stderr_path.read_text(errors="replace") ), "returncode": completed.returncode, "config": { "tp": config.tp, "mns": config.mns, "name": config.name, }, "trace_label": trace["label"], "offered_request_rate": trace["offered_request_rate"], "offered_request_rate_per_gpu": ( trace["offered_request_rate"] / config.tp ), "request_count": trace["requests"], "elapsed_seconds": time.time() - started, "trace_sha256": trace["sha256"], "stderr_sha256": BASE.sha256(stderr_path), } BASE.write_json(result_path, result) loads.append(result) print( json.dumps( { "config": config.name, "trace": trace["label"], "status": result["status"], "failure_kind": result["failure_kind"], }, sort_keys=True, ), flush=True, ) if not args.continue_on_failure: raise RuntimeError( f"Frontier failed for {config.name}/{trace['label']}: " f"{completed.returncode}" ) continue metrics = find_request_metrics(run_dir) result = { "status": "completed", "config": {"tp": config.tp, "mns": config.mns, "name": config.name}, "trace_label": trace["label"], "offered_request_rate": trace["offered_request_rate"], "offered_request_rate_per_gpu": trace["offered_request_rate"] / config.tp, "request_count": trace["requests"], "elapsed_seconds": time.time() - started, "trace_sha256": trace["sha256"], "request_metrics_sha256": BASE.sha256(metrics), "score": score(metrics, trace["shapes"]), } BASE.write_json(result_path, result) loads.append(result) print( json.dumps( { "config": config.name, "trace": trace["label"], "rate": trace["offered_request_rate"], "primary": result["score"]["slos"]["tpot_150ms"], }, sort_keys=True, ), flush=True, ) config_results.append( { "config": {"tp": config.tp, "mns": config.mns, "name": config.name}, "loads": loads, } ) rankings = {} for slo in (f"tpot_{int(value)}ms" for value in TPOT_SLOS_MS): records = [] for item in config_results: completed_loads = [ load for load in item["loads"] if load["status"] == "completed" ] invalid_loads = [ load for load in item["loads"] if load["status"] != "completed" ] feasible = [ load["offered_request_rate"] for load in completed_loads if load["score"]["slos"][slo]["feasible"] ] capacity = max(feasible, default=None) records.append( { "config": item["config"], "ranking_valid": not invalid_loads, "invalid_loads": [ { "trace_label": load["trace_label"], "failure_kind": load.get("failure_kind", "unknown"), } for load in invalid_loads ], "maximum_tested_feasible_request_rate": capacity, "maximum_tested_feasible_request_rate_per_gpu": ( capacity / item["config"]["tp"] if capacity is not None else None ), "lower_censored": capacity is None and not invalid_loads, "upper_censored": ( capacity == traces[-1]["offered_request_rate"] and not invalid_loads ), } ) records.sort( key=lambda row: ( -( row["maximum_tested_feasible_request_rate_per_gpu"] if row["maximum_tested_feasible_request_rate_per_gpu"] is not None else -1 ), row["config"]["name"], ) ) rankings[slo] = records has_invalid_cells = any( load["status"] != "completed" for item in config_results for load in item["loads"] ) if selected != list(BASE.GRID): manifest_status = "partial_not_decision_bearing" elif has_invalid_cells: manifest_status = "frozen_with_invalid_cells" else: manifest_status = "frozen_before_real" manifest = { "schema": "frontier-qwen30-exact-trace-surface-v1", "status": manifest_status, "contract": { "window_seconds": ( WINDOW_SECONDS if args.rate_contract == "trace-window" else None ), "rate_contract": args.rate_contract, "prefix_caching": args.prefix_caching, "arrival": "original_trace_timestamp_and_order", "input_output": "exact_source_values", "ttft_slo": "1000ms + 1000ms * input_tokens / 8000", "tpot_slos_ms": TPOT_SLOS_MS, "primary_tpot_slo_ms": 150.0, "target_pass_rate": TARGET_PASS_RATE, "predictor_training_job_threads": args.predictor_training_job_threads, "decode_cuda_graph_mode": args.decode_cuda_graph_mode, "align_real_graph_runtime": args.align_real_graph_runtime, "fresh_predictor_cache": args.fresh_predictor_cache, }, "frontier": { "source": str(args.frontier_source), "git_head": frontier_head, "git_status_short": subprocess.run( ["git", "-C", str(args.frontier_source), "status", "--short"], check=True, text=True, stdout=subprocess.PIPE, ).stdout, }, "profiles": { "root": str(args.profile_root), "coverage": coverage, "sha256": {name: BASE.sha256(path) for name, path in paths.items()}, }, "kernel_only_profiles": ( None if kernel_paths is None else { "root": str(args.kernel_profile_root), "coverage": kernel_coverage, "sha256": { name: BASE.sha256(path) for name, path in kernel_paths.items() }, } ), "runtime_alignment": { "capture_sizes_by_mns": ( GRAPH_CAPTURE_SIZES_BY_MNS if args.align_real_graph_runtime else None ), "num_blocks_by_config": ( { f"tp{tp}_mns{mns}": blocks for (tp, mns), blocks in REAL_NUM_BLOCKS_BY_CONFIG.items() } if args.align_real_graph_runtime else None ), }, "collective": { "backend": args.cc_backend, "allreduce_csv": str(args.allreduce_csv) if args.allreduce_csv else None, "allreduce_csv_sha256": BASE.sha256(args.allreduce_csv) if args.allreduce_csv else None, }, "traces": [trace_manifest_entry(trace) for trace in traces], "config_results": config_results, "rankings": rankings, } BASE.write_json(args.output_root / "frontier_trace_surface_frozen.json", manifest) print(args.output_root / "frontier_trace_surface_frozen.json") if __name__ == "__main__": main()