Implement clears in addMutations

VersionedMap.cpp

@@ -339,17 +339,26 @@ struct Finger {
     direction[searchPathSize_] = dir;
     ++searchPathSize_;
   }
-  void pop() { --searchPathSize_; }
+  void pop() {
+    assert(searchPathSize_ > 0);
+    --searchPathSize_;
+  }
   uint32_t backNode() const {
     assert(searchPathSize_ > 0);
     return searchPath[searchPathSize_ - 1];
   }
+  uint32_t &backNodeRef() {
+    assert(searchPathSize_ > 0);
+    return searchPath[searchPathSize_ - 1];
+  }
   bool backDirection() const {
     assert(searchPathSize_ > 0);
     return direction[searchPathSize_ - 1];
   }
   uint32_t searchPathSize() const { return searchPathSize_; }
 
+  void setSearchPathSizeUnsafe(int size) { searchPathSize_ = size; }
+
   Finger() : searchPathSize_(0) {}
 
   Finger(const Finger &other) {
@@ -494,6 +503,28 @@ struct VersionedMap::Impl {
     int len;
   };
 
+  Finger search(Key key, int64_t at) {
+    Finger finger;
+    bool ignored;
+    finger.push(latestRoot, ignored);
+
+    // Initialize finger to the search path of `key`
+    for (;;) {
+      auto n = finger.backNode();
+      if (n == 0) {
+        break;
+      }
+      auto c = key <=> mm.base[n];
+      if (c == 0) {
+        // No duplicates
+        break;
+      }
+      finger.push(child<std::memory_order_relaxed>(n, c > 0, latestVersion),
+                  c > 0);
+    }
+    return finger;
+  }
+
   // Infers `val` and `clearTo` if not set
   void insert(Key key, std::optional<Val> val, std::optional<bool> clearTo) {
     Finger finger;
@@ -501,7 +532,7 @@ struct VersionedMap::Impl {
     finger.push(latestRoot, ignored);
     bool inserted;
 
-    // Initialize finger to the search path of `m`
+    // Initialize finger to the search path of `key`
     for (;;) {
       auto n = finger.backNode();
       if (n == 0) {
@@ -577,6 +608,69 @@ struct VersionedMap::Impl {
     }
   }
 
+  // Removes `finger` from the tree, and leaves `finger` pointing to the next
+  // entry.
+  void remove(Finger &finger) {
+
+    // True if finger is pointing to an entry > than the entry we're removing
+    // after we rotate it down
+    bool greaterThan;
+
+    // Rotate down until we can remove the entry
+    for (;;) {
+      auto &node = finger.backNodeRef();
+      const auto l =
+          child<std::memory_order_relaxed>(node, false, latestVersion);
+      const auto r =
+          child<std::memory_order_relaxed>(node, true, latestVersion);
+      if (l == 0) {
+        node = r;
+        greaterThan = true;
+        break;
+      } else if (r == 0) {
+        node = l;
+        greaterThan = false;
+        break;
+      } else {
+        const bool direction =
+            mm.base[l].entry->priority > mm.base[r].entry->priority;
+        rotate(node, latestVersion, direction);
+        assert(node != 0);
+        finger.push(
+            child<std::memory_order_relaxed>(node, direction, latestVersion),
+            direction);
+      }
+    }
+
+    // propagate up the search path, all the way to the root since we may have
+    // more rotations to do even if an update doesn't change a node pointer
+    auto node = finger.backNode();
+    const auto oldSize = finger.searchPathSize();
+    for (;;) {
+      if (finger.searchPathSize() == 1) {
+        // Made it to the root
+        latestRoot = node;
+        break;
+      }
+      const bool direction = finger.backDirection();
+      finger.pop();
+      auto &parent = finger.backNodeRef();
+      auto old = parent;
+      parent = update(parent, direction, node, latestVersion);
+      node = parent;
+    }
+    finger.setSearchPathSizeUnsafe(oldSize);
+
+    if (greaterThan) {
+      while (auto c = child<std::memory_order_relaxed>(finger.backNode(), false,
+                                                       latestVersion) != 0) {
+        finger.push(c, false);
+      }
+    } else {
+      move<std::memory_order_relaxed>(finger, latestVersion, true);
+    }
+  }
+
   uint32_t newNode(int64_t version, const uint8_t *key, int keyLen,
                    const uint8_t *val, int valLen, bool clearTo) {
     auto result = mm.allocate();
@@ -596,7 +690,7 @@ struct VersionedMap::Impl {
 
   void printInOrder(int64_t version);
 
-  void printInOrderHelper(int64_t version, uint32_t node);
+  void printInOrderHelper(int64_t version, uint32_t node, int depth);
 
   void addMutations(const Mutation *mutations, int numMutations,
                     int64_t version) {
@@ -612,8 +706,15 @@ struct VersionedMap::Impl {
       } break;
       case Clear: {
         insert({m.param1, m.param1Len}, {{nullptr, -1}}, {});
-        // TODO erase (param1, param2)
+        if (m.param2Len > 0) {
+          auto iter = search({m.param1, m.param1Len}, latestVersion);
+          move<std::memory_order_relaxed>(iter, latestVersion, true);
+          while (iter.backNode() != 0 &&
+                 mm.base[iter.backNode()] < Key{m.param2, m.param2Len}) {
+            remove(iter);
+          }
           insert({m.param2, m.param2Len}, {}, true);
+        }
       } break;
       default:                   // GCOVR_EXCL_LINE
         __builtin_unreachable(); // GCOVR_EXCL_LINE
@@ -660,15 +761,20 @@ void VersionedMap::addMutations(const Mutation *mutations, int numMutations,
 
 void VersionedMap::Impl::printInOrder(int64_t version) {
   printInOrderHelper(version,
-                     roots.getThreadSafeHandle().rootForVersion(version));
+                     roots.getThreadSafeHandle().rootForVersion(version), 0);
 }
 
-void VersionedMap::Impl::printInOrderHelper(int64_t version, uint32_t node) {
+void VersionedMap::Impl::printInOrderHelper(int64_t version, uint32_t node,
+                                            int depth) {
   if (node == 0) {
     return;
   }
   printInOrderHelper(version,
-                     child<std::memory_order_relaxed>(node, false, version));
+                     child<std::memory_order_relaxed>(node, false, version),
+                     depth + 1);
+  for (int i = 0; i < depth; ++i) {
+    printf("  ");
+  }
   printf("%.*s", mm.base[node].entry->keyLen, mm.base[node].entry->getKey());
   if (mm.base[node].entry->valLen >= 0) {
     printf(" -> '%.*s'", mm.base[node].entry->valLen,
@@ -681,7 +787,8 @@ void VersionedMap::Impl::printInOrderHelper(int64_t version, uint32_t node) {
   }
   printf("\n");
   VersionedMap::Impl::printInOrderHelper(
-      version, child<std::memory_order_relaxed>(node, true, version));
+      version, child<std::memory_order_relaxed>(node, true, version),
+      depth + 1);
 }
 
 } // namespace weaselab
@@ -692,14 +799,34 @@ void VersionedMap::Impl::printInOrderHelper(int64_t version, uint32_t node) {
 int main() {
   {
     weaselab::VersionedMap::Impl impl;
+    {
       weaselab::VersionedMap::Mutation m[] = {
           {(const uint8_t *)"a", nullptr, 1, 0, weaselab::VersionedMap::Set},
           {(const uint8_t *)"b", nullptr, 1, 0, weaselab::VersionedMap::Set},
           {(const uint8_t *)"c", nullptr, 1, 0, weaselab::VersionedMap::Set},
+          {(const uint8_t *)"d", nullptr, 1, 0, weaselab::VersionedMap::Set},
+          {(const uint8_t *)"e", nullptr, 1, 0, weaselab::VersionedMap::Set},
+          {(const uint8_t *)"f", nullptr, 1, 0, weaselab::VersionedMap::Set},
       };
       impl.addMutations(m, sizeof(m) / sizeof(m[0]), 1);
-    impl.printInOrder(1);
     }
+    {
+      weaselab::VersionedMap::Mutation m[] = {
+          {(const uint8_t *)"a", (const uint8_t *)"d", 1, 1,
+           weaselab::VersionedMap::Clear},
+      };
+      impl.addMutations(m, sizeof(m) / sizeof(m[0]), 2);
+    }
+    {
+      weaselab::VersionedMap::Mutation m[] = {
+          {(const uint8_t *)"b", (const uint8_t *)"", 1, 0,
+           weaselab::VersionedMap::Clear},
+      };
+      impl.addMutations(m, sizeof(m) / sizeof(m[0]), 3);
+    }
+    impl.printInOrder(3);
+  }
+  return 0;
 
   ankerl::nanobench::Bench bench;
   bench.minEpochIterations(10000);

include/VersionedMap.h

@@ -31,7 +31,7 @@ namespace weaselab {
  *
  * Thread safety:
  *   - It's safe to operate on two different VersionedMaps in two different
- * threads concurrently
+ * threads concurrently.
  *   - It's safe to have multiple threads operating on the same VersionedMap
  * concurrently if all threads only call const methods.
  *   - Methods that make stronger guarantees about the safety of calling
@@ -51,7 +51,7 @@ struct VersionedMap {
     Clear,
   };
 
-  /** bytes ordered bitwise-lexicographically. */
+  /** Bytes ordered bitwise-lexicographically. */
   struct Key {
     const uint8_t *p;
     int len;
@@ -86,12 +86,15 @@ struct VersionedMap {
   /** The version of the most recent call to `setOldestVersion`. */
   int64_t getOldestVersion() const;
 
-  /** Iterates through a canonicalized view of all the mutations from
+  /** Iterates through a partially canonicalized[1] view of all the mutations
-   * `oldestVersion` to the iterator's version. There may be mutations from
+   * from `oldestVersion` to the iterator's version. There may be mutations from
    * versions < `oldestVersion`, but they won't affect the result. It's
    * thread-safe to operate on an iterator concurrently with any method of
    * `VersionedMap`, as long as it's not invalidated by `setOldestVersion`.
-   * @warning must not outlive its `VersionedMap`. */
+   * @warning must not outlive its `VersionedMap`.
+   *
+   * [1]: Mutations are sorted and non-overlapping, but may be adjacent.
+   */
   struct Iterator {
 
     Iterator() = default;