kernbench2/tests/attention/test_milestone_gqa_prefill_long_ctx_4cases.py

"""Tests for the long-context prefill 4-cases comparative-study bench.

Prefill counterpart to test_milestone_gqa_decode_long_ctx_4cases. The
4 cases differ in how KV cache is sharded across the 8 cubes and 8
PEs of a single KV-head group on LLaMA-3.1-70B GQA:

    Case 1  Cube-SP / PE-TP  → KV split by S_kv across cubes (Ring);
                               Q/O split T_q across all 64 ranks
    Case 2  Cube-Repl / PE-TP → full KV per cube; only CUBE 0's P PEs
                               split T_q (CUBEs 1..7 = pure memory waste)
    Case 3  Cube-Repl / PE-SP → full KV per cube; PEs SP on S_kv
                               (intra-cube AR); 8× redundant compute
    Case 4  Cube-SP / PE-SP  → KV split 64-way + Q split T_q across cubes;
                               Ring KV + intra-cube AR per cube  ★ optimal

Smoke sizes (T_q=256, S_kv=2048) keep the per-PE scratch comfortably
inside budget; the headline T_q=S_kv=4096 runs come from
``kernbench run --bench milestone-gqa-prefill-long-ctx-4cases``, not
pytest.
"""
from __future__ import annotations

import re
from pathlib import Path

from kernbench.runtime_api.bench_runner import run_bench
from kernbench.runtime_api.types import resolve_device
from kernbench.sim_engine.engine import GraphEngine
from kernbench.topology.builder import resolve_topology

TOPOLOGY_DEFAULT = Path(__file__).resolve().parents[2] / "topology.yaml"

_CASE4_PANEL = "single_kv_group_prefill_long_ctx_gqa_cube_sp_pe_sp"
_CASE3_PANEL = "single_kv_group_prefill_long_ctx_gqa_cube_repl_pe_sp"
_CASE2_PANEL = "single_kv_group_prefill_long_ctx_gqa_cube_repl_pe_tp"
_CASE1_PANEL = "single_kv_group_prefill_long_ctx_gqa_cube_sp_pe_tp"
_CUBE_RE = re.compile(r"\bcube(\d+)\b")

# Smoke sizes: T_q=256 (divisible by C·P=64 for Case 1, by P=8 for
# Case 2, by C=8 for Case 4) and S_kv=2048 (≥ 2·TILE_S_KV so the
# per-tile fold loop runs; Case 1/4 have S_local sufficient to exercise
# the Ring). Assertions are size-independent. The headline T_q=S_kv=4096
# bench runs come from the env-gated bench, not pytest.
# Smoke sizes — kept small so pytest stays fast:
#   T_q=64  : minimum that satisfies Case 1's ``T_q % (C·P) == 0``.
#   S_kv=512: enough to exercise multi-tile fold for Cases 2/3
#             (TILE_S_KV=64 → 8 tiles for Case 2; 1 tile/PE for Case 3
#             after intra-cube S_kv split) without burning minutes on
#             Case 2's single-rank serial path.
# Cases 2/3 also use TILE_S_KV=64 in their kernels to keep the
# persistent ``scores`` tensor inside the 1 MB scratch budget at
# T_q=64. Headline T_q=S_kv=4096 runs come from the env-gated bench,
# not pytest.
_SMOKE_T_Q = 64
_SMOKE_S_KV = 512

_HEADLINE_T_Q = 512
_HEADLINE_S_KV = 512


def _engine_factory(t, d):
    return GraphEngine(getattr(t, "topology_obj", t), enable_data=True)


def _count(op_log, name: str) -> int:
    return sum(1 for r in op_log if r.op_name == name)


def _dma_write_cubes(op_log) -> list[int]:
    cubes: list[int] = []
    for r in op_log:
        if r.op_name != "dma_write":
            continue
        m = _CUBE_RE.search(r.component_id)
        if m is not None:
            cubes.append(int(m.group(1)))
    return cubes


