dial order (#286)

Signed-off-by: turuslan <[email protected]>

Author: turuslan

include/libp2p/host/basic_host/basic_host.hpp

@@ -48,21 +48,15 @@ namespace libp2p::host {
     Connectedness connectedness(const peer::PeerInfo &p) const override;
 
     void connect(const peer::PeerInfo &peer_info,
-                 const ConnectionResultHandler &handler,
-                 std::chrono::milliseconds timeout) override;
+                 const ConnectionResultHandler &handler) override;
 
     void disconnect(const peer::PeerId &peer_id) override;
 
     void setProtocolHandler(StreamProtocols protocols,
                             StreamAndProtocolCb cb,
                             ProtocolPredicate predicate) override;
 
-    void newStream(const peer::PeerInfo &peer_info,
-                   StreamProtocols protocols,
-                   StreamAndProtocolOrErrorCb cb,
-                   std::chrono::milliseconds timeout = {}) override;
-
-    void newStream(const peer::PeerId &peer_id,
+    void newStream(const PeerInfo &peer_id,
                    StreamProtocols protocols,
                    StreamAndProtocolOrErrorCb cb) override;
 

include/libp2p/host/host.hpp

@@ -125,31 +125,18 @@ namespace libp2p {
      * @brief Initiates connection to the peer {@param peer_info}.
      * @param peer_info peer to connect.
      * @param handler callback, will be executed on success or fail
-     * @param timeout in milliseconds
      */
     virtual void connect(const peer::PeerInfo &peer_info,
-                         const ConnectionResultHandler &handler,
-                         std::chrono::milliseconds timeout) = 0;
-
-    /**
-     * @brief Initiates connection to the peer {@param peer_info}.
-     * @param peer_info peer to connect.
-     * @param handler callback, will be executed on success or fail
-     */
-    inline void connect(const peer::PeerInfo &peer_info,
-                        const ConnectionResultHandler &handler) {
-      connect(peer_info, handler, std::chrono::milliseconds::zero());
-    };
+                         const ConnectionResultHandler &handler) = 0;
 
     /**
      * @brief Initiates connection to the peer {@param peer_info}. If connection
      * exists, does nothing.
      * @param peer_info peer to connect.
      */
     inline void connect(const peer::PeerInfo &peer_info) {
-      connect(
-          peer_info, [](auto &&) {}, std::chrono::milliseconds::zero());
-    };
+      connect(peer_info, [](ConnectionResult &&) {});
+    }
 
     /**
      * Closes all connections (outbound and inbound) to given {@param peer_id}
@@ -158,16 +145,14 @@ namespace libp2p {
 
     /**
      * @brief Open new stream to the peer {@param peer_info} with protocol
-     * {@param protocol} with a specific timeout.
+     * {@param protocol}
      * @param peer_info stream will be opened to this peer
      * @param protocols "speak" using first supported protocol
      * @param cb callback, will be executed on success or fail
-     * @param timeout in milliseconds
      */
     virtual void newStream(const peer::PeerInfo &peer_info,
                            StreamProtocols protocols,
-                           StreamAndProtocolOrErrorCb cb,
-                           std::chrono::milliseconds timeout = {}) = 0;
+                           StreamAndProtocolOrErrorCb cb) = 0;
 
     /**
      * @brief Open new stream to the peer {@param peer} with protocol
@@ -176,9 +161,11 @@ namespace libp2p {
      * @param protocols "speak" using first supported protocol
      * @param cb callback, will be executed on success or fail
      */
