Reworking the representation for the beginning of a clear range

Most tests pass. Some tests fail.

VersionedMap.cpp

@@ -672,15 +672,8 @@ struct __attribute__((__visibility__("hidden"))) VersionedMap::Impl {
       // Beginning of a clear range
       pointVersion = -1; // Sentinel for "no point mutation here"
       if (inserted) {
-        rangeVersion = -1; // Sentinel for "no mutation ending here"
-#ifndef NDEBUG
         // If there were a clear range here, it wouldn't be canonical
-        Finger copy;
+        rangeVersion = -1; // Sentinel for "no mutation ending here"
-        finger.copyTo(copy);
-        move<std::memory_order_relaxed, true>(copy, latestVersion);
-        assert(copy.searchPathSize() == 0 ||
-               mm.base[copy.backNode()].entry->rangeVersion < 0);
-#endif
       } else {
         auto *entry = mm.base[finger.backNode()].entry;
         rangeVersion = entry->rangeVersion;
@@ -829,9 +822,9 @@ struct __attribute__((__visibility__("hidden"))) VersionedMap::Impl {
 
   int64_t getBytes() const { return totalMallocBytes + mm.getBytes(); }
 
-  void printInOrder(int64_t version);
+  void printInOrder(int64_t version) const;
 
-  void printInOrderHelper(int64_t version, uint32_t node, int depth);
+  void printInOrderHelper(int64_t version, uint32_t node, int depth) const;
 
   int accumulatedFuel = 0;
 
@@ -922,9 +915,9 @@ struct __attribute__((__visibility__("hidden"))) VersionedMap::Impl {
     // TODO Improve ILP?
     for (int i = 0; i < numMutations; ++i) {
       const auto &m = mutations[i];
-      Finger iter;
       switch (m.type) {
       case Set: {
+        Finger iter;
         search<std::memory_order_relaxed>({m.param1, m.param1Len}, latestRoot,
                                           latestVersion, iter);
         insert({m.param1, m.param1Len}, {{m.param2, m.param2Len}},
@@ -935,6 +928,7 @@ struct __attribute__((__visibility__("hidden"))) VersionedMap::Impl {
         // hand though.
 
         if (m.param2Len == 0) {
+          Finger iter;
           search<std::memory_order_relaxed>({m.param1, m.param1Len}, latestRoot,
                                             latestVersion, iter);
           const bool found = iter.searchPathSize() > 0 && iter.backNode() != 0;
@@ -972,43 +966,76 @@ struct __attribute__((__visibility__("hidden"))) VersionedMap::Impl {
                    iter);
           }
         } else {
+          // TODO ILP these
+          Finger begin;
           search<std::memory_order_relaxed>({m.param1, m.param1Len}, latestRoot,
-                                            latestVersion, iter);
+                                            latestVersion, begin);
-          insert({m.param1, m.param1Len}, {}, /*endRange*/ false, iter);
+          const bool foundBegin =
+              begin.searchPathSize() > 0 && begin.backNode() != 0;
+
+          Finger end;
+          search<std::memory_order_relaxed>({m.param2, m.param2Len}, latestRoot,
+                                            latestVersion, end);
+          const bool foundEnd = end.searchPathSize() > 0 && end.backNode() != 0;
 
           // Check if we can engulf on the left
-          {
+          bool engulfLeft;
-            const auto *entry = mm.base[iter.backNode()].entry;
+          Finger copy;
-            if (entry->clearTo()) {
+          begin.copyTo(copy);
-              remove(iter);
+          move<std::memory_order_relaxed, true>(copy, latestVersion);
-            }
+          if (foundBegin) {
+            engulfLeft = begin.searchPathSize() > 0 &&
+                         mm.base[begin.backNode()].entry->clearTo();
+          } else {
+            engulfLeft = copy.searchPathSize() > 0 &&
+                         mm.base[copy.backNode()].entry->clearTo();
           }
 
-          move<std::memory_order_relaxed, true>(iter, latestVersion);
-          while (iter.searchPathSize() > 0 &&
-                 mm.base[iter.backNode()] < Key{m.param2, m.param2Len}) {
-            remove(iter);
-            move<std::memory_order_relaxed, true>(iter, latestVersion);
-          }
-          // TODO reuse finger? It should be one rank away from its insertion
-          // point
-          search<std::memory_order_relaxed>({m.param2, m.param2Len}, latestRoot,
-                                            latestVersion, iter);
-
           // Check if we can engulf on the right
-          {
+          end.copyTo(copy);
-            Finger copy;
+          move<std::memory_order_relaxed, true>(copy, latestVersion);
-            iter.copyTo(copy);
+          const auto *next = copy.searchPathSize() > 0
-            move<std::memory_order_relaxed, true>(copy, latestVersion);
+                                 ? mm.base[copy.backNode()].entry
-            const auto *next = copy.searchPathSize() > 0
+                                 : nullptr;
-                                   ? mm.base[copy.backNode()].entry
+          const bool engulfRight = next && next->clearTo();
-                                   : nullptr;
+
-            if (next && next->clearTo()) {
+          if (engulfLeft && foundBegin) {
-              insert({next->getKey(), next->keyLen}, {}, /*endRange*/ true,
+            remove(begin);
-                     copy);
+          } else if (!engulfLeft) {
-            } else {
+            insert({m.param1, m.param1Len}, {}, /*rangeEntry*/ false, begin);
-              insert({m.param2, m.param2Len}, {}, /*endRange*/ true, iter);
+          }
+          move<std::memory_order_relaxed, true>(begin, latestVersion);
+
+          while (begin.searchPathSize() > 0 &&
+                 mm.base[begin.backNode()] < Key{m.param2, m.param2Len}) {
+            remove(begin);
+            move<std::memory_order_relaxed, true>(begin, latestVersion);
+          }
+
+#ifndef NDEBUG
+          if (foundEnd) {
+            [[maybe_unused]] bool beginEqEnd =
+                mm.base[begin.backNode()] <=> Key{m.param2, m.param2Len} == 0;
+            assert(beginEqEnd);
+          }
+#endif
+
+          if (engulfRight) {
+            if (foundEnd) {
+              remove(begin);
+              move<std::memory_order_relaxed, true>(begin, latestVersion);
             }
+            assert(begin.searchPathSize() > 0 && begin.backNode() != 0);
+            insert({mm.base[begin.backNode()].entry->getKey(),
+                    mm.base[begin.backNode()].entry->keyLen},
+                   {}, /*rangeEntry*/ true, begin);
+          } else {
+            if (!foundEnd) {
+              // TODO remove this search
+              search<std::memory_order_relaxed>(
+                  {m.param2, m.param2Len}, latestRoot, latestVersion, begin);
+            }
+            insert({m.param2, m.param2Len}, {}, /*rangeEntry*/ true, begin);
           }
         }
       } break;
@@ -1553,7 +1580,7 @@ int64_t VersionedMap::getBytes() const { return impl->getBytes(); }
 inline
 #endif
     void
-    VersionedMap::Impl::printInOrder(int64_t version) {
+    VersionedMap::Impl::printInOrder(int64_t version) const {
   printInOrderHelper(version,
                      roots.getThreadSafeHandle().rootForVersion(version), 0);
 }
@@ -1563,7 +1590,7 @@ inline
 #endif
     void
     VersionedMap::Impl::printInOrderHelper(int64_t version, uint32_t node,
-                                           int depth) {
+                                           int depth) const {
   if (node == 0) {
     return;
   }