Implement Turn/Turn first logic for connection selection.

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.
  */
 
@@ skipping 13 matching lines @@
 #include "webrtc/system_wrappers/include/field_trial.h"
 
 namespace {
 
 // messages for queuing up work for ourselves
 enum { MSG_SORT = 1, MSG_CHECK_AND_PING };
 
 // The minimum improvement in RTT that justifies a switch.
 static 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;
+}
+
 cricket::PortInterface::CandidateOrigin GetOrigin(cricket::PortInterface* port,
                                          cricket::PortInterface* origin_port) {
   if (!origin_port)
     return cricket::PortInterface::ORIGIN_MESSAGE;
   else if (port == origin_port)
     return cricket::PortInterface::ORIGIN_THIS_PORT;
   else
     return cricket::PortInterface::ORIGIN_OTHER_PORT;
 }
 
@@ skipping 186 matching lines @@
       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_delay_(MIN_CHECK_RECEIVING_DELAY * 5),
-      receiving_timeout_(MIN_CHECK_RECEIVING_DELAY * 50),
-      backup_connection_ping_interval_(0) {
+      config_(MIN_CHECK_RECEIVING_DELAY * 50,
+              0,
+              false,
+              false,

[pthatcher1, 2016/03/02 20:54:36]

Can you put a comment of what these are so it's easier to read?

[guoweis_webrtc, 2016/03/02 22:10:47]

Done.

+              MAX_CURRENT_STRONG_DELAY) {
   uint32_t weak_ping_delay = ::strtoul(
       webrtc::field_trial::FindFullName("WebRTC-StunInterPacketDelay").c_str(),
       nullptr, 10);
   if (weak_ping_delay) {
     weak_ping_delay_ =  weak_ping_delay;
   }
 }
 
 P2PTransportChannel::~P2PTransportChannel() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
@@ skipping 18 matching lines @@
   session->SignalPortReady.connect(this, &P2PTransportChannel::OnPortReady);
   session->SignalCandidatesReady.connect(
       this, &P2PTransportChannel::OnCandidatesReady);
   session->SignalCandidatesAllocationDone.connect(
       this, &P2PTransportChannel::OnCandidatesAllocationDone);
   session->StartGettingPorts();
 }
 
 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(receiving_timeout_);
+  connection->set_receiving_timeout(config_.receiving_timeout_ms);
   connection->SignalReadPacket.connect(
       this, &P2PTransportChannel::OnReadPacket);
   connection->SignalReadyToSend.connect(
       this, &P2PTransportChannel::OnReadyToSend);
   connection->SignalStateChange.connect(
       this, &P2PTransportChannel::OnConnectionStateChange);
   connection->SignalDestroyed.connect(
       this, &P2PTransportChannel::OnConnectionDestroyed);
   connection->SignalNominated.connect(this, &P2PTransportChannel::OnNominated);
   had_connection_ = true;
@@ skipping 89 matching lines @@
   for (Connection* conn : connections_) {
     conn->MaybeSetRemoteIceCredentials(ice_ufrag, ice_pwd);
   }
 }
 
 void P2PTransportChannel::SetRemoteIceMode(IceMode mode) {
   remote_ice_mode_ = mode;
 }
 
 void P2PTransportChannel::SetIceConfig(const IceConfig& config) {
-  gather_continually_ = config.gather_continually;
-  LOG(LS_INFO) << "Set gather_continually to " << gather_continually_;
+  config_.gather_continually = config.gather_continually;
+  LOG(LS_INFO) << "Set gather_continually to " << config_.gather_continually;
 
   if (config.backup_connection_ping_interval >= 0 &&
-      backup_connection_ping_interval_ !=
+      config_.backup_connection_ping_interval !=
           config.backup_connection_ping_interval) {
-    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 "
-                 << backup_connection_ping_interval_ << " milliseconds.";
+                 << config_.backup_connection_ping_interval << " milliseconds.";
   }
 
   if (config.receiving_timeout_ms >= 0 &&
-      receiving_timeout_ != config.receiving_timeout_ms) {
-    receiving_timeout_ = config.receiving_timeout_ms;
+      config_.receiving_timeout_ms != config.receiving_timeout_ms) {
+    config_.receiving_timeout_ms = config.receiving_timeout_ms;
     check_receiving_delay_ =
-        std::max(MIN_CHECK_RECEIVING_DELAY, receiving_timeout_ / 10);
+        std::max(MIN_CHECK_RECEIVING_DELAY, config_.receiving_timeout_ms / 10);
 
     for (Connection* connection : connections_) {
-      connection->set_receiving_timeout(receiving_timeout_);
+      connection->set_receiving_timeout(config_.receiving_timeout_ms);
     }
-    LOG(LS_INFO) << "Set ICE receiving timeout to " << receiving_timeout_
-                 << " milliseconds";
+    LOG(LS_INFO) << "Set ICE receiving timeout to "
+                 << config_.receiving_timeout_ms << " milliseconds";
   }
+
+  config_.prioritize_most_likely_candidate_pairs =
+      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_delay >= 0 &&
+      config_.max_strong_delay != config.max_strong_delay) {
+    config_.max_strong_delay = config.max_strong_delay;
+    LOG(LS_INFO) << "Set max strong delay to " << config_.max_strong_delay;
+  }
+}
+
+const IceConfig& P2PTransportChannel::GetIceConfig() const {

[pthatcher1, 2016/03/02 20:54:36]

Can you make this just config()?

[guoweis_webrtc, 2016/03/02 22:10:47]

Done.

+  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;
@@ skipping 198 matching lines @@
       return;
     } else {
       ASSERT(false);
       port->SendBindingErrorResponse(stun_msg, address,
                                      STUN_ERROR_SERVER_ERROR,
                                      STUN_ERROR_REASON_SERVER_ERROR);
       return;
     }
   }
 
