SPDY - implement greedy approach to read all the data and process it

net/spdy/spdy_session.cc

Index: net/spdy/spdy_session.cc
===================================================================
--- net/spdy/spdy_session.cc	(revision 180807)
+++ net/spdy/spdy_session.cc	(working copy)
@@ -240,7 +240,7 @@
       is_secure_(false),
       certificate_error_code_(OK),
       error_(OK),
-      state_(IDLE),
+      state_(STATE_IDLE),
       max_concurrent_streams_(initial_max_concurrent_streams == 0 ?
                               kInitialMaxConcurrentStreams :
                               initial_max_concurrent_streams),
@@ -251,7 +251,8 @@
       streams_pushed_count_(0),
       streams_pushed_and_claimed_count_(0),
       streams_abandoned_count_(0),
-      bytes_received_(0),
+      total_bytes_received_(0),
+      bytes_read_(0),

[Ryan Sleevi, 2013/02/06 23:04:46]

I'm still not sure I understand why we need this, since we could use a stack
variable. Yes, 4-8 bytes on the heap is probably cheap enough not to worry
about, but it looks so out of place for me, since it's only needed in the
context of a given iteration of DoLoop.

[ramant (doing other things), 2013/02/07 02:11:07]

As we had gone over the code, bytes_read_ is used in DoLoop (initializes to 0), 
DoRead (posts task if it goes over kMaxReadBytes) and DoReadComplete (keeps
increasing it as it reads bytes).

As we have discussed it, will leave this as is.

       sent_settings_(false),
       received_settings_(false),
       stalled_streams_(0),
@@ -311,8 +312,8 @@
 SpdySession::CallbackResultPair::~CallbackResultPair() {}
 
 SpdySession::~SpdySession() {
-  if (state_ != CLOSED) {
-    state_ = CLOSED;
+  if (state_ != STATE_CLOSED) {
+    state_ = STATE_CLOSED;
 
     // Cleanup all the streams.
     CloseAllStreams(net::ERR_ABORTED);
@@ -341,7 +342,7 @@
   base::StatsCounter spdy_sessions("spdy.sessions");
   spdy_sessions.Increment();
 
-  state_ = CONNECTED;
+  state_ = STATE_DO_READ;
   connection_.reset(connection);
   is_secure_ = is_secure;
   certificate_error_code_ = certificate_error_code;
@@ -373,17 +374,17 @@
 
   // Write out any data that we might have to send, such as the settings frame.
   WriteSocketLater();
-  net::Error error = ReadSocket();
+  int error = DoLoop(OK);
   if (error == ERR_IO_PENDING)
     return OK;
-  return error;
+  return static_cast<net::Error>(error);
 }
 
 bool SpdySession::VerifyDomainAuthentication(const std::string& domain) {
   if (!verify_domain_authentication_)
     return true;
 
-  if (state_ != CONNECTED)
+  if (!IsConnected())
     return false;
 
   SSLInfo ssl_info;
@@ -411,7 +412,7 @@
     const GURL& url,
     scoped_refptr<SpdyStream>* stream,
     const BoundNetLog& stream_net_log) {
-  CHECK_NE(state_, CLOSED);
+  CHECK_NE(state_, STATE_CLOSED);
 
   *stream = NULL;
 
@@ -771,7 +772,7 @@
 
   // If we're connecting, defer to the connection to give us the actual
   // LoadState.
-  if (state_ == CONNECTING)
+  if (state_ == STATE_CONNECTING)
     return connection_->GetLoadState();
 
   // Just report that we're idle since the session could be doing
@@ -780,11 +781,72 @@
 }
 
 void SpdySession::OnReadComplete(int bytes_read) {
+  DCHECK_NE(state_, STATE_DO_READ);
+  read_pending_ = false;
+  DoLoop(bytes_read);
+}
+
+void SpdySession::StartRead() {
+  DCHECK_NE(state_, STATE_DO_READ_COMPLETE);
+  read_pending_ = false;
+  DoLoop(OK);
+}
+
+int SpdySession::DoLoop(int result) {
+  bytes_read_ = 0;
+  do {
+    if (read_pending_)
+      return OK;
+
+    switch (state_) {
+      case STATE_DO_READ:
+        DCHECK_EQ(result, OK);
+        result = DoRead();
+        break;
+      case STATE_DO_READ_COMPLETE:
+        result = DoReadComplete(result);
+        break;
+      case STATE_CLOSED:
+        result = ERR_CONNECTION_CLOSED;
+        break;
+      default:
+        NOTREACHED() << "state_: " << state_;
+        break;
+    }
+  } while (result != ERR_IO_PENDING && result != ERR_CONNECTION_CLOSED);
+
+  return result;
+}
+
+int SpdySession::DoRead() {
+  DCHECK(!read_pending_);
+  if (bytes_read_ > kMaxReadBytes) {
+    state_ = STATE_DO_READ;
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        base::Bind(&SpdySession::StartRead,
+                   weak_factory_.GetWeakPtr()));
+    return ERR_IO_PENDING;
+  }
+
+  CHECK(connection_.get());
+  CHECK(connection_->socket());
+  state_ = STATE_DO_READ_COMPLETE;
+  int result = connection_->socket()->Read(
+      read_buffer_.get(),
+      kReadBufferSize,
+      base::Bind(&SpdySession::OnReadComplete, base::Unretained(this)));
+  if (result == net::ERR_IO_PENDING)
+    read_pending_ = true;

[Ryan Sleevi, 2013/02/06 23:04:46]

Why is this necessary? Returning result here should cause the loop to terminate,
as per line 816.

[ramant (doing other things), 2013/02/07 02:11:07]

Deleted read_pending_. This variable seems to indicate if there is a pending
read or not. It seems to have been added to prevent recursion previously.

IMO, we don't need this.

Whenever we call DoLoop, we are initializing read_pending_ to false.

What do you think of this change? Do you see any problems with that change?

+  return result;
+}
+
+int SpdySession::DoReadComplete(int bytes_read) {
   // Parse a frame.  For now this code requires that the frame fit into our
   // buffer (32KB).
   // TODO(mbelshe): support arbitrarily large frames!
 
-  read_pending_ = false;
+  DCHECK(!read_pending_);

[Ryan Sleevi, 2013/02/06 23:04:46]

This seems wrong - you set read_pending_ to true on line 840 on ERR_IO_PENDING,
with the callback bound to this function (line 838).

This means that if connection_->socket()->Read() were to return ERR_IO_PENDING,
and then call your callback, you should end up crashing.

[ramant (doing other things), 2013/02/07 02:11:07]

SpdySession::OnReadComplete initializes read_pending_ to false and calls DoLoop
which calls DoReadComplete.

As we had talked, will leave DCHECK as is.

 
   if (bytes_read <= 0) {
     // Session is tearing down.
@@ -792,10 +854,11 @@
     if (bytes_read == 0)
       error = ERR_CONNECTION_CLOSED;
     CloseSessionOnError(error, true, "bytes_read is <= 0.");
-    return;
+    return ERR_CONNECTION_CLOSED;
   }
 
-  bytes_received_ += bytes_read;
+  total_bytes_received_ += bytes_read;
+  bytes_read_ += bytes_read;

[Ryan Sleevi, 2013/02/06 23:04:46]

naming: having bytes_read_ (instance var) and bytes_read (scoped var) is very
subtle and confusing.

[ramant (doing other things), 2013/02/07 02:11:07]

Changed "bytes_read" to "result"

Done.

 
   last_activity_time_ = base::TimeTicks::Now();
 
@@ -818,8 +881,9 @@
       buffered_spdy_framer_->Reset();
   }
 
-  if (state_ != CLOSED)
-    ReadSocket();
+  if (IsConnected())
+    state_ = STATE_DO_READ;
+  return OK;
 }
 
 void SpdySession::OnWriteComplete(int result) {
@@ -874,49 +938,11 @@
   }
 }
 
