Fix the low qps bug

Author: YuCai18

conf/pika.conf

@@ -153,13 +153,6 @@ replication-num : 0
 # The default value of consensus-level is 0, which means this feature is not enabled.
 consensus-level : 0
 
-# consensus-batch-size: The maximum number of items in a consensus batch.
-consensus-batch-size : 1000
-
-# consensus-timeout: The timeout in milliseconds for waiting for a batch ACK from a slave.
-# Default: 1500
-consensus-timeout : 1500
-
 # The Prefix of dump file's name.
 # All the files that generated by command "bgsave" will be name with this prefix.
 dump-prefix :
@@ -320,6 +313,11 @@ write-binlog : yes
 # Supported Units [K|M|G], binlog-file-size default unit is in [bytes] and the default value is 100M.
 binlog-file-size : 104857600
 
+# The interval (in number of logs) for forcing a disk flush of the binlog.
+# A value of 1 means fsync for every log. A higher value improves performance at the cost of durability.
+# Default: 100
+binlog-fsync-interval : 100000
+
 # Automatically triggers a small compaction according to statistics
 # Use the cache to store up to 'max-cache-statistic-keys' keys
 # If 'max-cache-statistic-keys' set to '0', that means turn off the statistics function

include/pika_client_conn.h

@@ -8,6 +8,7 @@
 
 #include <bitset>
 #include <utility>
+#include <future>
 
 #include "acl.h"
 #include "include/pika_command.h"
@@ -52,6 +53,13 @@ class PikaClientConn : public net::RedisConn {
     bool cache_miss_in_rtc_;
   };
 
