feat(native): Add http2 support for exchange client

Author: kewang1024

presto-native-execution/presto_cpp/main/PrestoExchangeSource.cpp

@@ -19,6 +19,7 @@
 #include <sstream>
 
 #include "presto_cpp/main/QueryContextManager.h"
+#include "presto_cpp/main/common/Configs.h"
 #include "presto_cpp/main/common/Counters.h"
 #include "velox/common/base/Exceptions.h"
 #include "velox/common/testutil/TestValue.h"
@@ -278,9 +279,11 @@ void PrestoExchangeSource::processDataResponse(
     return;
   }
   auto* headers = response->headers();
-  VELOX_CHECK(
-      !headers->getIsChunked(),
-      "Chunked http transferring encoding is not supported.");
+  if (!SystemConfig::instance()->exchangeHttp2Enabled()) {
+    VELOX_CHECK(
+        !headers->getIsChunked(),
+        "Chunked http transferring encoding is not supported.");
+  }
   const uint64_t contentLength =
       atol(headers->getHeaders()
                .getSingleOrEmpty(proxygen::HTTP_HEADER_CONTENT_LENGTH)

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

@@ -232,6 +232,7 @@ SystemConfig::SystemConfig() {
           NUM_PROP(kLogNumZombieTasks, 20),
           NUM_PROP(kAnnouncementMaxFrequencyMs, 30'000), // 30s
           NUM_PROP(kHeartbeatFrequencyMs, 0),
+          BOOL_PROP(kExchangeHttp2Enabled, false),
           STR_PROP(kExchangeMaxErrorDuration, "3m"),
           STR_PROP(kExchangeRequestTimeout, "20s"),
           STR_PROP(kExchangeConnectTimeout, "20s"),
@@ -832,6 +833,10 @@ uint64_t SystemConfig::heartbeatFrequencyMs() const {
   return optionalProperty<uint64_t>(kHeartbeatFrequencyMs).value();
 }
 
+bool SystemConfig::exchangeHttp2Enabled() const {
+  return optionalProperty<bool>(kExchangeHttp2Enabled).value();
+}
+
 std::chrono::duration<double> SystemConfig::exchangeMaxErrorDuration() const {
   return velox::config::toDuration(
       optionalProperty(kExchangeMaxErrorDuration).value());

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

@@ -638,6 +638,10 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kHeartbeatFrequencyMs{
       "heartbeat-frequency-ms"};
 
+  /// Whether HTTP/2 should be enabled for exchange HTTP client.
+  static constexpr std::string_view kExchangeHttp2Enabled{
+      "exchange.http-client.http2-enabled"};
+
   static constexpr std::string_view kExchangeMaxErrorDuration{
       "exchange.max-error-duration"};
 
@@ -1013,6 +1017,8 @@ class SystemConfig : public ConfigBase {
 
   uint64_t heartbeatFrequencyMs() const;
 
+  bool exchangeHttp2Enabled() const;
+
   std::chrono::duration<double> exchangeMaxErrorDuration() const;
 
   std::chrono::duration<double> exchangeRequestTimeoutMs() const;

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

@@ -16,6 +16,7 @@
 #include <fmt/format.h>
 #include <folly/io/Cursor.h>
 #include <sys/resource.h>
+#include "presto_cpp/main/common/Configs.h"
 #include "velox/common/process/ThreadDebugInfo.h"
 
 namespace facebook::presto::util {
@@ -37,7 +38,11 @@ std::shared_ptr<folly::SSLContext> createSSLContext(
     sslContext->loadCertKeyPairFromFiles(
         clientCertAndKeyPath.c_str(), clientCertAndKeyPath.c_str());
     sslContext->setCiphersOrThrow(ciphers);
-    sslContext->setAdvertisedNextProtocols({"http/1.1"});
+    if (SystemConfig::instance()->exchangeHttp2Enabled()) {
+      sslContext->setAdvertisedNextProtocols({"h2", "http/1.1"});
+    } else {
+      sslContext->setAdvertisedNextProtocols({"http/1.1"});
+    }
     return sslContext;
   } catch (const std::exception& ex) {
     LOG(FATAL) << fmt::format(

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

@@ -18,6 +18,7 @@
 #endif // PRESTO_ENABLE_JWT
 #include <folly/io/async/EventBaseManager.h>
 #include <folly/synchronization/Latch.h>
+#include <proxygen/lib/http/codec/CodecProtocol.h>
 #include <velox/common/base/Exceptions.h>
 #include "presto_cpp/main/common/Configs.h"
 #include "presto_cpp/main/common/Counters.h"
@@ -200,13 +201,21 @@ class ResponseHandler : public proxygen::HTTPTransactionHandler {
     return promise_.getSemiFuture();
   }
 
-  void setTransaction(proxygen::HTTPTransaction* /* txn */) noexcept override {}
+  void setTransaction(proxygen::HTTPTransaction* txn) noexcept override {
+    if (txn) {
+      protocol_ = txn->getTransport().getCodec().getProtocol();
+    }
+  }
+
   void detachTransaction() noexcept override {
     self_.reset();
   }
 
   void onHeadersComplete(
       std::unique_ptr<proxygen::HTTPMessage> msg) noexcept override {
+    if (protocol_.has_value()) {
+      VLOG(2) << "HttpClient Receive " << proxygen::getCodecProtocolString(protocol_.value());
+    }
     response_ = std::make_unique<HttpResponse>(
         std::move(msg),
         client_->memoryPool(),
@@ -267,6 +276,7 @@ class ResponseHandler : public proxygen::HTTPTransactionHandler {
   folly::Promise<std::unique_ptr<HttpResponse>> promise_;
   std::shared_ptr<ResponseHandler> self_;
   std::shared_ptr<HttpClient> client_;
+  std::optional<proxygen::CodecProtocol> protocol_;
 };
 
 // Responsible for making an HTTP request. The request will be made in 2

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

@@ -18,6 +18,7 @@
 #include <proxygen/lib/http/connpool/SessionPool.h>
 #include <proxygen/lib/http/session/HTTPUpstreamSession.h>
 #include <velox/common/memory/MemoryPool.h>
+#include "presto_cpp/main/common/Configs.h"
 #include "presto_cpp/main/http/HttpConstants.h"
 #include "velox/common/base/Exceptions.h"
 
@@ -215,7 +216,9 @@ class HttpClient : public std::enable_shared_from_this<HttpClient> {
 class RequestBuilder {
  public:
   RequestBuilder() {
-    headers_.setHTTPVersion(1, 1);
+    if (!SystemConfig::instance()->exchangeHttp2Enabled()) {
+      headers_.setHTTPVersion(1, 1);
+    }
   }
 
   RequestBuilder& method(proxygen::HTTPMethod method) {