Use ScopedOperationsBatcher for media description pushdown Modify RtpTransceiver and SdpOfferAnswer to batch media description updates using ScopedOperationsBatcher. This replaces the previous approach where each transceiver update triggered a synchronous blocking call to the worker thread. Sequentially blocking the signaling thread for each transceiver during SDP renegotiation can cause performance bottlenecks and audio impairments, so the responsibility of "backing off" is now with the ScopedOperationsBatcher class instead of doing individual blocking calls. By batching these operations, the work on the worker thread is consolidated in a best-effort manner, and state updates to the signaling thread are handled via finalizer tasks once the batch completes. Bug: webrtc:42222804, webrtc:42223005 Change-Id: If826fdd30c1340afb0d426d2a7d1dac864976055 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/458360 Reviewed-by: Markus Handell <handellm@webrtc.org> Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47301}
diff --git a/pc/BUILD.gn b/pc/BUILD.gn index 5db70e7..9b04a2f 100644 --- a/pc/BUILD.gn +++ b/pc/BUILD.gn
@@ -1702,6 +1702,7 @@ ":rtp_sender", ":rtp_sender_proxy", ":rtp_transport_internal", + ":scoped_operations_batcher", ":session_description", ":video_rtp_receiver", "../api:audio_options_api",
diff --git a/pc/rtp_transceiver.cc b/pc/rtp_transceiver.cc index 0105748..946601a 100644 --- a/pc/rtp_transceiver.cc +++ b/pc/rtp_transceiver.cc
@@ -62,6 +62,7 @@ #include "pc/rtp_sender.h" #include "pc/rtp_sender_proxy.h" #include "pc/rtp_transport_internal.h" +#include "pc/scoped_operations_batcher.h" #include "pc/session_description.h" #include "pc/video_rtp_receiver.h" #include "rtc_base/checks.h" @@ -1299,32 +1300,34 @@ return channel_->SetRtpTransport(rtp_transport); } -RTCError RtpTransceiver::SetChannelLocalContent( +void RtpTransceiver::SetChannelLocalContent( const MediaContentDescription* content, - SdpType type) { + SdpType type, + ScopedOperationsBatcher& batcher) { RTC_DCHECK_RUN_ON(context()->signaling_thread()); - return SetChannelContent([&]() { - RTC_DCHECK_RUN_ON(context()->worker_thread()); - return channel_->SetLocalContent(content, type); - }); + SetChannelContent( + [this, content, type]() { + return channel_->SetLocalContent(content, type); + }, + batcher); } -RTCError RtpTransceiver::SetChannelRemoteContent( +void RtpTransceiver::SetChannelRemoteContent( const MediaContentDescription* content, - SdpType type) { + SdpType type, + ScopedOperationsBatcher& batcher) { RTC_DCHECK_RUN_ON(context()->signaling_thread()); - return SetChannelContent([&]() { - RTC_DCHECK_RUN_ON(context()->worker_thread()); - return channel_->SetRemoteContent(content, type); - }); + SetChannelContent( + [this, content, type]() { + return channel_->SetRemoteContent(content, type); + }, + batcher); } -RTCError RtpTransceiver::SetChannelContent( - absl::AnyInvocable<RTCError() &&> set_content) { +void RtpTransceiver::SetChannelContent( + absl::AnyInvocable<RTCError() &&> set_content, + ScopedOperationsBatcher& batcher) { RTC_DCHECK_RUN_ON(context()->signaling_thread()); - if (!channel_) { - return RTCError::InvalidState() << "No channel"; - } struct SenderParameters { const uint32_t ssrc; @@ -1339,29 +1342,35 @@ {.ssrc = sender->ssrc(), .sender = sender->internal()}); } - // Calls the callback on the worker thread, fetches and returns the - // RtpParameters for the senders. - RTCError result = context()->worker_thread()->BlockingCall([&]() { - RTCError error = std::move(set_content)(); - if (!error.ok()) { - return error; - } - for (auto& entry : sender_parameters) { - if (entry.ssrc != 0) { - entry.parameters = - channel_->media_send_channel()->GetRtpSendParameters(entry.ssrc); - } - } - return RTCError::OK(); - }); - - for (auto& entry : sender_parameters) { - if (entry.parameters) { - entry.sender->SetCachedParameters(std::move(*entry.parameters)); - } - } - - return result; + batcher.AddWithFinalizer( + [this, set_content = std::move(set_content), + sender_parameters = std::move(sender_parameters)]() mutable + -> RTCErrorOr<ScopedOperationsBatcher::FinalizerTask> { + RTC_DCHECK_RUN_ON(context()->worker_thread()); + if (!channel_) { + return RTCError::InvalidState() << "No channel"; + } + RTCError result = std::move(set_content)(); + if (!result.ok()) { + return result; + } + for (auto& entry : sender_parameters) { + if (entry.ssrc != 0) { + entry.parameters = + channel_->media_send_channel()->GetRtpSendParameters( + entry.ssrc); + } + } + return ScopedOperationsBatcher::FinalizerTask( + [sender_parameters = std::move(sender_parameters)]() mutable { + for (auto& entry : sender_parameters) { + if (entry.parameters) { + entry.sender->SetCachedParameters( + std::move(*entry.parameters)); + } + } + }); + }); } bool RtpTransceiver::SetChannelPayloadTypeDemuxingEnabled(bool enabled) {
diff --git a/pc/rtp_transceiver.h b/pc/rtp_transceiver.h index 730978b..a69d17b 100644 --- a/pc/rtp_transceiver.h +++ b/pc/rtp_transceiver.h
@@ -65,6 +65,7 @@ class DtlsTransport; class PeerConnectionSdpMethods; +class ScopedOperationsBatcher; // Implementation of the public RtpTransceiverInterface. // @@ -377,10 +378,20 @@ } bool SetChannelRtpTransport(RtpTransportInternal* rtp_transport); - RTCError SetChannelLocalContent(const MediaContentDescription* content, - SdpType type); - RTCError SetChannelRemoteContent(const MediaContentDescription* content, - SdpType type); + // Configures the channel with local content description. + // Pushes a multi-stage execution task into the provided + // `ScopedOperationsBatcher`. See `SetChannelContent` for details on the + // execution model. + void SetChannelLocalContent(const MediaContentDescription* content, + SdpType type, + ScopedOperationsBatcher& batcher); + // Configures the channel with remote content description. + // Pushes a multi-stage execution task into the provided + // `ScopedOperationsBatcher`. See `SetChannelContent` for details on the + // execution model. + void SetChannelRemoteContent(const MediaContentDescription* content, + SdpType type, + ScopedOperationsBatcher& batcher); bool SetChannelPayloadTypeDemuxingEnabled(bool enabled); void EnableChannel(bool enable); const std::vector<StreamParams>& channel_local_streams() const; @@ -431,7 +442,38 @@ GetOfferedAndImplementedHeaderExtensions( const MediaContentDescription* content) const; - RTCError SetChannelContent(absl::AnyInvocable<RTCError() &&> set_content); + // Configures the channel with the provided content description. + // Pushes a multi-stage execution task into the provided + // `ScopedOperationsBatcher`. + // + // Signaling Thread Worker Thread + // +-------------------+ +-------------------+ + // | SetChannelContent | | | + // | - Capture SSRC | | | + // | - Push Outer -----|------------->| [Outer Lambda] | + // | | | - Run set_content | + // | | | - GetRtpSendParams| + // | [Inner Lambda] <--|--------------| - Return Inner | + // | - SetCachedParams | | | + // +-------------------+ +-------------------+ + // + // Execution Stages: + // 1. **Preparation (Signaling Thread)**: Called when this method is invoked. + // Captures current sender SSRC and cached parameters. + // + // 2. **Execution (Worker Thread)**: The pushed outer lambda is executed + // by the ScopedOperationsBatcher on the worker thread. Runs + // `set_content()`. If successful, fetches updated `RtpSendParameters` + // from the media channel and returns an inner completion lambda. + // + // 3. **Completion (Signaling Thread)**: The inner lambda returned by the + // worker thread task is collected by `ScopedOperationsBatcher::Run()`. + // After all batched operations on the worker thread are complete, + // `Run()` executes these inner lambdas on the thread that called + // `Run()` (in this case, the Signaling Thread). This stage updates + // the cached parameters on the senders. + void SetChannelContent(absl::AnyInvocable<RTCError() &&> set_content, + ScopedOperationsBatcher& batcher); const Environment env_; // Enforce that this object is created, used and destroyed on one thread.
diff --git a/pc/sdp_offer_answer.cc b/pc/sdp_offer_answer.cc index 5a3ce99..9ef845c 100644 --- a/pc/sdp_offer_answer.cc +++ b/pc/sdp_offer_answer.cc
@@ -5220,25 +5220,22 @@ "types, ignoring all."; } - // This for-loop of invokes helps audio impairment during re-negotiations. - // One of the causes is that downstairs decoder creation is synchronous at - // the moment, and that a decoder is created for each codec listed in the - // SDP. - // - // TODO(bugs.webrtc.org/12840): consider merging the invokes again after - // these projects have shipped: - // - bugs.webrtc.org/12462 - // - crbug.com/1157227 - // - crbug.com/1187289 + // Batch up the calls to set the local/remote channel content. + ScopedOperationsBatcher batcher(context_->worker_thread()); + for (const auto& [transceiver, content] : channels) { - RTCError error = - (source == CS_LOCAL) - ? transceiver->SetChannelLocalContent(content, type) - : transceiver->SetChannelRemoteContent(content, type); - if (!error.ok()) { - return error; + if (source == CS_LOCAL) { + transceiver->SetChannelLocalContent(content, type, batcher); + } else { + transceiver->SetChannelRemoteContent(content, type, batcher); } } + + RTCError error = batcher.Run(); + if (!error.ok()) { + return error; + } + // If local and remote are both set, we assume that it's safe to trigger // CCFB. if (pc_->trials().IsEnabled("WebRTC-RFC8888CongestionControlFeedback")) {