Fixed multithreading bug

Author: YuCai18

conf/pika.conf

@@ -154,12 +154,12 @@ replication-num : 0
 consensus-level : 0
 
 # consensus-batch-size: The maximum number of items in a consensus batch.
-# Default: 1000
-consensus-batch-size : 1000
+# Default: 100
+consensus-batch-size : 100
 
 # consensus-timeout: The maximum waiting time during the master node batch sending process.
-# Default: 10
-consensus-timeout : 10
+# Default: 5
+consensus-timeout : 5
 
 # replication-ack-timeout: The timeout in milliseconds for waiting for a batch ACK from a slave.
 # Default: 5000

include/pika_consensus.h

@@ -275,13 +275,36 @@ class ConsensusCoordinator {
     prepared_id_ = offset;
   }
   void SetCommittedId(const LogOffset& offset) {
+    // Record desired committed id (from quorum/ACK) first
+    LogOffset fsynced_snapshot;
+    {
+      std::shared_lock fs_lock(fsynced_id_rwlock_);
+      fsynced_snapshot = last_fsynced_id_;
+    }
+    // If desired commit goes beyond fsynced, proactively trigger fsync
+    if (offset > fsynced_snapshot) {
+      std::lock_guard<pstd::Mutex> lk(sync_mu_);
+      needs_sync_.store(true);
+      sync_cv_.notify_one();
+    }
     std::lock_guard l(committed_id_rwlock_);
-    committed_id_ = offset;
+    if (offset > desired_committed_id_) {
+      desired_committed_id_ = offset;
+    }
+    LogOffset target = desired_committed_id_;
+    if (target > fsynced_snapshot) {
+      target = fsynced_snapshot;
+    }
+    if (target > committed_id_) {
+      committed_id_ = target;
     context_->UpdateAppliedIndex(committed_id_);
     committed_id_cv_.notify_all();
+    }
   }
   pstd::Mutex* GetCommittedIdMu() { return &committed_id_mu_; }
   pstd::CondVar* GetCommittedIdCv() { return &committed_id_cv_; }
+  // force next SendBinlog call to send immediately, bypassing coalesce wait
+  void TriggerImmediateSend() { immediate_send_once_.store(true); }
 
  private:
   void SyncBinlogLoop();
@@ -296,17 +319,23 @@ class ConsensusCoordinator {
   std::thread sync_thread_;
   pstd::Mutex promises_mu_;
   std::vector<std::promise<pstd::Status>> sync_promises_;
+  // one-shot switch to force immediate send on next SendBinlog
+  std::atomic<bool> immediate_send_once_{false};
 
   std::shared_mutex is_consistency_rwlock_;
   bool is_consistency_ = false;
   std::shared_mutex committed_id_rwlock_;
   pstd::Mutex committed_id_mu_;
   pstd::CondVar committed_id_cv_;
   LogOffset committed_id_ = LogOffset();
+  LogOffset desired_committed_id_ = LogOffset();
   std::shared_mutex prepared_id__rwlock_;
   LogOffset prepared_id_ = LogOffset();
   std::shared_ptr<Log> logs_;
   int binlog_fsync_counter_ = 0;
+  // Track last fsynced offset to gate commit advancement
+  std::shared_mutex fsynced_id_rwlock_;
+  LogOffset last_fsynced_id_ = LogOffset();
 };
 
 #endif  // INCLUDE_PIKA_CONSENSUS_H_

include/pika_rm.h

@@ -256,6 +256,18 @@ class PikaReplicaManager {
   // client for replica
   std::unique_ptr<PikaReplClient> pika_repl_client_;
   std::unique_ptr<PikaReplServer> pika_repl_server_;
+
+  // one-shot switch to force immediate send on next SendBinlog
+  std::atomic<bool> immediate_send_once_{false};
+ 
+  // consumer thread for write queue
+  std::thread bg_thread_;
+  pstd::CondVar bg_cv_;
+  std::atomic<bool> bg_thread_should_stop_{false};
+
+  std::shared_mutex is_consistency_rwlock_;
+  bool is_consistency_ = false;
+  std::shared_mutex committed_id_rwlock_;
 };
 
 #endif  //  PIKA_RM_H

src/pika_consensus.cc

