9 Commits
Author SHA1 Message Date
AharrypotterandClaude Opus 5 d7cb7efc94 fix: have the campaign worker ask zl_bench for its plugin list
The worker carried a hand-copied plugin array, taken from zl_bench before
backends/ existed, and it had drifted twice over. It still named
_device_guard_plugin, which is now a shim for _device_setup_plugin. More
consequentially it carried no timing plugin at all, so every measurement this
campaign ever sealed was timer=direct while zl_bench's own driver had been
capturing -- a difference the harness never reported, because the missing plugin
cannot announce itself.

It now calls plugin_plan.py with ZL_BACKEND=musa and uses whatever comes back,
so the caliber follows the backend's declared capabilities instead of a snapshot
of them. The plan is preflighted for non-emptiness, since an empty array would
silently run pytest with no plugins at all: no shape injection, no autotune
record, no IR dump, and a table that still looks well-formed.

Two effects on this campaign. _timing_plugin now enters the list, and since
MUSA declares graph capture the worker will capture where the operator allows
it; operators that cannot are marked per row, as on any other backend. And
_parallel_warmup_plugin joins, inert unless PARALLEL_WARMUP_GPUS is set.

The sealed 32-workload run therefore sits on the far side of a caliber change
and is not comparable with anything produced after this commit. Re-running it is
a campaign decision, not a harness one.

The worker test asserted the literal "-p _musa_benchmark_compat_plugin"; it now
asserts the delegation and that no -p is spelled out locally. What the plan
holds for musa is covered where it belongs, in zl_bench's test_backends.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-10 10:34:05 +08:00
AharrypotterandClaude Opus 5 75405cc555 Merge the S5000 MUSA RLC product campaign
Brings in campaigns/musa_rlc_product/: the runner, its worker, manifest
validation, the sealed 32-workload manifest, a topology re-accounting tool and
their tests. Developed on a fork branch that never reached main, and it merges
without conflict because it only adds a new directory; the one file it shares
with main, _musa_benchmark_compat_plugin.py, is byte-identical to the copy
already committed here.

The campaign layer sits above zl_bench rather than inside it: freezing a
workload set, classifying outcomes, and producing receipts, where zl_bench turns
one operator into one reproducible measurement. Two of its ideas are worth
naming because they are absent below it. Narrowing what a workload measures --
which correctness variant, which benchmark case -- requires a stated reason that
manifest validation enforces, so an exclusion stays visible in review. And
reaccount_musa_rlc_topology.py re-derives verdicts from sealed artifacts under a
corrected accounting rule without re-running anything, while refusing to promote
a workload on the new count, only to demote it.

Its worker still hardcodes a plugin list copied from a pre-backends zl_bench,
which is why the campaign has been timing with timer=direct. Switching it to
plugin_plan.py is the next change and will alter its caliber, so the sealed run
is not comparable across that boundary.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-10 10:32:33 +08:00
AharrypotterandClaude Opus 5 29af59be9b feat: absorb the MThreads compat plugin and the A/B range report
Three pieces from the fork's feature/musa-benchmark-support branch that never
reached main, rebased onto the backend abstraction rather than copied verbatim.

_musa_benchmark_compat_plugin drops FlagGems benchmark skip guards written as
`torch.cuda.is_available()` where the code under test dispatches through
flag_gems.device. On MThreads those turn a working operator into "not measured":
test_fft.py skipped entirely, and unskipped on an S5000 it produces 25 rows. The
plugin pins each override to one file and one exact skip reason and only acts
when flag_gems has dispatched to the mthreads vendor; two guards were probed on
device and deliberately left in place, which is why they are named in its
docstring rather than silently absent.

It is wired through a new DeviceBackend.compat_plugins rather than a name test
in the planner, so the policy stays where every other backend difference lives.
Compat plugins are planned first because they rewrite collection-time markers
and nothing may read the item list before them; a test pins that position and
another pins that they do not leak to other backends.

batch_summary gains --ab: the speedup column compares gems against torch, not
one build against another, so an A/B needs the two run.logs paired by operator,
dtype and size detail. Only rows that succeeded on both sides with a positive
latency count, so an incomplete pairing cannot look like a complete range, and
the range is annotated when either side carries a BENCHMARK_DIRECT_NO_CUDAGRAPH
marker, since a fallback row was timed under a different regime.

ab_rlc_test.sh is the same caliber as ab_fold_test.sh with the comparison axis
moved to FLAGTREE_<VENDOR>_RLC_ENHANCE. Both are caliber examples, not
regression scripts: if the switch under test no longer exists the script still
runs every round and prints a full table showing no difference, so confirm the
axis is live before reusing either.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-10 10:31:59 +08:00
AharrypotterandClaude Opus 5 c9f9b7885f feat: check the run's preconditions, and record which tuner cache it used
Two failure modes that cost this harness real results, both of them silent.

A wrong path does not fail as a wrong path. pytest cannot collect a benchmark
directory that is not there, so benchmark/conftest.py never registers --level
and --mode, and the run dies with "unrecognized arguments: --level core --mode
kernel" -- an error about the flags, several layers away from the missing
directory. run_pytest.sh now checks the FlagGems directory, the benchmark file,
SHAPE_FILE and REPLAY_FROM before starting pytest, prints
"PREFLIGHT_FAILED check=<the named check>" and exits 3, distinct from pytest's
own 1 and 4 so the batch layer can tell a broken environment from a broken
operator. Only the file half of a node is checked; a missing ::test_<fn> already
produces a clear "not found: ..." from pytest and needs no help.

FlagGems keeps libtuner winners in a sqlite DB under $FLAGGEMS_CACHE_DIR
(default ~/.flaggems), shared by every run on the machine. Whichever side of an
A/B runs first therefore hands its winners to the second, which then compiles
only those: measured on router_gemm, the second of two shared-cache runs
compiled 10 variants instead of 234 and finished in 17s instead of 52s. That is
how an mm A/B in this repo produced "945 variants -> 33, 136 converts -> 10" and
read as a compiler result when it was cache inheritance; isolated, both runs
compile 234.

ZL_ISOLATE_TUNER_CACHE=1 gives a run its own cache directory, dropped afterwards
with the Triton one since the chosen configs are already in autotune_records/.
It defaults to 0, because cold tuning is expensive and the standard REPLAY_FROM
A/B pins configs anyway; only record-against-record is exposed. The default is
unchanged, but the caliber is no longer invisible: either way it goes into the
run header and run_identity.json as tuner_cache, which is what was missing when
the mm numbers could not be explained after the fact.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-10 10:31:36 +08:00
Aharrypotter e970ed94c2 docs: document the MUSA RLC campaign
Covers the three screen cells and why replay locks them to one variable, how to
freeze and run a campaign, what the runner refuses to start on, the six
classifications, and the three scoping rules that decide whether a verdict
means anything.

Also states what a terminal campaign does not prove, since the manifest fixes
production_default_policy to unchanged-off and a passing run is not an argument
for flipping it.
2026-09-02 11:36:33 +08:00
Aharrypotter 8a685638bb feat: re-account a sealed MUSA RLC run without re-running it
A sealed run already holds every TTGIR dump and its launch index, so a change
in how convert topology is counted can be applied to it directly instead of
spending accelerator time.

The tool re-reads each cell's dump, recomputes the scoped count, re-runs the
campaign classifier against the sealed correctness verdicts, and writes a
derived receipt and matrix outside the run root. It refuses to write inside a
sealed run and rejects a directory with no campaign receipt.

A derived verdict can only demote a topology claim. Promoting one to G or C
needs reversed-order timing, which a sealed run holds only for tasks that
already showed a reduction, so a reduction appearing for the first time under
scoping is reported as X for missing round-b evidence rather than guessed.

Run against the sealed 32-workload S5000 campaign, a floor of 1 changes exactly
one task: gelu_backward moves from C to B, and the only excluded files in the
whole campaign are its five once-launched randn_kernel dumps. A floor of 0
reproduces the sealed counts with no task changed, which is the equivalence
that makes the scoped number auditable rather than merely different.
2026-09-02 11:35:43 +08:00
Aharrypotter 8bbea472bb feat: add the S5000 MUSA RLC product campaign
A frozen 32-workload campaign that measures how FlagTree's RemoveLayoutConversions
pass behaves on a Moore Threads S5000, through FlagGems public dispatch and this
repository's existing plugins.

The manifest is the contract: source commits, the single physical device, the
three screen cells off/mask0/candidate, the native-event timer, and per-workload
dtypes, timeouts and canary membership. prepare validates it against
ops/inventory_main.csv and freezes an immutable task matrix; the runner executes
the matrix serially, and the worker drives one operator cell through pytest with
_ir_meta_plugin, _autotune_record_plugin and the rest.

Three properties are worth calling out because they decide whether a verdict
means anything.

Convert topology is scoped to the measured operator. A benchmark call builds its
inputs once and then runs the operator many times under the timer, so counting
ttg.convert_layout over every executed kernel credits the candidate with
converts removed from input construction that no timer can observe. Each dumped
TTGIR file is joined to its launch count from index.tsv and only variants
launched more than measured_launch_floor contribute; the unscoped total and the
excluded files stay in the receipt, and a cell that cannot be joined to its
index is classified X rather than trusted.

Correctness is scoped to the variant a workload measures. Running a whole test
file lets an unrelated variant decide the verdict, so a workload may pin node
ids or a -k selector, never both, and any narrowing must state its reason.

Targets are resolved before the campaign starts. The inventory can lag the
pinned FlagGems commit, and such a task previously failed only inside the
worker's preflight, which exited silently and left an empty log. The runner now
resolves every benchmark and correctness target against the real tree, a
workload may name a target the inventory does not know about if it says why,
and each preflight check names itself on failure.
2026-09-02 11:35:30 +08:00
Aharrypotter 22139e4b87 chore: pin shipped text formats to LF
This repository's drivers are shell scripts executed on Linux accelerator
hosts, and its consumers record the SHA-256 of the files they ran. With Git's
default autocrlf on Windows a checkout rewrites those files with CRLF, which
breaks the shell drivers outright and changes every digest a receipt refers to.
2026-09-02 11:34:48 +08:00
Aharrypotter 3b54f458e6 feat: add a MThreads benchmark compatibility plugin
Several upstream FlagGems tests gate themselves on torch.cuda.is_available()
even though the code under test runs on whatever accelerator FlagGems
dispatches to. On MThreads those guards skip the operator instead of
exercising it, so a working path reports as "not measured".

The plugin drops such a guard only when it is pinned to one file and one exact
skip reason, only when FlagGems has dispatched to the MThreads vendor, and only
after the unskipped test was observed to pass on an S5000. It walks the node
chain rather than the item's own markers, because a module-level pytestmark
does not appear on the item.

Two guards were probed on device and deliberately left in place, which is why
they are named in the module docstring rather than silently absent.
test_grid_sample.py builds its inputs through a helper whose device argument
defaults to the literal "cuda": unskipped, all 90 collected cases fail in
aten::empty on a CUDA backend that is not present. test_cutlass_scaled_mm.py
gates on vLLM as well as CUDA, so removing its guard would raise on the missing
dependency instead of measuring anything.
2026-09-02 11:34:04 +08:00
18 changed files with 3355 additions and 6 deletions

No files matched your search

