feat(ACv2): support to handle large ranges

Author: bajajneha27

google/cloud/storage/async/object_descriptor.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/storage/async/object_descriptor.h"
+#include "google/cloud/storage/async/options.h"
 #include "google/cloud/internal/make_status.h"
 
 namespace google {
@@ -26,7 +27,13 @@ absl::optional<google::storage::v2::Object> ObjectDescriptor::metadata() const {
 
 std::pair<AsyncReader, AsyncToken> ObjectDescriptor::Read(std::int64_t offset,
                                                           std::int64_t limit) {
-  auto reader = impl_->Read({offset, limit});
+  auto max_range =
+      impl_->options().get<storage_experimental::MaximumRangeSizeOption>();
+  std::unique_ptr<storage_experimental::AsyncReaderConnection> reader;
+  if (limit > max_range) {
+    impl_->MakeSubsequentStream();
+  }
+  reader = impl_->Read({offset, limit});
   auto token = storage_internal::MakeAsyncToken(reader.get());
   return {AsyncReader(std::move(reader)), std::move(token)};
 }

google/cloud/storage/async/object_descriptor_connection.h

@@ -15,6 +15,7 @@
 #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_ASYNC_OBJECT_DESCRIPTOR_CONNECTION_H
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_ASYNC_OBJECT_DESCRIPTOR_CONNECTION_H
 
+#include "google/cloud/storage/async/reader.h"
 #include "google/cloud/storage/async/reader_connection.h"
 #include "google/cloud/options.h"
 #include "google/cloud/version.h"
@@ -59,6 +60,8 @@ class ObjectDescriptorConnection {
    * Starts a new range read in the current descriptor.
    */
   virtual std::unique_ptr<AsyncReaderConnection> Read(ReadParams p) = 0;
+
+  virtual void MakeSubsequentStream() = 0;
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/storage/internal/async/object_descriptor_connection_tracing.cc

@@ -13,6 +13,8 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/async/object_descriptor_connection_tracing.h"
+#include "google/cloud/storage/async/object_descriptor.h"
+#include "google/cloud/storage/async/reader.h"
 #include "google/cloud/storage/async/reader_connection.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/version.h"
@@ -61,6 +63,10 @@ class AsyncObjectDescriptorConnectionTracing
     return result;
   }
 
+  void MakeSubsequentStream() override {
+    return impl_->MakeSubsequentStream();
+  };
+
  private:
   opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span_;
   std::shared_ptr<storage_experimental::ObjectDescriptorConnection> impl_;

google/cloud/storage/internal/async/object_descriptor_impl.cc

@@ -37,24 +37,48 @@ ObjectDescriptorImpl::ObjectDescriptorImpl(
     : resume_policy_(std::move(resume_policy)),
       make_stream_(std::move(make_stream)),
       read_object_spec_(std::move(read_object_spec)),
-      stream_(std::move(stream)),
-      options_(std::move(options)) {}
+      options_(std::move(options)),
+      active_stream_(0),
+      streams_{std::move(stream)} {}
 
-ObjectDescriptorImpl::~ObjectDescriptorImpl() { stream_->Cancel(); }
+ObjectDescriptorImpl::~ObjectDescriptorImpl() {
+  for (auto stream : streams_) {
+    stream->Cancel();
+  }
+}
 
 void ObjectDescriptorImpl::Start(
     google::storage::v2::BidiReadObjectResponse first_response) {
   OnRead(std::move(first_response));
 }
 
-void ObjectDescriptorImpl::Cancel() { stream_->Cancel(); }
+void ObjectDescriptorImpl::Cancel() {
+  for (auto stream : streams_) {
+    stream->Cancel();
+  }
+}
 
 absl::optional<google::storage::v2::Object> ObjectDescriptorImpl::metadata()
     const {
   std::unique_lock<std::mutex> lk(mu_);
   return metadata_;
 }
 
+void ObjectDescriptorImpl::MakeSubsequentStream() {
+  auto request = google::storage::v2::BidiReadObjectRequest{};
+
+  *request.mutable_read_object_spec() = read_object_spec_;
+  auto stream_result = make_stream_(std::move(request)).get();
+  auto stream = std::move(stream_result->stream);
+
+  std::unique_lock<std::mutex> lk(mu_);
+  active_stream_ = streams_.size();
+  streams_.push_back(std::move(stream));
+  lk.unlock();
+
+  Start(std::move(stream_result->first_response));
+}
+
 std::unique_ptr<storage_experimental::AsyncReaderConnection>
 ObjectDescriptorImpl::Read(ReadParams p) {
   std::shared_ptr<storage::internal::HashFunction> hash_function =
@@ -200,7 +224,7 @@ void ObjectDescriptorImpl::Resume(google::rpc::Status const& proto_status) {
 void ObjectDescriptorImpl::OnResume(StatusOr<OpenStreamResult> result) {
   if (!result) return OnFinish(std::move(result).status());
   std::unique_lock<std::mutex> lk(mu_);
-  stream_ = std::move(result->stream);
+  streams_[0] = std::move(result->stream);
   // TODO(#15105) - this should be done without release the lock.
   Flush(std::move(lk));
   OnRead(std::move(result->first_response));

google/cloud/storage/internal/async/object_descriptor_impl.h

@@ -62,6 +62,8 @@ class ObjectDescriptorImpl
   std::unique_ptr<storage_experimental::AsyncReaderConnection> Read(
       ReadParams p) override;
 
+  void MakeSubsequentStream() override;
+
  private:
   std::weak_ptr<ObjectDescriptorImpl> WeakFromThis() {
     return shared_from_this();
@@ -77,7 +79,9 @@ class ObjectDescriptorImpl
     return CopyActiveRanges(std::unique_lock<std::mutex>(mu_));
   }
 
-  auto CurrentStream(std::unique_lock<std::mutex>) const { return stream_; }
+  auto CurrentStream(std::unique_lock<std::mutex>) const {
+    return streams_[active_stream_];
+  }
 
   void Flush(std::unique_lock<std::mutex> lk);
   void OnWrite(bool ok);
@@ -105,6 +109,8 @@ class ObjectDescriptorImpl
 
   std::unordered_map<std::int64_t, std::shared_ptr<ReadRange>> active_ranges_;
   Options options_;
+  std::vector<std::shared_ptr<OpenStream>> streams_ = {};
+  int active_stream_;
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/storage/internal/async/object_descriptor_reader_tracing.cc

@@ -41,7 +41,7 @@ class ObjectDescriptorReaderTracing : public ObjectDescriptorReader {
   ~ObjectDescriptorReaderTracing() override = default;
 
   future<ObjectDescriptorReader::ReadResponse> Read() override {
-    auto span = internal::MakeSpan("storage::AsyncConnection::ReadObjectRange");
+    auto span = internal::MakeSpan("storage::AsyncConnection::ReadRange");
     internal::OTelScope scope(span);
     return ObjectDescriptorReader::Read().then(
         [span = std::move(span),

google/cloud/storage/mocks/mock_async_object_descriptor_connection.h

@@ -31,6 +31,7 @@ class MockAsyncObjectDescriptorConnection
               (const, override));
   MOCK_METHOD(std::unique_ptr<storage_experimental::AsyncReaderConnection>,
               Read, (ReadParams), (override));
+  MOCK_METHOD(void, MakeSubsequentStream, (), (override));
   MOCK_METHOD(Options, options, (), (const, override));
 };