diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs index 30e513fa4a..3a3e4d27f0 100644 --- a/src/proto/h1/dispatch.rs +++ b/src/proto/h1/dispatch.rs @@ -196,18 +196,15 @@ where return Poll::Ready(Ok(())); } - // If we are continuing only because "wants_write_again", check if write is ready. + // If we are continuing only because "wants_write_again", re-check whether a second + // write poll can make progress. `poll_flush` can be ready even when there is no + // buffered data and the request body is still pending, so relying on the previous + // readiness can hot-loop. if !wants_read_again && wants_write_again { - // If write was ready, just proceed with the loop - if write_ready { - continue; - } // Write was previously pending, but may have become ready since polling flush, so - // we need to check it again. If we simply proceeded, the case of an unbuffered - // writer where flush is always ready would cause us to hot loop. + // we need to check it again. If it is still pending, it is safe to yield and rely + // on wake-up from the connection futures. if self.poll_write(cx)?.is_pending() { - // write is pending, so it is safe to yield and rely on wake-up from connection - // futures. return Poll::Ready(Ok(())); } } @@ -457,10 +454,10 @@ where self.conn.close_write(); } - /// If there is pending data in `body_rx`, we can make progress writing if the connection is - /// ready. + /// If there is pending data in `body_rx`, and the connection is still in a body-writing state, + /// we can make progress writing if the connection is ready. fn can_write_again(&mut self) -> bool { - self.body_rx.is_some() + !self.is_closing && self.body_rx.is_some() && self.conn.can_write_body() } fn is_done(&self) -> bool { diff --git a/tests/client.rs b/tests/client.rs index 864bf53725..f46e042977 100644 --- a/tests/client.rs +++ b/tests/client.rs @@ -1563,6 +1563,10 @@ mod conn { use std::io::{self, Read, Write}; use std::net::{SocketAddr, TcpListener}; use std::pin::Pin; + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; use std::task::{Context, Poll}; use std::thread; use std::time::Duration; @@ -3006,6 +3010,105 @@ mod conn { } } + struct CountingStream { + tcp: TokioIo, + flush_count: Arc, + } + + impl hyper::rt::Write for CountingStream { + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + Pin::new(&mut self.tcp).poll_shutdown(cx) + } + + fn poll_flush( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + self.flush_count.fetch_add(1, Ordering::Relaxed); + Pin::new(&mut self.tcp).poll_flush(cx) + } + + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.tcp).poll_write(cx, buf) + } + } + + impl hyper::rt::Read for CountingStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: hyper::rt::ReadBufCursor<'_>, + ) -> Poll> { + Pin::new(&mut self.tcp).poll_read(cx, buf) + } + } + + // https://github.com/hyperium/hyper/issues/4085 + #[tokio::test] + async fn http1_half_closed_peer_with_open_request_body_does_not_spin() { + let (listener, addr) = setup_tk_test_server().await; + let (headers_seen_tx, headers_seen_rx) = oneshot::channel(); + + tokio::spawn(async move { + let mut sock = listener.accept().await.unwrap().0; + let mut buf = [0; 1024]; + let mut received = Vec::new(); + + loop { + let n = sock.read(&mut buf).await.expect("server read request"); + assert_ne!(n, 0, "client closed before sending request headers"); + received.extend_from_slice(&buf[..n]); + + if received.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + + headers_seen_tx.send(()).unwrap(); + sock.shutdown().await.expect("server half-close write"); + tokio::time::sleep(Duration::from_millis(200)).await; + }); + + let flush_count = Arc::new(AtomicUsize::new(0)); + let io = CountingStream { + tcp: tcp_connect(&addr).await.unwrap(), + flush_count: flush_count.clone(), + }; + let (mut client, conn) = conn::http1::Builder::new() + .handshake::<_, StreamBody, io::Error>>>>(io) + .await + .expect("handshake"); + + let conn_task = tokio::spawn(async move { + let _ = conn.await; + }); + + let (_tx, rx) = mpsc::channel::, io::Error>>(0); + let req = Request::post("/a").body(StreamBody::new(rx)).unwrap(); + let response_task = tokio::spawn(async move { + let _ = client.send_request(req).await; + }); + + headers_seen_rx.await.unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + + let flushes = flush_count.load(Ordering::Relaxed); + assert!( + flushes < 100, + "client spun after peer half-close with open request body: poll_flush={flushes}", + ); + + response_task.abort(); + conn_task.abort(); + } + // https://github.com/hyperium/hyper/issues/4040 #[tokio::test] async fn h2_pipe_task_cancelled_on_response_future_drop() {