Merge branch 'master' into check-external-types

.github/workflows/CI.yml

@@ -177,7 +177,7 @@ jobs:
         uses: actions/checkout@v3
 
       - name: Install Rust
-        uses: dtolnay/rust-toolchain@nightly
+        uses: dtolnay/rust-toolchain@stable
 
       - name: Install cbindgen
         uses: taiki-e/cache-cargo-install-action@v1

CHANGELOG.md

@@ -1,3 +1,40 @@
+### v1.0.0-rc.4 (2023-07-10)
+
+
+#### Bug Fixes
+
+* **http1:**
+  * http1 server graceful shutdown fix (#3261) ([f4b51300](https://github.com/hyperium/hyper/commit/f4b513009d81083081d1c60c1981847bbb17dd5d))
+  * send error on Incoming body when connection errors (#3256) ([52f19259](https://github.com/hyperium/hyper/commit/52f192593fb9ebcf6d3894e0c85cbf710da4decd), closes [#3253](https://github.com/hyperium/hyper/issues/3253))
+  * properly end chunked bodies when it was known to be empty (#3254) ([fec64cf0](https://github.com/hyperium/hyper/commit/fec64cf0abdc678e30ca5f1b310c5118b2e01999), closes [#3252](https://github.com/hyperium/hyper/issues/3252))
+
+
+#### Features
+
+* **client:** Make clients able to use non-Send executor (#3184) ([d977f209](https://github.com/hyperium/hyper/commit/d977f209bc6068d8f878b22803fc42d90c887fcc), closes [#3017](https://github.com/hyperium/hyper/issues/3017))
+* **rt:**
+  * replace IO traits with hyper::rt ones (#3230) ([f9f65b7a](https://github.com/hyperium/hyper/commit/f9f65b7aa67fa3ec0267fe015945973726285bc2), closes [#3110](https://github.com/hyperium/hyper/issues/3110))
+  * add downcast on `Sleep` trait (#3125) ([d92d3917](https://github.com/hyperium/hyper/commit/d92d3917d950e4c61c37c2170f3ce273d2a0f7d1), closes [#3027](https://github.com/hyperium/hyper/issues/3027))
+* **service:** change Service::call to take &self (#3223) ([d894439e](https://github.com/hyperium/hyper/commit/d894439e009aa75103f6382a7ba98fb17da72f02), closes [#3040](https://github.com/hyperium/hyper/issues/3040))
+
+
+#### Breaking Changes
+
+* Any IO transport type provided must not 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))
+* `client::conn::http2` types now use another generic for an `Executor`.
+  Code that names `Connection` needs to include the additional generic parameter.
+ ([d977f209](https://github.com/hyperium/hyper/commit/d977f209bc6068d8f878b22803fc42d90c887fcc))
+* The Service::call function no longer takes a mutable reference to self.
+  The FnMut trait bound on the service::util::service_fn function and the trait bound
+  on the impl for the ServiceFn struct were changed from FnMut to Fn.
+
+ ([d894439e](https://github.com/hyperium/hyper/commit/d894439e009aa75103f6382a7ba98fb17da72f02))
+
+
+
 ### v1.0.0-rc.3 (2023-02-23)
 
 

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hyper"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.4"
 description = "A fast and correct HTTP library."
 readme = "README.md"
 homepage = "https://hyper.rs"
@@ -25,7 +25,7 @@ futures-channel = "0.3"
 futures-util = { version = "0.3", default-features = false }
 http = "0.2"
 http-body = "=1.0.0-rc.2"
-http-body-util = { version = "=0.1.0-rc.2", optional = true }
+http-body-util = { version = "=0.1.0-rc.3", optional = true }
 httpdate = "1.0"
 httparse = "1.8"
 h2 = { version = "0.3.9", optional = true }
@@ -41,7 +41,7 @@ libc = { version = "0.2", optional = true }
 
 [dev-dependencies]
 futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
-http-body-util = "=0.1.0-rc.2"
+http-body-util = "=0.1.0-rc.3"
 matches = "0.1"
 num_cpus = "1.0"
 pretty_env_logger = "0.4"

benches/end_to_end.rs

@@ -4,8 +4,7 @@
 extern crate test;
 mod support;
 
-// TODO: Reimplement Opts::bench using hyper::server::conn and hyper::client::conn
-// (instead of Server and HttpClient).
+// TODO: Reimplement parallel for HTTP/1
 
 use std::convert::Infallible;
 use std::net::SocketAddr;
@@ -315,7 +314,8 @@ impl Opts {
 
         let mut client = rt.block_on(async {
             if self.http2 {
-                let io = tokio::net::TcpStream::connect(&addr).await.unwrap();
+                let tcp = tokio::net::TcpStream::connect(&addr).await.unwrap();
+                let io = support::TokioIo::new(tcp);
                 let (tx, conn) = hyper::client::conn::http2::Builder::new(support::TokioExecutor)
                     .initial_stream_window_size(self.http2_stream_window)
                     .initial_connection_window_size(self.http2_conn_window)
@@ -328,7 +328,8 @@ impl Opts {
             } else if self.parallel_cnt > 1 {
                 todo!("http/1 parallel >1");
             } else {
-                let io = tokio::net::TcpStream::connect(&addr).await.unwrap();
+                let tcp = tokio::net::TcpStream::connect(&addr).await.unwrap();
+                let io = support::TokioIo::new(tcp);
                 let (tx, conn) = hyper::client::conn::http1::Builder::new()
                     .handshake(io)
                     .await
@@ -414,14 +415,15 @@ fn spawn_server(rt: &tokio::runtime::Runtime, opts: &Opts) -> SocketAddr {
     let opts = opts.clone();
     rt.spawn(async move {
         while let Ok((sock, _)) = listener.accept().await {
+            let io = support::TokioIo::new(sock);
             if opts.http2 {
                 tokio::spawn(
                     hyper::server::conn::http2::Builder::new(support::TokioExecutor)
                         .initial_stream_window_size(opts.http2_stream_window)
                         .initial_connection_window_size(opts.http2_conn_window)
                         .adaptive_window(opts.http2_adaptive_window)
                         .serve_connection(
-                            sock,
+                            io,
                             service_fn(move |req: Request<hyper::body::Incoming>| async move {
                                 let mut req_body = req.into_body();
                                 while let Some(_chunk) = req_body.frame().await {}
@@ -433,7 +435,7 @@ fn spawn_server(rt: &tokio::runtime::Runtime, opts: &Opts) -> SocketAddr {
                 );
             } else {
                 tokio::spawn(hyper::server::conn::http1::Builder::new().serve_connection(
-                    sock,
+                    io,
                     service_fn(move |req: Request<hyper::body::Incoming>| async move {
                         let mut req_body = req.into_body();
                         while let Some(_chunk) = req_body.frame().await {}

benches/pipeline.rs

@@ -3,6 +3,8 @@
 
 extern crate test;
 
+mod support;
+
 use std::convert::Infallible;
 use std::io::{Read, Write};
 use std::net::{SocketAddr, TcpStream};
@@ -40,11 +42,12 @@ fn hello_world_16(b: &mut test::Bencher) {
             rt.spawn(async move {
                 loop {
                     let (stream, _addr) = listener.accept().await.expect("accept");
+                    let io = support::TokioIo::new(stream);
 
                     http1::Builder::new()
                         .pipeline_flush(true)
                         .serve_connection(
-                            stream,
+                            io,
                             service_fn(|_| async {
                                 Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(
                                     "Hello, World!",

benches/server.rs

@@ -3,6 +3,8 @@
 
 extern crate test;
 
+mod support;
+
 use std::io::{Read, Write};
 use std::net::{SocketAddr, TcpListener, TcpStream};
 use std::sync::mpsc;
@@ -38,10 +40,11 @@ macro_rules! bench_server {
                 rt.spawn(async move {
                     loop {
                         let (stream, _) = listener.accept().await.expect("accept");
+                        let io = support::TokioIo::new(stream);
 
                         http1::Builder::new()
                             .serve_connection(
-                                stream,
+                                io,
                                 service_fn(|_| async {
                                     Ok::<_, hyper::Error>(
                                         Response::builder()

benches/support/mod.rs

@@ -1,2 +1,2 @@
 mod tokiort;
-pub use tokiort::{TokioExecutor, TokioTimer};
+pub use tokiort::{TokioExecutor, TokioIo, TokioTimer};

benches/support/tokiort.rs

@@ -41,6 +41,12 @@ impl Timer for TokioTimer {
             inner: tokio::time::sleep_until(deadline.into()),
         })
     }
+
+    fn reset(&self, sleep: &mut Pin<Box<dyn Sleep>>, new_deadline: Instant) {
+        if let Some(sleep) = sleep.as_mut().downcast_mut_pin::<TokioSleep>() {
+            sleep.reset(new_deadline.into())
+        }
+    }
 }
 
 struct TokioTimeout<T> {
@@ -75,7 +81,156 @@ impl Future for TokioSleep {
     }
 }
 
-// Use HasSleep to get tokio::time::Sleep to implement Unpin.
-// see https://docs.rs/tokio/latest/tokio/time/struct.Sleep.html
-
 impl Sleep for TokioSleep {}
+
+impl TokioSleep {
+    pub fn reset(self: Pin<&mut Self>, deadline: Instant) {
+        self.project().inner.as_mut().reset(deadline.into());
+    }
+}
+
+pin_project! {
+    #[derive(Debug)]
+    pub struct TokioIo<T> {
+        #[pin]
+        inner: T,
+    }
+}
+
+impl<T> TokioIo<T> {
+    pub fn new(inner: T) -> Self {
+        Self { inner }
+    }
+
+    pub fn inner(self) -> T {
+        self.inner
+    }
+}
+
+impl<T> hyper::rt::Read for TokioIo<T>
+where
+    T: tokio::io::AsyncRead,
+{
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        mut buf: hyper::rt::ReadBufCursor<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        let n = unsafe {
+            let mut tbuf = tokio::io::ReadBuf::uninit(buf.as_mut());
+            match tokio::io::AsyncRead::poll_read(self.project().inner, cx, &mut tbuf) {
+                Poll::Ready(Ok(())) => tbuf.filled().len(),
+                other => return other,
+            }
+        };
+
+        unsafe {
+            buf.advance(n);
+        }
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl<T> hyper::rt::Write for TokioIo<T>
+where
+    T: tokio::io::AsyncWrite,
+{
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        tokio::io::AsyncWrite::poll_write(self.project().inner, cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
+        tokio::io::AsyncWrite::poll_flush(self.project().inner, cx)
+    }
+
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        tokio::io::AsyncWrite::poll_shutdown(self.project().inner, cx)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        tokio::io::AsyncWrite::is_write_vectored(&self.inner)
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        tokio::io::AsyncWrite::poll_write_vectored(self.project().inner, cx, bufs)
+    }
+}
+
+impl<T> tokio::io::AsyncRead for TokioIo<T>
+where
+    T: hyper::rt::Read,
+{
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        tbuf: &mut tokio::io::ReadBuf<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        //let init = tbuf.initialized().len();
+        let filled = tbuf.filled().len();
+        let sub_filled = unsafe {
+            let mut buf = hyper::rt::ReadBuf::uninit(tbuf.unfilled_mut());
+
+            match hyper::rt::Read::poll_read(self.project().inner, cx, buf.unfilled()) {
+                Poll::Ready(Ok(())) => buf.filled().len(),
+                other => return other,
+            }
+        };
+
+        let n_filled = filled + sub_filled;
+        // At least sub_filled bytes had to have been initialized.
+        let n_init = sub_filled;
+        unsafe {
+            tbuf.assume_init(n_init);
+            tbuf.set_filled(n_filled);
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl<T> tokio::io::AsyncWrite for TokioIo<T>
+where
+    T: hyper::rt::Write,
+{
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        hyper::rt::Write::poll_write(self.project().inner, cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
+        hyper::rt::Write::poll_flush(self.project().inner, cx)
+    }
+
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        hyper::rt::Write::poll_shutdown(self.project().inner, cx)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        hyper::rt::Write::is_write_vectored(&self.inner)
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        hyper::rt::Write::poll_write_vectored(self.project().inner, cx, bufs)
+    }
+}

examples/client.rs

@@ -8,6 +8,10 @@ use hyper::Request;
 use tokio::io::{self, AsyncWriteExt as _};
 use tokio::net::TcpStream;
 
+#[path = "../benches/support/mod.rs"]
+mod support;
+use support::TokioIo;
+
 // A simple type alias so as to DRY.
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
 
@@ -40,8 +44,9 @@ async fn fetch_url(url: hyper::Uri) -> Result<()> {
     let port = url.port_u16().unwrap_or(80);
     let addr = format!("{}:{}", host, port);
     let stream = TcpStream::connect(addr).await?;
+    let io = TokioIo::new(stream);
 
-    let (mut sender, conn) = hyper::client::conn::http1::handshake(stream).await?;
+    let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await?;
     tokio::task::spawn(async move {
         if let Err(err) = conn.await {
             println!("Connection failed: {:?}", err);

examples/client_json.rs

@@ -7,6 +7,10 @@ use hyper::{body::Buf, Request};
 use serde::Deserialize;
 use tokio::net::TcpStream;
 
+#[path = "../benches/support/mod.rs"]
+mod support;
+use support::TokioIo;
+
 // A simple type alias so as to DRY.
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
 
@@ -29,8 +33,9 @@ async fn fetch_json(url: hyper::Uri) -> Result<Vec<User>> {
     let addr = format!("{}:{}", host, port);
 
     let stream = TcpStream::connect(addr).await?;
+    let io = TokioIo::new(stream);
 
-    let (mut sender, conn) = hyper::client::conn::http1::handshake(stream).await?;
+    let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await?;
     tokio::task::spawn(async move {
         if let Err(err) = conn.await {
             println!("Connection failed: {:?}", err);