Repository: hyperium/hyper
Pull request #3729 (2024-08-08 20:01:02)

Change graceful_shutdown function behavior.
Needed for implementing tonic max_connection_age_ms .
- https://github.com/hyperium/tonic/issues/1428

Related PRs:
- https://github.com/hyperium/tonic/pull/1865



seanmonstar: Why is this needed? If handshaking is taking a long time, it should still get closed, in my opinion.

Regarding the greater motivation, I believe it should be possible to implement a max age timer without any change to hyper. If any changes _are_ needed, it's best to open an issue to discuss and propose a design, before coding happens.
ionut-slaveanu: > Why is this needed? If handshaking is taking a long time, it should still get closed, in my opinion.
> 
> Regarding the greater motivation, I believe it should be possible to implement a max age timer without any change to hyper. If any changes _are_ needed, it's best to open an issue to discuss and propose a design, before coding happens.

The problem is that the requests send during handshake are lost. This can be tested using ghz tool that I used here as a client:

- https://github.com/hyperium/tonic/pull/1865
And with a little change in examples/src/max_connection_age/server.rs. I changed:
.max_connection_age_grace_ms(30000) to .max_connection_age_grace_ms(5)
These are the results:
./ghz --insecure   --proto /opt/work3/tonic/examples/proto/helloworld/helloworld.proto --call helloworld.Greeter.SayHello  -d '{"name":"Tonic"}' [::1]:50051 -n 100000 --connections=500 --cpus=16 --concurrency=500

Summary:
  Count:	100248
  Total:	2.22 s
  Slowest:	1.15 s
  Fastest:	0.03 ms
  Average:	3.63 ms
  Requests/sec:	45106.59

Response time histogram:
  0.032    [1]     |
  115.077  [23482] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  230.122  [148]   |
  345.167  [9]     |
  460.212  [0]     |
  575.257  [0]     |
  690.302  [0]     |
  805.348  [0]     |
  920.393  [1]     |
  1035.438 [76]    |
  1150.483 [19]    |

Latency distribution:
  10 % in 0.13 ms 
  25 % in 0.29 ms 
  50 % in 0.91 ms 
  75 % in 3.84 ms 
  90 % in 34.24 ms 
  95 % in 62.39 ms 
  99 % in 118.73 ms 

Status code distribution:
  [OK]            23736 responses   
  [Unavailable]   76512 responses   

