resolving comments

Author: shubham-up-47

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

@@ -29,6 +29,10 @@ namespace cloud {
 namespace storage_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
+// Defines the interface contract that any stream type (e.g., ReadStream,
+// WriteStream) managed by MultiStreamManager must implement. This explicit base
+// class ensures we have a clear, enforceable interface for operations like
+// CancelAll().
 class StreamBase {
  public:
   virtual ~StreamBase() = default;
@@ -41,8 +45,20 @@ class StreamBase {
 // are moved to the back of the queue for reuse.
 //
 // THREAD SAFETY:
-// This class is NOT thread-safe. The owner (ObjectDescriptorImpl) must
-// serialize access, typically by holding `mu_` while calling these methods.
+// This class is NOT thread-safe. The owner (e.g. ObjectDescriptorImpl
+// or AsyncWriterImpl etc) must serialize access, typically by holding
+// an external mutex while calling these methods.
+//
+// EXAMPLE USAGE:
+//   class MyOwner {
+//     std::mutex mu_;
+//     MultiStreamManager<MyStream, MyRange> manager_;
+//
+//     void StartRead() {
+//        std::unique_lock<std::mutex> lk(mu_);
+//        auto it = manager_.GetLeastBusyStream();
+//     }
+//   };
 template <typename StreamT, typename RangeT>
 class MultiStreamManager {
  public:
@@ -81,20 +97,20 @@ class MultiStreamManager {
     // In ObjectDescriptorImpl, we ensure there is always at least one stream,
     // but this assertion protects against future refactoring errors.
     assert(!streams_.empty());
-    auto best_it = streams_.begin();
+    auto least_busy_it = streams_.begin();
     // Track min_ranges to avoid calling .size() repeatedly if possible,
     // though for std::unordered_map .size() is O(1).
-    std::size_t min_ranges = best_it->active_ranges.size();
+    std::size_t min_ranges = least_busy_it->active_ranges.size();
 
     // 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) {
-        best_it = it;
+        least_busy_it = it;
         min_ranges = it->active_ranges.size();
       }
     }
-    return best_it;
+    return least_busy_it;
   }
 
   StreamIterator AddStream(std::shared_ptr<StreamT> stream) {

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

@@ -51,7 +51,6 @@ void ObjectDescriptorImpl::Start(
     google::storage::v2::BidiReadObjectResponse first_response) {
   std::unique_lock<std::mutex> lk(mu_);
   auto it = stream_manager_->GetLastStream();
-  // Unlock and start the Read loop first.
   lk.unlock();
   OnRead(it, std::move(first_response));
   // Acquire lock and queue the background stream.
@@ -87,7 +86,7 @@ void ObjectDescriptorImpl::MakeSubsequentStream() {
           [](StreamManager::Stream const& s) {
             auto const* rs = s.stream.get();
             return rs != nullptr && s.active_ranges.empty() &&
-                   !rs->write_pending;
+                   !rs->write_pending_;
           })) {
     return;
   }
@@ -153,7 +152,7 @@ ObjectDescriptorImpl::Read(ReadParams p) {
   auto it = stream_manager_->GetLeastBusyStream();
   auto const id = ++read_id_generator_;
   it->active_ranges.emplace(id, range);
-  auto& read_range = *it->stream->next_request.add_read_ranges();
+  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);
@@ -169,13 +168,13 @@ ObjectDescriptorImpl::Read(ReadParams p) {
 
 void ObjectDescriptorImpl::Flush(std::unique_lock<std::mutex> lk,
                                  StreamIterator it) {
-  if (it->stream->write_pending ||
-      it->stream->next_request.read_ranges().empty()) {
+  if (it->stream->write_pending_ ||
+      it->stream->next_request_.read_ranges().empty()) {
     return;
   }
-  it->stream->write_pending = true;
+  it->stream->write_pending_ = true;
   google::storage::v2::BidiReadObjectRequest request;
-  request.Swap(&it->stream->next_request);
+  request.Swap(&it->stream->next_request_);
 
   // Assign CurrentStream to a temporary variable to prevent
   // lifetime extension which can cause the lock to be held until the
@@ -191,14 +190,14 @@ void ObjectDescriptorImpl::Flush(std::unique_lock<std::mutex> lk,
 void ObjectDescriptorImpl::OnWrite(StreamIterator it, bool ok) {
   std::unique_lock<std::mutex> lk(mu_);
   if (!ok) return DoFinish(std::move(lk), it);
-  it->stream->write_pending = false;
+  it->stream->write_pending_ = false;
   Flush(std::move(lk), it);
 }
 
 void ObjectDescriptorImpl::DoRead(std::unique_lock<std::mutex> lk,
                                   StreamIterator it) {
-  if (it->stream->read_pending) return;
-  it->stream->read_pending = true;
+  if (it->stream->read_pending_) return;
+  it->stream->read_pending_ = true;
 
   // Assign CurrentStream to a temporary variable to prevent
   // lifetime extension which can cause the lock to be held until the
@@ -214,7 +213,7 @@ void ObjectDescriptorImpl::OnRead(
     StreamIterator it,
     absl::optional<google::storage::v2::BidiReadObjectResponse> response) {
   std::unique_lock<std::mutex> lk(mu_);
-  it->stream->read_pending = false;
+  it->stream->read_pending_ = false;
 
   if (!response) return DoFinish(std::move(lk), it);
   if (response->has_metadata()) {
@@ -243,7 +242,7 @@ void ObjectDescriptorImpl::OnRead(
 
 void ObjectDescriptorImpl::DoFinish(std::unique_lock<std::mutex> lk,
                                     StreamIterator it) {
-  it->stream->read_pending = false;
+  it->stream->read_pending_ = false;
   // Assign CurrentStream to a temporary variable to prevent
   // lifetime extension which can cause the lock to be held until the
   // end of the block.
@@ -293,8 +292,8 @@ void ObjectDescriptorImpl::OnResume(StreamIterator it,
 
   it->stream = std::make_shared<ReadStream>(std::move(result->stream),
                                             resume_policy_prototype_->clone());
-  it->stream->write_pending = false;
-  it->stream->read_pending = false;
+  it->stream->write_pending_ = false;
+  it->stream->read_pending_ = false;
 
   // TODO(#15105) - this should be done without release the lock.
   Flush(std::move(lk), it);

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

@@ -48,9 +48,9 @@ class ReadStream : public storage_internal::StreamBase {
 
   std::shared_ptr<OpenStream> stream_;
   std::unique_ptr<storage_experimental::ResumePolicy> resume_policy_;
-  google::storage::v2::BidiReadObjectRequest next_request;
-  bool write_pending = false;
-  bool read_pending = false;
+  google::storage::v2::BidiReadObjectRequest next_request_;
+  bool write_pending_ = false;
+  bool read_pending_ = false;
 };
 
 class ObjectDescriptorImpl