-net::Error SpdySession::ReadSocket() {
-  if (read_pending_)
-    return OK;
-
-  if (state_ == CLOSED) {
-    NOTREACHED();
-    return ERR_UNEXPECTED;
-  }
-
-  CHECK(connection_.get());
-  CHECK(connection_->socket());
-  int bytes_read = connection_->socket()->Read(
-      read_buffer_.get(),
-      kReadBufferSize,
-      base::Bind(&SpdySession::OnReadComplete, base::Unretained(this)));
-  switch (bytes_read) {
-    case 0:
-      // Socket is closed!
-      CloseSessionOnError(ERR_CONNECTION_CLOSED, true, "bytes_read is 0.");
-      return ERR_CONNECTION_CLOSED;
-    case net::ERR_IO_PENDING:
-      // Waiting for data.  Nothing to do now.
-      read_pending_ = true;
-      return ERR_IO_PENDING;
-    default:
-      // Data was read, process it.
-      // Schedule the work through the message loop to avoid recursive
-      // callbacks.
-      read_pending_ = true;
-      MessageLoop::current()->PostTask(
-          FROM_HERE,
-          base::Bind(&SpdySession::OnReadComplete,
-                     weak_factory_.GetWeakPtr(), bytes_read));
-      break;
-  }
-  return OK;
-}
-
 void SpdySession::WriteSocketLater() {
   if (delayed_write_pending_)
     return;
 
-  if (state_ < CONNECTED)
+  if (!IsConnected())
     return;
 
   delayed_write_pending_ = true;
@@ -933,7 +959,7 @@
   // If the socket isn't connected yet, just wait; we'll get called
   // again when the socket connection completes.  If the socket is
   // closed, just return.
-  if (state_ < CONNECTED || state_ == CLOSED)
+  if (!IsConnected())
     return;
 
   if (write_pending_)   // Another write is in progress still.
@@ -1056,8 +1082,8 @@
   // Don't close twice.  This can occur because we can have both
   // a read and a write outstanding, and each can complete with
   // an error.
-  if (state_ != CLOSED) {
-    state_ = CLOSED;
+  if (!IsClosed()) {
+    state_ = STATE_CLOSED;
     error_ = err;
     if (remove_from_pool)
       RemoveFromPool();
@@ -1921,16 +1947,16 @@
           // for larger volumes of data being sent.
           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd",
                                       val, 1, 200, 100);
-          if (bytes_received_ > 10 * 1024) {
+          if (total_bytes_received_ > 10 * 1024) {
             UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd10K",
                                         val, 1, 200, 100);
-            if (bytes_received_ > 25 * 1024) {
+            if (total_bytes_received_ > 25 * 1024) {
               UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd25K",
                                           val, 1, 200, 100);
-              if (bytes_received_ > 50 * 1024) {
+              if (total_bytes_received_ > 50 * 1024) {
                 UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd50K",
                                             val, 1, 200, 100);
-                if (bytes_received_ > 100 * 1024) {
+                if (total_bytes_received_ > 100 * 1024) {
                   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd100K",
                                               val, 1, 200, 100);
                 }