feat(tiering): Experimental QList::Node tiering

* Implement offload, onload and delete callback in TieringParams.
* Added ListNodeTieringTest suite.
* Added python test to verify replication with tiered list objects.

Signed-off-by: mkaruza <mario@dragonflydb.io>

Author: mkaruza

src/core/qlist.cc

@@ -426,6 +426,14 @@ void QList::Node::SetExternal(size_t offset, uint32_t size) {
   ext_size = size;
 }
 
+void QList::Node::Upload(QList* ql, std::string_view val) {
+  entry = static_cast<unsigned char*>(zmalloc(val.size()));
+  memcpy(entry, val.data(), val.size());
+  ql->AdjustMallocSize(val.size());
+  ql->AdjustOffloadNodeCount(-1);
+  offloaded = 0;
+}
+
 void QList::SetPackedThreshold(unsigned threshold) {
   packed_threshold = threshold;
 }
@@ -480,10 +488,9 @@ QList::QList(QList&& other) noexcept
       tiering_enabled_(other.tiering_enabled_),
       compress_(other.compress_),
       bookmark_count_(other.bookmark_count_),
-      num_offloaded_nodes_(other.num_offloaded_nodes_) {
+      tiering_params_(std::move(other.tiering_params_)) {
   other.head_ = nullptr;
   other.len_ = other.count_ = 0;
-  other.num_offloaded_nodes_ = 0;
 }
 
 QList::~QList() {
@@ -503,9 +510,9 @@ QList& QList::operator=(QList&& other) noexcept {
     tiering_enabled_ = other.tiering_enabled_;
     compress_ = other.compress_;
     bookmark_count_ = other.bookmark_count_;
-    num_offloaded_nodes_ = other.num_offloaded_nodes_;
     other.head_ = nullptr;
-    other.len_ = other.count_ = other.num_offloaded_nodes_ = 0;
+    other.len_ = other.count_ = 0;
+    tiering_params_ = std::move(other.tiering_params_);
   }
   return *this;
 }
@@ -519,7 +526,7 @@ void QList::Clear() noexcept {
     // If entry is offloaded we should skip freeing its memory.
     bool free_entry = current->offloaded == 0;
     if (tiering_enabled_ && (current->offloaded || current->io_pending)) {
-      CleanupOffloadedNode(current);
+      tiering_params_->cleanup(this, current);
     } else {
       if (current->encoding != QUICKLIST_NODE_ENCODING_RAW) {
         quicklistLZF* lzf = (quicklistLZF*)current->entry;
@@ -540,7 +547,6 @@ void QList::Clear() noexcept {
   head_ = nullptr;
   count_ = 0;
   malloc_size_ = 0;
-  num_offloaded_nodes_ = 0;
 }
 
 void QList::Push(string_view value, Where where) {
@@ -915,6 +921,9 @@ void QList::Replace(Iterator it, std::string_view elem) {
 
 void QList::CoolOff(Node* node, uint32_t node_id) {
   if (tiering_enabled_) {
+    uint32_t threshold = tiering_params_->node_depth_threshold;
+    uint32_t num_offloaded_nodes = tiering_params_->num_offloaded_nodes;
+
     // Dry run for offloading decision.
     // a. Node id is withing the offloadable depth - offload it if not already offloaded.
     // b. Node id is outside the offloadable depth - but we have too many nodes that are not
@@ -925,12 +934,11 @@ void QList::CoolOff(Node* node, uint32_t node_id) {
     //    we won't need to traverse them again for "trivial" access patterns unless they
     //    get accessed again. Another reason for missing offloaded nodes is that node_id can be
     //    off due to merges (can be improved in future).
-    if (node_id >= tiering_node_depth_threshold_ &&
-        node_id + tiering_node_depth_threshold_ < len_) {
+    if (node_id >= threshold && node_id + threshold < len_) {
       if (!node->offloaded && !node->io_pending) {
-        OffloadNode(node);
+        tiering_params_->offload(this, node);
       }
-    } else if (num_offloaded_nodes_ * 2 + tiering_node_depth_threshold_ * 2 < len_) {
+    } else if (num_offloaded_nodes * 2 + 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.
@@ -940,16 +948,15 @@ void QList::CoolOff(Node* node, uint32_t node_id) {
 
       // Traverse from both ends towards the middle as we expect more offloads towards the ends
       // due to usual access patterns of adding items via lpush/rpush.
-      while (traverse_node_id <= len_ / 2 &&
-             (num_offloaded_nodes_ + 2 * tiering_node_depth_threshold_) < len_) {
-        if (traverse_node_id >= tiering_node_depth_threshold_) {
+      while (traverse_node_id <= len_ / 2 && (num_offloaded_nodes + 2 * threshold) < len_) {
+        if (traverse_node_id >= threshold) {
           if (fw->offloaded == 0 && fw->io_pending == 0) {
-            OffloadNode(fw);
+            tiering_params_->offload(this, fw);
           }
 
           // Avoid offloading the same node twice when fw and rev meet in the middle.
           if (rev != fw && rev->offloaded == 0 && rev->io_pending == 0) {
-            OffloadNode(rev);
+            tiering_params_->offload(this, rev);
           }
         }
         fw = fw->next;
@@ -1047,12 +1054,12 @@ void QList::Materialize(Node* node) {
 
   // Cancel stash in progress before loading.
   if (node->io_pending) {
-    CleanupOffloadedNode(node);
+    tiering_params_->cleanup(this, node);
   }
 
   // Load the offloaded node data back into memory.
   if (node->offloaded) {
-    ReadOffloadedNode(node);
+    tiering_params_->load(this, node);
   }
 
   DCHECK(!node->offloaded);
@@ -1193,7 +1200,7 @@ void QList::DelNode(Node* node) {
   }
 
   if (tiering_enabled_ && (node->offloaded || node->io_pending)) {
-    CleanupOffloadedNode(node);
+    tiering_params_->cleanup(this, node);
   }
 
   /* If we deleted a node within our compress depth, we
@@ -1241,18 +1248,20 @@ bool QList::DelPackedIndex(Node* node, uint8_t* p) {
   return false;
 }
 
-void QList::OffloadNode(Node* node) const {
-  DCHECK(tiering_enabled_ && node->offloaded == 0 && node->io_pending == 0);
-  stats.offload_requests++;
-  node->io_pending = 1;
-}
-
-void QList::ReadOffloadedNode(QList::Node* node) const {
-  stats.onload_requests++;
-}
-
-void QList::CleanupOffloadedNode(QList::Node* node) const {
-  node->io_pending = 0;
+void QList::SetDbIndex(DbIndex db_id) {
+  if (db_id_ == db_id) {
+    return;
+  }
+  // Materialize all offloaded nodes before changing db_id
+  if (tiering_enabled_ && tiering_params_->num_offloaded_nodes > 0) {
+    Node* node = head_;
+    while (node) {
+      Node* next = node->next;
+      Materialize(node);
+      node = (next == head_) ? nullptr : next;
+    }
+  }
+  db_id_ = db_id;
 }
 
 void QList::InitIteratorEntry(Iterator* it) const {

src/core/qlist.h

@@ -10,8 +10,10 @@
 #include <functional>
 #include <memory>
 #include <string>
+#include <string_view>
 
 #include "core/collection_entry.h"
+#include "server/common_types.h"
 
 #define QL_COMP_BITS 16
 #define QL_BM_BITS 4
@@ -85,6 +87,7 @@ class QList {
     size_t GetLZF(void** data) const;
 
     void SetExternal(size_t offset, uint32_t sz);
+    void Upload(QList* ql, std::string_view val);
     std::pair<size_t, size_t> GetExternalSlice() const {
       return std::make_pair(size_t(ext_offset), size_t(ext_size));
     }
@@ -122,9 +125,22 @@ class QList {
   }
 
   // Add to the number of offloaded nodes by one.
-  void IncrementNumOffloadedNodes(int delta) {
-    num_offloaded_nodes_ += delta;
+  void AdjustOffloadNodeCount(int delta) {
+    if (tiering_enabled_) {
+      tiering_params_->num_offloaded_nodes += delta;
+    }
+  }
+
+  DbIndex GetDbIndex() const {
+    return db_id_;
   }
+  struct TieringParams {
+    uint32_t num_offloaded_nodes = 0;
+    uint32_t node_depth_threshold = 0;
+    void (*offload)(QList* ql, Node* node) = nullptr;
+    void (*load)(QList* ql, Node* node) = nullptr;
+    void (*cleanup)(QList* ql, Node* node) = nullptr;
+  };
 
   /**
    * fill: The number of entries allowed per internal list node can be specified
@@ -260,12 +276,17 @@ class QList {
     zstd_threshold_ = threshold;
   }
 
-  // Enable tiered storage and set node depth threshold
-  void EnableTiering(uint32_t threshold) {
+  // Enable tiered storage.
+  void EnableTiering(const TieringParams& params) {
     tiering_enabled_ = 1;
-    tiering_node_depth_threshold_ = threshold;
+    tiering_params_ = std::make_unique<TieringParams>(params);
   }
 
+  // Updates the db index associated with this list.
+  // If tiering is enabled and we have offloaded node materialize
+  // them if db index changed.
+  void SetDbIndex(DbIndex db_id);
+
   struct Stats {
     uint64_t compression_attempts = 0;
 
@@ -340,13 +361,6 @@ class QList {
   void DelNode(Node* node);
   bool DelPackedIndex(Node* node, uint8_t* p);
 
-  // Offload node to tiered storage
-  void OffloadNode(Node* node) const;
-  // Read offloaded node from tiered storage
-  void ReadOffloadedNode(Node* node) const;
-  // Delete offloaded node or cancel offloading of node
-  void CleanupOffloadedNode(Node* node) const;
-
   // Initializes iterator's zi_ to point to the element at offset_.
   // Decompresses the node if needed. Assumes current_ is not null.
   void InitIteratorEntry(Iterator* it) const;
@@ -364,9 +378,9 @@ class QList {
   unsigned compress_ : QL_COMP_BITS; /* depth of end nodes not to compress;0=off */
   unsigned bookmark_count_ : QL_BM_BITS;
   unsigned reserved2_ : 12;
-  uint32_t num_offloaded_nodes_ = 0;
-  uint32_t zstd_threshold_ = 0;                // 0 = disabled
-  uint32_t tiering_node_depth_threshold_ = 0;  // 0 = disabled
+  uint16_t db_id_ = kInvalidDbId;
+  uint32_t zstd_threshold_ = 0;  // 0 = disabled
+  std::unique_ptr<TieringParams> tiering_params_;
 };
 
 }  // namespace dfly

src/core/qlist_test.cc

@@ -411,20 +411,6 @@ TEST_F(QListTest, DefragmentListpackCompressed) {
   ASSERT_EQ(i, total_items);
 }
 
-TEST_F(QListTest, Tiering) {
-  QList::stats.offload_requests = 0;
-  QList tiered_ql;
-
-  // Enable tiering and set node_depth_threshold = 1
-  tiered_ql.EnableTiering(1);
-
-  for (int i = 0; i < 8000; i++) {
-    tiered_ql.Push(absl::StrCat("value", i), QList::TAIL);
-  }
-
-  EXPECT_EQ(QList::stats.offload_requests, 9);
-}
-
 TEST_F(QListTest, InsertPivotSplitMergeMallocSize) {
   QList ql(2, 0);  // fill=2: at most 2 elements per node
   ql.Push("x", QList::TAIL);

src/server/db_slice.cc

@@ -1566,6 +1566,12 @@ void DbSlice::RemoveOffloadedEntriesFromTieredStorage(absl::Span<const DbIndex>
         } else if (it->second.HasStashPending()) {
           tiered_storage->CancelStash(std::make_pair(index, it->first.GetSlice(&scratch)),
                                       &it->second);
+        } else if (it->second.ObjType() == OBJ_LIST && it->second.Encoding() == kEncodingQL2) {
+          // Nodes can be offloaded to tiered storage, but the main object doesn't have external or
+          // pending flag set. We need to clear the list explicitly.
+          if (auto* ql = static_cast<QList*>(it->second.RObjPtr()); ql) {
+            ql->Clear();
+          }
         }
       });
     } while (cursor);

src/server/generic_family.cc

@@ -972,6 +972,11 @@ OpStatus OpMove(const OpArgs& op_args, string_view key, DbIndex target_db) {
   auto& add_res = *op_result;
   add_res.it->first.SetSticky(sticky);
 
+  if (add_res.it->second.ObjType() == OBJ_LIST && add_res.it->second.Encoding() == kEncodingQL2) {
+    auto* ql = static_cast<QList*>(add_res.it->second.RObjPtr());
+    ql->SetDbIndex(target_db);
+  }
+
   auto bc = op_args.db_cntx.ns->GetBlockingController(op_args.shard->shard_id());
   if (add_res.it->second.ObjType() == OBJ_LIST && bc) {
     bc->Awaken(target_db, key);

src/server/list_family.cc

@@ -82,6 +82,37 @@ using time_point = Transaction::time_point;
 
 namespace {
 
+void OffloadListNode(QList* ql, QList::Node* node) {
+  TieredStorage* ts = EngineShard::tlocal()->tiered_storage();
+  DCHECK(ts);
+  QList::stats.offload_requests++;
+  StashListNode(ql->GetDbIndex(), ql, node, ts, nullptr);
+}
+
+void LoadListNode(QList* ql, QList::Node* node) {
+  TieredStorage* ts = EngineShard::tlocal()->tiered_storage();
+  DCHECK(ts);
+  QList::stats.onload_requests++;
+  auto res = ReadTieredListNode(ql->GetDbIndex(), ql, node, node->GetExternalSlice(), ts).Get();
+  if (!res) {
+    LOG(WARNING) << "Failed to load list node from tiered storage: " << res.error().message();
+  }
+}
+
+void CleanupListNode(QList* ql, QList::Node* node) {
+  TieredStorage* ts = EngineShard::tlocal()->tiered_storage();
+  DCHECK(ts);
+  if (!ts->IsClosed()) {
+    if (node->io_pending) {
+      ts->CancelStash(tiering::ListNodeId{ql->GetDbIndex(), ql, node}, node);
+    } else {
+      // We don't pass QList pointer so we need to decrease num_offloaded_nodes_ now.
+      ql->AdjustOffloadNodeCount(-1);
+      ts->Delete(ql->GetDbIndex(), node);
+    }
+  }
+}
+
 class ListWrapper {
   using LP = detail::ListPack;
 
@@ -105,9 +136,18 @@ class ListWrapper {
     QList* ql = CompactObj::AllocateMR<QList>(GetFlag(FLAGS_list_max_listpack_size),
                                               GetFlag(FLAGS_list_compress_depth));
 
+    // Set db index for new QList
+    ql->SetDbIndex(db_id_);
+
     const uint32_t tiering_node_depth_threshold = absl::GetFlag(FLAGS_list_tiering_threshold);
     if (tiering_node_depth_threshold > 0 && EngineShard::tlocal()->tiered_storage()) {
-      ql->EnableTiering(tiering_node_depth_threshold);
+      QList::TieringParams params{
+          .node_depth_threshold = tiering_node_depth_threshold,
+          .offload = OffloadListNode,
+          .load = LoadListNode,
+          .cleanup = CleanupListNode,
+      };
+      ql->EnableTiering(params);
     }
 
     if (uint32_t zstd_thresh = GetFlag(FLAGS_list_experimental_zstd_dict_threshold);