SPDY: flow-control fix: resume I/O once a WINDOW_UPDATE frame is received for a stalled stream.

BUG=none
TEST=net_unittests

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

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

13 files changed, 279 insertions, 111 deletions

diff --git a/net/http/http_network_transaction.h b/net/http/http_network_transaction.h
index 97a9d47..12f06e3 100644
--- a/net/http/http_network_transaction.h
+++ b/net/http/http_network_transaction.h
@@ -84,6 +84,9 @@ class HttpNetworkTransaction : public HttpTransaction {
 
  private:
   FRIEND_TEST_ALL_PREFIXES(HttpNetworkTransactionTest, ResetStateForRestart);
+  FRIEND_TEST_ALL_PREFIXES(SpdyNetworkTransactionTest, WindowUpdate);
+  FRIEND_TEST_ALL_PREFIXES(SpdyNetworkTransactionTest, WindowUpdateOverflow);
+  FRIEND_TEST_ALL_PREFIXES(SpdyNetworkTransactionTest, FlowControlStallResume);
 
   enum State {
     STATE_RESOLVE_PROXY,
diff --git a/net/socket/socket_test_util.cc b/net/socket/socket_test_util.cc
index 142a39e..a89f79c 100644
--- a/net/socket/socket_test_util.cc
+++ b/net/socket/socket_test_util.cc
@@ -548,6 +548,11 @@ void DelayedSocketData::CompleteRead() {
     socket()->OnReadComplete(GetNextRead());
 }
 
+void DelayedSocketData::ForceNextRead() {
+  write_delay_ = 0;
+  CompleteRead();
+}
+
 OrderedSocketData::OrderedSocketData(
     MockRead* reads, size_t reads_count, MockWrite* writes, size_t writes_count)
     : StaticSocketDataProvider(reads, reads_count, writes, writes_count),
diff --git a/net/socket/socket_test_util.h b/net/socket/socket_test_util.h
index 0ccf878..9367349 100644
--- a/net/socket/socket_test_util.h
+++ b/net/socket/socket_test_util.h
@@ -284,6 +284,7 @@ class DelayedSocketData : public StaticSocketDataProvider,
   virtual MockWriteResult OnWrite(const std::string& data);
   virtual void Reset();
   void CompleteRead();
+  void ForceNextRead();
 
  private:
   int write_delay_;
diff --git a/net/spdy/spdy_http_stream.cc b/net/spdy/spdy_http_stream.cc
index 4e62b45..3ba777a 100644
--- a/net/spdy/spdy_http_stream.cc
+++ b/net/spdy/spdy_http_stream.cc
@@ -340,9 +340,12 @@ int SpdyHttpStream::OnSendBody() {
                                   spdy::DATA_FLAG_FIN);
 }
 
-bool SpdyHttpStream::OnSendBodyComplete(int status) {
+void SpdyHttpStream::OnSendBodyComplete(int status) {
   CHECK(request_body_stream_.get());
   request_body_stream_->DidConsume(status);
+}
+
+bool SpdyHttpStream::IsFinishedSendingBody() {
   return request_body_stream_->eof();
 }
 
diff --git a/net/spdy/spdy_http_stream.h b/net/spdy/spdy_http_stream.h
index faf1388..95134fa 100644
--- a/net/spdy/spdy_http_stream.h
+++ b/net/spdy/spdy_http_stream.h
@@ -85,7 +85,8 @@ class SpdyHttpStream : public SpdyStream::Delegate, public HttpStream {
 
   virtual bool OnSendHeadersComplete(int status);
   virtual int OnSendBody();
-  virtual bool OnSendBodyComplete(int status);
+  virtual void OnSendBodyComplete(int status);
+  virtual bool IsFinishedSendingBody();
 
   // Called by the SpdySession when a response (e.g. a SYN_REPLY) has been
   // received for this stream.
diff --git a/net/spdy/spdy_network_transaction.h b/net/spdy/spdy_network_transaction.h
index fbf1318..1323932 100644
--- a/net/spdy/spdy_network_transaction.h
+++ b/net/spdy/spdy_network_transaction.h
@@ -55,9 +55,6 @@ class SpdyNetworkTransaction : public HttpTransaction {
   virtual uint64 GetUploadProgress() const;
 
  private:
-  FRIEND_TEST_ALL_PREFIXES(SpdyNetworkTransactionTest, WindowUpdate);
-  FRIEND_TEST_ALL_PREFIXES(SpdyNetworkTransactionTest, WindowUpdateOverflow);
-
   enum State {
     STATE_INIT_CONNECTION,
     STATE_INIT_CONNECTION_COMPLETE,
diff --git a/net/spdy/spdy_network_transaction_unittest.cc b/net/spdy/spdy_network_transaction_unittest.cc
index c95bcb3..d2da3ed 100644
--- a/net/spdy/spdy_network_transaction_unittest.cc
+++ b/net/spdy/spdy_network_transaction_unittest.cc
@@ -32,6 +32,7 @@ class SpdyNetworkTransactionTest
     // By default, all tests turn off compression.
     EnableCompression(false);
     google_get_request_initialized_ = false;
+    google_post_request_initialized_ = false;
   }
 
   virtual void TearDown() {
@@ -254,13 +255,25 @@ class SpdyNetworkTransactionTest
   const HttpRequestInfo& CreateGetRequest() {
     if (!google_get_request_initialized_) {
       google_get_request_.method = "GET";
-      google_get_request_.url = GURL("http://www.google.com/");
+      google_get_request_.url = GURL(kDefaultURL);
       google_get_request_.load_flags = 0;
       google_get_request_initialized_ = true;
     }
     return google_get_request_;
   }
 
+  const HttpRequestInfo& CreatePostRequest() {
+    if (!google_post_request_initialized_) {
+      google_post_request_.method = "POST";
+      google_post_request_.url = GURL(kDefaultURL);
+      google_post_request_.upload_data = new UploadData();
+      google_post_request_.upload_data->AppendBytes(kUploadData,
+                                                    kUploadDataSize);
+      google_post_request_initialized_ = true;
+    }
+    return google_post_request_;
+  }
+
   class RunServerPushTestCallback : public CallbackRunner< Tuple1<int> > {
    public:
     RunServerPushTestCallback(scoped_refptr<net::IOBufferWithSize> buffer,
@@ -352,7 +365,9 @@ class SpdyNetworkTransactionTest
 
  private:
   bool google_get_request_initialized_;
+  bool google_post_request_initialized_;
   HttpRequestInfo google_get_request_;
+  HttpRequestInfo google_post_request_;
   HttpRequestInfo google_get_push_request_;
 };
 
@@ -949,23 +964,7 @@ TEST_P(SpdyNetworkTransactionTest, ThreeGetsWithMaxConcurrentDelete) {
 
 // Test that a simple POST works.
 TEST_P(SpdyNetworkTransactionTest, Post) {
-  static const char upload[] = { "hello!" };
-
-  // Setup the request
-  HttpRequestInfo request;
-  request.method = "POST";
-  request.url = GURL("http://www.google.com/");
-  request.upload_data = new UploadData();
-  request.upload_data->AppendBytes(upload, strlen(upload));
-
-  // Http POST Content-Length is using UploadDataStream::size().
-  // It is the same as request.upload_data->GetContentLength().
-  scoped_ptr<UploadDataStream> stream(UploadDataStream::Create(
-      request.upload_data, NULL));
-  ASSERT_EQ(request.upload_data->GetContentLength(), stream->size());
-
-  scoped_ptr<spdy::SpdyFrame>
-      req(ConstructSpdyPost(request.upload_data->GetContentLength(), NULL, 0));
+  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyPost(kUploadDataSize, NULL, 0));
   scoped_ptr<spdy::SpdyFrame> body(ConstructSpdyBodyFrame(1, true));
   MockWrite writes[] = {
     CreateMockWrite(*req),
@@ -982,7 +981,7 @@ TEST_P(SpdyNetworkTransactionTest, Post) {
   scoped_refptr<DelayedSocketData> data(
       new DelayedSocketData(2, reads, arraysize(reads),
                             writes, arraysize(writes)));
-  NormalSpdyTransactionHelper helper(request,
+  NormalSpdyTransactionHelper helper(CreatePostRequest(),
                                      BoundNetLog(), GetParam());
   helper.RunToCompletion(data.get());
   TransactionHelperResult out = helper.output();
@@ -1096,7 +1095,6 @@ TEST_P(SpdyNetworkTransactionTest, PostWithEarlySynReply) {
   scoped_ptr<spdy::SpdyFrame> body(ConstructSpdyBodyFrame(1, true));
   MockWrite writes[] = {
     CreateMockWrite(*req.get(), 2),
-    CreateMockWrite(*body.get(), 3),  // POST upload frame
   };
 
   scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyPostSynReply(NULL, 0));
@@ -1114,6 +1112,8 @@ TEST_P(SpdyNetworkTransactionTest, PostWithEarlySynReply) {
   helper.RunPreTestSetup();
   helper.AddData(data.get());
   helper.RunDefaultTest();
+  helper.VerifyDataConsumed();
+
   TransactionHelperResult out = helper.output();
   EXPECT_EQ(ERR_SPDY_PROTOCOL_ERROR, out.rv);
 }
@@ -1218,37 +1218,15 @@ TEST_P(SpdyNetworkTransactionTest, ResponseWithTwoSynReplies) {
 // Test that sent data frames and received WINDOW_UPDATE frames change
 // the send_window_size_ correctly.
 TEST_P(SpdyNetworkTransactionTest, WindowUpdate) {
-  SpdySessionDependencies session_deps;
-  scoped_refptr<HttpNetworkSession> session =
-      SpdySessionDependencies::SpdyCreateSession(&session_deps);
-
-  SpdySession::SetSSLMode(false);
   SpdySession::SetFlowControl(true);
 
-  // Setup the request
-  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(kUploadData, kUploadDataSize);
-
-  // Http POST Content-Length is using UploadDataStream::size().
-  // It is the same as request.upload_data->GetContentLength().
-  scoped_ptr<UploadDataStream> stream(UploadDataStream::Create(
-      request.upload_data, NULL));
-  ASSERT_EQ(request.upload_data->GetContentLength(), stream->size());
-
-  scoped_ptr<spdy::SpdyFrame>
-      req(ConstructSpdyPost(request.upload_data->GetContentLength(), NULL, 0));
+  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyPost(kUploadDataSize, NULL, 0));
   scoped_ptr<spdy::SpdyFrame> body(ConstructSpdyBodyFrame(1, true));
   MockWrite writes[] = {
     CreateMockWrite(*req),
     CreateMockWrite(*body),
   };
 
-  // Response frames, send WINDOW_UPDATE first
   static const int kDeltaWindowSize = 0xff;
   scoped_ptr<spdy::SpdyFrame> window_update(
       ConstructSpdyWindowUpdate(1, kDeltaWindowSize));
@@ -1263,105 +1241,227 @@ TEST_P(SpdyNetworkTransactionTest, WindowUpdate) {
   scoped_refptr<DelayedSocketData> data(
       new DelayedSocketData(2, reads, arraysize(reads),
                             writes, arraysize(writes)));
-  session_deps.socket_factory.AddSocketDataProvider(data.get());
 
-  scoped_ptr<SpdyNetworkTransaction> trans(
-      new SpdyNetworkTransaction(session));
+  NormalSpdyTransactionHelper helper(CreatePostRequest(),
+                                     BoundNetLog(), GetParam());
+  helper.AddData(data.get());
+  helper.RunPreTestSetup();
 
-  TestCompletionCallback callback;
-  int rv = trans->Start(&request, &callback, BoundNetLog());
+  HttpNetworkTransaction* trans = helper.trans();
 
-  ASSERT_TRUE(trans->stream_ != NULL);
-  ASSERT_TRUE(trans->stream_->stream() != NULL);
-  EXPECT_EQ(spdy::kInitialWindowSize,
-            trans->stream_->stream()->send_window_size());
+  TestCompletionCallback callback;
+  int rv = trans->Start(&helper.request(), &callback, BoundNetLog());
 
   EXPECT_EQ(ERR_IO_PENDING, rv);
   rv = callback.WaitForResult();
   EXPECT_EQ(OK, rv);
 
-  ASSERT_TRUE(trans->stream_ != NULL);
-  ASSERT_TRUE(trans->stream_->stream() != NULL);
-  EXPECT_EQ(spdy::kInitialWindowSize + kDeltaWindowSize - kUploadDataSize,
-            trans->stream_->stream()->send_window_size());
-  EXPECT_TRUE(data->at_read_eof());
-  EXPECT_TRUE(data->at_write_eof());
-
+  // ASSERT_TRUE(trans->spdy_http_stream_ != NULL);
+  // ASSERT_TRUE(trans->spdy_http_stream_->stream() != NULL);
+  // EXPECT_EQ(spdy::kInitialWindowSize + kDeltaWindowSize - kUploadDataSize,
+  //           trans->spdy_http_stream_->stream()->send_window_size());
+  helper.VerifyDataConsumed();
   SpdySession::SetFlowControl(false);
 }
 
 // Test that WINDOW_UPDATE frame causing overflow is handled correctly.
+// Since WINDOW_UPDATEs should appear only when we're in the middle of a
+// long POST, we create a few full frame writes to force a WINDOW_UPDATE in
+// between.
 TEST_P(SpdyNetworkTransactionTest, WindowUpdateOverflow) {
-  SpdySessionDependencies session_deps;
-  scoped_refptr<HttpNetworkSession> session =
-      SpdySessionDependencies::SpdyCreateSession(&session_deps);
-
-  SpdySession::SetSSLMode(false);
   SpdySession::SetFlowControl(true);
 
-  // Setup the request
-  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(kUploadData, arraysize(kUploadData)-1);
+  // number of full frames we hope to write (but will not, used to
+  // set content-length header correctly)
+  static int kFrameCount = 3;
 
-  // Http POST Content-Length is using UploadDataStream::size().
-  // It is the same as request.upload_data->GetContentLength().
-  scoped_ptr<UploadDataStream> stream(UploadDataStream::Create(
-      request.upload_data, NULL));
-  ASSERT_EQ(request.upload_data->GetContentLength(), stream->size());
+  // Construct content for a data frame of maximum size.
+  scoped_ptr<std::string> content(
+      new std::string(kMaxSpdyFrameChunkSize, 'a'));
 
-  scoped_ptr<spdy::SpdyFrame>
-      req(ConstructSpdyPost(request.upload_data->GetContentLength(), NULL, 0));
-  scoped_ptr<spdy::SpdyFrame> body(ConstructSpdyBodyFrame(1, true));
+  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyPost(
+      kMaxSpdyFrameChunkSize * kFrameCount, NULL, 0));
+
+  scoped_ptr<spdy::SpdyFrame> body(
+      ConstructSpdyBodyFrame(1, content->c_str(), content->size(), false));
   scoped_ptr<spdy::SpdyFrame> rst(
       ConstructSpdyRstStream(1, spdy::FLOW_CONTROL_ERROR));
+
+  // We're not going to write a data frame with FIN, we'll receive a bad
+  // WINDOW_UPDATE while sending a request and will send a RST_STREAM frame.
   MockWrite writes[] = {
     CreateMockWrite(*req),
     CreateMockWrite(*body),
     CreateMockWrite(*rst),
   };
 
-  // Response frames, send WINDOW_UPDATE first
   static const int kDeltaWindowSize = 0x7fffffff;  // cause an overflow
   scoped_ptr<spdy::SpdyFrame> window_update(
       ConstructSpdyWindowUpdate(1, kDeltaWindowSize));
+  scoped_ptr<spdy::SpdyFrame> window_update2(
+      ConstructSpdyWindowUpdate(2, kDeltaWindowSize));
   scoped_ptr<spdy::SpdyFrame> reply(ConstructSpdyPostSynReply(NULL, 0));
+
+  // 1. Once we start writing a request, we do not read until the whole
+  // request is written completely, UNLESS, the first ReadSocket() call
+  // during the stream creation succeeded and caused another ReadSocket()
+  // to be scheduled.
+
+  // 2. We are trying to simulate WINDOW_UPDATE frame being received in the
+  // middle of a POST request's body being sent.
+
+  // In order to achieve 2, we force 1 to happen by giving
+  // DelayedSocketData a write delay of 0 and feeding SPDY dummy
+  // window_update2 frames which will be ignored, but will cause
+  // ReadSocket() to be scheduled and reads and writes to be interleaved,
+  // which will cause our window_update frame to be read right in the
+  // middle of a POST request being sent.
+
   MockRead reads[] = {
+    CreateMockRead(*window_update2),
+    CreateMockRead(*window_update2),
+    CreateMockRead(*window_update),
+    CreateMockRead(*window_update),
     CreateMockRead(*window_update),
-    CreateMockRead(*reply),
-    CreateMockRead(*body),
     MockRead(true, 0, 0)  // EOF
   };
 
   scoped_refptr<DelayedSocketData> data(
-      new DelayedSocketData(2, reads, arraysize(reads),
+      new DelayedSocketData(0, reads, arraysize(reads),
                             writes, arraysize(writes)));
-  session_deps.socket_factory.AddSocketDataProvider(data.get());
 
-  scoped_ptr<SpdyNetworkTransaction> trans(
-      new SpdyNetworkTransaction(session));
+  // Setup the request
+  HttpRequestInfo request;
+  request.method = "POST";
+  request.url = GURL("http://www.google.com/");
+  request.upload_data = new UploadData();
+  for (int i = 0; i < kFrameCount; ++i)
+    request.upload_data->AppendBytes(content->c_str(), content->size());
 
-  TestCompletionCallback callback;
-  int rv = trans->Start(&request, &callback, BoundNetLog());
+  NormalSpdyTransactionHelper helper(request,
+                                     BoundNetLog(), GetParam());
+  helper.AddData(data.get());
+  helper.RunPreTestSetup();
+
+  HttpNetworkTransaction* trans = helper.trans();
 
-  ASSERT_TRUE(trans->stream_ != NULL);
-  ASSERT_TRUE(trans->stream_->stream() != NULL);
-  EXPECT_EQ(spdy::kInitialWindowSize,
-            trans->stream_->stream()->send_window_size());
+  TestCompletionCallback callback;
+  int rv = trans->Start(&helper.request(), &callback, BoundNetLog());
 
   EXPECT_EQ(ERR_IO_PENDING, rv);
   rv = callback.WaitForResult();
   EXPECT_EQ(ERR_SPDY_PROTOCOL_ERROR, rv);
 
-  ASSERT_TRUE(session != NULL);
-  ASSERT_TRUE(session->spdy_session_pool() != NULL);
-  session->spdy_session_pool()->ClearSessions();
+  ASSERT_TRUE(helper.session() != NULL);
+  ASSERT_TRUE(helper.session()->spdy_session_pool() != NULL);
+  helper.session()->spdy_session_pool()->ClearSessions();
+  helper.VerifyDataConsumed();
 
-  EXPECT_TRUE(data->at_read_eof());
-  EXPECT_TRUE(data->at_write_eof());
+  SpdySession::SetFlowControl(false);
+}
+
+// Test that after hitting a send window size of 0, the write process
+// stalls and upon receiving WINDOW_UPDATE frame write resumes.
+//
+// This test constructs a POST request followed by enough data frames
+// containing 'a' that would make the window size 0, followed by another
+// data frame containing default content (which is "hello!") and this frame
+// also contains a FIN flag.  DelayedSocketData is used to enforce all
+// writes go through before a read could happen.  However, the last frame
+// ("hello!")  is not supposed to go through since by the time its turn
+// arrives, window size is 0.  At this point MessageLoop::Run() called via
+// callback would block.  Therefore we call MessageLoop::RunAllPending()
+// which returns after performing all possible writes.  We use DCHECKS to
+// ensure that last data frame is still there and stream has stalled.
+// After that, next read is artifically enforced, which causes a
+// WINDOW_UPDATE to be read and I/O process resumes.
+TEST_P(SpdyNetworkTransactionTest, FlowControlStallResume) {
+  SpdySession::SetFlowControl(true);
+
+  // Number of frames we need to send to zero out the window size: data
+  // frames plus SYN_STREAM plus the last data frame; also we need another
+  // data frame that we will send once the WINDOW_UPDATE is received,
+  // therefore +3.
+  size_t nwrites = spdy::kInitialWindowSize / kMaxSpdyFrameChunkSize + 3;
+
+  // Calculate last frame's size; 0 size data frame is legal.
+  size_t last_frame_size = spdy::kInitialWindowSize % kMaxSpdyFrameChunkSize;
+
+  // Construct content for a data frame of maximum size.
+  scoped_ptr<std::string> content(
+      new std::string(kMaxSpdyFrameChunkSize, 'a'));
+
+  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyPost(
+      spdy::kInitialWindowSize + kUploadDataSize, NULL, 0));
+
+  // Full frames.
+  scoped_ptr<spdy::SpdyFrame> body1(
+      ConstructSpdyBodyFrame(1, content->c_str(), content->size(), false));
+
+  // Last frame to zero out the window size.
+  scoped_ptr<spdy::SpdyFrame> body2(
+      ConstructSpdyBodyFrame(1, content->c_str(), last_frame_size, false));
+
+  // Data frame to be sent once WINDOW_UPDATE frame is received.
+  scoped_ptr<spdy::SpdyFrame> body3(ConstructSpdyBodyFrame(1, true));
+
+  // Fill in mock writes.
+  scoped_array<MockWrite> writes(new MockWrite[nwrites]);
+  size_t i = 0;
+  writes[i] = CreateMockWrite(*req);
+  for (i = 1; i < nwrites-2; i++)
+    writes[i] = CreateMockWrite(*body1);
+  writes[i++] = CreateMockWrite(*body2);
+  writes[i] = CreateMockWrite(*body3);
+
+  // Construct read frame, just give enough space to upload the rest of the
+  // data.
+  scoped_ptr<spdy::SpdyFrame> window_update(
+      ConstructSpdyWindowUpdate(1, kUploadDataSize));
+  scoped_ptr<spdy::SpdyFrame> reply(ConstructSpdyPostSynReply(NULL, 0));
+  MockRead reads[] = {
+    CreateMockRead(*window_update),
+    CreateMockRead(*window_update),
+    CreateMockRead(*reply),
+    CreateMockRead(*body2),
+    CreateMockRead(*body3),
+    MockRead(true, 0, 0)  // EOF
+  };
+
+  // Force all writes to happen before any read, last write will not
+  // actually queue a frame, due to window size being 0.
+  scoped_refptr<DelayedSocketData> data(
+      new DelayedSocketData(nwrites, reads, arraysize(reads),
+                            writes.get(), nwrites));
+
+  HttpRequestInfo request;
+  request.method = "POST";
+  request.url = GURL("http://www.google.com/");
+  request.upload_data = new UploadData();
+  scoped_ptr<std::string> upload_data(
+      new std::string(spdy::kInitialWindowSize, 'a'));
+  upload_data->append(kUploadData, kUploadDataSize);
+  request.upload_data->AppendBytes(upload_data->c_str(), upload_data->size());
+  NormalSpdyTransactionHelper helper(request,
+                                     BoundNetLog(), GetParam());
+  helper.AddData(data.get());
+  helper.RunPreTestSetup();
+
+  HttpNetworkTransaction* trans = helper.trans();
+
+  TestCompletionCallback callback;
+  int rv = trans->Start(&helper.request(), &callback, BoundNetLog());
+  EXPECT_EQ(ERR_IO_PENDING, rv);
+
+  MessageLoop::current()->RunAllPending(); // Write as much as we can.
+  // ASSERT_TRUE(trans->spdy_http_stream_ != NULL);
+  // ASSERT_TRUE(trans->spdy_http_stream_->stream() != NULL);
+  // EXPECT_EQ(0, trans->spdy_http_stream_->stream()->send_window_size());
+  // EXPECT_FALSE(trans->spdy_http_stream_->IsFinishedSendingBody());
+
+  data->ForceNextRead();   // Read in WINDOW_UPDATE frame.
+  rv = callback.WaitForResult();
+  helper.VerifyDataConsumed();
 
   SpdySession::SetFlowControl(false);
 }
diff --git a/net/spdy/spdy_session.cc b/net/spdy/spdy_session.cc
index 3e4d049..4f6504d 100644
--- a/net/spdy/spdy_session.cc
+++ b/net/spdy/spdy_session.cc
@@ -428,14 +428,18 @@ int SpdySession::WriteStreamData(spdy::SpdyStreamId stream_id,
 
   if (len > kMaxSpdyFrameChunkSize) {
     len = kMaxSpdyFrameChunkSize;
-    flags = spdy::DATA_FLAG_NONE;
+    flags = static_cast<spdy::SpdyDataFlags>(flags & ~spdy::DATA_FLAG_FIN);
   }
 
   // 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());
+    int new_len = std::min(len, stream->send_window_size());
+    if (new_len < len) {
+      len = new_len;
+      flags = static_cast<spdy::SpdyDataFlags>(flags & ~spdy::DATA_FLAG_FIN);
+    }
     stream->DecreaseSendWindowSize(len);
   }
 
diff --git a/net/spdy/spdy_stream.cc b/net/spdy/spdy_stream.cc
index 5fd5f14..fb5692a4 100644
--- a/net/spdy/spdy_stream.cc
+++ b/net/spdy/spdy_stream.cc
@@ -16,6 +16,7 @@ SpdyStream::SpdyStream(
     : continue_buffering_data_(true),
       stream_id_(stream_id),
       priority_(0),
+      window_update_write_pending_(false),
       send_window_size_(spdy::kInitialWindowSize),
       pushed_(pushed),
       metrics_(Singleton<BandwidthMetrics>::get()),
@@ -88,6 +89,12 @@ void SpdyStream::IncreaseSendWindowSize(int delta_window_size) {
   DCHECK_GE(delta_window_size, 1);
   int new_window_size = send_window_size_ + delta_window_size;
 
+  // We shouldn't be receving WINDOW_UPDATE before or after that state,
+  // since before means we've not written SYN_STREAM yet, and after means
+  // we've written a DATA frame with FIN bit.
+  if (io_state_ != STATE_SEND_BODY_COMPLETE)
+    return;
+
   // 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
@@ -105,6 +112,20 @@ void SpdyStream::IncreaseSendWindowSize(int delta_window_size) {
             << " send_window_size_ [current:" << send_window_size_ << "]"
             << " by " << delta_window_size << " bytes";
   send_window_size_ = new_window_size;
+
+  // If the stream was stalled due to consumed window size, restart the
+  // I/O loop.
+  if (!response_complete_ && !delegate_->IsFinishedSendingBody()) {
+    // Don't start the I/O loop for every WINDOW_UPDATE frame received,
+    // since we may receive many of them before the "write" due to first
+    // received WINDOW_UPDATE is completed.
+    if (window_update_write_pending_)
+      return;
+
+    window_update_write_pending_ = true;
+    io_state_ = STATE_SEND_BODY;
+    DoLoop(OK);
+  }
 }
 
 void SpdyStream::DecreaseSendWindowSize(int delta_window_size) {
@@ -221,6 +242,10 @@ void SpdyStream::OnWriteComplete(int status) {
   // TODO(mbelshe): Check for cancellation here.  If we're cancelled, we
   // should discontinue the DoLoop.
 
+  // Clear it just in case this write lead to a 0 size send window, so that
+  // incoming window updates will cause a write to be scheduled again.
+  window_update_write_pending_ = false;
+
   // It is possible that this stream was closed while we had a write pending.
   if (response_complete_)
     return;
@@ -425,7 +450,8 @@ int SpdyStream::DoSendBodyComplete(int result) {
   if (!delegate_)
     return ERR_UNEXPECTED;
 
-  if (!delegate_->OnSendBodyComplete(result))
+  delegate_->OnSendBodyComplete(result);
+  if (!delegate_->IsFinishedSendingBody())
     io_state_ = STATE_SEND_BODY;
   else
     io_state_ = STATE_READ_HEADERS;
diff --git a/net/spdy/spdy_stream.h b/net/spdy/spdy_stream.h
index cad23f2..1745867 100644
--- a/net/spdy/spdy_stream.h
+++ b/net/spdy/spdy_stream.h
@@ -48,8 +48,10 @@ class SpdyStream : public base::RefCounted<SpdyStream> {
 
     // Called when data has been sent. |status| indicates network error
     // or number of bytes has been sent.
+    virtual void OnSendBodyComplete(int status) = 0;
+
     // Returns true if no more data to be sent.
-    virtual bool OnSendBodyComplete(int status) = 0;
+    virtual bool IsFinishedSendingBody() = 0;
 
     // Called when SYN_REPLY received. |status| indicates network error.
     // Returns network error code.
@@ -223,6 +225,10 @@ class SpdyStream : public base::RefCounted<SpdyStream> {
   spdy::SpdyStreamId stream_id_;
   std::string path_;
   int priority_;
+
+  // Tracks when a window update has already triggered a write, but
+  // the write has not yet completed.
+  bool window_update_write_pending_;
   int send_window_size_;
 
   const bool pushed_;
diff --git a/net/spdy/spdy_stream_unittest.cc b/net/spdy/spdy_stream_unittest.cc
index 94b08ec..03a5180 100644
--- a/net/spdy/spdy_stream_unittest.cc
+++ b/net/spdy/spdy_stream_unittest.cc
@@ -55,10 +55,14 @@ class TestSpdyStreamDelegate : public SpdyStream::Delegate {
     ADD_FAILURE() << "OnSendBody should not be called";
     return ERR_UNEXPECTED;
   }
-  virtual bool OnSendBodyComplete(int status) {
+  virtual void OnSendBodyComplete(int status) {
     ADD_FAILURE() << "OnSendBodyComplete should not be called";
+  }
+
+  virtual bool IsFinishedSendingBody() {
     return true;
   }
+
   virtual int OnResponseReceived(const spdy::SpdyHeaderBlock& response,
                                  base::Time response_time,
                                  int status) {
diff --git a/net/spdy/spdy_test_util.cc b/net/spdy/spdy_test_util.cc
index 56d45ec..35e9968 100644
--- a/net/spdy/spdy_test_util.cc
+++ b/net/spdy/spdy_test_util.cc
@@ -599,12 +599,20 @@ spdy::SpdyFrame* ConstructSpdyPostSynReply(const char* const extra_headers[],
                                    arraysize(kStandardGetHeaders));
 }
 
-// Constructs a single SPDY data frame with the contents "hello!"
+// Constructs a single SPDY data frame with the default contents.
 spdy::SpdyFrame* ConstructSpdyBodyFrame(int stream_id, bool fin) {
   spdy::SpdyFramer framer;
-  return
-      framer.CreateDataFrame(stream_id, "hello!", 6,
-                             fin ? spdy::DATA_FLAG_FIN : spdy::DATA_FLAG_NONE);
+  return framer.CreateDataFrame(
+      stream_id, kUploadData, kUploadDataSize,
+      fin ? spdy::DATA_FLAG_FIN : spdy::DATA_FLAG_NONE);
+}
+
+// Constructs a single SPDY data frame with the given content.
+spdy::SpdyFrame* ConstructSpdyBodyFrame(int stream_id, const char* data,
+                                        uint32 len, bool fin) {
+  spdy::SpdyFramer framer;
+  return framer.CreateDataFrame(
+      stream_id, data, len, fin ? spdy::DATA_FLAG_FIN : spdy::DATA_FLAG_NONE);
 }
 
 // Construct an expected SPDY reply string.
diff --git a/net/spdy/spdy_test_util.h b/net/spdy/spdy_test_util.h
index ec60a30..e7df3f67 100644
--- a/net/spdy/spdy_test_util.h
+++ b/net/spdy/spdy_test_util.h
@@ -23,6 +23,12 @@
 
 namespace net {
 
+// Default upload data used by both, mock objects and framer when creating
+// data frames.
+const char kDefaultURL[] = "http://www.google.com";
+const char kUploadData[] = "hello!";
+const int kUploadDataSize = arraysize(kUploadData)-1;
+
 // NOTE: In GCC, on a Mac, this can't be in an anonymous namespace!
 // This struct holds information used to construct spdy control and data frames.
 struct SpdyHeaderInfo {
@@ -250,6 +256,10 @@ spdy::SpdyFrame* ConstructSpdyPostSynReply(const char* const extra_headers[],
 spdy::SpdyFrame* ConstructSpdyBodyFrame(int stream_id,
                                         bool fin);
 
+// Constructs a single SPDY data frame with the given content.
+spdy::SpdyFrame* ConstructSpdyBodyFrame(int stream_id, const char* data,
+                                        uint32 len, bool fin);
+
 // Create an async MockWrite from the given SpdyFrame.
 MockWrite CreateMockWrite(const spdy::SpdyFrame& req);