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),
+      should_write_block_(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;
+
+  // 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() {
+  // There are two async steps in the migration process due to write error:
+  // MigrateSessionOnWriteError (this method) and WriteToNewPacket.
+  // If migration_pending_ is false, a notification-triggered
+  // migration completed migration before this method was executed.
+  // If packet_ is null, a notification-triggered migration wrote the
+  // stashed packet on a new network post-migration before this method
+  // was executed. Either ways, do not trigger another migration.
+  if (migration_pending_ == false || packet_ == nullptr)
+    return;
+
+  if (stream_factory_ != nullptr &&
+      stream_factory_->MaybeMigrateSingleSession(this, WRITE_ERROR) ==
+          MigrationResult::SUCCESS)
+    return;
+
+  // 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() {
+  // This method can be called via a write error event or via an
+  // notifier (NetworkChangeNotifier) event. If there's only a notifier
+  // event that leads here, packet must be null and the connection is
+  // not write blocked.  If there's only a write error that leads

[Ryan Hamilton, 2016/09/11 15:38:42]

I'm not sure this statement is correct. Consider the following sequence of
events:

T0: Write starts, returns ERR_IO_PENDING, writer is write blocked
T1: NCN event fires which causes migration.

Doesn't that results in a connection and session which are expecting
OnCanWrite())?

In thinking about this more... When migration happens, an new socket and a new
writer are created. That new writer is not write blocked, obviously, because no
writes have happened. I think that the simplest approach to this situation would
be for MigrateToSocket() to call set_write_blocked(true) on the new writer. Then
it stays blocked (without a write ever happening) until the posted task runs.
Then when we get here, call set_write_blocked(false) and perform the IO via
either:

OnCanWrite()
SendPing()

or

WritePacketToSocket()
...
OnCanWrite()

What do you think?

[Jana, 2016/09/11 18:32:30]

So your example sequence of events is what I called interleaved events below...
FWIW, the code does exactly what you are suggesting I think. should_write_block
is set to false below and is set to true, though if you look at the latest
version of the code, this variable name has changed to write_pending_. I also
have a test to test for your example sequence of events
(MigrationOnNetworkDisconnectedWithWriteErrorQueued), and things work as
expected.

I've also removed this detailed comment in teh new version of the code, but if
you think it (or something like it) is useful, I'll bring it back.

+  // here, then packet must be non-null, since HandleWriteError stores
+  // a packet in the packet_ member. The packet_ member is cleared
+  // only in this method below.
+  //
+  // If the WriteError event and a notifier event are interleaved,
+  // leading to two WriteToNewSocket()s being posted and executed, it
+  // doesn't matter which WriteToNewSocket occurs first. The first one
+  // will write packet_ and unblock the connection (or close on
+  // error), and the second one will send a PING packet.
+
+  // Unblock the writer so it can be used.
+  should_write_block_ = false;
+
+  if (packet_ == nullptr) {
+    connection()->SendPing();
+    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.
+  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);
+  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();
 }
 
+bool QuicChromiumClientSession::ShouldWriteBlock() {
+  return should_write_block_;
+}
+
 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
+  // WriteToNewSocket completes.
+  task_runner_->PostTask(
+      FROM_HERE, base::Bind(&QuicChromiumClientSession::WriteToNewSocket,
+                            weak_factory_.GetWeakPtr()));
+  should_write_block_ = true;
+
+  // Migration completed.
+  migration_pending_ = false;
   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