Ordering Service: on request proposal strategy

irohad/consensus/CMakeLists.txt

@@ -8,6 +8,7 @@ add_library(consensus_round
     )
 target_link_libraries(consensus_round
     boost
+    shared_model_utils
     )
 
 add_library(gate_object

irohad/consensus/yac/impl/supermajority_checker_bft.cpp

@@ -19,6 +19,12 @@ namespace iroha {
             agreed, all, detail::kSupermajorityCheckerKfPlus1Bft);
       }
 
+      bool SupermajorityCheckerBft::hasMajority(PeersNumberType voted,
+                                                PeersNumberType all) const {
+        return checkKfPlus1Majority(
+            voted, all, detail::kSupermajorityCheckerKfPlus1Bft);
+      }
+
       bool SupermajorityCheckerBft::canHaveSupermajority(
           const VoteGroups &votes, PeersNumberType all) const {
         const PeersNumberType largest_group =

irohad/consensus/yac/impl/supermajority_checker_bft.hpp

@@ -23,6 +23,9 @@ namespace iroha {
         bool hasSupermajority(PeersNumberType current,
                               PeersNumberType all) const override;
 
+        bool hasMajority(PeersNumberType voted,
+                         PeersNumberType all) const override;
+
         bool canHaveSupermajority(const VoteGroups &votes,
                                   PeersNumberType all) const override;
       };

irohad/consensus/yac/impl/supermajority_checker_cft.cpp

@@ -19,6 +19,12 @@ namespace iroha {
             agreed, all, detail::kSupermajorityCheckerKfPlus1Cft);
       }
 
+      bool SupermajorityCheckerCft::hasMajority(PeersNumberType voted,
+                                                PeersNumberType all) const {
+        return checkKfPlus1Majority(
+            voted, all, detail::kSupermajorityCheckerKfPlus1Cft);
+      }
+
       bool SupermajorityCheckerCft::canHaveSupermajority(
           const VoteGroups &votes, PeersNumberType all) const {
         const PeersNumberType largest_group =

irohad/consensus/yac/impl/supermajority_checker_cft.hpp

@@ -23,6 +23,9 @@ namespace iroha {
         bool hasSupermajority(PeersNumberType current,
                               PeersNumberType all) const override;
 
+        bool hasMajority(PeersNumberType voted,
+                         PeersNumberType all) const override;
+
         bool canHaveSupermajority(const VoteGroups &votes,
                                   PeersNumberType all) const override;
       };

irohad/consensus/yac/impl/supermajority_checker_kf1.hpp

@@ -34,6 +34,15 @@ namespace iroha {
         return agreed * k >= (k - 1) * (all - 1) + k;
       }
 
+      inline bool checkKfPlus1Majority(PeersNumberType agreed,
+                                            PeersNumberType all,
+                                            unsigned int k) {
+        if (agreed > all) {
+          return false;
+        }
+        return agreed * k > all - 1;
+      }
+
     }  // namespace yac
   }    // namespace consensus
 }  // namespace iroha

irohad/consensus/yac/supermajority_checker.hpp

