feat(swarm): Make executor for connection tasks explicit (#3097)

Previously, the executor for connection tasks silently defaulted to a `futures::executor::ThreadPool`. This causes issues such as #2230.

With this patch, we force the user to choose, which executor they want to run the connection tasks on which results in overall simpler API with less footguns.

Closes #3068.

Cargo.toml

@@ -47,7 +47,8 @@ full = [
     "websocket",
     "yamux",
 ]
-async-std = ["libp2p-mdns?/async-io", "libp2p-tcp?/async-io", "libp2p-dns?/async-std", "libp2p-quic?/async-std"]
+
+async-std = ["libp2p-swarm/async-std", "libp2p-mdns?/async-io", "libp2p-tcp?/async-io", "libp2p-dns?/async-std", "libp2p-quic?/async-std"]
 autonat = ["dep:libp2p-autonat"]
 dcutr = ["dep:libp2p-dcutr", "libp2p-metrics?/dcutr"]
 deflate = ["dep:libp2p-deflate"]
@@ -74,7 +75,7 @@ rsa = ["libp2p-core/rsa"]
 secp256k1 = ["libp2p-core/secp256k1"]
 serde = ["libp2p-core/serde", "libp2p-kad?/serde", "libp2p-gossipsub?/serde"]
 tcp = ["dep:libp2p-tcp"]
-tokio = ["libp2p-mdns?/tokio", "libp2p-tcp?/tokio", "libp2p-dns?/tokio", "libp2p-quic?/tokio"]
+tokio = ["libp2p-swarm/tokio", "libp2p-mdns?/tokio", "libp2p-tcp?/tokio", "libp2p-dns?/tokio", "libp2p-quic?/tokio"]
 uds = ["dep:libp2p-uds"]
 wasm-bindgen = ["futures-timer/wasm-bindgen", "instant/wasm-bindgen", "getrandom/js"]
 wasm-ext = ["dep:libp2p-wasm-ext"]

core/CHANGELOG.md

@@ -4,8 +4,11 @@
 
 - Hide `prost::Error` from public API in `FromEnvelopeError::InvalidPeerRecord` and `signed_envelope::DecodingError`. See [PR 3058].
 
+- Move `Executor` to `libp2p-swarm`. See [PR 3097].
+
 [PR 3031]: https://github.com/libp2p/rust-libp2p/pull/3031
 [PR 3058]: https://github.com/libp2p/rust-libp2p/pull/3058
+[PR 3097]: https://github.com/libp2p/rust-libp2p/pull/3097
 
 # 0.37.0
 

core/src/lib.rs

@@ -83,23 +83,3 @@ pub use upgrade::{InboundUpgrade, OutboundUpgrade, ProtocolName, UpgradeError, U
 #[derive(thiserror::Error, Debug)]
 #[error(transparent)]
 pub struct DecodeError(prost::DecodeError);
-
-use std::{future::Future, pin::Pin};
-
-/// Implemented on objects that can run a `Future` in the background.
-///
-/// > **Note**: While it may be tempting to implement this trait on types such as
-/// >           [`futures::stream::FuturesUnordered`], please note that passing an `Executor` is
-/// >           optional, and that `FuturesUnordered` (or a similar struct) will automatically
-/// >           be used as fallback by libp2p. The `Executor` trait should therefore only be
-/// >           about running `Future`s in the background.
-pub trait Executor {
-    /// Run the given future in the background until it ends.
-    fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>);
-}
-
-impl<F: Fn(Pin<Box<dyn Future<Output = ()> + Send>>)> Executor for F {
-    fn exec(&self, f: Pin<Box<dyn Future<Output = ()> + Send>>) {
-        self(f)
-    }
-}

examples/chat-tokio.rs

@@ -39,7 +39,7 @@ use libp2p::{
         TokioMdns,
     },
     mplex, noise,
-    swarm::{NetworkBehaviour, SwarmBuilder, SwarmEvent},
+    swarm::{NetworkBehaviour, SwarmEvent},
     tcp, Multiaddr, PeerId, Transport,
 };
 use std::error::Error;
