core/: Concurrent connection attempts

core/CHANGELOG.md

@@ -36,6 +36,13 @@
 - Add `SignedEnvelope` and `PeerRecord` according to [RFC0002] and [RFC0003]
   (see [PR 2107]).
 
+- Concurrently dial up to 5 address candidates within a single dial attempt (see [PR 2248]).
+
+  - On success of a single address, provide errors of the thus far failed dials via
+    `NetworkEvent::ConnectionEstablished::outgoing`.
+
+  - On failure of all addresses, provide the errors via `NetworkEvent::DialError`.
+
 [PR 2145]: https://github.com/libp2p/rust-libp2p/pull/2145
 [PR 2213]: https://github.com/libp2p/rust-libp2p/pull/2213
 [PR 2142]: https://github.com/libp2p/rust-libp2p/pull/2142
@@ -44,6 +51,7 @@
 [PR 2191]: https://github.com/libp2p/rust-libp2p/pull/2191
 [PR 2195]: https://github.com/libp2p/rust-libp2p/pull/2195
 [PR 2107]: https://github.com/libp2p/rust-libp2p/pull/2107
+[PR 2248]: https://github.com/libp2p/rust-libp2p/pull/2248
 [RFC0002]: https://github.com/libp2p/specs/blob/master/RFC/0002-signed-envelopes.md
 [RFC0003]: https://github.com/libp2p/specs/blob/master/RFC/0003-routing-records.md
 

core/src/connection.rs

@@ -23,23 +23,40 @@ pub(crate) mod handler;
 mod listeners;
 mod substream;
 
-pub(crate) mod manager;
 pub(crate) mod pool;
 
-pub use error::{ConnectionError, PendingConnectionError};
+pub use error::{
+    ConnectionError, PendingConnectionError, PendingInboundConnectionError,
+    PendingOutboundConnectionError,
+};
 pub use handler::{ConnectionHandler, ConnectionHandlerEvent, IntoConnectionHandler};
 pub use listeners::{ListenerId, ListenersEvent, ListenersStream};
-pub use manager::ConnectionId;
 pub use pool::{ConnectionCounters, ConnectionLimits};
 pub use pool::{EstablishedConnection, EstablishedConnectionIter, PendingConnection};
 pub use substream::{Close, Substream, SubstreamEndpoint};
 
 use crate::muxing::StreamMuxer;
 use crate::{Multiaddr, PeerId};
+use futures::stream::Stream;
 use std::hash::Hash;
 use std::{error::Error, fmt, pin::Pin, task::Context, task::Poll};
 use substream::{Muxing, SubstreamEvent};
 
+/// Connection identifier.
+#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
+pub struct ConnectionId(usize);
+
+impl ConnectionId {
+    /// Creates a `ConnectionId` from a non-negative integer.
+    ///
+    /// This is primarily useful for creating connection IDs
+    /// in test environments. There is in general no guarantee
+    /// that all connection IDs are based on non-negative integers.
+    pub fn new(id: usize) -> Self {
+        ConnectionId(id)
+    }
+}
+
 /// The endpoint roles associated with a peer-to-peer communication channel.
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
 pub enum Endpoint {
@@ -72,7 +89,40 @@ impl Endpoint {
     }
 }
 
