Enhance asynchronous performance optimization

Author: wuxianrong

src/pika_kv.cc

@@ -12,8 +12,11 @@
 #include "include/pika_cache.h"
 #include "include/pika_conf.h"
 #include "include/pika_slot_command.h"
+#include "include/pika_server.h"
+#include "praft/praft.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())) {
@@ -67,24 +70,74 @@ void SetCmd::DoInitial() {
 void SetCmd::Do() {
   int32_t res = 1;
   STAGE_TIMER_GUARD(storage_duration_ms, true);
+  
+  bool is_raft_leader = false;
+  if (g_pika_server && g_pika_server->GetRaftManager()) {
+    auto node = g_pika_server->GetRaftManager()->GetRaftNode(db_->GetDBName());
+    is_raft_leader = (node && node->IsLeader());
+  }
+  
+  storage::CommitCallback callback = nullptr;
+  
+  if (is_raft_leader && db_->storage() && db_->storage()->IsRaftEnabled()) {
+    auto self = std::static_pointer_cast<SetCmd>(shared_from_this());
+    auto resp_ptr = std::make_shared<std::string>();
+    
+    auto pika_conn = std::dynamic_pointer_cast<PikaClientConn>(GetConn());
+    if (!pika_conn) {
+      res_.SetRes(CmdRes::kErrOther, "Invalid connection");
+      return;
+    }
+    
+    callback = [self, resp_ptr, pika_conn](rocksdb::Status status) {
+      int32_t result = (status.ok() || status.IsNotFound()) ? 1 : 0;
+      
+      if (status.ok() || status.IsNotFound()) {
+        if (self->condition_ == SetCmd::kVX) {
+          self->res_.AppendInteger(self->success_);
+        } else {
+          if (result == 1) {
+            self->res_.SetRes(CmdRes::kOk);
+            AddSlotKey("k", self->key_, self->db_);
+          } else {
+            self->res_.AppendStringLen(-1);
+          }
+        }
+      } else {
+        self->res_.SetRes(CmdRes::kErrOther, status.ToString());
+      }
+      
+      *resp_ptr = std::move(self->res_.message());
+      pika_conn->WriteResp(*resp_ptr);
+      pika_conn->NotifyEpoll(true);
+    };
+  }
+  
+  // Call storage layer with optional callback
   switch (condition_) {
     case SetCmd::kXX:
-      s_ = db_->storage()->Setxx(key_, value_, &res, static_cast<int32_t>(sec_));
+      s_ = db_->storage()->Setxx(key_, value_, &res, static_cast<int32_t>(sec_), callback);
       break;
     case SetCmd::kNX:
-      s_ = db_->storage()->Setnx(key_, value_, &res, static_cast<int32_t>(sec_));
+      s_ = db_->storage()->Setnx(key_, value_, &res, static_cast<int32_t>(sec_), callback);
       break;
     case SetCmd::kVX:
       s_ = db_->storage()->Setvx(key_, target_, value_, &success_, static_cast<int32_t>(sec_));
       break;
     case SetCmd::kEXORPX:
-      s_ = db_->storage()->Setex(key_, value_, static_cast<int32_t>(sec_));
+      s_ = db_->storage()->Setex(key_, value_, static_cast<int32_t>(sec_), callback);
       break;
     default:
-      s_ = db_->storage()->Set(key_, value_);
+      s_ = db_->storage()->Set(key_, value_, callback);
       break;
   }
 
+  // For async mode (Leader), response is handled by callback
+  if (is_raft_leader && db_->storage() && db_->storage()->IsRaftEnabled()) {
+    return;
+  }
+  
+  // For sync mode (non-Leader or non-Raft), set response immediately
   if (s_.ok() || s_.IsNotFound()) {
     if (condition_ == SetCmd::kVX) {
       res_.AppendInteger(success_);

src/pika_raft.cc

@@ -204,21 +204,9 @@ 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";
-          }
-        }
+        // Don't modify config file - braft manages cluster membership in raft_meta
+        // The raft-peers in config file is only used for initial bootstrap
+        LOG(INFO) << "Node added successfully to Raft cluster (managed by braft raft_meta)";
 
         res_.AppendStringRaw("+OK\r\n");
       } else {
@@ -232,38 +220,9 @@ 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";
-            }
-          }
-        }
+        // Don't modify config file - braft manages cluster membership in raft_meta
+        // The raft-peers in config file is only used for initial bootstrap
+        LOG(INFO) << "Node removed successfully from Raft cluster (managed by braft raft_meta)";
 
         res_.AppendStringRaw("+OK\r\n");
       } else {

src/pika_server.cc

@@ -117,10 +117,11 @@ PikaServer::PikaServer()
 
     std::lock_guard rwl(storage_options_rw_);
     storage_options_.append_log_function = 
-      [this](const ::pikiwidb::Binlog& binlog, std::promise<rocksdb::Status>&& promise) {
+      [this](const ::pikiwidb::Binlog& binlog, std::promise<rocksdb::Status>&& promise,
+             storage::CommitCallback callback) {
         std::string db_name = "db0";
 
-        raft_manager_->AppendLog(db_name, binlog, std::move(promise));
+        raft_manager_->AppendLog(db_name, binlog, std::move(promise), callback);
       };
     LOG(INFO) << "Raft append_log_function registered in storage_options";
   }

src/praft/include/praft/praft.h

@@ -21,8 +21,10 @@
 #include "pstd/include/pstd_mutex.h"
 #include "pstd/include/pstd_status.h"
 #include "rocksdb/status.h"
+#include "storage/batch.h"
 
 class PikaServer;
+class Cmd;
 
 // Forward declarations
 namespace storage {
@@ -33,6 +35,10 @@ namespace pikiwidb {
 class Binlog;
 }
 
+namespace net {
+class NetConn;
+}
+
 namespace pika_raft {
 
 // Raft log entry data structure
@@ -44,13 +50,22 @@ class WriteDoneClosure : public braft::Closure {
 
   void Run() override;
 
-  // Set promise for synchronous Raft apply
+  // Set promise for synchronous Raft apply (used in Follower on_apply)
   void SetPromise(std::shared_ptr<std::promise<rocksdb::Status>> p) {
     promise_ = p;
   }
+  
+  // Set callback for async response (used in Leader)
+  void SetCallback(storage::CommitCallback callback) {
+    callback_ = callback;
+  }
 
  private:
+  // For synchronous mode (Follower)
   std::shared_ptr<std::promise<rocksdb::Status>> promise_;
+  
+  // For asynchronous mode (Leader)
+  storage::CommitCallback callback_;
 };
 
 // Pika state machine implementation
@@ -162,10 +177,12 @@ class RaftManager {
   // Get cluster information
   pstd::Status GetClusterInfo(const std::string& db_name, std::string* info);
 
-  // Append binlog
+  // Append binlog (supports both sync and async modes)
+  // Sync mode: pass promise, async mode: pass callback
   void AppendLog(const std::string& db_name, 
                  const ::pikiwidb::Binlog& log, 
-                 std::promise<rocksdb::Status>&& promise);
+                 std::promise<rocksdb::Status>&& promise,
+                 storage::CommitCallback callback = nullptr);
 
   // Get Raft node for a specific DB
   std::shared_ptr<PikaRaftNode> GetRaftNode(const std::string& db_name);

src/praft/src/binlog.proto

@@ -14,12 +14,24 @@ enum OperateType {
   kDelete = 2;
 }
 
+// 数据类型（与 storage::DataType 对应）
+enum DataType {
+  kAll = 0;
+  kStrings = 1;
+  kHashes = 2;
+  kLists = 3;
+  kZSets = 4;
+  kSets = 5;
+  kStreams = 6;
+}
+
 // Binlog 条目（对应单个 RocksDB 操作）
 message BinlogEntry {
-  uint32 cf_idx = 1;            // 列族索引 (column family index)
-  OperateType op_type = 2;      // 操作类型
-  bytes key = 3;                // 已编码的 key
-  optional bytes value = 4;     // 已编码的 value（包含 TTL、version 等）
+  DataType data_type = 1;       // 数据类型（用于定位对应的 RocksDB 实例）
+  uint32 cf_idx = 2;            // 列族索引 (column family index, 0=meta/default, 1=data)
+  OperateType op_type = 3;      // 操作类型
+  bytes key = 4;                // 已编码的 key
+  optional bytes value = 5;     // 已编码的 value（包含 TTL、version 等）
 }
 
 // Binlog（对应一次 Raft 日志提交）