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/PrestoServer.cpp

@@ -257,8 +257,14 @@ void PrestoServer::run() {
             "Https Client Certificates are not configured correctly");
       }
 
-      sslContext_ =
-          util::createSSLContext(optionalClientCertPath.value(), ciphers);
+      sslContext_ = util::createSSLContext(
+          optionalClientCertPath.value(),
+          ciphers,
+          false);
+      exchangeSslContext_ = util::createSSLContext(
+          optionalClientCertPath.value(),
+          ciphers,
+          systemConfig->exchangeHttp2Enabled());
     }
 
     if (systemConfig->internalCommunicationJwtEnabled()) {
@@ -449,7 +455,7 @@ void PrestoServer::run() {
             exchangeHttpCpuExecutor_.get(),
             exchangeHttpIoExecutor_.get(),
             exchangeSourceConnectionPool_.get(),
-            sslContext_);
+            exchangeSslContext_);
       });
 
   velox::exec::ExchangeSource::registerFactory(

presto-native-execution/presto_cpp/main/PrestoServer.h

@@ -313,6 +313,7 @@ class PrestoServer {
   std::string nodeLocation_;
   std::string nodePoolType_;
   folly::SSLContextPtr sslContext_;
+  folly::SSLContextPtr exchangeSslContext_;
   std::string prestoBuiltinFunctionPrefix_;
 };
 

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

@@ -31,13 +31,18 @@ DateTime toISOTimestamp(uint64_t timeMilli) {
 
 std::shared_ptr<folly::SSLContext> createSSLContext(
     const std::string& clientCertAndKeyPath,
-    const std::string& ciphers) {
+    const std::string& ciphers,
+    bool http2Enabled) {
   try {
     auto sslContext = std::make_shared<folly::SSLContext>();
     sslContext->loadCertKeyPairFromFiles(
         clientCertAndKeyPath.c_str(), clientCertAndKeyPath.c_str());
     sslContext->setCiphersOrThrow(ciphers);
-    sslContext->setAdvertisedNextProtocols({"http/1.1"});
+    if (http2Enabled) {
+      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/common/Utils.h

@@ -30,7 +30,8 @@ DateTime toISOTimestamp(uint64_t timeMilli);
 
 std::shared_ptr<folly::SSLContext> createSSLContext(
     const std::string& clientCertAndKeyPath,
-    const std::string& ciphers);
+    const std::string& ciphers,
+    bool http2Enabled);
 
 /// Returns current process-wide CPU time in nanoseconds.
 long getProcessCpuTimeNs();

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,22 @@ 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 received response of "
+              << proxygen::getCodecProtocolString(protocol_.value());
+    }
     response_ = std::make_unique<HttpResponse>(
         std::move(msg),
         client_->memoryPool(),
@@ -267,6 +277,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

@@ -215,7 +215,6 @@ class HttpClient : public std::enable_shared_from_this<HttpClient> {
 class RequestBuilder {
  public:
   RequestBuilder() {
-    headers_.setHTTPVersion(1, 1);
   }
 
   RequestBuilder& method(proxygen::HTTPMethod method) {