Adding Performance Test Scripts

skills/ascendc/performance-analyzer/script/lingxi_perf_driver.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""Driver: 对每个 case × {reference, ascendc} 都用独立子进程跑 profiler，避免互相干扰。
+
+公平性保证：通过 torch_npu.profiler 解析 kernel_details.csv 拿到设备侧 kernel 总时延，
+覆盖 PyTorch 内置 aten op (aclnnXxx) 和 AscendC 自定义 kernel，避免 host wall-time
+中的 launch / dispatch 开销和不对等的 fallback 问题。
+
+用法:
+    python .claude/skills/performance-analyzer/scripts/lingxi_perf_driver.py \\
+        --output_dir /path/to/task_dir \\
+        --warmup 5 --active 20 --retry 2
+"""
+import argparse, json, os, re, statistics, subprocess, sys
+from pathlib import Path
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+SINGLE_WORKER = SCRIPT_DIR / "lingxi_perf_single.py"
+
+
+def pick_idle_npu(default=0):
+    """解析 `npu-smi info` 选空闲的 NPU（按 AICore% + HBM/显存占用排序）。失败回退 default。"""
+    try:
+        p = subprocess.run(["npu-smi", "info"], capture_output=True, text=True, timeout=10)
+        if p.returncode != 0:
+            return default
+    except Exception:
+        return default
+    # 兼容两种常见 npu-smi 输出（每卡两行）：
+    #   row1: | <dev_id>  <Name>  | OK | Power  Temp  Hugepages |
+    #   row2: | <chip>            | <Bus-Id> | AICore%  MemUsed/MemTot [HBMUsed/HBMTot] |
+    devices = {}
+    cur = None
+    head_re = re.compile(r"^\|\s+(\d+)\s+\S+\s+\|\s+\w+\s+\|")
+    # 通过 Bus-Id 判定第二行；AICore% 后紧跟 a/b [c/d] 数值
+    bus_re = re.compile(r"^\|\s+\d+\s+\|\s+[0-9A-Fa-f:.]+\s+\|\s+(\d+)\s+(\d+)\s*/\s*(\d+)(?:\s+(\d+)\s*/\s*(\d+))?")
+    for line in p.stdout.splitlines():
+        m2 = bus_re.match(line)
+        if m2 and cur is not None:
+            aicore = int(m2.group(1))
+            mem_used = int(m2.group(2)); mem_total = max(int(m2.group(3)), 1)
+            hbm_used = int(m2.group(4)) if m2.group(4) else 0
+            hbm_total = max(int(m2.group(5)), 1) if m2.group(5) else 1
+            mem_ratio = max(mem_used / mem_total, hbm_used / hbm_total)
+            devices[cur] = (aicore, mem_ratio)
+            cur = None
+            continue
+        m1 = head_re.match(line)
+        if m1:
+            cur = int(m1.group(1))
+    if not devices:
+        return default
+    best_id, _ = min(devices.items(), key=lambda kv: (kv[1][0], kv[1][1]))
+    return best_id
+
+
+def run_single(out_dir, idx, impl, warmup, active, device_id):
+    cmd = [
+        sys.executable, str(SINGLE_WORKER),
+        "--output_dir", str(out_dir),
+        "--case_idx", str(idx),
+        "--impl", impl,
+        "--warmup", str(warmup),
+        "--active", str(active),
+    ]
+    env = os.environ.copy()
+    # 子进程内 `torch.device("npu")` 会映射到这里指定的可见设备（逻辑 index 0）
+    env["ASCEND_RT_VISIBLE_DEVICES"] = str(device_id)
+    proc = subprocess.run(cmd, capture_output=True, text=True, timeout=600, env=env)
+    for line in proc.stdout.splitlines():
+        if line.startswith("__RESULT_JSON__"):
+            return json.loads(line[len("__RESULT_JSON__"):])
+    return {"case_idx": idx, "impl": impl, "avg_kernel_ms": None,
+            "error": f"no_result; stderr_tail={proc.stderr[-300:]}"}
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--output_dir", required=True)
+    ap.add_argument("--warmup", type=int, default=5)
+    ap.add_argument("--active", type=int, default=20)
+    ap.add_argument("--retry", type=int, default=2,
+                    help="重试次数（解析失败时）")
+    ap.add_argument("--output", help="输出 JSON 报告路径（供 agent 下游消费）")
+    ap.add_argument("--device", type=int, default=None,
+                    help="NPU 设备 id；未指定时自动选择空闲卡（按 AICore%% 最低）")
+    args = ap.parse_args()
+
+    out_dir = Path(args.output_dir).resolve()
+
+    # 设备选择：显式 --device > 环境变量 ASCEND_RT_VISIBLE_DEVICES > 自动挑空闲卡
+    if args.device is not None:
+        device_id = args.device
+        device_src = "cli"
+    elif os.environ.get("ASCEND_RT_VISIBLE_DEVICES"):
+        device_id = int(os.environ["ASCEND_RT_VISIBLE_DEVICES"].split(",")[0])
+        device_src = "env"
+    else:
+        device_id = pick_idle_npu(default=0)
+        device_src = "auto"
+    print(f"[INFO] Using NPU device {device_id} (source={device_src})")
+
+    json_files = sorted(out_dir.glob("*.json"))
+    json_path = next((f for f in json_files if not f.name.endswith(".bak")), None)
+    cases = []
+    if json_path:
+        with open(json_path) as f:
+            cases = [json.loads(line) for line in f if line.strip()]
+    n_cases = len(cases)
+
+    print("=" * 100)
+    print(f"Kernel-level Performance: {out_dir.name}  (warmup={args.warmup}, active={args.active})")
+    print("=" * 100)
+    print(f"{'Case':<5} {'Shape':<35} {'dtype':<10} {'Ref(ms)':>12} {'Asc(ms)':>12} {'Speedup':>10}")
+    print("-" * 100)
+
+    rows = []
+    speedups = []
+    for idx in range(n_cases):
+        ref_res, asc_res = None, None
+        for attempt in range(1 + args.retry):
+            ref_res = run_single(out_dir, idx, "reference", args.warmup, args.active, device_id)
+            if ref_res.get("avg_kernel_ms") is not None:
+                break
+        for attempt in range(1 + args.retry):
+            asc_res = run_single(out_dir, idx, "ascendc", args.warmup, args.active, device_id)
+            if asc_res.get("avg_kernel_ms") is not None:
+                break
+
+        shape = str(cases[idx]["inputs"][0]["shape"])
+        dtype = cases[idx]["inputs"][0]["dtype"]
+        ref_ms = ref_res.get("avg_kernel_ms")
+        asc_ms = asc_res.get("avg_kernel_ms")
+
+        if ref_ms is not None and asc_ms is not None and asc_ms > 0:
+            sp = ref_ms / asc_ms
+            speedups.append(sp)
+            print(f"{idx:<5} {shape:<35} {dtype:<10} {ref_ms:>12.6f} {asc_ms:>12.6f} {sp:>9.2f}x")
+        else:
+            print(f"{idx:<5} {shape:<35} {dtype:<10} "
+                  f"{'N/A' if ref_ms is None else f'{ref_ms:.6f}':>12} "
+                  f"{'N/A' if asc_ms is None else f'{asc_ms:.6f}':>12} "
+                  f"{'N/A':>10}  (ref_err={ref_res.get('error')}, asc_err={asc_res.get('error')})")
+
+        rows.append({"case": idx, "shape": shape, "dtype": dtype,
+                     "ref_kernel_ms": ref_ms, "asc_kernel_ms": asc_ms,
+                     "speedup": (ref_ms / asc_ms) if (ref_ms and asc_ms and asc_ms > 0) else None,
+                     "ref_breakdown": ref_res.get("breakdown") if isinstance(ref_res.get("breakdown"), dict) else None,
+                     "asc_breakdown": asc_res.get("breakdown") if isinstance(asc_res.get("breakdown"), dict) else None,
+                     "ref_error": ref_res.get("error"),
+                     "asc_error": asc_res.get("error")})
+
+    summary = {
+        "task": out_dir.name,
+        "task_dir": str(out_dir),
+        "n_cases_total": n_cases,
+        "n_cases_valid": len(speedups),
+        "geomean_speedup": statistics.geometric_mean(speedups) if speedups else None,
+        "mean_speedup": statistics.mean(speedups) if speedups else None,
+        "median_speedup": statistics.median(speedups) if speedups else None,
+        "min_speedup": min(speedups) if speedups else None,
+        "max_speedup": max(speedups) if speedups else None,
+        "warmup": args.warmup,
+        "active": args.active,
+        "device_id": device_id,
+        "device_select_source": device_src,
+        "timing_method": "torch_npu.profiler.kernel_details",
+        "per_case": rows,
+    }
+
+    print("-" * 100)
+    if speedups:
+        print(f"Mean speedup    : {summary['mean_speedup']:.2f}x")
+        print(f"Geomean speedup : {summary['geomean_speedup']:.2f}x  ← 主指标")
+        print(f"Median speedup  : {summary['median_speedup']:.2f}x")
+        print(f"Valid cases     : {len(speedups)}/{n_cases}")
+    print("=" * 100)
+    print("\nKernel breakdown:")
+    for r in rows:
+        if r["ref_breakdown"] or r["asc_breakdown"]:
+            print(f"\n[case {r['case']}] {r['shape']} {r['dtype']}")
+            if r["ref_breakdown"]:
+                print("  REF kernels:")
+                for k, v in r["ref_breakdown"].items():
+                    print(f"    {k:<45} calls={v['calls']:<3} total={v['total_us']:>8.2f}us  avg={v['avg_us']:>7.3f}us")
+            if r["asc_breakdown"]:
+                print("  ASC kernels:")
+                for k, v in r["asc_breakdown"].items():
+                    print(f"    {k:<45} calls={v['calls']:<3} total={v['total_us']:>8.2f}us  avg={v['avg_us']:>7.3f}us")
+
+    if args.output:
+        out_path = Path(args.output)
+        out_path.parent.mkdir(parents=True, exist_ok=True)
+        with open(out_path, "w", encoding="utf-8") as f:
+            json.dump(summary, f, indent=2, ensure_ascii=False)
+        print(f"\n[INFO] JSON report saved to: {out_path}")
+
+
+if __name__ == "__main__":
+    main()

skills/ascendc/performance-analyzer/script/lingxi_perf_single.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+"""单 case kernel-level 性能测试（被 perf_kernel_driver.py 通过子进程调用）。
+输出 JSON 到 stdout 最后一行。"""
+import argparse, importlib.util, inspect, json, os, shutil, sys, time
+from pathlib import Path
+import torch
+import torch.nn as nn
+
+
+def _load(path, name):
+    spec = importlib.util.spec_from_file_location(name, path)
+    m = importlib.util.module_from_spec(spec)
+    sys.modules[name] = m
+    spec.loader.exec_module(m)
+    return m
+
+
+def _find_cls(module, preferred):
+    c = getattr(module, preferred, None)
+    if inspect.isclass(c) and issubclass(c, nn.Module):
+        return c
+    for _, v in vars(module).items():
+        if inspect.isclass(v) and issubclass(v, nn.Module) and v is not nn.Module:
+            return v
+    raise AttributeError("no nn.Module")
+
+
+def _move(v, d):
+    if isinstance(v, torch.Tensor):
+        return v.to(d)
+    if isinstance(v, (list, tuple)):
+        return type(v)(_move(x, d) for x in v)
+    return v
+
+
+def _clone(v):
+    if isinstance(v, torch.Tensor):
+        return v.clone()
+    if isinstance(v, (list, tuple)):
+        return type(v)(_clone(x) for x in v)
+    return v
+
+
+def _find_file(root, name):
+    for r, _, files in os.walk(root):
+        if name in files:
+            return os.path.join(r, name)
+    return None
+
+
+def _parse_kernels(profile_path, active):
+    csv_path = _find_file(profile_path, "kernel_details.csv")
+    if not csv_path:
+        return None, "no_csv"
+    try:
+        import pandas as pd
+        df = pd.read_csv(csv_path)
+    except Exception as e:
+        return None, f"read_err:{e}"
+    if df.empty or "Duration(us)" not in df.columns:
+        return None, "empty_or_bad"
+    total_us = df["Duration(us)"].astype(float).sum()
+    avg_ms = (total_us / active) / 1000.0
+    breakdown = {}
+    for n, g in df.groupby("Name"):
+        breakdown[str(n)] = {
+            "calls": int(len(g)),
+            "total_us": float(g["Duration(us)"].astype(float).sum()),
+            "avg_us": float(g["Duration(us)"].astype(float).mean()),
+        }
+    return avg_ms, breakdown
+
+
+def _profile(model, inputs, warmup, active, tag):
+    import torch_npu
+
+    with torch.no_grad():
+        _ = model(*inputs)
+    torch.npu.synchronize()
+
+    profile_path = f"/tmp/perfk_{tag}_{int(time.time()*1000)}_{os.getpid()}"
+    if os.path.exists(profile_path):
+        shutil.rmtree(profile_path, ignore_errors=True)
+
+    exp = torch_npu.profiler._ExperimentalConfig(
+        aic_metrics=None,
+        profiler_level=torch_npu.profiler.ProfilerLevel.Level1,
+        l2_cache=False,
+        data_simplification=False,
+    )
+    skip_first = 1
+    total = skip_first + warmup + active
+
+    with torch_npu.profiler.profile(
+        activities=[torch_npu.profiler.ProfilerActivity.NPU,
+                    torch_npu.profiler.ProfilerActivity.CPU],
+        schedule=torch_npu.profiler.schedule(
+            wait=0, warmup=warmup, active=active, repeat=1, skip_first=skip_first
+        ),
+        on_trace_ready=torch_npu.profiler.tensorboard_trace_handler(profile_path),
+        record_shapes=False,
+        experimental_config=exp,
+    ) as prof:
+        for _ in range(total):
+            with torch.no_grad():
+                _ = model(*inputs)
+            prof.step()
+            torch.npu.synchronize()
+
+    avg_ms, breakdown = None, "init"
+    for _ in range(15):
+        time.sleep(1.0)
+        avg_ms, breakdown = _parse_kernels(profile_path, active)
+        if avg_ms is not None:
+            break
+    shutil.rmtree(profile_path, ignore_errors=True)
+    return avg_ms, breakdown
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--output_dir", required=True)
+    ap.add_argument("--case_idx", type=int, required=True)
+    ap.add_argument("--impl", choices=["reference", "ascendc"], required=True)
+    ap.add_argument("--warmup", type=int, default=5)
+    ap.add_argument("--active", type=int, default=20)
+    ap.add_argument("--device", type=int, default=None,
+                    help="NPU 设备 id；默认跟随 ASCEND_RT_VISIBLE_DEVICES，否则使用 npu:0")
+    args = ap.parse_args()
+
+    # 若显式传入 --device，覆写环境变量；driver 已通过 env 传入时会命中这一行之前
+    if args.device is not None:
+        os.environ["ASCEND_RT_VISIBLE_DEVICES"] = str(args.device)
+
+    out_dir = Path(args.output_dir).resolve()
+    sys.path.insert(0, str(out_dir / "kernel" / "build"))
+    sys.path.insert(0, str(out_dir))
+
+    if args.impl == "reference":
+        mod = _load(out_dir / "model.py", "ref_mod")
+        cls = _find_cls(mod, "Model")
+    else:
+        mod = _load(out_dir / "model_new_ascendc.py", "asc_mod")
+        cls = _find_cls(mod, "ModelNew")
+
+    init_args = getattr(_load(out_dir / "model.py", "ref_for_init"),
+                        "get_init_inputs", lambda: [])()
+    input_groups = _load(out_dir / "model.py", "ref_for_inputs").get_input_groups()
+
+    device = torch.device("npu")
+    torch.manual_seed(0)
+    if hasattr(torch, "npu"):
+        torch.npu.manual_seed(0)
+
+    model = cls(*_clone(init_args)).to(device).eval()
+    inputs = _move(_clone(input_groups[args.case_idx]), device)
+
+    avg_ms, breakdown = _profile(model, inputs, args.warmup, args.active,
+                                 f"{args.impl}_c{args.case_idx}")
+
+    result = {
+        "case_idx": args.case_idx,
+        "impl": args.impl,
+        "avg_kernel_ms": avg_ms,
+        "breakdown": breakdown if isinstance(breakdown, dict) else None,
+        "error": None if avg_ms is not None else str(breakdown),
+    }
+    print("__RESULT_JSON__" + json.dumps(result))
+
+
+if __name__ == "__main__":
+    main()