Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
200 changes: 200 additions & 0 deletions scripts/isa_resource_table.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,200 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# Copyright (c) 2025 FlyDSL Project Contributors
"""Summarize per-kernel ISA resource usage from a FLYDSL_DUMP_IR tree.

Delta table for codegen changes that alter address arithmetic: run a kernel
suite twice (before and after), dumping to two directories, then diff. A
functional pass/fail does not surface a silent VGPR or spill regression, so
these numbers have to be compared explicitly.

FLYDSL_DUMP_IR=1 FLYDSL_DUMP_DIR=<dir> python3 -m pytest <cases>
python3 isa_resource_table.py <dir> --json out.json
python3 isa_resource_table.py --diff before.json after.json

One dump directory can hold several kernels in a single ISA file, so entries are
keyed `<dump dir>::<kernel>` and every metric is scoped to that kernel: the
register counts come from that kernel's metadata entry and the instruction
counts from that kernel's body only.

--diff exits non-zero on a regression, and also whenever the comparison itself
is untrustworthy -- no kernels, a kernel on only one side, or a metric that
failed to parse. A gate that cannot tell "no regressions" from "no data" is
worse than no gate.
"""

import argparse
import json
import os
import re
import sys

KEYS = ("vgpr", "sgpr", "spill", "scratch_store", "scratch_load", "ds_read")
WORSE_IF_UP = ("vgpr", "spill", "scratch_store", "scratch_load")

# Kernel entries in the amdhsa.kernels list start at exactly two spaces; nested
# .args entries are indented further.
_ENTRY_SPLIT = re.compile(r"^ - ", re.M)


def _int_field(text, name):
m = re.search(r"\.%s:\s*(\d+)" % name, text)
return int(m.group(1)) if m else None


def _kernel_body(text, name, others):
"""Instruction text of one kernel: its label up to the end of its body.

The body ends at whichever comes first: the next `.Lfunc_endN:` or the label
of another kernel. Bounding by the label too means the result never depends
on `.Lfunc_end` numbering lining up with declaration order.

Returns None when no terminator is found. Falling back to "rest of file"
would silently fold the following kernels' instructions into this one's
counts -- a wrong number rather than a visible failure.
"""
m = re.search(r"^%s:\s*$" % re.escape(name), text, re.M)
if not m:
return None
rest = text[m.end() :]

ends = []
fn_end = re.search(r"^\.Lfunc_end\d+:", rest, re.M)
if fn_end:
ends.append(fn_end.start())
for other in others:
nxt = re.search(r"^%s:\s*$" % re.escape(other), rest, re.M)
if nxt:
ends.append(nxt.start())
return rest[: min(ends)] if ends else None


def parse_isa(path):
with open(path) as fh:
text = fh.read()

_, _, meta = text.partition("amdhsa.kernels:")
meta = meta.split(".end_amdgpu_metadata")[0]

names, chunks = [], []
for chunk in _ENTRY_SPLIT.split(meta)[1:]:
m = re.search(r"\.name:\s*(\S+)", chunk)
if not m:
continue
names.append(m.group(1))
chunks.append(chunk)

out = {}
for name, chunk in zip(names, chunks):
body = _kernel_body(text, name, [n for n in names if n != name])
out[name] = {
"vgpr": _int_field(chunk, "vgpr_count"),
"sgpr": _int_field(chunk, "sgpr_count"),
"spill": _int_field(chunk, "vgpr_spill_count"),
# None rather than 0 when the body is missing, so a parse failure is
# reported instead of silently looking like a clean kernel.
"scratch_store": body.count("scratch_store") if body is not None else None,
"scratch_load": body.count("scratch_load") if body is not None else None,
"ds_read": body.count("ds_read") if body is not None else None,
}
return out


def collect(root):
out = {}
for dirpath, _, files in os.walk(root):
for f in files:
if f.endswith("final_isa.s"):
mod = os.path.basename(dirpath)
for kern, metrics in parse_isa(os.path.join(dirpath, f)).items():
# The dump dir is usually named after its only kernel; keep
# the qualifier only where it actually disambiguates.
out[kern if kern == mod else "%s::%s" % (mod, kern)] = metrics
return out


def do_diff(before_path, after_path):
with open(before_path) as fh:
before = json.load(fh)
with open(after_path) as fh:
after = json.load(fh)

problems = []
if not before or not after:
problems.append("one or both inputs contain no kernels (before=%d, after=%d)" % (len(before), len(after)))

