Land Recent QUIC changes.

net/quic/reliable_quic_stream.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/reliable_quic_stream.h"
 
 #include "net/quic/quic_session.h"
 #include "net/quic/quic_spdy_decompressor.h"
 
 using base::StringPiece;
 using std::min;
 
 namespace net {
 
 ReliableQuicStream::ReliableQuicStream(QuicStreamId id,
                                        QuicSession* session)
     : sequencer_(this),
       id_(id),
       session_(session),
       visitor_(NULL),
       stream_bytes_read_(0),
       stream_bytes_written_(0),
       headers_decompressed_(false),
       headers_id_(0),
+      decompression_failed_(false),
       stream_error_(QUIC_STREAM_NO_ERROR),
       connection_error_(QUIC_NO_ERROR),
       read_side_closed_(false),
       write_side_closed_(false),
       fin_buffered_(false),
       fin_sent_(false) {
 }
 
 ReliableQuicStream::~ReliableQuicStream() {
 }
@@ skipping 15 matching lines @@
   if (read_side_closed_) {
     DLOG(INFO) << "Ignoring frame " << frame.stream_id;
     // We don't want to be reading: blackhole the data.
     return true;
   }
   // Note: This count include duplicate data received.
   stream_bytes_read_ += frame.data.length();
 
   bool accepted = sequencer_.OnStreamFrame(frame);
 
-  if (frame.fin) {
-    sequencer_.CloseStreamAtOffset(frame.offset + frame.data.size());
-  }
-
   return accepted;
 }
 
 void ReliableQuicStream::OnStreamReset(QuicRstStreamErrorCode error) {
   stream_error_ = error;
   TerminateFromPeer(false);  // Full close.
 }
 
 void ReliableQuicStream::ConnectionClose(QuicErrorCode error, bool from_peer) {
   if (read_side_closed_ && write_side_closed_) {
@@ skipping 186 matching lines @@
     }
     total_bytes_consumed += missing_size;
     StringPiece(data, missing_size).AppendToString(&headers_id_buffer_);
     DCHECK_EQ(4u, headers_id_buffer_.length());
     memcpy(&headers_id_, headers_id_buffer_.data(), 4);
     headers_id_buffer_.clear();
     data += missing_size;
     data_len -= missing_size;
   }
   DCHECK_NE(0u, headers_id_);
