feat(perf): support iperf-style intermittent results

protocols/perf/src/bin/perf.rs

@@ -29,8 +29,8 @@ use libp2p_core::{
     Transport as _,
 };
 use libp2p_identity::PeerId;
-use libp2p_perf::server::Event;
-use libp2p_perf::{Finished, Progressed, Run, RunParams, RunUpdate};
+use libp2p_perf::{client, server};
+use libp2p_perf::{Final, Intermediate, Run, RunParams, RunUpdate};
 use libp2p_swarm::{Config, NetworkBehaviour, Swarm, SwarmEvent};
 use log::{error, info};
 use serde::{Deserialize, Serialize};
@@ -135,7 +135,7 @@ async fn server(server_address: SocketAddr) -> Result<()> {
                     info!("Established connection to {:?} via {:?}", peer_id, endpoint);
                 }
                 SwarmEvent::ConnectionClosed { .. } => {}
-                SwarmEvent::Behaviour(Event { .. }) => {
+                SwarmEvent::Behaviour(server::Event { .. }) => {
                     info!("Finished run",)
                 }
                 e => panic!("{e:?}"),
@@ -237,7 +237,7 @@ async fn swarm<B: NetworkBehaviour + Default>() -> Result<Swarm<B>> {
 }
 
 async fn connect(
-    swarm: &mut Swarm<libp2p_perf::client::Behaviour>,
+    swarm: &mut Swarm<client::Behaviour>,
     server_address: Multiaddr,
 ) -> Result<PeerId> {
     let start = Instant::now();
@@ -260,21 +260,21 @@ async fn connect(
 }
 
 async fn perf(
-    swarm: &mut Swarm<libp2p_perf::client::Behaviour>,
+    swarm: &mut Swarm<client::Behaviour>,
     server_peer_id: PeerId,
     params: RunParams,
 ) -> Result<Run> {
     swarm.behaviour_mut().perf(server_peer_id, params)?;
 
     let duration = loop {
         match swarm.next().await.unwrap() {
-            SwarmEvent::Behaviour(libp2p_perf::client::Event {
+            SwarmEvent::Behaviour(client::Event {
                 id: _,
-                result: Ok(RunUpdate::Progressed(progressed)),
+                result: Ok(RunUpdate::Intermediate(progressed)),
             }) => {
                 info!("{progressed}");
 
-                let Progressed {
+                let Intermediate {
                     duration,
                     sent,
                     received,
@@ -291,9 +291,9 @@ async fn perf(
                     .unwrap()
                 );
             }
-            SwarmEvent::Behaviour(libp2p_perf::client::Event {
+            SwarmEvent::Behaviour(client::Event {
                 id: _,
-                result: Ok(RunUpdate::Finished(Finished { duration })),
+                result: Ok(RunUpdate::Final(Final { duration })),
             }) => break duration,
             e => panic!("{e:?}"),
         };

protocols/perf/src/client/handler.rs

@@ -37,7 +37,7 @@ use libp2p_swarm::{
 };
 use void::Void;
 
-use super::{RunError, RunId};
+use crate::client::{RunError, RunId};
 use crate::{RunParams, RunUpdate};
 
 #[derive(Debug)]
@@ -66,8 +66,6 @@ pub struct Handler {
     requested_streams: VecDeque<Command>,
 
     outbound: SelectAll<BoxStream<'static, (RunId, Result<crate::RunUpdate, std::io::Error>)>>,
-
-    keep_alive: KeepAlive,
 }
 
 impl Handler {
@@ -76,7 +74,6 @@ impl Handler {
             queued_events: Default::default(),
             requested_streams: Default::default(),
             outbound: Default::default(),
-            keep_alive: KeepAlive::Yes,
         }
     }
 }
@@ -157,7 +154,11 @@ impl ConnectionHandler for Handler {
     }
 
     fn connection_keep_alive(&self) -> KeepAlive {
-        self.keep_alive
+        if self.outbound.is_empty() {
+            KeepAlive::No
+        } else {
+            KeepAlive::Yes
+        }
     }
 
     fn poll(
@@ -182,12 +183,6 @@ impl ConnectionHandler for Handler {
             }));
         }
 
-        if self.outbound.is_empty() {
-            self.keep_alive = KeepAlive::No
-        } else {
-            self.keep_alive = KeepAlive::Yes
-        }
-
         Poll::Pending
     }
 }

protocols/perf/src/lib.rs

@@ -37,20 +37,20 @@ pub const PROTOCOL_NAME: StreamProtocol = StreamProtocol::new("/perf/1.0.0");
 
 #[derive(Debug, Clone, Copy)]
 pub enum RunUpdate {
-    Progressed(Progressed),
-    Finished(Finished),
+    Intermediate(Intermediate),
+    Final(Final),
 }
 
 #[derive(Debug, Clone, Copy)]
-pub struct Progressed {
+pub struct Intermediate {
     pub duration: Duration,
     pub sent: usize,
     pub received: usize,
 }
 
-impl Display for Progressed {
+impl Display for Intermediate {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let Progressed {
+        let Intermediate {
             duration,
             sent,
             received,
@@ -69,7 +69,7 @@ impl Display for Progressed {
 }
 
 #[derive(Debug, Clone, Copy)]
-pub struct Finished {
+pub struct Final {
     pub duration: RunDuration,
 }
 

protocols/perf/src/protocol.rs

@@ -27,7 +27,7 @@ use futures::{
     AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, FutureExt, SinkExt, Stream, StreamExt,
 };
 
-use crate::{Finished, Progressed, Run, RunDuration, RunParams, RunUpdate};
+use crate::{Final, Intermediate, Run, RunDuration, RunParams, RunUpdate};
 
 const BUF: [u8; 1024] = [0; 1024];
 const REPORT_INTERVAL: Duration = Duration::from_secs(1);
@@ -43,18 +43,18 @@ pub(crate) fn send_receive<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
     let inner = send_receive_inner(params, stream, sender).fuse();
 
     futures::stream::select(
-        receiver.map(|progressed| Ok(RunUpdate::Progressed(progressed))),
+        receiver.map(|progressed| Ok(RunUpdate::Intermediate(progressed))),
         inner
-            .map(|finished| finished.map(RunUpdate::Finished))
+            .map(|finished| finished.map(RunUpdate::Final))
             .into_stream(),
     )
 }
 
 async fn send_receive_inner<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
     params: RunParams,
     mut stream: S,
-    mut progress: futures::channel::mpsc::Sender<crate::Progressed>,
-) -> Result<Finished, std::io::Error> {
+    mut progress: futures::channel::mpsc::Sender<crate::Intermediate>,
+) -> Result<Final, std::io::Error> {
     let mut delay = Delay::new(REPORT_INTERVAL);
 
     let RunParams {
@@ -81,7 +81,7 @@ async fn send_receive_inner<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
                 Either::Left((_, _)) => {
                     delay.reset(REPORT_INTERVAL);
                     progress
-                        .send(Progressed {
+                        .send(Intermediate {
                             duration: intermittant_start.elapsed(),
                             sent: sent - intermittent_sent,
                             received: 0,
@@ -101,7 +101,7 @@ async fn send_receive_inner<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
             Either::Left((_, _)) => {
                 delay.reset(REPORT_INTERVAL);
                 progress
-                    .send(Progressed {
+                    .send(Intermediate {
                         duration: intermittant_start.elapsed(),
                         sent: sent - intermittent_sent,
                         received: 0,
@@ -127,7 +127,7 @@ async fn send_receive_inner<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
                 Either::Left((_, _)) => {
                     delay.reset(REPORT_INTERVAL);
                     progress
-                        .send(Progressed {
+                        .send(Intermediate {
                             duration: intermittant_start.elapsed(),
                             sent: sent - intermittent_sent,
                             received: received - intermittend_received,
@@ -145,7 +145,7 @@ async fn send_receive_inner<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
 
     let read_done = Instant::now();
 
-    Ok(Finished {
+    Ok(Final {
         duration: RunDuration {
             upload: write_done.duration_since(write_start),
             download: read_done.duration_since(write_done),

protocols/perf/src/server/handler.rs

@@ -41,14 +41,12 @@ pub struct Event {
 
 pub struct Handler {
     inbound: FuturesUnordered<BoxFuture<'static, Result<Run, std::io::Error>>>,
-    keep_alive: KeepAlive,
 }
 
 impl Handler {
     pub fn new() -> Self {
         Self {
             inbound: Default::default(),
-            keep_alive: KeepAlive::Yes,
         }
     }
 }
@@ -110,7 +108,11 @@ impl ConnectionHandler for Handler {
     }
 
     fn connection_keep_alive(&self) -> KeepAlive {
-        self.keep_alive
+        if self.inbound.is_empty() {
+            KeepAlive::No
+        } else {
+            KeepAlive::Yes
+        }
     }
 
     fn poll(
@@ -135,12 +137,6 @@ impl ConnectionHandler for Handler {
             }
         }
 
-        if self.inbound.is_empty() {
-            self.keep_alive = KeepAlive::No
-        } else {
-            self.keep_alive = KeepAlive::Yes
-        }
-
         Poll::Pending
     }
 }