| # 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() |
