Skip to main content

scx_mitosis/
stats.rs

1use std::collections::BTreeMap;
2use std::io::Write;
3use std::sync::atomic::AtomicBool;
4use std::sync::atomic::Ordering;
5use std::sync::Arc;
6use std::time::Duration;
7
8use anyhow::Result;
9use serde::Deserialize;
10use serde::Serialize;
11
12use scx_stats::prelude::*;
13use scx_stats_derive::stat_doc;
14use scx_stats_derive::Stats;
15
16use crate::DistributionStats;
17
18#[stat_doc]
19#[derive(Clone, Debug, Default, Serialize, Deserialize, Stats)]
20#[stat(_om_prefix = "c_")]
21#[stat(top)]
22pub struct CellMetrics {
23    #[stat(desc = "Number of cpus")]
24    pub num_cpus: u32,
25    #[stat(
26        desc = "Cgroup path of this cell, relative to the cgroup root (\"/\" for the root cell)",
27        _om_skip
28    )]
29    pub cgroup_path: String,
30    #[stat(desc = "Local queue %")]
31    pub local_q_pct: f64,
32    #[stat(desc = "CPU queue %")]
33    pub cpu_q_pct: f64,
34    #[stat(desc = "Cell queue %")]
35    pub cell_q_pct: f64,
36    #[stat(desc = "Borrowed CPU %")]
37    pub borrowed_pct: f64,
38    #[stat(desc = "Affinity violations % of global")]
39    pub affn_violations_pct: f64,
40    #[stat(desc = "Steal %")]
41    pub steal_pct: f64,
42    #[stat(desc = "Orphaned LLC DSQ drain events")]
43    pub drain_cnt: u64,
44    #[stat(desc = "Orphaned LLC DSQ affinity rescue events")]
45    pub drain_affn_cnt: u64,
46    #[stat(desc = "Pin reject skipped %")]
47    pub pin_skip_pct: f64,
48    #[stat(desc = "Slice shrink events")]
49    pub slice_shrink: u64,
50    #[stat(desc = "Slice shrink at max")]
51    pub slice_shrink_max: u64,
52    #[stat(desc = "Slice shrink proportional")]
53    pub slice_shrink_proportional: u64,
54    #[stat(desc = "Slice shrink at min")]
55    pub slice_shrink_min: u64,
56    #[stat(desc = "Decision share % of global")]
57    pub share_of_decisions_pct: f64,
58    #[stat(desc = "Cell scheduling decisions")]
59    total_decisions: u64,
60    #[stat(desc = "CPU utilization %")]
61    pub util_pct: f64,
62    #[stat(desc = "Borrowed CPU time % of running")]
63    pub demand_borrow_pct: f64,
64    #[stat(desc = "Lent CPU time %")]
65    pub lent_pct: f64,
66    #[stat(desc = "EWMA-smoothed utilization %")]
67    pub smoothed_util_pct: f64,
68}
69
70impl CellMetrics {
71    pub fn update(&mut self, ds: &DistributionStats) {
72        self.local_q_pct = ds.local_q_pct;
73        self.cpu_q_pct = ds.cpu_q_pct;
74        self.cell_q_pct = ds.cell_q_pct;
75        self.borrowed_pct = ds.borrowed_pct;
76        self.affn_violations_pct = ds.affn_viol_pct;
77        self.steal_pct = ds.steal_pct;
78        self.pin_skip_pct = ds.pin_skip_pct;
79        self.share_of_decisions_pct = ds.share_of_decisions_pct;
80        self.total_decisions = ds.total_decisions;
81    }
82
83    pub fn update_demand(&mut self, util_pct: f64, demand_borrow_pct: f64, lent_pct: f64) {
84        self.util_pct = util_pct;
85        self.demand_borrow_pct = demand_borrow_pct;
86        self.lent_pct = lent_pct;
87    }
88}
89
90#[stat_doc]
91#[derive(Clone, Debug, Default, Serialize, Deserialize, Stats)]
92#[stat(top)]
93pub struct Metrics {
94    #[stat(desc = "Number of cells")]
95    pub num_cells: u32,
96    #[stat(desc = "Local queue %")]
97    pub local_q_pct: f64,
98    #[stat(desc = "CPU queue %")]
99    pub cpu_q_pct: f64,
100    #[stat(desc = "Cell queue %")]
101    pub cell_q_pct: f64,
102    #[stat(desc = "Borrowed CPU %")]
103    pub borrowed_pct: f64,
104    #[stat(desc = "Affinity violations % of global")]
105    pub affn_violations_pct: f64,
106    #[stat(desc = "Steal %")]
107    pub steal_pct: f64,
108    #[stat(desc = "Orphaned LLC DSQ drain events")]
109    pub drain_cnt: u64,
110    #[stat(desc = "Orphaned LLC DSQ affinity rescue events")]
111    pub drain_affn_cnt: u64,
112    #[stat(desc = "Pin reject skipped %")]
113    pub pin_skip_pct: f64,
114    #[stat(desc = "Slice shrink events")]
115    pub slice_shrink: u64,
116    #[stat(desc = "Slice shrink at max")]
117    pub slice_shrink_max: u64,
118    #[stat(desc = "Slice shrink proportional")]
119    pub slice_shrink_proportional: u64,
120    #[stat(desc = "Slice shrink at min")]
121    pub slice_shrink_min: u64,
122    #[stat(desc = "Decision share % of global")]
123    pub share_of_decisions_pct: f64,
124    #[stat(desc = "Cell scheduling decisions")]
125    total_decisions: u64,
126    #[stat(desc = "CPU utilization %")]
127    pub util_pct: f64,
128    #[stat(desc = "Borrowed CPU time % of running")]
129    pub demand_borrow_pct: f64,
130    #[stat(desc = "Lent CPU time %")]
131    pub lent_pct: f64,
132    #[stat(desc = "Number of rebalancing events")]
133    pub rebalance_count: u64,
134    #[stat(
135        desc = "1 if the cell-0 holdout has taken a CPU already claimed by a workload cell, else 0"
136    )]
137    pub enforced_holdout: u64,
138    #[stat(desc = "Per-cell metrics")]
139    pub cells: BTreeMap<u32, CellMetrics>,
140}
141
142impl Metrics {
143    pub fn update(&mut self, ds: &DistributionStats) {
144        self.local_q_pct = ds.local_q_pct;
145        self.cpu_q_pct = ds.cpu_q_pct;
146        self.cell_q_pct = ds.cell_q_pct;
147        self.borrowed_pct = ds.borrowed_pct;
148        self.affn_violations_pct = ds.affn_viol_pct;
149        self.steal_pct = ds.steal_pct;
150        self.pin_skip_pct = ds.pin_skip_pct;
151        self.share_of_decisions_pct = ds.share_of_decisions_pct;
152        self.total_decisions = ds.total_decisions;
153    }
154
155    pub fn update_demand(&mut self, util_pct: f64, demand_borrow_pct: f64, lent_pct: f64) {
156        self.util_pct = util_pct;
157        self.demand_borrow_pct = demand_borrow_pct;
158        self.lent_pct = lent_pct;
159    }
160
161    fn delta(&self, _: &Self) -> Self {
162        Self { ..self.clone() }
163    }
164
165    fn format<W: Write>(&self, w: &mut W) -> Result<()> {
166        writeln!(w, "{}", serde_json::to_string_pretty(self)?)?;
167        Ok(())
168    }
169}
170
171pub fn server_data() -> StatsServerData<(), Metrics> {
172    let open: Box<dyn StatsOpener<(), Metrics>> = Box::new(move |(req_ch, res_ch)| {
173        req_ch.send(())?;
174        let mut prev = res_ch.recv()?;
175
176        let read: Box<dyn StatsReader<(), Metrics>> = Box::new(move |_args, (req_ch, res_ch)| {
177            req_ch.send(())?;
178            let cur = res_ch.recv()?;
179            let delta = cur.delta(&prev);
180            prev = cur;
181            delta.to_json()
182        });
183
184        Ok(read)
185    });
186
187    StatsServerData::new()
188        .add_meta(Metrics::meta())
189        .add_meta(CellMetrics::meta())
190        .add_ops("top", StatsOps { open, close: None })
191}
192
193pub fn monitor(intv: Duration, shutdown: Arc<AtomicBool>) -> Result<()> {
194    scx_utils::monitor_stats::<Metrics>(
195        &[],
196        intv,
197        || shutdown.load(Ordering::Relaxed),
198        |metrics| metrics.format(&mut std::io::stdout()),
199    )
200}