diff --git a/quinn-proto/src/connection.rs b/quinn-proto/src/connection.rs index 3125f5996..d865c6edd 100644 --- a/quinn-proto/src/connection.rs +++ b/quinn-proto/src/connection.rs @@ -2567,19 +2567,7 @@ impl Connection { .get_send_mut(id) .ok_or(FinishError::UnknownStream)?; ss.finish()?; - let space = &mut self.spaces[SpaceId::Data as usize]; - for frame in &mut space.pending.stream { - if frame.id == id && frame.offset + frame.data.len() as u64 == ss.offset { - frame.fin = true; - return Ok(()); - } - } - space.pending.stream.push_back(frame::Stream { - id, - data: Bytes::new(), - offset: ss.offset, - fin: true, - }); + self.spaces[SpaceId::Data as usize].finish_stream(id, ss.offset); Ok(()) } diff --git a/quinn-proto/src/spaces.rs b/quinn-proto/src/spaces.rs index 40f480207..11d061cdc 100644 --- a/quinn-proto/src/spaces.rs +++ b/quinn-proto/src/spaces.rs @@ -2,6 +2,7 @@ use std::collections::{BTreeMap, VecDeque}; use std::time::Instant; use std::{cmp, mem}; +use bytes::Bytes; use fnv::FnvHashSet; use crate::assembler::Assembler; @@ -121,6 +122,21 @@ impl PacketSpace { self.ecn_feedback = ecn; Ok(ce_increase != 0) } + + pub fn finish_stream(&mut self, id: StreamId, offset: u64) { + for frame in &mut self.pending.stream { + if frame.id == id && frame.offset + frame.data.len() as u64 == offset { + frame.fin = true; + return; + } + } + self.pending.stream.push_back(frame::Stream { + id, + data: Bytes::new(), + offset, + fin: true, + }); + } } /// Represents one or more packets subject to retransmission diff --git a/quinn/Cargo.toml b/quinn/Cargo.toml index adb567219..c2b6e82eb 100644 --- a/quinn/Cargo.toml +++ b/quinn/Cargo.toml @@ -21,9 +21,12 @@ travis-ci = { repository = "djc/quinn" } [dependencies] bytes = "0.4.7" +ct-logs = "0.5" err-derive = "0.1.5" fnv = "1.0.6" futures = "0.1.21" +libc = "0.2.49" +mio = "0.6" proto = { package = "quinn-proto", path = "../quinn-proto", version = "0.4.0" } rand = "0.6" rustls = { version = "0.15.2", features = ["quic"] } @@ -34,22 +37,20 @@ tokio-timer = "0.2.1" untrusted = "0.6.2" webpki = "0.19" webpki-roots = "0.16" -ct-logs = "0.5" -libc = "0.2.49" -mio = "0.6" [dev-dependencies] +crc = "1.8.1" +criterion = "0.2.10" +directories = "1.0.2" failure = "0.1" +rand = "0.6.5" +rcgen = "0.3" slog-term = "2" structopt = "0.2.7" tokio = "0.1.6" tokio-current-thread = "0.1" -url = "1.7" -rcgen = "0.3" -directories = "1.0.2" -rand = "0.6.5" -crc = "1.8.1" unwrap = "1.2.1" +url = "1.7" [[example]] name = "server" @@ -60,3 +61,7 @@ name = "client" [[example]] name = "insecure_connection" required-features = ["rustls/dangerous_configuration"] + +[[bench]] +name = "bench" +harness = false diff --git a/quinn/benches/bench.rs b/quinn/benches/bench.rs new file mode 100644 index 000000000..b71de3261 --- /dev/null +++ b/quinn/benches/bench.rs @@ -0,0 +1,182 @@ +use std::cell::RefCell; +use std::net::{IpAddr, Ipv6Addr, SocketAddr, UdpSocket}; +use std::rc::Rc; + +use criterion::{criterion_group, criterion_main, BatchSize, Benchmark, Criterion, Throughput}; +use futures::{Async, Future, Poll, Stream}; +use tokio; + +use quinn::{ClientConfigBuilder, Endpoint, NewStream, ReadError, RecvStream, ServerConfigBuilder}; + +criterion_group!(benches, throughput); +criterion_main!(benches); + +fn throughput(c: &mut Criterion) { + let mut server_config = ServerConfigBuilder::default(); + let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]); + let key = quinn::PrivateKey::from_der(&cert.serialize_private_key_der()).unwrap(); + let cert = quinn::Certificate::from_der(&cert.serialize_der()).unwrap(); + let cert_chain = quinn::CertificateChain::from_certs(vec![cert.clone()]); + server_config.certificate(cert_chain, key).unwrap(); + + let mut server = Endpoint::builder(); + server.listen(server_config.build()); + let server_sock = UdpSocket::bind(SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0)).unwrap(); + let server_addr = server_sock.local_addr().unwrap(); + let (server_driver, _, server_incoming) = server.with_socket(server_sock).unwrap(); + + let mut client_config = ClientConfigBuilder::default(); + client_config.add_certificate_authority(cert).unwrap(); + client_config.enable_keylog(); + let mut client = Endpoint::builder(); + client.default_client_config(client_config.build()); + let (client_driver, client, _) = client + .bind(SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0)) + .unwrap(); + + let mut runtime = tokio::runtime::current_thread::Runtime::new().unwrap(); + runtime.spawn(server_driver.map_err(|e| panic!("server driver failed: {}", e))); + runtime.spawn(client_driver.map_err(|e| panic!("client driver failed: {}", e))); + + let runtime = Rc::new(RefCell::new(runtime)); + runtime + .borrow_mut() + .spawn(server_incoming.for_each(move |connecting| { + connecting + .and_then(|(driver, _, incoming)| { + tokio::runtime::current_thread::spawn( + driver.map_err(|e| ignore_timeout("server connection driver", e)), + ); + incoming.for_each(|stream| { + if let NewStream::Uni(recv) = stream { + ReadAllUnordered { stream: recv }.map_err(|e| panic!(e)) + } else { + unreachable!("only benchmarking uni streams") + } + }) + }) + .map_err(|e| panic!("server connection establishment failed: {}", e)) + })); + + let (driver, connection, _) = runtime + .borrow_mut() + .block_on(client.connect(&server_addr, "localhost").unwrap()) + .unwrap(); + + runtime + .borrow_mut() + .spawn(driver.map_err(|e| ignore_timeout("client connection driver", e))); + + { + const DATA: &[u8] = &[0xAB; 128 * 1024]; + let runtime = runtime.clone(); + c.bench( + "throughput", + Benchmark::new("128kB", move |b| { + b.iter_batched( + || { + runtime + .borrow_mut() + .block_on(connection.open_uni()) + .expect("failed opening stream") + }, + |stream| { + runtime + .borrow_mut() + .block_on( + tokio::io::write_all(stream, DATA) + .map_err(|e| panic!("write to stream failed: {}", e)) + .and_then(|(stream, _)| { + tokio::io::shutdown(stream).map_err(|e| { + panic!("send stream shutdown failed: {}", e) + }) + }), + ) + .expect("failed writing data") + }, + BatchSize::PerIteration, + ) + }) + .throughput(Throughput::Bytes(DATA.len() as u32)), + ); + } + + /* + let (driver, connection, _) = runtime + .borrow_mut() + .block_on(client.connect(&server_addr, "localhost").unwrap()) + .unwrap(); + + runtime + .borrow_mut() + .spawn(driver.map_err(|e| ignore_timeout("client connection driver", e))); + + { + const DATA: &[u8] = &[0xAB; 32]; + let runtime = runtime.clone(); + c.bench( + "throughput", + Benchmark::new("32B", move |b| { + b.iter_batched( + || { + runtime + .borrow_mut() + .block_on(connection.open_uni()) + .expect("failed opening stream") + }, + |stream| { + runtime + .borrow_mut() + .block_on( + tokio::io::write_all(stream, DATA) + .map_err(|e| panic!("write to stream failed: {}", e)) + .and_then(|(stream, _)| { + tokio::io::shutdown(stream).map_err(|e| { + panic!("send stream shutdown failed: {}", e) + }) + }), + ) + .expect("failed writing data") + }, + BatchSize::PerIteration, + ) + }) + .throughput(Throughput::Bytes(DATA.len() as u32)), + ); + } + */ +} + +fn ignore_timeout(ty: &'static str, e: quinn::ConnectionError) { + use quinn::ConnectionError::*; + match e { + TimedOut => (), + e => panic!("{} failed: {:?}", ty, e), + } +} + +struct ReadAllUnordered { + stream: RecvStream, +} + +impl Future for ReadAllUnordered { + type Item = (); + type Error = ReadError; + + fn poll(&mut self) -> Poll { + loop { + match self.stream.poll_read_unordered() { + Ok(Async::Ready(_)) => {} + Ok(Async::NotReady) => { + return Ok(Async::NotReady); + } + Err(ReadError::Finished) => { + return Ok(Async::Ready(())); + } + Err(e) => { + return Err(e); + } + } + } + } +}