From dc5b2f88b22a49dccd72c4655a73a37299bbeaee Mon Sep 17 00:00:00 2001
From: sunnycase <sunnycase@live.cn>
Date: Mon, 8 Jun 2026 04:54:07 +0000
Subject: [PATCH 1/6] [TLE] Optimize hint manager lookup

---
 .../test/unit/language/test_hint_manager.py   | 29 +++++++++++++++++++
 python/triton/compiler/code_generator.py      |  9 ++++--
 .../nvidia/backend/nvidia_hint_handler.py     | 19 +++++++-----
 3 files changed, 47 insertions(+), 10 deletions(-)
 create mode 100644 python/test/unit/language/test_hint_manager.py

diff --git a/python/test/unit/language/test_hint_manager.py b/python/test/unit/language/test_hint_manager.py
new file mode 100644
index 0000000000..9dc83f5007
--- /dev/null
+++ b/python/test/unit/language/test_hint_manager.py
@@ -0,0 +1,29 @@
+from types import SimpleNamespace
+
+from triton.backends.nvidia.nvidia_hint_handler import NvidiaHintHandler
+
+
+class ParseMustNotRun:
+
+    def parse(self):
+        raise AssertionError("hint lookup must not reparse the JIT function")
+
+
+def test_nvidia_hint_lookup_uses_codegen_attached_map():
+    code_generator = SimpleNamespace(
+        flagtree_line_hints={17: "cache_global"},
+        jit_fn=ParseMustNotRun(),
+    )
+    node = SimpleNamespace(lineno=17)
+
+    assert NvidiaHintHandler.get_node_hints(code_generator, node) == "cache_global"
+
+
+def test_nvidia_hint_source_cache_returns_independent_dicts():
+    jit_fn = SimpleNamespace(src="def kernel(x):\n    y = x  # @hint:cache_global\n    return y\n")
+
+    first = NvidiaHintHandler.maps_line_numbers_to_comment_hints(jit_fn)
+    first[2] = "mutated"
+    second = NvidiaHintHandler.maps_line_numbers_to_comment_hints(jit_fn)
+
+    assert second == {2: "cache_global"}
diff --git a/python/triton/compiler/code_generator.py b/python/triton/compiler/code_generator.py
index bef96cc51f..250e14f54e 100644
--- a/python/triton/compiler/code_generator.py
+++ b/python/triton/compiler/code_generator.py
@@ -380,6 +380,7 @@ def __init__(self, context, prototype, gscope, function_name, jit_fn: JITFunctio
 
         self.lscope = {}
         self.jit_fn = jit_fn
+        self.flagtree_line_hints = {}
         # TODO: we currently generate illegal names for non-kernel functions involving constexprs!
         if is_kernel:
             function_name = function_name[function_name.rfind('.') + 1:]
@@ -1413,7 +1414,9 @@ def call_JitFunction(self, fn: JITFunction, args, kwargs, caller_context=None):
                                       module_map=self.builder.module_map, caller_context=caller_context,
                                       is_gluon=self.is_gluon)
             try:
-                generator.visit(fn.parse())
+                tree = fn.parse()
+                generator.flagtree_line_hints = getattr(tree.body[0], 'line_flagtree_hints', {}) or {}
+                generator.visit(tree)
             except Exception as e:
                 # Wrap the error in the callee with the location of the call.
                 if knobs.compilation.front_end_debugging:
@@ -1743,7 +1746,9 @@ def apply_constexpr_types(argument, indices, value):
     generator = CodeGenerator(context, prototype, gscope=fn.get_capture_scope(), function_name=fn.repr(proxy),
                               jit_fn=fn, is_kernel=True, file_name=file_name, begin_line=begin_line, options=options,
                               codegen_fns=codegen_fns, module_map=module_map, module=module, is_gluon=fn.is_gluon())
-    generator.visit(fn.parse())
+    tree = fn.parse()
+    generator.flagtree_line_hints = getattr(tree.body[0], 'line_flagtree_hints', {}) or {}
+    generator.visit(tree)
     module = generator.module
     # module takes ownership of the context
     module.context = context
diff --git a/third_party/nvidia/backend/nvidia_hint_handler.py b/third_party/nvidia/backend/nvidia_hint_handler.py
index f52f2c9c16..d19af4353b 100644
--- a/third_party/nvidia/backend/nvidia_hint_handler.py
+++ b/third_party/nvidia/backend/nvidia_hint_handler.py
@@ -1,4 +1,8 @@
 # should store at third_party/nvidia/backend/
+import functools
+import tokenize
+from io import StringIO
+
 from triton.compiler.hint_manager import BaseHintHandler
 
 
@@ -7,11 +11,8 @@ class NvidiaHintHandler(BaseHintHandler):
     # TODO : below can be reused by other backend which need to implement "hint";
     @staticmethod
     def get_node_hints(code_generator, node):
-        line_num = node.lineno
-        function_def = code_generator.jit_fn.parse()
-        line_flagtree_hints = getattr(function_def.body[0], 'line_flagtree_hints', {})
-        flagtree_hints = line_flagtree_hints.get(line_num)
-        return flagtree_hints
+        line_flagtree_hints = getattr(code_generator, 'flagtree_line_hints', {})
+        return line_flagtree_hints.get(node.lineno)
 
     @staticmethod
     def inject_kwargs_with_hints(fn, flagtree_hints, line_num, kws):
@@ -24,11 +25,13 @@ def inject_kwargs_with_hints(fn, flagtree_hints, line_num, kws):
 
     @staticmethod
     def maps_line_numbers_to_comment_hints(jit_fn):
-        import tokenize
-        from io import StringIO
+        return dict(NvidiaHintHandler._maps_line_numbers_to_comment_hints_from_source(jit_fn.src))
+
+    @staticmethod
+    @functools.lru_cache(maxsize=256)
+    def _maps_line_numbers_to_comment_hints_from_source(code_str):
         # Maps line numbers to comment hints
         line_flagtree_hints = {}
-        code_str = jit_fn.src
         g = tokenize.generate_tokens(StringIO(code_str).readline)
         for tok_type, tok_text, start, end, _ in g:
             if tok_type == tokenize.COMMENT:

From bd4313225992784d9ecb9488f5234db068306da6 Mon Sep 17 00:00:00 2001
From: sunnycase <sunnycase@live.cn>
Date: Tue, 9 Jun 2026 04:02:57 +0000
Subject: [PATCH 2/6] ci: add mthreads verify service test

---
 .github/workflows/mthreads-build-and-test.yml |  15 +
 .../python/test/test_verify_service.py        | 644 ++++++++++++++++++
 2 files changed, 659 insertions(+)
 create mode 100644 third_party/mthreads/python/test/test_verify_service.py

diff --git a/.github/workflows/mthreads-build-and-test.yml b/.github/workflows/mthreads-build-and-test.yml
index c94a16e39e..6d89a99549 100644
--- a/.github/workflows/mthreads-build-and-test.yml
+++ b/.github/workflows/mthreads-build-and-test.yml
@@ -46,3 +46,18 @@ jobs:
         run: |
           set -x
           python3 -m pytest -s third_party/mthreads/python/test/unit --device musa
