From ae2b50d7653a0b0d9f73d6419d3fe00765bb0eea Mon Sep 17 00:00:00 2001
From: Andrew Noyes <andrew@weaselab.dev>
Date: Mon, 3 Jun 2024 12:26:41 -0700
Subject: [PATCH] Interleave key search

---
 VersionedMap.cpp | 28 +++++++++++++++++++++++-----
 1 file changed, 23 insertions(+), 5 deletions(-)

diff --git a/VersionedMap.cpp b/VersionedMap.cpp
index dffa09c..7f5a51d 100644
--- a/VersionedMap.cpp
+++ b/VersionedMap.cpp
@@ -1229,6 +1229,9 @@ bool geq(const VersionedMap::Iterator::VersionedMutation &m,
 void VersionedMap::Impl::firstGeq(const weaselab::VersionedMap::Key *key,
                                   const int64_t *version, Iterator *iterator,
                                   int count) const {
+  if (count == 0) {
+    return;
+  }
 
   struct StepwiseFirstGeq {
 
@@ -1236,26 +1239,41 @@ void VersionedMap::Impl::firstGeq(const weaselab::VersionedMap::Key *key,
     const weaselab::VersionedMap::Key *key;
     const int64_t *version;
     weaselab::VersionedMap::Iterator *iterator;
-    uint32_t node;
 
     void begin(RootSet::ThreadSafeHandle handle) {
+      uint32_t root;
       if (iterator->impl != nullptr) {
-        node = iterator->impl->version == *version
+        root = iterator->impl->version == *version
                    ? iterator->impl->finger.root()
                    : handle.rootForVersion(*version);
         iterator->impl->~Impl();
         new (iterator->impl) Iterator::Impl();
       } else {
-        node = handle.rootForVersion(*version);
+        root = handle.rootForVersion(*version);
         iterator->impl =
             new (safe_malloc(sizeof(Iterator::Impl))) Iterator::Impl();
       }
+      Finger &finger = iterator->impl->finger;
+      finger.clear();
+      bool ignored = false;
+      finger.push(root, ignored);
     }
 
     bool step() {
       Finger &finger = iterator->impl->finger;
-      map->search<std::memory_order_acquire>(*key, node, *version, finger);
-      return true;
+
+      auto n = finger.backNode();
+      if (n == 0) {
+        return true;
+      }
+      auto c = *key <=> map->mm.base[n];
+      if (c == 0) {
+        // No duplicates
+        return true;
+      }
+      finger.push(map->child<std::memory_order_acquire>(n, c > 0, *version),
+                  c > 0);
+      return false;
     }
 
     void end() {