Expose reuse_port option for TCP and WebSocket transports (#69)

Add `reuse_port` option to TCP & WebSocket configs. When set, this both:
1. Sets `SO_REUSEPORT` on sockets.
2. Binds sockets to the listen address port when making outbound
connections.

I'm not sure if these two behaviors should be controlled by a single
option, but this follows libp2p convention.

Author: dmitry-markin

src/transport/tcp/config.rs

@@ -30,9 +30,17 @@ use crate::{
 pub struct Config {
     /// Listen address for the transport.
     ///
-    /// Default listen addres is `/ip6/::1/tcp`.
+    /// Default listen addresses are ["/ip4/0.0.0.0/tcp/0", "/ip6/::/tcp/0"].
     pub listen_addresses: Vec<multiaddr::Multiaddr>,
 
+    /// Whether to set `SO_REUSEPORT` and bind a socket to the listen address port for outbound
+    /// connections.
+    ///
+    /// Note that `SO_REUSEADDR` is always set on listening sockets.
+    ///
+    /// Defaults to `true`.
+    pub reuse_port: bool,
+
     /// Yamux configuration.
     pub yamux_config: crate::yamux::Config,
 
@@ -76,6 +84,7 @@ impl Default for Config {
                 "/ip4/0.0.0.0/tcp/0".parse().expect("valid address"),
                 "/ip6/::/tcp/0".parse().expect("valid address"),
             ],
+            reuse_port: true,
             yamux_config: Default::default(),
             noise_read_ahead_frame_count: MAX_READ_AHEAD_FACTOR,
             noise_write_buffer_size: MAX_WRITE_BUFFER_SIZE,

src/transport/tcp/listener.rs

@@ -56,40 +56,62 @@ pub struct TcpListener {
     listeners: Vec<TokioTcpListener>,
 }
 
-#[derive(Clone, Default)]
-pub struct DialAddresses {
-    /// Listen addresses.
-    listen_addresses: Arc<Vec<SocketAddr>>,
+/// Local addresses to use for outbound connections.
+#[derive(Clone)]
+pub enum DialAddresses {
+    /// Reuse port from listen addresses.
+    Reuse {
+        listen_addresses: Arc<Vec<SocketAddr>>,
+    },
+    /// Do not reuse port.
+    NoReuse,
+}
+
+impl Default for DialAddresses {
+    fn default() -> Self {
+        DialAddresses::NoReuse
+    }
 }
 
 impl DialAddresses {
     /// Get local dial address for an outbound connection.
-    pub(super) fn local_dial_address(&self, remote_address: &IpAddr) -> Option<SocketAddr> {
-        for address in self.listen_addresses.iter() {
-            if remote_address.is_ipv4() == address.is_ipv4()
-                && remote_address.is_loopback() == address.ip().is_loopback()
-            {
-                if remote_address.is_ipv4() {
-                    return Some(SocketAddr::new(
-                        IpAddr::V4(Ipv4Addr::UNSPECIFIED),
-                        address.port(),
-                    ));
-                } else {
-                    return Some(SocketAddr::new(
-                        IpAddr::V6(Ipv6Addr::UNSPECIFIED),
-                        address.port(),
-                    ));
+    pub(super) fn local_dial_address(
+        &self,
+        remote_address: &IpAddr,
+    ) -> Result<Option<SocketAddr>, ()> {
+        match self {
+            DialAddresses::Reuse { listen_addresses } => {
+                for address in listen_addresses.iter() {
+                    if remote_address.is_ipv4() == address.is_ipv4()
+                        && remote_address.is_loopback() == address.ip().is_loopback()
+                    {
+                        if remote_address.is_ipv4() {
+                            return Ok(Some(SocketAddr::new(
+                                IpAddr::V4(Ipv4Addr::UNSPECIFIED),
+                                address.port(),
+                            )));
+                        } else {
+                            return Ok(Some(SocketAddr::new(
+                                IpAddr::V6(Ipv6Addr::UNSPECIFIED),
+                                address.port(),
+                            )));
+                        }
+                    }
                 }
+
+                Err(())
             }
+            DialAddresses::NoReuse => Ok(None),
         }
-
-        None
     }
 }
 
 impl TcpListener {
     /// Create new [`TcpListener`]
-    pub fn new(addresses: Vec<Multiaddr>) -> (Self, Vec<Multiaddr>, DialAddresses) {
+    pub fn new(
+        addresses: Vec<Multiaddr>,
+        reuse_port: bool,
+    ) -> (Self, Vec<Multiaddr>, DialAddresses) {
         let (listeners, listen_addresses): (_, Vec<Vec<_>>) = addresses
             .into_iter()
             .filter_map(|address| {
@@ -117,7 +139,9 @@ impl TcpListener {
                 socket.set_nonblocking(true).ok()?;
                 socket.set_reuse_address(true).ok()?;
                 #[cfg(unix)]
-                socket.set_reuse_port(true).ok()?;
+                if reuse_port {
+                    socket.set_reuse_port(true).ok()?;
+                }
                 socket.bind(&address.into()).ok()?;
                 socket.listen(1024).ok()?;
 
@@ -176,14 +200,15 @@ impl TcpListener {
                     .with(Protocol::Tcp(address.port()))
             })
             .collect();
-
-        (
-            Self { listeners },
-            listen_multi_addresses,
-            DialAddresses {
+        let dial_addresses = if reuse_port {
+            DialAddresses::Reuse {
                 listen_addresses: Arc::new(listen_addresses),
-            },
-        )
+            }
+        } else {
+            DialAddresses::NoReuse
+        };
+
+        (Self { listeners }, listen_multi_addresses, dial_addresses)
     }
 
     /// Extract socket address and `PeerId`, if found, from `address`.
@@ -319,7 +344,7 @@ mod tests {
 
     #[tokio::test]
     async fn no_listeners() {
-        let (mut listener, _, _) = TcpListener::new(Vec::new());
+        let (mut listener, _, _) = TcpListener::new(Vec::new(), true);
 
         futures::future::poll_fn(|cx| match listener.poll_next_unpin(cx) {
             Poll::Pending => Poll::Ready(()),
@@ -331,7 +356,7 @@ mod tests {
     #[tokio::test]
     async fn one_listener() {
         let address: Multiaddr = "/ip6/::1/tcp/0".parse().unwrap();
-        let (mut listener, listen_addresses, _) = TcpListener::new(vec![address.clone()]);
+        let (mut listener, listen_addresses, _) = TcpListener::new(vec![address.clone()], true);
         let Some(Protocol::Tcp(port)) =
             listen_addresses.iter().next().unwrap().clone().iter().skip(1).next()
         else {
@@ -348,7 +373,7 @@ mod tests {
     async fn two_listeners() {
         let address1: Multiaddr = "/ip6/::1/tcp/0".parse().unwrap();
         let address2: Multiaddr = "/ip4/127.0.0.1/tcp/0".parse().unwrap();
-        let (mut listener, listen_addresses, _) = TcpListener::new(vec![address1, address2]);
+        let (mut listener, listen_addresses, _) = TcpListener::new(vec![address1, address2], true);
         let Some(Protocol::Tcp(port1)) =
             listen_addresses.iter().next().unwrap().clone().iter().skip(1).next()
         else {
@@ -373,7 +398,7 @@ mod tests {
 
     #[tokio::test]
     async fn local_dial_address() {
-        let dial_addresses = DialAddresses {
+        let dial_addresses = DialAddresses::Reuse {
             listen_addresses: Arc::new(vec![
                 "[2001:7d0:84aa:3900:2a5d:9e85::]:8888".parse().unwrap(),
                 "92.168.127.1:9999".parse().unwrap(),
@@ -382,20 +407,26 @@ mod tests {
 
         assert_eq!(
             dial_addresses.local_dial_address(&IpAddr::V4(Ipv4Addr::new(192, 168, 0, 1))),
-            Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 9999))
+            Ok(Some(SocketAddr::new(
+                IpAddr::V4(Ipv4Addr::UNSPECIFIED),
+                9999
+            ))),
         );
 
         assert_eq!(
             dial_addresses.local_dial_address(&IpAddr::V6(Ipv6Addr::new(0, 1, 2, 3, 4, 5, 6, 7))),
-            Some(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 8888))
+            Ok(Some(SocketAddr::new(
+                IpAddr::V6(Ipv6Addr::UNSPECIFIED),
+                8888
+            ))),
         );
     }
 
     #[tokio::test]
     async fn show_all_addresses() {
         let address1: Multiaddr = "/ip6/::/tcp/0".parse().unwrap();
         let address2: Multiaddr = "/ip4/0.0.0.0/tcp/0".parse().unwrap();
-        let (_, listen_addresses, _) = TcpListener::new(vec![address1, address2]);
+        let (_, listen_addresses, _) = TcpListener::new(vec![address1, address2], true);
 
         println!("{listen_addresses:#?}");
     }

src/transport/tcp/mod.rs

@@ -200,13 +200,14 @@ impl TcpTransport {
         socket.set_nonblocking(true)?;
 
         match dial_addresses.local_dial_address(&remote_address.ip()) {
-            Some(dial_address) => {
+            Ok(Some(dial_address)) => {
                 socket.set_reuse_address(true)?;
                 #[cfg(unix)]
                 socket.set_reuse_port(true)?;
                 socket.bind(&dial_address.into())?;
             }
-            None => {
+            Ok(None) => {}
+            Err(()) => {
                 tracing::debug!(
                     target: LOG_TARGET,
                     ?remote_address,
@@ -257,8 +258,10 @@ impl TransportBuilder for TcpTransport {
         );
 
         // start tcp listeners for all listen addresses
-        let (listener, listen_addresses, dial_addresses) =
-            TcpListener::new(std::mem::replace(&mut config.listen_addresses, Vec::new()));
+        let (listener, listen_addresses, dial_addresses) = TcpListener::new(
+            std::mem::replace(&mut config.listen_addresses, Vec::new()),
+            config.reuse_port,
+        );
 
         Ok((
             Self {

src/transport/websocket/config.rs

@@ -30,9 +30,17 @@ use crate::{
 pub struct Config {
     /// Listen address address for the transport.
     ///
-    /// Default listen addres is `/ip6/::1/tcp/ws`.
+    /// Default listen addreses are ["/ip4/0.0.0.0/tcp/0/ws", "/ip6/::/tcp/0/ws"].
     pub listen_addresses: Vec<multiaddr::Multiaddr>,
 
+    /// Whether to set `SO_REUSEPORT` and bind a socket to the listen address port for outbound
+    /// connections.
+    ///
+    /// Note that `SO_REUSEADDR` is always set on listening sockets.
+    ///
+    /// Defaults to `true`.
+    pub reuse_port: bool,
+
     /// Yamux configuration.
     pub yamux_config: crate::yamux::Config,
 
@@ -76,6 +84,7 @@ impl Default for Config {
                 "/ip4/0.0.0.0/tcp/0/ws".parse().expect("valid address"),
                 "/ip6/::/tcp/0/ws".parse().expect("valid address"),
             ],
+            reuse_port: true,
             yamux_config: Default::default(),
             noise_read_ahead_frame_count: MAX_READ_AHEAD_FACTOR,
             noise_write_buffer_size: MAX_WRITE_BUFFER_SIZE,