Short-circuiting, efficient checkRangeRead

ConflictSet.cpp

@@ -680,73 +680,101 @@ std::string getSearchPath(Node *n) {
 }
 } // namespace
 
-Iterator firstGeq(Node *n, const std::span<const uint8_t> key) {
+struct FirstGeqStepwise {
-  auto remaining = key;
+  Node *n;
+  std::span<const uint8_t> remaining;
   Node *nextSib = nullptr;
-  for (;;) {
+  int cmp;
-    if (n->partialKeyLen > 0) {
+
-      int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
+  enum Phase { Search, DownLeftSpine };
-      for (int i = 0; i < commonLen; ++i) {
+  Phase phase;
-        auto c = n->partialKey[i] <=> remaining[i];
+
-        if (c == 0) {
+  FirstGeqStepwise(Node *n, std::span<const uint8_t> remaining)
-          continue;
+      : n(n), remaining(remaining), phase(Search) {}
+
+  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 (c > 0) {
+        if (commonLen == n->partialKeyLen) {
-          goto downLeftSpine;
+          // 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;
+          return true;
+        }
+        int c = getChildGeq(n, 0);
+        assert(c >= 0);
+        n = getChildExists(n, c);
+        return downLeftSpine();
+      } else {
+        int c = getChildGeq(n, remaining[0]);
+        int c2 = getChildGeq(n, int(remaining[0]) + 1);
+        if (c2 >= 0) {
+          nextSib = getChildExists(n, c2);
+        }
+        if (c == remaining[0]) {
+          n = getChildExists(n, c);
+          remaining = remaining.subspan(1, remaining.size() - 1);
         } else {
-          n = nextSib;
+          if (c >= 0) {
-          goto downLeftSpine;
+            n = getChildExists(n, c);
+            return downLeftSpine();
+          } else {
+            n = nextSib;
+            return downLeftSpine();
+          }
         }
       }
-      if (commonLen == n->partialKeyLen) {
+      return false;
-        // partial key matches
+    case DownLeftSpine:
-        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
-        goto downLeftSpine;
-      }
-    }
-    if (remaining.size() == 0) {
       if (n->entryPresent) {
-        return {n, 0};
+        cmp = 1;
+        return true;
       }
       int c = getChildGeq(n, 0);
       assert(c >= 0);
       n = getChildExists(n, c);
-      goto downLeftSpine;
+      return false;
-    } else {
-      int c = getChildGeq(n, remaining[0]);
-      int c2 = getChildGeq(n, int(remaining[0]) + 1);
-      if (c2 >= 0) {
-        nextSib = getChildExists(n, c2);
-      }
-      if (c == remaining[0]) {
-        n = getChildExists(n, c);
-        remaining = remaining.subspan(1, remaining.size() - 1);
-      } else {
-        if (c >= 0) {
-          n = getChildExists(n, c);
-          goto downLeftSpine;
-        } else {
-          n = nextSib;
-          goto downLeftSpine;
-        }
-      }
     }
   }
-downLeftSpine:
+
-  if (n == nullptr) {
+  bool downLeftSpine() {
-    return {nullptr, 1};
+    if (n == nullptr) {
-  }
+      cmp = 1;
-  for (;;) {
+      return true;
-    if (n->entryPresent) {
-      return {n, 1};
     }
-    int c = getChildGeq(n, 0);
+    phase = DownLeftSpine;
-    assert(c >= 0);
+    return step();
-    n = getChildExists(n, c);
   }
+};
+
+Iterator firstGeq(Node *n, const std::span<const uint8_t> key) {
+  FirstGeqStepwise stepwise{n, key};
+  while (!stepwise.step())
+    ;
+  return {stepwise.n, stepwise.cmp};
 }
 
 Iterator firstGeq(Node *n, std::string_view key) {
@@ -832,84 +860,82 @@ downLeftSpine:
 
 bool checkRangeRead(Node *n, const std::span<const uint8_t> begin,
                     const std::span<const uint8_t> end, int64_t readVersion) {
-  auto left = firstGeq(n, begin);
+  auto left = FirstGeqStepwise{n, begin};
-  auto right = firstGeq(n, end);
+  auto right = FirstGeqStepwise{n, end};
-
+  bool leftDone;
-  Arena arena;
+  bool rightDone;
-  auto leftPath = vector<Node *>(arena);
+  for (;;) {
-  auto rightPath = vector<Node *>(arena);
+    if (left.phase == FirstGeqStepwise::Search &&
-  for (auto *iter = left.n; iter != nullptr; iter = iter->parent) {
+        right.phase == FirstGeqStepwise::Search &&
-    leftPath.push_back(iter);
+        left.n->maxVersion <= readVersion) {
-  }
+      return true;
-  for (auto *iter = right.n; iter != nullptr; iter = iter->parent) {
+    }
-    rightPath.push_back(iter);
+    leftDone = left.step();
-  }
+    rightDone = right.step();
-  Node *lca = n;
+    if (leftDone || rightDone) {
-  for (int i = 0; int(leftPath.size()) - 1 - i >= 0 &&
+      break;
-                  int(rightPath.size()) - 1 - i >= 0 &&
+    }
-                  leftPath[int(leftPath.size()) - 1 - i] ==
+    if (left.n != right.n) {
-                      rightPath[(rightPath.size()) - 1 - i];
+      break;
-       ++i) {
+    }
-    lca = leftPath[int(leftPath.size()) - 1 - i];
   }
 
-#if DEBUG_VERBOSE && !defined(NDEBUG)
+  if (!leftDone && !rightDone) {
-  fprintf(stderr, "firstGeq for `%s' got `%s'\n", printable(begin).c_str(),
+    assert(left.n->parent == right.n->parent);
-          getSearchPathPrintable(left.n).c_str());
+    for (int c = left.n->parentsIndex; c < right.n->parentsIndex;
-  fprintf(stderr, "firstGeq for `%s' got `%s'\n", printable(end).c_str(),
+         c = getChildGeq(left.n->parent, c + 1)) {
-          getSearchPathPrintable(right.n).c_str());
+      assert(c >= 0);
-  fprintf(stderr, "lca `%s'\n", getSearchPathPrintable(lca).c_str());
+      if (getChildExists(left.n->parent, c)->maxVersion > readVersion) {
-#endif
+        return false;
-  if (left.n != nullptr && left.cmp != 0 &&
+      }
-      left.n->entry.rangeVersion > 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 (left.n == right.n) {
+
+  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) {
+    // This would mean that left is overtaking right
+    __builtin_unreachable();
+  }
+
+  {
+    assert(left.n == right.n);
     return true;
   }
-  assert(left.n != nullptr);
-  auto boundaryVersion = left.n->entry.pointVersion;
-  if (left.cmp != 0) {
-    boundaryVersion = std::max(boundaryVersion, left.n->entry.rangeVersion);
-  }
-  if (right.n != nullptr) {
-    boundaryVersion = std::max(boundaryVersion, right.n->entry.rangeVersion);
-  }
-  if (boundaryVersion > readVersion) {
-    return false;
-  }
-
-  if (left.n != lca) {
-    while (left.n->parent != lca) {
-      for (int c = getChildGeq(left.n->parent, int(left.n->parentsIndex) + 1);
-           c >= 0; c = getChildGeq(left.n->parent, c + 1)) {
-        if (getChildExists(left.n->parent, c)->maxVersion > readVersion) {
-          return false;
-        }
-      }
-      left.n = left.n->parent;
-    }
-  }
-  if (right.n != nullptr && right.n != lca) {
-    while (right.n->parent != lca) {
-      for (int c = getChildLeq(right.n->parent, int(right.n->parentsIndex) - 1);
-           c >= 0; c = getChildLeq(right.n->parent, c - 1)) {
-        if (getChildExists(right.n->parent, c)->maxVersion > readVersion) {
-          return false;
-        }
-      }
-      right.n = right.n->parent;
-    }
-  }
-  for (int c = left.n != lca ? getChildGeq(lca, int(left.n->parentsIndex) + 1)
-                             : getChildGeq(lca, 0);
-       c >= 0 && (right.n == nullptr || c < right.n->parentsIndex);
-       c = getChildGeq(lca, c + 1)) {
-    if (getChildExists(lca, c)->maxVersion > readVersion) {
-      return false;
-    }
-  }
-  return true;
 }
 
 // Returns a pointer to the newly inserted node.  caller is reponsible for