-    virtual void newStream(const peer::PeerId &peer_id,
-                           StreamProtocols protocols,
-                           StreamAndProtocolOrErrorCb cb) = 0;
+    void newStream(const PeerId &peer_id,
+                   StreamProtocols protocols,
+                   StreamAndProtocolOrErrorCb cb) {
+      newStream(PeerInfo{.id = peer_id}, std::move(protocols), std::move(cb));
+    }
 
     /**
      * @brief Create listener on given multiaddress.

include/libp2p/injector/host_injector.hpp

@@ -10,7 +10,6 @@
 
 // implementations
 #include <libp2p/host/basic_host.hpp>
-#include <libp2p/peer/address_repository/inmem_address_repository.hpp>
 #include <libp2p/peer/impl/peer_repository_impl.hpp>
 #include <libp2p/peer/key_repository/inmem_key_repository.hpp>
 #include <libp2p/peer/protocol_repository/inmem_protocol_repository.hpp>
@@ -31,7 +30,6 @@ namespace libp2p::injector {
 
         // repositories
         di::bind<peer::PeerRepository>.to<peer::PeerRepositoryImpl>(),
-        di::bind<peer::AddressRepository>.to<peer::InmemAddressRepository>(),
         di::bind<peer::KeyRepository>.to<peer::InmemKeyRepository>(),
         di::bind<peer::ProtocolRepository>.to<peer::InmemProtocolRepository>(),
 

include/libp2p/injector/network_injector.hpp

@@ -31,6 +31,7 @@
 #include <libp2p/network/impl/network_impl.hpp>
 #include <libp2p/network/impl/router_impl.hpp>
 #include <libp2p/network/impl/transport_manager_impl.hpp>
+#include <libp2p/peer/address_repository/inmem_address_repository.hpp>
 #include <libp2p/peer/impl/identity_manager_impl.hpp>
 #include <libp2p/protocol_muxer/multiselect.hpp>
 #include <libp2p/security/noise.hpp>
@@ -347,6 +348,8 @@ namespace libp2p::injector {
         di::bind<muxer::MuxerAdaptor *[]>().to<muxer::Yamux, muxer::Mplex>(),  // NOLINT
         di::bind<transport::TransportAdaptor *[]>().to<transport::TcpTransport, transport::QuicTransport>(),  // NOLINT
 
+        di::bind<peer::AddressRepository>.to<peer::InmemAddressRepository>(),
+
         // user-defined overrides...
         std::forward<decltype(args)>(args)...
     );

include/libp2p/muxer/muxed_connection_config.hpp

@@ -32,5 +32,9 @@ namespace libp2p::muxer {
     static constexpr std::chrono::milliseconds kDefaultNoStreamsInterval =
         std::chrono::milliseconds(120000);
     std::chrono::milliseconds no_streams_interval = kDefaultNoStreamsInterval;
+
+    /// Dial timeout for outgoing connection
+    static constexpr std::chrono::seconds kDefaultDialTimeout{10};
+    std::chrono::milliseconds dial_timeout = kDefaultDialTimeout;
   };
 }  // namespace libp2p::muxer

include/libp2p/network/dialer.hpp

@@ -28,37 +28,24 @@ namespace libp2p::network {
         outcome::result<std::shared_ptr<connection::CapableConnection>>;
     using DialResultFunc = std::function<void(DialResult)>;
 
-    /**
-     * Establishes a connection or returns existing one to a given peer with a
-     * specific timeout
-     */
-    virtual void dial(const peer::PeerInfo &p,
-                      DialResultFunc cb,
-                      std::chrono::milliseconds timeout) = 0;
-
     /**
      * Establishes a connection or returns existing one to a given peer
      */
-    inline void dial(const peer::PeerInfo &p, DialResultFunc cb) {
-      dial(p, std::move(cb), std::chrono::milliseconds::zero());
-    }
-
-    /**
-     * NewStream returns a new stream to given peer p with a specific timeout.
-     * If there is no connection to p, attempts to create one.
-     */
-    virtual void newStream(const peer::PeerInfo &peer_info,
-                           StreamProtocols protocols,
-                           StreamAndProtocolOrErrorCb cb,
-                           std::chrono::milliseconds timeout = {}) = 0;
+    virtual void dial(const PeerInfo &p, DialResultFunc cb) = 0;
 
     /**
      * NewStream returns a new stream to given peer p.
      * If there is no connection to p, returns error.
      */
-    virtual void newStream(const peer::PeerId &peer_id,
+    virtual void newStream(const PeerInfo &peer_id,
                            StreamProtocols protocols,
                            StreamAndProtocolOrErrorCb cb) = 0;
+
+    void newStream(const PeerId &peer_id,
+                   StreamProtocols protocols,
+                   StreamAndProtocolOrErrorCb cb) {
+      newStream(PeerInfo{.id = peer_id}, std::move(protocols), std::move(cb));
+    }
   };
 
 }  // namespace libp2p::network

include/libp2p/network/impl/dialer_impl.hpp

@@ -6,14 +6,16 @@
 
 #pragma once
 
-#include <set>
+#include <deque>
 #include <unordered_map>
+#include <unordered_set>
 
 #include <libp2p/basic/scheduler.hpp>
 #include <libp2p/network/connection_manager.hpp>
 #include <libp2p/network/dialer.hpp>
 #include <libp2p/network/listener_manager.hpp>
 #include <libp2p/network/transport_manager.hpp>