+
+      - name: Verify Service Test on Mthreads
+        if: steps.check_backend.outputs.should_skip != 'true'
+        shell: bash
+        env:
+          MTHREADS_VERIFY_SERVICE_URL: ${{ vars.MTHREADS_VERIFY_SERVICE_URL }}
+          MTHREADS_VERIFY_SERVICE_KERNEL: ${{ vars.MTHREADS_VERIFY_SERVICE_KERNEL }}
+        run: |
+          set -x
+          verify_service_url="${MTHREADS_VERIFY_SERVICE_URL:-http://10.7.66.1:8912}"
+          verify_service_kernel="${MTHREADS_VERIFY_SERVICE_KERNEL:-gelu}"
+          python3 third_party/mthreads/python/test/test_verify_service.py \
+            --url "${verify_service_url}" \
+            --chip moore \
+            --kernel "${verify_service_kernel}"
diff --git a/third_party/mthreads/python/test/test_verify_service.py b/third_party/mthreads/python/test/test_verify_service.py
new file mode 100644
index 0000000000..c184215b08
--- /dev/null
+++ b/third_party/mthreads/python/test/test_verify_service.py
@@ -0,0 +1,644 @@
+"""
+验证服务测试脚本 — 支持多芯片、参数化 benchmark、HTML 性能表格
+
+用法:
+    python test_verify_service.py --url http://172.24.4.45:8912 --chip nvidia
+    python test_verify_service.py --url http://10.7.64.131:8912 --chip huawei --all
+    python test_verify_service.py --url http://10.7.66.1:8912 --chip moore --kernel gelu
+
+芯片:
+    nvidia / haiguang / tianshu / muxi / pingtouge  → CUDA 兼容, 代码不变
+    huawei → cuda 替换为 npu
+    moore  → cuda 替换为 musa
+
+测试算子:
+    gelu    GELU 激活函数 (参数化 benchmark: 12 组 dtype×shape)
+    matmul  矩阵乘法 FP16 (正确性验证)
+    add     向量加法 (正确性验证)
+"""
+
+import argparse
+import json
+import re
+import sys
+import time
+import urllib.request
+import urllib.error
+from html import unescape
+from typing import Dict, Optional
+
+
+# ============================================================
+# 芯片配置
+# ============================================================
+CHIP_CONFIG = {
+    "nvidia":     {"device_keyword": "cuda",  "desc": "NVIDIA CUDA"},
+    "haiguang":   {"device_keyword": "cuda",  "desc": "海光 DCU (CUDA兼容)"},
+    "tianshu":    {"device_keyword": "cuda",  "desc": "天数智芯 (CUDA兼容)"},
+    "muxi":       {"device_keyword": "cuda",  "desc": "沐曦 (CUDA兼容)"},
+    "pingtouge":  {"device_keyword": "cuda",  "desc": "平头哥 (CUDA兼容)"},
+    "huawei":     {"device_keyword": "npu",   "desc": "华为昇腾 Ascend"},
+    "moore":      {"device_keyword": "musa",  "desc": "摩尔线程 MUSA"},
+}
+
+
+# ============================================================
+# HTTP 工具
+# ============================================================
+def http_post(url: str, data: dict, timeout: int = 600) -> dict:
+    body = json.dumps(data).encode("utf-8")
+    req = urllib.request.Request(
+        url, data=body,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            return json.loads(resp.read().decode("utf-8"))
+    except urllib.error.HTTPError as e:
+        return {"error": True, "message": f"HTTP {e.code}: {e.reason}",
+                "detail": e.read().decode("utf-8", errors="replace")}
+    except urllib.error.URLError as e:
+        return {"error": True, "message": f"Connection failed: {e.reason}"}
+    except Exception as e:
+        return {"error": True, "message": str(e)}
+
+
+def http_get(url: str, timeout: int = 10) -> dict:
+    req = urllib.request.Request(url)
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            return json.loads(resp.read().decode("utf-8"))
+    except Exception as e:
+        return {"error": True, "message": str(e)}
+
+
+# ============================================================
+# 设备后处理
+# ============================================================
+def apply_device_postprocess(code: str, chip: str) -> str:
+    """将代码中的 cuda 替换为目标芯片的设备关键字"""
+    if not code:
+        return code
+    cfg = CHIP_CONFIG.get(chip, {})
+    target = cfg.get("device_keyword", "cuda")
+    if target == "cuda":
+        return code
+    return re.sub(r'\bcuda\b', target, code)
+
+
+# ============================================================
+# GELU — Triton 融合实现 (参数化 benchmark)
+# ============================================================
+GELU_TRITON_CODE = r'''
+import torch
+import triton
+import triton.language as tl
+
+
+@triton.jit
+def gelu_kernel(x_ptr, y_ptr, n_elements, BLOCK_SIZE: tl.constexpr):
+    pid = tl.program_id(0)
+    block_start = pid * BLOCK_SIZE
+    offsets = block_start + tl.arange(0, BLOCK_SIZE)
+    mask = offsets < n_elements
+    x = tl.load(x_ptr + offsets, mask=mask)
+    sqrt_2_over_pi = 0.7978845608028654
+    coeff = 0.044715
+    x3 = x * x * x
+    inner = sqrt_2_over_pi * (x + coeff * x3)
+    # tanh(x) = (exp(2x) - 1) / (exp(2x) + 1), avoids tl.math.tanh
+    # which is unavailable on some non-NVIDIA Triton backends (e.g. Muxi/Corex)
+    exp2x = tl.math.exp(2.0 * inner)
+    tanh_inner = (exp2x - 1.0) / (exp2x + 1.0)
+    y = 0.5 * x * (1.0 + tanh_inner)
+    tl.store(y_ptr + offsets, y, mask=mask)
+
+
+def gelu(x: torch.Tensor) -> torch.Tensor:
+    n_elements = x.numel()
+    y = torch.empty_like(x)
+    grid = lambda meta: (triton.cdiv(n_elements, meta["BLOCK_SIZE"]),)
+    gelu_kernel[grid](x, y, n_elements, BLOCK_SIZE=1024)
+    return y
+'''
+
+GELU_TORCH_CODE = r'''
+import torch
+import torch.nn.functional as F
+
+
+def gelu(x: torch.Tensor) -> torch.Tensor:
+    return F.gelu(x, approximate="tanh")
+'''
+
+GELU_TEST_FUNC = r'''
+import torch
+from kernel_module import gelu as triton_gelu
+from torch_module import gelu as torch_gelu
+
+
+def test():
+    shapes = [
+        (1024,), (4096,), (16384,), (65536,), (262144,),
+        (1024, 1024), (4096, 4096),
+    ]
+    total = len(shapes)
+    passed = 0
+    for shape in shapes:
+        x = torch.randn(*shape, device="cuda", dtype=torch.float32)
+        out_triton = triton_gelu(x)
+        out_torch = torch_gelu(x)
+        if torch.allclose(out_triton, out_torch, rtol=1e-3, atol=1e-5):
+            passed += 1
+        else:
+            diff = (out_triton - out_torch).abs().max().item()
+            raise AssertionError(f"Shape {shape}: max_diff={diff:.6e}")
+    return {"total_tests": total, "passed_tests": passed, "failed_tests": 0, "errors": []}
+'''
+
+GELU_BENCHMARK_FUNC = r'''
+import torch
+from bench.sandbox.test.test_parametrize import parametrize
+from kernel_module import gelu as triton_gelu
+from torch_module import gelu as torch_gelu
+
+
+@parametrize("dtype, shape", [
+    (torch.float16, (1024,)),
+    (torch.float16, (256, 256)),
+    (torch.float16, (4, 512, 512)),
+    (torch.float16, (8, 16, 64, 64)),
+    (torch.float32, (1024,)),
+    (torch.float32, (256, 256)),
+    (torch.float32, (4, 512, 512)),
+    (torch.float32, (8, 16, 64, 64)),
+    (torch.bfloat16, (1024,)),
+    (torch.bfloat16, (256, 256)),
+    (torch.bfloat16, (4, 512, 512)),
+    (torch.bfloat16, (8, 16, 64, 64)),
+])
+def benchmark_gelu(dtype, shape):
+    x = torch.randn(*shape, device="cuda", dtype=dtype)
+    n_runs = 100
+
+    # Warmup
+    for _ in range(10):
+        triton_gelu(x)
+        torch_gelu(x)
+    torch.cuda.synchronize()
+
+    # Triton
+    start = torch.cuda.Event(enable_timing=True)
+    end = torch.cuda.Event(enable_timing=True)
+    start.record()
+    for _ in range(n_runs):
+        triton_gelu(x)
+    end.record()
+    torch.cuda.synchronize()
+    triton_time_ms = start.elapsed_time(end) / n_runs
+
+    # PyTorch
+    start = torch.cuda.Event(enable_timing=True)
+    end = torch.cuda.Event(enable_timing=True)
+    start.record()
+    for _ in range(n_runs):
+        torch_gelu(x)
+    end.record()
+    torch.cuda.synchronize()
+    torch_time_ms = start.elapsed_time(end) / n_runs
+
+    speedup = torch_time_ms / triton_time_ms if triton_time_ms > 0 else 0.0
+    return {
+        "speedup": speedup,
+        "ref_time": torch_time_ms,
+        "res_time": triton_time_ms,
+    }
+'''
+
+# ============================================================
+# 矩阵乘法 — 正确性验证
+# ============================================================
+MATMUL_TRITON_CODE = r'''
+import torch
+import triton
+import triton.language as tl
+
+
+@triton.jit
+def matmul_kernel(
+    A_ptr, B_ptr, C_ptr, M, N, K,
+    stride_am, stride_ak, stride_bk, stride_bn, stride_cm, stride_cn,
+    BLOCK_M: tl.constexpr, BLOCK_N: tl.constexpr, BLOCK_K: tl.constexpr,
+):
+    pid_m = tl.program_id(0)
+    pid_n = tl.program_id(1)
+    rm = pid_m * BLOCK_M + tl.arange(0, BLOCK_M)
+    rn = pid_n * BLOCK_N + tl.arange(0, BLOCK_N)
+    rk = tl.arange(0, BLOCK_K)
+    A = A_ptr + rm[:, None] * stride_am + rk[None, :] * stride_ak
+    B = B_ptr + rk[:, None] * stride_bk + rn[None, :] * stride_bn
+    acc = tl.zeros((BLOCK_M, BLOCK_N), dtype=tl.float32)
+    for k in range(0, K, BLOCK_K):
+        a = tl.load(A, mask=(rm[:, None] < M) & (rk[None, :] < K - k), other=0.0)
+        b = tl.load(B, mask=(rk[:, None] < K - k) & (rn[None, :] < N), other=0.0)
+        acc += tl.dot(a, b)
+        A += BLOCK_K * stride_ak
+        B += BLOCK_K * stride_bk
+    c = acc.to(tl.float16)
+    C = C_ptr + rm[:, None] * stride_cm + rn[None, :] * stride_cn
+    tl.store(C, c, mask=(rm[:, None] < M) & (rn[None, :] < N))
+
+
+def matmul(A: torch.Tensor, B: torch.Tensor) -> torch.Tensor:
+    M, K1 = A.shape
+    K2, N = B.shape
+    assert K1 == K2
+    C = torch.empty((M, N), device=A.device, dtype=A.dtype)
+    grid = lambda meta: (triton.cdiv(M, meta["BLOCK_M"]), triton.cdiv(N, meta["BLOCK_N"]))
+    matmul_kernel[grid](
+        A, B, C, M, N, K1,
+        A.stride(0), A.stride(1),
+        B.stride(0), B.stride(1),
+        C.stride(0), C.stride(1),
+        BLOCK_M=64, BLOCK_N=64, BLOCK_K=32,
+    )
+    return C
+'''
+
+MATMUL_TORCH_CODE = r'''
+import torch
+
+
+def matmul(A: torch.Tensor, B: torch.Tensor) -> torch.Tensor:
+    return A @ B
+'''
+
+MATMUL_TEST_FUNC = r'''
+import torch
+from kernel_module import matmul as triton_matmul
+from torch_module import matmul as torch_matmul
+
+
+def test():
+    configs = [(128, 256, 128), (256, 512, 256), (512, 1024, 512)]
+    total = len(configs)
+    passed = 0
+    for M, N, K in configs:
+        A = torch.randn(M, K, device="cuda", dtype=torch.float16)
+        B = torch.randn(K, N, device="cuda", dtype=torch.float16)
+        out_triton = triton_matmul(A, B)
+        out_torch = torch_matmul(A, B)
+        if torch.allclose(out_triton, out_torch, rtol=1e-2, atol=1e-2):
+            passed += 1
+        else:
+            diff = (out_triton.float() - out_torch.float()).abs().max().item()
+            raise AssertionError(f"Shape ({M},{N},{K}): max_diff={diff:.6e}")
+    return {"total_tests": total, "passed_tests": passed, "failed_tests": 0, "errors": []}
+'''
+
+# ============================================================
+# 向量加法 — 正确性验证
+# ============================================================
+ADD_TRITON_CODE = r'''
+import torch
+import triton
+import triton.language as tl
+
+
+@triton.jit
+def add_kernel(x_ptr, y_ptr, out_ptr, n_elements, BLOCK_SIZE: tl.constexpr):
+    pid = tl.program_id(0)
+    block_start = pid * BLOCK_SIZE
+    offsets = block_start + tl.arange(0, BLOCK_SIZE)
+    mask = offsets < n_elements
+    x = tl.load(x_ptr + offsets, mask=mask)
+    y = tl.load(y_ptr + offsets, mask=mask)
+    tl.store(out_ptr + offsets, x + y, mask=mask)
+
+
+def add(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
+    n_elements = x.numel()
+    out = torch.empty_like(x)
+    grid = lambda meta: (triton.cdiv(n_elements, meta["BLOCK_SIZE"]),)
+    add_kernel[grid](x, y, out, n_elements, BLOCK_SIZE=1024)
+    return out
+'''
+
+ADD_TORCH_CODE = r'''
+import torch
+
+
+def add(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
+    return x + y
+'''
+
+ADD_TEST_FUNC = r'''
+import torch
+from kernel_module import add as triton_add
+from torch_module import add as torch_add
+
+
+def test():
+    total = 3
+    passed = 0
+    for size in [4096 * 4096, 8192 * 8192, 16384 * 4096]:
+        x = torch.randn(size, device="cuda", dtype=torch.float32)
+        y = torch.randn(size, device="cuda", dtype=torch.float32)
+        out_triton = triton_add(x, y)
+        out_torch = torch_add(x, y)
+        if torch.allclose(out_triton, out_torch):
+            passed += 1
+        else:
+            diff = (out_triton - out_torch).abs().max().item()
+            raise AssertionError(f"Size {size}: max_diff={diff:.6e}")
+    return {"total_tests": total, "passed_tests": passed, "failed_tests": 0, "errors": []}
+'''
+
+
+# ============================================================
+# 算子注册表
+# ============================================================
+KERNELS = {
+    "gelu": {
+        "kernel_name": "gelu",
+        "triton_code": GELU_TRITON_CODE,
+        "torch_code": GELU_TORCH_CODE,
+        "test_func_code": GELU_TEST_FUNC,
+        "benchmark_func_code": GELU_BENCHMARK_FUNC,
+        "desc": "GELU 激活函数 (Triton融合实现，参数化benchmark)",
+    },
+    "matmul": {
+        "kernel_name": "matmul",
+        "triton_code": MATMUL_TRITON_CODE,
+        "torch_code": MATMUL_TORCH_CODE,
+        "test_func_code": MATMUL_TEST_FUNC,
+        "benchmark_func_code": None,
+        "desc": "矩阵乘法 FP16 (正确性验证)",
+    },
+    "add": {
+        "kernel_name": "add",
+        "triton_code": ADD_TRITON_CODE,
+        "torch_code": ADD_TORCH_CODE,
+        "test_func_code": ADD_TEST_FUNC,
+        "benchmark_func_code": None,
+        "desc": "向量加法 (正确性验证)",
+    },
+}
+
+
+# ============================================================
+# 测试编排
+# ============================================================
+
+def run_accuracy_verify(base_url: str, info: dict, chip: str, gpu_id: int) -> dict:
+    """
+    Step 1: 正确性验证 — POST /verify (verify_type=accuracy)
+    """
+    pp = lambda code: apply_device_postprocess(code, chip)
+
+    payload = {
+        "kernel_name": info["kernel_name"],
+        "triton_code": pp(info["triton_code"]),
+        "torch_code": pp(info["torch_code"]),
+        "test_func_code": pp(info["test_func_code"]),
+        "verify_type": "accuracy",
+        "gpu_id": gpu_id,
+        "timeout": 300,
+    }
+
+    url = f"{base_url.rstrip('/')}/verify"
+    print(f"  [1/2] 正确性验证 → POST {url}")
+    sys.stdout.flush()
+
+    t0 = time.time()
+    result = http_post(url, payload, timeout=600)
+    elapsed = time.time() - t0
+
+    passed = result.get("passed", False)
+    total = result.get("total_tests", 0)
+    passed_tests = result.get("passed_tests", 0)
+    failed_tests = result.get("failed_tests", 0)
+
+    icon = "✅" if passed else "❌"
+    print(f"       {icon} passed={passed}, {passed_tests}/{total} 通过, {failed_tests} 失败, 耗时 {elapsed:.2f}s")
+
+    if result.get("errors"):
+        for e in result["errors"][:5]:
+            print(f"       ⚠️  {e[:160]}")
+
+    return result
+
+
+def run_performance_verify(base_url: str, info: dict, chip: str, gpu_id: int) -> dict:
+    """
+    Step 2: 性能验证 — POST /verify/triton (含 benchmark_func_code)
+    返回 legacy 格式，包含 speedup 列表和 html 表格
+    """
+    pp = lambda code: apply_device_postprocess(code, chip)
+    triton_code = pp(info["triton_code"])
+    torch_code = pp(info["torch_code"])
+    benchmark_code = pp(info.get("benchmark_func_code", ""))
+
+    payload = {
+        "triton_kernel_name": info["kernel_name"],
+        "triton_kernel_code": triton_code,
+        "torch_kernel_name": info["kernel_name"],
+        "torch_kernel_code": torch_code,
+        "test_func_code": "",
+        "benchmark_func_code": benchmark_code,
+        "language": "zh_CN",
+        "chip": chip,
+    }
+
+    url = f"{base_url.rstrip('/')}/verify/triton"
+    print(f"  [2/2] 性能验证 → POST {url}")
+    if benchmark_code:
+        lines = [l for l in benchmark_code.split('\n') if '@parametrize' in l]
+        if lines:
+            print(f"       @parametrize: {lines[0].strip()[:120]}...")
+    sys.stdout.flush()
+
+    t0 = time.time()
+    result = http_post(url, payload, timeout=600)
+    elapsed = time.time() - t0
+
+    speedup = result.get("speedup")
+    if speedup and isinstance(speedup, list):
+        # 显示汇总
+        avg_row = next((s for s in speedup if s.get("params") == "avg"), None)
+        if avg_row:
+            avg_sp = avg_row.get("speedup", 0)
+            icon = "🚀" if avg_sp >= 1.2 else ("✅" if avg_sp >= 1.0 else "⚠️")
+            print(f"       {icon} 平均加速比: {avg_sp:.3f}x, 耗时 {elapsed:.2f}s")
+        else:
+            print(f"       总条目: {len(speedup)}, 耗时 {elapsed:.2f}s")
+    elif result.get("success") is False:
+        print(f"       ❌ 失败: {result.get('traceback', '')[:200]}")
+    else:
+        print(f"       耗时 {elapsed:.2f}s")
+
+    # 打印 HTML 表格
+    info_data = result.get("info", {})
+    html = info_data.get("html", "")
+    if html:
+        print()
+        print(_render_html_table(html))
+
+    return result
+
+
+def _render_html_table(html: str) -> str:
+    """从 HTML 中提取 rich 表格文本并格式化到终端"""
+    # 尝试用 ANSI 颜色渲染（如果终端支持）
+    # 回退方案：提取 <pre> 中的纯文本
+    import textwrap
+
+    # 提取 <pre> 内容
+    pre_match = re.search(r'<pre[^>]*>(.*?)</pre>', html, re.DOTALL)
+    if pre_match:
+        text = pre_match.group(1)
+        text = unescape(text)
+        # 清理 HTML 标签（span 等）
+        text = re.sub(r'<[^>]+>', '', text)
+        # 清理多余的空白行
+        lines = text.split('\n')
+        # 过滤掉只有颜色代码的行
+        clean_lines = []
+        for line in lines:
+            stripped = line.strip()
+            if stripped:
+                clean_lines.append(line)
+        text = '\n'.join(clean_lines)
+        return textwrap.indent(text, '     ')
+
+    # 如果没找到 pre，返回 truncate 的 HTML
+    return textwrap.indent(html[:2000], '     ')
+
+
+def run_accuracy_only_test(base_url: str, info: dict, chip: str, gpu_id: int) -> dict:
+    """
+    仅正确性验证（适用于无 benchmark_func_code 的算子，如 matmul / add）
+    走 /verify?verify_type=both 自动做性能测试
+    """
+    pp = lambda code: apply_device_postprocess(code, chip)
+
+    payload = {
+        "kernel_name": info["kernel_name"],
+        "triton_code": pp(info["triton_code"]),
+        "torch_code": pp(info["torch_code"]),
+        "test_func_code": pp(info["test_func_code"]),
+        "verify_type": "both",
+        "gpu_id": gpu_id,
+        "timeout": 300,
+    }
+
+    url = f"{base_url.rstrip('/')}/verify"
+    print(f"  → POST {url} (verify_type=both)")
+    sys.stdout.flush()
+
+    t0 = time.time()
+    result = http_post(url, payload, timeout=600)
+    elapsed = time.time() - t0
+
+    passed = result.get("passed", False)
+    total = result.get("total_tests", 0)
+    passed_tests = result.get("passed_tests", 0)
+    speedup = result.get("speedup")
+
+    icon = "✅" if passed else "❌"
+    parts = [f"{icon} passed={passed}", f"{passed_tests}/{total} 通过"]
+    if speedup is not None:
+        sp_icon = "🚀" if speedup >= 1.2 else ("✅" if speedup >= 1.0 else "⚠️")
+        parts.append(f"加速比={speedup:.3f}x {sp_icon}")
+    parts.append(f"耗时 {elapsed:.2f}s")
+    print(f"       {' | '.join(parts)}")
+
+    if result.get("errors"):
+        for e in result["errors"][:3]:
+            print(f"       ⚠️  {e[:160]}")
+
+    return result
+
+
+def check_health(base_url: str):
+    data = http_get(f"{base_url.rstrip('/')}/health")
+    print(f"  Health:   {json.dumps(data, ensure_ascii=False) if 'error' not in data else '❌ 不可达'}")
+
+    gpu = http_get(f"{base_url.rstrip('/')}/gpu/status")
+    if "error" not in gpu:
+        print(f"  Vendor:   {gpu.get('vendor', '?')}")
+        print(f"  Devices:  {gpu.get('device_count', 0)} total, {gpu.get('available_devices', 0)} available")
+        for d in gpu.get("devices", []):
+            print(f"    GPU {d['device_id']}: {d['name']} ({d['free_memory_mb']}/{d['total_memory_mb']} MB free)")
+    else:
+        print(f"  GPU:      ⚠️  {gpu.get('message', '')}")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="KernelGen 验证服务测试",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""示例:
+  python test_verify_service.py --url http://172.24.4.45:8912 --chip nvidia
+  python test_verify_service.py --url http://10.7.64.131:8912 --chip huawei --all
+  python test_verify_service.py --url http://10.7.66.1:8912 --chip moore --kernel gelu""",
+    )
+    parser.add_argument("--url", default="http://localhost:8912", help="验证服务地址")
+    parser.add_argument("--chip", default="nvidia", choices=list(CHIP_CONFIG.keys()), help="芯片类型")
+    parser.add_argument("--gpu-id", type=int, default=-1, help="GPU ID")
+    parser.add_argument("--all", action="store_true", help="测试所有算子")
+    parser.add_argument("--kernel", default="gelu", choices=list(KERNELS.keys()), help="指定算子")
+    parser.add_argument("--health-only", action="store_true", help="仅检查健康状态")
+    args = parser.parse_args()
+
+    base_url = args.url.rstrip("/")
+    chip = args.chip
+    cfg = CHIP_CONFIG[chip]
+
+    print("=" * 72)
+    print("  KernelGen 验证服务测试")
+    print("=" * 72)
+    print(f"  服务:  {base_url}")
+    print(f"  芯片:  {chip} ({cfg['desc']})")
+    print(f"  设备:  {cfg['device_keyword']}" +
+          (f" (cuda → {cfg['device_keyword']})" if cfg['device_keyword'] != 'cuda' else " (CUDA兼容, 不替换)"))
+    print(f"  GPU:   {args.gpu_id}" + (" (自动)" if args.gpu_id == -1 else ""))
+    print()
+
+    check_health(base_url)
+
+    if args.health_only:
+        return
+
+    kernels_to_test = list(KERNELS.keys()) if args.all else [args.kernel]
+
+    all_passed = True
+    for name in kernels_to_test:
+        info = KERNELS[name]
+
+        print(f"\n{'─' * 72}")
+        print(f"  [{name}] {info['desc']}")
+        print(f"{'─' * 72}")
+
+        if info.get("benchmark_func_code"):
+            # GELU: 正确性 + 参数化性能
+            acc_result = run_accuracy_verify(base_url, info, chip, args.gpu_id)
+            if not acc_result.get("passed"):
+                all_passed = False
+            run_performance_verify(base_url, info, chip, args.gpu_id)
+        else:
+            # matmul / add: 正确性
+            result = run_accuracy_only_test(base_url, info, chip, args.gpu_id)
+            if not result.get("passed"):
+                all_passed = False
+
+    print(f"\n{'=' * 72}")
+    print(f"  测试完成  |  芯片: {chip}  |  服务: {base_url}")
+    print(f"{'=' * 72}")
+
+    sys.exit(0 if all_passed else 1)
+
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file

