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_;
+  
+  //Execute scheduled tasks
+  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,10 @@ class ConsensusCoordinator {
 
   SyncProgress& SyncPros() { return sync_pros_; }
   std::shared_ptr<StableLog> StableLogger() { return stable_logger_; }
-  std::shared_ptr<MemLog> MemLogger() { return mem_logger_; }
+  // Expose the internally maintained mem_logger_ object to the outside world.
+  std::shared_ptr<MemLog> MemLogger() { return mem_logger_; }  
+  // Allows an external user to inject a new StableLog instance into the current object and replace the original one
+  void SetStableLogger(std::shared_ptr<StableLog> logger) { stable_logger_ = logger; }
 
   LogOffset committed_index() {
     std::lock_guard lock(index_mu_);
@@ -186,35 +191,20 @@ 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();
+  // Called after committed_id_ is updated, waking up all threads waiting for the offset
+  void NotifyLogCommitted(const LogOffset& offset);
+  // Block until the specified offset ≤ the current committed_id_, or return false if the timeout expires.
+  bool WaitLogCommitted(const LogOffset& offset, std::chrono::milliseconds timeout);
 
  private:
   pstd::Status TruncateTo(const LogOffset& offset);
 
   pstd::Status InternalAppendLog(const std::shared_ptr<Cmd>& cmd_ptr);
   pstd::Status InternalAppendBinlog(const std::shared_ptr<Cmd>& cmd_ptr);
   void InternalApply(const MemLog::LogItem& log);
+  // Batch processing version
+  void InternalApplyFollower(const std::vector<std::shared_ptr<Cmd>>& cmds);
   void InternalApplyFollower(const std::shared_ptr<Cmd>& cmd_ptr);
 
   pstd::Status GetBinlogOffset(const BinlogOffset& start_offset, LogOffset* log_offset);
@@ -274,21 +264,27 @@ 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;                    // Update in-memory committed_id_
+      context_->UpdateAppliedIndex(committed_id_); // Persist to Context so it survives restart
+      log_commit_cv_.notify_all();              // Wake up any threads waiting in WaitLogCommitted()
+    }
   }
 
  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_;
+  // used to notify that the log has been submitted
+  std::condition_variable_any log_commit_cv_;
+  // Used to track threads waiting for a specific log to be submitted
+  std::mutex waiting_threads_mu_;
 };
 
 #endif  // INCLUDE_PIKA_CONSENSUS_H_

include/pika_db.h

@@ -148,6 +148,8 @@ class DB : public std::enable_shared_from_this<DB>, public pstd::noncopyable {
   bool IsBgSaving();
   BgSaveInfo bgsave_info();
   pstd::Status GetKeyNum(std::vector<storage::KeyInfo>* key_info);
+  // Execute a command synchronously inside DB context (used by consensus batch execution)
+  void ExecuteCmd(const std::shared_ptr<Cmd>& cmd_ptr, int dummy_flag = 0);
 
  private:
   bool opened_ = false;

include/pika_repl_client.h

@@ -86,13 +86,16 @@ 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);
   }
+  // Each asynchronous DB write task is triggered by the worker thread after completion.
+  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"
 
@@ -23,12 +24,13 @@
 #include "include/pika_stable_log.h"
 #include "include/rsync_client.h"
 
-#define kBinlogSendPacketNum 40
-#define kBinlogSendBatchNum 100
+// Batch processing parameters
+#define kBinlogSendPacketNum 500
+#define kBinlogSendBatchNum 1000
 
