UDP/QUIC/Http3 quiche::h3 Client/Connector integration

.github/workflows/build.yml

@@ -7,7 +7,7 @@ jobs:
     strategy:
       matrix:
         # nightly, MSRV, and latest stable
-        toolchain: [nightly, 1.72, 1.82.0]
+        toolchain: [nightly, 1.74, 1.82.0]
     runs-on: ubuntu-latest
     # Only run on "pull_request" event for external PRs. This is to avoid
     # duplicate builds for PRs created from internal branches.

pingora-core/Cargo.toml

@@ -68,6 +68,8 @@ zstd = "0"
 httpdate = "1"
 x509-parser = { version = "0.16.0", optional = true }
 ouroboros = { version = "0.18.4", optional = true }
+quiche = { git = 'https://github.com/cloudflare/quiche.git', rev = "5d2031ca", default-features = false, optional = true }
+ring = { version = "0.17.8", optional = true }
 
 [target.'cfg(unix)'.dependencies]
 daemonize = "0.5.0"
@@ -83,17 +85,21 @@ reqwest = { version = "0.11", features = [
 ], default-features = false }
 hyper = "0.14"
 rstest = "0.23.0"
+h3i = { git = 'https://github.com/cloudflare/quiche.git', rev = "5d2031ca" }
 
 [target.'cfg(unix)'.dev-dependencies]
 hyperlocal = "0.8"
 jemallocator = "0.5"
+histogram = "0.11"
+textplots = "0.8"
 
 [features]
-default = []
+default = ["boringssl"]
 openssl = ["pingora-openssl", "openssl_derived"]
-boringssl = ["pingora-boringssl", "openssl_derived"]
+boringssl = ["pingora-boringssl", "openssl_derived", "quic-boringssl"]
 rustls = ["pingora-rustls", "any_tls", "dep:x509-parser", "ouroboros"]
 patched_http1 = ["pingora-http/patched_http1"]
 openssl_derived = ["any_tls"]
 any_tls = []
 sentry = ["dep:sentry"]
+quic-boringssl = ["dep:quiche", "dep:ring", "pingora-boringssl", "quiche/boringssl-boring-crate"]

pingora-core/src/apps/mod.rs

@@ -23,8 +23,9 @@ use log::{debug, error};
 use std::future::poll_fn;
 use std::sync::Arc;
 
-use crate::protocols::http::v2::server;
-use crate::protocols::http::ServerSession;
+use crate::protocols::http::v2::server as h2_server;
+use crate::protocols::http::v3::server as h3_server;
+use crate::protocols::http::{HttpVersion, ServerSession};
 use crate::protocols::Digest;
 use crate::protocols::Stream;
 use crate::protocols::ALPN;
