Poe fix response (#225)

* change reponse num for poe

* change reponse num for poe

* add files

* add files

Author: cjcchen

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

@@ -69,6 +69,10 @@ PerformanceManager::~PerformanceManager() {
 
 int PerformanceManager::GetPrimary() { return primary_; }
 
+int PerformanceManager::NeedResponse() {
+  return config_.GetMinClientReceiveNum();  // f+1;
+}
+
 std::unique_ptr<Request> PerformanceManager::GenerateUserRequest() {
   std::unique_ptr<Request> request = std::make_unique<Request>();
   request->set_data(data_func_());
@@ -154,7 +158,7 @@ CollectorResultCode PerformanceManager::AddResponseMsg(
       return CollectorResultCode::OK;
     }
     response_[idx][seq]++;
-    if (response_[idx][seq] >= config_.GetMinClientReceiveNum()) {
+    if (response_[idx][seq] >= NeedResponse()) {
       response_[idx].erase(response_[idx].find(seq));
       done = true;
     }

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

@@ -46,6 +46,7 @@ class PerformanceManager {
 
  protected:
   virtual void SendMessage(const Request& request);
+  virtual int NeedResponse();
 
  private:
   // Add response messages which will be sent back to the caller

platform/consensus/ordering/pbft/checkpoint_manager_test.cpp

@@ -108,7 +108,8 @@ class CheckPointManagerTest : public Test {
 TEST_F(CheckPointManagerTest, SendCheckPoint) {
   config_.SetViewchangeCommitTimeout(100);
   SystemInfo sys_info;
-  CheckPointManager manager(config_, &replica_communicator_, nullptr, &sys_info);
+  CheckPointManager manager(config_, &replica_communicator_, nullptr,
+                            &sys_info);
 
   for (int i = 1; i <= 5; ++i) {
     std::unique_ptr<Request> request = std::make_unique<Request>();
@@ -135,7 +136,8 @@ TEST_F(CheckPointManagerTest, SendCheckPointOnce) {
   }));
 
   SystemInfo sys_info;
-  CheckPointManager manager(config_, &replica_communicator_, nullptr, &sys_info);
+  CheckPointManager manager(config_, &replica_communicator_, nullptr,
+                            &sys_info);
   for (int i = 1; i <= 5; ++i) {
     std::unique_ptr<Request> request = std::make_unique<Request>();
     request->set_seq(i);
@@ -163,7 +165,8 @@ TEST_F(CheckPointManagerTest, SendCheckPointTwo) {
   }));
 
   SystemInfo sys_info;
-  CheckPointManager manager(config_, &replica_communicator_, nullptr, &sys_info);
+  CheckPointManager manager(config_, &replica_communicator_, nullptr,
+                            &sys_info);
   std::unique_ptr<Request> request = std::make_unique<Request>();
   for (int i = 1; i <= 5; ++i) {
     std::unique_ptr<Request> request = std::make_unique<Request>();
@@ -261,7 +264,8 @@ TEST_F(CheckPointManagerTest, Votes) {
   std::future<bool> propose_done_future = propose_done.get_future();
 
   SystemInfo sys_info;
-  CheckPointManager manager(config_, &replica_communicator_, &mock_verifier, &sys_info);
+  CheckPointManager manager(config_, &replica_communicator_, &mock_verifier,
+                            &sys_info);
   EXPECT_CALL(replica_communicator_, BroadCast)
       .WillRepeatedly(Invoke([&](const google::protobuf::Message& message) {
         for (int i = 1; i <= 3; ++i) {

platform/consensus/ordering/pbft/commitment_test.cpp

@@ -58,7 +58,8 @@ class CommitmentTest : public Test {
         global_stats_(Stats::GetGlobalStats(1)),
         config_(GenerateConfig()),
         system_info_(config_),
-        checkpoint_manager_(config_, &replica_communicator_, &verifier_, &system_info_),
+        checkpoint_manager_(config_, &replica_communicator_, &verifier_,
+                            &system_info_),
         message_manager_(std::make_unique<MessageManager>(
             config_, nullptr, &checkpoint_manager_, &system_info_)),
         commitment_(

platform/consensus/ordering/pbft/query_test.cpp

@@ -59,7 +59,8 @@ class QueryTest : public Test {
       : global_stats_(Stats::GetGlobalStats(1)),
         config_(GenerateConfig()),
         system_info_(config_),
-        checkpoint_manager_(config_, &replica_communicator_, nullptr, &system_info_),
+        checkpoint_manager_(config_, &replica_communicator_, nullptr,
+                            &system_info_),
         message_manager_(config_, nullptr, &checkpoint_manager_, &system_info_),
         recovery_(config_, &checkpoint_manager_, &system_info_, nullptr),
         query_(config_, &recovery_),

platform/consensus/ordering/poe/framework/BUILD

@@ -26,8 +26,19 @@ cc_library(
         "//visibility:public",
     ],
     deps = [
+        ":performance_manager",
         "//common/utils",
         "//platform/consensus/ordering/common/framework:consensus",
         "//platform/consensus/ordering/poe/algorithm:poe",
     ],
 )
+
+cc_library(
+    name = "performance_manager",
+    srcs = ["performance_manager.cpp"],
+    hdrs = ["performance_manager.h"],
+    deps = [
+        "//platform/consensus/ordering/common/framework:performance_manager",
+        "//platform/consensus/ordering/poe/proto:proposal_cc_proto",
+    ],
+)

platform/consensus/ordering/poe/framework/consensus.cpp

@@ -27,12 +27,26 @@
 namespace resdb {
 namespace poe {
 
+std::unique_ptr<PoEPerformanceManager> Consensus::GetPerformanceManager() {
+  return config_.IsPerformanceRunning()
+             ? std::make_unique<PoEPerformanceManager>(
+                   config_, GetBroadCastClient(), GetSignatureVerifier())
+             : nullptr;
+}
+
 Consensus::Consensus(const ResDBConfig& config,
                      std::unique_ptr<TransactionManager> executor)
     : common::Consensus(config, std::move(executor)) {
   int total_replicas = config_.GetReplicaNum();
   int f = (total_replicas - 1) / 3;
 
+  if (config_.GetPublicKeyCertificateInfo()
+          .public_key()
+          .public_key_info()
+          .type() == CertificateKeyInfo::CLIENT) {
+    SetPerformanceManager(GetPerformanceManager());
+  }
+
   Init();
 
   start_ = 0;

platform/consensus/ordering/poe/framework/consensus.h

@@ -22,6 +22,7 @@
 #include "executor/common/transaction_manager.h"
 #include "platform/consensus/ordering/common/framework/consensus.h"
 #include "platform/consensus/ordering/poe/algorithm/poe.h"
+#include "platform/consensus/ordering/poe/framework/performance_manager.h"
 #include "platform/networkstrate/consensus_manager.h"
 
 namespace resdb {
@@ -41,6 +42,8 @@ class Consensus : public common::Consensus {
 
   int Prepare(const Transaction& txn);
 
+  std::unique_ptr<PoEPerformanceManager> GetPerformanceManager();
+
  protected:
   std::unique_ptr<PoE> poe_;
   Stats* global_stats_;

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

@@ -0,0 +1,41 @@
+/*
+ * 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 "platform/consensus/ordering/poe/framework/performance_manager.h"
+
+#include <glog/logging.h>
+
+#include "common/utils/utils.h"
+
+namespace resdb {
+namespace poe {
+
+using comm::CollectorResultCode;
+
+PoEPerformanceManager::PoEPerformanceManager(
+    const ResDBConfig& config, ReplicaCommunicator* replica_communicator,
+    SignatureVerifier* verifier)
+    : PerformanceManager(config, replica_communicator, verifier) {
+  f_ = config_.GetMaxMaliciousReplicaNum();
+}
+
+int PoEPerformanceManager::NeedResponse() { return 2 * f_ + 1; }
+
+}  // namespace poe
+}  // namespace resdb

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

@@ -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.
+ */
+
+#pragma once
+
+#include <thread>
+
+#include "platform/consensus/ordering/common/framework/performance_manager.h"
+#include "platform/consensus/ordering/poe/proto/proposal.pb.h"
+
+namespace resdb {
+namespace poe {
+
+class PoEPerformanceManager : public common::PerformanceManager {
+ public:
+  PoEPerformanceManager(const ResDBConfig& config,
+                        ReplicaCommunicator* replica_communicator,
+                        SignatureVerifier* verifier);
+
+  int NeedResponse() override;
+
+ private:
+  int f_;
+};
+
+}  // namespace poe
+}  // namespace resdb