Repository: cberner/redb
Pull request #203 (2022-02-28 07:03:08)

Add three levels of durability
The existing tests pass, but I didn't add any tests for the new durability level.

Why kinds of tests would be appropriate?

casey: I'm kind of sad that I removed `commit_helper_inner`, it would have been more fun to add a third layer, `commit_helper_inner_nested`.
casey: Why don't we duplicate a basic test that does a bunch of commits and change it to use eventually durable commits. It won't really test durability guarantees, but it will at least exercise the codepath, and it sounds like more testing infrastructure is needed to really exercise durability guarantees.
cberner: Ya, that seems good for now. I'd suggest `integration_tests::persistence` that tests that data is persisted across closing & re-opening the database
casey: Added a test for eventual persistence, this should be ready to go.

Patch:
diff --git a/src/db.rs b/src/db.rs
index 12f3f2b..29ed868 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -192,6 +192,22 @@ impl DatabaseBuilder {
     }
 }
 
+#[derive(Copy, Clone)]
+pub enum Durability {
+    /// Commits with this durability level will not be persisted to disk unless followed by a
+    /// commit with a higher durability level.
+    ///
+    /// Note: Pages are only freed during commits with higher durability levels. Exclusively using
+    /// this function may result in Error::OutOfSpace.
+    None,
+    /// Commits with this durability level have been queued for persitance to disk, and should be
+    /// persistant some time after [WriteTransaction::commit] returns.
+    Eventual,
+    /// Commits with this durability level are guaranteed to be persistant as soon as
+    /// [WriteTransaction::commit] returns.
+    Immediate,
+}
+
 #[allow(clippy::type_complexity)]
 pub struct WriteTransaction<'a> {
     storage: &'a Storage,
@@ -200,6 +216,7 @@ pub struct WriteTransaction<'a> {
     pending_table_root_changes: RefCell<HashMap<String, Rc<Cell<Option<PageNumber>>>>>,
     open_tables: RefCell<HashMap<String, &'static panic::Location<'static>>>,
     completed: AtomicBool,
+    durability: Durability,
 }
 
 impl<'a> WriteTransaction<'a> {
@@ -213,9 +230,14 @@ impl<'a> WriteTransaction<'a> {
             pending_table_root_changes: RefCell::new(Default::default()),
             open_tables: RefCell::new(Default::default()),
             completed: Default::default(),
+            durability: Durability::Immediate,
         })
     }
 
+    pub fn set_durability(&mut self, durability: Durability) {
+        self.durability = durability;
+    }
+
     /// Open the given table
     ///
     /// The table will be created if it does not exist
