Replace task vectors with ScopedOperationsBatcher Migrate task management in PeerConnection, RTCStatsCollector, and SdpOfferAnswerHandler from std::vector to ScopedOperationsBatcher. This simplifies collection and execution of cross-thread operations during shutdown and teardown. Key modifications include: - Updating CloseOnNetworkThread to return a BatchTaskWithFinalizer, allowing for coordinated cleanup between the network and signaling threads. - Refactoring the signatures of GetMediaChannelTeardownTasks and CancelPendingRequestAndGetShutdownTasks to use batcher references. - Removing rigid thread block count assertions in the PeerConnection destructor to allow for more flexible task yielding. - Simplifying task execution logic in unit tests. Bug: webrtc:42222804 Change-Id: Ifdcfba5f0bf9680eb78bd3e8b4888e71b873e6ff Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/461641 Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Reviewed-by: Harald Alvestrand <hta@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47334}
diff --git a/pc/BUILD.gn b/pc/BUILD.gn index 322488a..a603dee 100644 --- a/pc/BUILD.gn +++ b/pc/BUILD.gn
@@ -1016,6 +1016,7 @@ ":rtp_sender", ":rtp_sender_proxy", ":rtp_transceiver", + ":scoped_operations_batcher", ":track_media_info_map", ":transport_stats", "../api:candidate", @@ -1293,6 +1294,7 @@ ":rtp_transceiver", ":rtp_transmission_manager", ":rtp_transport_internal", + ":scoped_operations_batcher", ":sctp_data_channel", ":sctp_transport", ":sdp_offer_answer",
diff --git a/pc/peer_connection.cc b/pc/peer_connection.cc index 75b0b0e..cbc6b6d 100644 --- a/pc/peer_connection.cc +++ b/pc/peer_connection.cc
@@ -97,6 +97,7 @@ #include "pc/rtp_transceiver.h" #include "pc/rtp_transmission_manager.h" #include "pc/rtp_transport_internal.h" +#include "pc/scoped_operations_batcher.h" #include "pc/sctp_data_channel.h" #include "pc/sctp_transport.h" #include "pc/sdp_offer_answer.h" @@ -677,7 +678,6 @@ PeerConnection::~PeerConnection() { TRACE_EVENT0("webrtc", "PeerConnection::~PeerConnection"); RTC_DCHECK_RUN_ON(signaling_thread()); - RTC_LOG_THREAD_BLOCK_COUNT(); sdp_handler_->PrepareForShutdown(); @@ -685,8 +685,8 @@ // potentially pending operations. data_channel_controller_.PrepareForShutdown(); - std::vector<absl::AnyInvocable<void() &&>> network_tasks; - std::vector<absl::AnyInvocable<void() &&>> worker_tasks; + ScopedOperationsBatcher network_tasks(network_thread()); + ScopedOperationsBatcher worker_tasks(worker_thread()); // Stop transceivers before destroying the stats collector because // AudioRtpSender has a reference to the LegacyStatsCollector that it will @@ -697,30 +697,24 @@ legacy_stats_.reset(nullptr); stats_collector_.CancelPendingRequestAndGetShutdownTasks(network_tasks, worker_tasks); - - CloseOnNetworkThread(network_tasks); + network_tasks.AddWithFinalizer(MakeCloseOnNetworkThreadTask()); // call_ must be destroyed on the worker thread. - worker_thread()->BlockingCall([&] { + worker_tasks.Add([this]() { RTC_DCHECK_RUN_ON(worker_thread()); - for (auto& task : worker_tasks) { - std::move(task)(); - task = nullptr; - } worker_thread_safety_->SetNotAlive(); call_.reset(); media_engine_ref_.reset(); }); + network_tasks.Run(); + worker_tasks.Run(); + if (sdp_handler_) { sdp_handler_->ResetSessionDescFactory(); } data_channel_controller_.PrepareForShutdown(); - - // The expectation is that there will have been 1 blocking call for the worker - // thread and optionally 1 task for the network thread. - RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(2); } JsepTransportController* PeerConnection::InitializeNetworkThread( @@ -854,36 +848,38 @@ }); } -void PeerConnection::CloseOnNetworkThread( - std::vector<absl::AnyInvocable<void() &&>>& network_tasks) { +ScopedOperationsBatcher::BatchTaskWithFinalizer +PeerConnection::MakeCloseOnNetworkThreadTask() { RTC_DCHECK_RUN_ON(signaling_thread()); - if (transport_controller_copy_ || !network_tasks.empty()) { - network_thread()->BlockingCall([&] { - RTC_DCHECK_RUN_ON(network_thread()); - for (auto& task : network_tasks) { - std::move(task)(); - task = nullptr; - } - if (network_thread_safety_->alive()) { - // port_allocator_ and transport_controller_ live on the network thread - // and must be destroyed there. - TeardownDataChannelTransport_n(RTCError::OK()); - port_allocator_->DiscardCandidatePool(); - transport_controller_.reset(); - port_allocator_.reset(); - network_thread_safety_->SetNotAlive(); - } - }); + + if (!transport_controller_copy_) { + return nullptr; } - if (transport_controller_copy_) { - transport_controller_copy_ = nullptr; - sctp_mid_s_.reset(); - SetSctpTransportName(""); - } else { - RTC_DCHECK(!sctp_mid_s_); - RTC_DCHECK(sctp_transport_name_s_.empty()); - } + return [this]() -> RTCErrorOr<ScopedOperationsBatcher::FinalizerTask> { + RTC_DCHECK_RUN_ON(network_thread()); + if (network_thread_safety_->alive()) { + // port_allocator_ and transport_controller_ live on the network thread + // and must be destroyed there. + TeardownDataChannelTransport_n(RTCError::OK()); + port_allocator_->DiscardCandidatePool(); + transport_controller_.reset(); + port_allocator_.reset(); + network_thread_safety_->SetNotAlive(); + } + + return ScopedOperationsBatcher::FinalizerTask([this]() { + RTC_DCHECK_RUN_ON(signaling_thread()); + if (transport_controller_copy_) { + transport_controller_copy_ = nullptr; + sctp_mid_s_.reset(); + SetSctpTransportName(""); + } else { + RTC_DCHECK(!sctp_mid_s_); + RTC_DCHECK(sctp_transport_name_s_.empty()); + } + }); + }; } JsepTransportController* PeerConnection::InitializeTransportController_n( @@ -1942,10 +1938,12 @@ // worker thread (see `PushNewMediaChannelAndDeleteChannel`) and then // eventually freed on the signaling thread. // It would be good to combine those steps with the teardown steps here. - std::vector<absl::AnyInvocable<void() &&>> network_tasks; - std::vector<absl::AnyInvocable<void() &&>> worker_tasks; - sdp_handler_->GetMediaChannelTeardownTasks(network_tasks, worker_tasks); - CloseOnNetworkThread(network_tasks); + ScopedOperationsBatcher worker_tasks(worker_thread()); + { + ScopedOperationsBatcher network_tasks(network_thread()); + sdp_handler_->GetMediaChannelTeardownTasks(network_tasks, worker_tasks); + network_tasks.AddWithFinalizer(MakeCloseOnNetworkThreadTask()); + } // The event log is used in the transport controller, which must be outlived // by the former. CreateOffer by the peer connection is implemented @@ -1957,16 +1955,14 @@ rtp_manager_->Close(); } - worker_thread()->BlockingCall([&] { + worker_tasks.Add([this]() { RTC_DCHECK_RUN_ON(worker_thread()); - for (auto& task : worker_tasks) { - std::move(task)(); - task = nullptr; - } worker_thread_safety_->SetNotAlive(); call_.reset(); StopRtcEventLog_w(); }); + + worker_tasks.Run(); ReportUsagePattern(); ReportCloseUsageMetrics();
diff --git a/pc/peer_connection.h b/pc/peer_connection.h index 15743e2..b48c82b 100644 --- a/pc/peer_connection.h +++ b/pc/peer_connection.h
@@ -82,6 +82,7 @@ #include "pc/rtp_transceiver.h" #include "pc/rtp_transmission_manager.h" #include "pc/rtp_transport_internal.h" +#include "pc/scoped_operations_batcher.h" #include "pc/sdp_offer_answer.h" #include "pc/sdp_state_provider.h" #include "pc/session_description.h" @@ -100,6 +101,8 @@ namespace webrtc { +class ScopedOperationsBatcher; + // PeerConnection is the implementation of the PeerConnection object as defined // by the PeerConnectionInterface API surface. // The class currently is solely responsible for the following: @@ -490,8 +493,8 @@ JsepTransportController* InitializeNetworkThread( const ServerAddresses& stun_servers, const std::vector<RelayServerConfig>& turn_servers); - void CloseOnNetworkThread( - std::vector<absl::AnyInvocable<void() &&>>& network_tasks); + ScopedOperationsBatcher::BatchTaskWithFinalizer + MakeCloseOnNetworkThreadTask(); JsepTransportController* InitializeTransportController_n( std::unique_ptr<JsepTransportController> controller, const RTCConfiguration& configuration) RTC_RUN_ON(network_thread());
diff --git a/pc/rtc_stats_collector.cc b/pc/rtc_stats_collector.cc index ce8c2b3..598677b 100644 --- a/pc/rtc_stats_collector.cc +++ b/pc/rtc_stats_collector.cc
@@ -63,6 +63,7 @@ #include "pc/rtp_receiver_proxy.h" #include "pc/rtp_sender_proxy.h" #include "pc/rtp_transceiver.h" +#include "pc/scoped_operations_batcher.h" #include "pc/track_media_info_map.h" #include "pc/transport_stats.h" #include "rtc_base/checks.h" @@ -1412,12 +1413,12 @@ } void RTCStatsCollector::CancelPendingRequestAndGetShutdownTasks( - std::vector<absl::AnyInvocable<void() &&>>& network_tasks, - std::vector<absl::AnyInvocable<void() &&>>& worker_tasks) { + ScopedOperationsBatcher& network_tasks, + ScopedOperationsBatcher& worker_tasks) { RTC_DCHECK_RUN_ON(signaling_thread_); signaling_safety_->SetNotAlive(); - worker_tasks.push_back([flag = worker_safety_]() { flag->SetNotAlive(); }); - network_tasks.push_back([flag = network_safety_]() { flag->SetNotAlive(); }); + worker_tasks.Add([flag = worker_safety_]() { flag->SetNotAlive(); }); + network_tasks.Add([flag = network_safety_]() { flag->SetNotAlive(); }); } void RTCStatsCollector::ProducePartialResultsOnSignalingThread(
diff --git a/pc/rtc_stats_collector.h b/pc/rtc_stats_collector.h index f20a67d..6dd5795 100644 --- a/pc/rtc_stats_collector.h +++ b/pc/rtc_stats_collector.h
@@ -54,6 +54,7 @@ class RtpSenderInternal; class RtpReceiverInternal; +class ScopedOperationsBatcher; // Structure for tracking stats about each RtpTransceiver managed by the // PeerConnection. This can either by a Plan B style or Unified Plan style @@ -123,8 +124,8 @@ // on the worker and network threads before the RTCStatsCollector instance is // deleted. void CancelPendingRequestAndGetShutdownTasks( - std::vector<absl::AnyInvocable<void() &&>>& network_tasks, - std::vector<absl::AnyInvocable<void() &&>>& worker_tasks); + ScopedOperationsBatcher& network_tasks, + ScopedOperationsBatcher& worker_tasks); // Called by the PeerConnection instance when data channel states change. void OnSctpDataChannelStateChanged(int channel_id,
diff --git a/pc/rtc_stats_collector_unittest.cc b/pc/rtc_stats_collector_unittest.cc index 18b4df2..4aa3782 100644 --- a/pc/rtc_stats_collector_unittest.cc +++ b/pc/rtc_stats_collector_unittest.cc
@@ -69,6 +69,7 @@ #include "pc/peer_connection_internal.h" #include "pc/rtp_sender.h" #include "pc/rtp_transceiver.h" +#include "pc/scoped_operations_batcher.h" #include "pc/sctp_data_channel.h" #include "pc/stream_collection.h" #include "pc/test/fake_audio_track.h" @@ -4229,8 +4230,8 @@ // this posts a task to the worker/network threads which will be blocked by // the above task. wrapper.stats_collector().GetStatsReport(callback); - std::vector<absl::AnyInvocable<void() &&>> network_tasks; - std::vector<absl::AnyInvocable<void() &&>> worker_tasks; + ScopedOperationsBatcher network_tasks(worker_and_network.get()); + ScopedOperationsBatcher worker_tasks(worker_and_network.get()); // Now cancel any ongoing stats gathering operations. // This should cancel the outstanding operations and invoke pending callbacks @@ -4238,21 +4239,12 @@ wrapper.stats_collector().CancelPendingRequestAndGetShutdownTasks( network_tasks, worker_tasks); - // We should have one callback per conceptual thread. - EXPECT_EQ(network_tasks.size(), 1u); - EXPECT_EQ(worker_tasks.size(), 1u); - // Resume the network and worker threads. blocker.Set(); // Run the cleanup tasks. - auto quit = loop.QuitClosure(); - worker_and_network->PostTask([&]() { - std::move(network_tasks[0])(); - std::move(worker_tasks[0])(); - loop.task_queue()->PostTask([&]() { quit(); }); - }); - loop.Run(); + network_tasks.Run(); + worker_tasks.Run(); } // This covers the following steps: @@ -4269,16 +4261,14 @@ EXPECT_CALL(*callback, OnStatsDelivered(_)).Times(0); // Start by canceling any ongoing tasks. There aren't actually any ongoing // tasks, but this gives us the network cleanup task. - std::vector<absl::AnyInvocable<void() &&>> network_tasks; - std::vector<absl::AnyInvocable<void() &&>> worker_tasks; + ScopedOperationsBatcher network_tasks(pc->network_thread()); + ScopedOperationsBatcher worker_tasks(pc->worker_thread()); wrapper.stats_collector().CancelPendingRequestAndGetShutdownTasks( network_tasks, worker_tasks); // Clean up the state on the network thread. This will have the effect of // dropping any tasks targeting the network thread. - ASSERT_EQ(network_tasks.size(), 1u); - ASSERT_EQ(worker_tasks.size(), 1u); - std::move(network_tasks[0])(); - std::move(worker_tasks[0])(); + network_tasks.Run(); + worker_tasks.Run(); // Now, attempt to get a stats report. This will try to post a task to the // network thread, which will be dropped. wrapper.stats_collector().GetStatsReport(callback);
diff --git a/pc/sdp_offer_answer.cc b/pc/sdp_offer_answer.cc index 7a93480..ac88e6f 100644 --- a/pc/sdp_offer_answer.cc +++ b/pc/sdp_offer_answer.cc
@@ -5660,8 +5660,8 @@ } void SdpOfferAnswerHandler::GetMediaChannelTeardownTasks( - std::vector<absl::AnyInvocable<void() &&>>& network_tasks, - std::vector<absl::AnyInvocable<void() &&>>& worker_tasks) { + ScopedOperationsBatcher& network_tasks, + ScopedOperationsBatcher& worker_tasks) { RTC_DCHECK_RUN_ON(signaling_thread()); RTC_DCHECK_DISALLOW_THREAD_BLOCKING_CALLS(); if (!transceivers()) { @@ -5670,21 +5670,16 @@ auto list = transceivers()->List(); for (const auto& transceiver : list) { if (transceiver->media_type() == MediaType::VIDEO) { - if (auto task = transceiver->internal()->GetClearChannelNetworkTask()) - network_tasks.push_back(std::move(task)); - if (auto task = transceiver->internal()->GetDeleteChannelWorkerTask( - /*stop_senders=*/true)) { - worker_tasks.push_back(std::move(task)); - } + network_tasks.Add(transceiver->internal()->GetClearChannelNetworkTask()); + worker_tasks.Add(transceiver->internal()->GetDeleteChannelWorkerTask( + /*stop_senders=*/true)); } } for (const auto& transceiver : list) { if (transceiver->media_type() == MediaType::AUDIO) { - if (auto task = transceiver->internal()->GetClearChannelNetworkTask()) - network_tasks.push_back(std::move(task)); - if (auto task = transceiver->internal()->GetDeleteChannelWorkerTask( - /*stop_senders=*/true)) - worker_tasks.push_back(std::move(task)); + network_tasks.Add(transceiver->internal()->GetClearChannelNetworkTask()); + worker_tasks.Add(transceiver->internal()->GetDeleteChannelWorkerTask( + /*stop_senders=*/true)); } } }
diff --git a/pc/sdp_offer_answer.h b/pc/sdp_offer_answer.h index 802f518..a2b9ef1 100644 --- a/pc/sdp_offer_answer.h +++ b/pc/sdp_offer_answer.h
@@ -20,7 +20,6 @@ #include <string> #include <vector> -#include "absl/functional/any_invocable.h" #include "absl/strings/string_view.h" #include "api/audio_options.h" #include "api/candidate.h" @@ -53,6 +52,7 @@ #include "pc/rtp_receiver.h" #include "pc/rtp_transceiver.h" #include "pc/rtp_transmission_manager.h" +#include "pc/scoped_operations_batcher.h" #include "pc/sdp_payload_type_suggester.h" #include "pc/sdp_state_provider.h" #include "pc/session_description.h" @@ -191,9 +191,8 @@ // belongs to the network and worker threads. // The caller is responsible for invoking the callbacks on the correct threads // in the order 1st network thread, 2nd worker thread. - void GetMediaChannelTeardownTasks( - std::vector<absl::AnyInvocable<void() &&>>& network_tasks, - std::vector<absl::AnyInvocable<void() &&>>& worker_tasks); + void GetMediaChannelTeardownTasks(ScopedOperationsBatcher& network_tasks, + ScopedOperationsBatcher& worker_tasks); PLAN_B_ONLY scoped_refptr<StreamCollectionInterface> local_streams(); PLAN_B_ONLY scoped_refptr<StreamCollectionInterface> remote_streams();