-// unit seconds
-#define kSendKeepAliveTimeout (2 * 1000000)
-#define kRecvKeepAliveTimeout (20 * 1000000)
+// unit microseconds
+#define kSendKeepAliveTimeout (500 * 1000)  // 500ms发送一次心跳
+#define kRecvKeepAliveTimeout (3 * 1000000)  // 3 seconds receive timeout
 
 
 class SyncDB {
@@ -93,6 +95,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:
@@ -199,6 +203,9 @@ class PikaReplicaManager {
   void DropItemInOneWriteQueue(const std::string& ip, int port, const std::string& db_name);
   void DropItemInWriteQueue(const std::string& ip, int port);
   int ConsumeWriteQueue();
+  
+  // Static callback functions
+  static void ConsumeWriteQueueTask(void* arg);
 
   // Schedule Task
   void ScheduleReplServerBGTask(net::TaskFunc func, void* arg);
@@ -233,7 +240,11 @@ class PikaReplicaManager {
     return pika_repl_client_->GetUnfinishedAsyncWriteDBTaskCount(db_name);
   }
 
- private:
+  void NotifyAsyncWriteComplete(const std::string& db_name);
+  void WaitAsyncWriteComplete(const std::string& db_name);
+  void UpdateAllMasterDBCommittedID();
+  
+private:
   void InitDB();
   pstd::Status SelectLocalIp(const std::string& remote_ip, int remote_port, std::string* local_ip);
 
@@ -247,6 +258,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

include/pika_server.h

@@ -524,6 +524,9 @@ class PikaServer : public pstd::noncopyable {
     }
   }
 
+  // 添加批量处理命令的函数声明
+  void ScheduleWriteDBMulti(const std::vector<std::shared_ptr<Cmd>>& cmds, const std::string& db_name);
+
  private:
   /*
    * TimingTask use

src/pika_auxiliary_thread.cc

@@ -19,34 +19,93 @@ PikaAuxiliaryThread::~PikaAuxiliaryThread() {
   LOG(INFO) << "PikaAuxiliary thread " << thread_id() << " exit!!!";
 }
 
+void PikaAuxiliaryThread::DoTimingTask() {
+  uint64_t now = pstd::NowMicros();
+  
+  static uint64_t last_check_sync_timeout = 0;
+  static uint64_t last_update_committed_id = 0;
+  static uint64_t last_wake_up_binlog_sync = 0;
+  
+  // 降低CheckSyncTimeout频率为每5秒一次
+  if (now - last_check_sync_timeout > 5000000) { // 5秒
+  if (g_pika_server->role() & PIKA_ROLE_MASTER) {
+    g_pika_rm->CheckSyncTimeout(now);
+    }
+    last_check_sync_timeout = now;
+  }
+
+  // 提高UpdateAllMasterDBCommittedID频率为每2ms一次
+  if (now - last_update_committed_id > 2000) { // 2毫秒
+  g_pika_rm->UpdateAllMasterDBCommittedID();
+    last_update_committed_id = now;
+  }
+
+  // 控制WakeUpBinlogSync频率为每100ms一次
+  if (now - last_wake_up_binlog_sync > 100000) { // 100毫秒
+  g_pika_rm->WakeUpBinlogSync();
+    last_wake_up_binlog_sync = now;
+  }
+}
+
 void* PikaAuxiliaryThread::ThreadMain() {
+  // 记录上次同步的时间
+  uint64_t last_sync_time = 0;
+  uint64_t last_update_time = 0;
+  uint64_t last_timing_task_time = 0;
+  uint64_t last_check_time = 0;
+  const uint64_t sync_interval = 3000;  // 3ms，进一步提高同步频率
+  const uint64_t update_interval = 2000;  // 2ms，进一步提高更新频率
+  const uint64_t timing_task_interval = 200000;  // 200ms，进一步减少定时任务执行间隔
+  const uint64_t check_interval = 1000000;  // 1s，减少检查频率
+  
   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());
-    if (!s.ok()) {
+    // 降低CheckSyncTimeout和CheckLeaderProtectedMode调用频率
+    if (now - last_check_time >= check_interval) {
+    pstd::Status s = g_pika_rm->CheckSyncTimeout(now);
+      if (!s.ok() && !s.IsTimeout()) {  // 忽略超时错误的日志记录
       LOG(WARNING) << s.ToString();
+      }
+      g_pika_server->CheckLeaderProtectedMode();
+      last_check_time = now;
     }
 
-    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) {
+      pstd::Status s = g_pika_server->TriggerSendBinlogSync();
+      if (!s.ok() && !s.IsTimeout()) { // 忽略超时错误的日志记录
+        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, 500us); // 减少等待时间到500微秒
     } else {
-      // LOG_EVERY_N(INFO, 1000) << "Consume binlog number " << res;
+      // 处理完成后立即继续下一轮，不等待
+      continue;
     }
   }
   return nullptr;