Skip to content
Merged
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
39 changes: 33 additions & 6 deletions src/mldebug/client_debug.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
from mldebug.batch_runner import BatchRunner
from mldebug.debug_state import DebugState
from mldebug.debug_server import DebugServer
from mldebug.extra.kernel_info import format_kernel_info
from mldebug.extra.layer_report import format_layer_report
from mldebug.interactive_controller import InteractiveController
from mldebug.layer_info import LayerInfo
from mldebug.memory_dumper import MemoryDumper
Expand Down Expand Up @@ -248,22 +250,47 @@ def print_current_state(self, layer_order=None):
info_layer = self.state.get_layer_by_order(layer_order)
if info_layer:
print(f"{sep}\nInformation on layer: {layer_order}\n{sep}")
print(info_layer)
print(f"#Iterations: {info_layer.lcp.num_iter}")
print(info_layer.format_stamps())
self.print_kernel_functions(info_layer)
print(sep)
else:
print(f"Layer not found: {layer_order}. Note: TG Layers aren't supported.")
return

self.design_info.print_info()
if self.args.aie_only:
return

layer = self.state.get_current_layer()
if layer:
stamp_names = ", ".join([f"Stamp {i}: {stamp.name}" for i, stamp in enumerate(layer.stamps)])
LOGGER.log(f"Stopped at Start of Kernel(s): {stamp_names}")
LOGGER.log(f"Current Layer: {layer.layer_order}, Iteration: {self.state.cur_it}")
LOGGER.log(str(layer))
LOGGER.log(
f"Current Layer: {layer.layer_order}, Current Iteration: {self.state.cur_it}, "
f"#Iterations: {layer.lcp.num_iter}"
)
LOGGER.log(f"Stopped at Start of Kernel(s):\n{layer.format_stamps()}")
self.print_kernel_functions(layer)

def print_kernel_functions(self, layer):
"""
Print the subfunction call tree of a TG layer's kernel; other layers have none.
"""
if not layer.lcp.is_tg:
return
kernels = self.design_info.work_dir.get_kernel_info(layer.stamps)
if kernels:
LOGGER.log(format_kernel_info(kernels))

def dump_layers(self, filename=None):
"""
Print a text report of every debuggable layer, or write it to filename.
"""
report = format_layer_report(self.design_info)
if not filename:
print(report)
return
with open(filename, "w", encoding="utf-8") as fd:
fd.write(report + "\n")
print(f"[INFO] Layer report written to {filename}")

def read_lcp(self, col=None, row=None, ping=1):
"""
Expand Down
2 changes: 1 addition & 1 deletion src/mldebug/extra/aie_guidance.py
Original file line number Diff line number Diff line change
Expand Up @@ -479,7 +479,7 @@ def print_results(self, show_passed: bool = False, show_guidance: bool = True) -

# Show guidance and values if requested and failed
if show_guidance and not result.passed:
print(f" | {'':10} | {'':35} | {result.guidance}")
print(f" | {'':10} | {'':35} | -> {result.guidance}")
if result.actual_value is not None:
print(
f" | {'':10} | {'':35} | Actual: {result.actual_value}, Expected: {result.expected_value}"
Expand Down
12 changes: 6 additions & 6 deletions src/mldebug/extra/calltree.py
Original file line number Diff line number Diff line change
Expand Up @@ -249,8 +249,8 @@ def _visualize_tree(self, node, prefix="", is_last=True, is_root=True):
connector = ""
new_prefix = ""
else:
connector = "└── " if is_last else "├── "
new_prefix = prefix + (" " if is_last else " ")
connector = "|-- "
new_prefix = prefix + (" " if is_last else "| ")

tail_marker = " [tail-call]" if node.is_tail_call else ""
func_display = f"{node.func_name} (0x{node.pc:x}){tail_marker}"
Expand Down Expand Up @@ -335,8 +335,8 @@ def get_calltree(self, root_func=None, include_summary=False):

for root_addr in root_addrs:
root_name = self._addr_to_name.get(root_addr, f"<0x{root_addr:x}>")
output.append(f"\n┌─ Call tree for: {root_name}")
output.append("")
output.append(f"\n+- Call tree for: {root_name}")
output.append("|")

tree = self._build_call_tree(root_addr)
output.append(self._visualize_tree(tree))
Expand All @@ -362,11 +362,11 @@ def get_call_relationships(self):
lines.append(f"\n{func.name} (0x{addr:x}):")
for call_pc, target in func.calls:
target_name = self._addr_to_name.get(target, f"<unknown@0x{target:x}>")
lines.append(f" ├─ calls {target_name} at PC 0x{call_pc:x}")
lines.append(f" |- calls {target_name} at PC 0x{call_pc:x}")
if func.tail_jump_target and func.tail_jump_target in self._addr_to_name:
target_name = self._addr_to_name[func.tail_jump_target]
if not target_name.startswith("."):
lines.append(f" └─ tail-calls {target_name}")
lines.append(f" |- tail-calls {target_name}")

return "\n".join(lines)

Expand Down
279 changes: 279 additions & 0 deletions src/mldebug/extra/kernel_info.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,279 @@
# SPDX-License-Identifier: Apache-2.0
# Copyright (C) 2024-2026 Advanced Micro Devices, Inc. All rights reserved.

"""
Rebuild a kernel's call tree from the compiler's build artifacts.

