From 47c7d28023ebfd49a6959f8149d319da5fc22656 Mon Sep 17 00:00:00 2001 From: Advay Monga Date: Sun, 11 Oct 2026 14:51:00 -0400 Subject: [PATCH 1/4] Ceilings: every measured number with its physical bound and the reference - validity/ceiling.py: model spec from the HF config (dense and MoE), decode/prefill floors, largest batch under a TPOT limit, per-objective regime ceilings, pct_of and Amdahl ceiling_gain. Replaces the Gemma-only roofline.py. - validity/peaks.py: HBM bandwidth and bf16 GEMM measured once per GPU+driver (python -m validity peaks, run by vm-setup.sh, fetched by lab.vm setup); spec sheet as the labelled fallback. - lab/calibrate.py: bench/baseline results gain a `calibrated` block (value, % of ceiling, % of reference, shape, fingerprint); profile records carry each kernel's share of the window; kernel records the ncu launches' % of bandwidth peak. A ceiling or reference appears only under a matching config fingerprint, else why it is missing. - lab.validate measures the reference under the regimes once into knowledge/reference//.json. - Client rows record prompt_tokens; the worker reports the GPU a measurement ran on; the brief shows the calibrated baseline. Co-Authored-By: Claude Fable 5.1 --- CLAUDE.md | 14 +- lab/LEDGER.md | 4 +- lab/README.md | 4 +- lab/artifacts.py | 5 + lab/calibrate.py | 194 +++++++++++++++++++++ lab/evaltools.py | 20 ++- lab/proftools.py | 4 +- lab/session.py | 2 +- lab/tools.py | 6 +- lab/validate.py | 21 ++- lab/vm-setup.sh | 4 + lab/vm.py | 7 +- lab/worker.py | 9 +- regimes/client.py | 3 + tests/fake_engine.py | 6 +- tests/test_lab_calibrate.py | 86 +++++++++ tests/test_lab_evaltools.py | 4 + tests/test_lab_validate.py | 5 + tests/test_validity_ceiling.py | 91 ++++++++++ validity/__main__.py | 10 +- validity/ceiling.py | 181 +++++++++++++++++++ validity/peaks.py | 104 +++++++++++ validity/roofline.py | 310 --------------------------------- 23 files changed, 759 insertions(+), 335 deletions(-) create mode 100644 lab/calibrate.py create mode 100644 tests/test_lab_calibrate.py create mode 100644 tests/test_validity_ceiling.py create mode 100644 validity/ceiling.py create mode 100644 validity/peaks.py delete mode 100644 validity/roofline.py diff --git a/CLAUDE.md b/CLAUDE.md index a290788..9d6be99 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -33,6 +33,11 @@ pytest/ruff config). Each folder with commands has its own `python -m`. `agent.py` (the one place a model is called; the agent's jailed shell gets the engine's python on PATH; `LAB_AGENT_PROVIDER` claude = Agent SDK CLI, openai = `agent_openai.py`: any OpenAI-compatible API with its own key, https public hosts only, `LAB_MODEL_PRICE` required, loop outside the jail, Bash jailed with no network), + `calibrate.py` (every measured number with its context, decided 2026-10-11: `% of ceiling` from `validity/ceiling` + on measured peaks, `% of reference` from `knowledge/reference//.json` written by `lab.validate`, + and for kernels their share of the window; a ceiling or reference is shown only under a matching config + fingerprint (model, dtypes, max_model_len, tier, split, corpus version, GPU), else the field says why it is + missing; bench, baseline, profile and kernel records carry a `calibrated` block, the brief shows the baseline's), `tools.py` (metered tools: test, profile, ledger, budget, restore, note; every record's `cost` has wall seconds, and GPU tools charge them at `LAB_GPU_USD_PER_HOUR` / the target's `[cost] gpu_usd_per_hour` to the run's budget) + `proftools.py` (trace = nsys, kernel = ncu, hostprof = py-spy, each wrapping lab.profile's workload in the jail @@ -102,7 +107,11 @@ pytest/ruff config). Each folder with commands has its own `python -m`. (`verdict` re-judges saved results under the policy, no server). - `canaries/cheat_proxy.py` — cheat proxy in front of any OpenAI-compatible server (`truncate`, `early_eos`, `fake_first`, `drop`, `cache` (streamed and not), `inflate_usage`); `python -m canaries --upstream URL --cheat NAME`. -- `validity/` — `roofline.py` (analytical decode ceiling, no GPU; to be generalized into physical limits), `repeat_audit.py`: repeated-request audit (MLPerf TEST04-style), judges decode time per token, not TTFT, so prefix caching passes and replayed answers fail; `python -m validity repeat --url URL`. `tier_agreement.py`: `tier_agreement(changes)`: Spearman, Kendall tau-b, bootstrap CI, out-of-band disagreements; short tier usable if n ≥ 8, rho ≥ 0.8, CI low ≥ 0.5. `holdout.py`: `HoldoutGuard` (Thresholdout query budget, state persisted atomically under a file lock, deterministic from seed) and `SealedSplit` (hash manifest, one logged unseal per corpus version). Keep sigma small relative to the threshold. +- `validity/` — `ceiling.py` (physical bounds from the model's HF config and the GPU's peaks: active/total params, + bytes streamed per step with MoE experts-touched, decode and prefill floors, the largest batch under a TPOT limit, + per-objective regime ceilings; `pct_of`, `ceiling_gain` = Amdahl) + `peaks.py` (`python -m validity peaks`: HBM + bandwidth and bf16 GEMM measured once per GPU+driver into `knowledge/peaks/.json`, run by `vm-setup.sh`; spec + sheet as the labelled fallback), `repeat_audit.py`: repeated-request audit (MLPerf TEST04-style), judges decode time per token, not TTFT, so prefix caching passes and replayed answers fail; `python -m validity repeat --url URL`. `tier_agreement.py`: `tier_agreement(changes)`: Spearman, Kendall tau-b, bootstrap CI, out-of-band disagreements; short tier usable if n ≥ 8, rho ≥ 0.8, CI low ≥ 0.5. `holdout.py`: `HoldoutGuard` (Thresholdout query budget, state persisted atomically under a file lock, deterministic from seed) and `SealedSplit` (hash manifest, one logged unseal per corpus version). Keep sigma small relative to the threshold. - `kernels/` — kernel equivalence: `check_kernel` (seen + held-out shapes, per-dtype tolerances, edge cases incl. non-contiguous and NaN/Inf, inputs unchanged, no aliasing, memoization, determinism) and `time_kernel` / `check_memoization` (sync, rotated buffers, L2 flush on CUDA). Validated on CPU, MPS and CUDA (H200, 2026-10-04): all tests pass; the timer resolves ~20 µs kernels at ~9% IQR, 0.2 ms at 1.5%. `make_inputs` gets a seeded CPU generator (generate on CPU, then `.to(device)`). - `regimes/` — load generation: drives a live OpenAI-compatible server the eight ways it gets used, client-side. `client.py` (one streamed chat request: TTFT from *scheduled* send to first real token, TPOT = (E2E − TTFT)/(n − 1), @@ -232,6 +241,7 @@ a first `python -m lab.session --task ...` dry run with a tiny budget, which wai `engine/gpu-validated`) were judged by the old gate: re-judge with `python -m correctness verdict` on their saved results before deciding anything. lab `session.py --run` re-resolves `base` from HEAD rather than the run's original base (pre-existing; resuming after the engine moved would misaudit); -`validity/tier_agreement.py` needs real short/full runs; `validity/roofline.py` wants generalizing to the H200 + MoE +`validity/tier_agreement.py` needs real short/full runs; the ceilings need one GPU pass (measured peaks, floors +below every measurement, `lab.validate`'s reference regimes) target. The agent's tool set (e.g. a knowledge tool) is the user's design: ask first. README stays a few one-line feature bullets, not a dev doc. diff --git a/lab/LEDGER.md b/lab/LEDGER.md index f578c37..be96982 100644 --- a/lab/LEDGER.md +++ b/lab/LEDGER.md @@ -13,10 +13,10 @@ on write. Every record: `id`, `at`, `schema` (set by the writer), `kind`, `run`, | `test` | `test` | `lint`, `tests` (bools), returncode, output tails, `changed`, `scratch_left_out` | | `profile` | `profile` | `bundle` (ledger blob of the profiler output), `workspace_copy`, seconds | | `profile` | `trace`, `kernel`, `hostprof` | `bundle` (blob of the instrument's output dir), `workspace_copy`, `instrument` (resolved binary), `argv` (every command run), returncode, seconds, output tail | -| `bench` | `bench` | `config: {split: seen, tier, seed}`; `verdict: ok`, `metrics: {regime: {objective, value, better, valid, invalid_reasons}}`, `regimes` (full per-regime results), `ready_s`, `artifacts` | +| `bench` | `bench` | `config: {split: seen, tier, seed}`; `verdict: ok`, `metrics: {regime: {objective, value, better, valid, invalid_reasons}}`, `calibrated: {regime: {value, ceiling, pct_of_ceiling, bound, ceiling_basis, peaks_source, shape, reference, pct_of_reference, fingerprint}}` (a missing ceiling or reference comes with `ceiling_missing` / `reference_missing` saying why), `regimes` (full per-regime results), `ready_s`, `gpu`, `artifacts` | | `equiv` | `equiv` | `config: {split: seen, tier}`; `verdict: pass\|fail\|inconclusive`, `passed`, `reasons`, `gates`, `metrics` (accuracy with CI, length, consistency, divergence, per-task flips, unanswered), `thresholds`, `artifacts` | | `submit` | `submit` | **held-out shape**: no `result`; top-level `config: {split: heldout, tier: full}` and `metrics: {regime: {base, new, delta_pct, band_pct, verdict}}`, scalars only. The ledger refuses anything else that mentions the held-out split. | -| `baseline` | the loop, once per run before the first session | `config: {split: seen, tier: short, commit}`; `verdict: ok` with `metrics` shaped like `bench` (the base commit under bench's default regimes: the task's objective and constrained ones, from the cache submit compares against), or `verdict: error`, `reason` | +| `baseline` | the loop, once per run before the first session | `config: {split: seen, tier: short, commit}`; `verdict: ok` with `metrics` and `calibrated` shaped like `bench`'s (the base commit under bench's default regimes: the task's objective and constrained ones, from the cache submit compares against), or `verdict: error`, `reason` | | `finding` | `lab.ledger seed` | a `knowledge/` entry, whole, as `claim`; `source`, `content_sha` | | `note` | `note`, `restore`, refusals | free text, or `{verdict: refused, reason}` | diff --git a/lab/README.md b/lab/README.md index d7936dd..88a7147 100644 --- a/lab/README.md +++ b/lab/README.md @@ -105,7 +105,7 @@ with the referee's rights, snapshot the workspace and write the ledger on every | tool | does | |---|---| | `test` | lint and the fast suite on a pristine two-commit copy of the workspace, jailed | -| `profile` | `lab.profile` on the pristine copy, jailed, on `args.corpus_class` (default `steady_interactive`, seen split, `args.seed`) or `args.synthetic`; the bundle goes into the ledger as a blob | +| `profile` | `lab.profile` on the pristine copy, jailed, on `args.corpus_class` (default `steady_interactive`, seen split, `args.seed`) or `args.synthetic`; the bundle goes into the ledger as a blob, with each kernel's share of the window (the Amdahl denominator) in the record | | `trace` | `nsys profile` (`args.nsys_args`, default window-only CUDA/NVTX/OS-runtime with CUDA-graph nodes) around `lab.profile --profiler cuda-range` (window only), then `nsys stats` (kernel, CUDA API, memcpy, NVTX tables as CSV), both jailed; `.nsys-rep` + CSVs as a blob | | `kernel` | `ncu [--kernel-name regex:] --launch-skip N --launch-count N --set [ncu_args]` around the same workload, then `ncu --import --csv --page raw`; `.ncu-rep` + `metrics.csv` as a blob | | `hostprof` | `lab.profile --profiler pyspy`: py-spy record (speedscope, plus `args.pyspy_args`) for `seconds` under the repeated workload plus one `py-spy dump`; the bundle as a blob | @@ -113,7 +113,7 @@ with the referee's rights, snapshot the workspace and write the ledger on every | `budget` | dollars left | | `restore` | workspace back to a snapshot id (`base` resets) | | `note` | a note for the human; recorded, changes nothing | -| `bench` | serve the pristine copy and run the load regimes (`regimes/`) on the seen split, short tier; one headline per regime, raw (`args.regimes`, `args.tier`, `args.seed`) | +| `bench` | serve the pristine copy and run the load regimes (`regimes/`) on the seen split, short tier; one headline per regime, raw (`args.regimes`, `args.tier`, `args.seed`), each with its context (`lab/calibrate.py`: % of the physical ceiling, % of the reference engine measured the same way; only under a matching config fingerprint) | | `equiv` | serve it and run the correctness gate (`correctness/`) against the target's reference outputs; pass, fail or inconclusive with every metric; `tier=full` uses `-full` | | `submit` | the only thing that can produce a win. Needs a passing full-tier equiv and a seen bench on this snapshot. Measures the held-out split at the full tier, measures the base commit the same way (once per run, cached outside the jail), and records one aggregate per regime (base, new, delta %, noise band %, verdict improved/regressed/within_band/unknown_band) through the Thresholdout guard (`validity/holdout.py`, state in `lab/ledger/holdout_state.json`, seed `LAB_HOLDOUT_SEED`): the held-out numbers themselves never reach the ledger. Noise bands come from `knowledge/noise/.json` when measured | diff --git a/lab/artifacts.py b/lab/artifacts.py index 6ef4b79..280d701 100644 --- a/lab/artifacts.py +++ b/lab/artifacts.py @@ -19,6 +19,11 @@ def write_rows(work: Path, rows: list[dict]) -> None: (Path(work) / ROWS).write_text("".join(json.dumps(r) + "\n" for r in rows)) +def read_rows(work: Path) -> list[dict]: + p = Path(work) / ROWS + return [json.loads(x) for x in p.read_text().splitlines() if x.strip()] if p.exists() else [] + + def store(work: Path, root: Path) -> dict: """`work`'s passive data as ledger blobs, {device, telemetry, client_rows, serve_log} (None when absent); `work` is removed.""" diff --git a/lab/calibrate.py b/lab/calibrate.py new file mode 100644 index 0000000..22e3542 --- /dev/null +++ b/lab/calibrate.py @@ -0,0 +1,194 @@ +"""Every measured number with its context: % of the physical ceiling (validity/ceiling.py on measured peaks), % of +the reference engine measured the same way, and for parts of a step their share of the whole and the Amdahl gain. +A ceiling or a reference is only shown when its config fingerprint matches the measurement's; otherwise the field +says why it is missing. Facts for the agent, never a judgement.""" +from __future__ import annotations + +import csv +import json +import statistics +from pathlib import Path + +from lab import target +from validity import ceiling, peaks +from validity.ceiling import ModelSpec, Shape + +REFERENCE_DIR = target.ENV_ROOT / "knowledge" / "reference" +FINGERPRINT_KEYS = ("model", "w_dtype", "kv_dtype", "max_model_len", "tier", "split", "corpus_version", "gpu") + + +# -- model config ----------------------------------------------------------------------- +def model_config(model: str) -> dict | None: + """The model's HF config.json from the local hub cache, else the Hub; None offline without a cache.""" + from lab.safety.grader import HF_HUB + cached = sorted((HF_HUB / f"models--{model.replace('/', '--')}" / "snapshots").glob("*/config.json")) + if cached: + return json.loads(cached[-1].read_text()) + try: + from huggingface_hub import hf_hub_download + return json.loads(Path(hf_hub_download(model, "config.json")).read_text()) + except Exception: + return None + + +def model_spec(t: target.Target) -> ModelSpec | None: + cfg = model_config(t.model) + return ModelSpec.from_hf_config(cfg, t.model) if cfg else None + + +def dtypes(t: target.Target) -> tuple[str, str]: + m = t.raw.get("model") or {} + return m.get("dtype", "bf16"), m.get("kv_dtype", m.get("dtype", "bf16")) + + +# -- fingerprint -------------------------------------------------------------------------- +def fingerprint(t: target.Target, gpu: dict | None, tier: str, split: str) -> dict: + from regimes.workload import corpus_version + w, kv = dtypes(t) + return {"model": t.model, "w_dtype": w, "kv_dtype": kv, "max_model_len": t.max_model_len, "tier": tier, + "split": split, "corpus_version": corpus_version(t.corpus_dir) if t.corpus_dir.is_dir() else None, + "gpu": (gpu or {}).get("name"), "driver": (gpu or {}).get("driver_version"), + "sm_clock_mhz": (gpu or {}).get("clocks.sm")} + + +def same_config(a: dict, b: dict) -> list[str]: + """The fingerprint keys that differ (empty = comparable). Driver and clocks are reported, not required.""" + return [k for k in FINGERPRINT_KEYS if a.get(k) != b.get(k)] + + +# -- shapes from client rows -------------------------------------------------------------- +def shape_from_rows(rows: list[dict]) -> Shape | None: + ok = [r for r in rows if r.get("status") == "ok" and r.get("e2e_s") is not None] + if not ok: + return None + prompts = [r["prompt_tokens"] for r in ok if r.get("prompt_tokens")] + outs = [r.get("n_tokens") or 0 for r in ok] + start = min(r["scheduled_s"] for r in ok) + span = max(r["scheduled_s"] + r["e2e_s"] for r in ok) - start + conc = sum(r["e2e_s"] for r in ok) / span if span > 0 else 1.0 + if not prompts: + return None + p99 = sorted(prompts)[min(len(prompts) - 1, int(0.99 * len(prompts)))] + return Shape(statistics.fmean(prompts), float(p99), statistics.fmean(outs), max(1.0, conc)) + + +# -- reference numbers ----------------------------------------------------------------------- +def reference_path(t: target.Target, regime: str) -> Path: + return REFERENCE_DIR / t.name / f"{regime}.json" + + +def write_reference(t: target.Target, regime: str, result: dict, fp: dict, rows: list[dict] | None = None) -> Path: + """One regime of the reference server, measured by the same clients, stored with its fingerprint.""" + p = reference_path(t, regime) + p.parent.mkdir(parents=True, exist_ok=True) + d = {"target": t.name, "server": "reference", "regime": regime, "objective": result["objective"], + "value": result["value"] if result.get("valid") else None, "better": result["better"], + "valid": result.get("valid"), "summary": result.get("summary"), "fingerprint": fp} + sh = shape_from_rows(rows) if rows else None + if sh: + d["shape"] = sh.__dict__ + p.write_text(json.dumps(d, indent=1, default=str) + "\n") + return p + + +def reference_for(t: target.Target, regime: str, fp: dict) -> tuple[float | None, str | None]: + """(value, why_missing): the reference's objective for this regime, only under the same fingerprint.""" + p = reference_path(t, regime) + if not p.exists(): + return None, "no reference measurement for this regime (lab.validate writes one)" + d = json.loads(p.read_text()) + diff = same_config(d.get("fingerprint") or {}, fp) + if diff: + return None, f"reference measured under a different config: {', '.join(diff)}" + return d.get("value"), None if d.get("value") is not None else "reference measurement was invalid" + + +# -- regime results ---------------------------------------------------------------------------- +def calibrate_results(results: list[dict], rows: list[dict], t: target.Target, gpu: dict | None, tier: str, + split: str) -> dict: + """{regime: {value, ceiling, pct_of_ceiling, reference, pct_of_reference, ...}} for a bench's results.""" + fp = fingerprint(t, gpu, tier, split) + hw = peaks.load(fp["gpu"], fp.get("driver")) if fp["gpu"] else None + spec = model_spec(t) + w, kv = dtypes(t) + out = {} + for r in results: + name, obj = r["regime"], r["objective"] + value = r["value"] if r.get("valid") else None + row = {"objective": obj, "value": value, "better": r["better"], "fingerprint": fp} + shape = shape_from_rows([x for x in rows if x.get("regime") == name]) + if hw is None: + row["ceiling"], row["ceiling_missing"] = None, "no peaks for this GPU (python -m validity peaks)" + elif spec is None: + row["ceiling"], row["ceiling_missing"] = None, "model config unavailable" + elif shape is None: + row["ceiling"], row["ceiling_missing"] = None, "no completed requests with prompt token counts" + else: + limits = (t.interactive if name == "single_stream" else t.conversational).__dict__ + c = ceiling.regime_ceiling(obj, spec, hw, shape, limits, w, kv) + if c is None: + row["ceiling"], row["ceiling_missing"] = None, f"no physical bound for {obj}" + else: + row.update(ceiling=c["ceiling"], pct_of_ceiling=ceiling.pct_of(value, c["ceiling"], r["better"]), + bound=c["bound"], ceiling_basis=c["basis"], peaks_source=hw.source, + shape=shape.__dict__) + ref, why = reference_for(t, name, fp) + row["reference"] = ref + if why: + row["reference_missing"] = why + else: + row["pct_of_reference"] = ceiling.pct_of(value, ref, r["better"]) + out[name] = row + return out + + +# -- kernels and steps from a profile bundle ----------------------------------------------------- +def calibrate_kernels(bundle: Path, gpu: dict | None, top: int = 30) -> dict | None: + """From a torch-profiler bundle: each kernel's share of the profiled window (the Amdahl denominator).""" + kj, meta = bundle / "kernels.json", bundle / "meta.json" + if not kj.exists() or not meta.exists(): + return None + window_s = (json.loads(meta.read_text()).get("window") or {}).get("wall_s") + if not window_s: + return None + rows = [{"name": k["name"], "count": k["count"], "total_us": k["total_us"], + "share_of_window": k["total_us"] / (window_s * 1e6)} for k in json.loads(kj.read_text())["kernels"][:top]] + return {"window_s": window_s, "kernel_share_total": sum(r["share_of_window"] for r in rows), "kernels": rows, + "gpu": (gpu or {}).get("name")} + + +def calibrate_ncu(path: Path, gpu: dict | None) -> dict | None: + """From `ncu --csv --page raw`: each profiled launch's achieved DRAM rate as % of the measured bandwidth peak.""" + if not path.exists(): + return None + rows = list(csv.DictReader(path.open())) + if len(rows) < 2: + return None + units, data = rows[0], rows[1:] + hw = peaks.load(gpu["name"], gpu.get("driver_version")) if gpu and gpu.get("name") else None + col = "dram__bytes.sum.per_second" + scale = {"byte/s": 1, "Kbyte/s": 1e3, "Mbyte/s": 1e6, "Gbyte/s": 1e9, "Tbyte/s": 1e12}.get(units.get(col, ""), 1) + out = [] + for r in data: + bw = float(r[col]) * scale if r.get(col) else None + dur = r.get("gpu__time_duration.sum") + out.append({"name": r.get("Kernel Name"), "duration": float(dur) if dur else None, + "duration_unit": units.get("gpu__time_duration.sum"), "dram_bytes_s": bw, + "pct_of_bandwidth_peak": bw / hw.bw_bytes_s * 100 if bw and hw else None}) + return {"launches": out, "peaks_source": hw.source if hw else None, + "ceiling_missing": None if hw else "no peaks for this GPU (python -m validity peaks)"} + + +def facts(t: target.Target, gpu: dict | None) -> dict: + """The ceiling machinery's inputs, for the brief: peaks, the model's streamed bytes, and the single-stream floors.""" + spec, (w, kv) = model_spec(t), dtypes(t) + hw = peaks.load(gpu["name"], gpu.get("driver_version")) if gpu and gpu.get("name") else None + out: dict = {"peaks": hw.__dict__ if hw else None, "model": None} + if spec: + out["model"] = {"active_params": spec.active_params, "total_params": spec.total_params, + "weight_bytes_per_decode_step_batch1": spec.weight_bytes(1, w), + "kv_bytes_per_token": spec.kv_bytes_per_token(kv), "w_dtype": w, "kv_dtype": kv} + if spec and hw: + out["floors_s"] = {"decode_step_batch1_ctx1k": ceiling.decode_step_floor(spec, hw, 1, 1024, w, kv)["floor_s"], + "prefill_1k_prompt": ceiling.prefill_floor(spec, hw, 1024, w, kv)["floor_s"]} + return out diff --git a/lab/evaltools.py b/lab/evaltools.py index eb19f87..a060575 100644 --- a/lab/evaltools.py +++ b/lab/evaltools.py @@ -24,7 +24,7 @@ import time from pathlib import Path -from lab import artifacts, ledger, serve, target, worker +from lab import artifacts, calibrate, ledger, serve, target, worker from validity.holdout import HoldoutGuard, Refused HOLDOUT_BUDGET, HOLDOUT_THRESHOLD_PCT, HOLDOUT_SIGMA_PCT = 100, 2.0, 0.5 @@ -165,11 +165,13 @@ def bench(self, args: dict) -> str: "artifacts": artifacts.store(work, self.s.ledger_root)} self._record("bench", "bench", args, result, snap, config={"split": "seen", "tier": tier, "seed": seed}) return _said("bench", result) - result = {"verdict": "ok", "metrics": headline(r["results"]), "regimes": r["results"], "tier": tier, - "seconds": time.monotonic() - t0, "ready_s": r["ready_s"], + cal = calibrate.calibrate_results(r["results"], artifacts.read_rows(work), target.load(), r.get("gpu"), tier, "seen") + result = {"verdict": "ok", "metrics": headline(r["results"]), "calibrated": cal, "regimes": r["results"], + "tier": tier, "seconds": time.monotonic() - t0, "ready_s": r["ready_s"], "gpu": r.get("gpu"), "artifacts": artifacts.store(work, self.s.ledger_root)} self._record("bench", "bench", args, result, snap, config={"split": "seen", "tier": tier, "seed": seed}) - return json.dumps({"snapshot": snap.id, "split": "seen", "tier": tier, "metrics": result["metrics"]}, indent=1) + return json.dumps({"snapshot": snap.id, "split": "seen", "tier": tier, "metrics": result["metrics"], + "calibrated": cal}, indent=1) # -- equiv -------------------------------------------------------------------------- def equiv(self, args: dict) -> str: @@ -320,6 +322,11 @@ def _base_measure(self, t: target.Target, names: list[str], split: str, tier: st shutil.rmtree(work, ignore_errors=True) have.update(headline(r["results"])) cache.write_text(json.dumps(have, indent=1)) + if split == "seen": # the base's context travels with the baseline record + cal = cache.with_suffix(".calibrated.json") + prior = json.loads(cal.read_text()) if cal.exists() else {} + prior.update(calibrate.calibrate_results(r["results"], [], t, r.get("gpu"), tier, split)) + cal.write_text(json.dumps(prior, indent=1)) return have def baseline(self, commit: str) -> dict: @@ -331,7 +338,10 @@ def baseline(self, commit: str) -> dict: self._t0 = time.monotonic() try: seen = self.base_seen(target.load(), names) - result = {"verdict": "ok", "metrics": {n: seen[n] for n in names}} + cal = self.s.run_dir / f"base-seen-{BENCH_TIER}.calibrated.json" + calibrated = json.loads(cal.read_text()) if cal.exists() else {} + result = {"verdict": "ok", "metrics": {n: seen[n] for n in names}, + "calibrated": {n: calibrated[n] for n in names if n in calibrated}} except Exception as e: # no GPU or a base that will not serve is itself the fact result = _failed(e) return ledger.append({"kind": "baseline", "run": self.s.run_id, "session": self.s.session_id, diff --git a/lab/proftools.py b/lab/proftools.py index c8c3f14..5d3412b 100644 --- a/lab/proftools.py +++ b/lab/proftools.py @@ -134,7 +134,9 @@ def _instrumented(self, tool: str, args: dict) -> str: blob, visible = self._keep(files) if files.is_dir() and any(p.is_file() for p in files.rglob("*")) else ("", "") finally: shutil.rmtree(out, ignore_errors=True) - result = {"returncode": r["returncode"], "bundle": blob, "workspace_copy": visible, "instrument": r["instrument"], + from lab import calibrate + cal = calibrate.calibrate_ncu(self.s.ledger_root / blob / "metrics.csv", r.get("gpu")) if blob and tool == "kernel" else None + result = {"calibrated": cal, "returncode": r["returncode"], "bundle": blob, "workspace_copy": visible, "instrument": r["instrument"], "argv": r["argv"], "seconds": time.monotonic() - t0, "output": r["output"]} self._record("profile", tool, args, result, snap) if r["returncode"] != 0 or not r["head"]: diff --git a/lab/session.py b/lab/session.py index d63eb2f..0599640 100644 --- a/lab/session.py +++ b/lab/session.py @@ -89,7 +89,7 @@ def _baseline(s: Session) -> str: head = f"Baseline (base commit {c['commit']}, {c['split']} split, {c['tier']} tier, record {b['id']})" if r["verdict"] != "ok": return f"{head}: not measured; {r['reason']}" - return f"{head}:\n{json.dumps(r['metrics'])}" + return f"{head}:\n{json.dumps(r['metrics'])}\nCalibrated (ceiling = physical bound from measured peaks; reference = the reference engine measured the same way):\n{json.dumps(r.get('calibrated') or {})}" def brief(cfg: RunConfig, s: Session, snapshot_id: str, n: int) -> str: diff --git a/lab/tools.py b/lab/tools.py index b3a91ad..5dc5e6c 100644 --- a/lab/tools.py +++ b/lab/tools.py @@ -9,7 +9,7 @@ from pathlib import Path from typing import Any -from lab import corpus, engine, ledger, target +from lab import calibrate, corpus, engine, ledger, target from lab.agent import ToolSpec from lab.budget import Budget, BudgetExceeded, gpu_rate from lab.evaltools import EvalTools @@ -123,8 +123,10 @@ def profile(self, args: dict) -> str: blob, visible = self._keep(out / "files") if (out / "files").is_dir() else ("", "") finally: shutil.rmtree(out, ignore_errors=True) + bundles = sorted(p for p in (self.s.ledger_root / blob).glob("*") if p.is_dir()) if blob else [] result = {"returncode": r["returncode"], "bundle": blob, "workspace_copy": visible, - "seconds": time.monotonic() - t0, "output": r["output"]} + "seconds": time.monotonic() - t0, "output": r["output"], + "calibrated": calibrate.calibrate_kernels(bundles[-1], r.get("gpu")) if bundles else None} self._record("profile", "profile", args, result, snap) if r["returncode"] != 0 or not blob: return f"profile failed ({r['returncode']}):\n{result['output']}" diff --git a/lab/validate.py b/lab/validate.py index 1093318..01a2c3c 100644 --- a/lab/validate.py +++ b/lab/validate.py @@ -25,7 +25,7 @@ import time from pathlib import Path -from lab import serve, target +from lab import calibrate, gpu, serve, target def band_pct(values: list[float]) -> float | None: @@ -47,7 +47,8 @@ def write_noise(regime: str, values: list[float], t: target.Target, extra: dict def validate(t: target.Target, tier: str = "dev", noise_runs: int = 3, regimes: list[str] | None = None, - out: Path | None = None, encoder=None, log_dir: Path | None = None) -> dict: + out: Path | None = None, encoder=None, log_dir: Path | None = None, + reference_regimes: bool = True) -> dict: from correctness import client, run as crun, tasks from correctness.gate import Thresholds from regimes import suite @@ -94,6 +95,18 @@ def validate(t: target.Target, tier: str = "dev", noise_runs: int = 3, regimes: for n in names: report["noise"][n] = write_noise(n, values[n], t, {"tier": "short", "split": "seen"}) report["bands_measured"] = all(d["band_pct"] is not None for d in report["noise"].values()) if names and noise_runs else None + # 4. the reference engine under the same regimes, once: what every bench is shown next to + if reference_regimes and names: + with serve.Served(target.ENV_ROOT, t.reference, log=log_dir / "serve-reference-regimes.log", jailed=False) as srv: + rows: list = [] + ctx = suite.Ctx.from_target(t, srv.url, split="seen", tier="short", seed=0, rows=rows) + fp = calibrate.fingerprint(t, gpu.query(), "short", "seen") + for n in names: + k = len(rows) + r = suite.REGIMES[n](ctx) + calibrate.write_reference(t, n, r, fp, rows[k:]) + report.setdefault("reference_regimes", {})[n] = r["value"] if r["valid"] else None + srv.exclusive() (out / "report.json").write_text(json.dumps(report, indent=1, default=str)) return report @@ -103,11 +116,13 @@ def main(argv=None) -> int: target.add_argument(ap) ap.add_argument("--tier", choices=("dev", "full"), default="dev") ap.add_argument("--noise-runs", type=int, default=3) + ap.add_argument("--no-reference-regimes", action="store_true", help="skip measuring the reference under the regimes") ap.add_argument("--regimes", help="comma-separated (default: all)") ap.add_argument("--out") a = ap.parse_args(argv) t = target.load(a.target) - rep = validate(t, a.tier, a.noise_runs, a.regimes.split(",") if a.regimes else None, Path(a.out) if a.out else None) + rep = validate(t, a.tier, a.noise_runs, a.regimes.split(",") if a.regimes else None, Path(a.out) if a.out else None, + reference_regimes=not a.no_reference_regimes) for name, c in rep["candidates"].items(): print(f"{c['kind']:4s} {name}: {c['verdict'].upper()}" + (f" {c['reasons']}" if c["reasons"] else "")) for n, d in rep["noise"].items(): diff --git a/lab/vm-setup.sh b/lab/vm-setup.sh index 9484eb0..06f1b2b 100755 --- a/lab/vm-setup.sh +++ b/lab/vm-setup.sh @@ -83,6 +83,10 @@ else echo "clock lock: refused" fi +echo "== peaks: HBM bandwidth and bf16 GEMM, measured (validity/peaks.py; the ceilings' denominators)" +ENV_DIR="${LAB_VM_ENV_DIR:-$HOME/BlameGraph}"; ENGINE_DIR="$PWD" +(cd "$ENV_DIR" && "$ENGINE_DIR/.venv/bin/python" -m validity peaks --seconds 3) || echo "peaks: not measured (finding, not failure)" + echo "== model cache" mkdir -p "$HOME/.cache/huggingface" du -sh "$HOME/.cache/huggingface" 2>/dev/null || true diff --git a/lab/vm.py b/lab/vm.py index d9170f4..622cf9f 100644 --- a/lab/vm.py +++ b/lab/vm.py @@ -296,7 +296,12 @@ def main(argv: list[str] | None = None) -> int: print(p.types(), end="") return 0 if args.cmd == "setup": - return run(vm, f"bash {vm.env_dir}/lab/vm-setup.sh {vm.remote_dir}", keep=True) + code = run(vm, f"bash {vm.env_dir}/lab/vm-setup.sh {vm.remote_dir}", keep=True) + try: # the measured peaks come home: every ceiling is computed from them + fetch(vm, p.get(vm.name), "knowledge/peaks", engine.ENV_ROOT / "knowledge" / "peaks", base=vm.env_dir) + except ProviderError as e: + print(f"[lab.vm] peaks not fetched: {e}", file=sys.stderr) + return code words = args.command[1:] if args.command[:1] == ["--"] else args.command if not words: ap.error("run needs a command") diff --git a/lab/worker.py b/lab/worker.py index 1d895ce..355a7cb 100644 --- a/lab/worker.py +++ b/lab/worker.py @@ -70,7 +70,8 @@ def job_measure(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: artifacts=out if args["passive"] else None) as srv: results = run_regimes(srv.url, t, args["names"], args["split"], args["tier"], args["seed"], rows) srv.exclusive() - return {"results": results, "ready_s": srv.ready_s} + from lab import gpu + return {"results": results, "ready_s": srv.ready_s, "gpu": gpu.query()} finally: if rows is not None: artifacts.write_rows(out, rows) @@ -110,7 +111,8 @@ def job_profile(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: if bundles: T._drop_links(bundles[-1]) shutil.copytree(bundles[-1], files) - return {"returncode": proc.returncode, "output": (proc.stdout + proc.stderr)[-3000:]} + from lab import gpu + return {"returncode": proc.returncode, "output": (proc.stdout + proc.stderr)[-3000:], "gpu": gpu.query()} ins = proftools.instrument(kind, a) exe = shutil.which(ins["binary"]) if not exe: @@ -137,8 +139,9 @@ def job_profile(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: found = sorted(work.glob(ins["summary"])) if any(p.is_file() for p in work.rglob("*")): shutil.copytree(work, files) + from lab import gpu return {"returncode": proc.returncode, "instrument": exe, "argv": commands, "output": log[-3000:], - "head": proftools._head(found[-1] if found else None)} + "head": proftools._head(found[-1] if found else None), "gpu": gpu.query()} JOBS = {"test": job_test, "measure": job_measure, "equiv": job_equiv, "profile": job_profile} diff --git a/regimes/client.py b/regimes/client.py index 220e1d5..293e636 100644 --- a/regimes/client.py +++ b/regimes/client.py @@ -33,6 +33,7 @@ class Row: ttft_s: float | None = None e2e_s: float | None = None n_tokens: int = 0 + prompt_tokens: int | None = None # from the server's usage tpot_s: float | None = None http_status: int | None = None finish_reason: str | None = None @@ -95,6 +96,8 @@ def send(url: str, model: str, req: Request, scheduled: float, clock, timeout: f return row end = clock() row.text, row.finish_reason, row.e2e_s = "".join(parts), finish, end - scheduled + if usage and usage.get("prompt_tokens") is not None: + row.prompt_tokens = int(usage["prompt_tokens"]) if usage and usage.get("completion_tokens") is not None: row.n_tokens = int(usage["completion_tokens"]) else: diff --git a/tests/fake_engine.py b/tests/fake_engine.py index 2506ebd..f5ab98b 100644 --- a/tests/fake_engine.py +++ b/tests/fake_engine.py @@ -58,7 +58,8 @@ def do_POST(self): with lock, open(os.path.join(os.environ["TELEMETRY_DIR"], "requests.jsonl"), "a") as f: f.write(json.dumps({"trace_id": self.headers.get("X-Trace-Id"), "tokens_out": n}) + "\n") return self._json({"choices": [{"message": {"role": "assistant", "content": text}, "finish_reason": "stop"}], - "usage": {"completion_tokens": n}}) + "usage": {"prompt_tokens": sum(len(m["content"].split()) for m in body["messages"]), + "completion_tokens": n}}) with lock: if queue_limit is not None and waiting[0] >= queue_limit: data = b'{"error": "overloaded"}' @@ -78,7 +79,8 @@ def do_POST(self): time.sleep(token_s) self._chunk({"choices": [{"index": 0, "delta": {"content": t}}]}) self._chunk({"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}) - self._chunk({"choices": [], "usage": {"completion_tokens": len(text)}}) + self._chunk({"choices": [], "usage": {"prompt_tokens": sum(len(m["content"].split()) for m in body["messages"]), + "completion_tokens": len(text)}}) if os.environ.get("TELEMETRY_DIR"): with lock, open(os.path.join(os.environ["TELEMETRY_DIR"], "requests.jsonl"), "a") as f: f.write(json.dumps({"trace_id": self.headers.get("X-Trace-Id"), "tokens_out": len(text)}) + "\n") diff --git a/tests/test_lab_calibrate.py b/tests/test_lab_calibrate.py new file mode 100644 index 0000000..735b35e --- /dev/null +++ b/tests/test_lab_calibrate.py @@ -0,0 +1,86 @@ +"""Calibrated results: ceilings and reference only under a matching fingerprint, kernel shares, and the baseline +record carrying its context. No GPU.""" +from __future__ import annotations + +import json + +import pytest + +from lab import calibrate, target +from tests.test_validity_ceiling import QWEN3_30B_A3B + +GPU = {"name": "NVIDIA H200", "driver_version": "580.1", "clocks.sm": "1980"} + + +def rows(regime="single_stream", n=20, prompt=600, out=100): + return [{"regime": regime, "status": "ok", "scheduled_s": i * 1.0, "e2e_s": 0.9, "prompt_tokens": prompt, + "n_tokens": out} for i in range(n)] + + +def results(regime="single_stream", objective="tpot_p99_s", value=0.02, better="lower"): + return [{"regime": regime, "objective": objective, "value": value, "better": better, "valid": True, + "invalid_reasons": []}] + + +@pytest.fixture +def t(monkeypatch): + monkeypatch.setattr(calibrate, "model_config", lambda model: QWEN3_30B_A3B) + return target.load() + + +def test_results_get_ceiling_pct_and_shape(t): + cal = calibrate.calibrate_results(results(), rows(), t, GPU, "short", "seen")["single_stream"] + assert cal["value"] == 0.02 and cal["ceiling"] < 0.02 and 0 < cal["pct_of_ceiling"] < 100 + assert cal["bound"] == "memory" and cal["peaks_source"] == "spec" and cal["shape"]["prompt_tokens_mean"] == 600 + assert cal["fingerprint"]["gpu"] == "NVIDIA H200" and cal["fingerprint"]["w_dtype"] == "bf16" + assert cal["reference"] is None and "no reference" in cal["reference_missing"] + + +def test_missing_inputs_are_named_not_guessed(t, monkeypatch): + no_gpu = calibrate.calibrate_results(results(), rows(), t, None, "short", "seen")["single_stream"] + assert no_gpu["ceiling"] is None and "peaks" in no_gpu["ceiling_missing"] + no_rows = calibrate.calibrate_results(results(), [], t, GPU, "short", "seen")["single_stream"] + assert no_rows["ceiling"] is None and "prompt token" in no_rows["ceiling_missing"] + ratio = calibrate.calibrate_results(results("bursty", "speedup", 1.2, "higher"), rows("bursty"), t, GPU, + "short", "seen")["bursty"] + assert ratio["ceiling"] is None and "no physical bound" in ratio["ceiling_missing"] + monkeypatch.setattr(calibrate, "model_config", lambda model: None) + assert "model config" in calibrate.calibrate_results(results(), rows(), t, GPU, "short", "seen")["single_stream"]["ceiling_missing"] + + +def test_reference_only_under_the_same_fingerprint(t, tmp_path, monkeypatch): + monkeypatch.setattr(calibrate, "REFERENCE_DIR", tmp_path) + fp = calibrate.fingerprint(t, GPU, "short", "seen") + calibrate.write_reference(t, "single_stream", results(value=0.015)[0], fp, rows()) + same = calibrate.calibrate_results(results(), rows(), t, GPU, "short", "seen")["single_stream"] + assert same["reference"] == 0.015 and same["pct_of_reference"] == pytest.approx(75) + other_gpu = calibrate.calibrate_results(results(), rows(), t, {**GPU, "name": "NVIDIA H100"}, "short", "seen")["single_stream"] + assert other_gpu["reference"] is None and "gpu" in other_gpu["reference_missing"] + full = calibrate.calibrate_results(results(), rows(), t, GPU, "full", "seen")["single_stream"] + assert full["reference"] is None and "tier" in full["reference_missing"] + stored = json.loads((tmp_path / t.name / "single_stream.json").read_text()) + assert stored["server"] == "reference" and stored["shape"]["output_tokens_mean"] == 100 + + +def test_kernel_shares_and_ncu_rates(tmp_path): + b = tmp_path / "b1" + b.mkdir() + (b / "meta.json").write_text(json.dumps({"window": {"wall_s": 2.0}})) + (b / "kernels.json").write_text(json.dumps({"kernels": [ + {"name": "decode_attn", "count": 10, "total_us": 1_000_000}, {"name": "gemm", "count": 5, "total_us": 500_000}]})) + k = calibrate.calibrate_kernels(b, GPU) + assert k["kernels"][0]["share_of_window"] == pytest.approx(0.5) and k["kernel_share_total"] == pytest.approx(0.75) + assert calibrate.calibrate_kernels(tmp_path, GPU) is None + csvp = tmp_path / "metrics.csv" + csvp.write_text('"ID","Kernel Name","gpu__time_duration.sum","dram__bytes.sum.per_second"\n' + '"","","us","Tbyte/s"\n"0","gemm","12.5","2.4"\n"1","gemm","12.0","2.0"\n') + n = calibrate.calibrate_ncu(csvp, GPU) + assert [round(x["pct_of_bandwidth_peak"]) for x in n["launches"]] == [50, 42] and n["launches"][0]["duration_unit"] == "us" + assert calibrate.calibrate_ncu(csvp, None)["launches"][0]["pct_of_bandwidth_peak"] is None + + +def test_facts_for_the_brief(t): + f = calibrate.facts(t, GPU) + assert f["peaks"]["source"] == "spec" and f["model"]["active_params"] > 2.9e9 + assert 0.001 < f["floors_s"]["decode_step_batch1_ctx1k"] < 0.003 + assert calibrate.facts(t, None)["peaks"] is None diff --git a/tests/test_lab_evaltools.py b/tests/test_lab_evaltools.py index cec9d0f..ed0147c 100644 --- a/tests/test_lab_evaltools.py +++ b/tests/test_lab_evaltools.py @@ -119,6 +119,10 @@ def test_bench_serves_and_measures(lab): assert out["metrics"]["single_stream"]["value"] is not None rec = next(ledger.records(s.ledger_root, kind="bench")) assert rec["config"] == {"split": "seen", "tier": "short", "seed": tb.seed} and rec["result"]["verdict"] == "ok" + cal = out["calibrated"]["single_stream"] # every number with its context, or why it has none + assert cal["value"] == out["metrics"]["single_stream"]["value"] and cal["fingerprint"]["tier"] == "short" + assert ("ceiling_missing" in cal) == (cal.get("ceiling") is None) and "reference_missing" in cal + assert cal["shape"]["prompt_tokens_mean"] > 0 if cal.get("shape") else True assert rec["result"]["ready_s"] is not None and (s.run_dir / "serve-bench.log").exists() diff --git a/tests/test_lab_validate.py b/tests/test_lab_validate.py index 9df8bd9..4d4b940 100644 --- a/tests/test_lab_validate.py +++ b/tests/test_lab_validate.py @@ -62,7 +62,12 @@ def test_validate_separates_and_measures_bands(spec, tmp_path, monkeypatch): items = [{"id": f"mc-{k}", "task": "mmlu_pro", "gold": "C", "max_tokens": 6, "messages": [{"role": "user", "content": f"question {k} pick one"}]} for k in range(60)] monkeypatch.setattr(tasks, "load", lambda names, tier="dev", **kw: items) + from lab import calibrate + monkeypatch.setattr(calibrate, "REFERENCE_DIR", tmp_path / "reference") rep = validate.validate(spec, noise_runs=2, regimes=["single_stream"], out=tmp_path / "out", encoder=Enc()) + ref = json.loads((tmp_path / "reference" / spec.name / "single_stream.json").read_text()) + assert ref["server"] == "reference" and ref["fingerprint"]["tier"] == "short" and "single_stream" in rep["reference_regimes"] + assert ref["value"] is not None and ref["shape"]["prompt_tokens_mean"] > 0 assert rep["candidates"]["same"]["verdict"] == "pass" assert rep["candidates"]["wrong4"]["verdict"] == "fail", rep["candidates"]["wrong4"] assert rep["gate_separates"] is True diff --git a/tests/test_validity_ceiling.py b/tests/test_validity_ceiling.py new file mode 100644 index 0000000..7496748 --- /dev/null +++ b/tests/test_validity_ceiling.py @@ -0,0 +1,91 @@ +"""Physical ceilings from a model config and hardware peaks: Qwen3-30B-A3B's known counts, floors below any real +measurement, and the regime bounds. No GPU.""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from validity import ceiling, peaks +from validity.ceiling import ModelSpec, Shape + +QWEN3_30B_A3B = {"num_hidden_layers": 48, "hidden_size": 2048, "num_attention_heads": 32, "num_key_value_heads": 4, + "head_dim": 128, "vocab_size": 151936, "intermediate_size": 6144, "num_experts": 128, + "num_experts_per_tok": 8, "moe_intermediate_size": 768, "tie_word_embeddings": False} +LLAMA_8B = {"num_hidden_layers": 32, "hidden_size": 4096, "num_attention_heads": 32, "num_key_value_heads": 8, + "vocab_size": 128256, "intermediate_size": 14336} +H200 = peaks.Peaks("NVIDIA H200", 989e12, 4.8e12, 141e9, "spec") + + +def test_qwen3_moe_parameter_counts_match_the_model_card(): + m = ModelSpec.from_hf_config(QWEN3_30B_A3B, "qwen") + assert 2.9e9 < m.active_params < 3.4e9 # "3B activated" + assert 29e9 < m.total_params < 31e9 # "30B total" + assert m.kv_bytes_per_token("bf16") == 2 * 48 * 4 * 128 * 2 + assert m.experts_touched(1) == 8 and 120 < m.experts_touched(1000) <= 128 + + +def test_dense_llama_counts(): + m = ModelSpec.from_hf_config(LLAMA_8B, "llama") + assert 7.5e9 < m.active_params == m.total_params - LLAMA_8B["vocab_size"] * 4096 < 8.5e9 + assert m.head_dim == 128 and m.n_experts == 0 + + +def test_decode_floor_is_memory_bound_at_small_batch_and_grows_with_batch(): + m = ModelSpec.from_hf_config(QWEN3_30B_A3B) + f1 = ceiling.decode_step_floor(m, H200, 1, 1024) + f64 = ceiling.decode_step_floor(m, H200, 64, 1024) + assert f1["bound"] == "memory" and 0.001 < f1["floor_s"] < 0.003 # ~6 GB of weights at 4.8 TB/s + assert f64["floor_s"] > f1["floor_s"] and f64["floor_s"] < 64 * f1["floor_s"] + + +def test_prefill_floor_and_goodput_ceiling(): + m = ModelSpec.from_hf_config(QWEN3_30B_A3B) + p = ceiling.prefill_floor(m, H200, 4096) + assert p["floor_s"] > ceiling.prefill_floor(m, H200, 512)["floor_s"] + b = ceiling.max_batch_under(m, H200, 0.1, 1024) + assert b > 64 and ceiling.decode_step_floor(m, H200, b, 1024)["floor_s"] <= 0.1 + assert ceiling.decode_step_floor(m, H200, b + 1, 1024)["floor_s"] > 0.1 + assert ceiling.max_batch_under(m, H200, 1e-6, 1024) == 0 and ceiling.tokens_per_s_ceiling(m, H200, 1e-6, 1024) == 0 + + +def test_regime_ceilings_by_objective(): + m = ModelSpec.from_hf_config(QWEN3_30B_A3B) + shape = Shape(600, 2000, 200, concurrency=1) + lim = {"ttft_s": 2.0, "tpot_s": 0.1} + tpot = ceiling.regime_ceiling("tpot_p99_s", m, H200, shape, lim) + ttft = ceiling.regime_ceiling("ttft_p99_s", m, H200, shape, lim) + good = ceiling.regime_ceiling("goodput_req_per_s", m, H200, shape, lim) + assert tpot["better"] == "lower" and tpot["ceiling"] < 0.01 + assert ttft["better"] == "lower" and "p99 prompt" in ttft["basis"] + assert good["better"] == "higher" and good["ceiling"] > 10 + assert ceiling.regime_ceiling("speedup", m, H200, shape, lim) is None + + +def test_pct_of_and_ceiling_gain(): + assert ceiling.pct_of(0.004, 0.002, "lower") == pytest.approx(50) + assert ceiling.pct_of(50, 200, "higher") == pytest.approx(25) + assert ceiling.pct_of(None, 1, "lower") is None and ceiling.pct_of(1, 0, "higher") is None + assert ceiling.ceiling_gain(0.4, 25) == pytest.approx(0.3) and ceiling.ceiling_gain(0.4, None) is None + assert ceiling.ceiling_gain(0.4, 150) == 0 + + +def test_peaks_measured_file_wins_over_spec_and_driver_must_match(tmp_path): + assert peaks.load("NVIDIA H200").source == "spec" and peaks.load("Radeon") is None + p = peaks.Peaks("NVIDIA H200", 900e12, 4.5e12, 141e9, "measured", driver="580.1") + out = peaks.write(p, tmp_path) + assert out.name == "nvidia-h200.json" and json.loads(out.read_text())["bw_bytes_s"] == 4.5e12 + assert peaks.load("NVIDIA H200", "580.1", tmp_path).source == "measured" + assert peaks.load("NVIDIA H200", None, tmp_path).source == "measured" + assert peaks.load("NVIDIA H200", "581.0", tmp_path).source == "spec" # another driver: back to the sheet + + +def test_measure_needs_cuda(): + pytest.importorskip("torch") + import torch + if torch.cuda.is_available(): + pytest.skip("a GPU is present; the microbench is checked on the VM") + with pytest.raises(RuntimeError): + peaks.measure(0.1) + assert not (Path(peaks.PEAKS_DIR) / "radeon.json").exists() diff --git a/validity/__main__.py b/validity/__main__.py index 3458bd5..a655527 100644 --- a/validity/__main__.py +++ b/validity/__main__.py @@ -1,4 +1,5 @@ -"""python -m validity repeat --url URL [--n 20] [--api chat|completions] [--target T]""" +"""python -m validity repeat --url URL [--n 20] [--api chat|completions] [--target T] +python -m validity peaks [--seconds 3] [--out DIR] (measures this GPU's bandwidth and bf16 peaks; needs torch+CUDA)""" from __future__ import annotations import argparse @@ -14,7 +15,14 @@ def main(): sub = ap.add_subparsers(dest="kind", required=True) q = sub.add_parser("repeat"); q.add_argument("--url", required=True) q.add_argument("--n", type=int, default=20); q.add_argument("--api", choices=("chat", "completions"), default="chat") + pk = sub.add_parser("peaks"); pk.add_argument("--seconds", type=float, default=3.0); pk.add_argument("--out") a = ap.parse_args() + if a.kind == "peaks": + from pathlib import Path + from validity import peaks + p = peaks.measure(a.seconds) + print(peaks.write(p, Path(a.out) if a.out else None), json.dumps(p.__dict__), sep="\n") + return t = target.load(a.target) res = repeat_audit(a.url, t.model, PROMPTS[:a.n], api=a.api, chat_kwargs=t.chat_kwargs) print("PASS" if res["passed"] else "FAIL", *res["reasons"], json.dumps(res["evidence"]), sep="\n ") diff --git a/validity/ceiling.py b/validity/ceiling.py new file mode 100644 index 0000000..aea79f4 --- /dev/null +++ b/validity/ceiling.py @@ -0,0 +1,181 @@ +"""Physical ceilings for an LLM engine, from the model's HF config and the GPU's measured peaks (validity/peaks.py). +No GPU, no weights. Everything here is a bound, not a target: a step cannot be faster than streaming the weights +and KV it needs, or than the tensor cores can multiply. + +Per decode step the engine must read every weight a token touches (attention, router, lm_head, and the experts +the batch routes to), read each sequence's KV cache, and do 2 FLOP per active parameter per token. Prefill of P +tokens reads the weights once and does 2·P FLOP per active parameter plus the attention FLOPs. +step floor = max(bytes / bandwidth, FLOPs / peak). The gap from a floor to a measurement is overhead.""" +from __future__ import annotations + +from dataclasses import dataclass + +from validity.peaks import Peaks + +DTYPE_BYTES = {"bf16": 2, "fp16": 2, "fp8": 1, "int8": 1, "int4": 0.5} + + +@dataclass(frozen=True) +class ModelSpec: + name: str + n_layers: int + hidden: int + n_q_heads: int + n_kv_heads: int + head_dim: int + vocab: int + intermediate: int # dense MLP width (0 for a pure MoE) + n_experts: int = 0 + top_k: int = 0 + moe_intermediate: int = 0 + shared_intermediate: int = 0 # always-on shared expert width (DeepSeek-style), 0 if none + tie_embeddings: bool = False + + @classmethod + def from_hf_config(cls, cfg: dict, name: str = "") -> "ModelSpec": + """Llama-style dense and Qwen3/Mixtral/DeepSeek-style MoE configs.""" + heads = cfg["num_attention_heads"] + n_exp = cfg.get("num_experts") or cfg.get("num_local_experts") or cfg.get("n_routed_experts") or 0 + return cls(name or cfg.get("_name_or_path", ""), cfg["num_hidden_layers"], cfg["hidden_size"], heads, + cfg.get("num_key_value_heads", heads), cfg.get("head_dim") or cfg["hidden_size"] // heads, + cfg["vocab_size"], 0 if n_exp else cfg["intermediate_size"], n_exp, + cfg.get("num_experts_per_tok") or cfg.get("num_experts_per_token") or 0, + cfg.get("moe_intermediate_size") or (cfg["intermediate_size"] if n_exp else 0), + (cfg.get("moe_intermediate_size") or 0) * (cfg.get("n_shared_experts") or 0), + bool(cfg.get("tie_word_embeddings", False))) + + # -- parameter counts ---------------------------------------------------------------- + @property + def attn_per_layer(self) -> int: + return self.hidden * (self.n_q_heads + 2 * self.n_kv_heads) * self.head_dim + self.n_q_heads * self.head_dim * self.hidden + + @property + def expert_params(self) -> int: + """One expert: gate, up and down projections.""" + return 3 * self.hidden * self.moe_intermediate + + @property + def dense_mlp_per_layer(self) -> int: + return 3 * self.hidden * (self.intermediate + self.shared_intermediate) + + @property + def router_per_layer(self) -> int: + return self.hidden * self.n_experts + + @property + def lm_head(self) -> int: + return self.vocab * self.hidden + + @property + def always_on(self) -> int: + """Parameters every token reads regardless of routing (the embedding table is gathered, not streamed).""" + return self.n_layers * (self.attn_per_layer + self.dense_mlp_per_layer + self.router_per_layer) + self.lm_head + + @property + def active_params(self) -> int: + return self.always_on + self.n_layers * self.top_k * self.expert_params + + @property + def total_params(self) -> int: + return self.always_on + self.n_layers * self.n_experts * self.expert_params + (0 if self.tie_embeddings else self.vocab * self.hidden) + + def kv_bytes_per_token(self, kv_dtype: str = "bf16") -> float: + return 2 * self.n_layers * self.n_kv_heads * self.head_dim * DTYPE_BYTES[kv_dtype] + + def experts_touched(self, tokens: int) -> float: + """Expected distinct experts a layer reads for `tokens` routed tokens, uniform routing (the engine can do no + better than reading each touched expert once).""" + if not self.n_experts: + return 0.0 + return self.n_experts * (1 - (1 - self.top_k / self.n_experts) ** tokens) + + def weight_bytes(self, tokens: int, w_dtype: str = "bf16") -> float: + """Bytes of weights one forward over `tokens` tokens must stream.""" + return (self.always_on + self.n_layers * self.experts_touched(tokens) * self.expert_params) * DTYPE_BYTES[w_dtype] + + +# -- floors (seconds) --------------------------------------------------------------------- +def decode_step_floor(m: ModelSpec, hw: Peaks, batch: int, context: float, w_dtype: str = "bf16", + kv_dtype: str = "bf16") -> dict: + """One decode step for `batch` sequences at mean `context` tokens.""" + w = m.weight_bytes(batch, w_dtype) + kv = batch * context * m.kv_bytes_per_token(kv_dtype) + flops = 2 * m.active_params * batch + 4 * m.n_layers * m.n_q_heads * m.head_dim * context * batch + t_mem, t_compute = (w + kv) / hw.bw_bytes_s, flops / hw.bf16_flops + return {"floor_s": max(t_mem, t_compute), "t_mem_s": t_mem, "t_compute_s": t_compute, + "bound": "memory" if t_mem >= t_compute else "compute", "weight_bytes": w, "kv_bytes": kv, "flops": flops} + + +def prefill_floor(m: ModelSpec, hw: Peaks, prompt_tokens: int, w_dtype: str = "bf16", kv_dtype: str = "bf16") -> dict: + """Prefill of one prompt: weights once, KV written once, quadratic attention.""" + w = m.weight_bytes(prompt_tokens, w_dtype) + kv = prompt_tokens * m.kv_bytes_per_token(kv_dtype) + flops = 2 * m.active_params * prompt_tokens + 2 * m.n_layers * m.n_q_heads * m.head_dim * prompt_tokens * (prompt_tokens + 1) + t_mem, t_compute = (w + kv) / hw.bw_bytes_s, flops / hw.bf16_flops + return {"floor_s": max(t_mem, t_compute), "t_mem_s": t_mem, "t_compute_s": t_compute, + "bound": "memory" if t_mem >= t_compute else "compute", "weight_bytes": w, "kv_bytes": kv, "flops": flops} + + +def max_batch_under(m: ModelSpec, hw: Peaks, tpot_limit_s: float, context: float, w_dtype: str = "bf16", + kv_dtype: str = "bf16", cap: int = 1 << 16) -> int: + """Largest batch whose decode-step floor still meets the TPOT limit (the floor grows with batch).""" + lo, hi = 0, cap + while lo < hi: + mid = (lo + hi + 1) // 2 + if decode_step_floor(m, hw, mid, context, w_dtype, kv_dtype)["floor_s"] <= tpot_limit_s: + lo = mid + else: + hi = mid - 1 + return lo + + +def tokens_per_s_ceiling(m: ModelSpec, hw: Peaks, tpot_limit_s: float, context: float, w_dtype: str = "bf16", + kv_dtype: str = "bf16") -> float: + """Output tokens/s at the largest batch the TPOT limit allows; 0 when even one sequence cannot meet it.""" + b = max_batch_under(m, hw, tpot_limit_s, context, w_dtype, kv_dtype) + return b / decode_step_floor(m, hw, b, context, w_dtype, kv_dtype)["floor_s"] if b else 0.0 + + +# -- regime ceilings ------------------------------------------------------------------------ +@dataclass(frozen=True) +class Shape: + """What a regime's requests looked like: the ceiling depends on them.""" + prompt_tokens_mean: float + prompt_tokens_p99: float + output_tokens_mean: float + concurrency: float = 1.0 # mean sequences in flight + + +def regime_ceiling(objective: str, m: ModelSpec, hw: Peaks, shape: Shape, limits: dict, + w_dtype: str = "bf16", kv_dtype: str = "bf16") -> dict | None: + """The physical bound on a regime's objective, or None where none exists (speedup ratios, cold start).""" + ctx = shape.prompt_tokens_mean + shape.output_tokens_mean / 2 + if objective == "tpot_p99_s": + f = decode_step_floor(m, hw, max(1, round(shape.concurrency)), ctx, w_dtype, kv_dtype) + return {"ceiling": f["floor_s"], "better": "lower", "bound": f["bound"], + "basis": f"decode step floor at batch {max(1, round(shape.concurrency))}, context {ctx:.0f}"} + if objective == "ttft_p99_s": + f = prefill_floor(m, hw, int(shape.prompt_tokens_p99), w_dtype, kv_dtype) + return {"ceiling": f["floor_s"], "better": "lower", "bound": f["bound"], + "basis": f"prefill floor at the p99 prompt, {shape.prompt_tokens_p99:.0f} tokens"} + if objective in ("goodput_req_per_s", "good_req_per_s", "goodput_sessions_per_s"): + tps = tokens_per_s_ceiling(m, hw, limits["tpot_s"], ctx, w_dtype, kv_dtype) + per = tps / shape.output_tokens_mean if shape.output_tokens_mean else 0.0 + return {"ceiling": per, "better": "higher", "bound": "memory", + "basis": f"decode tokens/s at the largest batch meeting TPOT <= {limits['tpot_s']} s, " + f"context {ctx:.0f}, / {shape.output_tokens_mean:.0f} output tokens per request"} + return None + + +def pct_of(value: float | None, ceiling: float | None, better: str) -> float | None: + """How close a measurement is to its bound, 100 = at the bound; None without both numbers.""" + if value is None or ceiling is None or not value or not ceiling: + return None + return (ceiling / value if better == "lower" else value / ceiling) * 100 + + +def ceiling_gain(share_of_total: float, pct_of_ceiling: float | None) -> float | None: + """Fraction of the total that disappears if this part reached its ceiling (Amdahl); None without a ceiling.""" + if pct_of_ceiling is None or pct_of_ceiling <= 0: + return None + return share_of_total * (1 - min(pct_of_ceiling, 100.0) / 100) diff --git a/validity/peaks.py b/validity/peaks.py new file mode 100644 index 0000000..a72a3ee --- /dev/null +++ b/validity/peaks.py @@ -0,0 +1,104 @@ +"""Hardware peaks the ceilings are computed against: HBM bandwidth and bf16 tensor-core throughput, measured once +per GPU + driver (`python -m validity peaks`, run by lab/vm-setup.sh) into knowledge/peaks/.json. Spec-sheet +numbers are the fallback, labelled `source: spec`; a ceiling from them says so.""" +from __future__ import annotations + +import json +import re +import time +from dataclasses import asdict, dataclass +from pathlib import Path + +from lab import target + +PEAKS_DIR = target.ENV_ROOT / "knowledge" / "peaks" + +# Vendor spec sheets: dense bf16 tensor-core FLOP/s (no sparsity), HBM bytes/s, HBM bytes. +SPEC = { + "H200": (989e12, 4.8e12, 141e9), "H100": (989e12, 3.35e12, 80e9), "B200": (2250e12, 8e12, 192e9), + "A100": (312e12, 2.0e12, 80e9), "L40S": (362e12, 0.864e12, 48e9), "A10G": (125e12, 0.6e12, 24e9), +} + + +@dataclass(frozen=True) +class Peaks: + gpu: str # nvidia-smi's name + bf16_flops: float # dense bf16 FLOP/s + bw_bytes_s: float # HBM bytes/s + mem_bytes: float + source: str # measured | spec + driver: str | None = None + sm_clock_mhz: int | None = None + measured_at: str | None = None + method: str | None = None + + +def slug(gpu: str) -> str: + return re.sub(r"[^a-z0-9]+", "-", gpu.lower()).strip("-") + + +def spec_for(gpu: str) -> Peaks | None: + for key, (flops, bw, mem) in SPEC.items(): + if key.lower() in gpu.lower(): + return Peaks(gpu, flops, bw, mem, "spec") + return None + + +def load(gpu: str, driver: str | None = None, peaks_dir: Path | None = None) -> Peaks | None: + """Measured peaks for this GPU (and driver, when known); else the spec sheet; else None.""" + p = (peaks_dir or PEAKS_DIR) / f"{slug(gpu)}.json" + if p.exists(): + d = json.loads(p.read_text()) + if driver is None or d.get("driver") in (None, driver): + return Peaks(**d) + return spec_for(gpu) + + +def measure(seconds: float = 3.0) -> Peaks: + """Copy bandwidth and a large bf16 GEMM on the first CUDA device; the best of repeated trials (clocks warm).""" + import torch + if not torch.cuda.is_available(): + raise RuntimeError("no CUDA device") + dev = torch.device("cuda:0") + name = torch.cuda.get_device_name(dev) + props = torch.cuda.get_device_properties(dev) + n = 1 << 28 # 512 MiB per bf16 buffer, well past L2 + a, b = torch.empty(n, dtype=torch.bfloat16, device=dev), torch.empty(n, dtype=torch.bfloat16, device=dev) + best_bw = 0.0 + t_end = time.monotonic() + seconds + while time.monotonic() < t_end: + torch.cuda.synchronize() + t0 = time.perf_counter() + b.copy_(a) + torch.cuda.synchronize() + best_bw = max(best_bw, 2 * n * 2 / (time.perf_counter() - t0)) # read + write, 2 bytes each + del a, b + k = 8192 + x = torch.randn(k, k, dtype=torch.bfloat16, device=dev) + y = torch.randn(k, k, dtype=torch.bfloat16, device=dev) + best_flops = 0.0 + t_end = time.monotonic() + seconds + while time.monotonic() < t_end: + torch.cuda.synchronize() + t0 = time.perf_counter() + x @ y + torch.cuda.synchronize() + best_flops = max(best_flops, 2 * k ** 3 / (time.perf_counter() - t0)) + driver = sm = None + try: + import subprocess + out = subprocess.run(["nvidia-smi", "--query-gpu=driver_version,clocks.sm", "--format=csv,noheader,nounits"], + capture_output=True, text=True, timeout=10).stdout.strip().split(",") + driver, sm = out[0].strip(), int(out[1]) + except (OSError, ValueError, IndexError, subprocess.TimeoutExpired): + pass + return Peaks(name, best_flops, best_bw, float(props.total_memory), "measured", driver, sm, + time.strftime("%Y-%m-%dT%H:%M:%S"), f"best of {seconds}s: 512MiB bf16 copy; {k}^3 bf16 GEMM") + + +def write(p: Peaks, peaks_dir: Path | None = None) -> Path: + d = peaks_dir or PEAKS_DIR + d.mkdir(parents=True, exist_ok=True) + out = d / f"{slug(p.gpu)}.json" + out.write_text(json.dumps(asdict(p), indent=1) + "\n") + return out diff --git a/validity/roofline.py b/validity/roofline.py deleted file mode 100644 index 3ce541b..0000000 --- a/validity/roofline.py +++ /dev/null @@ -1,310 +0,0 @@ -"""Analytical roofline+ for LLM decode — offline ceiling, per-step breakdown, -ranked lever menu, gap attribution. No GPU, no weights. -Moved from inference-server 2026-10-06 as-is: still hard-wired to Gemma E4B/E2B on A100 (dense weights, embedded -measured anchors). Not yet generalized to the eval target (Qwen3-30B-A3B MoE, H200); not a limit check. - -The model: per decode step you (a) stream the dense weights from HBM once for the -whole batch, (b) stream each sequence's KV cache, (c) do 2*N_params*B FLOP of math. -Ideal step time = max(compute_time, memory_time); tok/s = B / step_time. The gap -between that ceiling and what we measure is overhead (launches, data-movement). -""" - -from dataclasses import dataclass - -# --------------------------------------------------------------------------- -# Hardware presets. flops = dense bf16 tensor-core FLOP/s; bw = HBM bytes/s. -# fp8_compute = has native fp8 tensor cores (A100 does NOT — quant is bytes-only there). -# --------------------------------------------------------------------------- -@dataclass -class HW: - name: str - flops: float # bf16 FLOP/s - bw: float # HBM bytes/s - mem: float # HBM bytes - fp8_compute: bool # native fp8 tensor cores (doubles compute roof) - -HARDWARE = { - "A100-80GB": HW("A100-80GB", 312e12, 2.039e12, 80e9, fp8_compute=False), - "A10G": HW("A10G", 125e12, 0.600e12, 24e9, fp8_compute=False), - "H100-80GB": HW("H100-80GB", 990e12, 3.350e12, 80e9, fp8_compute=True), -} - -DTYPE_BYTES = {"bf16": 2, "fp8": 1, "int8": 1} - - -# --------------------------------------------------------------------------- -# Model spec. Values read from each model's HF config.json (cited inline). -# --------------------------------------------------------------------------- -@dataclass -class Model: - name: str - n_layers: int - hidden: int - intermediate: int - n_q_heads: int - n_kv_heads: int - head_dim: int # sliding-layer head_dim - global_head_dim: int # full-layer head_dim - sliding_window: int - vocab: int - n_kv_shared: int # trailing layers that reuse an earlier layer's KV - hidden_per_layer: int # PLE per-layer embedding width - full_every: int = 6 # a full-attention layer every Nth (index % full_every == full_every-1) - - @property - def first_shared(self) -> int: - return self.n_layers - self.n_kv_shared - - def is_full(self, i: int) -> bool: - return i % self.full_every == self.full_every - 1 - - def stores_kv(self, i: int) -> bool: - return i < self.first_shared # trailing shared layers hold no KV of their own - - def layer_hd(self, i: int) -> int: - return self.global_head_dim if self.is_full(i) else self.head_dim - - -# google/gemma-4-E4B-it/config.json (42L, 8:2 GQA, hd 256/512, window 512, 18 shared) -E4B = Model("E4B", 42, 2560, 10240, 8, 2, 256, 512, 512, 262144, 18, 256) -# google/gemma-4-E2B-it/config.json (hidden 1536; smaller twin — for cross-model sanity) -E2B = Model("E2B", 30, 1536, 8192, 4, 1, 256, 256, 512, 262144, 12, 256, full_every=6) - - -# --------------------------------------------------------------------------- -# Empirical gap attribution — isolated compiled decode, A100/E4B, N=32 -# (benchmarks/tuning_log.md profiling, 2026-06-14). Fractions of achieved step time. -# --------------------------------------------------------------------------- -PROFILE = { - "attention (paged kernels)": 0.24, - "GEMM (weight matmuls)": 0.23, - "small-kernel launch tail": 0.19, - "DtoD memcpy (KV/act)": 0.16, - "KV scatter (index_put)": 0.15, - "norm/rope/act": 0.04, -} -# Which buckets are pure overhead the roofline says shouldn't exist: -OVERHEAD_BUCKETS = ["small-kernel launch tail", "DtoD memcpy (KV/act)", "KV scatter (index_put)"] - -# Measured tok/s to overlay (A100/E4B). Closed-loop = full serving sweep; isolated = decode-only. -MEASURED_CLOSED = {16: 658, 32: 1151} # compile-default, tuning_log anchor -MEASURED_ISOLATED = {32: 1870} # isolated decode, profile_decode_kernels_modal - - -# --------------------------------------------------------------------------- -# Core cost model -# --------------------------------------------------------------------------- -def dense_params(m: Model) -> dict: - """Params streamed from HBM every decode step, split into buckets.""" - attn = mlp = 0 - for i in range(m.n_layers): - hd = m.layer_hd(i) - attn += m.hidden * m.n_q_heads * hd # q_proj - attn += m.n_q_heads * hd * m.hidden # o_proj - if m.stores_kv(i): - attn += 2 * (m.hidden * m.n_kv_heads * hd) # k_proj + v_proj - mlp += 3 * m.hidden * m.intermediate # gate + up + down - lm_head = m.vocab * m.hidden # tied embed, read for logits - proj = m.hidden * m.n_layers * m.hidden_per_layer # per-layer-embed projection - return {"attn": attn, "mlp": mlp, "lm_head": lm_head, "proj": proj, - "total": attn + mlp + lm_head + proj} - - -def ple_table_params(m: Model) -> int: - """PLE embedding table — GATHERED per token, not streamed. The 'effective vs total' gap.""" - return m.vocab * m.n_layers * m.hidden_per_layer - - -def kv_bytes(m: Model, B: int, L: int, dtype: str) -> int: - """KV bytes for B sequences at context L — sliding-window + KV-sharing aware.""" - per_seq = 0 - for i in range(m.n_layers): - if not m.stores_kv(i): - continue - toks = min(L, m.sliding_window) if not m.is_full(i) else L - per_seq += 2 * m.n_kv_heads * m.layer_hd(i) * toks # K + V - return per_seq * DTYPE_BYTES[dtype] * B - - -def attn_flops(m: Model, B: int, L: int) -> float: - """Decode attention FLOP (1 query attends L_eff keys) — small but grows with L.""" - f = 0 - for i in range(m.n_layers): - L_eff = min(L, m.sliding_window) if not m.is_full(i) else L - f += 2 * 2 * m.n_q_heads * m.layer_hd(i) * L_eff # QK^T + AV - return f * B - - -def step(m: Model, hw: HW, B: int, L: int, w_dtype: str = "bf16", kv_dtype: str = "bf16") -> dict: - """Ideal per-step timing + regime. Times in seconds.""" - dp = dense_params(m) - w_bytes = dp["total"] * DTYPE_BYTES[w_dtype] - kvb = kv_bytes(m, B, L, kv_dtype) - flops = 2 * dp["total"] * B + attn_flops(m, B, L) - # fp8 doubles the compute roof only where the GPU has fp8 tensor cores. - peak = hw.flops * (2 if (w_dtype == "fp8" and hw.fp8_compute) else 1) - t_compute = flops / peak - t_weight = w_bytes / hw.bw - t_kv = kvb / hw.bw - t_mem = t_weight + t_kv - t_step = max(t_compute, t_mem) - return { - "t_compute": t_compute, "t_weight": t_weight, "t_kv": t_kv, "t_mem": t_mem, - "t_step": t_step, "tok_s": B / t_step, - "regime": "compute" if t_compute > t_mem else "memory", - "w_bytes": w_bytes, "kv_bytes": kvb, "flops": flops, - } - - -def ridge_batch(m: Model, hw: HW, L: int, w_dtype: str = "bf16") -> float: - """Batch where t_compute == t_mem (the roofline knee) — solved numerically.""" - lo, hi = 1.0, 4096.0 - for _ in range(40): - mid = (lo + hi) / 2 - s = step(m, hw, max(1, round(mid)), L, w_dtype) - if s["t_compute"] < s["t_mem"]: - lo = mid - else: - hi = mid - return (lo + hi) / 2 - - -def max_concurrent(m: Model, hw: HW, L: int, w_dtype: str = "bf16", kv_dtype: str = "bf16", - act_gb: float = 4.0) -> int: - """Max sequences at context L before HBM is exhausted (weights + activations + KV).""" - budget = hw.mem - dense_params(m)["total"] * DTYPE_BYTES[w_dtype] - act_gb * 1e9 - per_seq_kv = kv_bytes(m, 1, L, kv_dtype) - return max(0, int(budget / per_seq_kv)) - - -# --------------------------------------------------------------------------- -# Report sections -# --------------------------------------------------------------------------- -def fmt_ms(s: float) -> str: - return f"{s * 1e3:6.2f}ms" - - -def section_ceiling(m: Model, hw: HW): - print(f"\n{'='*74}\n CEILING SWEEP — {m.name} on {hw.name}, bf16\n{'='*74}") - dp = dense_params(m) - ple = ple_table_params(m) - print(f" dense params (streamed/step): {dp['total']/1e9:.2f}B " - f"({dp['total']*2/1e9:.1f}GB bf16) [attn {dp['attn']/1e9:.2f}B · " - f"mlp {dp['mlp']/1e9:.2f}B · lm_head {dp['lm_head']/1e9:.2f}B]") - print(f" PLE table (gathered, NOT streamed): {ple/1e9:.2f}B <- the 'effective-4B vs total' gap\n") - Ls = [128, 512, 2048] - header = " B".ljust(6) + "".join(f"L={L:<5} regime ".ljust(20) for L in Ls) - print(header) - for B in [1, 4, 8, 16, 32, 64, 128, 256]: - row = f" {B:<4}" - for L in Ls: - s = step(m, hw, B, L) - row += f"{s['tok_s']:7.0f} tok/s {s['regime'][:3]} ".ljust(20) - print(row) - print("\n ridge batch B* (knee, memory→compute):", - " ".join(f"L={L}: {ridge_batch(m, hw, L):.0f}" for L in Ls)) - - -def section_breakdown(m: Model, hw: HW, B: int, L: int): - print(f"\n{'='*74}\n PER-STEP BREAKDOWN — {m.name}/{hw.name}, B={B} L={L}, bf16\n{'='*74}") - s = step(m, hw, B, L) - ideal = s["t_step"] - print(f" weight-read {fmt_ms(s['t_weight'])} (dense weights, once/step — the floor)") - print(f" KV-read {fmt_ms(s['t_kv'])} ({s['kv_bytes']/1e6:.0f}MB for {B} seqs)") - print(f" compute {fmt_ms(s['t_compute'])} ({s['flops']/1e9:.0f} GFLOP) [{s['regime']}-bound]") - print(f" {'-'*46}") - print(f" IDEAL step {fmt_ms(ideal)} -> ceiling {s['tok_s']:.0f} tok/s") - - meas = MEASURED_ISOLATED.get(B) or MEASURED_CLOSED.get(B) - if meas: - t_meas = B / meas - overhead = t_meas - ideal - eff = ideal / t_meas - print(f"\n MEASURED {fmt_ms(t_meas)} -> {meas} tok/s ({eff*100:.0f}% of roof)") - print(f" OVERHEAD {fmt_ms(overhead)} ({overhead/t_meas*100:.0f}% of the step) <- the target") - # Attribute the overhead to profiled causes. - ov_frac = sum(PROFILE[b] for b in OVERHEAD_BUCKETS) - print("\n gap attribution (profile fractions of measured step):") - for b in OVERHEAD_BUCKETS: - print(f" {b:28s} {PROFILE[b]*100:4.0f}% ({fmt_ms(PROFILE[b]*t_meas)})") - print(f" {'= addressable overhead':28s} {ov_frac*100:4.0f}% " - f"(fusing these into the kernels is the #1 lever)") - - -def section_levers(m: Model, hw: HW, B: int, L: int): - print(f"\n{'='*74}\n LEVER MENU — {m.name}/{hw.name}, B={B} L={L} (ranked by predicted tok/s)\n{'='*74}") - # Additive-overhead model: measured_step = ideal_step + fixed_overhead. A lever that - # shrinks the ideal (quant) or the overhead (fusion) shifts its own term only. - base = step(m, hw, B, L) - meas = MEASURED_ISOLATED.get(B) or MEASURED_CLOSED.get(B) or base["tok_s"] - t_meas = B / meas - overhead = t_meas - base["t_step"] # fixed tax (s) - addressable = t_meas * sum(PROFILE[b] for b in OVERHEAD_BUCKETS) # fusable part of it - - def predict(new_ideal_s: float, new_overhead_s: float, tokens_mult: float = 1.0) -> float: - return tokens_mult * B / (new_ideal_s + new_overhead_s) - - levers = [] - # Kernel fusion: remove the fusable data-movement/launch overhead. - levers.append(("kernel fusion (kill KV-move+launches)", - predict(base["t_step"], overhead - addressable), - "fold scatter+DtoD+launch tail into the kernels")) - # int8/fp8 weights: shrinks the ideal weight-read term (A100: bytes only). - levers.append(("int8/fp8 weights (bytes)", - predict(step(m, hw, B, L, w_dtype="int8")["t_step"], overhead), - "halve weight bytes; A100 has no fp8 TC so compute roof unchanged")) - # fp8 KV: shrinks the ideal KV-read term. - levers.append(("fp8 KV cache", - predict(step(m, hw, B, L, kv_dtype="fp8")["t_step"], overhead), - "halve KV bytes moved")) - # Speculative decode: same step, ~(1+a*k) tokens out of it while memory-bound. - a, k = 0.7, 3 - levers.append((f"speculative decode (k={k}, accept={a})", - predict(base["t_step"], overhead, tokens_mult=1 + a * k), - "produce multiple tokens per weight-read while memory-bound")) - # Scale-out: run at target batch. Overhead held constant per step (optimistic — directional). - for Bt in (128, 256): - s_t = step(m, hw, Bt, L) - levers.append((f"scale to B={Bt} (bucketed graph)", - Bt / (s_t["t_step"] + overhead), - "more sequences share each weight-read (toward B*)")) - # Combined: fusion + scale + int8 weights (the realistic stack). - s256 = step(m, hw, 256, L, w_dtype="int8") - levers.append(("STACK: fusion + int8 wt + B=256", - 256 / (s256["t_step"] + (overhead - addressable)), - "the compounded ceiling if we do all three")) - - print(f" baseline (measured): {meas:.0f} tok/s ideal {base['tok_s']:.0f} tok/s " - f"({base['t_step']/t_meas*100:.0f}% of step) overhead {overhead*1e3:.1f}ms/step\n") - for name, tok, why in sorted(levers, key=lambda x: -x[1]): - print(f" {name:36s} {tok:7.0f} tok/s {tok/meas:4.1f}x {why}") - print("\n Additive-overhead model (overhead held fixed per step) — DIRECTIONAL. Batch-scaling") - print(" numbers are optimistic (overhead grows with B). Confirm the top pick on real A100.") - - -def section_capacity(m: Model, hw: HW): - print(f"\n{'='*74}\n KV CAPACITY — {m.name}/{hw.name} (max concurrent seqs before OOM)\n{'='*74}") - for L in [128, 512, 2048, 8192]: - n_bf16 = max_concurrent(m, hw, L, kv_dtype="bf16") - n_fp8 = max_concurrent(m, hw, L, kv_dtype="fp8") - print(f" L={L:<5} bf16 KV: {n_bf16:5d} seqs fp8 KV: {n_fp8:5d} seqs " - f"({kv_bytes(m,1,L,'bf16')/1e6:.0f}MB/seq bf16)") - print(" -> compare to the ~100-256 target: capacity-bound or not?") - - -def section_cross_hw(m: Model): - print(f"\n{'='*74}\n CROSS-HARDWARE — {m.name}, bf16, B=32 L=512 (why the bottleneck moves)\n{'='*74}") - for hw in HARDWARE.values(): - s = step(m, hw, 32, 512) - print(f" {hw.name:12s} ceiling {s['tok_s']:6.0f} tok/s [{s['regime']}-bound] " - f"B*={ridge_batch(m, hw, 512):.0f} (BW {hw.bw/1e12:.1f}TB/s, {hw.flops/1e12:.0f} TFLOP)") - - -if __name__ == "__main__": - m, hw = E4B, HARDWARE["A100-80GB"] - section_ceiling(m, hw) - section_breakdown(m, hw, B=32, L=160) - section_levers(m, hw, B=32, L=160) - section_capacity(m, hw) - section_cross_hw(m) - print() From 9708302272bf19bff59d054a6edb6782c6700daf Mon Sep 17 00:00:00 2001 From: Advay Monga Date: Sun, 11 Oct 2026 14:57:07 -0400 Subject: [PATCH 2/4] ceiling test runs as a script too (CI runs non-lab tests without pytest) Co-Authored-By: Claude Fable 5.1 --- tests/test_validity_ceiling.py | 39 ++++++++++++++++++++++------------ 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/tests/test_validity_ceiling.py b/tests/test_validity_ceiling.py index 7496748..0986e1a 100644 --- a/tests/test_validity_ceiling.py +++ b/tests/test_validity_ceiling.py @@ -3,12 +3,13 @@ from __future__ import annotations import json +import math +import sys from pathlib import Path -import pytest - -from validity import ceiling, peaks -from validity.ceiling import ModelSpec, Shape +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from validity import ceiling, peaks # noqa: E402 +from validity.ceiling import ModelSpec, Shape # noqa: E402 QWEN3_30B_A3B = {"num_hidden_layers": 48, "hidden_size": 2048, "num_attention_heads": 32, "num_key_value_heads": 4, "head_dim": 128, "vocab_size": 151936, "intermediate_size": 6144, "num_experts": 128, @@ -64,14 +65,16 @@ def test_regime_ceilings_by_objective(): def test_pct_of_and_ceiling_gain(): - assert ceiling.pct_of(0.004, 0.002, "lower") == pytest.approx(50) - assert ceiling.pct_of(50, 200, "higher") == pytest.approx(25) + assert math.isclose(ceiling.pct_of(0.004, 0.002, "lower"), 50) + assert math.isclose(ceiling.pct_of(50, 200, "higher"), 25) assert ceiling.pct_of(None, 1, "lower") is None and ceiling.pct_of(1, 0, "higher") is None - assert ceiling.ceiling_gain(0.4, 25) == pytest.approx(0.3) and ceiling.ceiling_gain(0.4, None) is None + assert math.isclose(ceiling.ceiling_gain(0.4, 25), 0.3) and ceiling.ceiling_gain(0.4, None) is None assert ceiling.ceiling_gain(0.4, 150) == 0 -def test_peaks_measured_file_wins_over_spec_and_driver_must_match(tmp_path): +def test_peaks_measured_file_wins_over_spec_and_driver_must_match(): + import tempfile + tmp_path = Path(tempfile.mkdtemp()) assert peaks.load("NVIDIA H200").source == "spec" and peaks.load("Radeon") is None p = peaks.Peaks("NVIDIA H200", 900e12, 4.5e12, 141e9, "measured", driver="580.1") out = peaks.write(p, tmp_path) @@ -82,10 +85,20 @@ def test_peaks_measured_file_wins_over_spec_and_driver_must_match(tmp_path): def test_measure_needs_cuda(): - pytest.importorskip("torch") - import torch + try: + import torch + except ImportError: + return if torch.cuda.is_available(): - pytest.skip("a GPU is present; the microbench is checked on the VM") - with pytest.raises(RuntimeError): + return # the microbench itself is checked on the VM + try: peaks.measure(0.1) - assert not (Path(peaks.PEAKS_DIR) / "radeon.json").exists() + except RuntimeError: + return + raise AssertionError("measure() must refuse without CUDA") + + +if __name__ == "__main__": + for name, fn in list(globals().items()): + if name.startswith("test_"): + fn(); print("ok", name) From f038b0fca2f76092e2788ae7971d6480fabce1dd Mon Sep 17 00:00:00 2001 From: Advay Monga Date: Mon, 12 Oct 2026 11:08:50 -0400 Subject: [PATCH 3/4] Ceilings: review fixes - Performance reference = the correctness reference minus its determinism switches (batch-invariant vLLM, deterministic SGLang); the stored reference records its launch and env. - Goodput ceiling pays for prefill compute and caps the batch by KV fitting in HBM beside the weights; sessions/s has no bound (it would need turns per session). Each basis states its assumptions. - dtype aliases (bfloat16, float16, ...); unknown dtypes are refused, not a KeyError. - The baseline gets its ceiling (seen base runs keep client rows); profile reads kernels.json at the blob root. - Calibration never raises: a failure is recorded instead of losing the measurement; ncu cells with separators or n/a and unit strings like Gbyte/second are parsed. - Invalid values say why; missing reasons are facts, not commands; seed recorded in the fingerprint, not required. - knowledge/peaks/nvidia-h200.json: measured on Verda (driver 580.126): 4.12 TB/s, 794 TFLOP/s bf16. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 8 +++- knowledge/peaks/nvidia-h200.json | 11 +++++ lab/calibrate.py | 70 +++++++++++++++++++++++++------- lab/evaltools.py | 9 ++-- lab/proftools.py | 5 ++- lab/tools.py | 8 ++-- lab/validate.py | 7 ++-- tests/test_lab_calibrate.py | 28 ++++++++++++- tests/test_validity_ceiling.py | 28 ++++++++++++- validity/ceiling.py | 41 +++++++++++++++---- 10 files changed, 176 insertions(+), 39 deletions(-) create mode 100644 knowledge/peaks/nvidia-h200.json diff --git a/CLAUDE.md b/CLAUDE.md index 9d6be99..acf3351 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -34,7 +34,8 @@ pytest/ruff config). Each folder with commands has its own `python -m`. `LAB_AGENT_PROVIDER` claude = Agent SDK CLI, openai = `agent_openai.py`: any OpenAI-compatible API with its own key, https public hosts only, `LAB_MODEL_PRICE` required, loop outside the jail, Bash jailed with no network), `calibrate.py` (every measured number with its context, decided 2026-10-11: `% of ceiling` from `validity/ceiling` - on measured peaks, `% of reference` from `knowledge/reference//.json` written by `lab.validate`, + on measured peaks, `% of reference` from `knowledge/reference//.json` written by `lab.validate` + with `perf_reference` (the correctness reference minus its determinism switches, which are slow by design), and for kernels their share of the window; a ceiling or reference is shown only under a matching config fingerprint (model, dtypes, max_model_len, tier, split, corpus version, GPU), else the field says why it is missing; bench, baseline, profile and kernel records carry a `calibrated` block, the brief shows the baseline's), @@ -109,7 +110,10 @@ pytest/ruff config). Each folder with commands has its own `python -m`. `fake_first`, `drop`, `cache` (streamed and not), `inflate_usage`); `python -m canaries --upstream URL --cheat NAME`. - `validity/` — `ceiling.py` (physical bounds from the model's HF config and the GPU's peaks: active/total params, bytes streamed per step with MoE experts-touched, decode and prefill floors, the largest batch under a TPOT limit, - per-objective regime ceilings; `pct_of`, `ceiling_gain` = Amdahl) + `peaks.py` (`python -m validity peaks`: HBM + per-objective regime ceilings: TPOT = decode step floor, TTFT = prefill floor at the p99 prompt, goodput = + 1 / (prefill compute + output tokens at the decode ceiling, batch capped by TPOT and by KV fitting in HBM); no + bound for speedups, cold start or sessions/s; assumptions (one token per step, uniform routing, declared dtypes) + are in each `basis`, so a value above 100% means one of them does not hold; `pct_of`, `ceiling_gain` = Amdahl) + `peaks.py` (`python -m validity peaks`: HBM bandwidth and bf16 GEMM measured once per GPU+driver into `knowledge/peaks/.json`, run by `vm-setup.sh`; spec sheet as the labelled fallback), `repeat_audit.py`: repeated-request audit (MLPerf TEST04-style), judges decode time per token, not TTFT, so prefix caching passes and replayed answers fail; `python -m validity repeat --url URL`. `tier_agreement.py`: `tier_agreement(changes)`: Spearman, Kendall tau-b, bootstrap CI, out-of-band disagreements; short tier usable if n ≥ 8, rho ≥ 0.8, CI low ≥ 0.5. `holdout.py`: `HoldoutGuard` (Thresholdout query budget, state persisted atomically under a file lock, deterministic from seed) and `SealedSplit` (hash manifest, one logged unseal per corpus version). Keep sigma small relative to the threshold. - `kernels/` — kernel equivalence: `check_kernel` (seen + held-out shapes, per-dtype tolerances, edge cases incl. non-contiguous and NaN/Inf, inputs unchanged, no aliasing, memoization, determinism) and `time_kernel` / `check_memoization` (sync, rotated buffers, L2 flush on CUDA). Validated on CPU, MPS and CUDA (H200, 2026-10-04): all tests pass; the timer resolves ~20 µs kernels at ~9% IQR, 0.2 ms at 1.5%. `make_inputs` gets a seeded CPU generator (generate on CPU, then `.to(device)`). diff --git a/knowledge/peaks/nvidia-h200.json b/knowledge/peaks/nvidia-h200.json new file mode 100644 index 0000000..8140f69 --- /dev/null +++ b/knowledge/peaks/nvidia-h200.json @@ -0,0 +1,11 @@ +{ + "gpu": "NVIDIA H200", + "bf16_flops": 793933120974445.2, + "bw_bytes_s": 4120900460756.014, + "mem_bytes": 150109880320.0, + "source": "measured", + "driver": "580.126.09", + "sm_clock_mhz": 1455, + "measured_at": "2026-10-12T05:51:02", + "method": "best of 3.0s: 512MiB bf16 copy; 8192^3 bf16 GEMM" +} diff --git a/lab/calibrate.py b/lab/calibrate.py index 22e3542..a4071cf 100644 --- a/lab/calibrate.py +++ b/lab/calibrate.py @@ -6,6 +6,7 @@ import csv import json +import re import statistics from pathlib import Path @@ -15,6 +16,8 @@ REFERENCE_DIR = target.ENV_ROOT / "knowledge" / "reference" FINGERPRINT_KEYS = ("model", "w_dtype", "kv_dtype", "max_model_len", "tier", "split", "corpus_version", "gpu") +DETERMINISM_ENV = ("VLLM_BATCH_INVARIANT",) +DETERMINISM_FLAGS = ("--enable-deterministic-inference",) # -- model config ----------------------------------------------------------------------- @@ -37,22 +40,37 @@ def model_spec(t: target.Target) -> ModelSpec | None: def dtypes(t: target.Target) -> tuple[str, str]: + """The weight and KV dtypes the target declares (`[model] dtype`, `kv_dtype`; bf16 when unset). The lab cannot see + what the agent's engine actually runs, so a quantized engine is measured against these declared ceilings.""" m = t.raw.get("model") or {} - return m.get("dtype", "bf16"), m.get("kv_dtype", m.get("dtype", "bf16")) + w = ceiling.dtype(m.get("dtype", "bf16")) + return w, ceiling.dtype(m.get("kv_dtype", w)) + + +def perf_reference(t: target.Target) -> target.Server | None: + """The reference server as a performance comparison: the correctness reference without its determinism switches + (batch-invariant vLLM, deterministic SGLang), which are slower by design.""" + import dataclasses + if t.reference is None: + return None + env = {k: v for k, v in t.reference.env.items() if k not in DETERMINISM_ENV} + launch = " ".join(w for w in t.reference.launch.split() if w not in DETERMINISM_FLAGS) + return dataclasses.replace(t.reference, env=env, launch=launch) # -- fingerprint -------------------------------------------------------------------------- -def fingerprint(t: target.Target, gpu: dict | None, tier: str, split: str) -> dict: +def fingerprint(t: target.Target, gpu: dict | None, tier: str, split: str, seed: int | None = None) -> dict: from regimes.workload import corpus_version w, kv = dtypes(t) return {"model": t.model, "w_dtype": w, "kv_dtype": kv, "max_model_len": t.max_model_len, "tier": tier, "split": split, "corpus_version": corpus_version(t.corpus_dir) if t.corpus_dir.is_dir() else None, "gpu": (gpu or {}).get("name"), "driver": (gpu or {}).get("driver_version"), - "sm_clock_mhz": (gpu or {}).get("clocks.sm")} + "sm_clock_mhz": (gpu or {}).get("clocks.sm"), "seed": seed} def same_config(a: dict, b: dict) -> list[str]: - """The fingerprint keys that differ (empty = comparable). Driver and clocks are reported, not required.""" + """The fingerprint keys that differ (empty = comparable). Driver, clocks and seed are reported, not required: + the seed picks a sample of the same split, it does not change what is measured.""" return [k for k in FINGERPRINT_KEYS if a.get(k) != b.get(k)] @@ -81,7 +99,9 @@ def write_reference(t: target.Target, regime: str, result: dict, fp: dict, rows: """One regime of the reference server, measured by the same clients, stored with its fingerprint.""" p = reference_path(t, regime) p.parent.mkdir(parents=True, exist_ok=True) - d = {"target": t.name, "server": "reference", "regime": regime, "objective": result["objective"], + srv = perf_reference(t) + d = {"target": t.name, "server": "reference", "launch": srv.launch if srv else None, "env": srv.env if srv else None, + "regime": regime, "objective": result["objective"], "value": result["value"] if result.get("valid") else None, "better": result["better"], "valid": result.get("valid"), "summary": result.get("summary"), "fingerprint": fp} sh = shape_from_rows(rows) if rows else None @@ -95,7 +115,7 @@ def reference_for(t: target.Target, regime: str, fp: dict) -> tuple[float | None """(value, why_missing): the reference's objective for this regime, only under the same fingerprint.""" p = reference_path(t, regime) if not p.exists(): - return None, "no reference measurement for this regime (lab.validate writes one)" + return None, "no reference measurement of this regime on record" d = json.loads(p.read_text()) diff = same_config(d.get("fingerprint") or {}, fp) if diff: @@ -105,9 +125,17 @@ def reference_for(t: target.Target, regime: str, fp: dict) -> tuple[float | None # -- regime results ---------------------------------------------------------------------------- def calibrate_results(results: list[dict], rows: list[dict], t: target.Target, gpu: dict | None, tier: str, - split: str) -> dict: - """{regime: {value, ceiling, pct_of_ceiling, reference, pct_of_reference, ...}} for a bench's results.""" - fp = fingerprint(t, gpu, tier, split) + split: str, seed: int | None = None) -> dict: + """{regime: {value, ceiling, pct_of_ceiling, reference, pct_of_reference, ...}} for a bench's results. Never + raises: a calibration that fails says why instead of costing the measurement it describes.""" + try: + return _calibrate_results(results, rows, t, gpu, tier, split, seed) + except Exception as e: # noqa: BLE001 + return {"error": f"calibration failed: {type(e).__name__}: {e}"} + + +def _calibrate_results(results, rows, t, gpu, tier, split, seed): + fp = fingerprint(t, gpu, tier, split, seed) hw = peaks.load(fp["gpu"], fp.get("driver")) if fp["gpu"] else None spec = model_spec(t) w, kv = dtypes(t) @@ -116,9 +144,11 @@ def calibrate_results(results: list[dict], rows: list[dict], t: target.Target, g name, obj = r["regime"], r["objective"] value = r["value"] if r.get("valid") else None row = {"objective": obj, "value": value, "better": r["better"], "fingerprint": fp} + if value is None: + row["value_missing"] = "the measurement was invalid: " + "; ".join(r.get("invalid_reasons") or []) shape = shape_from_rows([x for x in rows if x.get("regime") == name]) if hw is None: - row["ceiling"], row["ceiling_missing"] = None, "no peaks for this GPU (python -m validity peaks)" + row["ceiling"], row["ceiling_missing"] = None, "no measured or spec-sheet peaks for this GPU" elif spec is None: row["ceiling"], row["ceiling_missing"] = None, "model config unavailable" elif shape is None: @@ -167,16 +197,26 @@ def calibrate_ncu(path: Path, gpu: dict | None) -> dict | None: units, data = rows[0], rows[1:] hw = peaks.load(gpu["name"], gpu.get("driver_version")) if gpu and gpu.get("name") else None col = "dram__bytes.sum.per_second" - scale = {"byte/s": 1, "Kbyte/s": 1e3, "Mbyte/s": 1e6, "Gbyte/s": 1e9, "Tbyte/s": 1e12}.get(units.get(col, ""), 1) + unit = re.fullmatch(r"([KMGTP]?)byte/s(?:econd)?", units.get(col, "").strip()) + scale = {"": 1, "K": 1e3, "M": 1e6, "G": 1e9, "T": 1e12, "P": 1e15}[unit.group(1)] if unit else None out = [] for r in data: - bw = float(r[col]) * scale if r.get(col) else None - dur = r.get("gpu__time_duration.sum") - out.append({"name": r.get("Kernel Name"), "duration": float(dur) if dur else None, + v = _num(r.get(col)) + bw = v * scale if v is not None and scale is not None else None + dur = _num(r.get("gpu__time_duration.sum")) + out.append({"name": r.get("Kernel Name"), "duration": dur, "duration_unit": units.get("gpu__time_duration.sum"), "dram_bytes_s": bw, "pct_of_bandwidth_peak": bw / hw.bw_bytes_s * 100 if bw and hw else None}) return {"launches": out, "peaks_source": hw.source if hw else None, - "ceiling_missing": None if hw else "no peaks for this GPU (python -m validity peaks)"} + "ceiling_missing": None if hw else "no measured or spec-sheet peaks for this GPU"} + + +def _num(s) -> float | None: + """An ncu CSV cell as a number: thousands separators dropped, 'n/a' and blanks as None.""" + try: + return float(str(s).replace(",", "")) if s not in (None, "") else None + except ValueError: + return None def facts(t: target.Target, gpu: dict | None) -> dict: diff --git a/lab/evaltools.py b/lab/evaltools.py index a060575..634f668 100644 --- a/lab/evaltools.py +++ b/lab/evaltools.py @@ -165,7 +165,8 @@ def bench(self, args: dict) -> str: "artifacts": artifacts.store(work, self.s.ledger_root)} self._record("bench", "bench", args, result, snap, config={"split": "seen", "tier": tier, "seed": seed}) return _said("bench", result) - cal = calibrate.calibrate_results(r["results"], artifacts.read_rows(work), target.load(), r.get("gpu"), tier, "seen") + cal = calibrate.calibrate_results(r["results"], artifacts.read_rows(work), target.load(), r.get("gpu"), tier, + "seen", seed) result = {"verdict": "ok", "metrics": headline(r["results"]), "calibrated": cal, "regimes": r["results"], "tier": tier, "seconds": time.monotonic() - t0, "ready_s": r["ready_s"], "gpu": r.get("gpu"), "artifacts": artifacts.store(work, self.s.ledger_root)} @@ -317,7 +318,8 @@ def _base_measure(self, t: target.Target, names: list[str], split: str, tier: st work = self._work("base") try: # a contaminated base raises: never cached, every later submit reads it r = self._measure(tree, work, f"serve-base-{split}.log", names=missing, split=split, tier=tier, - passive=False) + passive=split == "seen") # held-out rows never leave the worker + base_rows = artifacts.read_rows(work) if split == "seen" else [] finally: shutil.rmtree(work, ignore_errors=True) have.update(headline(r["results"])) @@ -325,7 +327,8 @@ def _base_measure(self, t: target.Target, names: list[str], split: str, tier: st if split == "seen": # the base's context travels with the baseline record cal = cache.with_suffix(".calibrated.json") prior = json.loads(cal.read_text()) if cal.exists() else {} - prior.update(calibrate.calibrate_results(r["results"], [], t, r.get("gpu"), tier, split)) + prior.update(calibrate.calibrate_results(r["results"], base_rows, t, r.get("gpu"), tier, split, + self.seed)) cal.write_text(json.dumps(prior, indent=1)) return have diff --git a/lab/proftools.py b/lab/proftools.py index 5d3412b..bb2a3a4 100644 --- a/lab/proftools.py +++ b/lab/proftools.py @@ -135,7 +135,10 @@ def _instrumented(self, tool: str, args: dict) -> str: finally: shutil.rmtree(out, ignore_errors=True) from lab import calibrate - cal = calibrate.calibrate_ncu(self.s.ledger_root / blob / "metrics.csv", r.get("gpu")) if blob and tool == "kernel" else None + try: + cal = calibrate.calibrate_ncu(self.s.ledger_root / blob / "metrics.csv", r.get("gpu")) if blob and tool == "kernel" else None + except Exception as e: # noqa: BLE001 + cal = {"error": f"calibration failed: {type(e).__name__}: {e}"} result = {"calibrated": cal, "returncode": r["returncode"], "bundle": blob, "workspace_copy": visible, "instrument": r["instrument"], "argv": r["argv"], "seconds": time.monotonic() - t0, "output": r["output"]} self._record("profile", tool, args, result, snap) diff --git a/lab/tools.py b/lab/tools.py index 5dc5e6c..7cd1fdb 100644 --- a/lab/tools.py +++ b/lab/tools.py @@ -123,10 +123,12 @@ def profile(self, args: dict) -> str: blob, visible = self._keep(out / "files") if (out / "files").is_dir() else ("", "") finally: shutil.rmtree(out, ignore_errors=True) - bundles = sorted(p for p in (self.s.ledger_root / blob).glob("*") if p.is_dir()) if blob else [] + try: # the blob is the bundle itself (worker copies it to `files`) + cal = calibrate.calibrate_kernels(self.s.ledger_root / blob, r.get("gpu")) if blob else None + except Exception as e: # noqa: BLE001 never costs the measurement it describes + cal = {"error": f"calibration failed: {type(e).__name__}: {e}"} result = {"returncode": r["returncode"], "bundle": blob, "workspace_copy": visible, - "seconds": time.monotonic() - t0, "output": r["output"], - "calibrated": calibrate.calibrate_kernels(bundles[-1], r.get("gpu")) if bundles else None} + "seconds": time.monotonic() - t0, "output": r["output"], "calibrated": cal} self._record("profile", "profile", args, result, snap) if r["returncode"] != 0 or not blob: return f"profile failed ({r['returncode']}):\n{result['output']}" diff --git a/lab/validate.py b/lab/validate.py index 01a2c3c..16a2f76 100644 --- a/lab/validate.py +++ b/lab/validate.py @@ -95,12 +95,13 @@ def validate(t: target.Target, tier: str = "dev", noise_runs: int = 3, regimes: for n in names: report["noise"][n] = write_noise(n, values[n], t, {"tier": "short", "split": "seen"}) report["bands_measured"] = all(d["band_pct"] is not None for d in report["noise"].values()) if names and noise_runs else None - # 4. the reference engine under the same regimes, once: what every bench is shown next to + # 4. the reference engine under the same regimes, once, without its determinism switches: what every bench is shown next to if reference_regimes and names: - with serve.Served(target.ENV_ROOT, t.reference, log=log_dir / "serve-reference-regimes.log", jailed=False) as srv: + with serve.Served(target.ENV_ROOT, calibrate.perf_reference(t), log=log_dir / "serve-reference-regimes.log", + jailed=False) as srv: rows: list = [] ctx = suite.Ctx.from_target(t, srv.url, split="seen", tier="short", seed=0, rows=rows) - fp = calibrate.fingerprint(t, gpu.query(), "short", "seen") + fp = calibrate.fingerprint(t, gpu.query(), "short", "seen", 0) for n in names: k = len(rows) r = suite.REGIMES[n](ctx) diff --git a/tests/test_lab_calibrate.py b/tests/test_lab_calibrate.py index 735b35e..8215bab 100644 --- a/tests/test_lab_calibrate.py +++ b/tests/test_lab_calibrate.py @@ -33,7 +33,7 @@ def test_results_get_ceiling_pct_and_shape(t): assert cal["value"] == 0.02 and cal["ceiling"] < 0.02 and 0 < cal["pct_of_ceiling"] < 100 assert cal["bound"] == "memory" and cal["peaks_source"] == "spec" and cal["shape"]["prompt_tokens_mean"] == 600 assert cal["fingerprint"]["gpu"] == "NVIDIA H200" and cal["fingerprint"]["w_dtype"] == "bf16" - assert cal["reference"] is None and "no reference" in cal["reference_missing"] + assert cal["reference"] is None and "no reference measurement" in cal["reference_missing"] def test_missing_inputs_are_named_not_guessed(t, monkeypatch): @@ -84,3 +84,29 @@ def test_facts_for_the_brief(t): assert f["peaks"]["source"] == "spec" and f["model"]["active_params"] > 2.9e9 assert 0.001 < f["floors_s"]["decode_step_batch1_ctx1k"] < 0.003 assert calibrate.facts(t, None)["peaks"] is None + + +def test_perf_reference_drops_determinism_switches(t): + import dataclasses + srv = calibrate.perf_reference(t) + assert "VLLM_BATCH_INVARIANT" not in srv.env and srv.launch == t.reference.launch + sg = dataclasses.replace(t, reference=dataclasses.replace(t.reference, launch="python -m sglang.launch_server " + "--enable-deterministic-inference --port {port}")) + assert "--enable-deterministic-inference" not in calibrate.perf_reference(sg).launch + + +def test_calibration_never_raises_and_invalid_values_say_why(t, monkeypatch): + bad = calibrate.calibrate_results([{"regime": "x"}], [], t, GPU, "short", "seen") + assert "calibration failed" in bad["error"] + inv = calibrate.calibrate_results([{**results()[0], "valid": False, "invalid_reasons": ["3 silent drops"]}], rows(), + t, GPU, "short", "seen")["single_stream"] + assert inv["value"] is None and "3 silent drops" in inv["value_missing"] + + +def test_ncu_units_and_odd_cells(tmp_path): + p = tmp_path / "metrics.csv" + p.write_text('"ID","Kernel Name","gpu__time_duration.sum","dram__bytes.sum.per_second"\n' + '"","","usecond","Gbyte/second"\n"0","gemm","1,250.5","2,400"\n"1","gemm","n/a",""\n') + n = calibrate.calibrate_ncu(p, GPU)["launches"] + assert n[0]["duration"] == 1250.5 and round(n[0]["pct_of_bandwidth_peak"]) == 50 + assert n[1]["duration"] is None and n[1]["pct_of_bandwidth_peak"] is None diff --git a/tests/test_validity_ceiling.py b/tests/test_validity_ceiling.py index 0986e1a..b84374b 100644 --- a/tests/test_validity_ceiling.py +++ b/tests/test_validity_ceiling.py @@ -47,7 +47,8 @@ def test_prefill_floor_and_goodput_ceiling(): assert p["floor_s"] > ceiling.prefill_floor(m, H200, 512)["floor_s"] b = ceiling.max_batch_under(m, H200, 0.1, 1024) assert b > 64 and ceiling.decode_step_floor(m, H200, b, 1024)["floor_s"] <= 0.1 - assert ceiling.decode_step_floor(m, H200, b + 1, 1024)["floor_s"] > 0.1 + assert (b == ceiling.capacity_batch(m, H200, 1024) + or ceiling.decode_step_floor(m, H200, b + 1, 1024)["floor_s"] > 0.1) assert ceiling.max_batch_under(m, H200, 1e-6, 1024) == 0 and ceiling.tokens_per_s_ceiling(m, H200, 1e-6, 1024) == 0 @@ -75,7 +76,8 @@ def test_pct_of_and_ceiling_gain(): def test_peaks_measured_file_wins_over_spec_and_driver_must_match(): import tempfile tmp_path = Path(tempfile.mkdtemp()) - assert peaks.load("NVIDIA H200").source == "spec" and peaks.load("Radeon") is None + empty = Path(tempfile.mkdtemp()) + assert peaks.load("NVIDIA H200", peaks_dir=empty).source == "spec" and peaks.load("Radeon", peaks_dir=empty) is None p = peaks.Peaks("NVIDIA H200", 900e12, 4.5e12, 141e9, "measured", driver="580.1") out = peaks.write(p, tmp_path) assert out.name == "nvidia-h200.json" and json.loads(out.read_text())["bw_bytes_s"] == 4.5e12 @@ -98,6 +100,28 @@ def test_measure_needs_cuda(): raise AssertionError("measure() must refuse without CUDA") +def test_capacity_caps_the_batch_and_goodput_pays_for_prefill(): + m = ModelSpec.from_hf_config(QWEN3_30B_A3B) + cap = ceiling.capacity_batch(m, H200, 1024) + assert 0 < cap < 1200 # ~80 GB free for KV at ~100 KB/token + assert ceiling.max_batch_under(m, H200, 10.0, 1024) == cap # a loose TPOT limit stops at HBM, not at 65k + lim = {"ttft_s": 2.0, "tpot_s": 0.1} + short_out = ceiling.regime_ceiling("goodput_req_per_s", m, H200, Shape(8000, 9000, 50), lim) + long_out = ceiling.regime_ceiling("goodput_req_per_s", m, H200, Shape(8000, 9000, 2000), lim) + no_prefill = ceiling.tokens_per_s_ceiling(m, H200, 0.1, 8025) / 50 + assert short_out["ceiling"] < no_prefill and short_out["bound"] == "prefill" and long_out["bound"] == "decode" + assert ceiling.regime_ceiling("goodput_sessions_per_s", m, H200, Shape(600, 900, 200), lim) is None + + +def test_dtype_aliases(): + assert ceiling.dtype("bfloat16") == "bf16" and ceiling.dtype("FP8") == "fp8" + try: + ceiling.dtype("int3") + except ValueError: + return + raise AssertionError("unknown dtype accepted") + + if __name__ == "__main__": for name, fn in list(globals().items()): if name.startswith("test_"): diff --git a/validity/ceiling.py b/validity/ceiling.py index aea79f4..53cdedc 100644 --- a/validity/ceiling.py +++ b/validity/ceiling.py @@ -13,6 +13,15 @@ from validity.peaks import Peaks DTYPE_BYTES = {"bf16": 2, "fp16": 2, "fp8": 1, "int8": 1, "int4": 0.5} +ALIASES = {"bfloat16": "bf16", "float16": "fp16", "half": "fp16", "float8": "fp8", "fp8_e4m3": "fp8", "fp8_e5m2": "fp8"} + + +def dtype(name: str) -> str: + """A dtype name as DTYPE_BYTES knows it; ValueError for one it does not.""" + d = ALIASES.get(str(name).lower(), str(name).lower()) + if d not in DTYPE_BYTES: + raise ValueError(f"unknown dtype {name!r}; one of {sorted(DTYPE_BYTES)} or {sorted(ALIASES)}") + return d @dataclass(frozen=True) @@ -116,10 +125,17 @@ def prefill_floor(m: ModelSpec, hw: Peaks, prompt_tokens: int, w_dtype: str = "b "bound": "memory" if t_mem >= t_compute else "compute", "weight_bytes": w, "kv_bytes": kv, "flops": flops} +def capacity_batch(m: ModelSpec, hw: Peaks, context: float, w_dtype: str = "bf16", kv_dtype: str = "bf16") -> int: + """Sequences of `context` tokens whose KV fits in HBM beside all the weights (no activations or fragmentation: + an upper bound).""" + free = hw.mem_bytes - m.total_params * DTYPE_BYTES[w_dtype] + return max(0, int(free // (context * m.kv_bytes_per_token(kv_dtype)))) if context > 0 else 1 << 16 + + def max_batch_under(m: ModelSpec, hw: Peaks, tpot_limit_s: float, context: float, w_dtype: str = "bf16", kv_dtype: str = "bf16", cap: int = 1 << 16) -> int: - """Largest batch whose decode-step floor still meets the TPOT limit (the floor grows with batch).""" - lo, hi = 0, cap + """Largest batch whose decode-step floor still meets the TPOT limit and whose KV fits in HBM.""" + lo, hi = 0, min(cap, capacity_batch(m, hw, context, w_dtype, kv_dtype)) while lo < hi: mid = (lo + hi + 1) // 2 if decode_step_floor(m, hw, mid, context, w_dtype, kv_dtype)["floor_s"] <= tpot_limit_s: @@ -153,17 +169,24 @@ def regime_ceiling(objective: str, m: ModelSpec, hw: Peaks, shape: Shape, limits if objective == "tpot_p99_s": f = decode_step_floor(m, hw, max(1, round(shape.concurrency)), ctx, w_dtype, kv_dtype) return {"ceiling": f["floor_s"], "better": "lower", "bound": f["bound"], - "basis": f"decode step floor at batch {max(1, round(shape.concurrency))}, context {ctx:.0f}"} + "basis": f"decode step floor at batch {max(1, round(shape.concurrency))}, context {ctx:.0f}; one " + f"token per sequence per step, uniform expert routing, {w_dtype} weights, {kv_dtype} KV"} if objective == "ttft_p99_s": f = prefill_floor(m, hw, int(shape.prompt_tokens_p99), w_dtype, kv_dtype) return {"ceiling": f["floor_s"], "better": "lower", "bound": f["bound"], - "basis": f"prefill floor at the p99 prompt, {shape.prompt_tokens_p99:.0f} tokens"} - if objective in ("goodput_req_per_s", "good_req_per_s", "goodput_sessions_per_s"): + "basis": f"prefill floor at the p99 prompt, {shape.prompt_tokens_p99:.0f} tokens, {w_dtype} weights"} + if objective in ("goodput_req_per_s", "good_req_per_s"): + # each request costs its prefill FLOPs at peak (weights amortised over the batch) plus its output tokens at + # the decode ceiling; one GPU-second holds at most 1 / (that) requests tps = tokens_per_s_ceiling(m, hw, limits["tpot_s"], ctx, w_dtype, kv_dtype) - per = tps / shape.output_tokens_mean if shape.output_tokens_mean else 0.0 - return {"ceiling": per, "better": "higher", "bound": "memory", - "basis": f"decode tokens/s at the largest batch meeting TPOT <= {limits['tpot_s']} s, " - f"context {ctx:.0f}, / {shape.output_tokens_mean:.0f} output tokens per request"} + pf = prefill_floor(m, hw, int(shape.prompt_tokens_mean), w_dtype, kv_dtype) + per_req = pf["t_compute_s"] + (shape.output_tokens_mean / tps if tps else float("inf")) + return {"ceiling": 1 / per_req if per_req and per_req != float("inf") else 0.0, "better": "higher", + "bound": "decode" if shape.output_tokens_mean / max(tps, 1e-12) > pf["t_compute_s"] else "prefill", + "basis": f"1 / (prefill compute of {shape.prompt_tokens_mean:.0f} prompt tokens + " + f"{shape.output_tokens_mean:.0f} output tokens at the decode ceiling: the largest batch " + f"meeting TPOT <= {limits['tpot_s']} s whose KV fits in HBM, context {ctx:.0f}); " + f"one token per sequence per step, uniform expert routing, {w_dtype} weights, {kv_dtype} KV"} return None From ad612e54c32738a710db7f7414224433e7d70bfc Mon Sep 17 00:00:00 2001 From: Advay Monga Date: Tue, 13 Oct 2026 15:00:13 -0400 Subject: [PATCH 4/4] Ceilings: checked on an H200; prefill bounded by the server's uncached tokens - Goodput charges prefill only for the prompt tokens the server reports it did not take from its prefix cache (rows record usage.prompt_tokens_details.cached_tokens); with no report the bound is decode alone. On the H200, vLLM's prefix cache made the prefill-inclusive bound read 147-471%; now every floor sits below every measurement. - Peaks are stored per GPU and driver; both measured H200 drivers kept. - knowledge/reference/inference-server/: plain vLLM under single_stream, saturated, long_prompt_short_output. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 9 ++- ...-h200.json => nvidia-h200-580-126-09.json} | 0 knowledge/peaks/nvidia-h200-580-173-02.json | 11 ++++ .../long_prompt_short_output.json | 56 +++++++++++++++++++ .../reference/inference-server/saturated.json | 56 +++++++++++++++++++ .../inference-server/single_stream.json | 56 +++++++++++++++++++ lab/calibrate.py | 5 +- regimes/client.py | 3 + tests/test_lab_calibrate.py | 10 +++- tests/test_validity_ceiling.py | 17 +++++- validity/ceiling.py | 15 +++-- validity/peaks.py | 23 +++++--- 12 files changed, 241 insertions(+), 20 deletions(-) rename knowledge/peaks/{nvidia-h200.json => nvidia-h200-580-126-09.json} (100%) create mode 100644 knowledge/peaks/nvidia-h200-580-173-02.json create mode 100644 knowledge/reference/inference-server/long_prompt_short_output.json create mode 100644 knowledge/reference/inference-server/saturated.json create mode 100644 knowledge/reference/inference-server/single_stream.json diff --git a/CLAUDE.md b/CLAUDE.md index acf3351..ffb4790 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -245,7 +245,12 @@ a first `python -m lab.session --task ...` dry run with a tiny budget, which wai `engine/gpu-validated`) were judged by the old gate: re-judge with `python -m correctness verdict` on their saved results before deciding anything. lab `session.py --run` re-resolves `base` from HEAD rather than the run's original base (pre-existing; resuming after the engine moved would misaudit); -`validity/tier_agreement.py` needs real short/full runs; the ceilings need one GPU pass (measured peaks, floors -below every measurement, `lab.validate`'s reference regimes) +`validity/tier_agreement.py` needs real short/full runs. Ceilings checked on an H200 (Nebius, driver 580.173, +2026-10-13; peaks 4.14 TB/s, 794 TFLOP/s bf16): every floor below every measurement once goodput charges only the +prefill the server reports as uncached (vLLM's prefix cache made a prefill-inclusive bound read 147-471%). Plain vLLM: +single_stream TPOT p99 4.1 ms (37% of ceiling), saturated 5.8 req/s (19%), long prompt 68 req/s (94%); the engine: +21 ms (7%), 0.097 req/s (0.5%), long prompt invalid (111 requests without a response). Stored in +`knowledge/reference/inference-server/`. Finding for the corpus: long_prompt_short_output's window repeats each +prompt ~5x (23,185 requests over 4,794 prompts), which rewards prefix caching. target. The agent's tool set (e.g. a knowledge tool) is the user's design: ask first. README stays a few one-line feature bullets, not a dev doc. diff --git a/knowledge/peaks/nvidia-h200.json b/knowledge/peaks/nvidia-h200-580-126-09.json similarity index 100% rename from knowledge/peaks/nvidia-h200.json rename to knowledge/peaks/nvidia-h200-580-126-09.json diff --git a/knowledge/peaks/nvidia-h200-580-173-02.json b/knowledge/peaks/nvidia-h200-580-173-02.json new file mode 100644 index 0000000..0c150b9 --- /dev/null +++ b/knowledge/peaks/nvidia-h200-580-173-02.json @@ -0,0 +1,11 @@ +{ + "gpu": "NVIDIA H200", + "bf16_flops": 794182576833900.2, + "bw_bytes_s": 4135566038904.868, + "mem_bytes": 150109880320.0, + "source": "measured", + "driver": "580.173.02", + "sm_clock_mhz": 1455, + "measured_at": "2026-10-13T17:39:50", + "method": "best of 3.0s: 512MiB bf16 copy; 8192^3 bf16 GEMM" +} diff --git a/knowledge/reference/inference-server/long_prompt_short_output.json b/knowledge/reference/inference-server/long_prompt_short_output.json new file mode 100644 index 0000000..fdf17f1 --- /dev/null +++ b/knowledge/reference/inference-server/long_prompt_short_output.json @@ -0,0 +1,56 @@ +{ + "target": "inference-server", + "server": "reference", + "launch": "vllm serve Qwen/Qwen3-30B-A3B --dtype bfloat16 --served-model-name Qwen/Qwen3-30B-A3B --max-model-len 32768 --port {port}", + "env": {}, + "regime": "long_prompt_short_output", + "objective": "goodput_req_per_s", + "value": 68.0, + "better": "higher", + "valid": true, + "summary": { + "n": 4089, + "ok": 4089, + "error": 0, + "silent_drop": 0, + "truncated": 0, + "crash": 0, + "attainment": 1.0, + "meets_limits": true, + "ttft_p50_s": 0.22769945079134501, + "ttft_p99_s": 0.40794480072750794, + "tpot_p50_s": 0.055327961015874526, + "tpot_p99_s": 0.06561631411111396, + "output_tokens": 261696, + "elapsed_s": 60.0, + "req_per_s": 68.15, + "good_req_per_s": 68.15, + "output_tok_per_s": 4361.6, + "client_lag_p99_s": 0.00025313885493538635, + "limits": { + "ttft_s": 2.0, + "tpot_s": 0.1 + }, + "valid": true, + "invalid_reasons": [] + }, + "fingerprint": { + "model": "Qwen/Qwen3-30B-A3B", + "w_dtype": "bf16", + "kv_dtype": "bf16", + "max_model_len": 32768, + "tier": "short", + "split": "seen", + "corpus_version": "81739c469a304fac4a486917dee281ae8182881ab64128030b3c6ce1a428c613", + "gpu": "NVIDIA H200", + "driver": "580.173.02", + "sm_clock_mhz": "1980", + "seed": 0 + }, + "shape": { + "prompt_tokens_mean": 5374.133491481562, + "prompt_tokens_p99": 6748.0, + "output_tokens_mean": 64.0, + "concurrency": 2060.467756318616 + } +} diff --git a/knowledge/reference/inference-server/saturated.json b/knowledge/reference/inference-server/saturated.json new file mode 100644 index 0000000..97d3d92 --- /dev/null +++ b/knowledge/reference/inference-server/saturated.json @@ -0,0 +1,56 @@ +{ + "target": "inference-server", + "server": "reference", + "launch": "vllm serve Qwen/Qwen3-30B-A3B --dtype bfloat16 --served-model-name Qwen/Qwen3-30B-A3B --max-model-len 32768 --port {port}", + "env": {}, + "regime": "saturated", + "objective": "goodput_req_per_s", + "value": 5.819394739012707, + "better": "higher", + "valid": true, + "summary": { + "n": 528, + "ok": 528, + "error": 0, + "silent_drop": 0, + "truncated": 0, + "crash": 0, + "attainment": 1.0, + "meets_limits": true, + "ttft_p50_s": 0.08344380399989859, + "ttft_p99_s": 1.2710328759999356, + "tpot_p50_s": 0.023719139036450395, + "tpot_p99_s": 0.030884681000012886, + "output_tokens": 402714, + "elapsed_s": 90.73108522099983, + "req_per_s": 5.819394739012707, + "good_req_per_s": 5.819394739012707, + "output_tok_per_s": 4438.544948724931, + "client_lag_p99_s": 0.00012458699984563282, + "limits": { + "ttft_s": 2.0, + "tpot_s": 0.1 + }, + "valid": true, + "invalid_reasons": [] + }, + "fingerprint": { + "model": "Qwen/Qwen3-30B-A3B", + "w_dtype": "bf16", + "kv_dtype": "bf16", + "max_model_len": 32768, + "tier": "short", + "split": "seen", + "corpus_version": "81739c469a304fac4a486917dee281ae8182881ab64128030b3c6ce1a428c613", + "gpu": "NVIDIA H200", + "driver": "580.173.02", + "sm_clock_mhz": "1980", + "seed": 0 + }, + "shape": { + "prompt_tokens_mean": 710.1108091156178, + "prompt_tokens_p99": 5613.0, + "output_tokens_mean": 761.5467279949822, + "concurrency": 1062.8178219140848 + } +} diff --git a/knowledge/reference/inference-server/single_stream.json b/knowledge/reference/inference-server/single_stream.json new file mode 100644 index 0000000..316cfa8 --- /dev/null +++ b/knowledge/reference/inference-server/single_stream.json @@ -0,0 +1,56 @@ +{ + "target": "inference-server", + "server": "reference", + "launch": "vllm serve Qwen/Qwen3-30B-A3B --dtype bfloat16 --served-model-name Qwen/Qwen3-30B-A3B --max-model-len 32768 --port {port}", + "env": {}, + "regime": "single_stream", + "objective": "tpot_p99_s", + "value": 0.00414084390815831, + "better": "lower", + "valid": true, + "summary": { + "n": 20, + "ok": 20, + "error": 0, + "silent_drop": 0, + "truncated": 0, + "crash": 0, + "attainment": 1.0, + "meets_limits": true, + "ttft_p50_s": 0.026337838000017655, + "ttft_p99_s": 0.07710730499991314, + "tpot_p50_s": 0.00402400701940755, + "tpot_p99_s": 0.00414084390815831, + "output_tokens": 15988, + "elapsed_s": 65.61843898899997, + "req_per_s": 0.3047923770840194, + "good_req_per_s": 0.3047923770840194, + "output_tok_per_s": 243.6510262409651, + "client_lag_p99_s": 0.00013822099992921721, + "limits": { + "ttft_s": 0.5, + "tpot_s": 0.03 + }, + "valid": true, + "invalid_reasons": [] + }, + "fingerprint": { + "model": "Qwen/Qwen3-30B-A3B", + "w_dtype": "bf16", + "kv_dtype": "bf16", + "max_model_len": 32768, + "tier": "short", + "split": "seen", + "corpus_version": "81739c469a304fac4a486917dee281ae8182881ab64128030b3c6ce1a428c613", + "gpu": "NVIDIA H200", + "driver": "580.173.02", + "sm_clock_mhz": "1980", + "seed": 0 + }, + "shape": { + "prompt_tokens_mean": 1377.65, + "prompt_tokens_p99": 3406.0, + "output_tokens_mean": 799.4, + "concurrency": 1.0 + } +} diff --git a/lab/calibrate.py b/lab/calibrate.py index a4071cf..1fdd2ed 100644 --- a/lab/calibrate.py +++ b/lab/calibrate.py @@ -87,7 +87,10 @@ def shape_from_rows(rows: list[dict]) -> Shape | None: if not prompts: return None p99 = sorted(prompts)[min(len(prompts) - 1, int(0.99 * len(prompts)))] - return Shape(statistics.fmean(prompts), float(p99), statistics.fmean(outs), max(1.0, conc)) + reported = [r["prompt_tokens"] - r["cached_tokens"] for r in ok + if r.get("prompt_tokens") and r.get("cached_tokens") is not None] + uncached = statistics.fmean(reported) if len(reported) == len(ok) else None + return Shape(statistics.fmean(prompts), float(p99), statistics.fmean(outs), max(1.0, conc), uncached) # -- reference numbers ----------------------------------------------------------------------- diff --git a/regimes/client.py b/regimes/client.py index 293e636..d7ad81c 100644 --- a/regimes/client.py +++ b/regimes/client.py @@ -34,6 +34,7 @@ class Row: e2e_s: float | None = None n_tokens: int = 0 prompt_tokens: int | None = None # from the server's usage + cached_tokens: int | None = None # prompt tokens the server served from its prefix cache, when it reports them tpot_s: float | None = None http_status: int | None = None finish_reason: str | None = None @@ -98,6 +99,8 @@ def send(url: str, model: str, req: Request, scheduled: float, clock, timeout: f row.text, row.finish_reason, row.e2e_s = "".join(parts), finish, end - scheduled if usage and usage.get("prompt_tokens") is not None: row.prompt_tokens = int(usage["prompt_tokens"]) + cached = (usage.get("prompt_tokens_details") or {}).get("cached_tokens") + row.cached_tokens = int(cached) if cached is not None else None if usage and usage.get("completion_tokens") is not None: row.n_tokens = int(usage["completion_tokens"]) else: diff --git a/tests/test_lab_calibrate.py b/tests/test_lab_calibrate.py index 8215bab..0b150b1 100644 --- a/tests/test_lab_calibrate.py +++ b/tests/test_lab_calibrate.py @@ -23,8 +23,9 @@ def results(regime="single_stream", objective="tpot_p99_s", value=0.02, better=" @pytest.fixture -def t(monkeypatch): +def t(monkeypatch, tmp_path): monkeypatch.setattr(calibrate, "model_config", lambda model: QWEN3_30B_A3B) + monkeypatch.setattr(calibrate, "REFERENCE_DIR", tmp_path / "no-reference") # the committed ones are real return target.load() @@ -110,3 +111,10 @@ def test_ncu_units_and_odd_cells(tmp_path): n = calibrate.calibrate_ncu(p, GPU)["launches"] assert n[0]["duration"] == 1250.5 and round(n[0]["pct_of_bandwidth_peak"]) == 50 assert n[1]["duration"] is None and n[1]["pct_of_bandwidth_peak"] is None + + +def test_uncached_prompt_tokens_come_from_the_server_or_not_at_all(): + full = [{**r, "cached_tokens": 450} for r in rows()] + assert calibrate.shape_from_rows(full).uncached_prompt_tokens_mean == 150 + partial = full[:5] + rows()[5:] + assert calibrate.shape_from_rows(partial).uncached_prompt_tokens_mean is None diff --git a/tests/test_validity_ceiling.py b/tests/test_validity_ceiling.py index b84374b..0d24b84 100644 --- a/tests/test_validity_ceiling.py +++ b/tests/test_validity_ceiling.py @@ -80,7 +80,7 @@ def test_peaks_measured_file_wins_over_spec_and_driver_must_match(): assert peaks.load("NVIDIA H200", peaks_dir=empty).source == "spec" and peaks.load("Radeon", peaks_dir=empty) is None p = peaks.Peaks("NVIDIA H200", 900e12, 4.5e12, 141e9, "measured", driver="580.1") out = peaks.write(p, tmp_path) - assert out.name == "nvidia-h200.json" and json.loads(out.read_text())["bw_bytes_s"] == 4.5e12 + assert out.name == "nvidia-h200-580-1.json" and json.loads(out.read_text())["bw_bytes_s"] == 4.5e12 assert peaks.load("NVIDIA H200", "580.1", tmp_path).source == "measured" assert peaks.load("NVIDIA H200", None, tmp_path).source == "measured" assert peaks.load("NVIDIA H200", "581.0", tmp_path).source == "spec" # another driver: back to the sheet @@ -106,13 +106,24 @@ def test_capacity_caps_the_batch_and_goodput_pays_for_prefill(): assert 0 < cap < 1200 # ~80 GB free for KV at ~100 KB/token assert ceiling.max_batch_under(m, H200, 10.0, 1024) == cap # a loose TPOT limit stops at HBM, not at 65k lim = {"ttft_s": 2.0, "tpot_s": 0.1} - short_out = ceiling.regime_ceiling("goodput_req_per_s", m, H200, Shape(8000, 9000, 50), lim) - long_out = ceiling.regime_ceiling("goodput_req_per_s", m, H200, Shape(8000, 9000, 2000), lim) + short_out = ceiling.regime_ceiling("goodput_req_per_s", m, H200, Shape(8000, 9000, 50, uncached_prompt_tokens_mean=8000), lim) + long_out = ceiling.regime_ceiling("goodput_req_per_s", m, H200, Shape(8000, 9000, 2000, uncached_prompt_tokens_mean=8000), lim) no_prefill = ceiling.tokens_per_s_ceiling(m, H200, 0.1, 8025) / 50 assert short_out["ceiling"] < no_prefill and short_out["bound"] == "prefill" and long_out["bound"] == "decode" + unknown = ceiling.regime_ceiling("goodput_req_per_s", m, H200, Shape(8000, 9000, 50), lim) + assert unknown["ceiling"] == pytest_approx(no_prefill) and "may have been cached" in unknown["basis"] + half = ceiling.regime_ceiling("goodput_req_per_s", m, H200, Shape(8000, 9000, 50, uncached_prompt_tokens_mean=4000), lim) + assert short_out["ceiling"] < half["ceiling"] < unknown["ceiling"] assert ceiling.regime_ceiling("goodput_sessions_per_s", m, H200, Shape(600, 900, 200), lim) is None +def pytest_approx(x, rel=1e-9): + class A: + def __eq__(self, y): + return math.isclose(x, y, rel_tol=rel) + return A() + + def test_dtype_aliases(): assert ceiling.dtype("bfloat16") == "bf16" and ceiling.dtype("FP8") == "fp8" try: diff --git a/validity/ceiling.py b/validity/ceiling.py index 53cdedc..f91b2b1 100644 --- a/validity/ceiling.py +++ b/validity/ceiling.py @@ -160,6 +160,7 @@ class Shape: prompt_tokens_p99: float output_tokens_mean: float concurrency: float = 1.0 # mean sequences in flight + uncached_prompt_tokens_mean: float | None = None # what the server reported it had to prefill; None = unknown def regime_ceiling(objective: str, m: ModelSpec, hw: Peaks, shape: Shape, limits: dict, @@ -178,12 +179,18 @@ def regime_ceiling(objective: str, m: ModelSpec, hw: Peaks, shape: Shape, limits if objective in ("goodput_req_per_s", "good_req_per_s"): # each request costs its prefill FLOPs at peak (weights amortised over the batch) plus its output tokens at # the decode ceiling; one GPU-second holds at most 1 / (that) requests + # a prefix cache can skip any prompt token, so only what the server reports as uncached counts; when it + # reports nothing, prefill is left out and the bound is decode alone (still a true bound, a looser one) tps = tokens_per_s_ceiling(m, hw, limits["tpot_s"], ctx, w_dtype, kv_dtype) - pf = prefill_floor(m, hw, int(shape.prompt_tokens_mean), w_dtype, kv_dtype) - per_req = pf["t_compute_s"] + (shape.output_tokens_mean / tps if tps else float("inf")) + uncached = shape.uncached_prompt_tokens_mean or 0.0 + prefill_s = prefill_floor(m, hw, int(uncached), w_dtype, kv_dtype)["t_compute_s"] if uncached >= 1 else 0.0 + per_req = prefill_s + (shape.output_tokens_mean / tps if tps else float("inf")) + known = shape.uncached_prompt_tokens_mean is not None return {"ceiling": 1 / per_req if per_req and per_req != float("inf") else 0.0, "better": "higher", - "bound": "decode" if shape.output_tokens_mean / max(tps, 1e-12) > pf["t_compute_s"] else "prefill", - "basis": f"1 / (prefill compute of {shape.prompt_tokens_mean:.0f} prompt tokens + " + "bound": "decode" if shape.output_tokens_mean / max(tps, 1e-12) > prefill_s else "prefill", + "basis": (f"1 / (prefill compute of the {uncached:.0f} uncached prompt tokens per request the server " + f"reported" if known else "1 / (no prefill: the server reported no prefix-cache counts, so " + "every prompt token may have been cached") + " + " f"{shape.output_tokens_mean:.0f} output tokens at the decode ceiling: the largest batch " f"meeting TPOT <= {limits['tpot_s']} s whose KV fits in HBM, context {ctx:.0f}); " f"one token per sequence per step, uniform expert routing, {w_dtype} weights, {kv_dtype} KV"} diff --git a/validity/peaks.py b/validity/peaks.py index a72a3ee..c355919 100644 --- a/validity/peaks.py +++ b/validity/peaks.py @@ -1,5 +1,5 @@ """Hardware peaks the ceilings are computed against: HBM bandwidth and bf16 tensor-core throughput, measured once -per GPU + driver (`python -m validity peaks`, run by lab/vm-setup.sh) into knowledge/peaks/.json. Spec-sheet +per GPU + driver (`python -m validity peaks`, run by lab/vm-setup.sh) into knowledge/peaks/-.json. Spec-sheet numbers are the fallback, labelled `source: spec`; a ceiling from them says so.""" from __future__ import annotations @@ -44,14 +44,19 @@ def spec_for(gpu: str) -> Peaks | None: return None +def path_for(gpu: str, driver: str | None, peaks_dir: Path | None = None) -> Path: + return (peaks_dir or PEAKS_DIR) / (f"{slug(gpu)}-{slug(driver)}.json" if driver else f"{slug(gpu)}.json") + + def load(gpu: str, driver: str | None = None, peaks_dir: Path | None = None) -> Peaks | None: - """Measured peaks for this GPU (and driver, when known); else the spec sheet; else None.""" - p = (peaks_dir or PEAKS_DIR) / f"{slug(gpu)}.json" - if p.exists(): - d = json.loads(p.read_text()) - if driver is None or d.get("driver") in (None, driver): - return Peaks(**d) - return spec_for(gpu) + """Measured peaks for this GPU and driver; without a driver, the latest measurement of the GPU; else the spec + sheet; else None. A measurement on another driver is not used.""" + d = peaks_dir or PEAKS_DIR + if driver: + p = path_for(gpu, driver, d) + return Peaks(**json.loads(p.read_text())) if p.exists() else spec_for(gpu) + found = sorted(d.glob(f"{slug(gpu)}*.json"), key=lambda q: json.loads(q.read_text()).get("measured_at") or "") + return Peaks(**json.loads(found[-1].read_text())) if found else spec_for(gpu) def measure(seconds: float = 3.0) -> Peaks: @@ -99,6 +104,6 @@ def measure(seconds: float = 3.0) -> Peaks: def write(p: Peaks, peaks_dir: Path | None = None) -> Path: d = peaks_dir or PEAKS_DIR d.mkdir(parents=True, exist_ok=True) - out = d / f"{slug(p.gpu)}.json" + out = path_for(p.gpu, p.driver, d) out.write_text(json.dumps(asdict(p), indent=1) + "\n") return out