SPDY - implement greedy approach to read all the data and process it

net/spdy/spdy_session_spdy2_unittest.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 "base/memory/scoped_ptr.h"
+#include "base/memory/scoped_vector.h"
+#include "base/string_util.h"
 #include "net/base/cert_test_util.h"
 #include "net/base/host_cache.h"
+#include "net/base/io_buffer.h"
 #include "net/base/ip_endpoint.h"
 #include "net/base/net_log_unittest.h"
 #include "net/base/test_data_directory.h"
+#include "net/base/test_data_stream.h"
 #include "net/spdy/spdy_io_buffer.h"
 #include "net/spdy/spdy_session_pool.h"
 #include "net/spdy/spdy_stream.h"
 #include "net/spdy/spdy_test_util_spdy2.h"
 #include "testing/platform_test.h"
 
 using namespace net::test_spdy2;
 
 namespace net {
 
@@ skipping 139 matching lines @@
 
   scoped_refptr<TransportSocketParams> transport_params_;
   SpdySessionDependencies session_deps_;
   scoped_refptr<HttpNetworkSession> http_session_;
   SpdySessionPool* spdy_session_pool_;
   GURL test_url_;
   HostPortPair test_host_port_pair_;
   HostPortProxyPair pair_;
 };
 
+class SpdySessionTaskObserver : public MessageLoop::TaskObserver {
+ public:
+  explicit SpdySessionTaskObserver(const std::string& function_name)
+      : function_name_(function_name),
+        posted_(false) {
+    MessageLoop::current()->AddTaskObserver(this);
+  }
+
+  virtual ~SpdySessionTaskObserver() {
+    MessageLoop::current()->RemoveTaskObserver(this);
+  }
+
+  virtual void WillProcessTask(const base::PendingTask& pending_task) OVERRIDE {
+  }
+
+  virtual void DidProcessTask(const base::PendingTask& pending_task) OVERRIDE {
+    if (!posted_) {
+      posted_ = EndsWith(pending_task.posted_from.function_name(),
+                         function_name_, true);
+    }
+  }
+
+  bool posted() const { return posted_; }
+
+ private:
+  std::string function_name_;
+  bool posted_;

[Ryan Hamilton, 2013/02/05 05:03:45]

I think I might be inclined to make this a count instead of a bool.

[ramant (doing other things), 2013/02/05 20:43:43]

Done.

+};
+
 // Test the SpdyIOBuffer class.
 TEST_F(SpdySessionSpdy2Test, SpdyIOBuffer) {
   std::priority_queue<SpdyIOBuffer> queue_;
   const size_t kQueueSize = 100;
 
   // Insert items with random priority and increasing buffer size.
   for (size_t index = 0; index < kQueueSize; ++index) {
     queue_.push(SpdyIOBuffer(
         new IOBufferWithSize(index + 1),
         index + 1,
@@ skipping 1434 matching lines @@
   EXPECT_EQ(OK, session->InitializeWithSocket(connection.release(), true, OK));
 
   EXPECT_FALSE(session->NeedsCredentials());
 
   // Flush the SpdySession::OnReadComplete() task.
   MessageLoop::current()->RunUntilIdle();
 
   spdy_session_pool_->Remove(session);
 }
 
+TEST_F(SpdySessionSpdy2Test, SyncRead) {
+  MockConnect connect_data(SYNCHRONOUS, OK);
+  BufferedSpdyFramer framer(2, false);
+
+  scoped_ptr<SpdyFrame> req1(ConstructSpdyGet(NULL, 0, false, 1, MEDIUM));
+  MockWrite writes[] = {
+    CreateMockWrite(*req1, 0),
+  };
+
+  const int kPayloadSize = 1600;
+  TestDataStream test_stream;
+  scoped_refptr<net::IOBuffer> payload(new net::IOBuffer(kPayloadSize));
+  char* payload_data = payload->data();
+  test_stream.GetBytes(payload_data, kPayloadSize);
+
+  scoped_ptr<SpdyFrame> data_frame1(
+      framer.CreateDataFrame(1, payload_data, kPayloadSize, DATA_FLAG_NONE));
+  scoped_ptr<SpdyFrame> data_frame(
+      framer.CreateDataFrame(1, payload_data, kPayloadSize, DATA_FLAG_FIN));
+
+  scoped_ptr<SpdyFrame> resp1(ConstructSpdyGetSynReply(NULL, 0, 1));
+
+  size_t num_reads = 6;
+  scoped_array<MockRead> reads(new MockRead[num_reads]);
+  size_t i = 0;
+  reads[i] = CreateMockRead(*resp1, 1);
+  i = 1;
+  reads[i] = CreateMockRead(*data_frame1, 2);
+  for (i = 2; i < num_reads - 2; ++i)
+    reads[i] = CreateMockRead(*data_frame1, i + 1, SYNCHRONOUS);
+  reads[i] = CreateMockRead(*data_frame, i + 1, SYNCHRONOUS);

[Ryan Hamilton, 2013/02/05 05:03:45]

Any reason not to define the reads array with the normal:

MockReads reads[] = { ... };

pattern?

[ramant (doing other things), 2013/02/05 20:43:43]

Done.

+  ++i;
+  reads[i] = MockRead(ASYNC, 0, i + 1);  // EOF
+
+  DeterministicSocketData data(reads.get(), num_reads,
+                               writes, arraysize(writes));
+  data.set_connect_data(connect_data);
+  session_deps_.host_resolver->set_synchronous_mode(true);
+  session_deps_.deterministic_socket_factory->AddSocketDataProvider(&data);
+
+  SSLSocketDataProvider ssl(SYNCHRONOUS, OK);
+  session_deps_.deterministic_socket_factory->AddSSLSocketDataProvider(&ssl);
+
+  CreateDeterministicNetworkSession();
+
+  scoped_refptr<SpdySession> session = CreateInitializedSession();
+
+  scoped_refptr<SpdyStream> spdy_stream1;
+  TestCompletionCallback callback1;
+  GURL url1("http://www.google.com");
+  EXPECT_EQ(OK, session->CreateStream(url1, MEDIUM, &spdy_stream1,
+                                      BoundNetLog(), callback1.callback()));
+  EXPECT_EQ(0u, spdy_stream1->stream_id());
+
+  scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock);
+  (*headers)["method"] = "GET";
+  (*headers)["scheme"] = url1.scheme();
+  (*headers)["host"] = url1.host();
+  (*headers)["url"] = url1.path();
+  (*headers)["version"] = "HTTP/1.1";
+
+  spdy_stream1->set_spdy_headers(headers.Pass());
+  EXPECT_TRUE(spdy_stream1->HasUrl());
+  spdy_stream1->SendRequest(false);
+
+  SpdySessionTaskObserver observer("DoRead");

[Ryan Hamilton, 2013/02/05 05:03:45]

Only "DoRead"?  is there a way to check for the SpdySession part too?

[ramant (doing other things), 2013/02/05 20:43:43]

Done.

+
+  // Run until 1st read.
+  EXPECT_EQ(0u, spdy_stream1->stream_id());
+  data.RunFor(2);
+  EXPECT_EQ(1u, spdy_stream1->stream_id());
+  EXPECT_FALSE(observer.posted());
+
+  // Do 4 reads.
+  data.RunFor(4);
+  EXPECT_FALSE(observer.posted());
+  EXPECT_TRUE(data.at_write_eof());
+  EXPECT_TRUE(data.at_read_eof());
+}
+
+TEST_F(SpdySessionSpdy2Test, ASyncRead) {
+  MockConnect connect_data(SYNCHRONOUS, OK);
+  BufferedSpdyFramer framer(2, false);
+
+  scoped_ptr<SpdyFrame> req1(ConstructSpdyGet(NULL, 0, false, 1, MEDIUM));
+  MockWrite writes[] = {
+    CreateMockWrite(*req1, 0),
+  };
+
+  // Build buffers of size 8k.
+  ASSERT_EQ(32 * 1024, kMaxReadBytes);
+  const int kPayloadSize = (kMaxReadBytes / 4) - SpdyDataFrame::size();
+  TestDataStream test_stream;
+  scoped_refptr<net::IOBuffer> payload(new net::IOBuffer(kPayloadSize));
+  char* payload_data = payload->data();
+  test_stream.GetBytes(payload_data, kPayloadSize);
+
+  scoped_ptr<SpdyFrame> data_frame1(
+      framer.CreateDataFrame(1, payload_data, kPayloadSize, DATA_FLAG_NONE));
+  scoped_ptr<SpdyFrame> data_frame(
+      framer.CreateDataFrame(1, payload_data, kPayloadSize, DATA_FLAG_FIN));
+
+  scoped_ptr<SpdyFrame> resp1(ConstructSpdyGetSynReply(NULL, 0, 1));
+
+  // Send twice as many bytes as kMaxReadBytes.
+  size_t num_reads = 2 + (2 * kMaxReadBytes / kPayloadSize);
+  scoped_array<MockRead> reads(new MockRead[num_reads]);
+  size_t i = 0;
+  reads[i] = CreateMockRead(*resp1, 1);
+  i = 1;
+  reads[i] = CreateMockRead(*data_frame1, i + 1);
+  for (i = 2; i < num_reads - 2; ++i)
+    reads[i] = CreateMockRead(*data_frame1, i + 1, SYNCHRONOUS);
+  reads[i] = CreateMockRead(*data_frame, i + 1, SYNCHRONOUS);
+  ++i;
+  reads[i] = MockRead(ASYNC, 0, i + 1);  // EOF
+
+  DeterministicSocketData data(reads.get(), num_reads,
+                               writes, arraysize(writes));
+  data.set_connect_data(connect_data);
+  session_deps_.host_resolver->set_synchronous_mode(true);
+  session_deps_.deterministic_socket_factory->AddSocketDataProvider(&data);
+
+  SSLSocketDataProvider ssl(SYNCHRONOUS, OK);
+  session_deps_.deterministic_socket_factory->AddSSLSocketDataProvider(&ssl);
+
+  CreateDeterministicNetworkSession();
+
+  scoped_refptr<SpdySession> session = CreateInitializedSession();
+
+  scoped_refptr<SpdyStream> spdy_stream1;
+  TestCompletionCallback callback1;
+  GURL url1("http://www.google.com");
+  EXPECT_EQ(OK, session->CreateStream(url1, MEDIUM, &spdy_stream1,
+                                      BoundNetLog(), callback1.callback()));
+  EXPECT_EQ(0u, spdy_stream1->stream_id());
+
+  scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock);
+  (*headers)["method"] = "GET";
+  (*headers)["scheme"] = url1.scheme();
+  (*headers)["host"] = url1.host();
+  (*headers)["url"] = url1.path();
+  (*headers)["version"] = "HTTP/1.1";
+
+  spdy_stream1->set_spdy_headers(headers.Pass());
+  EXPECT_TRUE(spdy_stream1->HasUrl());
+  spdy_stream1->SendRequest(false);
+
+  SpdySessionTaskObserver observer("DoRead");
+
+  // Run until 1st read.
+  EXPECT_EQ(0u, spdy_stream1->stream_id());
+  data.RunFor(2);
+  EXPECT_EQ(1u, spdy_stream1->stream_id());
+  EXPECT_FALSE(observer.posted());
+
+  // Do 8 reads.
+  data.RunFor(8);
+  EXPECT_TRUE(observer.posted());
+  EXPECT_TRUE(data.at_write_eof());
+  EXPECT_TRUE(data.at_read_eof());
+}
+
 }  // namespace net