Error distribution:
  [34]    rpc error: code = Unavailable desc = write tcp [::1]:37928->[::1]:50051: write: broken pipe                                         
  [41]    rpc error: code = Unavailable desc = write tcp [::1]:38566->[::1]:50051: write: broken pipe                                         
  [186]   rpc error: code = Unavailable desc = write tcp [::1]:53256->[::1]:50051: write: broken pipe                                         
  [163]   rpc error: code = Unavailable desc = write tcp [::1]:38008->[::1]:50051: write: broken pipe                                         
  [111]   rpc error: code = Unavailable desc = write tcp [::1]:42796->[::1]:50051: write: broken pipe                                         
  [172]   rpc error: code = Unavailable desc = write tcp [::1]:54204->[::1]:50051: write: broken pipe                                         
  [116]   rpc error: code = Unavailable desc = write tcp [::1]:47742->[::1]:50051: write: broken pipe                                         
  [74]    rpc error: code = Unavailable desc = write tcp [::1]:60426->[::1]:50051: write: broken pipe                                         
  [116]   rpc error: code = Unavailable desc = write tcp [::1]:42470->[::1]:50051: write: broken pipe                                         
  [235]   rpc error: code = Unavailable desc = write tcp [::1]:37904->[::1]:50051: write: broken pipe                                         
  [166]   rpc error: code = Unavailable desc = write tcp [::1]:39546->[::1]:50051: write: broken pipe                                         
  [67]    rpc error: code = Unavailable desc = write tcp [::1]:52276->[::1]:50051: write: broken pipe                                         
  [98]    rpc error: code = Unavailable desc = write tcp [::1]:54342->[::1]:50051: write: broken pipe                                         
  [43]    rpc error: code = Unavailable desc = write tcp [::1]:38538->[::1]:50051: write: broken pipe                                         
  [128]   rpc error: code = Unavailable desc = write tcp [::1]:38110->[::1]:50051: write: broken pipe                                         
  [130]   rpc error: code = Unavailable desc = write tcp [::1]:37894->[::1]:50051: write: broken pipe                                         
  [94]    rpc error: code = Unavailable desc = write tcp [::1]:48438->[::1]:50051: write: broken pipe                                         
  [86]    rpc error: code = Unavailable desc = write tcp [::1]:59682->[::1]:50051: write: broken pipe                                         
  [72]    rpc error: code = Unavailable desc = write tcp [::1]:59316->[::1]:50051: write: broken pipe                                         
  [55]    rpc error: code = Unavailable desc = write tcp [::1]:34356->[::1]:50051: write: broken pipe                                         
  [203]   rpc error: code = Unavailable desc = write tcp [::1]:54196->[::1]:50051: write: broken pipe                                         
  [126]   rpc error: code = Unavailable desc = write tcp [::1]:45892->[::1]:50051: write: broken pipe                                         
  [55]    rpc error: code = Unavailable desc = write tcp [::1]:60098->[::1]:50051: write: broken pipe                                         
  [56]    rpc error: code = Unavailable desc = write tcp [::1]:59884->[::1]:50051: write: broken pipe                                         
  [89]    rpc error: code = Unavailable desc = write tcp [::1]:60014->[::1]:50051: write: broken pipe                                         
  [129]   rpc error: code = Unavailable desc = write tcp [::1]:33936->[::1]:50051: write: broken pipe                                         
  [202]   rpc error: code = Unavailable desc = write tcp [::1]:58490->[::1]:50051: write: broken pipe                                         
  [142]   rpc error: code = Unavailable desc = write tcp [::1]:58424->[::1]:50051: write: broken pipe                                         
  [282]   rpc error: code = Unavailable desc = write tcp [::1]:42666->[::1]:50051: write: broken pipe                                         
  [161]   rpc error: code = Unavailable desc = write tcp [::1]:48188->[::1]:50051: write: broken pipe                                         
  [108]   rpc error: code = Unavailable desc = write tcp [::1]:47336->[::1]:50051: write: broken pipe                                         
  [63]    rpc error: code = Unavailable desc = write tcp [::1]:33534->[::1]:50051: write: broken pipe                                         
  [5]     rpc error: code = Unavailable desc = write tcp [::1]:45998->[::1]:50051: write: broken pipe                                         
  [168]   rpc error: code = Unavailable desc = write tcp [::1]:56898->[::1]:50051: write: broken pipe                                         
  [136]   rpc error: code = Unavailable desc = write tcp [::1]:34526->[::1]:50051: write: broken pipe                                         
  [165]   rpc error: code = Unavailable desc = write tcp [::1]:40292->[::1]:50051: write: broken pipe                                         
  [152]   rpc error: code = Unavailable desc = write tcp [::1]:47172->[::1]:50051: write: broken pipe                                         
  [73]    rpc error: code = Unavailable desc = write tcp [::1]:56186->[::1]:50051: write: broken pipe                                         
  [41]    rpc error: code = Unavailable desc = write tcp [::1]:38942->[::1]:50051: write: broken pipe                                         
  [21]    rpc error: code = Unavailable desc = write tcp [::1]:42604->[::1]:50051: write: broken pipe                                         
  [64]    rpc error: code = Unavailable desc = write tcp [::1]:59666->[::1]:50051: write: broken pipe                                         
  [143]   rpc error: code = Unavailable desc = write tcp [::1]:50442->[::1]:50051: write: broken pipe                                         
  [227]   rpc error: code = Unavailable desc = write tcp [::1]:50780->[::1]:50051: write: broken pipe                                         
  [218]   rpc error: code = Unavailable desc = write tcp [::1]:34108->[::1]:50051: write: broken pipe 
So in this case we have:
Status code distribution:
  [OK]            23736 responses   

only 23736 OK responses, instead of having 100000 OK responses like here:
https://github.com/hyperium/tonic/issues/1428
seanmonstar: The unavailable error is because of graceful shutdown was triggered during the handshake? I'm struggling to make the connection, but perhaps there's a detail I'm not noticing.
ionut-slaveanu: > The unavailable error is because of graceful shutdown was triggered during the handshake? I'm struggling to make the connection, but perhaps there's a detail I'm not noticing.

The unavailable error is not a problem. So, If I have sent 100k requests (-n 100000), I expect to receive 100k OK responses like here:
https://github.com/hyperium/tonic/pull/1865
Status code distribution:
[OK] 100000 responses
[Unavailable] 184 responses

Error distribution:
[91] rpc error: code = Unavailable desc = the connection is draining
[93] rpc error: code = Unavailable desc = transport is closing

We received all 100k responses but some of them (184) got Unavailable error. 
In the failed test I sent the same 100k requests, but received only 24k OK responses:
 Status code distribution:
[OK] 23736 responses
[Unavailable] 76512 responses

Error distribution:
[34] rpc error: code = Unavailable desc = write tcp [::1]:37928->[::1]:50051: write: broken pipe.