@@ -390,7 +390,16 @@ Status ConsensusCoordinator::UpdateSlave(const std::string& ip, int port, const
     }
     {
       std::lock_guard l(slave_ptr->slave_mu);
-      slave_ptr->acked_offset = end;
+      // Treat this ACK as confirming everything before and including end
+      LogOffset updated_offset;
+      // Use empty start to indicate from the beginning of the window
+      slave_ptr->sync_win.Update(SyncWinItem(LogOffset()), SyncWinItem(end), &updated_offset);
+      if (!(updated_offset == LogOffset())) {
+        slave_ptr->acked_offset = updated_offset;
+      } else {
+        // Fallback to end if window was empty or no progress detected
+        slave_ptr->acked_offset = end;
+      }
       sync_pros_.AddMatchIndex(ip, port, slave_ptr->acked_offset);
       // LOG(INFO) << "PacificA slave ip: " << ip << ", port :" << port << ",slave acked_offset "
       //           << slave_ptr->acked_offset.ToString();
@@ -835,20 +844,44 @@ bool ConsensusCoordinator::checkFinished(const LogOffset& offset) {
 void ConsensusCoordinator::SyncBinlogLoop() {
   while (!thread_stop_.load()) {
     std::unique_lock<pstd::Mutex> lock(sync_mu_);
-    // timed wait to allow coalescing multiple appends
-    auto coalesce = std::chrono::milliseconds(g_pika_conf->consensus_timeout());
-    sync_cv_.wait_for(lock, coalesce, [this] { return needs_sync_.load() || thread_stop_.load(); });
-
+    // Wait until there is at least one pending append
+    sync_cv_.wait(lock, [this] { return needs_sync_.load() || thread_stop_.load(); });
     if (thread_stop_.load()) {
       break;
     }
-    if (!needs_sync_.load()) {
-      continue;
-    }
+    // Coalesce multiple appends in the next timeout window
+    auto coalesce = std::chrono::milliseconds(g_pika_conf->consensus_timeout());
+    lock.unlock();
+    std::this_thread::sleep_for(coalesce);
+    lock.lock();
 
     needs_sync_.store(false);
     pstd::Status s = stable_logger_->Logger()->Sync();
 
+    // Record fsynced offset (not beyond prepared_id_)
+    {
+      std::shared_lock prep_lock(prepared_id__rwlock_);
+      std::lock_guard fs_lock(fsynced_id_rwlock_);
+      if (prepared_id_ > last_fsynced_id_) {
+        last_fsynced_id_ = prepared_id_;
+      }
+    }
+
+    // After fsync, try to advance committed_id up to min(desired, fsynced)
+    {
+      std::shared_lock fs_lock(fsynced_id_rwlock_);
+      std::lock_guard commit_lock(committed_id_rwlock_);
+      LogOffset target = desired_committed_id_;
+      if (target > last_fsynced_id_) {
+        target = last_fsynced_id_;
+      }
+      if (target > committed_id_) {
+        committed_id_ = target;
+        context_->UpdateAppliedIndex(committed_id_);
+        committed_id_cv_.notify_all();
+      }
+    }
+
     std::lock_guard<pstd::Mutex> guard(promises_mu_);
     for (auto& p : sync_promises_) {
       p.set_value(s);
@@ -920,7 +953,9 @@ Status ConsensusCoordinator::AppendSlaveEntries(const std::shared_ptr<Cmd>& cmd_
                  << " cur last index " << last_index.l_offset.index;
     return Status::OK();
   }
+  auto start_us = pstd::NowMicros();
   Status s = PersistAppendBinlog(cmd_ptr);
+  auto end_us = pstd::NowMicros();
   if (!s.ok()) {
     return s;
   }
@@ -961,7 +996,7 @@ Status ConsensusCoordinator::UpdateCommittedID() {
   LogOffset slave_prepared_id = LogOffset();
 
   for (const auto& slave : slaves) {
-    if (slave.second->slave_state == kSlaveBinlogSync) {
+    if (slave.second->slave_state == kSlaveBinlogSync || slave.second->slave_state == SlaveState::KCandidate) {
       if (slave_prepared_id == LogOffset()) {
         slave_prepared_id = slave.second->acked_offset;
       } else if (slave.second->acked_offset < slave_prepared_id) {
@@ -1029,8 +1064,9 @@ pstd::Status ConsensusCoordinator::SendBinlog(const std::shared_ptr<SlaveNode>&
     return Status::OK();
   }
 
+  // Gate: allow only one in-flight batch until ACK clears the sync window
   int batch_size = g_pika_conf->consensus_batch_size();
-  for (int i = start_index; i < logs_->Size() && tasks.size() < batch_size; ++i) {
+  for (int i = start_index; i < logs_->Size() && static_cast<int>(tasks.size()) < batch_size; ++i) {
     const auto& item = logs_->At(i);
     tasks.emplace_back(RmNode(slave_ptr->Ip(), slave_ptr->Port(), db_name, slave_ptr->SessionId()),
                        BinlogChip(item.offset, item.binlog_), item.offset, committed_index);
@@ -1045,13 +1081,21 @@ pstd::Status ConsensusCoordinator::SendBinlog(const std::shared_ptr<SlaveNode>&
   // decide if we should send now based on size or timeout window
   bool size_triggered = (static_cast<int>(tasks.size()) >= batch_size);
   bool timeout_triggered = false;
-  if (slave_ptr->pending_since_us_ == 0 && !size_triggered) {
+
+  // one-shot immediate send to close current window
+  bool force_now = immediate_send_once_.exchange(false);
+
+  if (slave_ptr->pending_since_us_ == 0 && !size_triggered && !force_now) {
     // start pending window and wait for more logs or timeout
     slave_ptr->pending_since_us_ = now;
     return Status::OK();
   }
-  if (slave_ptr->pending_since_us_ > 0) {
-    timeout_triggered = (now - slave_ptr->pending_since_us_) >= (static_cast<uint64_t>(g_pika_conf->consensus_timeout()) * 1000ULL);
+  if (!size_triggered) {
+    if (force_now) {
+      timeout_triggered = true;
+    } else if (slave_ptr->pending_since_us_ > 0) {
+      timeout_triggered = (now - slave_ptr->pending_since_us_) >= (static_cast<uint64_t>(g_pika_conf->consensus_timeout()) * 1000ULL);
+    }
   }
   if (!size_triggered && !timeout_triggered) {
     return Status::OK();
@@ -1068,14 +1112,21 @@ pstd::Status ConsensusCoordinator::SendBinlog(const std::shared_ptr<SlaveNode>&
   std::vector<WriteTask> final_tasks_to_send;
   final_tasks_to_send.push_back(batched_task);
   g_pika_rm->ProduceWriteQueue(slave_ptr->Ip(), slave_ptr->Port(), db_name, final_tasks_to_send);
+  // Immediately consume the write queue to send over network
+  // g_pika_rm->ConsumeWriteQueue();
 
   // Update slave node's state
   slave_ptr->sent_offset = last_task.binlog_chip_.offset_;
+  // Track every log item so ACK can consume the window in order
   for (const auto& task : tasks) {
-    slave_ptr->sync_win.Push(SyncWinItem(task.binlog_chip_.offset_));
+    slave_ptr->sync_win.Push(SyncWinItem(task.binlog_chip_.offset_, task.binlog_chip_.binlog_.size()));
   }
   // reset pending timer after sending
   slave_ptr->pending_since_us_ = 0;
+  // start ACK timeout tracking for this in-flight batch
+  if (slave_ptr->ack_timeout_start_time_us_ == 0) {
+    slave_ptr->ack_timeout_start_time_us_ = now;
+  }
 
   // trigger fsync coalesced with network send
   {

src/pika_rm.cc

@@ -532,6 +532,7 @@ Status SyncMasterDB::AppendCandidateBinlog(const std::string& ip, int port, cons
 }
 
 pstd::Status SyncMasterDB::SyncBinlogAndWait() {
+  g_pika_rm->WakeUpBinlogSync();
   return coordinator_.SyncAndWait();
 }
 
@@ -540,7 +541,6 @@ Status SyncMasterDB::ConsensusProposeLog(const std::shared_ptr<Cmd>& cmd_ptr) {
     if (!coordinator_.GetISConsistency()) {
         return coordinator_.ProposeLog(cmd_ptr);
     }
-    //LOG(INFO) << "Master DB (" << db_info_.db_name_ << ") ConsensusProposeLog";
 
     // Batch append without immediate waiting to allow high concurrency
     Status s = coordinator_.AppendEntries(cmd_ptr); // Append the log entry to the coordinator
@@ -557,42 +557,63 @@ Status SyncMasterDB::ConsensusProposeLog(const std::shared_ptr<Cmd>& cmd_ptr) {
     //     // TODO: 这里暂时注掉了sleep等待，50ms耗时过长，影响写入链路，后期需要改成条件变量唤醒方式
     //     //std::this_thread::sleep_for(std::chrono::milliseconds(50));
     // }
-    // Batch-wait policy: only wait once per consensus_timeout window or after enough appends
-    static thread_local uint64_t window_start_us = 0;
-    static thread_local int accepted_since_window = 0;
+    // Per-DB global batching window across threads
+    struct WindowState {
+      std::atomic<uint64_t> start_us{0};
+      std::atomic<int> accepted{0};
+    };
+    static std::unordered_map<std::string, WindowState> g_db_windows;
+    static pstd::Mutex g_db_windows_mu;
+
+    WindowState* ws = nullptr;
+    {
+      std::lock_guard<pstd::Mutex> lk(g_db_windows_mu);
+      ws = &g_db_windows[db_info_.db_name_];
+    }
+
     const uint64_t now_us = pstd::NowMicros();
     const uint64_t timeout_us = static_cast<uint64_t>(g_pika_conf->consensus_timeout()) * 1000ULL;
     const int min_batch_wait = std::max(50, g_pika_conf->consensus_batch_size());
 
-    if (window_start_us == 0) {
-      window_start_us = now_us;
-      accepted_since_window = 0;
+    uint64_t expected0 = 0;
+    if (ws->start_us.compare_exchange_strong(expected0, now_us)) {
+      ws->accepted.store(0, std::memory_order_relaxed);
     }
 
-    accepted_since_window++;
-    bool window_elapsed = (now_us - window_start_us) >= timeout_us;
-    bool enough_accumulated = accepted_since_window >= min_batch_wait;
+    ws->accepted.fetch_add(1, std::memory_order_relaxed);
+    uint64_t leader_election_token = ws->start_us.load(std::memory_order_relaxed);
+    bool window_elapsed = (now_us - leader_election_token) >= timeout_us;
+    bool enough_accumulated = ws->accepted.load(std::memory_order_relaxed) >= min_batch_wait;
 
     if (!window_elapsed && !enough_accumulated) {
-      // do not wait this time; let caller return fast to accept more writes
       return Status::OK();
     }
-    // Wait for consensus to be achieved using condition variable (once per window or batch)
+
+    // Attempt to close the current window and become the leader for this batch
+    if (ws->start_us.compare_exchange_strong(leader_election_token, 0)) {
+      // Success, we are the leader. Reset count and trigger send.
+      ws->accepted.store(0, std::memory_order_relaxed);
+      coordinator_.TriggerImmediateSend();
+      g_pika_rm->WakeUpBinlogSync();
+    } else {
+      // Lost the election. Another thread will handle the batch.
+      // My command will be in the *next* batch. So, just return OK.
+      return Status::OK();
+    }
+
+    // Block once per window: wait until committed_id catches prepared_id (end of current window)
     pstd::Mutex* mu = coordinator_.GetCommittedIdMu();
     pstd::CondVar* cv = coordinator_.GetCommittedIdCv();
     std::unique_lock<pstd::Mutex> lock(*mu);
 
     auto timeout = std::chrono::seconds(10);
-    LogOffset offset = coordinator_.GetPreparedId();
-    while (offset > coordinator_.GetCommittedId()) {
-        if (cv->wait_for(lock, timeout) == std::cv_status::timeout) {
-            return Status::Timeout("No consistency achieved within 10 seconds");
-        }
+    LogOffset window_end = coordinator_.GetPreparedId();
+    while (window_end > coordinator_.GetCommittedId()) {
+      if (cv->wait_for(lock, timeout) == std::cv_status::timeout) {
+        return Status::Timeout("No consistency achieved within 10 seconds");
+      }
     }
 
-    // reset window
-    window_start_us = pstd::NowMicros();
-    accepted_since_window = 0;
     return Status::OK();
 }
 
@@ -745,6 +766,7 @@ PikaReplicaManager::PikaReplicaManager() {
   pika_repl_client_ = std::make_unique<PikaReplClient>(3000, 60);
   pika_repl_server_ = std::make_unique<PikaReplServer>(ips, port, 3000);
   InitDB();
+  bg_thread_should_stop_.store(false);
 }
 
 void PikaReplicaManager::Start() {
@@ -760,11 +782,27 @@ void PikaReplicaManager::Start() {
     LOG(FATAL) << "Start Repl Server Error: " << ret
                << (ret == net::kCreateThreadError ? ": create thread error " : ": other error");
   }
+
+  bg_thread_ = std::thread([this]() {
+    while (!bg_thread_should_stop_.load()) {
+      int consumed_count = ConsumeWriteQueue();
+      if (consumed_count == 0) {
+        std::unique_lock<pstd::Mutex> lock(write_queue_mu_);
+        bg_cv_.wait_for(lock, std::chrono::milliseconds(100),
+                        [this] { return bg_thread_should_stop_.load() || !write_queues_.empty(); });
+      }
+    }
+  });
 }
 
 void PikaReplicaManager::Stop() {
   pika_repl_client_->Stop();
   pika_repl_server_->Stop();
+  bg_thread_should_stop_.store(true);
+  bg_cv_.notify_one();
+  if (bg_thread_.joinable()) {
+    bg_thread_.join();
+  }
 }
 
 bool PikaReplicaManager::CheckMasterSyncFinished() {
@@ -800,6 +838,7 @@ void PikaReplicaManager::ProduceWriteQueue(const std::string& ip, int port, std:
   for (auto& task : tasks) {
     write_queues_[index][db_name].push(task);
   }
+  bg_cv_.notify_one();
 }
 
 int PikaReplicaManager::ConsumeWriteQueue() {