Add an implementation of the Raft consensus protocol

WORKSPACE

@@ -174,8 +174,8 @@ boost_deps()
 http_archive(
     name = "net_zlib_zlib",
     build_file = "@com_resdb_nexres//third_party:z.BUILD",
-    sha256 = "91844808532e5ce316b3c010929493c0244f3d37593afd6de04f71821d5136d9",
-    strip_prefix = "zlib-1.2.12",
+    sha256 = "9a93b2b7dfdac77ceba5a558a580e74667dd6fede4585b91eefb60f03b72df23",
+    strip_prefix = "zlib-1.3.1",
     urls = [
         "https://zlib.net/fossils/zlib-1.2.12.tar.gz",
         "https://downloads.sourceforge.net/project/libpng/zlib/1.2.12/zlib-1.2.12.tar.gz",
@@ -225,10 +225,14 @@ bind(
 
 http_archive(
     name = "com_zlib",
-    build_file = "@com_resdb_nexres//third_party:zlib.BUILD",
-    sha256 = "629380c90a77b964d896ed37163f5c3a34f6e6d897311f1df2a7016355c45eff",
-    strip_prefix = "zlib-1.2.11",
-    url = "https://github.com/madler/zlib/archive/v1.2.11.tar.gz",
+    build_file = "@com_resdb_nexres//third_party:z.BUILD",
+    sha256 = "9a93b2b7dfdac77ceba5a558a580e74667dd6fede4585b91eefb60f03b72df23",
+    strip_prefix = "zlib-1.3.1",
+    urls = [
+        "https://zlib.net/zlib-1.3.1.tar.gz",
+        "https://zlib.net/fossils/zlib-1.3.1.tar.gz",
+        "https://github.com/madler/zlib/releases/download/v1.3.1/zlib-1.3.1.tar.gz",
+    ],
 )
 
 http_archive(

benchmark/protocols/raft/BUILD

@@ -0,0 +1,43 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+package(default_visibility = ["//visibility:private"])
+
+load("@bazel_skylib//rules:common_settings.bzl", "bool_flag")
+
+cc_binary(
+    name = "kv_server_performance",
+    srcs = ["kv_server_performance.cpp"],
+    deps = [
+        "//chain/storage:memory_db",
+        "//executor/kv:kv_executor",
+        "//platform/config:resdb_config_utils",
+        "//platform/consensus/ordering/raft/framework:consensus",
+        "//service/utils:server_factory",
+    ],
+)
+
+cc_binary(
+    name = "kv_service_tools",
+    srcs = ["kv_service_tools.cpp"],
+    deps = [
+        "//common/proto:signature_info_cc_proto",
+        "//interface/kv:kv_client",
+        "//platform/config:resdb_config_utils",
+    ],
+)

benchmark/protocols/raft/kv_server_performance.cpp

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <glog/logging.h>
+
+#include "chain/storage/memory_db.h"
+#include "executor/kv/kv_executor.h"
+#include "platform/config/resdb_config_utils.h"
+#include "platform/consensus/ordering/raft/framework/consensus.h"
+#include "platform/networkstrate/service_network.h"
+#include "platform/statistic/stats.h"
+#include "proto/kv/kv.pb.h"
+
+using namespace resdb;
+using namespace resdb::raft;
+using namespace resdb::storage;
+
+void ShowUsage() {
+  printf("<config> <private_key> <cert_file> [logging_dir]\n");
+}
+
+std::string GetRandomKey() {
+  int num1 = rand() % 10;
+  int num2 = rand() % 10;
+  return std::to_string(num1) + std::to_string(num2);
+}
+
+int main(int argc, char** argv) {
+  if (argc < 3) {
+    ShowUsage();
+    exit(0);
+  }
+
+  // google::InitGoogleLogging(argv[0]);
+  // FLAGS_minloglevel = google::GLOG_WARNING;
+
+  char* config_file = argv[1];
+  char* private_key_file = argv[2];
+  char* cert_file = argv[3];
+
+  if (argc >= 5) {
+    auto monitor_port = Stats::GetGlobalStats(5);
+    monitor_port->SetPrometheus(argv[4]);
+  }
+
+  std::unique_ptr<ResDBConfig> config =
+      GenerateResDBConfig(config_file, private_key_file, cert_file);
+
+  config->RunningPerformance(true);
+  ResConfigData config_data = config->GetConfigData();
+
+  auto performance_consens = std::make_unique<Consensus>(
+      *config, std::make_unique<KVExecutor>(std::make_unique<MemoryDB>()));
+  performance_consens->SetupPerformanceDataFunc([]() {
+    KVRequest request;
+    request.set_cmd(KVRequest::SET);
+    request.set_key(GetRandomKey());
+    request.set_value("helloword");
+    std::string request_data;
+    request.SerializeToString(&request_data);
+    return request_data;
+  });
+
+  auto server =
+      std::make_unique<ServiceNetwork>(*config, std::move(performance_consens));
+  server->Run();
+}

benchmark/protocols/raft/kv_service_tools.cpp

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <fstream>
+
+#include "common/proto/signature_info.pb.h"
+#include "interface/kv/kv_client.h"
+#include "platform/config/resdb_config_utils.h"
+
+using resdb::GenerateReplicaInfo;
+using resdb::GenerateResDBConfig;
+using resdb::KVClient;
+using resdb::ReplicaInfo;
+using resdb::ResDBConfig;
+
+int main(int argc, char** argv) {
+  if (argc < 2) {
+    printf("<config path>\n");
+    return 0;
+  }
+  std::string client_config_file = argv[1];
+  ResDBConfig config = GenerateResDBConfig(client_config_file);
+
+  config.SetClientTimeoutMs(100000);
+
+  KVClient client(config);
+
+  client.Set("start", "value");
+  printf("start benchmark\n");
+}

chain/storage/leveldb.cpp

@@ -228,8 +228,10 @@ bool ResLevelDB::UpdateMetrics() {
   return true;
 }
 
-bool ResLevelDB::Flush() {
-  leveldb::Status status = db_->Write(leveldb::WriteOptions(), &batch_);
+bool ResLevelDB::Flush(bool should_sync) {
+  leveldb::WriteOptions opts = leveldb::WriteOptions();
+  opts.sync = should_sync;
+  leveldb::Status status = db_->Write(opts, &batch_);
   if (status.ok()) {
     batch_.Clear();
     return true;

chain/storage/leveldb.h

@@ -74,7 +74,7 @@ class ResLevelDB : public Storage {
 
   bool UpdateMetrics();
 
-  bool Flush() override;
+  bool Flush(bool should_sync = false) override;
 
   virtual uint64_t GetLastCheckpoint() override;
 

chain/storage/mock_storage.h

@@ -57,7 +57,7 @@ class MockStorage : public Storage {
   MOCK_METHOD(ItemsType, GetAllItems, (), (override));
   MOCK_METHOD(ValuesSeqType, GetAllItemsWithSeq, (), (override));
 
-  MOCK_METHOD(bool, Flush, (), (override));
+  MOCK_METHOD(bool, Flush, (bool should_sync), (override));
 };
 
 }  // namespace resdb

chain/storage/storage.h

@@ -62,7 +62,7 @@ class Storage {
   // Default no-op SQL execution for non-SQL backends.
   virtual std::string ExecuteSQL(const std::string& sql_string) { return ""; }
 
-  virtual bool Flush() { return true; };
+  virtual bool Flush(bool should_sync = false) { return true; };
 
   virtual uint64_t GetLastCheckpoint() { return 0; }
 

platform/consensus/ordering/common/algorithm/protocol_base.h

@@ -63,9 +63,9 @@ class ProtocolBase {
   }
 
  protected:
-  int SendMessage(int msg_type, const google::protobuf::Message& msg,
+  virtual int SendMessage(int msg_type, const google::protobuf::Message& msg,
                   int node_id);
-  int Broadcast(int msg_type, const google::protobuf::Message& msg);
+  virtual int Broadcast(int msg_type, const google::protobuf::Message& msg);
   int Commit(const google::protobuf::Message& msg);
 
   bool IsStop();

platform/consensus/ordering/common/framework/consensus.h

@@ -53,7 +53,7 @@ class Consensus : public ConsensusManager {
  protected:
   int SendMsg(int type, const google::protobuf::Message& msg, int node_id);
   int Broadcast(int type, const google::protobuf::Message& msg);
-  int ResponseMsg(const BatchUserResponse& batch_resp);
+  virtual int ResponseMsg(const BatchUserResponse& batch_resp);
   void AsyncSend();
   bool IsStop();
 

platform/consensus/ordering/common/framework/performance_manager.cpp

@@ -52,8 +52,8 @@ PerformanceManager::PerformanceManager(
   total_num_ = 0;
   replica_num_ = config_.GetReplicaNum();
   id_ = config_.GetSelfInfo().id();
-  primary_ = id_ % replica_num_;
-  if (primary_ == 0) primary_ = replica_num_;
+  primary_.store(id_ % replica_num_);
+  if (primary_ == 0) primary_.store(replica_num_);
   local_id_ = 1;
   sum_ = 0;
 }
@@ -67,7 +67,17 @@ PerformanceManager::~PerformanceManager() {
   }
 }
 
-int PerformanceManager::GetPrimary() { return primary_; }
+int PerformanceManager::GetPrimary() { return primary_.load(); }
+
+void PerformanceManager::SetPrimary(int id) {
+  int curr_primary = primary_.load();
+  while (id != curr_primary) {
+    if (primary_.compare_exchange_strong(curr_primary, id)) {
+    LOG(INFO) << "JIM -> " << __FUNCTION__ << ": primary updated to " << id;
+    return;
+    }
+  }  
+}
 
 int PerformanceManager::NeedResponse() {
   return config_.GetMinClientReceiveNum();  // f+1;
@@ -88,16 +98,17 @@ int PerformanceManager::StartEval() {
     return 0;
   }
   eval_started_ = true;
-  for (int i = 0; i < 100000000; ++i) {
-    std::unique_ptr<QueueItem> queue_item = std::make_unique<QueueItem>();
-    queue_item->context = nullptr;
-    queue_item->user_request = GenerateUserRequest();
-    batch_queue_.Push(std::move(queue_item));
-    if (i == 2000000) {
-      eval_ready_promise_.set_value(true);
+  std::thread([&](){
+    for (int i = 0; i < 100000000; ++i) {
+      std::unique_ptr<QueueItem> queue_item = std::make_unique<QueueItem>();
+      queue_item->context = nullptr;
+      queue_item->user_request = GenerateUserRequest();
+      batch_queue_.Push(std::move(queue_item));
+      if (i == 2000000) {
+        eval_ready_promise_.set_value(true);
+      }
     }
-  }
-  LOG(WARNING) << "start eval done";
+}).detach();
   return 0;
 }
 
@@ -176,8 +187,8 @@ void PerformanceManager::SendResponseToClient(
   if (create_time > 0) {
     uint64_t run_time = GetCurrentTime() - create_time;
     LOG(ERROR) << "receive current:" << GetCurrentTime()
-               << " create time:" << create_time << " run time:" << run_time
-               << " local id:" << batch_response.local_id();
+              << " create time:" << create_time << " run time:" << run_time
+              << " local id:" << batch_response.local_id();
     global_stats_->AddLatency(run_time);
   }
   send_num_--;
@@ -186,8 +197,8 @@ void PerformanceManager::SendResponseToClient(
 // =================== request ========================
 int PerformanceManager::BatchProposeMsg() {
   LOG(WARNING) << "batch wait time:" << config_.ClientBatchWaitTimeMS()
-               << " batch num:" << config_.ClientBatchNum()
-               << " max txn:" << config_.GetMaxProcessTxn();
+              << " batch num:" << config_.ClientBatchNum()
+              << " max txn:" << config_.GetMaxProcessTxn();
   std::vector<std::unique_ptr<QueueItem>> batch_req;
   eval_ready_future_.get();
   bool start = false;

platform/consensus/ordering/common/framework/performance_manager.h

@@ -39,6 +39,7 @@ class PerformanceManager {
   virtual ~PerformanceManager();
 
   int StartEval();
+  void SetPrimary(int id);
 
   int ProcessResponseMsg(std::unique_ptr<Context> context,
                          std::unique_ptr<Request> request);
@@ -89,7 +90,7 @@ class PerformanceManager {
   std::mutex response_lock_[response_set_size_];
   int replica_num_;
   int id_;
-  int primary_;
+  std::atomic<int> primary_;
   std::atomic<int> local_id_;
   std::atomic<int> sum_;
 };

platform/consensus/ordering/pbft/consensus_manager_pbft.cpp

@@ -53,9 +53,9 @@ ConsensusManagerPBFT::ConsensusManagerPBFT(
       view_change_manager_(std::make_unique<ViewChangeManager>(
           config_, checkpoint_manager_.get(), message_manager_.get(),
           system_info_.get(), GetBroadCastClient(), GetSignatureVerifier())),
-      recovery_(std::make_unique<Recovery>(config_, checkpoint_manager_.get(),
-                                           system_info_.get(),
-                                           message_manager_->GetStorage())),
+      recovery_(std::make_unique<PBFTRecovery>(
+          config_, checkpoint_manager_.get(), system_info_.get(),
+          message_manager_->GetStorage())),
       query_(std::make_unique<Query>(config_, recovery_.get(),
                                      std::move(query_executor))) {
   LOG(INFO) << "is running is performance mode:"

platform/consensus/ordering/pbft/consensus_manager_pbft.h

@@ -28,7 +28,7 @@
 #include "platform/consensus/ordering/pbft/query.h"
 #include "platform/consensus/ordering/pbft/response_manager.h"
 #include "platform/consensus/ordering/pbft/viewchange_manager.h"
-#include "platform/consensus/recovery/recovery.h"
+#include "platform/consensus/recovery/pbft_recovery.h"
 #include "platform/networkstrate/consensus_manager.h"
 
 namespace resdb {
@@ -84,7 +84,7 @@ class ConsensusManagerPBFT : public ConsensusManager {
   std::unique_ptr<ResponseManager> response_manager_;
   std::unique_ptr<PerformanceManager> performance_manager_;
   std::unique_ptr<ViewChangeManager> view_change_manager_;
-  std::unique_ptr<Recovery> recovery_;
+  std::unique_ptr<PBFTRecovery> recovery_;
   Stats* global_stats_;
   std::queue<std::pair<std::unique_ptr<Context>, std::unique_ptr<Request>>>
       request_pending_;