From c5f8655e249798e16a0810f93099b592769beec1 Mon Sep 17 00:00:00 2001
From: sunnycase <sunnycase@live.cn>
Date: Tue, 9 Jun 2026 04:10:13 +0000
Subject: [PATCH 3/6] test: run mthreads kernels directly

---
 .github/workflows/mthreads-build-and-test.yml |  12 +-
 .../python/test/test_verify_service.py        | 805 ++++++------------
 2 files changed, 277 insertions(+), 540 deletions(-)

diff --git a/.github/workflows/mthreads-build-and-test.yml b/.github/workflows/mthreads-build-and-test.yml
index 6d89a99549..c125431aa4 100644
--- a/.github/workflows/mthreads-build-and-test.yml
+++ b/.github/workflows/mthreads-build-and-test.yml
@@ -47,17 +47,15 @@ jobs:
           set -x
           python3 -m pytest -s third_party/mthreads/python/test/unit --device musa
 
-      - name: Verify Service Test on Mthreads
+      - name: Direct Kernel Test on Mthreads
         if: steps.check_backend.outputs.should_skip != 'true'
         shell: bash
         env:
-          MTHREADS_VERIFY_SERVICE_URL: ${{ vars.MTHREADS_VERIFY_SERVICE_URL }}
-          MTHREADS_VERIFY_SERVICE_KERNEL: ${{ vars.MTHREADS_VERIFY_SERVICE_KERNEL }}
+          MTHREADS_KERNEL_TEST_KERNEL: ${{ vars.MTHREADS_KERNEL_TEST_KERNEL }}
         run: |
           set -x
