Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(304)

Unified Diff: net/http/http_pipelined_connection.cc

Issue 7289006: Basic HTTP pipelining support (Closed) Base URL: svn://svn.chromium.org/chrome/trunk/src
Patch Set: Use HttpStreamFactoryImpl::Job Created 9 years, 5 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
Index: net/http/http_pipelined_connection.cc
diff --git a/net/http/http_pipelined_connection.cc b/net/http/http_pipelined_connection.cc
new file mode 100644
index 0000000000000000000000000000000000000000..b59c1d37a4fe3e69734a87b7f45176501b4d4d15
--- /dev/null
+++ b/net/http/http_pipelined_connection.cc
@@ -0,0 +1,478 @@
+// 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) {
+ 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

Powered by Google App Engine
This is Rietveld 408576698