projects/linux·blob master · refs

59 lines · 2.0 KiB

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