SPDY flow control: enforce obeying send window size via a command-line switch, initial support for receive window size.

BUG=48100
TEST=net_unittestss --gtest_filter="SpdyProtocolTest.ControlFrameStructs:SpdyNetworkTransactionTest.WindowUpdate:SpdyNetworkTransactionTest.WindowUpdateOverflow"

Review URL: http://codereview.chromium.org/3052005

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@53297 0039d316-1c4b-4281-b951-d872f2087c98

7 files changed, 103 insertions, 44 deletions

diff --git a/net/http/http_network_layer.cc b/net/http/http_network_layer.cc
index 7df286f..4950a7b 100644
--- a/net/http/http_network_layer.cc
+++ b/net/http/http_network_layer.cc
@@ -130,6 +130,12 @@ void HttpNetworkLayer::EnableSpdy(const std::string& mode) {
   static const char kDisableCompression[] = "no-compress";
   static const char kDisableAltProtocols[] = "no-alt-protocols";
 
+  // If flow-control is enabled, received WINDOW_UPDATE and SETTINGS
+  // messages are processed and outstanding window size is actually obeyed
+  // when sending data frames, and WINDOW_UPDATE messages are generated
+  // when data is consumed.
+  static const char kEnableFlowControl[] = "flow-control";
+
   // We want an A/B experiment between SPDY enabled and SPDY disabled,
   // but only for pages where SPDY *could have been* negotiated.  To do
   // this, we use NPN, but prevent it from negotiating SPDY.  If the
@@ -179,6 +185,8 @@ void HttpNetworkLayer::EnableSpdy(const std::string& mode) {
     } else if (option == kDisableAltProtocols) {
       use_alt_protocols = false;
       HttpNetworkTransaction::SetUseAlternateProtocols(false);
+    } else if (option == kEnableFlowControl) {
+      SpdySession::SetFlowControl(true);
     } else if (option.empty() && it == spdy_options.begin()) {
       continue;
     } else {
diff --git a/net/http/http_network_layer.h b/net/http/http_network_layer.h
index 8cc62ab..657fcd3 100644
--- a/net/http/http_network_layer.h
+++ b/net/http/http_network_layer.h
@@ -66,9 +66,11 @@ class HttpNetworkLayer : public HttpTransactionFactory, public NonThreadSafe {
 
   // Enable the spdy protocol.
   // Without calling this function, SPDY is disabled.  The mode can be:
-  //   ""            : (default) SSL and compression are enabled.
+  //   ""            : (default) SSL and compression are enabled, flow
+  //                   control disabled.
   //   "no-ssl"      : disables SSL.
   //   "no-compress" : disables compression.
+  //   "flow-control": enables flow control.
   //   "none"        : disables both SSL and compression.
   static void EnableSpdy(const std::string& mode);
 
diff --git a/net/spdy/spdy_network_transaction_unittest.cc b/net/spdy/spdy_network_transaction_unittest.cc
index 7dbc12c..2100e92 100644
--- a/net/spdy/spdy_network_transaction_unittest.cc
+++ b/net/spdy/spdy_network_transaction_unittest.cc
@@ -940,22 +940,24 @@ TEST_F(SpdyNetworkTransactionTest, ResponseWithTwoSynReplies) {
   helper.VerifyDataConsumed();
 }
 
-// Test that WINDOW_UPDATE frames change window_size correctly.
+// Test that sent data frames and received WINDOW_UPDATE frames change
+// the send_window_size_ correctly.
 TEST_F(SpdyNetworkTransactionTest, WindowUpdate) {
   SpdySessionDependencies session_deps;
   scoped_refptr<HttpNetworkSession> session =
       SpdySessionDependencies::SpdyCreateSession(&session_deps);
 
-  // We disable SSL for this test.
   SpdySession::SetSSLMode(false);
+  SpdySession::SetFlowControl(true);
 
   // Setup the request
-  static const char upload[] = { "hello!" };
+  static const char kUploadData[] = "hello!";
+  static const int kUploadDataSize = arraysize(kUploadData)-1;
   HttpRequestInfo request;
   request.method = "POST";
   request.url = GURL("http://www.google.com/");
   request.upload_data = new UploadData();
-  request.upload_data->AppendBytes(upload, strlen(upload));
+  request.upload_data->AppendBytes(kUploadData, kUploadDataSize);
 
   scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyPost(NULL, 0));
   scoped_ptr<spdy::SpdyFrame> body(ConstructSpdyBodyFrame(1, true));
@@ -989,7 +991,8 @@ TEST_F(SpdyNetworkTransactionTest, WindowUpdate) {
 
   ASSERT_TRUE(trans->stream_ != NULL);
   ASSERT_TRUE(trans->stream_->stream() != NULL);
-  EXPECT_EQ(spdy::kInitialWindowSize, trans->stream_->stream()->window_size());
+  EXPECT_EQ(spdy::kInitialWindowSize,
+            trans->stream_->stream()->send_window_size());
 
   EXPECT_EQ(ERR_IO_PENDING, rv);
   rv = callback.WaitForResult();
@@ -997,10 +1000,12 @@ TEST_F(SpdyNetworkTransactionTest, WindowUpdate) {
 
   ASSERT_TRUE(trans->stream_ != NULL);
   ASSERT_TRUE(trans->stream_->stream() != NULL);
-  EXPECT_EQ(spdy::kInitialWindowSize + kDeltaWindowSize,
-            trans->stream_->stream()->window_size());
+  EXPECT_EQ(spdy::kInitialWindowSize + kDeltaWindowSize - kUploadDataSize,
+            trans->stream_->stream()->send_window_size());
   EXPECT_TRUE(data->at_read_eof());
   EXPECT_TRUE(data->at_write_eof());
+
+  SpdySession::SetFlowControl(false);
 }
 
 // Test that WINDOW_UPDATE frame causing overflow is handled correctly.
@@ -1009,16 +1014,16 @@ TEST_F(SpdyNetworkTransactionTest, WindowUpdateOverflow) {
   scoped_refptr<HttpNetworkSession> session =
       SpdySessionDependencies::SpdyCreateSession(&session_deps);
 
-  // We disable SSL for this test.
   SpdySession::SetSSLMode(false);
+  SpdySession::SetFlowControl(true);
 
   // Setup the request
-  static const char upload[] = { "hello!" };
+  static const char kUploadData[] = "hello!";
   HttpRequestInfo request;
   request.method = "POST";
   request.url = GURL("http://www.google.com/");
   request.upload_data = new UploadData();
-  request.upload_data->AppendBytes(upload, strlen(upload));
+  request.upload_data->AppendBytes(kUploadData, arraysize(kUploadData)-1);
 
   scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyPost(NULL, 0));
   scoped_ptr<spdy::SpdyFrame> body(ConstructSpdyBodyFrame(1, true));
@@ -1055,7 +1060,8 @@ TEST_F(SpdyNetworkTransactionTest, WindowUpdateOverflow) {
 
   ASSERT_TRUE(trans->stream_ != NULL);
   ASSERT_TRUE(trans->stream_->stream() != NULL);
-  EXPECT_EQ(spdy::kInitialWindowSize, trans->stream_->stream()->window_size());
+  EXPECT_EQ(spdy::kInitialWindowSize,
+            trans->stream_->stream()->send_window_size());
 
   EXPECT_EQ(ERR_IO_PENDING, rv);
   rv = callback.WaitForResult();
@@ -1067,6 +1073,8 @@ TEST_F(SpdyNetworkTransactionTest, WindowUpdateOverflow) {
 
   EXPECT_FALSE(data->at_read_eof());
   EXPECT_TRUE(data->at_write_eof());
+
+  SpdySession::SetFlowControl(false);
 }
 
 TEST_F(SpdyNetworkTransactionTest, CancelledTransaction) {
diff --git a/net/spdy/spdy_session.cc b/net/spdy/spdy_session.cc
index 06ba7d5..a5bebf5 100644
--- a/net/spdy/spdy_session.cc
+++ b/net/spdy/spdy_session.cc
@@ -133,6 +133,9 @@ class NetLogSpdySettingsParameter : public NetLog::EventParameters {
 // static
 bool SpdySession::use_ssl_ = true;
 
+// static
+bool SpdySession::use_flow_control_ = true;
+
 SpdySession::SpdySession(const HostPortPair& host_port_pair,
                          HttpNetworkSession* session,
                          NetLog* net_log)
@@ -164,7 +167,7 @@ SpdySession::SpdySession(const HostPortPair& host_port_pair,
       sent_settings_(false),
       received_settings_(false),
       in_session_pool_(true),
-      initial_window_size_(spdy::kInitialWindowSize),
+      initial_send_window_size_(spdy::kInitialWindowSize),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_SPDY_SESSION)) {
   net_log_.BeginEvent(
       NetLog::TYPE_SPDY_SESSION,
@@ -372,7 +375,7 @@ int SpdySession::CreateStreamImpl(
   stream->set_priority(priority);
   stream->set_path(path);
   stream->set_net_log(stream_net_log);
-  stream->set_window_size(initial_window_size_);
+  stream->set_send_window_size(initial_send_window_size_);
   ActivateStream(stream);
 
   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyPriorityCount",
@@ -425,12 +428,6 @@ int SpdySession::WriteStreamData(spdy::SpdyStreamId stream_id,
                                  spdy::SpdyDataFlags flags) {
   LOG(INFO) << "Writing Stream Data for stream " << stream_id << " (" << len
             << " bytes)";
-  const int kMss = 1430;  // This is somewhat arbitrary and not really fixed,
-                          // but it will always work reasonably with ethernet.
-  // Chop the world into 2-packet chunks.  This is somewhat arbitrary, but
-  // is reasonably small and ensures that we elicit ACKs quickly from TCP
-  // (because TCP tries to only ACK every other packet).
-  const int kMaxSpdyFrameChunkSize = (2 * kMss) - spdy::SpdyFrame::size();
 
   // Find our stream
   DCHECK(IsStreamActive(stream_id));
@@ -444,6 +441,14 @@ int SpdySession::WriteStreamData(spdy::SpdyStreamId stream_id,
     flags = spdy::DATA_FLAG_NONE;
   }
 
+  // Obey send window size of the stream if flow control is enabled.
+  if (use_flow_control_) {
+    if (stream->send_window_size() <= 0)
+      return ERR_IO_PENDING;
+    len = std::min(len, stream->send_window_size());
+    stream->DecreaseSendWindowSize(len);
+  }
+
   // TODO(mbelshe): reduce memory copies here.
   scoped_ptr<spdy::SpdyDataFrame> frame(
       spdy_framer_.CreateDataFrame(stream_id, data->data(), len, flags));
@@ -1208,8 +1213,6 @@ void SpdySession::OnSettings(const spdy::SpdySettingsControlFrame& frame) {
     settings_storage->Set(host_port_pair_, settings);
   }
 
-  // TODO(agayev): Implement initial and per stream window size update.
-
   received_settings_ = true;
 
   net_log_.AddEvent(
@@ -1239,7 +1242,8 @@ void SpdySession::OnWindowUpdate(
   CHECK_EQ(stream->stream_id(), stream_id);
   CHECK(!stream->cancelled());
 
-  stream->UpdateWindowSize(delta_window_size);
+  if (use_flow_control_)
+    stream->IncreaseSendWindowSize(delta_window_size);
 }
 
 void SpdySession::SendSettings() {
diff --git a/net/spdy/spdy_session.h b/net/spdy/spdy_session.h
index 3add9b0..16f4517 100644
--- a/net/spdy/spdy_session.h
+++ b/net/spdy/spdy_session.h
@@ -31,6 +31,13 @@
 
 namespace net {
 
+// This is somewhat arbitrary and not really fixed, but it will always work
+// reasonably with ethernet. Chop the world into 2-packet chunks.  This is
+// somewhat arbitrary, but is reasonably small and ensures that we elicit
+// ACKs quickly from TCP (because TCP tries to only ACK every other packet).
+const int kMss = 1430;
+const int kMaxSpdyFrameChunkSize = (2 * kMss) - spdy::SpdyFrame::size();
+
 class SpdyStream;
 class HttpNetworkSession;
 class BoundNetLog;
@@ -118,6 +125,9 @@ class SpdySession : public base::RefCounted<SpdySession>,
   static void SetSSLMode(bool enable) { use_ssl_ = enable; }
   static bool SSLMode() { return use_ssl_; }
 
+  // Enable or disable flow control.
+  static void SetFlowControl(bool enable) { use_flow_control_ = enable; }
+
   // If session is closed, no new streams/transactions should be created.
   bool IsClosed() const { return state_ == CLOSED; }
 
@@ -327,14 +337,15 @@ class SpdySession : public base::RefCounted<SpdySession>,
 
   bool in_session_pool_;  // True if the session is currently in the pool.
 
-  int initial_window_size_; // Initial window size for the session; can be
-                            // changed by an arriving SETTINGS frame; newly
-                            // created streams use this value for the initial
-                            // window size.
+  // Initial send window size for the session; can be changed by an
+  // arriving SETTINGS frame; newly created streams use this value for the
+  // initial send window size.
+  int initial_send_window_size_;
 
   BoundNetLog net_log_;
 
   static bool use_ssl_;
+  static bool use_flow_control_;
 };
 
 }  // namespace net
diff --git a/net/spdy/spdy_stream.cc b/net/spdy/spdy_stream.cc
index a5f9d1f..1de67f2 100644
--- a/net/spdy/spdy_stream.cc
+++ b/net/spdy/spdy_stream.cc
@@ -71,23 +71,43 @@ void SpdyStream::set_spdy_headers(
   request_ = headers;
 }
 
-void SpdyStream::UpdateWindowSize(int delta_window_size) {
+void SpdyStream::IncreaseSendWindowSize(int delta_window_size) {
   DCHECK_GE(delta_window_size, 1);
-  int new_window_size = window_size_ + delta_window_size;
-
-  // it's valid for window_size_ to become negative (via an incoming
-  // SETTINGS frame which is handled in SpdySession::OnSettings), in which
-  // case incoming WINDOW_UPDATEs will eventually make it positive;
-  // however, if window_size_ is positive and incoming WINDOW_UPDATE makes
-  // it negative, we have an overflow.
-  if (window_size_ > 0 && new_window_size < 0) {
+  int new_window_size = send_window_size_ + delta_window_size;
+
+  // it's valid for send_window_size_ to become negative (via an incoming
+  // SETTINGS), in which case incoming WINDOW_UPDATEs will eventually make
+  // it positive; however, if send_window_size_ is positive and incoming
+  // WINDOW_UPDATE makes it negative, we have an overflow.
+  if (send_window_size_ > 0 && new_window_size < 0) {
     LOG(WARNING) << "Received WINDOW_UPDATE [delta:" << delta_window_size
                  << "] for stream " << stream_id_
-                 << " overflows window size [current:" << window_size_ << "]";
+                 << " overflows send_window_size_ [current:"
+                 << send_window_size_ << "]";
     session_->ResetStream(stream_id_, spdy::FLOW_CONTROL_ERROR);
     return;
   }
-  window_size_ = new_window_size;
+
+  LOG(INFO) << "Increasing stream " << stream_id_
+            << " send_window_size_ [current:" << send_window_size_ << "]"
+            << " by " << delta_window_size << " bytes";
+  send_window_size_ = new_window_size;
+}
+
+void SpdyStream::DecreaseSendWindowSize(int delta_window_size) {
+  // we only call this method when sending a frame, therefore
+  // |delta_window_size| should be within the valid frame size range.
+  DCHECK_GE(delta_window_size, 1);
+  DCHECK_LE(delta_window_size, kMaxSpdyFrameChunkSize);
+
+  // |send_window_size_| should have been at least |delta_window_size| for
+  // this call to happen.
+  DCHECK_GE(send_window_size_, delta_window_size);
+
+  LOG(INFO) << "Decreasing stream " << stream_id_
+            << " send_window_size_ [current:" << send_window_size_ << "]"
+            << " by " << delta_window_size  << " bytes";
+  send_window_size_ -= delta_window_size;
 }
 
 base::Time SpdyStream::GetRequestTime() const {
diff --git a/net/spdy/spdy_stream.h b/net/spdy/spdy_stream.h
index 13a6644..0d2ae44 100644
--- a/net/spdy/spdy_stream.h
+++ b/net/spdy/spdy_stream.h
@@ -102,13 +102,18 @@ class SpdyStream : public base::RefCounted<SpdyStream> {
   int priority() const { return priority_; }
   void set_priority(int priority) { priority_ = priority; }
 
-  int window_size() const { return window_size_; }
-  void set_window_size(int window_size) { window_size_ = window_size; }
+  int send_window_size() const { return send_window_size_; }
+  void set_send_window_size(int window_size) {
+    send_window_size_ = window_size;
+  }
 
-  // Updates |window_size_| with delta extracted from a WINDOW_UPDATE
-  // frame; sends a RST_STREAM if delta overflows |window_size_| and
+  // Increases |send_window_size_| with delta extracted from a WINDOW_UPDATE
+  // frame; sends a RST_STREAM if delta overflows |send_window_size_| and
   // removes the stream from the session.
-  void UpdateWindowSize(int delta_window_size);
+  void IncreaseSendWindowSize(int delta_window_size);
+
+  // Decreases |send_window_size_| by the given number of bytes.
+  void DecreaseSendWindowSize(int delta_window_size);
 
   const BoundNetLog& net_log() const { return net_log_; }
   void set_net_log(const BoundNetLog& log) { net_log_ = log; }
@@ -208,7 +213,8 @@ class SpdyStream : public base::RefCounted<SpdyStream> {
   spdy::SpdyStreamId stream_id_;
   std::string path_;
   int priority_;
-  int window_size_;
+  int send_window_size_;
+
   const bool pushed_;
   ScopedBandwidthMetrics metrics_;
   bool syn_reply_received_;