Get start_transaction working with spawned thread

Finish getting tests to work with shuttle

Author: anacrolix

.cargo/config.toml

@@ -1,3 +1,3 @@
 [build]
 # Setting cfg here means our IDE and CLI both use the same values.
-#rustflags = "--cfg shuttle"
+#rustflags = "--cfg loom"

Cargo.toml

@@ -43,6 +43,7 @@ tracing = { version = "0.1.40", features = ["log"] }
 twox-hash = { version = "1.6.3", optional = true }
 once_cell = "1.19.0"
 ctx-thread = "0.1.1"
+shuttle = { version = "0.7.1", optional = true }
 
 [target.'cfg(windows)'.dependencies.windows]
 version = "0.52.0"
@@ -66,12 +67,10 @@ test-log = "0.2.14"
 [target.'cfg(loom)'.dev-dependencies]
 loom = "0.7"
 
-[target.'cfg(shuttle)'.dependencies]
-shuttle = "0.7.1"
-
 [features]
 default = []
 testing = ["dep:fdlimit", "dep:rayon", "dep:twox-hash"]
+shuttle = ["dep:shuttle"]
 
 [[bench]]
 name = "possum"

src/concurrency/mod.rs

@@ -1,9 +1,54 @@
 pub(crate) mod sync;
 
-#[cfg(not(shuttle))]
+#[cfg(not(feature = "shuttle"))]
 pub use std::thread;
 
 // This isn't available in loom or shuttle yet. Unfortunately for shuttle it means threads are
 // spawned outside its control, and it doesn't work.
-#[cfg(shuttle)]
+#[cfg(feature = "shuttle")]
 pub use shuttle::thread;
+
+#[cfg(not(feature = "shuttle"))]
+pub(crate) fn run_blocking<F, R>(f: F) -> R
+where
+    F: FnOnce() -> R + Send,
+    R: Send,
+{
+    if false {
+        let (sender, receiver) = std::sync::mpsc::channel();
+        let tx_thread = std::thread::scope(|scope| {
+            scope.spawn(|| {
+                let res = f();
+                sender.send(res).unwrap();
+            });
+            receiver.recv().unwrap()
+        });
+        tx_thread
+    } else {
+        f()
+    }
+}
+
+#[cfg(feature = "shuttle")]
+pub(crate) fn run_blocking<F, R>(f: F) -> R
+    where
+        F: FnOnce() -> R + Send,
+        R: Send,
+{
+    use std::sync::mpsc;
+    let (sender, receiver) = mpsc::channel();
+    let tx_thread = std::thread::scope(|scope| {
+        scope.spawn(||{
+            let res = f();
+            sender.send(res).unwrap();
+        });
+        loop {
+            shuttle::thread::yield_now();
+            match receiver.try_recv() {
+                Err(mpsc::TryRecvError::Empty) => continue,
+                default => return default.unwrap()
+            }
+        }
+    });
+    tx_thread
+}

src/concurrency/sync.rs

@@ -1,9 +1,9 @@
 use crate::StableDeref;
 use std::ops::{Deref, DerefMut};
 
-#[cfg(shuttle)]
+#[cfg(feature = "shuttle")]
 use shuttle::sync;
-#[cfg(not(shuttle))]
+#[cfg(not(feature = "shuttle"))]
 use std::sync;
 
 use sync::Mutex as InnerMutex;

src/exclusive_file.rs

@@ -131,6 +131,6 @@ impl ExclusiveFile {
 
 impl Drop for ExclusiveFile {
     fn drop(&mut self) {
-        debug!("dropping exclusive file {}", self.id.deref());
+        debug!("dropping exclusive file {}", self.id);
     }
 }

src/file_id.rs

@@ -15,14 +15,6 @@ impl Debug for FileId {
     }
 }
 
