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 222 matching lines @@
       http_server_properties_(http_server_properties),
       connection_(new ClientSocketHandle),
       read_buffer_(new IOBuffer(kReadBufferSize)),
       read_pending_(false),
       stream_hi_water_mark_(kFirstStreamId),
       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),

[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),
       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();
+  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;
   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 339 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) {
+  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.
     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;
+    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();
 
   // 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 (IsConnected())
+    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 102 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 (!IsClosed()) {
+    state_ = STATE_CLOSED;
     error_ = err;
     if (remove_from_pool)
       RemoveFromPool();
     CloseAllStreams(err);
   }
 }
 
 Value* SpdySession::GetInfoAsValue() const {
   DictionaryValue* dict = new DictionaryValue();
 
@@ skipping 843 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