url/colibri
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #360 from woolcoxm/test/efficiency-suite

Browse Source 
tests: efficiency suite — tiny-model regression gates + full-model optimization dossier (#359)
This commit is contained in:
Vincenzo Fornaro
2026-07-20 17:09:21 +02:00
committed by GitHub
parent 1765ed80ac cf126cbcf9
commit af23f314fd
5 changed files with 1151 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
c/Makefile
+23
View File
@@ -363,6 +363,29 @@ test-python:
test: test-c test-python
# --- Efficiency / regression suite (issue: "test the program for inefficiencies") ---
# The tiny-model assertions live in test_inefficiency.py and run as part of
# test-python (they're discovered by the test_*.py glob). These targets are
# convenience entry points; the opt-in full-model report is NEVER in `make test`.
#
# make efficiency tiny-model asserted regression tests (CPU; part of test-python)
# make efficiency-cuda the CUDA-path tests (requires a CUDA build — see below)
# make efficiency-report opt-in full-model 🟢/🔴 diagnostic, never fails CI
#
# CUDA build (Windows): the CUDA tests need a host built with -DCOLI_CUDA plus
# the runtime DLL. Do this FIRST — note CUDA_DLL=1 on BOTH the host and the
# rule below, or `make glm.exe` will rebuild a CPU-only host and overwrite it:
# make clean && make glm.exe CUDA_DLL=1 && make cuda-dll
# The tests auto-skip with a clear message if the host is CPU-only.
efficiency: test-python
$(PYTHON) -m unittest tests.test_inefficiency -v
efficiency-cuda:
$(PYTHON) -m unittest tests.test_inefficiency.TinyCudaEfficiencyTest -v
efficiency-report:
$(PYTHON) tests/test_efficiency_report.py
# Local validation: one portable CPU build and dependency-free tests.
check:
$(MAKE) clean
c/tests/README_efficiency.md
+98
View File
@@ -0,0 +1,98 @@
# Efficiency suite — regression tests + optimization dossier
Two layers:
1. **`test_inefficiency.py`** — tiny-model *asserted* regression tests. Fast
(~0.15s/run), gate CI, catch breakage. Run as part of `make test`.
2. **`test_efficiency_report.py`** — an *opt-in optimization dossier* for a real
model. Runs every instrumentation flag, prints a 9-section report answering
*what is doing what, when, with what, is it inefficient, how to improve*.
Never fails CI (it's a report, not a gate).
## The dossier (what you run when optimizing)
```bash
# CPU-only (safe, fast to validate):
COLI_EFFICIENCY_MODEL=../glm52_i4_g64 make efficiency-report
# CUDA (dense + expert tiers — needs a CUDA build, see below):
COLI_EFFICIENCY_MODEL=../glm52_i4_g64 COLI_EFFICIENCY_CUDA=1 make efficiency-report
```
It turns ON every observability flag the engine supports — `PROF=1`,
`COLI_CUDA_PROFILE=1`, `CACHE_ROUTE=1` (auto-unlocks `route_agree`/`route_kl`),
`DISK_SPLIT=1`, `LOOKA=1` — so nothing the engine can tell you is left dark.
None of these change the computed output; they only add telemetry.
The 9 sections, and the question each answers:
| § | section | answers |
|---|---|---|
| 1 | PROVENANCE | what is running, on what CPU/backend, with what effective config |
| 2 | THROUGHPUT | tok/s + forward-latency p50/p90/p99/max (is the tail healthy?) |
| 3 | WHERE TIME GOES | the 5 PROFILE phases as % of decode + absolute seconds + verdict |
| 3a | ATTENTION BREAKDOWN | attention split into projection/RoPE, score-softmax-value, output |
| 4 | EXPERT CACHE | hit %, experts-loaded/token vs baseline topk |
| 5 | DISK I/O | GB fetched, MB/token, GB/s, read-service vs felt-wait, phase split |
| 5a | DISK-LOAD SPLIT | loads by decode phase (draft/absorb/verify) + MTP-vs-main bytes |
| 6 | ROUTING QUALITY | route_agree %, route_kl, cache swaps |
| 6a | ROUTING PREDICTABILITY | LOOKAHEAD recall per predictor (which prefetch wins) |
| 7 | SPECULATION | tokens/forward, MTP acceptance % |
| 8 | GPU TIERS | resident tensors, expert tier (count/GB/calls), H2D/kernel/D2H ms |
Every line that crosses an advisory threshold is marked `[FLAG]` with the
concrete lever to pull (raise RAM_GB, add PIN_GB, try DIRECT=1, lower CTX, …),
and all flags repeat in a summary at the end.
## Tunable thresholds
The `IS IT INEFFICIENT?` lines are advisory constants at the top of
`test_efficiency_report.py`:
| constant | default | meaning |
|---|---|---|
| `DISK_WAIT_DOMINANT` | 0.40 | >40% decode waiting on expert reads → I/O-bound |
| `LOW_HIT_RATE` | 0.30 | <30% cache hit → thrashing |
| `LOW_ROUTE_AGREE` | 0.80 | <80% routing overlap → prefetch guessing wrong |
| `HIGH_TAIL_RATIO` | 3.0 | p99 > 3× p50 → decode stalls |
| `LOW_MTP_ACCEPT` | 0.20 | <20% MTP acceptance → draft decoder is dead weight |
The tiny-model asserted floors live in `tools/efficiency.py` (`TINY_TOK_S_FLOOR`,
`MAX_DISK_WAIT_SHARE`, `MIN_CPU_CUDA_AGREEMENT`).
## The regression tests (what gates CI)
`test_inefficiency.py` runs on the bundled `glm_tiny` model and asserts:
- telemetry parses (no format drift)
- tiny tok/s ≥ floor (throughput regression)
- PROFILE phases present and non-negative (accounting sanity)
- disk-wait not dominant on a resident model (I/O-path regression)
- CPU determinism (two greedy runs agree)
- **CUDA** (skip unless CUDA built): init path, dense uploads VRAM, CPU-vs-CUDA
argmax agreement ≥ 70% (kernel-correctness guard)
```bash
make efficiency # tiny CPU tests
make efficiency-cuda # tiny CUDA tests (needs CUDA build)
```
## CUDA build prerequisite
The default `make glm.exe` builds **without** CUDA. The CUDA tests and the CUDA
dossier need a host built with `-DCOLI_CUDA` plus the runtime DLL:
```bash
make clean && make glm.exe CUDA_DLL=1 && make cuda-dll
```
`make efficiency-cuda` auto-skips with a clear message if the host is CPU-only
(it scans the binary for the "CPU-only" marker the engine embeds).
## Files
- `tools/efficiency.py` — shared harness: `parse_run()` (captures every
telemetry signal), `run_engine()`, thresholds. Reuses `PROFILE_RE`/`SPEED_RE`
from `tools/benchmark_cuda_fixture.py`.
- `tests/test_inefficiency.py` — tiny-model asserted tests (CPU + CUDA).
- `tests/test_efficiency_report.py` — the opt-in optimization dossier.
c/tests/test_efficiency_report.py
+315
View File
@@ -0,0 +1,315 @@
#!/usr/bin/env python3
"""Exhaustive optimization dossier for a colibri engine run.
This is NOT a pass/fail test. It runs the engine with every instrumentation flag
on (PROF, COLI_CUDA_PROFILE, CACHE_ROUTE, DISK_SPLIT, LOOKA) and prints a section-
by-section report answering, for each subsystem:
WHAT is doing it — which phase/kernel/tier
WHEN it is doing it — how much of decode wall-time it owns
WITH WHAT — the config/weights/tier it used
IS IT INEFFICIENT? — a verdict, with the threshold
HOW TO IMPROVE — the concrete knob, named
Activation (opt-in only — NOT in `make test`):
COLI_EFFICIENCY_MODEL=<model_dir> python tests/test_efficiency_report.py
Optional env:
COLI_EFFICIENCY_CUDA=1 also exercise the CUDA dense/expert tiers
COLI_EFFICIENCY_NGEN=N decode tokens (default 24)
COLI_EFFICIENCY_RAM_GB=N RAM budget (default 28)
COLI_EFFICIENCY_VRAM_GB=N CUDA expert-tier budget GB (default 4)
COLI_EFFICIENCY_PROMPT=... prompt (default: a code-gen prompt)
Exit code is always 0 (it's a dossier, not a gate). Lines marked FLAG point at
the most likely lever to move tok/s for the observed bottleneck.
"""
import os
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from tools.efficiency import run_engine, disk_wait_share # noqa: E402
# --- advisory thresholds (the "IS IT INEFFICIENT?" lines) ---
DISK_WAIT_DOMINANT = 0.40 # >40% decode waiting on expert reads -> I/O-bound
LOW_HIT_RATE = 0.30 # <30% cache hit -> thrashing (cap too small)
LOW_ROUTE_AGREE = 0.80 # <80% routing overlap -> prefetch is guessing wrong
HIGH_TAIL_RATIO = 3.0 # p99 > 3x p50 -> decode stalls (I/O hiccups / KV grow)
LOW_MTP_ACCEPT = 0.20 # <20% MTP acceptance -> draft decoder is dead weight
VRAM_WASTE_CALLS = 0 # experts pinned in VRAM but 0 calls served
def _flag(ok): return "OK " if ok else "FLAG"
def _bar(frac, width=24):
"""A simple ASCII bar for share visualization."""
n = max(0, min(width, round(frac * width)))
return "#" * n + "." * (width - n)
def _line(label, value, flag=None, note=""):
tag = f" [{flag}]" if flag else ""
print(f" {label:<22} {value}{tag} {note}" if note else f" {label:<22} {value}{tag}")
def main() -> int:
model = os.environ.get("COLI_EFFICIENCY_MODEL")
if not model:
print(__doc__)
print("\nNot activated: set COLI_EFFICIENCY_MODEL=<model_dir> to run.")
return 0
model = str(Path(model).resolve())
if not Path(model).is_dir():
print(f"ERROR: {model} is not a directory", file=sys.stderr)
return 0
ngen = int(os.environ.get("COLI_EFFICIENCY_NGEN", "24"))
ram_gb = os.environ.get("COLI_EFFICIENCY_RAM_GB", "28")
vram_gb = os.environ.get("COLI_EFFICIENCY_VRAM_GB", "4")
prompt = os.environ.get(
"COLI_EFFICIENCY_PROMPT",
"Write a Python function that computes the factorial of a number. "
"Include error handling and a docstring.")
use_cuda = os.environ.get("COLI_EFFICIENCY_CUDA") == "1"
# Turn ON every instrumentation flag so the dossier has maximum detail.
# These are all observability toggles (PROF/COLI_CUDA_PROFILE/CACHE_ROUTE/
# DISK_SPLIT/LOOKA); none change the computed output.
overlay = dict(
NGEN=str(ngen), TEMP="0", RAM_GB=ram_gb, PROMPT=prompt,
PROF="1", CACHE_ROUTE="1", DISK_SPLIT="1", LOOKA="1", ROUTE_AGREE="1",
)
if use_cuda:
overlay.update(COLI_CUDA="1", COLI_GPU="0", CUDA_DENSE="1",
COLI_CUDA_PROFILE="1", CUDA_EXPERT_GB=vram_gb)
print("=" * 78)
print(f"OPTIMIZATION DOSSIER — {Path(model).name}")
print(f" mode : {'CUDA (dense+expert tiers)' if use_cuda else 'CPU-only'} "
f"ngen : {ngen} ram : {ram_gb} GB" +
(f" vram : {vram_gb} GB" if use_cuda else ""))
print("=" * 78)
t0 = time.time()
t, proc = run_engine(overlay, snap=model, timeout=3600.0)
wall = time.time() - t0
flags = [] # collected FLAG lines for the summary
print(f"\n[0] RUN")
_line("wall clock", f"{wall:.0f}s")
_line("exit code", proc.returncode,
None if proc.returncode == 0 else "FLAG",
"" if proc.returncode == 0 else "non-zero exit")
if proc.returncode != 0:
print(" stderr tail:")
for ln in proc.stderr.strip().splitlines()[-8:]:
print(f" {ln}")
return 0
# ---------------------------------------------------------------- [1] WHO ----
print(f"\n[1] PROVENANCE — what is running, on what, with what config")
if t.get("machine"):
m = t["machine"]
_line("CPU", m["cpu"])
_line("cores / omp", f"{m['cores']} cores")
_line("backend", m["backend"])
if t.get("load"):
ld = t["load"]
_line("model load time", f"{ld['load_s']:.2f}s")
_line("resident dense", f"{ld['resident_dense_mb']:.1f} MB")
_line("layers / experts", f"{ld['layers']} layers, {ld['experts']} experts")
_line("MTP", f"{ld['mtp_status']} (draft={ld['draft']})")
if t.get("config_str"):
_line("resolved config", t["config_str"])
print(" (this is the EFFECTIVE config after auto-budgeting — not your env verbatim)")
# ---------------------------------------------------------------- [2] SPEED --
print(f"\n[2] THROUGHPUT — is it fast, is the tail healthy")
if t.get("tok_s") is not None:
_line("tok/s", f"{t['tok_s']:.3f}")
else:
flags.append("throughput line missing — engine output format may have changed")
if t.get("latency"):
la = t["latency"]
_line("decode forwards", f"{int(la['forwards'])}")
_line("p50 / p90", f"{la['p50_ms']:.1f} / {la['p90_ms']:.1f} ms")
_line("p99 / max", f"{la['p99_ms']:.1f} / {la['max_ms']:.1f} ms")
tail_ok = la["p99_ms"] <= HIGH_TAIL_RATIO * la["p50_ms"]
_line("tail ratio (p99/p50)", f"{la['p99_ms']/max(la['p50_ms'],1e-9):.2f}x",
_flag(tail_ok),
"high tail = decode stalls (I/O hiccups, KV growth, re-pin)")
if not tail_ok:
flags.append(f"tail latency p99={la['p99_ms']:.1f}ms >> p50={la['p50_ms']:.1f}ms "
"(look for REPIN swaps or disk stalls)")
# ---------------------------------------------------------------- [3] TIME ---
print(f"\n[3] WHERE TIME GOES — what is doing it, when (share of decode)")
ts = t.get("time_shares")
prof = t.get("profile")
if ts:
order = [("io", "expert-disk I/O", DISK_WAIT_DOMINANT, "the cache is too small / disk is slow"),
("matmul", "expert matmul", 0.40, "compute-bound; more cores or a GPU expert tier"),
("attention", "attention", 0.35, "context length is the cost; lower CTX"),
("head", "lm_head", 0.10, "vocab projection; unusual to dominate"),
("other", "other", 0.30, "scheduling / KV bookkeeping overhead")]
for key, name, thresh, lever in order:
f = ts[key]
ok = f < thresh
_line(name, f"{f:5.0%} {_bar(f)}", _flag(ok),
"" if ok else f"->{lever}")
if not ok:
flags.append(f"{name} dominates ({f:.0%}) -> {lever}")
if t.get("verdict"):
print(f" engine verdict : {t['verdict']}")
elif prof:
print(" (no [PROF] time shares — set PROF=1 for phase percentages)")
if prof:
print(" absolute seconds :")
for k in ("disk", "expert_matmul", "attention", "lm_head", "other"):
_line(k, f"{prof[k]:.3f}s")
# attention sub-breakdown: how is attention being read
ab = t.get("attn_breakdown")
if ab:
print(f"\n[3a] ATTENTION BREAKDOWN — how the attention phase is spent")
atot = sum(ab.values()) or 1.0
for k, label in (("proj_rope", "projection + RoPE"),
("score_sm_value", "score-softmax-value"),
("out_proj", "output projection")):
_line(label, f"{ab[k]:.3f}s ({ab[k]/atot:.0%} of attn)")
# ---------------------------------------------------------------- [4] CACHE --
print(f"\n[4] EXPERT CACHE — is the cache efficient")
hit = t.get("hit_pct")
if hit is not None:
ok = hit >= LOW_HIT_RATE * 100
_line("hit rate", f"{hit:.1f}%", _flag(ok),
"" if ok else "<30% = thrashing; raise RAM_GB or cap")
if not ok:
flags.append(f"cache hit {hit:.1f}% is low -> raise RAM_GB (or cap), add PIN_GB")
el = t.get("experts_loaded")
if el:
per_tok = el["per_tok"]
_line("experts loaded/token", f"{per_tok:.1f}")
_line(" per-layer", f"{el['per_layer']:.2f} across {el['n_sparse_layers']} sparse layers")
_line(" baseline", f"topk={el['baseline_topk']} active experts/token")
base_topk = el["baseline_topk"]
if base_topk > 0 and per_tok > 2 * base_topk:
flags.append(f"loading {per_tok:.0f} experts/token vs topk={base_topk} "
"-> redundant I/O; cache is re-fetching evicted experts")
# ---------------------------------------------------------------- [5] DISK ---
print(f"\n[5] DISK I/O — is I/O the bottleneck, and where")
eio = t.get("expert_io")
if eio:
_line("total fetched", f"{eio['gb_fetched']:.3f} GB")
_line("per token", f"{eio['mb_per_tok']:.1f} MB/token")
_line("disk throughput", f"{eio['gb_per_s']:.2f} GB/s over the run")
_line("read service", f"{eio['read_service_s']:.2f}s (on I/O threads)")
_line("felt wait", f"{eio['felt_wait_s']:.2f}s (stall compute felt)")
if eio["felt_wait_s"] > eio["read_service_s"] * 0.5 and eio["read_service_s"] > 0:
flags.append("felt wait is a large fraction of read service -> PIPE=1 may not be "
"overlapping fully, or DIRECT=1 on NVMe")
ds = t.get("disk_split")
if ds:
print(f"\n[5a] DISK-LOAD SPLIT — which decode phase reads the bytes")
_line("draft phase", f"{ds['draft']} loads")
_line("absorb phase", f"{ds['absorb']} loads")
_line("verify/main", f"{ds['verify_main']} loads")
_line("MTP-layer bytes", f"{ds['mtp_loads']} loads, {ds['mtp_gb']:.2f} GB")
_line("main-layer bytes", f"{ds['main_loads']} loads, {ds['main_gb']:.2f} GB")
if ds.get("mtp_bytes_pct") is not None:
_line("MTP share of bytes", f"{ds['mtp_bytes_pct']:.1f}%")
share = disk_wait_share(t)
if share is not None:
ok = share < DISK_WAIT_DOMINANT
_line("disk-wait share", f"{share:.0%}", _flag(ok),
"" if ok else "I/O-bound (see levers in [3])")
# ---------------------------------------------------------------- [6] ROUTE --
print(f"\n[6] ROUTING QUALITY — is the router / prefetch accurate")
ra = t.get("route_agree")
if ra:
ok = ra["agree_pct"] >= LOW_ROUTE_AGREE * 100
_line("route_agree", f"{ra['agree_pct']:.1f}% overlap with true top-K",
_flag(ok),
"" if ok else "prefetch is guessing wrong; CACHE_ROUTE params may need tuning")
_line("route_kl", f"{ra['kl']:.4f} mean KL (lower = closer to true routing)")
if not ok:
flags.append(f"route_agree {ra['agree_pct']:.1f}% low -> tune ROUTE_J/M/P, "
"or prefetch is hurting more than helping")
sw = t.get("swap")
if sw:
_line("cache swaps", f"{sw['swaps']}/{sw['slots']} ({sw['pct']:.1f}%)",
None, "high swap = churn between turns")
la = t.get("lookahead")
if la:
print(f"\n[6a] ROUTING PREDICTABILITY — recall of true experts in predicted top-8")
print(" (which predictor should drive prefetch? highest recall wins)")
for row in la:
_line(row["predictor"][:34], f"{row['pct']:5.1f}% ({row['hit']}/{row['tot']})")
# ---------------------------------------------------------------- [7] SPEC ---
print(f"\n[7] SPECULATION — is the draft decoder pulling weight")
sp = t.get("speculation")
if sp:
_line("tokens/forward", f"{sp['tok_per_fw']:.2f} (>1.0 means speculation helps)")
_line("forwards/tokens", f"{sp['forwards']} forwards for {sp['tokens']} tokens")
# Speculation helps only if acceptance is high enough that tok/forward > 1.
# tok_per_fw already == 1.0 when nothing verifies, so judge by acceptance.
ok = sp["mtp_accept_pct"] >= LOW_MTP_ACCEPT * 100
_line("MTP acceptance", f"{sp['mtp_accept_pct']:.0f}%", _flag(ok),
"" if ok else "<20% -> drafts rarely verify; DRAFT=0 may be faster")
if not ok:
flags.append(f"MTP acceptance {sp['mtp_accept_pct']:.0f}% low -> "
"drafts cost more I/O than they save; try DRAFT=0")
# ---------------------------------------------------------------- [8] GPU ----
print(f"\n[8] GPU TIERS — is the GPU actually used")
cuda = t.get("cuda") or {}
if not cuda.get("enabled"):
print(" (CUDA not enabled — CPU-only run)")
else:
if cuda.get("resident_tensors") is not None:
_line("resident dense tensors", f"{cuda['resident_tensors']} tensors, "
f"{cuda['resident_gb']:.2f} GB")
if cuda.get("expert_count") is not None:
waste = cuda["calls_served"] <= VRAM_WASTE_CALLS
_line("expert tier", f"{cuda['expert_count']} experts pinned "
f"({cuda['expert_gb']:.2f} GB)", _flag(not waste))
_line(" calls served", f"{cuda['calls_served']} from VRAM",
_flag(not waste),
"" if not waste else "WASTE: pinned but never routed -> lower CUDA_EXPERT_GB")
if waste:
flags.append("VRAM expert tier has 0 calls served -> experts pinned but unused; "
"PIN stats may not match this workload")
if cuda.get("groups"):
g = cuda["groups"]
_line("expert groups", f"{g['calls']} calls, {g['experts']} experts, "
f"{g['rows']} rows ({g['experts_per_call']:.1f} experts/call)")
if cuda.get("groups_timing"):
gt = cuda["groups_timing"]
_line("GPU timing", f"H2D {gt['h2d_ms']:.1f} ms | kernel {gt['kernel_ms']:.1f} ms | "
f"D2H {gt['d2h_ms']:.1f} ms")
if gt["h2d_ms"] + gt["d2h_ms"] > gt["kernel_ms"]:
flags.append("CUDA H2D+D2H > kernel time -> transfer-bound; "
"consider larger expert tier to keep weights resident")
# ---------------------------------------------------------------- summary ----
print("\n" + "=" * 78)
if flags:
print(f" {len(flags)} FLAG(s) — the most likely levers to move tok/s:")
for f in flags:
print(f" - {f}")
else:
print(" no flags — every measured subsystem is within advisory thresholds.")
print("=" * 78)
return 0
if __name__ == "__main__":
raise SystemExit(main())
c/tests/test_inefficiency.py
+257
View File
@@ -0,0 +1,257 @@
"""Inefficiency / regression tests for the colibri engine (tiny model, asserted).
These run against the bundled glm_tiny model (~0.6 MB resident, ~0.1s/run) and
gate CI: a regression here means something broke. They run on a plain CPU-only
`glm.exe` build; the CUDA_* tests auto-skip if the engine wasn't built with
CUDA_DLL=1 (see tests/README_efficiency.md for the build command).
The signals under test, and the inefficiency each catches:
- tok/s floor : a throughput regression (broken build / bad config)
- profile phases sum : telemetry accounting bug (other balloons)
- disk-wait not dominant: a tiny resident model should never be I/O-bound
- CPU determinism : greedy decode is reproducible (no stray RNG/threading)
- CUDA init path : COLI_CUDA=1 initializes and does not silently exit 2
- CUDA dense uses VRAM : CUDA_DENSE=1 actually uploads tensors (no silent CPU fallback)
- CPU vs CUDA TF-match : identical weights+inputs → identical argmax (kernel bug guard)
"""
import os
import shutil
import unittest
from pathlib import Path
from tools.efficiency import (
parse_run, run_engine, disk_wait_share, tf_agreement,
TINY_TOK_S_FLOOR, MAX_DISK_WAIT_SHARE, MIN_CPU_CUDA_AGREEMENT,
)
HERE = Path(__file__).resolve().parent
C_DIR = HERE.parent
ENGINE = C_DIR / "glm.exe"
TINY = C_DIR / "glm_tiny"
def _engine_present() -> bool:
"""True iff BOTH the built engine AND the tiny fixture are available.
These tests need glm.exe (a build artifact) AND glm_tiny/ (a generated
fixture, gitignored). CI runs `make check` = "dependency-free tests, no
model downloads" (workflow .github/workflows/check.yml, by design #140), so
neither is present there and these tests must SKIP rather than fail. They
run locally after `make glm.exe` (glm_tiny ships alongside the source, or
is regenerated by tools/make_glm_oracle.py).
"""
return ENGINE.exists() and (TINY / "config.json").exists()
def _skip_reason() -> str:
"""Name exactly which prerequisite is missing, so the skip is actionable."""
if not ENGINE.exists():
return "glm.exe not built (run: make glm.exe)"
if not (TINY / "config.json").exists():
return "glm_tiny fixture absent (gitignored; ship it locally or run tools/make_glm_oracle.py)"
return ""
def _cuda_available() -> bool:
"""True iff the engine binary has the CUDA loader compiled in AND the DLL is present.
The host is built with -DCOLI_CUDA only when CUDA_DLL=1 (Makefile). A binary
built without it embeds the string "this binary is CPU-only; rebuild" and
exits 2 on any CUDA env var — so we detect the CPU-only build by scanning
the binary for that marker (avoids a slow ldd/strings on every import; we
only read enough to find it). On Windows the DLL is also required
(backend_loader.c loads it at runtime); on Linux it's direct-linked.
"""
if not _engine_present():
return False
try:
# Read once; the marker is near the read-only string table. 256 KB is
# plenty for this string and avoids loading a 1 MB binary into memory.
with open(ENGINE, "rb") as f:
blob = f.read(2 * 1024 * 1024)
if b"this binary is CPU-only" in blob:
return False # CPU-only build: COLI_CUDA=1 would exit 2.
except OSError:
pass
# Windows: DLL required at runtime. Linux: direct-linked (no DLL).
if os.name == "nt":
return (C_DIR / "coli_cuda.dll").exists()
import subprocess
try:
out = subprocess.run(["ldd", str(ENGINE)], capture_output=True, text=True)
return "libcudart" in out.stdout
except (FileNotFoundError, OSError):
return False
@unittest.skipUnless(_engine_present(), _skip_reason() or "glm.exe + glm_tiny required")
class TinyEfficiencyTest(unittest.TestCase):
"""Asserted regression tests on the resident tiny model. Gates CI."""
def _run(self, **overlay):
return run_engine(overlay, engine=str(ENGINE), snap=str(TINY))[0]
# -- telemetry contract ---------------------------------------------------
def test_telemetry_parses(self):
"""A REPLAY run must emit the throughput + PROFILE lines the suite keys on.
If this fails, either the engine changed its output format (update the
parsers in tools/efficiency.py) or the run crashed early."""
t = self._run(REPLAY="1", TEMP="0", NGEN="4")
self.assertEqual(t["returncode"], 0, f"engine exited non-zero:\n{t['stderr']}")
self.assertIn("tok_s", t["parsed"], f"missing tok/s line:\n{t['stderr']}")
self.assertIn("profile", t["parsed"], f"missing PROFILE line:\n{t['stderr']}")
self.assertIn("hit_pct", t["parsed"], f"missing expert hit line:\n{t['stderr']}")
self.assertIsNotNone(t["tok_s"])
self.assertIsNotNone(t["profile"])
# -- throughput floor -----------------------------------------------------
def test_tiny_tok_s_floor(self):
"""Tiny decode must beat TINY_TOK_S_FLOOR.
The tiny model is fully resident and runs ~200 tok/s; the 20 tok/s
default floor is a 10x margin that catches broken builds or a pathological
config cascade (the cap=1 trap from ISSUE_new_model_resource_regression.md)
without flapping on machine noise."""
t = self._run(REPLAY="1", TEMP="0", NGEN="8")
self.assertEqual(t["returncode"], 0, f"engine exited non-zero:\n{t['stderr']}")
self.assertGreaterEqual(
t["tok_s"], TINY_TOK_S_FLOOR,
f"tok/s {t['tok_s']:.1f} below floor {TINY_TOK_S_FLOOR} "
f"(regression, or a config cascade starving the cache)",
)
# -- accounting sanity ----------------------------------------------------
def test_profile_phases_present_and_nonneg(self):
"""Every PROFILE phase must be present and non-negative.
`other` can go slightly negative from timer overhead (the engine allows
it), but a large negative means the timers are double-counting."""
t = self._run(REPLAY="1", TEMP="0", NGEN="4")
p = t["profile"]
for phase in ("disk", "expert_matmul", "attention", "lm_head"):
self.assertGreaterEqual(p[phase], -0.01, f"{phase} went negative: {p}")
# 'other' is a residual; allow a small negative from timer overlap.
self.assertGreaterEqual(p["other"], -0.05, f"other too negative (double-count): {p}")
# -- disk-wait not dominant on a resident model ---------------------------
def test_disk_wait_not_dominant(self):
"""A fully-resident tiny model must NOT be I/O-bound.
Everything fits in RAM; the expert-disk wait share should be ~0. If it
exceeds MAX_DISK_WAIT_SHARE, the cache/I/O path regressed — on a real
model this same regression would make decode I/O-bound (the exact
failure mode the [PROF] verdict flags)."""
t = self._run(REPLAY="1", TEMP="0", NGEN="8", PROF="1")
self.assertIn("time_shares", t["parsed"], f"missing [PROF] time shares:\n{t['stderr']}")
share = disk_wait_share(t)
self.assertIsNotNone(share)
self.assertLess(
share, MAX_DISK_WAIT_SHARE,
f"expert-I/O share {share:.0%} on a resident model — I/O path regressed",
)
# -- determinism ----------------------------------------------------------
def test_cpu_vs_cpu_determinism(self):
"""Two greedy REPLAY runs with the same seed produce identical telemetry.
TEMP=0 = greedy (no sampling), so tok/s and hit-rate must be reproducible.
A drift here means non-determinism crept into the decode path (stray
threading, uninitialized state) — which on a real model would make A/B
comparisons meaningless."""
a = self._run(REPLAY="1", TEMP="0", NGEN="8", SEED="1")
b = self._run(REPLAY="1", TEMP="0", NGEN="8", SEED="1")
self.assertEqual(a["returncode"], 0)
self.assertEqual(b["returncode"], 0)
self.assertEqual(a["hit_pct"], b["hit_pct"], "greedy hit-rate drifted between runs")
# tok/s within 25% — exact equality is too strict across scheduler noise.
self.assertLess(abs(a["tok_s"] - b["tok_s"]) / max(a["tok_s"], b["tok_s"]), 0.25)
@unittest.skipUnless(_cuda_available(),
_skip_reason() or "CUDA build not present (run: make clean && make glm.exe CUDA_DLL=1 && make cuda-dll)")
class TinyCudaEfficiencyTest(unittest.TestCase):
"""CUDA-path regression tests on the tiny model. Skip unless CUDA built.
glm_tiny is small and fully resident, so CUDA here is fast and exercises the
real GPU code path (init, dense upload, kernel correctness) without the
long load time or memory pressure of the full model. These guard the
silent-failure modes that are otherwise invisible:
- COLI_CUDA=1 silently falling back to CPU (loader/DLL missing)
- CUDA_DENSE=1 uploading nothing
- a CUDA kernel producing different argmax than CPU on identical inputs
"""
def _run(self, **overlay):
return run_engine(overlay, engine=str(ENGINE), snap=str(TINY))[0]
def test_cuda_init_path(self):
"""COLI_CUDA=1 must initialize the device and NOT exit 2.
Exit 2 is the engine's "requested backend is unavailable" path
(glm.c: g_cuda_enabled check). A clean init prints the [CUDA] device
banner to stderr. If this fails, the DLL is broken or the loader can't
resolve symbols (ABI drift between backend_cuda.h and the dll)."""
t = self._run(COLI_CUDA="1", COLI_GPU="0", REPLAY="1", TEMP="0", NGEN="4")
self.assertNotEqual(t["returncode"], 2,
f"engine refused CUDA backend:\n{t['stderr']}")
self.assertTrue(t["cuda"]["enabled"],
f"no [CUDA] device banner on stderr:\n{t['stderr']}")
def test_cuda_dense_uses_vram(self):
"""CUDA_DENSE=1 must actually upload dense tensors to VRAM.
The minimal GPU-exercising config (per backend_loader.c analysis):
COLI_CUDA=1 + CUDA_DENSE=1. Without CUDA_DENSE the dense path stays on
CPU and [CUDA] resident set reports 0 tensors — a silent no-op. This
catches that regression: after the run, resident_tensors > 0."""
t = self._run(COLI_CUDA="1", COLI_GPU="0", CUDA_DENSE="1",
REPLAY="1", TEMP="0", NGEN="4")
self.assertEqual(t["returncode"], 0, f"engine exited non-zero:\n{t['stderr']}")
rt = t["cuda"]["resident_tensors"]
self.assertIsNotNone(rt, f"no [CUDA] resident set line:\n{t['stderr']}")
self.assertGreater(rt, 0,
f"CUDA_DENSE=1 but {rt} tensors resident — silent CPU fallback")
def test_cpu_vs_cuda_tf_match(self):
"""CPU and CUDA teacher-forcing must agree on most positions (DIRECTLY).
Both paths prefill the SAME oracle sequence on the SAME weights, so their
argmaxes should match position-for-position — but not exactly: the two
backends accumulate dot-products in different orders (x86 SIMD vs CUDA
kernel), so a few near-tied logits flip. That divergence is expected
numeric behavior, not a kernel bug. A *catastrophic* kernel regression
(wrong GEMM, wrong scale, wrong fmt) would collapse agreement toward
random (~1/vocab = ~4%); the MIN_CPU_CUDA_AGREEMENT floor (default 70%)
catches that while tolerating harmless drift.
We compare CPU-vs-CUDA *directly* (not via the oracle match-counts),
because both backends differ from the oracle at different positions and
the summary line can't tell "CPU≠CUDA" from "CPU≠oracle"."""
import json
ref = json.loads((C_DIR / "ref_glm.json").read_text())
oracle = ref["tf_pred"]
cpu = self._run(TF="1", TEMP="0")
cuda = self._run(COLI_CUDA="1", COLI_GPU="0", CUDA_DENSE="1", TF="1", TEMP="0")
self.assertEqual(cpu["returncode"], 0, f"CPU TF run failed:\n{cpu['stderr']}")
self.assertEqual(cuda["returncode"], 0, f"CUDA TF run failed:\n{cuda['stderr']}")
self.assertIn("tf_mismatches", cpu["parsed"], "CPU run missing per-position mismatches")
self.assertIn("tf_mismatches", cuda["parsed"], "CUDA run missing per-position mismatches")
agree, diff = tf_agreement(cpu, cuda, oracle)
self.assertGreaterEqual(
agree, MIN_CPU_CUDA_AGREEMENT,
f"CPU-vs-CUDA argmax agreement {agree:.0%} below floor "
f"{MIN_CPU_CUDA_AGREEMENT:.0%} — CUDA kernel likely regressed. "
f"Differing positions: {diff[:10]}{'...' if len(diff)>10 else ''}",
)
if __name__ == "__main__":
unittest.main()
c/tools/efficiency.py
+458
View File
@@ -0,0 +1,458 @@
"""Efficiency / regression harness for the colibri engine.
The engine already emits rich telemetry (REPLAY tok/s, PROFILE phase timings,
[PROF] time shares + verdict, CUDA expert-tier utilization). Until now every
consumer of that telemetry — `benchmark_cuda_fixture.py`, `bench_full.sh`,
`bench_ux.sh` — has only *printed* it for a human to eyeball. This module turns
each signal into a parseable field so tests can assert on it.
Design:
- Reuses SPEED_RE / PROFILE_RE from tools.benchmark_cuda_fixture (no drift).
- parse_run() is pure: stdout+stderr in, dict out. Easy to unit-test against
captured strings (like the existing test_benchmark_cuda_fixture does).
- run_engine() is the subprocess wrapper. Captures stdout and stderr
separately, because the engine splits them: PROFILE/REPLAY/CUDA-tier go to
stdout, the [CUDA]/[PROF]/[prefill] banners go to stderr.
- Floor defaults are module constants (tunable in one place, not scattered).
No model file is required to import this module; only run_engine() invokes the
binary. parse_run() works on any captured text, so most test surface is covered
by string fixtures without spinning the engine at all.
"""
from __future__ import annotations
import os
import re
import subprocess
from pathlib import Path
from typing import Optional
# Reuse the validated regexes from the existing A/B benchmark harness so the
# PROFILE field order (disk, expert_matmul, attention, lm_head, other) and the
# tok/s capture stay identical. Drift here would silently break every consumer.
from tools.benchmark_cuda_fixture import SPEED_RE as _SPEED_RE_REPLAY, PROFILE_RE, PROFILE_KEYS
# SPEED_RE (from benchmark_cuda_fixture) matches the REPLAY-mode line only:
# "REPLAY decode: ... | 12.34 tok/s | ..."
# run_text / PROMPT mode uses a DIFFERENT format (glm.c:4682):
# "decode N tokens in X.XXs (12.34 tok/s) | expert hit rate ..."
# This alt regex catches the parenthesized form so the full-model report (which
# uses PROMPT mode) gets a real tok/s instead of reporting it missing.
SPEED_RE_TEXT = re.compile(r"decode \d+ tokens in [0-9.]+s \(([0-9.]+) tok/s\)")
def _first_speed(stdout: str):
"""Find tok/s in whichever run-mode format the engine used."""
for rx in (_SPEED_RE_REPLAY, SPEED_RE_TEXT):
m = rx.search(stdout)
if m:
return m
return None
# Public alias so existing imports keep working (tests reference SPEED_RE).
SPEED_RE = _SPEED_RE_REPLAY
# --- additional parsers (formats verified against glm.c printf strings) ---
# "expert hit rate 88.1%" (summary line) | "expert hit 95.0%" (REPLAY line)
HIT_RE = re.compile(r"expert hit(?:\s+rate)?\s+([0-9.]+)%")
# "[PROF] time shares: expert-I/O 3% | expert-matmul 12% | attention 56% | lm_head 2% | other 27%"
SHARES_RE = re.compile(
r"\[PROF\] time shares: expert-I/O\s+([0-9.]+)%\s*\|\s*expert-matmul\s+([0-9.]+)%\s*"
r"\|\s*attention\s+([0-9.]+)%\s*\|\s*lm_head\s+([0-9.]+)%\s*\|\s*other\s+([0-9.]+)%"
)
# "[PROF] verdict: I/O-bound — 60% of the time ..." (also compute-bound / attention-bound / balanced)
VERDICT_RE = re.compile(r"\[PROF\] verdict:\s*(I/O-bound|compute-bound|attention-bound|balanced)")
# "[PROF] expert I/O: ... hit 95.0% (76 hit / 4 load) | 4.0 loads/token"
LOADS_PER_TOK_RE = re.compile(r"\|\s*([0-9.]+)\s+loads/token")
# "CUDA expert tier: 111 resident experts (2.36 GB) | 5400 calls served from VRAM" (stdout)
CUDA_TIER_RE = re.compile(
r"CUDA expert tier:\s+(\d+)\s+resident experts\s+\(([0-9.]+)\s+GB\)\s*\|\s+(\d+)\s+calls served from VRAM"
)
# "[CUDA] device 0: NVIDIA ..., 14.4 GB VRAM, sm_120" (stderr, per device at init)
CUDA_DEVICE_RE = re.compile(r"\[CUDA\] device\s+\d+:")
# "[CUDA] resident set: 12 tensors, 0.45 GB VRAM" (stderr, cuda_stats_print)
CUDA_RESIDENT_RE = re.compile(
r"\[CUDA\] resident set:\s+(\d+)\s+tensors,\s+([0-9.]+)\s+GB\s+VRAM"
)
# "PREFILL (teacher-forcing) C vs oracle: 11/32 positions | 1700.4 pos/s" (TF=1 mode, stdout)
TF_MATCH_RE = re.compile(r"PREFILL \(teacher-forcing\).*:\s+(\d+)/(\d+)\s+positions")
# --- the six deeper signals (added after "are you gathering everything?" audit) ---
# "ATTENTION: projection/RoPE 0.050s | score-softmax-value 0.009s | output projection 0.011s"
# Sub-breakdown of the attention phase — answers "how is attention being read".
ATTN_BREAKDOWN_RE = re.compile(
r"ATTENTION: projection/RoPE\s+([0-9.]+)s\s*\|\s*score-softmax-value\s+([0-9.]+)s\s*"
r"\|\s*output projection\s+([0-9.]+)s"
)
# "[PROF] decode forwards: 20 | latency p50 7.3 ms | p90 8.0 ms | p99 8.1 ms | max 8.2 ms | 1.00 tok/forward"
# Per-forward tail latency — a p99 >> p50 means decode stalls (I/O hiccups, KV grow).
LATENCY_RE = re.compile(
r"\[PROF\] decode forwards:\s+(\d+)\s*\| latency p50\s+([0-9.]+)\s*ms\s*\| "
r"p90\s+([0-9.]+)\s*ms\s*\|\s*p99\s+([0-9.]+)\s*ms\s*\|\s*max\s+([0-9.]+)\s*ms"
)
# "[PROF] expert I/O: 0.004 GB fetched (0.2 MB/token, 0.03 GB/s over the run) | hit 95.0% ... |
# 4.0 loads/token | 0.0s read service / 0.0s felt wait"
# Absolute disk throughput + the felt-wait split (the [PROF] version, more detailed than PROFILE).
EXPERT_IO_RE = re.compile(
r"\[PROF\] expert I/O:\s+([0-9.]+)\s+GB fetched\s+\(([0-9.]+)\s+MB/token,\s*([0-9.]+)\s+GB/s"
r".*?\|\s*([0-9.]+)\s+loads/token\s*\|\s*([0-9.]+)s\s+read service\s*/\s*([0-9.]+)s\s+felt wait"
)
# "speculation: 1.05 tokens/forward (19 forwards per 20 tokens) | MTP acceptance 44% (7/16)"
# Draft efficiency — is the speculative decoder pulling weight or dead overhead?
SPECULATION_RE = re.compile(
r"speculation:\s+([0-9.]+)\s+tokens/forward\s+\((\d+)\s+forwards per\s+(\d+)\s+tokens\)"
r"\s*\|\s*MTP acceptance\s+([0-9.]+)%"
)
# "experts loaded/token: 450.0 (per-layer 56.25 across 8; baseline topk=8) | TOPK=0 TOPP=0.00"
# Fuller than loads_per_tok: includes the per-layer spread + the active topk/topp.
EXPERTS_LOADED_RE = re.compile(
r"experts loaded/token:\s+([0-9.]+)\s+\(per-layer\s+([0-9.]+)\s+across\s+(\d+);\s*baseline topk=(\d+)\)"
)
# "[PROF] machine: Intel(...) | 22 cores (22 omp threads) | RAM 34.1 GB total, 27.1 GB available | backend CUDA"
# Provenance — makes a report reproducible across machines/runs.
MACHINE_RE = re.compile(r"\[PROF\] machine:\s*(.*)\|\s*(\d+)\s+cores.*backend\s+(\S+)")
# "[PROF] config: RAM_GB=auto 23.9 CTX=4096 | expert cache cap 8/layer ... | DRAFT=0 PIPE=1 DIRECT=0 ..."
# Effective resolved config (after auto-budgeting). Answers "what config actually ran".
CONFIG_RE = re.compile(r"\[PROF\] config:\s*(.*)")
# "[ORACLE] mismatch pos=7 expected=197 got=22" (TF=1 mode, stderr, per position)
# Captures the engine's *actual* argmax at each position, so two backends can be
# compared DIRECTLY (independent of how each relates to the oracle).
TF_MISMATCH_RE = re.compile(r"\[ORACLE\] mismatch pos=(\d+) expected=\d+ got=(\d+)")
# --- routing-quality + disk-split + cuda-groups (the deep signals) ---
# summary suffix: " | swap 3.1% (12/384)" (CACHE_ROUTE inline)
SWAP_RE = re.compile(r"swap\s+([0-9.]+)%\s+\((\d+)/(\d+)\)")
# summary suffix: " | route_agree 85.0% | route_kl 0.0123" (CACHE_ROUTE/ROUTE_AGREE)
ROUTE_AGREE_RE = re.compile(r"route_agree\s+([0-9.]+)%\s*\|\s*route_kl\s+([0-9.]+)")
# "disk-load split: draft 8 + absorb 0 + verify/main 3 misses | MTP-layer 0 loads 0.00 GB |
# main-layers 11 loads 0.01 GB (MTP 0.0% of bytes)" (DISK_SPLIT=1)
DISK_SPLIT_RE = re.compile(
r"disk-load split: draft\s+(\d+)\s+\+\s+absorb\s+(\d+)\s+\+\s+verify/main\s+(\d+)\s+misses"
r"\s*\|\s*MTP-layer\s+(\d+)\s+loads\s+([0-9.]+)\s+GB\s*\|\s*main-layers\s+(\d+)\s+loads\s+([0-9.]+)\s+GB"
r"(?:\s+\(MTP\s+([0-9.]+)%\s+of bytes\))?"
)
# "[CUDA] expert groups: 120 call, 840 expert, 1200 righe (7.00 expert/call)"
CUDA_GROUPS_RE = re.compile(
r"\[CUDA\] expert groups:\s+(\d+)\s+call,\s+(\d+)\s+expert,\s+(\d+)\s+righe\s+\(([0-9.]+)\s+expert/call\)"
)
# "[CUDA] expert groups timing: H2D 12.3 ms | kernel 45.6 ms | D2H 7.8 ms" (COLI_CUDA_PROFILE=1)
CUDA_GROUPS_TIME_RE = re.compile(
r"\[CUDA\] expert groups timing: H2D\s+([0-9.]+)\s+ms\s*\|\s*kernel\s+([0-9.]+)\s+ms\s*\|\s*D2H\s+([0-9.]+)\s+ms"
)
# LOOKAHEAD recall block — 4 named rows. (name, pct, hit, tot)
LOOKAHEAD_RE = re.compile(
r"^\s*(.+?)\s+([0-9.]+)%\s+\((\d+)/(\d+)\)\s*$", re.MULTILINE
)
# "loaded in 0.02s | resident dense: 0.21 MB | layers=5 experts=8 | MTP absent (draft=0)"
LOAD_BANNER_RE = re.compile(
r"loaded in\s+([0-9.]+)s\s*\|\s*resident dense:\s+([0-9.]+)\s+MB\s*\|"
r"\s*layers=(\d+)\s+experts=(\d+)\s*\|\s*MTP\s+(\w+)\s+\(draft=(\d+)\)"
)
# --- tunable floors -----------------------------------------------------------
# These are deliberately generous so they catch *regressions* (broken builds,
# pathological configs, telemetry accounting bugs) without flapping on machine
# noise. The tiny model is fully resident at ~200 tok/s, so a 20 tok/s floor is
# a 10x margin. Tune per-host via env if needed (documented in README).
TINY_TOK_S_FLOOR = float(os.environ.get("COLI_TINY_TOK_S_FLOOR", "20.0"))
# On a fully-resident tiny model the expert-disk wait share should be tiny.
# If it exceeds this, something regressed in the I/O accounting or cache path.
MAX_DISK_WAIT_SHARE = float(os.environ.get("COLI_MAX_DISK_WAIT_SHARE", "0.20"))
# decode wall-time should be roughly the sum of PROFILE phases (other = residual).
PROFILE_SUM_TOLERANCE = float(os.environ.get("COLI_PROFILE_SUM_TOL", "0.05"))
# Minimum direct CPU-vs-CUDA argmax agreement on the tiny TF fixture. The two
# backends use different accumulation orders (SIMD dot vs CUDA kernel), so a
# few near-tied positions flip argmax — that's expected numeric divergence, not
# a kernel bug. Measured baseline ~84% (27/32) on this fixture; the 70% floor
# leaves headroom for machine noise while still catching a catastrophic kernel
# regression (e.g. a wrong GEMM would drop this to ~random = ~4%).
MIN_CPU_CUDA_AGREEMENT = float(os.environ.get("COLI_MIN_CPU_CUDA_AGREE", "0.70"))
def parse_run(stdout: str, stderr: str = "") -> dict:
"""Parse one engine run's output into a telemetry dict.
Returns keys: tok_s, hit_pct, profile (dict, seconds), profile_sum,
time_shares (dict, fractions 0..1), verdict, loads_per_tok, cuda (dict),
tf_match (tuple or None), parsed (set of field names found).
Raises RuntimeError only if the core throughput line is missing — everything
else is optional and absent on some run modes (e.g. [PROF] needs PROF=1,
CUDA tier needs gpu_expert_count>0).
"""
out = dict(
tok_s=None, hit_pct=None, profile=None, profile_sum=None,
time_shares=None, verdict=None, loads_per_tok=None,
cuda=None, tf_match=None, stderr=stderr,
)
parsed = set()
blob = stdout + "\n" + stderr # [PROF]/[CUDA] live on stderr; scan both.
m = _first_speed(stdout)
if m:
out["tok_s"] = float(m.group(1)); parsed.add("tok_s")
m = HIT_RE.search(blob)
if m:
out["hit_pct"] = float(m.group(1)); parsed.add("hit_pct")
m = PROFILE_RE.search(stdout)
if m:
service, wait, emm, attn, head, other = (float(x) for x in m.groups())
disk = service + (wait or 0.0)
out["profile"] = dict(zip(PROFILE_KEYS, (disk, emm, attn, head, other)))
out["profile_sum"] = disk + emm + attn + head + other
parsed.add("profile")
# ATTENTION sub-breakdown: projection/RoPE | score-softmax-value | output.
m = ATTN_BREAKDOWN_RE.search(stdout)
if m:
out["attn_breakdown"] = dict(zip(
("proj_rope", "score_sm_value", "out_proj"),
(float(x) for x in m.groups())))
parsed.add("attn_breakdown")
m = SHARES_RE.search(blob)
if m:
io, emm, attn, head, other = (float(x) / 100.0 for x in m.groups())
out["time_shares"] = dict(io=io, matmul=emm, attention=attn, head=head, other=other)
parsed.add("time_shares")
m = VERDICT_RE.search(blob)
if m:
out["verdict"] = m.group(1); parsed.add("verdict")
# [PROF] decode forwards + latency p50/p90/p99/max.
m = LATENCY_RE.search(blob)
if m:
out["latency"] = dict(zip(
("forwards", "p50_ms", "p90_ms", "p99_ms", "max_ms"),
(float(x) for x in m.groups())))
parsed.add("latency")
# [PROF] expert I/O throughput: GB fetched, MB/token, GB/s, service vs felt wait.
m = EXPERT_IO_RE.search(blob)
if m:
out["expert_io"] = dict(zip(
("gb_fetched", "mb_per_tok", "gb_per_s", "loads_per_tok",
"read_service_s", "felt_wait_s"),
(float(x) for x in m.groups())))
parsed.add("expert_io")
m = LOADS_PER_TOK_RE.search(blob)
if m:
out["loads_per_tok"] = float(m.group(1)); parsed.add("loads_per_tok")
# experts loaded/token with per-layer spread + baseline topk (run_text summary).
m = EXPERTS_LOADED_RE.search(stdout)
if m:
out["experts_loaded"] = dict(
per_tok=float(m.group(1)), per_layer=float(m.group(2)),
n_sparse_layers=int(m.group(3)), baseline_topk=int(m.group(4)))
parsed.add("experts_loaded")
# speculation: tokens/forward, forwards, tokens, MTP acceptance%.
m = SPECULATION_RE.search(stdout)
if m:
out["speculation"] = dict(zip(
("tok_per_fw", "forwards", "tokens", "mtp_accept_pct"),
(float(m.group(1)), int(m.group(2)), int(m.group(3)), float(m.group(4)))))
parsed.add("speculation")
# routing quality (CACHE_ROUTE inline suffixes on the summary line).
m = SWAP_RE.search(stdout)
if m:
out["swap"] = dict(pct=float(m.group(1)), swaps=int(m.group(2)), slots=int(m.group(3)))
parsed.add("swap")
m = ROUTE_AGREE_RE.search(stdout)
if m:
out["route_agree"] = dict(agree_pct=float(m.group(1)), kl=float(m.group(2)))
parsed.add("route_agree")
# disk-load split by decode phase (DISK_SPLIT=1).
m = DISK_SPLIT_RE.search(stdout)
if m:
out["disk_split"] = dict(zip(
("draft", "absorb", "verify_main", "mtp_loads", "mtp_gb",
"main_loads", "main_gb", "mtp_bytes_pct"),
(int(m.group(1)), int(m.group(2)), int(m.group(3)), int(m.group(4)),
float(m.group(5)), int(m.group(6)), float(m.group(7)),
float(m.group(8)) if m.group(8) else None)))
parsed.add("disk_split")
# provenance: machine + resolved config.
m = MACHINE_RE.search(blob)
if m:
out["machine"] = dict(cpu=m.group(1).strip(), cores=int(m.group(2)), backend=m.group(3))
parsed.add("machine")
m = CONFIG_RE.search(blob)
if m:
out["config_str"] = m.group(1).strip(); parsed.add("config")
# load banner: load time, resident dense MB, layers, experts, MTP status.
m = LOAD_BANNER_RE.search(stdout)
if m:
out["load"] = dict(zip(
("load_s", "resident_dense_mb", "layers", "experts", "mtp_status", "draft"),
(float(m.group(1)), float(m.group(2)), int(m.group(3)),
int(m.group(4)), m.group(5), int(m.group(6)))))
parsed.add("load")
# LOOKAHEAD routing-recall block (LOOKA=1): list of {predictor, pct, hit, tot}.
la_block = re.search(
r"LOOKAHEAD routing.*?recall.*?:\n((?:^\s+.+?\s+[0-9.]+%\s+\(\d+/\d+\)\s*$\n?)+)",
blob, re.MULTILINE)
if la_block:
out["lookahead"] = []
for row in LOOKAHEAD_RE.finditer(la_block.group(1)):
out["lookahead"].append(dict(
predictor=row.group(1).strip(), pct=float(row.group(2)),
hit=int(row.group(3)), tot=int(row.group(4))))
parsed.add("lookahead")
cuda = dict(enabled=False, expert_count=None, expert_gb=None,
calls_served=None, resident_tensors=None, resident_gb=None,
groups=None, groups_timing=None)
if CUDA_DEVICE_RE.search(stderr):
cuda["enabled"] = True
m = CUDA_TIER_RE.search(stdout)
if m:
cuda["expert_count"] = int(m.group(1))
cuda["expert_gb"] = float(m.group(2))
cuda["calls_served"] = int(m.group(3))
m = CUDA_RESIDENT_RE.search(stderr)
if m:
cuda["resident_tensors"] = int(m.group(1))
cuda["resident_gb"] = float(m.group(2))
m = CUDA_GROUPS_RE.search(stderr)
if m:
cuda["groups"] = dict(zip(
("calls", "experts", "rows", "experts_per_call"),
(int(m.group(1)), int(m.group(2)), int(m.group(3)), float(m.group(4)))))
m = CUDA_GROUPS_TIME_RE.search(stderr)
if m:
cuda["groups_timing"] = dict(zip(
("h2d_ms", "kernel_ms", "d2h_ms"),
(float(m.group(1)), float(m.group(2)), float(m.group(3)))))
out["cuda"] = cuda
m = TF_MATCH_RE.search(stdout)
if m:
out["tf_match"] = (int(m.group(1)), int(m.group(2))); parsed.add("tf_match")
# Capture per-position argmax divergences from the oracle, keyed by position.
mismatches = {int(mm.group(1)): int(mm.group(2))
for mm in TF_MISMATCH_RE.finditer(blob)}
if out["tf_match"] is not None:
out["tf_mismatches"] = mismatches
parsed.add("tf_mismatches")
out["parsed"] = parsed
return out
def run_engine(
env_overlay: dict,
*,
engine: Optional[str] = None,
cap: int = 4,
ebits: int = 4,
dbits: int = 4,
timeout: float = 600.0,
snap: Optional[str] = None,
) -> tuple[dict, subprocess.CompletedProcess]:
"""Run the engine with an env overlay; return (parsed_telemetry, proc).
`engine` defaults to ./glm.exe (colibri's Windows host). `snap` defaults to
the bundled tiny model (glm_tiny) so callers can omit it for fast tests.
The positional argv is `cap ebits dbits`, matching the engine's main().
"""
if engine is None:
engine = str(Path(__file__).resolve().parent.parent / "glm.exe")
env = os.environ.copy()
# Strip CUDA vars by default so a CPU run isn't accidentally GPU-accelerated
# by a leftover env; callers opt in by passing them in env_overlay.
for k in ("COLI_CUDA", "COLI_GPU", "COLI_GPUS", "CUDA_DENSE", "CUDA_EXPERT_GB"):
env.pop(k, None)
env.update(env_overlay)
if snap is not None:
env["SNAP"] = snap
elif "SNAP" not in env:
env["SNAP"] = str(Path(__file__).resolve().parent.parent / "glm_tiny")
proc = subprocess.run(
[engine, str(cap), str(ebits), str(dbits)],
env=env, capture_output=True, text=True, timeout=timeout,
)
telemetry = parse_run(proc.stdout, proc.stderr)
telemetry["returncode"] = proc.returncode
telemetry["env"] = {k: env_overlay[k] for k in env_overlay}
return telemetry, proc
def disk_wait_share(t: dict) -> Optional[float]:
"""Fraction of decode wall-time spent waiting on expert disk reads.
Preferred source: [PROF] time_shares (the engine's own accounting, which
separates felt-wait from read-service). Falls back to PROFILE disk / sum if
[PROF] wasn't emitted (PROF=0 runs). None if neither is available.
"""
if t.get("time_shares"):
return t["time_shares"]["io"]
if t.get("profile") and t.get("profile_sum"):
return t["profile"]["disk"] / t["profile_sum"]
return None
def tf_agreement(cpu: dict, cuda: dict, oracle: list[int]) -> tuple[float, list[int]]:
"""Direct CPU-vs-CUDA argmax agreement on the TF fixture.
Both runs prefilled the SAME oracle sequence; tf_mismatches holds each
backend's actual argmax where it diverged from the oracle. Where a backend
is ABSENT from the mismatch map, its prediction equals the oracle token at
that position. So the reconstructed per-position prediction is:
oracle[i] if i not in mismatches else mismatches[i]
and agreement is the fraction of positions where CPU and CUDA predictions
are identical — independent of how each relates to the oracle.
`oracle` is ref_glm.json's tf_pred (the per-position oracle argmax). Pass
n_positions = len(oracle).
Returns (agreement_fraction, list_of_differing_positions).
"""
cm = cpu.get("tf_mismatches") or {}
gm = cuda.get("tf_mismatches") or {}
diff = []
for i, orc in enumerate(oracle):
cpu_tok = cm.get(i, orc) # matched oracle => oracle token
cuda_tok = gm.get(i, orc)
if cpu_tok != cuda_tok:
diff.append(i)
agree = (len(oracle) - len(diff)) / len(oracle) if oracle else 0.0
return agree, diff