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 761 matching lines @@
   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!
 
   read_pending_ = false;
 
+  bool can_do_more = ProcessBytesRead(bytes_read);
+
+  if (can_do_more)
+    ReadSocket();
+}
+
+bool SpdySession::ProcessBytesRead(int bytes_read) {
   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 false;
   }
 
   bytes_received_ += 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();
+  return state_ != CLOSED;
 }
 
 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 35 matching lines @@
   } 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;
+  bool can_do_more = false;
+  net::Error result;
+  do {
+    if (read_pending_)
+      return OK;
 
-  if (state_ == CLOSED) {
-    NOTREACHED();
-    return ERR_UNEXPECTED;
-  }
+    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;
+    CHECK(connection_.get());
+    CHECK(connection_->socket());
+    int bytes_read = connection_->socket()->Read(
+        read_buffer_.get(),

[willchan no longer on Chromium, 2012/12/24 00:19:35]

don't we need to increment the offset we read into this buffer?

[Ryan Sleevi, 2012/12/24 05:38:44]

No, because ProcessBytesRead drains |read_buffer_|

+        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.");
+        result = ERR_CONNECTION_CLOSED;
+        break;
+      case net::ERR_IO_PENDING:
+        // Waiting for data.  Nothing to do now.
+        read_pending_ = true;
+        result = ERR_IO_PENDING;
+        break;
+      default:
+        can_do_more = ProcessBytesRead(bytes_read);
+        result = OK;
+        break;
+    }
+  } while (result == OK && can_do_more);

[Ryan Sleevi, 2012/12/24 05:38:44]

Like I expressed Friday, I'm a slightly concerned with this pattern, if only
because it's significantly different from how we normally handle reads and the
state machine.

In particular, the way this is currently implemented requires duplicating the
error handling logic (lines 894-903 and lines 789-795), splitting the handling
of values < 0 from line 905 (eg: not handled) and line 789 (invoked by the
callback from line 782)

It does seem like putting in an explicit state machine logic to flip between
these states makes it clearer to the reader the flow, and lets you keep error
handling consistently in the same place. It will also make the concept map
between (TCPClientSocket, SSLClientSocket, SpdyStream, ProxyService,
QuicHttpStream,  etc) and this code.

[Ryan Hamilton, 2012/12/24 15:26:35]

+1 on an explict STATE_DO_READ/STATE_DO_READ_COMPLETE to remove the duplication.

+  return result;
 }
 
 void SpdySession::WriteSocketLater() {
   if (delayed_write_pending_)
     return;
 
   if (state_ < CONNECTED)
     return;
 
   delayed_write_pending_ = true;
@@ skipping 1041 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