feat(tiering): Add experimental list node offloading

Extend tiered storage to offload individual QList nodes to disk,
gated by --tiered_experimental_list_support.

Key changes:
- Add ListNodeId (dbid, node*, ql*) as a new tiering identity type
  alongside KeyRef; introduce PendingId/ReadId variants so OpManager
  can handle both key-level and node-level I/O uniformly.
- Add tiering callbacks (offload_cb, onload_cb, delete_cb) to
  QList::TieringParams.  PromoteToQLIfNeeded now wires them up when
  tiered storage is available, using the new db_id_ member on
  ListWrapper to store DbIndex for usage.
- Update CancelStash / ReadInternal signatures to accept the generic
  PendingId / ReadId types instead of (dbid, key).
- Add IsClosed() / is_closed_ guard so delete_cb skips I/O after
  TieredStorage::Close() is called.
- Expose StashListNode / ReadTieredListNode free functions used by the
  ListWrapper callbacks.
- Added ListNodeTieringTest suite

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

Author: mkaruza

src/core/compact_object.cc

@@ -815,6 +815,8 @@ CompactObjType CompactObj::ObjType() const {
         return OBJ_STRING;
       case ExternalRep::SERIALIZED_MAP:
         return OBJ_HASH;
+      default:
+        break;
     };
   }
 

src/core/compact_object.h

