# SPDX-License-Identifier: GPL-2.0"""Live event session helper using perf.evlist. This module provides a LiveSession class that allows running a callbackfor each event collected live from the system, similar to perf.sessionbut 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()