fast_telemetry/metric/
counter.rs

1//! Thread-sharded atomic counter.
2//!
3//! Forked from https://crates.io/crates/fast-counter (MIT/Apache licensed)
4//! Originally authored by https://crates.io/users/JackThomson2
5//! Modified to use crossbeam's CachePadded for more correct cache line sizing,
6//! and to support swap operations and export operations.
7
8use crate::thread_id::thread_id;
9use crossbeam_utils::CachePadded;
10use std::fmt;
11use std::sync::atomic::{AtomicIsize, Ordering};
12
13fn make_padded_counter() -> CachePadded<AtomicIsize> {
14    CachePadded::new(AtomicIsize::new(0))
15}
16
17fn make_counter_cell() -> AtomicIsize {
18    AtomicIsize::new(0)
19}
20
21/// A sharded atomic counter.
22///
23/// Shards cache-line aligned AtomicIsize values across a vector for faster
24/// updates in high contention scenarios.
25pub struct Counter {
26    cells: Vec<CachePadded<AtomicIsize>>,
27}
28
29impl fmt::Debug for Counter {
30    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
31        f.debug_struct("Counter")
32            .field("sum", &self.sum())
33            .field("cells", &self.cells.len())
34            .finish()
35    }
36}
37
38impl Counter {
39    /// Creates a new Counter with at least `count` cells.
40    ///
41    /// The count is rounded up to the next power of two for fast modulo.
42    #[inline]
43    pub fn new(count: usize) -> Self {
44        let count = count.next_power_of_two();
45        Self {
46            cells: (0..count).map(|_| make_padded_counter()).collect(),
47        }
48    }
49
50    /// Adds a value to the counter using relaxed ordering.
51    #[inline]
52    pub fn add(&self, value: isize) {
53        self.add_with_ordering(value, Ordering::Relaxed)
54    }
55
56    /// Increments the counter by 1.
57    #[inline]
58    pub fn inc(&self) {
59        self.add(1)
60    }
61
62    #[inline]
63    fn add_with_thread_id(&self, thread_id: usize, value: isize, ordering: Ordering) {
64        let idx = thread_id & (self.cells.len() - 1);
65        // SAFETY: idx is always < cells.len() due to power-of-two masking
66        let cell = if cfg!(debug_assertions) {
67            self.cells.get(idx).expect("index out of bounds")
68        } else {
69            unsafe { self.cells.get_unchecked(idx) }
70        };
71        cell.fetch_add(value, ordering);
72    }
73
74    /// Adds a value to the counter with the specified ordering.
75    #[inline]
76    pub fn add_with_ordering(&self, value: isize, ordering: Ordering) {
77        self.add_with_thread_id(thread_id(), value, ordering);
78    }
79
80    /// Returns the sum of all shards using relaxed ordering.
81    ///
82    /// # Eventual Consistency
83    ///
84    /// Due to sharding, this may be slightly inaccurate under heavy concurrent
85    /// modification - writes to already-summed shards won't be reflected until
86    /// the next call. The total is eventually consistent.
87    #[inline]
88    pub fn sum(&self) -> isize {
89        self.sum_with_ordering(Ordering::Relaxed)
90    }
91
92    /// Returns the sum of all shards with the specified ordering.
93    #[inline]
94    pub fn sum_with_ordering(&self, ordering: Ordering) -> isize {
95        self.cells.iter().map(|c| c.load(ordering)).sum()
96    }
97
98    /// Resets all shards to zero and returns the previous sum.
99    ///
100    /// Useful for delta-style metrics export.
101    ///
102    /// # Eventual Consistency
103    ///
104    /// Writes that occur concurrently with `swap()` may be attributed to the
105    /// next window rather than the current one. This is because shards are
106    /// swapped sequentially - a write landing on an already-swapped shard
107    /// will be picked up by the next `swap()` call. No counts are lost; they
108    /// simply shift to the next export window. For telemetry purposes with
109    /// multi-second export intervals, this timing skew is negligible.
110    #[inline]
111    pub fn swap(&self) -> isize {
112        self.cells
113            .iter()
114            .map(|c| c.swap(0, Ordering::Relaxed))
115            .sum()
116    }
117}
118
119/// A grouped set of related sharded counters.
120///
121/// Use `CounterSet` when one hot-path operation commonly updates several
122/// counters together, such as request count, bytes processed, and cache hits.
123/// The counters share the same shard row, so updates by the same thread touch a
124/// compact row while rows for different shards remain padded apart to avoid
125/// false sharing.
126///
127/// `CounterSet` still supports individual counter updates by index. Resolve
128/// those indexes once during construction and keep direct handles/indexes on the
129/// hot path.
130///
131/// ```
132/// use fast_telemetry::CounterSet;
133///
134/// const REQUESTS: usize = 0;
135/// const BYTES: usize = 1;
136/// const HITS: usize = 2;
137///
138/// let counters = CounterSet::new(4, 3);
139/// counters.inc(REQUESTS);
140/// counters.add(BYTES, 4096);
141/// counters.inc(HITS);
142///
143/// assert_eq!(counters.sum(REQUESTS), 1);
144/// assert_eq!(counters.sum(BYTES), 4096);
145/// assert_eq!(counters.sum(HITS), 1);
146/// ```
147pub struct CounterSet {
148    cells: Vec<AtomicIsize>,
149    counters: usize,
150    stride: usize,
151    shard_mask: usize,
152}
153
154impl CounterSet {
155    /// Creates a grouped counter set with `counters` counters per shard.
156    ///
157    /// Shards are padded by row instead of by cell, so related counters updated
158    /// by the same thread sit contiguously while adjacent shard rows remain
159    /// separated enough to avoid false sharing.
160    pub fn new(shards: usize, counters: usize) -> Self {
161        assert!(counters >= 1, "counters must be >= 1");
162        let shards = shards.next_power_of_two();
163        let cells_per_padded_counter =
164            std::mem::size_of::<CachePadded<AtomicIsize>>() / std::mem::size_of::<AtomicIsize>();
165        let row_padding = cells_per_padded_counter.max(1);
166        let stride = counters.div_ceil(row_padding) * row_padding;
167        let cells = (0..(shards * stride))
168            .map(|_| make_counter_cell())
169            .collect();
170
171        Self {
172            cells,
173            counters,
174            stride,
175            shard_mask: shards - 1,
176        }
177    }
178
179    /// Returns the number of counters in each shard row.
180    #[inline]
181    pub fn len(&self) -> usize {
182        self.counters
183    }
184
185    /// Returns true if there are no counters in the set.
186    #[inline]
187    pub fn is_empty(&self) -> bool {
188        self.counters == 0
189    }
190
191    #[inline]
192    fn current_shard_offset(&self) -> usize {
193        (thread_id() & self.shard_mask) * self.stride
194    }
195
196    #[inline]
197    fn cell_at(&self, index: usize) -> &AtomicIsize {
198        if cfg!(debug_assertions) {
199            self.cells.get(index).expect("index out of bounds")
200        } else {
201            // SAFETY: callers compute indexes from checked counter indexes and
202            // row offsets derived from the shard mask.
203            unsafe { self.cells.get_unchecked(index) }
204        }
205    }
206
207    /// Increments one counter in the current thread's shard row.
208    #[inline]
209    pub fn inc(&self, counter_idx: usize) {
210        self.add(counter_idx, 1);
211    }
212
213    /// Adds a value to one counter in the current thread's shard row.
214    #[inline]
215    pub fn add(&self, counter_idx: usize, value: isize) {
216        assert!(counter_idx < self.counters, "counter index out of bounds");
217        let offset = self.current_shard_offset();
218        self.cell_at(offset + counter_idx)
219            .fetch_add(value, Ordering::Relaxed);
220    }
221
222    /// Increments all counters in the current thread's shard row.
223    #[inline]
224    pub fn inc_all(&self) {
225        self.add_all(1);
226    }
227
228    /// Adds the same value to all counters in the current thread's shard row.
229    #[inline(always)]
230    pub fn add_all(&self, value: isize) {
231        let offset = self.current_shard_offset();
232        let row = if cfg!(debug_assertions) {
233            self.cells
234                .get(offset..offset + self.counters)
235                .expect("row index out of bounds")
236        } else {
237            // SAFETY: current_shard_offset derives from shard_mask and stride,
238            // and counters <= stride by construction.
239            unsafe { std::slice::from_raw_parts(self.cells.as_ptr().add(offset), self.counters) }
240        };
241        for cell in row {
242            cell.fetch_add(value, Ordering::Relaxed);
243        }
244    }
245
246    /// Adds the same value to selected counters in the current thread's shard row.
247    #[inline]
248    pub fn add_indices(&self, indexes: &[usize], value: isize) {
249        let offset = self.current_shard_offset();
250        for index in indexes {
251            assert!(*index < self.counters, "counter index out of bounds");
252            self.cell_at(offset + *index)
253                .fetch_add(value, Ordering::Relaxed);
254        }
255    }
256
257    /// Adds one value per selected counter in the current thread's shard row.
258    #[inline]
259    pub fn add_index_values(&self, updates: &[(usize, isize)]) {
260        let offset = self.current_shard_offset();
261        for (index, value) in updates {
262            assert!(*index < self.counters, "counter index out of bounds");
263            self.cell_at(offset + *index)
264                .fetch_add(*value, Ordering::Relaxed);
265        }
266    }
267
268    /// Adds the same value to all counters in the current thread's shard row.
269    #[inline]
270    pub fn add_all_indexed(&self, value: isize) {
271        let offset = self.current_shard_offset();
272        for counter_idx in 0..self.counters {
273            self.cell_at(offset + counter_idx)
274                .fetch_add(value, Ordering::Relaxed);
275        }
276    }
277
278    /// Adds one value per counter in the current thread's shard row.
279    #[inline(always)]
280    pub fn add_values(&self, values: &[isize]) {
281        assert_eq!(values.len(), self.counters, "values must match counters");
282        let offset = self.current_shard_offset();
283        match values {
284            [a] => {
285                self.cell_at(offset).fetch_add(*a, Ordering::Relaxed);
286            }
287            [a, b] => {
288                self.cell_at(offset).fetch_add(*a, Ordering::Relaxed);
289                self.cell_at(offset + 1).fetch_add(*b, Ordering::Relaxed);
290            }
291            [a, b, c] => {
292                self.cell_at(offset).fetch_add(*a, Ordering::Relaxed);
293                self.cell_at(offset + 1).fetch_add(*b, Ordering::Relaxed);
294                self.cell_at(offset + 2).fetch_add(*c, Ordering::Relaxed);
295            }
296            [a, b, c, d] => {
297                self.cell_at(offset).fetch_add(*a, Ordering::Relaxed);
298                self.cell_at(offset + 1).fetch_add(*b, Ordering::Relaxed);
299                self.cell_at(offset + 2).fetch_add(*c, Ordering::Relaxed);
300                self.cell_at(offset + 3).fetch_add(*d, Ordering::Relaxed);
301            }
302            [a, b, c, d, e] => {
303                self.cell_at(offset).fetch_add(*a, Ordering::Relaxed);
304                self.cell_at(offset + 1).fetch_add(*b, Ordering::Relaxed);
305                self.cell_at(offset + 2).fetch_add(*c, Ordering::Relaxed);
306                self.cell_at(offset + 3).fetch_add(*d, Ordering::Relaxed);
307                self.cell_at(offset + 4).fetch_add(*e, Ordering::Relaxed);
308            }
309            [a, b, c, d, e, f] => {
310                self.cell_at(offset).fetch_add(*a, Ordering::Relaxed);
311                self.cell_at(offset + 1).fetch_add(*b, Ordering::Relaxed);
312                self.cell_at(offset + 2).fetch_add(*c, Ordering::Relaxed);
313                self.cell_at(offset + 3).fetch_add(*d, Ordering::Relaxed);
314                self.cell_at(offset + 4).fetch_add(*e, Ordering::Relaxed);
315                self.cell_at(offset + 5).fetch_add(*f, Ordering::Relaxed);
316            }
317            [a, b, c, d, e, f, g] => {
318                self.cell_at(offset).fetch_add(*a, Ordering::Relaxed);
319                self.cell_at(offset + 1).fetch_add(*b, Ordering::Relaxed);
320                self.cell_at(offset + 2).fetch_add(*c, Ordering::Relaxed);
321                self.cell_at(offset + 3).fetch_add(*d, Ordering::Relaxed);
322                self.cell_at(offset + 4).fetch_add(*e, Ordering::Relaxed);
323                self.cell_at(offset + 5).fetch_add(*f, Ordering::Relaxed);
324                self.cell_at(offset + 6).fetch_add(*g, Ordering::Relaxed);
325            }
326            [a, b, c, d, e, f, g, h] => {
327                self.cell_at(offset).fetch_add(*a, Ordering::Relaxed);
328                self.cell_at(offset + 1).fetch_add(*b, Ordering::Relaxed);
329                self.cell_at(offset + 2).fetch_add(*c, Ordering::Relaxed);
330                self.cell_at(offset + 3).fetch_add(*d, Ordering::Relaxed);
331                self.cell_at(offset + 4).fetch_add(*e, Ordering::Relaxed);
332                self.cell_at(offset + 5).fetch_add(*f, Ordering::Relaxed);
333                self.cell_at(offset + 6).fetch_add(*g, Ordering::Relaxed);
334                self.cell_at(offset + 7).fetch_add(*h, Ordering::Relaxed);
335            }
336            values => {
337                let row = if cfg!(debug_assertions) {
338                    self.cells
339                        .get(offset..offset + self.counters)
340                        .expect("row index out of bounds")
341                } else {
342                    // SAFETY: current_shard_offset derives from shard_mask and
343                    // stride, and counters <= stride by construction.
344                    unsafe {
345                        std::slice::from_raw_parts(self.cells.as_ptr().add(offset), self.counters)
346                    }
347                };
348                for (cell, value) in row.iter().zip(values.iter().copied()) {
349                    cell.fetch_add(value, Ordering::Relaxed);
350                }
351            }
352        }
353    }
354
355    /// Returns the sum for one counter across all shards.
356    #[inline]
357    pub fn sum(&self, counter_idx: usize) -> isize {
358        assert!(counter_idx < self.counters, "counter index out of bounds");
359        let shards = self.cells.len() / self.stride;
360        (0..shards)
361            .map(|shard| {
362                self.cell_at((shard * self.stride) + counter_idx)
363                    .load(Ordering::Relaxed)
364            })
365            .sum()
366    }
367
368    /// Returns the sum of all counters across all shards.
369    #[inline]
370    pub fn sum_all(&self) -> isize {
371        let shards = self.cells.len() / self.stride;
372        let mut total = 0isize;
373        for shard in 0..shards {
374            let offset = shard * self.stride;
375            for counter_idx in 0..self.counters {
376                total += self.cell_at(offset + counter_idx).load(Ordering::Relaxed);
377            }
378        }
379        total
380    }
381
382    /// Returns one value per counter in index order.
383    #[inline]
384    pub fn snapshot(&self) -> Vec<isize> {
385        (0..self.counters).map(|idx| self.sum(idx)).collect()
386    }
387
388    /// Resets one counter to zero and returns its previous sum.
389    ///
390    /// # Eventual Consistency
391    ///
392    /// Writes that occur concurrently with `sum_and_reset()` may be attributed
393    /// to the next window rather than the current one. No counts are lost; they
394    /// simply shift to the next export window.
395    #[inline]
396    pub fn sum_and_reset(&self, counter_idx: usize) -> isize {
397        assert!(counter_idx < self.counters, "counter index out of bounds");
398        let shards = self.cells.len() / self.stride;
399        (0..shards)
400            .map(|shard| {
401                self.cell_at((shard * self.stride) + counter_idx)
402                    .swap(0, Ordering::Relaxed)
403            })
404            .sum()
405    }
406
407    /// Returns one value per counter in index order, then resets all counters.
408    ///
409    /// # Eventual Consistency
410    ///
411    /// Writes that occur concurrently with `snapshot_and_reset()` may be
412    /// attributed to the next window rather than the current one. No counts are
413    /// lost; they simply shift to the next export window.
414    #[inline]
415    pub fn snapshot_and_reset(&self) -> Vec<isize> {
416        let mut values = vec![0; self.counters];
417        let shards = self.cells.len() / self.stride;
418        for shard in 0..shards {
419            let offset = shard * self.stride;
420            for (counter_idx, value) in values.iter_mut().enumerate() {
421                *value += self
422                    .cell_at(offset + counter_idx)
423                    .swap(0, Ordering::Relaxed);
424            }
425        }
426        values
427    }
428}
429
430/// A local write buffer for a [`CounterSet`].
431///
432/// Use one buffer per worker/thread when a hot path records several related
433/// counters per logical operation. The buffer accumulates deltas locally and
434/// flushes them to the backing `CounterSet` every `flush_every` completed
435/// operations, reducing shared atomic writes on very hot paths.
436///
437/// Individual updates are supported with [`CounterSetBuffer::inc`] and
438/// [`CounterSetBuffer::add`]. Call [`CounterSetBuffer::finish_op`] once after
439/// recording all updates for a logical operation. Full-group updates such as
440/// [`CounterSetBuffer::inc_all`] and [`CounterSetBuffer::add_values`] already
441/// finish the operation.
442///
443/// The buffer also flushes on drop, but long-lived services should still flush
444/// at request or task boundaries when they need prompt visibility to exporters.
445///
446/// ```
447/// use fast_telemetry::{CounterSet, CounterSetBuffer};
448///
449/// const REQUESTS: usize = 0;
450/// const BYTES: usize = 1;
451/// const ERRORS: usize = 2;
452///
453/// let counters = CounterSet::new(4, 3);
454/// let mut buffer = CounterSetBuffer::new(&counters, 64);
455///
456/// buffer.inc(REQUESTS);
457/// buffer.add(BYTES, 4096);
458/// buffer.finish_op();
459///
460/// buffer.inc(REQUESTS);
461/// buffer.inc(ERRORS);
462/// buffer.finish_op();
463///
464/// buffer.flush();
465///
466/// assert_eq!(counters.sum(REQUESTS), 2);
467/// assert_eq!(counters.sum(BYTES), 4096);
468/// assert_eq!(counters.sum(ERRORS), 1);
469/// ```
470pub struct CounterSetBuffer<'a> {
471    counters: &'a CounterSet,
472    deltas: Vec<isize>,
473    all_delta: isize,
474    ops_since_flush: usize,
475    flush_every: usize,
476    op_dirty: bool,
477}
478
479impl<'a> CounterSetBuffer<'a> {
480    /// Creates a local buffer for a grouped counter set.
481    #[inline]
482    pub fn new(counters: &'a CounterSet, flush_every: usize) -> Self {
483        assert!(flush_every >= 1, "flush_every must be >= 1");
484        Self {
485            counters,
486            deltas: vec![0; counters.len()],
487            all_delta: 0,
488            ops_since_flush: 0,
489            flush_every,
490            op_dirty: false,
491        }
492    }
493
494    /// Increments one buffered counter.
495    #[inline]
496    pub fn inc(&mut self, counter_idx: usize) {
497        self.add(counter_idx, 1);
498    }
499
500    /// Adds a value to one buffered counter.
501    #[inline]
502    pub fn add(&mut self, counter_idx: usize, value: isize) {
503        assert!(
504            counter_idx < self.deltas.len(),
505            "counter index out of bounds"
506        );
507        self.deltas[counter_idx] += value;
508        self.op_dirty = true;
509    }
510
511    /// Increments every buffered counter and marks one logical operation complete.
512    #[inline(always)]
513    pub fn inc_all(&mut self) {
514        self.all_delta += 1;
515        self.finish_group_op();
516    }
517
518    /// Adds the same value to every buffered counter and marks one logical operation complete.
519    #[inline(always)]
520    pub fn add_all(&mut self, value: isize) {
521        self.all_delta += value;
522        self.finish_group_op();
523    }
524
525    /// Adds one value per buffered counter and marks one logical operation complete.
526    #[inline]
527    pub fn add_values(&mut self, values: &[isize]) {
528        assert_eq!(
529            values.len(),
530            self.deltas.len(),
531            "values must match counters"
532        );
533        for (delta, value) in self.deltas.iter_mut().zip(values.iter().copied()) {
534            *delta += value;
535        }
536        self.finish_group_op();
537    }
538
539    /// Marks the current logical operation complete.
540    #[inline]
541    pub fn finish_op(&mut self) {
542        if !self.op_dirty {
543            return;
544        }
545
546        self.op_dirty = false;
547        self.finish_group_op();
548    }
549
550    #[inline(always)]
551    fn finish_group_op(&mut self) {
552        self.ops_since_flush += 1;
553        if self.ops_since_flush >= self.flush_every {
554            self.flush();
555        }
556    }
557
558    /// Flushes buffered deltas to the backing grouped counter set.
559    #[inline]
560    pub fn flush(&mut self) {
561        if self.ops_since_flush == 0 && !self.op_dirty {
562            return;
563        }
564
565        if self.all_delta != 0 {
566            self.counters.add_all(self.all_delta);
567            self.all_delta = 0;
568        }
569        if self.deltas.iter().any(|delta| *delta != 0) {
570            self.counters.add_values(&self.deltas);
571            self.deltas.fill(0);
572        }
573        self.ops_since_flush = 0;
574        self.op_dirty = false;
575    }
576}
577
578impl Drop for CounterSetBuffer<'_> {
579    fn drop(&mut self) {
580        self.flush();
581    }
582}
583
584#[cfg(test)]
585mod tests {
586    use super::*;
587
588    #[test]
589    fn basic_test() {
590        let counter = Counter::new(1);
591        counter.add(1);
592        assert_eq!(counter.sum(), 1);
593    }
594
595    #[test]
596    fn increment_multiple_times() {
597        let counter = Counter::new(1);
598        counter.add(1);
599        counter.add(1);
600        counter.add(1);
601        assert_eq!(counter.sum(), 3);
602    }
603
604    #[test]
605    fn test_inc() {
606        let counter = Counter::new(4);
607        counter.inc();
608        counter.inc();
609        assert_eq!(counter.sum(), 2);
610    }
611
612    #[test]
613    fn test_swap() {
614        let counter = Counter::new(4);
615        counter.add(100);
616        let val = counter.swap();
617        assert_eq!(val, 100);
618        assert_eq!(counter.sum(), 0);
619    }
620
621    #[test]
622    fn two_threads_incrementing_concurrently() {
623        let counter = Counter::new(2);
624
625        std::thread::scope(|s| {
626            for _ in 0..2 {
627                s.spawn(|| {
628                    counter.add(1);
629                });
630            }
631        });
632
633        assert_eq!(counter.sum(), 2);
634    }
635
636    #[test]
637    fn multiple_threads_incrementing_many_times() {
638        const WRITE_COUNT: isize = 1_000_000;
639        const THREAD_COUNT: isize = 8;
640
641        let counter = Counter::new(THREAD_COUNT as usize);
642
643        std::thread::scope(|s| {
644            for _ in 0..THREAD_COUNT {
645                s.spawn(|| {
646                    for _ in 0..WRITE_COUNT {
647                        counter.add(1);
648                    }
649                });
650            }
651        });
652
653        assert_eq!(counter.sum(), THREAD_COUNT * WRITE_COUNT);
654    }
655
656    #[test]
657    fn debug_format() {
658        let counter = Counter::new(8);
659        counter.add(42);
660        let debug = format!("{counter:?}");
661        assert!(debug.contains("sum: 42"));
662        assert!(debug.contains("cells: 8"));
663    }
664
665    #[test]
666    fn counter_set_updates_grouped_counters() {
667        let counters = CounterSet::new(4, 3);
668
669        counters.inc(0);
670        counters.add(1, 2);
671        counters.inc_all();
672        counters.add_values(&[2, 3, 4]);
673        counters.add_indices(&[0, 2], 1);
674        counters.add_index_values(&[(1, 2), (2, 3)]);
675
676        assert_eq!(counters.len(), 3);
677        assert!(!counters.is_empty());
678        assert_eq!(counters.sum(0), 5);
679        assert_eq!(counters.sum(1), 8);
680        assert_eq!(counters.sum(2), 9);
681        assert_eq!(counters.sum_all(), 22);
682        assert_eq!(counters.snapshot(), vec![5, 8, 9]);
683    }
684
685    #[test]
686    fn counter_set_sum_and_reset_resets_one_counter() {
687        let counters = CounterSet::new(4, 3);
688
689        counters.add_values(&[2, 3, 4]);
690        counters.add_values(&[5, 6, 7]);
691
692        assert_eq!(counters.sum_and_reset(1), 9);
693        assert_eq!(counters.snapshot(), vec![7, 0, 11]);
694    }
695
696    #[test]
697    fn counter_set_snapshot_and_reset_resets_all_counters() {
698        let counters = CounterSet::new(4, 3);
699
700        counters.add_values(&[2, 3, 4]);
701        counters.add_values(&[5, 6, 7]);
702
703        assert_eq!(counters.snapshot_and_reset(), vec![7, 9, 11]);
704        assert_eq!(counters.snapshot(), vec![0, 0, 0]);
705    }
706
707    #[test]
708    fn counter_set_buffer_flushes_grouped_and_individual_updates() {
709        let counters = CounterSet::new(4, 3);
710
711        {
712            let mut buffer = CounterSetBuffer::new(&counters, 2);
713
714            buffer.inc(0);
715            buffer.add(1, 2);
716            buffer.finish_op();
717            assert_eq!(counters.sum_all(), 0);
718
719            buffer.inc_all();
720            assert_eq!(counters.sum(0), 2);
721            assert_eq!(counters.sum(1), 3);
722            assert_eq!(counters.sum(2), 1);
723
724            buffer.add(2, 4);
725            buffer.finish_op();
726            buffer.flush();
727        }
728
729        assert_eq!(counters.sum(0), 2);
730        assert_eq!(counters.sum(1), 3);
731        assert_eq!(counters.sum(2), 5);
732        assert_eq!(counters.sum_all(), 10);
733    }
734
735    #[test]
736    fn counter_set_buffer_flushes_on_drop() {
737        let counters = CounterSet::new(4, 2);
738
739        {
740            let mut buffer = CounterSetBuffer::new(&counters, 64);
741            buffer.inc(0);
742            buffer.add(1, 5);
743            buffer.finish_op();
744        }
745
746        assert_eq!(counters.sum(0), 1);
747        assert_eq!(counters.sum(1), 5);
748    }
749}
fast_telemetry/metric/counter.rs

fast_telemetry/metric/
counter.rs
