Sublinear read range checking

Still slow though

ConflictSet.cpp

@@ -792,6 +792,88 @@ 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 right = firstGeq(n, end);
+
+  Arena arena;
+  auto leftPath = vector<Node *>(arena);
+  auto rightPath = vector<Node *>(arena);
+  for (auto *iter = left.n; iter != nullptr; iter = iter->parent) {
+    leftPath.push_back(iter);
+  }
+  for (auto *iter = right.n; iter != nullptr; iter = iter->parent) {
+    rightPath.push_back(iter);
+  }
+  Node *lca = n;
+  for (int i = 0; int(leftPath.size()) - 1 - i >= 0 &&
+                  int(rightPath.size()) - 1 - i >= 0 &&
+                  leftPath[int(leftPath.size()) - 1 - i] ==
+                      rightPath[(rightPath.size()) - 1 - i];
+       ++i) {
+    lca = leftPath[int(leftPath.size()) - 1 - i];
+  }
+
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+  fprintf(stderr, "firstGeq for `%s' got `%s'\n", printable(begin).c_str(),
+          getSearchPath(left.n).c_str());
+  fprintf(stderr, "firstGeq for `%s' got `%s'\n", printable(end).c_str(),
+          getSearchPath(right.n).c_str());
+  fprintf(stderr, "lca `%s'\n", getSearchPath(lca).c_str());
+#endif
+  if (left.n != nullptr && left.cmp != 0 &&
+      left.n->entry.rangeVersion > readVersion) {
+    return false;
+  }
+  if (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
 // setting 'entry' fields on the result, which may have !entryPresent. The
 // search path for `key` will have maxVersion at least `writeVersion` as a
@@ -871,64 +953,23 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
 
   void check(const ReadRange *reads, Result *result, int count) const {
     for (int i = 0; i < count; ++i) {
-      if (reads[i].readVersion < oldestVersion) {
-        result[i] = TooOld;
-        continue;
-      }
-      if (reads[i].end.len > 0) {
-        result[i] = Commit;
-        auto left = firstGeq(root, std::span<const uint8_t>(
-                                       reads[i].begin.p, reads[i].begin.len));
-        auto right = firstGeq(
-            root, std::span<const uint8_t>(reads[i].end.p, reads[i].end.len));
-#if DEBUG_VERBOSE && !defined(NDEBUG)
-        fprintf(stderr, "firstGeq for `%s' got `%s'\n",
-                printable(reads[i].begin).c_str(),
-                getSearchPath(left.n).c_str());
-        fprintf(stderr, "firstGeq for `%s' got `%s'\n",
-                printable(reads[i].end).c_str(),
-                getSearchPath(right.n).c_str());
-#endif
-        if (left.n != nullptr && left.cmp != 0 &&
-            left.n->entry.rangeVersion > reads[i].readVersion) {
-          result[i] = Conflict;
-          continue;
-        }
-        if (left.n == right.n) {
-          continue;
-        }
-        assert(left.n != nullptr);
-        auto boundaryVersion = left.n->entry.pointVersion;
-        if (left.cmp != 0) {
-          boundaryVersion =
-              std::max(boundaryVersion, left.n->entry.rangeVersion);
-        }
-        if (right.cmp == 0) {
-          boundaryVersion =
-              std::max(boundaryVersion, right.n->entry.rangeVersion);
-        }
-        if (boundaryVersion > reads[i].readVersion) {
-          result[i] = Conflict;
-          continue;
-        }
-        for (auto *iter = nextLogical(left.n); iter != right.n;
-             iter = nextLogical(iter)) {
-          if (std::max(iter->entry.pointVersion, iter->entry.rangeVersion) >
-              reads[i].readVersion) {
-            result[i] = Conflict;
-            break;
-          }
-        }
-      } else {
-        result[i] = checkPointRead(root,
+      result[i] =
+          reads[i].readVersion < oldestVersion ? TooOld
+          : (reads[i].end.len > 0
+                 ? checkRangeRead(root,
                                   std::span<const uint8_t>(reads[i].begin.p,
                                                            reads[i].begin.len),
+                                  std::span<const uint8_t>(reads[i].end.p,
+                                                           reads[i].end.len),
                                   reads[i].readVersion)
+                 : checkPointRead(root,
+                                  std::span<const uint8_t>(reads[i].begin.p,
+                                                           reads[i].begin.len),
+                                  reads[i].readVersion))
               ? Commit
               : Conflict;
     }
   }
-  }
 
   void addWrites(const WriteRange *writes, int count) {
     for (int i = 0; i < count; ++i) {

Internal.h

@@ -426,6 +426,10 @@ inline std::string printable(const Key &key) {
   return printable(std::string_view((const char *)key.p, key.len));
 }
 
+inline std::string printable(std::span<const uint8_t> key) {
+  return printable(std::string_view((const char *)key.data(), key.size()));
+}
+
 namespace {
 
 template <class ConflictSetImpl> struct TestDriver {