# ── Case 4 (Cube-SP × PE-SP) — ★ optimal ────────────────────────────


def _run_case4_smoke(*, T_q: int, S_kv: int):
    """Drive the Case 4 prefill panel via the case-specific runner."""
    from kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases import (
        _run_prefill_panel_long_ctx_cube_sp_pe_sp,
    )
    topo = resolve_topology(str(TOPOLOGY_DEFAULT))

    def _bench_fn(ctx):
        _run_prefill_panel_long_ctx_cube_sp_pe_sp(
            ctx, panel=_CASE4_PANEL,
            C=8, P=8,
            T_q=T_q, S_kv=S_kv,
            d_head=128, h_q=8, h_kv=1,
        )

    return run_bench(
        topology=topo, bench_fn=_bench_fn,
        device=resolve_device(None),
        engine_factory=_engine_factory,
    )


def test_case4_panel_registered():
    """Case 4 panel must be in ``_PANELS`` + ``_PANEL_DISPATCH`` with
    the LLaMA-3.1-70B single-KV-group prefill dims (T_q=S_kv=4096).
    """
    from kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases import (
        _PANEL_DISPATCH,
        _PANELS,
    )
    assert _CASE4_PANEL in _PANELS, f"{_CASE4_PANEL!r} not in {_PANELS}"
    assert _CASE4_PANEL in _PANEL_DISPATCH
    kind, params = _PANEL_DISPATCH[_CASE4_PANEL]
    assert kind == "prefill_long_ctx_cube_sp_pe_sp"
    assert params == {
        "C": 8, "P": 8,
        "T_q": _HEADLINE_T_Q, "S_kv": _HEADLINE_S_KV,
        "d_head": 128, "h_q": 8, "h_kv": 1,
    }


def test_case4_runner_smoke():
    """Case 4 runner drives the new kernel to completion at smoke sizes."""
    result = _run_case4_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok, (
        f"Case 4 prefill smoke at C=8 P=8 must complete; "
        f"got {result.completion}"
    )


def test_case4_writers_one_per_cube():
    """Case 4 prefill: each cube writes its own disjoint T_q/C rows
    (PE 0 of each cube). Output cubes must cover all 8.
    """
    result = _run_case4_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok
    cubes = _dma_write_cubes(result.engine.op_log)
    assert cubes, "expected at least one dma_write for the final O store"
    distinct = set(cubes)
    assert distinct == set(range(8)), (
        f"Case 4 expected PE 0 of every cube to write; got cubes={sorted(distinct)}"
    )


def test_case4_ipcq_copy_positive():
    """Case 4 prefill has both inter-CUBE Ring and intra-CUBE AR
    traffic; total ipcq_copy must be strictly positive.
    """
    result = _run_case4_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok
    n_copy = _count(result.engine.op_log, "ipcq_copy")
    assert n_copy > 0, f"Case 4 must have Ring + intra-CUBE comm; got 0 ipcq_copy"


# ── Case 2 (Cube-Repl × PE-TP) ──────────────────────────────────────


def _run_case2_smoke(*, T_q: int, S_kv: int):
    """Drive the Case 2 prefill panel via the case-specific runner."""
    from kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases import (
        _run_prefill_panel_long_ctx_cube_repl_pe_tp,
    )
    topo = resolve_topology(str(TOPOLOGY_DEFAULT))

    def _bench_fn(ctx):
        _run_prefill_panel_long_ctx_cube_repl_pe_tp(
            ctx, panel=_CASE2_PANEL,
            C=8, P=8,
            T_q=T_q, S_kv=S_kv,
            d_head=128, h_q=8, h_kv=1,
        )

    return run_bench(
        topology=topo, bench_fn=_bench_fn,
        device=resolve_device(None),
        engine_factory=_engine_factory,
    )


