feat(rt): replace IO traits with hyper::rt ones (#3230)

This replaces the usage of `tokio::io::{AsyncRead, AsyncWrite}` in hyper's public API with new traits in the `hyper::rt` module.

Closes #3110

BREAKING CHANGE: 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.

Author: seanmonstar

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!",