Revert to linear range checking, but keep short-circuiting

ConflictSet.cpp

@@ -686,39 +686,21 @@ struct FirstGeqStepwise {
   Node *nextSib = nullptr;
   int cmp;
 
-  enum Phase { Search, DownLeftSpine };
+  enum Phase {
+    Init,
+    // Being in this phase implies that the key matches the search path exactly
+    // up to this point
+    Search,
+    DownLeftSpine
+  };
   Phase phase;
 
   FirstGeqStepwise(Node *n, std::span<const uint8_t> remaining)
-      : n(n), remaining(remaining), phase(Search) {}
+      : n(n), remaining(remaining), phase(Init) {}
 
   bool step() {
     switch (phase) {
     case Search:
-      if (n->partialKeyLen > 0) {
-        int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
-        for (int i = 0; i < commonLen; ++i) {
-          auto c = n->partialKey[i] <=> remaining[i];
-          if (c == 0) {
-            continue;
-          }
-          if (c > 0) {
-            return downLeftSpine();
-          } else {
-            n = nextSib;
-            return downLeftSpine();
-          }
-        }
-        if (commonLen == n->partialKeyLen) {
-          // partial key matches
-          remaining =
-              remaining.subspan(commonLen, remaining.size() - commonLen);
-        } else if (n->partialKeyLen > int(remaining.size())) {
-          // n is the first physical node greater than remaining, and there's no
-          // eq node
-          return downLeftSpine();
-        }
-      }
       if (remaining.size() == 0) {
         if (n->entryPresent) {
           cmp = 0;
@@ -747,6 +729,32 @@ struct FirstGeqStepwise {
           }
         }
       }
+    case Init:
+      phase = Search;
+      if (n->partialKeyLen > 0) {
+        int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
+        for (int i = 0; i < commonLen; ++i) {
+          auto c = n->partialKey[i] <=> remaining[i];
+          if (c == 0) {
+            continue;
+          }
+          if (c > 0) {
+            return downLeftSpine();
+          } else {
+            n = nextSib;
+            return downLeftSpine();
+          }
+        }
+        if (commonLen == n->partialKeyLen) {
+          // partial key matches
+          remaining =
+              remaining.subspan(commonLen, remaining.size() - commonLen);
+        } else if (n->partialKeyLen > int(remaining.size())) {
+          // n is the first physical node greater than remaining, and there's no
+          // eq node
+          return downLeftSpine();
+        }
+      }
       return false;
     case DownLeftSpine:
       if (n->entryPresent) {
@@ -862,8 +870,10 @@ bool checkRangeRead(Node *n, const std::span<const uint8_t> begin,
                     const std::span<const uint8_t> end, int64_t readVersion) {
   auto left = FirstGeqStepwise{n, begin};
   auto right = FirstGeqStepwise{n, end};
-  bool leftDone;
-  bool rightDone;
+  bool leftDone = left.step();
+  bool rightDone = right.step();
+  assert(!leftDone);
+  assert(!rightDone);
   for (;;) {
     if (left.phase == FirstGeqStepwise::Search &&
         right.phase == FirstGeqStepwise::Search &&
@@ -880,62 +890,44 @@ bool checkRangeRead(Node *n, const std::span<const uint8_t> begin,
     }
   }
 
-  if (!leftDone && !rightDone) {
-    assert(left.n->parent == right.n->parent);
-    for (int c = left.n->parentsIndex; c < right.n->parentsIndex;
-         c = getChildGeq(left.n->parent, c + 1)) {
-      assert(c >= 0);
-      if (getChildExists(left.n->parent, c)->maxVersion > readVersion) {
-        return false;
-      }
-    }
-    // We've checked everything from begin to the search path of right.n
-    // Now check from the search path of right.n to end
-    for (;;) {
-      if (right.phase == FirstGeqStepwise::Search &&
-          right.n->maxVersion <= readVersion) {
-        return true;
-      }
-      rightDone = right.step();
-      if (rightDone) {
-        return right.n->entry.rangeVersion <= readVersion;
-      }
-    }
+  // TODO make it sublinear
+  if (!leftDone) {
+    while (!left.step())
+      ;
   }
-
-  if (leftDone) {
-    if (left.n == nullptr) {
-      return true;
-    }
-    if (right.phase == FirstGeqStepwise::DownLeftSpine) {
-      if (left.n->maxVersion > readVersion) {
-        return false;
-      }
-    }
-
-    // We've checked everything from begin to the search path of right.n
-    // Now check from the search path of right.n to end
-    for (;;) {
-      if (right.phase == FirstGeqStepwise::Search &&
-          right.n->maxVersion <= readVersion) {
-        return true;
-      }
-      rightDone = right.step();
-      if (rightDone) {
-        return right.n->entry.rangeVersion <= readVersion;
-      }
-    }
+  if (!rightDone) {
+    while (!right.step())
+      ;
   }
-
-  if (rightDone) {
-    // This would mean that left is overtaking right
-    __builtin_unreachable();
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+  fprintf(stderr, "Left firstGeq: %s, right firstGeq: %s\n",
+          getSearchPathPrintable(left.n).c_str(),
+          getSearchPathPrintable(right.n).c_str());
+#endif
+  if (left.n != nullptr && left.cmp != 0 &&
+      left.n->entry.rangeVersion > readVersion) {
+    return false;
   }
-
-  {
-    assert(left.n == right.n);
+  if (left.n == right.n) {
     return true;
   }
+  assert(left.n != nullptr);
+  auto boundaryVersion = left.n->entry.pointVersion;
+  // Already checked left rangeVersion
+  if (right.cmp == 0) {
+    boundaryVersion = std::max(boundaryVersion, right.n->entry.rangeVersion);
+  }
+  if (boundaryVersion > readVersion) {
+    return false;
+  }
+  for (auto *iter = nextLogical(left.n); iter != right.n;
+       iter = nextLogical(iter)) {
+    if (std::max(iter->entry.pointVersion, iter->entry.rangeVersion) >
+        readVersion) {
+      return false;
+    }
+  }
+  return true;
 }
 
 // Returns a pointer to the newly inserted node.  caller is reponsible for