def test_case2_panel_registered():
    """Case 2 panel registered with single-KV-group prefill dims."""
    from kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases import (
        _PANEL_DISPATCH,
        _PANELS,
    )
    assert _CASE2_PANEL in _PANELS, f"{_CASE2_PANEL!r} not in {_PANELS}"
    assert _CASE2_PANEL in _PANEL_DISPATCH
    kind, params = _PANEL_DISPATCH[_CASE2_PANEL]
    assert kind == "prefill_long_ctx_cube_repl_pe_tp"
    assert params == {
        "C": 8, "P": 8,
        "T_q": _HEADLINE_T_Q, "S_kv": _HEADLINE_S_KV,
        "d_head": 128, "h_q": 8, "h_kv": 1,
    }


def test_case2_runner_smoke():
    """Case 2 runner drives the new kernel to completion at smoke sizes."""
    result = _run_case2_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok, (
        f"Case 2 prefill smoke at C=8 P=8 must complete; "
        f"got {result.completion}"
    )


def test_case2_zero_ipcq_copy_no_comm():
    """Case 2's defining property: full KV per rank ⇒ NO inter-rank
    communication. Slide 11 lists comm cost as 'none'.
    """
    result = _run_case2_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok
    n_copy = _count(result.engine.op_log, "ipcq_copy")
    assert n_copy == 0, (
        f"Case 2 must have zero inter-rank comm; got ipcq_copy={n_copy}"
    )


def test_case2_writers_only_cube_0():
    """Case 2: only CUBE 0 active (PEs 0..P-1 each writing their T_q/P
    rows). All dma_writes come from cube 0.
    """
    result = _run_case2_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok
    cubes = _dma_write_cubes(result.engine.op_log)
    assert cubes, "expected at least one dma_write for the final O store"
    distinct = set(cubes)
    assert distinct == {0}, (
        f"Case 2 writers must be cube 0 only; got cubes={sorted(distinct)}"
    )


# ── Case 3 (Cube-Repl × PE-SP) ──────────────────────────────────────


def _run_case3_smoke(*, T_q: int, S_kv: int):
    """Drive the Case 3 prefill panel via the case-specific runner."""
    from kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases import (
        _run_prefill_panel_long_ctx_cube_repl_pe_sp,
    )
    topo = resolve_topology(str(TOPOLOGY_DEFAULT))

    def _bench_fn(ctx):
        _run_prefill_panel_long_ctx_cube_repl_pe_sp(
            ctx, panel=_CASE3_PANEL,
            C=8, P=8,
            T_q=T_q, S_kv=S_kv,
            d_head=128, h_q=8, h_kv=1,
        )

    return run_bench(
        topology=topo, bench_fn=_bench_fn,
        device=resolve_device(None),
        engine_factory=_engine_factory,
    )


def test_case3_panel_registered():
    """Case 3 panel registered with single-KV-group prefill dims."""
    from kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases import (
        _PANEL_DISPATCH,
        _PANELS,
    )
    assert _CASE3_PANEL in _PANELS, f"{_CASE3_PANEL!r} not in {_PANELS}"
    assert _CASE3_PANEL in _PANEL_DISPATCH
    kind, params = _PANEL_DISPATCH[_CASE3_PANEL]
    assert kind == "prefill_long_ctx_cube_repl_pe_sp"
    assert params == {
        "C": 8, "P": 8,
        "T_q": _HEADLINE_T_Q, "S_kv": _HEADLINE_S_KV,
        "d_head": 128, "h_q": 8, "h_kv": 1,
    }


def test_case3_runner_smoke():
    """Case 3 runner drives the new kernel to completion at smoke sizes."""
    result = _run_case3_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok, (
        f"Case 3 prefill smoke at C=8 P=8 must complete; "
        f"got {result.completion}"
    )


