Worklist algorithm for rotations

ConflictSet.cpp

@@ -305,10 +305,13 @@ void updateMaxVersion(Node *n) {
           std::max(maxVersion, n->child[i] != nullptr ? n->child[i]->maxVersion
                                                       : maxVersion);
     }
-    if (n->maxVersion == maxVersion || n->parent == nullptr) {
+    if (n->maxVersion == maxVersion) {
       break;
     }
     n->maxVersion = maxVersion;
+    if (n->parent == nullptr) {
+      break;
+    }
     n = n->parent;
   }
 }
@@ -346,6 +349,21 @@ void rotate(Node **node, bool dir) {
   updateMaxVersion(l);
 }
 
+int64_t checkMaxVersion(Node *node, bool &success) {
+  int64_t expected = std::max(node->pointVersion, node->rangeVersion);
+  for (int i = 0; i < 2; ++i) {
+    if (node->child[i] != nullptr) {
+      expected = std::max(expected, checkMaxVersion(node->child[i], success));
+    }
+  }
+  if (node->maxVersion != expected) {
+    fprintf(stderr, "%.*s has max version %d. Expected %d\n", node->len,
+            (const char *)(node + 1), int(node->maxVersion), int(expected));
+  }
+  success = false;
+  return expected;
+}
+
 bool checkInvariants(Node *node) {
   bool success = true;
   // Check bst invariant
@@ -366,7 +384,10 @@ bool checkInvariants(Node *node) {
   }
   assert(std::is_sorted(keys.begin(), keys.end()));
 
-  // TODO more invariants
+  checkMaxVersion(node, success);
+
+  // TODO Compare logical contents of map with
+  // reference implementation
   return success;
 }
 
@@ -478,47 +499,6 @@ struct ConflictSet::Impl {
     }
   };
 
-  struct StepwiseRotate {
-    // Rotation phase state. The heap invariant may be violated for n->parent.
-    // Once this phase is complete the heap invariant is restored for each
-    // n->parent encountered in a step of this phase.
-    Node *n;
-    Impl *impl;
-
-    StepwiseRotate() {}
-    StepwiseRotate(Node *n, Impl *impl) : n(n), impl(impl) {}
-
-    bool step() {
-#if DEBUG
-      fprintf(stderr, "Step rotate %.*s\n", n->len, (const char *)(n + 1));
-#endif
-      if (n->parent == nullptr) {
-        return true;
-      }
-      const bool dir = n == n->parent->child[1];
-      assert(dir || n == n->parent->child[0]);
-      // p is the address of the pointer to n->parent in the tree
-      Node **p = n->parent->parent == nullptr
-                     ? &impl->root
-                     : &n->parent->parent
-                            ->child[n->parent->parent->child[1] == n->parent];
-      assert(*p == n->parent);
-      if (n->parent->priority < n->priority) {
-#if DEBUG
-        fprintf(stderr, "\trotate %s\n", !dir ? "right" : "left");
-#endif
-        rotate(p, !dir);
-        // assert((*p)->child[0] == nullptr || (*p)->priority >=
-        // (*p)->child[0]->priority); assert((*p)->child[1] == nullptr ||
-        // (*p)->priority >= (*p)->child[1]->priority);
-        n = *p;
-      } else {
-        return true;
-      }
-      return false;
-    }
-  };
-
   void addWrites(const WriteRange *writes, int count) {
     auto stepwiseInserts =
         std::unique_ptr<StepwiseInsert[]>{new StepwiseInsert[count]};
@@ -534,12 +514,42 @@ struct ConflictSet::Impl {
 
     runInterleaved(std::span<StepwiseInsert>(stepwiseInserts.get(), count));
 
-    auto stepwiseRotates =
+    std::vector<Node *> workList;
-        std::unique_ptr<StepwiseRotate[]>{new StepwiseRotate[count]};
+    workList.reserve(count);
     for (int i = 0; i < count; ++i) {
-      stepwiseRotates[i] = StepwiseRotate{*stepwiseInserts[i].current, this};
+      workList.push_back(*stepwiseInserts[i].current);
+    }
+
+    while (!workList.empty()) {
+      Node *n = workList.back();
+      workList.pop_back();
+      if (n->parent == nullptr) {
+        continue;
+      }
+      const bool dir = n == n->parent->child[1];
+      assert(dir || n == n->parent->child[0]);
+      // p is the address of the pointer to n->parent in the tree
+      Node **p = n->parent->parent == nullptr
+                     ? &root
+                     : &n->parent->parent
+                            ->child[n->parent->parent->child[1] == n->parent];
+      assert(*p == n->parent);
+      if (n->parent->priority < n->priority) {
+#if DEBUG
+        fprintf(stderr, "\trotate %.*s %s\n", n->len, (const char *)(n + 1),
+                !dir ? "right" : "left");
+#endif
+        rotate(p, !dir);
+        workList.push_back(*p);
+        assert((*p)->child[!dir] != nullptr);
+        auto *lr = (*p)->child[!dir]->child[dir];
+        if (lr != nullptr) {
+          workList.push_back(lr);
+        }
+      } else {
+        continue;
+      }
     }
-    runSequential(std::span<StepwiseRotate>(stepwiseRotates.get(), count), 7);
   }
 
   void setOldestVersion(int64_t oldestVersion) {