Aharrypotter/zl_bench
1
0 Fork 0
forked from zhoulin/zl_bench
Code Pull Requests Activity

Compare commits

base: Aharrypotter/zl_bench:d7cb7efc94b0fcaca697097436fdd0cfa574bc9e
Aharrypotter/zl_bench:main Aharrypotter/zl_bench:feature/musa-benchmark-support Aharrypotter/zl_bench:feature/musa-rlc-product-harness zhoulin/zl_bench:main
..
compare: Aharrypotter/zl_bench:main
Aharrypotter/zl_bench:main Aharrypotter/zl_bench:feature/musa-benchmark-support Aharrypotter/zl_bench:feature/musa-rlc-product-harness zhoulin/zl_bench:main
18 Commits
d7cb7efc94 .. main
Author SHA1 Message Date
Aharrypotter 2786eb9886 bench: orchestrate a multi-card campaign without touching the measurement path 
campaign.sh fans a screened operator list across cards, one ab_repeat.sh per
card, serial within a card. It does no measuring: ab_repeat.sh is the path that
produced the four operator verdicts already on record, and it stays byte for
byte as it was.

Each card gets its own AB_ROOT. Eight ab_repeat.sh started in the same second
would otherwise compute the same timestamped directory name and two cards would
write into one.

ZL_ISOLATE_TUNER_CACHE=1 is forced, since eight processes on one sqlite is the
thing to avoid. The cost is that pinned configs differ from a shared-cache
batch, so absolute latencies cannot be quoted across the two; the caliber guard
refuses such comparisons on its own, which is why this is a note and not a
warning someone has to remember.

partition.py's shell output gained the per-card weight, because EXTRA_OPS has
to land on the lightest card and counting operators gets that wrong — one heavy
operator outweighs three light ones. It writes through stdout.buffer: Python
turns "\n" into CRLF on Windows, and the trailing \r then rides into the last
field and breaks the numeric comparison that picks the card.
 2026-09-10 19:43:37 +08:00
Aharrypotter 03fdcbab28 bench: partition operators across cards by measured cost, and rescore in place 
Two pieces the parallel campaign needs.

partition.py assigns screened operators to cards longest-first. run_batch.py's
existing hash partition ignores weight: on this batch it puts the critical path
at 120 minutes and leaves a card idle, where greedy gets 70. The floor is set
by the single heaviest operator — topk alone is 70.6 minutes against an ideal
share of 45.4 — so partitioning cannot beat it, only keep the other cards from
waiting. Weights come from screen.tsv's elapsed_before + elapsed_after, which
is one pair, i.e. 2 cells; N rounds is 2N cells, so the estimate is N times the
weight. Getting that wrong reported topk as 11.8 minutes.

RESCORE=<root> re-judges an existing screening root against the current
fingerprint and rewrites its screen.tsv, keeping the original alongside. The
criterion has already changed twice — index.tsv used to be hashed, and the
switch's own ttg.rlc-* annotations had to be stripped — and the second change
moved 24 operators from 23 IR_CHANGED to 16. Rescoring the artifacts already on
disk takes seconds; re-running that screen takes 79 minutes.
 2026-09-10 19:25:34 +08:00
Aharrypotter c6a7f834cb bench: ignore the switch's own annotations when fingerprinting the IR 
FlagTree's _apply_musa_rlc_policy(mod) stamps the whole RLC policy onto the
module as ttg.rlc-* attributes whenever the switch is on, regardless of whether
the pass then changes anything. Hashing the raw text therefore carried a
constant offset: every operator that compiles a kernel differed on that one
line and screened as IR_CHANGED. A filter that excludes nothing is not
filtering.

The null control could not catch it. Both of its sides run with the switch off,
so both carry the same attributes and the fingerprints match — it proved the
dump is deterministic, which was true, and said nothing about this.

Stripping ttg.rlc-* before hashing turns the screen back into a filter. Over
the same 24 operators, recomputed from the artifacts already on disk: 23
IR_CHANGED becomes 16, with 7 falling back to NO_IR_DIFF — tanh, gelu, mul,
softmax, mm, bmm, replication_pad3d, each byte-identical apart from that line.
replication_pad3d is the PR's largest NVIDIA win at +124.6%; on MUSA this pass
does not touch it.

The counts also show why the convert_layout count fails as a criterion in both
directions: addmm keeps cvt at 15 -> 15 while all 15 of its ttgir files differ
in substance, and tanh has no conversions and no change.
 2026-09-10 19:13:51 +08:00
Aharrypotter 4ac2a9acd8 bench: report when a foreign vendor's FlagGems implementation gets loaded 
FlagGems keeps per-vendor implementations under
flag_gems/runtime/backend/_<vendor>/ops/, and benchmarks commonly gate on
whether one imports:

    try:
        from flag_gems.runtime.backend._nvidia.hopper.ops.mm import router_gemm
        ROUTER_GEMM_AVAILABLE = True
    except Exception:
        ROUTER_GEMM_AVAILABLE = False

An import only proves the .py is on disk. It touches no device, so it succeeds
on every backend and the gate never fires. test_router_gemm.py therefore ran
the NVIDIA Hopper kernel on an S5000: its autotune and heuristics are keyed to
nvidia, supply no BLOCK_M/N/K/SPLIT_K on musa, and the run died deep inside the
JIT with "dynamic_func() missing 4 required positional arguments" — an error
that points nowhere near the cause. Meanwhile _mthreads/ops/mm.py holds a
working router_gemm the benchmark never imports. Gating on vendor_name first,
the way the Cambricon branch in the same file already does, makes it run: 10
shapes, 10 kernels, cvt 10 -> 2.

The plugin now scans sys.modules at session end and prints
FOREIGN_VENDOR_MODULE with the vendors it found. It reports rather than
refuses: loading another vendor's module is not proof of a problem, but when a
result looks wrong it is the first line to check. Vendor roots are compared
segment-wise, not by prefix, so a vendor whose name extends ours is not
mistaken for our own.

README gains the ordered checklist for an operator that will not run, with the
real instance behind each entry.
 2026-09-10 18:51:26 +08:00
Aharrypotter 2e456bdc74 bench: let the verdict govern the report, and put the caliber in the table 
A pair judged incomparable was still followed by a speedup range on the next
line. A NO_TRITON pair's range says nothing about the switch — nothing reached
the compiler — and a NO_IR_DIFF pair's range is the machine's noise floor for
that operator, which is useful but is not a result. Printing either unlabelled
produces exactly the thing this tooling exists to prevent: a table that looks
like a finding. NO_ROWS and NO_TRITON now print no range; NO_IR_DIFF prints one
and says what it is — how far it sits from zero is the confidence floor for a
single measurement of that operator.

screen.tsv gains the seven caliber columns, read back through
`batch_summary.py --caliber`, so CALIBER_FIELDS stays the single definition and
the screening table does not restate it. Two screen.tsv files taken at
different --level are otherwise indistinguishable side by side, and those
columns are the only thing that would stop the comparison.
 2026-09-10 17:10:59 +08:00
Aharrypotter a11863399a bench: bound each screening pair and record what it cost 
Screening is first contact with an operator, which makes it the place most
likely to meet one that hangs — and it had no upper bound at all, so a single
operator could hold the whole pass indefinitely. The entire argument for
having a screening step is that it is fast.

CELL_TIMEOUT (default 2400s, as in ab_repeat.sh) now caps each side, and a
truncated pair says so instead of being folded into the dtype explanation: rc
124/137 is a cost problem, not a verdict about the switch.

elapsed_before/elapsed_after join screen.tsv. The two-step design rests on the
ratio between a pair and six rounds — 0.8 minutes against 13-44 on S5000 — so
the pass should record its own half of that ratio rather than leave it to be
reconstructed from directory timestamps.
 2026-09-10 17:08:08 +08:00
Aharrypotter 1fd3abd56d bench: rename NO_KERNEL to NO_TRITON and say what it rules out 
"NO_KERNEL" reads as "this operator has no kernel", which is not what the
verdict means and not usually true. It means the run compiled no Triton
kernel: the operator may have run perfectly well through aten or a vendor
library, and a GPU kernel certainly executed. What is absent is anything for
the Triton compiler pass under test to act on.

The note now says that, and says the one alternative explanation is already
ruled out: run_pytest.sh gives every run its own TRITON_CACHE_DIR, so zero
.ttgir cannot be compilation skipped by a warm cache.
 2026-09-10 17:00:45 +08:00
Aharrypotter 4d69320139 bench: let ab_rlc_test.sh decide which operators are worth six rounds 
zl_bench already had the screening mechanics — ab_rlc_test.sh runs exactly one
config-pinned pair per operator (before records, after replays it) against the
vendor-prefixed RLC switch. I had started a second script for the same job
instead of extending this one; that is removed.

What the existing script could not tell you is whether the pair is worth
following with six rounds, which costs 13-44 minutes per operator against
about 0.8 for the pair. It now writes a verdict per operator to
<root>/screen.tsv, and three of the four are not "the switch does nothing":

  IR_CHANGED  the two sides' ttgir differ         -> measure it
  NO_IR_DIFF  byte-identical, configs pinned      -> skip it
  NO_KERNEL   rows but no .ttgir                  -> never reached Triton here
  NO_ROWS     no measurements                     -> usually the dtype

The criterion is the ttgir fingerprint, not the convert_layout count: equal
counts do not mean equal generated code, and what the A/B measures is whether
the generated code changed. Counts are still recorded — they say what kind of
change it was. The fingerprint comes from _ir_fingerprint.sh, shared with
ab_repeat.sh's round-1 check so the two cannot drift apart.

OPS entries take an optional per-operator dtype and default to the operator's
own set. Forcing one dtype across a list is how fft got written off as
unrunnable on MUSA: it only accepts complex64.

Output moves under runs/ab_rlc_<ts>/<op>_{before,after} with OUT_DIR set
explicitly, rather than recovering the directory with `ls -dt runs/<op>_*`.
That lookup is correct while one process runs operators in series, which is
the only way it has been used; it stops being correct the moment several cards
share runs/.
 2026-09-10 16:54:26 +08:00
Aharrypotter 2fbfe1dd16 bench: make screening a first-class step with the same caliber as the A/B 
Six rounds of A/B cost 13-44 minutes per operator on S5000 (rand, 15 shapes:
13.0 min; sort, 25 shapes: 43.4 min). Screening one costs about 0.8 minutes —
21 operators in 15.0 minutes, which knocked 13 of them out of the expensive
queue and cut it from ~7 hours to ~2.7. That ratio is the whole argument for
having a separate first step.

Screening was being done by ad-hoc scripts living in the results tree, and
those got two things wrong that this codebase already knows better than:

Both sides ran `record`. Each swept autotune independently, so the two sides
compiled different configs and produced different IR for reasons that have
nothing to do with the switch — every operator would screen as "changed".
screen.sh pins configs exactly as round 1 of ab_repeat does: A records, B
replays A.

The criterion was the convert_layout count. Equal counts do not mean equal
generated code, and what the A/B measures is whether the generated code
changed. The ttgir fingerprint decides; the cvt counts are still recorded,
because they explain what kind of change it was.

Four verdicts, and three of them are not "the switch does nothing":
IR_CHANGED (measure it), NO_IR_DIFF (skip it), NO_KERNEL (the operator never
reached Triton on this backend), NO_ROWS (it did not run — usually the dtype).
The last two are runnability problems being mistaken for switch results, which
is how `fft` got written off: the screen forced DTYPES=float16 on every
operator and fft only accepts complex64. OPS entries now carry an optional
per-operator dtype and default to the operator's own set.

ir_fingerprint moves to _ir_fingerprint.sh so screening and the A/B cannot
drift apart on the criterion.
 2026-09-10 16:49:32 +08:00
Aharrypotter 5236d0745f bench: fingerprint only the IR, and refuse to compare when there is none 
ir_fingerprint hashed every file under the dump directory, which includes
index.tsv and naming.md. An operator that compiles no Triton kernel at all
leaves only those two, identical on both sides, so the switch-effectiveness
check reported "byte-identical ttgir" and skipped the operator with
SWITCH_NO_IR_DIFF — "the switch changed no generated code" — when the truth
was that nothing was generated. router_gemm on MUSA is exactly this case: its
kernel cannot be instantiated there (missing BLOCK_M/N/K/SPLIT_K from the
autotune path), zero .ttgir files, and the check called it a no-op switch.

Now only *.ttgir files are hashed, and a run with none returns failure so the
caller treats it as "no comparable IR" rather than "identical". The message on
that path says so, and says explicitly that it does not mean the switch is
inert.
 2026-09-10 16:47:39 +08:00
Aharrypotter 551ffbbb77 bench: exit the sampler quietly when its reader goes away 
Piping `telemetry.py --run` into anything that stops reading first — head, or
the campaign shutting the sampler down — raised BrokenPipeError and printed a
traceback. Under ab_repeat that traceback lands in telemetry.err, where it
reads as "telemetry is broken" rather than "telemetry ended normally", and
telemetry.err is exactly the file someone checks when the clock record looks
short.

Swallowing the exception alone is not enough: the interpreter flushes stdout
again on exit and emits a second "Exception ignored" to stderr, so stdout is
pointed at devnull the way CPython documents.
 2026-09-10 16:10:15 +08:00
Aharrypotter 4329b0c835 docs: stop claiming non-CUDA backends never attempt graph capture 
Both the _timing_plugin docstring and the README env table still described the
behaviour from before MUSA declared supports_cuda_graph: "other backends do
not attempt CUDA Graph at all", "非 CUDA 即使设 1 也拒绝". Capture selection has
been driven by the backend capability, not by "is this CUDA", since the MUSA
fallback policy landed. Reading the old text is most likely why the S5000
campaigns were launched with ZL_ENABLE_CUDAGRAPH=0 without anyone checking
what auto would do.

Also states what timer=cuda_graph does and does not mean, since the two are
easy to conflate: it says capture was attempted. Measured on S5000 with auto
and configs pinned, dropout/rand/masked_select select the graph path and then
fall back on every single shape — BENCHMARK_DIRECT_NO_CUDAGRAPH on all 15
rows, zero poisoned, zero rows lost — so the whole table is direct-timed under
a cuda_graph label. ZL_ENABLE_CUDAGRAPH=0 selects the same path and records it
plainly, which is why the existing campaign data stands as measured.
 2026-09-10 16:00:44 +08:00
Aharrypotter ee5f70fa89 bench: record device clocks, temperature and power for the whole campaign 
A 6-round A/B here reports side noise at the 0.01% level. Clock throttling and
thermal drift move latency by far more than that and leave no trace in
run.log, so the report comes out looking exactly as clean as a run on a stable
machine. The NV run that reproduced PR763 sampled nvidia-smi every 5s
alongside the campaign; the MUSA campaigns had no equivalent, and the parallel
plan makes it worse by heating the whole box.

telemetry.py samples through the backend, so cuda and musa emit the same
columns (index, sm_clock_mhz, mem_clock_mhz, temp_c, power_w, util_pct) and
two machines' campaigns can be read side by side. mthreads-gmi offers no CSV
query mode, so the MUSA path parses --json and strips the units glued onto
each value ("36C", "1888MHz", "477.68W"); its "Power Draw " key carries a
trailing space, matched by stripping rather than by literal.

The device comes from the backend's own visibility variable when not pinned. A
list of devices resolves to no -i at all, sampling every one, rather than
silently picking the first. Dropping the placeholder drops its flag too: a
bare trailing -i fails the sampler instead of widening it.

Telemetry is evidence, not a gate. A missing or unpermitted sampler exits
quietly with its reason in telemetry.err and the campaign runs unchanged.
 2026-09-10 16:00:01 +08:00
Aharrypotter d45806eacd bench: log every A/B cell as it finishes, and cap it with a timeout 
The three markers the README requires be checked — AUTOTUNE_REPLAY_FALLBACK,
BENCHMARK_DIRECT_NO_CUDAGRAPH, BENCHMARK_CAPTURE_POISONED — were only
reachable by grepping run.log afterwards, which happens when someone
remembers to. The NV run that reproduced PR763 wrote a per-cell status row
instead; a 6-round campaign here had no equivalent, so a missing cell, a
stalled cell, or a marker count going from 0 to nonzero was invisible until
the batch ended.

run_cell now writes round/op/side/switch/rc/elapsed_s/success/failed plus the
three marker counts to <root>/status.tsv as each cell completes, and wraps the
cell in CELL_TIMEOUT (default 2400s, matching the NV run) so one hung cell
cannot hold the batch. rc 124/137 is called out by name.

grep -c prints 0 and exits 1 when it finds nothing, so each count swallows its
status; without that the bookkeeping itself would mark a clean cell failed.
 2026-09-10 15:55:36 +08:00
Aharrypotter 92f3d150c7 bench: preflight the interpreter, not just the paths 
triton normally comes from a PYTHONPATH pointing at a compiler build and
flag_gems from a source checkout, so a shell that reaches every correct path
can still be the wrong Python. run_pytest.sh checked that the FlagGems
directory and benchmark file exist and then handed the run to pytest, where
the real problem surfaced per cell as `pytest exit 4` from a conftest
ImportError.

