Move the "wait for QUIC handshake confirmation" logic to QuicChromiumClientSession::StreamRequest

net/quic/chromium/quic_chromium_client_session.cc

 // Copyright (c) 2012 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/quic/chromium/quic_chromium_client_session.h"
 
 #include <utility>
 
 #include "base/callback_helpers.h"
 #include "base/location.h"
@@ skipping 169 matching lines @@
   const GURL& GetURL() const override { return request_url_; }
 
  private:
   base::WeakPtr<QuicChromiumClientSession> session_;
   const GURL request_url_;
 };
 
 }  // namespace
 
 QuicChromiumClientSession::StreamRequest::StreamRequest(
-    const base::WeakPtr<QuicChromiumClientSession>& session)
-    : session_(session), stream_(nullptr) {}
+    const base::WeakPtr<QuicChromiumClientSession>& session,
+    bool requires_confirmation)
+    : session_(session),
+      requires_confirmation_(requires_confirmation),
+      stream_(nullptr),
+      next_state_(STATE_NONE),
+      weak_factory_(this) {}
 
 QuicChromiumClientSession::StreamRequest::~StreamRequest() {
   if (stream_)
     stream_->Reset(QUIC_STREAM_CANCELLED);
 
   if (session_)
     session_->CancelRequest(this);
 }
 
 int QuicChromiumClientSession::StreamRequest::StartRequest(
     const CompletionCallback& callback) {
   DCHECK(session_);
-  int rv = session_->TryCreateStream(this);
-  if (rv == ERR_IO_PENDING) {
+
+  next_state_ = STATE_WAIT_FOR_CONFIRMATION;
+  int rv = DoLoop(OK);
+  if (rv == ERR_IO_PENDING)
     callback_ = callback;
-  } else {
-    session_.reset();
-  }
 
   return rv;
 }
 
 QuicChromiumClientStream*
 QuicChromiumClientSession::StreamRequest::ReleaseStream() {
   DCHECK(stream_);
   QuicChromiumClientStream* stream = stream_;
   stream_ = nullptr;
   return stream;
 }
 
 void QuicChromiumClientSession::StreamRequest::OnRequestCompleteSuccess(
     QuicChromiumClientStream* stream) {
+  DCHECK_EQ(STATE_REQUEST_STREAM_COMPLETE, next_state_);
   session_.reset();
   stream_ = stream;
-  base::ResetAndReturn(&callback_).Run(OK);
+  // This method is called even when the request completes synchronously.
+  if (callback_)
+    DoCallback(OK);
 }
 
 void QuicChromiumClientSession::StreamRequest::OnRequestCompleteFailure(
     int rv) {
+  DCHECK_EQ(STATE_REQUEST_STREAM_COMPLETE, next_state_);
   session_.reset();
+  // This method is called even when the request completes synchronously.
+  if (callback_)
+    DoCallback(rv);
+}
+
+void QuicChromiumClientSession::StreamRequest::OnIOComplete(int rv) {
+  rv = DoLoop(rv);
+
+  if (rv != ERR_IO_PENDING && !callback_.is_null()) {
+    DoCallback(rv);
+  }
+}
+
+void QuicChromiumClientSession::StreamRequest::DoCallback(int rv) {
+  CHECK_NE(rv, ERR_IO_PENDING);
+  CHECK(!callback_.is_null());
+
+  // The client callback can do anything, including destroying this class,
+  // so any pending callback must be issued after everything else is done.
   base::ResetAndReturn(&callback_).Run(rv);
 }
 
