diff --git a/.github/workflows/collectivex-sweep.yml b/.github/workflows/collectivex-sweep.yml index cb9cc5b1cb..702ad39a48 100644 --- a/.github/workflows/collectivex-sweep.yml +++ b/.github/workflows/collectivex-sweep.yml @@ -224,6 +224,11 @@ jobs: || { echo "CollectiveX allocation cleanup failed; results may still be changing" >&2; exit 1; } python3 experimental/CollectiveX/summarize.py \ --results-dir experimental/CollectiveX/results >> "$GITHUB_STEP_SUMMARY" || true + # Second view over the same documents: latency answers how long a point took, + # this answers how fast it moved its payload. Both are renderers over the staged + # JSON and neither gates the leg, so both are `|| true`. + python3 experimental/CollectiveX/bandwidth.py \ + --results-dir experimental/CollectiveX/results >> "$GITHUB_STEP_SUMMARY" || true shopt -s nullglob results=(experimental/CollectiveX/results/*.json) if [ "${#results[@]}" -eq 0 ]; then diff --git a/experimental/CollectiveX/bandwidth.py b/experimental/CollectiveX/bandwidth.py new file mode 100644 index 0000000000..5d6833d810 --- /dev/null +++ b/experimental/CollectiveX/bandwidth.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +"""Per-component EP bandwidth, computed purely as a consumer of case-attempt artifacts. + +Two views over the bytes + latency every record already emits — no harness or GPU change, +and this gates nothing (see docs/methodology.md): + + A. effective bandwidth -- per-GPU GB/s at each ladder point: bytes / component latency. + B. alpha/beta fit -- OLS of p50 latency vs bytes, latency ~= alpha + bytes/beta, which + separates the bandwidth term from the per-call floor that dominates small T. + +Bytes are LOGICAL payload (one copy per unique (token, dest-rank) pair), excluding +protocol/padding. Reading them against a link's peak needs two adjustments: logical bytes +include the copies a rank routes to itself, which never cross the interconnect, so wire +traffic is (1 - routing.locality.local_rank_fraction) x these figures; and beta is a +MARGINAL rate, so it legitimately sits above every measured point. + +A line through a ladder is a MODEL and a positive slope alone is not evidence, so every Fit +carries its own quality and beta is withheld unless it clears both gates below. Deliberately +NOT added: a per-SKU peak table to check beta against; that hardware knowledge belongs in +platform_config.json, ages badly, and would mislead in its own right. +""" +from __future__ import annotations + +import argparse +import os +import sys +from typing import NamedTuple + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from summarize import load_results # noqa: E402 (sibling consumer, stdlib-only) + +# stage moves nothing (bytes always 0); isolated_sum is a derived percentile sum, not a real +# chained rate. Only these three carry a measured latency + bytes. +COMPONENTS = ("dispatch", "combine", "roundtrip") + +# Below this the line is not describing the ladder: two rungs of noise once yielded 1018 GB/s +# per GPU on a B200 -- above the link's per-direction peak -- at R2 = 0.29, printed +# indistinguishably from an R2 = 0.99 fit. A judgement call; good fits observed at 0.99+. +FIT_MIN_R2 = 0.9 +# R2 alone is NOT enough -- a latency-bound dispatch ladder came back at 3763 GB/s with +# R2 = 0.92, because alpha explained nearly everything and the slope was not identifiable. +# So also require the transfer term to be a real share of the top rung's latency; below this +# the honest reading is "not bandwidth-limited here", not "very fast". +FIT_MIN_BANDWIDTH_SHARE = 0.25 +# If the smallest rung is not within this fraction of the largest, alpha (the intercept at +# zero bytes) is an extrapolation, not a measured floor -- prefill ladders start at T=1024. +ALPHA_NEAR_ZERO_FRACTION = 0.1 +MIN_FIT_POINTS = 3 + + +class Fit(NamedTuple): + """A fitted ladder. `r2`, `points` and `bandwidth_share` are part of the result, not + diagnostics: a caller that ignores them can print an impossible bandwidth.""" + alpha_us: float + beta_gbps: float + r2: float + points: int + alpha_extrapolated: bool + excluded_rows: int + bandwidth_share: float # transfer term / largest measured latency, at the top rung + + @property + def beta_is_reliable(self) -> bool: + return self.r2 >= FIT_MIN_R2 and self.bandwidth_share >= FIT_MIN_BANDWIDTH_SHARE + + +def _ep(document: dict) -> int: + return int(document["identity"]["case_factors"]["case"]["ep"]) + + +def _algbw_per_gpu(total_logical_bytes: int, latency_us: float, ep: int) -> float | None: + """Per-GPU effective GB/s, or None when the latency cannot yield a rate. Bytes are the + AGGREGATE world payload (routed_copies, routing.py), hence the divide by EP size.""" + if latency_us <= 0: + return None + return total_logical_bytes / (latency_us * 1e-6) / 1e9 / ep + + +def _fit_line(xs: list[float], ys: list[float]) -> tuple[float, float, float]: + """(intercept, slope, r2) of a 2-parameter least-squares fit (stdlib only).""" + n = len(xs) + mean_x, mean_y = sum(xs) / n, sum(ys) / n + slope = (sum((x - mean_x) * (y - mean_y) for x, y in zip(xs, ys)) + / sum((x - mean_x) ** 2 for x in xs)) + intercept = mean_y - slope * mean_x + ss_tot = sum((y - mean_y) ** 2 for y in ys) + ss_res = sum((y - (intercept + slope * x)) ** 2 for x, y in zip(xs, ys)) + # ss_tot == 0: no variance to explain, so the fit cannot be credited with anything. + return intercept, slope, (1.0 - ss_res / ss_tot) if ss_tot else 0.0 + + +def fit_alpha_beta(document: dict, component: str, pct: str = "p50") -> Fit | None: + """Fit one component's ladder, or None when no fit is defensible at all. + + Regress against the emitted bytes, never T (routed_copies is sub-linear in T at high + fan-out); p50 by default, since p99 carries tail noise. Rungs that failed the correctness + gate are excluded — a corrupt payload is not a measurement of how fast correct data moves. + """ + usable, excluded = [], 0 + for row in document["measurement"]["rows"]: + percentiles = row["components"][component]["percentiles_us"] + if not percentiles: + continue + if not row.get("correctness", {}).get("passed", True): + excluded += 1 + continue + usable.append((row["byte_provenance"][component]["total_logical_bytes"], + percentiles[pct])) + if len(usable) < MIN_FIT_POINTS or len({x for x, _ in usable}) < 2: + return None # too few points, or zero-variance x (e.g. a degenerate ladder) + xs, ys = [x for x, _ in usable], [y for _, y in usable] + alpha_us, slope, r2 = _fit_line(xs, ys) + if slope <= 0: + return None # anti-correlated / noise-dominated: no meaningful bandwidth term + return Fit( # slope is us per aggregate byte; 1e-3/slope = aggregate GB/s; /ep = per-GPU + alpha_us=alpha_us, + beta_gbps=1e-3 / slope / _ep(document), + r2=r2, + points=len(usable), + alpha_extrapolated=min(xs) > ALPHA_NEAR_ZERO_FRACTION * max(xs), + excluded_rows=excluded, + bandwidth_share=(slope * max(xs) / max(ys)) if max(ys) > 0 else 0.0, + ) + + +def _format_fit(component: str, fit: Fit) -> str: + """One component's fit, withholding what the data does not support.""" + if not fit.beta_is_reliable: + why = "poor fit" if fit.r2 < FIT_MIN_R2 else "not bandwidth-bound" + return (f"{component}: beta=unreliable [{why}] " + f"(R2={fit.r2:.2f}, transfer share={fit.bandwidth_share:.2f}, n={fit.points})") + alpha = f"alpha={fit.alpha_us:6.1f}us" + ("*" if fit.alpha_extrapolated else " ") + return (f"{component}: beta={fit.beta_gbps:6.1f} GB/s {alpha} " + f"(R2={fit.r2:.3f}, n={fit.points})") + + +def _cell(row: dict, component: str, ep: int) -> str: + percentiles = row["components"][component]["percentiles_us"] + if not percentiles: + return f"{component}=n/a" + nbytes = row["byte_provenance"][component]["total_logical_bytes"] + p50 = _algbw_per_gpu(nbytes, percentiles["p50"], ep) + p99 = _algbw_per_gpu(nbytes, percentiles["p99"], ep) + return f"{component}=n/a" if p50 is None or p99 is None \ + else f"{component}={p50:6.1f}/{p99:<6.1f}" + + +def _sort_key(document: dict): + case = document["identity"]["case_factors"]["case"] + return (document["identity"]["case_factors"]["sku"], case["backend"], + case["precision"], case["phase"], case["ep"], + document.get("generated_at", "")) + + +def _provenance(document: dict) -> str: + """Enough identity to tell two attempts of one case apart — CI result directories + routinely hold both, and they can disagree.""" + run_id = document["identity"].get("allocation_factors", {}).get("run_id") + attempt = document["identity"].get("attempt_ordinal") + parts = ((document.get("generated_at") or "")[:19], + f"run {run_id}" if run_id else "", f"attempt {attempt}" if attempt else "") + return " · ".join(p for p in parts if p) + + +def render(documents: list[dict]) -> str: + lines = [ + "## CollectiveX EP bandwidth (per-GPU, logical payload)", + "", + "GB/s at p50/p99 *latency* -- the p99-latency figure is worst-case bandwidth, not a " + f"'p99 bandwidth'. `fit`: latency ~= alpha + bytes/beta over the ladder; beta is " + f"withheld below R2 {FIT_MIN_R2} or when the ladder never became bandwidth-bound, `*` " + "marks an extrapolated alpha, and rungs failing the correctness gate are excluded.", + "", + ] + for document in sorted(documents, key=_sort_key): + case = document["identity"]["case_factors"]["case"] + ep = _ep(document) + lines.append( + f"### {document['identity']['case_factors']['sku']} `{case['backend']}` " + f"{case['precision']} {case['phase']} ep{ep} — {document['outcome']['status']}" + ) + if provenance := _provenance(document): + lines.append(f" ({provenance})") + for row in document["measurement"]["rows"]: + cross_node = (row["routing"].get("locality") or {}).get("cross_node_fraction") + suffix = f" xnode={cross_node * 100:4.0f}%" if cross_node is not None else "" + if not row.get("correctness", {}).get("passed", True): + suffix += " [correctness FAILED]" + cells = " ".join(_cell(row, component, ep) for component in COMPONENTS) + lines.append(f" T={row['tokens_per_rank']:<5} {cells}{suffix}") + fits = [(c, f) for c in COMPONENTS if (f := fit_alpha_beta(document, c))] + dropped = max((f.excluded_rows for _, f in fits), default=0) + note = f" (excluded {dropped} gate-failed rung(s))" if dropped else "" + lines.append(" fit " + (" ".join(_format_fit(c, f) for c, f in fits) + if fits else "insufficient points") + note) + lines.append("") + if not documents: + lines.append("> No case-attempt documents found.") + return "\n".join(lines) + + +def main() -> int: + parser = argparse.ArgumentParser( + description="CollectiveX per-component EP bandwidth (pure consumer; gates nothing)") + parser.add_argument("--results-dir", default="results") + parser.add_argument("--runner") + parser.add_argument("--ts") + parser.add_argument("--all", action="store_true", + help="include non-success outcomes (skipped by default)") + args = parser.parse_args() + documents = load_results(args.results_dir, args.runner, args.ts) + if not args.all: + documents = [d for d in documents if d["outcome"]["status"] == "success"] + print(render(documents)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experimental/CollectiveX/tests/test_bandwidth.py b/experimental/CollectiveX/tests/test_bandwidth.py new file mode 100644 index 0000000000..3b7380f154 --- /dev/null +++ b/experimental/CollectiveX/tests/test_bandwidth.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +"""Math tests for the bandwidth consumer (unit conversion, alpha/beta fit, fit gating).""" +from __future__ import annotations + +import sys +import unittest +from pathlib import Path + +sys.path[:0] = [str(Path(__file__).resolve().parents[1])] + +import bandwidth # noqa: E402 + +COMPONENTS = bandwidth.COMPONENTS + + +def _row(tokens, nbytes, latency, passed=True): + """A measurement row. `latency` is a scalar, or a per-component dict whose None marks + that component unavailable.""" + lat = latency if isinstance(latency, dict) else dict.fromkeys(COMPONENTS, latency) + return { + "tokens_per_rank": tokens, + "components": {c: {"percentiles_us": None if lat[c] is None else { + "p50": lat[c], "p90": lat[c], "p95": lat[c], "p99": lat[c] * 2.0}} + for c in COMPONENTS}, + "byte_provenance": {c: {"total_logical_bytes": nbytes} for c in COMPONENTS}, + "correctness": {"passed": passed}, + "routing": {"locality": {"cross_node_fraction": 0.5}}, + } + + +def _doc(rows, ep=2): + case = {"ep": ep, "backend": "deepep-v2", "precision": "bf16", "phase": "decode", + "suite": "s", "routing": "uniform"} + return { + "generated_at": "2026-07-25T22:12:55.760511+00:00", + "identity": {"attempt_ordinal": 1, "allocation_factors": {"run_id": "30177021271"}, + "case_factors": {"sku": "h100", "case": case}}, + "measurement": {"rows": rows}, + "outcome": {"status": "success"}, + } + + +def _linear(pairs, passed=True): + """Rows exactly on latency = 10us + bytes * 2e-6, i.e. alpha=10, beta_agg=500 GB/s.""" + return [_row(t, b, 10.0 + b * 2e-6, passed) for t, b in pairs] + + +LADDER = ((8, 1e6), (16, 2e6), (32, 3e6)) + + +class BandwidthMath(unittest.TestCase): + def test_algbw_per_gpu(self): + # 1e9 bytes in 1000us = 1e12 B/s = 1000 GB/s aggregate; /ep(2) = 500 per GPU. + self.assertAlmostEqual(bandwidth._algbw_per_gpu(1e9, 1000.0, 2), 500.0) + self.assertIsNone(bandwidth._algbw_per_gpu(1e9, 0.0, 2)) + + def test_fit_recovers_alpha_beta(self): + fit = bandwidth.fit_alpha_beta(_doc(_linear(LADDER)), "dispatch") + self.assertAlmostEqual(fit.alpha_us, 10.0, places=4) + self.assertAlmostEqual(fit.beta_gbps, 250.0, places=4) # 500 aggregate / ep(2) + self.assertAlmostEqual(fit.r2, 1.0, places=6) + self.assertEqual(fit.points, 3) + self.assertTrue(fit.beta_is_reliable) + + def test_fit_is_none_when_undefensible(self): + flat = [_row(t, b, 12.0) for t, b in LADDER] # slope <= 0 + self.assertIsNone(bandwidth.fit_alpha_beta(_doc(_linear(LADDER[:2])), "dispatch")) + self.assertIsNone(bandwidth.fit_alpha_beta(_doc(flat), "dispatch")) + + def test_noisy_ladder_withholds_beta(self): + # A positive slope through noise still fits; printing its beta once produced a + # physically impossible 1018 GB/s per GPU on a B200 at R2 = 0.29. + rows = [_row(t, b, lat) for t, b, lat in ( + (8, 1e6, 300.0), (16, 2e6, 40.0), (32, 3e6, 260.0), + (64, 4e6, 60.0), (128, 5e6, 320.0))] + fit = bandwidth.fit_alpha_beta(_doc(rows), "dispatch") + self.assertLess(fit.r2, bandwidth.FIT_MIN_R2) + self.assertFalse(fit.beta_is_reliable) + out = bandwidth.render([_doc(rows)]) + self.assertIn("beta=unreliable", out) + self.assertNotIn("GB/s alpha", out) # no number presented as measured + + def test_latency_bound_ladder_withholds_beta_despite_high_r2(self): + # Real data gave beta = 3763 GB/s at R2 = 0.92: a near-zero slope explodes beta while + # the line still fits, so R2 cannot catch it — the transfer-share gate must. + rows = [_row(t, b, 500.0 + b * 1e-9) + for t, b in LADDER + ((64, 4e6), (128, 5e6))] + fit = bandwidth.fit_alpha_beta(_doc(rows), "dispatch") + self.assertGreater(fit.r2, bandwidth.FIT_MIN_R2) + self.assertLess(fit.bandwidth_share, bandwidth.FIT_MIN_BANDWIDTH_SHARE) + self.assertFalse(fit.beta_is_reliable) + self.assertIn("not bandwidth-bound", bandwidth._format_fit("dispatch", fit)) + + def test_alpha_marked_only_when_extrapolated(self): + prefill = bandwidth.fit_alpha_beta( # starts at T=1024: intercept is extrapolated + _doc(_linear(((1024, 1e9), (2048, 2e9), (4096, 4e9), (8192, 8e9)))), "dispatch") + decode = bandwidth.fit_alpha_beta( # reaches near zero bytes: alpha stands + _doc(_linear(((1, 1e5), (64, 6.4e6), (512, 5.12e7)))), "dispatch") + self.assertTrue(prefill.alpha_extrapolated) + self.assertFalse(decode.alpha_extrapolated) + self.assertIn("*", bandwidth._format_fit("dispatch", prefill)) + self.assertNotIn("*", bandwidth._format_fit("dispatch", decode)) + + def test_gate_failed_rung_excluded_from_fit_and_marked(self): + rows = _linear(LADDER) + [_row(64, 4e6, 999.0, passed=False)] + fit = bandwidth.fit_alpha_beta(_doc(rows), "dispatch") + self.assertEqual((fit.points, fit.excluded_rows), (3, 1)) + self.assertAlmostEqual(fit.beta_gbps, 250.0, places=4) # the corrupt rung didn't steer it + out = bandwidth.render([_doc(rows)]) + self.assertIn("[correctness FAILED]", out) + self.assertIn("excluded 1 gate-failed rung", out) + + def test_render_marks_unavailable_and_separates_attempts(self): + unavailable = [_row(t, b, {"dispatch": None, "combine": 5.0, "roundtrip": 6.0}) + for t, b in LADDER] + out = bandwidth.render([_doc(unavailable)]) + self.assertIn("dispatch=n/a", out) + self.assertIn("xnode= 50%", out) + second = _doc(_linear(LADDER)) + second["identity"]["attempt_ordinal"] = 2 + second["generated_at"] = "2026-07-25T23:00:00.000000+00:00" + out = bandwidth.render([_doc(_linear(LADDER)), second]) + self.assertIn("attempt 1", out) + self.assertIn("attempt 2", out) + self.assertIn("run 30177021271", out) + + +if __name__ == "__main__": + unittest.main()