path: root/tools/perf/python/arm-cs-trace-disasm.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
"""
arm-cs-trace-disasm.py: ARM CoreSight Trace Dump With Disassembler

Author: Tor Jeremiassen <tor@ti.com>
        Mathieu Poirier <mathieu.poirier@linaro.org>
        Leo Yan <leo.yan@linaro.org>
        Al Grant <Al.Grant@arm.com>

Below are some example commands for using this script.
Note a --kcore recording is required for accurate decode
due to the alternatives patching mechanism. In addition to this,
source line info comes from Perf, and when using kcore there is
no debug info. The following lists the supported features in each mode:

+-----------+-----------------+------------------+------------------+
| Recording | Accurate decode | Source line dump | Disassembly dump |
+-----------+-----------------+------------------+------------------+
| --kcore   | yes             | no               | yes              |
| normal    | no              | yes (inaccurate) | yes (inaccurate) |
+-----------+-----------------+------------------+------------------+

Output disassembly with objdump and auto detect vmlinux
(when running on same machine):
 perf script arm-cs-trace-disasm -- -d

Output disassembly with llvm-objdump:
 perf script arm-cs-trace-disasm -- -d llvm-objdump-11 -k path/to/vmlinux

Output accurate disassembly by passing kcore to script:
 perf script arm-cs-trace-disasm -- -d -k perf.data/kcore_dir/kcore

Output only source line and symbols:
 perf script arm-cs-trace-disasm
"""
from __future__ import annotations

import os
from os import path
import re
from subprocess import CalledProcessError, check_output
import argparse
import platform
import sys
from typing import Dict, List, Optional

import perf

# Initialize global dicts and regular expression
DISASM_CACHE: Dict[str, List[str]] = {}
DISASM_RE = re.compile(r"^\s*([0-9a-fA-F]+):")
DISASM_FUNC_RE = re.compile(r"^\s*([0-9a-fA-F]+)\s.*:")
CACHE_SIZE = 1024
class _State:
    sample_idx: int = -1
    source_file_name: Optional[str] = None
    line_number: Optional[int] = None
    dso: Optional[str] = None

_STATE = _State()

KVER = platform.release()
VMLINUX_PATHS = [
    f"/usr/lib/debug/boot/vmlinux-{KVER}.debug",
    f"/usr/lib/debug/lib/modules/{KVER}/vmlinux",
    f"/lib/modules/{KVER}/build/vmlinux",
    f"/usr/lib/debug/boot/vmlinux-{KVER}",
    f"/boot/vmlinux-{KVER}",
    "/boot/vmlinux",
    "vmlinux",
]

def default_objdump() -> str:
    """Return the default objdump path from perf config or 'objdump'."""
    try:
        config = perf.config_get("annotate.objdump")
        return str(config) if config else "objdump"
    except (AttributeError, TypeError):
        return "objdump"

def find_vmlinux() -> Optional[str]:
    """Find the vmlinux file in standard paths."""
    if hasattr(find_vmlinux, "path"):
        return getattr(find_vmlinux, "path")

    for v in VMLINUX_PATHS:
        if os.access(v, os.R_OK):
            setattr(find_vmlinux, "path", v)
            return v
    setattr(find_vmlinux, "path", None)
    return None

def get_dso_file_path(dso_name: str, dso_build_id: str, vmlinux: Optional[str]) -> str:
    """Return the path to the DSO file."""
    # Locate DSO binaries in the perf build-id cache (~/.debug or PERF_BUILDID_DIR).
    # Perf stores cached binaries under both <buildid_dir>/<dso_long_name>/<build_id>/elf
    # and the canonical symlink/file layout <buildid_dir>/.build-id/<bid[:2]>/<bid[2:]>/elf.
    buildid_dir = os.environ.get('PERF_BUILDID_DIR')
    if not buildid_dir:
        buildid_dir = os.path.join(os.environ.get('HOME', ''), '.debug')

    if (dso_name in ("[kernel.kallsyms]", "vmlinux", "kcore")
            or dso_name.endswith("/vmlinux") or dso_name.endswith("/kcore")):
        if vmlinux:
            return vmlinux
        if dso_build_id and dso_build_id != "[unknown]" and len(dso_build_id) > 2:
            bid_candidate = os.path.join(
                buildid_dir, ".build-id", dso_build_id[:2], dso_build_id[2:], "elf"
            )
            if os.access(bid_candidate, os.R_OK):
                return bid_candidate
            for kname in (dso_name, "vmlinux", "[kernel.kallsyms]"):
                candidate = os.path.join(buildid_dir, kname.lstrip("/"), dso_build_id, "elf")
                if os.access(candidate, os.R_OK):
                    return candidate
        return find_vmlinux() or dso_name

    if dso_name == "[vdso]":
        append = "/vdso"
    else:
        append = "/elf"

    if dso_build_id and dso_build_id != "[unknown]" and len(dso_build_id) > 2:
        bid_path = (buildid_dir + "/.build-id/" + dso_build_id[:2] + "/"
                    + dso_build_id[2:] + append).replace('//', '/')
        if os.access(bid_path, os.R_OK):
            return bid_path

    dso_path = buildid_dir + "/" + dso_name + "/" + dso_build_id + append
    # Replace duplicate slash chars to single slash char
    dso_path = dso_path.replace('//', '/')
    return dso_path

