Multiple connections per peer (#1440)

* Allow multiple connections per peer in libp2p-core.

Instead of trying to enforce a single connection per peer,
which involves quite a bit of additional complexity e.g.
to prioritise simultaneously opened connections and can
have other undesirable consequences [1], we now
make multiple connections per peer a feature.

The gist of these changes is as follows:

The concept of a "node" with an implicit 1-1 correspondence
to a connection has been replaced with the "first-class"
concept of a "connection". The code from `src/nodes` has moved
(with varying degrees of modification) to `src/connection`.
A `HandledNode` has become a `Connection`, a `NodeHandler` a
`ConnectionHandler`, the `CollectionStream` was the basis for
the new `connection::Pool`, and so forth.

Conceptually, a `Network` contains a `connection::Pool` which
in turn internally employs the `connection::Manager` for
handling the background `connection::manager::Task`s, one
per connection, as before. These are all considered implementation
details. On the public API, `Peer`s are managed as before through
the `Network`, except now the API has changed with the shift of focus
to (potentially multiple) connections per peer. The `NetworkEvent`s have
accordingly also undergone changes.

The Swarm APIs remain largely unchanged, except for the fact that
`inject_replaced` is no longer called. It may now practically happen
that multiple `ProtocolsHandler`s are associated with a single
`NetworkBehaviour`, one per connection. If implementations of
`NetworkBehaviour` rely somehow on communicating with exactly
one `ProtocolsHandler`, this may cause issues, but it is unlikely.

[1]: paritytech/substrate#4272

* Fix intra-rustdoc links.

* Update core/src/connection/pool.rs

Co-Authored-By: Max Inden <mail@max-inden.de>

* Address some review feedback and fix doc links.

* Allow responses to be sent on the same connection.

* Remove unnecessary remainders of inject_replaced.

* Update swarm/src/behaviour.rs

Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com>

* Update swarm/src/lib.rs

Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com>

* Update core/src/connection/manager.rs

Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com>

* Update core/src/connection/manager.rs

Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com>

* Update core/src/connection/pool.rs

Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com>

* Incorporate more review feedback.

* Move module declaration below imports.

* Update core/src/connection/manager.rs

Co-Authored-By: Toralf Wittner <tw@dtex.org>

* Update core/src/connection/manager.rs

Co-Authored-By: Toralf Wittner <tw@dtex.org>

* Simplify as per review.

* Fix rustoc link.

* Add try_notify_handler and simplify.

* Relocate DialingConnection and DialingAttempt.

For better visibility constraints.

* Small cleanup.

* Small cleanup. More robust EstablishedConnectionIter.

* Clarify semantics of `DialingPeer::connect`.

* Don't call inject_disconnected on InvalidPeerId.

To preserve the previous behavior and ensure calls to
`inject_disconnected` are always paired with calls to
`inject_connected`.

* Provide public ConnectionId constructor.

Mainly needed for testing purposes, e.g. in substrate.

* Move the established connection limit check to the right place.

* Clean up connection error handling.

Separate connection errors into those occuring during
connection setup or upon rejecting a newly established
connection (the `PendingConnectionError`) and those
errors occurring on previously established connections,
i.e. for which a `ConnectionEstablished` event has
been emitted by the connection pool earlier.

* Revert change in log level and clarify an invariant.

* Remove inject_replaced entirely.

* Allow notifying all connection handlers.

Thereby simplify by introducing a new enum `NotifyHandler`,
used with a single constructor `NetworkBehaviourAction::NotifyHandler`.

* Finishing touches.

Small API simplifications and code deduplication.
Some more useful debug logging.

Co-authored-by: Max Inden <mail@max-inden.de>
Co-authored-by: Pierre Krieger <pierre.krieger1708@gmail.com>
Co-authored-by: Toralf Wittner <tw@dtex.org>

core/Cargo.toml

@@ -13,6 +13,7 @@ categories = ["network-programming", "asynchronous"]
 asn1_der = "0.6.1"
 bs58 = "0.3.0"
 ed25519-dalek = "1.0.0-pre.3"
+either = "1.5"
 fnv = "1.0"
 futures = { version = "0.3.1", features = ["compat", "io-compat", "executor", "thread-pool"] }
 futures-timer = "3"

core/src/connection.rs

@@ -0,0 +1,336 @@
+// Copyright 2020 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+mod error;
+mod handler;
+mod listeners;
+mod substream;
+
+pub(crate) mod manager;
+pub(crate) mod pool;
+
+pub use error::{ConnectionError, PendingConnectionError};
+pub use handler::{ConnectionHandler, ConnectionHandlerEvent, IntoConnectionHandler};
+pub use listeners::{ListenerId, ListenersStream, ListenersEvent};
+pub use manager::ConnectionId;
+pub use substream::{Substream, SubstreamEndpoint, Close};
+pub use pool::{EstablishedConnection, EstablishedConnectionIter, PendingConnection};
+
+use crate::muxing::StreamMuxer;
+use crate::{Multiaddr, PeerId};
+use std::{fmt, pin::Pin, task::Context, task::Poll};
+use std::hash::Hash;
+use substream::{Muxing, SubstreamEvent};
+
+/// The endpoint roles associated with a peer-to-peer communication channel.
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
+pub enum Endpoint {
+    /// The socket comes from a dialer.
+    Dialer,
+    /// The socket comes from a listener.
+    Listener,
+}
+
+impl std::ops::Not for Endpoint {
+    type Output = Endpoint;
+
+    fn not(self) -> Self::Output {
+        match self {
+            Endpoint::Dialer => Endpoint::Listener,
+            Endpoint::Listener => Endpoint::Dialer
+        }
+    }
+}
+
+impl Endpoint {
+    /// Is this endpoint a dialer?
+    pub fn is_dialer(self) -> bool {
+        if let Endpoint::Dialer = self {
+            true
+        } else {
+            false
+        }
+    }
+
+    /// Is this endpoint a listener?
+    pub fn is_listener(self) -> bool {
+        if let Endpoint::Listener = self {
+            true
+        } else {
+            false
+        }
+    }
+}
+
+/// The endpoint roles associated with a peer-to-peer connection.
+#[derive(PartialEq, Eq, Debug, Clone, Hash)]
+pub enum ConnectedPoint {
+    /// We dialed the node.
+    Dialer {
+        /// Multiaddress that was successfully dialed.
+        address: Multiaddr,
+    },
+    /// We received the node.
+    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 Endpoint {
+    fn from(endpoint: &'_ ConnectedPoint) -> Endpoint {
+        endpoint.to_endpoint()
+    }
+}
+
+impl From<ConnectedPoint> for Endpoint {
+    fn from(endpoint: ConnectedPoint) -> Endpoint {
+        endpoint.to_endpoint()
+    }
+}
+
+impl ConnectedPoint {
+    /// Turns the `ConnectedPoint` into the corresponding `Endpoint`.
+    pub fn to_endpoint(&self) -> Endpoint {
+        match self {
+            ConnectedPoint::Dialer { .. } => Endpoint::Dialer,
+            ConnectedPoint::Listener { .. } => Endpoint::Listener
+        }
+    }
+
+    /// Returns true if we are `Dialer`.
+    pub fn is_dialer(&self) -> bool {
+        match self {
+            ConnectedPoint::Dialer { .. } => true,
+            ConnectedPoint::Listener { .. } => false
+        }
+    }
+
+    /// Returns true if we are `Listener`.
+    pub fn is_listener(&self) -> bool {
+        match self {
+            ConnectedPoint::Dialer { .. } => false,
+            ConnectedPoint::Listener { .. } => true
+        }
+    }
+}
+
+/// Information about a successfully established connection.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Connected<I> {
+    /// The connected endpoint, including network address information.
+    pub endpoint: ConnectedPoint,
+    /// Information obtained from the transport.
+    pub info: I,
+}
+
+impl<I> Connected<I>
+where
+    I: ConnectionInfo
+{
+    pub fn peer_id(&self) -> &I::PeerId {
+        self.info.peer_id()
+    }
+}
+
+/// Information about a connection.
+pub trait ConnectionInfo {
+    /// Identity of the node we are connected to.
+    type PeerId: Eq + Hash;
+
+    /// Returns the identity of the node we are connected to on this connection.
+    fn peer_id(&self) -> &Self::PeerId;
+}
+
+impl ConnectionInfo for PeerId {
+    type PeerId = PeerId;
+
+    fn peer_id(&self) -> &PeerId {
+        self
+    }
+}
+
+/// A multiplexed connection to a peer with an associated `ConnectionHandler`.
+pub struct Connection<TMuxer, THandler>
+where
+    TMuxer: StreamMuxer,
+    THandler: ConnectionHandler<Substream = Substream<TMuxer>>,
+{
+    /// Node that handles the muxing.
+    muxing: substream::Muxing<TMuxer, THandler::OutboundOpenInfo>,
+    /// Handler that processes substreams.
+    handler: THandler,
+}
+
+impl<TMuxer, THandler> fmt::Debug for Connection<TMuxer, THandler>
+where
+    TMuxer: StreamMuxer,
+    THandler: ConnectionHandler<Substream = Substream<TMuxer>> + fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Connection")
+            .field("muxing", &self.muxing)
+            .field("handler", &self.handler)
+            .finish()
+    }
+}
+
+impl<TMuxer, THandler> Unpin for Connection<TMuxer, THandler>
+where
+    TMuxer: StreamMuxer,
+    THandler: ConnectionHandler<Substream = Substream<TMuxer>>,
+{
+}
+
+impl<TMuxer, THandler> Connection<TMuxer, THandler>
+where
+    TMuxer: StreamMuxer,
+    THandler: ConnectionHandler<Substream = Substream<TMuxer>>,
+{
+    /// Builds a new `Connection` from the given substream multiplexer
+    /// and connection handler.
+    pub fn new(muxer: TMuxer, handler: THandler) -> Self {
+        Connection {
+            muxing: Muxing::new(muxer),
+            handler,
+        }
+    }
+
+    /// Returns a reference to the `ConnectionHandler`
+    pub fn handler(&self) -> &THandler {
+        &self.handler
+    }
+
+    /// Returns a mutable reference to the `ConnectionHandler`
+    pub fn handler_mut(&mut self) -> &mut THandler {
+        &mut self.handler
+    }
+
+    /// Notifies the connection handler of an event.
+    pub fn inject_event(&mut self, event: THandler::InEvent) {
+        self.handler.inject_event(event);
+    }
+
+    /// Returns `true` if the remote has shown any sign of activity
+    /// since the connection has been established.
+    ///
+    /// See also [`StreamMuxer::is_remote_acknowledged`].
+    pub fn is_remote_acknowledged(&self) -> bool {
+        self.muxing.is_remote_acknowledged()
+    }
+
+    /// Begins an orderly shutdown of the connection, returning a
+    /// `Future` that resolves when connection shutdown is complete.
+    pub fn close(self) -> Close<TMuxer> {
+        self.muxing.close().0
+    }
+
+    /// 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<THandler::OutEvent, ConnectionError<THandler::Error>>>
+    {
+        loop {
+            let mut io_pending = false;
+
+            // Perform I/O on the connection through the muxer, informing the handler
+            // of new substreams.
+            match self.muxing.poll(cx) {
+                Poll::Pending => io_pending = true,
+                Poll::Ready(Ok(SubstreamEvent::InboundSubstream { substream })) => {
+                    self.handler.inject_substream(substream, SubstreamEndpoint::Listener)
+                }
+                Poll::Ready(Ok(SubstreamEvent::OutboundSubstream { user_data, substream })) => {
+                    let endpoint = SubstreamEndpoint::Dialer(user_data);
+                    self.handler.inject_substream(substream, endpoint)
+                }
+                Poll::Ready(Err(err)) => return Poll::Ready(Err(ConnectionError::IO(err))),
+            }
+
+            // Poll the handler for new events.
+            match self.handler.poll(cx) {
+                Poll::Pending => {
+                    if io_pending {
+                        return Poll::Pending // Nothing to do
+                    }
+                }
+                Poll::Ready(Ok(ConnectionHandlerEvent::OutboundSubstreamRequest(user_data))) => {
+                    self.muxing.open_substream(user_data);
+                }
+                Poll::Ready(Ok(ConnectionHandlerEvent::Custom(event))) => {
+                    return Poll::Ready(Ok(event));
+                }
+                Poll::Ready(Err(err)) => return Poll::Ready(Err(ConnectionError::Handler(err))),
+            }
+        }
+    }
+}
+
+/// Borrowed information about an incoming connection currently being negotiated.
+#[derive(Debug, Copy, Clone)]
+pub struct IncomingInfo<'a> {
+    /// Local connection address.
+    pub local_addr: &'a Multiaddr,
+    /// Stack of protocols used to send back data to the remote.
+    pub send_back_addr: &'a Multiaddr,
+}
+
+impl<'a> IncomingInfo<'a> {
+    /// Builds the `ConnectedPoint` corresponding to the incoming connection.
+    pub fn to_connected_point(&self) -> ConnectedPoint {
+        ConnectedPoint::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, TPeerId> {
+    pub address: &'a Multiaddr,
+    pub peer_id: Option<&'a TPeerId>,
+}
+
+impl<'a, TPeerId> OutgoingInfo<'a, TPeerId> {
+    /// Builds a `ConnectedPoint` corresponding to the outgoing connection.
+    pub fn to_connected_point(&self) -> ConnectedPoint {
+        ConnectedPoint::Dialer {
+            address: self.address.clone()
+        }
+    }
+}
+
+/// Information about a connection limit.
+#[derive(Debug, Clone)]
+pub struct ConnectionLimit {
+    /// The maximum number of connections.
+    pub limit: usize,
+    /// The current number of connections.
+    pub current: usize,
+}
+
+impl fmt::Display for ConnectionLimit {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}/{}", self.current, self.limit)
+    }
+}