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();
 
@@ -65,31 +64,6 @@ class PikaCommandCollector {
   */
   bool AddCommand(const std::shared_ptr<Cmd>& cmd_ptr, CommandCallback callback);
 
-  /**
-    * @brief Called periodically by external systems to process batches
-    * @param force Force processing even if batch is not full or timeout not reached
-    * @return Number of commands processed
-  */
-
-  /**
-    * @brief Immediately process all currently collected commands
-    * @return The number of commands processed
-  */
-  size_t FlushCommands(bool force = false);
-
-
-  /**
-    * @brief Get the current number of pending commands
-    * @return number of commands
-  */
-  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
@@ -101,91 +75,17 @@ class PikaCommandCollector {
     * @return Pair of (total_processed_commands, total_batches)
   */
   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:
-
-  /**
-    * @brief batch processing command
-    * @param batch command batch
-    * @return Whether the processing is successful
-  */
-  pstd::Status ProcessBatch(const std::vector<std::shared_ptr<Cmd>>& commands, 
-                           const std::vector<CommandCallback>& callbacks);
-
-  /**
-    * @brief Check for conflicts based on command type and key name
-    * @param cmd_ptr command pointer
-    * @return true if there is a conflict (should be replaced), false if there is no conflict
-  */
-  bool CheckConflict(const std::shared_ptr<Cmd>& cmd_ptr) const;
-
-  /**
-    * @brief Handle key conflicts and remove conflicting commands
-    * @param cmd_ptr new command
-  */
-  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_;
 
-  // 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
-  static constexpr size_t kMetricsBufferSize = 100;
-  std::vector<BatchMetrics> recent_metrics_;
-  std::mutex metrics_mutex_;
 };
 
 #endif  // PIKA_COMMAND_COLLECTOR_H_

include/pika_command_queue.h

@@ -0,0 +1,99 @@
+// 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();
+  }
+};
+
+// New structure to group multiple CommandBatches for RocksDB processing
+struct BatchGroup {
+  std::vector<std::shared_ptr<CommandBatch>> batches;
+  LogOffset end_offset;  // Only store the final offset of the last batch
+  BatchGroup() = default;
+  BatchGroup(const std::vector<std::shared_ptr<CommandBatch>>& batches, 
+             const LogOffset& final_offset)
+      : batches(batches), end_offset(final_offset) {}
+  bool Empty() const { return batches.empty(); }
+  size_t BatchCount() const { return batches.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_consensus.h

@@ -155,8 +155,6 @@ class ConsensusCoordinator {
   pstd::Status Reset(const LogOffset& offset);
 
   pstd::Status ProposeLog(const std::shared_ptr<Cmd>& cmd_ptr);
-  // Batch processing of commands
-  pstd::Status BatchProposeLog(const std::vector<std::shared_ptr<Cmd>>& cmd_ptrs, std::vector<LogOffset>* offsets);
   pstd::Status UpdateSlave(const std::string& ip, int port, const LogOffset& start, const LogOffset& end);
   pstd::Status AddSlaveNode(const std::string& ip, int port, int session_id);
   pstd::Status RemoveSlaveNode(const std::string& ip, int port);
@@ -268,8 +266,6 @@ class ConsensusCoordinator {
   void BatchInternalApplyFollower(const std::vector<std::shared_ptr<Cmd>>& cmd_ptrs);
   pstd::Status ProcessCoordination();
 
-  // Batch operations for slave entries
-  pstd::Status BatchPersistAppendBinlog(const std::vector<std::shared_ptr<Cmd>>& cmd_ptrs,const std::vector<BinlogItem>& attributes,std::vector<LogOffset>* offsets);
   LogOffset GetCommittedId() {
     std::lock_guard l(committed_id_rwlock_);
     return committed_id_;
@@ -300,7 +296,7 @@ class ConsensusCoordinator {
 
  private:
   std::shared_mutex is_consistency_rwlock_;
-  bool is_consistency_ = true;
+  bool is_consistency_ = false;
   std::shared_mutex committed_id_rwlock_;
   LogOffset committed_id_ = LogOffset();
   std::atomic<uint64_t> notification_counter_{0};

include/pika_define.h

@@ -364,14 +364,6 @@ const int64_t kPoolSize = 1073741824;
 const std::string kBinlogPrefix = "write2file";
 const size_t kBinlogPrefixLen = 10;
 
-/* 
- * PIKA_BATCH_MAGIC: Core identifier for binlog batch processing.
- * - Master: Prefixes batched binlogs with this magic in SendBinlog
- * - Slave: Detects this magic in HandleBGWorkerWriteBinlog
- *   to switch between batch and single-binlog parsing modes.
- */
-const uint32_t PIKA_BATCH_MAGIC = 0x42544348; // "BTCH" in ASCII
-
 const std::string kPikaMeta = "meta";
 const std::string kManifest = "manifest";
 const std::string kContext = "context";