def read_disam(dso_fname: str, dso_start: int, start_addr: int,
               stop_addr: int, objdump: str) -> List[str]:
    """Read disassembly from a DSO file using objdump."""
    if stop_addr <= start_addr or (stop_addr - start_addr) > 0x100000:
        return []
    addr_range = f"{start_addr}:{stop_addr}:{dso_start}:{dso_fname}"

    # Don't let the cache get too big, clear it when it hits max size
    if len(DISASM_CACHE) > CACHE_SIZE:
        DISASM_CACHE.clear()

    if addr_range in DISASM_CACHE:
        disasm_output = DISASM_CACHE[addr_range]
    else:
        start_addr = start_addr - dso_start
        stop_addr = stop_addr - dso_start
        disasm = [objdump, "-d", "-z",
                  f"--start-address={start_addr:#x}",
                  f"--stop-address={stop_addr:#x}"]
        disasm += [dso_fname]
        try:
            disasm_output = check_output(disasm).decode('utf-8', errors='replace').split('\n')
        except (CalledProcessError, OSError):
            return []
        if len(disasm_output) <= 512:
            DISASM_CACHE[addr_range] = disasm_output

    return disasm_output

def print_disam(dso_fname: str, dso_start: int, start_addr: int,
                stop_addr: int, objdump: str) -> None:
    """Print disassembly for a given address range."""
    for line in read_disam(dso_fname, dso_start, start_addr, stop_addr, objdump):
        m = DISASM_FUNC_RE.search(line)
        if m is None:
            m = DISASM_RE.search(line)
            if m is None:
                continue
        print(f"\t{line}")

def print_sample(sample: perf.sample_event) -> None:
    """Print sample details."""
    print(f"Sample = {{ cpu: {sample.sample_cpu:04d} addr: {sample.sample_addr:016x} "
          f"phys_addr: {sample.sample_phys_addr:016x} ip: {sample.sample_ip:016x} "
          f"pid: {sample.sample_pid} tid: {sample.sample_tid} period: {sample.sample_period} "
          f"time: {sample.sample_time} index: {_STATE.sample_idx}}}")

def common_start_str(comm: str, sample: perf.sample_event) -> str:
    """Return common start string for sample output."""
    sec = int(sample.sample_time / 1000000000)
    ns = sample.sample_time % 1000000000
    cpu = sample.sample_cpu
    pid = sample.sample_pid
    tid = sample.sample_tid
    return f"{comm:>16s} {pid:5d}/{tid:<5d} [{cpu:04d}] {sec:9d}.{ns:09d}  "

def print_srccode(comm: str, sample: perf.sample_event, symbol: str, dso: str) -> None:
    """Print source code and symbols for a sample."""
    ip = sample.sample_ip
    if symbol == "[unknown]":
        start_str = common_start_str(comm, sample) + f"{ip:x}".rjust(16).ljust(40)
    else:
        symoff = 0
        symoff = getattr(sample, 'sym_offset', 0) or 0
        offs = f"+{symoff:#x}" if symoff != 0 else ""
        start_str = common_start_str(comm, sample) + (symbol + offs).ljust(40)

    source_file_name, line_number, source_line = sample.srccode() or (None, 0, None)
    if source_file_name:
        if _STATE.line_number == line_number and _STATE.source_file_name == source_file_name:
            src_str = ""
        else:
            if len(source_file_name) > 40:
                src_file = f"...{source_file_name[-37:]} "
            else:
                src_file = source_file_name.ljust(41)

            if source_line is None:
                src_str = f"{src_file}{line_number:>4d} <source not found>"
            else:
                src_str = f"{src_file}{line_number:>4d} {source_line}"
        _STATE.dso = None
    elif dso == _STATE.dso:
        src_str = ""
    else:
        src_str = dso
        _STATE.dso = dso

    _STATE.line_number = line_number
    _STATE.source_file_name = source_file_name

    print(start_str, src_str)

