aboutsummaryrefslogtreecommitdiff
path: root/tools/perf/python
diff options
context:
space:
mode:
Diffstat (limited to 'tools/perf/python')
-rwxr-xr-xtools/perf/python/SchedGui.py246
-rwxr-xr-xtools/perf/python/arm-cs-trace-disasm.py440
-rwxr-xr-xtools/perf/python/check-perf-trace.py223
-rwxr-xr-xtools/perf/python/compaction-times.py362
-rwxr-xr-xtools/perf/python/counting.py1
-rwxr-xr-xtools/perf/python/event_analyzing_sample.py317
-rwxr-xr-xtools/perf/python/export-to-postgresql.py1328
-rwxr-xr-xtools/perf/python/export-to-sqlite.py946
-rwxr-xr-xtools/perf/python/exported-sql-viewer.py5103
-rwxr-xr-xtools/perf/python/failed-syscalls-by-pid.py161
-rwxr-xr-xtools/perf/python/failed-syscalls.py94
-rwxr-xr-xtools/perf/python/flamegraph.py283
-rwxr-xr-xtools/perf/python/futex-contention.py107
-rwxr-xr-xtools/perf/python/gecko.py411
-rwxr-xr-xtools/perf/python/ilist.py86
-rwxr-xr-xtools/perf/python/intel-pt-events.py632
-rwxr-xr-xtools/perf/python/libxed.py123
-rwxr-xr-xtools/perf/python/mem-phys-addr.py136
-rwxr-xr-xtools/perf/python/net_dropmonitor.py173
-rwxr-xr-xtools/perf/python/netdev-times.py494
-rwxr-xr-xtools/perf/python/parallel-perf.py1250
-rw-r--r--tools/perf/python/perf.pyi122
-rwxr-xr-xtools/perf/python/perf_live.py10
-rwxr-xr-xtools/perf/python/powerpc-hcalls.py334
-rwxr-xr-xtools/perf/python/rw-by-file.py113
-rwxr-xr-xtools/perf/python/rw-by-pid.py205
-rwxr-xr-xtools/perf/python/rwtop.py256
-rwxr-xr-xtools/perf/python/sched-migration.py496
-rwxr-xr-xtools/perf/python/sctop.py253
-rwxr-xr-xtools/perf/python/stackcollapse.py145
-rwxr-xr-xtools/perf/python/stat-cpi.py231
-rwxr-xr-xtools/perf/python/syscall-counts-by-pid.py112
-rwxr-xr-xtools/perf/python/syscall-counts.py95
-rwxr-xr-xtools/perf/python/task-analyzer.py908
-rwxr-xr-xtools/perf/python/tracepoint.py5
-rwxr-xr-xtools/perf/python/treport.py807
-rwxr-xr-xtools/perf/python/ttimechart.py1806
-rwxr-xr-xtools/perf/python/twatch.py81
-rwxr-xr-xtools/perf/python/wakeup-latency.py103
39 files changed, 18886 insertions, 112 deletions
diff --git a/tools/perf/python/SchedGui.py b/tools/perf/python/SchedGui.py
new file mode 100755
index 000000000000..fa0d010fe528
--- /dev/null
+++ b/tools/perf/python/SchedGui.py
@@ -0,0 +1,246 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0
+# SchedGui.py - Python extension for perf script, basic GUI code for
+# traces drawing and overview.
+#
+# Copyright (C) 2010 by Frederic Weisbecker <fweisbec@gmail.com>
+#
+# Ported to modern directory structure.
+
+"""SchedGui.py - Python extension for perf script, basic GUI code for traces drawing and overview."""
+from __future__ import annotations
+
+import importlib
+from typing import Any
+
+class _DummyWx:
+ """Dummy wx module fallback when wxPython is not installed."""
+ Frame = object
+
+
+try:
+ wx: Any = importlib.import_module("wx")
+ WX_AVAILABLE = True
+except ImportError:
+ wx = _DummyWx
+ WX_AVAILABLE = False
+
+
+class RootFrame(wx.Frame):
+ """Main window frame for scheduling trace visualization."""
+ Y_OFFSET = 100
+ RECT_HEIGHT = 100
+ RECT_SPACE = 50
+ EVENT_MARKING_WIDTH = 5
+
+ def __init__(self, sched_tracer, title, parent=None, win_id=-1):
+ if not WX_AVAILABLE:
+ raise ImportError("You need to install the wxpython lib for this script")
+ wx.Frame.__init__(self, parent, win_id, title)
+
+ self.dc = None
+ (self.screen_width, self.screen_height) = wx.GetDisplaySize()
+ self.screen_width -= 10
+ self.screen_height -= 10
+ self.zoom = 0.5
+ self.scroll_scale = 20
+ self.sched_tracer = sched_tracer
+ self.sched_tracer.set_root_win(self)
+ (self.ts_start, self.ts_end) = sched_tracer.interval()
+ self.update_width_virtual()
+ self.nr_rects = sched_tracer.nr_rectangles() + 1
+ self.height_virtual = RootFrame.Y_OFFSET + \
+ (self.nr_rects * (RootFrame.RECT_HEIGHT + RootFrame.RECT_SPACE))
+
+ # whole window panel
+ self.panel = wx.Panel(self, size=(self.screen_width, self.screen_height))
+
+ # scrollable container
+ # Create SplitterWindow
+ self.splitter = wx.SplitterWindow(self.panel, style=wx.SP_3D)
+
+ # scrollable container (Top)
+ self.scroll = wx.ScrolledWindow(self.splitter)
+ self.scroll.SetScrollbars(self.scroll_scale, self.scroll_scale,
+ int(self.width_virtual // self.scroll_scale),
+ int(self.height_virtual // self.scroll_scale))
+ self.scroll.EnableScrolling(True, True)
+ self.scroll.SetFocus()
+
+ # scrollable drawing area
+ self.scroll_panel = wx.Panel(self.scroll,
+ size=(self.screen_width - 15, self.screen_height // 2))
+ self.scroll_panel.Bind(wx.EVT_PAINT, self.on_paint)
+ self.scroll_panel.Bind(wx.EVT_KEY_DOWN, self.on_key_press)
+ self.scroll_panel.Bind(wx.EVT_LEFT_DOWN, self.on_mouse_down)
+ self.scroll.Bind(wx.EVT_KEY_DOWN, self.on_key_press)
+ self.scroll.Bind(wx.EVT_LEFT_DOWN, self.on_mouse_down)
+
+ self.scroll_panel.SetSize(int(self.width_virtual), int(self.height_virtual))
+
+ # Create a separate panel for text (Bottom)
+ self.text_panel = wx.Panel(self.splitter)
+ self.text_sizer = wx.BoxSizer(wx.VERTICAL)
+ self.txt = wx.TextCtrl(self.text_panel, -1, "Click a bar to see details",
+ style=wx.TE_MULTILINE)
+ self.text_sizer.Add(self.txt, 1, wx.EXPAND | wx.ALL, 5)
+ self.text_panel.SetSizer(self.text_sizer)
+
+ # Split the window
+ self.splitter.SplitHorizontally(self.scroll, self.text_panel, (self.screen_height * 3) // 4)
+
+ # Main sizer to layout splitter
+ self.main_sizer = wx.BoxSizer(wx.VERTICAL)
+ self.main_sizer.Add(self.splitter, 1, wx.EXPAND)
+ self.panel.SetSizer(self.main_sizer)
+
+ self.scroll.Fit()
+ self.Fit()
+
+ self.Show(True)
+
+ def us_to_px(self, val):
+ """Convert microseconds to pixels."""
+ return val / (10 ** 3) * self.zoom
+
+ def px_to_us(self, val):
+ """Convert pixels to microseconds."""
+ return (val / self.zoom) * (10 ** 3)
+
+ def scroll_start(self):
+ """Get scroll start position in pixels."""
+ (x, y) = self.scroll.GetViewStart()
+ return (x * self.scroll_scale, y * self.scroll_scale)
+
+ def scroll_start_us(self):
+ """Get scroll start position in microseconds."""
+ (x, _) = self.scroll_start()
+ return self.px_to_us(x)
+
+ def paint_rectangle_zone(self, nr, color, top_color, start, end):
+ """Draw a rectangle zone for a CPU."""
+ offset_px = self.us_to_px(start - self.ts_start)
+ width_px = self.us_to_px(end - start)
+
+ offset_py = RootFrame.Y_OFFSET + (nr * (RootFrame.RECT_HEIGHT + RootFrame.RECT_SPACE))
+ width_py = RootFrame.RECT_HEIGHT
+
+ dc = self.dc
+
+ if top_color is not None:
+ (r, g, b) = top_color
+ top_color = wx.Colour(r, g, b)
+ brush = wx.Brush(top_color, wx.SOLID)
+ dc.SetBrush(brush)
+ dc.DrawRectangle(int(offset_px), int(offset_py),
+ int(width_px), RootFrame.EVENT_MARKING_WIDTH)
+ width_py -= RootFrame.EVENT_MARKING_WIDTH
+ offset_py += RootFrame.EVENT_MARKING_WIDTH
+
+ (r, g, b) = color
+ color = wx.Colour(r, g, b)
+ brush = wx.Brush(color, wx.SOLID)
+ dc.SetBrush(brush)
+ dc.DrawRectangle(int(offset_px), int(offset_py), int(width_px), int(width_py))
+
+ def update_rectangles(self, start, end):
+ """Update rectangles in the given time window."""
+ start += self.ts_start
+ end += self.ts_start
+ self.sched_tracer.fill_zone(start, end)
+
+ def on_paint(self, event):
+ """Handle paint event."""
+ window = event.GetEventObject()
+ dc = wx.PaintDC(window)
+
+ # Clear background to avoid ghosting
+ dc.SetBackground(wx.Brush(window.GetBackgroundColour()))
+ dc.Clear()
+
+ self.dc = dc
+ try:
+ width = min(self.width_virtual, self.screen_width)
+ (x, _) = self.scroll_start()
+ start = self.px_to_us(x)
+ end = self.px_to_us(x + width)
+ self.update_rectangles(start, end)
+
+ # Draw CPU labels at the left edge of the visible area
+ (x_scroll, _) = self.scroll_start()
+ for nr in range(self.nr_rects):
+ offset_py = RootFrame.Y_OFFSET + (nr * (RootFrame.RECT_HEIGHT + RootFrame.RECT_SPACE))
+ dc.DrawText(f"CPU {nr}", x_scroll + 10, offset_py + 10)
+ finally:
+ self.dc = None
+
+ def rect_from_ypixel(self, y):
+ y -= RootFrame.Y_OFFSET
+ rect = y // (RootFrame.RECT_HEIGHT + RootFrame.RECT_SPACE)
+ height = y % (RootFrame.RECT_HEIGHT + RootFrame.RECT_SPACE)
+
+ if rect < 0 or rect > self.nr_rects - 1 or height > RootFrame.RECT_HEIGHT:
+ return -1
+
+ return rect
+
+ def update_summary(self, txt):
+ self.txt.SetValue(txt)
+ self.text_panel.Layout()
+ self.splitter.Layout()
+ self.text_panel.Refresh()
+
+ def on_mouse_down(self, event):
+ pos = event.GetPosition()
+ x, y = pos.x, pos.y
+ rect = self.rect_from_ypixel(y)
+ if rect == -1:
+ return
+
+ t = self.px_to_us(x) + self.ts_start
+
+ self.sched_tracer.mouse_down(rect, t)
+
+ def update_width_virtual(self):
+ self.width_virtual = self.us_to_px(self.ts_end - self.ts_start)
+
+ def __zoom(self, x):
+ self.update_width_virtual()
+ (xpos, ypos) = self.scroll.GetViewStart()
+ xpos = int(self.us_to_px(x) // self.scroll_scale)
+ self.scroll_panel.SetSize((int(self.width_virtual), int(self.height_virtual)))
+ self.scroll.SetScrollbars(self.scroll_scale, self.scroll_scale,
+ int(self.width_virtual // self.scroll_scale),
+ int(self.height_virtual // self.scroll_scale),
+ xpos, ypos)
+ self.Refresh()
+
+ def zoom_in(self):
+ x = self.scroll_start_us()
+ self.zoom *= 2
+ self.__zoom(x)
+
+ def zoom_out(self):
+ x = self.scroll_start_us()
+ self.zoom /= 2
+ self.__zoom(x)
+
+ def on_key_press(self, event):
+ key = event.GetRawKeyCode()
+ if key == ord("+"):
+ self.zoom_in()
+ return
+ if key == ord("-"):
+ self.zoom_out()
+ return
+
+ key = event.GetKeyCode()
+ (x, y) = self.scroll.GetViewStart()
+ if key == wx.WXK_RIGHT:
+ self.scroll.Scroll(x + 1, y)
+ elif key == wx.WXK_LEFT:
+ self.scroll.Scroll(x - 1, y)
+ elif key == wx.WXK_DOWN:
+ self.scroll.Scroll(x, y + 1)
+ elif key == wx.WXK_UP:
+ self.scroll.Scroll(x, y - 1)
diff --git a/tools/perf/python/arm-cs-trace-disasm.py b/tools/perf/python/arm-cs-trace-disasm.py
new file mode 100755
index 000000000000..1812ccaf4b20
--- /dev/null
+++ b/tools/perf/python/arm-cs-trace-disasm.py
@@ -0,0 +1,440 @@
+#!/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,