feat(deferred-allocate): Part 3: Defer allocate

This implements deferred allocaion that delays allocating space for
nodes until commit time.

The nodestore refactor consisted of removing the 'new' hashmap, which
mapped newly allocated nodes to their contents in a proposal. Instead,
now we store those newly allocated nodes directly from the `Child`
struct.

The implementation constisted of:
 - A new type, NodeAllocator, to consolidate all the allocation code
 - Removing the 'new' hashmap, as freshly allocatred nodes are found
   by traversing the trie and inspecting their state
 - Use the most recently committed revision's freelist to allocate
   new nodes. This will change in a future revision when the freelist
   is tied to storage rather than the revision.
 - Rewrite flush_nodes to serialize then allocate. Previously, the node
   was serialized twice, once to determine the size during proposal
   creation (allocation time) and once to store it
 - Changes to the commit flow by separating the node persistence logic
   (see comments in commit.rs)

Some miscellaneous changes included in this diff (could be split out):
 - Branch children iterators can no longer return reference to hashes,
   since they are sometimes behind an ArcSwap
 - Propogate logger feature in fwdctl (helped with debugging)
 - Remove use of stored_len in tests
 - size_from_area_index was a method on nodestore, but didn't use
   nodestore
 - uncached_node_and_size was dead code
 - New clippy lint for hash_helper not needing &self in non-ethhash
   fixed

