Handling partial writes and fin-only writes in QUIC.

This handles an API problem for sending a fin with no associated data.

This can happen at the GFE if we're dechunking HTTP, have
sent all bytes read from a backend, and then we get a 0 byte terminal chunk.

In this case, we have no way of telling when we write data if the fin was
consumed, or if the write was blocked at an underlying layer and the end of
data would otherwise never be communicated to the peer.

Merge internal change: 40340063


Review URL: https://chromiumcodereview.appspot.com/11820003

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@176287 0039d316-1c4b-4281-b951-d872f2087c98

11 files changed, 94 insertions, 55 deletions

diff --git a/net/quic/quic_connection.cc b/net/quic/quic_connection.cc
index 28f8473..de5f7cd 100644
--- a/net/quic/quic_connection.cc
+++ b/net/quic/quic_connection.cc
@@ -394,13 +394,14 @@ void QuicConnection::OnPacketComplete() {
   }
 }
 
-size_t QuicConnection::SendStreamData(
+QuicConsumedData QuicConnection::SendStreamData(
     QuicStreamId id,
     StringPiece data,
     QuicStreamOffset offset,
     bool fin,
     QuicPacketSequenceNumber* last_packet) {
-  int total_bytes_consumed = 0;
+  size_t total_bytes_consumed = 0;
+  bool fin_consumed = false;
 
   while (queued_packets_.empty()) {
     vector<PacketPair> packets;
@@ -408,6 +409,7 @@ size_t QuicConnection::SendStreamData(
         packet_creator_.DataToStream(id, data, offset, fin, &packets);
     total_bytes_consumed += bytes_consumed;
     offset += bytes_consumed;
+    fin_consumed = fin && bytes_consumed == data.size();
     data.remove_prefix(bytes_consumed);
     DCHECK_LT(0u, packets.size());
 
@@ -432,7 +434,7 @@ size_t QuicConnection::SendStreamData(
       break;
     }
   }
-  return total_bytes_consumed;
+  return QuicConsumedData(total_bytes_consumed, fin_consumed);
 }
 
 void QuicConnection::SendRstStream(QuicStreamId id,
diff --git a/net/quic/quic_connection.h b/net/quic/quic_connection.h
index 467640c..382f729 100644
--- a/net/quic/quic_connection.h
+++ b/net/quic/quic_connection.h
@@ -127,13 +127,17 @@ class NET_EXPORT_PRIVATE QuicConnection : public QuicFramerVisitorInterface {
                  QuicConnectionHelperInterface* helper);
   virtual ~QuicConnection();
 
+
   // Send the data payload to the peer.
-  // TODO(wtc): document the return value.
-  size_t SendStreamData(QuicStreamId id,
-                        base::StringPiece data,
-                        QuicStreamOffset offset,
-                        bool fin,
-                        QuicPacketSequenceNumber* last_packet);
+  // Returns a pair with the number of bytes consumed from data, and a boolean
+  // indicating if the fin bit was consumed.  This does not indicate the data
+  // has been sent on the wire: it may have been turned into a packet and queued
+  // if the socket was unexpectedly blocked.
+  QuicConsumedData SendStreamData(QuicStreamId id,
+                                  base::StringPiece data,
+                                  QuicStreamOffset offset,
+                                  bool fin,
+                                  QuicPacketSequenceNumber* last_packet);
   // Send a stream reset frame to the peer.
   virtual void SendRstStream(QuicStreamId id,
                              QuicErrorCode error,
diff --git a/net/quic/quic_connection_test.cc b/net/quic/quic_connection_test.cc
index 728fbb92..00c7918 100644
--- a/net/quic/quic_connection_test.cc
+++ b/net/quic/quic_connection_test.cc
@@ -1009,8 +1009,8 @@ TEST_F(QuicConnectionTest, TestQueueLimitsOnSendStreamData) {
   // Queue the first packet.
   EXPECT_CALL(*scheduler_, TimeUntilSend(false)).WillOnce(testing::Return(
       QuicTime::Delta::FromMicroseconds(10)));
-  EXPECT_EQ(6u,
-            connection_.SendStreamData(1, "EnoughDataToQueue", 0, false, NULL));
+  EXPECT_EQ(6u, connection_.SendStreamData(
+      1, "EnoughDataToQueue", 0, false, NULL).bytes_consumed);
   EXPECT_EQ(6u, connection_.NumQueuedPackets());
 }
 
@@ -1022,8 +1022,8 @@ TEST_F(QuicConnectionTest, LoopThroughSendingPackets) {
 
   // Queue the first packet.
   EXPECT_CALL(*scheduler_, SentPacket(_, _, _)).Times(17);
-  EXPECT_EQ(17u,
-            connection_.SendStreamData(1, "EnoughDataToQueue", 0, false, NULL));
+  EXPECT_EQ(17u, connection_.SendStreamData(
+                1, "EnoughDataToQueue", 0, false, NULL).bytes_consumed);
 }
 
 }  // namespace
diff --git a/net/quic/quic_http_stream.cc b/net/quic/quic_http_stream.cc
index ca98073..7354bc4 100644
--- a/net/quic/quic_http_stream.cc
+++ b/net/quic/quic_http_stream.cc
@@ -347,8 +347,8 @@ int QuicHttpStream::DoSendHeaders() {
   bool has_upload_data = request_body_stream_ != NULL;
 
   io_state_ = STATE_SEND_HEADERS_COMPLETE;
-  int rv = stream_->WriteData(request_, !has_upload_data);
-  return rv;
+  QuicConsumedData rv = stream_->WriteData(request_, !has_upload_data);
+  return rv.bytes_consumed;
 }
 
 int QuicHttpStream::DoSendHeadersComplete(int rv) {
@@ -398,14 +398,13 @@ int QuicHttpStream::DoSendBody() {
   int len = request_body_buf_->BytesRemaining();
   if (len > 0 || eof) {
     base::StringPiece data(request_body_buf_->data(), len);
-    int rv = stream_->WriteData(data, eof);
-    request_body_buf_->DidConsume(rv);
-    DCHECK_NE(ERR_IO_PENDING, rv);
+    QuicConsumedData rv = stream_->WriteData(data, eof);
+    request_body_buf_->DidConsume(rv.bytes_consumed);
     if (eof) {
       io_state_ = STATE_OPEN;
       return OK;
     }
-    return rv;
+    return rv.bytes_consumed;
   }
 
   io_state_ = STATE_SEND_BODY_COMPLETE;
diff --git a/net/quic/quic_protocol.h b/net/quic/quic_protocol.h
index 471b797..9ca2502 100644
--- a/net/quic/quic_protocol.h
+++ b/net/quic/quic_protocol.h
@@ -34,6 +34,19 @@ typedef uint64 QuicStreamOffset;
 typedef uint64 QuicPacketSequenceNumber;
 typedef uint8 QuicFecGroupNumber;
 
+// A struct for functions which consume data payloads and fins.
+// The first member of the pair indicates bytes consumed.
+// The second member of the pair indicates if an incoming fin was consumed.
+struct QuicConsumedData {
+  QuicConsumedData(size_t bytes_consumed, bool fin_consumed)
+      : bytes_consumed(bytes_consumed),
+        fin_consumed(fin_consumed) {
+  }
+  size_t bytes_consumed;
+  bool fin_consumed;
+};
+
+
 // TODO(rch): Consider Quic specific names for these constants.
 const size_t kMaxPacketSize = 1200;  // Maximum size in bytes of a QUIC packet.
 
diff --git a/net/quic/quic_session.cc b/net/quic/quic_session.cc
index 02d2a6c..49490f6 100644
--- a/net/quic/quic_session.cc
+++ b/net/quic/quic_session.cc
@@ -145,8 +145,10 @@ bool QuicSession::OnCanWrite() {
   return write_blocked_streams_.empty();
 }
 
-int QuicSession::WriteData(QuicStreamId id, StringPiece data,
-                           QuicStreamOffset offset, bool fin) {
+QuicConsumedData QuicSession::WriteData(QuicStreamId id,
+                                        StringPiece data,
+                                        QuicStreamOffset offset,
+                                        bool fin) {
   // TODO(wtc): type mismatch -- connection_->SendStreamData() returns a
   // size_t.
   return connection_->SendStreamData(id, data, offset, fin, NULL);
diff --git a/net/quic/quic_session.h b/net/quic/quic_session.h
index 2bc06d2..9d9f848 100644
--- a/net/quic/quic_session.h
+++ b/net/quic/quic_session.h
@@ -46,10 +46,12 @@ class NET_EXPORT_PRIVATE QuicSession : public QuicConnectionVisitorInterface {
   virtual bool OnCanWrite() OVERRIDE;
 
   // Called by streams when they want to write data to the peer.
-  // TODO(wtc): document the return value -- whether it can be negative and
-  // how a failure is reported.
-  virtual int WriteData(QuicStreamId id, base::StringPiece data,
-                        QuicStreamOffset offset, bool fin);
+  // Returns a pair with the number of bytes consumed from data, and a boolean
+  // indicating if the fin bit was consumed.  This does not indicate the data
+  // has been sent on the wire: it may have been turned into a packet and queued
+  // if the socket was unexpectedly blocked.
+  virtual QuicConsumedData WriteData(QuicStreamId id, base::StringPiece data,
+                                     QuicStreamOffset offset, bool fin);
   // Called by streams when they want to close the stream in both directions.
   void SendRstStream(QuicStreamId id,
                      QuicErrorCode error,
diff --git a/net/quic/reliable_quic_stream.cc b/net/quic/reliable_quic_stream.cc
index 2aa477b..5c38cbd 100644
--- a/net/quic/reliable_quic_stream.cc
+++ b/net/quic/reliable_quic_stream.cc
@@ -99,24 +99,29 @@ const IPEndPoint& ReliableQuicStream::GetPeerAddress() const {
   return session_->peer_address();
 }
 
-int ReliableQuicStream::WriteData(StringPiece data, bool fin) {
+QuicConsumedData ReliableQuicStream::WriteData(StringPiece data, bool fin) {
   return WriteOrBuffer(data, fin);
 }
 
-int ReliableQuicStream::WriteOrBuffer(StringPiece data, bool fin) {
+QuicConsumedData ReliableQuicStream::WriteOrBuffer(StringPiece data, bool fin) {
   DCHECK(!fin_buffered_);
 
-  size_t bytes_written = 0;
+  QuicConsumedData consumed_data(0, false);
   fin_buffered_ = fin;
 
   if (queued_data_.empty()) {
-    bytes_written = WriteDataInternal(string(data.data(), data.length()), fin);
+    consumed_data = WriteDataInternal(string(data.data(), data.length()), fin);
   }
-  if (bytes_written != data.length()) {
-    queued_data_.push_back(string(data.data() + bytes_written,
-                                  data.length() - bytes_written));
+
+  // if there's unconsumed data or an unconsumed fin, queue it.
+  if (consumed_data.bytes_consumed != data.length() ||
+      (fin && !consumed_data.fin_consumed)) {
+    queued_data_.push_back(
+        string(data.data() + consumed_data.bytes_consumed,
+               data.length() - consumed_data.bytes_consumed));
   }
-  return data.size();
+
+  return QuicConsumedData(data.size(), true);
 }
 
 void ReliableQuicStream::OnCanWrite() {
@@ -126,35 +131,37 @@ void ReliableQuicStream::OnCanWrite() {
     if (queued_data_.size() == 1 && fin_buffered_) {
       fin = true;
     }
-    int bytes_written = WriteDataInternal(data, fin);
-    if (bytes_written == static_cast<int>(data.size())) {
+    QuicConsumedData consumed_data = WriteDataInternal(data, fin);
+    if (consumed_data.bytes_consumed == data.size() &&
+        fin == consumed_data.fin_consumed) {
       queued_data_.pop_front();
     } else {
-      queued_data_.front() = string(data.data() + bytes_written,
-                                    data.length() - bytes_written);
+      queued_data_.front().erase(0, consumed_data.bytes_consumed);
       break;
     }
   }
 }
 
-int ReliableQuicStream::WriteDataInternal(StringPiece data, bool fin) {
+QuicConsumedData ReliableQuicStream::WriteDataInternal(
+    StringPiece data, bool fin) {
   if (write_side_closed_) {
     DLOG(ERROR) << "Attempt to write when the write side is closed";
-    return 0;
+    return QuicConsumedData(0, false);
   }
 
-  int bytes_consumed = session()->WriteData(id(), data, offset_, fin);
-  offset_ += bytes_consumed;
-  stream_bytes_written_ += bytes_consumed;
-  if (bytes_consumed == static_cast<int>(data.length())) {
-    if (fin) {
+  QuicConsumedData consumed_data =
+      session()->WriteData(id(), data, offset_, fin);
+  offset_ += consumed_data.bytes_consumed;
+  stream_bytes_written_ += consumed_data.bytes_consumed;
+  if (consumed_data.bytes_consumed == data.length()) {
+    if (fin && consumed_data.fin_consumed) {
       fin_sent_ = true;
       CloseWriteSide();
     }
   } else {
     session_->MarkWriteBlocked(id());
   }
-  return bytes_consumed;
+  return consumed_data;
 }
 
 void ReliableQuicStream::CloseReadSide() {
diff --git a/net/quic/reliable_quic_stream.h b/net/quic/reliable_quic_stream.h
index c1d7406..50ed951 100644
--- a/net/quic/reliable_quic_stream.h
+++ b/net/quic/reliable_quic_stream.h
@@ -90,9 +90,15 @@ class NET_EXPORT_PRIVATE ReliableQuicStream {
   void set_visitor(Visitor* visitor) { visitor_ = visitor; }
 
  protected:
-  // TODO(alyssar): document the return value -- whether it can be negative and
-  // how a failure is reported.
-  virtual int WriteData(base::StringPiece data, bool fin);
+  // Returns a pair with the number of bytes consumed from data, and a boolean
+  // indicating if the fin bit was consumed.  This does not indicate the data
+  // has been sent on the wire: it may have been turned into a packet and queued
+  // if the socket was unexpectedly blocked.
+  //
+  // The default implementation always consumed all bytes and any fin, but
+  // this behavior is not guaranteed for subclasses so callers should check the
+  // return value.
+  virtual QuicConsumedData WriteData(base::StringPiece data, bool fin);
 
   // Close the read side of the socket.  Further frames will not be accepted.
   virtual void CloseReadSide();
@@ -104,13 +110,13 @@ class NET_EXPORT_PRIVATE ReliableQuicStream {
 
   // Sends as much of 'data' to the connection as the connection will consume,
   // and then buffers any remaining data in queued_data_.
-  // Returns the number of bytes consumed or buffered, which should always equal
-  // data.size()
-  int WriteOrBuffer(base::StringPiece data, bool fin);
+  // Returns (data.size(), true) as it always consumed all data: it returns for
+  // convenience to have the same return type as WriteDataInternal.
+  QuicConsumedData WriteOrBuffer(base::StringPiece data, bool fin);
 
   // Sends as much of 'data' to the connection as the connection will consume.
   // Returns the number of bytes consumed by the connection.
-  int WriteDataInternal(base::StringPiece data, bool fin);
+  QuicConsumedData WriteDataInternal(base::StringPiece data, bool fin);
 
  private:
   friend class test::ReliableQuicStreamPeer;
diff --git a/net/quic/test_tools/quic_test_utils.cc b/net/quic/test_tools/quic_test_utils.cc
index f6f4766..bb509f8 100644
--- a/net/quic/test_tools/quic_test_utils.cc
+++ b/net/quic/test_tools/quic_test_utils.cc
@@ -9,13 +9,14 @@
 using std::max;
 using std::min;
 using std::string;
+using testing::_;
 
 namespace net {
 namespace test {
 
 MockFramerVisitor::MockFramerVisitor() {
   // By default, we want to accept packets.
-  ON_CALL(*this, OnPacketHeader(testing::_))
+  ON_CALL(*this, OnPacketHeader(_))
       .WillByDefault(testing::Return(true));
 }
 
@@ -101,6 +102,8 @@ bool PacketSavingConnection::SendPacket(QuicPacketSequenceNumber number,
 
 MockSession::MockSession(QuicConnection* connection, bool is_server)
     : QuicSession(connection, is_server) {
+  ON_CALL(*this, WriteData(_, _, _, _))
+      .WillByDefault(testing::Return(QuicConsumedData(0, false)));
 }
 
 MockSession::~MockSession() {
diff --git a/net/quic/test_tools/quic_test_utils.h b/net/quic/test_tools/quic_test_utils.h
index 89d7103..544543a 100644
--- a/net/quic/test_tools/quic_test_utils.h
+++ b/net/quic/test_tools/quic_test_utils.h
@@ -199,8 +199,9 @@ class MockSession : public QuicSession {
   MOCK_METHOD0(CreateOutgoingReliableStream, ReliableQuicStream*());
   MOCK_METHOD3(WriteData,
                void(QuicStreamId id, base::StringPiece data, bool fin));
-  MOCK_METHOD4(WriteData, int(QuicStreamId id, base::StringPiece data,
-                              QuicStreamOffset offset, bool fin));
+  MOCK_METHOD4(WriteData, QuicConsumedData(QuicStreamId id,
+                                           base::StringPiece data,
+                                           QuicStreamOffset offset, bool fin));
   MOCK_METHOD0(IsHandshakeComplete, bool());
 
  private: