New Architecture

Author: YuCai18

include/pika_command_collector.h

@@ -36,7 +36,7 @@
   * 3. Send commands in batches to the consensus coordinator with batch-level synchronization
   * 4. Support asynchronous callback notification of command processing results
   * 5. Track performance metrics for batch processing
-  * 6. Provide intelligent retry mechanisms for failed batches
+
 */
 class PikaCommandCollector {
  public:
@@ -46,14 +46,13 @@ class PikaCommandCollector {
   /**
     * @brief constructor
     * @param coordinator consensus coordinator reference
-    * @param batch_size batch size (number of commands)
-    * @param batch_max_wait_time forced flush interval (milliseconds)
+    * @param batch_max_wait_time maximum wait time in milliseconds
   */
   // Constructor with raw pointer (original)
-  PikaCommandCollector(ConsensusCoordinator* coordinator, size_t batch_size = 100, int batch_max_wait_time = 5);
+  PikaCommandCollector(ConsensusCoordinator* coordinator, int batch_max_wait_time = 5);
 
   // Constructor with shared_ptr (for compatibility with make_shared calls)
-  PikaCommandCollector(std::shared_ptr<ConsensusCoordinator> coordinator, size_t batch_size = 100, int batch_max_wait_time = 5);
+  PikaCommandCollector(std::shared_ptr<ConsensusCoordinator> coordinator, int batch_max_wait_time = 5);
 
   ~PikaCommandCollector();
 
@@ -84,12 +83,6 @@ class PikaCommandCollector {
   */
   size_t PendingCommands() const;
 
-  /**
-    * @brief Set the batch size
-    * @param batch_size batch size
-  */
-  void SetBatchSize(size_t batch_size);
-  
   /**
     * @brief Set the batch max wait time
     * @param batch_max_wait_time maximum wait time in milliseconds
@@ -103,10 +96,6 @@ class PikaCommandCollector {
   std::pair<uint64_t, uint64_t> GetBatchStats() const;
 
   /**
-    * @brief Get average batch processing time in milliseconds
-    * @return Average processing time or nullopt if no batches processed
-  */
-  std::optional<double> GetAverageBatchTime() const;
 
  private:
 
@@ -131,55 +120,33 @@ class PikaCommandCollector {
   */
   void HandleConflict(const std::shared_ptr<Cmd>& cmd_ptr);
 
-  /**
-    * @brief Retry batch processing commands
-    * @param commands List of commands to retry
-    * @param callbacks Corresponding callback function list
-    * @param priority Priority level for the retry (higher means more urgent)
-    * @return Whether the commands were successfully requeued
-  */
-  bool RetryBatch(const std::vector<std::shared_ptr<Cmd>>& commands,
-                 const std::vector<CommandCallback>& callbacks,
-                 int priority = 100);
 
  private:
   //Consensus coordinator reference
   ConsensusCoordinator* coordinator_;
 
   // Batch processing configuration
-  std::atomic<size_t> batch_size_;
   std::atomic<int> batch_max_wait_time_;
+  std::chrono::time_point<std::chrono::steady_clock> batch_start_time_;
 
-  // Retry configuration
-  std::atomic<int> max_retry_attempts_{3};
-  std::atomic<int> retry_backoff_ms_{50};
-
   // Command collection and processing
   mutable std::mutex mutex_;
 
   // Pending command queue and callbacks
   std::list<std::pair<std::shared_ptr<Cmd>, CommandCallback>> pending_commands_;
 
-  // Priority queue for retries
-  std::deque<std::tuple<int, std::vector<std::shared_ptr<Cmd>>, std::vector<CommandCallback>>> retry_queue_;
-  
   // Command key mapping, used to handle same-key conflicts
   std::unordered_map<std::string, std::list<std::pair<std::shared_ptr<Cmd>, CommandCallback>>::iterator> key_map_;
 
   // Batch statistics
   std::atomic<uint64_t> total_processed_{0};
   std::atomic<uint64_t> total_batches_{0};
-  std::atomic<uint64_t> total_retries_{0};
   std::atomic<uint64_t> total_conflicts_{0};
-  std::atomic<uint64_t> total_batch_time_ms_{0};
-  std::chrono::time_point<std::chrono::steady_clock> batch_start_time_;
 
   // Performance tracking
   struct BatchMetrics {
     uint64_t batch_size;
-    uint64_t processing_time_ms;
     uint64_t wait_time_ms;
-    bool successful;
   };
 
   // Circular buffer for recent batch metrics

include/pika_command_queue.h

@@ -0,0 +1,87 @@
+// Copyright (c) 2015-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. An additional grant
+// of patent rights can be found in the PATENTS file in the same directory.
+
+#ifndef PIKA_COMMAND_QUEUE_H_
+#define PIKA_COMMAND_QUEUE_H_
+
+#include <memory>
+#include <queue>
+#include <vector>
+#include <string>
+#include <functional>
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
+
+#include "pstd/include/pstd_mutex.h"
+#include "pstd/include/env.h"
+#include "include/pika_command.h"
+#include "include/pika_define.h"
+#include "pstd/include/env.h"
+#include "include/pika_define.h"
+
+// Callback function type for command completion notification
+using CommandCallback = std::function<void(const LogOffset&, pstd::Status)>;
+
+// Structure representing a batch of commands
+struct CommandBatch {
+  std::vector<std::shared_ptr<Cmd>> commands;
+  std::vector<CommandCallback> callbacks;
+  uint64_t batch_id;
+  uint64_t create_time;
+  std::string db_name;
+  std::vector<LogOffset> binlog_offsets;  // Binlog offsets for each command
+  
+  CommandBatch(const std::vector<std::shared_ptr<Cmd>>& cmds,
+               const std::vector<CommandCallback>& cbs,
+               const std::string& db)
+      : commands(cmds), callbacks(cbs), db_name(db) {
+    static std::atomic<uint64_t> next_id{1};
+    batch_id = next_id.fetch_add(1);
+    create_time = pstd::NowMicros();
+  }
+  
+  bool Empty() const {
+    return commands.empty();
+  }
+  
+  size_t Size() const {
+    return commands.size();
+  }
+};
+
+// Thread-safe command queue for batched command processing
+class CommandQueue {
+public:
+  explicit CommandQueue(size_t max_size);
+  ~CommandQueue();
+  
+  // Enqueue a command batch (blocking if queue is full)
+  bool EnqueueBatch(std::shared_ptr<CommandBatch> batch);
+  
+  // Dequeue a command batch (blocking if queue is empty)
+  std::shared_ptr<CommandBatch> DequeueBatch();
+  
+  // Dequeue all available batches (non-blocking)
+  std::vector<std::shared_ptr<CommandBatch>> DequeueAllBatches();
+  
+  // Get current queue size
+  size_t Size() const;
+  
+  // Check if queue is empty
+  bool Empty() const;
+  
+  // Shutdown the queue
+  void Shutdown();
+
+private:
+  std::queue<std::shared_ptr<CommandBatch>> cmd_queue_;
+  mutable std::mutex queue_mutex_;
+  std::condition_variable queue_cv_;
+  size_t max_size_;
+  std::atomic<bool> shutdown_{false};
+};
+
+#endif  // PIKA_COMMAND_QUEUE_H_

include/pika_rm.h

@@ -12,6 +12,10 @@
 #include <string>
 #include <unordered_map>
 #include <vector>
+#include <thread>
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
 
 #include "pstd/include/pstd_status.h"
 
@@ -23,6 +27,7 @@
 #include "include/pika_stable_log.h"
 #include "include/rsync_client.h"
 #include "include/pika_command_collector.h"
+#include "include/pika_command_queue.h"
 
 #define kBinlogSendPacketNum 40
 #define kBinlogSendBatchNum 100
@@ -121,7 +126,7 @@ class SyncMasterDB : public SyncDB {
   pstd::Status UpdateCommittedID();
   pstd::Status CommitAppLog(const LogOffset& master_committed_id);
   pstd::Status Truncate(const LogOffset& offset);
-  pstd::Status WaitForSlaveAcks(const LogOffset& target_offset, int timeout_ms);
+  // pstd::Status WaitForSlaveAcks(const LogOffset& target_offset, int timeout_ms);
 };
 
 class SyncSlaveDB : public SyncDB {
@@ -242,6 +247,14 @@ class PikaReplicaManager {
     return pika_repl_client_->GetUnfinishedAsyncWriteDBTaskCount(db_name);
   }
 
+  // Command Queue related methods
+  void EnqueueCommandBatch(std::shared_ptr<CommandBatch> batch);
+  std::shared_ptr<CommandBatch> DequeueCommandBatch();
+  size_t GetCommandQueueSize() const;
+  bool IsCommandQueueEmpty() const;
+  // CommittedID notification for RocksDB thread
+  void NotifyCommittedID(const LogOffset& committed_id);
+
  private:
   void InitDB();
   pstd::Status SelectLocalIp(const std::string& remote_ip, int remote_port, std::string* local_ip);
@@ -273,6 +286,41 @@ class PikaReplicaManager {
   std::shared_mutex is_consistency_rwlock_;
   bool is_consistency_ = true;
   std::shared_mutex committed_id_rwlock_;
+
+  // Command queue for collected batches
+  std::unique_ptr<CommandQueue> command_queue_;
+  
+  // Background thread for processing command queue
+  std::unique_ptr<std::thread> command_queue_thread_;
+  std::atomic<bool> command_queue_running_{false};
+  std::mutex command_queue_mutex_;
+  std::condition_variable command_queue_cv_;
+  
+  // RocksDB background thread for Put operations and client responses
+  std::unique_ptr<std::thread> rocksdb_back_thread_;
+  std::atomic<bool> rocksdb_thread_running_{false};
+  std::mutex rocksdb_thread_mutex_;
+  std::condition_variable rocksdb_thread_cv_;
+  
+  // Pending batches waiting for CommittedID
+  std::queue<std::shared_ptr<CommandBatch>> pending_batches_;
+  std::mutex pending_batches_mutex_;
+  
+  // Last committed ID for RocksDB thread processing
+  LogOffset last_committed_id_;
+  std::mutex last_committed_id_mutex_;
+  
+  // Background thread processing methods
+  void StartCommandQueueThread();
+  void StopCommandQueueThread();
+  void CommandQueueLoop();
+  void ProcessCommandBatches(const std::vector<std::shared_ptr<CommandBatch>>& batches);
+  
+  // RocksDB background thread methods
+  void StartRocksDBThread();
+  void StopRocksDBThread();
+  void RocksDBThreadLoop();
+  void ProcessCommittedBatches(const LogOffset& committed_id);
 };
 
 #endif  //  PIKA_RM_H