diff --git a/CHANGELOG.md b/CHANGELOG.md index 8438cb51..cc1f43df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed — Hot-path performance: deep-review optimization pass (PR #168) + +- **+33% pipelined SET, +25% pipelined GET (single shard)** from a deep + performance review covering protocol → dispatch → shard → storage → + persistence → vector. 11 verified findings implemented: + - `Database::get`: 3 → 2 DashTable probes on a hit, 2 → 1 on a miss + (single-probe live/expired/absent classification). + - WAL v2 flush: 3 → 1 `write(2)` syscalls per 1ms tick via a reusable + staging buffer; on-disk format unchanged. + - WAL v3 record append no longer heap-allocates per non-FPI record + (`Cow::Borrowed`); checkpoint state machine no longer clones per tick. + - SPSC drain (`drain_spsc_shared`) reuses thread-local batch buffers + instead of allocating two `Vec`s per call. + - `Frame::Double` serialization is zero-alloc (stack `f64` Display + formatter, byte-identical wire output) in RESP2 and RESP3. + - SQ8 vector search no longer heap-allocates per query (HNSW + both + brute-force paths); IVF rotation/LUT buffers hoisted out of the + per-segment loop in `search_filtered` (parity with `search_mvcc`). + - io_uring `drain_completions` reuses persistent CQE/event buffers + (allocation-free drain loop at steady state). +- Multi-shard (2/4/8/16/auto): no regressions; +9–32% pipelined SET at + 2 and 8 shards. The unpipelined cross-shard latency floor (~1.7ms p50, + waker-relay + dispatch tick) is unchanged and tracked for follow-up. + ### Fixed — SQ8 vector quantization now works across the full FT lifecycle - **`FT.CREATE ... QUANTIZATION SQ8` was a declared-but-unimplemented diff --git a/src/io/uring_driver.rs b/src/io/uring_driver.rs index 08193df4..30a3d826 100644 --- a/src/io/uring_driver.rs +++ b/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, } 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 { + 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) { + 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 { + 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 { + pub fn drain_completions_into(&mut self, events: &mut Vec) { 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; } // ----------------------------------------------------------------------- diff --git a/src/persistence/checkpoint.rs b/src/persistence/checkpoint.rs index 62bd12ec..7b853083 100644 --- a/src/persistence/checkpoint.rs +++ b/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, diff --git a/src/persistence/wal.rs b/src/persistence/wal.rs index 8a701eb2..bf6f8523 100644 --- a/src/persistence/wal.rs +++ b/src/persistence/wal.rs @@ -63,6 +63,10 @@ pub struct WalWriter { shard_id: usize, /// Buffered RESP bytes awaiting flush. buf: Vec, + /// 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, /// File handle for the WAL. file: Option, /// 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; diff --git a/src/persistence/wal_v3/record.rs b/src/persistence/wal_v3/record.rs index ab035bb1..5f87c154 100644 --- a/src/persistence/wal_v3/record.rs +++ b/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) diff --git a/src/protocol/serialize.rs b/src/protocol/serialize.rs index a6fd0b4b..c48d43d5 100644 --- a/src/protocol/serialize.rs +++ b/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); diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 44e6dfb9..47155c12 100644 --- a/src/shard/event_loop.rs +++ b/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 diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 996f52ee..f5c37ebc 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -79,8 +79,20 @@ pub(crate) fn drain_spsc_shared( let mut drained = 0; // Collect all messages first, then batch Execute/PipelineBatch under single borrow. - let mut execute_batch: Vec = Vec::new(); - let mut other_messages: Vec = Vec::new(); + // + // Scratch buffers are thread-local (one shard per OS thread) so this + // function — called from the 1ms tick and every I/O select arm — does + // not heap-allocate two Vecs per invocation. `mem::take` (instead of + // holding the RefCell borrow) keeps re-entrancy safe: a nested call + // would simply fall back to fresh empty Vecs. + thread_local! { + static DRAIN_SCRATCH: RefCell<(Vec, Vec)> = + const { RefCell::new((Vec::new(), Vec::new())) }; + } + let (mut execute_batch, mut other_messages) = + DRAIN_SCRATCH.with(|s| std::mem::take(&mut *s.borrow_mut())); + execute_batch.clear(); + other_messages.clear(); let mut snapshot_seen = false; for consumer in consumers.iter_mut() { @@ -156,7 +168,7 @@ pub(crate) fn drain_spsc_shared( // Process Execute/PipelineBatch/MultiExecute batch under single borrow_mut if !execute_batch.is_empty() { - for msg in execute_batch { + for msg in execute_batch.drain(..) { handle_shard_message_shared( shard_databases, pubsub_registry, @@ -188,7 +200,7 @@ pub(crate) fn drain_spsc_shared( } // Process other messages (PubSubPublish, SnapshotBegin, etc.) - for msg in other_messages { + for msg in other_messages.drain(..) { handle_shard_message_shared( shard_databases, pubsub_registry, @@ -213,6 +225,12 @@ pub(crate) fn drain_spsc_shared( aof_pool, // FIX-W1-2: thread AOF pool through SPSC drain ); } + + // Return the (now drained) scratch buffers so their capacity is reused + // by the next drain cycle. + DRAIN_SCRATCH.with(|s| { + *s.borrow_mut() = (execute_batch, other_messages); + }); } /// Process a single cross-shard message using shared database access. diff --git a/src/storage/db.rs b/src/storage/db.rs index 0ec35bfd..588af056 100644 --- a/src/storage/db.rs +++ b/src/storage/db.rs @@ -673,53 +673,63 @@ impl Database { /// Get an entry by key, performing lazy expiration. /// /// Returns `None` if the key does not exist or has expired. - /// Optimized: immutable lookup for expiry check, no LRU touch on reads - /// (LRU is only needed when eviction is active; callers requiring LRU - /// updates should use `get_mut()`). This reduces the hot path from - /// get_mut+get (2 SIMD probes) to get+get (1 probe for non-expired keys - /// since the compiler can often elide the second immutable lookup). + /// Optimized: a single immutable probe classifies the key as live, + /// expired, or absent. NLL forces one re-probe on the live path + /// (the cold fallback below mutates `self`, so the first borrow cannot + /// be returned directly): 2 probes on a live hit, 1 on a miss — down + /// from 3/2 with the previous expiry-check + `is_some()` + get chain. + /// No LRU touch on reads (callers requiring LRU updates use `get_mut()`). pub fn get(&mut self, key: &[u8]) -> Option<&Entry> { let now_ms = self.cached_now_ms; let base_ts = self.base_timestamp; - // Single immutable lookup: check existence + expiry - let expired = self - .data - .get(key) - .is_some_and(|e| e.is_expired_at(base_ts, now_ms)); - if expired { - let removed = self.data.remove(key)?; - self.used_memory = self - .used_memory - .saturating_sub(entry_overhead(key, &removed)); - return None; - } - // Hot path: DashTable lookup - if self.data.get(key).is_some() { - return self.data.get(key); - } - // Cold fallback: read from disk DataFile via cold_read helper. - // Extract owned result first to drop immutable borrows before mutation. - let cold_result = self.cold_shard_dir.as_ref().and_then(|shard_dir| { - self.cold_index.as_ref().and_then(|ci| { - crate::storage::tiered::cold_read::cold_read_through(ci, shard_dir, key, now_ms) - }) - }); - if let Some((redis_value, ttl_ms)) = cold_result { - let key_bytes = Bytes::copy_from_slice(key); - // Build an entry from the RedisValue (works for strings and collections) - let mut entry = Entry::new_string(Bytes::new()); // placeholder - entry.value = - crate::storage::compact_value::CompactValue::from_redis_value(redis_value); - if let Some(ttl) = ttl_ms { - entry.set_expires_at_ms(self.base_timestamp, ttl); + enum KeyState { + Live, + Expired, + Absent, + } + let state = match self.data.get(key) { + Some(e) if e.is_expired_at(base_ts, now_ms) => KeyState::Expired, + Some(_) => KeyState::Live, + None => KeyState::Absent, + }; + match state { + // Hot path: single re-probe, borrow returned to caller. + KeyState::Live => self.data.get(key), + KeyState::Expired => { + let removed = self.data.remove(key)?; + self.used_memory = self + .used_memory + .saturating_sub(entry_overhead(key, &removed)); + None } - self.set(key_bytes, entry); - if let Some(ref mut ci) = self.cold_index { - ci.remove(key); + KeyState::Absent => { + // Cold fallback: read from disk DataFile via cold_read helper. + // Extract owned result first to drop immutable borrows before mutation. + let cold_result = self.cold_shard_dir.as_ref().and_then(|shard_dir| { + self.cold_index.as_ref().and_then(|ci| { + crate::storage::tiered::cold_read::cold_read_through( + ci, shard_dir, key, now_ms, + ) + }) + }); + if let Some((redis_value, ttl_ms)) = cold_result { + let key_bytes = Bytes::copy_from_slice(key); + // Build an entry from the RedisValue (works for strings and collections) + let mut entry = Entry::new_string(Bytes::new()); // placeholder + entry.value = + crate::storage::compact_value::CompactValue::from_redis_value(redis_value); + if let Some(ttl) = ttl_ms { + entry.set_expires_at_ms(self.base_timestamp, ttl); + } + self.set(key_bytes, entry); + if let Some(ref mut ci) = self.cold_index { + ci.remove(key); + } + return self.data.get(key); + } + None } - return self.data.get(key); } - None } /// Get a mutable reference to an entry by key, performing lazy expiration and access tracking. diff --git a/src/storage/eviction.rs b/src/storage/eviction.rs index 54dbeee2..03f42fd4 100644 --- a/src/storage/eviction.rs +++ b/src/storage/eviction.rs @@ -106,7 +106,7 @@ pub fn aggregate_used_memory(databases: &[Database]) -> usize { } /// Eviction policy variants matching Redis maxmemory-policy. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EvictionPolicy { NoEviction, AllKeysLru, @@ -120,17 +120,26 @@ pub enum EvictionPolicy { impl EvictionPolicy { /// Parse a policy name string (case-insensitive) into an EvictionPolicy. + /// + /// Allocation-free: called per `try_evict_if_needed` invocation (i.e. per + /// write command under memory pressure), so it must not build a lowercase + /// `String` on every call. pub fn from_str(s: &str) -> Self { - match s.to_ascii_lowercase().as_str() { - "allkeys-lru" => EvictionPolicy::AllKeysLru, - "allkeys-lfu" => EvictionPolicy::AllKeysLfu, - "allkeys-random" => EvictionPolicy::AllKeysRandom, - "volatile-lru" => EvictionPolicy::VolatileLru, - "volatile-lfu" => EvictionPolicy::VolatileLfu, - "volatile-random" => EvictionPolicy::VolatileRandom, - "volatile-ttl" => EvictionPolicy::VolatileTtl, - _ => EvictionPolicy::NoEviction, + const TABLE: [(&str, EvictionPolicy); 7] = [ + ("allkeys-lru", EvictionPolicy::AllKeysLru), + ("allkeys-lfu", EvictionPolicy::AllKeysLfu), + ("allkeys-random", EvictionPolicy::AllKeysRandom), + ("volatile-lru", EvictionPolicy::VolatileLru), + ("volatile-lfu", EvictionPolicy::VolatileLfu), + ("volatile-random", EvictionPolicy::VolatileRandom), + ("volatile-ttl", EvictionPolicy::VolatileTtl), + ]; + for (name, policy) in TABLE { + if s.eq_ignore_ascii_case(name) { + return policy; + } } + EvictionPolicy::NoEviction } /// Return the canonical string name for this policy. diff --git a/src/vector/hnsw/search.rs b/src/vector/hnsw/search.rs index 8cfeafb1..603c04f3 100644 --- a/src/vector/hnsw/search.rs +++ b/src/vector/hnsw/search.rs @@ -257,8 +257,10 @@ pub fn hnsw_search_filtered( // L2) and feeds a dedicated ADC in the distance closures below. The entire TQ // LUT setup is skipped for SQ8. let is_sq8 = collection.quantization == QuantizationConfig::Sq8; - let sq8_query: Vec = if is_sq8 { - let mut q = query.to_vec(); + // Inline buffer keeps the per-query SQ8 path heap-free for dim ≤ 512 + // (preserves the VEC-HNSW-03 zero-allocation guarantee). + let sq8_query: SmallVec<[f32; 512]> = if is_sq8 { + let mut q: SmallVec<[f32; 512]> = SmallVec::from_slice(query); if collection.metric != DistanceMetric::L2 { let n: f32 = q.iter().map(|x| x * x).sum::().sqrt(); if n > 0.0 { @@ -270,7 +272,7 @@ pub fn hnsw_search_filtered( } q } else { - Vec::new() + SmallVec::new() }; let q_rot = scratch.query_rotated.as_mut_slice(); diff --git a/src/vector/segment/holder.rs b/src/vector/segment/holder.rs index c5945a60..943b3b05 100644 --- a/src/vector/segment/holder.rs +++ b/src/vector/segment/holder.rs @@ -293,9 +293,15 @@ impl SegmentHolder { let dim = query_f32.len(); let pdim = padded_dimension(dim as u32) as usize; + // Allocate query rotation + LUT buffers ONCE, reuse across all IVF + // segments (same pattern as search_mvcc below — previously these + // were allocated per-segment-per-query, 12KB+ × n_segments). + let mut q_rotated = vec![0.0f32; pdim]; + let mut lut_buf = vec![0u8; pdim * 16]; + for ivf_seg in &snapshot.ivf { - // Rotate query using this IVF segment's sign flips. - let mut q_rotated = vec![0.0f32; pdim]; + // Reset and re-rotate for this segment (different sign_flips per segment) + q_rotated.iter_mut().for_each(|v| *v = 0.0); q_rotated[..dim].copy_from_slice(query_f32); // Normalize before FWHT. let qnorm: f32 = query_f32.iter().map(|x| x * x).sum::().sqrt(); @@ -307,9 +313,6 @@ impl SegmentHolder { } fwht::fwht(&mut q_rotated, ivf_seg.sign_flips()); - // LUT buffer on the stack (16KB for 1024-dim, well within 8MB stack). - let mut lut_buf = vec![0u8; pdim * 16]; - if let Some(bm) = filter_bitmap { all.extend(ivf_seg.search_filtered( query_f32, diff --git a/src/vector/segment/mutable.rs b/src/vector/segment/mutable.rs index 320f52be..bf57be8e 100644 --- a/src/vector/segment/mutable.rs +++ b/src/vector/segment/mutable.rs @@ -24,6 +24,47 @@ use crate::vector::types::{DistanceMetric, SearchResult, VectorId}; /// Maximum byte size before a mutable segment is considered full (128 MB). const MUTABLE_SEGMENT_MAX: usize = 128 * 1024 * 1024; +/// SQ8 ADC query preparation, heap-free on the per-query hot path. +/// +/// L2 borrows the caller's query directly (encode side stores raw values, +/// no normalization needed). Unit-sphere metrics (Cosine + InnerProduct) +/// copy into an inline buffer (stack for dim ≤ 512) and normalize to match +/// the encode-side normalization. +// The 2KB inline variant is deliberate: a stack buffer that keeps the +// per-query SQ8 hot path heap-free. The enum lives only as a short-lived +// stack local during search; boxing it would reintroduce the allocation. +#[allow(clippy::large_enum_variant)] +enum Sq8Query<'a> { + Borrowed(&'a [f32]), + Normalized(SmallVec<[f32; 512]>), +} + +impl<'a> Sq8Query<'a> { + #[inline] + fn prepare(query_f32: &'a [f32], metric: DistanceMetric) -> Self { + if metric == DistanceMetric::L2 { + return Sq8Query::Borrowed(query_f32); + } + let mut q: SmallVec<[f32; 512]> = SmallVec::from_slice(query_f32); + let n: f32 = q.iter().map(|x| x * x).sum::().sqrt(); + if n > 0.0 { + let inv = 1.0 / n; + for v in q.iter_mut() { + *v *= inv; + } + } + Sq8Query::Normalized(q) + } + + #[inline] + fn as_slice(&self) -> &[f32] { + match self { + Sq8Query::Borrowed(s) => s, + Sq8Query::Normalized(q) => q, + } + } +} + /// 48 bytes. MVCC fields prepared for Phase 65. #[repr(C)] pub struct MutableEntry { @@ -349,18 +390,9 @@ impl MutableSegment { // Query is normalized for unit-sphere metrics (Cosine + InnerProduct) to // match the encode-side normalization. if self.collection.quantization == QuantizationConfig::Sq8 { - let mut q = query_f32.to_vec(); - // Normalize for unit-sphere metrics (Cosine + InnerProduct) to match the - // encode side; L2 keeps the raw query for true Euclidean ADC. - if self.collection.metric != DistanceMetric::L2 { - let n: f32 = q.iter().map(|x| x * x).sum::().sqrt(); - if n > 0.0 { - let inv = 1.0 / n; - for v in q.iter_mut() { - *v *= inv; - } - } - } + // Heap-free query prep: borrow for L2, inline-normalize otherwise. + let q = Sq8Query::prepare(query_f32, self.collection.metric); + let q = q.as_slice(); let mut heap: BinaryHeap = BinaryHeap::with_capacity(k + 1); for entry in &inner.entries { if entry.delete_lsn != 0 { @@ -376,7 +408,7 @@ impl MutableSegment { let off = id * bytes_per_code; let slot = &inner.tq_codes[off..off + bytes_per_code]; let (min, scale) = sq8_params(slot, dim); - let dist = sq8_l2_adc(&q, &slot[..dim], min, scale); + let dist = sq8_l2_adc(q, &slot[..dim], min, scale); let global_id = inner.global_id_base + entry.internal_id; if heap.len() < k { heap.push(DistF32(dist, global_id, entry.key_hash)); @@ -540,18 +572,9 @@ impl MutableSegment { // SQ8: per-vector affine decode + true squared-L2 ADC (MVCC-visible scan). if self.collection.quantization == QuantizationConfig::Sq8 { - let mut q = query_f32.to_vec(); - // Normalize for unit-sphere metrics (Cosine + InnerProduct) to match the - // encode side; L2 keeps the raw query for true Euclidean ADC. - if self.collection.metric != DistanceMetric::L2 { - let n: f32 = q.iter().map(|x| x * x).sum::().sqrt(); - if n > 0.0 { - let inv = 1.0 / n; - for v in q.iter_mut() { - *v *= inv; - } - } - } + // Heap-free query prep: borrow for L2, inline-normalize otherwise. + let q = Sq8Query::prepare(query_f32, self.collection.metric); + let q = q.as_slice(); let mut heap: BinaryHeap = BinaryHeap::with_capacity(k + 1); for entry in &inner.entries { if !is_visible( @@ -574,7 +597,7 @@ impl MutableSegment { let off = id * bytes_per_code; let slot = &inner.tq_codes[off..off + bytes_per_code]; let (min, scale) = sq8_params(slot, dim); - let dist = sq8_l2_adc(&q, &slot[..dim], min, scale); + let dist = sq8_l2_adc(q, &slot[..dim], min, scale); let global_id = inner.global_id_base + entry.internal_id; if heap.len() < k { heap.push(DistF32(dist, global_id, entry.key_hash));