SPDY - Re-land greedy read support for SpdySession

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 <algorithm>
 #include <map>
 
 #include "base/basictypes.h"
@@ skipping 311 matching lines @@
                          size_t max_concurrent_streams_limit,
                          TimeFunc time_func,
                          const HostPortPair& trusted_spdy_proxy,
                          NetLog* net_log)
     : ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)),
       host_port_proxy_pair_(host_port_proxy_pair),
       spdy_session_pool_(spdy_session_pool),
       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),
       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_state_(FLOW_CONTROL_NONE),
       stream_initial_send_window_size_(kSpdyStreamInitialWindowSize),
       stream_initial_recv_window_size_(stream_initial_recv_window_size == 0 ?
@@ skipping 20 matching lines @@
   DCHECK(HttpStreamFactory::spdy_enabled());
   net_log_.BeginEvent(
       NetLog::TYPE_SPDY_SESSION,
       base::Bind(&NetLogSpdySessionCallback, &host_port_proxy_pair_));
   next_unclaimed_push_stream_sweep_time_ = time_func_() +
       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
   // TODO(mbelshe): consider randomization of the stream_hi_water_mark.
 }
 
 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();
   }
 
@@ skipping 11 matching lines @@
   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 30 matching lines @@
     DCHECK_GT(kDefaultInitialRecvWindowSize, session_recv_window_size_);
     // This condition implies that |kDefaultInitialRecvWindowSize| -
     // |session_recv_window_size_| doesn't overflow.
     DCHECK_GT(session_recv_window_size_, 0);
     IncreaseRecvWindowSize(
         kDefaultInitialRecvWindowSize - session_recv_window_size_);
   }
 
   // Write out any data that we might have to send, such as the settings frame.
   WriteSocketLater();
-  net::Error error = ReadSocket();
+  int error = DoLoop(OK);
+  UMA_HISTOGRAM_COUNTS("Net.SpdySession.Init.BytesRead", bytes_read_);
   if (error == ERR_IO_PENDING)
     return OK;
-  return error;
+  if (error != OK)
+    UMA_HISTOGRAM_COUNTS("Net.SpdySession.InitAndError.BytesRead", bytes_read_);

[Ryan Hamilton, 2013/04/14 02:21:46]

We had also talked about adding a histogram of the number of bytes when the
session was closed.  That would be somewhat redundant with this, I guess?

[ramant (doing other things), 2013/04/16 19:12:23]

Added new histograms per our discussion.
Done.

+  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 349 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) {
-  // Parse a frame.  For now this code requires that the frame fit into our
-  // buffer (32KB).
-  // TODO(mbelshe): support arbitrarily large frames!
+  DCHECK_NE(state_, STATE_DO_READ);
+  DoLoop(bytes_read);
+}
 
-  read_pending_ = false;
+void SpdySession::StartRead() {
+  DCHECK_NE(state_, STATE_DO_READ_COMPLETE);
+  DoLoop(OK);
+}
 
-  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;
-  }
-
-  bytes_received_ += bytes_read;
-
-  last_activity_time_ = base::TimeTicks::Now();
+int SpdySession::DoLoop(int result) {
+  bytes_read_ = 0;
 
   // 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);
 
+  do {
+    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 && state_ != STATE_CLOSED);
+
+  return result;
+}
+
+int SpdySession::DoRead() {
+  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;
+  return connection_->socket()->Read(
+      read_buffer_.get(),
+      kReadBufferSize,
+      base::Bind(&SpdySession::OnReadComplete, weak_factory_.GetWeakPtr()));
+}
+
+int SpdySession::DoReadComplete(int result) {
+  // Parse a frame.  For now this code requires that the frame fit into our
+  // buffer (32KB).
+  // TODO(mbelshe): support arbitrarily large frames!
+
+  if (result <= 0) {
+    // Session is tearing down.
+    net::Error error = static_cast<net::Error>(result);
+    if (result == 0)
+      error = ERR_CONNECTION_CLOSED;
+    CloseSessionOnError(error, true, "result is <= 0.");
+    return ERR_CONNECTION_CLOSED;
+  }
+
+  total_bytes_received_ += result;
+  bytes_read_ += result;
+
+  last_activity_time_ = base::TimeTicks::Now();
+
   DCHECK(buffered_spdy_framer_.get());
   char* data = read_buffer_->data();
-  while (bytes_read &&
+  while (result &&
          buffered_spdy_framer_->error_code() ==
              SpdyFramer::SPDY_NO_ERROR) {
     uint32 bytes_processed =
-        buffered_spdy_framer_->ProcessInput(data, bytes_read);
-    bytes_read -= bytes_processed;
+        buffered_spdy_framer_->ProcessInput(data, result);
+    result -= bytes_processed;
     data += bytes_processed;
   }
 
-  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 36 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);
   }
 }
 
 base::Value* SpdySession::GetInfoAsValue() const {
   base::DictionaryValue* dict = new base::DictionaryValue();
 
@@ skipping 936 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 158 matching lines @@
   }
 
   session_recv_window_size_ -= delta_window_size;
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_UPDATE_RECV_WINDOW,
       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
                  -delta_window_size, session_recv_window_size_));
 }
 
 }  // namespace net