feat(perf): implement libp2p perf protocol

Cargo.toml

@@ -28,6 +28,7 @@ members = [
     "protocols/identify",
     "protocols/kad",
     "protocols/mdns",
+    "protocols/perf",
     "protocols/ping",
     "protocols/relay",
     "protocols/request-response",

libp2p/Cargo.toml

@@ -28,6 +28,7 @@ full = [
     "metrics",
     "mplex",
     "noise",
+    "perf",
     "ping",
     "plaintext",
     "pnet",
@@ -66,6 +67,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"]
@@ -120,6 +122,7 @@ pin-project = "1.0.0"
 libp2p-deflate = { version = "0.39.0", path = "../transports/deflate", optional = true }
 libp2p-dns = { version = "0.39.0", path = "../transports/dns", optional = true }
 libp2p-mdns = { version = "0.43.0", path = "../protocols/mdns", optional = true }
+libp2p-perf = { version = "0.1.0", path = "../protocols/perf", optional = true }
 libp2p-quic = { version = "0.7.0-alpha.3", path = "../transports/quic", optional = true }
 libp2p-tcp = { version = "0.39.0", path = "../transports/tcp", optional = true }
 libp2p-tls = { version = "0.1.0", path = "../transports/tls", optional = true }

protocols/perf/Cargo.toml

@@ -0,0 +1,42 @@
+[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"
+async-std = { version = "1.9.0", features = ["attributes"] }
+clap = { version = "4.1.6", features = ["derive"] }
+env_logger = "0.10.0"
+futures = "0.3.26"
+instant = "0.1.11"
+libp2p-core = { version = "0.39.0", path = "../../core" }
+libp2p-dns = { version = "0.39.0", path = "../../transports/dns", features = ["async-std"] }
+libp2p-identity = { version = "0.1.0", path = "../../identity" }
+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-swarm-test = { path = "../../swarm-test"}
+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,22 @@
+# syntax=docker/dockerfile:1.5-labs
+FROM rust:1.67.0 as builder
+
+# Run with access to the target cache to speed up builds
+WORKDIR /workspace
+ADD . .
+RUN --mount=type=cache,target=./target \
+    --mount=type=cache,target=/usr/local/cargo/registry \
+    cargo build --release --package libp2p-perf
+
+RUN --mount=type=cache,target=./target \
+    mv ./target/release/perf-server /usr/local/bin/perf-server
+
+RUN --mount=type=cache,target=./target \
+    mv ./target/release/perf-client /usr/local/bin/perf-client
+
+FROM debian:bullseye-slim
+
+COPY --from=builder /usr/local/bin/perf-server /usr/local/bin/perf-server
+COPY --from=builder /usr/local/bin/perf-client /usr/local/bin/perf-client
+
+ENTRYPOINT [ "perf-server"]

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

@@ -0,0 +1,133 @@
+// 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::{future::Either, StreamExt};
+use libp2p_core::{muxing::StreamMuxerBox, transport::OrTransport, upgrade, Multiaddr, Transport};
+use libp2p_dns::DnsConfig;
+use libp2p_identity::PeerId;
+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,
+}
+
+#[async_std::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+
+    let opts = Opts::parse();
+
+    // Create a random PeerId
+    let local_key = libp2p_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 = DnsConfig::system(OrTransport::new(quic, tcp))
+            .await
+            .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::default(),
+        local_peer_id,
+    )
+    .substream_upgrade_protocol_override(upgrade::Version::V1Lazy)
+    .build();
+
+    swarm.dial(opts.server_address).unwrap();
+    let server_peer_id = loop {
+        match swarm.next().await.unwrap() {
+            SwarmEvent::ConnectionEstablished { peer_id, .. } => break 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 stats = 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::Event { id: _, result }) => break result?,
+            e => panic!("{e:?}"),
+        }
+    };
+
+    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(())
+}