Let QUIC streams write 16k before ceding. Behind FLAG_quic_batch_writes.

net/quic/quic_session_test.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/quic/quic_session.h"
 
 #include <set>
 
 #include "base/basictypes.h"
 #include "base/containers/hash_tables.h"
@@ skipping 78 matching lines @@
       : QuicSpdyStream(id, session) {}
 
   using ReliableQuicStream::CloseWriteSide;
 
   void OnDataAvailable() override {}
 
   void SendBody(const string& data, bool fin) {
     WriteOrBufferData(data, fin, nullptr);
   }
 
+  using QuicSpdyStream::set_priority;
+
   MOCK_METHOD0(OnCanWrite, void());
 };
 
 // Poor man's functor for use as callback in a mock.
 class StreamBlocker {
  public:
   StreamBlocker(QuicSession* session, QuicStreamId stream_id)
       : session_(session),
         stream_id_(stream_id) {
   }
 
   void MarkConnectionLevelWriteBlocked() {
     session_->MarkConnectionLevelWriteBlocked(stream_id_, kSomeMiddlePriority);
   }
 
+  void MarkHighPriorityWriteBlocked() {
+    session_->MarkConnectionLevelWriteBlocked(stream_id_, kHighestPriority);
+  }
+
  private:
   QuicSession* const session_;
   const QuicStreamId stream_id_;
 };
 
 class TestSession : public QuicSpdySession {
  public:
   explicit TestSession(QuicConnection* connection)
       : QuicSpdySession(connection, DefaultQuicConfig()),
         crypto_stream_(this),
@@ skipping 27 matching lines @@
     return QuicSpdySession::GetOrCreateDynamicStream(stream_id);
   }
 
   QuicConsumedData WritevData(
       QuicStreamId id,
       QuicIOVector data,
       QuicStreamOffset offset,
       bool fin,
       FecProtection fec_protection,
       QuicAckListenerInterface* ack_notifier_delegate) override {
-    // Always consumes everything.
-    if (writev_consumes_all_data_) {
-      return QuicConsumedData(data.total_length, fin);
-    } else {
-      return QuicSession::WritevData(id, data, offset, fin, fec_protection,
-                                     ack_notifier_delegate);
+    QuicConsumedData consumed(data.total_length, fin);
+    if (!writev_consumes_all_data_) {
+      consumed = QuicSession::WritevData(id, data, offset, fin, fec_protection,
+                                         ack_notifier_delegate);
     }
+    QuicSessionPeer::GetWriteBlockedStreams(this)->UpdateBytesForStream(
+        id, consumed.bytes_consumed);
+    return consumed;
   }
 
   void set_writev_consumes_all_data(bool val) {
     writev_consumes_all_data_ = val;
   }
 
   QuicConsumedData SendStreamData(QuicStreamId id) {
     struct iovec iov;
     return WritevData(id, MakeIOVector("not empty", &iov), 0, true,
                       MAY_FEC_PROTECT, nullptr);
   }
 
+  QuicConsumedData SendLargeFakeData(QuicStreamId id, int bytes) {
+    DCHECK(writev_consumes_all_data_);
+    struct iovec iov;
+    iov.iov_base = nullptr;  // should not be read.
+    iov.iov_len = static_cast<size_t>(bytes);
+    return WritevData(id, QuicIOVector(&iov, 1, bytes), 0, true,
+                      MAY_FEC_PROTECT, nullptr);
+  }
+
   using QuicSession::PostProcessAfterData;
 
  private:
   StrictMock<TestCryptoStream> crypto_stream_;
 
   bool writev_consumes_all_data_;
 };
 
 class QuicSessionTestBase : public ::testing::TestWithParam<QuicVersion> {
  protected:
@@ skipping 193 matching lines @@
   TestStream* stream2 = session_.CreateOutgoingDynamicStream();
   TestStream* stream4 = session_.CreateOutgoingDynamicStream();
   TestStream* stream6 = session_.CreateOutgoingDynamicStream();
 
   session_.MarkConnectionLevelWriteBlocked(stream2->id(), kSomeMiddlePriority);
   session_.MarkConnectionLevelWriteBlocked(stream6->id(), kSomeMiddlePriority);
   session_.MarkConnectionLevelWriteBlocked(stream4->id(), kSomeMiddlePriority);
 
   InSequence s;
   StreamBlocker stream2_blocker(&session_, stream2->id());
-  // Reregister, to test the loop limit.
-  EXPECT_CALL(*stream2, OnCanWrite())
-      .WillOnce(Invoke(&stream2_blocker,
-                       &StreamBlocker::MarkConnectionLevelWriteBlocked));
-  EXPECT_CALL(*stream6, OnCanWrite());
-  EXPECT_CALL(*stream4, OnCanWrite());
+
+  if (FLAGS_quic_batch_writes) {
+    // Reregister, to test the loop limit.
+    EXPECT_CALL(*stream2, OnCanWrite())
+        .WillOnce(Invoke(&stream2_blocker,
+                         &StreamBlocker::MarkConnectionLevelWriteBlocked));
+    // 2 will get called a second time as it didn't finish its block
+    EXPECT_CALL(*stream2, OnCanWrite());
+    EXPECT_CALL(*stream6, OnCanWrite());
+    // 4 will not get called, as we exceeded the loop limit.
+  } else {
+    // Reregister, to test the loop limit.
+    EXPECT_CALL(*stream2, OnCanWrite())
+        .WillOnce(Invoke(&stream2_blocker,
+                         &StreamBlocker::MarkConnectionLevelWriteBlocked));
+    EXPECT_CALL(*stream6, OnCanWrite());
+    EXPECT_CALL(*stream4, OnCanWrite());
+  }
   session_.OnCanWrite();
   EXPECT_TRUE(session_.WillingAndAbleToWrite());
 }
 
