fix some bugs

Author: YuCai18

include/pika_client_conn.h

@@ -113,7 +113,7 @@ class PikaClientConn : public net::RedisConn {
   std::vector<std::shared_ptr<std::string>> resp_array;
 
   std::shared_ptr<TimeStat> time_stat_;
-
+  void TryWriteResp();
  private:
   net::ServerThread* const server_thread_;
   std::string current_db_;
@@ -134,7 +134,7 @@ class PikaClientConn : public net::RedisConn {
   void ProcessMonitor(const PikaCmdArgsType& argv);
 
   void ExecRedisCmd(const PikaCmdArgsType& argv, std::shared_ptr<std::string>& resp_ptr, bool cache_miss_in_rtc);
-  void TryWriteResp();
+  // void TryWriteResp();
 };
 
 struct ClientInfo {

include/pika_command_collector.h

@@ -62,7 +62,7 @@ class PikaCommandCollector {
     * @param callback callback function after processing is completed
     * @return whether the addition was successful
   */
-  bool AddCommand(const std::shared_ptr<Cmd>& cmd_ptr, CommandCallback callback);
+  bool AddCommand(std::shared_ptr<Cmd> cmd_ptr, CommandCallback callback);
 
   /**
     * @brief Set the batch max wait time

include/pika_kv.h

@@ -24,6 +24,7 @@ class SetCmd : public Cmd {
     res.push_back(key_);
     return res;
   }
+  ~SetCmd();
   void Do() override;
   void DoUpdateCache() override;
   void DoThroughDB() override;

src/pika_client_conn.cc

@@ -591,8 +591,13 @@ void PikaClientConn::ExecRedisCmd(const PikaCmdArgsType& argv, std::shared_ptr<s
   }
 
   std::shared_ptr<Cmd> cmd_ptr = DoCmd(argv, opt, resp_ptr, cache_miss_in_rtc);
-  *resp_ptr = std::move(cmd_ptr->res().message());
-  resp_num--;
+  // *resp_ptr = std::move(cmd_ptr->res().message());
+  // resp_num--;
+  if (opt == kCmdNameSet) {
+  } else {
+    *resp_ptr = std::move(cmd_ptr->res().message());
+    resp_num--;
+  }
 }
 
 std::queue<std::shared_ptr<Cmd>> PikaClientConn::GetTxnCmdQue() { return txn_cmd_que_; }

src/pika_command_collector.cc

@@ -30,7 +30,7 @@ PikaCommandCollector::~PikaCommandCollector() {
             << " commands, " << total_batches_.load() << " batches";
 }
 
-bool PikaCommandCollector::AddCommand(const std::shared_ptr<Cmd>& cmd_ptr, CommandCallback callback) {
+bool PikaCommandCollector::AddCommand(std::shared_ptr<Cmd> cmd_ptr, CommandCallback callback) {
   if (!cmd_ptr || !cmd_ptr->is_write()) {
     LOG(WARNING) << "Attempt to add non-write command to CommandCollector";
     return false;

src/pika_consensus.cc

@@ -868,7 +868,7 @@ Status ConsensusCoordinator::AppendEntries(const std::shared_ptr<Cmd>& cmd_ptr,
     return s;
   }
 
-  g_pika_server->SignalAuxiliary();
+  // g_pika_server->SignalAuxiliary();
   return Status::OK();
 }
 Status ConsensusCoordinator::AppendSlaveEntries(const std::shared_ptr<Cmd>& cmd_ptr, const BinlogItem& attribute) {
@@ -914,7 +914,8 @@ Status ConsensusCoordinator::UpdateCommittedID() {
   LogOffset slave_prepared_id = LogOffset();
 
   for (const auto& slave : slaves) {
-    if (slave.second->slave_state == kSlaveBinlogSync) {
+    // Consider both kSlaveBinlogSync and KCandidate slaves for CommittedID calculation
+    if (slave.second->slave_state == kSlaveBinlogSync || slave.second->slave_state == KCandidate) {
       if (slave_prepared_id == LogOffset()) {
         slave_prepared_id = slave.second->acked_offset;
       } else if (slave.second->acked_offset < slave_prepared_id) {
@@ -990,86 +991,28 @@ void ConsensusCoordinator::BatchInternalApplyFollower(const std::vector<std::sha
 
 Status ConsensusCoordinator::SendBinlog(std::shared_ptr<SlaveNode> slave_ptr, std::string db_name) {
   std::vector<WriteTask> tasks;
-  const int MAX_BATCH_SIZE = 100; // Maximum number of logs to send in a single batch
-
-  // Get current committed_id to ensure it's sent to the slave
-  LogOffset current_committed_id = GetCommittedId();
-  LOG(INFO) << "SendBinlog: [Thread " << std::this_thread::get_id() << "] Current committed_id: " << current_committed_id.ToString()
-            << ", sending to slave " << slave_ptr->Ip() << ":" << slave_ptr->Port()
-            << ", logs_ addr: " << logs_.get() << ", db_name: " << db_name_;
 
   // Check if there are new log entries that need to be sent to the slave
-  LOG(INFO) << "SendBinlog: logs_->LastOffset()=" << logs_->LastOffset().ToString() 
-            << ", slave_ptr->acked_offset=" << slave_ptr->acked_offset.ToString()
-            << ", logs_->Size()=" << logs_->Size();
-            
-  if (logs_->Size() > 0 && logs_->LastOffset() >= slave_ptr->acked_offset) {
+  if (logs_->LastOffset() >= slave_ptr->acked_offset) {
     // Find the index of the log entry corresponding to the slave's acknowledged offset
     int index = logs_->FindOffset(slave_ptr->acked_offset);
-    int entries_to_send = logs_->Size() - index;
-    LOG(INFO) << "SendBinlog: Found " << entries_to_send << " new log entries to send, "
-              << "starting from index " << index << " of " << logs_->Size();
-    
     if (index < logs_->Size()) {
-      // Send log entries in optimized batches
-      RmNode rm_node(slave_ptr->Ip(), slave_ptr->Port(), db_name, slave_ptr->SessionId());
-      
-      // For large batches, use specialized batch handling
-      if (entries_to_send > MAX_BATCH_SIZE) {
-        LOG(INFO) << "SendBinlog: Using optimized batch sending for " << entries_to_send << " entries";
-        
-        // Process in chunks of MAX_BATCH_SIZE
-        for (int batch_start = index; batch_start < logs_->Size(); batch_start += MAX_BATCH_SIZE) {
-          int batch_end = std::min(batch_start + MAX_BATCH_SIZE, logs_->Size());
-          std::vector<WriteTask> batch_tasks;
-          
-          for (int i = batch_start; i < batch_end; ++i) {
-            Log::LogItem item = logs_->At(i);
-            WriteTask task(rm_node, BinlogChip(item.offset, item.binlog_), item.offset, current_committed_id);
-            batch_tasks.push_back(task);
-          }
-          
-          g_pika_rm->ProduceWriteQueue(slave_ptr->Ip(), slave_ptr->Port(), db_name, batch_tasks);
-          LOG(INFO) << "SendBinlog: Sent batch " << (batch_start - index) / MAX_BATCH_SIZE + 1 
-                    << " with " << (batch_end - batch_start) << " entries";
-        }
-      } else {
-        // Send all entries in a single batch
-        for (int i = index; i < logs_->Size(); ++i) {
-          Log::LogItem item = logs_->At(i);
-          WriteTask task(rm_node, BinlogChip(item.offset, item.binlog_), item.offset, current_committed_id);
-          tasks.push_back(task);
-        }
+      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_), slave_ptr->sent_offset, GetCommittedId());
+        tasks.emplace_back(std::move(task));
+
+        slave_ptr->sent_offset = item.offset;
       }
-    } else {
-      LOG(INFO) << "SendBinlog: No new log entries to send, index " << index << " is out of range (logs size: " << logs_->Size() << ")";
-    }
-  } else {
-    if (logs_->Size() == 0) {
-      LOG(INFO) << "SendBinlog: No logs available yet (logs_->Size()=0), will send empty binlog to maintain connection";
-    } else {
-      LOG(INFO) << "SendBinlog: Slave is already up to date, last offset: " << logs_->LastOffset().ToString()
-                << ", slave acked offset: " << slave_ptr->acked_offset.ToString();
     }
   }
 
-  // Only send empty binlog if there are no actual log entries to send
-  // This prevents the deadlock where master waits for slave ACK and slave waits for master data
-  if (tasks.empty() && logs_->Size() == 0) {
-    // LOG(INFO) << "SendBinlog: Sending empty binlog with current committed_id: " << current_committed_id.ToString();
-    RmNode rm_node(slave_ptr->Ip(), slave_ptr->Port(), db_name, slave_ptr->SessionId());
-    // Create an empty WriteTask that includes the current committed_id
-    WriteTask empty_task(rm_node, BinlogChip(LogOffset(), ""), LogOffset(), current_committed_id);
-    tasks.push_back(empty_task);
-  }
-
-  // Send the tasks to the slave
   if (!tasks.empty()) {
-    LOG(INFO) << "SendBinlog: Sending " << tasks.size() << " tasks to slave " << slave_ptr->Ip() << ":" << slave_ptr->Port();
-    extern std::unique_ptr<PikaReplicaManager> g_pika_rm;
     g_pika_rm->ProduceWriteQueue(slave_ptr->Ip(), slave_ptr->Port(), db_name, tasks);
-  } else {
-    LOG(INFO) << "SendBinlog: No tasks to send to slave " << slave_ptr->Ip() << ":" << slave_ptr->Port();
   }
   return Status::OK();
 }

src/pika_kv.cc

@@ -15,7 +15,22 @@
 
 extern std::unique_ptr<PikaConf> g_pika_conf;
 /* SET key value [NX] [XX] [EX <seconds>] [PX <milliseconds>] */
+SetCmd::~SetCmd() {
+  auto tmp_conn = GetConn();
+  if (!tmp_conn) {
+    return;
+  }
+
+  auto pc = dynamic_cast<PikaClientConn*>(tmp_conn.get());
+  std::shared_ptr<std::string> resp_ptr = std::make_shared<std::string>();
+  *resp_ptr = std::move(res().message());
+  pc->resp_num--;
+  pc->resp_array.push_back(resp_ptr);
+  pc->TryWriteResp();
+  LOG(INFO) << "SetCmd::~SetCmd() is completed";
+}
 void SetCmd::DoInitial() {
+  LOG(INFO) << "SetCmd::DoInitial() is started";
   if (!CheckArg(argv_.size())) {
     res_.SetRes(CmdRes::kWrongNum, kCmdNameSet);
     return;
@@ -65,6 +80,7 @@ void SetCmd::DoInitial() {
 }
 
 void SetCmd::Do() {
+  LOG(INFO) << "SetCmd::Do() is started";
   int32_t res = 1;
   STAGE_TIMER_GUARD(storage_duration_ms, true);
   switch (condition_) {
@@ -102,10 +118,12 @@ void SetCmd::Do() {
 }
 
 void SetCmd::DoThroughDB() {
+  LOG(INFO) << "SetCmd::DoThroughDB() is started";
   Do();
 }
 
 void SetCmd::DoUpdateCache() {
+  LOG(INFO) << "SetCmd::DoUpdateCache() is started";
   if (SetCmd::kNX == condition_ || IsTooLargeKey(g_pika_conf->max_key_size_in_cache())) {
     return;
   }
@@ -121,6 +139,7 @@ void SetCmd::DoUpdateCache() {
 }
 
 std::string SetCmd::ToRedisProtocol() {
+  LOG(INFO) << "SetCmd::ToRedisProtocol() is started";
   if (condition_ == SetCmd::kEXORPX) {
     std::string content;
     content.reserve(RAW_ARGS_LEN);

src/pika_rm.cc

@@ -1448,43 +1448,23 @@ void PikaReplicaManager::ProcessCommandBatches(const std::vector<std::shared_ptr
       const auto& cmd_ptr = all_commands[cmd_idx];
       LogOffset cmd_offset;
 
-      if (sync_db->GetISConsistency()) {
-        // For consistency mode, use ConsensusProposeLog directly  
-        pstd::Status s = sync_db->ConsensusProposeLog(cmd_ptr);
-        if (!s.ok()) {
-          LOG(ERROR) << "Failed to append command " << cmd_ptr->name() << ": " << s.ToString();
-          // Call callbacks with error for remaining commands
-          for (size_t error_idx = cmd_idx; error_idx < all_commands.size(); ++error_idx) {
-            auto& batch_info = command_to_batch_map[error_idx];
-            auto batch = batch_info.first;
-            size_t batch_cmd_idx = batch_info.second;
-            if (batch_cmd_idx < batch->callbacks.size() && batch->callbacks[batch_cmd_idx]) {
-              batch->callbacks[batch_cmd_idx](LogOffset(), s);
-            }
-          }
-          return;
-        }
-        // Get the prepared ID as the offset
-        cmd_offset = sync_db->GetPreparedId();
-      } else {
-        // For non-consistency mode, use ConsensusProposeLog as well
-        pstd::Status s = sync_db->ConsensusProposeLog(cmd_ptr);
-        if (!s.ok()) {
-          LOG(ERROR) << "Failed to propose command " << cmd_ptr->name() << ": " << s.ToString();
-          // Call callbacks with error for remaining commands
-          for (size_t error_idx = cmd_idx; error_idx < all_commands.size(); ++error_idx) {
-            auto& batch_info = command_to_batch_map[error_idx];
-            auto batch = batch_info.first;
-            size_t batch_cmd_idx = batch_info.second;
-            if (batch_cmd_idx < batch->callbacks.size() && batch->callbacks[batch_cmd_idx]) {
-              batch->callbacks[batch_cmd_idx](LogOffset(), s);
-            }
+      pstd::Status s = sync_db->ConsensusProposeLog(cmd_ptr);
+      if (!s.ok()) {
+        LOG(ERROR) << "Failed to " << (sync_db->GetISConsistency() ? "append" : "propose") 
+                   << " command " << cmd_ptr->name() << ": " << s.ToString();
+        // Call callbacks with error for remaining commands
+        for (size_t error_idx = cmd_idx; error_idx < all_commands.size(); ++error_idx) {
+          auto& batch_info = command_to_batch_map[error_idx];
+          auto batch = batch_info.first;
+          size_t batch_cmd_idx = batch_info.second;
+          if (batch_cmd_idx < batch->callbacks.size() && batch->callbacks[batch_cmd_idx]) {
+            batch->callbacks[batch_cmd_idx](LogOffset(), s);
           }
-          return;
         }
-        // Get the prepared ID as the offset
-        cmd_offset = sync_db->GetPreparedId();
+        return;
       }
+      // Get the prepared ID as the offset
+      cmd_offset = sync_db->GetPreparedId();
 
       all_offsets[cmd_idx] = cmd_offset;
 
@@ -1597,6 +1577,7 @@ void PikaReplicaManager::RocksDBThreadLoop() {
 }
 
 size_t PikaReplicaManager::ProcessCommittedBatchGroups(const LogOffset& committed_id) {
+  LOG(INFO) << "ProcessCommittedBatchGroups started";
   std::queue<std::shared_ptr<BatchGroup>> groups_to_process;
   // Get pending BatchGroups and separate committed from uncommitted
   {
@@ -1662,6 +1643,7 @@ size_t PikaReplicaManager::ProcessCommittedBatchGroups(const LogOffset& committe
       }
     }
   }
+  
   return groups_count;
 }
 

tests/integration/clean_start.sh

@@ -38,17 +38,17 @@ echo "Replication setup successful."
 
 echo "Running benchmark..."
 
-redis-benchmark -p 9301 -t set -r 100000 -n 100000 -c 10000 --threads 4
-echo "Benchmark finished." 
+#redis-benchmark -p 9301 -t set -r 100000 -n 100000 -c 10000 --threads 4
+# echo "Benchmark finished." 
 
-echo -e "\n==== 主节点 INFO 日志 ===="
-tail -n 150 ./pacifica_test/master/log/pika.INFO
+# echo -e "\n==== 主节点 INFO 日志 ===="
+# tail -n 150 ./pacifica_test/master/log/pika.INFO
 
-echo -e "\n==== 主节点 WARNING 日志 ===="
-tail -n 150 ./pacifica_test/master/log/pika.WARNING 
+# echo -e "\n==== 主节点 WARNING 日志 ===="
+# tail -n 150 ./pacifica_test/master/log/pika.WARNING 
 
-echo -e "\n==== 从节点 INFO 日志 ===="
-tail -n 150 ./pacifica_test/slave1/log/pika.INFO
+# echo -e "\n==== 从节点 INFO 日志 ===="
+# tail -n 150 ./pacifica_test/slave1/log/pika.INFO
 
-echo -e "\n==== 从节点 WARNING 日志 ===="
-tail -n 150 ./pacifica_test/slave1/log/pika.WARNING 
+# echo -e "\n==== 从节点 WARNING 日志 ===="
+# tail -n 150 ./pacifica_test/slave1/log/pika.WARNING