o_sfu_telemetry/metrics/
counter.rs

1//! lock-free metric counters for runtime hot paths
2//!
3//! counters use relaxed atomics because snapshots are observational telemetry
4//! and never synchronize runtime state
5//! padded counters isolate high-frequency worker labels onto cache lines
6
7use std::{
8    marker::PhantomData,
9    sync::atomic::{AtomicI64, AtomicU64, Ordering},
10    time::Duration,
11};
12
13pub(super) trait MetricLabel: Copy + 'static {
14    const VARIANTS: &'static [Self];
15    const COUNT: usize;
16
17    fn as_index(self) -> usize;
18}
19
20pub(super) trait ExportedMetricLabel: MetricLabel {
21    fn label_value(self) -> &'static str;
22}
23
24pub(super) trait MetricBucketLabel: MetricLabel {
25    fn upper_bound(self) -> &'static str;
26}
27
28pub(super) trait HistogramBucketLabel: MetricBucketLabel {
29    /// Returns the first inclusive finite bound or `None` above the largest bound.
30    fn from_duration(duration: Duration) -> Option<Self>;
31}
32
33#[repr(align(64))]
34#[derive(Debug, Default)]
35pub(super) struct PaddedCounter {
36    value: Counter,
37}
38
39impl PaddedCounter {
40    pub(super) fn increment(&self) {
41        self.value.increment();
42    }
43
44    pub(super) fn add(&self, value: usize) {
45        self.value.add(value);
46    }
47
48    pub(super) fn add_u64(&self, value: u64) {
49        self.value.add_u64(value);
50    }
51
52    pub(super) fn load(&self) -> u64 {
53        self.value.load()
54    }
55}
56
57#[derive(Debug, Default)]
58pub(super) struct Counter {
59    value: AtomicU64,
60}
61
62impl Counter {
63    pub(super) fn increment(&self) {
64        self.value.fetch_add(1, Ordering::Relaxed);
65    }
66
67    pub(super) fn add(&self, value: usize) {
68        if let Ok(value) = u64::try_from(value) {
69            self.value.fetch_add(value, Ordering::Relaxed);
70        }
71    }
72
73    pub(super) fn add_u64(&self, value: u64) {
74        self.value.fetch_add(value, Ordering::Relaxed);
75    }
76
77    pub(super) fn load(&self) -> u64 {
78        self.value.load(Ordering::Relaxed)
79    }
80}
81
82#[derive(Debug, Default)]
83pub(super) struct UpDownCounter {
84    value: AtomicI64,
85}
86
87impl UpDownCounter {
88    pub(super) fn add(&self, delta: i64) {
89        self.value.fetch_add(delta, Ordering::Relaxed);
90    }
91
92    pub(super) fn load(&self) -> i64 {
93        self.value.load(Ordering::Relaxed)
94    }
95}
96
97#[derive(Debug)]
98pub(super) struct UpDownCounterFamily<L: MetricLabel> {
99    counters: Box<[UpDownCounter]>,
100    _label: PhantomData<L>,
101}
102
103impl<L: MetricLabel> Default for UpDownCounterFamily<L> {
104    fn default() -> Self {
105        let counters = (0..L::COUNT)
106            .map(|_| UpDownCounter::default())
107            .collect::<Vec<_>>()
108            .into_boxed_slice();
109        Self {
110            counters,
111            _label: PhantomData,
112        }
113    }
114}
115
116impl<L: MetricLabel> UpDownCounterFamily<L> {
117    pub(super) fn add(&self, label: L, delta: i64) {
118        if let Some(counter) = self.counters.get(label.as_index()) {
119            counter.add(delta);
120        }
121    }
122
123    pub(super) fn load(&self, label: L) -> i64 {
124        self.counters
125            .get(label.as_index())
126            .map_or(0, UpDownCounter::load)
127    }
128}
129
130#[derive(Debug)]
131pub(super) struct PaddedCounterFamily<L: MetricLabel> {
132    counters: Box<[PaddedCounter]>,
133    _label: PhantomData<L>,
134}
135
136impl<L: MetricLabel> Default for PaddedCounterFamily<L> {
137    fn default() -> Self {
138        let counters = (0..L::COUNT)
139            .map(|_| PaddedCounter::default())
140            .collect::<Vec<_>>()
141            .into_boxed_slice();
142        Self {
143            counters,
144            _label: PhantomData,
145        }
146    }
147}
148
149impl<L: MetricLabel> PaddedCounterFamily<L> {
150    pub(super) fn accumulate_into(&self, totals: &mut [u64]) {
151        for label in L::VARIANTS {
152            let count = self.load(*label);
153            if let Some(total) = totals.get_mut(label.as_index()) {
154                *total = total.saturating_add(count);
155            }
156        }
157    }
158
159    pub(super) fn increment(&self, label: L) {
160        if let Some(counter) = self.counters.get(label.as_index()) {
161            counter.increment();
162        }
163    }
164
165    pub(super) fn add(&self, label: L, value: usize) {
166        if let Some(counter) = self.counters.get(label.as_index()) {
167            counter.add(value);
168        }
169    }
170
171    pub(super) fn add_u64(&self, label: L, value: u64) {
172        if let Some(counter) = self.counters.get(label.as_index()) {
173            counter.add_u64(value);
174        }
175    }
176
177    pub(super) fn load(&self, label: L) -> u64 {
178        self.counters
179            .get(label.as_index())
180            .map_or(0, PaddedCounter::load)
181    }
182}
183
184#[derive(Debug)]
185pub(super) struct CounterFamily<L: MetricLabel> {
186    counters: Box<[Counter]>,
187    _label: PhantomData<L>,
188}
189
190impl<L: MetricLabel> Default for CounterFamily<L> {
191    fn default() -> Self {
192        let counters = (0..L::COUNT)
193            .map(|_| Counter::default())
194            .collect::<Vec<_>>()
195            .into_boxed_slice();
196        Self {
197            counters,
198            _label: PhantomData,
199        }
200    }
201}
202
203impl<L: MetricLabel> CounterFamily<L> {
204    pub(super) fn increment(&self, label: L) {
205        if let Some(counter) = self.counters.get(label.as_index()) {
206            counter.increment();
207        }
208    }
209
210    pub(super) fn add(&self, label: L, value: usize) {
211        if let Some(counter) = self.counters.get(label.as_index()) {
212            counter.add(value);
213        }
214    }
215
216    pub(super) fn add_u64(&self, label: L, value: u64) {
217        if let Some(counter) = self.counters.get(label.as_index()) {
218            counter.add_u64(value);
219        }
220    }
221
222    pub(super) fn load(&self, label: L) -> u64 {
223        self.counters.get(label.as_index()).map_or(0, Counter::load)
224    }
225}
226
227#[derive(Debug)]
228pub(super) struct Histogram<B: HistogramBucketLabel> {
229    buckets: Box<[Counter]>,
230    count: Counter,
231    sum_micros: Counter,
232    _bucket: PhantomData<B>,
233}
234
235impl<B: HistogramBucketLabel> Default for Histogram<B> {
236    fn default() -> Self {
237        let buckets = (0..B::COUNT)
238            .map(|_| Counter::default())
239            .collect::<Vec<_>>()
240            .into_boxed_slice();
241        Self {
242            buckets,
243            count: Counter::default(),
244            sum_micros: Counter::default(),
245            _bucket: PhantomData,
246        }
247    }
248}
249
250impl<B: HistogramBucketLabel> Histogram<B> {
251    /// records one cumulative Prometheus histogram observation
252    ///
253    /// every bucket at or above the selected bound is incremented so snapshots can
254    /// be rendered directly as `_bucket{le=...}` samples. Observations above every
255    /// finite bound still contribute to count, sum and the exported `+Inf` bucket.
256    pub(super) fn observe(&self, duration: Duration) {
257        self.count.increment();
258        self.sum_micros
259            .add_u64(u64::try_from(duration.as_micros()).unwrap_or(u64::MAX));
260        let Some(bucket) = B::from_duration(duration) else {
261            return;
262        };
263        for counter in self.buckets.iter().skip(bucket.as_index()) {
264            counter.increment();
265        }
266    }
267
268    pub(super) fn load_bucket(&self, bucket: B) -> u64 {
269        self.buckets.get(bucket.as_index()).map_or(0, Counter::load)
270    }
271
272    pub(super) fn load_count(&self) -> u64 {
273        self.count.load()
274    }
275
276    pub(super) fn load_sum_micros(&self) -> u64 {
277        self.sum_micros.load()
278    }
279}
280
281#[derive(Debug)]
282pub(super) struct HistogramFamily<L: MetricLabel, B: HistogramBucketLabel> {
283    histograms: Box<[Histogram<B>]>,
284    _label: PhantomData<L>,
285}
286
287impl<L: MetricLabel, B: HistogramBucketLabel> Default for HistogramFamily<L, B> {
288    fn default() -> Self {
289        let histograms = (0..L::COUNT)
290            .map(|_| Histogram::default())
291            .collect::<Vec<_>>()
292            .into_boxed_slice();
293        Self {
294            histograms,
295            _label: PhantomData,
296        }
297    }
298}
299
300impl<L: MetricLabel, B: HistogramBucketLabel> HistogramFamily<L, B> {
301    pub(super) fn observe(&self, label: L, duration: Duration) {
302        if let Some(histogram) = self.histograms.get(label.as_index()) {
303            histogram.observe(duration);
304        }
305    }
306
307    pub(super) fn load_bucket(&self, label: L, bucket: B) -> u64 {
308        self.histograms
309            .get(label.as_index())
310            .map_or(0, |histogram| histogram.load_bucket(bucket))
311    }
312
313    pub(super) fn load_count(&self, label: L) -> u64 {
314        self.histograms
315            .get(label.as_index())
316            .map_or(0, Histogram::load_count)
317    }
318
319    pub(super) fn load_sum_micros(&self, label: L) -> u64 {
320        self.histograms
321            .get(label.as_index())
322            .map_or(0, Histogram::load_sum_micros)
323    }
324}
o_sfu_telemetry/metrics/counter.rs

o_sfu_telemetry/metrics/
counter.rs
