Something went wrong. Try again.
noq, a QUIC implementation in Rust
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210use std::{net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
use anyhow::{Context, Result};use bytes::Bytes;use clap::Parser;use noq::{TokioRuntime, crypto::rustls::QuicServerConfig};use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, pem::PemObject};use tracing::{debug, error, info};
use crate::{CommonOpt, PERF_CIPHER_SUITES, noprotection::NoProtectionServerConfig};
#[derive(Parser)]#[clap(name = "server")]pub struct Opt { /// Address to listen on #[clap(long = "listen", default_value = "[::]:4433")] listen: SocketAddr, /// TLS private key in DER format #[clap(short = 'k', long = "key", requires = "cert")] key: Option<PathBuf>, /// TLS certificate in PEM format #[clap(short = 'c', long = "cert", requires = "key")] cert: Option<PathBuf>, /// Common options #[command(flatten)] common: CommonOpt,}
pub async fn run(opt: Opt) -> Result<()> { let (key, cert) = match (&opt.key, &opt.cert) { (Some(key), Some(cert)) => ( PrivateKeyDer::from_pem_file(key).context("reading private key")?, CertificateDer::pem_file_iter(cert) .context("reading certificate chain file")? .collect::<Result<_, _>>() .context("reading certificate chain")?, ), _ => { let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); ( PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()).into(), vec![CertificateDer::from(cert.cert)], ) } };
let default_provider = rustls::crypto::ring::default_provider(); let provider = rustls::crypto::CryptoProvider { cipher_suites: PERF_CIPHER_SUITES.into(), ..default_provider };
let mut crypto = rustls::ServerConfig::builder_with_provider(provider.into()) .with_protocol_versions(&[&rustls::version::TLS13]) .unwrap() .with_no_client_auth() .with_single_cert(cert, key) .unwrap(); crypto.alpn_protocols = vec![b"perf".to_vec()];
if opt.common.keylog { crypto.key_log = Arc::new(rustls::KeyLogFile::new()); }
let transport = opt.common.build_transport_config( #[cfg(feature = "qlog")] "perf-server", )?;
let crypto = QuicServerConfig::try_from(crypto)?; let mut config = noq::ServerConfig::with_crypto(match opt.common.no_protection { true => Arc::new(NoProtectionServerConfig::new(crypto)), false => Arc::new(crypto), }); config.transport_config(Arc::new(transport));
let socket = opt.common.bind_socket(opt.listen)?;
let mut endpoint_cfg = noq::EndpointConfig::default(); endpoint_cfg.max_udp_payload_size(opt.common.max_udp_payload_size)?;
let endpoint = noq::Endpoint::new(endpoint_cfg, Some(config), socket, Arc::new(TokioRuntime)) .context("creating endpoint")?;
info!("listening on {}", endpoint.local_addr().unwrap());
let opt = Arc::new(opt);
while let Some(handshake) = endpoint.accept().await { let opt = opt.clone(); tokio::spawn(async move { if let Err(e) = handle(handshake, opt).await { error!("connection lost: {:#}", e); } }); }
Ok(())}
async fn handle(handshake: noq::Incoming, opt: Arc<Opt>) -> Result<()> { let connection = handshake.await.context("handshake failed")?;
debug!( "{} connected", connection .path(noq::PathId::ZERO) .expect("path open after connect") .remote_address() .expect("path is alive") ); tokio::try_join!( drive_uni(connection.clone()), drive_bi(connection.clone()), conn_stats(connection, opt) )?; Ok(())}
async fn conn_stats(connection: noq::Connection, opt: Arc<Opt>) -> Result<()> { if opt.common.conn_stats { loop { tokio::time::sleep(Duration::from_secs(2)).await; println!("{:?}\n", connection.stats()); } }
Ok(())}
async fn drive_uni(connection: noq::Connection) -> Result<()> { while let Ok(stream) = connection.accept_uni().await { 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: noq::Connection, stream: noq::RecvStream) -> Result<()> { let bytes = read_req(stream).await?; let response = connection.open_uni().await?; respond(bytes, response).await?; Ok(())}
async fn drive_bi(connection: noq::Connection) -> Result<()> { while let Ok((send, recv)) = connection.accept_bi().await { tokio::spawn(async move { if let Err(e) = handle_bi(send, recv).await { error!("request failed: {:#}", e); } }); } Ok(())}
async fn handle_bi(send: noq::SendStream, recv: noq::RecvStream) -> Result<()> { let bytes = read_req(recv).await?; respond(bytes, send).await?; Ok(())}
async fn read_req(mut stream: noq::RecvStream) -> Result<u64> { let mut buf = [0; 8]; stream .read_exact(&mut buf) .await .context("reading request")?; let n = u64::from_be_bytes(buf); debug!("got req for {} bytes on {}", n, stream.id()); drain_stream(stream).await?; Ok(n)}
async fn drain_stream(mut stream: noq::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_many_chunks(&mut bufs[..]).await?.is_some() {} debug!("finished reading {}", stream.id()); Ok(())}
async fn respond(mut bytes: u64, mut stream: noq::SendStream) -> Result<()> { static DATA: [u8; 1024 * 1024] = [42; 1024 * 1024];
while bytes > 0 { let chunk_len = bytes.min(DATA.len() as u64); stream .write_chunk(Bytes::from_static(&DATA[..chunk_len as usize])) .await .context("sending response")?; bytes -= chunk_len; } debug!("finished responding on {}", stream.id()); Ok(())}