Index: net/quic/quic_stream_sequencer.cc |
diff --git a/net/quic/quic_stream_sequencer.cc b/net/quic/quic_stream_sequencer.cc |
index b9b7ad32700ac00fc43f26cc067143fc355f7e15..663ee3584a6fda7685ccbf29ed6098b8087ec053 100644 |
--- a/net/quic/quic_stream_sequencer.cc |
+++ b/net/quic/quic_stream_sequencer.cc |
@@ -17,6 +17,10 @@ using std::string; |
namespace net { |
+QuicStreamSequencer::FrameData::FrameData(QuicStreamOffset offset, |
+ string segment) |
+ : offset(offset), segment(segment) {} |
+ |
QuicStreamSequencer::QuicStreamSequencer(ReliableQuicStream* quic_stream) |
: stream_(quic_stream), |
num_bytes_consumed_(0), |
@@ -33,13 +37,14 @@ QuicStreamSequencer::~QuicStreamSequencer() { |
void QuicStreamSequencer::OnStreamFrame(const QuicStreamFrame& frame) { |
++num_frames_received_; |
- if (IsDuplicate(frame)) { |
+ FrameList::iterator insertion_point = FindInsertionPoint(frame); |
+ if (IsDuplicate(frame, insertion_point)) { |
++num_duplicate_frames_received_; |
// Silently ignore duplicates. |
return; |
} |
- if (FrameOverlapsBufferedData(frame)) { |
+ if (FrameOverlapsBufferedData(frame, insertion_point)) { |
stream_->CloseConnectionWithDetails( |
QUIC_INVALID_STREAM_FRAME, "Stream frame overlaps with buffered data."); |
return; |
@@ -65,7 +70,7 @@ void QuicStreamSequencer::OnStreamFrame(const QuicStreamFrame& frame) { |
++num_early_frames_received_; |
} |
- // If the frame has arrived in-order then we can process it immediately, only |
+ // If the frame has arrived in-order then process it immediately, only |
// buffering if the stream is unable to process it. |
size_t bytes_consumed = 0; |
if (!blocked_ && byte_offset == num_bytes_consumed_) { |
@@ -91,9 +96,11 @@ void QuicStreamSequencer::OnStreamFrame(const QuicStreamFrame& frame) { |
if (bytes_consumed < data_len) { |
DVLOG(1) << "Buffering stream data at offset " << byte_offset; |
const size_t remaining_length = data_len - bytes_consumed; |
- buffered_frames_.insert(std::make_pair( |
- byte_offset + bytes_consumed, |
- string(frame.data.data() + bytes_consumed, remaining_length))); |
+ // Inserting an empty string and then copying to avoid the extra copy. |
+ insertion_point = buffered_frames_.insert( |
+ insertion_point, FrameData(byte_offset + bytes_consumed, "")); |
+ insertion_point->segment = |
+ string(frame.data.data() + bytes_consumed, remaining_length); |
num_bytes_buffered_ += remaining_length; |
} |
} |
@@ -101,8 +108,7 @@ void QuicStreamSequencer::OnStreamFrame(const QuicStreamFrame& frame) { |
void QuicStreamSequencer::CloseStreamAtOffset(QuicStreamOffset offset) { |
const QuicStreamOffset kMaxOffset = numeric_limits<QuicStreamOffset>::max(); |
- // If we have a scheduled termination or close, any new offset should match |
- // it. |
+ // 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; |
@@ -130,16 +136,18 @@ bool QuicStreamSequencer::MaybeCloseStream() { |
int QuicStreamSequencer::GetReadableRegions(iovec* iov, size_t iov_len) const { |
DCHECK(!blocked_); |
- FrameMap::const_iterator it = buffered_frames_.begin(); |
+ 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->first != offset) return index; |
+ if (it->offset != offset) { |
+ return index; |
+ } |
- iov[index].iov_base = static_cast<void*>( |
- const_cast<char*>(it->second.data())); |
- iov[index].iov_len = it->second.size(); |
- offset += it->second.size(); |
+ 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; |
@@ -149,21 +157,19 @@ int QuicStreamSequencer::GetReadableRegions(iovec* iov, size_t iov_len) const { |
int QuicStreamSequencer::Readv(const struct iovec* iov, size_t iov_len) { |
DCHECK(!blocked_); |
- FrameMap::iterator it = buffered_frames_.begin(); |
+ 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->first == 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->second.size() - frame_offset); |
+ it->segment.size() - frame_offset); |
char* iov_ptr = static_cast<char*>(iov[iov_index].iov_base) + iov_offset; |
- memcpy(iov_ptr, |
- it->second.data() + frame_offset, bytes_to_read); |
+ memcpy(iov_ptr, it->segment.data() + frame_offset, bytes_to_read); |
frame_offset += bytes_to_read; |
iov_offset += bytes_to_read; |
@@ -172,67 +178,79 @@ int QuicStreamSequencer::Readv(const struct iovec* iov, size_t iov_len) { |
iov_offset = 0; |
++iov_index; |
} |
- if (it->second.size() == frame_offset) { |
+ if (it->segment.size() == frame_offset) { |
// We've copied this whole frame |
- RecordBytesConsumed(it->second.size()); |
+ RecordBytesConsumed(it->segment.size()); |
buffered_frames_.erase(it); |
it = buffered_frames_.begin(); |
frame_offset = 0; |
} |
} |
- // We've finished copying. If we have a partial frame, update it. |
+ // Done copying. If there is a partial frame, update it. |
if (frame_offset != 0) { |
- buffered_frames_.insert(std::make_pair(it->first + frame_offset, |
- it->second.substr(frame_offset))); |
- buffered_frames_.erase(buffered_frames_.begin()); |
+ 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); |
} |
bool QuicStreamSequencer::HasBytesToRead() const { |
- FrameMap::const_iterator it = buffered_frames_.begin(); |
- |
- return it != buffered_frames_.end() && it->first == num_bytes_consumed_; |
+ return !buffered_frames_.empty() && |
+ buffered_frames_.begin()->offset == num_bytes_consumed_; |
} |
bool QuicStreamSequencer::IsClosed() const { |
return num_bytes_consumed_ >= close_offset_; |
} |
-bool QuicStreamSequencer::FrameOverlapsBufferedData( |
- const QuicStreamFrame& frame) const { |
+QuicStreamSequencer::FrameList::iterator |
+QuicStreamSequencer::FindInsertionPoint(const QuicStreamFrame& frame) { |
if (buffered_frames_.empty()) { |
- return false; |
+ 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; |
+} |
- FrameMap::const_iterator next_frame = |
- buffered_frames_.lower_bound(frame.offset); |
- // Duplicate frames should have been dropped in IsDuplicate. |
- DCHECK(next_frame == buffered_frames_.end() || |
- next_frame->first != frame.offset); |
- |
- // If there is a buffered frame with a higher starting offset, then we check |
- // to see if the new frame runs into the higher frame. |
- if (next_frame != buffered_frames_.end() && |
- (frame.offset + frame.data.size()) > next_frame->first) { |
+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() << " > " << next_frame->first; |
+ << frame.data.size() << " > " << insertion_point->offset; |
return true; |
} |
- |
- // If there is a buffered frame with a lower starting offset, then we check |
- // to see if the buffered frame runs into the new frame. |
- if (next_frame != buffered_frames_.begin()) { |
- FrameMap::const_iterator preceeding_frame = --next_frame; |
- QuicStreamOffset offset = preceeding_frame->first; |
- uint64 data_length = preceeding_frame->second.length(); |
- if ((offset + data_length) > frame.offset) { |
- DVLOG(1) << "Preceeding frame overlaps new frame: " << offset << " + " |
- << data_length << " > " << frame.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; |
} |
@@ -240,40 +258,42 @@ 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_) { |
- FrameMap::iterator it = buffered_frames_.begin(); |
- if (it->first != 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->first |
- << " length: " << it->second.length(); |
+ << " end_offset: " << end_offset << " offset: " << it->offset |
+ << " length: " << it->segment.length(); |
stream_->Reset(QUIC_ERROR_PROCESSING_STREAM); |
return; |
} |
- if (it->first + it->second.length() <= end_offset) { |
- num_bytes_consumed_ += it->second.length(); |
- num_bytes_buffered_ -= it->second.length(); |
+ if (it->offset + it->segment.length() <= end_offset) { |
+ num_bytes_consumed_ += it->segment.length(); |
+ num_bytes_buffered_ -= it->segment.length(); |
// This chunk is entirely consumed. |
buffered_frames_.erase(it); |
continue; |
} |
// Partially consume this frame. |
- size_t delta = end_offset - it->first; |
+ size_t delta = end_offset - it->offset; |
RecordBytesConsumed(delta); |
- buffered_frames_.insert(make_pair(end_offset, it->second.substr(delta))); |
+ string new_data = it->segment.substr(delta); |
buffered_frames_.erase(it); |
+ buffered_frames_.push_front(FrameData(num_bytes_consumed_, new_data)); |
break; |
} |
} |
-bool QuicStreamSequencer::IsDuplicate(const QuicStreamFrame& frame) const { |
- // A frame is duplicate if the frame offset is smaller than our bytes consumed |
- // or we have stored the frame in our map. |
- // TODO(pwestin): Is it possible that a new frame contain more data even if |
- // the offset is the same? |
+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_ || |
- buffered_frames_.find(frame.offset) != buffered_frames_.end(); |
+ (insertion_point != buffered_frames_.end() && |
+ frame.offset == insertion_point->offset); |
} |
void QuicStreamSequencer::SetBlockedUntilFlush() { |
@@ -282,10 +302,10 @@ void QuicStreamSequencer::SetBlockedUntilFlush() { |
void QuicStreamSequencer::FlushBufferedFrames() { |
blocked_ = false; |
- FrameMap::iterator it = buffered_frames_.find(num_bytes_consumed_); |
- while (it != buffered_frames_.end()) { |
- DVLOG(1) << "Flushing buffered packet at offset " << it->first; |
- string* data = &it->second; |
+ FrameList::iterator it = buffered_frames_.begin(); |
+ while (it != buffered_frames_.end() && it->offset == num_bytes_consumed_) { |
+ DVLOG(1) << "Flushing buffered packet at offset " << it->offset; |
+ const string* data = &it->segment; |
size_t bytes_consumed = stream_->ProcessRawData(data->c_str(), |
data->size()); |
RecordBytesConsumed(bytes_consumed); |
@@ -295,15 +315,14 @@ void QuicStreamSequencer::FlushBufferedFrames() { |
if (bytes_consumed > data->size()) { |
stream_->Reset(QUIC_ERROR_PROCESSING_STREAM); // Programming error |
return; |
- } else if (bytes_consumed == data->size()) { |
- buffered_frames_.erase(it); |
- it = buffered_frames_.find(num_bytes_consumed_); |
- } else { |
- string new_data = it->second.substr(bytes_consumed); |
+ } |
+ if (bytes_consumed < data->size()) { |
+ string new_data = data->substr(bytes_consumed); |
buffered_frames_.erase(it); |
- buffered_frames_.insert(std::make_pair(num_bytes_consumed_, new_data)); |
+ buffered_frames_.push_front(FrameData(num_bytes_consumed_, new_data)); |
return; |
} |
+ buffered_frames_.erase(it++); |
} |
MaybeCloseStream(); |
} |