A multi-round campaign burns every cell that way before anyone reads a log:
today's noise-calibration run lost 12 measurement cells and 160 load runs to
`ModuleNotFoundError: No module named 'triton'`, each failing in under a
second and each looking, from the outside, like a cell that ran.

The check names the interpreter it used, since which Python is on PATH is the
whole question. Its stderr redirect stays on the child — putting `2>/dev/null`
on the preflight call swallows preflight's own `>&2` message and turns a named
refusal into a bare exit 3.
 2026-09-10 15:18:00 +08:00
AharrypotterandClaude Opus 5 c95474a432 bench: refuse to report a comparison whose two sides have different calibers 
Two runs on different calibers produce a report that looks exactly like a
valid one — no step errors, the numbers just mean something else. This repo
has published three figures that were an artifact of such a difference:
router_gemm -8.2% (record vs record), mm "945 variants -> 33" (shared vs
inherited libtuner cache), replication_pad3d +43% vs +124.5% (timer).

batch_summary --ab / --ab-rounds now reads every participating run's
run_identity.json and compares backend/device/timer/bench_level/shape_mode/
tuner_cache/direct_repeat before parsing a single latency. Mismatch, or a
missing identity, prints !!! [caliber] and exits 4. autotune_mode is judged
separately: record pinning a replay is the normal shape, both sides record
is the exact trap above and is refused by name. --allow-caliber-mismatch
releases it, still printing what it found.

Recording on a *shared* tuner cache warns rather than refuses: the pinned
config depends on what else ran on the machine, which leaves relative
conclusions intact and absolute latencies unquotable. Refusing it would
reject the default every A/B here currently uses.

ab_repeat.sh closes the one axis that had no check: after round 1 it
fingerprints both sides' ttgir (relative path + content, so a changed
variant *set* counts too). Byte-identical means the switch changed no
generated code for this op, and the remaining N-1 rounds would measure
noise and report it as "no difference" — so the op is skipped with
SWITCH_NO_IR_DIFF unless SWITCH_ALLOW_NO_IR_DIFF=1.

README leads with a caliber table: each axis, its measured magnitude, what
pins it, and which code checks it. The fourth column is code because
documentation did not hold — the README already said "the B side must use
REPLAY_FROM" when the record-vs-record number was published.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-09-10 14:32:49 +08:00
AharrypotterandClaude Opus 5 aa147c938e feat: repeat an A/B over rounds and report whether the verdict holds 
A single pair cannot say whether its own verdict is stable, and on this device
that is not a theoretical gap. Measured on an S5000, topk eliminated 32
convert_layout ops yet 17 of its 42 shapes changed the SIGN of their delta
between two equally valid runs. The cause is not the timer: graph capture and
direct timing agree to 1.00x on that operator, and the verdict moved a median of
1.1pp between them, inside the 1.7pp noise floor a same-timer control
established. The deltas are simply near zero, and one measurement cannot orient
them.

ab_repeat.sh runs an A/B over N rounds in independent processes. Round 1's A
side records the autotune configs and every later run -- including A's own later
rounds -- replays that one set, so configs are pinned across rounds and sides and
only the switch under test and measurement noise remain. Round order alternates
to cancel one-directional drift. An operator with a failed round gets no report,
since a report missing rounds reads exactly like a complete one.

batch_summary --ab-rounds aggregates <root>/r<k>_<A|B>/run.log and separates the
two dispersions, which answer different questions and were previously conflated:
the spread within one side's samples says how noisy the machine was, while the
spread of the per-round deltas says whether the verdict holds. Shapes whose
delta changes sign across rounds are counted and called out, because their
direction is not a conclusion at any timer.

The path is already validated on NVIDIA: six rounds narrowed router_gemm to
-16.6% ~ -16.2% with 0.4% side noise, which is what turned a single pair into a
statement.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-09-10 13:11:37 +08:00
AharrypotterandClaude Opus 5 9665d36a7c feat: allow the direct timer to repeat and take the median 
do_bench already returns a median over the iterations inside one call, which
suppresses iteration-level jitter. It cannot see drift across a call: clocks
ramping, another process arriving. Direct timing is where that shows, because
it walks the full Python launch path once per iteration, so on small kernels the
number carries launch overhead and CPU jitter that graph capture removes.
ZL_DIRECT_REPEAT=N repeats the call and takes the median over calls, which is
the layer the inner median misses. It defaults to 1, so nothing changes unless
asked.

The knob covers both ways a row ends up on the direct timer: a run that selected
timer=direct outright, and the rows a captured run falls back on, whose marker
now carries repeat=N since those rows were already a different caliber from the
captured ones and can now also be a different sample count. Repetition is
skipped when the caller asks for quantiles or return_mode "all", because the
result is then a sequence and a median across calls is undefined for it.

On MThreads this is not a refinement. A convert-reduction screen over 21
operators puts the largest reductions at sort (65->15), dropout (48->0) and
rand (24->0), and all three are exactly the operators that cannot be
graph-captured there -- RNG seeding and a capture-illegal stream op. The
headline numbers on that device are produced by the noisiest timer, which until
now had no repetition at all.

_cudagraph_plugin imports triton at module scope, so the tests load it against a
stub: the repeat logic is arithmetic and should stay testable without a GPU.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-09-10 12:56:08 +08:00
16 changed files with 1837 additions and 44 deletions
Expand all files Collapse all files

No files matched your search

