chore(DenseSet): defrag all links in a chain (#4019)

* add defrag for all nodes in a link chain
* add tests

---------

Signed-off-by: kostas <kostas@dragonflydb.io>

src/core/dense_set.cc

@@ -840,4 +840,22 @@ size_t DenseSet::SizeSlow() {
   return size_;
 }
 
+size_t DenseSet::IteratorBase::TraverseApply(DensePtr* ptr, std::function<void(DensePtr*)> fun) {
+  size_t links_traversed = 0;
+  while (ptr->IsLink()) {
+    DenseLinkKey* link = ptr->AsLink();
+    fun(link);
+    ptr = &link->next;
+    ++links_traversed;
+  }
+
+  // The last ptr in the link always returns ptr->IsLink() = false
+  DCHECK(!ptr->IsEmpty());
+  DCHECK(ptr->IsObject());
+  fun(ptr);
+  ++links_traversed;
+
+  return links_traversed;
+}
+
 }  // namespace dfly

src/core/dense_set.h

@@ -202,6 +202,11 @@ class DenseSet {
 
     void Advance();
 
+    // If ptr is a link, it calls fun on all links in the chain.
+    // Otherwise it calls it only once on the object.
+    // ptr must be non empty.
+    size_t TraverseApply(DensePtr* ptr, std::function<void(DensePtr*)> fun);
+
     DenseSet* owner_;
     ChainVectorIterator curr_list_;
     DensePtr* curr_entry_;

src/core/score_map.cc

@@ -164,23 +164,21 @@ pair<sds, bool> ReallocIfNeededGeneric(void* obj, float ratio) {
 }  // namespace
 
 bool ScoreMap::iterator::ReallocIfNeeded(float ratio, std::function<void(sds, sds)> cb) {
-  // Unwrap all links to correctly call SetObject()
-  auto* ptr = curr_entry_;
-
-  if (ptr->IsLink()) {
-    ptr = ptr->AsLink();
-  }
-
-  // Note: we do not iterate over the links. Although we could that...
-  auto* obj = ptr->GetObject();
-  auto [new_obj, reallocated] = ReallocIfNeededGeneric(obj, ratio);
-  if (reallocated) {
-    if (cb) {
-      cb((sds)obj, (sds)new_obj);
+  bool reallocated = false;
+  auto body = [ratio, &cb, &reallocated](auto* ptr) {
+    auto* obj = ptr->GetObject();
+    auto [new_obj, realloc] = ReallocIfNeededGeneric(obj, ratio);
+    if (realloc) {
+      if (cb) {
+        cb((sds)obj, (sds)new_obj);
+      }
+      sdsfree((sds)obj);
+      ptr->SetObject(new_obj);
     }
-    sdsfree((sds)obj);
-    ptr->SetObject(new_obj);
-  }
+    reallocated |= realloc;
+  };
+
+  TraverseApply(curr_entry_, body);
 
   return reallocated;
 }

src/core/string_map.cc

@@ -298,17 +298,17 @@ detail::SdsPair StringMap::iterator::BreakToPair(void* obj) {
 }
 
 bool StringMap::iterator::ReallocIfNeeded(float ratio) {
-  // Unwrap all links to correctly call SetObject()
-  auto* ptr = curr_entry_;
-  if (ptr->IsLink()) {
-    ptr = ptr->AsLink();
-  }
+  bool reallocated = false;
+  auto body = [this, ratio, &reallocated](auto* ptr) {
+    auto* obj = ptr->GetObject();
+    auto [new_obj, realloc] = static_cast<StringMap*>(owner_)->ReallocIfNeeded(obj, ratio);
+    ptr->SetObject(new_obj);
+    reallocated |= realloc;
+  };
 
-  // Note: we do not iterate over the links. Although we could that...
-  auto* obj = ptr->GetObject();
-  auto [new_obj, realloced] = static_cast<StringMap*>(owner_)->ReallocIfNeeded(obj, ratio);
-  ptr->SetObject(new_obj);
-  return realloced;
+  TraverseApply(curr_entry_, body);
+
+  return reallocated;
 }
 
 }  // namespace dfly