feat(storage): experimental options to tune stall timeouts (#9593)

Some applications need more progress than just "at least 1 byte per
second for K seconds".  These new options allow them to require "at
least N bytes per second for K seconds".

google/cloud/internal/curl_impl.cc

@@ -178,7 +178,9 @@ void CurlImpl::ApplyOptions(Options const& options) {
   user_agent_ = absl::StrCat(absl::StrJoin(agents, " "), UserAgentSuffix());
   http_version_ = std::move(options.get<HttpVersionOption>());
   transfer_stall_timeout_ = options.get<TransferStallTimeoutOption>();
+  transfer_stall_minimum_rate_ = options.get<TransferStallMinimumRateOption>();
   download_stall_timeout_ = options.get<DownloadStallTimeoutOption>();
+  transfer_stall_minimum_rate_ = options.get<DownloadStallMinimumRateOption>();
   ignored_http_error_codes_ = options.get<IgnoredHttpErrorCodes>();
 }
 
@@ -726,11 +728,13 @@ Status CurlImpl::MakeRequest(CurlImpl::HttpMethod method,
     if (download_stall_timeout_.count() != 0) {
       // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* `long`
       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(std::move(status));
       // Timeout if the request sends or receives less than 1 byte/second (i.e.
       // effectively no bytes) for `download_stall_timeout_` seconds.
-      status = handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, 1L);
+      status = handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, limit);
       if (!status.ok()) return OnTransferError(std::move(status));
       status = handle_.SetOption(CURLOPT_LOW_SPEED_TIME, timeout);
       if (!status.ok()) return OnTransferError(std::move(status));
@@ -741,11 +745,13 @@ Status CurlImpl::MakeRequest(CurlImpl::HttpMethod method,
   if (transfer_stall_timeout_.count() != 0) {
     // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* `long`
     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(std::move(status));
     // Timeout if the request sends or receives less than 1 byte/second (i.e.
     // effectively no bytes) for `transfer_stall_timeout_` seconds.
-    status = handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, 1L);
+    status = handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, limit);
     if (!status.ok()) return OnTransferError(std::move(status));
     status = handle_.SetOption(CURLOPT_LOW_SPEED_TIME, timeout);
     if (!status.ok()) return OnTransferError(std::move(status));

google/cloud/internal/curl_impl.h

@@ -127,7 +127,9 @@ class CurlImpl {
   bool logging_enabled_ = false;
   CurlHandle::SocketOptions socket_options_;
   std::chrono::seconds transfer_stall_timeout_;
+  std::uint32_t transfer_stall_minimum_rate_ = 1;
   std::chrono::seconds download_stall_timeout_;
+  std::uint32_t download_stall_minimum_rate_ = 1;
   std::string http_version_;
   std::int32_t http_code_;
   std::set<std::int32_t> ignored_http_error_codes_;

google/cloud/internal/rest_options.h

@@ -18,6 +18,7 @@
 #include "google/cloud/options.h"
 #include "google/cloud/version.h"
 #include <chrono>
+#include <cstdint>
 #include <memory>
 #include <string>
 
@@ -56,6 +57,16 @@ struct TransferStallTimeoutOption {
   using Type = std::chrono::seconds;
 };
 
+/**
+ * The minimum accepted bytes/second transfer rate.
+ *
+ * If the average rate is below this value for the `TransferStallTimeoutOption`
+ * then the transfer is aborted.
+ */
+struct TransferStallMinimumRateOption {
+  using Type = std::int32_t;
+};
+
 /**
  * Sets the download stall timeout.
  *
@@ -74,6 +85,16 @@ struct DownloadStallTimeoutOption {
   using Type = std::chrono::seconds;
 };
 
+/**
+ * The minimum accepted bytes/second download rate.
+ *
+ * If the average rate is below this value for the `DownloadStallTimeoutOption`
+ * then the download is aborted.
+ */
+struct DownloadStallMinimumRateOption {
+  using Type = std::int32_t;
+};
+
 /**
  * Some services appropriate Http error codes for their own use. If any such
  * error codes need to be treated as non-failures, this option can indicate
@@ -84,10 +105,10 @@ struct IgnoredHttpErrorCodes {
 };
 
 /// The complete list of options accepted by `CurlRestClient`
-using RestOptionList =
-    ::google::cloud::OptionList<UserIpOption, TransferStallTimeoutOption,
-                                DownloadStallTimeoutOption,
-                                IgnoredHttpErrorCodes>;
+using RestOptionList = ::google::cloud::OptionList<
+    UserIpOption, TransferStallTimeoutOption, TransferStallMinimumRateOption,
+    DownloadStallTimeoutOption, DownloadStallMinimumRateOption,
+    IgnoredHttpErrorCodes>;
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace rest_internal

google/cloud/storage/client_options.cc

@@ -187,6 +187,8 @@ Options DefaultOptions(std::shared_ptr<oauth2::Credentials> credentials,
           .set<MaximumCurlSocketSendSizeOption>(0)
           .set<TransferStallTimeoutOption>(std::chrono::seconds(
               GOOGLE_CLOUD_CPP_STORAGE_DEFAULT_DOWNLOAD_STALL_TIMEOUT))
+          .set<storage_experimental::TransferStallMinimumRateOption>(1)
+          .set<storage_experimental::DownloadStallMinimumRateOption>(1)
           .set<RetryPolicyOption>(
               LimitedTimeRetryPolicy(
                   STORAGE_CLIENT_DEFAULT_MAXIMUM_RETRY_PERIOD)
@@ -238,8 +240,12 @@ Options DefaultOptions(std::shared_ptr<oauth2::Credentials> credentials,
       Options{}
           .set<rest::DownloadStallTimeoutOption>(
               o.get<DownloadStallTimeoutOption>())
+          .set<rest::DownloadStallMinimumRateOption>(
+              o.get<storage_experimental::DownloadStallMinimumRateOption>())
           .set<rest::TransferStallTimeoutOption>(
               o.get<TransferStallTimeoutOption>())
+          .set<rest::TransferStallMinimumRateOption>(
+              o.get<storage_experimental::TransferStallMinimumRateOption>())
           .set<rest::MaximumCurlSocketRecvSizeOption>(
               o.get<MaximumCurlSocketRecvSizeOption>())
           .set<rest::MaximumCurlSocketSendSizeOption>(

google/cloud/storage/client_options_test.cc

@@ -325,13 +325,17 @@ TEST_F(ClientOptionsTest, MakeOptionsFromDefault) {
   EXPECT_EQ("test-project-id", opts.get<ProjectIdOption>());
   EXPECT_LT(0, opts.get<ConnectionPoolSizeOption>());
   EXPECT_LT(0, opts.get<DownloadBufferSizeOption>());
+  EXPECT_LT(0,
+            opts.get<storage_experimental::DownloadStallMinimumRateOption>());
   EXPECT_LT(0, opts.get<UploadBufferSizeOption>());
   EXPECT_LT(0, opts.get<MaximumSimpleUploadSizeOption>());
   EXPECT_TRUE(opts.has<EnableCurlSslLockingOption>());
   EXPECT_TRUE(opts.has<EnableCurlSigpipeHandlerOption>());
   EXPECT_EQ(0, opts.get<MaximumCurlSocketSendSizeOption>());
   EXPECT_EQ(0, opts.get<MaximumCurlSocketRecvSizeOption>());
   EXPECT_LT(0, opts.get<TransferStallTimeoutOption>().count());
+  EXPECT_LT(0,
+            opts.get<storage_experimental::TransferStallMinimumRateOption>());
   EXPECT_THAT(opts.get<CARootsFilePathOption>(), IsEmpty());
 }
 
@@ -360,13 +364,19 @@ TEST_F(ClientOptionsTest, DefaultOptions) {
   EXPECT_EQ(0, o.get<MaximumCurlSocketRecvSizeOption>());
   EXPECT_EQ(0, o.get<MaximumCurlSocketSendSizeOption>());
   EXPECT_LT(std::chrono::seconds(0), o.get<TransferStallTimeoutOption>());
+  EXPECT_LT(0, o.get<storage_experimental::TransferStallMinimumRateOption>());
   EXPECT_LT(std::chrono::seconds(0), o.get<DownloadStallTimeoutOption>());
+  EXPECT_LT(0, o.get<storage_experimental::DownloadStallMinimumRateOption>());
 
   namespace rest = ::google::cloud::rest_internal;
   EXPECT_EQ(o.get<rest::DownloadStallTimeoutOption>(),
             o.get<DownloadStallTimeoutOption>());
+  EXPECT_EQ(o.get<rest::DownloadStallMinimumRateOption>(),
+            o.get<storage_experimental::DownloadStallMinimumRateOption>());
   EXPECT_EQ(o.get<rest::TransferStallTimeoutOption>(),
             o.get<TransferStallTimeoutOption>());
+  EXPECT_EQ(o.get<rest::TransferStallMinimumRateOption>(),
+            o.get<storage_experimental::TransferStallMinimumRateOption>());
   EXPECT_EQ(o.get<rest::MaximumCurlSocketRecvSizeOption>(),
             o.get<MaximumCurlSocketRecvSizeOption>());
   EXPECT_EQ(o.get<rest::MaximumCurlSocketSendSizeOption>(),

google/cloud/storage/internal/curl_download_request.cc

@@ -261,11 +261,13 @@ Status CurlDownloadRequest::SetOptions() {
   if (download_stall_timeout_.count() != 0) {
     // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* `long`
     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(std::move(status));
-    // Timeout if the download receives less than 1 byte/second (i.e.
-    // effectively no bytes) for `transfer_stall_timeout_` seconds.
-    status = handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, 1L);
+    // Timeout if the download receives less than `limit` bytes in `timeout`
+    // seconds for `transfer_stall_timeout_` seconds.
+    status = handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, limit);
     if (!status.ok()) return OnTransferError(std::move(status));
     status = handle_.SetOption(CURLOPT_LOW_SPEED_TIME, timeout);
     if (!status.ok()) return OnTransferError(std::move(status));

google/cloud/storage/internal/curl_download_request.h

@@ -135,6 +135,7 @@ class CurlDownloadRequest : public ObjectReadSource {
   bool logging_enabled_ = false;
   CurlHandle::SocketOptions socket_options_;
   std::chrono::seconds download_stall_timeout_;
+  std::uint32_t download_stall_minimum_rate_ = 0;
   CurlHandle handle_;
   rest_internal::CurlMulti multi_;
   std::shared_ptr<rest_internal::CurlHandleFactory> factory_;

google/cloud/storage/internal/curl_request.cc

@@ -120,10 +120,12 @@ StatusOr<HttpResponse> CurlRequest::MakeRequestImpl() {
   if (transfer_stall_timeout_.count() != 0) {
     // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* `long`
     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_);
     handle_.SetOption(CURLOPT_CONNECTTIMEOUT, timeout);
     // Timeout if the request sends or receives less than 1 byte/second (i.e.
     // effectively no bytes) for `transfer_stall_timeout_` seconds.
-    handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, 1L);
+    handle_.SetOption(CURLOPT_LOW_SPEED_LIMIT, limit);
     handle_.SetOption(CURLOPT_LOW_SPEED_TIME, timeout);
   }
   auto status = handle_.EasyPerform();

google/cloud/storage/internal/curl_request.h

@@ -78,6 +78,7 @@ class CurlRequest {
   bool logging_enabled_ = false;
   CurlHandle::SocketOptions socket_options_;
   std::chrono::seconds transfer_stall_timeout_;
+  std::uint32_t transfer_stall_minimum_rate_;
   CurlHandle handle_;
   std::shared_ptr<rest_internal::CurlHandleFactory> factory_;
 };

google/cloud/storage/internal/curl_request_builder.cc

@@ -59,6 +59,7 @@ CurlRequest CurlRequestBuilder::BuildRequest() && {
   request.logging_enabled_ = logging_enabled_;
   request.socket_options_ = socket_options_;
   request.transfer_stall_timeout_ = transfer_stall_timeout_;
+  request.transfer_stall_minimum_rate_ = transfer_stall_minimum_rate_;
   return request;
 }
 
@@ -75,6 +76,7 @@ CurlRequestBuilder::BuildDownloadRequest() && {
   request->logging_enabled_ = logging_enabled_;
   request->socket_options_ = socket_options_;
   request->download_stall_timeout_ = download_stall_timeout_;
+  request->download_stall_minimum_rate_ = download_stall_minimum_rate_;
   auto status = request->SetOptions();
   if (!status.ok()) return status;
   return request;
@@ -95,7 +97,11 @@ CurlRequestBuilder& CurlRequestBuilder::ApplyClientOptions(
   http_version_ =
       std::move(options.get<storage_experimental::HttpVersionOption>());
   transfer_stall_timeout_ = options.get<TransferStallTimeoutOption>();
+  transfer_stall_minimum_rate_ =
+      options.get<storage_experimental::TransferStallMinimumRateOption>();
   download_stall_timeout_ = options.get<DownloadStallTimeoutOption>();
+  download_stall_minimum_rate_ =
+      options.get<storage_experimental::DownloadStallMinimumRateOption>();
   return *this;
 }
 

google/cloud/storage/internal/curl_request_builder.h

@@ -197,7 +197,9 @@ class CurlRequestBuilder {
   bool logging_enabled_ = false;
   CurlHandle::SocketOptions socket_options_;
   std::chrono::seconds transfer_stall_timeout_;
+  std::uint32_t transfer_stall_minimum_rate_ = 0;
   std::chrono::seconds download_stall_timeout_;
+  std::uint32_t download_stall_minimum_rate_ = 0;
   std::string http_version_;
 };
 

google/cloud/storage/internal/grpc_client.cc

@@ -114,6 +114,14 @@ void MaybeFinalize(google::storage::v2::WriteObjectRequest& write_request,
   }
 }
 
+std::chrono::milliseconds ScaleStallTimeout(std::chrono::milliseconds timeout,
+                                            std::uint32_t size,
+                                            std::uint32_t quantum) {
+  if (timeout == std::chrono::milliseconds(0)) return timeout;
+  if (quantum <= size || size == 0) return timeout;
+  return timeout * quantum / size;
+}
+
 Status TimeoutError(std::chrono::milliseconds timeout, std::string const& op) {
   return Status(StatusCode::kDeadlineExceeded,
                 "timeout [" + absl::FormatDuration(absl::FromChrono(timeout)) +
@@ -346,8 +354,11 @@ StatusOr<ObjectMetadata> GrpcClient::InsertObjectMedia(
   if (!r) return std::move(r).status();
   auto proto_request = *r;
 
-  auto timeout = google::cloud::internal::CurrentOptions()
-                     .get<TransferStallTimeoutOption>();
+  auto const& current = google::cloud::internal::CurrentOptions();
+  auto timeout = ScaleStallTimeout(
+      current.get<TransferStallTimeoutOption>(),
+      current.get<storage_experimental::TransferStallMinimumRateOption>(),
+      google::storage::v2::ServiceConstants::MAX_WRITE_CHUNK_BYTES);
   auto create_watchdog = [cq = background_->cq(), timeout]() mutable {
     if (timeout == std::chrono::seconds(0)) {
       return make_ready_future(false);
@@ -606,8 +617,12 @@ StatusOr<EmptyResponse> GrpcClient::DeleteResumableUpload(
 
 StatusOr<QueryResumableUploadResponse> GrpcClient::UploadChunk(
     UploadChunkRequest const& request) {
-  auto const timeout = google::cloud::internal::CurrentOptions()
-                           .get<TransferStallTimeoutOption>();
+  auto const& current = google::cloud::internal::CurrentOptions();
+  auto const timeout = ScaleStallTimeout(
+      current.get<TransferStallTimeoutOption>(),
+      current.get<storage_experimental::TransferStallMinimumRateOption>(),
+      google::storage::v2::ServiceConstants::MAX_WRITE_CHUNK_BYTES);
+
   auto create_watchdog = [cq = background_->cq(), timeout]() mutable {
     if (timeout == std::chrono::seconds(0)) {
       return make_ready_future(false);

google/cloud/storage/options.h

@@ -23,6 +23,7 @@
 #include "google/cloud/credentials.h"
 #include "google/cloud/options.h"
 #include <chrono>
+#include <cstdint>
 #include <memory>
 #include <string>
 
@@ -48,6 +49,27 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 struct HttpVersionOption {
   using Type = std::string;
 };
+
+/**
+ * The minimum accepted bytes/second transfer rate.
+ *
+ * If the average rate is below this value for the `TransferStallTimeoutOption`
+ * then the transfer is aborted.
+ */
+struct TransferStallMinimumRateOption {
+  using Type = std::int32_t;
+};
+
+/**
+ * The minimum accepted bytes/second download rate.
+ *
+ * If the average rate is below this value for the `DownloadStallTimeoutOption`
+ * then the download is aborted.
+ */
+struct DownloadStallMinimumRateOption {
+  using Type = std::int32_t;
+};
+
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_experimental