submission 745574

#!POPCORN leaderboard amd-mixed-mla
#!POPCORN gpu MI355X
"""
MLA decode: v24-style with optimal split-K tuning.
Uses persistent mode with stage1_asm + mla_reduce_v1 for multi-split cases.
"""

import torch
from task import input_t, output_t

from aiter import dtypes as aiter_dtypes
from aiter import get_mla_metadata_info_v1, get_mla_metadata_v1
from aiter.mla import mla_decode_fwd
from aiter import mla_decode_stage1_asm_fwd, mla_reduce_v1

NUM_HEADS = 16
NUM_KV_HEADS = 1
KV_LORA_RANK = 512
QK_ROPE_HEAD_DIM = 64
QK_HEAD_DIM = KV_LORA_RANK + QK_ROPE_HEAD_DIM   # 576
V_HEAD_DIM = KV_LORA_RANK                        # 512
SM_SCALE = 1.0 / (QK_HEAD_DIM ** 0.5)
PAGE_SIZE = 1
FP8_DTYPE = aiter_dtypes.fp8

_CACHE = {}


def quantize_fp8(tensor: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
    finfo = torch.finfo(FP8_DTYPE)
    amax = tensor.abs().amax().clamp(min=1e-12)
    scale = amax / finfo.max
    fp8_tensor = (tensor / scale).clamp(min=finfo.min, max=finfo.max).to(FP8_DTYPE)
    return fp8_tensor, scale.to(torch.float32).reshape(1)


def _choose_num_kv_splits(batch_size: int, kv_seq_len: int) -> int:
    """Tuned split-K: 16 for mid-batch short-context, 32 otherwise."""
    if kv_seq_len == 1024 and batch_size in (32, 64):
        return 16
    return 32


def _should_use_direct(batch_size: int, kv_seq_len: int) -> bool:
    """Use direct single-split for bs=64 kv=1k."""
    return kv_seq_len == 1024 and batch_size == 64


def _get_cached_state(batch_size, kv_seq_len, total_q, total_kv_len, device,
                      q_dtype, kv_dtype, qo_indptr, kv_indptr, num_kv_splits):
    cache_key = (batch_size, kv_seq_len, total_q, total_kv_len, num_kv_splits)
    cached = _CACHE.get(cache_key)
    if cached is not None:
        return cached

    kv_indices = torch.arange(total_kv_len, dtype=torch.int32, device=device)
    kv_last_page_len = (kv_indptr[1:] - kv_indptr[:-1]).to(torch.int32)
    o = torch.empty((total_q, NUM_HEADS, V_HEAD_DIM), dtype=torch.bfloat16, device=device)

    cached = {
        "kv_indices": kv_indices,
        "kv_last_page_len": kv_last_page_len,
        "o": o,
    }

    if num_kv_splits > 1:
        # Multi-split persistent mode metadata
        info = get_mla_metadata_info_v1(
            batch_size, 1, NUM_HEADS, 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=device) 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,
            NUM_HEADS // NUM_KV_HEADS,
            NUM_KV_HEADS,
            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,
        )
        cached["meta"] = {
            "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,
        }
        cached["logits"] = torch.empty(
            (reduce_partial_map.size(0), 1, NUM_HEADS, V_HEAD_DIM),
            dtype=torch.float32, device=device,
        )
        cached["attn_lse"] = torch.empty(
            (reduce_partial_map.size(0), 1, NUM_HEADS, 1),
            dtype=torch.float32, device=device,
        )

    _CACHE[cache_key] = cached
    return cached


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

    batch_size = config["batch_size"]
    kv_seq_len = config["kv_seq_len"]
    total_q = q.shape[0]
    total_kv_len = int(kv_indptr[-1].item())

    q_fp8, q_scale = quantize_fp8(q)
    kv_buffer_fp8, kv_scale = kv_data["fp8"]
    kv_buffer_4d = kv_buffer_fp8.view(total_kv_len, PAGE_SIZE, NUM_KV_HEADS, -1)

    use_direct = _should_use_direct(batch_size, kv_seq_len)
    num_kv_splits = 1 if use_direct else _choose_num_kv_splits(batch_size, kv_seq_len)

    state = _get_cached_state(
        batch_size, kv_seq_len, total_q, total_kv_len, q.device,
        FP8_DTYPE, FP8_DTYPE, qo_indptr, kv_indptr, num_kv_splits,
    )

    o = state["o"]
    q_reshaped = q_fp8.view(-1, NUM_HEADS, QK_HEAD_DIM)

    if use_direct:
        # Direct output: single-split via mla_decode_fwd
        mla_decode_fwd(
            q_reshaped, kv_buffer_4d, o,
            qo_indptr, kv_indptr,
            state["kv_indices"], state["kv_last_page_len"],
            1, page_size=PAGE_SIZE, nhead_kv=NUM_KV_HEADS,
            sm_scale=SM_SCALE, logit_cap=0.0, num_kv_splits=1,
            q_scale=q_scale, kv_scale=kv_scale,
            intra_batch_mode=True,
        )
    else:
        # Multi-split persistent: stage1_asm + mla_reduce_v1
        meta = state["meta"]
        mla_decode_stage1_asm_fwd(
            q_reshaped, kv_buffer_4d,
            qo_indptr, kv_indptr,
            state["kv_indices"], state["kv_last_page_len"],
            None,
            meta["work_meta_data"], meta["work_indptr"], meta["work_info_set"],
            1, PAGE_SIZE, NUM_KV_HEADS, SM_SCALE,
            state["logits"], state["attn_lse"], o,
            q_scale, kv_scale,
        )
        mla_reduce_v1(
            state["logits"], state["attn_lse"],
            meta["reduce_indptr"], meta["reduce_final_map"],
            meta["reduce_partial_map"],
            1, o, None,
        )
    return o