feat(server): Redesign and simplify tiered storage module. (dragonflydb#589)

1. Allow offloading blobs larger than 2KB.
2. Totally redesign the offloading algorithm for blobs smaller than 2KB.
3. Fix bugs around IO request cancelations.

Signed-off-by: Roman Gershman <[email protected]>

Signed-off-by: Roman Gershman <[email protected]>

Author: romange

src/core/compact_object.h

@@ -89,8 +89,8 @@ class CompactObj {
 
   enum MaskBit {
     REF_BIT = 1,
-    EXPIRE_BIT = 2,
-    FLAG_BIT = 4,
+    EXPIRE_BIT = 2,  // Mark objects that have expiry timestamp assigned.
+    FLAG_BIT = 4,    // Used to mark keys that have memcache flags assigned.
 
     // ascii encoding is not an injective function. it compresses 8 bytes to 7 but also 7 to 7.
     // therefore, in order to know the original length we introduce 2 flags that
@@ -266,14 +266,15 @@ class CompactObj {
   bool IsExternal() const {
     return taglen_ == EXTERNAL_TAG;
   }
+
   void SetExternal(size_t offset, size_t sz);
   std::pair<size_t, size_t> GetExternalPtr() const;
 
   // In case this object a single blob, returns number of bytes allocated on heap
   // for that blob. Otherwise returns 0.
   size_t MallocUsed() const;
 
-  // Resets the object to empty state.
+  // Resets the object to empty state (string).
   void Reset();
 
   bool IsInline() const {
@@ -346,9 +347,9 @@ class CompactObj {
   //
   static_assert(sizeof(u_) == 16, "");
 
-  // Maybe it's possible to merge those 2 together and gain another byte
-  // but lets postpone it to 2023.
   mutable uint8_t mask_ = 0;
+
+  // We currently reserve 5 bits for tags and 3 bits for extending the mask. currently reserved.
   uint8_t taglen_ = 0;
 };
 
@@ -362,4 +363,45 @@ inline bool CompactObj::operator==(std::string_view sv) const {
   return EqualNonInline(sv);
 }
 
+class CompactObjectView {
+ public:
+  CompactObjectView(const CompactObj& src) : obj_(src.AsRef()) {
+  }
+  CompactObjectView(const CompactObjectView& o) : obj_(o.obj_.AsRef()) {
+  }
+  CompactObjectView(CompactObjectView&& o) = default;
+
+  operator CompactObj() const {
+    return obj_.AsRef();
+  }
+
+  const CompactObj* operator->() const {
+    return &obj_;
+  }
+
+  bool operator==(const CompactObjectView& o) const {
+    return obj_ == o.obj_;
+  }
+
+  uint64_t Hash() const {
+    return obj_.HashCode();
+  }
+
+  CompactObjectView& operator=(const CompactObjectView& o) {
+    obj_ = o.obj_.AsRef();
+    return *this;
+  }
+
+  bool defined() const {
+    return obj_.IsRef();
+  }
+
+  void Reset() {
+    obj_.Reset();
+  }
+
+ private:
+  CompactObj obj_;
+};
+
 }  // namespace dfly

src/core/generate_bin_sizes.py

@@ -10,13 +10,12 @@
 
 def print_small_bins():
     prev_val = 0
-    for i in range(64, 1, -1):
-        val = 4096 // i
-        val = (val // 16)*16  # make it 16 bytes aligned
-        if val != prev_val:
-            print(val, end=', ')
-            prev_val = val
-
+    for i in range(56, 1, -1):
+        len = (4096 - i*8)  # reduce by size of hashes
+        len = (len // 8)*8  # make it 8 bytes aligned
+        if len != prev_val:
+            print(i, len)
+            prev_val = len
     print()
 
 

src/server/common.cc

@@ -193,12 +193,13 @@ bool ParseDouble(string_view src, double* value) {
 #define ADD(x) (x) += o.x
 
 TieredStats& TieredStats::operator+=(const TieredStats& o) {
-  static_assert(sizeof(TieredStats) == 32);
+  static_assert(sizeof(TieredStats) == 40);
 
-  ADD(external_reads);
-  ADD(external_writes);
+  ADD(tiered_reads);
+  ADD(tiered_writes);
   ADD(storage_capacity);
   ADD(storage_reserved);
+  ADD(aborted_offloads);
   return *this;
 }
 

src/server/common.h

@@ -91,13 +91,14 @@ struct OpArgs {
 };
 
 struct TieredStats {
-  size_t external_reads = 0;
-  size_t external_writes = 0;
+  size_t tiered_reads = 0;
+  size_t tiered_writes = 0;
 
   size_t storage_capacity = 0;
 
   // how much was reserved by actively stored items.
   size_t storage_reserved = 0;
+  size_t aborted_offloads = 0;
 
   TieredStats& operator+=(const TieredStats&);
 };

src/server/db_slice.cc

@@ -703,8 +703,11 @@ void DbSlice::PreUpdate(DbIndex db_ind, PrimeIterator it) {
       tiered->Free(offset, size);
       it->second.Reset();
 
-      stats->external_entries -= 1;
-      stats->external_size -= size;
+      stats->tiered_entries -= 1;
+      stats->tiered_size -= size;
+    } else if (it->second.HasIoPending()) {
+      TieredStorage* tiered = shard_owner()->tiered_storage();
+      tiered->CancelIo(db_ind, it);
     }
   }
 

src/server/server_family.cc

@@ -1313,12 +1313,13 @@ void ServerFamily::Info(CmdArgList args, ConnectionContext* cntx) {
 
   if (should_enter("TIERED", true)) {
     ADD_HEADER("# TIERED");
-    append("external_entries", total.external_entries);
-    append("external_bytes", total.external_size);
-    append("external_reads", m.tiered_stats.external_reads);
-    append("external_writes", m.tiered_stats.external_writes);
-    append("external_reserved", m.tiered_stats.storage_reserved);
-    append("external_capacity", m.tiered_stats.storage_capacity);
+    append("tiered_entries", total.tiered_entries);
+    append("tiered_bytes", total.tiered_size);
+    append("tiered_reads", m.tiered_stats.tiered_reads);
+    append("tiered_writes", m.tiered_stats.tiered_writes);
+    append("tiered_reserved", m.tiered_stats.storage_reserved);
+    append("tiered_capacity", m.tiered_stats.storage_capacity);
+    append("tiered_aborted_writes", m.tiered_stats.aborted_offloads);
   }
 
   if (should_enter("PERSISTENCE", true)) {

src/server/string_family.cc

@@ -413,7 +413,7 @@ OpStatus SetCmd::Set(const SetParams& params, string_view key, string_view value
       TieredStorage::EligibleForOffload(value)) {  // external storage enabled.
     // TODO: we may have a bug if we block the fiber inside UnloadItem - "it" may be invalid
     // afterwards.
-    shard->tiered_storage()->UnloadItem(op_args_.db_cntx.db_index, it);
+    shard->tiered_storage()->ScheduleOffload(op_args_.db_cntx.db_index, it);
   }
 
   return OpStatus::OK;
@@ -458,14 +458,14 @@ OpStatus SetCmd::SetExisting(const SetParams& params, PrimeIterator it, ExpireIt
 
   // overwrite existing entry.
   prime_value.SetString(value);
+  DCHECK(!prime_value.HasIoPending());
 
   if (value.size() >= kMinTieredLen) {  // external storage enabled.
-
     // TODO: if UnloadItem can block the calling fiber, then we have the bug because then "it"
     // can be invalid after the function returns and the functions that follow may access invalid
     // entry.
     if (shard->tiered_storage()) {
-      shard->tiered_storage()->UnloadItem(op_args_.db_cntx.db_index, it);
+      shard->tiered_storage()->ScheduleOffload(op_args_.db_cntx.db_index, it);
     }
   }
 

src/server/table.cc

@@ -23,16 +23,15 @@ DbTableStats& DbTableStats::operator+=(const DbTableStats& o) {
   ADD(update_value_amount);
   ADD(listpack_blob_cnt);
   ADD(listpack_bytes);
-  ADD(external_entries);
-  ADD(external_size);
+  ADD(tiered_entries);
+  ADD(tiered_size);
 
   return *this;
 }
 
 DbTable::DbTable(std::pmr::memory_resource* mr)
     : prime(kInitSegmentLog, detail::PrimeTablePolicy{}, mr),
-      expire(0, detail::ExpireTablePolicy{}, mr),
-      mcflag(0, detail::ExpireTablePolicy{}, mr) {
+      expire(0, detail::ExpireTablePolicy{}, mr), mcflag(0, detail::ExpireTablePolicy{}, mr) {
 }
 
 DbTable::~DbTable() {

src/server/table.h

@@ -48,8 +48,8 @@ struct DbTableStats {
   ssize_t update_value_amount = 0;
   size_t listpack_blob_cnt = 0;
   size_t listpack_bytes = 0;
-  size_t external_entries = 0;
-  size_t external_size = 0;
+  size_t tiered_entries = 0;
+  size_t tiered_size = 0;
 
   DbTableStats& operator+=(const DbTableStats& o);
 };