braft Performance optimization

Author: wuxianrong

conf/pika.conf

@@ -656,11 +656,35 @@ 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
-# Default value is 1000ms (1 second)
-raft-election-timeout-ms : 1000
+# Optimized default: 300ms (for faster failover, was 1000ms)
+# Range: 100-5000ms, Recommended: 300-500ms for LAN
+raft-election-timeout-ms : 300
 
 # Raft snapshot interval in seconds
 # This determines how often Raft takes snapshots of the state machine
 # Snapshots are used to compact the log and speed up node recovery
 # Default value is 3600 seconds (1 hour)
 raft-snapshot-interval-s : 3600
+
+## Raft Performance Tuning (批处理性能优化)
+
+# Enable request batching for better throughput
+# When enabled, multiple client requests are batched into a single Raft log entry
+# This significantly reduces Raft consensus overhead
+# Default: yes (enabled for optimal performance)
+raft-enable-batching : yes
+
+# Batch timeout in milliseconds
+# How long to wait before flushing a batch (if max-batch-size not reached)
+# Lower values = lower latency but less batching benefit
+# Higher values = more batching but higher P99 latency
+# Range: 1-10ms, Recommended: 2-5ms
+# Default: 2ms
+raft-batch-timeout-ms : 2
+
+# Maximum number of requests in a single batch
+# When this limit is reached, the batch is flushed immediately
+# Higher values = more batching but higher memory usage
+# Range: 100-10000, Recommended: 500-2000
+# Default: 1000
+raft-max-batch-size : 1000

include/pika_conf.h

@@ -906,6 +906,20 @@ class PikaConf : public pstd::BaseConf {
     return raft_snapshot_interval_s_;
   }
 
+  // Raft 性能优化相关配置
+  bool raft_enable_batching() {
+    std::shared_lock l(rwlock_);
+    return raft_enable_batching_;
+  }
+  int raft_batch_timeout_ms() {
+    std::shared_lock l(rwlock_);
+    return raft_batch_timeout_ms_;
+  }
+  int raft_max_batch_size() {
+    std::shared_lock l(rwlock_);
+    return raft_max_batch_size_;
+  }
+  
   int Load();
   int ConfigRewrite();
   int ConfigRewriteSlaveOf();
@@ -1093,8 +1107,13 @@ class PikaConf : public pstd::BaseConf {
   bool raft_enabled_ = false;
   std::string raft_group_id_;
   std::string raft_peers_;
-  int raft_election_timeout_ms_ = 1000;
+  int raft_election_timeout_ms_ = 300;  // 优化后的默认值：300ms
   int raft_snapshot_interval_s_ = 3600;
+  
+  // Raft 性能优化配置
+  bool raft_enable_batching_ = true;    // 启用批处理（默认开启）
+  int raft_batch_timeout_ms_ = 2;       // 批处理超时：2ms
+  int raft_max_batch_size_ = 1000;      // 最大批量大小：1000
 };
 
 #endif

src/pika_conf.cc

@@ -717,13 +717,26 @@ int PikaConf::Load() {
 
   GetConfInt("raft-election-timeout-ms", &raft_election_timeout_ms_);
   if (raft_election_timeout_ms_ <= 0) {
-    raft_election_timeout_ms_ = 1000;
+    raft_election_timeout_ms_ = 300;  // 优化后的默认值
   }
 
   GetConfInt("raft-snapshot-interval-s", &raft_snapshot_interval_s_);
   if (raft_snapshot_interval_s_ <= 0) {
     raft_snapshot_interval_s_ = 3600;
   }
+  
+  // Raft 性能优化配置
+  GetConfBool("raft-enable-batching", &raft_enable_batching_);
+  
+  GetConfInt("raft-batch-timeout-ms", &raft_batch_timeout_ms_);
+  if (raft_batch_timeout_ms_ <= 0) {
+    raft_batch_timeout_ms_ = 2;  // 默认 2ms
+  }
+  
+  GetConfInt("raft-max-batch-size", &raft_max_batch_size_);
+  if (raft_max_batch_size_ <= 0) {
+    raft_max_batch_size_ = 1000;  // 默认 1000
+  }
 
   return ret;
 }

src/praft/CMakeLists.txt

@@ -21,6 +21,11 @@ add_dependencies(binlog_pb protobuf)
 # Collect all source files (excluding binlog.proto which is in binlog_pb)
 aux_source_directory(./src DIR_SRCS)
 
+# Add batch_manager source files explicitly
+set(DIR_SRCS ${DIR_SRCS} 
+    ${CMAKE_CURRENT_SOURCE_DIR}/batch_manager.cc
+)
+
 # Create static library (WITHOUT binlog protobuf sources, link binlog_pb instead)
 add_library(praft STATIC ${DIR_SRCS})
 

src/praft/batch_manager.cc