class TraceDisasm:
    """Class to handle trace disassembly."""
    def __init__(self, cli_options: argparse.Namespace):
        self.options = cli_options
        self.sample_idx = -1
        self.cpu_data: Dict[str, int] = {}
        self.session: Optional[perf.session] = None

    def process_event(self, sample: perf.sample_event) -> None:
        """Process a single perf event."""
        self.sample_idx += 1
        _STATE.sample_idx = self.sample_idx

        if self.options.start_time is not None and sample.sample_time < self.options.start_time:
            return
        if self.options.stop_time is not None and sample.sample_time > self.options.stop_time:
            sys.exit(0)
        if self.options.start_sample is not None and self.sample_idx < self.options.start_sample:
            return
        if self.options.stop_sample is not None and self.sample_idx > self.options.stop_sample:
            sys.exit(0)

        ev_name = str(sample.evsel)
        if self.options.verbose:
            print(f"Event type: {ev_name}")
            print_sample(sample)

        # Initialize CPU data if it's empty, and directly return back
        # if this is the first tracing event for this CPU. This must happen
        # before the dso == '[unknown]' check because the initial
        # CS_ETM_TRACE_ON sample has ip == 0 (dso == '[unknown]') and
        # carries the starting target address in sample.sample_addr.
        if self.cpu_data.get(f"{sample.sample_cpu}addr") is None:
            self.cpu_data[f"{sample.sample_cpu}addr"] = sample.sample_addr
            return

        dso = getattr(sample, 'dso_long_name', None) or sample.dso or '[unknown]'
        symbol = sample.symbol or '[unknown]'
        dso_bid = sample.dso_bid or '[unknown]'
        dso_start = sample.map_start
        dso_end = sample.map_end
        map_pgoff = sample.map_pgoff or 0

        comm = "[unknown]"
        try:
            if self.session:
                thread_info = self.session.find_thread(sample.sample_pid, sample.sample_tid)
                if thread_info:
                    comm = thread_info.comm() or "[unknown]"
        except (TypeError, AttributeError):
            pass

        if dso == '[unknown]':
            return

        if dso_start is None or dso_end is None:
            print(f"Failed to find valid dso map for dso {dso}")
            return

        if "instructions" in ev_name:
            print_srccode(comm, sample, symbol, dso)
            return

        if "branches" not in ev_name:
            return

        self._process_branch(sample, comm, symbol, dso, dso_bid, dso_start, dso_end, map_pgoff)

    def _process_branch(self, sample: perf.sample_event, comm: str, symbol: str, dso: str,
                        dso_bid: str, dso_start: int, dso_end: int, map_pgoff: int) -> None:
        """Helper to process branch events."""
        cpu = sample.sample_cpu
        ip = sample.sample_ip
        addr = sample.sample_addr

        # The format for packet is:
        #
        #                 +------------+------------+------------+
        #  sample_prev:   |    addr    |     ip     |    cpu     |
        #                 +------------+------------+------------+
        #  sample_next:   |    addr    |     ip     |    cpu     |
        #                 +------------+------------+------------+
        #
        # Combine the two continuous packets to get the instruction range for
        # sample_prev::cpu:
        #
        #     [ sample_prev::addr .. sample_next::ip ]
        #
        # sample_prev::addr is stored into cpu_data and read back as
        # 'start_addr' when the new packet arrives, and sample_next::ip + 4 is
        # used for 'stop_addr' so objdump includes the branch instruction at
        # sample_next::ip.
        start_addr = self.cpu_data[str(cpu) + 'addr']
        stop_addr  = ip + 4

        # Record for previous sample packet
        self.cpu_data[str(cpu) + 'addr'] = addr

        # Filter out zero start_address. Optionally identify CS_ETM_TRACE_ON packet
        if start_addr == 0:
            if stop_addr == 4 and self.options.verbose:
                print(f"CPU{cpu}: CS_ETM_TRACE_ON packet is inserted")
            return

        if start_addr < dso_start or start_addr > dso_end:
            print(f"Start address {start_addr:#x} is out of range [ {dso_start:#x} .. "
                  f"{dso_end:#x} ] for dso {dso}")
            return

        if stop_addr < dso_start or stop_addr > dso_end:
            print(f"Stop address {stop_addr:#x} is out of range [ {dso_start:#x} .. "
                  f"{dso_end:#x} ] for dso {dso}")
            return

        if self.options.objdump is not None:
            # Kernel images (vmlinux / kcore) and non-PIE executables linked at
            # the default 0x400000 base already have their virtual load addresses
            # baked into the ELF symbol/section headers read by objdump, so
            # dso_vm_start and map_pgoff_local must be 0 to avoid subtracting
            # map_start from virtual sample addresses.
            if (dso in ("[kernel.kallsyms]", "vmlinux", "kcore")
                    or dso.endswith("/vmlinux") or dso.endswith("/kcore")):
                dso_vm_start = 0
                map_pgoff_local = 0
            elif dso_start == 0x400000:
                dso_vm_start = 0
                map_pgoff_local = 0
            else:
                dso_vm_start = dso_start
                map_pgoff_local = map_pgoff

            dso_fname = get_dso_file_path(dso, dso_bid, self.options.vmlinux)
            if path.exists(dso_fname):
                print_disam(dso_fname, dso_vm_start, start_addr + map_pgoff_local,
                            stop_addr + map_pgoff_local, self.options.objdump)
            else:
                print(f"Failed to find dso {dso} for address range [ "
                      f"{start_addr + map_pgoff_local:#x} .. {stop_addr + map_pgoff_local:#x} ]")

        print_srccode(comm, sample, symbol, dso)

    def run(self) -> None:
        """Run the trace disassembly session."""
        input_file = self.options.input or "perf.data"
        if input_file != "-" and not os.path.exists(input_file):
            print(f"Error: {input_file} not found.", file=sys.stderr)
            sys.exit(1)

        print('ARM CoreSight Trace Data Assembler Dump')
        try:
            session_kwargs = {
                "sample": self.process_event,
                "itrace": self.options.itrace,
            }
            if self.options.vmlinux:
                session_kwargs["vmlinux"] = self.options.vmlinux
            try:
                self.session = perf.session(perf.data(input_file), **session_kwargs)
            except TypeError:
                session_kwargs.pop("vmlinux", None)
                self.session = perf.session(perf.data(input_file), **session_kwargs)
        except (OSError, ValueError, KeyError, RuntimeError, TypeError, AttributeError) as e:
            print(f"Error opening session: {e}", file=sys.stderr)
            sys.exit(1)

        try:
            self.session.process_events()
        except (KeyboardInterrupt, SystemExit):
            pass
        finally:
            # Break the reference cycle between self.session and the bound
            # self.process_event callback so the C perf.session object is freed.
            self.session = None
        print('End')

