How do I call into libp2p rather than having it call my functions in a loop?

[Frederik-Baetens]

Hi!

I'm trying to use libp2p in conjunction with an rpc library, so libp2p isn't the main point where requests arrive. However, in the tokio chat example, it seems to be necessary to continiously loop over event = swarm.next() to drive? libp2p.

Is there a different way to submit that call to tokio, or am I required to create a separate thread which has the same kind of loop as in the tokio chat example, which communicates with the rest of my program through channels?

[thomaseizinger]

Is there a different way to submit that call to tokio, or am I required to create a separate thread which has the same kind of loop as in the tokio chat example, which communicates with the rest of my program through channels?

It doesn't have to be a separate thread. What I've found to work reasonably well is to use the tokio or futures select! macro to poll multiple futures simultaneously. You can see an example of this here:

https://github.com/comit-network/xmr-btc-swap/blob/475aac1387846c8e8351bcd4d3bef4a7e019e806/swap/src/protocol/bob/event_loop.rs#L89-L193

The naming might be bit odd but the way it works is that we have a thing called EventLoop that has ownership of the Swarm and can hence call functions that require &mut self. We then also have a thing called EventLoopHandle which holds either sending or receiving ends of bmrng (request-response) channels. The EventLoop basically "listens" on events from either the Swarm or any of the channels and whatever resolves first gets processed, distributed to wherever it needs to and then the loop starts again.

Another thing that might be interesting in our use of rust-libp2p is that we more or less exclusively use event_process = false on the NetworkBehaviour. That forces you to only implement From to bubble the events up and all the actually processing happens in the EventLoop and not in NetworkBehaviourEventProcess implementations.

An alternative way of doing things is to wrap the Swarm in an Arc<Mutex> and simply have a task that continuously locks and polls the Swarm and you give a copy of the Arc to any part of the program that needs to deal with the Swarm.

[Frederik-Baetens]

Thanks for the great explanation!

I saw that in that xmr-btc example, the run method of the eventloop is still polled as well, kind of partially defeating the purpose of the eventloop abstraction (still useful for ownership purposes of course). https://github.com/comit-network/xmr-btc-swap/blob/master/swap/src/bin/swap.rs#L213

Is there a way that I could avoid leaking the polling of that run method with a select! macro outside of the impl Eventloop code?

So something like this:

impl SwarmOwner {
    pub async fn new() -> SwarmOwner {
        let swarm = create_gossipsub_swarm;
        let (sender, receiver) = mpsc::channel(100);
        let swarmowner = SwarmOwner {swarm, sender, receiver,};
        tokio::spawn(swarmowner.run_swarm(swarm));
        swarmowner
    }
...
}

Should I just spawn a separate thread and .await it there?

Or is spawning a task without awaiting it a sane thing to do here, and is the select! just for graceful termination purposes?

Edit: i think i can, i'm starting to figure it out, i'll post my solution later for others who might read this

[thomaseizinger]

is the select! just for graceful termination purposes?

Primarily that yes. It is also exposed for testing purposes so we can manually start and stop things to test loss of network connection and such.

Having said that, no software design is ever ideal so I am sure there are improvements to be made on the code I linked to 😁

[Frederik-Baetens]

So this is what I ended up with:

I also created a handle with a channel sender in it, and then an actual struct which owns the swarm, and serves as an actor. I'm just using tokio::spawn to run this actor instead of using a select! macro in the top level. The run function itself still utilizes the select macro, which I think is obligatory.

The create_gossipsub_swarm function contains similar code to the swarm creation in the tokio chat example, but then adapted with gossipsub code. If you think this would be interesting, I could make a pr adding this example. I think it would be helpful since there's enough differences between floodsub & gossipsub that converting between the two isn't always immediately obvious. My modified tokiochat example also shows off the "event_process=false" which I think could also be educational.

#[derive(Serialize, Deserialize, Debug)]
pub struct MyMessage {
    pub message: String,
}

//Currently it's a bit weird in this code to pass the topic with every message in the channel,
//since my swarm is only subscribed on one hardcoded topic in this example.
//If you would add methods to subscribe to new channels,
//you could pass different topics with every channel message that you want to be published,
//allowing you to publish to different topics on the fly
#[derive(Clone)]
pub struct SwarmHandle {
    sender: Sender<(MyMessage, IdentTopic)>,
    topic: IdentTopic,
}

struct SwarmOwner {
    swarm: Swarm<MyBehaviour>,
    receiver: Receiver<(MyMessage, IdentTopic)>,
}

impl SwarmHandle {
    pub async fn new() -> Result<SwarmHandle, Box<dyn Error>> {
        let (sender, receiver) = mpsc::channel(100);
        
        //A vec is passed here for better peer disovery in the future, but for now it's just empty
        let (swarm, topic) = create_gossipsub_swarm(Vec::new()).await?;
        let swarm_owner = SwarmOwner { swarm, receiver };
        let swarm_handle = SwarmHandle {sender: sender.clone(), topic };
        tokio::spawn(swarm_owner.run_swarm());
        Ok(swarm_handle)
    }
    
    pub async fn publish(&self, message: String ) {
        self.sender.send((MyMessage{message}, self.topic.clone())).await.unwrap();
    }

}

