diff --git a/google/cloud/internal/curl_impl.cc b/google/cloud/internal/curl_impl.cc index 29eeac7c68c70..a066ca7dd927d 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(); @@ -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(fallback); + if (connect_timeout == std::chrono::milliseconds::zero()) return {}; + // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* long + auto const timeout_ms = static_cast(connect_timeout.count()); + return handle_.SetOption(CURLOPT_CONNECTTIMEOUT_MS, timeout_ms); +} + void CurlImpl::MergeAndWriteHeaders( std::function 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(download_stall_timeout_.count()); // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* long auto const limit = static_cast(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(transfer_stall_timeout_.count()); // NOLINTNEXTLINE(google-runtime-int) - libcurl *requires* long auto const limit = static_cast(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); diff --git a/google/cloud/internal/curl_impl.h b/google/cloud/internal/curl_impl.h index 8e08ce43811a0..c3e2b86d23770 100644 --- a/google/cloud/internal/curl_impl.h +++ b/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_; diff --git a/google/cloud/internal/rest_options.h b/google/cloud/internal/rest_options.h index d5644c05ccd42..a95ef33e926cf 100644 --- a/google/cloud/internal/rest_options.h +++ b/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 diff --git a/google/cloud/storage/client.cc b/google/cloud/storage/client.cc index c12d8d623b5f1..2456f8f4e7014 100644 --- a/google/cloud/storage/client.cc +++ b/google/cloud/storage/client.cc @@ -581,6 +581,26 @@ 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(64 * 1024 * 1024); + } + 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 +653,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/client_test.cc b/google/cloud/storage/client_test.cc index 8f37f558c4292..887d6ba6a7b99 100644 --- a/google/cloud/storage/client_test.cc +++ b/google/cloud/storage/client_test.cc @@ -470,6 +470,21 @@ TEST_F(ClientTest, Timeouts) { internal::DefaultOptions().get()); } +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()); + + auto const options = internal::DefaultOptions( + Options{}.set( + std::chrono::milliseconds(1500))); + EXPECT_EQ(std::chrono::milliseconds(1500), + options.get()); +} + } // namespace GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace storage 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..41c9e1297c639 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,37 @@ 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(); + auto const max_buffer = + current->get(); + + 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( - 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, max_buffer)); } 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..ebd0a8441c2a1 --- /dev/null +++ b/google/cloud/storage/internal/hedged_object_read_source.cc @@ -0,0 +1,166 @@ +// 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/internal/make_status.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 (std::nothrow) char[n]); + if (!buffer) { + if (!resolve_on_open_error) return; + auto expected = false; + if (state->resolved.compare_exchange_strong(expected, true)) { + state->promise.set_value(RaceResult{ + google::cloud::internal::ResourceExhaustedError( + "Out of memory allocating hedge buffer", GCP_ERROR_INFO()), + nullptr, + {}}); + } + return; + } + 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, std::size_t max_buffer) + : hedge_pool_(std::move(hedge_pool)), + child_factory_(std::move(child_factory)), + delay_(delay), + max_hedges_(max_hedges), + max_buffer_(max_buffer) {} + +bool HedgedObjectReadSource::IsOpen() const { + if (active_child_) return active_child_->IsOpen(); + return !is_closed_; +} + +StatusOr HedgedObjectReadSource::Close() { + is_closed_ = true; + 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) { + if (is_closed_) return ReadSourceResult{}; + + // 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); + + // Racing requires one staging buffer of `n` bytes per attempt, on top of the + // caller's own buffer. For a large read that multiplication is worse than + // the tail latency it avoids, so open the stream without hedging and read + // straight into the caller's buffer. + if (n > max_buffer_) { + auto child = child_factory_(); + if (!child) return std::move(child).status(); + active_child_ = *std::move(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..b7c8930fcfc7b --- /dev/null +++ b/google/cloud/storage/internal/hedged_object_read_source.h @@ -0,0 +1,85 @@ +// 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. + * + * Each racing attempt reads into its own buffer, because a losing attempt + * keeps writing until it completes and must not touch the caller's buffer. + * Peak memory for the race is therefore proportional to the size of the first + * read. Reads larger than @p max_buffer are served without hedging, directly + * into the caller's buffer, so a large read cannot multiply memory use. + */ +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, + std::size_t max_buffer); + + ~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::size_t max_buffer_; + + std::unique_ptr active_child_; + bool is_closed_ = false; +}; + +} // 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..ab28f2e1a2dae --- /dev/null +++ b/google/cloud/storage/internal/hedged_object_read_source_test.cc @@ -0,0 +1,258 @@ +// 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; + +// Large enough that no test read is treated as oversized. +auto constexpr kUnlimitedBuffer = std::size_t{1} << 30; + +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, kUnlimitedBuffer); + + 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, kUnlimitedBuffer); + + 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, kUnlimitedBuffer); + + 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, kUnlimitedBuffer); + + 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, kUnlimitedBuffer); + EXPECT_TRUE(source.IsOpen()); + EXPECT_THAT(source.Close(), IsOk()); +} + +TEST(HedgedObjectReadSourceTest, CloseBeforeRead) { + auto pool = std::make_shared(1, 0.0, 0.0, 0); + auto factory = []() { + return std::unique_ptr( + std::make_unique()); + }; + HedgedObjectReadSource source(pool, factory, std::chrono::milliseconds(10), 2, + kUnlimitedBuffer); + EXPECT_TRUE(source.IsOpen()); + EXPECT_STATUS_OK(source.Close()); + EXPECT_FALSE(source.IsOpen()); + auto const res = source.Read(nullptr, 1024); + EXPECT_TRUE(res.ok()); + EXPECT_EQ(res->bytes_received, 0); +} + +TEST(HedgedObjectReadSourceTest, OversizedReadIsNotHedged) { + // A read larger than the limit must open exactly one child and read into the + // caller's buffer, with no racing attempts to stage copies of the data. + auto calls = std::make_shared>(0); + auto factory = [calls]() -> StatusOr> { + ++*calls; + auto mock = std::make_unique(); + EXPECT_CALL(*mock, Read).WillOnce([](char* buf, std::size_t) { + std::string const payload = "direct"; + std::copy(payload.begin(), payload.end(), buf); + return MakeReadResult(payload); + }); + return std::unique_ptr(std::move(mock)); + }; + + // A zero delay would let a hedge start immediately if the limit were not + // honored, so any race would be observable as extra factory calls. + HedgedObjectReadSource source(MakeUnlimitedPool(), factory, + std::chrono::milliseconds(0), + /*max_hedges=*/2, /*max_buffer=*/8); + + std::vector buffer(64); + auto result = source.Read(buffer.data(), buffer.size()); + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(result->bytes_received, Eq(6)); + EXPECT_THAT(std::string(buffer.data(), result->bytes_received), Eq("direct")); + EXPECT_THAT(calls->load(), Eq(1)); +} + +TEST(HedgedObjectReadSourceTest, OversizedReadPropagatesOpenError) { + auto factory = []() -> StatusOr> { + return Status(StatusCode::kPermissionDenied, "uh-oh"); + }; + + HedgedObjectReadSource source(MakeUnlimitedPool(), factory, + std::chrono::milliseconds(0), + /*max_hedges=*/2, /*max_buffer=*/8); + + std::vector buffer(64); + EXPECT_THAT(source.Read(buffer.data(), buffer.size()), + StatusIs(StatusCode::kPermissionDenied)); +} + +TEST(HedgedObjectReadSourceTest, SubsequentReadsIgnoreBufferLimit) { + // The limit only decides whether the *open* is hedged. Once a child exists, + // reads of any size continue on it without staging buffers. + auto calls = std::make_shared>(0); + auto factory = [calls]() -> StatusOr> { + ++*calls; + auto mock = std::make_unique(); + EXPECT_CALL(*mock, Read) + .WillOnce(Return(MakeReadResult("small"))) + .WillOnce(Return(MakeReadResult("large"))); + return std::unique_ptr(std::move(mock)); + }; + + HedgedObjectReadSource source(MakeUnlimitedPool(), factory, + std::chrono::milliseconds(500), + /*max_hedges=*/2, /*max_buffer=*/64); + + std::vector small(8); + EXPECT_THAT(source.Read(small.data(), small.size()), IsOk()); + // Well past the limit, but the winner is already open. + std::vector large(4096); + EXPECT_THAT(source.Read(large.data(), large.size()), IsOk()); + EXPECT_THAT(calls->load(), Eq(1)); +} + +} // 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..c6ef44dc32d58 --- /dev/null +++ b/google/cloud/storage/internal/hedging_thread_pool.h @@ -0,0 +1,197 @@ +// 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_((std::max)(1.0, capacity)), + tokens_((std::max)(1.0, 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_]() mutable { + 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..043f40232ce17 --- /dev/null +++ b/google/cloud/storage/internal/hedging_thread_pool_test.cc @@ -0,0 +1,128 @@ +// 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) { + // Declare the promises *before* the pool. `get()` returns as soon as the + // shared state is ready, which may be before the worker has returned from + // `set_value()`. Destroying the pool joins the workers, so declaring it last + // means the workers are done before the promises they reference go away. + std::promise p1; + std::promise p2; + auto f1 = p1.get_future(); + auto f2 = p2.get_future(); + + HedgingThreadPool pool(2, 0.0, 0.0, 0); + EXPECT_TRUE(pool.Enqueue([&p1] { p1.set_value(); })); + EXPECT_TRUE(pool.Enqueue([&p2] { p2.set_value(); })); + + f1.get(); + f2.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, FractionalRateLimiter) { + // A rate limit of 0.5 tokens per second (one token per 2 seconds). + // The capacity is set to 0.5, which the pool must clamp to a floor of 1.0. + HedgingThreadPool pool(5, 0.5, 0.5, 0); + + // Since capacity is clamped to 1.0, we must be able to acquire at least one + // token. + EXPECT_TRUE(pool.TryAcquireHedgeToken()); + EXPECT_FALSE(pool.TryAcquireHedgeToken()); +} + +TEST(HedgingThreadPoolTest, SafeDestructionOnWorkerThread) { + // Dropping the last reference to the pool from inside a task runs the pool + // destructor on one of its own worker threads. It must detach that thread + // rather than join itself. + // + // Every synchronization object is shared and captured by value: this worker + // is detached, so it can still be running after this function returns and + // must not reference anything on the test's stack. + auto started = std::make_shared>(); + auto destroyed = std::make_shared>(); + auto release = std::make_shared>(); + auto started_future = started->get_future(); + auto destroyed_future = destroyed->get_future(); + auto release_future = release->get_future().share(); + + auto pool = std::make_shared(1, 0.0, 0.0, 0); + ASSERT_TRUE(pool->Enqueue( + [pool_copy = pool, started, destroyed, release_future]() mutable { + started->set_value(); + release_future.wait(); + // The test thread has dropped its reference by now, so this is the + // last one: the pool destructor runs on this worker thread. + pool_copy.reset(); + destroyed->set_value(); + })); + + started_future.get(); + pool.reset(); + release->set_value(); + // Wait for the destructor to finish on the worker thread. This replaces a + // timing-based sleep: the test cannot return while the pool is still being + // destroyed. + destroyed_future.get(); +} + +} // 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..a7d192a2463b0 100644 --- a/google/cloud/storage/options.h +++ b/google/cloud/storage/options.h @@ -33,6 +33,83 @@ 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 largest read, in bytes, that is eligible for hedging. + * + * Racing requests each buffer their own copy of the data, so the memory used + * while opening a stream grows with the size of the first read. A read larger + * than this value is served without hedging, reading directly into the + * application's buffer, which bounds that growth. Note this is the size the + * application asks for in a single read (e.g. `stream.read(buf, n)`), not the + * size of the object or of a requested range. + * + * The default is 64 MiB (64 * 1024 * 1024). + * + * @ingroup storage-options + */ +struct MaximumHedgeBufferOption { + using Type = std::size_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 +143,24 @@ 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. + * + * This applies to all requests, not just downloads, and it only bounds + * establishing the connection: it has no effect once bytes start flowing. Use + * `TransferStallTimeoutOption` and `DownloadStallTimeoutOption` to bound + * stalled transfers. + * + * @ingroup storage-options + */ +struct HttpConnectTimeoutOption { + using Type = std::chrono::milliseconds; +}; + GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace storage_experimental @@ -392,6 +487,13 @@ 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::MaximumHedgeBufferOption, + 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",