fix some bugs

YuCai18 · YuCai18 · commit d6c0db3a7bc8 · 2025-09-05T11:18:25.000+08:00
diff --git a/include/pika_client_conn.h b/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 {
diff --git a/include/pika_command_collector.h b/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
diff --git a/include/pika_kv.h b/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;
diff --git a/src/pika_client_conn.cc b/src/pika_client_conn.cc
@@ -229,14 +229,16 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
       if (command_collector) {
         // Create callback to handle command completion
         auto callback = [this, c_ptr](const LogOffset& offset, pstd::Status status) {
-          if (status.ok()) {
-            // Command was successfully processed through the pipeline
-            LOG(INFO) << "Command " << c_ptr->name() << " completed via CommandCollector";
-          } else {
-            // Set error response
-            c_ptr->res().SetRes(CmdRes::kErrOther, "Command processing failed: " + status.ToString());
-            LOG(ERROR) << "Command " << c_ptr->name() << " failed in CommandCollector: " << status.ToString();
-          }
+            LOG(INFO) << "Command completed";
+            auto pc = dynamic_cast<PikaClientConn*>(c_ptr->GetConn().get());
+            if (pc) {
+                auto resp_ptr = c_ptr->GetResp();
+                if (resp_ptr) {
+                    *resp_ptr = std::move(c_ptr->res().message());
+                }
+                pc->resp_num--;
+                pc->TryWriteResp();
+            }
         };
         
         // Add command to collector for batch processing
@@ -591,8 +593,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_; }
diff --git a/src/pika_command.cc b/src/pika_command.cc
@@ -952,35 +952,36 @@ bool Cmd::DoReadCommandInCache() {
 
 
 void Cmd::DoBinlog() {
-  if (res().ok() && is_write() && g_pika_conf->write_binlog()) {
-    std::shared_ptr<net::NetConn> conn_ptr = GetConn();
-    std::shared_ptr<std::string> resp_ptr = GetResp();
-    // Consider that dummy cmd appended by system, both conn and resp are null.
-    if ((!conn_ptr || !resp_ptr) && (name_ != kCmdDummy)) {
-      if (!conn_ptr) {
-        LOG(WARNING) << sync_db_->SyncDBInfo().ToString() << " conn empty.";
-      }
-      if (!resp_ptr) {
-        LOG(WARNING) << sync_db_->SyncDBInfo().ToString() << " resp empty.";
-      }
-      res().SetRes(CmdRes::kErrOther);
-      return;
-    }
-
-    Status s = sync_db_->ConsensusProposeLog(shared_from_this());
-    if (!s.ok()) {
-       if(g_pika_server->IsConsistency()&&s.IsTimeout()){
-         res().SetRes(CmdRes::kConsistencyTimeout, "Timeout waiting for consistency");
-         LOG(WARNING) << sync_db_->SyncDBInfo().ToString() << " Slave node consistency timeout"
-                        << s.ToString();
-       }else{
-        LOG(WARNING) << sync_db_->SyncDBInfo().ToString() << " Writing binlog failed, maybe no space left on device "
-                        << s.ToString();
-        res().SetRes(CmdRes::kErrOther, s.ToString());
-       }
-      return;
-    }
-  }
+//   if (res().ok() && is_write() && g_pika_conf->write_binlog()) {
+//     std::shared_ptr<net::NetConn> conn_ptr = GetConn();
+//     std::shared_ptr<std::string> resp_ptr = GetResp();
+//     // Consider that dummy cmd appended by system, both conn and resp are null.
+//     if ((!conn_ptr || !resp_ptr) && (name_ != kCmdDummy)) {
+//       if (!conn_ptr) {
+//         LOG(WARNING) << sync_db_->SyncDBInfo().ToString() << " conn empty.";
+//       }
+//       if (!resp_ptr) {
+//         LOG(WARNING) << sync_db_->SyncDBInfo().ToString() << " resp empty.";
+//       }
+//       res().SetRes(CmdRes::kErrOther);
+//       return;
+//     }
+
+//     Status s = sync_db_->ConsensusProposeLog(shared_from_this());
+//     if (!s.ok()) {
+//        if(g_pika_server->IsConsistency()&&s.IsTimeout()){
+//          res().SetRes(CmdRes::kConsistencyTimeout, "Timeout waiting for consistency");
+//          LOG(WARNING) << sync_db_->SyncDBInfo().ToString() << " Slave node consistency timeout"
+//                         << s.ToString();
+//        }else{
+//         LOG(WARNING) << sync_db_->SyncDBInfo().ToString() << " Writing binlog failed, maybe no space left on device "
+//                         << s.ToString();
+//         res().SetRes(CmdRes::kErrOther, s.ToString());
+//        }
+//       return;
+//     }
+//   }
+    return;
 }
 
 #define PIKA_STAGE_DURATION_OUTPUT(duration)      \
diff --git a/src/pika_command_collector.cc b/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;
diff --git a/src/pika_consensus.cc b/src/pika_consensus.cc
@@ -824,7 +824,6 @@ bool ConsensusCoordinator::checkFinished(const LogOffset& offset) {
 //// pacificA private:
 
 Status ConsensusCoordinator::PersistAppendBinlog(const std::shared_ptr<Cmd>& cmd_ptr, LogOffset& cur_offset) {
-  std::lock_guard l(order_mu_);
   std::string content = cmd_ptr->ToRedisProtocol();
   std::string binlog = std::string();
   LogOffset offset = LogOffset();
@@ -843,12 +842,16 @@ Status ConsensusCoordinator::PersistAppendBinlog(const std::shared_ptr<Cmd>& cmd
   // If successful, append the log entry to the logs
   // TODO: 这里logs_的appendlog操作和上边的stable_logger_->Logger()->Put不是原子的，可能导致offset大的先被追加到logs_中，
   // 多线程写入的时候窗口会对不上，最终主从断开连接。需要加逻辑保证原子性
-  LOG(INFO) << "PersistAppendBinlog: About to append to logs_, current size=" << logs_->Size() 
-            << ", offset=" << cur_offset.ToString() << ", cmd=" << cmd_ptr->name();
-  logs_->AppendLog(Log::LogItem(cur_offset, cmd_ptr, binlog));
-  LOG(INFO) << "PersistAppendBinlog: Successfully appended to logs_, new size=" << logs_->Size();
-
-  SetPreparedId(cur_offset);
+  //LOG(INFO) << "PersistAppendBinlog: About to append to logs_, current size=" << logs_->Size() 
+  //          << ", offset=" << cur_offset.ToString() << ", cmd=" << cmd_ptr->name();
+  // logs_->AppendLog(Log::LogItem(cur_offset, cmd_ptr, binlog));
+  //LOG(INFO) << "PersistAppendBinlog: Successfully appended to logs_, new size=" << logs_->Size();
+  {
+    std::lock_guard l(order_mu_);
+    // Append to logs_ under order lock to maintain ordering
+    logs_->AppendLog(Log::LogItem(cur_offset, cmd_ptr, binlog));
+    SetPreparedId(cur_offset);
+  }
 
   return stable_logger_->Logger()->IsOpened();
 }
@@ -868,7 +871,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) {
@@ -922,6 +925,23 @@ Status ConsensusCoordinator::UpdateCommittedID() {
       }
     }
   }
+  // if (!has_active_slaves) {
+  //   LogOffset master_prepared_id = GetPreparedId();
+  //   if (master_prepared_id.IsValid() && master_prepared_id >= GetCommittedId()) {
+  //     SetCommittedId(master_prepared_id);
+  //     LOG(INFO) << "PacificA update CommittedID (no active slaves): " << GetCommittedId().ToString()
+  //               << ", Total slaves: " << total_slaves 
+  //               << ", kSlaveBinlogSync: " << binlog_sync_slaves
+  //               << ", Other states: " << other_state_slaves;
+  //   } else {
+  //     LOG(INFO) << "PacificA update CommittedID: No active slaves, keeping current CommittedID: " << GetCommittedId().ToString()
+  //               << ", Total slaves: " << total_slaves 
+  //               << ", kSlaveBinlogSync: " << binlog_sync_slaves
+  //               << ", Other states: " << other_state_slaves;
+  //   }
+  //   // g_pika_rm->NotifyCommittedID(GetCommittedId());
+  //   return Status::OK();
+  // }
   if (slave_prepared_id < GetCommittedId()) {
     LOG(WARNING) << "Error: slave_prepared_id (" << slave_prepared_id.ToString() << ") < master_committedId ("
                  << GetCommittedId().ToString() << ")";
@@ -953,7 +973,9 @@ Status ConsensusCoordinator::ProcessCoordination() {
 // Execute the operation of writing to DB
 Status ConsensusCoordinator::ApplyBinlog(const std::shared_ptr<Cmd>& cmd_ptr) {
   auto opt = cmd_ptr->argv()[0];
+  LOG(INFO) << "[ApplyBinlog] Received command: " << opt << " for db: " << db_name_;
   if (pstd::StringToLower(opt) != kCmdNameFlushdb) {
+    LOG(INFO) << "[ApplyBinlog] Scheduling async task for " << opt;
     InternalApplyFollower(cmd_ptr);
   } else {
     int32_t wait_ms = 250;
@@ -990,86 +1012,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();
 }
diff --git a/src/pika_kv.cc b/src/pika_kv.cc
@@ -15,6 +15,20 @@
 
 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() {
   if (!CheckArg(argv_.size())) {
     res_.SetRes(CmdRes::kWrongNum, kCmdNameSet);
diff --git a/src/pika_repl_server.cc b/src/pika_repl_server.cc
@@ -17,7 +17,7 @@ extern PikaServer* g_pika_server;
 extern std::unique_ptr<PikaReplicaManager> g_pika_rm;
 
 PikaReplServer::PikaReplServer(const std::set<std::string>& ips, int port, int cron_interval) {
-  server_tp_ = std::make_unique<net::ThreadPool>(PIKA_REPL_SERVER_TP_SIZE, 100000, "PikaReplServer");
+  server_tp_ = std::make_unique<net::ThreadPool>(1, 100000, "PikaReplServer");
   pika_repl_server_thread_ = std::make_unique<PikaReplServerThread>(ips, port, cron_interval);
   pika_repl_server_thread_->set_thread_name("PikaReplServer");
 }
diff --git a/src/pika_rm.cc b/src/pika_rm.cc
diff --git a/src/pika_server.cc b/src/pika_server.cc
diff --git a/tests/integration/clean_start.sh b/tests/integration/clean_start.sh

Original file line number	Diff line number	Diff line change
`@@ -24,6 +24,7 @@ class SetCmd : public Cmd {`
`24`	`24`	`res.push_back(key_);`
`25`	`25`	`return res;`
`26`	`26`	`}`
	`27`	`+ ~SetCmd() {}`
`27`	`28`	`void Do() override;`
`28`	`29`	`void DoUpdateCache() override;`
`29`	`30`	`void DoThroughDB() override;`
Original file line number	Diff line number	Diff line change
`@@ -30,7 +30,7 @@ PikaCommandCollector::~PikaCommandCollector() {`
`30`	`30`	`<< " commands, " << total_batches_.load() << " batches";`
`31`	`31`	`}`
`32`	`32`
`33`		`-bool PikaCommandCollector::AddCommand(const std::shared_ptr<Cmd>& cmd_ptr, CommandCallback callback) {`
	`33`	`+bool PikaCommandCollector::AddCommand(std::shared_ptr<Cmd> cmd_ptr, CommandCallback callback) {`
`34`	`34`	`if (!cmd_ptr \|\| !cmd_ptr->is_write()) {`
`35`	`35`	`LOG(WARNING) << "Attempt to add non-write command to CommandCollector";`
`36`	`36`	`return false;`
Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@ extern PikaServer* g_pika_server;`
`17`	`17`	`extern std::unique_ptr<PikaReplicaManager> g_pika_rm;`
`18`	`18`
`19`	`19`	`PikaReplServer::PikaReplServer(const std::set<std::string>& ips, int port, int cron_interval) {`
`20`		`- server_tp_ = std::make_unique<net::ThreadPool>(PIKA_REPL_SERVER_TP_SIZE, 100000, "PikaReplServer");`
	`20`	`+ server_tp_ = std::make_unique<net::ThreadPool>(1, 100000, "PikaReplServer");`
`21`	`21`	`pika_repl_server_thread_ = std::make_unique<PikaReplServerThread>(ips, port, cron_interval);`
`22`	`22`	`pika_repl_server_thread_->set_thread_name("PikaReplServer");`
`23`	`23`	`}`