CheckRangeRightSide struct

ConflictSet.cpp

@@ -1082,106 +1082,145 @@ struct CheckRangeLeftSide {
 
 // Return true if the max version among all keys that start with key[:prefixLen]
 // that are < key is <= readVersion
-bool checkRangeRightSide(Node *n, std::span<const uint8_t> key, int prefixLen,
+struct CheckRangeRightSide {
-                         int64_t readVersion) {
+  CheckRangeRightSide(Node *n, std::span<const uint8_t> key, int prefixLen,
+                      int64_t readVersion)
+      : n(n), key(key), remaining(key), prefixLen(prefixLen),
+        readVersion(readVersion) {
 #if DEBUG_VERBOSE && !defined(NDEBUG)
-  fprintf(stderr, "Check range right side to %s for keys starting with %s\n",
+    fprintf(stderr, "Check range right side to %s for keys starting with %s\n",
-          printable(key).c_str(), printable(key.subspan(0, prefixLen)).c_str());
+            printable(key).c_str(),
+            printable(key.subspan(0, prefixLen)).c_str());
 #endif
-  auto remaining = key;
+  }
+
+  Node *n;
+  std::span<const uint8_t> key;
+  std::span<const uint8_t> remaining;
+  int prefixLen;
+  int64_t readVersion;
   int searchPathLen = 0;
-  for (;;) {
+  bool ok;
-    if (n->maxVersion <= readVersion) {
-      return true;
-    }
-    if (remaining.size() == 0) {
-      assert(searchPathLen >= prefixLen);
-      return true;
-    }
 
-    if (searchPathLen >= prefixLen) {
+  enum Phase { Search, DownLeftSpine } phase = Search;
-      if (maxBetweenExclusive(n, -1, remaining[0]) > readVersion) {
-        return false;
-      }
-    }
 
-    if (searchPathLen >= prefixLen && searchPathLen < int(key.size()) &&
+  bool step() {
-        n->entryPresent && n->entry.pointVersion > readVersion) {
+    switch (phase) {
-      return false;
+    case Search: {
-    }
+      if (n->maxVersion <= readVersion) {
-
+        ok = true;
-    int c = getChildGeq(n, remaining[0]);
+        return true;
-    if (c == remaining[0]) {
+      }
-      n = getChildExists(n, c);
+      if (remaining.size() == 0) {
-      remaining = remaining.subspan(1, remaining.size() - 1);
+        assert(searchPathLen >= prefixLen);
-      ++searchPathLen;
+        ok = true;
-    } else {
-      if (c >= 0) {
-        if (searchPathLen < prefixLen) {
-          n = getChildExists(n, c);
-          goto downLeftSpine;
-        }
-        if (maxBetweenExclusive(n, -1, c) > readVersion) {
-          return false;
-        }
         return true;
-      } else {
-        n = nextSibling(n);
-        goto downLeftSpine;
       }
-    }
 
-    if (n->partialKeyLen > 0) {
+      if (searchPathLen >= prefixLen) {
-      int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
+        if (maxBetweenExclusive(n, -1, remaining[0]) > readVersion) {
-      for (int i = 0; i < commonLen; ++i) {
+          ok = false;
-        auto c = n->partialKey[i] <=> remaining[i];
+          return true;
-        if (c == 0) {
-          ++searchPathLen;
-          continue;
         }
-        if (c > 0) {
+      }
+
+      if (searchPathLen >= prefixLen && searchPathLen < int(key.size()) &&
+          n->entryPresent && n->entry.pointVersion > readVersion) {
+        ok = false;
+        return true;
+      }
+
+      int c = getChildGeq(n, remaining[0]);
+      if (c == remaining[0]) {
+        n = getChildExists(n, c);
+        remaining = remaining.subspan(1, remaining.size() - 1);
+        ++searchPathLen;
+      } else {
+        if (c >= 0) {
           if (searchPathLen < prefixLen) {
-            goto downLeftSpine;
+            n = getChildExists(n, c);
-          }
+            phase = DownLeftSpine;
-          if (n->entryPresent && n->entry.rangeVersion > readVersion) {
             return false;
           }
+          if (maxBetweenExclusive(n, -1, c) > readVersion) {
+            ok = false;
+            return true;
+          }
+          ok = true;
           return true;
         } else {
-          if (searchPathLen >= prefixLen && n->maxVersion > readVersion) {
-            return false;
-          }
           n = nextSibling(n);
-          goto downLeftSpine;
+          phase = DownLeftSpine;
-        }
-      }
-      if (commonLen == n->partialKeyLen) {
-        // partial key matches
-        remaining = remaining.subspan(commonLen, remaining.size() - commonLen);
-      } else if (n->partialKeyLen > int(remaining.size())) {
-        if (searchPathLen < prefixLen) {
-          goto downLeftSpine;
-        }
-        if (n->entryPresent && n->entry.rangeVersion > readVersion) {
           return false;
         }
+      }
+
+      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) {
+            ++searchPathLen;
+            continue;
+          }
+          if (c > 0) {
+            if (searchPathLen < prefixLen) {
+              phase = DownLeftSpine;
+              return false;
+            }
+            if (n->entryPresent && n->entry.rangeVersion > readVersion) {
+              ok = false;
+              return true;
+            }
+            ok = true;
+            return true;
+          } else {
+            if (searchPathLen >= prefixLen && n->maxVersion > readVersion) {
+              ok = false;
+              return true;
+            }
+            n = nextSibling(n);
+            phase = DownLeftSpine;
+            return false;
+          }
+        }
+        if (commonLen == n->partialKeyLen) {
+          // partial key matches
+          remaining =
+              remaining.subspan(commonLen, remaining.size() - commonLen);
+        } else if (n->partialKeyLen > int(remaining.size())) {
+          if (searchPathLen < prefixLen) {
+            phase = DownLeftSpine;
+            return false;
+          }
+          if (n->entryPresent && n->entry.rangeVersion > readVersion) {
+            ok = false;
+            return true;
+          }
+          ok = true;
+          return true;
+        }
+      }
+    } break;
+    case DownLeftSpine:
+      if (n == nullptr) {
+        ok = true;
         return true;
       }
+      for (;;) {
+        if (n->entryPresent) {
+          ok = n->entry.rangeVersion <= readVersion;
+          return true;
+        }
+        int c = getChildGeq(n, 0);
+        assert(c >= 0);
+        n = getChildExists(n, c);
+      }
+      break;
     }
+    return false;
   }
-downLeftSpine:
+};
-  if (n == nullptr) {
-    return true;
-  }
-  for (;;) {
-    if (n->entryPresent) {
-      return n->entry.rangeVersion <= readVersion;
-    }
-    int c = getChildGeq(n, 0);
-    assert(c >= 0);
-    n = getChildExists(n, c);
-  }
-}
 
 bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
                     std::span<const uint8_t> end, int64_t readVersion) {
@@ -1238,7 +1277,11 @@ bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
     return false;
   }
 
-  if (!checkRangeRightSide(n, end, lcp + 1, readVersion)) {
+  CheckRangeRightSide checkRangeRightSide{n, end, lcp + 1, readVersion};
+  while (!checkRangeRightSide.step())
+    ;
+
+  if (!checkRangeRightSide.ok) {
     return false;
   }