xref: /linux/tools/perf/python/perf_live.py (revision 67f8bc848ee31831336bd478e57d2f993551902e)

# SPDX-License-Identifier: GPL-2.0
"""
Live event session helper using perf.evlist.

This module provides a LiveSession class that allows running a callback
for each event collected live from the system, similar to perf.session
but without requiring a perf.data file.
"""

import perf


class LiveSession:
    """Represents a live event collection session."""

    def __init__(self, event_string: str, sample_callback):
        self.event_string = event_string
        self.sample_callback = sample_callback
        # Create a cpu map for all online CPUs
        self.cpus = perf.cpu_map()
        # Parse events and set maps
        self.evlist = perf.parse_events(self.event_string, self.cpus)
        self.evlist.config()

    def run(self):
        """Run the live session."""
        try:
            self.evlist.open()
            self.evlist.mmap()
            self.evlist.enable()

            while True:
                # Poll for events with 100ms timeout
                try:
                    self.evlist.poll(100)
                except InterruptedError:
                    continue
                for cpu in self.cpus:
                    for _ in range(1000): # Limit to 1000 events per CPU per poll to prevent starvation
                        try:
                            event = self.evlist.read_on_cpu(cpu)
                        except TypeError as e:
                            if "Unknown CPU" in str(e):
                                # CPU might be unmapped or offline, wait for mmap event
                                break
                            if "Unexpected header type" in str(e):
                                # Ignore valid but unsupported event types
                                continue
                            raise

                        if event is None:
                            break

                        if event.type == perf.RECORD_SAMPLE:
                            self.sample_callback(event)
        except KeyboardInterrupt:
            pass
        finally:
            self.evlist.close()