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]>

Author: romange

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,28 @@ 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_) {
+      auto* fw = head_;
+      auto* rev = head_->prev;
+      node_id = 0;
+      while (node_id <= len_ / 2 &&
+             (num_offloaded_nodes_ + 2 * tiering_params_->node_depth_threshold) < len_) {
+        if (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;
+        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 {
 
@@ -99,6 +101,10 @@ class ListWrapper {
     if (ShouldPromoteToQL(lp.GetPointer(), value.size())) {
       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());
       } else {

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;