CodeQL / Analyze (actions) (push) Canceled after 0s
CodeQL / Analyze (python) (push) Canceled after 0s
pre-commit / pre-commit (push) Canceled after 0s
Multi-Arch CI / setup (push) Canceled after 0s
Publish build_manylinux_x86_64 images / publish_build_manylinux_x86_64 (push) Canceled after 0s
Setup (Multi-Arch) / setup (push) Canceled after 0s
Multi-Arch CI / Linux::${{ fromJSON(needs.setup.outputs.linux_build_config || '{}').build_variant_label || 'skip' }} (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Publish no_rocm_image_ubi10 image / publish_no_rocm_image_ubi10 (push) Canceled after 0s
Multi-Arch CI / Windows::${{ fromJSON(needs.setup.outputs.windows_build_config || '{}').build_variant_label || 'skip' }} (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Multi-Arch CI / Manifest Diff (push) Canceled after 0s
TheRock Manifest Diff Report / Generate Manifest Diff Report (push) Canceled after 0s
Multi-Arch CI / CI Summary (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04 (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04_ocl_rt images / publish_no_rocm_image_ubuntu24_04_ocl_rt (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04_rocgdb (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04_rocgdb images / publish_no_rocm_image_ubuntu24_04_rocgdb (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04_ocl_rt (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Unit Tests / Unit Tests :: ubuntu-24.04 (push) Canceled after 0s
Unit Tests / Unit Tests :: windows-2022 (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04_media (push) Canceled after 0s
Unit Tests / Unit Tests Summary (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04_openmpi (push) Canceled after 0s
823 lines
28 KiB
Python
823 lines
28 KiB
Python
# MIT License
|
|
#
|
|
# Copyright (c) 2024-2026 Advanced Micro Devices, Inc. All rights reserved.
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
# of this software and associated documentation files (the "Software"), to deal
|
|
# in the Software without restriction, including without limitation the rights
|
|
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
# copies of the Software, and to permit persons to whom the Software is
|
|
# furnished to do so, subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be included in
|
|
# all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
# THE SOFTWARE.
|
|
|
|
"""Build in-memory code metadata from explicit code object inputs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import bisect
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from collections import OrderedDict
|
|
from collections.abc import Iterable
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from .code_index import CodeIndex
|
|
|
|
TOOL_VERSION = "snapshot_dwarf-1.1"
|
|
HEADER = "ISA, _, LineNumber, Source, Codeobj, Vaddr, Hit, Latency, Stall, Idle"
|
|
SEPARATOR = " -> " # matches Instruction::separator in code_printing.hpp
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CodeObject:
|
|
path: str | Path
|
|
code_object_id: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CodeArtifacts:
|
|
code_index: CodeIndex
|
|
source_paths: tuple[Path, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _SqttFuncmapEntry:
|
|
marker_id: int
|
|
kind: str
|
|
name: str
|
|
source_loc: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _SqttFuncmap:
|
|
entries: tuple[_SqttFuncmapEntry, ...]
|
|
shader_clock_bits: int
|
|
shader_clock_shift: int
|
|
extra_payload_counts: dict[int, int]
|
|
|
|
|
|
def _find_tool(name: str) -> str:
|
|
candidates: list[Path] = []
|
|
for root in _rocm_roots():
|
|
candidates.append(root / "llvm" / "bin" / name)
|
|
candidates.append(root / "bin" / name)
|
|
|
|
for found in (shutil.which(name),):
|
|
if found:
|
|
candidates.append(Path(found))
|
|
|
|
for candidate in candidates:
|
|
if candidate.is_file():
|
|
return str(candidate)
|
|
raise FileNotFoundError(f"Could not locate {name}")
|
|
|
|
|
|
def _rocm_roots() -> list[Path]:
|
|
roots: list[Path] = []
|
|
seen: set[Path] = set()
|
|
for var in ("ROCM_HOME", "ROCM_PATH"):
|
|
value = os.environ.get(var)
|
|
if value:
|
|
root = Path(value).expanduser()
|
|
if root not in seen:
|
|
roots.append(root)
|
|
seen.add(root)
|
|
|
|
default = Path("/opt/rocm")
|
|
if default not in seen:
|
|
roots.append(default)
|
|
return roots
|
|
|
|
|
|
_LLVM_OBJDUMP: str | None = None
|
|
|
|
|
|
def llvm_objdump() -> str:
|
|
"""Resolve llvm-objdump lazily so the module can be imported without it."""
|
|
global _LLVM_OBJDUMP
|
|
if _LLVM_OBJDUMP is None:
|
|
_LLVM_OBJDUMP = _find_tool("llvm-objdump")
|
|
return _LLVM_OBJDUMP
|
|
|
|
|
|
def is_elf(path: str | Path) -> bool:
|
|
try:
|
|
with Path(path).open("rb") as f:
|
|
return f.read(4) == b"\x7fELF"
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SQTT marker metadata
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _parse_u32(value: str) -> int | None:
|
|
if not value or any(char < "0" or char > "9" for char in value):
|
|
return None
|
|
parsed = int(value)
|
|
return parsed if parsed <= 0xFFFFFFFF else None
|
|
|
|
|
|
def _split_funcmap_name(value: str) -> tuple[str, str]:
|
|
name, sep, source_loc = value.partition("@")
|
|
return (name, source_loc if sep else "")
|
|
|
|
|
|
def _parse_sqtt_funcmap(data: bytes) -> _SqttFuncmap:
|
|
"""Parse the subset of ``.sqtt_funcmap`` needed by code.json.
|
|
|
|
The output intentionally mirrors the C++ funcmap reader, but malformed
|
|
rows are ignored here: generating code metadata must remain best-effort
|
|
for ordinary code objects and older marker producers.
|
|
"""
|
|
|
|
entries: list[_SqttFuncmapEntry] = []
|
|
extra_payload_counts: dict[int, int] = {}
|
|
shader_clock_bits = 0
|
|
shader_clock_shift = 0
|
|
|
|
for raw_line in re.split(r"[\n\x00]", data.decode("utf-8", errors="replace")):
|
|
line = raw_line.rstrip("\r \t")
|
|
if len(line) < 2 or line[1] != ":":
|
|
continue
|
|
|
|
kind = line[0]
|
|
payload = line[2:]
|
|
if kind == "M":
|
|
bits = 0
|
|
shift = 0
|
|
saw_bits = False
|
|
saw_shift = False
|
|
valid = True
|
|
for attribute in payload.split(";"):
|
|
key, separator, value = attribute.partition("=")
|
|
if not separator:
|
|
continue
|
|
parsed = _parse_u32(value)
|
|
if key == "shader_clock_bits":
|
|
if parsed is None:
|
|
valid = False
|
|
else:
|
|
bits = parsed
|
|
saw_bits = True
|
|
elif key == "shader_clock_shift":
|
|
if parsed is None:
|
|
valid = False
|
|
else:
|
|
shift = parsed
|
|
saw_shift = True
|
|
|
|
if valid and saw_bits and (bits == 0 or saw_shift) and bits <= 29 and (
|
|
bits == 0 or (shift < 32 and bits <= 32 - shift)
|
|
):
|
|
shader_clock_bits = bits
|
|
shader_clock_shift = shift if bits else 0
|
|
continue
|
|
|
|
if kind == "R":
|
|
marker_id_text, separator, attributes = payload.partition(":")
|
|
marker_id = _parse_u32(marker_id_text) if separator else None
|
|
if marker_id is None:
|
|
continue
|
|
for attribute in attributes.split(";"):
|
|
key, equals, value = attribute.partition("=")
|
|
if key != "extra_payload_count" or not equals:
|
|
continue
|
|
count = _parse_u32(value)
|
|
if count is not None:
|
|
extra_payload_counts[marker_id] = count
|
|
continue
|
|
|
|
if kind == "K":
|
|
name, source_loc = _split_funcmap_name(payload)
|
|
if name:
|
|
entries.append(_SqttFuncmapEntry(0, kind, name, source_loc))
|
|
continue
|
|
|
|
if kind not in {"F", "U", "P"}:
|
|
continue
|
|
|
|
marker_id_text, separator, name_and_source = payload.partition(":")
|
|
marker_id = _parse_u32(marker_id_text) if separator else None
|
|
if marker_id is None:
|
|
continue
|
|
name, source_loc = _split_funcmap_name(name_and_source)
|
|
if name:
|
|
entries.append(_SqttFuncmapEntry(marker_id, kind, name, source_loc))
|
|
|
|
return _SqttFuncmap(
|
|
entries=tuple(entries),
|
|
shader_clock_bits=shader_clock_bits,
|
|
shader_clock_shift=shader_clock_shift,
|
|
extra_payload_counts=extra_payload_counts,
|
|
)
|
|
|
|
|
|
def _read_sqtt_funcmap(elf_path: str) -> _SqttFuncmap | None:
|
|
"""Return embedded marker metadata, or ``None`` when no section exists."""
|
|
|
|
from elftools.common import exceptions as elftools_exceptions
|
|
from elftools.elf.elffile import ELFFile
|
|
|
|
try:
|
|
with Path(elf_path).open("rb") as file:
|
|
section = ELFFile(file).get_section_by_name(".sqtt_funcmap")
|
|
if section is None:
|
|
return None
|
|
data = section.data()
|
|
except (OSError, elftools_exceptions.ELFError, ValueError):
|
|
return None
|
|
except Exception:
|
|
LOGGER.debug("Unexpected error reading .sqtt_funcmap from %s", elf_path, exc_info=True)
|
|
return None
|
|
|
|
return _parse_sqtt_funcmap(data) if data else None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DWARF helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _decode(b) -> str:
|
|
if isinstance(b, bytes):
|
|
return b.decode("utf-8", errors="replace")
|
|
return b or ""
|
|
|
|
|
|
def _resolve_file(line_program, file_index: int) -> str | None:
|
|
"""Resolve a DWARF file index to a full path using the CU line program."""
|
|
if line_program is None:
|
|
return None
|
|
header = line_program.header
|
|
file_entries = header.get("file_entry", [])
|
|
include_dirs = header.get("include_directory", [])
|
|
|
|
# In DWARF <=4 file_entry is 1-based; in DWARF 5 it's 0-based.
|
|
version = header.get("version", 4)
|
|
idx = file_index if version >= 5 else file_index - 1
|
|
if idx < 0 or idx >= len(file_entries):
|
|
return None
|
|
fe = file_entries[idx]
|
|
name = _decode(fe.name)
|
|
d_idx = fe.dir_index
|
|
# Same indexing rule for include_directory.
|
|
if version >= 5:
|
|
d = _decode(include_dirs[d_idx]) if 0 <= d_idx < len(include_dirs) else ""
|
|
else:
|
|
if d_idx == 0:
|
|
d = ""
|
|
else:
|
|
j = d_idx - 1
|
|
d = _decode(include_dirs[j]) if 0 <= j < len(include_dirs) else ""
|
|
if d and not os.path.isabs(name):
|
|
return os.path.normpath(os.path.join(d, name))
|
|
return os.path.normpath(name)
|
|
|
|
|
|
def _get_high_pc(die, low_pc: int | None) -> int | None:
|
|
if "DW_AT_high_pc" not in die.attributes:
|
|
return None
|
|
attr = die.attributes["DW_AT_high_pc"]
|
|
form = attr.form
|
|
if form == "DW_FORM_addr":
|
|
return attr.value
|
|
if form.startswith("DW_FORM_data") or form in (
|
|
"DW_FORM_udata",
|
|
"DW_FORM_sdata",
|
|
"DW_FORM_implicit_const",
|
|
):
|
|
return (low_pc or 0) + attr.value
|
|
return attr.value
|
|
|
|
|
|
def _get_die_ranges(die, dwarfinfo, cu) -> list[tuple[int, int]]:
|
|
"""Return list of (begin, end) for a DIE, handling DW_AT_ranges and lo/hi."""
|
|
if "DW_AT_ranges" in die.attributes:
|
|
rl_reader = dwarfinfo.range_lists()
|
|
if rl_reader is None:
|
|
return []
|
|
offset = die.attributes["DW_AT_ranges"].value
|
|
try:
|
|
entries = rl_reader.get_range_list_at_offset(offset, cu=cu)
|
|
except TypeError:
|
|
entries = rl_reader.get_range_list_at_offset(offset)
|
|
# CU base address (default 0).
|
|
base = 0
|
|
top = cu.get_top_DIE()
|
|
if "DW_AT_low_pc" in top.attributes:
|
|
base = top.attributes["DW_AT_low_pc"].value
|
|
|
|
ranges: list[tuple[int, int]] = []
|
|
for e in entries or []:
|
|
tname = type(e).__name__
|
|
if tname == "BaseAddressEntry":
|
|
base = e.base_address
|
|
elif tname == "RangeEntry":
|
|
# pyelftools: is_absolute means begin/end are absolute addrs.
|
|
is_abs = getattr(e, "is_absolute", False)
|
|
begin = e.begin_offset
|
|
end = e.end_offset
|
|
if not is_abs:
|
|
begin += base
|
|
end += base
|
|
if end > begin:
|
|
ranges.append((begin, end))
|
|
return ranges
|
|
|
|
if "DW_AT_low_pc" in die.attributes:
|
|
low = die.attributes["DW_AT_low_pc"].value
|
|
high = _get_high_pc(die, low)
|
|
if high is not None and high > low:
|
|
return [(low, high)]
|
|
return []
|
|
|
|
|
|
class _Inlined:
|
|
"""Mirror of DIEInfo for a single CU's DIE tree."""
|
|
|
|
__slots__ = (
|
|
"ranges",
|
|
"children",
|
|
"total_low",
|
|
"total_high",
|
|
"children_low",
|
|
"children_high",
|
|
"file_and_line",
|
|
"is_inlined",
|
|
)
|
|
|
|
def __init__(self):
|
|
self.ranges: list[tuple[int, int]] = []
|
|
self.children: list[_Inlined] = []
|
|
self.total_low = 1 << 63
|
|
self.total_high = 0
|
|
self.children_low = 1 << 63
|
|
self.children_high = 0
|
|
self.file_and_line: str = ""
|
|
self.is_inlined = False
|
|
|
|
def add_range(self, lo: int, hi: int) -> None:
|
|
self.ranges.append((lo, hi))
|
|
if lo < self.total_low:
|
|
self.total_low = lo
|
|
if hi > self.total_high:
|
|
self.total_high = hi
|
|
|
|
def expand_children(self, lo: int, hi: int) -> None:
|
|
if lo < self.children_low:
|
|
self.children_low = lo
|
|
if hi > self.children_high:
|
|
self.children_high = hi
|
|
|
|
def contains_total(self, addr: int) -> bool:
|
|
return self.total_low <= addr < self.total_high
|
|
|
|
def contains_children(self, addr: int) -> bool:
|
|
return self.children_low <= addr < self.children_high
|
|
|
|
def get_call_stack(self, addr: int, out: list[str]) -> bool:
|
|
if not self.contains_children(addr):
|
|
return False
|
|
added = False
|
|
for c in self.children:
|
|
if c.get_call_stack(addr, out):
|
|
added = True
|
|
break
|
|
if self.is_inlined and self.contains_total(addr):
|
|
for lo, hi in self.ranges:
|
|
if lo <= addr < hi:
|
|
out.append(self.file_and_line)
|
|
return True
|
|
return added
|
|
|
|
|
|
def _build_die_tree(die, dwarfinfo, cu, line_program) -> _Inlined:
|
|
info = _Inlined()
|
|
if die.tag == "DW_TAG_inlined_subroutine":
|
|
info.is_inlined = True
|
|
for lo, hi in _get_die_ranges(die, dwarfinfo, cu):
|
|
info.add_range(lo, hi)
|
|
# Resolve call-site (where this inlined subroutine was called).
|
|
call_file_attr = die.attributes.get("DW_AT_call_file")
|
|
call_line_attr = die.attributes.get("DW_AT_call_line")
|
|
if call_file_attr is not None and call_line_attr is not None:
|
|
fname = _resolve_file(line_program, call_file_attr.value)
|
|
if fname:
|
|
info.file_and_line = f"{fname}:{call_line_attr.value}"
|
|
# Always include this node's own range into children_range so parents
|
|
# can find it via children_range (matches the C++ comment).
|
|
info.children_low = info.total_low
|
|
info.children_high = info.total_high
|
|
|
|
for child in die.iter_children():
|
|
sub = _build_die_tree(child, dwarfinfo, cu, line_program)
|
|
if sub.children_high > sub.children_low:
|
|
info.children.append(sub)
|
|
info.expand_children(sub.children_low, sub.children_high)
|
|
return info
|
|
|
|
|
|
def build_address_ranges(elf_path: str) -> list[tuple[int, int, str]]:
|
|
"""Return sorted ``(begin, end, comment)`` source ranges."""
|
|
from elftools.common import exceptions as elftools_exceptions
|
|
from elftools.elf.elffile import ELFFile
|
|
|
|
expected_errors = (
|
|
OSError,
|
|
elftools_exceptions.ELFError,
|
|
getattr(elftools_exceptions, "DWARFError", elftools_exceptions.ELFError),
|
|
AssertionError,
|
|
KeyError,
|
|
IndexError,
|
|
ValueError,
|
|
)
|
|
range_map: list[tuple[int, int, str]] = []
|
|
|
|
with Path(elf_path).open("rb") as f:
|
|
try:
|
|
elf = ELFFile(f)
|
|
if not elf.has_dwarf_info():
|
|
return []
|
|
dwarfinfo = elf.get_dwarf_info()
|
|
cu_iter = iter(dwarfinfo.iter_CUs())
|
|
except expected_errors:
|
|
return []
|
|
except Exception:
|
|
LOGGER.debug("Unexpected error reading DWARF metadata from %s", elf_path, exc_info=True)
|
|
return []
|
|
|
|
while True:
|
|
try:
|
|
cu = next(cu_iter)
|
|
except StopIteration:
|
|
break
|
|
except expected_errors:
|
|
break
|
|
except Exception:
|
|
LOGGER.debug("Unexpected error reading a DWARF CU from %s", elf_path, exc_info=True)
|
|
break
|
|
|
|
try:
|
|
line_program = dwarfinfo.line_program_for_CU(cu)
|
|
except expected_errors:
|
|
line_program = None
|
|
except Exception:
|
|
LOGGER.debug(
|
|
"Unexpected error reading a DWARF line program from %s", elf_path, exc_info=True
|
|
)
|
|
line_program = None
|
|
if line_program is None:
|
|
continue
|
|
|
|
# Build inlined-subroutine DIE tree once per CU.
|
|
try:
|
|
top = cu.get_top_DIE()
|
|
die_root = _build_die_tree(top, dwarfinfo, cu, line_program)
|
|
except expected_errors:
|
|
continue
|
|
except Exception:
|
|
LOGGER.debug("Unexpected error reading DWARF DIEs from %s", elf_path, exc_info=True)
|
|
continue
|
|
|
|
# Walk the line program; each entry covers [addr, next_addr).
|
|
try:
|
|
entries = list(line_program.get_entries())
|
|
except expected_errors:
|
|
continue
|
|
except Exception:
|
|
LOGGER.debug(
|
|
"Unexpected error reading DWARF line entries from %s", elf_path, exc_info=True
|
|
)
|
|
continue
|
|
states = [(e.state, e) for e in entries if e.state is not None]
|
|
line_ranges: dict[int, tuple[int, str]] = {}
|
|
for i in range(len(states) - 1):
|
|
st, _ = states[i]
|
|
nxt, _ = states[i + 1]
|
|
if st.end_sequence:
|
|
continue
|
|
addr = st.address
|
|
end_addr = nxt.address
|
|
if end_addr <= addr:
|
|
continue
|
|
src = _resolve_file(line_program, st.file)
|
|
if not src:
|
|
continue
|
|
line_str = f"{src}:{st.line}" if st.line != 0 else f"{src}:?"
|
|
line_ranges[addr] = (end_addr, line_str)
|
|
|
|
for addr, (end_addr, line_str) in line_ranges.items():
|
|
stack: list[str] = []
|
|
die_root.get_call_stack(addr, stack)
|
|
if stack:
|
|
line_str = line_str + SEPARATOR + SEPARATOR.join(stack)
|
|
range_map.append((addr, end_addr, line_str))
|
|
|
|
# Sort by start address; later we'll binary-search.
|
|
range_map.sort(key=lambda t: t[0])
|
|
return range_map
|
|
|
|
|
|
def lookup_range(ranges: list[tuple[int, int, str]], addr: int) -> str:
|
|
"""Return the comment for the range containing addr, or ''."""
|
|
if not ranges:
|
|
return ""
|
|
i = bisect.bisect_right(ranges, addr, key=lambda row: row[0]) - 1
|
|
if i < 0:
|
|
return ""
|
|
begin, end, text = ranges[i]
|
|
if begin <= addr < end:
|
|
return text
|
|
return ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Disassembly
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Instruction lines: leading whitespace, mnemonic+operands, "// HEX: HEX..."
|
|
_RE_INST = re.compile(r"^\s+(.*?)\s*//\s*([0-9A-Fa-f]+):\s*(.+)\s*$")
|
|
_RE_DECIMAL = re.compile(r"^-?\d+$")
|
|
|
|
|
|
def _normalize_branch_operand(inst: str, raw_words: str) -> str:
|
|
"""Make branch targets compatible with the decoder stitcher.
|
|
|
|
The stitcher expects the last token of branch instructions to be the SOPP
|
|
16-bit branch displacement in decimal form. llvm-objdump may print symbolic
|
|
labels for ELF-local branch targets, e.g. ``s_cbranch_vccz label``. The
|
|
encoded immediate is still present in the raw instruction word, so use that
|
|
as the source of truth. Keep the raw unsigned 16-bit value to match the
|
|
decoder/control CSV disassembly; the stitcher converts values >= 32768 to
|
|
signed internally.
|
|
"""
|
|
if "branch" not in inst:
|
|
return inst
|
|
|
|
parts = inst.rsplit(None, 1)
|
|
if len(parts) != 2:
|
|
return inst
|
|
|
|
if _RE_DECIMAL.match(parts[1]):
|
|
return inst
|
|
|
|
first_word = raw_words.split()[0] if raw_words.split() else ""
|
|
if len(first_word) < 4:
|
|
return inst
|
|
|
|
try:
|
|
imm = int(first_word[-4:], 16)
|
|
except ValueError:
|
|
return inst
|
|
return f"{parts[0]} {imm}"
|
|
|
|
|
|
def read_symbol_labels(elf_path: str) -> tuple[dict[int, tuple[str, str]], dict[int, list[str]]]:
|
|
"""Return (function labels, debug labels) keyed by virtual address."""
|
|
from elftools.elf.elffile import ELFFile
|
|
|
|
functions: dict[int, tuple[str, str]] = {}
|
|
labels: dict[int, list[str]] = {}
|
|
with Path(elf_path).open("rb") as f:
|
|
elf = ELFFile(f)
|
|
sections = [
|
|
section
|
|
for name in (".symtab", ".dynsym")
|
|
if (section := elf.get_section_by_name(name)) is not None
|
|
]
|
|
|
|
for section in sections:
|
|
for symbol in section.iter_symbols():
|
|
name = symbol.name
|
|
if not name:
|
|
continue
|
|
vaddr = int(symbol["st_value"])
|
|
symbol_type = symbol["st_info"]["type"]
|
|
if symbol_type == "STT_FUNC" and vaddr not in functions:
|
|
functions[vaddr] = (name, name)
|
|
elif symbol_type == "STT_NOTYPE" and name.startswith("label"):
|
|
shndx = symbol["st_shndx"]
|
|
if shndx not in ("SHN_UNDEF", "SHN_ABS"):
|
|
labels.setdefault(vaddr, [])
|
|
if name not in labels[vaddr]:
|
|
labels[vaddr].append(name)
|
|
return functions, labels
|
|
|
|
|
|
def parse_disassembly(elf_path: str) -> list[tuple[int, str]]:
|
|
"""Return [(vaddr, instruction_text)] from llvm-objdump output."""
|
|
proc = subprocess.run(
|
|
[llvm_objdump(), "-d", "--no-show-raw-insn", elf_path],
|
|
capture_output=True,
|
|
text=True,
|
|
errors="replace",
|
|
)
|
|
if proc.returncode != 0:
|
|
detail = (proc.stderr or proc.stdout).strip()
|
|
message = f"llvm-objdump failed for {elf_path} with exit code {proc.returncode}"
|
|
if detail:
|
|
message = f"{message}: {detail}"
|
|
raise RuntimeError(message)
|
|
|
|
insts: list[tuple[int, str]] = []
|
|
|
|
for raw in proc.stdout.splitlines():
|
|
line = raw.rstrip()
|
|
if not line:
|
|
continue
|
|
if line.lstrip().startswith(";"):
|
|
continue
|
|
im = _RE_INST.match(line)
|
|
if im:
|
|
inst = _normalize_branch_operand(im.group(1).strip(), im.group(3))
|
|
vaddr = int(im.group(2), 16)
|
|
insts.append((vaddr, inst))
|
|
return insts
|
|
|
|
|
|
def _code_label(name: str) -> str:
|
|
# RCV recognizes semicolon label rows only when the text starts with "; _".
|
|
# Keep the real symbol in kernels[]; this marker affects only the code row.
|
|
return f"; {name}" if name.startswith("_") else f"; _{name}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Snapshots
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_RE_PATH_LINE = re.compile(r"((?:/|\.{1,2}/)?[\w./+\-]+\.\w+):(\d+|\?)")
|
|
|
|
|
|
def extract_paths_from_comments(comments: list[str]) -> list[str]:
|
|
found: OrderedDict[str, None] = OrderedDict()
|
|
for c in comments:
|
|
for m in _RE_PATH_LINE.finditer(c):
|
|
found.setdefault(os.path.normpath(m.group(1)), None)
|
|
return list(found.keys())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Driver
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _normalize_code_objects(code_objects: Iterable[CodeObject]) -> list[CodeObject]:
|
|
out: list[CodeObject] = []
|
|
seen_ids: dict[int, Path] = {}
|
|
for code_object in code_objects:
|
|
path = Path(code_object.path).expanduser().resolve()
|
|
code_object_id = int(code_object.code_object_id)
|
|
if not path.is_file():
|
|
raise FileNotFoundError(path)
|
|
if not is_elf(path):
|
|
raise ValueError(f"Not an ELF code object: {path}")
|
|
if code_object_id in seen_ids:
|
|
raise ValueError(
|
|
f"Code object id {code_object_id} is used by both "
|
|
f"{seen_ids[code_object_id]} and {path}."
|
|
)
|
|
seen_ids[code_object_id] = path
|
|
out.append(CodeObject(path, code_object_id))
|
|
if not out:
|
|
raise ValueError("No code objects were provided.")
|
|
return out
|
|
|
|
|
|
def generate_code_artifacts(code_objects: Iterable[CodeObject]) -> CodeArtifacts:
|
|
"""Build RCV code metadata and source snapshot inputs from code objects.
|
|
|
|
The caller supplies explicit code object ids. This function disassembles
|
|
each ELF, combines that ISA with DWARF source ranges and symbol labels, and
|
|
returns an in-memory CodeIndex plus the source paths referenced by the ISA
|
|
comments.
|
|
"""
|
|
normalized = _normalize_code_objects(code_objects)
|
|
rows: list[list] = []
|
|
comments: list[str] = []
|
|
kernels_out: list[dict] = []
|
|
sqtt_funcmap_rows: list[list[object]] = []
|
|
sqtt_funcmap_layouts: list[list[int]] = []
|
|
sqtt_funcmap_payloads: list[list[int]] = []
|
|
line_no = 1
|
|
|
|
for code_object in normalized:
|
|
elf_path = str(code_object.path)
|
|
codeobj_id = code_object.code_object_id
|
|
range_map = build_address_ranges(elf_path)
|
|
insts = parse_disassembly(elf_path)
|
|
kernels, debug_labels = read_symbol_labels(elf_path)
|
|
funcmap = _read_sqtt_funcmap(elf_path)
|
|
|
|
if funcmap is not None:
|
|
# Keep the existing sqtt_funcmap row shape used by rocprofv3
|
|
# consumers. The two new side tables below are additive, so older
|
|
# readers can ignore them without losing their original schema.
|
|
if funcmap.shader_clock_bits:
|
|
sqtt_funcmap_layouts.append(
|
|
[codeobj_id, funcmap.shader_clock_bits, funcmap.shader_clock_shift]
|
|
)
|
|
symbol_vaddrs: dict[str, int] = {}
|
|
for address, (name, _demangled) in sorted(kernels.items()):
|
|
symbol_vaddrs.setdefault(name, address)
|
|
for entry in funcmap.entries:
|
|
vaddr = symbol_vaddrs.get(entry.name, 0) if entry.kind in {"F", "K"} else 0
|
|
sqtt_funcmap_rows.append(
|
|
[codeobj_id, entry.marker_id, entry.kind, entry.name, entry.source_loc, vaddr]
|
|
)
|
|
payload_count = funcmap.extra_payload_counts.get(entry.marker_id, 0)
|
|
if entry.kind != "K" and payload_count:
|
|
sqtt_funcmap_payloads.append([codeobj_id, entry.marker_id, payload_count])
|
|
|
|
for vaddr, inst in insts:
|
|
# Kernel and debug labels are emitted as code rows immediately
|
|
# before the instruction at the same vaddr so branch targets in RCV
|
|
# can land on the following ISA line.
|
|
kernel = kernels.get(vaddr)
|
|
if kernel is not None:
|
|
kernel_name, kernel_demangled = kernel
|
|
rows.append(
|
|
[
|
|
_code_label(kernel_name),
|
|
0,
|
|
line_no,
|
|
kernel_demangled,
|
|
codeobj_id,
|
|
vaddr,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
]
|
|
)
|
|
line_no += 1
|
|
for label in debug_labels.get(vaddr, []):
|
|
rows.append(
|
|
[
|
|
_code_label(label),
|
|
0,
|
|
line_no,
|
|
"",
|
|
codeobj_id,
|
|
vaddr,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
]
|
|
)
|
|
line_no += 1
|
|
|
|
comment = lookup_range(range_map, vaddr)
|
|
comments.append(comment)
|
|
rows.append([inst, 0, line_no, comment, codeobj_id, vaddr, 0, 0, 0, 0])
|
|
line_no += 1
|
|
|
|
for address, (name, demangled) in sorted(kernels.items()):
|
|
kernels_out.append(
|
|
{
|
|
"address": address,
|
|
"codeobj": codeobj_id,
|
|
"name": name,
|
|
"demangled": demangled,
|
|
}
|
|
)
|
|
|
|
code_doc = {
|
|
"code": rows,
|
|
"version": TOOL_VERSION,
|
|
"header": HEADER,
|
|
"kernels": kernels_out,
|
|
"hsaco": [str(code_object.path) for code_object in normalized],
|
|
}
|
|
if sqtt_funcmap_rows:
|
|
code_doc["sqtt_funcmap"] = sqtt_funcmap_rows
|
|
if sqtt_funcmap_layouts:
|
|
code_doc["sqtt_funcmap_layout"] = sqtt_funcmap_layouts
|
|
if sqtt_funcmap_payloads:
|
|
code_doc["sqtt_funcmap_payloads"] = sqtt_funcmap_payloads
|
|
|
|
source_paths = tuple(Path(path) for path in extract_paths_from_comments(comments))
|
|
return CodeArtifacts(CodeIndex.from_document(code_doc), source_paths)
|
|
|
|
|
|
def generate_code_index(code_objects: Iterable[CodeObject]) -> CodeIndex:
|
|
return generate_code_artifacts(code_objects).code_index
|