scx_rustsched_loader/
queues.rs

1// SPDX-License-Identifier: GPL-2.0-only
2// Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES.
3
4//! The per-CID queues, observed from the loader. The arena the BPF program
5//! keeps them in is mapped into this process too, at the addresses the
6//! program itself uses, so the loader reads a queue through the same
7//! `edq` code the program runs, over a memory that can only load.
8//!
9//! Nothing is added to the scheduling paths for this. A queue already
10//! counts the tasks it holds and the tasks ever queued on it, and both
11//! are read without its lock. The time a task waits in a queue follows
12//! from the two: by Little's law the average wait is the average depth
13//! over the arrival rate. The arrivals are exact, and the depth is
14//! sampled by a thread that runs only while a stats client is polling.
15
16use std::{
17    ffi::OsStr,
18    sync::{
19        Arc, Mutex,
20        atomic::{AtomicBool, AtomicU64, Ordering},
21    },
22    thread,
23    time::{Duration, Instant},
24};
25
26use anyhow::{Context, Result, ensure};
27use libbpf_rs::{AsRawLibbpf, MapCore, MapFlags};
28
29use crate::{
30    Loaded, OpsProfiler,
31    edq::{self, View},
32};
33
34/// The arena, read through the mapping libbpf made when it created the
35/// map. The addresses the BPF program stores are user addresses in that
36/// mapping, so they are read as they are. An address outside of it reads
37/// as 0. The view keeps the object, and with it the mapping, alive.
38#[derive(Clone)]
39pub struct ArenaView {
40    base: u64,
41    bytes: u64,
42    _object: Loaded,
43}
44
45impl ArenaView {
46    /// The arena of `object`.
47    pub fn new(object: &Loaded) -> Result<Self> {
48        let map = object
49            .maps()
50            .find(|map| map.name() == OsStr::new("arena"))
51            .context("find arena map")?;
52        let mut size = 0;
53        // SAFETY: the map belongs to `object`, which is alive.
54        let base = unsafe {
55            libbpf_rs::libbpf_sys::bpf_map__initial_value(
56                map.as_libbpf_object().as_ptr(),
57                &mut size,
58            )
59        };
60        ensure!(!base.is_null(), "the arena is not mapped");
61        // SAFETY: sysconf() has no preconditions.
62        let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }.max(1) as u64;
63        Ok(Self {
64            base: base as u64,
65            bytes: u64::from(map.max_entries()) * page,
66            _object: object.clone(),
67        })
68    }
69
70    fn contains(&self, address: u64, bytes: u64) -> bool {
71        address >= self.base && bytes <= self.bytes && address - self.base <= self.bytes - bytes
72    }
73
74    /// The word at arena address `address`, or 0 for an address that is
75    /// not an aligned word of the arena. The BPF program may write the
76    /// word at the same time: the read is atomic, of no particular order.
77    pub fn read(&self, address: u64) -> u64 {
78        if !address.is_multiple_of(8) || !self.contains(address, 8) {
79            return 0;
80        }
81        // SAFETY: the word is aligned and inside the arena's mapping,
82        // which lives as long as the object this view keeps.
83        unsafe { (*(address as *const AtomicU64)).load(Ordering::Relaxed) }
84    }
85}
86
87impl View for ArenaView {
88    fn load(&self, address: u64) -> u64 {
89        self.read(address)
90    }
91}
92
93/// A per-CID table of the scheduler's, read from the loader: the one whose
94/// header is in the `.data` section `section`, see scx_rustsched's `cid_slot!`,
95/// with `words` 64-bit words per CID. For counters each CPU keeps for
96/// itself, which cost the scheduling paths no shared write.
97pub struct CidTableReader {
98    view: ArenaView,
99    table: u64,
100    count: u64,
101    words: u64,
102}
103
104impl CidTableReader {
105    /// The table of the attached scheduler in `object`.
106    pub fn new(object: &Loaded, section: &str, words: usize) -> Result<Self> {
107        let view = ArenaView::new(object)?;
108        let header = object
109            .maps()
110            .find(|map| map.name() == OsStr::new(section))
111            .with_context(|| format!("find the table {section}"))?;
112        let mut data = [0_u8; 16];
113        ensure!(
114            header.lookup_into(&0_u32.to_ne_bytes(), &mut data, MapFlags::ANY)?,
115            "read the header of {section}"
116        );
117        let table = u64::from_ne_bytes(data[..8].try_into()?);
118        let count = u64::from_ne_bytes(data[8..].try_into()?);
119        let words = words as u64;
120        ensure!(
121            count != 0 && count <= u64::from(u32::MAX) && view.contains(table, count * words * 8),
122            "the table {section} is not in the arena"
123        );
124        Ok(Self {
125            view,
126            table,
127            count,
128            words,
129        })
130    }
131
132    /// Each word of the entry, summed over the CIDs.
133    pub fn sum(&self) -> Vec<u64> {
134        let mut sums = vec![0_u64; self.words as usize];
135        for cid in 0..self.count {
136            for (word, sum) in sums.iter_mut().enumerate() {
137                let address = self.table + (cid * self.words + word as u64) * 8;
138                *sum = sum.wrapping_add(self.view.read(address));
139            }
140        }
141        sums
142    }
143}
144
145/// What the sampler saw of one CID's queue, cumulative.
146#[derive(Clone, Copy, Debug, Default)]
147pub struct QueueStats {
148    /// Tasks ever queued, the queue's own count.
149    pub inserts: u64,
150    /// Sum of the depths sampled.
151    pub depth_sum: u64,
152    /// Samples that found tasks queued.
153    pub busy: u64,
154    /// Samples that found the queue's lock held.
155    pub locked: u64,
156}
157
158struct State {
159    last_read: Instant,
160    passes: u64,
161    queues: Vec<QueueStats>,
162}
163
164struct Shared {
165    view: ArenaView,
166    table: u64,
167    stop: AtomicBool,
168    state: Mutex<State>,
169}
170
171impl Shared {
172    fn queue(&self, cid: usize) -> u64 {
173        self.table + cid as u64 * edq::QUEUE_BYTES
174    }
175
176    /// Sample every queue once.
177    fn pass(&self) {
178        let mut state = self.state.lock().unwrap();
179        for (cid, queue) in state.queues.iter_mut().enumerate() {
180            let address = self.queue(cid);
181            let depth = edq::nr_queued(&self.view, address);
182            queue.depth_sum += depth;
183            queue.busy += u64::from(depth != 0);
184            queue.locked += u64::from(edq::locked(&self.view, address));
185        }
186        state.passes += 1;
187    }
188
189    fn run(&self) {
190        // xorshift, seeded with anything nonzero.
191        let mut random = self.table | 1;
192        while !self.stop.load(Ordering::Relaxed) {
193            let idle = self.state.lock().unwrap().last_read.elapsed() > OpsProfiler::IDLE;
194            if idle {
195                thread::park();
196                continue;
197            }
198            self.pass();
199            random ^= random << 13;
200            random ^= random >> 7;
201            random ^= random << 17;
202            thread::sleep(
203                QueueSampler::PERIOD_MIN
204                    + QueueSampler::PERIOD_SPREAD
205                        .mul_f64((random >> 11) as f64 / (1_u64 << 53) as f64),
206            );
207        }
208    }
209}
210
211/// Samples the depth of every CID's queue while a stats client is polling,
212/// and parks once nobody has asked for a while, as the callback profiler
213/// drops its accounting.
214pub struct QueueSampler {
215    shared: Arc<Shared>,
216    thread: Option<thread::JoinHandle<()>>,
217}
218
219impl QueueSampler {
220    /// The time between two passes is uniform between PERIOD_MIN and
221    /// PERIOD_MIN + PERIOD_SPREAD. A fixed period would beat against the
222    /// tick and against periodic workloads, and see the same phase of
223    /// them every time.
224    const PERIOD_MIN: Duration = Duration::from_millis(5);
225    const PERIOD_SPREAD: Duration = Duration::from_millis(10);
226
227    /// Start sampling the queues of the attached scheduler in `object`.
228    pub fn new(object: &Loaded) -> Result<Self> {
229        let view = ArenaView::new(object)?;
230        let queues = object
231            .maps()
232            .find(|map| map.name() == OsStr::new(".data.queues"))
233            .context("find the queue table")?;
234        let mut data = [0_u8; 24];
235        ensure!(
236            queues.lookup_into(&0_u32.to_ne_bytes(), &mut data, MapFlags::ANY)?,
237            "read the queue table"
238        );
239        let table = u64::from_ne_bytes(data[..8].try_into()?);
240        let count = u64::from_ne_bytes(data[8..16].try_into()?);
241        ensure!(
242            count != 0
243                && count <= u64::from(u32::MAX)
244                && view.contains(table, count * edq::QUEUE_BYTES),
245            "the queue table is not in the arena"
246        );
247        let shared = Arc::new(Shared {
248            view,
249            table,
250            stop: AtomicBool::new(false),
251            state: Mutex::new(State {
252                // Parked until the first client asks.
253                last_read: Instant::now() - OpsProfiler::IDLE * 2,
254                passes: 0,
255                queues: vec![QueueStats::default(); count as usize],
256            }),
257        });
258        let sampler = shared.clone();
259        let thread = thread::Builder::new()
260            .name("edq-sampler".into())
261            .spawn(move || sampler.run())
262            .context("start the queue sampler")?;
263        Ok(Self {
264            shared,
265            thread: Some(thread),
266        })
267    }
268
269    /// The number of sampling passes so far and every queue's statistics,
270    /// indexed by CID, with the insert counts read now.
271    pub fn sample(&self) -> (u64, Vec<QueueStats>) {
272        let (passes, mut queues) = {
273            let mut state = self.shared.state.lock().unwrap();
274            state.last_read = Instant::now();
275            (state.passes, state.queues.clone())
276        };
277        if let Some(thread) = &self.thread {
278            thread.thread().unpark();
279        }
280        for (cid, queue) in queues.iter_mut().enumerate() {
281            queue.inserts = edq::nr_inserted(&self.shared.view, self.shared.queue(cid));
282        }
283        (passes, queues)
284    }
285}
286
287impl Drop for QueueSampler {
288    fn drop(&mut self) {
289        self.shared.stop.store(true, Ordering::Relaxed);
290        if let Some(thread) = self.thread.take() {
291            thread.thread().unpark();
292            let _ = thread.join();
293        }
294    }
295}
scx_rustsched_loader/queues.rs

scx_rustsched_loader/
queues.rs
