Google Git
Sign in
webrtc/src/f201b69970^!/.
commit
f201b699704b2a1348db26ac75e5874c22ed5db3
[log]
author
Tommi <tommi@webrtc.org>
Tue Apr 07 17:06:04 2026
committer
WebRTC LUCI CQ <webrtc-scoped@luci-project-accounts.iam.gserviceaccount.com>
Wed Apr 08 04:18:38 2026
tree
5f032f399870cf568b9974eb2560a422624a157c
parent
d7181bf827c9d6cf020923d41c1e7398dd7e059e [diff]
Batch resetting of unsignaled receive streams

Batch up resetting of receive streams when payload type demuxing is
disabled. This introduces a new method,
`GetResetUnsignaledRecvStreamCallback`, which follows the same pattern
and behavior as `GetStatsCallback`.

Bug: webrtc:42222804
Change-Id: I523d340046e9544091b4b1417c947bc439d00efc
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/461660
Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org>
Reviewed-by: Harald Alvestrand <hta@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#47338}

diff --git a/media/base/fake_media_engine.h b/media/base/fake_media_engine.h
index acff856..18c278b 100644
--- a/media/base/fake_media_engine.h
+++ b/media/base/fake_media_engine.h

@@ -125,6 +125,10 @@
   bool CheckNoRtcp() { return rtcp_packets_.empty(); }
   void set_fail_set_recv_codecs(bool fail) { fail_set_recv_codecs_ = fail; }
   void ResetUnsignaledRecvStream() override {}
+  absl::AnyInvocable<void() &&> GetResetUnsignaledRecvStreamCallback()
+      override {
+    return [this]() { ResetUnsignaledRecvStream(); };
+  }
   std::optional<uint32_t> GetUnsignaledSsrc() const override {
     return std::nullopt;
   }

diff --git a/media/base/media_channel.h b/media/base/media_channel.h
index 95d1e7e..641b6e9 100644
--- a/media/base/media_channel.h
+++ b/media/base/media_channel.h

@@ -279,6 +279,11 @@
   // Resets any cached StreamParams for an unsignaled RecvStream, and removes
   // any existing unsignaled streams.
   virtual void ResetUnsignaledRecvStream() = 0;
+  // Returns a callback that can be used to reset unsignaled receive streams.
+  // The purpose is to allow binding the channel's safety flag to a callback
+  // that is run on the worker thread, ensuring the channel is alive.
+  virtual absl::AnyInvocable<void() &&>
+  GetResetUnsignaledRecvStreamCallback() = 0;
   // Sets the abstract interface class for sending RTP/RTCP data.
   virtual void SetInterface(MediaChannelNetworkInterface* iface) = 0;
   // Called on the network when an RTP packet is received.

diff --git a/media/engine/webrtc_video_engine.cc b/media/engine/webrtc_video_engine.cc
index 0e6ce01..cf1ff5d 100644
--- a/media/engine/webrtc_video_engine.cc
+++ b/media/engine/webrtc_video_engine.cc

@@ -3143,11 +3143,6 @@
 }
 
 void WebRtcVideoReceiveChannel::ResetUnsignaledRecvStream() {
-  if (!worker_thread_->IsCurrent()) {
-    worker_thread_->PostTask(SafeTask(
-        task_safety_.flag(), [this]() { ResetUnsignaledRecvStream(); }));
-    return;
-  }
   RTC_DCHECK_RUN_ON(&thread_checker_);
   RTC_LOG(LS_INFO) << "ResetUnsignaledRecvStream.";
   unsignaled_stream_params_ = StreamParams();
@@ -3168,6 +3163,12 @@
   }
 }
 
+absl::AnyInvocable<void() &&>
+WebRtcVideoReceiveChannel::GetResetUnsignaledRecvStreamCallback() {
+  return SafeTask(task_safety_.flag(),
+                  [this]() { ResetUnsignaledRecvStream(); });
+}
+
 std::optional<uint32_t> WebRtcVideoReceiveChannel::GetUnsignaledSsrc() const {
   RTC_DCHECK_RUN_ON(&thread_checker_);
   std::optional<uint32_t> ssrc;

diff --git a/media/engine/webrtc_video_engine.h b/media/engine/webrtc_video_engine.h
index 8c5442b..3630eb5 100644
--- a/media/engine/webrtc_video_engine.h
+++ b/media/engine/webrtc_video_engine.h

@@ -529,6 +529,7 @@
   }
   bool RemoveRecvStream(uint32_t ssrc) override;
   void ResetUnsignaledRecvStream() override;
+  absl::AnyInvocable<void() &&> GetResetUnsignaledRecvStreamCallback() override;
   std::optional<uint32_t> GetUnsignaledSsrc() const override;
   void OnDemuxerCriteriaUpdatePending() override;
   void OnDemuxerCriteriaUpdateComplete() override;