def test_case3_intra_cube_ar_only_ipcq():
    """Case 3 reduce pattern: per-CUBE 8-way PE-SP AR (same structural
    cost as decode Case 3's intra-CUBE phase = 21 ipcq_copy per cube),
    and NO inter-CUBE traffic (each cube has a full copy of KV).

    Q-axis tiling tradeoff: prefill Case 3 wraps the intra-CUBE AR
    INSIDE the outer Q-tile loop (the kernel can't hold full
    (m, ℓ, O) for the whole T_q in scratch). So the AR runs once
    per Q-tile → total ipcq = n_q_tiles × per-tile cost.

      per-CUBE intra (2×4 PE grid), PER Q-TILE:
        row chain along intra_W: cols 1,2,3 each row × 2 rows ×
          3 tensors (m, ℓ, O)                          = 18
        col bridge along intra_N: pe4 only × 3 tensors =  3
        per-CUBE intra total                           = 21
        × 8 CUBEs                                      = 168
        × n_q_tiles (= G·T_q / TILE_Q = 8·64/64 = 8)   = 1344

      inter-CUBE: 0 (replicated KV ⇒ no AllReduce needed).
    """
    from kernbench.benches.gqa_helpers.long_ctx._gqa_attention_prefill_long_ctx_cube_repl_pe_sp import (
        TILE_Q,
    )

    result = _run_case3_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok
    G = 8  # h_q / h_kv
    n_q_tiles = (G * _SMOKE_T_Q + TILE_Q - 1) // TILE_Q
    expected = n_q_tiles * 168
    n_copy = _count(result.engine.op_log, "ipcq_copy")
    assert n_copy == expected, (
        f"Case 3 expected {expected} ipcq_copy "
        f"({n_q_tiles} Q-tiles × 168 per Q-tile intra-CUBE AR); "
        f"got {n_copy}"
    )


def test_case3_single_dma_write_at_cube_0():
    """Every cube ends with the full answer after intra-CUBE AR; only
    the designated writer (cube 0, PE 0) stores O.
    """
    result = _run_case3_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok
    cubes = _dma_write_cubes(result.engine.op_log)
    assert cubes, "expected at least one dma_write for the final O store"
    distinct = set(cubes)
    assert distinct == {0}, (
        f"Case 3 designated writer must be cube 0; got cubes={sorted(distinct)}"
    )


# ── Case 1 (Cube-SP × PE-TP) ────────────────────────────────────────


def _run_case1_smoke(*, T_q: int, S_kv: int):
    """Drive the Case 1 prefill panel via the case-specific runner."""
    from kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases import (
        _run_prefill_panel_long_ctx_cube_sp_pe_tp,
    )
    topo = resolve_topology(str(TOPOLOGY_DEFAULT))

    def _bench_fn(ctx):
        _run_prefill_panel_long_ctx_cube_sp_pe_tp(
            ctx, panel=_CASE1_PANEL,
            C=8, P=8,
            T_q=T_q, S_kv=S_kv,
            d_head=128, h_q=8, h_kv=1,
        )

    return run_bench(
        topology=topo, bench_fn=_bench_fn,
        device=resolve_device(None),
        engine_factory=_engine_factory,
    )


def test_case1_panel_registered():
    """Case 1 panel registered with single-KV-group prefill dims."""
    from kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases import (
        _PANEL_DISPATCH,
        _PANELS,
    )
    assert _CASE1_PANEL in _PANELS, f"{_CASE1_PANEL!r} not in {_PANELS}"
    assert _CASE1_PANEL in _PANEL_DISPATCH
    kind, params = _PANEL_DISPATCH[_CASE1_PANEL]
    assert kind == "prefill_long_ctx_cube_sp_pe_tp"
    assert params == {
        "C": 8, "P": 8,
        "T_q": _HEADLINE_T_Q, "S_kv": _HEADLINE_S_KV,
        "d_head": 128, "h_q": 8, "h_kv": 1,
    }