-          verify_service_url="${MTHREADS_VERIFY_SERVICE_URL:-http://10.7.66.1:8912}"
-          verify_service_kernel="${MTHREADS_VERIFY_SERVICE_KERNEL:-gelu}"
+          export TRITON_DEFAULT_BACKEND=mthreads
+          kernel="${MTHREADS_KERNEL_TEST_KERNEL:-gelu}"
           python3 third_party/mthreads/python/test/test_verify_service.py \
-            --url "${verify_service_url}" \
             --chip moore \
-            --kernel "${verify_service_kernel}"
+            --kernel "${kernel}"
diff --git a/third_party/mthreads/python/test/test_verify_service.py b/third_party/mthreads/python/test/test_verify_service.py
index c184215b08..f6866efa60 100644
--- a/third_party/mthreads/python/test/test_verify_service.py
+++ b/third_party/mthreads/python/test/test_verify_service.py
@@ -1,99 +1,37 @@
 """
-验证服务测试脚本 — 支持多芯片、参数化 benchmark、HTML 性能表格
-
-用法:
-    python test_verify_service.py --url http://172.24.4.45:8912 --chip nvidia
-    python test_verify_service.py --url http://10.7.64.131:8912 --chip huawei --all
-    python test_verify_service.py --url http://10.7.66.1:8912 --chip moore --kernel gelu
-
-芯片:
-    nvidia / haiguang / tianshu / muxi / pingtouge  → CUDA 兼容, 代码不变
-    huawei → cuda 替换为 npu
-    moore  → cuda 替换为 musa
-
-测试算子:
-    gelu    GELU 激活函数 (参数化 benchmark: 12 组 dtype×shape)
-    matmul  矩阵乘法 FP16 (正确性验证)
-    add     向量加法 (正确性验证)
+Direct kernel verification for mthreads/MUSA.
+
+Examples:
+    python test_verify_service.py --chip moore --kernel gelu
+    python test_verify_service.py --chip moore --all
+    python test_verify_service.py --chip moore --kernel gelu --benchmark
 """
 
 import argparse
-import json
-import re
+import os
 import sys
 import time
-import urllib.request
-import urllib.error
-from html import unescape
-from typing import Dict, Optional
-
-
-# ============================================================
-# 芯片配置
-# ============================================================
-CHIP_CONFIG = {
-    "nvidia":     {"device_keyword": "cuda",  "desc": "NVIDIA CUDA"},
-    "haiguang":   {"device_keyword": "cuda",  "desc": "海光 DCU (CUDA兼容)"},
-    "tianshu":    {"device_keyword": "cuda",  "desc": "天数智芯 (CUDA兼容)"},
-    "muxi":       {"device_keyword": "cuda",  "desc": "沐曦 (CUDA兼容)"},
-    "pingtouge":  {"device_keyword": "cuda",  "desc": "平头哥 (CUDA兼容)"},
-    "huawei":     {"device_keyword": "npu",   "desc": "华为昇腾 Ascend"},
-    "moore":      {"device_keyword": "musa",  "desc": "摩尔线程 MUSA"},
-}
+from dataclasses import dataclass
+from typing import Callable, Iterable
 