@@ -171,8 +171,9 @@ class CompactObj {
 
   // Different representations of external values
   enum class ExternalRep : uint8_t {
-    STRING,         // OBJ_STRING, Basic representation with various string encodings
-    SERIALIZED_MAP  // OBJ_HASH, Serialized map
+    STRING,          // OBJ_STRING, Basic representation with various string encodings
+    SERIALIZED_MAP,  // OBJ_HASH, Serialized map
+    LIST_NODE        // OBJ_LIST, QList::Node
   };
 
   explicit CompactObj(bool is_key)

src/core/qlist.cc

@@ -17,6 +17,7 @@ extern "C" {
 
 #include "base/logging.h"
 #include "core/page_usage/page_usage_stats.h"
+#include "core/tiering_types.h"
 
 using namespace std;
 
@@ -48,7 +49,7 @@ namespace dfly {
 namespace {
 
 static_assert(sizeof(QList) == 48);
-static_assert(sizeof(QList::Node) == 40);
+static_assert(sizeof(QList::Node) == 48);
 
 enum IterDir : uint8_t { FWD = 1, REV = 0 };
 
@@ -166,6 +167,7 @@ QList::Node* CreateRAW(int container, uint8_t* entry, size_t sz) {
   node->recompress = 0;
   node->dont_compress = 0;
   node->offloaded = 0;
+  node->io_pending = 0;
 
   return node;
 }
@@ -459,6 +461,14 @@ void QList::Clear() noexcept {
 
   while (len_) {
     Node* next = current->next;
+
+    // Clean up offloaded/pending nodes before freeing.
+    if (current->offloaded || current->io_pending) {
+      if (tiering_params_ && tiering_params_->delete_cb) {
+        tiering_params_->delete_cb(current);
+      }
+    }
+
     if (current->encoding != QUICKLIST_NODE_ENCODING_RAW) {
       quicklistLZF* lzf = (quicklistLZF*)current->entry;
       stats.compressed_bytes -= lzf->sz;
@@ -849,7 +859,7 @@ void QList::CoolOff(Node* node, uint32_t node_id) {
     //    off due to merges (can be improved in future).
     if (node_id >= tiering_params_->node_depth_threshold &&
         node_id + tiering_params_->node_depth_threshold < len_) {
-      if (!node->offloaded) {
+      if (!node->offloaded && !node->io_pending) {
         OffloadNode(node);
       }
     } else if (num_offloaded_nodes_ * 2 + tiering_params_->node_depth_threshold * 2 < len_) {
@@ -865,12 +875,12 @@ void QList::CoolOff(Node* node, uint32_t node_id) {
       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) {
+          if (fw->offloaded == 0 && fw->io_pending == 0) {
             OffloadNode(fw);
           }
 
           // Avoid offloading the same node twice when fw and rev meet in the middle.
-          if (rev != fw && rev->offloaded == 0) {
+          if (rev != fw && rev->offloaded == 0 && rev->io_pending == 0) {
             OffloadNode(rev);
           }
         }
@@ -938,12 +948,28 @@ void QList::CompressByDepth(Node* node) {
 void QList::AccessForReads(bool recompress, Node* node) {
   DCHECK(node);
   stats.total_node_reads++;
+
+  if (node->io_pending) {
+    // Offload in progres so cancel it and read the node synchronously.
+    DCHECK(tiering_params_);
+    if (tiering_params_->delete_cb) {
+      tiering_params_->delete_cb(node);
+    }
+    DCHECK(!node->io_pending);
+  }
+
   if (node->offloaded) {
     DCHECK(tiering_params_);
     stats.onload_requests++;
+    if (tiering_params_->onload_cb) {
+      tiering_params_->onload_cb(node);
+    }
+    // After onload_cb returns, the node entry must be restored.
+    DCHECK(!node->offloaded);
+    DCHECK(node->entry != nullptr);
     num_offloaded_nodes_--;
-    node->offloaded = 0;
   }
+
   if (len_ > 2 && node != head_ && node->next != nullptr) {
     stats.interior_node_reads++;
   }
@@ -1065,8 +1091,16 @@ void QList::DelNode(Node* node) {
   len_--;
   count_ -= node->count;
   malloc_size_ -= node->sz;
-  if (node->offloaded) {
-    num_offloaded_nodes_--;
+
+  if (node->offloaded || node->io_pending) {
+    if (node->offloaded)
+      num_offloaded_nodes_--;
+    else if (node->io_pending)
+      num_offloaded_nodes_--;  // was pre-counted in OffloadNode
+    // Clean up disk segment and deregister fragment.
+    if (tiering_params_ && tiering_params_->delete_cb) {
+      tiering_params_->delete_cb(node);
+    }
   }
 
   /* If we deleted a node within our compress depth, we
@@ -1102,9 +1136,14 @@ bool QList::DelPackedIndex(Node* node, uint8_t* p) {
 
 void QList::OffloadNode(Node* node) {
   DCHECK(tiering_params_ && node->offloaded == 0);
+  if (node->io_pending)
+    return;
+
   num_offloaded_nodes_++;
   stats.offload_requests++;
-  node->offloaded = 1;
+  if (tiering_params_->offload_cb) {
+    tiering_params_->offload_cb(node);
+  }
 }
 
 void QList::InitIteratorEntry(Iterator* it) const {
@@ -1396,4 +1435,13 @@ auto QList::Iterator::Get() const -> Entry {
   return ptr ? Entry(reinterpret_cast<char*>(ptr), sz) : Entry(val);
 }
 
+void QList::Node::SetExternal(size_t offset, uint32_t size) {
+  DCHECK(entry);
+  zfree(entry);
+  offloaded = 1;
+  entry = nullptr;
+  ext.offset = offset;
+  ext.size = size;
+}
+
 }  // namespace dfly

src/core/qlist.h

@@ -7,6 +7,7 @@
 #include <absl/functional/function_ref.h>
 
 #include <cstdint>
+#include <functional>
 #include <memory>
 #include <string>
 
@@ -61,16 +62,27 @@ class QList {
     uint16_t attempted_compress : 1; /* node can't compress; too small */
     uint16_t dont_compress : 1;      /* prevent compression of entry that will be used later */
     uint16_t offloaded : 1;          /* node is offloaded to colder storage */
-    uint16_t reserved1 : 8;          /* reserved for future use */
-
-    uint16_t reserved2; /* more bits to steal for future usage */
-    uint32_t reserved3; /* more bits to steal for future usage */
+    uint16_t io_pending : 1;         /* node has pending io operation (offload or onload) */
+    uint16_t reserved1 : 7;          /* reserved for future use */
 
     bool IsCompressed() const {
       return encoding != QUICKLIST_NODE_ENCODING_RAW;
     }
 
     size_t GetLZF(void** data) const;
+
+    struct __attribute__((__packed__)) ExternalRecord {
+      uint32_t size;
+      size_t offset;
+    };
+    static_assert(sizeof(ExternalRecord) == 12);
+
+    ExternalRecord ext;
+
+    void SetExternal(size_t offset, uint32_t sz);
+    std::pair<size_t, size_t> GetExternalSlice() const {
+      return std::make_pair(size_t(ext.offset), size_t(ext.size));
+    }
   };
 
   using Entry = CollectionEntry;
@@ -100,9 +112,18 @@ class QList {
   using IterateFunc = absl::FunctionRef<bool(Entry)>;
   enum InsertOpt : uint8_t { BEFORE, AFTER };
 
+  void AdjustMallocSize(size_t delta) {
+    malloc_size_ += delta;
+  }
+
   struct TieringParams {
-    // TODO: hook functions and params that allow qlist offloading nodes to colder storage.
     uint32_t node_depth_threshold = 2;
+    // Called when a node should be offloaded to disk.
+    std::function<void(Node*)> offload_cb;
+    // Called when an offloaded node needs its data loaded back from disk.
+    std::function<void(Node*)> onload_cb;
+    // Called when an offloaded or io_pending node is being deleted.
+    std::function<void(Node*)> delete_cb;
   };
 
   /**

src/core/tiering_types.cc

@@ -4,10 +4,44 @@
 
 #include "core/tiering_types.h"
 
+#include <utility>
+
+#include "core/compact_object.h"
+#include "core/overloaded.h"
 #include "redis/redis_aux.h"
 
 namespace dfly::tiering {
 
+bool FragmentRef::IsOffloaded() const {
+  return std::visit(Overloaded{[](CompactValue* pv) { return pv->IsExternal(); },
+                               [](QList::Node* node) { return node->offloaded != 0; }},
+                    val_);
+}
+
+void FragmentRef::ClearOffloaded() {
+  std::visit(Overloaded{[](CompactValue* pv) { pv->RemoveExternal(); },
+                        [](QList::Node* node) { node->offloaded = 0; }},
+             val_);
+}
+
+bool FragmentRef::HasStashPending() const {
+  return std::visit(Overloaded{[](CompactValue* pv) { return pv->HasStashPending(); },
+                               [](QList::Node* node) { return node->io_pending != 0; }},
+                    val_);
+}
+
+void FragmentRef::SetStashPending(bool b) {
+  std::visit(Overloaded{[b](CompactValue* pv) { pv->SetStashPending(b); },
+                        [b](QList::Node* node) { node->io_pending = b ? 1 : 0; }},
+             val_);
+}
+
+CompactObjType FragmentRef::ObjType() const {
+  return std::visit(Overloaded{[](CompactValue* pv) { return pv->ObjType(); },
+                               [](QList::Node*) -> CompactObjType { return OBJ_LIST; }},
+                    val_);
+}
+
 auto FragmentRef::GetDescr(const CompactValue* pv) -> SerializationDescr {
   switch (pv->ObjType()) {
     case OBJ_STRING: {
@@ -27,12 +61,31 @@ auto FragmentRef::GetDescr(const CompactValue* pv) -> SerializationDescr {
   };
 }
 
+auto FragmentRef::GetDescr(const QList::Node* node) -> SerializationDescr {
+  if (!node->entry || node->sz == 0)
+    return {};
+
+  std::string_view entry_view{reinterpret_cast<const char*>(node->entry), node->sz};
+  return {std::array<std::string_view, 2>{entry_view, {}}, CompactObj::ExternalRep::LIST_NODE};
+}
+
+auto FragmentRef::GetSerializationDescr() const -> SerializationDescr {
+  return std::visit(Overloaded{[](CompactValue* pv) { return GetDescr(pv); },
+                               [](QList::Node* node) { return GetDescr(node); }},
+                    val_);
+}
+
 TieredCoolRecord* FragmentRef::GetCoolRecord() const {
-  return std::visit(
-      [](auto* pv) -> TieredCoolRecord* {
-        return pv->IsExternal() && pv->IsCool() ? pv->GetCool().record : nullptr;
-      },
-      val_);
+  return std::visit(Overloaded{[](CompactValue* pv) -> TieredCoolRecord* {
+                                 return pv->IsExternal() && pv->IsCool() ? pv->GetCool().record
+                                                                         : nullptr;
+                               },
+                               [](QList::Node* node) -> TieredCoolRecord* { return nullptr; }},
+                    val_);
+}
+
+std::pair<size_t, size_t> FragmentRef::GetExternalSlice() const {
+  return std::visit([](auto* val) { return val->GetExternalSlice(); }, val_);
 }
 
 }  // namespace dfly::tiering

src/core/tiering_types.h

@@ -7,6 +7,7 @@
 #include <boost/intrusive/list_hook.hpp>
 
 #include "core/compact_object.h"
+#include "core/qlist.h"
 
 namespace dfly::tiering {
 
@@ -39,45 +40,36 @@ class FragmentRef {
   FragmentRef(CompactValue* pv) : val_(pv) {  // NOLINT
   }
 
-  bool IsOffloaded() const {
-    return std::visit([](auto* pv) { return pv->IsExternal(); }, val_);
+  FragmentRef(QList::Node& node) : val_(&node) {  // NOLINT
   }
 
-  // Resets offloaded state for this fragment.
-  void ClearOffloaded() {
-    std::visit([](auto* pv) { pv->RemoveExternal(); }, val_);
+  FragmentRef(QList::Node* node) : val_(node) {  // NOLINT
   }
 
-  bool HasStashPending() const {
-    return std::visit([](auto* pv) { return pv->HasStashPending(); }, val_);
-  }
+  bool IsOffloaded() const;
 
-  void ClearStashPending() {
-    std::visit([](auto* pv) { pv->SetStashPending(false); }, val_);
-  }
+  // Resets offloaded state for this fragment.
+  void ClearOffloaded();
 
-  CompactObjType ObjType() const {
-    return std::visit([](auto* pv) { return pv->ObjType(); }, val_);
-  }
+  bool HasStashPending() const;
+  void SetStashPending(bool b);
+
+  CompactObjType ObjType() const;
 
   // Determine required byte size and encoding type based on value.
-  SerializationDescr GetSerializationDescr() const {
-    return std::visit([](auto* pv) { return GetDescr(pv); }, val_);
-  }
+  SerializationDescr GetSerializationDescr() const;
 
   // Returns a pointer to TieredCoolRecord if this fragment is cool, and null otherwise.
   TieredCoolRecord* GetCoolRecord() const;
 
   // Returns the external slice of the offloaded value. Only valid if IsOffloaded() is true.
-  std::pair<size_t, size_t> GetExternalSlice() const {
-    return std::visit([](auto* pv) { return pv->GetExternalSlice(); }, val_);
-  }
+  std::pair<size_t, size_t> GetExternalSlice() const;
 
  private:
   static SerializationDescr GetDescr(const CompactValue* pv);
+  static SerializationDescr GetDescr(const QList::Node* node);
 
-  // TODO: to support more types, for example Node* from qlist.h.
-  std::variant<CompactValue*> val_;
+  std::variant<CompactValue*, QList::Node*> val_;
 };
 
 }  // namespace dfly::tiering