Capture channel pointers on signaling thread to fix data race The stats collector previously accessed channel getters directly from transceivers during the collection process. This created a data race if a transceiver's channel was cleared on the signaling thread while the collection was still ongoing/queued on the worker thread. This change introduces a callback-based mechanism for retrieving media channel statistics and RTP parameters. By capturing these as absl::AnyInvocable objects on the signaling thread at the start of the stats collection process, the collector can safely execute them later on the worker thread. Bug: webrtc:481443652 Change-Id: Ia8ee334df710bc8655e2617e079bf200ab04173e Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/461260 Reviewed-by: Evan Shrubsole <eshr@webrtc.org> Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47317}
diff --git a/media/base/fake_media_engine.cc b/media/base/fake_media_engine.cc index cad3c58..2603c7b 100644 --- a/media/base/fake_media_engine.cc +++ b/media/base/fake_media_engine.cc
@@ -20,6 +20,7 @@ #include <utility> #include <vector> +#include "absl/functional/any_invocable.h" #include "absl/strings/match.h" #include "api/audio/audio_device.h" #include "api/audio_options.h" @@ -173,6 +174,16 @@ bool /* get_and_clear_legacy_stats */) { return false; } +absl::AnyInvocable<std::optional<VoiceMediaReceiveInfo>()> +FakeVoiceMediaReceiveChannel::GetStatsCallback(bool reset_legacy) { + return [this, reset_legacy]() -> std::optional<VoiceMediaReceiveInfo> { + VoiceMediaReceiveInfo info; + if (GetStats(&info, reset_legacy)) { + return info; + } + return std::nullopt; + }; +} void FakeVoiceMediaReceiveChannel::SetRawAudioSink( uint32_t /* ssrc */, std::unique_ptr<AudioSinkInterface> sink) { @@ -318,6 +329,16 @@ bool FakeVoiceMediaSendChannel::GetStats(VoiceMediaSendInfo* /* info */) { return false; } +absl::AnyInvocable<std::optional<VoiceMediaSendInfo>()> +FakeVoiceMediaSendChannel::GetStatsCallback() { + return [this]() -> std::optional<VoiceMediaSendInfo> { + VoiceMediaSendInfo info; + if (GetStats(&info)) { + return info; + } + return std::nullopt; + }; +} bool FakeVoiceMediaSendChannel::SetSendCodecs( const std::vector<Codec>& codecs) { if (fail_set_send_codecs()) { @@ -419,6 +440,16 @@ bool FakeVideoMediaSendChannel::GetStats(VideoMediaSendInfo* /* info */) { return false; } +absl::AnyInvocable<std::optional<VideoMediaSendInfo>()> +FakeVideoMediaSendChannel::GetStatsCallback() { + return [this]() -> std::optional<VideoMediaSendInfo> { + VideoMediaSendInfo info; + if (GetStats(&info)) { + return info; + } + return std::nullopt; + }; +} bool FakeVideoMediaSendChannel::SetSendCodecs( const std::vector<Codec>& codecs) { if (fail_set_send_codecs()) { @@ -559,6 +590,16 @@ bool FakeVideoMediaReceiveChannel::GetStats(VideoMediaReceiveInfo* /* info */) { return false; } +absl::AnyInvocable<std::optional<VideoMediaReceiveInfo>()> +FakeVideoMediaReceiveChannel::GetStatsCallback() { + return [this]() -> std::optional<VideoMediaReceiveInfo> { + VideoMediaReceiveInfo info; + if (GetStats(&info)) { + return info; + } + return std::nullopt; + }; +} FakeVoiceEngine::FakeVoiceEngine() : encoder_factory_(make_ref_counted<FakeVoiceEncoderFactory>(this)),
diff --git a/media/base/fake_media_engine.h b/media/base/fake_media_engine.h index 4e5e43a..acff856 100644 --- a/media/base/fake_media_engine.h +++ b/media/base/fake_media_engine.h
@@ -337,6 +337,11 @@ } return RtpParameters(); } + + absl::AnyInvocable<RtpParameters(uint32_t)> GetRtpSendParametersCallback() + const override { + return [this](uint32_t ssrc) { return GetRtpSendParameters(ssrc); }; + } RTCError SetRtpSendParameters(uint32_t ssrc, const RtpParameters& parameters, SetParametersCallback callback) override { @@ -519,6 +524,8 @@ bool GetStats(VoiceMediaReceiveInfo* info, bool get_and_clear_legacy_stats) override; + absl::AnyInvocable<std::optional<VoiceMediaReceiveInfo>()> GetStatsCallback( + bool reset_legacy) override; void SetRawAudioSink(uint32_t ssrc, std::unique_ptr<AudioSinkInterface> sink) override; @@ -611,6 +618,8 @@ std::optional<Codec> GetSendCodec() const override; bool GetStats(VoiceMediaSendInfo* stats) override; + absl::AnyInvocable<std::optional<VoiceMediaSendInfo>()> GetStatsCallback() + override; private: class VoiceChannelAudioSink : public AudioSource::Sink { @@ -695,6 +704,8 @@ void ClearRecordableEncodedFrameCallback(uint32_t ssrc) override; void RequestRecvKeyFrame(uint32_t ssrc) override; bool GetStats(VideoMediaReceiveInfo* info) override; + absl::AnyInvocable<std::optional<VideoMediaReceiveInfo>()> GetStatsCallback() + override; bool AddDefaultRecvStreamForTesting(const StreamParams& /* sp */) override { RTC_CHECK_NOTREACHED(); @@ -755,6 +766,8 @@ bool SendCodecHasNack() const override { return false; } bool GetStats(VideoMediaSendInfo* info) override; + absl::AnyInvocable<std::optional<VideoMediaSendInfo>()> GetStatsCallback() + override; bool SetOptions(const VideoOptions& options) override; private:
diff --git a/media/base/media_channel.h b/media/base/media_channel.h index 7075694..95d1e7e 100644 --- a/media/base/media_channel.h +++ b/media/base/media_channel.h
@@ -238,6 +238,18 @@ uint32_t /* ssrc */, VideoEncoderFactory::EncoderSelectorInterface* /* encoder_selector */) {} virtual RtpParameters GetRtpSendParameters(uint32_t ssrc) const = 0; + // Returns a callback that is to be used to retrieve RTP send parameters + // in the same way as `GetRtpSendParameters` does. + // The difference between GetRtpSendParametersCallback and + // `GetRtpSendParameters` is that GetRtpSendParametersCallback is callable + // from the signaling thread and allows binding the `alive` state of the + // channel with the returned callback object. The callback object itself + // however must be called on the worker thread, but does not require the + // caller to have a pointer to the channel. This avoids running into a race + // conditions during asynchronous teardown where signaling thread and worker + // thread state may be torn down asynchronously. + virtual absl::AnyInvocable<RtpParameters(uint32_t)> + GetRtpSendParametersCallback() const = 0; virtual bool SendCodecHasNack() const = 0; // Called whenever the list of sending SSRCs changes. virtual void SetSsrcListChangedCallback( @@ -911,6 +923,13 @@ // DTMF event 0-9, *, #, A-D. virtual bool InsertDtmf(uint32_t ssrc, int event, int duration) = 0; virtual bool GetStats(VoiceMediaSendInfo* stats) = 0; + // Returns a callback that can be used to retrieve stats. + // The purpose is to allow binding state as it exists on the signaling thread + // to a callback that is run on the worker thread. This avoids race conditions + // during asynchronous teardown where signaling thread and worker thread state + // may be torn down asynchronously. + virtual absl::AnyInvocable<std::optional<VoiceMediaSendInfo>()> + GetStatsCallback() = 0; virtual bool SenderNackEnabled() const = 0; virtual bool SenderNonSenderRttEnabled() const = 0; virtual bool SetOptions(const AudioOptions& options) = 0; @@ -936,6 +955,13 @@ virtual void SetDefaultRawAudioSink( std::unique_ptr<AudioSinkInterface> sink) = 0; virtual bool GetStats(VoiceMediaReceiveInfo* stats, bool reset_legacy) = 0; + // Returns a callback that can be used to retrieve stats. + // The purpose is to allow binding state as it exists on the signaling thread + // to a callback that is run on the worker thread. This avoids race conditions + // during asynchronous teardown where signaling thread and worker thread state + // may be torn down asynchronously. + virtual absl::AnyInvocable<std::optional<VoiceMediaReceiveInfo>()> + GetStatsCallback(bool reset_legacy) = 0; virtual enum RtcpMode RtcpMode() const = 0; virtual void SetRtcpMode(enum RtcpMode mode) = 0; virtual void SetReceiveNackEnabled(bool enabled) = 0; @@ -979,6 +1005,13 @@ virtual void GenerateSendKeyFrame(uint32_t ssrc, const std::vector<std::string>& rids) = 0; virtual bool GetStats(VideoMediaSendInfo* stats) = 0; + // Returns a callback that can be used to retrieve stats. + // The purpose is to allow binding state as it exists on the signaling thread + // to a callback that is run on the worker thread. This avoids race conditions + // during asynchronous teardown where signaling thread and worker thread state + // may be torn down asynchronously. + virtual absl::AnyInvocable<std::optional<VideoMediaSendInfo>()> + GetStatsCallback() = 0; // This fills the "bitrate parts" (rtx, video bitrate) of the // BandwidthEstimationInfo, since that part that isn't possible to get // through Call::GetStats, as they are statistics of the send @@ -1016,6 +1049,13 @@ // Clear recordable encoded frame callback for `ssrc` virtual void ClearRecordableEncodedFrameCallback(uint32_t ssrc) = 0; virtual bool GetStats(VideoMediaReceiveInfo* stats) = 0; + // Returns a callback that can be used to retrieve stats. + // The purpose is to allow binding state as it exists on the signaling thread + // to a callback that is run on the worker thread. This avoids race conditions + // during asynchronous teardown where signaling thread and worker thread state + // may be torn down asynchronously. + virtual absl::AnyInvocable<std::optional<VideoMediaReceiveInfo>()> + GetStatsCallback() = 0; virtual bool AddDefaultRecvStreamForTesting(const StreamParams& sp) = 0; };
diff --git a/media/engine/webrtc_video_engine.cc b/media/engine/webrtc_video_engine.cc index 6b3e0b5..172c7a1 100644 --- a/media/engine/webrtc_video_engine.cc +++ b/media/engine/webrtc_video_engine.cc
@@ -1408,6 +1408,16 @@ return rtp_params; } +absl::AnyInvocable<RtpParameters(uint32_t)> +WebRtcVideoSendChannel::GetRtpSendParametersCallback() const { + return [this, safety = task_safety_.flag()]( + uint32_t ssrc) mutable -> RtpParameters { + if (!safety->alive()) + return RtpParameters(); + return GetRtpSendParameters(ssrc); + }; +} + bool WebRtcVideoSendChannel::SetOptions(const VideoOptions& options) { RTC_DCHECK_RUN_ON(&thread_checker_); default_send_options_ = options; @@ -1726,6 +1736,21 @@ } } +absl::AnyInvocable<std::optional<VideoMediaSendInfo>()> +WebRtcVideoSendChannel::GetStatsCallback() { + return [this, safety = task_safety_.flag()]() mutable + -> std::optional<VideoMediaSendInfo> { + if (!safety->alive()) { + return std::nullopt; + } + VideoMediaSendInfo info; + if (GetStats(&info)) { + return info; + } + return std::nullopt; + }; +} + void WebRtcVideoSendChannel::OnPacketSent(const SentPacketInfo& sent_packet) { RTC_DCHECK_RUN_ON(&network_thread_checker_); // TODO(tommi): We shouldn't need to go through call_ to deliver this @@ -3230,6 +3255,21 @@ } } +absl::AnyInvocable<std::optional<VideoMediaReceiveInfo>()> +WebRtcVideoReceiveChannel::GetStatsCallback() { + return [this, safety = task_safety_.flag()]() mutable + -> std::optional<VideoMediaReceiveInfo> { + if (!safety->alive()) { + return std::nullopt; + } + VideoMediaReceiveInfo info; + if (GetStats(&info)) { + return info; + } + return std::nullopt; + }; +} + void WebRtcVideoReceiveChannel::OnPacketReceived(RtpPacketReceived packet) { // Note: the network_thread_checker may refer to the worker thread if the two // threads are combined, but this is either always true or always false
diff --git a/media/engine/webrtc_video_engine.h b/media/engine/webrtc_video_engine.h index b28211a..d497e42 100644 --- a/media/engine/webrtc_video_engine.h +++ b/media/engine/webrtc_video_engine.h
@@ -200,6 +200,8 @@ const RtpParameters& parameters, SetParametersCallback callback) override; RtpParameters GetRtpSendParameters(uint32_t ssrc) const override; + absl::AnyInvocable<RtpParameters(uint32_t)> GetRtpSendParametersCallback() + const override; std::optional<Codec> GetSendCodec() const override; bool SetSend(bool send) override; bool SetVideoSend(uint32_t ssrc, @@ -209,6 +211,8 @@ bool RemoveSendStream(uint32_t ssrc) override; void FillBitrateInfo(BandwidthEstimationInfo* bwe_info) override; bool GetStats(VideoMediaSendInfo* info) override; + absl::AnyInvocable<std::optional<VideoMediaSendInfo>()> GetStatsCallback() + override; void OnPacketSent(const SentPacketInfo& sent_packet) override; void OnReadyToSend(bool ready) override; @@ -531,6 +535,8 @@ bool SetSink(uint32_t ssrc, VideoSinkInterface<VideoFrame>* sink) override; void SetDefaultSink(VideoSinkInterface<VideoFrame>* sink) override; bool GetStats(VideoMediaReceiveInfo* info) override; + absl::AnyInvocable<std::optional<VideoMediaReceiveInfo>()> GetStatsCallback() + override; void OnPacketReceived(RtpPacketReceived packet) override; bool SetBaseMinimumPlayoutDelayMs(uint32_t ssrc, int delay_ms) override; @@ -764,6 +770,7 @@ RTC_GUARDED_BY(thread_checker_); const int receive_buffer_size_; + ScopedTaskSafety task_safety_; }; // Keeping the old name "WebRtcVideoChannel" around because some external
diff --git a/media/engine/webrtc_voice_engine.cc b/media/engine/webrtc_voice_engine.cc index a3c8c98..0448da0 100644 --- a/media/engine/webrtc_voice_engine.cc +++ b/media/engine/webrtc_voice_engine.cc
@@ -1861,6 +1861,21 @@ return true; } +absl::AnyInvocable<std::optional<VoiceMediaSendInfo>()> +WebRtcVoiceSendChannel::GetStatsCallback() { + return [this, safety = task_safety_.flag()]() mutable + -> std::optional<VoiceMediaSendInfo> { + if (!safety->alive()) { + return std::nullopt; + } + VoiceMediaSendInfo info; + if (GetStats(&info)) { + return info; + } + return std::nullopt; + }; +} + bool WebRtcVoiceSendChannel::SenderNackEnabled() const { RTC_DCHECK_RUN_ON(worker_thread_); if (!send_codec_spec_) { @@ -1925,6 +1940,16 @@ return rtp_params; } +absl::AnyInvocable<RtpParameters(uint32_t)> +WebRtcVoiceSendChannel::GetRtpSendParametersCallback() const { + return [this, safety = task_safety_.flag()]( + uint32_t ssrc) mutable -> RtpParameters { + if (!safety->alive()) + return RtpParameters(); + return GetRtpSendParameters(ssrc); + }; +} + RTCError WebRtcVoiceSendChannel::SetRtpSendParameters( uint32_t ssrc, const RtpParameters& parameters, @@ -2779,6 +2804,22 @@ } } +absl::AnyInvocable<std::optional<VoiceMediaReceiveInfo>()> +WebRtcVoiceReceiveChannel::GetStatsCallback(bool get_and_clear_legacy_stats) { + return + [this, get_and_clear_legacy_stats, safety = task_safety_.flag()]() mutable + -> std::optional<VoiceMediaReceiveInfo> { + if (!safety->alive()) { + return std::nullopt; + } + VoiceMediaReceiveInfo info; + if (GetStats(&info, get_and_clear_legacy_stats)) { + return info; + } + return std::nullopt; + }; +} + void WebRtcVoiceReceiveChannel::SetRawAudioSink( uint32_t ssrc, std::unique_ptr<AudioSinkInterface> sink) {
diff --git a/media/engine/webrtc_voice_engine.h b/media/engine/webrtc_voice_engine.h index c38a47f..f033c21 100644 --- a/media/engine/webrtc_voice_engine.h +++ b/media/engine/webrtc_voice_engine.h
@@ -218,6 +218,8 @@ bool SetSenderParameters(const AudioSenderParameter& params) override; RtpParameters GetRtpSendParameters(uint32_t ssrc) const override; + absl::AnyInvocable<RtpParameters(uint32_t)> GetRtpSendParametersCallback() + const override; RTCError SetRtpSendParameters(uint32_t ssrc, const RtpParameters& parameters, SetParametersCallback callback) override; @@ -248,6 +250,8 @@ const NetworkRoute& network_route) override; void OnReadyToSend(bool ready) override; bool GetStats(VoiceMediaSendInfo* info) override; + absl::AnyInvocable<std::optional<VoiceMediaSendInfo>()> GetStatsCallback() + override; bool SetOptions(const AudioOptions& options) override; // Sets a frame transformer between encoder and packetizer, to transform @@ -377,6 +381,8 @@ void OnPacketReceived(RtpPacketReceived packet) override; bool GetStats(VoiceMediaReceiveInfo* info, bool get_and_clear_legacy_stats) override; + absl::AnyInvocable<std::optional<VoiceMediaReceiveInfo>()> GetStatsCallback( + bool get_and_clear_legacy_stats) override; // Set the audio sink for an existing stream. void SetRawAudioSink(uint32_t ssrc,
diff --git a/pc/rtc_stats_collector.cc b/pc/rtc_stats_collector.cc index 8030dd4..ce8c2b3 100644 --- a/pc/rtc_stats_collector.cc +++ b/pc/rtc_stats_collector.cc
@@ -2274,16 +2274,6 @@ std::vector<RtpTransceiverStatsInfo> transceiver_stats_infos; std::vector<TransceiverReferences> transceiver_references; - // These are used to invoke GetStats for all the media channels together in - // one worker thread hop. - std::map<VoiceMediaSendChannelInterface*, VoiceMediaSendInfo> - voice_send_stats; - std::map<VideoMediaSendChannelInterface*, VideoMediaSendInfo> - video_send_stats; - std::map<VoiceMediaReceiveChannelInterface*, VoiceMediaReceiveInfo> - voice_receive_stats; - std::map<VideoMediaReceiveChannelInterface*, VideoMediaReceiveInfo> - video_receive_stats; auto transceivers = pc_->GetTransceiversInternal(); @@ -2319,18 +2309,28 @@ stats.has_receivers = !refs.receivers.empty(); if (stats.has_channel) { + auto* send_channel = transceiver->media_send_channel(); + auto* receive_channel = transceiver->media_receive_channel(); if (stats.media_type == MediaType::AUDIO) { - AddChannelStats(voice_send_stats, - transceiver->voice_media_send_channel()); - AddChannelStats(voice_receive_stats, - transceiver->voice_media_receive_channel()); + VoiceMediaSendChannelInterface* voice_send = + send_channel->AsVoiceSendChannel(); + RTC_CHECK(voice_send); + VoiceMediaReceiveChannelInterface* voice_receive = + receive_channel->AsVoiceReceiveChannel(); + RTC_CHECK(voice_receive); + refs.get_send_stats_voice = voice_send->GetStatsCallback(); + refs.get_receive_stats_voice = voice_receive->GetStatsCallback(false); + refs.get_send_parameters = voice_send->GetRtpSendParametersCallback(); } else if (stats.media_type == MediaType::VIDEO) { - AddChannelStats(video_send_stats, - transceiver->video_media_send_channel()); - AddChannelStats(video_receive_stats, - transceiver->video_media_receive_channel()); - } else { - RTC_DCHECK_NOTREACHED(); + VideoMediaSendChannelInterface* video_send = + send_channel->AsVideoSendChannel(); + RTC_CHECK(video_send); + VideoMediaReceiveChannelInterface* video_receive = + receive_channel->AsVideoReceiveChannel(); + RTC_CHECK(video_receive); + refs.get_send_stats_video = video_send->GetStatsCallback(); + refs.get_receive_stats_video = video_receive->GetStatsCallback(); + refs.get_send_parameters = video_send->GetRtpSendParametersCallback(); } } @@ -2343,11 +2343,8 @@ // as GetCallStats(). At the same time we construct the TrackMediaInfoMaps, // which also needs info from the worker thread. return [this, transceiver_stats_infos = std::move(transceiver_stats_infos), - transceiver_references = std::move(transceiver_references), - voice_send_stats = std::move(voice_send_stats), - voice_receive_stats = std::move(voice_receive_stats), - video_send_stats = std::move(video_send_stats), - video_receive_stats = std::move(video_receive_stats)]() mutable { + transceiver_references = + std::move(transceiver_references)]() mutable { Thread::ScopedDisallowBlockingCalls no_blocking_calls; WorkerThreadResult worker_result; @@ -2355,28 +2352,6 @@ std::move(transceiver_stats_infos); worker_result.transceiver_references = std::move(transceiver_references); - for (auto& pair : voice_send_stats) { - if (!pair.first->GetStats(&pair.second)) { - RTC_LOG(LS_WARNING) << "Failed to get voice send stats."; - } - } - for (auto& pair : voice_receive_stats) { - if (!pair.first->GetStats(&pair.second, - /*get_and_clear_legacy_stats=*/false)) { - RTC_LOG(LS_WARNING) << "Failed to get voice receive stats."; - } - } - for (auto& pair : video_send_stats) { - if (!pair.first->GetStats(&pair.second)) { - RTC_LOG(LS_WARNING) << "Failed to get video send stats."; - } - } - for (auto& pair : video_receive_stats) { - if (!pair.first->GetStats(&pair.second)) { - RTC_LOG(LS_WARNING) << "Failed to get video receive stats."; - } - } - // Create the TrackMediaInfoMap for each transceiver stats object // and keep track of whether we have at least one audio receiver. bool has_audio_receiver = false; @@ -2396,16 +2371,7 @@ for (const auto& sender : stats.sender_infos) { RtpParameters params; if (sender.ssrc != 0 && stats.has_channel) { - auto& transceiver = refs.transceiver; - MediaSendChannelInterface* media_channel = - (stats.media_type == MediaType::AUDIO) - ? static_cast<MediaSendChannelInterface*>( - transceiver->voice_media_send_channel()) - : static_cast<MediaSendChannelInterface*>( - transceiver->video_media_send_channel()); - if (media_channel) { - params = media_channel->GetRtpSendParameters(sender.ssrc); - } + params = refs.get_send_parameters(sender.ssrc); } // `sender_parameters` must be the same size as `stats.sender_infos` to // allow 1:1 mapping when TrackMediaInfoMap is constructed. Empty params @@ -2421,21 +2387,22 @@ std::optional<VoiceMediaInfo> voice_media_info; std::optional<VideoMediaInfo> video_media_info; if (stats.has_channel) { - auto& transceiver = refs.transceiver; if (stats.media_type == MediaType::AUDIO) { - auto voice_send_channel = transceiver->voice_media_send_channel(); - auto voice_receive_channel = - transceiver->voice_media_receive_channel(); + std::optional<VoiceMediaSendInfo> voice_send = + refs.get_send_stats_voice(); + std::optional<VoiceMediaReceiveInfo> voice_receive = + refs.get_receive_stats_voice(); voice_media_info = VoiceMediaInfo( - std::move(voice_send_stats[voice_send_channel]), - std::move(voice_receive_stats[voice_receive_channel])); + std::move(voice_send).value_or(VoiceMediaSendInfo{}), + std::move(voice_receive).value_or(VoiceMediaReceiveInfo{})); } else if (stats.media_type == MediaType::VIDEO) { - auto video_send_channel = transceiver->video_media_send_channel(); - auto video_receive_channel = - transceiver->video_media_receive_channel(); + std::optional<VideoMediaSendInfo> video_send = + refs.get_send_stats_video(); + std::optional<VideoMediaReceiveInfo> video_receive = + refs.get_receive_stats_video(); video_media_info = VideoMediaInfo( - std::move(video_send_stats[video_send_channel]), - std::move(video_receive_stats[video_receive_channel])); + std::move(video_send).value_or(VideoMediaSendInfo{}), + std::move(video_receive).value_or(VideoMediaReceiveInfo{})); } }
diff --git a/pc/rtc_stats_collector.h b/pc/rtc_stats_collector.h index c873006..f20a67d 100644 --- a/pc/rtc_stats_collector.h +++ b/pc/rtc_stats_collector.h
@@ -35,6 +35,7 @@ #include "api/task_queue/task_queue_base.h" #include "api/units/timestamp.h" #include "call/call.h" +#include "media/base/media_channel.h" #include "pc/data_channel_utils.h" #include "pc/peer_connection_internal.h" #include "pc/rtp_receiver.h" @@ -78,6 +79,13 @@ struct TransceiverReferences { scoped_refptr<RtpTransceiver> transceiver; std::vector<scoped_refptr<RtpReceiverInternal>> receivers; + absl::AnyInvocable<std::optional<VoiceMediaSendInfo>()> get_send_stats_voice; + absl::AnyInvocable<std::optional<VideoMediaSendInfo>()> get_send_stats_video; + absl::AnyInvocable<std::optional<VoiceMediaReceiveInfo>()> + get_receive_stats_voice; + absl::AnyInvocable<std::optional<VideoMediaReceiveInfo>()> + get_receive_stats_video; + absl::AnyInvocable<RtpParameters(uint32_t ssrc)> get_send_parameters; }; // All public methods of the collector are to be called on the signaling thread.
diff --git a/pc/rtc_stats_collector_unittest.cc b/pc/rtc_stats_collector_unittest.cc index 35905ed..18b4df2 100644 --- a/pc/rtc_stats_collector_unittest.cc +++ b/pc/rtc_stats_collector_unittest.cc
@@ -371,6 +371,7 @@ }); EXPECT_CALL(*sender, AttachmentId()).WillRepeatedly(Return(attachment_id)); EXPECT_CALL(*sender, Stop()).WillRepeatedly(Return()); + EXPECT_CALL(*sender, SetSendCodecs(_)).WillRepeatedly(Return()); RTC_ALLOW_PLAN_B_DEPRECATION_BEGIN(); pc_->AddSender(sender); RTC_ALLOW_PLAN_B_DEPRECATION_END(); @@ -887,6 +888,70 @@ EXPECT_NE(c.get(), b.get()); } +TEST_P(RTCStatsCollectorTest, StatsPreservedWhenChannelClearedConcurrently) { + VoiceMediaInfo voice_media_info; + voice_media_info.receivers.emplace_back(); + voice_media_info.receivers[0].add_ssrc(1); + voice_media_info.receivers[0].packets_received = 123; + + RTC_ALLOW_PLAN_B_DEPRECATION_BEGIN(); + pc_->AddVoiceChannel("AudioMid", "TransportName", voice_media_info); + RTC_ALLOW_PLAN_B_DEPRECATION_END(); + auto remote_receiver = stats_->SetupRemoteTrackAndReceiver( + MediaType::AUDIO, "RemoteAudioTrackID", "RemoteStreamId", 1); + auto local_sender = stats_->SetupLocalTrackAndSender( + MediaType::AUDIO, "LocalAudioTrackID", 2, + /*add_stream=*/true, /*attachment_id=*/2); + + EXPECT_CALL(*local_sender, SetSendCodecs(_)).WillRepeatedly(Return()); + EXPECT_CALL(*local_sender, SetMediaChannel(_)).WillRepeatedly(Return()); + EXPECT_CALL(*local_sender, DetachTrackAndGetStopTask()).WillRepeatedly([] { + return nullptr; + }); + EXPECT_CALL(*remote_receiver, SetMediaChannel(_)).WillRepeatedly(Return()); + + scoped_refptr<const RTCStatsReport> report; + bool stats_obtained = false; + stats_->stats_collector().GetStatsReport( + RTCStatsObtainer::Create(&report, [&] { + stats_obtained = true; + main_thread_.Quit(); + })); + + auto transceivers = pc_->GetTransceiversInternal(); + ASSERT_FALSE(transceivers.empty()); + auto* transceiver = transceivers[0]->internal(); + EXPECT_TRUE(transceiver->HasChannel()); + + auto network_task = transceiver->GetClearChannelNetworkTask(); + pc_->network_thread()->BlockingCall([&]() { std::move(network_task)(); }); + + auto delete_worker_task = + transceiver->GetDeleteChannelWorkerTask(/*stop_senders=*/false); + // Now the state of the transceiver is that on the signaling thread, + // there is no channel. However, the work of actually deleting the channels + // belongs to the worker thread task, which hasn't run yet. + // Let's post the execution of that task so that it ends up running + // *after* the task that the `stats_collector` has queued up for + // gathering the stats from the channel. The stats collector should + // be able to get the stats safely even though technically the + // transceiver by now, doesn't have a channel. + EXPECT_FALSE(transceiver->HasChannel()); + Event worker_task_done; + pc_->worker_thread()->PostTask([&]() { + std::move(delete_worker_task)(); + worker_task_done.Set(); + }); + + main_thread_.Run(); + + EXPECT_TRUE(report); + auto inbound_rtps = report->GetStatsOfType<RTCInboundRtpStreamStats>(); + EXPECT_EQ(inbound_rtps.size(), 1u); + // Wait for the cleanup task to complete. + worker_task_done.Wait(Event::kForever); +} + TEST_P(RTCStatsCollectorTest, ToJsonProducesParseableJson) { ExampleStatsGraph graph = SetupExampleStatsGraphForSelectorTests(); scoped_refptr<const RTCStatsReport> report =
diff --git a/pc/test/fake_peer_connection_for_stats.h b/pc/test/fake_peer_connection_for_stats.h index 1e38082..c2f0f07 100644 --- a/pc/test/fake_peer_connection_for_stats.h +++ b/pc/test/fake_peer_connection_for_stats.h
@@ -20,6 +20,7 @@ #include <utility> #include <vector> +#include "absl/functional/any_invocable.h" #include "absl/strings/string_view.h" #include "api/audio/audio_device.h" #include "api/audio_options.h" @@ -35,6 +36,7 @@ #include "api/rtp_transceiver_direction.h" #include "api/scoped_refptr.h" #include "api/sequence_checker.h" +#include "api/task_queue/pending_task_safety_flag.h" #include "api/task_queue/task_queue_base.h" #include "api/transport/data_channel_transport_interface.h" #include "call/call.h" @@ -111,7 +113,23 @@ return false; } + absl::AnyInvocable<std::optional<VoiceMediaSendInfo>()> GetStatsCallback() + override { + return [this, safety = task_safety_.flag()]() mutable + -> std::optional<VoiceMediaSendInfo> { + if (!safety->alive()) { + return std::nullopt; + } + VoiceMediaSendInfo info; + if (GetStats(&info)) { + return info; + } + return std::nullopt; + }; + } + private: + ScopedTaskSafety task_safety_; std::optional<VoiceMediaSendInfo> send_stats_; }; @@ -138,7 +156,23 @@ return false; } + absl::AnyInvocable<std::optional<VoiceMediaReceiveInfo>()> GetStatsCallback( + bool get_and_clear_legacy_stats) override { + return [this, safety = task_safety_.flag()]() mutable + -> std::optional<VoiceMediaReceiveInfo> { + if (!safety->alive()) { + return std::nullopt; + } + VoiceMediaReceiveInfo info; + if (GetStats(&info, /*get_and_clear_legacy_stats=*/false)) { + return info; + } + return std::nullopt; + }; + } + private: + ScopedTaskSafety task_safety_; std::optional<VoiceMediaReceiveInfo> receive_stats_; }; @@ -164,7 +198,23 @@ return false; } + absl::AnyInvocable<std::optional<VideoMediaSendInfo>()> GetStatsCallback() + override { + return [this, safety = task_safety_.flag()]() mutable + -> std::optional<VideoMediaSendInfo> { + if (!safety->alive()) { + return std::nullopt; + } + VideoMediaSendInfo info; + if (GetStats(&info)) { + return info; + } + return std::nullopt; + }; + } + private: + ScopedTaskSafety task_safety_; std::optional<VideoMediaSendInfo> send_stats_; }; @@ -189,7 +239,23 @@ return false; } + absl::AnyInvocable<std::optional<VideoMediaReceiveInfo>()> GetStatsCallback() + override { + return [this, safety = task_safety_.flag()]() mutable + -> std::optional<VideoMediaReceiveInfo> { + if (!safety->alive()) { + return std::nullopt; + } + VideoMediaReceiveInfo info; + if (GetStats(&info)) { + return info; + } + return std::nullopt; + }; + } + private: + ScopedTaskSafety task_safety_; std::optional<VideoMediaReceiveInfo> receive_stats_; };
diff --git a/pc/test/mock_voice_media_receive_channel_interface.h b/pc/test/mock_voice_media_receive_channel_interface.h index 1c1f5e3..feac0ee 100644 --- a/pc/test/mock_voice_media_receive_channel_interface.h +++ b/pc/test/mock_voice_media_receive_channel_interface.h
@@ -16,6 +16,7 @@ #include <type_traits> #include <vector> +#include "absl/functional/any_invocable.h" #include "api/audio_options.h" #include "api/call/audio_sink.h" #include "api/crypto/frame_decryptor_interface.h" @@ -74,6 +75,10 @@ GetStats, (webrtc::VoiceMediaReceiveInfo * stats, bool reset_legacy), (override)); + MOCK_METHOD(absl::AnyInvocable<std::optional<VoiceMediaReceiveInfo>()>, + GetStatsCallback, + (bool reset_legacy), + (override)); MOCK_METHOD(::webrtc::RtcpMode, RtcpMode, (), (const, override)); MOCK_METHOD(void, SetRtcpMode, (::webrtc::RtcpMode mode), (override)); MOCK_METHOD(void, SetReceiveNackEnabled, (bool enabled), (override));