diff --git a/Cargo.lock b/Cargo.lock index 5e0fb43b..f0892e04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -807,6 +807,7 @@ dependencies = [ "bytes", "criterion", "itoa", + "memchr", "thiserror 2.0.18", ] diff --git a/Cargo.toml b/Cargo.toml index 49697165..a5494c6e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,9 +21,10 @@ categories = ["caching", "database-implementations"] [profile.release] overflow-checks = true -lto = "thin" +lto = true codegen-units = 1 strip = "symbols" +panic = "abort" [workspace.dependencies] # async runtime diff --git a/crates/ember-core/src/keyspace.rs b/crates/ember-core/src/keyspace.rs index a4b23ad2..c574552d 100644 --- a/crates/ember-core/src/keyspace.rs +++ b/crates/ember-core/src/keyspace.rs @@ -508,6 +508,10 @@ impl Keyspace { /// Randomly samples `EVICTION_SAMPLE_SIZE` keys and removes the one /// with the oldest `last_access` time. Returns `true` if a key was /// evicted, `false` if the keyspace is empty. + /// + /// Uses reservoir sampling with k=1 to avoid allocating a Vec on + /// every eviction attempt. The victim key index is remembered + /// rather than cloned, eliminating a heap allocation on the hot path. fn try_evict(&mut self) -> bool { if self.entries.is_empty() { return false; @@ -515,18 +519,38 @@ impl Keyspace { let mut rng = rand::rng(); - // randomly sample keys and find the least recently accessed one - let victim = self - .entries - .iter() - .choose_multiple(&mut rng, EVICTION_SAMPLE_SIZE) - .into_iter() - .min_by_key(|(_, entry)| entry.last_access_ms) - .map(|(k, _)| k.clone()); + // reservoir sample k=1 from the iterator, tracking the oldest + // entry by last_access_ms. this replaces choose_multiple() which + // allocates a Vec internally. + let mut best_key: Option<&str> = None; + let mut best_access = u64::MAX; + let mut seen = 0usize; + + for (key, entry) in &self.entries { + // reservoir sampling: include this item with probability + // EVICTION_SAMPLE_SIZE / (seen + 1), capped once we have + // enough candidates + seen += 1; + if seen <= EVICTION_SAMPLE_SIZE { + if entry.last_access_ms < best_access { + best_access = entry.last_access_ms; + best_key = Some(key.as_str()); + } + } else { + use rand::Rng; + let j = rng.random_range(0..seen); + if j < EVICTION_SAMPLE_SIZE && entry.last_access_ms < best_access { + best_access = entry.last_access_ms; + best_key = Some(key.as_str()); + } + } + } - if let Some(key) = victim { - if let Some(entry) = self.entries.remove(&key) { - self.memory.remove(&key, &entry.value); + if let Some(victim) = best_key { + // we need an owned key to remove from the map + let victim = victim.to_owned(); + if let Some(entry) = self.entries.remove(&victim) { + self.memory.remove(&victim, &entry.value); self.decrement_expiry_if_set(&entry); self.evicted_total += 1; self.defer_drop(entry.value); @@ -792,7 +816,7 @@ impl Keyspace { // Redis strips trailing zeros: "10.5" not "10.50000..." // but keeps at least one decimal if the result is a whole number let formatted = format_float(new_val); - let new_bytes = Bytes::from(formatted.clone()); + let new_bytes = Bytes::copy_from_slice(formatted.as_bytes()); match self.set(key.to_owned(), new_bytes, existing_expire) { SetResult::Ok => Ok(formatted), @@ -856,10 +880,11 @@ impl Keyspace { "KEYS on large keyspace, consider SCAN instead" ); } + let compiled = GlobPattern::new(pattern); self.entries .iter() .filter(|(_, entry)| !entry.is_expired()) - .filter(|(key, _)| glob_match(pattern, key)) + .filter(|(key, _)| compiled.matches(key)) .map(|(key, _)| key.clone()) .collect() } @@ -962,6 +987,8 @@ impl Keyspace { let mut position = 0u64; let target_count = if count == 0 { 10 } else { count }; + let compiled = pattern.map(GlobPattern::new); + for (key, entry) in self.entries.iter() { // skip expired entries if entry.is_expired() { @@ -975,8 +1002,8 @@ impl Keyspace { } // pattern matching - if let Some(pat) = pattern { - if !glob_match(pat, key) { + if let Some(ref pat) = compiled { + if !pat.matches(key) { position += 1; continue; } @@ -2192,8 +2219,36 @@ pub(crate) fn format_float(val: f64) -> String { /// /// Uses an iterative two-pointer algorithm with backtracking for O(n*m) /// worst-case performance, where n is pattern length and m is text length. +/// +/// Prefer [`GlobPattern::new`] + [`GlobPattern::matches`] when matching +/// the same pattern against many strings (KEYS, SCAN) to avoid +/// re-collecting the pattern chars on every call. pub(crate) fn glob_match(pattern: &str, text: &str) -> bool { let pat: Vec = pattern.chars().collect(); + glob_match_compiled(&pat, text) +} + +/// Pre-compiled glob pattern that avoids re-allocating pattern chars +/// on every match call. Use for KEYS/SCAN where the same pattern is +/// tested against every key in the keyspace. +pub(crate) struct GlobPattern { + chars: Vec, +} + +impl GlobPattern { + pub(crate) fn new(pattern: &str) -> Self { + Self { + chars: pattern.chars().collect(), + } + } + + pub(crate) fn matches(&self, text: &str) -> bool { + glob_match_compiled(&self.chars, text) + } +} + +/// Core glob matching against a pre-compiled pattern char slice. +fn glob_match_compiled(pat: &[char], text: &str) -> bool { let txt: Vec = text.chars().collect(); let mut pi = 0; // pattern index @@ -2536,12 +2591,12 @@ mod tests { #[test] fn safety_margin_rejects_near_raw_limit() { - // one entry = 1 (key) + 3 (val) + 96 (overhead) = 100 bytes. - // configure max_memory = 112. effective limit = 112 * 90 / 100 = 100. + // one entry = 1 (key) + 3 (val) + 128 (overhead) = 132 bytes. + // configure max_memory = 147. effective limit = 147 * 90 / 100 = 132. // the entry fills exactly the effective limit, so a second entry should - // be rejected even though the raw limit has 12 bytes of headroom. + // be rejected even though the raw limit has 15 bytes of headroom. let config = ShardConfig { - max_memory: Some(112), + max_memory: Some(147), eviction_policy: EvictionPolicy::NoEviction, ..ShardConfig::default() }; @@ -3400,8 +3455,12 @@ mod tests { #[test] fn lpush_evicts_under_lru_policy() { + // "a" entry = 1 + 3 + 128 = 132 bytes. + // list entry = 4 + 24 + (4 + 32) + 128 = 192 bytes. + // effective limit = 250 * 90 / 100 = 225. fits one entry but not both, + // so lpush should evict "a" to make room for the list. let config = ShardConfig { - max_memory: Some(200), + max_memory: Some(250), eviction_policy: EvictionPolicy::AllKeysLru, ..ShardConfig::default() }; diff --git a/crates/ember-core/src/memory.rs b/crates/ember-core/src/memory.rs index ea5d08d6..683c5e08 100644 --- a/crates/ember-core/src/memory.rs +++ b/crates/ember-core/src/memory.rs @@ -43,16 +43,20 @@ pub fn effective_limit(max_bytes: usize) -> usize { /// Estimated overhead per entry in the HashMap. /// -/// Accounts for: HashMap bucket pointer (8), Entry struct fields -/// (Option = 16, last_access Instant = 8, Value enum tag + padding), -/// plus HashMap per-entry bookkeeping. +/// Accounts for: the String key struct (24 bytes ptr+len+cap), Entry struct +/// fields (Value enum tag + Bytes/collection inline storage + expires_at_ms +/// + last_access_ms), plus hashbrown per-entry bookkeeping (1 control byte +/// + empty slot waste at ~87.5% load factor). /// -/// This is an approximation measured empirically on x86-64 linux. The exact -/// value varies by platform and compiler version, but precision isn't critical — -/// we use this for eviction triggers and memory reporting, not for correctness. +/// This is calibrated from `std::mem::size_of` on 64-bit platforms. The +/// exact value varies by compiler version, but precision isn't critical — +/// we use this for eviction triggers and memory reporting, not correctness. /// Overestimating is fine (triggers eviction earlier); underestimating could -/// theoretically let memory grow slightly beyond the configured limit. -pub(crate) const ENTRY_OVERHEAD: usize = 96; +/// let memory grow slightly beyond the configured limit. +/// +/// The `entry_overhead_not_too_small` test validates this constant against +/// the actual struct sizes on each platform. +pub(crate) const ENTRY_OVERHEAD: usize = 128; /// Tracks memory usage for a single keyspace. /// @@ -301,10 +305,31 @@ mod tests { fn entry_size_accounts_for_key_and_value() { let val = string_val("test"); let size = entry_size("mykey", &val); - // 5 (key) + 4 (value) + 96 (overhead) + // 5 (key) + 4 (value) + ENTRY_OVERHEAD assert_eq!(size, 5 + 4 + ENTRY_OVERHEAD); } + /// Validates that ENTRY_OVERHEAD is at least as large as the actual + /// struct sizes, so we never underestimate memory usage. + #[test] + fn entry_overhead_not_too_small() { + use crate::keyspace::Entry; + + let entry_size = std::mem::size_of::(); + let key_struct_size = std::mem::size_of::(); + // hashbrown uses 1 control byte per slot + ~14% empty slot waste. + // 8 bytes is a conservative lower bound for per-entry hash overhead. + let hashmap_per_entry = 8; + let minimum = entry_size + key_struct_size + hashmap_per_entry; + + assert!( + ENTRY_OVERHEAD >= minimum, + "ENTRY_OVERHEAD ({ENTRY_OVERHEAD}) is less than measured minimum \ + ({minimum} = Entry({entry_size}) + String({key_struct_size}) + \ + hashmap({hashmap_per_entry}))" + ); + } + #[test] fn list_value_size() { let mut deque = std::collections::VecDeque::new(); diff --git a/crates/ember-core/src/types/sorted_set.rs b/crates/ember-core/src/types/sorted_set.rs index 8345b194..93a68eef 100644 --- a/crates/ember-core/src/types/sorted_set.rs +++ b/crates/ember-core/src/types/sorted_set.rs @@ -4,12 +4,15 @@ //! are ordered by (score, member) — ties in score are broken //! lexicographically, matching Redis semantics. //! -//! Implementation uses a `BTreeMap<(OrderedFloat, String), ()>` for -//! ordered iteration and a `HashMap>` for O(1) -//! member→score lookups. This is simpler and more correct than a -//! hand-rolled skip list. +//! Implementation uses a `BTreeMap<(OrderedFloat, Arc), ()>` for +//! ordered iteration and a `HashMap, OrderedFloat>` for O(1) +//! member→score lookups. Member strings are shared via `Arc` between +//! both indexes, cutting per-member memory roughly in half vs storing +//! two independent `String`s. `Arc` (vs `Rc`) is required because shards +//! are spawned as tokio tasks which require `Send`. use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; use ordered_float::OrderedFloat; @@ -49,12 +52,15 @@ impl AddResult { /// /// Members are ordered by `(score, member_name)`. Rank is determined by /// position in this ordering (0-based, lowest score first). +/// +/// Member strings are shared between the score index and the member index +/// via `Arc`, so each string is stored once on the heap. #[derive(Debug, Clone)] pub struct SortedSet { /// Score→member index for ordered iteration. - tree: BTreeMap<(OrderedFloat, String), ()>, + tree: BTreeMap<(OrderedFloat, Arc), ()>, /// Member→score index for O(1) lookups. - scores: HashMap>, + scores: HashMap, OrderedFloat>, } impl SortedSet { @@ -76,7 +82,7 @@ impl SortedSet { pub fn add_with_flags(&mut self, member: String, score: f64, flags: &ZAddFlags) -> AddResult { let new_score = OrderedFloat(score); - if let Some(&old_score) = self.scores.get(&member) { + if let Some(&old_score) = self.scores.get(member.as_str()) { // member exists — skip if any flag condition prevents the update if flags.nx || (flags.gt && new_score <= old_score) @@ -85,10 +91,16 @@ impl SortedSet { { return AddResult::UNCHANGED; } - // update: remove old entry, insert new - self.tree.remove(&(old_score, member.clone())); - self.scores.insert(member.clone(), new_score); - self.tree.insert((new_score, member), ()); + // update: remove old tree entry, reuse the Rc for the new one + let name: Arc = self + .scores + .get_key_value(member.as_str()) + .unwrap() + .0 + .clone(); + self.tree.remove(&(old_score, name.clone())); + self.scores.insert(name.clone(), new_score); + self.tree.insert((new_score, name), ()); AddResult { added: false, updated: true, @@ -98,8 +110,9 @@ impl SortedSet { if flags.xx { return AddResult::UNCHANGED; } - self.scores.insert(member.clone(), new_score); - self.tree.insert((new_score, member), ()); + let name: Arc = Arc::from(member); + self.scores.insert(name.clone(), new_score); + self.tree.insert((new_score, name), ()); AddResult { added: true, updated: false, @@ -109,8 +122,8 @@ impl SortedSet { /// Removes a member from the sorted set. Returns `true` if it existed. pub fn remove(&mut self, member: &str) -> bool { - if let Some(score) = self.scores.remove(member) { - self.tree.remove(&(score, member.to_owned())); + if let Some((name, score)) = self.scores.remove_entry(member) { + self.tree.remove(&(score, name)); true } else { false @@ -129,8 +142,8 @@ impl SortedSet { /// small-to-medium sets; a skip list with rank counts would give /// O(log n) if this becomes a bottleneck. pub fn rank(&self, member: &str) -> Option { - let score = self.scores.get(member)?; - let key = (*score, member.to_owned()); + let (name, &score) = self.scores.get_key_value(member)?; + let key = (score, name.clone()); // count entries before this one Some(self.tree.range(..&key).count()) } @@ -151,7 +164,7 @@ impl SortedSet { .keys() .skip(s) .take(e - s + 1) - .map(|(score, member)| (member.as_str(), score.0)) + .map(|(score, member)| (&**member, score.0)) .collect() } @@ -167,9 +180,7 @@ impl SortedSet { /// Returns an iterator over (member, score) pairs in sorted order. pub fn iter(&self) -> impl Iterator { - self.tree - .keys() - .map(|(score, member)| (member.as_str(), score.0)) + self.tree.keys().map(|(score, member)| (&**member, score.0)) } /// Estimates memory usage in bytes. @@ -193,11 +204,16 @@ impl SortedSet { /// Estimates the memory cost of storing a single member. /// /// Includes BTreeMap entry overhead (64), HashMap entry overhead (56), - /// the member string stored in both collections, and the OrderedFloat. + /// two Arc pointers (8 bytes each), the shared string data once + /// (with Arc overhead of 16 bytes for strong + weak counts), and the + /// OrderedFloat in both. pub fn estimated_member_cost(member: &str) -> usize { const BTREE_ENTRY: usize = 64; const HASHMAP_ENTRY: usize = 56; - BTREE_ENTRY + HASHMAP_ENTRY + member.len() * 2 + 8 + const RC_PTR: usize = 8; // pointer per Rc clone + const RC_HEADER: usize = 16; // strong + weak counts + // string data stored once + Rc header + 2 Rc pointers + 2 OrderedFloat + BTREE_ENTRY + HASHMAP_ENTRY + member.len() + RC_HEADER + RC_PTR * 2 + 16 } } diff --git a/crates/ember-protocol/Cargo.toml b/crates/ember-protocol/Cargo.toml index 4a293095..6d8155cb 100644 --- a/crates/ember-protocol/Cargo.toml +++ b/crates/ember-protocol/Cargo.toml @@ -12,6 +12,7 @@ readme = "README.md" [dependencies] bytes = { workspace = true } itoa = "1" +memchr = "2" thiserror = { workspace = true } [dev-dependencies] diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index be62dfc2..bf6646bd 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -710,7 +710,7 @@ fn extract_string(frame: &Frame) -> Result { fn extract_bytes(frame: &Frame) -> Result { match frame { Frame::Bulk(data) => Ok(data.clone()), - Frame::Simple(s) => Ok(Bytes::from(s.clone().into_bytes())), + Frame::Simple(s) => Ok(Bytes::copy_from_slice(s.as_bytes())), _ => Err(ProtocolError::InvalidCommandFrame( "expected bulk or simple string argument".into(), )), diff --git a/crates/ember-protocol/src/parse.rs b/crates/ember-protocol/src/parse.rs index 89c74528..d61bfcfe 100644 --- a/crates/ember-protocol/src/parse.rs +++ b/crates/ember-protocol/src/parse.rs @@ -6,6 +6,14 @@ //! The parser uses a `Cursor<&[u8]>` to track its position through the //! input buffer without consuming it, allowing the caller to retry once //! more data arrives. +//! +//! # Single-pass design +//! +//! Earlier versions used a two-pass approach: `check()` to validate a +//! complete frame exists, then `parse()` to build Frame values. This +//! scanned every byte twice. The current implementation does a single +//! pass that builds Frame values directly, returning `Incomplete` if +//! the buffer doesn't contain enough data yet. use std::io::Cursor; @@ -39,11 +47,8 @@ pub fn parse_frame(buf: &[u8]) -> Result, ProtocolError> let mut cursor = Cursor::new(buf); - match check(&mut cursor, 0) { - Ok(()) => { - // we know a complete frame exists — reset and parse it - cursor.set_position(0); - let frame = parse(&mut cursor, 0)?; + match try_parse(&mut cursor, 0) { + Ok(frame) => { let consumed = cursor.position() as usize; Ok(Some((frame, consumed))) } @@ -53,103 +58,12 @@ pub fn parse_frame(buf: &[u8]) -> Result, ProtocolError> } // --------------------------------------------------------------------------- -// check: validates a complete frame exists without allocating -// --------------------------------------------------------------------------- - -/// Peeks through the buffer to verify a complete frame is present. -/// Advances the cursor past the frame on success. -fn check(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result<(), ProtocolError> { - let prefix = read_byte(cursor)?; - - match prefix { - b'+' | b'-' => check_line(cursor), - b':' => check_line(cursor), - b'$' => check_bulk(cursor), - b'*' => check_array(cursor, depth), - b'_' => check_line(cursor), - b'%' => check_map(cursor, depth), - other => Err(ProtocolError::InvalidPrefix(other)), - } -} - -fn check_line(cursor: &mut Cursor<&[u8]>) -> Result<(), ProtocolError> { - find_crlf(cursor)?; - Ok(()) -} - -fn check_bulk(cursor: &mut Cursor<&[u8]>) -> Result<(), ProtocolError> { - let len = read_integer_line(cursor)?; - if len < 0 { - return Err(ProtocolError::InvalidFrameLength(len)); - } - if len > MAX_BULK_LEN { - return Err(ProtocolError::BulkStringTooLarge(len as usize)); - } - let len = len as usize; - - // need `len` bytes of data + \r\n - let remaining = remaining(cursor); - if remaining < len + 2 { - return Err(ProtocolError::Incomplete); - } - - let pos = cursor.position() as usize; - // verify trailing \r\n - let buf = cursor.get_ref(); - if buf[pos + len] != b'\r' || buf[pos + len + 1] != b'\n' { - return Err(ProtocolError::InvalidFrameLength(len as i64)); - } - - cursor.set_position((pos + len + 2) as u64); - Ok(()) -} - -fn check_array(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result<(), ProtocolError> { - let next_depth = depth + 1; - if next_depth > MAX_NESTING_DEPTH { - return Err(ProtocolError::NestingTooDeep(MAX_NESTING_DEPTH)); - } - - let count = read_integer_line(cursor)?; - if count < 0 { - return Err(ProtocolError::InvalidFrameLength(count)); - } - if count as usize > MAX_ARRAY_ELEMENTS { - return Err(ProtocolError::TooManyElements(count as usize)); - } - - for _ in 0..count { - check(cursor, next_depth)?; - } - Ok(()) -} - -fn check_map(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result<(), ProtocolError> { - let next_depth = depth + 1; - if next_depth > MAX_NESTING_DEPTH { - return Err(ProtocolError::NestingTooDeep(MAX_NESTING_DEPTH)); - } - - let count = read_integer_line(cursor)?; - if count < 0 { - return Err(ProtocolError::InvalidFrameLength(count)); - } - if count as usize > MAX_ARRAY_ELEMENTS { - return Err(ProtocolError::TooManyElements(count as usize)); - } - - for _ in 0..count { - check(cursor, next_depth)?; // key - check(cursor, next_depth)?; // value - } - Ok(()) -} - -// --------------------------------------------------------------------------- -// parse: actually builds Frame values (only called after check succeeds) +// single-pass parser: validates and builds Frame values in one traversal // --------------------------------------------------------------------------- -fn parse(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result { +/// Parses a complete RESP3 frame from the cursor position, returning +/// `Incomplete` if the buffer doesn't contain enough data. +fn try_parse(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result { let prefix = read_byte(cursor)?; match prefix { @@ -172,18 +86,51 @@ fn parse(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result { - let len = read_integer_line(cursor)? as usize; + let len = read_integer_line(cursor)?; + if len < 0 { + return Err(ProtocolError::InvalidFrameLength(len)); + } + if len > MAX_BULK_LEN { + return Err(ProtocolError::BulkStringTooLarge(len as usize)); + } + let len = len as usize; + + // need `len` bytes of data + \r\n + let remaining = remaining(cursor); + if remaining < len + 2 { + return Err(ProtocolError::Incomplete); + } + let pos = cursor.position() as usize; - let data = &cursor.get_ref()[pos..pos + len]; - cursor.set_position((pos + len + 2) as u64); // skip data + \r\n + let buf = cursor.get_ref(); + + // verify trailing \r\n + if buf[pos + len] != b'\r' || buf[pos + len + 1] != b'\n' { + return Err(ProtocolError::InvalidFrameLength(len as i64)); + } + + let data = &buf[pos..pos + len]; + cursor.set_position((pos + len + 2) as u64); Ok(Frame::Bulk(Bytes::copy_from_slice(data))) } b'*' => { let next_depth = depth + 1; - let count = read_integer_line(cursor)? as usize; + if next_depth > MAX_NESTING_DEPTH { + return Err(ProtocolError::NestingTooDeep(MAX_NESTING_DEPTH)); + } + + let count = read_integer_line(cursor)?; + if count < 0 { + return Err(ProtocolError::InvalidFrameLength(count)); + } + if count as usize > MAX_ARRAY_ELEMENTS { + return Err(ProtocolError::TooManyElements(count as usize)); + } + + let count = count as usize; let mut frames = Vec::with_capacity(count); for _ in 0..count { - frames.push(parse(cursor, next_depth)?); + frames.push(try_parse(cursor, next_depth)?); } Ok(Frame::Array(frames)) } @@ -194,16 +141,27 @@ fn parse(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result { let next_depth = depth + 1; - let count = read_integer_line(cursor)? as usize; + if next_depth > MAX_NESTING_DEPTH { + return Err(ProtocolError::NestingTooDeep(MAX_NESTING_DEPTH)); + } + + let count = read_integer_line(cursor)?; + if count < 0 { + return Err(ProtocolError::InvalidFrameLength(count)); + } + if count as usize > MAX_ARRAY_ELEMENTS { + return Err(ProtocolError::TooManyElements(count as usize)); + } + + let count = count as usize; let mut pairs = Vec::with_capacity(count); for _ in 0..count { - let key = parse(cursor, next_depth)?; - let val = parse(cursor, next_depth)?; + let key = try_parse(cursor, next_depth)?; + let val = try_parse(cursor, next_depth)?; pairs.push((key, val)); } Ok(Frame::Map(pairs)) } - // check() already validated the prefix, so this shouldn't happen other => Err(ProtocolError::InvalidPrefix(other)), } } @@ -245,12 +203,17 @@ fn find_crlf(cursor: &mut Cursor<&[u8]>) -> Result { return Err(ProtocolError::Incomplete); } - // scan for \r\n - for i in start..buf.len().saturating_sub(1) { - if buf[i] == b'\r' && buf[i + 1] == b'\n' { - cursor.set_position((i + 2) as u64); - return Ok(i); + // SIMD-accelerated scan for \r, then verify \n follows. + // memchr processes 16-32 bytes per cycle vs 1 byte in a naive loop. + let mut pos = start; + while let Some(offset) = memchr::memchr(b'\r', &buf[pos..]) { + let cr = pos + offset; + if cr + 1 < buf.len() && buf[cr + 1] == b'\n' { + cursor.set_position((cr + 2) as u64); + return Ok(cr); } + // bare \r without \n — keep scanning past it + pos = cr + 1; } Err(ProtocolError::Incomplete) diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index 756f9640..bc54689b 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -166,6 +166,7 @@ where let mut buf = BytesMut::with_capacity(BUF_CAPACITY); let mut out = BytesMut::with_capacity(BUF_CAPACITY); + let mut frames = Vec::new(); loop { // guard against unbounded buffer growth from incomplete frames @@ -190,7 +191,7 @@ where // them concurrently to shards. this allows pipelined commands to // be processed in parallel rather than serially. out.clear(); - let mut frames = Vec::new(); + frames.clear(); loop { match parse_frame(&buf) { Ok(Some((frame, consumed))) => { @@ -210,7 +211,7 @@ where // when not yet authenticated, process frames serially so that an // AUTH command in a pipeline takes effect for subsequent frames if !authenticated { - for frame in frames { + for frame in frames.drain(..) { if is_auth_frame(&frame) { let (response, success) = try_auth(frame, ctx); response.serialize(&mut out); @@ -245,7 +246,7 @@ where if enter_sub { // process any non-subscribe commands that came before let mut sub_frames = Vec::new(); - for frame in frames { + for frame in frames.drain(..) { if is_subscribe_frame(&frame) { sub_frames.push(frame); } else { @@ -273,7 +274,7 @@ where // each dispatch is just an mpsc send (fast, completes // immediately when the channel has capacity). let mut pending = Vec::with_capacity(frames.len()); - for frame in frames { + for frame in frames.drain(..) { let p = dispatch_command(frame, &engine, ctx, slow_log, pubsub).await; pending.push(p); }