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 176979)
+++ 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),
       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;
 
@@ -767,7 +768,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
@@ -776,11 +777,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;
+  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_);
 
   if (bytes_read <= 0) {
     // Session is tearing down.
@@ -788,10 +850,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;
 
   last_activity_time_ = base::TimeTicks::Now();
 
@@ -814,8 +877,9 @@
       buffered_spdy_framer_->Reset();
   }
 
-  if (state_ != CLOSED)
-    ReadSocket();
+  if (IsConnected())
+    state_ = STATE_DO_READ;
+  return OK;
 }
 
 void SpdySession::OnWriteComplete(int result) {
@@ -870,49 +934,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;
@@ -929,7 +955,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.
@@ -1051,8 +1077,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();
@@ -1916,16 +1942,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);
                 }