Communicating between a libp2p thread and a UI thread

[pieterdd]

I have a network thread running a libp2p swarm, and I have a UI thread. What would be a convenient design pattern to make the two threads communicate?

My idea was to establish two mpsc channels: one for network events and one for UI events. That sounds nice, but there is one problem. The network thread already blocks when awaiting the next swarm event. I'm not sure how to make it wait for both a swarm event or a message from the UI thread (whichever happens to trigger first in that iteration of the loop). I might be able to poll both every 50 ms without blocking, but you guys probably know a cleaner way from personal experience.

[thomaseizinger]

future::future::select is your friend :)

[pieterdd]

Aha. I've looked at the docs, but I'm not yet sure how I'd be able to move the mpsc::Receiver into a Future object I can run select! on in a loop without running into object ownership issues. Could you elaborate a bit or refer me to a real-world implementation? Rust's future crate is still new to me.

[thomaseizinger]

Right, a Receiver implements Stream and so does Swarm. Meaning, via the StreamExt extension trait, you can call .next() on it and do:

futures::future::select(receiver.next(), swarm.next()).await

This will give you an Either where Either::Left contains a tuple of: (Some(command), unfinished_swarm_future) and Either::Right contains a tuple of (Some(swarm_event), unfinished_receiver_future).

Both, Receiver and Swarm return cancellation-safe futures, meaning you can just drop the unfinished futures, i.e. do nothing with them.

Here is a good example from a workshop I did recently: https://github.com/thomaseizinger/libp2p-workshop/blob/iteration-4/src/event_loop.rs

[pieterdd]

Good that you brought this up along with the libp2p-workshop example. I was using std::sync::mpsc and wasn't aware of the futures::channel::mpsc implementation yet. The refactor will take a bit of work, but conceptually it seems fairly straightforward.

[thomaseizinger]

The std channels block the thread on reading which doesn't work well with the async ecosystem as that will block a worker thread of your runtime (e.g. tokio).

[dariusc93]

The general idea would be to use channels while using a oneshot for returning any values you are awaiting on while polling swarm. So to break it down you would do something like the following:

Have an enum being something like

enum Command {
  CommandOne(oneshot::Sender<Result<(), Error>>)
}

1. Create a channel (bound or unbound, suggest bound with 1 slot)
2. Pass a receiver to the thread or task handling polling the swarm while passing the sender to your UI
3. In your task, you would use tokio::select! or future::select! to poll both your receiver and the swarm (would suggest having separate functions where you push the events into to keep it clean but each their own on that)
4. When you wish to talk to libp2p, say wanting to connect to a peer, you would send something like Command::Connect (similar to above but it being Command::Connect(Multiaddr, oneshot::Sender<Result<(), Error>>)) while passing a oneshot sender to it. In your thread or task, you would dial either an address or peer upon receiving the command and push the oneshot off into a map with say your address or peerid of what youre dialing as your key (not the best way since you can dial a multiaddr without a peerid, usually suggest doing this in a behaviour if possible to make use of ConnectionId from DialOpts). As you poll swarm, you would listen on one of the swarm events dealing with connections. If it was successful, you would get SwarmEvent::ConnectionEstablished, or if failed you may get OutgoingConnectionError. You would check your map, remove the oneshot sender and send the result
5. In your UI function, you would await on the completion of your oneshot and return the results

To make it easier you could have a behaviour that emits your commands from your sender to swarm so you could process it while polling swarm. Eg this is an old code I tried (not compatible with current version of libp2p but not hard to update)

use core::task::{Context, Poll};
use futures::{channel::mpsc::Receiver, StreamExt};
use libp2p::swarm::{
    self, dummy::ConnectionHandler as DummyConnectionHandler, NetworkBehaviour, PollParameters,
};
use std::pin::Pin;

type NetworkBehaviourAction<T> = swarm::NetworkBehaviourAction<T, void::Void>;

pub struct Behaviour<T> {
    rx: Receiver<T>,
}

impl<T> Behaviour<T> {
    pub fn new(rx: Receiver<T>) -> Self {
        Behaviour { rx }
    }
}

impl<T> NetworkBehaviour for Behaviour<T>
where
    T: 'static + Send,
{
    type ConnectionHandler = DummyConnectionHandler;
    type OutEvent = T;

    fn new_handler(&mut self) -> Self::ConnectionHandler {
        DummyConnectionHandler
    }

    fn on_swarm_event(&mut self, _: swarm::FromSwarm<Self::ConnectionHandler>) {}

    fn on_connection_handler_event(
        &mut self,
        _peer_id: libp2p::PeerId,
        _connection_id: swarm::ConnectionId,
        _event: swarm::THandlerOutEvent<Self>,
    ) {
    }

    fn poll(
        &mut self,
        cx: &mut Context,
        _: &mut impl PollParameters,
    ) -> Poll<NetworkBehaviourAction<T>> {
        match Pin::new(&mut self.rx).as_mut().poll_next_unpin(cx) {
            Poll::Ready(Some(event)) => Poll::Ready(NetworkBehaviourAction::GenerateEvent(event)),
            _ => Poll::Pending,
        }
    }
}

Then in your main behaviour, you would simply Behaviour<Command> and pass your receiver into it. That way you dont have to worry about using a select macro, however, the downside would be that you would have to poll swarm, which may have a lot of events prior to yours, before you could process your commands, while with select, if youre using tokio select, you could use biased so it would poll from top to bottom instead of doing being pseudo-random.

To see examples, you could take a look at beetle, rust-ipfs (i maintain this project), ipfs-embed to see how they allow polling the swarm in a separate task while sending commands

[pieterdd]

Thank you for the thorough documentation. It's good to have an overview of the different patterns people apply when implementing libp2p. In my case however I don't have a custom Behaviour implemented. So for the repository I'm currently working on, I think the suggestion of a futures::future::select call with the receiver and swarm in the main event loop is the most straightforward solution.