diff --git a/core/src/update.rs b/core/src/update.rs index 5aa857a38b..f86f3a3ae6 100644 --- a/core/src/update.rs +++ b/core/src/update.rs @@ -160,9 +160,16 @@ pub fn build_trie( } let common_after_prefix = |k1: &KeyPath, k2: &KeyPath| { - let x = &k1.view_bits::()[skip..]; - let y = &k2.view_bits::()[skip..]; - shared_bits(x, y) + let first_byte = skip / 8; + let shared = k1[first_byte..] + .iter() + .zip(&k2[first_byte..]) + .position(|(a, b)| a != b) + .map_or(256, |i| { + (first_byte + i) * 8 + + (k1[first_byte + i] ^ k2[first_byte + i]).leading_zeros() as usize + }); + shared - skip }; while let Some((this_key, this_val)) = b { diff --git a/nomt/src/beatree/index.rs b/nomt/src/beatree/index.rs index 6537c77541..6cf7362215 100644 --- a/nomt/src/beatree/index.rs +++ b/nomt/src/beatree/index.rs @@ -43,6 +43,11 @@ impl Index { self.first_key_map.insert(separator, branch) } + /// Iterate all bottom-level branch nodes. + pub fn branches(&self) -> impl Iterator> { + self.first_key_map.values() + } + #[cfg(test)] pub fn into_iter(self) -> impl Iterator)> { self.first_key_map.into_iter() diff --git a/nomt/src/beatree/iterator.rs b/nomt/src/beatree/iterator.rs index 6f23963edd..b5a74fd979 100644 --- a/nomt/src/beatree/iterator.rs +++ b/nomt/src/beatree/iterator.rs @@ -235,6 +235,27 @@ impl LeafIterator { } fn next<'a>(&'a mut self) -> Option> { + // Most steps remain within the same leaf. Advance its cursor in place instead + // of moving the entire state, including its branch and leaf handles, twice. + let within_leaf = match &self.state { + LeafIteratorState::Proceeding { current, .. } => { + let next = current.index + 1; + next < current.leaf.n() + && self + .end + .as_ref() + .map_or(true, |end| current.leaf.key(next) < *end) + } + _ => false, + }; + if within_leaf { + let LeafIteratorState::Proceeding { current, .. } = &mut self.state else { + unreachable!(); + }; + current.index += 1; + return Some(current.last_output()); + } + // This is a so-called "Streaming Iterator", where the items borrow the lifetime // of the callsite rather than consuming items from the object - these are somewhat awkward // because we first have to advance the state of the iterator and then return a borrow diff --git a/nomt/src/beatree/leaf_cache.rs b/nomt/src/beatree/leaf_cache.rs index da18639d86..67aa48e4ed 100644 --- a/nomt/src/beatree/leaf_cache.rs +++ b/nomt/src/beatree/leaf_cache.rs @@ -43,6 +43,15 @@ impl LeafCache { shard.cache.get(&page_number).map(|x| x.clone()) } + /// The maximum number of leaves the cache holds. + pub fn capacity(&self) -> usize { + self.inner + .shards + .iter() + .map(|shard| shard.lock().max_items) + .sum() + } + /// Insert a cache entry. This does not evict anything. pub fn insert(&self, page_number: PageNumber, node: Arc) { let mut shard = self.inner.shard_for(page_number); @@ -50,6 +59,11 @@ impl LeafCache { shard.cache.put(page_number, node); } + /// Remove a superseded page. Existing readers can still reload it from disk. + pub(super) fn remove(&self, page_number: PageNumber) { + self.inner.shard_for(page_number).cache.pop(&page_number); + } + /// Evict all excess items from the cache. pub fn evict(&self) { for shard in &self.inner.shards { diff --git a/nomt/src/beatree/mod.rs b/nomt/src/beatree/mod.rs index 547a763e34..220a909e3e 100644 --- a/nomt/src/beatree/mod.rs +++ b/nomt/src/beatree/mod.rs @@ -141,6 +141,36 @@ impl Tree { }) } + /// Fill the leaf cache with leaves, up to its capacity, reading them on `threads` threads. + pub fn prepopulate_leaf_cache(&self, threads: usize) { + let shared = self.shared.read(); + let leaf_cache = shared.leaf_cache.clone(); + let leaf_store = shared.leaf_store_rd.clone(); + let leaves: Vec = shared + .bbn_index + .branches() + .flat_map(|branch| (0..branch.n() as usize).map(|i| PageNumber(branch.node_pointer(i)))) + .take(leaf_cache.capacity()) + .collect(); + drop(shared); + + // `chunks` panics on 0, so keep the size at least 1 even if there are no leaves. + let chunk_size = leaves.len().div_ceil(threads.max(1)).max(1); + let (leaf_cache, leaf_store) = (&leaf_cache, &leaf_store); + std::thread::scope(|scope| { + for chunk in leaves.chunks(chunk_size) { + scope.spawn(move || { + for &pn in chunk { + let leaf = Arc::new(leaf::node::LeafNode { + inner: leaf_store.query(pn), + }); + leaf_cache.insert(pn, leaf); + } + }); + } + }); + } + /// Lookup a key in the btree. This blocks the current thread. pub fn lookup(&self, key: Key) -> Option> { let shared = self.shared.read(); @@ -483,13 +513,22 @@ impl SyncController { /// Finishes sync. /// - /// Has to be called after the manifest is updated. Must be invoked by the sync - /// thread. Blocking. - pub fn post_meta(&mut self) { - let pre_swap_rx = self.inner.pre_swap_rx.lock().take().unwrap(); - join_task(&pre_swap_rx); - let bbn_index = self.inner.bbn_index.lock().take().unwrap(); - Tree::finish_sync(&self.inner.shared, bbn_index); + /// Has to be called after the manifest is updated. Non-blocking: the returned task must be + /// joined before the commit completes. + pub fn post_meta(&mut self) -> Receiver> { + let inner = self.inner.clone(); + let (tx, rx) = crossbeam_channel::bounded(1); + spawn_task( + &self.inner.sync.tp, + move || { + let pre_swap_rx = inner.pre_swap_rx.lock().take().unwrap(); + join_task(&pre_swap_rx); + let bbn_index = inner.bbn_index.lock().take().unwrap(); + Tree::finish_sync(&inner.shared, bbn_index); + }, + tx, + ); + rx } } diff --git a/nomt/src/beatree/ops/update/leaf_stage.rs b/nomt/src/beatree/ops/update/leaf_stage.rs index 8ddb37fbab..8eed7fe8ad 100644 --- a/nomt/src/beatree/ops/update/leaf_stage.rs +++ b/nomt/src/beatree/ops/update/leaf_stage.rs @@ -57,6 +57,9 @@ impl PostIoWork { pub(super) fn run(self, leaf_cache: &LeafCache) { for worker_output in self.worker_changes.into_iter() { for (_, leaf_entry) in worker_output.leaves_tracker.inner { + if let Some(pn) = leaf_entry.deleted { + leaf_cache.remove(pn); + } if let Some((leaf, pn)) = leaf_entry.inserted { leaf_cache.insert(pn, leaf); } diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs index bab4adad00..6a81f379ff 100644 --- a/nomt/src/lib.rs +++ b/nomt/src/lib.rs @@ -124,14 +124,17 @@ impl KeyReadWrite { } } + /// Hashes the values. The read value hash lets the merkle update recognize the key's own leaf + /// without fetching it. fn to_compact(&self) -> crate::merkle::KeyReadWrite { let hash = |v: &Value| T::hash_value(v); match self { - KeyReadWrite::Read(_) => crate::merkle::KeyReadWrite::Read, + KeyReadWrite::Read(read) => crate::merkle::KeyReadWrite::Read(read.as_ref().map(hash)), KeyReadWrite::Write(val) => crate::merkle::KeyReadWrite::Write(val.as_ref().map(hash)), - KeyReadWrite::ReadThenWrite(_, val) => { - crate::merkle::KeyReadWrite::ReadThenWrite(val.as_ref().map(hash)) - } + KeyReadWrite::ReadThenWrite(read, val) => crate::merkle::KeyReadWrite::ReadThenWrite( + read.as_ref().map(hash), + val.as_ref().map(hash), + ), } } } @@ -197,6 +200,7 @@ impl From<[u8; 32]> for Root { /// An instance of the Nearly-Optimal Merkle Trie Database. pub struct Nomt { merkle_update_pool: UpdatePool, + actuals_concurrency: usize, /// The handle to the page cache. page_cache: PageCache, page_pool: PagePool, @@ -234,9 +238,13 @@ impl Nomt { let io_handle = store.io_pool().make_handle(); merkle::prepopulate_cache(io_handle, &page_cache, &store, o.page_cache_upper_levels)?; } + if o.prepopulate_leaf_cache { + store.prepopulate_leaf_cache(o.commit_concurrency); + } Ok(Self { merkle_update_pool: UpdatePool::new(o.commit_concurrency, o.warm_up), + actuals_concurrency: o.actuals_concurrency, page_cache, page_pool, store, @@ -328,6 +336,7 @@ impl Nomt { Session { store, + actuals_concurrency: self.actuals_concurrency, merkle_updater: self.merkle_update_pool.begin::( self.page_cache.clone(), self.page_pool.clone(), @@ -477,6 +486,7 @@ impl SessionParams { /// correctness of replaying the same operations. pub struct Session { store: Store, + actuals_concurrency: usize, merkle_updater: Updater, metrics: Metrics, rollback_delta: Option, @@ -580,20 +590,18 @@ impl Session { ); } } + let merkle_update_handle = self.merkle_updater.update_and_prove::( + compact_actuals::(&actuals, self.actuals_concurrency), + self.witness_mode.0, + )?; + + // Built while the merkle update runs: both only read the store, which is not modified + // until the finished session is committed. let rollback_delta = self .rollback_delta .take() .map(|delta_builder| delta_builder.finalize(&actuals)); - let mut compact_actuals = Vec::with_capacity(actuals.len()); - for (path, read_write) in &actuals { - compact_actuals.push((path.clone(), read_write.to_compact::())); - } - - let merkle_update_handle = self - .merkle_updater - .update_and_prove::(compact_actuals, self.witness_mode.0)?; - let mut tx = self.store.new_value_tx(); for (path, read_write) in actuals { if let KeyReadWrite::Write(value) | KeyReadWrite::ReadThenWrite(_, value) = read_write { @@ -613,6 +621,52 @@ impl Session { } } +/// Hashes the values of `actuals` for the merkle update, splitting large inputs across scoped threads. +fn compact_actuals( + actuals: &[(KeyPath, KeyReadWrite)], + max_threads: usize, +) -> Vec<(KeyPath, crate::merkle::KeyReadWrite)> { + const MIN_HASHES_PER_THREAD: usize = 8192; + let compact = |chunk: &[(KeyPath, KeyReadWrite)]| { + chunk + .iter() + .map(|(path, read_write)| (*path, read_write.to_compact::())) + .collect::>() + }; + if max_threads == 1 || actuals.len() < 2 * MIN_HASHES_PER_THREAD { + return compact(actuals); + } + let hashes = actuals + .iter() + .filter(|(_, read_write)| { + matches!( + read_write, + KeyReadWrite::Write(Some(_)) | KeyReadWrite::ReadThenWrite(_, Some(_)) + ) + }) + .count(); + let threads = max_threads.min(hashes / MIN_HASHES_PER_THREAD); + if threads < 2 { + return compact(actuals); + } + let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(threads)); + if threads == 1 { + return compact(actuals); + } + let chunk = actuals.len().div_ceil(threads); + std::thread::scope(|scope| { + let handles: Vec<_> = actuals + .chunks(chunk) + .map(|chunk| scope.spawn(move || compact(chunk))) + .collect(); + let mut output = Vec::with_capacity(actuals.len()); + for handle in handles { + output.extend(handle.join().expect("compacting thread panicked")); + } + output + }) +} + /// A finished session. /// /// This is the result of completing a session and computing the merkle root and merkle DB changes, @@ -691,18 +745,13 @@ impl FinishedSession { shared.last_commit_marker = None; } - if let Some(rollback_delta) = self.rollback_delta { - // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`. - let rollback = nomt.store.rollback().unwrap(); - rollback.commit(rollback_delta)?; - } - nomt.store.commit( self.value_transaction.into_iter(), nomt.page_cache.clone(), self.merkle_output .updated_pages .into_frozen_iter(/* into_overlay */ false), + self.rollback_delta, ) } @@ -714,7 +763,7 @@ impl FinishedSession { /// The changeset may be invalidated if another competing session, overlay, or rollback was /// committed. pub fn try_commit_nonblocking( - mut self, + self, nomt: &Nomt, ) -> Result, anyhow::Error> { let write_guard = self @@ -725,15 +774,6 @@ impl FinishedSession { return Ok(Some(self)); } - if let Some(rollback_delta) = self.rollback_delta { - // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`. - let rollback = nomt.store.rollback().unwrap(); - if let Some(delta) = rollback.commit_nonblocking(rollback_delta)? { - self.rollback_delta = Some(delta); - return Ok(Some(self)); - } - } - { let mut shared = nomt.shared.lock(); if shared.root != self.prev_root { @@ -753,6 +793,7 @@ impl FinishedSession { self.merkle_output .updated_pages .into_frozen_iter(/* into_overlay */ false), + self.rollback_delta, )?; Ok(None) @@ -768,29 +809,13 @@ impl Overlay { /// overlay has an uncommitted parent. An overlay may be invalidated by a competing commit or /// rollback. pub fn commit(self, nomt: &Nomt) -> anyhow::Result<()> { - if !self.parent_matches_marker(nomt.shared.lock().last_commit_marker.as_ref()) { - anyhow::bail!("Overlay parent not committed"); - } - - let root = self.root(); - let page_changes: Vec<_> = self - .page_changes() - .into_iter() - .map(|(page_id, dirty_page)| (page_id.clone(), dirty_page.clone())) - .collect(); - let values: Vec<_> = self - .value_changes() - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect(); - let rollback_delta = self.rollback_delta().map(|delta| delta.clone()); - let _write_guard = nomt.access_lock.write(); - let marker = self.mark_committed(); - { let mut shared = nomt.shared.lock(); + if !self.parent_matches_marker(shared.last_commit_marker.as_ref()) { + anyhow::bail!("Overlay parent not committed"); + } if shared.root != self.prev_root() { anyhow::bail!( "Changeset no longer valid (expected previous root {:?}, got {:?})", @@ -798,18 +823,18 @@ impl Overlay { shared.root ); } - shared.root = root; - shared.last_commit_marker = Some(marker); + shared.root = self.root(); + shared.last_commit_marker = Some(self.mark_committed()); } - if let Some(rollback_delta) = rollback_delta { - // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`. - let rollback = nomt.store.rollback().unwrap(); - rollback.commit(rollback_delta)?; - } + let (page_changes, values, rollback_delta) = self.into_committed_changes(); - nomt.store - .commit(values, nomt.page_cache.clone(), page_changes) + nomt.store.commit( + values, + nomt.page_cache.clone(), + page_changes, + rollback_delta, + ) } /// Commit the changes from this overlay to the underlying database without blocking. @@ -823,32 +848,16 @@ impl Overlay { self, nomt: &Nomt, ) -> anyhow::Result> { - if !self.parent_matches_marker(nomt.shared.lock().last_commit_marker.as_ref()) { - anyhow::bail!("Overlay parent not committed"); - } - - let root = self.root(); - let page_changes: Vec<_> = self - .page_changes() - .into_iter() - .map(|(page_id, dirty_page)| (page_id.clone(), dirty_page.clone())) - .collect(); - let values: Vec<_> = self - .value_changes() - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect(); - let rollback_delta = self.rollback_delta().map(|delta| delta.clone()); - let write_guard = nomt.access_lock.try_write(); if write_guard.is_none() { return Ok(Some(self)); } - let marker = self.mark_committed(); - { let mut shared = nomt.shared.lock(); + if !self.parent_matches_marker(shared.last_commit_marker.as_ref()) { + anyhow::bail!("Overlay parent not committed"); + } if shared.root != self.prev_root() { anyhow::bail!( "Changeset no longer valid (expected previous root {:?}, got {:?})", @@ -856,18 +865,18 @@ impl Overlay { shared.root ); } - shared.root = root; - shared.last_commit_marker = Some(marker); + shared.root = self.root(); + shared.last_commit_marker = Some(self.mark_committed()); } - if let Some(rollback_delta) = rollback_delta { - // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`. - let rollback = nomt.store.rollback().unwrap(); - rollback.commit(rollback_delta)?; - } + let (page_changes, values, rollback_delta) = self.into_committed_changes(); - nomt.store - .commit(values, nomt.page_cache.clone(), page_changes)?; + nomt.store.commit( + values, + nomt.page_cache.clone(), + page_changes, + rollback_delta, + )?; Ok(None) } @@ -981,6 +990,54 @@ pub fn validate_hashtable(options: &Options) -> anyhow::Result = (0u64..131_073) + .map(|i| { + let mut path = [0; 32]; + path[..8].copy_from_slice(&i.to_be_bytes()); + let value = Some(i.to_le_bytes().to_vec()); + let read_write = match i % 6 { + 0 => KeyReadWrite::Read(None), + 1 => KeyReadWrite::Read(value), + 2 => KeyReadWrite::Write(None), + 3 => KeyReadWrite::Write(value), + 4 => KeyReadWrite::ReadThenWrite(value, None), + _ => KeyReadWrite::ReadThenWrite(None, value), + }; + (path, read_write) + }) + .collect(); + + for len in [0, 1, 16_383, 16_384, 49_151, 49_152, 49_153, 131_073] { + let sequential = compact_actuals::(&actuals[..len], 1); + for threads in [2, 16] { + let parallel = compact_actuals::(&actuals[..len], threads); + assert_eq!(parallel.len(), sequential.len()); + for ((path, op), (expected_path, expected_op)) in parallel.iter().zip(&sequential) { + assert_eq!(path, expected_path); + match (op, expected_op) { + (Compact::Read(value), Compact::Read(expected)) + | (Compact::Write(value), Compact::Write(expected)) => { + assert_eq!(value, expected); + } + ( + Compact::ReadThenWrite(read, value), + Compact::ReadThenWrite(expected_read, expected), + ) => { + assert_eq!(read, expected_read); + assert_eq!(value, expected); + } + _ => panic!("operation changed: {op:?} != {expected_op:?}"), + } + } + } + } + } + #[test] fn session_is_sync() { fn is_sync() {} diff --git a/nomt/src/merkle/mod.rs b/nomt/src/merkle/mod.rs index 809f2f8a11..12391914e0 100644 --- a/nomt/src/merkle/mod.rs +++ b/nomt/src/merkle/mod.rs @@ -142,33 +142,42 @@ pub enum BucketInfo { /// Whether a key was read or written. #[derive(Debug, Clone)] pub enum KeyReadWrite { - /// The key was read. - Read, + /// The key was read. Contains the read value hash or `None` if absent. + Read(Option), /// The key was written. Contains the written value or `None` if deleted. Write(Option), - /// The key was both read and written. Contains the written value of `None` if deleted. - ReadThenWrite(Option), + /// The key was both read and written. Contains the read value hash (or `None` if absent) and + /// the written value hash (or `None` if deleted). + ReadThenWrite(Option, Option), } impl KeyReadWrite { fn is_write(&self) -> bool { match self { - KeyReadWrite::Read => false, - KeyReadWrite::Write(_) | KeyReadWrite::ReadThenWrite(_) => true, + KeyReadWrite::Read(_) => false, + KeyReadWrite::Write(_) | KeyReadWrite::ReadThenWrite(_, _) => true, } } fn is_read(&self) -> bool { match self { - KeyReadWrite::Read | KeyReadWrite::ReadThenWrite(_) => true, + KeyReadWrite::Read(_) | KeyReadWrite::ReadThenWrite(_, _) => true, KeyReadWrite::Write(_) => false, } } + /// The hash of the value the key was read with, if it was read and present. + fn read_value_hash(&self) -> Option { + match self { + KeyReadWrite::Read(v) | KeyReadWrite::ReadThenWrite(v, _) => *v, + KeyReadWrite::Write(_) => None, + } + } + fn written_value(&self) -> Option> { match self { - KeyReadWrite::Read => None, - KeyReadWrite::Write(v) | KeyReadWrite::ReadThenWrite(v) => { + KeyReadWrite::Read(_) => None, + KeyReadWrite::Write(v) | KeyReadWrite::ReadThenWrite(_, v) => { Some(v.as_ref().map(|vh| *vh)) } } @@ -339,7 +348,7 @@ impl Updater { true, // collect witness ); - seeker.push(key_path); + seeker.push(key_path, None); let mut page_set = PageSet::new(self.page_pool.clone(), None); // Blocking I/O loop. diff --git a/nomt/src/merkle/page_walker.rs b/nomt/src/merkle/page_walker.rs index 93044a8fc0..a924b191bb 100644 --- a/nomt/src/merkle/page_walker.rs +++ b/nomt/src/merkle/page_walker.rs @@ -952,38 +952,24 @@ impl PageWalker { /// Count the number of leaves present *only* in the provided page, /// without jumping into child pages. fn count_leaves(page: &PageMut) -> u64 { - // A simpler linear scan cannot be done because the page could contain some garbage. + // Follow only reachable nodes: unused slots can contain garbage. A page has + // DEPTH levels, so depth-first traversal needs at most DEPTH + 1 pending nodes. + let mut pending = [0usize; DEPTH + 1]; + pending[1] = 1; + let mut len = 2; let mut counter = 0; - - // We just need the node indexes within a page, - // so we treat this as the root page starting from the root position. - let mut pos = TriePosition::new(); - let initial_depth = pos.depth(); - pos.down(false); - - loop { - let node = page.node(pos.node_index()); - // Continue to traverse the left child if the current node is internal, - // stop if we reach the end of the page. - if trie::is_internal::(&node) && pos.depth_in_page() != DEPTH { - pos.down(false); - continue; - } - - if trie::is_leaf::(&node) { - counter += 1; - } - - // Going up until I reach a leaf sibling or the node I started from. - while pos.depth() != initial_depth && pos.peek_last_bit() { - pos.up(1); - } - - if pos.depth() == initial_depth { - break; + while len != 0 { + len -= 1; + let index = pending[len]; + match NodeKind::of::(&page.node(index)) { + NodeKind::Leaf => counter += 1, + NodeKind::Internal if index < (1 << DEPTH) - 2 => { + pending[len] = index * 2 + 2; + pending[len + 1] = index * 2 + 3; + len += 2; + } + _ => {} } - - pos.sibling(); } counter } diff --git a/nomt/src/merkle/seek.rs b/nomt/src/merkle/seek.rs index c46dca66c4..d9257acbf1 100644 --- a/nomt/src/merkle/seek.rs +++ b/nomt/src/merkle/seek.rs @@ -38,6 +38,9 @@ const MAX_INFLIGHT: usize = 1024; struct SeekRequest { key: KeyPath, + /// Hash of the value the caller read for `key`, if it exists. A terminal leaf whose hash + /// matches `(key, known_value_hash)` is this key's leaf, so it needs no beatree fetch. + known_value_hash: Option, position: TriePosition, page_id: Option, siblings: Vec, @@ -50,18 +53,27 @@ impl SeekRequest { read_transaction: &BeatreeReadTx, overlay: &LiveOverlay, key: KeyPath, + known_value_hash: Option, root: Node, ) -> SeekRequest { let state = if trie::is_terminator::(&root) { RequestState::Completed(None) } else if trie::is_leaf::(&root) { - RequestState::begin_leaf_fetch::(read_transaction, overlay, &TriePosition::new()) + RequestState::resolve_leaf::( + read_transaction, + overlay, + &TriePosition::new(), + &root, + key, + known_value_hash, + ) } else { RequestState::Seeking }; let mut request = SeekRequest { key, + known_value_hash, position: TriePosition::new(), page_id: None, siblings: Vec::new(), @@ -144,8 +156,14 @@ impl SeekRequest { } if trie::is_leaf::(&cur_node) { - self.state = - RequestState::begin_leaf_fetch::(read_transaction, overlay, &self.position); + self.state = RequestState::resolve_leaf::( + read_transaction, + overlay, + &self.position, + &cur_node, + self.key, + self.known_value_hash, + ); if let RequestState::FetchingLeaf { .. } = self.state { self.continue_leaf_fetch::(None); } @@ -412,6 +430,28 @@ enum RequestState { } impl RequestState { + /// Resolves the leaf node `leaf_node` at `pos`: without I/O if it is the leaf of `key` with the + /// known value hash, otherwise by fetching the leaf data from the beatree. + fn resolve_leaf( + read_transaction: &BeatreeReadTx, + overlay: &LiveOverlay, + pos: &TriePosition, + leaf_node: &Node, + key: KeyPath, + known_value_hash: Option, + ) -> Self { + if let Some(value_hash) = known_value_hash { + let leaf = trie::LeafData { + key_path: key, + value_hash, + }; + if H::hash_leaf(&leaf) == *leaf_node { + return RequestState::Completed(Some(leaf)); + } + } + Self::begin_leaf_fetch::(read_transaction, overlay, pos) + } + fn begin_leaf_fetch( read_transaction: &BeatreeReadTx, overlay: &LiveOverlay, @@ -634,12 +674,15 @@ impl Seeker { } /// Push a request for key path. - pub fn push(&mut self, key: KeyPath) { + /// + /// `known_value_hash` is the hash of the value the caller read for `key`, if any. + pub fn push(&mut self, key: KeyPath, known_value_hash: Option) { let request_index = self.processed + self.requests.len(); self.requests.push_back(SeekRequest::new::( &self.beatree_read_transaction, &self.overlay, key, + known_value_hash, self.root, )); self.idle_requests.push_back(request_index); diff --git a/nomt/src/merkle/worker.rs b/nomt/src/merkle/worker.rs index cb6f060983..9d19015332 100644 --- a/nomt/src/merkle/worker.rs +++ b/nomt/src/merkle/worker.rs @@ -171,7 +171,7 @@ fn warm_up_phase( Err(_) => panic!("Warm-Up worker, unexpected failure of the warmup channel"), }; - seeker.push(warm_up_command.key_path); + seeker.push(warm_up_command.key_path, None); } else if index == page_idx { seeker.try_recv_page(&mut page_set)?; } else { @@ -510,7 +510,8 @@ impl RangeUpdater { break; } } else { - seeker.push(self.shared.read_write[next_push].0); + let (key, read_write) = &self.shared.read_write[next_push]; + seeker.push(*key, read_write.read_value_hash()); seeker.submit_all(page_set); } } @@ -538,10 +539,10 @@ fn subtrie_ops(read_write: &[(KeyPath, KeyReadWrite)]) -> Vec<(KeyPath, Option { + KeyReadWrite::Write(val) | KeyReadWrite::ReadThenWrite(_, val) => { Some((key.clone(), val.clone())) } - KeyReadWrite::Read => None, + KeyReadWrite::Read(_) => None, }) .collect::>() } diff --git a/nomt/src/options.rs b/nomt/src/options.rs index a44652b733..6d24f7668d 100644 --- a/nomt/src/options.rs +++ b/nomt/src/options.rs @@ -7,6 +7,7 @@ pub struct Options { pub(crate) path: PathBuf, /// The number of commit workers. Values over 64 will be rounded down to 64. pub(crate) commit_concurrency: usize, + pub(crate) actuals_concurrency: usize, /// The number of io_uring instances, or I/O threads on non-Linux platforms. pub(crate) io_workers: usize, /// Enable or disable metrics collection. @@ -29,6 +30,8 @@ pub struct Options { /// Whether to prepopulate the upper layers of the page cache on startup. /// This incurs some I/O on startup but leads to predictable worst-case performance. pub(crate) prepopulate_page_cache: bool, + /// Whether to fill the leaf cache with beatree leaves on startup. + pub(crate) prepopulate_leaf_cache: bool, pub(crate) page_cache_upper_levels: usize, } @@ -40,6 +43,7 @@ impl Options { Self { path: PathBuf::from("nomt_db"), commit_concurrency: 1, + actuals_concurrency: 16, io_workers: 3, metrics: false, bitbox_num_pages: 64_000, @@ -52,6 +56,7 @@ impl Options { page_cache_size: 256, leaf_cache_size: 256, prepopulate_page_cache: false, + prepopulate_leaf_cache: false, page_cache_upper_levels: 2, } } @@ -73,6 +78,19 @@ impl Options { self.commit_concurrency = commit_concurrency; } + /// Set the maximum number of threads hashing actuals per session. + /// + /// Limited by available CPU parallelism. Set to 1 to hash sequentially. + /// Independent of commit and I/O worker concurrency. + /// + /// Default: 16. + /// + /// Panics if `max_threads` is zero. + pub fn actuals_concurrency(&mut self, max_threads: usize) { + assert!(max_threads > 0); + self.actuals_concurrency = max_threads; + } + /// Set metrics collection on or off. /// /// Default: off. @@ -170,6 +188,15 @@ impl Options { self.prepopulate_page_cache = prepopulate; } + /// Sets whether to fill the leaf cache with beatree leaves on startup, up to its capacity. + /// + /// This reads up to [`Self::leaf_cache_size`] MiB of leaves on startup, so that sessions after + /// a restart do not pay for cold leaf reads. + /// Default: false + pub fn prepopulate_leaf_cache(&mut self, prepopulate: bool) { + self.prepopulate_leaf_cache = prepopulate; + } + /// Sets the number of upper levels of the page tree to keep permanently /// cached. /// diff --git a/nomt/src/overlay.rs b/nomt/src/overlay.rs index b6b4a28873..8cd9c453b9 100644 --- a/nomt/src/overlay.rs +++ b/nomt/src/overlay.rs @@ -55,19 +55,29 @@ impl Overlay { } } - /// Get the merkle page changes associated uniquely with this overlay. - pub(super) fn page_changes(&self) -> &HashMap { - &self.inner.data.pages - } - - /// Get the merkle page changes associated uniquely with this overlay. - pub(super) fn value_changes(&self) -> &HashMap { - &self.inner.data.values - } - - /// Get the rollback delta associated uniquely with this overlay. - pub(super) fn rollback_delta(&self) -> Option<&crate::rollback::Delta> { - self.inner.rollback_delta.as_ref() + /// Move the changes out of the overlay, cloning only what is still shared. + /// + /// Must be called after marking the overlay committed so dropping the extracted `Data` + /// does not mark the overlay as dropped. + pub(super) fn into_committed_changes( + self, + ) -> ( + HashMap, + HashMap, + Option, + ) { + let (data, rollback_delta) = match Arc::try_unwrap(self.inner) { + Ok(inner) => (inner.data, inner.rollback_delta), + Err(inner) => (Arc::clone(&inner.data), inner.rollback_delta.clone()), + }; + let (pages, values) = match Arc::try_unwrap(data) { + Ok(mut data) => ( + std::mem::take(&mut data.pages), + std::mem::take(&mut data.values), + ), + Err(data) => (data.pages.clone(), data.values.clone()), + }; + (pages, values, rollback_delta) } /// Mark the overlay as committed and return a marker. @@ -454,6 +464,93 @@ mod tests { assert!(LiveOverlay::new(None).is_ok()); } + fn overlay_with_changes() -> super::Overlay { + LiveOverlay::new(None).unwrap().finish( + [0; 32], + [1; 32], + [( + ROOT_PAGE_ID, + dummy_page(ROOT_PAGE_ID, 1, BucketInfo::FreshWithNoDependents), + )] + .into_iter() + .collect(), + [([1; 32], ValueChange::Insert(vec![1, 2, 3]))] + .into_iter() + .collect(), + Some(crate::rollback::Delta { + priors: [([1; 32], Some(vec![4, 5, 6]))].into_iter().collect(), + }), + ) + } + + #[test] + fn committed_changes_move_with_weak_descendants() { + let overlay = overlay_with_changes(); + let child = LiveOverlay::new([&overlay]).unwrap().finish( + [1; 32], + [2; 32], + HashMap::new(), + HashMap::new(), + None, + ); + let value_ptr = overlay.inner.data.values[&[1; 32]] + .as_option() + .unwrap() + .as_ptr(); + let prior_ptr = overlay.inner.rollback_delta.as_ref().unwrap().priors[&[1; 32]] + .as_ref() + .unwrap() + .as_ptr(); + let marker = overlay.mark_committed(); + + let (pages, values, delta) = overlay.into_committed_changes(); + + assert_eq!(pages[&ROOT_PAGE_ID].page.node(0), [1; 32]); + assert_eq!(values[&[1; 32]], ValueChange::Insert(vec![1, 2, 3])); + assert_eq!(values[&[1; 32]].as_option().unwrap().as_ptr(), value_ptr); + let prior = delta.as_ref().unwrap().priors[&[1; 32]].as_ref().unwrap(); + assert_eq!(prior, &[4, 5, 6]); + assert_eq!(prior.as_ptr(), prior_ptr); + assert!(child.parent_matches_marker(Some(&marker))); + assert!(LiveOverlay::new([&child]).is_ok()); + assert!(marker.0.is_committed()); + } + + #[test] + fn committed_changes_preserve_live_parent_and_ancestor_data() { + for retain_parent in [true, false] { + let overlay = overlay_with_changes(); + let child = LiveOverlay::new([&overlay]).unwrap().finish( + [1; 32], + [2; 32], + HashMap::new(), + HashMap::new(), + None, + ); + let live = if retain_parent { + LiveOverlay::new([&overlay]).unwrap() + } else { + LiveOverlay::new([&child, &overlay]).unwrap() + }; + let marker = overlay.mark_committed(); + + let (pages, values, delta) = overlay.into_committed_changes(); + + assert_eq!(pages[&ROOT_PAGE_ID].page.node(0), [1; 32]); + assert_eq!(values[&[1; 32]], ValueChange::Insert(vec![1, 2, 3])); + assert_eq!(delta.unwrap().priors[&[1; 32]], Some(vec![4, 5, 6])); + assert_eq!(live.page(&ROOT_PAGE_ID).unwrap().page.node(0), [1; 32]); + assert_eq!( + live.value(&[1; 32]), + Some(ValueChange::Insert(vec![1, 2, 3])) + ); + assert!(child.parent_matches_marker(Some(&marker))); + assert!(LiveOverlay::new([&child]).is_ok()); + drop(live); + assert!(marker.0.is_committed()); + } + } + #[test] fn not_ancestor() { let a = LiveOverlay::new(None).unwrap().finish( diff --git a/nomt/src/rollback/mod.rs b/nomt/src/rollback/mod.rs index b2d0f176a6..3deaa65b0b 100644 --- a/nomt/src/rollback/mod.rs +++ b/nomt/src/rollback/mod.rs @@ -173,27 +173,6 @@ impl Rollback { Ok(()) } - /// Saves the delta into the log, unless the locks are already held by another thread. - /// - /// If commit is blocked, it returns the delta back to the caller. - pub fn commit_nonblocking(&self, delta: Delta) -> anyhow::Result> { - let delta_bytes = delta.encode(); - - // Try to lock the in-memory log and the seglog. - let mut in_memory = match self.shared.in_memory.try_lock() { - Some(lock) => lock, - None => return Ok(Some(delta)), // Another thread is holding the lock. - }; - let mut seglog = match self.shared.seglog.try_lock() { - Some(lock) => lock, - None => return Ok(Some(delta)), // Another thread is holding the lock. - }; - - let record_id = seglog.append(&delta_bytes)?; - in_memory.push_recent(record_id, delta); - Ok(None) - } - /// Truncates the rollback log by removing the last `n` deltas. /// /// This function returns the keys and values that we should apply to the database to restore diff --git a/nomt/src/store/mod.rs b/nomt/src/store/mod.rs index 7cb81a6886..6eebdebd7c 100644 --- a/nomt/src/store/mod.rs +++ b/nomt/src/store/mod.rs @@ -243,6 +243,11 @@ impl Store { } } + /// Fill the beatree leaf cache on `threads` threads, up to its capacity. + pub fn prepopulate_leaf_cache(&self, threads: usize) { + self.shared.values.prepopulate_leaf_cache(threads); + } + /// Creates a new [`beatree::ReadTransaction`]. `sync` will be blocked until this is dropped. pub fn read_transaction(&self) -> beatree::ReadTransaction { self.shared.values.read_transaction() @@ -278,6 +283,7 @@ impl Store { value_tx: impl IntoIterator + Send + 'static, page_cache: PageCache, updated_pages: impl IntoIterator + Send + 'static, + rollback_delta: Option, ) -> anyhow::Result<()> { let mut sync = self.sync.lock(); @@ -295,6 +301,7 @@ impl Store { self.shared.pages.clone(), self.shared.values.clone(), self.shared.rollback.clone(), + rollback_delta, page_cache, updated_pages, ) { diff --git a/nomt/src/store/sync.rs b/nomt/src/store/sync.rs index 0ac076351a..12b53b01d5 100644 --- a/nomt/src/store/sync.rs +++ b/nomt/src/store/sync.rs @@ -4,7 +4,9 @@ use super::{ meta::{self, Meta}, DirtyPage, Shared, }; -use crate::{beatree, bitbox, options::PanicOnSyncMode, page_cache::PageCache, rollback}; +use crate::{ + beatree, bitbox, options::PanicOnSyncMode, page_cache::PageCache, rollback, task::join_task, +}; pub struct Sync { pub(crate) sync_seqn: u32, @@ -35,6 +37,7 @@ impl Sync { bitbox: bitbox::DB, beatree: beatree::Tree, rollback: Option, + rollback_delta: Option, page_cache: PageCache, updated_pages: impl IntoIterator + Send + 'static, ) -> anyhow::Result<()> { @@ -42,18 +45,29 @@ impl Sync { let mut bitbox_sync = bitbox.sync(); let mut beatree_sync = beatree.sync(); - let mut rollback_sync = rollback.map(|rollback| rollback.sync()); + let mut rollback_sync = rollback.as_ref().map(|rollback| rollback.sync()); bitbox_sync.begin_sync(sync_seqn, page_cache, updated_pages); beatree_sync.begin_sync(value_tx); + + // Persist the rollback delta while the trees are being prepared. All of them must + // succeed before the manifest references their new state. + let rollback_result = match rollback_delta { + // UNWRAP: a rollback delta is only produced when rollback is enabled. + Some(delta) => rollback.as_ref().unwrap().commit(delta), + None => Ok(()), + }; + let bitbox_result = bitbox_sync.wait_pre_meta(); + let beatree_result = beatree_sync.wait_pre_meta(); + rollback_result?; + bitbox_result?; + let beatree_meta_wd = beatree_result?; + let (rollback_start_live, rollback_end_live) = match rollback_sync { Some(ref mut rollback) => rollback.begin_sync(), None => (0, 0), }; - bitbox_sync.wait_pre_meta()?; - let beatree_meta_wd = beatree_sync.wait_pre_meta()?; - if let Some(PanicOnSyncMode::PostWal) = self.panic_on_sync { panic!("panic_on_sync is true (post-wal)") } @@ -82,8 +96,11 @@ impl Sync { rollback.post_meta(); } - bitbox_sync.post_meta(shared.io_pool.make_handle())?; - beatree_sync.post_meta(); + // Finish the beatree sync while the hash-table pages are written out. + let beatree_post_meta = beatree_sync.post_meta(); + let bitbox_result = bitbox_sync.post_meta(shared.io_pool.make_handle()); + join_task(&beatree_post_meta); + bitbox_result?; if let Some(ref rollback) = rollback_sync { rollback.wait_post_meta()?; diff --git a/nomt/tests/prepopulate_leaf_cache.rs b/nomt/tests/prepopulate_leaf_cache.rs new file mode 100644 index 0000000000..0d2e8b39e3 --- /dev/null +++ b/nomt/tests/prepopulate_leaf_cache.rs @@ -0,0 +1,42 @@ +use nomt::{hasher::Blake3Hasher, KeyReadWrite, Nomt, Options, SessionParams}; +use nomt_test_utils::account_path; +use std::path::PathBuf; + +fn open(path: &PathBuf, prepopulate: bool) -> Nomt { + let mut o = Options::new(); + o.path(path); + o.bitbox_seed([0; 16]); + o.commit_concurrency(2); + o.io_workers(1); + o.leaf_cache_size(1); + o.prepopulate_leaf_cache(prepopulate); + Nomt::open(o).unwrap() +} + +fn value(id: u64) -> Vec { + vec![id as u8; 400] +} + +#[test] +fn prepopulate_leaf_cache_reads_back_values() { + let path = PathBuf::from("test/prepopulate_leaf_cache"); + let _ = std::fs::remove_dir_all(&path); + // ~1.6 MiB of values: more leaves than a 1 MiB leaf cache holds. + let n = 4000; + + { + let nomt = open(&path, false); + let session = nomt.begin_session(SessionParams::default()); + let mut actuals: Vec<_> = (0..n) + .map(|id| (account_path(id), KeyReadWrite::Write(Some(value(id))))) + .collect(); + actuals.sort_by_key(|(k, _)| *k); + session.finish(actuals).unwrap().commit(&nomt).unwrap(); + } + + let nomt = open(&path, true); + let session = nomt.begin_session(SessionParams::default()); + for id in 0..n { + assert_eq!(session.read(account_path(id)).unwrap(), Some(value(id))); + } +}