diff --git a/google/cloud/internal/curl_impl.cc b/google/cloud/internal/curl_impl.cc index 29eeac7c68c70..b25affc8a62f8 100644 --- a/google/cloud/internal/curl_impl.cc +++ b/google/cloud/internal/curl_impl.cc @@ -197,6 +197,10 @@ CurlImpl::CurlImpl(CurlHandle handle, http_version_ = options.get(); + if (options.has()) { + connect_timeout_ms_ = options.get(); + } + transfer_stall_timeout_ = options.get(); transfer_stall_minimum_rate_ = options.get(); download_stall_timeout_ = options.get(); @@ -449,6 +453,15 @@ Status CurlImpl::MakeRequest(HttpMethod method, RestContext& context, status = handle_.SetOption(CURLOPT_LOW_SPEED_TIME, timeout); if (!status.ok()) return OnTransferError(context, std::move(status)); } + // Set after the stall timeouts: `CURLOPT_CONNECTTIMEOUT_MS` and + // `CURLOPT_CONNECTTIMEOUT` configure the same setting in libcurl, an + // explicitly configured connect timeout should win. + if (connect_timeout_ms_ != std::chrono::milliseconds::zero()) { + // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* long + auto const timeout_ms = static_cast(connect_timeout_ms_.count()); + status = handle_.SetOption(CURLOPT_CONNECTTIMEOUT_MS, timeout_ms); + if (!status.ok()) return OnTransferError(context, std::move(status)); + } return MakeRequestImpl(context); } diff --git a/google/cloud/internal/curl_impl.h b/google/cloud/internal/curl_impl.h index 8e08ce43811a0..e1cfeb9cf32e8 100644 --- a/google/cloud/internal/curl_impl.h +++ b/google/cloud/internal/curl_impl.h @@ -146,6 +146,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_; diff --git a/google/cloud/internal/rest_options.h b/google/cloud/internal/rest_options.h index d5644c05ccd42..266c615ab9035 100644 --- a/google/cloud/internal/rest_options.h +++ b/google/cloud/internal/rest_options.h @@ -56,6 +56,17 @@ 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. + */ +struct HttpConnectTimeoutOption { + using Type = std::chrono::milliseconds; +}; + /** * Sets the download stall timeout. * @@ -101,9 +112,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 diff --git a/google/cloud/storage/client.cc b/google/cloud/storage/client.cc index c12d8d623b5f1..f9f67de64ed2e 100644 --- a/google/cloud/storage/client.cc +++ b/google/cloud/storage/client.cc @@ -581,6 +581,23 @@ Options DefaultOptions(Options opts) { "/iamapi"); } + if (!o.has()) { + o.set(false); + } + if (!o.has()) { + o.set(0.0); + } + if (!o.has()) { + o.set(0); + } + if (!o.has()) { + o.set( + std::chrono::milliseconds(500)); + } + if (!o.has()) { + o.set(2); + } + auto logging = GetEnv("CLOUD_STORAGE_ENABLE_TRACING"); if (logging) { for (auto c : absl::StrSplit(*logging, ',')) { @@ -633,6 +650,12 @@ Options DefaultOptions(Options opts) { rest_defaults.set(o.get()); } + // The (experimental) connect timeout is mapped the same way. + if (o.has()) { + rest_defaults.set( + o.get()); + } + return google::cloud::internal::MergeOptions(std::move(o), std::move(rest_defaults)); } diff --git a/google/cloud/storage/google_cloud_cpp_storage.bzl b/google/cloud/storage/google_cloud_cpp_storage.bzl index 086ab4553a07b..2c85f8260d0bb 100644 --- a/google/cloud/storage/google_cloud_cpp_storage.bzl +++ b/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", diff --git a/google/cloud/storage/google_cloud_cpp_storage.cmake b/google/cloud/storage/google_cloud_cpp_storage.cmake index 25b47c411f7dd..9b88e828efca4 100644 --- a/google/cloud/storage/google_cloud_cpp_storage.cmake +++ b/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 diff --git a/google/cloud/storage/internal/connection_impl.cc b/google/cloud/storage/internal/connection_impl.cc index 4e51345c18893..c208a6a06e162 100644 --- a/google/cloud/storage/internal/connection_impl.cc +++ b/google/cloud/storage/internal/connection_impl.cc @@ -14,6 +14,7 @@ #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" @@ -21,6 +22,7 @@ #include "google/cloud/internal/rest_retry_loop.h" #include "google/cloud/log.h" #include "absl/strings/match.h" +#include #include #include #include @@ -155,7 +157,27 @@ std::shared_ptr StorageConnectionImpl::Create( StorageConnectionImpl::StorageConnectionImpl( std::unique_ptr stub, Options options) : stub_(std::move(stub)), - options_(MergeOptions(std::move(options), stub_->options())) {} + options_(MergeOptions(std::move(options), stub_->options())) { + if (options_.get()) { + // 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(); + if (pool_size == 0) { + pool_size = + (std::max)(4, std::thread::hardware_concurrency()); + } + auto const max_threads = 2 * pool_size; + auto const rate_limit = + options_.get(); + auto const max_concurrent = + options_.get(); + // Allow bursts of up to one second worth of hedges. + hedge_pool_ = std::make_shared( + max_threads, rate_limit, rate_limit, max_concurrent); + } +} Options StorageConnectionImpl::options() const { return options_; } @@ -392,15 +414,32 @@ StatusOr> StorageConnectionImpl::ReadObject( *current, request, where); }; - auto retry_policy = current->get()->clone(); - auto backoff_policy = current->get()->clone(); - auto child = factory(request, *retry_policy, *backoff_policy); - if (!child) return child; + auto retry_source_factory = + [factory, current, + request]() -> StatusOr> { + auto retry_policy = current->get()->clone(); + auto backoff_policy = current->get()->clone(); + auto child = factory(request, *retry_policy, *backoff_policy); + if (!child) return child; + return std::unique_ptr( + std::make_unique( + factory, current, request, *std::move(child), + std::move(retry_policy), std::move(backoff_policy))); + }; + + auto const enable_hedging = + current->get(); + auto const delay = current->get(); + auto const max_hedges = + current->get(); + + if (!enable_hedging || max_hedges <= 0 || !hedge_pool_) { + return retry_source_factory(); + } return std::unique_ptr( - std::make_unique( - std::move(factory), std::move(current), request, *std::move(child), - std::move(retry_policy), std::move(backoff_policy))); + std::make_unique( + hedge_pool_, std::move(retry_source_factory), delay, max_hedges)); } StatusOr StorageConnectionImpl::ListObjects( diff --git a/google/cloud/storage/internal/connection_impl.h b/google/cloud/storage/internal/connection_impl.h index b487aa6fd6efa..b1e2b36ae7cc7 100644 --- a/google/cloud/storage/internal/connection_impl.h +++ b/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 stub_; Options options_; + std::shared_ptr hedge_pool_; google::cloud::internal::InvocationIdGenerator invocation_id_generator_; }; diff --git a/google/cloud/storage/internal/hedged_object_read_source.cc b/google/cloud/storage/internal/hedged_object_read_source.cc new file mode 100644 index 0000000000000..198a48e2768bf --- /dev/null +++ b/google/cloud/storage/internal/hedged_object_read_source.cc @@ -0,0 +1,138 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "google/cloud/storage/internal/hedged_object_read_source.h" +#include +#include +#include +#include + +namespace google { +namespace cloud { +namespace storage { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +namespace internal { +namespace { + +struct RaceResult { + StatusOr result; + std::unique_ptr source; + std::unique_ptr buffer; +}; + +struct RaceState { + std::promise promise; + std::atomic resolved{false}; +}; + +// Opens a new child and performs its initial read, resolving the race if this +// attempt finishes first. Losing attempts close their child. Only the primary +// attempt resolves the race on an open error: a hedge that fails to open must +// not mask a slower, but successful, primary. +void RunAttempt(std::shared_ptr const& state, + HedgedObjectReadSource::ChildFactory const& factory, + std::size_t n, bool resolve_on_open_error, + std::shared_ptr release_slot) { + struct SlotGuard { + std::shared_ptr pool; + ~SlotGuard() { + if (pool) pool->ReleaseHedgeSlot(); + } + } guard{std::move(release_slot)}; + + auto source = factory(); + if (!source) { + if (!resolve_on_open_error) return; + auto expected = false; + if (state->resolved.compare_exchange_strong(expected, true)) { + state->promise.set_value( + RaceResult{std::move(source).status(), nullptr, {}}); + } + return; + } + std::unique_ptr buffer(new char[n]); + auto result = (*source)->Read(buffer.get(), n); + auto expected = false; + if (state->resolved.compare_exchange_strong(expected, true)) { + state->promise.set_value( + RaceResult{std::move(result), *std::move(source), std::move(buffer)}); + } else { + (*source)->Close(); + } +} + +} // namespace + +HedgedObjectReadSource::HedgedObjectReadSource( + std::shared_ptr hedge_pool, ChildFactory child_factory, + std::chrono::milliseconds delay, int max_hedges) + : hedge_pool_(std::move(hedge_pool)), + child_factory_(std::move(child_factory)), + delay_(delay), + max_hedges_(max_hedges) {} + +bool HedgedObjectReadSource::IsOpen() const { + if (active_child_) return active_child_->IsOpen(); + return true; +} + +StatusOr HedgedObjectReadSource::Close() { + if (active_child_) return active_child_->Close(); + // The source was never read from, there is no child (or HTTP response) to + // close. + return HttpResponse{HttpStatusCode::kOk, {}, {}}; +} + +StatusOr HedgedObjectReadSource::Read(char* buf, + std::size_t n) { + // Only the stream open is hedged. Once a child has won the race all + // subsequent reads continue on it, at its current offset, without any + // thread hops or extra copies. + if (active_child_) return active_child_->Read(buf, n); + + auto state = std::make_shared(); + auto future = state->promise.get_future(); + + auto primary = [state, factory = child_factory_, n] { + RunAttempt(state, factory, n, /*resolve_on_open_error=*/true, nullptr); + }; + // If the pool is shutting down run the attempt inline, the read must + // complete either way. + if (!hedge_pool_->Enqueue(primary)) primary(); + + for (int i = 0; i != max_hedges_; ++i) { + if (future.wait_for(delay_) != std::future_status::timeout) break; + if (!hedge_pool_->TryAcquireHedgeToken()) continue; + auto hedge = [state, factory = child_factory_, n, pool = hedge_pool_] { + RunAttempt(state, factory, n, /*resolve_on_open_error=*/false, pool); + }; + if (!hedge_pool_->Enqueue(hedge)) { + hedge_pool_->ReleaseHedgeSlot(); + break; + } + } + + auto race = future.get(); + active_child_ = std::move(race.source); + if (race.result.ok() && race.result->bytes_received > 0) { + std::memcpy(buf, race.buffer.get(), race.result->bytes_received); + } + return race.result; +} + +} // namespace internal +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace storage +} // namespace cloud +} // namespace google diff --git a/google/cloud/storage/internal/hedged_object_read_source.h b/google/cloud/storage/internal/hedged_object_read_source.h new file mode 100644 index 0000000000000..d9d6cd511d18c --- /dev/null +++ b/google/cloud/storage/internal/hedged_object_read_source.h @@ -0,0 +1,76 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_HEDGED_OBJECT_READ_SOURCE_H +#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_HEDGED_OBJECT_READ_SOURCE_H + +#include "google/cloud/storage/internal/hedging_thread_pool.h" +#include "google/cloud/storage/internal/object_read_source.h" +#include "google/cloud/storage/version.h" +#include +#include +#include + +namespace google { +namespace cloud { +namespace storage { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +namespace internal { + +/** + * Hedge the *open* of an `ObjectReadSource` to reduce tail latency. + * + * The first `Read()` races one or more children created by `child_factory`: + * a primary attempt starts immediately, and up to @p max_hedges additional + * attempts start, staggered by @p delay, while no attempt has completed. The + * first attempt to complete its initial read wins; losing attempts are closed + * when they eventually complete. + * + * Only the initial open is hedged. `ObjectReadSource` is a stream, so a hedge + * started mid-stream would restart from the request's initial offset and + * could return the wrong bytes. After the race, all subsequent reads simply + * continue on the winning child at its current offset, with no extra threads + * or copies. + */ +class HedgedObjectReadSource : public ObjectReadSource { + public: + using ChildFactory = + std::function>()>; + + HedgedObjectReadSource(std::shared_ptr hedge_pool, + ChildFactory child_factory, + std::chrono::milliseconds delay, int max_hedges); + + ~HedgedObjectReadSource() override = default; + + bool IsOpen() const override; + StatusOr Close() override; + StatusOr Read(char* buf, std::size_t n) override; + + private: + std::shared_ptr hedge_pool_; + ChildFactory child_factory_; + std::chrono::milliseconds delay_; + int max_hedges_; + + std::unique_ptr active_child_; +}; + +} // namespace internal +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace storage +} // namespace cloud +} // namespace google + +#endif // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_HEDGED_OBJECT_READ_SOURCE_H diff --git a/google/cloud/storage/internal/hedged_object_read_source_test.cc b/google/cloud/storage/internal/hedged_object_read_source_test.cc new file mode 100644 index 0000000000000..2c72a1f6d2493 --- /dev/null +++ b/google/cloud/storage/internal/hedged_object_read_source_test.cc @@ -0,0 +1,171 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "google/cloud/storage/internal/hedged_object_read_source.h" +#include "google/cloud/storage/testing/mock_client.h" +#include "google/cloud/testing_util/status_matchers.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace google { +namespace cloud { +namespace storage { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +namespace internal { +namespace { + +using ::google::cloud::storage::testing::MockObjectReadSource; +using ::google::cloud::testing_util::IsOk; +using ::google::cloud::testing_util::StatusIs; +using ::testing::Eq; +using ::testing::Return; + +std::shared_ptr MakeUnlimitedPool() { + return std::make_shared( + /*max_threads=*/4, /*rate_limit=*/0.0, /*capacity=*/0.0, + /*max_concurrent=*/0); +} + +ReadSourceResult MakeReadResult(std::string const& payload) { + auto result = + ReadSourceResult{payload.size(), HttpResponse{HttpStatusCode::kOk, + /*payload=*/{}, + /*headers=*/{}}}; + return result; +} + +TEST(HedgedObjectReadSourceTest, PrimaryWins) { + auto factory = []() -> StatusOr> { + auto mock = std::make_unique(); + EXPECT_CALL(*mock, Read).WillOnce([](char* buf, std::size_t) { + std::string const payload = "payload"; + std::copy(payload.begin(), payload.end(), buf); + return MakeReadResult(payload); + }); + return std::unique_ptr(std::move(mock)); + }; + + HedgedObjectReadSource source(MakeUnlimitedPool(), factory, + std::chrono::milliseconds(500), + /*max_hedges=*/2); + + std::vector buffer(100); + auto result = source.Read(buffer.data(), buffer.size()); + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(result->bytes_received, Eq(7)); + EXPECT_THAT(std::string(buffer.data(), result->bytes_received), + Eq("payload")); +} + +TEST(HedgedObjectReadSourceTest, SubsequentReadsContinueOnWinner) { + // The factory must be called exactly once: after the open race is decided, + // reads must continue on the winning child without creating new children, + // otherwise the stream would restart at the wrong offset. + auto factory_calls = std::make_shared>(0); + auto factory = + [factory_calls]() -> StatusOr> { + ++*factory_calls; + auto mock = std::make_unique(); + EXPECT_CALL(*mock, Read) + .WillOnce(Return(MakeReadResult("chunk-1"))) + .WillOnce(Return(MakeReadResult("chunk-2"))); + return std::unique_ptr(std::move(mock)); + }; + + HedgedObjectReadSource source(MakeUnlimitedPool(), factory, + std::chrono::milliseconds(500), + /*max_hedges=*/2); + + std::vector buffer(100); + EXPECT_THAT(source.Read(buffer.data(), buffer.size()), IsOk()); + EXPECT_THAT(source.Read(buffer.data(), buffer.size()), IsOk()); + EXPECT_THAT(factory_calls->load(), Eq(1)); +} + +TEST(HedgedObjectReadSourceTest, HedgeWinsWhenPrimaryStalls) { + // The primary blocks until the end of the test, the hedge answers + // immediately. The read must complete with the hedge's data, and the + // (losing) primary must be closed once it completes. + auto unblock_primary = std::make_shared>(); + auto primary_closed = std::make_shared>(); + auto calls = std::make_shared>(0); + auto factory = [unblock_primary, primary_closed, + calls]() -> StatusOr> { + auto mock = std::make_unique(); + if (++*calls == 1) { + EXPECT_CALL(*mock, Read).WillOnce([unblock_primary](char*, std::size_t) { + unblock_primary->get_future().get(); + return MakeReadResult("slow"); + }); + EXPECT_CALL(*mock, Close).WillOnce([primary_closed]() { + primary_closed->set_value(); + return make_status_or(HttpResponse{HttpStatusCode::kOk, {}, {}}); + }); + } else { + EXPECT_CALL(*mock, Read).WillOnce(Return(MakeReadResult("hedge"))); + } + return std::unique_ptr(std::move(mock)); + }; + + auto source = std::make_unique( + MakeUnlimitedPool(), factory, std::chrono::milliseconds(1), + /*max_hedges=*/2); + + std::vector buffer(100); + auto result = source->Read(buffer.data(), buffer.size()); + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(result->bytes_received, Eq(5)); + + unblock_primary->set_value(); + primary_closed->get_future().get(); +} + +TEST(HedgedObjectReadSourceTest, PrimaryOpenErrorPropagates) { + auto factory = []() -> StatusOr> { + return Status(StatusCode::kPermissionDenied, "uh-oh"); + }; + + HedgedObjectReadSource source(MakeUnlimitedPool(), factory, + std::chrono::milliseconds(500), + /*max_hedges=*/2); + + std::vector buffer(100); + EXPECT_THAT(source.Read(buffer.data(), buffer.size()), + StatusIs(StatusCode::kPermissionDenied)); +} + +TEST(HedgedObjectReadSourceTest, CloseWithoutReadSucceeds) { + auto factory = []() -> StatusOr> { + return Status(StatusCode::kUnimplemented, "never called"); + }; + HedgedObjectReadSource source(MakeUnlimitedPool(), factory, + std::chrono::milliseconds(500), + /*max_hedges=*/2); + EXPECT_TRUE(source.IsOpen()); + EXPECT_THAT(source.Close(), IsOk()); +} + +} // namespace +} // namespace internal +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace storage +} // namespace cloud +} // namespace google diff --git a/google/cloud/storage/internal/hedging_thread_pool.h b/google/cloud/storage/internal/hedging_thread_pool.h new file mode 100644 index 0000000000000..995e7d0f31b51 --- /dev/null +++ b/google/cloud/storage/internal/hedging_thread_pool.h @@ -0,0 +1,198 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_HEDGING_THREAD_POOL_H +#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_HEDGING_THREAD_POOL_H + +#include "google/cloud/storage/version.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace google { +namespace cloud { +namespace storage { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +namespace internal { + +/** + * A lazy, dynamically-scaling thread pool with integrated hedge throttling. + * + * The pool starts with no threads and spawns workers on demand, up to + * @p max_threads. Hedged requests are gated by `TryAcquireHedgeToken()`, + * which enforces two limits: a maximum number of concurrently active hedges, + * and a maximum rate of new hedges per second (a token bucket). + */ +class HedgingThreadPool { + private: + struct State { + std::size_t const max_threads; + std::size_t idle_threads = 0; + std::queue> tasks; + std::mutex queue_mutex; + std::condition_variable cv; + bool stop = false; + + // Concurrency limiter. + std::int64_t const max_concurrent_hedges; + std::atomic active_concurrent_hedges{0}; + + explicit State(std::size_t mt, std::int64_t mc) + : max_threads(mt), max_concurrent_hedges(mc) {} + }; + + public: + HedgingThreadPool(std::size_t max_threads, double rate_limit, double capacity, + std::int64_t max_concurrent) + : state_(std::make_shared(max_threads, max_concurrent)), + rate_limit_(rate_limit), + tokens_capacity_(capacity), + tokens_(capacity), + last_refill_(std::chrono::steady_clock::now()) {} + + ~HedgingThreadPool() { + { + std::lock_guard lock(state_->queue_mutex); + state_->stop = true; + } + state_->cv.notify_all(); + for (auto& worker : workers_) { + if (worker.joinable()) { + if (worker.get_id() == std::this_thread::get_id()) { + worker.detach(); + } else { + worker.join(); + } + } + } + } + + /** + * Schedule @p task to run on a pool thread. + * + * Returns false if the pool is shutting down, in which case the task is + * *not* scheduled. Callers waiting on the task's side effects must handle + * this case (e.g. by running the task inline), or they would block forever. + */ + bool Enqueue(std::function task) { + { + std::lock_guard lock(state_->queue_mutex); + if (state_->stop) return false; + state_->tasks.push(std::move(task)); + // Only spawn a new thread if there are no idle threads and the pool has + // not reached its thread ceiling. + if (state_->idle_threads == 0 && workers_.size() < state_->max_threads) { + SpawnWorker(); + } + } + state_->cv.notify_one(); + return true; + } + + /** + * Try to reserve capacity for one hedged request. + * + * On success the caller *must* eventually call `ReleaseHedgeSlot()`. + */ + bool TryAcquireHedgeToken() { + // Gate 1: the ceiling on concurrently active hedges. + if (state_->max_concurrent_hedges > 0) { + auto current = + state_->active_concurrent_hedges.load(std::memory_order_relaxed); + do { + if (current >= state_->max_concurrent_hedges) return false; + } while (!state_->active_concurrent_hedges.compare_exchange_weak( + current, current + 1, std::memory_order_relaxed)); + } + + // Gate 2: the rate limit on new hedges (token bucket). + if (rate_limit_ > 0.0) { + std::lock_guard lock(limiter_mutex_); + Refill(); + if (tokens_ < 1.0) { + if (state_->max_concurrent_hedges > 0) { + state_->active_concurrent_hedges.fetch_sub( + 1, std::memory_order_relaxed); + } + return false; + } + tokens_ -= 1.0; + } + + return true; + } + + void ReleaseHedgeSlot() { + if (state_->max_concurrent_hedges > 0) { + state_->active_concurrent_hedges.fetch_sub(1, std::memory_order_relaxed); + } + } + + private: + void SpawnWorker() { + workers_.emplace_back([state = state_]() { + while (true) { + std::function task; + { + std::unique_lock lock(state->queue_mutex); + ++state->idle_threads; + state->cv.wait(lock, [&state] { + return state->stop || !state->tasks.empty(); + }); + --state->idle_threads; + if (state->stop && state->tasks.empty()) return; + task = std::move(state->tasks.front()); + state->tasks.pop(); + } + task(); + } + }); + } + + void Refill() { + auto now = std::chrono::steady_clock::now(); + auto const elapsed = + std::chrono::duration_cast>(now - + last_refill_) + .count(); + last_refill_ = now; + tokens_ = (std::min)(tokens_capacity_, tokens_ + elapsed * rate_limit_); + } + + std::shared_ptr state_; + std::vector workers_; + + // Token bucket rate limiter. + double rate_limit_; + double tokens_capacity_; + double tokens_; + std::chrono::steady_clock::time_point last_refill_; + std::mutex limiter_mutex_; +}; + +} // namespace internal +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace storage +} // namespace cloud +} // namespace google + +#endif // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_HEDGING_THREAD_POOL_H diff --git a/google/cloud/storage/internal/hedging_thread_pool_test.cc b/google/cloud/storage/internal/hedging_thread_pool_test.cc new file mode 100644 index 0000000000000..2b07aa04f7193 --- /dev/null +++ b/google/cloud/storage/internal/hedging_thread_pool_test.cc @@ -0,0 +1,113 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "google/cloud/storage/internal/hedging_thread_pool.h" +#include +#include +#include +#include +#include + +namespace google { +namespace cloud { +namespace storage { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +namespace internal { +namespace { + +TEST(HedgingThreadPoolTest, EnqueueAndExecute) { + HedgingThreadPool pool(2, 0.0, 0.0, 0); + std::promise p1; + std::promise p2; + + EXPECT_TRUE(pool.Enqueue([&p1] { p1.set_value(); })); + EXPECT_TRUE(pool.Enqueue([&p2] { p2.set_value(); })); + + p1.get_future().get(); + p2.get_future().get(); +} + +TEST(HedgingThreadPoolTest, MaxConcurrentHedgesLimit) { + // Only one concurrent hedge allowed. + HedgingThreadPool pool(5, 0.0, 0.0, 1); + + EXPECT_TRUE(pool.TryAcquireHedgeToken()); + // Fails because one hedge is active. + EXPECT_FALSE(pool.TryAcquireHedgeToken()); + + pool.ReleaseHedgeSlot(); + + EXPECT_TRUE(pool.TryAcquireHedgeToken()); +} + +TEST(HedgingThreadPoolTest, RateLimiter) { + // A rate limit of 5.0 tokens per second (one token per 200ms), and a burst + // capacity of 2 tokens. + HedgingThreadPool pool(5, 5.0, 2.0, 0); + + EXPECT_TRUE(pool.TryAcquireHedgeToken()); + EXPECT_TRUE(pool.TryAcquireHedgeToken()); + // The burst capacity is exhausted. + EXPECT_FALSE(pool.TryAcquireHedgeToken()); + + // The refill is time-based, there is no way to inject a fake clock. Wait + // longer than one token's refill period, with margin for slow machines. + std::this_thread::sleep_for(std::chrono::milliseconds(250)); + + EXPECT_TRUE(pool.TryAcquireHedgeToken()); +} + +TEST(HedgingThreadPoolTest, SafeDestructionOnWorkerThread) { + // We need to test the scenario where a worker thread drops the *last* + // reference to the HedgingThreadPool while executing a task, triggering + // its destruction on the worker thread itself. + + auto pool = std::make_shared(1, 0.0, 0.0, 0); + + std::promise thread_started; + std::promise main_cleared; + auto thread_started_future = thread_started.get_future(); + auto main_cleared_future = main_cleared.get_future(); + + pool->Enqueue([pool_copy = pool, &thread_started, + &main_cleared_future]() mutable { + // 1. Signal main thread that the worker is running and owns a copy + thread_started.set_value(); + + // 2. Wait for the main thread to drop its reference to the pool + main_cleared_future.wait(); + + // 3. At this exact moment, `pool_copy` is the LAST reference to the pool. + // When this lambda returns, the worker thread will destroy the lambda, + // which destroys `pool_copy`, triggering the pool's destructor on this + // very worker thread. It must detach and not deadlock/terminate. + pool_copy.reset(); + }); + + // Wait for worker to start and take ownership + thread_started_future.wait(); + + // Drop main thread's reference + pool.reset(); + + // Signal worker to proceed with destruction + main_cleared.set_value(); +} + +} // namespace +} // namespace internal +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace storage +} // namespace cloud +} // namespace google diff --git a/google/cloud/storage/options.h b/google/cloud/storage/options.h index 2f76f7a215738..7caf80398ecf9 100644 --- a/google/cloud/storage/options.h +++ b/google/cloud/storage/options.h @@ -33,6 +33,65 @@ namespace cloud { namespace storage_experimental { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +/** + * Enable experimental request hedging for `ReadObject()` streams. + * + * When enabled, opening a download races the initial request against one or + * more delayed, duplicate ("hedged") requests, and the first to respond wins. + * This reduces tail latency at the cost of additional requests. + * + * @ingroup storage-options + */ +struct EnableReadHedgingOption { + using Type = bool; +}; + +/** + * The maximum rate of hedged requests per second across the connection. + * + * The default is 0.0, meaning no rate limit. + * + * @ingroup storage-options + */ +struct ReadHedgeRateLimitOption { + using Type = double; +}; + +/** + * The maximum number of concurrently active hedged requests across the + * connection. + * + * The default is 0, meaning no concurrency limit. + * + * @ingroup storage-options + */ +struct MaxConcurrentHedgesOption { + using Type = std::int64_t; +}; + +/** + * The delay before starting a hedged request. + * + * The default is 500 milliseconds. + * + * @ingroup storage-options + */ +struct ReadHedgeDelayOption { + using Type = std::chrono::milliseconds; +}; + +/** + * The maximum number of hedged requests per stream open. + * + * The default is 2. Set to 0 to disable hedging for reads even when + * `EnableReadHedgingOption` is set. + * + * @ingroup storage-options + */ +struct MaxReadHedgesOption { + using Type = int; +}; + /** * Set the HTTP version used by the client. * @@ -66,6 +125,19 @@ struct OTelSpanEnrichmentOption { using Type = bool; }; +/** + * 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. + * + * @ingroup storage-options + */ +struct HttpConnectTimeoutOption { + using Type = std::chrono::milliseconds; +}; + GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace storage_experimental @@ -392,6 +464,12 @@ using ClientOptionList = ::google::cloud::OptionList< IdempotencyPolicyOption, CARootsFilePathOption, UploadChecksumValidationOption, DownloadChecksumValidationOption, PrecomputedChecksumsOption, storage_experimental::HttpVersionOption, + storage_experimental::HttpConnectTimeoutOption, + storage_experimental::EnableReadHedgingOption, + storage_experimental::ReadHedgeRateLimitOption, + storage_experimental::MaxConcurrentHedgesOption, + storage_experimental::ReadHedgeDelayOption, + storage_experimental::MaxReadHedgesOption, storage_experimental::OTelSpanEnrichmentOption>; GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END diff --git a/google/cloud/storage/storage_client_unit_tests.bzl b/google/cloud/storage/storage_client_unit_tests.bzl index 5b0b074ae6a8b..8f6c874b32e8a 100644 --- a/google/cloud/storage/storage_client_unit_tests.bzl +++ b/google/cloud/storage/storage_client_unit_tests.bzl @@ -67,6 +67,8 @@ storage_client_unit_tests = [ "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",