Some superkernels fuse many ops behind a single wrapper that the debugger can
only break on as a whole -- templated-graph (TG) kernels today. The caller
decides which layers are worth breaking down; this module just builds the tree
for whatever kernel entry PC it is handed, so the interactive 'i'/info() command
can show what a layer's kernel is actually made of.

Where the call edges come from depends on the compiler:

Chess. The bridge map file's PM section lists every function with its address
range, stack frame and callees, which is all we need:

0x000009e0..0x00000b57 ( 376 items) : <obj>::<symbol> (Function, Global, .text) ...

Called functions : <symbol>
<symbol>

Peano. The lld map file is a plain symbol table with no call graph in it at
all, so the edges come from the disassembly instead -- the `jl` targets that
AIECallTree already recovers. Sizes and stack frames are map-only, so they read
as unknown for Peano designs.
"""

import re
import textwrap
from collections import namedtuple
from dataclasses import dataclass, field

from mldebug.extra.calltree import AIECallTree

_MEM_SECTION = re.compile(r"^Memory map for memory '(\S+)':")
_ENTRY = re.compile(r"^\s+0x([0-9a-f]+)\.\.0x[0-9a-f]+\s+\(\s*(\d+) items\)\s*:\s*(.+?)\s*$")
_FUNCTION = re.compile(
r"^(?P<sym>\S+)\s+\(Function,\s*\w+,\s*\.text\S*\)"
r"(?:\s+\(stack frame size = (?P<stack>\d+)\))?$"
)
_CALLED = re.compile(r"^\s+Called functions\s*:\s*(\S+)\s*$")
_CALLED_MORE = re.compile(r"^\s{8,}(\S+)\s*$")
# Leftover mangling when c++filt cannot demangle: _Z[L]<len><name>, lowercased.
_MANGLED = re.compile(r"^_+z[a-z]*?(\d+)(.+)$")

_PM_SECTION = "PM"
_REPEAT_MARK = " (*)"
_EMPTY = "-"
_TRUNC_SUFFIX = "..."
_LOCATION = "on {locations}"


@dataclass
class MapFunction:
"""One PM function entry of a map file."""

symbol: str
start_addr: int
size: int
stack: int
callees: list = field(default_factory=list)


@dataclass
class KernelNode:
"""One function of a kernel, plus its position in that kernel's call tree."""

func: MapFunction
name: str
branch: str
repeated: bool
aie_func: object = None


@dataclass
class KernelInfo:
"""A kernel's call tree, and a label for where it runs."""

location: str
nodes: list = field(default_factory=list)


def parse_map_functions(map_path):
"""Parse the PM section of a map file into {symbol: MapFunction}."""
functions = {}
section = None
current = None
in_callees = False
with open(map_path, encoding="utf-8", errors="replace") as fd:
for line in fd:
line = line.rstrip("\n")
m_section = _MEM_SECTION.match(line)
if m_section:
section, current, in_callees = m_section.group(1), None, False
continue
m_called = _CALLED.match(line)
if m_called:
in_callees = current is not None
if in_callees:
current.callees.append(m_called.group(1))
continue
m_entry = _ENTRY.match(line)
if m_entry:
in_callees = False
current = _make_function(m_entry) if section == _PM_SECTION else None
if current:
functions[current.symbol] = current
continue
if in_callees:
m_more = _CALLED_MORE.match(line)
if m_more:
current.callees.append(m_more.group(1))
else:
in_callees = False
return functions


def _make_function(m_entry):
"""Build a MapFunction from an entry match, or None when the entry is not a function."""
m_func = _FUNCTION.match(m_entry.group(3))
if not m_func:
return None
return MapFunction(
symbol=m_func.group("sym").rsplit("::", 1)[-1],
start_addr=int(m_entry.group(1), 16),
size=int(m_entry.group(2)),
stack=int(m_func.group("stack") or 0),
)


def _readable(name):
"""Recover the identifier from a name c++filt left mangled, using its length prefix."""
m_mangled = _MANGLED.match(name)
if not m_mangled:
return name
return m_mangled.group(2)[: int(m_mangled.group(1))] or name


def _build_tree(root_symbol, functions, name_of):
"""Flatten the call tree under root_symbol into pre-order KernelNodes."""
nodes = []
expanded = set()

