moq_uring/metrics.rs
1//! Per-worker counters: buffer-pool health, batch effectiveness, ring traffic,
2//! and scheduling.
3//!
4//! A worker is a thread that never yields to anything an ops surface can see,
5//! so these are how its health leaves the thread. Every write is a relaxed
6//! atomic add on the worker's own thread and every read is a relaxed load from
7//! whichever thread scrapes, which makes a [`Snapshot`] a cheap, slightly
8//! skewed reading rather than a consistent instant. Rates and ratios are what
9//! these are for; two counters in one snapshot may be a few operations apart.
10
11use std::sync::Arc;
12use std::sync::atomic::{AtomicU64, Ordering};
13
14/// One cumulative counter.
15#[derive(Default)]
16pub(crate) struct Counter(AtomicU64);
17
18impl Counter {
19 pub(crate) fn add(&self, count: u64) {
20 // Relaxed: the writer is one thread and readers want a magnitude, not
21 // an ordering against the operation being counted.
22 self.0.fetch_add(count, Ordering::Relaxed);
23 }
24
25 fn get(&self) -> u64 {
26 self.0.load(Ordering::Relaxed)
27 }
28}
29
30/// The counters themselves, shared by the worker, its sockets, its timer heap,
31/// and its park word.
32#[derive(Default)]
33pub(crate) struct Counters {
34 pub rx_datagrams: Counter,
35 pub rx_receives: Counter,
36 pub rx_enobufs: Counter,
37 pub rx_exhausted: Counter,
38 pub tx_datagrams: Counter,
39 pub tx_sends: Counter,
40 pub tx_stalls: Counter,
41 pub submissions: Counter,
42 pub completions: Counter,
43 pub enters: Counter,
44 pub parks: Counter,
45 pub wakes: Counter,
46 pub timers_armed: Counter,
47 pub timers_fired: Counter,
48 pub timers_cancelled: Counter,
49}
50
51/// A worker's counters, readable from any thread.
52///
53/// Hand one to [`crate::Config::metrics`] to keep a copy the process can scrape
54/// while the worker runs, or take the worker's own through
55/// [`crate::Handle::metrics`]. Clones share one set of counters, so give each
56/// worker its own.
57#[derive(Clone, Default)]
58pub struct Metrics(Arc<Counters>);
59
60impl Metrics {
61 pub(crate) fn counters(&self) -> &Arc<Counters> {
62 &self.0
63 }
64
65 pub(crate) fn from_counters(counters: Arc<Counters>) -> Self {
66 Self(counters)
67 }
68
69 /// Read every counter.
70 pub fn snapshot(&self) -> Snapshot {
71 Snapshot {
72 rx_datagrams: self.0.rx_datagrams.get(),
73 rx_receives: self.0.rx_receives.get(),
74 rx_enobufs: self.0.rx_enobufs.get(),
75 rx_exhausted: self.0.rx_exhausted.get(),
76 tx_datagrams: self.0.tx_datagrams.get(),
77 tx_sends: self.0.tx_sends.get(),
78 tx_stalls: self.0.tx_stalls.get(),
79 submissions: self.0.submissions.get(),
80 completions: self.0.completions.get(),
81 enters: self.0.enters.get(),
82 parks: self.0.parks.get(),
83 wakes: self.0.wakes.get(),
84 timers_armed: self.0.timers_armed.get(),
85 timers_fired: self.0.timers_fired.get(),
86 timers_cancelled: self.0.timers_cancelled.get(),
87 }
88 }
89}
90
91impl std::fmt::Debug for Metrics {
92 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
93 self.snapshot().fmt(f)
94 }
95}
96
97/// A reading of one worker's counters, all cumulative since it started.
98#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
99#[non_exhaustive]
100pub struct Snapshot {
101 /// UDP datagrams received, counting each `UDP_GRO` coalesced segment.
102 pub rx_datagrams: u64,
103 /// Receive completions that delivered datagrams. `rx_datagrams` over this
104 /// is the GRO coalescing actually achieved.
105 pub rx_receives: u64,
106 /// Receives the kernel ended with `ENOBUFS`: the provided-buffer ring was
107 /// empty, so no buffer could be selected and the receive was never
108 /// performed. The datagram stays in the socket queue, so this is
109 /// receive-side backpressure rather than a confirmed loss; sustained, it
110 /// becomes one, once the socket buffer fills. The first thing to look at
111 /// when throughput sags.
112 pub rx_enobufs: u64,
113 /// Re-arms that found no free receive buffer at all, so the socket was left
114 /// unarmed until a packet released one. The pool is at its ceiling and
115 /// every buffer is held by an unread packet.
116 pub rx_exhausted: u64,
117 /// UDP datagrams sent, counting each `UDP_SEGMENT` segment of a GSO train.
118 pub tx_datagrams: u64,
119 /// `sendmsg` operations staged. `tx_datagrams` over this is the GSO
120 /// batching actually achieved.
121 pub tx_sends: u64,
122 /// Times the send-buffer pool became drained at its ceiling and an
123 /// acquisition had to wait. Send-side backpressure.
124 pub tx_stalls: u64,
125 /// Submission queue entries the kernel accepted.
126 pub submissions: u64,
127 /// Completion queue entries dispatched.
128 pub completions: u64,
129 /// `io_uring_enter` calls. Datagrams over this is the syscall amortization
130 /// the runtime exists for.
131 pub enters: u64,
132 /// Times the worker parked in `io_uring_enter` with nothing left to poll.
133 pub parks: u64,
134 /// `futex` wakes another thread had to issue because the worker was parked.
135 pub wakes: u64,
136 /// Timers armed, re-arms included (a re-arm is a cancel plus an arm).
137 pub timers_armed: u64,
138 /// Timers that reached their deadline.
139 pub timers_fired: u64,
140 /// Timers dropped or re-armed before their deadline.
141 pub timers_cancelled: u64,
142}
143
144impl Snapshot {
145 /// Timers currently in the heap: armed, less those fired and cancelled.
146 pub fn timers_active(&self) -> u64 {
147 self.timers_armed
148 .saturating_sub(self.timers_fired)
149 .saturating_sub(self.timers_cancelled)
150 }
151}