@@ -49,6 +49,15 @@ namespace iroha {
         virtual bool hasSupermajority(PeersNumberType current,
                                       PeersNumberType all) const = 0;
 
+        /**
+         * Check if majority of votes is achieved
+         * @param voted - number of voted peers
+         * @param all - number of all peers in network
+         * @return true if majority is reached
+         */
+        virtual bool hasMajority(PeersNumberType voted,
+                                 PeersNumberType all) const = 0;
+
         /**
          * Check if supermajority is possible
          * @param voted - numbers of peers voted for each option

irohad/main/application.cpp

@@ -36,6 +36,7 @@
 #include "multi_sig_transactions/transport/mst_transport_stub.hpp"
 #include "network/impl/block_loader_impl.hpp"
 #include "network/impl/peer_communication_service_impl.hpp"
+#include "ordering/impl/kick_out_proposal_creation_strategy.hpp"
 #include "ordering/impl/on_demand_common.hpp"
 #include "ordering/impl/on_demand_ordering_gate.hpp"
 #include "pending_txs_storage/impl/pending_txs_storage_impl.hpp"
@@ -388,6 +389,14 @@ void Irohad::initOrderingGate() {
     return reject_delay;
   };
 
+  std::shared_ptr<iroha::ordering::ProposalCreationStrategy> proposal_strategy =
+      std::make_shared<ordering::KickOutProposalCreationStrategy>(
+          getSupermajorityChecker(kConsensusConsistencyModel));
+
+  auto field_validator =
+      std::make_shared<shared_model::validation::FieldValidator>(
+          validators_config_);
+
   ordering_gate =
       ordering_init.initOrderingGate(max_proposal_size_,
                                      proposal_delay_,
@@ -401,7 +410,10 @@ void Irohad::initOrderingGate() {
                                      proposal_factory,
                                      persistent_cache,
                                      delay,
-                                     log_manager_->getChild("Ordering"));
+                                     log_manager_->getChild("Ordering"),
+                                     field_validator,
+                                     proposal_strategy,
+                                     keypair.publicKey());
   log_->info("[Init] => init ordering gate - [{}]",
              logger::logBool(ordering_gate));
 }

irohad/main/impl/on_demand_ordering_init.cpp

@@ -13,6 +13,7 @@
 #include "interfaces/common_objects/types.hpp"
 #include "logger/logger.hpp"
 #include "logger/logger_manager.hpp"
+#include "ordering/impl/kick_out_proposal_creation_strategy.hpp"
 #include "ordering/impl/on_demand_common.hpp"
 #include "ordering/impl/on_demand_connection_manager.hpp"
 #include "ordering/impl/on_demand_ordering_gate.hpp"
@@ -53,12 +54,14 @@ namespace iroha {
             async_call,
         std::shared_ptr<TransportFactoryType> proposal_transport_factory,
         std::chrono::milliseconds delay,
+        shared_model::crypto::PublicKey my_key,
         const logger::LoggerManagerTreePtr &ordering_log_manager) {
       return std::make_shared<ordering::transport::OnDemandOsClientGrpcFactory>(
           std::move(async_call),
           std::move(proposal_transport_factory),
           [] { return std::chrono::system_clock::now(); },
           delay,
+          my_key,
           ordering_log_manager->getChild("NetworkClient")->getLogger());
     }
 
@@ -69,6 +72,7 @@ namespace iroha {
         std::shared_ptr<TransportFactoryType> proposal_transport_factory,
         std::chrono::milliseconds delay,
         std::vector<shared_model::interface::types::HashType> initial_hashes,
+        shared_model::crypto::PublicKey my_key,
         const logger::LoggerManagerTreePtr &ordering_log_manager) {
       // since top block will be the first in commit_notifier observable,
       // hashes of two previous blocks are prepended
@@ -193,6 +197,7 @@ namespace iroha {
           createNotificationFactory(std::move(async_call),
                                     std::move(proposal_transport_factory),
                                     delay,
+                                    my_key,
                                     ordering_log_manager),
           peers,
           ordering_log_manager->getChild("ConnectionManager")->getLogger());
@@ -205,6 +210,7 @@ namespace iroha {
         std::shared_ptr<shared_model::interface::UnsafeProposalFactory>
             proposal_factory,
         std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
+        std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
         std::function<std::chrono::milliseconds(
             const synchronizer::SynchronizationEvent &)> delay_func,
         size_t max_number_of_transactions,
@@ -264,6 +270,7 @@ namespace iroha {
           std::move(cache),
           std::move(proposal_factory),
           std::move(tx_cache),
+          std::move(creation_strategy),
           max_number_of_transactions,
           ordering_log_manager->getChild("Gate")->getLogger());
     }
@@ -273,11 +280,13 @@ namespace iroha {
         std::shared_ptr<shared_model::interface::UnsafeProposalFactory>
             proposal_factory,
         std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
+        std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
         const logger::LoggerManagerTreePtr &ordering_log_manager) {
       return std::make_shared<ordering::OnDemandOrderingServiceImpl>(
           max_number_of_transactions,
           std::move(proposal_factory),
           std::move(tx_cache),
+          creation_strategy,
           ordering_log_manager->getChild("Service")->getLogger());
     }
 
@@ -307,16 +316,23 @@ namespace iroha {
         std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
         std::function<std::chrono::milliseconds(
             const synchronizer::SynchronizationEvent &)> delay_func,
-        logger::LoggerManagerTreePtr ordering_log_manager) {
+        logger::LoggerManagerTreePtr ordering_log_manager,
+        std::shared_ptr<shared_model::validation::FieldValidator>
+            field_validator,
+        std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
+        shared_model::crypto::PublicKey my_key) {
       auto ordering_service = createService(max_number_of_transactions,
                                             proposal_factory,
                                             tx_cache,
+                                            creation_strategy,
                                             ordering_log_manager);
       service = std::make_shared<ordering::transport::OnDemandOsServerGrpc>(
           ordering_service,
           std::move(transaction_factory),
           std::move(batch_parser),
           std::move(transaction_batch_factory),
+          field_validator,
+          creation_strategy,
           ordering_log_manager->getChild("Server")->getLogger());
       return createGate(
           ordering_service,
@@ -325,10 +341,12 @@ namespace iroha {
                                   std::move(proposal_transport_factory),
                                   delay,
                                   std::move(initial_hashes),
+                                  my_key,
                                   ordering_log_manager),
           std::make_shared<ordering::cache::OnDemandCache>(),
           std::move(proposal_factory),
           std::move(tx_cache),
+          std::move(creation_strategy),
           std::move(delay_func),
           max_number_of_transactions,
           ordering_log_manager);

irohad/main/impl/on_demand_ordering_init.hpp

@@ -21,7 +21,7 @@
 #include "ordering/impl/on_demand_os_server_grpc.hpp"
 #include "ordering/impl/ordering_gate_cache/ordering_gate_cache.hpp"
 #include "ordering/on_demand_ordering_service.hpp"
-#include "ordering/on_demand_os_transport.hpp"
+#include "ordering/ordering_service_proposal_creation_strategy.hpp"
 
 namespace iroha {
   namespace network {
@@ -46,6 +46,7 @@ namespace iroha {
               async_call,
           std::shared_ptr<TransportFactoryType> proposal_transport_factory,
           std::chrono::milliseconds delay,
+          shared_model::crypto::PublicKey my_key,
           const logger::LoggerManagerTreePtr &ordering_log_manager);
 
       /**
@@ -60,6 +61,7 @@ namespace iroha {
           std::shared_ptr<TransportFactoryType> proposal_transport_factory,
           std::chrono::milliseconds delay,
           std::vector<shared_model::interface::types::HashType> initial_hashes,
+          shared_model::crypto::PublicKey my_key,
           const logger::LoggerManagerTreePtr &ordering_log_manager);
 
       /**
@@ -73,6 +75,7 @@ namespace iroha {
           std::shared_ptr<shared_model::interface::UnsafeProposalFactory>
               proposal_factory,
           std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
+          std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
           std::function<std::chrono::milliseconds(
               const synchronizer::SynchronizationEvent &)> delay_func,
           size_t max_number_of_transactions,
@@ -87,6 +90,7 @@ namespace iroha {
           std::shared_ptr<shared_model::interface::UnsafeProposalFactory>
               proposal_factory,
           std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
+          std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
           const logger::LoggerManagerTreePtr &ordering_log_manager);
 
       rxcpp::composite_subscription sync_event_notifier_lifetime_;
@@ -119,6 +123,10 @@ namespace iroha {
        * requests to ordering service and processing responses
        * @param proposal_factory factory required by ordering service to produce
        * proposals
+       * @param field_validator - provides a dependency for validating peer keys
+       * @param creation_strategy - provides a strategy for creaing proposals in
+       * OS
+       * @param my_key - public key of instantiated peer
        * @return initialized ordering gate
        */
       std::shared_ptr<network::OrderingGate> initOrderingGate(
@@ -142,7 +150,11 @@ namespace iroha {
           std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
           std::function<std::chrono::milliseconds(
               const synchronizer::SynchronizationEvent &)> delay_func,
-          logger::LoggerManagerTreePtr ordering_log_manager);
+          logger::LoggerManagerTreePtr ordering_log_manager,
+          std::shared_ptr<shared_model::validation::FieldValidator>
+              field_validator,
+          std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
+          shared_model::crypto::PublicKey my_key);
 
       /// gRPC service for ordering service
       std::shared_ptr<ordering::proto::OnDemandOrdering::Service> service;

irohad/ordering/CMakeLists.txt

@@ -11,6 +11,7 @@ target_link_libraries(on_demand_common
 
 add_library(on_demand_ordering_service
     impl/on_demand_ordering_service_impl.cpp
+    impl/kick_out_proposal_creation_strategy.cpp
     )
 
 target_link_libraries(on_demand_ordering_service

irohad/ordering/impl/kick_out_proposal_creation_strategy.cpp

@@ -0,0 +1,59 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "ordering/impl/kick_out_proposal_creation_strategy.hpp"
+
+#include <algorithm>
+#include "interfaces/common_objects/peer.hpp"
+
+using namespace iroha::ordering;
+
+KickOutProposalCreationStrategy::KickOutProposalCreationStrategy(
+    std::shared_ptr<SupermajorityCheckerType> majority_checker)
+    : majority_checker_(majority_checker) {}
+
+void KickOutProposalCreationStrategy::onCollaborationOutcome(
+    const PeerList &peers) {
+  std::lock_guard<std::mutex> guard(mutex_);
+  RoundCollectionType last_requested;
+  for (const auto &peer : peers) {
+    auto iter = last_requested_.find(peer.hex());
+    if (iter != last_requested_.end()) {
+      last_requested.insert(*iter);
+    } else {
+      last_requested.insert({peer.hex(), RoundType{0, 0}});
+    }
+  }
+  last_requested_ = last_requested;
+}
+
+bool KickOutProposalCreationStrategy::shouldCreateRound(RoundType round) {
+  uint64_t counter = 0;
+  {
+    std::lock_guard<std::mutex> guard(mutex_);
+    counter = std::count_if(
+        last_requested_.begin(),
+        last_requested_.end(),
+        [&round](const auto &elem) { return elem.second >= round; });
+  }
+
+  auto has_majority =
+      majority_checker_->hasMajority(counter, last_requested_.size());
+  return not has_majority;
+}
+
+boost::optional<ProposalCreationStrategy::RoundType>
+KickOutProposalCreationStrategy::onProposal(const PeerType &who,
+                                            RoundType requested_round) {
+  {
+    std::lock_guard<std::mutex> guard(mutex_);
+    auto iter = last_requested_.find(who.hex());
+    if (iter != last_requested_.end() and iter->second < requested_round) {
+      iter->second = requested_round;
+    }
+  }
+
+  return boost::none;
+}