From f15189dbce12acd94c0703ff5b8a7b8aa9f1f167 Mon Sep 17 00:00:00 2001 From: Tomas Gunnarsson Date: Wed, 13 Apr 2022 09:03:52 +0000 Subject: [PATCH] [CallbackList] Use CallbackList in AsyncPacketSocket for close events. This removes use of the SignalClose sigslot. This CL includes thread checks for the callback list and updates some call sites to unsubscribe from events before deletion or detaching from a socket instance. Bug: webrtc:11943 Change-Id: Ib66d39aa5cc795b750c9e3eaa85ed6af8b55b2b5 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/258561 Reviewed-by: Niels Moller Commit-Queue: Tomas Gunnarsson Cr-Commit-Position: refs/heads/main@{#36540} --- p2p/base/port_unittest.cc | 6 +++--- p2p/base/tcp_port.cc | 9 ++++++++- p2p/base/turn_port.cc | 11 ++++++++++- p2p/base/turn_port_unittest.cc | 2 +- p2p/base/turn_server.cc | 6 +++++- rtc_base/BUILD.gn | 1 + rtc_base/async_packet_socket.cc | 16 +++++++++++++++- rtc_base/async_packet_socket.h | 30 +++++++++++++++++++++++++++--- rtc_base/async_tcp_socket.cc | 3 +-- rtc_base/test_echo_server.h | 3 ++- 10 files changed, 73 insertions(+), 14 deletions(-) diff --git a/p2p/base/port_unittest.cc b/p2p/base/port_unittest.cc index 81d5cfeda5..1c73c96407 100644 --- a/p2p/base/port_unittest.cc +++ b/p2p/base/port_unittest.cc @@ -672,8 +672,8 @@ class PortTest : public ::testing::Test, public sigslot::has_slots<> { // Ensure redundant SignalClose events on TcpConnection won't break tcp // reconnection. Chromium will fire SignalClose for all outstanding IPC // packets during reconnection. - tcp_conn1->socket()->SignalClose(tcp_conn1->socket(), 0); - tcp_conn2->socket()->SignalClose(tcp_conn2->socket(), 0); + tcp_conn1->socket()->NotifyClosedForTest(0); + tcp_conn2->socket()->NotifyClosedForTest(0); // 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. @@ -1625,7 +1625,7 @@ TEST_F(PortTest, TestDisableInterfaceOfTcpPort) { lconn->Ping(0); // Now disconnect the client socket... - socket->SignalClose(socket, 1); + socket->NotifyClosedForTest(1); // And prevent new sockets from being created. socket_factory.set_next_client_tcp_socket(nullptr); diff --git a/p2p/base/tcp_port.cc b/p2p/base/tcp_port.cc index b9d8e8572c..2964c8b078 100644 --- a/p2p/base/tcp_port.cc +++ b/p2p/base/tcp_port.cc @@ -519,6 +519,7 @@ void TCPConnection::OnClose(rtc::AsyncPacketSocket* socket, int error) { // initial connect() (i.e. `pretending_to_be_writable_` is false) . We have // to manually destroy here as this connection, as never connected, will not // be scheduled for ping to trigger destroy. + socket_->UnsubscribeClose(this); Destroy(); } } @@ -557,6 +558,11 @@ void TCPConnection::CreateOutgoingTcpSocket() { int opts = (remote_candidate().protocol() == SSLTCP_PROTOCOL_NAME) ? rtc::PacketSocketFactory::OPT_TLS_FAKE : 0; + + if (socket_) { + socket_->UnsubscribeClose(this); + } + rtc::PacketSocketTcpOptions tcp_opts; tcp_opts.opts = opts; socket_.reset(port()->socket_factory()->CreateClientTcpSocket( @@ -590,7 +596,8 @@ void TCPConnection::ConnectSocketSignals(rtc::AsyncPacketSocket* socket) { } socket->SignalReadPacket.connect(this, &TCPConnection::OnReadPacket); socket->SignalReadyToSend.connect(this, &TCPConnection::OnReadyToSend); - socket->SignalClose.connect(this, &TCPConnection::OnClose); + socket->SubscribeClose( + this, [this](rtc::AsyncPacketSocket* s, int err) { OnClose(s, err); }); } } // namespace cricket diff --git a/p2p/base/turn_port.cc b/p2p/base/turn_port.cc index e15f35211b..31b7272972 100644 --- a/p2p/base/turn_port.cc +++ b/p2p/base/turn_port.cc @@ -306,6 +306,10 @@ TurnPort::~TurnPort() { while (!entries_.empty()) { DestroyEntry(entries_.front()); } + + if (socket_) + socket_->UnsubscribeClose(this); + if (!SharedSocket()) { delete socket_; } @@ -451,7 +455,9 @@ bool TurnPort::CreateTurnClientSocket() { if (server_address_.proto == PROTO_TCP || server_address_.proto == PROTO_TLS) { socket_->SignalConnect.connect(this, &TurnPort::OnSocketConnect); - socket_->SignalClose.connect(this, &TurnPort::OnSocketClose); + socket_->SubscribeClose(this, [this](rtc::AsyncPacketSocket* s, int err) { + OnSocketClose(s, err); + }); } else { state_ = STATE_CONNECTED; } @@ -542,6 +548,9 @@ void TurnPort::OnAllocateMismatch() { << ": Allocating a new socket after " "STUN_ERROR_ALLOCATION_MISMATCH, retry: " << allocate_mismatch_retries_ + 1; + + socket_->UnsubscribeClose(this); + if (SharedSocket()) { ResetSharedSocket(); } else { diff --git a/p2p/base/turn_port_unittest.cc b/p2p/base/turn_port_unittest.cc index d1b911f837..2244b9d78d 100644 --- a/p2p/base/turn_port_unittest.cc +++ b/p2p/base/turn_port_unittest.cc @@ -1305,7 +1305,7 @@ TEST_F(TurnPortTest, TestSocketCloseWillDestroyConnection) { Port::ORIGIN_MESSAGE); EXPECT_NE(nullptr, conn); EXPECT_TRUE(!turn_port_->connections().empty()); - turn_port_->socket()->SignalClose(turn_port_->socket(), 1); + turn_port_->socket()->NotifyClosedForTest(1); EXPECT_TRUE_SIMULATED_WAIT(turn_port_->connections().empty(), kConnectionDestructionDelay, fake_clock_); } diff --git a/p2p/base/turn_server.cc b/p2p/base/turn_server.cc index 6a5d7a9a3f..85baf75819 100644 --- a/p2p/base/turn_server.cc +++ b/p2p/base/turn_server.cc @@ -195,7 +195,10 @@ void TurnServer::AcceptConnection(rtc::Socket* server_socket) { cricket::AsyncStunTCPSocket* tcp_socket = new cricket::AsyncStunTCPSocket(accepted_socket); - tcp_socket->SignalClose.connect(this, &TurnServer::OnInternalSocketClose); + tcp_socket->SubscribeClose(this, + [this](rtc::AsyncPacketSocket* s, int err) { + OnInternalSocketClose(s, err); + }); // Finally add the socket so it can start communicating with the client. AddInternalSocket(tcp_socket, info.proto); } @@ -564,6 +567,7 @@ void TurnServer::DestroyInternalSocket(rtc::AsyncPacketSocket* socket) { InternalSocketMap::iterator iter = server_sockets_.find(socket); if (iter != server_sockets_.end()) { rtc::AsyncPacketSocket* socket = iter->first; + socket->UnsubscribeClose(this); socket->SignalReadPacket.disconnect(this); server_sockets_.erase(iter); std::unique_ptr socket_to_delete = diff --git a/rtc_base/BUILD.gn b/rtc_base/BUILD.gn index 5b3776a3a8..bf39ebe927 100644 --- a/rtc_base/BUILD.gn +++ b/rtc_base/BUILD.gn @@ -935,6 +935,7 @@ rtc_library("rtc_base") { deps = [ ":async_resolver_interface", ":async_socket", + ":callback_list", ":checks", ":ip_address", ":logging", diff --git a/rtc_base/async_packet_socket.cc b/rtc_base/async_packet_socket.cc index d5435d71d0..1ce0d3b122 100644 --- a/rtc_base/async_packet_socket.cc +++ b/rtc_base/async_packet_socket.cc @@ -24,10 +24,24 @@ PacketOptions::PacketOptions(DiffServCodePoint dscp) : dscp(dscp) {} PacketOptions::PacketOptions(const PacketOptions& other) = default; PacketOptions::~PacketOptions() = default; -AsyncPacketSocket::AsyncPacketSocket() = default; +AsyncPacketSocket::AsyncPacketSocket() { + network_checker_.Detach(); +} AsyncPacketSocket::~AsyncPacketSocket() = default; +void AsyncPacketSocket::SubscribeClose( + const void* removal_tag, + std::function callback) { + RTC_DCHECK_RUN_ON(&network_checker_); + on_close_.AddReceiver(removal_tag, std::move(callback)); +} + +void AsyncPacketSocket::UnsubscribeClose(const void* removal_tag) { + RTC_DCHECK_RUN_ON(&network_checker_); + on_close_.RemoveReceivers(removal_tag); +} + void CopySocketInformationToPacketInfo(size_t packet_size_bytes, const AsyncPacketSocket& socket_from, bool is_connectionless, diff --git a/rtc_base/async_packet_socket.h b/rtc_base/async_packet_socket.h index 2e334ec36d..aa31e25eab 100644 --- a/rtc_base/async_packet_socket.h +++ b/rtc_base/async_packet_socket.h @@ -13,9 +13,12 @@ #include +#include "api/sequence_checker.h" +#include "rtc_base/callback_list.h" #include "rtc_base/dscp.h" #include "rtc_base/network/sent_packet.h" #include "rtc_base/socket.h" +#include "rtc_base/system/no_unique_address.h" #include "rtc_base/system/rtc_export.h" #include "rtc_base/third_party/sigslot/sigslot.h" #include "rtc_base/time_utils.h" @@ -100,6 +103,11 @@ class RTC_EXPORT AsyncPacketSocket : public sigslot::has_slots<> { virtual int GetError() const = 0; virtual void SetError(int error) = 0; + // Register a callback to be called when the socket is closed. + void SubscribeClose(const void* removal_tag, + std::function callback); + void UnsubscribeClose(const void* removal_tag); + // Emitted each time a packet is read. Used only for UDP and // connected TCP sockets. sigslot::signal5 { // CONNECTING to CONNECTED. sigslot::signal1 SignalConnect; - // Emitted for client TCP sockets when state is changed from - // CONNECTED to CLOSED. - sigslot::signal2 SignalClose; + void NotifyClosedForTest(int err) { NotifyClosed(err); } + + protected: + // TODO(bugs.webrtc.org/11943): Remove after updating downstream code. + void SignalClose(AsyncPacketSocket* s, int err) { + RTC_DCHECK_EQ(s, this); + NotifyClosed(err); + } + + void NotifyClosed(int err) { + RTC_DCHECK_RUN_ON(&network_checker_); + on_close_.Send(this, err); + } + + RTC_NO_UNIQUE_ADDRESS webrtc::SequenceChecker network_checker_; + + private: + webrtc::CallbackList on_close_ + RTC_GUARDED_BY(&network_checker_); }; // Listen socket, producing an AsyncPacketSocket when a peer connects. diff --git a/rtc_base/async_tcp_socket.cc b/rtc_base/async_tcp_socket.cc index c2480755b4..d29eafddb9 100644 --- a/rtc_base/async_tcp_socket.cc +++ b/rtc_base/async_tcp_socket.cc @@ -68,7 +68,6 @@ AsyncTCPSocketBase::AsyncTCPSocketBase(Socket* socket, max_outsize_(max_packet_size) { inbuf_.EnsureCapacity(kMinimumRecvSize); - RTC_DCHECK(socket_.get() != nullptr); socket_->SignalConnectEvent.connect(this, &AsyncTCPSocketBase::OnConnectEvent); socket_->SignalReadEvent.connect(this, &AsyncTCPSocketBase::OnReadEvent); @@ -237,7 +236,7 @@ void AsyncTCPSocketBase::OnWriteEvent(Socket* socket) { } void AsyncTCPSocketBase::OnCloseEvent(Socket* socket, int error) { - SignalClose(this, error); + NotifyClosed(error); } // AsyncTCPSocket diff --git a/rtc_base/test_echo_server.h b/rtc_base/test_echo_server.h index ba5f997287..e4f70caee2 100644 --- a/rtc_base/test_echo_server.h +++ b/rtc_base/test_echo_server.h @@ -45,7 +45,8 @@ class TestEchoServer : public sigslot::has_slots<> { if (raw_socket) { AsyncTCPSocket* packet_socket = new AsyncTCPSocket(raw_socket); packet_socket->SignalReadPacket.connect(this, &TestEchoServer::OnPacket); - packet_socket->SignalClose.connect(this, &TestEchoServer::OnClose); + packet_socket->SubscribeClose( + this, [this](AsyncPacketSocket* s, int err) { OnClose(s, err); }); client_sockets_.push_back(packet_socket); } }