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_REGATHER_ON_FAILED_NETWORKS,
+  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 if any existing networks do not have any connection
+// and regather on those failed networks.
+static const int DEFAULT_REGATHER_ON_FAILED_NETWORKS_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),
-      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 */) {
+      config_(
+          MIN_CHECK_RECEIVING_INTERVAL * 50 /* receiving_timeout */,
+          0 /* backup_connection_ping_interval */,
+          GATHER_ONCE,
+          false /* prioritize_most_likely_candidate_pairs */,
+          MAX_CURRENT_STRONG_INTERVAL /* max_strong_interval */,
+          DEFAULT_REGATHER_ON_FAILED_NETWORKS_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);
   }
 }
 
 P2PTransportChannel::~P2PTransportChannel() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 }
 
 // Add the allocator session to our list so that we know which sessions
 // are still active.
 void P2PTransportChannel::AddAllocatorSession(
     std::unique_ptr<PortAllocatorSession> session) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   session->set_generation(static_cast<uint32_t>(allocator_sessions_.size()));
   session->SignalPortReady.connect(this, &P2PTransportChannel::OnPortReady);
   session->SignalCandidatesReady.connect(
       this, &P2PTransportChannel::OnCandidatesReady);
+  session->SignalCandidatesRemoved.connect(
+      this, &P2PTransportChannel::OnCandidatesRemoved);
   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();
 
   allocator_sessions_.push_back(std::move(session));
 }
@@ skipping 109 matching lines @@
   }
   // Updating the remote ICE candidate generation could change the sort order.
   RequestSort();
 }
 
 void P2PTransportChannel::SetRemoteIceMode(IceMode mode) {
   remote_ice_mode_ = mode;
 }
 
 void P2PTransportChannel::SetIceConfig(const IceConfig& config) {
-  config_.gather_continually = config.gather_continually;
-  LOG(LS_INFO) << "Set gather_continually to " << config_.gather_continually;
+  config_.gathering_policy = config.gathering_policy;
+  LOG(LS_INFO) << "Set gathering_policy to " << config_.gathering_policy;
 
   if (config.backup_connection_ping_interval >= 0 &&
       config_.backup_connection_ping_interval !=
           config.backup_connection_ping_interval) {
     config_.backup_connection_ping_interval =
         config.backup_connection_ping_interval;
     LOG(LS_INFO) << "Set backup connection ping interval to "
                  << config_.backup_connection_ping_interval << " milliseconds.";
   }
 
@@ skipping 14 matching lines @@
       config.prioritize_most_likely_candidate_pairs;
   LOG(LS_INFO) << "Set ping most likely connection to "
                << config_.prioritize_most_likely_candidate_pairs;
 
   if (config.max_strong_interval >= 0 &&
       config_.max_strong_interval != config.max_strong_interval) {
     config_.max_strong_interval = config.max_strong_interval;
     LOG(LS_INFO) << "Set max strong interval to "
                  << config_.max_strong_interval;
   }
+
+  if (config.regather_on_failed_networks_interval >= 0 &&
+      config.regather_on_failed_networks_interval !=
+          config_.regather_on_failed_networks_interval) {
+    config_.regather_on_failed_networks_interval =
+        config.regather_on_failed_networks_interval;
+    LOG(LS_INFO) << "Set regather_on_failed_networks_interval to "
+                 << config_.regather_on_failed_networks_interval;
+  }
 }
 
 const IceConfig& P2PTransportChannel::config() const {
   return config_;
 }
 
 // Go into the state of processing candidates, and running in general
 void P2PTransportChannel::Connect() {
   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(RTC_FROM_HERE, this, MSG_CHECK_AND_PING);
+  thread()->PostDelayed(RTC_FROM_HERE,
+                        config_.regather_on_failed_networks_interval, this,
+                        MSG_REGATHER_ON_FAILED_NETWORKS);
 }
 
 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 44 matching lines @@
   // Remember the ports and candidates, and signal that candidates are ready.
   // The session will handle this, and send an initiate/accept/modify message
   // if one is pending.
 
   port->SetIceRole(ice_role_);
   port->SetIceTiebreaker(tiebreaker_);
   ports_.push_back(port);
   port->SignalUnknownAddress.connect(
       this, &P2PTransportChannel::OnUnknownAddress);
   port->SignalDestroyed.connect(this, &P2PTransportChannel::OnPortDestroyed);