-
-# ============================================================
-# HTTP 工具
-# ============================================================
-def http_post(url: str, data: dict, timeout: int = 600) -> dict:
-    body = json.dumps(data).encode("utf-8")
-    req = urllib.request.Request(
-        url, data=body,
-        headers={"Content-Type": "application/json"},
-        method="POST",
-    )
-    try:
-        with urllib.request.urlopen(req, timeout=timeout) as resp:
-            return json.loads(resp.read().decode("utf-8"))
-    except urllib.error.HTTPError as e:
-        return {"error": True, "message": f"HTTP {e.code}: {e.reason}",
-                "detail": e.read().decode("utf-8", errors="replace")}
-    except urllib.error.URLError as e:
-        return {"error": True, "message": f"Connection failed: {e.reason}"}
-    except Exception as e:
-        return {"error": True, "message": str(e)}
-
-
-def http_get(url: str, timeout: int = 10) -> dict:
-    req = urllib.request.Request(url)
-    try:
-        with urllib.request.urlopen(req, timeout=timeout) as resp:
-            return json.loads(resp.read().decode("utf-8"))
-    except Exception as e:
-        return {"error": True, "message": str(e)}
-
-
-# ============================================================
-# 设备后处理
-# ============================================================
-def apply_device_postprocess(code: str, chip: str) -> str:
-    """将代码中的 cuda 替换为目标芯片的设备关键字"""
-    if not code:
-        return code
-    cfg = CHIP_CONFIG.get(chip, {})
-    target = cfg.get("device_keyword", "cuda")
-    if target == "cuda":
-        return code
-    return re.sub(r'\bcuda\b', target, code)
-
-
-# ============================================================
-# GELU — Triton 融合实现 (参数化 benchmark)
-# ============================================================
-GELU_TRITON_CODE = r'''
 import torch
+import torch.nn.functional as F
 import triton
 import triton.language as tl
+import triton.testing as triton_testing
+
+__test__ = False
+
+
+CHIP_CONFIG = {
+    "nvidia": {"device_keyword": "cuda", "desc": "NVIDIA CUDA"},
+    "haiguang": {"device_keyword": "cuda", "desc": "Haiguang DCU (CUDA-compatible)"},
+    "tianshu": {"device_keyword": "cuda", "desc": "Tianshu (CUDA-compatible)"},
+    "muxi": {"device_keyword": "cuda", "desc": "Muxi (CUDA-compatible)"},
+    "pingtouge": {"device_keyword": "cuda", "desc": "Pingtouge (CUDA-compatible)"},
+    "huawei": {"device_keyword": "npu", "desc": "Huawei Ascend"},
+    "moore": {"device_keyword": "musa", "desc": "Moore Threads MUSA"},
+}
 
 
 @triton.jit
@@ -107,8 +45,6 @@ def gelu_kernel(x_ptr, y_ptr, n_elements, BLOCK_SIZE: tl.constexpr):
     coeff = 0.044715
     x3 = x * x * x
     inner = sqrt_2_over_pi * (x + coeff * x3)
-    # tanh(x) = (exp(2x) - 1) / (exp(2x) + 1), avoids tl.math.tanh
-    # which is unavailable on some non-NVIDIA Triton backends (e.g. Muxi/Corex)
     exp2x = tl.math.exp(2.0 * inner)
     tanh_inner = (exp2x - 1.0) / (exp2x + 1.0)
     y = 0.5 * x * (1.0 + tanh_inner)
@@ -121,115 +57,25 @@ def gelu(x: torch.Tensor) -> torch.Tensor:
     grid = lambda meta: (triton.cdiv(n_elements, meta["BLOCK_SIZE"]),)
     gelu_kernel[grid](x, y, n_elements, BLOCK_SIZE=1024)
     return y
-'''
-
-GELU_TORCH_CODE = r'''
-import torch
-import torch.nn.functional as F
-
-
-def gelu(x: torch.Tensor) -> torch.Tensor:
-    return F.gelu(x, approximate="tanh")
-'''
-
-GELU_TEST_FUNC = r'''
-import torch
-from kernel_module import gelu as triton_gelu
-from torch_module import gelu as torch_gelu
-
-
-def test():
-    shapes = [
-        (1024,), (4096,), (16384,), (65536,), (262144,),
-        (1024, 1024), (4096, 4096),
-    ]
-    total = len(shapes)
-    passed = 0
-    for shape in shapes:
-        x = torch.randn(*shape, device="cuda", dtype=torch.float32)
-        out_triton = triton_gelu(x)
-        out_torch = torch_gelu(x)
-        if torch.allclose(out_triton, out_torch, rtol=1e-3, atol=1e-5):
-            passed += 1
-        else:
-            diff = (out_triton - out_torch).abs().max().item()
-            raise AssertionError(f"Shape {shape}: max_diff={diff:.6e}")
-    return {"total_tests": total, "passed_tests": passed, "failed_tests": 0, "errors": []}
-'''
-
-GELU_BENCHMARK_FUNC = r'''
-import torch
-from bench.sandbox.test.test_parametrize import parametrize
-from kernel_module import gelu as triton_gelu
-from torch_module import gelu as torch_gelu
-
-
-@parametrize("dtype, shape", [
-    (torch.float16, (1024,)),
-    (torch.float16, (256, 256)),
-    (torch.float16, (4, 512, 512)),
-    (torch.float16, (8, 16, 64, 64)),
-    (torch.float32, (1024,)),
-    (torch.float32, (256, 256)),
-    (torch.float32, (4, 512, 512)),
-    (torch.float32, (8, 16, 64, 64)),
-    (torch.bfloat16, (1024,)),
-    (torch.bfloat16, (256, 256)),
-    (torch.bfloat16, (4, 512, 512)),
-    (torch.bfloat16, (8, 16, 64, 64)),
-])
-def benchmark_gelu(dtype, shape):
-    x = torch.randn(*shape, device="cuda", dtype=dtype)
-    n_runs = 100
-
-    # Warmup
-    for _ in range(10):
-        triton_gelu(x)
-        torch_gelu(x)
-    torch.cuda.synchronize()
-
-    # Triton
-    start = torch.cuda.Event(enable_timing=True)
-    end = torch.cuda.Event(enable_timing=True)
-    start.record()
-    for _ in range(n_runs):
-        triton_gelu(x)
-    end.record()
-    torch.cuda.synchronize()
-    triton_time_ms = start.elapsed_time(end) / n_runs
-
-    # PyTorch
-    start = torch.cuda.Event(enable_timing=True)
-    end = torch.cuda.Event(enable_timing=True)
-    start.record()
-    for _ in range(n_runs):
-        torch_gelu(x)
-    end.record()
-    torch.cuda.synchronize()
-    torch_time_ms = start.elapsed_time(end) / n_runs
-
-    speedup = torch_time_ms / triton_time_ms if triton_time_ms > 0 else 0.0
-    return {
-        "speedup": speedup,
-        "ref_time": torch_time_ms,
-        "res_time": triton_time_ms,
-    }
-'''
-
-# ============================================================
-# 矩阵乘法 — 正确性验证
-# ============================================================
-MATMUL_TRITON_CODE = r'''
-import torch
-import triton
-import triton.language as tl
 
 
 @triton.jit
 def matmul_kernel(
-    A_ptr, B_ptr, C_ptr, M, N, K,
-    stride_am, stride_ak, stride_bk, stride_bn, stride_cm, stride_cn,
-    BLOCK_M: tl.constexpr, BLOCK_N: tl.constexpr, BLOCK_K: tl.constexpr,
+    A_ptr,
+    B_ptr,
+    C_ptr,
+    M,
+    N,
+    K,
+    stride_am,
+    stride_ak,
+    stride_bk,
+    stride_bn,
+    stride_cm,
+    stride_cn,
+    BLOCK_M: tl.constexpr,
+    BLOCK_N: tl.constexpr,
+    BLOCK_K: tl.constexpr,
 ):
     pid_m = tl.program_id(0)
     pid_n = tl.program_id(1)
@@ -253,57 +99,28 @@ def matmul_kernel(
 def matmul(A: torch.Tensor, B: torch.Tensor) -> torch.Tensor:
     M, K1 = A.shape
     K2, N = B.shape
-    assert K1 == K2
+    if K1 != K2:
+        raise ValueError(f"incompatible matmul shapes: {A.shape} and {B.shape}")
     C = torch.empty((M, N), device=A.device, dtype=A.dtype)
     grid = lambda meta: (triton.cdiv(M, meta["BLOCK_M"]), triton.cdiv(N, meta["BLOCK_N"]))
     matmul_kernel[grid](
-        A, B, C, M, N, K1,
-        A.stride(0), A.stride(1),
-        B.stride(0), B.stride(1),
-        C.stride(0), C.stride(1),
-        BLOCK_M=64, BLOCK_N=64, BLOCK_K=32,
+        A,
+        B,
+        C,
+        M,
+        N,
+        K1,
+        A.stride(0),
+        A.stride(1),
+        B.stride(0),
+        B.stride(1),
+        C.stride(0),
+        C.stride(1),
+        BLOCK_M=64,
+        BLOCK_N=64,
+        BLOCK_K=32,
     )
     return C
-'''
-
-MATMUL_TORCH_CODE = r'''
-import torch
-
-
-def matmul(A: torch.Tensor, B: torch.Tensor) -> torch.Tensor:
-    return A @ B
-'''
-
-MATMUL_TEST_FUNC = r'''
-import torch
-from kernel_module import matmul as triton_matmul
-from torch_module import matmul as torch_matmul
-
-
-def test():
-    configs = [(128, 256, 128), (256, 512, 256), (512, 1024, 512)]
-    total = len(configs)
-    passed = 0
-    for M, N, K in configs:
-        A = torch.randn(M, K, device="cuda", dtype=torch.float16)
-        B = torch.randn(K, N, device="cuda", dtype=torch.float16)
-        out_triton = triton_matmul(A, B)
-        out_torch = torch_matmul(A, B)
-        if torch.allclose(out_triton, out_torch, rtol=1e-2, atol=1e-2):
-            passed += 1
-        else:
-            diff = (out_triton.float() - out_torch.float()).abs().max().item()
-            raise AssertionError(f"Shape ({M},{N},{K}): max_diff={diff:.6e}")
-    return {"total_tests": total, "passed_tests": passed, "failed_tests": 0, "errors": []}
-'''
-
-# ============================================================
-# 向量加法 — 正确性验证
-# ============================================================
-ADD_TRITON_CODE = r'''
-import torch
-import triton
-import triton.language as tl
 
 
 @triton.jit
@@ -323,322 +140,244 @@ def add(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
     grid = lambda meta: (triton.cdiv(n_elements, meta["BLOCK_SIZE"]),)
     add_kernel[grid](x, y, out, n_elements, BLOCK_SIZE=1024)
     return out
-'''
 
