From 25d60cca9f23cc21eebc2ff4e952bc1483f2fc16 Mon Sep 17 00:00:00 2001 From: Benjamin Saunders Date: Sat, 30 Mar 2019 11:37:54 -0700 Subject: [PATCH] Refactor close behavior 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. --- interop/src/main.rs | 98 ++++++++++++++++----------------- quinn-proto/src/connection.rs | 7 +++ quinn/examples/client.rs | 4 +- quinn/examples/server.rs | 3 +- quinn/src/connection.rs | 101 ++++++++++++---------------------- quinn/src/tests.rs | 4 +- 6 files changed, 97 insertions(+), 120 deletions(-) diff --git a/interop/src/main.rs b/interop/src/main.rs index 229e21953..a0bc27a58 100644 --- a/interop/src/main.rs +++ b/interop/src/main.rs @@ -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) }), diff --git a/quinn-proto/src/connection.rs b/quinn-proto/src/connection.rs index b30c4cce0..f9cf5eae0 100644 --- a/quinn-proto/src/connection.rs +++ b/quinn-proto/src/connection.rs @@ -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 for ConnectionError { @@ -2931,6 +2937,7 @@ impl From 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"), } } } diff --git a/quinn/examples/client.rs b/quinn/examples/client.rs index 0d4fc9baf..c0f84302b 100644 --- a/quinn/examples/client.rs +++ b/quinn/examples/client.rs @@ -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")) }), diff --git a/quinn/examples/server.rs b/quinn/examples/server.rs index de958612b..e2d521746 100644 --- a/quinn/examples/server.rs +++ b/quinn/examples/server.rs @@ -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( diff --git a/quinn/src/connection.rs b/quinn/src/connection.rs index 067751fbd..56dc162e0 100644 --- a/quinn/src/connection.rs +++ b/quinn/src/connection.rs @@ -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 { 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 { - 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, 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, 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>>, incoming_streams_reader: Option, finishing: FnvHashMap>>, + /// Always set to Some before the connection becomes drained error: Option, - closing: Option>, } 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 { diff --git a/quinn/src/tests.rs b/quinn/src/tests.rs index 64b4f72ed..c9526c0c0 100644 --- a/quinn/src/tests.rs +++ b/quinn/src/tests.rs @@ -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"); }) }), )