linux/tools/perf/python/stat-cpi.py

#!/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})")

    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:
            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:
                last_err = e
            except OSError as e:
                if e.errno == 13:
                    last_err = e
                else:
                    raise
        if last_err is not None:
            raise last_err
        raise RuntimeError("Failed to open events")

    def run_live(self) -> None:
        """Read counters live."""
        try:
            evlist = self._open_live_evlist()
        except OSError as e:
            print(f"Failed to open events: {e}", file=sys.stderr)
            sys.exit(1)

        def handle_signal(_signum: int, _frame: Any) -> None:
            raise KeyboardInterrupt

        signal.signal(signal.SIGINT, signal.default_int_handler)
        signal.signal(signal.SIGTERM, handle_signal)

        print("Live mode started. Press Ctrl+C to stop.")
        try:
            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.")
        finally:
            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()