-ADD_TORCH_CODE = r'''
-import torch
 
+@dataclass(frozen=True)
+class KernelSpec:
+    name: str
+    desc: str
+    runner: Callable[[str, str], int]
+    benchmark: Callable[[str, str, int, int], None] | None = None
+
+
+def _import_backend_extension(device_type: str) -> None:
+    if device_type == "musa":
+        try:
+            import torch_musa  # noqa: F401
+        except ImportError:
+            pass
+    elif device_type == "npu":
+        try:
+            import torch_npu  # noqa: F401
+        except ImportError:
+            pass
+
+
+def _configure_device(chip: str, gpu_id: int) -> tuple[str, str]:
+    device_type = CHIP_CONFIG[chip]["device_keyword"]
+    if device_type == "musa":
+        os.environ.setdefault("TRITON_DEFAULT_BACKEND", "mthreads")
+
+    _import_backend_extension(device_type)
+    device_interface = getattr(torch, device_type, None)
+    if device_interface is None:
+        raise RuntimeError(f"torch.{device_type} is not available")
+
+    is_available = getattr(device_interface, "is_available", None)
+    if callable(is_available) and not is_available():
+        raise RuntimeError(f"{device_type} device is not available")
+
+    if gpu_id >= 0:
+        set_device = getattr(device_interface, "set_device", None)
+        if callable(set_device):
+            set_device(gpu_id)
+        return device_type, f"{device_type}:{gpu_id}"
+    return device_type, device_type
+
+
+def _synchronize(device_type: str) -> None:
+    device_interface = getattr(torch, device_type)
+    synchronize = getattr(device_interface, "synchronize", None)
+    if callable(synchronize):
+        synchronize()
+
+
+def _assert_close(
+    label: str,
+    actual: torch.Tensor,
+    expected: torch.Tensor,
+    *,
+    rtol: float,
+    atol: float,
+) -> None:
+    if torch.allclose(actual, expected, rtol=rtol, atol=atol):
+        return
+    diff = (actual.float() - expected.float()).abs().max().item()
+    raise AssertionError(f"{label}: max_diff={diff:.6e}, rtol={rtol}, atol={atol}")
 
-def add(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
-    return x + y
-'''
 
-ADD_TEST_FUNC = r'''
-import torch
-from kernel_module import add as triton_add
-from torch_module import add as torch_add
+def _run_cases(
+    name: str,
+    cases: Iterable,
+    run_case: Callable[[object], None],
+) -> int:
+    passed = 0
+    for case in cases:
+        t0 = time.time()
+        run_case(case)
+        elapsed = time.time() - t0
+        print(f"    PASS {name} {case} ({elapsed:.2f}s)")
+        passed += 1
+    return passed
+
+
+def run_gelu(device_type: str, device: str) -> int:
+    shapes = [
+        (1024,),
+        (4096,),
+        (16384,),
+        (65536,),
+        (262144,),
+        (1024, 1024),
+        (4096, 4096),
+    ]
+
+    def case(shape):
+        x = torch.randn(*shape, device=device, dtype=torch.float32)
+        actual = gelu(x)
+        expected = F.gelu(x, approximate="tanh")
+        _synchronize(device_type)
+        _assert_close(f"gelu shape={shape}", actual, expected, rtol=1e-3, atol=1e-5)
+
+    return _run_cases("gelu", shapes, case)
+
+
+def benchmark_gelu(device_type: str, device: str, warmup_ms: int, rep_ms: int) -> None:
+    cases = [
+        (torch.float16, (1024,)),
+        (torch.float16, (256, 256)),
+        (torch.float16, (4, 512, 512)),
+        (torch.float16, (8, 16, 64, 64)),
+        (torch.float32, (1024,)),
+        (torch.float32, (256, 256)),
+        (torch.float32, (4, 512, 512)),
+        (torch.float32, (8, 16, 64, 64)),
+        (torch.bfloat16, (1024,)),
+        (torch.bfloat16, (256, 256)),
+        (torch.bfloat16, (4, 512, 512)),
+        (torch.bfloat16, (8, 16, 64, 64)),
+    ]
+
+    print("    dtype        shape              triton_ms  torch_ms  speedup")
+    for dtype, shape in cases:
+        x = torch.randn(*shape, device=device, dtype=dtype)
+        triton_ms = triton_testing.do_bench(
+            lambda: gelu(x),
+            warmup=warmup_ms,
+            rep=rep_ms,
+            device_type=device_type,
+        )
+        torch_ms = triton_testing.do_bench(
+            lambda: F.gelu(x, approximate="tanh"),
+            warmup=warmup_ms,
+            rep=rep_ms,
+            device_type=device_type,
+        )
+        speedup = torch_ms / triton_ms if triton_ms > 0 else 0.0
+        print(f"    {str(dtype):<12} {str(shape):<18} {triton_ms:9.4f} {torch_ms:8.4f} {speedup:7.3f}x")
+
+
+def run_matmul(device_type: str, device: str) -> int:
+    configs = [(128, 256, 128), (256, 512, 256), (512, 1024, 512)]
+
+    def case(config):
+        M, N, K = config
+        A = torch.randn(M, K, device=device, dtype=torch.float16)
+        B = torch.randn(K, N, device=device, dtype=torch.float16)
+        actual = matmul(A, B)
+        expected = A @ B
+        _synchronize(device_type)
+        _assert_close(f"matmul shape=({M},{N},{K})", actual, expected, rtol=1e-2, atol=1e-2)
+
+    return _run_cases("matmul", configs, case)
+
+
+def run_add(device_type: str, device: str) -> int:
+    sizes = [4096 * 4096, 8192 * 8192, 16384 * 4096]
+
+    def case(size):
+        x = torch.randn(size, device=device, dtype=torch.float32)
+        y = torch.randn(size, device=device, dtype=torch.float32)
+        actual = add(x, y)
+        expected = x + y
+        _synchronize(device_type)
+        _assert_close(f"add size={size}", actual, expected, rtol=1e-5, atol=1e-8)
+
+    return _run_cases("add", sizes, case)
 
 
-def test():
-    total = 3
-    passed = 0
-    for size in [4096 * 4096, 8192 * 8192, 16384 * 4096]:
-        x = torch.randn(size, device="cuda", dtype=torch.float32)
-        y = torch.randn(size, device="cuda", dtype=torch.float32)
-        out_triton = triton_add(x, y)
-        out_torch = torch_add(x, y)
-        if torch.allclose(out_triton, out_torch):
-            passed += 1
-        else:
-            diff = (out_triton - out_torch).abs().max().item()
-            raise AssertionError(f"Size {size}: max_diff={diff:.6e}")
-    return {"total_tests": total, "passed_tests": passed, "failed_tests": 0, "errors": []}
-'''
-
-
-# ============================================================
-# 算子注册表
-# ============================================================
 KERNELS = {
-    "gelu": {
-        "kernel_name": "gelu",
-        "triton_code": GELU_TRITON_CODE,
-        "torch_code": GELU_TORCH_CODE,
-        "test_func_code": GELU_TEST_FUNC,
-        "benchmark_func_code": GELU_BENCHMARK_FUNC,
-        "desc": "GELU 激活函数 (Triton融合实现，参数化benchmark)",
-    },
-    "matmul": {
-        "kernel_name": "matmul",
-        "triton_code": MATMUL_TRITON_CODE,
-        "torch_code": MATMUL_TORCH_CODE,
-        "test_func_code": MATMUL_TEST_FUNC,
-        "benchmark_func_code": None,
-        "desc": "矩阵乘法 FP16 (正确性验证)",
-    },
-    "add": {
-        "kernel_name": "add",
-        "triton_code": ADD_TRITON_CODE,
-        "torch_code": ADD_TORCH_CODE,
-        "test_func_code": ADD_TEST_FUNC,
-        "benchmark_func_code": None,
-        "desc": "向量加法 (正确性验证)",
-    },
+    "gelu": KernelSpec(
+        name="gelu",
+        desc="GELU activation (Triton direct run)",
+        runner=run_gelu,
+        benchmark=benchmark_gelu,
+    ),
+    "matmul": KernelSpec(
+        name="matmul",
+        desc="FP16 matrix multiplication correctness",
+        runner=run_matmul,
+    ),
+    "add": KernelSpec(
+        name="add",
+        desc="Vector addition correctness",
+        runner=run_add,
+    ),
 }
 
 
