submission 586081

"""
MLA decode submission: same aiter fp8 kernel as reference,
but with per-(batch_size, kv_seq_len) tuned num_kv_splits
instead of the hardcoded 32 the reference uses.

On first call for a given (batch_size, kv_seq_len) config, we
benchmark a range of num_kv_splits values and cache the winner.
Subsequent calls use the cached value directly.
"""

import time
import torch
from task import input_t, output_t
from aiter.mla import mla_decode_fwd
from aiter import dtypes as aiter_dtypes
from aiter import get_mla_metadata_info_v1, get_mla_metadata_v1

# ---------------------------------------------------------------------------
# Constants (match reference exactly)
# ---------------------------------------------------------------------------
FP8_DTYPE   = aiter_dtypes.fp8
PAGE_SIZE   = 1
SM_SCALE    = 1.0 / (576 ** 0.5)

# Candidates to try during tuning. Powers of 2 + a few extras.
# Reference hardcodes 32 — we search around and beyond it.
KV_SPLITS_CANDIDATES = [1, 2, 4, 8, 16, 24, 32, 48, 64]

# Cache: (batch_size, kv_seq_len) -> best num_kv_splits
_tuned_splits: dict[tuple[int, int], int] = {}


# ---------------------------------------------------------------------------
# FP8 quantization (identical to reference)
# ---------------------------------------------------------------------------

def _quantize_fp8(tensor: torch.Tensor):
    finfo = torch.finfo(FP8_DTYPE)
    amax  = tensor.abs().amax().clamp(min=1e-12)
    scale = amax / finfo.max
    fp8   = (tensor / scale).clamp(finfo.min, finfo.max).to(FP8_DTYPE)
    return fp8, scale.to(torch.float32).reshape(1)


# ---------------------------------------------------------------------------
# Metadata builder (identical to reference, parameterised on num_kv_splits)
# ---------------------------------------------------------------------------

def _make_meta(batch_size, nq, nkv, q_dtype, kv_dtype,
               qo_indptr, kv_indptr, kv_last_page_len, num_kv_splits):
    info = get_mla_metadata_info_v1(
        batch_size, 1, nq, q_dtype, kv_dtype,
        is_sparse=False, fast_mode=False,
        num_kv_splits=num_kv_splits, intra_batch_mode=True,
    )
    work = [torch.empty(s, dtype=t, device="cuda") for s, t in info]
    (work_metadata, work_indptr, work_info_set,
     reduce_indptr, reduce_final_map, reduce_partial_map) = work

    get_mla_metadata_v1(
        qo_indptr, kv_indptr, kv_last_page_len,
        nq // nkv, nkv, True,
        work_metadata, work_info_set, work_indptr,
        reduce_indptr, reduce_final_map, reduce_partial_map,
        page_size=PAGE_SIZE,
        kv_granularity=max(PAGE_SIZE, 16),
        max_seqlen_qo=1,
        uni_seqlen_qo=1,
        fast_mode=False,
        max_split_per_batch=num_kv_splits,
        intra_batch_mode=True,
        dtype_q=q_dtype,
        dtype_kv=kv_dtype,
    )
    return {
        "work_meta_data":    work_metadata,
        "work_indptr":       work_indptr,
        "work_info_set":     work_info_set,
        "reduce_indptr":     reduce_indptr,
        "reduce_final_map":  reduce_final_map,
        "reduce_partial_map":reduce_partial_map,
    }


# ---------------------------------------------------------------------------
# Single kernel call (fp8 Q + fp8 KV, matches reference exactly)
# ---------------------------------------------------------------------------

def _run(q_fp8, q_scale, kv_fp8, kv_scale,
         qo_indptr, kv_indptr, config, num_kv_splits, out=None):
    bs   = config["batch_size"]
    nq   = config["num_heads"]
    nkv  = config["num_kv_heads"]
    dq   = config["qk_head_dim"]
    dv   = config["v_head_dim"]

    total_kv = int(kv_indptr[-1].item())
    kv_idx   = torch.arange(total_kv, dtype=torch.int32, device="cuda")
    kv_last  = (kv_indptr[1:] - kv_indptr[:-1]).to(torch.int32)

    kv_4d = kv_fp8.view(kv_fp8.shape[0], PAGE_SIZE, nkv, dq)
    meta  = _make_meta(bs, nq, nkv, q_fp8.dtype, kv_fp8.dtype,
                       qo_indptr, kv_indptr, kv_last, num_kv_splits)

    if out is None:
        out = torch.empty((q_fp8.shape[0], nq, dv), dtype=torch.bfloat16, device="cuda")

    mla_decode_fwd(
        q_fp8.view(-1, nq, dq), kv_4d, out,
        qo_indptr, kv_indptr, kv_idx, kv_last,
        1,                          # max_q_len (decode = 1)
        page_size=PAGE_SIZE,
        nhead_kv=nkv,
        sm_scale=SM_SCALE,
        logit_cap=0.0,
        num_kv_splits=num_kv_splits,
        q_scale=q_scale,
        kv_scale=kv_scale,
        intra_batch_mode=True,
        **meta,
    )
    return out


# ---------------------------------------------------------------------------
# Tuning: benchmark candidates and pick fastest for this config
# ---------------------------------------------------------------------------

def _tune(q_fp8, q_scale, kv_fp8, kv_scale,
          qo_indptr, kv_indptr, config) -> int:
    """
    Try each candidate num_kv_splits value, return the fastest.
    Uses CUDA events for accurate GPU timing (3 warm + 5 timed runs).
    """
    bs      = config["batch_size"]
    kv_len  = config["kv_seq_len"]
    key     = (bs, kv_len)

    if key in _tuned_splits:
        return _tuned_splits[key]

    best_splits = 32
    best_time   = float("inf")

    out = torch.empty(
        (q_fp8.shape[0], config["num_heads"], config["v_head_dim"]),
        dtype=torch.bfloat16, device="cuda"
    )

    for splits in KV_SPLITS_CANDIDATES:
        try:
            # warm up
            for _ in range(3):
                _run(q_fp8, q_scale, kv_fp8, kv_scale,
                     qo_indptr, kv_indptr, config, splits, out)
            torch.cuda.synchronize()

            # time
            start = torch.cuda.Event(enable_timing=True)
            end   = torch.cuda.Event(enable_timing=True)
            start.record()
            for _ in range(5):
                _run(q_fp8, q_scale, kv_fp8, kv_scale,
                     qo_indptr, kv_indptr, config, splits, out)
            end.record()
            torch.cuda.synchronize()

            elapsed = start.elapsed_time(end) / 5  # ms per call

            if elapsed < best_time:
                best_time   = elapsed
                best_splits = splits

        except Exception:
            # Some split counts may be invalid for small configs — skip
            continue

    _tuned_splits[key] = best_splits
    return best_splits


# ---------------------------------------------------------------------------
# Public entry point
# ---------------------------------------------------------------------------

def custom_kernel(data: input_t) -> output_t:
    q, kv_data, qo_indptr, kv_indptr, config = data

    # Quantize Q to fp8 (same as reference)
    q_fp8, q_scale = _quantize_fp8(q)

    # Use pre-quantized fp8 KV (same as reference)
    kv_fp8, kv_scale = kv_data["fp8"]

    # Get (or tune) the best num_kv_splits for this config
    splits = _tune(q_fp8, q_scale, kv_fp8, kv_scale,
                   qo_indptr, kv_indptr, config)

    return _run(q_fp8, q_scale, kv_fp8, kv_scale,
                qo_indptr, kv_indptr, config, splits)