Merge branch 'main' into rkuris/s3fifo

Author: rkuris

benchmark/README.md

@@ -169,7 +169,7 @@ If you're looking for detailed logging, there are some command line options to e
     cargo run --profile release --bin benchmark -- -l debug -n 10000 single
 ```
 
-# Using opentelemetry
+## Using opentelemetry
 
 To use the opentelemetry server and record timings, just run a docker image that collects the data using:
 
@@ -178,6 +178,3 @@ docker run   -p 127.0.0.1:4318:4318   -p 127.0.0.1:55679:55679   otel/openteleme
 ```
 
 Then, pass the `-e` option to the benchmark.
-```
-
-Then, pass the `-e` option to the benchmark.

firewood/Cargo.toml

@@ -18,7 +18,6 @@ readme = "../README.md"
 [dependencies]
 aquamarine = "0.6.0"
 async-trait = "0.1.77"
-storage = { version = "0.0.4", path = "../storage" }
 futures = "0.3.30"
 hex = "0.4.3"
 metrics = "0.24.0"
@@ -29,14 +28,13 @@ thiserror = "2.0.3"
 typed-builder = "0.20.0"
 bincode = "1.3.3"
 integer-encoding = "4.0.0"
-io-uring = {version = "0.7", optional = true }
 smallvec = "1.6.1"
 fastrace = { version = "0.7.4" }
 
 [features]
 default = []
 nightly = []
-iouring = ["io-uring"]
+io-uring = ["storage/io-uring"]
 logger = ["storage/logger"]
 branch_factor_256 = [ "storage/branch_factor_256" ]
 
@@ -60,3 +58,9 @@ unwrap_used = "warn"
 indexing_slicing = "warn"
 explicit_deref_methods = "warn"
 missing_const_for_fn = "warn"
+
+[target.'cfg(target_os = "linux")'.dependencies]
+storage = { version = "0.0.4", path = "../storage", features = ["io-uring"] }
+
+[target.'cfg(not(target_os = "linux"))'.dependencies]
+storage = { version = "0.0.4", path = "../storage" }

firewood/src/db.rs

@@ -425,7 +425,7 @@ impl<'a> api::Proposal for Proposal<'a> {
         match Arc::into_inner(self) {
             Some(proposal) => {
                 let mut manager = proposal.db.manager.write().expect("poisoned lock");
-                Ok(manager.commit(proposal.nodestore.clone())?)
+                Ok(manager.commit(proposal.nodestore)?)
             }
             None => Err(api::Error::CannotCommitClonedProposal),
         }
@@ -438,7 +438,7 @@ impl Proposal<'_> {
         match Arc::into_inner(self) {
             Some(proposal) => {
                 let mut manager = proposal.db.manager.write().expect("poisoned lock");
-                Ok(manager.commit(proposal.nodestore.clone())?)
+                Ok(manager.commit(proposal.nodestore)?)
             }
             None => Err(api::Error::CannotCommitClonedProposal),
         }

firewood/src/merkle.rs

@@ -16,8 +16,8 @@ use std::num::NonZeroUsize;
 use std::sync::Arc;
 use storage::{
     BranchNode, Child, Hashable, HashedNodeReader, ImmutableProposal, LeafNode, LinearAddress,
-    MutableProposal, NibblesIterator, Node, NodeStore, Path, ReadableStorage, TrieHash, TrieReader,
-    ValueDigest,
+    MutableProposal, NibblesIterator, Node, NodeStore, Path, ReadableStorage, SharedNode, TrieHash,
+    TrieReader, ValueDigest,
 };
 
 use thiserror::Error;
@@ -93,7 +93,7 @@ fn get_helper<T: TrieReader>(
     nodestore: &T,
     node: &Node,
     key: &[u8],
-) -> Result<Option<Arc<Node>>, MerkleError> {
+) -> Result<Option<SharedNode>, MerkleError> {
     // 4 possibilities for the position of the `key` relative to `node`:
     // 1. The node is at `key`
     // 2. The key is above the node (i.e. its ancestor)
@@ -111,7 +111,7 @@ fn get_helper<T: TrieReader>(
             // Case (2) or (4)
             Ok(None)
         }
-        (None, None) => Ok(Some(Arc::new(node.clone()))), // 1. The node is at `key`
+        (None, None) => Ok(Some(node.clone().into())), // 1. The node is at `key`
         (Some((child_index, remaining_key)), None) => {
             // 3. The key is below the node (i.e. its descendant)
             match node {
@@ -152,7 +152,7 @@ impl<T> From<T> for Merkle<T> {
 }
 
 impl<T: TrieReader> Merkle<T> {
-    pub(crate) fn root(&self) -> Option<Arc<Node>> {
+    pub(crate) fn root(&self) -> Option<SharedNode> {
         self.nodestore.root_node()
     }
 
@@ -161,7 +161,7 @@ impl<T: TrieReader> Merkle<T> {
         &self.nodestore
     }
 
-    fn read_node(&self, addr: LinearAddress) -> Result<Arc<Node>, MerkleError> {
+    fn read_node(&self, addr: LinearAddress) -> Result<SharedNode, MerkleError> {
         self.nodestore.read_node(addr).map_err(Into::into)
     }
 
@@ -335,7 +335,7 @@ impl<T: TrieReader> Merkle<T> {
         Ok(node.value().map(|v| v.to_vec().into_boxed_slice()))
     }
 
-    pub(crate) fn get_node(&self, key: &[u8]) -> Result<Option<Arc<Node>>, MerkleError> {
+    pub(crate) fn get_node(&self, key: &[u8]) -> Result<Option<SharedNode>, MerkleError> {
         let Some(root) = self.root() else {
             return Ok(None);
         };
@@ -591,16 +591,18 @@ impl<S: ReadableStorage> Merkle<NodeStore<MutableProposal, S>> {
         let root = self.nodestore.mut_root();
         let Some(root_node) = std::mem::take(root) else {
             // The trie is empty. There is nothing to remove.
-            counter!("firewood.remove", "result" => "nonexistent").increment(1);
+            counter!("firewood.remove", "prefix" => "false", "result" => "nonexistent")
+                .increment(1);
             return Ok(None);
         };
 
         let (root_node, removed_value) = self.remove_helper(root_node, &key)?;
         *self.nodestore.mut_root() = root_node;
         if removed_value.is_some() {
-            counter!("firewood.remove", "result" => "success").increment(1);
+            counter!("firewood.remove", "prefix" => "false", "result" => "success").increment(1);
         } else {
-            counter!("firewood.remove", "result" => "nonexistent").increment(1);
+            counter!("firewood.remove", "prefix" => "false", "result" => "nonexistent")
+                .increment(1);
         }
         Ok(removed_value)
     }
@@ -816,11 +818,14 @@ impl<S: ReadableStorage> Merkle<NodeStore<MutableProposal, S>> {
         let root = self.nodestore.mut_root();
         let Some(root_node) = std::mem::take(root) else {
             // The trie is empty. There is nothing to remove.
+            counter!("firewood.remove", "prefix" => "true", "result" => "nonexistent").increment(1);
             return Ok(0);
         };
 
         let mut deleted = 0;
         let root_node = self.remove_prefix_helper(root_node, &prefix, &mut deleted)?;
+        counter!("firewood.remove", "prefix" => "true", "result" => "success")
+            .increment(deleted as u64);
         *self.nodestore.mut_root() = root_node;
         Ok(deleted)
     }

firewood/src/stream.rs

@@ -8,17 +8,16 @@ use futures::stream::FusedStream;
 use futures::{Stream, StreamExt};
 use std::cmp::Ordering;
 use std::iter::once;
-use std::sync::Arc;
 use std::task::Poll;
-use storage::{BranchNode, Child, NibblesIterator, Node, PathIterItem, TrieReader};
+use storage::{BranchNode, Child, NibblesIterator, Node, PathIterItem, SharedNode, TrieReader};
 
 /// Represents an ongoing iteration over a node and its children.
 enum IterationNode {
     /// This node has not been returned yet.
     Unvisited {
         /// The key (as nibbles) of this node.
         key: Key,
-        node: Arc<Node>,
+        node: SharedNode,
     },
     /// This node has been returned. Track which child to visit next.
     Visited {
@@ -94,7 +93,7 @@ impl<'a, T: TrieReader> MerkleNodeStream<'a, T> {
 }
 
 impl<T: TrieReader> Stream for MerkleNodeStream<'_, T> {
-    type Item = Result<(Key, Arc<Node>), api::Error>;
+    type Item = Result<(Key, SharedNode), api::Error>;
 
     fn poll_next(
         mut self: std::pin::Pin<&mut Self>,
@@ -141,7 +140,7 @@ impl<T: TrieReader> Stream for MerkleNodeStream<'_, T> {
 
                             let child = match child {
                                 Child::AddressWithHash(addr, _) => merkle.read_node(addr)?,
-                                Child::Node(node) => Arc::new(node.clone()),
+                                Child::Node(node) => node.clone().into(),
                             };
 
                             let child_partial_path = child.partial_path().iter().copied();
@@ -251,7 +250,7 @@ fn get_iterator_intial_state<T: TrieReader>(
                     node = match child {
                         None => return Ok(NodeStreamState::Iterating { iter_stack }),
                         Some(Child::AddressWithHash(addr, _)) => merkle.read_node(*addr)?,
-                        Some(Child::Node(node)) => Arc::new((*node).clone()), // TODO can we avoid cloning this?
+                        Some(Child::Node(node)) => (*node).clone().into(), // TODO can we avoid cloning this?
                     };
 
                     matched_key_nibbles.push(next_unmatched_key_nibble);
@@ -374,7 +373,7 @@ enum PathIteratorState<'a> {
         /// prefix of the key we're traversing to.
         matched_key: Vec<u8>,
         unmatched_key: NibblesIterator<'a>,
-        node: Arc<Node>,
+        node: SharedNode,
     },
     Exhausted,
 }
@@ -504,7 +503,7 @@ impl<T: TrieReader> Iterator for PathIterator<'_, '_, T> {
                                         matched_key.push(next_unmatched_key_nibble);
 
                                         let ret = node.clone();
-                                        *node = Arc::new(child.clone());
+                                        *node = child.clone().into();
 
                                         Some(Ok(PathIterItem {
                                             key_nibbles: node_key,
@@ -584,6 +583,8 @@ fn key_from_nibble_iter<Iter: Iterator<Item = u8>>(mut nibbles: Iter) -> Key {
 #[cfg(test)]
 #[allow(clippy::indexing_slicing, clippy::unwrap_used)]
 mod tests {
+    use std::sync::Arc;
+
     use storage::{MemStore, MutableProposal, NodeStore};
 
     use crate::merkle::Merkle;

storage/Cargo.toml

@@ -22,8 +22,8 @@ bytemuck = "1.7.0"
 bytemuck_derive = "1.7.0"
 bitfield = "0.18.1"
 fastrace = { version = "0.7.4" }
-strum = "0.27.0"
-strum_macros = "0.27.0"
+io-uring = { version = "0.7.4", optional = true }
+triomphe = "0.1.14"
 s3-fifo = { git = "https://github.com/rkuris/s3-fifo", branch = "main" }
 
 [dev-dependencies]
@@ -36,6 +36,7 @@ tempfile = "3.12.0"
 [features]
 logger = ["log"]
 branch_factor_256 = []
+io-uring = ["dep:io-uring"]
 
 [[bench]]
 name = "serializer"

storage/src/lib.rs

@@ -1,7 +1,7 @@
 // Copyright (C) 2023, Ava Labs, Inc. All rights reserved.
 // See the file LICENSE.md for licensing terms.
 #![warn(missing_debug_implementations, rust_2018_idioms, missing_docs)]
-#![forbid(unsafe_code)]
+#![deny(unsafe_code)]
 
 //! # storage implements the storage of a [Node] on top of a LinearStore
 //!
@@ -33,14 +33,16 @@ pub use nodestore::{
 pub use linear::filebacked::FileBacked;
 pub use linear::memory::MemStore;
 
-use strum_macros::VariantArray;
 pub use trie_hash::TrieHash;
 
+/// A shared node, which is just a triophe Arc of a node
+pub type SharedNode = triomphe::Arc<Node>;
+
 /// The strategy for caching nodes that are read
 /// from the storage layer. Generally, we only want to
 /// cache write operations, but for some read-heavy workloads
 /// you can enable caching of branch reads or all reads.
-#[derive(Clone, Debug, VariantArray)]
+#[derive(Clone, Debug)]
 pub enum CacheReadStrategy {
     /// Only cache writes (no reads will be cached)
     WritesOnly,