Interleave checks for point reads

This doesn't actually seem faster, but it should prepare us to implement range reads non-naively. It probably should be faster. To be investigated.
2024-02-05 16:44:57 -08:00
parent 57ec97f2ee
commit d78afe0823
1 changed files with 166 additions and 94 deletions
--- a/ConflictSet.cpp
+++ b/ConflictSet.cpp
@@ -656,80 +656,109 @@ std::string_view getSearchPath(Arena &arena, Node *n) {
  }
 }

-Iterator lastLeq(Node *n, const std::span<const uint8_t> key) {
-  auto remaining = key;
-  for (;;) {
-    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;
+struct StepwiseLastLeq {
+  StepwiseLastLeq() {}
+
+  Node *n;
+  std::span<const uint8_t> remaining;
+  StepwiseLastLeq(Node *n, const std::span<const uint8_t> key)
+      : n(n), remaining(key) {}
+
+  int cmp;
+
+  enum Phase { Search, ScanBackward, DownRightSpine };
+  Phase 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) {
+            n = prevPhysical(n);
+            phase = ScanBackward;
+            return false;
+          } else {
+            phase = DownRightSpine;
+            return false;
+          }
        }
-        if (c > 0) {
+        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
          n = prevPhysical(n);
-          goto scanBackward;
-        } else {
-          goto downRightSpine;
+          phase = ScanBackward;
+          return false;
        }
      }
-      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
-        n = prevPhysical(n);
-        goto scanBackward;
-      }
-    }
-    {
-      Arena arena;
-      assert((std::string(getSearchPath(arena, n)) +
-              std::string((const char *)remaining.data(), remaining.size()))
-                 .ends_with(std::string((const char *)key.data(), key.size())));
-    }
-    if (remaining.size() == 0) {
-      // We've found the physical node corresponding to search path `key`
-      if (n->entryPresent) {
-        return {n, 0};
+      if (remaining.size() == 0) {
+        // We've found the physical node corresponding to search path `key`
+        if (n->entryPresent) {
+          cmp = 0;
+          return true;
+        } else {
+          phase = ScanBackward;
+          return false;
+        }
      } else {
-        goto scanBackward;
+        int c = getChildLeq(n, remaining[0]);
+        if (c == remaining[0]) {
+          n = getChildExists(n, c);
+          remaining = remaining.subspan(1, remaining.size() - 1);
+        } else {
+          if (c >= 0) {
+            n = getChildExists(n, c);
+            phase = DownRightSpine;
+            return false;
+          } else {
+            phase = ScanBackward;
+            return false;
+          }
+        }
      }
-    } else {
-      int c = getChildLeq(n, remaining[0]);
-      if (c == remaining[0]) {
-        n = getChildExists(n, c);
-        remaining = remaining.subspan(1, remaining.size() - 1);
-      } else {
+      return false;
+    case DownRightSpine:
+      // The physical node corresponding to search path `key` does not
+      // exist. Let's find the physical node corresponding to the highest
+      // search key (not necessarily present) less than key.
+      // Move down the right spine
+      {
+        int c = getChildLeq(n, 255);
        if (c >= 0) {
          n = getChildExists(n, c);
-          goto downRightSpine;
        } else {
-          goto scanBackward;
+          phase = ScanBackward;
        }
+        return false;
      }
+    case ScanBackward:
+      // Iterate backwards along existing physical nodes until we find a present
+      // entry
+      if (!n->entryPresent) {
+        n = prevPhysical(n);
+        return false;
+      }
+      cmp = -1;
+      return true;
    }
+    __builtin_unreachable(); // GCOVR_EXCL_LINE
  }
-downRightSpine:
-  // The physical node corresponding to search path `key` does not
-  // exist. Let's find the physical node corresponding to the highest
-  // search key (not necessarily present) less than key.
-  // Move down the right spine
-  for (;;) {
-    int c = getChildLeq(n, 255);
-    if (c >= 0) {
-      n = getChildExists(n, c);
-    } else {
-      goto scanBackward;
-    }
-  }
-scanBackward:
-  // Iterate backwards along existing physical nodes until we find a present
-  // entry
-  for (; !n->entryPresent; n = prevPhysical(n)) {
-  }
-  return {n, -1};
+};
+
+Iterator lastLeq(Node *n, const std::span<const uint8_t> key) {
+  StepwiseLastLeq l{n, key};
+  while (!l.step())
+    ;
+  return {l.n, l.cmp};
 }

 // Returns a pointer to the newly inserted node.  caller is reponsible for
