diff --git a/google/cloud/storage/async/object_responses.h b/google/cloud/storage/async/object_responses.h index 5a5970d1dc39c..7d05659a7c7f3 100644 --- a/google/cloud/storage/async/object_responses.h +++ b/google/cloud/storage/async/object_responses.h @@ -56,6 +56,23 @@ class ReadPayload { return {impl_.chunk_begin(), impl_.chunk_end()}; } + ///@{ + /** + * The payload contents as an `absl::Cord`. + * + * Copying an `absl::Cord` is relatively cheap, as the underlying buffers are + * reference counted and shared between copies. Applications that need to + * retain the payload data beyond the lifetime of this object can copy (or + * move) the returned value instead of copying the data in `contents()`. + * + * Retaining a copy of this value keeps the underlying buffers alive, but + * it does not extend the lifetime of any other resources (such as the RPC + * that received the data). + */ + absl::Cord const& cord_contents() const& { return impl_; } + absl::Cord cord_contents() && { return std::move(impl_); } + ///@} + /// The object metadata. std::optional metadata() const { return metadata_; diff --git a/google/cloud/storage/internal/async/read_payload_impl_test.cc b/google/cloud/storage/internal/async/read_payload_impl_test.cc index 197f7a062b5b5..a67c719556b32 100644 --- a/google/cloud/storage/internal/async/read_payload_impl_test.cc +++ b/google/cloud/storage/internal/async/read_payload_impl_test.cc @@ -14,9 +14,15 @@ #include "google/cloud/storage/internal/async/read_payload_impl.h" #include "google/cloud/testing_util/is_proto_equal.h" +#include "absl/strings/cord.h" #include "absl/strings/string_view.h" #include "google/storage/v2/storage.pb.h" #include +#include +#include +#include +#include +#include namespace google { namespace cloud { @@ -27,6 +33,7 @@ namespace { using ::google::cloud::testing_util::IsProtoEqual; using ::testing::AllOf; using ::testing::ElementsAre; +using ::testing::Eq; using ::testing::Field; using ::testing::IsEmpty; using ::testing::Optional; @@ -175,6 +182,52 @@ TEST(ReadPayload, AccumulateEmptyIntoEmpty) { EXPECT_FALSE(actual.metadata().has_value()); } +TEST(ReadPayload, CordContents) { + storage::ReadPayload const actual = ReadPayloadImpl::Make(absl::Cord(kQuick)); + EXPECT_THAT(actual.cord_contents(), Eq(std::string_view(kQuick))); + + // Copying the Cord shares the underlying buffer, the data is not copied. + std::optional const original = + actual.cord_contents().TryFlat(); + ASSERT_TRUE(original.has_value()); + // NOLINTNEXTLINE(performance-unnecessary-copy-initialization) + absl::Cord const copy = actual.cord_contents(); + EXPECT_THAT(copy, Eq(std::string_view(kQuick))); + std::optional const copied = copy.TryFlat(); + ASSERT_TRUE(copied.has_value()); + EXPECT_THAT(copied->data(), Eq(original->data())); +} + +TEST(ReadPayload, CordContentsFromVector) { + storage::ReadPayload const actual = storage::ReadPayload( + std::vector({std::string(kQuick), std::string(kQuick)})); + EXPECT_THAT(std::string(actual.cord_contents()), + Eq(std::string(kQuick) + kQuick)); + EXPECT_THAT(actual.cord_contents().size(), Eq(actual.size())); +} + +TEST(ReadPayload, CordContentsMove) { + bool released = false; + storage::ReadPayload payload = ReadPayloadImpl::Make( + absl::MakeCordFromExternal(kQuick, [&released] { released = true; })); + std::optional const original = + payload.cord_contents().TryFlat(); + ASSERT_TRUE(original.has_value()); + absl::Cord moved = std::move(payload).cord_contents(); + EXPECT_THAT(moved, Eq(std::string_view(kQuick))); + std::optional const flat = moved.TryFlat(); + ASSERT_TRUE(flat.has_value()); + EXPECT_THAT(flat->data(), Eq(original->data())); + EXPECT_FALSE(released); + moved.Clear(); + EXPECT_TRUE(released); +} + +TEST(ReadPayload, CordContentsEmpty) { + storage::ReadPayload const actual; + EXPECT_THAT(actual.cord_contents(), IsEmpty()); +} + } // namespace GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace storage_internal diff --git a/google/cloud/storage/tests/async_client_integration_test.cc b/google/cloud/storage/tests/async_client_integration_test.cc index 45c6c4c4bda9a..24dcea55f90ad 100644 --- a/google/cloud/storage/tests/async_client_integration_test.cc +++ b/google/cloud/storage/tests/async_client_integration_test.cc @@ -25,11 +25,14 @@ #include "google/cloud/internal/getenv.h" #include "google/cloud/testing_util/is_proto_equal.h" #include "google/cloud/testing_util/status_matchers.h" +#include "absl/strings/cord.h" +#include "absl/strings/str_join.h" #include #include #include #include #include +#include #include #include #include @@ -45,12 +48,29 @@ using ::google::cloud::internal::GetEnv; using ::google::cloud::testing_util::IsOk; using ::google::cloud::testing_util::IsProtoEqual; using ::google::cloud::testing_util::StatusIs; +using ::testing::Eq; +using ::testing::Gt; using ::testing::IsEmpty; using ::testing::Le; using ::testing::Not; using ::testing::Optional; +using ::testing::Pointwise; +using ::testing::SizeIs; using ::testing::VariantWith; +// Matches a (string_view, string_view) tuple if both refer to the same buffer, +// i.e., they have the same data pointer and size. +MATCHER(SameBuffer, "refers to the same buffer") { + std::string_view const lhs = std::get<0>(arg); + std::string_view const rhs = std::get<1>(arg); + if (lhs.data() == rhs.data() && lhs.size() == rhs.size()) return true; + *result_listener << "lhs={data=" << static_cast(lhs.data()) + << ", size=" << lhs.size() + << "}, rhs={data=" << static_cast(rhs.data()) + << ", size=" << rhs.size() << "}"; + return false; +} + class AsyncClientIntegrationTest : public google::cloud::storage::testing::StorageIntegrationTest { protected: @@ -72,6 +92,37 @@ class AsyncClientIntegrationTest .set_generation(object.generation())); } + /// Returns ~5MiB of random data, enough to require multiple `Read()` calls. + std::vector MakeCordTestData() { + std::size_t constexpr kBlockSize = 1024 * 1024; + std::size_t constexpr kBlockCount = 5; + std::vector blocks(kBlockCount); + std::generate(blocks.begin(), blocks.end(), + [&] { return MakeRandomData(kBlockSize); }); + return blocks; + } + + /// Creates an appendable object, supported in zonal buckets and the emulator. + StatusOr CreateAppendableObject( + AsyncClient& async, std::string const& object_name, + std::vector const& blocks) { + StatusOr> w = + async + .StartAppendableObjectUpload(BucketName(bucket_name()), object_name) + .get(); + if (!w) return std::move(w).status(); + AsyncWriter writer; + AsyncToken token; + std::tie(writer, token) = *std::move(w); + for (std::string const& block : blocks) { + StatusOr p = + writer.Write(std::move(token), WritePayload(block)).get(); + if (!p) return std::move(p).status(); + token = *std::move(p); + } + return writer.Finalize(std::move(token)).get(); + } + private: std::string bucket_name_; }; @@ -90,6 +141,46 @@ auto AlwaysRetry() { MakeAlwaysRetryAsyncIdempotencyPolicy); } +// Reads all the data from @p reader. For each payload, verifies that +// `cord_contents()` refers to the same buffers as `contents()`, and that +// copying the Cord does not copy the data. Retains each Cord in @p retained. +void ReadAndRetainCords(AsyncReader reader, AsyncToken token, + std::vector& retained) { + while (token.valid()) { + StatusOr> p = + reader.Read(std::move(token)).get(); + ASSERT_STATUS_OK(p); + ReadPayload payload; + std::tie(payload, token) = *std::move(p); + + absl::Cord const& cord = payload.cord_contents(); + EXPECT_EQ(cord.size(), payload.size()); + std::vector const chunks(cord.chunk_begin(), + cord.chunk_end()); + EXPECT_THAT(chunks, Pointwise(SameBuffer(), payload.contents())); + + // NOLINTNEXTLINE(performance-unnecessary-copy-initialization) + absl::Cord const copy = payload.cord_contents(); + std::vector const copy_chunks(copy.chunk_begin(), + copy.chunk_end()); + EXPECT_THAT(copy_chunks, Pointwise(SameBuffer(), chunks)); + + if (payload.size() == 0) continue; + retained.push_back(std::move(payload).cord_contents()); + } +} + +// Verifies the Cords retained by `ReadAndRetainCords()` contain @p expected. +// Called after the payloads, reader, and client have been destroyed. +void VerifyRetainedCords(std::vector retained, + std::string const& expected) { + EXPECT_THAT(retained, SizeIs(Gt(1))); + absl::Cord actual; + for (absl::Cord& c : retained) actual.Append(std::move(c)); + EXPECT_EQ(actual.size(), expected.size()); + EXPECT_THAT(actual, Eq(expected)); +} + TEST_F(AsyncClientIntegrationTest, ObjectCRUD) { auto async = AsyncClient(TestOptions()); auto object_name = MakeRandomObjectName(); @@ -295,6 +386,72 @@ TEST_F(AsyncClientIntegrationTest, StreamingReadRange) { absl::string_view(contents).substr(kReadOffset)); } +TEST_F(AsyncClientIntegrationTest, StreamingReadCordContents) { + auto async = AsyncClient(TestOptions()); + std::string const object_name = MakeRandomObjectName(); + std::vector const blocks = MakeCordTestData(); + std::string const expected = absl::StrJoin(blocks, ""); + StatusOr metadata = + async + .InsertObject(BucketName(bucket_name()), object_name, blocks, + AlwaysRetry()) + .get(); + ASSERT_STATUS_OK(metadata); + ScheduleForDelete(*metadata); + + std::vector retained; + { + // Use a separate client, so we can verify the retained Cords outlive it. + auto reader_client = AsyncClient(TestOptions()); + StatusOr> r = + reader_client.ReadObject(BucketName(bucket_name()), object_name).get(); + ASSERT_STATUS_OK(r); + AsyncReader reader; + AsyncToken token; + std::tie(reader, token) = *std::move(r); + ReadAndRetainCords(std::move(reader), std::move(token), retained); + if (HasFatalFailure()) return; + } + VerifyRetainedCords(std::move(retained), expected); +} + +TEST_F(AsyncClientIntegrationTest, OpenReadCordContents) { + if (!UsingEmulator()) GTEST_SKIP(); + auto async = AsyncClient(TestOptions()); + auto client = MakeIntegrationTestClient(true, TestOptions()); + std::string const object_name = MakeRandomObjectName(); + + StatusOr create = client.CreateBucket( + bucket_name(), storage::BucketMetadata{}.set_location("us-west4")); + if (!create && create.status().code() != StatusCode::kAlreadyExists) { + GTEST_FAIL() << "cannot create bucket: " << create.status(); + } + + std::vector const blocks = MakeCordTestData(); + std::string const expected = absl::StrJoin(blocks, ""); + StatusOr metadata = + CreateAppendableObject(async, object_name, blocks); + ASSERT_STATUS_OK(metadata); + ScheduleForDelete(*metadata); + + std::vector retained; + { + // Use a separate client, so we can verify the retained Cords outlive it. + auto reader_client = AsyncClient(TestOptions()); + auto spec = google::storage::v2::BidiReadObjectSpec{}; + spec.set_bucket(BucketName(bucket_name()).FullName()); + spec.set_object(object_name); + StatusOr descriptor = reader_client.Open(spec).get(); + ASSERT_STATUS_OK(descriptor); + AsyncReader reader; + AsyncToken token; + std::tie(reader, token) = descriptor->Read(0, expected.size()); + ReadAndRetainCords(std::move(reader), std::move(token), retained); + if (HasFatalFailure()) return; + } + VerifyRetainedCords(std::move(retained), expected); +} + TEST_F(AsyncClientIntegrationTest, StartUnbufferedUploadEmpty) { auto client = AsyncClient(TestOptions()); auto object_name = MakeRandomObjectName();