| #!/usr/bin/env python3 | |
| # SPDX-License-Identifier: GPL-2.0 | |
| """Calculate CPI from perf stat data or live.""" | |
| from __future__ import annotations | |
| import argparse | |
| import os | |
| import signal | |
| import sys | |
| import time | |
| from typing import Any, Optional | |
| import perf | |
| class StatCpiAnalyzer: | |
| """Accumulates cycles and instructions and calculates CPI.""" | |
| def __init__(self, args: argparse.Namespace) -> None: | |
| self.args = args | |
| self.data: dict[str, float] = {} | |
| self.prev_data: dict[str, tuple[int, int, int]] = {} | |
| self.recorded_pairs: set[tuple[int, int]] = set() | |
| def get_key(self, event: str, cpu: int, thread: int) -> str: | |
| """Get key for data dictionary.""" | |
| return f"{event}-{cpu}-{thread}" | |
| def store_key(self, cpu: int, thread: int) -> None: | |
| """Store CPU and thread IDs.""" | |
| self.recorded_pairs.add((cpu, thread)) | |
| def store(self, event: str, cpu: int, thread: int, | |
| counts: tuple[int, int, int], is_delta: bool = False, | |
| raw_name: Optional[str] = None) -> None: | |
| """Store counter values, computing difference from previous | |
| absolute values if not already deltas.""" | |
| self.store_key(cpu, thread) | |
| key = self.get_key(event, cpu, thread) | |
| prev_key = self.get_key(raw_name or event, cpu, thread) | |
| val, ena, run = counts | |
| if is_delta: | |
| # counts are already deltas | |
| cur_val = val | |
| cur_ena = ena | |
| cur_run = run | |
| else: | |
| if prev_key in self.prev_data: | |
| prev_val, prev_ena, prev_run = self.prev_data[prev_key] | |
| cur_val = val - prev_val | |
| cur_ena = ena - prev_ena | |
| cur_run = run - prev_run | |
| else: | |
| cur_val = val | |
| cur_ena = ena | |
| cur_run = run | |
| self.prev_data[prev_key] = counts # Store absolute value for next time | |
| # Scale each raw event's delta by its own multiplexing ratio before | |
| # summing across PMUs (e.g. cpu_core and cpu_atom on hybrid systems) so | |
| # enabled time from an idle PMU does not inflate the multiplier of an | |
| # active PMU. | |
| scaled_val = cur_val * (cur_ena / float(cur_run)) if cur_run > 0 else float(cur_val) | |
| self.data[key] = self.data.get(key, 0.0) + scaled_val | |
| def get(self, event: str, cpu: int, thread: int) -> float: | |
| """Get scaled counter value.""" | |
| key = self.get_key(event, cpu, thread) | |
| return self.data.get(key, 0.0) | |
| @staticmethod | |
| def _classify_event(name: str) -> Optional[str]: | |
| """Classify an event name as 'cycles' or 'instructions'.""" | |
| ev = name[6:-1] if name.startswith("evsel(") and name.endswith(")") else name | |
| ev = ev.split(":", 1)[0] | |
| if "/" in ev: | |
| parts = [p for p in ev.split("/") if p] | |
| if len(parts) >= 2: | |
| ev = parts[1] | |
| if ev in ("cycles", "cpu-cycles"): | |
| return "cycles" | |
| if ev == "instructions": | |
| return "instructions" | |
| return None | |
| def process_stat_event(self, event: Any, name: Optional[str] = None) -> None: | |
| """Process PERF_RECORD_STAT and PERF_RECORD_STAT_ROUND events.""" | |
| if event.type == perf.RECORD_STAT: | |
| if name: | |
| event_name = self._classify_event(name) | |
| if not event_name: | |
| return | |
| self.store(event_name, event.cpu, event.thread, | |
| (event.val, event.ena, event.run), raw_name=name) | |
| elif event.type == perf.RECORD_STAT_ROUND: | |
| timestamp = getattr(event, "time", 0) | |
| self.print_interval(timestamp) | |
| self.data.clear() | |
| self.recorded_pairs.clear() | |
| def print_interval(self, timestamp: int) -> None: | |
| """Print CPI for the current interval.""" | |
| for cpu, thread in sorted(self.recorded_pairs): | |
| cyc = self.get("cycles", cpu, thread) | |
| ins = self.get("instructions", cpu, thread) | |
| cpi = 0.0 | |
| if ins != 0: | |
| cpi = cyc / float(ins) | |
| t_sec = timestamp / 1000000000.0 | |
| print(f"{t_sec:15f}: cpu {cpu}, thread {thread} -> cpi {cpi:f} ({cyc:.0f}/{ins:.0f})", | |
| flush=True) | |
| def read_counters(self, evlist: Any) -> None: | |
| """Read counters live.""" | |
| for evsel in evlist: | |
| name = str(evsel) | |
| event_name = self._classify_event(name) | |
| if not event_name: | |
| continue | |
| for cpu in evsel.cpus(): | |
| for thread in evsel.threads(): | |
| try: | |
| counts = evsel.read(cpu, thread) | |
| self.store(event_name, cpu, thread, | |
| (counts.val, counts.ena, counts.run), | |
| is_delta=True, raw_name=name) | |
| except OSError: | |
| pass | |
| def run_file(self) -> None: | |
| """Process events from file.""" | |
| session: Optional[perf.session] = perf.session( | |
| perf.data(self.args.input), stat=self.process_stat_event | |
| ) | |
| try: | |
| assert session is not None | |
| session.process_events() | |
| finally: | |
| session = None | |
| def _open_live_evlist(self) -> Any: | |
| """Open evlist for live mode, falling back to user-space or process scope on EACCES.""" | |
| threads = perf.thread_map(self.args.pid) if self.args.pid else None | |
| candidates = [ | |
| ("cycles,instructions", threads), | |
| ("cycles:u,instructions:u", threads), | |
| ] | |
| if threads is None: | |
| self_threads = perf.thread_map(os.getpid()) | |
| candidates.append(("cycles,instructions", self_threads)) | |
| candidates.append(("cycles:u,instructions:u", self_threads)) | |
| last_err: Optional[OSError] = None | |
| for events, tmap in candidates: | |
| evlist = None | |
| try: | |
| evlist = perf.parse_events(events, None, tmap) | |
| for evsel in evlist: | |
| evsel.read_format |= ( | |
| perf.FORMAT_TOTAL_TIME_ENABLED | perf.FORMAT_TOTAL_TIME_RUNNING | |
| ) | |
| evlist.open() | |
| evlist.enable() | |
| return evlist | |
| except PermissionError as e: | |
| if evlist is not None: | |
| evlist.close() | |
| last_err = e | |
| except OSError as e: | |
| if evlist is not None: | |
| evlist.close() | |
| if e.errno == 13: | |
| last_err = e | |
| else: | |
| raise | |
| except BaseException: | |
| if evlist is not None: | |
| evlist.close() | |
| raise | |
| if last_err is not None: | |
| raise last_err | |
| raise RuntimeError("Failed to open events") | |
| def run_live(self) -> None: | |
| """Read counters live.""" | |
| def handle_signal(_signum: int, _frame: Any) -> None: | |
| raise KeyboardInterrupt | |
| signal.signal(signal.SIGINT, signal.default_int_handler) | |
| signal.signal(signal.SIGTERM, handle_signal) | |
| evlist = None | |
| try: | |
| try: | |
| evlist = self._open_live_evlist() | |
| except OSError as e: | |
| print(f"Failed to open events: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| print("Live mode started. Press Ctrl+C to stop.", flush=True) | |
| while True: | |
| time.sleep(self.args.interval) | |
| timestamp = time.time_ns() | |
| self.read_counters(evlist) | |
| self.print_interval(timestamp) | |
| self.data.clear() | |
| self.recorded_pairs.clear() | |
| except KeyboardInterrupt: | |
| print("\nStopped.", flush=True) | |
| finally: | |
| if evlist is not None: | |
| evlist.close() | |
| def main() -> None: | |
| """Main function.""" | |
| ap = argparse.ArgumentParser(description="Calculate CPI from perf stat data or live") | |
| ap.add_argument("-i", "--input", help="Input file name (enables file mode)") | |
| ap.add_argument("-I", "--interval", type=float, default=1.0, | |
| help="Interval in seconds for live mode") | |
| ap.add_argument("-p", "--pid", type=int, | |
| help="Monitor specific process ID in live mode") | |
| args = ap.parse_args() | |
| analyzer = StatCpiAnalyzer(args) | |
| if args.input: | |
| analyzer.run_file() | |
| else: | |
| analyzer.run_live() | |
| if __name__ == "__main__": | |
| main() |