Make deletion of Connection objects more deterministic.

This changes most deletion paths of Connection objects to go through
the owner class of the Connection instances, Port.

In situations where Connection objects still need to be deleted
asynchronously, `async = true` can be passed to
`Port::DestroyConnection` and get the same behavior as
`Connection::Destroy` formerly gave.

The `Destroy()` method still exists for downstream compatibility, but
instead of deleting connection objects asynchronously, the deletion
now happens synchronously via the Port class.

Bug: webrtc:13892, webrtc:13865
Change-Id: I07edb7bb5e5d93b33542581b4b09def548de9e12
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/259826
Reviewed-by: Danil Chapovalov <danilchap@webrtc.org>
Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#36676}

9 files changed, 100 insertions, 72 deletions

diff --git a/p2p/base/connection.cc b/p2p/base/connection.cc
index 25a4d012a9..04da6bdc4b 100644
--- a/p2p/base/connection.cc
+++ b/p2p/base/connection.cc
@@ -18,7 +18,6 @@
 #include <vector>
 
 #include "absl/algorithm/container.h"
-#include "absl/memory/memory.h"
 #include "absl/strings/match.h"
 #include "p2p/base/port_allocator.h"
 #include "rtc_base/checks.h"
@@ -321,6 +320,7 @@ Connection::Connection(rtc::WeakPtr<Port> port,
 
 Connection::~Connection() {
   RTC_DCHECK_RUN_ON(network_thread_);
+  RTC_DCHECK(!port_);
 }
 
 webrtc::TaskQueueBase* Connection::network_thread() const {
@@ -832,8 +832,14 @@ void Connection::Prune() {
 
 void Connection::Destroy() {
   RTC_DCHECK_RUN_ON(network_thread_);
+  if (port_)
+    port_->DestroyConnection(this);
+}
+
+bool Connection::Shutdown() {
+  RTC_DCHECK_RUN_ON(network_thread_);
   if (!port_)
-    return;
+    return false;  // already shut down.
 
   RTC_DLOG(LS_VERBOSE) << ToString() << ": Connection destroyed";
 
@@ -850,20 +856,7 @@ void Connection::Destroy() {
   // information required for logging needs access to `port_`.
   port_.reset();
 
-  // Unwind the stack before deleting the object in case upstream callers
-  // need to refer to the Connection's state as part of teardown.
-  // NOTE: We move ownership of 'this' into the capture section of the lambda
-  // so that the object will always be deleted, including if PostTask fails.
-  // In such a case (only tests), deletion would happen inside of the call
-  // to `Destroy()`.
-  network_thread_->PostTask(
-      webrtc::ToQueuedTask([me = absl::WrapUnique(this)]() {}));
-}
-
-void Connection::FailAndDestroy() {
-  RTC_DCHECK_RUN_ON(network_thread_);
-  set_state(IceCandidatePairState::FAILED);
-  Destroy();
+  return true;
 }
 
 void Connection::FailAndPrune() {
@@ -873,9 +866,9 @@ void Connection::FailAndPrune() {
   // and Connection. In some cases (Port dtor), a Connection object is deleted
   // without using the `Destroy` method (port_ won't be nulled and some
   // functionality won't run as expected), while in other cases
-  // (AddOrReplaceConnection), the Connection object is deleted asynchronously
-  // via the `Destroy` method and in that case `port_` will be nulled.
-  // However, in that case, there's a chance that the Port object gets
+  // the Connection object is deleted asynchronously and in that case `port_`
+  // will be nulled.
+  // In such a case, there's a chance that the Port object gets
   // deleted before the Connection object ends up being deleted.
   if (!port_)
     return;
@@ -975,7 +968,7 @@ void Connection::UpdateState(int64_t now) {
   // Update the receiving state.
   UpdateReceiving(now);
   if (dead(now)) {
-    Destroy();
+    port_->DestroyConnectionAsync(this);
   }
 }
 
@@ -1393,7 +1386,8 @@ void Connection::OnConnectionRequestErrorResponse(ConnectionRequest* request,
     RTC_LOG(LS_ERROR) << ToString()
                       << ": Received STUN error response, code=" << error_code
                       << "; killing connection";
-    FailAndDestroy();
+    set_state(IceCandidatePairState::FAILED);
+    port_->DestroyConnectionAsync(this);
   }
 }
 
diff --git a/p2p/base/connection.h b/p2p/base/connection.h
index e0f8ed4874..405a1def90 100644
--- a/p2p/base/connection.h
+++ b/p2p/base/connection.h
@@ -188,11 +188,18 @@ class Connection : public CandidatePairInterface, public sigslot::has_slots<> {
   int receiving_timeout() const;
   void set_receiving_timeout(absl::optional<int> receiving_timeout_ms);
 
-  // Makes the connection go away.
-  void Destroy();
-
-  // Makes the connection go away, in a failed state.
-  void FailAndDestroy();
+  // The preferred way of destroying a `Connection` instance is by calling
+  // the `DestroyConnection` method in `Port`. This method is provided
+  // for some external callers that still need it. Basically just forwards
+  // the call to port_.
+  [[deprecated]] void Destroy();
+
+  // Signals object destruction, releases outstanding references and performs
+  // final logging.
+  // The function will return `true` when shutdown was performed, signals
+  // emitted and outstanding references released. If the function returns
+  // `false`, `Shutdown()` has previously been called.
+  bool Shutdown();
 
   // Prunes the connection and sets its state to STATE_FAILED,
   // It will not be used or send pings although it can still receive packets.
diff --git a/p2p/base/p2p_transport_channel.cc b/p2p/base/p2p_transport_channel.cc
index 1daec120d9..1bdda7b219 100644
--- a/p2p/base/p2p_transport_channel.cc
+++ b/p2p/base/p2p_transport_channel.cc
@@ -222,9 +222,10 @@ P2PTransportChannel::~P2PTransportChannel() {
   TRACE_EVENT0("webrtc", "P2PTransportChannel::~P2PTransportChannel");
   RTC_DCHECK_RUN_ON(network_thread_);
   std::vector<Connection*> copy(connections().begin(), connections().end());
-  for (Connection* con : copy) {
-    con->SignalDestroyed.disconnect(this);
-    con->Destroy();
+  for (Connection* connection : copy) {
+    connection->SignalDestroyed.disconnect(this);
+    ice_controller_->OnConnectionDestroyed(connection);
+    connection->port()->DestroyConnection(connection);
   }
   resolvers_.clear();
 }
@@ -1695,7 +1696,7 @@ void P2PTransportChannel::RemoveConnectionForTest(Connection* connection) {
   RTC_DCHECK(!FindConnection(connection));
   if (selected_connection_ == connection)
     selected_connection_ = nullptr;
-  connection->Destroy();
+  connection->port()->DestroyConnection(connection);
 }
 
 // Monitor connection states.
@@ -2040,7 +2041,7 @@ void P2PTransportChannel::HandleAllTimedOut() {
     }
     connection->SignalDestroyed.disconnect(this);
     ice_controller_->OnConnectionDestroyed(connection);
-    connection->Destroy();
+    connection->port()->DestroyConnection(connection);
   }
 
   if (update_selected_connection)
diff --git a/p2p/base/port.cc b/p2p/base/port.cc
index 3bd09dc024..a8b4e8ab15 100644
--- a/p2p/base/port.cc
+++ b/p2p/base/port.cc
@@ -18,6 +18,7 @@
 #include <vector>
 
 #include "absl/algorithm/container.h"
+#include "absl/memory/memory.h"
 #include "absl/strings/match.h"
 #include "absl/strings/string_view.h"
 #include "p2p/base/connection.h"
@@ -186,22 +187,7 @@ void Port::Construct() {
 Port::~Port() {
   RTC_DCHECK_RUN_ON(thread_);
   CancelPendingTasks();
-
-  // Delete all of the remaining connections.  We copy the list up front
-  // because each deletion will cause it to be modified.
-
-  std::vector<Connection*> list;
-
-  AddressMap::iterator iter = connections_.begin();
-  while (iter != connections_.end()) {
-    list.push_back(iter->second);
-    ++iter;
-  }
-
-  for (uint32_t i = 0; i < list.size(); i++) {
-    list[i]->SignalDestroyed.disconnect(this);
-    delete list[i];
-  }
+  DestroyAllConnections();
 }
 
 const std::string& Port::Type() const {
@@ -352,11 +338,11 @@ void Port::AddOrReplaceConnection(Connection* conn) {
         << ": A new connection was created on an existing remote address. "
            "New remote candidate: "
         << conn->remote_candidate().ToSensitiveString();
-    ret.first->second->SignalDestroyed.disconnect(this);
-    ret.first->second->Destroy();
+    auto old_conn = absl::WrapUnique(ret.first->second);
     ret.first->second = conn;
+    HandleConnectionDestroyed(old_conn.get());
+    old_conn->Shutdown();
   }
-  conn->SignalDestroyed.connect(this, &Port::OnConnectionDestroyed);
 }
 
 void Port::OnReadPacket(const char* data,
@@ -614,9 +600,9 @@ rtc::DiffServCodePoint Port::StunDscpValue() const {
 
 void Port::DestroyAllConnections() {
   RTC_DCHECK_RUN_ON(thread_);
-  for (auto kv : connections_) {
-    kv.second->SignalDestroyed.disconnect(this);
-    kv.second->Destroy();
+  for (auto& [unused, connection] : connections_) {
+    connection->Shutdown();
+    delete connection;
   }
   connections_.clear();
 }
@@ -928,6 +914,24 @@ void Port::OnConnectionDestroyed(Connection* conn) {
   }
 }
 
+void Port::DestroyConnectionInternal(Connection* conn, bool async) {
+  RTC_DCHECK_RUN_ON(thread_);
+  OnConnectionDestroyed(conn);
+  conn->Shutdown();
+  if (async) {
+    // Unwind the stack before deleting the object in case upstream callers
+    // need to refer to the Connection's state as part of teardown.
+    // NOTE: We move ownership of `conn` into the capture section of the lambda
+    // so that the object will always be deleted, including if PostTask fails.
+    // In such a case (only tests), deletion would happen inside of the call
+    // to `DestroyConnection()`.
+    thread_->PostTask(
+        webrtc::ToQueuedTask([conn = absl::WrapUnique(conn)]() {}));
+  } else {
+    delete conn;
+  }
+}
+
 void Port::Destroy() {
   RTC_DCHECK(connections_.empty());
   RTC_LOG(LS_INFO) << ToString() << ": Port deleted";
diff --git a/p2p/base/port.h b/p2p/base/port.h
index b1dab5e92b..9f24b7b3ff 100644
--- a/p2p/base/port.h
+++ b/p2p/base/port.h
@@ -293,6 +293,20 @@ class Port : public PortInterface,
   // Returns the connection to the given address or NULL if none exists.
   Connection* GetConnection(const rtc::SocketAddress& remote_addr) override;
 
+  // Removes and deletes a connection object. By default the connection will
+  // be deleted directly inside the function, but a caller may optionally pass
+  // `async = true` in order to defer the actual `delete` call to happen when
+  // the current call stack has been unwound. This may be needed when a
+  // connection object needs to be deleted but pointers to the object are held
+  // call stack.
+  void DestroyConnection(Connection* conn) {
+    DestroyConnectionInternal(conn, false);
+  }
+
+  void DestroyConnectionAsync(Connection* conn) {
+    DestroyConnectionInternal(conn, true);
+  }
+
   // In a shared socket mode each port which shares the socket will decide
   // to accept the packet based on the `remote_addr`. Currently only UDP
   // port implemented this method.
@@ -452,9 +466,14 @@ class Port : public PortInterface,
 
  private:
   void Construct();
-  // Called when one of our connections deletes itself.
+
+  // Called internally when deleting a connection object.
   void OnConnectionDestroyed(Connection* conn);
 
+  // Private implementation of DestroyConnection to keep the async usage
+  // distinct.
+  void DestroyConnectionInternal(Connection* conn, bool async);
+
   void OnNetworkTypeChanged(const rtc::Network* network);
 
   rtc::Thread* const thread_;
diff --git a/p2p/base/port_unittest.cc b/p2p/base/port_unittest.cc
index 1c73c96407..b9115f51b5 100644
--- a/p2p/base/port_unittest.cc
+++ b/p2p/base/port_unittest.cc
@@ -274,11 +274,7 @@ class TestChannel : public sigslot::has_slots<> {
         [this](PortInterface* port) { OnSrcPortDestroyed(port); });
   }
 
-  ~TestChannel() {
-    if (conn_) {
-      conn_->SignalDestroyed.disconnect(this);
-    }
-  }
+  ~TestChannel() { Stop(); }
 
   int complete_count() { return complete_count_; }
   Connection* conn() { return conn_; }
@@ -287,6 +283,7 @@ class TestChannel : public sigslot::has_slots<> {
 
   void Start() { port_->PrepareAddress(); }
   void CreateConnection(const Candidate& remote_candidate) {
+    RTC_DCHECK(!conn_);
     conn_ = port_->CreateConnection(remote_candidate, Port::ORIGIN_MESSAGE);
     IceMode remote_ice_mode =
         (ice_mode_ == ICEMODE_FULL) ? ICEMODE_LITE : ICEMODE_FULL;
@@ -298,6 +295,7 @@ class TestChannel : public sigslot::has_slots<> {
                                      &TestChannel::OnConnectionReadyToSend);
     connection_ready_to_send_ = false;
   }
+
   void OnConnectionStateChange(Connection* conn) {
     if (conn->write_state() == Connection::STATE_WRITABLE) {
       conn->set_use_candidate_attr(true);
@@ -305,6 +303,10 @@ class TestChannel : public sigslot::has_slots<> {
     }
   }
   void AcceptConnection(const Candidate& remote_candidate) {
+    if (conn_) {
+      conn_->SignalDestroyed.disconnect(this);
+      conn_ = nullptr;
+    }
     ASSERT_TRUE(remote_request_.get() != NULL);
     Candidate c = remote_candidate;
     c.set_address(remote_address_);
@@ -317,8 +319,7 @@ class TestChannel : public sigslot::has_slots<> {
   void Ping(int64_t now) { conn_->Ping(now); }
   void Stop() {
     if (conn_) {
-      conn_->SignalDestroyed.disconnect(this);
-      conn_->Destroy();
+      port_->DestroyConnection(conn_);
       conn_ = nullptr;
     }
   }
@@ -358,7 +359,7 @@ class TestChannel : public sigslot::has_slots<> {
   void OnDestroyed(Connection* conn) {
     ASSERT_EQ(conn_, conn);
     RTC_LOG(LS_INFO) << "OnDestroy connection " << conn << " deleted";
-    conn_ = NULL;
+    conn_ = nullptr;
     // When the connection is destroyed, also clear these fields so future
     // connections are possible.
     remote_request_.reset();
@@ -677,7 +678,7 @@ class PortTest : public ::testing::Test, public sigslot::has_slots<> {
 
     // Speed up destroying ch2's connection such that the test is ready to
     // accept a new connection from ch1 before ch1's connection destroys itself.
-    ch2->conn()->Destroy();
+    ch2->Stop();
     EXPECT_TRUE_WAIT(ch2->conn() == NULL, kDefaultTimeout);
   }
 
diff --git a/p2p/base/tcp_port.cc b/p2p/base/tcp_port.cc
index 2964c8b078..90842e1fb6 100644
--- a/p2p/base/tcp_port.cc
+++ b/p2p/base/tcp_port.cc
@@ -509,9 +509,8 @@ void TCPConnection::OnClose(rtc::AsyncPacketSocket* socket, int error) {
     port()->thread()->PostDelayedTask(
         webrtc::ToQueuedTask(network_safety_,
                              [this]() {
-                               if (pretending_to_be_writable_) {
-                                 Destroy();
-                               }
+                               if (pretending_to_be_writable_)
+                                 port()->DestroyConnection(this);
                              }),
         reconnection_timeout());
   } else if (!pretending_to_be_writable_) {
@@ -520,7 +519,7 @@ void TCPConnection::OnClose(rtc::AsyncPacketSocket* socket, int error) {
     // to manually destroy here as this connection, as never connected, will not
     // be scheduled for ping to trigger destroy.
     socket_->UnsubscribeClose(this);
-    Destroy();
+    port()->DestroyConnectionAsync(this);
   }
 }
 
diff --git a/p2p/base/turn_port.h b/p2p/base/turn_port.h
index 74d249317f..9308c5ed6b 100644
--- a/p2p/base/turn_port.h
+++ b/p2p/base/turn_port.h
@@ -201,12 +201,11 @@ class TurnPort : public Port {
   // Finds the turn entry with `address` and sets its channel id.
   // Returns true if the entry is found.
   bool SetEntryChannelId(const rtc::SocketAddress& address, int channel_id);
-  // Visible for testing.
-  // Shuts down the turn port, usually because of some fatal errors.
-  void Close();
 
   void HandleConnectionDestroyed(Connection* conn) override;
 
+  void CloseForTest() { Close(); }
+
  protected:
   TurnPort(rtc::Thread* thread,
            rtc::PacketSocketFactory* factory,
@@ -249,6 +248,9 @@ class TurnPort : public Port {
 
   rtc::DiffServCodePoint StunDscpValue() const override;
 
+  // Shuts down the turn port, frees requests and deletes connections.
+  void Close();
+
  private:
   enum {
     MSG_ALLOCATE_ERROR = MSG_FIRST_AVAILABLE,
diff --git a/p2p/base/turn_port_unittest.cc b/p2p/base/turn_port_unittest.cc
index 2244b9d78d..496b6a226f 100644
--- a/p2p/base/turn_port_unittest.cc
+++ b/p2p/base/turn_port_unittest.cc
@@ -664,7 +664,8 @@ class TurnPortTest : public ::testing::Test,
 
     // Destroy the connection on the TURN port. The TurnEntry still exists, so
     // the TURN port should still process a ping from an unknown address.
-    conn2->Destroy();
+    turn_port_->DestroyConnection(conn2);
+
     conn1->Ping(0);
     EXPECT_TRUE_SIMULATED_WAIT(turn_unknown_address_, kSimulatedRtt,
                                fake_clock_);
@@ -1268,7 +1269,7 @@ TEST_F(TurnPortTest, TestStopProcessingPacketsAfterClosed) {
   EXPECT_EQ_SIMULATED_WAIT(Connection::STATE_WRITABLE, conn2->write_state(),
                            kSimulatedRtt * 2, fake_clock_);
 
-  turn_port_->Close();
+  turn_port_->CloseForTest();
   SIMULATED_WAIT(false, kSimulatedRtt, fake_clock_);
   turn_unknown_address_ = false;
   conn2->Ping(0);