if __name__ == "__main__":
    def int_arg(v: str) -> int:
        """Helper for integer command line arguments."""
        val = int(v)
        if val < 0:
            raise argparse.ArgumentTypeError("Argument must be a positive integer")
        return val

    arg_parser = argparse.ArgumentParser(description="ARM CoreSight Trace Dump With Disassembler")
    arg_parser.add_argument("-i", "--input", help="input perf.data file")
    arg_parser.add_argument("-k", "--vmlinux", default=os.environ.get("PERF_SYMBOL_VMLINUX"),
                            help="Set path to vmlinux file. Omit to autodetect")
    arg_parser.add_argument("-d", "--objdump", nargs="?", const=default_objdump(),
                            help="Show disassembly. Can also be used to change the objdump path")
    arg_parser.add_argument("-v", "--verbose", action="store_true", help="Enable debugging log")
    arg_parser.add_argument("--start-time", type=int_arg,
                            help="Monotonic clock time of sample to start from.")
    arg_parser.add_argument("--stop-time", type=int_arg,
                            help="Monotonic clock time of sample to stop at.")
    arg_parser.add_argument("--itrace", default=os.environ.get("PERF_ITRACE") or "b",
                            help="Instruction tracing options.")
    arg_parser.add_argument("--start-sample", type=int_arg,
                            help="Index of sample to start from.")
    arg_parser.add_argument("--stop-sample", type=int_arg,
                            help="Index of sample to stop at.")

    parsed_options = arg_parser.parse_args()
    if (parsed_options.start_time is not None and parsed_options.stop_time is not None and
            parsed_options.start_time >= parsed_options.stop_time):
        print("--start-time must less than --stop-time")
        sys.exit(2)
    if (parsed_options.start_sample is not None and parsed_options.stop_sample is not None and
            parsed_options.start_sample >= parsed_options.stop_sample):
        print("--start-sample must less than --stop-sample")
        sys.exit(2)

    td = TraceDisasm(parsed_options)
    td.run()