From 8a685638bb98b42fbccda8e2fc4cc5cbcc9b934a Mon Sep 17 00:00:00 2001 From: Aharrypotter <34+aharrypotter@noreply.localhost> Date: Wed, 2 Sep 2026 11:35:43 +0800 Subject: [PATCH] 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. --- .../reaccount_musa_rlc_topology.py | 223 ++++++++++++++++++ .../tests/test_reaccount_musa_rlc_topology.py | 157 ++++++++++++ 2 files changed, 380 insertions(+) create mode 100644 campaigns/musa_rlc_product/reaccount_musa_rlc_topology.py create mode 100644 campaigns/musa_rlc_product/tests/test_reaccount_musa_rlc_topology.py diff --git a/campaigns/musa_rlc_product/reaccount_musa_rlc_topology.py b/campaigns/musa_rlc_product/reaccount_musa_rlc_topology.py new file mode 100644 index 0000000..3a8e9fd --- /dev/null +++ b/campaigns/musa_rlc_product/reaccount_musa_rlc_topology.py @@ -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()) diff --git a/campaigns/musa_rlc_product/tests/test_reaccount_musa_rlc_topology.py b/campaigns/musa_rlc_product/tests/test_reaccount_musa_rlc_topology.py new file mode 100644 index 0000000..f9ffafe --- /dev/null +++ b/campaigns/musa_rlc_product/tests/test_reaccount_musa_rlc_topology.py @@ -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()