Change to parallel execution

Author: YuCai18

conf/pika.conf

@@ -154,11 +154,16 @@ replication-num : 0
 consensus-level : 0
 
 # consensus-batch-size: The maximum number of items in a consensus batch.
+# Default: 1000
 consensus-batch-size : 1000
 
-# consensus-timeout: The timeout in milliseconds for waiting for a batch ACK from a slave.
-# Default: 1500
-consensus-timeout : 1500
+# consensus-timeout: The maximum waiting time during the master node batch sending process.
+# Default: 10
+consensus-timeout : 10
+
+# replication-ack-timeout: The timeout in milliseconds for waiting for a batch ACK from a slave.
+# Default: 5000
+replication-ack-timeout : 5000
 
 # The Prefix of dump file's name.
 # All the files that generated by command "bgsave" will be name with this prefix.
@@ -320,6 +325,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 : 100
+
 # 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

@@ -350,6 +350,18 @@ class PikaConf : public pstd::BaseConf {
   int max_conn_rbuf_size() { return max_conn_rbuf_size_.load(); }
   int consensus_level() { return consensus_level_.load(); }
   int replication_num() { return replication_num_.load(); }
+  int consensus_batch_size() {
+    std::shared_lock l(rwlock_);
+    return consensus_batch_size_;
+  }
+  int consensus_timeout() {
+    std::shared_lock l(rwlock_);
+    return consensus_timeout_;
+  }
+  int replication_ack_timeout() {
+    std::shared_lock l(rwlock_);
+    return replication_ack_timeout_;
+  }
   int rate_limiter_mode() {
     std::shared_lock l(rwlock_);
     return rate_limiter_mode_;
@@ -665,6 +677,21 @@ class PikaConf : public pstd::BaseConf {
     TryPushDiffCommands("max-conn-rbuf-size", std::to_string(value));
     max_conn_rbuf_size_.store(value);
   }
+  void SetConsensusBatchSize(const int value) {
+    std::lock_guard l(rwlock_);
+    TryPushDiffCommands("consensus-batch-size", std::to_string(value));
+    consensus_batch_size_ = value;
+  }
+  void SetConsensusTimeout(const int value) {
+    std::lock_guard l(rwlock_);
+    TryPushDiffCommands("consensus-timeout", std::to_string(value));
+    consensus_timeout_ = value;
+  }
+  void SetReplicationAckTimeout(const int value) {
+    std::lock_guard l(rwlock_);
+    TryPushDiffCommands("replication-ack-timeout", std::to_string(value));
+    replication_ack_timeout_ = value;
+  }
   void SetMaxCacheFiles(const int& value) {
     std::lock_guard l(rwlock_);
     TryPushDiffCommands("max-cache-files", std::to_string(value));
@@ -748,6 +775,12 @@ class PikaConf : public pstd::BaseConf {
     TryPushDiffCommands("rsync-timeout-ms", std::to_string(value));
     rsync_timeout_ms_.store(value);
   }
+ 
+  void SetBinlogFsyncSize(const int value) {
+    std::lock_guard l(rwlock_);
+    TryPushDiffCommands("binlog-fsync-size", std::to_string(value));
+    binlog_fsync_size_ = value;
+  }
 
   void SetProtoMaxBulkLen(const int64_t value) {
     std::lock_guard l(rwlock_);
@@ -888,9 +921,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;
@@ -982,11 +1012,14 @@ class PikaConf : public pstd::BaseConf {
   int64_t rate_limiter_refill_period_us_ = 0;
   int64_t rate_limiter_fairness_ = 0;
   bool rate_limiter_auto_tuned_ = true;
+  int replication_ack_timeout_ = 5000;
 
   std::atomic<int> sync_window_size_;
   std::atomic<int> max_conn_rbuf_size_;
   std::atomic<int> consensus_level_;
   std::atomic<int> replication_num_;
+  int consensus_batch_size_ = 1000;
+  int consensus_timeout_ = 1500;
 
   std::string network_interface_;
 
@@ -1050,6 +1083,7 @@ class PikaConf : public pstd::BaseConf {
   int throttle_bytes_per_second_ = 200 << 20; // 200MB/s
   int max_rsync_parallel_num_ = kMaxRsyncParallelNum;
   std::atomic_int64_t rsync_timeout_ms_ = 1000;
+  int binlog_fsync_size_ = 8 * 1024 * 1024;
 
   /*
   kUninitialized = 0,             // unknown setting
@@ -1068,10 +1102,6 @@ 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_;
 };
 
 #endif

include/pika_consensus.h

@@ -13,6 +13,7 @@
 #include "include/pika_slave_node.h"
 #include "include/pika_stable_log.h"
 #include "pstd/include/env.h"
+#include <future>
 
 class Context : public pstd::noncopyable {
  public:
@@ -129,7 +130,7 @@ class Log {
 
   Log();
   int Size();
-  void AppendLog(const LogItem& item);
+  pstd::Status AppendLog(const std::shared_ptr<StableLog>& stable_log, const std::shared_ptr<Cmd>& cmd_ptr);
   LogOffset LastOffset();
   LogOffset FirstOffset();
   LogItem At(int index);
@@ -154,6 +155,7 @@ class ConsensusCoordinator {
   // invoked by dbsync process
   pstd::Status Reset(const LogOffset& offset);
 
+  pstd::Status SyncAndWait();
   pstd::Status ProposeLog(const std::shared_ptr<Cmd>& cmd_ptr);
   pstd::Status UpdateSlave(const std::string& ip, int port, const LogOffset& start, const LogOffset& end);
   pstd::Status AddSlaveNode(const std::string& ip, int port, int session_id);
@@ -249,10 +251,10 @@ class ConsensusCoordinator {
  public:
   void InitContext() { context_->Init(); }
   bool checkFinished(const LogOffset& offset);
-  pstd::Status AppendEntries(const std::shared_ptr<Cmd>& cmd_ptr, LogOffset& cur_logoffset);
+  pstd::Status AppendEntries(const std::shared_ptr<Cmd>& cmd_ptr);
   void SetConsistency(bool is_consistency);
   bool GetISConsistency();
-  pstd::Status SendBinlog(std::shared_ptr<SlaveNode> slave_ptr, std::string db_name);
+  pstd::Status SendBinlog(const std::shared_ptr<SlaveNode>& slave_ptr, const std::string& db_name);
   pstd::Status Truncate(const LogOffset& offset);
   pstd::Status AppendSlaveEntries(const std::shared_ptr<Cmd>& cmd_ptr, const BinlogItem& attribute);
   pstd::Status CommitAppLog(const LogOffset& master_committed_id);
@@ -282,9 +284,19 @@ class ConsensusCoordinator {
   pstd::CondVar* GetCommittedIdCv() { return &committed_id_cv_; }
 
  private:
-  pstd::Status PersistAppendBinlog(const std::shared_ptr<Cmd>& cmd_ptr, LogOffset& cur_offset);
+  void SyncBinlogLoop();
+  pstd::Status PersistAppendBinlog(const std::shared_ptr<Cmd>& cmd_ptr);
 
  private:
+  // For async sync
+  pstd::Mutex sync_mu_;
+  pstd::CondVar sync_cv_;
+  std::atomic<bool> needs_sync_ = false;
+  std::atomic<bool> thread_stop_ = false;
+  std::thread sync_thread_;
+  pstd::Mutex promises_mu_;
+  std::vector<std::promise<pstd::Status>> sync_promises_;
+
   std::shared_mutex is_consistency_rwlock_;
   bool is_consistency_ = false;
   std::shared_mutex committed_id_rwlock_;
@@ -294,6 +306,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_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_repl_server_conn.h

@@ -21,7 +21,6 @@ class PikaReplServerConn : public net::PbConn {
   PikaReplServerConn(int fd, const std::string& ip_port, net::Thread* thread, void* worker_specific_data,
                      net::NetMultiplexer* mpx);
   ~PikaReplServerConn() override;
-
   static void HandleMetaSyncRequest(void* arg);
   static void HandleTrySyncRequest(void* arg);
 

include/pika_rm.h

@@ -30,6 +30,7 @@
 #define kSendKeepAliveTimeout (2 * 1000000)
 #define kRecvKeepAliveTimeout (20 * 1000000)
 
+std::string CreateBatchFromTasks(const std::vector<WriteTask>& tasks);
 
 class SyncDB {
  public:
@@ -83,11 +84,11 @@ class SyncMasterDB : public SyncDB {
     return coordinator_.StableLogger()->Logger();
   }
 
+  std::shared_ptr<SlaveNode> GetSlaveNode(const std::string& ip, int port);
  private:
   // invoker need to hold slave_mu_
   pstd::Status ReadBinlogFileToWq(const std::shared_ptr<SlaveNode>& slave_ptr);
 
-  std::shared_ptr<SlaveNode> GetSlaveNode(const std::string& ip, int port);
   std::unordered_map<std::string, std::shared_ptr<SlaveNode>> GetAllSlaveNodes();
 
   pstd::Mutex session_mu_;
@@ -112,8 +113,7 @@ class SyncMasterDB : public SyncDB {
   pstd::Status UpdateCommittedID();
   pstd::Status CommitAppLog(const LogOffset& master_committed_id);
   pstd::Status Truncate(const LogOffset& offset);
-
-
+  pstd::Status SyncBinlogAndWait();
 };
 
 class SyncSlaveDB : public SyncDB {
@@ -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_;
 };

include/pika_slave_node.h

@@ -78,6 +78,12 @@ class SlaveNode : public RmNode {
   pstd::Status Update(const LogOffset& start, const LogOffset& end, LogOffset* updated_offset);
 
   pstd::Mutex slave_mu;
+
+  // For replication batch
+  std::vector<WriteTask> task_buffer_;
+  uint64_t buffer_start_time_us_ = 0;
+  uint64_t ack_timeout_start_time_us_ = 0;
+  uint64_t pending_since_us_ = 0;
 };
 
 #endif  // PIKA_SLAVE_NODE_H

src/pika_auxiliary_thread.cc

@@ -8,6 +8,7 @@
 #include "include/pika_define.h"
 #include "include/pika_rm.h"
 #include "include/pika_server.h"
+#include "include/pika_conf.h"
 
 extern PikaServer* g_pika_server;
 extern std::unique_ptr<PikaReplicaManager> g_pika_rm;
@@ -34,17 +35,23 @@ void* PikaAuxiliaryThread::ThreadMain() {
 
     g_pika_server->CheckLeaderProtectedMode();
 
-    // TODO(whoiami) timeout
-    s = g_pika_server->TriggerSendBinlogSync();
-    if (!s.ok()) {
-      LOG(WARNING) << s.ToString();
-    }
-    // send to peer
+    // send to peer first if there are queued packets
     int res = g_pika_server->SendToPeer();
     if (res == 0) {
-      // sleep 100 ms
+      // idle: wait for a short period (consensus-timeout) or notification
       std::unique_lock lock(mu_);
-      cv_.wait_for(lock, 100ms);
+      auto to = std::chrono::milliseconds(g_pika_conf->consensus_timeout());
+      if (to.count() <= 0) {
+        to = 10ms;
+      }
+      cv_.wait_for(lock, to);
+      // after wait, trigger replication send once (size/timeout gating will decide to send or not)
+      s = g_pika_server->TriggerSendBinlogSync();
+      if (!s.ok()) {
+        LOG(WARNING) << s.ToString();
+      }
+      // consume what may have been produced by trigger
+      g_pika_server->SendToPeer();
     } else {
       // LOG_EVERY_N(INFO, 1000) << "Consume binlog number " << res;
     }