diff --git a/pc/channel.cc b/pc/channel.cc index aad7c54381..34269a13b0 100644 --- a/pc/channel.cc +++ b/pc/channel.cc @@ -170,7 +170,9 @@ std::string BaseChannel::ToString() const { bool BaseChannel::ConnectToRtpTransport() { RTC_DCHECK_RUN_ON(network_thread()); RTC_DCHECK(rtp_transport_); - if (!RegisterRtpDemuxerSink_n()) { + // TODO(bugs.webrtc.org/12230): This accesses demuxer_criteria_ on the + // networking thread. + if (!rtp_transport_->RegisterRtpDemuxerSink(demuxer_criteria_, this)) { RTC_LOG(LS_ERROR) << "Failed to set up demuxing for " << ToString(); return false; } @@ -291,13 +293,42 @@ bool BaseChannel::SetRemoteContent(const MediaContentDescription* content, Bind(&BaseChannel::SetRemoteContent_w, this, content, type, error_desc)); } -bool BaseChannel::SetPayloadTypeDemuxingEnabled(bool enabled) { +void BaseChannel::SetPayloadTypeDemuxingEnabled(bool enabled) { TRACE_EVENT0("webrtc", "BaseChannel::SetPayloadTypeDemuxingEnabled"); - return InvokeOnWorker( + InvokeOnWorker( RTC_FROM_HERE, Bind(&BaseChannel::SetPayloadTypeDemuxingEnabled_w, this, enabled)); } +bool BaseChannel::UpdateRtpTransport(std::string* error_desc) { + return network_thread_->Invoke(RTC_FROM_HERE, [this, error_desc] { + RTC_DCHECK_RUN_ON(network_thread()); + RTC_DCHECK(rtp_transport_); + // TODO(bugs.webrtc.org/12230): This accesses demuxer_criteria_ on the + // networking thread. + if (!rtp_transport_->RegisterRtpDemuxerSink(demuxer_criteria_, this)) { + RTC_LOG(LS_ERROR) << "Failed to set up demuxing for " << ToString(); + rtc::StringBuilder desc; + desc << "Failed to set up demuxing for m-section with mid='" + << content_name() << "'."; + SafeSetError(desc.str(), error_desc); + return false; + } + // NOTE: This doesn't take the BUNDLE case in account meaning the RTP header + // extension maps are not merged when BUNDLE is enabled. This is fine + // because the ID for MID should be consistent among all the RTP transports, + // and that's all RtpTransport uses this map for. + // + // TODO(deadbeef): Move this call to JsepTransport, there is no reason + // BaseChannel needs to be involved here. + if (media_type() != cricket::MEDIA_TYPE_DATA) { + rtp_transport_->UpdateRtpHeaderExtensionMap( + receive_rtp_header_extensions_); + } + return true; + }); +} + bool BaseChannel::IsReadyToReceiveMedia_w() const { // Receive data if we are enabled and have local content, return enabled() && @@ -305,12 +336,6 @@ bool BaseChannel::IsReadyToReceiveMedia_w() const { } bool BaseChannel::IsReadyToSendMedia_w() const { - // Need to access some state updated on the network thread. - return network_thread_->Invoke( - RTC_FROM_HERE, Bind(&BaseChannel::IsReadyToSendMedia_n, this)); -} - -bool BaseChannel::IsReadyToSendMedia_n() const { // Send outgoing data if we are enabled, have local and remote content, // and we have had some form of connectivity. return enabled() && @@ -508,38 +533,6 @@ void BaseChannel::OnRtpPacket(const webrtc::RtpPacketReceived& parsed_packet) { }); } -void BaseChannel::UpdateRtpHeaderExtensionMap( - const RtpHeaderExtensions& header_extensions) { - // Update the header extension map on network thread in case there is data - // race. - // - // NOTE: This doesn't take the BUNDLE case in account meaning the RTP header - // extension maps are not merged when BUNDLE is enabled. This is fine because - // the ID for MID should be consistent among all the RTP transports. - network_thread_->Invoke(RTC_FROM_HERE, [this, &header_extensions] { - RTC_DCHECK_RUN_ON(network_thread()); - rtp_transport_->UpdateRtpHeaderExtensionMap(header_extensions); - }); -} - -bool BaseChannel::RegisterRtpDemuxerSink_w() { - // Copy demuxer criteria, since they're a worker-thread variable - // and we want to pass them to the network thread - return network_thread_->Invoke( - RTC_FROM_HERE, [this, demuxer_criteria = demuxer_criteria_] { - RTC_DCHECK_RUN_ON(network_thread()); - RTC_DCHECK(rtp_transport_); - return rtp_transport_->RegisterRtpDemuxerSink(demuxer_criteria, this); - }); -} - -bool BaseChannel::RegisterRtpDemuxerSink_n() { - RTC_DCHECK(rtp_transport_); - // TODO(bugs.webrtc.org/12230): This accesses demuxer_criteria_ on the - // networking thread. - return rtp_transport_->RegisterRtpDemuxerSink(demuxer_criteria_, this); -} - void BaseChannel::EnableMedia_w() { RTC_DCHECK(worker_thread_ == rtc::Thread::Current()); if (enabled_) @@ -573,22 +566,28 @@ void BaseChannel::ChannelWritable_n() { if (writable_) { return; } + writable_ = true; RTC_LOG(LS_INFO) << "Channel writable (" << ToString() << ")" - << (was_ever_writable_ ? "" : " for the first time"); - - was_ever_writable_ = true; - writable_ = true; - UpdateMediaSendRecvState(); + << (was_ever_writable_n_ ? "" : " for the first time"); + // We only have to do this AsyncInvoke once, when first transitioning to + // writable. + if (!was_ever_writable_n_) { + invoker_.AsyncInvoke(RTC_FROM_HERE, worker_thread_, [this] { + RTC_DCHECK_RUN_ON(worker_thread()); + was_ever_writable_ = true; + UpdateMediaSendRecvState_w(); + }); + } + was_ever_writable_n_ = true; } void BaseChannel::ChannelNotWritable_n() { - if (!writable_) + if (!writable_) { return; - - RTC_LOG(LS_INFO) << "Channel not writable (" << ToString() << ")"; + } writable_ = false; - UpdateMediaSendRecvState(); + RTC_LOG(LS_INFO) << "Channel not writable (" << ToString() << ")"; } bool BaseChannel::AddRecvStream_w(const StreamParams& sp) { @@ -604,9 +603,9 @@ void BaseChannel::ResetUnsignaledRecvStream_w() { media_channel()->ResetUnsignaledRecvStream(); } -bool BaseChannel::SetPayloadTypeDemuxingEnabled_w(bool enabled) { +void BaseChannel::SetPayloadTypeDemuxingEnabled_w(bool enabled) { if (enabled == payload_type_demuxing_enabled_) { - return true; + return; } payload_type_demuxing_enabled_ = enabled; if (!enabled) { @@ -617,21 +616,10 @@ bool BaseChannel::SetPayloadTypeDemuxingEnabled_w(bool enabled) { // there is no straightforward way to identify those streams. media_channel()->ResetUnsignaledRecvStream(); demuxer_criteria_.payload_types.clear(); - if (!RegisterRtpDemuxerSink_w()) { - RTC_LOG(LS_ERROR) << "Failed to disable payload type demuxing for " - << ToString(); - return false; - } } else if (!payload_types_.empty()) { demuxer_criteria_.payload_types.insert(payload_types_.begin(), payload_types_.end()); - if (!RegisterRtpDemuxerSink_w()) { - RTC_LOG(LS_ERROR) << "Failed to enable payload type demuxing for " - << ToString(); - return false; - } } - return true; } bool BaseChannel::UpdateLocalStreams_w(const std::vector& streams, @@ -772,11 +760,6 @@ bool BaseChannel::UpdateRemoteStreams_w( demuxer_criteria_.ssrcs.insert(new_stream.ssrcs.begin(), new_stream.ssrcs.end()); } - // Re-register the sink to update the receiving ssrcs. - if (!RegisterRtpDemuxerSink_w()) { - RTC_LOG(LS_ERROR) << "Failed to set up demuxing for " << ToString(); - ret = false; - } remote_streams_ = streams; return ret; } @@ -795,6 +778,10 @@ RtpHeaderExtensions BaseChannel::GetFilteredRtpHeaderExtensions( return webrtc::RtpExtension::FilterDuplicateNonEncrypted(extensions); } +void BaseChannel::SetReceiveExtensions(const RtpHeaderExtensions& extensions) { + receive_rtp_header_extensions_ = extensions; +} + void BaseChannel::OnMessage(rtc::Message* pmsg) { TRACE_EVENT0("webrtc", "BaseChannel::OnMessage"); switch (pmsg->message_id) { @@ -873,12 +860,6 @@ VoiceChannel::~VoiceChannel() { Deinit(); } -void BaseChannel::UpdateMediaSendRecvState() { - RTC_DCHECK_RUN_ON(network_thread()); - invoker_.AsyncInvoke(RTC_FROM_HERE, worker_thread_, - [this] { UpdateMediaSendRecvState_w(); }); -} - void VoiceChannel::Init_w(webrtc::RtpTransportInternal* rtp_transport) { BaseChannel::Init_w(rtp_transport); } @@ -916,7 +897,7 @@ bool VoiceChannel::SetLocalContent_w(const MediaContentDescription* content, RtpHeaderExtensions rtp_header_extensions = GetFilteredRtpHeaderExtensions(audio->rtp_header_extensions()); - UpdateRtpHeaderExtensionMap(rtp_header_extensions); + SetReceiveExtensions(rtp_header_extensions); media_channel()->SetExtmapAllowMixed(audio->extmap_allow_mixed()); AudioRecvParameters recv_params = last_recv_params_; @@ -936,11 +917,6 @@ bool VoiceChannel::SetLocalContent_w(const MediaContentDescription* content, for (const AudioCodec& codec : audio->codecs()) { MaybeAddHandledPayloadType(codec.id); } - // Need to re-register the sink to update the handled payload. - if (!RegisterRtpDemuxerSink_w()) { - RTC_LOG(LS_ERROR) << "Failed to set up audio demuxing for " << ToString(); - return false; - } } last_recv_params_ = recv_params; @@ -1003,10 +979,6 @@ bool VoiceChannel::SetRemoteContent_w(const MediaContentDescription* content, "disable payload type demuxing for " << ToString(); ClearHandledPayloadTypes(); - if (!RegisterRtpDemuxerSink_w()) { - RTC_LOG(LS_ERROR) << "Failed to update audio demuxing for " << ToString(); - return false; - } } // TODO(pthatcher): Move remote streams into AudioRecvParameters, @@ -1087,7 +1059,7 @@ bool VideoChannel::SetLocalContent_w(const MediaContentDescription* content, RtpHeaderExtensions rtp_header_extensions = GetFilteredRtpHeaderExtensions(video->rtp_header_extensions()); - UpdateRtpHeaderExtensionMap(rtp_header_extensions); + SetReceiveExtensions(rtp_header_extensions); media_channel()->SetExtmapAllowMixed(video->extmap_allow_mixed()); VideoRecvParameters recv_params = last_recv_params_; @@ -1130,11 +1102,6 @@ bool VideoChannel::SetLocalContent_w(const MediaContentDescription* content, for (const VideoCodec& codec : video->codecs()) { MaybeAddHandledPayloadType(codec.id); } - // Need to re-register the sink to update the handled payload. - if (!RegisterRtpDemuxerSink_w()) { - RTC_LOG(LS_ERROR) << "Failed to set up video demuxing for " << ToString(); - return false; - } } last_recv_params_ = recv_params; @@ -1241,10 +1208,6 @@ bool VideoChannel::SetRemoteContent_w(const MediaContentDescription* content, "disable payload type demuxing for " << ToString(); ClearHandledPayloadTypes(); - if (!RegisterRtpDemuxerSink_w()) { - RTC_LOG(LS_ERROR) << "Failed to update video demuxing for " << ToString(); - return false; - } } // TODO(pthatcher): Move remote streams into VideoRecvParameters, @@ -1355,11 +1318,6 @@ bool RtpDataChannel::SetLocalContent_w(const MediaContentDescription* content, for (const DataCodec& codec : data->codecs()) { MaybeAddHandledPayloadType(codec.id); } - // Need to re-register the sink to update the handled payload. - if (!RegisterRtpDemuxerSink_w()) { - RTC_LOG(LS_ERROR) << "Failed to set up data demuxing for " << ToString(); - return false; - } last_recv_params_ = recv_params; diff --git a/pc/channel.h b/pc/channel.h index 1fb2a3978c..0ba23eb425 100644 --- a/pc/channel.h +++ b/pc/channel.h @@ -119,9 +119,6 @@ class BaseChannel : public ChannelInterface, RTC_DCHECK_RUN_ON(network_thread()); return srtp_active(); } - - bool writable() const { return writable_; } - // Set an RTP level transport which could be an RtpTransport without // encryption, an SrtpTransport for SDES or a DtlsSrtpTransport for DTLS-SRTP. // This can be called from any thread and it hops to the network thread @@ -143,7 +140,8 @@ class BaseChannel : public ChannelInterface, return rtp_transport(); } - // Channel control + // Channel control. Must call UpdateRtpTransport afterwards to apply any + // changes to the RtpTransport on the network thread. bool SetLocalContent(const MediaContentDescription* content, webrtc::SdpType type, std::string* error_desc) override; @@ -158,7 +156,11 @@ class BaseChannel : public ChannelInterface, // This method will also remove any existing streams that were bound to this // channel on the basis of payload type, since one of these streams might // actually belong to a new channel. See: crbug.com/webrtc/11477 - bool SetPayloadTypeDemuxingEnabled(bool enabled) override; + // + // As with SetLocalContent/SetRemoteContent, must call UpdateRtpTransport + // afterwards to apply changes to the RtpTransport on the network thread. + void SetPayloadTypeDemuxingEnabled(bool enabled) override; + bool UpdateRtpTransport(std::string* error_desc) override; bool Enable(bool enable) override; @@ -198,7 +200,7 @@ class BaseChannel : public ChannelInterface, protected: bool was_ever_writable() const { - RTC_DCHECK_RUN_ON(network_thread()); + RTC_DCHECK_RUN_ON(worker_thread()); return was_ever_writable_; } void set_local_content_direction(webrtc::RtpTransceiverDirection direction) { @@ -256,7 +258,7 @@ class BaseChannel : public ChannelInterface, bool AddRecvStream_w(const StreamParams& sp) RTC_RUN_ON(worker_thread()); bool RemoveRecvStream_w(uint32_t ssrc) RTC_RUN_ON(worker_thread()); void ResetUnsignaledRecvStream_w() RTC_RUN_ON(worker_thread()); - bool SetPayloadTypeDemuxingEnabled_w(bool enabled) + void SetPayloadTypeDemuxingEnabled_w(bool enabled) RTC_RUN_ON(worker_thread()); bool AddSendStream_w(const StreamParams& sp) RTC_RUN_ON(worker_thread()); bool RemoveSendStream_w(uint32_t ssrc) RTC_RUN_ON(worker_thread()); @@ -264,7 +266,6 @@ class BaseChannel : public ChannelInterface, // Should be called whenever the conditions for // IsReadyToReceiveMedia/IsReadyToSendMedia are satisfied (or unsatisfied). // Updates the send/recv state of the media channel. - void UpdateMediaSendRecvState(); virtual void UpdateMediaSendRecvState_w() = 0; bool UpdateLocalStreams_w(const std::vector& streams, @@ -286,6 +287,9 @@ class BaseChannel : public ChannelInterface, // non-encrypted and encrypted extension is present for the same URI. RtpHeaderExtensions GetFilteredRtpHeaderExtensions( const RtpHeaderExtensions& extensions); + // Set a list of RTP extensions we should prepare to receive on the next + // UpdateRtpTransport call. + void SetReceiveExtensions(const RtpHeaderExtensions& extensions); // From MessageHandler void OnMessage(rtc::Message* pmsg) override; @@ -302,13 +306,6 @@ class BaseChannel : public ChannelInterface, void MaybeAddHandledPayloadType(int payload_type) RTC_RUN_ON(worker_thread()); void ClearHandledPayloadTypes() RTC_RUN_ON(worker_thread()); - - void UpdateRtpHeaderExtensionMap( - const RtpHeaderExtensions& header_extensions); - - bool RegisterRtpDemuxerSink_w() RTC_RUN_ON(worker_thread()); - bool RegisterRtpDemuxerSink_n() RTC_RUN_ON(network_thread()); - // Return description of media channel to facilitate logging std::string ToString() const; @@ -319,7 +316,6 @@ class BaseChannel : public ChannelInterface, void DisconnectFromRtpTransport(); void SignalSentPacket_n(const rtc::SentPacket& sent_packet) RTC_RUN_ON(network_thread()); - bool IsReadyToSendMedia_n() const RTC_RUN_ON(network_thread()); rtc::Thread* const worker_thread_; rtc::Thread* const network_thread_; @@ -344,10 +340,9 @@ class BaseChannel : public ChannelInterface, RTC_GUARDED_BY(network_thread()); std::vector > rtcp_socket_options_ RTC_GUARDED_BY(network_thread()); - // TODO(bugs.webrtc.org/12230): writable_ is accessed in tests - // outside of the network thread. - bool writable_ = false; - bool was_ever_writable_ RTC_GUARDED_BY(network_thread()) = false; + bool writable_ RTC_GUARDED_BY(network_thread()) = false; + bool was_ever_writable_n_ RTC_GUARDED_BY(network_thread()) = false; + bool was_ever_writable_ RTC_GUARDED_BY(worker_thread()) = false; const bool srtp_required_ = true; const webrtc::CryptoOptions crypto_options_; @@ -371,9 +366,10 @@ class BaseChannel : public ChannelInterface, // Cached list of payload types, used if payload type demuxing is re-enabled. std::set payload_types_ RTC_GUARDED_BY(worker_thread()); - // TODO(bugs.webrtc.org/12239): Modified on worker thread, accessed - // on network thread in RegisterRtpDemuxerSink_n (called from Init_w) + // TODO(bugs.webrtc.org/12239): These two variables are modified on the worker + // thread, accessed on the network thread in UpdateRtpTransport. webrtc::RtpDemuxerCriteria demuxer_criteria_; + RtpHeaderExtensions receive_rtp_header_extensions_; // This generator is used to generate SSRCs for local streams. // This is needed in cases where SSRCs are not negotiated or set explicitly // like in Simulcast. diff --git a/pc/channel_interface.h b/pc/channel_interface.h index 68b6486304..4580a2fd60 100644 --- a/pc/channel_interface.h +++ b/pc/channel_interface.h @@ -52,7 +52,8 @@ class ChannelInterface { virtual bool SetRemoteContent(const MediaContentDescription* content, webrtc::SdpType type, std::string* error_desc) = 0; - virtual bool SetPayloadTypeDemuxingEnabled(bool enabled) = 0; + virtual void SetPayloadTypeDemuxingEnabled(bool enabled) = 0; + virtual bool UpdateRtpTransport(std::string* error_desc) = 0; // Access to the local and remote streams that were set on the channel. virtual const std::vector& local_streams() const = 0; diff --git a/pc/channel_unittest.cc b/pc/channel_unittest.cc index c4071475d0..fb62b08df5 100644 --- a/pc/channel_unittest.cc +++ b/pc/channel_unittest.cc @@ -323,19 +323,26 @@ class ChannelTest : public ::testing::Test, public sigslot::has_slots<> { fake_rtcp_packet_transport2_.get(), asymmetric); } }); + // The transport becoming writable will asynchronously update the send state + // on the worker thread; since this test uses the main thread as the worker + // thread, we must process the message queue for this to occur. + WaitForThreads(); } bool SendInitiate() { bool result = channel1_->SetLocalContent(&local_media_content1_, - SdpType::kOffer, NULL); + SdpType::kOffer, NULL) && + channel1_->UpdateRtpTransport(nullptr); if (result) { channel1_->Enable(true); result = channel2_->SetRemoteContent(&remote_media_content1_, - SdpType::kOffer, NULL); + SdpType::kOffer, NULL) && + channel2_->UpdateRtpTransport(nullptr); if (result) { ConnectFakeTransports(); result = channel2_->SetLocalContent(&local_media_content2_, - SdpType::kAnswer, NULL); + SdpType::kAnswer, NULL) && + channel2_->UpdateRtpTransport(nullptr); } } return result; @@ -344,27 +351,32 @@ class ChannelTest : public ::testing::Test, public sigslot::has_slots<> { bool SendAccept() { channel2_->Enable(true); return channel1_->SetRemoteContent(&remote_media_content2_, - SdpType::kAnswer, NULL); + SdpType::kAnswer, NULL) && + channel1_->UpdateRtpTransport(nullptr); } bool SendOffer() { bool result = channel1_->SetLocalContent(&local_media_content1_, - SdpType::kOffer, NULL); + SdpType::kOffer, NULL) && + channel1_->UpdateRtpTransport(nullptr); if (result) { channel1_->Enable(true); result = channel2_->SetRemoteContent(&remote_media_content1_, - SdpType::kOffer, NULL); + SdpType::kOffer, NULL) && + channel2_->UpdateRtpTransport(nullptr); } return result; } bool SendProvisionalAnswer() { bool result = channel2_->SetLocalContent(&local_media_content2_, - SdpType::kPrAnswer, NULL); + SdpType::kPrAnswer, NULL) && + channel2_->UpdateRtpTransport(nullptr); if (result) { channel2_->Enable(true); result = channel1_->SetRemoteContent(&remote_media_content2_, - SdpType::kPrAnswer, NULL); + SdpType::kPrAnswer, NULL) && + channel1_->UpdateRtpTransport(nullptr); ConnectFakeTransports(); } return result; @@ -372,10 +384,12 @@ class ChannelTest : public ::testing::Test, public sigslot::has_slots<> { bool SendFinalAnswer() { bool result = channel2_->SetLocalContent(&local_media_content2_, - SdpType::kAnswer, NULL); + SdpType::kAnswer, NULL) && + channel2_->UpdateRtpTransport(nullptr); if (result) result = channel1_->SetRemoteContent(&remote_media_content2_, - SdpType::kAnswer, NULL); + SdpType::kAnswer, NULL) && + channel1_->UpdateRtpTransport(nullptr); return result; } @@ -608,10 +622,12 @@ class ChannelTest : public ::testing::Test, public sigslot::has_slots<> { CreateContent(0, kPcmuCodec, kH264Codec, &content1); content1.AddStream(stream1); EXPECT_TRUE(channel1_->SetLocalContent(&content1, SdpType::kOffer, NULL)); + EXPECT_TRUE(channel1_->UpdateRtpTransport(nullptr)); EXPECT_TRUE(channel1_->Enable(true)); EXPECT_EQ(1u, media_channel1_->send_streams().size()); EXPECT_TRUE(channel2_->SetRemoteContent(&content1, SdpType::kOffer, NULL)); + EXPECT_TRUE(channel2_->UpdateRtpTransport(nullptr)); EXPECT_EQ(1u, media_channel2_->recv_streams().size()); ConnectFakeTransports(); @@ -619,8 +635,10 @@ class ChannelTest : public ::testing::Test, public sigslot::has_slots<> { typename T::Content content2; CreateContent(0, kPcmuCodec, kH264Codec, &content2); EXPECT_TRUE(channel1_->SetRemoteContent(&content2, SdpType::kAnswer, NULL)); + EXPECT_TRUE(channel1_->UpdateRtpTransport(nullptr)); EXPECT_EQ(0u, media_channel1_->recv_streams().size()); EXPECT_TRUE(channel2_->SetLocalContent(&content2, SdpType::kAnswer, NULL)); + EXPECT_TRUE(channel2_->UpdateRtpTransport(nullptr)); EXPECT_TRUE(channel2_->Enable(true)); EXPECT_EQ(0u, media_channel2_->send_streams().size()); @@ -633,10 +651,12 @@ class ChannelTest : public ::testing::Test, public sigslot::has_slots<> { CreateContent(0, kPcmuCodec, kH264Codec, &content3); content3.AddStream(stream2); EXPECT_TRUE(channel2_->SetLocalContent(&content3, SdpType::kOffer, NULL)); + EXPECT_TRUE(channel2_->UpdateRtpTransport(nullptr)); ASSERT_EQ(1u, media_channel2_->send_streams().size()); EXPECT_EQ(stream2, media_channel2_->send_streams()[0]); EXPECT_TRUE(channel1_->SetRemoteContent(&content3, SdpType::kOffer, NULL)); + EXPECT_TRUE(channel1_->UpdateRtpTransport(nullptr)); ASSERT_EQ(1u, media_channel1_->recv_streams().size()); EXPECT_EQ(stream2, media_channel1_->recv_streams()[0]); @@ -644,9 +664,11 @@ class ChannelTest : public ::testing::Test, public sigslot::has_slots<> { typename T::Content content4; CreateContent(0, kPcmuCodec, kH264Codec, &content4); EXPECT_TRUE(channel1_->SetLocalContent(&content4, SdpType::kAnswer, NULL)); + EXPECT_TRUE(channel1_->UpdateRtpTransport(nullptr)); EXPECT_EQ(0u, media_channel1_->send_streams().size()); EXPECT_TRUE(channel2_->SetRemoteContent(&content4, SdpType::kAnswer, NULL)); + EXPECT_TRUE(channel2_->UpdateRtpTransport(nullptr)); EXPECT_EQ(0u, media_channel2_->recv_streams().size()); SendCustomRtp2(kSsrc2, 0); @@ -915,8 +937,6 @@ class ChannelTest : public ::testing::Test, public sigslot::has_slots<> { EXPECT_FALSE(channel2_->SrtpActiveForTesting()); EXPECT_TRUE(SendInitiate()); WaitForThreads(); - EXPECT_TRUE(channel1_->writable()); - EXPECT_TRUE(channel2_->writable()); EXPECT_TRUE(SendAccept()); EXPECT_TRUE(channel1_->SrtpActiveForTesting()); EXPECT_TRUE(channel2_->SrtpActiveForTesting()); diff --git a/pc/sdp_offer_answer.cc b/pc/sdp_offer_answer.cc index fd697ce8b1..f924c4060d 100644 --- a/pc/sdp_offer_answer.cc +++ b/pc/sdp_offer_answer.cc @@ -2473,11 +2473,6 @@ RTCError SdpOfferAnswerHandler::UpdateSessionState( // But all call-sites should be verifying this before calling us! RTC_DCHECK(session_error() == SessionError::kNone); - // If this is answer-ish we're ready to let media flow. - if (type == SdpType::kPrAnswer || type == SdpType::kAnswer) { - EnableSending(); - } - // Update the signaling state according to the specified state machine (see // https://w3c.github.io/webrtc-pc/#rtcsignalingstate-enum). if (type == SdpType::kOffer) { @@ -4201,21 +4196,6 @@ void SdpOfferAnswerHandler::UpdateRemoteSendersList( } } -void SdpOfferAnswerHandler::EnableSending() { - RTC_DCHECK_RUN_ON(signaling_thread()); - for (const auto& transceiver : transceivers()->List()) { - cricket::ChannelInterface* channel = transceiver->internal()->channel(); - if (channel && !channel->enabled()) { - channel->Enable(true); - } - } - - if (data_channel_controller()->rtp_data_channel() && - !data_channel_controller()->rtp_data_channel()->enabled()) { - data_channel_controller()->rtp_data_channel()->Enable(true); - } -} - RTCError SdpOfferAnswerHandler::PushdownMediaDescription( SdpType type, cricket::ContentSource source) { @@ -4225,15 +4205,13 @@ RTCError SdpOfferAnswerHandler::PushdownMediaDescription( RTC_DCHECK_RUN_ON(signaling_thread()); RTC_DCHECK(sdesc); - if (!UpdatePayloadTypeDemuxingState(source)) { - // Note that this is never expected to fail, since RtpDemuxer doesn't return - // an error when changing payload type demux criteria, which is all this - // does. - LOG_AND_RETURN_ERROR(RTCErrorType::INTERNAL_ERROR, - "Failed to update payload type demuxing state."); - } + // Gather lists of updates to be made on cricket channels on the signaling + // thread, before performing them all at once on the worker thread. Necessary + // due to threading restrictions. + auto payload_type_demuxing_updates = GetPayloadTypeDemuxingUpdates(source); + std::vector content_updates; - // Push down the new SDP media section for each audio/video transceiver. + // Collect updates for each audio/video transceiver. for (const auto& transceiver : transceivers()->List()) { const ContentInfo* content_info = FindMediaSectionForTransceiver(transceiver, sdesc); @@ -4243,19 +4221,12 @@ RTCError SdpOfferAnswerHandler::PushdownMediaDescription( } const MediaContentDescription* content_desc = content_info->media_description(); - if (!content_desc) { - continue; - } - std::string error; - bool success = (source == cricket::CS_LOCAL) - ? channel->SetLocalContent(content_desc, type, &error) - : channel->SetRemoteContent(content_desc, type, &error); - if (!success) { - LOG_AND_RETURN_ERROR(RTCErrorType::INVALID_PARAMETER, error); + if (content_desc) { + content_updates.emplace_back(channel, content_desc); } } - // If using the RtpDataChannel, push down the new SDP section for it too. + // If using the RtpDataChannel, add it to the list of updates. if (data_channel_controller()->rtp_data_channel()) { const ContentInfo* data_content = cricket::GetFirstDataContent(sdesc->description()); @@ -4263,21 +4234,21 @@ RTCError SdpOfferAnswerHandler::PushdownMediaDescription( const MediaContentDescription* data_desc = data_content->media_description(); if (data_desc) { - std::string error; - bool success = (source == cricket::CS_LOCAL) - ? data_channel_controller() - ->rtp_data_channel() - ->SetLocalContent(data_desc, type, &error) - : data_channel_controller() - ->rtp_data_channel() - ->SetRemoteContent(data_desc, type, &error); - if (!success) { - LOG_AND_RETURN_ERROR(RTCErrorType::INVALID_PARAMETER, error); - } + content_updates.push_back( + {data_channel_controller()->rtp_data_channel(), data_desc}); } } } + RTCError error = pc_->worker_thread()->Invoke( + RTC_FROM_HERE, + rtc::Bind(&SdpOfferAnswerHandler::ApplyChannelUpdates, this, type, source, + std::move(payload_type_demuxing_updates), + std::move(content_updates))); + if (!error.ok()) { + return error; + } + // Need complete offer/answer with an SCTP m= section before starting SCTP, // according to https://tools.ietf.org/html/draft-ietf-mmusic-sctp-sdp-19 if (pc_->sctp_mid() && local_description() && remote_description()) { @@ -4306,6 +4277,49 @@ RTCError SdpOfferAnswerHandler::PushdownMediaDescription( return RTCError::OK(); } +RTCError SdpOfferAnswerHandler::ApplyChannelUpdates( + SdpType type, + cricket::ContentSource source, + std::vector payload_type_demuxing_updates, + std::vector content_updates) { + RTC_DCHECK_RUN_ON(pc_->worker_thread()); + // If this is answer-ish we're ready to let media flow. + bool enable_sending = type == SdpType::kPrAnswer || type == SdpType::kAnswer; + std::set modified_channels; + for (const auto& update : payload_type_demuxing_updates) { + modified_channels.insert(update.channel); + update.channel->SetPayloadTypeDemuxingEnabled(update.enabled); + } + for (const auto& update : content_updates) { + modified_channels.insert(update.channel); + std::string error; + bool success = (source == cricket::CS_LOCAL) + ? update.channel->SetLocalContent( + update.content_description, type, &error) + : update.channel->SetRemoteContent( + update.content_description, type, &error); + if (!success) { + LOG_AND_RETURN_ERROR(RTCErrorType::INVALID_PARAMETER, error); + } + if (enable_sending && !update.channel->enabled()) { + update.channel->Enable(true); + } + } + // The above calls may have modified properties of the channel (header + // extension mappings, demuxer criteria) which still need to be applied to the + // RtpTransport. + return pc_->network_thread()->Invoke( + RTC_FROM_HERE, [modified_channels] { + for (auto channel : modified_channels) { + std::string error; + if (!channel->UpdateRtpTransport(&error)) { + LOG_AND_RETURN_ERROR(RTCErrorType::INVALID_PARAMETER, error); + } + } + return RTCError::OK(); + }); +} + RTCError SdpOfferAnswerHandler::PushdownTransportDescription( cricket::ContentSource source, SdpType type) { @@ -4904,7 +4918,8 @@ const std::string SdpOfferAnswerHandler::GetTransportName( return ""; } -bool SdpOfferAnswerHandler::UpdatePayloadTypeDemuxingState( +std::vector +SdpOfferAnswerHandler::GetPayloadTypeDemuxingUpdates( cricket::ContentSource source) { RTC_DCHECK_RUN_ON(signaling_thread()); // We may need to delete any created default streams and disable creation of @@ -4976,8 +4991,7 @@ bool SdpOfferAnswerHandler::UpdatePayloadTypeDemuxingState( // Gather all updates ahead of time so that all channels can be updated in a // single Invoke; necessary due to thread guards. - std::vector> - channels_to_update; + std::vector channel_updates; for (const auto& transceiver : transceivers()->List()) { cricket::ChannelInterface* channel = transceiver->internal()->channel(); const ContentInfo* content = @@ -4990,38 +5004,22 @@ bool SdpOfferAnswerHandler::UpdatePayloadTypeDemuxingState( if (source == cricket::CS_REMOTE) { local_direction = RtpTransceiverDirectionReversed(local_direction); } - channels_to_update.emplace_back(local_direction, - transceiver->internal()->channel()); + cricket::MediaType media_type = channel->media_type(); + bool in_bundle_group = + (bundle_group && bundle_group->HasContentName(channel->content_name())); + bool payload_type_demuxing_enabled = false; + if (media_type == cricket::MediaType::MEDIA_TYPE_AUDIO) { + payload_type_demuxing_enabled = + (!in_bundle_group || pt_demuxing_enabled_audio) && + RtpTransceiverDirectionHasRecv(local_direction); + } else if (media_type == cricket::MediaType::MEDIA_TYPE_VIDEO) { + payload_type_demuxing_enabled = + (!in_bundle_group || pt_demuxing_enabled_video) && + RtpTransceiverDirectionHasRecv(local_direction); + } + channel_updates.emplace_back(channel, payload_type_demuxing_enabled); } - - if (channels_to_update.empty()) { - return true; - } - return pc_->worker_thread()->Invoke( - RTC_FROM_HERE, [&channels_to_update, bundle_group, - pt_demuxing_enabled_audio, pt_demuxing_enabled_video]() { - for (const auto& it : channels_to_update) { - RtpTransceiverDirection local_direction = it.first; - cricket::ChannelInterface* channel = it.second; - cricket::MediaType media_type = channel->media_type(); - bool in_bundle_group = (bundle_group && bundle_group->HasContentName( - channel->content_name())); - if (media_type == cricket::MediaType::MEDIA_TYPE_AUDIO) { - if (!channel->SetPayloadTypeDemuxingEnabled( - (!in_bundle_group || pt_demuxing_enabled_audio) && - RtpTransceiverDirectionHasRecv(local_direction))) { - return false; - } - } else if (media_type == cricket::MediaType::MEDIA_TYPE_VIDEO) { - if (!channel->SetPayloadTypeDemuxingEnabled( - (!in_bundle_group || pt_demuxing_enabled_video) && - RtpTransceiverDirectionHasRecv(local_direction))) { - return false; - } - } - } - return true; - }); + return channel_updates; } } // namespace webrtc diff --git a/pc/sdp_offer_answer.h b/pc/sdp_offer_answer.h index 43a3dbb5a8..4b14f20708 100644 --- a/pc/sdp_offer_answer.h +++ b/pc/sdp_offer_answer.h @@ -455,15 +455,32 @@ class SdpOfferAnswerHandler : public SdpStateProvider, cricket::MediaType media_type, StreamCollection* new_streams); - // Enables media channels to allow sending of media. - // This enables media to flow on all configured audio/video channels and the - // RtpDataChannel. - void EnableSending(); // Push the media parts of the local or remote session description - // down to all of the channels. + // down to all of the channels, and enable sending if applicable. RTCError PushdownMediaDescription(SdpType type, cricket::ContentSource source); + struct PayloadTypeDemuxingUpdate { + PayloadTypeDemuxingUpdate(cricket::ChannelInterface* channel, bool enabled) + : channel(channel), enabled(enabled) {} + cricket::ChannelInterface* channel; + bool enabled; + }; + struct ContentUpdate { + ContentUpdate(cricket::ChannelInterface* channel, + const cricket::MediaContentDescription* content_description) + : channel(channel), content_description(content_description) {} + cricket::ChannelInterface* channel; + const cricket::MediaContentDescription* content_description; + }; + // Helper method used by PushdownMediaDescription to apply a batch of updates + // to BaseChannels on the worker thread. + RTCError ApplyChannelUpdates( + SdpType type, + cricket::ContentSource source, + std::vector payload_type_demuxing_updates, + std::vector content_updates); + RTCError PushdownTransportDescription(cricket::ContentSource source, SdpType type); // Helper function to remove stopped transceivers. @@ -550,9 +567,14 @@ class SdpOfferAnswerHandler : public SdpStateProvider, const std::string& mid) const; const std::string GetTransportName(const std::string& content_name); - // Based on number of transceivers per media type, enabled or disable - // payload type based demuxing in the affected channels. - bool UpdatePayloadTypeDemuxingState(cricket::ContentSource source); + + // Based on number of transceivers per media type, and their bundle status and + // payload types, determine whether payload type based demuxing should be + // enabled or disabled. Returns a list of channels and the corresponding + // value to be passed into SetPayloadTypeDemuxingEnabled, so that this action + // can be combined with other operations on the worker thread. + std::vector GetPayloadTypeDemuxingUpdates( + cricket::ContentSource source); // ================================================================== // Access to pc_ variables diff --git a/pc/test/mock_channel_interface.h b/pc/test/mock_channel_interface.h index 1be4dcb0ce..3a73225239 100644 --- a/pc/test/mock_channel_interface.h +++ b/pc/test/mock_channel_interface.h @@ -46,7 +46,8 @@ class MockChannelInterface : public cricket::ChannelInterface { webrtc::SdpType, std::string*), (override)); - MOCK_METHOD(bool, SetPayloadTypeDemuxingEnabled, (bool), (override)); + MOCK_METHOD(void, SetPayloadTypeDemuxingEnabled, (bool), (override)); + MOCK_METHOD(bool, UpdateRtpTransport, (std::string*), (override)); MOCK_METHOD(const std::vector&, local_streams, (),