peer_manager_impl.hpp

Go to the documentation of this file.

#ifndef KAGOME_NETWORK_PEERMANAGERIMPL
#define KAGOME_NETWORK_PEERMANAGERIMPL

#include "network/peer_manager.hpp"

#include <memory>
#include <queue>

#include <libp2p/basic/scheduler.hpp>
#include <libp2p/event/bus.hpp>
#include <libp2p/host/host.hpp>
#include <libp2p/protocol/identify/identify.hpp>
#include <libp2p/protocol/kademlia/kademlia.hpp>

#include "application/app_configuration.hpp"
#include "application/app_state_manager.hpp"
#include "application/chain_spec.hpp"
#include "blockchain/block_storage.hpp"
#include "blockchain/block_tree.hpp"
#include "clock/clock.hpp"
#include "consensus/grandpa/voting_round.hpp"
#include "crypto/hasher.hpp"
#include "log/logger.hpp"
#include "metrics/metrics.hpp"
#include "network/impl/protocols/block_announce_protocol.hpp"
#include "network/impl/protocols/propagate_transactions_protocol.hpp"
#include "network/impl/protocols/protocol_factory.hpp"
#include "network/impl/stream_engine.hpp"
#include "network/protocols/sync_protocol.hpp"
#include "network/reputation_repository.hpp"
#include "network/router.hpp"
#include "network/types/block_announce.hpp"
#include "network/types/bootstrap_nodes.hpp"
#include "network/types/own_peer_info.hpp"
#include "scale/libp2p_types.hpp"
#include "storage/buffer_map_types.hpp"

namespace kagome::network {

  enum class PeerType { PEER_TYPE_IN = 0, PEER_TYPE_OUT };

  struct PeerDescriptor {
    PeerType peer_type;
    clock::SteadyClock::TimePoint time_point;
  };

  class PeerManagerImpl : public PeerManager,
                          public std::enable_shared_from_this<PeerManagerImpl> {
   public:
    static constexpr std::chrono::seconds kTimeoutForConnecting{15};

    enum class Error { UNDECLARED_COLLATOR = 1, OUT_OF_VIEW, DUPLICATE };

    PeerManagerImpl(
        std::shared_ptr<application::AppStateManager> app_state_manager,
        libp2p::Host &host,
        std::shared_ptr<libp2p::protocol::Identify> identify,
        std::shared_ptr<libp2p::protocol::kademlia::Kademlia> kademlia,
        std::shared_ptr<libp2p::basic::Scheduler> scheduler,
        std::shared_ptr<StreamEngine> stream_engine,
        const application::AppConfiguration &app_config,
        std::shared_ptr<clock::SteadyClock> clock,
        const BootstrapNodes &bootstrap_nodes,
        const OwnPeerInfo &own_peer_info,
        std::shared_ptr<network::Router> router,
        std::shared_ptr<storage::BufferStorage> storage,
        std::shared_ptr<crypto::Hasher> hasher,
        std::shared_ptr<ReputationRepository> reputation_repository);

    bool prepare();

    bool start();

    void stop();

    void connectToPeer(const PeerInfo &peer_info) override;

    void reserveStreams(const PeerId &peer_id) const override;

    std::shared_ptr<StreamEngine> getStreamEngine() override;

    size_t activePeersNumber() const override;

    void setCollating(const PeerId &peer_id,
                      network::CollatorPublicKey const &collator_id,
                      network::ParachainId para_id) override;

    ParachainState &parachainState() override;

    outcome::result<
        std::pair<network::CollatorPublicKey const &, network::ParachainId>>
    insert_advertisement(PeerState &peer_state,
                         ParachainState &parachain_state,
                         primitives::BlockHash para_hash) override;

    void forEachPeer(std::function<void(const PeerId &)> func) const override;

    void forOnePeer(const PeerId &peer_id,
                    std::function<void(const PeerId &)> func) const override;

    void keepAlive(const PeerId &peer_id) override;

    void startPingingPeer(const PeerId &peer_id) override;

    void updatePeerState(const PeerId &peer_id, const Status &status) override;

    void updatePeerState(const PeerId &peer_id,
                         const BlockAnnounce &announce) override;

    void updatePeerState(
        const PeerId &peer_id,
        const GrandpaNeighborMessage &neighbor_message) override;

    std::optional<std::reference_wrapper<PeerState>> getPeerState(
        const PeerId &peer_id) override;

   private:
    bool isSelfPeer(const PeerId &peer_id) const;

    void align();

    void processDiscoveredPeer(const PeerId &peer_id);

    void processFullyConnectedPeer(const PeerId &peer_id);

    void connectToPeer(const PeerId &peer_id);

    void disconnectFromPeer(const PeerId &peer_id);

    std::vector<scale::PeerInfoSerializable> loadLastActivePeers();

    void storeActivePeers();

    void clearClosedPingingConnections();

    std::shared_ptr<application::AppStateManager> app_state_manager_;
    libp2p::Host &host_;
    std::shared_ptr<libp2p::protocol::Identify> identify_;
    std::shared_ptr<libp2p::protocol::kademlia::Kademlia> kademlia_;
    std::shared_ptr<libp2p::basic::Scheduler> scheduler_;
    std::shared_ptr<StreamEngine> stream_engine_;
    const application::AppConfiguration &app_config_;
    std::shared_ptr<clock::SteadyClock> clock_;
    const BootstrapNodes &bootstrap_nodes_;
    const OwnPeerInfo &own_peer_info_;
    std::shared_ptr<network::Router> router_;
    std::shared_ptr<storage::BufferStorage> storage_;
    std::shared_ptr<crypto::Hasher> hasher_;
    std::shared_ptr<ReputationRepository> reputation_repository_;

    libp2p::event::Handle add_peer_handle_;
    libp2p::event::Handle peer_disconnected_handler_;
    std::unordered_set<PeerId> peers_in_queue_;
    std::deque<std::reference_wrapper<const PeerId>> queue_to_connect_;
    std::unordered_set<PeerId> connecting_peers_;
    std::unordered_set<libp2p::network::ConnectionManager::ConnectionSPtr>
        pinging_connections_;

    std::map<PeerId, PeerDescriptor> active_peers_;
    std::unordered_map<PeerId, PeerState> peer_states_;
    libp2p::basic::Scheduler::Handle align_timer_;
    std::set<PeerId> recently_active_peers_;

    // metrics
    metrics::RegistryPtr registry_ = metrics::createRegistry();
    metrics::Gauge *sync_peer_num_;

    // parachain
    ParachainState parachain_state_;

    log::Logger log_;
  };

}  // namespace kagome::network

OUTCOME_HPP_DECLARE_ERROR(kagome::network, PeerManagerImpl::Error)

#endif  // KAGOME_NETWORK_PEERMANAGERIMPL