fix(tiering): implement qlist symmetric cooloff and fix memory handling

Implements the symmetric cooloff logic for QList tiering that walks
from both ends of the list and offloads nodes beyond the threshold.
Fixes memory handling bugs in move operator and null pointer check.

Changes:
- Implement symmetric cooloff in CoolOff() for balanced tiering
- Fix InsertNode() null pointer check when insert_opt is AFTER
- Fix move operator to properly reset num_offloaded_nodes_
- Add Stats::operator+= for cross-shard aggregation
- Add Prometheus metrics for list reads and tiering events
- Add list_tiering_threshold flag for configuration
- Fix typo: "overlaoding" -> "overloading"

Co-Authored-By: Claude Sonnet 4.5 <[email protected]>
Signed-off-by: Roman Gershman <[email protected]>

Author: romange

src/core/detail/listpack.h

@@ -76,6 +76,10 @@ class ListPack {
     return lp_;
   }
 
+  size_t BytesSize() const {
+    return lpBytes(lp_);
+  }
+
  private:
   static CollectionEntry GetEntry(uint8_t* pos);
 

src/core/qlist.cc

@@ -417,6 +417,24 @@ QList::Node* SplitNode(QList::Node* node, int offset, bool after, ssize_t* diff)
 
 __thread QList::Stats QList::stats;
 
+QList::Stats& QList::Stats::operator+=(const Stats& other) {
+#define ADD_FIELD(field) this->field += other.field;
+
+  ADD_FIELD(compression_attempts);
+  ADD_FIELD(bad_compression_attempts);
+  ADD_FIELD(decompression_calls);
+  ADD_FIELD(compressed_bytes);
+  ADD_FIELD(raw_compressed_bytes);
+  ADD_FIELD(interior_node_reads);
+  ADD_FIELD(total_node_reads);
+  ADD_FIELD(offload_requests);
+  ADD_FIELD(onload_requests);
+
+#undef ADD_FIELD
+
+  return *this;
+}
+
 size_t QList::Node::GetLZF(void** data) const {
   DCHECK(encoding == QUICKLIST_NODE_ENCODING_LZF || encoding == QLIST_NODE_ENCODING_LZ4);
   quicklistLZF* lzf = (quicklistLZF*)entry;
@@ -485,9 +503,9 @@ QList& QList::operator=(QList&& other) noexcept {
     compress_ = other.compress_;
     bookmark_count_ = other.bookmark_count_;
     tiering_params_ = std::move(other.tiering_params_);
-
+    num_offloaded_nodes_ = other.num_offloaded_nodes_;
     other.head_ = nullptr;
-    other.len_ = other.count_ = 0;
+    other.len_ = other.count_ = other.num_offloaded_nodes_ = 0;
   }
   return *this;
 }
