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"
 #include "base/compiler_specific.h"
+#include "base/debug/trace_event.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/metrics/field_trial.h"
 #include "base/metrics/histogram.h"
 #include "base/metrics/stats_counters.h"
 #include "base/stl_util.h"
 #include "base/string_number_conversions.h"
 #include "base/string_util.h"
 #include "base/stringprintf.h"
 #include "base/time.h"
@@ skipping 208 matching lines @@
       http_server_properties_(http_server_properties),
       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),
+      state_(STATE_IDLE),
       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),
+      total_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 39 matching lines @@
 
 SpdySession::CallbackResultPair::CallbackResultPair(
     const CompletionCallback& callback_in, int result_in)
     : callback(callback_in),
       result(result_in) {
 }
 
 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);
   }
 
   if (connection_->is_initialized()) {
     // With SPDY we can't recycle sockets.
     connection_->socket()->Disconnect();
   }
 
   // Streams should all be gone now.
   DCHECK_EQ(0u, num_active_streams());
   DCHECK_EQ(0u, num_unclaimed_pushed_streams());
 
   DCHECK(pending_callback_map_.empty());
 
   RecordHistograms();
 
   net_log_.EndEvent(NetLog::TYPE_SPDY_SESSION);
 }
 
 net::Error SpdySession::InitializeWithSocket(
     ClientSocketHandle* connection,
     bool is_secure,
     int certificate_error_code) {
   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;
 
   NextProto protocol = default_protocol_;
   NextProto protocol_negotiated = connection->socket()->GetNegotiatedProtocol();
   if (protocol_negotiated != kProtoUnknown) {
     protocol = protocol_negotiated;
   }
 
@@ skipping 11 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();
+  net::Error error = static_cast<net::Error>(DoLoop(OK));

[Ryan Hamilton, 2012/12/29 18:41:52]

nit:  int rv = DoLoop(OK);

[ramant (doing other things), 2013/01/16 00:31:58]

Done.

   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)
+  if (!IsConnected())
     return false;
 
   SSLInfo ssl_info;
   bool was_npn_negotiated;
   NextProto protocol_negotiated = kProtoUnknown;
   if (!GetSSLInfo(&ssl_info, &was_npn_negotiated, &protocol_negotiated))
     return true;   // This is not a secure session, so all domains are okay.
 
   return !ssl_info.client_cert_sent &&
       (enable_credential_frames_ || !ssl_info.channel_id_sent ||
        ServerBoundCertService::GetDomainForHost(domain) ==
        ServerBoundCertService::GetDomainForHost(
            host_port_proxy_pair_.first.host())) &&
       ssl_info.cert->VerifyNameMatch(domain);
 }
 
 void SpdySession::SetStreamHasWriteAvailable(SpdyStream* stream,
                                              SpdyIOBufferProducer* producer) {
   write_queue_.push(producer);
   stream_producers_[producer] = stream;
   WriteSocketLater();
 }
 
 int SpdySession::GetPushStream(
     const GURL& url,
     scoped_refptr<SpdyStream>* stream,
     const BoundNetLog& stream_net_log) {
-  CHECK_NE(state_, CLOSED);
+  CHECK_NE(state_, STATE_CLOSED);
 
   *stream = NULL;
 
   // Don't allow access to secure push streams over an unauthenticated, but
   // encrypted SSL socket.
   if (is_secure_ && certificate_error_code_ != OK &&
       (url.SchemeIs("https") || url.SchemeIs("wss"))) {
     RecordProtocolErrorHistogram(
         PROTOCOL_ERROR_REQUEST_FOR_SECURE_CONTENT_OVER_INSECURE_SESSION);
     CloseSessionOnError(
@@ skipping 335 matching lines @@
   return ContainsKey(active_streams_, stream_id);
 }
 
 LoadState SpdySession::GetLoadState() const {
   // NOTE: The application only queries the LoadState via the
   //       SpdyNetworkTransaction, and details are only needed when
   //       we're in the process of connecting.
 
   // 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
   // many things concurrently.
   return LOAD_STATE_IDLE;
 }
 
 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) {
+  TRACE_EVENT0("spdy", "SpdySession::DoLoop");
+  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);
+
+  if (result == net::ERR_IO_PENDING)
+    read_pending_ = true;

[Ryan Hamilton, 2012/12/29 18:41:52]

nit: I would move these two lines to DoRead.

[ramant (doing other things), 2013/01/16 00:31:58]

Done.

+  return result;
+}
+
+int SpdySession::DoRead() {
+  TRACE_EVENT0("spdy", "SpdySession::DoRead");
+  const int kMaxReadBytes = 32 * 1024;
+  if (bytes_read_ > kMaxReadBytes) {

[Ryan Hamilton, 2012/12/29 18:41:52]

bytes_read_ is currently being set to 0 in DoLoop().  How does that interact
with this expression?

[ramant (doing other things), 2013/01/16 00:31:58]

We increment bytes_read_ in DoReadComplete. It is initialized to 0, before we
start reading and we read continuously 32k bytes without yielding and then we
yield by calling DoLoop (from StartRead) and we start reading continuously for
another 32k bytes.

+    state_ = STATE_DO_READ;
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        base::Bind(&SpdySession::StartRead,
+                   weak_factory_.GetWeakPtr()));
+    return ERR_IO_PENDING;
+  }
+
+  DCHECK(!read_pending_);