@@ -97,23 +97,12 @@ async fn main() -> Result<(), Box<dyn Error>> {
     }
 
     // Create a Swarm to manage peers and events.
-    let mut swarm = {
-        let mdns = TokioMdns::new(Default::default())?;
-        let mut behaviour = MyBehaviour {
-            floodsub: Floodsub::new(peer_id),
-            mdns,
-        };
-
-        behaviour.floodsub.subscribe(floodsub_topic.clone());
-
-        SwarmBuilder::new(transport, behaviour, peer_id)
-            // We want the connection background tasks to be spawned
-            // onto the tokio runtime.
-            .executor(Box::new(|fut| {
-                tokio::spawn(fut);
-            }))
-            .build()
+    let mdns = TokioMdns::new(Default::default())?;
+    let behaviour = MyBehaviour {
+        floodsub: Floodsub::new(peer_id),
+        mdns,
     };
+    let mut swarm = libp2p_swarm::Swarm::with_tokio_executor(transport, behaviour, peer_id);
 
     // Reach out to another node if specified
     if let Some(to_dial) = std::env::args().nth(1) {

examples/chat.rs

@@ -115,7 +115,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         };
 
         behaviour.floodsub.subscribe(floodsub_topic.clone());
-        Swarm::new(transport, behaviour, local_peer_id)
+        Swarm::with_threadpool_executor(transport, behaviour, local_peer_id)
     };
 
     // Reach out to another node if specified

examples/distributed-key-value-store.rs

@@ -99,7 +99,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         let kademlia = Kademlia::new(local_peer_id, store);
         let mdns = Mdns::new(MdnsConfig::default())?;
         let behaviour = MyBehaviour { kademlia, mdns };
-        Swarm::new(transport, behaviour, local_peer_id)
+        Swarm::with_async_std_executor(transport, behaviour, local_peer_id)
     };
 
     // Read full lines from stdin

examples/file-sharing.rs

@@ -219,9 +219,7 @@ mod network {
         ProtocolSupport, RequestId, RequestResponse, RequestResponseCodec, RequestResponseEvent,
         RequestResponseMessage, ResponseChannel,
     };
-    use libp2p::swarm::{
-        ConnectionHandlerUpgrErr, NetworkBehaviour, Swarm, SwarmBuilder, SwarmEvent,
-    };
+    use libp2p::swarm::{ConnectionHandlerUpgrErr, NetworkBehaviour, Swarm, SwarmEvent};
     use std::collections::{hash_map, HashMap, HashSet};
     use std::iter;
 
@@ -252,7 +250,7 @@ mod network {
 
         // Build the Swarm, connecting the lower layer transport logic with the
         // higher layer network behaviour logic.
-        let swarm = SwarmBuilder::new(
+        let swarm = Swarm::with_threadpool_executor(
             libp2p::development_transport(id_keys).await?,
             ComposedBehaviour {
                 kademlia: Kademlia::new(peer_id, MemoryStore::new(peer_id)),
@@ -263,8 +261,7 @@ mod network {
                 ),
             },
             peer_id,
-        )
-        .build();
+        );
 
         let (command_sender, command_receiver) = mpsc::channel(0);
         let (event_sender, event_receiver) = mpsc::channel(0);

examples/gossipsub-chat.rs

@@ -110,7 +110,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     let mut swarm = {
         let mdns = Mdns::new(MdnsConfig::default())?;
         let behaviour = MyBehaviour { gossipsub, mdns };
-        Swarm::new(transport, behaviour, local_peer_id)
+        Swarm::with_async_std_executor(transport, behaviour, local_peer_id)
     };
 
     // Read full lines from stdin

examples/ipfs-kad.rs

@@ -68,7 +68,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
             behaviour.add_address(&PeerId::from_str(peer)?, bootaddr.clone());
         }
 
