Change kMinChildrenNode4 to 2, fixing the induction

ConflictSet.cpp

@@ -259,19 +259,26 @@ struct Node256 : Node {
   uint8_t *partialKey() { return (uint8_t *)(this + 1); }
 };
 
+namespace {
+std::string getSearchPathPrintable(Node *n);
+}
+
 // Bound memory usage following the analysis in the ART paper
 
-constexpr int kBytesPerKey = 121;
-constexpr int kMinChildrenNode4 = 1;
+constexpr int kBytesPerKey = 120;
+constexpr int kMinChildrenNode4 = 2;
 constexpr int kMinChildrenNode16 = 5;
 constexpr int kMinChildrenNode48 = 17;
 constexpr int kMinChildrenNode256 = 49;
 
-static_assert(sizeof(Node256) < kMinChildrenNode256 * kBytesPerKey);
-static_assert(sizeof(Node48) < kMinChildrenNode48 * kBytesPerKey);
-static_assert(sizeof(Node16) < kMinChildrenNode16 * kBytesPerKey);
-static_assert(sizeof(Node4) < kMinChildrenNode4 * kBytesPerKey);
-static_assert(sizeof(Node0) < kBytesPerKey);
+static_assert(sizeof(Node256) + kBytesPerKey <=
+              kMinChildrenNode256 * kBytesPerKey);
+static_assert(sizeof(Node48) + kBytesPerKey <=
+              kMinChildrenNode48 * kBytesPerKey);
+static_assert(sizeof(Node16) + kBytesPerKey <=
+              kMinChildrenNode16 * kBytesPerKey);
+static_assert(sizeof(Node4) + kBytesPerKey <= kMinChildrenNode4 * kBytesPerKey);
+static_assert(sizeof(Node0) <= kBytesPerKey);
 
 // setOldestVersion will additionally try to maintain this property:
 //     `max(children, 1) * length >= capacity`
@@ -736,21 +743,14 @@ Node *nextLogical(Node *node) {
   return node;
 }
 
