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

Refactor close behavior

Browse Source 
Closing the connection is now considered an instantaneous operation,
which causes outstanding I/O futures to fail immediately with a
distinct error and gracefully terminates IncomingStreams and, one
draining has completed, ConnectionDriver.
This commit is contained in:
Benjamin Saunders
2019-03-30 11:37:54 -07:00
parent 83da359acc
commit 25d60cca9f
6 changed files with 97 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
interop/src/main.rs
+49 -49
View File
@@ -109,55 +109,55 @@ fn run(log: Logger, options: Opt) -> Result<()> {
let conn = conn.connection;
let stream = conn.open_bi();
let stream_data = &mut stream_data;
stream
.map_err(|e| format_err!("failed to open stream: {}", e))
.and_then(move |stream| get(stream))
.and_then(move |data| {
.map(move |data| {
println!("read {} bytes, closing", data.len());
*stream_data = true;
conn.close(0, b"done").map_err(|_| unreachable!())
conn.close(0, b"done");
})
.map(|()| {
close = true;
})
})
.and_then(|_| {
println!("attempting resumption");
state.lock().unwrap().saw_cert = false;
endpoint
.connect_with(&client_config, &remote, host)
.unwrap()
.map_err(|e| format_err!("failed to connect: {}", e))
.and_then(|conn| {
tokio_current_thread::spawn(
conn.driver.map_err(|e| eprintln!("connection lost: {}", e)),
);
resumption = !state.lock().unwrap().saw_cert;
let conn = conn.connection;
conn.force_key_update();
let stream = conn.open_bi();
let stream2 = conn.open_bi();
let rebinding = &mut rebinding;
stream
.map_err(|e| format_err!("failed to open stream: {}", e))
.and_then(move |stream| get(stream))
.inspect(|_| {
key_update = true;
})
.and_then(move |_| {
let socket = std::net::UdpSocket::bind("[::]:0").unwrap();
let addr = socket.local_addr().unwrap();
println!("rebinding to {}", addr);
endpoint
.rebind(socket, &tokio_reactor::Handle::default())
.expect("rebind failed");
stream2
.and_then(|_| {
println!("attempting resumption");
state.lock().unwrap().saw_cert = false;
endpoint
.connect_with(&client_config, &remote, host)
.unwrap()
.map_err(|e| format_err!("failed to connect: {}", e))
.and_then(|conn| {
tokio_current_thread::spawn(
conn.driver.map_err(|e| eprintln!("connection lost: {}", e)),
);
resumption = !state.lock().unwrap().saw_cert;
let conn = conn.connection;
conn.force_key_update();
let stream = conn.open_bi();
let stream2 = conn.open_bi();
let rebinding = &mut rebinding;
stream
.map_err(|e| format_err!("failed to open stream: {}", e))
.and_then(move |stream| get(stream))
})
.and_then(move |_| {
*rebinding = true;
conn.close(0, b"done").map_err(|_| unreachable!())
.inspect(|_| {
key_update = true;
})
.and_then(move |_| {
let socket = std::net::UdpSocket::bind("[::]:0").unwrap();
let addr = socket.local_addr().unwrap();
println!("rebinding to {}", addr);
endpoint
.rebind(socket, &tokio_reactor::Handle::default())
.expect("rebind failed");
stream2
.map_err(|e| {
format_err!("failed to open stream: {}", e)
})
.and_then(move |stream| get(stream))
})
.map(move |_| {
*rebinding = true;
conn.close(0, b"done");
})
})
})
}),
@@ -177,13 +177,12 @@ fn run(log: Logger, options: Opt) -> Result<()> {
endpoint
.connect_with(&client_config, &remote, host)?
.and_then(|conn| {
tokio_current_thread::spawn(
conn.driver.map_err(|e| eprintln!("connection lost: {}", e)),
);
retry = true;
conn.connection
.close(0, b"done")
.map_err(|_| unreachable!())
conn.connection.close(0, b"done");
conn.driver
})
.map(|()| {
close = true;
}),
);
if let Err(e) = result {
@@ -203,6 +202,7 @@ fn run(log: Logger, options: Opt) -> Result<()> {
};
let mut h3 = false;
println!("trying h3");
let result = runtime.block_on(
endpoint
.connect_with(&h3_client_config, &remote, host)?
@@ -232,13 +232,13 @@ fn run(log: Logger, options: Opt) -> Result<()> {
let req_fut = req_stream
.map_err(|e| format_err!("failed to open request stream: {}", e))
.and_then(|req_stream| h3_get(req_stream))
.and_then(move |data| {
.map(move |data| {
println!(
"read {} bytes: \n\n{}\n\n closing",
data.len(),
String::from_utf8_lossy(&data)
);
conn.close(0, b"done").map_err(|_| unreachable!())
conn.close(0, b"done");
});
control_fut.and_then(|_| req_fut).map(|_| h3 = true)
}),
quinn-proto/src/connection.rs
+7
View File
@@ -1170,6 +1170,9 @@ impl Connection {
State::closed(err)
}
ConnectionError::VersionMismatch => State::Draining,
ConnectionError::LocallyClosed => {
unreachable!("LocallyClosed isn't generated by packet processing")
}
};
}
@@ -2906,6 +2909,9 @@ pub enum ConnectionError {
/// The peer has become unreachable.
#[error(display = "timed out")]
TimedOut,
/// The local application closed the connection.
#[error(display = "closed")]
LocallyClosed,
}
impl From<TransportError> for ConnectionError {
@@ -2931,6 +2937,7 @@ impl From<ConnectionError> for io::Error {
),
TransportError(x) => io::Error::new(io::ErrorKind::Other, format!("{}", x)),
VersionMismatch => io::Error::new(io::ErrorKind::Other, "version mismatch"),
LocallyClosed => io::Error::new(io::ErrorKind::Other, "locally closed"),
}
}
}
quinn/examples/client.rs
+2 -2
View File
@@ -142,7 +142,7 @@ fn run(log: Logger, options: Opt) -> Result<()> {
.map_err(|e| format_err!("failed to read response: {}", e))
.map(move |x| (x, response_start))
})
.and_then(move |((_, data), response_start)| {
.map(move |((_, data), response_start)| {
let duration = response_start.elapsed();
eprintln!(
"response received in {:?} - {} KiB/s",
@@ -151,7 +151,7 @@ fn run(log: Logger, options: Opt) -> Result<()> {
);
io::stdout().write_all(&data).unwrap();
io::stdout().flush().unwrap();
conn.close(0, b"done").map_err(|_| unreachable!())
conn.close(0, b"done");
})
.map(|()| eprintln!("drained"))
}),
quinn/examples/server.rs
+2 -1
View File
@@ -175,7 +175,8 @@ fn handle_connection(root: &PathBuf, log: &Logger, conn: quinn::NewConnection) {
let log2 = log.clone();
let root = root.clone();
tokio_current_thread::spawn(driver.map_err(|e| panic!(e)));
// We ignore errors from the driver because they'll be reported by the `incoming` handler anyway.
tokio_current_thread::spawn(driver.map_err(|_| ()));
// Each stream initiated by the client constitutes a new request.
tokio_current_thread::spawn(
quinn/src/connection.rs
+35 -66
View File
@@ -87,8 +87,10 @@ impl NewConnection {
/// `Connection` API object to the `Endpoint` task and the related stream-related interfaces.
/// It also keeps track of outstanding timeouts for the `Connection`.
///
/// If the connection encounters an error condition, this future will yield an error. It
/// will terminate (yielding `Ready(())`) if the connection was closed without error.
/// If the connection encounters an error condition, this future will yield an error. It will
/// terminate (yielding `Ok(())`) if the connection was closed without error. Unlike other
/// connection-related futures, this waits for the draining period to complete to ensure that
/// packets still in flight from the peer are handled gracefully.
#[must_use = "connection drivers must be spawned for their connections to function"]
pub struct ConnectionDriver(pub(crate) ConnectionRef);
@@ -98,10 +100,6 @@ impl Future for ConnectionDriver {
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
let conn = &mut *self.0.lock().unwrap();
if conn.driver.is_none() {
conn.driver = Some(task::current());
}
let now = Instant::now();
loop {
let mut keep_going = false;
@@ -111,17 +109,19 @@ impl Future for ConnectionDriver {
keep_going |= conn.handle_timer_updates();
conn.forward_endpoint_events();
conn.forward_app_events();
if !keep_going || conn.closed {
if !keep_going || conn.inner.is_drained() {
break;
}
}
if !conn.inner.is_drained() {
Ok(Async::NotReady)
} else if let Some(e) = &conn.pending.error {
Err(e.clone())
} else {
Ok(Async::Ready(()))
conn.driver = Some(task::current());
return Ok(Async::NotReady);
}
match conn.pending.error {
Some(ConnectionError::LocallyClosed) => Ok(Async::Ready(())),
Some(ref e) => Err(e.clone()),
None => unreachable!("drained connections always have an error"),
}
}
}
@@ -176,8 +176,9 @@ impl Connection {
/// Close the connection immediately.
///
/// This does not ensure delivery of outstanding data. It is the application's responsibility
/// to call this only when all important communications have been completed.
/// Pending operations will fail immediately with `ConnectionError::LocallyClosed`. Delivery of
/// data on unfinished streams is not guaranteed, so the application must call this only when
/// all important communications have been completed.
///
/// `error_code` and `reason` are not interpreted, and are provided directly to the peer.
///
@@ -186,24 +187,10 @@ impl Connection {
///
/// # Panics
/// - If called more than once on handles to the same connection
// FIXME: Infallible
pub fn close(&self, error_code: u16, reason: &[u8]) -> impl Future<Item = (), Error = ()> {
let (send, recv) = oneshot::channel();
{
let conn = &mut *self.0.lock().unwrap();
assert!(
conn.pending.closing.is_none(),
"a connection can only be closed once"
);
conn.pending.closing = Some(send);
conn.inner.close(Instant::now(), error_code, reason.into());
}
let handle = self.clone();
recv.then(move |_| {
// Ensure the connection isn't dropped until it's fully drained.
let _ = handle;
Ok(())
})
pub fn close(self, error_code: u16, reason: &[u8]) {
let conn = &mut *self.0.lock().unwrap();
conn.inner.close(Instant::now(), error_code, reason.into());
conn.pending.terminate(ConnectionError::LocallyClosed);
}
/// The peer's UDP address.
@@ -236,20 +223,18 @@ impl FuturesStream for IncomingStreams {
type Error = ConnectionError;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
let mut conn = self.0.lock().unwrap();
if conn.closed {
return Ok(Async::Ready(None));
}
if let Some(x) = conn.inner.accept() {
if let Some(ConnectionError::LocallyClosed) = conn.pending.error {
Ok(Async::Ready(None))
} else if let Some(ref e) = conn.pending.error {
Err(e.clone())
} else if let Some(x) = conn.inner.accept() {
let stream = BiStream::new(self.0.clone(), x);
let stream = if x.directionality() == Directionality::Uni {
NewStream::Uni(RecvStream(stream))
} else {
NewStream::Bi(stream)
};
return Ok(Async::Ready(Some(stream)));
}
if let Some(ref x) = conn.pending.error {
Err(x.clone())
Ok(Async::Ready(Some(stream)))
} else {
conn.pending.incoming_streams_reader = Some(task::current());
Ok(Async::NotReady)
@@ -288,7 +273,6 @@ impl ConnectionRef {
conn_events,
endpoint_events,
connected: false,
closed: false,
})))
}
}
@@ -325,7 +309,6 @@ pub struct ConnectionInner {
conn_events: mpsc::UnboundedReceiver<ConnectionEvent>,
endpoint_events: mpsc::UnboundedSender<(ConnectionHandle, EndpointEvent)>,
connected: bool,
closed: bool,
}
impl ConnectionInner {
@@ -339,15 +322,6 @@ impl ConnectionInner {
fn forward_endpoint_events(&mut self) {
while let Some(event) = self.inner.poll_endpoint_events() {
if let quinn::EndpointEvent::Drained = event {
self.closed = true;
self.pending
.fail(ConnectionError::TransportError(quinn::TransportError {
code: quinn::TransportErrorCode::NO_ERROR,
frame: None,
reason: "connection is closing".to_string(),
}));
}
self.endpoint_events
.unbounded_send((self.handle, EndpointEvent::Proto(event)))
.unwrap();
@@ -361,13 +335,13 @@ impl ConnectionInner {
self.inner.handle_event(event);
}
Ok(Async::Ready(Some(ConnectionEvent::DriverLost))) => {
self.closed = true;
self.pending
.fail(ConnectionError::TransportError(quinn::TransportError {
self.pending.terminate(ConnectionError::TransportError(
quinn::TransportError {
code: quinn::TransportErrorCode::INTERNAL_ERROR,
frame: None,
reason: "driver future was dropped".to_string(),
}));
reason: "endpoint driver future was dropped".to_string(),
},
));
}
Ok(Async::Ready(None)) | Ok(Async::NotReady) => {
return Ok(());
@@ -387,7 +361,7 @@ impl ConnectionInner {
self.connected = true;
}
ConnectionLost { reason } => {
self.pending.fail(reason);
self.pending.terminate(reason);
}
StreamWritable { stream } => {
if let Some(writer) = self.pending.blocked_writers.remove(&stream) {
@@ -435,11 +409,6 @@ impl ConnectionInner {
trace!(self.log, "{timer:?} timeout", timer = timer);
self.inner
.handle_event(quinn::ConnectionEvent::Timer(now, timer));
if timer == quinn::Timer::Close {
if let Some(x) = self.pending.closing.take() {
let _ = x.send(());
}
}
// Timeout call may have queued sends
keep_going = true;
}
@@ -512,8 +481,8 @@ pub struct Pending {
bi_opening: VecDeque<oneshot::Sender<Result<StreamId, ConnectionError>>>,
incoming_streams_reader: Option<Task>,
finishing: FnvHashMap<StreamId, oneshot::Sender<Option<ConnectionError>>>,
/// Always set to Some before the connection becomes drained
error: Option<ConnectionError>,
closing: Option<oneshot::Sender<()>>,
}
impl Pending {
@@ -526,11 +495,11 @@ impl Pending {
incoming_streams_reader: None,
finishing: FnvHashMap::default(),
error: None,
closing: None,
}
}
fn fail(&mut self, reason: ConnectionError) {
/// Used to wake up all blocked futures when the connection becomes closed for any reason
fn terminate(&mut self, reason: ConnectionError) {
self.error = Some(reason.clone());
for (_, writer) in self.blocked_writers.drain() {
writer.notify()
@@ -740,7 +709,7 @@ impl Drop for BiStream {
Directionality::Uni => (ours, !ours),
};
if conn.pending.closing.is_some() || conn.pending.error.is_some() {
if conn.pending.error.is_some() {
return;
}
if send && !self.finished {
quinn/src/tests.rs
+2 -2
View File
@@ -154,9 +154,9 @@ fn run_echo(client_addr: SocketAddr, server_addr: SocketAddr) {
read_to_end(stream, usize::max_value())
.map_err(|e| panic!("read: {}", e))
})
.and_then(move |(_, data)| {
.map(move |(_, data)| {
assert_eq!(&data[..], b"foo");
conn.close(0, b"done").map_err(|_| unreachable!())
conn.close(0, b"done");
})
}),
)