src/net/quic/quic_stream_sequencer.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "net/quic/quic_stream_sequencer.h"
   6
   7 #include <algorithm>
   8 #include <limits>
   9
  10 #include "base/logging.h"
  11 #include "net/quic/reliable_quic_stream.h"
  12
  13 using std::min;
  14 using std::numeric_limits;
  15
  16 namespace net {
  17
  18 QuicStreamSequencer::QuicStreamSequencer(ReliableQuicStream* quic_stream)
  19     : stream_(quic_stream),
  20       num_bytes_consumed_(0),
  21       max_frame_memory_(numeric_limits<size_t>::max()),
  22       close_offset_(numeric_limits<QuicStreamOffset>::max()) {
  23 }
  24
  25 QuicStreamSequencer::QuicStreamSequencer(size_t max_frame_memory,
  26                                          ReliableQuicStream* quic_stream)
  27     : stream_(quic_stream),
  28       num_bytes_consumed_(0),
  29       max_frame_memory_(max_frame_memory),
  30       close_offset_(numeric_limits<QuicStreamOffset>::max()) {
  31   if (max_frame_memory < kMaxPacketSize) {
  32     LOG(DFATAL) << "Setting max frame memory to " << max_frame_memory
  33                 << ".  Some frames will be impossible to handle.";
  34   }
  35 }
  36
  37 QuicStreamSequencer::~QuicStreamSequencer() {
  38 }
  39
  40 bool QuicStreamSequencer::WillAcceptStreamFrame(
  41     const QuicStreamFrame& frame) const {
  42   size_t data_len = frame.data.size();
  43   DCHECK_LE(data_len, max_frame_memory_);
  44
  45   if (IsDuplicate(frame)) {
  46     return true;
  47   }
  48   QuicStreamOffset byte_offset = frame.offset;
  49   if (data_len > max_frame_memory_) {
  50     // We're never going to buffer this frame and we can't pass it up.
  51     // The stream might only consume part of it and we'd need a partial ack.
  52     //
  53     // Ideally this should never happen, as we check that
  54     // max_frame_memory_ > kMaxPacketSize and lower levels should reject
  55     // frames larger than that.
  56     return false;
  57   }
  58   if (byte_offset + data_len - num_bytes_consumed_ > max_frame_memory_) {
  59     // We can buffer this but not right now.  Toss it.
  60     // It might be worth trying an experiment where we try best-effort buffering
  61     return false;
  62   }
  63   return true;
  64 }
  65
  66 bool QuicStreamSequencer::OnStreamFrame(const QuicStreamFrame& frame) {
  67   if (!WillAcceptStreamFrame(frame)) {
  68     // This should not happen, as WillAcceptFrame should be called before
  69     // OnStreamFrame.  Error handling should be done by the caller.
  70     return false;
  71   }
  72   if (IsDuplicate(frame)) {
  73     // Silently ignore duplicates.
  74     return true;
  75   }
  76
  77   QuicStreamOffset byte_offset = frame.offset;
  78   const char* data = frame.data.data();
  79   size_t data_len = frame.data.size();
  80
  81   if (data_len == 0 && !frame.fin) {
  82     // Stream frames must have data or a fin flag.
  83     stream_->CloseConnectionWithDetails(QUIC_INVALID_STREAM_FRAME,
  84                                         "Empty stream frame without FIN set.");
  85     return false;
  86   }
  87
  88   if (frame.fin) {
  89     CloseStreamAtOffset(frame.offset + frame.data.size());
  90     if (data_len == 0) {
  91       return true;
  92     }
  93   }
  94
  95   if (byte_offset == num_bytes_consumed_) {
  96     DVLOG(1) << "Processing byte offset " << byte_offset;
  97     size_t bytes_consumed = stream_->ProcessRawData(data, data_len);
  98     num_bytes_consumed_ += bytes_consumed;
  99
 100     if (MaybeCloseStream()) {
 101       return true;
 102     }
 103     if (bytes_consumed > data_len) {
 104       stream_->Close(QUIC_ERROR_PROCESSING_STREAM);
 105       return false;
 106     } else if (bytes_consumed == data_len) {
 107       FlushBufferedFrames();
 108       return true;  // it's safe to ack this frame.
 109     } else {
 110       // Set ourselves up to buffer what's left
 111       data_len -= bytes_consumed;
 112       data += bytes_consumed;
 113       byte_offset += bytes_consumed;
 114     }
 115   }
 116   DVLOG(1) << "Buffering packet at offset " << byte_offset;
 117   frames_.insert(make_pair(byte_offset, string(data, data_len)));
 118   return true;
 119 }
 120
 121 void QuicStreamSequencer::CloseStreamAtOffset(QuicStreamOffset offset) {
 122   const QuicStreamOffset kMaxOffset = numeric_limits<QuicStreamOffset>::max();
 123
 124   // If we have a scheduled termination or close, any new offset should match
 125   // it.
 126   if (close_offset_ != kMaxOffset && offset != close_offset_) {
 127     stream_->Close(QUIC_MULTIPLE_TERMINATION_OFFSETS);
 128     return;
 129   }
 130
 131   close_offset_ = offset;
 132
 133   MaybeCloseStream();
 134 }
 135
 136 bool QuicStreamSequencer::MaybeCloseStream() {
 137   if (IsHalfClosed()) {
 138     DVLOG(1) << "Passing up termination, as we've processed "
 139              << num_bytes_consumed_ << " of " << close_offset_
 140              << " bytes.";
 141     // Technically it's an error if num_bytes_consumed isn't exactly
 142     // equal, but error handling seems silly at this point.
 143     stream_->TerminateFromPeer(true);
 144     return true;
 145   }
 146   return false;
 147 }
 148
 149 int QuicStreamSequencer::GetReadableRegions(iovec* iov, size_t iov_len) {
 150   FrameMap::iterator it = frames_.begin();
 151   size_t index = 0;
 152   QuicStreamOffset offset = num_bytes_consumed_;
 153   while (it != frames_.end() && index < iov_len) {
 154     if (it->first != offset) return index;
 155
 156     iov[index].iov_base = static_cast<void*>(
 157         const_cast<char*>(it->second.data()));
 158     iov[index].iov_len = it->second.size();
 159     offset += it->second.size();
 160
 161     ++index;
 162     ++it;
 163   }
 164   return index;
 165 }
 166
 167 int QuicStreamSequencer::Readv(const struct iovec* iov, size_t iov_len) {
 168   FrameMap::iterator it = frames_.begin();
 169   size_t iov_index = 0;
 170   size_t iov_offset = 0;
 171   size_t frame_offset = 0;
 172   size_t initial_bytes_consumed = num_bytes_consumed_;
 173
 174   while (iov_index < iov_len &&
 175          it != frames_.end() &&
 176          it->first == num_bytes_consumed_) {
 177     int bytes_to_read = min(iov[iov_index].iov_len - iov_offset,
 178                             it->second.size() - frame_offset);
 179
 180     char* iov_ptr = static_cast<char*>(iov[iov_index].iov_base) + iov_offset;
 181     memcpy(iov_ptr,
 182            it->second.data() + frame_offset, bytes_to_read);
 183     frame_offset += bytes_to_read;
 184     iov_offset += bytes_to_read;
 185
 186     if (iov[iov_index].iov_len == iov_offset) {
 187       // We've filled this buffer.
 188       iov_offset = 0;
 189       ++iov_index;
 190     }
 191     if (it->second.size() == frame_offset) {
 192       // We've copied this whole frame
 193       num_bytes_consumed_ += it->second.size();
 194       frames_.erase(it);
 195       it = frames_.begin();
 196       frame_offset = 0;
 197     }
 198   }
 199   // We've finished copying.  If we have a partial frame, update it.
 200   if (frame_offset != 0) {
 201     frames_.insert(make_pair(it->first + frame_offset,
 202                              it->second.substr(frame_offset)));
 203     frames_.erase(frames_.begin());
 204     num_bytes_consumed_ += frame_offset;
 205   }
 206   return num_bytes_consumed_ - initial_bytes_consumed;
 207 }
 208
 209 void QuicStreamSequencer::MarkConsumed(size_t num_bytes_consumed) {
 210   size_t end_offset = num_bytes_consumed_ + num_bytes_consumed;
 211   while (!frames_.empty() && end_offset != num_bytes_consumed_) {
 212     FrameMap::iterator it = frames_.begin();
 213     if (it->first != num_bytes_consumed_) {
 214       LOG(DFATAL) << "Invalid argument to MarkConsumed. "
 215                   << " num_bytes_consumed_: " << num_bytes_consumed_
 216                   << " end_offset: " << end_offset
 217                   << " offset: " << it->first
 218                   << " length: " << it->second.length();
 219       stream_->Close(QUIC_ERROR_PROCESSING_STREAM);
 220       return;
 221     }
 222
 223     if (it->first + it->second.length() <= end_offset) {
 224       num_bytes_consumed_ += it->second.length();
 225       // This chunk is entirely consumed.
 226       frames_.erase(it);
 227       continue;
 228     }
 229
 230     // Partially consume this frame.
 231     size_t delta = end_offset - it->first;
 232     num_bytes_consumed_ += delta;
 233     frames_.insert(make_pair(end_offset, it->second.substr(delta)));
 234     frames_.erase(it);
 235     break;
 236   }
 237 }
 238
 239 bool QuicStreamSequencer::HasBytesToRead() const {
 240   FrameMap::const_iterator it = frames_.begin();
 241
 242   return it != frames_.end() && it->first == num_bytes_consumed_;
 243 }
 244
 245 bool QuicStreamSequencer::IsHalfClosed() const {
 246   return num_bytes_consumed_ >= close_offset_;
 247 }
 248
 249 bool QuicStreamSequencer::IsDuplicate(const QuicStreamFrame& frame) const {
 250   // A frame is duplicate if the frame offset is smaller than our bytes consumed
 251   // or we have stored the frame in our map.
 252   // TODO(pwestin): Is it possible that a new frame contain more data even if
 253   // the offset is the same?
 254   return frame.offset < num_bytes_consumed_ ||
 255       frames_.find(frame.offset) != frames_.end();
 256 }
 257
 258 void QuicStreamSequencer::FlushBufferedFrames() {
 259   FrameMap::iterator it = frames_.find(num_bytes_consumed_);
 260   while (it != frames_.end()) {
 261     DVLOG(1) << "Flushing buffered packet at offset " << it->first;
 262     string* data = &it->second;
 263     size_t bytes_consumed = stream_->ProcessRawData(data->c_str(),
 264                                                     data->size());
 265     num_bytes_consumed_ += bytes_consumed;
 266     if (MaybeCloseStream()) {
 267       return;
 268     }
 269     if (bytes_consumed > data->size()) {
 270       stream_->Close(QUIC_ERROR_PROCESSING_STREAM);  // Programming error
 271       return;
 272     } else if (bytes_consumed == data->size()) {
 273       frames_.erase(it);
 274       it = frames_.find(num_bytes_consumed_);
 275     } else {
 276       string new_data = it->second.substr(bytes_consumed);
 277       frames_.erase(it);
 278       frames_.insert(make_pair(num_bytes_consumed_, new_data));
 279       return;
 280     }
 281   }
 282 }
 283
 284 }  // namespace net