def visit(symbol, branch, child_prefix):
func = functions[symbol]
repeated = symbol in expanded
nodes.append(KernelNode(func, _readable(name_of(symbol)), branch, repeated))
if repeated:
return
expanded.add(symbol)
callees = [c for c in func.callees if c in functions]
for i, callee in enumerate(callees):
last = i == len(callees) - 1
visit(
callee,
child_prefix + "|-- ",
child_prefix + (" " if last else "| "),
)

visit(root_symbol, "", "")
return nodes


def _attach_pcs(nodes, aie_functions):
"""Link each node to its AIEFunction from the work dir's parsed ELF listing."""
# Keyed on entry PC: names cannot identify a function, since the LST parser
# records locals under a debug label and template clones demangle alike.
by_pc = {func.start_pc: func for func in aie_functions}
for node in nodes:
node.aie_func = by_pc.get(node.func.start_addr)


def build_kernel_info(map_path, start_pc, aie_functions, demangle, location):
"""Call tree of the kernel entered at start_pc, or None when the map has none there."""
functions = parse_map_functions(map_path)
root = next((sym for sym, f in functions.items() if f.start_addr == start_pc), None)
if not root:
return None
nodes = _build_tree(root, functions, demangle)
_attach_pcs(nodes, aie_functions)
return KernelInfo(location, nodes)


def _functions_from_lst(lst):
"""MapFunction view of Peano disassembly, keyed by entry PC; sizes are map-only."""
functions = {}
for addr, func in AIECallTree.from_string(lst).functions.items():
callees = list(dict.fromkeys(target for _, target in func.calls))
functions[addr] = MapFunction(addr, addr, None, None, callees)
return functions


def build_kernel_info_from_lst(lst, start_pc, aie_functions, location):
"""Same as build_kernel_info, for Peano designs whose map file has no call graph."""
functions = _functions_from_lst(lst)
if start_pc not in functions:
return None
# The LST parser's own names stop at the first non-word character, so take
# them from the work dir's function database instead.
names = {func.start_pc: func.name for func in aie_functions}
nodes = _build_tree(start_pc, functions, lambda pc: names.get(pc) or f"<0x{pc:x}>")
_attach_pcs(nodes, aie_functions)
return KernelInfo(location, nodes)


_Column = namedtuple("_Column", "label width align value")


def _pc(node, attr):
"""PC from the work dir function database, 0 when the function is not in it."""
return getattr(node.aie_func, attr, 0) if node.aie_func else 0


def _hex(value):
"""Format a PC as hex, or '-' when it is unknown."""
return f"0x{value:06x}" if value else _EMPTY


def _label(node):
"""Tree prefix plus function name, marking calls whose subtree was already shown."""
return node.branch + node.name + (_REPEAT_MARK if node.repeated else "")


# FUNCTION is wide enough for a 3-deep tree prefix plus a templated kernel name.
_COLUMNS = (
_Column("FUNCTION", 56, "<", _label),
_Column("START_PC", 9, ">", lambda n: _hex(n.func.start_addr)),
_Column("END_PC", 9, ">", lambda n: _hex(_pc(n, "end_pc"))),
_Column("LOCK_REL", 9, ">", lambda n: _hex(_pc(n, "final_lock_release_pc"))),
_Column("SIZE", 6, ">", lambda n: n.func.size),
_Column("STACK", 6, ">", lambda n: n.func.stack),
)

_HEADER = " ".join(f"{c.label:{c.align}{c.width}}" for c in _COLUMNS)


def _clip(value, width):
"""Fit a cell value into width, marking truncated values with a trailing '...'."""
text = _EMPTY if value is None or value == "" else str(value)
if len(text) <= width:
return text
return text[: width - len(_TRUNC_SUFFIX)] + _TRUNC_SUFFIX


def _row(node):
"""Format one tree node as a table row."""
cells = [f"{_clip(c.value(node), c.width):{c.align}{c.width}}" for c in _COLUMNS]
return " ".join(cells)


def _group_identical(kernels):
"""Group trees that match down to every PC; a layer's stamps usually share one."""
groups = {}
for kernel in kernels:
signature = tuple(
(n.func.symbol, n.func.start_addr, _pc(n, "end_pc"), _pc(n, "final_lock_release_pc"))
for n in kernel.nodes
)
groups.setdefault(signature, []).append(kernel)
return list(groups.values())


def format_kernel_info(kernels):
"""Render kernel call trees as text, or '' when there are none."""
groups = _group_identical(kernels)
lines = []
for same in groups:
lines.append("")
# A single tree covers every stamp; only say who is who when they diverge.
if len(groups) > 1:
lines += textwrap.wrap(
_LOCATION.format(locations=", ".join(k.location for k in same)),
width=len(_HEADER),
subsequent_indent=" ",
)
lines += [_HEADER, "-" * len(_HEADER)]
lines += [_row(node) for node in same[0].nodes]
return "\n".join(lines)
Loading
Loading