From a9aaf40943ccf8cf557a2fbf8f192f5da2b52bcf Mon Sep 17 00:00:00 2001 From: Benjamin Saunders Date: Mon, 22 Feb 2021 17:09:20 -0800 Subject: [PATCH] Implement perf protocol for standardized benchmarking --- Cargo.toml | 4 +- perf/Cargo.toml | 28 ++++++ perf/src/client.rs | 225 +++++++++++++++++++++++++++++++++++++++++++++ perf/src/server.rs | 176 +++++++++++++++++++++++++++++++++++ 4 files changed, 431 insertions(+), 2 deletions(-) create mode 100644 perf/Cargo.toml create mode 100644 perf/src/client.rs create mode 100644 perf/src/server.rs diff --git a/Cargo.toml b/Cargo.toml index 8c61b44f3..157b5fd61 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [workspace] -members = ["quinn", "quinn-proto", "quinn-h3", "interop", "bench", "fuzz"] -default-members = ["quinn", "quinn-proto", "quinn-h3", "interop", "bench"] +members = ["quinn", "quinn-proto", "quinn-h3", "interop", "bench", "perf", "fuzz"] +default-members = ["quinn", "quinn-proto", "quinn-h3", "interop", "bench", "perf"] [profile.bench] debug = true diff --git a/perf/Cargo.toml b/perf/Cargo.toml new file mode 100644 index 000000000..03cf58dcb --- /dev/null +++ b/perf/Cargo.toml @@ -0,0 +1,28 @@ +[package] +name = "perf" +version = "0.1.0" +authors = ["Benjamin Saunders "] +edition = "2018" +license = "MIT OR Apache-2.0" +publish = false + +[dependencies] +anyhow = "1.0.22" +futures = "0.3.8" +quinn = { path = "../quinn" } +rcgen = "0.8" +rustls = { version = "0.19", features = ["dangerous_configuration"] } +webpki = "0.21" +structopt = "0.3" +tokio = { version = "1.0.1", features = ["rt", "macros", "signal", "net", "sync"] } +tracing = "0.1.10" +tracing-subscriber = { version = "0.2.5", default-features = false, features = ["env-filter", "fmt", "ansi", "chrono"]} +bytes = "1" + +[[bin]] +name = "server" +path = "src/server.rs" + +[[bin]] +name = "client" +path = "src/client.rs" diff --git a/perf/src/client.rs b/perf/src/client.rs new file mode 100644 index 000000000..6e8560e2f --- /dev/null +++ b/perf/src/client.rs @@ -0,0 +1,225 @@ +use std::{ + net::{IpAddr, Ipv6Addr, SocketAddr}, + sync::Arc, +}; + +use anyhow::{Context, Result}; +use bytes::Bytes; +use futures::StreamExt; +use structopt::StructOpt; +use tokio::sync::Semaphore; +use tracing::{error, info, trace}; + +/// Connects to a QUIC perf server and maintains a specified pattern of requests until interrupted +#[derive(StructOpt)] +#[structopt(name = "client")] +struct Opt { + /// Host to connect to + #[structopt(default_value = "localhost:4433")] + host: String, + /// Override DNS resolution for host + #[structopt(long)] + ip: Option, + /// Number of unidirectional requests to maintain concurrently + #[structopt(long, default_value = "0")] + uni_requests: u64, + /// Number of bidirectional requests to maintain concurrently + #[structopt(long, default_value = "1")] + bi_requests: u64, + /// Number of bytes to request + #[structopt(long, default_value = "1048576")] + download_size: u64, + /// Number of bytes to transmit, in addition to the request header + #[structopt(long, default_value = "1048576")] + upload_size: u64, + /// Whether to skip certificate validation + #[structopt(long)] + insecure: bool, +} + +#[tokio::main(flavor = "current_thread")] +async fn main() { + let opt = Opt::from_args(); + + tracing_subscriber::fmt::init(); + + if let Err(e) = run(opt).await { + error!("{:#}", e); + } +} + +async fn run(opt: Opt) -> Result<()> { + let mut host_parts = opt.host.split(':'); + let host_name = host_parts.next().unwrap(); + let host_port = host_parts + .next() + .map_or(Ok(443), |x| x.parse()) + .context("parsing port")?; + let addr = match opt.ip { + None => tokio::net::lookup_host(&opt.host) + .await + .context("resolving host")? + .next() + .unwrap(), + Some(ip) => SocketAddr::new(ip, host_port), + }; + + info!("connecting to {} at {}", host_name, addr); + + let endpoint = quinn::EndpointBuilder::default(); + + let (endpoint, _) = endpoint + .bind(&SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0)) + .context("binding endpoint")?; + + let mut cfg = quinn::ClientConfigBuilder::default(); + cfg.protocols(&[b"perf"]); + let mut cfg = cfg.build(); + if opt.insecure { + let tls_cfg: &mut rustls::ClientConfig = Arc::get_mut(&mut cfg.crypto).unwrap(); + tls_cfg + .dangerous() + .set_certificate_verifier(SkipServerVerification::new()); + } + + let quinn::NewConnection { + connection, + mut uni_streams, + .. + } = endpoint + .connect_with(cfg, &addr, &host_name)? + .await + .context("connecting")?; + + info!("established"); + + tokio::spawn(async move { + while let Some(Ok(stream)) = uni_streams.next().await { + tokio::spawn(async move { + if let Err(e) = drain_stream(stream).await { + error!("reading response failed: {:#}", e); + } + }); + } + }); + + tokio::select! { + x = drive_uni(connection.clone(), opt.uni_requests, opt.upload_size, opt.download_size) => x?, + x = drive_bi(connection.clone(), opt.bi_requests, opt.upload_size, opt.download_size) => x?, + _ = tokio::signal::ctrl_c() => { + info!("shutting down"); + connection.close(0u32.into(), b"interrupted"); + } + } + + endpoint.wait_idle().await; + + // TODO: Print stats + + Ok(()) +} + +async fn drain_stream(mut stream: quinn::RecvStream) -> Result<()> { + #[rustfmt::skip] + let mut bufs = [ + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + ]; + while stream.read_chunks(&mut bufs[..]).await?.is_some() {} + trace!("response finished on {}", stream.id()); + Ok(()) +} + +async fn drive_uni( + connection: quinn::Connection, + concurrency: u64, + upload: u64, + download: u64, +) -> Result<()> { + let sem = Arc::new(Semaphore::new(concurrency as usize)); + + loop { + let permit = sem.clone().acquire_owned().await.unwrap(); + let send = connection.open_uni().await?; + trace!("sending request on {}", send.id()); + tokio::spawn(async move { + if let Err(e) = request(send, upload, download).await { + error!("sending request failed: {:#}", e); + } + drop(permit); + }); + } +} + +async fn request(mut send: quinn::SendStream, mut upload: u64, download: u64) -> Result<()> { + send.write_all(&download.to_be_bytes()).await?; + let buf = [42; 4 * 1024]; + while upload > 0 { + let n = send + .write(&buf[..upload.min(buf.len() as u64) as usize]) + .await + .context("sending response")?; + upload -= n as u64; + } + send.finish().await?; + trace!("upload finished on {}", send.id()); + Ok(()) +} + +async fn drive_bi( + connection: quinn::Connection, + concurrency: u64, + upload: u64, + download: u64, +) -> Result<()> { + let sem = Arc::new(Semaphore::new(concurrency as usize)); + + loop { + let permit = sem.clone().acquire_owned().await.unwrap(); + let (send, recv) = connection.open_bi().await?; + trace!("sending request on {}", send.id()); + tokio::spawn(async move { + if let Err(e) = request_bi(send, recv, upload, download).await { + error!("request failed: {:#}", e); + } + drop(permit); + }); + } +} + +async fn request_bi( + send: quinn::SendStream, + recv: quinn::RecvStream, + upload: u64, + download: u64, +) -> Result<()> { + request(send, upload, download).await?; + drain_stream(recv).await?; + Ok(()) +} + +struct SkipServerVerification; + +impl SkipServerVerification { + fn new() -> Arc { + Arc::new(Self) + } +} + +impl rustls::ServerCertVerifier for SkipServerVerification { + fn verify_server_cert( + &self, + _roots: &rustls::RootCertStore, + _presented_certs: &[rustls::Certificate], + _dns_name: webpki::DNSNameRef, + _ocsp_response: &[u8], + ) -> Result { + Ok(rustls::ServerCertVerified::assertion()) + } +} diff --git a/perf/src/server.rs b/perf/src/server.rs new file mode 100644 index 000000000..3ff17d8e3 --- /dev/null +++ b/perf/src/server.rs @@ -0,0 +1,176 @@ +use std::{fs, net::SocketAddr, path::PathBuf}; + +use anyhow::{Context, Result}; +use bytes::Bytes; +use futures::StreamExt; +use structopt::StructOpt; +use tracing::{debug, error, info, trace}; + +#[derive(StructOpt)] +#[structopt(name = "server")] +struct Opt { + /// Address to listen on + #[structopt(long = "listen", default_value = "[::]:4433")] + listen: SocketAddr, + /// TLS private key in PEM format + #[structopt(parse(from_os_str), short = "k", long = "key", requires = "cert")] + key: Option, + /// TLS certificate in PEM format + #[structopt(parse(from_os_str), short = "c", long = "cert", requires = "key")] + cert: Option, +} + +#[tokio::main(flavor = "current_thread")] +async fn main() { + let opt = Opt::from_args(); + + tracing_subscriber::fmt::init(); + + if let Err(e) = run(opt).await { + error!("{}", e); + } +} + +async fn run(opt: Opt) -> Result<()> { + let (key, cert) = match (&opt.key, &opt.cert) { + (&Some(ref key), &Some(ref cert)) => { + let key = fs::read(key).context("reading key")?; + let cert = fs::read(cert).expect("reading cert"); + ( + quinn::PrivateKey::from_pem(&key).context("parsing key")?, + quinn::CertificateChain::from_pem(&cert).context("parsing cert")?, + ) + } + _ => { + let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + ( + quinn::PrivateKey::from_der(&cert.serialize_private_key_der()).unwrap(), + quinn::CertificateChain::from_certs(vec![quinn::Certificate::from_der( + &cert.serialize_der().unwrap(), + ) + .unwrap()]), + ) + } + }; + + let mut server_config = quinn::ServerConfigBuilder::default(); + server_config.certificate(cert, key).unwrap(); + server_config.protocols(&[b"perf"]); + + let server_config = server_config.build(); + + let mut endpoint = quinn::EndpointBuilder::default(); + endpoint.listen(server_config); + + let (endpoint, mut incoming) = endpoint.bind(&opt.listen).context("binding endpoint")?; + + info!("listening on {}", endpoint.local_addr().unwrap()); + + while let Some(handshake) = incoming.next().await { + tokio::spawn(async move { + if let Err(e) = handle(handshake).await { + error!("connection lost: {:#}", e); + } + }); + } + + Ok(()) +} + +async fn handle(handshake: quinn::Connecting) -> Result<()> { + let quinn::NewConnection { + uni_streams, + bi_streams, + connection, + .. + } = handshake.await.context("handshake failed")?; + debug!("{} connected", connection.remote_address()); + tokio::select! { + x = drive_uni(connection, uni_streams) => x?, + x = drive_bi(bi_streams) => x?, + } + Ok(()) +} + +async fn drive_uni( + connection: quinn::Connection, + mut streams: quinn::IncomingUniStreams, +) -> Result<()> { + while let Some(stream) = streams.next().await { + let stream = stream?; + let connection = connection.clone(); + tokio::spawn(async move { + if let Err(e) = handle_uni(connection, stream).await { + error!("request failed: {:#}", e); + } + }); + } + Ok(()) +} + +async fn handle_uni(connection: quinn::Connection, stream: quinn::RecvStream) -> Result<()> { + let bytes = read_req(stream).await?; + let response = connection.open_uni().await?; + respond(bytes, response).await?; + Ok(()) +} + +async fn drive_bi(mut streams: quinn::IncomingBiStreams) -> Result<()> { + while let Some(stream) = streams.next().await { + let (send, recv) = stream?; + tokio::spawn(async move { + if let Err(e) = handle_bi(send, recv).await { + error!("request failed: {:#}", e); + } + }); + } + Ok(()) +} + +async fn handle_bi(send: quinn::SendStream, recv: quinn::RecvStream) -> Result<()> { + let bytes = read_req(recv).await?; + respond(bytes, send).await?; + Ok(()) +} + +async fn read_req(mut stream: quinn::RecvStream) -> Result { + let mut buf = [0; 8]; + stream + .read_exact(&mut buf) + .await + .context("reading request")?; + let n = u64::from_be_bytes(buf); + trace!("got req for {} bytes on {}", n, stream.id()); + drain_stream(stream).await?; + Ok(n) +} + +async fn drain_stream(mut stream: quinn::RecvStream) -> Result<()> { + #[rustfmt::skip] + let mut bufs = [ + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + Bytes::new(), Bytes::new(), Bytes::new(), Bytes::new(), + ]; + while stream.read_chunks(&mut bufs[..]).await?.is_some() {} + trace!("finished reading {}", stream.id()); + Ok(()) +} + +async fn respond(mut bytes: u64, mut stream: quinn::SendStream) -> Result<()> { + let buf = [42; 4 * 1024]; + while bytes > 0 { + let n = stream + .write(&buf[..bytes.min(buf.len() as u64) as usize]) + .await + .context("sending response")?; + bytes -= n as u64; + } + trace!("finished responding on {}", stream.id()); + Ok(()) +}