compio_pool/
metrics.rs

1//! Pool instrumentation.
2
3use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
4
5/// A point-in-time snapshot of pool activity, from [`Pool::metrics`](crate::Pool::metrics).
6///
7/// Gauges (`live`, `idle`, `parked`) are summed across threads without a lock,
8/// so they can be momentarily inconsistent with each other. Counters are
9/// monotonic.
10#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
11pub struct Metrics {
12    /// Connections currently owned by a shard, idle or checked out.
13    pub live: u64,
14    /// Connections sitting in a shard free list.
15    pub idle: u64,
16    /// Connections sitting in the cross-thread reservoir.
17    pub parked: u64,
18    /// Connections opened since the pool was created.
19    pub created: u64,
20    /// Connections closed since the pool was created.
21    pub closed: u64,
22    /// Successful checkouts.
23    pub acquires: u64,
24    /// Checkouts that had to wait for a connection to come back.
25    pub waits: u64,
26    /// Checkouts that gave up at the acquire timeout.
27    pub timeouts: u64,
28    /// Connections discarded because a checkout was cancelled mid-operation.
29    pub poisoned: u64,
30    /// Connections discarded because `recycle` rejected them.
31    pub recycle_failures: u64,
32    /// Connections taken from the reservoir by a thread that did not open them.
33    pub unparked: u64,
34}
35
36#[derive(Debug, Default)]
37pub(crate) struct Counters {
38    pub live: AtomicU64,
39    pub idle: AtomicU64,
40    pub created: AtomicU64,
41    pub closed: AtomicU64,
42    pub acquires: AtomicU64,
43    pub waits: AtomicU64,
44    pub timeouts: AtomicU64,
45    pub poisoned: AtomicU64,
46    pub recycle_failures: AtomicU64,
47    pub unparked: AtomicU64,
48}
49
50impl Counters {
51    pub(crate) fn inc(c: &AtomicU64) {
52        c.fetch_add(1, Relaxed);
53    }
54
55    pub(crate) fn dec(c: &AtomicU64) {
56        c.fetch_sub(1, Relaxed);
57    }
58
59    pub(crate) fn snapshot(&self, parked: u64) -> Metrics {
60        Metrics {
61            live: self.live.load(Relaxed),
62            idle: self.idle.load(Relaxed),
63            parked,
64            created: self.created.load(Relaxed),
65            closed: self.closed.load(Relaxed),
66            acquires: self.acquires.load(Relaxed),
67            waits: self.waits.load(Relaxed),
68            timeouts: self.timeouts.load(Relaxed),
69            poisoned: self.poisoned.load(Relaxed),
70            recycle_failures: self.recycle_failures.load(Relaxed),
71            unparked: self.unparked.load(Relaxed),
72        }
73    }
74}
75
76#[cfg(test)]
77mod tests {
78    use super::*;
79
80    #[test]
81    fn counters_start_at_zero() {
82        assert_eq!(Counters::default().snapshot(0), Metrics::default());
83    }
84
85    #[test]
86    fn inc_and_dec_move_one_at_a_time() {
87        let c = Counters::default();
88        Counters::inc(&c.live);
89        Counters::inc(&c.live);
90        Counters::dec(&c.live);
91        assert_eq!(c.live.load(Relaxed), 1);
92    }
93
94    /// Each field must be read from its own atomic — a copy-paste slip in
95    /// `snapshot` would otherwise go unnoticed.
96    #[test]
97    fn snapshot_maps_every_field_to_its_own_counter() {
98        let c = Counters::default();
99        c.live.store(1, Relaxed);
100        c.idle.store(2, Relaxed);
101        c.created.store(3, Relaxed);
102        c.closed.store(4, Relaxed);
103        c.acquires.store(5, Relaxed);
104        c.waits.store(6, Relaxed);
105        c.timeouts.store(7, Relaxed);
106        c.poisoned.store(8, Relaxed);
107        c.recycle_failures.store(9, Relaxed);
108        c.unparked.store(10, Relaxed);
109
110        assert_eq!(
111            c.snapshot(11),
112            Metrics {
113                live: 1,
114                idle: 2,
115                parked: 11,
116                created: 3,
117                closed: 4,
118                acquires: 5,
119                waits: 6,
120                timeouts: 7,
121                poisoned: 8,
122                recycle_failures: 9,
123                unparked: 10,
124            }
125        );
126    }
127
128    /// `parked` lives in the exchange, not in `Counters`, so it is passed in
129    /// rather than loaded.
130    #[test]
131    fn parked_comes_from_the_argument() {
132        let c = Counters::default();
133        assert_eq!(c.snapshot(0).parked, 0);
134        assert_eq!(c.snapshot(42).parked, 42);
135    }
136
137    #[test]
138    fn metrics_is_copy_eq_and_debug() {
139        let m = Counters::default().snapshot(1);
140        let copy = m;
141        assert_eq!(m, copy);
142        assert_ne!(m, Metrics::default());
143        assert!(format!("{m:?}").contains("parked"));
144        assert!(format!("{:?}", Counters::default()).contains("live"));
145    }
146}
compio_pool/metrics.rs

compio_pool/
metrics.rs
