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 174489)
+++ net/spdy/spdy_session.cc	(working copy)
@@ -42,6 +42,7 @@
 namespace {
 
 const int kReadBufferSize = 8 * 1024;
+const int kMaxReadBytes = 64 * 1024;
 const int kDefaultConnectionAtRiskOfLossSeconds = 10;
 const int kHungIntervalSeconds = 10;
 
@@ -238,6 +239,7 @@
       certificate_error_code_(OK),
       error_(OK),
       state_(IDLE),
+      io_state_(STATE_NONE),
       max_concurrent_streams_(initial_max_concurrent_streams == 0 ?
                               kInitialMaxConcurrentStreams :
                               initial_max_concurrent_streams),
@@ -249,6 +251,7 @@
       streams_pushed_and_claimed_count_(0),
       streams_abandoned_count_(0),
       bytes_received_(0),
+      bytes_read_(0),
       sent_settings_(false),
       received_settings_(false),
       stalled_streams_(0),
@@ -370,7 +373,10 @@
 
   // Write out any data that we might have to send, such as the settings frame.
   WriteSocketLater();
-  net::Error error = ReadSocket();
+  io_state_ = STATE_DO_READ;
+  bytes_read_ = 0;
+  int result = DoLoop(OK);
+  net::Error error =   static_cast<net::Error>(result);
   if (error == ERR_IO_PENDING)
     return OK;
   return error;
@@ -772,23 +778,74 @@
   return LOAD_STATE_IDLE;
 }
 
-void SpdySession::OnReadComplete(int bytes_read) {
+int SpdySession::DoLoop(int result) {
+  do {
+    if (read_pending_)
+      return OK;
+
+    if (state_ == CLOSED)
+      return ERR_UNEXPECTED;
+
+    IoState io_state = io_state_;
+    io_state_ = STATE_NONE;
+    switch (io_state) {
+      case STATE_DO_READ:
+        result = DoRead();
+        break;
+      case STATE_DO_READ_COMPLETE:
+        result = DoReadComplete(result);
+        break;
+      default:
+        NOTREACHED() << "io_state_: " << io_state;
+        break;
+    }
+  } while (io_state_ != STATE_NONE && result != ERR_IO_PENDING);

[Ryan Sleevi, 2012/12/27 05:29:48]

Design Suggestion: You can probably make it clearer how kMaxReadBytes relates by
moving it into here, rather than being part of DoRead. That way, even if you add
additional states, it's clear how much is processed at a time.


int bytes_read = 0;
do {
  switch (io_state_) {
  ...
    case STATE_DO_READ_COMPLETE:
      if (result > 0)
        bytes_read += result;
      result = DoReadComplete(result);
  ...
  }
} while (io_state_ != STATE_NONE && result != ERR_IO_PENDING &&
         bytes_read < kMaxReadBytes);
if (bytes_read >= kMaxReadBytes) {
  result = ERR_IO_PENDING;
  MessageLoop::current()->PostTask(...);
}
return result;


This lets you avoid having to track both bytes_read_ and read_pending_ on the
class itself. It also makes it clear that kMaxReadBytes is the terminator for
the processing loop overall - in case more states are introduced, for example.

+  return result;
+}
+
+int SpdySession::DoRead() {
+  io_state_ = STATE_DO_READ_COMPLETE;
+
+  CHECK(connection_.get());
+  CHECK(connection_->socket());
+  int bytes_read = connection_->socket()->Read(
+      read_buffer_.get(),
+      kReadBufferSize,
+      base::Bind(&SpdySession::OnReadComplete, base::Unretained(this)));
+  if (bytes_read == net::ERR_IO_PENDING) {
+    read_pending_ = true;
+  } else if (bytes_read > 0) {
+    if (bytes_read_ > kMaxReadBytes) {
+      // Data was read, process it.
+      // Schedule the work through the message loop to yield for other tasks to
+      // run on IO thread.
+      read_pending_ = true;
+      MessageLoop::current()->PostTask(
+          FROM_HERE,
+          base::Bind(&SpdySession::OnReadComplete,
+                     weak_factory_.GetWeakPtr(), bytes_read));
+    }
+  }

[Ryan Sleevi, 2012/12/27 05:29:48]

Here, it would be as simple as
CHECK(...)
return connection_->socket()->Read(...)

[We do this in the Http layers as an idiom, for example]

[ramant (doing other things), 2012/12/29 03:02:20]

Done.

+  return bytes_read;
+}
+
+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;
-
   if (bytes_read <= 0) {
     // Session is tearing down.
     net::Error error = static_cast<net::Error>(bytes_read);
     if (bytes_read == 0)
       error = ERR_CONNECTION_CLOSED;
     CloseSessionOnError(error, true, "bytes_read is <= 0.");
-    return;
+    io_state_ = STATE_NONE;

[Ryan Sleevi, 2012/12/27 05:29:48]

If you're relying on this pattern to set io_state_ (and you seem to be mixing
two patterns - having the loop set it on 789/790 like the SSL/TCP sockets do,
but then setting it in the handlers here, like the HTTP streams do), then you
should do it at line 836 to ensure that all early-returns have the proper state.
Then you can drop lines 873 & 874.

[ramant (doing other things), 2012/12/29 03:02:20]

Done.

+    return error;
   }
 
   bytes_received_ += bytes_read;
+  bytes_read_ += bytes_read;
 
   last_activity_time_ = base::TimeTicks::Now();
 
@@ -812,9 +869,19 @@
   }
 
   if (state_ != CLOSED)
-    ReadSocket();
+    io_state_ = STATE_DO_READ;
+  else
+    io_state_ = STATE_NONE;
+  return OK;
 }
 
+void SpdySession::OnReadComplete(int bytes_read) {
+  io_state_ = STATE_DO_READ_COMPLETE;

[Ryan Sleevi, 2012/12/27 05:29:48]

Normally we don't set the state in the bound callbacks, and we instead let code
line like 807 handle the asynchronous restarting.

[ramant (doing other things), 2012/12/29 03:02:20]

Done.

+  read_pending_ = false;
+  bytes_read_ = 0;

[Ryan Sleevi, 2012/12/27 05:29:48]

My primary reason for suggesting the loop shuffling above was when I was reading
this code, I missed the _ and was trying to figure out why you were clearing
|bytes_read|.

Had line 878 said something like "rv" (which is our normal shorthand), I
probably would not have been as confused.

+  DoLoop(bytes_read);
+}
+
 void SpdySession::OnWriteComplete(int result) {
   DCHECK(write_pending_);
   DCHECK(in_flight_write_.size());
@@ -867,44 +934,6 @@
   }
 }
 
-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;