mirror of
https://github.com/n0-computer/noq.git
synced 2026-09-21 02:33:23 +00:00
Implement perf protocol for standardized benchmarking
This commit is contained in:
Benjamin Saunders
committed by
Dirkjan Ochtman
Dirkjan Ochtman
parent
14db88562d
commit
a9aaf40943
+2
-2
@@ -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
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
[package]
|
||||
name = "perf"
|
||||
version = "0.1.0"
|
||||
authors = ["Benjamin Saunders <ben.e.saunders@gmail.com>"]
|
||||
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"
|
||||
@@ -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<IpAddr>,
|
||||
/// 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<Self> {
|
||||
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<rustls::ServerCertVerified, rustls::TLSError> {
|
||||
Ok(rustls::ServerCertVerified::assertion())
|
||||
}
|
||||
}
|
||||
@@ -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<PathBuf>,
|
||||
/// TLS certificate in PEM format
|
||||
#[structopt(parse(from_os_str), short = "c", long = "cert", requires = "key")]
|
||||
cert: Option<PathBuf>,
|
||||
}
|
||||
|
||||
#[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<u64> {
|
||||
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(())
|
||||
}
|
||||
Reference in New Issue
Block a user