perf: deep-review hot-path optimizations (storage, WAL, SPSC, protocol, vector, io_uring)

src/io/uring_driver.rs

@@ -223,6 +223,12 @@ pub struct UringDriver {
     /// Eventfd registered with io_uring for CQE notifications.
     /// When CQEs arrive, the kernel writes to this fd, waking tokio's epoll.
     cqe_eventfd: RawFd,
+    /// Reusable scratch for decoded CQEs so `drain_completions*` doesn't
+    /// heap-allocate per batch (called on every eventfd wake + 1ms tick).
+    cqe_scratch: Vec<(u8, u32, u32, i32, u32)>,
+    /// Reusable event buffer loaned to the event loop via
+    /// `take_event_scratch` / `return_event_scratch`.
+    event_scratch: Vec<IoEvent>,
 }
 
 impl UringDriver {
@@ -293,6 +299,8 @@ impl UringDriver {
             pending_sqes: 0,
             tick: 0,
             cqe_eventfd: efd,
+            cqe_scratch: Vec::with_capacity(256),
+            event_scratch: Vec::with_capacity(256),
         })
     }
 
@@ -586,28 +594,54 @@ impl UringDriver {
         Ok(n)
     }
 
+    /// Take the reusable event buffer (capacity retained across drains).
+    ///
+    /// The event loop borrows this around its drain loop so per-batch event
+    /// collection is allocation-free; give it back with
+    /// `return_event_scratch` when done.
+    #[inline]
+    pub fn take_event_scratch(&mut self) -> Vec<IoEvent> {
+        std::mem::take(&mut self.event_scratch)
+    }
+
+    /// Return the event buffer taken with `take_event_scratch`.
+    #[inline]
+    pub fn return_event_scratch(&mut self, mut events: Vec<IoEvent>) {
+        events.clear();
+        self.event_scratch = events;
+    }
+
     /// Drain all completed CQEs, returning events for the caller to process.
     ///
+    /// Allocating variant of [`Self::drain_completions_into`]; prefer the
+    /// `_into` form with `take_event_scratch` on the hot path.
+    pub fn drain_completions(&mut self) -> Vec<IoEvent> {
+        let mut events = Vec::new();
+        self.drain_completions_into(&mut events);
+        events
+    }
+
+    /// Drain all completed CQEs, appending events for the caller to process.
+    ///
     /// The caller (shard event loop) handles command dispatch based on event type.
     /// Buffer lifecycle: recv data is copied from the provided buffer before
     /// the buffer is returned to the ring (per pitfall 1 in research).