-# ============================================================
-# 测试编排
-# ============================================================
-
-def run_accuracy_verify(base_url: str, info: dict, chip: str, gpu_id: int) -> dict:
-    """
-    Step 1: 正确性验证 — POST /verify (verify_type=accuracy)
-    """
-    pp = lambda code: apply_device_postprocess(code, chip)
-
-    payload = {
-        "kernel_name": info["kernel_name"],
-        "triton_code": pp(info["triton_code"]),
-        "torch_code": pp(info["torch_code"]),
-        "test_func_code": pp(info["test_func_code"]),
-        "verify_type": "accuracy",
-        "gpu_id": gpu_id,
-        "timeout": 300,
-    }
-
-    url = f"{base_url.rstrip('/')}/verify"
-    print(f"  [1/2] 正确性验证 → POST {url}")
-    sys.stdout.flush()
-
-    t0 = time.time()
-    result = http_post(url, payload, timeout=600)
-    elapsed = time.time() - t0
-
-    passed = result.get("passed", False)
-    total = result.get("total_tests", 0)
-    passed_tests = result.get("passed_tests", 0)
-    failed_tests = result.get("failed_tests", 0)
-
-    icon = "✅" if passed else "❌"
-    print(f"       {icon} passed={passed}, {passed_tests}/{total} 通过, {failed_tests} 失败, 耗时 {elapsed:.2f}s")
-
-    if result.get("errors"):
-        for e in result["errors"][:5]:
-            print(f"       ⚠️  {e[:160]}")
-
-    return result
-
-
-def run_performance_verify(base_url: str, info: dict, chip: str, gpu_id: int) -> dict:
-    """
-    Step 2: 性能验证 — POST /verify/triton (含 benchmark_func_code)
-    返回 legacy 格式，包含 speedup 列表和 html 表格
-    """
-    pp = lambda code: apply_device_postprocess(code, chip)
-    triton_code = pp(info["triton_code"])
-    torch_code = pp(info["torch_code"])
-    benchmark_code = pp(info.get("benchmark_func_code", ""))
-
-    payload = {
-        "triton_kernel_name": info["kernel_name"],
-        "triton_kernel_code": triton_code,
-        "torch_kernel_name": info["kernel_name"],
-        "torch_kernel_code": torch_code,
-        "test_func_code": "",
-        "benchmark_func_code": benchmark_code,
-        "language": "zh_CN",
-        "chip": chip,
-    }
-
-    url = f"{base_url.rstrip('/')}/verify/triton"
-    print(f"  [2/2] 性能验证 → POST {url}")
-    if benchmark_code:
-        lines = [l for l in benchmark_code.split('\n') if '@parametrize' in l]
-        if lines:
-            print(f"       @parametrize: {lines[0].strip()[:120]}...")
-    sys.stdout.flush()
-
-    t0 = time.time()
-    result = http_post(url, payload, timeout=600)
-    elapsed = time.time() - t0
-
-    speedup = result.get("speedup")
-    if speedup and isinstance(speedup, list):
-        # 显示汇总
-        avg_row = next((s for s in speedup if s.get("params") == "avg"), None)
-        if avg_row:
-            avg_sp = avg_row.get("speedup", 0)
-            icon = "🚀" if avg_sp >= 1.2 else ("✅" if avg_sp >= 1.0 else "⚠️")
-            print(f"       {icon} 平均加速比: {avg_sp:.3f}x, 耗时 {elapsed:.2f}s")
-        else:
-            print(f"       总条目: {len(speedup)}, 耗时 {elapsed:.2f}s")
-    elif result.get("success") is False:
-        print(f"       ❌ 失败: {result.get('traceback', '')[:200]}")
-    else:
-        print(f"       耗时 {elapsed:.2f}s")
-
-    # 打印 HTML 表格
-    info_data = result.get("info", {})
-    html = info_data.get("html", "")
-    if html:
-        print()
-        print(_render_html_table(html))
-
-    return result
-
-
-def _render_html_table(html: str) -> str:
-    """从 HTML 中提取 rich 表格文本并格式化到终端"""
-    # 尝试用 ANSI 颜色渲染（如果终端支持）
-    # 回退方案：提取 <pre> 中的纯文本
-    import textwrap
-
-    # 提取 <pre> 内容
-    pre_match = re.search(r'<pre[^>]*>(.*?)</pre>', html, re.DOTALL)
-    if pre_match:
-        text = pre_match.group(1)
-        text = unescape(text)
-        # 清理 HTML 标签（span 等）
-        text = re.sub(r'<[^>]+>', '', text)
-        # 清理多余的空白行
-        lines = text.split('\n')
-        # 过滤掉只有颜色代码的行
-        clean_lines = []
-        for line in lines:
-            stripped = line.strip()
-            if stripped:
-                clean_lines.append(line)
-        text = '\n'.join(clean_lines)
-        return textwrap.indent(text, '     ')
-
-    # 如果没找到 pre，返回 truncate 的 HTML
-    return textwrap.indent(html[:2000], '     ')
-
-
-def run_accuracy_only_test(base_url: str, info: dict, chip: str, gpu_id: int) -> dict:
-    """
-    仅正确性验证（适用于无 benchmark_func_code 的算子，如 matmul / add）
-    走 /verify?verify_type=both 自动做性能测试
-    """
-    pp = lambda code: apply_device_postprocess(code, chip)
-
-    payload = {
-        "kernel_name": info["kernel_name"],
-        "triton_code": pp(info["triton_code"]),
-        "torch_code": pp(info["torch_code"]),
-        "test_func_code": pp(info["test_func_code"]),
-        "verify_type": "both",
-        "gpu_id": gpu_id,
-        "timeout": 300,
-    }
-
-    url = f"{base_url.rstrip('/')}/verify"
-    print(f"  → POST {url} (verify_type=both)")
-    sys.stdout.flush()
-
-    t0 = time.time()
-    result = http_post(url, payload, timeout=600)
-    elapsed = time.time() - t0
-
-    passed = result.get("passed", False)
-    total = result.get("total_tests", 0)
-    passed_tests = result.get("passed_tests", 0)
-    speedup = result.get("speedup")
-
-    icon = "✅" if passed else "❌"
-    parts = [f"{icon} passed={passed}", f"{passed_tests}/{total} 通过"]
-    if speedup is not None:
-        sp_icon = "🚀" if speedup >= 1.2 else ("✅" if speedup >= 1.0 else "⚠️")
-        parts.append(f"加速比={speedup:.3f}x {sp_icon}")
-    parts.append(f"耗时 {elapsed:.2f}s")
-    print(f"       {' | '.join(parts)}")
-
-    if result.get("errors"):
-        for e in result["errors"][:3]:
-            print(f"       ⚠️  {e[:160]}")
-
-    return result
-
-
-def check_health(base_url: str):
-    data = http_get(f"{base_url.rstrip('/')}/health")
-    print(f"  Health:   {json.dumps(data, ensure_ascii=False) if 'error' not in data else '❌ 不可达'}")
-
-    gpu = http_get(f"{base_url.rstrip('/')}/gpu/status")
-    if "error" not in gpu:
-        print(f"  Vendor:   {gpu.get('vendor', '?')}")
-        print(f"  Devices:  {gpu.get('device_count', 0)} total, {gpu.get('available_devices', 0)} available")
-        for d in gpu.get("devices", []):
-            print(f"    GPU {d['device_id']}: {d['name']} ({d['free_memory_mb']}/{d['total_memory_mb']} MB free)")
-    else:
-        print(f"  GPU:      ⚠️  {gpu.get('message', '')}")
-
-
-def main():
+def main() -> None:
     parser = argparse.ArgumentParser(
-        description="KernelGen 验证服务测试",
+        description="Run Triton kernels directly on the selected backend.",
         formatter_class=argparse.RawDescriptionHelpFormatter,
-        epilog="""示例:
-  python test_verify_service.py --url http://172.24.4.45:8912 --chip nvidia
-  python test_verify_service.py --url http://10.7.64.131:8912 --chip huawei --all
-  python test_verify_service.py --url http://10.7.66.1:8912 --chip moore --kernel gelu""",
+        epilog="""Examples:
+  python test_verify_service.py --chip moore --kernel gelu
+  python test_verify_service.py --chip moore --all
+  python test_verify_service.py --chip moore --kernel gelu --benchmark""",
     )
-    parser.add_argument("--url", default="http://localhost:8912", help="验证服务地址")
-    parser.add_argument("--chip", default="nvidia", choices=list(CHIP_CONFIG.keys()), help="芯片类型")
+    parser.add_argument("--chip", default="moore", choices=list(CHIP_CONFIG.keys()), help="chip type")
     parser.add_argument("--gpu-id", type=int, default=-1, help="GPU ID")
-    parser.add_argument("--all", action="store_true", help="测试所有算子")
-    parser.add_argument("--kernel", default="gelu", choices=list(KERNELS.keys()), help="指定算子")
-    parser.add_argument("--health-only", action="store_true", help="仅检查健康状态")
+    parser.add_argument("--all", action="store_true", help="run all kernels")
+    parser.add_argument("--kernel", default="gelu", choices=list(KERNELS.keys()), help="kernel to run")
+    parser.add_argument("--benchmark", action="store_true", help="also run benchmark for kernels that support it")
+    parser.add_argument("--benchmark-warmup-ms", type=int, default=25, help="benchmark warmup window in ms")
+    parser.add_argument("--benchmark-rep-ms", type=int, default=100, help="benchmark repeat window in ms")
     args = parser.parse_args()
 
-    base_url = args.url.rstrip("/")
-    chip = args.chip
-    cfg = CHIP_CONFIG[chip]
+    cfg = CHIP_CONFIG[args.chip]
+    device_type, device = _configure_device(args.chip, args.gpu_id)
+    kernels_to_run = list(KERNELS) if args.all else [args.kernel]
 
     print("=" * 72)
-    print("  KernelGen 验证服务测试")
+    print("  Direct Triton Kernel Test")
     print("=" * 72)
-    print(f"  服务:  {base_url}")
-    print(f"  芯片:  {chip} ({cfg['desc']})")
-    print(f"  设备:  {cfg['device_keyword']}" +
-          (f" (cuda → {cfg['device_keyword']})" if cfg['device_keyword'] != 'cuda' else " (CUDA兼容, 不替换)"))
-    print(f"  GPU:   {args.gpu_id}" + (" (自动)" if args.gpu_id == -1 else ""))
+    print(f"  chip:   {args.chip} ({cfg['desc']})")
+    print(f"  device: {device}")
     print()
 
