submission 716151

import os
import torch
from task import input_t, output_t

os.environ["AITER_USE_NT"] = "0"

from aiter import ActivationType, QuantType
from aiter.fused_moe import fused_moe, get_ksplit


def custom_kernel(data: input_t) -> output_t:
    (
        hidden_states, gate_up_weight, down_weight,
        gate_up_weight_scale, down_weight_scale,
        gate_up_weight_shuffled, down_weight_shuffled,
        gate_up_weight_scale_shuffled, down_weight_scale_shuffled,
        topk_weights, topk_ids, config,
    ) = data

    hidden_pad = config["d_hidden_pad"] - config["d_hidden"]
    intermediate_pad = config["d_expert_pad"] - config["d_expert"]

    # Per-case ksplit: split-K for small-batch cases, no split for large batch
    if config["bs"] <= 128:
        if config["n_routed_experts"] > 100:
            os.environ["AITER_KSPLIT"] = "4"
        else:
            os.environ["AITER_KSPLIT"] = "2"
    else:
        os.environ["AITER_KSPLIT"] = "0"
    get_ksplit.cache_clear()

    # block_m tuning for E=33 cases (only effective when ksplit=0, i.e. ck2stages path)
    block_size_M = None
    if config["n_routed_experts"] < 100:
        if config["d_expert"] == 512 and config["bs"] == 512:
            block_size_M = 128
        elif config["d_expert"] == 2048 and config["bs"] == 512:
            block_size_M = 64

    return fused_moe(
        hidden_states, gate_up_weight_shuffled, down_weight_shuffled,
        topk_weights, topk_ids,
        expert_mask=None, activation=ActivationType.Silu,
        quant_type=QuantType.per_1x32, doweight_stage1=False,
        w1_scale=gate_up_weight_scale_shuffled,
        w2_scale=down_weight_scale_shuffled,
        a1_scale=None, a2_scale=None,
        hidden_pad=hidden_pad, intermediate_pad=intermediate_pad,
        block_size_M=block_size_M,
    )