Basic HTTP pipelining support

net/http/http_pipelined_connection.cc

+// 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.
+
+#include "net/http/http_pipelined_connection.h"
+
+#include "base/message_loop.h"
+#include "base/stl_util-inl.h"
+#include "net/base/io_buffer.h"
+#include "net/http/http_pipelined_host.h"
+#include "net/http/http_pipelined_stream.h"
+#include "net/http/http_request_info.h"
+#include "net/http/http_stream_parser.h"
+#include "net/socket/client_socket_handle.h"
+
+namespace net {
+
+HttpPipelinedConnection::HttpPipelinedConnection(
+    ClientSocketHandle* connection,
+    HttpPipelinedHost* host,
+    const SSLConfig& used_ssl_config,
+    const ProxyInfo& used_proxy_info,
+    const BoundNetLog& net_log,
+    bool was_npn_negotiated)
+    : host_(host),
+      connection_(connection),
+      used_ssl_config_(used_ssl_config),
+      used_proxy_info_(used_proxy_info),
+      net_log_(net_log),
+      was_npn_negotiated_(was_npn_negotiated),
+      read_buf_(new GrowableIOBuffer()),
+      next_pipeline_id_(1),
+      active_(false),
+      usable_(true),
+      ALLOW_THIS_IN_INITIALIZER_LIST(method_factory_(this)),
+      send_next_state_(SEND_STATE_NONE),
+      ALLOW_THIS_IN_INITIALIZER_LIST(send_io_callback_(
+          this, &HttpPipelinedConnection::OnSendIOCallback)),
+      send_user_callback_(NULL),
+      read_next_state_(READ_STATE_NONE),
+      ALLOW_THIS_IN_INITIALIZER_LIST(read_io_callback_(
+          this, &HttpPipelinedConnection::OnReadIOCallback)),
+      read_user_callback_(NULL) {
+  DCHECK(connection_.get());
+}
+
+HttpPipelinedConnection::~HttpPipelinedConnection() {
+  DCHECK(depth() == 0);
+  DCHECK(parser_map_.empty());
+  DCHECK(callback_map_.empty());
+  DCHECK(deferred_request_queue_.empty());
+  DCHECK(request_order_.empty());
+  DCHECK_EQ(send_next_state_, SEND_STATE_NONE);
+  DCHECK_EQ(read_next_state_, READ_STATE_NONE);
+  DCHECK(!send_user_callback_);
+  DCHECK(!read_user_callback_);
+  if (!usable_) {
+    connection_->socket()->Disconnect();
+  }
+  connection_->Reset();
+}
+
+int HttpPipelinedConnection::AddStream() {
+  int pipeline_id = next_pipeline_id_++;
+  DCHECK(pipeline_id);
+  stream_state_map_.insert(std::make_pair(pipeline_id, STREAM_CREATED));
+  return pipeline_id;
+}
+
+void HttpPipelinedConnection::InitializeParser(int pipeline_id,
+                                               const HttpRequestInfo* request,
+                                               const BoundNetLog& net_log) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK(!ContainsKey(parser_map_, pipeline_id));
+  stream_state_map_[pipeline_id] = STREAM_BOUND;
+  HttpStreamParser* parser = new HttpStreamParser(connection_.get(),
+                                                  request,
+                                                  read_buf_.get(),
+                                                  net_log);
+  parser_map_.insert(std::make_pair(pipeline_id, parser));
+  if (!active_) {
+    active_ = true;
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &HttpPipelinedConnection::FillPipeline));
+  }
+}
+
+void HttpPipelinedConnection::FillPipeline() {
+  host_->OnPipelineHasCapacity(this);
+}
+
+void HttpPipelinedConnection::RemoveStream(int pipeline_id) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+
+  if (stream_state_map_[pipeline_id] != STREAM_CREATED) {
+    DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_CLOSED);
+    DCHECK(ContainsKey(parser_map_, pipeline_id));
+    DCHECK(!ContainsKey(callback_map_, pipeline_id));
+
+    stream_state_map_.erase(pipeline_id);
+    ParserMap::iterator it = parser_map_.find(pipeline_id);
+    delete it->second;
+    parser_map_.erase(it);
+  }
+
+  MessageLoop::current()->PostTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(
+          &HttpPipelinedConnection::FillPipeline));
+}
+
+int HttpPipelinedConnection::SendRequest(int pipeline_id,
+                                         const std::string& request_line,
+                                         const HttpRequestHeaders& headers,
+                                         UploadDataStream* request_body,
+                                         HttpResponseInfo* response,
+                                         CompletionCallback* callback) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_BOUND);
+  if (!usable_) {
+    return ERR_PIPELINE_EVICTION;
+  }
+
+  DeferredSendRequest deferred_request;
+  deferred_request.pipeline_id = pipeline_id;
+  deferred_request.request_line = request_line;
+  deferred_request.headers = headers;
+  deferred_request.request_body = request_body;
+  deferred_request.response = response;
+  deferred_request.callback = callback;
+  deferred_request_queue_.push(deferred_request);
+
+  if (send_next_state_ == SEND_STATE_NONE) {
+    send_next_state_ = SEND_STATE_NEXT_REQUEST;
+    return DoSendRequestLoop(OK);
+  } else {
+    return ERR_IO_PENDING;
+  }
+}
+
+int HttpPipelinedConnection::DoSendRequestLoop(int result) {
+  int rv = result;
+  do {
+    SendRequestState state = send_next_state_;
+    send_next_state_ = SEND_STATE_NONE;
+    switch (state) {
+      case SEND_STATE_NEXT_REQUEST:
+        rv = DoSendNextRequest(rv);
+        break;
+      case SEND_STATE_COMPLETE:
+        rv = DoSendComplete(rv);
+        break;
+      case SEND_STATE_UNUSABLE:
+        rv = DoEvictPendingSendRequests(rv);
+        break;
+      default:
+        NOTREACHED() << "bad send state";
+        rv = ERR_FAILED;
+        break;
+    }
+  } while (rv != ERR_IO_PENDING && send_next_state_ != SEND_STATE_NONE);
+  return rv;
+}
+
+void HttpPipelinedConnection::OnSendIOCallback(int result) {
+  DCHECK(send_user_callback_);
+  DoSendRequestLoop(result);
+}
+
+int HttpPipelinedConnection::DoSendNextRequest(int result) {
+  DCHECK(!deferred_request_queue_.empty());
+  const DeferredSendRequest& deferred_request = deferred_request_queue_.front();
+  DCHECK(ContainsKey(stream_state_map_, deferred_request.pipeline_id));
+  if (stream_state_map_[deferred_request.pipeline_id] == STREAM_CLOSED) {
+    deferred_request_queue_.pop();
+    if (deferred_request_queue_.empty()) {
+      send_next_state_ = SEND_STATE_NONE;
+    } else {
+      send_next_state_ = SEND_STATE_NEXT_REQUEST;
+    }
+    return OK;
+  }
+  DCHECK(ContainsKey(parser_map_, deferred_request.pipeline_id));
+  int rv = parser_map_[deferred_request.pipeline_id]->SendRequest(
+      deferred_request.request_line,
+      deferred_request.headers,
+      deferred_request.request_body,
+      deferred_request.response,
+      &send_io_callback_);
+  if (result == ERR_IO_PENDING || rv == ERR_IO_PENDING) {
+    send_user_callback_ = deferred_request.callback;
+  }
+  request_order_.push(deferred_request.pipeline_id);
+  stream_state_map_[deferred_request.pipeline_id] = STREAM_SENT;
+  deferred_request_queue_.pop();
+  send_next_state_ = SEND_STATE_COMPLETE;
+  return rv;
+}
+
+int HttpPipelinedConnection::DoSendComplete(int result) {
+  if (send_user_callback_) {
+    CompletionCallback* callback = send_user_callback_;
+    send_user_callback_ = NULL;
+    callback->Run(result);
+  }
+  if (result != OK) {
+    send_next_state_ = SEND_STATE_UNUSABLE;
+    usable_ = false;
+    return result;
+  }
+  if (deferred_request_queue_.empty()) {
+    send_next_state_ = SEND_STATE_NONE;
+  } else {
+    send_next_state_ = SEND_STATE_NEXT_REQUEST;
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &HttpPipelinedConnection::DoSendRequestLoop,
+            ERR_IO_PENDING));
+  }
+  return OK;
+}
+
+int HttpPipelinedConnection::DoEvictPendingSendRequests(int result) {
+  send_next_state_ = SEND_STATE_NONE;
+  while (!deferred_request_queue_.empty()) {
+    const DeferredSendRequest& evicted_send = deferred_request_queue_.front();
+    if (stream_state_map_[evicted_send.pipeline_id] == STREAM_CLOSED) {

[mmenke, 2011/08/03 21:09:39]

This case will result in an infinite loop without a 
deferred_request_queue_.pop().

[James Simonsen, 2011/08/05 01:39:00]

Done. Added a unit test too.

+      continue;
+    }
+    evicted_send.callback->Run(ERR_PIPELINE_EVICTION);
+    deferred_request_queue_.pop();
+  }
+  return result;
+}
+
+int HttpPipelinedConnection::ReadResponseHeaders(int pipeline_id,
+                                                 CompletionCallback* callback) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_SENT);
+  DCHECK(!ContainsKey(callback_map_, pipeline_id));
+  if (!usable_) {
+    return ERR_PIPELINE_EVICTION;
+  }
+  callback_map_[pipeline_id] = callback;
+  if (read_next_state_ == READ_STATE_NONE) {
+    read_next_state_ = READ_STATE_NEXT_HEADERS;
+    return DoReadHeadersLoop(OK);
+  } else {
+    return ERR_IO_PENDING;
+  }
+}
+
+int HttpPipelinedConnection::DoReadHeadersLoop(int result) {
+  int rv = result;
+  do {
+    ReadHeadersState state = read_next_state_;
+    read_next_state_ = READ_STATE_NONE;
+    switch (state) {
+      case READ_STATE_NEXT_HEADERS:
+        rv = DoReadNextHeaders(rv);
+        break;
+      case READ_STATE_COMPLETE:
+        rv = DoReadHeadersComplete(rv);
+        break;
+      case READ_STATE_STREAM_CLOSED:
+        rv = DoStreamClosed();
+        break;
+      case READ_STATE_UNUSABLE:
+        rv = DoEvictPendingReadHeaders(rv);
+        break;
+      default:
+        NOTREACHED() << "bad read state";
+        rv = ERR_FAILED;
+        break;
+    }
+  } while (rv != ERR_IO_PENDING && read_next_state_ != READ_STATE_NONE);
+  return rv;
+}
+
+void HttpPipelinedConnection::OnReadIOCallback(int result) {
+  DCHECK(read_user_callback_ != NULL);
+  DoReadHeadersLoop(result);
+}
+
+int HttpPipelinedConnection::DoReadNextHeaders(int result) {
+  DCHECK(!request_order_.empty());
+  int pipeline_id = request_order_.front();
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  if (stream_state_map_[pipeline_id] == STREAM_CLOSED) {
+    // Since nobody will read whatever data is on the pipeline associated with
+    // this request, we must shut down the rest of the pipeline.
+    read_next_state_ = READ_STATE_UNUSABLE;
+    return OK;
+  }
+  CallbackMap::iterator it = callback_map_.find(pipeline_id);
+  if (it == callback_map_.end()) {
+    return ERR_IO_PENDING;
+  }
+  DCHECK(ContainsKey(parser_map_, pipeline_id));
+  int rv = parser_map_[pipeline_id]->ReadResponseHeaders(
+      &read_io_callback_);
+  if (rv == ERR_IO_PENDING) {
+    read_next_state_ = READ_STATE_COMPLETE;
+    read_user_callback_ = it->second;
+  } else if (result == ERR_IO_PENDING) {
+    read_next_state_ = READ_STATE_STREAM_CLOSED;
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &HttpPipelinedConnection::FireUserCallback,
+            it->second,
+            rv));
+    rv = ERR_IO_PENDING;
+  }
+  callback_map_.erase(it);
+  return rv;
+}
+
+int HttpPipelinedConnection::DoReadHeadersComplete(int result) {
+  read_next_state_ = READ_STATE_STREAM_CLOSED;
+  if (read_user_callback_) {
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &HttpPipelinedConnection::FireUserCallback,
+            read_user_callback_,
+            result));
+    read_user_callback_ = NULL;
+  }
+  return ERR_IO_PENDING;
+}
+
+void HttpPipelinedConnection::FireUserCallback(CompletionCallback* callback,
+                                               int result) {
+  DCHECK(callback);
+  callback->Run(result);
+}
+
+int HttpPipelinedConnection::DoStreamClosed() {
+  DCHECK(!request_order_.empty());
+  request_order_.pop();
+  if (request_order_.empty()) {
+    read_next_state_ = READ_STATE_NONE;
+    return OK;
+  } else {
+    read_next_state_ = READ_STATE_NEXT_HEADERS;
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &HttpPipelinedConnection::DoReadHeadersLoop,
+            ERR_IO_PENDING));
+    return ERR_IO_PENDING;  // Wait for the task to fire.
+  }
+}
+
+int HttpPipelinedConnection::DoEvictPendingReadHeaders(int result) {
+  while (!request_order_.empty()) {
+    int evicted_id = request_order_.front();
+    request_order_.pop();
+    CallbackMap::iterator cb_it = callback_map_.find(evicted_id);
+    if (cb_it == callback_map_.end()) {
+      continue;
+    }
+    callback_map_.erase(cb_it);
+    if (stream_state_map_[evicted_id] == STREAM_CLOSED) {
+      continue;
+    }
+    cb_it->second->Run(ERR_PIPELINE_EVICTION);
+  }
+  DCHECK(callback_map_.empty());
+  read_next_state_ = READ_STATE_NONE;
+  return result;
+}
+
+void HttpPipelinedConnection::Close(int pipeline_id,
+                                    bool not_reusable) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  if (stream_state_map_[pipeline_id] == STREAM_CLOSED) {
+    // TODO(simonjam): Why is Close() sometimes called twice?
+    return;
+  }
+  stream_state_map_[pipeline_id] = STREAM_CLOSED;
+  // TODO(simonjam): If it hasn't sent yet, then we can just ignore it instead
+  // of shutting the pipeline down.
+  bool is_active_stream = !request_order_.empty() &&
+      pipeline_id == request_order_.front();
+  if (not_reusable || !is_active_stream) {
+    usable_ = false;
+  }
+  if (is_active_stream) {
+    read_next_state_ = READ_STATE_STREAM_CLOSED;
+    read_user_callback_ = NULL;
+    DoReadHeadersLoop(OK);
+  } else {
+    if (send_next_state_ == SEND_STATE_NONE) {
+      send_next_state_ = SEND_STATE_UNUSABLE;
+      DoSendRequestLoop(OK);
+    }
+    if (read_next_state_ == READ_STATE_NONE) {
+      read_next_state_ = READ_STATE_UNUSABLE;
+      DoReadHeadersLoop(OK);
+    }
+  }
+}
+
+int HttpPipelinedConnection::ReadResponseBody(int pipeline_id,
+                                              IOBuffer* buf, int buf_len,
+                                              CompletionCallback* callback) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK(!request_order_.empty());
+  DCHECK(pipeline_id == request_order_.front());
+  DCHECK(ContainsKey(parser_map_, pipeline_id));
+  return parser_map_[pipeline_id]->ReadResponseBody(buf, buf_len, callback);
+}
+
+uint64 HttpPipelinedConnection::GetUploadProgress(int pipeline_id) const {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK(ContainsKey(parser_map_, pipeline_id));
+  return parser_map_.find(pipeline_id)->second->GetUploadProgress();
+}
+
+HttpResponseInfo* HttpPipelinedConnection::GetResponseInfo(int pipeline_id) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK(ContainsKey(parser_map_, pipeline_id));
+  return parser_map_[pipeline_id]->GetResponseInfo();
+}
+
+bool HttpPipelinedConnection::IsResponseBodyComplete(int pipeline_id) const {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK(ContainsKey(parser_map_, pipeline_id));
+  return parser_map_.find(pipeline_id)->second->IsResponseBodyComplete();
+}
+
+bool HttpPipelinedConnection::CanFindEndOfResponse(int pipeline_id) const {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK(ContainsKey(parser_map_, pipeline_id));
+  return parser_map_.find(pipeline_id)->second->CanFindEndOfResponse();
+}
+
+bool HttpPipelinedConnection::IsMoreDataBuffered(int pipeline_id) const {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  return read_buf_->offset();
+}
+
+bool HttpPipelinedConnection::IsConnectionReused(int pipeline_id) const {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  if (pipeline_id > 1) {
+    return true;
+  }
+  ClientSocketHandle::SocketReuseType reuse_type = connection_->reuse_type();
+  return connection_->is_reused() ||
+         reuse_type == ClientSocketHandle::UNUSED_IDLE;
+}
+
+void HttpPipelinedConnection::SetConnectionReused(int pipeline_id) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  connection_->set_is_reused(true);
+}
+
+void HttpPipelinedConnection::GetSSLInfo(int pipeline_id,
+                                         SSLInfo* ssl_info) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK(ContainsKey(parser_map_, pipeline_id));
+  return parser_map_[pipeline_id]->GetSSLInfo(ssl_info);
+}
+
+void HttpPipelinedConnection::GetSSLCertRequestInfo(
+    int pipeline_id,
+    SSLCertRequestInfo* cert_request_info) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK(ContainsKey(parser_map_, pipeline_id));
+  return parser_map_[pipeline_id]->GetSSLCertRequestInfo(cert_request_info);
+}
+
+}  // namespace net