-    check_health(base_url)
-
-    if args.health_only:
-        return
-
-    kernels_to_test = list(KERNELS.keys()) if args.all else [args.kernel]
-
-    all_passed = True
-    for name in kernels_to_test:
-        info = KERNELS[name]
-
-        print(f"\n{'─' * 72}")
-        print(f"  [{name}] {info['desc']}")
-        print(f"{'─' * 72}")
-
-        if info.get("benchmark_func_code"):
-            # GELU: 正确性 + 参数化性能
-            acc_result = run_accuracy_verify(base_url, info, chip, args.gpu_id)
-            if not acc_result.get("passed"):
-                all_passed = False
-            run_performance_verify(base_url, info, chip, args.gpu_id)
-        else:
-            # matmul / add: 正确性
-            result = run_accuracy_only_test(base_url, info, chip, args.gpu_id)
-            if not result.get("passed"):
-                all_passed = False
-
-    print(f"\n{'=' * 72}")
-    print(f"  测试完成  |  芯片: {chip}  |  服务: {base_url}")
-    print(f"{'=' * 72}")
+    total_passed = 0
+    try:
+        for name in kernels_to_run:
+            spec = KERNELS[name]
+            print(f"{'-' * 72}")
+            print(f"  [{spec.name}] {spec.desc}")
+            print(f"{'-' * 72}")
+            passed = spec.runner(device_type, device)
+            total_passed += passed
+            if args.benchmark and spec.benchmark is not None:
+                print("    benchmark:")
+                spec.benchmark(device_type, device, args.benchmark_warmup_ms, args.benchmark_rep_ms)
+            elif args.benchmark:
+                print("    benchmark: skipped (not available)")
+            print()
+    except Exception as exc:
+        print(f"FAILED: {exc}", file=sys.stderr)
+        sys.exit(1)
 
-    sys.exit(0 if all_passed else 1)
+    print("=" * 72)
+    print(f"  PASS: {total_passed} case(s)")
+    print("=" * 72)
 
 
 if __name__ == "__main__":
-    main()
\ No newline at end of file
+    main()

From d3e0aca5a7f8ae77b80797085c05c47210c80336 Mon Sep 17 00:00:00 2001
From: flagtree-bot <flagtree_ai@163.com>
Date: Tue, 9 Jun 2026 05:24:09 +0000
Subject: [PATCH 4/6] Apply code-format changes

---
 .../python/test/test_verify_service.py        | 21 +++++++++----------
 1 file changed, 10 insertions(+), 11 deletions(-)

diff --git a/third_party/mthreads/python/test/test_verify_service.py b/third_party/mthreads/python/test/test_verify_service.py
index f6866efa60..02675d0f61 100644
--- a/third_party/mthreads/python/test/test_verify_service.py
+++ b/third_party/mthreads/python/test/test_verify_service.py
@@ -22,7 +22,6 @@
 
 __test__ = False
 
-
 CHIP_CONFIG = {
     "nvidia": {"device_keyword": "cuda", "desc": "NVIDIA CUDA"},
     "haiguang": {"device_keyword": "cuda", "desc": "Haiguang DCU (CUDA-compatible)"},
@@ -54,7 +53,7 @@ def gelu_kernel(x_ptr, y_ptr, n_elements, BLOCK_SIZE: tl.constexpr):
 def gelu(x: torch.Tensor) -> torch.Tensor:
     n_elements = x.numel()
     y = torch.empty_like(x)
-    grid = lambda meta: (triton.cdiv(n_elements, meta["BLOCK_SIZE"]),)
+    grid = lambda meta: (triton.cdiv(n_elements, meta["BLOCK_SIZE"]), )
     gelu_kernel[grid](x, y, n_elements, BLOCK_SIZE=1024)
     return y
 
@@ -137,7 +136,7 @@ def add_kernel(x_ptr, y_ptr, out_ptr, n_elements, BLOCK_SIZE: tl.constexpr):
 def add(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
     n_elements = x.numel()
     out = torch.empty_like(x)
-    grid = lambda meta: (triton.cdiv(n_elements, meta["BLOCK_SIZE"]),)
+    grid = lambda meta: (triton.cdiv(n_elements, meta["BLOCK_SIZE"]), )
     add_kernel[grid](x, y, out, n_elements, BLOCK_SIZE=1024)
     return out
 
@@ -223,11 +222,11 @@ def _run_cases(
 
 def run_gelu(device_type: str, device: str) -> int:
     shapes = [
-        (1024,),
-        (4096,),
-        (16384,),
-        (65536,),
-        (262144,),
+        (1024, ),
+        (4096, ),
+        (16384, ),
+        (65536, ),
+        (262144, ),
         (1024, 1024),
         (4096, 4096),
     ]
@@ -244,15 +243,15 @@ def case(shape):
 
 def benchmark_gelu(device_type: str, device: str, warmup_ms: int, rep_ms: int) -> None:
     cases = [
-        (torch.float16, (1024,)),
+        (torch.float16, (1024, )),
         (torch.float16, (256, 256)),
         (torch.float16, (4, 512, 512)),
         (torch.float16, (8, 16, 64, 64)),
-        (torch.float32, (1024,)),
+        (torch.float32, (1024, )),
         (torch.float32, (256, 256)),
         (torch.float32, (4, 512, 512)),
         (torch.float32, (8, 16, 64, 64)),
-        (torch.bfloat16, (1024,)),
+        (torch.bfloat16, (1024, )),
         (torch.bfloat16, (256, 256)),
         (torch.bfloat16, (4, 512, 512)),
         (torch.bfloat16, (8, 16, 64, 64)),

From f69a9dc2172a4dc4b4c5acabbd0c1e73c6420312 Mon Sep 17 00:00:00 2001
From: sunnycase <sunnycase@live.cn>
Date: Tue, 9 Jun 2026 05:54:35 +0000
Subject: [PATCH 5/6] ci: run all mthreads direct kernel tests

---
 .github/workflows/mthreads-build-and-test.yml | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/.github/workflows/mthreads-build-and-test.yml b/.github/workflows/mthreads-build-and-test.yml
index c125431aa4..920edec59a 100644
--- a/.github/workflows/mthreads-build-and-test.yml
+++ b/.github/workflows/mthreads-build-and-test.yml
@@ -50,12 +50,9 @@ jobs:
       - name: Direct Kernel Test on Mthreads
         if: steps.check_backend.outputs.should_skip != 'true'
         shell: bash
-        env:
-          MTHREADS_KERNEL_TEST_KERNEL: ${{ vars.MTHREADS_KERNEL_TEST_KERNEL }}
         run: |
           set -x
           export TRITON_DEFAULT_BACKEND=mthreads
-          kernel="${MTHREADS_KERNEL_TEST_KERNEL:-gelu}"
           python3 third_party/mthreads/python/test/test_verify_service.py \
             --chip moore \
-            --kernel "${kernel}"
+            --all

From 543d7f54a90b80e1a6e3c6daeb5bfc1aa8eb0a13 Mon Sep 17 00:00:00 2001
From: sunnycase <sunnycase@live.cn>
Date: Wed, 10 Jun 2026 05:53:44 +0000
Subject: [PATCH 6/6] test: reproduce mthreads tl.dot loop matmul

---
 .../python/test/test_verify_service.py        | 20 +++++++++++++------
 1 file changed, 14 insertions(+), 6 deletions(-)

diff --git a/third_party/mthreads/python/test/test_verify_service.py b/third_party/mthreads/python/test/test_verify_service.py
index 02675d0f61..52694aa612 100644
--- a/third_party/mthreads/python/test/test_verify_service.py
+++ b/third_party/mthreads/python/test/test_verify_service.py
@@ -90,9 +90,8 @@ def matmul_kernel(
         acc += tl.dot(a, b)
         A += BLOCK_K * stride_ak
         B += BLOCK_K * stride_bk
-    c = acc.to(tl.float16)
     C = C_ptr + rm[:, None] * stride_cm + rn[None, :] * stride_cn
-    tl.store(C, c, mask=(rm[:, None] < M) & (rn[None, :] < N))
+    tl.store(C, acc, mask=(rm[:, None] < M) & (rn[None, :] < N))
 
 
 def matmul(A: torch.Tensor, B: torch.Tensor) -> torch.Tensor:
@@ -281,12 +280,21 @@ def run_matmul(device_type: str, device: str) -> int:
 
     def case(config):
         M, N, K = config
-        A = torch.randn(M, K, device=device, dtype=torch.float16)
-        B = torch.randn(K, N, device=device, dtype=torch.float16)
+        torch.manual_seed(42)
+        A = torch.randn(M, K, device=device, dtype=torch.float32)
+        B = torch.randn(K, N, device=device, dtype=torch.float32)
         actual = matmul(A, B)
         expected = A @ B
         _synchronize(device_type)
-        _assert_close(f"matmul shape=({M},{N},{K})", actual, expected, rtol=1e-2, atol=1e-2)
+        diff = (actual.float() - expected.float()).abs()
+        max_diff = diff.max().item()
+        ok = torch.allclose(actual, expected, rtol=1e-2, atol=1e-2)
+        print(f"    ({M},{N},{K}): max_diff={max_diff:.6e} {'PASS' if ok else 'FAIL'}")
+        if not ok:
+            tile = diff[:64, :64]
+            for row in range(0, 64, 16):
+                print(f"      Row {row:2d} max diff: {tile[row].max().item():.4e}")
+            raise AssertionError(f"matmul shape=({M},{N},{K}): max_diff={max_diff:.6e}, rtol=1e-2, atol=1e-2")
 
     return _run_cases("matmul", configs, case)
 
@@ -314,7 +322,7 @@ def case(size):
     ),
     "matmul": KernelSpec(
         name="matmul",
-        desc="FP16 matrix multiplication correctness",
+        desc="FP32 tl.dot loop accumulation correctness",
         runner=run_matmul,
     ),
     "add": KernelSpec(