krabka_raft/
event_queue.rs

1//! The controller's event-queue measurements.
2//!
3//! These are Kafka's `ControllerEventManager` `EventQueueTimeMs`, `EventQueueProcessingTimeMs`
4//! and `AvgIdleRatio`, and `KafkaController`'s
5//! `EventQueueOperationsStartedCount` and `EventQueueOperationsTimedOutCount`.
6//!
7//! Kafka's `QuorumController` runs every operation as an event on one queue
8//! thread. Krabka's engine loop is that thread: each turn handles one command
9//! from the engine's mpsc or one timer, so a turn is one event. A command
10//! waited on the mpsc for its queue time; a timer has none, as a Kafka
11//! deferred event sets `DOES_NOT_UPDATE_QUEUE_TIME`. A write that parks for
12//! its commit keeps its processing time running until the commit reply, as a
13//! Kafka `ControllerWriteEvent` does.
14//!
15//! The engine reports to an [`EventQueueRecorder`]. The broker registers its
16//! metrics as one; [`NoEventQueueRecorder`] is the default.
17
18use {
19    std::sync::Arc,
20    tokio::time::{Duration, Instant},
21};
22
23/// Receives the engine loop's event-queue measurements. Every method is
24/// called from the engine task, so an implementation must not block.
25pub trait EventQueueRecorder: Send + Sync {
26    /// A command waited `waited` between its send and the turn that started
27    /// it: Kafka's `EventQueueTimeMs`.
28    fn queue_time(&self, waited: Duration);
29    /// An event ran for `took`, to its commit reply for a write: Kafka's
30    /// `EventQueueProcessingTimeMs`.
31    fn processing_time(&self, took: Duration);
32    /// The loop started an event: Kafka's `EventQueueOperationsStartedCount`.
33    fn operation_started(&self);
34    /// The loop reached a command whose caller had already stopped waiting
35    /// for the reply: Kafka's `EventQueueOperationsTimedOutCount`. Kafka drops
36    /// an event whose deadline passed in the queue; the engine still runs the
37    /// command, so it also counts as started.
38    fn operation_timed_out(&self);
39    /// The loop waited `idle` for its next event. Its rate is Kafka's
40    /// `AvgIdleRatio`.
41    fn idle_time(&self, idle: Duration);
42}
43
44/// The recorder of a controller nobody measures.
45#[derive(Debug, Clone, Copy, Default)]
46pub struct NoEventQueueRecorder;
47
48impl EventQueueRecorder for NoEventQueueRecorder {
49    fn queue_time(&self, _waited: Duration) {}
50
51    fn processing_time(&self, _took: Duration) {}
52
53    fn operation_started(&self) {}
54
55    fn operation_timed_out(&self) {}
56
57    fn idle_time(&self, _idle: Duration) {}
58}
59
60/// How a command reached the turn that runs it.
61#[derive(Debug, Clone, Copy)]
62pub(crate) struct Arrival {
63    /// When the command was sent.
64    pub(crate) enqueued_at: Instant,
65    /// Whether its caller had stopped waiting for the reply by now.
66    pub(crate) caller_gone: bool,
67}
68
69/// Meters the engine loop's turns for an [`EventQueueRecorder`].
70pub(crate) struct EventQueueMeter {
71    recorder: Arc<dyn EventQueueRecorder>,
72    idle_since: Instant,
73}
74
75impl EventQueueMeter {
76    /// A meter whose loop starts idle now.
77    pub(crate) fn new(recorder: Arc<dyn EventQueueRecorder>) -> Self {
78        Self {
79            recorder,
80            idle_since: Instant::now(),
81        }
82    }
83
84    /// The loop woke for one event: a command, with its `arrival`, or a timer
85    /// with `None`. Records the idle wait that ended and the start, and
86    /// returns the timer of the event's processing.
87    pub(crate) fn start(&self, arrival: Option<Arrival>) -> EventTimer {
88        let now = Instant::now();
89        self.recorder
90            .idle_time(now.saturating_duration_since(self.idle_since));
91        self.recorder.operation_started();
92        if let Some(arrival) = arrival {
93            self.recorder
94                .queue_time(now.saturating_duration_since(arrival.enqueued_at));
95            if arrival.caller_gone {
96                self.recorder.operation_timed_out();
97            }
98        }
99        EventTimer {
100            started: now,
101            recorder: Arc::clone(&self.recorder),
102        }
103    }
104
105    /// The turn is over; the loop is idle from now.
106    pub(crate) fn finish_turn(&mut self) { self.idle_since = Instant::now(); }
107}
108
109/// The running processing time of one event. Dropping it records the time
110/// since the event started, so a write keeps it in its commit waiter until it
111/// replies.
112pub(crate) struct EventTimer {
113    started: Instant,
114    recorder: Arc<dyn EventQueueRecorder>,
115}
116
117impl Drop for EventTimer {
118    fn drop(&mut self) { self.recorder.processing_time(self.started.elapsed()); }
119}
120
121#[cfg(test)]
122pub(crate) mod test_support {
123    use {super::EventQueueRecorder, std::sync::Mutex, tokio::time::Duration};
124
125    /// What a [`RecordingEventQueue`] has been told, in arrival order.
126    #[derive(Debug, Clone, Default, PartialEq, Eq)]
127    pub struct EventQueueRecord {
128        pub(crate) queue_times: Vec<Duration>,
129        pub(crate) processing_times: Vec<Duration>,
130        pub(crate) started: u64,
131        pub(crate) timed_out: u64,
132        pub(crate) idle_times: Vec<Duration>,
133    }
134
135    /// An [`EventQueueRecorder`] that keeps every measurement.
136    #[derive(Debug, Default)]
137    pub struct RecordingEventQueue(Mutex<EventQueueRecord>);
138
139    impl RecordingEventQueue {
140        pub(crate) fn record(&self) -> EventQueueRecord {
141            self.0.lock().expect("the record is not poisoned").clone()
142        }
143
144        fn with(&self, update: impl FnOnce(&mut EventQueueRecord)) {
145            update(&mut self.0.lock().expect("the record is not poisoned"));
146        }
147    }
148
149    impl EventQueueRecorder for RecordingEventQueue {
150        fn queue_time(&self, waited: Duration) { self.with(|r| r.queue_times.push(waited)); }
151
152        fn processing_time(&self, took: Duration) { self.with(|r| r.processing_times.push(took)); }
153
154        fn operation_started(&self) { self.with(|r| r.started += 1); }
155
156        fn operation_timed_out(&self) { self.with(|r| r.timed_out += 1); }
157
158        fn idle_time(&self, idle: Duration) { self.with(|r| r.idle_times.push(idle)); }
159    }
160}
161
162#[cfg(test)]
163mod tests {
164    use {
165        super::{
166            Arrival, EventQueueMeter,
167            test_support::{EventQueueRecord, RecordingEventQueue},
168        },
169        assert2::check,
170        std::sync::Arc,
171        tokio::time::{Duration, Instant, advance},
172    };
173
174    /// One turn under a paused clock. A command, when there is one, is sent
175    /// at the start; the previous turn then runs for `busy`, the loop idles
176    /// for `idle`, and this turn runs for `runs`.
177    struct TurnCase {
178        name: &'static str,
179        command: Option<Caller>,
180        busy: Duration,
181        idle: Duration,
182        runs: Duration,
183        expected: EventQueueRecord,
184    }
185
186    /// Whether the command's caller is still waiting when the turn starts.
187    #[derive(Clone, Copy)]
188    enum Caller {
189        Waiting,
190        Gone,
191    }
192
193    #[tokio::test(start_paused = true)]
194    async fn one_turn_records_kafkas_event_queue_measurements() {
195        let ms = Duration::from_millis;
196        let cases = [
197            TurnCase {
198                name: "a timer has no queue time",
199                command: None,
200                busy: ms(9),
201                idle: ms(40),
202                runs: ms(3),
203                expected: EventQueueRecord {
204                    queue_times: vec![],
205                    processing_times: vec![ms(3)],
206                    started: 1,
207                    timed_out: 0,
208                    idle_times: vec![ms(40)],
209                },
210            },
211            TurnCase {
212                name: "a command queued behind a busy turn",
213                command: Some(Caller::Waiting),
214                busy: ms(25),
215                idle: ms(0),
216                runs: ms(7),
217                expected: EventQueueRecord {
218                    queue_times: vec![ms(25)],
219                    processing_times: vec![ms(7)],
220                    started: 1,
221                    timed_out: 0,
222                    idle_times: vec![ms(0)],
223                },
224            },
225            TurnCase {
226                name: "a command whose caller left still runs, and counts as timed out",
227                command: Some(Caller::Gone),
228                busy: ms(30_000),
229                idle: ms(5),
230                runs: ms(1),
231                expected: EventQueueRecord {
232                    queue_times: vec![ms(30_005)],
233                    processing_times: vec![ms(1)],
234                    started: 1,
235                    timed_out: 1,
236                    idle_times: vec![ms(5)],
237                },
238            },
239        ];
240        for case in cases {
241            let recorder = Arc::new(RecordingEventQueue::default());
242            let mut meter = EventQueueMeter::new(Arc::<RecordingEventQueue>::clone(&recorder));
243            let enqueued_at = Instant::now();
244            advance(case.busy).await;
245            meter.finish_turn();
246            advance(case.idle).await;
247            let timer = meter.start(case.command.map(|caller| Arrival {
248                enqueued_at,
249                caller_gone: matches!(caller, Caller::Gone),
250            }));
251            advance(case.runs).await;
252            drop(timer);
253            check!(recorder.record() == case.expected, "{}", case.name);
254        }
255    }
256}
krabka_raft/event_queue.rs

krabka_raft/
event_queue.rs
