feat (server) : introduce no point in time replication (#5103)

* introduce no point in time replication algorithm, under flag point_in_time_snapshot

Signed-off-by: adi_holden <[email protected]>

Author: adiholden

src/core/dash.h

@@ -268,6 +268,9 @@ class DashTable : public detail::DashTableBase {
   const_bucket_iterator CursorToBucketIt(Cursor c) const {
     return const_bucket_iterator{this, c.segment_id(global_depth_), c.bucket_id(), 0};
   }
+  bucket_iterator CursorToBucketIt(Cursor c) {
+    return bucket_iterator{this, c.segment_id(global_depth_), c.bucket_id(), 0};
+  }
 
   // Capture Version Change. Runs cb(it) on every bucket! (not entry) in the table whose version
   // would potentially change upon insertion of 'k'.
@@ -932,8 +935,8 @@ void DashTable<_Key, _Value, Policy>::Split(uint32_t seg_id, EvictionPolicy& ev)
       std::move(hash_fn), target,
       [&](uint32_t segment_from, detail::PhysicalBid from, uint32_t segment_to,
           detail::PhysicalBid to) {
-        // OnMove is used to notify eviction policy about the moves across buckets/segments
-        // during the split.
+        // OnMove is used to notify eviction policy about the moves across
+        // buckets/segments during the split.
         ev.OnMove(Cursor{global_depth_, segment_from, from}, Cursor{global_depth_, segment_to, to});
       });
 

src/server/db_slice.cc

@@ -99,6 +99,7 @@ class PrimeEvictionPolicy {
     DVLOG(2) << "split: " << segment->SlowSize() << "/" << segment->capacity();
   }
   void OnMove(PrimeTable::Cursor source, PrimeTable::Cursor dest) {
+    moved_items_.push_back(std::make_pair(source, dest));
   }
 
   bool CanGrow(const PrimeTable& tbl) const;
@@ -113,8 +114,12 @@ class PrimeEvictionPolicy {
   unsigned checked() const {
     return checked_;
   }
+  const DbSlice::MovedItemsVec& moved_items() {
+    return moved_items_;
+  }
 
  private:
+  DbSlice::MovedItemsVec moved_items_;
   DbSlice* db_slice_;
   ssize_t mem_offset_;
   ssize_t soft_limit_ = 0;
@@ -364,7 +369,15 @@ class DbSlice::PrimeBumpPolicy {
     return !obj.IsSticky();
   }
   void OnMove(PrimeTable::Cursor source, PrimeTable::Cursor dest) {
+    moved_items_.push_back(std::make_pair(source, dest));
+  }
+
+  const DbSlice::MovedItemsVec& moved_items() {
+    return moved_items_;
   }
+
+ private:
+  DbSlice::MovedItemsVec moved_items_;
 };
 
 DbSlice::DbSlice(uint32_t index, bool cache_mode, EngineShard* owner)
@@ -708,6 +721,7 @@ OpResult<DbSlice::ItAndUpdater> DbSlice::AddOrFindInternal(const Context& cntx,
     events_.insertion_rejections++;
     return OpStatus::OUT_OF_MEMORY;
   }
+  CallMovedCallbacks(cntx.db_index, evp.moved_items());
 
   events_.mutations++;
   ssize_t table_increase = db.prime.mem_usage() - table_before;
@@ -1252,6 +1266,12 @@ uint64_t DbSlice::RegisterOnChange(ChangeCallback cb) {
   return change_cb_.emplace_back(NextVersion(), std::move(cb)).first;
 }
 
+uint64_t DbSlice::RegisterOnMove(MovedCallback cb) {
+  ++next_moved_id_;
+  moved_cb_.emplace_back(next_moved_id_, cb);
+  return next_moved_id_;
+}
+
 void DbSlice::FlushChangeToEarlierCallbacks(DbIndex db_ind, Iterator it, uint64_t upper_bound) {
   unique_lock<LocalLatch> lk(serialization_latch_);
 
@@ -1284,6 +1304,14 @@ void DbSlice::UnregisterOnChange(uint64_t id) {
   change_cb_.erase(it);
 }
 
+void DbSlice::UnregisterOnMoved(uint64_t id) {
+  serialization_latch_.Wait();
+  auto it =
+      find_if(moved_cb_.begin(), moved_cb_.end(), [id](const auto& cb) { return cb.first == id; });
+  CHECK(it != moved_cb_.end());
+  moved_cb_.erase(it);
+}
+
 auto DbSlice::DeleteExpiredStep(const Context& cntx, unsigned count) -> DeleteExpiredStats {
   auto& db = *db_arr_[cntx.db_index];
   DeleteExpiredStats result;
@@ -1754,6 +1782,7 @@ void DbSlice::OnCbFinishBlocking() {
       if (bump_it != it) {  // the item was bumped
         ++events_.bumpups;
       }
+      CallMovedCallbacks(db_index, policy.moved_items());
     }
   }
 
@@ -1777,4 +1806,21 @@ void DbSlice::CallChangeCallbacks(DbIndex id, const ChangeReq& cr) const {
   }
 }
 
+void DbSlice::CallMovedCallbacks(
+    DbIndex id, const std::vector<std::pair<PrimeTable::Cursor, PrimeTable::Cursor>>& moved_items) {
+  if (moved_cb_.empty())
+    return;
+
+  // does not preempt, just increments the counter.
+  unique_lock<LocalLatch> lk(serialization_latch_);
+
+  const size_t limit = moved_cb_.size();
+  auto ccb = moved_cb_.begin();
+  for (size_t i = 0; i < limit; ++i) {
+    CHECK(ccb->second);
+    ccb->second(id, moved_items);
+    ++ccb;
+  }
+}
+
 }  // namespace dfly