-impl Deref for FileId {
-    type Target = FileIdInner;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
-
 impl std::str::FromStr for FileId {
     type Err = std::num::ParseIntError;
 

src/handle.rs

@@ -226,7 +226,15 @@ impl Handle {
     ) -> rusqlite::Result<OwnedTx> {
         Ok(self
             .start_transaction(|conn, handle| {
-                let rtx = conn.transaction_with_behavior(behaviour)?;
+                let tx_res = run_blocking(|| {
+                    // We're holding the write lock around the Connection, I think we're safe to
+                    // pass it to another thread. For some reason the return type, the
+                    // rusqlite::Transaction is what spits the dummy. It doesn't implement Send,
+                    // even though it only has a reference to Connection internally. So we put it
+                    // inside CanSend to return it from the thread then pop it out.
+                    conn.transaction_with_behavior(behaviour).map(CanSend)
+                });
+                let rtx = tx_res?.0;
                 Ok(Transaction::new(rtx, handle))
             })?
             .into())
@@ -261,15 +269,6 @@ impl Handle {
         Ok(reader)
     }
 
-    // pub(crate) fn associated_read<'h, H>(handle: H) -> rusqlite::Result<Reader<'h, H>> where H: WithHandle {
-    //     let reader = Reader {
-    //         owned_tx: handle.as_ref().start_deferred_transaction()?,
-    //         handle,
-    //         reads: Default::default(),
-    //     };
-    //     Ok(reader)
-    // }
-
     pub fn read_single(&self, key: &[u8]) -> Result<Option<SnapshotValue<Value>>> {
         let mut reader = self.read()?;
         let Some(value) = reader.add(key)? else {
@@ -454,9 +453,9 @@ impl Handle {
         Ok(())
     }
 
-    pub fn delete_prefix(&self, prefix: &[u8]) -> PubResult<()> {
+    pub fn delete_prefix(&self, prefix: impl AsRef<[u8]>) -> PubResult<()> {
         let mut tx = self.start_deferred_transaction()?;
-        for item in tx.list_items(prefix)? {
+        for item in tx.list_items(prefix.as_ref())? {
             tx.delete_key(&item.key)?;
         }
         tx.commit()?.complete();
@@ -567,3 +566,7 @@ impl AsRef<Handle> for Rc<RwLockReadGuard<'_, Handle>> {
         self.deref()
     }
 }
+
+struct CanSend<T>(T);
+
+unsafe impl<T> Send for CanSend<T> {}

src/testing.rs

@@ -1,7 +1,7 @@
 pub mod torrent_storage;
 
 use std::hash::Hasher;
-use std::io::{copy, SeekFrom, Write};
+use std::io::{BufReader, copy, SeekFrom, Write};
 
 use anyhow::{ensure, Result};
 use rand::Rng;
@@ -90,6 +90,7 @@ pub fn readable_repeated_bytes(byte: u8, limit: usize) -> Vec<u8> {
 pub fn condense_repeated_bytes(r: impl Read) -> (Option<u8>, u64) {
     let mut count = 0;
     let mut byte = None;
+    let r = BufReader::new(r);
     for b in r.bytes() {
         let b = b.unwrap();
         match byte {
@@ -118,12 +119,12 @@ pub fn check_concurrency(
         loom::model(move || f().unwrap());
         Ok(())
     }
-    #[cfg(shuttle)]
+    #[cfg(feature = "shuttle")]
     {
         shuttle::check_random(move || f().unwrap(), iterations_hint);
         Ok(())
     }
-    #[cfg(all(not(loom), not(shuttle)))]
+    #[cfg(all(not(loom), not(feature = "shuttle")))]
     if false {
         for _ in 0..1000 {
             f()?

src/tests.rs

@@ -53,14 +53,16 @@ fn test_inc_array() {
 /// Show that replacing keys doesn't cause a key earlier in the same values file to be punched. This
 /// occurred because there were file_id values in the manifest file that had the wrong type, and so
 /// the query that looked for the starting offset for hole punching would punch out the whole file
-/// thinking it was empty.
+/// thinking it was empty. Note sometimes this test fails and there's extra values files floating
+/// around. I haven't figured out why.
 #[test]
 #[cfg(not(miri))]
 fn test_replace_keys() -> Result<()> {
     check_concurrency(
         || {
             let tempdir = test_tempdir("test_replace_keys")?;
             let handle = Handle::new(tempdir.path.clone())?;
+            handle.delete_prefix("")?;
             let a = "a".as_bytes().to_vec();
             let b = "b".as_bytes().to_vec();
             let block_size: usize = handle.block_size().try_into()?;
@@ -92,9 +94,13 @@ fn test_replace_keys() -> Result<()> {
             // There can be multiple value files if the value puncher is holding onto a file when another
             // write occurs.
             for value_file in values_files {
-                let mut file = File::open(&value_file.path)?;
+                let path = &value_file.path;
+                eprintln!("{:?}", path);
+                let mut file = File::open(path)?;
+                // file.sync_all()?;
                 for region in seekhole::Iter::new(&mut file) {
                     let region = region?;
+                    eprintln!("{:?}", region);
                     if matches!(region.region_type, seekhole::RegionType::Data) {
                         allocated_space += region.length();
                     }

tests/simple_tests.rs

@@ -228,7 +228,7 @@ fn torrent_storage_small() -> Result<()> {
             }
             Ok(())
         },
-        100,
+        1,
     )
 }
 
@@ -244,7 +244,7 @@ fn torrent_storage_big() -> Result<()> {
                 view_snapshot_values: true,
             })
         },
-        100,
+        1,
     )
 }
 
@@ -288,7 +288,7 @@ fn torrent_storage_inner(opts: TorrentStorageOpts) -> Result<()> {
         let piece_data = Arc::clone(&piece_data);
         let start_delay = Duration::from_micros(1000 * (index / 2) as u64);
         let handle = Arc::clone(&handle);
-        join_handles.push(std::thread::spawn(move || -> Result<()> {
+        join_handles.push(thread::spawn(move || -> Result<()> {
             let key = offset_key(offset);
             sleep(start_delay);
             debug!("starting block write");
@@ -481,13 +481,14 @@ fn reads_update_last_used() -> Result<()> {
             let uniform = UniformDuration::new(Duration::from_nanos(0), LAST_USED_RESOLUTION);
             for _ in 0..100 {
                 let dither = uniform.sample(&mut rng);
-                sleep(LAST_USED_RESOLUTION + dither);
+                // This needs to be a real sleep or the timestamps sqlite generates don't progress.
+                std::thread::sleep(LAST_USED_RESOLUTION + dither);
                 let new_read_ts = handle.read_single(&key)?.unwrap().last_used();
                 assert!(new_read_ts > read_ts);
             }
             Ok(())
         },
-        100,
+        10,
     )
 }