-  port->SignalNetworkInactive.connect(
-      this, &P2PTransportChannel::OnPortNetworkInactive);
+  port->SignalClosed.connect(this, &P2PTransportChannel::OnPortClosed);
+
   port->SignalRoleConflict.connect(
       this, &P2PTransportChannel::OnRoleConflict);
   port->SignalSentPacket.connect(this, &P2PTransportChannel::OnSentPacket);
 
   // Attempt to create a connection from this new port to all of the remote
   // candidates that we were given so far.
 
   std::vector<RemoteCandidate>::iterator iter;
   for (iter = remote_candidates_.begin(); iter != remote_candidates_.end();
        ++iter) {
@@ skipping 714 matching lines @@
   }
   set_receiving(receiving);
 }
 
 void P2PTransportChannel::MaybeStopPortAllocatorSessions() {
   if (!IsGettingPorts()) {
     return;
   }
 
   for (const auto& session : allocator_sessions_) {
-    if (!session->IsGettingPorts()) {
+    if (!session->CanGetPorts()) {
+      // The session has completely stopped.
       continue;
     }
-    // If gathering continually, keep the last session running so that it
-    // will gather candidates if the networks change.
-    if (config_.gather_continually && session == allocator_sessions_.back()) {
+    // If gathering continually, keep the last session at STATE_CLEARED so that
+    // it can gather candidates if the networks change.
+    if (config_.gather_continually() && session == allocator_sessions_.back()) {
       session->ClearGettingPorts();
-      break;
+    } else {
+      session->StopGettingPorts();
     }
-    session->StopGettingPorts();
   }
 }
 
 // If all connections timed out, delete them all.
 void P2PTransportChannel::HandleAllTimedOut() {
   for (Connection* connection : connections_) {
     connection->Destroy();
   }
 }
 
@@ skipping 20 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_REGATHER_ON_FAILED_NETWORKS:
+      RegatherOnFailedNetworks();
+      break;
+    case MSG_SIGNAL_CANDIDATES_REMOVED:
+      if (!candidates_to_remove_.empty()) {
+        LOG(LS_INFO) << "Signal removing " << candidates_to_remove_.size()
+                     << " candidates";
+        SignalCandidatesRemoved(this, candidates_to_remove_);
+        candidates_to_remove_.clear();
+      }
+      break;
     default:
       ASSERT(false);
       break;
   }
 }
 
 // Handle queued up sort request
 void P2PTransportChannel::OnSort() {
   // Resort the connections based on the new statistics.
   SortConnections();
@@ skipping 181 matching lines @@
   // 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.";
     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";
+  if (RemovePort(port)) {
+    LOG(INFO) << "Removed port because it is destroyed: " << 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.
-  if (!config_.gather_continually) {
+void P2PTransportChannel::OnPortClosed(PortInterface* port) {
+  ASSERT(worker_thread_ == rtc::Thread::Current());
+
+  if (RemovePort(port)) {
+    LOG(INFO) << "Removed port because it is closed: " << ports_.size()
+              << " remaining";
+  }
+}
+
+bool P2PTransportChannel::RemovePort(PortInterface* port) {
+  auto new_end = std::remove(ports_.begin(), ports_.end(), port);
+  if (new_end != ports_.end()) {
+    ports_.erase(new_end, ports_.end());
+    return true;
+  }
+  return false;
+}
+
+void P2PTransportChannel::OnCandidatesRemoved(
+    PortAllocatorSession* session,
+    const std::vector<Candidate>& candidates) {
+  ASSERT(worker_thread_ == rtc::Thread::Current());
+  // Do not signal candidate removals if continual gathering is not enabled.
+  // Do not signal candidate removals if this is not the last session because
+  // an ICE restart would have signaled the remote side about all candidate
+  // removals in previous sessions.
+  if (!config_.gather_continually() || session != allocator_session()) {
     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()) {
+
+  // If |candidates_to_remove_| is not empty, a msg
+  // MSG_SIGNAL_CANDIDATES_REMOVED should have been posted.
+  if (candidates_to_remove_.empty()) {
+    thread()->Post(this, MSG_SIGNAL_CANDIDATES_REMOVED);
+  }
+
+  for (Candidate candidate : candidates) {
+    candidate.set_transport_name(transport_name());
+    candidates_to_remove_.push_back(candidate);
+  }
+}
+
+void P2PTransportChannel::OnRegatherOnFailedNetworks() {
+  if (!config_.gather_continually() || allocator_sessions_.empty()) {
     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) {
-    candidate.set_transport_name(transport_name());
-  }
-  SignalCandidatesRemoved(this, candidates);
+  allocator_session()->RegatherOnFailedNetworks();
+
+  thread()->PostDelayed(config_.regather_on_failed_network_interval_, this,
+                        MSG_REGATHER_ON_FAILED_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