Dampening connection switch.

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 62 matching lines @@
 const int WEAK_PING_INTERVAL = 1000 * PING_PACKET_SIZE / 10000;
 
 // Writable connections are pinged at a faster rate while stabilizing.
 const int STABILIZING_WRITABLE_CONNECTION_PING_INTERVAL = 900;  // ms
 
 // Writable connections are pinged at a slower rate once stabilized.
 const int STABLE_WRITABLE_CONNECTION_PING_INTERVAL = 2500;  // ms
 
 static const int MIN_CHECK_RECEIVING_INTERVAL = 50;  // ms
 
+static const int RECEIVING_DAMPENING_INTERVAL = 500;  // ms

[pthatcher1, 2016/07/13 21:17:23]

I think we should be conservative with the value for now and stick with 1000ms. 
If we feel confident that we can lower it later, we can.  But I think it's
better to stick with a bigger value and then test going down rather than risk
being too low.

[honghaiz3, 2016/07/14 04:54:20]

There is actually some risk of choosing the same dampening value and the
receiving timeout value. 
Because by the time we re-check it again the backup connection may become not
receiving if only one stun-ping response message was received on the backup
connection. This may cause some switching not happening while it should happen. 
I set it to 900ms for now. 

+
 // We periodically check if any existing networks do not have any connection
 // and regather on those networks.
 static const int DEFAULT_REGATHER_ON_FAILED_NETWORKS_INTERVAL = 5 * 60 * 1000;
 static constexpr int a_is_better = 1;
 static constexpr int b_is_better = -1;
 
 P2PTransportChannel::P2PTransportChannel(const std::string& transport_name,
                                          int component,
                                          P2PTransport* transport,
                                          PortAllocator* allocator)
@@ skipping 12 matching lines @@
       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 */,
               GATHER_ONCE /* continual_gathering_policy */,
               false /* prioritize_most_likely_candidate_pairs */,
               STABLE_WRITABLE_CONNECTION_PING_INTERVAL,
               true /* presume_writable_when_fully_relayed */,
-              DEFAULT_REGATHER_ON_FAILED_NETWORKS_INTERVAL) {
+              DEFAULT_REGATHER_ON_FAILED_NETWORKS_INTERVAL,
+              RECEIVING_DAMPENING_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());
@@ skipping 51 matching lines @@
 // criteria is as follows:
 // i) write/receiving/connected states
 // ii) For controlled side,
 //        a) nomination state,
 //        b) last data received time.
 // iii) Lower cost / higher priority.
 // iv) rtt.
 // TODO(honghaiz): Stop the aggressive nomination on the controlling side and
 // implement the ice-renomination option.
 bool P2PTransportChannel::ShouldSwitchSelectedConnection(
-    Connection* new_connection) const {
+    Connection* new_connection) {
   if (!new_connection || selected_connection_ == new_connection) {
     return false;
   }
 
   if (selected_connection_ == nullptr) {
     return true;
   }
 
-  int cmp = CompareConnections(selected_connection_, new_connection);
+  bool connection_dampened = false;
+  int cmp = CompareConnectionStatesWithDampening(
+      selected_connection_, new_connection, &connection_dampened);
   if (cmp != 0) {
     return cmp < 0;
   }
+  // A connection is dampened indicates that it is in a better receiving state,
+  // so we need to remember it and check it again periodically.
+  if (connection_dampened) {
+    pending_selected_connection_ = new_connection;
+  }

[pthatcher1, 2016/07/13 21:17:23]

I don't like the idea of ShouldSwitchSelectedConnection being non-const and
changing state.

And I don't think we need to store the pending_selected_connection_.  I think
there are alternatives that are more simple.  

For example, if we know we need to wait X ms to see if we're past the dampening
period, we could just schedule a MSG_SORT_AND_UPDATE_STATE that will take care
of switching, if it should.

[honghaiz3, 2016/07/14 04:54:20]

Done for other parts except that
ShouldSwitchSelectedConnection needs to schedule a event, which access this,
which is not a const argument. 
So there is no easy way to keep the method const. 

[pthatcher1, 2016/07/14 18:01:44]

Can't the *caller* of ShouldSwitchSelectedConnection schedule the re-comparison?
 Then it can remain const.

+
+  cmp = CompareConnectionsWithoutStates(selected_connection_, new_connection);

[pthatcher1, 2016/07/13 21:17:23]

I also think it would be more simple to keep it as one CompareConnections and
not break it up into two (WithDampening and WithoutStates).

[honghaiz3, 2016/07/14 04:54:21]

Done.

+  if (cmp != 0) {
+    return cmp < 0;
+  }
 
   // If everything else is the same, switch only if rtt has improved by
   // a margin.
   return new_connection->rtt() <= selected_connection_->rtt() - kMinImprovement;
 }
 
 void P2PTransportChannel::SetIceRole(IceRole ice_role) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   if (ice_role_ != ice_role) {
     ice_role_ = ice_role;
@@ skipping 152 matching lines @@
                    << config_.presume_writable_when_fully_relayed;
     }
   }
 
   if (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;
   }