-  Connection* connection = port->CreateConnection(
-      remote_candidate, cricket::PortInterface::ORIGIN_THIS_PORT);
+  Connection* connection =
+      port->CreateConnection(remote_candidate, PortInterface::ORIGIN_THIS_PORT);
   if (!connection) {
     ASSERT(false);
     port->SendBindingErrorResponse(stun_msg, address, STUN_ERROR_SERVER_ERROR,
                                    STUN_ERROR_REASON_SERVER_ERROR);
     return;
   }
 
   LOG(LS_INFO) << "Adding connection from "
                << (remote_candidate_is_new ? "peer reflexive" : "resurrected")
                << " candidate: " << remote_candidate.ToString();
@@ skipping 156 matching lines @@
                 << connection->remote_candidate().ToString()
                 << "New remote candidate: "
                 << remote_candidate.ToString();
       return false;
     }
   } else {
     PortInterface::CandidateOrigin origin = GetOrigin(port, origin_port);
 
     // Don't create connection if this is a candidate we received in a
     // message and we are not allowed to make outgoing connections.
-    if (origin == cricket::PortInterface::ORIGIN_MESSAGE && incoming_only_)
+    if (origin == PortInterface::ORIGIN_MESSAGE && incoming_only_)
       return false;
 
     connection = port->CreateConnection(remote_candidate, origin);
     if (!connection)
       return false;
 
     AddConnection(connection);
 
     LOG_J(LS_INFO, this) << "Created connection with origin=" << origin << ", ("
                          << connections_.size() << " total)";
   }
 
   return true;
 }
 
