Optimize the log submission using the proto method

Author: wuxianrong

include/pika_raft.h

@@ -104,10 +104,5 @@ class RaftConfigCmd : public Cmd {
   std::string db_name_;
 };
 
-// Thread-local flag to detect if we're in on_apply context
-namespace pika_raft {
-extern thread_local bool g_in_raft_apply;
-}
-
 #endif  // PIKA_RAFT_H_
 

src/pika_command.cc

@@ -6,8 +6,6 @@
 #include <memory>
 #include <sstream>
 #include <utility>
-#include <future>
-#include <chrono>
 
 #include <glog/logging.h>
 #include "include/pika_acl.h"
@@ -897,17 +895,16 @@ void Cmd::InternalProcessCommand(const HintKeys& hint_keys) {
 
   uint64_t before_do_binlog_us = pstd::NowMicros();
   this->command_duration_ms = (before_do_binlog_us - before_do_command_us) / 1000;
-  
-  // Release locks BEFORE calling DoBinlog() to avoid deadlock in Raft mode
-  // In Raft mode, DoBinlog() will wait for on_apply which needs the same lock
+  DoBinlog(); 
+
+
   if (!IsSuspend()) {
     db_->DBUnlockShared();
   }
   if (is_write()) {
     record_lock.Unlock(current_key());
   }
 
-  DoBinlog();
 
   uint64_t end_us = pstd::NowMicros();
   this->binlog_duration_ms = (end_us - before_do_binlog_us) / 1000;
@@ -963,52 +960,8 @@ bool Cmd::DoReadCommandInCache() {
 
 
 void Cmd::DoBinlog() {
-  // Skip binlog entirely if we're in on_apply context
-  if (pika_raft::g_in_raft_apply) {
-    return;
-  }
-  
-  // Check if Raft is enabled
-  auto raft_manager = g_pika_server->GetRaftManager();
-  if (raft_manager && is_write() && res().ok()) {
-    // Plan A: Submit command to Raft using Redis protocol
-    // Skip if we're already in on_apply context (avoid recursion)
-    std::string redis_proto = ToRedisProtocol();
-    
-    // Prepend db_name to Redis protocol for extraction in on_apply
-    std::string log_data = db_name_ + "|" + redis_proto;
-    
-    
-    // Create promise/future for synchronous waiting
-    std::promise<rocksdb::Status> promise;
-    auto future = promise.get_future();
-    
-    // Submit to Raft
-    auto status = raft_manager->SubmitCommandWithPromise(
-        db_name_, log_data, std::move(promise));
-    
-    if (!status.ok()) {
-      LOG(ERROR) << "Failed to submit command to Raft: " << status.ToString();
-      res_.SetRes(CmdRes::kErrOther, "Raft submit failed: " + status.ToString());
-      return;
-    }
-    
-    // Wait for Raft to apply (with 10 second timeout)
-    auto wait_status = future.wait_for(std::chrono::seconds(10));
-    if (wait_status == std::future_status::timeout) {
-      LOG(ERROR) << "Raft apply timeout for command: " << name_;
-      res_.SetRes(CmdRes::kErrOther, "Raft apply timeout");
-      return;
-    }
-    
-    // Get the result
-    rocksdb::Status raft_result = future.get();
-    if (!raft_result.ok()) {
-      LOG(ERROR) << "Raft apply failed: " << raft_result.ToString();
-      res_.SetRes(CmdRes::kErrOther, "Raft apply failed: " + raft_result.ToString());
-      return;
-    }
-    
+  // 如果是 Raft 模式，跳过写 binlog（改用 Protobuf binlog）
+  if (g_pika_server->GetRaftManager()) {
     return;
   }
 

src/pika_kv.cc

@@ -11,12 +11,9 @@
 
 #include "include/pika_cache.h"
 #include "include/pika_conf.h"
-#include "include/pika_raft.h"
-#include "include/pika_server.h"
 #include "include/pika_slot_command.h"
 
 extern std::unique_ptr<PikaConf> g_pika_conf;
-extern PikaServer* g_pika_server;
 /* SET key value [NX] [XX] [EX <seconds>] [PX <milliseconds>] */
 void SetCmd::DoInitial() {
   if (!CheckArg(argv_.size())) {
@@ -68,17 +65,6 @@ void SetCmd::DoInitial() {
 }
 
 void SetCmd::Do() {
-  // Plan A: If Raft is enabled, skip actual write on first call
-  // The write will happen when on_apply executes this command
-  // Use thread-local flag to detect if we're in on_apply context
-  if (g_pika_server->GetRaftManager() && !pika_raft::g_in_raft_apply) {
-    // Raft mode: First call from client, skip write
-    // Set OK response, actual write will happen in on_apply
-    res_.SetRes(CmdRes::kOk);
-    return;
-  }
-  
-  // Normal path: Either non-Raft mode or called from on_apply
   int32_t res = 1;
   STAGE_TIMER_GUARD(storage_duration_ms, true);
   switch (condition_) {

src/pika_raft.cc

@@ -94,9 +94,17 @@ void RaftClusterCmd::Do() {
                 << " with peers: " << argv_[2];
       status = raft_mgr->InitCluster(db_name_, args_);
       if (status.ok()) {
-        // Raft 模式下关闭 RocksDB WAL（Raft Log 已提供持久化）
-        db_->storage()->DisableWal(true);
-        LOG(INFO) << "Disabled RocksDB WAL for Raft mode (DB: " << db_name_ << ")";
+        // Update raft-peers in config file and persist
+        std::string peers_str = argv_[2];
+        if (g_pika_conf->SetConfStr("raft-peers", peers_str)) {
+          if (g_pika_conf->WriteBack()) {
+            LOG(INFO) << "Updated raft-peers in config file: " << peers_str;
+          } else {
+            LOG(WARNING) << "Failed to write raft-peers to config file";
+          }
+        } else {
+          LOG(WARNING) << "Failed to update raft-peers in config";
+        }
         res_.AppendStringRaw("+OK\r\n");
       } else {
         res_.SetRes(CmdRes::kErrOther, "Failed to initialize cluster: " + status.ToString());
@@ -109,9 +117,6 @@ void RaftClusterCmd::Do() {
                 << " to leader: " << args_[0];
       status = raft_mgr->JoinCluster(db_name_, args_[0]);
       if (status.ok()) {
-        // Raft 模式下关闭 RocksDB WAL（Raft Log 已提供持久化）
-        db_->storage()->DisableWal(true);
-        LOG(INFO) << "Disabled RocksDB WAL for Raft mode (DB: " << db_name_ << ")";
         res_.AppendStringRaw("+OK\r\n");
       } else {
         res_.SetRes(CmdRes::kErrOther, "Failed to join cluster: " + status.ToString());
@@ -123,7 +128,6 @@ void RaftClusterCmd::Do() {
       std::string info;
       status = raft_mgr->GetClusterInfo(db_name_, &info);
       if (status.ok()) {
-        // 拆分成多行，以数组格式返回
         std::vector<std::string> lines;
         std::stringstream ss(info);
         std::string line;
@@ -200,6 +204,22 @@ void RaftNodeCmd::Do() {
                 << ", peer: " << peer_addr_;
       status = raft_mgr->AddNode(db_name_, peer_addr_);
       if (status.ok()) {
+        // Update raft-peers in config file
+        std::string current_peers = g_pika_conf->raft_peers();
+        if (current_peers.empty()) {
+          current_peers = peer_addr_;
+        } else if (current_peers.find(peer_addr_) == std::string::npos) {
+          current_peers += "," + peer_addr_;
+        }
+        
+        if (g_pika_conf->SetConfStr("raft-peers", current_peers)) {
+          if (g_pika_conf->WriteBack()) {
+            LOG(INFO) << "Updated raft-peers in config file: " << current_peers;
+          } else {
+            LOG(WARNING) << "Failed to write raft-peers to config file";
+          }
+        }
+        
         res_.AppendStringRaw("+OK\r\n");
       } else {
         res_.SetRes(CmdRes::kErrOther, "Failed to add node: " + status.ToString());
@@ -212,6 +232,39 @@ void RaftNodeCmd::Do() {
                 << ", peer: " << peer_addr_;
       status = raft_mgr->RemoveNode(db_name_, peer_addr_);
       if (status.ok()) {
+        // Update raft-peers in config file
+        std::string current_peers = g_pika_conf->raft_peers();
+        if (!current_peers.empty()) {
+          // Parse and rebuild peer list without the removed peer
+          std::vector<std::string> peer_list;
+          std::stringstream ss(current_peers);
+          std::string peer;
+          while (std::getline(ss, peer, ',')) {
+            peer.erase(0, peer.find_first_not_of(" \t"));
+            peer.erase(peer.find_last_not_of(" \t") + 1);
+            if (!peer.empty() && peer != peer_addr_) {
+              peer_list.push_back(peer);
+            }
+          }
+          
+          // Rebuild peers string
+          std::string new_peers;
+          for (size_t i = 0; i < peer_list.size(); i++) {
+            new_peers += peer_list[i];
+            if (i < peer_list.size() - 1) {
+              new_peers += ",";
+            }
+          }
+          
+          if (g_pika_conf->SetConfStr("raft-peers", new_peers)) {
+            if (g_pika_conf->WriteBack()) {
+              LOG(INFO) << "Updated raft-peers in config file: " << new_peers;
+            } else {
+              LOG(WARNING) << "Failed to write raft-peers to config file";
+            }
+          }
+        }
+        
         res_.AppendStringRaw("+OK\r\n");
       } else {
         res_.SetRes(CmdRes::kErrOther, "Failed to remove node: " + status.ToString());

src/pika_server.cc

@@ -114,6 +114,15 @@ PikaServer::PikaServer()
       LOG(FATAL) << "Failed to initialize Raft manager: " << status.ToString();
     }
     LOG(INFO) << "Raft manager initialized successfully";
+    
+    std::lock_guard rwl(storage_options_rw_);
+    storage_options_.append_log_function = 
+      [this](const ::pikiwidb::Binlog& binlog, std::promise<rocksdb::Status>&& promise) {
+        std::string db_name = "db0";
+        
+        raft_manager_->AppendLog(db_name, binlog, std::move(promise));
+      };
+    LOG(INFO) << "Raft append_log_function registered in storage_options";
   }
 
   bgsave_thread_.set_thread_name("PikaServer::bgsave_thread_");
@@ -237,58 +246,13 @@ void PikaServer::Start() {
     } else {
       LOG(INFO) << "Raft manager started successfully";
 
-      // 为每个数据库注册 binlog 回调
+      // Disable WAL for all databases when Raft is enabled
       for (const auto& db_item : dbs_) {
-        std::string db_name = db_item.first;
-        auto storage = db_item.second->storage();  // Returns std::shared_ptr<storage::Storage>
-        
-        if (!storage) {
-          LOG(WARNING) << "数据库 " << db_name << " 的 storage 为空，跳过回调注册";
-          continue;
+        auto storage = db_item.second->storage();
+        if (storage) {
+          storage->DisableWal(true);
+          LOG(INFO) << "Disabled WAL for DB: " << db_item.first;
         }
-        
-        // 设置存储引擎引用 (使用原始指针)
-        raft_manager_->SetStorage(storage.get());
-        
-        // 注册 binlog 回调（使用 promise/future 同步）
-        storage->SetBinlogWriteCallback(
-          [this, db_name](const pikiwidb::Binlog& binlog, std::promise<rocksdb::Status>&& promise) {
-            // 序列化 binlog
-            std::string binlog_data;
-            if (!binlog.SerializeToString(&binlog_data)) {
-              LOG(ERROR) << "Failed to serialize binlog";
-              promise.set_value(rocksdb::Status::Corruption("Binlog serialization failed"));
-              return;
-            }
-            
-            LOG(INFO) << "收到 binlog 回调，数据库: " << db_name 
-                      << ", binlog 大小: " << binlog_data.size() << " 字节"
-                      << ", entries: " << binlog.entries_size();
-            
-            // 创建异步回调闭包
-            auto* closure = new pika_raft::WriteDoneClosure(nullptr, nullptr);
-            closure->SetBinlogData(binlog_data);
-            
-            // TODO: 这里需要将 promise 传递给 closure，让 Raft apply 完成后设置结果
-            // 目前先立即返回 OK，实际应该等待 Raft 应用完成
-            
-            // 提交到 Raft
-            auto apply_status = raft_manager_->ApplyBinlog(db_name, binlog_data, closure);
-            if (!apply_status.ok()) {
-              LOG(ERROR) << "提交 binlog 到 Raft 失败: " << apply_status.ToString();
-              // 设置错误状态并调用 Run，让 closure 自己清理
-              closure->status().set_error(-1, "%s", apply_status.ToString().c_str());
-              closure->Run();
-              promise.set_value(rocksdb::Status::IOError(apply_status.ToString()));
-            } else {
-              // TODO: 应该在 Raft apply 完成后才 set_value
-              // 目前先简单地立即返回 OK
-              promise.set_value(rocksdb::Status::OK());
-            }
-          }
-        );
-        
-        LOG(INFO) << "已为数据库 " << db_name << " 注册 binlog 回调";
       }
     }
   }