submission 619842

import torch

from task import input_t, output_t
from aiter import ActivationType, QuantType
import aiter.fused_moe as _fmoe_module
from aiter.fused_moe import fused_moe


def _maybe_patch_sorting() -> None:
    target = getattr(_fmoe_module, "moe_sorting_fwd", None)
    if target is None:
        return
    if getattr(target, "__name__", "") == "_compact_moe_sorting_fwd":
        return

    def _compact_moe_sorting_fwd(*args, **kwargs):
        if "dispatch_policy" not in kwargs and len(args) >= 8:
            try:
                num_experts = int(args[7])
                if num_experts >= 257:
                    kwargs["dispatch_policy"] = 1
            except:
                pass
        return target(*args, **kwargs)

    _compact_moe_sorting_fwd.__name__ = "_compact_moe_sorting_fwd"
    setattr(_fmoe_module, "moe_sorting_fwd", _compact_moe_sorting_fwd)


@torch.inference_mode()
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

    _maybe_patch_sorting()

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

    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,
    )