more than POc

dranikpg · dranikpg · commit 4e183f94b5f6 · 2025-12-02T15:50:42.000+03:00
Signed-off-by: Vladislav Oleshko &lt;vlad@dragonflydb.io&gt;
diff --git a/src/core/compact_object.cc b/src/core/compact_object.cc
@@ -882,13 +882,10 @@ CompactObjType CompactObj::ObjType() const {
     return OBJ_STRING;
 
   if (taglen_ == EXTERNAL_TAG) {
-    VLOG(0) << "My type is external";
     switch (static_cast<ExternalRep>(u_.ext_ptr.representation)) {
       case ExternalRep::STRING:
-        VLOG(0) << "My type is a string";
         return OBJ_STRING;
       case ExternalRep::SERIALIZED_MAP:
-        VLOG(0) << "Mype is a hash map";
         return OBJ_HASH;
     };
   }
diff --git a/src/core/detail/listpack_wrap.cc b/src/core/detail/listpack_wrap.cc
@@ -33,16 +33,15 @@ void ListpackWrap::Iterator::Read() {
   next_ptr_ = lpNext(lp_, next_ptr_);
 }
 
-ListpackWrap::~ListpackWrap() {
-  DCHECK(!dirty_);
+ListpackWrap ListpackWrap::WithCapacity(size_t capacity) {
+  return ListpackWrap{lpNew(capacity)};
 }
 
 ListpackWrap ListpackWrap::WithCapacity(size_t capacity) {
   return ListpackWrap{lpNew(capacity)};
 }
 
 uint8_t* ListpackWrap::GetPointer() {
-  dirty_ = false;
   return lp_;
 }
 
@@ -63,7 +62,6 @@ bool ListpackWrap::Delete(std::string_view key) {
     return false;
 
   lp_ = lpDeleteRangeWithEntry(lp_, &ptr, 2);
-  dirty_ = true;
   return true;
 }
 
@@ -90,7 +88,6 @@ bool ListpackWrap::Insert(std::string_view key, std::string_view value, bool ski
       lp_ = lpReplace(lp_, &vptr, vsrc, value.size());
       DCHECK_EQ(0u, lpLength(lp_) % 2);
 
-      dirty_ = true;
       updated = true;
     }
   }
@@ -100,7 +97,6 @@ bool ListpackWrap::Insert(std::string_view key, std::string_view value, bool ski
     // TODO: we should at least allocate once for both elements
     lp_ = lpAppend(lp_, fsrc, key.size());
     lp_ = lpAppend(lp_, vsrc, value.size());
-    dirty_ = true;
   }
 
   return !updated;
diff --git a/src/core/detail/listpack_wrap.h b/src/core/detail/listpack_wrap.h
@@ -15,8 +15,6 @@ struct ListpackWrap {
   using IntBuf = uint8_t[2][24];
 
  public:
-  ~ListpackWrap();
-
   struct Iterator {
     using iterator_category = std::forward_iterator_tag;
     using difference_type = std::ptrdiff_t;
@@ -66,7 +64,6 @@ struct ListpackWrap {
  private:
   uint8_t* lp_;            // the listpack itself
   mutable IntBuf intbuf_;  // buffer for integers decoded to strings
-  bool dirty_ = false;     // whether lp_ was updated, but never retrieved with GetPointer
 };
 
 }  // namespace dfly::detail
diff --git a/src/server/hset_family.cc b/src/server/hset_family.cc
@@ -75,6 +75,7 @@ struct HMapWrap {
   }
 
  public:
+  // Create from non-external prime value
   HMapWrap(const PrimeValue& pv, DbContext db_cntx) {
     DCHECK(!pv.IsExternal() || pv.IsCool());
     if (pv.Encoding() == kEncodingListPack)
@@ -83,6 +84,9 @@ struct HMapWrap {
       impl_ = GetStringMap(pv, db_cntx);
   }
 
+  explicit HMapWrap(detail::ListpackWrap lw) : impl_{std::move(lw)} {
+  }
+
   explicit HMapWrap(tiering::SerializedMap* sm) : impl_{sm} {
   }
 
@@ -193,7 +197,12 @@ OpResult<T> ExecuteRO(Transaction* tx, F&& f) {
       using D = tiering::SerializedMapDecoder;
       util::fb2::Future<OpResult<T>> fut;
       auto read_cb = [fut, f = std::move(f)](io::Result<D*> res) mutable {
-        HMapWrap hw{res.value()->Get()};
+        // Create wrapper from different types
+        Overloaded ov{
+            [](tiering::SerializedMap* sm) { return HMapWrap{sm}; },
+            [](detail::ListpackWrap* lw) { return HMapWrap{*lw}; },
+        };
+        auto hw = visit(ov, res.value()->Read());
         fut.Resolve(f(hw));
       };
 
@@ -216,15 +225,34 @@ OpResult<T> ExecuteRO(Transaction* tx, F&& f) {
 }
 
 // Wrap write handler
-template <typename F> auto WrapW(F&& f) {
-  using RT = std::invoke_result_t<F, HMapWrap&>;
-  return [f = std::forward<F>(f)](Transaction* t, EngineShard* es) -> RT {
+template <typename F> auto ExecuteW(Transaction* tx, F&& f) {
+  using T = typename std::invoke_result_t<F, HMapWrap&>::Type;
+  auto shard_cb = [f = std::forward<F>(f)](Transaction* t,
+                                           EngineShard* es) -> OpResult<CbVariant<T>> {
+    // Fetch value of hash type
     auto [key, op_args] = KeyAndArgs(t, es);
 
     auto it_res = op_args.GetDbSlice().FindMutable(op_args.db_cntx, key, OBJ_HASH);
     RETURN_ON_BAD_STATUS(it_res);
     auto& pv = it_res->it->second;
 
+    // Enqueue read for future values
+    if (pv.IsExternal() && !pv.IsCool()) {
+      using D = tiering::SerializedMapDecoder;
+      util::fb2::Future<OpResult<T>> fut;
+      auto read_cb = [fut, f = std::move(f)](io::Result<D*> res) mutable {
+        // Create wrapper from different types
+        HMapWrap hw{*res.value()->Write()};
+        fut.Resolve(f(hw));
+
+        // soak listpack wrapper back to get updated value
+        *res.value()->Write() = *hw.Get<detail::ListpackWrap>();
+      };
+
+      es->tiered_storage()->Read(op_args.db_cntx.db_index, key, pv, D{}, std::move(read_cb));
+      return CbVariant<T>{std::move(fut)};
+    }
+
     // Remove document before modification
     op_args.shard->search_indices()->RemoveDoc(key, op_args.db_cntx, pv);
 
@@ -240,8 +268,11 @@ template <typename F> auto WrapW(F&& f) {
     else
       op_args.shard->search_indices()->AddDoc(key, op_args.db_cntx, pv);
 
-    return res;
+    RETURN_ON_BAD_STATUS(res);
+    return CbVariant<T>{std::move(res).value()};
   };
+
+  return Unwrap(tx->ScheduleSingleHopT(std::move(shard_cb)));
 }
 
 size_t EstimateListpackMinBytes(CmdArgList members) {
@@ -391,28 +422,27 @@ OpResult<vector<OptStr>> OpHMGet(const HMapWrap& hw, CmdArgList fields) {
   DCHECK(!fields.empty());
 
   std::vector<OptStr> result(fields.size());
-  if (auto lw = hw.Get<detail::ListpackWrap>(); lw) {
+  if (auto sm = hw.Get<StringMap*>(); sm) {
+    for (size_t i = 0; i < fields.size(); ++i) {
+      if (auto it = (*sm)->Find(fields[i]); it != (*sm)->end()) {
+        result[i].emplace(it->second, sdslen(it->second));
+      }
+    }
+  } else {
     absl::flat_hash_map<string_view, absl::InlinedVector<size_t, 3>> reverse;
     reverse.reserve(fields.size() + 1);
     for (size_t i = 0; i < fields.size(); ++i) {
       reverse[ArgS(fields, i)].push_back(i);  // map fields to their index.
     }
 
-    for (const auto [key, value] : *lw) {
+    for (const auto [key, value] : hw.Range()) {
       if (auto it = reverse.find(key); it != reverse.end()) {
         for (size_t index : it->second) {
           DCHECK_LT(index, result.size());
           result[index].emplace(value);
         }
       }
     }
-  } else {
-    StringMap* sm = *hw.Get<StringMap*>();
-    for (size_t i = 0; i < fields.size(); ++i) {
-      if (auto it = sm->Find(fields[i]); it != sm->end()) {
-        result[i].emplace(it->second, sdslen(it->second));
-      }
-    }
   }
 
   return result;
@@ -424,8 +454,8 @@ struct OpSetParams {
   bool keepttl = false;
 };
 
-OpResult<uint32_t> OpSet(const OpArgs& op_args, string_view key, CmdArgList values,
-                         const OpSetParams& op_sp = OpSetParams{}) {
+OpResult<CbVariant<uint32_t>> OpSet(const OpArgs& op_args, string_view key, CmdArgList values,
+                                    const OpSetParams& op_sp = OpSetParams{}) {
   DCHECK(!values.empty() && 0 == values.size() % 2);
   VLOG(2) << "OpSet(" << key << ")";
 
@@ -438,6 +468,26 @@ OpResult<uint32_t> OpSet(const OpArgs& op_args, string_view key, CmdArgList valu
   auto& it = add_res.it;
   PrimeValue& pv = it->second;
 
+  // If the value is external, enqueue read and modify it there
+  if (pv.IsExternal() && !pv.IsCool()) {
+    CHECK(op_sp.ttl == UINT32_MAX);  // TODO: remove
+    using D = tiering::SerializedMapDecoder;
+    util::fb2::Future<OpResult<uint32_t>> fut;
+    auto read_cb = [fut, values, &op_sp](io::Result<D*> res) mutable {
+      // Create wrapper from different types
+      auto& lw = *res.value()->Write();
+      uint32_t created = 0;
+      for (size_t i = 0; i < values.size(); i += 2) {
+        created += lw.Insert(values[i], values[i + 1], op_sp.skip_if_exists);
+      }
+      fut.Resolve(created);
+    };
+
+    op_args.shard->tiered_storage()->Read(op_args.db_cntx.db_index, key, pv, D{},
+                                          std::move(read_cb));
+    return CbVariant<uint32_t>{std::move(fut)};
+  }
+
   if (add_res.is_new) {
     if (op_sp.ttl == UINT32_MAX) {
       lp = lpNew(0);
@@ -495,7 +545,7 @@ OpResult<uint32_t> OpSet(const OpArgs& op_args, string_view key, CmdArgList valu
   if (auto* ts = op_args.shard->tiered_storage(); ts)
     ts->TryStash(op_args.db_cntx.db_index, key, &pv);
 
-  return created;
+  return CbVariant<uint32_t>{created};
 }
 
 void HGetGeneric(CmdArgList args, uint8_t getall_mask, Transaction* tx, SinkReplyBuilder* builder) {
@@ -587,7 +637,8 @@ void HSetEx(CmdArgList args, const CommandContext& cmd_cntx) {
     return OpSet(t->GetOpArgs(shard), key, fields, op_sp);
   };
 
-  OpResult<uint32_t> result = cmd_cntx.tx->ScheduleSingleHopT(std::move(cb));
+  auto delayed_result = cmd_cntx.tx->ScheduleSingleHopT(std::move(cb));
+  OpResult<uint32_t> result = Unwrap(std::move(delayed_result));
   if (result) {
     cmd_cntx.rb->SendLong(*result);
   } else {
@@ -618,7 +669,7 @@ void HSetFamily::HDel(CmdArgList args, const CommandContext& cmd_cntx) {
       deleted += hw.Erase(s);
     return deleted;
   };
-  HSetReplies{cmd_cntx.rb}.Send(cmd_cntx.tx->ScheduleSingleHopT(WrapW(cb)));
+  HSetReplies{cmd_cntx.rb}.Send(ExecuteW(cmd_cntx.tx, std::move(cb)));
 }
 
 void HSetFamily::HExpire(CmdArgList args, const CommandContext& cmd_cntx) {
@@ -861,7 +912,8 @@ void HSetFamily::HSet(CmdArgList args, const CommandContext& cmd_cntx) {
     return OpSet(t->GetOpArgs(shard), key, args);
   };
 
-  OpResult<uint32_t> result = cmd_cntx.tx->ScheduleSingleHopT(std::move(cb));
+  auto delayed_result = cmd_cntx.tx->ScheduleSingleHopT(std::move(cb));
+  OpResult<uint32_t> result = Unwrap(std::move(delayed_result));
 
   if (result && cmd == "HSET") {
     cmd_cntx.rb->SendLong(*result);
@@ -876,7 +928,7 @@ void HSetFamily::HSetNx(CmdArgList args, const CommandContext& cmd_cntx) {
   auto cb = [&](Transaction* t, EngineShard* shard) {
     return OpSet(t->GetOpArgs(shard), key, args.subspan(1), OpSetParams{.skip_if_exists = true});
   };
-  HSetReplies{cmd_cntx.rb}.Send(cmd_cntx.tx->ScheduleSingleHopT(cb));
+  HSetReplies{cmd_cntx.rb}.Send(Unwrap(cmd_cntx.tx->ScheduleSingleHopT(cb)));
 }
 
 void StrVecEmplaceBack(StringVec& str_vec, const listpackEntry& lp) {
diff --git a/src/server/tiered_storage.cc b/src/server/tiered_storage.cc
@@ -196,7 +196,6 @@ class TieredStorage::ShardOpManager : public tiering::OpManager {
   // Set value to be an in-memory type again. Update memory stats.
   void Upload(DbIndex dbid, string_view value, PrimeValue* pv) {
     DCHECK(!value.empty());
-
     switch (pv->GetExternalRep()) {
       case CompactObj::ExternalRep::STRING:
         pv->Materialize(value, true);
@@ -322,6 +321,7 @@ bool TieredStorage::ShardOpManager::NotifyFetched(EntryId id, tiering::DiskSegme
   auto key = get<OpManager::KeyRef>(id);
   auto* pv = Find(key);
   if (pv && pv->IsExternal() && segment == pv->GetExternalSlice()) {
+    VLOG(0) << "Touched? " << pv->WasTouched();
     if (metrics.modified || pv->WasTouched()) {
       ++stats_.total_uploads;
       decoder->Upload(pv);
diff --git a/src/server/tiering/decoders.cc b/src/server/tiering/decoders.cc
@@ -4,7 +4,11 @@
 
 #include "server/tiering/decoders.h"
 
+<<<<<<< HEAD
 #include "base/logging.h"
+=======
+#include "core/compact_object.h"
+>>>>>>> fbcc8d59 (more than POc)
 #include "core/detail/listpack_wrap.h"
 #include "server/tiering/serialized_map.h"
 
@@ -78,19 +82,40 @@ void SerializedMapDecoder::Initialize(std::string_view slice) {
 }
 
 Decoder::UploadMetrics SerializedMapDecoder::GetMetrics() const {
-  return UploadMetrics{.modified = false,
+  return UploadMetrics{.modified = modified_,
                        .estimated_mem_usage = map_->DataBytes() + map_->size() * 2 * 8};
 }
 
 void SerializedMapDecoder::Upload(CompactObj* obj) {
+  if (std::holds_alternative<std::unique_ptr<SerializedMap>>(map_))
+    MakeOwned();
+
+  obj->InitRobj(OBJ_HASH, kEncodingListPack, Write()->GetPointer());
+}
+
+std::variant<SerializedMap*, detail::ListpackWrap*> SerializedMapDecoder::Read() const {
+  using RT = std::variant<SerializedMap*, detail::ListpackWrap*>;
+  return std::visit([](auto& ptr) -> RT { return ptr.get(); }, map_);
+}
+
+detail::ListpackWrap* SerializedMapDecoder::Write() {
+  if (std::holds_alternative<std::unique_ptr<detail::ListpackWrap>>(map_))
+    return std::get<std::unique_ptr<detail::ListpackWrap>>(map_).get();
+
+  // Convert SerializedMap to listpack
+  MakeOwned();
+  modified_ = true;
+  return Write();
+}
+
+void SerializedMapDecoder::MakeOwned() {
+  auto& map = std::get<std::unique_ptr<SerializedMap>>(map_);
+
   auto lw = detail::ListpackWrap::WithCapacity(GetMetrics().estimated_mem_usage);
-  for (const auto& [key, value] : *map_)
+  for (const auto& [key, value] : *map)
     lw.Insert(key, value, true);
-  obj->InitRobj(OBJ_HASH, kEncodingListPack, lw.GetPointer());
-}
 
-SerializedMap* SerializedMapDecoder::Get() const {
-  return map_.get();
+  map_ = std::make_unique<detail::ListpackWrap>(lw);
 }
 
 }  // namespace dfly::tiering
diff --git a/src/server/tiering/decoders.h b/src/server/tiering/decoders.h
@@ -12,6 +12,10 @@
 #include "core/compact_object.h"
 #include "core/string_or_view.h"
 
+namespace dfly::detail {
+struct ListpackWrap;
+}
+
 namespace dfly::tiering {
 
 struct SerializedMap;
@@ -77,10 +81,14 @@ struct SerializedMapDecoder : public Decoder {
   UploadMetrics GetMetrics() const override;
   void Upload(CompactObj* obj) override;
 
-  SerializedMap* Get() const;
+  std::variant<SerializedMap*, dfly::detail::ListpackWrap*> Read() const;
+  dfly::detail::ListpackWrap* Write();
 
  private:
-  std::unique_ptr<SerializedMap> map_;
+  void MakeOwned();  // Convert to listpack
+
+  bool modified_;
+  std::variant<std::unique_ptr<SerializedMap>, std::unique_ptr<dfly::detail::ListpackWrap>> map_;
 };
 
 }  // namespace dfly::tiering

Original file line number	Diff line number	Diff line change
`@@ -33,16 +33,15 @@ void ListpackWrap::Iterator::Read() {`
`33`	`33`	`next_ptr_ = lpNext(lp_, next_ptr_);`
`34`	`34`	`}`
`35`	`35`
`36`		`-ListpackWrap::~ListpackWrap() {`
`37`		`- DCHECK(!dirty_);`
	`36`	`+ListpackWrap ListpackWrap::WithCapacity(size_t capacity) {`
	`37`	`+ return ListpackWrap{lpNew(capacity)};`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`ListpackWrap ListpackWrap::WithCapacity(size_t capacity) {`
`41`	`41`	`return ListpackWrap{lpNew(capacity)};`
`42`	`42`	`}`
`43`	`43`
`44`	`44`	`uint8_t* ListpackWrap::GetPointer() {`
`45`		`- dirty_ = false;`
`46`	`45`	`return lp_;`
`47`	`46`	`}`
`48`	`47`
`@@ -63,7 +62,6 @@ bool ListpackWrap::Delete(std::string_view key) {`
`63`	`62`	`return false;`
`64`	`63`
`65`	`64`	`lp_ = lpDeleteRangeWithEntry(lp_, &ptr, 2);`
`66`		`- dirty_ = true;`
`67`	`65`	`return true;`
`68`	`66`	`}`
`69`	`67`
`@@ -90,7 +88,6 @@ bool ListpackWrap::Insert(std::string_view key, std::string_view value, bool ski`
`90`	`88`	`lp_ = lpReplace(lp_, &vptr, vsrc, value.size());`
`91`	`89`	`DCHECK_EQ(0u, lpLength(lp_) % 2);`
`92`	`90`
`93`		`- dirty_ = true;`
`94`	`91`	`updated = true;`
`95`	`92`	`}`
`96`	`93`	`}`
`@@ -100,7 +97,6 @@ bool ListpackWrap::Insert(std::string_view key, std::string_view value, bool ski`
`100`	`97`	`// TODO: we should at least allocate once for both elements`
`101`	`98`	`lp_ = lpAppend(lp_, fsrc, key.size());`
`102`	`99`	`lp_ = lpAppend(lp_, vsrc, value.size());`
`103`		`- dirty_ = true;`
`104`	`100`	`}`
`105`	`101`
`106`	`102`	`return !updated;`