Basic HTTP pipelining support

net/http/http_pipelined_connection_impl.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_impl.h"
+
+#include "base/message_loop.h"
+#include "base/stl_util.h"
+#include "net/base/io_buffer.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 {
+
+HttpPipelinedConnectionImpl::HttpPipelinedConnectionImpl(
+    ClientSocketHandle* connection,
+    HttpPipelinedConnection::Delegate* delegate,
+    const SSLConfig& used_ssl_config,
+    const ProxyInfo& used_proxy_info,
+    const BoundNetLog& net_log,
+    bool was_npn_negotiated)
+    : delegate_(delegate),
+      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),
+      completed_one_request_(false),
+      ALLOW_THIS_IN_INITIALIZER_LIST(method_factory_(this)),
+      send_next_state_(SEND_STATE_NONE),
+      ALLOW_THIS_IN_INITIALIZER_LIST(send_io_callback_(
+          this, &HttpPipelinedConnectionImpl::OnSendIOCallback)),
+      send_user_callback_(NULL),
+      read_next_state_(READ_STATE_NONE),
+      ALLOW_THIS_IN_INITIALIZER_LIST(read_io_callback_(
+          this, &HttpPipelinedConnectionImpl::OnReadIOCallback)),
+      read_user_callback_(NULL) {
+  DCHECK(connection_.get());
+}
+
+HttpPipelinedConnectionImpl::~HttpPipelinedConnectionImpl() {
+  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();
+}
+
+HttpStream* HttpPipelinedConnectionImpl::CreateNewStream() {
+  int pipeline_id = next_pipeline_id_++;
+  DCHECK(pipeline_id);
+  HttpPipelinedStream* stream = new HttpPipelinedStream(this, pipeline_id);
+  stream_state_map_.insert(std::make_pair(pipeline_id, STREAM_CREATED));
+  return stream;
+}
+
+void HttpPipelinedConnectionImpl::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(
+            &HttpPipelinedConnectionImpl::FillPipeline));
+  }
+}
+
+void HttpPipelinedConnectionImpl::FillPipeline() {
+  delegate_->OnPipelineHasCapacity(this);
+}
+
+void HttpPipelinedConnectionImpl::OnStreamDeleted(int pipeline_id) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  Close(pipeline_id, false);
+  DCHECK(!ContainsKey(callback_map_, pipeline_id));
+
+  if (stream_state_map_[pipeline_id] != STREAM_CREATED &&
+      stream_state_map_[pipeline_id] != STREAM_UNUSED) {
+    DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_CLOSED);
+    DCHECK(ContainsKey(parser_map_, pipeline_id));
+
+    ParserMap::iterator it = parser_map_.find(pipeline_id);
+    delete it->second;
+    parser_map_.erase(it);
+  }
+  DCHECK(!ContainsKey(parser_map_, pipeline_id));
+  stream_state_map_.erase(pipeline_id);
+
+  MessageLoop::current()->PostTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(
+          &HttpPipelinedConnectionImpl::FillPipeline));
+}
+
+int HttpPipelinedConnectionImpl::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 HttpPipelinedConnectionImpl::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: " << state;
+        rv = ERR_FAILED;
+        break;
+    }
+  } while (rv != ERR_IO_PENDING && send_next_state_ != SEND_STATE_NONE);
+  return rv;
+}
+
+void HttpPipelinedConnectionImpl::OnSendIOCallback(int result) {
+  DCHECK(send_user_callback_);
+  DoSendRequestLoop(result);
+}
+
+int HttpPipelinedConnectionImpl::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_);
+  // |result| == ERR_IO_PENDING means this function was *not* called on the same
+  // stack as SendRequest(). That means we returned ERR_IO_PENDING to
+  // SendRequest() earlier and will need to invoke its callback.
+  if (result == ERR_IO_PENDING || rv == ERR_IO_PENDING) {
+    send_user_callback_ = deferred_request.callback;
+  }
+  stream_state_map_[deferred_request.pipeline_id] = STREAM_SENDING;
+  send_next_state_ = SEND_STATE_COMPLETE;
+  return rv;
+}
+
+int HttpPipelinedConnectionImpl::DoSendComplete(int result) {
+  DCHECK(!deferred_request_queue_.empty());
+  const DeferredSendRequest& deferred_request = deferred_request_queue_.front();
+  DCHECK_EQ(stream_state_map_[deferred_request.pipeline_id], STREAM_SENDING);
+  request_order_.push(deferred_request.pipeline_id);
+  stream_state_map_[deferred_request.pipeline_id] = STREAM_SENT;
+  deferred_request_queue_.pop();
+  if (result == ERR_SOCKET_NOT_CONNECTED && completed_one_request_) {
+    result = ERR_PIPELINE_EVICTION;
+  }
+  if (send_user_callback_) {
+    CompletionCallback* callback = send_user_callback_;
+    send_user_callback_ = NULL;
+    callback->Run(result);

[mmenke, 2011/09/15 19:28:16]

Maybe we should mark this as unusable before running the callback?  Otherwise,
it may try to get a new pipeline, get us, and then be evicted again...  I think
things turn out fine, but seems a little funky.

Might want to test some of the pipeline assignment logic, too.

[James Simonsen, 2011/09/17 01:23:02]

Yeah, that sounds good.
The assignment logic should be tested in http_pipelined_host_unittest.cc.

Do you mean to add integration tests? That seems like a good idea. The
relationship between objects and the order they do things is important. The unit
tests try to simulate that, but probably not perfectly.

I can look into some integration tests next week. I think the important pieces
are the transaction, pool, host, pipeline, and stream.

[mmenke, 2011/09/17 01:41:22]

Yea, I was just thinking that we don't test reassignment to a new pipelined
connection after eviction.  Think you're right about the components we'd need to
adequately test that.

On 2011/09/17 01:23:02, James Simonsen wrote:

+  }
+  if (result < OK) {
+    send_next_state_ = SEND_STATE_UNUSABLE;
+    usable_ = false;
+    return result;
+  }
+  if (deferred_request_queue_.empty()) {
+    send_next_state_ = SEND_STATE_NONE;
+    return OK;
+  }
+  send_next_state_ = SEND_STATE_NEXT_REQUEST;
+  MessageLoop::current()->PostTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(
+          &HttpPipelinedConnectionImpl::DoSendRequestLoop,
+          ERR_IO_PENDING));
+  return ERR_IO_PENDING;  // Wait for the task to fire.
+}
+
+int HttpPipelinedConnectionImpl::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) {
+      evicted_send.callback->Run(ERR_PIPELINE_EVICTION);
+    }
+    deferred_request_queue_.pop();
+  }
+  return result;
+}
+
+int HttpPipelinedConnectionImpl::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;
+  }
+  stream_state_map_[pipeline_id] = STREAM_READ_PENDING;
+  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 HttpPipelinedConnectionImpl::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_WAITING_FOR_CLOSE:
+        rv = DoReadWaitingForClose(rv);
+        return rv;
+      case READ_STATE_STREAM_CLOSED:
+        rv = DoReadStreamClosed();
+        break;
+      case READ_STATE_UNUSABLE:
+        rv = DoEvictPendingReadHeaders(rv);
+        break;
+      case READ_STATE_NONE:
+        break;
+      default:
+        NOTREACHED() << "bad read state";
+        rv = ERR_FAILED;
+        break;
+    }
+  } while (rv != ERR_IO_PENDING && read_next_state_ != READ_STATE_NONE);
+  return rv;
+}
+
+void HttpPipelinedConnectionImpl::OnReadIOCallback(int result) {
+  DCHECK(read_user_callback_ != NULL);
+  DoReadHeadersLoop(result);
+}
+
+int HttpPipelinedConnectionImpl::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 closed 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));
+
+  if (result == ERR_IO_PENDING) {
+    DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_ACTIVE);
+  } else {
+    DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_READ_PENDING);
+    stream_state_map_[pipeline_id] = STREAM_ACTIVE;
+  }
+
+  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 (rv < OK) {
+    read_next_state_ = READ_STATE_WAITING_FOR_CLOSE;
+    if (rv == ERR_SOCKET_NOT_CONNECTED && completed_one_request_)
+      rv = ERR_PIPELINE_EVICTION;
+  } else {
+    DCHECK_LE(OK, rv);
+    read_next_state_ = READ_STATE_WAITING_FOR_CLOSE;
+  }
+
+  // |result| == ERR_IO_PENDING means this function was *not* called on the same
+  // stack as ReadResponseHeaders(). That means we returned ERR_IO_PENDING to
+  // ReadResponseHeaders() earlier and now need to invoke its callback.
+  if (rv != ERR_IO_PENDING && result == ERR_IO_PENDING) {
+    read_next_state_ = READ_STATE_WAITING_FOR_CLOSE;
+    read_user_callback_ = it->second;
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &HttpPipelinedConnectionImpl::FireReadUserCallback,
+            rv));
+  }
+  return rv;
+}
+
+int HttpPipelinedConnectionImpl::DoReadHeadersComplete(int result) {
+  read_next_state_ = READ_STATE_WAITING_FOR_CLOSE;
+  if (read_user_callback_) {
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &HttpPipelinedConnectionImpl::FireReadUserCallback,
+            result));
+  }
+  return result;
+}
+
+int HttpPipelinedConnectionImpl::DoReadWaitingForClose(int result) {
+  read_next_state_ = READ_STATE_WAITING_FOR_CLOSE;
+  return result;
+}
+
+int HttpPipelinedConnectionImpl::DoReadStreamClosed() {
+  DCHECK(!request_order_.empty());
+  int pipeline_id = request_order_.front();
+  DCHECK(ContainsKey(callback_map_, pipeline_id));
+  DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_CLOSED);
+  callback_map_.erase(pipeline_id);
+  request_order_.pop();
+  if (!usable_) {
+    read_next_state_ = READ_STATE_UNUSABLE;
+    return OK;
+  } else {
+    completed_one_request_ = true;
+    if (request_order_.empty() ||
+        !ContainsKey(callback_map_, request_order_.front())) {
+      read_next_state_ = READ_STATE_NONE;
+      return OK;
+    } else {
+      stream_state_map_[request_order_.front()] = STREAM_ACTIVE;
+      read_next_state_ = READ_STATE_NEXT_HEADERS;
+      MessageLoop::current()->PostTask(
+          FROM_HERE,
+          method_factory_.NewRunnableMethod(
+              &HttpPipelinedConnectionImpl::DoReadHeadersLoop,
+              ERR_IO_PENDING));
+      return ERR_IO_PENDING;  // Wait for the task to fire.
+    }
+  }
+}
+
+int HttpPipelinedConnectionImpl::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;
+    }
+    if (stream_state_map_[evicted_id] != STREAM_CLOSED) {
+      cb_it->second->Run(ERR_PIPELINE_EVICTION);
+    }
+    callback_map_.erase(cb_it);
+  }
+  DCHECK(callback_map_.empty());
+  read_next_state_ = READ_STATE_NONE;
+  return result;
+}
+
+void HttpPipelinedConnectionImpl::Close(int pipeline_id,
+                                        bool not_reusable) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  switch (stream_state_map_[pipeline_id]) {
+    case STREAM_CREATED:
+      stream_state_map_[pipeline_id] = STREAM_UNUSED;
+      break;
+
+    case STREAM_BOUND:
+      stream_state_map_[pipeline_id] = STREAM_CLOSED;
+      break;
+
+    case STREAM_SENDING:
+      usable_ = false;
+      stream_state_map_[pipeline_id] = STREAM_CLOSED;
+      send_user_callback_ = NULL;
+      send_next_state_ = SEND_STATE_UNUSABLE;
+      DoSendRequestLoop(OK);
+      break;
+
+    case STREAM_SENT:
+    case STREAM_READ_PENDING:
+      usable_ = false;
+      stream_state_map_[pipeline_id] = STREAM_CLOSED;
+      if (read_next_state_ == READ_STATE_NONE) {
+        read_next_state_ = READ_STATE_UNUSABLE;
+        DoReadHeadersLoop(OK);
+      }
+      break;
+
+    case STREAM_ACTIVE:
+      stream_state_map_[pipeline_id] = STREAM_CLOSED;
+      if (not_reusable) {
+        usable_ = false;
+      }
+      read_next_state_ = READ_STATE_STREAM_CLOSED;
+      read_user_callback_ = NULL;
+      DoReadHeadersLoop(OK);
+      break;
+
+    case STREAM_CLOSED:
+    case STREAM_UNUSED:
+      // TODO(simonjam): Why is Close() sometimes called twice?
+      break;
+
+    default:
+      NOTREACHED();
+      break;
+  }
+}
+
+int HttpPipelinedConnectionImpl::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 HttpPipelinedConnectionImpl::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* HttpPipelinedConnectionImpl::GetResponseInfo(
+    int pipeline_id) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  DCHECK(ContainsKey(parser_map_, pipeline_id));
+  return parser_map_[pipeline_id]->GetResponseInfo();
+}
+
+bool HttpPipelinedConnectionImpl::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 HttpPipelinedConnectionImpl::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 HttpPipelinedConnectionImpl::IsMoreDataBuffered(int pipeline_id) const {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  return read_buf_->offset();
+}
+
+bool HttpPipelinedConnectionImpl::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 HttpPipelinedConnectionImpl::SetConnectionReused(int pipeline_id) {
+  DCHECK(ContainsKey(stream_state_map_, pipeline_id));
+  connection_->set_is_reused(true);
+}
+
+void HttpPipelinedConnectionImpl::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 HttpPipelinedConnectionImpl::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);
+}
+
+void HttpPipelinedConnectionImpl::FireReadUserCallback(
+    int result) {
+  if (read_user_callback_) {
+    read_user_callback_->Run(result);
+    read_user_callback_ = NULL;
+  }
+}
+
+}  // namespace net