Split RespDeserializer into RespDeserializer + RespFrameParser (#125)

- Improve performance by parsing the RESP response on the network thread to detect the end of the frame and deserializing the RESP response on the client thread, based on the result of the parsing from the network thread

Author: mcatanzariti

Cargo.toml

@@ -106,10 +106,14 @@ harness = false
 required-features = ["bench"]
 
 [[bench]]
-name = "large_array_parsing"
+name = "parse_integer"
 harness = false
 required-features = ["bench"]
 
+[[bench]]
+name = "large_array_parsing"
+harness = false
+required-features = ["bench"]
 
 [[example]]
 name = "actix_crud"

benches/generic_api.rs

@@ -113,10 +113,10 @@ fn bench_rustis_simple_getsetdel_async(b: &mut Bencher) {
                 let key = "test_key";
 
                 client
-                    .send(cmd("SET").arg(key).arg(42.423456), None)
+                    .send::<()>(cmd("SET").arg(key).arg(42.423456), None)
                     .await?;
-                let _: f64 = client.send(cmd("GET").arg(key), None).await?.to()?;
-                client.send(cmd("DEL").arg(key), None).await?;
+                let _: f64 = client.send(cmd("GET").arg(key), None).await?;
+                client.send::<u32>(cmd("DEL").arg(key), None).await?;
 
                 Ok::<_, Error>(())
             })

benches/parse_integer.rs

@@ -0,0 +1,88 @@
+use criterion::{Criterion, criterion_group, criterion_main};
+use memchr::memchr;
+use std::hint::black_box;
+
+#[derive(Debug)]
+pub enum Error {
+    EOF,
+    CannotParseInteger,
+}
+
+// --- VERSION 1 : MEMCHR + ATOI ---
+fn parse_integer_atoi(buf: &[u8], pos: &mut usize) -> Result<i64, Error> {
+    let rem = &buf[*pos..];
+    let i = memchr(b'\r', rem).ok_or(Error::EOF)?;
+    if i + 1 >= rem.len() || rem[i + 1] != b'\n' {
+        return Err(Error::EOF);
+    }
+
+    let val = atoi::atoi::<i64>(&rem[..i]).ok_or(Error::CannotParseInteger)?;
+
+    *pos += i + 2;
+    Ok(val)
+}
+
+// --- VERSION 2 : MANUAL (*10) ---
+fn parse_integer_manual(buf: &[u8], pos: &mut usize) -> Result<i64, Error> {
+    let mut n = 0i64;
+    let b_slice = &buf[*pos..];
+
+    for (i, &b) in b_slice.iter().enumerate() {
+        match b {
+            b'0'..=b'9' => {
+                n = n.wrapping_mul(10).wrapping_add((b - b'0') as i64);
+            }
+            b'\r' => {
+                if i + 1 < b_slice.len() && b_slice[i + 1] == b'\n' {
+                    *pos += i + 2;
+                    return Ok(n);
+                }
+                return Err(Error::CannotParseInteger);
+            }
+            _ => return Err(Error::CannotParseInteger),
+        }
+    }
+    Err(Error::EOF)
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    // On teste sur un petit entier (cas typique des longueurs RESP)
+    let data_small = b"12\r\n";
+    // On teste sur un grand entier
+    let data_large = b"1234567890\r\n";
+
+    let mut group = c.benchmark_group("Integer Parsing");
+
+    group.bench_function("atoi_small", |b| {
+        b.iter(|| {
+            let mut pos = 0;
+            let _ = parse_integer_atoi(black_box(data_small), &mut pos);
+        })
+    });
+
+    group.bench_function("manual_small", |b| {
+        b.iter(|| {
+            let mut pos = 0;
+            let _ = parse_integer_manual(black_box(data_small), &mut pos);
+        })
+    });
+
+    group.bench_function("atoi_large", |b| {
+        b.iter(|| {
+            let mut pos = 0;
+            let _ = parse_integer_atoi(black_box(data_large), &mut pos);
+        })
+    });
+
+    group.bench_function("manual_large", |b| {
+        b.iter(|| {
+            let mut pos = 0;
+            let _ = parse_integer_manual(black_box(data_large), &mut pos);
+        })
+    });
+
+    group.finish();
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);

examples/resp_profiling.rs