+int QuicChromiumClientSession::StreamRequest::DoLoop(int rv) {
+  do {
+    State state = next_state_;
+    next_state_ = STATE_NONE;
+    switch (state) {
+      case STATE_WAIT_FOR_CONFIRMATION:
+        CHECK_EQ(OK, rv);
+        rv = DoWaitForConfirmation();
+        break;
+      case STATE_WAIT_FOR_CONFIRMATION_COMPLETE:
+        rv = DoWaitForConfirmationComplete(rv);
+        break;
+      case STATE_REQUEST_STREAM:
+        CHECK_EQ(OK, rv);
+        rv = DoRequestStream();
+        break;
+      case STATE_REQUEST_STREAM_COMPLETE:
+        rv = DoRequestStreamComplete(rv);
+        break;
+      default:
+        NOTREACHED() << "next_state_: " << next_state_;
+        break;
+    }
+  } while (next_state_ != STATE_NONE && next_state_ && rv != ERR_IO_PENDING);
+
+  return rv;
+}
+
+int QuicChromiumClientSession::StreamRequest::DoWaitForConfirmation() {
+  next_state_ = STATE_WAIT_FOR_CONFIRMATION_COMPLETE;
+  if (requires_confirmation_) {
+    return session_->WaitForHandshakeConfirmation(
+        base::Bind(&QuicChromiumClientSession::StreamRequest::OnIOComplete,
+                   weak_factory_.GetWeakPtr()));
+  }
+
+  return OK;
+}
+
+int QuicChromiumClientSession::StreamRequest::DoWaitForConfirmationComplete(
+    int rv) {
+  DCHECK_NE(ERR_IO_PENDING, rv);
+  if (rv < 0)
+    return rv;
+
+  next_state_ = STATE_REQUEST_STREAM;
+  return OK;
+}
+
+int QuicChromiumClientSession::StreamRequest::DoRequestStream() {
+  next_state_ = STATE_REQUEST_STREAM_COMPLETE;
+
+  return session_->TryCreateStream(this);
+}
+
+int QuicChromiumClientSession::StreamRequest::DoRequestStreamComplete(int rv) {
+  DCHECK(rv == OK || !stream_);
+  session_.reset();
+
+  return rv;
+}
+
 QuicChromiumClientSession::QuicChromiumClientSession(
     QuicConnection* connection,
     std::unique_ptr<DatagramClientSocket> socket,
     QuicStreamFactory* stream_factory,
     QuicCryptoClientStreamFactory* crypto_client_stream_factory,
     QuicClock* clock,
     TransportSecurityState* transport_security_state,
     std::unique_ptr<QuicServerInfo> server_info,
     const QuicServerId& server_id,
     bool require_confirmation,
@@ skipping 55 matching lines @@
                                    kAdditionalOverheadForIPv6);
   }
   connect_timing_.dns_start = dns_resolution_start_time;
   connect_timing_.dns_end = dns_resolution_end_time;
 }
 
 QuicChromiumClientSession::~QuicChromiumClientSession() {
   DCHECK(callback_.is_null());
 
   net_log_.EndEvent(NetLogEventType::QUIC_SESSION);
+  DCHECK(waiting_for_confirmation_callbacks_.empty());
   if (!dynamic_streams().empty())
     RecordUnexpectedOpenStreams(DESTRUCTOR);
   if (!observers_.empty())
     RecordUnexpectedObservers(DESTRUCTOR);
   if (!going_away_)
     RecordUnexpectedNotGoingAway(DESTRUCTOR);
 
   while (!dynamic_streams().empty() || !observers_.empty() ||
          !stream_requests_.empty()) {
     // The session must be closed before it is destroyed.
@@ skipping 128 matching lines @@
   DCHECK(!base::ContainsKey(observers_, observer));
   observers_.insert(observer);
 }
 
 void QuicChromiumClientSession::RemoveObserver(Observer* observer) {
   DCHECK(base::ContainsKey(observers_, observer));
   observers_.erase(observer);
 }
 
 std::unique_ptr<QuicChromiumClientSession::StreamRequest>
-QuicChromiumClientSession::CreateStreamRequest() {
+QuicChromiumClientSession::CreateStreamRequest(bool requires_confirmation) {
   // base::MakeUnique does not work because the StreamRequest constructor
   // is private.
   return std::unique_ptr<StreamRequest>(
-      new StreamRequest(weak_factory_.GetWeakPtr()));
+      new StreamRequest(weak_factory_.GetWeakPtr(), requires_confirmation));
+}
+
+int QuicChromiumClientSession::WaitForHandshakeConfirmation(
+    const CompletionCallback& callback) {
+  if (!connection()->connected())
+    return ERR_CONNECTION_CLOSED;
+
+  if (IsCryptoHandshakeConfirmed())
+    return OK;
+
+  waiting_for_confirmation_callbacks_.push_back(callback);
+  return ERR_IO_PENDING;
 }
 
 int QuicChromiumClientSession::TryCreateStream(StreamRequest* request) {
   if (goaway_received()) {
     DVLOG(1) << "Going away.";
     return ERR_CONNECTION_CLOSED;
   }
 
   if (!connection()->connected()) {
     DVLOG(1) << "Already closed.";
@@ skipping 384 matching lines @@
           "Net.QuicSession.HostResolution.HandshakeConfirmedTime",
           base::TimeTicks::Now() - connect_timing_.dns_end);
     }
 
     ObserverSet::iterator it = observers_.begin();
     while (it != observers_.end()) {
       Observer* observer = *it;
       ++it;
       observer->OnCryptoHandshakeConfirmed();
     }
+
+    NotifyRequestsOfConfirmation(OK);
   }
   QuicSpdySession::OnCryptoHandshakeEvent(event);
 }
 
 void QuicChromiumClientSession::OnCryptoHandshakeMessageSent(
     const CryptoHandshakeMessage& message) {
   logger_->OnCryptoHandshakeMessageSent(message);
 }
 
 void QuicChromiumClientSession::OnCryptoHandshakeMessageReceived(
@@ skipping 122 matching lines @@
     base::ResetAndReturn(&callback_).Run(ERR_QUIC_PROTOCOL_ERROR);
   }
 
   for (auto& socket : sockets_) {
     socket->Close();
   }
   DCHECK(dynamic_streams().empty());
   CloseAllStreams(ERR_UNEXPECTED);
   CloseAllObservers(ERR_UNEXPECTED);
   CancelAllRequests(ERR_CONNECTION_CLOSED);