-/// The endpoint roles associated with a peer-to-peer connection.
+/// The endpoint roles associated with a pending peer-to-peer connection.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum PendingPoint {
+    /// The socket comes from a dialer.
+    ///
+    /// There is no single address associated with the Dialer of a pending
+    /// connection. Addresses are dialed in parallel. Only once the first dial
+    /// is successful is the address of the connection known.
+    Dialer,
+    /// The socket comes from a listener.
+    Listener {
+        /// Local connection address.
+        local_addr: Multiaddr,
+        /// Stack of protocols used to send back data to the remote.
+        send_back_addr: Multiaddr,
+    },
+}
+
+impl From<ConnectedPoint> for PendingPoint {
+    fn from(endpoint: ConnectedPoint) -> Self {
+        match endpoint {
+            ConnectedPoint::Dialer { .. } => PendingPoint::Dialer,
+            ConnectedPoint::Listener {
+                local_addr,
+                send_back_addr,
+            } => PendingPoint::Listener {
+                local_addr,
+                send_back_addr,
+            },
+        }
+    }
+}
+
+/// The endpoint roles associated with an established peer-to-peer connection.
 #[derive(PartialEq, Eq, Debug, Clone, Hash)]
 pub enum ConnectedPoint {
     /// We dialed the node.
@@ -234,13 +284,18 @@ where
     pub fn close(self) -> (THandler, Close<TMuxer>) {
         (self.handler, self.muxing.close().0)
     }
+}
+
+impl<TMuxer, THandler> Stream for Connection<TMuxer, THandler>
+where
+    TMuxer: StreamMuxer,
+    THandler: ConnectionHandler<Substream = Substream<TMuxer>>,
+{
+    type Item = Result<Event<THandler::OutEvent>, ConnectionError<THandler::Error>>;
 
     /// Polls the connection for events produced by the associated handler
     /// as a result of I/O activity on the substream multiplexer.
-    pub fn poll(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Event<THandler::OutEvent>, ConnectionError<THandler::Error>>> {
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         loop {
             let mut io_pending = false;
 
@@ -260,9 +315,9 @@ where
                 }
                 Poll::Ready(Ok(SubstreamEvent::AddressChange(address))) => {
                     self.handler.inject_address_change(&address);
-                    return Poll::Ready(Ok(Event::AddressChange(address)));
+                    return Poll::Ready(Some(Ok(Event::AddressChange(address))));
                 }
-                Poll::Ready(Err(err)) => return Poll::Ready(Err(ConnectionError::IO(err))),
+                Poll::Ready(Err(err)) => return Poll::Ready(Some(Err(ConnectionError::IO(err)))),
             }
 
             // Poll the handler for new events.
@@ -276,9 +331,11 @@ where
                     self.muxing.open_substream(user_data);
                 }
                 Poll::Ready(Ok(ConnectionHandlerEvent::Custom(event))) => {
-                    return Poll::Ready(Ok(Event::Handler(event)));
+                    return Poll::Ready(Some(Ok(Event::Handler(event))));
+                }
+                Poll::Ready(Err(err)) => {
+                    return Poll::Ready(Some(Err(ConnectionError::Handler(err))))
                 }
-                Poll::Ready(Err(err)) => return Poll::Ready(Err(ConnectionError::Handler(err))),
             }
         }
     }
@@ -295,26 +352,17 @@ pub struct IncomingInfo<'a> {
 
 impl<'a> IncomingInfo<'a> {
     /// Builds the `ConnectedPoint` corresponding to the incoming connection.
-    pub fn to_connected_point(&self) -> ConnectedPoint {
-        ConnectedPoint::Listener {
+    pub fn to_pending_point(&self) -> PendingPoint {
+        PendingPoint::Listener {
             local_addr: self.local_addr.clone(),
             send_back_addr: self.send_back_addr.clone(),
         }
     }
-}
-
-/// Borrowed information about an outgoing connection currently being negotiated.
-#[derive(Debug, Copy, Clone)]
-pub struct OutgoingInfo<'a> {
-    pub address: &'a Multiaddr,
-    pub peer_id: Option<&'a PeerId>,
-}
-
-impl<'a> OutgoingInfo<'a> {
-    /// Builds a `ConnectedPoint` corresponding to the outgoing connection.
+    /// Builds the `ConnectedPoint` corresponding to the incoming connection.
     pub fn to_connected_point(&self) -> ConnectedPoint {
-        ConnectedPoint::Dialer {
-            address: self.address.clone(),
+        ConnectedPoint::Listener {
+            local_addr: self.local_addr.clone(),
+            send_back_addr: self.send_back_addr.clone(),
         }
     }
 }

core/src/connection/error.rs

@@ -20,6 +20,7 @@
 
 use crate::connection::ConnectionLimit;
 use crate::transport::TransportError;
+use crate::Multiaddr;
 use std::{fmt, io};
 
 /// Errors that can occur in the context of an established `Connection`.
@@ -29,10 +30,6 @@ pub enum ConnectionError<THandlerErr> {
     // TODO: Eventually this should also be a custom error?
     IO(io::Error),
 
-    /// The connection was dropped because the connection limit
-    /// for a peer has been reached.
-    ConnectionLimit(ConnectionLimit),
-
     /// The connection handler produced an error.
     Handler(THandlerErr),
 }
@@ -45,9 +42,6 @@ where
         match self {
             ConnectionError::IO(err) => write!(f, "Connection error: I/O error: {}", err),
             ConnectionError::Handler(err) => write!(f, "Connection error: Handler error: {}", err),
-            ConnectionError::ConnectionLimit(l) => {
-                write!(f, "Connection error: Connection limit: {}.", l)
-            }
         }
     }
 }
