comments

dragonflydb · Dec 5, 2024 · 88b5116 · 88b5116
1 parent 8ce1463
commit 88b5116
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 5 deletions.
diff --git a/src/server/db_slice.cc b/src/server/db_slice.cc
@@ -1090,7 +1090,6 @@ DbSlice::PrimeItAndExp DbSlice::ExpireIfNeeded(const Context& cntx, PrimeIterato
     LOG(ERROR) << "Invalid call to ExpireIfNeeded";
     return {it, ExpireIterator{}};
   }
-  block_counter_.Wait();
 
   auto& db = db_arr_[cntx.db_index];
 

diff --git a/src/server/dflycmd.cc b/src/server/dflycmd.cc
@@ -369,13 +369,17 @@ void DflyCmd::StartStable(CmdArgList args, Transaction* tx, RedisReplyBuilder* r
 
     auto cb = [this, &status, replica_ptr = replica_ptr](EngineShard* shard) {
       FlowInfo* flow = &replica_ptr->flows[shard->shard_id()];
+
       status = StopFullSyncInThread(flow, &replica_ptr->cntx, shard);
       if (*status != OpStatus::OK) {
         return;
       }
       StartStableSyncInThread(flow, &replica_ptr->cntx, shard);
     };
     shard_set->RunBlockingInParallel(std::move(cb));
+
+    if (*status != OpStatus::OK)
+      return rb->SendError(kInvalidState);
   }
 
   LOG(INFO) << "Transitioned into stable sync with replica " << replica_ptr->address << ":"

diff --git a/src/server/engine_shard.cc b/src/server/engine_shard.cc
@@ -667,8 +667,8 @@ void EngineShard::Heartbeat() {
   if (db_slice.WillBlockOnJournalWrite()) {
     const auto elapsed = std::chrono::system_clock::now() - start;
     if (elapsed > std::chrono::seconds(1)) {
-      LOG(WARNING) << "Stalled heartbeat() fiber for " << elapsed.count()
-                   << " seconds because of big value serialization";
+      LOG_EVERY_T(WARNING, 5) << "Stalled heartbeat() fiber for " << elapsed.count()
+                              << " seconds because of big value serialization";
     }
     return;
   }

diff --git a/src/server/generic_family.cc b/src/server/generic_family.cc
@@ -597,8 +597,9 @@ void OpScan(const OpArgs& op_args, const ScanOpts& scan_opts, uint64_t* cursor,
   auto& db_slice = op_args.GetDbSlice();
   DCHECK(db_slice.IsDbValid(op_args.db_cntx.db_index));
 
-  // We need to make sure we don't preempt below, because we don't hold any locks to the keys
-  // and ExpireIfNeeded requires that.
+  // ScanCb can preempt due to journaling expired entries and we need to make sure that
+  // we enter the callback in a timing when journaling will not cause preemptions. Otherwise,
+  // the bucket might change as we Traverse and yield.
   db_slice.BlockingCounter()->Wait();
 
   util::FiberAtomicGuard guard;