Enhance asynchronous performance optimization

Author: wuxianrong

conf/pika.conf

@@ -647,12 +647,6 @@ raft-enabled : no
 # All nodes in the same Raft cluster should have the same group-id
 raft-group-id : pika_raft_group
 
-# Initial Raft peers (comma separated, e.g., 127.0.0.1:12221,127.0.0.1:12222)
-# Format: ip:port,ip:port,ip:port
-# The port should be: pika_port + 3000 (e.g., if pika port is 9221, raft port is 12221)
-# Leave empty if initializing a single-node cluster
-raft-peers : 
-
 # Raft election timeout in milliseconds
 # This is the time to wait before starting a new election if no heartbeat is received
 # A larger value reduces the chance of unnecessary elections but increases failover time

include/pika_command.h

@@ -70,7 +70,6 @@ const std::string kCmdNameClearCache = "clearcache";
 // Raft commands
 const std::string kCmdNameRaftCluster = "raft.cluster";
 const std::string kCmdNameRaftNode = "raft.node";
-const std::string kCmdNameRaftConfig = "raft.config";
 
 // Migrate slot
 const std::string kCmdNameSlotsMgrtSlot = "slotsmgrtslot";

include/pika_conf.h

@@ -893,10 +893,6 @@ class PikaConf : public pstd::BaseConf {
     std::shared_lock l(rwlock_);
     return raft_group_id_;
   }
-  std::string raft_peers() {
-    std::shared_lock l(rwlock_);
-    return raft_peers_;
-  }
   int raft_election_timeout_ms() {
     std::shared_lock l(rwlock_);
     return raft_election_timeout_ms_;
@@ -1092,7 +1088,6 @@ class PikaConf : public pstd::BaseConf {
   // Raft configuration
   bool raft_enabled_ = false;
   std::string raft_group_id_;
-  std::string raft_peers_;
   int raft_election_timeout_ms_ = 1000;
   int raft_snapshot_interval_s_ = 3600;
 };

include/pika_raft.h

@@ -21,10 +21,9 @@
  * Raft Commands
  */
 
-// RAFT.CLUSTER [INIT|JOIN|INFO] [args...]
-// INIT: RAFT.CLUSTER INIT peer1,peer2,peer3
-// JOIN: RAFT.CLUSTER JOIN leader_address
-// INFO: RAFT.CLUSTER INFO
+// RAFT.CLUSTER [INIT|INFO] [args...]
+// INIT: RAFT.CLUSTER INIT [peer1,peer2,peer3]  (peers optional, no peers = prepare for cluster expansion)
+// INFO: RAFT.CLUSTER INFO [db_name]
 class RaftClusterCmd : public Cmd {
  public:
   RaftClusterCmd(const std::string& name, int arity, uint32_t flag)
@@ -36,7 +35,7 @@ class RaftClusterCmd : public Cmd {
   Cmd* Clone() override { return new RaftClusterCmd(*this); }
 
  private:
-  enum class Operation { INIT, JOIN, INFO, UNKNOWN };
+  enum class Operation { INIT, INFO, UNKNOWN };
 
   void DoInitial() override;
   void Clear() override {
@@ -76,33 +75,5 @@ class RaftNodeCmd : public Cmd {
   std::string db_name_;
 };
 
-// RAFT.CONFIG GET|SET key [value] [db_name]
-class RaftConfigCmd : public Cmd {
- public:
-  RaftConfigCmd(const std::string& name, int arity, uint32_t flag)
-      : Cmd(name, arity, flag, static_cast<uint32_t>(AclCategory::ADMIN)) {}
-  
-  void Do() override;
-  void Split(const HintKeys& hint_keys) override {};
-  void Merge() override {};
-  Cmd* Clone() override { return new RaftConfigCmd(*this); }
-
- private:
-  enum class Operation { GET, SET, UNKNOWN };
-  
-  void DoInitial() override;
-  void Clear() override {
-    operation_ = Operation::UNKNOWN;
-    config_key_.clear();
-    config_value_.clear();
-    db_name_.clear();
-  }
-  
-  Operation operation_;
-  std::string config_key_;
-  std::string config_value_;
-  std::string db_name_;
-};
-
 #endif  // PIKA_RAFT_H_
 

src/pika_command.cc

@@ -168,10 +168,6 @@ void InitCmdTable(CmdTable* cmd_table) {
       std::make_unique<RaftNodeCmd>(kCmdNameRaftNode, -3, kCmdFlagsWrite | kCmdFlagsAdmin | kCmdFlagsSlow);
   cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNameRaftNode, std::move(raftnodeptr)));
 
-  std::unique_ptr<Cmd> raftconfigptr =
-      std::make_unique<RaftConfigCmd>(kCmdNameRaftConfig, -2, kCmdFlagsRead | kCmdFlagsAdmin | kCmdFlagsSlow);
-  cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNameRaftConfig, std::move(raftconfigptr)));
-
 #ifdef WITH_COMMAND_DOCS
   std::unique_ptr<Cmd> commandptr =
       std::make_unique<CommandCmd>(kCmdNameCommand, -1, kCmdFlagsRead | kCmdFlagsAdmin | kCmdFlagsSlow);

src/pika_conf.cc

@@ -713,8 +713,6 @@ int PikaConf::Load() {
     raft_group_id_ = "pika_raft_group";
   }
 
-  GetConfStr("raft-peers", &raft_peers_);
-  
   GetConfInt("raft-election-timeout-ms", &raft_election_timeout_ms_);
   if (raft_election_timeout_ms_ <= 0) {
     raft_election_timeout_ms_ = 1000;

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_);