feat(native): Split HTTP2 client flow control into three separate wi… (#26502)

```
== NO RELEASE NOTE ==
```

Author: kewang1024

presto-native-execution/presto_cpp/main/common/Configs.cpp

@@ -234,8 +234,10 @@ SystemConfig::SystemConfig() {
           NUM_PROP(kAnnouncementMaxFrequencyMs, 30'000), // 30s
           NUM_PROP(kHeartbeatFrequencyMs, 0),
           BOOL_PROP(kHttpClientHttp2Enabled, false),
-          NUM_PROP(kHttpClientHttp2MaxStreamsPerConnection, 1),
-          NUM_PROP(kHttpClientHttp2FlowControlWindow, 1 << 24 /*16MB*/),
+          NUM_PROP(kHttpClientHttp2MaxStreamsPerConnection, 8),
+          NUM_PROP(kHttpClientHttp2InitialStreamWindow, 1 << 23 /*8MB*/),
+          NUM_PROP(kHttpClientHttp2StreamWindow, 1 << 23 /*8MB*/),
+          NUM_PROP(kHttpClientHttp2SessionWindow, 1 << 26 /*64MB*/),
           STR_PROP(kExchangeMaxErrorDuration, "3m"),
           STR_PROP(kExchangeRequestTimeout, "20s"),
           STR_PROP(kExchangeConnectTimeout, "20s"),
@@ -868,8 +870,17 @@ uint32_t SystemConfig::httpClientHttp2MaxStreamsPerConnection() const {
       .value();
 }
 
-uint32_t SystemConfig::httpClientHttp2FlowControlWindow() const {
-  return optionalProperty<uint32_t>(kHttpClientHttp2FlowControlWindow).value();
+uint32_t SystemConfig::httpClientHttp2InitialStreamWindow() const {
+  return optionalProperty<uint32_t>(kHttpClientHttp2InitialStreamWindow)
+      .value();
+}
+
+uint32_t SystemConfig::httpClientHttp2StreamWindow() const {
+  return optionalProperty<uint32_t>(kHttpClientHttp2StreamWindow).value();
+}
+
+uint32_t SystemConfig::httpClientHttp2SessionWindow() const {
+  return optionalProperty<uint32_t>(kHttpClientHttp2SessionWindow).value();
 }
 
 std::chrono::duration<double> SystemConfig::exchangeMaxErrorDuration() const {

presto-native-execution/presto_cpp/main/common/Configs.h

@@ -661,10 +661,17 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kHttpClientHttp2MaxStreamsPerConnection{
       "http-client.http2.max-streams-per-connection"};
 
-  /// HTTP/2 flow control window size in bytes.
-  /// Controls the initial receive window, stream window, and session window.
-  static constexpr std::string_view kHttpClientHttp2FlowControlWindow{
-      "http-client.http2.flow-control-window"};
+  /// HTTP/2 initial stream window size in bytes.
+  static constexpr std::string_view kHttpClientHttp2InitialStreamWindow{
+      "http-client.http2.initial-stream-window"};
+
+  /// HTTP/2 stream window size in bytes.
+  static constexpr std::string_view kHttpClientHttp2StreamWindow{
+      "http-client.http2.stream-window"};
+
+  /// HTTP/2 session window size in bytes.
+  static constexpr std::string_view kHttpClientHttp2SessionWindow{
+      "http-client.http2.session-window"};
 
   static constexpr std::string_view kExchangeMaxErrorDuration{
       "exchange.max-error-duration"};
@@ -1058,7 +1065,11 @@ class SystemConfig : public ConfigBase {
 
   uint32_t httpClientHttp2MaxStreamsPerConnection() const;
 
-  uint32_t httpClientHttp2FlowControlWindow() const;
+  uint32_t httpClientHttp2InitialStreamWindow() const;
+
+  uint32_t httpClientHttp2StreamWindow() const;
+
+  uint32_t httpClientHttp2SessionWindow() const;
 
   std::chrono::duration<double> exchangeMaxErrorDuration() const;
 

presto-native-execution/presto_cpp/main/http/HttpClient.cpp

@@ -47,8 +47,12 @@ HttpClient::HttpClient(
       http2Enabled_(SystemConfig::instance()->httpClientHttp2Enabled()),
       maxConcurrentStreams_(
           SystemConfig::instance()->httpClientHttp2MaxStreamsPerConnection()),
-      http2FlowControlWindow_(
-          SystemConfig::instance()->httpClientHttp2FlowControlWindow()),
+      http2InitialStreamWindow_(
+          SystemConfig::instance()->httpClientHttp2InitialStreamWindow()),
+      http2StreamWindow_(
+          SystemConfig::instance()->httpClientHttp2StreamWindow()),
+      http2SessionWindow_(
+          SystemConfig::instance()->httpClientHttp2SessionWindow()),
       pool_(std::move(pool)),
       sslContext_(sslContext),
       reportOnBodyStatsFunc_(std::move(reportOnBodyStatsFunc)),
@@ -310,7 +314,9 @@ class ConnectionHandler : public proxygen::HTTPConnector::Callback {
       std::chrono::milliseconds connectTimeout,
       bool http2Enabled,
       uint32_t maxConcurrentStreams,
-      uint32_t http2FlowControlWindow,
+      uint32_t http2InitialStreamWindow,
+      uint32_t http2StreamWindow,
+      uint32_t http2SessionWindow,
       folly::EventBase* eventBase,
       const folly::SocketAddress& address,
       folly::SSLContextPtr sslContext)
@@ -320,7 +326,9 @@ class ConnectionHandler : public proxygen::HTTPConnector::Callback {
         connectTimeout_(connectTimeout),
         http2Enabled_(http2Enabled),
         maxConcurrentStreams_(maxConcurrentStreams),
-        http2FlowControlWindow_(http2FlowControlWindow),
+        http2InitialStreamWindow_(http2InitialStreamWindow),
+        http2StreamWindow_(http2StreamWindow),
+        http2SessionWindow_(http2SessionWindow),
         eventBase_(eventBase),
         address_(address),
         sslContext_(std::move(sslContext)) {}
@@ -343,9 +351,7 @@ class ConnectionHandler : public proxygen::HTTPConnector::Callback {
   void connectSuccess(proxygen::HTTPUpstreamSession* session) override {
     if (http2Enabled_) {
       session->setFlowControl(
-          http2FlowControlWindow_,
-          http2FlowControlWindow_,
-          http2FlowControlWindow_);
+          http2InitialStreamWindow_, http2StreamWindow_, http2SessionWindow_);
       session->setMaxConcurrentOutgoingStreams(maxConcurrentStreams_);
     }
     auto txn = session->newTransaction(responseHandler_.get());
@@ -372,7 +378,9 @@ class ConnectionHandler : public proxygen::HTTPConnector::Callback {
   const std::chrono::milliseconds connectTimeout_;
   const bool http2Enabled_;
   const uint32_t maxConcurrentStreams_;
-  const uint32_t http2FlowControlWindow_;
+  const uint32_t http2InitialStreamWindow_;
+  const uint32_t http2StreamWindow_;
+  const uint32_t http2SessionWindow_;
   folly::EventBase* const eventBase_;
   const folly::SocketAddress address_;
   const folly::SSLContextPtr sslContext_;
@@ -541,7 +549,9 @@ void HttpClient::sendRequest(std::shared_ptr<ResponseHandler> responseHandler) {
         connectTimeout_,
         http2Enabled_,
         maxConcurrentStreams_,
-        http2FlowControlWindow_,
+        http2InitialStreamWindow_,
+        http2StreamWindow_,
+        http2SessionWindow_,
         eventBase_,
         address_,
         sslContext_);

presto-native-execution/presto_cpp/main/http/HttpClient.h

@@ -202,7 +202,9 @@ class HttpClient : public std::enable_shared_from_this<HttpClient> {
   const std::chrono::milliseconds connectTimeout_;
   const bool http2Enabled_;
   const uint32_t maxConcurrentStreams_;
-  const uint32_t http2FlowControlWindow_;
+  const uint32_t http2InitialStreamWindow_;
+  const uint32_t http2StreamWindow_;
+  const uint32_t http2SessionWindow_;
   const std::shared_ptr<velox::memory::MemoryPool> pool_;
   const folly::SSLContextPtr sslContext_;
   const std::function<void(int)> reportOnBodyStatsFunc_;