@@ -348,17 +370,7 @@ impl<'a> WriteTransaction<'a> {
     }
 
     pub fn commit(self) -> Result {
-        self.commit_helper(false)
-    }
-
-    /// Note: pages are only freed during commit(). So exclusively using this function may result
-    /// in Error::OutOfSpace
-    pub fn non_durable_commit(self) -> Result {
-        self.commit_helper(true)
-    }
-
-    fn commit_helper(self, non_durable: bool) -> Result {
-        match self.commit_helper_inner(non_durable) {
+        match self.commit_inner() {
             Ok(_) => Ok(()),
             Err(err) => match err {
                 // Rollback the transaction if we ran out of space during commit, so that user may
@@ -372,7 +384,7 @@ impl<'a> WriteTransaction<'a> {
         }
     }
 
-    fn commit_helper_inner(&self, non_durable: bool) -> Result {
+    fn commit_inner(&self) -> Result {
         // Update all the table roots in the master table, before committing
         for (name, update) in self.pending_table_root_changes.borrow_mut().drain() {
             let new_root = self.storage.update_table_root(
@@ -383,14 +395,23 @@ impl<'a> WriteTransaction<'a> {
             )?;
             self.root_page.set(Some(new_root));
         }
-        if non_durable {
-            self.storage
-                .non_durable_commit(self.root_page.get(), self.transaction_id)?;
-        } else {
-            self.storage
-                .commit(self.root_page.get(), self.transaction_id)?;
+
+        match self.durability {
+            Durability::None => self
+                .storage
+                .non_durable_commit(self.root_page.get(), self.transaction_id)?,
+            Durability::Eventual => {
+                self.storage
+                    .durable_commit(self.root_page.get(), self.transaction_id, true)?
+            }
+            Durability::Immediate => {
+                self.storage
+                    .durable_commit(self.root_page.get(), self.transaction_id, false)?
+            }
         }
+
         self.completed.store(true, Ordering::SeqCst);
+
         Ok(())
     }
 
diff --git a/src/lib.rs b/src/lib.rs
index a4a2316..3ea6a77 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,6 +1,6 @@
 pub use db::{
-    Database, DatabaseBuilder, MultimapTableDefinition, ReadTransaction, TableDefinition,
-    WriteTransaction,
+    Database, DatabaseBuilder, Durability, MultimapTableDefinition, ReadTransaction,
+    TableDefinition, WriteTransaction,
 };
 pub use error::Error;
 pub use multimap_table::{
diff --git a/src/tree_store/page_store/mmap.rs b/src/tree_store/page_store/mmap.rs
index 2ecb576..b2370f5 100644
--- a/src/tree_store/page_store/mmap.rs
+++ b/src/tree_store/page_store/mmap.rs
@@ -26,11 +26,15 @@ impl Mmap {
         self.inner.len()
     }
 
+    #[cfg(not(target_os = "macos"))]
+    pub(crate) fn flush(&self) -> Result {
+        self.inner.flush()?;
+        Ok(())
+    }
+
     #[cfg(target_os = "macos")]
     pub(crate) fn flush(&self) -> Result {
-        // TODO: It may be unsafe to mix F_BARRIERFSYNC with writes to the mmap.
-        //       Investigate switching to `write()`
-        let code = unsafe { libc::fcntl(self.file.as_raw_fd(), libc::F_BARRIERFSYNC) };
+        let code = unsafe { libc::fcntl(self.file.as_raw_fd(), libc::F_FULLFSYNC) };
         if code == -1 {
             return Err(io::Error::last_os_error().into());
         }
@@ -38,8 +42,18 @@ impl Mmap {
     }
 
     #[cfg(not(target_os = "macos"))]
-    pub(crate) fn flush(&self) -> Result {
-        self.inner.flush()?;
+    pub(crate) fn eventual_flush(&self) -> Result {
+        self.flush()
+    }
+
+    #[cfg(target_os = "macos")]
+    pub(crate) fn eventual_flush(&self) -> Result {
+        // TODO: It may be unsafe to mix F_BARRIERFSYNC with writes to the mmap.
+        //       Investigate switching to `write()`
+        let code = unsafe { libc::fcntl(self.file.as_raw_fd(), libc::F_BARRIERFSYNC) };
+        if code == -1 {
+            return Err(io::Error::last_os_error().into());
+        }
         Ok(())
     }
 
diff --git a/src/tree_store/page_store/page_manager.rs b/src/tree_store/page_store/page_manager.rs
index 86caa35..74fd339 100644
--- a/src/tree_store/page_store/page_manager.rs
+++ b/src/tree_store/page_store/page_manager.rs
@@ -743,7 +743,7 @@ impl TransactionalMemory {
     }
 
     // Commit all outstanding changes and make them visible as the primary
-    pub(crate) fn commit(&self, transaction_id: u64) -> Result {
+    pub(crate) fn commit(&self, transaction_id: u64, eventual: bool) -> Result {
         // All mutable pages must be dropped, this ensures that when a transaction completes
         // no more writes can happen to the pages it allocated. Thus it is safe to make them visible
         // to future read transactions
@@ -756,7 +756,11 @@ impl TransactionalMemory {
         mutator.set_last_committed_transaction_id(transaction_id);
         drop(mutator);
 
-        self.mmap.flush()?;
+        if eventual {
+            self.mmap.eventual_flush()?;
+        } else {
+            self.mmap.flush()?;
+        }
 
         let god_byte = self.mmap.get_memory(0..DB_METAPAGE_SIZE)[GOD_BYTE_OFFSET];
         let mut next = match god_byte & PRIMARY_BIT {
@@ -1068,7 +1072,7 @@ impl Drop for TransactionalMemory {
         // Commit any non-durable transactions that are outstanding
         if self.read_from_secondary.load(Ordering::SeqCst) {
             if let Ok(non_durable_transaction_id) = self.get_last_committed_transaction_id() {
-                if self.commit(non_durable_transaction_id).is_err() {
+                if self.commit(non_durable_transaction_id, false).is_err() {
                     eprintln!(
                         "Failure while finalizing non-durable commit. Database may have rolled back"
                     );
diff --git a/src/tree_store/storage.rs b/src/tree_store/storage.rs
index 0b09615..f29725f 100644
--- a/src/tree_store/storage.rs
+++ b/src/tree_store/storage.rs
@@ -271,7 +271,7 @@ impl Storage {
             let (new_root, _) =
                 make_mut_single_leaf(FREED_TABLE.as_bytes(), &freed_table.to_bytes(), &mem)?;
             mem.set_secondary_root_page(new_root)?;
-            mem.commit(next_transaction_id)?;
+            mem.commit(next_transaction_id, false)?;
             next_transaction_id += 1;
         }
 
@@ -521,10 +521,11 @@ impl Storage {
         self.mem.get_primary_root_page()
     }
 
-    pub(crate) fn commit(
+    pub(crate) fn durable_commit(
         &self,
         mut new_master_root: Option<PageNumber>,
         transaction_id: TransactionId,
+        eventual: bool,
     ) -> Result {
         let oldest_live_read = self
             .live_read_transactions
@@ -555,7 +556,7 @@ impl Storage {
             self.mem.set_secondary_root_page(PageNumber::null())?;
         }
 
-        self.mem.commit(transaction_id)?;
+        self.mem.commit(transaction_id, eventual)?;
         assert_eq!(
             Some(transaction_id),
             self.live_write_transaction.lock().unwrap().take()


Test patch:
diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs
index dd0ad3d..b351b2e 100644
--- a/tests/integration_tests.rs
+++ b/tests/integration_tests.rs
@@ -5,7 +5,8 @@ use tempfile::NamedTempFile;
 use rand::prelude::SliceRandom;
 use rand::Rng;
 use redb::{
-    Database, DatabaseBuilder, Error, MultimapTableDefinition, ReadableTable, TableDefinition,
+    Database, DatabaseBuilder, Durability, Error, MultimapTableDefinition, ReadableTable,
+    TableDefinition,
 };
 
 const ELEMENTS: usize = 100;
@@ -33,11 +34,12 @@ fn mixed_durable_commit() {
 
     let db_size = 129 * 4096;
     let db = unsafe { Database::create(tmpfile.path(), db_size).unwrap() };
-    let txn = db.begin_write().unwrap();
+    let mut txn = db.begin_write().unwrap();
+    txn.set_durability(Durability::None);
     let mut table = txn.open_table(U64_TABLE).unwrap();
 
     table.insert(&0, &0).unwrap();
-    txn.non_durable_commit().unwrap();
+    txn.commit().unwrap();
 
     let txn = db.begin_write().unwrap();
     txn.commit().unwrap();
@@ -49,7 +51,8 @@ fn non_durable_commit_persistence() {
 
     let db_size = 16 * 1024 * 1024;
     let db = unsafe { Database::create(tmpfile.path(), db_size).unwrap() };
-    let txn = db.begin_write().unwrap();
+    let mut txn = db.begin_write().unwrap();
+    txn.set_durability(Durability::None);
     let mut table = txn.open_table(SLICE_TABLE).unwrap();
 
     let pairs = gen_data(100, 16, 20);
@@ -60,7 +63,7 @@ fn non_durable_commit_persistence() {
             table.insert(key, value).unwrap();
         }
     }
-    txn.non_durable_commit().unwrap();
+    txn.commit().unwrap();
 
     // Check that cleanly closing the database persists the non-durable commit
     drop(db);
@@ -80,13 +83,13 @@ fn non_durable_commit_persistence() {
     }
 }
 
-#[test]
-fn persistence() {
+fn test_persistence(durability: Durability) {
     let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
 
     let db_size = 16 * 1024 * 1024;
     let db = unsafe { Database::create(tmpfile.path(), db_size).unwrap() };
-    let txn = db.begin_write().unwrap();
+    let mut txn = db.begin_write().unwrap();
+    txn.set_durability(durability);
     let mut table = txn.open_table(SLICE_TABLE).unwrap();
 
     let pairs = gen_data(100, 16, 20);
@@ -116,6 +119,16 @@ fn persistence() {
     }
 }
 
+#[test]
+fn eventual_persistence() {
+    test_persistence(Durability::Eventual);
+}
+
+#[test]
+fn immediate_persistence() {
+    test_persistence(Durability::Immediate);
+}
+
 #[test]
 fn change_db_size() {
     let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
@@ -203,7 +216,8 @@ fn regression4() {
     let db = unsafe { Database::create(tmpfile.path(), db_size).unwrap() };
     let mut i = 0u64;
     loop {
-        let txn = db.begin_write().unwrap();
+        let mut txn = db.begin_write().unwrap();
+        txn.set_durability(Durability::None);
         let mut table = txn.open_table(U64_TABLE).unwrap();
         // Fill the database
         match table.insert(&i, &i) {
@@ -216,7 +230,7 @@ fn regression4() {
                 _ => unreachable!(),
             },
         }
-        match txn.non_durable_commit() {
+        match txn.commit() {
             Ok(_) => {}
             Err(err) => match err {
                 Error::OutOfSpace => break,
@@ -476,22 +490,24 @@ fn regression3() {
 fn non_durable_read_isolation() {
     let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
     let db = unsafe { Database::create(tmpfile.path(), 1024 * 1024).unwrap() };
-    let write_txn = db.begin_write().unwrap();
+    let mut write_txn = db.begin_write().unwrap();
+    write_txn.set_durability(Durability::None);
     let mut table = write_txn.open_table(SLICE_TABLE).unwrap();
 
     table.insert(b"hello", b"world").unwrap();
-    write_txn.non_durable_commit().unwrap();
+    write_txn.commit().unwrap();
 
     let read_txn = db.begin_read().unwrap();
     let read_table = read_txn.open_table(SLICE_TABLE).unwrap();
     assert_eq!(b"world", read_table.get(b"hello").unwrap().unwrap());
 
-    let write_txn = db.begin_write().unwrap();
+    let mut write_txn = db.begin_write().unwrap();
+    write_txn.set_durability(Durability::None);
     let mut table = write_txn.open_table(SLICE_TABLE).unwrap();
     table.remove(b"hello").unwrap();
     table.insert(b"hello2", b"world2").unwrap();
     table.insert(b"hello3", b"world3").unwrap();
-    write_txn.non_durable_commit().unwrap();
+    write_txn.commit().unwrap();
 
     let read_txn2 = db.begin_read().unwrap();
     let read_table2 = read_txn2.open_table(SLICE_TABLE).unwrap();