+  if (data_len == 0) {
+    return total_bytes_consumed;
+  }
 
   // Once the headers are finished, we simply pass the data through.
   if (headers_decompressed_) {
     // Some buffered header data remains.
     if (!decompressed_headers_.empty()) {
       ProcessHeaderData();
     }
-    if (decompressed_headers_.empty() && data_len > 0) {
+    if (decompressed_headers_.empty()) {
       DVLOG(1) << "Delegating procesing to ProcessData";
       total_bytes_consumed += ProcessData(data, data_len);
     }
     return total_bytes_consumed;
   }
 
   QuicHeaderId current_header_id =
       session_->decompressor()->current_header_id();
   // Ensure that this header id looks sane.
   if (headers_id_ < current_header_id ||
       headers_id_ > kMaxHeaderIdDelta + current_header_id) {
     DVLOG(1) << "Invalid headers for stream: " << id()
              << " header_id: " << headers_id_
              << " current_header_id: " << current_header_id;
     session_->connection()->SendConnectionClose(QUIC_INVALID_HEADER_ID);
     return total_bytes_consumed;
   }
 
   // If we are head-of-line blocked on decompression, then back up.
   if (current_header_id != headers_id_) {
     session_->MarkDecompressionBlocked(headers_id_, id());
     DVLOG(1) << "Unable to decompress header data for stream: " << id()
              << " header_id: " << headers_id_;
     return total_bytes_consumed;
   }
 
   // Decompressed data will be delivered to decompressed_headers_.
   size_t bytes_consumed = session_->decompressor()->DecompressData(
       StringPiece(data, data_len), this);
+  DCHECK_NE(0u, bytes_consumed);
+  if (bytes_consumed > data_len) {
+    DCHECK(false) << "DecompressData returned illegal value";
+    OnDecompressionError();
+    return total_bytes_consumed;
+  }
   total_bytes_consumed += bytes_consumed;
+  data += bytes_consumed;
+  data_len -= bytes_consumed;
+
+  if (decompression_failed_) {
+    // The session will have been closed in OnDecompressionError.
+    return total_bytes_consumed;
+  }
 
   // Headers are complete if the decompressor has moved on to the
   // next stream.
   headers_decompressed_ =
       session_->decompressor()->current_header_id() != headers_id_;
+  if (!headers_decompressed_) {
+    DCHECK_EQ(0u, data_len);
+  }
 
   ProcessHeaderData();
 
+  if (!headers_decompressed_ || !decompressed_headers_.empty()) {
+    return total_bytes_consumed;
+  }
+
   // We have processed all of the decompressed data but we might
   // have some more raw data to process.
-  if (decompressed_headers_.empty() && bytes_consumed < data_len) {
-    total_bytes_consumed += ProcessData(data + bytes_consumed,
-                                        data_len - bytes_consumed);
+  if (data_len > 0) {
+    total_bytes_consumed += ProcessData(data, data_len);
   }
 
   // The sequencer will push any additional buffered frames if this data
   // has been completely consumed.
   return total_bytes_consumed;
 }
 
 uint32 ReliableQuicStream::ProcessHeaderData() {
   if (decompressed_headers_.empty()) {
     return 0;
   }
 
   size_t bytes_processed = ProcessData(decompressed_headers_.data(),
                                        decompressed_headers_.length());
   if (bytes_processed == decompressed_headers_.length()) {
     decompressed_headers_.clear();
   } else {
     decompressed_headers_ = decompressed_headers_.erase(0, bytes_processed);
   }
   return bytes_processed;
 }
 
 void ReliableQuicStream::OnDecompressorAvailable() {
   DCHECK_EQ(headers_id_,
             session_->decompressor()->current_header_id());
   DCHECK(!headers_decompressed_);
+  DCHECK(!decompression_failed_);
   DCHECK_EQ(0u, decompressed_headers_.length());
 
   size_t total_bytes_consumed = 0;
   struct iovec iovecs[5];
   while (!headers_decompressed_) {
     size_t num_iovecs =
         sequencer_.GetReadableRegions(iovecs, arraysize(iovecs));
 
     if (num_iovecs == 0) {
       return;
     }
     for (size_t i = 0; i < num_iovecs && !headers_decompressed_; i++) {
       total_bytes_consumed += session_->decompressor()->DecompressData(
           StringPiece(static_cast<char*>(iovecs[i].iov_base),
                       iovecs[i].iov_len), this);
+      if (decompression_failed_) {
+        return;
+      }
 
       headers_decompressed_ =
           session_->decompressor()->current_header_id() != headers_id_;
     }
   }
 
   // Either the headers are complete, or the all data as been consumed.
   sequencer_.MarkConsumed(total_bytes_consumed);
   ProcessHeaderData();  // Unprocessed headers remain in decompressed_headers_.
   if (IsHalfClosed()) {
     TerminateFromPeer(true);
   } else if (headers_decompressed_ && decompressed_headers_.empty()) {
     sequencer_.FlushBufferedFrames();
   }
 }
 
 bool ReliableQuicStream::OnDecompressedData(StringPiece data) {
   data.AppendToString(&decompressed_headers_);
   return true;
 }
 
 void ReliableQuicStream::OnDecompressionError() {
+  DCHECK(!decompression_failed_);
+  decompression_failed_ = true;
   session_->connection()->SendConnectionClose(QUIC_DECOMPRESSION_FAILURE);
 }
 
 
 void ReliableQuicStream::CloseWriteSide() {
   if (write_side_closed_) {
     return;
   }
   DLOG(INFO) << "Done writing to stream " << id();
 
@@ skipping 11 matching lines @@
   if (visitor_) {
     Visitor* visitor = visitor_;
     // Calling Visitor::OnClose() may result the destruction of the visitor,
     // so we need to ensure we don't call it again.
     visitor_ = NULL;
     visitor->OnClose(this);
   }
 }
 
 }  // namespace net