Add benchmarking tests and integrate it with CI/CD

Signed-off-by: Pavel Abramov <[email protected]>

.github/workflows/rust.yml

@@ -23,3 +23,18 @@ jobs:
       run: cargo build --verbose
     - name: Run tests
       run: cargo test --verbose
+
+    - name: Run benchmarks
+      run: cargo bench -- --output-format json > benchmark_report.json
+
+    - name: Install criterion-json-to-markdown
+      run: cargo install criterion-json-to-markdown
+
+    - name: Convert JSON to Markdown
+      run: criterion-json-to-markdown benchmark_report.json > benchmark_report.md
+
+    - name: Upload benchmark report
+      uses: actions/upload-artifact@v3
+      with:
+        name: benchmark-report
+        path: benchmark_report.md

Cargo.toml

@@ -15,6 +15,9 @@ tempfile = "3.10.1"
 tokio = { version = "1.39.1", features = ["full"] }
 uuid = { version = "1.4", features = ["v4"] }
 
+[dev-dependencies]
+criterion = { version = "0.4", features = ["async_tokio", "html_reports"] }
+
 [lib]
 name = "nkv"
 path = "src/lib.rs"
@@ -27,6 +30,10 @@ path = "src/server/main.rs"
 name = "nkv-client"
 path = "src/client/main.rs"
 
+[[bench]]
+name = "nkv_bench"
+harness = false
+
 [profile.release]
 strip = true  # Automatically strip symbols from the binary
 opt-level = "z"  # Optimize for size.

benches/nkv_bench.rs