So, yes I think this is the problem the graceful shutdown was triggered during the handshake. i that case the client will receive a broken pipe error:

- [34] rpc error: code = Unavailable desc = write tcp [::1]:37928->[::1]:50051: write: broken pipe.

 when trying to write to that stream and the request, response will be lost.
 This may never happen in production with with a proper max_connection_age_ms value. I set it to 5 ms only for stress testing of this feature. So, yes we could use directly the current graceful_shutdown function with any changes in hyper and hyper-utils as you said before, but in this case the client that use max_connection_age_ms configuration option in tonic needs to know to set it to a bigger value like 30000ms (30s) to make sure that the graceful_shutdown does not happen during http2 handshake.

Anyway, I think the best option to fix somehow also the case when the client of tonic that use this library sets  max_connection_age_ms to a small value like 5ms and in that case tonic should not miss any OK responses.
seanmonstar: But this change fixes the problem? It seems this just cancels graceful shutdown if triggered while a connection is in the process of starting.
ionut-slaveanu: > But this change fixes the problem? It seems this just cancels graceful shutdown if triggered while a connection is in the process of starting.
Yes, this is what I'm trying to do, to not shutdown the connections that are in handshaking state. That's because, in this case the responses will be lost. Is there any external method to know that the http2 connection is in handshaking state? I'm thinking to change my tonic PR to something like this:
if connection.state != HANDSHAKING {
   connection .graceful_shutdown()
}

seanmonstar: That doesn't seem to me like what most people would want from this method. Consider: as implemented in this PR, if graceful shutdown is triggered a couple milliseconds after the connection starts, and thus gets ignored because of handshaking, then this connection will stay in an "active" state indefinitely, and then you're much more likely to have active streams that get abruptly interrupted when the final shutdown step just kills everything that hasn't stopped. That can be even worse.

I could imagine a solution here: instead of just closing immediately, a flag could be set that as soon as the handshake is done, _then_ it starts the HTTP/2 graceful shutdown. So, literally all requests will get rejected, but as least the client will _know_ from the `GOAWAY` frame that none of the requests were handled.
ionut-slaveanu: > That doesn't seem to me like what most people would want from this method. Consider: as implemented in this PR, if graceful shutdown is triggered a couple milliseconds after the connection starts, and thus gets ignored because of handshaking, then this connection will stay in an "active" state indefinitely, and then you're much more likely to have active streams that get abruptly interrupted when the final shutdown step just kills everything that hasn't stopped. That can be even worse.
> 
> I could imagine a solution here: instead of just closing immediately, a flag could be set that as soon as the handshake is done, _then_ it starts the HTTP/2 graceful shutdown. So, literally all requests will get rejected, but as least the client will _know_ from the `GOAWAY` frame that none of the requests were handled.

I tried to implement your suggested solution.


