Add more extensive tests (#45)

.github/workflows/ci.yml

@@ -48,6 +48,23 @@ jobs:
     - name: "Test: Wisckey and Monoio"
       run: just monoio-wisckey-tests
       timeout-minutes: 10
+  big-test:
+    runs-on: ubuntu-24.04
+    steps:
+    - uses: actions/checkout@v4
+    - name: Install rustc and clippy nightly
+      uses: dtolnay/rust-toolchain@stable
+      with:
+         toolchain: nightly
+         components: cargo, rustc, clippy, rustfmt
+    - name: Install Just
+      uses: taiki-e/install-action@v2
+      with:
+        tool: just
+    - name: Insert many entries 
+      run: just bigtest-many
+    - name: Insert large entries 
+      run: just bigtest-large
   lint:
     runs-on: ubuntu-24.04
     steps:
@@ -61,7 +78,6 @@ jobs:
       uses: taiki-e/install-action@v2
       with:
         tool: just
-
     - name: "Lint Checks: Tokio (with sync FS)"
       run: just async-lint
     - name: "Lint Checks: Tokio-Uring"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lsm"
-version = "0.4.1"
+version = "0.5.0-dev"
 authors = ["Kai Mast <[email protected]>"]
 edition = "2024"
 repository = "https://github.com/kaimast/lsm-rs"
@@ -26,7 +26,7 @@ tokio-condvar = { version="0.3", features=["parking_lot"] }
 tokio-uring = { version="0.5", optional=true }
 bloomfilter = { version="3", optional=true }
 monoio = { version="0.2", optional=true, features=["sync"] }
-kioto-uring-executor = { version="0.2", optional=true }
+kioto-uring-executor = { version="0.3.0-dev", optional=true }
 bitvec = { version="1", optional=true }
 
 [dependencies.tokio]
@@ -41,7 +41,7 @@ tempfile = "3"
 tracing-tracy = "0.11"
 tokio = { version="1", default-features=false, features=["rt-multi-thread"] }
 tracing-subscriber = { version="0.3", default-features=false }
-rand = "0.8.5"
+rand = "0.8"
 
 [lib]
 path = "src/lib.rs"
@@ -68,4 +68,7 @@ name = "lsm-benchmark"
 path = "benchmarks/async.rs"
 
 [workspace]
-members = ["sync"]
+members = ["sync", "bigtest"]
+
+[patch.crates-io]
+kioto-uring-executor = { git = "https://github.com/kaimast/kioto-uring-executor" }

NEWS

@@ -1,4 +1,5 @@
-UNRELEASED:
+UNRELEASED (0.5):
+    - More extensive testing
     - Add support for monoio
     - Codebase moved from Rust 2021 to 2024
     - Updated bloomfilter to 0.3

README.md

@@ -31,8 +31,8 @@ However, we may add a converter tool in the future or an `endianess` feature fla
 ## Feature Flags
 * `snappy-compression`: Use the [snappy format](https://docs.rs/snap/1.0.5/snap/) to compress data on disk *(enabled by default)*
 * `bloom-filters`: Add bloom filters to data blocks for more efficient searching. *(enabled by default)*
-* `monoio`: Use `monoio` as async runtime I/O instead of `tokio. ` *(disabled by default)*
-* `async-io`: Use `tokio_uring` as async runtime instead of regular `tokio`. *(disabled by default)*
+* `monoio`: Use `monoio` as async runtime I/O instead of `tokio. Note, this will not spawn any additional OS threads. ` *(disabled by default)*
+* `tokio-uring`: Use `tokio_uring` as async runtime (using `toio-uring-executor`) instead of regular `tokio`. *(disabled by default)*
 * `wisckey`: Store keys and values separately. This usually results in higher throughput with slightly higher CPU-usage. *(disabled by default)*
 
 ## Synchronous API

bigtest/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "lsm-bigtest"
+version = "0.5.0-dev"
+edition = "2024"
+authors = ["Kai Mast <[email protected]>"]
+license = "MIT"
+description = "Runs a longer test with lots of data"
+readme = "../README.md"
+
+[dependencies]
+kioto-uring-executor = "0.3.0-dev"
+clap = { version="4", features=["derive"] }
+env_logger = "0.11"
+tempfile = "3"
+rand = "0.8"
+
+[dependencies.lsm]
+path = ".."
+features = ["tokio-uring"]

bigtest/src/main.rs

@@ -0,0 +1,103 @@
+use std::sync::Arc;
+
+use clap::Parser;
+use rand::Rng;
+
+use lsm::{Database, Params, StartMode};
+
+#[derive(Parser)]
+struct Args {
+    #[clap(long, short = 'n', default_value_t = 100_000)]
+    #[clap(help = "The number of insertions per thread")]
+    num_insertions: usize,
+
+    #[clap(long, short = 't', default_value_t = 10)]
+    num_threads: usize,
+
+    #[clap(long, default_value_t = 1_000_000)]
+    key_range: usize,
+
+    #[clap(long, default_value_t = 1024)]
+    entry_size: usize,
+
+    #[clap(long, default_value = "/tmp")]
+    #[clap(
+        help = "Where to create the temporary working directory? Note, this is the parent directoy of the directory not the directoy itself.
+It is recommended to use a tmpfs to not wear out a physical disk"
+    )]
+    workdir_location: String,
+}
+
+#[kioto_uring_executor::main]
+async fn main() {
+    env_logger::init();
+
+    let args = Args::parse();
+
+    if args.num_insertions == 0 {
+        panic!("Need to insert at least one entry");
+    }
+
+    if args.key_range == 0 {
+        panic!("Key range cannot be zero");
+    }
+
+    println!("Creating working directory and empty database");
+    let tmp_dir = tempfile::Builder::new()
+        .prefix("lsm-bigest-")
+        .tempdir_in(args.workdir_location)
+        .expect("Failed to create working directory");
+
+    let mut db_path = tmp_dir.path().to_path_buf();
+    db_path.push("storage.lsm");
+
+    let params = Params {
+        db_path,
+        ..Default::default()
+    };
+
+    let database = Arc::new(
+        Database::new_with_params(StartMode::CreateOrOverride, params)
+            .await
+            .expect("Failed to create database instance"),
+    );
+
+    println!(
+        "Inserting a total of {} entries of size {} across {} threads",
+        args.num_insertions * args.num_threads,
+        args.entry_size,
+        args.num_threads
+    );
+
+    let tasks: Vec<_> = (1..=args.num_threads)
+        .map(|idx| {
+            let database = database.clone();
+            kioto_uring_executor::spawn_with(move || {
+                let mut rng = rand::thread_rng();
+                Box::pin(async move {
+                    for count in 1..=args.num_insertions {
+                        let key_idx = rng.gen_range(0..args.key_range);
+                        let key = format!("key{key_idx}").as_bytes().to_vec();
+
+                        let mut value = vec![0; args.entry_size];
+                        rng.fill(value.as_mut_slice());
+
+                        database.put(key, value).await.expect("Insert failed");
+
+                        if count % 10_000 == 0 {
+                            println!(
+                                "Thread #{idx} inserted {count} entries so far ({}%)",
+                                (count as f64) * 100.0 / (args.num_insertions as f64)
+                            );
+                        }
+                    }
+                    println!("Thread #{idx} is done");
+                })
+            })
+        })
+        .collect();
+
+    for task in tasks {
+        task.join().await;
+    }
+}

