diff --git a/runs/frontier-fidelity-envelope-v1/analyze_qwen30_latency_case.py b/runs/frontier-fidelity-envelope-v1/analyze_qwen30_latency_case.py new file mode 100644 index 0000000..2b7e734 --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/analyze_qwen30_latency_case.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +"""Compare one complete Qwen30 Frontier latency surface with real vLLM.""" + +from __future__ import annotations + +import argparse +import json +import math +from itertools import combinations +from pathlib import Path +from typing import Any + + +CONFIGS = tuple(f"tp{tp}_mns{mns}" for tp in (1, 2, 4) for mns in (8, 16, 32, 64)) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--sim-root", type=Path, required=True) + parser.add_argument("--real-audit", 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 number(value: Any, field: str) -> float: + if not isinstance(value, (int, float)) or isinstance(value, bool): + raise ValueError(f"{field} is not numeric: {value!r}") + result = float(value) + if not math.isfinite(result) or result < 0: + raise ValueError(f"{field} is invalid: {value!r}") + return result + + +def parse_config(name: str) -> tuple[int, int]: + tp, mns = name.split("_", 1) + return int(tp.removeprefix("tp")), int(mns.removeprefix("mns")) + + +def rank(values: dict[str, float]) -> list[str]: + return [name for name, _ in sorted(values.items(), key=lambda item: (item[1], item[0]))] + + +def pairwise(sim: dict[str, float], real: dict[str, float]) -> dict[str, int | float | None]: + concordant = discordant = ties = 0 + for left, right in combinations(CONFIGS, 2): + sim_delta = sim[left] - sim[right] + real_delta = real[left] - real[right] + if math.isclose(sim_delta, 0.0, abs_tol=1e-9) or math.isclose(real_delta, 0.0, abs_tol=1e-9): + ties += 1 + elif (sim_delta > 0) == (real_delta > 0): + concordant += 1 + else: + discordant += 1 + informative = concordant + discordant + return { + "concordant_pairs": concordant, + "discordant_pairs": discordant, + "tied_pairs": ties, + "informative_pairs": informative, + "agreement": concordant / informative if informative else None, + } + + +def main() -> None: + args = parse_args() + root = args.sim_root.resolve() + audit = json.loads(args.real_audit.read_text()) + metrics = tuple(audit.get("applicable_metrics") or []) + if not metrics or set(audit.get("configs") or {}) != set(CONFIGS): + raise ValueError("real audit is incomplete") + expected_requests = int(audit["trace_manifests"]["tp1"]["requests"]) + cells: dict[str, Any] = {} + for name in CONFIGS: + tp, mns = parse_config(name) + path = root / "runs" / name / f"tp{tp}" / "result.json" + result = json.loads(path.read_text()) + if result.get("status") != "completed" or result.get("request_count") != expected_requests: + raise ValueError(f"incomplete simulator result: {path}") + if result.get("config") != {"tp": tp, "mns": mns, "name": name}: + raise ValueError(f"simulator config drift: {path}") + score = result.get("score") + if not isinstance(score, dict): + raise ValueError(f"simulator score missing: {path}") + values: dict[str, Any] = {} + for metric in metrics: + prefix = metric.removesuffix("_ms") + values[metric] = { + statistic: number(score.get(f"{prefix}_{statistic}_ms"), f"{path}:{metric}:{statistic}") + for statistic in ("mean", "p90") + } + cells[name] = {"path": str(path), "metrics": values} + selection: dict[str, Any] = {} + for metric in metrics: + for statistic in ("mean", "p90"): + sim_values = {name: cells[name]["metrics"][metric][statistic] for name in CONFIGS} + real_values = { + name: number( + audit["configs"][name]["metrics"][metric][f"pooled_{statistic}_ms"], + f"real:{name}:{metric}:{statistic}", + ) + for name in CONFIGS + } + sim_ranking, real_ranking = rank(sim_values), rank(real_values) + sim_choice, real_best = sim_ranking[0], real_ranking[0] + selection[f"{metric}:{statistic}"] = { + "sim_winner": sim_choice, + "real_winner": real_best, + "winner_match": sim_choice == real_best, + "selected_config_real_regret": (real_values[sim_choice] - real_values[real_best]) / real_values[real_best], + "sim_ranking": sim_ranking, + "real_ranking": real_ranking, + **pairwise(sim_values, real_values), + } + payload = { + "schema": "qwen30-latency-case-frontier-real-comparison-v1", + "sim_root": str(root), + "real_audit": str(args.real_audit.resolve()), + "prefill_only": bool(audit["prefill_only"]), + "applicable_metrics": metrics, + "selection": selection, + "cells": cells, + } + lines = ["# Qwen30 Frontier vs real latency selection", "", "| Objective | Frontier | Real | Match | Regret | Pairwise |", "|---|---|---|---:|---:|---:|"] + for objective, row in selection.items(): + agreement = "N/A" if row["agreement"] is None else f"{row['agreement']:.1%}" + lines.append(f"| {objective} | {row['sim_winner']} | {row['real_winner']} | {'yes' if row['winner_match'] else 'no'} | {row['selected_config_real_regret']:.1%} | {agreement} |") + args.json_output.parent.mkdir(parents=True, exist_ok=True) + args.markdown_output.parent.mkdir(parents=True, exist_ok=True) + args.json_output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + args.markdown_output.write_text("\n".join(lines) + "\n") + print(json.dumps(selection, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/runs/frontier-fidelity-envelope-v1/audit_qwen30_latency_case.py b/runs/frontier-fidelity-envelope-v1/audit_qwen30_latency_case.py new file mode 100644 index 0000000..ed06274 --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/audit_qwen30_latency_case.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 +"""Audit one completed Qwen30 latency-selection real surface. + +This reader intentionally treats a prefill-only case as a four-objective +surface: TTFT/E2E mean and p90. TPOT is omitted rather than coerced to zero. +""" + +from __future__ import annotations + +import argparse +import json +import math +import statistics +from pathlib import Path +from typing import Any + + +CONFIGS = tuple(f"tp{tp}_mns{mns}" for tp in (1, 2, 4) for mns in (8, 16, 32, 64)) +TRIALS = (1, 2, 3) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--case-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 nearest_rank(values: list[float], fraction: float) -> float: + if not values: + raise ValueError("cannot calculate a percentile of an empty metric") + return sorted(values)[math.ceil(len(values) * fraction) - 1] + + +def finite(value: Any, field: str) -> float: + if not isinstance(value, (int, float)) or isinstance(value, bool): + raise ValueError(f"{field} is not numeric: {value!r}") + value = float(value) + if not math.isfinite(value) or value < 0: + raise ValueError(f"{field} is invalid: {value!r}") + return value + + +def read_manifest(root: Path, tp: int) -> dict[str, Any]: + path = root / "traces" / f"tp{tp}" / "public" / "manifest.json" + manifest = json.loads(path.read_text()) + if manifest.get("schema") != "qwen30-latency-case-v1": + raise ValueError(f"unexpected trace schema in {path}") + if manifest.get("tensor_parallel_size") != tp or int(manifest.get("requests", 0)) <= 0: + raise ValueError(f"invalid trace contract in {path}") + outputs = manifest.get("output_tokens") + if not isinstance(outputs, list) or len(outputs) != 1 or int(outputs[0]) <= 0: + raise ValueError(f"non-uniform output contract in {path}") + return manifest + + +def metric_stats(values: list[float]) -> dict[str, float]: + return {"samples": len(values), "mean_ms": statistics.fmean(values), "p90_ms": nearest_rank(values, 0.90)} + + +def validate_trial(path: Path, manifest: dict[str, Any]) -> tuple[dict[str, list[float]], dict[str, Any]]: + payload = json.loads(path.read_text()) + if payload.get("schema") != "qwen30-exact-trace-anchor-v1": + raise ValueError(f"unexpected real result schema in {path}") + contract = payload.get("contract") + summary = payload.get("summary") + records = payload.get("requests") + if not isinstance(contract, dict) or not isinstance(summary, dict) or not isinstance(records, list): + raise ValueError(f"malformed result payload: {path}") + expected = int(manifest["requests"]) + expected_contract = { + "requests": expected, + "requests_file_sha256": manifest["private_jsonl_sha256"], + "row_vector_sha256": manifest["row_vector_sha256"], + "first_arrival_s": manifest["first_arrival_s"], + "last_arrival_s": manifest["last_arrival_s"], + } + for name, wanted in expected_contract.items(): + actual = contract.get(name) + if isinstance(wanted, float): + if not isinstance(actual, (int, float)) or not math.isclose(float(actual), wanted, abs_tol=1e-9): + raise ValueError(f"{path}: contract drift for {name}") + elif actual != wanted: + raise ValueError(f"{path}: contract drift for {name}") + if len(records) != expected or summary.get("completed") != expected or summary.get("failed") != 0: + raise ValueError(f"{path}: incomplete real replay") + + metrics: dict[str, list[float]] = {"ttft_ms": [], "tpot_ms": [], "e2e_ms": []} + indices: set[int] = set() + expected_output = int(manifest["output_tokens"][0]) + for record in records: + if record.get("success") is not True: + raise ValueError(f"{path}: failed request record") + index = record.get("source_index") + if not isinstance(index, int) or index in indices: + raise ValueError(f"{path}: duplicate/non-integer source index") + indices.add(index) + if record.get("actual_input_tokens") != record.get("input_tokens"): + raise ValueError(f"{path}: input usage drift") + if record.get("requested_output_tokens") != expected_output or record.get("actual_output_tokens") != expected_output: + raise ValueError(f"{path}: output usage drift") + for metric in ("ttft_ms", "e2e_ms"): + metrics[metric].append(finite(record.get(metric), metric)) + tpot = record.get("tpot_ms") + if expected_output == 1: + if tpot is not None: + raise ValueError(f"{path}: OSL=1 must report TPOT=null") + else: + metrics["tpot_ms"].append(finite(tpot, "tpot_ms")) + if len(indices) != expected: + raise ValueError(f"{path}: missing source rows") + return metrics, {"result_path": str(path), "requests": expected} + + +def main() -> None: + args = parse_args() + root = args.case_root.resolve() + manifests = {tp: read_manifest(root, tp) for tp in (1, 2, 4)} + prefill_only = int(manifests[1]["output_tokens"][0]) == 1 + if any((int(manifest["output_tokens"][0]) == 1) != prefill_only for manifest in manifests.values()): + raise ValueError("TP-specific output contracts differ") + applicable = ("ttft_ms", "e2e_ms") if prefill_only else ("ttft_ms", "tpot_ms", "e2e_ms") + configs: dict[str, Any] = {} + for name in CONFIGS: + tp = int(name.split("_", 1)[0].removeprefix("tp")) + trial_rows = [] + pooled = {metric: [] for metric in applicable} + for trial in TRIALS: + path = root / "real" / name / f"trial{trial}" / "results" / "result.json" + values, provenance = validate_trial(path, manifests[tp]) + trial_rows.append({ + **provenance, + "trial": trial, + "metrics": {metric: metric_stats(values[metric]) for metric in applicable}, + }) + for metric in applicable: + pooled[metric].extend(values[metric]) + configs[name] = { + "trials": trial_rows, + "metrics": { + metric: { + "pooled_samples": len(pooled[metric]), + "pooled_mean_ms": statistics.fmean(pooled[metric]), + "pooled_p90_ms": nearest_rank(pooled[metric], 0.90), + "trial_mean_of_means_ms": statistics.fmean( + row["metrics"][metric]["mean_ms"] for row in trial_rows + ), + "trial_stddev_of_means_ms": statistics.stdev( + row["metrics"][metric]["mean_ms"] for row in trial_rows + ), + } + for metric in applicable + }, + } + winners: dict[str, Any] = {} + for metric in applicable: + for statistic in ("pooled_mean_ms", "pooled_p90_ms"): + ranked = sorted((row["metrics"][metric][statistic], name) for name, row in configs.items()) + winners[f"{metric}:{statistic}"] = { + "winner": ranked[0][1], + "winner_value_ms": ranked[0][0], + "ranking": [name for _, name in ranked], + } + payload = { + "schema": "qwen30-latency-case-real-audit-v1", + "case_root": str(root), + "prefill_only": prefill_only, + "applicable_metrics": list(applicable), + "trace_manifests": {f"tp{tp}": manifests[tp] for tp in manifests}, + "configs": configs, + "winners": winners, + } + lines = ["# Qwen30 real latency case audit", "", f"Prefill-only: `{prefill_only}`.", ""] + lines += ["| Objective | Real winner | Value (ms) |", "|---|---|---:|"] + for objective, winner in winners.items(): + lines.append(f"| {objective} | {winner['winner']} | {winner['winner_value_ms']:.2f} |") + args.json_output.parent.mkdir(parents=True, exist_ok=True) + args.markdown_output.parent.mkdir(parents=True, exist_ok=True) + args.json_output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + args.markdown_output.write_text("\n".join(lines) + "\n") + print(json.dumps(winners, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/runs/frontier-fidelity-envelope-v1/prepare_qwen30_latency_case.py b/runs/frontier-fidelity-envelope-v1/prepare_qwen30_latency_case.py new file mode 100644 index 0000000..e4bfa6a --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/prepare_qwen30_latency_case.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +"""Materialize Qwen30 Fixed/Trace latency cases without changing their contract. + +The private request file deliberately contains either an exact source prompt +(trace cases) or deterministic token IDs (fixed cases). The public Frontier +fixture contains only arrivals, shapes, sessions, and legal complete prefix +block identities. +""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import math +from pathlib import Path +from typing import Any + + +FIELDS = ( + "arrived_at", + "num_prefill_tokens", + "num_decode_tokens", + "session_id", + "block_hash_ids", +) + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def row_digest(rows: list[dict[str, Any]]) -> str: + digest = hashlib.sha256() + for row in rows: + digest.update( + json.dumps( + [ + row["source_index"], + row["arrived_at"], + row["input_length"], + row["output_length"], + row["session_id"], + row["runtime_block_ids"], + ], + separators=(",", ":"), + ).encode() + ) + digest.update(b"\n") + return digest.hexdigest() + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + sub = parser.add_subparsers(dest="kind", required=True) + + trace = sub.add_parser("trace") + trace.add_argument("--base-public", type=Path, required=True) + trace.add_argument("--base-private", type=Path, required=True) + trace.add_argument("--output-tokens", type=int, required=True) + trace.add_argument("--tp", type=int, required=True) + trace.add_argument("--output-root", type=Path, required=True) + + fixed = sub.add_parser("fixed") + fixed.add_argument("--model", type=Path, required=True) + fixed.add_argument("--input-tokens", type=int, required=True) + fixed.add_argument("--output-tokens", type=int, required=True) + fixed.add_argument("--requests", type=int, required=True) + fixed.add_argument("--per-gpu-rate", type=float, required=True) + fixed.add_argument("--tp", type=int, required=True) + fixed.add_argument("--output-root", type=Path, required=True) + return parser.parse_args() + + +def materialize_trace(args: argparse.Namespace) -> tuple[list[dict[str, Any]], bool, str, float]: + with args.base_public.open(newline="") as source: + public = list(csv.DictReader(source)) + private = [json.loads(line) for line in args.base_private.open() if line.strip()] + if not public or len(public) != len(private): + raise ValueError("trace public/private request count mismatch") + rows: list[dict[str, Any]] = [] + for public_row, private_row in zip(public, private, strict=True): + if int(public_row["num_prefill_tokens"]) != int(private_row["input_length"]): + raise ValueError("trace input-length drift") + if float(public_row["arrived_at"]) != float(private_row["arrived_at"]): + raise ValueError("trace arrival drift") + # The real vLLM request artifact retains its final partial prompt block + # for provenance, while the Frontier CSV deliberately projects only + # legal complete cache blocks. Use the latter as the shared + # simulator-facing identity vector; vLLM itself derives cache hashes + # from the exact private prompt text. + runtime_ids = [int(value) for value in public_row["block_hash_ids"].split("|") if value] + if len(runtime_ids) != int(private_row["input_length"]) // 16: + raise ValueError("public trace exposes an incomplete runtime prefix block") + if int(public_row["session_id"]) != int(private_row["session_id"]): + raise ValueError("trace session-id drift") + body = dict(private_row["body"]) + body.update( + { + "min_tokens": args.output_tokens, + "max_tokens": args.output_tokens, + "ignore_eos": True, + } + ) + rows.append( + { + "source_index": int(private_row["source_index"]), + "arrived_at": float(private_row["arrived_at"]), + "input_length": int(private_row["input_length"]), + "output_length": args.output_tokens, + "session_id": int(private_row["session_id"]), + "runtime_block_ids": runtime_ids, + "body": body, + } + ) + return ( + rows, + True, + "trace-derived: exact input/arrival/session/prefix; output override only", + len(rows) / 600.0 * args.tp, + ) + + +def materialize_fixed(args: argparse.Namespace) -> tuple[list[dict[str, Any]], bool, str, float]: + if min(args.input_tokens, args.output_tokens, args.requests, args.tp) <= 0: + raise ValueError("fixed dimensions must be positive") + if args.input_tokens + args.output_tokens > 40960: + raise ValueError("fixed shape exceeds max model length") + if not math.isfinite(args.per_gpu_rate) or args.per_gpu_rate <= 0: + raise ValueError("per-GPU rate must be positive") + from transformers import AutoTokenizer + + tokenizer = AutoTokenizer.from_pretrained(args.model, trust_remote_code=True) + candidates = [ + token for token in range(tokenizer.vocab_size) if token not in set(tokenizer.all_special_ids) + ] + if len(candidates) < args.requests + 1: + raise ValueError("tokenizer does not provide enough non-special tokens") + rate = args.per_gpu_rate * args.tp + base = candidates[0] + rows = [] + for index in range(args.requests): + rows.append( + { + "source_index": index, + "arrived_at": index / rate, + "input_length": args.input_tokens, + "output_length": args.output_tokens, + "session_id": index, + "runtime_block_ids": [], + "body": { + "prompt": [candidates[index + 1], *([base] * (args.input_tokens - 1))], + "min_tokens": args.output_tokens, + "max_tokens": args.output_tokens, + "ignore_eos": True, + }, + } + ) + return ( + rows, + False, + "fixed-shape: deterministic token IDs, uniform TP-normalized QPS, no prefix reuse", + rate, + ) + + +def write_case( + rows: list[dict[str, Any]], + *, + prefix_caching: bool, + description: str, + global_rate: float, + tp: int, + root: Path, +) -> None: + if not rows: + raise ValueError("empty case") + root = root.resolve() + public_root = root / "public" + private_root = root / "private" + public_root.mkdir(parents=True, exist_ok=True) + private_root.mkdir(parents=True, exist_ok=True) + public_path = public_root / "frontier.csv" + private_path = private_root / "real_requests.jsonl" + with public_path.open("w", newline="") as output: + writer = csv.DictWriter(output, fieldnames=FIELDS, lineterminator="\n") + writer.writeheader() + for row in rows: + writer.writerow( + { + "arrived_at": f"{row['arrived_at']:.12f}", + "num_prefill_tokens": row["input_length"], + "num_decode_tokens": row["output_length"], + "session_id": row["session_id"], + "block_hash_ids": "|".join(str(value) for value in row["runtime_block_ids"]), + } + ) + with private_path.open("w") as output: + for row in rows: + output.write(json.dumps(row, separators=(",", ":")) + "\n") + arrivals = [float(row["arrived_at"]) for row in rows] + manifest = { + "schema": "qwen30-latency-case-v1", + "description": description, + "tensor_parallel_size": tp, + "requests": len(rows), + "output_tokens": sorted({int(row["output_length"]) for row in rows}), + "prefix_caching": prefix_caching, + "public_csv": str(public_path), + "public_csv_sha256": sha256(public_path), + "private_jsonl": str(private_path), + "private_jsonl_sha256": sha256(private_path), + "row_vector_sha256": row_digest(rows), + "first_arrival_s": arrivals[0], + "last_arrival_s": arrivals[-1], + "global_offered_request_rate": global_rate, + "per_gpu_offered_request_rate": global_rate / tp, + } + (public_root / "manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") + print(json.dumps(manifest, sort_keys=True)) + + +def main() -> None: + args = parse_args() + if args.kind == "trace": + rows, prefix, description, global_rate = materialize_trace(args) + else: + rows, prefix, description, global_rate = materialize_fixed(args) + write_case( + rows, + prefix_caching=prefix, + description=description, + global_rate=global_rate, + tp=args.tp, + root=args.output_root, + ) + + +if __name__ == "__main__": + main() diff --git a/runs/frontier-fidelity-envelope-v1/qwen30-latency-expansion-card.md b/runs/frontier-fidelity-envelope-v1/qwen30-latency-expansion-card.md new file mode 100644 index 0000000..8ed2b8c --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/qwen30-latency-expansion-card.md @@ -0,0 +1,49 @@ +# EXP-SIMFID-Q30-LATENCY-EXPANSION: remaining Fixed/Trace × PD/P cases + +> Status: prepared for runtime-alignment preflight (2026-07-18). The already +> completed Qwen3-30B-A3B Trace-PD surface is excluded from this card. + +## Question and fixed boundary + +Does the same graph-aligned Frontier configuration that selected the correct +winner on Trace-PD also select the real vLLM winner when either the arrival/ +prefix state or the decode phase is removed? This is a selection test, not an +SLO/capacity evaluation. + +| Case | Input / output | arrival and cache | Applicable objectives | +|---|---|---|---| +| Fixed-PD | 2048 / 128 | 129 uniform requests; `t'=t/TP`; prefix cache off | mean/p90 TTFT, TPOT, E2E | +| Trace-P | exact held-out input, arrival, session and complete block-16 prefix relation / 1 | same TP-normalized 129-row trace; prefix cache on | mean/p90 TTFT, E2E; TPOT=N/A | +| Fixed-P | 2048 / 1 | 129 uniform requests; `t'=t/TP`; prefix cache off | mean/p90 TTFT, E2E; TPOT=N/A | + +All cases use Qwen3-30B-A3B BF16, community vLLM 0.20.0, H20, `TP∈{1,2,4}` +and `MNS∈{8,16,32,64}`, MBT=8192, chunked prefill, three fresh-server trials +per cell. The fixed QPS is the Trace-PD base offered rate (`129/600` req/s per +GPU), hence the global arrival rate is multiplied by TP. It makes throughput +per GPU comparable without claiming that different TP values see identical +cluster-level load. + +## Simulator and real contracts + +- Frontier uses commit `deadc4a…`, `piecewise`, the frozen CUDA-event profile + for prefill/mixed batches and KERNEL_ONLY profile for captured pure decode. +- Trace-P reuses the already verified graph buckets/KV capacities because the + server CLI is unchanged from Trace-PD. The only workload change is `OSL→1`. +- Fixed-P/PD disables vLLM prefix caching on both sides. Before freezing their + simulator surfaces, a no-request server-start preflight records the actual + graph captures and KV-block capacities for every `(TP,MNS)`; no request + latency or winner is used as calibration. +- Each real result verifies every request's input/output usage and row-vector + digest. `OSL=1` produces JSON `null` TPOT samples and is never converted to + zero. + +## Decision rule and cost + +For every applicable objective, compare the complete 12-cell simulator and +pooled three-trial real surface: winner match, selected-config real regret, +and non-tied pair direction agreement. A simulator crash or missing request +metric is a coverage failure, not a high-latency cell. + +The no-request Fixed runtime preflight is capped at 2 H20-GPUh. Each 36-run +real surface is estimated at 13 nominal / 41 worst-case H20-GPUh, plus CPU-only +Frontier replay. Launch logs record the resolved inputs, paths, and duration. diff --git a/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py b/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py index cac0e74..a2c39a0 100644 --- a/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py +++ b/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py @@ -279,19 +279,23 @@ def score(path: Path, expected_shapes: list[tuple[int, int]]) -> dict[str, Any]: 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] + def summary(values: list[float]) -> dict[str, float | None]: + if not values: + return {"mean": None, "p50": None, "p90": None, "p95": None} + return { + "mean": sum(values) / len(values), + "p50": percentile(values, 0.50), + "p90": percentile(values, 0.90), + "p95": percentile(values, 0.95), + } + + ttft_summary = summary(ttfts) + tpot_summary = summary(tpots) + e2e_summary = summary(e2es) 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), + **{f"ttft_{name}_ms": value for name, value in ttft_summary.items()}, + **{f"tpot_{name}_ms": value for name, value in tpot_summary.items()}, + **{f"e2e_{name}_ms": value for name, value in e2e_summary.items()}, "slos": slos, } diff --git a/runs/frontier-fidelity-envelope-v1/run_qwen30_exact_trace_real_anchor.sh b/runs/frontier-fidelity-envelope-v1/run_qwen30_exact_trace_real_anchor.sh index 1b582e0..1037a54 100755 --- a/runs/frontier-fidelity-envelope-v1/run_qwen30_exact_trace_real_anchor.sh +++ b/runs/frontier-fidelity-envelope-v1/run_qwen30_exact_trace_real_anchor.sh @@ -8,10 +8,12 @@ TP="${TP:?TP is required}" MNS="${MNS:?MNS is required}" TRACE_LABEL="${TRACE_LABEL:?TRACE_LABEL is required}" SERVER_PORT="${SERVER_PORT:?SERVER_PORT is required}" +PREFIX_CACHING="${PREFIX_CACHING:-true}" VENV_ROOT="${VENV_ROOT:-/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1}" MODEL_ROOT="${MODEL_ROOT:-/home/admin/cpfs/wjh/models/Qwen/Qwen3-30B-A3B}" FLASHINFER_WORKSPACE_BASE="${FLASHINFER_WORKSPACE_BASE:-/tmp/wjh/flashinfer-workspace-vllm020-profiler-v1}" SERVED_MODEL="qwen3-30b-exact-trace" +EXACT_TRACE_CLIENT="${EXACT_TRACE_CLIENT:-qwen30_exact_trace_client.py}" SERVER_PID="" mkdir -p "${OUTPUT_ROOT}/logs" "${OUTPUT_ROOT}/provenance" "${OUTPUT_ROOT}/results" \ @@ -37,8 +39,21 @@ if [[ "${#GPU_IDS[@]}" -ne "${TP}" ]]; then exit 1 fi +case "${PREFIX_CACHING}" in + true) + PREFIX_CACHING_FLAG="--enable-prefix-caching" + ;; + false) + PREFIX_CACHING_FLAG="--no-enable-prefix-caching" + ;; + *) + echo "ERROR: PREFIX_CACHING must be true or false, got ${PREFIX_CACHING}" >&2 + exit 1 + ;; +esac + REQUEST_COUNT="$(wc -l < "${REQUESTS_FILE}")" -echo "QWEN30_EXACT_TRACE_REAL_LAUNCH_ECHO host=$(hostname) gpus=${CUDA_VISIBLE_DEVICES} model=${MODEL_ROOT} runtime=vLLM-0.20.0+cu129 dtype=BF16 config=TP${TP}_MNS${MNS}_MBT8192 trace=${TRACE_LABEL} requests=${REQUEST_COUNT} source=${REQUESTS_FILE} arrivals=original600s input_output_prompt=exact prefix=on block=16 tpot_slo=150ms flashinfer_workspace=${FLASHINFER_WORKSPACE_BASE} output=${OUTPUT_ROOT} expected_wall=12-35m hard_wall=3600s" +echo "QWEN30_EXACT_TRACE_REAL_LAUNCH_ECHO host=$(hostname) gpus=${CUDA_VISIBLE_DEVICES} model=${MODEL_ROOT} runtime=vLLM-0.20.0+cu129 dtype=BF16 config=TP${TP}_MNS${MNS}_MBT8192 trace=${TRACE_LABEL} requests=${REQUEST_COUNT} source=${REQUESTS_FILE} arrivals=manifest prefix=${PREFIX_CACHING} block=16 metrics=TTFT,TPOT-if-OSL-gt-1,E2E flashinfer_workspace=${FLASHINFER_WORKSPACE_BASE} output=${OUTPUT_ROOT} expected_wall=12-35m hard_wall=3600s" date -u +"START_UTC=%Y-%m-%dT%H:%M:%SZ" sha256sum qwen30_exact_trace_client.py run_qwen30_exact_trace_real_anchor.sh \ ../frontier-phase-factorial-v0/qwen30_prefill_client.py \ @@ -60,7 +75,7 @@ setsid "${VENV_ROOT}/bin/vllm" serve "${MODEL_ROOT}" \ --host 127.0.0.1 --port "${SERVER_PORT}" --served-model-name "${SERVED_MODEL}" \ --tensor-parallel-size "${TP}" --gpu-memory-utilization 0.92 \ --max-model-len 40960 --max-num-batched-tokens 8192 --max-num-seqs "${MNS}" \ - --enable-prefix-caching --enable-chunked-prefill --no-enable-log-requests \ + "${PREFIX_CACHING_FLAG}" --enable-chunked-prefill --no-enable-log-requests \ > "${OUTPUT_ROOT}/logs/server.log" 2>&1 & SERVER_PID=$! READY=0 @@ -89,7 +104,7 @@ fi --model-path "${MODEL_ROOT}" --rate 1 --requests 4 --input-tokens 512 \ --output-tokens 1 --output "${OUTPUT_ROOT}/results/warmup.json" -"${VENV_ROOT}/bin/python" qwen30_exact_trace_client.py \ +"${VENV_ROOT}/bin/python" "${EXACT_TRACE_CLIENT}" \ --port "${SERVER_PORT}" --requests-file "${REQUESTS_FILE}" \ --output "${OUTPUT_ROOT}/results/result.json" --tpot-slo-ms 150 \ --timeout-seconds 1800