Repository: hatoo/oha
Pull request #326 (2023-10-09 17:49:38)

Support HTTP/2
Implement #201 



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index bbf69f1..933b667 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,8 @@
 # Unreleased
 
+- Support HTTP/2 #224 #201
+    - Added `-p` option to set number of HTTP/2 parallel requests
+
 # 0.6.5 (2023-09-04)
 
 - Fix Apple Silicon's binary release #323
diff --git a/Cargo.lock b/Cargo.lock
index d7a36a7..b8c877d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -135,8 +135,8 @@ dependencies = [
  "bytes",
  "futures-util",
  "http",
- "http-body",
- "hyper",
+ "http-body 0.4.5",
+ "hyper 0.14.27",
  "itoa",
  "matchit",
  "memchr",
@@ -165,7 +165,7 @@ dependencies = [
  "bytes",
  "futures-util",
  "http",
- "http-body",
+ "http-body 0.4.5",
  "mime",
  "rustversion",
  "tower-layer",
@@ -633,6 +633,29 @@ dependencies = [
  "pin-project-lite",
 ]
 
+[[package]]
+name = "http-body"
+version = "1.0.0-rc.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "951dfc2e32ac02d67c90c0d65bd27009a635dc9b381a2cc7d284ab01e3a0150d"
+dependencies = [
+ "bytes",
+ "http",
+]
+
+[[package]]
+name = "http-body-util"
+version = "0.1.0-rc.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08ef12f041acdd397010e5fb6433270c147d3b8b2d0a840cd7fff8e531dca5c8"
+dependencies = [
+ "bytes",
+ "futures-util",
+ "http",
+ "http-body 1.0.0-rc.2",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "httparse"
 version = "1.8.0"
@@ -663,7 +686,7 @@ dependencies = [
  "futures-util",
  "h2",
  "http",
- "http-body",
+ "http-body 0.4.5",
  "httparse",
  "httpdate",
  "itoa",
@@ -675,6 +698,25 @@ dependencies = [
  "want",
 ]
 
+[[package]]
+name = "hyper"
+version = "1.0.0-rc.4"
+source = "git+https://github.com/hyperium/hyper.git#5a8f4c0f60cb99f5935f4ed6aca268e3ed641ab1"
+dependencies = [
+ "bytes",
+ "futures-channel",
+ "futures-util",
+ "h2",
+ "http",
+ "http-body 1.0.0-rc.2",
+ "httparse",
+ "httpdate",
+ "itoa",
+ "pin-project-lite",
+ "tokio",
+ "want",
+]
+
 [[package]]
 name = "idna"
 version = "0.4.0"
@@ -935,13 +977,17 @@ dependencies = [
  "float-ord",
  "flume",
  "futures",
+ "futures-util",
  "http",
+ "http-body-util",
  "humantime",
- "hyper",
+ "hyper 0.14.27",
+ "hyper 1.0.0-rc.4",
  "jemallocator",
  "lazy_static",
  "libc",
  "native-tls",
+ "pin-project-lite",
  "rand",
  "rand_regex",
  "ratatui",
diff --git a/Cargo.toml b/Cargo.toml
index 166ffcc..d5266b9 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -42,7 +42,7 @@ tui = { package = "ratatui", version = "0.23", default-features = false, feature
 ] }
 
 http = "0.2"
-hyper = { version = "0.14.27", features = ["full"] }
+hyper = { git = "https://github.com/hyperium/hyper.git", features = ["full"] }
 
 # native-tls
 native-tls = { version = "0.2.11", optional = true }
@@ -62,14 +62,18 @@ trust-dns-resolver = "0.23.0"
 rand_regex = "0.16.0"
 regex-syntax = "0.7.4"
 url = "2.4.0"
+futures-util = "0.3.28"
+pin-project-lite = "0.2.13"
+http-body-util = "0.1.0-rc.3"
 
 [target.'cfg(unix)'.dependencies]
 rlimit = "0.10.0"
 jemallocator = "0.5.0"
 
 [dev-dependencies]
+hyper14 = { package = "hyper", version = "0.14", features = ["full"] }
 assert_cmd = "2.0.2"
-axum = "0.6.20"
+axum = { version = "0.6.20", features = ["http2"] }
 bytes = "1.0"
 http = "0.2"
 lazy_static = "1.4.0"
diff --git a/README.md b/README.md
index 6906768..86c5ade 100644
--- a/README.md
+++ b/README.md
@@ -57,7 +57,8 @@ Arguments:
 
 Options:
   -n <N_REQUESTS>                     Number of requests to run. [default: 200]
-  -c <N_WORKERS>                      Number of workers to run concurrently. You may should increase limit to number of open files for larger `-c`. [default: 50]
+  -c <N_CONNECTIONS>                  Number of connections to run concurrently. You may should increase limit to number of open files for larger `-c`. [default: 50]
+  -p <N_HTTP2_PARALLEL>               Number of parallel requests to send on HTTP/2. `oha` will run c * p concurrent workers in total. [default: 1]
   -z <DURATION>                       Duration of application to send requests. If duration is specified, n is ignored.
                                       Examples: -z 10s -z 3m.
   -q <QUERY_PER_SECOND>               Rate limit for all, in queries per second (QPS)
@@ -80,10 +81,11 @@ Options:
   -T <CONTENT_TYPE>                   Content-Type.
   -a <BASIC_AUTH>                     Basic authentication, username:password
       --http-version <HTTP_VERSION>   HTTP version. Available values 0.9, 1.0, 1.1.
+      --http2                         Use HTTP/2. Shorthand for --http-version=2
       --host <HOST>                   HTTP Host header
       --disable-compression           Disable compression.
-  -r, --redirect <REDIRECT>           Limit for number of Redirect. Set 0 for no redirection. [default: 10]
-      --disable-keepalive             Disable keep-alive, prevents re-use of TCP connections between different HTTP requests.
+  -r, --redirect <REDIRECT>           Limit for number of Redirect. Set 0 for no redirection. Redirection isn't supported for HTTP/2. [default: 10]
+      --disable-keepalive             Disable keep-alive, prevents re-use of TCP connections between different HTTP requests. This isn't supported for HTTP/2.
       --ipv6                          Lookup only ipv6.
       --ipv4                          Lookup only ipv4.
       --insecure                      Accept invalid certs.
diff --git a/src/client.rs b/src/client.rs
index a750645..b202bff 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -1,14 +1,21 @@
 use futures::future::FutureExt;
-use futures::StreamExt;
+use http_body_util::Full;
+use hyper::body::{Body, Incoming};
 use rand::prelude::*;
+use std::pin::Pin;
 use std::sync::Arc;
 use thiserror::Error;
+use tokio::net::TcpStream;
 use url::{ParseError, Url};
 
+use crate::tokiort::{TokioExecutor, TokioIo};
 use crate::url_generator::{UrlGenerator, UrlGeneratorError};
 use crate::ConnectToEntry;
 
-#[derive(Debug, Clone)]
+type SendRequestHttp1 = hyper::client::conn::http1::SendRequest<Full<&'static [u8]>>;
+type SendRequestHttp2 = hyper::client::conn::http2::SendRequest<Full<&'static [u8]>>;
+
+#[derive(Debug, Clone, Copy)]
 pub struct ConnectionTime {
     pub dns_lookup: std::time::Instant,
     pub dialup: std::time::Instant,
@@ -149,7 +156,6 @@ pub struct Client {
     pub headers: http::header::HeaderMap,
     pub body: Option<&'static [u8]>,
     pub dns: Dns,
-    // client: Option<hyper::client::conn::SendRequest<hyper::Body>>,
     pub timeout: Option<std::time::Duration>,
     pub redirect_limit: usize,
     pub disable_keepalive: bool,
@@ -158,12 +164,12 @@ pub struct Client {
     pub unix_socket: Option<std::path::PathBuf>,
 }
 
-struct ClientState {
+struct ClientStateHttp1 {
     rng: StdRng,
-    send_request: Option<hyper::client::conn::SendRequest<hyper::Body>>,
+    send_request: Option<SendRequestHttp1>,
 }
 
-impl Default for ClientState {
+impl Default for ClientStateHttp1 {
     fn default() -> Self {
         Self {
             rng: StdRng::from_entropy(),
@@ -172,32 +178,111 @@ impl Default for ClientState {
     }
 }
 
+struct ClientStateHttp2 {
+    rng: StdRng,
+    send_request: SendRequestHttp2,
+}
+
+impl Clone for ClientStateHttp2 {
+    fn clone(&self) -> Self {
+        Self {
+            rng: StdRng::from_entropy(),
+            send_request: self.send_request.clone(),
+        }
+    }
+}
+
 pub enum QueryLimit {
     Qps(usize),
     Burst(std::time::Duration, usize),
 }
 
+// To avoid dynamic dispatch
+// I'm not sure how much this is effective
+enum Stream {
+    Tcp(TcpStream),
+    #[cfg(all(feature = "native-tls", not(feature = "rustls")))]
+    Tls(tokio_native_tls::TlsStream<TcpStream>),
+    #[cfg(feature = "rustls")]
+    Tls(tokio_rustls::client::TlsStream<TcpStream>),
+    #[cfg(unix)]
+    Unix(tokio::net::UnixStream),
+}
+
+impl Stream {
+    async fn handshake_http1(self) -> Result<SendRequestHttp1, ClientError> {
+        match self {
+            Stream::Tcp(stream) => {
+                let (send_request, conn) =
+                    hyper::client::conn::http1::handshake(TokioIo::new(stream)).await?;
+                tokio::spawn(conn);
+                Ok(send_request)
+            }
+            Stream::Tls(stream) => {
+                let (send_request, conn) =
+                    hyper::client::conn::http1::handshake(TokioIo::new(stream)).await?;
+                tokio::spawn(conn);
+                Ok(send_request)
+            }
+            #[cfg(unix)]
+            Stream::Unix(stream) => {
+                let (send_request, conn) =
+                    hyper::client::conn::http1::handshake(TokioIo::new(stream)).await?;
+                tokio::spawn(conn);
+                Ok(send_request)
+            }
+        }
+    }
+    async fn handshake_http2(self) -> Result<SendRequestHttp2, ClientError> {
+        match self {
+            Stream::Tcp(stream) => {
+                let (send_request, conn) =
+                    hyper::client::conn::http2::handshake(TokioExecutor, TokioIo::new(stream))
+                        .await?;
+                tokio::spawn(conn);
+                Ok(send_request)
+            }
+            Stream::Tls(stream) => {
+                let (send_request, conn) =
+                    hyper::client::conn::http2::handshake(TokioExecutor, TokioIo::new(stream))
+                        .await?;
+                tokio::spawn(conn);
+                Ok(send_request)
+            }
+            #[cfg(unix)]
+            Stream::Unix(stream) => {
+                let (send_request, conn) =
+                    hyper::client::conn::http2::handshake(TokioExecutor, TokioIo::new(stream))
+                        .await?;
+                tokio::spawn(conn);
+                Ok(send_request)
+            }
+        }
+    }
+}
+
 impl Client {
+    fn is_http2(&self) -> bool {
+        self.http_version == http::Version::HTTP_2
+    }
+
     #[cfg(unix)]
     async fn client(
         &self,
         addr: (std::net::IpAddr, u16),
         url: &Url,
-    ) -> Result<hyper::client::conn::SendRequest<hyper::Body>, ClientError> {
+    ) -> Result<Stream, ClientError> {
         if url.scheme() == "https" {
             self.tls_client(addr, url).await
         } else if let Some(socket_path) = &self.unix_socket {
-            let stream = tokio::net::UnixStream::connect(socket_path).await?;
-            let (send, conn) = hyper::client::conn::handshake(stream).await?;
-            tokio::spawn(conn);
-            Ok(send)
+            Ok(Stream::Unix(
+                tokio::net::UnixStream::connect(socket_path).await?,
+            ))
         } else {
             let stream = tokio::net::TcpStream::connect(addr).await?;
             stream.set_nodelay(true)?;
             // stream.set_keepalive(std::time::Duration::from_secs(1).into())?;
-            let (send, conn) = hyper::client::conn::handshake(stream).await?;
-            tokio::spawn(conn);
-            Ok(send)
+            Ok(Stream::Tcp(stream))
         }
     }
 
@@ -206,16 +291,14 @@ impl Client {
         &self,
         addr: (std::net::IpAddr, u16),
         url: &Url,
-    ) -> Result<hyper::client::conn::SendRequest<hyper::Body>, ClientError> {
+    ) -> Result<Stream, ClientError> {
         if url.scheme() == "https" {
             self.tls_client(addr, url).await
         } else {
             let stream = tokio::net::TcpStream::connect(addr).await?;
             stream.set_nodelay(true)?;
             // stream.set_keepalive(std::time::Duration::from_secs(1).into())?;
-            let (send, conn) = hyper::client::conn::handshake(stream).await?;
-            tokio::spawn(conn);
-            Ok(send)
+            Ok(Stream::Tcp(stream))
         }
     }
 
@@ -224,7 +307,7 @@ impl Client {
         &self,
         addr: (std::net::IpAddr, u16),
         url: &Url,
-    ) -> Result<hyper::client::conn::SendRequest<hyper::Body>, ClientError> {
+    ) -> Result<Stream, ClientError> {
         let stream = tokio::net::TcpStream::connect(addr).await?;
         stream.set_nodelay(true)?;
 
@@ -241,9 +324,7 @@ impl Client {
             .connect(url.host_str().ok_or(ClientError::HostNotFound)?, stream)
             .await?;
 
-        let (send, conn) = hyper::client::conn::handshake(stream).await?;
-        tokio::spawn(conn);
-        Ok(send)
+        Ok(Stream::Tls(stream))
     }
 
     #[cfg(feature = "rustls")]
@@ -251,7 +332,7 @@ impl Client {
         &self,
         addr: (std::net::IpAddr, u16),
         url: &Url,
-    ) -> Result<hyper::client::conn::SendRequest<hyper::Body>, ClientError> {
+    ) -> Result<Stream, ClientError> {
         let stream = tokio::net::TcpStream::connect(addr).await?;
         stream.set_nodelay(true)?;
 
@@ -273,14 +354,25 @@ impl Client {
             rustls::ServerName::try_from(url.host_str().ok_or(ClientError::HostNotFound)?)?;
         let stream = connector.connect(domain, stream).await?;
 
-        let (send, conn) = hyper::client::conn::handshake(stream).await?;
-        tokio::spawn(conn);
-        Ok(send)
+        Ok(Stream::Tls(stream))
+    }
+
+    async fn client_http1(
+        &self,
+        addr: (std::net::IpAddr, u16),
+        url: &Url,
+    ) -> Result<SendRequestHttp1, ClientError> {
+        let stream = self.client(addr, url).await?;
+        stream.handshake_http1().await
     }
 
-    fn request(&self, url: &Url) -> Result<http::Request<hyper::Body>, ClientError> {
+    fn request(&self, url: &Url) -> Result<http::Request<Full<&'static [u8]>>, ClientError> {
         let mut builder = http::Request::builder()
-            .uri(&url[url::Position::BeforePath..])
+            .uri(if self.is_http2() {
+                &url[..]
+            } else {
+                &url[url::Position::BeforePath..]
+            })
             .method(self.method.clone())
             .version(self.http_version);
 
@@ -289,13 +381,16 @@ impl Client {
             .ok_or(ClientError::GetHeaderFromBuilderError)? = self.headers.clone();
 
         if let Some(body) = self.body {
-            Ok(builder.body(hyper::Body::from(body))?)
+            Ok(builder.body(Full::new(body))?)
         } else {
-            Ok(builder.body(hyper::Body::empty())?)
+            Ok(builder.body(Full::default())?)
         }
     }
 
-    async fn work(&self, client_state: &mut ClientState) -> Result<RequestResult, ClientError> {
+    async fn work_http1(
+        &self,
+        client_state: &mut ClientStateHttp1,
+    ) -> Result<RequestResult, ClientError> {
         let timeout = if let Some(timeout) = self.timeout {
             tokio::time::sleep(timeout).boxed()
         } else {
@@ -312,7 +407,7 @@ impl Client {
             } else {
                 let addr = self.dns.lookup(&url, &mut client_state.rng).await?;
                 let dns_lookup = std::time::Instant::now();
-                let send_request = self.client(addr, &url).await?;
+                let send_request = self.client_http1(addr, &url).await?;
                 let dialup = std::time::Instant::now();
 
                 connection_time = Some(ConnectionTime { dns_lookup, dialup });
@@ -325,7 +420,7 @@ impl Client {
                 start = std::time::Instant::now();
                 let addr = self.dns.lookup(&url, &mut client_state.rng).await?;
                 let dns_lookup = std::time::Instant::now();
-                send_request = self.client(addr, &url).await?;
+                send_request = self.client_http1(addr, &url).await?;
                 let dialup = std::time::Instant::now();
                 connection_time = Some(ConnectionTime { dns_lookup, dialup });
             }
@@ -336,8 +431,12 @@ impl Client {
                     let mut status = parts.status;
 
                     let mut len_sum = 0;
-                    while let Some(chunk) = stream.next().await {
-                        len_sum += chunk?.len();
+                    while let Some(chunk) = futures::future::poll_fn(|cx| {
+                        Incoming::poll_frame(Pin::new(&mut stream), cx)
+                    })
+                    .await
+                    {
+                        len_sum += chunk?.data_ref().map(|d| d.len()).unwrap_or_default();
                     }
 
                     if self.redirect_limit != 0 {
@@ -391,25 +490,87 @@ impl Client {
             }
         }
     }
+    async fn connect_http2<R: Rng>(
+        &self,
+        url: &Url,
+        rng: &mut R,
+    ) -> Result<(ConnectionTime, SendRequestHttp2), ClientError> {
+        let addr = self.dns.lookup(url, rng).await?;
+        let dns_lookup = std::time::Instant::now();
+        let stream = self.client(addr, url).await?;
+        let send_request = stream.handshake_http2().await?;
+        let dialup = std::time::Instant::now();
+        Ok((ConnectionTime { dns_lookup, dialup }, send_request))
+    }
+
+    async fn work_http2(
+        &self,
+        client_state: &mut ClientStateHttp2,
+    ) -> Result<RequestResult, ClientError> {
+        let timeout = if let Some(timeout) = self.timeout {
+            tokio::time::sleep(timeout).boxed()
+        } else {
+            std::future::pending().boxed()
+        };
+
+        let do_req = async {
+            let url = self.url_generator.generate(&mut client_state.rng)?;
+            let start = std::time::Instant::now();
+            let connection_time: Option<ConnectionTime> = None;
+
+            let request = self.request(&url)?;
+            match client_state.send_request.send_request(request).await {
+                Ok(res) => {
+                    let (parts, mut stream) = res.into_parts();
+                    let status = parts.status;
+
+                    let mut len_sum = 0;
+                    while let Some(chunk) = futures::future::poll_fn(|cx| {
+                        Incoming::poll_frame(Pin::new(&mut stream), cx)
+                    })
+                    .await
+                    {
+                        len_sum += chunk?.data_ref().map(|d| d.len()).unwrap_or_default();
+                    }
+
+                    let end = std::time::Instant::now();
+
+                    let result = RequestResult {
+                        start_latency_correction: None,
+                        start,
+                        end,
+                        status,
+                        len_bytes: len_sum,
+                        connection_time,
+                    };
+
+                    Ok::<_, ClientError>(result)
+                }
+                Err(e) => Err(e.into()),
+            }
+        };
+
+        tokio::select! {
+            res = do_req => {
+                res
+            }
+            _ = timeout => {
+                Err(ClientError::Timeout)
+            }
+        }
+    }
 
     #[allow(clippy::type_complexity)]
     fn redirect<'a, R: Rng + Send>(
         &'a self,
-        send_request: hyper::client::conn::SendRequest<hyper::Body>,
+        send_request: SendRequestHttp1,
         base_url: &'a Url,
         location: &'a http::header::HeaderValue,
         limit: usize,
         rng: &'a mut R,
     ) -> futures::future::BoxFuture<
         'a,
-        Result<
-            (
-                hyper::client::conn::SendRequest<hyper::Body>,
-                http::StatusCode,
-                usize,
-            ),
-            ClientError,
-        >,
+        Result<(SendRequestHttp1, http::StatusCode, usize), ClientError>,
     > {
         async move {
             if limit == 0 {
@@ -429,7 +590,7 @@ impl Client {
                     (send_request, None)
                 } else {
                     let addr = self.dns.lookup(&url, rng).await?;
-                    (self.client(addr, &url).await?, Some(send_request))
+                    (self.client_http1(addr, &url).await?, Some(send_request))
                 };
 
             while futures::future::poll_fn(|ctx| send_request.poll_ready(ctx))
@@ -437,7 +598,7 @@ impl Client {
                 .is_err()
             {
                 let addr = self.dns.lookup(&url, rng).await?;
-                send_request = self.client(addr, &url).await?;
+                send_request = self.client_http1(addr, &url).await?;
             }
 
             let mut request = self.request(&url)?;
@@ -452,8 +613,10 @@ impl Client {
             let mut status = parts.status;
 
             let mut len_sum = 0;
-            while let Some(chunk) = stream.next().await {
-                len_sum += chunk?.len();
+            while let Some(chunk) =
+                futures::future::poll_fn(|cx| Incoming::poll_frame(Pin::new(&mut stream), cx)).await
+            {
+                len_sum += chunk?.data_ref().map(|d| d.len()).unwrap_or_default();
             }
 
             if let Some(location) = parts.headers.get("Location") {
@@ -523,36 +686,98 @@ fn is_too_many_open_files(res: &Result<RequestResult, ClientError>) -> bool {
         .unwrap_or(false)
 }
 
+async fn setup_http2(client: &Client) -> (ConnectionTime, ClientStateHttp2) {
+    let mut rng = StdRng::from_entropy();
+    let url = client.url_generator.generate(&mut rng).unwrap();
+    let (connection_time, send_request) = client.connect_http2(&url, &mut rng).await.unwrap();
+
+    let client_state = ClientStateHttp2 { rng, send_request };
+
+    (connection_time, client_state)
+}
+
+fn set_connection_time<E>(res: &mut Result<RequestResult, E>, connection_time: ConnectionTime) {
+    if let Ok(res) = res {
+        res.connection_time = Some(connection_time);
+    }
+}
+
+fn set_start_latency_correction<E>(
+    res: &mut Result<RequestResult, E>,
+    start_latency_correction: std::time::Instant,
+) {
+    if let Ok(res) = res {
+        res.start_latency_correction = Some(start_latency_correction);
+    }
+}
+
 /// Run n tasks by m workers
 pub async fn work(
     client: Client,
     report_tx: flume::Sender<Result<RequestResult, ClientError>>,
     n_tasks: usize,
-    n_workers: usize,
+    n_connections: usize,
+    n_http2_parallel: usize,
 ) {
     use std::sync::atomic::{AtomicUsize, Ordering};
     let counter = Arc::new(AtomicUsize::new(0));
 
     let client = Arc::new(client);
 
-    let futures = (0..n_workers)
-        .map(|_| {
-            let report_tx = report_tx.clone();
-            let counter = counter.clone();
-            let client = client.clone();
-            tokio::spawn(async move {
-                let mut client_state = ClientState::default();
-                while counter.fetch_add(1, Ordering::Relaxed) < n_tasks {
-                    let res = client.work(&mut client_state).await;
-                    let is_cancel = is_too_many_open_files(&res);
-                    report_tx.send_async(res).await.unwrap();
-                    if is_cancel {
-                        break;
+    let futures = if client.is_http2() {
+        (0..n_connections)
+            .map(|_| {
+                let report_tx = report_tx.clone();
+                let counter = counter.clone();
+                let client = client.clone();
+                tokio::spawn(async move {
+                    let (connection_time, client_state) = setup_http2(&client).await;
+                    let futures = (0..n_http2_parallel)
+                        .map(move |_| {
+                            let report_tx = report_tx.clone();
+                            let counter = counter.clone();
+                            let client = client.clone();
+
+                            let mut client_state = client_state.clone();
+                            tokio::spawn(async move {
+                                while counter.fetch_add(1, Ordering::Relaxed) < n_tasks {
+                                    let mut res = client.work_http2(&mut client_state).await;
+                                    let is_cancel = is_too_many_open_files(&res);
+                                    set_connection_time(&mut res, connection_time);
+                                    report_tx.send_async(res).await.unwrap();
+                                    if is_cancel {
+                                        break;
+                                    }
+                                }
+                            })
+                        })
+                        .collect::<Vec<_>>();
+                    for f in futures {
+                        let _ = f.await;
                     }
-                }
+                })
             })
-        })
-        .collect::<Vec<_>>();
+            .collect::<Vec<_>>()
+    } else {
+        (0..n_connections)
+            .map(|_| {
+                let report_tx = report_tx.clone();
+                let counter = counter.clone();
+                let client = client.clone();
+                tokio::spawn(async move {
+                    let mut client_state = ClientStateHttp1::default();
+                    while counter.fetch_add(1, Ordering::Relaxed) < n_tasks {
+                        let res = client.work_http1(&mut client_state).await;
+                        let is_cancel = is_too_many_open_files(&res);
+                        report_tx.send_async(res).await.unwrap();
+                        if is_cancel {
+                            break;
+                        }
+                    }
+                })
+            })
+            .collect::<Vec<_>>()
+    };
 
     for f in futures {
         let _ = f.await;
@@ -565,7 +790,8 @@ pub async fn work_with_qps(
     report_tx: flume::Sender<Result<RequestResult, ClientError>>,
     query_limit: QueryLimit,
     n_tasks: usize,
-    n_workers: usize,
+    n_connections: usize,
+    n_http2_parallel: usize,
 ) {
     let (tx, rx) = flume::unbounded();
 
@@ -608,24 +834,59 @@ pub async fn work_with_qps(
 
     let client = Arc::new(client);
 
-    let futures = (0..n_workers)
-        .map(|_| {
-            let report_tx = report_tx.clone();
-            let rx = rx.clone();
-            let client = client.clone();
-            tokio::spawn(async move {
-                let mut client_state = ClientState::default();
-                while let Ok(()) = rx.recv_async().await {
-                    let res = client.work(&mut client_state).await;
-                    let is_cancel = is_too_many_open_files(&res);
-                    report_tx.send_async(res).await.unwrap();
-                    if is_cancel {
-                        break;
+    let futures = if client.is_http2() {
+        (0..n_connections)
+            .map(|_| {
+                let report_tx = report_tx.clone();
+                let rx = rx.clone();
+                let client = client.clone();
+                tokio::spawn(async move {
+                    let (connection_time, client_state) = setup_http2(&client).await;
+                    let futures = (0..n_http2_parallel)
+                        .map(|_| {
+                            let report_tx = report_tx.clone();
+                            let rx = rx.clone();
+                            let client = client.clone();
+                            let mut client_state = client_state.clone();
+                            tokio::spawn(async move {
+                                while let Ok(()) = rx.recv_async().await {
+                                    let mut res = client.work_http2(&mut client_state).await;
+                                    let is_cancel = is_too_many_open_files(&res);
+                                    set_connection_time(&mut res, connection_time);
+                                    report_tx.send_async(res).await.unwrap();
+                                    if is_cancel {
+                                        break;
+                                    }
+                                }
+                            })
+                        })
+                        .collect::<Vec<_>>();
+                    for f in futures {
+                        let _ = f.await;
                     }
-                }
+                })
             })
-        })
-        .collect::<Vec<_>>();
+            .collect::<Vec<_>>()
+    } else {
+        (0..n_connections)
+            .map(|_| {
+                let report_tx = report_tx.clone();
+                let rx = rx.clone();
+                let client = client.clone();
+                tokio::spawn(async move {
+                    let mut client_state = ClientStateHttp1::default();
+                    while let Ok(()) = rx.recv_async().await {
+                        let res = client.work_http1(&mut client_state).await;
+                        let is_cancel = is_too_many_open_files(&res);
+                        report_tx.send_async(res).await.unwrap();
+                        if is_cancel {
+                            break;
+                        }
+                    }
+                })
+            })
+            .collect::<Vec<_>>()
+    };
 
     for f in futures {
         let _ = f.await;
@@ -638,7 +899,8 @@ pub async fn work_with_qps_latency_correction(
     report_tx: flume::Sender<Result<RequestResult, ClientError>>,
     query_limit: QueryLimit,
     n_tasks: usize,
-    n_workers: usize,
+    n_connections: usize,
+    n_http2_parallel: usize,
 ) {
     let (tx, rx) = flume::unbounded();
 
@@ -683,29 +945,61 @@ pub async fn work_with_qps_latency_correction(
 
     let client = Arc::new(client);
 
-    let futures = (0..n_workers)
-        .map(|_| {
-            let client = client.clone();
-            let mut client_state = ClientState::default();
-            let report_tx = report_tx.clone();
-            let rx = rx.clone();
-            tokio::spawn(async move {
-                while let Ok(start) = rx.recv_async().await {
-                    let mut res = client.work(&mut client_state).await;
-
-                    if let Ok(request_result) = &mut res {
-                        request_result.start_latency_correction = Some(start);
+    let futures = if client.is_http2() {
+        (0..n_connections)
+            .map(|_| {
+                let report_tx = report_tx.clone();
+                let rx = rx.clone();
+                let client = client.clone();
+                tokio::spawn(async move {
+                    let (connection_time, client_state) = setup_http2(&client).await;
+                    let futures = (0..n_http2_parallel)
+                        .map(|_| {
+                            let report_tx = report_tx.clone();
+                            let rx = rx.clone();
+                            let client = client.clone();
+                            let mut client_state = client_state.clone();
+                            tokio::spawn(async move {
+                                while let Ok(start) = rx.recv_async().await {
+                                    let mut res = client.work_http2(&mut client_state).await;
+                                    let is_cancel = is_too_many_open_files(&res);
+                                    set_connection_time(&mut res, connection_time);
+                                    set_start_latency_correction(&mut res, start);
+                                    report_tx.send_async(res).await.unwrap();
+                                    if is_cancel {
+                                        break;
+                                    }
+                                }
+                            })
+                        })
+                        .collect::<Vec<_>>();
+                    for f in futures {
+                        let _ = f.await;
                     }
-
-                    let is_cancel = is_too_many_open_files(&res);
-                    report_tx.send_async(res).await.unwrap();
-                    if is_cancel {
-                        break;
+                })
+            })
+            .collect::<Vec<_>>()
+    } else {
+        (0..n_connections)
+            .map(|_| {
+                let client = client.clone();
+                let mut client_state = ClientStateHttp1::default();
+                let report_tx = report_tx.clone();
+                let rx = rx.clone();
+                tokio::spawn(async move {
+                    while let Ok(start) = rx.recv_async().await {
+                        let mut res = client.work_http1(&mut client_state).await;
+                        set_start_latency_correction(&mut res, start);
+                        let is_cancel = is_too_many_open_files(&res);
+                        report_tx.send_async(res).await.unwrap();
+                        if is_cancel {
+                            break;
+                        }
                     }
-                }
+                })
             })
-        })
-        .collect::<Vec<_>>();
+            .collect::<Vec<_>>()
+    };
 
     for f in futures {
         let _ = f.await;
@@ -717,31 +1011,70 @@ pub async fn work_until(
     client: Client,
     report_tx: flume::Sender<Result<RequestResult, ClientError>>,
     dead_line: std::time::Instant,
-    n_workers: usize,
+    n_connections: usize,
+    n_http2_parallel: usize,
 ) {
     let client = Arc::new(client);
-    let futures = (0..n_workers)
-        .map(|_| {
-            let client = client.clone();
-            let report_tx = report_tx.clone();
-            let mut client_state = ClientState::default();
-            tokio::spawn(async move {
-                loop {
-                    let res = client.work(&mut client_state).await;
-                    let is_cancel = is_too_many_open_files(&res);
-                    report_tx.send_async(res).await.unwrap();
-                    if is_cancel {
-                        break;
+    if client.is_http2() {
+        let futures = (0..n_connections)
+            .map(|_| {
+                let client = client.clone();
+                let report_tx = report_tx.clone();
+                tokio::spawn(async move {
+                    let (connection_time, client_state) = setup_http2(&client).await;
+
+                    let futures = (0..n_http2_parallel)
+                        .map(|_| {
+                            let client = client.clone();
+                            let report_tx = report_tx.clone();
+                            let mut client_state = client_state.clone();
+                            tokio::spawn(async move {
+                                loop {
+                                    let mut res = client.work_http2(&mut client_state).await;
+                                    let is_cancel = is_too_many_open_files(&res);
+                                    set_connection_time(&mut res, connection_time);
+                                    report_tx.send_async(res).await.unwrap();
+                                    if is_cancel {
+                                        break;
+                                    }
+                                }
+                            })
+                        })
+                        .collect::<Vec<_>>();
+
+                    tokio::time::sleep_until(dead_line.into()).await;
+                    for f in futures {
+                        f.abort();
                     }
-                }
+                })
             })
-        })
-        .collect::<Vec<_>>();
-
-    tokio::time::sleep_until(dead_line.into()).await;
-    for f in futures {
-        f.abort();
-    }
+            .collect::<Vec<_>>();
+        for f in futures {
+            let _ = f.await;
+        }
+    } else {
+        let futures = (0..n_connections)
+            .map(|_| {
+                let client = client.clone();
+                let report_tx = report_tx.clone();
+                let mut client_state = ClientStateHttp1::default();
+                tokio::spawn(async move {
+                    loop {
+                        let res = client.work_http1(&mut client_state).await;
+                        let is_cancel = is_too_many_open_files(&res);
+                        report_tx.send_async(res).await.unwrap();
+                        if is_cancel {
+                            break;
+                        }
+                    }
+                })
+            })
+            .collect::<Vec<_>>();
+        tokio::time::sleep_until(dead_line.into()).await;
+        for f in futures {
+            f.abort();
+        }
+    };
 }
 
 /// Run until dead_line by n workers limit to qps works in a second
@@ -751,7 +1084,8 @@ pub async fn work_until_with_qps(
     query_limit: QueryLimit,
     start: std::time::Instant,
     dead_line: std::time::Instant,
-    n_workers: usize,
+    n_connections: usize,
+    n_http2_parallel: usize,
 ) {
     let rx = match query_limit {
         QueryLimit::Qps(qps) => {
@@ -795,28 +1129,69 @@ pub async fn work_until_with_qps(
 
     let client = Arc::new(client);
 
-    let futures = (0..n_workers)
-        .map(|_| {
-            let client = client.clone();
-            let mut client_state = ClientState::default();
-            let report_tx = report_tx.clone();
-            let rx = rx.clone();
-            tokio::spawn(async move {
-                while let Ok(()) = rx.recv_async().await {
-                    let res = client.work(&mut client_state).await;
-                    let is_cancel = is_too_many_open_files(&res);
-                    report_tx.send_async(res).await.unwrap();
-                    if is_cancel {
-                        break;
+    if client.is_http2() {
+        let futures = (0..n_connections)
+            .map(|_| {
+                let client = client.clone();
+                let report_tx = report_tx.clone();
+                let rx = rx.clone();
+                tokio::spawn(async move {
+                    let (connection_time, client_state) = setup_http2(&client).await;
+
+                    let futures = (0..n_http2_parallel)
+                        .map(|_| {
+                            let client = client.clone();
+                            let report_tx = report_tx.clone();
+                            let rx = rx.clone();
+                            let mut client_state = client_state.clone();
+                            tokio::spawn(async move {
+                                while let Ok(()) = rx.recv_async().await {
+                                    let mut res = client.work_http2(&mut client_state).await;
+                                    let is_cancel = is_too_many_open_files(&res);
+                                    set_connection_time(&mut res, connection_time);
+                                    report_tx.send_async(res).await.unwrap();
+                                    if is_cancel {
+                                        break;
+                                    }
+                                }
+                            })
+                        })
+                        .collect::<Vec<_>>();
+                    tokio::time::sleep_until(dead_line.into()).await;
+                    for f in futures {
+                        f.abort();
                     }
-                }
+                })
             })
-        })
-        .collect::<Vec<_>>();
+            .collect::<Vec<_>>();
 
-    tokio::time::sleep_until(dead_line.into()).await;
-    for f in futures {
-        f.abort();
+        for f in futures {
+            let _ = f.await;
+        }
+    } else {
+        let futures = (0..n_connections)
+            .map(|_| {
+                let client = client.clone();
+                let mut client_state = ClientStateHttp1::default();
+                let report_tx = report_tx.clone();
+                let rx = rx.clone();
+                tokio::spawn(async move {
+                    while let Ok(()) = rx.recv_async().await {
+                        let res = client.work_http1(&mut client_state).await;
+                        let is_cancel = is_too_many_open_files(&res);
+                        report_tx.send_async(res).await.unwrap();
+                        if is_cancel {
+                            break;
+                        }
+                    }
+                })
+            })
+            .collect::<Vec<_>>();
+
+        tokio::time::sleep_until(dead_line.into()).await;
+        for f in futures {
+            f.abort();
+        }
     }
 }
 
@@ -827,7 +1202,8 @@ pub async fn work_until_with_qps_latency_correction(
     query_limit: QueryLimit,
     start: std::time::Instant,
     dead_line: std::time::Instant,
-    n_workers: usize,
+    n_connections: usize,
+    n_http2_parallel: usize,
 ) {
     let (tx, rx) = flume::unbounded();
     match query_limit {
@@ -870,32 +1246,69 @@ pub async fn work_until_with_qps_latency_correction(
 
     let client = Arc::new(client);
 
-    let futures = (0..n_workers)
-        .map(|_| {
-            let client = client.clone();
-            let mut client_state = ClientState::default();
-            let report_tx = report_tx.clone();
-            let rx = rx.clone();
-            tokio::spawn(async move {
-                while let Ok(start) = rx.recv_async().await {
-                    let mut res = client.work(&mut client_state).await;
-
-                    if let Ok(request_result) = &mut res {
-                        request_result.start_latency_correction = Some(start);
+    if client.is_http2() {
+        let futures = (0..n_connections)
+            .map(|_| {
+                let client = client.clone();
+                let report_tx = report_tx.clone();
+                let rx = rx.clone();
+                tokio::spawn(async move {
+                    let (connection_time, client_state) = setup_http2(&client).await;
+                    let futures = (0..n_http2_parallel)
+                        .map(|_| {
+                            let client = client.clone();
+                            let report_tx = report_tx.clone();
+                            let rx = rx.clone();
+                            let mut client_state = client_state.clone();
+                            tokio::spawn(async move {
+                                while let Ok(start) = rx.recv_async().await {
+                                    let mut res = client.work_http2(&mut client_state).await;
+                                    set_start_latency_correction(&mut res, start);
+                                    set_connection_time(&mut res, connection_time);
+                                    let is_cancel = is_too_many_open_files(&res);
+                                    report_tx.send_async(res).await.unwrap();
+                                    if is_cancel {
+                                        break;
+                                    }
+                                }
+                            })
+                        })
+                        .collect::<Vec<_>>();
+                    tokio::time::sleep_until(dead_line.into()).await;
+                    for f in futures {
+                        f.abort();
                     }
+                })
+            })
+            .collect::<Vec<_>>();
 
-                    let is_cancel = is_too_many_open_files(&res);
-                    report_tx.send_async(res).await.unwrap();
-                    if is_cancel {
-                        break;
+        for f in futures {
+            let _ = f.await;
+        }
+    } else {
+        let futures = (0..n_connections)
+            .map(|_| {
+                let client = client.clone();
+                let mut client_state = ClientStateHttp1::default();
+                let report_tx = report_tx.clone();
+                let rx = rx.clone();
+                tokio::spawn(async move {
+                    while let Ok(start) = rx.recv_async().await {
+                        let mut res = client.work_http1(&mut client_state).await;
+                        set_start_latency_correction(&mut res, start);
+                        let is_cancel = is_too_many_open_files(&res);
+                        report_tx.send_async(res).await.unwrap();
+                        if is_cancel {
+                            break;
+                        }
                     }
-                }
+                })
             })
-        })
-        .collect::<Vec<_>>();
+            .collect::<Vec<_>>();
 
-    tokio::time::sleep_until(dead_line.into()).await;
-    for f in futures {
-        f.abort();
+        tokio::time::sleep_until(dead_line.into()).await;
+        for f in futures {
+            f.abort();
+        }
     }
 }
diff --git a/src/main.rs b/src/main.rs
index 92f0ae7..59f0808 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -16,6 +16,7 @@ mod histogram;
 mod monitor;
 mod printer;
 mod timescale;
+mod tokiort;
 mod url_generator;
 
 #[cfg(unix)]
@@ -36,11 +37,17 @@ struct Opts {
     )]
     n_requests: usize,
     #[clap(
-        help = "Number of workers to run concurrently. You may should increase limit to number of open files for larger `-c`.",
+        help = "Number of connections to run concurrently. You may should increase limit to number of open files for larger `-c`.",
         short = 'c',
         default_value = "50"
     )]
-    n_workers: usize,
+    n_connections: usize,
+    #[clap(
+        help = "Number of parallel requests to send on HTTP/2. `oha` will run c * p concurrent workers in total.",
+        short = 'p',
+        default_value = "1"
+    )]
+    n_http2_parallel: usize,
     #[clap(
         help = "Duration of application to send requests. If duration is specified, n is ignored.
 Examples: -z 10s -z 3m.",
@@ -115,19 +122,21 @@ Note: If qps is specified, burst will be ignored",
         long = "http-version"
     )]
     http_version: Option<String>,
+    #[clap(help = "Use HTTP/2. Shorthand for --http-version=2", long = "http2")]
+    http2: bool,
     #[clap(help = "HTTP Host header", long = "host")]
     host: Option<String>,
     #[clap(help = "Disable compression.", long = "disable-compression")]
     disable_compression: bool,
     #[clap(
-        help = "Limit for number of Redirect. Set 0 for no redirection.",
+        help = "Limit for number of Redirect. Set 0 for no redirection. Redirection isn't supported for HTTP/2.",
         default_value = "10",
         short = 'r',
         long = "redirect"
     )]
     redirect: usize,
     #[clap(
-        help = "Disable keep-alive, prevents re-use of TCP connections between different HTTP requests.",
+        help = "Disable keep-alive, prevents re-use of TCP connections between different HTTP requests. This isn't supported for HTTP/2.",
         long = "disable-keepalive"
     )]
     disable_keepalive: bool,
@@ -201,17 +210,18 @@ impl FromStr for ConnectToEntry {
 async fn main() -> anyhow::Result<()> {
     let mut opts: Opts = Opts::parse();
 
-    let http_version: http::Version = if let Some(http_version) = opts.http_version {
-        match http_version.trim() {
+    let http_version: http::Version = match (opts.http2, opts.http_version) {
+        (true, Some(_)) => anyhow::bail!("--http2 and --http-version are exclusive"),
+        (true, None) => http::Version::HTTP_2,
+        (false, Some(http_version)) => match http_version.trim() {
             "0.9" => http::Version::HTTP_09,
             "1.0" => http::Version::HTTP_10,
             "1.1" => http::Version::HTTP_11,
-            "2.0" | "2" => anyhow::bail!("HTTP/2 is not supported yet."),
+            "2.0" | "2" => http::Version::HTTP_2,
             "3.0" | "3" => anyhow::bail!("HTTP/3 is not supported yet."),
-            _ => anyhow::bail!("Unknown HTTP version. Valid versions are 0.9, 1.0, 1.1."),
-        }
-    } else {
-        http::Version::HTTP_11
+            _ => anyhow::bail!("Unknown HTTP version. Valid versions are 0.9, 1.0, 1.1, 2."),
+        },
+        (false, None) => http::Version::HTTP_11,
     };
 
     let url_generator = if opts.rand_regex_url {
@@ -449,8 +459,14 @@ async fn main() -> anyhow::Result<()> {
         match opts.query_per_second {
             Some(0) | None => match opts.burst_duration {
                 None => {
-                    client::work_until(client, result_tx, start + duration.into(), opts.n_workers)
-                        .await
+                    client::work_until(
+                        client,
+                        result_tx,
+                        start + duration.into(),
+                        opts.n_connections,
+                        opts.n_http2_parallel,
+                    )
+                    .await
                 }
                 Some(burst_duration) => {
                     if opts.latency_correction {
@@ -463,7 +479,8 @@ async fn main() -> anyhow::Result<()> {
                             ),
                             start,
                             start + duration.into(),
-                            opts.n_workers,
+                            opts.n_connections,
+                            opts.n_http2_parallel,
                         )
                         .await
                     } else {
@@ -476,7 +493,8 @@ async fn main() -> anyhow::Result<()> {
                             ),
                             start,
                             start + duration.into(),
-                            opts.n_workers,
+                            opts.n_connections,
+                            opts.n_http2_parallel,
                         )
                         .await
                     }
@@ -490,7 +508,8 @@ async fn main() -> anyhow::Result<()> {
                         client::QueryLimit::Qps(qps),
                         start,
                         start + duration.into(),
-                        opts.n_workers,
+                        opts.n_connections,
+                        opts.n_http2_parallel,
                     )
                     .await
                 } else {
@@ -500,7 +519,8 @@ async fn main() -> anyhow::Result<()> {
                         client::QueryLimit::Qps(qps),
                         start,
                         start + duration.into(),
-                        opts.n_workers,
+                        opts.n_connections,
+                        opts.n_http2_parallel,
                     )
                     .await
                 }
@@ -509,7 +529,16 @@ async fn main() -> anyhow::Result<()> {
     } else {
         match opts.query_per_second {
             Some(0) | None => match opts.burst_duration {
-                None => client::work(client, result_tx, opts.n_requests, opts.n_workers).await,
+                None => {
+                    client::work(
+                        client,
+                        result_tx,
+                        opts.n_requests,
+                        opts.n_connections,
+                        opts.n_http2_parallel,
+                    )
+                    .await
+                }
                 Some(burst_duration) => {
                     if opts.latency_correction {
                         client::work_with_qps_latency_correction(
@@ -520,7 +549,8 @@ async fn main() -> anyhow::Result<()> {
                                 opts.burst_requests.unwrap_or(1),
                             ),
                             opts.n_requests,
-                            opts.n_workers,
+                            opts.n_connections,
+                            opts.n_http2_parallel,
                         )
                         .await
                     } else {
@@ -532,7 +562,8 @@ async fn main() -> anyhow::Result<()> {
                                 opts.burst_requests.unwrap_or(1),
                             ),
                             opts.n_requests,
-                            opts.n_workers,
+                            opts.n_connections,
+                            opts.n_http2_parallel,
                         )
                         .await
                     }
@@ -545,7 +576,8 @@ async fn main() -> anyhow::Result<()> {
                         result_tx,
                         client::QueryLimit::Qps(qps),
                         opts.n_requests,
-                        opts.n_workers,
+                        opts.n_connections,
+                        opts.n_http2_parallel,
                     )
                     .await
                 } else {
@@ -554,7 +586,8 @@ async fn main() -> anyhow::Result<()> {
                         result_tx,
                         client::QueryLimit::Qps(qps),
                         opts.n_requests,
-                        opts.n_workers,
+                        opts.n_connections,
+                        opts.n_http2_parallel,
                     )
                     .await
                 }
diff --git a/src/printer.rs b/src/printer.rs
index e02e24a..5dc66f6 100644
--- a/src/printer.rs
+++ b/src/printer.rs
@@ -682,7 +682,7 @@ fn calcuate_connection_times_base<E>(
 ) -> Vec<(Instant, ConnectionTime)> {
     res.iter()
         .filter_map(|r| r.as_ref().ok())
-        .filter_map(|r| r.connection_time.clone().map(|c| (r.start, c)))
+        .filter_map(|r| r.connection_time.map(|c| (r.start, c)))
         .collect()
 }
 
diff --git a/src/tokiort.rs b/src/tokiort.rs
new file mode 100644
index 0000000..c5adc75
--- /dev/null
+++ b/src/tokiort.rs
@@ -0,0 +1,259 @@
+// This entire codes are copied from https://github.com/hyperium/hyper/blob/f9f65b7aa67fa3ec0267fe015945973726285bc2/benches/support/tokiort.rs
+/*
+Copyright (c) 2014-2021 Sean McArthur
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
+*/
+
+#![allow(dead_code)]
+//! Various runtimes for hyper
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+    time::{Duration, Instant},
+};
+
+use futures_util::Future;
+use hyper::rt::{Sleep, Timer};
+use pin_project_lite::pin_project;
+
+#[derive(Clone)]
+/// An Executor that uses the tokio runtime.
+pub struct TokioExecutor;
+
+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);
+    }
+}
+
+/// A Timer that uses the tokio runtime.
+
+#[derive(Clone, Debug)]
+pub struct TokioTimer;
+
+impl Timer for TokioTimer {
+    fn sleep(&self, duration: Duration) -> Pin<Box<dyn Sleep>> {
+        Box::pin(TokioSleep {
+            inner: tokio::time::sleep(duration),
+        })
+    }
+
+    fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn Sleep>> {
+        Box::pin(TokioSleep {
+            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)
+        }
+    }
+}
+
+struct TokioTimeout<T> {
+    inner: Pin<Box<tokio::time::Timeout<T>>>,
+}
+
+impl<T> Future for TokioTimeout<T>
+where
+    T: Future,
+{
+    type Output = Result<T::Output, tokio::time::error::Elapsed>;
+
+    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
+        self.inner.as_mut().poll(context)
+    }
+}
+
+// Use TokioSleep to get tokio::time::Sleep to implement Unpin.
+// see https://docs.rs/tokio/latest/tokio/time/struct.Sleep.html
+pin_project! {
+    pub(crate) struct TokioSleep {
+        #[pin]
+        pub(crate) inner: tokio::time::Sleep,
+    }
+}
+
+impl Future for TokioSleep {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().inner.poll(cx)
+    }
+}
+
+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)
+    }
+}


Test patch:
diff --git a/tests/tests.rs b/tests/tests.rs
index 4317ea3..320b3d2 100644
--- a/tests/tests.rs
+++ b/tests/tests.rs
@@ -11,8 +11,8 @@ use axum::{
     routing::{any, get},
     Router,
 };
-use http::{HeaderMap, Response};
-use hyper::{
+use http::{HeaderMap, Request, Response};
+use hyper14::{
     body::to_bytes,
     server::conn::AddrIncoming,
     service::{make_service_fn, service_fn},
@@ -22,14 +22,14 @@ use hyper::{
 // Port 5111- is reserved for testing
 static PORT: AtomicU16 = AtomicU16::new(5111);
 
-fn bind_port() -> (hyper::server::Builder<AddrIncoming>, u16) {
+fn bind_port() -> (hyper14::server::Builder<AddrIncoming>, u16) {
     let port = PORT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
     let addr = SocketAddr::new("127.0.0.1".parse().unwrap(), port);
 
     (axum::Server::bind(&addr), port)
 }
 
-fn bind_port_ipv6() -> (hyper::server::Builder<AddrIncoming>, u16) {
+fn bind_port_ipv6() -> (hyper14::server::Builder<AddrIncoming>, u16) {
     let port = PORT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
     let addr = SocketAddr::new(std::net::IpAddr::V6(Ipv6Addr::LOCALHOST), port);
 
@@ -409,6 +409,39 @@ async fn test_enable_compression_default() {
     assert!(accept_encoding.contains(&"br"));
 }
 
+async fn get_http_version(args: &[&str]) -> http::Version {
+    let (tx, rx) = flume::unbounded();
+
+    let app = Router::new().route(
+        "/",
+        get(|req: Request<hyper14::Body>| async move {
+            tx.send(req.version()).unwrap();
+            "Hello World"
+        }),
+    );
+
+    let (server, port) = bind_port();
+
+    tokio::spawn(async {
+        server.serve(app.into_make_service()).await.unwrap();
+    });
+
+    let args: Vec<String> = args.iter().map(|s| s.to_string()).collect();
+    tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("oha")
+            .unwrap()
+            .args(["-n", "1", "--no-tui"])
+            .args(args)
+            .arg(format!("http://127.0.0.1:{port}"))
+            .assert()
+            .success();
+    })
+    .await
+    .unwrap();
+
+    rx.try_recv().unwrap()
+}
+
 #[tokio::test]
 async fn test_setting_custom_header() {
     let header = get_header_body(&["-H", "foo: bar", "--"]).await.0;
@@ -595,3 +628,14 @@ async fn test_ipv6() {
 async fn test_query_limit() {
     assert_eq!(burst_10_req_delay_2s_rate_4(10, &[],).await, 10);
 }
+
+#[tokio::test]
+async fn test_http2() {
+    // FIXME: Ideally, we need to test all of above with HTTP/2
+    assert_eq!(get_http_version(&[]).await, http::Version::HTTP_11);
+    assert_eq!(get_http_version(&["--http2"]).await, http::Version::HTTP_2);
+    assert_eq!(
+        get_http_version(&["--http-version", "2"]).await,
+        http::Version::HTTP_2
+    );
+}