impl SwarmOwner {
    async fn run_swarm(mut self) {
        let mut listening = false;
        loop {
            let to_publish = {
                tokio::select! {
                    sibling = self.receiver.recv() => {
                        let (sibling, topic) = sibling.unwrap();
                        println!("publishing sibling: {:?}", sibling);
                        Some((topic, sibling))
                    }
                    event = self.swarm.next() => {
                        match event {
                            OutEvent::MdnsEvent(mdnsevent) => {
                                match mdnsevent {
                                    MdnsEvent::Discovered(list) => {
                                        for (peer, _) in list {
                                            self.swarm.gossipsub.add_explicit_peer(&peer);
                                        }
                                        None
                                    },
                                    MdnsEvent::Expired(list) => {
                                        for (peer, _) in list {
                                            if !self.swarm.mdns.has_node(&peer) {
                                                self.swarm.gossipsub.remove_explicit_peer(&peer);
                                            }
                                        }
                                        None
                                    }
                                }
                            },
                            OutEvent::GossipsubEvent(gossipsubevent) => {
                                if let GossipsubEvent::Message { message, message_id, propagation_source } = gossipsubevent {
                                    let MyMessage { message } = deserialize(&message.data).unwrap();

                                    dbg!{message}; //You would probably want to do something more useful than just printing with your message

                                    println!("Received prefix: '{}', address: {:?}, from {:?}", prefix, address, message.source);
                                    println!("message id was {:?} and propagation_source was {:?}", message_id, propagation_source);
                                }
                                None
                            }
                        }
                    }
                }
            };
            if let Some((topic, sibling)) = to_publish {
                //dbg!(&topic);
                self.swarm.gossipsub.publish(topic, serialize(&sibling).unwrap()).unwrap();
            }
            if !listening {
                for addr in Swarm::listeners(&self.swarm) {
                    println!("Listening on {:?}", addr);
                    listening = true;
                }
            }
        }
    }
}

[mxinden]

I could make a pr adding this example. I think it would be helpful since there's enough differences between floodsub & gossipsub that converting between the two isn't always immediately obvious.

Generally I am very much in favor of having more examples, but would an additional GossipSub example add much value given that there is already gossipsub-chat.rs?

I would very much appreciate a pull request adding an example showcasing:

• Combining multiple NetworkBehaviours, e.g. GossipSub and Kademlia. With event_process=false if you like.
• Communication from user to swarm and swarm to user.
  • User -> Swarm: Can be done via a MPSC channel as you showed above.
  • Swarm -> User: Would be dependent on GossipSub and Kademlia. E.g. one could register a subscriber for a GossipSub topic, and pass a oneshot::Receiver for the response to the swarm when looking up a value via Kademlia.
• Ideally adding both a file in examples/ and extending src/tutorial.rs.

[thomaseizinger]

@Frederik-Baetens A couple of suggestions if you don't mind:

1. Having tokio::select! evaluate to something could be unnecessarily limiting depending on how complex your Swarm gets. You might want to consider doing all the handling inside the select branch.
2. Anything that should only happen once could be done before entering the loop or even before calling run_swarm. That would avoid the listening state altogether which is otherwise a bit hard to understand (raises the question "Is the variable ever set to "false" again?)
3. If you are not exposing the run_swarm function, you could also just completely get rid of the SwarmOwner component and simply use an async move block that you spawn into a runtime.

Something like:

#[derive(Serialize, Deserialize, Debug)]
pub struct MyMessage {
    pub message: String,
}

//Currently it's a bit weird in this code to pass the topic with every message in the channel,
//since my swarm is only subscribed on one hardcoded topic in this example.
//If you would add methods to subscribe to new channels,
//you could pass different topics with every channel message that you want to be published,
//allowing you to publish to different topics on the fly
#[derive(Clone)]
pub struct SwarmHandle {
    sender: Sender<(MyMessage, IdentTopic)>,
    topic: IdentTopic,
}

impl SwarmHandle {
    pub async fn new() -> Result<SwarmHandle, Box<dyn Error>> {
        let (sender, receiver) = mpsc::channel(100);
        
        //A vec is passed here for better peer disovery in the future, but for now it's just empty
        let (swarm, topic) = create_gossipsub_swarm(Vec::new()).await?;
        let swarm_handle = SwarmHandle {sender: sender.clone(), topic };

        tokio::spawn(async move {
            for addr in Swarm::listeners(&swarm) {
                println!("Listening on {:?}", addr);
                listening = true;
            }

            loop {
                tokio::select! {
                    sibling = receiver.recv() => {
                        let (sibling, topic) = sibling.unwrap();
                        println!("publishing sibling: {:?}", sibling);
                        
                       swarm.gossipsub.publish(topic, serialize(&sibling).unwrap()).unwrap();
                    }
                    event = swarm.next() => {
                        match event {
                            OutEvent::MdnsEvent(mdnsevent) => {
                                match mdnsevent {
                                    MdnsEvent::Discovered(list) => {
                                        for (peer, _) in list {
                                            swarm.gossipsub.add_explicit_peer(&peer);
                                        }
                                    },
                                    MdnsEvent::Expired(list) => {
                                        for (peer, _) in list {
                                            if !swarm.mdns.has_node(&peer) {
                                                swarm.gossipsub.remove_explicit_peer(&peer);
                                            }
                                        }
                                    }
                                }
                            },
                            OutEvent::GossipsubEvent(gossipsubevent) => {
                                if let GossipsubEvent::Message { message, message_id, propagation_source } = gossipsubevent {
                                    let MyMessage { message } = deserialize(&message.data).unwrap();

                                    dbg!{message}; //You would probably want to do something more useful than just printing with your message

                                    println!("Received prefix: '{}', address: {:?}, from {:?}", prefix, address, message.source);
                                    println!("message id was {:?} and propagation_source was {:?}", message_id, propagation_source);
                                }
                            }
                        }
                    }
                }
            }
        });

        Ok(swarm_handle)
    }
    
    pub async fn publish(&self, message: String ) {
        self.sender.send((MyMessage{message}, self.topic.clone())).await.unwrap();
    }
}