submission 622022

"""
Optimized MLA (Multi-head Latent Attention) decode kernel using aiter backend.

DeepSeek R1 forward_absorb MLA: absorbed q (576), compressed kv_buffer (576),
output v_head_dim = kv_lora_rank = 512.

Key optimizations:
1. Use aiter mla_decode_fwd kernel - highly optimized persistent mode
2. FP8 quantization for Q and KV (a8w8) - fastest on MI355X
3. Dynamic NUM_KV_SPLITS based on batch size and seq length
4. Cached metadata buffers to reduce allocation overhead
5. Optimized Q quantization path

Performance: a8w8 is ~2-3x faster than bf16 on MI355X
"""

import torch
import torch.nn.functional as F
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
from aiter.utility.fp4_utils import (
    mxfp4_to_f32,
    e8m0_to_f32,
)

# ---------------------------------------------------------------------------
# DeepSeek R1 latent MQA constants (forward_absorb path)
# ---------------------------------------------------------------------------
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 (platform-specific via aiter)
FP8_DTYPE = aiter_dtypes.fp8

# Block size for MXFP4 quantization
MXFP4_BLOCK_SIZE = 32
NUM_BLOCKS = QK_HEAD_DIM // MXFP4_BLOCK_SIZE  # 18

# Cache for metadata buffers
_metadata_cache = {}


# ---------------------------------------------------------------------------
# Dynamic NUM_KV_SPLITS calculation
# ---------------------------------------------------------------------------
def get_optimal_kv_splits(batch_size: int, kv_seq_len: int) -> int:
    """
    Calculate optimal NUM_KV_SPLITS based on batch size and sequence length.
    
    Tuning strategy:
    - Small batch + short seq: fewer splits (less overhead)
    - Large batch + long seq: more splits (better parallelism)
    - Balance between parallelism and reduction overhead
    """
    # Base splits based on batch size
    if batch_size <= 4:
        base_splits = 8
    elif batch_size <= 16:
        base_splits = 16
    elif batch_size <= 64:
        base_splits = 32
    else:
        base_splits = 64
    
    # Adjust based on sequence length
    if kv_seq_len <= 1024:
        splits = min(base_splits, 8)
    elif kv_seq_len <= 4096:
        splits = min(base_splits, 16)
    else:
        splits = base_splits
    
    # Ensure at least 1 split
    return max(1, splits)


