Something went wrong. Try again.
(I'm so sorry!) Linux kernel mirror (for testing) git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960# 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()