README.md
+247 -4
View File
@@ -6,6 +6,30 @@
2. **可解释**：自动按 shape 收集每次运行实际使用的 ttgir（带可读的变体命名），供 IR 级 diff；
3. **口径统一**：cudagraph 计时消除 launch 开销，小 kernel 的对比不被 CPU 侧噪声淹没；capture 前按 warmup 预算显式预热吸收 autotune/JIT，首轮即稳态、多轮一致。
## 先读这个：口径决定数字的含义
同一个算子、同一台机器、同一份代码，换一条测试口径，测出来的数字可以完全不同。
下表每一行都是本仓踩过的，量级是实测的，不是估计：
| 口径轴 | 改一下的量级 | 由谁固定 | 由谁检查 |
|---|---|---|---|
| autotune config（record vs replay） | router_gemm 差 8pp | `REPLAY_FROM` | `caliber_complaints`：两侧都 record 直接拒绝出报告 |
| timer（cuda_graph vs direct） | replication_pad3d 从 +43% 变成 +124.5% | `ZL_TIMER` / backend 能力 | 同上，`timer` 字段不一致即拒绝 |
| libtuner 持久缓存（共享 vs 隔离） | mm 看起来从 945 个变体降到 33 个——实际是 B 侧继承了 A 侧的缓存 | `ZL_ISOLATE_TUNER_CACHE=1` | 共享缓存上 record 会打 `警告 [caliber]` |
| `--level core` vs `comprehensive` | mm 的 shape 数 5 → 29，能不能覆盖到大 GEMM 全看这个 | `BENCH_LEVEL` | `bench_level` 字段不一致即拒绝 |
| 重复轮数 | 收益落在零附近时，单次测量给不出方向：topk 42 个 shape 有 17 个在两次等效运行间翻了符号 | `ROUNDS` / `ZL_DIRECT_REPEAT` | `direct_repeat` 字段不一致即拒绝 |
| 被测开关本身是否真的接上 | 开关名拼错，两侧跑同一份代码，脚本照样报出"无差异" | 开关名正确 | `ab_repeat.sh`：第 1 轮后比两侧 ttgir 指纹，相同即跳过该算子 |
**结论：一次运行的数字没有意义，只有一对同口径运行的差值有意义。** 每次运行把自己的
口径写进 `runs/<run>/run_identity.json`；`batch_summary.py --ab` / `--ab-rounds` 在出
报告前核对这些字段，不一致就拒绝，因为**口径不同的两次运行会产出一份和有效报告长得
一模一样的表**——没有任何一步会报错，只是数字换了含义。确认两次运行确实可比时，用
`--allow-caliber-mismatch` 显式放行。
文档拦不住人：README 里早就写着"B 侧必须走 REPLAY_FROM"和"只有 record 数据支撑的
结论按未验证处理"，本仓仍然发布过一个 record 对 record 得到的 router_gemm 数字。
所以上表第四列是代码，不是提醒。
单算子精测流程之上另有批量筛查层（`run_batch.py`，见「批量多算子测试」），可对双仓库上千个算子做例行体检，产物结构与单算子完全同构。
## 快速开始
@@ -67,9 +91,10 @@ softmax:
| `USE_FLAGTUNE` | `0` | `0` 走普通 autotune（跳过搜索、快速验证）；`1` 用 FlagTune 扩展调优空间（首跑全量搜索、慢）。A/B 两侧须取同值 |
| `BENCH_LEVEL` | `core` | 透传上游 `--level`。`comprehensive` 相对 `core` 做两件事：① 在 `set_shapes()` 里并入子类 `set_more_shapes()`（`mm` 因此从 5 个 shape 变 29 个，含 N/K 到 152064 的大 K 与 `models_shapes/*.yaml` 的模型 shape）；② 在 `get_input_iter()` 里对每个 shape 再产出一次变体（BLAS 是转置侧），用例数翻倍。注意 ① 只在 `SHAPE_MODE=upstream` 下可见——传了 `SHAPE_FILE` 时 `_shape_inject_plugin` 会在上游合并之后再覆盖一次、把 shape 集拉回 yaml 内容（BLAS 那类 arity 不一致的情形除外，插件会退让）；② 不经插件，任何 shape 来源下都生效。sweep 成本同步放大，shape 多时配合 `PARALLEL_WARMUP_GPUS`。A/B 两侧须取同值 |
| `ZL_ISOLATE_TUNER_CACHE` | `0` | FlagGems 的 libtuner winner 存在 `$FLAGGEMS_CACHE_DIR/config_cache` 的 sqlite 里（默认 `~/.flaggems`），跨 run 长存。`1` 给本次 run 一个私有缓存目录，两侧各自冷调优（README「各自最优」口径），跑完随 Triton 缓存一起删——选中的 config 已在 `autotune_records/` 里，DB 本身无留存价值。共享缓存的风险是单向的：先跑的一侧写入 winner，后跑的一侧直接命中、只编译那几个胜出变体。实测 `router_gemm` 共享模式下第二次运行编译变体从 234 掉到 10、耗时 52s→17s，隔离模式下两次都是 234 个变体。走 `REPLAY_FROM` 的标准 A/B 不受影响（config 本就锁死），record 对 record 才会中招。取值无论如何都写进 `run_identity.json` 的 `tuner_cache` |
| `ZL_DIRECT_REPEAT` | `1` | direct 计时的重复次数，取中位。只作用于 direct 路径：整体 `timer=direct` 的运行，以及图捕获运行里逐行回退的那些点（标记会带上 `repeat=N`）；捕获成功的行不受影响。`do_bench` 自己已在调用内取中位，压住的是迭代间抖动；跨调用重复补的是它看不见的那层——时钟爬坡、邻居进程进来这类漂移。调用方要 `quantiles` 或 `return_mode="all"` 时返回序列、跨调用中位无定义，自动不重复。在 MThreads 上这不是可选项：cvt 消除量最大的 `sort`/`dropout`/`rand` 恰好都无法捕获，那批结论正是由这条路径产出的。取值写进 `run_identity.json` 的 `direct_repeat` |
| `DTYPES` | 空（上游默认 dtype 扫描） | 空格分隔的 dtype 白名单（如 `"bfloat16 float16"`），逐个转为上游 `--dtypes`。算子不支持时上游报 `can't be supported by this op` |
| `PARALLEL_WARMUP_GPUS` | 空（串行） | 设为 `N`（≥2）时把 autotune sweep 分片到 N 卡并行，测量仍单卡串行；不设或 `<2` 则完全串行（见「加速：多卡并行 sweep」） |
| `ZL_ENABLE_CUDAGRAPH` | `auto` | `auto`/`1` 仅在 CUDA 上走 Graph 计时；非 CUDA 即使设 `1` 也拒绝并保持 `timer=direct` |
| `ZL_ENABLE_CUDAGRAPH` | `auto` | `auto`/`1` 在**声明了 `supports_cuda_graph` 的后端**上走 Graph 计时（目前 cuda 与 musa），其余后端即使设 `1` 也拒绝并保持 `timer=direct`。注意 `timer=cuda_graph` 只表示**尝试过**捕获，不表示有任何一行捕获成功——S5000 上 `dropout`/`rand`/`masked_select` 是逐 shape 全部回退（每行 `BENCHMARK_DIRECT_NO_CUDAGRAPH`，零毒化、不丢行），整张表实际是 direct 计时。这种情况下设 `0` 走的是同一条路径，只是记录更诚实 |
| `ZL_ENABLE_AUTOTUNE` | `1` | `0` 关闭 autotune record/replay 插件 |
| `ZL_ENABLE_IR` | `1` | `0` 关闭 TTGIR / 编译元数据插件 |
| `ZL_ENABLE_MM_CLUSTER` | `0` | `1` 且后端支持时加载 Hopper mm cluster 规避 |
@@ -88,7 +113,9 @@ softmax:
```
runs/softmax_20260716-031752/
├── run.log # 完整日志（含 SUCCESS 行的 latency/speedup 表；已去 ANSI 色的纯文本）
├── run_identity.json # provider/backend/device/timer/vendor（插件合并写入）
├── run_identity.json # 本次运行的口径：provider/backend/device/timer/vendor/
│ # bench_level/shape_mode/tuner_cache/direct_repeat/autotune_mode
│ # （shell 写 stub，插件合并写入；对比工具据此判可比性）
├── shapes.yaml # 本次实际使用的 shape（存档）
├── autotune_records/ # Triton autotune 选中的 config（供 replay）
│ ├── softmax.json
@@ -110,6 +137,137 @@ ttgir 关键设计：
- **命名 = 组内有区分度的参数**：同一 (shape, kernel) 组内取值相同的 constexpr 不进文件名；多词参数缩写为首字母（`BLOCK_SIZE_M` → `BSM`，图例见 naming.md）；同名冲突依次用 dtype（`__fp16`）、参数对齐特化（`__EMdiv16`）、hash 前缀消歧。
- **崩溃也能拿到 IR**：dump 挂在进程 atexit 上，CUDA crash 后仍会落盘已编译部分。
## 多卡跑一批：campaign.sh
`ab_repeat.sh` 是单卡串行的。一批算子要并行时用 `campaign.sh` 编排——它只分活，
测量仍然全部由 `ab_repeat.sh` 做，那个脚本一行不动。
```bash
SWITCH=FLAGTREE_MUSA_RLC_ENHANCE DEVICES=0,1,2,3,4,5,6,7 SCREEN=runs/ab_rlc_XXX/screen.tsv ZL_BACKEND=musa FLAGGEMS_DIR=/path/to/FlagGems DRY_RUN=1 bash campaign.sh
```
`DRY_RUN=1` 先看方案，确认了再去掉它起跑。
- **只取筛选表里 `IR_CHANGED` 的算子**——第一步的判定在这里生效
- **按实测耗时贪心分片**（`partition.py`），不是 hash 取模。S5000 上 16 个算子实测：
hash 分片关键路径 120 分钟、有一张卡空转；贪心 70 分钟。**下界由最重的单个算子
决定**（`topk` 一个就 70.6 分钟），加卡不解决，分片只保证别的卡不空等
- **每卡独立 `AB_ROOT`**——八个 `ab_repeat.sh` 同一秒启动，默认的时间戳目录名会撞
- **强制 `ZL_ISOLATE_TUNER_CACHE=1`**。8 个进程写同一个 sqlite 是要避开的事。代价是
pin 的 config 与共享缓存批次不同，**两批的绝对延迟不可互引**——口径守卫会自己拒绝
- `EXTRA_OPS` 追加筛选表里没有的算子（如 benchmark 刚修好的），放到**负载最小**的卡上
- 跑完合并各卡 `status.tsv`（首列是卡号），并统计非零退出的格与三个标记的总数
设备可见性变量名从 backends 层取，不写死 `MUSA_VISIBLE_DEVICES`。
## 两步：先筛选，再实测
跑一个算子的六轮 A/B 要 13~44 分钟（S5000 实测：`rand` 15 shape 13.0 分钟，
`sort` 25 shape 43.4 分钟）。用 `ab_rlc_test.sh` 筛一对只要约 0.8 分钟。
**差 20~50 倍**，所以先筛后测：
```bash
# 第一步：一对定去留（A 侧 record，B 侧 REPLAY_FROM 它，config 锁死）
OPS="col2im:col2im topk:topk perf_fft:fft" VENDOR=MUSA bash ab_rlc_test.sh
# 第二步：只把 IR_CHANGED 的送进来
SWITCH=FLAGTREE_MUSA_RLC_ENHANCE ROUNDS=6 OPS="topk:topk" bash ab_repeat.sh
```
`ab_rlc_test.sh` 除了原有的 speedup 区间，还给每个算子写一行判定到
`<root>/screen.tsv`：
| 判定 | 含义 | 该怎么办 |
|---|---|---|
| `IR_CHANGED` | 两侧 ttgir 不同 | 排六轮 A/B |
| `NO_IR_DIFF` | 逐字节相同（config 已锁死） | 跳过。跑六轮只会测到噪声，并报成一个"无差异"结论 |
| `NO_TRITON` | 跑出了行，但没编出 `.ttgir` | 本次运行没有任何东西经过 Triton。算子本身可能完全正常——多半跑的是 aten 或厂商库的 kernel——只是被测的编译器 pass 没有作用对象。**既不是"开关无作用"，也不是"没有 kernel"**。`run_pytest.sh` 每个 run 用独立 `TRITON_CACHE_DIR`，所以"缓存命中没重新编译"这个解释已被排除 |
| `NO_ROWS` | 没有测量行 | 先修可运行性：多半是 dtype 不被该算子接受，或它在本后端拿不到 config |
两件容易做错的：
- **判据是 ttgir 指纹，不是 `convert_layout` 计数。** cvt 数相同不等于生成代码
相同，而要测的是"生成代码变了没有"。两个方向都会骗人：`addmm` 的 cvt 是 15→15
没变，15 个 ttgir 文件却实质全不同；`tanh`/`softmax` 的 cvt 是 0→0，生成代码也
真的没动。cvt 数照样记，它解释的是变化的性质。
- **指纹先抹掉开关自己盖的注解。** FlagTree 的 `_apply_musa_rlc_policy(mod)` 在开关
打开时把整套 RLC 策略旋钮盖成 module 属性（`ttg.rlc-*`），跟这个 pass 后面有没有
真的改动什么无关。不抹的话判据有个常量偏移——**凡是能编出 kernel 的算子都必然判成
"变了"，筛选器谁都不排除**。24 个算子实测：抹之前 23 个 `IR_CHANGED`，抹之后 7 个
落回 `NO_IR_DIFF`。零对照抓不到这个，因为两侧同值时属性也相同。
- **不要全局写死一个 `DTYPES`。** `OPS` 元素支持 `OP:OP_FILE:DTYPES`，留空表示用
该算子自己的全套。`fft` 只吃 complex64，被一刀切的 `DTYPES=float16` 判死过一次
——那是 harness 的问题，被记成了算子的问题。
同一份 `_ir_fingerprint.sh` 也被 `ab_repeat.sh` 用于第 1 轮后的对比轴有效性检查，
两处判据不会漂开。
判定还管住报告：`NO_ROWS` 与 `NO_TRITON` 不出 speedup 区间（前者没有可比的测量，
后者与被测开关无关）；`NO_IR_DIFF` 照出，但明说那是**本机对该算子的噪声底**，不是
开关的收益——它离零多远，就是该算子单次测量的可信下限。判定刚说完这一对不可比，
下一行就给个 `+0.3%`，正是这套工具一路在防的"看起来像结论的表"。
`screen.tsv` 末七列是口径（`backend device timer bench_level shape_mode
tuner_cache direct_repeat`），由 `python batch_summary.py --caliber <run_dir>` 取——
`CALIBER_FIELDS` 只有那一处定义，筛选表不重列一遍。两份不同 `--level` 的
`screen.tsv` 放一起比时，这几列是唯一能拦住的东西。
每对有 `CELL_TIMEOUT`（默认 2400 秒）兜底，超时打 `!!! CELL_TIMEOUT` 并按耗时问题
报告，不混进 dtype 那条解释里。`elapsed_before/after` 也进表：两步流程的全部依据
就是"一对 0.8 分钟对六轮 13~44 分钟"这个比例，那这一步该记下自己那一半。
## 算子跑不起来时的排查清单
`NO_ROWS` / `NO_TRITON` 说的是"这一格没测出东西"，不是"开关无作用"。按下面的顺序查，
每一条都踩过：
| # | 症状 | 查什么 | 实例 |
|---|---|---|---|
| 1 | `PREFLIGHT_FAILED check=interpreter imports triton + flag_gems` | 用的是哪个 python。triton 通常来自 PYTHONPATH 指向的编译器构建 | 一次噪声标定因此废掉 12 格 + 160 次负载 run |
| 2 | `PREFLIGHT_FAILED check=benchmark file exists` | `OPS` 元素是 `OP:OP_FILE`，写反就找不到文件 | `test_perf_sort` 在 `test_sort.py` 里，要写 `perf_sort:sort` |
| 3 | `ValueError: Given dtype(s) '...' can't be supported by this op` | 该算子只吃特定 dtype，别全局写死 `DTYPES` | `fft` 只吃 complex64，被一刀切的 fp16 判成"跑不出行"，而它是 cvt 消除量最大的算子 |
| 4 | `dynamic_func() missing ... BLOCK_M, BLOCK_N ...` 之类 JIT 深处的报错 | **benchmark 是不是导入了别家厂商的实现**（见下） | `router_gemm` 拿 NVIDIA Hopper 的 kernel 跑在 MUSA 上 |
| 5 | 跑出行但 `kernels=0` | 该算子在本后端没走 Triton，走的是 aten 或厂商库 | — |
### 第 4 条：import 成功不等于这个后端支持
FlagGems 的厂商实现放在 `flag_gems/runtime/backend/_<vendor>/ops/`。benchmark 里常见
这种可用性判据：
```python
try:
from flag_gems.runtime.backend._nvidia.hopper.ops.mm import router_gemm
ROUTER_GEMM_AVAILABLE = True # ← import 成功就置 True
except Exception:
ROUTER_GEMM_AVAILABLE = False
```
**`import` 只证明那个 `.py` 在磁盘上，它不碰设备，所以在任何后端上都成功。** gate 于是
形同虚设：MUSA 上照样置 True，benchmark 拿 NVIDIA Hopper 的 kernel 跑在摩尔线程卡上，
最后死在 JIT 深处的 `dynamic_func() missing BLOCK_M, BLOCK_N, BLOCK_K, SPLIT_K`——
**那个报错完全指不向真正的原因**。而 `_mthreads/ops/mm.py` 下明明有一份能用的实现，
benchmark 从头到尾没导入过。
正确写法是先判 vendor（同一个文件里 Cambricon 那支就是对的）：
```python
if base.vendor_name == "nvidia":
try:
from flag_gems.runtime.backend._nvidia.hopper.ops.mm import router_gemm
ROUTER_GEMM_AVAILABLE = True
except Exception:
pass
```
`_device_setup_plugin` 会在 session 结束时扫 `sys.modules`，载入过别家厂商实现时打：
```
FOREIGN_VENDOR_MODULE active=mthreads loaded=nvidia
```
**它不拦，只报**——载入别家模块本身不一定有问题（有些是共享工具），但一个算子结果可疑
时，这行是第一个该看的地方。
## A/B 对比测试标准流程
```bash
@@ -161,6 +319,91 @@ benchmark 目录，注册 `--level`/`--mode` 的 `conftest.py` 就没被加载
只校验节点的文件那一半；`::test_<fn>` 交给 pytest，函数不存在时它自己会清楚地报
`not found: ...`。
还校验**解释器**——`python -c "import triton, flag_gems"`。triton 通常来自 PYTHONPATH
指向的编译器构建产物、flag_gems 来自源码检出，所以一个能摸到全部正确路径的 shell
仍然可能是错的那个 Python。不查的话，失败以每格一次 `pytest exit 4`（conftest 的
ImportError）出现，多轮 campaign 会在有人读日志之前把每一格都烧掉——这不是假想，
2026-09-10 的噪声标定实验就这么废了 12 格加 160 次负载 run。报错里带上
`$(command -v python)`，因为要看的就是"用的到底是哪个解释器"。
### 口径核对：出报告前的拒绝
`batch_summary.py --ab` / `--ab-rounds` 在解析任何 latency 之前，先读参与对比的每个
run 目录的 `run_identity.json`，逐字段核对 `backend / device / timer / bench_level /
shape_mode / tuner_cache / direct_repeat`。任何一项不一致，或任何一侧缺 identity，
就打 `!!! [caliber] ...` 并以 **4** 退出，不出报告。
```
!!! [caliber] sort: timer 不一致：before=cuda_graph, after=direct
!!! [caliber] sort: 拒绝出报告。确认口径确实可比后加 --allow-caliber-mismatch
```
单独判 `autotune_mode`：一侧 record 定 config、另一侧 replay 它是正常形态；**两侧都
record 直接拒绝**——那正是本仓发布过错误 router_gemm 数字的那次形态。
不拒绝但会警告的一类：在**共享** libtuner 缓存上 record。被 pin 的 config 取决于此前
在这台机器上跑过什么，相对结论仍成立，绝对延迟与变体数不可跨机器引用。
`--allow-caliber-mismatch` 放行，仍会把问题打出来。用它之前先能说清为什么这两次运行
可比；跨口径引用别人的数字，比自己重跑一遍贵得多。
### 多轮重复：ab_repeat.sh 与 --ab-rounds
单对观测无法说明自己稳不稳。S5000 实测 `topk` 消除了 32 个 `convert_layout`，
42 个 shape 里却有 17 个在两次同等有效的运行之间**改变了 delta 的符号**——那不是
计时口径问题（图捕获与 direct 的延迟比是 1.00x），是收益本身落在零附近、单次
测量给不出方向。
```bash
SWITCH=FLAGTREE_MUSA_RLC_ENHANCE OPS="topk masked_select" ROUNDS=6 ZL_BACKEND=musa FLAGGEMS_DIR=/path/to/FlagGems bash ab_repeat.sh
```
口径：第 1 轮 A 侧 record config，此后每一次运行（含 A 侧的后续轮次）都
`REPLAY_FROM` 那一份，config 因此跨轮跨侧全部锁死；逐轮交替先后顺序抵消单向漂移。
有轮次失败的算子不出报告——缺轮的报告会让人误以为是完整的 N 轮。
每跑完一格往 `<root>/status.tsv` 追一行：`round op side switch rc elapsed_s success
failed replay_fallback no_cudagraph capture_poisoned`。README 要求必查的三个标记就在
这张表里，跑的时候就能看见它们从 0 变成非 0，不用等结束再 grep——事后 grep 只在有人
想起来的时候才发生。每格另有 `CELL_TIMEOUT`（默认 2400 秒）兜底，超时记 `rc=124` 并打
`!!! CELL_TIMEOUT`，不会让一格卡死拖住整批。
全程还采一份 `<root>/telemetry.tsv`：
```
timestamp index sm_clock_mhz mem_clock_mhz temp_c power_w util_pct
2026-09-10T07:41:12 7 1888 2500 36 477.68 87
```
列在 cuda 与 musa 上完全一致（`nvidia-smi --query-gpu` 与 `mthreads-gmi -q --json`
各自解析后归一），所以两台机器的 campaign 可以并排读。理由很直接：六轮 A/B 声称的是
0.01% 量级的单侧噪声，**降频和热漂移能移动比这大得多的延迟，而且在 run.log 里不留
任何痕迹**——报告照样干净。`TELEMETRY_INTERVAL`（默认 5 秒）设 0 关闭；采样器缺失或
没权限时 `telemetry.py` 安静退出并把原因写进 `telemetry.err`，不影响测试本身。
第 1 轮 A/B 跑完后，脚本比两侧 `ttgir/` 的指纹（相对路径 + 内容）。**逐字节相同就
跳过这个算子**，并打 `!!! SWITCH_NO_IR_DIFF op=... switch=...`：这个开关没改变该算子的
生成代码，剩下 N-1 轮测的全是噪声，最后还会报成一个"无差异"的结论。先查开关名拼写
与它在被测源码里的读取点；确属只影响运行时、不影响 IR 的开关，用
`SWITCH_ALLOW_NO_IR_DIFF=1` 放行。
报告分开给两种离散度，它们回答不同的问题：
| 指标 | 含义 |
|---|---|
| 逐轮区间 | N 轮各自 delta 的 min~max，说明**判定**稳不稳 |
| 单侧噪声 | 同一侧 N 次延迟的极差，说明**机器**吵不吵 |
| 符号不稳行数 | delta 在轮次间改变符号的 shape 数，这些行的方向不可结论 |
已有产物可以单独重出报告，目录形态是 `<root>/r<k>_<A|B>/run.log`：
```bash
python batch_summary.py --ab-rounds runs/ab_repeat_XXX/topk --label topk
```
NV 上验证过这条路径：`router_gemm` 六轮把逐轮区间收窄到 -16.6% ~ -16.2%、单侧
噪声 0.4%，把单对观测升级成了结论。
### ab_fold_test.sh / ab_rlc_test.sh：A/B 测试参考示例
`ab_fold_test.sh` 是上述 A/B 流程的现成参考：以 FlagGems `FLAGGEMS_MXFP4_FOLDSCALE`（fold_scale 优化）开关为对比对象，对 4 个 DeepSeek-V4-Flash 真实 trace shape 集（`shape/DeepSeek-V4-Flash-p{1024,4096,32768,65536}d1024.yaml`，随仓库提供）各跑一对 FOLD=0（基线，record）→ FOLD=1（replay 同 config），共 8 轮：
@@ -268,8 +511,8 @@ python run_batch.py --ops-file my_ops.txt --replay-root runs/batch_xxx
| `_shape_iter_inject_plugin` | 覆盖在 `get_input_iter` 里硬编码 shape 的类（conv/pool 等） | 不关 |
| `_bespoke_shape_plugin` | 覆盖特殊输入构造的算子（upsample/flash_mla/cutlass 等，按类名注册） | 测这些算子之外可关 |
| `_autotune_record_plugin` | record/replay Triton autotune 选择；`autotune_records/meta.json` 记录 backend，跨后端 replay 拒绝 | `ZL_ENABLE_AUTOTUNE=0` |
| `_timing_plugin` | 按后端选择计时路径：CUDA 可走 Graph（内部调用 `_cudagraph_plugin`），其余后端一开始就是上游 `do_bench`（`timer=direct`） | `ZL_ENABLE_CUDAGRAPH=0` 强制 direct |
| `_cudagraph_plugin` | 仅由 `_timing_plugin` 在 CUDA 上调用。`do_bench` → `do_bench_cudagraph`，capture 前按 warmup 预算显式预热；无法 capture 时回退并打 `BENCHMARK_DIRECT_NO_CUDAGRAPH` | 不要直接 `-p` 加到非 CUDA 运行 |
| `_timing_plugin` | 按后端的 `supports_cuda_graph` 选计时路径：声明了的走 Graph（内部调用 `_cudagraph_plugin`，逐 shape 回退），没声明的一开始就是上游 `do_bench`（`timer=direct`） | `ZL_ENABLE_CUDAGRAPH=0` 强制 direct |
| `_cudagraph_plugin` | 仅由 `_timing_plugin` 在声明了 `supports_cuda_graph` 的后端上调用。`do_bench` → `do_bench_cudagraph`，capture 前按 warmup 预算显式预热；无法 capture 时回退并打 `BENCHMARK_DIRECT_NO_CUDAGRAPH` | 不要直接 `-p` 加到非 CUDA 运行 |
| `_musa_benchmark_compat_plugin` | 剥掉 FlagGems 上游若干 benchmark 上过时的 `torch.cuda.is_available()` 跳过门——被测代码本就经 `flag_gems.device` 分发，在 MThreads 上那些门把可用算子变成「未测量」。每条覆盖都钉死到一个文件与一条确切的 skip 理由，且仅在 `flag_gems.vendor_name == "mthreads"` 时生效 | 由 `backends/musa.py` 的 `compat_plugins` 声明，其他后端不加载 |
| `_pretty_report_plugin` | 结果表整理 + 着色：全表相同的输入折叠成表头下一行图例，每行 Size Detail 只留随行变化的部分（shape 保持 `torch.Size([...])` 原样，MoE 类算子单行从 ~400 字符缩到一屏内）；SUCCESS 绿 / FAILED 红。列名与 `SUCCESS` 字样保持上游原文，`run.log` 的 grep/解析不受影响 | 需要与上游原始表格逐字对齐时去掉 |
| `_ir_meta_plugin` | 编译期记录每个变体的 constexpr/签名/特化；launch 钩子统计每个 (kernel, shape) 的实际使用；退出时按上述结构 dump ttgir。Triton knobs 缺失时 fail-soft | `ZL_ENABLE_IR=0` |
_cudagraph_plugin.py
+48 -9
View File
@@ -33,6 +33,47 @@ import triton.testing as _tt
_ORIGINAL_DO_BENCH = _tt.do_bench
def direct_repeat() -> int:
"""ZL_DIRECT_REPEAT, clamped to >= 1. 1 keeps today's single call."""
try:
return max(1, int(os.environ.get("ZL_DIRECT_REPEAT", "1")))
except ValueError:
return 1
def run_direct(fn, warmup, rep, grad_to_none, quantiles, return_mode):
"""Upstream do_bench, optionally repeated with the median taken.
Direct timing walks the full Python launch path once per iteration, so on
small kernels it carries launch overhead and CPU jitter that graph capture
does not. do_bench already returns a median over the iterations inside one
call, which suppresses iteration-level jitter but not drift across a call:
clocks ramping, another process arriving. Repeating the call and taking the
median over calls is the part the inner median cannot see.
It matters most where it is least optional. On MThreads the operators with
the largest RLC convert reductions -- sort, dropout, rand -- are exactly the
ones that cannot be graph-captured, so the headline numbers there are
produced by this path.
Repetition is skipped when the caller asks for quantiles or return_mode
"all", because the result is then a sequence and a median across calls is
not defined for it; the single call is returned unchanged.
"""
n = direct_repeat()
if n == 1 or quantiles is not None or return_mode == "all":
return _ORIGINAL_DO_BENCH(
fn, warmup=warmup, rep=rep, grad_to_none=grad_to_none,
quantiles=quantiles, return_mode=return_mode)
import statistics
samples = [
_ORIGINAL_DO_BENCH(fn, warmup=warmup, rep=rep, grad_to_none=grad_to_none,
quantiles=None, return_mode=return_mode)
for _ in range(n)
]
return statistics.median(samples)
def _device_interface():
"""Triton's active device interface: torch.cuda on NVIDIA, torch_musa on MThreads."""
from triton.runtime import driver
@@ -194,16 +235,14 @@ def _patched_do_bench(fn, warmup=25, rep=100, grad_to_none=None,
def _direct(note):
from _term_style import YELLOW, paint
print(paint(f"BENCHMARK_DIRECT_NO_CUDAGRAPH op={op} {note}", YELLOW),
n = direct_repeat()
# The repeat count belongs in the marker: these rows are already a
# different caliber from the captured ones, and now they can also be a
# different number of samples.
rpt = "" if n == 1 else f" repeat={n}"
print(paint(f"BENCHMARK_DIRECT_NO_CUDAGRAPH op={op} {note}{rpt}", YELLOW),
flush=True)
return _ORIGINAL_DO_BENCH(
fn,
warmup=warmup,
rep=rep,
grad_to_none=grad_to_none,
quantiles=quantiles,
return_mode=return_mode,
)
return run_direct(fn, warmup, rep, grad_to_none, quantiles, return_mode)
backend = None
try:
_device_setup_plugin.py
+43
View File
@@ -67,3 +67,46 @@ def pytest_configure(config):
_setup_at_import()
# FlagGems 的厂商实现按 flag_gems.runtime.backend._<vendor>.ops.* 组织。benchmark
# 常写成 `try: import <某厂商实现>; AVAILABLE = True`——但 import 只证明那个 .py 在
# 磁盘上，它不碰设备，所以在任何后端上都会成功。于是 gate 形同虚设，测出来的是别家
# 的 kernel 跑在本机上。router_gemm 就是这样：test_router_gemm.py 无条件导入
# _nvidia.hopper 版，在 MUSA 上 gate 照过，最后死在 JIT 深处的
# "dynamic_func() missing BLOCK_M, BLOCK_N, BLOCK_K, SPLIT_K"——那个报错完全指不
# 向真正的原因，而 _mthreads 下明明有一份能用的实现。
#
# 这里不拦，只报：跑完看一眼载入过谁家的实现。
_VENDOR_MOD_PREFIX = "flag_gems.runtime.backend._"
def foreign_vendor_modules(active_vendor: str, modules=None) -> list:
"""本次进程里载入过的、不属于当前厂商的 FlagGems 后端模块。"""
if modules is None:
modules = sys.modules
out = set()
for name in list(modules):
if not name.startswith(_VENDOR_MOD_PREFIX):
continue
# 只保留厂商根，_nvidia.hopper.ops.mm 与 _nvidia.ops.mm 算同一条。
# 按段比而不是按前缀：前缀会让 _<vendor>_something 被误认成自家。
root = name[len(_VENDOR_MOD_PREFIX):].split(".")[0]
if root != active_vendor:
out.add(root)
return sorted(out)
def pytest_sessionfinish(session, exitstatus):
from _term_style import tag
try:
vendor = get_backend().flaggems_vendor
except Exception:
return
foreign = foreign_vendor_modules(vendor)
if not foreign:
return
print(f"{tag('[device-setup-plugin]')} FOREIGN_VENDOR_MODULE active={vendor} "
f"loaded={','.join(foreign)}", file=sys.stderr, flush=True)
print(f"{tag('[device-setup-plugin]')} 本次载入了别家厂商的 FlagGems 实现。"
"如果这个算子的结果可疑，先查 benchmark 的可用性判据是不是只做了 import "
"而没判 vendor_name", file=sys.stderr, flush=True)
_ir_fingerprint.sh
+31
View File
@@ -0,0 +1,31 @@
#!/bin/bash
# 由 ab_repeat.sh 与 screen.sh 共用；两处必须用同一个判据，否则筛选放行的算子
# 到了 A/B 那步又被拦下，或者反过来。
# 一个 run 生成代码的指纹：只取 .ttgir 的相对路径 + 内容，并先抹掉开关自己盖上去的
# 注解。
#
# 只认 .ttgir：dump 目录里还有 index.tsv 和 naming.md，一个 kernel 都没编出来的 run
# 只剩这两个文件、两侧完全相同，会把"什么都没生成"报成"开关没改变生成代码"。
#
# 抹掉 ttg.rlc-*：FlagTree 的 _apply_musa_rlc_policy(mod) 在开关打开时把整套 RLC 策略
# 旋钮盖成 module 属性，跟这个 pass 后面有没有真的改动什么无关。不抹的话判据有个常量
# 偏移——凡是能编出 kernel 的算子都必然判成"变了"，筛选器谁都不排除。实测 24 个算子
# 里有 7 个（tanh gelu mul softmax mm bmm replication_pad3d）除这行属性外逐字节相同。
# 零对照抓不到它：两侧同值时属性也相同。
#
# 零 kernel 返回非零，交给调用方当作"没有可比的 IR"，而不是"一致"。
ir_fingerprint() {
local dir="$1/ttgir"
[[ -d "$dir" ]] || return 1
local n
n=$(find "$dir" -name '*.ttgir' -type f 2>/dev/null | wc -l)
(( n > 0 )) || return 1
local out
out=$( { find "$dir" -name '*.ttgir' -type f | LC_ALL=C sort | while IFS= read -r f; do
printf '%s
' "${f#"$dir"/}"
sed -E 's/"ttg[.]rlc-[^"]*" = [0-9-]+ : i32, ?//g' "$f"
done; } | sha256sum | cut -d' ' -f1 )
[[ -n "$out" ]] && printf '%s' "$out"
}
_timing_plugin.py
+31 -8
View File
@@ -1,10 +1,15 @@
"""Select the timing path from backend capabilities, before any shape runs.
CUDA keeps the existing `_cudagraph_plugin` behaviour (graph capture, with
per-shape fallback to plain do_bench). Other backends do not attempt CUDA
Graph at all: they leave `triton.testing.do_bench` in place and record
`timer=direct`. Trying capture-then-fallback on a non-CUDA device would
produce FATAL_DEVICE noise and mixed-caliber rows.
Whether capture is attempted comes from the backend's `supports_cuda_graph`,
not from "is this CUDA". CUDA and MUSA both declare it, so both get graph
capture with per-shape fallback to plain do_bench; a backend that declares
nothing leaves `triton.testing.do_bench` in place and records `timer=direct`.
`timer=cuda_graph` therefore means capture was *attempted*, not that any row
was captured. On S5000 every shape of dropout/rand/masked_select falls back
(BENCHMARK_DIRECT_NO_CUDAGRAPH on every row, nothing poisoned, no rows lost),
so the whole table is direct-timed under a cuda_graph label. Setting
ZL_ENABLE_CUDAGRAPH=0 there selects the same timing path and says so plainly.
"""
from __future__ import annotations
@@ -37,11 +42,29 @@ def pytest_configure(config):
if _graph_requested(backend):
import _cudagraph_plugin as _cg
_cg.pytest_configure(config)
update_identity(timer="cuda_graph", graph_capture="attempted")
# Recorded on this branch too: the fallback rows a captured run produces
# are timed by the direct path and take the same repeat count.
update_identity(timer="cuda_graph", graph_capture="attempted",
direct_repeat=_cg.direct_repeat())
print(f"{tag('[timing-plugin]')} timer=cuda_graph backend={backend.name}",
file=sys.stderr, flush=True)
return
update_identity(timer="direct", graph_capture="disabled")
# A wholly-direct run gets the same repetition the fallback rows get, so the
# "put everything on one timer" advice does not silently cost the samples.
# Patching only when repeat > 1 keeps the default path untouched: upstream's
# do_bench stays in place exactly as before.
import _cudagraph_plugin as _cg
repeat = _cg.direct_repeat()
if repeat > 1:
def _repeated(fn, warmup=25, rep=100, grad_to_none=None,
quantiles=None, return_mode="mean"):
return _cg.run_direct(fn, warmup, rep, grad_to_none,
quantiles, return_mode)
import triton.testing as _tt
_tt.do_bench = _repeated
update_identity(timer="direct", graph_capture="disabled",
direct_repeat=repeat)
print(f"{tag('[timing-plugin]')} timer=direct backend={backend.name} "
"(upstream do_bench; graph capture not selected)",
f"(upstream do_bench; graph capture not selected)"
+ (f"; median of {repeat} calls" if repeat > 1 else ""),
file=sys.stderr, flush=True)
ab_repeat.sh
+168
View File
@@ -0,0 +1,168 @@
#!/bin/bash
# 多轮 A/B：同一对比在 N 轮独立进程里各跑一次，逐 shape 取轮次中位。
#
# 为什么需要它：单对观测无法说明自己稳不稳。S5000 实测 topk 消除了 32 个
# convert_layout，42 个 shape 里却有 17 个在两次同等有效的运行之间改变了 delta
# 的符号——那不是计时口径的问题（图捕获与 direct 的延迟比是 1.00x），是收益本身
# 落在零附近、单次测量给不出方向。多轮之后才谈得上"这个判定成立"。
#
# 口径：第 1 轮 A 侧 record autotune config，此后每一次运行（含 A 侧自己的后续
# 轮次）都 REPLAY_FROM 那一份。config 因此跨轮跨侧全部锁死，剩下的变量只有被测
# 开关与测量噪声。逐轮交替先后顺序，抵消单向漂移。
#
# 用法：
# SWITCH=FLAGTREE_MUSA_RLC_ENHANCE OPS="topk masked_select" ROUNDS=6 \
# ZL_BACKEND=musa FLAGGEMS_DIR=/path/to/FlagGems bash ab_repeat.sh
#
# OPS 元素为 OP 或 OP:OP_FILE（省略时 OP_FILE=OP）。
# SWITCH 是被测环境变量，A 侧设 0、B 侧设 1。
#
# 对比轴有效性由脚本自己查：第 1 轮 A/B 跑完后比两侧的 ttgir 指纹，完全一致说明
# 这个开关在这个算子上没改变任何生成代码——继续跑满 N 轮只会得到一个"无差异"的假
# 结论（同 ab_fold_test.sh 的告诫）。这类算子默认跳过；开关确实只影响运行时而不影响
# IR 时，用 SWITCH_ALLOW_NO_IR_DIFF=1 放行。
#
# 产物：runs/ab_repeat_<开关>_<时间戳>/<op>/r<k>_<A|B>/，跑完对每个算子调用
# batch_summary.py --ab-rounds 出报告。
set -uo pipefail
cd "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SWITCH_ALLOW_NO_IR_DIFF="${SWITCH_ALLOW_NO_IR_DIFF:-0}"
source "$(dirname "${BASH_SOURCE[0]}")/_ir_fingerprint.sh"
: "${SWITCH:?SWITCH is required, e.g. FLAGTREE_MUSA_RLC_ENHANCE}"
OPS="${OPS:-}"
: "${OPS:?OPS is required, e.g. \"topk masked_select\"}"
ROUNDS="${ROUNDS:-6}"
CELL_TIMEOUT="${CELL_TIMEOUT:-2400}"
if [[ -t 1 && -z "${NO_COLOR:-}" ]]; then
export FLAGGEMS_PERF_COLOR=always
fi
# AB_ROOT 供 campaign.sh 指定每卡独立的产物根：八个 ab_repeat.sh 同一秒启动时，
# 默认的时间戳目录名会撞在一起，两张卡往同一个目录里写。
ROOT="${AB_ROOT:-runs/ab_repeat_${SWITCH}_$(date +%Y%m%d-%H%M%S)}"
mkdir -p "$ROOT"
LOG="$ROOT/summary.log"
STATUS="$ROOT/status.tsv"
# 全程采时钟/温度/功耗。六轮 A/B 声称的是 0.01% 量级的单侧噪声，降频和热漂移
# 能移动比这大得多的延迟，而且在 run.log 里不留痕迹——报告照样干净。
# TELEMETRY_INTERVAL=0 关闭。采不到时 telemetry.py 自己安静退出，不影响测试。
TELEMETRY_INTERVAL="${TELEMETRY_INTERVAL:-5}"
TELE_PID=""
if [[ "$TELEMETRY_INTERVAL" != "0" ]]; then
python telemetry.py --backend "${ZL_BACKEND:-cuda}" --device "${ZL_DEVICE:-}" \
--interval "$TELEMETRY_INTERVAL" --run \
> "$ROOT/telemetry.tsv" 2> "$ROOT/telemetry.err" &
TELE_PID=$!
trap '[[ -n "$TELE_PID" ]] && kill "$TELE_PID" 2>/dev/null' EXIT
fi
printf 'round\top\tside\tswitch\trc\telapsed_s\tsuccess\tfailed\treplay_fallback\tno_cudagraph\tcapture_poisoned\n' > "$STATUS"
# 跑一格并逐格登记。事后 grep 也能拿到这些数，但只在有人想起来时才发生；
# 缺格、超时、标记从 0 变成非 0，都要在跑的时候就看得见。
run_cell() { # round side op op_file out_dir [replay_from]
local round="$1" side="$2" op="$3" op_file="$4" dir="$5" replay="${6:-}"
local val=0; [[ "$side" == B ]] && val=1
local start rc elapsed log ok fail fb ng po
start=$(date +%s)
if [[ -n "$replay" ]]; then
OP="$op" OP_FILE="$op_file" OUT_DIR="$dir" REPLAY_FROM="$replay" \
timeout -k 30 "$CELL_TIMEOUT" env "$SWITCH=$val" bash run_pytest.sh
else
OP="$op" OP_FILE="$op_file" OUT_DIR="$dir" \
timeout -k 30 "$CELL_TIMEOUT" env "$SWITCH=$val" bash run_pytest.sh
fi
rc=$?
elapsed=$(( $(date +%s) - start ))
(( rc == 124 || rc == 137 )) && echo "!!! CELL_TIMEOUT ${CELL_TIMEOUT}s op=$op round=$round side=$side"
log="$dir/run.log"
# grep -c 没命中时打印 0 但以 1 退出；不吞掉的话会被当成本格失败
ok=$(grep -c '^SUCCESS' "$log" 2>/dev/null || true)
fail=$(grep -c '^FAILED' "$log" 2>/dev/null || true)
fb=$(grep -c 'AUTOTUNE_REPLAY_FALLBACK' "$log" 2>/dev/null || true)
ng=$(grep -c 'BENCHMARK_DIRECT_NO_CUDAGRAPH' "$log" 2>/dev/null || true)
po=$(grep -c 'BENCHMARK_CAPTURE_POISONED' "$log" 2>/dev/null || true)
printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
"$round" "$op" "$side" "$val" "$rc" "$elapsed" \
"${ok:-0}" "${fail:-0}" "${fb:-0}" "${ng:-0}" "${po:-0}" >> "$STATUS"
return $rc
}
{
echo ">>> switch=$SWITCH rounds=$ROUNDS ops=$OPS"
for entry in $OPS; do
op="${entry%%:*}"
op_file="${entry#*:}"
[[ "$op_file" == "$entry" ]] && op_file="$op"
out="$ROOT/$op"
mkdir -p "$out"
# 第 1 轮 A 侧 record，其余全部回放它，config 跨轮跨侧一致
echo "########## $op | round 1 | A ($SWITCH=0, record) ##########"
if ! run_cell 1 A "$op" "$op_file" "$out/r1_A"; then
echo "!!! FAILED $op round 1 A; skipping this operator"
continue
fi
REC="$out/r1_A"
ok=1; reason="有失败轮次"
for ((k = 1; k <= ROUNDS; k++)); do
# 奇数轮先 A 后 B，偶数轮反过来
if (( k % 2 == 1 )); then order="A B"; else order="B A"; fi
for side in $order; do
[[ $k -eq 1 && $side == A ]] && continue # 已经跑过
val=$([[ $side == A ]] && echo 0 || echo 1)
echo "########## $op | round $k | $side ($SWITCH=$val, replay) ##########"
if ! run_cell "$k" "$side" "$op" "$op_file" "$out/r${k}_${side}" "$REC"; then
echo "!!! FAILED $op round $k $side"
ok=0
fi
done
if (( k == 1 && ok )); then
fa=$(ir_fingerprint "$out/r1_A") || fa=""
fb=$(ir_fingerprint "$out/r1_B") || fb=""
if [[ -z "$fa" || -z "$fb" ]]; then
echo "??? $op: 没有编出任何 .ttgir（可能该算子没走 Triton，或本轮失败），"
echo "??? 跳过对比轴有效性检查——这不等于开关无作用"
elif [[ "$fa" == "$fb" ]]; then
echo "!!! SWITCH_NO_IR_DIFF op=$op switch=$SWITCH"
echo "!!! 两侧 ttgir 逐字节相同：这个开关没有改变 $op 的生成代码。"
echo "!!! 继续跑 $ROUNDS 轮只会测出噪声，并报成一个'无差异'结论。"
echo "!!! 先确认开关名拼写与它在被测源码里的读取点；确属只影响运行时的"
echo "!!! 开关，用 SWITCH_ALLOW_NO_IR_DIFF=1 重跑。"
if [[ "$SWITCH_ALLOW_NO_IR_DIFF" != "1" ]]; then
ok=0; reason="开关未改变该算子的生成代码"
break
fi
else
echo ">>> 对比轴有效：$op 两侧 ttgir 不同（A=${fa:0:12} B=${fb:0:12}）"
fi
fi
done
echo "########## $op | report ##########"
if (( ok )); then
python batch_summary.py --ab-rounds "$out" --label "$op" \
|| echo "!!! report failed: $op"
else
# 缺轮的算子照样出报告会让人误以为是完整的 N 轮
echo "!!! $op $reason，不出报告；产物在 $out"
fi
done
echo "########## ALL DONE ##########"
} 2>&1 | tee "$LOG"
sed -i -E $'s/\x1b\\[[0-9;]*[A-Za-z]//g' "$LOG"
echo ">>> 总 log: $(pwd)/$LOG"
echo ">>> 逐格状态表: $(pwd)/$STATUS"
[[ -s "$ROOT/telemetry.tsv" ]] && echo ">>> 时钟/温度: $(pwd)/$ROOT/telemetry.tsv"
[[ -s "$ROOT/telemetry.err" ]] && echo ">>> 遥测提示: $(cat "$ROOT/telemetry.err")"
echo ">>> 逐算子重出报告: python batch_summary.py --ab-rounds $ROOT/<op> --label <op>"
echo ">>> 解读要点: 逐轮区间宽说明判定不稳；单侧噪声大说明机器不稳；两者含义不同"
ab_rlc_test.sh
+149 -19
View File
@@ -10,15 +10,79 @@
#
# 跑完用 batch_summary.py --ab <before> <after> 得到该算子的 speedup 区间。
#
# 这一轮同时充当 ab_repeat.sh 的筛选前置：一对的代价约 0.8 分钟，六轮 A/B 是
# 13~44 分钟（S5000 实测），差 20~50 倍。所以先用它过一遍，只把 IR_CHANGED 的
# 算子送进 ab_repeat.sh。判定写在 <root>/screen.tsv，四种之一：
#
# IR_CHANGED 两侧 ttgir 不同 -> 排六轮
# NO_IR_DIFF 逐字节相同（config 已锁死）-> 跳过，跑六轮只会测到噪声
# NO_TRITON 跑出行但没编出 .ttgir -> 该算子在本后端没走 Triton，
# 编译器开关无从谈起。不是"开关无作用"
# NO_ROWS 没有测量行 -> 先修可运行性，多半是 dtype
#
# 判据是 ttgir 指纹而不是 convert_layout 计数：cvt 数相同不等于生成代码相同。
# cvt 数照样记，它解释的是变化的性质。
#
# 用法：
# OPS="softmax gelu_backward" bash ab_rlc_test.sh
# OPS="sort:perf_sort topk" bash ab_rlc_test.sh # 冒号分隔 OP:OP_FILE
# OPS="sort:perf_sort topk" bash ab_rlc_test.sh # OP:OP_FILE
# OPS="perf_fft:fft:complex64 topk:topk" bash ab_rlc_test.sh # 再加 DTYPES
#
# 注意：RLC 开关名带厂商前缀，用 VENDOR 指定（默认 MUSA）。套用前先确认开关有效：
# grep 该开关名于 FlagTree 源码应有命中，否则两侧跑的是同一份代码。
# DTYPES 留空表示用该算子自己的全套。**不要全局写死一个 dtype**：fft 只吃
# complex64，被一刀切的 DTYPES=float16 判死过一次——那是 harness 的问题，
# 却被记成了算子的问题。
#
# 注意：RLC 开关名带厂商前缀，用 VENDOR 指定（默认 MUSA）。开关名拼错时两侧跑
# 同一份代码，此时判定会是 NO_IR_DIFF——先查拼写，再下"开关无作用"的结论。
set -uo pipefail
cd "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
source "./_ir_fingerprint.sh"
# RESCORE=<已有的 root>：判据改了之后，用当前的 ir_fingerprint 对盘上已有的 ttgir
# 重新打分并改写该 root 的 screen.tsv，不重跑任何测量。
#
# 需要它是因为判据确实改过：_ir_fingerprint.sh 起初把 index.tsv 也算进哈希，后来又
# 发现要抹掉开关自己盖的 ttg.rlc-* 属性。同一批 24 个算子，结论从 23 个 IR_CHANGED
# 变成 16 个。重打分是秒级的，重跑那批筛选要 79 分钟。
#
# 原表存为 screen.tsv.before-rescore，判定有变化的算子逐个打出来。
RESCORE="${RESCORE:-}"
if [[ -n "$RESCORE" ]]; then
[[ -d "$RESCORE" ]] || { echo "!!! 目录不存在: $RESCORE" >&2; exit 2; }
TSV="$RESCORE/screen.tsv"
[[ -f "$TSV" ]] || { echo "!!! 没有 screen.tsv: $TSV" >&2; exit 2; }
cp "$TSV" "$TSV.before-rescore"
hdr=$(head -1 "$TSV.before-rescore")
# 列名 -> 序号，不写死列序
idx() { echo "$hdr" | tr '\t' '\n' | grep -nxF "$1" | cut -d: -f1; }
c_op=$(idx op); c_rb=$(idx rows_before); c_ra=$(idx rows_after); c_v=$(idx verdict)
[[ -n "$c_op" && -n "$c_v" ]] || { echo "!!! screen.tsv 缺 op/verdict 列" >&2; exit 2; }
printf '%s\n' "$hdr" > "$TSV"
changed=0
while IFS= read -r line; do
op=$(printf '%s' "$line" | cut -f"$c_op")
old=$(printf '%s' "$line" | cut -f"$c_v")
rb=$(printf '%s' "$line" | cut -f"${c_rb:-0}"); ra=$(printf '%s' "$line" | cut -f"${c_ra:-0}")
fa=$(ir_fingerprint "$RESCORE/${op}_before") || fa=""
fb=$(ir_fingerprint "$RESCORE/${op}_after") || fb=""
if [[ "${rb:-0}" == 0 || "${ra:-0}" == 0 ]]; then v=NO_ROWS
elif [[ -z "$fa" || -z "$fb" ]]; then v=NO_TRITON
elif [[ "$fa" == "$fb" ]]; then v=NO_IR_DIFF
else v=IR_CHANGED; fi
printf '%s\n' "$line" | awk -v FS='\t' -v OFS='\t' -v i="$c_v" -v v="$v" '{$i=v; print}' >> "$TSV"
if [[ "$old" != "$v" ]]; then
printf ' %-22s %-12s -> %s\n' "$op" "$old" "$v" >&2
changed=$((changed+1))
fi
done < <(tail -n +2 "$TSV.before-rescore")
echo ">>> 重打分完成，$changed 个算子的判定变了；原表存为 $TSV.before-rescore"
awk -v i="$c_v" -F'\t' 'NR>1{c[$i]++} END{for (k in c) printf(" %-12s %d\n", k, c[k])}' "$TSV"
exit 0
fi
VENDOR="${VENDOR:-MUSA}"
ENHANCE_VAR="FLAGTREE_${VENDOR}_RLC_ENHANCE"
@@ -26,38 +90,103 @@ ENHANCE_VAR="FLAGTREE_${VENDOR}_RLC_ENHANCE"
# 空格分隔的算子表。元素为 OP 或 OP:OP_FILE（省略时 OP_FILE=OP）。
OPS="${OPS:-softmax}"
# 筛选是第一次接触一个算子的地方，最容易撞上挂死的那个。没有上限时一个算子
# 能把整批停在那里，而筛选的全部意义就是快。
CELL_TIMEOUT="${CELL_TIMEOUT:-2400}"
if [[ -t 1 && -z "${NO_COLOR:-}" ]]; then
export FLAGGEMS_PERF_COLOR=always
fi
LOG="runs/ab_rlc_$(date +%Y%m%d-%H%M%S).log"
mkdir -p runs
# 显式 OUT_DIR，不再用 `ls -dt runs/<op>_*` 猜"最新那个就是我刚跑的"。串行下
# 那个假设成立，多卡并行共用 runs/ 就不成立了。
ROOT="runs/ab_rlc_$(date +%Y%m%d-%H%M%S)"
mkdir -p "$ROOT"
LOG="$ROOT/summary.log"
TSV="$ROOT/screen.tsv"
printf 'op op_file dtypes rows_before rows_after kernels_before kernels_after cvt_before cvt_after elapsed_before elapsed_after verdict backend device timer bench_level shape_mode tuner_cache direct_repeat\n' > "$TSV"
count_cvt() { find "$1/ttgir" -name '*.ttgir' -type f -exec cat {} + 2>/dev/null | grep -c convert_layout || true; }
count_kernels() { find "$1/ttgir" -name '*.ttgir' -type f 2>/dev/null | wc -l; }
count_rows() { grep -c '^SUCCESS' "$1/run.log" 2>/dev/null || true; }
{
echo ">>> vendor=$VENDOR switch=$ENHANCE_VAR ops=$OPS"
for entry in $OPS; do
op="${entry%%:*}"
op_file="${entry#*:}"
[[ "$op_file" == "$entry" ]] && op_file="$op"
op="${entry%%:*}"; rest="${entry#*:}"
if [[ "$rest" == "$entry" ]]; then op_file="$op"; dtypes=""
else
op_file="${rest%%:*}"; d="${rest#*:}"
[[ "$d" == "$rest" ]] && dtypes="" || dtypes="$d"
fi
BEFORE="$ROOT/${op}_before"; AFTER="$ROOT/${op}_after"
echo "########## $op | $ENHANCE_VAR=0 before (record) ##########"
if ! OP="$op" OP_FILE="$op_file" env "$ENHANCE_VAR=0" bash run_pytest.sh; then
echo "!!! FAILED before: $op"
continue
fi
BEFORE=$(ls -dt "runs/${op}_"* | head -1)
t0=$(date +%s)
OP="$op" OP_FILE="$op_file" OUT_DIR="$BEFORE" DTYPES="$dtypes" \
timeout -k 30 "$CELL_TIMEOUT" env "$ENHANCE_VAR=0" bash run_pytest.sh
rc_before=$?; el_before=$(( $(date +%s) - t0 ))
echo "########## $op | $ENHANCE_VAR=1 after (replay $BEFORE) ##########"
if ! OP="$op" OP_FILE="$op_file" REPLAY_FROM="$BEFORE" env "$ENHANCE_VAR=1" bash run_pytest.sh; then
echo "!!! FAILED after: $op"
continue
fi
AFTER=$(ls -dt "runs/${op}_"* | head -1)
t0=$(date +%s)
OP="$op" OP_FILE="$op_file" OUT_DIR="$AFTER" DTYPES="$dtypes" \
REPLAY_FROM="$BEFORE" timeout -k 30 "$CELL_TIMEOUT" env "$ENHANCE_VAR=1" bash run_pytest.sh
rc_after=$?; el_after=$(( $(date +%s) - t0 ))
for rc in "$rc_before" "$rc_after"; do
(( rc == 124 || rc == 137 )) && echo "!!! CELL_TIMEOUT ${CELL_TIMEOUT}s op=$op"
done
rb=$(count_rows "$BEFORE"); ra=$(count_rows "$AFTER")
kb=$(count_kernels "$BEFORE"); ka=$(count_kernels "$AFTER")
cb=$(count_cvt "$BEFORE"); ca=$(count_cvt "$AFTER")
fb=$(ir_fingerprint "$BEFORE") || fb=""
fa=$(ir_fingerprint "$AFTER") || fa=""
if (( rc_before != 0 || rc_after != 0 )) || [[ "${rb:-0}" == 0 || "${ra:-0}" == 0 ]]; then
verdict=NO_ROWS
if (( rc_before == 124 || rc_after == 124 || rc_before == 137 || rc_after == 137 )); then
note="超过 CELL_TIMEOUT=${CELL_TIMEOUT}s 被截断（rc=$rc_before/$rc_after）。这是耗时问题，不是对开关的判定——先看它为什么这么慢"
else
note="跑不出测量行（rc=$rc_before/$rc_after）。看 run.log——多半是 dtype 不被该算子接受，或它在本后端拿不到 config"
fi
elif [[ -z "$fb" || -z "$fa" ]]; then
verdict=NO_TRITON
note="跑出了行但没编出 .ttgir：本次运行没有任何东西经过 Triton。算子本身可能完全正常（多半跑的是 aten 或厂商库），只是被测的编译器 pass 没有作用对象。不是'开关无作用'，也不是'没有 kernel'"
elif [[ "$fb" == "$fa" ]]; then
verdict=NO_IR_DIFF
note="两侧 ttgir 逐字节相同（config 已锁死）：开关没改变生成代码。先查开关名拼写，再下结论"
else
verdict=IR_CHANGED
note="生成代码有变（cvt ${cb}→${ca}，指纹 ${fb:0:8}/${fa:0:8}）：值得排六轮 A/B"
fi
# 口径从 batch_summary.py 取，那里是 CALIBER_FIELDS 的唯一定义。两份不同
# --level 的 screen.tsv 放一起比时，这几列是唯一能拦住的东西。
cal=$(python batch_summary.py --caliber "$BEFORE" 2>/dev/null)
printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
"$op" "$op_file" "${dtypes:-<all>}" "${rb:-0}" "${ra:-0}" \
"$kb" "$ka" "${cb:-0}" "${ca:-0}" "$el_before" "$el_after" "$verdict" \
"$cal" >> "$TSV"
echo ">>> $op: $verdict — $note"
# 判定说这一对不可比，就别在下一行给出一个看起来像结论的区间
case "$verdict" in
NO_ROWS)
echo "!!! $op 没有可对比的测量，不出 speedup 区间"; continue ;;
NO_TRITON)
echo "!!! $op 没有经过 Triton，speedup 区间与被测开关无关，不出"; continue ;;
NO_IR_DIFF)
echo ">>> $op 两侧生成代码相同：下面这个区间是本机对该算子的噪声底，"
echo ">>> 不是开关的收益。它离零多远，就是该算子单次测量的可信下限。" ;;
esac
echo "########## $op | summary ##########"
python batch_summary.py --ab "$BEFORE/run.log" "$AFTER/run.log" --label "$op" \
|| echo "!!! summary failed: $op"
done
echo "########## 筛选汇总 ##########"
awk -F'\t' 'NR>1{printf " %-12s %-22s cvt %s->%s kernels %s/%s rows %s/%s %ss\n", \
$12,$1,$8,$9,$6,$7,$4,$5,$10+$11}' "$TSV" | sort
echo ">>> 只把 IR_CHANGED 的算子送进 ab_repeat.sh。"
echo "########## ALL DONE ##########"
} 2>&1 | tee "$LOG"
@@ -65,4 +194,5 @@ mkdir -p runs
sed -i -E $'s/\x1b\\[[0-9;]*[A-Za-z]//g' "$LOG"
echo ">>> 总 log: $(pwd)/$LOG"
echo ">>> 每算子两份 run.log 在 runs/<op>_<时间戳>/ 下，按时间先后即 before / after"
echo ">>> 筛选表: $(pwd)/$TSV"
echo ">>> 每算子两份 run.log 在 $ROOT/<op>_{before,after}/ 下"
backends/base.py
+28
View File
@@ -72,6 +72,34 @@ class DeviceBackend:
# backend. Skipped up front so one of them cannot cost a whole run.
capture_deny_ops: Tuple[str, ...] = ()
# One-shot device telemetry sampler. Empty means this backend has none, and
# the campaign runs without a clock/thermal record rather than pretending.
# "{dev}" is dropped from the argv when no device is pinned, so the sampler
# covers every visible device instead of guessing an index.
telemetry_argv: Tuple[str, ...] = ()
def telemetry_command(self, device: str = "") -> List[str]:
if not self.telemetry_argv:
return []
out: List[str] = []
for a in self.telemetry_argv:
if "{dev}" in a:
if not device:
# Drop the flag the placeholder belonged to as well; a bare
# trailing "-i" makes the sampler fail instead of widening.
if out and out[-1].startswith("-"):
out.pop()
continue
a = a.replace("{dev}", device)
out.append(a)
return out
def parse_telemetry(self, raw: str) -> List[Dict[str, str]]:
"""One dict per device: index, sm_clock_mhz, mem_clock_mhz, temp_c,
power_w, util_pct. Normalised so a CUDA and a MUSA campaign produce the
same columns and can be read side by side."""
return []
def classify_capture_error(self, exc: BaseException) -> str:
"""-> "fatal" | "poisoned" | "recoverable".
backends/cuda.py
+21
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
from typing import Dict, List
from .base import DeviceBackend
@@ -26,3 +28,22 @@ class CudaBackend(DeviceBackend):
# No CUDA wording is known to poison the context without also matching one
# of the fatal markers above, so nothing is listed here rather than guessed.
poisoning_error_markers = ()
telemetry_argv = (
"nvidia-smi",
"--query-gpu=index,clocks.sm,clocks.mem,temperature.gpu,power.draw,utilization.gpu",
"--format=csv,noheader,nounits",
"-i", "{dev}",
)
def parse_telemetry(self, raw: str) -> List[Dict[str, str]]:
out: List[Dict[str, str]] = []
for line in raw.splitlines():
parts = [p.strip() for p in line.split(",")]
if len(parts) != 6 or not parts[0].isdigit():
continue
out.append(dict(zip(
("index", "sm_clock_mhz", "mem_clock_mhz",
"temp_c", "power_w", "util_pct"), parts)))
return out
backends/musa.py
+49
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
from typing import Dict, List
from .base import DeviceBackend
@@ -50,3 +52,50 @@ class MusaBackend(DeviceBackend):
# Device count comes from torch.musa or MUSA_VISIBLE_DEVICES. Do not guess
# a vendor CLI here; confirm the real listing tool before adding it.
list_command = None
# mthreads-gmi has no CSV query mode; --json is the only machine-readable
# form it offers. Values arrive with their units glued on ("36C", "1888MHz",
# "477.68W", "87%"), so the parser strips trailing non-numerics rather than
# trusting a fixed suffix.
telemetry_argv = (
"mthreads-gmi", "-q",
"-d", "TEMPERATURE,CLOCK,POWER,UTILIZATION",
"--json",
"-i", "{dev}",
)
def parse_telemetry(self, raw: str) -> List[Dict[str, str]]:
import json
import re
def num(v):
m = re.match(r"^\s*(-?[\d.]+)", str(v or ""))
return m.group(1) if m else ""
start = raw.find("{")
if start < 0:
return []
try:
data = json.loads(raw[start:])
except Exception:
return []
gpus = data.get("GPU") or []
if isinstance(gpus, dict):
gpus = [gpus]
out: List[Dict[str, str]] = []
for g in gpus:
clocks = g.get("Clocks") or {}
power = g.get("Power Readings") or {}
# 键名带尾随空格（"Power Draw "），别按字面写死
draw = next((v for k, v in power.items()
if k.strip() == "Power Draw"), "")
out.append({
"index": str(g.get("Index", "")).strip(),
"sm_clock_mhz": num(clocks.get("Graphics")),
"mem_clock_mhz": num(clocks.get("Memory")),
"temp_c": num((g.get("Temperature") or {}).get("GPU Current Temp")),
"power_w": num(draw),
"util_pct": num((g.get("Utilization") or {}).get("Gpu")),
})
return out
batch_summary.py
+253 -4
View File
@@ -16,8 +16,10 @@ from __future__ import annotations
import argparse
import csv
import itertools
import os
import re
import statistics
import sys
from pathlib import Path
from typing import Dict, List, Optional, Tuple
@@ -220,8 +222,16 @@ def ab_rows(before_text: str, after_text: str) -> List[Dict[str, str]]:
return paired
def report_ab(before: Path, after: Path, label: str = "") -> List[Dict[str, str]]:
"""打印一个算子的前后对比：逐行 speedup 与整体区间。"""
def report_ab(before: Path, after: Path, label: str = "",
strict: bool = True) -> List[Dict[str, str]]:
"""打印一个算子的前后对比：逐行 speedup 与整体区间。
先核对两侧口径；不一致时默认拒绝出报告，因为不可比的两次运行会产出
一份和有效报告长得一模一样的表。
"""
if not _report_caliber([("before", before.parent), ("after", after.parent)],
label or "ab", strict):
raise CaliberMismatch()
before_text = before.read_text(encoding="utf-8", errors="replace")
after_text = after.read_text(encoding="utf-8", errors="replace")
rows = ab_rows(before_text, after_text)
@@ -244,6 +254,225 @@ def report_ab(before: Path, after: Path, label: str = "") -> List[Dict[str, str]
return rows
# Fields whose disagreement makes two runs incomparable. Measured magnitudes,
# so the list is not precautionary: an unpinned autotune config moved
# router_gemm by 8pp; the timer moved replication_pad3d from +43% to +124.5%;
# a shared tuner cache made mm look like 945 compiled variants had become 33;
# --level core sees 5 mm shapes where comprehensive sees 29.
CALIBER_FIELDS = (
"backend", "device", "timer", "bench_level", "shape_mode",
"tuner_cache", "direct_repeat",
)
def caliber_of(run_dir: Path) -> Optional[Dict[str, str]]:
"""Caliber fields from a run's run_identity.json, or None if unreadable."""
if not (run_dir / "run_identity.json").is_file():
return None
try:
from _run_identity import load_identity
ident = load_identity(str(run_dir))
except Exception:
return None
return {k: str(ident.get(k, "")) for k in CALIBER_FIELDS} | {
"autotune_mode": str(ident.get("autotune_mode", ""))
}
def caliber_complaints(runs: List[Tuple[str, Path]]) -> List[str]:
"""Everything about this set of runs that makes their comparison unsound.
Reported rather than silently tolerated because the failure is invisible in
the numbers: two runs on different calibers produce a report that looks
exactly like a valid one. This repository has already published three
figures that were an artifact of one such difference.
"""
out: List[str] = []
cals = [(name, caliber_of(d)) for name, d in runs]
missing = [n for n, c in cals if c is None]
if missing:
out.append(f"口径不可核实：{', '.join(missing)} 缺 run_identity.json"
"（早于该字段的产物）")
known = [(n, c) for n, c in cals if c is not None]
if len(known) < 2:
return out
for field in CALIBER_FIELDS:
seen = {c[field] for _, c in known}
if len(seen) > 1:
vals = ", ".join(f"{n}={c[field] or '<空>'}" for n, c in known)
out.append(f"{field} 不一致：{vals}")
# A/B needs one side to pin the configs the other replays. Two record runs
# each sweep independently, and the winner can differ; the README puts that
# at the 10% level and one-directional per shape, which is enough to swamp
# what is being measured.
modes = [(n, c["autotune_mode"]) for n, c in known]
if all(m.startswith("record") for _, m in modes):
out.append("两侧都是 record 模式：config 未锁定，各自 sweep 可能选出不同 "
"winner。按 README，只有 record 数据支撑的结论按未验证处理")
return out
def caliber_hazards(runs: List[Tuple[str, Path]]) -> List[str]:
"""Not mismatches — shared state that makes a matched caliber less solid.
Warned about rather than refused, because refusing would reject the
default configuration every A/B in this repo currently uses.
"""
cals = [(n, c) for n, c in ((n, caliber_of(d)) for n, d in runs) if c]
recording = [n for n, c in cals if c["autotune_mode"].startswith("record")]
shared = [n for n, c in cals if c["tuner_cache"].startswith("shared")]
if recording and shared:
# 一次 record 跑在共享 libtuner cache 上，sweep 的起点是别的运行留下的
# 结果。曾把 mm 的 945 个变体读成 33 个，误报为编译期收益。A/B 两侧都
# replay 同一份 config 时结论仍成立，绝对值不成立。
return [f"{', '.join(recording)} 在共享 tuner cache 上 record："
"被 pin 的 config 取决于此前跑过什么。相对结论不受影响，"
"绝对延迟/变体数不可跨机器引用（ZL_ISOLATE_TUNER_CACHE=1 可消除）"]
return []
class CaliberMismatch(SystemExit):
"""Strict-mode refusal. Exit 4 so a driver script can tell it from a crash."""
def __init__(self) -> None:
super().__init__(4)
def _report_caliber(runs: List[Tuple[str, Path]], label: str, strict: bool) -> bool:
"""Print caliber problems. Returns False when strict mode should stop."""
for h in caliber_hazards(runs):
print(f"警告 [caliber] {label}: {h}")
problems = caliber_complaints(runs)
if not problems:
return True
head = "!!!" if strict else "警告"
for p in problems:
print(f"{head} [caliber] {label}: {p}")
if strict:
print(f"!!! [caliber] {label}: 拒绝出报告。确认口径确实可比后加 "
"--allow-caliber-mismatch")
return False
return True
def _lat_index(text: str) -> Dict[Tuple[str, str, int], float]:
"""(dtype, size_detail, occurrence) -> gems latency, SUCCESS rows only.
Occurrence disambiguates a shape a benchmark measures twice in one table,
which BLAS does at --level comprehensive.
"""
out: Dict[Tuple[str, str, int], float] = {}
seen: Dict[Tuple[str, str], int] = {}
for row in parse_run_log(text):
if row.get("row_status") != "SUCCESS":
continue
try:
lat = float(row.get("gems_lat", ""))
except ValueError:
continue
if lat <= 0:
continue
base = (row.get("dtype", ""), row.get("size_detail", ""))
i = seen.get(base, 0)
seen[base] = i + 1
out[base + (i,)] = lat
return out
def ab_round_rows(root: Path) -> Tuple[List[Dict[str, str]], List[float], int]:
"""Aggregate an N-round A/B laid out as <root>/r<k>_<A|B>/run.log.
Returns the per-shape rows, the per-round deltas, and the round count.
A single pair cannot say whether its own verdict is stable, which is the
question that matters once a delta lands near zero: measured on an S5000,
topk eliminated 32 converts yet 17 of its 42 shapes changed the SIGN of
their delta between two equally valid runs. Two dispersions answer different
halves of that and both are reported:
side spread max/min-1 within one side's samples -> how noisy the machine
was while measuring
round spread min~max of the per-round deltas -> whether the VERDICT holds,
which is what a single pair cannot show
"""
rounds: List[Tuple[Dict, Dict]] = []
for k in itertools.count(1):
a, b = root / f"r{k}_A" / "run.log", root / f"r{k}_B" / "run.log"
if not (a.is_file() and b.is_file()):
break
rounds.append((
_lat_index(a.read_text(encoding="utf-8", errors="replace")),
_lat_index(b.read_text(encoding="utf-8", errors="replace")),
))
if not rounds:
return [], [], 0
keys = set(rounds[0][0]) & set(rounds[0][1])
for a, b in rounds[1:]:
keys &= set(a) & set(b)
per_round = [
statistics.median((a[k] / b[k] - 1.0) * 100.0 for k in keys)
for a, b in rounds
] if keys else []
rows: List[Dict[str, str]] = []
for key in sorted(keys):
av = [a[key] for a, _ in rounds]
bv = [b[key] for _, b in rounds]
am, bm = statistics.median(av), statistics.median(bv)
deltas = [(a / b - 1.0) * 100.0 for a, b in zip(av, bv)]
rows.append({
"dtype": key[0], "size_detail": key[1],
"before_lat": f"{am:.6f}", "after_lat": f"{bm:.6f}",
"speedup_percent": f"{(am / bm - 1.0) * 100.0:.2f}",
"delta_min": f"{min(deltas):.2f}", "delta_max": f"{max(deltas):.2f}",
"sign_stable": "yes" if all(d > 0 for d in deltas)
or all(d <= 0 for d in deltas) else "NO",
"before_spread": f"{(max(av) / min(av) - 1.0) * 100.0:.2f}",
"after_spread": f"{(max(bv) / min(bv) - 1.0) * 100.0:.2f}",
})
return rows, per_round, len(rounds)
def report_ab_rounds(root: Path, label: str = "",
strict: bool = True) -> List[Dict[str, str]]:
"""Print the multi-round A/B: per-shape medians plus both dispersions."""
name = label or root.name
cells = sorted(d for d in root.glob("r*_[AB]") if (d / "run.log").is_file())
if cells and not _report_caliber([(d.name, d) for d in cells], name, strict):
raise CaliberMismatch()
rows, per_round, n = ab_round_rows(root)
if n < 2:
print(f"[ab-rounds] {name}: 需要至少 2 轮 <root>/r<k>_<A|B>/run.log"
f"（找到 {n} 轮）")
return rows
if not rows:
print(f"[ab-rounds] {name}: {n} 轮，但没有全轮次都成功的配对行")
return rows
for r in rows:
flag = "" if r["sign_stable"] == "yes" else " 符号不稳"
print(f" {r['dtype']:<16} {float(r['speedup_percent']):+7.2f}% "
f"[{float(r['delta_min']):+.2f}%~{float(r['delta_max']):+.2f}%] "
f"{r['before_lat']} -> {r['after_lat']} "
f"{r['size_detail']}{flag}")
med = [float(r["speedup_percent"]) for r in rows]
spread = [float(r["before_spread"]) for r in rows] + \
[float(r["after_spread"]) for r in rows]
unstable = sum(1 for r in rows if r["sign_stable"] != "yes")
print(f"[ab-rounds] {name}: {n} 轮 × {len(rows)} shape，"
f"speedup 中位 {statistics.median(med):+.2f}%，"
f"逐轮 {min(per_round):+.2f}% ~ {max(per_round):+.2f}%，"
f"单侧噪声中位 {statistics.median(spread):.2f}%")
if unstable:
# The verdict, not the machine, is what these rows fail to pin down.
print(f"[ab-rounds] {name}: {unstable}/{len(rows)} 个 shape 的 delta "
"在轮次间改变符号——这些行的方向不可结论，与计时口径无关")
return rows
def main() -> None:
p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("batch_dir", type=Path, nargs="?",
@@ -252,13 +481,33 @@ def main() -> None:
help="输出 CSV（默认 <batch_dir>/summary.csv）")
p.add_argument("--ab", nargs=2, type=Path, metavar=("BEFORE", "AFTER"),
help="两份 run.log，按 shape 配对输出该算子的 speedup 区间")
p.add_argument("--label", default="", help="--ab 报告里的算子名")
p.add_argument("--ab-rounds", type=Path, default=None, metavar="ROOT",
help="多轮 A/B 目录，含 r<k>_A/run.log 与 r<k>_B/run.log；"
"逐 shape 取轮次中位，并报逐轮区间与符号稳定性")
p.add_argument("--label", default="", help="--ab / --ab-rounds 报告里的算子名")
p.add_argument("--caliber", type=Path, default=None, metavar="RUN_DIR",
help="打印该 run 的口径字段（TSV 一行），供筛选表复用")
p.add_argument("--allow-caliber-mismatch", action="store_true",
help="口径不一致时仍出报告（默认拒绝）。只在确认两次运行"
"确实可比时使用")
args = p.parse_args()
if args.caliber is not None:
# 口径的定义只有 CALIBER_FIELDS 一处；筛选表等复用它，不要在别处重列
cal = caliber_of(args.caliber) or {}
print(" ".join(str(cal.get(k, "")) for k in CALIBER_FIELDS))
return
if args.ab_rounds:
if not args.ab_rounds.is_dir():
raise SystemExit(f"目录不存在: {args.ab_rounds}")
report_ab_rounds(args.ab_rounds, args.label,
strict=not args.allow_caliber_mismatch)
return
if args.ab:
for path in args.ab:
if not path.is_file():
raise SystemExit(f"run.log 不存在: {path}")
report_ab(args.ab[0], args.ab[1], args.label)
report_ab(args.ab[0], args.ab[1], args.label,
strict=not args.allow_caliber_mismatch)
return
if args.batch_dir is None:
raise SystemExit("需要 batch_dir，或用 --ab BEFORE AFTER")
campaign.sh
+128
View File
@@ -0,0 +1,128 @@
#!/bin/bash
# 多卡跑一批算子的六轮 A/B：每卡一个 ab_repeat.sh，卡内串行。
#
# 它只做编排，不做测量。ab_repeat.sh 一行不动——那是已经跑出结论的、验证过的串行
# 路径，不该为了并行去改它。分片交给 partition.py（按实测耗时贪心，比 hash 分片在
# 这批活上快 50 分钟）。
#
# 并行必须隔离 tuner cache：8 个进程写同一个 sqlite 是要避开的事。代价是 pin 下来
# 的 config 与共享缓存下跑的批次不同，两批的绝对延迟不可互引——batch_summary.py
# 的口径守卫会自己拒绝这种对比，不用记着。
#
# 用法：
# SWITCH=FLAGTREE_MUSA_RLC_ENHANCE DEVICES=0,1,2,3,4,5,6,7 \
# SCREEN=runs/ab_rlc_XXX/screen.tsv \
# ZL_BACKEND=musa FLAGGEMS_DIR=/path/to/FlagGems bash campaign.sh
#
# DRY_RUN=1 只打印分片方案，不起跑
# EXTRA_OPS=".." 追加筛选表里没有的算子（如 benchmark 刚修好的），塞给最空的卡
# ROUNDS=6 CELL_TIMEOUT=2400 VERDICT=IR_CHANGED
set -uo pipefail
cd "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
: "${SWITCH:?SWITCH is required, e.g. FLAGTREE_MUSA_RLC_ENHANCE}"
: "${SCREEN:?SCREEN is required, 指向 ab_rlc_test.sh 的 screen.tsv}"
DEVICES="${DEVICES:-0}"
ROUNDS="${ROUNDS:-6}"
CELL_TIMEOUT="${CELL_TIMEOUT:-2400}"
VERDICT="${VERDICT:-IR_CHANGED}"
EXTRA_OPS="${EXTRA_OPS:-}"
DRY_RUN="${DRY_RUN:-0}"
TAB=$'\t'
[[ -f "$SCREEN" ]] || { echo "!!! 筛选表不存在: $SCREEN" >&2; exit 2; }
# 设备可见性变量名从 backends 层拿，不写死 MUSA_VISIBLE_DEVICES
VIS_KEY=$(python -c "import sys; sys.path.insert(0, '.')
from backends import get_backend
print(get_backend().visibility_env_keys[0])" 2>/dev/null) || VIS_KEY=""
[[ -n "$VIS_KEY" ]] || { echo "!!! 取不到后端的可见性变量名，检查 ZL_BACKEND" >&2; exit 2; }
echo ">>> switch=$SWITCH rounds=$ROUNDS devices=$DEVICES 可见性变量=$VIS_KEY"
python partition.py --screen "$SCREEN" --devices "$DEVICES" \
--verdict "$VERDICT" --rounds "$ROUNDS" || exit 2
# 每卡一行 "<device><TAB><OPS><TAB><权重秒>"
mapfile -t PLAN < <(python partition.py --screen "$SCREEN" --devices "$DEVICES" \
--verdict "$VERDICT" --format shell) || exit 2
if [[ -n "$EXTRA_OPS" ]]; then
# 追加到当前负载最小的卡。按算子个数选是错的——一个重算子能顶三个轻的，
# partition.py 的第三列给的就是权重秒数
lightest=0; least=""
for i in "${!PLAN[@]}"; do
IFS="$TAB" read -r _d _o _w <<< "${PLAN[$i]}"
if [[ -z "$least" ]] || (( ${_w:-0} < least )); then least=${_w:-0}; lightest=$i; fi
done
IFS="$TAB" read -r d o w <<< "${PLAN[$lightest]}"
PLAN[$lightest]="${d}${TAB}${o:+$o }${EXTRA_OPS}${TAB}${w}"
echo ">>> EXTRA_OPS 追加到卡${d}（当前最空，$(( ${w:-0} * ROUNDS / 60 )) 分钟）: $EXTRA_OPS"
fi
if [[ "$DRY_RUN" == "1" ]]; then
echo ">>> DRY_RUN=1，不起跑。逐卡将执行："
for row in "${PLAN[@]}"; do
IFS="$TAB" read -r d o w <<< "$row"
printf ' 卡%s %4s 分钟 OPS="%s"\n' "$d" "$(( ${w:-0} * ROUNDS / 60 ))" "$o"
done
exit 0
fi
ROOT="runs/campaign_${SWITCH}_$(date +%Y%m%d-%H%M%S)"
mkdir -p "$ROOT"
printf '%s\n' "${PLAN[@]}" > "$ROOT/plan.tsv"
echo ">>> 产物: $(pwd)/$ROOT"
declare -a PIDS=() CARDS=()
for row in "${PLAN[@]}"; do
IFS="$TAB" read -r dev ops _w <<< "$row"
[[ -n "$ops" ]] || { echo ">>> 卡$dev 没有分到活，跳过"; continue; }
out="$ROOT/card${dev}"
mkdir -p "$out"
# AB_ROOT 让每卡有独立产物根：八个 ab_repeat.sh 同一秒启动，默认的时间戳目录名会撞
env "$VIS_KEY=$dev" ZL_DEVICE="$dev" ZL_ISOLATE_TUNER_CACHE=1 \
AB_ROOT="$out" SWITCH="$SWITCH" ROUNDS="$ROUNDS" \
CELL_TIMEOUT="$CELL_TIMEOUT" OPS="$ops" NO_COLOR=1 \
bash ab_repeat.sh > "$ROOT/card${dev}.log" 2>&1 &
PIDS+=($!); CARDS+=("$dev")
echo ">>> 卡$dev pid=$! ops=$ops"
done
(( ${#PIDS[@]} )) || { echo "!!! 没有一张卡分到活" >&2; exit 2; }
echo ">>> ${#PIDS[@]} 张卡已启动，等待全部结束……"
fail=0
for i in "${!PIDS[@]}"; do
if wait "${PIDS[$i]}"; then
echo ">>> 卡${CARDS[$i]} 结束 rc=0"
else
rc=$?; fail=$((fail+1))
echo "!!! 卡${CARDS[$i]} 结束 rc=$rc —— 看 $ROOT/card${CARDS[$i]}.log"
fi
done
# 合并逐格记录：卡号加在最前面，便于事后按卡追
{
first=1
for d in "${CARDS[@]}"; do
f="$ROOT/card${d}/status.tsv"
[[ -f "$f" ]] || continue
if (( first )); then printf 'device%s%s\n' "$TAB" "$(head -1 "$f")"; first=0; fi
tail -n +2 "$f" | sed "s/^/${d}${TAB}/"
done
} > "$ROOT/status.tsv"
echo
echo "########## 汇总 ##########"
echo ">>> 失败的卡: $fail / ${#PIDS[@]}"
if [[ -s "$ROOT/status.tsv" ]]; then
echo ">>> 逐格记录: $(pwd)/$ROOT/status.tsv ($(( $(wc -l < "$ROOT/status.tsv") - 1 )) 格)"
awk -F"$TAB" 'NR>1 && ($6+0)!=0 {n++} END{printf ">>> 非零退出的格: %d\n", n+0}' "$ROOT/status.tsv"
awk -F"$TAB" 'NR>1 {fb+=$10; ng+=$11; po+=$12}
END{printf ">>> 三个标记合计: replay_fallback=%d no_cudagraph=%d capture_poisoned=%d\n",
fb+0, ng+0, po+0}' "$ROOT/status.tsv"
fi
echo ">>> 逐算子报告在各卡的 summary.log 里；单独重出："
echo " python batch_summary.py --ab-rounds $ROOT/card<N>/<op> --label <op>"
echo ">>> 注意：本批 ZL_ISOLATE_TUNER_CACHE=1，绝对延迟不可与共享缓存的批次互引"
partition.py
+145
View File
@@ -0,0 +1,145 @@
#!/usr/bin/env python3
"""把筛选通过的算子按实测耗时分到各张卡上。
为什么不用 `run_batch.py` 的 hash 分片：它按算子名的 hash 取模，不看重量。S5000 上
17 个算子的六轮 campaign，hash 分片关键路径 120 分钟、有一张卡空转；按耗时贪心是
70 分钟。同一批活，差 50 分钟。
贪心（最长优先塞进当前最空的卡）在这里够用，也不可能更好：`topk` 一个算子就要 70.6
分钟，而理想均分是 45.4——**关键路径的下界由最重的那个算子决定，加卡不解决**。分片
只能保证除它之外的卡不空等。
权重来自 `screen.tsv` 的 `elapsed_before + elapsed_after`（**一对 = 2 格**的秒数）。
N 轮是 2N 格，所以预估耗时是权重的 N 倍——不是 N/6 倍，这里错过一次，把 topk 的 70.6
分钟报成了 11.8。缺列时退化成等权重，并在 stderr 说明——等权重的分片仍然
比不分好，但别让人以为它按耗时排过。
用法：
python partition.py --screen runs/ab_rlc_XXX/screen.tsv --devices 0,1,2,3
python partition.py --screen ... --devices 0,1 --verdict IR_CHANGED --format shell
"""
from __future__ import annotations
import argparse
import csv
import sys
from pathlib import Path
from typing import Dict, List, Optional, Sequence, Tuple
Item = Tuple[str, float]
def plan(items: Sequence[Item], slots: Sequence[str]) -> Dict[str, List[Item]]:
"""最长优先贪心：每次把最重的活放进当前总量最小的卡。
卡少于 1 张是调用方的错，直接抛；活比卡少时后面的卡就是空的，这不是错误。
"""
if not slots:
raise ValueError("需要至少一张卡")
out: Dict[str, List[Item]] = {s: [] for s in slots}
load: Dict[str, float] = {s: 0.0 for s in slots}
# 同重量时按名字排，保证同一份输入每次得到同一个方案
for name, weight in sorted(items, key=lambda kv: (-kv[1], kv[0])):
target = min(slots, key=lambda s: (load[s], s))
out[target].append((name, weight))
load[target] += weight
return out
def imbalance(assignment: Dict[str, List[Item]]) -> float:
"""最慢的卡 / 理想均分。1.0 是完美；下界由最重的单个算子决定。"""
loads = [sum(w for _, w in v) for v in assignment.values()]
total = sum(loads)
if total <= 0:
return 1.0
return max(loads) / (total / len(loads))
def read_screen(path: Path, verdict: str = "IR_CHANGED"
) -> Tuple[List[Item], Dict[str, str], List[str]]:
"""(算子, 一对耗时秒), op -> op_file, 被判定挡掉的算子。"""
with path.open(newline="", encoding="utf-8") as f:
rows = list(csv.DictReader(f, delimiter="\t"))
if not rows:
raise SystemExit(f"筛选表没有数据行: {path}")
items: List[Item] = []
files: Dict[str, str] = {}
skipped: List[str] = []
warned = False
for r in rows:
op = r.get("op", "")
if not op:
continue
if verdict and r.get("verdict") != verdict:
skipped.append(f"{op}({r.get('verdict', '?')})")
continue
try:
w = float(r["elapsed_before"]) + float(r["elapsed_after"])
except (KeyError, TypeError, ValueError):
w = 1.0
if not warned:
print("!!! 筛选表没有 elapsed_* 列，退化成等权重分片——"
"分片仍然比不分好，但它没有按耗时排过", file=sys.stderr)
warned = True
items.append((op, w))
files[op] = r.get("op_file") or op
return items, files, skipped
def main(argv: Optional[List[str]] = None) -> int:
p = argparse.ArgumentParser(description=__doc__.splitlines()[0],
formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--screen", type=Path, required=True, help="ab_rlc_test.sh 的 screen.tsv")
p.add_argument("--devices", required=True, help="逗号分隔的设备列表，如 0,1,2,3")
p.add_argument("--verdict", default="IR_CHANGED",
help="只取这个判定的算子；空字符串表示全要")
# 权重是一对（2 格）的秒数；N 轮是 2N 格，所以预估耗时是权重的 N 倍
p.add_argument("--rounds", type=int, default=6, help="仅用于打印预估耗时")
p.add_argument("--format", choices=("human", "shell"), default="human",
help="shell 输出每卡一行 `<device>\\t<OPS>`，供驱动脚本读")
args = p.parse_args(argv)
if not args.screen.is_file():
raise SystemExit(f"筛选表不存在: {args.screen}")
devices = [d.strip() for d in args.devices.split(",") if d.strip()]
if not devices:
raise SystemExit("--devices 为空")
items, files, skipped = read_screen(args.screen, args.verdict)
if not items:
raise SystemExit(f"筛选表里没有 verdict={args.verdict} 的算子——"
"先确认筛选跑过，且判据没有把所有算子都挡掉")
assignment = plan(items, devices)
if args.format == "shell":
# 第三列是该卡的权重秒数，供驱动脚本决定往哪张卡追加算子——按负载选，不是
# 按算子个数：一个重算子能顶三个轻的。
# 显式写 "\n" 而不用 print：Windows 上 print 会发 CRLF，读的一侧会把 \r
# 带进最后一个字段，数值比较就废了。
for d in devices:
ops = " ".join(f"{op}:{files[op]}" for op, _ in assignment[d])
load = sum(w for _, w in assignment[d])
sys.stdout.buffer.write(f"{d}\t{ops}\t{load:.0f}\n".encode())
sys.stdout.buffer.flush()
return 0
total = sum(w for _, w in items) * args.rounds
print(f">>> {len(items)} 个算子 / {len(devices)} 张卡 / {args.rounds} 轮")
if skipped:
print(f">>> 判定挡掉 {len(skipped)} 个: {', '.join(skipped)}")
for d in devices:
load = sum(w for _, w in assignment[d]) * args.rounds
names = " ".join(op for op, _ in assignment[d]) or "（空）"
print(f" 卡{d} {load/60:6.1f} 分钟 {names}")
crit = max(sum(w for _, w in v) for v in assignment.values()) * args.rounds
heaviest = max(items, key=lambda kv: kv[1])
print(f">>> 单卡串行 {total/3600:.2f} 小时；{len(devices)} 卡关键路径 "
f"{crit/60:.1f} 分钟，失衡 {imbalance(assignment):.2f}x")
print(f">>> 下界由最重的算子决定：{heaviest[0]} 单独就要 "
f"{heaviest[1]*args.rounds/60:.1f} 分钟，加卡不解决")
return 0
if __name__ == "__main__":
raise SystemExit(main())
run_pytest.sh
+6
View File
@@ -114,6 +114,12 @@ preflight "FlagGems directory exists: $FLAGGEMS_DIR" test -d "$FLAGGEMS_DIR"
# The node's file half; the ::test_<fn> half is left to pytest, which reports a
# missing function clearly ("not found: ...") while a missing FILE does not.
preflight "benchmark file exists: ${TEST_FILE%%::*}" test -f "${TEST_FILE%%::*}"
# The interpreter, not just the paths. triton and flag_gems normally come from a
# PYTHONPATH pointing at a compiler build and a source checkout, so a shell that
# merely reaches the right files can still be the wrong Python. Without this the
# failure surfaces per cell as `pytest exit 4` from a conftest ImportError, and a
# multi-round campaign burns every cell before anyone reads a log.
preflight "interpreter imports triton + flag_gems ($(command -v python))" sh -c 'python -c "import triton, flag_gems" 2>/dev/null'
if [[ -n "$SHAPE_FILE" ]]; then
preflight "shape file exists: $SHAPE_FILE" test -f "$SHAPE_FILE"
fi
telemetry.py
+110
View File
@@ -0,0 +1,110 @@
#!/usr/bin/env python3
"""采样设备时钟/温度/功耗，跨后端归一化成同一组列。
为什么要有它：六轮 A/B 会声称单侧噪声在 0.01% 量级。降频和热漂移足以移动比这大
得多的延迟，而且在 run.log 里不留任何痕迹——报告照样干净。NV 上复现 PR763 的那次
全程 `nvidia-smi -l 5` 写 clocks.csv，S5000 这边一直没有对应的东西。
输出每设备每次采样一行 TSV：
timestamp index sm_clock_mhz mem_clock_mhz temp_c power_w util_pct
列在 cuda 与 musa 上完全一致，所以两边的 campaign 可以并排读。采样器不存在时安静
退出：拿不到遥测是缺一份证据，不是让整批测试失败的理由。
用法：
python telemetry.py --backend musa --device 7 --interval 5 --run > telemetry.tsv
python telemetry.py --backend cuda # 只打印将要执行的采样命令
"""
from __future__ import annotations
import argparse
import os
import subprocess
import sys
import time
from typing import List, Optional
FIELDS = ("index", "sm_clock_mhz", "mem_clock_mhz", "temp_c", "power_w", "util_pct")
def resolve_device(backend, explicit: Optional[str]) -> str:
"""显式设备优先，否则从后端自己的可见性变量取。
单值才用得上：`MUSA_VISIBLE_DEVICES=7` 指一张卡，`0,1,2` 指一组，后者不传
`-i`、采全部，比挑第一张更诚实。
"""
if explicit:
return explicit.strip()
for key in getattr(backend, "visibility_env_keys", ()):
raw = os.environ.get(key, "").strip()
if raw and "," not in raw:
return raw
return ""
def sample_once(backend, argv: List[str]) -> List[dict]:
try:
out = subprocess.run(argv, capture_output=True, text=True, timeout=30)
except (OSError, subprocess.SubprocessError):
return []
return backend.parse_telemetry(out.stdout or "")
def run(backend, device: str, interval: float, out=sys.stdout) -> int:
argv = backend.telemetry_command(device)
if not argv:
print(f"[telemetry] backend={backend.name} 没有采样器；本批不记时钟/温度",
file=sys.stderr, flush=True)
return 0
if not sample_once(backend, argv):
# 第一次就取不到，说明工具缺失或没权限。不要留一个空文件让人以为采过了
print(f"[telemetry] {argv[0]} 取不到读数；本批不记时钟/温度",
file=sys.stderr, flush=True)
return 0
print("timestamp\t" + "\t".join(FIELDS), file=out, flush=True)
while True:
ts = time.strftime("%Y-%m-%dT%H:%M:%S")
for row in sample_once(backend, argv):
print(ts + "\t" + "\t".join(row.get(f, "") for f in FIELDS),
file=out, flush=True)
time.sleep(interval)
def main(argv: Optional[List[str]] = None) -> int:
p = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--backend", default=os.environ.get("ZL_BACKEND", "cuda"))
p.add_argument("--device", default=os.environ.get("ZL_DEVICE", ""))
p.add_argument("--interval", type=float, default=5.0)
p.add_argument("--run", action="store_true",
help="持续采样直到被杀；不给则只打印采样命令")
args = p.parse_args(argv)
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from backends import get_backend
backend = get_backend(args.backend)
device = resolve_device(backend, args.device)
if not args.run:
print(" ".join(backend.telemetry_command(device)))
return 0
try:
return run(backend, device, args.interval)
except KeyboardInterrupt:
return 0
except BrokenPipeError:
# 读端先走（管给 head，或 campaign 结束时收掉）。这是正常终止，不该在
# telemetry.err 里留一段 traceback 让人以为遥测本身坏了。按 CPython 的
# 惯用做法把 stdout 指到 devnull——只吞异常不够，解释器退出时还会 flush
# 一次，再抛一条 "Exception ignored" 到 stderr。
try:
devnull = os.open(os.devnull, os.O_WRONLY)
os.dup2(devnull, sys.stdout.fileno())
except Exception:
pass
return 0
if __name__ == "__main__":
raise SystemExit(main())
test_backends.py
+380
View File
@@ -10,6 +10,7 @@ import json
import os
import sys
import tempfile
import types
import unittest
from pathlib import Path
@@ -279,6 +280,82 @@ class CaptureClassificationTests(unittest.TestCase):
self.assertFalse(musa.denies_capture("unique"))
class DirectRepeatTests(unittest.TestCase):
"""Repetition on the direct timer.
_cudagraph_plugin imports triton at module scope, so it is loaded here
against a stub: the repeat logic is plain arithmetic and should stay
testable on a machine with no GPU and no Triton.
"""
@classmethod
def setUpClass(cls):
cls._saved = {k: sys.modules.get(k) for k in ("triton", "triton.testing")}
triton = types.ModuleType("triton")
testing = types.ModuleType("triton.testing")
testing.do_bench = lambda *a, **k: 0.0
testing.do_bench_cudagraph = lambda *a, **k: 0.0
testing._summarize_statistics = lambda *a, **k: 0.0
triton.testing = testing
sys.modules["triton"] = triton
sys.modules["triton.testing"] = testing
import _cudagraph_plugin # noqa: E402
cls.cg = _cudagraph_plugin
@classmethod
def tearDownClass(cls):
for name, mod in cls._saved.items():
if mod is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = mod
sys.modules.pop("_cudagraph_plugin", None)
def _run(self, value, **kwargs):
"""-> (result, how many times do_bench was called)"""
seq, calls = [3.0, 1.0, 2.0, 9.0, 5.0], []
self.cg._ORIGINAL_DO_BENCH = (
lambda fn, **kw: (calls.append(1), seq[(len(calls) - 1) % len(seq)])[1])
old = os.environ.get("ZL_DIRECT_REPEAT")
os.environ["ZL_DIRECT_REPEAT"] = value
try:
out = self.cg.run_direct(lambda: None, 25, 100, None,
kwargs.get("quantiles"),
kwargs.get("return_mode", "mean"))
finally:
if old is None:
os.environ.pop("ZL_DIRECT_REPEAT", None)
else:
os.environ["ZL_DIRECT_REPEAT"] = old
return out, len(calls)
def test_default_is_one_call(self):
self.assertEqual(self._run("1"), (3.0, 1))
def test_unset_is_one_call(self):
old = os.environ.pop("ZL_DIRECT_REPEAT", None)
try:
self.assertEqual(self.cg.direct_repeat(), 1)
finally:
if old is not None:
os.environ["ZL_DIRECT_REPEAT"] = old
def test_median_over_calls(self):
# 3, 1, 2, 9, 5 -> 3
self.assertEqual(self._run("5"), (3.0, 5))
def test_garbage_and_zero_fall_back_to_one(self):
self.assertEqual(self._run("abc"), (3.0, 1))
self.assertEqual(self._run("0"), (3.0, 1))
self.assertEqual(self._run("-4"), (3.0, 1))
def test_sequence_results_are_not_repeated(self):
# quantiles / return_mode="all" make do_bench return a sequence, and a
# median across calls is undefined for it; take the single call.
self.assertEqual(self._run("5", quantiles=[0.5])[1], 1)
self.assertEqual(self._run("5", return_mode="all")[1], 1)
class DeviceArgTests(unittest.TestCase):
def test_devices_preferred(self):
self.assertEqual(resolve_device_arg("0,1", None), "0,1")
@@ -354,5 +431,308 @@ class AutotuneReplayBackendTests(unittest.TestCase):
self.assertFalse(_replay_backend_ok(tmp))
class CaliberGuardTests(unittest.TestCase):
"""两次运行口径不同产出的报告，和有效报告长得一模一样。
这些用例钉住的是本仓已经踩过的三次：record 对 record、共享 tuner cache、
timer 不同。它们都不会让任何一步报错，只会让数字变成另一个含义。
"""
BASE = {
"backend": "musa", "device": "S5000", "timer": "cuda_graph",
"bench_level": "core", "shape_mode": "default",
"tuner_cache": "isolated", "direct_repeat": 5,
"autotune_mode": "replay",
}
def _run(self, root, name, **over):
d = Path(root, name)
d.mkdir(parents=True)
if over.get("_no_identity"):
return d
ident = dict(self.BASE)
ident.update({k: v for k, v in over.items() if not k.startswith("_")})
Path(d, "run_identity.json").write_text(
json.dumps(ident), encoding="utf-8")
return d
def test_same_caliber_has_no_complaints(self):
from batch_summary import caliber_complaints
with tempfile.TemporaryDirectory() as tmp:
runs = [("A", self._run(tmp, "a")), ("B", self._run(tmp, "b"))]
self.assertEqual(caliber_complaints(runs), [])
def test_record_side_pinning_the_other_is_normal(self):
"""A/B 的正常形态：一侧 record 定 config，另一侧 replay 它。"""
from batch_summary import caliber_complaints
with tempfile.TemporaryDirectory() as tmp:
runs = [("r1_A", self._run(tmp, "a", autotune_mode="record")),
("r1_B", self._run(tmp, "b"))]
self.assertEqual(caliber_complaints(runs), [])
def test_record_against_record_is_flagged(self):
from batch_summary import caliber_complaints
with tempfile.TemporaryDirectory() as tmp:
runs = [("A", self._run(tmp, "a", autotune_mode="record")),
("B", self._run(tmp, "b", autotune_mode="record"))]
self.assertTrue(any("record" in c for c in caliber_complaints(runs)))
def test_each_caliber_field_is_checked(self):
from batch_summary import CALIBER_FIELDS, caliber_complaints
others = {"backend": "cuda", "device": "H20", "timer": "direct",
"bench_level": "comprehensive", "shape_mode": "yaml",
"tuner_cache": "shared", "direct_repeat": 1}
self.assertEqual(set(others), set(CALIBER_FIELDS))
for field, other in others.items():
with tempfile.TemporaryDirectory() as tmp:
runs = [("A", self._run(tmp, "a")),
("B", self._run(tmp, "b", **{field: other}))]
got = caliber_complaints(runs)
self.assertTrue(any(c.startswith(field) for c in got),
f"{field} 不一致时没被拦下: {got}")
def test_missing_identity_is_flagged_not_assumed_equal(self):
from batch_summary import caliber_complaints
with tempfile.TemporaryDirectory() as tmp:
runs = [("A", self._run(tmp, "a")),
("B", self._run(tmp, "b", _no_identity=True))]
self.assertTrue(any("run_identity" in c
for c in caliber_complaints(runs)))
def test_shared_cache_record_warns_but_does_not_block(self):
from batch_summary import caliber_complaints, caliber_hazards
with tempfile.TemporaryDirectory() as tmp:
runs = [("r1_A", self._run(tmp, "a", autotune_mode="record",
tuner_cache="shared(~/.flaggems)")),
("r1_B", self._run(tmp, "b",
tuner_cache="shared(~/.flaggems)"))]
self.assertEqual(caliber_complaints(runs), [])
self.assertTrue(caliber_hazards(runs))
def test_isolated_cache_has_no_hazard(self):
from batch_summary import caliber_hazards
with tempfile.TemporaryDirectory() as tmp:
runs = [("A", self._run(tmp, "a", autotune_mode="record")),
("B", self._run(tmp, "b"))]
self.assertEqual(caliber_hazards(runs), [])
def test_strict_refuses_and_permissive_continues(self):
from batch_summary import _report_caliber
with tempfile.TemporaryDirectory() as tmp:
runs = [("A", self._run(tmp, "a")),
("B", self._run(tmp, "b", timer="direct"))]
self.assertFalse(_report_caliber(runs, "op", strict=True))
self.assertTrue(_report_caliber(runs, "op", strict=False))
LOG = ("Operator: op (dtype=torch.float16, mode=kernel, level=core)\n"
"SUCCESS 1.000 2.000 0.500 shape\n")
def test_report_ab_refuses_mismatched_pair(self):
from batch_summary import CaliberMismatch, report_ab
with tempfile.TemporaryDirectory() as tmp:
a = self._run(tmp, "a")
b = self._run(tmp, "b", timer="direct")
for d in (a, b):
Path(d, "run.log").write_text(self.LOG, encoding="utf-8")
with self.assertRaises(CaliberMismatch):
report_ab(a / "run.log", b / "run.log", "op")
# 放行开关在，但必须是显式的
self.assertEqual(
len(report_ab(a / "run.log", b / "run.log", "op", strict=False)), 1)
class TelemetryTests(unittest.TestCase):
"""采样器命令与解析。两边归一到同一组列，campaign 才能并排读。"""
# 取自 S5000 实机：值带单位，"Power Draw " 的键名有尾随空格
MUSA_JSON = """
{
"Attached GPUs": "8",
"GPU": [
{
"Index": "7",
"Temperature": {"GPU Current Temp": "36C"},
"Clocks": {"Graphics": "1888MHz", "Memory": "2500MHz", "Video": "0MHz"},
"Power Readings": {"Power Draw ": "477.68W", "Current Power Limit": "950.00W"},
"Utilization": {"Gpu": "87%", "Memory": "14%"}
}
]
}
"""
CUDA_CSV = "0, 1980, 2619, 41, 122.35, 63\n"
def test_musa_parses_units_off_values(self):
rows = get_backend("musa").parse_telemetry(self.MUSA_JSON)
self.assertEqual(rows, [{
"index": "7", "sm_clock_mhz": "1888", "mem_clock_mhz": "2500",
"temp_c": "36", "power_w": "477.68", "util_pct": "87"}])
def test_cuda_parses_csv(self):
rows = get_backend("cuda").parse_telemetry(self.CUDA_CSV)
self.assertEqual(rows, [{
"index": "0", "sm_clock_mhz": "1980", "mem_clock_mhz": "2619",
"temp_c": "41", "power_w": "122.35", "util_pct": "63"}])
def test_both_backends_yield_the_same_columns(self):
import telemetry
a = get_backend("musa").parse_telemetry(self.MUSA_JSON)[0]
b = get_backend("cuda").parse_telemetry(self.CUDA_CSV)[0]
self.assertEqual(set(a), set(b))
self.assertEqual(set(a), set(telemetry.FIELDS))
def test_garbage_parses_to_nothing_rather_than_raising(self):
for backend in ("musa", "cuda"):
for raw in ("", "no such tool", "{broken json"):
self.assertEqual(get_backend(backend).parse_telemetry(raw), [])
def test_no_device_drops_the_flag_not_just_the_value(self):
# 留下裸 "-i" 会让采样器直接失败，比采全部更糟
argv = get_backend("musa").telemetry_command("")
self.assertNotIn("-i", argv)
self.assertIn("--json", argv)
self.assertEqual(get_backend("musa").telemetry_command("7")[-2:], ["-i", "7"])
def test_backend_without_sampler_returns_empty(self):
from backends.base import DeviceBackend
self.assertEqual(DeviceBackend().telemetry_command("0"), [])
def test_device_comes_from_the_visibility_var_when_unpinned(self):
import telemetry
backend = get_backend("musa")
old = os.environ.get("MUSA_VISIBLE_DEVICES")
try:
os.environ["MUSA_VISIBLE_DEVICES"] = "7"
self.assertEqual(telemetry.resolve_device(backend, ""), "7")
# 一组卡不挑第一张，交给采样器采全部
os.environ["MUSA_VISIBLE_DEVICES"] = "0,1,2"
self.assertEqual(telemetry.resolve_device(backend, ""), "")
self.assertEqual(telemetry.resolve_device(backend, "3"), "3")
finally:
if old is None:
os.environ.pop("MUSA_VISIBLE_DEVICES", None)
else:
os.environ["MUSA_VISIBLE_DEVICES"] = old
class ForeignVendorModuleTests(unittest.TestCase):
"""benchmark 的可用性判据常写成 `try: import <厂商实现>; AVAILABLE = True`。
import 只证明那个 .py 在磁盘上——它不碰设备，所以在任何后端上都成功，gate 于是
形同虚设。router_gemm 就这样让 NVIDIA Hopper 的 kernel 跑在了 MUSA 卡上，最后
死在 JIT 深处的 "dynamic_func() missing BLOCK_M ..."，报错完全指不向真正的原因。
"""
P = "flag_gems.runtime.backend._"
def _mods(self, *names):
return {self.P + n: object() for n in names} | {"torch": object()}
def test_only_own_vendor_is_clean(self):
from _device_setup_plugin import foreign_vendor_modules
mods = self._mods("mthreads", "mthreads.ops.mm")
self.assertEqual(foreign_vendor_modules("mthreads", mods), [])
def test_foreign_vendor_is_reported_once_per_root(self):
from _device_setup_plugin import foreign_vendor_modules
# 同一厂商的多个子模块只报一条
mods = self._mods("mthreads.ops.mm",
"nvidia.hopper.ops.mm", "nvidia.ops.mm", "nvidia")
self.assertEqual(foreign_vendor_modules("mthreads", mods), ["nvidia"])
def test_several_foreign_vendors(self):
from _device_setup_plugin import foreign_vendor_modules
mods = self._mods("mthreads", "nvidia.ops.mm", "cambricon.ops.mm")
self.assertEqual(foreign_vendor_modules("mthreads", mods),
["cambricon", "nvidia"])
def test_a_vendor_whose_name_extends_ours_is_still_foreign(self):
# 按前缀比会把 _mthreads_extra 认成自家
from _device_setup_plugin import foreign_vendor_modules
mods = self._mods("mthreads", "mthreads_extra.ops")
self.assertEqual(foreign_vendor_modules("mthreads", mods),
["mthreads_extra"])
def test_unrelated_modules_are_ignored(self):
from _device_setup_plugin import foreign_vendor_modules
mods = {"torch": object(), "triton": object(),
"flag_gems.ops.mm": object()}
self.assertEqual(foreign_vendor_modules("mthreads", mods), [])
class PartitionTests(unittest.TestCase):
"""按耗时把算子分到各卡。hash 分片在这批活上比它慢 50 分钟。"""
HEADER = "op\top_file\telapsed_before\telapsed_after\tverdict"
def _load(self, assignment):
return sorted(sum(w for _, w in v) for v in assignment.values())
def _screen(self, tmp, rows, header=None):
p = Path(tmp, "screen.tsv")
p.write_text((header or self.HEADER) + "\n" + "\n".join(rows) + "\n",
encoding="utf-8")
return p
def test_heavy_first_keeps_cards_even(self):
from partition import plan
items = [("a", 10), ("b", 9), ("c", 8), ("d", 7), ("e", 6), ("f", 5)]
self.assertEqual(self._load(plan(items, ["0", "1", "2"])), [15, 15, 15])
def test_one_item_heavier_than_the_ideal_share_sets_the_floor(self):
# topk 单个 70 分钟、理想均分 45——加卡不解决，分片只保证别人不空等
from partition import imbalance, plan
items = [("topk", 70)] + [(f"op{i}", 5) for i in range(10)]
a = plan(items, [str(i) for i in range(8)])
self.assertEqual(max(sum(w for _, w in v) for v in a.values()), 70)
self.assertGreater(imbalance(a), 1.0)
def test_more_cards_than_operators_leaves_some_empty(self):
from partition import plan
a = plan([("a", 1), ("b", 1)], ["0", "1", "2", "3"])
self.assertEqual(sum(1 for v in a.values() if not v), 2)
def test_same_input_gives_the_same_plan(self):
from partition import plan
items = [("a", 5), ("b", 5), ("c", 5), ("d", 5)]
first = plan(items, ["0", "1"])
for _ in range(5):
self.assertEqual(plan(list(reversed(items)), ["0", "1"]), first)
def test_no_cards_is_the_callers_error(self):
from partition import plan
with self.assertRaises(ValueError):
plan([("a", 1)], [])
def test_perfect_balance_is_one(self):
from partition import imbalance, plan
self.assertEqual(imbalance(plan([("a", 5), ("b", 5)], ["0", "1"])), 1.0)
def test_verdict_filters_and_reports_what_it_dropped(self):
from partition import read_screen
with tempfile.TemporaryDirectory() as tmp:
f = self._screen(tmp, ["a\ta\t10\t10\tIR_CHANGED",
"b\tbb\t5\t5\tNO_IR_DIFF",
"c\tc\t1\t1\tNO_TRITON"])
items, files, skipped = read_screen(f)
self.assertEqual(items, [("a", 20.0)])
self.assertEqual(files["a"], "a")
self.assertEqual(len(skipped), 2)
def test_op_file_defaults_to_op_name(self):
from partition import read_screen
with tempfile.TemporaryDirectory() as tmp:
f = self._screen(tmp, ["a\t\t1\t1\tIR_CHANGED"])
_, files, _ = read_screen(f)
self.assertEqual(files["a"], "a")
def test_missing_elapsed_falls_back_to_equal_weights(self):
from partition import read_screen
with tempfile.TemporaryDirectory() as tmp:
f = self._screen(tmp, ["a\ta\tIR_CHANGED", "b\tb\tIR_CHANGED"],
header="op\top_file\tverdict")
items, _, _ = read_screen(f)
self.assertEqual([w for _, w in items], [1.0, 1.0])
if __name__ == "__main__":
unittest.main()