analytical-viz: 'Chip roofline & B*' tab — AI, B*, L*, cost curves

New tab surfaces the arithmetic-intensity story from LLM-serving
practice for the current sidebar chip + model. All derived from
MachineParams + ModelConfig; no new configuration.

chip_roofline module (pure functions):
- arithmetic_intensity      = C / W   (FLOPs per byte HBM BW)
- critical_batch (B*)       = C * b / (2 * W) * sparsity
- balance_context (L*)      = 2 * N_active * W / (C * kv_bpt)
- knee_batch (B_knee)       = B* / (1 - S_kv/L*)  (None past L*)
- per_token_latency_curve   = weight/B + compute + KV, per B
- bound_regime              = which term dominates now

Peak roofline convention (no compute_util factor) so weight_s ==
compute_s exactly at B*. Comm + TP/CP sharding intentionally
excluded — stage_latencies is the full latency model; this is the
back-of-envelope chip-vs-model view.

Tab shows:
- 4 KPI cards: AI, B*, L*, current regime at (B, S_kv)
- MoE hint when preset flags MoE ('300 x sparsity' rule)
- Plot 1: cost vs B at current S_kv (weight, compute floor, KV
  floor, total) with B* marker line
- Plot 2: cost curves at S_kv/L* = 0.25, 0.5, 1, 2, 5 — shows the
  no-knee regime past L*
- Plot 3: B_knee vs S_kv — knee slides right, diverges at L*
- Formulas + interpretation table

