blob: 99e73d58a96ffdeb876a799748017be7be9ed5a0 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
# 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)
# CPython's PyErr_SetFromErrno() invokes PyErr_CheckSignals() on
# EINTR, converting SIGINT into KeyboardInterrupt, which inherits
# from BaseException rather than InterruptedError/OSError, so
# Ctrl-C exits the loop and triggers finally: self.evlist.close()
# while non-fatal signals retry poll().
except InterruptedError:
continue
for cpu in self.cpus:
# Limit to 1000 events per CPU per poll to prevent starvation
for _ in range(1000):
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)
finally:
self.evlist.close()
|
