relnote: refactoring the buffering logic in QUIC's stream sequencer.

net/quic/quic_stream_sequencer.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/quic/quic_stream_sequencer.h"
 
 #include <algorithm>
 #include <limits>
 #include <utility>
 
 #include "base/logging.h"
+#include "net/quic/quic_frame_list.h"
 #include "net/quic/reliable_quic_stream.h"
 
 using std::min;
 using std::numeric_limits;
 using std::string;
 
 namespace net {
 
-QuicStreamSequencer::FrameData::FrameData(QuicStreamOffset offset,
-                                          const string& segment)
-    : offset(offset), segment(segment) {}
-
 QuicStreamSequencer::QuicStreamSequencer(ReliableQuicStream* quic_stream)
     : stream_(quic_stream),
       num_bytes_consumed_(0),
       close_offset_(numeric_limits<QuicStreamOffset>::max()),
       blocked_(false),
       num_bytes_buffered_(0),
       num_frames_received_(0),
       num_duplicate_frames_received_(0),
-      num_early_frames_received_(0) {
-}
+      num_early_frames_received_(0) {}
 
-QuicStreamSequencer::~QuicStreamSequencer() {
-}
+QuicStreamSequencer::~QuicStreamSequencer() {}
 
 void QuicStreamSequencer::OnStreamFrame(const QuicStreamFrame& frame) {
   ++num_frames_received_;
-  FrameList::iterator insertion_point = FindInsertionPoint(frame);
-  if (IsDuplicate(frame, insertion_point)) {
-    ++num_duplicate_frames_received_;
-    // Silently ignore duplicates.
-    return;
-  }
-
-  if (FrameOverlapsBufferedData(frame, insertion_point)) {
-    stream_->CloseConnectionWithDetails(
-        QUIC_INVALID_STREAM_FRAME, "Stream frame overlaps with buffered data.");
-    return;
-  }
-
   const QuicStreamOffset byte_offset = frame.offset;
   const size_t data_len = frame.data.length();
   if (data_len == 0 && !frame.fin) {
     // Stream frames must have data or a fin flag.
     stream_->CloseConnectionWithDetails(QUIC_INVALID_STREAM_FRAME,
                                         "Empty stream frame without FIN set.");
     return;
   }
 
   if (frame.fin) {
     CloseStreamAtOffset(frame.offset + data_len);
     if (data_len == 0) {
       return;
     }
   }
+  size_t bytes_written;
+  QuicErrorCode result =
+      buffered_frames_.WriteAtOffset(byte_offset, frame.data, &bytes_written);
+
+  if (result == QUIC_INVALID_STREAM_DATA) {
+    stream_->CloseConnectionWithDetails(
+        QUIC_INVALID_STREAM_FRAME, "Stream frame overlaps with buffered data.");
+    return;
+  }
+  if (result == QUIC_NO_ERROR && bytes_written == 0) {
+    ++num_duplicate_frames_received_;
+    // Silently ignore duplicates.
+    return;
+  }
 
   if (byte_offset > num_bytes_consumed_) {
     ++num_early_frames_received_;
   }
 
-  DVLOG(1) << "Buffering stream data at offset " << byte_offset;
-  // Inserting an empty string and then copying to avoid the extra copy.
-  insertion_point =
-      buffered_frames_.insert(insertion_point, FrameData(byte_offset, ""));
-  frame.data.CopyToString(&insertion_point->segment);
   num_bytes_buffered_ += data_len;
 
   if (blocked_) {
     return;
   }
 
   if (byte_offset == num_bytes_consumed_) {
     stream_->OnDataAvailable();
   }
 }
 
 void QuicStreamSequencer::CloseStreamAtOffset(QuicStreamOffset offset) {
   const QuicStreamOffset kMaxOffset = numeric_limits<QuicStreamOffset>::max();
 
   // If there is a scheduled close, the new offset should match it.
   if (close_offset_ != kMaxOffset && offset != close_offset_) {
     stream_->Reset(QUIC_MULTIPLE_TERMINATION_OFFSETS);
     return;
   }
 
   close_offset_ = offset;
 
   MaybeCloseStream();
 }
 
 bool QuicStreamSequencer::MaybeCloseStream() {
   if (!blocked_ && IsClosed()) {
     DVLOG(1) << "Passing up termination, as we've processed "
-             << num_bytes_consumed_ << " of " << close_offset_
-             << " bytes.";
+             << num_bytes_consumed_ << " of " << close_offset_ << " bytes.";
     // This will cause the stream to consume the fin.
     // Technically it's an error if num_bytes_consumed isn't exactly
     // equal, but error handling seems silly at this point.
     stream_->OnDataAvailable();
-    buffered_frames_.clear();
+    buffered_frames_.Clear();
     num_bytes_buffered_ = 0;
     return true;
   }
   return false;
 }
 
 int QuicStreamSequencer::GetReadableRegions(iovec* iov, size_t iov_len) const {
   DCHECK(!blocked_);
-  FrameList::const_iterator it = buffered_frames_.begin();
-  size_t index = 0;
-  QuicStreamOffset offset = num_bytes_consumed_;
-  while (it != buffered_frames_.end() && index < iov_len) {
-    if (it->offset != offset) {
-      return index;
-    }
-
-    iov[index].iov_base =
-        static_cast<void*>(const_cast<char*>(it->segment.data()));
-    iov[index].iov_len = it->segment.size();
-    offset += it->segment.size();
-
-    ++index;
-    ++it;
-  }
-  return index;
+  return buffered_frames_.GetReadableRegions(iov, iov_len);
 }
 
 int QuicStreamSequencer::Readv(const struct iovec* iov, size_t iov_len) {
   DCHECK(!blocked_);
-  FrameList::iterator it = buffered_frames_.begin();
-  size_t iov_index = 0;
-  size_t iov_offset = 0;
-  size_t frame_offset = 0;
-  QuicStreamOffset initial_bytes_consumed = num_bytes_consumed_;
-
-  while (iov_index < iov_len && it != buffered_frames_.end() &&
-         it->offset == num_bytes_consumed_) {
-    int bytes_to_read = min(iov[iov_index].iov_len - iov_offset,
-                            it->segment.size() - frame_offset);
-
-    char* iov_ptr = static_cast<char*>(iov[iov_index].iov_base) + iov_offset;
-    memcpy(iov_ptr, it->segment.data() + frame_offset, bytes_to_read);
-    frame_offset += bytes_to_read;
-    iov_offset += bytes_to_read;
-
-    if (iov[iov_index].iov_len == iov_offset) {
-      // We've filled this buffer.
-      iov_offset = 0;
-      ++iov_index;
-    }
-    if (it->segment.size() == frame_offset) {
-      // We've copied this whole frame
-      RecordBytesConsumed(it->segment.size());
-      buffered_frames_.erase(it);
-      it = buffered_frames_.begin();
-      frame_offset = 0;
-    }
-  }
-  // Done copying.  If there is a partial frame, update it.
-  if (frame_offset != 0) {
-    buffered_frames_.push_front(
-        FrameData(it->offset + frame_offset, it->segment.substr(frame_offset)));
-    buffered_frames_.erase(it);
-    RecordBytesConsumed(frame_offset);
-  }
-  return static_cast<int>(num_bytes_consumed_ - initial_bytes_consumed);
+  size_t bytes_read = buffered_frames_.ReadvAndInvalidate(iov, iov_len);
+  RecordBytesConsumed(bytes_read);
+  return static_cast<int>(bytes_read);
 }
 
 bool QuicStreamSequencer::HasBytesToRead() const {
-  return !buffered_frames_.empty() &&
-         buffered_frames_.begin()->offset == num_bytes_consumed_;
+  return buffered_frames_.HasBytesToRead();
 }
 
 bool QuicStreamSequencer::IsClosed() const {
   return num_bytes_consumed_ >= close_offset_;
 }
 
-QuicStreamSequencer::FrameList::iterator
-QuicStreamSequencer::FindInsertionPoint(const QuicStreamFrame& frame) {
-  if (buffered_frames_.empty()) {
-    return buffered_frames_.begin();
-  }
-  // If it's after all buffered_frames, return the end.
-  if (frame.offset >= (buffered_frames_.rbegin()->offset +
-                       buffered_frames_.rbegin()->segment.length())) {
-    return buffered_frames_.end();
-  }
-  FrameList::iterator iter = buffered_frames_.begin();
-  // Only advance the iterator if the data begins after the already received
-  // frame.  If the new frame overlaps with an existing frame, the iterator will
-  // still point to the frame it overlaps with.
-  while (iter != buffered_frames_.end() &&
-         frame.offset >= iter->offset + iter->segment.length()) {
-    ++iter;
-  }
-  return iter;
-}
-
-bool QuicStreamSequencer::FrameOverlapsBufferedData(
-    const QuicStreamFrame& frame,
-    FrameList::const_iterator insertion_point) const {
-  if (buffered_frames_.empty() || insertion_point == buffered_frames_.end()) {
-    return false;
-  }
-  // If there is a buffered frame with a higher starting offset, then check to
-  // see if the new frame overlaps the beginning of the higher frame.
-  if (frame.offset < insertion_point->offset &&
-      frame.offset + frame.data.length() > insertion_point->offset) {
-    DVLOG(1) << "New frame overlaps next frame: " << frame.offset << " + "
-             << frame.data.size() << " > " << insertion_point->offset;
-    return true;
-  }
-  // If there is a buffered frame with a lower starting offset, then check to
-  // see if the buffered frame runs into the new frame.
-  if (frame.offset >= insertion_point->offset &&
-      frame.offset <
-          insertion_point->offset + insertion_point->segment.length()) {
-    DVLOG(1) << "Preceeding frame overlaps new frame: "
-             << insertion_point->offset << " + "
-             << insertion_point->segment.length() << " > " << frame.offset;
-    return true;
-  }
-
-  return false;
-}
-
 void QuicStreamSequencer::MarkConsumed(size_t num_bytes_consumed) {
   DCHECK(!blocked_);
-  size_t end_offset = num_bytes_consumed_ + num_bytes_consumed;
-  while (!buffered_frames_.empty() && end_offset != num_bytes_consumed_) {
-    FrameList::iterator it = buffered_frames_.begin();
-    if (it->offset != num_bytes_consumed_) {
-      LOG(DFATAL) << "Invalid argument to MarkConsumed. "
-                  << " num_bytes_consumed_: " << num_bytes_consumed_
-                  << " end_offset: " << end_offset << " offset: " << it->offset
-                  << " length: " << it->segment.length();
-      stream_->Reset(QUIC_ERROR_PROCESSING_STREAM);
-      return;
-    }
-
-    if (it->offset + it->segment.length() <= end_offset) {
-      RecordBytesConsumed(it->segment.length());
-      // This chunk is entirely consumed.
-      buffered_frames_.erase(it);
-      continue;
-    }
-
-    // Partially consume this frame.
-    size_t delta = end_offset - it->offset;
-    RecordBytesConsumed(delta);
-    string new_data = it->segment.substr(delta);
-    buffered_frames_.erase(it);
-    buffered_frames_.push_front(FrameData(num_bytes_consumed_, new_data));
-    break;
+  bool result =
+      buffered_frames_.IncreaseTotalReadAndInvalidate(num_bytes_consumed);
+  if (!result) {
+    LOG(DFATAL) << "Invalid argument to MarkConsumed."
+                << " expect to consume: " << num_bytes_consumed
+                << ", but not enough bytes available.";
+    stream_->Reset(QUIC_ERROR_PROCESSING_STREAM);
+    return;
   }
-}
-
-bool QuicStreamSequencer::IsDuplicate(
-    const QuicStreamFrame& frame,
-    FrameList::const_iterator insertion_point) const {
-  // A frame is duplicate if the frame offset is smaller than the bytes consumed
-  // or identical to an already received frame.
-  return frame.offset < num_bytes_consumed_ ||
-         (insertion_point != buffered_frames_.end() &&
-          frame.offset == insertion_point->offset);
+  RecordBytesConsumed(num_bytes_consumed);
 }
 
 void QuicStreamSequencer::SetBlockedUntilFlush() {
   blocked_ = true;
 }
 
 void QuicStreamSequencer::SetUnblocked() {
   blocked_ = false;
   if (IsClosed() || HasBytesToRead()) {
     stream_->OnDataAvailable();
   }
 }
 
 void QuicStreamSequencer::RecordBytesConsumed(size_t bytes_consumed) {
   num_bytes_consumed_ += bytes_consumed;
   num_bytes_buffered_ -= bytes_consumed;
 
   stream_->AddBytesConsumed(bytes_consumed);
 }
 
 }  // namespace net