Background compaction in a thread pool

.github/workflows/rust.yml

@@ -17,9 +17,9 @@ jobs:
     steps:
     - uses: actions/checkout@v4
     - name: Build
-      run: cargo build --verbose
+      run: cargo build
     - name: Run tests
-      run: cargo test --release --verbose -- --nocapture
+      run: cargo test --release -- --nocapture
     - name: Run simple example
       run: cargo run --example simple
     - name: Run multithreaded example

Cargo.toml

@@ -19,6 +19,7 @@ uuid = { version = "1.10.0" }
 rand = "0.8.5"
 fslock = "0.2.1"
 libc = "0.2.158"
+crossbeam-channel = "0.5.13"
 
 [features]
 whitebox_testing = []

candy-crasher/src/main.rs

@@ -15,19 +15,15 @@ const CONFIG: Config = Config {
     truncate_up: true,
     clear_on_unsupported_version: true,
     mlock_headers: false,
+    num_compaction_threads: 4,
 };
 
 fn child_inserts() -> Result<()> {
     // our job is to create 1M entries while being killed by our evil parent
 
     let store = CandyStore::open("dbdir", CONFIG)?;
     let highest_bytes = store.get("highest")?.unwrap_or(vec![0, 0, 0, 0]);
-    let highest = u32::from_le_bytes([
-        highest_bytes[0],
-        highest_bytes[1],
-        highest_bytes[2],
-        highest_bytes[3],
-    ]);
+    let highest = u32::from_le_bytes(highest_bytes.try_into().unwrap());
 
     if highest == TARGET - 1 {
         println!("child finished (already at {highest})");
@@ -50,12 +46,7 @@ fn child_removals() -> Result<()> {
 
     let store = CandyStore::open("dbdir", CONFIG)?;
     let lowest_bytes = store.get("lowest")?.unwrap_or(vec![0, 0, 0, 0]);
-    let lowest = u32::from_le_bytes([
-        lowest_bytes[0],
-        lowest_bytes[1],
-        lowest_bytes[2],
-        lowest_bytes[3],
-    ]);
+    let lowest = u32::from_le_bytes(lowest_bytes.try_into().unwrap());
 
     if lowest == TARGET - 1 {
         println!("child finished (already at {lowest})");
@@ -79,12 +70,7 @@ fn child_list_inserts() -> Result<()> {
     let store = CandyStore::open("dbdir", CONFIG)?;
 
     let highest_bytes = store.get("list_highest")?.unwrap_or(vec![0, 0, 0, 0]);
-    let highest = u32::from_le_bytes([
-        highest_bytes[0],
-        highest_bytes[1],
-        highest_bytes[2],
-        highest_bytes[3],
-    ]);
+    let highest = u32::from_le_bytes(highest_bytes.try_into().unwrap());
 
     if highest == TARGET - 1 {
         println!("child finished (already at {highest})");
@@ -108,12 +94,7 @@ fn child_list_removals() -> Result<()> {
     let store = CandyStore::open("dbdir", CONFIG)?;
 
     let lowest_bytes = store.get("list_lowest")?.unwrap_or(vec![0, 0, 0, 0]);
-    let lowest = u32::from_le_bytes([
-        lowest_bytes[0],
-        lowest_bytes[1],
-        lowest_bytes[2],
-        lowest_bytes[3],
-    ]);
+    let lowest = u32::from_le_bytes(lowest_bytes.try_into().unwrap());
 
     if lowest == TARGET - 1 {
         println!("child finished (already at {lowest})");
@@ -258,6 +239,7 @@ fn main() -> Result<()> {
     //     "dbdir",
     //     Config {
     //         expected_number_of_keys: 1_000_000,
+    //         clear_on_unsupported_version: true,
     //         ..Default::default()
     //     },
     // )?;
@@ -277,7 +259,7 @@ fn main() -> Result<()> {
         for res in store.iter() {
             let (k, v) = res?;
             assert_eq!(v, b"i am a key");
-            let k = u32::from_le_bytes([k[0], k[1], k[2], k[3]]);
+            let k = u32::from_le_bytes(k.try_into().unwrap());
             assert!(k < TARGET);
             count += 1;
         }
@@ -296,7 +278,12 @@ fn main() -> Result<()> {
             store.remove("lowest")?,
             Some((TARGET - 1).to_le_bytes().to_vec())
         );
-        assert_eq!(store.iter().count(), 0);
+        assert_eq!(
+            store.iter().count(),
+            0,
+            "{:?}",
+            store.iter().collect::<Vec<_>>()
+        );
 
         println!("DB validated successfully");
     }
@@ -314,14 +301,14 @@ fn main() -> Result<()> {
 
         for (i, res) in store.iter_list("xxx").enumerate() {
             let (k, v) = res?;
-            assert_eq!(k, (i as u32).to_le_bytes());
+            assert_eq!(u32::from_le_bytes(k.try_into().unwrap()), i as u32);
             assert_eq!(v, b"yyy");
         }
 
         println!("DB validated successfully");
     }
 
-    parent_run(shared_stuff, child_list_removals, 10..30)?;
+    parent_run(shared_stuff, child_list_removals, 10..80)?;
 
     {
         println!("Parent starts validating the DB...");

candy-longliving/src/main.rs

@@ -16,7 +16,13 @@ fn main() -> Result<()> {
     let num_iters: usize = args[2].parse().expect("num_iters not a number");
     let tail_length: usize = args[3].parse().expect("tail_length not a number");
 
-    let db = Arc::new(CandyStore::open("dbdir", Config::default())?);
+    let db = Arc::new(CandyStore::open(
+        "dbdir",
+        Config {
+            min_compaction_threashold: 1024 * 1024,
+            ..Default::default()
+        },
+    )?);
     db.clear()?;
 
     let mut handles = vec![];
@@ -35,7 +41,7 @@ fn main() -> Result<()> {
                 if i % 10000 == 0 {
                     let t1 = Instant::now();
                     println!(
-                        "thread {thd} at {i} {:?} rate={}us",
+                        "thread {thd} at {i} {} rate={}us",
                         db.stats(),
                         t1.duration_since(t0).as_micros() / 10_000,
                     );

candy-perf/src/main.rs

@@ -245,7 +245,8 @@ fn test_concurrency_without_contention(num_threads: u32, num_keys: u32) -> Resul
                 {
                     let t0 = Instant::now();
                     for i in thd * num_keys..(thd + 1) * num_keys {
-                        db.set(&i.to_le_bytes(), &thd.to_le_bytes())?;
+                        let status = db.set(&i.to_le_bytes(), &thd.to_le_bytes())?;
+                        debug_assert!(status.was_created());
                     }
                     insert_time_ns.fetch_add(
                         Instant::now().duration_since(t0).as_nanos() as u64,
@@ -257,7 +258,7 @@ fn test_concurrency_without_contention(num_threads: u32, num_keys: u32) -> Resul
                     let t0 = Instant::now();
                     for i in thd * num_keys..(thd + 1) * num_keys {
                         let val = db.get(&i.to_le_bytes())?;
-                        debug_assert_eq!(val, Some(thd.to_le_bytes().to_vec()));
+                        debug_assert_eq!(val, Some(thd.to_le_bytes().to_vec()), "thd={thd} i={i}");
                         black_box(val.unwrap());
                     }
                     get_time_ns.fetch_add(

src/hashing.rs

@@ -52,6 +52,7 @@ impl PartedHash {
         self.0 as u32
     }
 
+    #[allow(dead_code)]
     pub fn as_u64(&self) -> u64 {
         self.0
     }

src/lib.rs

@@ -70,17 +70,13 @@ pub enum CandyError {
     KeyTooLong(usize),
     ValueTooLong(usize),
     EntryCannotFitInShard(usize, usize),
-    KeyAlreadyExists(Vec<u8>, u64),
 }
 
 impl Display for CandyError {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         match self {
             Self::KeyTooLong(sz) => write!(f, "key too long {sz}"),
             Self::ValueTooLong(sz) => write!(f, "value too long {sz}"),
-            Self::KeyAlreadyExists(key, ph) => {
-                write!(f, "key {key:?} already exists (0x{ph:016x})")
-            }
             Self::EntryCannotFitInShard(sz, max) => {
                 write!(f, "entry too big ({sz}) for a single shard file ({max})")
             }
@@ -111,6 +107,8 @@ pub struct Config {
     pub clear_on_unsupported_version: bool,
     /// whether or not to mlock the shard headers to RAM (POSIX only)
     pub mlock_headers: bool,
+    /// number of background compaction threads
+    pub num_compaction_threads: usize,
     /// optionally delay modifying operations before for the given duration before flushing data to disk,
     /// to ensure reboot consistency
     #[cfg(feature = "flush_aggregation")]
@@ -128,6 +126,7 @@ impl Default for Config {
             truncate_up: true,
             clear_on_unsupported_version: false,
             mlock_headers: false,
+            num_compaction_threads: 4,
             #[cfg(feature = "flush_aggregation")]
             flush_aggregation_delay: None,
         }

src/lists.rs

@@ -174,7 +174,6 @@ impl CandyStore {
         // if the item already exists, it's already part of the list. just update it and preserve the index
         if let Some(mut existing_val) = self.get_raw(&item_key)? {
             match mode {
-                InsertMode::MustCreate => unreachable!(),
                 InsertMode::GetOrCreate => {
                     existing_val.truncate(existing_val.len() - size_of::<u64>());
                     return Ok(InsertToListStatus::ExistingValue(existing_val));