# ---------------------------------------------------------------------------
# FP8 quantization (optimized)
# ---------------------------------------------------------------------------
def quantize_fp8_fast(tensor: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
    """
    Optimized dynamic per-tensor FP8 quantization.
    
    Uses fused operations where possible to reduce kernel launches.
    """
    finfo = torch.finfo(FP8_DTYPE)
    amax = tensor.abs().amax().clamp(min=1e-12)
    scale = amax / finfo.max
    
    # Fused: divide, clamp, cast
    fp8_tensor = (tensor / scale).clamp(min=finfo.min, max=finfo.max).to(FP8_DTYPE)
    
    return fp8_tensor, scale.to(torch.float32).reshape(1)


# ---------------------------------------------------------------------------
# MXFP4 Dequantization (for fallback path)
# ---------------------------------------------------------------------------
def dequantize_mxfp4_batch(
    fp4_data: torch.Tensor,
    scale_e8m0: torch.Tensor,
) -> torch.Tensor:
    """Batch dequantize MXFP4 to bf16."""
    total_kv = fp4_data.shape[0]
    N = QK_HEAD_DIM
    
    fp4_data_2d = fp4_data.view(total_kv, N // 2)
    float_vals = mxfp4_to_f32(fp4_data_2d)
    
    scale_f32 = e8m0_to_f32(scale_e8m0)
    scale_f32 = scale_f32[:, :NUM_BLOCKS]
    
    float_vals_blocked = float_vals.view(total_kv, NUM_BLOCKS, MXFP4_BLOCK_SIZE)
    scaled = float_vals_blocked * scale_f32.unsqueeze(-1)
    
    return scaled.view(total_kv, 1, N).to(torch.bfloat16)


# ---------------------------------------------------------------------------
# Cached metadata generation
# ---------------------------------------------------------------------------
def _get_or_create_metadata(
    batch_size: int,
    max_q_len: int,
    nhead: int,
    nhead_kv: int,
    q_dtype: torch.dtype,
    kv_dtype: torch.dtype,
    qo_indptr: torch.Tensor,
    kv_indptr: torch.Tensor,
    kv_last_page_len: torch.Tensor,
    num_kv_splits: int,
):
    """
    Get cached metadata or create new one.
    
    Caching strategy:
    - Cache key: (batch_size, max_q_len, num_kv_splits, dtypes)
    - Reuse buffers when possible to reduce allocation overhead
    """
    cache_key = (batch_size, max_q_len, num_kv_splits, q_dtype, kv_dtype)
    
    if cache_key in _metadata_cache:
        cached = _metadata_cache[cache_key]
        # Reuse cached buffers, just update metadata
        get_mla_metadata_v1(
            qo_indptr, kv_indptr, kv_last_page_len,
            nhead // nhead_kv,
            nhead_kv,
            True,
            cached["work_meta_data"],
            cached["work_info_set"],
            cached["work_indptr"],
            cached["reduce_indptr"],
            cached["reduce_final_map"],
            cached["reduce_partial_map"],
            page_size=PAGE_SIZE,
            kv_granularity=max(PAGE_SIZE, 16),
            max_seqlen_qo=max_q_len,
            uni_seqlen_qo=max_q_len,
            fast_mode=False,
            max_split_per_batch=num_kv_splits,
            intra_batch_mode=True,
            dtype_q=q_dtype,
            dtype_kv=kv_dtype,
        )
        return cached
    
    # Create new metadata buffers
    info = get_mla_metadata_info_v1(
        batch_size, max_q_len, nhead, 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,
        nhead // nhead_kv,
        nhead_kv,
        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=max_q_len,
        uni_seqlen_qo=max_q_len,
        fast_mode=False,
        max_split_per_batch=num_kv_splits,
        intra_batch_mode=True,
        dtype_q=q_dtype,
        dtype_kv=kv_dtype,
    )
    
    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,
    }
    
    # Cache for future use (limit cache size)
    if len(_metadata_cache) < 16:
        _metadata_cache[cache_key] = meta
    
    return meta


# ---------------------------------------------------------------------------
# Optimized Aiter MLA decode kernel
# ---------------------------------------------------------------------------
def aiter_mla_decode_optimized(
    q: torch.Tensor,
    kv_buffer: torch.Tensor,
    qo_indptr: torch.Tensor,
    kv_indptr: torch.Tensor,
    config: dict,
    q_scale: torch.Tensor,
    kv_scale: torch.Tensor,
) -> torch.Tensor:
    """
    Optimized MLA decode attention using aiter a8w8 persistent-mode kernel.
    
    Optimizations:
    1. Dynamic NUM_KV_SPLITS based on workload
    2. Cached metadata buffers
    3. Efficient memory allocation
    
    Args:
        q: (total_q, num_heads, 576) fp8 - quantized queries
        kv_buffer: (total_kv, 1, 576) fp8 - quantized KV cache
        qo_indptr: (batch_size + 1,) int32 - query segment pointers
        kv_indptr: (batch_size + 1,) int32 - KV segment pointers
        config: dict with MLA parameters
        q_scale: scalar float32 - Q scale factor
        kv_scale: scalar float32 - KV scale factor
    
    Returns:
        attention_output: (total_q, num_heads, 512) bfloat16
    """
    batch_size = config["batch_size"]
    nq = config["num_heads"]
    nkv = config["num_kv_heads"]
    dq = config["qk_head_dim"]
    dv = config["v_head_dim"]
    q_seq_len = config["q_seq_len"]
    kv_seq_len = config["kv_seq_len"]
    
    # Dynamic NUM_KV_SPLITS
    num_kv_splits = get_optimal_kv_splits(batch_size, kv_seq_len)
    
    total_kv_len = int(kv_indptr[-1].item())
    kv_indices = torch.arange(total_kv_len, dtype=torch.int32, device="cuda")
    
    # Reshape kv_buffer to 4D for aiter: (total_kv, page_size, nhead_kv, dim)
    kv_buffer_4d = kv_buffer.view(kv_buffer.shape[0], PAGE_SIZE, nkv, kv_buffer.shape[-1])
    
    max_q_len = q_seq_len
    kv_last_page_len = (kv_indptr[1:] - kv_indptr[:-1]).to(torch.int32)
    
    # Get or create metadata (with caching)
    meta = _get_or_create_metadata(
        batch_size, max_q_len, nq, nkv,
        q.dtype, kv_buffer.dtype,
        qo_indptr, kv_indptr, kv_last_page_len,
        num_kv_splits,
    )
    
    # Allocate output
    o = torch.empty((q.shape[0], nq, dv), dtype=torch.bfloat16, device="cuda")
    
    # Call aiter kernel
    mla_decode_fwd(
        q.view(-1, nq, dq),
        kv_buffer_4d,
        o,
        qo_indptr,
        kv_indptr,
        kv_indices,
        kv_last_page_len,
        max_q_len,
        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 o


# ---------------------------------------------------------------------------
# Fallback: PyTorch implementation with MXFP4
# ---------------------------------------------------------------------------
def pytorch_mla_mxfp4(
    q: torch.Tensor,
    kv_buffer_mxfp4: torch.Tensor,
    kv_scale_mxfp4: torch.Tensor,
    qo_indptr: torch.Tensor,
    kv_indptr: torch.Tensor,
    config: dict,
) -> torch.Tensor:
    """
    Fallback PyTorch implementation with MXFP4 KV.
    
    Used when FP8 KV is not available.
    """
    batch_size = config["batch_size"]
    nq = config["num_heads"]
    dv = config["v_head_dim"]
    
    total_q = q.shape[0]
    total_kv = kv_buffer_mxfp4.shape[0]
    
    # Batch dequantize MXFP4 KV
    kv_bf16 = dequantize_mxfp4_batch(kv_buffer_mxfp4, kv_scale_mxfp4)
    k_bf16 = kv_bf16.view(total_kv, QK_HEAD_DIM)
    v_bf16 = kv_bf16[:, :, :dv].view(total_kv, dv)
    
    output = torch.empty((total_q, nq, dv), dtype=torch.bfloat16, device=q.device)
    
    for b in range(batch_size):
        q_start = qo_indptr[b].item()
        q_end = qo_indptr[b + 1].item()
        kv_start = kv_indptr[b].item()
        kv_end = kv_indptr[b + 1].item()
        
        kv_len = kv_end - kv_start
        
        if kv_len == 0:
            continue
        
        q_b = q[q_start:q_end]
        k_b = k_bf16[kv_start:kv_end]
        v_b = v_bf16[kv_start:kv_end]
        
        # Compute attention
        scores = torch.matmul(q_b.float(), k_b.t().unsqueeze(0)) * SM_SCALE
        scores_max = scores.amax(dim=-1, keepdim=True)
        exp_scores = torch.exp(scores - scores_max)
        sum_exp = exp_scores.sum(dim=-1, keepdim=True)
        
        out = torch.matmul(exp_scores, v_b.unsqueeze(0).float()) / sum_exp
        output[q_start:q_end] = out.to(torch.bfloat16)
    
    return output


# ---------------------------------------------------------------------------
# Main Kernel Entry Point
# ---------------------------------------------------------------------------
def custom_kernel(data: input_t) -> output_t:
    """
    Optimized MLA decode attention using aiter backend.
    
    Strategy:
    1. Prefer aiter a8w8 kernel (fp8 Q + fp8 KV) - fastest
    2. Dynamic NUM_KV_SPLITS based on workload
    3. Cached metadata buffers
    4. Fallback to PyTorch with MXFP4 if fp8 KV not available
    
    The aiter a8w8 kernel is ~2-3x faster than bf16 on MI355X.
    """
    q, kv_data, qo_indptr, kv_indptr, config = data
    
    # Check if fp8 KV is available
    if "fp8" in kv_data and kv_data["fp8"] is not None:
        # Use aiter a8w8 kernel (fastest path)
        kv_buffer_fp8, kv_scale = kv_data["fp8"]
        
        # Quantize Q to fp8 (optimized)
        q_fp8, q_scale = quantize_fp8_fast(q)
        
        # Call optimized aiter kernel
        return aiter_mla_decode_optimized(
            q_fp8, kv_buffer_fp8, qo_indptr, kv_indptr, config,
            q_scale=q_scale, kv_scale=kv_scale,
        )
    else:
        # Fallback to PyTorch with MXFP4
        kv_buffer_mxfp4, kv_scale_mxfp4 = kv_data["mxfp4"]
        return pytorch_mla_mxfp4(
            q, kv_buffer_mxfp4, kv_scale_mxfp4, qo_indptr, kv_indptr, config
        )