Fix timeout bug

Author: YuCai18

conf/pika.conf

@@ -153,6 +153,13 @@ 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 :

include/pika_binlog.h

@@ -61,6 +61,8 @@ class Binlog : public pstd::noncopyable {
    * Set Producer pro_num and pro_offset with lock
    */
   pstd::Status SetProducerStatus(uint32_t pro_num, uint64_t pro_offset, uint32_t term = 0, uint64_t index = 0);
+  // Force sync data to disk
+  pstd::Status Sync();
   // Need to hold Lock();
   pstd::Status Truncate(uint32_t pro_num, uint64_t pro_offset, uint64_t index);
 

include/pika_conf.h

@@ -888,6 +888,9 @@ 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;
@@ -1065,6 +1068,10 @@ 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

@@ -257,7 +257,7 @@ class ConsensusCoordinator {
   pstd::Status AppendSlaveEntries(const std::shared_ptr<Cmd>& cmd_ptr, const BinlogItem& attribute);
   pstd::Status CommitAppLog(const LogOffset& master_committed_id);
   pstd::Status UpdateCommittedID();
-  pstd::Status ApplyBinlog(const std::shared_ptr<Cmd>& cmd_ptr);
+  pstd::Status ApplyBinlog(const std::vector<Log::LogItem>& logs);
   pstd::Status ProcessCoordination();
 
   LogOffset GetCommittedId() {
@@ -276,7 +276,10 @@ class ConsensusCoordinator {
     std::lock_guard l(committed_id_rwlock_);
     committed_id_ = offset;
     context_->UpdateAppliedIndex(committed_id_);
+    committed_id_cv_.notify_all();
   }
+  pstd::Mutex* GetCommittedIdMu() { return &committed_id_mu_; }
+  pstd::CondVar* GetCommittedIdCv() { return &committed_id_cv_; }
 
  private:
   pstd::Status PersistAppendBinlog(const std::shared_ptr<Cmd>& cmd_ptr, LogOffset& cur_offset);
@@ -285,6 +288,8 @@ class ConsensusCoordinator {
   std::shared_mutex is_consistency_rwlock_;
   bool is_consistency_ = false;
   std::shared_mutex committed_id_rwlock_;
+  pstd::Mutex committed_id_mu_;
+  pstd::CondVar committed_id_cv_;
   LogOffset committed_id_ = LogOffset();
   std::shared_mutex prepared_id__rwlock_;
   LogOffset prepared_id_ = LogOffset();

include/pika_define.h

@@ -354,6 +354,14 @@ 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_client.h

@@ -88,14 +88,22 @@ class PikaReplClient {
     async_write_db_task_counts_[db_index].fetch_sub(incr_step, std::memory_order::memory_order_seq_cst);
   }
 
-  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);
-  }
+  int32_t GetUnfinishedAsyncWriteDBTaskCount(const std::string& db_name);
+  void SignalAsyncWriteDBTaskEnd(const std::string& db_name);
+  void WaitForAsyncWriteDBTaskEnd(const std::string& db_name);
+
+  // unfinished_async_write_db_tasks related
+  pstd::Mutex unfinished_async_write_db_tasks_mu_;
+  std::unordered_map<std::string, int32_t> unfinished_async_write_db_tasks_;
+  pstd::CondVar async_write_db_tasks_cond_;
+
+  // db_write_block_fds_ related
+  pstd::Mutex db_write_block_fds_mu_;
+  std::set<int> db_write_block_fds_;
 
  private:
-  size_t GetBinlogWorkerIndexByDBName(const std::string &db_name);
+  size_t GetBinlogWorkerIndexByDBName(const std::string& db_name);
+  size_t GetDBWorkerIndexByDBName(const std::string& db_name);
   size_t GetHashIndexByKey(const std::string& key);
   void UpdateNextAvail() { next_avail_ = (next_avail_ + 1) % static_cast<int32_t>(write_binlog_workers_.size()); }
 

include/pika_rm.h

@@ -207,6 +207,8 @@ class PikaReplicaManager {
                                const std::shared_ptr<InnerMessage::InnerResponse>& res,
                                const std::shared_ptr<net::PbConn>& conn, void* res_private_data);
   void ScheduleWriteDBTask(const std::shared_ptr<Cmd>& cmd_ptr, const std::string& db_name);
+  void SignalAsyncWriteDBTaskEnd(const std::string& db_name);
+  void WaitForAsyncWriteDBTaskEnd(const std::string& db_name);
   void ScheduleReplClientBGTaskByDBName(net::TaskFunc , void* arg, const std::string &db_name);
   void ReplServerRemoveClientConn(int fd);
   void ReplServerUpdateClientConnMap(const std::string& ip_port, int fd);
@@ -244,7 +246,7 @@ class PikaReplicaManager {
   pstd::Mutex write_queue_mu_;
 
   // every host owns a queue, the key is "ip + port"
-  std::unordered_map<std::string, std::unordered_map<std::string, std::queue<WriteTask>>> write_queues_;
+  std::unordered_map<std::string, std::unordered_map<std::string, std::queue<std::pair<WriteTask, uint64_t>>>> write_queues_;
   std::unique_ptr<PikaReplClient> pika_repl_client_;
   std::unique_ptr<PikaReplServer> pika_repl_server_;
 };

src/net/src/pb_conn.cc

@@ -153,7 +153,7 @@ WriteStatus PbConn::SendReply() {
     if (item_len - write_buf_.item_pos_ != 0) {
       return kWriteHalf;
     }
-    LOG(ERROR) << "write item success";
+    //LOG(ERROR) << "write item success";
   }
   return kWriteAll;
 }

src/pika_binlog.cc

@@ -469,3 +469,10 @@ Status Binlog::Truncate(uint32_t pro_num, uint64_t pro_offset, uint64_t index) {
 
   return Status::OK();
 }
+
+Status Binlog::Sync() {
+  if (queue_) {
+    return queue_->Sync();
+  }
+  return Status::Corruption("Logger not initialized");
+}

src/pika_conf.cc

@@ -707,6 +707,11 @@ 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;
 }
 
@@ -912,3 +917,7 @@ 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_; }