scx_flow/rust/
stats.rs

1// SPDX-License-Identifier: GPL-2.0
2//! Stats server for the flow scheduler.
3//!
4//! Copyright (c) 2026 Galih Tama <galpt@v.recipes>
5
6//! Exports the counters view plus the dashboard view from the BPF maps.
7//! The stats server carries deltas while the dashboard carries raw
8//! counters plus per CPU cards for the loopback page.
9
10use std::io::Write;
11use std::sync::Arc;
12use std::sync::atomic::AtomicBool;
13use std::sync::atomic::Ordering;
14use std::time::Duration;
15
16use anyhow::Result;
17use scx_stats::prelude::*;
18use scx_stats_derive::Stats;
19use scx_stats_derive::stat_doc;
20use serde::Deserialize;
21use serde::Serialize;
22
23#[stat_doc]
24#[derive(Clone, Debug, Default, Serialize, Deserialize, Stats)]
25#[stat(top)]
26/// Counters with placement, admission, and miss detail.
27/// BPF holds 17 counters, Rust adds display-only uptime for eighteen.
28pub struct Metrics {
29    #[stat(desc = "Tasks now on a CPU")]
30    #[serde(default)]
31    pub on_cpu: u64,
32    #[stat(desc = "Total runtime in nanoseconds")]
33    #[serde(default)]
34    pub total_runtime: u64,
35    /// Display-only uptime since attach in nanos with no BPF use.
36    /// Filled from the start instant, never from the BPF counters.
37    #[stat(desc = "Uptime since attach in nanoseconds")]
38    #[serde(default)]
39    pub uptime_ns: u64,
40    #[stat(desc = "Fresh deadline joins")]
41    #[serde(default)]
42    pub inserts: u64,
43    #[stat(desc = "Runnable slice ends with requeue")]
44    #[serde(default)]
45    pub requeues: u64,
46    #[stat(desc = "Blocks and exits")]
47    #[serde(default)]
48    pub completions: u64,
49    #[stat(desc = "Moves from the local tier")]
50    #[serde(default)]
51    pub local_moves: u64,
52    #[stat(desc = "Moves from the node tier")]
53    #[serde(default)]
54    pub node_moves: u64,
55    #[stat(desc = "Moves from the machine tier")]
56    #[serde(default)]
57    pub machine_moves: u64,
58    #[stat(desc = "Idle wakeup kicks sent after insert")]
59    #[serde(default)]
60    pub kicks: u64,
61    #[stat(desc = "Tasks admitted with no bound")]
62    #[serde(default)]
63    pub admits: u64,
64    #[stat(
65        desc = "Rejects dead at zero whitelisted for wire compat with no writer; real rejects count in gate_rejects"
66    )]
67    #[serde(default)]
68    pub rejects: u64,
69    #[stat(desc = "Wall completions past deadline")]
70    #[serde(default)]
71    pub misses: u64,
72    #[stat(desc = "Closed gate rejects on stale CPUs plus tasks")]
73    #[serde(default)]
74    pub gate_rejects: u64,
75    #[stat(desc = "Busy preempts sent with margin plus tail")]
76    #[serde(default)]
77    pub preempt_kicks: u64,
78    #[stat(desc = "Preempts held by margin plus tail plus eligibility")]
79    #[serde(default)]
80    pub preempt_skipped: u64,
81    #[stat(desc = "RED overload rejects to the value ordered queue")]
82    #[serde(default)]
83    pub red_rejects: u64,
84    #[stat(desc = "RED reclaims from the value ordered queue")]
85    #[serde(default)]
86    pub red_reclaims: u64,
87}
88
89/// One card of the per CPU grid.
90/// Id plus SMT stay fixed while pid plus slice refresh on each poll.
91/// Pid holds zero when idle and slice holds the shared quantum.
92/// SMT marks the second thread of one core for display only with
93/// no placement use and false on old JSON.
94#[derive(Clone, Debug, Default, Serialize, Deserialize)]
95pub struct PerCpuMetrics {
96    /// CPU id.
97    #[serde(default)]
98    pub id: u32,
99    /// True for the second thread of one core with false on single thread.
100    /// Display only with no placement use plus false on old JSON.
101    #[serde(default)]
102    pub smt: bool,
103    /// Pid now on the CPU with zero when idle.
104    #[serde(default)]
105    pub running_pid: u32,
106    /// Fixed slice in nanos with the shared quantum.
107    #[serde(default)]
108    pub slice_ns: u64,
109}
110
111/// Snapshot for the web dashboard.
112/// Counters stay raw with the on CPU gauge plus the live pid view.
113/// The run loop pushes one per poll and the web thread keeps the
114/// newest behind a lock for the page plus the JSON routes.
115#[derive(Clone, Debug, Default, Serialize, Deserialize)]
116pub struct WebMetrics {
117    /// Scheduler wide counters with raw values.
118    pub stats: Metrics,
119    /// One entry per online CPU in rank order.
120    #[serde(default)]
121    pub per_cpu: Vec<PerCpuMetrics>,
122    /// Scheduler version for the page plus the log.
123    #[serde(default)]
124    pub version: String,
125    /// Wall time in nanos since epoch for the log.
126    #[serde(default)]
127    pub timestamp_ns: u64,
128    /// One line topology summary for the page.
129    #[serde(default)]
130    pub topology: String,
131}
132
133/// Stats printer loop for the monitor flag.
134/// Polls the stats server on the interval with plain text lines.
135pub fn monitor(intv: Duration, shutdown: Arc<AtomicBool>) -> Result<()> {
136    scx_utils::monitor_stats::<Metrics>(
137        &[],
138        intv,
139        || shutdown.load(Ordering::Relaxed),
140        |metrics| metrics.format(&mut std::io::stdout()),
141    )
142}
143
144/// Server data for the stats server.
145/// A single top op reports interval deltas of the counters.
146pub fn server_data() -> StatsServerData<(), Metrics> {
147    let open: Box<dyn StatsOpener<(), Metrics>> = Box::new(move |(req_ch, res_ch)| {
148        req_ch.send(())?;
149        let mut prev = res_ch.recv()?;
150        let read: Box<dyn StatsReader<(), Metrics>> = Box::new(move |_args, (req_ch, res_ch)| {
151            req_ch.send(())?;
152            let cur = res_ch.recv()?;
153            let delta = cur.delta(&prev);
154            prev = cur;
155            delta.to_json()
156        });
157        Ok(read)
158    });
159    StatsServerData::new()
160        .add_meta(Metrics::meta())
161        .add_ops("top", StatsOps { open, close: None })
162}
163
164impl Metrics {
165    fn format<W: Write>(&self, w: &mut W) -> Result<()> {
166        writeln!(
167            w,
168            "[{}] run={} runtime_ns={} uptime_ns={} ins={} req={} done={} \
169             local={} node={} machine={} kick={} adm={} rej={} \
170             miss={} gate={} pkick={} pskip={} redrej={} redrec={}",
171            crate::SCHEDULER_NAME,
172            self.on_cpu,
173            self.total_runtime,
174            self.uptime_ns,
175            self.inserts,
176            self.requeues,
177            self.completions,
178            self.local_moves,
179            self.node_moves,
180            self.machine_moves,
181            self.kicks,
182            self.admits,
183            self.rejects,
184            self.misses,
185            self.gate_rejects,
186            self.preempt_kicks,
187            self.preempt_skipped,
188            self.red_rejects,
189            self.red_reclaims,
190        )?;
191        Ok(())
192    }
193
194    /// Interval delta of the counters over the poll interval.
195    /// Gauges like on_cpu plus uptime_ns pass through as live values.
196    pub fn delta(&self, rhs: &Self) -> Self {
197        Self {
198            on_cpu: self.on_cpu,
199            total_runtime: self.total_runtime.wrapping_sub(rhs.total_runtime),
200            uptime_ns: self.uptime_ns,
201            inserts: self.inserts.wrapping_sub(rhs.inserts),
202            requeues: self.requeues.wrapping_sub(rhs.requeues),
203            completions: self.completions.wrapping_sub(rhs.completions),
204            local_moves: self.local_moves.wrapping_sub(rhs.local_moves),
205            node_moves: self.node_moves.wrapping_sub(rhs.node_moves),
206            machine_moves: self.machine_moves.wrapping_sub(rhs.machine_moves),
207            kicks: self.kicks.wrapping_sub(rhs.kicks),
208            admits: self.admits.wrapping_sub(rhs.admits),
209            rejects: self.rejects.wrapping_sub(rhs.rejects),
210            misses: self.misses.wrapping_sub(rhs.misses),
211            gate_rejects: self.gate_rejects.wrapping_sub(rhs.gate_rejects),
212            preempt_kicks: self.preempt_kicks.wrapping_sub(rhs.preempt_kicks),
213            preempt_skipped: self.preempt_skipped.wrapping_sub(rhs.preempt_skipped),
214            red_rejects: self.red_rejects.wrapping_sub(rhs.red_rejects),
215            red_reclaims: self.red_reclaims.wrapping_sub(rhs.red_reclaims),
216        }
217    }
218}
scx_flow/rust/stats.rs

scx_flow/rust/
stats.rs