-        Swarm::new(transport, behaviour, local_peer_id)
+        Swarm::with_async_std_executor(transport, behaviour, local_peer_id)
     };
 
     // Order Kademlia to search for a peer.

examples/ipfs-private.rs

@@ -204,7 +204,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
         println!("Subscribing to {gossipsub_topic:?}");
         behaviour.gossipsub.subscribe(&gossipsub_topic).unwrap();
-        Swarm::new(transport, behaviour, local_peer_id)
+        Swarm::with_async_std_executor(transport, behaviour, local_peer_id)
     };
 
     // Reach out to other nodes if specified

examples/mdns-passive-discovery.rs

@@ -45,7 +45,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     // Create a Swarm that establishes connections through the given transport.
     // Note that the MDNS behaviour itself will not actually inititiate any connections,
     // as it only uses UDP.
-    let mut swarm = Swarm::new(transport, behaviour, peer_id);
+    let mut swarm = Swarm::with_async_std_executor(transport, behaviour, peer_id);
     swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
 
     loop {

examples/ping.rs

@@ -54,7 +54,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let transport = libp2p::development_transport(local_key).await?;
 
-    let mut swarm = Swarm::new(transport, Behaviour::default(), local_peer_id);
+    let mut swarm = Swarm::with_async_std_executor(transport, Behaviour::default(), local_peer_id);
 
     // Tell the swarm to listen on all interfaces and a random, OS-assigned
     // port.

misc/metrics/examples/metrics/main.rs

@@ -70,7 +70,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     let local_peer_id = PeerId::from(local_key.public());
     info!("Local peer id: {:?}", local_peer_id);
 
-    let mut swarm = Swarm::new(
+    let mut swarm = Swarm::without_executor(
         block_on(libp2p::development_transport(local_key))?,
         Behaviour::default(),
         local_peer_id,

misc/multistream-select/tests/transport.rs

@@ -61,8 +61,10 @@ fn transport_upgrade() {
 
         let listen_addr = Multiaddr::from(Protocol::Memory(random::<u64>()));
 
-        let mut dialer = Swarm::new(dialer_transport, dummy::Behaviour, dialer_id);
-        let mut listener = Swarm::new(listener_transport, dummy::Behaviour, listener_id);
+        let mut dialer =
+            Swarm::with_async_std_executor(dialer_transport, dummy::Behaviour, dialer_id);
+        let mut listener =
+            Swarm::with_async_std_executor(listener_transport, dummy::Behaviour, listener_id);
 
         listener.listen_on(listen_addr).unwrap();
         let (addr_sender, addr_receiver) = oneshot::channel();

protocols/autonat/examples/autonat_client.rs

@@ -67,7 +67,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let behaviour = Behaviour::new(local_key.public());
 
-    let mut swarm = Swarm::new(transport, behaviour, local_peer_id);
+    let mut swarm = Swarm::with_async_std_executor(transport, behaviour, local_peer_id);
     swarm.listen_on(
         Multiaddr::empty()
             .with(Protocol::Ip4(Ipv4Addr::UNSPECIFIED))

protocols/autonat/examples/autonat_server.rs

@@ -57,7 +57,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let behaviour = Behaviour::new(local_key.public());
 
-    let mut swarm = Swarm::new(transport, behaviour, local_peer_id);
+    let mut swarm = Swarm::with_async_std_executor(transport, behaviour, local_peer_id);
     swarm.listen_on(
         Multiaddr::empty()
             .with(Protocol::Ip4(Ipv4Addr::UNSPECIFIED))

protocols/autonat/tests/test_client.rs

@@ -40,7 +40,7 @@ async fn init_swarm(config: Config) -> Swarm<Behaviour> {
     let local_id = PeerId::from_public_key(&keypair.public());
     let transport = development_transport(keypair).await.unwrap();
     let behaviour = Behaviour::new(local_id, config);
-    Swarm::new(transport, behaviour, local_id)
+    Swarm::with_async_std_executor(transport, behaviour, local_id)
 }
 
 async fn spawn_server(kill: oneshot::Receiver<()>) -> (PeerId, Multiaddr) {

protocols/autonat/tests/test_server.rs

@@ -39,7 +39,7 @@ async fn init_swarm(config: Config) -> Swarm<Behaviour> {
     let local_id = PeerId::from_public_key(&keypair.public());
     let transport = development_transport(keypair).await.unwrap();
     let behaviour = Behaviour::new(local_id, config);
-    Swarm::new(transport, behaviour, local_id)
+    Swarm::with_async_std_executor(transport, behaviour, local_id)
 }
 
 async fn init_server(config: Option<Config>) -> (Swarm<Behaviour>, PeerId, Multiaddr) {

protocols/dcutr/examples/dcutr.rs

@@ -19,7 +19,7 @@
 // DEALINGS IN THE SOFTWARE.
 
 use clap::Parser;
-use futures::executor::block_on;
+use futures::executor::{block_on, ThreadPool};
 use futures::future::FutureExt;
 use futures::stream::StreamExt;
 use libp2p::core::multiaddr::{Multiaddr, Protocol};
@@ -155,9 +155,12 @@ fn main() -> Result<(), Box<dyn Error>> {
         dcutr: dcutr::behaviour::Behaviour::new(),
     };
 
-    let mut swarm = SwarmBuilder::new(transport, behaviour, local_peer_id)
-        .dial_concurrency_factor(10_u8.try_into().unwrap())
-        .build();
+    let mut swarm = match ThreadPool::new() {
+        Ok(tp) => SwarmBuilder::with_executor(transport, behaviour, local_peer_id, tp),
+        Err(_) => SwarmBuilder::without_executor(transport, behaviour, local_peer_id),
+    }
+    .dial_concurrency_factor(10_u8.try_into().unwrap())
+    .build();
 
     swarm
         .listen_on(

protocols/dcutr/tests/lib.rs

@@ -98,7 +98,7 @@ fn build_relay() -> Swarm<relay::Relay> {
 
     let transport = build_transport(MemoryTransport::default().boxed(), local_public_key);
 
-    Swarm::new(
+    Swarm::with_threadpool_executor(
         transport,
         relay::Relay::new(
             local_peer_id,
@@ -122,7 +122,7 @@ fn build_client() -> Swarm<Client> {
         local_public_key,
     );
 
-    Swarm::new(
+    Swarm::with_threadpool_executor(
         transport,
         Client {
             relay: behaviour,

protocols/gossipsub/tests/smoke.rs

@@ -171,7 +171,7 @@ fn build_node() -> (Multiaddr, Swarm<Gossipsub>) {
         .build()
         .unwrap();
     let behaviour = Gossipsub::new(MessageAuthenticity::Author(peer_id), config).unwrap();
-    let mut swarm = Swarm::new(transport, behaviour, peer_id);
+    let mut swarm = Swarm::without_executor(transport, behaviour, peer_id);
 
     let port = 1 + random::<u64>();
     let mut addr: Multiaddr = Protocol::Memory(port).into();

protocols/identify/examples/identify.rs

@@ -55,7 +55,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         local_key.public(),
     ));
 
-    let mut swarm = Swarm::new(transport, behaviour, local_peer_id);
+    let mut swarm = Swarm::with_async_std_executor(transport, behaviour, local_peer_id);
 
     // Tell the swarm to listen on all interfaces and a random, OS-assigned
     // port.

protocols/identify/src/behaviour.rs

@@ -584,7 +584,7 @@ mod tests {
             let protocol = Behaviour::new(
                 Config::new("a".to_string(), pubkey.clone()).with_agent_version("b".to_string()),
             );
-            let swarm = Swarm::new(transport, protocol, pubkey.to_peer_id());
+            let swarm = Swarm::with_async_std_executor(transport, protocol, pubkey.to_peer_id());
             (swarm, pubkey)
         };
 
@@ -593,7 +593,7 @@ mod tests {
             let protocol = Behaviour::new(
                 Config::new("c".to_string(), pubkey.clone()).with_agent_version("d".to_string()),
             );
-            let swarm = Swarm::new(transport, protocol, pubkey.to_peer_id());
+            let swarm = Swarm::with_async_std_executor(transport, protocol, pubkey.to_peer_id());
             (swarm, pubkey)
         };
 
@@ -661,7 +661,7 @@ mod tests {
         let (mut swarm1, pubkey1) = {
             let (pubkey, transport) = transport();
             let protocol = Behaviour::new(Config::new("a".to_string(), pubkey.clone()));
-            let swarm = Swarm::new(transport, protocol, pubkey.to_peer_id());
+            let swarm = Swarm::with_async_std_executor(transport, protocol, pubkey.to_peer_id());
             (swarm, pubkey)
         };
 
@@ -670,7 +670,7 @@ mod tests {
             let protocol = Behaviour::new(
                 Config::new("a".to_string(), pubkey.clone()).with_agent_version("b".to_string()),
             );
-            let swarm = Swarm::new(transport, protocol, pubkey.to_peer_id());
+            let swarm = Swarm::with_async_std_executor(transport, protocol, pubkey.to_peer_id());
             (swarm, pubkey)
         };
 
@@ -742,7 +742,7 @@ mod tests {
                     .with_initial_delay(Duration::from_secs(10)),
             );
 
-            Swarm::new(transport, protocol, pubkey.to_peer_id())
+            Swarm::with_async_std_executor(transport, protocol, pubkey.to_peer_id())
         };
 
         let mut swarm2 = {
@@ -751,7 +751,7 @@ mod tests {
                 Config::new("a".to_string(), pubkey.clone()).with_agent_version("b".to_string()),
             );
 
-            Swarm::new(transport, protocol, pubkey.to_peer_id())
+            Swarm::with_async_std_executor(transport, protocol, pubkey.to_peer_id())
         };
 
         let swarm1_peer_id = *swarm1.local_peer_id();

protocols/kad/src/behaviour/test.rs

@@ -66,7 +66,7 @@ fn build_node_with_config(cfg: KademliaConfig) -> (Multiaddr, TestSwarm) {
     let store = MemoryStore::new(local_id);
     let behaviour = Kademlia::with_config(local_id, store, cfg);
 
-    let mut swarm = Swarm::new(transport, behaviour, local_id);
+    let mut swarm = Swarm::without_executor(transport, behaviour, local_id);
 
     let address: Multiaddr = Protocol::Memory(random::<u64>()).into();
     swarm.listen_on(address.clone()).unwrap();

protocols/mdns/tests/use-async-std.rs

@@ -62,7 +62,7 @@ async fn create_swarm(config: MdnsConfig) -> Result<Swarm<Mdns>, Box<dyn Error>>
     let peer_id = PeerId::from(id_keys.public());
     let transport = libp2p::development_transport(id_keys).await?;
     let behaviour = Mdns::new(config)?;
-    let mut swarm = Swarm::new(transport, behaviour, peer_id);
+    let mut swarm = Swarm::with_async_std_executor(transport, behaviour, peer_id);
     swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
     Ok(swarm)
 }

protocols/mdns/tests/use-tokio.rs

@@ -58,7 +58,7 @@ async fn create_swarm(config: MdnsConfig) -> Result<Swarm<TokioMdns>, Box<dyn Er
     let peer_id = PeerId::from(id_keys.public());
     let transport = libp2p::tokio_development_transport(id_keys)?;
     let behaviour = TokioMdns::new(config)?;
-    let mut swarm = Swarm::new(transport, behaviour, peer_id);
+    let mut swarm = Swarm::with_tokio_executor(transport, behaviour, peer_id);
     swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
     Ok(swarm)
 }