scx_flow/
main.rs

1// SPDX-License-Identifier: GPL-2.0
2//! Flow scheduler front end.
3//!
4//! Copyright (c) 2026 Galih Tama <galpt@v.recipes>
5
6//! Loads the BPF object, seeds the topology view, then drives the loop.
7//! Observability is counters through the stats server plus the loopback
8//! dashboard with per CPU cards plus a JSON snapshot for debugging.
9
10mod bpf_skel;
11pub use bpf_skel::*;
12pub mod bpf_intf;
13pub use bpf_intf::*;
14#[path = "rust/config.rs"]
15mod config;
16#[path = "rust/snapshot.rs"]
17mod snapshot;
18#[path = "rust/stats.rs"]
19mod stats;
20#[path = "rust/topology.rs"]
21mod topology;
22#[path = "rust/webui.rs"]
23mod webui;
24
25use std::mem::MaybeUninit;
26use std::sync::Arc;
27use std::sync::atomic::AtomicBool;
28use std::sync::atomic::Ordering;
29use std::time::Duration;
30
31use anyhow::Result;
32use clap::CommandFactory;
33use clap::Parser;
34use clap_complete::Shell;
35use clap_complete::generate;
36use crossbeam::channel::RecvTimeoutError;
37use log::info;
38use scx_stats::prelude::*;
39use scx_utils::UserExitInfo;
40use scx_utils::build_id;
41use scx_utils::libbpf_clap_opts::LibbpfOpts;
42use scx_utils::scx_ops_attach;
43use scx_utils::scx_ops_load;
44use scx_utils::scx_ops_open;
45use scx_utils::try_set_rlimit_infinity;
46use scx_utils::uei_exited;
47use scx_utils::uei_report;
48
49use config::Config;
50use stats::Metrics;
51
52/* Binary name used in logs and stats. */
53const SCHEDULER_NAME: &str = "scx_flow";
54/* CPU bound shared with the BPF header. */
55#[cfg(test)]
56const MAX_CPUS: usize = crate::bpf_intf::flow_consts_FLOW_MAX_CPUS as usize;
57
58fn full_version() -> String {
59    build_id::full_version(env!("CARGO_PKG_VERSION"))
60}
61
62/// Validated poll interval from float seconds.
63/// Rejects NaN plus infinite plus non-positive plus over one hour, so
64/// a bad flag fails closed before any timer starts with no panic.
65fn poll_interval_secs(v: f64) -> Result<Duration> {
66    if !v.is_finite() || v <= 0.0 || v > 3600.0 {
67        anyhow::bail!("interval bad {v}");
68    }
69    Ok(Duration::from_secs_f64(v))
70}
71
72#[derive(Debug, Parser)]
73#[command(name = SCHEDULER_NAME, version, disable_version_flag = true)]
74struct Opts {
75    /* Poll interval for the stats printer. */
76    #[clap(long)]
77    stats: Option<f64>,
78    /* Run the stats printer only. */
79    #[clap(long)]
80    monitor: Option<f64>,
81    /* Verbose BPF logging. */
82    #[clap(short = 'd', long, action = clap::ArgAction::SetTrue)]
83    debug: bool,
84    /* Verbose output with libbpf detail. */
85    #[clap(short = 'v', long, action = clap::ArgAction::SetTrue)]
86    verbose: bool,
87    /* Exit dump buffer length in bytes. */
88    #[clap(long, default_value = "1048576")]
89    exit_dump_len: u32,
90    /* Print version and exit. */
91    #[clap(short = 'V', long, action = clap::ArgAction::SetTrue)]
92    version: bool,
93    /* Show stat descriptions. */
94    #[clap(long)]
95    help_stats: bool,
96    /* Generate shell completions and exit. */
97    #[clap(long, value_name = "SHELL", hide = true)]
98    completions: Option<Shell>,
99    /* Disable the loopback dashboard thread. */
100    #[clap(long = "no-webui", action = clap::ArgAction::SetTrue)]
101    no_webui: bool,
102    #[clap(flatten, next_help_heading = "Libbpf Options")]
103    libbpf: LibbpfOpts,
104}
105
106/*
107 * Scheduler owns the skeleton, the link, the stats
108 * server plus the dashboard channel. It drives the run
109 * loop until shutdown or exit.
110 */
111pub(crate) struct Scheduler<'a> {
112    skel: BpfSkel<'a>,
113    struct_ops: Option<libbpf_rs::Link>,
114    stats_server: StatsServer<(), Metrics>,
115    started_at: std::time::Instant,
116    /* Dashboard sender with None when disabled. */
117    webui_tx: Option<crossbeam::channel::Sender<stats::WebMetrics>>,
118    /* Online ids once at init in rank order. */
119    online_cpus: Vec<u32>,
120    /* SMT flag per online CPU in rank order for the cards. */
121    smt: Vec<bool>,
122    /* One line topology summary for the page. */
123    topology: String,
124}
125
126impl<'a> Scheduler<'a> {
127    fn init(
128        opts: &'a Opts,
129        open_object: &'a mut MaybeUninit<libbpf_rs::OpenObject>,
130        shutdown: Arc<AtomicBool>,
131    ) -> Result<Self> {
132        try_set_rlimit_infinity();
133        let mut bld = BpfSkelBuilder::default();
134        bld.obj_builder.debug(opts.debug || opts.verbose);
135        let open_opts = opts.libbpf.clone().into_bpf_open_opts();
136        let mut skel = scx_ops_open!(bld, open_object, flow_ops, open_opts)?;
137        /* Validate the constants before load. */
138        let cfg = Config::default();
139        cfg.validate()?;
140        info!("Config: {}", cfg.describe());
141        /* Ops flags live in the BPF object for kernel 7.2 and up. */
142        skel.struct_ops.flow_ops_mut().exit_dump_len = opts.exit_dump_len;
143        let mut skel = scx_ops_load!(skel, flow_ops, uei)?;
144        let _ = &mut skel;
145        /* Seed the BPF topology view with sibling plus node rows. */
146        /* Failures keep the BPF defaults with node zero. */
147        let rows = topology::topo_rows();
148        Self::seed_topo_with(&mut skel, &rows);
149        let struct_ops = scx_ops_attach!(skel, flow_ops)?;
150        let stats_server = StatsServer::new(stats::server_data()).launch()?;
151        /* Bounded dashboard channel drops a frame when full. */
152        let webui_tx = if opts.no_webui {
153            None
154        } else {
155            let (tx, rx) = crossbeam::channel::bounded::<stats::WebMetrics>(16);
156            let sd = shutdown.clone();
157            std::thread::spawn(move || {
158                webui::start(rx, sd);
159            });
160            Some(tx)
161        };
162        let online_cpus: Vec<u32> = rows.iter().map(|(cpu, _, _)| *cpu).collect();
163        /* SMT flags cache the sibling view with no sysfs use on poll. */
164        let smt: Vec<bool> = rows
165            .iter()
166            .map(|(cpu, sib, _)| topology::is_smt_thread(*cpu, *sib))
167            .collect();
168        let topology = topology::describe_topology(&rows);
169        info!("Topology: {topology}");
170        Ok(Self {
171            skel,
172            struct_ops: Some(struct_ops),
173            stats_server,
174            started_at: std::time::Instant::now(),
175            webui_tx,
176            online_cpus,
177            smt,
178            topology,
179        })
180    }
181
182    fn exited(&self) -> bool {
183        uei_exited!(&self.skel, uei)
184    }
185
186    /* Seed one topology row per CPU into the BPF view. */
187    /* Each row carries the thread sibling and the node id. */
188    /* A failed update keeps the BPF default with no trap. */
189    fn seed_topo_with(skel: &mut BpfSkel<'_>, rows: &[(u32, u32, u32)]) {
190        use libbpf_rs::MapCore;
191        for (cpu, sib, node) in rows {
192            let key = cpu.to_ne_bytes();
193            let mut val = [0u8; 8];
194            val[0..4].copy_from_slice(&sib.to_ne_bytes());
195            val[4..8].copy_from_slice(&node.to_ne_bytes());
196            if let Err(e) = skel
197                .maps
198                .topo_stor
199                .update(&key, &val, libbpf_rs::MapFlags::ANY)
200            {
201                log::warn!("topo seed failed for cpu {cpu}: {e}");
202            }
203        }
204    }
205
206    fn run(&mut self, shutdown: Arc<AtomicBool>) -> Result<UserExitInfo> {
207        let (res_ch, req_ch) = self.stats_server.channels();
208        /* Short tick keeps stats polls prompt while the page polls */
209        /* once per second, so most ticks only refresh the bound */
210        /* channel when the dashboard is on plus the queue has room. */
211        /* One BPF read serves both the stats reply plus the page. */
212        while !shutdown.load(Ordering::Relaxed) && !self.exited() {
213            match req_ch.recv_timeout(Duration::from_millis(100)) {
214                Ok(()) => {
215                    if let Some(ref tx) = self.webui_tx {
216                        if !tx.is_full() {
217                            let web = self.get_web_metrics();
218                            let stats = web.stats.clone();
219                            let _ = tx.try_send(web);
220                            res_ch.send(stats)?
221                        } else {
222                            res_ch.send(self.get_metrics())?
223                        }
224                    } else {
225                        res_ch.send(self.get_metrics())?
226                    }
227                }
228                Err(RecvTimeoutError::Timeout) => {
229                    if let Some(ref tx) = self.webui_tx
230                        && !tx.is_full()
231                    {
232                        let web = self.get_web_metrics();
233                        let _ = tx.try_send(web);
234                    }
235                }
236                Err(e) => Err(e)?,
237            }
238        }
239        let m = self.get_metrics();
240        info!(
241            "exit ins={} req={} done={} local={} node={} machine={} kick={} adm={} rej={} miss={} gate={} pkick={} pskip={} redrej={} redrec={} runtime={} oncpu={}",
242            m.inserts,
243            m.requeues,
244            m.completions,
245            m.local_moves,
246            m.node_moves,
247            m.machine_moves,
248            m.kicks,
249            m.admits,
250            m.rejects,
251            m.misses,
252            m.gate_rejects,
253            m.preempt_kicks,
254            m.preempt_skipped,
255            m.red_rejects,
256            m.red_reclaims,
257            m.total_runtime,
258            m.on_cpu,
259        );
260        let _ = self.struct_ops.take();
261        uei_report!(&self.skel, uei)
262    }
263}
264
265fn main() -> Result<()> {
266    let opts = Opts::parse();
267    if let Some(shell) = opts.completions {
268        generate(
269            shell,
270            &mut Opts::command(),
271            SCHEDULER_NAME,
272            &mut std::io::stdout(),
273        );
274        return Ok(());
275    }
276    let only = opts.monitor.is_some();
277    if opts.version {
278        println!("{} {}", SCHEDULER_NAME, full_version());
279        return Ok(());
280    }
281    if opts.help_stats {
282        println!("stats: top");
283        return Ok(());
284    }
285    if !only {
286        simplelog::SimpleLogger::init(
287            if opts.debug {
288                simplelog::LevelFilter::Debug
289            } else {
290                simplelog::LevelFilter::Info
291            },
292            simplelog::Config::default(),
293        )?;
294        info!("{} {}", SCHEDULER_NAME, full_version());
295        info!("Starting {} scheduler", SCHEDULER_NAME);
296    }
297    let shutdown = Arc::new(AtomicBool::new(false));
298    let sd = shutdown.clone();
299    ctrlc::set_handler(move || {
300        sd.store(true, Ordering::Relaxed);
301    })?;
302    if let Some(intv) = opts.monitor.or(opts.stats) {
303        let dur = poll_interval_secs(intv)?;
304        let sd = shutdown.clone();
305        let jh = std::thread::spawn(move || {
306            if let Err(e) = stats::monitor(dur, sd) {
307                log::warn!("monitor failed: {e}");
308            }
309        });
310        if only {
311            let _ = jh.join();
312            return Ok(());
313        }
314    }
315    let mut open_object = MaybeUninit::<libbpf_rs::OpenObject>::uninit();
316    let mut sched = Scheduler::init(&opts, &mut open_object, shutdown.clone())?;
317    sched.run(shutdown)?;
318    info!("Scheduler exited");
319    Ok(())
320}
321
322#[cfg(test)]
323mod tests {
324    use super::*;
325
326    #[test]
327    fn scheduler_name_is_flow() {
328        assert_eq!(SCHEDULER_NAME, "scx_flow");
329        assert!(poll_interval_secs(1.0).is_ok());
330        assert!(poll_interval_secs(0.0).is_err());
331        assert!(poll_interval_secs(-1.0).is_err());
332        assert!(poll_interval_secs(f64::NAN).is_err());
333        assert!(poll_interval_secs(f64::INFINITY).is_err());
334        assert!(poll_interval_secs(3601.0).is_err());
335    }
336
337    #[test]
338    fn max_cpus_matches_header() {
339        assert_eq!(
340            MAX_CPUS,
341            crate::bpf_intf::flow_consts_FLOW_MAX_CPUS as usize
342        );
343    }
344
345    #[test]
346    fn quantum_matches_header() {
347        assert_eq!(
348            crate::config::QUANTUM_NS,
349            crate::bpf_intf::flow_consts_FLOW_QUANTUM_NS as u64
350        );
351        assert_eq!(crate::config::QUANTUM_NS, 1_000_000);
352        assert_eq!(crate::config::WEIGHT_BASE, 128);
353        assert_eq!(crate::config::WEIGHT_MIN, 1);
354        assert_eq!(crate::config::WEIGHT_MAX, 16_384);
355    }
356
357    #[test]
358    fn slot_matches_header() {
359        assert_eq!(crate::bpf_intf::flow_consts_FLOW_MACHINE as u64, 0x5A00);
360        assert_eq!(crate::bpf_intf::flow_consts_FLOW_OVERFLOW as u64, 0x5A01);
361        assert_eq!(crate::bpf_intf::flow_consts_FLOW_MAX_DSQS as u64, 1042);
362    }
363
364    #[test]
365    fn period_matches_header() {
366        assert_eq!(
367            crate::config::PERIOD_NS,
368            crate::bpf_intf::flow_consts_FLOW_PERIOD_NS as u64
369        );
370        assert_eq!(crate::config::PERIOD_NS, 80_000_000);
371    }
372
373    #[test]
374    fn task_size_is_72() {
375        assert_eq!(std::mem::size_of::<crate::bpf_intf::flow_task_ctx>(), 72);
376    }
377
378    #[test]
379    fn cpu_size_is_16() {
380        assert_eq!(std::mem::size_of::<crate::bpf_intf::flow_cpu_state>(), 16);
381    }
382
383    #[test]
384    fn topo_size_is_8() {
385        assert_eq!(std::mem::size_of::<crate::bpf_intf::flow_topo>(), 8);
386    }
387
388    #[test]
389    fn sched_stats_size_is_136() {
390        assert_eq!(
391            std::mem::size_of::<crate::bpf_intf::flow_sched_stats>(),
392            136
393        );
394    }
395
396    #[test]
397    fn fixed_red_matches_header() {
398        assert_eq!(
399            crate::bpf_intf::flow_consts_FLOW_RED_TOL_NS as u64,
400            crate::config::RED_TOL_NS
401        );
402        assert_eq!(crate::config::RED_TOL_NS, 64_000);
403        assert_eq!(
404            crate::bpf_intf::flow_consts_FLOW_RED_EMAX_NS as u64,
405            crate::config::RED_EMAX_NS
406        );
407        assert_eq!(crate::config::RED_EMAX_NS, 256_000);
408    }
409
410    #[test]
411    fn kick_rule_matches_header() {
412        assert_eq!(
413            crate::bpf_intf::flow_consts_FLOW_PREEMPT_MARGIN_NS as u64,
414            crate::config::PREEMPT_MARGIN_NS
415        );
416        assert_eq!(crate::config::PREEMPT_MARGIN_NS, 100_000);
417        assert_eq!(
418            crate::bpf_intf::flow_consts_FLOW_PREEMPT_TAIL_NS as u64,
419            crate::config::PREEMPT_TAIL_NS
420        );
421        assert_eq!(crate::config::PREEMPT_TAIL_NS, 100_000);
422    }
423
424    #[test]
425    fn hint_matches_header() {
426        assert_eq!(
427            crate::config::HINT_MAX,
428            crate::bpf_intf::flow_consts_FLOW_HINT_MAX as u64
429        );
430        assert_eq!(crate::config::HINT_MAX, 8192);
431    }
432}
scx_flow/main.rs

scx_flow/
main.rs
