Simple example with stream between two peers

[alexander-camuto]

Hello all.

I'm a newcomer to rust-libp2p, having been using js-libp2p and go-libp2p for a while. At first glance much of the interfaces are structured quite differently and I'm struggling to match some of the functionality I could achieve in go and js. Let's say for instance that I have a remote peer in go that is serving data using a stream handler as such:

h.SetStreamHandler("/bencher", func(s inet.Stream) {
		defer s.Close()
		r := rand.New(rand.NewSource(time.Now().UnixNano()))
		start := time.Now()

		n, err := io.CopyN(s, r, 100<<20)
		if err != nil {
			fmt.Println("copy error: ", err)
		}
		took := time.Since(start)
		fmt.Printf("transfer took %s (%d bps)\n", took, int(float64(n)/took.Seconds()))
	})

Now fetching that data in js-libp2p or go-libp2p is rather trivial. In js I can simply call dialProtocol on an instantiated libp2p node:

const {stream} = await libp2p.dialProtocol(pid, '/bencher');
for await (const _ of stream.source) { }

Or in go I can call NewStream on an instantiated libp2p host:

s, err := h.NewStream(context.TODO(), pid, "/bencher")
n, err := io.Copy(ioutil.Discard, s)

I've been trying to achieve similar results using Swarm in rust but end up digging through the implementation of libp2p::Transport or multistream_select::dialer_select_proto to attempt to replicate this behaviour. I was wondering if this is the right approach or if there might be a simpler way of doing this that is currently going over my head.

[thomaseizinger]

rust-libp2p's design is centered around the idea of modelling each protocol as a pluggable component.

There is a hierachy of NetworkBehaviour > ProtocolsHandler > InboundUpgrade/OutboundUpgrade.

The upgrade traits define how to go from bytes to higher-level messages on a protocol stream.

The ProtocolsHandler is responsible for connection-level state management with one peer (per protocol).

The NetworkBehaviour is responsible for managing state across multiple peers AND connections per protocol.

To achieve what you want is unfortunately not as light-weight as in Go or JS because there is currently no way of not using these abstractions.

However, whilst it might not be as appealing for prototyping, these abstractions make it possible to very easily write fast and isolated tests for NetworkBehaviours, something that is quite nice once the networking logic in your app grows.

[alexander-camuto]

Ah gotcha -- thanks for the prompt reply.

What would the stream example above look like at the NetworkBehaviour level ?

I've been going down a separate route which doesn't use these abstractions and it feels a bit clunky. My approach so far has been to use the libp2p::Transport interface, directly dialing the remote peer using the dial method; and subsequent to that creating an outbound substream using libp2p::core::muxing::outbound_from_ref_and_wrap called on the returned StreamMuxer.

[thomaseizinger]

What would the stream example above look like at the NetworkBehaviour level ?

You would need to create an implementation for all of these abstractions and compose them together. For details, I suggest looking how other protocols have been implemented. Perhaps a good start is perhaps libp2p-request-response.

Note that personally, I would really like rust-libp2p to be more prototype friendly here. We currently don't support the "creating an arbitrary stream between two peers" usecase particularly well :(

See also: #2078 and #2012.

[alexander-camuto]

@thomaseizinger I somehow managed to get this to work + compile to wasm without the NetworkBehaviour and ProtocolsHandler abstractions. I only use libp2p::Transport and some functions from the multistream-select crate. Not the prettiest way of doing things but if anyone else is interested -- here's how I did it (I'm new to rust so forgive me for bad syntax etc...):

#[cfg(feature = "browser")]
mod browser;

use libp2p::core::muxing::outbound_from_ref_and_wrap;
use libp2p::core::{self, transport::OptionalTransport};

use async_std;
use libp2p::wasm_ext;
use libp2p::{identity, Multiaddr, PeerId};
use libp2p::{mplex, noise, Transport};
use multistream_select::dialer_select_proto;
use std::sync::Arc;

#[cfg(feature = "native")]
use libp2p::websocket;

#[cfg(feature = "native")]
use libp2p_dns as dns;

#[cfg(feature = "native")]
use libp2p_tcp as tcp;

pub async fn service(wasm_external_transport: Option<wasm_ext::ExtTransport>, dial: String) {
    // Create a random PeerId
    let local_key = identity::Keypair::generate_ed25519();
    let local_peer_id = PeerId::from(local_key.public());
    log::info!("Local peer id: {:?}", local_peer_id);

    let dh_keys = noise::Keypair::<noise::X25519Spec>::new()
        .into_authentic(&local_key)
        .unwrap();
    let noise = noise::NoiseConfig::xx(dh_keys).into_authenticated();

    log::info!("Attempting to dial: {:?}", dial);

    let transport = if let Some(t) = wasm_external_transport {
        OptionalTransport::some(t)
    } else {
        OptionalTransport::none()
    };

    // if not compiling for wasm
    #[cfg(feature = "native")]
    let transport = transport.or_transport({
        let desktop_trans = tcp::TcpConfig::new();
        let desktop_trans =
            websocket::WsConfig::new(desktop_trans.clone()).or_transport(desktop_trans);
        OptionalTransport::some(
            if let Ok(dns) = dns::DnsConfig::system(desktop_trans.clone()).await {
                dns.boxed()
            } else {
                desktop_trans.map_err(dns::DnsErr::Transport).boxed()
            },
        )
    });

    // gets an outbound stream from Muxer when dialing
    let transport = transport
        .upgrade(core::upgrade::Version::V1)
        .authenticate(noise)
        .multiplex(mplex::MplexConfig::new())
        .map(|(peer, muxer), _| (peer, outbound_from_ref_and_wrap(Arc::new(muxer))));

    let dial = dial.parse::<Multiaddr>().ok().unwrap();
    let protos = vec![b"/bencher"];

    // Reach out to another node
    let connect = transport.dial(dial.clone()).ok().unwrap().await.unwrap();
    // select for a custom protocol ID
    let (proto, mut io) = dialer_select_proto(
        connect.1.await.unwrap(),
        protos.into_iter(),
        multistream_select::Version::V1,
    )
    .await
    .unwrap();
    log::info!("Dialed proto {:?}!", proto);
    // empty out stream into the void !
    async_std::io::copy(&mut io, &mut async_std::io::sink())
        .await
        .ok();
}

[thomaseizinger]

Thanks for posting this!

I haven't looked into it in detail but at a first glance, this does not seem to actually make use of multiplexing. In other words, you are opening a new connection per protocol instead of multiplexing multiple protocols over the same physical connection. Might not be an issue in this example but likely limiting if you want to do more than just a PoC.

[alexander-camuto]

Thanks for pointing that out. I'm just using this to benchmark the speed of fetching data through a raw I/O stream (between two local peers) for different libp2p implementations to get an upper bound on data transfer performance. Surprisingly I'm only getting 40 mbps for the above examples (20 mbps when compiled to wasm) -- vs up to 700mbps using go-libp2p and 80mbps for js-libp2p.

Feel like I'm missing something here.

[thomaseizinger]

Did you compile in release mode? :)

[alexander-camuto]

that was it ! 🤦‍♂️ thank you -- I get closer to 700mbps now and 481mbps when compiling to wasm.

[thomaseizinger]

Happens to the best of us 😁

[0x77dev]

Just to update the thread, the #5027 has been merged into the master branch. You can find an easier method for establishing basic stream between peers in the stream protocol example.