@@ -1,23 +1,25 @@
-use rustis::resp::RespDeserializer;
-use serde::Deserialize;
+// use rustis::resp::RespDeserializer;
+// use serde::Deserialize;
 
 fn main() -> rustis::Result<()> {
-    let mut raw_data = Vec::new();
-    raw_data.extend_from_slice(b"$1024\r\n");
-    raw_data.extend_from_slice(&vec![b'A'; 1024]);
-    raw_data.extend_from_slice(b"\r\n");
+    // let mut raw_data = Vec::new();
+    // raw_data.extend_from_slice(b"$1024\r\n");
+    // raw_data.extend_from_slice(&vec![b'A'; 1024]);
+    // raw_data.extend_from_slice(b"\r\n");
 
-    println!("desrializer stress-test startup...");
+    // println!("desrializer stress-test startup...");
 
-    for i in 0..500_000 {
-        let mut resp_deserializer = RespDeserializer::new(&raw_data);
-        let result = String::deserialize(&mut resp_deserializer);
-        let _ = std::hint::black_box(result);
+    // for i in 0..500_000 {
+    //     let (frame, _) = RespFrameParser::new(&raw_data).parse()?;
+    //     let response = RespResponse::new(RespBuf::from(Bytes::copy_from_slice(buf)), frame);
+    //     let mut resp_deserializer = RespDeserializer::new(&raw_data);
+    //     let result = String::deserialize(&mut resp_deserializer);
+    //     let _ = std::hint::black_box(result);
 
-        if i % 100_000 == 0 {
-            println!("Processed {} iterations", i);
-        }
-    }
+    //     if i % 100_000 == 0 {
+    //         println!("Processed {} iterations", i);
+    //     }
+    // }
 
     Ok(())
 }

run_tests.sh

@@ -0,0 +1 @@
+cargo test --features tokio-rustls,pool,json,client-cache -- --test-threads=1

src/cache.rs

@@ -8,20 +8,20 @@ use crate::{
         StringCommands, ZRangeOptions,
     },
     resp::{
-        BulkString, Command, CommandArgsMut, RespBuf, RespDeserializer, RespSerializer, Response,
-        Value, cmd,
+        BulkString, Command, CommandArgsMut, FastPathCommandBuilder, RespDeserializer,
+        RespResponse, Response,
     },
 };
-use bytes::BytesMut;
+use bytes::Bytes;
 use dashmap::DashMap;
 use futures_util::StreamExt;
-use serde::{Deserialize, Serialize, de::DeserializeOwned};
-use std::{fmt::Write, sync::Arc, time::Duration};
+use serde::{Serialize, de::DeserializeOwned};
+use std::{sync::Arc, time::Duration};
 
 /// Re-export the moka cache builder.
 pub use moka::future::CacheBuilder;
 
-type SubCache = DashMap<Command, RespBuf>;
+type SubCache = DashMap<Bytes, RespResponse>;
 type MokaCache = moka::future::Cache<BulkString, Arc<SubCache>>;
 type MokaCacheBuilder = moka::future::CacheBuilder<BulkString, Arc<SubCache>, MokaCache>;
 
