From ae5277bf78b606f07d3a6ab050fadb0c9d8d1de7 Mon Sep 17 00:00:00 2001
From: Andrew Noyes <andrew@weaselab.dev>
Date: Tue, 20 Feb 2024 11:28:58 -0800
Subject: [PATCH] Interleave checking left and right side

---
 ConflictSet.cpp | 112 ++++++++++++++++++++++++++----------------------
 1 file changed, 61 insertions(+), 51 deletions(-)

diff --git a/ConflictSet.cpp b/ConflictSet.cpp
index dbf4a9b..f78f20a 100644
--- a/ConflictSet.cpp
+++ b/ConflictSet.cpp
@@ -999,8 +999,7 @@ struct CheckRangeLeftSide {
         if (c >= 0) {
           if (searchPathLen < prefixLen) {
             n = getChildExists(n, c);
-            phase = DownLeftSpine;
-            return false;
+            return downLeftSpine();
           }
           if (maxBetweenExclusive(n, c, 256) > readVersion) {
             ok = false;
@@ -1011,8 +1010,7 @@ struct CheckRangeLeftSide {
           return true;
         } else {
           n = nextSibling(n);
-          phase = DownLeftSpine;
-          return false;
+          return downLeftSpine();
         }
       }
 
@@ -1026,8 +1024,7 @@ struct CheckRangeLeftSide {
           }
           if (c > 0) {
             if (searchPathLen < prefixLen) {
-              phase = DownLeftSpine;
-              return false;
+              return downLeftSpine();
             }
             if (n->entryPresent && n->entry.rangeVersion > readVersion) {
               ok = false;
@@ -1037,8 +1034,7 @@ struct CheckRangeLeftSide {
             return true;
           } else {
             n = nextSibling(n);
-            phase = DownLeftSpine;
-            return false;
+            return downLeftSpine();
           }
         }
         if (commonLen == n->partialKeyLen) {
@@ -1047,8 +1043,7 @@ struct CheckRangeLeftSide {
               remaining.subspan(commonLen, remaining.size() - commonLen);
         } else if (n->partialKeyLen > int(remaining.size())) {
           if (searchPathLen < prefixLen) {
-            phase = DownLeftSpine;
-            return false;
+            return downLeftSpine();
           }
           if (n->entryPresent && n->entry.rangeVersion > readVersion) {
             ok = false;
@@ -1061,23 +1056,26 @@ struct CheckRangeLeftSide {
       break;
     }
     case DownLeftSpine:
-      if (n == nullptr) {
-        ok = true;
+      if (n->entryPresent) {
+        ok = n->entry.rangeVersion <= readVersion;
         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);
-      }
+      int c = getChildGeq(n, 0);
+      assert(c >= 0);
+      n = getChildExists(n, c);
       break;
     }
     return false;
   }
+
+  bool downLeftSpine() {
+    phase = DownLeftSpine;
+    if (n == nullptr) {
+      ok = true;
+      return true;
+    }
+    return false;
+  }
 };
 
 // Return true if the max version among all keys that start with key[:prefixLen]
@@ -1139,8 +1137,7 @@ struct CheckRangeRightSide {
         if (c >= 0) {
           if (searchPathLen < prefixLen) {
             n = getChildExists(n, c);
-            phase = DownLeftSpine;
-            return false;
+            return downLeftSpine();
           }
           if (maxBetweenExclusive(n, -1, c) > readVersion) {
             ok = false;
@@ -1150,8 +1147,7 @@ struct CheckRangeRightSide {
           return true;
         } else {
           n = nextSibling(n);
-          phase = DownLeftSpine;
-          return false;
+          return downLeftSpine();
         }
       }
 
@@ -1165,8 +1161,7 @@ struct CheckRangeRightSide {
           }
           if (c > 0) {
             if (searchPathLen < prefixLen) {
-              phase = DownLeftSpine;
-              return false;
+              return downLeftSpine();
             }
             if (n->entryPresent && n->entry.rangeVersion > readVersion) {
               ok = false;
@@ -1180,8 +1175,7 @@ struct CheckRangeRightSide {
               return true;
             }
             n = nextSibling(n);
-            phase = DownLeftSpine;
-            return false;
+            return downLeftSpine();
           }
         }
         if (commonLen == n->partialKeyLen) {
@@ -1190,8 +1184,7 @@ struct CheckRangeRightSide {
               remaining.subspan(commonLen, remaining.size() - commonLen);
         } else if (n->partialKeyLen > int(remaining.size())) {
           if (searchPathLen < prefixLen) {
-            phase = DownLeftSpine;
-            return false;
+            return downLeftSpine();
           }
           if (n->entryPresent && n->entry.rangeVersion > readVersion) {
             ok = false;
@@ -1203,23 +1196,25 @@ struct CheckRangeRightSide {
       }
     } break;
     case DownLeftSpine:
-      if (n == nullptr) {
-        ok = true;
+      if (n->entryPresent) {
+        ok = n->entry.rangeVersion <= readVersion;
         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);
-      }
+      int c = getChildGeq(n, 0);
+      assert(c >= 0);
+      n = getChildExists(n, c);
       break;
     }
     return false;
   }
+  bool downLeftSpine() {
+    phase = DownLeftSpine;
+    if (n == nullptr) {
+      ok = true;
+      return true;
+    }
+    return false;
+  }
 };
 
 bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
@@ -1265,22 +1260,37 @@ bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
     return true;
   }
 
-  CheckRangeLeftSide checkRangeLeftSide{n, begin, lcp + 1, readVersion};
-  while (!checkRangeLeftSide.step())
-    ;
-  if (!checkRangeLeftSide.ok) {
-    return false;
-  }
-
   if (!checkRangeStartsWith(n, begin.subspan(0, lcp), begin[lcp], end[lcp],
                             readVersion)) {
     return false;
   }
 
+  CheckRangeLeftSide checkRangeLeftSide{n, begin, lcp + 1, readVersion};
   CheckRangeRightSide checkRangeRightSide{n, end, lcp + 1, readVersion};
-  while (!checkRangeRightSide.step())
-    ;
 
+  for (;;) {
+    bool leftDone = checkRangeLeftSide.step();
+    bool rightDone = checkRangeRightSide.step();
+    if (!leftDone && !rightDone) {
+      continue;
+    }
+    if (leftDone && rightDone) {
+      break;
+    } else if (leftDone) {
+      while (!checkRangeRightSide.step())
+        ;
+      break;
+    } else {
+      assert(rightDone);
+      while (!checkRangeLeftSide.step())
+        ;
+    }
+    break;
+  }
+
+  if (!checkRangeLeftSide.ok) {
+    return false;
+  }
   if (!checkRangeRightSide.ok) {
     return false;
   }