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

net/spdy/spdy_session.cc

 // Copyright (c) 2012 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_session.h"
 
 #include <map>
 
 #include "base/basictypes.h"
 #include "base/bind.h"
@@ skipping 24 matching lines @@
 #include "net/spdy/spdy_http_utils.h"
 #include "net/spdy/spdy_protocol.h"
 #include "net/spdy/spdy_session_pool.h"
 #include "net/spdy/spdy_stream.h"
 
 namespace net {
 
 namespace {
 
 const int kReadBufferSize = 8 * 1024;
+const int kMaxReadBytes = 64 * 1024;
 const int kDefaultConnectionAtRiskOfLossSeconds = 10;
 const int kHungIntervalSeconds = 10;
 
 // Minimum seconds that unclaimed pushed streams will be kept in memory.
 const int kMinPushedStreamLifetimeSeconds = 300;
 
 Value* NetLogSpdySynCallback(const SpdyHeaderBlock* headers,
                              bool fin,
                              bool unidirectional,
                              SpdyStreamId stream_id,
@@ skipping 176 matching lines @@
       connection_(new ClientSocketHandle),
       read_buffer_(new IOBuffer(kReadBufferSize)),
       read_pending_(false),
       stream_hi_water_mark_(1),  // Always start at 1 for the first stream id.
       write_pending_(false),
       delayed_write_pending_(false),
       is_secure_(false),
       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),
       max_concurrent_streams_limit_(max_concurrent_streams_limit == 0 ?
                                     kMaxConcurrentStreamLimit :
                                     max_concurrent_streams_limit),
       streams_initiated_count_(0),
       streams_pushed_count_(0),
       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),
       pings_in_flight_(0),
       next_ping_id_(1),
       last_activity_time_(base::TimeTicks::Now()),
       check_ping_status_pending_(false),
       flow_control_(false),
       initial_send_window_size_(kSpdyStreamInitialWindowSize),
       initial_recv_window_size_(initial_recv_window_size == 0 ?
@@ skipping 101 matching lines @@
   flow_control_ = (protocol >= kProtoSPDY3);
 
   buffered_spdy_framer_.reset(new BufferedSpdyFramer(version,
                                                      enable_compression_));
   buffered_spdy_framer_->set_visitor(this);
   SendInitialSettings();
   UMA_HISTOGRAM_ENUMERATION("Net.SpdyVersion", protocol, kProtoMaximumVersion);
 
   // 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;
 }
 
 bool SpdySession::VerifyDomainAuthentication(const std::string& domain) {
   if (!verify_domain_authentication_)
     return true;
 
   if (state_ != CONNECTED)
@@ skipping 381 matching lines @@
   // If we're connecting, defer to the connection to give us the actual
   // LoadState.
   if (state_ == CONNECTING)
     return connection_->GetLoadState();
 
   // Just report that we're idle since the session could be doing
   // many things concurrently.
   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();
 
   // The SpdyFramer will use callbacks onto |this| as it parses frames.
   // When errors occur, those callbacks can lead to teardown of all references
   // to |this|, so maintain a reference to self during this call for safe
   // cleanup.
   scoped_refptr<SpdySession> self(this);
 
   DCHECK(buffered_spdy_framer_.get());
   char *data = read_buffer_->data();
   while (bytes_read &&
          buffered_spdy_framer_->error_code() ==
              SpdyFramer::SPDY_NO_ERROR) {
     uint32 bytes_processed =
         buffered_spdy_framer_->ProcessInput(data, bytes_read);
     bytes_read -= bytes_processed;
     data += bytes_processed;
     if (buffered_spdy_framer_->state() == SpdyFramer::SPDY_DONE)
       buffered_spdy_framer_->Reset();
   }
 
   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());
 
   last_activity_time_ = base::TimeTicks::Now();
   write_pending_ = false;
 
   scoped_refptr<SpdyStream> stream = in_flight_write_.stream();
@@ skipping 34 matching lines @@
     WriteSocketLater();
   } else {
     in_flight_write_.release();
 
     // The stream is now errored.  Close it down.
     CloseSessionOnError(
         static_cast<net::Error>(result), true, "The stream has errored.");
   }
 }
 
-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)
     return;
 
   delayed_write_pending_ = true;
   MessageLoop::current()->PostTask(
       FROM_HERE,
@@ skipping 1039 matching lines @@
 SSLClientSocket* SpdySession::GetSSLClientSocket() const {
   if (!is_secure_)
     return NULL;
   SSLClientSocket* ssl_socket =
       reinterpret_cast<SSLClientSocket*>(connection_->socket());
   DCHECK(ssl_socket);
   return ssl_socket;
 }
 
 }  // namespace net