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 : 1
+
 # 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"
@@ -55,6 +56,8 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
   c_ptr->SetConn(shared_from_this());
   c_ptr->SetResp(resp_ptr);
 
+  LOG(INFO) << "PikaClientConn::DoCmd command: " << c_ptr->name() << ", keys: " << c_ptr->current_key().size();
+
   // Check authed
   if (AuthRequired()) {  // the user is not authed, need to do auth
     if (!(c_ptr->flag() & kCmdFlagsNoAuth)) {
@@ -357,14 +360,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_; }