Batch up blocking calls when creating media channels in RtpTransceiver Refactor RtpTransceiver to support media channel and sender creation asynchronously on the worker thread. Construction still can't be considered complete until all the same steps have completed. This change batches up multiple blocking calls into a single one to the worker thread during transceiver initialization. Key modifications include: * Utilizing ScopedOperationsBatcher to gather channel and sender initialization tasks on the worker thread. * Splitting transceiver setup into a worker thread task (for media creation) and a finalizer task on the signaling thread (to update internal state). * Updating SdpOfferAnswerHandler and RtpTransmissionManager to pass the batcher through the transceiver association and creation workflows. * Ensuring all transceivers are fully constructed before proceeding with channel updates during SDP negotiation. Bug: webrtc:42222117 Change-Id: I572a4b5cf474df583a3e4dd9328d57bb1d51cab5 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/463900 Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Reviewed-by: Harald Alvestrand <hta@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47491}
diff --git a/pc/BUILD.gn b/pc/BUILD.gn index 41d66c3..c001c2d 100644 --- a/pc/BUILD.gn +++ b/pc/BUILD.gn
@@ -1767,6 +1767,7 @@ ":rtp_sender", ":rtp_sender_proxy", ":rtp_transceiver", + ":scoped_operations_batcher", ":simulcast_description", ":transceiver_list", ":usage_pattern",
diff --git a/pc/peer_connection.cc b/pc/peer_connection.cc index a3bc777..9a3e98a 100644 --- a/pc/peer_connection.cc +++ b/pc/peer_connection.cc
@@ -1233,13 +1233,17 @@ std::string sender_id = (track && !rtp_manager()->FindSenderById(track->id()) ? track->id() : CreateRandomUuid()); + ScopedOperationsBatcher worker_tasks(context_->worker_thread()); auto transceiver = rtp_manager()->CreateAndAddTransceiver( configuration_.media_config, sdp_handler_->audio_options(), sdp_handler_->video_options(), configuration_.crypto_options, sdp_handler_->video_bitrate_allocator_factory(), media_type, track, init.stream_ids, parameters.encodings, /*header_extensions_to_negotiate=*/{}, - /*simulcast_rejected=*/false, /*initial_simulcast_layers=*/{}, sender_id); + /*simulcast_rejected=*/false, /*initial_simulcast_layers=*/{}, + worker_tasks, sender_id); + RTCError error = worker_tasks.Run(); + RTC_DCHECK(error.ok()); transceiver->internal()->set_direction(init.direction); if (update_negotiation_needed) {
diff --git a/pc/rtp_transceiver.cc b/pc/rtp_transceiver.cc index 69520ab..12db4ac 100644 --- a/pc/rtp_transceiver.cc +++ b/pc/rtp_transceiver.cc
@@ -379,6 +379,7 @@ std::vector<RtpHeaderExtensionCapability> header_extensions_to_negotiate, bool simulcast_rejected, const std::vector<SimulcastLayer>& initial_simulcast_layers, + ScopedOperationsBatcher& worker_tasks, absl::AnyInvocable<void()> on_negotiation_needed) : env_(env), thread_(context->signaling_thread()), @@ -402,37 +403,50 @@ RTC_DCHECK(context_->is_configured_for_media()); RTC_DCHECK(media_type_ == MediaType::AUDIO || media_type_ == MediaType::VIDEO); - RTC_LOG_THREAD_BLOCK_COUNT(); + RTC_DCHECK_DISALLOW_THREAD_BLOCKING_CALLS(); if (media_type_ == MediaType::VIDEO) { ConfigureExtraVideoHeaderExtensions(init_send_encodings, header_extensions_to_negotiate_); } - // This should be possible without a blocking call to the worker, perhaps done - // asynchronously. At the moment this is complicated by the fact that - // construction of the channels actually changes the settings of the engine. - context_->worker_thread()->BlockingCall([&]() { - RTC_DCHECK_RUN_ON(this->context()->worker_thread()); - auto channels = CreateMediaContentChannels( - media_type_, env_, media_engine(), call, media_config, audio_options, - video_options, crypto_options, video_bitrate_allocator_factory, - GetEncoderSwitchRequestCallback()); - owned_send_channel_ = std::move(channels.first); - owned_receive_channel_ = std::move(channels.second); - senders_.push_back(CreateSender( - media_type_, env_, context_, legacy_stats_, set_streams_observer_, - sender_id, owned_send_channel_.get(), init_send_encodings, - simulcast_rejected, initial_simulcast_layers)); - }); + auto encoder_switch_callback = GetEncoderSwitchRequestCallback(); - ConfigureSender(senders_.back(), track.get(), stream_ids, init_send_encodings, - codec_vendor()); + worker_tasks.AddWithFinalizer( + [this, call, media_config, audio_options, video_options, crypto_options, + video_bitrate_allocator_factory, + encoder_switch_callback = std::move(encoder_switch_callback), + sender_id = std::string(sender_id), init_send_encodings, + simulcast_rejected, initial_simulcast_layers, track, stream_ids, + receiver_id = std::string( + receiver_id)]() mutable -> ScopedOperationsBatcher::FinalizerTask { + RTC_DCHECK_RUN_ON(this->context()->worker_thread()); + auto channels = CreateMediaContentChannels( + media_type_, env_, media_engine(), call, media_config, + audio_options, video_options, crypto_options, + video_bitrate_allocator_factory, + std::move(encoder_switch_callback)); + auto sender = CreateSender( + media_type_, env_, context_, legacy_stats_, set_streams_observer_, + sender_id, channels.first.get(), init_send_encodings, + simulcast_rejected, initial_simulcast_layers); + return ScopedOperationsBatcher::FinalizerTask( + [this, channels = std::move(channels), sender = std::move(sender), + track, stream_ids, init_send_encodings, receiver_id]() mutable { + RTC_DCHECK_RUN_ON(thread_); + owned_send_channel_ = std::move(channels.first); + owned_receive_channel_ = std::move(channels.second); + senders_.push_back(std::move(sender)); - receivers_.push_back(CreateReceiver( - media_type_, context_->signaling_thread(), context_->worker_thread(), - receiver_id.empty() ? CreateRandomUuid() : receiver_id, - owned_receive_channel_.get())); - RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(1); + ConfigureSender(senders_.back(), track.get(), stream_ids, + init_send_encodings, codec_vendor()); + + receivers_.push_back(CreateReceiver( + media_type_, context_->signaling_thread(), + context_->worker_thread(), + receiver_id.empty() ? CreateRandomUuid() : receiver_id, + owned_receive_channel_.get())); + }); + }); } RtpTransceiver::~RtpTransceiver() {
diff --git a/pc/rtp_transceiver.h b/pc/rtp_transceiver.h index 5386205..ee39dc9 100644 --- a/pc/rtp_transceiver.h +++ b/pc/rtp_transceiver.h
@@ -144,6 +144,7 @@ std::vector<RtpHeaderExtensionCapability> header_extensions_to_negotiate, bool simulcast_rejected, const std::vector<SimulcastLayer>& initial_simulcast_layers, + ScopedOperationsBatcher& worker_tasks, absl::AnyInvocable<void()> on_negotiation_needed); ~RtpTransceiver() override;
diff --git a/pc/rtp_transceiver_unittest.cc b/pc/rtp_transceiver_unittest.cc index 8575a3e..546b148 100644 --- a/pc/rtp_transceiver_unittest.cc +++ b/pc/rtp_transceiver_unittest.cc
@@ -1017,7 +1017,9 @@ AudioOptions audio_options; audio_options.audio_network_adaptor = true; - auto transceiver = make_ref_counted<RtpTransceiver>( + scoped_refptr<RtpTransceiver> transceiver; + ScopedOperationsBatcher worker_tasks(context()->worker_thread()); + transceiver = make_ref_counted<RtpTransceiver>( env(), call_.get(), MediaConfig(), /*sender_id=*/"sender", /*receiver_id=*/"receiver", MediaType::AUDIO, /*track=*/nullptr, @@ -1029,11 +1031,10 @@ /*video_bitrate_allocator_factory=*/nullptr, /*header_extensions=*/std::vector<RtpHeaderExtensionCapability>(), /*simulcast_rejected=*/false, - /*initial_simulcast_layers=*/std::vector<SimulcastLayer>(), + /*initial_simulcast_layers=*/std::vector<SimulcastLayer>(), worker_tasks, /*on_negotiation_needed=*/[] {}); - EXPECT_FALSE(transceiver->HasChannel()); - ScopedOperationsBatcher worker_tasks(context()->worker_thread()); + EXPECT_TRUE(worker_tasks.Run().ok()); ScopedOperationsBatcher network_tasks(context()->network_thread()); transceiver->CreateChannel( "0", call_.get(), MediaConfig(), /*srtp_required=*/false, CryptoOptions(), @@ -1049,8 +1050,9 @@ auto* fake_channel = static_cast<FakeVoiceMediaSendChannel*>(voice_channel); EXPECT_TRUE(fake_channel->options().audio_network_adaptor); - transceiver->ClearChannel(); - transceiver->StopStandard(); + network_tasks.Add(transceiver->GetClearChannelNetworkTask()); + worker_tasks.Add( + transceiver->GetDeleteChannelWorkerTask(/*stop_senders=*/true)); } // Sframe tests
diff --git a/pc/rtp_transmission_manager.cc b/pc/rtp_transmission_manager.cc index 64f1a07..e0ecad9 100644 --- a/pc/rtp_transmission_manager.cc +++ b/pc/rtp_transmission_manager.cc
@@ -47,6 +47,7 @@ #include "pc/rtp_sender.h" #include "pc/rtp_sender_proxy.h" #include "pc/rtp_transceiver.h" +#include "pc/scoped_operations_batcher.h" #include "pc/simulcast_description.h" #include "pc/usage_pattern.h" #include "pc/video_rtp_receiver.h" @@ -249,6 +250,7 @@ if (FindSenderById(sender_id)) { sender_id = CreateRandomUuid(); } + ScopedOperationsBatcher worker_tasks(context_->worker_thread()); transceiver = CreateAndAddTransceiver( media_config, audio_options, video_options, crypto_options, video_bitrate_allocator_factory, media_type, track, stream_ids, @@ -257,7 +259,9 @@ : std::vector<RtpEncodingParameters>(1, RtpEncodingParameters{}), /*header_extensions_to_negotiate=*/{}, /*simulcast_rejected=*/false, /*initial_simulcast_layers=*/{}, - sender_id, /*receiver_id=*/""); + worker_tasks, sender_id, /*receiver_id=*/""); + RTCError error = worker_tasks.Run(); + RTC_DCHECK(error.ok()); transceiver->internal()->set_created_by_addtrack(true); transceiver->internal()->set_direction(RtpTransceiverDirection::kSendRecv); } @@ -279,6 +283,7 @@ header_extensions_to_negotiate, bool simulcast_rejected, const std::vector<SimulcastLayer>& initial_simulcast_layers, + ScopedOperationsBatcher& worker_tasks, absl::string_view sender_id, absl::string_view receiver_id) { RTC_DCHECK_RUN_ON(signaling_thread()); @@ -315,7 +320,7 @@ codec_lookup_helper_, legacy_stats_, observer, audio_options, video_options, crypto_options, video_bitrate_allocator_factory, std::move(header_extensions), simulcast_rejected, - initial_simulcast_layers, + initial_simulcast_layers, worker_tasks, [this_weak_ptr = weak_ptr_factory_.GetWeakPtr()]() { if (this_weak_ptr) { this_weak_ptr->OnNegotiationNeeded();
diff --git a/pc/rtp_transmission_manager.h b/pc/rtp_transmission_manager.h index 6238d35..264d775 100644 --- a/pc/rtp_transmission_manager.h +++ b/pc/rtp_transmission_manager.h
@@ -114,6 +114,7 @@ header_extensions_to_negotiate, bool simulcast_rejected, const std::vector<SimulcastLayer>& initial_simulcast_layers, + ScopedOperationsBatcher& worker_tasks, absl::string_view sender_id, absl::string_view receiver_id = "");
diff --git a/pc/sdp_offer_answer.cc b/pc/sdp_offer_answer.cc index 108f102..2493ca6 100644 --- a/pc/sdp_offer_answer.cc +++ b/pc/sdp_offer_answer.cc
@@ -115,6 +115,40 @@ namespace webrtc { namespace { +void MaybeHandleLocallyRejectedTransceiver( + ContentSource source, + const SessionDescriptionInterface& new_session, + const ContentInfo& new_content, + scoped_refptr<RtpTransceiverProxyWithInternal<RtpTransceiver>> transceiver, + ScopedOperationsBatcher& worker_tasks) { + if (source != ContentSource::CS_LOCAL || !new_content.rejected) { + return; + } + if (new_session.GetType() == SdpType::kOffer) { + // If the RtpTransceiver API was used, it would already have made the + // transceiver stopping. But if the rejection was caused by SDP + // munging then we need to ensure the transceiver is stopping here. + if (!transceiver->internal()->stopping()) { + worker_tasks.AddWithFinalizer( + transceiver->internal()->StopStandardAsync()); + } + RTC_DCHECK(transceiver->internal()->stopping()); + } else { + RTC_DCHECK(new_session.GetType() == SdpType::kAnswer || + new_session.GetType() == SdpType::kPrAnswer); + // When RtpTransceiver API is used, rejection happens in the offer and + // the transceiver will already be stopped at local answer time + // (calling stop between SRD(offer) and SLD(answer) would not reject + // the content in the answer - instead this would trigger a follow-up + // O/A exchange). So if the content was rejected but the transceiver + // is not already stopped, SDP munging has happened and we need to + // ensure the transceiver is stopped. + if (!transceiver->internal()->stopped()) { + worker_tasks.Add(transceiver->internal()->GetStopTransceiverProcedure()); + } + RTC_DCHECK(transceiver->internal()->stopped()); + } +} struct DtlsTransportAndName { scoped_refptr<DtlsTransport> transport; @@ -4190,13 +4224,16 @@ ScopedOperationsBatcher worker_tasks(context_->worker_thread()); ScopedOperationsBatcher network_init_tasks(context_->network_thread()); const ContentInfos& new_contents = new_session.description()->contents(); + struct TransceiverUpdate { + scoped_refptr<RtpTransceiverProxyWithInternal<RtpTransceiver>> transceiver; + const ContentInfo& content; + }; + std::vector<TransceiverUpdate> transceivers_to_update; + for (size_t i = 0; i < new_contents.size(); ++i) { const ContentInfo& new_content = new_contents[i]; MediaType media_type = new_content.media_description()->type(); mid_generator_.AddKnownId(new_content.mid()); - auto it = bundle_groups_by_mid.find(new_content.mid()); - const ContentGroup* bundle_group = - it != bundle_groups_by_mid.end() ? it->second : nullptr; if (media_type == MediaType::AUDIO || media_type == MediaType::VIDEO) { const ContentInfo* old_local_content = nullptr; if (old_local_description && @@ -4210,9 +4247,9 @@ old_remote_content = &old_remote_description->description()->contents()[i]; } - auto transceiver_or_error = - AssociateTransceiver(source, new_session.GetType(), i, new_content, - old_local_content, old_remote_content); + auto transceiver_or_error = AssociateTransceiver( + source, new_session.GetType(), i, new_content, old_local_content, + old_remote_content, worker_tasks); if (!transceiver_or_error.ok()) { // In the case where a transceiver is rejected locally prior to being // associated, we don't expect to find a transceiver, but might find it @@ -4222,50 +4259,19 @@ } return transceiver_or_error.MoveError(); } - auto transceiver = transceiver_or_error.MoveValue(); - UpdateTransceiverChannel(transceiver, new_content, bundle_group, - 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(). - if (source == ContentSource::CS_LOCAL && new_content.rejected) { - // Local offer. - if (new_session.GetType() == SdpType::kOffer) { - // If the RtpTransceiver API was used, it would already have made the - // transceiver stopping. But if the rejection was caused by SDP - // munging then we need to ensure the transceiver is stopping here. - if (!transceiver->internal()->stopping()) { - worker_tasks.AddWithFinalizer( - transceiver->internal()->StopStandardAsync()); - } - RTC_DCHECK(transceiver->internal()->stopping()); - } else { - // Local answer. - RTC_DCHECK(new_session.GetType() == SdpType::kAnswer || - new_session.GetType() == SdpType::kPrAnswer); - // When RtpTransceiver API is used, rejection happens in the offer and - // the transceiver will already be stopped at local answer time - // (calling stop between SRD(offer) and SLD(answer) would not reject - // the content in the answer - instead this would trigger a follow-up - // O/A exchange). So if the content was rejected but the transceiver - // is not already stopped, SDP munging has happened and we need to - // ensure the transceiver is stopped. - if (!transceiver->internal()->stopped()) { - worker_tasks.Add( - transceiver->internal()->GetStopTransceiverProcedure()); - } - RTC_DCHECK(transceiver->internal()->stopped()); - } - } + transceivers_to_update.push_back( + {transceiver_or_error.MoveValue(), new_content}); } else if (media_type == MediaType::DATA) { - const auto data_mid = pc_->sctp_mid(); + const std::optional<std::string> data_mid = pc_->sctp_mid(); if (data_mid && new_content.mid() != data_mid.value()) { // Ignore all but the first data section. RTC_LOG(LS_INFO) << "Ignoring data media section with MID=" << new_content.mid(); continue; } + auto it = bundle_groups_by_mid.find(new_content.mid()); + const ContentGroup* bundle_group = + it != bundle_groups_by_mid.end() ? it->second : nullptr; RTCError error = UpdateDataChannelTransport(source, new_content, bundle_group); if (!error.ok()) { @@ -4279,7 +4285,30 @@ } } - RTCError error = network_teardown_tasks.Run(); + // Run transceiver creation tasks to ensure transceivers are fully constructed + // before UpdateTransceiverChannel is called. + RTCError error = worker_tasks.Run(); + if (!error.ok()) { + return error; + } + + for (TransceiverUpdate& update : transceivers_to_update) { + auto it = bundle_groups_by_mid.find(update.content.mid()); + const ContentGroup* bundle_group = + it != bundle_groups_by_mid.end() ? it->second : nullptr; + + UpdateTransceiverChannel(update.transceiver, update.content, bundle_group, + 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(). + MaybeHandleLocallyRejectedTransceiver(source, new_session, update.content, + std::move(update.transceiver), + worker_tasks); + } + + 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. @@ -4293,7 +4322,8 @@ size_t mline_index, const ContentInfo& content, const ContentInfo* old_local_content, - const ContentInfo* old_remote_content) { + const ContentInfo* old_remote_content, + ScopedOperationsBatcher& worker_tasks) { TRACE_EVENT0("webrtc", "SdpOfferAnswerHandler::AssociateTransceiver"); RTC_DCHECK(IsUnifiedPlan()); #if RTC_DCHECK_IS_ON @@ -4369,7 +4399,7 @@ pc_->GetCryptoOptions(), video_bitrate_allocator_factory_.get(), media_desc->type(), nullptr, {}, send_encodings, /*header_extensions_to_negotiate=*/{}, simulcast_rejected, - initial_simulcast_layers, sender_id, receiver_id); + initial_simulcast_layers, worker_tasks, sender_id, receiver_id); transceiver->internal()->set_direction( RtpTransceiverDirection::kRecvOnly); transceiver->internal()->ApplySframeEnabled(media_desc->sframe_enabled());
diff --git a/pc/sdp_offer_answer.h b/pc/sdp_offer_answer.h index 98f26b2..913ddce 100644 --- a/pc/sdp_offer_answer.h +++ b/pc/sdp_offer_answer.h
@@ -373,7 +373,8 @@ size_t mline_index, const ContentInfo& content, const ContentInfo* old_local_content, - const ContentInfo* old_remote_content) + const ContentInfo* old_remote_content, + ScopedOperationsBatcher& worker_tasks) RTC_RUN_ON(signaling_thread()); // Returns the media section in the given session description that is