Add SSRC tracking and receive sink clearing scaffolding Add scaffolding changes to prepare the channel layer for direct packet delivery on the network thread. This brings architectural updates to allow upcoming changes for routing packets on the network thread. - Add ClearReceiveSinks_n, SetReceiveSsrcs_n, and GetUnsignaledSsrcs methods to MediaReceiveChannelInterface. - Update BaseChannel methods (DisconnectFromRtpTransport_n, RegisterRtpDemuxerSink_w, and UpdateRemoteStreams_w) to pass active and removed SSRCs down to the receive channels. - Implement extraction and tracking of removed SSRCs, including unsignaled streams, during remote stream updates. Bug: webrtc:42222117 Change-Id: I6f28468dbf1dcf270ee360d9d44b8c0256e638b4 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/474580 Reviewed-by: Harald Alvestrand <hta@webrtc.org> Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47776}
diff --git a/media/BUILD.gn b/media/BUILD.gn index 871fd49..7b37914 100644 --- a/media/BUILD.gn +++ b/media/BUILD.gn
@@ -355,6 +355,7 @@ "../rtc_base:network_route", "../rtc_base:socket", "../rtc_base:stringutils", + "../rtc_base/containers:flat_set", "../rtc_base/network:sent_packet", "../video/config:encoder_config", "//third_party/abseil-cpp/absl/container:flat_hash_map", @@ -890,6 +891,7 @@ "../rtc_base:stringutils", "../rtc_base:threading", "../rtc_base:timeutils", + "../rtc_base/containers:flat_set", "../rtc_base/network:sent_packet", "../rtc_base/synchronization:mutex", "../rtc_base/system:file_wrapper",
diff --git a/media/base/fake_media_engine.h b/media/base/fake_media_engine.h index 1aee698..190ea69 100644 --- a/media/base/fake_media_engine.h +++ b/media/base/fake_media_engine.h
@@ -75,6 +75,7 @@ #include "modules/rtp_rtcp/source/rtp_packet_received.h" #include "rtc_base/async_packet_socket.h" #include "rtc_base/checks.h" +#include "rtc_base/containers/flat_set.h" #include "rtc_base/copy_on_write_buffer.h" #include "rtc_base/network/sent_packet.h" #include "rtc_base/network_route.h" @@ -139,6 +140,25 @@ std::optional<uint32_t> GetUnsignaledSsrc() const override { return std::nullopt; } + void SetReceiveSsrcs_n(const flat_set<uint32_t>& ssrcs) override { + receive_ssrcs_n_ = ssrcs; + } + void ClearReceiveSinks_n( + std::optional<std::vector<uint32_t>> ssrcs) override { + clear_receive_sinks_calls_.push_back(ssrcs); + } + std::vector<uint32_t> GetUnsignaledSsrcs() const override { + return fake_unsignaled_ssrcs_; + } + void AddFakeUnsignaledSsrc(uint32_t ssrc) { + fake_unsignaled_ssrcs_.push_back(ssrc); + } + void ClearFakeUnsignaledSsrcs() { fake_unsignaled_ssrcs_.clear(); } + const flat_set<uint32_t>& receive_ssrcs_n() const { return receive_ssrcs_n_; } + const std::vector<std::optional<std::vector<uint32_t>>>& + clear_receive_sinks_calls() const { + return clear_receive_sinks_calls_; + } virtual bool SetLocalSsrc(const StreamParams& /* sp */) { return true; } void OnDemuxerCriteriaUpdatePending() override {} @@ -229,6 +249,10 @@ bool fail_set_recv_codecs() const { return fail_set_recv_codecs_; } private: + flat_set<uint32_t> receive_ssrcs_n_; + std::vector<std::optional<std::vector<uint32_t>>> clear_receive_sinks_calls_; + std::vector<uint32_t> fake_unsignaled_ssrcs_; + bool playout_; std::vector<RtpExtension> recv_extensions_; std::list<std::string> rtp_packets_;
diff --git a/media/base/media_channel.h b/media/base/media_channel.h index 3a77b05..e18dd1d 100644 --- a/media/base/media_channel.h +++ b/media/base/media_channel.h
@@ -59,6 +59,7 @@ #include "modules/rtp_rtcp/include/report_block_data.h" #include "modules/rtp_rtcp/source/rtp_packet_received.h" #include "rtc_base/async_packet_socket.h" +#include "rtc_base/containers/flat_set.h" #include "rtc_base/copy_on_write_buffer.h" #include "rtc_base/network/sent_packet.h" #include "rtc_base/network_route.h" @@ -289,10 +290,17 @@ virtual absl::AnyInvocable<void() &&> GetResetUnsignaledRecvStreamTask() = 0; // Sets the abstract interface class for sending RTP/RTCP data. virtual void SetInterface(MediaChannelNetworkInterface* iface) = 0; + virtual void SetReceiveSsrcs_n(const flat_set<uint32_t>& ssrcs) {} + // Clears receive sinks on the network thread. + // If `ssrcs` is std::nullopt, all receive sinks are cleared (permanent + // teardown). Otherwise, only the specified `ssrcs` are cleared. + virtual void ClearReceiveSinks_n(std::optional<std::vector<uint32_t>> ssrcs) { + } // Called on the network when an RTP packet is received. virtual void OnPacketReceived(RtpPacketReceived packet) = 0; // Gets the current unsignaled receive stream's SSRC, if there is one. virtual std::optional<uint32_t> GetUnsignaledSsrc() const = 0; + virtual std::vector<uint32_t> GetUnsignaledSsrcs() const = 0; // This is currently a workaround because of the demuxer state being managed // across two separate threads. Once the state is consistently managed on // the same thread (network), this workaround can be removed.
diff --git a/media/engine/webrtc_video_engine.cc b/media/engine/webrtc_video_engine.cc index 61824c8..a235231 100644 --- a/media/engine/webrtc_video_engine.cc +++ b/media/engine/webrtc_video_engine.cc
@@ -3178,6 +3178,17 @@ return ssrc; } +std::vector<uint32_t> WebRtcVideoReceiveChannel::GetUnsignaledSsrcs() const { + RTC_DCHECK_RUN_ON(&thread_checker_); + std::vector<uint32_t> ssrcs; + for (const auto& [ssrc, stream] : receive_streams_) { + if (stream->IsDefaultStream()) { + ssrcs.push_back(ssrc); + } + } + return ssrcs; +} + void WebRtcVideoReceiveChannel::OnDemuxerCriteriaUpdatePending() { RTC_DCHECK_RUN_ON(&thread_checker_); ++demuxer_criteria_id_;
diff --git a/media/engine/webrtc_video_engine.h b/media/engine/webrtc_video_engine.h index 57941e2..3db65ef 100644 --- a/media/engine/webrtc_video_engine.h +++ b/media/engine/webrtc_video_engine.h
@@ -542,6 +542,7 @@ void ResetUnsignaledRecvStream() override; absl::AnyInvocable<void() &&> GetResetUnsignaledRecvStreamTask() override; std::optional<uint32_t> GetUnsignaledSsrc() const override; + std::vector<uint32_t> GetUnsignaledSsrcs() const override; void OnDemuxerCriteriaUpdatePending() override; void OnDemuxerCriteriaUpdateComplete() override; bool SetSink(uint32_t ssrc, VideoSinkInterface<VideoFrame>* sink) override;
diff --git a/media/engine/webrtc_video_engine_unittest.cc b/media/engine/webrtc_video_engine_unittest.cc index cf7d4c5..93d4723 100644 --- a/media/engine/webrtc_video_engine_unittest.cc +++ b/media/engine/webrtc_video_engine_unittest.cc
@@ -7146,6 +7146,22 @@ EXPECT_EQ(receivers2[0]->GetConfig().rtp.remote_ssrc, kIncomingSignalledSsrc); } +TEST_F(WebRtcVideoChannelTest, GetUnsignaledSsrcs) { + EXPECT_TRUE(receive_channel_->GetUnsignaledSsrcs().empty()); + + // Receive a packet with an unsignaled SSRC. + RtpPacketReceived packet; + packet.SetSsrc(kIncomingUnsignalledSsrc); + ReceivePacketAndAdvanceTime(packet); + + EXPECT_THAT(receive_channel_->GetUnsignaledSsrcs(), + testing::ElementsAre(kIncomingUnsignalledSsrc)); + + // Resetting the unsignaled stream should clear the SSRCs. + receive_channel_->ResetUnsignaledRecvStream(); + EXPECT_TRUE(receive_channel_->GetUnsignaledSsrcs().empty()); +} + TEST_F(WebRtcVideoChannelTest, RecentlyAddedSsrcsDoNotCreateUnsignalledRecvStreams) { const uint32_t kSsrc1 = 1;
diff --git a/media/engine/webrtc_voice_engine.cc b/media/engine/webrtc_voice_engine.cc index e2c7838..d0a125a 100644 --- a/media/engine/webrtc_voice_engine.cc +++ b/media/engine/webrtc_voice_engine.cc
@@ -2509,6 +2509,11 @@ return unsignaled_recv_ssrcs_.back(); } +std::vector<uint32_t> WebRtcVoiceReceiveChannel::GetUnsignaledSsrcs() const { + RTC_DCHECK_RUN_ON(worker_thread_); + return unsignaled_recv_ssrcs_; +} + // Not implemented. // TODO(https://crbug.com/webrtc/12676): Implement a fix for the unsignalled // SSRC race that can happen when an m= section goes from receiving to not
diff --git a/media/engine/webrtc_voice_engine.h b/media/engine/webrtc_voice_engine.h index 0cba300..9d268b8 100644 --- a/media/engine/webrtc_voice_engine.h +++ b/media/engine/webrtc_voice_engine.h
@@ -352,6 +352,7 @@ void ResetUnsignaledRecvStream() override; absl::AnyInvocable<void() &&> GetResetUnsignaledRecvStreamTask() override; std::optional<uint32_t> GetUnsignaledSsrc() const override; + std::vector<uint32_t> GetUnsignaledSsrcs() const override; void OnDemuxerCriteriaUpdatePending() override; void OnDemuxerCriteriaUpdateComplete() override;
diff --git a/media/engine/webrtc_voice_engine_unittest.cc b/media/engine/webrtc_voice_engine_unittest.cc index bf5b4a9..99f7701 100644 --- a/media/engine/webrtc_voice_engine_unittest.cc +++ b/media/engine/webrtc_voice_engine_unittest.cc
@@ -2885,6 +2885,28 @@ EXPECT_EQ(0u, receivers2.size()); } +TEST_P(WebRtcVoiceEngineTestFake, GetUnsignaledSsrcs) { + ASSERT_TRUE(SetupChannel()); + // No receive streams to start with. + ASSERT_TRUE(call_.GetAudioReceiveStreams().empty()); + EXPECT_TRUE(receive_channel_->GetUnsignaledSsrcs().empty()); + + // Deliver a couple packets with unsignaled SSRCs. + uint8_t packet[sizeof(kPcmuFrame)]; + memcpy(packet, kPcmuFrame, sizeof(kPcmuFrame)); + SetBE32(std::span<uint8_t>(&packet[8], 4), 0x1234); + DeliverPacket(packet); + SetBE32(std::span<uint8_t>(&packet[8], 4), 0x5678); + DeliverPacket(packet); + + EXPECT_THAT(receive_channel_->GetUnsignaledSsrcs(), + testing::ElementsAre(0x1234, 0x5678)); + + // Should remove all default streams. + receive_channel_->ResetUnsignaledRecvStream(); + EXPECT_TRUE(receive_channel_->GetUnsignaledSsrcs().empty()); +} + // Test that receiving N unsignaled stream works (streams will be created), and // that packets are forwarded to them all. TEST_P(WebRtcVoiceEngineTestFake, RecvMultipleUnsignaled) {
diff --git a/pc/channel.cc b/pc/channel.cc index 1231adc..218c876 100644 --- a/pc/channel.cc +++ b/pc/channel.cc
@@ -253,7 +253,7 @@ return true; } -void BaseChannel::DisconnectFromRtpTransport_n() { +void BaseChannel::DisconnectFromRtpTransport_n(bool permanent_teardown) { RTC_DCHECK(rtp_transport_); RTC_DCHECK(media_send_channel()); rtp_transport_->UnregisterRtpDemuxerSink(this); @@ -262,6 +262,9 @@ rtp_transport_->UnsubscribeSentPacket(this); rtp_transport_ = nullptr; media_send_channel()->SetInterface(nullptr); + if (permanent_teardown) { + media_receive_channel()->ClearReceiveSinks_n(std::nullopt); + } media_receive_channel()->SetInterface(nullptr); } @@ -273,7 +276,7 @@ } if (rtp_transport_) { - DisconnectFromRtpTransport_n(); + DisconnectFromRtpTransport_n(rtp_transport == nullptr); } RTC_DCHECK(!rtp_transport_); @@ -536,7 +539,8 @@ } bool BaseChannel::RegisterRtpDemuxerSink_w( - const MediaContentDescription* content) { + const MediaContentDescription* content, + std::vector<uint32_t> removed_ssrcs) { bool clear_payload_types = false; if (!RtpTransceiverDirectionHasSend(content->direction())) { RTC_DLOG(LS_VERBOSE) @@ -560,6 +564,10 @@ return false; } + if (!removed_ssrcs.empty()) { + media_receive_channel()->ClearReceiveSinks_n(std::move(removed_ssrcs)); + } + bool needs_re_registration = false; if (clear_payload_types && !payload_types_.empty()) { @@ -571,6 +579,7 @@ ssrcs_ = std::move(ssrcs); needs_re_registration = true; } + media_receive_channel()->SetReceiveSsrcs_n(ssrcs_); if (!needs_re_registration) { return true; @@ -973,26 +982,46 @@ const bool new_has_unsignaled_ssrcs = HasStreamWithNoSsrcs(streams); const bool old_has_unsignaled_ssrcs = HasStreamWithNoSsrcs(remote_streams_); + // Check for streams that have been removed. + std::vector<uint32_t> removed_ssrcs; + bool reset_unsignaled_streams = false; + for (const StreamParams& old_stream : remote_streams_) { + // If we no longer have an unsignaled stream, we would like to remove + // the unsignaled stream params that are cached. + if (!old_stream.has_ssrcs() && !new_has_unsignaled_ssrcs) { + if (!reset_unsignaled_streams) { + reset_unsignaled_streams = true; + std::vector<uint32_t> unsignaled = + media_receive_channel()->GetUnsignaledSsrcs(); + removed_ssrcs.insert(removed_ssrcs.end(), unsignaled.begin(), + unsignaled.end()); + RTC_LOG(LS_INFO) << "Reset unsignaled remote stream for " << ToString() + << "."; + } + } else if (old_stream.has_ssrcs() && + !GetStreamBySsrc(streams, old_stream.first_ssrc())) { + removed_ssrcs.push_back(old_stream.first_ssrc()); + } + } + RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(0); // Re-register the sink to update after changing the demuxer criteria first. - if (!RegisterRtpDemuxerSink_w(content)) { + if (!RegisterRtpDemuxerSink_w(content, std::move(removed_ssrcs))) { return RTCError::InvalidParameter() << "Failed to set up audio demuxing for mid='" << mid() << "'."; } RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(1); - // Check for streams that have been removed. + if (reset_unsignaled_streams) { + media_receive_channel()->ResetUnsignaledRecvStream(); + } + + // Now remove the old streams on the worker thread. for (const StreamParams& old_stream : remote_streams_) { - // If we no longer have an unsignaled stream, we would like to remove - // the unsignaled stream params that are cached. - if (!old_stream.has_ssrcs() && !new_has_unsignaled_ssrcs) { - media_receive_channel()->ResetUnsignaledRecvStream(); - RTC_LOG(LS_INFO) << "Reset unsignaled remote stream for " << ToString() - << "."; - } else if (old_stream.has_ssrcs() && - !GetStreamBySsrc(streams, old_stream.first_ssrc())) { + if (old_stream.has_ssrcs() && + !GetStreamBySsrc(streams, old_stream.first_ssrc())) { if (media_receive_channel()->RemoveRecvStream(old_stream.first_ssrc())) { RTC_LOG(LS_INFO) << "Remove remote ssrc: " << old_stream.first_ssrc() << " from " << ToString() << ".";
diff --git a/pc/channel.h b/pc/channel.h index f08e7c7..a318d63 100644 --- a/pc/channel.h +++ b/pc/channel.h
@@ -265,7 +265,8 @@ // Registers a demuxer criteria with the transport, on the network thread. // This function will fail if there's no transport of if a sink is already // registered for this channel's demuxer_critera(). - bool RegisterRtpDemuxerSink_w(const MediaContentDescription* content) + bool RegisterRtpDemuxerSink_w(const MediaContentDescription* content, + std::vector<uint32_t> removed_ssrcs = {}) RTC_RUN_ON(worker_thread()); // Return description of media channel to facilitate logging @@ -277,7 +278,8 @@ private: bool ConnectToRtpTransport_n(RtpTransportInternal* rtp_transport) RTC_RUN_ON(network_thread()); - void DisconnectFromRtpTransport_n() RTC_RUN_ON(network_thread()); + void DisconnectFromRtpTransport_n(bool permanent_teardown) + RTC_RUN_ON(network_thread()); void SignalSentPacket_n(const SentPacketInfo& sent_packet); // Only called on the network thread. RtpDemuxerCriteria demuxer_criteria() const RTC_RUN_ON(network_thread());
diff --git a/pc/channel_unittest.cc b/pc/channel_unittest.cc index 15959db..837cf56 100644 --- a/pc/channel_unittest.cc +++ b/pc/channel_unittest.cc
@@ -1706,6 +1706,82 @@ channel2_->media_receive_channel()); } + void TestSetRtpTransportNullClearsReceiveSinks() { + CreateChannels(0, 0); + SendTask(channel1_->network_thread(), + [this]() { channel1_->SetRtpTransport(nullptr); }); + EXPECT_TRUE(absl::c_linear_search( + media_receive_channel1_impl()->clear_receive_sinks_calls(), + std::nullopt)); + } + + void TestRegisterRtpDemuxerSinkPropagatesReceiveSsrcs() { + CreateChannels(0, 0); + typename T::Content content; + CreateContent(0, kPcmuCodec, kH264Codec, &content); + AddLegacyStreamInContent(kSsrc1, 0, &content); + + EXPECT_TRUE(channel1_->SetRemoteContent(&content, SdpType::kOffer).ok()); + EXPECT_GT(media_receive_channel1_impl()->receive_ssrcs_n().count(kSsrc1), + 0u); + } + + void TestUpdateRemoteStreamsClearsRemovedSinks() { + CreateChannels(0, 0); + typename T::Content content1; + CreateContent(0, kPcmuCodec, kH264Codec, &content1); + AddLegacyStreamInContent(kSsrc1, 0, &content1); + AddLegacyStreamInContent(kSsrc2, 0, &content1); + + EXPECT_TRUE(channel1_->SetRemoteContent(&content1, SdpType::kOffer).ok()); + EXPECT_TRUE(media_receive_channel1_impl()->HasRecvStream(kSsrc1)); + EXPECT_TRUE(media_receive_channel1_impl()->HasRecvStream(kSsrc2)); + + typename T::Content content2; + CreateContent(0, kPcmuCodec, kH264Codec, &content2); + AddLegacyStreamInContent(kSsrc2, 0, &content2); + + EXPECT_TRUE(channel1_->SetRemoteContent(&content2, SdpType::kOffer).ok()); + + bool found_cleared_ssrc = false; + for (const auto& call : + media_receive_channel1_impl()->clear_receive_sinks_calls()) { + if (call.has_value() && absl::c_linear_search(*call, kSsrc1)) { + found_cleared_ssrc = true; + break; + } + } + EXPECT_TRUE(found_cleared_ssrc); + } + + void TestUpdateRemoteStreamsClearsUnsignaledSinks() { + CreateChannels(0, 0); + + typename T::Content content1; + CreateContent(0, kPcmuCodec, kH264Codec, &content1); + StreamParams unsignaled_stream; + content1.AddStream(unsignaled_stream); + + EXPECT_TRUE(channel1_->SetRemoteContent(&content1, SdpType::kOffer).ok()); + + media_receive_channel1_impl()->AddFakeUnsignaledSsrc(kSsrc3); + + typename T::Content content2; + CreateContent(0, kPcmuCodec, kH264Codec, &content2); + + EXPECT_TRUE(channel1_->SetRemoteContent(&content2, SdpType::kOffer).ok()); + + bool found_cleared_ssrc = false; + for (const auto& call : + media_receive_channel1_impl()->clear_receive_sinks_calls()) { + if (call.has_value() && absl::c_linear_search(*call, kSsrc3)) { + found_cleared_ssrc = true; + break; + } + } + EXPECT_TRUE(found_cleared_ssrc); + } + webrtc::test::RunLoop main_thread_; // TODO(pbos): Remove playout from all media channels and let renderers mute // themselves. @@ -2060,6 +2136,26 @@ Base::TestRtpHeaderExtensionIdHistoryReassignment(); } +TEST_F(VoiceChannelSingleThreadTest, + TestSetRtpTransportNullClearsReceiveSinks) { + Base::TestSetRtpTransportNullClearsReceiveSinks(); +} + +TEST_F(VoiceChannelSingleThreadTest, + TestRegisterRtpDemuxerSinkPropagatesReceiveSsrcs) { + Base::TestRegisterRtpDemuxerSinkPropagatesReceiveSsrcs(); +} + +TEST_F(VoiceChannelSingleThreadTest, + TestUpdateRemoteStreamsClearsRemovedSinks) { + Base::TestUpdateRemoteStreamsClearsRemovedSinks(); +} + +TEST_F(VoiceChannelSingleThreadTest, + TestUpdateRemoteStreamsClearsUnsignaledSinks) { + Base::TestUpdateRemoteStreamsClearsUnsignaledSinks(); +} + // VoiceChannelDoubleThreadTest TEST_F(VoiceChannelDoubleThreadTest, TestInit) { Base::TestInit(); @@ -2669,6 +2765,26 @@ webrtc::kPacketizationParamRaw)))); } +TEST_F(VideoChannelSingleThreadTest, + TestSetRtpTransportNullClearsReceiveSinks) { + Base::TestSetRtpTransportNullClearsReceiveSinks(); +} + +TEST_F(VideoChannelSingleThreadTest, + TestRegisterRtpDemuxerSinkPropagatesReceiveSsrcs) { + Base::TestRegisterRtpDemuxerSinkPropagatesReceiveSsrcs(); +} + +TEST_F(VideoChannelSingleThreadTest, + TestUpdateRemoteStreamsClearsRemovedSinks) { + Base::TestUpdateRemoteStreamsClearsRemovedSinks(); +} + +TEST_F(VideoChannelSingleThreadTest, + TestUpdateRemoteStreamsClearsUnsignaledSinks) { + Base::TestUpdateRemoteStreamsClearsUnsignaledSinks(); +} + // VideoChannelDoubleThreadTest TEST_F(VideoChannelDoubleThreadTest, TestInit) { Base::TestInit();
diff --git a/pc/test/mock_voice_media_receive_channel_interface.h b/pc/test/mock_voice_media_receive_channel_interface.h index 8156ddf..81a164d 100644 --- a/pc/test/mock_voice_media_receive_channel_interface.h +++ b/pc/test/mock_voice_media_receive_channel_interface.h
@@ -114,6 +114,7 @@ GetUnsignaledSsrc, (), (const, override)); + MOCK_METHOD(std::vector<uint32_t>, GetUnsignaledSsrcs, (), (const, override)); MOCK_METHOD(void, OnDemuxerCriteriaUpdatePending, (), (override)); MOCK_METHOD(void, OnDemuxerCriteriaUpdateComplete, (), (override)); MOCK_METHOD(void,