feat(storage): implement TTFB speculative hedging with configurable connect timeouts (#16344)

Author: ajayky-os

google/cloud/internal/curl_impl.cc

@@ -197,6 +197,10 @@ CurlImpl::CurlImpl(CurlHandle handle,
 
   http_version_ = options.get<HttpVersionOption>();
 
+  if (options.has<HttpConnectTimeoutOption>()) {
+    connect_timeout_ms_ = options.get<HttpConnectTimeoutOption>();
+  }
+
   transfer_stall_timeout_ = options.get<TransferStallTimeoutOption>();
   transfer_stall_minimum_rate_ = options.get<TransferStallMinimumRateOption>();
   download_stall_timeout_ = options.get<DownloadStallTimeoutOption>();
@@ -262,6 +266,22 @@ void CurlImpl::WriteHeader(std::string const& header) {
   request_headers_.reset(headers);
 }
 
+Status CurlImpl::SetConnectTimeout(std::chrono::seconds fallback) {
+  // libcurl stores `CURLOPT_CONNECTTIMEOUT` and `CURLOPT_CONNECTTIMEOUT_MS` in
+  // a single setting, so there must be exactly one place that decides the
+  // value. An explicitly configured connect timeout wins; otherwise fall back
+  // to the timeout implied by the stall options, preserving the historical
+  // behavior for applications that do not set one.
+  auto const connect_timeout =
+      connect_timeout_ms_ != std::chrono::milliseconds::zero()
+          ? connect_timeout_ms_
+          : std::chrono::duration_cast<std::chrono::milliseconds>(fallback);
+  if (connect_timeout == std::chrono::milliseconds::zero()) return {};
+  // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* long
+  auto const timeout_ms = static_cast<long>(connect_timeout.count());
+  return handle_.SetOption(CURLOPT_CONNECTTIMEOUT_MS, timeout_ms);
+}
+
 void CurlImpl::MergeAndWriteHeaders(
     std::function<void(HttpHeader const&)> const& write_fn) {
   // There are some headers that we do not want to merge. These headers
@@ -432,6 +452,13 @@ Status CurlImpl::MakeRequest(HttpMethod method, RestContext& context,
 #endif
   }
 
+  // Nothing else sets a connection timeout, so this can be decided once, before
+  // the per-method options.
+  status =
+      SetConnectTimeout(method == HttpMethod::kGet ? download_stall_timeout_
+                                                   : transfer_stall_timeout_);
+  if (!status.ok()) return OnTransferError(context, std::move(status));
+
   if (method == HttpMethod::kGet) {
     status = handle_.SetOption(CURLOPT_NOPROGRESS, 1L);
     if (!status.ok()) return OnTransferError(context, std::move(status));
@@ -440,8 +467,6 @@ Status CurlImpl::MakeRequest(HttpMethod method, RestContext& context,
       auto const timeout = static_cast<long>(download_stall_timeout_.count());
       // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* long
       auto const limit = static_cast<long>(download_stall_minimum_rate_);
-      status = handle_.SetOption(CURLOPT_CONNECTTIMEOUT, timeout);
-      if (!status.ok()) return OnTransferError(context, std::move(status));
       // Timeout if the request sends or receives less than 1 byte/second
       // (i.e.  effectively no bytes) for download_stall_timeout_.
       status = handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, limit);
@@ -457,8 +482,6 @@ Status CurlImpl::MakeRequest(HttpMethod method, RestContext& context,
     auto const timeout = static_cast<long>(transfer_stall_timeout_.count());
     // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* long
     auto const limit = static_cast<long>(transfer_stall_minimum_rate_);
-    status = handle_.SetOption(CURLOPT_CONNECTTIMEOUT, timeout);
-    if (!status.ok()) return OnTransferError(context, std::move(status));
     // Timeout if the request sends or receives less than 1 byte/second
     // (i.e.  effectively no bytes) for transfer_stall_timeout_.
     status = handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, limit);

google/cloud/internal/curl_impl.h

@@ -123,6 +123,13 @@ class CurlImpl {
 
   void WriteHeader(std::string const& header);
 
+  // Sets the connection timeout, using `HttpConnectTimeoutOption` when the
+  // application configured one and @p fallback (the relevant stall timeout)
+  // otherwise. This is the only place that sets a connection timeout: libcurl
+  // keeps a single value for `CURLOPT_CONNECTTIMEOUT` and
+  // `CURLOPT_CONNECTTIMEOUT_MS`.
+  Status SetConnectTimeout(std::chrono::seconds fallback);
+
   // Cleanup the CURL handles, leaving them ready for reuse.
   void CleanupHandles();
   // Perform at least part of the request.
@@ -146,6 +153,7 @@ class CurlImpl {
   CurlHandle::SocketOptions socket_options_;
   std::string user_agent_;
   std::string http_version_;
+  std::chrono::milliseconds connect_timeout_ms_{0};
   std::chrono::seconds transfer_stall_timeout_;
   std::uint32_t transfer_stall_minimum_rate_;
   std::chrono::seconds download_stall_timeout_;

google/cloud/internal/rest_options.h

@@ -56,6 +56,23 @@ struct TransferStallMinimumRateOption {
   using Type = std::int32_t;
 };
 
+/**
+ * Sets the TCP/TLS connection timeout.
+ *
+ * If the connection cannot be established within this time, the request is
+ * aborted. This is useful as a fail-safe against OS-level TCP locks during
+ * severe network routing anomalies.
+ *
+ * This applies to all HTTP methods, and it only bounds establishing the
+ * connection: it has no effect once bytes start flowing. Note that this takes
+ * precedence over the connection timeout implied by
+ * `TransferStallTimeoutOption` and `DownloadStallTimeoutOption`, as libcurl
+ * uses a single setting for all of them.
+ */
+struct HttpConnectTimeoutOption {
+  using Type = std::chrono::milliseconds;
+};
+
 /**
  * Sets the download stall timeout.
  *
@@ -101,9 +118,10 @@ struct TargetApiVersionOption {
 
 /// The complete list of options accepted by `CurlRestClient`
 using RestInternalOptionList = ::google::cloud::OptionList<
-    TransferStallTimeoutOption, TransferStallMinimumRateOption,
-    DownloadStallTimeoutOption, DownloadStallMinimumRateOption,
-    LongrunningEndpointOption, TargetApiVersionOption>;
+    HttpConnectTimeoutOption, TransferStallTimeoutOption,
+    TransferStallMinimumRateOption, DownloadStallTimeoutOption,
+    DownloadStallMinimumRateOption, LongrunningEndpointOption,
+    TargetApiVersionOption>;
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace rest_internal

google/cloud/storage/client.cc

@@ -581,6 +581,26 @@ Options DefaultOptions(Options opts) {
                                                                 "/iamapi");
   }
 
+  if (!o.has<storage_experimental::EnableReadHedgingOption>()) {
+    o.set<storage_experimental::EnableReadHedgingOption>(false);
+  }
+  if (!o.has<storage_experimental::ReadHedgeRateLimitOption>()) {
+    o.set<storage_experimental::ReadHedgeRateLimitOption>(0.0);
+  }
+  if (!o.has<storage_experimental::MaxConcurrentHedgesOption>()) {
+    o.set<storage_experimental::MaxConcurrentHedgesOption>(0);
+  }
+  if (!o.has<storage_experimental::MaximumHedgeBufferOption>()) {
+    o.set<storage_experimental::MaximumHedgeBufferOption>(64 * 1024 * 1024);
+  }
+  if (!o.has<storage_experimental::ReadHedgeDelayOption>()) {
+    o.set<storage_experimental::ReadHedgeDelayOption>(
+        std::chrono::milliseconds(500));
+  }
+  if (!o.has<storage_experimental::MaxReadHedgesOption>()) {
+    o.set<storage_experimental::MaxReadHedgesOption>(2);
+  }
+
   auto logging = GetEnv("CLOUD_STORAGE_ENABLE_TRACING");
   if (logging) {
     for (auto c : absl::StrSplit(*logging, ',')) {
@@ -633,6 +653,12 @@ Options DefaultOptions(Options opts) {
     rest_defaults.set<rest::CAPathOption>(o.get<internal::CAPathOption>());
   }
 
+  // The (experimental) connect timeout is mapped the same way.
+  if (o.has<storage_experimental::HttpConnectTimeoutOption>()) {
+    rest_defaults.set<rest::HttpConnectTimeoutOption>(
+        o.get<storage_experimental::HttpConnectTimeoutOption>());
+  }
+
   return google::cloud::internal::MergeOptions(std::move(o),
                                                std::move(rest_defaults));
 }

google/cloud/storage/client_test.cc

@@ -470,6 +470,21 @@ TEST_F(ClientTest, Timeouts) {
             internal::DefaultOptions().get<DownloadStallTimeoutOption>());
 }
 
+TEST_F(ClientTest, ConnectTimeout) {
+  namespace rest = ::google::cloud::rest_internal;
+
+  // The connect timeout is opt-in: when the application does not set it the
+  // REST layer keeps libcurl's own default.
+  EXPECT_FALSE(
+      internal::DefaultOptions().has<rest::HttpConnectTimeoutOption>());
+
+  auto const options = internal::DefaultOptions(
+      Options{}.set<storage_experimental::HttpConnectTimeoutOption>(
+          std::chrono::milliseconds(1500)));
+  EXPECT_EQ(std::chrono::milliseconds(1500),
+            options.get<rest::HttpConnectTimeoutOption>());
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage

google/cloud/storage/google_cloud_cpp_storage.bzl

@@ -74,6 +74,8 @@ google_cloud_cpp_storage_hdrs = [
     "internal/hash_validator.h",
     "internal/hash_validator_impl.h",
     "internal/hash_values.h",
+    "internal/hedged_object_read_source.h",
+    "internal/hedging_thread_pool.h",
     "internal/hmac_key_metadata_parser.h",
     "internal/hmac_key_requests.h",
     "internal/http_response.h",
@@ -184,6 +186,7 @@ google_cloud_cpp_storage_srcs = [
     "internal/hash_validator.cc",
     "internal/hash_validator_impl.cc",
     "internal/hash_values.cc",
+    "internal/hedged_object_read_source.cc",
     "internal/hmac_key_metadata_parser.cc",
     "internal/hmac_key_requests.cc",
     "internal/http_response.cc",

google/cloud/storage/google_cloud_cpp_storage.cmake

@@ -117,6 +117,9 @@ add_library(
     internal/hash_validator_impl.h
     internal/hash_values.cc
     internal/hash_values.h
+    internal/hedged_object_read_source.cc
+    internal/hedged_object_read_source.h
+    internal/hedging_thread_pool.h
     internal/hmac_key_metadata_parser.cc
     internal/hmac_key_metadata_parser.h
     internal/hmac_key_requests.cc
@@ -447,6 +450,8 @@ if (BUILD_TESTING)
         internal/hash_function_impl_test.cc
         internal/hash_validator_test.cc
         internal/hash_values_test.cc
+        internal/hedged_object_read_source_test.cc
+        internal/hedging_thread_pool_test.cc
         internal/hmac_key_requests_test.cc
         internal/http_response_test.cc
         internal/logging_stub_test.cc

google/cloud/storage/internal/connection_impl.cc

@@ -14,13 +14,15 @@
 
 #include "google/cloud/internal/disable_deprecation_warnings.inc"
 #include "google/cloud/storage/internal/connection_impl.h"
+#include "google/cloud/storage/internal/hedged_object_read_source.h"
 #include "google/cloud/storage/internal/retry_object_read_source.h"
 #include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/internal/filesystem.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/internal/rest_retry_loop.h"
 #include "google/cloud/log.h"
 #include "absl/strings/match.h"
+#include <algorithm>
 #include <chrono>
 #include <fstream>
 #include <functional>
@@ -155,7 +157,27 @@ std::shared_ptr<StorageConnectionImpl> StorageConnectionImpl::Create(
 StorageConnectionImpl::StorageConnectionImpl(
     std::unique_ptr<storage_internal::GenericStub> stub, Options options)
     : stub_(std::move(stub)),
-      options_(MergeOptions(std::move(options), stub_->options())) {}
+      options_(MergeOptions(std::move(options), stub_->options())) {
+  if (options_.get<storage_experimental::EnableReadHedgingOption>()) {
+    // The pool only runs stream-open attempts: one primary and (at most) a few
+    // hedges per stream being opened. Size it to the number of connections the
+    // REST layer can use, falling back to the hardware concurrency when the
+    // connection pool is unbounded (`ConnectionPoolSizeOption == 0`).
+    auto pool_size = options_.get<ConnectionPoolSizeOption>();
+    if (pool_size == 0) {
+      pool_size =
+          (std::max<std::size_t>)(4, std::thread::hardware_concurrency());
+    }
+    auto const max_threads = 2 * pool_size;
+    auto const rate_limit =
+        options_.get<storage_experimental::ReadHedgeRateLimitOption>();
+    auto const max_concurrent =
+        options_.get<storage_experimental::MaxConcurrentHedgesOption>();
+    // Allow bursts of up to one second worth of hedges.
+    hedge_pool_ = std::make_shared<HedgingThreadPool>(
+        max_threads, rate_limit, rate_limit, max_concurrent);
+  }
+}
 
 Options StorageConnectionImpl::options() const { return options_; }
 
@@ -392,15 +414,37 @@ StatusOr<std::unique_ptr<ObjectReadSource>> StorageConnectionImpl::ReadObject(
         *current, request, where);
   };
 
-  auto retry_policy = current->get<RetryPolicyOption>()->clone();
-  auto backoff_policy = current->get<BackoffPolicyOption>()->clone();
-  auto child = factory(request, *retry_policy, *backoff_policy);
-  if (!child) return child;
+  auto retry_source_factory =
+      [factory, current,
+       request]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    auto retry_policy = current->get<RetryPolicyOption>()->clone();
+    auto backoff_policy = current->get<BackoffPolicyOption>()->clone();
+    auto child = factory(request, *retry_policy, *backoff_policy);
+    if (!child) return child;
+    return std::unique_ptr<ObjectReadSource>(
+        std::make_unique<RetryObjectReadSource>(
+            factory, current, request, *std::move(child),
+            std::move(retry_policy), std::move(backoff_policy)));
+  };
+
+  auto const enable_hedging =
+      current->get<storage_experimental::EnableReadHedgingOption>();
+  auto const delay = current->get<storage_experimental::ReadHedgeDelayOption>();
+  auto const max_hedges =
+      current->get<storage_experimental::MaxReadHedgesOption>();
+  auto const max_buffer =
+      current->get<storage_experimental::MaximumHedgeBufferOption>();
+
+  if (!enable_hedging || max_hedges <= 0 || !hedge_pool_) {
+    return retry_source_factory();
+  }
 
+  // `max_buffer` bounds the size of an individual read, which is only known
+  // when the application calls `Read()`; the source applies it there.
   return std::unique_ptr<ObjectReadSource>(
-      std::make_unique<RetryObjectReadSource>(
-          std::move(factory), std::move(current), request, *std::move(child),
-          std::move(retry_policy), std::move(backoff_policy)));
+      std::make_unique<HedgedObjectReadSource>(hedge_pool_,
+                                               std::move(retry_source_factory),
+                                               delay, max_hedges, max_buffer));
 }
 
 StatusOr<ListObjectsResponse> StorageConnectionImpl::ListObjects(

google/cloud/storage/internal/connection_impl.h

@@ -17,6 +17,7 @@
 
 #include "google/cloud/storage/idempotency_policy.h"
 #include "google/cloud/storage/internal/generic_stub.h"
+#include "google/cloud/storage/internal/hedging_thread_pool.h"
 #include "google/cloud/storage/internal/storage_connection.h"
 #include "google/cloud/storage/object_read_stream.h"
 #include "google/cloud/storage/retry_policy.h"
@@ -187,6 +188,7 @@ class StorageConnectionImpl
 
   std::unique_ptr<storage_internal::GenericStub> stub_;
   Options options_;
+  std::shared_ptr<HedgingThreadPool> hedge_pool_;
   google::cloud::internal::InvocationIdGenerator invocation_id_generator_;
 };