Problem integrating libp2p with bevy

[Looooong]

I'm using libp2p version 0.36.0. I have some problem trying to integrate libp2p with bevy. I have the following class:

pub struct PartyNetwork {
    pub swarm: Mutex<Swarm<Gossipsub>>,
}

impl PartyNetwork {
    pub fn new() -> Self {
        let local_key = Keypair::generate_ed25519();
        let local_peer_id = PeerId::from_public_key(local_key.public());

        let transport =
            libp2p::futures::executor::block_on(libp2p::development_transport(local_key.clone()))
                .unwrap();

        let gossipsub_config = GossipsubConfigBuilder::default()
            .heartbeat_interval(Duration::from_secs(10))
            .build()
            .unwrap();
        let mut gossipsub = Gossipsub::new(
            MessageAuthenticity::Signed(local_key.clone()),
            gossipsub_config,
        )
        .unwrap();
        gossipsub.subscribe(&IdentTopic::new("chat")).unwrap();

        let mut swarm = Swarm::new(transport, gossipsub, local_peer_id);
        Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse().unwrap()).unwrap();
        let swarm = Mutex::new(swarm);

        info!("Local peer ID: {:?}", local_peer_id);

        Self { swarm }
    }
}

And the following system:

fn poll_party_network(party_network: Res<PartyNetwork>) {
    let mut party_network = party_network.swarm.lock().unwrap();

    match party_network.next_event().now_or_never() {
        Some(SwarmEvent::NewListenAddr(addr)) => info!("Listening to: {:?}", addr),
        Some(SwarmEvent::ConnectionEstablished {
            endpoint, peer_id, ..
        }) => info!("Connected to peer {:?} at: {:?}", peer_id, endpoint),
        Some(event) => info!("Swarm event: {:?}", event),
        _ => (),
    }
}

The NewListenAddr events are fired. But when I try to dial to other peer in localhost, there is no event until I receive UnknownPeerUnreachableAddr.

On the other hand, I created a similar example that connects properly without bevy:

use libp2p::futures::executor::block_on;
use libp2p::futures::prelude::*;
use libp2p::gossipsub::{IdentTopic as Topic, MessageAuthenticity};
use libp2p::{gossipsub, identity, PeerId};
use std::time::Duration;

fn main() {
    // Create a random PeerId
    let local_key = identity::Keypair::generate_ed25519();
    let local_peer_id = PeerId::from(local_key.public());
    println!("Local peer id: {:?}", local_peer_id);

    // Set up an encrypted TCP Transport over the Mplex and Yamux protocols
    let transport = block_on(libp2p::development_transport(local_key.clone())).unwrap();

    // Create a Gossipsub topic
    let topic = Topic::new("test-net");

    // Create a Swarm to manage peers and events
    let mut swarm = {
        // Set a custom gossipsub
        let gossipsub_config = gossipsub::GossipsubConfigBuilder::default()
            .heartbeat_interval(Duration::from_secs(10)) // This is set to aid debugging by not cluttering the log space
            .build()
            .expect("Valid config");
        // build a gossipsub network behaviour
        let mut gossipsub: gossipsub::Gossipsub =
            gossipsub::Gossipsub::new(MessageAuthenticity::Signed(local_key), gossipsub_config)
                .expect("Correct configuration");

        // subscribes to our topic
        gossipsub.subscribe(&topic).unwrap();

        // build the swarm
        libp2p::Swarm::new(transport, gossipsub, local_peer_id)
    };

    // Listen on all interfaces and whatever port the OS assigns
    libp2p::Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse().unwrap()).unwrap();

    // Reach out to another node if specified
    if let Some(to_dial) = std::env::args().nth(1) {
        let dialing = to_dial.clone();
        match to_dial.parse() {
            Ok(to_dial) => match libp2p::Swarm::dial_addr(&mut swarm, to_dial) {
                Ok(_) => println!("Dialed {:?}", dialing),
                Err(e) => println!("Dial {:?} failed: {:?}", dialing, e),
            },
            Err(err) => println!("Failed to parse address to dial: {:?}", err),
        }
    }

    // Kick it off
    loop {
        match swarm.next_event().now_or_never() {
            Some(event) => println!("Event: {:?}", event),
            None => (),
        }
    }
}

What am I missing here?

[thomaseizinger]

Are systems in bevy called regularly? You can think of the Swarm as a giant Stream. It will not make progress unless you poll it (which is what happens under the hood when you call next_event.

1. Make sure poll_party_network is called on a regular basis. Unless it is called, the Swarm will not do anything, that includes even basic things like reading from the network buffer to react to an incoming connection.
2. Instead of a Mutex, try to use ResMut to get a mutable reference.

[Looooong]

1. Yes, the system is called frequently.

2. I still need Mutex because Swarm is not safe to be shared between thread. Also, changing to ResMut doesn't change the outcome.

Another thing to note that, when I switch from now_or_never() to futures::executor::block_on(), it connects correctly.

[Looooong]

I can reproduce the same behaviour in the example without bevy. When I add thread::sleep between the loop, even with a small duration, it won't connect and will result in dial timeout error.

    loop {
        match swarm.next_event().now_or_never() {
            Some(event) => println!("Event: {:?}", event),
            None => (),
        }

        thread::sleep(Duration::from_secs_f32(0.01));
    }

[thomaseizinger]

I think that using futures::executor::block_on() within a bevy system is unlikely desirable because it will block your thread and may cause problems in producing stable frame rates?

I am guessing your issue comes from the use of now_or_never. This uses a "noop-waker` and god knows what issues that causes with low-level futures like tcp sockets.

I think the proper way of integrating libp2p into bevy might be through some sort of async-system support where you can properly .await the swarm. Some quick research turned up bevyengine/bevy#1900.

[Looooong]

I just add futures::executor::block_on() for testing purpose. Of course, it will block the thread and prevent bevy from processing further. Still, I managed to integrate libp2p by spawning a separate task thread in the IoTaskPool. That task thread owns the swarm object, poll for swarm events and communicates with bevy app through mcsp.

impl FromWorld for NetworkManager {
    fn from_world(world: &mut World) -> Self {
        let (command_tx, mut command_rx) =
            libp2p::futures::channel::mpsc::unbounded::<NetworkCommand>();
        let (event_tx, event_rx) =
            std::sync::mpsc::channel::<SwarmEvent<GossipsubEvent, GossipsubHandlerError>>();
        let event_rx = Mutex::new(event_rx);

        let io_task_pool = world.get_resource::<IoTaskPool>().unwrap();
        io_task_pool
            .spawn(async move {
                let mut swarm = create_network_swarm();
                swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse().unwrap()).unwrap();

                loop {
                    select! {
                        command = command_rx.select_next_some() => {
                            match command {
                                NetworkCommand::Dial(addr) => swarm.dial_addr(addr).unwrap(),
                                NetworkCommand::Subscribe(topic) => swarm.behaviour_mut().subscribe(&topic).map(|_| ()).unwrap(),
                                NetworkCommand::Unsubscribe(topic) => swarm.behaviour_mut().unsubscribe(&topic).map(|_| ()).unwrap(),
                                NetworkCommand::Publish(topic, data) => swarm.behaviour_mut().publish(topic, data).map(|_| ()).unwrap(),
                            };
                        }
                        event = swarm.next_event().fuse() => event_tx.send(event).unwrap(),
                    }
                }
            })
            .detach();

        Self {
            command_tx,
            event_rx,
        }
    }
}