feat(perf): implement libp2p perf protocol

.dockerignore

@@ -1 +0,0 @@
-target

Cargo.toml

@@ -26,6 +26,7 @@ full = [
     "metrics",
     "mplex",
     "noise",
+    "perf",
     "ping",
     "plaintext",
     "pnet",
@@ -64,6 +65,7 @@ mdns = ["dep:libp2p-mdns"]
 metrics = ["dep:libp2p-metrics"]
 mplex = ["dep:libp2p-mplex"]
 noise = ["dep:libp2p-noise"]
+perf = ["dep:libp2p-perf"]
 ping = ["dep:libp2p-ping", "libp2p-metrics?/ping"]
 plaintext = ["dep:libp2p-plaintext"]
 pnet = ["dep:libp2p-pnet"]
@@ -101,6 +103,7 @@ libp2p-kad = { version = "0.43.0", path = "protocols/kad", optional = true }
 libp2p-metrics = { version = "0.12.0", path = "misc/metrics", optional = true }
 libp2p-mplex = { version = "0.39.0", path = "muxers/mplex", optional = true }
 libp2p-noise = { version = "0.42.0", path = "transports/noise", optional = true }
+libp2p-perf = { version = "0.1.0", path = "protocols/perf", optional = true }
 libp2p-ping = { version = "0.42.0", path = "protocols/ping", optional = true }
 libp2p-plaintext = { version = "0.39.0", path = "transports/plaintext", optional = true }
 libp2p-pnet = { version = "0.22.2", path = "transports/pnet", optional = true }
@@ -159,6 +162,7 @@ members = [
     "protocols/identify",
     "protocols/kad",
     "protocols/mdns",
+    "protocols/perf",
     "protocols/ping",
     "protocols/relay",
     "protocols/request-response",

protocols/perf/Cargo.toml

@@ -0,0 +1,38 @@
+[package]
+name = "libp2p-perf"
+edition = "2021"
+rust-version = "1.64.0"
+description = "libp2p perf protocol implementation"
+version = "0.1.0"
+authors = ["Max Inden <mail@max-inden.de>"]
+license = "MIT"
+repository = "https://github.com/libp2p/rust-libp2p"
+keywords = ["peer-to-peer", "libp2p", "networking"]
+categories = ["network-programming", "asynchronous"]
+
+[dependencies]
+anyhow = "1"
+clap = { version = "4.1.6", features = ["derive"] }
+env_logger = "0.10.0"
+futures = "0.3.26"
+libp2p-core = { version = "0.39.0", path = "../../core" }
+libp2p-dns = { version = "0.39.0", path = "../../transports/dns", features = ["async-std"] }
+libp2p-noise = { version = "0.42.0", path = "../../transports/noise" }
+libp2p-quic = { version = "0.7.0-alpha.2", path = "../../transports/quic", features = ["async-std"] }
+libp2p-swarm = { version = "0.42.0", path = "../../swarm", features = ["macros", "async-std"] }
+libp2p-tcp = { version = "0.39.0", path = "../../transports/tcp", features = ["async-io"] }
+libp2p-yamux = { version = "0.43.0", path = "../../muxers/yamux" }
+log = "0.4"
+thiserror = "1.0"
+void = "1"
+
+[dev-dependencies]
+rand = "0.8"
+libp2p-plaintext = { path = "../../transports/plaintext" }
+
+# Passing arguments to the docsrs builder in order to properly document cfg's.
+# More information: https://docs.rs/about/builds#cross-compiling
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]
+rustc-args = ["--cfg", "docsrs"]

protocols/perf/Dockerfile

@@ -0,0 +1,4 @@
+FROM ubuntu:kinetic
+ADD ./target/release/perf-client /usr/local/bin/perf-client
+ADD ./target/release/perf-server /usr/local/bin/perf-server
+ENTRYPOINT [ "perf-server"]

protocols/perf/src/bin/perf-client.rs

@@ -0,0 +1,139 @@
+// Copyright 2023 Protocol Labs.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use anyhow::{bail, Result};
+use clap::Parser;
+use futures::{executor::block_on, future::Either, StreamExt};
+use libp2p_core::{
+    identity, muxing::StreamMuxerBox, transport::OrTransport, upgrade, Multiaddr, PeerId, Transport,
+};
+use libp2p_dns::DnsConfig;
+use libp2p_perf::client::RunParams;
+use libp2p_swarm::{SwarmBuilder, SwarmEvent};
+use log::info;
+
+#[derive(Debug, Parser)]
+#[clap(name = "libp2p perf client")]
+struct Opts {
+    #[arg(long)]
+    server_address: Multiaddr,
+}
+
+fn main() -> Result<()> {
+    env_logger::init();
+
+    let opts = Opts::parse();
+
+    // Create a random PeerId
+    let local_key = identity::Keypair::generate_ed25519();
+    let local_peer_id = PeerId::from(local_key.public());
+
+    let transport = {
+        let tcp =
+            libp2p_tcp::async_io::Transport::new(libp2p_tcp::Config::default().port_reuse(true))
+                .upgrade(upgrade::Version::V1Lazy)
+                .authenticate(
+                    libp2p_noise::NoiseAuthenticated::xx(&local_key)
+                        .expect("Signing libp2p-noise static DH keypair failed."),
+                )
+                .multiplex(libp2p_yamux::YamuxConfig::default());
+
+        let quic = {
+            let mut config = libp2p_quic::Config::new(&local_key);
+            config.support_draft_29 = true;
+            libp2p_quic::async_std::Transport::new(config)
+        };
+
+        let dns = block_on(DnsConfig::system(OrTransport::new(quic, tcp))).unwrap();
+
+        dns.map(|either_output, _| match either_output {
+            Either::Left((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+            Either::Right((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+        })
+        .boxed()
+    };
+
+    let mut swarm = SwarmBuilder::with_async_std_executor(
+        transport,
+        libp2p_perf::client::behaviour::Behaviour::default(),
+        local_peer_id,
+    )
+    .substream_upgrade_protocol_override(upgrade::Version::V1Lazy)
+    .build();
+
+    swarm.dial(opts.server_address).unwrap();
+    let server_peer_id = block_on(async {
+        loop {
+            match swarm.next().await.unwrap() {
+                SwarmEvent::ConnectionEstablished { peer_id, .. } => return Ok(peer_id),
+                SwarmEvent::OutgoingConnectionError { peer_id, error } => {
+                    bail!("Outgoing connection error to {:?}: {:?}", peer_id, error);
+                }
+                e => panic!("{e:?}"),
+            }
+        }
+    })?;
+
+    swarm.behaviour_mut().perf(
+        server_peer_id,
+        RunParams {
+            to_send: 10 * 1024 * 1024,
+            to_receive: 10 * 1024 * 1024,
+        },
+    )?;
+
+    let result = block_on(async {
+        loop {
+            match swarm.next().await.unwrap() {
+                SwarmEvent::ConnectionEstablished {
+                    peer_id, endpoint, ..
+                } => {
+                    info!("Established connection to {:?} via {:?}", peer_id, endpoint);
+                }
+                SwarmEvent::OutgoingConnectionError { peer_id, error } => {
+                    info!("Outgoing connection error to {:?}: {:?}", peer_id, error);
+                }
+                SwarmEvent::Behaviour(libp2p_perf::client::behaviour::Event { id: _, result }) => {
+                    break result
+                }
+                e => panic!("{e:?}"),
+            }
+        }
+    });
+
+    let stats = result?;
+
+    let sent_mebibytes = stats.params.to_send as f64 / 1024.0 / 1024.0;
+    let sent_time = (stats.timers.write_done - stats.timers.write_start).as_secs_f64();
+    let sent_bandwidth_mebibit_second = (sent_mebibytes * 8.0) / sent_time;
+
+    let received_mebibytes = stats.params.to_receive as f64 / 1024.0 / 1024.0;
+    let receive_time = (stats.timers.read_done - stats.timers.write_done).as_secs_f64();
+    let receive_bandwidth_mebibit_second = (received_mebibytes * 8.0) / receive_time;
+
+    println!(
+        "Finished run: Sent {sent_mebibytes} MiB in {sent_time} s with \
+         {sent_bandwidth_mebibit_second} MiBit/s and received \
+         {received_mebibytes} MiB in {receive_time} s with \
+         {receive_bandwidth_mebibit_second} MiBit/s",
+    );
+
+    Ok(())
+}

protocols/perf/src/bin/perf-server.rs

@@ -0,0 +1,131 @@
+// Copyright 2023 Protocol Labs.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use clap::Parser;
+use futures::{executor::block_on, future::Either, StreamExt};
+use libp2p_core::{
+    identity, muxing::StreamMuxerBox, transport::OrTransport, upgrade, PeerId, Transport,
+};
+use libp2p_dns::DnsConfig;
+use libp2p_swarm::{SwarmBuilder, SwarmEvent};
+use log::{error, info};
+
+#[derive(Debug, Parser)]
+#[clap(name = "libp2p perf server")]
+struct Opts {}
+
+fn main() {
+    env_logger::init();
+
+    let _opts = Opts::parse();
+
+    // 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:?}");
+
+    let transport = {
+        let tcp =
+            libp2p_tcp::async_io::Transport::new(libp2p_tcp::Config::default().port_reuse(true))
+                .upgrade(upgrade::Version::V1Lazy)
+                .authenticate(
+                    libp2p_noise::NoiseAuthenticated::xx(&local_key)
+                        .expect("Signing libp2p-noise static DH keypair failed."),
+                )
+                .multiplex(libp2p_yamux::YamuxConfig::default());
+
+        let quic = {
+            let mut config = libp2p_quic::Config::new(&local_key);
+            config.support_draft_29 = true;
+            libp2p_quic::async_std::Transport::new(config)
+        };
+
+        let dns = block_on(DnsConfig::system(OrTransport::new(quic, tcp))).unwrap();
+
+        dns.map(|either_output, _| match either_output {
+            Either::Left((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+            Either::Right((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+        })
+        .boxed()
+    };
+
+    let mut swarm = SwarmBuilder::with_async_std_executor(
+        transport,
+        libp2p_perf::server::behaviour::Behaviour::default(),
+        local_peer_id,
+    )
+    .substream_upgrade_protocol_override(upgrade::Version::V1Lazy)
+    .build();
+
+    swarm
+        .listen_on("/ip4/0.0.0.0/tcp/4001".parse().unwrap())
+        .unwrap();
+
+    swarm
+        .listen_on("/ip4/0.0.0.0/udp/4001/quic-v1".parse().unwrap())
+        .unwrap();
+
+    block_on(async {
+        loop {
+            match swarm.next().await.unwrap() {
+                SwarmEvent::NewListenAddr { address, .. } => {
+                    info!("Listening on {:?}", address);
+                }
+                SwarmEvent::IncomingConnection { .. } => {}
+                e @ SwarmEvent::IncomingConnectionError { .. } => {
+                    error!("{e:?}");
+                }
+                SwarmEvent::ConnectionEstablished {
+                    peer_id, endpoint, ..
+                } => {
+                    info!("Established connection to {:?} via {:?}", peer_id, endpoint);
+                }
+                SwarmEvent::ConnectionClosed { .. } => {}
+                SwarmEvent::Behaviour(libp2p_perf::server::behaviour::Event::Finished {
+                    remote_peer_id,
+                    stats,
+                }) => {
+                    let received_mebibytes = stats.params.received as f64 / 1024.0 / 1024.0;
+                    let receive_time =
+                        (stats.timers.read_done - stats.timers.read_start).as_secs_f64();
+                    let receive_bandwidth_mebibit_second =
+                        (received_mebibytes * 8.0) / receive_time;
+
+                    let sent_mebibytes = stats.params.sent as f64 / 1024.0 / 1024.0;
+                    let sent_time =
+                        (stats.timers.write_done - stats.timers.read_done).as_secs_f64();
+                    let sent_bandwidth_mebibit_second = (sent_mebibytes * 8.0) / sent_time;
+
+                    info!(
+                        "Finished run with {}: Received {} MiB in {} s with {} MiBit/s and sent {} MiB in {} s with {} MiBit/s",
+                        remote_peer_id,
+                        received_mebibytes,
+                        receive_time,
+                        receive_bandwidth_mebibit_second,
+                        sent_mebibytes,
+                        sent_time,
+                        sent_bandwidth_mebibit_second,
+                    )
+                }
+                e => panic!("{e:?}"),
+            }
+        }
+    })
+}