names = sorted(set(before) | set(after))
hdr = "%-52s " % "kernel" + " ".join("%15s" % k for k in KEYS)
print(hdr)
print("-" * len(hdr))
worsened = improved = unchanged = 0
for n in names:
b, x = before.get(n), after.get(n)
if b is None or x is None:
side = "ONLY IN BEFORE" if x is None else "ONLY IN AFTER"
print("%-52s %s" % (n, side))
problems.append("%s: %s" % (n, side))
continue
missing = [k for k in KEYS if b.get(k) is None or x.get(k) is None]
if missing:
print("%-52s UNPARSED: %s" % (n, ",".join(missing)))
problems.append("%s: unparsed metrics %s" % (n, ",".join(missing)))
continue
cells, changed = [], False
for k in KEYS:
bv, xv = b[k], x[k]
if bv == xv:
cells.append("%15s" % xv)
continue
changed = True
d = xv - bv
if k in WORSE_IF_UP:
if d > 0:
worsened += 1
else:
improved += 1
cells.append(("%s->%s(%+d)" % (bv, xv, d)).rjust(15))
if changed:
print("%-52s " % n + " ".join(cells))
else:
unchanged += 1

print(
"\n%d kernels compared; %d unchanged; worsened metrics: %d; improved: %d"
% (len(names), unchanged, worsened, improved)
)
if problems:
print("\nCOMPARISON NOT TRUSTWORTHY (%d problem(s)):" % len(problems), file=sys.stderr)
for p in problems[:20]:
print(" " + p, file=sys.stderr)
return 2
return 1 if worsened else 0


def main():
ap = argparse.ArgumentParser()
ap.add_argument("dump_dir", nargs="?")
ap.add_argument("--json")
ap.add_argument("--diff", nargs=2, metavar=("BEFORE", "AFTER"))
a = ap.parse_args()

if a.diff:
return do_diff(a.diff[0], a.diff[1])

if not a.dump_dir:
ap.error("dump_dir required unless --diff is used")
data = collect(a.dump_dir)
if a.json:
with open(a.json, "w") as fh:
json.dump(data, fh, indent=1, sort_keys=True)
hdr = "%-52s " % "kernel" + " ".join("%8s" % k for k in KEYS)
print(hdr)
print("-" * len(hdr))
for n in sorted(data):
print("%-52s " % n + " ".join("%8s" % data[n].get(k) for k in KEYS))
print("\n%d kernels" % len(data))
return 0 if data else 2


if __name__ == "__main__":
sys.exit(main())
161 changes: 161 additions & 0 deletions tests/unit/test_isa_resource_table.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,161 @@
#!/usr/bin/env python3

# SPDX-License-Identifier: Apache-2.0
# Copyright (c) 2026 FlyDSL Project Contributors

"""Tests for scripts/isa_resource_table.py.

The parser reads LLVM's AMDGPU assembly text, which is not a stable interface.
These tests pin the exact shape it depends on -- the `amdhsa.kernels` metadata
list, the two-space ` - ` entry indent, and the `<name>:` / `.Lfunc_endN:`
body delimiters -- so a future LLVM format change fails here rather than in a
resource comparison.

The parser must also never guess: when the format does not match it reports
None, and --diff turns that into a non-zero exit. Silently returning a wrong
number would be worse than reporting nothing, because the whole point of the
tool is to catch regressions a passing test suite hides.
"""

import importlib.util
import json
import pathlib
import sys

import pytest

_SCRIPT = pathlib.Path(__file__).resolve().parents[2] / "scripts" / "isa_resource_table.py"
_spec = importlib.util.spec_from_file_location("isa_resource_table", _SCRIPT)
irt = importlib.util.module_from_spec(_spec)
sys.modules["isa_resource_table"] = irt
_spec.loader.exec_module(irt)


TWO_KERNEL_ISA = """\t.text
first_kernel_0:
\tds_read_b64 v[0:1], v2
\tds_read_b64 v[2:3], v4
\tscratch_store_dword off, v0, s0
\ts_endpgm
.Lfunc_end0:
\t.size\tfirst_kernel_0, .Lfunc_end0-first_kernel_0
second_kernel_1:
\tds_read_b64 v[6:7], v8
\ts_endpgm
.Lfunc_end1:
\t.size\tsecond_kernel_1, .Lfunc_end1-second_kernel_1
\t.amdgpu_metadata
amdhsa.kernels:
- .agpr_count: 0
.args:
- .offset: 0
.size: 8
.value_kind: global_buffer
.name: first_kernel_0
.sgpr_count: 47
.vgpr_count: 16
.vgpr_spill_count: 4
- .agpr_count: 0
.name: second_kernel_1
.sgpr_count: 18
.vgpr_count: 32
.vgpr_spill_count: 0
.end_amdgpu_metadata
"""


