[no merge] optimistic 1.1.0

core/src/update.rs

@@ -160,9 +160,16 @@ pub fn build_trie<H: NodeHasher>(
     }
 
     let common_after_prefix = |k1: &KeyPath, k2: &KeyPath| {
-        let x = &k1.view_bits::<Msb0>()[skip..];
-        let y = &k2.view_bits::<Msb0>()[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 {

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<Item = &Arc<BranchNode>> {
+        self.first_key_map.values()
+    }
+
     #[cfg(test)]
     pub fn into_iter(self) -> impl Iterator<Item = (Key, Arc<BranchNode>)> {
         self.first_key_map.into_iter()

nomt/src/beatree/iterator.rs

@@ -235,6 +235,27 @@ impl LeafIterator {
     }
 
     fn next<'a>(&'a mut self) -> Option<IterOutput<'a>> {
+        // 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

nomt/src/beatree/leaf_cache.rs

@@ -43,13 +43,27 @@ 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<LeafNode>) {
         let mut shard = self.inner.shard_for(page_number);
 
         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 {

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<PageNumber> = 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<Vec<u8>> {
         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<TaskResult<()>> {
+        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
     }
 }
 

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);
                 }