on_demand_ordering_service_impl.hpp

Go to the documentation of this file.

#ifndef IROHA_ON_DEMAND_ORDERING_SERVICE_IMPL_HPP
#define IROHA_ON_DEMAND_ORDERING_SERVICE_IMPL_HPP

#include "ordering/on_demand_ordering_service.hpp"

#include <boost/range/adaptor/filtered.hpp>
#include <boost/range/adaptor/indirected.hpp>
#include <map>
#include <mutex>
#include <shared_mutex>

#include <tbb/concurrent_unordered_set.h>
#include "interfaces/iroha_internal/unsafe_proposal_factory.hpp"
#include "logger/logger_fwd.hpp"
#include "multi_sig_transactions/hash.hpp"
// TODO 2019-03-15 andrei: IR-403 Separate BatchHashEquality and MstState
#include "multi_sig_transactions/state/mst_state.hpp"
#include "ordering/impl/on_demand_common.hpp"
#include "ordering/ordering_service_proposal_creation_strategy.hpp"

namespace iroha {
  namespace ametsuchi {
    class TxPresenceCache;
  }
  namespace ordering {
    namespace detail {
      using BatchSetType = tbb::concurrent_unordered_set<
          transport::OdOsNotification::TransactionBatchType,
          model::PointerBatchHasher,
          shared_model::interface::BatchHashEquality>;

      using ProposalMapType =
          std::map<consensus::Round,
                   boost::optional<std::shared_ptr<
                       const transport::OdOsNotification::ProposalType>>>;
    }  // namespace detail

    class OnDemandOrderingServiceImpl : public OnDemandOrderingService {
     public:
      OnDemandOrderingServiceImpl(
          size_t transaction_limit,
          std::shared_ptr<shared_model::interface::UnsafeProposalFactory>
              proposal_factory,
          std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
          std::shared_ptr<ProposalCreationStrategy> proposal_creation_strategy,
          logger::LoggerPtr log,
          size_t number_of_proposals = 3);

      // --------------------- | OnDemandOrderingService |_---------------------

      void onCollaborationOutcome(consensus::Round round) override;

      void onTxsCommitted(const HashesSetType &hashes) override {
        removeFromBatchesCache(hashes);
      }

      // ----------------------- | OdOsNotification | --------------------------

      void onBatches(CollectionType batches) override;

      boost::optional<std::shared_ptr<const ProposalType>> onRequestProposal(
          consensus::Round round) override;

     private:
      boost::optional<std::shared_ptr<shared_model::interface::Proposal>>
      packNextProposals(const consensus::Round &round);

      boost::optional<
          std::shared_ptr<const OnDemandOrderingServiceImpl::ProposalType>>
      uploadProposal(consensus::Round round);

      using TransactionsCollectionType =
          std::vector<std::shared_ptr<shared_model::interface::Transaction>>;

      boost::optional<std::shared_ptr<shared_model::interface::Proposal>>
      tryCreateProposal(
          consensus::Round const &round,
          const TransactionsCollectionType &txs,
          shared_model::interface::types::TimestampType created_time);

      void tryErase(const consensus::Round &current_round);

      bool batchAlreadyProcessed(
          const shared_model::interface::TransactionBatch &batch);

      void insertBatchToCache(
          std::shared_ptr<shared_model::interface::TransactionBatch> const
              &batch);

      void removeFromBatchesCache(
          const OnDemandOrderingService::HashesSetType &hashes);

      bool isEmptyBatchesCache() const;

      void forCachedBatches(
          std::function<void(const transport::OdOsNotification::BatchesSetType
                                 &)> const &f) override;

      std::vector<std::shared_ptr<shared_model::interface::Transaction>>
      getTransactionsFromBatchesCache(size_t requested_tx_amount);

      size_t transaction_limit_;

      size_t number_of_proposals_;

      detail::ProposalMapType proposal_map_;

      std::mutex proposals_mutex_;

      mutable std::shared_timed_mutex batches_cache_cs_;
      BatchesSetType batches_cache_;

      std::shared_ptr<shared_model::interface::UnsafeProposalFactory>
          proposal_factory_;

      std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache_;

      std::shared_ptr<ProposalCreationStrategy> proposal_creation_strategy_;

      logger::LoggerPtr log_;

      consensus::Round current_round_;
    };
  }  // namespace ordering
}  // namespace iroha

#endif  // IROHA_ON_DEMAND_ORDERING_SERVICE_IMPL_HPP