test_chip_roofline covers B* against H100 reference (295), sparsity
scaling, L* scaling with HBM BW, knee divergence at L*, per-token
curve monotonicity + asymptote to compute+KV floor, regime
classification. Smoke test guards the tab is wired in app.py.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-07-29 11:59:19 -07:00
parent 5e92b89821
commit 1ddd1baa7a
3 changed files with 582 additions and 1 deletions
@@ -0,0 +1,207 @@
"""Tests for chip_roofline: AI, B*, L*, per-token latency curves."""
from __future__ import annotations
import math
import pytest
from tests.analytical_visualization.chip_roofline import (
arithmetic_intensity,
balance_context,
bound_regime,
critical_batch,
knee_batch,
kv_bytes_per_token,
per_token_latency_curve,
total_active_params,
)
from tests.analytical_visualization.model_config import (
MachineParams, ModelConfig,
)
from tests.analytical_visualization.model_presets import PRESETS
# ── Arithmetic intensity ────────────────────────────────────────────
def test_arithmetic_intensity_matches_ratio():
"""AI = peak_flops / bw_hbm — pure ratio, no util factor."""
m = MachineParams(peak_tflops_f16=8.0, bw_hbm_gbs=256.0)
assert arithmetic_intensity(m) == pytest.approx(8e12 / 256e9)
def test_ai_scales_with_flops_and_bandwidth():
m1 = MachineParams(peak_tflops_f16=8.0, bw_hbm_gbs=256.0)
m2 = MachineParams(peak_tflops_f16=16.0, bw_hbm_gbs=256.0)
m3 = MachineParams(peak_tflops_f16=8.0, bw_hbm_gbs=128.0)
assert arithmetic_intensity(m2) == pytest.approx(2 * arithmetic_intensity(m1))
assert arithmetic_intensity(m3) == pytest.approx(2 * arithmetic_intensity(m1))
# ── Critical batch B* ───────────────────────────────────────────────
def test_critical_batch_h100_reference():
"""H100-class: 989 TFLOPs BF16, 3.35 TB/s HBM3 → B* ≈ 295."""
m = MachineParams(peak_tflops_f16=989.0, bw_hbm_gbs=3350.0)
model = PRESETS["Llama 3 8B"].model # bf16 (b=2)
b_star = critical_batch(m, model)
assert b_star == pytest.approx(295, rel=0.05), b_star
def test_critical_batch_scales_with_sparsity():
"""MoE 8× sparsity gives 8× B*."""
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
b_dense = critical_batch(m, model, sparsity=1)
b_moe8 = critical_batch(m, model, sparsity=8)
assert b_moe8 == pytest.approx(8 * b_dense)
def test_critical_batch_bf16_formula():
"""For BF16 (b=2), B* = AI in flops-per-byte units → numerically."""
m = MachineParams(peak_tflops_f16=8.0, bw_hbm_gbs=256.0)
model = ModelConfig(bytes_per_elem=2)
ai = arithmetic_intensity(m)
assert critical_batch(m, model) == pytest.approx(ai), (
critical_batch(m, model), ai,
)
# ── Balance context L* ──────────────────────────────────────────────
def test_balance_context_positive_finite():
"""L* for a real model on real machine is a positive finite number."""
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
l_star = balance_context(m, model)
assert 0 < l_star < 1e9, l_star
def test_balance_context_scales_with_bandwidth():
"""L* = 2N*W/(C*kv_bpt): doubling HBM BW doubles L*.
Slower memory hits the KV wall at a shorter context. Machine-only
change so N_active and kv_bpt stay fixed."""
fast = MachineParams(bw_hbm_gbs=512.0)
slow = MachineParams(bw_hbm_gbs=256.0)
model = PRESETS["Llama 3 8B"].model
assert (balance_context(fast, model)
== pytest.approx(2 * balance_context(slow, model)))
# ── Knee ────────────────────────────────────────────────────────────
def test_knee_equals_b_star_at_short_context():
"""S_kv → 0: B_knee → B*."""
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
b_star = critical_batch(m, model)
assert knee_batch(m, model, s_kv=1) == pytest.approx(b_star, rel=1e-3)
def test_knee_diverges_at_balance_context():
"""S_kv = L*: knee is infinite; S_kv > L*: no knee (None)."""
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
l_star = balance_context(m, model)
assert knee_batch(m, model, s_kv=int(2 * l_star)) is None
# At 0.9 * L*, knee ~= 10 * B*; assert at least 5x to survive rounding.
below = knee_batch(m, model, s_kv=int(0.9 * l_star))
assert below is not None and below > 5 * critical_batch(m, model)
def test_knee_slides_right_with_context():
"""S_kv = L*/2 → B_knee = 2 * B*."""
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
b_star = critical_batch(m, model)
l_star = balance_context(m, model)
got = knee_batch(m, model, s_kv=int(l_star / 2))
assert got == pytest.approx(2 * b_star, rel=0.01), (got, 2 * b_star)
# ── Per-token latency curve ─────────────────────────────────────────
def test_latency_curve_monotonically_decreasing():
"""total_s must strictly decrease as batch increases (weight/B shrinks)."""
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
pts = per_token_latency_curve(m, model, [1, 2, 4, 8, 16, 32, 64],
s_kv=1024)
totals = [p.total_s for p in pts]
for a, b in zip(totals, totals[1:]):
assert b < a, (a, b)
def test_latency_curve_asymptotes_to_compute_plus_kv():
"""At very large B, total → compute + KV (weight/B → 0)."""
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
pts = per_token_latency_curve(m, model, [10_000_000], s_kv=1024)
p = pts[0]
assert p.total_s == pytest.approx(p.compute_s + p.kv_s, rel=1e-3)
def test_latency_at_b_star_is_roughly_2x_floor():
"""At B*, weight = compute (dense, S_kv small), so total ≈ 2 * compute."""
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
b_star = int(round(critical_batch(m, model)))
pts = per_token_latency_curve(m, model, [b_star], s_kv=1)
p = pts[0]
# weight_s ≈ compute_s at B*
assert p.weight_s == pytest.approx(p.compute_s, rel=0.05), (
p.weight_s, p.compute_s,
)
# ── Regime classification ───────────────────────────────────────────
def test_regime_memory_bound_at_b1():
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
assert bound_regime(m, model, batch=1, s_kv=1024) == "memory-bound"
def test_regime_kv_bound_past_balance_context():
"""S_kv > L* with reasonable batch → KV dominates."""
m = MachineParams()
model = PRESETS["Llama 3 8B"].model
l_star = balance_context(m, model)
b_star = int(round(critical_batch(m, model)))
assert bound_regime(m, model, batch=b_star * 4,
s_kv=int(3 * l_star)) == "kv-bound"
# ── Component sanity ────────────────────────────────────────────────
def test_total_active_params_llama3_8b_ballpark():
"""Llama 3 8B — attn+FFN alone (no embeddings/LM head) is ~6.9B,
HF-reported total 8.03B. Guard the ballpark."""
n = total_active_params(PRESETS["Llama 3 8B"].model)
assert 6.5e9 < n < 8e9, n
def test_kv_bytes_per_token_llama3_8b():
"""Llama 3 8B: 2*8*128*2 bytes/layer/token × 32 layers = 131072 bytes."""
kv_bpt = kv_bytes_per_token(PRESETS["Llama 3 8B"].model)
assert kv_bpt == 2 * 8 * 128 * 2 * 32
# ── App wiring: tab exists on the Streamlit app ────────────────────
def test_roofline_tab_registered_in_app():
"""The new 'Chip roofline & B*' tab is listed in st.tabs and gets a
corresponding `with tab_roofline:` block."""
from pathlib import Path
src = (Path(__file__).parent / "app.py").resolve().read_text(
encoding="utf-8"
)
assert src.count('"Chip roofline & B*"') == 1
assert "with tab_roofline:" in src