Function worker

test/kernel/deepseekv3_fp8_block_gemm_tuning.py:74–106

(
    M: int,
    N: int,
    K: int,
    block_size: int,
    dtype: torch.dtype,
    test_count: int,
    test_configs,
    queue,
)

def worker(
    M: int,
    N: int,
    K: int,
    block_size: int,
    dtype: torch.dtype,
    test_count: int,
    test_configs,
    queue,
):
    dog = Watchdog(timeout=10)
    dog.start()

    try:
        for index in range(len(test_configs)):
            tuning_config = test_configs[index]
            cost_time = test_fp8_block_gemm(
                M=M,
                N=N,
                K=K,
                block_size=block_size,
                dtype=dtype,
                test_count=test_count,
                **tuning_config,
            )
            dog.heartbeat()
            queue.put(cost_time)  # Put result in queue
    except Exception as ex:
        logger.exception(str(ex) + f"config {tuning_config}")
        import sys

        sys.exit(-1)
        pass


def get_test_configs(split_id, split_count):