Removing MessageHandler dependency from Connection.

Bug: webrtc:11988
Change-Id: Ic35bb5baeafbda7210012dceb0d6d5f5b3eb95c9
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/249941
Reviewed-by: Niels Moller <nisse@webrtc.org>
Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#35890}

p2p/base/connection.cc

@@ -18,6 +18,7 @@
 #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"
@@ -833,15 +834,25 @@ void Connection::Prune() {
 
 void Connection::Destroy() {
   RTC_DCHECK_RUN_ON(network_thread_);
-  // TODO(deadbeef, nisse): This may leak if an application closes a
-  // PeerConnection and then quickly destroys the PeerConnectionFactory (along
-  // with the networking thread on which this message is posted). Also affects
-  // tests, with a workaround in
-  // AutoSocketServerThread::~AutoSocketServerThread.
-  RTC_LOG(LS_VERBOSE) << ToString() << ": Connection destroyed";
-  // TODO(bugs.webrtc.org/11988): Use PostTask.
-  port_->thread()->Post(RTC_FROM_HERE, this, MSG_DELETE);
+  RTC_DLOG(LS_VERBOSE) << ToString() << ": Connection destroyed";
+
+  // Fire the 'destroyed' event before deleting the object. This is done
+  // intentionally to avoid a situation whereby the signal might have dangling
+  // pointers to objects that have been deleted by the time the async task
+  // that deletes the connection object runs.
+  SignalDestroyed(this);
+  SignalDestroyed.disconnect_all();
+
   LogCandidatePairConfig(webrtc::IceCandidatePairConfigType::kDestroyed);
+
+  // 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() {
@@ -1422,15 +1433,6 @@ void Connection::MaybeUpdatePeerReflexiveCandidate(
   }
 }
 
-void Connection::OnMessage(rtc::Message* pmsg) {
-  RTC_DCHECK_RUN_ON(network_thread_);
-  RTC_DCHECK(pmsg->message_id == MSG_DELETE);
-  RTC_LOG(LS_INFO) << "Connection deleted with number of pings sent: "
-                   << num_pings_sent_;
-  SignalDestroyed(this);
-  delete this;
-}
-
 int64_t Connection::last_received() const {
   RTC_DCHECK_RUN_ON(network_thread_);
   return std::max(last_data_received_,

p2p/base/connection.h

@@ -70,9 +70,7 @@ class ConnectionRequest : public StunRequest {
 
 // Represents a communication link between a port on the local client and a
 // port on the remote client.
-class Connection : public CandidatePairInterface,
-                   public rtc::MessageHandlerAutoCleanup,
-                   public sigslot::has_slots<> {
+class Connection : public CandidatePairInterface, public sigslot::has_slots<> {
  public:
   struct SentPing {
     SentPing(const std::string id, int64_t sent_time, uint32_t nomination)
@@ -320,8 +318,6 @@ class Connection : public CandidatePairInterface,
   void set_remote_nomination(uint32_t remote_nomination);
 
  protected:
-  enum { MSG_DELETE = 0, MSG_FIRST_AVAILABLE };
-
   // Constructs a new connection to the given remote port.
   Connection(Port* port, size_t index, const Candidate& candidate);
 
@@ -351,8 +347,6 @@ class Connection : public CandidatePairInterface,
   void set_state(IceCandidatePairState state);
   void set_connected(bool value);
 
-  void OnMessage(rtc::Message* pmsg) override;
-
   // The local port where this connection sends and receives packets.
   Port* port() { return port_; }
   const Port* port() const { return port_; }

p2p/base/p2p_transport_channel_unittest.cc

@@ -483,6 +483,9 @@ class P2PTransportChannelTestBase : public ::testing::Test,
     ep2_.cd1_.ch_.reset();
     ep1_.cd2_.ch_.reset();
     ep2_.cd2_.ch_.reset();
+    // Process pending tasks that need to run for cleanup purposes such as
+    // pending deletion of Connection objects (see Connection::Destroy).
+    rtc::Thread::Current()->ProcessMessages(0);
   }
   P2PTransportChannel* ep1_ch1() { return ep1_.cd1_.ch_.get(); }
   P2PTransportChannel* ep1_ch2() { return ep1_.cd2_.ch_.get(); }

p2p/base/tcp_port.cc

@@ -79,6 +79,7 @@
 #include "rtc_base/logging.h"
 #include "rtc_base/net_helper.h"
 #include "rtc_base/rate_tracker.h"
+#include "rtc_base/task_utils/to_queued_task.h"
 #include "rtc_base/third_party/sigslot/sigslot.h"
 
 namespace cricket {
@@ -366,7 +367,9 @@ TCPConnection::TCPConnection(TCPPort* port,
   }
 }
 
-TCPConnection::~TCPConnection() {}
+TCPConnection::~TCPConnection() {
+  RTC_DCHECK_RUN_ON(network_thread_);
+}
 
 int TCPConnection::Send(const void* data,
                         size_t size,
@@ -493,11 +496,20 @@ void TCPConnection::OnClose(rtc::AsyncPacketSocket* socket, int error) {
     // events.
     pretending_to_be_writable_ = true;
 
+    // If this connection can't become connected and writable again in 5
+    // seconds, it's time to tear this down. This is the case for the original
+    // TCP connection on passive side during a reconnect.
     // We don't attempt reconnect right here. This is to avoid a case where the
     // shutdown is intentional and reconnect is not necessary. We only reconnect
     // when the connection is used to Send() or Ping().
-    port()->thread()->PostDelayed(RTC_FROM_HERE, reconnection_timeout(), this,
-                                  MSG_TCPCONNECTION_DELAYED_ONCLOSE);
+    port()->thread()->PostDelayedTask(
+        webrtc::ToQueuedTask(network_safety_,
+                             [this]() {
+                               if (pretending_to_be_writable_) {
+                                 Destroy();
+                               }
+                             }),
+        reconnection_timeout());
   } else if (!pretending_to_be_writable_) {
     // OnClose could be called when the underneath socket times out during the
     // initial connect() (i.e. `pretending_to_be_writable_` is false) . We have
@@ -507,24 +519,6 @@ void TCPConnection::OnClose(rtc::AsyncPacketSocket* socket, int error) {
   }
 }
 
-void TCPConnection::OnMessage(rtc::Message* pmsg) {
-  switch (pmsg->message_id) {
-    case MSG_TCPCONNECTION_DELAYED_ONCLOSE:
-      // If this connection can't become connected and writable again in 5
-      // seconds, it's time to tear this down. This is the case for the original
-      // TCP connection on passive side during a reconnect.
-      if (pretending_to_be_writable_) {
-        Destroy();
-      }
-      break;
-    case MSG_TCPCONNECTION_FAILED_CREATE_SOCKET:
-      FailAndPrune();
-      break;
-    default:
-      Connection::OnMessage(pmsg);
-  }
-}
-
 void TCPConnection::MaybeReconnect() {
   // Only reconnect for an outgoing TCPConnection when OnClose was signaled and
   // no outstanding reconnect is pending.
@@ -576,13 +570,13 @@ void TCPConnection::CreateOutgoingTcpSocket() {
   } else {
     RTC_LOG(LS_WARNING) << ToString() << ": Failed to create connection to "
                         << remote_candidate().address().ToSensitiveString();
+    set_state(IceCandidatePairState::FAILED);
     // We can't FailAndPrune directly here. FailAndPrune and deletes all
     // the StunRequests from the request_map_. And if this is in the stack
     // of Connection::Ping(), we are still using the request.
     // Unwind the stack and defer the FailAndPrune.
-    set_state(IceCandidatePairState::FAILED);
-    port()->thread()->Post(RTC_FROM_HERE, this,
-                           MSG_TCPCONNECTION_FAILED_CREATE_SOCKET);
+    port()->thread()->PostTask(
+        webrtc::ToQueuedTask(network_safety_, [this]() { FailAndPrune(); }));
   }
 }
 

p2p/base/tcp_port.h

@@ -20,6 +20,7 @@
 #include "p2p/base/port.h"
 #include "rtc_base/async_packet_socket.h"
 #include "rtc_base/containers/flat_map.h"
+#include "rtc_base/task_utils/pending_task_safety_flag.h"
 
 namespace cricket {
 
@@ -135,8 +136,6 @@ class TCPConnection : public Connection {
 
   rtc::AsyncPacketSocket* socket() { return socket_.get(); }
 
-  void OnMessage(rtc::Message* pmsg) override;
-
   // Allow test cases to overwrite the default timeout period.
   int reconnection_timeout() const { return reconnection_timeout_; }
   void set_reconnection_timeout(int timeout_in_ms) {
@@ -144,11 +143,6 @@ class TCPConnection : public Connection {
   }
 
  protected:
-  enum {
-    MSG_TCPCONNECTION_DELAYED_ONCLOSE = Connection::MSG_FIRST_AVAILABLE,
-    MSG_TCPCONNECTION_FAILED_CREATE_SOCKET,
-  };
-
   // Set waiting_for_stun_binding_complete_ to false to allow data packets in
   // addition to what Port::OnConnectionRequestResponse does.
   void OnConnectionRequestResponse(ConnectionRequest* req,
@@ -190,6 +184,8 @@ class TCPConnection : public Connection {
   // Allow test case to overwrite the default timeout period.
   int reconnection_timeout_;
 
+  webrtc::ScopedTaskSafety network_safety_;
+
   friend class TCPPort;
 };
 

pc/scenario_tests/goog_cc_test.cc

@@ -100,6 +100,9 @@ TEST(GoogCcPeerScenarioTest, MAYBE_NoBweChangeFromVideoUnmute) {
     s.ProcessMessages(TimeDelta::Seconds(1));
     EXPECT_GE(get_bwe(), initial_bwe);
   }
+
+  caller->pc()->Close();
+  callee->pc()->Close();
 }
 
 }  // namespace test

test/peer_scenario/tests/remote_estimate_test.cc

@@ -102,6 +102,8 @@ TEST(RemoteEstimateEndToEnd, AudioUsesAbsSendTimeExtension) {
         }
       });
   RTC_CHECK(s.WaitAndProcess(&received_abs_send_time));
+  caller->pc()->Close();
+  callee->pc()->Close();
 }
 }  // namespace test
 }  // namespace webrtc

test/peer_scenario/tests/unsignaled_stream_test.cc

@@ -254,6 +254,8 @@ TEST_P(UnsignaledStreamTest, ReplacesUnsignaledStreamOnCompletedSignaling) {
       });
   EXPECT_TRUE(s.WaitAndProcess(&offer_exchange_done));
   EXPECT_TRUE(s.WaitAndProcess(&second_sink.frame_observed_));
+  caller->pc()->Close();
+  callee->pc()->Close();
 }
 
 INSTANTIATE_TEST_SUITE_P(