Function worker

test/kernel/fuse_moe_tuning.py:229–268

(
    expert_num: int,
    m: int,
    n: int,
    k: int,
    topk: int,
    dtype: torch.dtype,
    test_count: int,
    use_fp8_w8a8: bool,
    is_up: bool,
    block_shape,
    num_fused_shared_experts: int,
    test_configs,
    queue,
)

def worker(
    expert_num: int,
    m: int,
    n: int,
    k: int,
    topk: int,
    dtype: torch.dtype,
    test_count: int,
    use_fp8_w8a8: bool,
    is_up: bool,
    block_shape,
    num_fused_shared_experts: int,
    test_configs,
    queue,
):
    try:
        for index in range(len(test_configs)):
            cost_time = test_kernel(
                expert_num=expert_num,
                m=m,
                n=n,
                k=k,
                topk=topk,
                dtype=dtype,
                test_count=test_count,
                use_fp8_w8a8=use_fp8_w8a8,
                is_up=is_up,
                block_shape=block_shape,
                num_fused_shared_experts=num_fused_shared_experts,
                **test_configs[index],
            )
            queue.put(cost_time)  # Put result in queue

    except Exception as ex:
        logger.error(str(ex))
        logger.exception(str(ex))
        import sys

        sys.exit(-1)
        pass


def get_test_configs(split_id, split_count):