def test_case1_runner_smoke():
    """Case 1 runner drives the new kernel to completion at smoke sizes."""
    result = _run_case1_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok, (
        f"Case 1 prefill smoke at C=8 P=8 must complete; "
        f"got {result.completion}"
    )


def test_case1_ipcq_copy_positive_ring_only():
    """Case 1 has inter-CUBE Ring (P parallel lanes) only; total ipcq
    must be positive. No intra-CUBE comm (PEs are disjoint on T_q).
    """
    result = _run_case1_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok
    n_copy = _count(result.engine.op_log, "ipcq_copy")
    assert n_copy > 0, f"Case 1 must have Ring KV comm; got 0 ipcq_copy"


def test_case1_writers_all_64_ranks():
    """Case 1 prefill: every rank owns disjoint T_q/(C·P) rows and
    writes them. All 8 cubes appear in dma_write component ids.
    """
    result = _run_case1_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok
    cubes = _dma_write_cubes(result.engine.op_log)
    assert cubes, "expected at least one dma_write for the final O store"
    distinct = set(cubes)
    assert distinct == set(range(8)), (
        f"Case 1 expected every cube to write; got cubes={sorted(distinct)}"
    )


# ── Panel-metrics helpers + _run_panel wiring ───────────────────────


_EXPECTED_ENGINES = {
    "pe_gemm", "pe_math", "pe_dma", "pe_fetch_store", "pe_ipcq", "pe_cpu",
}


def test_panel_metrics_helpers_present_and_correct():
    """Bench file must expose ``_end_to_end_ns`` and
    ``_engine_occupancy_ns`` for the comparative plot script.
    Exercised via the Case 2 smoke runner's op_log (cheapest case).
    """
    from kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases import (
        _end_to_end_ns,
        _engine_occupancy_ns,
    )
    result = _run_case2_smoke(T_q=_SMOKE_T_Q, S_kv=_SMOKE_S_KV)
    assert result.completion.ok
    op_log = result.engine.op_log

    lat = _end_to_end_ns(op_log)
    assert lat > 0, f"expected positive end-to-end latency; got {lat}"

    occ = _engine_occupancy_ns(op_log)
    assert isinstance(occ, dict)
    assert set(occ.keys()) >= _EXPECTED_ENGINES, (
        f"engine_occupancy_ns missing required engines; "
        f"got {set(occ.keys())}"
    )
    assert occ["pe_gemm"] > 0, (
        f"expected pe_gemm occupancy > 0; got {occ['pe_gemm']}"
    )


def test_run_panel_returns_latency_and_engine_occupancy(monkeypatch):
    """``_run_panel`` must include ``latency_ns`` + ``engine_occupancy_ns``
    alongside ``op_log_summary`` so sweep.json carries them for the
    comparative plot script. Uses monkeypatch to swap the Case 2 panel's
    T_q/S_kv to smoke sizes for speed.
    """
    import kernbench.benches.gqa_helpers.long_ctx.gqa_prefill_long_ctx_4cases as mod

    orig_kind, orig_params = mod._PANEL_DISPATCH[_CASE2_PANEL]
    fast_params = {**orig_params, "T_q": _SMOKE_T_Q, "S_kv": _SMOKE_S_KV}
    monkeypatch.setitem(
        mod._PANEL_DISPATCH, _CASE2_PANEL, (orig_kind, fast_params),
    )

    row = mod._run_panel(_CASE2_PANEL, str(TOPOLOGY_DEFAULT))
    assert "op_log_summary" in row
    assert "latency_ns" in row, (
        f"_run_panel row missing latency_ns; keys={sorted(row.keys())}"
    )
    assert row["latency_ns"] > 0
    assert "engine_occupancy_ns" in row, (
        f"_run_panel row missing engine_occupancy_ns; "
        f"keys={sorted(row.keys())}"
    )
    assert isinstance(row["engine_occupancy_ns"], dict)
    assert set(row["engine_occupancy_ns"].keys()) >= _EXPECTED_ENGINES