+#include <libp2p/peer/address_repository.hpp>
 #include <libp2p/protocol_muxer/protocol_muxer.hpp>
 
 namespace libp2p::network {
@@ -27,36 +29,25 @@ namespace libp2p::network {
                std::shared_ptr<TransportManager> tmgr,
                std::shared_ptr<ConnectionManager> cmgr,
                std::shared_ptr<ListenerManager> listener,
+               std::shared_ptr<peer::AddressRepository> addr_repo,
                std::shared_ptr<basic::Scheduler> scheduler);
 
     // Establishes a connection to a given peer
-    void dial(const peer::PeerInfo &p,
-              DialResultFunc cb,
-              std::chrono::milliseconds timeout) override;
+    void dial(const PeerInfo &p, DialResultFunc cb) override;
 
-    // NewStream returns a new stream to given peer p.
-    // If there is no connection to p, attempts to create one.
-    void newStream(const peer::PeerInfo &p,
-                   StreamProtocols protocols,
-                   StreamAndProtocolOrErrorCb cb,
-                   std::chrono::milliseconds timeout = {}) override;
-
-    void newStream(const peer::PeerId &peer_id,
+    void newStream(const PeerInfo &peer_id,
                    StreamProtocols protocols,
                    StreamAndProtocolOrErrorCb cb) override;
 
    private:
     // A context to handle an intermediary state of the peer we are dialing to
     // but the connection is not yet established
     struct DialCtx {
-      /// Known and scheduled addresses to try to dial via
-      std::set<multi::Multiaddress> addresses;
-
-      /// Timeout for a single connection attempt
-      std::chrono::milliseconds timeout;
+      /// Queue of addresses to try connect to
+      std::deque<Multiaddress> addr_queue;
 
-      /// Addresses we already tried, but no connection was established
-      std::set<multi::Multiaddress> tried_addresses;
+      /// Tracks addresses added to `addr_queue`
+      std::unordered_set<Multiaddress> addr_seen;
 
       /// Callbacks for all who requested a connection to the peer
       std::vector<Dialer::DialResultFunc> callbacks;
@@ -86,6 +77,7 @@ namespace libp2p::network {
     std::shared_ptr<TransportManager> tmgr_;
     std::shared_ptr<ConnectionManager> cmgr_;
     std::shared_ptr<ListenerManager> listener_;
+    std::shared_ptr<peer::AddressRepository> addr_repo_;
     std::shared_ptr<basic::Scheduler> scheduler_;
     log::Logger log_;
 

include/libp2p/peer/address_repository.hpp

@@ -123,6 +123,13 @@ namespace libp2p::peer {
     virtual outcome::result<void> updateAddresses(const PeerId &p,
                                                   Milliseconds ttl) = 0;
 
+    /**
+     * Move failed address to back.
+     * That way dialer will try other addresses first.
+     */
+    virtual void dialFailed(const PeerId &peer_id,
+                            const Multiaddress &addr) = 0;
+
     /**
      * @brief Get all addresses associated with this Peer {@param p}. May
      * contain duplicates.

include/libp2p/peer/address_repository/inmem_address_repository.hpp

@@ -50,6 +50,8 @@ namespace libp2p::peer {
     outcome::result<void> updateAddresses(const PeerId &p,
                                           Milliseconds ttl) override;
 
+    void dialFailed(const PeerId &peer_id, const Multiaddress &addr) override;
+
     outcome::result<std::vector<multi::Multiaddress>> getAddresses(
         const PeerId &p) const override;
 
@@ -60,13 +62,15 @@ namespace libp2p::peer {
     std::unordered_set<PeerId> getPeers() const override;
 
    private:
-    using ttlmap = std::unordered_map<multi::Multiaddress, Clock::time_point>;
-    using ttlmap_ptr = std::shared_ptr<ttlmap>;
-    using peer_db = std::unordered_map<PeerId, ttlmap_ptr>;
+    struct Peer {
+      std::unordered_map<Multiaddress, Clock::time_point> expires;
+      std::vector<Multiaddress> order;
 
-    bool isNewDnsAddr(const multi::Multiaddress &ma);
+      bool eraseOrder(const Multiaddress &addr);
+    };
+    using peer_db = std::unordered_map<PeerId, Peer>;
 
-    peer_db::iterator findOrInsert(const PeerId &p);
+    bool isNewDnsAddr(const multi::Multiaddress &ma);
 
     std::shared_ptr<network::DnsaddrResolver> dnsaddr_resolver_;
     peer_db db_;

include/libp2p/peer/peer_info.hpp

@@ -40,6 +40,10 @@ namespace libp2p::peer {
 
 }  // namespace libp2p::peer
 
+namespace libp2p {
+  using peer::PeerInfo;
+}  // namespace libp2p
+
 namespace std {
   template <>
   struct hash<libp2p::peer::PeerInfo> {