feat(swarm): smart dialing

This PR tries to port smart dialing logic from `go-libp2p`
https://github.com/libp2p/go-libp2p/blob/v0.45.0/p2p/net/swarm/dial_ranker.go
libp2p/go-libp2p#2260

Pull-Request: #6229.

Author: hanabi1224

swarm/CHANGELOG.md

@@ -6,6 +6,12 @@
 - Raise MSRV to 1.88.0.
   See [PR 6273](https://github.com/libp2p/rust-libp2p/pull/6273).
 
+- Add smart dialing support (`with_smart_dial`).
+  Ranks dial addresses by transport priority (QUIC > TCP, IPv6 > IPv4)
+  with Happy Eyeballs (RFC 8305) staggered delays. All addresses are
+  dialed concurrently.
+  See [PR 6229](https://github.com/libp2p/rust-libp2p/pull/6229).
+
 ## 0.47.1
 
 - Replace `lru::LruCache` with `hashlink::LruCache`.

swarm/src/connection/pool.rs

@@ -18,6 +18,7 @@
 // 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.
+
 use std::{
     collections::HashMap,
     convert::Infallible,
@@ -31,7 +32,7 @@ use concurrent_dial::ConcurrentDial;
 use fnv::FnvHashMap;
 use futures::{
     channel::{mpsc, oneshot},
-    future::{BoxFuture, Either, poll_fn},
+    future::{Either, poll_fn},
     prelude::*,
     ready,
     stream::{FuturesUnordered, SelectAll},
@@ -49,11 +50,13 @@ use crate::{
     connection::{
         Connected, Connection, ConnectionError, ConnectionId, IncomingInfo,
         PendingInboundConnectionError, PendingOutboundConnectionError, PendingPoint,
+        pool::concurrent_dial::{PendingDial, SmartDial},
     },
     transport::TransportError,
 };
 
-mod concurrent_dial;
+pub(crate) mod concurrent_dial;
+pub(crate) mod dial_ranker;
 mod task;
 
 enum ExecSwitch {
@@ -142,6 +145,9 @@ where
 
     /// How long a connection should be kept alive once it starts idling.
     idle_connection_timeout: Duration,
+
+    /// Enables smart dialing.
+    smart_dial: bool,
 }
 
 #[derive(Debug)]
@@ -333,6 +339,7 @@ where
             no_established_connections_waker: None,
             established_connection_events: Default::default(),
             new_connection_dropped_listeners: Default::default(),
+            smart_dial: config.smart_dial,
         }
     }
 
@@ -413,15 +420,7 @@ where
     /// that establishes and negotiates the connection.
     pub(crate) fn add_outgoing(
         &mut self,
-        dials: Vec<
-            BoxFuture<
-                'static,
-                (
-                    Multiaddr,
-                    Result<(PeerId, StreamMuxerBox), TransportError<std::io::Error>>,
-                ),
-            >,
-        >,
+        dials: Vec<PendingDial>,
         peer: Option<PeerId>,
         role_override: Endpoint,
         port_use: PortUse,
@@ -435,15 +434,27 @@ where
 
         let (abort_notifier, abort_receiver) = oneshot::channel();
 
-        self.executor.spawn(
-            task::new_for_pending_outgoing_connection(
-                connection_id,
-                ConcurrentDial::new(dials, concurrency_factor),
-                abort_receiver,
-                self.pending_connection_events_tx.clone(),
-            )
-            .instrument(span),
-        );
+        if self.smart_dial {
+            self.executor.spawn(
+                task::new_for_pending_outgoing_connection(
+                    connection_id,
+                    SmartDial::new(dials),
+                    abort_receiver,
+                    self.pending_connection_events_tx.clone(),
+                )
+                .instrument(span),
+            );
+        } else {
+            self.executor.spawn(
+                task::new_for_pending_outgoing_connection(
+                    connection_id,
+                    ConcurrentDial::new(dials, concurrency_factor),
+                    abort_receiver,
+                    self.pending_connection_events_tx.clone(),
+                )
+                .instrument(span),
+            );
+        }
 
         let endpoint = PendingPoint::Dialer {
             role_override,
@@ -984,6 +995,8 @@ pub(crate) struct PoolConfig {
     pub(crate) per_connection_event_buffer_size: usize,
     /// Number of addresses concurrently dialed for a single outbound connection attempt.
     pub(crate) dial_concurrency_factor: NonZeroU8,
+    /// Ranker that determines the ranking of outgoing connection attempts.
+    pub(crate) smart_dial: bool,
     /// How long a connection should be kept alive once it is idling.
     pub(crate) idle_connection_timeout: Duration,
     /// The configured override for substream protocol upgrades, if any.
@@ -1005,6 +1018,7 @@ impl PoolConfig {
             idle_connection_timeout: Duration::from_secs(10),
             substream_upgrade_protocol_override: None,
             max_negotiating_inbound_streams: 128,
+            smart_dial: false,
         }
     }
 

swarm/src/connection/pool/concurrent_dial.rs

@@ -22,66 +22,96 @@ use std::{
     num::NonZeroU8,
     pin::Pin,
     task::{Context, Poll},
+    vec::IntoIter,
 };
 
 use futures::{
+    FutureExt,
     future::{BoxFuture, Future},
     ready,
     stream::{FuturesUnordered, StreamExt},
 };
+use futures_timer::Delay;
 use libp2p_core::muxing::StreamMuxerBox;
 use libp2p_identity::PeerId;
 
-use crate::{Multiaddr, transport::TransportError};
+use crate::{Multiaddr, connection::pool::dial_ranker::rank_dials, transport::TransportError};
 
-type Dial = BoxFuture<
+/// A pending outbound dial that hasn't started yet.
+///
+/// The `fut` is created upfront by `Transport::dial()` but won't be polled
+/// until `delay` elapses. If `smart_dial` is enabled, `rank_dials` sets the
+/// `delay` and reorders the vec so preferred addresses start first. Without
+/// smart dial, delays are `None` and the concurrency factor limits how many
+/// are polled at once.
+pub(crate) struct PendingDial {
+    pub(crate) addr: Multiaddr,
+    pub(crate) fut: DialFuture,
+}
+
+/// The async dial operation. Owns the actual transport dial and returns the
+/// dialed address along with the result so the pool can match failures
+/// back to specific addresses.
+pub(crate) type DialFuture = BoxFuture<
     'static,
     (
         Multiaddr,
         Result<(PeerId, StreamMuxerBox), TransportError<std::io::Error>>,
     ),
 >;
 
+/// The result of a concurrent or smart dial to a single peer.
+///
+/// Returns the first successful address and its negotiated connection, along
+/// with any errors from earlier failed dials. Returns all errors if every
+/// dial failed.
+pub(crate) type DialResult = Result<
+    (
+        Multiaddr,
+        (PeerId, StreamMuxerBox),
+        Vec<(Multiaddr, TransportError<std::io::Error>)>,
+    ),
+    Vec<(Multiaddr, TransportError<std::io::Error>)>,
+>;
+
+/// Drives concurrent dial attempts to a single peer, limited by a concurrency factor.
+///
+/// Starts up to `concurrency_factor` dials simultaneously. On failure, the next
+/// pending dial starts immediately, keeping the concurrency window filled until
+/// either a dial succeeds or all pending dials are exhausted.
 pub(crate) struct ConcurrentDial {
-    dials: FuturesUnordered<Dial>,
-    pending_dials: Box<dyn Iterator<Item = Dial> + Send>,
+    dials: FuturesUnordered<DialFuture>,
+    pending_dials: IntoIter<PendingDial>,
     errors: Vec<(Multiaddr, TransportError<std::io::Error>)>,
 }
 
 impl Unpin for ConcurrentDial {}
 
 impl ConcurrentDial {
-    pub(crate) fn new(pending_dials: Vec<Dial>, concurrency_factor: NonZeroU8) -> Self {
+    pub(crate) fn new(pending_dials: Vec<PendingDial>, concurrency_factor: NonZeroU8) -> Self {
         let mut pending_dials = pending_dials.into_iter();
 
+        // Fill the concurrency window: start up to `concurrency_factor` dials.
+        // As each dial completes (success or failure), `dial_pending()` is called
+        // to refill the window from remaining pending dials.
         let dials = FuturesUnordered::new();
-        for dial in pending_dials.by_ref() {
-            dials.push(dial);
-            if dials.len() == concurrency_factor.get() as usize {
-                break;
-            }
+        for dial in pending_dials
+            .by_ref()
+            .take(concurrency_factor.get() as usize)
+        {
+            dials.push(dial.fut);
         }
 
         Self {
             dials,
             errors: Default::default(),
-            pending_dials: Box::new(pending_dials),
+            pending_dials,
         }
     }
 }
 
 impl Future for ConcurrentDial {
-    type Output = Result<
-        // Either one dial succeeded, returning the negotiated [`PeerId`], the address, the
-        // muxer and the addresses and errors of the dials that failed before.
-        (
-            Multiaddr,
-            (PeerId, StreamMuxerBox),
-            Vec<(Multiaddr, TransportError<std::io::Error>)>,
-        ),
-        // Or all dials failed, thus returning the address and error for each dial.
-        Vec<(Multiaddr, TransportError<std::io::Error>)>,
-    >;
+    type Output = DialResult;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
         loop {
@@ -93,8 +123,66 @@ impl Future for ConcurrentDial {
                 Some((addr, Err(e))) => {
                     self.errors.push((addr, e));
                     if let Some(dial) = self.pending_dials.next() {
-                        self.dials.push(dial)
+                        self.dials.push(dial.fut);
+                    }
+                }
+                None => {
+                    return Poll::Ready(Err(std::mem::take(&mut self.errors)));
+                }
+            }
+        }
+    }
+}
+
+/// Drives ranked dial attempts to a single peer with staggered delays.
+///
+/// All dials start immediately via [`rank_dials`], which assigns delays based on
+/// transport priority (QUIC > TCP, IPv6 > IPv4) and Happy Eyeballs (RFC 8305).
+/// The delays pace the dials, giving faster transports a head start while slower
+/// paths wait, with all dials overlapping in flight.
+pub(crate) struct SmartDial {
+    dials: FuturesUnordered<DialFuture>,
+    errors: Vec<(Multiaddr, TransportError<std::io::Error>)>,
+}
+
+impl Unpin for SmartDial {}
+
+impl SmartDial {
+    pub(crate) fn new(pending_dials: Vec<PendingDial>) -> Self {
+        let pending_dials = rank_dials(pending_dials);
+
+        let dials = FuturesUnordered::new();
+        for (delay, dial) in pending_dials {
+            dials.push(
+                async move {
+                    if !delay.is_zero() {
+                        Delay::new(delay).await;
                     }
+                    dial.fut.await
+                }
+                .boxed(),
+            );
+        }
+
+        Self {
+            dials,
+            errors: Default::default(),
+        }
+    }
+}
+
+impl Future for SmartDial {
+    type Output = DialResult;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        loop {
+            match ready!(self.dials.poll_next_unpin(cx)) {
+                Some((addr, Ok(output))) => {
+                    let errors = std::mem::take(&mut self.errors);
+                    return Poll::Ready(Ok((addr, output, errors)));
+                }
+                Some((addr, Err(e))) => {
+                    self.errors.push((addr, e));
                 }
                 None => {
                     return Poll::Ready(Err(std::mem::take(&mut self.errors)));