Modify the SPDY stream to be buffered....

net/spdy/spdy_stream.cc

 // Copyright (c) 2010 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/spdy/spdy_stream.h"
 
 #include "base/logging.h"
+#include "base/message_loop.h"
 #include "net/http/http_request_info.h"
 #include "net/http/http_response_info.h"
 #include "net/spdy/spdy_session.h"
 
 namespace net {
 
 SpdyStream::SpdyStream(SpdySession* session, spdy::SpdyStreamId stream_id,
                        bool pushed, LoadLog* log)
     : stream_id_(stream_id),
       priority_(0),
       pushed_(pushed),
       download_finished_(false),
       metrics_(Singleton<BandwidthMetrics>::get()),
       session_(session),
       response_(NULL),
       request_body_stream_(NULL),
       response_complete_(false),
       io_state_(STATE_NONE),
       response_status_(OK),
       user_callback_(NULL),
       user_buffer_(NULL),
       user_buffer_len_(0),
       cancelled_(false),
       load_log_(log),
       send_bytes_(0),
       recv_bytes_(0),
-      histograms_recorded_(false) {}
+      histograms_recorded_(false),
+      buffered_read_callback_pending_(false),
+      more_read_data_pending_(false) {}
 
 SpdyStream::~SpdyStream() {
   DLOG(INFO) << "Deleting SpdyStream for stream " << stream_id_;
 
   // TODO(willchan): We're still calling CancelStream() too many times, because
   // inactive pending/pushed streams will still have stream_id_ set.
   if (stream_id_) {
     session_->CancelStream(stream_id_);
   } else if (!response_complete_) {
     NOTREACHED();
@@ skipping 149 matching lines @@
 
   if (user_callback_)
     DoCallback(rv);
 }
 
 bool SpdyStream::OnDataReceived(const char* data, int length) {
   DCHECK_GE(length, 0);
   LOG(INFO) << "SpdyStream: Data (" << length << " bytes) received for "
             << stream_id_;
 
+  CHECK(!response_complete_);
+
   // If we don't have a response, then the SYN_REPLY did not come through.
   // We cannot pass data up to the caller unless the reply headers have been
   // received.
   if (!response_->headers) {
     OnClose(ERR_SYN_REPLY_NOT_RECEIVED);
     return false;
   }
 
-  if (length > 0)
-    recv_bytes_ += length;
-  recv_last_byte_time_ = base::TimeTicks::Now();
-
   // A zero-length read means that the stream is being closed.
   if (!length) {
     metrics_.StopStream();
     download_finished_ = true;
+    response_complete_ = true;
+
+    // We need to complete any pending buffered read now.
+    DoBufferedReadCallback();
+
     OnClose(net::OK);
     return true;
   }
 
   // Track our bandwidth.
   metrics_.RecordBytes(length);
+  recv_bytes_ += length;
+  recv_last_byte_time_ = base::TimeTicks::Now();
 
-  if (length > 0) {
-    // TODO(mbelshe): If read is pending, we should copy the data straight into
-    // the read buffer here.  For now, we'll queue it always.
-    // TODO(mbelshe): We need to have some throttling on this.  We shouldn't
-    //                buffer an infinite amount of data.
-
-    IOBufferWithSize* io_buffer = new IOBufferWithSize(length);
-    memcpy(io_buffer->data(), data, length);
-
-    response_body_.push_back(io_buffer);
-  }
+  // Save the received data.
+  IOBufferWithSize* io_buffer = new IOBufferWithSize(length);
+  memcpy(io_buffer->data(), data, length);
+  response_body_.push_back(io_buffer);
 
   // Note that data may be received for a SpdyStream prior to the user calling

[wtc, 2010/02/25 20:04:49]

Please update this comment.  It refers to the test for
user_callback_, which is no longer here.

[Mike Belshe, 2010/02/25 23:30:17]

Done.

   // ReadResponseBody(), therefore user_callback_ may be NULL.  This may often
   // happen for server initiated streams.
-  if (user_callback_) {
-    int rv;
-    if (user_buffer_) {
-      rv = ReadResponseBody(user_buffer_, user_buffer_len_, user_callback_);
-      CHECK(rv != ERR_IO_PENDING);
-      user_buffer_ = NULL;
-      user_buffer_len_ = 0;
-    } else {
-      rv = OK;
-    }
-    DoCallback(rv);
-  }
+  if (user_buffer_)
+    ScheduleBufferedReadCallback();
 
   return true;
 }
 
 void SpdyStream::OnWriteComplete(int status) {
   // TODO(mbelshe): Check for cancellation here.  If we're cancelled, we
   // should discontinue the DoLoop.
 
   if (status > 0)
     send_bytes_ += status;
@@ skipping 71 matching lines @@
         break;
       default:
         NOTREACHED();
         break;
     }
   } while (result != ERR_IO_PENDING && io_state_ != STATE_NONE);
 
   return result;
 }
 
