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;
+
+  // 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;
+
+  // 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;
+  // Prevent any pending migration or write tasks from executing.
+  write_pending_ = false;
+  migration_pending_ = false;
+
+  DCHECK(!connection()->writer()->IsWriteBlocked());
+
+  if (packet_ == nullptr) {
+    connection()->SendPing();

[Ryan Hamilton, 2016/09/11 19:24:12]

Maybe I'm not looking in the right place, but I don't see OnCanWrite here?

[Jana, 2016/09/11 19:44:04]

If there was a write block just before this WriteToNewSocket is executed,
HandleWriteError is called, which sets packet_. So when this task gets
scheduled, it will find packet_ to be non null, since it's the same member
variable that HandleWriteError sets. (This task also sets write_pending_ and
migration_pending_ to false above so that the migration task posted by
HandleWriteError aborts itself when it runs.) 

What am I missing?

+    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 write_pending_;
+}
+
 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()));
+  // 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