-    pub fn drain_completions(&mut self) -> Vec<IoEvent> {
+    pub fn drain_completions_into(&mut self, events: &mut Vec<IoEvent>) {
         self.tick += 1;
         let current_tick = self.tick;
-        let mut events = Vec::new();
 
         // Collect CQEs first to release the mutable borrow on self.ring,
         // allowing return_buf to access the ring's submission queue below.
-        let cqes: Vec<(u8, u32, u32, i32, u32)> = self
-            .ring
-            .completion()
-            .map(|cqe| {
-                let (event_type, conn_id, aux) = decode_user_data(cqe.user_data());
-                (event_type, conn_id, aux, cqe.result(), cqe.flags())
-            })
-            .collect();
-
-        for (event_type, conn_id, _aux, result, flags) in cqes {
+        // Reuses the persistent scratch (mem::take avoids a borrow conflict
+        // with the buf_ring/connections accesses in the loop body).
+        let mut cqes = std::mem::take(&mut self.cqe_scratch);
+        cqes.clear();
+        cqes.extend(self.ring.completion().map(|cqe| {
+            let (event_type, conn_id, aux) = decode_user_data(cqe.user_data());
+            (event_type, conn_id, aux, cqe.result(), cqe.flags())
+        }));
+
+        for (event_type, conn_id, _aux, result, flags) in cqes.drain(..) {
             match event_type {
                 EVENT_ACCEPT => {
                     if result >= 0 {
@@ -691,7 +725,8 @@ impl UringDriver {
             }
         }
 
-        events
+        // Put the (drained) CQE scratch back for the next batch.
+        self.cqe_scratch = cqes;
     }
 
     // -----------------------------------------------------------------------

src/persistence/checkpoint.rs

@@ -67,7 +67,10 @@ impl CheckpointTrigger {
 }
 
 /// Internal state of the checkpoint protocol.
-#[derive(Debug, Clone, PartialEq, Eq)]
+///
+/// All fields are scalar — `Copy` keeps `advance_tick` (1ms tick path)
+/// free of `clone()` calls.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum CheckpointState {
     /// No checkpoint in progress.
     Idle,
@@ -163,7 +166,7 @@ impl CheckpointManager {
     /// - `FlushPages(n)` — flush n dirty pages
     /// - `Finalize { redo_lsn }` — all pages done, write WAL checkpoint record
     pub fn advance_tick(&mut self) -> CheckpointAction {
-        match self.state.clone() {
+        match self.state {
             CheckpointState::Idle => CheckpointAction::Nothing,
             CheckpointState::InProgress {
                 redo_lsn,

src/persistence/wal.rs

@@ -63,6 +63,10 @@ pub struct WalWriter {
     shard_id: usize,
     /// Buffered RESP bytes awaiting flush.
     buf: Vec<u8>,
+    /// Staging buffer for assembling the full block frame
+    /// (header + payload + CRC) so each flush is a single write syscall.
+    /// Cleared after every flush; capacity is retained.
+    frame_buf: Vec<u8>,
     /// File handle for the WAL.
     file: Option<std::fs::File>,
     /// Path to the WAL file.
@@ -97,6 +101,7 @@ impl WalWriter {
         let mut writer = Self {
             shard_id,
             buf: Vec::with_capacity(8192),
+            frame_buf: Vec::with_capacity(8192),
             file: Some(file),
             file_path,
             write_offset,
@@ -233,15 +238,19 @@ impl WalWriter {
             // block_len = cmd_count(2) + db_idx(1) + payload.len() + crc32(4)
             let block_len = (2 + 1 + self.buf.len() + 4) as u32;
 
-            // Write block frame: 7-byte header on stack, then payload, then CRC
-            let mut block_header = [0u8; 7];
-            block_header[0..4].copy_from_slice(&block_len.to_le_bytes());
-            block_header[4..6].copy_from_slice(&cmd_count_bytes);
-            block_header[6] = db_idx;
-
-            file.write_all(&block_header)?;
-            file.write_all(&self.buf)?;
-            file.write_all(&crc.to_le_bytes())?;
+            // Assemble the full block frame (header + payload + CRC) in the
+            // reusable staging buffer so the flush is ONE write syscall
+            // instead of three. On-disk layout is unchanged:
+            // [block_len:4 LE][cmd_count:2 LE][db_idx:1][payload:var][crc32:4 LE]
+            self.frame_buf.clear();
+            self.frame_buf.reserve(7 + self.buf.len() + 4);
+            self.frame_buf.extend_from_slice(&block_len.to_le_bytes());
+            self.frame_buf.extend_from_slice(&cmd_count_bytes);
+            self.frame_buf.push(db_idx);
+            self.frame_buf.extend_from_slice(&self.buf);
+            self.frame_buf.extend_from_slice(&crc.to_le_bytes());
+
+            file.write_all(&self.frame_buf)?;
 
             let total_written = 4 + block_len as u64; // block_len prefix + block_len contents
             self.write_offset += total_written;

src/persistence/wal_v3/record.rs

@@ -132,13 +132,15 @@ pub fn write_wal_v3_record(
     let should_compress =
         record_type == WalRecordType::FullPageImage && payload.len() > FPI_COMPRESS_THRESHOLD;
 
-    let (actual_payload, flags) = if should_compress {
+    // Cow keeps the common non-FPI path allocation-free: only compressed
+    // FPI payloads own a buffer; everything else borrows `payload` directly.
+    let (actual_payload, flags): (std::borrow::Cow<'_, [u8]>, u8) = if should_compress {
         (
-            lz4_flex::compress_prepend_size(payload),
+            std::borrow::Cow::Owned(lz4_flex::compress_prepend_size(payload)),
             FLAG_LZ4_COMPRESSED,
         )
     } else {
-        (payload.to_vec(), 0u8)
+        (std::borrow::Cow::Borrowed(payload), 0u8)
     };
 
     // record_len = 4 (len field) + 12 (header) + payload + 4 (crc)

src/protocol/serialize.rs

@@ -2,6 +2,52 @@ use bytes::{BufMut, BytesMut};
 
 use super::frame::Frame;
 
+/// Stack buffer holding the `Display` output of an `f64`.
+///
+/// `{}` formatting of `f64` never uses scientific notation, so the longest
+/// output is a subnormal like `5e-324`: `-0.` + 323 zeros + up to 17
+/// significant digits ≈ 343 bytes. 400 leaves comfortable margin.
+/// Produces byte-identical output to `format!("{}", f)` without the heap
+/// allocation (response serialization is a hot path).
+struct F64Display {
+    buf: [u8; 400],
+    len: usize,
+}
+
+impl F64Display {
+    #[inline]
+    fn format(f: f64) -> Self {
+        use std::fmt::Write;
+        let mut this = F64Display {
+            buf: [0u8; 400],
+            len: 0,
+        };
+        // Cannot fail: buffer is sized for the worst-case f64 Display output
+        // (see struct doc). On the impossible overflow, output is truncated
+        // rather than panicking.
+        let _ = write!(this, "{f}");
+        this
+    }
+
+    #[inline]
+    fn as_bytes(&self) -> &[u8] {
+        &self.buf[..self.len]
+    }
+}
+
+impl std::fmt::Write for F64Display {
+    fn write_str(&mut self, s: &str) -> std::fmt::Result {
+        let bytes = s.as_bytes();
+        let end = self.len + bytes.len();
+        let Some(dst) = self.buf.get_mut(self.len..end) else {
+            return Err(std::fmt::Error);
+        };
+        dst.copy_from_slice(bytes);
+        self.len = end;
+        Ok(())
+    }
+}
+
 /// Serialize a Frame into RESP2 wire format, appending to the buffer.
 pub fn serialize(frame: &Frame, buf: &mut BytesMut) {
     match frame {
@@ -64,23 +110,26 @@ pub fn serialize(frame: &Frame, buf: &mut BytesMut) {
             }
         }
         Frame::Double(f) => {
-            // Downgrade: BulkString of formatted float
-            let s = if f.is_infinite() {
+            // Downgrade: BulkString of formatted float (zero-alloc: stack
+            // buffer for finite values, static slices for inf/nan)
+            let formatted;
+            let s: &[u8] = if f.is_infinite() {
                 if f.is_sign_positive() {
-                    "inf".to_string()
+                    b"inf"
                 } else {
-                    "-inf".to_string()
+                    b"-inf"
                 }
             } else if f.is_nan() {
-                "nan".to_string()
+                b"nan"
             } else {
-                format!("{}", f)
+                formatted = F64Display::format(*f);
+                formatted.as_bytes()
             };
             buf.put_u8(b'$');
             let mut itoa_buf = itoa::Buffer::new();
             buf.put_slice(itoa_buf.format(s.len()).as_bytes());
             buf.put_slice(b"\r\n");
-            buf.put_slice(s.as_bytes());
+            buf.put_slice(s);
             buf.put_slice(b"\r\n");
         }
         Frame::Boolean(b) => {
@@ -173,8 +222,8 @@ pub fn serialize_resp3(frame: &Frame, buf: &mut BytesMut) {
             } else if f.is_nan() {
                 buf.put_slice(b"nan");
             } else {
-                let s = format!("{}", f);
-                buf.put_slice(s.as_bytes());
+                // Zero-alloc: stack buffer, byte-identical to format!("{}", f)
+                buf.put_slice(F64Display::format(*f).as_bytes());
             }
             buf.put_slice(b"\r\n");
         }
@@ -618,6 +667,52 @@ mod tests {
         assert_eq!(&buf[..], b"$3\r\n1.5\r\n");
     }
 
+    // === F64Display stack formatter: must match format!("{}", f) exactly ===
+
+    #[test]
+    fn test_f64_display_matches_std_format() {
+        // Includes the worst-case Display lengths: f64::MAX (~309 digits)
+        // and the smallest subnormal (~326 chars), which guard the stack
+        // buffer capacity in F64Display.
+        let cases = [
+            0.0,
+            -0.0,
+            1.5,
+            -42.5,
+            1.23,
+            3.141592653589793,
+            f64::MAX,
+            f64::MIN,
+            f64::MIN_POSITIVE,
+            5e-324, // smallest subnormal — longest Display output
+            1e308,
+            -1e-308,
+        ];
+        for f in cases {
+            let expected = format!("{}", f);
+            let got = F64Display::format(f);
+            assert_eq!(
+                got.as_bytes(),
+                expected.as_bytes(),
+                "F64Display mismatch for {f:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn test_serialize_double_extreme_values_resp2_and_resp3() {
+        for f in [f64::MAX, 5e-324, -1e-308] {
+            let s = format!("{}", f);
+            let buf = serialize_frame(&Frame::Double(f));
+            let expected = format!("${}\r\n{}\r\n", s.len(), s);
+            assert_eq!(&buf[..], expected.as_bytes());
+
+            let buf3 = serialize_resp3_frame(&Frame::Double(f));
+            let expected3 = format!(",{}\r\n", s);
+            assert_eq!(&buf3[..], expected3.as_bytes());
+        }
+    }
+
     #[test]
     fn test_resp2_null_still_dollar_minus_one() {
         let buf = serialize_frame(&Frame::Null);

src/shard/event_loop.rs

@@ -1072,19 +1072,24 @@ impl super::Shard {
                     #[cfg(target_os = "linux")]
                     if let Some(ref mut driver) = uring_state {
                         driver.drain_eventfd();
+                        // Borrow the driver's reusable event buffer: zero
+                        // allocations per CQE batch in this drain loop.
+                        let mut events = driver.take_event_scratch();
                         loop {
                             let _ = driver.submit_and_wait_nonblocking();
-                            let events = driver.drain_completions();
+                            events.clear();
+                            driver.drain_completions_into(&mut events);
                             if events.is_empty() {
                                 break;
                             }
-                            for event in events {
+                            for event in events.drain(..) {
                                 uring_handler::handle_uring_event(
                                     event, driver, &shard_databases, shard_id, &mut uring_parse_bufs,
                                     &mut inflight_sends, uring_listener_fd, &cached_clock,
                                 );
                             }
                         }
+                        driver.return_event_scratch(events);
                     }
                 }
                 // Per-shard SO_REUSEPORT accept (unix, non-uring tokio path)
@@ -1465,13 +1470,16 @@ impl super::Shard {
                     #[cfg(target_os = "linux")]
                     if let Some(ref mut driver) = uring_state {
                         let _ = driver.submit_and_wait_nonblocking();
-                        let events = driver.drain_completions();
-                        for event in events {
+                        // Reuse the driver's event buffer (1ms tick path).
+                        let mut events = driver.take_event_scratch();
+                        driver.drain_completions_into(&mut events);
+                        for event in events.drain(..) {
                             uring_handler::handle_uring_event(
                                 event, driver, &shard_databases, shard_id, &mut uring_parse_bufs,
                                 &mut inflight_sends, uring_listener_fd, &cached_clock,
                             );
                         }
+                        driver.return_event_scratch(events);
                     }
                 }
                 // WAL fsync + MVCC sweep on 1-second interval