+  NotifyRequestsOfConfirmation(ERR_CONNECTION_CLOSED);
   NotifyFactoryOfSessionClosedLater();
 }
 
 void QuicChromiumClientSession::OnSuccessfulVersionNegotiation(
     const QuicVersion& version) {
   logger_->OnSuccessfulVersionNegotiation(version);
   QuicSpdySession::OnSuccessfulVersionNegotiation(version);
 
   ObserverSet::iterator it = observers_.begin();
   while (it != observers_.end()) {
@@ skipping 239 matching lines @@
   UMA_HISTOGRAM_COUNTS_1000("Net.QuicSession.AbortedPendingStreamRequests",
                             stream_requests_.size());
 
   while (!stream_requests_.empty()) {
     StreamRequest* request = stream_requests_.front();
     stream_requests_.pop_front();
     request->OnRequestCompleteFailure(net_error);
   }
 }
 
+void QuicChromiumClientSession::NotifyRequestsOfConfirmation(int net_error) {
+  // Post tasks to avoid reentrancy.
+  for (auto callback : waiting_for_confirmation_callbacks_)
+    task_runner_->PostTask(FROM_HERE, base::Bind(callback, net_error));
+
+  waiting_for_confirmation_callbacks_.clear();
+}
+
 std::unique_ptr<base::Value> QuicChromiumClientSession::GetInfoAsValue(
     const std::set<HostPortPair>& aliases) {
   std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
   dict->SetString("version", QuicVersionToString(connection()->version()));
   dict->SetInteger("open_streams", GetNumOpenOutgoingStreams());
   std::unique_ptr<base::ListValue> stream_list(new base::ListValue());
   for (DynamicStreamMap::const_iterator it = dynamic_streams().begin();
        it != dynamic_streams().end(); ++it) {
     stream_list->AppendString(base::UintToString(it->second->id()));
   }
@@ skipping 210 matching lines @@
 }
 
 size_t QuicChromiumClientSession::EstimateMemoryUsage() const {
   // TODO(xunjieli): Estimate |crypto_stream_|, QuicSpdySession's
   // QuicHeaderList, QuicSession's QuiCWriteBlockedList, open streams and
   // unacked packet map.
   return base::trace_event::EstimateMemoryUsage(packet_readers_);
 }
 
 }  // namespace net