@@ -807,51 +836,94 @@ void destroyTree(Node *root) {
  }
 }

-struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
-  void check(const ReadRange *reads, Result *result, int count) const {
-    for (int i = 0; i < count; ++i) {
-      const auto &r = reads[i];
-      if (r.readVersion < oldestVersion) {
-        result[i] = TooOld;
-        continue;
+struct CheckStepWise {
+  ConflictSet::Result *result;
+  const ConflictSet::ReadRange *read;
+  StepwiseLastLeq left;
+  StepwiseLastLeq right;
+
+  CheckStepWise() {}
+
+  CheckStepWise(Node *root, ConflictSet::Result *result,
+                const ConflictSet::ReadRange *r)
+      : result(result), read(r),
+        left(root, std::span<const uint8_t>(r->begin.p, r->begin.len)),
+        right(root, std::span<const uint8_t>(r->end.p, r->end.len)),
+        phase(r->end.len == 0 ? PointRead : RangeRead) {}
+
+  enum Phase {
+    PointRead,
+    RangeRead,
+  };
+
+  Phase phase;
+
+  bool step() {
+    switch (phase) {
+    case PointRead:
+      if (left.step()) {
+        auto *l = left.n;
+        int c = left.cmp;
+        assert(l != nullptr);
+        assert(l->entryPresent);
+        *result = (c == 0 ? l->entry.pointVersion : l->entry.rangeVersion) >
+                          read->readVersion
+                      ? ConflictSet::Conflict
+                      : ConflictSet::Commit;
+        return true;
      }
-      auto [l, c] =
-          lastLeq(root, std::span<const uint8_t>(r.begin.p, r.begin.len));
-#if DEBUG_VERBOSE && !defined(NDEBUG)
-      Arena arena;
-      fprintf(stderr, "LastLeq for `%s' got `%s'\n", printable(r.begin).c_str(),
-              printable(getSearchPath(arena, l)).c_str());
-#endif
-      assert(l != nullptr);
-      assert(l->entryPresent);
-      result[i] = (c == 0 ? l->entry.pointVersion : l->entry.rangeVersion) >
-                          r.readVersion
-                      ? Conflict
-                      : Commit;
-      if (result[i] == Commit && r.end.len > 0) {
-        auto [e, c] =
-            lastLeq(root, std::span<const uint8_t>(r.end.p, r.end.len));
-#if DEBUG_VERBOSE && !defined(NDEBUG)
-        Arena arena;
-        fprintf(stderr, "LastLeq for `%s' got `%s'\n", printable(r.end).c_str(),
-                printable(getSearchPath(arena, e)).c_str());
-#endif
+      return false;
+    case RangeRead: {
+      while (!left.step())
+        ;
+      while (!right.step())
+        ;
+      auto *l = left.n;
+      auto c = left.cmp;
+      *result = (c == 0 ? l->entry.pointVersion : l->entry.rangeVersion) >
+                        read->readVersion
+                    ? ConflictSet::Conflict
+                    : ConflictSet::Commit;
+      if (*result == ConflictSet::Commit) {
+        auto *e = right.n;
+        auto c = right.cmp;
        if (l == e) {
-          continue;
+          return true;
        }
        if (c != 0) {
          e = nextLogical(e);
        }
        for (auto iter = nextLogical(l); iter != e; iter = nextLogical(iter)) {
-          if (iter->entry.pointVersion > r.readVersion ||
-              iter->entry.rangeVersion > r.readVersion) {
-            result[i] = Conflict;
-            break;
+          if (iter->entry.pointVersion > read->readVersion ||
+              iter->entry.rangeVersion > read->readVersion) {
+            *result = ConflictSet::Conflict;
+            return true;
          }
        }
      }
+      return true;
    }
+    }
+    __builtin_unreachable(); // GCOVR_EXCL_LINE
  }
+};
+
+struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
+
+  void check(const ReadRange *reads, Result *result, int count) const {
+    Arena arena;
+    CheckStepWise *checks = new (arena) CheckStepWise[count];
+    int index = 0;
+    for (int i = 0; i < count; ++i) {
+      if (reads[i].readVersion < oldestVersion) {
+        result[i] = TooOld;
+      } else {
+        checks[index++] = CheckStepWise(root, result + i, reads + i);
+      }
+    }
+    runInterleaved(std::span<CheckStepWise>(checks, index));
+  }
+
  void addWrites(const WriteRange *writes, int count) {
    for (int i = 0; i < count; ++i) {
      const auto &w = writes[i];