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,

[pthatcher1, 2016/06/07 18:54:33]

It seems like a better name would be MSG_REGATHER_FAILED_NETWORKS

[honghaiz3, 2016/06/29 02:48:22]

MSG_REGATHER_ON_FAILED_NETWORKS better?

+  MSG_SIGNAL_CANDIDATES_REMOVED

[pthatcher1, 2016/06/07 18:54:33]

Why does this require a thread post?  To coalesce candidate removals?

[honghaiz3, 2016/06/29 02:48:22]

Yes. We can merge multiple candidate removals into one. 

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

[pthatcher1, 2016/06/07 18:54:33]

I think a better name would be DEFAULT_REGATHER_FAILED_NETWORKS_INTERVAL.?

+
 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),

[pthatcher1, 2016/06/07 18:54:33]

Can you put this in config_ instead of as its own thing?

       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();

[pthatcher1, 2016/06/07 18:54:33]

This feels like implicit magic to me.  Could we instead know when we're removing
candidates because of an ICE restart and then suppress them from being signaled
in the first place instead of clearing them here?

 
   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()) {

[pthatcher1, 2016/06/07 18:54:33]

I don't like the look of the complexity added to hide the disconnected state
from the application.  I think it would be better instead to keep it simple and
just expose the disconnected state to applications that opt into continual
gathering.

     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.
+  // TODO(honghaiz): Expose the recovering state to the application.
+  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 || !allocator_->network_manager()) {
+    return;
+  }
+  rtc::NetworkManager::NetworkList networks;
+  allocator_->network_manager()->GetNetworks(&networks);

[pthatcher1, 2016/06/07 18:54:33]

I'm not a fan of reaching into the allocator like this.  Would it make sense,
instead, to have a GetFailedNetworks method on the Allocator?  The Allocator has
the Ports.  The Ports have the Connections.  The Connections have the states. 
So the Allocator should know which ports have non-failed connections.  I think
the code would end up being more simple and would have a more clear division of
labor between the P2pTransportChannel and the PortAllocator.

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

[pthatcher1, 2016/06/07 18:54:33]

This logic seems a bit complicated.   Would it make more sense to have one
std::remove_if with networks, checking if any non-failed connection has a
matching network name?

[honghaiz3, 2016/06/29 02:48:23]

Done.

+  if (!networks.empty()) {
+    StartRegathering(&networks);
+  }

[pthatcher1, 2016/06/07 18:54:33]

It seems like a lot of this logic should be in a separate methods called
RegatherNetworksWithAllFailedConnections or RegatherFailedNetworks, unless we
move most of it to PortAllocator::GetFailedNetworks(), in which case it would be
StartRegathering(allocator->GetFailedNetworks()).  

In fact, why even have StartRegathering logic in P2pTransportChannel?  Why not
have that in PortAllocator?  Why not have the method be
PortAllocator::RegatherFailedNetworks()?   That would seem even more clear to
me. 

[honghaiz3, 2016/06/29 02:48:23]

Done.

+
+  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(

[pthatcher1, 2016/06/07 18:54:33]

It sounds like this method should be called RegatherFailedNetworksEvery(int
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;
+    }

[pthatcher1, 2016/06/07 18:54:33]

***

     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 connection was removed and it was writable. If it was not
+    // 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();

[pthatcher1, 2016/06/07 18:54:33]

Should this be last_regather_time_, and should it be set inside of
StartRegathering?

[honghaiz3, 2016/06/29 02:48:23]

Done.

+        StartRegathering(nullptr);
+      }

[pthatcher1, 2016/06/07 18:54:33]

Can we put this in a separate methods called MaybeRegather() or
RegatherIfAllConnectionsJustFailed()?

[honghaiz3, 2016/06/29 02:48:23]

Removed. 

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

[pthatcher1, 2016/06/07 18:54:33]

It feels like this should be Close(), which should then fire
SignalCandidatesRemoved, which we can listen to.

+
+  std::vector<Candidate> candidates =
+      allocator_session()->ReadyCandidates(port);
+  for (Candidate candidate : candidates) {
     candidate.set_transport_name(transport_name());
+    candidates_removed_.push_back(candidate);
   }
-  SignalCandidatesRemoved(this, candidates);
+  thread()->Post(this, MSG_SIGNAL_CANDIDATES_REMOVED);

[pthatcher1, 2016/06/07 18:54:33]

Why does this need a thread hop?

[honghaiz3, 2016/06/29 02:48:23]

We can combine multiple candidate removal signals into one. 

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

[pthatcher1, 2016/06/07 18:54:33]

I feel like we're confounding two things by triggering OnPortNetworkInactive
directly here.  If we have a different signal and that triggers similar
behavior, I think that's OK.  But confounding the two gets too complex.

For example, we could, here, call port->Close() which would then fire a
port->SignalCandidatesRemoved, which we could listen to.  

+  }
+
+  allocator_session()->GetPortsOnNetworks(networks);
 }
 
 // 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