From 690f6e19ef41765e39dcc35814dffb65b37d58dc Mon Sep 17 00:00:00 2001 From: Nikolai Golub Date: Wed, 21 Oct 2026 10:46:01 +0200 Subject: [PATCH 1/3] Hash session actuals for the merkle update on multiple threads Co-Authored-By: Claude Opus 5.5 (1M context) --- nomt/src/lib.rs | 35 +++++++++++++++++++++++++++++------ 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs index 3f543f78ca..b5ad8ff04b 100644 --- a/nomt/src/lib.rs +++ b/nomt/src/lib.rs @@ -580,14 +580,9 @@ impl Session { ); } } - 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)?; + .update_and_prove::(compact_actuals::(&actuals), 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. @@ -615,6 +610,34 @@ impl Session { } } +/// Hashes the values of `actuals` for the merkle update, splitting large inputs across scoped threads. +fn compact_actuals( + actuals: &[(KeyPath, KeyReadWrite)], +) -> Vec<(KeyPath, crate::merkle::KeyReadWrite)> { + const MIN_CHUNK: usize = 2048; + let compact = |chunk: &[(KeyPath, KeyReadWrite)]| { + chunk + .iter() + .map(|(path, read_write)| (*path, read_write.to_compact::())) + .collect::>() + }; + let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(16)); + let chunk = actuals.len().div_ceil(threads).max(MIN_CHUNK); + if actuals.len() <= chunk { + return compact(actuals); + } + std::thread::scope(|scope| { + let handles: Vec<_> = actuals + .chunks(chunk) + .map(|chunk| scope.spawn(move || compact(chunk))) + .collect(); + handles + .into_iter() + .flat_map(|handle| handle.join().expect("compacting thread panicked")) + .collect() + }) +} + /// A finished session. /// /// This is the result of completing a session and computing the merkle root and merkle DB changes, From 48bc2eeb65bbce1beba4c1d201a73252e72011fc Mon Sep 17 00:00:00 2001 From: Nikolai Golub Date: Wed, 21 Oct 2026 11:48:31 +0200 Subject: [PATCH 2/3] Parametrize concurrency --- nomt/src/lib.rs | 14 ++++++++++---- nomt/src/options.rs | 15 +++++++++++++++ 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs index b5ad8ff04b..0b68643e06 100644 --- a/nomt/src/lib.rs +++ b/nomt/src/lib.rs @@ -197,6 +197,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, @@ -237,6 +238,7 @@ impl Nomt { 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 +330,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 +480,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,9 +584,10 @@ impl Session { ); } } - let merkle_update_handle = self - .merkle_updater - .update_and_prove::(compact_actuals::(&actuals), self.witness_mode.0)?; + 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. @@ -613,6 +618,7 @@ 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_CHUNK: usize = 2048; let compact = |chunk: &[(KeyPath, KeyReadWrite)]| { @@ -621,7 +627,7 @@ fn compact_actuals( .map(|(path, read_write)| (*path, read_write.to_compact::())) .collect::>() }; - let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(16)); + let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(max_threads)); let chunk = actuals.len().div_ceil(threads).max(MIN_CHUNK); if actuals.len() <= chunk { return compact(actuals); diff --git a/nomt/src/options.rs b/nomt/src/options.rs index a44652b733..c0f9a1436e 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. @@ -40,6 +41,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, @@ -73,6 +75,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. From 6b802a0f4167c657f3ed7da5cf1c79b0e722e4f0 Mon Sep 17 00:00:00 2001 From: Nikolai Golub Date: Wed, 21 Oct 2026 11:57:58 +0200 Subject: [PATCH 3/3] smarter parallelism --- nomt/src/lib.rs | 75 +++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 67 insertions(+), 8 deletions(-) diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs index 0b68643e06..0852cdf4b9 100644 --- a/nomt/src/lib.rs +++ b/nomt/src/lib.rs @@ -620,27 +620,44 @@ fn compact_actuals( actuals: &[(KeyPath, KeyReadWrite)], max_threads: usize, ) -> Vec<(KeyPath, crate::merkle::KeyReadWrite)> { - const MIN_CHUNK: usize = 2048; + const MIN_HASHES_PER_THREAD: usize = 8192; let compact = |chunk: &[(KeyPath, KeyReadWrite)]| { chunk .iter() .map(|(path, read_write)| (*path, read_write.to_compact::())) .collect::>() }; - let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(max_threads)); - let chunk = actuals.len().div_ceil(threads).max(MIN_CHUNK); - if actuals.len() <= chunk { + 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(); - handles - .into_iter() - .flat_map(|handle| handle.join().expect("compacting thread panicked")) - .collect() + let mut output = Vec::with_capacity(actuals.len()); + for handle in handles { + output.extend(handle.join().expect("compacting thread panicked")); + } + output }) } @@ -984,6 +1001,48 @@ 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, Compact::Read) => {} + (Compact::Write(value), Compact::Write(expected)) + | (Compact::ReadThenWrite(value), Compact::ReadThenWrite(expected)) => { + assert_eq!(value, expected); + } + _ => panic!("operation changed: {op:?} != {expected_op:?}"), + } + } + } + } + } + #[test] fn session_is_sync() { fn is_sync() {}