Task completion status:
  [x] Roots may not be persisted (Part 1: #1041)
  [x] BranchNode children can now be Node, LinearAddress or a
MaybePersistedNode (Part 2: #1045 and #1047)
  [x] When converting a `MutableProposal` to an `ImmutableProposal`,
don't actually allocate space, just create `SharedNode`s from the
`Node`s. (Part 3: #1055 and this PR)
  [ ] Remove `NodeStoreHeader` from `NodeStore`. The header should move
into the `RevisionManager`.
  [ ] Allocate new nodes from the recently-freed nodes from an expired
revision, then from the freelist. This can be done by maintaining a
local freelist from the nodes deleted in the expiring revision and
falling back to the actual freelist. Deleted nodes that are not reused
must still be pushed onto the freelist.

Author: rkuris

firewood/src/manager.rs

@@ -145,21 +145,10 @@ impl RevisionManager {
     ///    It only contains the address of the nodes that are deleted, which should be very small.
     /// 3. Revision reaping. If more than the maximum number of revisions are kept in memory, the
     ///    oldest revision is reaped.
-    /// 4. Set last committed revision.
+    /// 4. Persist to disk. This includes flushing everything to disk.
+    /// 5. Set last committed revision.
     ///    Set last committed revision in memory.
-    ///    Another commit can start after this but before the node flush is completed.
-    /// 5. Free list flush.
-    ///    Persist/write the free list header.
-    ///    The free list is flushed first to prevent future allocations from using the space allocated to this proposal.
-    ///    This should be done in a single write since the free list headers are small, and must be persisted to disk before starting the next step.
-    /// 6. Node flush.
-    ///    Persist/write all the nodes to disk.
-    ///    Note that since these are all freshly allocated nodes, they will never be referred to by any prior commit.
-    ///    After flushing all nodes, the file should be flushed to disk (fsync) before performing the next step.
-    /// 7. Root move.
-    ///    The root address on disk must be updated.
-    ///    This write can be delayed, but would mean that recovery will not roll forward to this revision.
-    /// 8. Proposal Cleanup.
+    /// 6. Proposal Cleanup.
     ///    Any other proposals that have this proposal as a parent should be reparented to the committed version.
     #[fastrace::trace(short_name = true)]
     #[crate::metrics("firewood.proposal.commit", "proposal commit to storage")]
@@ -170,7 +159,7 @@ impl RevisionManager {
             return Err(RevisionManagerError::NotLatest);
         }
 
-        let mut committed = proposal.as_committed();
+        let mut committed = proposal.as_committed(current_revision);
 
         // 2. Persist delete list for this committed revision to disk for recovery
 
@@ -212,7 +201,12 @@ impl RevisionManager {
             gauge!("firewood.max_revisions").set(self.max_revisions as f64);
         }
 
-        // 4. Set last committed revision
+        // 4. Persist to disk.
+        // TODO: We can probably do this in another thread, but it requires that
+        // we move the header out of NodeStore, which is in a future PR.
+        committed.persist()?;
+
+        // 5. Set last committed revision
         let committed: CommittedRevision = committed.into();
         self.historical
             .write()
@@ -224,18 +218,8 @@ impl RevisionManager {
                 .expect("poisoned lock")
                 .insert(hash, committed.clone());
         }
-        // TODO: We could allow other commits to start here using the pending list
-
-        // 5. Free list flush, which will prevent allocating on top of the nodes we are about to write
-        proposal.flush_freelist()?;
-
-        // 6. Node flush
-        proposal.flush_nodes()?;
-
-        // 7. Root move
-        proposal.flush_header()?;
 
-        // 8. Proposal Cleanup
+        // 6. Proposal Cleanup
         // Free proposal that is being committed as well as any proposals no longer
         // referenced by anyone else.
         self.proposals

firewood/src/proof.rs

@@ -106,11 +106,11 @@ impl Hashable for ProofNode {
         })
     }
 
-    fn children(&self) -> impl Iterator<Item = (usize, &HashType)> + Clone {
+    fn children(&self) -> impl Iterator<Item = (usize, HashType)> + Clone {
         self.child_hashes
             .iter()
             .enumerate()
-            .filter_map(|(i, hash)| hash.as_ref().map(|h| (i, h)))
+            .filter_map(|(i, hash)| hash.as_ref().map(|h| (i, h.clone())))
     }
 }
 

fwdctl/Cargo.toml

@@ -38,6 +38,7 @@ nonzero_ext = "0.3.0"
 
 [features]
 ethhash = ["firewood/ethhash"]
+logger = ["firewood/logger"]
 
 [dev-dependencies]
 # Workspace dependencies

storage/src/checker/mod.rs

@@ -5,10 +5,10 @@ mod range_set;
 use range_set::LinearAddressRangeSet;
 
 use crate::logger::warn;
-use crate::nodestore::alloc::{AREA_SIZES, AreaIndex, FreeAreaWithMetadata};
+use crate::nodestore::alloc::{AREA_SIZES, AreaIndex, FreeAreaWithMetadata, size_from_area_index};
 use crate::{
     CheckerError, Committed, HashType, HashedNodeReader, LinearAddress, Node, NodeReader,
-    NodeStore, Path, StoredAreaParent, TrieNodeParent, WritableStorage, hash_node,
+    NodeStore, Path, RootReader, StoredAreaParent, TrieNodeParent, WritableStorage, hash_node,
 };
 
 use std::cmp::Ordering;
@@ -42,19 +42,35 @@ impl<S: WritableStorage> NodeStore<Committed, S> {
         let mut visited = LinearAddressRangeSet::new(db_size)?;
 
         // 2. traverse the trie and check the nodes
-        if let (Some(root_address), Some(root_hash)) = (self.root_address(), self.root_hash()) {
+        if let Some(root) = self.root_as_maybe_persisted_node() {
             // the database is not empty, traverse the trie
-            self.check_area_aligned(
-                root_address,
-                StoredAreaParent::TrieNode(TrieNodeParent::Root),
-            )?;
-            self.visit_trie(
-                root_address,
-                HashType::from(root_hash),
-                Path::new(),
-                &mut visited,
-                opt.hash_check,
-            )?;
+            if let Some(root_address) = root.as_linear_address() {
+                self.check_area_aligned(
+                    root_address,
+                    StoredAreaParent::TrieNode(TrieNodeParent::Root),
+                )?;
+                // For hash checking, we need the root hash if it exists
+                #[cfg_attr(not(feature = "ethhash"), expect(clippy::useless_conversion))]
+                let root_hash = self.root_hash().map(HashType::from);
+                if let Some(root_hash) = root_hash {
+                    self.visit_trie(
+                        root_address,
+                        root_hash,
+                        Path::new(),
+                        &mut visited,
+                        opt.hash_check,
+                    )?;
+                } else {
+                    // No root hash available, visit without hash checking
+                    self.visit_trie(
+                        root_address,
+                        HashType::default(), // dummy hash
+                        Path::new(),
+                        &mut visited,
+                        false, // can't check hashes without root hash
+                    )?;
+                }
+            }
         }
 
         // 3. check the free list - this can happen in parallel with the trie traversal
@@ -71,7 +87,7 @@ impl<S: WritableStorage> NodeStore<Committed, S> {
         Ok(())
     }
 
-    /// Recursively traverse the trie from the given root address.
+    /// Recursively traverse the trie from the given root node.
     fn visit_trie(
         &self,
         subtree_root_address: LinearAddress,
@@ -134,7 +150,7 @@ impl<S: WritableStorage> NodeStore<Committed, S> {
                 parent,
             } = free_area?;
             self.check_area_aligned(addr, StoredAreaParent::FreeList(parent))?;
-            let area_size = Self::size_from_area_index(area_index);
+            let area_size = size_from_area_index(area_index);
             if free_list_id != area_index {
                 return Err(CheckerError::FreelistAreaSizeMismatch {
                     address: addr,

storage/src/hashednode.rs

@@ -124,7 +124,7 @@ pub trait Hashable {
     fn value_digest(&self) -> Option<ValueDigest<&[u8]>>;
     /// Each element is a child's index and hash.
     /// Yields 0 elements if the node is a leaf.
-    fn children(&self) -> impl Iterator<Item = (usize, &HashType)> + Clone;
+    fn children(&self) -> impl Iterator<Item = (usize, HashType)> + Clone;
 }
 
 /// A preimage of a hash.
@@ -138,7 +138,7 @@ pub trait Preimage {
 trait HashableNode {
     fn partial_path(&self) -> impl Iterator<Item = u8> + Clone;
     fn value(&self) -> Option<&[u8]>;
-    fn children_iter(&self) -> impl Iterator<Item = (usize, &HashType)> + Clone;
+    fn children_iter(&self) -> impl Iterator<Item = (usize, HashType)> + Clone;
 }
 
 impl HashableNode for BranchNode {
@@ -150,7 +150,7 @@ impl HashableNode for BranchNode {
         self.value.as_deref()
     }
 
-    fn children_iter(&self) -> impl Iterator<Item = (usize, &HashType)> + Clone {
+    fn children_iter(&self) -> impl Iterator<Item = (usize, HashType)> + Clone {
         self.children_hashes()
     }
 }
@@ -164,7 +164,7 @@ impl HashableNode for LeafNode {
         Some(&self.value)
     }
 
-    fn children_iter(&self) -> impl Iterator<Item = (usize, &HashType)> + Clone {
+    fn children_iter(&self) -> impl Iterator<Item = (usize, HashType)> + Clone {
         iter::empty()
     }
 }
@@ -198,7 +198,7 @@ impl<'a, N: HashableNode> Hashable for NodeAndPrefix<'a, N> {
         self.node.value().map(ValueDigest::Value)
     }
 
-    fn children(&self) -> impl Iterator<Item = (usize, &HashType)> + Clone {
+    fn children(&self) -> impl Iterator<Item = (usize, HashType)> + Clone {
         self.node.children_iter()
     }
 }

storage/src/hashers/ethhash.rs

@@ -173,7 +173,7 @@ impl<T: Hashable> Preimage for T {
             let mut rlp = RlpStream::new_list(17);
             let mut child_iter = self.children().peekable();
             for index in 0..=15 {
-                if let Some(&(child_index, digest)) = child_iter.peek() {
+                if let Some(&(child_index, ref digest)) = child_iter.peek() {
                     if child_index == index {
                         match digest {
                             HashType::Hash(hash) => rlp.append(&hash.as_slice()),
@@ -225,7 +225,7 @@ impl<T: Hashable> Preimage for T {
                                     self.partial_path().collect::<Box<_>>(),
                                     true,
                                 ));
-                                rlp.append_raw(rlp_bytes, 1);
+                                rlp.append_raw(&rlp_bytes, 1);
                                 let bytes = rlp.out();
                                 TrieHash::from(Keccak256::digest(bytes))
                             }

storage/src/lib.rs

@@ -47,7 +47,7 @@ pub use node::{
 };
 pub use nodestore::{
     Committed, HashedNodeReader, ImmutableProposal, LinearAddress, MutableProposal, NodeReader,
-    NodeStore, Parentable, ReadInMemoryNode, RootReader, TrieReader,
+    NodeStore, Parentable, RootReader, TrieReader,
 };
 
 pub use linear::filebacked::FileBacked;

storage/src/linear/filebacked.rs

@@ -25,7 +25,7 @@
 
 use std::fs::{File, OpenOptions};
 use std::io::Read;
-use std::num::NonZero;
+use std::num::{NonZero, NonZeroU64};
 #[cfg(unix)]
 use std::os::unix::fs::FileExt;
 #[cfg(windows)]
@@ -212,13 +212,13 @@ impl WritableStorage for FileBacked {
         }
     }
 
-    fn write_cached_nodes<'a>(
+    fn write_cached_nodes(
         &self,
-        nodes: impl Iterator<Item = (LinearAddress, &'a SharedNode)>,
+        nodes: impl Iterator<Item = (NonZeroU64, SharedNode)>,
     ) -> Result<(), FileIoError> {
         let mut guard = self.cache.lock().expect("poisoned lock");
         for (addr, node) in nodes {
-            guard.put(addr, node.clone());
+            guard.put(addr.into(), node);
         }
         Ok(())
     }

storage/src/linear/mod.rs

@@ -19,7 +19,7 @@
     reason = "Found 4 occurrences after enabling the lint."
 )]
 
-use std::fmt::Debug;
+use std::{fmt::Debug, num::NonZero};
 use std::io::Read;
 use std::ops::Deref;
 use std::path::PathBuf;
@@ -181,9 +181,9 @@ pub trait WritableStorage: ReadableStorage {
     fn write(&self, offset: u64, object: &[u8]) -> Result<usize, FileIoError>;
 
     /// Write all nodes to the cache (if any)
-    fn write_cached_nodes<'a>(
+    fn write_cached_nodes(
         &self,
-        _nodes: impl Iterator<Item = (LinearAddress, &'a SharedNode)>,
+        _nodes: impl Iterator<Item = (NonZero<u64>, SharedNode)>,
     ) -> Result<(), FileIoError> {
         Ok(())
     }

storage/src/node/branch.rs

@@ -412,8 +412,13 @@ impl BranchNode {
     }
 
     /// Returns (index, hash) for each child that has a hash set.
-    pub fn children_hashes(&self) -> impl Iterator<Item = (usize, &HashType)> + Clone {
-        self.children_iter().map(|(idx, (_, hash))| (idx, hash))
+    pub fn children_hashes(&self) -> impl Iterator<Item = (usize, HashType)> + Clone {
+        self.children.iter().enumerate().filter_map(|(i, child)| {
+            child
+                .as_ref()
+                .and_then(|child| child.hash().cloned())
+                .map(|hash| (i, hash))
+        })
     }
 
     /// Returns (index, address) for each child that has a hash set.