+TEST_P(QuicSessionTestServer, TestBatchedWrites) {
+  FLAGS_quic_batch_writes = true;
+  TestStream* stream2 = session_.CreateOutgoingDynamicStream();
+  TestStream* stream4 = session_.CreateOutgoingDynamicStream();
+  TestStream* stream6 = session_.CreateOutgoingDynamicStream();
+
+  session_.set_writev_consumes_all_data(true);
+  session_.MarkConnectionLevelWriteBlocked(stream2->id(), kSomeMiddlePriority);
+  session_.MarkConnectionLevelWriteBlocked(stream4->id(), kSomeMiddlePriority);
+
+  StreamBlocker stream2_blocker(&session_, stream2->id());
+  StreamBlocker stream4_blocker(&session_, stream4->id());
+  StreamBlocker stream6_blocker(&session_, stream6->id());
+  // With two sessions blocked, we should get two write calls.  They should both
+  // go to the first stream as it will only write 6k and mark itself blocked
+  // again.
+  InSequence s;
+  EXPECT_CALL(*stream2, OnCanWrite())
+      .WillOnce(DoAll(testing::IgnoreResult(Invoke(CreateFunctor(
+          &session_, &TestSession::SendLargeFakeData, stream2->id(), 6000))),
+                      Invoke(&stream2_blocker,
+                             &StreamBlocker::MarkConnectionLevelWriteBlocked)));
+  EXPECT_CALL(*stream2, OnCanWrite())
+      .WillOnce(DoAll(testing::IgnoreResult(Invoke(CreateFunctor(
+          &session_, &TestSession::SendLargeFakeData, stream2->id(), 6000))),
+                      Invoke(&stream2_blocker,
+                             &StreamBlocker::MarkConnectionLevelWriteBlocked)));
+  session_.OnCanWrite();
+
+  // We should get one more call for stream2, at which point it has used its
+  // write quota and we move over to stream 4.
+  EXPECT_CALL(*stream2, OnCanWrite())
+      .WillOnce(DoAll(testing::IgnoreResult(Invoke(CreateFunctor(
+          &session_, &TestSession::SendLargeFakeData, stream2->id(), 6000))),
+                      Invoke(&stream2_blocker,
+                             &StreamBlocker::MarkConnectionLevelWriteBlocked)));
+  EXPECT_CALL(*stream4, OnCanWrite())
+      .WillOnce(DoAll(testing::IgnoreResult(Invoke(CreateFunctor(
+          &session_, &TestSession::SendLargeFakeData, stream4->id(), 6000))),
+                      Invoke(&stream4_blocker,
+                             &StreamBlocker::MarkConnectionLevelWriteBlocked)));
+  session_.OnCanWrite();
+
+  // Now let stream 4 do the 2nd of its 3 writes, but add a block for a high
+  // priority stream 6.  4 should be preempted.  6 will write but *not* block so
+  // will cede back to 4.
+  stream6->set_priority(kHighestPriority);
+  EXPECT_CALL(*stream4, OnCanWrite())
+      .WillOnce(DoAll(testing::IgnoreResult(Invoke(CreateFunctor(
+          &session_, &TestSession::SendLargeFakeData, stream4->id(), 6000))),
+                      Invoke(&stream4_blocker,
+                             &StreamBlocker::MarkConnectionLevelWriteBlocked),
+                      Invoke(&stream6_blocker,
+                             &StreamBlocker::MarkHighPriorityWriteBlocked)));
+  EXPECT_CALL(*stream6, OnCanWrite())
+      .WillOnce(testing::IgnoreResult(Invoke(CreateFunctor(
+          &session_, &TestSession::SendLargeFakeData, stream4->id(), 6000))));
+    session_.OnCanWrite();
+
+  // Stream4 alread did 6k worth of writes, so after doing another 12k it should
+  // cede and 2 should resume.
+  EXPECT_CALL(*stream4, OnCanWrite())
+      .WillOnce(DoAll(testing::IgnoreResult(Invoke(CreateFunctor(
+          &session_, &TestSession::SendLargeFakeData, stream4->id(), 12000))),
+                      Invoke(&stream4_blocker,
+                             &StreamBlocker::MarkConnectionLevelWriteBlocked)));
+  EXPECT_CALL(*stream2, OnCanWrite())
+      .WillOnce(DoAll(testing::IgnoreResult(Invoke(CreateFunctor(
+          &session_, &TestSession::SendLargeFakeData, stream2->id(), 6000))),
+                      Invoke(&stream2_blocker,
+                             &StreamBlocker::MarkConnectionLevelWriteBlocked)));
+  session_.OnCanWrite();
+}
+
 TEST_P(QuicSessionTestServer, OnCanWriteBundlesStreams) {
   // Drive congestion control manually.
   MockSendAlgorithm* send_algorithm = new StrictMock<MockSendAlgorithm>;
   QuicConnectionPeer::SetSendAlgorithm(session_.connection(), send_algorithm);
 
   TestStream* stream2 = session_.CreateOutgoingDynamicStream();
   TestStream* stream4 = session_.CreateOutgoingDynamicStream();
   TestStream* stream6 = session_.CreateOutgoingDynamicStream();
 
   session_.MarkConnectionLevelWriteBlocked(stream2->id(), kSomeMiddlePriority);
@@ skipping 96 matching lines @@
   InSequence s;
   // Force most streams to re-register, which is common scenario when we block
   // the Crypto stream, and only the crypto stream can "really" write.
 
   // Due to prioritization, we *should* be asked to write the crypto stream
   // first.
   // Don't re-register the crypto stream (which signals complete writing).
   TestCryptoStream* crypto_stream = session_.GetCryptoStream();
   EXPECT_CALL(*crypto_stream, OnCanWrite());
 
-  // Re-register all other streams, to show they weren't able to proceed.
-  EXPECT_CALL(*stream2, OnCanWrite())
-      .WillOnce(Invoke(&stream2_blocker,
-                       &StreamBlocker::MarkConnectionLevelWriteBlocked));
-  EXPECT_CALL(*stream3, OnCanWrite())
-      .WillOnce(Invoke(&stream3_blocker,
-                       &StreamBlocker::MarkConnectionLevelWriteBlocked));
+  EXPECT_CALL(*stream2, OnCanWrite());
+  EXPECT_CALL(*stream3, OnCanWrite());
   EXPECT_CALL(*stream4, OnCanWrite())
       .WillOnce(Invoke(&stream4_blocker,
                        &StreamBlocker::MarkConnectionLevelWriteBlocked));
 
   session_.OnCanWrite();
   EXPECT_TRUE(session_.WillingAndAbleToWrite());
   EXPECT_FALSE(session_.HasPendingHandshake());  // Crypto stream wrote.
 }
 
 TEST_P(QuicSessionTestServer, OnCanWriteWithClosedStream) {
@@ skipping 552 matching lines @@
   // Verify that there is no entry for the stream in
   // locally_closed_streams_highest_offset_.
   EXPECT_EQ(
       FLAGS_quic_fix_fin_accounting ? 0u : 1u,
       QuicSessionPeer::GetLocallyClosedStreamsHighestOffset(&session_).size());
 }
 
 }  // namespace
 }  // namespace test
 }  // namespace net