diff --git a/bench/src/bulk.rs b/bench/src/bulk.rs index 5f966a9ee..68253501c 100644 --- a/bench/src/bulk.rs +++ b/bench/src/bulk.rs @@ -25,7 +25,7 @@ fn main() { let mut endpoint = quinn::EndpointBuilder::default(); endpoint.listen(server_config.build()); let mut runtime = rt(); - let (driver, endpoint, incoming) = runtime.enter(|| { + let (endpoint, incoming) = runtime.enter(|| { endpoint .bind(&SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0)) .unwrap() @@ -33,13 +33,9 @@ fn main() { let server_addr = endpoint.local_addr().unwrap(); drop(endpoint); // Ensure server shuts down when finished let thread = std::thread::spawn(move || { - let handle = runtime.spawn(async { - driver.await.expect("server endpoint driver"); - }); if let Err(e) = runtime.block_on(server(incoming)) { eprintln!("server failed: {:#}", e); } - runtime.block_on(handle).expect("server run"); }); let mut runtime = rt(); @@ -53,13 +49,8 @@ fn main() { async fn server(mut incoming: quinn::Incoming) -> Result<()> { let handshake = incoming.next().await.unwrap(); let quinn::NewConnection { - driver, - mut uni_streams, - .. + mut uni_streams, .. } = handshake.await.context("handshake failed")?; - tokio::spawn(async { - driver.await.expect("server conn driver"); - }); let mut stream = uni_streams .next() .await @@ -75,27 +66,19 @@ async fn server(mut incoming: quinn::Incoming) -> Result<()> { } async fn client(server_addr: SocketAddr, server_cert: quinn::Certificate) -> Result<()> { - let (driver, endpoint, _) = quinn::EndpointBuilder::default() + let (endpoint, _) = quinn::EndpointBuilder::default() .bind(&SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0)) .unwrap(); - tokio::spawn(async { - driver.await.expect("client endpoint driver"); - }); let mut client_config = quinn::ClientConfigBuilder::default(); client_config .add_certificate_authority(server_cert) .unwrap(); - let quinn::NewConnection { - driver, connection, .. - } = endpoint + let quinn::NewConnection { connection, .. } = endpoint .connect_with(client_config.build(), &server_addr, "localhost") .unwrap() .await .context("unable to connect")?; - tokio::spawn(async { - let _ = driver.await; - }); trace!("connected"); let mut stream = connection diff --git a/interop/src/main.rs b/interop/src/main.rs index d4b1bc938..27c227ce0 100644 --- a/interop/src/main.rs +++ b/interop/src/main.rs @@ -7,7 +7,7 @@ use std::{ }; use anyhow::{anyhow, Result}; -use futures::{future, TryFutureExt}; +use futures::future; use lazy_static::lazy_static; use quinn_h3::Settings; use structopt::StructOpt; @@ -303,8 +303,7 @@ impl State { let mut endpoint = quinn::Endpoint::builder(); endpoint.default_client_config(client_config.clone()); - let (endpoint_driver, endpoint, _) = endpoint.bind(&"[::]:0".parse().unwrap())?; - tokio::spawn(endpoint_driver.unwrap_or_else(|e| eprintln!("IO error: {}", e))); + let (endpoint, _) = endpoint.bind(&"[::]:0".parse().unwrap())?; let h3_client = match peer.alpn { Alpn::Hq => None, @@ -336,10 +335,6 @@ impl State { .await .map_err(|e| anyhow!("failed to connect: {}", e))?; result.handshake = true; - let close_handle = tokio::spawn(tokio::time::timeout( - Duration::from_secs(2), - new_conn.driver, - )); let stream = new_conn .connection .open_bi() @@ -367,7 +362,6 @@ impl State { .into_0rtt() { Ok((new_conn, _)) => { - tokio::spawn(new_conn.driver.unwrap_or_else(|_| ())); let stream = new_conn .connection .open_bi() @@ -384,14 +378,13 @@ impl State { let new_conn = conn .await .map_err(|e| anyhow!("failed to connect: {}", e))?; - tokio::spawn(new_conn.driver.unwrap_or_else(|_| ())); new_conn.connection } }; result.resumption = !*saw_cert.lock().unwrap(); conn.close(0u32.into(), b"done"); - result.close = close_handle.await.is_ok(); + self.endpoint.wait_idle().await; Ok(result) } @@ -402,7 +395,6 @@ impl State { .connect(&self.remote, &self.host)? .await .map_err(|e| anyhow!("failed to connect: {}", e))?; - tokio::spawn(new_conn.driver.unwrap_or_else(|_| ())); let conn = new_conn.connection; // Make sure some traffic has gone both ways before the key update let stream = conn @@ -429,7 +421,6 @@ impl State { .connect(&self.remote, &self.host)? .await .map_err(|e| anyhow!("failed to connect: {}", e))?; - tokio::spawn(new_conn.driver.unwrap_or_else(|_| ())); let stream = new_conn .connection .open_bi() @@ -443,14 +434,12 @@ impl State { async fn rebind(&self) -> Result<()> { let mut endpoint = quinn::Endpoint::builder(); endpoint.default_client_config(self.client_config.clone()); - let (endpoint_driver, endpoint, _) = endpoint.bind(&"[::]:0".parse().unwrap())?; - tokio::spawn(endpoint_driver.unwrap_or_else(|e| eprintln!("IO error: {}", e))); + let (endpoint, _) = endpoint.bind(&"[::]:0".parse().unwrap())?; let new_conn = endpoint .connect(&self.remote, &self.host)? .await .map_err(|e| anyhow!("failed to connect: {}", e))?; - tokio::spawn(new_conn.driver.unwrap_or_else(|_| ())); let socket = std::net::UdpSocket::bind("[::]:0").unwrap(); endpoint.rebind(socket)?; let stream = new_conn @@ -467,7 +456,7 @@ impl State { if let Alpn::Hq = self.peer.alpn { return Err(anyhow!("H3 not implemented on this peer")); } - let (quic_driver, h3_driver, conn) = self + let conn = self .h3_client .as_ref() .unwrap() @@ -475,9 +464,6 @@ impl State { .await .map_err(|e| anyhow!("h3 failed to connect: {}", e))?; - tokio::spawn(h3_driver.unwrap_or_else(|_| ())); - tokio::spawn(quic_driver.unwrap_or_else(|_| ())); - h3_get(&conn, &self.peer.uri("/")) .await .map_err(|e| anyhow!("h3 request failed: {}", e))?; @@ -488,19 +474,13 @@ impl State { async fn core_h3(&self) -> Result { let mut result = InteropResult::default(); - let (quic_driver, h3_driver, conn) = self + let conn = self .h3_client .as_ref() .unwrap() .connect(&self.remote, &self.host)? .await .map_err(|e| anyhow!("h3 failed to connect: {}", e))?; - let close_handle = tokio::spawn(tokio::time::timeout(Duration::from_secs(2), quic_driver)); - tokio::spawn(async { - if let Err(e) = h3_driver.await { - error!("h3 driver error: {}", e); - } - }); result.handshake = true; h3_get(&conn, &self.peer.uri("/")) .await @@ -525,9 +505,7 @@ impl State { .connect_with(client_config, &self.remote, &self.host)? .into_0rtt() { - Ok((quic_driver, driver, conn, _)) => { - tokio::spawn(quic_driver.unwrap_or_else(|_| ())); - tokio::spawn(driver.unwrap_or_else(|_| ())); + Ok((conn, _)) => { h3_get(&conn, &self.peer.uri("/")) .await .map_err(|e| anyhow!("0-RTT request failed: {}", e))?; @@ -536,31 +514,29 @@ impl State { } Err(connecting) => { info!("0-RTT unsupported"); - let (quic_driver, _, new_conn) = connecting + connecting .await - .map_err(|e| anyhow!("failed to connect: {}", e))?; - tokio::spawn(quic_driver.unwrap_or_else(|_| ())); - new_conn + .map_err(|e| anyhow!("failed to connect: {}", e))? } }; result.resumption = !*saw_cert.lock().unwrap(); conn.close(); - result.close = close_handle.await.is_ok(); + self.endpoint.wait_idle().await; + + result.close = true; Ok(result) } async fn key_update_h3(&self) -> Result<()> { - let (quic_driver, h3_driver, conn) = self + let conn = self .h3_client .as_ref() .unwrap() .connect(&self.remote, &self.host)? .await .map_err(|e| anyhow!("h3 failed to connect: {}", e))?; - tokio::spawn(quic_driver.unwrap_or_else(|_| ())); - tokio::spawn(h3_driver.unwrap_or_else(|_| ())); // Make sure some traffic has gone both ways before the key update h3_get(&conn, &self.peer.uri("/")) .await @@ -577,15 +553,13 @@ impl State { let mut remote = self.remote; remote.set_port(self.peer.retry_port); - let (quic_driver, h3_driver, conn) = self + let conn = self .h3_client .as_ref() .unwrap() .connect(&self.remote, &self.host)? .await .map_err(|e| anyhow!("h3 failed to connect: {}", e))?; - tokio::spawn(quic_driver.unwrap_or_else(|_| ())); - tokio::spawn(h3_driver.unwrap_or_else(|_| ())); h3_get(&conn, &self.peer.uri("/")) .await .map_err(|e| anyhow!("request failed on retry port: {}", e))?; @@ -594,17 +568,13 @@ impl State { } async fn rebind_h3(&self) -> Result<()> { - let (endpoint_driver, endpoint, _) = - quinn::Endpoint::builder().bind(&"[::]:0".parse().unwrap())?; - tokio::spawn(endpoint_driver.unwrap_or_else(|e| eprintln!("IO error: {}", e))); + let (endpoint, _) = quinn::Endpoint::builder().bind(&"[::]:0".parse().unwrap())?; let h3_client = quinn_h3::client::Builder::default().endpoint(self.endpoint.clone()); - let (quic_driver, h3_driver, conn) = h3_client + let conn = h3_client .connect(&self.remote, &self.host)? .await .map_err(|e| anyhow!("h3 failed to connect: {}", e))?; - tokio::spawn(quic_driver.unwrap_or_else(|_| ())); - tokio::spawn(h3_driver.unwrap_or_else(|_| ())); let socket = std::net::UdpSocket::bind("[::]:0").unwrap(); endpoint.rebind(socket)?; h3_get(&conn, &self.peer.uri("/")) diff --git a/quinn-debug b/quinn-debug new file mode 100644 index 000000000..cf3699861 --- /dev/null +++ b/quinn-debug @@ -0,0 +1,390 @@ +Wireshark SSL debug log + +Wireshark version: 3.1.0rc0-765-g3234152b (v3.1.0rc0-765-g3234152b) +GnuTLS version: 3.4.17 +Libgcrypt version: 1.7.7 + + +dissect_ssl enter frame #736 (first time) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x7fcf29431510 + record: offset = 0, reported_length_remaining = 50 +ssl_try_set_version found version 0x0303 -> state 0x10 +dissect_ssl3_record: content_type 23 Application Data +decrypt_ssl3_record: app_data len 45, ssl state 0x10 +packet_from_server: is from server - FALSE +decrypt_ssl3_record: using client decoder +decrypt_ssl3_record: no decoder available + +dissect_ssl enter frame #738 (first time) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x7fcf294321f0 + record: offset = 0, reported_length_remaining = 50 +ssl_try_set_version found version 0x0303 -> state 0x10 +dissect_ssl3_record: content_type 23 Application Data +decrypt_ssl3_record: app_data len 45, ssl state 0x10 +packet_from_server: is from server - FALSE +decrypt_ssl3_record: using client decoder +decrypt_ssl3_record: no decoder available + +dissect_ssl enter frame #739 (first time) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x7fcf29432cc0 + record: offset = 0, reported_length_remaining = 50 +ssl_try_set_version found version 0x0303 -> state 0x10 +dissect_ssl3_record: content_type 23 Application Data +decrypt_ssl3_record: app_data len 45, ssl state 0x10 +packet_from_server: is from server - FALSE +decrypt_ssl3_record: using client decoder +decrypt_ssl3_record: no decoder available + +dissect_ssl enter frame #742 (first time) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x7fcf29433bb0 + record: offset = 0, reported_length_remaining = 50 +ssl_try_set_version found version 0x0303 -> state 0x10 +dissect_ssl3_record: content_type 23 Application Data +decrypt_ssl3_record: app_data len 45, ssl state 0x10 +packet_from_server: is from server - FALSE +decrypt_ssl3_record: using client decoder +decrypt_ssl3_record: no decoder available + +dissect_ssl enter frame #744 (first time) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x7fcf29432cc0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data +decrypt_ssl3_record: app_data len 45, ssl state 0x10 +packet_from_server: is from server - FALSE +decrypt_ssl3_record: using client decoder +decrypt_ssl3_record: no decoder available + +dissect_ssl enter frame #745 (first time) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x7fcf294321f0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data +decrypt_ssl3_record: app_data len 45, ssl state 0x10 +packet_from_server: is from server - FALSE +decrypt_ssl3_record: using client decoder +decrypt_ssl3_record: no decoder available + +dissect_ssl enter frame #746 (first time) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x7fcf29433bb0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data +decrypt_ssl3_record: app_data len 45, ssl state 0x10 +packet_from_server: is from server - FALSE +decrypt_ssl3_record: using client decoder +decrypt_ssl3_record: no decoder available + +dissect_ssl enter frame #750 (first time) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x7fcf29431510 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data +decrypt_ssl3_record: app_data len 45, ssl state 0x10 +packet_from_server: is from server - FALSE +decrypt_ssl3_record: using client decoder +decrypt_ssl3_record: no decoder available + +dissect_ssl enter frame #736 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #738 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #739 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #742 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #744 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #745 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #746 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #750 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #750 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #750 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #736 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #738 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #739 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #742 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #744 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #745 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #746 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #750 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #736 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #738 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #739 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #742 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #744 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #745 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #746 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #750 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #736 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #738 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #739 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #742 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #744 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #745 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #746 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #750 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #736 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #738 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #739 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #742 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #744 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #745 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #746 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #750 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #736 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #738 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #739 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #742 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #744 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29432610, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #745 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29431b40, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #746 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29433500, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data + +dissect_ssl enter frame #750 (already visited) +packet_from_server: is from server - FALSE + conversation = 0x7fcf29430e60, ssl_session = 0x0 + record: offset = 0, reported_length_remaining = 50 +dissect_ssl3_record: content_type 23 Application Data diff --git a/quinn-h3/examples/h3_client.rs b/quinn-h3/examples/h3_client.rs index fc858631c..9c2b3d06d 100644 --- a/quinn-h3/examples/h3_client.rs +++ b/quinn-h3/examples/h3_client.rs @@ -50,12 +50,7 @@ async fn main() -> Result<()> { } } - let (endpoint_driver, client) = client.build()?; - tokio::spawn(async move { - if let Err(e) = endpoint_driver.await { - eprintln!("quic driver error: {}", e) - } - }); + let client = client.build()?; match request(client, &options.uri).await { Ok(_) => println!("client finished"), @@ -70,23 +65,11 @@ async fn request(client: Client, uri: &Uri) -> Result<()> { .to_socket_addrs()? .next() .ok_or_else(|| anyhow!("couldn't resolve to an address"))?; - let (quic_driver, h3_driver, conn) = client + let conn = client .connect(&remote, uri.host().unwrap_or("localhost"))? .await .map_err(|e| anyhow!("failed ot connect: {:?}", e))?; - tokio::spawn(async move { - if let Err(e) = h3_driver.await { - eprintln!("h3 client error: {}", e) - } - }); - - tokio::spawn(async move { - if let Err(e) = quic_driver.await { - eprintln!("h3 client error: {}", e) - } - }); - let request = Request::get(uri) .header("client", "quinn-h3:0.0.1") .body(()) diff --git a/quinn-h3/examples/h3_server.rs b/quinn-h3/examples/h3_server.rs index 585e66e61..5afb88fca 100644 --- a/quinn-h3/examples/h3_server.rs +++ b/quinn-h3/examples/h3_server.rs @@ -6,10 +6,8 @@ use http::{Response, StatusCode}; use structopt::{self, StructOpt}; use tracing::error; -use quinn::ConnectionDriver as QuicDriver; use quinn_h3::{ self, - connection::ConnectionDriver, server::{Builder as ServerBuilder, IncomingRequest, RecvRequest}, }; @@ -47,17 +45,11 @@ async fn main() -> Result<()> { .certificate(certs.0, certs.2) .expect("failed to add cert"); - let (endpoint_driver, mut incoming) = { - let (driver, _server, incoming) = server.build().expect("bind failed"); - (driver, incoming) + let mut incoming = { + let (_server, incoming) = server.build().expect("bind failed"); + incoming }; - tokio::spawn(async move { - if let Err(e) = endpoint_driver.await { - eprintln!("h3 server error: {}", e) - } - }); - println!("server listening"); while let Some(connecting) = incoming.next().await { println!("server received connection"); @@ -76,27 +68,13 @@ async fn main() -> Result<()> { Ok(()) } -async fn handle_connection(conn: (QuicDriver, ConnectionDriver, IncomingRequest)) -> Result<()> { - let (quic_driver, h3_driver, mut incoming) = conn; - - tokio::spawn(async move { - if let Err(e) = h3_driver.await { - eprintln!("h3 connection driver error: {}", e) - } - }); - - tokio::spawn(async move { - while let Some(request) = incoming.next().await { - tokio::spawn(async move { - if let Err(e) = handle_request(request).await { - eprintln!("request error: {}", e) - } - }); - } - }); - - if let Err(e) = quic_driver.await { - eprintln!("quic connection driver error: {}", e) +async fn handle_connection(mut incoming: IncomingRequest) -> Result<()> { + while let Some(request) = incoming.next().await { + tokio::spawn(async move { + if let Err(e) = handle_request(request).await { + eprintln!("request error: {}", e) + } + }); } Ok(()) diff --git a/quinn-h3/src/client.rs b/quinn-h3/src/client.rs index a0dd9e98e..7d44d31ab 100644 --- a/quinn-h3/src/client.rs +++ b/quinn-h3/src/client.rs @@ -72,18 +72,15 @@ impl Builder { } } - pub fn build(self) -> Result<(quinn::EndpointDriver, Client), quinn::EndpointError> { + pub fn build(self) -> Result { let mut endpoint_builder = quinn::Endpoint::builder(); endpoint_builder.default_client_config(self.client_config.build()); - let (endpoint_driver, endpoint, _) = endpoint_builder.bind(&"[::]:0".parse().unwrap())?; + let (endpoint, _) = endpoint_builder.bind(&"[::]:0".parse().unwrap())?; - Ok(( - endpoint_driver, - Client { - endpoint, - settings: self.settings, - }, - )) + Ok(Client { + endpoint, + settings: self.settings, + }) } } @@ -182,17 +179,7 @@ pub struct Connecting { } impl Connecting { - pub fn into_0rtt( - self, - ) -> Result< - ( - quinn::ConnectionDriver, - ConnectionDriver, - Connection, - ZeroRttAccepted, - ), - Self, - > { + pub fn into_0rtt(self) -> Result<(Connection, ZeroRttAccepted), Self> { let Self { connecting, settings, @@ -204,7 +191,6 @@ impl Connecting { }), Ok((new_conn, zero_rtt)) => { let quinn::NewConnection { - driver, connection, uni_streams, bi_streams, @@ -213,23 +199,18 @@ impl Connecting { let conn_ref = ConnectionRef::new(connection, Side::Client, uni_streams, bi_streams, settings) .expect("error in h3 settings"); // FIXME return an error type - Ok(( - driver, - ConnectionDriver(conn_ref.clone()), - Connection(conn_ref), - ZeroRttAccepted(zero_rtt), - )) + tokio::spawn(ConnectionDriver(conn_ref.clone())); + Ok((Connection(conn_ref), ZeroRttAccepted(zero_rtt))) } } } } impl Future for Connecting { - type Output = Result<(quinn::ConnectionDriver, ConnectionDriver, Connection), Error>; + type Output = Result; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let quinn::NewConnection { - driver, connection, uni_streams, bi_streams, @@ -242,11 +223,8 @@ impl Future for Connecting { bi_streams, self.settings.clone(), )?; - Poll::Ready(Ok(( - driver, - ConnectionDriver(conn_ref.clone()), - Connection(conn_ref), - ))) + tokio::spawn(ConnectionDriver(conn_ref.clone())); + Poll::Ready(Ok(Connection(conn_ref))) } } diff --git a/quinn-h3/src/connection.rs b/quinn-h3/src/connection.rs index 0b9f914ad..af09b4c4f 100644 --- a/quinn-h3/src/connection.rs +++ b/quinn-h3/src/connection.rs @@ -28,19 +28,19 @@ use crate::{ Error, Settings, }; -pub struct ConnectionDriver(pub(crate) ConnectionRef); +pub(crate) struct ConnectionDriver(pub(crate) ConnectionRef); impl Future for ConnectionDriver { - type Output = Result<(), Error>; + type Output = (); fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll { let res = self.0.h3.lock().unwrap().drive(cx); match res { Ok(false) => Poll::Pending, - Ok(true) => Poll::Ready(Ok(())), - Err(DriverError(err, code, msg)) => { + Ok(true) => Poll::Ready(()), + Err(DriverError(_err, code, msg)) => { self.0.quic.close(code.into(), msg.as_bytes()); - Poll::Ready(Err(err)) + Poll::Ready(()) } } } diff --git a/quinn-h3/src/server.rs b/quinn-h3/src/server.rs index 60529c6f1..45752d357 100644 --- a/quinn-h3/src/server.rs +++ b/quinn-h3/src/server.rs @@ -82,14 +82,13 @@ impl Builder { pub fn endpoint( self, endpoint: EndpointBuilder, - ) -> Result<(quinn::EndpointDriver, Server, IncomingConnection), quinn::EndpointError> { + ) -> Result<(Server, IncomingConnection), quinn::EndpointError> { let listen = self .listen .unwrap_or_else(|| "[::]:4433".parse().expect("valid listen address")); - let (endpoint_driver, _, incoming) = endpoint.bind(&listen)?; + let (_, incoming) = endpoint.bind(&listen)?; Ok(( - endpoint_driver, Server, IncomingConnection { incoming, @@ -98,19 +97,16 @@ impl Builder { )) } - pub fn build( - self, - ) -> Result<(quinn::EndpointDriver, Server, IncomingConnection), quinn::EndpointError> { + pub fn build(self) -> Result<(Server, IncomingConnection), quinn::EndpointError> { let mut endpoint_builder = quinn::Endpoint::builder(); endpoint_builder.listen(self.config.build()); let listen = self .listen .unwrap_or_else(|| "[::]:4433".parse().expect("valid listen address")); - let (endpoint_driver, _, incoming) = endpoint_builder.bind(&listen)?; + let (_, incoming) = endpoint_builder.bind(&listen)?; Ok(( - endpoint_driver, Server, IncomingConnection { incoming, @@ -146,11 +142,10 @@ pub struct Connecting { } impl Future for Connecting { - type Output = Result<(quinn::ConnectionDriver, ConnectionDriver, IncomingRequest), Error>; + type Output = Result; fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { let quinn::NewConnection { - driver, connection, bi_streams, uni_streams, @@ -163,11 +158,8 @@ impl Future for Connecting { bi_streams, self.settings.clone(), )?; - Poll::Ready(Ok(( - driver, - ConnectionDriver(conn_ref.clone()), - IncomingRequest(conn_ref), - ))) + tokio::spawn(ConnectionDriver(conn_ref.clone())); + Poll::Ready(Ok(IncomingRequest(conn_ref))) } } diff --git a/quinn/Cargo.toml b/quinn/Cargo.toml index 1ce0ab44c..60231866b 100644 --- a/quinn/Cargo.toml +++ b/quinn/Cargo.toml @@ -46,7 +46,7 @@ directories = "2.0.0" rand = "0.7" rcgen = "0.7" structopt = "0.3.0" -tokio = { version = "0.2.2", features = ["rt-core", "rt-threaded", "time", "macros"] } +tokio = { version = "0.2.6", features = ["rt-threaded", "time", "macros"] } tracing-subscriber = "0.1.5" tracing-futures = { version = "0.2.0", default-features = false, features = ["std-future"] } unwrap = "1.2.1" diff --git a/quinn/benches/bench.rs b/quinn/benches/bench.rs index 0ddf53d77..f2558298f 100644 --- a/quinn/benches/bench.rs +++ b/quinn/benches/bench.rs @@ -6,10 +6,7 @@ use std::{ use criterion::{criterion_group, criterion_main, Criterion, Throughput}; use futures::StreamExt; -use tokio::{ - runtime::{Builder, Runtime}, - task::JoinHandle, -}; +use tokio::runtime::{Builder, Runtime}; use tracing::error_span; use tracing_futures::Instrument as _; @@ -30,7 +27,7 @@ fn throughput(c: &mut Criterion) { let mut group = c.benchmark_group("throughput"); { let (addr, thread) = ctx.spawn_server(); - let (client, mut runtime, handle) = ctx.make_client(addr); + let (endpoint, client, mut runtime) = ctx.make_client(addr); const DATA: &[u8] = &[0xAB; 128 * 1024]; group.throughput(Throughput::Bytes(DATA.len() as u64)); group.bench_function("large streams", |b| { @@ -43,13 +40,13 @@ fn throughput(c: &mut Criterion) { }) }); drop(client); - runtime.block_on(handle).unwrap(); + runtime.block_on(endpoint.wait_idle()); thread.join().unwrap(); } { let (addr, thread) = ctx.spawn_server(); - let (client, mut runtime, handle) = ctx.make_client(addr); + let (endpoint, client, mut runtime) = ctx.make_client(addr); const DATA: &[u8] = &[0xAB; 1]; group.throughput(Throughput::Elements(1)); group.bench_function("small streams", |b| { @@ -61,7 +58,7 @@ fn throughput(c: &mut Criterion) { }) }); drop(client); - runtime.block_on(handle).unwrap(); + runtime.block_on(endpoint.wait_idle()); thread.join().unwrap(); } @@ -104,27 +101,17 @@ impl Context { let mut endpoint = Endpoint::builder(); endpoint.listen(config); let mut runtime = rt(); - let (driver, _, mut incoming) = runtime.enter(|| endpoint.with_socket(sock).unwrap()); - runtime.spawn(async { driver.instrument(error_span!("server")).await.unwrap() }); + let (_, mut incoming) = runtime.enter(|| endpoint.with_socket(sock).unwrap()); let handle = runtime.spawn( async move { let quinn::NewConnection { - driver, - mut uni_streams, - .. + mut uni_streams, .. } = incoming .next() .await .expect("accept") .await .expect("connect"); - tokio::spawn(async move { - match driver.instrument(error_span!("server")).await { - Ok(()) => panic!("unexpected success"), - Err(quinn::ConnectionError::ApplicationClosed { .. }) => {} - Err(e) => panic!("connection lost: {}", e), - } - }); while let Some(Ok(mut stream)) = uni_streams.next().await { while let Some(_) = stream.read_unordered().await.unwrap() {} } @@ -139,22 +126,14 @@ impl Context { pub fn make_client( &self, server_addr: SocketAddr, - ) -> (quinn::Connection, Runtime, JoinHandle<()>) { + ) -> (quinn::Endpoint, quinn::Connection, Runtime) { let mut runtime = rt(); - let (endpoint_driver, endpoint, _) = runtime.enter(|| { + let (endpoint, _) = runtime.enter(|| { Endpoint::builder() .bind(&SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0)) .unwrap() }); - runtime.spawn(async move { - endpoint_driver - .instrument(error_span!("client")) - .await - .unwrap(); - }); - let quinn::NewConnection { - driver, connection, .. - } = runtime + let quinn::NewConnection { connection, .. } = runtime .block_on( endpoint .connect_with(self.client_config.clone(), &server_addr, "localhost") @@ -162,10 +141,7 @@ impl Context { .instrument(error_span!("client")), ) .unwrap(); - let handle = runtime.spawn(async move { - driver.instrument(error_span!("client")).await.unwrap(); - }); - (connection, runtime, handle) + (endpoint, connection, runtime) } } diff --git a/quinn/examples/client.rs b/quinn/examples/client.rs index cb5f8c564..ccc405c65 100644 --- a/quinn/examples/client.rs +++ b/quinn/examples/client.rs @@ -7,9 +7,7 @@ use std::{ }; use anyhow::{anyhow, Result}; -use futures::TryFutureExt; use structopt::StructOpt; -use tokio::runtime::Builder; use tracing::{error, info}; use url::Url; @@ -57,7 +55,8 @@ fn main() { ::std::process::exit(code); } -fn run(options: Opt) -> Result<()> { +#[tokio::main] +async fn run(options: Opt) -> Result<()> { let url = options.url; let remote = (url.host_str().unwrap(), url.port().unwrap_or(4433)) .to_socket_addrs()? @@ -90,10 +89,7 @@ fn run(options: Opt) -> Result<()> { endpoint.default_client_config(client_config.build()); - let mut runtime = Builder::new().basic_scheduler().enable_all().build()?; - let (endpoint_driver, endpoint, _) = - runtime.enter(|| endpoint.bind(&"[::]:0".parse().unwrap()))?; - let handle = runtime.spawn(endpoint_driver.unwrap_or_else(|e| eprintln!("IO error: {}", e))); + let (endpoint, _) = endpoint.bind(&"[::]:0".parse().unwrap())?; let request = format!("GET {}\r\n", url.path()); let start = Instant::now(); @@ -103,59 +99,47 @@ fn run(options: Opt) -> Result<()> { .as_ref() .map_or_else(|| url.host_str(), |x| Some(&x)) .ok_or_else(|| anyhow!("no hostname specified"))?; - let r: Result<()> = runtime.block_on(async { - let new_conn = endpoint - .connect(&remote, &host)? - .await - .map_err(|e| anyhow!("failed to connect: {}", e))?; - eprintln!("connected at {:?}", start.elapsed()); - let quinn::NewConnection { - driver, - connection: conn, - .. - } = { new_conn }; - tokio::spawn(driver.unwrap_or_else(|e| eprintln!("connection lost: {}", e))); - let (mut send, recv) = conn - .open_bi() - .await - .map_err(|e| anyhow!("failed to open stream: {}", e))?; - if rebind { - let socket = std::net::UdpSocket::bind("[::]:0").unwrap(); - let addr = socket.local_addr().unwrap(); - eprintln!("rebinding to {}", addr); - endpoint.rebind(socket).expect("rebind failed"); - } - send.write_all(request.as_bytes()) - .await - .map_err(|e| anyhow!("failed to send request: {}", e))?; - send.finish() - .await - .map_err(|e| anyhow!("failed to shutdown stream: {}", e))?; - let response_start = Instant::now(); - eprintln!("request sent at {:?}", response_start - start); - let resp = recv - .read_to_end(usize::max_value()) - .await - .map_err(|e| anyhow!("failed to read response: {}", e))?; - let duration = response_start.elapsed(); - eprintln!( - "response received in {:?} - {} KiB/s", - duration, - resp.len() as f32 / (duration_secs(&duration) * 1024.0) - ); - io::stdout().write_all(&resp).unwrap(); - io::stdout().flush().unwrap(); - conn.close(0u32.into(), b"done"); - Ok(()) - }); - r?; + let new_conn = endpoint + .connect(&remote, &host)? + .await + .map_err(|e| anyhow!("failed to connect: {}", e))?; + eprintln!("connected at {:?}", start.elapsed()); + let quinn::NewConnection { + connection: conn, .. + } = { new_conn }; + let (mut send, recv) = conn + .open_bi() + .await + .map_err(|e| anyhow!("failed to open stream: {}", e))?; + if rebind { + let socket = std::net::UdpSocket::bind("[::]:0").unwrap(); + let addr = socket.local_addr().unwrap(); + eprintln!("rebinding to {}", addr); + endpoint.rebind(socket).expect("rebind failed"); + } - // Allow the endpoint driver to automatically shut down - drop(endpoint); - - // Let the connection finish closing gracefully - runtime.block_on(handle).unwrap(); + send.write_all(request.as_bytes()) + .await + .map_err(|e| anyhow!("failed to send request: {}", e))?; + send.finish() + .await + .map_err(|e| anyhow!("failed to shutdown stream: {}", e))?; + let response_start = Instant::now(); + eprintln!("request sent at {:?}", response_start - start); + let resp = recv + .read_to_end(usize::max_value()) + .await + .map_err(|e| anyhow!("failed to read response: {}", e))?; + let duration = response_start.elapsed(); + eprintln!( + "response received in {:?} - {} KiB/s", + duration, + resp.len() as f32 / (duration_secs(&duration) * 1024.0) + ); + io::stdout().write_all(&resp).unwrap(); + io::stdout().flush().unwrap(); + conn.close(0u32.into(), b"done"); Ok(()) } diff --git a/quinn/examples/common/mod.rs b/quinn/examples/common/mod.rs index 9b9dd5305..d65429cf0 100644 --- a/quinn/examples/common/mod.rs +++ b/quinn/examples/common/mod.rs @@ -1,8 +1,8 @@ //! Commonly used code in most examples. use quinn::{ - Certificate, CertificateChain, ClientConfig, ClientConfigBuilder, Endpoint, EndpointDriver, - Incoming, PrivateKey, ServerConfig, ServerConfigBuilder, TransportConfig, + Certificate, CertificateChain, ClientConfig, ClientConfigBuilder, Endpoint, Incoming, + PrivateKey, ServerConfig, ServerConfigBuilder, TransportConfig, }; use std::{error::Error, net::SocketAddr, sync::Arc}; @@ -15,12 +15,12 @@ use std::{error::Error, net::SocketAddr, sync::Arc}; pub fn make_client_endpoint( bind_addr: SocketAddr, server_certs: &[&[u8]], -) -> Result<(Endpoint, EndpointDriver), Box> { +) -> Result> { let client_cfg = configure_client(server_certs)?; let mut endpoint_builder = Endpoint::builder(); endpoint_builder.default_client_config(client_cfg); - let (driver, endpoint, _incoming) = endpoint_builder.bind(&bind_addr)?; - Ok((endpoint, driver)) + let (endpoint, _incoming) = endpoint_builder.bind(&bind_addr)?; + Ok(endpoint) } /// Constructs a QUIC endpoint configured to listen for incoming connections on a certain address @@ -28,18 +28,15 @@ pub fn make_client_endpoint( /// /// ## Returns /// -/// - UDP socket driver /// - a sream of incoming QUIC connections /// - server certificate serialized into DER format #[allow(unused)] -pub fn make_server_endpoint( - bind_addr: SocketAddr, -) -> Result<(EndpointDriver, Incoming, Vec), Box> { +pub fn make_server_endpoint(bind_addr: SocketAddr) -> Result<(Incoming, Vec), Box> { let (server_config, server_cert) = configure_server()?; let mut endpoint_builder = Endpoint::builder(); endpoint_builder.listen(server_config); - let (driver, _endpoint, incoming) = endpoint_builder.bind(&bind_addr)?; - Ok((driver, incoming, server_cert)) + let (_endpoint, incoming) = endpoint_builder.bind(&bind_addr)?; + Ok((incoming, server_cert)) } /// Builds default quinn client config and trusts given certificates. diff --git a/quinn/examples/connection.rs b/quinn/examples/connection.rs index 534ee0287..abdba63c9 100644 --- a/quinn/examples/connection.rs +++ b/quinn/examples/connection.rs @@ -22,9 +22,7 @@ use common::{make_client_endpoint, make_server_endpoint}; #[tokio::main] async fn main() -> Result<(), Box> { let server_addr = "127.0.0.1:5000".parse().unwrap(); - let (driver, mut incoming, server_cert) = make_server_endpoint(server_addr)?; - // drive server's UDP socket - tokio::spawn(async { driver.await.unwrap() }); + let (mut incoming, server_cert) = make_server_endpoint(server_addr)?; // accept a single connection tokio::spawn(async move { let incoming_conn = incoming.next().await.unwrap(); @@ -33,28 +31,21 @@ async fn main() -> Result<(), Box> { "[server] connection accepted: addr={}", new_conn.connection.remote_address() ); - // Drive the connection to completion - if let Err(e) = new_conn.driver.await { - println!("[server] connection lost: {}", e); - } }); - let (endpoint, driver) = make_client_endpoint("0.0.0.0:0".parse().unwrap(), &[&server_cert])?; - // drive client's UDP socket - tokio::spawn(async { driver.await.unwrap() }); + let endpoint = make_client_endpoint("0.0.0.0:0".parse().unwrap(), &[&server_cert])?; // connect to server - let quinn::NewConnection { - driver, connection, .. - } = endpoint + let quinn::NewConnection { connection, .. } = endpoint .connect(&server_addr, "localhost") .unwrap() .await .unwrap(); println!("[client] connected: addr={}", connection.remote_address()); - drop((endpoint, connection)); + drop(connection); - driver.await.unwrap(); + // Make sure the server has a chance to clean up + endpoint.wait_idle().await; Ok(()) } diff --git a/quinn/examples/insecure_connection.rs b/quinn/examples/insecure_connection.rs index b639939f5..ec7c0dc1e 100644 --- a/quinn/examples/insecure_connection.rs +++ b/quinn/examples/insecure_connection.rs @@ -24,9 +24,7 @@ async fn main() -> Result<(), Box> { /// Runs a QUIC server bound to given address. async fn run_server(addr: SocketAddr) { - let (driver, mut incoming, _server_cert) = make_server_endpoint(addr).unwrap(); - // drive UDP socket - tokio::spawn(async { driver.await.unwrap() }); + let (mut incoming, _server_cert) = make_server_endpoint(addr).unwrap(); // accept a single connection let incoming_conn = incoming.next().await.unwrap(); let new_conn = incoming_conn.await.unwrap(); @@ -34,10 +32,6 @@ async fn run_server(addr: SocketAddr) { "[server] connection accepted: addr={}", new_conn.connection.remote_address() ); - // Drive the connection to completion - if let Err(e) = new_conn.driver.await { - println!("[server] connection lost: {}", e); - } } async fn run_client(server_addr: SocketAddr) -> Result<(), Box> { @@ -45,22 +39,19 @@ async fn run_client(server_addr: SocketAddr) -> Result<(), Box> { let mut endpoint_builder = Endpoint::builder(); endpoint_builder.default_client_config(client_cfg); - let (driver, endpoint, _) = endpoint_builder.bind(&"127.0.0.1:0".parse().unwrap())?; - tokio::spawn(async { driver.await.unwrap() }); + let (endpoint, _) = endpoint_builder.bind(&"127.0.0.1:0".parse().unwrap())?; // connect to server - let quinn::NewConnection { - driver, connection, .. - } = endpoint + let quinn::NewConnection { connection, .. } = endpoint .connect(&server_addr, "localhost") .unwrap() .await .unwrap(); println!("[client] connected: addr={}", connection.remote_address()); // Dropping handles allows the corresponding objects to automatically shut down - drop((endpoint, connection)); - // Drive the connection to completion - driver.await.unwrap(); + drop(connection); + // Make sure the server has a chance to clean up + endpoint.wait_idle().await; Ok(()) } diff --git a/quinn/examples/server.rs b/quinn/examples/server.rs index d3c897dbb..a906aea8b 100644 --- a/quinn/examples/server.rs +++ b/quinn/examples/server.rs @@ -9,7 +9,6 @@ use std::{ use anyhow::{anyhow, bail, Context, Result}; use futures::{StreamExt, TryFutureExt}; use structopt::{self, StructOpt}; -use tokio::runtime::Builder; use tracing::{error, info, info_span}; use tracing_futures::Instrument as _; @@ -57,12 +56,12 @@ fn main() { ::std::process::exit(code); } -fn run(options: Opt) -> Result<()> { +#[tokio::main] +async fn run(options: Opt) -> Result<()> { let mut transport_config = quinn::TransportConfig::default(); transport_config.stream_window_uni(0); let mut server_config = quinn::ServerConfig::default(); server_config.transport = Arc::new(transport_config); - let mut server_config = quinn::ServerConfigBuilder::new(server_config); server_config.protocols(common::ALPN_QUIC_HTTP); @@ -122,32 +121,26 @@ fn run(options: Opt) -> Result<()> { bail!("root path does not exist"); } - let mut runtime = Builder::new().threaded_scheduler().enable_all().build()?; - - let (endpoint_driver, mut incoming) = { - let (driver, endpoint, incoming) = runtime.enter(|| endpoint.bind(&options.listen))?; + let mut incoming = { + let (endpoint, incoming) = endpoint.bind(&options.listen)?; info!("listening on {}", endpoint.local_addr()?); - (driver, incoming) + incoming }; - runtime.spawn(async move { - while let Some(conn) = incoming.next().await { - info!("connection incoming"); - tokio::spawn( - handle_connection(root.clone(), conn).unwrap_or_else(move |e| { - error!("connection failed: {reason}", reason = e.to_string()) - }), - ); - } - }); - runtime.block_on(endpoint_driver)?; + while let Some(conn) = incoming.next().await { + info!("connection incoming"); + tokio::spawn( + handle_connection(root.clone(), conn).unwrap_or_else(move |e| { + error!("connection failed: {reason}", reason = e.to_string()) + }), + ); + } Ok(()) } async fn handle_connection(root: Arc, conn: quinn::Connecting) -> Result<()> { let quinn::NewConnection { - driver, connection, mut bi_streams, .. @@ -157,12 +150,9 @@ async fn handle_connection(root: Arc, conn: quinn::Connecting) -> Result<( remote = %connection.remote_address(), protocol = %connection.protocol().map_or_else(|| "".into(), |x| String::from_utf8_lossy(&x).into_owned()) ); - tokio::spawn(driver.unwrap_or_else(|_| ()).instrument(span.clone())); async { info!("established"); - // We ignore errors from the driver because they'll be reported by the `streams` handler anyway. - // Each stream initiated by the client constitutes a new request. while let Some(stream) = bi_streams.next().await { let stream = match stream { diff --git a/quinn/examples/single_socket.rs b/quinn/examples/single_socket.rs index 70717bbc1..ae491540e 100644 --- a/quinn/examples/single_socket.rs +++ b/quinn/examples/single_socket.rs @@ -41,50 +41,43 @@ async fn main() -> Result<(), Box> { let server2_cert = run_server(addr2)?; let server3_cert = run_server(addr3)?; - let (client, driver) = make_client_endpoint( + let client = make_client_endpoint( "127.0.0.1:0".parse().unwrap(), &[&server1_cert, &server2_cert, &server3_cert], )?; // connect to multiple endpoints using the same socket/endpoint - run_client(&client, addr1); - run_client(&client, addr2); - run_client(&client, addr3); - drop(client); + futures::future::join_all(vec![ + run_client(&client, addr1), + run_client(&client, addr2), + run_client(&client, addr3), + ]) + .await; + + // Make sure the server has a chance to clean up + client.wait_idle().await; - // drive client endpoint to completion - driver.await.unwrap(); Ok(()) } /// Runs a QUIC server bound to given address and returns server certificate. fn run_server(addr: SocketAddr) -> Result, Box> { - let (driver, mut incoming, server_cert) = make_server_endpoint(addr)?; - // drive UDP socket - tokio::spawn(async { driver.await.unwrap() }); + let (mut incoming, server_cert) = make_server_endpoint(addr)?; // accept a single connection tokio::spawn(async move { - let quinn::NewConnection { - driver, connection, .. - } = incoming.next().await.unwrap().await.unwrap(); + let quinn::NewConnection { connection, .. } = incoming.next().await.unwrap().await.unwrap(); println!( "[server] incoming connection: addr={}", connection.remote_address() ); - let _ = driver.await; }); Ok(server_cert) } /// Attempt QUIC connection with the given server address. -fn run_client(endpoint: &Endpoint, server_addr: SocketAddr) { +async fn run_client(endpoint: &Endpoint, server_addr: SocketAddr) { let connect = endpoint.connect(&server_addr, "localhost").unwrap(); - tokio::spawn(async { - let quinn::NewConnection { - driver, connection, .. - } = connect.await.unwrap(); - tokio::spawn(async { driver.await.unwrap() }); - println!("[client] connected: addr={}", connection.remote_address()); - }); + let quinn::NewConnection { connection, .. } = connect.await.unwrap(); + println!("[client] connected: addr={}", connection.remote_address()); } diff --git a/quinn/src/broadcast.rs b/quinn/src/broadcast.rs index 3638c8cd7..51463fed8 100644 --- a/quinn/src/broadcast.rs +++ b/quinn/src/broadcast.rs @@ -19,6 +19,7 @@ use std::task::{Context, Waker}; /// wakeup is genuine but the condition of interest has already passed, then the task's generation /// no longer matches the counter, and we infer that the task's `Waker` is no longer stored and a /// new one must be recorded. +#[derive(Debug)] pub struct Broadcast { wakers: Vec, generation: u64, diff --git a/quinn/src/builders.rs b/quinn/src/builders.rs index e32810932..da772d80b 100644 --- a/quinn/src/builders.rs +++ b/quinn/src/builders.rs @@ -3,6 +3,7 @@ use std::{io, net::SocketAddr, str, sync::Arc}; use err_derive::Error; use proto::{ClientConfig, EndpointConfig, ServerConfig}; use rustls::TLSError; +use tracing::error; use crate::{ endpoint::{Endpoint, EndpointDriver, EndpointRef, Incoming}, @@ -32,11 +33,10 @@ impl EndpointBuilder { } } - /// Build an endpoint bound to `addr`. - pub fn bind( - self, - addr: &SocketAddr, - ) -> Result<(EndpointDriver, Endpoint, Incoming), EndpointError> { + /// Build an endpoint bound to `addr` + /// + /// Must be called from within a tokio runtime context. + pub fn bind(self, addr: &SocketAddr) -> Result<(Endpoint, Incoming), EndpointError> { let socket = std::net::UdpSocket::bind(addr).map_err(EndpointError::Socket)?; self.with_socket(socket) } @@ -45,7 +45,7 @@ impl EndpointBuilder { pub fn with_socket( self, socket: std::net::UdpSocket, - ) -> Result<(EndpointDriver, Endpoint, Incoming), EndpointError> { + ) -> Result<(Endpoint, Incoming), EndpointError> { let addr = socket.local_addr().map_err(EndpointError::Socket)?; let socket = UdpSocket::from_std(socket).map_err(EndpointError::Socket)?; let rc = EndpointRef::new( @@ -53,8 +53,13 @@ impl EndpointBuilder { proto::Endpoint::new(Arc::new(self.config), self.server_config.map(Arc::new)), addr.is_ipv6(), ); + let driver = EndpointDriver(rc.clone()); + tokio::spawn(async { + if let Err(e) = driver.await { + error!("I/O error: {}", e); + } + }); Ok(( - EndpointDriver(rc.clone()), Endpoint { inner: rc.clone(), default_client_config: self.client_config, diff --git a/quinn/src/connection.rs b/quinn/src/connection.rs index 0126e9f1d..b101dbe16 100644 --- a/quinn/src/connection.rs +++ b/quinn/src/connection.rs @@ -26,8 +26,7 @@ use crate::{ }; /// In-progress connection attempt future -/// -/// Be sure to spawn the `ConnectionDriver` when complete. +#[derive(Debug)] pub struct Connecting(Option); impl Connecting { @@ -35,8 +34,7 @@ impl Connecting { Self(Some(ConnectionDriver(conn))) } - /// Convert into a 0-RTT or 0.5-RTT connection at the cost of weakened security. Be sure to - /// spawn the `ConnectionDriver`. + /// Convert into a 0-RTT or 0.5-RTT connection at the cost of weakened security /// /// Opens up the connection for use before the handshake finishes, allowing the API user to /// send data with 0-RTT encryption if the necessary key material is available. This is useful @@ -87,7 +85,7 @@ impl Future for Connecting { fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { let connected = match self.0 { Some(ref mut driver) => { - let r = driver.poll_unpin(cx)?; + let r = driver.poll_unpin(cx); let driver = driver.0.lock().unwrap(); match r { Poll::Ready(()) => { @@ -132,8 +130,6 @@ impl Future for ZeroRttAccepted { /// Components of a newly established connection /// -/// Ensure `driver` runs or the connection will not work. -/// /// All fields of this struct, in addition to any other handles constructed later, must be dropped /// for a connection to be implicitly closed. If the `NewConnection` is stored in a long-lived /// variable, moving individual fields won't cause remaining unused fields to be dropped, even with @@ -145,7 +141,7 @@ impl Future for ZeroRttAccepted { /// ```rust /// # use quinn::NewConnection; /// # fn dummy(new_connection: NewConnection) { -/// let NewConnection { driver, connection, .. } = { new_connection }; +/// let NewConnection { connection, .. } = { new_connection }; /// # } /// ``` /// @@ -153,8 +149,6 @@ impl Future for ZeroRttAccepted { #[derive(Debug)] #[non_exhaustive] pub struct NewConnection { - /// The future responsible for handling I/O on the connection - pub driver: ConnectionDriver, /// Handle for interacting with the connection pub connection: Connection, /// Unidirectional streams initiated by the peer, in the order they were opened @@ -171,8 +165,8 @@ pub struct NewConnection { impl NewConnection { fn new(conn: ConnectionRef) -> Self { + tokio::spawn(ConnectionDriver(conn.clone())); Self { - driver: ConnectionDriver(conn.clone()), connection: Connection(conn.clone()), uni_streams: IncomingUniStreams(conn.clone()), bi_streams: IncomingBiStreams(conn.clone()), @@ -193,15 +187,15 @@ impl NewConnection { /// packets still in flight from the peer are handled gracefully. #[must_use = "connection drivers must be spawned for their connections to function"] #[derive(Debug)] -pub struct ConnectionDriver(pub(crate) ConnectionRef); +pub(crate) struct ConnectionDriver(pub(crate) ConnectionRef); impl Future for ConnectionDriver { - type Output = Result<(), ConnectionError>; + type Output = (); fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { let conn = &mut *self.0.lock().unwrap(); if let Some(ref e) = conn.error { if *e != ConnectionError::LocallyClosed { - return Poll::Ready(Err(e.clone())); + return Poll::Ready(()); } } @@ -212,8 +206,8 @@ impl Future for ConnectionDriver { let now = Instant::now(); let mut keep_going = false; if let Err(e) = conn.process_conn_events(cx) { - conn.terminate(e.clone()); - return Poll::Ready(Err(e)); + conn.terminate(e); + return Poll::Ready(()); } conn.drive_transmit(now); keep_going |= conn.drive_timer(cx, now); @@ -229,8 +223,8 @@ impl Future for ConnectionDriver { return Poll::Pending; } match conn.error { - Some(ConnectionError::LocallyClosed) => Poll::Ready(Ok(())), - Some(ref e) => Poll::Ready(Err(e.clone())), + Some(ConnectionError::LocallyClosed) => Poll::Ready(()), + Some(_) => Poll::Ready(()), None => unreachable!("drained connections always have an error"), } } diff --git a/quinn/src/endpoint.rs b/quinn/src/endpoint.rs index 783bbaa2d..c210bf772 100644 --- a/quinn/src/endpoint.rs +++ b/quinn/src/endpoint.rs @@ -15,6 +15,7 @@ use futures::{channel::mpsc, FutureExt, StreamExt}; use proto::{self as proto, ClientConfig, ConnectError, ConnectionHandle, DatagramEvent}; use crate::{ + broadcast::{self, Broadcast}, builders::EndpointBuilder, connection::{Connecting, ConnectionDriver, ConnectionRef}, udp::UdpSocket, @@ -39,7 +40,7 @@ impl Endpoint { EndpointBuilder::default() } - /// Connect to a remote endpoint. Be sure to spawn the `ConnectionDriver` after connecting. + /// Connect to a remote endpoint /// /// `server_name` must be covered by the certificate presented by the server. This prevents a /// connection from being intercepted by an attacker with a valid certificate for some other @@ -115,6 +116,28 @@ impl Endpoint { task.wake(); } } + + /// Wait for all connections on the endpoint to be cleanly shut down + /// + /// Waiting for this condition before terminating the endpoint improves the odds that peers are + /// notified about recently closed connection, which is preferred to making them wait for an + /// idle timeout. + /// + /// Does not proactively close existing connections or cause incoming connections to be + /// rejected. Consider calling `Endpoint::close` and dropping the `Incoming` stream if that is + /// desired. + pub async fn wait_idle(&self) { + let mut state = broadcast::State::default(); + futures::future::poll_fn(|cx| { + let endpoint = &mut *self.inner.lock().unwrap(); + if endpoint.connections.is_empty() { + return Poll::Ready(()); + } + endpoint.idle.register(cx, &mut state); + Poll::Pending + }) + .await; + } } /// A future that drives IO on an endpoint @@ -129,7 +152,7 @@ impl Endpoint { /// have been dropped, or when an I/O error occurs. #[must_use = "endpoint drivers must be spawned for I/O to occur"] #[derive(Debug)] -pub struct EndpointDriver(pub(crate) EndpointRef); +pub(crate) struct EndpointDriver(pub(crate) EndpointRef); impl Future for EndpointDriver { type Output = Result<(), io::Error>; @@ -191,6 +214,7 @@ pub(crate) struct EndpointInner { close: Option<(VarInt, Bytes)>, driver_lost: bool, recv_buf: Box<[u8]>, + idle: Broadcast, } impl EndpointInner { @@ -247,7 +271,7 @@ impl EndpointInner { fn drive_incoming(&mut self, cx: &mut Context) { for i in (0..self.incoming.len()).rev() { match self.incoming[i].poll_unpin(cx) { - Poll::Ready(Ok(())) | Poll::Ready(Err(_)) if !self.incoming_live => { + Poll::Ready(()) if !self.incoming_live => { let _ = self.incoming.swap_remove_back(i); } // It's safe to poll an already dead connection @@ -297,6 +321,9 @@ impl EndpointInner { Proto(e) => { if e.is_drained() { self.connections.remove(&ch); + if self.connections.is_empty() { + self.idle.wake(); + } } if let Some(event) = self.inner.handle_event(ch, e) { // Ignoring errors from dropped connections that haven't yet been cleaned up @@ -404,6 +431,7 @@ impl EndpointRef { close: None, driver_lost: false, recv_buf: vec![0; 64 * 1024].into(), + idle: Broadcast::new(), }))) } } diff --git a/quinn/src/lib.rs b/quinn/src/lib.rs index 5cff34b24..a1e9a95b9 100644 --- a/quinn/src/lib.rs +++ b/quinn/src/lib.rs @@ -8,14 +8,13 @@ //! //! The entry point of this crate is the [`Endpoint`](struct.Endpoint.html). //! -//! ``` +//! ```no_run //! # use futures::TryFutureExt; -//! let mut runtime = tokio::runtime::Builder::new().basic_scheduler().enable_all().build().unwrap(); //! let mut builder = quinn::Endpoint::builder(); -//! // -//! let (endpoint_driver, endpoint, _) = runtime.enter(|| builder.bind(&"[::]:0".parse().unwrap()).unwrap()); -//! runtime.spawn(endpoint_driver.unwrap_or_else(|e| panic!("I/O error: {}", e))); -//! // +//! // ... configure builder ... +//! // Ensure you're inside a tokio runtime context +//! let (endpoint, _) = builder.bind(&"[::]:0".parse().unwrap()).unwrap(); +//! // ... use endpoint ... //! ``` //! # About QUIC //! @@ -64,12 +63,12 @@ pub use crate::builders::{ mod connection; pub use connection::{ - Connecting, Connection, ConnectionDriver, Datagrams, IncomingBiStreams, IncomingUniStreams, - NewConnection, OpenBi, OpenUni, ZeroRttAccepted, + Connecting, Connection, Datagrams, IncomingBiStreams, IncomingUniStreams, NewConnection, + OpenBi, OpenUni, ZeroRttAccepted, }; mod endpoint; -pub use endpoint::{Endpoint, EndpointDriver, Incoming}; +pub use endpoint::{Endpoint, Incoming}; mod streams; pub use streams::{ diff --git a/quinn/src/tests.rs b/quinn/src/tests.rs index 1f0cf9153..2d269fa23 100644 --- a/quinn/src/tests.rs +++ b/quinn/src/tests.rs @@ -5,7 +5,7 @@ use std::{ sync::Arc, }; -use futures::{future, FutureExt, StreamExt, TryFutureExt}; +use futures::{future, FutureExt, StreamExt}; use tokio::{ runtime::{Builder, Runtime}, time::{Duration, Instant}, @@ -14,7 +14,7 @@ use tracing::{info, info_span}; use tracing_futures::Instrument as _; use super::{ - ClientConfigBuilder, Endpoint, EndpointDriver, Incoming, NewConnection, RecvStream, SendStream, + ClientConfigBuilder, Endpoint, Incoming, NewConnection, RecvStream, SendStream, ServerConfigBuilder, }; @@ -22,14 +22,12 @@ use super::{ fn handshake_timeout() { let _guard = subscribe(); let mut runtime = rt_threaded(); - let (client_driver, client, _) = runtime.enter(|| { + let (client, _) = runtime.enter(|| { Endpoint::builder() .bind(&SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)) .unwrap() }); - runtime.spawn(client_driver.unwrap_or_else(|e| panic!("client endpoint driver failed: {}", e))); - let mut client_config = crate::ClientConfig::default(); const IDLE_TIMEOUT: Duration = Duration::from_millis(500); let mut transport_config = crate::TransportConfig::default(); @@ -59,64 +57,12 @@ fn handshake_timeout() { assert!(dt > IDLE_TIMEOUT && dt < 2 * IDLE_TIMEOUT); } -#[test] -fn drop_endpoint() { - let _guard = subscribe(); - let mut runtime = rt_basic(); - let (driver, endpoint, _) = runtime.enter(|| { - Endpoint::builder() - .bind(&SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)) - .unwrap() - }); - - let handle = runtime.spawn( - endpoint - .connect( - &SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1234), - "localhost", - ) - .unwrap() - .map(|x| match x { - Err(crate::ConnectionError::TransportError(proto::TransportError { - code: proto::TransportErrorCode::INTERNAL_ERROR, - .. - })) => {} - Err(e) => panic!("unexpected error: {}", e), - Ok(_) => { - panic!("unexpected success"); - } - }), - ); - - drop((driver, endpoint)); - runtime.block_on(handle).unwrap(); -} - -#[test] -fn drop_endpoint_driver() { - let _guard = subscribe(); - let endpoint = Endpoint::builder(); - let runtime = rt_basic(); - let (_, endpoint, _) = runtime.enter(|| { - endpoint - .bind(&SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)) - .unwrap() - }); - - assert!(endpoint - .connect( - &SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1234), - "localhost", - ) - .is_err()); -} - #[test] fn close_endpoint() { let _guard = subscribe(); let endpoint = Endpoint::builder(); let mut runtime = rt_basic(); - let (_driver, endpoint, incoming) = runtime.enter(|| { + let (endpoint, incoming) = runtime.enter(|| { endpoint .bind(&SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)) .unwrap() @@ -152,7 +98,7 @@ fn local_addr() { let socket = UdpSocket::bind("[::1]:0").unwrap(); let addr = socket.local_addr().unwrap(); let runtime = rt_basic(); - let (_, ep, _) = runtime.enter(|| Endpoint::builder().with_socket(socket).unwrap()); + let (ep, _) = runtime.enter(|| Endpoint::builder().with_socket(socket).unwrap()); assert_eq!( addr, ep.local_addr() @@ -164,8 +110,7 @@ fn local_addr() { fn read_after_close() { let _guard = subscribe(); let mut runtime = rt_basic(); - let (driver, endpoint, mut incoming) = runtime.enter(endpoint); - runtime.spawn(driver.unwrap_or_else(|e| panic!("{}", e))); + let (endpoint, mut incoming) = runtime.enter(endpoint); const MSG: &[u8] = b"goodbye!"; runtime.spawn(async move { let new_conn = incoming @@ -174,7 +119,6 @@ fn read_after_close() { .expect("endpoint") .await .expect("connection"); - tokio::spawn(new_conn.driver.unwrap_or_else(|_| ())); let mut s = new_conn.connection.open_uni().await.unwrap(); s.write_all(MSG).await.unwrap(); s.finish().await.unwrap(); @@ -185,7 +129,6 @@ fn read_after_close() { .unwrap() .await .expect("connect"); - tokio::spawn(new_conn.driver.unwrap_or_else(|_| ())); tokio::time::delay_until(Instant::now() + Duration::from_millis(100)).await; let stream = new_conn .uni_streams @@ -202,7 +145,7 @@ fn read_after_close() { } /// Construct an endpoint suitable for connecting to itself -fn endpoint() -> (EndpointDriver, Endpoint, Incoming) { +fn endpoint() -> (Endpoint, Incoming) { let mut endpoint = Endpoint::builder(); let mut server_config = ServerConfigBuilder::default(); @@ -217,29 +160,25 @@ fn endpoint() -> (EndpointDriver, Endpoint, Incoming) { client_config.add_certificate_authority(cert).unwrap(); endpoint.default_client_config(client_config.build()); - let (x, y, z) = endpoint + let (x, y) = endpoint .bind(&SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)) .unwrap(); - (x, y, z) + (x, y) } -#[test] -fn zero_rtt() { +#[tokio::test] +async fn zero_rtt() { let _guard = subscribe(); - let mut runtime = rt_basic(); - let (driver, endpoint, incoming) = runtime.enter(endpoint); + let (endpoint, incoming) = endpoint(); - runtime.spawn(driver.unwrap_or_else(|e| panic!("{}", e))); const MSG: &[u8] = b"goodbye!"; - runtime.spawn(incoming.take(2).for_each(|incoming| { + tokio::spawn(incoming.take(2).for_each(|incoming| { async { let NewConnection { - driver, mut uni_streams, connection, .. } = incoming.into_0rtt().unwrap_or_else(|_| unreachable!()).0; - tokio::spawn(driver.unwrap_or_else(|_| ())); tokio::spawn(async move { while let Some(Ok(x)) = uni_streams.next().await { let msg = x.read_to_end(usize::max_value()).await.unwrap(); @@ -251,59 +190,21 @@ fn zero_rtt() { s.finish().await.expect("finish"); } })); - runtime.block_on(async { - let NewConnection { - driver, - mut uni_streams, - .. - } = endpoint - .connect(&endpoint.local_addr().unwrap(), "localhost") - .unwrap() - .into_0rtt() - .err() - .expect("0-RTT succeeded without keys") - .await - .expect("connect"); - tokio::spawn(async move { - // Buy time for the driver to process the server's NewSessionTicket - tokio::time::delay_until(Instant::now() + Duration::from_millis(100)).await; - let stream = uni_streams - .next() - .await - .expect("incoming streams") - .expect("missing stream"); - let msg = stream - .read_to_end(usize::max_value()) - .await - .expect("read_to_end"); - assert_eq!(msg, MSG); - }); - driver.unwrap_or_else(|_| ()).await - }); - info!("initial connection complete"); - let ( - NewConnection { - connection, - driver, - mut uni_streams, - .. - }, - zero_rtt, - ) = endpoint + let NewConnection { + mut uni_streams, .. + } = endpoint .connect(&endpoint.local_addr().unwrap(), "localhost") .unwrap() .into_0rtt() - .ok() - .expect("missing 0-RTT keys"); - // Send something before the driver starts to ensure it's 0-RTT - runtime.spawn(async move { - let mut s = connection.open_uni().await.expect("0-RTT open uni"); - s.write_all(MSG).await.expect("0-RTT write"); - s.finish().await.expect("0-RTT finish"); - }); - let handle = runtime.spawn(driver.unwrap_or_else(|_| ())); - runtime.block_on(async move { + .err() + .expect("0-RTT succeeded without keys") + .await + .expect("connect"); + + tokio::spawn(async move { + // Buy time for the driver to process the server's NewSessionTicket + tokio::time::delay_until(Instant::now() + Duration::from_millis(100)).await; let stream = uni_streams .next() .await @@ -314,13 +215,45 @@ fn zero_rtt() { .await .expect("read_to_end"); assert_eq!(msg, MSG); - assert_eq!(zero_rtt.await, true); + }); + endpoint.wait_idle().await; + + info!("initial connection complete"); + + let ( + NewConnection { + connection, + mut uni_streams, + .. + }, + zero_rtt, + ) = endpoint + .connect(&endpoint.local_addr().unwrap(), "localhost") + .unwrap() + .into_0rtt() + .expect("missing 0-RTT keys"); + // Send something ASAP to use 0-RTT + tokio::spawn(async move { + let mut s = connection.open_uni().await.expect("0-RTT open uni"); + s.write_all(MSG).await.expect("0-RTT write"); + s.finish().await.expect("0-RTT finish"); }); - // The endpoint driver won't finish if we could still create new connections - drop(endpoint); + let stream = uni_streams + .next() + .await + .expect("incoming streams") + .expect("missing stream"); + let msg = stream + .read_to_end(usize::max_value()) + .await + .expect("read_to_end"); + assert_eq!(msg, MSG); + assert_eq!(zero_rtt.await, true); - runtime.block_on(handle).unwrap(); + drop(uni_streams); + + endpoint.wait_idle().await; } #[test] @@ -365,7 +298,7 @@ fn run_echo(client_addr: SocketAddr, server_addr: SocketAddr) { server.listen(server_config.build()); let server_sock = UdpSocket::bind(server_addr).unwrap(); let server_addr = server_sock.local_addr().unwrap(); - let (server_driver, _, mut server_incoming) = + let (server, mut server_incoming) = runtime.enter(|| server.with_socket(server_sock).unwrap()); let mut client_config = ClientConfigBuilder::default(); @@ -373,39 +306,19 @@ fn run_echo(client_addr: SocketAddr, server_addr: SocketAddr) { client_config.enable_keylog(); let mut client = Endpoint::builder(); client.default_client_config(client_config.build()); - let (client_driver, client, _) = runtime.enter(|| client.bind(&client_addr).unwrap()); + let (client, _) = runtime.enter(|| client.bind(&client_addr).unwrap()); - let handle = runtime.spawn( - server_driver - .unwrap_or_else(|e| panic!("server driver failed: {}", e)) - .instrument(info_span!("server endpoint")), - ); - let handle = future::join( - handle, - runtime.spawn( - client_driver - .unwrap_or_else(|e| panic!("client driver failed: {}", e)) - .instrument(info_span!("client endpoint")), - ), - ); - let handle = future::join( - handle, - runtime.spawn(async move { - let incoming = server_incoming.next().await.unwrap(); - let new_conn = incoming.instrument(info_span!("server")).await.unwrap(); - tokio::spawn( - new_conn - .bi_streams - .take_while(|x| future::ready(x.is_ok())) - .for_each(|s| echo(s.unwrap())), - ); + let handle = runtime.spawn(async move { + let incoming = server_incoming.next().await.unwrap(); + let new_conn = incoming.instrument(info_span!("server")).await.unwrap(); + tokio::spawn( new_conn - .driver - .unwrap_or_else(|_| ()) - .instrument(info_span!("server")) - .await - }), - ); + .bi_streams + .take_while(|x| future::ready(x.is_ok())) + .for_each(|s| echo(s.unwrap())), + ); + server.wait_idle().await; + }); info!("connecting from {} to {}", client_addr, server_addr); runtime.block_on(async move { @@ -415,26 +328,17 @@ fn run_echo(client_addr: SocketAddr, server_addr: SocketAddr) { .instrument(info_span!("client")) .await .expect("connect"); - let driver = tokio::spawn( - new_conn - .driver - .unwrap_or_else(|e| eprintln!("outgoing connection lost: {}", e)) - .instrument(info_span!("client")), - ); let (mut send, recv) = new_conn.connection.open_bi().await.expect("stream open"); send.write_all(b"foo").await.expect("write"); send.finish().await.expect("finish"); let data = recv.read_to_end(usize::max_value()).await.expect("read"); assert_eq!(&data[..], b"foo"); new_conn.connection.close(0u32.into(), b"done"); - driver.await.unwrap(); + client.wait_idle().await; }); handle }; - let ((r1, r2), r3) = runtime.block_on(handle); - r1.unwrap(); - r2.unwrap(); - r3.unwrap(); + runtime.block_on(handle).unwrap(); } async fn echo((mut send, recv): (SendStream, RecvStream)) { diff --git a/quinn/tests/many_connections.rs b/quinn/tests/many_connections.rs index 445073db3..cd6c978ff 100644 --- a/quinn/tests/many_connections.rs +++ b/quinn/tests/many_connections.rs @@ -30,9 +30,8 @@ fn connect_n_nodes_to_1_and_send_1mb_data() { let (cfg, listener_cert) = configure_listener(); let mut ep_builder = quinn::Endpoint::builder(); ep_builder.listen(cfg); - let (driver, endpoint, incoming_conns) = + let (endpoint, incoming_conns) = unwrap!(runtime.enter(|| ep_builder.bind(&"127.0.0.1:0".parse().unwrap()))); - runtime.spawn(driver.unwrap_or_else(|e| panic!("Listener IO error: {}", e))); let listener_addr = unwrap!(endpoint.local_addr()); let expected_messages = 50; @@ -43,7 +42,6 @@ fn connect_n_nodes_to_1_and_send_1mb_data() { .take(expected_messages) .for_each(move |new_conn| { let conn = new_conn.connection; - tokio::spawn(new_conn.driver.unwrap_or_else(|_| ())); let shared = shared2.clone(); let task = new_conn @@ -62,7 +60,7 @@ fn connect_n_nodes_to_1_and_send_1mb_data() { future::ready(()) }); - let handle = runtime.spawn(read_incoming_data); + runtime.spawn(read_incoming_data); let client_cfg = configure_connector(&listener_cert); @@ -70,34 +68,25 @@ fn connect_n_nodes_to_1_and_send_1mb_data() { let data = random_data_with_hash(1024 * 1024); let shared = shared.clone(); let task = unwrap!(endpoint.connect_with(client_cfg.clone(), &listener_addr, "localhost")) - .and_then(move |new_conn| { - tokio::spawn( - write_to_peer(new_conn.connection, data).unwrap_or_else(move |e| { - // Error will also be propagated to the driver - eprintln!("write failed: {}", e); - }), - ); - new_conn.driver - }) + .map_err(WriteError::ConnectionClosed) + .and_then(move |new_conn| write_to_peer(new_conn.connection, data)) .unwrap_or_else(move |e| { + use quinn::ConnectionError::*; match e { - quinn::ConnectionError::ApplicationClosed { .. } - | quinn::ConnectionError::Reset => {} - // TODO: Determine why packet loss during connection close leads to this timing out - // even though valid stateless reset packets are sent. - _ => match e { - quinn::ConnectionError::TimedOut => {} - _ => shared.lock().unwrap().errors.push(e), + WriteError::ConnectionClosed(ApplicationClosed { .. }) + | WriteError::ConnectionClosed(Reset) => {} + WriteError::ConnectionClosed(e) => match e { + _ => { + shared.lock().unwrap().errors.push(e); + } }, + _ => panic!("unexpected write error"), } }); runtime.spawn(task); } - // we don't need it anymore, this will make EndpointDriver finish after all connections are - // finished. - drop(endpoint); - unwrap!(runtime.block_on(handle)); + runtime.block_on(endpoint.wait_idle()); let shared = shared.lock().unwrap(); if !shared.errors.is_empty() { panic!("some connections failed: {:?}", shared.errors);