resolving comments

Author: shubham-up-47

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

@@ -41,7 +41,7 @@ class StreamBase {
 // Manages a collection of streams.
 //
 // This class implements the "Subsequent Stream" logic where idle streams
-// are moved to the back of the queue for reuse.
+// are moved to the front of the queue for reuse.
 //
 // THREAD SAFETY:
 // This class is NOT thread-safe. The owner (e.g. ObjectDescriptorImpl
@@ -83,9 +83,9 @@ class MultiStreamManager {
     streams_.push_back(Stream{std::move(initial_stream), {}});
   }
 
-  StreamIterator GetLastStream() {
+  StreamIterator GetFirstStream() {
     if (streams_.empty()) return streams_.end();
-    return std::prev(streams_.end());
+    return streams_.begin();
   }
 
   StreamIterator GetLeastBusyStream() {
@@ -94,21 +94,24 @@ class MultiStreamManager {
     // Track min_ranges to avoid calling .size() repeatedly if possible,
     // though for std::unordered_map .size() is O(1).
     std::size_t min_ranges = least_busy_it->active_ranges.size();
+    if (min_ranges == 0) return least_busy_it;
 
     // Start checking from the second element
     for (auto it = std::next(streams_.begin()); it != streams_.end(); ++it) {
       // Strict less-than ensures stability (preferring older streams if tied)
-      if (it->active_ranges.size() < min_ranges) {
+      auto size = it->active_ranges.size();
+      if (size < min_ranges) {
         least_busy_it = it;
-        min_ranges = it->active_ranges.size();
+        min_ranges = size;
+        if (min_ranges == 0) return least_busy_it;
       }
     }
     return least_busy_it;
   }
 
   StreamIterator AddStream(std::shared_ptr<StreamT> stream) {
-    streams_.push_back(Stream{std::move(stream), {}});
-    return std::prev(streams_.end());
+    streams_.push_front(Stream{std::move(stream), {}});
+    return streams_.begin();
   }
 
   void CancelAll() {
@@ -141,16 +144,14 @@ class MultiStreamManager {
   }
 
   template <typename Pred>
-  bool ReuseIdleStreamToBack(Pred pred) {
+  bool ReuseIdleStreamToFront(Pred pred) {
     for (auto it = streams_.begin(); it != streams_.end(); ++it) {
       if (!pred(*it)) continue;
 
-      // If the idle stream is already at the back, we don't
-      // need to move it. If it's elsewhere, use splice() to move the node.
-      // splice() is O(1) and, crucially, does not invalidate iterators
-      // or copy the Stream object.
-      if (std::next(it) != streams_.end()) {
-        streams_.splice(streams_.end(), streams_, it);
+      // If the idle stream is already at the front, we don't
+      // need to move it. Otherwise splice to the front in O(1).
+      if (it != streams_.begin()) {
+        streams_.splice(streams_.begin(), streams_, it);
       }
       return true;
     }

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

@@ -50,26 +50,26 @@ TEST(MultiStreamManagerTest, ConstructsWithFactoryAndHasOneStream) {
   auto mgr = MultiStreamManagerTest::MakeManager();
   EXPECT_FALSE(mgr.Empty());
   EXPECT_EQ(mgr.Size(), 1U);
-  auto it = mgr.GetLastStream();
+  auto it = mgr.GetFirstStream();
   ASSERT_TRUE(it->stream);
 }
 
 TEST(MultiStreamManagerTest, ConstructsWithInitialStream) {
   auto initial = std::make_shared<FakeStream>();
   Manager mgr([] { return nullptr; }, initial);
   EXPECT_EQ(mgr.Size(), 1U);
-  auto it = mgr.GetLastStream();
+  auto it = mgr.GetFirstStream();
   EXPECT_EQ(it->stream, initial);
 }
 
-TEST(MultiStreamManagerTest, AddStreamAppendsAndGetLastReturnsNew) {
+TEST(MultiStreamManagerTest, AddStreamAppendsAndGetFirstReturnsNew) {
   auto mgr = MultiStreamManagerTest::MakeManager();
   auto s1 = std::make_shared<FakeStream>();
   auto it1 = mgr.AddStream(s1);
   EXPECT_EQ(mgr.Size(), 2U);
   EXPECT_EQ(it1->stream.get(), s1.get());
-  auto it_last = mgr.GetLastStream();
-  EXPECT_EQ(it_last->stream.get(), s1.get());
+  auto it_first = mgr.GetFirstStream();
+  EXPECT_EQ(it_first->stream.get(), s1.get());
 }
 
 TEST(MultiStreamManagerTest, GetLeastBusyPrefersFewestActiveRanges) {
@@ -78,7 +78,7 @@ TEST(MultiStreamManagerTest, GetLeastBusyPrefersFewestActiveRanges) {
   // The manager starts with an initial stream (size 0).
   // We must make it "busy" so it doesn't win the comparison against our test
   // streams.
-  auto it_init = mgr.GetLastStream();
+  auto it_init = mgr.GetFirstStream();
   it_init->active_ranges.emplace(999, std::make_shared<FakeRange>());
   it_init->active_ranges.emplace(998, std::make_shared<FakeRange>());
 
@@ -103,7 +103,7 @@ TEST(MultiStreamManagerTest, GetLeastBusyPrefersFewestActiveRanges) {
 
 TEST(MultiStreamManagerTest, CleanupDoneRangesRemovesFinished) {
   auto mgr = MultiStreamManagerTest::MakeManager();
-  auto it = mgr.GetLastStream();
+  auto it = mgr.GetFirstStream();
   auto r1 = std::make_shared<FakeRange>();
   r1->done = false;
   auto r2 = std::make_shared<FakeRange>();
@@ -120,7 +120,7 @@ TEST(MultiStreamManagerTest, CleanupDoneRangesRemovesFinished) {
 
 TEST(MultiStreamManagerTest, RemoveStreamAndNotifyRangesCallsOnFinish) {
   auto mgr = MultiStreamManagerTest::MakeManager();
-  auto it = mgr.GetLastStream();
+  auto it = mgr.GetFirstStream();
   auto r1 = std::make_shared<FakeRange>();
   auto r2 = std::make_shared<FakeRange>();
   it->active_ranges.emplace(11, r1);
@@ -142,49 +142,50 @@ TEST(MultiStreamManagerTest, CancelAllInvokesCancel) {
   EXPECT_EQ(s2->cancelled, 1);
 }
 
-TEST(MultiStreamManagerTest, ReuseIdleStreamToBackMovesElement) {
+TEST(MultiStreamManagerTest, ReuseIdleStreamToFrontMovesElement) {
   auto mgr = MultiStreamManagerTest::MakeManager();
   // Capture the factory-created stream pointer (initial element)
-  auto* factory_ptr = mgr.GetLastStream()->stream.get();
+  auto* factory_ptr = mgr.GetFirstStream()->stream.get();
   auto s1 = std::make_shared<FakeStream>();
   mgr.AddStream(s1);
-  bool moved = mgr.ReuseIdleStreamToBack([](Manager::Stream const& s) {
+  bool moved = mgr.ReuseIdleStreamToFront([](Manager::Stream const& s) {
     auto* fs = s.stream.get();
     return fs != nullptr && s.active_ranges.empty() && !fs->write_pending;
   });
   EXPECT_TRUE(moved);
-  auto it_last = mgr.GetLastStream();
-  // After move, the factory stream should be last
-  EXPECT_EQ(it_last->stream.get(), factory_ptr);
-  EXPECT_NE(it_last->stream.get(), s1.get());
+  auto it_first = mgr.GetFirstStream();
+  // After move, the s1 stream should be first
+  EXPECT_EQ(it_first->stream.get(), s1.get());
+  EXPECT_NE(it_first->stream.get(), factory_ptr);
 }
 
 TEST(MultiStreamManagerTest,
-     ReuseIdleStreamAlreadyAtBackReturnsTrueWithoutMove) {
+     ReuseIdleStreamAlreadyAtFrontReturnsTrueWithoutMove) {
   auto mgr = MultiStreamManagerTest::MakeManager();
-  // The manager starts with one stream. It is the last stream, and it is idle.
-  auto initial_last = mgr.GetLastStream();
-  bool reused = mgr.ReuseIdleStreamToBack(
+  // The manager starts with one stream. It is the first stream, and it is idle.
+  auto initial_first = mgr.GetFirstStream();
+  bool reused = mgr.ReuseIdleStreamToFront(
       [](Manager::Stream const& s) { return s.active_ranges.empty(); });
   EXPECT_TRUE(reused);
-  // Pointer should remain the same (it was already at the back)
-  EXPECT_EQ(mgr.GetLastStream(), initial_last);
+  // Pointer should remain the same (it was already at the front)
+  EXPECT_EQ(mgr.GetFirstStream(), initial_first);
 }
 
 TEST(MultiStreamManagerTest, ReuseIdleStreamDoesNotMoveWhenWritePending) {
   auto mgr = MultiStreamManagerTest::MakeManager();
+  auto factory_ptr = mgr.GetFirstStream()->stream.get();
   // Mark factory stream as not reusable
-  mgr.GetLastStream()->stream->write_pending = true;
+  mgr.GetFirstStream()->stream->write_pending = true;
   auto s1 = std::make_shared<FakeStream>();
   s1->write_pending = true;  // also mark appended stream as not reusable
   mgr.AddStream(s1);
-  bool moved = mgr.ReuseIdleStreamToBack([](Manager::Stream const& s) {
+  bool moved = mgr.ReuseIdleStreamToFront([](Manager::Stream const& s) {
     auto* fs = s.stream.get();
     return fs != nullptr && s.active_ranges.empty() && !fs->write_pending;
   });
   EXPECT_FALSE(moved);
-  auto it_last = mgr.GetLastStream();
-  EXPECT_EQ(it_last->stream.get(), s1.get());
+  auto it_first = mgr.GetFirstStream();
+  EXPECT_EQ(it_first->stream.get(), factory_ptr);
 }
 
 TEST(MultiStreamManagerTest, MoveActiveRangesTransfersAllEntries) {
@@ -204,24 +205,24 @@ TEST(MultiStreamManagerTest, MoveActiveRangesTransfersAllEntries) {
   EXPECT_TRUE(it2->active_ranges.count(202));
 }
 
-TEST(MultiStreamManagerTest, GetLastStreamReflectsRecentAppendAndReuse) {
+TEST(MultiStreamManagerTest, GetFirstStreamReflectsFrontReuse) {
   auto mgr = MultiStreamManagerTest::MakeManager();
   auto s1 = std::make_shared<FakeStream>();
   mgr.AddStream(s1);
-  EXPECT_EQ(mgr.GetLastStream()->stream.get(), s1.get());
-  bool moved = mgr.ReuseIdleStreamToBack([](Manager::Stream const& s) {
+  EXPECT_EQ(mgr.GetFirstStream()->stream.get(), s1.get());
+  bool moved = mgr.ReuseIdleStreamToFront([](Manager::Stream const& s) {
     return s.stream != nullptr && s.active_ranges.empty();
   });
   EXPECT_TRUE(moved);
-  auto it_last = mgr.GetLastStream();
-  EXPECT_NE(it_last->stream.get(), s1.get());
+  auto it_first = mgr.GetFirstStream();
+  EXPECT_EQ(it_first->stream.get(), s1.get());
 }
 
 TEST(MultiStreamManagerTest, EmptyAndSizeTransitions) {
   auto mgr = MultiStreamManagerTest::MakeManager();
   EXPECT_FALSE(mgr.Empty());
   EXPECT_EQ(mgr.Size(), 1U);
-  auto it = mgr.GetLastStream();
+  auto it = mgr.GetFirstStream();
   mgr.RemoveStreamAndNotifyRanges(it, Status());
   EXPECT_TRUE(mgr.Empty());
   EXPECT_EQ(mgr.Size(), 0U);

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

@@ -50,7 +50,7 @@ ObjectDescriptorImpl::~ObjectDescriptorImpl() { Cancel(); }
 void ObjectDescriptorImpl::Start(
     google::storage::v2::BidiReadObjectResponse first_response) {
   std::unique_lock<std::mutex> lk(mu_);
-  auto it = stream_manager_->GetLastStream();
+  auto it = stream_manager_->GetFirstStream();
   if (it == stream_manager_->End()) return;
   lk.unlock();
   OnRead(it, std::move(first_response));
@@ -83,7 +83,7 @@ void ObjectDescriptorImpl::AssurePendingStreamQueued() {
 void ObjectDescriptorImpl::MakeSubsequentStream() {
   std::unique_lock<std::mutex> lk(mu_);
   // Reuse an idle stream if possible.
-  if (stream_manager_->ReuseIdleStreamToBack(
+  if (stream_manager_->ReuseIdleStreamToFront(
           [](StreamManager::Stream const& s) {
             auto const* rs = s.stream.get();
             return rs != nullptr && s.active_ranges.empty() &&
@@ -151,16 +151,6 @@ ObjectDescriptorImpl::Read(ReadParams p) {
   }
 
   auto it = stream_manager_->GetLeastBusyStream();
-  if (it == stream_manager_->End()) {
-    lk.unlock();
-    range->OnFinish(Status(StatusCode::kFailedPrecondition,
-                           "Cannot read object, all streams failed"));
-    if (!internal::TracingEnabled(options_)) {
-      return std::unique_ptr<storage_experimental::AsyncReaderConnection>(
-          std::make_unique<ObjectDescriptorReader>(std::move(range)));
-    }
-    return MakeTracingObjectDescriptorReader(std::move(range));
-  }
   auto const id = ++read_id_generator_;
   it->active_ranges.emplace(id, range);
   auto& read_range = *it->stream->next_request.add_read_ranges();