@@ -0,0 +1,191 @@
+// Copyright (c) 2024-present, Qihoo, Inc.  All rights reserved.
+// This source code is licensed under the BSD-style license found in the
+// LICENSE file in the root directory of this source tree.
+
+#include "batch_manager.h"
+#include <glog/logging.h>
+#include "pstd/include/pstd_defer.h"
+
+namespace pika_raft {
+
+// BatchClosure 实现
+void BatchClosure::Run() {
+  std::unique_ptr<BatchClosure> self_guard(this);
+  
+  // 根据 Raft 结果通知所有 promise
+  if (status().ok()) {
+    for (auto& promise : promises_) {
+      if (promise) {
+        promise->set_value(rocksdb::Status::OK());
+      }
+    }
+  } else {
+    rocksdb::Status error_status = rocksdb::Status::IOError(status().error_str());
+    for (auto& promise : promises_) {
+      if (promise) {
+        promise->set_value(error_status);
+      }
+    }
+  }
+}
+
+// BatchManager 实现
+BatchManager::BatchManager(braft::Node* raft_node, const BatchConfig& config)
+    : raft_node_(raft_node), config_(config) {
+  pending_items_.reserve(config_.max_batch_size);
+}
+
+BatchManager::~BatchManager() {
+  Stop();
+}
+
+void BatchManager::Submit(const std::string& log_data,
+                          std::shared_ptr<std::promise<rocksdb::Status>> promise) {
+  if (!config_.enable_batching) {
+    // 批处理未启用，直接提交
+    auto* closure = new BatchClosure({promise});
+    braft::Task task;
+    butil::IOBuf buf;
+    buf.append(log_data);
+    task.data = &buf;
+    task.done = closure;
+    raft_node_->apply(task);
+    return;
+  }
+  
+  std::lock_guard<std::mutex> lock(mutex_);
+  
+  BatchItem item;
+  item.log_data = log_data;
+  item.promise = promise;
+  item.submit_time_us = butil::gettimeofday_us();
+  
+  pending_items_.push_back(std::move(item));
+  total_requests_.fetch_add(1, std::memory_order_relaxed);
+  
+  // 如果队列满了，立即通知工作线程
+  if (pending_items_.size() >= config_.max_batch_size) {
+    cv_.notify_one();
+  }
+}
+
+void BatchManager::Start() {
+  if (running_.exchange(true)) {
+    return;  // 已经在运行
+  }
+  
+  batch_thread_ = std::thread([this]() { BatchWorker(); });
+  LOG(INFO) << "BatchManager started with config: timeout=" << config_.batch_timeout_ms 
+            << "ms, max_batch_size=" << config_.max_batch_size;
+}
+
+void BatchManager::Stop() {
+  if (!running_.exchange(false)) {
+    return;  // 已经停止
+  }
+  
+  cv_.notify_one();
+  if (batch_thread_.joinable()) {
+    batch_thread_.join();
+  }
+  
+  LOG(INFO) << "BatchManager stopped. Total requests: " << total_requests_.load()
+            << ", Total batches: " << total_batches_.load();
+}
+
+void BatchManager::BatchWorker() {
+  LOG(INFO) << "BatchManager worker thread started";
+  
+  while (running_.load(std::memory_order_relaxed)) {
+    std::unique_lock<std::mutex> lock(mutex_);
+    
+    // 等待有数据或超时
+    cv_.wait_for(lock, std::chrono::milliseconds(config_.batch_timeout_ms), 
+                 [this]() {
+                   return !pending_items_.empty() || 
+                          !running_.load(std::memory_order_relaxed);
+                 });
+    
+    if (!running_.load(std::memory_order_relaxed)) {
+      // 处理剩余请求
+      if (!pending_items_.empty()) {
+        FlushBatch();
+      }
+      break;
+    }
+    
+    if (!pending_items_.empty()) {
+      FlushBatch();
+    }
+  }
+  
+  LOG(INFO) << "BatchManager worker thread stopped";
+}
+
+void BatchManager::FlushBatch() {
+  // 假设调用者已持有锁
+  
+  if (pending_items_.empty()) {
+    return;
+  }
+  
+  size_t batch_size = pending_items_.size();
+  
+  // 合并所有日志数据
+  // 格式：<count>\n<log1_len>\n<log1_data><log2_len>\n<log2_data>...
+  std::string batch_data;
+  batch_data.reserve(batch_size * 100);  // 预估大小
+  
+  // 写入批次大小
+  batch_data.append(std::to_string(batch_size));
+  batch_data.append("\n");
+  
+  // 收集所有 promise
+  std::vector<std::shared_ptr<std::promise<rocksdb::Status>>> promises;
+  promises.reserve(batch_size);
+  
+  for (auto& item : pending_items_) {
+    // 写入单个日志长度和数据
+    batch_data.append(std::to_string(item.log_data.size()));
+    batch_data.append("\n");
+    batch_data.append(item.log_data);
+    
+    promises.push_back(std::move(item.promise));
+  }
+  
+  pending_items_.clear();
+  
+  // 创建批量 closure
+  auto* closure = new BatchClosure(std::move(promises));
+  
+  // 提交到 Raft
+  braft::Task task;
+  butil::IOBuf buf;
+  buf.append(batch_data);
+  task.data = &buf;
+  task.done = closure;
+  
+  raft_node_->apply(task);
+  
+  total_batches_.fetch_add(1, std::memory_order_relaxed);
+  
+  VLOG(1) << "Flushed batch of " << batch_size << " requests to Raft";
+}
+
+BatchManager::Stats BatchManager::GetStats() const {
+  Stats stats;
+  stats.total_requests = total_requests_.load(std::memory_order_relaxed);
+  stats.total_batches = total_batches_.load(std::memory_order_relaxed);
+  
+  if (stats.total_batches > 0) {
+    stats.avg_batch_size = static_cast<double>(stats.total_requests) / stats.total_batches;
+  }
+  
+  std::lock_guard<std::mutex> lock(const_cast<std::mutex&>(mutex_));
+  stats.pending_count = pending_items_.size();
+  
+  return stats;
+}
+
+} // namespace pika_raft
+