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    fn from_duration(duration: Duration) -> Self;
30}
31
32#[repr(align(64))]
33#[derive(Debug, Default)]
34pub(super) struct PaddedCounter {
35    value: Counter,
36}
37
38impl PaddedCounter {
39    pub(super) fn increment(&self) {
40        self.value.increment();
41    }
42
43    pub(super) fn add(&self, value: usize) {
44        self.value.add(value);
45    }
46
47    pub(super) fn add_u64(&self, value: u64) {
48        self.value.add_u64(value);
49    }
50
51    pub(super) fn load(&self) -> u64 {
52        self.value.load()
53    }
54}
55
56#[derive(Debug, Default)]
57pub(super) struct Counter {
58    value: AtomicU64,
59}
60
61impl Counter {
62    pub(super) fn increment(&self) {
63        self.value.fetch_add(1, Ordering::Relaxed);
64    }
65
66    pub(super) fn add(&self, value: usize) {
67        if let Ok(value) = u64::try_from(value) {
68            self.value.fetch_add(value, Ordering::Relaxed);
69        }
70    }
71
72    pub(super) fn add_u64(&self, value: u64) {
73        self.value.fetch_add(value, Ordering::Relaxed);
74    }
75
76    pub(super) fn load(&self) -> u64 {
77        self.value.load(Ordering::Relaxed)
78    }
79}
80
81#[derive(Debug, Default)]
82pub(super) struct UpDownCounter {
83    value: AtomicI64,
84}
85
86impl UpDownCounter {
87    pub(super) fn add(&self, delta: i64) {
88        self.value.fetch_add(delta, Ordering::Relaxed);
89    }
90
91    pub(super) fn load(&self) -> i64 {
92        self.value.load(Ordering::Relaxed)
93    }
94}
95
96#[derive(Debug)]
97pub(super) struct UpDownCounterFamily<L: MetricLabel> {
98    counters: Box<[UpDownCounter]>,
99    _label: PhantomData<L>,
100}
101
102impl<L: MetricLabel> Default for UpDownCounterFamily<L> {
103    fn default() -> Self {
104        let counters = (0..L::COUNT)
105            .map(|_| UpDownCounter::default())
106            .collect::<Vec<_>>()
107            .into_boxed_slice();
108        Self {
109            counters,
110            _label: PhantomData,
111        }
112    }
113}
114
115impl<L: MetricLabel> UpDownCounterFamily<L> {
116    pub(super) fn add(&self, label: L, delta: i64) {
117        if let Some(counter) = self.counters.get(label.as_index()) {
118            counter.add(delta);
119        }
120    }
121
122    pub(super) fn load(&self, label: L) -> i64 {
123        self.counters
124            .get(label.as_index())
125            .map_or(0, UpDownCounter::load)
126    }
127}
128
129#[derive(Debug)]
130pub(super) struct PaddedCounterFamily<L: MetricLabel> {
131    counters: Box<[PaddedCounter]>,
132    _label: PhantomData<L>,
133}
134
135impl<L: MetricLabel> Default for PaddedCounterFamily<L> {
136    fn default() -> Self {
137        let counters = (0..L::COUNT)
138            .map(|_| PaddedCounter::default())
139            .collect::<Vec<_>>()
140            .into_boxed_slice();
141        Self {
142            counters,
143            _label: PhantomData,
144        }
145    }
146}
147
148impl<L: MetricLabel> PaddedCounterFamily<L> {
149    pub(super) fn increment(&self, label: L) {
150        if let Some(counter) = self.counters.get(label.as_index()) {
151            counter.increment();
152        }
153    }
154
155    pub(super) fn add(&self, label: L, value: usize) {
156        if let Some(counter) = self.counters.get(label.as_index()) {
157            counter.add(value);
158        }
159    }
160
161    pub(super) fn add_u64(&self, label: L, value: u64) {
162        if let Some(counter) = self.counters.get(label.as_index()) {
163            counter.add_u64(value);
164        }
165    }
166
167    pub(super) fn load(&self, label: L) -> u64 {
168        self.counters
169            .get(label.as_index())
170            .map_or(0, PaddedCounter::load)
171    }
172}
173
174#[derive(Debug)]
175pub(super) struct CounterFamily<L: MetricLabel> {
176    counters: Box<[Counter]>,
177    _label: PhantomData<L>,
178}
179
180impl<L: MetricLabel> Default for CounterFamily<L> {
181    fn default() -> Self {
182        let counters = (0..L::COUNT)
183            .map(|_| Counter::default())
184            .collect::<Vec<_>>()
185            .into_boxed_slice();
186        Self {
187            counters,
188            _label: PhantomData,
189        }
190    }
191}
192
193impl<L: MetricLabel> CounterFamily<L> {
194    pub(super) fn increment(&self, label: L) {
195        if let Some(counter) = self.counters.get(label.as_index()) {
196            counter.increment();
197        }
198    }
199
200    pub(super) fn add(&self, label: L, value: usize) {
201        if let Some(counter) = self.counters.get(label.as_index()) {
202            counter.add(value);
203        }
204    }
205
206    pub(super) fn add_u64(&self, label: L, value: u64) {
207        if let Some(counter) = self.counters.get(label.as_index()) {
208            counter.add_u64(value);
209        }
210    }
211
212    pub(super) fn load(&self, label: L) -> u64 {
213        self.counters.get(label.as_index()).map_or(0, Counter::load)
214    }
215}
216
217#[derive(Debug)]
218pub(super) struct Histogram<B: HistogramBucketLabel> {
219    buckets: Box<[Counter]>,
220    count: Counter,
221    sum_micros: Counter,
222    _bucket: PhantomData<B>,
223}
224
225impl<B: HistogramBucketLabel> Default for Histogram<B> {
226    fn default() -> Self {
227        let buckets = (0..B::COUNT)
228            .map(|_| Counter::default())
229            .collect::<Vec<_>>()
230            .into_boxed_slice();
231        Self {
232            buckets,
233            count: Counter::default(),
234            sum_micros: Counter::default(),
235            _bucket: PhantomData,
236        }
237    }
238}
239
240impl<B: HistogramBucketLabel> Histogram<B> {
241    /// records one cumulative Prometheus histogram observation
242    ///
243    /// every bucket at or above the selected bound is incremented so snapshots can
244    /// be rendered directly as `_bucket{le=...}` samples
245    pub(super) fn observe(&self, duration: Duration) {
246        self.count.increment();
247        self.sum_micros
248            .add_u64(u64::try_from(duration.as_micros()).unwrap_or(u64::MAX));
249        let bucket_index = B::from_duration(duration).as_index();
250        for counter in self.buckets.iter().skip(bucket_index) {
251            counter.increment();
252        }
253    }
254
255    pub(super) fn load_bucket(&self, bucket: B) -> u64 {
256        self.buckets.get(bucket.as_index()).map_or(0, Counter::load)
257    }
258
259    pub(super) fn load_count(&self) -> u64 {
260        self.count.load()
261    }
262
263    pub(super) fn load_sum_micros(&self) -> u64 {
264        self.sum_micros.load()
265    }
266}
267
268#[derive(Debug)]
269pub(super) struct HistogramFamily<L: MetricLabel, B: HistogramBucketLabel> {
270    histograms: Box<[Histogram<B>]>,
271    _label: PhantomData<L>,
272}
273
274impl<L: MetricLabel, B: HistogramBucketLabel> Default for HistogramFamily<L, B> {
275    fn default() -> Self {
276        let histograms = (0..L::COUNT)
277            .map(|_| Histogram::default())
278            .collect::<Vec<_>>()
279            .into_boxed_slice();
280        Self {
281            histograms,
282            _label: PhantomData,
283        }
284    }
285}
286
287impl<L: MetricLabel, B: HistogramBucketLabel> HistogramFamily<L, B> {
288    pub(super) fn observe(&self, label: L, duration: Duration) {
289        if let Some(histogram) = self.histograms.get(label.as_index()) {
290            histogram.observe(duration);
291        }
292    }
293
294    pub(super) fn load_bucket(&self, label: L, bucket: B) -> u64 {
295        self.histograms
296            .get(label.as_index())
297            .map_or(0, |histogram| histogram.load_bucket(bucket))
298    }
299
300    pub(super) fn load_count(&self, label: L) -> u64 {
301        self.histograms
302            .get(label.as_index())
303            .map_or(0, Histogram::load_count)
304    }
305
306    pub(super) fn load_sum_micros(&self, label: L) -> u64 {
307        self.histograms
308            .get(label.as_index())
309            .map_or(0, Histogram::load_sum_micros)
310    }
311}
o_sfu_telemetry/metrics/counter.rs

o_sfu_telemetry/metrics/
counter.rs
