diff --git a/google/cloud/storage/async/client.cc b/google/cloud/storage/async/client.cc index 9f9baf89da6ca..4999e3527546b 100644 --- a/google/cloud/storage/async/client.cc +++ b/google/cloud/storage/async/client.cc @@ -61,8 +61,10 @@ future> AsyncClient::Open( future> AsyncClient::Open( google::storage::v2::BidiReadObjectSpec spec, Options opts) { return connection_ - ->Open({std::move(spec), google::cloud::internal::MergeOptions( - std::move(opts), connection_->options())}) + ->Open({std::move(spec), + google::cloud::internal::MergeOptions(std::move(opts), + connection_->options()), + /*initial_read_range=*/absl::nullopt}) .then([](auto f) -> StatusOr { auto connection = f.get(); if (!connection) return std::move(connection).status(); diff --git a/google/cloud/storage/async/connection.h b/google/cloud/storage/async/connection.h index 8e9df06d9f9e9..6017de35d09cd 100644 --- a/google/cloud/storage/async/connection.h +++ b/google/cloud/storage/async/connection.h @@ -76,6 +76,15 @@ class AsyncConnection { virtual future> InsertObject( InsertObjectParams p) = 0; + /// A parameter to specify the initial offset and length of an eagerly-started + /// read. + struct InitialReadRange { + /// The starting offset of the read. + std::int64_t offset; + /// The number of bytes to read. + std::int64_t length; + }; + /** * A thin wrapper around the `Open()` parameters. * @@ -85,6 +94,7 @@ class AsyncConnection { struct OpenParams { google::storage::v2::BidiReadObjectSpec read_spec; Options options; + absl::optional initial_read_range; }; /// Open an object to perform multiple reads. diff --git a/google/cloud/storage/internal/async/connection_impl.cc b/google/cloud/storage/internal/async/connection_impl.cc index 976fea52f2337..6f4ffcb7c3fa3 100644 --- a/google/cloud/storage/internal/async/connection_impl.cc +++ b/google/cloud/storage/internal/async/connection_impl.cc @@ -192,6 +192,12 @@ future>> AsyncConnectionImpl::Open(OpenParams p) { auto initial_request = google::storage::v2::BidiReadObjectRequest{}; *initial_request.mutable_read_object_spec() = p.read_spec; + if (p.initial_read_range) { + auto* range = initial_request.add_read_ranges(); + range->set_read_offset(p.initial_read_range->offset); + range->set_read_length(p.initial_read_range->length); + range->set_read_id(1); + } auto current = internal::MakeImmutableOptions(p.options); // Get the policy factory and immediately create a policy. auto resume_policy = current->get()(); @@ -234,8 +240,8 @@ AsyncConnectionImpl::Open(OpenParams p) { using ReturnType = std::shared_ptr; return pending.then([rp = std::move(resume_policy), fa = std::move(factory), rs = std::move(p.read_spec), - options = std::move(p.options), refresh = refresh_]( - auto f) mutable -> StatusOr { + options = std::move(p.options), initial_read_range = p.initial_read_range, + refresh = refresh_](auto f) mutable -> StatusOr { auto result = f.get(); if (!result) return std::move(result).status(); @@ -255,7 +261,7 @@ AsyncConnectionImpl::Open(OpenParams p) { }; auto impl = std::make_shared( std::move(rp), std::move(fa), std::move(rs), std::move(result->stream), - std::move(options), std::move(transport_ok)); + initial_read_range, std::move(options), std::move(transport_ok)); impl->Start(std::move(result->first_response)); return ReturnType(impl); }); diff --git a/google/cloud/storage/internal/async/connection_impl_open_test.cc b/google/cloud/storage/internal/async/connection_impl_open_test.cc index 941df81f692de..837e7af8e96cb 100644 --- a/google/cloud/storage/internal/async/connection_impl_open_test.cc +++ b/google/cloud/storage/internal/async/connection_impl_open_test.cc @@ -214,7 +214,8 @@ TEST(AsyncConnectionImplTest, OpenSimple) { auto request = google::storage::v2::BidiReadObjectSpec{}; ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &request)); - auto pending = connection->Open({std::move(request), connection->options()}); + auto pending = connection->Open( + {std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt}); auto next = sequencer.PopFrontWithName(); EXPECT_EQ(next.second, "Start"); @@ -310,7 +311,8 @@ TEST(AsyncConnectionImplTest, HandleRedirectErrors) { auto request = google::storage::v2::BidiReadObjectSpec{}; ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest0, &request)); - auto pending = connection->Open({std::move(request), connection->options()}); + auto pending = connection->Open( + {std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt}); for (int i = 0; i != kRetryAttempts + 1; ++i) { auto next = sequencer.PopFrontWithName(); @@ -354,7 +356,8 @@ TEST(AsyncConnectionImplTest, StopOnPermanentError) { auto request = google::storage::v2::BidiReadObjectSpec{}; ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &request)); - auto pending = connection->Open({std::move(request), connection->options()}); + auto pending = connection->Open( + {std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt}); auto next = sequencer.PopFrontWithName(); EXPECT_EQ(next.second, "Start"); @@ -390,7 +393,8 @@ TEST(AsyncConnectionImplTest, TooManyTransienErrors) { auto request = google::storage::v2::BidiReadObjectSpec{}; ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &request)); - auto pending = connection->Open({std::move(request), connection->options()}); + auto pending = connection->Open( + {std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt}); for (int i = 0; i != kRetryAttempts + 1; ++i) { auto next = sequencer.PopFrontWithName(); diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.cc b/google/cloud/storage/internal/async/object_descriptor_impl.cc index 4f145986521ad..9545f98acffd5 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl.cc +++ b/google/cloud/storage/internal/async/object_descriptor_impl.cc @@ -39,11 +39,16 @@ ObjectDescriptorImpl::ObjectDescriptorImpl( std::unique_ptr resume_policy, OpenStreamFactory make_stream, google::storage::v2::BidiReadObjectSpec read_object_spec, - std::shared_ptr stream, Options options, + std::shared_ptr stream, + absl::optional initial_read_range, + Options options, std::function transport_ok) : resume_policy_prototype_(std::move(resume_policy)), make_stream_(std::move(make_stream)), read_object_spec_(std::move(read_object_spec)), + initial_read_range_(std::move(initial_read_range)), + initial_cache_consumed_(!initial_read_range.has_value()), + read_id_generator_(initial_read_range.has_value() ? 1 : 0), options_(std::move(options)), transport_ok_(std::move(transport_ok)) { stream_manager_ = std::make_unique( @@ -203,12 +208,26 @@ std::unique_ptr ObjectDescriptorImpl::Read( } auto it = stream_manager_->GetLeastBusyStream(); - auto const id = ++read_id_generator_; + bool is_cache_hit = false; + if (!initial_cache_consumed_ && initial_read_range_.has_value()) { + auto const& initial = *initial_read_range_; + // Ensure both offset and length match. + if (p.start == initial.offset && limit.value_or(0) == initial.length) { + is_cache_hit = true; + } + initial_cache_consumed_ = true; + } + + auto const id = is_cache_hit ? 1 : ++read_id_generator_; it->active_ranges.emplace(id, range); - auto& read_range = *it->stream->next_request.add_read_ranges(); - read_range.set_read_id(id); - read_range.set_read_offset(p.start); - read_range.set_read_length(p.length); + + if (!is_cache_hit) { + auto& read_range = *it->stream->next_request.add_read_ranges(); + read_range.set_read_id(id); + read_range.set_read_offset(p.start); + read_range.set_read_length(p.length); + } + Flush(std::move(lk), it); if (!internal::TracingEnabled(options_)) { diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.h b/google/cloud/storage/internal/async/object_descriptor_impl.h index 7f5f654951f3b..7ac8228927b4b 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl.h +++ b/google/cloud/storage/internal/async/object_descriptor_impl.h @@ -15,6 +15,7 @@ #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OBJECT_DESCRIPTOR_IMPL_H #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OBJECT_DESCRIPTOR_IMPL_H +#include "google/cloud/storage/async/connection.h" #include "google/cloud/storage/async/object_descriptor_connection.h" #include "google/cloud/storage/async/resume_policy.h" #include "google/cloud/storage/internal/async/multi_stream_manager.h" @@ -57,11 +58,14 @@ class ObjectDescriptorImpl : public storage::ObjectDescriptorConnection, public std::enable_shared_from_this { public: - ObjectDescriptorImpl(std::unique_ptr resume_policy, - OpenStreamFactory make_stream, - google::storage::v2::BidiReadObjectSpec read_object_spec, - std::shared_ptr stream, Options options = {}, - std::function transport_ok = {}); + ObjectDescriptorImpl( + std::unique_ptr resume_policy, + OpenStreamFactory make_stream, + google::storage::v2::BidiReadObjectSpec read_object_spec, + std::shared_ptr stream, + absl::optional + initial_read_range, + Options options = {}, std::function transport_ok = {}); ~ObjectDescriptorImpl() override; // Start the read loop. @@ -122,6 +126,9 @@ class ObjectDescriptorImpl mutable std::mutex mu_; google::storage::v2::BidiReadObjectSpec read_object_spec_; absl::optional metadata_; + absl::optional + initial_read_range_; + bool initial_cache_consumed_ = false; std::int64_t read_id_generator_ = 0; Options options_; diff --git a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc index 5423423b4a08d..c4d479fea1848 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc +++ b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc @@ -72,12 +72,14 @@ auto MakeTested( OpenStreamFactory make_stream, google::storage::v2::BidiReadObjectSpec read_object_spec, std::shared_ptr stream, + absl::optional initial_read_range = absl::nullopt, std::function transport_ok = [] { return true; }) { Options options; options.set(true); return std::make_shared( std::move(resume_policy), std::move(make_stream), - std::move(read_object_spec), std::move(stream), std::move(options), + std::move(read_object_spec), std::move(stream), + std::move(initial_read_range), std::move(options), std::move(transport_ok)); } @@ -286,6 +288,188 @@ TEST(ObjectDescriptorImpl, ReadSingleRange) { next.first.set_value(true); } +/// @test Verify that reading a pre-warmed cache range skips making a request. +TEST(ObjectDescriptorImpl, ReadCacheHit) { + auto constexpr kResponse0 = R"pb( + metadata { + bucket: "projects/_/buckets/test-bucket" + name: "test-object" + generation: 42 + } + read_handle { handle: "handle-12345" } + )pb"; + + auto constexpr kResponse1 = R"pb( + read_handle { handle: "handle-23456" } + object_data_ranges { + range_end: true + read_range { read_id: 1 read_offset: 20000 } + checksummed_data { + content: "The quick brown fox jumps over the lazy dog" + } + } + )pb"; + + AsyncSequencer sequencer; + auto stream = std::make_unique(); + // Cache hit means no new request. + EXPECT_CALL(*stream, Write).Times(0); + + EXPECT_CALL(*stream, Read) + .WillOnce([=, &sequencer]() { + return sequencer.PushBack("Read[1]").then([&](auto) { + auto response = Response{}; + EXPECT_TRUE(TextFormat::ParseFromString(kResponse1, &response)); + return absl::make_optional(response); + }); + }) + .WillOnce([&sequencer]() { + return sequencer.PushBack("Read[2]").then( + [](auto) { return absl::optional{}; }); + }); + EXPECT_CALL(*stream, Finish).WillOnce([&sequencer]() { + return sequencer.PushBack("Finish").then( + [](auto) { return PermanentError(); }); + }); + EXPECT_CALL(*stream, Cancel).Times(AtMost(1)); + + MockFactory factory; + EXPECT_CALL(factory, Call).WillOnce([](Request const&) { + return make_ready_future(StatusOr(PermanentError())); + }); + + auto tested = MakeTested(NoResume(), factory.AsStdFunction(), + google::storage::v2::BidiReadObjectSpec{}, + std::make_shared(std::move(stream)), + storage::AsyncConnection::InitialReadRange{20000, 100}); + auto response = Response{}; + EXPECT_TRUE(TextFormat::ParseFromString(kResponse0, &response)); + tested->Start(std::move(response)); + + auto read1 = sequencer.PopFrontWithName(); + EXPECT_EQ(read1.second, "Read[1]"); + + // Call ReadParams exactly matching the cache. + auto s1 = tested->Read({20000, 100}); + auto s1r1 = s1->Read(); + + read1.first.set_value(true); + + EXPECT_THAT(s1r1.get(), + VariantWith(ResultOf( + "contents are", + [](storage::ReadPayload const& p) { return p.contents(); }, + ElementsAre(absl::string_view{ + "The quick brown fox jumps over the lazy dog"})))); + + EXPECT_THAT(s1->Read().get(), VariantWith(IsOk())); + + auto next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read[2]"); + next.first.set_value(true); + + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); +} + +/// @test Verify that reading a cache miss spawns a request with ID 2. +TEST(ObjectDescriptorImpl, ReadCacheMiss) { + auto constexpr kResponse0 = R"pb( + metadata { + bucket: "projects/_/buckets/test-bucket" + name: "test-object" + generation: 42 + } + read_handle { handle: "handle-12345" } + )pb"; + + auto constexpr kRequest1 = R"pb( + read_ranges { read_id: 2 read_offset: 35000 read_length: 100 } + )pb"; + auto constexpr kResponse1 = R"pb( + read_handle { handle: "handle-23456" } + object_data_ranges { + range_end: true + read_range { read_id: 2 read_offset: 35000 } + checksummed_data { + content: "Data for ID 2" + } + } + )pb"; + + AsyncSequencer sequencer; + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Write) + .WillOnce([&](Request const& request, grpc::WriteOptions) { + auto expected = Request{}; + EXPECT_TRUE(TextFormat::ParseFromString(kRequest1, &expected)); + EXPECT_THAT(request, IsProtoEqual(expected)); + return sequencer.PushBack("Write[1]").then([](auto f) { + return f.get(); + }); + }); + + EXPECT_CALL(*stream, Read) + .WillOnce([=, &sequencer]() { + return sequencer.PushBack("Read[1]").then([&](auto) { + auto response = Response{}; + EXPECT_TRUE(TextFormat::ParseFromString(kResponse1, &response)); + return absl::make_optional(response); + }); + }) + .WillOnce([&sequencer]() { + return sequencer.PushBack("Read[2]").then( + [](auto) { return absl::optional{}; }); + }); + EXPECT_CALL(*stream, Finish).WillOnce([&sequencer]() { + return sequencer.PushBack("Finish").then( + [](auto) { return PermanentError(); }); + }); + EXPECT_CALL(*stream, Cancel).Times(AtMost(1)); + + MockFactory factory; + EXPECT_CALL(factory, Call).WillOnce([](Request const&) { + return make_ready_future(StatusOr(PermanentError())); + }); + + auto tested = MakeTested(NoResume(), factory.AsStdFunction(), + google::storage::v2::BidiReadObjectSpec{}, + std::make_shared(std::move(stream)), + storage::AsyncConnection::InitialReadRange{20000, 100}); + auto response = Response{}; + EXPECT_TRUE(TextFormat::ParseFromString(kResponse0, &response)); + tested->Start(std::move(response)); + + auto read1 = sequencer.PopFrontWithName(); + EXPECT_EQ(read1.second, "Read[1]"); + + // Call ReadParams that misses the cache. (Requests offset 35000 instead of 20000). + auto s1 = tested->Read({35000, 100}); + auto s1r1 = s1->Read(); + + auto next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Write[1]"); + next.first.set_value(true); + read1.first.set_value(true); + + EXPECT_THAT(s1r1.get(), + VariantWith(ResultOf( + "contents are", + [](storage::ReadPayload const& p) { return p.contents(); }, + ElementsAre(absl::string_view{"Data for ID 2"})))); + + EXPECT_THAT(s1->Read().get(), VariantWith(IsOk())); + + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read[2]"); + next.first.set_value(true); + + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); +} + /// @test Reading multiple ranges creates a single request. TEST(ObjectDescriptorImpl, ReadMultipleRanges) { auto constexpr kLength = 100; @@ -1780,7 +1964,7 @@ TEST(ObjectDescriptorImpl, MultiStreamOptimizationDisabled) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); tested->MakeSubsequentStream(); @@ -1825,6 +2009,7 @@ TEST(ObjectDescriptorImpl, IsOpenFalseOnPermanentError) { auto tested = MakeTested(NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, std::make_shared(std::move(stream)), + /*initial_read_range=*/absl::nullopt, transport_ok_callback); EXPECT_TRUE(tested->IsOpen()); @@ -1843,7 +2028,7 @@ TEST(ObjectDescriptorImpl, IsOpenFalseOnTransportFailure) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), Options{}, + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, Options{}, std::move(transport_ok)); EXPECT_FALSE(tested->IsOpen()); } @@ -1859,7 +2044,7 @@ TEST(ObjectDescriptorImpl, IsOpenTrueOnTransportSuccess) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), Options{}, + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, Options{}, std::move(transport_ok)); EXPECT_TRUE(tested->IsOpen()); } @@ -1936,7 +2121,7 @@ TEST(ObjectDescriptorImpl, FullReadChecksumValidationSuccess) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponseMetadata, &response)); @@ -2043,7 +2228,7 @@ TEST(ObjectDescriptorImpl, FullReadChecksumValidationMismatchCrc32c) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponseMetadata, &response)); @@ -2151,7 +2336,7 @@ TEST(ObjectDescriptorImpl, FullReadChecksumValidationMismatchMd5) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponseMetadata, &response)); @@ -2259,7 +2444,7 @@ TEST(ObjectDescriptorImpl, PartialReadChecksumValidationBypassed) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponseMetadata, &response));