justfile

@@ -42,11 +42,13 @@ wisckey-sync-tests:
 
 lint: sync-lint async-lint wisckey-lint \
       wisckey-no-compression-lint tokio-uring-lint \
-      tokio-uring-wisckey-lint monoio-lint monoio-wisckey-lint
+      tokio-uring-wisckey-lint monoio-lint monoio-wisckey-lint \
+      bigtest-lint
 
 fix-formatting:
     cargo fmt
     cd sync && just fix-formatting
+    cd bigtest && cargo fmt
 
 check-formatting:
     cargo fmt --check
@@ -86,3 +88,13 @@ wisckey-no-compression-lint:
 
 tokio-uring-wisckey-lint:
     {{CLIPPY}} --features=tokio-uring,snappy-compression,wisckey -- -D warnings
+
+bigtest-lint:
+    {{CLIPPY}} --package=lsm-bigtest
+
+bigtest-many:
+    cargo run --release --package=lsm-bigtest -- -n100000 --entry-size=1024
+
+bigtest-large:
+    cargo run --release --package=lsm-bigtest -- -n100 --entry-size=100000
+

src/tasks.rs

@@ -184,7 +184,7 @@ impl TaskManager {
         let level_update_cond = Arc::new(UpdateCond::new());
 
         #[cfg(feature = "tokio-uring")]
-        let mut spawn_pos = kioto_uring_executor::new_spawn_ring();
+        let mut sring = kioto_uring_executor::new_spawn_ring();
 
         {
             let memtable_update_cond = Arc::new(UpdateCond::new());
@@ -200,10 +200,9 @@ impl TaskManager {
 
                 cfg_if::cfg_if! {
                     if #[cfg(feature="tokio-uring")] {
-                        unsafe {
-                            kioto_uring_executor::unsafe_spawn_at(spawn_pos.get(), task);
-                            spawn_pos.advance();
-                        }
+                        sring.spawn_with(move || {
+                            Box::pin(task)
+                        });
                     } else if #[cfg(feature="monoio")] {
                         monoio::spawn(task);
                     } else {
@@ -234,10 +233,9 @@ impl TaskManager {
 
                     cfg_if::cfg_if! {
                         if #[cfg(feature="tokio-uring")] {
-                            unsafe {
-                                kioto_uring_executor::unsafe_spawn_at(spawn_pos.get(), task);
-                                spawn_pos.advance();
-                            }
+                                sring.spawn_with(move || {
+                                    Box::pin(task)
+                                });
                         } else if #[cfg(feature="monoio")] {
                             monoio::spawn(task);
                         } else {