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