fix bug with low QPS

Author: YuCai18

include/pika_auxiliary_thread.h

@@ -16,6 +16,9 @@ class PikaAuxiliaryThread : public net::Thread {
   ~PikaAuxiliaryThread() override;
   pstd::Mutex mu_;
   pstd::CondVar cv_;
+  
+  // 执行定时任务
+  void DoTimingTask();
 
  private:
   void* ThreadMain() override;

include/pika_consensus.h

@@ -6,6 +6,8 @@
 #define PIKA_CONSENSUS_H_
 
 #include <utility>
+#include <condition_variable>
+#include <unordered_map>
 
 #include "include/pika_binlog_transverter.h"
 #include "include/pika_client_conn.h"
@@ -170,7 +172,7 @@ class ConsensusCoordinator {
 
   SyncProgress& SyncPros() { return sync_pros_; }
   std::shared_ptr<StableLog> StableLogger() { return stable_logger_; }
-  std::shared_ptr<MemLog> MemLogger() { return mem_logger_; }
+  std::shared_ptr<MemLog> MemLogger() { return mem_logger_; }  void SetStableLogger(std::shared_ptr<StableLog> logger) { stable_logger_ = logger; }
 
   LogOffset committed_index() {
     std::lock_guard lock(index_mu_);
@@ -186,28 +188,9 @@ class ConsensusCoordinator {
   };
   static int InitCmd(net::RedisParser* parser, const net::RedisCmdArgsType& argv);
 
-  std::string ToStringStatus() {
-    std::stringstream tmp_stream;
-    {
-      std::lock_guard lock(index_mu_);
-      tmp_stream << "  Committed_index: " << committed_index_.ToString() << "\r\n";
-    }
-    tmp_stream << "  Context: "
-               << "\r\n"
-               << context_->ToString();
-    {
-      std::shared_lock lock(term_rwlock_);
-      tmp_stream << "  Term: " << term_ << "\r\n";
-    }
-    tmp_stream << "  Mem_logger size: " << mem_logger_->Size() << " last offset "
-               << mem_logger_->last_offset().ToString() << "\r\n";
-    tmp_stream << "  Stable_logger first offset " << stable_logger_->first_offset().ToString() << "\r\n";
-    LogOffset log_status;
-    stable_logger_->Logger()->GetProducerStatus(&(log_status.b_offset.filenum), &(log_status.b_offset.offset),
-                                                &(log_status.l_offset.term), &(log_status.l_offset.index));
-    tmp_stream << "  Physical Binlog Status: " << log_status.ToString() << "\r\n";
-    return tmp_stream.str();
-  }
+  std::string ToStringStatus();
+  void NotifyLogCommitted(const LogOffset& offset);
+  bool WaitLogCommitted(const LogOffset& offset, std::chrono::milliseconds timeout);
 
  private:
   pstd::Status TruncateTo(const LogOffset& offset);
@@ -274,21 +257,28 @@ class ConsensusCoordinator {
   }
   void SetCommittedId(const LogOffset& offset) {
     std::lock_guard l(committed_id_rwlock_);
-    committed_id_ = offset;
-    context_->UpdateAppliedIndex(committed_id_);
+    if (offset > committed_id_) {
+      committed_id_ = offset;
+      context_->UpdateAppliedIndex(committed_id_);
+      // 通知所有等待的线程
+      log_commit_cv_.notify_all();
+    }
   }
 
  private:
   pstd::Status PersistAppendBinlog(const std::shared_ptr<Cmd>& cmd_ptr, LogOffset& cur_offset);
 
- private:
+  std::shared_ptr<Log> logs_;
   std::shared_mutex is_consistency_rwlock_;
   bool is_consistency_ = false;
   std::shared_mutex committed_id_rwlock_;
   LogOffset committed_id_ = LogOffset();
   std::shared_mutex prepared_id__rwlock_;
   LogOffset prepared_id_ = LogOffset();
-  std::shared_ptr<Log> logs_;
+  // 条件变量，用于通知日志已提交
+  std::condition_variable_any log_commit_cv_;
+  // 用于跟踪等待特定日志提交的线程
+  std::mutex waiting_threads_mu_;
 };
 
 #endif  // INCLUDE_PIKA_CONSENSUS_H_

include/pika_repl_client.h

@@ -86,13 +86,18 @@ class PikaReplClient {
     int32_t db_index = db_name.back() - '0';
     assert(db_index >= 0 && db_index <= 7);
     async_write_db_task_counts_[db_index].fetch_sub(incr_step, std::memory_order::memory_order_seq_cst);
+    
+    NotifyAsyncWriteComplete(db_name);
   }
 
   int32_t GetUnfinishedAsyncWriteDBTaskCount(const std::string& db_name) {
     int32_t db_index = db_name.back() - '0';
     assert(db_index >= 0 && db_index <= 7);
     return async_write_db_task_counts_[db_index].load(std::memory_order_seq_cst);
   }
+  
+  // 声明通知方法，实现在cpp文件中
+  void NotifyAsyncWriteComplete(const std::string& db_name);
 
  private:
   size_t GetBinlogWorkerIndexByDBName(const std::string &db_name);

include/pika_rm.h

@@ -12,6 +12,7 @@
 #include <string>
 #include <unordered_map>
 #include <vector>
+#include <condition_variable>
 
 #include "pstd/include/pstd_status.h"
 
@@ -93,6 +94,8 @@ class SyncMasterDB : public SyncDB {
   pstd::Mutex session_mu_;
   int32_t session_id_ = 0;
   ConsensusCoordinator coordinator_;
+  std::mutex cv_mu_;
+  std::condition_variable cv_;
 
   //pacificA public:
  public:
@@ -233,7 +236,14 @@ class PikaReplicaManager {
     return pika_repl_client_->GetUnfinishedAsyncWriteDBTaskCount(db_name);
   }
 
- private:
+  // 添加条件变量相关方法
+  void NotifyAsyncWriteComplete(const std::string& db_name);
+  void WaitAsyncWriteComplete(const std::string& db_name);
+
+  // 在PikaReplicaManager类中添加以下函数声明
+  void UpdateAllMasterDBCommittedID();
+  
+private:
   void InitDB();
   pstd::Status SelectLocalIp(const std::string& remote_ip, int remote_port, std::string* local_ip);
 
@@ -247,6 +257,10 @@ class PikaReplicaManager {
   std::unordered_map<std::string, std::unordered_map<std::string, std::queue<WriteTask>>> write_queues_;
   std::unique_ptr<PikaReplClient> pika_repl_client_;
   std::unique_ptr<PikaReplServer> pika_repl_server_;
+
+  std::mutex async_write_mu_;
+  std::condition_variable async_write_cv_;
+  std::unordered_map<std::string, int> unfinished_async_write_count_;
 };
 
 #endif  //  PIKA_RM_H

src/pika_auxiliary_thread.cc

@@ -19,34 +19,74 @@ PikaAuxiliaryThread::~PikaAuxiliaryThread() {
   LOG(INFO) << "PikaAuxiliary thread " << thread_id() << " exit!!!";
 }
 
+void PikaAuxiliaryThread::DoTimingTask() {
+  uint64_t now = pstd::NowMicros();
+  
+  if (g_pika_server->role() & PIKA_ROLE_MASTER) {
+    g_pika_rm->CheckSyncTimeout(now);
+  }
+
+  // 定期更新主库的CommittedID
+  g_pika_rm->UpdateAllMasterDBCommittedID();
+
+  // wake up binlog sender
+  g_pika_rm->WakeUpBinlogSync();
+}
+
 void* PikaAuxiliaryThread::ThreadMain() {
+  // 记录上次同步的时间
+  uint64_t last_sync_time = 0;
+  uint64_t last_update_time = 0;
+  uint64_t last_timing_task_time = 0;
+  const uint64_t sync_interval = 5000;  // 5ms，大幅提高同步频率
+  const uint64_t update_interval = 5000;  // 5ms，大幅提高更新频率
+  const uint64_t timing_task_interval = 500000;  // 500ms，减少定时任务执行间隔
+  
   while (!should_stop()) {
+    uint64_t now = pstd::NowMicros();
+    
     if (g_pika_server->ShouldMetaSync()) {
       g_pika_rm->SendMetaSyncRequest();
     } else if (g_pika_server->MetaSyncDone()) {
       g_pika_rm->RunSyncSlaveDBStateMachine();
     }
 
-    pstd::Status s = g_pika_rm->CheckSyncTimeout(pstd::NowMicros());
+    pstd::Status s = g_pika_rm->CheckSyncTimeout(now);
     if (!s.ok()) {
       LOG(WARNING) << s.ToString();
     }
 
     g_pika_server->CheckLeaderProtectedMode();
 
-    // TODO(whoiami) timeout
-    s = g_pika_server->TriggerSendBinlogSync();
-    if (!s.ok()) {
-      LOG(WARNING) << s.ToString();
+    // 增加同步频率，减少同步间隔
+    if (now - last_sync_time >= sync_interval) {
+      s = g_pika_server->TriggerSendBinlogSync();
+      if (!s.ok()) {
+        LOG(WARNING) << s.ToString();
+      }
+      last_sync_time = now;
+    }
+    
+    // 单独处理更新committed_id，提高更新频率
+    if (now - last_update_time >= update_interval) {
+      g_pika_rm->UpdateAllMasterDBCommittedID();
+      last_update_time = now;
+    }
+    
+    // 定期执行定时任务
+    if (now - last_timing_task_time >= timing_task_interval) {
+      DoTimingTask();
+      last_timing_task_time = now;
     }
+    
     // send to peer
     int res = g_pika_server->SendToPeer();
     if (res == 0) {
-      // sleep 100 ms
       std::unique_lock lock(mu_);
-      cv_.wait_for(lock, 100ms);
+      cv_.wait_for(lock, 1ms);
     } else {
-      // LOG_EVERY_N(INFO, 1000) << "Consume binlog number " << res;
+      // 处理完成后立即继续下一轮，不等待
+      continue;
     }
   }
   return nullptr;