Use mutationIndex instead of materializeClearEndingHere

VersionedMap.cpp

@@ -812,9 +812,17 @@ struct VersionedMap::Iterator::Impl {
   int64_t version;
   const VersionedMap::Impl *map;
   int cmp;
-  // True if this is a point mutation and a range mutation, and we're
+
-  // materializing the range mutation instead of the point mutation.
+  // State for materializing mutations associated with the entry at `finger`.
-  bool materializeClearEndingHere = false;
+  // Cases:
+  // - If finger is a set and the end of a clear, then mutation[0] is the clear
+  // and mutation[1] is the set.
+  // - If finger is a set and not the end of a clear, then mutation[0] is the
+  // set
+  // - If finger is a clear and not a set, then mutation[0] is the clear
+  int mutationCount;
+  int mutationIndex;
+  VersionedMutation mutations[2];
 };
 
 VersionedMap::Iterator::~Iterator() {
@@ -850,48 +858,78 @@ VersionedMap::Iterator::operator=(Iterator &&other) noexcept {
   return *this;
 }
 
-// TODO once we have good testing for this, design a state machine that fuses
-// adjacent point and range clears at the same version.
 VersionedMap::Iterator::VersionedMutation
 VersionedMap::Iterator::operator*() const {
-  assert(impl->finger.backNode() != 0);
   assert(impl->finger.searchPathSize() != 0);
-  const auto &entry = *impl->map->mm.base[impl->finger.backNode()].entry;
+  assert(impl->mutationIndex < impl->mutationCount);
-  if (impl->materializeClearEndingHere) {
+  assert(impl->mutationIndex >= 0);
-    assert(entry.pointMutation() && entry.clearTo());
+  return impl->mutations[impl->mutationIndex];
-    auto prev = *this;
+}
-    --prev;
+
-    const auto &prevEntry =
+void materializeMutations(VersionedMap::Iterator::Impl *impl, const Entry *prev,
-        *prev.impl->map->mm.base[prev.impl->finger.backNode()].entry;
+                          const Entry *next) {
-    return {prevEntry.getKey(), entry.getKey(), prevEntry.keyLen + 1,
+  if (prev == nullptr) {
-            entry.keyLen,       Clear,          entry.rangeVersion};
+    auto copy = impl->finger;
+    impl->map->move<std::memory_order_acquire>(copy, impl->version, false);
+    if (copy.searchPathSize() > 0) {
+      prev = impl->map->mm.base[copy.backNode()].entry;
+    } else {
+      assert(!impl->map->mm.base[impl->finger.backNode()].entry->clearTo());
+    }
   }
-  if (entry.valLen >= 0) {
+  if (next == nullptr) {
-    return {entry.getKey(),    entry.getVal(), entry.keyLen, entry.valLen, Set,
+    auto copy = impl->finger;
-            entry.pointVersion};
+    impl->map->move<std::memory_order_acquire>(copy, impl->version, true);
-  } else {
+    if (copy.searchPathSize() > 0) {
-    return {entry.getKey(), nullptr,           entry.keyLen, 0,
+      next = impl->map->mm.base[copy.backNode()].entry;
-            Clear,          entry.pointVersion};
+    }
+  }
+
+  const auto &entry = *impl->map->mm.base[impl->finger.backNode()].entry;
+  impl->mutationCount = 0;
+  if (entry.clearTo()) {
+    impl->mutations[impl->mutationCount++] = {
+        prev->getKey(),
+        entry.getKey(),
+        prev->pointMutation() && prev->valLen < 0 &&
+                prev->pointVersion == entry.rangeVersion
+            ? prev->keyLen
+            : prev->keyLen + 1,
+        entry.keyLen,
+        VersionedMap::Clear,
+        entry.rangeVersion};
+  }
+  if (entry.pointMutation()) {
+    if (entry.valLen < 0) {
+      if (next == nullptr ||
+          !(next->clearTo() && next->rangeVersion == entry.pointVersion)) {
+        impl->mutations[impl->mutationCount++] = {
+            entry.getKey(),      nullptr,           entry.keyLen, 0,
+            VersionedMap::Clear, entry.pointVersion};
+      }
+    } else {
+      impl->mutations[impl->mutationCount++] = {
+          entry.getKey(), entry.getVal(),    entry.keyLen,
+          entry.valLen,   VersionedMap::Set, entry.pointVersion};
+    }
   }
 }
 
 VersionedMap::Iterator &VersionedMap::Iterator::operator++() {
-  const auto &entry = *impl->map->mm.base[impl->finger.backNode()].entry;
+  if (impl->mutationIndex < impl->mutationCount - 1) {
-  if (impl->materializeClearEndingHere) {
+    ++impl->mutationIndex;
-    assert(entry.pointMutation() && entry.clearTo());
-    impl->materializeClearEndingHere = false;
     return *this;
   }
 
-  impl->map->move<std::memory_order_acquire>(impl->finger, impl->version, true);
+  do {
-  if (impl->finger.searchPathSize() > 0) {
     const auto &entry = *impl->map->mm.base[impl->finger.backNode()].entry;
-    impl->materializeClearEndingHere = entry.pointMutation() && entry.clearTo();
+    impl->map->move<std::memory_order_acquire>(impl->finger, impl->version,
-    assert(entry.pointVersion <= impl->version);
+                                               true);
-    assert(entry.rangeVersion <= impl->version);
+    if (impl->finger.searchPathSize() > 0) {
-  } else {
+      materializeMutations(impl, &entry, nullptr);
-    impl->materializeClearEndingHere = false;
+    }
-  }
+  } while (impl->mutationCount == 0);
+  impl->mutationIndex = 0;
 
   return *this;
 }
@@ -903,16 +941,20 @@ VersionedMap::Iterator VersionedMap::Iterator::operator++(int) {
 }
 
 VersionedMap::Iterator &VersionedMap::Iterator::operator--() {
-  const auto &entry = *impl->map->mm.base[impl->finger.backNode()].entry;
+  if (impl->mutationIndex > 0) {
-  if (entry.pointMutation() && entry.clearTo() &&
+    --impl->mutationIndex;
-      !impl->materializeClearEndingHere) {
-    impl->materializeClearEndingHere = true;
     return *this;
   }
 
-  impl->map->move<std::memory_order_acquire>(impl->finger, impl->version,
+  do {
-                                             false);
+    const auto &entry = *impl->map->mm.base[impl->finger.backNode()].entry;
-  impl->materializeClearEndingHere = false;
+    impl->map->move<std::memory_order_acquire>(impl->finger, impl->version,
+                                               false);
+    if (impl->finger.searchPathSize() > 0) {
+      materializeMutations(impl, nullptr, &entry);
+    }
+  } while (impl->mutationCount == 0);
+  impl->mutationIndex = impl->mutationCount - 1;
   return *this;
 }
 
@@ -930,8 +972,7 @@ bool VersionedMap::Iterator::operator==(const Iterator &other) const {
     return impl->finger.searchPathSize() == other.impl->finger.searchPathSize();
   }
   return impl->finger.backNode() == other.impl->finger.backNode() &&
-         impl->materializeClearEndingHere ==
+         impl->mutationIndex == other.impl->mutationIndex;
-             other.impl->materializeClearEndingHere;
 }
 
 void VersionedMap::Impl::firstGeq(const Key *key, const int64_t *version,
@@ -960,9 +1001,6 @@ void VersionedMap::Impl::firstGeq(const Key *key, const int64_t *version,
       move<std::memory_order_acquire>(finger, version[i], true);
       if (finger.searchPathSize() > 0) {
         assert(finger.backNode() != 0);
-        const auto &entry = *mm.base[finger.backNode()].entry;
-        iterator[i].impl->materializeClearEndingHere =
-            entry.clearTo() && entry.pointMutation();
       }
     } else {
       iterator[i].impl->cmp = 0;
@@ -971,6 +1009,24 @@ void VersionedMap::Impl::firstGeq(const Key *key, const int64_t *version,
     iterator[i].impl->finger = finger;
     iterator[i].impl->version = version[i];
     iterator[i].impl->map = this;
+
+    const Entry *prev = nullptr;
+    for (;;) {
+      if (iterator[i].impl->finger.searchPathSize() > 0) {
+        materializeMutations(iterator[i].impl, prev, nullptr);
+        if (iterator[i].impl->mutationCount > 0) {
+          break;
+        }
+      } else {
+        break;
+      }
+      prev = iterator[i]
+                 .impl->map->mm.base[iterator[i].impl->finger.backNode()]
+                 .entry;
+      iterator[i].impl->map->move<std::memory_order_acquire>(
+          iterator[i].impl->finger, iterator[i].impl->version, true);
+    }
+    iterator[i].impl->mutationIndex = 0;
   }
 }
 
@@ -1003,7 +1059,23 @@ VersionedMap::Iterator VersionedMap::begin(int64_t version) const {
     }
   }
   result.impl->map = impl;
-  result.impl->materializeClearEndingHere = false;
+
+  const Entry *prev = nullptr;
+  for (;;) {
+    if (result.impl->finger.searchPathSize() > 0) {
+      materializeMutations(result.impl, prev, nullptr);
+      if (result.impl->mutationCount > 0) {
+        break;
+      }
+    } else {
+      break;
+    }
+    prev = result.impl->map->mm.base[result.impl->finger.backNode()].entry;
+    result.impl->map->move<std::memory_order_acquire>(
+        result.impl->finger, result.impl->version, true);
+  }
+  result.impl->mutationIndex = 0;
+
   result.impl->version = version;
   return result;
 }
@@ -1013,7 +1085,7 @@ VersionedMap::Iterator VersionedMap::end(int64_t version) const {
   result.impl = new (malloc(sizeof(Iterator::Impl))) Iterator::Impl();
   result.impl->cmp = 1;
   result.impl->map = impl;
-  result.impl->materializeClearEndingHere = false;
+  result.impl->mutationIndex = 0;
   result.impl->version = version;
   return result;
 }
@@ -1101,7 +1173,7 @@ int main() {
       };
       versionedMap.addMutations(m, sizeof(m) / sizeof(m[0]), 3);
     }
-    const int64_t v = 2;
+    const int64_t v = 3;
     cast(versionedMap)->printInOrder(v);
     weaselab::VersionedMap::Key k = {(const uint8_t *)"a", 1};
     weaselab::VersionedMap::Iterator iter;