@@ -134,65 +134,78 @@ impl Cache {
     /// Executes the `MGET` command with client-side caching.
     pub async fn mget<R: Response + DeserializeOwned>(&self, keys: impl Serialize) -> Result<R> {
         let prepared_command = self.client.mget::<R>(keys);
-        let mut collection_buf = BytesMut::new();
-        let _ =
-            collection_buf.write_fmt(format_args!("*{}\r\n", prepared_command.command.num_args()));
-
-        for arg in (0..prepared_command.command.num_args())
-            .filter_map(|i| prepared_command.command.get_arg(i))
-        {
-            let key = BulkString::from(arg.to_vec());
-
-            let Some(values) = self.cache.get(&key).await else {
-                collection_buf.clear();
-                break;
-            };
-
-            let prepared_command = self.client.get::<R>(arg);
-            let Some(buf) = values.get(&prepared_command.command) else {
-                collection_buf.clear();
-                break;
-            };
-
-            collection_buf.extend(buf.iter());
+        let mut responses = Vec::with_capacity(prepared_command.command.num_args());
+        let mut missing_indices = Vec::new();
+        let mut missing_keys = Vec::new();
+
+        // 1. check cache
+        for (i, arg) in prepared_command.command.args().enumerate() {
+            let key = BulkString::from(arg.clone());
+
+            if let Some(values) = self.cache.get(&key).await
+                && let Some(response) = values.get(FastPathCommandBuilder::get(key.clone()).bytes())
+            {
+                log::debug!(
+                    "[{}] Cache hit on key `{}`",
+                    self.client.connection_tag(),
+                    key
+                );
+                responses.push(response.clone());
+            } else {
+                log::debug!(
+                    "[{}] Cache miss on key `{}`",
+                    self.client.connection_tag(),
+                    key
+                );
+                responses.push(RespResponse::null());
+                missing_indices.push(i);
+                missing_keys.push(key);
+            }
         }
 
-        if !collection_buf.is_empty() {
-            log::debug!("[{}] Cache hit on mget", self.client.connection_tag(),);
+        // 2. Fetch missing keys from Redis server if any
+        if !missing_keys.is_empty() {
+            let missing_prepared_command = self.client.mget::<R>(missing_keys);
+            let response = self
+                .client
+                .internal_send(missing_prepared_command.command, None)
+                .await?;
+            let Ok(array_iter) = response.clone().into_array_iter() else {
+                return Err(Error::Client(ClientError::ExpectedArrayForMGet));
+            };
 
-            let mut deserializer = RespDeserializer::new(&collection_buf);
-            return R::deserialize(&mut deserializer);
-        }
+            for (idx_in_missing, response) in array_iter.enumerate() {
+                let original_idx = missing_indices[idx_in_missing];
+
+                let Some(key) = prepared_command
+                    .command
+                    .get_arg(original_idx)
+                    .map(BulkString::from)
+                else {
+                    break;
+                };
+
+                // Insert into cache
+                self.cache
+                    .entry(key.clone())
+                    .or_insert_with(async { Arc::new(DashMap::new()) })
+                    .await
+                    .value()
+                    .insert(
+                        FastPathCommandBuilder::get(key).bytes().clone(),
+                        response.clone(),
+                    );
 
-        let buf = self
-            .client
-            .send(prepared_command.command.clone(), None)
-            .await?;
-        let mut deserializer = RespDeserializer::new(&buf);
-        let Value::Array(values) = Value::deserialize(&mut deserializer)? else {
-            return Err(Error::Client(ClientError::ExpectedArrayForMGet));
-        };
-
-        for (value, key) in values.iter().zip(
-            (0..prepared_command.command.num_args())
-                .filter_map(|i| prepared_command.command.get_arg(i)),
-        ) {
-            let mut serializer = RespSerializer::new();
-            value.serialize(&mut serializer)?;
-
-            // Insert into cache
-            self.cache
-                .entry(key.to_vec().into())
-                .or_insert_with(async { Arc::new(DashMap::new()) })
-                .await
-                .value()
-                .insert(
-                    cmd("GET").arg(key).into(),
-                    RespBuf::new(serializer.get_output().into()),
-                );
+                responses[original_idx] = response;
+            }
+        } else {
+            log::debug!("[{}] Cache hit on mget", self.client.connection_tag());
         }
 
-        R::deserialize(&Value::Array(values))
+        // 3. deserialize
+        let response = RespResponse::owned_array(responses);
+        let deserializer = RespDeserializer::new(response.view());
+        R::deserialize(deserializer)
     }
 
     /// Executes the `GETRANGE` command with client-side caching.
@@ -458,15 +471,15 @@ impl Cache {
         R: Response + DeserializeOwned,
     {
         if let Some(values) = self.cache.get(&key).await
-            && let Some(buf) = values.get(&command)
+            && let Some(response) = values.get(command.bytes())
         {
             log::debug!(
                 "[{}] Cache hit on key `{}`",
                 self.client.connection_tag(),
                 key
             );
-            let mut deserializer = RespDeserializer::new(&buf);
-            return R::deserialize(&mut deserializer);
+            let deserializer = RespDeserializer::new(response.view());
+            return R::deserialize(deserializer);
         }
 
         // Cache miss: fetch from Redis
@@ -476,17 +489,18 @@ impl Cache {
             key
         );
 
-        let buf = self.client.send(command.clone(), None).await?;
-        let mut deserializer = RespDeserializer::new(&buf);
-        let deserialized = R::deserialize(&mut deserializer)?;
+        let command_bytes = command.bytes().clone();
+        let response = self.client.internal_send(command, None).await?;
+        let deserializer = RespDeserializer::new(response.view());
+        let deserialized = R::deserialize(deserializer)?;
 
         // Insert into cache
         self.cache
             .entry(key)
             .or_insert_with(async { Arc::new(DashMap::new()) })
             .await
             .value()
-            .insert(command, buf);
+            .insert(command_bytes, response);
 
         Ok(deserialized)
     }