net/http/http_pipelined_connection_impl.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "net/http/http_pipelined_connection_impl.h"
   6
   7 #include "base/bind.h"
   8 #include "base/bind_helpers.h"
   9 #include "base/message_loop/message_loop.h"
  10 #include "base/stl_util.h"
  11 #include "base/values.h"
  12 #include "net/base/io_buffer.h"
  13 #include "net/http/http_pipelined_stream.h"
  14 #include "net/http/http_request_info.h"
  15 #include "net/http/http_response_body_drainer.h"
  16 #include "net/http/http_response_headers.h"
  17 #include "net/http/http_stream_parser.h"
  18 #include "net/http/http_version.h"
  19 #include "net/socket/client_socket_handle.h"
  20
  21 namespace net {
  22
  23 namespace {
  24
  25 base::Value* NetLogReceivedHeadersCallback(const NetLog::Source& source,
  26                                            const std::string* feedback,
  27                                            NetLog::LogLevel /* log_level */) {
  28   base::DictionaryValue* dict = new base::DictionaryValue;
  29   source.AddToEventParameters(dict);
  30   dict->SetString("feedback", *feedback);
  31   return dict;
  32 }
  33
  34 base::Value* NetLogStreamClosedCallback(const NetLog::Source& source,
  35                                         bool not_reusable,
  36                                         NetLog::LogLevel /* log_level */) {
  37   base::DictionaryValue* dict = new base::DictionaryValue;
  38   source.AddToEventParameters(dict);
  39   dict->SetBoolean("not_reusable", not_reusable);
  40   return dict;
  41 }
  42
  43 base::Value* NetLogHostPortPairCallback(const HostPortPair* host_port_pair,
  44                                         NetLog::LogLevel /* log_level */) {
  45   base::DictionaryValue* dict = new base::DictionaryValue;
  46   dict->SetString("host_and_port", host_port_pair->ToString());
  47   return dict;
  48 }
  49
  50 }  // anonymous namespace
  51
  52 HttpPipelinedConnection*
  53 HttpPipelinedConnectionImpl::Factory::CreateNewPipeline(
  54     ClientSocketHandle* connection,
  55     HttpPipelinedConnection::Delegate* delegate,
  56     const HostPortPair& origin,
  57     const SSLConfig& used_ssl_config,
  58     const ProxyInfo& used_proxy_info,
  59     const BoundNetLog& net_log,
  60     bool was_npn_negotiated,
  61     NextProto protocol_negotiated) {
  62   return new HttpPipelinedConnectionImpl(connection, delegate, origin,
  63                                          used_ssl_config, used_proxy_info,
  64                                          net_log, was_npn_negotiated,
  65                                          protocol_negotiated);
  66 }
  67
  68 HttpPipelinedConnectionImpl::HttpPipelinedConnectionImpl(
  69     ClientSocketHandle* connection,
  70     HttpPipelinedConnection::Delegate* delegate,
  71     const HostPortPair& origin,
  72     const SSLConfig& used_ssl_config,
  73     const ProxyInfo& used_proxy_info,
  74     const BoundNetLog& net_log,
  75     bool was_npn_negotiated,
  76     NextProto protocol_negotiated)
  77     : delegate_(delegate),
  78       connection_(connection),
  79       used_ssl_config_(used_ssl_config),
  80       used_proxy_info_(used_proxy_info),
  81       net_log_(BoundNetLog::Make(net_log.net_log(),
  82                                  NetLog::SOURCE_HTTP_PIPELINED_CONNECTION)),
  83       was_npn_negotiated_(was_npn_negotiated),
  84       protocol_negotiated_(protocol_negotiated),
  85       read_buf_(new GrowableIOBuffer()),
  86       next_pipeline_id_(1),
  87       active_(false),
  88       usable_(true),
  89       completed_one_request_(false),
  90       weak_factory_(this),
  91       send_next_state_(SEND_STATE_NONE),
  92       send_still_on_call_stack_(false),
  93       read_next_state_(READ_STATE_NONE),
  94       active_read_id_(0),
  95       read_still_on_call_stack_(false) {
  96   CHECK(connection_.get());
  97   net_log_.BeginEvent(
  98       NetLog::TYPE_HTTP_PIPELINED_CONNECTION,
  99       base::Bind(&NetLogHostPortPairCallback, &origin));
 100 }
 101
 102 HttpPipelinedConnectionImpl::~HttpPipelinedConnectionImpl() {
 103   CHECK_EQ(depth(), 0);
 104   CHECK(stream_info_map_.empty());
 105   CHECK(pending_send_request_queue_.empty());
 106   CHECK(request_order_.empty());
 107   CHECK_EQ(send_next_state_, SEND_STATE_NONE);
 108   CHECK_EQ(read_next_state_, READ_STATE_NONE);
 109   CHECK(!active_send_request_.get());
 110   CHECK(!active_read_id_);
 111   if (!usable_) {
 112     connection_->socket()->Disconnect();
 113   }
 114   connection_->Reset();
 115   net_log_.EndEvent(NetLog::TYPE_HTTP_PIPELINED_CONNECTION);
 116 }
 117
 118 HttpPipelinedStream* HttpPipelinedConnectionImpl::CreateNewStream() {
 119   int pipeline_id = next_pipeline_id_++;
 120   CHECK(pipeline_id);
 121   HttpPipelinedStream* stream = new HttpPipelinedStream(this, pipeline_id);
 122   stream_info_map_.insert(std::make_pair(pipeline_id, StreamInfo()));
 123   return stream;
 124 }
 125
 126 void HttpPipelinedConnectionImpl::InitializeParser(
 127     int pipeline_id,
 128     const HttpRequestInfo* request,
 129     const BoundNetLog& net_log) {
 130   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 131   CHECK(!stream_info_map_[pipeline_id].parser.get());
 132   stream_info_map_[pipeline_id].state = STREAM_BOUND;
 133   stream_info_map_[pipeline_id].parser.reset(new HttpStreamParser(
 134       connection_.get(), request, read_buf_.get(), net_log));
 135   stream_info_map_[pipeline_id].source = net_log.source();
 136
 137   // In case our first stream doesn't SendRequest() immediately, we should still
 138   // allow others to use this pipeline.
 139   if (pipeline_id == 1) {
 140     base::MessageLoop::current()->PostTask(
 141         FROM_HERE,
 142         base::Bind(&HttpPipelinedConnectionImpl::ActivatePipeline,
 143                    weak_factory_.GetWeakPtr()));
 144   }
 145 }
 146
 147 void HttpPipelinedConnectionImpl::ActivatePipeline() {
 148   if (!active_) {
 149     active_ = true;
 150     delegate_->OnPipelineHasCapacity(this);
 151   }
 152 }
 153
 154 void HttpPipelinedConnectionImpl::OnStreamDeleted(int pipeline_id) {
 155   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 156   Close(pipeline_id, false);
 157
 158   if (stream_info_map_[pipeline_id].state != STREAM_CREATED &&
 159       stream_info_map_[pipeline_id].state != STREAM_UNUSED) {
 160     CHECK_EQ(stream_info_map_[pipeline_id].state, STREAM_CLOSED);
 161     CHECK(stream_info_map_[pipeline_id].parser.get());
 162     stream_info_map_[pipeline_id].parser.reset();
 163   }
 164   CHECK(!stream_info_map_[pipeline_id].parser.get());
 165   stream_info_map_.erase(pipeline_id);
 166
 167   delegate_->OnPipelineHasCapacity(this);
 168 }
 169
 170 int HttpPipelinedConnectionImpl::SendRequest(
 171     int pipeline_id,
 172     const std::string& request_line,
 173     const HttpRequestHeaders& headers,
 174     HttpResponseInfo* response,
 175     const CompletionCallback& callback) {
 176   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 177   CHECK_EQ(stream_info_map_[pipeline_id].state, STREAM_BOUND);
 178   if (!usable_) {
 179     return ERR_PIPELINE_EVICTION;
 180   }
 181
 182   PendingSendRequest* send_request = new PendingSendRequest;
 183   send_request->pipeline_id = pipeline_id;
 184   send_request->request_line = request_line;
 185   send_request->headers = headers;
 186   send_request->response = response;
 187   send_request->callback = callback;
 188   pending_send_request_queue_.push(send_request);
 189
 190   int rv;
 191   if (send_next_state_ == SEND_STATE_NONE) {
 192     send_next_state_ = SEND_STATE_START_IMMEDIATELY;
 193     rv = DoSendRequestLoop(OK);
 194   } else {
 195     rv = ERR_IO_PENDING;
 196   }
 197   ActivatePipeline();
 198   return rv;
 199 }
 200
 201 int HttpPipelinedConnectionImpl::DoSendRequestLoop(int result) {
 202   int rv = result;
 203   do {
 204     SendRequestState state = send_next_state_;
 205     send_next_state_ = SEND_STATE_NONE;
 206     switch (state) {
 207       case SEND_STATE_START_IMMEDIATELY:
 208         rv = DoStartRequestImmediately(rv);
 209         break;
 210       case SEND_STATE_START_NEXT_DEFERRED_REQUEST:
 211         rv = DoStartNextDeferredRequest(rv);
 212         break;
 213       case SEND_STATE_SEND_ACTIVE_REQUEST:
 214         rv = DoSendActiveRequest(rv);
 215         break;
 216       case SEND_STATE_COMPLETE:
 217         rv = DoSendComplete(rv);
 218         break;
 219       case SEND_STATE_EVICT_PENDING_REQUESTS:
 220         rv = DoEvictPendingSendRequests(rv);
 221         break;
 222       default:
 223         CHECK(false) << "bad send state: " << state;
 224         rv = ERR_FAILED;
 225         break;
 226     }
 227   } while (rv != ERR_IO_PENDING && send_next_state_ != SEND_STATE_NONE);
 228   send_still_on_call_stack_ = false;
 229   return rv;
 230 }
 231
 232 void HttpPipelinedConnectionImpl::OnSendIOCallback(int result) {
 233   CHECK(active_send_request_.get());
 234   DoSendRequestLoop(result);
 235 }
 236
 237 int HttpPipelinedConnectionImpl::DoStartRequestImmediately(int result) {
 238   CHECK(!active_send_request_.get());
 239   CHECK_EQ(static_cast<size_t>(1), pending_send_request_queue_.size());
 240   // If SendRequest() completes synchronously, then we need to return the value
 241   // directly to the caller. |send_still_on_call_stack_| will track this.
 242   // Otherwise, asynchronous completions will notify the caller via callback.
 243   send_still_on_call_stack_ = true;
 244   active_send_request_.reset(pending_send_request_queue_.front());
 245   pending_send_request_queue_.pop();
 246   send_next_state_ = SEND_STATE_SEND_ACTIVE_REQUEST;
 247   return OK;
 248 }
 249
 250 int HttpPipelinedConnectionImpl::DoStartNextDeferredRequest(int result) {
 251   CHECK(!send_still_on_call_stack_);
 252   CHECK(!active_send_request_.get());
 253
 254   while (!pending_send_request_queue_.empty()) {
 255     scoped_ptr<PendingSendRequest> next_request(
 256         pending_send_request_queue_.front());
 257     pending_send_request_queue_.pop();
 258     CHECK(ContainsKey(stream_info_map_, next_request->pipeline_id));
 259     if (stream_info_map_[next_request->pipeline_id].state != STREAM_CLOSED) {
 260       active_send_request_.reset(next_request.release());
 261       send_next_state_ = SEND_STATE_SEND_ACTIVE_REQUEST;
 262       return OK;
 263     }
 264   }
 265
 266   send_next_state_ = SEND_STATE_NONE;
 267   return OK;
 268 }
 269
 270 int HttpPipelinedConnectionImpl::DoSendActiveRequest(int result) {
 271   CHECK(stream_info_map_[active_send_request_->pipeline_id].parser.get());
 272   int rv = stream_info_map_[active_send_request_->pipeline_id].parser->
 273       SendRequest(active_send_request_->request_line,
 274                   active_send_request_->headers,
 275                   active_send_request_->response,
 276                   base::Bind(&HttpPipelinedConnectionImpl::OnSendIOCallback,
 277                              base::Unretained(this)));
 278   stream_info_map_[active_send_request_->pipeline_id].state = STREAM_SENDING;
 279   send_next_state_ = SEND_STATE_COMPLETE;
 280   return rv;
 281 }
 282
 283 int HttpPipelinedConnectionImpl::DoSendComplete(int result) {
 284   CHECK(active_send_request_.get());
 285   CHECK_EQ(STREAM_SENDING,
 286            stream_info_map_[active_send_request_->pipeline_id].state);
 287
 288   request_order_.push(active_send_request_->pipeline_id);
 289   stream_info_map_[active_send_request_->pipeline_id].state = STREAM_SENT;
 290   net_log_.AddEvent(
 291       NetLog::TYPE_HTTP_PIPELINED_CONNECTION_SENT_REQUEST,
 292       stream_info_map_[active_send_request_->pipeline_id].source.
 293           ToEventParametersCallback());
 294
 295   if (result == ERR_SOCKET_NOT_CONNECTED && completed_one_request_) {
 296     result = ERR_PIPELINE_EVICTION;
 297   }
 298   if (result < OK) {
 299     usable_ = false;
 300   }
 301
 302   if (!send_still_on_call_stack_) {
 303     QueueUserCallback(active_send_request_->pipeline_id,
 304                       active_send_request_->callback, result, FROM_HERE);
 305   }
 306
 307   active_send_request_.reset();
 308
 309   if (send_still_on_call_stack_) {
 310     // It should be impossible for another request to appear on the queue while
 311     // this send was on the call stack.
 312     CHECK(pending_send_request_queue_.empty());
 313     send_next_state_ = SEND_STATE_NONE;
 314   } else if (!usable_) {
 315     send_next_state_ = SEND_STATE_EVICT_PENDING_REQUESTS;
 316   } else {
 317     send_next_state_ = SEND_STATE_START_NEXT_DEFERRED_REQUEST;
 318   }
 319
 320   return result;
 321 }
 322
 323 int HttpPipelinedConnectionImpl::DoEvictPendingSendRequests(int result) {
 324   while (!pending_send_request_queue_.empty()) {
 325     scoped_ptr<PendingSendRequest> evicted_send(
 326         pending_send_request_queue_.front());
 327     pending_send_request_queue_.pop();
 328     if (ContainsKey(stream_info_map_, evicted_send->pipeline_id) &&
 329         stream_info_map_[evicted_send->pipeline_id].state != STREAM_CLOSED) {
 330       evicted_send->callback.Run(ERR_PIPELINE_EVICTION);
 331     }
 332   }
 333   send_next_state_ = SEND_STATE_NONE;
 334   return result;
 335 }
 336
 337 int HttpPipelinedConnectionImpl::ReadResponseHeaders(
 338     int pipeline_id, const CompletionCallback& callback) {
 339   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 340   CHECK_EQ(STREAM_SENT, stream_info_map_[pipeline_id].state);
 341   CHECK(stream_info_map_[pipeline_id].read_headers_callback.is_null());
 342
 343   if (!usable_)
 344     return ERR_PIPELINE_EVICTION;
 345
 346   stream_info_map_[pipeline_id].state = STREAM_READ_PENDING;
 347   stream_info_map_[pipeline_id].read_headers_callback = callback;
 348   if (read_next_state_ == READ_STATE_NONE &&
 349       pipeline_id == request_order_.front()) {
 350     read_next_state_ = READ_STATE_START_IMMEDIATELY;
 351     return DoReadHeadersLoop(OK);
 352   }
 353   return ERR_IO_PENDING;
 354 }
 355
 356 void HttpPipelinedConnectionImpl::StartNextDeferredRead() {
 357   if (read_next_state_ == READ_STATE_NONE) {
 358     read_next_state_ = READ_STATE_START_NEXT_DEFERRED_READ;
 359     DoReadHeadersLoop(OK);
 360   }
 361 }
 362
 363 int HttpPipelinedConnectionImpl::DoReadHeadersLoop(int result) {
 364   int rv = result;
 365   do {
 366     ReadHeadersState state = read_next_state_;
 367     read_next_state_ = READ_STATE_NONE;
 368     switch (state) {
 369       case READ_STATE_START_IMMEDIATELY:
 370         rv = DoStartReadImmediately(rv);
 371         break;
 372       case READ_STATE_START_NEXT_DEFERRED_READ:
 373         rv = DoStartNextDeferredRead(rv);
 374         break;
 375       case READ_STATE_READ_HEADERS:
 376         rv = DoReadHeaders(rv);
 377         break;
 378       case READ_STATE_READ_HEADERS_COMPLETE:
 379         rv = DoReadHeadersComplete(rv);
 380         break;
 381       case READ_STATE_WAITING_FOR_CLOSE:
 382         // This is a holding state. We return instead of continuing to run hte
 383         // loop. The state will advance when the stream calls Close().
 384         rv = DoReadWaitForClose(rv);
 385         read_still_on_call_stack_ = false;
 386         return rv;
 387       case READ_STATE_STREAM_CLOSED:
 388         rv = DoReadStreamClosed();
 389         break;
 390       case READ_STATE_EVICT_PENDING_READS:
 391         rv = DoEvictPendingReadHeaders(rv);
 392         break;
 393       case READ_STATE_NONE:
 394         break;
 395       default:
 396         CHECK(false) << "bad read state";
 397         rv = ERR_FAILED;
 398         break;
 399     }
 400   } while (rv != ERR_IO_PENDING && read_next_state_ != READ_STATE_NONE);
 401   read_still_on_call_stack_ = false;
 402   return rv;
 403 }
 404
 405 void HttpPipelinedConnectionImpl::OnReadIOCallback(int result) {
 406   DoReadHeadersLoop(result);
 407 }
 408
 409 int HttpPipelinedConnectionImpl::DoStartReadImmediately(int result) {
 410   CHECK(!active_read_id_);
 411   CHECK(!read_still_on_call_stack_);
 412   CHECK(!request_order_.empty());
 413   // If ReadResponseHeaders() completes synchronously, then we need to return
 414   // the value directly to the caller. |read_still_on_call_stack_| will track
 415   // this. Otherwise, asynchronous completions will notify the caller via
 416   // callback.
 417   read_still_on_call_stack_ = true;
 418   read_next_state_ = READ_STATE_READ_HEADERS;
 419   active_read_id_ = request_order_.front();
 420   request_order_.pop();
 421   return OK;
 422 }
 423
 424 int HttpPipelinedConnectionImpl::DoStartNextDeferredRead(int result) {
 425   CHECK(!active_read_id_);
 426   CHECK(!read_still_on_call_stack_);
 427
 428   if (request_order_.empty()) {
 429     read_next_state_ = READ_STATE_NONE;
 430     return OK;
 431   }
 432
 433   int next_id = request_order_.front();
 434   CHECK(ContainsKey(stream_info_map_, next_id));
 435   switch (stream_info_map_[next_id].state) {
 436     case STREAM_READ_PENDING:
 437       read_next_state_ = READ_STATE_READ_HEADERS;
 438       active_read_id_ = next_id;
 439       request_order_.pop();
 440       break;
 441
 442     case STREAM_CLOSED:
 443       // Since nobody will read whatever data is on the pipeline associated with
 444       // this closed request, we must shut down the rest of the pipeline.
 445       read_next_state_ = READ_STATE_EVICT_PENDING_READS;
 446       break;
 447
 448     case STREAM_SENT:
 449       read_next_state_ = READ_STATE_NONE;
 450       break;
 451
 452     default:
 453       CHECK(false) << "Unexpected read state: "
 454                    << stream_info_map_[next_id].state;
 455   }
 456
 457   return OK;
 458 }
 459
 460 int HttpPipelinedConnectionImpl::DoReadHeaders(int result) {
 461   CHECK(active_read_id_);
 462   CHECK(ContainsKey(stream_info_map_, active_read_id_));
 463   CHECK_EQ(STREAM_READ_PENDING, stream_info_map_[active_read_id_].state);
 464   stream_info_map_[active_read_id_].state = STREAM_ACTIVE;
 465   int rv = stream_info_map_[active_read_id_].parser->ReadResponseHeaders(
 466       base::Bind(&HttpPipelinedConnectionImpl::OnReadIOCallback,
 467                  base::Unretained(this)));
 468   read_next_state_ = READ_STATE_READ_HEADERS_COMPLETE;
 469   return rv;
 470 }
 471
 472 int HttpPipelinedConnectionImpl::DoReadHeadersComplete(int result) {
 473   CHECK(active_read_id_);
 474   CHECK(ContainsKey(stream_info_map_, active_read_id_));
 475   CHECK_EQ(STREAM_ACTIVE, stream_info_map_[active_read_id_].state);
 476
 477   read_next_state_ = READ_STATE_WAITING_FOR_CLOSE;
 478   if (result < OK) {
 479     if (completed_one_request_ &&
 480         (result == ERR_CONNECTION_CLOSED ||
 481          result == ERR_EMPTY_RESPONSE ||
 482          result == ERR_SOCKET_NOT_CONNECTED)) {
 483       // These usually indicate that pipelining failed on the server side. In
 484       // that case, we should retry without pipelining.
 485       result = ERR_PIPELINE_EVICTION;
 486     }
 487     usable_ = false;
 488   }
 489
 490   CheckHeadersForPipelineCompatibility(active_read_id_, result);
 491
 492   if (!read_still_on_call_stack_) {
 493     QueueUserCallback(active_read_id_,
 494                       stream_info_map_[active_read_id_].read_headers_callback,
 495                       result, FROM_HERE);
 496   }
 497
 498   return result;
 499 }
 500
 501 int HttpPipelinedConnectionImpl::DoReadWaitForClose(int result) {
 502   read_next_state_ = READ_STATE_WAITING_FOR_CLOSE;
 503   return result;
 504 }
 505
 506 int HttpPipelinedConnectionImpl::DoReadStreamClosed() {
 507   CHECK(active_read_id_);
 508   CHECK(ContainsKey(stream_info_map_, active_read_id_));
 509   CHECK_EQ(stream_info_map_[active_read_id_].state, STREAM_CLOSED);
 510   active_read_id_ = 0;
 511   if (!usable_) {
 512     // TODO(simonjam): Don't wait this long to evict.
 513     read_next_state_ = READ_STATE_EVICT_PENDING_READS;
 514     return OK;
 515   }
 516   completed_one_request_ = true;
 517   base::MessageLoop::current()->PostTask(
 518       FROM_HERE,
 519       base::Bind(&HttpPipelinedConnectionImpl::StartNextDeferredRead,
 520                  weak_factory_.GetWeakPtr()));
 521   read_next_state_ = READ_STATE_NONE;
 522   return OK;
 523 }
 524
 525 int HttpPipelinedConnectionImpl::DoEvictPendingReadHeaders(int result) {
 526   while (!request_order_.empty()) {
 527     int evicted_id = request_order_.front();
 528     request_order_.pop();
 529     if (!ContainsKey(stream_info_map_, evicted_id)) {
 530       continue;
 531     }
 532     if (stream_info_map_[evicted_id].state == STREAM_READ_PENDING) {
 533       stream_info_map_[evicted_id].state = STREAM_READ_EVICTED;
 534       stream_info_map_[evicted_id].read_headers_callback.Run(
 535           ERR_PIPELINE_EVICTION);
 536     }
 537   }
 538   read_next_state_ = READ_STATE_NONE;
 539   return result;
 540 }
 541
 542 void HttpPipelinedConnectionImpl::Close(int pipeline_id,
 543                                         bool not_reusable) {
 544   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 545   net_log_.AddEvent(
 546       NetLog::TYPE_HTTP_PIPELINED_CONNECTION_STREAM_CLOSED,
 547       base::Bind(&NetLogStreamClosedCallback,
 548                  stream_info_map_[pipeline_id].source, not_reusable));
 549   switch (stream_info_map_[pipeline_id].state) {
 550     case STREAM_CREATED:
 551       stream_info_map_[pipeline_id].state = STREAM_UNUSED;
 552       break;
 553
 554     case STREAM_BOUND:
 555       stream_info_map_[pipeline_id].state = STREAM_CLOSED;
 556       break;
 557
 558     case STREAM_SENDING:
 559       usable_ = false;
 560       stream_info_map_[pipeline_id].state = STREAM_CLOSED;
 561       active_send_request_.reset();
 562       send_next_state_ = SEND_STATE_EVICT_PENDING_REQUESTS;
 563       DoSendRequestLoop(OK);
 564       break;
 565
 566     case STREAM_SENT:
 567     case STREAM_READ_PENDING:
 568       usable_ = false;
 569       stream_info_map_[pipeline_id].state = STREAM_CLOSED;
 570       if (!request_order_.empty() &&
 571           pipeline_id == request_order_.front() &&
 572           read_next_state_ == READ_STATE_NONE) {
 573         read_next_state_ = READ_STATE_EVICT_PENDING_READS;
 574         DoReadHeadersLoop(OK);
 575       }
 576       break;
 577
 578     case STREAM_ACTIVE:
 579       stream_info_map_[pipeline_id].state = STREAM_CLOSED;
 580       if (not_reusable) {
 581         usable_ = false;
 582       }
 583       read_next_state_ = READ_STATE_STREAM_CLOSED;
 584       DoReadHeadersLoop(OK);
 585       break;
 586
 587     case STREAM_READ_EVICTED:
 588       stream_info_map_[pipeline_id].state = STREAM_CLOSED;
 589       break;
 590
 591     case STREAM_CLOSED:
 592     case STREAM_UNUSED:
 593       // TODO(simonjam): Why is Close() sometimes called twice?
 594       break;
 595
 596     default:
 597       CHECK(false);
 598       break;
 599   }
 600 }
 601
 602 int HttpPipelinedConnectionImpl::ReadResponseBody(
 603     int pipeline_id, IOBuffer* buf, int buf_len,
 604     const CompletionCallback& callback) {
 605   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 606   CHECK_EQ(active_read_id_, pipeline_id);
 607   CHECK(stream_info_map_[pipeline_id].parser.get());
 608   return stream_info_map_[pipeline_id].parser->ReadResponseBody(
 609       buf, buf_len, callback);
 610 }
 611
 612 UploadProgress HttpPipelinedConnectionImpl::GetUploadProgress(
 613     int pipeline_id) const {
 614   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 615   CHECK(stream_info_map_.find(pipeline_id)->second.parser.get());
 616   return stream_info_map_.find(pipeline_id)->second.parser->GetUploadProgress();
 617 }
 618
 619 HttpResponseInfo* HttpPipelinedConnectionImpl::GetResponseInfo(
 620     int pipeline_id) {
 621   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 622   CHECK(stream_info_map_.find(pipeline_id)->second.parser.get());
 623   return stream_info_map_.find(pipeline_id)->second.parser->GetResponseInfo();
 624 }
 625
 626 bool HttpPipelinedConnectionImpl::IsResponseBodyComplete(
 627     int pipeline_id) const {
 628   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 629   CHECK(stream_info_map_.find(pipeline_id)->second.parser.get());
 630   return stream_info_map_.find(pipeline_id)->second.parser->
 631       IsResponseBodyComplete();
 632 }
 633
 634 bool HttpPipelinedConnectionImpl::CanFindEndOfResponse(int pipeline_id) const {
 635   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 636   CHECK(stream_info_map_.find(pipeline_id)->second.parser.get());
 637   return stream_info_map_.find(pipeline_id)->second.parser->
 638       CanFindEndOfResponse();
 639 }
 640
 641 bool HttpPipelinedConnectionImpl::IsConnectionReused(int pipeline_id) const {
 642   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 643   if (pipeline_id > 1) {
 644     return true;
 645   }
 646   ClientSocketHandle::SocketReuseType reuse_type = connection_->reuse_type();
 647   return connection_->is_reused() ||
 648          reuse_type == ClientSocketHandle::UNUSED_IDLE;
 649 }
 650
 651 void HttpPipelinedConnectionImpl::SetConnectionReused(int pipeline_id) {
 652   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 653   connection_->set_is_reused(true);
 654 }
 655
 656 bool HttpPipelinedConnectionImpl::GetLoadTimingInfo(
 657     int pipeline_id, LoadTimingInfo* load_timing_info) const {
 658   return connection_->GetLoadTimingInfo(IsConnectionReused(pipeline_id),
 659                                         load_timing_info);
 660 }
 661
 662 void HttpPipelinedConnectionImpl::GetSSLInfo(int pipeline_id,
 663                                              SSLInfo* ssl_info) {
 664   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 665   CHECK(stream_info_map_[pipeline_id].parser.get());
 666   stream_info_map_[pipeline_id].parser->GetSSLInfo(ssl_info);
 667 }
 668
 669 void HttpPipelinedConnectionImpl::GetSSLCertRequestInfo(
 670     int pipeline_id,
 671     SSLCertRequestInfo* cert_request_info) {
 672   CHECK(ContainsKey(stream_info_map_, pipeline_id));
 673   CHECK(stream_info_map_[pipeline_id].parser.get());
 674   stream_info_map_[pipeline_id].parser->GetSSLCertRequestInfo(
 675       cert_request_info);
 676 }
 677
 678 void HttpPipelinedConnectionImpl::Drain(HttpPipelinedStream* stream,
 679                                         HttpNetworkSession* session) {
 680   HttpResponseHeaders* headers = stream->GetResponseInfo()->headers.get();
 681   if (!stream->CanFindEndOfResponse() || headers->IsChunkEncoded() ||
 682       !usable_) {
 683     // TODO(simonjam): Drain chunk-encoded responses if they're relatively
 684     // common.
 685     stream->Close(true);
 686     delete stream;
 687     return;
 688   }
 689   HttpResponseBodyDrainer* drainer = new HttpResponseBodyDrainer(stream);
 690   drainer->StartWithSize(session, headers->GetContentLength());
 691   // |drainer| will delete itself when done.
 692 }
 693
 694 void HttpPipelinedConnectionImpl::CheckHeadersForPipelineCompatibility(
 695     int pipeline_id,
 696     int result) {
 697   if (result < OK) {
 698     switch (result) {
 699       // TODO(simonjam): Ignoring specific errors like this may not work.
 700       // Collect metrics to see if this code is useful.
 701       case ERR_ABORTED:
 702       case ERR_INTERNET_DISCONNECTED:
 703       case ERR_NETWORK_CHANGED:
 704         // These errors are no fault of the server.
 705         break;
 706
 707       default:
 708         ReportPipelineFeedback(pipeline_id, PIPELINE_SOCKET_ERROR);
 709         break;
 710     }
 711     return;
 712   }
 713   HttpResponseInfo* info = GetResponseInfo(pipeline_id);
 714   const HttpVersion required_version(1, 1);
 715   if (info->headers->GetParsedHttpVersion() < required_version) {
 716     ReportPipelineFeedback(pipeline_id, OLD_HTTP_VERSION);
 717     return;
 718   }
 719   if (!info->headers->IsKeepAlive() || !CanFindEndOfResponse(pipeline_id)) {
 720     usable_ = false;
 721     ReportPipelineFeedback(pipeline_id, MUST_CLOSE_CONNECTION);
 722     return;
 723   }
 724   if (info->headers->HasHeader(
 725       HttpAuth::GetChallengeHeaderName(HttpAuth::AUTH_SERVER))) {
 726     ReportPipelineFeedback(pipeline_id, AUTHENTICATION_REQUIRED);
 727     return;
 728   }
 729   ReportPipelineFeedback(pipeline_id, OK);
 730 }
 731
 732 void HttpPipelinedConnectionImpl::ReportPipelineFeedback(int pipeline_id,
 733                                                          Feedback feedback) {
 734   std::string feedback_str;
 735   switch (feedback) {
 736     case OK:
 737       feedback_str = "OK";
 738       break;
 739
 740     case PIPELINE_SOCKET_ERROR:
 741       feedback_str = "PIPELINE_SOCKET_ERROR";
 742       break;
 743
 744     case OLD_HTTP_VERSION:
 745       feedback_str = "OLD_HTTP_VERSION";
 746       break;
 747
 748     case MUST_CLOSE_CONNECTION:
 749       feedback_str = "MUST_CLOSE_CONNECTION";
 750       break;
 751
 752     case AUTHENTICATION_REQUIRED:
 753       feedback_str = "AUTHENTICATION_REQUIRED";
 754       break;
 755
 756     default:
 757       NOTREACHED();
 758       feedback_str = "UNKNOWN";
 759       break;
 760   }
 761   net_log_.AddEvent(
 762       NetLog::TYPE_HTTP_PIPELINED_CONNECTION_RECEIVED_HEADERS,
 763       base::Bind(&NetLogReceivedHeadersCallback,
 764                  stream_info_map_[pipeline_id].source, &feedback_str));
 765   delegate_->OnPipelineFeedback(this, feedback);
 766 }
 767
 768 void HttpPipelinedConnectionImpl::QueueUserCallback(
 769     int pipeline_id, const CompletionCallback& callback, int rv,
 770     const tracked_objects::Location& from_here) {
 771   CHECK(stream_info_map_[pipeline_id].pending_user_callback.is_null());
 772   stream_info_map_[pipeline_id].pending_user_callback = callback;
 773   base::MessageLoop::current()->PostTask(
 774       from_here,
 775       base::Bind(&HttpPipelinedConnectionImpl::FireUserCallback,
 776                  weak_factory_.GetWeakPtr(), pipeline_id, rv));
 777 }
 778
 779 void HttpPipelinedConnectionImpl::FireUserCallback(int pipeline_id,
 780                                                    int result) {
 781   if (ContainsKey(stream_info_map_, pipeline_id)) {
 782     CHECK(!stream_info_map_[pipeline_id].pending_user_callback.is_null());
 783     CompletionCallback callback =
 784         stream_info_map_[pipeline_id].pending_user_callback;
 785     stream_info_map_[pipeline_id].pending_user_callback.Reset();
 786     callback.Run(result);
 787   }
 788 }
 789
 790 int HttpPipelinedConnectionImpl::depth() const {
 791   return stream_info_map_.size();
 792 }
 793
 794 bool HttpPipelinedConnectionImpl::usable() const {
 795   return usable_;
 796 }
 797
 798 bool HttpPipelinedConnectionImpl::active() const {
 799   return active_;
 800 }
 801
 802 const SSLConfig& HttpPipelinedConnectionImpl::used_ssl_config() const {
 803   return used_ssl_config_;
 804 }
 805
 806 const ProxyInfo& HttpPipelinedConnectionImpl::used_proxy_info() const {
 807   return used_proxy_info_;
 808 }
 809
 810 const BoundNetLog& HttpPipelinedConnectionImpl::net_log() const {
 811   return net_log_;
 812 }
 813
 814 bool HttpPipelinedConnectionImpl::was_npn_negotiated() const {
 815   return was_npn_negotiated_;
 816 }
 817
 818 NextProto HttpPipelinedConnectionImpl::protocol_negotiated()
 819     const {
 820   return protocol_negotiated_;
 821 }
 822
 823 HttpPipelinedConnectionImpl::PendingSendRequest::PendingSendRequest()
 824     : pipeline_id(0),
 825       response(NULL) {
 826 }
 827
 828 HttpPipelinedConnectionImpl::PendingSendRequest::~PendingSendRequest() {
 829 }
 830
 831 HttpPipelinedConnectionImpl::StreamInfo::StreamInfo()
 832     : state(STREAM_CREATED) {
 833 }
 834
 835 HttpPipelinedConnectionImpl::StreamInfo::~StreamInfo() {
 836 }
 837
 838 }  // namespace net