@@ -0,0 +1,238 @@
+use criterion::{black_box, criterion_group, criterion_main, BatchSize, Criterion};
+use nkv::nkv::NotifyKeyValue;
+use nkv::srv::{BaseMsg, GetMsg, PutMsg, Server};
+use tempfile::TempDir;
+use tokio::runtime::Runtime;
+use tokio::sync::mpsc;
+
+pub fn bench_nkv(c: &mut Criterion) {
+    let mut group = c.benchmark_group("nkv_group");
+
+    // Create a Tokio runtime
+    let rt = Runtime::new().unwrap();
+
+    // Define different input sizes to test
+    let input_sizes = [1024, 2048, 4096, 8192, 16384];
+
+    group.bench_function(format!("nkv_new"), |b| {
+        b.to_async(&rt).iter(|| async {
+            let temp_dir = TempDir::new().unwrap();
+            let result = NotifyKeyValue::new(temp_dir.path().to_path_buf());
+            black_box(result)
+        })
+    });
+
+    for &size in &input_sizes {
+        group.bench_function(format!("nkv_single_put_size_{}", size), |b| {
+            b.to_async(&rt).iter_batched(
+                || {
+                    // Setup code: create a Box<[u8]> with the given size
+                    let data = vec![0u8; size].into_boxed_slice();
+                    data
+                },
+                |input| async {
+                    let temp_dir = TempDir::new().unwrap();
+                    let mut nkv = NotifyKeyValue::new(temp_dir.path().to_path_buf()).unwrap();
+                    let result = nkv.put("key1", input).await;
+                    black_box(result)
+                },
+                BatchSize::SmallInput,
+            );
+        });
+
+        group.bench_function(format!("nkv_single_update_size_{}", size), |b| {
+            b.to_async(&rt).iter_batched(
+                || {
+                    // Setup code: create a Box<[u8]> with the given size
+                    let data = vec![0u8; size].into_boxed_slice();
+                    let new_data = vec![6u8; size].into_boxed_slice();
+                    (data, new_data)
+                },
+                |(data, new_data)| async {
+                    let temp_dir = TempDir::new().unwrap();
+                    let mut nkv = NotifyKeyValue::new(temp_dir.path().to_path_buf()).unwrap();
+                    nkv.put("key1", data).await;
+                    let result = nkv.put("key1", new_data).await;
+                    black_box(result)
+                },
+                BatchSize::SmallInput,
+            );
+        });
+
+        group.bench_function(format!("nkv_single_get_size{}", size), |b| {
+            b.iter_batched(
+                || {
+                    let data = vec![0u8; size].into_boxed_slice();
+                    let temp_dir = TempDir::new().unwrap();
+                    let mut nkv = NotifyKeyValue::new(temp_dir.path().to_path_buf()).unwrap();
+                    let rt = Runtime::new().unwrap();
+                    rt.block_on(nkv.put("key1", data));
+                    nkv
+                },
+                |nkv| black_box(nkv.get("key1")),
+                BatchSize::SmallInput, // Adjust based on expected input size
+            );
+        });
+
+        group.bench_function(format!("nkv_single_delete_size{}", size), |b| {
+            b.to_async(&rt).iter_batched(
+                || {
+                    let data = vec![0u8; size].into_boxed_slice();
+                    data
+                },
+                |data| async {
+                    let temp_dir = TempDir::new().unwrap();
+                    let mut nkv = NotifyKeyValue::new(temp_dir.path().to_path_buf()).unwrap();
+
+                    nkv.put("key1", data).await;
+                    let result = nkv.delete("key1").await;
+                    black_box(result)
+                },
+                BatchSize::SmallInput, // Adjust based on expected input size
+            );
+        });
+    }
+
+    group.finish();
+}
+
+fn bench_server(c: &mut Criterion) {
+    let mut group = c.benchmark_group("server_group");
+
+    let rt = Runtime::new().unwrap();
+    group.bench_function(format!("nkv_new"), |b| {
+        b.to_async(&rt).iter(|| async {
+            let temp_dir = TempDir::new().expect("Failed to create temporary directory");
+            let url = "127.0.0.1:8091";
+
+            let result = Server::new(url.to_string(), temp_dir.path().to_path_buf()).await;
+            black_box(result)
+        })
+    });
+
+    let input_sizes = [1024, 2048, 4096, 8192, 16384];
+
+    for &size in &input_sizes {
+        group.bench_function(format!("server_put_size_{}", size), |b| {
+            b.to_async(&rt).iter_batched(
+                || {
+                    // Setup code: create a Box<[u8]> with the given size
+                    let data = vec![0u8; size].into_boxed_slice();
+                    data
+                },
+                |input| async {
+                    let temp_dir = TempDir::new().expect("Failed to create temporary directory");
+                    // not used with channels
+                    let url = "127.0.0.1:8091";
+
+                    let srv = Server::new(url.to_string(), temp_dir.path().to_path_buf())
+                        .await
+                        .unwrap();
+
+                    let put_tx = srv.put_tx();
+
+                    let key = "key1".to_string();
+                    let (resp_tx, mut resp_rx) = mpsc::channel(1);
+
+                    let _ = put_tx.send(PutMsg {
+                        key,
+                        value: input,
+                        resp_tx,
+                    });
+
+                    let result = resp_rx.recv().await;
+
+                    black_box(result)
+                },
+                BatchSize::SmallInput,
+            );
+        });
+
+        group.bench_function(format!("server_put_get_size_{}", size), |b| {
+            b.to_async(&rt).iter_batched(
+                || {
+                    // Setup code: create a Box<[u8]> with the given size
+                    let data = vec![0u8; size].into_boxed_slice();
+                    data
+                },
+                |input| async {
+                    let temp_dir = TempDir::new().expect("Failed to create temporary directory");
+                    // not used with channels
+                    let url = "127.0.0.1:8091";
+
+                    let srv = Server::new(url.to_string(), temp_dir.path().to_path_buf())
+                        .await
+                        .unwrap();
+
+                    let put_tx = srv.put_tx();
+                    let get_tx = srv.get_tx();
+
+                    let (resp_tx, mut resp_rx) = mpsc::channel(1);
+
+                    let _ = put_tx.send(PutMsg {
+                        key: "key1".to_string(),
+                        value: input,
+                        resp_tx,
+                    });
+
+                    let _ = resp_rx.recv().await.unwrap();
+
+                    let (get_resp_tx, mut get_resp_rx) = mpsc::channel(1);
+                    let _ = get_tx.send(GetMsg {
+                        key: "key1".to_string(),
+                        resp_tx: get_resp_tx,
+                    });
+                    let result = get_resp_rx.recv().await.unwrap();
+
+                    black_box(result)
+                },
+                BatchSize::SmallInput,
+            );
+        });
+
+        group.bench_function(format!("server_put_delete_size_{}", size), |b| {
+            b.to_async(&rt).iter_batched(
+                || {
+                    // Setup code: create a Box<[u8]> with the given size
+                    let data = vec![0u8; size].into_boxed_slice();
+                    data
+                },
+                |input| async {
+                    let temp_dir = TempDir::new().expect("Failed to create temporary directory");
+                    // not used with channels
+                    let url = "127.0.0.1:8091";
+
+                    let srv = Server::new(url.to_string(), temp_dir.path().to_path_buf())
+                        .await
+                        .unwrap();
+
+                    let put_tx = srv.put_tx();
+                    let del_tx = srv.del_tx();
+
+                    let (resp_tx, mut resp_rx) = mpsc::channel(1);
+
+                    let _ = put_tx.send(PutMsg {
+                        key: "key1".to_string(),
+                        value: input,
+                        resp_tx,
+                    });
+
+                    let _ = resp_rx.recv().await.unwrap();
+
+                    let (del_resp_tx, mut del_resp_rx) = mpsc::channel(1);
+                    let _ = del_tx.send(BaseMsg {
+                        key: "key1".to_string(),
+                        resp_tx: del_resp_tx,
+                    });
+                    let result = del_resp_rx.recv().await.unwrap();
+
+                    black_box(result)
+                },
+                BatchSize::SmallInput,
+            );
+        });
+    }
+}
+
+criterion_group!(benches, bench_nkv, bench_server);
+criterion_main!(benches);

