1# SPDX-License-Identifier: GPL-2.0 2""" 3Live event session helper using perf.evlist. 4 5This module provides a LiveSession class that allows running a callback 6for each event collected live from the system, similar to perf.session 7but without requiring a perf.data file. 8""" 9 10import perf 11 12 13class LiveSession: 14 """Represents a live event collection session.""" 15 16 def __init__(self, event_string: str, sample_callback): 17 self.event_string = event_string 18 self.sample_callback = sample_callback 19 # Create a cpu map for all online CPUs 20 self.cpus = perf.cpu_map() 21 # Parse events and set maps 22 self.evlist = perf.parse_events(self.event_string, self.cpus) 23 self.evlist.config() 24 25 def run(self): 26 """Run the live session.""" 27 try: 28 self.evlist.open() 29 self.evlist.mmap() 30 self.evlist.enable() 31 32 while True: 33 # Poll for events with 100ms timeout 34 try: 35 self.evlist.poll(100) 36 except InterruptedError: 37 continue 38 for cpu in self.cpus: 39 for _ in range(1000): # Limit to 1000 events per CPU per poll to prevent starvation 40 try: 41 event = self.evlist.read_on_cpu(cpu) 42 except TypeError as e: 43 if "Unknown CPU" in str(e): 44 # CPU might be unmapped or offline, wait for mmap event 45 break 46 if "Unexpected header type" in str(e): 47 # Ignore valid but unsupported event types 48 continue 49 raise 50 51 if event is None: 52 break 53 54 if event.type == perf.RECORD_SAMPLE: 55 self.sample_callback(event) 56 except KeyboardInterrupt: 57 pass 58 finally: 59 self.evlist.close() 60