scx_flow/rust/
snapshot.rs

1// SPDX-License-Identifier: GPL-2.0
2//! Snapshot reads for the flow scheduler.
3//!
4//! Copyright (c) 2026 Galih Tama <galpt@v.recipes>
5
6//! Builds the counters view plus the dashboard view from the BPF maps.
7//! Each poll reads the per CPU stats rows plus the per CPU pid view
8//! with no extra sysfs use, so the page stays cheap beside the slice.
9//! A dashboard poll costs 3N BPF map reads for N online CPUs: N for
10//! the stats sum plus N for the on CPU gauge plus N for the per CPU
11//! cards, with one alloc and no sysfs on poll.
12
13use std::mem::MaybeUninit;
14use std::os::fd::AsFd;
15use std::os::fd::AsRawFd;
16
17use crate::Scheduler;
18use crate::stats;
19
20impl<'a> Scheduler<'a> {
21    /// Count live pids for the on CPU gauge with no BPF counter.
22    /// Walks the cached online list with one map read per CPU, so the
23    /// gauge costs N reads with no atomic cost on the hot paths. The
24    /// BPF on CPU field stays zero for wire compat and never drives
25    /// this count.
26    pub(crate) fn count_on_cpu(&self) -> u64 {
27        let mut live = 0u64;
28        for &id in self.online_cpus.iter() {
29            let st = self.read_cpu(id as usize);
30            if st.running_pid != 0 {
31                live = live.saturating_add(1);
32            }
33        }
34        live
35    }
36
37    pub(crate) fn get_metrics(&self) -> stats::Metrics {
38        let mut total_runtime = 0u64;
39        let mut inserts = 0u64;
40        let mut requeues = 0u64;
41        let mut completions = 0u64;
42        let mut local_moves = 0u64;
43        let mut node_moves = 0u64;
44        let mut machine_moves = 0u64;
45        let mut kicks = 0u64;
46        let mut admits = 0u64;
47        let mut misses = 0u64;
48        let mut gate_rejects = 0u64;
49        let mut preempt_kicks = 0u64;
50        let mut preempt_skipped = 0u64;
51        let mut red_rejects = 0u64;
52        let mut red_reclaims = 0u64;
53        for &id in self.online_cpus.iter() {
54            let s = self.read_stats(id as usize);
55            total_runtime = total_runtime.saturating_add(s.total_runtime);
56            inserts = inserts.saturating_add(s.inserts);
57            requeues = requeues.saturating_add(s.requeues);
58            completions = completions.saturating_add(s.completions);
59            local_moves = local_moves.saturating_add(s.local_moves);
60            node_moves = node_moves.saturating_add(s.node_moves);
61            machine_moves = machine_moves.saturating_add(s.machine_moves);
62            kicks = kicks.saturating_add(s.kicks);
63            admits = admits.saturating_add(s.admits);
64            misses = misses.saturating_add(s.misses);
65            gate_rejects = gate_rejects.saturating_add(s.gate_rejects);
66            preempt_kicks = preempt_kicks.saturating_add(s.preempt_kicks);
67            preempt_skipped = preempt_skipped.saturating_add(s.preempt_skipped);
68            red_rejects = red_rejects.saturating_add(s.red_rejects);
69            red_reclaims = red_reclaims.saturating_add(s.red_reclaims);
70        }
71        stats::Metrics {
72            on_cpu: self.count_on_cpu(),
73            total_runtime,
74            uptime_ns: self.started_at.elapsed().as_nanos().min(u64::MAX as u128) as u64,
75            inserts,
76            requeues,
77            completions,
78            local_moves,
79            node_moves,
80            machine_moves,
81            kicks,
82            admits,
83            // Dead compat: rejects stays zero with no BPF writer for wire
84            // compat only; readers must use gate_rejects for drops.
85            rejects: 0,
86            misses,
87            gate_rejects,
88            preempt_kicks,
89            preempt_skipped,
90            red_rejects,
91            red_reclaims,
92        }
93    }
94
95    /// Read one per CPU stats row without heap use.
96    /// Failed lookups yield a zero row with no trap.
97    pub(crate) fn read_stats(&self, cpu: usize) -> crate::bpf_intf::flow_sched_stats {
98        let zero = crate::bpf_intf::flow_sched_stats {
99            on_cpu: 0,
100            total_runtime: 0,
101            inserts: 0,
102            requeues: 0,
103            completions: 0,
104            local_moves: 0,
105            node_moves: 0,
106            machine_moves: 0,
107            kicks: 0,
108            admits: 0,
109            rejects: 0,
110            misses: 0,
111            gate_rejects: 0,
112            preempt_kicks: 0,
113            preempt_skipped: 0,
114            red_rejects: 0,
115            red_reclaims: 0,
116        };
117        if cpu >= crate::bpf_intf::flow_consts_FLOW_MAX_CPUS as usize {
118            return zero;
119        }
120        let fd = self.skel.maps.cpu_stats_stor.as_fd().as_raw_fd();
121        let key = cpu as u32;
122        let mut out = MaybeUninit::<crate::bpf_intf::flow_sched_stats>::zeroed();
123        let ret = unsafe {
124            libbpf_rs::libbpf_sys::bpf_map_lookup_elem(
125                fd,
126                &key as *const _ as *const std::ffi::c_void,
127                out.as_mut_ptr() as *mut std::ffi::c_void,
128            )
129        };
130        if ret == 0 {
131            unsafe { out.assume_init() }
132        } else {
133            zero
134        }
135    }
136
137    /// Read one CPU pid view without heap use.
138    /// Failed lookups yield an idle view with zero pid.
139    pub(crate) fn read_cpu(&self, cpu: usize) -> crate::bpf_intf::flow_cpu_state {
140        let idle = crate::bpf_intf::flow_cpu_state {
141            running_pid: 0,
142            cursor: 0,
143            min_vruntime: 0,
144        };
145        if cpu >= crate::bpf_intf::flow_consts_FLOW_MAX_CPUS as usize {
146            return idle;
147        }
148        let fd = self.skel.maps.cpu_state_stor.as_fd().as_raw_fd();
149        let key = cpu as u32;
150        let mut out = MaybeUninit::<crate::bpf_intf::flow_cpu_state>::zeroed();
151        let ret = unsafe {
152            libbpf_rs::libbpf_sys::bpf_map_lookup_elem(
153                fd,
154                &key as *const _ as *const std::ffi::c_void,
155                out.as_mut_ptr() as *mut std::ffi::c_void,
156            )
157        };
158        if ret == 0 {
159            unsafe { out.assume_init() }
160        } else {
161            idle
162        }
163    }
164
165    /// Dashboard snapshot with raw counters plus live pid cards.
166    /// Counters stay raw with no deltas and the on CPU gauge counts live
167    /// pids with the pid view. Costs 3N map reads for N online CPUs: N
168    /// for the stats sum via get_metrics plus N for the gauge plus N for
169    /// the cards, with one alloc sized to the online count. SMT comes
170    /// from the cached init flags with no sysfs use on poll and stays
171    /// display only. Offline stays out, so per CPU count matches the
172    /// cached online count. Version plus timestamp plus topology join the
173    /// counters for the page plus the log.
174    pub(crate) fn get_web_metrics(&self) -> stats::WebMetrics {
175        let mut per_cpu = Vec::with_capacity(self.online_cpus.len());
176        for (rank, &id) in self.online_cpus.iter().enumerate() {
177            let st = self.read_cpu(id as usize);
178            let smt = self.smt.get(rank).copied().unwrap_or(false);
179            per_cpu.push(stats::PerCpuMetrics {
180                id,
181                smt,
182                running_pid: st.running_pid,
183                slice_ns: crate::config::QUANTUM_NS,
184            });
185        }
186        let timestamp_ns = std::time::SystemTime::now()
187            .duration_since(std::time::UNIX_EPOCH)
188            .map(|v| v.as_nanos().min(u64::MAX as u128) as u64)
189            .unwrap_or(0);
190        let metrics = self.get_metrics();
191        stats::WebMetrics {
192            stats: metrics,
193            per_cpu,
194            version: env!("CARGO_PKG_VERSION").to_string(),
195            timestamp_ns,
196            topology: self.topology.clone(),
197        }
198    }
199}
200
201#[cfg(test)]
202mod tests {
203    /// Poll cost in BPF map reads for N online CPUs.
204    /// Dashboard polls read N for the stats sum via get_metrics plus N
205    /// for the on CPU gauge plus N for the per CPU cards, so the total
206    /// stays 3N with one alloc and no sysfs on poll.
207    fn snapshot_reads(n: usize) -> usize {
208        n.saturating_mul(3)
209    }
210
211    #[test]
212    fn poll_cost_is_three_n() {
213        assert_eq!(snapshot_reads(0), 0);
214        assert_eq!(snapshot_reads(1), 3);
215        assert_eq!(snapshot_reads(4), 12);
216        // One alloc holds N cards with no extra walk.
217        let per_cpu = Vec::<u32>::with_capacity(4);
218        assert!(per_cpu.capacity() >= 4);
219    }
220}
scx_flow/rust/snapshot.rs

scx_flow/rust/
snapshot.rs