src/lib.rs

@@ -8,6 +8,7 @@ pub mod nkv;
 pub mod notifier;
 mod persist_value;
 pub mod request_msg;
+pub mod srv;
 
 use crate::notifier::{Message, Subscriber};
 use crate::request_msg::*;

src/server/main.rs

@@ -1,7 +1,7 @@
 use std::env;
 use tempfile::TempDir;
 
-mod srv;
+use nkv::srv;
 
 const DEFAULT_URL: &str = "127.0.0.1:8091";
 

src/server/srv.rs → src/srv.rs

@@ -9,17 +9,17 @@ use http::StatusCode;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use tokio::io::{split, AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
-use tokio::net::{TcpListener, TcpStream};
+use tokio::net::TcpListener;
 use tokio::sync::mpsc;
 
-use ::nkv::nkv;
-use ::nkv::notifier::WriteStream;
-use ::nkv::request_msg::{self, BaseMessage, PutMessage, ServerRequest, ServerResponse};
+use crate::nkv;
+use crate::notifier::WriteStream;
+use crate::request_msg::{self, BaseMessage, PutMessage, ServerRequest, ServerResponse};
 
 pub struct PutMsg {
-    key: String,
-    value: Box<[u8]>,
-    resp_tx: mpsc::Sender<nkv::NotifyKeyValueError>,
+    pub key: String,
+    pub value: Box<[u8]>,
+    pub resp_tx: mpsc::Sender<nkv::NotifyKeyValueError>,
 }
 
 pub struct NkvGetResp {
@@ -28,13 +28,13 @@ pub struct NkvGetResp {
 }
 
 pub struct GetMsg {
-    key: String,
-    resp_tx: mpsc::Sender<NkvGetResp>,
+    pub key: String,
+    pub resp_tx: mpsc::Sender<NkvGetResp>,
 }
 
 pub struct BaseMsg {
-    key: String,
-    resp_tx: mpsc::Sender<nkv::NotifyKeyValueError>,
+    pub key: String,
+    pub resp_tx: mpsc::Sender<nkv::NotifyKeyValueError>,
 }
 
 pub struct SubMsg {
@@ -321,10 +321,10 @@ impl Server {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use ::nkv::notifier::Message;
-    use ::nkv::NkvClient;
+    use crate::notifier::Message;
+    use crate::NkvClient;
     use tempfile::TempDir;
-    use tokio;
+    use tokio::{self, net::TcpStream};
 
     #[tokio::test]
     async fn test_server() {