Makes migration on write error asynchronous to avoid reentrancy issues

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 219 matching lines @@
       logger_(new QuicConnectionLogger(this,
                                        connection_description,
                                        std::move(socket_performance_watcher),
                                        net_log_)),
       going_away_(false),
       port_migration_detected_(false),
       disabled_reason_(QUIC_DISABLED_NOT),
       token_binding_signatures_(kTokenBindingSignatureMapSize),
       streams_pushed_count_(0),
       streams_pushed_and_claimed_count_(0),
-      error_code_from_rewrite_(OK),
-      use_error_code_from_rewrite_(false),
+      migration_pending_(false),
+      write_pending_(false),
       weak_factory_(this) {
   sockets_.push_back(std::move(socket));
   packet_readers_.push_back(base::WrapUnique(new QuicChromiumPacketReader(
       sockets_.back().get(), clock, this, yield_after_packets,
       yield_after_duration, net_log_)));
   crypto_stream_.reset(
       crypto_client_stream_factory->CreateQuicCryptoClientStream(
           server_id, this, base::WrapUnique(new ProofVerifyContextChromium(
                                cert_verify_flags, net_log_)),
           crypto_config));
@@ skipping 693 matching lines @@
 void QuicChromiumClientSession::OnSuccessfulVersionNegotiation(
     const QuicVersion& version) {
   logger_->OnSuccessfulVersionNegotiation(version);
   QuicSpdySession::OnSuccessfulVersionNegotiation(version);
 }
 
 int QuicChromiumClientSession::HandleWriteError(
     int error_code,
     scoped_refptr<StringIOBuffer> packet) {
   DCHECK(packet != nullptr);
-  use_error_code_from_rewrite_ = false;
-  if (stream_factory_) {
-    stream_factory_->MaybeMigrateSingleSession(this, WRITE_ERROR, packet);
+  DCHECK_NE(ERR_IO_PENDING, error_code);
+  DCHECK_GT(0, error_code);
+  DCHECK(!migration_pending_);
+  DCHECK(packet_ == nullptr);
+
+  // Post a task to migrate the session onto a new network.
+  task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&QuicChromiumClientSession::MigrateSessionOnWriteError,
+                 weak_factory_.GetWeakPtr()));
+
+  // Store packet in the session since the actual migration and packet rewrite
+  // can happen via this posted task or via an async network notification.
+  packet_ = packet;
+  migration_pending_ = true;

[Ryan Hamilton, 2016/09/12 00:52:18]

This is the only place migration_pending_ is set to true, right? Looks like the
only purpose of this is to prevent MigrateSessionOnWriteError from doing work if
the NCN event fired. Do I have this right? If so perhaps a better name might be:

  need_to_migrate_for_write_error_

Or some such. (this is a nit ;>)

Oh! Here's another possibly better idea. When you post the task for
MigrateSessionOnError, pass in sockets_.size(). Then in MigrateSessionOnError
check to see if the current sockets_.size() == the passed in value. Then you
don't need an additional state variable.

[Jana, 2016/09/12 21:08:02]

This is used (later in the pause CL) by the NCN code path as well, so this won't
be a good name then.
Per conversation, leaving this for later.

+  static_cast<QuicChromiumPacketWriter*>(connection()->writer())
+      ->set_write_blocked(true);

[Ryan Hamilton, 2016/09/12 00:52:18]

This is not needed is it? By returning ERR_IO_PENDING below the writer is
guaranteed to be write blocked.

[Jana, 2016/09/12 21:08:02]

Oh of course -- I am not sure why I added this here. Removed.

+
+  // Cause the packet writer to return ERR_IO_PENDING and block so
+  // that the actual migration happens from the message loop instead
+  // of under the call stack of QuicConnection::WritePacket.
+  return ERR_IO_PENDING;
+}
+
+void QuicChromiumClientSession::MigrateSessionOnWriteError() {
+  // If migration_pending_ is false, an earlier task completed migration.
+  if (!migration_pending_)
+    return;
+
+  if (stream_factory_ != nullptr &&
+      stream_factory_->MaybeMigrateSingleSession(this, WRITE_ERROR) ==
+          MigrationResult::SUCCESS)
+    return;

[Ryan Hamilton, 2016/09/12 00:52:18]

nit: {}s needed.

[Jana, 2016/09/12 21:08:02]

Done.

+
+  // Close the connection if migration failed. Do not cause a
+  // connection close packet to be sent since socket may be borked.
+  connection()->CloseConnection(QUIC_PACKET_WRITE_ERROR,
+                                "Write and subsequent migration failed",
+                                ConnectionCloseBehavior::SILENT_CLOSE);
+}
+
+void QuicChromiumClientSession::WriteToNewSocket() {
+  // If write_pending_ is false, an earlier task wrote to the new socket.
+  if (!write_pending_)
+    return;

[Ryan Hamilton, 2016/09/12 01:32:57]

You can take this or leave it... If we removed this early return and this method
ended up firing twice (possible but unlikely) would the worst thing that would
happen be that an extra ping gets sent? If so you might consider just dropping
this variable altogether and skipping this check. 

Or totally keep it :> It's completely your call.

[Jana, 2016/09/12 21:08:02]

I did have the code doing that earlier, but enforcing one packet instead of two
makes reasoning about behavior consistent (If there's a write error, packet is
resent, otherwise PING is sent), and that helps with keeping tests more
consistent across interleaved write error + NCN events. I'd prefer to keep it.

+  // Prevent any pending migration or write tasks from executing.
+  write_pending_ = false;
+  migration_pending_ = false;
+
+  static_cast<QuicChromiumPacketWriter*>(connection()->writer())
+      ->set_write_blocked(false);
+  DCHECK(!connection()->writer()->IsWriteBlocked());
+
+  if (packet_ == nullptr) {
+    connection()->SendPing();
+    // The connection may have been blocked before the migration
+    // started. Unblock the connection if sending the PING packet did
+    // not leave the writer blocked.

[Ryan Hamilton, 2016/09/12 00:52:18]

I think the best thing to do is to call OnCanWrite() before calling SendPing()
unless there is some really compelling reason to do otherwise. Recall that the
Async Packet Writer contract is to call OnCanWrite after a write completes
async. So by calling SendPing first, we're sending a ping "in the middle" of
sending the previous packet.

[Jana, 2016/09/12 21:08:02]

Hm. Yeah, this does make reasoning simpler. Done.

+    if (!connection()->writer()->IsWriteBlocked())
+      connection()->OnCanWrite();
+    return;
   }
-  return use_error_code_from_rewrite_ ? error_code_from_rewrite_ : error_code;
+
+  // Set packet_ to null first. We cannot set packet_ to null after
+  // the following write since the write may result in packet_ being
+  // reused via a write error.

[Ryan Hamilton, 2016/09/12 00:52:18]

Avoid first person in comments:

  // Set packet to null before calling WritePacketToSocket because that method
may set packet_ if there is a write error.

(or somesuch)

[Jana, 2016/09/12 21:08:02]

Done.

+  scoped_refptr<StringIOBuffer> packet = packet_;
+  packet_ = nullptr;
+
+  // The connection is waiting for the original write to complete
+  // asynchronously. The new writer will notify the connection if the
+  // write below completes asynchronously, but a synchronous competion
+  // must be propagated back to the connection here.
+  WriteResult result =
+      static_cast<QuicChromiumPacketWriter*>(connection()->writer())
+          ->WritePacketToSocket(packet);
+
+  if (result.error_code == ERR_IO_PENDING)
+    return;
+  // All write errors should be mapped into ERR_IO_PENDING by
+  // HandleWriteError.
+  DCHECK(result.error_code >= 0);

[Ryan Hamilton, 2016/09/12 00:52:18]

nit: DCHECK_LT(0, result.error_code);

[Jana, 2016/09/12 21:08:02]

Done.

+  connection()->OnCanWrite();
 }
 
 void QuicChromiumClientSession::OnWriteError(int error_code) {
   DCHECK_NE(ERR_IO_PENDING, error_code);
   DCHECK_GT(0, error_code);
   connection()->OnWriteError(error_code);
 }
 
 void QuicChromiumClientSession::OnWriteUnblocked() {
   connection()->OnCanWrite();
 }
 
 void QuicChromiumClientSession::OnPathDegrading() {
   if (stream_factory_) {
-    stream_factory_->MaybeMigrateSingleSession(this, EARLY_MIGRATION, nullptr);
+    stream_factory_->MaybeMigrateSingleSession(this, EARLY_MIGRATION);
   }
 }
 
 bool QuicChromiumClientSession::HasOpenDynamicStreams() const {
   return QuicSession::HasOpenDynamicStreams() ||
          GetNumDrainingOutgoingStreams() > 0;
 }
 
 void QuicChromiumClientSession::OnProofValid(
     const QuicCryptoClientConfig::CachedState& cached) {
@@ skipping 186 matching lines @@
   going_away_ = true;
   DCHECK_EQ(0u, GetNumActiveStreams());
   // Will delete |this|.
   if (stream_factory_)
     stream_factory_->OnSessionClosed(this);
 }
 
 bool QuicChromiumClientSession::MigrateToSocket(
     std::unique_ptr<DatagramClientSocket> socket,
     std::unique_ptr<QuicChromiumPacketReader> reader,
-    std::unique_ptr<QuicChromiumPacketWriter> writer,
-    scoped_refptr<StringIOBuffer> packet) {
+    std::unique_ptr<QuicChromiumPacketWriter> writer) {
   DCHECK_EQ(sockets_.size(), packet_readers_.size());
-  if (sockets_.size() >= kMaxReadersPerQuicSession) {
+  if (sockets_.size() >= kMaxReadersPerQuicSession)
     return false;
-  }
+
   // TODO(jri): Make SetQuicPacketWriter take a scoped_ptr.
   packet_readers_.push_back(std::move(reader));
   sockets_.push_back(std::move(socket));
   StartReading();
-  QuicChromiumPacketWriter* raw_writer = writer.get();
   connection()->SetQuicPacketWriter(writer.release(), /*owns_writer=*/true);
-  if (packet == nullptr) {
-    connection()->SendPing();
-    return true;
-  }
-  // Packet rewrite after migration on socket write error.
-  error_code_from_rewrite_ = raw_writer->WritePacketToSocket(packet.get());
-  use_error_code_from_rewrite_ = true;
+
+  // Post task to write the pending packet or a PING packet to the new
+  // socket.  Also block the writer to prevent is being used until

[Ryan Hamilton, 2016/09/12 00:52:18]

Can you add some language here which explains that the motivation is to avoid
reentrancy issues if there is a write error?

[Jana, 2016/09/12 21:08:02]

Done.

+  // WriteToNewSocket completes.

[Ryan Hamilton, 2016/09/12 00:52:18]

nit: move this sentence down to the call to block the writer so it's next to the
code.

[Jana, 2016/09/12 21:08:02]

Done.

+  task_runner_->PostTask(
+      FROM_HERE, base::Bind(&QuicChromiumClientSession::WriteToNewSocket,
+                            weak_factory_.GetWeakPtr()));
+
+  // Block the writer until the task posted is executed.
+  static_cast<QuicChromiumPacketWriter*>(connection()->writer())
+      ->set_write_blocked(true);
+
+  // Migration completed and write task posted.
+  migration_pending_ = false;
+  write_pending_ = true;
   return true;
 }
 
 void QuicChromiumClientSession::PopulateNetErrorDetails(
     NetErrorDetails* details) {
   details->quic_port_migration_detected = port_migration_detected_;
 }
 
 const DatagramClientSocket* QuicChromiumClientSession::GetDefaultSocket()
     const {
@@ skipping 27 matching lines @@
 }
 
 void QuicChromiumClientSession::DeletePromised(
     QuicClientPromisedInfo* promised) {
   if (IsOpenStream(promised->id()))
     streams_pushed_and_claimed_count_++;
   QuicClientSessionBase::DeletePromised(promised);
 }
 
 }  // namespace net