Add Dyn object-safe storage engine, and use for Raft log

Author: erikgrinaker

Diff for: src/bin/toydb.rs

@@ -37,13 +37,10 @@ async fn main() -> Result<()> {
     let path = std::path::Path::new(&cfg.data_dir);
     let raft_log = match cfg.storage_raft.as_str() {
         "bitcask" | "" => raft::Log::new(
-            Box::new(storage::engine::BitCask::new_compact(
-                path.join("log"),
-                cfg.compact_threshold,
-            )?),
+            storage::engine::BitCask::new_compact(path.join("log"), cfg.compact_threshold)?,
             cfg.sync,
         )?,
-        "memory" => raft::Log::new(Box::new(storage::engine::Memory::new()), false)?,
+        "memory" => raft::Log::new(storage::engine::Memory::new(), false)?,
         name => return Err(Error::Config(format!("Unknown Raft storage engine {}", name))),
     };
     let raft_state: Box<dyn raft::State> = match cfg.storage_sql.as_str() {

Diff for: src/raft/log.rs

@@ -56,8 +56,9 @@ impl KeyPrefix {
 }
 /// A Raft log.
 pub struct Log {
-    /// The underlying storage engine.
-    engine: Box<dyn Engine>,
+    /// The underlying storage engine. Uses a trait object to avoid leaking
+    /// generic type parameters throughout the Raft module.
+    engine: Box<dyn storage::engine::Dyn>,
     /// The index of the last stored entry.
     last_index: Index,
     /// The term of the last stored entry.
@@ -72,7 +73,7 @@ pub struct Log {
 
 impl Log {
     /// Creates a new log, using the given storage engine.
-    pub fn new(mut engine: Box<dyn Engine>, sync: bool) -> Result<Self> {
+    pub fn new(mut engine: impl storage::engine::Engine + 'static, sync: bool) -> Result<Self> {
         let (last_index, last_term) = engine
             .scan_prefix(&KeyPrefix::Entry.encode()?)
             .last()
@@ -86,7 +87,14 @@ impl Log {
             .map(|v| bincode::deserialize(&v))
             .transpose()?
             .unwrap_or((0, 0));
-        Ok(Self { engine, last_index, last_term, commit_index, commit_term, sync })
+        Ok(Self {
+            engine: Box::new(engine),
+            last_index,
+            last_term,
+            commit_index,
+            commit_term,
+            sync,
+        })
     }
 
     /// Decodes an entry from a log key/value pair.
@@ -210,7 +218,7 @@ impl Log {
                 std::ops::Bound::Included(Key::Entry(Index::MAX).encode()?)
             }
         };
-        Ok(self.engine.scan(from, to).map(|r| r.and_then(|(k, v)| Self::decode_entry(&k, &v))))
+        Ok(self.engine.scan((from, to)).map(|r| r.and_then(|(k, v)| Self::decode_entry(&k, &v))))
     }
 
     /// Splices a set of entries into the log. The entries must be contiguous,
@@ -260,83 +268,14 @@ impl Log {
     }
 }
 
-/// A Raft log storage engine. This is a wrapper trait around storage::Engine
-/// with a blanket implementation, to make it object-safe (this is otherwise
-/// prevented by the generic scan() method). This wrapper allows Log to use a
-/// trait object for the engine, which prevents leaking generics throughout the
-/// Raft implementation.
-///
-/// TODO: Consider getting rid of this and using generics throughout.
-pub trait Engine: std::fmt::Display + Send + Sync {
-    fn delete(&mut self, key: &[u8]) -> Result<()>;
-
-    fn flush(&mut self) -> Result<()>;
-
-    fn get(&mut self, key: &[u8]) -> Result<Option<Vec<u8>>>;
-
-    #[allow(clippy::type_complexity)]
-    fn scan(
-        &mut self,
-        from: std::ops::Bound<Vec<u8>>,
-        to: std::ops::Bound<Vec<u8>>,
-    ) -> Box<dyn DoubleEndedIterator<Item = Result<(Vec<u8>, Vec<u8>)>> + '_>;
-
-    #[allow(clippy::type_complexity)]
-    fn scan_prefix(
-        &mut self,
-        prefix: &[u8],
-    ) -> Box<dyn DoubleEndedIterator<Item = Result<(Vec<u8>, Vec<u8>)>> + '_>;
-
-    fn set(&mut self, key: &[u8], value: Vec<u8>) -> Result<()>;
-
-    fn status(&mut self) -> Result<storage::engine::Status>;
-}
-
-impl<E: storage::engine::Engine> Engine for E {
-    fn delete(&mut self, key: &[u8]) -> Result<()> {
-        self.delete(key)
-    }
-
-    fn flush(&mut self) -> Result<()> {
-        self.flush()
-    }
-
-    fn get(&mut self, key: &[u8]) -> Result<Option<Vec<u8>>> {
-        self.get(key)
-    }
-
-    fn scan(
-        &mut self,
-        from: std::ops::Bound<Vec<u8>>,
-        to: std::ops::Bound<Vec<u8>>,
-    ) -> Box<dyn DoubleEndedIterator<Item = Result<(Vec<u8>, Vec<u8>)>> + '_> {
-        Box::new(self.scan((from, to)))
-    }
-
-    fn scan_prefix(
-        &mut self,
-        prefix: &[u8],
-    ) -> Box<dyn DoubleEndedIterator<Item = Result<(Vec<u8>, Vec<u8>)>> + '_> {
-        Box::new(self.scan_prefix(prefix))
-    }
-
-    fn set(&mut self, key: &[u8], value: Vec<u8>) -> Result<()> {
-        self.set(key, value)
-    }
-
-    fn status(&mut self) -> Result<storage::engine::Status> {
-        self.status()
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::storage::engine::Memory;
     use pretty_assertions::assert_eq;
 
     fn setup() -> Log {
-        Log::new(Box::new(Memory::new()), false).expect("empty engine should never fail to open")
+        Log::new(Memory::new(), false).expect("empty engine should never fail to open")
     }
 
     #[test]

Diff for: src/raft/mod.rs

@@ -4,7 +4,7 @@ mod node;
 mod server;
 mod state;
 
-pub use self::log::{Engine, Entry, Index, Log};
+pub use log::{Entry, Index, Log};
 pub use message::{Address, Event, Message, Request, RequestID, Response};
 pub use node::{Node, NodeID, Status, Term};
 pub use server::Server;

Diff for: src/raft/node/candidate.rs

@@ -176,7 +176,7 @@ mod tests {
     )> {
         let (node_tx, node_rx) = mpsc::unbounded_channel();
         let (state_tx, state_rx) = mpsc::unbounded_channel();
-        let mut log = Log::new(Box::new(storage::engine::Memory::new()), false)?;
+        let mut log = Log::new(storage::engine::Memory::new(), false)?;
         log.append(1, Some(vec![0x01]))?;
         log.append(1, Some(vec![0x02]))?;
         log.append(2, Some(vec![0x03]))?;

Diff for: src/raft/node/follower.rs

@@ -284,7 +284,7 @@ pub mod tests {
     )> {
         let (node_tx, node_rx) = mpsc::unbounded_channel();
         let (state_tx, state_rx) = mpsc::unbounded_channel();
-        let mut log = Log::new(Box::new(storage::engine::Memory::new()), false)?;
+        let mut log = Log::new(storage::engine::Memory::new(), false)?;
         log.append(1, Some(vec![0x01]))?;
         log.append(1, Some(vec![0x02]))?;
         log.append(2, Some(vec![0x03]))?;
@@ -604,7 +604,7 @@ pub mod tests {
         // TODO: Move this into a setup function.
         let (node_tx, mut node_rx) = mpsc::unbounded_channel();
         let (state_tx, mut state_rx) = mpsc::unbounded_channel();
-        let mut log = Log::new(Box::new(storage::engine::Memory::new()), false)?;
+        let mut log = Log::new(storage::engine::Memory::new(), false)?;
         log.append(1, Some(vec![0x01]))?;
         log.append(1, Some(vec![0x02]))?;
         log.append(2, Some(vec![0x03]))?;

Diff for: src/raft/node/leader.rs

@@ -306,7 +306,7 @@ mod tests {
         let (node_tx, node_rx) = mpsc::unbounded_channel();
         let (state_tx, state_rx) = mpsc::unbounded_channel();
         let peers = HashSet::from([2, 3, 4, 5]);
-        let mut log = Log::new(Box::new(storage::engine::Memory::new()), false)?;
+        let mut log = Log::new(storage::engine::Memory::new(), false)?;
         log.append(1, Some(vec![0x01]))?;
         log.append(1, Some(vec![0x02]))?;
         log.append(2, Some(vec![0x03]))?;

Diff for: src/raft/node/mod.rs

@@ -383,7 +383,7 @@ mod tests {
             id: 1,
             peers: HashSet::from_iter(peers),
             term: 1,
-            log: Log::new(Box::new(storage::engine::Memory::new()), false)?,
+            log: Log::new(storage::engine::Memory::new(), false)?,
             node_tx,
             state_tx,
         };
@@ -396,7 +396,7 @@ mod tests {
         let node = Node::new(
             1,
             HashSet::from([2, 3]),
-            Log::new(Box::new(storage::engine::Memory::new()), false)?,
+            Log::new(storage::engine::Memory::new(), false)?,
             Box::new(TestState::new(0)),
             node_tx,
         )
@@ -415,7 +415,7 @@ mod tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn new_state_apply_all() -> Result<()> {
         let (node_tx, _) = mpsc::unbounded_channel();
-        let mut log = Log::new(Box::new(storage::engine::Memory::new()), false)?;
+        let mut log = Log::new(storage::engine::Memory::new(), false)?;
         log.append(1, Some(vec![0x01]))?;
         log.append(2, None)?;
         log.append(2, Some(vec![0x02]))?;
@@ -433,7 +433,7 @@ mod tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn new_state_apply_partial() -> Result<()> {
         let (node_tx, _) = mpsc::unbounded_channel();
-        let mut log = Log::new(Box::new(storage::engine::Memory::new()), false)?;
+        let mut log = Log::new(storage::engine::Memory::new(), false)?;
         log.append(1, Some(vec![0x01]))?;
         log.append(2, None)?;
         log.append(2, Some(vec![0x02]))?;
@@ -452,7 +452,7 @@ mod tests {
     #[should_panic(expected = "applied index above commit index")]
     async fn new_state_apply_missing() {
         let (node_tx, _) = mpsc::unbounded_channel();
-        let mut log = Log::new(Box::new(storage::engine::Memory::new()), false).unwrap();
+        let mut log = Log::new(storage::engine::Memory::new(), false).unwrap();
         log.append(1, Some(vec![0x01])).unwrap();
         log.append(2, None).unwrap();
         log.append(2, Some(vec![0x02])).unwrap();
@@ -469,7 +469,7 @@ mod tests {
         let node = Node::new(
             1,
             HashSet::new(),
-            Log::new(Box::new(storage::engine::Memory::new()), false)?,
+            Log::new(storage::engine::Memory::new(), false)?,
             Box::new(TestState::new(0)),
             node_tx,
         )

Diff for: src/storage/engine/dyn.rs

@@ -0,0 +1,61 @@
+use crate::error::Result;
+
+use super::{Engine, Status};
+
+/// A variant of Engine that's object-safe. This is otherwise prevented by the
+/// generic Engine::scan() method. This uses dynamic dispatch, including for the
+/// scan iterator, which comes with a minor performance penalty.
+pub trait Dyn: Send + Sync {
+    fn delete(&mut self, key: &[u8]) -> Result<()>;
+
+    fn flush(&mut self) -> Result<()>;
+
+    fn get(&mut self, key: &[u8]) -> Result<Option<Vec<u8>>>;
+
+    fn scan(
+        &mut self,
+        range: (std::ops::Bound<Vec<u8>>, std::ops::Bound<Vec<u8>>),
+    ) -> ScanIterator<'_>;
+
+    fn scan_prefix(&mut self, prefix: &[u8]) -> ScanIterator<'_>;
+
+    fn set(&mut self, key: &[u8], value: Vec<u8>) -> Result<()>;
+
+    fn status(&mut self) -> Result<Status>;
+}
+
+type ScanIterator<'a> = Box<dyn DoubleEndedIterator<Item = Result<(Vec<u8>, Vec<u8>)>> + 'a>;
+
+/// Blanket implementation of Dyn for the Engine trait.
+impl<E: Engine> Dyn for E {
+    fn delete(&mut self, key: &[u8]) -> Result<()> {
+        self.delete(key)
+    }
+
+    fn flush(&mut self) -> Result<()> {
+        self.flush()
+    }
+
+    fn get(&mut self, key: &[u8]) -> Result<Option<Vec<u8>>> {
+        self.get(key)
+    }
+
+    fn scan(
+        &mut self,
+        range: (std::ops::Bound<Vec<u8>>, std::ops::Bound<Vec<u8>>),
+    ) -> ScanIterator<'_> {
+        Box::new(self.scan(range))
+    }
+
+    fn scan_prefix(&mut self, prefix: &[u8]) -> ScanIterator<'_> {
+        Box::new(self.scan_prefix(prefix))
+    }
+
+    fn set(&mut self, key: &[u8], value: Vec<u8>) -> Result<()> {
+        self.set(key, value)
+    }
+
+    fn status(&mut self) -> Result<Status> {
+        self.status()
+    }
+}

Diff for: src/storage/engine/mod.rs

@@ -1,10 +1,12 @@
 mod bitcask;
+mod r#dyn;
 mod memory;
 
 #[cfg(test)]
 pub use super::debug::Engine as Debug;
 pub use bitcask::BitCask;
 pub use memory::Memory;
+pub use r#dyn::Dyn;
 
 use crate::error::Result;
 

Diff for: tests/setup.rs

@@ -79,7 +79,7 @@ pub async fn server(
     let mut srv = Server::new(
         id,
         peers,
-        raft::Log::new(Box::new(storage::engine::BitCask::new(dir.path().join("log"))?), false)?,
+        raft::Log::new(storage::engine::BitCask::new(dir.path().join("log"))?, false)?,
         Box::new(sql::engine::Raft::new_state(storage::engine::Memory::new())?),
     )
     .await?;