Introduce SetSsrcTask and batch SetSsrc in ApplyLocalDescription The existing SetSsrc implementation in RtpSender required multiple synchronous blocking calls to the worker thread. In scenarios with many transceivers this was a significant performance bottleneck. This change introduces SetSsrcTask, which allows SetSsrc operations to be decomposed into asynchronous tasks. By using ScopedOperationsBatcher, these tasks are now batched and executed efficiently on the worker thread. The existing SetSsrc will be for PlanB only. Key modifications: * Added SetSsrcTask to the RtpSenderInternal interface to return a batchable task containing both worker and signaling thread logic. * Updated SdpOfferAnswer::ApplyLocalDescription to use ScopedOperationsBatcher, reducing the number of thread context switches. * Moved media channel configurations (frame encryptors, transformers, and selectors) into the batched worker task. * Marked the legacy SetSsrc method as Plan B only. Bug: webrtc:42222117 Change-Id: I49cfbeef4764d8423d678eec4f5c8017374e3187 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/465020 Reviewed-by: Danil Chapovalov <danilchap@webrtc.org> Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47507}
diff --git a/pc/BUILD.gn b/pc/BUILD.gn index 57026b6..1b7768f 100644 --- a/pc/BUILD.gn +++ b/pc/BUILD.gn
@@ -2029,6 +2029,7 @@ deps = [ ":dtmf_sender", ":legacy_stats_collector_interface", + ":scoped_operations_batcher", ":simulcast_description", "../api:audio_options_api", "../api:dtls_transport_interface", @@ -2060,6 +2061,7 @@ "../rtc_base:rtc_event", "../rtc_base:threading", "../rtc_base/synchronization:mutex", + "../rtc_base/system:plan_b_only", "//third_party/abseil-cpp/absl/algorithm:container", "//third_party/abseil-cpp/absl/functional:any_invocable", "//third_party/abseil-cpp/absl/strings:string_view", @@ -3417,6 +3419,7 @@ ":pc_test_utils", ":rtp_sender", ":rtp_transport_internal", + ":scoped_operations_batcher", ":video_rtp_receiver", ":video_track", "../api:audio_options_api", @@ -3995,6 +3998,7 @@ ":rtp_transceiver", ":rtp_transmission_manager", ":rtp_transport_internal", + ":scoped_operations_batcher", ":sctp_data_channel", ":sctp_utils", ":session_description",
diff --git a/pc/rtc_stats_collector_unittest.cc b/pc/rtc_stats_collector_unittest.cc index 60c4f13..ef92482 100644 --- a/pc/rtc_stats_collector_unittest.cc +++ b/pc/rtc_stats_collector_unittest.cc
@@ -3976,12 +3976,14 @@ /*stats=*/nullptr, /*set_streams_observer=*/nullptr, /*media_channel=*/nullptr); - worker_thread->BlockingCall([&] { - fake_media_channel->AddSendStream(StreamParams::CreateLegacy(1234)); - sender->SetMediaChannel(fake_media_channel.get()); - }); - - sender->SetSsrc(1234); + { + ScopedOperationsBatcher worker_tasks(pc_->worker_thread()); + worker_tasks.Add([&]() { + fake_media_channel->AddSendStream(StreamParams::CreateLegacy(1234)); + sender->SetMediaChannel(fake_media_channel.get()); + }); + worker_tasks.AddWithFinalizer(sender->SetSsrcTask(1234)); + } sender->SetTrack(track.get()); RTC_ALLOW_PLAN_B_DEPRECATION_BEGIN();
diff --git a/pc/rtp_sender.cc b/pc/rtp_sender.cc index 2a03b20..5fbb37f 100644 --- a/pc/rtp_sender.cc +++ b/pc/rtp_sender.cc
@@ -48,6 +48,7 @@ #include "media/base/media_engine.h" #include "pc/dtmf_sender.h" #include "pc/legacy_stats_collector_interface.h" +#include "pc/scoped_operations_batcher.h" #include "pc/simulcast_description.h" #include "rtc_base/checks.h" #include "rtc_base/crypto_random.h" @@ -839,6 +840,99 @@ } } +ScopedOperationsBatcher::BatchTaskWithFinalizer RtpSenderBase::SetSsrcTask( + uint32_t ssrc) { + RTC_DCHECK_RUN_ON(signaling_thread_); + if (stopped_ || ssrc == ssrc_) { + return nullptr; + } + + cached_parameters_.reset(); + + // If we are already sending with a particular SSRC, stop sending. + if (can_send_track()) { + ClearSend(); + RemoveTrackFromStats(); + } + ssrc_ = ssrc; + if (can_send_track()) { + SetSend(); + AddTrackToStats(); + } + + return [this, ssrc]() mutable + -> RTCErrorOr<ScopedOperationsBatcher::FinalizerTask> { + RTC_DCHECK_RUN_ON(worker_thread_); + + RtpParameters current_parameters; + bool params_modified = false; + + if (!init_parameters_.encodings.empty() || + init_parameters_.degradation_preference.has_value()) { + if (ssrc != 0) { + RTC_DCHECK(media_channel_); + // Get the current parameters, which are constructed from the SDP. The + // number of layers in the SDP is currently authoritative to support SDP + // munging for Plan-B simulcast with "a=ssrc-group:SIM <ssrc-id>..." + // lines as described in RFC 5576. All fields should be default + // constructed and the SSRC field set, which we need to copy. + current_parameters = media_channel_->GetRtpSendParameters(ssrc); + // SSRC 0 has special meaning as "no stream". In this case, + // current_parameters may have size 0. + RTC_CHECK_GE(current_parameters.encodings.size(), + init_parameters_.encodings.size()); + for (size_t i = 0; i < init_parameters_.encodings.size(); ++i) { + init_parameters_.encodings[i].ssrc = + current_parameters.encodings[i].ssrc; + init_parameters_.encodings[i].rid = + current_parameters.encodings[i].rid; + current_parameters.encodings[i] = init_parameters_.encodings[i]; + } + current_parameters.degradation_preference = + init_parameters_.degradation_preference; + params_modified = + media_channel_ + ->SetRtpSendParameters(ssrc, current_parameters, nullptr) + .ok(); + if (params_modified) { + // The parameters may change as they're applied. + current_parameters = media_channel_->GetRtpSendParameters(ssrc); + } + } + // Clear the `init_parameters_` after they have been applied to the + // media channel. This prevents stale values from being used in + // subsequent calls to `SetSsrc`, which could happen if `SetSsrc` is + // called multiple times on the same sender. See + // https://issues.webrtc.org/issues/500993975 for details. + init_parameters_.encodings.clear(); + init_parameters_.degradation_preference = std::nullopt; + } + + // While we're on the worker thread, attach the frame decryptor, transformer + // and selector to the current media channel. + if (frame_encryptor_ != nullptr) { + media_channel_->SetFrameEncryptor(ssrc, frame_encryptor_); + } + if (frame_transformer_ != nullptr) { + media_channel_->SetEncoderToPacketizerFrameTransformer( + ssrc, frame_transformer_); + } + if (encoder_selector_ != nullptr) { + media_channel_->SetEncoderSelector(ssrc, encoder_selector_); + } + + if (params_modified) { + return ScopedOperationsBatcher::FinalizerTask( + [this, current_parameters = std::move(current_parameters)]() mutable { + RTC_DCHECK_RUN_ON(signaling_thread_); + cached_parameters_ = std::move(current_parameters); + }); + } else { + return ScopedOperationsBatcher::FinalizerTask(); + } + }; +} + void RtpSenderBase::Stop() { RTC_DCHECK_RUN_ON(signaling_thread_); TRACE_EVENT0("webrtc", "RtpSenderBase::Stop");
diff --git a/pc/rtp_sender.h b/pc/rtp_sender.h index 60b3ce0..d0b6f80 100644 --- a/pc/rtp_sender.h +++ b/pc/rtp_sender.h
@@ -46,8 +46,10 @@ #include "media/base/media_channel.h" #include "pc/dtmf_sender.h" #include "pc/legacy_stats_collector_interface.h" +#include "pc/scoped_operations_batcher.h" #include "pc/simulcast_description.h" #include "rtc_base/synchronization/mutex.h" +#include "rtc_base/system/plan_b_only.h" #include "rtc_base/thread.h" #include "rtc_base/thread_annotations.h" @@ -68,7 +70,10 @@ // If `ssrc` is 0, this indiates that the sender should disconnect from the // underlying transport (this occurs if the sender isn't seen in a local // description). - virtual void SetSsrc(uint32_t ssrc) = 0; + PLAN_B_ONLY virtual void SetSsrc(uint32_t ssrc) = 0; + + [[nodiscard]] virtual ScopedOperationsBatcher::BatchTaskWithFinalizer + SetSsrcTask(uint32_t ssrc) = 0; virtual void set_stream_ids(const std::vector<std::string>& stream_ids) = 0; virtual void set_init_send_encodings( @@ -169,7 +174,10 @@ // If `ssrc` is 0, this indiates that the sender should disconnect from the // underlying transport (this occurs if the sender isn't seen in a local // description). - void SetSsrc(uint32_t ssrc) override; + PLAN_B_ONLY void SetSsrc(uint32_t ssrc) override; + ScopedOperationsBatcher::BatchTaskWithFinalizer SetSsrcTask( + uint32_t ssrc) override; + uint32_t ssrc() const override { RTC_DCHECK_RUN_ON(signaling_thread_); return ssrc_;
diff --git a/pc/rtp_sender_receiver_unittest.cc b/pc/rtp_sender_receiver_unittest.cc index b44a550..ee13647 100644 --- a/pc/rtp_sender_receiver_unittest.cc +++ b/pc/rtp_sender_receiver_unittest.cc
@@ -58,6 +58,7 @@ #include "pc/media_stream.h" #include "pc/rtp_sender.h" #include "pc/rtp_transport_internal.h" +#include "pc/scoped_operations_batcher.h" #include "pc/test/fake_video_track_source.h" #include "pc/video_rtp_receiver.h" #include "pc/video_track.h" @@ -216,7 +217,7 @@ ASSERT_TRUE(audio_rtp_sender_->SetTrack(audio_track_.get())); EXPECT_CALL(*set_streams_observer, OnSetStreams()); audio_rtp_sender_->SetStreams({local_stream_->id()}); - audio_rtp_sender_->SetSsrc(kAudioSsrc); + SetSsrc(kAudioSsrc, *audio_rtp_sender_); VerifyVoiceChannelInput(); } @@ -284,7 +285,7 @@ ASSERT_TRUE(video_rtp_sender_->SetTrack(video_track_.get())); EXPECT_CALL(*set_streams_observer, OnSetStreams()); video_rtp_sender_->SetStreams({local_stream_->id()}); - video_rtp_sender_->SetSsrc(ssrc); + SetSsrc(ssrc, *video_rtp_sender_); VerifyVideoChannelInput(ssrc); } void CreateVideoRtpSenderWithNoTrack() { @@ -568,6 +569,11 @@ voice_media_receive_channel_.get()); } + void SetSsrc(uint32_t ssrc, RtpSenderInternal& sender) { + ScopedOperationsBatcher worker_tasks(worker_thread_.get()); + worker_tasks.AddWithFinalizer(sender.SetSsrcTask(ssrc)); + } + test::RunLoop run_loop_; // Initialize `signaling_thread_` to point to the current thread. // This is the internal thread owned by `run_loop_`. @@ -800,7 +806,7 @@ // SSRC but no track. EXPECT_TRUE(audio_rtp_sender_->SetTrack(nullptr)); - audio_rtp_sender_->SetSsrc(kAudioSsrc); + SetSsrc(kAudioSsrc, *audio_rtp_sender_); VerifyVoiceChannelNoInput(); } @@ -815,7 +821,7 @@ // SSRC but no track. EXPECT_TRUE(video_rtp_sender_->SetTrack(nullptr)); - video_rtp_sender_->SetSsrc(kVideoSsrc); + SetSsrc(kVideoSsrc, *video_rtp_sender_); VerifyVideoChannelNoInput(); } @@ -825,7 +831,7 @@ CreateAudioRtpSenderWithNoTrack(); scoped_refptr<AudioTrackInterface> track = AudioTrack::Create(kAudioTrackId, nullptr); - audio_rtp_sender_->SetSsrc(kAudioSsrc); + SetSsrc(kAudioSsrc, *audio_rtp_sender_); audio_rtp_sender_->SetTrack(track.get()); VerifyVoiceChannelInput(); @@ -839,7 +845,7 @@ scoped_refptr<AudioTrackInterface> track = AudioTrack::Create(kAudioTrackId, nullptr); audio_rtp_sender_->SetTrack(track.get()); - audio_rtp_sender_->SetSsrc(kAudioSsrc); + SetSsrc(kAudioSsrc, *audio_rtp_sender_); VerifyVoiceChannelInput(); DestroyAudioRtpSender(); @@ -850,7 +856,7 @@ TEST_F(RtpSenderReceiverTest, VideoSenderEarlyWarmupSsrcThenTrack) { AddVideoTrack(); CreateVideoRtpSenderWithNoTrack(); - video_rtp_sender_->SetSsrc(kVideoSsrc); + SetSsrc(kVideoSsrc, *video_rtp_sender_); video_rtp_sender_->SetTrack(video_track_.get()); VerifyVideoChannelInput(); @@ -863,7 +869,7 @@ AddVideoTrack(); CreateVideoRtpSenderWithNoTrack(); video_rtp_sender_->SetTrack(video_track_.get()); - video_rtp_sender_->SetSsrc(kVideoSsrc); + SetSsrc(kVideoSsrc, *video_rtp_sender_); VerifyVideoChannelInput(); DestroyVideoRtpSender(); @@ -874,7 +880,7 @@ TEST_F(RtpSenderReceiverTest, AudioSenderSsrcSetToZero) { CreateAudioRtpSender(); - audio_rtp_sender_->SetSsrc(0); + SetSsrc(0, *audio_rtp_sender_); VerifyVoiceChannelNoInput(); } @@ -883,7 +889,7 @@ TEST_F(RtpSenderReceiverTest, VideoSenderSsrcSetToZero) { CreateAudioRtpSender(); - audio_rtp_sender_->SetSsrc(0); + SetSsrc(0, *audio_rtp_sender_); VerifyVideoChannelNoInput(); } @@ -901,7 +907,7 @@ TEST_F(RtpSenderReceiverTest, VideoSenderTrackSetToNull) { CreateVideoRtpSender(); - video_rtp_sender_->SetSsrc(0); + SetSsrc(0, *video_rtp_sender_); VerifyVideoChannelNoInput(); } @@ -910,7 +916,7 @@ TEST_F(RtpSenderReceiverTest, AudioSenderSsrcChanged) { CreateAudioRtpSender(); - audio_rtp_sender_->SetSsrc(kAudioSsrc2); + SetSsrc(kAudioSsrc2, *audio_rtp_sender_); VerifyVoiceChannelNoInput(kAudioSsrc); VerifyVoiceChannelInput(kAudioSsrc2); @@ -923,7 +929,7 @@ TEST_F(RtpSenderReceiverTest, VideoSenderSsrcChanged) { CreateVideoRtpSender(); - video_rtp_sender_->SetSsrc(kVideoSsrc2); + SetSsrc(kVideoSsrc2, *video_rtp_sender_); VerifyVideoChannelNoInput(kVideoSsrc); VerifyVideoChannelInput(kVideoSsrc2); @@ -1034,7 +1040,7 @@ audio_rtp_sender_->SetMediaChannel( voice_media_send_channel()->AsVoiceSendChannel()); }); - audio_rtp_sender_->SetSsrc(1); + SetSsrc(1, *audio_rtp_sender_); params = audio_rtp_sender_->GetParameters(); ASSERT_EQ(1u, params.encodings.size()); @@ -1308,7 +1314,7 @@ video_rtp_sender_->SetMediaChannel( video_media_send_channel()->AsVideoSendChannel()); }); - video_rtp_sender_->SetSsrc(kVideoSsrcSimulcast); + SetSsrc(kVideoSsrcSimulcast, *video_rtp_sender_); params = video_rtp_sender_->GetParameters(); ASSERT_EQ(2u, params.encodings.size()); @@ -1353,7 +1359,7 @@ video_rtp_sender_->SetMediaChannel( video_media_send_channel()->AsVideoSendChannel()); }); - video_rtp_sender_->SetSsrc(kVideoSsrcSimulcast); + SetSsrc(kVideoSsrcSimulcast, *video_rtp_sender_); params = video_rtp_sender_->GetParameters(); ASSERT_EQ(2u, params.encodings.size()); @@ -1415,7 +1421,13 @@ video_rtp_sender->SetMediaChannel( video_media_send_channel->AsVideoSendChannel()); - EXPECT_DEATH(video_rtp_sender->SetSsrc(kVideoSsrcSimulcast), ""); + EXPECT_DEATH( + { + ScopedOperationsBatcher worker_tasks(thread); + worker_tasks.AddWithFinalizer( + video_rtp_sender->SetSsrcTask(kVideoSsrcSimulcast)); + }, + ""); video_rtp_sender->Stop(); } #endif @@ -1856,7 +1868,7 @@ // Verify that the content hint is accounted for when video_rtp_sender_ does // get enabled. - video_rtp_sender_->SetSsrc(kVideoSsrc); + SetSsrc(kVideoSsrc, *video_rtp_sender_); EXPECT_EQ(true, video_media_send_channel()->options().is_screencast); // And removing the hint should go back to false (to verify that false was @@ -2074,7 +2086,7 @@ mock_channel_ptr->last_set_frame_encryptor_ = nullptr; // Expect propagation to media channel when SSRC is set. - video_rtp_sender_->SetSsrc(kVideoSsrc); + SetSsrc(kVideoSsrc, *video_rtp_sender_); EXPECT_EQ(mock_channel_ptr->last_set_frame_encryptor_, frame_encryptor); }
diff --git a/pc/sdp_offer_answer.cc b/pc/sdp_offer_answer.cc index 2493ca6..c3a66d6 100644 --- a/pc/sdp_offer_answer.cc +++ b/pc/sdp_offer_answer.cc
@@ -2157,43 +2157,7 @@ } if (IsUnifiedPlan()) { - if (ConfiguredForMedia()) { - // We must use List and not ListInternal here because - // transceivers()->StableState() is indexed by the non-internal refptr. - for (const auto& transceiver_ext : transceivers()->List()) { - auto transceiver = transceiver_ext->internal(); - if (transceiver->stopped()) { - continue; - } - const ContentInfo* content = - FindMediaSectionForTransceiver(transceiver, local_description()); - if (!content) { - continue; - } - if (content->rejected || !transceiver->HasChannel() || - transceiver->channel_local_streams().empty()) { - // 0 is a special value meaning "this sender has no associated send - // stream". Need to call this so the sender won't attempt to configure - // a no longer existing stream and run into DCHECKs in the lower - // layers. - transceiver->sender_internal()->SetSsrc(0); - } else { - // Get the StreamParams from the channel which could generate SSRCs. - const std::vector<StreamParams>& streams = - transceiver->channel_local_streams(); - transceiver->sender_internal()->set_stream_ids( - streams[0].stream_ids()); - auto encodings = - transceiver->sender_internal()->init_send_encodings(); - transceiver->sender_internal()->SetSsrc(streams[0].first_ssrc()); - if (!encodings.empty()) { - transceivers() - ->StableState(transceiver_ext) - ->SetInitSendEncodings(encodings); - } - } - } - } + UpdateSenderSsrcsFromLocalDescription(); } else { // Plan B semantics. @@ -2244,6 +2208,50 @@ return RTCError::OK(); } +void SdpOfferAnswerHandler::UpdateSenderSsrcsFromLocalDescription() { + RTC_DCHECK_RUN_ON(signaling_thread()); + RTC_DCHECK(IsUnifiedPlan()); + if (!ConfiguredForMedia()) { + return; + } + ScopedOperationsBatcher worker_tasks(context_->worker_thread()); + // We must use List and not ListInternal here because + // transceivers()->StableState() is indexed by the non-internal refptr. + for (const auto& transceiver_ext : transceivers()->List()) { + RtpTransceiver* transceiver = transceiver_ext->internal(); + if (transceiver->stopped()) { + continue; + } + const ContentInfo* content = + FindMediaSectionForTransceiver(transceiver, local_description()); + if (content == nullptr) { + continue; + } + scoped_refptr<RtpSenderInternal> sender = transceiver->sender_internal(); + if (content->rejected || !transceiver->HasChannel() || + transceiver->channel_local_streams().empty()) { + // 0 is a special value meaning "this sender has no associated send + // stream". Need to call this so the sender won't attempt to configure + // a no longer existing stream and run into DCHECKs in the lower + // layers. + worker_tasks.AddWithFinalizer(sender->SetSsrcTask(0)); + } else { + const std::vector<StreamParams>& streams = + transceiver->channel_local_streams(); + sender->set_stream_ids(streams[0].stream_ids()); + std::vector<RtpEncodingParameters> encodings = + sender->init_send_encodings(); + worker_tasks.AddWithFinalizer( + sender->SetSsrcTask(streams[0].first_ssrc())); + if (!encodings.empty()) { + transceivers() + ->StableState(transceiver_ext) + ->SetInitSendEncodings(std::move(encodings)); + } + } + } +} + void SdpOfferAnswerHandler::SetRemoteDescription( scoped_refptr<SetSessionDescriptionObserver> observer, std::unique_ptr<SessionDescriptionInterface> desc) {
diff --git a/pc/sdp_offer_answer.h b/pc/sdp_offer_answer.h index 913ddce..5db241a 100644 --- a/pc/sdp_offer_answer.h +++ b/pc/sdp_offer_answer.h
@@ -284,6 +284,10 @@ // Part of ApplyRemoteDescription steps specific to Unified Plan. void ApplyRemoteDescriptionUpdateTransceiverState(SdpType sdp_type); + // Updates sender SSRCs and init send encodings from transceivers. + // Part of ApplyLocalDescription steps specific to Unified Plan. + void UpdateSenderSsrcsFromLocalDescription() RTC_RUN_ON(signaling_thread()); + // Part of ApplyRemoteDescription steps specific to plan b. PLAN_B_ONLY void PlanBUpdateSendersAndReceivers( const ContentInfo* audio_content,
diff --git a/pc/test/mock_rtp_sender_internal.h b/pc/test/mock_rtp_sender_internal.h index 24f19bc..8b3c9a0 100644 --- a/pc/test/mock_rtp_sender_internal.h +++ b/pc/test/mock_rtp_sender_internal.h
@@ -32,6 +32,7 @@ #include "media/base/codec.h" #include "media/base/media_channel.h" #include "pc/rtp_sender.h" +#include "pc/scoped_operations_batcher.h" #include "test/gmock.h" namespace webrtc { @@ -121,6 +122,10 @@ (webrtc::MediaSendChannelInterface*), (override)); MOCK_METHOD(void, SetSsrc, (uint32_t), (override)); + MOCK_METHOD(ScopedOperationsBatcher::BatchTaskWithFinalizer, + SetSsrcTask, + (uint32_t), + (override)); MOCK_METHOD(void, set_stream_ids, (const std::vector<std::string>&),
diff --git a/pc/transceiver_list.cc b/pc/transceiver_list.cc index 8b1a59e..5234038 100644 --- a/pc/transceiver_list.cc +++ b/pc/transceiver_list.cc
@@ -13,6 +13,7 @@ #include <cstddef> #include <optional> #include <string> +#include <utility> #include <vector> #include "absl/strings/string_view.h" @@ -48,8 +49,8 @@ } void TransceiverStableState::SetInitSendEncodings( - const std::vector<RtpEncodingParameters>& encodings) { - init_send_encodings_ = encodings; + std::vector<RtpEncodingParameters> encodings) { + init_send_encodings_ = std::move(encodings); } std::vector<RtpTransceiver*> TransceiverList::ListInternal() const {
diff --git a/pc/transceiver_list.h b/pc/transceiver_list.h index 0f41e56..5136449 100644 --- a/pc/transceiver_list.h +++ b/pc/transceiver_list.h
@@ -42,8 +42,7 @@ void SetMSectionIfUnset(std::optional<std::string> mid, std::optional<size_t> mline_index); void SetRemoteStreamIds(const std::vector<std::string>& ids); - void SetInitSendEncodings( - const std::vector<RtpEncodingParameters>& encodings); + void SetInitSendEncodings(std::vector<RtpEncodingParameters> encodings); void SetFiredDirection( std::optional<RtpTransceiverDirection> fired_direction) { fired_direction_ = fired_direction;