@@ -61,8 +62,8 @@ pub trait ServerApp {
 #[derive(Default)]
 /// HTTP Server options that control how the server handles some transport types.
 pub struct HttpServerOptions {
-    /// Use HTTP/2 for plaintext.
-    pub h2c: bool,
+    /// HTTP version to use.
+    pub http_version: HttpVersion,
 }
 
 /// This trait defines the interface of an HTTP application.
@@ -84,7 +85,15 @@ pub trait HttpServerApp {
     /// every time a new HTTP/2 **connection** needs to be established.
     ///
     /// A `None` means to use the built-in default options. See [`server::H2Options`] for more details.
-    fn h2_options(&self) -> Option<server::H2Options> {
+    fn h2_options(&self) -> Option<h2_server::H2Options> {
+        None
+    }
+
+    /// Provide options on how HTTP/3 connection should be established. This function will be called
+    /// every time a new HTTP/3 **connection** needs to be established.
+    ///
+    /// A `None` means to use the built-in default options. See [`server::H2Options`] for more details.
+    fn h3_options(&self) -> Option<&h3_server::Http3Options> {
         None
     }
 
@@ -109,10 +118,17 @@ where
         mut stream: Stream,
         shutdown: &ShutdownWatch,
     ) -> Option<Stream> {
-        let mut h2c = self.server_options().as_ref().map_or(false, |o| o.h2c);
+        let mut http_version = self
+            .server_options()
+            .as_ref()
+            .map_or(HttpVersion::V1, |o| o.http_version);
+
+        if stream.quic_connection_state().is_some() {
+            http_version = HttpVersion::V3;
+        }
 
         // try to read h2 preface
-        if h2c {
+        if matches!(http_version, HttpVersion::V2) {
             let mut buf = [0u8; H2_PREFACE.len()];
             let peeked = stream
                 .try_peek(&mut buf)
@@ -126,10 +142,30 @@ where
             // not all streams support peeking
             if peeked {
                 // turn off h2c (use h1) if h2 preface doesn't exist
-                h2c = buf == H2_PREFACE;
+                http_version = match buf == H2_PREFACE {
+                    true => HttpVersion::V2,
+                    false => HttpVersion::V1,
+                };
             }
         }
-        if h2c || matches!(stream.selected_alpn_proto(), Some(ALPN::H2)) {
+
+        // TODO: logic for Http3 to Http2/1 fallback. Requires Http2/1 listener being present.
+        if matches!(http_version, HttpVersion::V3)
+            && (matches!(stream.selected_alpn_proto(), Some(ALPN::H1))
+                || matches!(stream.selected_alpn_proto(), Some(ALPN::H2))
+                || matches!(stream.selected_alpn_proto(), Some(ALPN::H2H1)))
+        {
+            error!(
+                "Server is configured for {:?}. Received ALPN: {}. \
+             Fallback from Http3 to Http2/1 is currently not supported.",
+                http_version,
+                stream.selected_alpn_proto().unwrap()
+            )
+        }
+
+        if matches!(http_version, HttpVersion::V2)
+            || matches!(stream.selected_alpn_proto(), Some(ALPN::H2))
+        {
             // create a shared connection digest
             let digest = Arc::new(Digest {
                 ssl_digest: stream.get_ssl_digest(),
@@ -140,7 +176,7 @@ where
             });
 
             let h2_options = self.h2_options();
-            let h2_conn = server::handshake(stream, h2_options).await;
+            let h2_conn = h2_server::handshake(stream, h2_options).await;
             let mut h2_conn = match h2_conn {
                 Err(e) => {
                     error!("H2 handshake error {e}");
@@ -160,7 +196,7 @@ where
                             .await.map_err(|e| error!("H2 error waiting for shutdown {e}"));
                         return None;
                     }
-                    h2_stream = server::HttpSession::from_h2_conn(&mut h2_conn, digest.clone()) => h2_stream
+                    h2_stream = h2_server::HttpSession::from_h2_conn(&mut h2_conn, digest.clone()) => h2_stream
                 };
                 let h2_stream = match h2_stream {
                     Err(e) => {
@@ -178,6 +214,57 @@ where
                         .await;
                 });
             }
+        } else if matches!(http_version, HttpVersion::V3) {
+            // create a shared connection digest
+            let digest = Arc::new(Digest {
+                ssl_digest: stream.get_ssl_digest(),
+                // TODO: log h3 handshake time
+                timing_digest: stream.get_timing_digest(),
+                proxy_digest: stream.get_proxy_digest(),
+                socket_digest: stream.get_socket_digest(),
+            });
+
+            let h3_options = self.h3_options();
+            let h3_conn = h3_server::handshake(stream, h3_options).await;
+            let mut h3_conn = match h3_conn {
+                Err(e) => {
+                    error!("H3 handshake error {e}");
+                    return None;
+                }
+                Ok(c) => c,
+            };
+
+            let mut shutdown = shutdown.clone();
+            loop {
+                // this loop ends when the client decides to close the h3 conn
+                let h3_stream = tokio::select! {
+                    _ = shutdown.changed() => {
+                        match h3_conn.graceful_shutdown().await {
+                            Ok(()) => {}
+                            Err(e) => { error!("H3 error waiting for shutdown {e}") }
+                        };
+                        return None;
+                    }
+                    h3_stream = h3_server::Http3Session::from_h3_conn(&mut h3_conn, digest.clone()) => h3_stream
+                };
+
+                let h3_stream = match h3_stream {
+                    Err(e) => {
+                        // It is common for the client to just disconnect TCP without properly
+                        // closing H2. So we don't log the errors here
+                        debug!("H3 error when accepting new stream {e}");
+                        return None;
+                    }
+                    Ok(s) => s?, // None means the connection is ready to be closed
+                };
+
+                let app = self.clone();
+                let shutdown = shutdown.clone();
+                pingora_runtime::current_handle().spawn(async move {
+                    app.process_new_http(ServerSession::new_http3(h3_stream), &shutdown)
+                        .await;
+                });
+            }
         } else {
             // No ALPN or ALPN::H1 and h2c was not configured, fallback to HTTP/1.1
             self.process_new_http(ServerSession::new_http1(stream), shutdown)

pingora-core/src/connectors/http/mod.rs

@@ -16,23 +16,30 @@
 
 use crate::connectors::ConnectorOptions;
 use crate::protocols::http::client::HttpSession;
+use crate::protocols::{UniqueID, UniqueIDType};
 use crate::upstreams::peer::Peer;
+use parking_lot::RwLock;
 use pingora_error::Result;
+use pingora_pool::PoolNode;
+use std::collections::HashMap;
 use std::time::Duration;
 
 pub mod v1;
 pub mod v2;
+pub mod v3;
 
 pub struct Connector {
     h1: v1::Connector,
     h2: v2::Connector,
+    h3: v3::Connector,
 }
 
 impl Connector {
     pub fn new(options: Option<ConnectorOptions>) -> Self {
         Connector {
             h1: v1::Connector::new(options.clone()),
-            h2: v2::Connector::new(options),
+            h2: v2::Connector::new(options.clone()),
+            h3: v3::Connector::new(options),
         }
     }
 
@@ -50,7 +57,15 @@ impl Connector {
         let h1_only = peer
             .get_peer_options()
             .map_or(true, |o| o.alpn.get_max_http_version() == 1);
-        if h1_only {
+
+        if peer.udp_http3() {
+            if let Some(h3) = self.h3.reused_http_session(peer).await? {
+                Ok((HttpSession::H3(h3), true))
+            } else {
+                let session = self.h3.new_http_session(peer).await?;
+                Ok((session, false))
+            }
+        } else if h1_only {
             let (h1, reused) = self.h1.get_http_session(peer).await?;
             Ok((HttpSession::H1(h1), reused))
         } else {
@@ -85,6 +100,7 @@ impl Connector {
         match session {
             HttpSession::H1(h1) => self.h1.release_http_session(h1, peer, idle_timeout).await,
             HttpSession::H2(h2) => self.h2.release_http_session(h2, peer, idle_timeout),
+            HttpSession::H3(h3) => self.h3.release_http_session(h3, peer, idle_timeout),
         }
     }
 
@@ -94,6 +110,52 @@ impl Connector {
     }
 }
 
+pub(crate) struct InUsePool<T: UniqueID> {
+    // TODO: use pingora hashmap to shard the lock contention
+    pools: RwLock<HashMap<u64, PoolNode<T>>>,
+}
+
+impl<T: UniqueID> InUsePool<T> {
+    pub(crate) fn new() -> Self {
+        InUsePool {
+            pools: RwLock::new(HashMap::new()),
+        }
+    }
+    pub(crate) fn insert(&self, reuse_hash: u64, conn: T) {
+        {
+            let pools = self.pools.read();
+            if let Some(pool) = pools.get(&reuse_hash) {
+                pool.insert(conn.id(), conn);
+                return;
+            }
+        } // drop read lock
+
+        let pool = PoolNode::new();
+        pool.insert(conn.id(), conn);
+        let mut pools = self.pools.write();
+        pools.insert(reuse_hash, pool);
+    }
+
+    // retrieve a `<T>` to create a new stream
+    // the caller should return the <T> to this pool if there is still
+    // capacity left for more streams
+    pub(crate) fn get(&self, reuse_hash: u64) -> Option<T> {
+        let pools = self.pools.read();
+        pools.get(&reuse_hash)?.get_any().map(|v| v.1)
+    }
+
+    // release a http stream, this functional will cause an `<T>` to be returned (if exist)
+    // the caller should update the ref and then decide where to put it (in use pool or idle)
+    pub(crate) fn release(&self, reuse_hash: u64, id: UniqueIDType) -> Option<T> {
+        let pools = self.pools.read();
+        if let Some(pool) = pools.get(&reuse_hash) {
+            pool.remove(id)
+        } else {
+            None
+        }
+    }
+}
+
 #[cfg(test)]
 #[cfg(feature = "any_tls")]
 mod tests {
@@ -121,8 +183,8 @@ mod tests {
         let (h2, reused) = connector.get_http_session(&peer).await.unwrap();
         assert!(!reused);
         match &h2 {
-            HttpSession::H1(_) => panic!("expect h2"),
             HttpSession::H2(h2_stream) => assert!(!h2_stream.ping_timedout()),
+            _ => panic!("expect h2"),
         }
 
         connector.release_http_session(h2, &peer, None).await;
@@ -131,8 +193,8 @@ mod tests {
         // reused this time
         assert!(reused);
         match &h2 {
-            HttpSession::H1(_) => panic!("expect h2"),
             HttpSession::H2(h2_stream) => assert!(!h2_stream.ping_timedout()),
+            _ => panic!("expect h2"),
         }
     }
 
@@ -147,7 +209,7 @@ mod tests {
             HttpSession::H1(http) => {
                 get_http(http, 200).await;
             }
-            HttpSession::H2(_) => panic!("expect h1"),
+            _ => panic!("expect h1"),
         }
         connector.release_http_session(h1, &peer, None).await;
 
@@ -156,7 +218,7 @@ mod tests {
         assert!(reused);
         match &mut h1 {
             HttpSession::H1(_) => {}
-            HttpSession::H2(_) => panic!("expect h1"),
+            _ => panic!("expect h1"),
         }
     }
 
@@ -177,7 +239,7 @@ mod tests {
             HttpSession::H1(http) => {
                 get_http(http, 200).await;
             }
-            HttpSession::H2(_) => panic!("expect h1"),
+            _ => panic!("expect h1"),
         }
         connector.release_http_session(h1, &peer, None).await;
 
@@ -189,7 +251,7 @@ mod tests {
         assert!(reused);
         match &mut h1 {
             HttpSession::H1(_) => {}
-            HttpSession::H2(_) => panic!("expect h1"),
+            _ => panic!("expect h1"),
         }
     }
 
@@ -206,7 +268,7 @@ mod tests {
             HttpSession::H1(http) => {
                 get_http(http, 200).await;
             }
-            HttpSession::H2(_) => panic!("expect h1"),
+            _ => panic!("expect h1"),
         }
         connector.release_http_session(h1, &peer, None).await;
 
@@ -216,7 +278,7 @@ mod tests {
         assert!(reused);
         match &mut h1 {
             HttpSession::H1(_) => {}
-            HttpSession::H2(_) => panic!("expect h1"),
+            _ => panic!("expect h1"),
         }
     }
 }