@@ -701,7 +719,7 @@ void QList::InsertNode(Node* old_node, Node* new_node, uint32_t old_node_id, Ins
 
   // Calculate final positions AFTER all linkage and len_ updates are complete.
   uint32_t new_node_id;
-  if (insert_opt == AFTER) {
+  if (insert_opt == AFTER && old_node) {
     new_node_id = old_node_id + 1;  // new_node inserted after, old_node position unchanged
   } else {
     new_node_id = old_node_id;  // new_node takes old_node's position
@@ -892,8 +910,34 @@ void QList::CoolOff(Node* node, uint32_t node_id) {
         stats.offload_requests++;
         node->offloaded = 1;
       }
-    } else if (len_ > num_offloaded_nodes_ * 2 + tiering_params_->node_depth_threshold * 2) {
-      // TBD.
+    } else if (num_offloaded_nodes_ * 2 + tiering_params_->node_depth_threshold * 2 < len_) {
+      // We check `num_offloaded_nodes_ * 2` above to avoid frequent traversals.
+      // So only when the gap between offloaded and non-offloaded nodes is large enough,
+      // we do a traversal to offload more nodes.
+      auto* fw = head_;
+      auto* rev = head_->prev;
+      uint32_t traverse_node_id = 0;
+
+      // Traverse from both ends towards the middle as we expect more offloads towards the ends
+      // due to usuall access patterns of adding items via lpush/rpush.
+      while (traverse_node_id <= len_ / 2 &&
+             (num_offloaded_nodes_ + 2 * tiering_params_->node_depth_threshold) < len_) {
+        if (traverse_node_id >= tiering_params_->node_depth_threshold) {
+          if (fw->offloaded == 0) {
+            num_offloaded_nodes_++;
+            stats.offload_requests++;
+            fw->offloaded = 1;
+          }
+          if (rev->offloaded == 0) {
+            num_offloaded_nodes_++;
+            stats.offload_requests++;
+            rev->offloaded = 1;
+          }
+        }
+        fw = fw->next;
+        rev = rev->prev;
+        traverse_node_id++;
+      }
     }
   }
 

src/core/qlist.h

@@ -248,6 +248,8 @@ class QList {
     uint64_t total_node_reads = 0;
     uint64_t offload_requests = 0;
     uint64_t onload_requests = 0;
+
+    Stats& operator+=(const Stats& other);
   };
   static __thread Stats stats;
 

src/server/list_family.cc

@@ -61,6 +61,8 @@ ABSL_FLAG(int32_t, list_max_listpack_size, -2, "Maximum listpack size, default i
  */
 
 ABSL_FLAG(int32_t, list_compress_depth, 0, "Compress depth of the list. Default is no compression");
+ABSL_FLAG(unsigned, list_tiering_threshold, 0,
+          "Tiering threshold for lists. Default - no tiering.");
 
 namespace dfly {
 
@@ -86,22 +88,30 @@ class ListWrapper {
     return std::visit(Overload{[&f](auto* s) { return f(*s); }, f}, impl_);
   }
 
-  static bool ShouldPromoteToQL(uint8_t* lp, size_t new_size) {
-    size_t sz = lpBytes(lp);
-    return !ShouldStoreAsListPack(sz + new_size);
+  static QList* PromoteToQLIfNeeded(LP lp, size_t additional_size) {
+    size_t sz = lp.BytesSize();
+    if (ShouldStoreAsListPack(sz + additional_size)) {
+      return nullptr;
+    }
+    QList* ql = CompactObj::AllocateMR<QList>(GetFlag(FLAGS_list_max_listpack_size),
+                                              GetFlag(FLAGS_list_compress_depth));
+    if (GetFlag(FLAGS_list_tiering_threshold) > 0) {
+      ql->SetTieringParams(
+          QList::TieringParams{.node_depth_threshold = GetFlag(FLAGS_list_tiering_threshold)});
+    }
+    if (lp.Size() > 0) {
+      ql->AppendListpack(lp.GetPointer());
+    }
+    return ql;
   }
 
   void PushInternal(string_view value, QList::Where where, QList& ql) {
     ql.Push(value, where);
   }
 
   void PushInternal(string_view value, QList::Where where, LP& lp) {
-    if (ShouldPromoteToQL(lp.GetPointer(), value.size())) {
-      QList* ql = CompactObj::AllocateMR<QList>(GetFlag(FLAGS_list_max_listpack_size),
-                                                GetFlag(FLAGS_list_compress_depth));
-      if (lp.Size() > 0) {
-        ql->AppendListpack(lp.GetPointer());
-      } else {
+    if (QList* ql = PromoteToQLIfNeeded(lp, value.size()); ql) {
+      if (lp.Size() == 0) {  // otherwise we already appended it in PromoteToQLIfNeeded.
         lpFree(lp.GetPointer());
       }
       ql->Push(value, where);
@@ -120,14 +130,12 @@ class ListWrapper {
     if (!p)
       return false;
 
-    if (ShouldPromoteToQL(lp.GetPointer(), elem.size())) {
-      QList* ql = CompactObj::AllocateMR<QList>(GetFlag(FLAGS_list_max_listpack_size),
-                                                GetFlag(FLAGS_list_compress_depth));
-      DCHECK_GT(lp.Size(), 0u);  // otherwise we would not find anything
+    if (QList* ql = PromoteToQLIfNeeded(lp, elem.size()); ql) {
+      DCHECK_GT(ql->Size(), 0u);  // otherwise we would not Find the pivot.
       impl_ = ql;
-      ql->AppendListpack(lp.GetPointer());
       return ql->Insert(pivot, elem, insert_opt);
     }
+
     lp.Insert(p, elem, insert_opt);
     return true;
   }
@@ -141,12 +149,9 @@ class ListWrapper {
     if (!p)
       return false;
 
-    if (ShouldPromoteToQL(lp.GetPointer(), elem.size())) {
-      QList* ql = CompactObj::AllocateMR<QList>(GetFlag(FLAGS_list_max_listpack_size),
-                                                GetFlag(FLAGS_list_compress_depth));
-      DCHECK_GT(lp.Size(), 0u);  // otherwise we would not seek
+    if (QList* ql = PromoteToQLIfNeeded(lp, elem.size()); ql) {
+      DCHECK_GT(ql->Size(), 0u);  // otherwise we would not seek
       impl_ = ql;
-      ql->AppendListpack(lp.GetPointer());
       return ql->Replace(index, elem);
     }
     lp.Replace(p, elem);

src/server/server_family.cc

@@ -1963,6 +1963,12 @@ void PrintPrometheusMetrics(uint64_t uptime, const Metrics& m, DflyCmd* dfly_cmd
   AppendMetricWithoutLabels("huffman_tables_built", "Huffman tables built",
                             m.shard_stats.huffman_tables_built, MetricType::COUNTER, &resp->body());
 
+  AppendMetricHeader("list_reads", "List Reads Patterns", MetricType::COUNTER, &resp->body());
+  AppendMetricValue("list_reads", m.qlist_stats.total_node_reads, {"type"}, {"total"},
+                    &resp->body());
+  AppendMetricValue("list_reads", m.qlist_stats.interior_node_reads, {"type"}, {"interior"},
+                    &resp->body());
+
   // Tiered metrics
   {
     AppendMetricWithoutLabels("tiered_entries", "Tiered entries", total.tiered_entries,
@@ -1998,12 +2004,19 @@ void PrintPrometheusMetrics(uint64_t uptime, const Metrics& m, DflyCmd* dfly_cmd
     AppendMetricValue("tiered_hits", m.events.ram_misses, {"type"}, {"disk"}, &resp->body());
 
     // Potential problems due to overloading system
-    AppendMetricHeader("tiered_overload", "Potential problems due to overlaoding",
+    AppendMetricHeader("tiered_overload", "Potential problems due to overloading",
                        MetricType::COUNTER, &resp->body());
     AppendMetricValue("tiered_overload", m.tiered_stats.total_clients_throttled, {"type"},
                       {"client throttling"}, &resp->body());
     AppendMetricValue("tiered_overload", m.tiered_stats.total_stash_overflows, {"type"},
                       {"stash overflows"}, &resp->body());
+
+    AppendMetricHeader("tiered_list_events", "Tiered List Events", MetricType::COUNTER,
+                       &resp->body());
+    AppendMetricValue("tiered_list_events", m.qlist_stats.offload_requests, {"type"}, {"offload"},
+                      &resp->body());
+    AppendMetricValue("tiered_list_events", m.qlist_stats.onload_requests, {"type"}, {"onload"},
+                      &resp->body());
   }
 }
 
@@ -2837,6 +2850,8 @@ Metrics ServerFamily::GetMetrics(Namespace* ns) const {
         result.search_stats += shard->search_indices()->GetStats();
       }
 
+      result.qlist_stats += QList::stats;
+
       result.traverse_ttl_per_sec += shard->GetMovingSum6(EngineShard::TTL_TRAVERSE);
       result.delete_ttl_per_sec += shard->GetMovingSum6(EngineShard::TTL_DELETE);
       if (result.tx_queue_len < shard->txq()->size())

src/server/server_family.h

@@ -9,6 +9,7 @@
 #include <optional>
 #include <string>
 
+#include "core/qlist.h"
 #include "facade/dragonfly_listener.h"
 #include "server/detail/save_stages_controller.h"
 #include "server/dflycmd.h"
@@ -22,7 +23,6 @@
 namespace util {
 
 class AcceptServer;
-class ListenerInterface;
 class HttpListenerBase;
 
 }  // namespace util
@@ -86,6 +86,7 @@ struct Metrics {
   SearchStats search_stats;
   ServerState::Stats coordinator_stats;  // stats on transaction running
   PeakStats peak_stats;
+  QList::Stats qlist_stats;
 
   size_t qps = 0;