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,

[xunjieli, 2017/05/01 20:15:06]

Does |session| always outlive the StreamRequest? Will we have a case where
|session| weak ptr is invalidated? 
If not, we could also use a raw QuicChromiumClientSession* ptr here to simplify
the lifetime relationship between StreamRequest and the Session. Not suggesting
to do it in this CL, but I would like to check to see if this is a viable
option.

[Ryan Hamilton, 2017/05/01 21:42:33]

The StreamRequest may well outlive the session. The StreamRequest is owned by
the caller, not by the session so it's lifetime can not be bound by the session.
I don't think this is workable, but I'm working on a followup which gets rid of
the WeakPtr in favor of a "Handle", which hides the lifetime issues from the
callers. Hopefully this will be a good end-state for things which need to
interact with the session! *fingers crossed*

https://codereview.chromium.org/2334943002/ (not ready for review, but if you're
curious...)

[xunjieli, 2017/05/02 17:23:05]

Got it. That makes sense. The "Handle" approach is neat! I like the idea that
the Handle is guaranteed to be valid throughout.

+    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);
+  if (callback_)

[xunjieli, 2017/05/01 20:15:05]

I think callback_ is always valid, right? It's never null-ed once set in the
constructor. Same for line 239 and line 246.

[Ryan Hamilton, 2017/05/01 21:42:35]

Ah. Good question. No, it's not guaranteed to be valid because these methods do
not follow the "net-style" for async operations. They are invoked even if the
operation returns synchronously. (This is another opportunity for cleanup later!
So many yaks)

Added comments to clarify this.

[xunjieli, 2017/05/02 17:23:05]

Ah, that's subtle. Thanks for the comment.

+    DoCallback(OK);
 }
 
 void QuicChromiumClientSession::StreamRequest::OnRequestCompleteFailure(
     int rv) {
+  DCHECK_EQ(STATE_REQUEST_STREAM_COMPLETE, next_state_);
   session_.reset();
+  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_)

[xunjieli, 2017/05/01 20:15:05]

nit: add curly braces or use early return.

[Ryan Hamilton, 2017/05/01 21:42:35]

Done.

+    return session_->WaitForHandshakeConfirmation(
+        base::Bind(&QuicChromiumClientSession::StreamRequest::OnIOComplete,
+                   weak_factory_.GetWeakPtr()));
+
+  return OK;
+}
+
+int QuicChromiumClientSession::StreamRequest::DoWaitForConfirmationComplete(
+    int rv) {

[xunjieli, 2017/05/01 20:15:05]

nit: maybe add a DCHECK_NE(ERR_IO_PENDING, rv) here.

[Ryan Hamilton, 2017/05/01 21:42:35]

Done.

+  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 51 matching lines @@
   IPEndPoint address;
   if (socket && socket->GetLocalAddress(&address) == OK &&
       address.GetFamily() == ADDRESS_FAMILY_IPV6) {
     connection->SetMaxPacketLength(connection->max_packet_length() -
                                    kAdditionalOverheadForIPv6);
   }
   connect_timing_.dns_start = dns_resolution_start_time;
   connect_timing_.dns_end = dns_resolution_end_time;
 }
 
 QuicChromiumClientSession::~QuicChromiumClientSession() {

[xunjieli, 2017/05/01 20:15:06]

Maybe we can add a DCHECK(waiting_for_confirmation_callbacks_.empty()) here?

[Ryan Hamilton, 2017/05/01 21:42:35]

Done.

   DCHECK(callback_.is_null());
 
   net_log_.EndEvent(NetLogEventType::QUIC_SESSION);
   if (!dynamic_streams().empty())
     RecordUnexpectedOpenStreams(DESTRUCTOR);
   if (!observers_.empty())
     RecordUnexpectedObservers(DESTRUCTOR);
   if (!going_away_)
     RecordUnexpectedNotGoingAway(DESTRUCTOR);
 
@@ skipping 131 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();
     }
+
+    NotifyAllWaitingForConfirmation(OK);

[xunjieli, 2017/05/01 20:15:05]

I like the new notification API!

[Ryan Hamilton, 2017/05/01 21:42:34]

\o/ Thanks!

   }
   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);
+  NotifyAllWaitingForConfirmation(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 248 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::NotifyAllWaitingForConfirmation(int net_error) {
+  // Post tasks to avoid reentrancy.
+  for (auto callback : waiting_for_confirmation_callbacks_)

[xunjieli, 2017/05/01 20:15:06]

nit: need a pair of {} here for the for-loop.

[Ryan Hamilton, 2017/05/01 21:42:32]

Hm. I thought this was not needed because the body is a single line, by the same
rule for no {}s in one-line if()s. For example:

void SdchManager::ClearData() {
  blacklisted_domains_.clear();
  allow_latency_experiment_.clear();
  dictionaries_.clear();
  for (auto& observer : observers_)
    observer.OnClearDictionaries();
}

[xunjieli, 2017/05/02 17:23:05]

Got it. I didn't know that braces are optional for for-loops.
I doublechecked, and you are right. 
https://google.github.io/styleguide/cppguide.html says:
"Braces are optional for single-statement loops."

+    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