@@ -60,16 +54,31 @@ where
         match self {
             ConnectionError::IO(err) => Some(err),
             ConnectionError::Handler(err) => Some(err),
-            ConnectionError::ConnectionLimit(..) => None,
         }
     }
 }
 
+/// Errors that can occur in the context of a pending outgoing `Connection`.
+///
+/// Note: Addresses for an outbound connection are dialed in parallel. Thus, compared to
+/// [`PendingInboundConnectionError`], one or more [`TransportError`]s can occur for a single
+/// connection.
+pub type PendingOutboundConnectionError<TTransErr> =
+    PendingConnectionError<Vec<(Multiaddr, TransportError<TTransErr>)>>;
+
+/// Errors that can occur in the context of a pending incoming `Connection`.
+pub type PendingInboundConnectionError<TTransErr> =
+    PendingConnectionError<TransportError<TTransErr>>;
+
 /// Errors that can occur in the context of a pending `Connection`.
 #[derive(Debug)]
-pub enum PendingConnectionError<TTransErr> {
-    /// An error occurred while negotiating the transport protocol(s).
-    Transport(TransportError<TTransErr>),
+pub enum PendingConnectionError<TransportError> {
+    /// An error occurred while negotiating the transport protocol(s) on a connection.
+    Transport(TransportError),
+
+    /// The connection was dropped because the connection limit
+    /// for a peer has been reached.
+    ConnectionLimit(ConnectionLimit),
 
     /// Pending connection attempt has been aborted.
     Aborted,
@@ -83,16 +92,23 @@ pub enum PendingConnectionError<TTransErr> {
     IO(io::Error),
 }
 
-impl<TTransErr> fmt::Display for PendingConnectionError<TTransErr>
+impl<TransportError> fmt::Display for PendingConnectionError<TransportError>
 where
-    TTransErr: fmt::Display,
+    TransportError: fmt::Display + fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             PendingConnectionError::IO(err) => write!(f, "Pending connection: I/O error: {}", err),
             PendingConnectionError::Aborted => write!(f, "Pending connection: Aborted."),
             PendingConnectionError::Transport(err) => {
-                write!(f, "Pending connection: Transport error: {}", err)
+                write!(
+                    f,
+                    "Pending connection: Transport error on connection: {}",
+                    err
+                )
+            }
+            PendingConnectionError::ConnectionLimit(l) => {
+                write!(f, "Connection error: Connection limit: {}.", l)
             }
             PendingConnectionError::InvalidPeerId => {
                 write!(f, "Pending connection: Invalid peer ID.")
@@ -101,16 +117,17 @@ where
     }
 }
 
-impl<TTransErr> std::error::Error for PendingConnectionError<TTransErr>
+impl<TransportError> std::error::Error for PendingConnectionError<TransportError>
 where
-    TTransErr: std::error::Error + 'static,
+    TransportError: std::error::Error + 'static,
 {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
             PendingConnectionError::IO(err) => Some(err),
-            PendingConnectionError::Transport(err) => Some(err),
+            PendingConnectionError::Transport(_) => None,
             PendingConnectionError::InvalidPeerId => None,
             PendingConnectionError::Aborted => None,
+            PendingConnectionError::ConnectionLimit(..) => None,
         }
     }
 }