feat(http,web): expose config for H2 window sizes (#3926)

* feat(http,web): expose config for H2 window sizes

* changelog

actix-http/CHANGES.md

@@ -3,10 +3,13 @@
 ## Unreleased
 
 - Minimum supported Rust version (MSRV) is now 1.88.
+- Increase default HTTP/2 flow control window sizes. [#3638]
+- Expose configuration methods to improve upload throughput. [#3638]
 - Fix truncated body ending without error when connection closed abnormally. [#3067]
 - Add config/method for `TCP_NODELAY`. [#3918]
 - Do not compress 206 Partial Content responses. [#3191]
 
+[#3638]: https://github.com/actix/actix-web/issues/3638
 [#3067]: https://github.com/actix/actix-web/pull/3067
 [#3918]: https://github.com/actix/actix-web/pull/3918
 [#3191]: https://github.com/actix/actix-web/issues/3191

actix-http/src/builder.rs

@@ -5,6 +5,7 @@ use actix_service::{IntoServiceFactory, Service, ServiceFactory};
 
 use crate::{
     body::{BoxBody, MessageBody},
+    config::{DEFAULT_H2_CONN_WINDOW_SIZE, DEFAULT_H2_STREAM_WINDOW_SIZE},
     h1::{self, ExpectHandler, H1Service, UpgradeHandler},
     service::HttpService,
     ConnectCallback, Extensions, KeepAlive, Request, Response, ServiceConfigBuilder,
@@ -21,6 +22,8 @@ pub struct HttpServiceBuilder<T, S, X = ExpectHandler, U = UpgradeHandler> {
     secure: bool,
     local_addr: Option<net::SocketAddr>,
     h1_allow_half_closed: bool,
+    h2_conn_window_size: u32,
+    h2_stream_window_size: u32,
     expect: X,
     upgrade: Option<U>,
     on_connect_ext: Option<Rc<ConnectCallback<T>>>,
@@ -44,6 +47,8 @@ where
             secure: false,
             local_addr: None,
             h1_allow_half_closed: true,
+            h2_conn_window_size: DEFAULT_H2_CONN_WINDOW_SIZE,
+            h2_stream_window_size: DEFAULT_H2_STREAM_WINDOW_SIZE,
 
             // dispatcher parts
             expect: ExpectHandler,
@@ -146,6 +151,22 @@ where
         self
     }
 
+    /// Sets initial stream-level flow control window size for HTTP/2 connections.
+    ///
+    /// See [`ServiceConfigBuilder::h2_initial_window_size`] for more details.
+    pub fn h2_initial_window_size(mut self, size: u32) -> Self {
+        self.h2_stream_window_size = size;
+        self
+    }
+
+    /// Sets initial connection-level flow control window size for HTTP/2 connections.
+    ///
+    /// See [`ServiceConfigBuilder::h2_initial_connection_window_size`] for more details.
+    pub fn h2_initial_connection_window_size(mut self, size: u32) -> Self {
+        self.h2_conn_window_size = size;
+        self
+    }
+
     /// Provide service for `EXPECT: 100-Continue` support.
     ///
     /// Service get called with request that contains `EXPECT` header.
@@ -166,6 +187,8 @@ where
             secure: self.secure,
             local_addr: self.local_addr,
             h1_allow_half_closed: self.h1_allow_half_closed,
+            h2_conn_window_size: self.h2_conn_window_size,
+            h2_stream_window_size: self.h2_stream_window_size,
             expect: expect.into_factory(),
             upgrade: self.upgrade,
             on_connect_ext: self.on_connect_ext,
@@ -192,6 +215,8 @@ where
             secure: self.secure,
             local_addr: self.local_addr,
             h1_allow_half_closed: self.h1_allow_half_closed,
+            h2_conn_window_size: self.h2_conn_window_size,
+            h2_stream_window_size: self.h2_stream_window_size,
             expect: self.expect,
             upgrade: Some(upgrade.into_factory()),
             on_connect_ext: self.on_connect_ext,
@@ -229,6 +254,8 @@ where
             .secure(self.secure)
             .local_addr(self.local_addr)
             .h1_allow_half_closed(self.h1_allow_half_closed)
+            .h2_initial_window_size(self.h2_stream_window_size)
+            .h2_initial_connection_window_size(self.h2_conn_window_size)
             .build();
 
         H1Service::with_config(cfg, service.into_factory())
@@ -256,6 +283,8 @@ where
             .secure(self.secure)
             .local_addr(self.local_addr)
             .h1_allow_half_closed(self.h1_allow_half_closed)
+            .h2_initial_window_size(self.h2_stream_window_size)
+            .h2_initial_connection_window_size(self.h2_conn_window_size)
             .build();
 
         crate::h2::H2Service::with_config(cfg, service.into_factory())
@@ -280,6 +309,8 @@ where
             .secure(self.secure)
             .local_addr(self.local_addr)
             .h1_allow_half_closed(self.h1_allow_half_closed)
+            .h2_initial_window_size(self.h2_stream_window_size)
+            .h2_initial_connection_window_size(self.h2_conn_window_size)
             .build();
 
         HttpService::with_config(cfg, service.into_factory())

actix-http/src/config.rs

@@ -8,6 +8,16 @@ use bytes::BytesMut;
 
 use crate::{date::DateService, KeepAlive};
 
+/// Default HTTP/2 initial connection-level flow control window size.
+///
+/// Matches awc's defaults to avoid poor throughput on high-BDP links.
+pub(crate) const DEFAULT_H2_CONN_WINDOW_SIZE: u32 = 1024 * 1024 * 2; // 2MiB
+
+/// Default HTTP/2 initial stream-level flow control window size.
+///
+/// Matches awc's defaults to avoid poor throughput on high-BDP links.
+pub(crate) const DEFAULT_H2_STREAM_WINDOW_SIZE: u32 = 1024 * 1024; // 1MiB
+
 /// A builder for creating a [`ServiceConfig`]
 #[derive(Default, Debug)]
 pub struct ServiceConfigBuilder {
@@ -76,6 +86,28 @@ impl ServiceConfigBuilder {
         self
     }
 
+    /// Sets initial stream-level flow control window size for HTTP/2 connections.
+    ///
+    /// Higher values can improve upload performance on high-latency links at the cost of higher
+    /// worst-case memory usage per connection.
+    ///
+    /// The default value is 1MiB.
+    pub fn h2_initial_window_size(mut self, size: u32) -> Self {
+        self.inner.h2_stream_window_size = size;
+        self
+    }
+
+    /// Sets initial connection-level flow control window size for HTTP/2 connections.
+    ///
+    /// Higher values can improve upload performance on high-latency links at the cost of higher
+    /// worst-case memory usage per connection.
+    ///
+    /// The default value is 2MiB.
+    pub fn h2_initial_connection_window_size(mut self, size: u32) -> Self {
+        self.inner.h2_conn_window_size = size;
+        self
+    }
+
     /// Builds a [`ServiceConfig`] from this [`ServiceConfigBuilder`] instance
     pub fn build(self) -> ServiceConfig {
         ServiceConfig(Rc::new(self.inner))
@@ -96,6 +128,8 @@ struct Inner {
     tcp_nodelay: Option<bool>,
     date_service: DateService,
     h1_allow_half_closed: bool,
+    h2_conn_window_size: u32,
+    h2_stream_window_size: u32,
 }
 
 impl Default for Inner {
@@ -109,6 +143,8 @@ impl Default for Inner {
             tcp_nodelay: None,
             date_service: DateService::new(),
             h1_allow_half_closed: true,
+            h2_conn_window_size: DEFAULT_H2_CONN_WINDOW_SIZE,
+            h2_stream_window_size: DEFAULT_H2_STREAM_WINDOW_SIZE,
         }
     }
 }
@@ -131,6 +167,8 @@ impl ServiceConfig {
             tcp_nodelay: None,
             date_service: DateService::new(),
             h1_allow_half_closed: true,
+            h2_conn_window_size: DEFAULT_H2_CONN_WINDOW_SIZE,
+            h2_stream_window_size: DEFAULT_H2_STREAM_WINDOW_SIZE,
         }))
     }
 
@@ -195,6 +233,16 @@ impl ServiceConfig {
         self.0.tcp_nodelay
     }
 
+    /// HTTP/2 initial stream-level flow control window size (in bytes).
+    pub fn h2_initial_window_size(&self) -> u32 {
+        self.0.h2_stream_window_size
+    }
+
+    /// HTTP/2 initial connection-level flow control window size (in bytes).
+    pub fn h2_initial_connection_window_size(&self) -> u32 {
+        self.0.h2_conn_window_size
+    }
+
     pub(crate) fn now(&self) -> Instant {
         self.0.date_service.now()
     }

actix-http/src/h2/mod.rs

@@ -11,7 +11,7 @@ use actix_rt::time::{sleep_until, Sleep};
 use bytes::Bytes;
 use futures_core::{ready, Stream};
 use h2::{
-    server::{handshake, Connection, Handshake},
+    server::{Builder, Connection, Handshake},
     RecvStream,
 };
 
@@ -61,8 +61,13 @@ pub(crate) fn handshake_with_timeout<T>(io: T, config: &ServiceConfig) -> Handsh
 where
     T: AsyncRead + AsyncWrite + Unpin,
 {
+    let mut builder = Builder::new();
+    builder
+        .initial_window_size(config.h2_initial_window_size())
+        .initial_connection_window_size(config.h2_initial_connection_window_size());
+
     HandshakeWithTimeout {
-        handshake: handshake(io),
+        handshake: builder.handshake(io),
         timer: config
             .client_request_deadline()
             .map(|deadline| Box::pin(sleep_until(deadline.into()))),

actix-http/tests/test_server.rs

@@ -10,7 +10,10 @@ use actix_http::{
     header, Error, HttpService, KeepAlive, Request, Response, StatusCode, Version,
 };
 use actix_http_test::test_server;
-use actix_rt::{net::TcpStream, time::sleep};
+use actix_rt::{
+    net::TcpStream,
+    time::{sleep, timeout},
+};
 use actix_service::fn_service;
 use actix_utils::future::{err, ok, ready};
 use bytes::Bytes;
@@ -953,3 +956,62 @@ async fn h2c_auto() {
 
     srv.stop().await;
 }
+
+#[actix_rt::test]
+async fn h2_flow_control_window_sizes() {
+    let mut srv = test_server(|| {
+        HttpService::build()
+            .keep_alive(KeepAlive::Disabled)
+            .finish(|_req: Request| ok::<_, Infallible>(Response::ok()))
+            .tcp_auto_h2c()
+    })
+    .await;
+
+    let tcp = TcpStream::connect(srv.addr()).await.unwrap();
+
+    let mut builder = h2::client::Builder::new();
+    builder.max_send_buffer_size(4 * 1024 * 1024);
+
+    let (h2, connection) = builder.handshake(tcp).await.unwrap();
+    tokio::spawn(async move { connection.await.unwrap() });
+    let mut h2 = h2.ready().await.unwrap();
+
+    let request = ::http::Request::builder()
+        .method("POST")
+        .uri("/")
+        .body(())
+        .unwrap();
+
+    let (response, mut send) = h2.send_request(request, false).unwrap();
+
+    // request more than the default 64KiB. if server is advertising larger flow control windows,
+    // we should get at least 1MiB assigned.
+    send.reserve_capacity(2 * 1024 * 1024);
+
+    let cap = timeout(Duration::from_secs(2), async {
+        loop {
+            let cap = std::future::poll_fn(|cx| send.poll_capacity(cx))
+                .await
+                .unwrap()
+                .unwrap();
+
+            if cap >= 1024 * 1024 {
+                break cap;
+            }
+        }
+    })
+    .await
+    .unwrap();
+
+    assert!(
+        cap >= 1024 * 1024,
+        "expected >= 1MiB send capacity, got {cap}"
+    );
+
+    send.send_data(Bytes::new(), true).unwrap();
+
+    let res = response.await.unwrap();
+    assert!(res.status().is_success());
+
+    srv.stop().await;
+}

actix-web/CHANGES.md

@@ -3,6 +3,8 @@
 ## Unreleased
 
 - Minimum supported Rust version (MSRV) is now 1.88.
+- Improve HTTP/2 upload throughput by increasing default flow control window sizes. [#3638]
+- Add `HttpServer::{h2_initial_window_size, h2_initial_connection_window_size}` methods for tuning. [#3638]
 - Add `HttpRequest::url_for_map` and `HttpRequest::url_for_iter` methods for named URL parameters. [#3895]
 - Ignore unparsable cookies in `Cookie` request header.
 - Add `experimental-introspection` feature to report configured routes [#3594]
@@ -13,6 +15,7 @@
 [#3895]: https://github.com/actix/actix-web/pull/3895
 [#3594]: https://github.com/actix/actix-web/pull/3594
 [#3918]: https://github.com/actix/actix-web/pull/3918
+[#3638]: https://github.com/actix/actix-web/issues/3638
 [#3562]: https://github.com/actix/actix-web/issues/3562
 [#3191]: https://github.com/actix/actix-web/issues/3191
 

actix-web/src/server.rs

@@ -33,6 +33,8 @@ struct Config {
     client_request_timeout: Duration,
     client_disconnect_timeout: Duration,
     h1_allow_half_closed: bool,
+    h2_initial_window_size: Option<u32>,
+    h2_initial_connection_window_size: Option<u32>,
     #[allow(dead_code)] // only dead when no TLS features are enabled
     tls_handshake_timeout: Option<Duration>,
 }
@@ -120,6 +122,8 @@ where
                 client_request_timeout: Duration::from_secs(5),
                 client_disconnect_timeout: Duration::from_secs(1),
                 h1_allow_half_closed: true,
+                h2_initial_window_size: None,
+                h2_initial_connection_window_size: None,
                 tls_handshake_timeout: None,
             })),
             backlog: 1024,
@@ -282,6 +286,33 @@ where
         self
     }
 
+    /// Sets initial stream-level flow control window size for HTTP/2 connections.
+    ///
+    /// Higher values can improve upload performance on high-latency links at the cost of higher
+    /// worst-case memory usage per connection.
+    ///
+    /// The default value is 1MiB.
+    #[cfg(feature = "http2")]
+    pub fn h2_initial_window_size(self, size: u32) -> Self {
+        self.config.lock().unwrap().h2_initial_window_size = Some(size);
+        self
+    }
+
+    /// Sets initial connection-level flow control window size for HTTP/2 connections.
+    ///
+    /// Higher values can improve upload performance on high-latency links at the cost of higher
+    /// worst-case memory usage per connection.
+    ///
+    /// The default value is 2MiB.
+    #[cfg(feature = "http2")]
+    pub fn h2_initial_connection_window_size(self, size: u32) -> Self {
+        self.config
+            .lock()
+            .unwrap()
+            .h2_initial_connection_window_size = Some(size);
+        self
+    }
+
     /// Sets function that will be called once before each connection is handled.
     ///
     /// It will receive a `&std::any::Any`, which contains underlying connection type and an
@@ -590,6 +621,14 @@ where
                         svc = svc.tcp_nodelay(enabled);
                     }
 
+                    if let Some(val) = cfg.h2_initial_window_size {
+                        svc = svc.h2_initial_window_size(val);
+                    }
+
+                    if let Some(val) = cfg.h2_initial_connection_window_size {
+                        svc = svc.h2_initial_connection_window_size(val);
+                    }
+
                     if let Some(handler) = on_connect_fn.clone() {
                         svc =
                             svc.on_connect_ext(move |io: &_, ext: _| (handler)(io as &dyn Any, ext))
@@ -639,6 +678,14 @@ where
                         svc = svc.tcp_nodelay(enabled);
                     }
 
+                    if let Some(val) = cfg.h2_initial_window_size {
+                        svc = svc.h2_initial_window_size(val);
+                    }
+
+                    if let Some(val) = cfg.h2_initial_connection_window_size {
+                        svc = svc.h2_initial_connection_window_size(val);
+                    }
+
                     if let Some(handler) = on_connect_fn.clone() {
                         svc =
                             svc.on_connect_ext(move |io: &_, ext: _| (handler)(io as &dyn Any, ext))
@@ -719,6 +766,14 @@ where
                         svc = svc.tcp_nodelay(enabled);
                     }
 
+                    if let Some(val) = c.h2_initial_window_size {
+                        svc = svc.h2_initial_window_size(val);
+                    }
+
+                    if let Some(val) = c.h2_initial_connection_window_size {
+                        svc = svc.h2_initial_connection_window_size(val);
+                    }
+
                     if let Some(handler) = on_connect_fn.clone() {
                         svc = svc
                             .on_connect_ext(move |io: &_, ext: _| (handler)(io as &dyn Any, ext));
@@ -774,6 +829,14 @@ where
                         svc = svc.tcp_nodelay(enabled);
                     }
 
+                    if let Some(val) = c.h2_initial_window_size {
+                        svc = svc.h2_initial_window_size(val);
+                    }
+
+                    if let Some(val) = c.h2_initial_connection_window_size {
+                        svc = svc.h2_initial_connection_window_size(val);
+                    }
+
                     if let Some(handler) = on_connect_fn.clone() {
                         svc = svc
                             .on_connect_ext(move |io: &_, ext: _| (handler)(io as &dyn Any, ext));
@@ -844,6 +907,14 @@ where
                         svc = svc.tcp_nodelay(enabled);
                     }
 
+                    if let Some(val) = c.h2_initial_window_size {
+                        svc = svc.h2_initial_window_size(val);
+                    }
+
+                    if let Some(val) = c.h2_initial_connection_window_size {
+                        svc = svc.h2_initial_connection_window_size(val);
+                    }
+
                     if let Some(handler) = on_connect_fn.clone() {
                         svc = svc
                             .on_connect_ext(move |io: &_, ext: _| (handler)(io as &dyn Any, ext));
@@ -914,6 +985,14 @@ where
                         svc = svc.tcp_nodelay(enabled);
                     }
 
+                    if let Some(val) = c.h2_initial_window_size {
+                        svc = svc.h2_initial_window_size(val);
+                    }
+
+                    if let Some(val) = c.h2_initial_connection_window_size {
+                        svc = svc.h2_initial_connection_window_size(val);
+                    }
+
                     if let Some(handler) = on_connect_fn.clone() {
                         svc = svc
                             .on_connect_ext(move |io: &_, ext: _| (handler)(io as &dyn Any, ext));
@@ -985,6 +1064,14 @@ where
                         svc = svc.tcp_nodelay(enabled);
                     }
 
+                    if let Some(val) = c.h2_initial_window_size {
+                        svc = svc.h2_initial_window_size(val);
+                    }
+
+                    if let Some(val) = c.h2_initial_connection_window_size {
+                        svc = svc.h2_initial_connection_window_size(val);
+                    }
+
                     if let Some(handler) = on_connect_fn.clone() {
                         svc = svc
                             .on_connect_ext(move |io: &_, ext: _| (handler)(io as &dyn Any, ext));