+void SpdyStream::ScheduleBufferedReadCallback() {
+  // If there is already a scheduled DoBufferedReadCallback, don't issue
+  // another one.  Mark that we have received more data and return.
+  if (buffered_read_callback_pending_) {
+    more_read_data_pending_ = true;
+    return;
+  }
+
+  more_read_data_pending_ = false;
+  buffered_read_callback_pending_ = true;
+  const int kBufferTimeMs = 1;
+  MessageLoop::current()->PostDelayedTask(FROM_HERE, NewRunnableMethod(
+    this, &SpdyStream::DoBufferedReadCallback), kBufferTimeMs);
+}
+
+// Checks to see if we should wait for more buffered data before notifying
+// the caller.  Returns true if we should wait, false otherwise.
+bool SpdyStream::ShouldWaitForMoreBufferedData() {

[willchan no longer on Chromium, 2010/02/25 20:47:35]

I think this member function can be const

[Mike Belshe, 2010/02/25 23:30:17]

Done.

+  // If the response is complete, there is no point in waiting.
+  if (response_complete_)
+    return false;
+
+  int bytes_buffered = 0;
+  std::list<scoped_refptr<IOBufferWithSize> >::iterator it;
+  for (it = response_body_.begin();
+       it != response_body_.end() && bytes_buffered < user_buffer_len_;
+       ++it)
+    bytes_buffered += (*it)->size();
+
+  return bytes_buffered < user_buffer_len_;
+}
+
+void SpdyStream::DoBufferedReadCallback() {
+  buffered_read_callback_pending_ = false;
+
+  // When more_read_data_pending_ is true, it means that more data has
+  // arrived since we started waiting.  Wait a little longer and continue
+  // to buffer.
+  if (more_read_data_pending_ && ShouldWaitForMoreBufferedData()) {
+    ScheduleBufferedReadCallback();
+    return;
+  }
+
+  int rv = 0;
+  if (user_buffer_) {
+    rv = ReadResponseBody(user_buffer_, user_buffer_len_, user_callback_);
+    CHECK(rv != ERR_IO_PENDING);
+    user_buffer_ = NULL;
+    user_buffer_len_ = 0;
+  }
+
+  if (user_callback_)
+    DoCallback(rv);
+}
+
 void SpdyStream::DoCallback(int rv) {
   CHECK(rv != ERR_IO_PENDING);
   CHECK(user_callback_);
 
   // Since Run may result in being called back, clear user_callback_ in advance.
   CompletionCallback* c = user_callback_;
   user_callback_ = NULL;
   c->Run(rv);
 }
 
@@ skipping 92 matching lines @@
   UMA_HISTOGRAM_TIMES("Net.SpdyStreamDownloadTime",
       recv_last_byte_time_ - recv_first_byte_time_);
   UMA_HISTOGRAM_TIMES("Net.SpdyStreamTime",
       recv_last_byte_time_ - send_time_);
 
   UMA_HISTOGRAM_COUNTS("Net.SpdySendBytes", send_bytes_);
   UMA_HISTOGRAM_COUNTS("Net.SpdyRecvBytes", recv_bytes_);
 }
 
 }  // namespace net