1512 lines
58 KiB
Python
1512 lines
58 KiB
Python
#!/usr/bin/env python3
|
|
"""Blind Frontier/community-vLLM rank comparison on one fixed prompt cohort.
|
|
|
|
The earlier grid selected a different ``sampling_u`` subset at every load. That
|
|
confounds offered load with prompt mix and makes binary-search monotonicity an
|
|
untested assumption. This runner selects one length-stratified cohort, changes
|
|
load only by scaling its complete arrival timeline, evaluates every discrete
|
|
load, freezes Frontier first, and only then permits the community-vLLM run.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import random
|
|
import statistics
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from dataclasses import asdict, replace
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
REPO_ROOT = SCRIPT_DIR.parents[2]
|
|
for search_path in (SCRIPT_DIR, REPO_ROOT / "src"):
|
|
if str(search_path) not in sys.path:
|
|
sys.path.insert(0, str(search_path))
|
|
|
|
import community_prefill_grid as community_grid # noqa: E402
|
|
import frontier_prefill_grid as frontier_grid # noqa: E402
|
|
from aituner.engine import build_launch_recipe # noqa: E402
|
|
from aituner.spec import ConfigPatch, load_study_spec # noqa: E402
|
|
from aituner.trace import TraceRequest, load_trace_requests # noqa: E402
|
|
from aituner.worker import ( # noqa: E402
|
|
_ignore_sigterm_if_main,
|
|
_install_sigterm_as_keyboardinterrupt,
|
|
_replay_requests,
|
|
_restore_sigterm,
|
|
_terminate_process_tree,
|
|
_wait_for_server_or_exit,
|
|
)
|
|
|
|
|
|
PROTOCOL_SCHEMA = "qwen235b-prefill-fixed-cohort-protocol-v1"
|
|
FRONTIER_SCHEMA = "frontier-qwen235b-prefill-fixed-cohort-v1"
|
|
COMMUNITY_SCHEMA = "community-qwen235b-prefill-fixed-cohort-v1"
|
|
COMPARISON_SCHEMA = "frontier-community-qwen235b-rank-comparison-v1"
|
|
COHORT_SIZE = 64
|
|
COHORT_SEED = 2026071501
|
|
CONFIG_ORDER_SEED = 2026071502
|
|
RATE_ORDER_SEED = 2026071503
|
|
WARMUP_SEED = 2026071504
|
|
INPUT_BINS = (0, 1024, 2048, 4096, 8192, 16384, 32769)
|
|
OFFERED_RATES = (0.15, 0.25, 0.35, 0.50, 0.75, 1.00, 1.50)
|
|
TARGET_PASS_RATE = 0.95
|
|
EXPECTED_QUANT_SIGNATURE = "method=fp8|act=dynamic|serialized=True|block=128x128"
|
|
|
|
SLO_VARIANTS: dict[str, dict[str, Any]] = {
|
|
"linear_8k_primary": {
|
|
"description": "TTFT <= 1000 ms + input_tokens / 8000 tokens/s",
|
|
"kind": "linear_ms",
|
|
"intercept_ms": 1000.0,
|
|
"per_token_ms": 0.125,
|
|
},
|
|
"linear_6k": {
|
|
"description": "TTFT <= 1000 ms + input_tokens / 6000 tokens/s",
|
|
"kind": "linear_ms",
|
|
"intercept_ms": 1000.0,
|
|
"per_token_ms": 1.0 / 6.0,
|
|
},
|
|
"linear_10k": {
|
|
"description": "TTFT <= 1000 ms + input_tokens / 10000 tokens/s",
|
|
"kind": "linear_ms",
|
|
"intercept_ms": 1000.0,
|
|
"per_token_ms": 0.1,
|
|
},
|
|
"legacy_step_1s_2s": {
|
|
"description": "Original strict 1 s/2 s step SLO (null-capacity sensitivity)",
|
|
"kind": "step_ms",
|
|
"buckets": [
|
|
{"max_input_tokens": 8191, "threshold_ms": 1000.0},
|
|
{"threshold_ms": 2000.0},
|
|
],
|
|
},
|
|
}
|
|
PRIMARY_SLO = "linear_8k_primary"
|
|
|
|
|
|
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 sha256_text(value: str) -> str:
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
|
|
def order_hash(values: Iterable[object]) -> str:
|
|
return sha256_text("\n".join(str(value) for value in values))
|
|
|
|
|
|
def write_json(path: Path, payload: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def rate_key(rate: float) -> str:
|
|
return f"r_{rate:.3f}".replace(".", "p")
|
|
|
|
|
|
def _percentile(values: list[float], fraction: float) -> float | None:
|
|
if not values:
|
|
return None
|
|
ordered = sorted(values)
|
|
index = min(len(ordered) - 1, max(0, math.ceil(fraction * len(ordered)) - 1))
|
|
return float(ordered[index])
|
|
|
|
|
|
def _latency_summary(values: list[float]) -> dict[str, Any]:
|
|
return {
|
|
"count": len(values),
|
|
"mean": statistics.fmean(values) if values else None,
|
|
"p50": _percentile(values, 0.50),
|
|
"p95": _percentile(values, 0.95),
|
|
"p99": _percentile(values, 0.99),
|
|
"max": max(values) if values else None,
|
|
}
|
|
|
|
|
|
def slo_threshold_ms(variant: dict[str, Any], input_tokens: int) -> float:
|
|
if variant["kind"] == "linear_ms":
|
|
return float(variant["intercept_ms"]) + float(variant["per_token_ms"]) * input_tokens
|
|
if variant["kind"] != "step_ms":
|
|
raise ValueError(f"unsupported SLO kind: {variant['kind']}")
|
|
for bucket in variant["buckets"]:
|
|
ceiling = bucket.get("max_input_tokens")
|
|
if ceiling is None or input_tokens <= int(ceiling):
|
|
return float(bucket["threshold_ms"])
|
|
raise AssertionError("step SLO must have a terminal bucket")
|
|
|
|
|
|
def _input_bin(input_tokens: int) -> int:
|
|
for index, (low, high) in enumerate(zip(INPUT_BINS, INPUT_BINS[1:])):
|
|
if low <= input_tokens < high:
|
|
return index
|
|
raise ValueError(f"input length outside protocol bins: {input_tokens}")
|
|
|
|
|
|
def _allocate_quotas(counts: list[int], cohort_size: int) -> list[int]:
|
|
total = sum(counts)
|
|
if cohort_size <= 0 or cohort_size > total:
|
|
raise ValueError(f"invalid cohort size {cohort_size} for {total} rows")
|
|
exact = [cohort_size * count / total for count in counts]
|
|
quotas = [min(count, math.floor(value)) for count, value in zip(counts, exact)]
|
|
remaining = cohort_size - sum(quotas)
|
|
order = sorted(
|
|
range(len(counts)),
|
|
key=lambda index: (-(exact[index] - math.floor(exact[index])), index),
|
|
)
|
|
while remaining:
|
|
advanced = False
|
|
for index in order:
|
|
if quotas[index] >= counts[index]:
|
|
continue
|
|
quotas[index] += 1
|
|
remaining -= 1
|
|
advanced = True
|
|
if not remaining:
|
|
break
|
|
if not advanced:
|
|
raise AssertionError("could not allocate complete cohort")
|
|
return quotas
|
|
|
|
|
|
def load_source_rows(trace: Path) -> list[dict[str, Any]]:
|
|
rows = []
|
|
with trace.open(encoding="utf-8") as source:
|
|
for source_index, line in enumerate(source):
|
|
if not line.strip():
|
|
continue
|
|
raw = json.loads(line)
|
|
input_tokens = int(raw["input_length"])
|
|
if not 0 <= input_tokens <= 32768:
|
|
continue
|
|
prompt = raw.get("prompt")
|
|
if not isinstance(prompt, str) or not prompt:
|
|
raise ValueError(f"row {source_index} lacks raw prompt")
|
|
rows.append(
|
|
{
|
|
"source_row_index": source_index,
|
|
"source_request_id": str(
|
|
raw.get("request_id") or raw.get("id") or source_index
|
|
),
|
|
"source_arrival_s": float(raw["timestamp"]),
|
|
"sampling_u": float(raw["sampling_u"]),
|
|
"input_tokens": input_tokens,
|
|
"prompt_sha256": sha256_text(prompt),
|
|
"input_bin": _input_bin(input_tokens),
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def select_cohort(
|
|
rows: list[dict[str, Any]], *, cohort_size: int, seed: int
|
|
) -> tuple[list[dict[str, Any]], list[int]]:
|
|
by_bin = [[] for _ in range(len(INPUT_BINS) - 1)]
|
|
for row in rows:
|
|
by_bin[int(row["input_bin"])].append(row)
|
|
quotas = _allocate_quotas([len(items) for items in by_bin], cohort_size)
|
|
selected = []
|
|
for bin_index, (items, quota) in enumerate(zip(by_bin, quotas)):
|
|
ranked = sorted(
|
|
items,
|
|
key=lambda row: sha256_text(
|
|
f"{seed}|{bin_index}|{row['source_row_index']}|{row['prompt_sha256']}"
|
|
),
|
|
)
|
|
selected.extend(ranked[:quota])
|
|
selected.sort(key=lambda row: (row["source_arrival_s"], row["source_row_index"]))
|
|
if len(selected) != cohort_size:
|
|
raise AssertionError("cohort selection returned wrong size")
|
|
return selected, quotas
|
|
|
|
|
|
def prepare_protocol(args: argparse.Namespace) -> Path:
|
|
trace = args.trace.resolve()
|
|
actual_sha = sha256(trace)
|
|
if args.expected_trace_sha256 and actual_sha != args.expected_trace_sha256:
|
|
raise ValueError(
|
|
f"trace SHA256 mismatch: expected={args.expected_trace_sha256}, actual={actual_sha}"
|
|
)
|
|
rows = load_source_rows(trace)
|
|
cohort, quotas = select_cohort(rows, cohort_size=args.cohort_size, seed=args.seed)
|
|
first_arrival = float(cohort[0]["source_arrival_s"])
|
|
source_span = float(cohort[-1]["source_arrival_s"]) - first_arrival
|
|
if source_span <= 0:
|
|
raise ValueError("selected cohort has no arrival span")
|
|
|
|
output_root = args.output_root.resolve()
|
|
trace_dir = output_root / "traces"
|
|
trace_dir.mkdir(parents=True, exist_ok=True)
|
|
fields = [
|
|
"arrived_at",
|
|
"num_prefill_tokens",
|
|
"num_decode_tokens",
|
|
"source_row_index",
|
|
"source_request_id",
|
|
"source_arrival_s",
|
|
"sampling_u",
|
|
"slo_ttft_ms",
|
|
]
|
|
rates = {}
|
|
for rate in args.rate:
|
|
duration_s = len(cohort) / rate
|
|
target = trace_dir / f"{rate_key(rate)}.csv"
|
|
with target.open("w", encoding="utf-8", newline="") as output:
|
|
writer = csv.DictWriter(output, fieldnames=fields)
|
|
writer.writeheader()
|
|
for row in cohort:
|
|
arrival_s = (
|
|
(float(row["source_arrival_s"]) - first_arrival)
|
|
/ source_span
|
|
* duration_s
|
|
)
|
|
writer.writerow(
|
|
{
|
|
"arrived_at": f"{arrival_s:.12f}",
|
|
"num_prefill_tokens": row["input_tokens"],
|
|
"num_decode_tokens": 1,
|
|
"source_row_index": row["source_row_index"],
|
|
"source_request_id": row["source_request_id"],
|
|
"source_arrival_s": f"{row['source_arrival_s']:.12f}",
|
|
"sampling_u": f"{row['sampling_u']:.12f}",
|
|
"slo_ttft_ms": f"{slo_threshold_ms(SLO_VARIANTS[PRIMARY_SLO], int(row['input_tokens'])):.6f}",
|
|
}
|
|
)
|
|
rates[rate_key(rate)] = {
|
|
"offered_request_rate": rate,
|
|
"duration_s": duration_s,
|
|
"path": str(target.resolve()),
|
|
"sha256": sha256(target),
|
|
"request_count": len(cohort),
|
|
"request_rate_recomputed": len(cohort) / duration_s,
|
|
"source_row_order_sha256": order_hash(
|
|
row["source_row_index"] for row in cohort
|
|
),
|
|
"input_length_order_sha256": order_hash(
|
|
row["input_tokens"] for row in cohort
|
|
),
|
|
}
|
|
|
|
bin_counts = [0] * (len(INPUT_BINS) - 1)
|
|
for row in rows:
|
|
bin_counts[int(row["input_bin"])] += 1
|
|
manifest = {
|
|
"schema": PROTOCOL_SCHEMA,
|
|
"created_unix_s": time.time(),
|
|
"source": {
|
|
"path": str(trace),
|
|
"sha256": actual_sha,
|
|
"eligible_request_count": len(rows),
|
|
},
|
|
"selection": {
|
|
"method": "length_stratified_smallest_sha256",
|
|
"seed": args.seed,
|
|
"cohort_size": len(cohort),
|
|
"input_bin_edges": list(INPUT_BINS),
|
|
"source_bin_counts": bin_counts,
|
|
"cohort_bin_quotas": quotas,
|
|
"stable_order": ["source_arrival_s", "source_row_index"],
|
|
"cohort_source_row_order_sha256": order_hash(
|
|
row["source_row_index"] for row in cohort
|
|
),
|
|
"source_arrival_span_s": source_span,
|
|
},
|
|
"load_contract": {
|
|
"only_mutated_variable": "uniform_arrival_timeline_scale",
|
|
"request_count_per_point": len(cohort),
|
|
"completion_tokens_override": 1,
|
|
"target_pass_rate": TARGET_PASS_RATE,
|
|
"binary_search": False,
|
|
"monotonicity_assumed": False,
|
|
"offered_rates": list(args.rate),
|
|
},
|
|
"slo_variants": SLO_VARIANTS,
|
|
"primary_slo": PRIMARY_SLO,
|
|
"cohort": cohort,
|
|
"rates": rates,
|
|
}
|
|
manifest_path = output_root / "protocol_manifest.json"
|
|
write_json(manifest_path, manifest)
|
|
audit_protocol(manifest_path)
|
|
print(manifest_path)
|
|
return manifest_path
|
|
|
|
|
|
def audit_protocol(manifest_path: Path) -> dict[str, Any]:
|
|
manifest = json.loads(manifest_path.read_text())
|
|
if manifest.get("schema") != PROTOCOL_SCHEMA:
|
|
raise ValueError(f"unexpected protocol schema: {manifest.get('schema')}")
|
|
expected_ids = [str(row["source_row_index"]) for row in manifest["cohort"]]
|
|
expected_lengths = [int(row["input_tokens"]) for row in manifest["cohort"]]
|
|
rate_checks = []
|
|
for record in manifest["rates"].values():
|
|
path = Path(record["path"])
|
|
if sha256(path) != record["sha256"]:
|
|
raise ValueError(f"trace hash mismatch: {path}")
|
|
with path.open(newline="") as source:
|
|
rows = list(csv.DictReader(source))
|
|
ids = [row["source_row_index"] for row in rows]
|
|
lengths = [int(row["num_prefill_tokens"]) for row in rows]
|
|
arrivals = [float(row["arrived_at"]) for row in rows]
|
|
if ids != expected_ids or lengths != expected_lengths:
|
|
raise ValueError(f"cohort/order drift in {path}")
|
|
if arrivals != sorted(arrivals) or abs(arrivals[0]) > 1e-9:
|
|
raise ValueError(f"invalid arrival sequence in {path}")
|
|
recomputed = len(rows) / (arrivals[-1] - arrivals[0])
|
|
if not math.isclose(
|
|
recomputed, float(record["offered_request_rate"]), rel_tol=1e-9
|
|
):
|
|
raise ValueError(f"offered-rate mismatch in {path}: {recomputed}")
|
|
rate_checks.append(
|
|
{"rate": record["offered_request_rate"], "recomputed_rate": recomputed}
|
|
)
|
|
result = {
|
|
"status": "passed",
|
|
"protocol_manifest_sha256": sha256(manifest_path),
|
|
"cohort_size": len(expected_ids),
|
|
"rate_checks": rate_checks,
|
|
}
|
|
write_json(manifest_path.parent / "protocol_audit.json", result)
|
|
return result
|
|
|
|
|
|
def _read_trace(path: Path) -> list[dict[str, Any]]:
|
|
with path.open(encoding="utf-8", newline="") as source:
|
|
return list(csv.DictReader(source))
|
|
|
|
|
|
def score_requests(requests: list[dict[str, Any]]) -> dict[str, Any]:
|
|
scores = {}
|
|
for name, variant in SLO_VARIANTS.items():
|
|
passed = 0
|
|
for request in requests:
|
|
threshold = slo_threshold_ms(variant, int(request["input_tokens"]))
|
|
passed += int(
|
|
bool(request["success"])
|
|
and request.get("ttft_ms") is not None
|
|
and float(request["ttft_ms"]) <= threshold
|
|
)
|
|
count = len(requests)
|
|
pass_rate = passed / count if count else 0.0
|
|
scores[name] = {
|
|
"request_count": count,
|
|
"passed_request_count": passed,
|
|
"slo_pass_rate": pass_rate,
|
|
"target_pass_rate": TARGET_PASS_RATE,
|
|
"feasible": pass_rate >= TARGET_PASS_RATE,
|
|
}
|
|
ttfts = [
|
|
float(request["ttft_ms"])
|
|
for request in requests
|
|
if request.get("ttft_ms") is not None
|
|
]
|
|
return {"scores": scores, "ttft_ms": _latency_summary(ttfts)}
|
|
|
|
|
|
def _capacity_record(
|
|
results: list[dict[str, Any]], *, config: dict[str, Any], slo_name: str
|
|
) -> dict[str, Any]:
|
|
ordered = sorted(results, key=lambda item: float(item["offered_request_rate"]))
|
|
feasible = [
|
|
float(item["offered_request_rate"])
|
|
for item in ordered
|
|
if item["scores"][slo_name]["feasible"]
|
|
]
|
|
vector = [bool(item["scores"][slo_name]["feasible"]) for item in ordered]
|
|
violations = []
|
|
for lower_index in range(len(vector)):
|
|
for higher_index in range(lower_index + 1, len(vector)):
|
|
if not vector[lower_index] and vector[higher_index]:
|
|
violations.append(
|
|
[
|
|
float(ordered[lower_index]["offered_request_rate"]),
|
|
float(ordered[higher_index]["offered_request_rate"]),
|
|
]
|
|
)
|
|
capacity = max(feasible) if feasible else None
|
|
tp = int(config["tp"])
|
|
return {
|
|
"config": config,
|
|
"slo": slo_name,
|
|
"tested_rates": [float(item["offered_request_rate"]) for item in ordered],
|
|
"pass_rates": [float(item["scores"][slo_name]["slo_pass_rate"]) for item in ordered],
|
|
"feasibility": vector,
|
|
"maximum_tested_feasible_request_rate": capacity,
|
|
"maximum_tested_feasible_request_rate_per_gpu": (
|
|
capacity / tp if capacity is not None else None
|
|
),
|
|
"lower_censored": capacity is None,
|
|
"upper_censored": bool(feasible) and capacity == max(item["offered_request_rate"] for item in ordered),
|
|
"monotonicity_violations": violations,
|
|
}
|
|
|
|
|
|
def rank_surface(
|
|
config_results: list[dict[str, Any]], *, slo_name: str
|
|
) -> list[dict[str, Any]]:
|
|
records = [
|
|
_capacity_record(item["loads"], config=item["config"], slo_name=slo_name)
|
|
for item in config_results
|
|
]
|
|
records.sort(
|
|
key=lambda item: (
|
|
-(
|
|
item["maximum_tested_feasible_request_rate_per_gpu"]
|
|
if item["maximum_tested_feasible_request_rate_per_gpu"] is not None
|
|
else -1.0
|
|
),
|
|
item["config"]["name"],
|
|
)
|
|
)
|
|
previous_value = object()
|
|
previous_rank = 0
|
|
for index, record in enumerate(records, start=1):
|
|
value = record["maximum_tested_feasible_request_rate_per_gpu"]
|
|
if value != previous_value:
|
|
previous_rank = index
|
|
previous_value = value
|
|
record["rank"] = previous_rank
|
|
return records
|
|
|
|
|
|
def _validate_profile_contract(profile_paths: dict[str, Path]) -> dict[str, Any]:
|
|
payload = json.loads(profile_paths["manifest"].read_text())
|
|
contract = payload.get("contract") or {}
|
|
if contract.get("quant_signature") != EXPECTED_QUANT_SIGNATURE:
|
|
raise ValueError(f"unexpected FP8 profile contract: {contract}")
|
|
if contract.get("profiling_precision") != "BF16":
|
|
raise ValueError(f"unexpected output/accumulation precision: {contract}")
|
|
for name, path in (
|
|
("linear_op.csv", profile_paths["linear"]),
|
|
("attention.csv", profile_paths["attention"]),
|
|
("moe.csv", profile_paths["moe"]),
|
|
):
|
|
expected = payload["outputs"][name]["sha256"]
|
|
if sha256(path) != expected:
|
|
raise ValueError(f"profile output hash mismatch: {path}")
|
|
return contract
|
|
|
|
|
|
def run_frontier_load(
|
|
*,
|
|
args: argparse.Namespace,
|
|
profile_paths: dict[str, Path],
|
|
config: Any,
|
|
rate_record: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
rate = float(rate_record["offered_request_rate"])
|
|
load_dir = args.output_root / "runs" / config.name / rate_key(rate)
|
|
result_path = load_dir / "result.json"
|
|
if result_path.is_file():
|
|
result = json.loads(result_path.read_text())
|
|
if result.get("status") == "completed":
|
|
return result
|
|
load_dir.mkdir(parents=True, exist_ok=True)
|
|
trace = Path(rate_record["path"])
|
|
command = frontier_grid.build_command(
|
|
python=args.python,
|
|
frontier_source=args.frontier_source,
|
|
profile_root=args.profile_root,
|
|
profile_paths=profile_paths,
|
|
trace=trace,
|
|
config=config,
|
|
probe_dir=load_dir,
|
|
run_id=f"{config.name}_{rate_key(rate)}",
|
|
cache_root=args.output_root / "cache" / config.name,
|
|
)
|
|
write_json(load_dir / "command.json", command)
|
|
environment = os.environ.copy()
|
|
environment.update(
|
|
{
|
|
"PYTHONPATH": str(args.frontier_source),
|
|
"WANDB_DISABLED": "true",
|
|
"VIDUR_DISABLE_WANDB": "1",
|
|
}
|
|
)
|
|
started = time.time()
|
|
with (load_dir / "stdout.log").open("w", encoding="utf-8") as output:
|
|
completed = subprocess.run(
|
|
command,
|
|
cwd=args.frontier_source,
|
|
env=environment,
|
|
stdout=output,
|
|
stderr=subprocess.STDOUT,
|
|
check=False,
|
|
)
|
|
if completed.returncode != 0:
|
|
raise RuntimeError(
|
|
f"Frontier failed: config={config.name}, rate={rate}, rc={completed.returncode}"
|
|
)
|
|
request_metrics = frontier_grid.find_request_metrics(load_dir)
|
|
trace_rows = _read_trace(trace)
|
|
with request_metrics.open(encoding="utf-8", newline="") as source:
|
|
metric_rows = list(csv.DictReader(source))
|
|
metrics_by_id = {int(row["Request Id"]): row for row in metric_rows}
|
|
if set(metrics_by_id) != set(range(len(trace_rows))):
|
|
raise ValueError(f"Frontier Request Id mismatch: {request_metrics}")
|
|
requests = [
|
|
{
|
|
"request_id": row["source_row_index"],
|
|
"input_tokens": int(row["num_prefill_tokens"]),
|
|
"success": True,
|
|
"ttft_ms": float(metrics_by_id[index]["ttft"]),
|
|
}
|
|
for index, row in enumerate(trace_rows)
|
|
]
|
|
score = score_requests(requests)
|
|
result = {
|
|
"status": "completed",
|
|
"config": asdict(config) | {"name": config.name},
|
|
"offered_request_rate": rate,
|
|
"request_rate_per_gpu": rate / config.gpu_count,
|
|
"elapsed_seconds": time.time() - started,
|
|
"trace_path": str(trace),
|
|
"trace_sha256": sha256(trace),
|
|
"request_metrics_path": str(request_metrics),
|
|
"request_metrics_sha256": sha256(request_metrics),
|
|
"requests": requests,
|
|
**score,
|
|
}
|
|
write_json(result_path, result)
|
|
return result
|
|
|
|
|
|
def run_frontier(args: argparse.Namespace) -> Path:
|
|
protocol = json.loads(args.protocol_manifest.read_text())
|
|
if protocol.get("schema") != PROTOCOL_SCHEMA:
|
|
raise ValueError("invalid fixed-cohort protocol manifest")
|
|
audit_protocol(args.protocol_manifest)
|
|
profile_paths = frontier_grid.resolve_profile_paths(args.profile_root)
|
|
profile_contract = _validate_profile_contract(profile_paths)
|
|
cache_equivalence = json.loads(args.cache_equivalence_artifact.read_text())
|
|
if (
|
|
cache_equivalence.get("schema")
|
|
!= "frontier-ep-critical-lane-cache-equivalence-v1"
|
|
or not cache_equivalence.get("byte_identical")
|
|
):
|
|
raise ValueError("EP critical-lane cache lacks byte-identical evidence")
|
|
if Path(cache_equivalence["new_source"]).resolve() != args.frontier_source.resolve():
|
|
raise ValueError("cache equivalence was measured on a different Frontier source")
|
|
configs = frontier_grid.selected_configs(args.config)
|
|
args.output_root.mkdir(parents=True, exist_ok=True)
|
|
run_manifest = {
|
|
"schema": FRONTIER_SCHEMA,
|
|
"created_unix_s": time.time(),
|
|
"protocol_manifest": {
|
|
"path": str(args.protocol_manifest.resolve()),
|
|
"sha256": sha256(args.protocol_manifest),
|
|
},
|
|
"frontier": {
|
|
"source": str(args.frontier_source.resolve()),
|
|
"python": str(args.python.resolve()),
|
|
"fingerprint": frontier_grid.frontier_source_fingerprint(
|
|
args.frontier_source, args.frontier_commit
|
|
),
|
|
},
|
|
"profiles": {
|
|
"contract": profile_contract,
|
|
"files": {
|
|
name: {"path": str(path.resolve()), "sha256": sha256(path)}
|
|
for name, path in profile_paths.items()
|
|
},
|
|
},
|
|
"ep_critical_lane_cache_equivalence": {
|
|
"path": str(args.cache_equivalence_artifact.resolve()),
|
|
"sha256": sha256(args.cache_equivalence_artifact),
|
|
"request_metrics_byte_identical": True,
|
|
"request_metrics_sha256": cache_equivalence["new_sha256"],
|
|
},
|
|
"kv_capacity_evidence": {
|
|
"tp4": {
|
|
"path": str(args.tp4_capacity_artifact.resolve()),
|
|
"sha256": sha256(args.tp4_capacity_artifact),
|
|
"num_gpu_blocks": 26101,
|
|
},
|
|
"tp8": {
|
|
"path": str(args.tp8_capacity_artifact.resolve()),
|
|
"sha256": sha256(args.tp8_capacity_artifact),
|
|
"num_gpu_blocks": 62351,
|
|
},
|
|
},
|
|
"best_effort_contract": {
|
|
"arrival_semantics": "preserve_materialized_trace_arrivals",
|
|
"quantization": "block_fp8_w8a8_dynamic_with_bf16_output_accumulation",
|
|
"execution_profiles": "community_vllm_0.10.2_serving_entrypoints",
|
|
"moe_ep_prefill": "global_routing_then_slowest_local_ep_lane",
|
|
"moe_ep_critical_lane_cache": "exact_(total_routed_tokens,layer_id)_memoization",
|
|
"kv_blocks": "measured_community_vllm_and_fixed_per_topology",
|
|
"cpu_overhead_modeling": "disabled_no_native_community_vllm_records",
|
|
"end_to_end_action_specific_calibration": False,
|
|
"real_serving_results_observed_before_freeze": False,
|
|
},
|
|
"orchestration": {
|
|
"max_parallel_configs": args.max_parallel_configs,
|
|
"cache_isolation": "one_cache_root_per_config",
|
|
"loads_within_config": "sequential",
|
|
},
|
|
"configs": [asdict(config) | {"name": config.name} for config in configs],
|
|
}
|
|
write_json(args.output_root / "run_manifest.json", run_manifest)
|
|
|
|
rate_records = sorted(
|
|
protocol["rates"].values(), key=lambda item: item["offered_request_rate"]
|
|
)
|
|
|
|
def run_config(config: Any) -> dict[str, Any]:
|
|
loads = []
|
|
for record in rate_records:
|
|
result = run_frontier_load(
|
|
args=args,
|
|
profile_paths=profile_paths,
|
|
config=config,
|
|
rate_record=record,
|
|
)
|
|
loads.append(result)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"system": "Frontier",
|
|
"config": config.name,
|
|
"rate": result["offered_request_rate"],
|
|
"pass_rate": result["scores"][PRIMARY_SLO]["slo_pass_rate"],
|
|
"feasible": result["scores"][PRIMARY_SLO]["feasible"],
|
|
"elapsed_seconds": result["elapsed_seconds"],
|
|
},
|
|
sort_keys=True,
|
|
),
|
|
flush=True,
|
|
)
|
|
summary = {"config": asdict(config) | {"name": config.name}, "loads": loads}
|
|
write_json(args.output_root / "results" / f"{config.name}.json", summary)
|
|
return summary
|
|
|
|
results_by_name = {}
|
|
with ThreadPoolExecutor(max_workers=args.max_parallel_configs) as pool:
|
|
futures = {pool.submit(run_config, config): config for config in configs}
|
|
for future in as_completed(futures):
|
|
config = futures[future]
|
|
results_by_name[config.name] = future.result()
|
|
config_results = [results_by_name[config.name] for config in configs]
|
|
freeze = {
|
|
"schema": FRONTIER_SCHEMA,
|
|
"status": "frozen_before_community_run",
|
|
"created_unix_s": time.time(),
|
|
"run_manifest_sha256": sha256(args.output_root / "run_manifest.json"),
|
|
"protocol_manifest_sha256": sha256(args.protocol_manifest),
|
|
"primary_slo": PRIMARY_SLO,
|
|
"rankings": {
|
|
name: rank_surface(config_results, slo_name=name) for name in SLO_VARIANTS
|
|
},
|
|
"config_results": config_results,
|
|
}
|
|
path = args.output_root / "frontier_ranking_frozen.json"
|
|
write_json(path, freeze)
|
|
print(path)
|
|
return path
|
|
|
|
|
|
def _community_study_payload(
|
|
*,
|
|
tp: int,
|
|
repo: Path,
|
|
python: Path,
|
|
vllm: Path,
|
|
model: Path,
|
|
trace: Path,
|
|
windows: Path,
|
|
port: int,
|
|
) -> dict[str, Any]:
|
|
payload = community_grid.study_payload(
|
|
tp=tp,
|
|
repo=repo,
|
|
python=python,
|
|
vllm=vllm,
|
|
model=model,
|
|
trace=trace,
|
|
windows=windows,
|
|
port=port,
|
|
)
|
|
payload["study_id"] = f"community-qwen235b-prefill-fixed-cohort-tp{tp}-v1"
|
|
payload["slo"] = {
|
|
"target_pass_rate": TARGET_PASS_RATE,
|
|
"ttft_rule": {
|
|
key: value
|
|
for key, value in SLO_VARIANTS[PRIMARY_SLO].items()
|
|
if key != "description"
|
|
},
|
|
}
|
|
# The dedicated runner supplies exact fixed-cohort arrivals and disables
|
|
# SLO-based early stopping; these trace values only define raw prompt loading.
|
|
payload["trace"]["early_stop_max_lag_s"] = None
|
|
payload["trace"]["early_stop_max_elapsed_s"] = None
|
|
return payload
|
|
|
|
|
|
def _validate_frontier_freeze(path: Path, protocol_manifest: Path) -> dict[str, Any]:
|
|
freeze = json.loads(path.read_text())
|
|
if freeze.get("schema") != FRONTIER_SCHEMA:
|
|
raise ValueError(f"unexpected Frontier freeze schema: {freeze.get('schema')}")
|
|
if freeze.get("status") != "frozen_before_community_run":
|
|
raise ValueError("Frontier result is not marked frozen")
|
|
if freeze.get("protocol_manifest_sha256") != sha256(protocol_manifest):
|
|
raise ValueError("Frontier and community protocol manifests differ")
|
|
if len(freeze.get("config_results") or []) != len(frontier_grid.GRID):
|
|
raise ValueError("Frontier freeze is incomplete")
|
|
expected_rates = set(OFFERED_RATES)
|
|
for item in freeze["config_results"]:
|
|
rates = {float(load["offered_request_rate"]) for load in item["loads"]}
|
|
if rates != expected_rates:
|
|
raise ValueError(f"Frontier freeze lacks rates for {item['config']['name']}")
|
|
return freeze
|
|
|
|
|
|
def prepare_community(args: argparse.Namespace) -> Path:
|
|
protocol = json.loads(args.protocol_manifest.read_text())
|
|
if protocol.get("schema") != PROTOCOL_SCHEMA:
|
|
raise ValueError("invalid fixed-cohort protocol manifest")
|
|
_validate_frontier_freeze(args.frontier_freeze, args.protocol_manifest)
|
|
repo = args.repo.resolve()
|
|
python = args.python.resolve()
|
|
vllm = args.vllm.resolve()
|
|
model = args.model.resolve()
|
|
windows = args.windows.resolve()
|
|
trace = Path(protocol["source"]["path"])
|
|
for path in (repo, python, vllm, model, windows, trace):
|
|
if not path.exists():
|
|
raise FileNotFoundError(path)
|
|
if sha256(trace) != protocol["source"]["sha256"]:
|
|
raise ValueError("raw prompt trace changed after protocol freeze")
|
|
output_root = args.output_root.resolve()
|
|
output_root.mkdir(parents=True, exist_ok=True)
|
|
studies = {}
|
|
for tp in (4, 8):
|
|
path = output_root / "studies" / f"tp{tp}" / "study.json"
|
|
write_json(
|
|
path,
|
|
_community_study_payload(
|
|
tp=tp,
|
|
repo=repo,
|
|
python=python,
|
|
vllm=vllm,
|
|
model=model,
|
|
trace=trace,
|
|
windows=windows,
|
|
port=args.port,
|
|
),
|
|
)
|
|
studies[str(tp)] = {"path": str(path), "sha256": sha256(path)}
|
|
|
|
configs = list(community_grid.GRID)
|
|
random.Random(args.config_order_seed).shuffle(configs)
|
|
records = []
|
|
for execution_index, config in enumerate(configs, start=1):
|
|
rates = list(OFFERED_RATES)
|
|
random.Random(f"{args.rate_order_seed}|{config.name}").shuffle(rates)
|
|
records.append(
|
|
{
|
|
"execution_index": execution_index,
|
|
"config": asdict(config) | {"name": config.name},
|
|
"rate_order": rates,
|
|
"study_path": studies[str(config.tp)]["path"],
|
|
"result_path": str(output_root / "results" / f"{config.name}.json"),
|
|
"engine_log_path": str(output_root / "runs" / config.name / "engine.log"),
|
|
}
|
|
)
|
|
total_replay_seconds = sum(
|
|
float(record["duration_s"]) for record in protocol["rates"].values()
|
|
) * len(configs)
|
|
manifest = {
|
|
"schema": COMMUNITY_SCHEMA,
|
|
"created_unix_s": time.time(),
|
|
"repository": community_grid.git_fingerprint(repo),
|
|
"protocol_manifest": {
|
|
"path": str(args.protocol_manifest.resolve()),
|
|
"sha256": sha256(args.protocol_manifest),
|
|
},
|
|
"frontier_freeze": {
|
|
"path": str(args.frontier_freeze.resolve()),
|
|
"sha256": sha256(args.frontier_freeze),
|
|
},
|
|
"runtime": {"python": str(python), "vllm": str(vllm)},
|
|
"model": {
|
|
"path": str(model),
|
|
"config_sha256": sha256(model / "config.json"),
|
|
},
|
|
"studies": studies,
|
|
"execution_contract": {
|
|
"request_mode": "raw_completion",
|
|
"completion_tokens_override": 1,
|
|
"prefix_caching": False,
|
|
"client_max_concurrency": community_grid.CLIENT_MAX_CONCURRENCY,
|
|
"slo_early_stop": False,
|
|
"one_server_launch_per_config": True,
|
|
"warmup": {
|
|
"method": "one_disjoint_request_per_input_length_bin",
|
|
"request_count": len(INPUT_BINS) - 1,
|
|
"seed": WARMUP_SEED,
|
|
"latencies_discarded": True,
|
|
},
|
|
"rate_order_randomized_within_config": True,
|
|
"config_order_seed": args.config_order_seed,
|
|
"rate_order_seed": args.rate_order_seed,
|
|
"estimated_replay_wall_seconds_excluding_8_model_loads": total_replay_seconds,
|
|
},
|
|
"configs": records,
|
|
}
|
|
path = output_root / "run_manifest.json"
|
|
write_json(path, manifest)
|
|
print(path)
|
|
return path
|
|
|
|
|
|
def _materialize_real_requests(
|
|
*,
|
|
all_requests: list[TraceRequest],
|
|
rate_record: dict[str, Any],
|
|
cohort: list[dict[str, Any]],
|
|
) -> list[TraceRequest]:
|
|
by_id = {request.row_id: request for request in all_requests}
|
|
expected_prompt_hash = {
|
|
str(row["source_row_index"]): row["prompt_sha256"] for row in cohort
|
|
}
|
|
trace_rows = _read_trace(Path(rate_record["path"]))
|
|
requests = []
|
|
for row in trace_rows:
|
|
request_id = row["source_row_index"]
|
|
if request_id not in by_id:
|
|
raise ValueError(f"cohort request is absent from raw trace loader: {request_id}")
|
|
request = by_id[request_id]
|
|
prompt = request.body.get("prompt")
|
|
if not isinstance(prompt, str) or sha256_text(prompt) != expected_prompt_hash[request_id]:
|
|
raise ValueError(f"raw prompt fingerprint changed: request={request_id}")
|
|
requests.append(replace(request, arrival_s=float(row["arrived_at"])))
|
|
return requests
|
|
|
|
|
|
def _select_warmup_requests(
|
|
all_requests: list[TraceRequest], *, cohort_ids: set[str]
|
|
) -> list[TraceRequest]:
|
|
by_bin = [[] for _ in range(len(INPUT_BINS) - 1)]
|
|
for request in all_requests:
|
|
if request.row_id in cohort_ids or request.prompt_tokens_hint is None:
|
|
continue
|
|
input_tokens = int(request.prompt_tokens_hint)
|
|
if not 0 <= input_tokens <= 32768:
|
|
continue
|
|
by_bin[_input_bin(input_tokens)].append(request)
|
|
selected = []
|
|
for bin_index, candidates in enumerate(by_bin):
|
|
if not candidates:
|
|
raise ValueError(f"no disjoint warmup request in input bin {bin_index}")
|
|
selected.append(
|
|
min(
|
|
candidates,
|
|
key=lambda request: sha256_text(
|
|
f"{WARMUP_SEED}|{bin_index}|{request.row_id}"
|
|
),
|
|
)
|
|
)
|
|
return [replace(request, arrival_s=0.0) for request in selected]
|
|
|
|
|
|
def _run_real_replay(
|
|
requests: list[TraceRequest], *, recipe: Any, study: Any, max_elapsed_s: float
|
|
) -> tuple[list[Any], bool, str]:
|
|
return _replay_requests(
|
|
requests,
|
|
base_url=recipe.base_url,
|
|
timeout_s=recipe.request_timeout_s,
|
|
max_concurrency=study.trace.max_concurrency,
|
|
# target=0 makes the mathematical SLO early-stop bound unreachable
|
|
# while retaining the existing replay engine.
|
|
target_pass_rate=0.0,
|
|
max_lag_s=None,
|
|
max_elapsed_s=max_elapsed_s,
|
|
evaluate_outcome=lambda outcome: type(
|
|
"NoEarlyStopEvaluation", (), {"passed": bool(outcome.success)}
|
|
)(),
|
|
)
|
|
|
|
|
|
def _real_request_records(
|
|
requests: list[TraceRequest], outcomes: list[Any]
|
|
) -> list[dict[str, Any]]:
|
|
outcomes_by_id = {outcome.request_id: outcome for outcome in outcomes}
|
|
records = []
|
|
for request in requests:
|
|
outcome = outcomes_by_id.get(request.row_id)
|
|
records.append(
|
|
{
|
|
"request_id": request.row_id,
|
|
"arrival_s": request.arrival_s,
|
|
"input_tokens": int(request.prompt_tokens_hint or 0),
|
|
"success": bool(outcome and outcome.success),
|
|
"ttft_ms": outcome.ttft_ms if outcome is not None else None,
|
|
"tpot_ms": outcome.tpot_ms if outcome is not None else None,
|
|
"completion_tokens": outcome.completion_tokens if outcome is not None else None,
|
|
"completion_tokens_source": (
|
|
outcome.completion_tokens_source if outcome is not None else ""
|
|
),
|
|
"error": outcome.error if outcome is not None else "missing_outcome",
|
|
}
|
|
)
|
|
return records
|
|
|
|
|
|
def run_community_config(
|
|
*,
|
|
manifest: dict[str, Any],
|
|
record: dict[str, Any],
|
|
protocol: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
result_path = Path(record["result_path"])
|
|
if result_path.is_file():
|
|
result = json.loads(result_path.read_text())
|
|
if result.get("status") == "completed":
|
|
return result
|
|
config = record["config"]
|
|
study_path = Path(record["study_path"])
|
|
study = load_study_spec(study_path)
|
|
_, all_requests = load_trace_requests(study, study_spec_path=study_path)
|
|
recipe = build_launch_recipe(
|
|
study.engine,
|
|
ConfigPatch(
|
|
flag_patch={
|
|
"max-num-seqs": int(config["mns"]),
|
|
"max-num-batched-tokens": int(config["mbt"]),
|
|
}
|
|
),
|
|
)
|
|
run_dir = Path(record["engine_log_path"]).parent
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
write_json(run_dir / "engine_command.json", recipe.argv)
|
|
marker_env = {
|
|
"AITUNER_STUDY_ID": study.study_id,
|
|
"AITUNER_TRIAL_ID": f"fixed-cohort-{config['name']}",
|
|
}
|
|
loads = []
|
|
load_by_rate = {
|
|
float(item["offered_request_rate"]): item
|
|
for item in protocol["rates"].values()
|
|
}
|
|
started = time.time()
|
|
with Path(record["engine_log_path"]).open("w", encoding="utf-8") as engine_log:
|
|
process = subprocess.Popen( # noqa: S603
|
|
recipe.argv,
|
|
cwd=recipe.cwd,
|
|
env={**recipe.env, **marker_env},
|
|
stdout=engine_log,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
start_new_session=True,
|
|
)
|
|
previous_sigterm = _install_sigterm_as_keyboardinterrupt()
|
|
try:
|
|
_wait_for_server_or_exit(
|
|
process,
|
|
base_url=recipe.base_url,
|
|
healthcheck_path=recipe.healthcheck_path,
|
|
ready_timeout_s=recipe.ready_timeout_s,
|
|
)
|
|
cohort_ids = {
|
|
str(row["source_row_index"]) for row in protocol["cohort"]
|
|
}
|
|
warmup_requests = _select_warmup_requests(
|
|
all_requests, cohort_ids=cohort_ids
|
|
)
|
|
warmup_started = time.time()
|
|
warmup_outcomes, warmup_early_stopped, warmup_stop_reason = _run_real_replay(
|
|
warmup_requests,
|
|
recipe=recipe,
|
|
study=study,
|
|
max_elapsed_s=1200.0,
|
|
)
|
|
warmup_records = _real_request_records(warmup_requests, warmup_outcomes)
|
|
warmup_result = {
|
|
"status": "completed"
|
|
if not warmup_early_stopped
|
|
and all(item["success"] for item in warmup_records)
|
|
else "failed",
|
|
"elapsed_seconds": time.time() - warmup_started,
|
|
"early_stopped": warmup_early_stopped,
|
|
"early_stop_reason": warmup_stop_reason,
|
|
"cohort_disjoint": not bool(
|
|
cohort_ids & {item["request_id"] for item in warmup_records}
|
|
),
|
|
"requests": warmup_records,
|
|
}
|
|
write_json(run_dir / "warmup.json", warmup_result)
|
|
if warmup_result["status"] != "completed":
|
|
raise RuntimeError(f"server warmup failed: {warmup_result}")
|
|
time.sleep(2.0)
|
|
for rate in record["rate_order"]:
|
|
rate = float(rate)
|
|
load_dir = run_dir / rate_key(rate)
|
|
load_result_path = load_dir / "result.json"
|
|
if load_result_path.is_file():
|
|
load_result = json.loads(load_result_path.read_text())
|
|
if load_result.get("status") == "completed":
|
|
loads.append(load_result)
|
|
continue
|
|
load_dir.mkdir(parents=True, exist_ok=True)
|
|
rate_record = load_by_rate[rate]
|
|
requests = _materialize_real_requests(
|
|
all_requests=all_requests,
|
|
rate_record=rate_record,
|
|
cohort=protocol["cohort"],
|
|
)
|
|
load_started = time.time()
|
|
outcomes, early_stopped, early_stop_reason = _run_real_replay(
|
|
requests,
|
|
recipe=recipe,
|
|
study=study,
|
|
max_elapsed_s=max(float(rate_record["duration_s"]) + 900.0, 1200.0),
|
|
)
|
|
request_records = _real_request_records(requests, outcomes)
|
|
score = score_requests(request_records)
|
|
load_result = {
|
|
"status": "completed",
|
|
"config": config,
|
|
"offered_request_rate": rate,
|
|
"request_rate_per_gpu": rate / int(config["tp"]),
|
|
"elapsed_seconds": time.time() - load_started,
|
|
"trace_path": rate_record["path"],
|
|
"trace_sha256": rate_record["sha256"],
|
|
"early_stopped": early_stopped,
|
|
"early_stop_reason": early_stop_reason,
|
|
"requests": request_records,
|
|
**score,
|
|
}
|
|
if early_stopped:
|
|
raise RuntimeError(
|
|
f"complete replay stopped early: {config['name']} rate={rate}: "
|
|
f"{early_stop_reason}"
|
|
)
|
|
write_json(load_result_path, load_result)
|
|
loads.append(load_result)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"system": "community_vllm",
|
|
"config": config["name"],
|
|
"rate": rate,
|
|
"pass_rate": score["scores"][PRIMARY_SLO]["slo_pass_rate"],
|
|
"feasible": score["scores"][PRIMARY_SLO]["feasible"],
|
|
"elapsed_seconds": load_result["elapsed_seconds"],
|
|
},
|
|
sort_keys=True,
|
|
),
|
|
flush=True,
|
|
)
|
|
time.sleep(2.0)
|
|
finally:
|
|
_ignore_sigterm_if_main()
|
|
_terminate_process_tree(process, timeout_s=30.0, marker_env=marker_env)
|
|
_restore_sigterm(previous_sigterm)
|
|
loads.sort(key=lambda item: float(item["offered_request_rate"]))
|
|
result = {
|
|
"schema": COMMUNITY_SCHEMA,
|
|
"status": "completed",
|
|
"config": config,
|
|
"elapsed_seconds": time.time() - started,
|
|
"loads": loads,
|
|
}
|
|
write_json(result_path, result)
|
|
return result
|
|
|
|
|
|
def run_community(args: argparse.Namespace) -> Path:
|
|
manifest = json.loads(args.manifest.read_text())
|
|
if manifest.get("schema") != COMMUNITY_SCHEMA:
|
|
raise ValueError(f"unexpected community manifest schema: {manifest.get('schema')}")
|
|
protocol_path = Path(manifest["protocol_manifest"]["path"])
|
|
frontier_freeze = Path(manifest["frontier_freeze"]["path"])
|
|
if sha256(protocol_path) != manifest["protocol_manifest"]["sha256"]:
|
|
raise ValueError("protocol changed after community manifest preparation")
|
|
if sha256(frontier_freeze) != manifest["frontier_freeze"]["sha256"]:
|
|
raise ValueError("Frontier freeze changed after community manifest preparation")
|
|
_validate_frontier_freeze(frontier_freeze, protocol_path)
|
|
protocol = json.loads(protocol_path.read_text())
|
|
config_results = []
|
|
for record in sorted(manifest["configs"], key=lambda item: item["execution_index"]):
|
|
result = run_community_config(
|
|
manifest=manifest,
|
|
record=record,
|
|
protocol=protocol,
|
|
)
|
|
config_results.append(result)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"config": result["config"]["name"],
|
|
"config_completed": True,
|
|
"elapsed_seconds": result["elapsed_seconds"],
|
|
},
|
|
sort_keys=True,
|
|
),
|
|
flush=True,
|
|
)
|
|
freeze = {
|
|
"schema": COMMUNITY_SCHEMA,
|
|
"status": "completed",
|
|
"created_unix_s": time.time(),
|
|
"run_manifest_sha256": sha256(args.manifest),
|
|
"protocol_manifest_sha256": sha256(protocol_path),
|
|
"frontier_freeze_sha256": sha256(frontier_freeze),
|
|
"primary_slo": PRIMARY_SLO,
|
|
"rankings": {
|
|
name: rank_surface(config_results, slo_name=name) for name in SLO_VARIANTS
|
|
},
|
|
"config_results": config_results,
|
|
}
|
|
path = args.manifest.parent / "community_ranking_frozen.json"
|
|
write_json(path, freeze)
|
|
print(path)
|
|
return path
|
|
|
|
|
|
def _rank_map(ranking: list[dict[str, Any]]) -> dict[str, int]:
|
|
return {item["config"]["name"]: int(item["rank"]) for item in ranking}
|
|
|
|
|
|
def _capacity_map(ranking: list[dict[str, Any]]) -> dict[str, float | None]:
|
|
return {
|
|
item["config"]["name"]: item["maximum_tested_feasible_request_rate_per_gpu"]
|
|
for item in ranking
|
|
}
|
|
|
|
|
|
def _pearson(left: list[float], right: list[float]) -> float | None:
|
|
if len(left) < 2:
|
|
return None
|
|
left_mean = statistics.fmean(left)
|
|
right_mean = statistics.fmean(right)
|
|
numerator = sum((a - left_mean) * (b - right_mean) for a, b in zip(left, right))
|
|
denominator = math.sqrt(
|
|
sum((a - left_mean) ** 2 for a in left)
|
|
* sum((b - right_mean) ** 2 for b in right)
|
|
)
|
|
return numerator / denominator if denominator else None
|
|
|
|
|
|
def _average_ranks(capacities: dict[str, float | None]) -> dict[str, float]:
|
|
ordered = sorted(
|
|
capacities,
|
|
key=lambda name: (
|
|
-(
|
|
capacities[name]
|
|
if capacities[name] is not None
|
|
else float("-inf")
|
|
),
|
|
name,
|
|
),
|
|
)
|
|
ranks = {}
|
|
start = 0
|
|
while start < len(ordered):
|
|
value = capacities[ordered[start]]
|
|
end = start + 1
|
|
while end < len(ordered) and capacities[ordered[end]] == value:
|
|
end += 1
|
|
average_rank = ((start + 1) + end) / 2.0
|
|
for name in ordered[start:end]:
|
|
ranks[name] = average_rank
|
|
start = end
|
|
return ranks
|
|
|
|
|
|
def _pairwise_accuracy(
|
|
frontier_capacity: dict[str, float | None], real_capacity: dict[str, float | None]
|
|
) -> dict[str, Any]:
|
|
names = sorted(set(frontier_capacity) & set(real_capacity))
|
|
comparable = 0
|
|
correct = 0
|
|
frontier_ties = 0
|
|
real_ties = 0
|
|
for left_index, left in enumerate(names):
|
|
for right in names[left_index + 1 :]:
|
|
f_left = frontier_capacity[left]
|
|
f_right = frontier_capacity[right]
|
|
r_left = real_capacity[left]
|
|
r_right = real_capacity[right]
|
|
if None in (f_left, f_right, r_left, r_right):
|
|
continue
|
|
f_sign = (f_left > f_right) - (f_left < f_right)
|
|
r_sign = (r_left > r_right) - (r_left < r_right)
|
|
frontier_ties += int(f_sign == 0)
|
|
real_ties += int(r_sign == 0)
|
|
if f_sign == 0 or r_sign == 0:
|
|
continue
|
|
comparable += 1
|
|
correct += int(f_sign == r_sign)
|
|
return {
|
|
"comparable_non_tied_pairs": comparable,
|
|
"correct_pairs": correct,
|
|
"accuracy": correct / comparable if comparable else None,
|
|
"frontier_tied_pairs": frontier_ties,
|
|
"real_tied_pairs": real_ties,
|
|
}
|
|
|
|
|
|
def _request_residuals(
|
|
frontier_results: list[dict[str, Any]], real_results: list[dict[str, Any]]
|
|
) -> dict[str, Any]:
|
|
def index(results: list[dict[str, Any]]) -> dict[tuple[str, float, str], dict[str, Any]]:
|
|
indexed = {}
|
|
for config in results:
|
|
name = config["config"]["name"]
|
|
for load in config["loads"]:
|
|
rate = float(load["offered_request_rate"])
|
|
for request in load["requests"]:
|
|
indexed[(name, rate, str(request["request_id"]))] = request
|
|
return indexed
|
|
|
|
frontier_index = index(frontier_results)
|
|
real_index = index(real_results)
|
|
keys = sorted(set(frontier_index) & set(real_index))
|
|
errors = []
|
|
absolute_percentage_errors = []
|
|
by_bin: dict[str, list[float]] = {}
|
|
for key in keys:
|
|
frontier_request = frontier_index[key]
|
|
real_request = real_index[key]
|
|
if not real_request["success"] or real_request.get("ttft_ms") is None:
|
|
continue
|
|
predicted = float(frontier_request["ttft_ms"])
|
|
observed = float(real_request["ttft_ms"])
|
|
error = predicted - observed
|
|
errors.append(error)
|
|
if observed > 0:
|
|
absolute_percentage_errors.append(abs(error) / observed)
|
|
input_tokens = int(real_request["input_tokens"])
|
|
bin_index = _input_bin(input_tokens)
|
|
label = f"[{INPUT_BINS[bin_index]},{INPUT_BINS[bin_index + 1]})"
|
|
by_bin.setdefault(label, []).append(error)
|
|
return {
|
|
"matched_successful_requests": len(errors),
|
|
"signed_error_ms_mean": statistics.fmean(errors) if errors else None,
|
|
"mae_ms": statistics.fmean(abs(value) for value in errors) if errors else None,
|
|
"mape": (
|
|
statistics.fmean(absolute_percentage_errors)
|
|
if absolute_percentage_errors
|
|
else None
|
|
),
|
|
"by_input_bin": {
|
|
label: {
|
|
"count": len(values),
|
|
"signed_error_ms_mean": statistics.fmean(values),
|
|
"mae_ms": statistics.fmean(abs(value) for value in values),
|
|
}
|
|
for label, values in sorted(by_bin.items())
|
|
},
|
|
}
|
|
|
|
|
|
def _pass_rate_residuals(
|
|
frontier_results: list[dict[str, Any]], real_results: list[dict[str, Any]]
|
|
) -> dict[str, Any]:
|
|
def index(
|
|
results: list[dict[str, Any]], slo_name: str
|
|
) -> dict[tuple[str, float], float]:
|
|
return {
|
|
(config["config"]["name"], float(load["offered_request_rate"])): float(
|
|
load["scores"][slo_name]["slo_pass_rate"]
|
|
)
|
|
for config in results
|
|
for load in config["loads"]
|
|
}
|
|
|
|
variants = {}
|
|
for slo_name in SLO_VARIANTS:
|
|
frontier_index = index(frontier_results, slo_name)
|
|
real_index = index(real_results, slo_name)
|
|
keys = sorted(set(frontier_index) & set(real_index))
|
|
errors = [frontier_index[key] - real_index[key] for key in keys]
|
|
variants[slo_name] = {
|
|
"matched_config_load_points": len(errors),
|
|
"signed_error_mean": statistics.fmean(errors) if errors else None,
|
|
"mae": statistics.fmean(abs(value) for value in errors) if errors else None,
|
|
"max_absolute_error": max((abs(value) for value in errors), default=None),
|
|
}
|
|
return variants
|
|
|
|
|
|
def compare(args: argparse.Namespace) -> Path:
|
|
frontier = json.loads(args.frontier_freeze.read_text())
|
|
community = json.loads(args.community_freeze.read_text())
|
|
if frontier.get("schema") != FRONTIER_SCHEMA:
|
|
raise ValueError("unexpected Frontier freeze")
|
|
if community.get("schema") != COMMUNITY_SCHEMA or community.get("status") != "completed":
|
|
raise ValueError("unexpected/incomplete community freeze")
|
|
if frontier["protocol_manifest_sha256"] != community["protocol_manifest_sha256"]:
|
|
raise ValueError("comparison inputs used different protocols")
|
|
variants = {}
|
|
for slo_name in SLO_VARIANTS:
|
|
frontier_ranking = frontier["rankings"][slo_name]
|
|
real_ranking = community["rankings"][slo_name]
|
|
frontier_ranks = _rank_map(frontier_ranking)
|
|
real_ranks = _rank_map(real_ranking)
|
|
names = sorted(set(frontier_ranks) & set(real_ranks))
|
|
frontier_capacity = _capacity_map(frontier_ranking)
|
|
real_capacity = _capacity_map(real_ranking)
|
|
frontier_average_ranks = _average_ranks(frontier_capacity)
|
|
real_average_ranks = _average_ranks(real_capacity)
|
|
best_frontier = max(
|
|
(value for value in frontier_capacity.values() if value is not None),
|
|
default=None,
|
|
)
|
|
best_real = max(
|
|
(value for value in real_capacity.values() if value is not None),
|
|
default=None,
|
|
)
|
|
frontier_top_set = [
|
|
name for name, value in frontier_capacity.items() if value == best_frontier
|
|
]
|
|
real_top_set = [name for name, value in real_capacity.items() if value == best_real]
|
|
selected_real_capacities = [
|
|
real_capacity[name]
|
|
for name in frontier_top_set
|
|
if real_capacity[name] is not None
|
|
]
|
|
regret_values = [
|
|
(best_real - value) / best_real
|
|
for value in selected_real_capacities
|
|
if best_real not in (None, 0)
|
|
]
|
|
variants[slo_name] = {
|
|
"frontier_top1_set": frontier_top_set,
|
|
"real_top1_set": real_top_set,
|
|
"top1_set_intersection": sorted(set(frontier_top_set) & set(real_top_set)),
|
|
"top1_set_match": set(frontier_top_set) == set(real_top_set),
|
|
"top1_regret_fraction_best_tie_break": min(regret_values)
|
|
if regret_values
|
|
else None,
|
|
"top1_regret_fraction_worst_tie_break": max(regret_values)
|
|
if regret_values
|
|
else None,
|
|
"spearman_rank_correlation": _pearson(
|
|
[frontier_average_ranks[name] for name in names],
|
|
[real_average_ranks[name] for name in names],
|
|
),
|
|
"pairwise_ordering": _pairwise_accuracy(frontier_capacity, real_capacity),
|
|
"frontier_ranking": frontier_ranking,
|
|
"real_ranking": real_ranking,
|
|
}
|
|
result = {
|
|
"schema": COMPARISON_SCHEMA,
|
|
"created_unix_s": time.time(),
|
|
"frontier_freeze": {
|
|
"path": str(args.frontier_freeze.resolve()),
|
|
"sha256": sha256(args.frontier_freeze),
|
|
},
|
|
"community_freeze": {
|
|
"path": str(args.community_freeze.resolve()),
|
|
"sha256": sha256(args.community_freeze),
|
|
},
|
|
"primary_slo": PRIMARY_SLO,
|
|
"variants": variants,
|
|
"request_level_ttft_residuals": _request_residuals(
|
|
frontier["config_results"], community["config_results"]
|
|
),
|
|
"pass_rate_surface_residuals": _pass_rate_residuals(
|
|
frontier["config_results"], community["config_results"]
|
|
),
|
|
}
|
|
write_json(args.output, result)
|
|
print(args.output)
|
|
return args.output
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser()
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
prepare = subparsers.add_parser("prepare-protocol")
|
|
prepare.add_argument("--trace", type=Path, required=True)
|
|
prepare.add_argument("--expected-trace-sha256", default=frontier_grid.TRACE_SHA256)
|
|
prepare.add_argument("--output-root", type=Path, required=True)
|
|
prepare.add_argument("--cohort-size", type=int, default=COHORT_SIZE)
|
|
prepare.add_argument("--seed", type=int, default=COHORT_SEED)
|
|
prepare.add_argument("--rate", action="append", type=float, default=[])
|
|
|
|
audit = subparsers.add_parser("audit-protocol")
|
|
audit.add_argument("--manifest", type=Path, required=True)
|
|
|
|
frontier = subparsers.add_parser("run-frontier")
|
|
frontier.add_argument("--frontier-source", type=Path, required=True)
|
|
frontier.add_argument(
|
|
"--frontier-commit", default="d9cfeb6d8791fbf2f295dd9744c56a666171776e"
|
|
)
|
|
frontier.add_argument("--python", type=Path, required=True)
|
|
frontier.add_argument("--profile-root", type=Path, required=True)
|
|
frontier.add_argument("--protocol-manifest", type=Path, required=True)
|
|
frontier.add_argument("--output-root", type=Path, required=True)
|
|
frontier.add_argument("--tp4-capacity-artifact", type=Path, required=True)
|
|
frontier.add_argument("--tp8-capacity-artifact", type=Path, required=True)
|
|
frontier.add_argument("--cache-equivalence-artifact", type=Path, required=True)
|
|
frontier.add_argument("--max-parallel-configs", type=int, default=1)
|
|
frontier.add_argument("--config", action="append")
|
|
|
|
community = subparsers.add_parser("prepare-community")
|
|
community.add_argument("--repo", type=Path, required=True)
|
|
community.add_argument("--python", type=Path, required=True)
|
|
community.add_argument("--vllm", type=Path, required=True)
|
|
community.add_argument("--model", type=Path, required=True)
|
|
community.add_argument("--windows", type=Path, required=True)
|
|
community.add_argument("--protocol-manifest", type=Path, required=True)
|
|
community.add_argument("--frontier-freeze", type=Path, required=True)
|
|
community.add_argument("--output-root", type=Path, required=True)
|
|
community.add_argument("--port", type=int, default=18918)
|
|
community.add_argument("--config-order-seed", type=int, default=CONFIG_ORDER_SEED)
|
|
community.add_argument("--rate-order-seed", type=int, default=RATE_ORDER_SEED)
|
|
|
|
run_real = subparsers.add_parser("run-community")
|
|
run_real.add_argument("--manifest", type=Path, required=True)
|
|
|
|
compare_parser = subparsers.add_parser("compare")
|
|
compare_parser.add_argument("--frontier-freeze", type=Path, required=True)
|
|
compare_parser.add_argument("--community-freeze", type=Path, required=True)
|
|
compare_parser.add_argument("--output", type=Path, required=True)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
if args.command == "prepare-protocol":
|
|
if not args.rate:
|
|
args.rate = list(OFFERED_RATES)
|
|
if sorted(set(args.rate)) != list(args.rate):
|
|
raise ValueError("--rate values must be unique and ascending")
|
|
prepare_protocol(args)
|
|
elif args.command == "audit-protocol":
|
|
print(json.dumps(audit_protocol(args.manifest), indent=2))
|
|
elif args.command == "run-frontier":
|
|
run_frontier(args)
|
|
elif args.command == "prepare-community":
|
|
prepare_community(args)
|
|
elif args.command == "run-community":
|
|
run_community(args)
|
|
elif args.command == "compare":
|
|
compare(args)
|
|
else:
|
|
raise AssertionError(args.command)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|