Add QuicConnectivityProbingManager which sends connectivity probing

packet on new path. The manager will send out a new connectivity probing
packet if there is no response from the server on the probed path. The
manager will retry after a silent period from the previous sent event with
exponential backoff.
1 unit timeout is the smaller of 2 smoothed rtt from old path, or 600ms.


Bug: 774622
Cq-Include-Trybots: master.tryserver.chromium.android:android_cronet_tester;master.tryserver.chromium.mac:ios-simulator-cronet
Change-Id: I623b8a5aa8da6d553ca16d9f67655d4a32a4a1c9
Reviewed-on: https://chromium-review.googlesource.com/774299
Commit-Queue: Zhongyi Shi <zhongyi@chromium.org>
Reviewed-by: Ryan Hamilton <rch@chromium.org>
Cr-Commit-Position: refs/heads/master@{#517737}

net/BUILD.gn

@@ -1123,6 +1123,8 @@ component("net") {
       "quic/chromium/quic_clock_skew_detector.h",
       "quic/chromium/quic_connection_logger.cc",
       "quic/chromium/quic_connection_logger.h",
+      "quic/chromium/quic_connectivity_probing_manager.cc",
+      "quic/chromium/quic_connectivity_probing_manager.h",
       "quic/chromium/quic_crypto_client_stream_factory.cc",
       "quic/chromium/quic_crypto_client_stream_factory.h",
       "quic/chromium/quic_http_stream.cc",
@@ -4961,6 +4963,7 @@ test("net_unittests") {
     "quic/chromium/quic_chromium_client_stream_test.cc",
     "quic/chromium/quic_chromium_connection_helper_test.cc",
     "quic/chromium/quic_clock_skew_detector_test.cc",
+    "quic/chromium/quic_connectivity_probing_manager_test.cc",
     "quic/chromium/quic_end_to_end_unittest.cc",
     "quic/chromium/quic_http_stream_test.cc",
     "quic/chromium/quic_http_utils_test.cc",

net/quic/chromium/quic_chromium_client_session.cc

@@ -26,6 +26,7 @@
 #include "net/quic/chromium/crypto/proof_verifier_chromium.h"
 #include "net/quic/chromium/quic_chromium_connection_helper.h"
 #include "net/quic/chromium/quic_chromium_packet_writer.h"
+#include "net/quic/chromium/quic_connectivity_probing_manager.h"
 #include "net/quic/chromium/quic_crypto_client_stream_factory.h"
 #include "net/quic/chromium/quic_server_info.h"
 #include "net/quic/chromium/quic_stream_factory.h"
@@ -64,6 +65,10 @@ const size_t kTokenBindingSignatureMapSize = 10;
 // migrating sessions need to wait for a new network to connect.
 const size_t kWaitTimeForNewNetworkSecs = 10;
 
+// Maximum RTT time for this session when set initial timeout for probing
+// network.
+const int kDefaultRTTMilliSecs = 300;
+
 // The maximum size of uncompressed QUIC headers that will be allowed.
 const size_t kMaxUncompressedHeaderSize = 256 * 1024;
 
@@ -658,7 +663,7 @@ QuicChromiumClientSession::QuicChromiumClientSession(
     base::TimeTicks dns_resolution_end_time,
     QuicClientPushPromiseIndex* push_promise_index,
     ServerPushDelegate* push_delegate,
-    base::TaskRunner* task_runner,
+    base::SequencedTaskRunner* task_runner,
     std::unique_ptr<SocketPerformanceWatcher> socket_performance_watcher,
     NetLog* net_log)
     : QuicSpdyClientSessionBase(connection, push_promise_index, config),
@@ -692,6 +697,7 @@ QuicChromiumClientSession::QuicChromiumClientSession(
       streams_pushed_and_claimed_count_(0),
       bytes_pushed_count_(0),
       bytes_pushed_and_unclaimed_count_(0),
+      probing_manager_(this, task_runner_),
       migration_pending_(false),
       weak_factory_(this) {
   sockets_.push_back(std::move(socket));
@@ -1449,6 +1455,15 @@ void QuicChromiumClientSession::OnSuccessfulVersionNegotiation(
   QuicSpdySession::OnSuccessfulVersionNegotiation(version);
 }
 
+void QuicChromiumClientSession::OnConnectivityProbeReceived(
+    const QuicSocketAddress& self_address,
+    const QuicSocketAddress& peer_address) {
+  DVLOG(1) << "Probing response from ip:port: " << peer_address.ToString()
+           << " to ip:port: " << self_address.ToString() << " is received";
+  // Notify the probing manager that a connectivity probing packet is received.
+  probing_manager_.OnConnectivityProbingReceived(self_address, peer_address);
+}
+
 int QuicChromiumClientSession::HandleWriteError(
     int error_code,
     scoped_refptr<QuicChromiumPacketWriter::ReusableIOBuffer> packet) {
@@ -1570,6 +1585,43 @@ void QuicChromiumClientSession::OnMigrationTimeout(size_t num_sockets) {
                       QUIC_CONNECTION_MIGRATION_NO_NEW_NETWORK);
 }
 
+void QuicChromiumClientSession::OnProbeNetworkSucceeded(
+    NetworkChangeNotifier::NetworkHandle network,
+    const QuicSocketAddress& self_address,
+    std::unique_ptr<DatagramClientSocket> socket,
+    std::unique_ptr<QuicChromiumPacketWriter> writer,
+    std::unique_ptr<QuicChromiumPacketReader> reader) {
+  DCHECK(socket);
+  DCHECK(writer);
+  DCHECK(reader);
+
+  // Set |this| to listen on socket write events on the packet writer
+  // that was used for probing.
+  writer->set_delegate(this);
+
+  // TODO(zhongyi): check |network| to see if we want to migrate to
+  // the network immediately.
+
+  // Mark probing socket as the default socket.
+  sockets_.push_back(std::move(socket));
+  packet_readers_.push_back(std::move(reader));
+
+  connection()->SetSelfAddress(self_address);
+  connection()->SetQuicPacketWriter(writer.release(),
+                                    /*owns_writer*/ true);
+}
+
+void QuicChromiumClientSession::OnProbeNetworkFailed(
+    NetworkChangeNotifier::NetworkHandle network) {
+  DVLOG(1) << "Connectivity probing fails on NetworkHandle " << network;
+}
+
+void QuicChromiumClientSession::OnSendConnectivityProbingPacket(
+    QuicChromiumPacketWriter* writer,
+    const QuicSocketAddress& peer_address) {
+  connection()->SendConnectivityProbingPacket(writer, peer_address);
+}
+
 void QuicChromiumClientSession::OnNetworkConnected(
     NetworkChangeNotifier::NetworkHandle network,
     const NetLogWithSource& net_log) {
@@ -1797,6 +1849,84 @@ void QuicChromiumClientSession::NotifyRequestsOfConfirmation(int net_error) {
   waiting_for_confirmation_callbacks_.clear();
 }
 
+ProbingResult QuicChromiumClientSession::StartProbeNetwork(
+    NetworkChangeNotifier::NetworkHandle network,
+    IPEndPoint peer_address,
+    const NetLogWithSource& migration_net_log) {
+  if (!stream_factory_)
+    return ProbingResult::FAILURE;
+
+  CHECK_NE(NetworkChangeNotifier::kInvalidNetworkHandle, network);
+
+  // Close idle session.
+  if (GetNumActiveStreams() == 0) {
+    DVLOG(1) << "Client closes session before sends connectivity probe "
+             << "due to being idle";
+    HistogramAndLogMigrationFailure(migration_net_log,
+                                    MIGRATION_STATUS_NO_MIGRATABLE_STREAMS,
+                                    connection_id(), "Session being idle");
+    CloseSessionOnError(ERR_NETWORK_CHANGED,
+                        QUIC_CONNECTION_MIGRATION_NO_MIGRATABLE_STREAMS);
+    return ProbingResult::DISABLED_WITH_IDLE_SESSION;
+  }
+
+  // Abort probing if connection migration is disabled by config.
+  if (config()->DisableConnectionMigration()) {
+    DVLOG(1) << "Client disables probing network with connection migration "
+             << "disabled by config";
+    HistogramAndLogMigrationFailure(migration_net_log,
+                                    MIGRATION_STATUS_DISABLED, connection_id(),
+                                    "Migration disabled by config");
+    // TODO(zhongyi): do we want to close the session?
+    return ProbingResult::DISABLED_BY_CONFIG;
+  }
+
+  // TODO(zhongyi): migrate the session still, but reset non-migratable stream.
+  // TODO(zhongyi): migrate non-migrtable stream if moving from Cellular to
+  // wifi.
+  // Abort probing if there is stream marked as non-migratable.
+  if (HasNonMigratableStreams()) {
+    DVLOG(1) << "Clients disables probing network with non-migratable streams";
+    HistogramAndLogMigrationFailure(migration_net_log,
+                                    MIGRATION_STATUS_NON_MIGRATABLE_STREAM,
+                                    connection_id(), "Non-migratable stream");
+    return ProbingResult::DISABLED_BY_NON_MIGRABLE_STREAM;
+  }
+
+  // Create and configure socket on |network|.
+  std::unique_ptr<DatagramClientSocket> probing_socket =
+      stream_factory_->CreateSocket(net_log_.net_log(), net_log_.source());
+  if (stream_factory_->ConfigureSocket(probing_socket.get(), peer_address,
+                                       network) != OK) {
+    HistogramAndLogMigrationFailure(
+        migration_net_log, MIGRATION_STATUS_INTERNAL_ERROR, connection_id(),
+        "Socket configuration failed");
+    return ProbingResult::INTERNAL_ERROR;
+  }
+
+  // Create new packet writer and reader on the probing socket.
+  std::unique_ptr<QuicChromiumPacketWriter> probing_writer(
+      new QuicChromiumPacketWriter(probing_socket.get()));
+  std::unique_ptr<QuicChromiumPacketReader> probing_reader(
+      new QuicChromiumPacketReader(probing_socket.get(), clock_, this,
+                                   yield_after_packets_, yield_after_duration_,
+                                   net_log_));
+
+  int timeout_ms = connection()
+                       ->sent_packet_manager()
+                       .GetRttStats()
+                       ->smoothed_rtt()
+                       .ToMilliseconds();
+  if (timeout_ms == 0 || timeout_ms > kDefaultRTTMilliSecs)
+    timeout_ms = kDefaultRTTMilliSecs;
+
+  probing_manager_.StartProbing(
+      network, QuicSocketAddress(QuicSocketAddressImpl(peer_address)),
+      std::move(probing_socket), std::move(probing_writer),
+      std::move(probing_reader), base::TimeDelta::FromMilliseconds(timeout_ms));
+  return ProbingResult::PENDING;
+}
+
 std::unique_ptr<base::Value> QuicChromiumClientSession::GetInfoAsValue(
     const std::set<HostPortPair>& aliases) {
   std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());

net/quic/chromium/quic_chromium_client_session.h

@@ -32,6 +32,7 @@
 #include "net/quic/chromium/quic_chromium_packet_reader.h"
 #include "net/quic/chromium/quic_chromium_packet_writer.h"
 #include "net/quic/chromium/quic_connection_logger.h"
+#include "net/quic/chromium/quic_connectivity_probing_manager.h"
 #include "net/quic/core/quic_client_push_promise_index.h"
 #include "net/quic/core/quic_crypto_client_stream.h"
 #include "net/quic/core/quic_packets.h"
@@ -68,9 +69,20 @@ enum class MigrationResult {
   FAILURE          // Migration failed for other reasons.
 };
 
+// Result of a connectivity probing attempt.
+enum class ProbingResult {
+  PENDING,                          // Probing started, pending result.
+  DISABLED_WITH_IDLE_SESSION,       // Probing disabled with idle session.
+  DISABLED_BY_CONFIG,               // Probing disabled by config.
+  DISABLED_BY_NON_MIGRABLE_STREAM,  // Probing disabled by special stream.
+  INTERNAL_ERROR,                   // Probing failed for internal reason.
+  FAILURE,                          // Probing failed for other reason.
+};
+
 class NET_EXPORT_PRIVATE QuicChromiumClientSession
     : public QuicSpdyClientSessionBase,
       public MultiplexedSession,
+      public QuicConnectivityProbingManager::Delegate,
       public QuicChromiumPacketReader::Visitor,
       public QuicChromiumPacketWriter::Delegate {
  public:
@@ -314,7 +326,7 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
       base::TimeTicks dns_resolution_end_time,
       QuicClientPushPromiseIndex* push_promise_index,
       ServerPushDelegate* push_delegate,
-      base::TaskRunner* task_runner,
+      base::SequencedTaskRunner* task_runner,
       std::unique_ptr<SocketPerformanceWatcher> socket_performance_watcher,
       NetLog* net_log);
   ~QuicChromiumClientSession() override;
@@ -350,6 +362,21 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
   void OnWriteError(int error_code) override;
   void OnWriteUnblocked() override;
 
+  // QuicConnectivityProbingManager::Delegate override.
+  void OnProbeNetworkSucceeded(
+      NetworkChangeNotifier::NetworkHandle network,
+      const QuicSocketAddress& self_address,
+      std::unique_ptr<DatagramClientSocket> socket,
+      std::unique_ptr<QuicChromiumPacketWriter> writer,
+      std::unique_ptr<QuicChromiumPacketReader> reader) override;
+
+  void OnProbeNetworkFailed(
+      NetworkChangeNotifier::NetworkHandle network) override;
+
+  void OnSendConnectivityProbingPacket(
+      QuicChromiumPacketWriter* writer,
+      const QuicSocketAddress& peer_address) override;
+
   // QuicSpdySession methods:
   void OnHeadersHeadOfLineBlocking(QuicTime::Delta delta) override;
 
@@ -382,6 +409,9 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
                           ConnectionCloseSource source) override;
   void OnSuccessfulVersionNegotiation(
       const QuicTransportVersion& version) override;
+  void OnConnectivityProbeReceived(
+      const QuicSocketAddress& self_address,
+      const QuicSocketAddress& peer_address) override;
   void OnPathDegrading() override;
   bool HasOpenDynamicStreams() const override;
 
@@ -569,6 +599,10 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
   void CancelAllRequests(int net_error);
   void NotifyRequestsOfConfirmation(int net_error);
 
+  ProbingResult StartProbeNetwork(NetworkChangeNotifier::NetworkHandle network,
+                                  IPEndPoint peer_address,
+                                  const NetLogWithSource& migration_net_log);
+
   // Notifies the factory that this session is going away and no more streams
   // should be created from it.  This needs to be called before closing any
   // streams, because closing a stream may cause a new stream to be created.
@@ -609,7 +643,7 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
   std::vector<CompletionCallback> waiting_for_confirmation_callbacks_;
   CompletionCallback callback_;
   size_t num_total_streams_;
-  base::TaskRunner* task_runner_;
+  base::SequencedTaskRunner* task_runner_;
   NetLogWithSource net_log_;
   std::vector<std::unique_ptr<QuicChromiumPacketReader>> packet_readers_;
   LoadTimingInfo::ConnectTiming connect_timing_;
@@ -631,6 +665,7 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
   // Stores packet that witnesses socket write error. This packet is
   // written to a new socket after migration completes.
   scoped_refptr<QuicChromiumPacketWriter::ReusableIOBuffer> packet_;
+  QuicConnectivityProbingManager probing_manager_;
   // TODO(jri): Replace use of migration_pending_ sockets_.size().
   // When a task is posted for MigrateSessionOnError, pass in
   // sockets_.size(). Then in MigrateSessionOnError, check to see if

net/quic/chromium/quic_connectivity_probing_manager.cc

+// Copyright (c) 2017 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_connectivity_probing_manager.h"
+
+namespace net {
+
+namespace {
+
+// Default to 2 seconds timeout as the maximum timeout.
+const int64_t kMaxProbingTimeoutMs = 2000;
+
+}  // namespace
+
+QuicConnectivityProbingManager::QuicConnectivityProbingManager(
+    Delegate* delegate,
+    base::SequencedTaskRunner* task_runner)
+    : delegate_(delegate),
+      retry_count_(0),
+      task_runner_(task_runner),
+      weak_factory_(this) {
+  retransmit_timer_.SetTaskRunner(task_runner_);
+}
+
+QuicConnectivityProbingManager::~QuicConnectivityProbingManager() {
+  CancelProbing();
+}
+
+int QuicConnectivityProbingManager::HandleWriteError(
+    int error_code,
+    scoped_refptr<QuicChromiumPacketWriter::ReusableIOBuffer> packet) {
+  // Write error on the probing network is not recoverable.
+  DVLOG(1) << "Probing packet encounters write error";
+  // Post a task to notify |delegate_| that this probe failed and cancel
+  // undergoing probing, which will delete the packet writer.
+  task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&QuicConnectivityProbingManager::NotifyDelegateProbeFailed,
+                 weak_factory_.GetWeakPtr()));
+  return error_code;
+}
+
+void QuicConnectivityProbingManager::OnWriteError(int error_code) {
+  // Write error on the probing network.
+  NotifyDelegateProbeFailed();
+}
+
+void QuicConnectivityProbingManager::OnWriteUnblocked() {}
+
+void QuicConnectivityProbingManager::CancelProbing() {
+  network_ = NetworkChangeNotifier::kInvalidNetworkHandle;
+  peer_address_ = QuicSocketAddress();
+  socket_.reset();
+  writer_.reset();
+  reader_.reset();
+  retry_count_ = 0;
+  initial_timeout_ = base::TimeDelta();
+  retransmit_timer_.Stop();
+}
+
+void QuicConnectivityProbingManager::StartProbing(
+    NetworkChangeNotifier::NetworkHandle network,
+    const QuicSocketAddress& peer_address,
+    std::unique_ptr<DatagramClientSocket> socket,
+    std::unique_ptr<QuicChromiumPacketWriter> writer,
+    std::unique_ptr<QuicChromiumPacketReader> reader,
+    base::TimeDelta initial_timeout) {
+  // Start a new probe will always cancel the previous one.
+  CancelProbing();
+
+  network_ = network;
+  peer_address_ = peer_address;
+  socket_ = std::move(socket);
+  writer_ = std::move(writer);
+  // |this| will listen to all socket write events for the probing
+  // packet writer.
+  writer_->set_delegate(this);
+  reader_ = std::move(reader);
+  initial_timeout_ = initial_timeout;
+
+  reader_->StartReading();
+  SendConnectivityProbingPacket(initial_timeout_);
+}
+
+void QuicConnectivityProbingManager::OnConnectivityProbingReceived(
+    const QuicSocketAddress& self_address,
+    const QuicSocketAddress& peer_address) {
+  if (!socket_) {
+    DVLOG(1) << "Probing response is ignored as probing was cancelled "
+             << "or succeeded.";
+    return;
+  }
+
+  IPEndPoint local_address;
+  socket_->GetLocalAddress(&local_address);
+
+  DVLOG(1) << "Current probing is live at self ip:port "
+           << local_address.ToString() << ", to peer ip:port "
+           << peer_address_.ToString();
+
+  if (QuicSocketAddressImpl(local_address) != self_address.impl() ||
+      peer_address_ != peer_address) {
+    DVLOG(1) << "Received probing response from peer ip:port "
+             << peer_address.ToString() << ", to self ip:port "
+             << self_address.ToString() << ". Ignored.";
+    return;
+  }
+
+  // TODO(zhongyi): add metrics collection.
+  // Notify the delegate that the probe succeeds and reset everything.
+  delegate_->OnProbeNetworkSucceeded(network_, self_address, std::move(socket_),
+                                     std::move(writer_), std::move(reader_));
+  CancelProbing();
+}
+
+void QuicConnectivityProbingManager::SendConnectivityProbingPacket(
+    base::TimeDelta timeout) {
+  delegate_->OnSendConnectivityProbingPacket(writer_.get(), peer_address_);
+  retransmit_timer_.Start(
+      FROM_HERE, timeout,
+      base::Bind(
+          &QuicConnectivityProbingManager::MaybeResendConnectivityProbingPacket,
+          weak_factory_.GetWeakPtr()));
+}
+
+void QuicConnectivityProbingManager::NotifyDelegateProbeFailed() {
+  if (network_ != NetworkChangeNotifier::kInvalidNetworkHandle) {
+    delegate_->OnProbeNetworkFailed(network_);
+    CancelProbing();
+  }
+}
+
+void QuicConnectivityProbingManager::MaybeResendConnectivityProbingPacket() {
+  // Use exponential backoff for the timeout
+  retry_count_++;
+  int64_t timeout_ms =
+      (UINT64_C(1) << retry_count_) * initial_timeout_.InMilliseconds();
+  if (timeout_ms > kMaxProbingTimeoutMs) {
+    NotifyDelegateProbeFailed();
+    return;
+  }
+  SendConnectivityProbingPacket(base::TimeDelta::FromMilliseconds(timeout_ms));
+}
+
+}  // namespace net

net/quic/chromium/quic_connectivity_probing_manager.h

+// Copyright (c) 2017 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.
+
+#ifndef NET_QUIC_CHROMIUM_QUIC_CONNECTIVITY_PROBING_MANAGER_H_
+#define NET_QUIC_CHROMIUM_QUIC_CONNECTIVITY_PROBING_MANAGER_H_
+
+#include "base/time/time.h"
+#include "net/base/net_export.h"
+#include "net/log/net_log_with_source.h"
+#include "net/quic/chromium/quic_chromium_packet_reader.h"
+#include "net/quic/chromium/quic_chromium_packet_writer.h"
+#include "net/quic/core/quic_time.h"
+
+namespace net {
+
+// Responsible for sending and retransmitting connectivity probing packet on a
+// designated path to the specified peer, and for notifying associated session
+// when connectivity probe fails or succeeds.
+class NET_EXPORT_PRIVATE QuicConnectivityProbingManager
+    : public QuicChromiumPacketWriter::Delegate {
+ public:
+  // Delegate interface which receives notifications on network probing
+  // results.
+  class NET_EXPORT_PRIVATE Delegate {
+   public:
+    virtual ~Delegate() {}
+    // Called when probing on |network| succeeded. Caller hands off ownership
+    // of |socket|, |writer| and |reader| for |network| to delegate.
+    virtual void OnProbeNetworkSucceeded(
+        NetworkChangeNotifier::NetworkHandle network,
+        const QuicSocketAddress& self_address,
+        std::unique_ptr<DatagramClientSocket> socket,
+        std::unique_ptr<QuicChromiumPacketWriter> writer,
+        std::unique_ptr<QuicChromiumPacketReader> reader) = 0;
+
+    // Called when probing on |network| fails.
+    virtual void OnProbeNetworkFailed(
+        NetworkChangeNotifier::NetworkHandle network) = 0;
+
+    // Called when a connectivity probing packet needs to be sent to
+    // |peer_address| using |writer|.
+    virtual void OnSendConnectivityProbingPacket(
+        QuicChromiumPacketWriter* writer,
+        const QuicSocketAddress& peer_address) = 0;
+  };
+
+  QuicConnectivityProbingManager(Delegate* delegate,
+                                 base::SequencedTaskRunner* task_runner);
+  ~QuicConnectivityProbingManager();
+
+  // QuicChromiumPacketWriter::Delegate interface.
+  int HandleWriteError(int error_code,
+                       scoped_refptr<QuicChromiumPacketWriter::ReusableIOBuffer>
+                           last_packet) override;
+
+  void OnWriteError(int error_code) override;
+
+  void OnWriteUnblocked() override;
+
+  // Starts probe |network| to |peer_address|. |this| will take ownership of
+  // |socket|, |writer| and |reader|. |writer| and |reader| should be bound
+  // to |socket| and |writer| will be used to send connectivity probing packets.
+  // Connectivity probing packets will be resent after initial timeout. Mutilple
+  // trials will be attempted with exponential backoff until a connectivity
+  // probing packet response is received from the peer by |reader| or final
+  // time out.
+  void StartProbing(NetworkChangeNotifier::NetworkHandle network,
+                    const QuicSocketAddress& peer_address,
+                    std::unique_ptr<DatagramClientSocket> socket,
+                    std::unique_ptr<QuicChromiumPacketWriter> writer,
+                    std::unique_ptr<QuicChromiumPacketReader> reader,
+                    base::TimeDelta initial_timeout);
+
+  // Cancels undergoing probing.
+  void CancelProbing();
+
+  // Called when a connectivity probing packet has been received from
+  // |peer_address| on a socket with |self_address|.
+  void OnConnectivityProbingReceived(const QuicSocketAddress& self_address,
+                                     const QuicSocketAddress& peer_address);
+
+ private:
+  // Called when a connectivity probing needs to be sent to |peer_address_| and
+  // set a timer to resend a connectivity probing packet to peer after
+  // |timeout|.
+  void SendConnectivityProbingPacket(base::TimeDelta timeout);
+
+  // Called when no connectivity probing packet response has been received on
+  // the currrent probing path after timeout.
+  void MaybeResendConnectivityProbingPacket();
+
+  void NotifyDelegateProbeFailed();
+
+  Delegate* delegate_;  // Unowned, must outlive |this|.
+
+  NetworkChangeNotifier::NetworkHandle network_;
+  QuicSocketAddress peer_address_;
+
+  std::unique_ptr<DatagramClientSocket> socket_;
+  std::unique_ptr<QuicChromiumPacketWriter> writer_;
+  std::unique_ptr<QuicChromiumPacketReader> reader_;
+
+  int64_t retry_count_;
+  base::TimeDelta initial_timeout_;
+  base::OneShotTimer retransmit_timer_;
+
+  base::SequencedTaskRunner* task_runner_;
+
+  base::WeakPtrFactory<QuicConnectivityProbingManager> weak_factory_;
+  DISALLOW_COPY_AND_ASSIGN(QuicConnectivityProbingManager);
+};
+
+}  // namespace net
+
+#endif  // NET_QUIC_CHROMIUM_QUIC_CONNECTIVITY_PROBING_MANAGER_H_

net/quic/chromium/quic_stream_factory.h

@@ -522,7 +522,7 @@ class NET_EXPORT_PRIVATE QuicStreamFactory
 
   QuicClientPushPromiseIndex push_promise_index_;
 
-  base::TaskRunner* task_runner_;
+  base::SequencedTaskRunner* task_runner_;
 
   const scoped_refptr<SSLConfigService> ssl_config_service_;
 

net/quic/chromium/quic_stream_factory_peer.cc

@@ -64,8 +64,9 @@ bool QuicStreamFactoryPeer::IsLiveSession(QuicStreamFactory* factory,
   return false;
 }
 
-void QuicStreamFactoryPeer::SetTaskRunner(QuicStreamFactory* factory,
-                                          base::TaskRunner* task_runner) {
+void QuicStreamFactoryPeer::SetTaskRunner(
+    QuicStreamFactory* factory,
+    base::SequencedTaskRunner* task_runner) {
   factory->task_runner_ = task_runner;
 }
 

net/quic/chromium/quic_stream_factory_peer.h

@@ -9,7 +9,7 @@
 #include <stdint.h>
 
 #include "base/macros.h"
-#include "base/task_runner.h"
+#include "base/sequenced_task_runner.h"
 #include "net/base/host_port_pair.h"
 #include "net/base/privacy_mode.h"
 #include "net/quic/core/quic_packets.h"
@@ -51,7 +51,7 @@ class QuicStreamFactoryPeer {
                             QuicChromiumClientSession* session);
 
   static void SetTaskRunner(QuicStreamFactory* factory,
-                            base::TaskRunner* task_runner);
+                            base::SequencedTaskRunner* task_runner);
 
   static QuicTime::Delta GetPingTimeout(QuicStreamFactory* factory);
 

net/quic/chromium/test_task_runner.cc

@@ -26,6 +26,12 @@ bool TestTaskRunner::PostDelayedTask(const base::Location& from_here,
   return false;
 }
 
+bool TestTaskRunner::PostNonNestableDelayedTask(const base::Location& from_here,
+                                                base::OnceClosure task,
+                                                base::TimeDelta delay) {
+  return PostDelayedTask(from_here, std::move(task), delay);
+}
+
 bool TestTaskRunner::RunsTasksInCurrentSequence() const {
   return true;
 }

net/quic/chromium/test_task_runner.h

@@ -22,7 +22,7 @@ namespace test {
 
 typedef base::TestPendingTask PostedTask;
 
-class TestTaskRunner : public base::TaskRunner {
+class TestTaskRunner : public base::SequencedTaskRunner {
  public:
   explicit TestTaskRunner(MockClock* clock);
 
@@ -30,6 +30,10 @@ class TestTaskRunner : public base::TaskRunner {
   bool PostDelayedTask(const base::Location& from_here,
                        base::OnceClosure task,
                        base::TimeDelta delay) override;
+  bool PostNonNestableDelayedTask(const base::Location& from_here,
+                                  base::OnceClosure task,
+                                  base::TimeDelta delay) override;
+
   bool RunsTasksInCurrentSequence() const override;
 
   const std::vector<PostedTask>& GetPostedTasks() const;