[Ryan Hamilton, 2012/12/29 18:41:52]

nit Should this be moved above the if()?

[ramant (doing other things), 2013/01/16 00:31:58]

Done.

+  CHECK(connection_.get());
+  CHECK(connection_->socket());
+  state_ = STATE_DO_READ_COMPLETE;
+  return connection_->socket()->Read(
+      read_buffer_.get(),
+      kReadBufferSize,
+      base::Bind(&SpdySession::OnReadComplete, base::Unretained(this)));
+}
+
+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;
+  TRACE_EVENT0("spdy", "SpdySession::DoReadComplete");
+  DCHECK(!read_pending_);
 
   if (bytes_read <= 0) {
     // Session is tearing down.
     net::Error error = static_cast<net::Error>(bytes_read);

[Ryan Hamilton, 2012/12/29 18:41:52]

nit: please get rid of this cast and assignment. 

[ramant (doing other things), 2013/01/16 00:31:58]

"CloseSessionOnError(error,..." use net::Error. On windows, passing int for
net::Error results in a compilation error. Sorry for the cast.

     if (bytes_read == 0)
       error = ERR_CONNECTION_CLOSED;
     CloseSessionOnError(error, true, "bytes_read is <= 0.");
-    return;
+    return ERR_CONNECTION_CLOSED;

[Ryan Hamilton, 2012/12/29 18:41:52]

Shouldn't this say "return error;"?

[ramant (doing other things), 2013/01/16 00:31:58]

CloseSessionOnError is logging the error. Thought it is better to return what is
current state of connection. 

If state is ERR_CONNECTION_CLOSED, higher level code knows the socket is closed
and it doesn't perform extra reads (for example during tear down of tests).

Done. 

   }
 
-  bytes_received_ += bytes_read;
+  total_bytes_received_ += bytes_read;
+  bytes_read_ += bytes_read;

[Ryan Hamilton, 2012/12/29 18:41:52]

I'm confused about the difference between bytes_read_ and total_bytes_read_. 
Can you add a comment?

[ramant (doing other things), 2013/01/16 00:31:58]

Added a comment in the header file.

Done.

 
   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();
+  if (state_ != STATE_CLOSED)
+    state_ = STATE_DO_READ;
+  return OK;
 }
 
 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)
+  if (!IsConnected())
     return;
 
   delayed_write_pending_ = true;
   MessageLoop::current()->PostTask(
       FROM_HERE,
       base::Bind(&SpdySession::WriteSocket, weak_factory_.GetWeakPtr()));
 }
 
 void SpdySession::WriteSocket() {
   // This function should only be called via WriteSocketLater.
   DCHECK(delayed_write_pending_);
   delayed_write_pending_ = false;
 
   // 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.
     return;
 
   // Loop sending frames until we've sent everything or until the write
   // returns error (or ERR_IO_PENDING).
   DCHECK(buffered_spdy_framer_.get());
   while (in_flight_write_.buffer() || !write_queue_.empty()) {
     if (!in_flight_write_.buffer()) {
@@ skipping 101 matching lines @@
   scoped_refptr<SpdySession> self(this);
 
   DCHECK_LT(err, OK);
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_CLOSE,
       base::Bind(&NetLogSpdySessionCloseCallback, err, &description));
 
   // 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 (state_ != STATE_CLOSED) {

[Ryan Hamilton, 2012/12/29 18:41:52]

if (!IsConnected()) {

[ramant (doing other things), 2013/01/16 00:31:58]

Changed it to !IsClosed() (in case connect failed).
Done.

+    state_ = STATE_CLOSED;
     error_ = err;
     if (remove_from_pool)
       RemoveFromPool();
     CloseAllStreams(err);
   }
 }
 
 Value* SpdySession::GetInfoAsValue() const {
   DictionaryValue* dict = new DictionaryValue();
 
@@ skipping 838 matching lines @@
     SettingsMap::const_iterator it;
     for (it = settings_map.begin(); it != settings_map.end(); ++it) {
       const SpdySettingsIds id = it->first;
       const uint32 val = it->second.second;
       switch (id) {
         case SETTINGS_CURRENT_CWND:
           // Record several different histograms to see if cwnd converges
           // 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);
                 }
               }
             }
           }
           break;
         case SETTINGS_ROUND_TRIP_TIME:
           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsRTT",
                                       val, 1, 1200, 100);
@@ skipping 26 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