Revert "Background compaction in a thread pool"

This reverts commit a4b6682.

Cargo.toml

@@ -19,7 +19,6 @@ 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,15 +15,19 @@ 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.try_into().unwrap());
+    let highest = u32::from_le_bytes([
+        highest_bytes[0],
+        highest_bytes[1],
+        highest_bytes[2],
+        highest_bytes[3],
+    ]);
 
     if highest == TARGET - 1 {
         println!("child finished (already at {highest})");
@@ -46,7 +50,12 @@ 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.try_into().unwrap());
+    let lowest = u32::from_le_bytes([
+        lowest_bytes[0],
+        lowest_bytes[1],
+        lowest_bytes[2],
+        lowest_bytes[3],
+    ]);
 
     if lowest == TARGET - 1 {
         println!("child finished (already at {lowest})");
@@ -70,7 +79,12 @@ 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.try_into().unwrap());
+    let highest = u32::from_le_bytes([
+        highest_bytes[0],
+        highest_bytes[1],
+        highest_bytes[2],
+        highest_bytes[3],
+    ]);
 
     if highest == TARGET - 1 {
         println!("child finished (already at {highest})");
@@ -94,7 +108,12 @@ 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.try_into().unwrap());
+    let lowest = u32::from_le_bytes([
+        lowest_bytes[0],
+        lowest_bytes[1],
+        lowest_bytes[2],
+        lowest_bytes[3],
+    ]);
 
     if lowest == TARGET - 1 {
         println!("child finished (already at {lowest})");
@@ -239,7 +258,6 @@ fn main() -> Result<()> {
     //     "dbdir",
     //     Config {
     //         expected_number_of_keys: 1_000_000,
-    //         clear_on_unsupported_version: true,
     //         ..Default::default()
     //     },
     // )?;
@@ -259,7 +277,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.try_into().unwrap());
+            let k = u32::from_le_bytes([k[0], k[1], k[2], k[3]]);
             assert!(k < TARGET);
             count += 1;
         }
@@ -278,12 +296,7 @@ fn main() -> Result<()> {
             store.remove("lowest")?,
             Some((TARGET - 1).to_le_bytes().to_vec())
         );
-        assert_eq!(
-            store.iter().count(),
-            0,
-            "{:?}",
-            store.iter().collect::<Vec<_>>()
-        );
+        assert_eq!(store.iter().count(), 0);
 
         println!("DB validated successfully");
     }
@@ -301,14 +314,14 @@ fn main() -> Result<()> {
 
         for (i, res) in store.iter_list("xxx").enumerate() {
             let (k, v) = res?;
-            assert_eq!(u32::from_le_bytes(k.try_into().unwrap()), i as u32);
+            assert_eq!(k, (i as u32).to_le_bytes());
             assert_eq!(v, b"yyy");
         }
 
         println!("DB validated successfully");
     }
 
-    parent_run(shared_stuff, child_list_removals, 10..80)?;
+    parent_run(shared_stuff, child_list_removals, 10..30)?;
 
     {
         println!("Parent starts validating the DB...");

candy-longliving/src/main.rs

@@ -35,7 +35,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,
                     );

src/hashing.rs

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

src/lib.rs

@@ -70,13 +70,17 @@ 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})")
             }
@@ -107,8 +111,6 @@ 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")]
@@ -126,7 +128,6 @@ 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,6 +174,7 @@ 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));