+  struct ParallelTask {
+    std::vector<std::shared_ptr<std::string>> resps;
+    std::vector<std::future<void>> futures;
+    std::atomic<size_t> completed_count{0};
+    size_t total_tasks{0};
+  };
+
   struct TxnStateBitMask {
    public:
     static constexpr uint8_t Start = 0;
@@ -72,6 +80,7 @@ class PikaClientConn : public net::RedisConn {
   void BatchExecRedisCmd(const std::vector<net::RedisCmdArgsType>& argvs, bool cache_miss_in_rtc);
   int DealMessage(const net::RedisCmdArgsType& argv, std::string* response) override { return 0; }
   static void DoBackgroundTask(void* arg);
+  static void ParallelExecRedisCmd(void* arg);
 
   bool IsPubSub() { return is_pubsub_; }
   void SetIsPubSub(bool is_pubsub) { is_pubsub_ = is_pubsub; }

include/pika_conf.h

@@ -749,6 +749,8 @@ class PikaConf : public pstd::BaseConf {
     rsync_timeout_ms_.store(value);
   }
 
+  int binlog_fsync_interval() const;
+
   void SetProtoMaxBulkLen(const int64_t value) {
     std::lock_guard l(rwlock_);
     TryPushDiffCommands("proto-max-bulk-len", std::to_string(value));
@@ -888,9 +890,6 @@ class PikaConf : public pstd::BaseConf {
   int ConfigRewriteSlaveOf();
   int ConfigRewriteReplicationID();
 
-  int consensus_batch_size() const;
-  int consensus_timeout() const;
-
  private:
   int port_ = 0;
   int slave_priority_ = 100;
@@ -1069,9 +1068,8 @@ class PikaConf : public pstd::BaseConf {
   //Internal used metrics Persisted by pika.conf
   std::unordered_set<std::string> internal_used_unfinished_full_sync_;
 
-  // Consensus configuration
-  int consensus_batch_size_;
-  int consensus_timeout_;
+  // Binlog fsync interval
+  int binlog_fsync_interval_;
 };
 
 #endif

include/pika_consensus.h

@@ -294,6 +294,7 @@ class ConsensusCoordinator {
   std::shared_mutex prepared_id__rwlock_;
   LogOffset prepared_id_ = LogOffset();
   std::shared_ptr<Log> logs_;
+  int binlog_fsync_counter_ = 0;
 };
 
 #endif  // INCLUDE_PIKA_CONSENSUS_H_

include/pika_define.h

@@ -354,14 +354,6 @@ const int64_t kPoolSize = 1073741824;
 const std::string kBinlogPrefix = "write2file";
 const size_t kBinlogPrefixLen = 10;
 
-/* 
- * PIKA_BATCH_MAGIC: Core identifier for binlog batch processing.
- * - Master: Prefixes batched binlogs with this magic in SendBinlog
- * - Slave: Detects this magic in HandleBGWorkerWriteBinlog
- *   to switch between batch and single-binlog parsing modes.
- */
-const uint32_t PIKA_BATCH_MAGIC = 0x42544348; // "BTCH" in ASCII
-
 const std::string kPikaMeta = "meta";
 const std::string kManifest = "manifest";
 const std::string kContext = "context";

include/pika_repl_bgworker.h

@@ -42,6 +42,7 @@ class PikaReplBgWorker {
   net::RedisParser redis_parser_;
   std::string ip_port_;
   std::string db_name_;
+  int binlog_fsync_counter_ = 0;
 
  private:
   net::BGThread bg_thread_;

include/pika_rm.h

@@ -245,8 +245,15 @@ class PikaReplicaManager {
 
   pstd::Mutex write_queue_mu_;
 
+  // db_name -> a queue of write task
+  using DBWriteTaskQueue = std::map<std::string, std::queue<WriteTask>>;
+  // ip:port -> a map of DBWriteTaskQueue
+  using SlaveWriteTaskQueue = std::map<std::string, DBWriteTaskQueue>;
+
   // every host owns a queue, the key is "ip + port"
-  std::unordered_map<std::string, std::unordered_map<std::string, std::queue<std::pair<WriteTask, uint64_t>>>> write_queues_;
+  SlaveWriteTaskQueue write_queues_;
+
+  // client for replica
   std::unique_ptr<PikaReplClient> pika_repl_client_;
   std::unique_ptr<PikaReplServer> pika_repl_server_;
 };

src/pika_client_conn.cc

@@ -19,6 +19,7 @@
 #include "net/src/dispatch_thread.h"
 #include "net/src/worker_thread.h"
 #include "src/pstd/include/scope_record_lock.h"
+#include <future>
 
 #include "rocksdb/perf_context.h"
 #include "rocksdb/iostats_context.h"
@@ -357,14 +358,64 @@ void PikaClientConn::DoBackgroundTask(void* arg) {
 }
 
 void PikaClientConn::BatchExecRedisCmd(const std::vector<net::RedisCmdArgsType>& argvs, bool cache_miss_in_rtc) {
-  resp_num.store(static_cast<int32_t>(argvs.size()));
-  for (const auto& argv : argvs) {
-    std::shared_ptr<std::string> resp_ptr = std::make_shared<std::string>();
-    resp_array.push_back(resp_ptr);
-    ExecRedisCmd(argv, resp_ptr, cache_miss_in_rtc);
+  if (argvs.empty()) {
+    return;
   }
-  time_stat_->process_done_ts_ = pstd::NowMicros();
-  TryWriteResp();
+  if (argvs.size() > 1) {
+    auto task = std::make_shared<ParallelTask>();
+    task->total_tasks = argvs.size();
+    task->resps.resize(argvs.size());
+
+    for (size_t i = 0; i < argvs.size(); ++i) {
+      task->resps[i] = std::make_shared<std::string>();
+      std::promise<void> promise;
+      task->futures.push_back(promise.get_future());
+      
+      g_pika_server->ScheduleClientPool(&PikaClientConn::ParallelExecRedisCmd, new std::tuple(shared_from_this(), argvs[i], task, i, std::move(promise), cache_miss_in_rtc), false, false);
+    }
+
+    for (auto& f : task->futures) {
+      f.get();
+    }
+
+    for (const auto& resp : task->resps) {
+      WriteResp(*resp);
+    }
+    if (write_completed_cb_) {
+      write_completed_cb_();
+      write_completed_cb_ = nullptr;
+    }
+    NotifyEpoll(true);
+  } else {
+    resp_num.store(static_cast<int32_t>(argvs.size()));
+    for (const auto& argv : argvs) {
+      std::shared_ptr<std::string> resp_ptr = std::make_shared<std::string>();
+      resp_array.push_back(resp_ptr);
+      ExecRedisCmd(argv, resp_ptr, cache_miss_in_rtc);
+    }
+    time_stat_->process_done_ts_ = pstd::NowMicros();
+    TryWriteResp();
+  }
+}
+
+void PikaClientConn::ParallelExecRedisCmd(void* arg) {
+    auto* task_args = static_cast<std::tuple<std::shared_ptr<PikaClientConn>, net::RedisCmdArgsType, std::shared_ptr<ParallelTask>, size_t, std::promise<void>, bool>*>(arg);
+    auto [conn, argv, task, index, promise, cache_miss_in_rtc] = std::move(*task_args);
+    delete task_args;
+
+    std::string opt = argv[0];
+    pstd::StringToLower(opt);
+    if (opt == kClusterPrefix) {
+        if (argv.size() >= 2) {
+            opt += argv[1];
+            pstd::StringToLower(opt);
+        }
+    }
+
+    std::shared_ptr<Cmd> cmd_ptr = conn->DoCmd(argv, opt, task->resps[index], cache_miss_in_rtc);
+    *(task->resps[index]) = std::move(cmd_ptr->res().message());
+
+    promise.set_value();
 }
 
 bool PikaClientConn::ReadCmdInCache(const net::RedisCmdArgsType& argv, const std::string& opt) {

src/pika_conf.cc

@@ -528,6 +528,10 @@ int PikaConf::Load() {
   if (binlog_file_size_ < 1024 || static_cast<int64_t>(binlog_file_size_) > (1024LL * 1024 * 1024)) {
     binlog_file_size_ = 100 * 1024 * 1024;  // 100M
   }
+  GetConfInt("binlog-fsync-interval", &binlog_fsync_interval_);
+  if (binlog_fsync_interval_ < 0) {
+    binlog_fsync_interval_ = 0;
+  }
   GetConfStr("pidfile", &pidfile_);
 
   // db sync
@@ -707,11 +711,6 @@ int PikaConf::Load() {
     rsync_timeout_ms_.store(tmp_rsync_timeout_ms);
   }
 
-  consensus_batch_size_ = 100;
-  consensus_timeout_ = 1000; // 1s
-
-  GetConfInt("consensus-batch-size", &consensus_batch_size_);
-  GetConfInt("consensus-timeout", &consensus_timeout_);
   return ret;
 }
 
@@ -918,6 +917,4 @@ std::vector<rocksdb::CompressionType> PikaConf::compression_per_level() {
   return types;
 }
 
-int PikaConf::consensus_batch_size() const { return consensus_batch_size_; }
-
-int PikaConf::consensus_timeout() const { return consensus_timeout_; }
+int PikaConf::binlog_fsync_interval() const { return binlog_fsync_interval_; }

src/pika_consensus.cc

@@ -842,11 +842,12 @@ Status ConsensusCoordinator::PersistAppendBinlog(const std::shared_ptr<Cmd>& cmd
     return s;
   }
 
-  // Force flush to ensure data persistence on master
-  s = stable_logger_->Logger()->Sync();
-  if (!s.ok()) {
-    LOG(WARNING) << "Failed to sync binlog to disk on master: " << s.ToString();
-    return s;
+  if (++binlog_fsync_counter_ % g_pika_conf->binlog_fsync_interval() == 0) {
+    s = stable_logger_->Logger()->Sync();
+    if (!s.ok()) {
+      LOG(WARNING) << "Failed to sync binlog to disk on master: " << s.ToString();
+      return s;
+    }
   }
 
   // If successful, append the log entry to the logs
@@ -998,33 +999,39 @@ Status ConsensusCoordinator::ApplyBinlog(const std::vector<Log::LogItem>& logs)
 
 Status ConsensusCoordinator::SendBinlog(std::shared_ptr<SlaveNode> slave_ptr, std::string db_name) {
   std::vector<WriteTask> tasks;
-  LogOffset prev_offset = slave_ptr->sent_offset;
-  LOG(INFO) << "SendBinlog: logs_->LastOffset()=" << logs_->LastOffset().ToString()
-            << ", slave_ptr->sent_offset=" << slave_ptr->sent_offset.ToString();
-  // Check if there are new log entries that need to be sent to the slave
-  if (logs_->LastOffset() >= slave_ptr->acked_offset) {
-    LOG(INFO) << "SendBinlog: logs_->Size()=" << logs_->Size();
-    // Find the index of the log entry corresponding to the slave's acknowledged offset
-    int index = logs_->FindOffset(slave_ptr->acked_offset);
-    LOG(INFO) << "SendBinlog: index=" << index;
-    if (index < logs_->Size()) {
-      for (int i = index; i < logs_->Size(); ++i) {
-        const Log::LogItem& item = logs_->At(i);
-
-        slave_ptr->SetLastSendTime(pstd::NowMicros());
-
-        RmNode rm_node(slave_ptr->Ip(), slave_ptr->Port(), slave_ptr->DBName(), slave_ptr->SessionId());
-        WriteTask task(rm_node, BinlogChip(item.offset, item.binlog_), prev_offset, GetCommittedId());
-        tasks.emplace_back(std::move(task));
-        
-        prev_offset = item.offset;
-        slave_ptr->sent_offset = item.offset;
-      }
+  if (!g_pika_server->IsConsistency()) {
+    return Status::OK();
+  }
+  std::string ip = slave_ptr->Ip();
+  int port = slave_ptr->Port();
+  int32_t session_id = slave_ptr->SessionId();
+
+  LogOffset last_sent = slave_ptr->sent_offset;
+  if (logs_->LastOffset() > last_sent) {
+    int send_start_index = logs_->FindOffset(last_sent);
+    if (send_start_index < 0) {
+      LOG(WARNING) << "Binlog offset not found, maybe purged. last_sent: " << last_sent.ToString();
+      return Status::Corruption("cant find the file_num");
+    }
+
+    if (send_start_index < logs_->Size() && logs_->At(send_start_index).offset == last_sent) {
+      send_start_index++;
+    }
+
+    if (send_start_index < logs_->Size()) {
+      LogOffset prev_offset = send_start_index > 0 ? logs_->At(send_start_index - 1).offset : LogOffset();
+      const auto& item = logs_->At(send_start_index);
+      RmNode rm_node(ip, port, db_name, session_id);
+      WriteTask task(rm_node, BinlogChip(item.offset, item.binlog_), prev_offset, GetCommittedId());
+      tasks.emplace_back(std::move(task));
     }
   }
 
   if (!tasks.empty()) {
-    g_pika_rm->ProduceWriteQueue(slave_ptr->Ip(), slave_ptr->Port(), db_name, tasks);
+    g_pika_rm->ProduceWriteQueue(ip, port, db_name, tasks);
+    slave_ptr->sent_offset = tasks.back().binlog_chip_.offset_;
+    LOG(INFO) << "SendBinlog tasks to queue, slave: " << ip << ":" << port << " tasks: " << tasks.size()
+              << " new sent_offset: " << slave_ptr->sent_offset.ToString();
   }
   return Status::OK();
 }