src/server/db_slice.h

@@ -434,12 +434,20 @@ class DbSlice {
   }
 
   using ChangeCallback = std::function<void(DbIndex, const ChangeReq&)>;
+  // Holds pairs of source and destination cursors for items moved in the dash table
+  using MovedItemsVec = std::vector<std::pair<PrimeTable::Cursor, PrimeTable::Cursor>>;
+  using MovedCallback = std::function<void(DbIndex, const MovedItemsVec&)>;
 
   //! Registers the callback to be called for each change.
   //! Returns the registration id which is also the unique version of the dbslice
   //! at a time of the call.
   uint64_t RegisterOnChange(ChangeCallback cb);
 
+  //! Registers the callback to be called after items are moved in table.
+  //! Returns the registration id which is also the unique version of the dbslice
+  //! at a time of the call.
+  uint64_t RegisterOnMove(MovedCallback cb);
+
   bool HasRegisteredCallbacks() const {
     return !change_cb_.empty();
   }
@@ -450,6 +458,8 @@ class DbSlice {
   //! Unregisters the callback.
   void UnregisterOnChange(uint64_t id);
 
+  void UnregisterOnMoved(uint64_t id);
+
   struct DeleteExpiredStats {
     uint32_t deleted = 0;         // number of deleted items due to expiry (less than traversed).
     uint32_t deleted_bytes = 0;   // total bytes of deleted items.
@@ -605,6 +615,7 @@ class DbSlice {
   }
 
   void CallChangeCallbacks(DbIndex id, const ChangeReq& cr) const;
+  void CallMovedCallbacks(DbIndex id, const MovedItemsVec& moved_items);
 
   // We need this because registered callbacks might yield and when they do so we want
   // to avoid Heartbeat or Flushing the db.
@@ -620,6 +631,7 @@ class DbSlice {
   bool expire_allowed_ = true;
 
   uint64_t version_ = 1;  // Used to version entries in the PrimeTable.
+  uint64_t next_moved_id_ = 1;
   ssize_t memory_budget_ = SSIZE_MAX / 2;
   size_t bytes_per_object_ = 0;
   size_t soft_budget_limit_ = 0;
@@ -649,6 +661,8 @@ class DbSlice {
   // ordered from the smallest to largest version.
   std::list<std::pair<uint64_t, ChangeCallback>> change_cb_;
 
+  std::list<std::pair<uint32_t, MovedCallback>> moved_cb_;
+
   // Used in temporary computations in Find item and CbFinish
   // This set is used to hold fingerprints of key accessed during the run of
   // a transaction callback (not the whole transaction).
@@ -676,7 +690,8 @@ class DbSlice {
   // and polymorphic allocator (new C++ features)
   // the declarations below meant to say:
   // absl::flat_hash_map<std::string,
-  //                    absl::flat_hash_set<facade::Connection::WeakRef, Hash>> client_tracking_map_
+  //                    absl::flat_hash_set<facade::Connection::WeakRef, Hash>>
+  //                    client_tracking_map_
   using HashSetAllocator = PMR_NS::polymorphic_allocator<facade::Connection::WeakRef>;
 
   using ConnectionHashSet =

src/server/replica.cc

@@ -847,6 +847,13 @@ void DflyShardReplica::FullSyncDflyFb(std::string eof_token, BlockingCounter bc,
     }
   });
 
+  // In the no point-in-time replication flow, it's possible to serialize a journal change
+  // before serializing the bucket that the key was updated in on the master side. As a result,
+  // when loading the serialized bucket data on the replica, it may overwrite the earlier entry
+  // added by the journal change. This is an expected and valid scenario, so to avoid unnecessary
+  // warnings, we enable SetOverrideExistingKeys(true).
+  rdb_loader_->SetOverrideExistingKeys(true);
+
   // Load incoming rdb stream.
   if (std::error_code ec = rdb_loader_->Load(&ps); ec) {
     cntx->ReportError(ec, "Error loading rdb format");

src/server/server_family.cc

@@ -885,6 +885,7 @@ void ServerFamily::Init(util::AcceptServer* acceptor, std::vector<facade::Listen
   config_registry.RegisterMutable("tls_ca_cert_dir");
   config_registry.RegisterMutable("replica_priority");
   config_registry.RegisterMutable("lua_undeclared_keys_shas");
+  config_registry.RegisterMutable("point_in_time_snapshot");
 
   pb_task_ = shard_set->pool()->GetNextProactor();
   if (pb_task_->GetKind() == ProactorBase::EPOLL) {