feat: hset/hdel wrote by braft and binlog

src/storage/include/storage/storage.h

@@ -62,6 +62,7 @@ struct StorageOptions {
   size_t small_compaction_duration_threshold = 10000;
   size_t db_instance_num = 3;  // default = 3
   bool is_use_raft = true;
+  uint32_t raft_timeout = 10;
   Status ResetOptions(const OptionType& option_type, const std::unordered_map<std::string, std::string>& options_map);
 };
 

src/storage/src/batch.h

@@ -1,3 +1,10 @@
+/*
+ * 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. An additional grant
+ * of patent rights can be found in the PATENTS file in the same directory.
+ */
+
 #pragma once
 
 #include <chrono>
@@ -47,7 +54,7 @@ class RocksBatch : public Batch {
 
 class BinlogBatch : public Batch {
  public:
-  BinlogBatch(int32_t index, int32_t seconds = 10) : seconds_(seconds) {
+  BinlogBatch(int32_t index, uint32_t seconds = 10) : seconds_(seconds) {
     binlog_.set_db_id(0);
     binlog_.set_slot_idx(index);
   }
@@ -71,10 +78,6 @@ class BinlogBatch : public Batch {
     std::promise<Status> promise;
     auto future = promise.get_future();
     pikiwidb::PRaft::Instance().AppendLog(binlog_, std::move(promise));
-    if (seconds_ == -1) {  // if do not require timeout
-      return future.get();
-    }
-
     auto status = future.wait_for(std::chrono::seconds(seconds_));
     if (status == std::future_status::timeout) {
       return Status::Incomplete("Wait for write timeout");
@@ -84,12 +87,12 @@ class BinlogBatch : public Batch {
 
  private:
   pikiwidb::Binlog binlog_;
-  int32_t seconds_;
+  uint32_t seconds_;
 };
 
 inline auto Batch::CreateBatch(Redis* redis, bool use_binlog) -> std::unique_ptr<Batch> {
   if (use_binlog) {
-    return std::make_unique<BinlogBatch>(redis->GetIndex());
+    return std::make_unique<BinlogBatch>(redis->GetIndex(), redis->GetRaftTimeout());
   }
   return std::make_unique<RocksBatch>(redis->GetDB(), redis->GetWriteOptions(), redis->GetColumnFamilyHandles());
 }

src/storage/src/redis.cc

@@ -53,6 +53,7 @@ Redis::~Redis() {
 
 Status Redis::Open(const StorageOptions& storage_options, const std::string& db_path) {
   is_use_raft_ = storage_options.is_use_raft;
+  raft_timeout_ = storage_options.raft_timeout;
   statistics_store_->SetCapacity(storage_options.statistics_max_size);
   small_compaction_threshold_ = storage_options.small_compaction_threshold;
 

src/storage/src/redis.h

@@ -212,6 +212,7 @@ class Redis {
   void GetRocksDBInfo(std::string& info, const char* prefix);
   auto GetWriteOptions() const -> const rocksdb::WriteOptions& { return default_write_options_; }
   auto GetColumnFamilyHandles() const -> const std::vector<rocksdb::ColumnFamilyHandle*>& { return handles_; }
+  auto GetRaftTimeout() const -> uint32_t { return raft_timeout_; }
 
   // Sets Commands
   Status SAdd(const Slice& key, const std::vector<std::string>& members, int32_t* ret);
@@ -358,6 +359,7 @@ class Redis {
 
   // For raft
   bool is_use_raft_;
+  uint32_t raft_timeout_;
 
   Status UpdateSpecificKeyStatistics(const DataType& dtype, const std::string& key, uint64_t count);
   Status UpdateSpecificKeyDuration(const DataType& dtype, const std::string& key, uint64_t duration);