| #!/usr/bin/env python3 | |
| # SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) | |
| """Interactive perf list.""" | |
| from abc import ABC, abstractmethod | |
| import argparse | |
| from dataclasses import dataclass | |
| import math | |
| from typing import Any, Dict, Optional, Tuple | |
| import perf | |
| from textual import on | |
| from textual.app import App, ComposeResult | |
| from textual.binding import Binding | |
| from textual.containers import Horizontal, HorizontalGroup, Vertical, VerticalScroll | |
| from textual.css.query import NoMatches | |
| from textual.command import SearchIcon | |
| from textual.screen import ModalScreen | |
| from textual.widgets import Button, Footer, Header, Input, Label, Sparkline, Static, Tree | |
| from textual.widgets.tree import TreeNode | |
| def get_info(info: Dict[str, str], key: str): | |
| return (info[key] + "\n") if key in info else "" | |
| class TreeValue(ABC): | |
| """Abstraction for the data of value in the tree.""" | |
| @abstractmethod | |
| def name(self) -> str: | |
| pass | |
| @abstractmethod | |
| def description(self) -> str: | |
| pass | |
| @abstractmethod | |
| def matches(self, query: str) -> bool: | |
| pass | |
| @abstractmethod | |
| def parse(self) -> perf.evlist: | |
| pass | |
| @abstractmethod | |
| def value(self, evlist: perf.evlist, evsel: perf.evsel, cpu: int, thread: int) -> float: | |
| pass | |
| @dataclass | |
| class Metric(TreeValue): | |
| """A metric in the tree.""" | |
| metric_name: str | |
| metric_pmu: str | |
| def name(self) -> str: | |
| return self.metric_name | |
| def description(self) -> str: | |
| """Find and format metric description.""" | |
| for metric in perf.metrics(): | |
| if metric["MetricName"] != self.metric_name: | |
| continue | |
| if self.metric_pmu and metric["PMU"] != self.metric_pmu: | |
| continue | |
| desc = get_info(metric, "BriefDescription") | |
| desc += get_info(metric, "PublicDescription") | |
| desc += get_info(metric, "MetricExpr") | |
| desc += get_info(metric, "MetricThreshold") | |
| return desc | |
| return "description" | |
| def matches(self, query: str) -> bool: | |
| return query in self.metric_name | |
| def parse(self) -> perf.evlist: | |
| return perf.parse_metrics(self.metric_name, self.metric_pmu) | |
| def value(self, evlist: perf.evlist, evsel: perf.evsel, cpu: int, thread: int) -> float: | |
| try: | |
| val = evlist.compute_metric(self.metric_name, cpu, thread) | |
| return 0 if math.isnan(val) else val | |
| except: | |
| # Be tolerant of failures to compute metrics on particular CPUs/threads. | |
| return 0 | |
| @dataclass | |
| class PmuEvent(TreeValue): | |
| """A PMU and event within the tree.""" | |
| pmu: str | |
| event: str | |
| def name(self) -> str: | |
| if self.event.startswith(self.pmu) or ':' in self.event: | |
| return self.event | |
| else: | |
| return f"{self.pmu}/{self.event}/" | |
| def description(self) -> str: | |
| """Find and format event description for {pmu}/{event}/.""" | |
| for p in perf.pmus(): | |
| if p.name() != self.pmu: | |
| continue | |
| for info in p.events(): | |
| if "name" not in info or info["name"] != self.event: | |
| continue | |
| desc = get_info(info, "topic") | |
| desc += get_info(info, "event_type_desc") | |
| desc += get_info(info, "desc") | |
| desc += get_info(info, "long_desc") | |
| desc += get_info(info, "encoding_desc") | |
| return desc | |
| return "description" | |
| def matches(self, query: str) -> bool: | |
| return query in self.pmu or query in self.event | |
| def parse(self) -> perf.evlist: | |
| return perf.parse_events(self.name()) | |
| def value(self, evlist: perf.evlist, evsel: perf.evsel, cpu: int, thread: int) -> float: | |
| return evsel.read(cpu, thread).val | |
| class ErrorScreen(ModalScreen[bool]): | |
| """Pop up dialog for errors.""" | |
| CSS = """ | |
| ErrorScreen { | |
| align: center middle; | |
| } | |
| """ | |
| def __init__(self, error: str): | |
| self.error = error | |
| super().__init__() | |
| def compose(self) -> ComposeResult: | |
| yield Button(f"Error: {self.error}", variant="primary", id="error") | |
| def on_button_pressed(self, event: Button.Pressed) -> None: | |
| self.dismiss(True) | |
| class SearchScreen(ModalScreen[str]): | |
| """Pop up dialog for search.""" | |
| CSS = """ | |
| SearchScreen Horizontal { | |
| align: center middle; | |
| margin-top: 1; | |
| } | |
| SearchScreen Input { | |
| width: 1fr; | |
| } | |
| """ | |
| def compose(self) -> ComposeResult: | |
| yield Horizontal(SearchIcon(), Input(placeholder="Event name")) | |
| def on_input_submitted(self, event: Input.Submitted) -> None: | |
| """Handle the user pressing Enter in the input field.""" | |
| self.dismiss(event.value) | |
| class Counter(HorizontalGroup): | |
| """Two labels for a CPU and its counter value.""" | |
| CSS = """ | |
| Label { | |
| gutter: 1; | |
| } | |
| """ | |
| def __init__(self, cpu: int) -> None: | |
| self.cpu = cpu | |
| super().__init__() | |
| def compose(self) -> ComposeResult: | |
| label = f"cpu{self.cpu}" if self.cpu >= 0 else "total" | |
| yield Label(label + " ") | |
| yield Label("0", id=f"counter_{label}") | |
| class CounterSparkline(HorizontalGroup): | |
| """A Sparkline for a performance counter.""" | |
| def __init__(self, cpu: int) -> None: | |
| self.cpu = cpu | |
| super().__init__() | |
| def compose(self) -> ComposeResult: | |
| label = f"cpu{self.cpu}" if self.cpu >= 0 else "total" | |
| yield Label(label) | |
| yield Sparkline([], summary_function=max, id=f"sparkline_{label}") | |
| class IListApp(App): | |
| TITLE = "Interactive Perf List" | |
| BINDINGS = [ | |
| Binding(key="s", action="search", description="Search", | |
| tooltip="Search events and PMUs"), | |
| Binding(key="n", action="next", description="Next", | |
| tooltip="Next search result or item"), | |
| Binding(key="p", action="prev", description="Previous", | |
| tooltip="Previous search result or item"), | |
| Binding(key="c", action="collapse", description="Collapse", | |
| tooltip="Collapse the current PMU"), | |
| Binding(key="^q", action="quit", description="Quit", | |
| tooltip="Quit the app"), | |
| ] | |
| CSS = """ | |
| /* Make the 'total' sparkline a different color. */ | |
| #sparkline_total > .sparkline--min-color { | |
| color: $accent; | |
| } | |
| #sparkline_total > .sparkline--max-color { | |
| color: $accent 30%; | |
| } | |
| /* | |
| * Make the active_search initially not displayed with the text in | |
| * the middle of the line. | |
| */ | |
| #active_search { | |
| display: none; | |
| width: 100%; | |
| text-align: center; | |
| } | |
| """ | |
| def __init__(self, interval: float) -> None: | |
| self.interval = interval | |
| self.evlist = None | |
| self.selected: Optional[TreeValue] = None | |
| self.search_results: list[TreeNode[TreeValue]] = [] | |
| self.cur_search_result: TreeNode[TreeValue] | None = None | |
| super().__init__() | |
| def expand_and_select(self, node: TreeNode[Any]) -> None: | |
| """Expand select a node in the tree.""" | |
| if node.parent: | |
| node.parent.expand() | |
| if node.parent.parent: | |
| node.parent.parent.expand() | |
| node.expand() | |
| node.tree.select_node(node) | |
| node.tree.scroll_to_node(node) | |
| def set_searched_tree_node(self, previous: bool) -> None: | |
| """Set the cur_search_result node to either the next or previous.""" | |
| l = len(self.search_results) | |
| if l < 1: | |
| tree: Tree[TreeValue] = self.query_one("#root", Tree) | |
| if previous: | |
| tree.action_cursor_up() | |
| else: | |
| tree.action_cursor_down() | |
| return | |
| if self.cur_search_result: | |
| idx = self.search_results.index(self.cur_search_result) | |
| if previous: | |
| idx = idx - 1 if idx > 0 else l - 1 | |
| else: | |
| idx = idx + 1 if idx < l - 1 else 0 | |
| else: | |
| idx = l - 1 if previous else 0 | |
| node = self.search_results[idx] | |
| if node == self.cur_search_result: | |
| return | |
| self.cur_search_result = node | |
| self.expand_and_select(node) | |
| def action_search(self) -> None: | |
| """Search was chosen.""" | |
| def set_initial_focus(event: str | None) -> None: | |
| """Sets the focus after the SearchScreen is dismissed.""" | |
| search_label = self.query_one("#active_search", Label) | |
| search_label.display = True if event else False | |
| if not event: | |
| return | |
| event = event.lower() | |
| search_label.update(f'Searching for events matching "{event}"') | |
| tree: Tree[str] = self.query_one("#root", Tree) | |
| def find_search_results(event: str, node: TreeNode[str], | |
| cursor_seen: bool = False, | |
| match_after_cursor: Optional[TreeNode[str]] = None | |
| ) -> Tuple[bool, Optional[TreeNode[str]]]: | |
| """Find nodes that match the search remembering the one after the cursor.""" | |
| if not cursor_seen and node == tree.cursor_node: | |
| cursor_seen = True | |
| if node.data and node.data.matches(event): | |
| if cursor_seen and not match_after_cursor: | |
| match_after_cursor = node | |
| self.search_results.append(node) | |
| if node.children: | |
| for child in node.children: | |
| (cursor_seen, match_after_cursor) = \ | |
| find_search_results(event, child, cursor_seen, match_after_cursor) | |
| return (cursor_seen, match_after_cursor) | |
| self.search_results.clear() | |
| (_, self.cur_search_result) = find_search_results(event, tree.root) | |
| if len(self.search_results) < 1: | |
| self.push_screen(ErrorScreen(f"Failed to find pmu/event or metric {event}")) | |
| search_label.display = False | |
| elif self.cur_search_result: | |
| self.expand_and_select(self.cur_search_result) | |
| else: | |
| self.set_searched_tree_node(previous=False) | |
| self.push_screen(SearchScreen(), set_initial_focus) | |
| def action_next(self) -> None: | |
| """Next was chosen.""" | |
| self.set_searched_tree_node(previous=False) | |
| def action_prev(self) -> None: | |
| """Previous was chosen.""" | |
| self.set_searched_tree_node(previous=True) | |
| def action_collapse(self) -> None: | |
| """Collapse the part of the tree currently on.""" | |
| tree: Tree[str] = self.query_one("#root", Tree) | |
| node = tree.cursor_node | |
| if node and node.parent: | |
| node.parent.collapse_all() | |
| node.tree.scroll_to_node(node.parent) | |
| def update_counts(self) -> None: | |
| """Called every interval to update counts.""" | |
| if not self.selected or not self.evlist: | |
| return | |
| def update_count(cpu: int, count: int): | |
| # Update the raw count display. | |
| counter: Label = self.query(f"#counter_cpu{cpu}" if cpu >= 0 else "#counter_total") | |
| if not counter: | |
| return | |
| counter = counter.first(Label) | |
| counter.update(str(count)) | |
| # Update the sparkline. | |
| line: Sparkline = self.query(f"#sparkline_cpu{cpu}" if cpu >= 0 else "#sparkline_total") | |
| if not line: | |
| return | |
| line = line.first(Sparkline) | |
| # If there are more events than the width, remove the front event. | |
| if len(line.data) > line.size.width: | |
| line.data.pop(0) | |
| line.data.append(count) | |
| line.mutate_reactive(Sparkline.data) | |
| # Update the total and each CPU counts, assume there's just 1 evsel. | |
| total = 0 | |
| self.evlist.disable() | |
| for evsel in self.evlist: | |
| for cpu in evsel.cpus(): | |
| aggr = 0 | |
| for thread in evsel.threads(): | |
| aggr += self.selected.value(self.evlist, evsel, cpu, thread) | |
| update_count(cpu, aggr) | |
| total += aggr | |
| update_count(-1, total) | |
| self.evlist.enable() | |
| def on_mount(self) -> None: | |
| """When App starts set up periodic event updating.""" | |
| self.update_counts() | |
| self.set_interval(self.interval, self.update_counts) | |
| def set_selected(self, value: TreeValue) -> None: | |
| """Updates the event/description and starts the counters.""" | |
| try: | |
| label_name = self.query_one("#event_name", Label) | |
| event_description = self.query_one("#event_description", Static) | |
| lines = self.query_one("#lines") | |
| except NoMatches: | |
| # A race with rendering, ignore the update as we can't | |
| # mount the assumed output widgets. | |
| return | |
| self.selected = value | |
| # Remove previous event information. | |
| if self.evlist: | |
| self.evlist.disable() | |
| self.evlist.close() | |
| old_lines = self.query(CounterSparkline) | |
| for line in old_lines: | |
| line.remove() | |
| old_counters = self.query(Counter) | |
| for counter in old_counters: | |
| counter.remove() | |
| # Update event/metric text and description. | |
| label_name.update(value.name()) | |
| event_description.update(value.description()) | |
| # Open the event. | |
| try: | |
| self.evlist = value.parse() | |
| if self.evlist: | |
| self.evlist.open() | |
| self.evlist.enable() | |
| except: | |
| self.evlist = None | |
| if not self.evlist: | |
| self.push_screen(ErrorScreen(f"Failed to open {value.name()}")) | |
| return | |
| # Add spark lines for all the CPUs. Note, must be done after | |
| # open so that the evlist CPUs have been computed by propagate | |
| # maps. | |
| line = CounterSparkline(cpu=-1) | |
| lines.mount(line) | |
| for cpu in self.evlist.all_cpus(): | |
| line = CounterSparkline(cpu) | |
| lines.mount(line) | |
| line = Counter(cpu=-1) | |
| lines.mount(line) | |
| for cpu in self.evlist.all_cpus(): | |
| line = Counter(cpu) | |
| lines.mount(line) | |
| def compose(self) -> ComposeResult: | |
| """Draws the app.""" | |
| def metric_event_tree() -> Tree: | |
| """Create tree of PMUs and metricgroups with events or metrics under.""" | |
| tree: Tree[TreeValue] = Tree("Root", id="root") | |
| pmus = tree.root.add("PMUs") | |
| for pmu in perf.pmus(): | |
| pmu_name = pmu.name().lower() | |
| pmu_node = pmus.add(pmu_name) | |
| try: | |
| for event in sorted(pmu.events(), key=lambda x: x["name"]): | |
| if "deprecated" in event: | |
| continue | |
| if "name" in event: | |
| e = event["name"].lower() | |
| if "alias" in event: | |
| pmu_node.add_leaf(f'{e} ({event["alias"]})', | |
| data=PmuEvent(pmu_name, e)) | |
| else: | |
| pmu_node.add_leaf(e, data=PmuEvent(pmu_name, e)) | |
| except: | |
| # Reading events may fail with EPERM, ignore. | |
| pass | |
| metrics = tree.root.add("Metrics") | |
| groups = set() | |
| for metric in perf.metrics(): | |
| groups.update(metric["MetricGroup"]) | |
| def add_metrics_to_tree(node: TreeNode[TreeValue], parent: str, pmu: str = None): | |
| for metric in sorted(perf.metrics(), key=lambda x: x["MetricName"]): | |
| metric_pmu = metric.get('PMU') | |
| if pmu and metric_pmu and metric_pmu != pmu: | |
| continue | |
| if parent in metric["MetricGroup"]: | |
| name = metric["MetricName"] | |
| display_name = name | |
| if metric_pmu: | |
| display_name += f" ({metric_pmu})" | |
| node.add_leaf(display_name, data=Metric(name, metric_pmu)) | |
| child_group_name = f'{name}_group' | |
| if child_group_name in groups: | |
| display_child_group_name = child_group_name | |
| if metric_pmu: | |
| display_child_group_name += f" ({metric_pmu})" | |
| add_metrics_to_tree(node.add(display_child_group_name), | |
| child_group_name, | |
| metric_pmu) | |
| for group in sorted(groups): | |
| if group.endswith('_group'): | |
| continue | |
| add_metrics_to_tree(metrics.add(group), group) | |
| tree.root.expand() | |
| return tree | |
| yield Header(id="header") | |
| yield Horizontal(Vertical(metric_event_tree(), id="events"), | |
| Vertical(Label("event name", id="event_name"), | |
| Static("description", markup=False, id="event_description"), | |
| )) | |
| yield Label(id="active_search") | |
| yield VerticalScroll(id="lines") | |
| yield Footer(id="footer") | |
| @on(Tree.NodeSelected) | |
| def on_tree_node_selected(self, event: Tree.NodeSelected[TreeValue]) -> None: | |
| """Called when a tree node is selected, selecting the event.""" | |
| if event.node.data: | |
| self.set_selected(event.node.data) | |
| if __name__ == "__main__": | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument('-I', '--interval', help="Counter update interval in seconds", default=0.1) | |
| args = ap.parse_args() | |
| app = IListApp(float(args.interval)) | |
| app.run() |