Part 13 — The Integration: One Event Loop, Three Sources
You’ve built twelve instruments. Each one works on its own. Now the question becomes: how do you run them all at once without turning your monitoring system into the thing it’s supposed to observe?
The answer is a single Tokio event loop that polls all three data sources — PMC file descriptors, eBPF ring buffers, and procfs/sysfs files — on their own schedules. Tokio’s select! macro makes this natural: each source gets its own future, and select! runs them concurrently, waking whenever any source has data ready.
This part doesn’t introduce new instrumentation. It wires together what Parts 3–12 built. The reward is a single binary that prints structured metrics every second — PMCs, scheduler events, NUMA stats, thermal readings, block I/O, and histograms — all from one process.
The architecture, revisited
In Part 1, we showed the hybrid architecture as a diagram. Here’s the same architecture, but now with the concrete components we’ve built filled in:
┌─────────────────────────────────────────────────────────────────┐
│ userspace (monitor binary) │
│ │
│ tokio::select! { │
│ │
│ ┌──────────────────┐ ┌────────────────────┐ ┌────────────┐ │
│ │ PMC poller │ │ ring buffer reader │ │ file │ │
│ │ (Parts 3-5, 8) │ │ (Parts 6, 10-12) │ │ poller │ │
│ │ │ │ │ │ (7, 9) │ │
│ │ Every 1s: │ │ Whenever data │ │ Every 5s: │ │
│ │ read_counter() │ │ arrives in ring_buf │ │ read │ │
│ │ on each PMC fd │ │ .next() │ │ /sys, │ │
│ │ │ │ │ │ /proc │ │
│ └────────┬─────────┘ └──────────┬──────────┘ └─────┬──────┘ │
│ │ │ │ │
│ └───────────┬───────────┘ │ │
│ │ │ │
│ ┌────────▼────────┐ │ │
│ │ metrics sink │◄──────────────────────┘ │
│ │ (aggregate + │ │
│ │ emit JSON) │ │
│ └─────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│ │ │
perf_event_open() eBPF maps file I/O
▲ ▲ ▲
│ │ │
┌──────────┼────────────────┼────────────────┼───────────────────┐
│ │ Linux Kernel │ │ │
│ ┌───────┴────────┐ │ │ │
│ │ PMC counters │ │ │ │
│ │ (fd per event │ │ │ │
│ │ per CPU) │ │ │ │
│ └────────────────┘ │ │ │
│ │ │ │
│ ┌────────────────────────┴──────────────┐ │ │
│ │ eBPF programs │ │ │
│ │ scheduler (Part 6) │ │ │
│ │ block I/O (Part 10) │ │ │
│ │ vhost/virtio (Part 11) │ │ │
│ │ histograms (Part 12) │ │ │
│ └───────────────────────────────────────┘ │ │
│ │ │
│ ┌────────────────┐ ┌────────────────────┴──────────────┐ │
│ │ /proc/vmstat │ │/sys/class/thermal/ │ │
│ │ /proc/stat │ │/sys/devices/system/node/ │ │
│ └────────────────┘ └──────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘
The three sources have three different rhythms:
- PMC counters are polled on a timer. You read the file descriptor, compute a rate (delta / interval), and reset. Every 1 second is a common choice.
- eBPF ring buffers are push from the kernel’s side, but polled from userspace. The kernel writes events as they happen; userspace calls
ring_buf.next()to drain them. aya’sRingBuf::next()is non-blocking — it returnsNoneimmediately when the buffer is empty. To integrate withtokio::select!, we poll on a short timer (every 100 ms) and drain batches at each tick. - procfs/sysfs are polled on a slower timer. Thermal zones and NUMA stats don’t change every second — every 5 or 10 seconds is plenty.
The event loop needs to accommodate all three rhythms without any one source blocking the others. That’s what tokio::select! does.
The main event loop
// monitor/src/main.rs
use anyhow::Result;
use std::time::Duration;
use tokio::time;
mod metrics;
mod numa;
mod pmc;
mod thermal;
/// How often to poll PMC counters
const PMC_INTERVAL: Duration = Duration::from_secs(1);
/// How often to poll procfs/sysfs
const FILE_INTERVAL: Duration = Duration::from_secs(5);
/// Maximum events to drain from the ring buffer per tick
const RINGBUF_BATCH_SIZE: usize = 256;
/// How often to poll the eBPF ring buffer
const RINGBUF_INTERVAL: Duration = Duration::from_millis(100);
#[tokio::main]
async fn main() -> Result<()> {
tracing_subscriber::fmt::init();
// Load eBPF programs and attach tracepoints
let mut ebpf = load_and_attach_ebpf()?;
// Open PMC file descriptors for the events we care about
let pmc_fds = pmc::open_counters()?;
// Read initial counter values so the first delta is meaningful
let mut prev_pmc = pmc::read_all_counters(&pmc_fds)?;
// Timer ticks for each source
let mut pmc_tick = time::interval(PMC_INTERVAL);
let mut file_tick = time::interval(FILE_INTERVAL);
// Ring buffer for eBPF events
let map = ebpf
.map_mut("events")
.ok_or_else(|| anyhow::anyhow!("events map not found"))?;
let mut ring_buf = aya::maps::RingBuf::try_from(map)?;
// Timer for ring buffer polling
let mut ringbuf_tick = time::interval(RINGBUF_INTERVAL);
// Aggregated metrics for the current interval
let mut metrics = metrics::Metrics::new();
loop {
tokio::select! {
_ = pmc_tick.tick() => {
let curr = pmc::read_all_counters(&pmc_fds)?;
let deltas = pmc::compute_deltas(&prev_pmc, &curr);
prev_pmc = curr;
metrics.update_pmc(&deltas);
}
_ = ringbuf_tick.tick() => {
for event in drain_ringbuf(&mut ring_buf, RINGBUF_BATCH_SIZE) {
metrics.update_ebpf(&event);
}
}
_ = file_tick.tick() => {
let thermal = thermal::read_all_thermal_zones()?;
let numa = numa::read_numa_stats()?;
metrics.update_thermal(&thermal);
metrics.update_numa(&numa);
}
}
// Emit metrics whenever the PMC tick fires (every 1s)
if pmc_tick.tick().is_completed() {
let output = metrics.format_output();
println!("{}", output);
metrics.reset();
}
}
}PMC helpers
The pmc module needs a few helper types and functions that the event loop uses. Here is a self-contained pmc.rs that includes everything needed for the integration:
#![allow(unused)]
fn main() {
// monitor/src/pmc.rs
use anyhow::Result;
use std::io;
/// Delta between two PMC counter samples.
#[derive(Debug, Default)]
pub struct CounterDeltas {
pub instructions: u64,
pub cycles: u64,
pub cache_references: u64,
pub cache_misses: u64,
pub branch_misses: u64,
}
/// A PMC event specification.
pub struct EventSpec {
pub name: String,
pub perf_type: u32,
pub config: u64,
}
/// Open counters for all available events (used in the first example above).
pub fn open_counters() -> io::Result<Vec<(String, i32)>> {
let mut fds = Vec::new();
for spec in available_events() {
let fd = open_pmc(spec.perf_type, spec.config, 0, -1)?;
fds.push((spec.name, fd));
}
Ok(fds)
}
/// Read all counters at once.
pub fn read_all_counters(fds: &[(String, i32)]) -> io::Result<Vec<(String, u64)>> {
let mut vals = Vec::new();
for (name, fd) in fds {
let mut val: u64 = 0;
let n = unsafe { libc::read(*fd, &mut val as *mut _ as *mut libc::c_void, 8) };
if n < 0 {
return Err(io::Error::last_os_error());
}
vals.push((name.clone(), val));
}
Ok(vals)
}
/// Compute per-second deltas between two counter readings.
pub fn compute_deltas(
prev: &[(String, u64)],
curr: &[(String, u64)],
) -> CounterDeltas {
let get = |name: &str, samples: &[(String, u64)]| -> u64 {
samples.iter().find(|(n, _)| n == name).map(|(_, v)| *v).unwrap_or(0)
};
CounterDeltas {
instructions: get("instructions", curr).saturating_sub(get("instructions", prev)),
cycles: get("cycles", curr).saturating_sub(get("cycles", prev)),
cache_references: get("cache_references", curr).saturating_sub(get("cache_references", prev)),
cache_misses: get("cache_misses", curr).saturating_sub(get("cache_misses", prev)),
branch_misses: get("branch_misses", curr).saturating_sub(get("branch_misses", prev)),
}
}
fn open_pmc(type_: u32, config: u64, pid: i32, cpu: i32) -> io::Result<i32> {
#[repr(C)]
struct PerfEventAttr {
type_: u32,
size: u32,
config: u64,
sample_period: u64,
sample_type: u64,
read_format: u64,
flags: u64,
}
let attr = PerfEventAttr {
type_,
size: std::mem::size_of::<PerfEventAttr>() as u32,
config,
sample_period: 0,
sample_type: 0,
read_format: 0,
flags: 0b101, // disabled=1, pinned=1
};
let fd = unsafe {
libc::syscall(libc::SYS_perf_event_open, &attr as *const _, pid, cpu, -1, 0)
};
if fd < 0 {
return Err(io::Error::last_os_error());
}
Ok(fd as i32)
}
fn available_events() -> Vec<EventSpec> {
vec![
EventSpec { name: "instructions".into(), perf_type: 0, config: 1 },
EventSpec { name: "cycles".into(), perf_type: 0, config: 0 },
EventSpec { name: "cache_references".into(), perf_type: 0, config: 3 },
EventSpec { name: "cache_misses".into(), perf_type: 0, config: 4 },
EventSpec { name: "branch_misses".into(), perf_type: 0, config: 5 },
]
}
}⚠️ The code below has two bugs. We’ll walk through them and fix both in the next section. Don’t copy this version — it’s here to show the problem.
There are two problems with this.
First, the if pmc_tick.tick().is_completed() block outside select! calls tick() a second time. That call waits for the next tick — it doesn’t check whether the previous one fired. The result is a full extra interval delay before each emission. The PMC branch inside select! already consumed the tick; calling tick() again starts a fresh wait.
Second, the ring buffer branch uses drain_one, which returns a single event. If the scheduler is producing thousands of events per second, we’d only process one per select! iteration while the PMC and file ticks wait. We need to batch.
Let’s fix both.
Batched ring buffer draining
The fix: drain the ring buffer up to a batch size on each wake, not one at a time:
#[tokio::main]
async fn main() -> Result<()> {
tracing_subscriber::fmt::init();
let mut ebpf = load_and_attach_ebpf()?;
let pmc_fds = pmc::open_counters()?;
let mut prev_pmc = pmc::read_all_counters(&pmc_fds)?;
let mut pmc_tick = time::interval(PMC_INTERVAL);
let mut file_tick = time::interval(FILE_INTERVAL);
let mut ringbuf_tick = time::interval(RINGBUF_INTERVAL);
let map = ebpf
.map_mut("events")
.ok_or_else(|| anyhow::anyhow!("events map not found"))?;
let mut ring_buf = aya::maps::RingBuf::try_from(map)?;
let mut metrics = metrics::Metrics::new();
loop {
tokio::select! {
_ = pmc_tick.tick() => {
let curr = pmc::read_all_counters(&pmc_fds)?;
let deltas = pmc::compute_deltas(&prev_pmc, &curr);
prev_pmc = curr;
metrics.update_pmc(&deltas);
// Emit and reset on every PMC tick
let output = metrics.format_output();
println!("{}", output);
metrics.reset();
}
_ = ringbuf_tick.tick() => {
for event in drain_ringbuf(&mut ring_buf, RINGBUF_BATCH_SIZE) {
metrics.update_ebpf(&event);
}
}
_ = file_tick.tick() => {
let thermal = thermal::read_all_thermal_zones()?;
let numa = numa::read_numa_stats()?;
metrics.update_thermal(&thermal);
metrics.update_numa(&numa);
}
}
}
}
/// Drain up to `limit` events from the ring buffer.
/// Returns immediately if the buffer is empty — doesn't block.
/// This is synchronous because aya's `RingBuf::next()` is non-blocking;
/// it returns `None` when the buffer is empty rather than waiting.
fn drain_ringbuf<T: std::borrow::Borrow<aya::maps::MapData>>(
ring_buf: &mut aya::maps::RingBuf<T>,
limit: usize,
) -> Vec<SchedulerEvent> {
let mut events = Vec::with_capacity(limit);
for _ in 0..limit {
match ring_buf.next() {
Some(item) => events.push(parse_ringbuf_item(&*item)),
None => break,
}
}
events
}drain_ringbuf is synchronous — it calls ring_buf.next() in a tight loop up to the batch limit. This is fine because ring_buf.next() is non-blocking: it returns None immediately when the buffer is empty. We call it on a timer tick (every 100 ms) rather than trying to make it a future that blocks on data arrival. aya’s RingBuf doesn’t provide an async next(), so the timer-based approach is how we integrate it with tokio::select! without busy-looping.
Why batch? Without batching, a burst of 10,000 scheduler events would require 10,000 iterations through select!, each one checking the PMC timer and file timer before getting back to the ring buffer. With batching, we drain up to 256 events in one timer tick, then yield back to select! so the other sources get their turn. The batch size is a tuning knob: too small and you waste cycles on select! overhead; too large and you starve the PMC and file polls. 256 is a reasonable starting point — adjust based on your event rate.
Why a timer, not a blocking wait? aya’s RingBuf::next() is non-blocking — it returns None when the buffer is empty, rather than sleeping until data arrives. We can’t use it directly in select! as an always-ready future (that would busy-loop). The timer approach is honest: we poll the buffer every 100 ms, drain what’s there, and move on. If aya adds an async ring buffer API in the future, you could replace the timer with a true async poll — the rest of the loop wouldn’t change.
Unified event types
The eBPF programs from Parts 2, 6, and 10 all write to the same EVENTS ring buffer. To distinguish them in userspace, we use a unified event struct with an event_type tag at the front. Your eBPF programs should set this tag before writing to the ring buffer.
#![allow(unused)]
fn main() {
// monitor/src/events.rs
/// Event discriminant written by eBPF programs.
/// Use an explicit `u32` tag instead of a C enum so that the
/// userspace parser knows the exact size and alignment.
/// C enums are implementation-defined in size (often 4 bytes on Linux,
/// but the compiler chooses). A `u32` tag removes the ambiguity.
#[derive(Clone, Copy, Debug)]
pub enum EventType {
ContextSwitch = 0,
Wakeup = 1,
BioQueue = 2,
}
/// Unified event read from the ring buffer.
///
/// The layout must match what the eBPF program writes.
/// The eBPF side writes a struct with a `u32` tag followed by a `u32` cpu_id,
/// totaling 8 bytes with natural alignment. If you change this struct,
/// you must update the eBPF side to match.
#[derive(Clone, Copy, Debug)]
#[repr(C)]
pub struct RawEvent {
pub tag: u32,
pub cpu_id: u32,
}
/// Parsed event with a typed discriminant.
pub struct SchedulerEvent {
pub event_type: EventType,
pub cpu_id: u32,
}
/// Parse a raw `RingBufItem` into a `SchedulerEvent`.
///
/// The eBPF program writes an 8-byte struct: a `u32` tag
/// followed by a `u32` cpu_id. We read both fields at their
/// natural offsets (0 and 4) rather than assuming a packed
/// 5-byte layout. If the eBPF side uses a C enum for the tag,
/// the enum is 4 bytes on Linux (not 1), so reading the tag
/// as a single byte at offset 0 would miss the upper 3 bytes
/// and read `cpu_id` from the wrong offset.
pub fn parse_ringbuf_item(item: &[u8]) -> SchedulerEvent {
// Read the 8-byte struct at its natural layout
let raw = if item.len() >= std::mem::size_of::<RawEvent>() {
let tag = u32::from_le_bytes([
item[0], item[1], item[2], item[3],
]);
let cpu_id = u32::from_le_bytes([
item[4], item[5], item[6], item[7],
]);
RawEvent { tag, cpu_id }
} else {
RawEvent { tag: 0, cpu_id: 0 }
};
let event_type = match raw.tag {
0 => EventType::ContextSwitch,
1 => EventType::Wakeup,
2 => EventType::BioQueue,
_ => EventType::ContextSwitch,
};
SchedulerEvent { event_type, cpu_id: raw.cpu_id }
}
}Loading and attaching eBPF programs
The load_and_attach_ebpf function combines what Part 2 introduced and Part 6 elaborated:
#![allow(unused)]
fn main() {
use aya::programs::TracePoint;
use aya::Ebpf;
fn load_and_attach_ebpf() -> Result<Ebpf> {
// Load the eBPF object embedded at compile time
let mut ebpf = Ebpf::load(aya::include_bytes_aligned!(
concat!(env!("OUT_DIR"), "/perf-monitor")
))?;
// Attach scheduler tracepoints (Part 6)
attach_tracepoint(&mut ebpf, "sched_switch")?;
attach_tracepoint(&mut ebpf, "sched_waking")?;
// Attach block I/O tracepoint (Part 10)
attach_tracepoint(&mut ebpf, "block_bio_queue")?;
Ok(ebpf)
}
fn attach_tracepoint(ebpf: &mut Ebpf, name: &str) -> Result<()> {
let program: &mut TracePoint = ebpf
.program_mut(name)
.ok_or_else(|| anyhow::anyhow!("program '{}' not found", name))?
.try_into()?;
program.load()?;
// Map program names to their tracepoint (category, name)
let (category, tp_name) = match name {
"sched_switch" => ("sched", "sched_switch"),
"sched_waking" => ("sched", "sched_waking"),
"block_bio_queue" => ("block", "block_bio_queue"),
_ => return Err(anyhow::anyhow!("unknown program: {}", name)),
};
program.attach(category, tp_name)?;
Ok(())
}
}The mapping from program name to tracepoint category and event name is explicit. You could derive it — the program name often matches the tracepoint name — but being explicit avoids surprises when the two diverge (as they do for some kprobes).
NUMA helper
The numa module provides per-node memory stats. Your earlier numa.rs from Part 7 may have a different NumaStats for vmstat. For integration we need the sysfs meminfo version:
#![allow(unused)]
fn main() {
// monitor/src/numa.rs
use std::fs;
use std::io;
use serde::Serialize;
#[derive(Debug, Default, Serialize, Clone)]
pub struct NumaMemStats {
pub nodes: Vec<NodeMemInfo>,
}
#[derive(Debug, Default, Serialize, Clone)]
pub struct NodeMemInfo {
pub node: u32,
pub total_mb: u64,
pub free_mb: u64,
}
/// Read per-node memory info from sysfs and return total / free in MiB.
pub fn read_numa_stats() -> io::Result<NumaMemStats> {
let mut nodes = Vec::new();
let node_dir = std::path::Path::new("/sys/devices/system/node");
for entry in fs::read_dir(node_dir)?.flatten() {
let name_str = entry.file_name().to_string_lossy();
if !name_str.starts_with("node") {
continue;
}
let node_id: u32 = name_str.trim_start_matches("node").parse().unwrap_or(0);
let meminfo_path = entry.path().join("meminfo");
if !meminfo_path.exists() {
continue;
}
let content = fs::read_to_string(&meminfo_path)?;
let mut total_kb: u64 = 0;
let mut free_kb: u64 = 0;
for line in content.lines() {
let mut parts = line.split_whitespace();
let field = parts.next().unwrap_or("").trim_end_matches(':');
let value: u64 = parts.next().unwrap_or("0").parse().unwrap_or(0);
match field {
"MemTotal" => total_kb = value,
"MemFree" => free_kb = value,
_ => {}
}
}
nodes.push(NodeMemInfo {
node: node_id,
total_mb: total_kb / 1024,
free_mb: free_kb / 1024,
});
}
Ok(NumaMemStats { nodes })
}
}The metrics aggregator
The Metrics struct collects data from all three sources and formats it for output. Each source updates a different field:
#![allow(unused)]
fn main() {
// monitor/src/metrics.rs
use crate::events::{EventType, SchedulerEvent};
use crate::numa::NumaMemStats;
use crate::thermal::ThermalZone;
use serde::Serialize;
#[derive(Debug, Default)]
pub struct Metrics {
pub pmc: PmcMetrics,
pub scheduler: SchedulerMetrics,
pub block_io: BlockIoMetrics,
pub numa: Option<NumaMetrics>,
pub thermal: Option<ThermalMetrics>,
pub histogram: Option<HistogramMetrics>,
}
#[derive(Debug, Default, Serialize)]
pub struct PmcMetrics {
pub instructions: u64,
pub cycles: u64,
pub cache_references: u64,
pub cache_misses: u64,
pub branch_misses: u64,
}
#[derive(Debug, Default, Serialize)]
pub struct SchedulerMetrics {
pub context_switches: u64,
pub wakeups: u64,
pub per_cpu_switches: std::collections::HashMap<u32, u64>,
}
#[derive(Debug, Default, Serialize)]
pub struct BlockIoMetrics {
pub bio_queue_events: u64,
pub total_bytes: u64,
}
#[derive(Debug, Default, Serialize)]
pub struct HistogramMetrics {
pub buckets: Vec<u64>,
pub overflow: u64,
}
#[derive(Debug, Default, Serialize)]
pub struct NumaMetrics {
pub nodes: Vec<NodeMemInfo>,
}
#[derive(Debug, Default, Serialize, Clone)]
pub struct NodeMemInfo {
pub node: u32,
pub total_mb: u64,
pub free_mb: u64,
}
#[derive(Debug, Default, Serialize)]
pub struct ThermalMetrics {
pub zones: Vec<ZoneReading>,
}
#[derive(Debug, Default, Serialize)]
pub struct ZoneReading {
pub zone: String,
pub temp_c: f64,
pub trip_c: f64,
}
impl Metrics {
pub fn new() -> Self {
Self::default()
}
pub fn update_pmc(&mut self, deltas: &pmc::CounterDeltas) {
self.pmc.instructions = deltas.instructions;
self.pmc.cycles = deltas.cycles;
self.pmc.cache_references = deltas.cache_references;
self.pmc.cache_misses = deltas.cache_misses;
self.pmc.branch_misses = deltas.branch_misses;
}
pub fn update_ebpf(&mut self, event: &SchedulerEvent) {
match event.event_type {
EventType::ContextSwitch => {
self.scheduler.context_switches += 1;
*self.scheduler.per_cpu_switches
.entry(event.cpu_id)
.or_insert(0) += 1;
}
EventType::Wakeup => {
self.scheduler.wakeups += 1;
}
EventType::BioQueue => {
self.block_io.bio_queue_events += 1;
}
}
}
pub fn update_thermal(&mut self, zones: &[thermal::ThermalZone]) {
self.thermal = Some(ThermalMetrics {
zones: zones.iter().map(|z| {
// Convert from Part 9's millidegrees to Celsius.
// Find the lowest critical trip point for the trip_c field.
let trip_c = z.trip_points.iter()
.filter(|tp| tp.trip_type == "critical")
.map(|tp| tp.temp_millicelsius as f64 / 1000.0)
.fold(f64::MAX, f64::min);
ZoneReading {
zone: z.name.clone(),
temp_c: z.temp_millicelsius as f64 / 1000.0,
trip_c,
}
}).collect(),
});
}
pub fn update_numa(&mut self, numa: &numa::NumaMemStats) {
self.numa = Some(NumaMetrics {
nodes: numa.nodes.clone(),
});
}
pub fn update_histogram(&mut self, hist: &HistogramMetrics) {
self.histogram = Some(HistogramMetrics {
buckets: hist.buckets.clone(),
overflow: hist.overflow,
});
}
pub fn reset(&mut self) {
self.pmc = PmcMetrics::default();
self.scheduler = SchedulerMetrics::default();
self.block_io = BlockIoMetrics::default();
self.histogram = None;
// Don't reset numa and thermal — they're point-in-time snapshots,
// not interval counters. They keep their last-read values until
// the next file poll overwrites them.
}
pub fn format_output(&self) -> String {
let ipc = if self.pmc.cycles > 0 {
self.pmc.instructions as f64 / self.pmc.cycles as f64
} else {
0.0
};
let cache_miss_rate = if self.pmc.cache_references > 0 {
self.pmc.cache_misses as f64 / self.pmc.cache_references as f64 * 100.0
} else {
0.0
};
let branch_miss_rate = if self.pmc.instructions > 0 {
self.pmc.branch_misses as f64 / self.pmc.instructions as f64 * 100.0
} else {
0.0
};
let mut lines = Vec::new();
lines.push(format!(
"IPC={:.2} cache_miss={:.1}% branch_miss={:.3}% switches={} wakeups={}",
ipc, cache_miss_rate, branch_miss_rate,
self.scheduler.context_switches,
self.scheduler.wakeups,
));
if self.block_io.bio_queue_events > 0 {
lines.push(format!(
" bio_queue={} bio_bytes={}",
self.block_io.bio_queue_events,
self.block_io.total_bytes,
));
}
if let Some(ref hist) = self.histogram {
lines.push(format!(
" histogram: overflow={} buckets={}",
hist.overflow,
hist.buckets.iter().sum::<u64>(),
));
}
if let Some(ref numa) = self.numa {
for node in &numa.nodes {
lines.push(format!(
" node{}: {}/{} MB free",
node.node, node.free_mb, node.total_mb,
));
}
}
if let Some(ref thermal) = self.thermal {
for zone in &thermal.zones {
lines.push(format!(
" {}: {:.0}°C (trip: {:.0}°C)",
zone.zone, zone.temp_c, zone.trip_c,
));
}
}
lines.join("\n")
}
}
}Why reset() doesn’t clear thermal and NUMA. PMC counters and scheduler events are interval metrics — they accumulate over the reporting period and need to be zeroed. Thermal readings and NUMA memory stats are point-in-time snapshots. The last reading is still valid until the next poll replaces it. If you zeroed them on reset, the output would flicker between “last known value” and “no data” on every tick, which is confusing.
The format_output method uses conditional formatting. If there were no block I/O events in this interval, the bio_queue line is suppressed. If thermal zones haven’t been read yet (first 5 seconds), no thermal lines appear. This keeps the output clean without needing a separate “disable this source” flag.
Structured output: JSON mode
Human-readable output is good for development. For production monitoring, you want structured output that a downstream consumer — Prometheus, a log aggregator, a dashboard — can parse. The serde::Serialize derive on the metric types makes this straightforward:
#![allow(unused)]
fn main() {
impl Metrics {
/// Format metrics as a single JSON object.
/// Includes a timestamp so the consumer can align
/// readings from different sources.
pub fn format_json(&self) -> Result<String> {
#[derive(Serialize)]
struct Output<'a> {
timestamp: String,
pmc: &'a PmcMetrics,
scheduler: &'a SchedulerMetrics,
block_io: &'a BlockIoMetrics,
numa: &'a Option<NumaMetrics>,
thermal: &'a Option<ThermalMetrics>,
}
let output = Output {
timestamp: chrono::Utc::now().to_rfc3339(),
pmc: &self.pmc,
scheduler: &self.scheduler,
block_io: &self.block_io,
numa: &self.numa,
thermal: &self.thermal,
};
Ok(serde_json::to_string(&output)?)
}
}
}Add a CLI flag to choose the format:
// monitor/src/main.rs
use clap::Parser;
#[derive(Clone, clap::ValueEnum)]
enum OutputFormat {
Text,
Json,
}
#[derive(Parser)]
#[command(name = "perf-monitor")]
struct Args {
/// Output format: text (human-readable) or json (structured)
#[arg(long, default_value = "text")]
format: OutputFormat,
}
#[tokio::main]
async fn main() -> Result<()> {
let args = Args::parse();
// ... (same setup as before) ...
loop {
tokio::select! {
_ = pmc_tick.tick() => {
let curr = pmc::read_all_counters(&pmc_fds)?;
let deltas = pmc::compute_deltas(&prev_pmc, &curr);
prev_pmc = curr;
metrics.update_pmc(&deltas);
// Emit and reset on every PMC tick (inside the branch,
// not after select! — see the earlier "double-tick" fix)
let output = match args.format {
OutputFormat::Text => metrics.format_output(),
OutputFormat::Json => metrics.format_json()?,
};
println!("{}", output);
metrics.reset();
}
_ = ringbuf_tick.tick() => {
// ring_buf was created once before the loop —
// just drain it on each tick
for event in drain_ringbuf(&mut ring_buf, RINGBUF_BATCH_SIZE) {
metrics.update_ebpf(&event);
}
}
_ = file_tick.tick() => {
if let Ok(zones) = thermal::read_all_thermal_zones() {
metrics.update_thermal(&zones);
}
if let Ok(numa) = numa::read_numa_stats() {
metrics.update_numa(&numa);
}
}
}
}
}This adds clap and serde_json to Cargo.toml:
[dependencies]
aya = { version = "0.13", features = ["async_tokio"] }
tokio = { version = "1", features = ["full"] }
anyhow = "1"
clap = { version = "4", features = ["derive"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tracing = "0.1"
tracing-subscriber = "0.3"
chrono = "0.4"
Configuration: What to poll and how often
Hard-coded intervals are fine for a tutorial. For a real tool, you want the intervals to be configurable without recompiling. A simple config file works:
# perf-monitor.toml
[pmc]
interval_secs = 1
events = ["instructions", "cycles", "cache_references", "cache_misses", "branch_misses"]
[scheduler]
enabled = true
[block_io]
enabled = true
[file_poll]
interval_secs = 5
[output]
format = "text" # "text" or "json"
Reading it:
#![allow(unused)]
fn main() {
use serde::Deserialize;
#[derive(Debug, Deserialize)]
struct Config {
pmc: PmcConfig,
scheduler: SchedulerConfig,
block_io: BlockIoConfig,
file_poll: FilePollConfig,
output: OutputConfig,
}
#[derive(Debug, Deserialize)]
struct PmcConfig {
interval_secs: u64,
events: Vec<String>,
}
#[derive(Debug, Deserialize)]
struct SchedulerConfig {
enabled: bool,
}
#[derive(Debug, Deserialize)]
struct BlockIoConfig {
enabled: bool,
}
#[derive(Debug, Deserialize)]
struct FilePollConfig {
interval_secs: u64,
}
#[derive(Debug, Deserialize)]
struct OutputConfig {
format: String,
}
fn load_config(path: &std::path::Path) -> Result<Config> {
let text = std::fs::read_to_string(path)?;
let config: Config = toml::from_str(&text)?;
Ok(config)
}
}The config file controls which eBPF programs get attached (if scheduler.enabled = false, don’t attach sched_switch and sched_waking), which PMC events to open, and the polling intervals. This means you can run the same binary on a development machine (where you only care about scheduler events) and a production server (where you want everything) by switching config files.
Why TOML? It’s the lingua franca of Rust configuration. serde + toml is a single dependency, and the format is readable by humans and scripts alike. You could use JSON, but then you’d need to explain the schema somewhere — TOML’s section headers serve as built-in documentation.
Add toml to Cargo.toml:
toml = "0.8"
Conditional program attachment
With the config loaded, program attachment becomes conditional:
#![allow(unused)]
fn main() {
fn load_and_attach_ebpf(config: &Config) -> Result<Ebpf> {
let mut ebpf = Ebpf::load(aya::include_bytes_aligned!(
concat!(env!("OUT_DIR"), "/perf-monitor")
))?;
if config.scheduler.enabled {
attach_tracepoint(&mut ebpf, "sched_switch")?;
attach_tracepoint(&mut ebpf, "sched_waking")?;
}
if config.block_io.enabled {
attach_tracepoint(&mut ebpf, "block_bio_queue")?;
}
Ok(ebpf)
}
}When a source is disabled, its eBPF program is never loaded or attached. The kernel doesn’t fire it. The ring buffer receives no events for it. The update_ebpf method never sees those event types. The output never mentions them. Zero overhead for disabled sources — not just “not printed,” but “not running.”
PMC event selection from config
The PMC reader opens one file descriptor per event per CPU. Opening events you don’t need wastes file descriptors and scheduler time. The config file controls which events get opened:
#![allow(unused)]
fn main() {
// monitor/src/pmc.rs
use anyhow::{anyhow, Result};
/// A PMC event specification parsed from the config file.
pub struct EventSpec {
pub name: String,
pub perf_type: u32,
pub config: u64,
}
/// Open file descriptors for the events listed in the config.
pub fn open_counters_for(events: &[String]) -> Result<Vec<(String, i32)>> {
let all_events = available_events();
let mut fds = Vec::new();
for name in events {
let spec = all_events.iter()
.find(|e| e.name == *name)
.ok_or_else(|| anyhow!("unknown PMC event: {}", name))?;
let fd = open_pmc(spec.perf_type, spec.config, 0, -1)?;
fds.push((name.clone(), fd));
}
Ok(fds)
}
/// The built-in events this monitor supports.
/// Part 4's CPU detection code would extend this
/// with microarchitecture-specific raw events.
fn available_events() -> Vec<EventSpec> {
vec![
EventSpec { name: "instructions".into(), perf_type: 0, config: 1 },
EventSpec { name: "cycles".into(), perf_type: 0, config: 0 },
EventSpec { name: "cache_references".into(), perf_type: 0, config: 3 },
EventSpec { name: "cache_misses".into(), perf_type: 0, config: 4 },
EventSpec { name: "branch_misses".into(), perf_type: 0, config: 5 },
// Raw events (Part 4-5) would be added here based on
// the detected CPU microarchitecture:
// EventSpec { name: "llc_misses".into(), perf_type: 4, config: 0x2e_412e },
// EventSpec { name: "dtlb_walks".into(), perf_type: 4, config: 0x4f_01 },
]
}
}The available_events() function returns a static list of hardware events. Part 4’s CPU detection would extend this with raw PMC events for the detected microarchitecture (LLC misses, TLB walks, etc.). The key design point: the event list is a data structure, not a compile-time constant. The config file selects from it at runtime.
Putting it all together
The complete main.rs with configuration, conditional attachment, and structured output:
// monitor/src/main.rs
use anyhow::Result;
use clap::Parser;
use std::path::PathBuf;
use std::time::Duration;
use tokio::time;
mod metrics;
mod numa;
mod pmc;
mod thermal;
const RINGBUF_BATCH_SIZE: usize = 256;
const RINGBUF_INTERVAL: Duration = Duration::from_millis(100);
#[derive(Clone, clap::ValueEnum)]
enum OutputFormat {
Text,
Json,
}
#[derive(Parser)]
#[command(name = "perf-monitor")]
struct Args {
/// Path to config file
#[arg(short, long, default_value = "perf-monitor.toml")]
config: PathBuf,
/// Output format: text or json (overrides config file)
#[arg(long)]
format: Option<String>,
}
#[tokio::main]
async fn main() -> Result<()> {
tracing_subscriber::fmt::init();
let args = Args::parse();
let config = load_config(&args.config)?;
let output_format = match args.format.as_deref().unwrap_or(&config.output.format) {
"json" => OutputFormat::Json,
_ => OutputFormat::Text,
};
// Load and attach eBPF programs (conditional on config)
let mut ebpf = load_and_attach_ebpf(&config)?;
// Open PMC counters (only the events the config requests)
let pmc_fds = pmc::open_counters_for(&config.pmc.events)?;
let mut prev_pmc = pmc::read_all_counters(&pmc_fds)?;
// Timers
let pmc_interval = Duration::from_secs(config.pmc.interval_secs);
let file_interval = Duration::from_secs(config.file_poll.interval_secs);
let mut pmc_tick = time::interval(pmc_interval);
let mut file_tick = time::interval(file_interval);
let mut ringbuf_tick = time::interval(RINGBUF_INTERVAL);
// Ring buffer for eBPF events (created once, drained on each tick)
let mut ring_buf = {
let map = ebpf
.take_map("events")
.ok_or_else(|| anyhow::anyhow!("events map not found"))?;
aya::maps::RingBuf::try_from(map)?
};
let mut metrics = metrics::Metrics::new();
loop {
// Build the select! branches conditionally.
// If there's no ring buffer, we skip that branch.
tokio::select! {
_ = pmc_tick.tick() => {
let curr = pmc::read_all_counters(&pmc_fds)?;
let deltas = pmc::compute_deltas(&prev_pmc, &curr);
prev_pmc = curr;
metrics.update_pmc(&deltas);
// Read histograms from eBPF maps (Part 12)
if config.scheduler.enabled {
if let Ok(hist) = read_histogram(&mut ebpf) {
metrics.update_histogram(&hist);
}
}
// Emit and reset on every PMC tick
let output = match output_format {
OutputFormat::Text => metrics.format_output(),
OutputFormat::Json => metrics.format_json()?,
};
println!("{}", output);
metrics.reset();
}
_ = ringbuf_tick.tick() => {
if config.scheduler.enabled || config.block_io.enabled {
for event in drain_ringbuf(&mut ring_buf, RINGBUF_BATCH_SIZE) {
metrics.update_ebpf(&event);
}
}
}
_ = file_tick.tick() => {
if let Ok(zones) = thermal::read_all_thermal_zones() {
metrics.update_thermal(&zones);
}
if let Ok(numa) = numa::read_numa_stats() {
metrics.update_numa(&numa);
}
}
}
}
}
fn drain_ringbuf<T: std::borrow::Borrow<aya::maps::MapData>>(
ring_buf: &mut aya::maps::RingBuf<T>,
limit: usize,
) -> Vec<SchedulerEvent> {
let mut events = Vec::with_capacity(limit);
for _ in 0..limit {
match ring_buf.next() {
Some(item) => events.push(parse_ringbuf_item(&*item)),
None => break,
}
}
events
}The ringbuf_tick when all eBPF sources are disabled. The ring buffer is always created (it’s tied to the events map, which always exists in the eBPF object). When all eBPF sources are disabled, no programs write to the ring buffer, so drain_ringbuf returns an empty Vec on every tick. The timer still fires every 100 ms, but the cost is negligible — a single iteration through drain_ringbuf that immediately returns None from ring_buf.next() because the buffer is empty.
Example output
Running perf-monitor with default settings on a busy system:
IPC=1.84 cache_miss=3.2% branch_miss=0.412% switches=4821 wakeups=3847
bio_queue=142 bio_bytes=727040
node0: 12480/32168 MB free
node1: 9832/32168 MB free
x86_pkg_temp: 67°C (trip: 95°C)
acpitz: 64°C (trip: 95°C)
Same system, JSON mode (--format json):
{"timestamp":"2026-06-20T02:23:15Z","pmc":{"instructions":1842300,"cycles":1001200,"cache_references":923400,"cache_misses":29812,"branch_misses":7615},"scheduler":{"context_switches":4821,"wakeups":3847,"per_cpu_switches":{"0":1200,"1":1180,"2":1230,"3":1211}},"block_io":{"bio_queue_events":142,"total_bytes":727040},"numa":{"nodes":[{"node":0,"total_mb":32168,"free_mb":12480},{"node":1,"total_mb":32168,"free_mb":9832}]},"thermal":{"zones":[{"zone":"x86_pkg_temp","temp_c":67.0,"trip_c":95.0},{"zone":"acpitz","temp_c":64.0,"trip_c":95.0}]}}
Reading histograms from eBPF maps
Part 12 defined the eBPF side of queue depth histograms. In userspace, read the PerCpuArray and aggregate:
#![allow(unused)]
fn main() {
// monitor/src/main.rs (or a dedicated histogram module)
use aya::maps::{PerCpuArray, PerCpuValues};
use crate::metrics::HistogramMetrics;
const BOUNDARIES: [u32; 8] = [1, 2, 5, 9, 17, 33, 65, u32::MAX];
fn read_histogram(ebpf: &mut aya::Ebpf) -> anyhow::Result<HistogramMetrics> {
let hist: PerCpuArray<u64> = PerCpuArray::try_from(ebpf.map_mut("queue_hist")?)?;
let mut bucket_counts = vec![0u64; 8];
for idx in 0..8u32 {
let per_cpu_values: PerCpuValues<u64> = hist.get(&idx, 0)?;
bucket_counts[idx as usize] = per_cpu_values.iter().sum();
}
Ok(HistogramMetrics {
buckets: bucket_counts,
overflow: 0,
})
}
}What about histograms?
Part 12’s queue depth histograms use PerCpuArray, not RingBuf. They’re read on the PMC tick — same interval, same select! branch:
#![allow(unused)]
fn main() {
_ = pmc_tick.tick() => {
let curr = pmc::read_all_counters(&pmc_fds)?;
let deltas = pmc::compute_deltas(&prev_pmc, &curr);
prev_pmc = curr;
metrics.update_pmc(&deltas);
// Read histograms from eBPF maps (Part 12)
if config.scheduler.enabled {
if let Ok(hist) = read_histogram(&mut ebpf) {
metrics.update_histogram(&hist);
}
}
// Emit and reset
let output = match output_format {
OutputFormat::Text => metrics.format_output(),
OutputFormat::Json => metrics.format_json()?,
};
println!("{}", output);
metrics.reset();
}
}The histogram read is synchronous — you read all 8 per-CPU buckets and sum them. It takes microseconds. No need for a separate timer or a separate select! branch. It’s read alongside the PMC counters because both are “poll on a timer, read a value” — the histogram just happens to read from an eBPF map instead of a file descriptor.
Performance overhead
The whole point of this architecture is low overhead. Let’s quantify what “low” means:
PMC reading: One read() syscall per event per tick. 5 events × 1 tick/second = 5 syscalls/second. Each read() is a few microseconds. Total: ~25 μs/second, or 0.0025% of one CPU.
Ring buffer draining: Up to 256 events per batch, polled every 100 ms. ring_buf.next() is a memory read from a shared ring buffer — no syscall, no kernel transition. On a system with 5,000 scheduler events/second, that’s ~20 batches/second at the 256-event batch size. Each batch is a tight loop of memcpy + struct parsing. Total: well under 1 ms/second. The 100 ms poll interval adds 10 timer wakes/second, but each is a simple check — if the buffer is empty, the cost is near zero.
File polling: Reading /sys/class/thermal/thermal_zone*/temp and /sys/devices/system/node/node*/meminfo involves opening files, reading them, and closing them. Two thermal zones + two NUMA nodes = ~8 file reads every 5 seconds. Each read is ~100 μs. Total: ~800 μs every 5 seconds, or 0.016% of one CPU.
eBPF programs: These run in-kernel. Their overhead is proportional to the event rate. A sched_switch handler that reads three fields and writes a 24-byte event to a ring buffer takes ~1 μs per invocation. At 5,000 switches/second, that’s 5 ms/second on one CPU — 0.5%.
Total overhead: under 1% of one CPU. This is the benefit of the three-source architecture. PMC counters are read from file descriptors — no eBPF overhead. eBPF programs do in-kernel aggregation — only summaries cross the kernel/userspace boundary. File reads are infrequent. Nothing spins.
Summary
The integration chapter ties together the three data sources into a single event loop:
-
tokio::select!runs all three sources concurrently — PMC polling, ring buffer draining, and file reading — without any source blocking the others. -
Batched ring buffer draining (polled every 100 ms) prevents eBPF event bursts from starving the PMC and file polls. The batch size and poll interval are tuning knobs.
-
The
Metricsaggregator collects data from all sources, formats it as human-readable text or structured JSON, and resets interval counters on each tick. -
Configuration files control which sources are enabled, which PMC events to read, and the polling intervals — without recompiling.
-
Conditional program attachment means disabled sources have zero overhead — their eBPF programs are never loaded, their ring buffer events never arrive, their file reads never happen.
-
Histogram reads happen on the PMC tick alongside counter reads — they’re both “poll a map/file, read a value” operations, not push-based events.
-
The total overhead is under 1% of one CPU. PMC reading is a handful of syscalls. Ring buffer draining is memory copies. File polling is infrequent. eBPF programs run in-kernel with microsecond-scale handlers.
You now have a complete eBPF performance monitoring system: hardware counters, kernel tracepoints, and procfs/sysfs — all wired into a single binary with configurable sources, structured output, and minimal overhead. Parts 1–2 gave you the mental model and project structure. Parts 3–9 gave you each data source. Parts 10–12 gave you specialized instrumentation. This part gave you the assembly.
What you do with it from here depends on what you’re monitoring. The data sources are modular — add a new eBPF program, add a new update_* method, add a new config section. The event loop doesn’t care how many sources there are. select! just grows another branch.