Asynchronously clear media channels in RtpTransceiver Refactor the channel destruction process to eliminate synchronous blocking calls to the worker and network threads. This replaces direct calls to ClearChannel with batched tasks that handle teardown asynchronously. Bug: webrtc:42222804 Change-Id: If09f94655f37c274955d56219d7252ca097b627c Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/463100 Reviewed-by: Danil Chapovalov <danilchap@webrtc.org> Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47473}
diff --git a/pc/BUILD.gn b/pc/BUILD.gn index 3e96991..45bb726 100644 --- a/pc/BUILD.gn +++ b/pc/BUILD.gn
@@ -1220,6 +1220,7 @@ "../rtc_base/system:plan_b_only", "../system_wrappers:metrics", "//third_party/abseil-cpp/absl/algorithm:container", + "//third_party/abseil-cpp/absl/cleanup", "//third_party/abseil-cpp/absl/functional:any_invocable", "//third_party/abseil-cpp/absl/memory", "//third_party/abseil-cpp/absl/strings",
diff --git a/pc/channel.cc b/pc/channel.cc index 9c6fe64..daf42c4 100644 --- a/pc/channel.cc +++ b/pc/channel.cc
@@ -211,9 +211,8 @@ // Eats any outstanding messages or packets. alive_->SetNotAlive(); - // The media channel is destroyed at the end of the destructor, since it - // is a std::unique_ptr. The transport channel (rtp_transport) must outlive - // the media channel. + // The channel must be disconnected from the transport before destruction. + RTC_DCHECK(rtp_transport_ == nullptr); } std::string BaseChannel::ToString() const {
diff --git a/pc/sdp_offer_answer.cc b/pc/sdp_offer_answer.cc index 3e13b30..7fc773a 100644 --- a/pc/sdp_offer_answer.cc +++ b/pc/sdp_offer_answer.cc
@@ -26,6 +26,7 @@ #include <vector> #include "absl/algorithm/container.h" +#include "absl/cleanup/cleanup.h" #include "absl/functional/any_invocable.h" #include "absl/memory/memory.h" #include "absl/strings/match.h" @@ -3478,7 +3479,16 @@ std::vector<scoped_refptr<MediaStreamInterface>> all_added_streams; std::vector<scoped_refptr<MediaStreamInterface>> all_removed_streams; std::vector<scoped_refptr<RtpReceiverInterface>> removed_receivers; + // Keep to-be-removed transceivers alive until after tasks for them have been + // run. + std::vector<RtpTransceiverProxyRefPtr> transceivers_to_remove; + absl::Cleanup cleanup_remove = [&] { + for (const auto& transceiver : transceivers_to_remove) { + transceivers()->Remove(transceiver); + } + }; ScopedOperationsBatcher worker_tasks(context_->worker_thread()); + ScopedOperationsBatcher network_tasks(context_->network_thread()); for (auto&& transceivers_stable_state_pair : transceivers()->StableStates()) { auto transceiver = transceivers_stable_state_pair.first; @@ -3530,7 +3540,9 @@ // newly created (newly_created) or if remote streams were not set. RTC_DCHECK(transceiver->internal()->mid().has_value()); - transceiver->internal()->ClearChannel(); + network_tasks.Add(transceiver->internal()->GetClearChannelNetworkTask()); + worker_tasks.Add(transceiver->internal()->GetDeleteChannelWorkerTask( + /*stop_senders=*/false)); if (signaling_state() == PeerConnectionInterface::kHaveRemoteOffer && transceiver->receiver()) { @@ -3542,7 +3554,7 @@ } else { worker_tasks.Add( transceiver->internal()->GetStopTransceiverProcedure()); - transceivers()->Remove(transceiver); + transceivers_to_remove.push_back(transceiver); } } auto sender_internal = transceiver->internal()->sender_internal(); @@ -3556,7 +3568,12 @@ transceiver->internal()->set_mline_index(stable_state.mline_index()); } } - RTCError e = transport_controller_s()->RollbackTransports(); + + RTCError e = network_tasks.Run(); + RTC_DCHECK(e.ok()); // only void tasks queued. + e = worker_tasks.Run(); + RTC_DCHECK(e.ok()); // only void tasks queued. + e = transport_controller_s()->RollbackTransports(); if (!e.ok()) { return e; } @@ -4114,8 +4131,15 @@ } } + // Thread safety requires specific execution order for tasks: + // - Construction: Worker tasks before Network tasks. + // (Network depends on Worker setup). + // - Destruction: Network tasks before Worker tasks. + // (Worker cleanup may depend on Network teardown). + // The batchers below are declared and executed to enforce these orders. + ScopedOperationsBatcher network_teardown_tasks(context_->network_thread()); ScopedOperationsBatcher worker_tasks(context_->worker_thread()); - ScopedOperationsBatcher network_tasks(context_->network_thread()); + ScopedOperationsBatcher network_init_tasks(context_->network_thread()); const ContentInfos& new_contents = new_session.description()->contents(); for (size_t i = 0; i < new_contents.size(); ++i) { const ContentInfo& new_content = new_contents[i]; @@ -4151,7 +4175,8 @@ } auto transceiver = transceiver_or_error.MoveValue(); UpdateTransceiverChannel(transceiver, new_content, bundle_group, - worker_tasks, network_tasks); + network_teardown_tasks, worker_tasks, + network_init_tasks); // Handle locally rejected content. This code path is only needed for apps // that SDP munge. Remote rejected content is handled in // ApplyRemoteDescriptionUpdateTransceiverState(). @@ -4204,11 +4229,11 @@ } } - RTCError error = worker_tasks.Run(); - if (!error.ok()) { - return error; - } - return network_tasks.Run(); + RTCError error = network_teardown_tasks.Run(); + RTC_DCHECK(error.ok()); // Teardown tasks cannot fail. + error = worker_tasks.Run(); + RTC_DCHECK(error.ok()); // Cleanup and construction tasks cannot fail. + return network_init_tasks.Run(); } RTCErrorOr<scoped_refptr<RtpTransceiverProxyWithInternal<RtpTransceiver>>> @@ -4350,14 +4375,18 @@ scoped_refptr<RtpTransceiverProxyWithInternal<RtpTransceiver>> transceiver, const ContentInfo& content, const ContentGroup* bundle_group, + ScopedOperationsBatcher& network_teardown_tasks, ScopedOperationsBatcher& worker_tasks, - ScopedOperationsBatcher& network_tasks) { + ScopedOperationsBatcher& network_init_tasks) { TRACE_EVENT0("webrtc", "SdpOfferAnswerHandler::UpdateTransceiverChannel"); RTC_DCHECK(IsUnifiedPlan()); RTC_DCHECK(transceiver); if (content.rejected) { if (transceiver->internal()->HasChannel()) { - transceiver->internal()->ClearChannel(); + network_teardown_tasks.Add( + transceiver->internal()->GetClearChannelNetworkTask()); + worker_tasks.Add(transceiver->internal()->GetDeleteChannelWorkerTask( + /*stop_senders=*/false)); } } else { if (!transceiver->internal()->HasChannel()) { @@ -4369,7 +4398,7 @@ RTC_DCHECK_RUN_ON(network_thread()); return transport_controller_n()->GetRtpTransport(mid); }, - worker_tasks, network_tasks); + worker_tasks, network_init_tasks); } } } @@ -5474,6 +5503,8 @@ void SdpOfferAnswerHandler::RemoveUnusedChannels( const SessionDescription* desc) { RTC_DCHECK_RUN_ON(signaling_thread()); + RTC_DCHECK(!IsUnifiedPlan()); + if (ConfiguredForMedia()) { // Destroy video channel first since it may have a pointer to the // voice channel.
diff --git a/pc/sdp_offer_answer.h b/pc/sdp_offer_answer.h index e2d2b6f..98f26b2 100644 --- a/pc/sdp_offer_answer.h +++ b/pc/sdp_offer_answer.h
@@ -391,8 +391,10 @@ transceiver, const ContentInfo& content, const ContentGroup* bundle_group, + ScopedOperationsBatcher& network_teardown_tasks, ScopedOperationsBatcher& worker_tasks, - ScopedOperationsBatcher& network_tasks) RTC_RUN_ON(signaling_thread()); + ScopedOperationsBatcher& network_init_tasks) + RTC_RUN_ON(signaling_thread()); // Either creates or destroys the local data channel according to the given // media section.