def _write(tmp_path, text, name="21_final_isa.s"):
p = tmp_path / name
p.write_text(text)
return str(p)


def test_each_kernel_parsed_separately(tmp_path):
"""Registers come from the kernel's own metadata entry, not the first match."""
got = irt.parse_isa(_write(tmp_path, TWO_KERNEL_ISA))
assert set(got) == {"first_kernel_0", "second_kernel_1"}
assert (got["first_kernel_0"]["vgpr"], got["first_kernel_0"]["sgpr"]) == (16, 47)
assert (got["second_kernel_1"]["vgpr"], got["second_kernel_1"]["sgpr"]) == (32, 18)
assert got["first_kernel_0"]["spill"] == 4
assert got["second_kernel_1"]["spill"] == 0


def test_instruction_counts_scoped_to_one_body(tmp_path):
"""Counting file-wide would give the first kernel 3 ds_read instead of 2."""
got = irt.parse_isa(_write(tmp_path, TWO_KERNEL_ISA))
assert got["first_kernel_0"]["ds_read"] == 2
assert got["second_kernel_1"]["ds_read"] == 1
assert got["first_kernel_0"]["scratch_store"] == 1
assert got["second_kernel_1"]["scratch_store"] == 0


def test_body_bounded_by_next_label_when_end_marker_renamed(tmp_path):
"""The next kernel's label also terminates a body, so counts stay correct
even if .Lfunc_end numbering stops lining up with declaration order."""
text = TWO_KERNEL_ISA.replace(".Lfunc_end0:", ".Lsomething_else0:")
got = irt.parse_isa(_write(tmp_path, text))
assert got["first_kernel_0"]["ds_read"] == 2 # not 3: second kernel excluded
assert got["first_kernel_0"]["vgpr"] == 16


def test_unterminated_body_reports_none(tmp_path):
"""With no terminator at all, refuse to guess rather than run to EOF."""
text = TWO_KERNEL_ISA.replace(".Lfunc_end0:", ".Lx0:").replace(".Lfunc_end1:", ".Lx1:")
got = irt.parse_isa(_write(tmp_path, text))
# second_kernel_1 is last and now has neither a .Lfunc_end nor a following label
assert got["second_kernel_1"]["ds_read"] is None
assert got["second_kernel_1"]["vgpr"] == 32 # metadata is still readable


def test_unknown_metadata_layout_reports_none(tmp_path):
"""A renamed register field must not silently read as zero."""
text = TWO_KERNEL_ISA.replace(".vgpr_count:", ".vector_gpr_count:")
got = irt.parse_isa(_write(tmp_path, text))
assert got["first_kernel_0"]["vgpr"] is None


def test_collect_qualifies_only_ambiguous_names(tmp_path):
d = tmp_path / "first_kernel_0"
d.mkdir()
_write(d, TWO_KERNEL_ISA)
got = irt.collect(str(tmp_path))
assert "first_kernel_0" in got
assert "first_kernel_0::second_kernel_1" in got


def _diff(tmp_path, before, after):
a, b = tmp_path / "b.json", tmp_path / "a.json"
a.write_text(json.dumps(before))
b.write_text(json.dumps(after))
return irt.do_diff(str(a), str(b))


_OK = {"vgpr": 10, "sgpr": 10, "spill": 0, "scratch_store": 0, "scratch_load": 0, "ds_read": 0}


def test_diff_clean_run_succeeds(tmp_path):
assert _diff(tmp_path, {"k": dict(_OK)}, {"k": dict(_OK)}) == 0


def test_diff_reports_regression(tmp_path):
worse = dict(_OK, vgpr=12)
assert _diff(tmp_path, {"k": dict(_OK)}, {"k": worse}) == 1


def test_diff_ignores_improvement(tmp_path):
better = dict(_OK, vgpr=8)
assert _diff(tmp_path, {"k": dict(_OK)}, {"k": better}) == 0


@pytest.mark.parametrize(
"before,after,why",
[
({}, {}, "no kernels on either side"),
({"k": dict(_OK)}, {}, "after side empty"),
({"k": dict(_OK)}, {"other": dict(_OK)}, "kernel only on one side"),
({"k": dict(_OK, vgpr=None)}, {"k": dict(_OK, vgpr=None)}, "unparsed metric"),
],
)
def test_diff_fails_when_data_is_untrustworthy(tmp_path, before, after, why):
"""Cannot distinguish 'no regressions' from 'no data' -- must not return 0."""
assert _diff(tmp_path, before, after) == 2, why
Loading