diff --git a/media/engine/webrtc_voice_engine.cc b/media/engine/webrtc_voice_engine.cc
index 4000fb1..05bfafe 100644
--- a/media/engine/webrtc_voice_engine.cc
+++ b/media/engine/webrtc_voice_engine.cc

@@ -2495,11 +2495,6 @@
 }
 
 void WebRtcVoiceReceiveChannel::ResetUnsignaledRecvStream() {
-  if (!worker_thread_->IsCurrent()) {
-    worker_thread_->PostTask(SafeTask(
-        task_safety_.flag(), [this]() { ResetUnsignaledRecvStream(); }));
-    return;
-  }
   RTC_DCHECK_RUN_ON(worker_thread_);
   RTC_LOG(LS_INFO) << "ResetUnsignaledRecvStream.";
   unsignaled_stream_params_ = StreamParams();
@@ -2510,6 +2505,12 @@
   }
 }
 
+absl::AnyInvocable<void() &&>
+WebRtcVoiceReceiveChannel::GetResetUnsignaledRecvStreamCallback() {
+  return SafeTask(task_safety_.flag(),
+                  [this]() { ResetUnsignaledRecvStream(); });
+}
+
 std::optional<uint32_t> WebRtcVoiceReceiveChannel::GetUnsignaledSsrc() const {
   RTC_DCHECK_RUN_ON(worker_thread_);
   if (unsignaled_recv_ssrcs_.empty()) {

diff --git a/media/engine/webrtc_voice_engine.h b/media/engine/webrtc_voice_engine.h
index f033c21..450e99a 100644
--- a/media/engine/webrtc_voice_engine.h
+++ b/media/engine/webrtc_voice_engine.h

@@ -358,6 +358,7 @@
   bool AddRecvStream(const StreamParams& sp) override;
   bool RemoveRecvStream(uint32_t ssrc) override;
   void ResetUnsignaledRecvStream() override;
+  absl::AnyInvocable<void() &&> GetResetUnsignaledRecvStreamCallback() override;
   std::optional<uint32_t> GetUnsignaledSsrc() const override;
 
   void OnDemuxerCriteriaUpdatePending() override;

diff --git a/pc/rtp_transceiver.cc b/pc/rtp_transceiver.cc
index 58da29f..59e4ef9 100644
--- a/pc/rtp_transceiver.cc
+++ b/pc/rtp_transceiver.cc

@@ -1378,13 +1378,6 @@
   channel_->Enable(enable);
 }
 
-void RtpTransceiver::ResetUnsignaledRecvStream() {
-  RTC_DCHECK_RUN_ON(thread_);
-  if (MediaReceiveChannelInterface* receive_channel = media_receive_channel()) {
-    receive_channel->ResetUnsignaledRecvStream();
-  }
-}
-
 const std::vector<StreamParams>& RtpTransceiver::channel_local_streams() const {
   RTC_DCHECK_RUN_ON(thread_);
   RTC_DCHECK(channel_);

diff --git a/pc/rtp_transceiver.h b/pc/rtp_transceiver.h
index 86e76f8..66aa443 100644
--- a/pc/rtp_transceiver.h
+++ b/pc/rtp_transceiver.h

@@ -393,7 +393,6 @@
                                SdpType type,
                                ScopedOperationsBatcher& batcher);
   void EnableChannel(bool enable);
-  void ResetUnsignaledRecvStream();
 
   const std::vector<StreamParams>& channel_local_streams() const;
   const std::vector<StreamParams>& channel_remote_streams() const;

diff --git a/pc/sdp_offer_answer.cc b/pc/sdp_offer_answer.cc
index ac88e6f..e5dae81 100644
--- a/pc/sdp_offer_answer.cc
+++ b/pc/sdp_offer_answer.cc

@@ -56,6 +56,7 @@
 #include "api/video/video_codec_constants.h"
 #include "media/base/codec.h"
 #include "media/base/codec_comparators.h"
+#include "media/base/media_channel.h"
 #include "media/base/media_constants.h"
 #include "media/base/media_engine.h"
 #include "media/base/rid_description.h"
@@ -5166,9 +5167,12 @@
   RTC_DCHECK(sdesc);
 
   if (ConfiguredForMedia()) {
+    // Batch up operations for the worker thread.
+    ScopedOperationsBatcher batcher(context_->worker_thread());
+
     // Note: This may perform a BlockingCall over to the network thread, which
     // we'll also do in a loop below.
-    UpdatePayloadTypeDemuxingState(source, bundle_groups_by_mid);
+    UpdatePayloadTypeDemuxingState(source, bundle_groups_by_mid, batcher);
 
     // Push down the new SDP media section for each audio/video transceiver.
     auto rtp_transceivers = transceivers()->ListInternal();
@@ -5215,9 +5219,6 @@
                            "types, ignoring all.";
     }
 
-    // Batch up the calls to set the local/remote channel content.
-    ScopedOperationsBatcher batcher(context_->worker_thread());
-
     for (const auto& [transceiver, content] : channels) {
       if (source == CS_LOCAL) {
         transceiver->SetChannelLocalContent(content, type, batcher);
@@ -5770,7 +5771,8 @@
 
 void SdpOfferAnswerHandler::UpdatePayloadTypeDemuxingState(
     ContentSource source,
-    const flat_map<std::string, const ContentGroup*>& bundle_groups_by_mid) {
+    const flat_map<std::string, const ContentGroup*>& bundle_groups_by_mid,
+    ScopedOperationsBatcher& worker_tasks) {
   TRACE_EVENT0("webrtc",
                "SdpOfferAnswerHandler::UpdatePayloadTypeDemuxingState");
   RTC_DCHECK_RUN_ON(signaling_thread());
@@ -5939,10 +5941,6 @@
   // unsignaled streams that might have been created previously based on
   // payload type matches. This prevents SSRC collisions if those SSRCs
   // are later reused.
-  // TODO(tommi): Could/should we rather issue the calls to the transceivers
-  // on the network thread? This will post tasks per transceiver to the worker
-  // thread. A common configuration will have joined worker and network threads,
-  // so bundling with the network thread may be more efficient.
   for (RtpTransceiver* transceiver : transceivers()->ListInternal()) {
     const ContentInfo* content =
         FindMediaSectionForTransceiver(transceiver, sdesc);
@@ -5951,13 +5949,17 @@
     }
     auto it = transports_to_update.find(content->mid());
     if (it != transports_to_update.end() && !it->second) {
-      // TODO: bugs.webrtc.org/42221580 - This will remove *all* unsignaled
-      // streams (those without an explicitly signaled SSRC), which may include
-      // streams that were matched to this channel by MID or RID. Ideally we'd
-      // remove only the streams that were matched based on payload type alone,
-      // but currently there is no straightforward way to identify those
-      // streams.
-      transceiver->ResetUnsignaledRecvStream();
+      if (MediaReceiveChannelInterface* receive_channel =
+              transceiver->media_receive_channel()) {
+        // TODO: bugs.webrtc.org/42221580 - This will remove *all* unsignaled
+        // streams (those without an explicitly signaled SSRC), which may
+        // include streams that were matched to this channel by MID or RID.
+        // Ideally we'd remove only the streams that were matched based on
+        // payload type alone, but currently there is no straightforward way to
+        // identify those streams.
+        worker_tasks.Add(
+            receive_channel->GetResetUnsignaledRecvStreamCallback());
+      }
     }
   }
 }

diff --git a/pc/sdp_offer_answer.h b/pc/sdp_offer_answer.h
index a2b9ef1..919459d 100644
--- a/pc/sdp_offer_answer.h
+++ b/pc/sdp_offer_answer.h

@@ -72,6 +72,8 @@
 
 namespace webrtc {
 
+class ScopedOperationsBatcher;
+
 // SdpOfferAnswerHandler is a component
 // of the PeerConnection object as defined
 // by the PeerConnectionInterface API surface.
@@ -575,7 +577,8 @@
   // payload type based demuxing in the affected channels.
   void UpdatePayloadTypeDemuxingState(
       ContentSource source,
-      const flat_map<std::string, const ContentGroup*>& bundle_groups_by_mid);
+      const flat_map<std::string, const ContentGroup*>& bundle_groups_by_mid,
+      ScopedOperationsBatcher& worker_tasks);
 
   // Updates the error state, signaling if necessary.
   void SetSessionError(SessionError error, const std::string& error_desc);

diff --git a/pc/test/mock_voice_media_receive_channel_interface.h b/pc/test/mock_voice_media_receive_channel_interface.h
index feac0ee..688120d 100644
--- a/pc/test/mock_voice_media_receive_channel_interface.h
+++ b/pc/test/mock_voice_media_receive_channel_interface.h

@@ -101,6 +101,10 @@
               (override));
   MOCK_METHOD(bool, RemoveRecvStream, (uint32_t ssrc), (override));
   MOCK_METHOD(void, ResetUnsignaledRecvStream, (), (override));
+  MOCK_METHOD(absl::AnyInvocable<void() &&>,
+              GetResetUnsignaledRecvStreamCallback,
+              (),
+              (override));
   MOCK_METHOD(void,
               SetInterface,
               (webrtc::MediaChannelNetworkInterface * iface),