gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-09-19 01:36:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Spawn driver tasks implicitly

Browse Source
This commit is contained in:
Benjamin Saunders
2020-01-05 21:16:07 -08:00
committed by Dirkjan Ochtman
parent debaabeaf3
commit 5774071b48
23 changed files with 699 additions and 583 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bench/src/bulk.rs
+4 -21
View File
@@ -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
interop/src/main.rs
+16 -46
View File
@@ -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<InteropResult> {
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("/"))
quinn-debug
+390
View File
@@ -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
quinn-h3/examples/h3_client.rs
+2 -19
View File
@@ -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(())
quinn-h3/examples/h3_server.rs
+10 -32
View File
@@ -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(())
quinn-h3/src/client.rs
+12 -34
View File
@@ -72,18 +72,15 @@ impl Builder {
}
}
pub fn build(self) -> Result<(quinn::EndpointDriver, Client), quinn::EndpointError> {
pub fn build(self) -> Result<Client, quinn::EndpointError> {
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<Connection, Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
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)))
}
}
quinn-h3/src/connection.rs
+5 -5
View File
@@ -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<Self::Output> {
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(())
}
}
}
quinn-h3/src/server.rs
+7 -15
View File
@@ -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<IncomingRequest, Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
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)))
}
}
quinn/Cargo.toml
+1 -1
View File
@@ -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"
quinn/benches/bench.rs
+11 -35
View File
@@ -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)
}
}
quinn/examples/client.rs
+42 -58
View File
@@ -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(())
}
quinn/examples/common/mod.rs
+8 -11
View File
@@ -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<dyn Error>> {
) -> Result<Endpoint, Box<dyn Error>> {
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<u8>), Box<dyn Error>> {
pub fn make_server_endpoint(bind_addr: SocketAddr) -> Result<(Incoming, Vec<u8>), Box<dyn Error>> {
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.
quinn/examples/connection.rs
+6 -15
View File
@@ -22,9 +22,7 @@ use common::{make_client_endpoint, make_server_endpoint};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
"[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(())
}
quinn/examples/insecure_connection.rs
+6 -15
View File
@@ -24,9 +24,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
/// 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<dyn Error>> {
@@ -45,22 +39,19 @@ async fn run_client(server_addr: SocketAddr) -> Result<(), Box<dyn Error>> {
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(())
}
quinn/examples/server.rs
+13 -23
View File
@@ -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<Path>, conn: quinn::Connecting) -> Result<()> {
let quinn::NewConnection {
driver,
connection,
mut bi_streams,
..
@@ -157,12 +150,9 @@ async fn handle_connection(root: Arc<Path>, conn: quinn::Connecting) -> Result<(
remote = %connection.remote_address(),
protocol = %connection.protocol().map_or_else(|| "<none>".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 {
quinn/examples/single_socket.rs
+15 -22
View File
@@ -41,50 +41,43 @@ async fn main() -> Result<(), Box<dyn Error>> {
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<Vec<u8>, Box<dyn Error>> {
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());
}
quinn/src/broadcast.rs
+1
View File
@@ -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<Waker>,
generation: u64,
quinn/src/builders.rs
+12 -7
View File
@@ -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,
quinn/src/connection.rs
+12 -18
View File
@@ -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<ConnectionDriver>);
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<Self::Output> {
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<Self::Output> {
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"),
}
}
quinn/src/endpoint.rs
+31 -3
View File
@@ -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(),
})))
}
}
quinn/src/lib.rs
+8 -9
View File
@@ -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();
//! // <configure 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)));
//! // <use endpoint>
//! // ... 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::{
quinn/src/tests.rs
+74 -170
View File
@@ -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)) {
quinn/tests/many_connections.rs
+13 -24
View File
@@ -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);