Implement phase 2

ConflictSet.cpp

@@ -264,7 +264,7 @@ struct Node {
   /* begin section that's copied to the next node */
   union {
     Entry entry;
-    /* Set to the forwarding point for this node if deferredReleased is set */
+    /* Set to the forwarding point for this node if releaseDeferred is set */
     Node *forwardTo;
   };
   Node *parent;
@@ -278,15 +278,12 @@ struct Node {
 
   /* If set, this node has been replaced and the next node in the forwarding
    * chain is `forwardTo`*/
-  bool deferredReleased;
+  bool releaseDeferred;
 
   uint8_t *partialKey();
-  Type getType() const {
+  Type getType() const { return type; }
-    assert(!deferredReleased);
-    return type;
-  }
   int32_t getCapacity() const {
-    assert(!deferredReleased);
+    assert(!releaseDeferred);
     return partialKeyCapacity;
   }
 
@@ -320,7 +317,7 @@ struct Node0 : Node {
   constexpr static auto kType = Type_Node0;
 
   uint8_t *partialKey() {
-    assert(!deferredReleased);
+    assert(!releaseDeferred);
     return (uint8_t *)(this + 1);
   }
   void copyChildrenAndKeyFrom(const Node0 &other);
@@ -338,7 +335,7 @@ struct Node3 : Node {
   uint8_t index[kMaxNodes];
 
   uint8_t *partialKey() {
-    assert(!deferredReleased);
+    assert(!releaseDeferred);
     return (uint8_t *)(this + 1);
   }
   void copyChildrenAndKeyFrom(const Node0 &other);
@@ -357,7 +354,7 @@ struct Node16 : Node {
   uint8_t index[kMaxNodes];
 
   uint8_t *partialKey() {
-    assert(!deferredReleased);
+    assert(!releaseDeferred);
     return (uint8_t *)(this + 1);
   }
   void copyChildrenAndKeyFrom(const Node3 &other);
@@ -382,7 +379,7 @@ struct Node48 : Node {
   int8_t index[256];
 
   uint8_t *partialKey() {
-    assert(!deferredReleased);
+    assert(!releaseDeferred);
     return (uint8_t *)(this + 1);
   }
   void copyChildrenAndKeyFrom(const Node16 &other);
@@ -405,7 +402,7 @@ struct Node256 : Node {
   InternalVersionT maxOfMax[kMaxOfMaxTotalPages];
 
   uint8_t *partialKey() {
-    assert(!deferredReleased);
+    assert(!releaseDeferred);
     return (uint8_t *)(this + 1);
   }
   void copyChildrenAndKeyFrom(const Node48 &other);
@@ -708,7 +705,7 @@ template <class T> struct BoundedFreeListAllocator {
   T *allocate(int partialKeyCapacity) {
     T *result = allocate_helper(partialKeyCapacity);
     result->endOfRange = false;
-    result->deferredReleased = false;
+    result->releaseDeferred = false;
     if constexpr (!std::is_same_v<T, Node0>) {
       memset(result->children, 0, sizeof(result->children));
       const auto z = InternalVersionT::zero;
@@ -852,10 +849,13 @@ struct WriteContext {
   }
 
   // Place in a list to be released in the next call to releaseDeferred.
-  void deferRelease(Node *n) {
+  void deferRelease(Node *n, Node *forwardTo) {
+    n->releaseDeferred = true;
+    n->forwardTo = forwardTo;
     n->parent = deferredList;
     deferredList = n;
   }
+
   // Release all nodes passed to deferRelease since the last call to
   // releaseDeferred.
   void releaseDeferred() {
@@ -1672,7 +1672,7 @@ TaggedNodePointer &getOrCreateChild(TaggedNodePointer &self,
 
     auto *newSelf = writeContext->allocate<Node3>(self->partialKeyLen);
     newSelf->copyChildrenAndKeyFrom(*self0);
-    writeContext->release(self0);
+    writeContext->deferRelease(self0, newSelf);
     self = newSelf;
 
     goto insert3;
@@ -1682,7 +1682,7 @@ TaggedNodePointer &getOrCreateChild(TaggedNodePointer &self,
       auto *self3 = static_cast<Node3 *>(self);
       auto *newSelf = writeContext->allocate<Node16>(self->partialKeyLen);
       newSelf->copyChildrenAndKeyFrom(*self3);
-      writeContext->release(self3);
+      writeContext->deferRelease(self3, newSelf);
       self = newSelf;
       goto insert16;
     }
@@ -1711,7 +1711,7 @@ TaggedNodePointer &getOrCreateChild(TaggedNodePointer &self,
       auto *self16 = static_cast<Node16 *>(self);
       auto *newSelf = writeContext->allocate<Node48>(self->partialKeyLen);
       newSelf->copyChildrenAndKeyFrom(*self16);
-      writeContext->release(self16);
+      writeContext->deferRelease(self16, newSelf);
       self = newSelf;
       goto insert48;
     }
@@ -1742,7 +1742,7 @@ TaggedNodePointer &getOrCreateChild(TaggedNodePointer &self,
       auto *self48 = static_cast<Node48 *>(self);
       auto *newSelf = writeContext->allocate<Node256>(self->partialKeyLen);
       newSelf->copyChildrenAndKeyFrom(*self48);
-      writeContext->release(self48);
+      writeContext->deferRelease(self48, newSelf);
       self = newSelf;
       goto insert256;
     }
@@ -4738,6 +4738,16 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
     // Phase 2: Perform insertions. Nodes may be upsized during this phase, but
     // old nodes get forwarding pointers installed and are released after
     // phase 2.
+
+    for (int i = 0; i < count; ++i) {
+      while (context.results[i].insertionPoint->releaseDeferred) {
+        context.results[i].insertionPoint =
+            context.results[i].insertionPoint->forwardTo;
+      }
+      addPointWrite(getInTree(context.results[i].insertionPoint, this),
+                    context.results[i].remaining, writeVersion, &writeContext);
+    }
+    safe_free(context.results, sizeof(interleaved_insert::Result) * count);
   }
 
   void addWrites(const WriteRange *writes, int count, int64_t writeVersion) {
@@ -4788,7 +4798,7 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
         break;
       }
     }
-    if (0 && allPointWrites) {
+    if (allPointWrites) {
       interleavedPointWrites(writes, count, InternalVersionT(writeVersion));
     } else {
       for (int i = 0; i < count; ++i) {
@@ -4806,6 +4816,8 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
       }
     }
 
+    writeContext.releaseDeferred();
+
     // Run gc at least 200% the rate we're inserting entries
     keyUpdates += std::max<int64_t>(writeContext.accum.entries_inserted -
                                         writeContext.accum.entries_erased,