+  if (config.receiving_dampening_interval) {
+    config_.receiving_dampening_interval = config.receiving_dampening_interval;
+    LOG(LS_INFO) << "Set receiving_dampening_interval to"
+                 << *config_.receiving_dampening_interval;
+  }
 }
 
 const IceConfig& P2PTransportChannel::config() const {
   return config_;
 }
 
 void P2PTransportChannel::MaybeStartGathering() {
   if (ice_ufrag_.empty() || ice_pwd_.empty()) {
     return;
   }
@@ skipping 246 matching lines @@
   }
 
   if (!ShouldSwitchSelectedConnection(conn)) {
     LOG(LS_INFO)
         << "Not switching the selected connection on controlled side yet: "
         << conn->ToString();
     return;
   }
 
   LOG(LS_INFO)
-      << "Switching selected connection on controlled side due to nomination: "
-      << conn->ToString();
+      << "Switching selected connection on controlled side due to nomination.";
   SwitchSelectedConnection(conn);
   // Now that we have selected a connection, it is time to prune other
   // connections and update the read/write state of the channel.
   RequestSortAndStateUpdate();
 }
 
 void P2PTransportChannel::AddRemoteCandidate(const Candidate& candidate) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   uint32_t generation = GetRemoteCandidateGeneration(candidate);
@@ skipping 299 matching lines @@
 
 // Monitor connection states.
 void P2PTransportChannel::UpdateConnectionStates() {
   int64_t now = rtc::TimeMillis();
 
   // We need to copy the list of connections since some may delete themselves
   // when we call UpdateState.
   for (Connection* c : connections_) {
     c->UpdateState(now);
   }
+
+  if (!selected_connection_) {
+    was_strong_ = false;
+    return;
+  }
+  bool is_strong = !selected_connection_->weak();
+  if (was_strong_ && !is_strong) {
+    // If the selected connection just changes to weak, reset the number of
+    // pings sent on the selected connection so that it will be pinged at least
+    // a few times.
+    selected_connection_->reset_num_pings_sent();
+    for (Connection* c : connections_) {
+      c->reset_first_received_since_channel_weak();
+    }
+  }
+  was_strong_ = is_strong;

[pthatcher1, 2016/07/13 21:17:23]

I think we can calculate the "has it been receiving long enough" logic
independently for each Connection and avoid having to reach into the connection
with things like reset_num_pings_sent() and
reset_first_received_since_channel_weak().

For example, we can simply have a "continuously_receiving_since()" that give the
timestamp of the time we went from non receiving -> receiving.  

[honghaiz3, 2016/07/14 04:54:21]

We can do that. But a potential problem is that if the backup connection was
receiving (for example due to the periodic ping message on the backup
connection), it will switch right away. 
So I just reset the continuously_receiving_since to |now| if the selected
connection becomes weak. 

I realize reset_num_pings_sent is not needed any more because if the switching
is dampened, it will be in weak state and so it will be pinged in fast rate. 

 }
 
 // Prepare for best candidate sorting.
 void P2PTransportChannel::RequestSortAndStateUpdate() {
   if (!sort_dirty_) {
     worker_thread_->Post(RTC_FROM_HERE, this, MSG_SORT_AND_UPDATE_STATE);
     sort_dirty_ = true;
   }
 }
 
 void P2PTransportChannel::MaybeStartPinging() {
   if (started_pinging_) {
     return;
   }
 
   int64_t now = rtc::TimeMillis();
   if (std::any_of(
           connections_.begin(), connections_.end(),
           [this, now](const Connection* c) { return IsPingable(c, now); })) {
     LOG_J(LS_INFO, this) << "Have a pingable connection for the first time; "
                          << "starting to ping.";
     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);
     started_pinging_ = true;
   }
 }
 
-// Compare two connections based on their writing, receiving, and connected
-// states.
-int P2PTransportChannel::CompareConnectionStates(const Connection* a,
-                                                 const Connection* b) const {
+int P2PTransportChannel::CompareConnectionWriteAndConnectedStates(
+    const Connection* a,
+    const Connection* b) const {
   // First, prefer a connection that's writable or presumed writable over
   // one that's not writable.
   bool a_writable = a->writable() || PresumedWritable(a);
   bool b_writable = b->writable() || PresumedWritable(b);
   if (a_writable && !b_writable) {
     return a_is_better;
   }
   if (!a_writable && b_writable) {
     return b_is_better;
   }
 
-  // Sort based on write-state. Better states have lower values.
-  if (a->write_state() < b->write_state()) {
-    return a_is_better;
-  }
-  if (b->write_state() < a->write_state()) {
-    return b_is_better;
-  }
-
-  // We prefer a receiving connection to a non-receiving, higher-priority
-  // connection when sorting connections and choosing which connection to
-  // switch to.
-  if (a->receiving() && !b->receiving()) {
-    return a_is_better;
-  }
-  if (!a->receiving() && b->receiving()) {
-    return b_is_better;
+  // Better write-states have lower values, and a negative return value
+  // indicates b is better.
+  int cmp = b->write_state() - a->write_state();
+  if (cmp != 0) {
+    return cmp;
   }
 
   // WARNING: Some complexity here about TCP reconnecting.
   // When a TCP connection fails because of a TCP socket disconnecting, the
   // active side of the connection will attempt to reconnect for 5 seconds while
   // pretending to be writable (the connection is not set to the unwritable
   // state).  On the passive side, the connection also remains writable even
   // though it is disconnected, and a new connection is created when the active
   // side connects.  At that point, there are two TCP connections on the passive
   // side: 1. the old, disconnected one that is pretending to be writable, and
@@ skipping 16 matching lines @@
     if (a->connected() && !b->connected()) {
       return a_is_better;
     }
     if (!a->connected() && b->connected()) {
       return b_is_better;
     }
   }
   return 0;
 }
 
+// Compare two connections based on their writing, connected, and receiving
+// states.
+int P2PTransportChannel::CompareConnectionStates(const Connection* a,
+                                                 const Connection* b) const {
+  int cmp = CompareConnectionWriteAndConnectedStates(a, b);
+  if (cmp != 0) {
+    return cmp;
+  }
+
+  // We prefer a receiving connection to a non-receiving, higher-priority
+  // connection when sorting connections.
+  if (a->receiving() && !b->receiving()) {
+    return a_is_better;
+  }
+  if (!a->receiving() && b->receiving()) {
+    return b_is_better;
+  }

[pthatcher1, 2016/07/13 21:17:22]

Doesn't this change the order so that connected vs not takes precedence over
receiving?

[honghaiz3, 2016/07/14 04:54:20]

I did not think that should have a major impact although we don't have to change
that in this CL. 

+
+  return 0;
+}
+
+int P2PTransportChannel::CompareConnectionStatesWithDampening(
+    const Connection* selected_conn,
+    const Connection* new_conn,

[pthatcher1, 2016/07/13 21:17:23]

CompareConnectionX should know about "selected" and "new" connections.  I think
it would make more sense to pass call them "a" and "b" and then have a
"b_receiving_delay" passed in.  Or better yet, combine "now" with
"b_receiving_delay" into a "b_receiving_threshold".

[honghaiz3, 2016/07/14 04:54:20]

Done

+    bool* new_conn_dampened) const {
+  int cmp = CompareConnectionWriteAndConnectedStates(selected_conn, new_conn);
+  if (cmp != 0) {
+    return cmp;
+  }
+
+  if (selected_conn->receiving() && !new_conn->receiving()) {
+    return a_is_better;
+  }
+
+  if (!selected_conn->receiving() && new_conn->receiving()) {
+    // If |selected_conn| is not receiving and |new_conn| is receiving, we wait
+    // for a short while before switching to the new connection.
+    if (new_conn->first_received_since_channel_weak() != 0 &&
+        rtc::TimeMillis() - new_conn->first_received_since_channel_weak() >=
+            *config_.receiving_dampening_interval) {

[pthatcher1, 2016/07/13 21:17:22]

I think it would make more sense to have the Connection have a method with
something like continuously_receiving_since() that returns a timestamp of the
earliest time of the currently continuous time span that it's been receiving. 
Then this could be:

if (!a->receiving() && b->receiving() &&
    b->continuously_receiving_since() < b_receiving_threshold) {
  return b_is_better;
}
return 0;

[honghaiz3, 2016/07/14 04:54:20]

Done, somewhat different though. 

+      return b_is_better;
+    }
+    *new_conn_dampened = true;
+  }
+
+  return 0;
+}
+
 // Compares two connections based only on the candidate and network information.
 // Returns positive if |a| is better than |b|.
 int P2PTransportChannel::CompareConnectionCandidates(
     const Connection* a,
     const Connection* b) const {
   // Prefer lower network cost.
   uint32_t a_cost = a->ComputeNetworkCost();
   uint32_t b_cost = b->ComputeNetworkCost();
   // Smaller cost is better.
   if (a_cost < b_cost) {
@@ skipping 10 matching lines @@
   if (a->priority() < b->priority()) {
     return b_is_better;
   }
 
   // If we're still tied at this point, prefer a younger generation.
   // (Younger generation means a larger generation number).
   return (a->remote_candidate().generation() + a->port()->generation()) -
          (b->remote_candidate().generation() + b->port()->generation());
 }
 
-int P2PTransportChannel::CompareConnections(const Connection* a,
-                                            const Connection* b) const {
+int P2PTransportChannel::CompareConnectionsWithoutStates(
+    const Connection* a,
+    const Connection* b) const {
   RTC_CHECK(a != nullptr);
   RTC_CHECK(b != nullptr);
 
-  // We prefer to switch to a writable and receiving connection over a
-  // non-writable or non-receiving connection, even if the latter has
-  // been nominated by the controlling side.
-  int state_cmp = CompareConnectionStates(a, b);
-  if (state_cmp != 0) {
-    return state_cmp;
-  }
-
   if (ice_role_ == ICEROLE_CONTROLLED) {
     // Compare the connections based on the nomination states and the last data
     // received time if this is on the controlled side.
     if (a->nominated() && !b->nominated()) {
       return a_is_better;
     }
     if (!a->nominated() && b->nominated()) {
       return b_is_better;
     }
 
     if (a->last_data_received() > b->last_data_received()) {
       return a_is_better;
     }
     if (a->last_data_received() < b->last_data_received()) {
       return b_is_better;
     }
   }
 
   // Compare the network cost and priority.
   return CompareConnectionCandidates(a, b);
 }
 
+int P2PTransportChannel::CompareConnections(const Connection* a,
+                                            const Connection* b) const {
+  // We prefer to switch to a writable and receiving connection over a
+  // non-writable or non-receiving connection, even if the latter has
+  // been nominated by the controlling side.
+  int cmp = CompareConnectionStates(a, b);
+  if (cmp != 0) {
+    return cmp;
+  }
+  cmp = CompareConnectionsWithoutStates(a, b);
+  if (cmp != 0) {
+    return cmp;
+  }
+  // Otherwise, compare the latency estimate. Note that a negative return value
+  // indicates |b| is better.
+  return b->rtt() - a->rtt();
+}
+
 bool P2PTransportChannel::PresumedWritable(const Connection* conn) const {
   return (conn->write_state() == Connection::STATE_WRITE_INIT &&
           config_.presume_writable_when_fully_relayed &&
           conn->local_candidate().type() == RELAY_PORT_TYPE &&
           (conn->remote_candidate().type() == RELAY_PORT_TYPE ||
            conn->remote_candidate().type() == PRFLX_PORT_TYPE));
 }
 
 // Sort the available connections to find the best one.  We also monitor
 // the number of available connections and the current state.
 void P2PTransportChannel::SortConnectionsAndUpdateState() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   // Make sure the connection states are up-to-date since this affects how they
   // will be sorted.
   UpdateConnectionStates();
 
   // Any changes after this point will require a re-sort.
   sort_dirty_ = false;
 
   // Find the best alternative connection by sorting.  It is important to note
   // that amongst equal preference, writable connections, this will choose the
   // one whose estimated latency is lowest.  So it is the only one that we
   // need to consider switching to.
   std::stable_sort(connections_.begin(), connections_.end(),
                    [this](const Connection* a, const Connection* b) {
-                     int cmp = CompareConnections(a, b);
-                     if (cmp != 0) {
-                       return cmp > 0;
-                     }
+                     return CompareConnections(a, b) > 0;

[pthatcher1, 2016/07/13 21:17:22]

I think it would be less risky to leave the CompareConnections methods as-is
(with the exception off adding the b_receiving_threhsold).

[honghaiz3, 2016/07/14 04:54:20]

Done.

 
-                     // Otherwise, sort based on latency estimate.
-                     return a->rtt() < b->rtt();
                    });
 
   LOG(LS_VERBOSE) << "Sorting " << connections_.size()
                   << " available connections:";
   for (size_t i = 0; i < connections_.size(); ++i) {
     LOG(LS_VERBOSE) << connections_[i]->ToString();
   }
 
   Connection* top_connection =
       (connections_.size() > 0) ? connections_[0] : nullptr;
 
   // If necessary, switch to the new choice. Note that |top_connection| doesn't
   // have to be writable to become the selected connection although it will
   // have higher priority if it is writable.
   if (ShouldSwitchSelectedConnection(top_connection)) {
-    LOG(LS_INFO) << "Switching selected connection after sorting: "
-                 << top_connection->ToString();
+    LOG(LS_INFO) << "Switching selected connection after sorting";
     SwitchSelectedConnection(top_connection);
   }
 
   // The controlled side can prune only if the selected connection has been
   // nominated because otherwise it may prune the connection that will be
   // selected by the controlling side.
   // TODO(honghaiz): This is not enough to prevent a connection from being
   // pruned too early because with aggressive nomination, the controlling side
   // will nominate every connection until it becomes writable.
   if (ice_role_ == ICEROLE_CONTROLLING ||
@@ skipping 60 matching lines @@
   }
 }
 
 // Change the selected connection, and let listeners know.
 void P2PTransportChannel::SwitchSelectedConnection(Connection* conn) {
   // Note: if conn is NULL, the previous |selected_connection_| has been
   // destroyed, so don't use it.
   Connection* old_selected_connection = selected_connection_;
   selected_connection_ = conn;
   if (selected_connection_) {
+    if (pending_selected_connection_ == selected_connection_) {
+      pending_selected_connection_ = nullptr;
+    }
     if (old_selected_connection) {
       LOG_J(LS_INFO, this) << "Previous selected connection: "
                            << old_selected_connection->ToString();
     }
     LOG_J(LS_INFO, this) << "New selected connection: "
                          << selected_connection_->ToString();
     SignalRouteChange(this, selected_connection_->remote_candidate());
     // This is a temporary, but safe fix to webrtc issue 5705.
     // TODO(honghaiz): Make all EWOULDBLOCK error routed through the transport
     // channel so that it knows whether the media channel is allowed to
@@ skipping 148 matching lines @@
       ASSERT(false);
       break;
   }
 }
 
 // Handle queued up check-and-ping request
 void P2PTransportChannel::OnCheckAndPing() {
   // Make sure the states of the connections are up-to-date (since this affects
   // which ones are pingable).
   UpdateConnectionStates();
+  if (pending_selected_connection_) {
+    if (ShouldSwitchSelectedConnection(pending_selected_connection_)) {
+      LOG(LS_INFO) << "Switching selected connection to a pending one";
+      SwitchSelectedConnection(pending_selected_connection_);
+    } else if (!weak()) {
+      pending_selected_connection_ = nullptr;
+    }
+  }

[pthatcher1, 2016/07/13 21:17:23]

This code, or the equivalent, really should be inside of UpdateConnectionStates.

[honghaiz3, 2016/07/14 04:54:20]

This serves the purpose of periodical checking the dampened connection. 
Now that is not needed, just remove this. 

+
   // When the selected connection is not receiving or not writable, or any
   // active connection has not been pinged enough times, use the weak ping
   // interval.
   bool need_more_pings_at_weak_interval = std::any_of(
       connections_.begin(), connections_.end(), [](Connection* conn) {
         return conn->active() &&
                conn->num_pings_sent() < MIN_PINGS_AT_WEAK_PING_INTERVAL;
       });
   int ping_interval = (weak() || need_more_pings_at_weak_interval)
                           ? weak_ping_interval_
@@ skipping 187 matching lines @@
   LOG_J(LS_INFO, this) << "Removed connection ("
     << static_cast<int>(connections_.size()) << " remaining)";
 
   // If this is currently the selected connection, then we need to pick a new
   // one. The call to SortConnectionsAndUpdateState will pick a new one. It
   // looks at the current selected connection in order to avoid switching
   // between fairly similar ones. Since this connection is no longer an option,
   // we can just set selected to nullptr and re-choose a best assuming that
   // there was no selected connection.
   if (selected_connection_ == connection) {
-    LOG(LS_INFO) << "selected connection destroyed. Will choose a new one.";
+    LOG(LS_INFO) << "Selected connection destroyed. Will choose a new one.";
     SwitchSelectedConnection(nullptr);
     RequestSortAndStateUpdate();
   } else {
     // If a non-selected connection was destroyed, we don't need to re-sort but
     // we do need to update state, because we could be switching to "failed" or
     // "completed".
     UpdateState();
   }
 }
 
@@ skipping 86 matching lines @@
     return;
 
   // Let the client know of an incoming packet
   SignalReadPacket(this, data, len, packet_time, 0);
 
   // May need to switch the sending connection based on the receiving media path
   // if this is the controlled side.
   if (ice_role_ == ICEROLE_CONTROLLED &&
       ShouldSwitchSelectedConnection(connection)) {
     LOG(LS_INFO) << "Switching selected connection on controlled side due to "
-                 << "data received: " << connection->ToString();
+                 << "data received";
     SwitchSelectedConnection(connection);
   }
 }
 
 void P2PTransportChannel::OnSentPacket(const rtc::SentPacket& sent_packet) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   SignalSentPacket(this, sent_packet);
 }
 
@@ skipping 114 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