path: root/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 "base/logging.h"
#include "base/metrics/sparse_histogram.h"
#include "net/quic/reliable_quic_stream.h"

using std::min;
using std::numeric_limits;

namespace net {

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) {
}

QuicStreamSequencer::~QuicStreamSequencer() {
}

bool QuicStreamSequencer::OnStreamFrame(const QuicStreamFrame& frame) {
  ++num_frames_received_;
  if (IsDuplicate(frame)) {
    ++num_duplicate_frames_received_;
    // Silently ignore duplicates.
    return true;
  }

  QuicStreamOffset byte_offset = frame.offset;
  size_t data_len = frame.data.TotalBufferSize();
  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 false;
  }

  if (frame.fin) {
    CloseStreamAtOffset(frame.offset + data_len);
    if (data_len == 0) {
      return true;
    }
  }

  IOVector data;
  data.AppendIovec(frame.data.iovec(), frame.data.Size());

  // If the frame has arrived in-order then we can process it immediately, only
  // buffering if the stream is unable to process it.
  if (!blocked_ && byte_offset == num_bytes_consumed_) {
    DVLOG(1) << "Processing byte offset " << byte_offset;
    size_t bytes_consumed = 0;
    for (size_t i = 0; i < data.Size(); ++i) {
      bytes_consumed += stream_->ProcessRawData(
          static_cast<char*>(data.iovec()[i].iov_base),
          data.iovec()[i].iov_len);
    }
    num_bytes_consumed_ += bytes_consumed;
    stream_->AddBytesConsumed(bytes_consumed);
    stream_->MaybeSendWindowUpdate();

    if (MaybeCloseStream()) {
      return true;
    }
    if (bytes_consumed > data_len) {
      stream_->Reset(QUIC_ERROR_PROCESSING_STREAM);
      return false;
    } else if (bytes_consumed == data_len) {
      FlushBufferedFrames();
      return true;  // it's safe to ack this frame.
    } else {
      // Set ourselves up to buffer what's left.
      data_len -= bytes_consumed;
      data.Consume(bytes_consumed);
      byte_offset += bytes_consumed;
    }
  }

  // Buffer any remaining data to be consumed by the stream when ready.
  for (size_t i = 0; i < data.Size(); ++i) {
    DVLOG(1) << "Buffering stream data at offset " << byte_offset;
    const iovec& iov = data.iovec()[i];
    frames_.insert(make_pair(
        byte_offset, string(static_cast<char*>(iov.iov_base), iov.iov_len)));
    byte_offset += iov.iov_len;
    num_bytes_buffered_ += iov.iov_len;
    stream_->AddBytesBuffered(iov.iov_len);
  }
  return true;
}

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 (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.";
    // Technically it's an error if num_bytes_consumed isn't exactly
    // equal, but error handling seems silly at this point.
    stream_->OnFinRead();
    frames_.clear();
    num_bytes_buffered_ = 0;
    return true;
  }
  return false;
}

int QuicStreamSequencer::GetReadableRegions(iovec* iov, size_t iov_len) {
  DCHECK(!blocked_);
  FrameMap::iterator it = frames_.begin();
  size_t index = 0;
  QuicStreamOffset offset = num_bytes_consumed_;
  while (it != frames_.end() && index < iov_len) {
    if (it->first != 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();

    ++index;
    ++it;
  }
  return index;
}

int QuicStreamSequencer::Readv(const struct iovec* iov, size_t iov_len) {
  DCHECK(!blocked_);
  FrameMap::iterator it = frames_.begin();
  size_t iov_index = 0;
  size_t iov_offset = 0;
  size_t frame_offset = 0;
  size_t initial_bytes_consumed = num_bytes_consumed_;

  while (iov_index < iov_len &&
         it != frames_.end() &&
         it->first == num_bytes_consumed_) {
    int bytes_to_read = min(iov[iov_index].iov_len - iov_offset,
                            it->second.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);
    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->second.size() == frame_offset) {
      // We've copied this whole frame
      RecordBytesConsumed(it->second.size());
      frames_.erase(it);
      it = frames_.begin();
      frame_offset = 0;
    }
  }
  // We've finished copying.  If we have a partial frame, update it.
  if (frame_offset != 0) {
    frames_.insert(make_pair(it->first + frame_offset,
                             it->second.substr(frame_offset)));
    frames_.erase(frames_.begin());
    RecordBytesConsumed(frame_offset);
  }
  return num_bytes_consumed_ - initial_bytes_consumed;
}

bool QuicStreamSequencer::HasBytesToRead() const {
  FrameMap::const_iterator it = frames_.begin();

  return it != frames_.end() && it->first == num_bytes_consumed_;
}

bool QuicStreamSequencer::IsClosed() const {
  return num_bytes_consumed_ >= close_offset_;
}

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?
  return frame.offset < num_bytes_consumed_ ||
      frames_.find(frame.offset) != frames_.end();
}

void QuicStreamSequencer::SetBlockedUntilFlush() {
  blocked_ = true;
}

void QuicStreamSequencer::FlushBufferedFrames() {
  blocked_ = false;
  FrameMap::iterator it = frames_.find(num_bytes_consumed_);
  while (it != frames_.end()) {
    DVLOG(1) << "Flushing buffered packet at offset " << it->first;
    string* data = &it->second;
    size_t bytes_consumed = stream_->ProcessRawData(data->c_str(),
                                                    data->size());
    RecordBytesConsumed(bytes_consumed);
    if (MaybeCloseStream()) {
      return;
    }
    if (bytes_consumed > data->size()) {
      stream_->Reset(QUIC_ERROR_PROCESSING_STREAM);  // Programming error
      return;
    } else if (bytes_consumed == data->size()) {
      frames_.erase(it);
      it = frames_.find(num_bytes_consumed_);
    } else {
      string new_data = it->second.substr(bytes_consumed);
      frames_.erase(it);
      frames_.insert(make_pair(num_bytes_consumed_, new_data));
      return;
    }
  }
  MaybeCloseStream();
}

void QuicStreamSequencer::RecordBytesConsumed(size_t bytes_consumed) {
  num_bytes_consumed_ += bytes_consumed;
  num_bytes_buffered_ -= bytes_consumed;

  stream_->AddBytesConsumed(bytes_consumed);
  stream_->RemoveBytesBuffered(bytes_consumed);
  stream_->MaybeSendWindowUpdate();
}

}  // namespace net