+17 -1
View File
@@ -66,6 +66,7 @@ softmax:
| `REPLAY_FROM` | 空(record 模式) | 指向某次历史 run 目录,replay 其 autotune 选择。做 A/B 时 B 侧必须设(见「A/B 对比测试标准流程」) |
| `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` |
| `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` |
@@ -148,7 +149,19 @@ libtuner 的持久缓存何时失效:FlagGems kernel 源码改动、tune_confi
同一层缓存也决定了 `_autotune_record_plugin` 的实现方式:record 不能靠"cache 新增了哪个 key"来判断本次选中的 config(缓存一热就走 cached 分支、不写新 key,键集差集恒为空),改为 `run()` 之后直接读回本次调用的 `self.cache[key]`;replay 同理不能加"key 不在 cache 里才注入"的前置条件,否则注入被跳过、该 run 表面在 replay 实际在自调优。想确认 record 真的覆盖到目标 kernel,查 `runs/<run>/autotune_records/<op>.json` 里有无对应 kernel 条目——漏记时该文件照样生成,只是少了 libtuner 那几个。
### ab_fold_test.sh:A/B 测试参考示例
### 前置检查
`run_pytest.sh` 在启动 pytest 之前校验 FlagGems 目录、benchmark 文件、`SHAPE_FILE`
与 `REPLAY_FROM`,失败时打 `!!! PREFLIGHT_FAILED check=<具名检查>` 并以 3 退出。
这些检查存在的理由是**静默失败会指向错误的方向**:路径不存在时 pytest 收集不到
benchmark 目录,注册 `--level`/`--mode` 的 `conftest.py` 就没被加载,报错变成
`unrecognized arguments: --level core --mode kernel`,看起来像参数或版本问题。
只校验节点的文件那一半;`::test_<fn>` 交给 pytest,函数不存在时它自己会清楚地报
`not found: ...`。
### 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 轮:
@@ -157,6 +170,8 @@ bash ab_fold_test.sh
# 汇总 log:runs/ab_fold_<时间戳>.log(已去色);快速对账:grep -E '^#####|FAILED' <log>
```
`ab_rlc_test.sh` 是同一口径的另一个实例,对比轴换成 FlagTree 的 `FLAGTREE_<VENDOR>_RLC_ENHANCE`(`VENDOR` 默认 `MUSA`),算子表由 `OPS` 指定,跑完调 `batch_summary.py --ab <before> <after>` 出该算子的 speedup 区间。
可复用的是**口径**——同 config replay、cudagraph 计时、真实 trace shape 集、逐对成组——对比其它开关 / 编译器改动 / 算子,改循环变量与环境变量即可,口径部分不用动。
但**它是口径示例,不是可长期直接执行的回归脚本**:对比轴是 FlagGems 侧的算子开关,会随上游演进被改名、改语义或移除,本仓库不跟随同步更新(该脚本引用的开关目前就已被上游移除)。失效后果是**静默的**:脚本照样跑完全部轮次、照样输出完整对比表,只是两侧执行同一份代码,得出"无差异"的假结论。所以套用前先确认对比轴仍有效——`grep -rn '<开关名>' $FLAGGEMS_DIR/src` 要有命中,且两侧 latency 表确有差异。
@@ -255,6 +270,7 @@ python run_batch.py --ops-file my_ops.txt --replay-root runs/batch_xxx
| `_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 运行 |
| `_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` |
| `_mm_cluster_fix_plugin` | Hopper fp16 mm cluster kernel 越界崩溃的运行时规避 | 默认关;`ZL_ENABLE_MM_CLUSTER=1` 或 `ZL_EXTRA_PLUGINS` |
+89
View File
@@ -0,0 +1,89 @@
"""Narrow compatibility fixes for the frozen MUSA product corpus.
Several upstream FlagGems tests gate themselves on ``torch.cuda.is_available()``
even though the code under test runs on any accelerator FlagGems dispatches to.
On MThreads those guards skip the test instead of exercising the operator, which
turns a working path into a verdict of "not measured".
Every override here is pinned to one file and one exact skip reason, applies
only when FlagGems has dispatched to the MThreads vendor, and is only added
after the unskipped test was observed to pass on the device.
Two files were probed on S5000 and deliberately left out, because their guards
are not stale:
- ``test_cutlass_scaled_mm.py`` gates on vLLM as well as CUDA, so removing the
guard would raise on the missing dependency instead of measuring anything.
- ``test_grid_sample.py`` builds its inputs through a helper whose device
argument defaults to the literal ``"cuda"``. Its guard is consistent with its
own body: unskipped, all 90 collected cases fail in ``aten::empty`` on a CUDA
backend that does not exist here. Making that file portable means editing
upstream test source, which is outside what a compatibility plugin may do.
"""
from __future__ import annotations
import sys
from pathlib import Path
# file name -> exact skipif reasons that are stale on a dispatching backend.
_STALE_CUDA_GUARDS: dict[str, frozenset[str]] = {
"test_fft.py": frozenset({"CUDA is not available"}),
"test_top_k_per_row_decode.py": frozenset({"CUDA device required"}),
}
# Guards that are only stale for one test in the file stay pinned to it.
_PINNED_TESTS: dict[str, str] = {
"test_fft.py": "test_perf_fft",
}
def _is_stale_cuda_guard(item, mark) -> bool:
if mark.name != "skipif":
return False
name = Path(str(item.path)).name
reasons = _STALE_CUDA_GUARDS.get(name)
if reasons is None or mark.kwargs.get("reason") not in reasons:
return False
pinned = _PINNED_TESTS.get(name)
return pinned is None or item.name == pinned
# Retained under its historical name: the sealed campaign's tests reference it.
def _is_stale_fft_cuda_guard(item, mark) -> bool:
return Path(str(item.path)).name == "test_fft.py" and _is_stale_cuda_guard(
item, mark
)
def pytest_collection_modifyitems(items):
import flag_gems
if flag_gems.vendor_name != "mthreads":
return
removed: dict[str, int] = {}
# A guard can sit on the test, its class, or the module's pytestmark, and
# skipif is resolved by walking that whole chain. Strip every level.
seen: set[int] = set()
for item in items:
for node in item.listchain():
if id(node) in seen:
continue
seen.add(id(node))
markers = getattr(node, "own_markers", None)
if not markers:
continue
kept = [mark for mark in markers if not _is_stale_cuda_guard(item, mark)]
dropped = len(markers) - len(kept)
if dropped:
markers[:] = kept
name = Path(str(item.path)).name
removed[name] = removed.get(name, 0) + dropped
for name, count in sorted(removed.items()):
print(
f"[musa-benchmark-compat] removed {count} stale CUDA-only skip "
f"guard(s) from {name}",
file=sys.stderr,
flush=True,
)
+5
View File
@@ -71,6 +71,10 @@ def main(argv=None) -> int:
p.add_argument("--test-file", default=os.environ.get("TEST_FILE", ""))
p.add_argument("--shape-mode", default=os.environ.get("SHAPE_MODE", ""))
p.add_argument("--bench-level", default=os.environ.get("BENCH_LEVEL", "core"))
# "isolated" or "shared(<path>)". Recorded because a shared libtuner cache
# lets whichever side runs first hand its winners to the other, and after
# the fact the two runs are otherwise indistinguishable.
p.add_argument("--tuner-cache", default="")
args = p.parse_args(argv)
init_identity(
args.out_dir,
@@ -83,6 +87,7 @@ def main(argv=None) -> int:
test_file=args.test_file,
shape_mode=args.shape_mode,
bench_level=args.bench_level,
tuner_cache=args.tuner_cache,
)
return 0
+68
View File
@@ -0,0 +1,68 @@
#!/bin/bash
# A/B 对比:FlagTree RemoveLayoutConversions 增强关 vs 开,逐算子两轮。
#
# 口径:每个算子先跑关(record autotune config),再用 REPLAY_FROM 回放同一套
# config 跑开,两侧只差 FLAGTREE_<VENDOR>_RLC_ENHANCE 这一个开关。产物沿用
# run_pytest.sh 的目录结构,所以每个算子恰好留下两份 run.log:
#
# runs/<op>_<时间戳>/run.log 关(before)
# runs/<op>_<时间戳>/run.log 开(after)
#
# 跑完用 batch_summary.py --ab <before> <after> 得到该算子的 speedup 区间。
#
# 用法:
# OPS="softmax gelu_backward" bash ab_rlc_test.sh
# OPS="sort:perf_sort topk" bash ab_rlc_test.sh # 冒号分隔 OP:OP_FILE
#
# 注意:RLC 开关名带厂商前缀,用 VENDOR 指定(默认 MUSA)。套用前先确认开关有效:
# grep 该开关名于 FlagTree 源码应有命中,否则两侧跑的是同一份代码。
set -uo pipefail
cd "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
VENDOR="${VENDOR:-MUSA}"
ENHANCE_VAR="FLAGTREE_${VENDOR}_RLC_ENHANCE"
# 空格分隔的算子表。元素为 OP 或 OP:OP_FILE(省略时 OP_FILE=OP)。
OPS="${OPS:-softmax}"
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
{
echo ">>> vendor=$VENDOR switch=$ENHANCE_VAR ops=$OPS"
for entry in $OPS; do
op="${entry%%:*}"
op_file="${entry#*:}"
[[ "$op_file" == "$entry" ]] && op_file="$op"
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)
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)
echo "########## $op | summary ##########"
python batch_summary.py --ab "$BEFORE/run.log" "$AFTER/run.log" --label "$op" \
|| echo "!!! summary failed: $op"
done
echo "########## ALL DONE ##########"
} 2>&1 | tee "$LOG"
# 去掉 ANSI 转义以便 grep/diff(终端输出保留颜色)
sed -i -E $'s/\x1b\\[[0-9;]*[A-Za-z]//g' "$LOG"
echo ">>> 总 log: $(pwd)/$LOG"
echo ">>> 每算子两份 run.log 在 runs/<op>_<时间戳>/ 下,按时间先后即 before / after"
+7
View File
@@ -90,6 +90,13 @@ class DeviceBackend:
def denies_capture(self, op: str) -> bool:
return bool(op) and op in self.capture_deny_ops
# Plugins that patch around upstream assumptions this backend violates, e.g.
# FlagGems benchmarks that gate themselves on torch.cuda.is_available() even
# though the operator under test runs on whatever device FlagGems dispatched
# to. Loaded first so collection-time overrides land before anything reads
# the item list. Each such plugin must scope itself to its own vendor.
compat_plugins: Tuple[str, ...] = ()
# Optional listing command; (argv, line_prefix). prefix="" means count
# non-empty lines. Must be a real vendor CLI; do not invent flags.
list_command: Optional[Tuple[Sequence[str], str]] = None
+5
View File
@@ -42,6 +42,11 @@ class MusaBackend(DeviceBackend):
# timing precision instead of its measurement. Revisit if a newer torch_musa
# changes the behaviour.
capture_deny_ops = ("unique2",)
# Several FlagGems benchmarks skip on `not torch.cuda.is_available()` while
# the code under test dispatches through flag_gems.device. On MThreads that
# turns a working operator into "not measured"; the plugin drops those
# guards, pinned to one file and one exact skip reason each.
compat_plugins = ("_musa_benchmark_compat_plugin",)
# 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
+71 -1
View File
@@ -186,12 +186,82 @@ def summarize(batch_dir: Path, out_path: Optional[Path] = None) -> Path:
return out_path
def ab_rows(before_text: str, after_text: str) -> List[Dict[str, str]]:
"""把两次 run.log 的测量行按 (op, dtype, size_detail) 配对。
speedup 列是 gems 对 torch,不是前后对比,所以这里用 gems_lat 自己算
before/after - 1。只有两侧都 SUCCESS 且延迟为正的行才计入;对不上的行
单独返回,避免用不完整的配对报一个看起来完整的区间。
"""
def index(text: str) -> Dict[Tuple[str, str, str], float]:
out: Dict[Tuple[str, str, str], float] = {}
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:
out[(row.get("fg_op_name", ""), row.get("dtype", ""),
row.get("size_detail", ""))] = lat
return out
before, after = index(before_text), index(after_text)
paired: List[Dict[str, str]] = []
for key in sorted(before.keys() & after.keys()):
op, dtype, size = key
b, a = before[key], after[key]
paired.append({
"fg_op_name": op, "dtype": dtype, "size_detail": size,
"before_lat": f"{b:.6f}", "after_lat": f"{a:.6f}",
"speedup_percent": f"{(b / a - 1.0) * 100.0:.2f}",
})
return paired
def report_ab(before: Path, after: Path, label: str = "") -> List[Dict[str, str]]:
"""打印一个算子的前后对比:逐行 speedup 与整体区间。"""
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)
name = label or (rows[0]["fg_op_name"] if rows else "?")
if not rows:
print(f"[ab] {name}: 没有可配对的测量行"
f"(before={before} after={after})")
return rows
values = [float(r["speedup_percent"]) for r in rows]
for r in rows:
print(f" {r['dtype']:<16} {float(r['speedup_percent']):+7.2f}% "
f"{r['before_lat']} -> {r['after_lat']} {r['size_detail']}")
fallback = ""
if DIRECT_NO_CUDAGRAPH_RE.search(before_text) or \
DIRECT_NO_CUDAGRAPH_RE.search(after_text):
# 回退行用的是 do_bench 口径,与 cudagraph 行不同表可比性
fallback = " [含 BENCHMARK_DIRECT_NO_CUDAGRAPH 回退行]"
print(f"[ab] {name}: {len(rows)} 行配对,speedup "
f"{min(values):+.2f}% ~ {max(values):+.2f}%{fallback}")
return rows
def main() -> None:
p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("batch_dir", type=Path, help="run_batch.py 的产物目录")
p.add_argument("batch_dir", type=Path, nargs="?",
help="run_batch.py 的产物目录")
p.add_argument("-o", "--output", type=Path, default=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 报告里的算子名")
args = p.parse_args()
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)
return
if args.batch_dir is None:
raise SystemExit("需要 batch_dir,或用 --ab BEFORE AFTER")
if not args.batch_dir.is_dir():
raise SystemExit(f"目录不存在: {args.batch_dir}")
out = summarize(args.batch_dir, args.output)
+103
View File
@@ -0,0 +1,103 @@
# S5000 MUSA RLC 产品验证 campaign
在 Moore Threads S5000 上,经由 FlagGems public dispatch,测量 FlagTree
`RemoveLayoutConversions`(RLC)增强对 32 个冻结 workload 的影响。
本目录只提供 campaign 层:契约、冻结、执行、判定、重记账。单算子插件、shape、
inventory 仍然复用仓库根目录的既有能力。
## 组成
| 文件 | 作用 |
|---|---|
| `manifest/musa-rlc-32-product-v1.json` | 冻结契约:源码提交、设备、cell、计时口径、32 个 workload |
| `prepare_musa_rlc_product.py` | 校验 manifest 与 `ops/inventory_main.csv`,冻结不可变 task matrix |
| `run_musa_rlc_product.py` | 串行执行 task matrix,产出逐 cell、逐 task、campaign 三级收据 |
| `run_pytest_musa_rlc_product.sh` | 单个算子单个 cell 的 worker,挂载根目录插件 |
| `reaccount_musa_rlc_topology.py` | 对已封存 run 按新口径重记账,不占用设备 |
| `tests/` | 上述四者的单元测试 |
根目录的 `_musa_benchmark_compat_plugin.py` 属于插件族,不在本目录。
## 三个 cell
| cell | `FLAGTREE_MUSA_RLC_ENHANCE` | phase mask | 含义 |
|---|---|---|---|
| `off` | 0 | — | 增强关闭的基线 |
| `mask0` | 1 | 0 | 增强开启但所有 phase 关闭;它的 topology 必须等于 `off`,否则控制面本身有问题 |
| `candidate` | 1 | 13 | 1a、2、3 启用,1b 未启用 |
`off` 记录 autotune config,`mask0` 与 `candidate` 回放同一套 config。回放回退即判定失败,
这样 A/B 之间的差异才收敛到编译器改动这一个变量。
## 跑法
```bash
python campaigns/musa_rlc_product/prepare_musa_rlc_product.py \
campaigns/musa_rlc_product/manifest/musa-rlc-32-product-v1.json \
--inventory-dir ops --selection canary --dry-run
```
`--dry-run` 只做校验与预算估算。去掉它并给出 `--output-root` 才会冻结 task matrix。
`--selection all` 需要 `--canary-receipt` 指向一份通过的 canary 收据。
执行:
```bash
python campaigns/musa_rlc_product/run_musa_rlc_product.py \
--manifest <frozen>/manifest.json \
--task-matrix <frozen>/task-matrix.json \
--build-root <flagtree build root> \
--flaggems <flaggems> --zl-bench <this repo> \
--worker campaigns/musa_rlc_product/run_pytest_musa_rlc_product.sh \
--output-root <run root>
```
runner 在开跑前校验:manifest 与 matrix 的哈希绑定、FlagTree/FlagGems/zl_bench 三处源码身份、
Triton 与 FlagGems 确实从冻结路径导入、设备可见数为 1、所有 benchmark 与 correctness 目标
在真实源码树里存在、设备当前没有其它进程。任何一条不成立就整批拒绝。
单元测试直接跑文件:
```bash
python campaigns/musa_rlc_product/tests/test_musa_rlc_product.py
python campaigns/musa_rlc_product/tests/test_reaccount_musa_rlc_topology.py
```
## 判定
| 类 | 含义 |
|---|---|
| G | 正确,convert topology 下降,且双执行顺序下均稳定提速 ≥5% |
| C | 正确,topology 下降,但收益不足 5% |
| B | 正确,但 candidate 没有减少被测算子的 convert topology |
| R | 正确,但双顺序稳定退化 ≤ -3% |
| K | candidate 特有的正确性失败 |
| X | 不支持、超时、harness 失败,或同配置回放被破坏 |
只有 topology 相对 incumbent 下降时才追加反向顺序计时。convert 数下降不等于变快,
这一点在本 campaign 内已有反例。
## 三条口径边界
**convert 计数只算被测算子。** benchmark 调用会先构造一次输入再在计时区内反复执行算子。
把所有执行过的 kernel 都计入,会把输入构造 kernel 里消掉的 convert 算成收益,而计时器
根本观测不到它。因此每个 dump 的 TTGIR 都与 `index.tsv` 的 launch 数关联,只有 launch
数超过 `measured_launch_floor` 的变体才计入。未收敛总数与被排除文件保留在收据里;
无法与 index 关联的 cell 判为 X,而不是相信一个界定不了的 topology 声明。
**correctness 只跑该 workload 实际测量的变体。** 以整个测试文件为粒度,会让无关变体
决定被测算子的结论。workload 可以钉 node id 或 `-k` 选择器,两者互斥,且任何收窄都必须
写明理由。完整 argv 落在收据里。
**目标在开跑前解析。** inventory 可能落后于固定的 FlagGems 提交,例如上游把一个 benchmark
文件拆成多个。这类任务过去只在 worker 前置检查里静默失败,留下空日志。现在 runner 会
对真实源码树解析全部目标,workload 可以声明 inventory 里没有的目标但必须给出理由,
worker 的每个前置检查失败时自报名称。
## 不属于本 campaign 的结论
一次 terminal 的 campaign 只证明:该冻结语料在该设备该提交下的 dispatch 正确性、
layout topology 响应、以及同契约下的 native-event 延迟。它不证明跨厂商正确性、
性能对等、CI 状态,也不构成打开生产默认开关的依据。manifest 的
`production_default_policy` 固定为 `unchanged-off`。
@@ -0,0 +1,551 @@
{
"schema_version": 1,
"campaign_id": "s5000-musa-rlc-32-product-v1-20260902-v3",
"experiment_kind": "musa-rlc-product-validation",
"permitted_claim": "S5000 MUSA product-dispatch correctness, layout-topology response, and same-contract native-event latency for this frozen corpus only",
"production_default_policy": "unchanged-off",
"source": {
"flagtree_commit": "3059917e2df16f8ea35c5efe1ed1af2d87117396",
"flaggems_commit": "46ef8080329190d8da8738340240de2d5a475950",
"zl_bench_commit": "da2288564584039366e7be07fe4e94539a560c81"
},
"device": {
"ssh_alias": "S5000",
"container": "zixu_flagos_S5000",
"backend": "mthreads",
"model": "Moore Threads S5000",
"physical_device": "7",
"logical_device": "0",
"visible_device_count": 1
},
"protocol": {
"screen_cells": [
"off",
"mask0",
"candidate"
],
"timing_cells": [
"off",
"candidate"
],
"canary_order": [
"dropout",
"softmax",
"instance_norm",
"fft",
"sort",
"batch_norm",
"topk",
"conv1d"
],
"candidate_phase_mask": 13,
"candidate_profitability_policy": true,
"candidate_product_launch_count": 1,
"default_benchmark_dtypes": [
"float16"
],
"timer": "triton.testing.do_bench via MUSA device-interface Event",
"autotune_contract": "off-record then mask0/candidate replay from off",
"fresh_process_per_operator_cell": true,
"same_physical_device_for_all_cells": true,
"serial_measurement": true,
"full_campaign_requires_canary_terminal": true,
"correctness": "FlagGems public dispatch quick tests under candidate, narrowed to the variant each workload measures and run with the MThreads compatibility plugin, plus benchmark execution for every cell",
"classification": {
"G": "correct, convert count decreases, and stable speedup >= 5%",
"C": "correct and convert count decreases, but latency intervals overlap or speedup < 5%",
"B": "correct but candidate does not reduce the executed convert topology",
"R": "correct but stable regression <= -3%",
"K": "candidate correctness failure",
"X": "unsupported, timeout, harness failure, or same-contract replay violation"
},
"timeout_seconds_by_cost": {
"light": 300,
"medium": 900,
"heavy": 1800
},
"measured_launch_floor": 1,
"supported_benchmark_dtypes": [
"float16",
"float32",
"bfloat16",
"int16",
"int32",
"bool",
"complex64"
]
},
"workloads": [
{
"id": "batch_norm",
"repo": "main",
"op": "batch_norm",
"op_file": "batch_norm",
"shape_key": "batch_norm",
"correctness_file": "tests/test_batch_norm.py",
"family": "normalization",
"nv_evidence": "strong-positive",
"nv_gain_percent": 43.1,
"prior_s5000": "B 0->0",
"cost": "medium",
"canary": true
},
{
"id": "replication_pad3d",
"repo": "main",
"op": "replication_pad3d",
"op_file": "replication_pad3d",
"shape_key": "replication_pad3d",
"correctness_file": "tests/test_replication_pad3d.py",
"family": "padding",
"nv_evidence": "strong-positive",
"nv_gain_percent": 38.6,
"prior_s5000": "B 0->0 at author shape",
"cost": "medium",
"canary": false
},
{
"id": "fft",
"repo": "main",
"op": "perf_fft",
"op_file": "fft",
"shape_key": "fft",
"correctness_file": "tests/test_fft.py",
"family": "fft",
"nv_evidence": "strong-positive",
"nv_gain_percent": 36.5,
"prior_s5000": "G 38->2",
"cost": "heavy",
"canary": true,
"benchmark_dtypes": [
"complex64"
]
},
{
"id": "topk",
"repo": "main",
"op": "topk",
"op_file": "topk",
"shape_key": "topk",
"correctness_file": "tests/test_topk.py",
"family": "selection",
"nv_evidence": "strong-positive-shared-network-risk",
"nv_gain_percent": 20.8,
"prior_s5000": "B 0->0 at large author shape",
"cost": "medium",
"canary": true
},
{
"id": "instance_norm",
"repo": "main",
"op": "instance_norm",
"op_file": "instance_norm",
"shape_key": "instance_norm",
"correctness_file": "tests/test_instance_norm.py",
"family": "normalization",
"nv_evidence": "strong-positive",
"nv_gain_percent": 17.8,
"prior_s5000": "G 1->0",
"cost": "medium",
"canary": true
},
{
"id": "dropout",
"repo": "main",
"op": "dropout",
"op_file": "dropout",
"shape_key": "dropout",
"correctness_file": "tests/test_dropout.py",
"family": "rng",
"nv_evidence": "strong-positive-shared-network-risk",
"nv_gain_percent": 15.2,
"prior_s5000": "product G 8->0 +9.15%",
"cost": "light",
"canary": true
},
{
"id": "router_gemm",
"repo": "main",
"op": "perf_router_gemm",
"op_file": "router_gemm",
"shape_key": "router_gemm",
"correctness_file": "tests/test_router_gemm.py",
"family": "mma",
"nv_evidence": "strong-positive-mma",
"nv_gain_percent": 14.5,
"prior_s5000": "public screen available; product v1 pending",
"cost": "heavy",
"canary": false,
"benchmark_dtypes": [
"bfloat16"
]
},
{
"id": "fused_recurrent_gated_delta_rule",
"repo": "main",
"op": "perf_fused_recurrent_gated_delta_rule",
"op_file": "FLA/test_fused_recurrent_gated_delta_rule_perf",
"shape_key": null,
"correctness_file": "tests/test_FLA/test_fused_recurrent_gated_delta_rule.py",
"family": "recurrent",
"nv_evidence": "strong-positive-recurrent",
"nv_gain_percent": 14.2,
"prior_s5000": "not product-qualified",
"cost": "heavy",
"canary": false,
"benchmark_dtypes": [
"bfloat16"
]
},
{
"id": "top_k_per_row_decode",
"repo": "main",
"op": "top_k_per_row_decode",
"op_file": "top_k_per_row_decode",
"shape_key": "top_k_per_row_decode",
"correctness_file": "tests/test_top_k_per_row_decode.py",
"family": "selection",
"nv_evidence": "strong-positive-shared-network-risk",
"nv_gain_percent": 13.0,
"prior_s5000": "not product-qualified",
"cost": "medium",
"canary": false
},
{
"id": "sort",
"repo": "main",
"op": "perf_sort",
"op_file": "sort",
"shape_key": "sort",
"correctness_file": "tests/test_sort.py",
"family": "sort",
"nv_evidence": "strong-positive-shared-network-risk",
"nv_gain_percent": 12.3,
"prior_s5000": "G 2->1",
"cost": "medium",
"canary": true
},
{
"id": "sort_stable",
"repo": "main",
"op": "perf_sort_stable",
"op_file": "sort",
"shape_key": "sort_stable",
"correctness_file": "tests/test_sort.py",
"family": "sort",
"nv_evidence": "strong-positive-shared-network-risk",
"nv_gain_percent": 12.3,
"prior_s5000": "G 2->1",
"cost": "medium",
"canary": false
},
{
"id": "uniform",
"repo": "main",
"op": "uniform_",
"op_file": "uniform_",
"shape_key": "uniform_",
"correctness_file": "tests/test_uniform_.py",
"family": "rng",
"nv_evidence": "strong-positive-rng",
"nv_gain_percent": 10.3,
"prior_s5000": "public boundary risk; product v1 pending",
"cost": "light",
"canary": false
},
{
"id": "rand",
"repo": "main",
"op": "rand",
"op_file": "rand",
"shape_key": "rand",
"correctness_file": "tests/test_rand.py",
"family": "rng",
"nv_evidence": "strong-positive-rng",
"nv_gain_percent": 8.7,
"prior_s5000": "product v1 pending",
"cost": "light",
"canary": false
},
{
"id": "rand_like",
"repo": "main",
"op": "rand_like",
"op_file": "rand_like",
"shape_key": "rand_like",
"correctness_file": "tests/test_rand_like.py",
"family": "rng",
"nv_evidence": "strong-positive-rng",
"nv_gain_percent": 8.6,
"prior_s5000": "product v1 pending",
"cost": "light",
"canary": false
},
{
"id": "max_pool2d_with_indices",
"repo": "main",
"op": "max_pool2d_with_indices",
"op_file": "max_pool2d",
"shape_key": "max_pool2d_with_indices",
"correctness_file": "tests/test_max_pool2d.py",
"family": "reduction-index",
"nv_evidence": "positive-reduction-index",
"nv_gain_percent": 6.1,
"prior_s5000": "B 0->0",
"cost": "medium",
"canary": false
},
{
"id": "randperm",
"repo": "main",
"op": "randperm",
"op_file": "randperm",
"shape_key": "randperm",
"correctness_file": "tests/test_randperm.py",
"family": "rng",
"nv_evidence": "small-positive-rng",
"nv_gain_percent": 4.0,
"prior_s5000": "C 7->1 latency overlap",
"cost": "medium",
"canary": false,
"benchmark_dtypes": [
"int32"
]
},
{
"id": "randn",
"repo": "main",
"op": "randn",
"op_file": "randn",
"shape_key": "randn",
"correctness_file": "tests/test_randn.py",
"family": "rng",
"nv_evidence": "diluted-positive-rng",
"nv_gain_percent": 4.0,
"prior_s5000": "product v1 pending",
"cost": "light",
"canary": false
},
{
"id": "randn_like",
"repo": "main",
"op": "randn_like",
"op_file": "randn_like",
"shape_key": "randn_like",
"correctness_file": "tests/test_randn_like.py",
"family": "rng",
"nv_evidence": "diluted-positive-rng",
"nv_gain_percent": 4.0,
"prior_s5000": "product v1 pending",
"cost": "light",
"canary": false
},
{
"id": "upsample_bicubic2d_aa",
"repo": "main",
"op": "upsample_bicubic2d_aa",
"op_file": "upsample_bicubic2d_aa",
"shape_key": "upsample_bicubic2d_aa",
"correctness_file": "tests/test_upsample_bicubic2d_aa.py",
"family": "interpolation",
"nv_evidence": "diluted-positive-cold-path",
"nv_gain_percent": 1.7,
"prior_s5000": "C 10->0 latency flat",
"cost": "medium",
"canary": false,
"correctness_selector": "not backward",
"correctness_scope_reason": "the benchmarked op is the forward path (benchmark/test_upsample_bicubic2d_aa.py::test_upsample_bicubic2d_aa); the backward baseline defect fails identically under off/mask0/candidate and is tracked as a separate MUSA numerics issue"
},
{
"id": "grid_sample",
"repo": "main",
"op": "grid_sample",
"op_file": "grid_sample",
"shape_key": "grid_sample",
"correctness_file": "tests/test_grid_sample.py",
"family": "sampling",
"nv_evidence": "structure-change-no-win",
"nv_gain_percent": 0.0,
"prior_s5000": "B 0->0",
"cost": "medium",
"canary": false,
"benchmark_selector": "dtype0",
"benchmark_scope_reason": "benchmark/test_grid_sample.py parametrises over consts.FLOAT_DTYPES and declares dtypes=[dtype] per case, so a campaign-wide --dtypes float16 filter makes the float32 and bfloat16 cases raise; dtype0 is the float16 case this workload measures"
},
{
"id": "max_pool2d_backward",
"repo": "main",
"op": "max_pool2d_backward",
"op_file": "max_pool2d",
"shape_key": "max_pool2d_backward",
"correctness_file": "tests/test_max_pool2d.py",
"family": "reduction-index",
"nv_evidence": "structure-change-no-win",
"nv_gain_percent": 0.0,
"prior_s5000": "B 0->0",
"cost": "medium",
"canary": false
},
{
"id": "conv1d",
"repo": "main",
"op": "conv1d",
"op_file": "conv1d",
"shape_key": "conv1d",
"correctness_file": "tests/test_conv1d.py",
"family": "convolution",
"nv_evidence": "light-regression-review",
"nv_gain_percent": -4.2,
"prior_s5000": "C 1->0 via phase3, slight slowdown",
"cost": "heavy",
"canary": true
},
{
"id": "layer_norm_backward",
"repo": "main",
"op": "layer_norm_backward",
"op_file": "layer_norm",
"shape_key": "layer_norm_backward",
"correctness_file": "tests/test_layer_norm.py",
"family": "normalization",
"nv_evidence": "light-regression-review",
"nv_gain_percent": -1.9,
"prior_s5000": "C 4->2 latency flat",
"cost": "medium",
"canary": false
},
{
"id": "cutlass_scaled_mm",
"repo": "main",
"op": "cutlass_scaled_mm_benchmark",
"op_file": "cutlass_scaled_mm",
"shape_key": "cutlass_scaled_mm",
"correctness_file": "tests/test_cutlass_scaled_mm.py",
"family": "mma",
"nv_evidence": "light-regression-mma",
"nv_gain_percent": -1.2,
"prior_s5000": "likely unsupported CUDA-specific control",
"cost": "heavy",
"canary": false
},
{
"id": "gelu_backward",
"repo": "main",
"op": "gelu_backward",
"op_file": "gelu",
"shape_key": "gelu_backward",
"correctness_file": "tests/test_gelu.py",
"family": "elementwise-backward",
"nv_evidence": "direct-no-cudagraph-noise",
"nv_gain_percent": null,
"prior_s5000": "B 0->0 with LLVM optimizations",
"cost": "medium",
"canary": false
},
{
"id": "fused_marlin_moe_mxfp4",
"repo": "main",
"op": "fused_marlin_moe_w4a16_mxfp4",
"op_file": "fused_marlin_moe_w4a16_mxfp4",
"shape_key": null,
"correctness_file": "tests/test_fused_marlin_moe.py",
"family": "quantized-moe",
"nv_evidence": "autotune-config-change",
"nv_gain_percent": null,
"prior_s5000": "likely unsupported CUDA-specific control",
"cost": "heavy",
"canary": false,
"correctness_selector": "mxfp4",
"correctness_scope_reason": "tests/test_fused_marlin_moe.py also holds w4a16_int4 and w8a16_int8 variants; only the mxfp4 variant corresponds to this workload, and the int8 failures are a separate baseline defect",
"inventory_override_reason": "zl_bench da2288564584039366e7be07fe4e94539a560c81 lists benchmark/test_fused_marlin_moe.py, but FlagGems 46ef8080329190d8da8738340240de2d5a475950 split it into per-quantisation files; the mxfp4 target is benchmark/test_fused_marlin_moe_w4a16_mxfp4.py::test_fused_marlin_moe_w4a16_mxfp4"
},
{
"id": "bincount",
"repo": "main",
"op": "bincount",
"op_file": "bincount",
"shape_key": "bincount",
"correctness_file": "tests/test_bincount.py",
"family": "histogram",
"nv_evidence": "autotune-config-change",
"nv_gain_percent": null,
"prior_s5000": "product v1 pending",
"cost": "medium",
"canary": false,
"benchmark_dtypes": [
"float32"
]
},
{
"id": "masked_select",
"repo": "main",
"op": "masked_select",
"op_file": "masked_select",
"shape_key": "masked_select",
"correctness_file": "tests/test_masked_select.py",
"family": "selection",
"nv_evidence": "holdout-shared-network",
"nv_gain_percent": null,
"prior_s5000": "public qualify available",
"cost": "medium",
"canary": false
},
{
"id": "mm",
"repo": "main",
"op": "mm",
"op_file": "mm",
"shape_key": "mm",
"correctness_file": "tests/test_mm.py",
"family": "mma-control",
"nv_evidence": "holdout-mma-control",
"nv_gain_percent": null,
"prior_s5000": "control only",
"cost": "heavy",
"canary": false
},
{
"id": "scaled_dot_product_attention",
"repo": "main",
"op": "scaled_dot_product_attention",
"op_file": "scaled_dot_product_attention",
"shape_key": "scaled_dot_product_attention",
"correctness_file": "tests/test_scaled_dot_product_attention.py",
"family": "attention-control",
"nv_evidence": "holdout-attention-control",
"nv_gain_percent": null,
"prior_s5000": "control only",
"cost": "heavy",
"canary": false
},
{
"id": "softmax",
"repo": "main",
"op": "softmax",
"op_file": "softmax",
"shape_key": "softmax",
"correctness_file": "tests/test_softmax.py",
"family": "reduction-control",
"nv_evidence": "holdout-reduction-control",
"nv_gain_percent": null,
"prior_s5000": "product B 0->0",
"cost": "light",
"canary": true
},
{
"id": "layer_norm",
"repo": "main",
"op": "layer_norm",
"op_file": "layer_norm",
"shape_key": "layer_norm",
"correctness_file": "tests/test_layer_norm.py",
"family": "normalization-control",
"nv_evidence": "holdout-normalization-control",
"nv_gain_percent": null,
"prior_s5000": "control only",
"cost": "medium",
"canary": false
}
],
"predecessor_campaign_id": "s5000-musa-rlc-32-product-v1-20260901-v2"
}
@@ -0,0 +1,478 @@
#!/usr/bin/env python3
"""Validate and freeze the single-device MUSA RLC product campaign."""
from __future__ import annotations
import argparse
import csv
import hashlib
import json
import os
import shutil
from pathlib import Path
from typing import Any
COSTS = {"light", "medium", "heavy"}
CELLS = ["off", "mask0", "candidate"]
TIMING_CELLS = ["off", "candidate"]
ESTIMATED_SECONDS = {"light": 180, "medium": 600, "heavy": 1200}
class CampaignError(RuntimeError):
pass
def _canonical(value: object) -> bytes:
return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
def _is_git_sha(value: object) -> bool:
return (
isinstance(value, str)
and len(value) == 40
and all(char in "0123456789abcdef" for char in value)
)
def _correctness_scope_errors(row: dict[str, Any], index: int) -> list[str]:
"""Check an optional narrowing of a workload's correctness gate.
Running a whole test file lets an unrelated variant decide the verdict for
the operator actually being measured. A workload may therefore pin node ids
or a ``-k`` selector, but never both, and narrowing always requires a stated
reason so the exclusion stays visible in review.
"""
name = row.get("id", index)
selector = row.get("correctness_selector")
node_ids = row.get("correctness_node_ids")
errors: list[str] = []
if selector is not None and node_ids is not None:
errors.append(
f"workload {name} must not set both correctness_selector and "
"correctness_node_ids"
)
if selector is not None and (not isinstance(selector, str) or not selector.strip()):
errors.append(f"workload {name} correctness_selector must be a non-empty string")
if node_ids is not None and (
not isinstance(node_ids, list)
or not node_ids
or not all(isinstance(value, str) and value.strip() for value in node_ids)
):
errors.append(
f"workload {name} correctness_node_ids must be non-empty strings"
)
if selector is None and node_ids is None:
return errors
reason = row.get("correctness_scope_reason")
if not isinstance(reason, str) or not reason.strip():
errors.append(
f"workload {name} narrows correctness and must state "
"correctness_scope_reason"
)
return errors
def _benchmark_scope_errors(row: dict[str, Any], index: int) -> list[str]:
"""Check an optional narrowing of a workload's benchmark run.
Some benchmarks parametrise one dtype per case, so the campaign-wide dtype
filter makes every non-matching case raise. A workload may pin the case it
measures, but narrowing what gets timed always requires a stated reason.
"""
name = row.get("id", index)
selector = row.get("benchmark_selector")
if selector is None:
return []
errors: list[str] = []
if not isinstance(selector, str) or not selector.strip():
errors.append(f"workload {name} benchmark_selector must be a non-empty string")
reason = row.get("benchmark_scope_reason")
if not isinstance(reason, str) or not reason.strip():
errors.append(
f"workload {name} narrows the benchmark and must state "
"benchmark_scope_reason"
)
return errors
def _inventory_override_errors(row: dict[str, Any], index: int) -> list[str]:
"""Check an explicit departure from the pinned zl_bench inventory.
The inventory can lag the pinned FlagGems commit, for example when upstream
splits one benchmark file into several. A workload may then name the target
that actually exists, but it must say why, so a stale inventory is never
silently routed around.
"""
name = row.get("id", index)
reason = row.get("inventory_override_reason")
if reason is None:
return []
if not isinstance(reason, str) or not reason.strip():
return [
f"workload {name} inventory_override_reason must be a non-empty string"
]
return []
def validate_manifest(manifest: dict[str, Any]) -> list[str]:
errors: list[str] = []
if manifest.get("schema_version") != 1:
errors.append("schema_version must be 1")
if manifest.get("experiment_kind") != "musa-rlc-product-validation":
errors.append("experiment_kind must be musa-rlc-product-validation")
if manifest.get("production_default_policy") != "unchanged-off":
errors.append("production_default_policy must be unchanged-off")
if "S5000 MUSA product-dispatch" not in manifest.get("permitted_claim", ""):
errors.append("permitted_claim must bind S5000 MUSA product dispatch")
source = manifest.get("source")
if not isinstance(source, dict):
errors.append("source must be an object")
else:
for key in ("flagtree_commit", "flaggems_commit", "zl_bench_commit"):
if not _is_git_sha(source.get(key)):
errors.append(f"source.{key} must be a lowercase 40-character Git SHA")
device = manifest.get("device")
expected_device = {
"ssh_alias": "S5000",
"container": "zixu_flagos_S5000",
"backend": "mthreads",
"model": "Moore Threads S5000",
"physical_device": "7",
"logical_device": "0",
"visible_device_count": 1,
}
if not isinstance(device, dict):
errors.append("device must be an object")
else:
for key, expected in expected_device.items():
if device.get(key) != expected:
errors.append(f"device.{key} must equal {expected!r}")
protocol = manifest.get("protocol")
if not isinstance(protocol, dict):
errors.append("protocol must be an object")
else:
if protocol.get("screen_cells") != CELLS:
errors.append("protocol.screen_cells must be off/mask0/candidate")
if protocol.get("timing_cells") != TIMING_CELLS:
errors.append("protocol.timing_cells must be off/candidate")
if protocol.get("candidate_phase_mask") != 13:
errors.append("candidate_phase_mask must be 13")
canary_order = protocol.get("canary_order")
if not isinstance(canary_order, list) or len(canary_order) != 8:
errors.append("protocol.canary_order must contain eight workload ids")
for flag in (
"candidate_profitability_policy",
"fresh_process_per_operator_cell",
"same_physical_device_for_all_cells",
"serial_measurement",
"full_campaign_requires_canary_terminal",
):
if protocol.get(flag) is not True:
errors.append(f"protocol.{flag} must be true")
if protocol.get("timer") != "triton.testing.do_bench via MUSA device-interface Event":
errors.append("protocol.timer must bind the MUSA native-event path")
default_dtypes = protocol.get("default_benchmark_dtypes")
if not isinstance(default_dtypes, list) or not default_dtypes or not all(
isinstance(value, str) and value for value in default_dtypes
):
errors.append("protocol.default_benchmark_dtypes must be non-empty strings")
supported = protocol.get("supported_benchmark_dtypes")
if not isinstance(supported, list) or not supported or not all(
isinstance(value, str) and value for value in supported
):
errors.append(
"protocol.supported_benchmark_dtypes must be non-empty strings"
)
timeouts = protocol.get("timeout_seconds_by_cost")
if not isinstance(timeouts, dict) or set(timeouts) != COSTS:
errors.append("timeout_seconds_by_cost must define light/medium/heavy")
elif not all(isinstance(value, int) and value > 0 for value in timeouts.values()):
errors.append("timeout_seconds_by_cost values must be positive integers")
floor = protocol.get("measured_launch_floor")
if not isinstance(floor, int) or isinstance(floor, bool) or floor < 0:
errors.append(
"protocol.measured_launch_floor must be a non-negative integer"
)
workloads = manifest.get("workloads")
if not isinstance(workloads, list) or len(workloads) != 32:
errors.append("workloads must contain exactly 32 entries")
return errors
ids: list[str] = []
nodes: list[tuple[str, str, str]] = []
canaries = 0
for index, row in enumerate(workloads):
if not isinstance(row, dict):
errors.append(f"workloads[{index}] must be an object")
continue
for key in (
"id", "repo", "op", "op_file", "correctness_file", "family",
"nv_evidence", "prior_s5000", "cost",
):
if not isinstance(row.get(key), str) or not row[key]:
errors.append(f"workloads[{index}] missing {key}")
if row.get("repo") != "main":
errors.append(f"workload {row.get('id', index)} must use the main FlagGems repo")
if row.get("cost") not in COSTS:
errors.append(f"workload {row.get('id', index)} has invalid cost")
if not isinstance(row.get("canary"), bool):
errors.append(f"workload {row.get('id', index)} canary must be boolean")
benchmark_dtypes = row.get(
"benchmark_dtypes", protocol.get("default_benchmark_dtypes", [])
)
if not isinstance(benchmark_dtypes, list) or not benchmark_dtypes or not all(
isinstance(value, str) and value for value in benchmark_dtypes
):
errors.append(
f"workload {row.get('id', index)} benchmark_dtypes must be non-empty strings"
)
else:
# The benchmark CLI rejects a dtype it does not define, and that
# only surfaced as an argparse error inside a device run.
vocabulary = protocol.get("supported_benchmark_dtypes")
if isinstance(vocabulary, list):
unknown = [
value for value in benchmark_dtypes if value not in vocabulary
]
if unknown:
errors.append(
f"workload {row.get('id', index)} declares benchmark dtype(s) "
f"the benchmark CLI does not accept: {', '.join(unknown)}"
)
errors.extend(_correctness_scope_errors(row, index))
errors.extend(_benchmark_scope_errors(row, index))
errors.extend(_inventory_override_errors(row, index))
canaries += row.get("canary") is True
ids.append(row.get("id", ""))
nodes.append((row.get("repo", ""), row.get("op", ""), row.get("op_file", "")))
if len(ids) != len(set(ids)):
errors.append("workload ids must be unique")
if len(nodes) != len(set(nodes)):
errors.append("workload repo/op/op_file triples must be unique")
if canaries != 8:
errors.append("exactly eight workloads must be canaries")
elif set(manifest["protocol"].get("canary_order", [])) != {
row["id"] for row in workloads if row["canary"]
}:
errors.append("protocol.canary_order must contain every canary exactly once")
return errors
def _load_inventory(inventory_dir: Path) -> dict[tuple[str, str, str], dict[str, str]]:
path = inventory_dir / "inventory_main.csv"
if not path.is_file():
raise CampaignError(f"missing inventory: {path}")
rows: dict[tuple[str, str, str], dict[str, str]] = {}
with path.open(newline="", encoding="utf-8") as handle:
for row in csv.DictReader(handle):
key = ("main", row.get("op", ""), row.get("op_file", ""))
if key in rows:
raise CampaignError(f"duplicate inventory row: {key}")
rows[key] = row
return rows
def validate_inventory(workloads: list[dict[str, Any]], inventory_dir: Path) -> list[str]:
inventory = _load_inventory(inventory_dir)
errors: list[str] = []
for row in workloads:
key = (row["repo"], row["op"], row["op_file"])
found = inventory.get(key)
if found is None:
# A stated override admits a target the pinned inventory does not
# know about. The runner still resolves it against the real tree.
if row.get("inventory_override_reason"):
continue
errors.append(f"inventory missing {key}")
continue
if found.get("collected") == "no":
errors.append(f"inventory marks {key} collected=no")
if found.get("marked_skip") == "yes":
errors.append(f"inventory marks {key} skipped")
return errors
def build_plan(manifest: dict[str, Any], selection: str) -> tuple[dict[str, Any], dict[str, Any]]:
workloads = manifest["workloads"]
if selection == "canary":
by_id = {row["id"]: row for row in workloads if row["canary"]}
workloads = [by_id[task_id] for task_id in manifest["protocol"]["canary_order"]]
elif selection != "all":
raise CampaignError(f"unsupported selection: {selection}")
timeouts = manifest["protocol"]["timeout_seconds_by_cost"]
tasks = []
ordinal = 0
for row in workloads:
ordinal += 1
tasks.append(
{
"ordinal": ordinal,
"task_id": row["id"],
"repo": row["repo"],
"op": row["op"],
"op_file": row["op_file"],
"shape_key": row.get("shape_key"),
"correctness_file": row["correctness_file"],
"correctness_selector": row.get("correctness_selector"),
"correctness_node_ids": row.get("correctness_node_ids"),
"correctness_scope_reason": row.get("correctness_scope_reason"),
"benchmark_selector": row.get("benchmark_selector"),
"benchmark_scope_reason": row.get("benchmark_scope_reason"),
"inventory_override_reason": row.get("inventory_override_reason"),
"measured_launch_floor": manifest["protocol"]["measured_launch_floor"],
"family": row["family"],
"nv_evidence": row["nv_evidence"],
"nv_gain_percent": row.get("nv_gain_percent"),
"prior_s5000": row["prior_s5000"],
"cost": row["cost"],
"benchmark_dtypes": row.get(
"benchmark_dtypes",
manifest["protocol"]["default_benchmark_dtypes"],
),
"timeout_seconds": timeouts[row["cost"]],
"estimated_seconds_per_cell": ESTIMATED_SECONDS[row["cost"]],
"cells": CELLS,
"status": "PENDING",
}
)
estimate = sum(task["estimated_seconds_per_cell"] * len(CELLS) for task in tasks)
matrix = {
"schema_version": 1,
"campaign_id": manifest["campaign_id"],
"selection": selection,
"device": manifest["device"],
"screen_cells": CELLS,
"timing_cells": TIMING_CELLS,
"tasks": tasks,
}
budget = {
"schema_version": 1,
"campaign_id": manifest["campaign_id"],
"selection": selection,
"task_count": len(tasks),
"gpu_count": 1,
"serial_measurement": True,
"estimated_wall_seconds": estimate,
"estimated_gpu_hours": round(estimate / 3600.0, 4),
"note": "Scheduling estimate only; it is not performance evidence.",
}
return matrix, budget
def validate_canary_receipt(
manifest: dict[str, Any], selection: str, canary_receipt: Path | None
) -> list[str]:
if selection != "all":
return []
if canary_receipt is None:
return ["full campaign requires --canary-receipt"]
try:
receipt = json.loads(canary_receipt.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
return [f"cannot read canary receipt: {exc}"]
errors = []
if receipt.get("campaign_id") != manifest.get("campaign_id"):
errors.append("canary receipt campaign_id mismatch")
if receipt.get("selection") != "canary" or receipt.get("status") != "TERMINAL":
errors.append("canary receipt must be a terminal canary")
if receipt.get("canary_gate_pass") is not True:
errors.append("canary receipt does not authorize the full campaign")
return errors
def prepare(
manifest_path: Path,
inventory_dir: Path,
output_root: Path,
selection: str,
canary_receipt: Path | None = None,
) -> Path:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
if not isinstance(manifest, dict):
raise CampaignError("manifest root must be an object")
errors = validate_manifest(manifest)
if not errors:
errors.extend(validate_inventory(manifest["workloads"], inventory_dir))
if not errors:
errors.extend(validate_canary_receipt(manifest, selection, canary_receipt))
if errors:
raise CampaignError("campaign validation failed:\n- " + "\n- ".join(errors))
matrix, budget = build_plan(manifest, selection)
output = output_root.resolve() / f"{manifest['campaign_id']}--{selection}"
if output.exists():
raise CampaignError(f"campaign output already exists and is immutable: {output}")
output.parent.mkdir(parents=True, exist_ok=True)
stage = output.parent / f".{output.name}.staging-{os.getpid()}"
if stage.exists():
raise CampaignError(f"staging directory already exists: {stage}")
try:
stage.mkdir()
shutil.copy2(manifest_path, stage / "manifest.json")
manifest_sha = hashlib.sha256((stage / "manifest.json").read_bytes()).hexdigest()
matrix["manifest_sha256"] = manifest_sha
budget["manifest_sha256"] = manifest_sha
(stage / "manifest.sha256").write_text(
f"{manifest_sha} manifest.json\n", encoding="utf-8", newline="\n"
)
(stage / "task-matrix.json").write_bytes(_canonical(matrix))
(stage / "budget.json").write_bytes(_canonical(budget))
(stage / "ops.txt").write_text(
"".join(f"{task['repo']}:{task['op']}@{task['op_file']}\n" for task in matrix["tasks"]),
encoding="utf-8",
newline="\n",
)
stage.replace(output)
except Exception:
if stage.exists():
shutil.rmtree(stage)
raise
return output
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("manifest", type=Path)
parser.add_argument("--inventory-dir", type=Path, required=True)
parser.add_argument("--selection", choices=("canary", "all"), default="canary")
parser.add_argument("--output-root", type=Path)
parser.add_argument("--canary-receipt", type=Path)
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args()
try:
manifest = json.loads(args.manifest.read_text(encoding="utf-8"))
errors = validate_manifest(manifest) if isinstance(manifest, dict) else ["manifest root must be an object"]
if not errors:
errors.extend(validate_inventory(manifest["workloads"], args.inventory_dir))
if not errors and not args.dry_run:
errors.extend(validate_canary_receipt(manifest, args.selection, args.canary_receipt))
if errors:
raise CampaignError("campaign validation failed:\n- " + "\n- ".join(errors))
matrix, budget = build_plan(manifest, args.selection)
if args.dry_run:
print(json.dumps({"status": "DRY_RUN", "budget": budget, "tasks": matrix["tasks"]}, indent=2, sort_keys=True))
return 0
if args.output_root is None:
raise CampaignError("--output-root is required unless --dry-run is used")
output = prepare(
args.manifest,
args.inventory_dir,
args.output_root,
args.selection,
args.canary_receipt,
)
except (OSError, json.JSONDecodeError, CampaignError) as exc:
print(str(exc))
return 2
print(json.dumps({"status": "PREPARED", "output": str(output)}, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,223 @@
#!/usr/bin/env python3
"""Re-account a sealed MUSA RLC run under measured-operator convert scoping.
The sealed campaign counted every executed kernel's ``ttg.convert_layout`` ops,
including the kernels a benchmark launches once to build its inputs. Those sit
outside the timed region, so a candidate could be credited with a topology
reduction that no timer could ever observe.
This tool re-reads the sealed TTGIR dumps and their launch index, recomputes the
convert counts scoped to repeatedly launched kernels, and re-runs the campaign
classifier. It never writes inside the sealed run.
A re-accounted verdict can only demote a topology claim. Promoting a task to G
or C needs reversed-order timing, which a sealed run only holds for tasks that
already showed a reduction, so a newly appearing reduction is reported as X for
missing round-b evidence rather than guessed.
"""
from __future__ import annotations
import argparse
import csv
import importlib.util
import json
import sys
from pathlib import Path
from typing import Any
HERE = Path(__file__).resolve().parent
class ReaccountError(RuntimeError):
pass
def _load_runner():
spec = importlib.util.spec_from_file_location(
"run_musa_rlc_product", HERE / "run_musa_rlc_product.py"
)
if spec is None or spec.loader is None:
raise ReaccountError("cannot load run_musa_rlc_product.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _canonical(value: object) -> bytes:
return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
def reaccount_task(
runner,
task_root: Path,
receipt: dict[str, Any],
measured_launch_floor: int,
) -> dict[str, Any]:
"""Recompute one sealed task receipt under the scoped convert count."""
cells: dict[str, dict[str, Any]] = {}
for key, sealed in receipt.get("cells", {}).items():
cell = dict(sealed)
cell_dir = task_root / str(sealed.get("round")) / str(sealed.get("cell"))
if cell_dir.is_dir():
scope = runner._convert_count(
cell_dir, measured_launch_floor=measured_launch_floor
)
cell.update(
{
"convert_count": scope["measured_convert_count"],
"convert_count_all": scope["total_convert_count"],
"convert_scope_complete": scope["scope_complete"],
"measured_launch_floor": scope["measured_launch_floor"],
"convert_excluded_unmeasured": scope["excluded_unmeasured"],
"convert_unindexed_files": scope["unindexed_files"],
}
)
else:
cell["convert_scope_complete"] = False
cells[key] = cell
classification, detail = runner.classify_task(
receipt["correctness"], cells, receipt.get("correctness_isolation")
)
sealed_class = receipt.get("classification")
excluded = sorted(
{
row["file"]
for cell in cells.values()
for row in cell.get("convert_excluded_unmeasured", [])
}
)
return {
"task_id": receipt["task_id"],
"op": receipt.get("op"),
"family": receipt.get("family"),
"sealed_classification": sealed_class,
"reaccounted_classification": classification,
"changed": classification != sealed_class,
"sealed_reason": receipt.get("classification_detail", {}).get("reason"),
"reaccounted_reason": detail["reason"],
"sealed_convert": {
key: receipt["cells"][key].get("convert_count")
for key in sorted(receipt.get("cells", {}))
},
"reaccounted_convert": {
key: cells[key].get("convert_count") for key in sorted(cells)
},
"excluded_unmeasured_files": excluded,
"classification_detail": detail,
}
def reaccount(
run_root: Path, output_root: Path, measured_launch_floor: int
) -> dict[str, Any]:
run_root = run_root.resolve()
output_root = output_root.resolve()
if not (run_root / "campaign-receipt.json").is_file():
raise ReaccountError(f"not a sealed campaign run root: {run_root}")
if output_root == run_root or run_root in output_root.parents:
raise ReaccountError("refusing to write inside the sealed run root")
runner = _load_runner()
sealed = json.loads((run_root / "campaign-receipt.json").read_text(encoding="utf-8"))
rows: list[dict[str, Any]] = []
for path in sorted((run_root / "tasks").glob("*/task-receipt.json")):
receipt = json.loads(path.read_text(encoding="utf-8"))
rows.append(
reaccount_task(runner, path.parent, receipt, measured_launch_floor)
)
terminal = sorted(runner.TERMINAL)
result = {
"schema_version": 1,
"kind": "musa-rlc-topology-reaccount",
"sealed_campaign_id": sealed.get("campaign_id"),
"sealed_run_root": str(run_root),
"measured_launch_floor": measured_launch_floor,
"task_count": len(rows),
"sealed_counts": {
key: sum(row["sealed_classification"] == key for row in rows)
for key in terminal
},
"reaccounted_counts": {
key: sum(row["reaccounted_classification"] == key for row in rows)
for key in terminal
},
"changed_task_ids": [row["task_id"] for row in rows if row["changed"]],
"note": (
"Derived accounting over sealed artifacts. It supersedes no receipt "
"and produces no new device measurement."
),
"tasks": rows,
}
output_root.mkdir(parents=True, exist_ok=True)
(output_root / "reaccount-receipt.json").write_bytes(_canonical(result))
with (output_root / "reaccount-matrix.csv").open(
"w", newline="", encoding="utf-8"
) as handle:
writer = csv.writer(handle)
writer.writerow(
[
"task_id",
"sealed_class",
"reaccounted_class",
"changed",
"sealed_off",
"sealed_candidate",
"scoped_off",
"scoped_candidate",
"reaccounted_reason",
]
)
for row in rows:
writer.writerow(
[
row["task_id"],
row["sealed_classification"],
row["reaccounted_classification"],
"yes" if row["changed"] else "no",
row["sealed_convert"].get("a_off"),
row["sealed_convert"].get("a_candidate"),
row["reaccounted_convert"].get("a_off"),
row["reaccounted_convert"].get("a_candidate"),
row["reaccounted_reason"],
]
)
return result
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--run-root", type=Path, required=True)
parser.add_argument("--output-root", type=Path, required=True)
parser.add_argument("--measured-launch-floor", type=int, default=1)
args = parser.parse_args()
try:
if args.measured_launch_floor < 0:
raise ReaccountError("--measured-launch-floor must be non-negative")
result = reaccount(
args.run_root, args.output_root, args.measured_launch_floor
)
except (OSError, json.JSONDecodeError, ReaccountError) as exc:
print(f"MUSA_RLC_REACCOUNT_ERROR: {exc}", file=sys.stderr)
return 2
summary = {
key: result[key]
for key in (
"sealed_campaign_id",
"measured_launch_floor",
"task_count",
"sealed_counts",
"reaccounted_counts",
"changed_task_ids",
)
}
print(json.dumps(summary, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,834 @@
#!/usr/bin/env python3
"""Run the frozen S5000 MUSA RLC product campaign serially."""
from __future__ import annotations
import argparse
import csv
import hashlib
import importlib.util
import json
import math
import os
import signal
import statistics
import subprocess
import sys
import time
from pathlib import Path
from typing import Any
TERMINAL = {"G", "C", "B", "R", "K", "X"}
RLC_ENV_KEYS = {
"FLAGTREE_RLC_ENHANCE",
"FLAGTREE_RLC_PHASE_MASK",
"FLAGTREE_MUSA_RLC_ENHANCE",
"FLAGTREE_MUSA_RLC_PHASE_MASK",
"FLAGTREE_MUSA_RLC_PROFITABILITY_POLICY",
"FLAGTREE_MUSA_RLC_PRODUCT_LAUNCH_COUNT",
"FLAGTREE_MUSA_RLC_MIN_ADJUSTED_SAVED_COST_PER_TENSOR_OP",
"FLAGTREE_MUSA_RLC_PHASE3_SAVED_COST_MULTIPLIER",
"FLAGTREE_MUSA_RLC_MAX_EXTERNAL_USE_EDGES",
"FLAGTREE_MUSA_RLC_MIN_REMOVED_CONVERT_DENSITY_PER_1024_PROPOSAL_VALUES",
"FLAGTREE_MUSA_RLC_PRESERVE_INT_TO_FP_CONTIGUITY",
"FLAGTREE_MUSA_RLC_PRESERVE_FP_TO_FP_CONTIGUITY",
}
class CampaignError(RuntimeError):
pass
def _canonical(value: object) -> bytes:
return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _atomic_json(path: Path, value: object) -> None:
temporary = path.with_name(path.name + ".tmp")
temporary.write_bytes(_canonical(value))
os.replace(temporary, path)
def _load_module(path: Path, name: str):
spec = importlib.util.spec_from_file_location(name, path)
if spec is None or spec.loader is None:
raise CampaignError(f"cannot load module: {path}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _identity(path: Path) -> dict[str, str]:
rows: dict[str, str] = {}
for raw in path.read_text(encoding="utf-8").splitlines():
key, separator, value = raw.partition("=")
if separator:
rows[key] = value
return rows
def _base_env(site: Path, flaggems: Path) -> dict[str, str]:
env = os.environ.copy()
for key in list(env):
if key.startswith("FLAGTREE_") and "RLC" in key:
env.pop(key)
env.pop("DISABLE_LLVM_OPT", None)
env.update(
{
"PYTHONNOUSERSITE": "1",
"PYTHONDONTWRITEBYTECODE": "1",
"PYTHONUNBUFFERED": "1",
"GEMS_VENDOR": "mthreads",
"FLAGGEMS_VENDOR": "mthreads",
"FLAGTREE_BACKEND": "mthreads",
"TRITON_BACKENDS_IN_TREE": "1",
"MUSA_VISIBLE_DEVICES": "7",
"MTHREADS_VISIBLE_DEVICES": "7",
"PYTHONPATH": f"{flaggems / 'src'}:{site}",
"FLAGTREE_RLC_TRACE_REJECTS": "1",
}
)
return env
def _cell_env(base: dict[str, str], cell: str) -> dict[str, str]:
env = base.copy()
configs = {
"off": {"enhance": "0", "mask": "13", "policy": "0", "launches": "0"},
"mask0": {"enhance": "1", "mask": "0", "policy": "1", "launches": "1"},
"candidate": {"enhance": "1", "mask": "13", "policy": "1", "launches": "1"},
}
try:
row = configs[cell]
except KeyError as exc:
raise CampaignError(f"unsupported cell: {cell}") from exc
env.update(
{
"FLAGTREE_MUSA_RLC_ENHANCE": row["enhance"],
"FLAGTREE_MUSA_RLC_PHASE_MASK": row["mask"],
"FLAGTREE_MUSA_RLC_PROFITABILITY_POLICY": row["policy"],
"FLAGTREE_MUSA_RLC_PRODUCT_LAUNCH_COUNT": row["launches"],
"FLAGTREE_MUSA_RLC_MIN_ADJUSTED_SAVED_COST_PER_TENSOR_OP": "1800",
"FLAGTREE_MUSA_RLC_PHASE3_SAVED_COST_MULTIPLIER": "2",
"FLAGTREE_MUSA_RLC_MAX_EXTERNAL_USE_EDGES": "8",
"FLAGTREE_MUSA_RLC_MIN_REMOVED_CONVERT_DENSITY_PER_1024_PROPOSAL_VALUES": "0",
"FLAGTREE_MUSA_RLC_PRESERVE_INT_TO_FP_CONTIGUITY": "1",
"FLAGTREE_MUSA_RLC_PRESERVE_FP_TO_FP_CONTIGUITY": "1",
}
)
return env
def _run_process(
command: list[str], *, cwd: Path, env: dict[str, str], log: Path, timeout: int
) -> tuple[int, bool, float]:
started = time.time()
with log.open("w", encoding="utf-8", newline="\n") as handle:
process = subprocess.Popen(
command,
cwd=str(cwd),
env=env,
stdout=handle,
stderr=subprocess.STDOUT,
text=True,
start_new_session=True,
)
timed_out = False
try:
return_code = process.wait(timeout=timeout)
except subprocess.TimeoutExpired:
timed_out = True
try:
os.killpg(process.pid, signal.SIGTERM)
process.wait(timeout=10)
except subprocess.TimeoutExpired:
os.killpg(process.pid, signal.SIGKILL)
process.wait()
return_code = 124
return return_code, timed_out, time.time() - started
def _indexed_launches(ttgir_root: Path) -> dict[Path, int] | None:
"""Map each dumped TTGIR file to its executed launch count.
Returns None when ``index.tsv`` is missing or unreadable, which means the
measured-operator scope cannot be established for this cell.
"""
index = ttgir_root / "index.tsv"
if not index.is_file():
return None
try:
lines = index.read_text(encoding="utf-8", errors="replace").splitlines()
except OSError:
return None
if not lines:
return None
header = lines[0].split("\t")
try:
shape_at = header.index("shape")
kernel_at = header.index("kernel")
file_at = header.index("file")
launch_at = header.index("launches")
except ValueError:
return None
width = max(shape_at, kernel_at, file_at, launch_at)
launches: dict[Path, int] = {}
for raw in lines[1:]:
fields = raw.split("\t")
if len(fields) <= width:
continue
name = fields[file_at].strip()
# Autotune losers are indexed for audit but never dumped; they carry a
# parenthesised placeholder instead of a file name.
if not name or name.startswith("("):
continue
try:
count = int(fields[launch_at])
except ValueError:
continue
path = ttgir_root / fields[shape_at] / fields[kernel_at] / name
launches[path] = max(launches.get(path, 0), count)
return launches
def _convert_count(path: Path, *, measured_launch_floor: int = 1) -> dict[str, Any]:
"""Count executed convert_layout ops, scoped to the measured operator.
A benchmark call constructs its inputs once and then runs the operator many
times under the timer. Counting every executed kernel therefore credits the
candidate with converts removed from input construction, outside the timed
region. Only variants launched more than ``measured_launch_floor`` times
contribute to the measured count; the unscoped total is preserved too.
"""
ttgir_root = path / "ttgir"
files = sorted(ttgir_root.glob("**/*.ttgir"))
launches = _indexed_launches(ttgir_root)
total = 0
measured = 0
excluded: list[dict[str, Any]] = []
unindexed: list[str] = []
for file in files:
converts = file.read_text(encoding="utf-8", errors="replace").count(
"ttg.convert_layout"
)
total += converts
relative = file.relative_to(ttgir_root).as_posix()
if launches is None:
measured += converts
continue
launch_count = launches.get(file)
if launch_count is None:
# Dumped but absent from the index: keep it and report the gap
# rather than silently dropping executed topology.
measured += converts
unindexed.append(relative)
continue
if launch_count > measured_launch_floor:
measured += converts
elif converts:
excluded.append(
{
"file": relative,
"launches": launch_count,
"convert_count": converts,
}
)
return {
"measured_convert_count": measured,
"total_convert_count": total,
"ttgir_files": len(files),
"measured_launch_floor": measured_launch_floor,
"scope_complete": launches is not None and not unindexed,
"excluded_unmeasured": excluded,
"unindexed_files": unindexed,
}
def _median_gems_latency(rows: list[dict[str, str]]) -> float | None:
values = []
for row in rows:
if row.get("row_status") != "SUCCESS":
continue
try:
value = float(row.get("gems_lat", ""))
except ValueError:
continue
if math.isfinite(value) and value > 0:
values.append(value)
return statistics.median(values) if values else None
def missing_targets(matrix: dict[str, Any], flaggems: Path) -> list[str]:
"""Resolve every task's benchmark and correctness target against the tree.
The inventory a matrix is built from can lag the pinned FlagGems commit, for
example when upstream splits one benchmark file into several. Before this
check, such a task only failed inside the worker's preflight, which exited
silently and left a zero-byte log and a cell receipt with nothing in it.
"""
missing: list[str] = []
for task in matrix["tasks"]:
benchmark = flaggems / "benchmark" / f"test_{task['op_file']}.py"
if not benchmark.is_file():
missing.append(f"{task['task_id']} benchmark -> {benchmark}")
correctness = flaggems / task["correctness_file"]
if not correctness.is_file():
missing.append(f"{task['task_id']} correctness -> {correctness}")
return missing
def validate_contract(
manifest_path: Path,
matrix_path: Path,
build_root: Path,
flaggems: Path,
zl_bench: Path,
python: Path,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
matrix = json.loads(matrix_path.read_text(encoding="utf-8"))
if not isinstance(manifest, dict) or not isinstance(matrix, dict):
raise CampaignError("manifest and task matrix roots must be objects")
manifest_sha = _sha256(manifest_path)
if matrix.get("manifest_sha256") != manifest_sha:
raise CampaignError("task matrix manifest_sha256 does not match manifest")
if matrix.get("campaign_id") != manifest.get("campaign_id"):
raise CampaignError("campaign id mismatch")
if matrix.get("screen_cells") != ["off", "mask0", "candidate"]:
raise CampaignError("task matrix cell contract mismatch")
missing = missing_targets(matrix, flaggems)
if missing:
raise CampaignError(
"targets missing from the frozen FlagGems tree: " + "; ".join(missing)
)
site = build_root / "site"
identity = _identity(build_root / "identity.txt")
expected = manifest["source"]
if identity.get("git_commit") != expected["flagtree_commit"]:
raise CampaignError("FlagTree build identity does not match manifest")
if not (build_root / "build.done").is_file():
raise CampaignError("FlagTree product build is not terminal")
if (flaggems / ".source-commit").read_text(encoding="utf-8").strip() != expected["flaggems_commit"]:
raise CampaignError("FlagGems source identity does not match manifest")
if (zl_bench / ".source-commit").read_text(encoding="utf-8").strip() != expected["zl_bench_commit"]:
raise CampaignError("zl_bench source identity does not match manifest")
probe_env = _base_env(site, flaggems)
probe_code = (
"import json,torch,torch_musa,triton,flag_gems;"
"print(json.dumps({'count':torch.musa.device_count(),"
"'device':str(flag_gems.device),'vendor':str(flag_gems.vendor_name),"
"'triton':triton.__file__,'flaggems':flag_gems.__file__},sort_keys=True))"
)
probe = subprocess.run(
[str(python), "-B", "-c", probe_code],
cwd=str(flaggems),
env=probe_env,
text=True,
capture_output=True,
timeout=60,
)
if probe.returncode:
raise CampaignError(f"product import probe failed: {probe.stderr.strip()}")
observed = json.loads(probe.stdout.strip().splitlines()[-1])
if observed.get("count") != 1 or observed.get("vendor") != "mthreads":
raise CampaignError(f"MUSA device/vendor mismatch: {observed}")
if not Path(observed["triton"]).is_relative_to(site):
raise CampaignError("imported Triton does not come from the frozen product site")
if not Path(observed["flaggems"]).is_relative_to(flaggems / "src"):
raise CampaignError("imported FlagGems does not come from the frozen product source")
observed.update(
{
"flagtree_commit": identity["git_commit"],
"libtriton_sha256": _sha256(site / "triton/_C/libtriton.so"),
"flaggems_commit": expected["flaggems_commit"],
"zl_bench_commit": expected["zl_bench_commit"],
"python": str(python),
}
)
return manifest, matrix, observed
def run_cell(
task: dict[str, Any],
cell: str,
round_name: str,
replay_from: Path | None,
*,
output_root: Path,
worker: Path,
python: Path,
zl_bench: Path,
flaggems: Path,
site: Path,
shape_file: Path,
batch_summary: Any,
base_env: dict[str, str],
) -> dict[str, Any]:
out = output_root / "tasks" / task["task_id"] / round_name / cell
env = _cell_env(base_env, cell)
env.update(
{
"PYTHON_BIN": str(python),
"ZL_BENCH_DIR": str(zl_bench),
"FLAGGEMS_DIR": str(flaggems),
"FLAGTREE_SITE": str(site),
"OP": task["op"],
"OP_FILE": task["op_file"],
"SHAPE_FILE": str(shape_file),
"OUT_DIR": str(out),
"DTYPES": " ".join(task["benchmark_dtypes"]),
}
)
if task.get("benchmark_selector"):
env["BENCH_SELECTOR"] = str(task["benchmark_selector"])
if replay_from is not None:
env["REPLAY_FROM"] = str(replay_from)
command = ["bash", str(worker)]
# The shell worker owns run.log, so use a sibling controller log.
controller_log = out.parent / f"{cell}.controller.log"
controller_log.parent.mkdir(parents=True, exist_ok=True)
return_code, timed_out, elapsed = _run_process(
command,
cwd=worker.parent,
env=env,
log=controller_log,
timeout=int(task["timeout_seconds"]),
)
run_log = out / "run.log"
text = run_log.read_text(encoding="utf-8", errors="replace") if run_log.is_file() else ""
rows = batch_summary.parse_run_log(text)
if timed_out:
status = "TIMEOUT"
else:
status = batch_summary.classify(return_code, rows, text)
scope = _convert_count(
out, measured_launch_floor=int(task.get("measured_launch_floor", 1))
)
# A worker can fail its preflight before creating OUT_DIR (for example,
# when a frozen inventory entry has no backend benchmark file). Preserve
# that failure as a cell receipt instead of aborting the whole campaign.
out.mkdir(parents=True, exist_ok=True)
receipt = {
"task_id": task["task_id"],
"round": round_name,
"cell": cell,
"status": status,
"return_code": return_code,
"timed_out": timed_out,
"elapsed_seconds": round(elapsed, 3),
"rows": rows,
"rows_success": sum(row.get("row_status") == "SUCCESS" for row in rows),
"median_gems_latency_ms": _median_gems_latency(rows),
"convert_count": scope["measured_convert_count"],
"convert_count_all": scope["total_convert_count"],
"convert_scope_complete": scope["scope_complete"],
"measured_launch_floor": scope["measured_launch_floor"],
"convert_excluded_unmeasured": scope["excluded_unmeasured"],
"convert_unindexed_files": scope["unindexed_files"],
"ttgir_files": scope["ttgir_files"],
"replay_fallback_count": text.count("AUTOTUNE_REPLAY_FALLBACK"),
"benchmark_selector": task.get("benchmark_selector"),
"benchmark_dtypes": list(task["benchmark_dtypes"]),
"effective_rlc": {
key: env[key]
for key in sorted(RLC_ENV_KEYS)
if key in env
},
}
_atomic_json(out / "cell-receipt.json", receipt)
return receipt
def correctness_command(task: dict[str, Any], python: Path) -> list[str]:
"""Build the correctness argv, narrowed to the variant this task measures.
A task-level file runs every variant in that file, so an unrelated failing
variant can mask or fabricate a verdict for the operator under test. A
manifest may therefore pin explicit node ids or a ``-k`` selector; the
manifest schema requires a stated reason whenever it does.
"""
command = [str(python), "-B", "-m", "pytest", "-q"]
# The same vendor-scoped compatibility plugin the benchmark worker loads.
# Without it a stale CUDA-only guard skips the operator here too, and a
# SKIP would be read as "not measured" rather than "not gated".
command.extend(["-p", "_musa_benchmark_compat_plugin"])
node_ids = task.get("correctness_node_ids")
if node_ids:
command.extend(str(node) for node in node_ids)
else:
command.append(task["correctness_file"])
selector = task.get("correctness_selector")
if selector:
command.extend(["-k", str(selector)])
# -rs records why a variant was skipped, so a SKIP verdict stays auditable
# without a second run.
command.extend(["-rs", "--quick"])
return command
def run_correctness(
task: dict[str, Any],
cell: str,
*,
output_root: Path,
python: Path,
flaggems: Path,
site: Path,
base_env: dict[str, str],
plugin_dir: Path,
) -> dict[str, Any]:
out = output_root / "correctness" / task["task_id"] / cell
out.mkdir(parents=True, exist_ok=True)
env = _cell_env(base_env, cell)
env["PYTEST_DISABLE_PLUGIN_AUTOLOAD"] = "1"
env["PYTHONPATH"] = f"{plugin_dir}{os.pathsep}{env['PYTHONPATH']}"
env["TRITON_CACHE_DIR"] = str(out / "triton-cache")
env["FLAGGEMS_CACHE_DIR"] = str(out / "flaggems-cache")
(out / "triton-cache").mkdir()
(out / "flaggems-cache").mkdir()
command = correctness_command(task, python)
return_code, timed_out, elapsed = _run_process(
command,
cwd=flaggems,
env=env,
log=out / "correctness.log",
timeout=int(task["timeout_seconds"]),
)
log_text = (out / "correctness.log").read_text(encoding="utf-8", errors="replace")
if timed_out:
status = "TIMEOUT"
elif return_code != 0:
status = "FAIL"
elif " passed" in log_text:
status = "PASS"
elif " skipped" in log_text:
status = "SKIP"
else:
status = "FAIL"
receipt = {
"task_id": task["task_id"],
"correctness_file": task["correctness_file"],
"correctness_node_ids": task.get("correctness_node_ids"),
"correctness_selector": task.get("correctness_selector"),
"correctness_scope_reason": task.get("correctness_scope_reason"),
"command": command,
"cell": cell,
"status": status,
"return_code": return_code,
"timed_out": timed_out,
"elapsed_seconds": round(elapsed, 3),
}
_atomic_json(out / "correctness-receipt.json", receipt)
return receipt
def _speedup(off: dict[str, Any], candidate: dict[str, Any]) -> float | None:
left = off.get("median_gems_latency_ms")
right = candidate.get("median_gems_latency_ms")
if not isinstance(left, (int, float)) or not isinstance(right, (int, float)) or right <= 0:
return None
return (left / right - 1.0) * 100.0
def classify_task(
correctness: dict[str, Any],
cells: dict[str, dict[str, Any]],
correctness_isolation: dict[str, dict[str, Any]] | None = None,
) -> tuple[str, dict[str, Any]]:
detail: dict[str, Any] = {"reason": ""}
if correctness["status"] == "FAIL":
isolation = correctness_isolation or {}
off_status = isolation.get("off", {}).get("status")
mask0_status = isolation.get("mask0", {}).get("status")
detail.update(
{
"off_correctness_status": off_status,
"mask0_correctness_status": mask0_status,
"candidate_correctness_status": correctness["status"],
}
)
if off_status == "FAIL" and mask0_status == "FAIL":
detail["reason"] = "baseline correctness fails under off/mask0/candidate"
return "X", detail
detail["reason"] = "candidate correctness failed"
return "K", detail
if correctness["status"] != "PASS":
detail["reason"] = f"candidate correctness is {correctness['status']}"
return "X", detail
required = ("a_off", "a_mask0", "a_candidate")
if any(cells.get(key, {}).get("status") != "PASS" for key in required):
detail["reason"] = "one or more screen cells did not pass"
return "X", detail
if any(cells[key].get("replay_fallback_count") for key in ("a_mask0", "a_candidate")):
detail["reason"] = "same-config replay fell back"
return "X", detail
# A topology claim is only admissible when every screen cell could be scoped
# to the measured operator. Older receipts predate scoping and are trusted.
if any(cells[key].get("convert_scope_complete", True) is False for key in required):
detail["reason"] = "measured convert topology scope is incomplete"
return "X", detail
off_count = cells["a_off"]["convert_count"]
mask0_count = cells["a_mask0"]["convert_count"]
candidate_count = cells["a_candidate"]["convert_count"]
detail.update(
{
"off_convert_count": off_count,
"mask0_convert_count": mask0_count,
"candidate_convert_count": candidate_count,
"off_convert_count_all": cells["a_off"].get("convert_count_all"),
"candidate_convert_count_all": cells["a_candidate"].get("convert_count_all"),
"round_a_speedup_percent": _speedup(cells["a_off"], cells["a_candidate"]),
}
)
if off_count != mask0_count:
detail["reason"] = "mask0 topology differs from off"
return "X", detail
if candidate_count >= off_count:
speedup = detail["round_a_speedup_percent"]
if speedup is not None and speedup <= -3.0:
detail["reason"] = "no topology reduction and screening regression"
return "R", detail
detail["reason"] = "candidate did not reduce executed convert topology"
return "B", detail
if cells.get("b_off", {}).get("status") != "PASS" or cells.get("b_candidate", {}).get("status") != "PASS":
detail["reason"] = "topology changed but reversed-order timing did not pass"
return "X", detail
if cells["b_off"].get("replay_fallback_count") or cells["b_candidate"].get("replay_fallback_count"):
detail["reason"] = "reversed-order timing replay fell back"
return "X", detail
round_b = _speedup(cells["b_off"], cells["b_candidate"])
round_a = detail["round_a_speedup_percent"]
detail["round_b_speedup_percent"] = round_b
if round_a is None or round_b is None:
detail["reason"] = "missing comparable native-event latency"
return "X", detail
if round_a >= 5.0 and round_b >= 5.0:
detail["reason"] = "topology reduction and stable >=5% speedup"
return "G", detail
if round_a <= -3.0 and round_b <= -3.0:
detail["reason"] = "topology reduction but stable regression"
return "R", detail
detail["reason"] = "topology reduction without stable >=5% speedup"
return "C", detail
def run_campaign(
manifest: dict[str, Any],
matrix: dict[str, Any],
observed: dict[str, Any],
*,
output_root: Path,
worker: Path,
python: Path,
zl_bench: Path,
flaggems: Path,
build_root: Path,
resume: bool,
) -> dict[str, Any]:
if output_root.exists() and not resume:
raise CampaignError(f"output exists; use --resume: {output_root}")
observed = dict(observed)
observed.update(
{
"benchmark_runner_sha256": _sha256(Path(__file__)),
"benchmark_worker_sha256": _sha256(worker),
"musa_benchmark_compat_plugin_sha256": _sha256(
zl_bench / "_musa_benchmark_compat_plugin.py"
),
}
)
output_root.mkdir(parents=True, exist_ok=True)
site = build_root / "site"
shape_file = zl_bench / "ops/shapes_single.yaml"
batch_summary = _load_module(zl_bench / "batch_summary.py", "zl_batch_summary")
base_env = _base_env(site, flaggems)
status_csv = output_root / "status.csv"
rows: list[dict[str, Any]] = []
occupancy = subprocess.run(
["mthreads-gmi"], text=True, capture_output=True, timeout=30
)
if occupancy.returncode:
raise CampaignError(f"mthreads-gmi failed: {occupancy.stderr.strip()}")
(output_root / "occupancy-before.txt").write_text(occupancy.stdout, encoding="utf-8", newline="\n")
if "No running processes found" not in occupancy.stdout:
raise CampaignError("S5000 has an active process; campaign not admitted")
for task in matrix["tasks"]:
task_root = output_root / "tasks" / task["task_id"]
task_receipt_path = task_root / "task-receipt.json"
if resume and task_receipt_path.is_file():
receipt = json.loads(task_receipt_path.read_text(encoding="utf-8"))
if receipt.get("classification") in TERMINAL:
rows.append(receipt)
continue
cells: dict[str, dict[str, Any]] = {}
off_root = task_root / "round-a" / "off"
cells["a_off"] = run_cell(
task, "off", "round-a", None,
output_root=output_root, worker=worker, python=python,
zl_bench=zl_bench, flaggems=flaggems, site=site,
shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
)
cells["a_mask0"] = run_cell(
task, "mask0", "round-a", off_root,
output_root=output_root, worker=worker, python=python,
zl_bench=zl_bench, flaggems=flaggems, site=site,
shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
)
cells["a_candidate"] = run_cell(
task, "candidate", "round-a", off_root,
output_root=output_root, worker=worker, python=python,
zl_bench=zl_bench, flaggems=flaggems, site=site,
shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
)
correctness = run_correctness(
task, "candidate", output_root=output_root, python=python, flaggems=flaggems,
site=site, base_env=base_env, plugin_dir=zl_bench,
)
correctness_isolation: dict[str, dict[str, Any]] = {
"candidate": correctness
}
if correctness["status"] != "PASS":
correctness_isolation["off"] = run_correctness(
task, "off", output_root=output_root, python=python,
flaggems=flaggems, site=site, base_env=base_env,
plugin_dir=zl_bench,
)
correctness_isolation["mask0"] = run_correctness(
task, "mask0", output_root=output_root, python=python,
flaggems=flaggems, site=site, base_env=base_env,
plugin_dir=zl_bench,
)
if (
cells["a_off"]["status"] == "PASS"
and cells["a_candidate"]["status"] == "PASS"
and cells["a_candidate"]["convert_count"] < cells["a_off"]["convert_count"]
):
cells["b_candidate"] = run_cell(
task, "candidate", "round-b", off_root,
output_root=output_root, worker=worker, python=python,
zl_bench=zl_bench, flaggems=flaggems, site=site,
shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
)
cells["b_off"] = run_cell(
task, "off", "round-b", off_root,
output_root=output_root, worker=worker, python=python,
zl_bench=zl_bench, flaggems=flaggems, site=site,
shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
)
classification, detail = classify_task(
correctness, cells, correctness_isolation
)
receipt = {
"task_id": task["task_id"],
"op": task["op"],
"op_file": task["op_file"],
"family": task["family"],
"nv_evidence": task["nv_evidence"],
"nv_gain_percent": task.get("nv_gain_percent"),
"prior_s5000": task["prior_s5000"],
"classification": classification,
"classification_detail": detail,
"correctness": correctness,
"correctness_isolation": correctness_isolation,
"cells": cells,
}
task_root.mkdir(parents=True, exist_ok=True)
_atomic_json(task_receipt_path, receipt)
rows.append(receipt)
with status_csv.with_name(status_csv.name + ".tmp").open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(
handle,
fieldnames=["task_id", "op", "family", "classification", "reason"],
)
writer.writeheader()
for row in rows:
writer.writerow(
{
"task_id": row["task_id"],
"op": row["op"],
"family": row["family"],
"classification": row["classification"],
"reason": row["classification_detail"]["reason"],
}
)
os.replace(status_csv.with_name(status_csv.name + ".tmp"), status_csv)
print(
f"MUSA_RLC_PRODUCT_TASK task={task['task_id']} class={classification} "
f"reason={detail['reason']}",
flush=True,
)
counts = {key: sum(row["classification"] == key for row in rows) for key in sorted(TERMINAL)}
key_rows = {row["task_id"]: row for row in rows}
canary_gate = None
if matrix.get("selection") == "canary":
canary_gate = (
key_rows.get("dropout", {}).get("classification") in {"G", "C"}
and key_rows.get("softmax", {}).get("classification") == "B"
and counts["K"] == 0
)
final = {
"schema_version": 1,
"campaign_id": manifest["campaign_id"],
"selection": matrix["selection"],
"status": "TERMINAL",
"task_count": len(rows),
"counts": counts,
"canary_gate_pass": canary_gate,
"source_observed": observed,
"production_default_policy": "unchanged-off",
"claim_scope": manifest["permitted_claim"],
}
_atomic_json(output_root / "campaign-receipt.json", final)
return final
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--manifest", type=Path, required=True)
parser.add_argument("--task-matrix", type=Path, required=True)
parser.add_argument("--build-root", type=Path, required=True)
parser.add_argument("--flaggems", type=Path, required=True)
parser.add_argument("--zl-bench", type=Path, required=True)
parser.add_argument("--worker", type=Path, required=True)
parser.add_argument("--python", type=Path, default=Path("/usr/bin/python3"))
parser.add_argument("--output-root", type=Path, required=True)
parser.add_argument("--resume", action="store_true")
args = parser.parse_args()
try:
manifest, matrix, observed = validate_contract(
args.manifest.resolve(), args.task_matrix.resolve(), args.build_root.resolve(),
args.flaggems.resolve(), args.zl_bench.resolve(), args.python.resolve(),
)
final = run_campaign(
manifest, matrix, observed,
output_root=args.output_root.resolve(), worker=args.worker.resolve(),
python=args.python.resolve(), zl_bench=args.zl_bench.resolve(),
flaggems=args.flaggems.resolve(), build_root=args.build_root.resolve(),
resume=args.resume,
)
except (OSError, json.JSONDecodeError, CampaignError, subprocess.SubprocessError) as exc:
print(f"MUSA_RLC_PRODUCT_ERROR: {exc}", file=sys.stderr)
return 2
print(json.dumps(final, indent=2, sort_keys=True))
return 0 if final.get("canary_gate_pass") is not False else 1
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,110 @@
#!/usr/bin/env bash
set -Eeuo pipefail
: "${PYTHON_BIN:?PYTHON_BIN is required}"
: "${ZL_BENCH_DIR:?ZL_BENCH_DIR is required}"
: "${FLAGGEMS_DIR:?FLAGGEMS_DIR is required}"
: "${FLAGTREE_SITE:?FLAGTREE_SITE is required}"
: "${OP:?OP is required}"
: "${OP_FILE:?OP_FILE is required}"
: "${SHAPE_FILE:?SHAPE_FILE is required}"
: "${OUT_DIR:?OUT_DIR is required}"
worker_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# A silent preflight exit leaves a zero-byte controller log and no run.log, so
# the campaign records a failed cell with nothing to diagnose it. Name the check
# that failed instead.
preflight() {
local description="$1"
shift
if ! "$@"; then
printf 'MUSA_RLC_PRODUCT_PREFLIGHT_FAILED op=%s check=%s\n' \
"${OP:-?}" "$description" >&2
exit 3
fi
}
preflight "python interpreter is executable: $PYTHON_BIN" test -x "$PYTHON_BIN"
preflight "zl_bench directory exists: $ZL_BENCH_DIR" test -d "$ZL_BENCH_DIR"
preflight "ir meta plugin exists: $ZL_BENCH_DIR/_ir_meta_plugin.py" \
test -f "$ZL_BENCH_DIR/_ir_meta_plugin.py"
preflight "benchmark file exists: $FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py" \
test -f "$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py"
preflight "FlagGems source exists: $FLAGGEMS_DIR/src/flag_gems" \
test -d "$FLAGGEMS_DIR/src/flag_gems"
preflight "libtriton is non-empty: $FLAGTREE_SITE/triton/_C/libtriton.so" \
test -s "$FLAGTREE_SITE/triton/_C/libtriton.so"
preflight "shape file exists: $SHAPE_FILE" test -f "$SHAPE_FILE"
preflight "output directory is fresh: $OUT_DIR" test ! -e "$OUT_DIR"
mkdir -p "$OUT_DIR/autotune_records" "$OUT_DIR/triton-cache" "$OUT_DIR/flaggems-cache"
export PYTHONNOUSERSITE=1
export PYTHONDONTWRITEBYTECODE=1
export PYTHONUNBUFFERED=1
export GEMS_VENDOR=mthreads
export FLAGGEMS_VENDOR=mthreads
export FLAGTREE_BACKEND=mthreads
export TRITON_BACKENDS_IN_TREE=1
export MUSA_VISIBLE_DEVICES=7
export MTHREADS_VISIBLE_DEVICES=7
export PYTHONPATH="$worker_dir:$ZL_BENCH_DIR:$FLAGGEMS_DIR/src:$FLAGTREE_SITE"
export FLAGGEMS_PERF_CURRENT_OP="$OP"
export FLAGGEMS_PERF_COLOR=never
export NO_COLOR=1
# The plugin list comes from zl_bench, not from here. Hand-copying it is how
# this campaign drifted: the copy was taken before backends/ existed, so it
# still named _device_guard_plugin (now _device_setup_plugin) and, more
# importantly, carried no timing plugin at all -- which is why every sealed
# measurement is timer=direct while zl_bench's own driver had been capturing.
# Asking the planner means the caliber tracks whatever the backend declares.
export ZL_BACKEND=musa
mapfile -t plugins < <("$PYTHON_BIN" "$ZL_BENCH_DIR/plugin_plan.py" \
--backend "$ZL_BACKEND" --pytest-args | tr ' ' '\n')
preflight "plugin plan is non-empty" test "${#plugins[@]}" -gt 0
dtype_args=()
for dtype in ${DTYPES:-float16}; do
dtype_args+=(--dtypes "$dtype")
done
# Some benchmarks parametrise one dtype per case, so a campaign-wide --dtypes
# filter makes every non-matching case raise. A workload may then pin the case
# it measures; the manifest requires a stated reason for that narrowing.
selector_args=()
if [[ -n "${BENCH_SELECTOR:-}" ]]; then
selector_args+=(-k "$BENCH_SELECTOR")
fi
if [[ -n "${REPLAY_FROM:-}" ]]; then
autotune_env="FLAGGEMS_PERF_AUTOTUNE_REPLAY_DIR=$REPLAY_FROM/autotune_records"
mode_desc="replay($REPLAY_FROM)"
else
autotune_env="FLAGGEMS_PERF_AUTOTUNE_RECORD_DIR=$OUT_DIR/autotune_records"
mode_desc=record
fi
cp "$SHAPE_FILE" "$OUT_DIR/shapes.yaml"
test_file="$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py::test_${OP}"
status=0
{
printf 'MUSA_RLC_PRODUCT_RUN op=%s mode=%s out=%s\n' "$OP" "$mode_desc" "$OUT_DIR"
env \
TRITON_CACHE_DIR="$OUT_DIR/triton-cache" \
FLAGGEMS_CACHE_DIR="$OUT_DIR/flaggems-cache" \
FLAGGEMS_PERF_TTGIR_DUMP_DIR="$OUT_DIR/ttgir" \
"$autotune_env" \
USE_FLAGTUNE=0 \
"$PYTHON_BIN" -u -m pytest -s "$test_file" \
"${plugins[@]}" \
--color=no \
--shape_file "$SHAPE_FILE" \
"${dtype_args[@]}" \
"${selector_args[@]}" \
--level core --mode kernel || status=$?
printf 'MUSA_RLC_PRODUCT_RUN_RC=%s\n' "$status"
exit "$status"
} 2>&1 | tee "$OUT_DIR/run.log" || status=$?
exit "$status"
@@ -0,0 +1,573 @@
from __future__ import annotations
import copy
import importlib.util
import json
import sys
import tempfile
import unittest
from types import SimpleNamespace
from pathlib import Path
HERE = Path(__file__).resolve().parent
CAMPAIGN = HERE.parent
REPO_ROOT = CAMPAIGN.parents[1]
REFERENCE = CAMPAIGN / "manifest" / "musa-rlc-32-product-v1.json"
SPEC = importlib.util.spec_from_file_location(
"prepare_musa_rlc_product", CAMPAIGN / "prepare_musa_rlc_product.py"
)
assert SPEC is not None and SPEC.loader is not None
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)
RUN_SPEC = importlib.util.spec_from_file_location(
"run_musa_rlc_product", CAMPAIGN / "run_musa_rlc_product.py"
)
assert RUN_SPEC is not None and RUN_SPEC.loader is not None
RUNNER = importlib.util.module_from_spec(RUN_SPEC)
RUN_SPEC.loader.exec_module(RUNNER)
COMPAT_SPEC = importlib.util.spec_from_file_location(
"_musa_benchmark_compat_plugin", REPO_ROOT / "_musa_benchmark_compat_plugin.py"
)
assert COMPAT_SPEC is not None and COMPAT_SPEC.loader is not None
COMPAT = importlib.util.module_from_spec(COMPAT_SPEC)
COMPAT_SPEC.loader.exec_module(COMPAT)
def _manifest():
return json.loads(REFERENCE.read_text(encoding="utf-8"))
class MusaRlcProductTests(unittest.TestCase):
def test_musa_fft_compatibility_override_is_narrow(self):
stale = SimpleNamespace(
name="skipif", args=(True,), kwargs={"reason": "CUDA is not available"}
)
unrelated = SimpleNamespace(
name="skipif", args=(True,), kwargs={"reason": "another reason"}
)
fft = SimpleNamespace(path=Path("benchmark/test_fft.py"), name="test_perf_fft")
other = SimpleNamespace(path=Path("benchmark/test_fft.py"), name="test_other")
self.assertTrue(COMPAT._is_stale_fft_cuda_guard(fft, stale))
self.assertFalse(COMPAT._is_stale_fft_cuda_guard(fft, unrelated))
self.assertFalse(COMPAT._is_stale_fft_cuda_guard(other, stale))
def test_runner_preserves_preflight_failure_receipts(self):
runner = (CAMPAIGN / "run_musa_rlc_product.py").read_text(encoding="utf-8")
ensure_parent = "out.mkdir(parents=True, exist_ok=True)"
write_receipt = '_atomic_json(out / "cell-receipt.json", receipt)'
self.assertIn(ensure_parent, runner)
self.assertLess(runner.index(ensure_parent), runner.index(write_receipt))
def test_worker_dumps_only_executed_ttgir_variants(self):
worker = (CAMPAIGN / "run_pytest_musa_rlc_product.sh").read_text(
encoding="utf-8"
)
self.assertIn(
'FLAGGEMS_PERF_TTGIR_DUMP_DIR="$OUT_DIR/ttgir"', worker
)
# The plugin list is asked for, not written here: a hand-copied list is
# what let this worker keep _device_guard_plugin and no timing plugin
# long after zl_bench had moved on. What the plan contains for musa --
# the compat plugin included -- is asserted in zl_bench's test_backends.
self.assertIn("plugin_plan.py", worker)
self.assertIn('--backend "$ZL_BACKEND"', worker)
self.assertNotIn("-p _", worker)
self.assertIn(
'PYTHONPATH="$worker_dir:$ZL_BENCH_DIR:$FLAGGEMS_DIR/src:$FLAGTREE_SITE"',
worker,
)
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
dumped = root / "ttgir" / "shape" / "kernel" / "used.ttgir"
cached = root / "triton-cache" / "hash" / "loser.ttgir"
dumped.parent.mkdir(parents=True)
cached.parent.mkdir(parents=True)
dumped.write_text("ttg.convert_layout\nttg.convert_layout\n", encoding="utf-8")
cached.write_text("ttg.convert_layout\n" * 99, encoding="utf-8")
scope = RUNNER._convert_count(root)
self.assertEqual(scope["total_convert_count"], 2)
self.assertEqual(scope["ttgir_files"], 1)
# Without an index the measured scope cannot be established, so the
# count stays unscoped and the cell is marked incomplete.
self.assertEqual(scope["measured_convert_count"], 2)
self.assertFalse(scope["scope_complete"])
@staticmethod
def _write_cell(root, rows):
"""Materialise a cell's ttgir dump plus the index the dumper writes."""
ttgir = root / "ttgir"
ttgir.mkdir(parents=True)
lines = [
"\t".join(
[
"shape", "kernel", "file", "launches", "sweep_launches",
"num_warps", "num_stages", "constexprs", "cache_hash",
]
)
]
for shape, kernel, name, launches, converts in rows:
if name.startswith("("):
lines.append(
"\t".join([shape, kernel, name, str(launches), "0", "4", "3", "{}", "H"])
)
continue
path = ttgir / shape / kernel / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("ttg.convert_layout\n" * converts, encoding="utf-8")
lines.append(
"\t".join([shape, kernel, name, str(launches), "0", "4", "3", "{}", "H"])
)
(ttgir / "index.tsv").write_text("\n".join(lines) + "\n", encoding="utf-8")
def test_convert_count_excludes_input_construction_kernels(self):
# Reproduces the sealed gelu_backward shape: every convert sits in the
# once-launched input builder, while the timed kernel has none.
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self._write_cell(
root,
[
("1048576", "randn_kernel", "w16s3.ttgir", 1, 4),
("1048576", "gelu_none_kernel_rank_1", "w4s3.ttgir", 1, 0),
("1048576", "gelu_backward_none_kernel_rank_1", "w4s3.ttgir", 3795, 0),
],
)
scope = RUNNER._convert_count(root)
self.assertEqual(scope["total_convert_count"], 4)
self.assertEqual(scope["measured_convert_count"], 0)
self.assertTrue(scope["scope_complete"])
self.assertEqual(
[row["file"] for row in scope["excluded_unmeasured"]],
["1048576/randn_kernel/w16s3.ttgir"],
)
def test_convert_count_keeps_repeatedly_launched_kernels(self):
# Reproduces the sealed sort shape: converts in the dominant kernel and
# in an autotune loser that was indexed but never dumped.
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self._write_cell(
root,
[
("1024x1024", "sweep", "w4s3.ttgir", 280, 3),
("1024x1024", "broadcast_to_kernel", "w4s3.ttgir", 140, 1),
("-", "arange_kernel", "(sweep loser, not dumped)", 0, 0),
],
)
scope = RUNNER._convert_count(root)
self.assertEqual(scope["measured_convert_count"], 4)
self.assertEqual(scope["total_convert_count"], 4)
self.assertTrue(scope["scope_complete"])
self.assertEqual(scope["excluded_unmeasured"], [])
def test_convert_count_reports_dumped_but_unindexed_files(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self._write_cell(root, [("s", "measured", "w4s3.ttgir", 140, 2)])
stray = root / "ttgir" / "s" / "stray" / "w4s3.ttgir"
stray.parent.mkdir(parents=True)
stray.write_text("ttg.convert_layout\n", encoding="utf-8")
scope = RUNNER._convert_count(root)
# An unindexed dump is counted rather than dropped, and flagged.
self.assertEqual(scope["measured_convert_count"], 3)
self.assertFalse(scope["scope_complete"])
self.assertEqual(scope["unindexed_files"], ["s/stray/w4s3.ttgir"])
def test_incomplete_convert_scope_fails_closed(self):
def cell(convert_count, latency, complete=True):
return {
"status": "PASS",
"convert_count": convert_count,
"convert_scope_complete": complete,
"median_gems_latency_ms": latency,
"replay_fallback_count": 0,
}
cells = {
"a_off": cell(8, 0.14),
"a_mask0": cell(8, 0.14),
"a_candidate": cell(0, 0.12, complete=False),
"b_candidate": cell(0, 0.121),
"b_off": cell(8, 0.141),
}
classification, detail = RUNNER.classify_task({"status": "PASS"}, cells)
self.assertEqual(classification, "X")
self.assertEqual(
detail["reason"], "measured convert topology scope is incomplete"
)
def test_correctness_command_narrows_to_the_measured_variant(self):
python = Path("/usr/bin/python3")
plain = RUNNER.correctness_command(
{"correctness_file": "tests/test_sort.py"}, python
)
self.assertEqual(plain[-3:], ["tests/test_sort.py", "-rs", "--quick"])
self.assertNotIn("-k", plain)
selected = RUNNER.correctness_command(
{
"correctness_file": "tests/test_upsample_bicubic2d_aa.py",
"correctness_selector": "not backward",
},
python,
)
self.assertIn("tests/test_upsample_bicubic2d_aa.py", selected)
self.assertEqual(selected[selected.index("-k") + 1], "not backward")
pinned = RUNNER.correctness_command(
{
"correctness_file": "tests/test_mm.py",
"correctness_node_ids": ["tests/test_mm.py::test_mm"],
},
python,
)
self.assertIn("tests/test_mm.py::test_mm", pinned)
self.assertNotIn("tests/test_mm.py", pinned)
def test_narrowed_correctness_requires_a_stated_reason(self):
manifest = copy.deepcopy(_manifest())
target = next(
row for row in manifest["workloads"] if row["id"] == "upsample_bicubic2d_aa"
)
self.assertEqual(target["correctness_selector"], "not backward")
target.pop("correctness_scope_reason")
errors = MODULE.validate_manifest(manifest)
self.assertIn(
"workload upsample_bicubic2d_aa narrows correctness and must state "
"correctness_scope_reason",
errors,
)
both = copy.deepcopy(_manifest())
row = next(item for item in both["workloads"] if item["id"] == "sort")
row["correctness_selector"] = "sort"
row["correctness_node_ids"] = ["tests/test_sort.py::test_sort"]
row["correctness_scope_reason"] = "reason"
self.assertIn(
"workload sort must not set both correctness_selector and "
"correctness_node_ids",
MODULE.validate_manifest(both),
)
def test_measured_launch_floor_is_a_frozen_protocol_field(self):
manifest = copy.deepcopy(_manifest())
self.assertEqual(manifest["protocol"]["measured_launch_floor"], 1)
plan, _ = MODULE.build_plan(manifest, "all")
self.assertTrue(
all(task["measured_launch_floor"] == 1 for task in plan["tasks"])
)
scoped = next(
task for task in plan["tasks"] if task["task_id"] == "fused_marlin_moe_mxfp4"
)
self.assertEqual(scoped["correctness_selector"], "mxfp4")
self.assertTrue(scoped["correctness_scope_reason"])
manifest["protocol"]["measured_launch_floor"] = -1
self.assertIn(
"protocol.measured_launch_floor must be a non-negative integer",
MODULE.validate_manifest(manifest),
)
def test_compat_plugin_only_overrides_verified_stale_guards(self):
def mark(reason, name="skipif"):
return SimpleNamespace(name=name, args=(True,), kwargs={"reason": reason})
def item(file_name, test_name):
return SimpleNamespace(path=Path("tests") / file_name, name=test_name)
# Verified on S5000: unskipped, these pass.
self.assertTrue(
COMPAT._is_stale_cuda_guard(
item("test_top_k_per_row_decode.py", "test_accuracy_top_k"),
mark("CUDA device required"),
)
)
self.assertTrue(
COMPAT._is_stale_fft_cuda_guard(
item("test_fft.py", "test_perf_fft"), mark("CUDA is not available")
)
)
# The fft guard stays pinned to the one test it was verified against.
self.assertFalse(
COMPAT._is_stale_cuda_guard(
item("test_fft.py", "test_other"), mark("CUDA is not available")
)
)
# Probed and rejected: grid_sample hardcodes device="cuda" in its own
# helper, and cutlass_scaled_mm gates on vLLM as well as CUDA.
self.assertFalse(
COMPAT._is_stale_cuda_guard(
item("test_grid_sample.py", "test_nearest_zeros_4d_small"),
mark("CUDA is required."),
)
)
self.assertFalse(
COMPAT._is_stale_cuda_guard(
item("test_cutlass_scaled_mm.py", "test_cutlass_scaled_mm"),
mark("Feature requires vLLM and NVIDIA Hopper architecture"),
)
)
# Only skipif marks, and only the exact reason string.
self.assertFalse(
COMPAT._is_stale_cuda_guard(
item("test_top_k_per_row_decode.py", "test_accuracy_top_k"),
mark("CUDA device required", name="xfail"),
)
)
self.assertFalse(
COMPAT._is_stale_cuda_guard(
item("test_top_k_per_row_decode.py", "test_accuracy_top_k"),
mark("some other reason"),
)
)
def test_compat_plugin_strips_module_level_guards(self):
# top_k_per_row_decode carries its guard as a module-level pytestmark,
# which resolves through the node chain rather than the item's own marks.
stale = SimpleNamespace(
name="skipif", args=(True,), kwargs={"reason": "CUDA device required"}
)
keep = SimpleNamespace(
name="skipif", args=(True,), kwargs={"reason": "vLLM not installed"}
)
module = SimpleNamespace(own_markers=[stale, keep])
function = SimpleNamespace(
path=Path("tests/test_top_k_per_row_decode.py"),
name="test_accuracy_top_k",
own_markers=[],
)
function.listchain = lambda: [module, function]
sys.modules.setdefault(
"flag_gems", SimpleNamespace(vendor_name="mthreads")
)
COMPAT.pytest_collection_modifyitems([function])
self.assertEqual(module.own_markers, [keep])
def test_benchmark_selector_requires_a_stated_reason(self):
manifest = copy.deepcopy(_manifest())
target = next(
row for row in manifest["workloads"] if row["id"] == "grid_sample"
)
self.assertEqual(target["benchmark_selector"], "dtype0")
target.pop("benchmark_scope_reason")
self.assertIn(
"workload grid_sample narrows the benchmark and must state "
"benchmark_scope_reason",
MODULE.validate_manifest(manifest),
)
def test_declared_benchmark_dtypes_must_exist_in_the_cli_vocabulary(self):
manifest = copy.deepcopy(_manifest())
self.assertEqual(
next(row for row in manifest["workloads"] if row["id"] == "randperm")[
"benchmark_dtypes"
],
["int32"],
)
# int64 is not a --dtypes choice; the sealed run only found that out on
# the device, as an argparse error inside a screen cell.
next(row for row in manifest["workloads"] if row["id"] == "randperm")[
"benchmark_dtypes"
] = ["int32", "int64"]
self.assertIn(
"workload randperm declares benchmark dtype(s) the benchmark CLI "
"does not accept: int64",
MODULE.validate_manifest(manifest),
)
def test_stale_inventory_needs_an_override_reason(self):
manifest = copy.deepcopy(_manifest())
marlin = next(
row
for row in manifest["workloads"]
if row["id"] == "fused_marlin_moe_mxfp4"
)
self.assertEqual(marlin["op_file"], "fused_marlin_moe_w4a16_mxfp4")
self.assertTrue(marlin["inventory_override_reason"])
with tempfile.TemporaryDirectory() as temporary:
inventory = Path(temporary)
(inventory / "inventory_main.csv").write_text(
"repo,op,op_file,file,marked_skip,collected,op_names\n"
+ "".join(
f"main,{row['op']},{row['op_file']},f.py,,yes,{row['id']}\n"
for row in manifest["workloads"]
if row["id"] != "fused_marlin_moe_mxfp4"
),
encoding="utf-8",
)
# The stated override lets the missing row through.
self.assertEqual(
MODULE.validate_inventory(manifest["workloads"], inventory), []
)
marlin.pop("inventory_override_reason")
self.assertEqual(
MODULE.validate_inventory(manifest["workloads"], inventory),
[
"inventory missing ('main', 'fused_marlin_moe_w4a16_mxfp4', "
"'fused_marlin_moe_w4a16_mxfp4')"
],
)
def test_missing_benchmark_target_fails_before_the_campaign_starts(self):
matrix = {
"tasks": [
{
"task_id": "present",
"op_file": "sort",
"correctness_file": "tests/test_sort.py",
},
{
# The shape of the sealed marlin defect: the inventory named
# a merged benchmark file that upstream had already split.
"task_id": "fused_marlin_moe_mxfp4",
"op_file": "fused_marlin_moe",
"correctness_file": "tests/test_fused_marlin_moe.py",
},
]
}
with tempfile.TemporaryDirectory() as temporary:
flaggems = Path(temporary)
(flaggems / "benchmark").mkdir()
(flaggems / "tests").mkdir()
(flaggems / "benchmark" / "test_sort.py").write_text("", encoding="utf-8")
(flaggems / "tests" / "test_sort.py").write_text("", encoding="utf-8")
(flaggems / "tests" / "test_fused_marlin_moe.py").write_text(
"", encoding="utf-8"
)
missing = RUNNER.missing_targets(matrix, flaggems)
self.assertEqual(len(missing), 1)
self.assertIn("fused_marlin_moe_mxfp4 benchmark", missing[0])
self.assertIn("test_fused_marlin_moe.py", missing[0])
def test_worker_names_the_failed_preflight_check(self):
worker = (CAMPAIGN / "run_pytest_musa_rlc_product.sh").read_text(encoding="utf-8")
self.assertIn("MUSA_RLC_PRODUCT_PREFLIGHT_FAILED", worker)
self.assertIn(
'preflight "benchmark file exists: '
'$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py"',
worker,
)
# The benchmark selector must reach pytest.
self.assertIn('selector_args+=(-k "$BENCH_SELECTOR")', worker)
self.assertIn('"${selector_args[@]}"', worker)
def test_correctness_runs_with_the_vendor_compat_plugin(self):
command = RUNNER.correctness_command(
{"correctness_file": "tests/test_top_k_per_row_decode.py"},
Path("/usr/bin/python3"),
)
self.assertEqual(
command[command.index("-p") + 1], "_musa_benchmark_compat_plugin"
)
def test_valid_manifest_and_plan_sizes(self):
manifest = _manifest()
self.assertEqual(MODULE.validate_manifest(manifest), [])
canary, canary_budget = MODULE.build_plan(manifest, "canary")
full, full_budget = MODULE.build_plan(manifest, "all")
self.assertEqual(len(canary["tasks"]), 8)
self.assertEqual(len(full["tasks"]), 32)
self.assertEqual(canary["tasks"][0]["task_id"], "dropout")
self.assertEqual(canary["tasks"][1]["task_id"], "softmax")
self.assertEqual(canary["tasks"][0]["benchmark_dtypes"], ["float16"])
self.assertEqual(
next(task for task in canary["tasks"] if task["task_id"] == "fft")[
"benchmark_dtypes"
],
["complex64"],
)
self.assertEqual(canary_budget["gpu_count"], 1)
self.assertEqual(full_budget["gpu_count"], 1)
self.assertTrue(canary_budget["serial_measurement"])
def test_invalid_device_and_canary_count_fail_closed(self):
manifest = copy.deepcopy(_manifest())
manifest["device"]["physical_device"] = "0"
manifest["workloads"][0]["canary"] = False
errors = MODULE.validate_manifest(manifest)
self.assertIn("device.physical_device must equal '7'", errors)
self.assertIn("exactly eight workloads must be canaries", errors)
def test_invalid_timer_and_default_policy_fail_closed(self):
manifest = copy.deepcopy(_manifest())
manifest["protocol"]["timer"] = "host wall clock"
manifest["production_default_policy"] = "enabled"
errors = MODULE.validate_manifest(manifest)
self.assertIn("protocol.timer must bind the MUSA native-event path", errors)
self.assertIn("production_default_policy must be unchanged-off", errors)
def test_runtime_classifier_positive_and_fail_closed_paths(self):
def cell(convert_count, latency, status="PASS", replay=0):
return {
"status": status,
"convert_count": convert_count,
"median_gems_latency_ms": latency,
"replay_fallback_count": replay,
}
correct = {"status": "PASS"}
cells = {
"a_off": cell(8, 0.14),
"a_mask0": cell(8, 0.14),
"a_candidate": cell(0, 0.12),
"b_candidate": cell(0, 0.121),
"b_off": cell(8, 0.141),
}
classification, _ = RUNNER.classify_task(correct, cells)
self.assertEqual(classification, "G")
invalid = copy.deepcopy(cells)
invalid["a_mask0"]["convert_count"] = 7
classification, detail = RUNNER.classify_task(correct, invalid)
self.assertEqual(classification, "X")
self.assertEqual(detail["reason"], "mask0 topology differs from off")
classification, _ = RUNNER.classify_task({"status": "FAIL"}, cells)
self.assertEqual(classification, "K")
classification, detail = RUNNER.classify_task(
{"status": "FAIL"},
cells,
{
"candidate": {"status": "FAIL"},
"mask0": {"status": "FAIL"},
"off": {"status": "FAIL"},
},
)
self.assertEqual(classification, "X")
self.assertEqual(
detail["reason"],
"baseline correctness fails under off/mask0/candidate",
)
def test_full_preparation_requires_passing_canary_receipt(self):
manifest = _manifest()
self.assertEqual(
MODULE.validate_canary_receipt(manifest, "all", None),
["full campaign requires --canary-receipt"],
)
with tempfile.TemporaryDirectory() as temporary:
receipt = Path(temporary) / "campaign-receipt.json"
receipt.write_text(
json.dumps(
{
"campaign_id": manifest["campaign_id"],
"selection": "canary",
"status": "TERMINAL",
"canary_gate_pass": True,
}
),
encoding="utf-8",
)
self.assertEqual(
MODULE.validate_canary_receipt(manifest, "all", receipt), []
)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,157 @@
from __future__ import annotations
import importlib.util
import json
import tempfile
import unittest
from pathlib import Path
HERE = Path(__file__).resolve().parent
CAMPAIGN = HERE.parent
SPEC = importlib.util.spec_from_file_location(
"reaccount_musa_rlc_topology", CAMPAIGN / "reaccount_musa_rlc_topology.py"
)
assert SPEC is not None and SPEC.loader is not None
TOOL = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(TOOL)
def _write_cell(cell_dir: Path, rows):
ttgir = cell_dir / "ttgir"
ttgir.mkdir(parents=True)
lines = [
"\t".join(
[
"shape", "kernel", "file", "launches", "sweep_launches",
"num_warps", "num_stages", "constexprs", "cache_hash",
]
)
]
for shape, kernel, name, launches, converts in rows:
path = ttgir / shape / kernel / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("ttg.convert_layout\n" * converts, encoding="utf-8")
lines.append(
"\t".join([shape, kernel, name, str(launches), "0", "4", "3", "{}", "H"])
)
(ttgir / "index.tsv").write_text("\n".join(lines) + "\n", encoding="utf-8")
def _seal_run(root: Path):
"""A minimal sealed run shaped like the gelu_backward task."""
(root / "campaign-receipt.json").write_text(
json.dumps({"campaign_id": "sealed-v2"}), encoding="utf-8"
)
task_root = root / "tasks" / "gelu_backward"
cells = {}
for key, round_name, cell_name, setup, measured in (
("a_off", "round-a", "off", 4, 0),
("a_mask0", "round-a", "mask0", 4, 0),
("a_candidate", "round-a", "candidate", 0, 0),
("b_off", "round-b", "off", 4, 0),
("b_candidate", "round-b", "candidate", 0, 0),
):
cell_dir = task_root / round_name / cell_name
_write_cell(
cell_dir,
[
("s", "randn_kernel", "w16s3.ttgir", 1, setup),
("s", "gelu_backward_kernel", "w4s3.ttgir", 3795, measured),
],
)
cells[key] = {
"round": round_name,
"cell": cell_name,
"status": "PASS",
"convert_count": setup + measured,
"median_gems_latency_ms": 0.1168 if cell_name == "off" else 0.1166,
"replay_fallback_count": 0,
}
task_root.mkdir(parents=True, exist_ok=True)
(task_root / "task-receipt.json").write_text(
json.dumps(
{
"task_id": "gelu_backward",
"op": "gelu_backward",
"family": "elementwise-backward",
"classification": "C",
"classification_detail": {
"reason": "topology reduction without stable >=5% speedup"
},
"correctness": {"status": "PASS"},
"correctness_isolation": {"candidate": {"status": "PASS"}},
"cells": cells,
}
),
encoding="utf-8",
)
class ReaccountTests(unittest.TestCase):
def test_input_construction_reduction_is_demoted_to_b(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
run_root = base / "sealed"
run_root.mkdir()
_seal_run(run_root)
result = TOOL.reaccount(run_root, base / "derived", 1)
self.assertEqual(result["changed_task_ids"], ["gelu_backward"])
task = result["tasks"][0]
self.assertEqual(task["sealed_classification"], "C")
self.assertEqual(task["reaccounted_classification"], "B")
self.assertEqual(task["sealed_convert"]["a_off"], 4)
self.assertEqual(task["reaccounted_convert"]["a_off"], 0)
self.assertEqual(
task["excluded_unmeasured_files"], ["s/randn_kernel/w16s3.ttgir"]
)
def test_a_zero_floor_reproduces_the_sealed_verdict(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
run_root = base / "sealed"
run_root.mkdir()
_seal_run(run_root)
result = TOOL.reaccount(run_root, base / "derived", 0)
# Floor 0 admits the once-launched builder again, so the derived verdict
# must match the sealed one. That equivalence is what makes the scoped
# count auditable rather than an unexplained new number.
self.assertEqual(result["changed_task_ids"], [])
self.assertEqual(result["tasks"][0]["reaccounted_classification"], "C")
def test_derived_output_is_written_and_never_inside_the_sealed_run(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
run_root = base / "sealed"
run_root.mkdir()
_seal_run(run_root)
with self.assertRaises(TOOL.ReaccountError):
TOOL.reaccount(run_root, run_root / "derived", 1)
with self.assertRaises(TOOL.ReaccountError):
TOOL.reaccount(run_root, run_root, 1)
output = base / "derived"
TOOL.reaccount(run_root, output, 1)
self.assertTrue((output / "reaccount-receipt.json").is_file())
self.assertTrue((output / "reaccount-matrix.csv").is_file())
# The sealed receipt must survive untouched.
sealed = json.loads(
(run_root / "tasks" / "gelu_backward" / "task-receipt.json").read_text(
encoding="utf-8"
)
)
self.assertEqual(sealed["classification"], "C")
self.assertEqual(sealed["cells"]["a_off"]["convert_count"], 4)
def test_a_directory_without_a_campaign_receipt_is_rejected(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
with self.assertRaises(TOOL.ReaccountError):
TOOL.reaccount(base, base.parent / "derived", 1)
if __name__ == "__main__":
unittest.main()
+3 -1
View File
@@ -41,7 +41,9 @@ def plan_plugins(
enable_mm_cluster: Optional[bool] = None,
) -> List[str]:
backend = get_backend(backend_name)
plugins: List[str] = list(CORE_PLUGINS)
# Compat first: these override collection-time markers, so they have to be
# in place before anything else inspects the collected items.
plugins: List[str] = list(backend.compat_plugins) + list(CORE_PLUGINS)
autotune = (
backend.supports_triton_hooks
+50 -3
View File
@@ -75,8 +75,30 @@ case "$BENCH_LEVEL" in
*) echo "!!! BENCH_LEVEL 只能是 core 或 comprehensive(当前:$BENCH_LEVEL)" >&2; exit 2 ;;
esac
# FlagGems 的 libtuner winner 存在 $FLAGGEMS_CACHE_DIR/config_cache 的 sqlite 里,
# 默认 ~/.flaggems,跨 run 长存。1=本次 run 用独立缓存目录,两侧各自冷调优
# (README「各自最优」口径);0=沿用共享缓存(默认,快)。
#
# 共享缓存的风险是单向的:先跑的一侧把 winner 写进去,后跑的一侧直接命中,
# 于是它只编译那几个胜出变体。实测 mm 上后跑侧的 TTGIR 变体从 945 塌到 33、
# cvt 从 136 变 10,看起来像被测改动的效果,其实只是继承了对侧的 config。
# 走 REPLAY_FROM 的标准 A/B 不受此影响(config 本来就锁死),record 对 record
# 才会中招。无论取哪个值都会写进 run_identity.json,口径不再隐形。
ZL_ISOLATE_TUNER_CACHE="${ZL_ISOLATE_TUNER_CACHE:-0}"
# ============================== 执行逻辑 ==============================
# 前置检查。静默失败最贵:路径不存在时 pytest 收集不到 benchmark 目录,注册
# --level/--mode 的 conftest 就没加载,报错变成 "unrecognized arguments",指向
# 完全错误的方向。每项检查自报名字,失败即停。
preflight() {
local what="$1"; shift
if ! "$@"; then
echo "!!! PREFLIGHT_FAILED check=$what" >&2
exit 3
fi
}
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
export PYTHONPATH="$SCRIPT_DIR${PYTHONPATH:+:$PYTHONPATH}"
eval "$(python "$SCRIPT_DIR/providers/flaggems.py" --emit-shell \
@@ -88,6 +110,17 @@ eval "$(python "$SCRIPT_DIR/providers/flaggems.py" --emit-shell \
--shape-mode "$SHAPE_MODE")"
: "${TEST_FILE:=$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py::test_${OP}}"
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%%::*}"
if [[ -n "$SHAPE_FILE" ]]; then
preflight "shape file exists: $SHAPE_FILE" test -f "$SHAPE_FILE"
fi
if [[ -n "$REPLAY_FROM" ]]; then
preflight "replay run directory exists: $REPLAY_FROM" test -d "$REPLAY_FROM"
fi
# 产物目录,可用 OUT_DIR 指定。时间戳只到秒,并发启动会撞名,故用不带 -p 的
# mkdir 抢占(已存在即失败),撞上就退让到 -2、-3……
if [[ -n "${OUT_DIR:-}" ]]; then
@@ -162,6 +195,18 @@ else
SHAPE_ARGS=(--shape_file "$SHAPE_FILE")
fi
# libtuner 缓存:隔离时用本 run 的私有目录,跑完随 Triton 缓存一起删(选中的
# config 已经在 autotune_records/ 里,DB 本身没有留存价值)。
TUNER_CACHE_ARGS=()
if [[ "$ZL_ISOLATE_TUNER_CACHE" == "1" ]]; then
TUNER_CACHE_DIR="$OUT_DIR/.flaggems_cache"
rm -rf "$TUNER_CACHE_DIR"; mkdir -p "$TUNER_CACHE_DIR"
TUNER_CACHE_ARGS=(FLAGGEMS_CACHE_DIR="$TUNER_CACHE_DIR")
TUNER_DESC=isolated
else
TUNER_DESC="shared(${FLAGGEMS_CACHE_DIR:-~/.flaggems})"
fi
python "$SCRIPT_DIR/_run_identity.py" "$OUT_DIR" \
--backend "$ZL_BACKEND" \
--op "$OP" \
@@ -171,11 +216,12 @@ python "$SCRIPT_DIR/_run_identity.py" "$OUT_DIR" \
--device "${ZL_DEVICE:-}" \
--test-file "$TEST_FILE" \
--shape-mode "$SHAPE_MODE" \
--bench-level "$BENCH_LEVEL"
--bench-level "$BENCH_LEVEL" \
--tuner-cache "$TUNER_DESC"
status=0
{
echo "${C_BOLD}>>> op=$OP backend=$ZL_BACKEND mode=$MODE_DESC USE_FLAGTUNE=$USE_FLAGTUNE shape=$SHAPE_MODE level=$BENCH_LEVEL${C_RESET}"
echo "${C_BOLD}>>> op=$OP backend=$ZL_BACKEND mode=$MODE_DESC USE_FLAGTUNE=$USE_FLAGTUNE shape=$SHAPE_MODE level=$BENCH_LEVEL tuner=$TUNER_DESC${C_RESET}"
echo "${C_DIM}>>> node=$TEST_FILE${C_RESET}"
echo "${C_DIM}>>> out=$OUT_DIR${C_RESET}"
@@ -186,13 +232,14 @@ status=0
status=0
TRITON_CACHE_DIR="$CACHE_DIR" \
FLAGGEMS_PERF_TTGIR_DUMP_DIR="$OUT_DIR/ttgir" \
env "$AUTOTUNE_ENV" \
env "$AUTOTUNE_ENV" ${TUNER_CACHE_ARGS[@]+"${TUNER_CACHE_ARGS[@]}"} \
USE_FLAGTUNE=$USE_FLAGTUNE python -u -m pytest -s "$TEST_FILE" \
"${PLUGINS[@]}" "${PYTEST_COLOR[@]}" \
${SHAPE_ARGS[@]+"${SHAPE_ARGS[@]}"} ${DTYPE_ARGS[@]+"${DTYPE_ARGS[@]}"} \
--level "$BENCH_LEVEL" --mode kernel || status=$?
rm -rf "$CACHE_DIR"
[[ -n "${TUNER_CACHE_DIR:-}" ]] && rm -rf "$TUNER_CACHE_DIR"
if (( status == 0 )); then
echo "${C_GREEN}>>> done. outputs in $OUT_DIR${C_RESET}"
+11
View File
@@ -173,6 +173,17 @@ class PluginPlanTests(unittest.TestCase):
self.assertNotIn("_cudagraph_plugin", names)
self.assertNotIn("_mm_cluster_fix_plugin", names)
def test_musa_loads_its_compat_plugin_first(self):
names = self._plan("musa")
self.assertIn("_musa_benchmark_compat_plugin", names)
# It rewrites collection-time markers, so nothing may inspect the item
# list before it runs.
self.assertEqual(names[0], "_musa_benchmark_compat_plugin")
def test_compat_plugins_do_not_leak_to_other_backends(self):
for name in ("cuda", "rocm", "ascend"):
self.assertNotIn("_musa_benchmark_compat_plugin", self._plan(name), name)
def test_musa_same_core_without_cluster_fix(self):
# _cudagraph_plugin is absent from the plan on every backend: the timing
# plugin loads it, so its absence here says nothing about capture support.