ionut-slaveanu: Fixed workflow:
[![CI](https://github.com/ionut-slaveanu/hyper/actions/workflows/CI.yml/badge.svg)](https://github.com/ionut-slaveanu/hyper/actions/workflows/CI.yml)
jeromegn: This would be useful to us as we noticed we're closing connections abruptly if the h2 handshake is in progress and we're trying to gracefully shut down the server.

Patch:
diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml
index 1a362d88..3877ac58 100644
--- a/.github/workflows/CI.yml
+++ b/.github/workflows/CI.yml
@@ -118,6 +118,7 @@ jobs:
       - name: Pin some dependencies
         run: |
           cargo update -p tokio --precise 1.38.1
+          cargo update -p tokio-util --precise 0.7.11
 
       - name: Check
         run: cargo check --features full
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5f8c90a3..c385ad71 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,21 @@
+## v1.5.0 (2024-10-15)
+
+
+#### Bug Fixes
+
+* **http1:**
+  * improve performance of parsing sequentially partial messages (#3764) ([3900a23](https://github.com/hyperium/hyper/commit/3900a2381b96a7e7f608a5e031b3e90ddcdfcd74))
+  * send 'connection: close' when connection is ending (#3725) ([c86a6bcb](https://github.com/hyperium/hyper/commit/c86a6bcb4acb0f92e731ea2e4c1e4a839248a600), closes [#3720](https://github.com/hyperium/hyper/issues/3720))
+  * make `date_header` effective (#3718) ([7de02373](https://github.com/hyperium/hyper/commit/7de02373f5e4ce392587a4d9d7710c6faf9c6165))
+* **http2:** strip content-length header in response to CONNECT requests (#3748) ([67a4a498](https://github.com/hyperium/hyper/commit/67a4a498d8bbdce4e604bc578da4693fb048f83d))
+
+
+#### Features
+
+* **client:** Add HTTP/2 builder options `header_table_size()` and `max_concurrent_streams()`  ([4c84e8c1](https://github.com/hyperium/hyper/commit/4c84e8c1c26a1464221de96b9f39816ce7251a5f))
+* **rt:** add `ReadBufCursor` methods `remaining()` and `put_slice()` (#3700) ([5a13041e](https://github.com/hyperium/hyper/commit/5a13041ed7033c9dab6e2adafd08b6af20cd33fb))
+
+
 ### v1.4.1 (2024-07-09)
 
 
@@ -212,7 +230,7 @@ Be sure to check out the [upgrading guide](https://hyper.rs/guides/1/upgrading).
 
 #### Breaking Changes
 
-* Any IO transport type provided must not implement `hyper::rt::{Read, Write}` instead of
+* Any IO transport type provided must now implement `hyper::rt::{Read, Write}` instead of
   `tokio::io` traits. You can grab a helper type from `hyper-util` to wrap Tokio types, or implement the traits yourself,
   if it's a custom type.
  ([f9f65b7a](https://github.com/hyperium/hyper/commit/f9f65b7aa67fa3ec0267fe015945973726285bc2))
@@ -1600,7 +1618,7 @@ Be sure to check out the [upgrading guide](https://hyper.rs/guides/1/upgrading).
 
 * **client:**
   * check for dead connections in Pool ([44af2738](https://github.com/hyperium/hyper/commit/44af273853f82b81591b813d13627e143a14a6b7), closes [#1429](https://github.com/hyperium/hyper/issues/1429))
-  * error on unsupport 101 responses, ignore other 1xx codes ([22774222](https://github.com/hyperium/hyper/commit/227742221fa7830a14c18becbbc6137d97b57729))
+  * error on unsupported 101 responses, ignore other 1xx codes ([22774222](https://github.com/hyperium/hyper/commit/227742221fa7830a14c18becbbc6137d97b57729))
 * **server:**
   * send 400 responses on parse errors before closing connection ([7cb72d20](https://github.com/hyperium/hyper/commit/7cb72d2019bffbc667b9ad2d8cbc19c1a513fcf7))
   * error if Response code is 1xx ([44c34ce9](https://github.com/hyperium/hyper/commit/44c34ce9adc888916bd67656cc54c35f7908f536))
diff --git a/Cargo.toml b/Cargo.toml
index a1cdd9e7..f569ba22 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "hyper"
-version = "1.4.1"
+version = "1.5.0"
 description = "A fast and correct HTTP library."
 readme = "README.md"
 homepage = "https://hyper.rs"
diff --git a/benches/pipeline.rs b/benches/pipeline.rs
index b79232de..d36f054a 100644
--- a/benches/pipeline.rs
+++ b/benches/pipeline.rs
@@ -76,7 +76,7 @@ fn hello_world_16(b: &mut test::Bencher) {
         tcp.write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
             .unwrap();
         let mut buf = Vec::new();
-        tcp.read_to_end(&mut buf).unwrap()
+        tcp.read_to_end(&mut buf).unwrap() - "connection: close\r\n".len()
     } * PIPELINED_REQUESTS;
 
     let mut tcp = TcpStream::connect(addr).unwrap();
diff --git a/benches/server.rs b/benches/server.rs
index c5424105..6e9d3742 100644
--- a/benches/server.rs
+++ b/benches/server.rs
@@ -72,7 +72,7 @@ macro_rules! bench_server {
             tcp.write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
                 .unwrap();
             let mut buf = Vec::new();
-            tcp.read_to_end(&mut buf).unwrap()
+            tcp.read_to_end(&mut buf).unwrap() - "connection: close\r\n".len()
         };
 
         let mut tcp = TcpStream::connect(addr).unwrap();
diff --git a/examples/hello-http2.rs b/examples/hello-http2.rs
new file mode 100644
index 00000000..54bdef08
--- /dev/null
+++ b/examples/hello-http2.rs
@@ -0,0 +1,81 @@
+#![deny(warnings)]
+
+use std::convert::Infallible;
+use std::net::SocketAddr;
+
+use http_body_util::Full;
+use hyper::body::Bytes;
+use hyper::server::conn::http2;
+use hyper::service::service_fn;
+use hyper::{Request, Response};
+use tokio::net::TcpListener;
+
+// This would normally come from the `hyper-util` crate, but we can't depend
+// on that here because it would be a cyclical dependency.
+#[path = "../benches/support/mod.rs"]
+mod support;
+use support::TokioIo;
+
+// An async function that consumes a request, does nothing with it and returns a
+// response.
+async fn hello(_: Request<hyper::body::Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
+    Ok(Response::new(Full::new(Bytes::from("Hello, World!"))))
+}
+
+#[derive(Clone)]
+// An Executor that uses the tokio runtime.
+pub struct TokioExecutor;
+
+// Implement the `hyper::rt::Executor` trait for `TokioExecutor` so that it can be used to spawn
+// tasks in the hyper runtime.
+// An Executor allows us to manage execution of tasks which can help us improve the efficiency and
+// scalability of the server.
+impl<F> hyper::rt::Executor<F> for TokioExecutor
+where
+    F: std::future::Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    fn execute(&self, fut: F) {
+        tokio::task::spawn(fut);
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    pretty_env_logger::init();
+
+    // This address is localhost
+    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
+
+    // Bind to the port and listen for incoming TCP connections
+    let listener = TcpListener::bind(addr).await?;
+
+    loop {
+        // When an incoming TCP connection is received grab a TCP stream for
+        // client-server communication.
+        //
+        // Note, this is a .await point, this loop will loop forever but is not a busy loop. The
+        // .await point allows the Tokio runtime to pull the task off of the thread until the task
+        // has work to do. In this case, a connection arrives on the port we are listening on and
+        // the task is woken up, at which point the task is then put back on a thread, and is
+        // driven forward by the runtime, eventually yielding a TCP stream.
+        let (stream, _) = listener.accept().await?;
+        // Use an adapter to access something implementing `tokio::io` traits as if they implement
+        // `hyper::rt` IO traits.
+        let io = TokioIo::new(stream);
+
+        // Spin up a new task in Tokio so we can continue to listen for new TCP connection on the
+        // current task without waiting for the processing of the HTTP/2 connection we just received
+        // to finish
+        tokio::task::spawn(async move {
+            // Handle the connection from the client using HTTP/2 with an executor and pass any
+            // HTTP requests received on that connection to the `hello` function
+            if let Err(err) = http2::Builder::new(TokioExecutor)
+                .serve_connection(io, service_fn(hello))
+                .await
+            {
+                eprintln!("Error serving connection: {}", err);
+            }
+        });
+    }
+}
diff --git a/src/client/conn/http1.rs b/src/client/conn/http1.rs
index 647171e7..ecfe6eb8 100644
--- a/src/client/conn/http1.rs
+++ b/src/client/conn/http1.rs
@@ -48,6 +48,8 @@ pub struct Parts<T> {
 ///
 /// In most cases, this should just be spawned into an executor, so that it
 /// can process incoming and outgoing messages, notice hangups, and the like.
+///
+/// Instances of this type are typically created via the [`handshake`] function
 #[must_use = "futures do nothing unless polled"]
 pub struct Connection<T, B>
 where
diff --git a/src/client/conn/http2.rs b/src/client/conn/http2.rs
index ce86c6dc..3db28957 100644
--- a/src/client/conn/http2.rs
+++ b/src/client/conn/http2.rs
@@ -37,6 +37,8 @@ impl<B> Clone for SendRequest<B> {
 ///
 /// In most cases, this should just be spawned into an executor, so that it
 /// can process incoming and outgoing messages, notice hangups, and the like.
+///
+/// Instances of this type are typically created via the [`handshake`] function
 #[must_use = "futures do nothing unless polled"]
 pub struct Connection<T, B, E>
 where
diff --git a/src/lib.rs b/src/lib.rs
index 23087a0e..4683fd65 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -52,19 +52,22 @@
 //!
 //! [feature flags]: https://doc.rust-lang.org/cargo/reference/manifest.html#the-features-section
 //!
-//! # Unstable Features
+//! ## Unstable Features
 //!
 //! hyper includes a set of unstable optional features that can be enabled through the use of a
 //! feature flag and a [configuration flag].
 //!
 //! The following is a list of feature flags and their corresponding `RUSTFLAG`:
+//!
 //! - `ffi`: Enables C API for hyper `hyper_unstable_ffi`.
 //! - `tracing`: Enables debug logging with `hyper_unstable_tracing`.
 //!
-//! Enabling an unstable feature is possible with the following `cargo` command, as of version `1.64.0`:
+//! For example:
+//!
 //! ```notrust
-//! RUSTFLAGS="--cfg hyper_unstable_tracing" cargo rustc --features client,http1,http2,tracing --crate-type cdylib
-//!```
+//! RUSTFLAGS="--cfg hyper_unstable_tracing" cargo build
+//! ```
+//!
 //! [configuration flag]: https://doc.rust-lang.org/reference/conditional-compilation.html
 //!
 //! # Stability
@@ -93,10 +96,8 @@ pub use http;
 #[cfg(all(test, feature = "nightly"))]
 extern crate test;
 
-pub use crate::http::{header, Method, Request, Response, StatusCode, Uri, Version};
-
 #[doc(no_inline)]
-pub use crate::http::HeaderMap;
+pub use http::{header, HeaderMap, Method, Request, Response, StatusCode, Uri, Version};
 
 pub use crate::error::{Error, Result};
 
diff --git a/src/proto/h1/conn.rs b/src/proto/h1/conn.rs
index 10f4f87b..8ddf7558 100644
--- a/src/proto/h1/conn.rs
+++ b/src/proto/h1/conn.rs
@@ -21,7 +21,7 @@ use super::{Decoder, Encode, EncodedBuf, Encoder, Http1Transaction, ParseContext
 use crate::body::DecodedLength;
 #[cfg(feature = "server")]
 use crate::common::time::Time;
-use crate::headers::connection_keep_alive;
+use crate::headers;
 use crate::proto::{BodyLength, MessageHead};
 #[cfg(feature = "server")]
 use crate::rt::Sleep;
@@ -657,7 +657,7 @@ where
         let outgoing_is_keep_alive = head
             .headers
             .get(CONNECTION)
-            .map_or(false, connection_keep_alive);
+            .map_or(false, headers::connection_keep_alive);
 
         if !outgoing_is_keep_alive {
             match head.version {
@@ -680,12 +680,21 @@ where
     // If we know the remote speaks an older version, we try to fix up any messages
     // to work with our older peer.
     fn enforce_version(&mut self, head: &mut MessageHead<T::Outgoing>) {
-        if let Version::HTTP_10 = self.state.version {
-            // Fixes response or connection when keep-alive header is not present
-            self.fix_keep_alive(head);
-            // If the remote only knows HTTP/1.0, we should force ourselves
-            // to do only speak HTTP/1.0 as well.
-            head.version = Version::HTTP_10;
+        match self.state.version {
+            Version::HTTP_10 => {
+                // Fixes response or connection when keep-alive header is not present
+                self.fix_keep_alive(head);
+                // If the remote only knows HTTP/1.0, we should force ourselves
+                // to do only speak HTTP/1.0 as well.
+                head.version = Version::HTTP_10;
+            }
+            Version::HTTP_11 => {
+                if let KA::Disabled = self.state.keep_alive.status() {
+                    head.headers
+                        .insert(CONNECTION, HeaderValue::from_static("close"));
+                }
+            }
+            _ => (),
         }
         // If the remote speaks HTTP/1.1, then it *should* be fine with
         // both HTTP/1.0 and HTTP/1.1 from us. So again, we just let
diff --git a/src/proto/h1/io.rs b/src/proto/h1/io.rs
index 4ad2fca1..5f7b836d 100644
--- a/src/proto/h1/io.rs
+++ b/src/proto/h1/io.rs
@@ -32,6 +32,7 @@ const MAX_BUF_LIST_BUFFERS: usize = 16;
 pub(crate) struct Buffered<T, B> {
     flush_pipeline: bool,
     io: T,
+    partial_len: Option<usize>,
     read_blocked: bool,
     read_buf: BytesMut,
     read_buf_strategy: ReadStrategy,
@@ -65,6 +66,7 @@ where
         Buffered {
             flush_pipeline: false,
             io,
+            partial_len: None,
             read_blocked: false,
             read_buf: BytesMut::with_capacity(0),
             read_buf_strategy: ReadStrategy::default(),
@@ -176,6 +178,7 @@ where
         loop {
             match super::role::parse_headers::<S>(
                 &mut self.read_buf,
+                self.partial_len,
                 ParseContext {
                     cached_headers: parse_ctx.cached_headers,
                     req_method: parse_ctx.req_method,
@@ -191,14 +194,19 @@ where
             )? {
                 Some(msg) => {
                     debug!("parsed {} headers", msg.head.headers.len());
+                    self.partial_len = None;
                     return Poll::Ready(Ok(msg));
                 }
                 None => {
                     let max = self.read_buf_strategy.max();
-                    if self.read_buf.len() >= max {
+                    let curr_len = self.read_buf.len();
+                    if curr_len >= max {
                         debug!("max_buf_size ({}) reached, closing", max);
                         return Poll::Ready(Err(crate::Error::new_too_large()));
                     }
+                    if curr_len > 0 {
+                        self.partial_len = Some(curr_len);
+                    }
                 }
             }
             if ready!(self.poll_read_from_io(cx)).map_err(crate::Error::new_io)? == 0 {
diff --git a/src/proto/h1/role.rs b/src/proto/h1/role.rs
index e5a88721..4f04acec 100644
--- a/src/proto/h1/role.rs
+++ b/src/proto/h1/role.rs
@@ -66,6 +66,7 @@ macro_rules! maybe_panic {
 
 pub(super) fn parse_headers<T>(
     bytes: &mut BytesMut,
+    prev_len: Option<usize>,
     ctx: ParseContext<'_>,
 ) -> ParseResult<T::Incoming>
 where
@@ -78,9 +79,37 @@ where
 
     let _entered = trace_span!("parse_headers");
 
+    if let Some(prev_len) = prev_len {
+        if !is_complete_fast(bytes, prev_len) {
+            return Ok(None);
+        }
+    }
+
     T::parse(bytes, ctx)
 }
 
+/// A fast scan for the end of a message.
+/// Used when there was a partial read, to skip full parsing on a
+/// a slow connection.
+fn is_complete_fast(bytes: &[u8], prev_len: usize) -> bool {
+    let start = if prev_len < 3 { 0 } else { prev_len - 3 };
+    let bytes = &bytes[start..];
+
+    for (i, b) in bytes.iter().copied().enumerate() {
+        if b == b'\r' {
+            if bytes[i + 1..].chunks(3).next() == Some(&b"\n\r\n"[..]) {
+                return true;
+            }
+        } else if b == b'\n' {
+            if bytes.get(i + 1) == Some(&b'\n') {
+                return true;
+            }
+        }
+    }
+
+    false
+}
+
 pub(super) fn encode_headers<T>(
     enc: Encode<'_, T::Outgoing>,
     dst: &mut Vec<u8>,
@@ -2827,6 +2856,28 @@ mod tests {
         parse(Some(200), 210, false);
     }
 
+    #[test]
+    fn test_is_complete_fast() {
+        let s = b"GET / HTTP/1.1\r\na: b\r\n\r\n";
+        for n in 0..s.len() {
+            assert!(is_complete_fast(s, n), "{:?}; {}", s, n);
+        }
+        let s = b"GET / HTTP/1.1\na: b\n\n";
+        for n in 0..s.len() {
+            assert!(is_complete_fast(s, n));
+        }
+
+        // Not
+        let s = b"GET / HTTP/1.1\r\na: b\r\n\r";
+        for n in 0..s.len() {
+            assert!(!is_complete_fast(s, n));
+        }
+        let s = b"GET / HTTP/1.1\na: b\n";
+        for n in 0..s.len() {
+            assert!(!is_complete_fast(s, n));
+        }
+    }
+
     #[test]
     fn test_write_headers_orig_case_empty_value() {
         let mut headers = HeaderMap::new();
diff --git a/src/proto/h2/server.rs b/src/proto/h2/server.rs
index ee2d08ea..a434ba9e 100644
--- a/src/proto/h2/server.rs
+++ b/src/proto/h2/server.rs
@@ -89,6 +89,7 @@ pin_project! {
         service: S,
         state: State<T, B>,
         date_header: bool,
+        close_pending: bool
     }
 }
 
@@ -101,7 +102,6 @@ where
         hs: Handshake<Compat<T>, SendBuf<B::Data>>,
     },
     Serving(Serving<T, B>),
-    Closed,
 }
 
 struct Serving<T, B>
@@ -172,6 +172,7 @@ where
             },
             service,
             date_header: config.date_header,
+            close_pending: false,
         }
     }
 
@@ -179,7 +180,8 @@ where
         trace!("graceful_shutdown");
         match self.state {
             State::Handshaking { .. } => {
-                // fall-through, to replace state with Closed
+                self.close_pending = true;
+                return;
             }
             State::Serving(ref mut srv) => {
                 if srv.closing.is_none() {
@@ -187,11 +189,7 @@ where
                 }
                 return;
             }
-            State::Closed => {
-                return;
-            }
         }
-        self.state = State::Closed;
     }
 }
 
@@ -228,12 +226,11 @@ where
                     })
                 }
                 State::Serving(ref mut srv) => {
-                    ready!(srv.poll_server(cx, &mut me.service, &mut me.exec))?;
-                    return Poll::Ready(Ok(Dispatched::Shutdown));
-                }
-                State::Closed => {
                     // graceful_shutdown was called before handshaking finished,
-                    // nothing to do here...
+                    if true == me.close_pending && srv.closing.is_none() {
+                        srv.conn.graceful_shutdown();
+                    }
+                    ready!(srv.poll_server(cx, &mut me.service, &mut me.exec))?;
                     return Poll::Ready(Ok(Dispatched::Shutdown));
                 }
             };
@@ -485,6 +482,13 @@ where
                                 me.reply.send_reset(h2::Reason::INTERNAL_ERROR);
                                 return Poll::Ready(Err(crate::Error::new_user_header()));
                             }
+                            if res
+                                .headers_mut()
+                                .remove(::http::header::CONTENT_LENGTH)
+                                .is_some()
+                            {
+                                warn!("successful response to CONNECT request disallows content-length header");
+                            }
                             let send_stream = reply!(me, res, false);
                             connect_parts.pending.fulfill(Upgraded::new(
                                 H2Upgraded {
diff --git a/src/rt/timer.rs b/src/rt/timer.rs
index c486ecc1..c6a6f1db 100644
--- a/src/rt/timer.rs
+++ b/src/rt/timer.rs
@@ -57,7 +57,7 @@
 //!         self.project().inner.as_mut().reset(deadline.into());
 //!     }
 //! }
-//! ````
+//! ```
 
 use std::{
     any::TypeId,
diff --git a/src/upgrade.rs b/src/upgrade.rs
index 03d7e98d..9d23a290 100644
--- a/src/upgrade.rs
+++ b/src/upgrade.rs
@@ -58,7 +58,7 @@ use crate::common::io::Rewind;
 ///
 /// This type holds a trait object internally of the original IO that
 /// was used to speak HTTP before the upgrade. It can be used directly
-/// as a `Read` or `Write` for convenience.
+/// as a [`Read`] or [`Write`] for convenience.
 ///
 /// Alternatively, if the exact type is known, this can be deconstructed
 /// into its parts.


Test patch:
diff --git a/tests/client.rs b/tests/client.rs
index 6b6a6df2..4bdbecb4 100644
--- a/tests/client.rs
+++ b/tests/client.rs
@@ -2044,22 +2044,20 @@ mod conn {
     #[tokio::test]
     async fn test_try_send_request() {
         use std::future::Future;
-        let (listener, addr) = setup_tk_test_server().await;
         let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>();
+        let (io_srv, io_cli) = tokio_test::io::Builder::new()
+            .write(b"GET / HTTP/1.1\r\n\r\n")
+            .read(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n")
+            .build_with_handle();
 
         tokio::spawn(async move {
-            let mut sock = listener.accept().await.unwrap().0;
-            let mut buf = [0u8; 8192];
-            sock.read(&mut buf).await.expect("read 1");
-            sock.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n")
-                .await
-                .expect("write 1");
+            let _io = io_cli;
             let _ = done_rx.await;
         });
 
         // make polling fair by putting both in spawns
         tokio::spawn(async move {
-            let io = tcp_connect(&addr).await.expect("tcp connect");
+            let io = TokioIo::new(io_srv);
             let (mut client, mut conn) = conn::http1::Builder::new()
                 .handshake::<_, Empty<Bytes>>(io)
                 .await
diff --git a/tests/server.rs b/tests/server.rs
index 5120ad77..f72cf627 100644
--- a/tests/server.rs
+++ b/tests/server.rs
@@ -1140,6 +1140,8 @@ fn pipeline_enabled() {
 
         assert_eq!(s(lines.next().unwrap()), "HTTP/1.1 200 OK\r");
         assert_eq!(s(lines.next().unwrap()), "content-length: 12\r");
+        // close because the last request said to close
+        assert_eq!(s(lines.next().unwrap()), "connection: close\r");
         lines.next().unwrap(); // Date
         assert_eq!(s(lines.next().unwrap()), "\r");
         assert_eq!(s(lines.next().unwrap()), "Hello World");
@@ -1181,7 +1183,7 @@ fn http_11_uri_too_long() {
     let mut req = connect(server.addr());
     req.write_all(request_line.as_bytes()).unwrap();
 
-    let expected = "HTTP/1.1 414 URI Too Long\r\ncontent-length: 0\r\n";
+    let expected = "HTTP/1.1 414 URI Too Long\r\nconnection: close\r\ncontent-length: 0\r\n";
     let mut buf = [0; 256];
     let n = req.read(&mut buf).unwrap();
     assert!(n >= expected.len(), "read: {:?} >= {:?}", n, expected.len());
@@ -1208,6 +1210,12 @@ async fn disable_keep_alive_mid_request() {
             "should receive OK response, but buf: {:?}",
             buf,
         );
+        let sbuf = s(&buf);
+        assert!(
+            sbuf.contains("connection: close\r\n"),
+            "response should have sent close: {:?}",
+            sbuf,
+        );
     });
 
     let (socket, _) = listener.accept().await.unwrap();
@@ -2366,7 +2374,7 @@ fn streaming_body() {
         buf.starts_with(b"HTTP/1.1 200 OK\r\n"),
         "response is 200 OK"
     );
-    assert_eq!(buf.len(), 100_789, "full streamed body read");
+    assert_eq!(buf.len(), 100_808, "full streamed body read");
 }
 
 #[test]