Implement Drain() on HttpPipelinedStream.

BUG=102385
TEST=net_unittests


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

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

net/http/http_pipelined_connection_impl.cc

@@ -9,6 +9,8 @@
 #include "net/base/io_buffer.h"
 #include "net/http/http_pipelined_stream.h"
 #include "net/http/http_request_info.h"
+#include "net/http/http_response_body_drainer.h"
+#include "net/http/http_response_headers.h"
 #include "net/http/http_stream_parser.h"
 #include "net/socket/client_socket_handle.h"
 
@@ -606,6 +608,22 @@ void HttpPipelinedConnectionImpl::GetSSLCertRequestInfo(
       cert_request_info);
 }
 
+void HttpPipelinedConnectionImpl::Drain(HttpPipelinedStream* stream,
+                                        HttpNetworkSession* session) {
+  HttpResponseHeaders* headers = stream->GetResponseInfo()->headers;
+  if (!stream->CanFindEndOfResponse() || headers->IsChunkEncoded() ||
+      !usable_) {
+    // TODO(simonjam): Drain chunk-encoded responses if they're relatively
+    // common.
+    stream->Close(true);
+    delete stream;
+    return;
+  }
+  HttpResponseBodyDrainer* drainer = new HttpResponseBodyDrainer(stream);
+  drainer->StartWithSize(session, headers->GetContentLength());
+  // |drainer| will delete itself when done.
+}
+
 void HttpPipelinedConnectionImpl::QueueUserCallback(
     int pipeline_id,
     OldCompletionCallback* callback,

net/http/http_pipelined_connection_impl.h

@@ -28,6 +28,7 @@ namespace net {
 
 class ClientSocketHandle;
 class GrowableIOBuffer;
+class HttpNetworkSession;
 class HttpRequestHeaders;
 class HttpResponseInfo;
 class IOBuffer;
@@ -117,6 +118,10 @@ class NET_EXPORT_PRIVATE HttpPipelinedConnectionImpl
   void GetSSLCertRequestInfo(int pipeline_id,
                              SSLCertRequestInfo* cert_request_info);
 
+  // Attempts to drain the response body for |stream| so that the pipeline may
+  // be reused.
+  void Drain(HttpPipelinedStream* stream, HttpNetworkSession* session);
+
  private:
   enum StreamState {
     STREAM_CREATED,

net/http/http_pipelined_connection_impl_unittest.cc

@@ -1086,6 +1086,131 @@ TEST_F(HttpPipelinedConnectionImplTest, CloseBeforeReadCallbackRuns) {
   MessageLoop::current()->RunAllPending();
 }
 
+TEST_F(HttpPipelinedConnectionImplTest, RecoverFromDrainOnRedirect) {
+  MockWrite writes[] = {
+    MockWrite(false, 0, "GET /redirect.html HTTP/1.1\r\n\r\n"),
+    MockWrite(false, 1, "GET /ok.html HTTP/1.1\r\n\r\n"),
+  };
+  MockRead reads[] = {
+    MockRead(false, 2,
+             "HTTP/1.1 302 OK\r\n"
+             "Content-Length: 8\r\n\r\n"
+             "redirect"),
+    MockRead(false, 3,
+             "HTTP/1.1 200 OK\r\n"
+             "Content-Length: 7\r\n\r\n"
+             "ok.html"),
+  };
+  Initialize(reads, arraysize(reads), writes, arraysize(writes));
+
+  scoped_ptr<HttpStream> stream1(NewTestStream("redirect.html"));
+  scoped_ptr<HttpStream> stream2(NewTestStream("ok.html"));
+
+  HttpRequestHeaders headers1;
+  HttpResponseInfo response1;
+  EXPECT_EQ(OK, stream1->SendRequest(headers1, NULL, &response1, &callback_));
+  HttpRequestHeaders headers2;
+  HttpResponseInfo response2;
+  EXPECT_EQ(OK, stream2->SendRequest(headers2, NULL, &response2, &callback_));
+
+  EXPECT_EQ(OK, stream1->ReadResponseHeaders(&callback_));
+  stream1.release()->Drain(NULL);
+
+  EXPECT_EQ(OK, stream2->ReadResponseHeaders(&callback_));
+  ExpectResponse("ok.html", stream2, false);
+  stream2->Close(false);
+}
+
+TEST_F(HttpPipelinedConnectionImplTest, EvictAfterDrainOfUnknownSize) {
+  MockWrite writes[] = {
+    MockWrite(false, 0, "GET /redirect.html HTTP/1.1\r\n\r\n"),
+    MockWrite(false, 1, "GET /ok.html HTTP/1.1\r\n\r\n"),
+  };
+  MockRead reads[] = {
+    MockRead(false, 2,
+             "HTTP/1.1 302 OK\r\n\r\n"
+             "redirect"),
+  };
+  Initialize(reads, arraysize(reads), writes, arraysize(writes));
+
+  scoped_ptr<HttpStream> stream1(NewTestStream("redirect.html"));
+  scoped_ptr<HttpStream> stream2(NewTestStream("ok.html"));
+
+  HttpRequestHeaders headers1;
+  HttpResponseInfo response1;
+  EXPECT_EQ(OK, stream1->SendRequest(headers1, NULL, &response1, &callback_));
+  HttpRequestHeaders headers2;
+  HttpResponseInfo response2;
+  EXPECT_EQ(OK, stream2->SendRequest(headers2, NULL, &response2, &callback_));
+
+  EXPECT_EQ(OK, stream1->ReadResponseHeaders(&callback_));
+  stream1.release()->Drain(NULL);
+
+  EXPECT_EQ(ERR_PIPELINE_EVICTION, stream2->ReadResponseHeaders(&callback_));
+  stream2->Close(false);
+}
+
+TEST_F(HttpPipelinedConnectionImplTest, EvictAfterFailedDrain) {
+  MockWrite writes[] = {
+    MockWrite(false, 0, "GET /redirect.html HTTP/1.1\r\n\r\n"),
+    MockWrite(false, 1, "GET /ok.html HTTP/1.1\r\n\r\n"),
+  };
+  MockRead reads[] = {
+    MockRead(false, 2,
+             "HTTP/1.1 302 OK\r\n"
+             "Content-Length: 8\r\n\r\n"),
+    MockRead(false, ERR_SOCKET_NOT_CONNECTED, 3),
+  };
+  Initialize(reads, arraysize(reads), writes, arraysize(writes));
+
+  scoped_ptr<HttpStream> stream1(NewTestStream("redirect.html"));
+  scoped_ptr<HttpStream> stream2(NewTestStream("ok.html"));
+
+  HttpRequestHeaders headers1;
+  HttpResponseInfo response1;
+  EXPECT_EQ(OK, stream1->SendRequest(headers1, NULL, &response1, &callback_));
+  HttpRequestHeaders headers2;
+  HttpResponseInfo response2;
+  EXPECT_EQ(OK, stream2->SendRequest(headers2, NULL, &response2, &callback_));
+
+  EXPECT_EQ(OK, stream1->ReadResponseHeaders(&callback_));
+  stream1.release()->Drain(NULL);
+
+  EXPECT_EQ(ERR_PIPELINE_EVICTION, stream2->ReadResponseHeaders(&callback_));
+  stream2->Close(false);
+}
+
+TEST_F(HttpPipelinedConnectionImplTest, EvictIfDrainingChunkedEncoding) {
+  MockWrite writes[] = {
+    MockWrite(false, 0, "GET /redirect.html HTTP/1.1\r\n\r\n"),
+    MockWrite(false, 1, "GET /ok.html HTTP/1.1\r\n\r\n"),
+  };
+  MockRead reads[] = {
+    MockRead(false, 2,
+             "HTTP/1.1 302 OK\r\n"
+             "Transfer-Encoding: chunked\r\n\r\n"),
+    MockRead(false, 3,
+             "jibberish"),
+  };
+  Initialize(reads, arraysize(reads), writes, arraysize(writes));
+
+  scoped_ptr<HttpStream> stream1(NewTestStream("redirect.html"));
+  scoped_ptr<HttpStream> stream2(NewTestStream("ok.html"));
+
+  HttpRequestHeaders headers1;
+  HttpResponseInfo response1;
+  EXPECT_EQ(OK, stream1->SendRequest(headers1, NULL, &response1, &callback_));
+  HttpRequestHeaders headers2;
+  HttpResponseInfo response2;
+  EXPECT_EQ(OK, stream2->SendRequest(headers2, NULL, &response2, &callback_));
+
+  EXPECT_EQ(OK, stream1->ReadResponseHeaders(&callback_));
+  stream1.release()->Drain(NULL);
+
+  EXPECT_EQ(ERR_PIPELINE_EVICTION, stream2->ReadResponseHeaders(&callback_));
+  stream2->Close(false);
+}
+
 TEST_F(HttpPipelinedConnectionImplTest, OnPipelineHasCapacity) {
   MockWrite writes[] = {
     MockWrite(false, 0, "GET /ok.html HTTP/1.1\r\n\r\n"),

net/http/http_pipelined_network_transaction_unittest.cc

@@ -419,6 +419,49 @@ TEST_F(HttpPipelinedNetworkTransactionTest, SendErrorEvictsToNewPipeline) {
   ExpectResponse("two.html", two_transaction);
 }
 
+TEST_F(HttpPipelinedNetworkTransactionTest, RedirectDrained) {
+  Initialize();
+
+  MockWrite writes[] = {
+    MockWrite(false, 0, "GET /redirect.html HTTP/1.1\r\n"
+              "Host: localhost\r\n"
+              "Connection: keep-alive\r\n\r\n"),
+    MockWrite(false, 3, "GET /two.html HTTP/1.1\r\n"
+              "Host: localhost\r\n"
+              "Connection: keep-alive\r\n\r\n"),
+  };
+  MockRead reads[] = {
+    MockRead(false, 1, "HTTP/1.1 302 OK\r\n"),
+    MockRead(false, 2, "Content-Length: 8\r\n\r\n"),
+    MockRead(true, 4, "redirect"),
+    MockRead(false, 5, "HTTP/1.1 200 OK\r\n"),
+    MockRead(false, 6, "Content-Length: 8\r\n\r\n"),
+    MockRead(false, 7, "two.html"),
+  };
+  AddExpectedConnection(reads, arraysize(reads), writes, arraysize(writes));
+
+  scoped_ptr<HttpNetworkTransaction> one_transaction(
+      new HttpNetworkTransaction(session_.get()));
+  TestOldCompletionCallback one_callback;
+  EXPECT_EQ(ERR_IO_PENDING,
+            one_transaction->Start(GetRequestInfo("redirect.html"),
+                                   &one_callback, BoundNetLog()));
+  EXPECT_EQ(OK, one_callback.WaitForResult());
+
+  HttpNetworkTransaction two_transaction(session_.get());
+  TestOldCompletionCallback two_callback;
+  EXPECT_EQ(ERR_IO_PENDING,
+            two_transaction.Start(GetRequestInfo("two.html"), &two_callback,
+                                  BoundNetLog()));
+
+  one_transaction.reset();
+  data_vector_[0]->RunFor(2);
+  data_vector_[0]->SetStop(10);
+
+  EXPECT_EQ(OK, two_callback.WaitForResult());
+  ExpectResponse("two.html", two_transaction);
+}
+
 TEST_F(HttpPipelinedNetworkTransactionTest, BasicHttpAuthentication) {
   Initialize();
 

net/http/http_pipelined_stream.cc

@@ -118,11 +118,8 @@ void HttpPipelinedStream::LogNumRttVsBytesMetrics() const {
   // TODO(simonjam): I don't want to copy & paste this from http_basic_stream.
 }
 
-void HttpPipelinedStream::Drain(HttpNetworkSession*) {
-  // On errors, we already evict everything from the pipeline and close it.
-  // TODO(simonjam): Consider trying to drain the pipeline in the same way that
-  // HttpBasicStream does.
-  delete this;
+void HttpPipelinedStream::Drain(HttpNetworkSession* session) {
+  pipeline_->Drain(this, session);
 }
 
 const SSLConfig& HttpPipelinedStream::used_ssl_config() const {

net/http/http_response_body_drainer.cc

-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Copyright (c) 2011 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.
 
@@ -25,7 +25,25 @@ HttpResponseBodyDrainer::HttpResponseBodyDrainer(HttpStream* stream)
 HttpResponseBodyDrainer::~HttpResponseBodyDrainer() {}
 
 void HttpResponseBodyDrainer::Start(HttpNetworkSession* session) {
-  read_buf_ = new IOBuffer(kDrainBodyBufferSize);
+  StartWithSize(session, kDrainBodyBufferSize);
+}
+
+void HttpResponseBodyDrainer::StartWithSize(HttpNetworkSession* session,
+                                            int num_bytes_to_drain) {
+  DCHECK_LE(0, num_bytes_to_drain);
+  // TODO(simonjam): Consider raising this limit if we're pipelining. If we have
+  // a bunch of responses in the pipeline, we should be less willing to give up
+  // while draining.
+  if (num_bytes_to_drain > kDrainBodyBufferSize) {
+    Finish(ERR_RESPONSE_BODY_TOO_BIG_TO_DRAIN);
+    return;
+  } else if (num_bytes_to_drain == 0) {
+    Finish(OK);
+    return;
+  }
+
+  read_size_ = num_bytes_to_drain;
+  read_buf_ = new IOBuffer(read_size_);
   next_state_ = STATE_DRAIN_RESPONSE_BODY;
   int rv = DoLoop(OK);
 
@@ -71,7 +89,7 @@ int HttpResponseBodyDrainer::DoDrainResponseBody() {
   next_state_ = STATE_DRAIN_RESPONSE_BODY_COMPLETE;
 
   return stream_->ReadResponseBody(
-      read_buf_, kDrainBodyBufferSize - total_read_,
+      read_buf_, read_size_ - total_read_,
       &io_callback_);
 }
 

net/http/http_response_body_drainer.h

@@ -36,6 +36,9 @@ class NET_EXPORT_PRIVATE HttpResponseBodyDrainer {
   // doesn't complete immediately, it will add itself to |session|.
   void Start(HttpNetworkSession* session);
 
+  // As above, but stop reading once |num_bytes_to_drain| has been reached.
+  void StartWithSize(HttpNetworkSession* session, int num_bytes_to_drain);
+
  private:
   enum State {
     STATE_DRAIN_RESPONSE_BODY,
@@ -52,6 +55,7 @@ class NET_EXPORT_PRIVATE HttpResponseBodyDrainer {
   void OnTimerFired();
   void Finish(int result);
 
+  int read_size_;
   scoped_refptr<IOBuffer> read_buf_;
   const scoped_ptr<HttpStream> stream_;
   State next_state_;

net/http/http_response_body_drainer_unittest.cc

@@ -255,6 +255,24 @@ TEST_F(HttpResponseBodyDrainerTest, DrainBodyTooLarge) {
   EXPECT_TRUE(result_waiter_.WaitForResult());
 }
 
+TEST_F(HttpResponseBodyDrainerTest, StartBodyTooLarge) {
+  TestOldCompletionCallback callback;
+  int too_many_chunks =
+      HttpResponseBodyDrainer::kDrainBodyBufferSize / kMagicChunkSize;
+  too_many_chunks += 1;  // Now it's too large.
+
+  mock_stream_->set_num_chunks(0);
+  drainer_->StartWithSize(session_, too_many_chunks * kMagicChunkSize);
+  EXPECT_TRUE(result_waiter_.WaitForResult());
+}
+
+TEST_F(HttpResponseBodyDrainerTest, StartWithNothingToDo) {
+  TestOldCompletionCallback callback;
+  mock_stream_->set_num_chunks(0);
+  drainer_->StartWithSize(session_, 0);
+  EXPECT_FALSE(result_waiter_.WaitForResult());
+}
+
 }  // namespace
 
 }  // namespace net

net/http/http_response_headers.cc

@@ -1291,4 +1291,10 @@ bool HttpResponseHeaders::GetContentRange(int64* first_byte_position,
   return true;
 }
 
+bool HttpResponseHeaders::IsChunkEncoded() const {
+  // Ignore spurious chunked responses from HTTP/1.0 servers and proxies.
+  return GetHttpVersion() >= HttpVersion(1, 1) &&
+      HasHeaderValue("Transfer-Encoding", "chunked");
+}
+
 }  // namespace net

net/http/http_response_headers.h

@@ -239,6 +239,9 @@ class NET_EXPORT HttpResponseHeaders
                        int64* last_byte_position,
                        int64* instance_length) const;
 
+  // Returns true if the response is chunk-encoded.
+  bool IsChunkEncoded() const;
+
   // Returns the HTTP response code.  This is 0 if the response code text seems
   // to exist but could not be parsed.  Otherwise, it defaults to 200 if the
   // response code is not found in the raw headers.

net/http/http_stream_parser.cc

@@ -721,11 +721,8 @@ void HttpStreamParser::CalculateResponseBodySize() {
     response_body_length_ = 0;
 
   if (response_body_length_ == -1) {
-    // Ignore spurious chunked responses from HTTP/1.0 servers and
-    // proxies. Otherwise "Transfer-Encoding: chunked" trumps
-    // "Content-Length: N"
-    if (response_->headers->GetHttpVersion() >= HttpVersion(1, 1) &&
-        response_->headers->HasHeaderValue("Transfer-Encoding", "chunked")) {
+    // "Transfer-Encoding: chunked" trumps "Content-Length: N"
+    if (response_->headers->IsChunkEncoded()) {
       chunked_decoder_.reset(new HttpChunkedDecoder());
     } else {
       response_body_length_ = response_->headers->GetContentLength();