connection_reuse: drop dead connections. (#178)

Currently, connection substreams are added to
`connection_reuse::Shared::active_connections`, but never removed. This
is not least because the `StreamMuxer` trait defines its inbound and
outbound substream futures to always yield a substream and contains no
provision to signal that no more substreams can be created, which would
allow client code (e.g. `ConnectionReuse`) to detect this and purge its
caches.

This PR defines the `StreamMuxer` trait to optionally yield
inbound/outbound substreams and changes `libp2p-mplex` to handle
stream EOFs by marking the underlying resource as closed.
`ConnectionReuse` will remove stream muxers from its active connections
cache if a `None` substream is returned.

swarm/src/connection_reuse.rs

@@ -40,7 +40,7 @@
 //! `MuxedTransport` trait.
 
 use fnv::FnvHashMap;
-use futures::future::{self, FutureResult, IntoFuture};
+use futures::future::{self, Either, FutureResult, IntoFuture};
 use futures::{Async, Future, Poll, Stream};
 use futures::stream::FuturesUnordered;
 use futures::stream::Fuse as StreamFuse;
@@ -48,7 +48,7 @@ use futures::sync::mpsc;
 use multiaddr::Multiaddr;
 use muxing::StreamMuxer;
 use parking_lot::Mutex;
-use std::io::Error as IoError;
+use std::io::{self, Error as IoError};
 use std::sync::Arc;
 use transport::{MuxedTransport, Transport, UpgradedNode};
 use upgrade::ConnectionUpgrade;
@@ -152,52 +152,59 @@ where
 
     fn dial(self, addr: Multiaddr) -> Result<Self::Dial, (Self, Multiaddr)> {
         // If we already have an active connection, use it!
-        if let Some(connec) = self.shared
+        let substream = if let Some(muxer) = self.shared
             .lock()
             .active_connections
             .get(&addr)
-            .map(|c| c.clone())
+            .map(|muxer| muxer.clone())
         {
-            debug!(target: "libp2p-swarm", "Using existing multiplexed connection to {}", addr);
-            let future = connec.outbound().map(|s| (s, addr));
-            return Ok(Box::new(future) as Box<_>);
-        }
-
-        debug!(target: "libp2p-swarm", "No existing connection to {} ; dialing", addr);
-
-        // TODO: handle if we're already in the middle in dialing that same node?
-        // TODO: try dialing again if the existing connection has dropped
-
-        let dial = match self.inner.dial(addr) {
-            Ok(l) => l,
-            Err((inner, addr)) => {
-                warn!(target: "libp2p-swarm", "Failed to dial {} because the underlying \
-                                               transport doesn't support this address", addr);
-                return Err((
-                    ConnectionReuse {
-                        inner: inner,
-                        shared: self.shared,
-                    },
-                    addr,
-                ));
-            }
+            let a = addr.clone();
+            Either::A(muxer.outbound().map(move |s| s.map(move |s| (s, a))))
+        } else {
+            Either::B(future::ok(None))
         };
 
         let shared = self.shared.clone();
-        let dial = dial.into_future().and_then(move |(connec, addr)| {
-            // Always replace the active connection because we are the most recent.
-            let mut lock = shared.lock();
-            lock.active_connections.insert(addr.clone(), connec.clone());
-            // TODO: doesn't need locking ; the sender could be extracted
-            let _ = lock.add_to_next_tx.unbounded_send((
-                connec.clone(),
-                connec.clone().inbound(),
-                addr.clone(),
-            ));
-            connec.outbound().map(|s| (s, addr))
+        let inner = self.inner;
+        let future = substream.and_then(move |outbound| {
+            if let Some(o) = outbound {
+                debug!(target: "libp2p-swarm", "Using existing multiplexed connection to {}", addr);
+                return Either::A(future::ok(o));
+            }
+            // The previous stream muxer did not yield a new substream => start new dial
+            debug!(target: "libp2p-swarm", "No existing connection to {}; dialing", addr);
+            match inner.dial(addr.clone()) {
+                Ok(dial) => {
+                    let future = dial.into_future().and_then(move |(muxer, addr)| {
+                        muxer.clone().outbound().and_then(move |substream| {
+                            if let Some(s) = substream {
+                                // Replace the active connection because we are the most recent.
+                                let mut lock = shared.lock();
+                                lock.active_connections.insert(addr.clone(), muxer.clone());
+                                // TODO: doesn't need locking ; the sender could be extracted
+                                let _ = lock.add_to_next_tx.unbounded_send((
+                                    muxer.clone(),
+                                    muxer.inbound(),
+                                    addr.clone(),
+                                ));
+                                Ok((s, addr))
+                            } else {
+                                error!(target: "libp2p-swarm", "failed to dial to {}", addr);
+                                shared.lock().active_connections.remove(&addr);
+                                Err(io::Error::new(io::ErrorKind::Other, "dial failed"))
+                            }
+                        })
+                    });
+                    Either::B(Either::A(future))
+                }
+                Err(_) => {
+                    let e = io::Error::new(io::ErrorKind::Other, "transport rejected dial");
+                    Either::B(Either::B(future::err(e)))
+                }
+            }
         });
 
-        Ok(Box::new(dial) as Box<_>)
+        Ok(Box::new(future) as Box<_>)
     }
 
     #[inline]
@@ -281,12 +288,6 @@ where
                 let next_incoming = muxer.clone().inbound();
                 self.connections
                     .push((muxer.clone(), next_incoming, client_addr.clone()));
-                // We overwrite any current active connection to that multiaddr because we
-                // are the freshest possible connection.
-                self.shared
-                    .lock()
-                    .active_connections
-                    .insert(client_addr, muxer);
             }
             Err(err) => {
                 // Insert the rest of the pending upgrades, but not the current one.
@@ -301,7 +302,18 @@ where
         for n in (0..self.connections.len()).rev() {
             let (muxer, mut next_incoming, client_addr) = self.connections.swap_remove(n);
             match next_incoming.poll() {
-                Ok(Async::Ready(incoming)) => {
+                Ok(Async::Ready(None)) => {
+                    // stream muxer gave us a `None` => connection should be considered closed
+                    debug!(target: "libp2p-swarm", "no more inbound substreams on {}", client_addr);
+                    self.shared.lock().active_connections.remove(&client_addr);
+                }
+                Ok(Async::Ready(Some(incoming))) => {
+                    // We overwrite any current active connection to that multiaddr because we
+                    // are the freshest possible connection.
+                    self.shared
+                        .lock()
+                        .active_connections
+                        .insert(client_addr.clone(), muxer.clone());
                     // A new substream is ready.
                     let mut new_next = muxer.clone().inbound();
                     self.connections
@@ -366,7 +378,11 @@ where
         for n in (0..lock.next_incoming.len()).rev() {
             let (muxer, mut future, addr) = lock.next_incoming.swap_remove(n);
             match future.poll() {
-                Ok(Async::Ready(value)) => {
+                Ok(Async::Ready(None)) => {
+                    debug!(target: "libp2p-swarm", "no inbound substream for {}", addr);
+                    lock.active_connections.remove(&addr);
+                }
+                Ok(Async::Ready(Some(value))) => {
                     // A substream is ready ; push back the muxer for the next time this function
                     // is called, then return.
                     debug!(target: "libp2p-swarm", "New incoming substream");

swarm/src/either.rs

@@ -102,56 +102,79 @@ where
     B: StreamMuxer,
 {
     type Substream = EitherSocket<A::Substream, B::Substream>;
-    type InboundSubstream = EitherTransportFuture<A::InboundSubstream, B::InboundSubstream>;
-    type OutboundSubstream = EitherTransportFuture<A::OutboundSubstream, B::OutboundSubstream>;
+    type InboundSubstream = EitherInbound<A, B>;
+    type OutboundSubstream = EitherOutbound<A, B>;
 
     #[inline]
     fn inbound(self) -> Self::InboundSubstream {
         match self {
-            EitherSocket::First(a) => EitherTransportFuture::First(a.inbound()),
-            EitherSocket::Second(b) => EitherTransportFuture::Second(b.inbound()),
+            EitherSocket::First(a) => EitherInbound::A(a.inbound()),
+            EitherSocket::Second(b) => EitherInbound::B(b.inbound()),
         }
     }
 
     #[inline]
     fn outbound(self) -> Self::OutboundSubstream {
         match self {
-            EitherSocket::First(a) => EitherTransportFuture::First(a.outbound()),
-            EitherSocket::Second(b) => EitherTransportFuture::Second(b.outbound()),
+            EitherSocket::First(a) => EitherOutbound::A(a.outbound()),
+            EitherSocket::Second(b) => EitherOutbound::B(b.outbound()),
         }
     }
 }
 
-/// Implements `Future` and redirects calls to either `First` or `Second`.
-///
-/// Additionally, the output will be wrapped inside a `EitherSocket`.
-// TODO: This type is needed because of the lack of `impl Trait` in stable Rust.
-//         If Rust had impl Trait we could use the Either enum from the futures crate and add some
-//         modifiers to it. This custom enum is a combination of Either and these modifiers.
 #[derive(Debug, Copy, Clone)]
-pub enum EitherTransportFuture<A, B> {
-    First(A),
-    Second(B),
+pub enum EitherInbound<A: StreamMuxer, B: StreamMuxer> {
+    A(A::InboundSubstream),
+    B(B::InboundSubstream),
 }
 
-impl<A, B> Future for EitherTransportFuture<A, B>
+impl<A, B> Future for EitherInbound<A, B>
 where
-    A: Future<Error = IoError>,
-    B: Future<Error = IoError>,
+    A: StreamMuxer,
+    B: StreamMuxer,
 {
-    type Item = EitherSocket<A::Item, B::Item>;
+    type Item = Option<EitherSocket<A::Substream, B::Substream>>;
     type Error = IoError;
 
     #[inline]
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        match self {
-            &mut EitherTransportFuture::First(ref mut a) => {
+        match *self {
+            EitherInbound::A(ref mut a) => {
                 let item = try_ready!(a.poll());
-                Ok(Async::Ready(EitherSocket::First(item)))
+                Ok(Async::Ready(item.map(EitherSocket::First)))
             }
-            &mut EitherTransportFuture::Second(ref mut b) => {
+            EitherInbound::B(ref mut b) => {
                 let item = try_ready!(b.poll());
-                Ok(Async::Ready(EitherSocket::Second(item)))
+                Ok(Async::Ready(item.map(EitherSocket::Second)))
+            }
+        }
+    }
+}
+
+#[derive(Debug, Copy, Clone)]
+pub enum EitherOutbound<A: StreamMuxer, B: StreamMuxer> {
+    A(A::OutboundSubstream),
+    B(B::OutboundSubstream),
+}
+
+impl<A, B> Future for EitherOutbound<A, B>
+where
+    A: StreamMuxer,
+    B: StreamMuxer,
+{
+    type Item = Option<EitherSocket<A::Substream, B::Substream>>;
+    type Error = IoError;
+
+    #[inline]
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match *self {
+            EitherOutbound::A(ref mut a) => {
+                let item = try_ready!(a.poll());
+                Ok(Async::Ready(item.map(EitherSocket::First)))
+            }
+            EitherOutbound::B(ref mut b) => {
+                let item = try_ready!(b.poll());
+                Ok(Async::Ready(item.map(EitherSocket::Second)))
             }
         }
     }

swarm/src/muxing.rs

@@ -29,10 +29,18 @@ use tokio_io::{AsyncRead, AsyncWrite};
 pub trait StreamMuxer {
     /// Type of the object that represents the raw substream where data can be read and written.
     type Substream: AsyncRead + AsyncWrite;
+
     /// Future that will be resolved when a new incoming substream is open.
-    type InboundSubstream: Future<Item = Self::Substream, Error = IoError>;
+    ///
+    /// A `None` item signals that the underlying resource has been exhausted and
+    /// no more substreams can be created.
+    type InboundSubstream: Future<Item = Option<Self::Substream>, Error = IoError>;
+
     /// Future that will be resolved when the outgoing substream is open.
-    type OutboundSubstream: Future<Item = Self::Substream, Error = IoError>;
+    ///
+    /// A `None` item signals that the underlying resource has been exhausted and
+    /// no more substreams can be created.
+    type OutboundSubstream: Future<Item = Option<Self::Substream>, Error = IoError>;
 
     /// Produces a future that will be resolved when a new incoming substream arrives.
     fn inbound(self) -> Self::InboundSubstream;