Single threaded RtpTransceiver construction Move the creation of media content channels and the RTP sender/receiver directly to the signaling thread within the RtpTransceiver constructor. This is now possible following some previous updates to the other classes. Previously, these operations were delegated to the worker thread using a ScopedOperationsBatcher (`worker_tasks`). By executing them synchronously on the signaling thread, this change simplifies the initialization process and entirely removes the need for task batching during transceiver construction. * Update `CreateAndAddTransceiver` and `AssociateTransceiver` methods to remove the `worker_tasks` parameter across all call sites. * Adapts related `RtpTransceiver` tests. Bug: webrtc:42224170 Change-Id: I2497af614d0d36520ac9793d6c59e00370385eea Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/475100 Reviewed-by: Danil Chapovalov <danilchap@webrtc.org> Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47829}
diff --git a/pc/peer_connection.cc b/pc/peer_connection.cc index f36189d..eb0fde1 100644 --- a/pc/peer_connection.cc +++ b/pc/peer_connection.cc
@@ -1114,17 +1114,14 @@ 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=*/{}, - worker_tasks, sender_id); - RTCError error = worker_tasks.Run(); - RTC_DCHECK(error.ok()); + /*simulcast_rejected=*/false, + /*initial_simulcast_layers=*/{}, sender_id); transceiver->internal()->set_direction(init.direction); if (update_negotiation_needed) {
diff --git a/pc/rtp_transceiver.cc b/pc/rtp_transceiver.cc index 7133a44..4f8a3e4 100644 --- a/pc/rtp_transceiver.cc +++ b/pc/rtp_transceiver.cc
@@ -364,7 +364,6 @@ 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()), @@ -394,50 +393,32 @@ } auto encoder_switch_callback = GetEncoderSwitchRequestCallback(); - auto parameters_changed_callback = GetParametersChangedCallback(); std::vector<Codec> send_codecs = GetSendCodecs(); - 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), - parameters_changed_callback = std::move(parameters_changed_callback), - sender_id = std::string(sender_id), init_send_encodings, - simulcast_rejected, initial_simulcast_layers, track, stream_ids, - send_codecs = std::move(send_codecs), - receiver_id = std::string( - receiver_id)]() mutable -> ScopedOperationsBatcher::FinalizerTask { - RTC_DCHECK_RUN_ON(this->context()->worker_thread()); - auto channels = CreateMediaContentChannels( - media_type_, env_, voice_channel_factory(), video_channel_factory(), - call, media_config, audio_options, video_options, crypto_options, - video_bitrate_allocator_factory, std::move(encoder_switch_callback), - std::move(parameters_changed_callback)); - auto sender = CreateSender( - media_type_, env_, context_, legacy_stats_, set_streams_observer_, - sender_id, - absl::bind_front(&RtpTransceiver::TryToEnableSframe, this), - channels.first.get(), init_send_encodings, simulcast_rejected, - initial_simulcast_layers, stream_ids, std::move(send_codecs)); - return ScopedOperationsBatcher::FinalizerTask( - [this, channels = std::move(channels), sender = std::move(sender), - track, 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)); + auto channels = CreateMediaContentChannels( + media_type_, env_, voice_channel_factory(), video_channel_factory(), call, + media_config, audio_options, video_options, crypto_options, + video_bitrate_allocator_factory, std::move(encoder_switch_callback), + GetParametersChangedCallback()); - bool set_track_succeeded = senders_.back()->SetTrack(track.get()); - RTC_DCHECK(set_track_succeeded); + auto sender = CreateSender( + media_type_, env_, context_, legacy_stats_, set_streams_observer_, + sender_id, absl::bind_front(&RtpTransceiver::TryToEnableSframe, this), + channels.first.get(), init_send_encodings, simulcast_rejected, + initial_simulcast_layers, stream_ids, std::move(send_codecs)); - receivers_.push_back(CreateReceiver( - media_type_, context_->signaling_thread(), - context_->worker_thread(), - receiver_id.empty() ? CreateRandomUuid() : receiver_id, - owned_receive_channel_.get(), - absl::bind_front(&RtpTransceiver::TryToEnableSframe, this))); - }); - }); + owned_send_channel_ = std::move(channels.first); + owned_receive_channel_ = std::move(channels.second); + senders_.push_back(std::move(sender)); + + bool set_track_succeeded = senders_.back()->SetTrack(track.get()); + RTC_DCHECK(set_track_succeeded); + + receivers_.push_back(CreateReceiver( + media_type_, context_->signaling_thread(), context_->worker_thread(), + receiver_id.empty() ? CreateRandomUuid() : receiver_id, + owned_receive_channel_.get(), + absl::bind_front(&RtpTransceiver::TryToEnableSframe, this))); } RtpTransceiver::~RtpTransceiver() {
diff --git a/pc/rtp_transceiver.h b/pc/rtp_transceiver.h index a7e0985..f6c090b 100644 --- a/pc/rtp_transceiver.h +++ b/pc/rtp_transceiver.h
@@ -146,7 +146,6 @@ 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 235e6f4..f7d7873 100644 --- a/pc/rtp_transceiver_unittest.cc +++ b/pc/rtp_transceiver_unittest.cc
@@ -93,16 +93,14 @@ absl::AnyInvocable<RtpTransportInternal*() &&> transport_lookup = []() { return nullptr; }) { - ScopedOperationsBatcher worker_tasks(context->worker_thread()); auto transceiver = make_ref_counted<RtpTransceiver>( env, call, MediaConfig(), "sender", "receiver", MediaType::AUDIO, nullptr, std::vector<std::string>(), std::vector<RtpEncodingParameters>(), context, codec_lookup_helper, nullptr, nullptr, audio_options, VideoOptions(), CryptoOptions(), nullptr, std::vector<RtpHeaderExtensionCapability>(), - false, std::vector<SimulcastLayer>(), worker_tasks, [] {}); + false, std::vector<SimulcastLayer>(), [] {}); - RTC_CHECK(worker_tasks.Run().ok()); - + ScopedOperationsBatcher worker_tasks(context->worker_thread()); ScopedOperationsBatcher network_tasks(context->network_thread()); transceiver->CreateChannel("0", call, MediaConfig(), false, CryptoOptions(), audio_options, VideoOptions(), nullptr, @@ -124,16 +122,14 @@ absl::AnyInvocable<RtpTransportInternal*() &&> transport_lookup = []() { return nullptr; }) { - ScopedOperationsBatcher worker_tasks(context->worker_thread()); auto transceiver = make_ref_counted<RtpTransceiver>( env, call, MediaConfig(), "sender", "receiver", MediaType::VIDEO, nullptr, std::vector<std::string>(), std::vector<RtpEncodingParameters>(), context, codec_lookup_helper, nullptr, nullptr, AudioOptions(), video_options, CryptoOptions(), nullptr, std::vector<RtpHeaderExtensionCapability>(), - false, std::vector<SimulcastLayer>(), worker_tasks, [] {}); + false, std::vector<SimulcastLayer>(), [] {}); - RTC_CHECK(worker_tasks.Run().ok()); - + ScopedOperationsBatcher worker_tasks(context->worker_thread()); ScopedOperationsBatcher network_tasks(context->network_thread()); transceiver->CreateChannel("0", call, MediaConfig(), false, CryptoOptions(), AudioOptions(), video_options, nullptr,
diff --git a/pc/rtp_transmission_manager.cc b/pc/rtp_transmission_manager.cc index a40d287..c75534b 100644 --- a/pc/rtp_transmission_manager.cc +++ b/pc/rtp_transmission_manager.cc
@@ -47,7 +47,6 @@ #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" @@ -250,7 +249,6 @@ 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, @@ -259,9 +257,7 @@ : std::vector<RtpEncodingParameters>(1, RtpEncodingParameters{}), /*header_extensions_to_negotiate=*/{}, /*simulcast_rejected=*/false, /*initial_simulcast_layers=*/{}, - worker_tasks, sender_id, /*receiver_id=*/""); - RTCError error = worker_tasks.Run(); - RTC_DCHECK(error.ok()); + sender_id, /*receiver_id=*/""); transceiver->internal()->set_created_by_addtrack(true); transceiver->internal()->set_direction(RtpTransceiverDirection::kSendRecv); } @@ -283,7 +279,6 @@ 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()); @@ -320,7 +315,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, worker_tasks, + initial_simulcast_layers, [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 264d775..6238d35 100644 --- a/pc/rtp_transmission_manager.h +++ b/pc/rtp_transmission_manager.h
@@ -114,7 +114,6 @@ 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 ba28817..b08fe20 100644 --- a/pc/sdp_offer_answer.cc +++ b/pc/sdp_offer_answer.cc
@@ -4336,9 +4336,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, worker_tasks); + auto transceiver_or_error = + AssociateTransceiver(source, new_session.GetType(), i, new_content, + old_local_content, old_remote_content); 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 @@ -4374,13 +4374,6 @@ } } - // 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 = @@ -4399,7 +4392,7 @@ update.transceiver, worker_tasks); } - error = network_teardown_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. @@ -4413,8 +4406,7 @@ size_t mline_index, const ContentInfo& content, const ContentInfo* old_local_content, - const ContentInfo* old_remote_content, - ScopedOperationsBatcher& worker_tasks) { + const ContentInfo* old_remote_content) { TRACE_EVENT0("webrtc", "SdpOfferAnswerHandler::AssociateTransceiver"); RTC_DCHECK(IsUnifiedPlan()); #if RTC_DCHECK_IS_ON @@ -4490,7 +4482,7 @@ pc_->GetCryptoOptions(), video_bitrate_allocator_factory_.get(), media_desc->type(), nullptr, {}, send_encodings, /*header_extensions_to_negotiate=*/{}, simulcast_rejected, - initial_simulcast_layers, worker_tasks, sender_id, receiver_id); + initial_simulcast_layers, 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 5db241a..de62ef5 100644 --- a/pc/sdp_offer_answer.h +++ b/pc/sdp_offer_answer.h
@@ -377,8 +377,7 @@ size_t mline_index, const ContentInfo& content, const ContentInfo* old_local_content, - const ContentInfo* old_remote_content, - ScopedOperationsBatcher& worker_tasks) + const ContentInfo* old_remote_content) RTC_RUN_ON(signaling_thread()); // Returns the media section in the given session description that is