active ranges for each stream

Author: bajajneha27

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

@@ -38,8 +38,10 @@ ObjectDescriptorImpl::ObjectDescriptorImpl(
       make_stream_(std::move(make_stream)),
       read_object_spec_(std::move(read_object_spec)),
       options_(std::move(options)),
-      active_stream_(0),
-      streams_{std::move(stream)} {}
+      streams_{std::move(stream)} {
+        std::unordered_map<std::int64_t, std::shared_ptr<ReadRange>> initial_ranges;
+        active_ranges_.push_back(std::move(initial_ranges));
+      }
 
 ObjectDescriptorImpl::~ObjectDescriptorImpl() {
   for (auto const& stream : streams_) {
@@ -74,9 +76,10 @@ void ObjectDescriptorImpl::MakeSubsequentStream() {
   std::unique_lock<std::mutex> lk(mu_);
   active_stream_ = streams_.size();
   streams_.push_back(std::move(stream));
+  std::unordered_map<std::int64_t, std::shared_ptr<ReadRange>> active_ranges;
+  active_ranges_.push_back(std::move(active_ranges));
   lk.unlock();
-
-  Start(std::move(stream_result->first_response));
+  OnRead(std::move(stream_result->first_response));
 }
 
 std::unique_ptr<storage_experimental::AsyncReaderConnection>
@@ -93,7 +96,7 @@ ObjectDescriptorImpl::Read(ReadParams p) {
 
   std::unique_lock<std::mutex> lk(mu_);
   auto const id = ++read_id_generator_;
-  active_ranges_.emplace(id, range);
+  active_ranges_[active_stream_].emplace(id, range);
   auto& read_range = *next_request_.add_read_ranges();
   read_range.set_read_id(id);
   read_range.set_read_offset(p.start);
@@ -170,9 +173,10 @@ void ObjectDescriptorImpl::OnRead(
 
 void ObjectDescriptorImpl::CleanupDoneRanges(
     std::unique_lock<std::mutex> const&) {
-  for (auto i = active_ranges_.begin(); i != active_ranges_.end();) {
+  auto &active_ranges = active_ranges_[active_stream_];
+  for (auto i = active_ranges.begin(); i != active_ranges.end();) {
     if (i->second->IsDone()) {
-      i = active_ranges_.erase(i);
+      i = active_ranges.erase(i);
     } else {
       ++i;
     }
@@ -209,7 +213,7 @@ void ObjectDescriptorImpl::Resume(google::rpc::Status const& proto_status) {
   ApplyRedirectErrors(read_object_spec_, proto_status);
   auto request = google::storage::v2::BidiReadObjectRequest{};
   *request.mutable_read_object_spec() = read_object_spec_;
-  for (auto const& kv : active_ranges_) {
+  for (auto const& kv : active_ranges_[active_stream_]) {
     auto range = kv.second->RangeForResume(kv.first);
     if (!range) continue;
     *request.add_read_ranges() = *std::move(range);

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

@@ -72,7 +72,7 @@ class ObjectDescriptorImpl
   // This may seem expensive, but it is less bug-prone than iterating over
   // the map with the lock held.
   auto CopyActiveRanges(std::unique_lock<std::mutex> const&) const {
-    return active_ranges_;
+    return active_ranges_[active_stream_];
   }
 
   auto CopyActiveRanges() const {
@@ -106,10 +106,10 @@ class ObjectDescriptorImpl
   bool write_pending_ = false;
   google::storage::v2::BidiReadObjectRequest next_request_;
 
-  std::unordered_map<std::int64_t, std::shared_ptr<ReadRange>> active_ranges_;
+  std::vector<std::unordered_map<std::int64_t, std::shared_ptr<ReadRange>>> active_ranges_;
   Options options_;
-  std::int64_t active_stream_ = 0;
-  std::vector<std::shared_ptr<OpenStream>> streams_ = {};
+  std::size_t active_stream_ = 0;
+  std::vector<std::shared_ptr<OpenStream>> streams_;
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/storage/tests/async_client_integration_test.cc

@@ -918,13 +918,24 @@ TEST_F(AsyncClientIntegrationTest, Open) {
 
   auto constexpr kSize = 8 * 1024;
   auto constexpr kStride = 2 * kSize;
+  auto constexpr kBlockCount = 4;
+  auto const block = MakeRandomData(kSize);
 
-  auto os = client.WriteObject(bucket_name(), object_name);
-  for (char c : {'0', '1', '2', '3', '4'}) {
-    os << std::string(kStride, c);
+  auto w =
+      async.StartAppendableObjectUpload(BucketName(bucket_name()), object_name)
+          .get();
+  ASSERT_STATUS_OK(w);
+  AsyncWriter writer;
+  AsyncToken token;
+  std::tie(writer, token) = *std::move(w);
+  for (int i = 0; i != kBlockCount; ++i) {
+    auto p = writer.Write(std::move(token), WritePayload(block)).get();
+    ASSERT_STATUS_OK(p);
+    token = *std::move(p);
   }
-  os.Close();
-  ASSERT_STATUS_OK(os.metadata());
+
+  auto metadata = writer.Finalize(std::move(token)).get();
+  ASSERT_STATUS_OK(metadata);
 
   auto spec = google::storage::v2::BidiReadObjectSpec{};
   spec.set_bucket(BucketName(bucket_name()).FullName());
@@ -933,13 +944,7 @@ TEST_F(AsyncClientIntegrationTest, Open) {
   ASSERT_STATUS_OK(descriptor);
 
   AsyncReader r0;
-  AsyncReader r1;
-  AsyncReader r2;
   AsyncToken t0;
-  AsyncToken t1;
-  AsyncToken t2;
-  std::tie(r1, t1) = descriptor->Read(1 * kStride, kSize);
-  std::tie(r2, t2) = descriptor->Read(1 * kStride, kSize);
   auto actual0 = std::string{};
   std::tie(r0, t0) = descriptor->Read(0 * kStride, kSize);
   while (t0.valid()) {
@@ -952,9 +957,9 @@ TEST_F(AsyncClientIntegrationTest, Open) {
     t0 = std::move(t);
   }
 
-  EXPECT_EQ(actual0, std::string(kSize, '0'));
+  EXPECT_EQ(actual0.size(), kSize);
   client.DeleteObject(bucket_name(), object_name,
-                      storage::Generation(os.metadata()->generation()));
+                      storage::Generation(metadata->generation()));
 }
 
 }  // namespace