diff --git a/google/cloud/storage/internal/connection_impl.cc b/google/cloud/storage/internal/connection_impl.cc index bac46cfd22f53..69ded70ddcbcb 100644 --- a/google/cloud/storage/internal/connection_impl.cc +++ b/google/cloud/storage/internal/connection_impl.cc @@ -159,6 +159,7 @@ StorageConnectionImpl::StorageConnectionImpl( : stub_(std::move(stub)), options_(MergeOptions(std::move(options), stub_->options())) { if (options_.get()) { + hedged_read_metrics_ = std::make_shared(); // `DefaultOptions()` normally resolves these, but a connection can be // built without it, in which case the option is left at 0 ("automatic"). // A pool sized 0 would accept reads it never runs, hanging the caller. @@ -448,9 +449,9 @@ StatusOr> StorageConnectionImpl::ReadObject( // `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(read_pool_, hedge_pool_, - std::move(retry_source_factory), - delay, max_hedges, max_buffer)); + std::make_unique( + read_pool_, hedge_pool_, std::move(retry_source_factory), delay, + max_hedges, max_buffer, hedged_read_metrics_)); } StatusOr StorageConnectionImpl::ListObjects( diff --git a/google/cloud/storage/internal/connection_impl.h b/google/cloud/storage/internal/connection_impl.h index 895e943a93227..abef746724bcf 100644 --- a/google/cloud/storage/internal/connection_impl.h +++ b/google/cloud/storage/internal/connection_impl.h @@ -32,6 +32,8 @@ namespace cloud { namespace storage { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN namespace internal { +class HedgedReadMetrics; + /** * Decorates a `StorageConnection` to retry each operation. */ @@ -190,6 +192,7 @@ class StorageConnectionImpl Options options_; std::shared_ptr read_pool_; std::shared_ptr hedge_pool_; + std::shared_ptr hedged_read_metrics_; 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 index 8378ba8f6ad29..3a65ed087fbb9 100644 --- a/google/cloud/storage/internal/hedged_object_read_source.cc +++ b/google/cloud/storage/internal/hedged_object_read_source.cc @@ -14,6 +14,8 @@ #include "google/cloud/storage/internal/hedged_object_read_source.h" #include "google/cloud/internal/make_status.h" +#include +#include #include #include #include @@ -24,12 +26,42 @@ namespace cloud { namespace storage { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN namespace internal { + +HedgedReadMetrics::HedgedReadMetrics() + : HedgedReadMetrics(opentelemetry::metrics::Provider::GetMeterProvider()) {} + +HedgedReadMetrics::HedgedReadMetrics( + opentelemetry::nostd::shared_ptr< + opentelemetry::metrics::MeterProvider> const& provider) { + if (!provider) return; + opentelemetry::nostd::shared_ptr meter = + provider->GetMeter("gl-cpp", version_string()); + if (!meter) return; + + hedges_dispatched_ = meter->CreateUInt64Counter( + "storage.read_hedging.hedges_dispatched", + "Total number of speculative hedge read attempts dispatched", "{hedge}"); + hedge_won_ = meter->CreateUInt64Counter( + "storage.read_hedging.hedge_won", + "Total number of hedged read operations won by a secondary hedge attempt", + "{request}"); +} + +void HedgedReadMetrics::IncrementHedgesDispatched() { + if (hedges_dispatched_) hedges_dispatched_->Add(1); +} + +void HedgedReadMetrics::IncrementHedgeWon() { + if (hedge_won_) hedge_won_->Add(1); +} + namespace { struct RaceResult { StatusOr result; std::unique_ptr source; std::unique_ptr buffer; + bool is_primary; }; struct RaceState { @@ -39,12 +71,13 @@ struct RaceState { // 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 +// attempt resolves the race on open or read errors: a hedge that fails 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) { + std::shared_ptr release_slot, + bool is_primary) { // Releases the acquired hedge concurrency slot upon function exit across // all code paths (early return on open/allocation error, race winner, or // race loser). For primary attempts, release_slot is nullptr. @@ -55,13 +88,13 @@ void RunAttempt(std::shared_ptr const& state, } } guard{std::move(release_slot)}; - auto source = factory(); + StatusOr> source = factory(); if (!source) { if (!resolve_on_open_error) return; bool expected = false; if (state->resolved.compare_exchange_strong(expected, true)) { state->promise.set_value( - RaceResult{std::move(source).status(), nullptr, {}}); + RaceResult{std::move(source).status(), nullptr, {}, is_primary}); } return; } @@ -74,32 +107,79 @@ void RunAttempt(std::shared_ptr const& state, google::cloud::internal::ResourceExhaustedError( "Out of memory allocating hedge buffer", GCP_ERROR_INFO()), nullptr, - {}}); + {}, + is_primary}); } return; } - auto result = (*source)->Read(buffer.get(), n); + StatusOr result = (*source)->Read(buffer.get(), n); + if (!result && !resolve_on_open_error) { + (*source)->Close(); + return; + } bool expected = false; if (state->resolved.compare_exchange_strong(expected, true)) { - state->promise.set_value( - RaceResult{std::move(result), *std::move(source), std::move(buffer)}); + state->promise.set_value(RaceResult{std::move(result), *std::move(source), + std::move(buffer), is_primary}); } else { (*source)->Close(); } } +void DispatchHedges(std::shared_ptr const& state, + std::future& future, std::size_t n, + int max_hedges, std::chrono::milliseconds delay, + std::shared_ptr const& hedge_pool, + HedgedObjectReadSource::ChildFactory const& child_factory, + HedgedReadMetrics* metrics) { + for (int hedges_dispatched = 0; hedges_dispatched < max_hedges;) { + if (future.wait_for(delay) != std::future_status::timeout) break; + if (!hedge_pool->TryAcquireHedgeToken()) { + // When delay is 0ms (or token acquisition fails), back off briefly on + // the future instead of busy-spinning if tokens or concurrency slots are + // temporarily exhausted. + if (delay == std::chrono::milliseconds::zero() && + future.wait_for(std::chrono::milliseconds(10)) != + std::future_status::timeout) { + break; + } + continue; + } + auto hedge = [state, factory = child_factory, n, pool = hedge_pool] { + RunAttempt(state, factory, n, /*resolve_on_open_error=*/false, pool, + /*is_primary=*/false); + }; + if (!hedge_pool->Enqueue(hedge)) { + hedge_pool->ReleaseHedgeSlot(); + break; + } + ++hedges_dispatched; + if (metrics) metrics->IncrementHedgesDispatched(); + } +} + } // namespace HedgedObjectReadSource::HedgedObjectReadSource( std::shared_ptr read_pool, std::shared_ptr hedge_pool, ChildFactory child_factory, std::chrono::milliseconds delay, int max_hedges, std::size_t max_buffer) + : HedgedObjectReadSource(std::move(read_pool), std::move(hedge_pool), + std::move(child_factory), delay, max_hedges, + max_buffer, nullptr) {} + +HedgedObjectReadSource::HedgedObjectReadSource( + std::shared_ptr read_pool, + std::shared_ptr hedge_pool, ChildFactory child_factory, + std::chrono::milliseconds delay, int max_hedges, std::size_t max_buffer, + std::shared_ptr metrics) : read_pool_(std::move(read_pool)), hedge_pool_(std::move(hedge_pool)), child_factory_(std::move(child_factory)), delay_(delay), max_hedges_(max_hedges), - max_buffer_(max_buffer) {} + max_buffer_(max_buffer), + metrics_(std::move(metrics)) {} bool HedgedObjectReadSource::IsOpen() const { if (active_child_) return active_child_->IsOpen(); @@ -130,48 +210,32 @@ StatusOr HedgedObjectReadSource::Read(char* buf, // 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_(); + StatusOr> 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(); + std::future future = state->promise.get_future(); auto primary = [state, factory = child_factory_, n] { - RunAttempt(state, factory, n, /*resolve_on_open_error=*/true, nullptr); + RunAttempt(state, factory, n, /*resolve_on_open_error=*/true, nullptr, + /*is_primary=*/true); }; // The primary attempt is scheduled on the dedicated read pool. // If the pool is shutting down run the attempt inline, the read must // complete either way. if (!read_pool_->Enqueue(primary)) primary(); - for (int hedges_dispatched = 0; hedges_dispatched < max_hedges_;) { - if (future.wait_for(delay_) != std::future_status::timeout) break; - if (!hedge_pool_->TryAcquireHedgeToken()) { - // When delay_ is 0ms (or token acquisition fails), back off briefly on - // the future instead of busy-spinning if tokens or concurrency slots are - // temporarily exhausted. - if (delay_ == std::chrono::milliseconds::zero()) { - if (future.wait_for(std::chrono::milliseconds(10)) != - std::future_status::timeout) { - break; - } - } - 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; - } - ++hedges_dispatched; + DispatchHedges(state, future, n, max_hedges_, delay_, hedge_pool_, + child_factory_, metrics_.get()); + + RaceResult race = future.get(); + if (metrics_ && !race.is_primary && race.result.ok()) { + metrics_->IncrementHedgeWon(); } - 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); diff --git a/google/cloud/storage/internal/hedged_object_read_source.h b/google/cloud/storage/internal/hedged_object_read_source.h index c224b3ecedafb..5e26a5db4f7a5 100644 --- a/google/cloud/storage/internal/hedged_object_read_source.h +++ b/google/cloud/storage/internal/hedged_object_read_source.h @@ -18,7 +18,12 @@ #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 +#include #include +#include #include #include @@ -28,6 +33,37 @@ namespace storage { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN namespace internal { +/** + * OpenTelemetry metrics helper for read hedging operations. + */ +class HedgedReadMetrics { + public: + HedgedReadMetrics(); + explicit HedgedReadMetrics( + opentelemetry::nostd::shared_ptr< + opentelemetry::metrics::MeterProvider> const& provider); + + ~HedgedReadMetrics() = default; + HedgedReadMetrics(HedgedReadMetrics const&) = delete; + HedgedReadMetrics& operator=(HedgedReadMetrics const&) = delete; + HedgedReadMetrics(HedgedReadMetrics&&) = default; + HedgedReadMetrics& operator=(HedgedReadMetrics&&) = default; + + /// Records that a speculative hedge read attempt was dispatched. + void IncrementHedgesDispatched(); + + /// Records that a hedged read operation was won by a hedge attempt. + void IncrementHedgeWon(); + + private: + opentelemetry::nostd::unique_ptr< + opentelemetry::metrics::Counter> + hedges_dispatched_; + opentelemetry::nostd::unique_ptr< + opentelemetry::metrics::Counter> + hedge_won_; +}; + /** * Hedge the *open* of an `ObjectReadSource` to reduce tail latency. * @@ -60,6 +96,13 @@ class HedgedObjectReadSource : public ObjectReadSource { std::chrono::milliseconds delay, int max_hedges, std::size_t max_buffer); + HedgedObjectReadSource(std::shared_ptr read_pool, + std::shared_ptr hedge_pool, + ChildFactory child_factory, + std::chrono::milliseconds delay, int max_hedges, + std::size_t max_buffer, + std::shared_ptr metrics); + ~HedgedObjectReadSource() override = default; bool IsOpen() const override; @@ -73,6 +116,7 @@ class HedgedObjectReadSource : public ObjectReadSource { std::chrono::milliseconds delay_; int max_hedges_; std::size_t max_buffer_; + std::shared_ptr metrics_; std::unique_ptr active_child_; bool is_closed_ = false; diff --git a/google/cloud/storage/internal/hedged_object_read_source_test.cc b/google/cloud/storage/internal/hedged_object_read_source_test.cc index 0b934d3a31216..b3668124fc0e1 100644 --- a/google/cloud/storage/internal/hedged_object_read_source_test.cc +++ b/google/cloud/storage/internal/hedged_object_read_source_test.cc @@ -14,15 +14,18 @@ #include "google/cloud/storage/internal/hedged_object_read_source.h" #include "google/cloud/storage/testing/mock_client.h" +#include "google/cloud/testing_util/mock_opentelemetry_metrics.h" #include "google/cloud/testing_util/status_matchers.h" #include #include #include #include #include +#include #include #include #include +#include #include namespace google { @@ -34,8 +37,13 @@ namespace { using ::google::cloud::storage::testing::MockObjectReadSource; using ::google::cloud::testing_util::IsOk; +using ::google::cloud::testing_util::MockCounter; +using ::google::cloud::testing_util::MockMeter; +using ::google::cloud::testing_util::MockMeterProvider; using ::google::cloud::testing_util::StatusIs; +using ::testing::A; using ::testing::Eq; +using ::testing::Return; // Large enough that no test read is treated as oversized. std::size_t constexpr kUnlimitedBuffer = std::size_t{1} << 30; @@ -526,6 +534,389 @@ TEST(HedgedObjectReadSourceTest, SubsequentReadsIgnoreBufferLimit) { EXPECT_THAT(calls->load(), Eq(1)); } +TEST(HedgedObjectReadSourceTest, MetricsInterface) { + HedgedReadMetrics metrics; + metrics.IncrementHedgesDispatched(); + metrics.IncrementHedgeWon(); +} + +TEST(HedgedObjectReadSourceTest, PrimaryWinsMetrics) { + auto mock_hedges_dispatched = std::make_unique>(); + EXPECT_CALL(*mock_hedges_dispatched, Add(A())).Times(0); + + auto mock_hedge_won = std::make_unique>(); + EXPECT_CALL(*mock_hedge_won, Add(A())).Times(0); + + auto mock_meter = std::make_shared(); + EXPECT_CALL(*mock_meter, CreateUInt64Counter) + .WillOnce([mock = std::move(mock_hedges_dispatched)]( + opentelemetry::nostd::string_view name, + opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) mutable { + EXPECT_THAT(name, Eq("storage.read_hedging.hedges_dispatched")); + return std::move(mock); + }) + .WillOnce([mock = std::move(mock_hedge_won)]( + opentelemetry::nostd::string_view name, + opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) mutable { + EXPECT_THAT(name, Eq("storage.read_hedging.hedge_won")); + return std::move(mock); + }); + + auto mock_provider = std::make_shared(); + EXPECT_CALL(*mock_provider, GetMeter).WillOnce(Return(mock_meter)); + + auto metrics = std::make_shared(mock_provider); + + 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(MakeUnlimitedReadPool(), + MakeUnlimitedHedgePool(), factory, + std::chrono::milliseconds(500), + /*max_hedges=*/2, kUnlimitedBuffer, metrics); + + std::vector buffer(100); + StatusOr result = source.Read(buffer.data(), buffer.size()); + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(result->bytes_received, Eq(7)); +} + +TEST(HedgedObjectReadSourceTest, HedgeWinsMetrics) { + auto mock_hedges_dispatched = std::make_unique>(); + EXPECT_CALL(*mock_hedges_dispatched, Add(std::uint64_t{1})); + + auto mock_hedge_won = std::make_unique>(); + EXPECT_CALL(*mock_hedge_won, Add(std::uint64_t{1})); + + auto mock_meter = std::make_shared(); + EXPECT_CALL(*mock_meter, CreateUInt64Counter) + .WillOnce([mock = std::move(mock_hedges_dispatched)]( + opentelemetry::nostd::string_view name, + opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) mutable { + EXPECT_THAT(name, Eq("storage.read_hedging.hedges_dispatched")); + return std::move(mock); + }) + .WillOnce([mock = std::move(mock_hedge_won)]( + opentelemetry::nostd::string_view name, + opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) mutable { + EXPECT_THAT(name, Eq("storage.read_hedging.hedge_won")); + return std::move(mock); + }); + + auto mock_provider = std::make_shared(); + EXPECT_CALL(*mock_provider, GetMeter).WillOnce(Return(mock_meter)); + + auto metrics = std::make_shared(mock_provider); + + auto unblock_primary = std::make_shared>(); + auto primary_closed = std::make_shared>(); + auto calls = std::make_shared>(0); + auto factory = + MakeStallingPrimaryFactory(unblock_primary, primary_closed, calls); + + HedgedObjectReadSource source(MakeUnlimitedReadPool(), + MakeUnlimitedHedgePool(), factory, + std::chrono::milliseconds(10), + /*max_hedges=*/1, kUnlimitedBuffer, metrics); + + std::vector buffer(100); + StatusOr result = source.Read(buffer.data(), buffer.size()); + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(result->bytes_received, Eq(5)); + EXPECT_THAT(std::string(buffer.data(), result->bytes_received), Eq("hedge")); + + unblock_primary->set_value(); + primary_closed->get_future().get(); +} + +TEST(HedgedObjectReadSourceTest, NullProviderSafe) { + HedgedReadMetrics metrics(nullptr); + metrics.IncrementHedgesDispatched(); + metrics.IncrementHedgeWon(); +} + +TEST(HedgedObjectReadSourceTest, NullMeterSafe) { + auto mock_provider = std::make_shared(); + EXPECT_CALL(*mock_provider, GetMeter).WillOnce(Return(nullptr)); + + HedgedReadMetrics metrics(mock_provider); + metrics.IncrementHedgesDispatched(); + metrics.IncrementHedgeWon(); +} + +TEST(HedgedObjectReadSourceTest, NullMetricsSafe) { + 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(MakeUnlimitedReadPool(), + MakeUnlimitedHedgePool(), factory, + std::chrono::milliseconds(500), + /*max_hedges=*/2, kUnlimitedBuffer, + /*metrics=*/nullptr); + + std::vector buffer(100); + StatusOr result = source.Read(buffer.data(), buffer.size()); + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(result->bytes_received, Eq(7)); +} + +TEST(HedgedObjectReadSourceTest, NullMetricsSafeHedgeWins) { + auto unblock_primary = std::make_shared>(); + auto primary_closed = std::make_shared>(); + auto calls = std::make_shared>(0); + auto factory = + MakeStallingPrimaryFactory(unblock_primary, primary_closed, calls); + + HedgedObjectReadSource source(MakeUnlimitedReadPool(), + MakeUnlimitedHedgePool(), factory, + std::chrono::milliseconds(10), + /*max_hedges=*/1, kUnlimitedBuffer, + /*metrics=*/nullptr); + + std::vector buffer(100); + StatusOr result = source.Read(buffer.data(), buffer.size()); + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(result->bytes_received, Eq(5)); + EXPECT_THAT(std::string(buffer.data(), result->bytes_received), Eq("hedge")); + + unblock_primary->set_value(); + primary_closed->get_future().get(); +} + +TEST(HedgedObjectReadSourceTest, MultipleHedgesDispatchedPrimaryWins) { + auto mock_hedges_dispatched = std::make_unique>(); + EXPECT_CALL(*mock_hedges_dispatched, Add(std::uint64_t{1})).Times(2); + + auto mock_hedge_won = std::make_unique>(); + EXPECT_CALL(*mock_hedge_won, Add(A())).Times(0); + + auto mock_meter = std::make_shared(); + EXPECT_CALL(*mock_meter, CreateUInt64Counter) + .WillOnce([mock = std::move(mock_hedges_dispatched)]( + opentelemetry::nostd::string_view name, + opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) mutable { + EXPECT_THAT(name, Eq("storage.read_hedging.hedges_dispatched")); + return std::move(mock); + }) + .WillOnce([mock = std::move(mock_hedge_won)]( + opentelemetry::nostd::string_view name, + opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) mutable { + EXPECT_THAT(name, Eq("storage.read_hedging.hedge_won")); + return std::move(mock); + }); + + auto mock_provider = std::make_shared(); + EXPECT_CALL(*mock_provider, GetMeter).WillOnce(Return(mock_meter)); + + auto metrics = std::make_shared(mock_provider); + + auto unblock_primary = std::make_shared>(); + auto unblock_hedges = std::make_shared>(); + std::shared_future hedge_future = unblock_hedges->get_future().share(); + auto hedges_dispatched_count = std::make_shared>(0); + auto calls = std::make_shared>(0); + + auto factory = [unblock_primary, hedge_future, hedges_dispatched_count, + calls]() -> StatusOr> { + int call_count = ++*calls; + auto mock = std::make_unique(); + if (call_count == 1) { + EXPECT_CALL(*mock, Read) + .WillOnce([unblock_primary](char* buf, std::size_t) { + unblock_primary->get_future().get(); + std::string const payload = "primary"; + std::copy(payload.begin(), payload.end(), buf); + return MakeReadResult(payload); + }); + } else { + EXPECT_CALL(*mock, Read) + .WillOnce( + [hedge_future, hedges_dispatched_count](char* buf, std::size_t) { + ++*hedges_dispatched_count; + hedge_future.get(); + std::string const payload = "hedge"; + std::copy(payload.begin(), payload.end(), buf); + return MakeReadResult(payload); + }); + EXPECT_CALL(*mock, Close).WillOnce([]() { + return make_status_or(HttpResponse{HttpStatusCode::kOk, {}, {}}); + }); + } + return std::unique_ptr(std::move(mock)); + }; + + HedgedObjectReadSource source(MakeUnlimitedReadPool(), + MakeUnlimitedHedgePool(), factory, + std::chrono::milliseconds(5), + /*max_hedges=*/2, kUnlimitedBuffer, metrics); + + std::vector buffer(100); + std::thread trigger([unblock_primary, hedges_dispatched_count] { + while (hedges_dispatched_count->load() < 2) { + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + unblock_primary->set_value(); + }); + + StatusOr result = source.Read(buffer.data(), buffer.size()); + trigger.join(); + + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(std::string(buffer.data(), result->bytes_received), + Eq("primary")); + + unblock_hedges->set_value(); +} + +TEST(HedgedObjectReadSourceTest, HedgeReadErrorDoesNotIncrementHedgeWon) { + auto mock_hedges_dispatched = std::make_unique>(); + EXPECT_CALL(*mock_hedges_dispatched, Add(std::uint64_t{1})).Times(1); + + auto mock_hedge_won = std::make_unique>(); + EXPECT_CALL(*mock_hedge_won, Add(A())).Times(0); + + auto mock_meter = std::make_shared(); + EXPECT_CALL(*mock_meter, CreateUInt64Counter) + .WillOnce([mock = std::move(mock_hedges_dispatched)]( + opentelemetry::nostd::string_view name, + opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) mutable { + EXPECT_THAT(name, Eq("storage.read_hedging.hedges_dispatched")); + return std::move(mock); + }) + .WillOnce([mock = std::move(mock_hedge_won)]( + opentelemetry::nostd::string_view name, + opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) mutable { + EXPECT_THAT(name, Eq("storage.read_hedging.hedge_won")); + return std::move(mock); + }); + + auto mock_provider = std::make_shared(); + EXPECT_CALL(*mock_provider, GetMeter).WillOnce(Return(mock_meter)); + + auto metrics = std::make_shared(mock_provider); + + auto unblock_primary = std::make_shared>(); + auto calls = std::make_shared>(0); + auto factory = [unblock_primary, + calls]() -> StatusOr> { + int call_count = ++*calls; + auto mock = std::make_unique(); + if (call_count == 1) { + EXPECT_CALL(*mock, Read) + .WillOnce([unblock_primary](char* buf, std::size_t) { + unblock_primary->get_future().get(); + std::string const payload = "primary"; + std::copy(payload.begin(), payload.end(), buf); + return MakeReadResult(payload); + }); + } else { + EXPECT_CALL(*mock, Read).WillOnce([](char*, std::size_t) { + return Status(StatusCode::kUnavailable, "hedge read failed"); + }); + EXPECT_CALL(*mock, Close).WillOnce([]() { + return make_status_or(HttpResponse{HttpStatusCode::kOk, {}, {}}); + }); + } + return std::unique_ptr(std::move(mock)); + }; + + HedgedObjectReadSource source(MakeUnlimitedReadPool(), + MakeUnlimitedHedgePool(), factory, + std::chrono::milliseconds(5), + /*max_hedges=*/1, kUnlimitedBuffer, metrics); + + std::vector buffer(100); + std::thread unblocker([unblock_primary] { + std::this_thread::sleep_for(std::chrono::milliseconds(30)); + unblock_primary->set_value(); + }); + + StatusOr result = source.Read(buffer.data(), buffer.size()); + unblocker.join(); + + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(std::string(buffer.data(), result->bytes_received), + Eq("primary")); + EXPECT_THAT(calls->load(), Eq(2)); +} + +TEST(HedgedObjectReadSourceTest, HedgeReadErrorIgnoredWhenPrimarySucceeds) { + auto unblock_primary = std::make_shared>(); + auto calls = std::make_shared>(0); + auto factory = [unblock_primary, + calls]() -> StatusOr> { + int call_count = ++*calls; + auto mock = std::make_unique(); + if (call_count == 1) { + // Primary attempt: stalls until unblocked, then succeeds. + EXPECT_CALL(*mock, Read) + .WillOnce([unblock_primary](char* buf, std::size_t) { + unblock_primary->get_future().get(); + std::string const payload = "primary"; + std::copy(payload.begin(), payload.end(), buf); + return MakeReadResult(payload); + }); + } else { + // Hedge attempt: opens successfully, but Read fails. + EXPECT_CALL(*mock, Read).WillOnce([](char*, std::size_t) { + return Status(StatusCode::kUnavailable, "hedge read failed"); + }); + EXPECT_CALL(*mock, Close).WillOnce([]() { + return make_status_or(HttpResponse{HttpStatusCode::kOk, {}, {}}); + }); + } + return std::unique_ptr(std::move(mock)); + }; + + auto hedge_pool = std::make_shared( + /*max_threads=*/2, /*rate_limit=*/0.0, /*capacity=*/0.0, + /*max_concurrent=*/1); + + HedgedObjectReadSource source(MakeUnlimitedReadPool(), hedge_pool, factory, + std::chrono::milliseconds(1), + /*max_hedges=*/1, kUnlimitedBuffer); + + std::vector buffer(100); + std::thread unblocker([unblock_primary] { + std::this_thread::sleep_for(std::chrono::milliseconds(30)); + unblock_primary->set_value(); + }); + + StatusOr result = source.Read(buffer.data(), buffer.size()); + unblocker.join(); + + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(std::string(buffer.data(), result->bytes_received), + Eq("primary")); + EXPECT_THAT(calls->load(), Eq(2)); + + // Verify that the hedge concurrency slot was released and not leaked. + EXPECT_TRUE(hedge_pool->TryAcquireHedgeToken()); + hedge_pool->ReleaseHedgeSlot(); +} + } // namespace } // namespace internal GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END