diff --git a/perf/src/bin/perf_client.rs b/perf/src/bin/perf_client.rs index 97b677b06..0b2f6af34 100644 --- a/perf/src/bin/perf_client.rs +++ b/perf/src/bin/perf_client.rs @@ -51,6 +51,9 @@ struct Opt { /// Specify the local socket address #[structopt(long)] local_addr: Option, + /// Whether to print connection statistics + #[structopt(long)] + conn_stats: bool, } #[tokio::main(flavor = "current_thread")] @@ -162,6 +165,9 @@ async fn run(opt: Opt) -> Result<()> { { let guard = stats.lock().unwrap(); guard.print(); + if opt.conn_stats { + println!("{:?}\n", connection.stats()); + } } } }; diff --git a/perf/src/bin/perf_server.rs b/perf/src/bin/perf_server.rs index 330e72891..0f7578b34 100644 --- a/perf/src/bin/perf_server.rs +++ b/perf/src/bin/perf_server.rs @@ -1,4 +1,4 @@ -use std::{fs, net::SocketAddr, path::PathBuf, sync::Arc}; +use std::{fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration}; use anyhow::{Context, Result}; use bytes::Bytes; @@ -26,6 +26,9 @@ struct Opt { /// Receive buffer size in bytes #[structopt(long, default_value = "2097152")] recv_buffer_size: usize, + /// Whether to print connection statistics + #[structopt(long)] + conn_stats: bool, } #[tokio::main(flavor = "current_thread")] @@ -89,9 +92,12 @@ async fn run(opt: Opt) -> Result<()> { info!("listening on {}", endpoint.local_addr().unwrap()); + let opt = Arc::new(opt); + while let Some(handshake) = incoming.next().await { + let opt = opt.clone(); tokio::spawn(async move { - if let Err(e) = handle(handshake).await { + if let Err(e) = handle(handshake, opt).await { error!("connection lost: {:#}", e); } }); @@ -100,7 +106,7 @@ async fn run(opt: Opt) -> Result<()> { Ok(()) } -async fn handle(handshake: quinn::Connecting) -> Result<()> { +async fn handle(handshake: quinn::Connecting, opt: Arc) -> Result<()> { let quinn::NewConnection { uni_streams, bi_streams, @@ -108,7 +114,22 @@ async fn handle(handshake: quinn::Connecting) -> Result<()> { .. } = handshake.await.context("handshake failed")?; debug!("{} connected", connection.remote_address()); - tokio::try_join!(drive_uni(connection, uni_streams), drive_bi(bi_streams))?; + tokio::try_join!( + drive_uni(connection.clone(), uni_streams), + drive_bi(bi_streams), + conn_stats(connection, opt) + )?; + Ok(()) +} + +async fn conn_stats(connection: quinn::Connection, opt: Arc) -> Result<()> { + if opt.conn_stats { + loop { + tokio::time::sleep(Duration::from_secs(2)).await; + println!("{:?}\n", connection.stats()); + } + } + Ok(()) }