-bool P2PTransportChannel::FindConnection(
-    cricket::Connection* connection) const {
+bool P2PTransportChannel::FindConnection(Connection* connection) const {
   std::vector<Connection*>::const_iterator citer =
       std::find(connections_.begin(), connections_.end(), connection);
   return citer != connections_.end();
 }
 
 uint32_t P2PTransportChannel::GetRemoteCandidateGeneration(
     const Candidate& candidate) {
   // If the candidate has a ufrag, use it to find the generation.
   if (!candidate.username().empty()) {
     uint32_t generation = 0;
@@ skipping 309 matching lines @@
   if (!IsGettingPorts()) {
     return;
   }
 
   for (PortAllocatorSession* session : allocator_sessions_) {
     if (!session->IsGettingPorts()) {
       continue;
     }
     // If gathering continually, keep the last session running so that it
     // will gather candidates if the networks change.
-    if (gather_continually_ && session == allocator_sessions_.back()) {
+    if (config_.gather_continually && session == allocator_sessions_.back()) {
       session->ClearGettingPorts();
       break;
     }
     session->StopGettingPorts();
   }
 }
 
 // If all connections timed out, delete them all.
 void P2PTransportChannel::HandleAllTimedOut() {
   for (Connection* connection : connections_) {
@@ skipping 89 matching lines @@
 
   // If the channel is weakly connected, ping all connections.
   if (weak()) {
     return true;
   }
 
   // Always ping active connections regardless whether the channel is completed
   // or not, but backup connections are pinged at a slower rate.
   if (IsBackupConnection(conn)) {
     return (now >= conn->last_ping_response_received() +
-                       backup_connection_ping_interval_);
+                       config_.backup_connection_ping_interval);
   }
   return conn->active();
 }
 
 // Returns the next pingable connection to ping.  This will be the oldest
 // pingable connection unless we have a connected, writable connection that is
 // past the maximum acceptable ping delay. When reconnecting a TCP connection,
 // the best connection is disconnected, although still WRITABLE while
 // reconnecting. The newly created connection should be selected as the ping
 // target to become writable instead. See the big comment in CompareConnections.
 Connection* P2PTransportChannel::FindNextPingableConnection() {
   uint32_t now = rtc::Time();
+  Connection* conn_to_ping = nullptr;
   if (best_connection_ && best_connection_->connected() &&
       best_connection_->writable() &&
-      (best_connection_->last_ping_sent() + MAX_CURRENT_STRONG_DELAY <= now)) {
-    return best_connection_;
+      (best_connection_->last_ping_sent() + config_.max_strong_delay <= now)) {
+    conn_to_ping = best_connection_;
+  } else {
+    conn_to_ping = FindConnectionToPing(now);
   }
-
-  // First, find "triggered checks".  We ping first those connections
-  // that have received a ping but have not sent a ping since receiving
-  // it (last_received_ping > last_sent_ping).  But we shouldn't do
-  // triggered checks if the connection is already writable.
-  Connection* oldest_needing_triggered_check = nullptr;
-  Connection* oldest = nullptr;
-  for (Connection* conn : connections_) {
-    if (!IsPingable(conn, now)) {
-      continue;
-    }
-    bool needs_triggered_check =
-        (!conn->writable() &&
-         conn->last_ping_received() > conn->last_ping_sent());
-    if (needs_triggered_check &&
-        (!oldest_needing_triggered_check ||
-         (conn->last_ping_received() <
-          oldest_needing_triggered_check->last_ping_received()))) {
-      oldest_needing_triggered_check = conn;
-    }
-    if (!oldest || (conn->last_ping_sent() < oldest->last_ping_sent())) {
-      oldest = conn;
-    }
+  if (conn_to_ping && pinged_connections_.insert(conn_to_ping).second) {
+    unpinged_connections_.erase(conn_to_ping);

[pthatcher1, 2016/03/02 20:54:36]

I think it would be more correct to move this to OnCheckAndPing:

Connection* conn = FindNextPingableConnection();
if (conn) {
  PingConnection(conn);
  if (conn && pinged_connections_.insert(conn_to_ping).second) {
    unpinged_connections_.erase(conn_to_ping);
  } 
}

In other words, FindNext shouldn't change the pinged state, Ping should.

[guoweis_webrtc, 2016/03/02 22:10:47]

Done.

   }
-
-  if (oldest_needing_triggered_check) {
-    LOG(LS_INFO) << "Selecting connection for triggered check: " <<
-        oldest_needing_triggered_check->ToString();
-    return oldest_needing_triggered_check;
-  }
-  return oldest;
+  return conn_to_ping;
 }
 
 // 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
@@ skipping 49 matching lines @@
 void P2PTransportChannel::OnConnectionDestroyed(Connection* connection) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   // Note: the previous best_connection_ may be destroyed by now, so don't
   // use it.
 
   // Remove this connection from the list.
   std::vector<Connection*>::iterator iter =
       std::find(connections_.begin(), connections_.end(), connection);
   ASSERT(iter != connections_.end());
+  pinged_connections_.erase(*iter);
+  unpinged_connections_.erase(*iter);
   connections_.erase(iter);
 
   LOG_J(LS_INFO, this) << "Removed connection ("
     << static_cast<int>(connections_.size()) << " remaining)";
 
   if (pending_best_connection_ == connection) {
     pending_best_connection_ = NULL;
   }
 
   // If this is currently the best connection, then we need to pick a new one.
@@ skipping 24 matching lines @@
   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.
-  if (!gather_continually_) {
+  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";
@@ skipping 27 matching lines @@
 
   SignalSentPacket(this, sent_packet);
 }
 
 void P2PTransportChannel::OnReadyToSend(Connection* connection) {
   if (connection == best_connection_ && writable()) {
     SignalReadyToSend(this);
   }
 }
 
+// Find "triggered checks".  We ping first those connections that have
+// received a ping but have not sent a ping since receiving it
+// (last_received_ping > last_sent_ping).  But we shouldn't do
+// triggered checks if the connection is already writable.
+Connection* P2PTransportChannel::FindOldestConnectionNeedingTriggeredCheck(
+    uint32_t now) {
+  Connection* oldest_needing_triggered_check = nullptr;
+  for (auto conn : connections_) {
+    if (!IsPingable(conn, now)) {
+      continue;
+    }
+    bool needs_triggered_check =
+        (!conn->writable() &&
+         conn->last_ping_received() > conn->last_ping_sent());
+    if (needs_triggered_check &&
+        (!oldest_needing_triggered_check ||
+         (conn->last_ping_received() <
+          oldest_needing_triggered_check->last_ping_received()))) {
+      oldest_needing_triggered_check = conn;
+    }
+  }
+
+  if (oldest_needing_triggered_check) {
+    LOG(LS_INFO) << "Selecting connection for triggered check: "
+                 << oldest_needing_triggered_check->ToString();
+  }
+  return oldest_needing_triggered_check;
+}
+
+Connection* P2PTransportChannel::FindConnectionToPing(uint32_t now) {
+  RTC_CHECK(connections_.size() ==
+            pinged_connections_.size() + unpinged_connections_.size());
+
+  // If there is nothing pingable in the |unpinged_connections_|, copy over from
+  // |pinged_connections_|.
+  if (std::find_if(unpinged_connections_.begin(), unpinged_connections_.end(),
+                   [this, now](Connection* conn) {
+                     return this->IsPingable(conn, now);
+                   }) == unpinged_connections_.end()) {
+    unpinged_connections_.insert(pinged_connections_.begin(),
+                                 pinged_connections_.end());
+    pinged_connections_.clear();
+  }

[pthatcher1, 2016/03/02 20:54:36]

I think this should also happen in OnCheckAndPing.  Whenever we move something
out of unpinged_connections_, then check if it's "empty".

[guoweis_webrtc, 2016/03/02 22:10:47]

I put it here on purpose such that if we get a new connection after we ping the
last existing one, the new one will have high priority.

+
+  Connection* conn_to_ping = FindOldestConnectionNeedingTriggeredCheck(now);
+  if (conn_to_ping) {
+    return conn_to_ping;
+  }
+
+  for (Connection* conn : unpinged_connections_) {
+    if (!IsPingable(conn, now)) {
+      continue;
+    }
+    if (!conn_to_ping ||
+        SelectMorePingableConnection(conn_to_ping, conn) == conn) {
+      conn_to_ping = conn;
+    }
+  }
+  return conn_to_ping;
+}
+
+Connection* P2PTransportChannel::MostLikelyToWork(Connection* conn1,
+                                                  Connection* conn2) {
+  bool rr1 = IsRelayRelay(conn1);
+  bool rr2 = IsRelayRelay(conn2);
+  if (rr1 && !rr2) {
+    return conn1;
+  } else if (rr2 && !rr1) {
+    return conn2;
+  } else if (rr1 && rr2) {
+    bool udp1 = IsUdp(conn1);
+    bool udp2 = IsUdp(conn2);
+    if (udp1 && !udp2) {
+      return conn1;
+    } else if (udp2 && udp1) {
+      return conn2;
+    }
+  }
+  return nullptr;
+}
+
+Connection* P2PTransportChannel::LeastRecentlyPinged(Connection* conn1,
+                                                     Connection* conn2) {
+  if (conn1->last_ping_sent() < conn2->last_ping_sent()) {
+    return conn1;
+  }
+  if (conn1->last_ping_sent() > conn2->last_ping_sent()) {
+    return conn2;
+  }
+  return nullptr;
+}
+
+Connection* P2PTransportChannel::SelectMorePingableConnection(

[pthatcher1, 2016/03/02 20:54:36]

To keep up with your naming conventions, this should be MostPingable

[guoweis_webrtc, 2016/03/02 22:10:47]

Done.

+    Connection* conn1,
+    Connection* conn2) {
+  RTC_DCHECK(conn1 != conn2);
+  if (config_.prioritize_most_likely_candidate_pairs) {
+    Connection* most_likely_to_work_conn = MostLikelyToWork(conn1, conn2);
+    if (most_likely_to_work_conn) {
+      return most_likely_to_work_conn;
+    }
+  }
+
+  Connection* least_recently_pinged_conn = LeastRecentlyPinged(conn1, conn2);
+  if (least_recently_pinged_conn) {
+    return least_recently_pinged_conn;
+  }
+
+  // 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