From 6cd405850467683cf10d05028ea0f644a68a91a4 Mon Sep 17 00:00:00 2001 From: Tomas Gunnarsson Date: Mon, 8 Feb 2021 14:20:08 +0100 Subject: [PATCH] Fix unsynchronized access to mid_to_transport_ in JsepTransportController * Added several thread checks to JTC to help with programmer errors. * Avoid a few Invokes() to the network thread here and there such as for fetching sctp transport name for getStats(). The transport name is now cached when it changes on the network thread. * JsepTransportController instances now get deleted on the network thread rather than on the signaling thread + issuing an Invoke() in the dtor. * Moved some thread hops from JTC over to PC which is where the problem exists and also (imho) makes it easier to see where hops happen in the PC code. * The sctp transport is now started asynchronously when we push down the media description. * PeerConnection proxy calls GetSctpTransport directly on the network thread instead of to the signaling thread + blocking on the network thread. * The above changes simplified things for webrtc::SctpTransport which allowed for removing locking from that class and delete some code. Bug: webrtc:9987, webrtc:12445 Change-Id: Ic89a9426e314e1b93c81751d4f732f05fa448fbc Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/205620 Commit-Queue: Tommi Reviewed-by: Harald Alvestrand Cr-Commit-Position: refs/heads/master@{#33191} --- api/peer_connection_proxy.h | 10 +- pc/jsep_transport_controller.cc | 37 ++++++-- pc/jsep_transport_controller.h | 8 +- pc/jsep_transport_controller_unittest.cc | 3 + pc/peer_connection.cc | 98 +++++++++++++++---- pc/peer_connection.h | 27 +++--- pc/peer_connection_factory.cc | 9 +- pc/peer_connection_integrationtest.cc | 8 +- pc/sctp_transport.cc | 115 +++++++++-------------- pc/sctp_transport.h | 18 ++-- pc/sdp_offer_answer.cc | 47 ++++++--- 11 files changed, 238 insertions(+), 142 deletions(-) diff --git a/api/peer_connection_proxy.h b/api/peer_connection_proxy.h index 2d4cb5cad0..5c1d4b768a 100644 --- a/api/peer_connection_proxy.h +++ b/api/peer_connection_proxy.h @@ -20,9 +20,12 @@ namespace webrtc { +// PeerConnection proxy objects will be constructed with two thread pointers, +// signaling and network. The proxy macros don't have 'network' specific macros +// and support for a secondary thread is provided via 'WORKER' macros. // TODO(deadbeef): Move this to .cc file and out of api/. What threads methods // are called on is an implementation detail. -BEGIN_SIGNALING_PROXY_MAP(PeerConnection) +BEGIN_PROXY_MAP(PeerConnection) PROXY_SIGNALING_THREAD_DESTRUCTOR() PROXY_METHOD0(rtc::scoped_refptr, local_streams) PROXY_METHOD0(rtc::scoped_refptr, remote_streams) @@ -133,7 +136,10 @@ PROXY_METHOD1(void, SetAudioRecording, bool) PROXY_METHOD1(rtc::scoped_refptr, LookupDtlsTransportByMid, const std::string&) -PROXY_CONSTMETHOD0(rtc::scoped_refptr, GetSctpTransport) +// This method will be invoked on the network thread. See +// PeerConnectionFactory::CreatePeerConnectionOrError for more details. +PROXY_WORKER_CONSTMETHOD0(rtc::scoped_refptr, + GetSctpTransport) PROXY_METHOD0(SignalingState, signaling_state) PROXY_METHOD0(IceConnectionState, ice_connection_state) PROXY_METHOD0(IceConnectionState, standardized_ice_connection_state) diff --git a/pc/jsep_transport_controller.cc b/pc/jsep_transport_controller.cc index 542dae4181..0ded1de84f 100644 --- a/pc/jsep_transport_controller.cc +++ b/pc/jsep_transport_controller.cc @@ -105,10 +105,8 @@ JsepTransportController::JsepTransportController( JsepTransportController::~JsepTransportController() { // Channel destructors may try to send packets, so this needs to happen on // the network thread. - network_thread_->Invoke(RTC_FROM_HERE, [this] { - RTC_DCHECK_RUN_ON(network_thread_); - DestroyAllJsepTransports_n(); - }); + RTC_DCHECK_RUN_ON(network_thread_); + DestroyAllJsepTransports_n(); } RTCError JsepTransportController::SetLocalDescription( @@ -145,6 +143,7 @@ RTCError JsepTransportController::SetRemoteDescription( RtpTransportInternal* JsepTransportController::GetRtpTransport( const std::string& mid) const { + RTC_DCHECK_RUN_ON(network_thread_); auto jsep_transport = GetJsepTransportForMid(mid); if (!jsep_transport) { return nullptr; @@ -154,6 +153,7 @@ RtpTransportInternal* JsepTransportController::GetRtpTransport( DataChannelTransportInterface* JsepTransportController::GetDataChannelTransport( const std::string& mid) const { + RTC_DCHECK_RUN_ON(network_thread_); auto jsep_transport = GetJsepTransportForMid(mid); if (!jsep_transport) { return nullptr; @@ -163,6 +163,7 @@ DataChannelTransportInterface* JsepTransportController::GetDataChannelTransport( cricket::DtlsTransportInternal* JsepTransportController::GetDtlsTransport( const std::string& mid) { + RTC_DCHECK_RUN_ON(network_thread_); auto jsep_transport = GetJsepTransportForMid(mid); if (!jsep_transport) { return nullptr; @@ -172,6 +173,7 @@ cricket::DtlsTransportInternal* JsepTransportController::GetDtlsTransport( const cricket::DtlsTransportInternal* JsepTransportController::GetRtcpDtlsTransport(const std::string& mid) const { + RTC_DCHECK_RUN_ON(network_thread_); auto jsep_transport = GetJsepTransportForMid(mid); if (!jsep_transport) { return nullptr; @@ -181,6 +183,7 @@ JsepTransportController::GetRtcpDtlsTransport(const std::string& mid) const { rtc::scoped_refptr JsepTransportController::LookupDtlsTransportByMid(const std::string& mid) { + RTC_DCHECK_RUN_ON(network_thread_); auto jsep_transport = GetJsepTransportForMid(mid); if (!jsep_transport) { return nullptr; @@ -190,6 +193,7 @@ JsepTransportController::LookupDtlsTransportByMid(const std::string& mid) { rtc::scoped_refptr JsepTransportController::GetSctpTransport( const std::string& mid) const { + RTC_DCHECK_RUN_ON(network_thread_); auto jsep_transport = GetJsepTransportForMid(mid); if (!jsep_transport) { return nullptr; @@ -236,11 +240,16 @@ bool JsepTransportController::NeedsIceRestart( absl::optional JsepTransportController::GetDtlsRole( const std::string& mid) const { + // TODO(tommi): Remove this hop. Currently it's called from the signaling + // thread during negotiations, potentially multiple times. + // WebRtcSessionDescriptionFactory::InternalCreateAnswer is one example. if (!network_thread_->IsCurrent()) { return network_thread_->Invoke>( RTC_FROM_HERE, [&] { return GetDtlsRole(mid); }); } + RTC_DCHECK_RUN_ON(network_thread_); + const cricket::JsepTransport* t = GetJsepTransportForMid(mid); if (!t) { return absl::optional(); @@ -846,24 +855,34 @@ bool JsepTransportController::HandleBundledContent( bool JsepTransportController::SetTransportForMid( const std::string& mid, cricket::JsepTransport* jsep_transport) { - RTC_DCHECK(jsep_transport); - if (mid_to_transport_[mid] == jsep_transport) { - return true; - } RTC_DCHECK_RUN_ON(network_thread_); + RTC_DCHECK(jsep_transport); + + auto it = mid_to_transport_.find(mid); + if (it != mid_to_transport_.end() && it->second == jsep_transport) + return true; + pending_mids_.push_back(mid); - mid_to_transport_[mid] = jsep_transport; + + if (it == mid_to_transport_.end()) { + mid_to_transport_.insert(std::make_pair(mid, jsep_transport)); + } else { + it->second = jsep_transport; + } + return config_.transport_observer->OnTransportChanged( mid, jsep_transport->rtp_transport(), jsep_transport->RtpDtlsTransport(), jsep_transport->data_channel_transport()); } void JsepTransportController::RemoveTransportForMid(const std::string& mid) { + RTC_DCHECK_RUN_ON(network_thread_); bool ret = config_.transport_observer->OnTransportChanged(mid, nullptr, nullptr, nullptr); // Calling OnTransportChanged with nullptr should always succeed, since it is // only expected to fail when adding media to a transport (not removing). RTC_DCHECK(ret); + mid_to_transport_.erase(mid); } diff --git a/pc/jsep_transport_controller.h b/pc/jsep_transport_controller.h index 506a41808a..59d66a24f2 100644 --- a/pc/jsep_transport_controller.h +++ b/pc/jsep_transport_controller.h @@ -363,8 +363,9 @@ class JsepTransportController : public sigslot::has_slots<> { // transports are bundled on (In current implementation, it is the first // content in the BUNDLE group). const cricket::JsepTransport* GetJsepTransportForMid( - const std::string& mid) const; - cricket::JsepTransport* GetJsepTransportForMid(const std::string& mid); + const std::string& mid) const RTC_RUN_ON(network_thread_); + cricket::JsepTransport* GetJsepTransportForMid(const std::string& mid) + RTC_RUN_ON(network_thread_); // Get the JsepTransport without considering the BUNDLE group. Return nullptr // if the JsepTransport is destroyed. @@ -460,7 +461,8 @@ class JsepTransportController : public sigslot::has_slots<> { jsep_transports_by_name_ RTC_GUARDED_BY(network_thread_); // This keeps track of the mapping between media section // (BaseChannel/SctpTransport) and the JsepTransport underneath. - std::map mid_to_transport_; + std::map mid_to_transport_ + RTC_GUARDED_BY(network_thread_); // Keep track of mids that have been mapped to transports. Used for rollback. std::vector pending_mids_ RTC_GUARDED_BY(network_thread_); // Aggregate states for Transports. diff --git a/pc/jsep_transport_controller_unittest.cc b/pc/jsep_transport_controller_unittest.cc index 5361f904aa..9efa205368 100644 --- a/pc/jsep_transport_controller_unittest.cc +++ b/pc/jsep_transport_controller_unittest.cc @@ -904,6 +904,9 @@ TEST_F(JsepTransportControllerTest, IceSignalingOccursOnSignalingThread) { EXPECT_EQ(2, candidates_signal_count_); EXPECT_TRUE(!signaled_on_non_signaling_thread_); + + network_thread_->Invoke(RTC_FROM_HERE, + [&] { transport_controller_.reset(); }); } // Test that if the TransportController was created with the diff --git a/pc/peer_connection.cc b/pc/peer_connection.cc index 2cb43bf408..f82fe35c6d 100644 --- a/pc/peer_connection.cc +++ b/pc/peer_connection.cc @@ -489,12 +489,17 @@ PeerConnection::~PeerConnection() { sdp_handler_->ResetSessionDescFactory(); } - transport_controller_.reset(); - // port_allocator_ lives on the network thread and should be destroyed there. + // port_allocator_ and transport_controller_ live on the network thread and + // should be destroyed there. network_thread()->Invoke(RTC_FROM_HERE, [this] { RTC_DCHECK_RUN_ON(network_thread()); + transport_controller_.reset(); port_allocator_.reset(); + if (network_thread_safety_) { + network_thread_safety_->SetNotAlive(); + network_thread_safety_ = nullptr; + } }); // call_ and event_log_ must be destroyed on the worker thread. worker_thread()->Invoke(RTC_FROM_HERE, [this] { @@ -527,13 +532,15 @@ RTCError PeerConnection::Initialize( } // The port allocator lives on the network thread and should be initialized - // there. + // there. Also set up the task safety flag for canceling pending tasks on + // the network thread when closing. // TODO(bugs.webrtc.org/12427): See if we can piggyback on this call and // initialize all the |transport_controller_->Subscribe*| calls below on the // network thread via this invoke. const auto pa_result = network_thread()->Invoke( RTC_FROM_HERE, [this, &stun_servers, &turn_servers, &configuration] { + network_thread_safety_ = PendingTaskSafetyFlag::Create(); return InitializePortAllocator_n(stun_servers, turn_servers, configuration); }); @@ -832,6 +839,16 @@ PeerConnection::AddTransceiver( return AddTransceiver(track, RtpTransceiverInit()); } +RtpTransportInternal* PeerConnection::GetRtpTransport(const std::string& mid) { + RTC_DCHECK_RUN_ON(signaling_thread()); + return network_thread()->Invoke( + RTC_FROM_HERE, [this, &mid] { + auto rtp_transport = transport_controller_->GetRtpTransport(mid); + RTC_DCHECK(rtp_transport); + return rtp_transport; + }); +} + RTCErrorOr> PeerConnection::AddTransceiver( rtc::scoped_refptr track, @@ -1588,11 +1605,11 @@ PeerConnection::LookupDtlsTransportByMidInternal(const std::string& mid) { rtc::scoped_refptr PeerConnection::GetSctpTransport() const { - RTC_DCHECK_RUN_ON(signaling_thread()); - if (!sctp_mid_s_) { + RTC_DCHECK_RUN_ON(network_thread()); + if (!sctp_mid_n_) return nullptr; - } - return transport_controller_->GetSctpTransport(*sctp_mid_s_); + + return transport_controller_->GetSctpTransport(*sctp_mid_n_); } const SessionDescriptionInterface* PeerConnection::local_description() const { @@ -1673,11 +1690,16 @@ void PeerConnection::Close() { // WebRTC session description factory, the session description factory would // call the transport controller. sdp_handler_->ResetSessionDescFactory(); - transport_controller_.reset(); rtp_manager_->Close(); - network_thread()->Invoke( - RTC_FROM_HERE, [this] { port_allocator_->DiscardCandidatePool(); }); + network_thread()->Invoke(RTC_FROM_HERE, [this] { + transport_controller_.reset(); + port_allocator_->DiscardCandidatePool(); + if (network_thread_safety_) { + network_thread_safety_->SetNotAlive(); + network_thread_safety_ = nullptr; + } + }); worker_thread()->Invoke(RTC_FROM_HERE, [this] { RTC_DCHECK_RUN_ON(worker_thread()); @@ -1810,6 +1832,17 @@ absl::optional PeerConnection::GetDataMid() const { } } +void PeerConnection::SetSctpDataMid(const std::string& mid) { + RTC_DCHECK_RUN_ON(signaling_thread()); + sctp_mid_s_ = mid; +} + +void PeerConnection::ResetSctpDataMid() { + RTC_DCHECK_RUN_ON(signaling_thread()); + sctp_mid_s_.reset(); + sctp_transport_name_s_.clear(); +} + void PeerConnection::OnSctpDataChannelClosed(DataChannelInterface* channel) { // Since data_channel_controller doesn't do signals, this // signal is relayed here. @@ -2023,13 +2056,8 @@ std::vector PeerConnection::GetDataChannelStats() const { absl::optional PeerConnection::sctp_transport_name() const { RTC_DCHECK_RUN_ON(signaling_thread()); - if (sctp_mid_s_ && transport_controller_) { - auto dtls_transport = transport_controller_->GetDtlsTransport(*sctp_mid_s_); - if (dtls_transport) { - return dtls_transport->transport_name(); - } - return absl::optional(); - } + if (sctp_mid_s_ && transport_controller_) + return sctp_transport_name_s_; return absl::optional(); } @@ -2268,6 +2296,15 @@ bool PeerConnection::SetupDataChannelTransport_n(const std::string& mid) { data_channel_controller_.set_data_channel_transport(transport); data_channel_controller_.SetupDataChannelTransport_n(); sctp_mid_n_ = mid; + auto dtls_transport = transport_controller_->GetDtlsTransport(mid); + if (dtls_transport) { + signaling_thread()->PostTask( + ToQueuedTask(signaling_thread_safety_.flag(), + [this, name = dtls_transport->transport_name()] { + RTC_DCHECK_RUN_ON(signaling_thread()); + sctp_transport_name_s_ = std::move(name); + })); + } // Note: setting the data sink and checking initial state must be done last, // after setting up the data channel. Setting the data sink may trigger @@ -2612,9 +2649,19 @@ bool PeerConnection::OnTransportChanged( if (base_channel) { ret = base_channel->SetRtpTransport(rtp_transport); } + if (mid == sctp_mid_n_) { data_channel_controller_.OnTransportChanged(data_channel_transport); + if (dtls_transport) { + signaling_thread()->PostTask(ToQueuedTask( + signaling_thread_safety_.flag(), + [this, name = dtls_transport->internal()->transport_name()] { + RTC_DCHECK_RUN_ON(signaling_thread()); + sctp_transport_name_s_ = std::move(name); + })); + } } + return ret; } @@ -2624,6 +2671,23 @@ PeerConnectionObserver* PeerConnection::Observer() const { return observer_; } +void PeerConnection::StartSctpTransport(int local_port, + int remote_port, + int max_message_size) { + RTC_DCHECK_RUN_ON(signaling_thread()); + if (!sctp_mid_s_) + return; + + network_thread()->PostTask(ToQueuedTask( + network_thread_safety_, + [this, mid = *sctp_mid_s_, local_port, remote_port, max_message_size] { + rtc::scoped_refptr sctp_transport = + transport_controller()->GetSctpTransport(mid); + if (sctp_transport) + sctp_transport->Start(local_port, remote_port, max_message_size); + })); +} + CryptoOptions PeerConnection::GetCryptoOptions() { RTC_DCHECK_RUN_ON(signaling_thread()); // TODO(bugs.webrtc.org/9891) - Remove PeerConnectionFactory::CryptoOptions diff --git a/pc/peer_connection.h b/pc/peer_connection.h index 4bab90a4b1..92e33d2858 100644 --- a/pc/peer_connection.h +++ b/pc/peer_connection.h @@ -404,14 +404,15 @@ class PeerConnection : public PeerConnectionInternal, // channels are configured this will return nullopt. absl::optional GetDataMid() const; - void SetSctpDataMid(const std::string& mid) { - RTC_DCHECK_RUN_ON(signaling_thread()); - sctp_mid_s_ = mid; - } - void ResetSctpDataMid() { - RTC_DCHECK_RUN_ON(signaling_thread()); - sctp_mid_s_.reset(); - } + void SetSctpDataMid(const std::string& mid); + + void ResetSctpDataMid(); + + // Asynchronously calls SctpTransport::Start() on the network thread for + // |sctp_mid()| if set. Called as part of setting the local description. + void StartSctpTransport(int local_port, + int remote_port, + int max_message_size); // Returns the CryptoOptions for this PeerConnection. This will always // return the RTCConfiguration.crypto_options if set and will only default @@ -427,12 +428,7 @@ class PeerConnection : public PeerConnectionInternal, bool fire_callback = true); // Returns rtp transport, result can not be nullptr. - RtpTransportInternal* GetRtpTransport(const std::string& mid) { - RTC_DCHECK_RUN_ON(signaling_thread()); - auto rtp_transport = transport_controller_->GetRtpTransport(mid); - RTC_DCHECK(rtp_transport); - return rtp_transport; - } + RtpTransportInternal* GetRtpTransport(const std::string& mid); // Returns true if SRTP (either using DTLS-SRTP or SDES) is required by // this session. @@ -648,6 +644,8 @@ class PeerConnection : public PeerConnectionInternal, // The unique_ptr belongs to the worker thread, but the Call object manages // its own thread safety. std::unique_ptr call_ RTC_GUARDED_BY(worker_thread()); + ScopedTaskSafety signaling_thread_safety_; + rtc::scoped_refptr network_thread_safety_; std::unique_ptr call_safety_ RTC_GUARDED_BY(worker_thread()); @@ -677,6 +675,7 @@ class PeerConnection : public PeerConnectionInternal, // thread, but applied first on the networking thread via an invoke(). absl::optional sctp_mid_s_ RTC_GUARDED_BY(signaling_thread()); absl::optional sctp_mid_n_ RTC_GUARDED_BY(network_thread()); + std::string sctp_transport_name_s_ RTC_GUARDED_BY(signaling_thread()); // The machinery for handling offers and answers. Const after initialization. std::unique_ptr sdp_handler_ diff --git a/pc/peer_connection_factory.cc b/pc/peer_connection_factory.cc index c65b2f5fca..a8d64fa739 100644 --- a/pc/peer_connection_factory.cc +++ b/pc/peer_connection_factory.cc @@ -265,8 +265,15 @@ PeerConnectionFactory::CreatePeerConnectionOrError( if (!result.ok()) { return result.MoveError(); } + // We configure the proxy with a pointer to the network thread for methods + // that need to be invoked there rather than on the signaling thread. + // Internally, the proxy object has a member variable named |worker_thread_| + // which will point to the network thread (and not the factory's + // worker_thread()). All such methods have thread checks though, so the code + // should still be clear (outside of macro expansion). rtc::scoped_refptr result_proxy = - PeerConnectionProxy::Create(signaling_thread(), result.MoveValue()); + PeerConnectionProxy::Create(signaling_thread(), network_thread(), + result.MoveValue()); return result_proxy; } diff --git a/pc/peer_connection_integrationtest.cc b/pc/peer_connection_integrationtest.cc index df8d6c15ec..c813c5ed71 100644 --- a/pc/peer_connection_integrationtest.cc +++ b/pc/peer_connection_integrationtest.cc @@ -5944,9 +5944,11 @@ TEST_F(PeerConnectionIntegrationTestUnifiedPlan, callee()->AddAudioVideoTracks(); caller()->CreateAndSetAndSignalOffer(); ASSERT_TRUE_WAIT(SignalingStateStable(), kDefaultTimeout); - ASSERT_EQ_WAIT(SctpTransportState::kConnected, - caller()->pc()->GetSctpTransport()->Information().state(), - kDefaultTimeout); + network_thread()->Invoke(RTC_FROM_HERE, [this] { + ASSERT_EQ_WAIT(SctpTransportState::kConnected, + caller()->pc()->GetSctpTransport()->Information().state(), + kDefaultTimeout); + }); ASSERT_TRUE_WAIT(callee()->data_channel(), kDefaultTimeout); ASSERT_TRUE_WAIT(callee()->data_observer()->IsOpen(), kDefaultTimeout); } diff --git a/pc/sctp_transport.cc b/pc/sctp_transport.cc index f3e40b838b..0f7e4fc9d0 100644 --- a/pc/sctp_transport.cc +++ b/pc/sctp_transport.cc @@ -45,7 +45,7 @@ SctpTransport::~SctpTransport() { } SctpTransportInformation SctpTransport::Information() const { - MutexLock lock(&lock_); + RTC_DCHECK_RUN_ON(owner_thread_); return info_; } @@ -71,103 +71,78 @@ rtc::scoped_refptr SctpTransport::dtls_transport() void SctpTransport::Clear() { RTC_DCHECK_RUN_ON(owner_thread_); RTC_DCHECK(internal()); - { - MutexLock lock(&lock_); - // Note that we delete internal_sctp_transport_, but - // only drop the reference to dtls_transport_. - dtls_transport_ = nullptr; - internal_sctp_transport_ = nullptr; - } + // Note that we delete internal_sctp_transport_, but + // only drop the reference to dtls_transport_. + dtls_transport_ = nullptr; + internal_sctp_transport_ = nullptr; UpdateInformation(SctpTransportState::kClosed); } void SctpTransport::SetDtlsTransport( rtc::scoped_refptr transport) { RTC_DCHECK_RUN_ON(owner_thread_); - SctpTransportState next_state; - { - MutexLock lock(&lock_); - next_state = info_.state(); - dtls_transport_ = transport; - if (internal_sctp_transport_) { - if (transport) { - internal_sctp_transport_->SetDtlsTransport(transport->internal()); - transport->internal()->SignalDtlsState.connect( - this, &SctpTransport::OnDtlsStateChange); - if (info_.state() == SctpTransportState::kNew) { - next_state = SctpTransportState::kConnecting; - } - } else { - internal_sctp_transport_->SetDtlsTransport(nullptr); + SctpTransportState next_state = info_.state(); + dtls_transport_ = transport; + if (internal_sctp_transport_) { + if (transport) { + internal_sctp_transport_->SetDtlsTransport(transport->internal()); + transport->internal()->SignalDtlsState.connect( + this, &SctpTransport::OnDtlsStateChange); + if (info_.state() == SctpTransportState::kNew) { + next_state = SctpTransportState::kConnecting; } + } else { + internal_sctp_transport_->SetDtlsTransport(nullptr); } } + UpdateInformation(next_state); } void SctpTransport::Start(int local_port, int remote_port, int max_message_size) { - { - MutexLock lock(&lock_); - // Record max message size on calling thread. - info_ = SctpTransportInformation(info_.state(), info_.dtls_transport(), - max_message_size, info_.MaxChannels()); - } - if (owner_thread_->IsCurrent()) { - if (!internal()->Start(local_port, remote_port, max_message_size)) { - RTC_LOG(LS_ERROR) << "Failed to push down SCTP parameters, closing."; - UpdateInformation(SctpTransportState::kClosed); - } - } else { - owner_thread_->Invoke( - RTC_FROM_HERE, [this, local_port, remote_port, max_message_size] { - Start(local_port, remote_port, max_message_size); - }); + RTC_DCHECK_RUN_ON(owner_thread_); + info_ = SctpTransportInformation(info_.state(), info_.dtls_transport(), + max_message_size, info_.MaxChannels()); + + if (!internal()->Start(local_port, remote_port, max_message_size)) { + RTC_LOG(LS_ERROR) << "Failed to push down SCTP parameters, closing."; + UpdateInformation(SctpTransportState::kClosed); } } void SctpTransport::UpdateInformation(SctpTransportState state) { RTC_DCHECK_RUN_ON(owner_thread_); - bool must_send_update; - SctpTransportInformation info_copy(SctpTransportState::kNew); - { - MutexLock lock(&lock_); - must_send_update = (state != info_.state()); - // TODO(https://bugs.webrtc.org/10358): Update max channels from internal - // SCTP transport when available. - if (internal_sctp_transport_) { - info_ = SctpTransportInformation( - state, dtls_transport_, info_.MaxMessageSize(), info_.MaxChannels()); - } else { - info_ = SctpTransportInformation( - state, dtls_transport_, info_.MaxMessageSize(), info_.MaxChannels()); - } - if (observer_ && must_send_update) { - info_copy = info_; - } + bool must_send_update = (state != info_.state()); + // TODO(https://bugs.webrtc.org/10358): Update max channels from internal + // SCTP transport when available. + if (internal_sctp_transport_) { + info_ = SctpTransportInformation( + state, dtls_transport_, info_.MaxMessageSize(), info_.MaxChannels()); + } else { + info_ = SctpTransportInformation( + state, dtls_transport_, info_.MaxMessageSize(), info_.MaxChannels()); } - // We call the observer without holding the lock. + if (observer_ && must_send_update) { - observer_->OnStateChange(info_copy); + observer_->OnStateChange(info_); } } void SctpTransport::OnAssociationChangeCommunicationUp() { RTC_DCHECK_RUN_ON(owner_thread_); - { - MutexLock lock(&lock_); - RTC_DCHECK(internal_sctp_transport_); - if (internal_sctp_transport_->max_outbound_streams() && - internal_sctp_transport_->max_inbound_streams()) { - int max_channels = - std::min(*(internal_sctp_transport_->max_outbound_streams()), - *(internal_sctp_transport_->max_inbound_streams())); - // Record max channels. - info_ = SctpTransportInformation(info_.state(), info_.dtls_transport(), - info_.MaxMessageSize(), max_channels); - } + RTC_DCHECK(internal_sctp_transport_); + if (internal_sctp_transport_->max_outbound_streams() && + internal_sctp_transport_->max_inbound_streams()) { + int max_channels = + std::min(*(internal_sctp_transport_->max_outbound_streams()), + *(internal_sctp_transport_->max_inbound_streams())); + // Record max channels. + info_ = SctpTransportInformation(info_.state(), info_.dtls_transport(), + info_.MaxMessageSize(), max_channels); } + UpdateInformation(SctpTransportState::kConnected); } diff --git a/pc/sctp_transport.h b/pc/sctp_transport.h index d916a00897..4bb42748fc 100644 --- a/pc/sctp_transport.h +++ b/pc/sctp_transport.h @@ -20,7 +20,6 @@ #include "media/sctp/sctp_transport_internal.h" #include "p2p/base/dtls_transport_internal.h" #include "pc/dtls_transport.h" -#include "rtc_base/synchronization/mutex.h" #include "rtc_base/third_party/sigslot/sigslot.h" #include "rtc_base/thread.h" #include "rtc_base/thread_annotations.h" @@ -54,12 +53,12 @@ class SctpTransport : public SctpTransportInterface, // internal() to be functions on the webrtc::SctpTransport interface, // and make the internal() function private. cricket::SctpTransportInternal* internal() { - MutexLock lock(&lock_); + RTC_DCHECK_RUN_ON(owner_thread_); return internal_sctp_transport_.get(); } const cricket::SctpTransportInternal* internal() const { - MutexLock lock(&lock_); + RTC_DCHECK_RUN_ON(owner_thread_); return internal_sctp_transport_.get(); } @@ -75,15 +74,12 @@ class SctpTransport : public SctpTransportInterface, void OnDtlsStateChange(cricket::DtlsTransportInternal* transport, cricket::DtlsTransportState state); - // Note - owner_thread never changes, but can't be const if we do - // Invoke() on it. - rtc::Thread* owner_thread_; - mutable Mutex lock_; - // Variables accessible off-thread, guarded by lock_ - SctpTransportInformation info_ RTC_GUARDED_BY(lock_); + // NOTE: |owner_thread_| is the thread that the SctpTransport object is + // constructed on. In the context of PeerConnection, it's the network thread. + rtc::Thread* const owner_thread_; + SctpTransportInformation info_ RTC_GUARDED_BY(owner_thread_); std::unique_ptr internal_sctp_transport_ - RTC_GUARDED_BY(lock_); - // Variables only accessed on-thread + RTC_GUARDED_BY(owner_thread_); SctpTransportObserverInterface* observer_ RTC_GUARDED_BY(owner_thread_) = nullptr; rtc::scoped_refptr dtls_transport_ diff --git a/pc/sdp_offer_answer.cc b/pc/sdp_offer_answer.cc index 4dd5b6f1af..9fa4188e10 100644 --- a/pc/sdp_offer_answer.cc +++ b/pc/sdp_offer_answer.cc @@ -729,6 +729,21 @@ bool CanAddLocalMediaStream(webrtc::StreamCollectionInterface* current_streams, return true; } +rtc::scoped_refptr LookupDtlsTransportByMid( + rtc::Thread* network_thread, + JsepTransportController* controller, + const std::string& mid) { + // TODO(tommi): Can we post this (and associated operations where this + // function is called) to the network thread and avoid this Invoke? + // We might be able to simplify a few things if we set the transport on + // the network thread and then update the implementation to check that + // the set_ and relevant get methods are always called on the network + // thread (we'll need to update proxy maps). + return network_thread->Invoke>( + RTC_FROM_HERE, + [controller, &mid] { return controller->LookupDtlsTransportByMid(mid); }); +} + } // namespace // Used by parameterless SetLocalDescription() to create an offer or answer. @@ -1308,8 +1323,8 @@ RTCError SdpOfferAnswerHandler::ApplyLocalDescription( // Note that code paths that don't set MID won't be able to use // information about DTLS transports. if (transceiver->mid()) { - auto dtls_transport = transport_controller()->LookupDtlsTransportByMid( - *transceiver->mid()); + auto dtls_transport = LookupDtlsTransportByMid( + pc_->network_thread(), transport_controller(), *transceiver->mid()); transceiver->internal()->sender_internal()->set_transport( dtls_transport); transceiver->internal()->receiver_internal()->set_transport( @@ -1725,9 +1740,9 @@ RTCError SdpOfferAnswerHandler::ApplyRemoteDescription( transceiver->internal()->set_current_direction(local_direction); // 2.2.8.1.11.[3-6]: Set the transport internal slots. if (transceiver->mid()) { - auto dtls_transport = - transport_controller()->LookupDtlsTransportByMid( - *transceiver->mid()); + auto dtls_transport = LookupDtlsTransportByMid(pc_->network_thread(), + transport_controller(), + *transceiver->mid()); transceiver->internal()->sender_internal()->set_transport( dtls_transport); transceiver->internal()->receiver_internal()->set_transport( @@ -4276,13 +4291,11 @@ RTCError SdpOfferAnswerHandler::PushdownMediaDescription( // 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()) { - rtc::scoped_refptr sctp_transport = - transport_controller()->GetSctpTransport(*(pc_->sctp_mid())); auto local_sctp_description = cricket::GetFirstSctpDataContentDescription( local_description()->description()); auto remote_sctp_description = cricket::GetFirstSctpDataContentDescription( remote_description()->description()); - if (sctp_transport && local_sctp_description && remote_sctp_description) { + if (local_sctp_description && remote_sctp_description) { int max_message_size; // A remote max message size of zero means "any size supported". // We configure the connection with our own max message size. @@ -4293,8 +4306,9 @@ RTCError SdpOfferAnswerHandler::PushdownMediaDescription( std::min(local_sctp_description->max_message_size(), remote_sctp_description->max_message_size()); } - sctp_transport->Start(local_sctp_description->port(), - remote_sctp_description->port(), max_message_size); + pc_->StartSctpTransport(local_sctp_description->port(), + remote_sctp_description->port(), + max_message_size); } } @@ -4520,8 +4534,16 @@ bool SdpOfferAnswerHandler::ReadyToUseRemoteCandidate( return false; } - std::string transport_name = GetTransportName(result.value()->name); - return !transport_name.empty(); + bool has_transport = false; + cricket::ChannelInterface* channel = pc_->GetChannel(result.value()->name); + if (channel) { + has_transport = !channel->transport_name().empty(); + } else if (data_channel_controller()->data_channel_transport()) { + auto sctp_mid = pc_->sctp_mid(); + RTC_DCHECK(sctp_mid); + has_transport = (result.value()->name == *sctp_mid); + } + return has_transport; } void SdpOfferAnswerHandler::ReportRemoteIceCandidateAdded( @@ -4644,6 +4666,7 @@ cricket::VoiceChannel* SdpOfferAnswerHandler::CreateVoiceChannel( cricket::VideoChannel* SdpOfferAnswerHandler::CreateVideoChannel( const std::string& mid) { RTC_DCHECK_RUN_ON(signaling_thread()); + // NOTE: This involves a non-ideal hop (Invoke) over to the network thread. RtpTransportInternal* rtp_transport = pc_->GetRtpTransport(mid); // TODO(bugs.webrtc.org/11992): CreateVideoChannel internally switches to the