Fix the low qps bug

Author: YuCai18

src/pika_consensus.cc

@@ -998,33 +998,43 @@ Status ConsensusCoordinator::ApplyBinlog(const std::vector<Log::LogItem>& logs)
 
 Status ConsensusCoordinator::SendBinlog(std::shared_ptr<SlaveNode> slave_ptr, std::string db_name) {
   std::vector<WriteTask> tasks;
-  LogOffset prev_offset = slave_ptr->sent_offset;
-  LOG(INFO) << "SendBinlog: logs_->LastOffset()=" << logs_->LastOffset().ToString()
-            << ", slave_ptr->sent_offset=" << slave_ptr->sent_offset.ToString();
-  // Check if there are new log entries that need to be sent to the slave
-  if (logs_->LastOffset() >= slave_ptr->acked_offset) {
-    LOG(INFO) << "SendBinlog: logs_->Size()=" << logs_->Size();
-    // Find the index of the log entry corresponding to the slave's acknowledged offset
-    int index = logs_->FindOffset(slave_ptr->acked_offset);
-    LOG(INFO) << "SendBinlog: index=" << index;
-    if (index < logs_->Size()) {
-      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());
+  if (!g_pika_server->IsConsistency()) {
+    return Status::OK();
+  }
+  std::string ip = slave_ptr->Ip();
+  int port = slave_ptr->Port();
+  int32_t session_id = slave_ptr->SessionId();
+
+  LogOffset last_sent = slave_ptr->sent_offset;
+  if (logs_->LastOffset() > last_sent) {
+    int send_start_index = logs_->FindOffset(last_sent);
+    if (send_start_index < 0) {
+      LOG(WARNING) << "Binlog offset not found, maybe purged. last_sent: " << last_sent.ToString();
+      return Status::Corruption("cant find the file_num");
+    }
+
+    if (send_start_index < logs_->Size() && logs_->At(send_start_index).offset == last_sent) {
+      send_start_index++;
+    }
+
+    if (send_start_index < logs_->Size()) {
+      LogOffset prev_offset = send_start_index > 0 ? logs_->At(send_start_index - 1).offset : LogOffset();
+      for (int i = send_start_index; i < logs_->Size(); ++i) {
+        const auto& item = logs_->At(i);
+        RmNode rm_node(ip, port, db_name, session_id);
         WriteTask task(rm_node, BinlogChip(item.offset, item.binlog_), prev_offset, GetCommittedId());
         tasks.emplace_back(std::move(task));
 
         prev_offset = item.offset;
-        slave_ptr->sent_offset = item.offset;
       }
     }
   }
 
   if (!tasks.empty()) {
-    g_pika_rm->ProduceWriteQueue(slave_ptr->Ip(), slave_ptr->Port(), db_name, tasks);
+    g_pika_rm->ProduceWriteQueue(ip, port, db_name, tasks);
+    slave_ptr->sent_offset = tasks.back().binlog_chip_.offset_;
+    LOG(INFO) << "SendBinlog tasks to queue, slave: " << ip << ":" << port << " tasks: " << tasks.size()
+              << " new sent_offset: " << slave_ptr->sent_offset.ToString();
   }
   return Status::OK();
 }

src/pika_rm.cc

@@ -238,20 +238,18 @@ Status SyncMasterDB::WakeUpSlaveBinlogSync() {
     for (auto& slave_iter : slaves) {
         std::shared_ptr<SlaveNode> slave_ptr = slave_iter.second;
         std::lock_guard l(slave_ptr->slave_mu);
-        if (slave_ptr->sent_offset == slave_ptr->acked_offset) {
-          Status s;
-          if (coordinator_.GetISConsistency()) {
-            if(slave_ptr->slave_state == SlaveState::kSlaveBinlogSync||slave_ptr->slave_state == SlaveState::KCandidate){
-              s = coordinator_.SendBinlog(slave_ptr, db_info_.db_name_);
-            }
-          } else {
-            s = ReadBinlogFileToWq(slave_ptr);
-          }
-          if (!s.ok()) {
-            to_del.push_back(slave_ptr);
-            LOG(WARNING) << "WakeUpSlaveBinlogSync failed, marking for deletion: "
-                             << slave_ptr->ToStringStatus() << " - " << s.ToString();
+        Status s;
+        if (coordinator_.GetISConsistency()) {
+          if(slave_ptr->slave_state == SlaveState::kSlaveBinlogSync||slave_ptr->slave_state == SlaveState::KCandidate){
+            s = coordinator_.SendBinlog(slave_ptr, db_info_.db_name_);
           }
+        } else {
+          s = ReadBinlogFileToWq(slave_ptr);
+        }
+        if (!s.ok()) {
+          to_del.push_back(slave_ptr);
+          LOG(WARNING) << "WakeUpSlaveBinlogSync failed, marking for deletion: "
+                           << slave_ptr->ToStringStatus() << " - " << s.ToString();
         }
     }