-// Fix larger-than-desired capacities. Does not return nodes to freelists,
-// since that wouldn't actually reclaim the memory used for partial key
-// capacity.
-void maybeDecreaseCapacity(Node *&self, NodeAllocators *allocators,
-                           ConflictSet::Impl *impl) {
-  const int maxCapacity =
-      std::max<int>(self->numChildren, 1) * self->partialKeyLen;
-  if (self->partialKeyCapacity <= maxCapacity) {
-    return;
-  }
-
+// Invalidates `self`, replacing it with a node of at least capacity.
+// Does not return nodes to freelists.
+void makeCapacityAtLeast(Node *&self, int capacity, NodeAllocators *allocators,
+                         ConflictSet::Impl *impl) {
   switch (self->type) {
   case Type::Node0: {
     auto *self0 = (Node0 *)self;
-    auto *newSelf = allocators->node0.allocate(maxCapacity);
+    auto *newSelf = allocators->node0.allocate(capacity);
     memcpy((char *)newSelf + kNodeCopyBegin, (char *)self + kNodeCopyBegin,
            kNodeCopySize);
     memcpy(newSelf->partialKey(), self0->partialKey(), self->partialKeyLen);
@@ -760,7 +760,7 @@ void maybeDecreaseCapacity(Node *&self, NodeAllocators *allocators,
   } break;
   case Type::Node4: {
     auto *self4 = (Node4 *)self;
-    auto *newSelf = allocators->node4.allocate(maxCapacity);
+    auto *newSelf = allocators->node4.allocate(capacity);
     memcpy((char *)newSelf + kNodeCopyBegin, (char *)self + kNodeCopyBegin,
            kNodeCopySize);
     memcpy(newSelf->partialKey(), self4->partialKey(), self->partialKeyLen);
@@ -776,7 +776,7 @@ void maybeDecreaseCapacity(Node *&self, NodeAllocators *allocators,
   } break;
   case Type::Node16: {
     auto *self16 = (Node16 *)self;
-    auto *newSelf = allocators->node16.allocate(maxCapacity);
+    auto *newSelf = allocators->node16.allocate(capacity);
     memcpy((char *)newSelf + kNodeCopyBegin, (char *)self + kNodeCopyBegin,
            kNodeCopySize);
     memcpy(newSelf->partialKey(), self16->partialKey(), self->partialKeyLen);
@@ -792,7 +792,7 @@ void maybeDecreaseCapacity(Node *&self, NodeAllocators *allocators,
   } break;
   case Type::Node48: {
     auto *self48 = (Node48 *)self;
-    auto *newSelf = allocators->node48.allocate(maxCapacity);
+    auto *newSelf = allocators->node48.allocate(capacity);
     memcpy((char *)newSelf + kNodeCopyBegin, (char *)self + kNodeCopyBegin,
            kNodeCopySize);
     memcpy(newSelf->partialKey(), self48->partialKey(), self->partialKeyLen);
@@ -812,7 +812,7 @@ void maybeDecreaseCapacity(Node *&self, NodeAllocators *allocators,
   } break;
   case Type::Node256: {
     auto *self256 = (Node256 *)self;
-    auto *newSelf = allocators->node256.allocate(maxCapacity);
+    auto *newSelf = allocators->node256.allocate(capacity);
     memcpy((char *)newSelf + kNodeCopyBegin, (char *)self + kNodeCopyBegin,
            kNodeCopySize);
     memcpy(newSelf->partialKey(), self256->partialKey(), self->partialKeyLen);
@@ -827,9 +827,27 @@ void maybeDecreaseCapacity(Node *&self, NodeAllocators *allocators,
   }
 }
 
+// Fix larger-than-desired capacities. Does not return nodes to freelists,
+// since that wouldn't actually reclaim the memory used for partial key
+// capacity.
+void maybeDecreaseCapacity(Node *&self, NodeAllocators *allocators,
+                           ConflictSet::Impl *impl) {
+  const int maxCapacity =
+      std::max<int>(self->numChildren, 1) * self->partialKeyLen;
+  if (self->partialKeyCapacity <= maxCapacity) {
+    return;
+  }
+  makeCapacityAtLeast(self, maxCapacity, allocators, impl);
+}
+
 // TODO fuse into erase child so we don't need to repeat branches on type
 void maybeDownsize(Node *self, NodeAllocators *allocators,
-                   ConflictSet::Impl *impl) {
+                   ConflictSet::Impl *impl, Node *&dontInvalidate) {
+
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+  fprintf(stderr, "maybeDownsize: %s\n", getSearchPathPrintable(self).c_str());
+#endif
+
   switch (self->type) {
   case Type::Node0:
     __builtin_unreachable(); // GCOVR_EXCL_LINE
@@ -849,11 +867,11 @@ void maybeDownsize(Node *self, NodeAllocators *allocators,
         int minCapacity = self4->partialKeyLen + 1 + child->partialKeyLen;
 
         if (minCapacity > child->partialKeyCapacity) {
-          // TODO resize child? It seems to be quite challenging to implement,
-          // since callers would now have to account for erase invalidating
-          // not on the search path. We could lower kBytesPerKey by doing this
-          // though.
-          return;
+          const bool update = child == dontInvalidate;
+          makeCapacityAtLeast(child, minCapacity, allocators, impl);
+          if (update) {
+            dontInvalidate = child;
+          }
         }
 
         // Merge partial key with child
@@ -885,7 +903,7 @@ void maybeDownsize(Node *self, NodeAllocators *allocators,
     }
   } break;
   case Type::Node16:
-    if (self->numChildren < kMinChildrenNode16) {
+    if (self->numChildren + int(self->entryPresent) < kMinChildrenNode16) {
       auto *self16 = (Node16 *)self;
       auto *newSelf = allocators->node4.allocate(self->partialKeyLen);
       memcpy((char *)newSelf + kNodeCopyBegin, (char *)self + kNodeCopyBegin,
@@ -902,7 +920,7 @@ void maybeDownsize(Node *self, NodeAllocators *allocators,
     }
     break;
   case Type::Node48:
-    if (self->numChildren < kMinChildrenNode48) {
+    if (self->numChildren + int(self->entryPresent) < kMinChildrenNode48) {
       auto *self48 = (Node48 *)self;
       auto *newSelf = allocators->node16.allocate(self->partialKeyLen);
       memcpy((char *)newSelf + kNodeCopyBegin, (char *)self + kNodeCopyBegin,
@@ -915,7 +933,7 @@ void maybeDownsize(Node *self, NodeAllocators *allocators,
             // Suppress a false positive -Waggressive-loop-optimizations warning
             // in gcc. `assume` doesn't work for some reason.
             if (!(i < 16)) {
-              __builtin_unreachable();
+              __builtin_unreachable(); // GCOVR_EXCL_LINE
             }
             newSelf->index[i] = c;
             newSelf->children[i] = self48->children[self48->index[c]];
@@ -929,7 +947,7 @@ void maybeDownsize(Node *self, NodeAllocators *allocators,
     }
     break;
   case Type::Node256:
-    if (self->numChildren < kMinChildrenNode256) {
+    if (self->numChildren + int(self->entryPresent) < kMinChildrenNode256) {
       auto *self256 = (Node256 *)self;
       auto *newSelf = allocators->node48.allocate(self->partialKeyLen);
       memcpy((char *)newSelf + kNodeCopyBegin, (char *)self + kNodeCopyBegin,
@@ -954,16 +972,28 @@ void maybeDownsize(Node *self, NodeAllocators *allocators,
 }
 
 // Precondition: self is not the root. May invalidate nodes along the search
-// path to self.
-Node *erase(Node *self, NodeAllocators *allocators, ConflictSet::Impl *impl) {
+// path to self. May invalidate children of self->parent. Returns a pointer to
+// the node after self. If erase invalidates the pointee of `dontInvalidate`, it
+// will update it to its new pointee as well.
+Node *erase(Node *self, NodeAllocators *allocators, ConflictSet::Impl *impl,
+            Node *&dontInvalidate) {
   assert(self->parent != nullptr);
 
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+  fprintf(stderr, "Erase: %s\n", getSearchPathPrintable(self).c_str());
+#endif
+
   Node *parent = self->parent;
   uint8_t parentsIndex = self->parentsIndex;
 
   auto *result = nextLogical(self);
   self->entryPresent = false;
   if (self->numChildren != 0) {
+    const bool update = result == dontInvalidate;
+    maybeDownsize(self, allocators, impl, result);
+    if (update) {
+      dontInvalidate = result;
+    }
     return result;
   }
 
@@ -1017,9 +1047,13 @@ Node *erase(Node *self, NodeAllocators *allocators, ConflictSet::Impl *impl) {
   --parent->numChildren;
   if (parent->numChildren == 0 && !parent->entryPresent &&
       parent->parent != nullptr) {
-    erase(parent, allocators, impl);
+    return erase(parent, allocators, impl, dontInvalidate);
   } else {
-    maybeDownsize(parent, allocators, impl);
+    const bool update = result == dontInvalidate;
+    maybeDownsize(parent, allocators, impl, result);
+    if (update) {
+      dontInvalidate = result;
+    }
   }
   return result;
 }
@@ -1223,10 +1257,6 @@ struct SearchStepWise {
   }
 };
 
-namespace {
-std::string getSearchPathPrintable(Node *n);
-}
-
 // Logically this is the same as performing firstGeq and then checking against
 // point or range version according to cmp, but this version short circuits as
 // soon as it can prove that there's no conflict.
@@ -2015,7 +2045,7 @@ void addWriteRange(Node *&root, int64_t oldestVersion,
   }
 
   for (beginNode = nextLogical(beginNode); beginNode != endNode;
-       beginNode = erase(beginNode, allocators, impl)) {
+       beginNode = erase(beginNode, allocators, impl, endNode)) {
   }
 }
 
@@ -2141,7 +2171,8 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
         // There's no way to insert a range such that range version of the right
         // node is greater than the point version of the left node
         assert(n->entry.rangeVersion <= oldestVersion);
-        n = erase(n, &allocators, this);
+        Node *dummy = nullptr;
+        n = erase(n, &allocators, this, dummy);
       } else {
         maybeDecreaseCapacity(n, &allocators, this);
         n = nextLogical(n);
@@ -2493,11 +2524,12 @@ Iterator firstGeq(Node *n, std::string_view key) {
     minNumChildren = kMinChildrenNode256;
     break;
   }
-  if (node->numChildren < minNumChildren) {
+  if (node->numChildren + int(node->entryPresent) < minNumChildren) {
     fprintf(stderr,
-            "%s has %d children, which is less than the minimum required %d\n",
+            "%s has %d children + %d entries, which is less than the minimum "
+            "required %d\n",
             getSearchPathPrintable(node).c_str(), node->numChildren,
-            minNumChildren);
+            int(node->entryPresent), minNumChildren);
     success = false;
   }
   // TODO check that the max capacity property eventually holds

Internal.h

@@ -482,7 +482,7 @@ template <class ConflictSetImpl> struct TestDriver {
   ConflictSetImpl cs{oldestVersion};
   ReferenceImpl refImpl{oldestVersion};
 
-  constexpr static auto kMaxKeyLen = 64;
+  constexpr static auto kMaxKeyLen = 128;
 
   bool ok = true;
 

corpus/05a79f06cf3f67f726dae68d18a2290f6c9a50c9

@@ -0,0 +1 @@
+: