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

@@ -897,17 +897,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 +962,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

@@ -68,17 +68,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_server.cc

@@ -114,6 +114,21 @@ PikaServer::PikaServer()
       LOG(FATAL) << "Failed to initialize Raft manager: " << status.ToString();
     }
     LOG(INFO) << "Raft manager initialized successfully";
+    
+    // 设置 Raft binlog 回调（pikiwidb_raft 风格）
+    // 这样后续创建的所有 DB 都会使用这个回调
+    std::lock_guard rwl(storage_options_rw_);
+    storage_options_.append_log_function = 
+      [this](const ::pikiwidb::Binlog& binlog, std::promise<rocksdb::Status>&& promise) {
+        // 使用 db_id 确定是哪个数据库（pikiwidb_raft 的实现方式）
+        // 这里简化处理：使用第一个 DB（"db0"）
+        // 更完整的实现需要根据 binlog.db_id() 查找对应的 DB
+        std::string db_name = "db0";  // TODO: 支持多 DB
+        
+        // 调用 RaftManager::AppendLog()
+        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 +252,23 @@ void PikaServer::Start() {
     } else {
       LOG(INFO) << "Raft manager started successfully";
 
-      // 为每个数据库注册 binlog 回调
+      // 为每个数据库设置 Storage 引用并禁用 RocksDB WAL（供 on_apply 使用）
       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>
+        auto storage = db_item.second->storage();
 
         if (!storage) {
-          LOG(WARNING) << "数据库 " << db_name << " 的 storage 为空，跳过回调注册";
+          LOG(WARNING) << "数据库 " << db_name << " 的 storage 为空，跳过";
           continue;
         }
 
+        storage->DisableWal(true);
+        LOG(INFO) << "已为数据库 " << db_name << " 禁用 RocksDB WAL（Raft 模式）";
+        
         // 设置存储引擎引用 (使用原始指针)
         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 回调";
+        LOG(INFO) << "已为数据库 " << db_name << " 设置 Storage 引用";
       }
     }
   }

src/praft/include/praft/praft.h

@@ -29,6 +29,10 @@ namespace storage {
 class Storage;
 }
 
+namespace pikiwidb {
+class Binlog;
+}
+
 namespace net {
 class RedisConn;
 }
@@ -38,21 +42,6 @@ class Cmd;
 namespace pika_raft {
 
 // Raft log entry data structure
-struct RaftLogEntry {
-  std::string cmd_name;
-  std::vector<std::string> args;
-  std::string db_name;
-  int64_t timestamp;
-  
-  RaftLogEntry() : timestamp(0) {}
-  
-  // Serialize to string
-  std::string SerializeAsString() const;
-  
-  // Deserialize from string
-  bool ParseFromString(const std::string& data);
-};
-
 // Write done closure for asynchronous Raft callback
 class WriteDoneClosure : public braft::Closure {
  public:
@@ -137,9 +126,6 @@ class PikaRaftNode {
   // Remove peer from the cluster
   pstd::Status RemovePeer(const braft::PeerId& peer);
 
-  // Apply a command to Raft
-  pstd::Status Apply(const RaftLogEntry& entry);
-
   // Get cluster status information
   void GetStatus(std::string* status_str);
 
@@ -192,23 +178,17 @@ class RaftManager {
   // Check if Raft is enabled for a specific DB
   bool IsRaftEnabled(const std::string& db_name) const;
 
-  // Apply a command through Raft
-  pstd::Status ApplyCommand(const std::string& db_name, const RaftLogEntry& entry);
-
   // Apply binlog to Raft (called by storage callback)
   pstd::Status ApplyBinlog(const std::string& db_name, 
-                          const std::string& binlog_data,
-                          WriteDoneClosure* done);
+                           const std::string& binlog_data, 
+                           WriteDoneClosure* done);
 
-  // Submit command to Raft with promise (for synchronous waiting)
-  pstd::Status SubmitCommandWithPromise(const std::string& db_name,
-                                        const std::string& log_data,
-                                        std::promise<rocksdb::Status>&& promise);
+  // Append binlog (pikiwidb_raft 风格，直接接收 Binlog + promise)
+  void AppendLog(const std::string& db_name, 
+                 const ::pikiwidb::Binlog& log, 
+                 std::promise<rocksdb::Status>&& promise);
 
-  // Apply command from Redis protocol (called in on_apply)
-  rocksdb::Status ApplyCommandFromRedisProtocol(const std::string& redis_proto_data,
-                                                  const std::string& db_name);
-
+  // Submit command to Raft with promise (for synchronous waiting)
   // Get Raft node for a specific DB
   std::shared_ptr<PikaRaftNode> GetRaftNode(const std::string& db_name);
 

src/praft/src/binlog.proto

@@ -7,16 +7,6 @@ package pikiwidb;
 
 option optimize_for = LITE_RUNTIME;
 
-// 数据类型枚举
-enum DataType {
-  kStrings = 0;
-  kHashes = 1;
-  kLists = 2;
-  kSets = 3;
-  kZSets = 4;
-  kStreams = 5;
-}
-
 // 操作类型
 enum OperateType {
   kNoOperate = 0;
@@ -26,28 +16,15 @@ enum OperateType {
 
 // Binlog 条目（对应单个 RocksDB 操作）
 message BinlogEntry {
-  DataType data_type = 1;       // 数据类型（对应哪个 RocksDB）
+  uint32 cf_idx = 1;            // 列族索引 (column family index)
   OperateType op_type = 2;      // 操作类型
   bytes key = 3;                // 已编码的 key
   optional bytes value = 4;     // 已编码的 value（包含 TTL、version 等）
-  
-  // 用于日志恢复的元信息
-  uint64 timestamp = 5;         // 操作时间戳
-  optional uint64 ttl = 6;      // TTL（秒），0 表示永久
 }
 
 // Binlog（对应一次 Raft 日志提交）
 message Binlog {
-  uint32 db_id = 1;             // 数据库 ID
-  uint32 slot_idx = 2;          // 槽位索引（预留）
-  
-  // Raft 日志索引（对应 LogicOffset）
-  uint32 term = 3;              // Raft term
-  uint64 log_index = 4;         // Raft log index
-  
-  // Binlog 文件位置（对应 BinlogOffset）
-  uint32 filenum = 5;           // Binlog 文件编号
-  uint64 offset = 6;            // Binlog 文件内偏移量
-  
-  repeated BinlogEntry entries = 7;  // 批量操作条目
+  uint32 db_id = 1;                  // 数据库 ID
+  uint32 slot_idx = 2;               // 槽位索引（预留）
+  repeated BinlogEntry entries = 3;  // 批量操作条目
 }