Use continual gathering to restore backup connections

webrtc/p2p/base/p2ptransportchannel.cc

 /*
  *  Copyright 2004 The WebRTC Project Authors. All rights reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
 #include "webrtc/p2p/base/p2ptransportchannel.h"
 
 #include <algorithm>
 #include <set>
 
 #include "webrtc/base/common.h"
 #include "webrtc/base/crc32.h"
 #include "webrtc/base/logging.h"
 #include "webrtc/base/stringencode.h"
 #include "webrtc/p2p/base/candidate.h"
 #include "webrtc/p2p/base/candidatepairinterface.h"
 #include "webrtc/p2p/base/common.h"
 #include "webrtc/p2p/base/relayport.h"  // For RELAY_PORT_TYPE.
 #include "webrtc/p2p/base/stunport.h"   // For STUN_PORT_TYPE.
 #include "webrtc/system_wrappers/include/field_trial.h"
 
 namespace {
 
 // messages for queuing up work for ourselves
-enum { MSG_SORT = 1, MSG_CHECK_AND_PING };
+enum {
+  MSG_SORT = 1,
+  MSG_CHECK_AND_PING,
+  MSG_CHECK_RESTORE_BACKUP_CONNECTION,
+  MSG_SIGNAL_CANDIDATES_REMOVED
+};
 
 // The minimum improvement in RTT that justifies a switch.
-static const double kMinImprovement = 10;
+const double kMinImprovement = 10;
 
 bool IsRelayRelay(cricket::Connection* conn) {
   return conn->local_candidate().type() == cricket::RELAY_PORT_TYPE &&
          conn->remote_candidate().type() == cricket::RELAY_PORT_TYPE;
 }
 
 bool IsUdp(cricket::Connection* conn) {
   return conn->local_candidate().relay_protocol() == cricket::UDP_PROTOCOL_NAME;
 }
 
@@ skipping 173 matching lines @@
 // writable or not receiving.
 const int WEAK_PING_INTERVAL = 1000 * PING_PACKET_SIZE / 10000;
 
 // If the current best connection is both writable and receiving, then we will
 // also try hard to make sure it is pinged at this rate (a little less than
 // 2 * STRONG_PING_INTERVAL).
 static const int MAX_CURRENT_STRONG_INTERVAL = 900;  // ms
 
 static const int MIN_CHECK_RECEIVING_INTERVAL = 50;  // ms
 
+// We periodically check and restore backup connections on existing networks
+// that do not have any connection at this interval.
+static const int CHECK_RESTORE_BACKUP_CONNECTION_INTERVAL = 5 * 60 * 1000;
+
 P2PTransportChannel::P2PTransportChannel(const std::string& transport_name,
                                          int component,
                                          P2PTransport* transport,
                                          PortAllocator* allocator)
     : P2PTransportChannel(transport_name, component, allocator) {}
 
 P2PTransportChannel::P2PTransportChannel(const std::string& transport_name,
                                          int component,
                                          PortAllocator* allocator)
     : TransportChannelImpl(transport_name, component),
       allocator_(allocator),
       worker_thread_(rtc::Thread::Current()),
       incoming_only_(false),
       error_(0),
       best_connection_(NULL),
       pending_best_connection_(NULL),
       sort_dirty_(false),
       remote_ice_mode_(ICEMODE_FULL),
       ice_role_(ICEROLE_UNKNOWN),
       tiebreaker_(0),
       gathering_state_(kIceGatheringNew),
       check_receiving_interval_(MIN_CHECK_RECEIVING_INTERVAL * 5),
+      check_restore_backup_connection_interval_(
+          CHECK_RESTORE_BACKUP_CONNECTION_INTERVAL),
       config_(MIN_CHECK_RECEIVING_INTERVAL * 50 /* receiving_timeout */,
               0 /* backup_connection_ping_interval */,
               false /* gather_continually */,
               false /* prioritize_most_likely_candidate_pairs */,
               MAX_CURRENT_STRONG_INTERVAL /* max_strong_interval */) {
   uint32_t weak_ping_interval = ::strtoul(
       webrtc::field_trial::FindFullName("WebRTC-StunInterPacketDelay").c_str(),
       nullptr, 10);
   if (weak_ping_interval) {
     weak_ping_interval_ = static_cast<int>(weak_ping_interval);
@@ skipping 14 matching lines @@
   session->SignalPortReady.connect(this, &P2PTransportChannel::OnPortReady);
   session->SignalCandidatesReady.connect(
       this, &P2PTransportChannel::OnCandidatesReady);
   session->SignalCandidatesAllocationDone.connect(
       this, &P2PTransportChannel::OnCandidatesAllocationDone);
 
   // We now only want to apply new candidates that we receive to the ports
   // created by this new session because these are replacing those of the
   // previous sessions.
   ports_.clear();
+  // We do not need to signal the candidate removals because a new allocator
+  // session indicates an ICE restart, which will remove all old remote
+  // candidates on the other side.
+  candidates_removed_.clear();
 
   allocator_sessions_.push_back(std::move(session));
 }
 
 void P2PTransportChannel::AddConnection(Connection* connection) {
   connections_.push_back(connection);
   unpinged_connections_.insert(connection);
   connection->set_remote_ice_mode(remote_ice_mode_);
   connection->set_receiving_timeout(config_.receiving_timeout);
   connection->SignalReadPacket.connect(
@@ skipping 40 matching lines @@
   if (!had_connection_) {
     return TransportChannelState::STATE_INIT;
   }
 
   std::vector<Connection*> active_connections;
   for (Connection* connection : connections_) {
     if (connection->active()) {
       active_connections.push_back(connection);
     }
   }
-  if (active_connections.empty()) {
+  // TODO(honghaiz): We should expose one transport channel state to the
+  // the external world (e.g. TransportController), so that we won't need to
+  // make both writable and channel state depend on IsRecoveringConnectivity.
+  if (active_connections.empty() && !IsRecoveringConnectivity()) {
     return TransportChannelState::STATE_FAILED;
   }
 
   std::set<rtc::Network*> networks;
   for (Connection* connection : active_connections) {
     rtc::Network* network = connection->port()->Network();
     if (networks.find(network) == networks.end()) {
       networks.insert(network);
     } else {
       LOG_J(LS_VERBOSE, this) << "Ice not completed yet for this channel as "
@@ skipping 95 matching lines @@
   ASSERT(worker_thread_ == rtc::Thread::Current());
   if (ice_ufrag_.empty() || ice_pwd_.empty()) {
     ASSERT(false);
     LOG(LS_ERROR) << "P2PTransportChannel::Connect: The ice_ufrag_ and the "
                   << "ice_pwd_ are not set.";
     return;
   }
 
   // Start checking and pinging as the ports come in.
   thread()->Post(this, MSG_CHECK_AND_PING);
+  thread()->PostDelayed(check_restore_backup_connection_interval_, this,
+                        MSG_CHECK_RESTORE_BACKUP_CONNECTION);
 }
 
 void P2PTransportChannel::MaybeStartGathering() {
   // Start gathering if we never started before, or if an ICE restart occurred.
   if (allocator_sessions_.empty() ||
       IceCredentialsChanged(allocator_sessions_.back()->ice_ufrag(),
                             allocator_sessions_.back()->ice_pwd(), ice_ufrag_,
                             ice_pwd_)) {
     if (gathering_state_ != kIceGatheringGathering) {
       gathering_state_ = kIceGatheringGathering;
@@ skipping 761 matching lines @@
         RTC_DCHECK(state == STATE_CONNECTING || state == STATE_COMPLETED);
         break;
       default:
         RTC_DCHECK(false);
         break;
     }
     state_ = state;
     SignalStateChanged(this);
   }
 
-  bool writable = best_connection_ && best_connection_->writable();
+  // If it is recovering connectivity, it is considered writable to the
+  // external world.

[Taylor Brandstetter, 2016/06/06 18:18:37]

When you say "external world", do you mean applications (so they see ICE
"connected" instead of "disconnected"), or do you mean the audio/video engines?
I think in this situation, an application *should* see "disconnected". If
they've turned continual gathering on, they should expect this, and just wait
for it to resolve itself, and maybe display a message to the user in the
meantime.

[honghaiz3, 2016/06/07 16:42:06]

I meant the TransportController and the layers above. 
If the application see the "disconnected", it will try to kick off "full ICE
restart", which we don't want it to happen when a continual gathering may
resolve the issue. I added a TODO to expose the recovering state the app,
although the app won't do much about it except that "it maybe display a message
to the user". 

[Taylor Brandstetter, 2016/06/07 17:13:26]

If the application is using continual gathering, it should know that
"disconnected" can resolve itself. It should expect a sequence of these states:

connected/not gathering ->
disconnected/gathering ->
checking/not gathering ->
connected/not gathering

If at the end, the state is "disconnected/not gathering", *then* the application
may want to do a full ICE restart.

However, we don't even switch to "checking" properly right now. So I'm ok with
the code as it is now. Just be aware that we may need to change it later.

+  bool writable = (best_connection_ && best_connection_->writable()) ||
+                  IsRecoveringConnectivity();
   set_writable(writable);
 
   bool receiving = false;
   for (const Connection* connection : connections_) {
     if (connection->receiving()) {
       receiving = true;
       break;
     }
   }
   set_receiving(receiving);
@@ skipping 48 matching lines @@
 
 // Handle any queued up requests
 void P2PTransportChannel::OnMessage(rtc::Message *pmsg) {
   switch (pmsg->message_id) {
     case MSG_SORT:
       OnSort();
       break;
     case MSG_CHECK_AND_PING:
       OnCheckAndPing();
       break;
+    case MSG_CHECK_RESTORE_BACKUP_CONNECTION:
+      OnCheckAndRestoreBackupConnection();
+      break;
+    case MSG_SIGNAL_CANDIDATES_REMOVED:
+      if (!candidates_removed_.empty()) {
+        LOG(LS_INFO) << "Signal removing " << candidates_removed_.size()
+                     << " candidates";
+        SignalCandidatesRemoved(this, candidates_removed_);
+        candidates_removed_.clear();
+      }
+      break;
     default:
       ASSERT(false);
       break;
   }
 }
 
 // Handle queued up sort request
 void P2PTransportChannel::OnSort() {
   // Resort the connections based on the new statistics.
   SortConnections();
@@ skipping 18 matching lines @@
     Connection* conn = FindNextPingableConnection();
     if (conn) {
       PingConnection(conn);
       MarkConnectionPinged(conn);
     }
   }
   int delay = std::min(ping_interval, check_receiving_interval_);
   thread()->PostDelayed(delay, this, MSG_CHECK_AND_PING);
 }
 
+void P2PTransportChannel::OnCheckAndRestoreBackupConnection() {
+  if (!config_.gather_continually) {
+    return;
+  }
+  rtc::NetworkManager::NetworkList networks;
+  allocator_->GetNetworks(&networks);
+  // Filter out loopback networks.
+  auto network_end = std::remove_if(
+      networks.begin(), networks.end(), [](rtc::Network* network) {
+        return network->type() == rtc::ADAPTER_TYPE_LOOPBACK;
+      });
+
+  // Filter out networks having connections in non-failed state.
+  for (Connection* conn : connections_) {
+    if (conn->state() == Connection::STATE_FAILED) {
+      continue;
+    }
+    const std::string& interface_name = conn->port()->Network()->name();
+    network_end = std::remove_if(networks.begin(), network_end,
+                                 [interface_name](rtc::Network* net) {
+                                   return net->name() == interface_name;
+                                 });
+    if (networks.begin() == network_end) {
+      break;
+    }
+  }
+  networks.erase(network_end, networks.end());
+  if (!networks.empty()) {
+    StartRegathering(&networks);
+  }
+
+  thread()->PostDelayed(check_restore_backup_connection_interval_, this,
+                        MSG_CHECK_RESTORE_BACKUP_CONNECTION);
+}
+
 // A connection is considered a backup connection if the channel state
 // is completed, the connection is not the best connection and it is active.
 bool P2PTransportChannel::IsBackupConnection(Connection* conn) const {
   return state_ == STATE_COMPLETED && conn != best_connection_ &&
          conn->active();
 }
 
 // Is the connection in a state for us to even consider pinging the other side?
 // We consider a connection pingable even if it's not connected because that's
 // how a TCP connection is kicked into reconnecting on the active side.
@@ skipping 47 matching lines @@
   }
   return conn_to_ping;
 }
 
 void P2PTransportChannel::MarkConnectionPinged(Connection* conn) {
   if (conn && pinged_connections_.insert(conn).second) {
     unpinged_connections_.erase(conn);
   }
 }
 
+void P2PTransportChannel::set_check_restore_backup_connection_interval(
+    int interval) {
+  check_restore_backup_connection_interval_ = interval;
+  thread()->Clear(this, MSG_CHECK_RESTORE_BACKUP_CONNECTION, nullptr);
+  thread()->PostDelayed(check_restore_backup_connection_interval_, this,
+                        MSG_CHECK_RESTORE_BACKUP_CONNECTION);
+}
+
 // Apart from sending ping from |conn| this method also updates
 // |use_candidate_attr| flag. The criteria to update this flag is
 // explained below.
 // Set USE-CANDIDATE if doing ICE AND this channel is in CONTROLLING AND
 //    a) Channel is in FULL ICE AND
 //      a.1) |conn| is the best connection OR
 //      a.2) there is no best connection OR
 //      a.3) the best connection is unwritable OR
 //      a.4) |conn| has higher priority than best_connection.
 //    b) we're doing LITE ICE AND
@@ skipping 31 matching lines @@
 
   // May stop the allocator session when at least one connection becomes
   // strongly connected after starting to get ports and the local candidate of
   // the connection is at the latest generation. It is not enough to check
   // that the connection becomes weakly connected because the connection may be
   // changing from (writable, receiving) to (writable, not receiving).
   bool strongly_connected = !connection->weak();
   bool latest_generation = connection->local_candidate().generation() >=
                            allocator_session()->generation();
   if (strongly_connected && latest_generation) {
+    if (recovering_start_time_) {
+      recovering_start_time_ = 0;
+    }
     MaybeStopPortAllocatorSessions();
   }
 
   // We have to unroll the stack before doing this because we may be changing
   // the state of connections while sorting.
   RequestSort();
 }
 
 // When a connection is removed, edit it out, and then update our best
 // connection.
@@ skipping 18 matching lines @@
     pending_best_connection_ = NULL;
   }
 
   // If this is currently the best connection, then we need to pick a new one.
   // The call to SortConnections will pick a new one.  It looks at the current
   // best connection in order to avoid switching between fairly similar ones.
   // Since this connection is no longer an option, we can just set best to NULL
   // and re-choose a best assuming that there was no best connection.
   if (best_connection_ == connection) {
     LOG(LS_INFO) << "Best connection destroyed.  Will choose a new one.";
+    // When continual gathering is enabled, we will attempt re-gathering only
+    // if the best connecion was removed and it was writable. If it was not

[Taylor Brandstetter, 2016/06/06 18:18:37]

"connecion" => "connection"

[honghaiz3, 2016/06/07 16:42:06]

Done.

+    // writable, most likely it is in the process of gathering on the current
+    // session.
+    if (config_.gather_continually && writable()) {
+      bool all_connections_failed = std::all_of(
+          connections_.begin(), connections_.end(), [](Connection* conn) {
+            return conn->state() == Connection::STATE_FAILED;
+          });
+      // Start regathering on all active networks if all connections failed.
+      if (all_connections_failed && !IsRecoveringConnectivity()) {
+        recovering_start_time_ = rtc::TimeMillis();
+        StartRegathering(nullptr);
+      }
+    }
     SwitchBestConnectionTo(NULL);
     RequestSort();
   }
 
   UpdateState();
 }
 
 // When a port is destroyed remove it from our list of ports to use for
 // connection attempts.
 void P2PTransportChannel::OnPortDestroyed(PortInterface* port) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   // Remove this port from the list (if we didn't drop it already).
   std::vector<PortInterface*>::iterator iter =
       std::find(ports_.begin(), ports_.end(), port);
   if (iter != ports_.end())
     ports_.erase(iter);
 
   LOG(INFO) << "Removed port from p2p socket: "
             << static_cast<int>(ports_.size()) << " remaining";
 }
 
 void P2PTransportChannel::OnPortNetworkInactive(PortInterface* port) {
   // If it does not gather continually, the port will be removed from the list
-  // when ICE restarts.
+  // when ICE restarts, so don't need to remove it here.
   if (!config_.gather_continually) {
     return;
   }
   auto it = std::find(ports_.begin(), ports_.end(), port);
   // Don't need to do anything if the port has been deleted from the port list.
   if (it == ports_.end()) {
     return;
   }
   ports_.erase(it);
   LOG(INFO) << "Removed port due to inactive networks: " << ports_.size()
             << " remaining";
-  std::vector<Candidate> candidates = port->Candidates();
-  for (Candidate& candidate : candidates) {
+  port->FailAndDestroyConnections();
+
+  for (Candidate candidate : port->Candidates()) {
     candidate.set_transport_name(transport_name());
+    candidates_removed_.push_back(candidate);

[Taylor Brandstetter, 2016/06/06 18:18:37]

Something I didn't notice until now: This needs to sanitize candidates just like
BasicPortAllocator does. By which I mean: don't signal candidates that don't
match the candidate filter, and zero out the "related address" if necessary.
See:
https://cs.chromium.org/chromium/src/third_party/webrtc/p2p/client/basicporta...

If we don't do this, then we'll end up leaking IPs in the "candidate removed"
event.

[honghaiz3, 2016/06/07 16:42:06]

Done.

   }
-  SignalCandidatesRemoved(this, candidates);
+  thread()->Post(this, MSG_SIGNAL_CANDIDATES_REMOVED);
+}
+
+void P2PTransportChannel::StartRegathering(
+    std::vector<rtc::Network*>* networks) {
+  std::string log_msg;
+  std::vector<PortInterface*> ports_to_remove;
+  if (networks == nullptr) {
+    log_msg = "Regathering on all networks";
+    ports_to_remove = ports_;
+  } else {
+    log_msg = "Regathering on networks:";
+    for (rtc::Network* network : *networks) {
+      log_msg += " " + network->ToString();
+    }
+    // Remove all ports using |networks|.
+    for (PortInterface* port : ports_) {
+      if (std::find(networks->begin(), networks->end(), port->Network()) !=
+          networks->end()) {
+        ports_to_remove.push_back(port);
+      }
+    }
+  }
+  LOG(LS_INFO) << log_msg;
+  for (PortInterface* port : ports_to_remove) {
+    // Remove the port and signal the other side to remove the respective remote
+    // candidates.
+    OnPortNetworkInactive(port);
+  }
+
+  PortAllocatorSession* session = allocator_session();
+  // By deactivating all networks in existing allocation sequences,
+  // it will force re-gathering on all active networks.
+  session->InactivateNetworksInExistingSequences(networks);
+  session->StartGettingPorts();
 }
 
 // We data is available, let listeners know
 void P2PTransportChannel::OnReadPacket(Connection* connection,
                                        const char* data,
                                        size_t len,
                                        const rtc::PacketTime& packet_time) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   // Do not deliver, if packet doesn't belong to the correct transport channel.
@@ skipping 134 matching lines @@
 
   // During the initial state when nothing has been pinged yet, return the first
   // one in the ordered |connections_|.
   return *(std::find_if(connections_.begin(), connections_.end(),
                         [conn1, conn2](Connection* conn) {
                           return conn == conn1 || conn == conn2;
                         }));
 }
 
 }  // namespace cricket