Sonic MoE GPT-5.6 Sol

"""SM90 grouped BF16 GEMM with a fused SwiGLU epilogue.

The routing used by this workload is balanced, so a rectangular launch over
experts is both faster and simpler than building a host-side grouped-GEMM
descriptor.  Bounds still come from the device offsets: partial and empty
expert tiles are masked without synchronizing those offsets to the CPU.
"""
from __future__ import annotations

import torch
import torch.nn as nn
import triton
import triton.language as tl


@triton.jit
def _grouped_up_swiglu(
    x_ptr,
    offsets_ptr,
    gate_ptr,
    up_ptr,
    out_ptr,
    H: tl.constexpr,
    I: tl.constexpr,
    MAX_ROWS: tl.constexpr,
    EXPERT_STRIDE: tl.constexpr,
    BLOCK_M: tl.constexpr,
    BLOCK_N: tl.constexpr,
    BLOCK_K: tl.constexpr,
    GROUP_M: tl.constexpr,
    GROUP_SPAN: tl.constexpr,
):
    tile = tl.program_id(0)
    expert = tl.program_id(1)
    tiles_m: tl.constexpr = tl.cdiv(MAX_ROWS, BLOCK_M)
    group = tile // GROUP_SPAN
    first_m = group * GROUP_M
    group_m = tl.minimum(tiles_m - first_m, GROUP_M)
    within = tile - group * GROUP_SPAN
    tile_m = first_m + (within % group_m)
    tile_n = within // group_m

    begin = tl.load(offsets_ptr + expert)
    end = tl.load(offsets_ptr + expert + 1)
    rows = begin + tile_m * BLOCK_M + tl.arange(0, BLOCK_M)
    cols = tile_n * BLOCK_N + tl.arange(0, BLOCK_N)
    row_ok = rows < end
    col_ok = cols < I

    gate_acc = tl.zeros((BLOCK_M, BLOCK_N), tl.float32)
    up_acc = tl.zeros((BLOCK_M, BLOCK_N), tl.float32)

    # A is deliberately loaded once and feeds both tensor-core operations.
    for k0 in tl.range(0, H, BLOCK_K):
        ks = k0 + tl.arange(0, BLOCK_K)
        a = tl.load(
            x_ptr + rows[:, None] * H + ks[None, :],
            mask=row_ok[:, None],
            other=0.0,
        )
        w_index = expert * EXPERT_STRIDE + ks[:, None] * I + cols[None, :]
        wg = tl.load(gate_ptr + w_index, mask=col_ok[None, :], other=0.0)
        wu = tl.load(up_ptr + w_index, mask=col_ok[None, :], other=0.0)
        gate_acc = tl.dot(a, wg, gate_acc)
        up_acc = tl.dot(a, wu, up_acc)

    value = (gate_acc * tl.sigmoid(gate_acc)) * up_acc
    tl.store(
        out_ptr + rows[:, None] * I + cols[None, :],
        value,
        mask=row_ok[:, None] & col_ok[None, :],
    )


class Model(nn.Module):
    def __init__(self, T_total: int, H: int, I: int, E: int, K: int):  # noqa: E741
        super().__init__()
        self.T_total = T_total
        self.H = H
        self.I = I
        self.E = E
        self.K = K
        self.W_gate = nn.Parameter(torch.empty(E, H, I, dtype=torch.bfloat16))
        self.W_up = nn.Parameter(torch.empty(E, H, I, dtype=torch.bfloat16))
        nn.init.normal_(self.W_gate, std=0.02)
        nn.init.normal_(self.W_up, std=0.02)

    def forward(
        self,
        hidden_states: torch.Tensor,
        expert_offsets: torch.Tensor,
    ) -> torch.Tensor:
        t_perm = hidden_states.shape[0]
        out = torch.empty(
            (t_perm, self.I), dtype=torch.bfloat16, device=hidden_states.device
        )

        block_m = 128
        if self.I == 1024:
            # The small experts have only two M tiles; a narrower N tile with a
            # deeper K step wins by reducing the dual-accumulator footprint.
            block_n, block_k, stages = 64, 64, 3
        else:
            block_n, block_k = 128, 32
            stages = 6 if self.H == 4096 else 5
        max_rows = triton.cdiv(t_perm, self.E)
        grid = (
            triton.cdiv(max_rows, block_m) * triton.cdiv(self.I, block_n),
            self.E,
        )
        _grouped_up_swiglu[grid](
            hidden_states,
            expert_offsets,
            self.W_gate,
            self.W_up,
            out,
            H=self.H,
            I=self.I,
            MAX_ROWS=max_rows,
            EXPERT_STRIDE=self.H * self.I,
            BLOCK_M=block_m,
            BLOCK_N=block_n,
            BLOCK_K=block_k,
            GROUP_M=16,
            GROUP_SPAN=16 * triton.cdiv(self.I, block_n),
            num_warps=8,
            num_stages=stages,
        )
        return out


T_total = 32768
H = 4096
I = 1536  # noqa: E741
E = 128
K = 8


def _build_routing(T_total: int, E: int, K: int, device: str = "cpu") -> torch.Tensor:
    t_perm = T_total * K
    base = t_perm // E
    rem = t_perm - base * E
    counts = torch.full((E,), base, dtype=torch.int32, device=device)
    counts[:rem] += 1
    offsets = torch.zeros(E + 1, dtype=torch.int32, device=device)
    offsets[1:] = torch.cumsum(counts, dim=0)
    return offsets


def get_inputs():
    t_perm = T_total * K
    hidden_states = torch.randn(t_perm, H, dtype=torch.bfloat16) * 0.1
    expert_offsets = _build_routing(T_total, E, K)
    return [hidden_states, expert_offsets]


def get_init_inputs():
    return [T_total, H, I, E, K]