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/api/sctp_transport_interface.h b/api/sctp_transport_interface.h index 6af0bfce34..7080889fcf 100644 --- a/api/sctp_transport_interface.h +++ b/api/sctp_transport_interface.h @@ -35,6 +35,8 @@ enum class SctpTransportState { // http://w3c.github.io/webrtc-pc/#rtcsctptransport-interface class RTC_EXPORT SctpTransportInformation { public: + SctpTransportInformation() = default; + SctpTransportInformation(const SctpTransportInformation&) = default; explicit SctpTransportInformation(SctpTransportState state); SctpTransportInformation( SctpTransportState 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 406c44de54..087cffc4c4 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()); @@ -1831,6 +1853,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. @@ -2044,13 +2077,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(); } @@ -2289,6 +2317,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 @@ -2633,9 +2670,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; } @@ -2645,6 +2692,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 4a2561918a..4ed92adfce 100644 --- a/pc/peer_connection_integrationtest.cc +++ b/pc/peer_connection_integrationtest.cc @@ -5969,9 +5969,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..bb579af9c1 100644 --- a/pc/sctp_transport.cc +++ b/pc/sctp_transport.cc @@ -45,7 +45,15 @@ SctpTransport::~SctpTransport() { } SctpTransportInformation SctpTransport::Information() const { - MutexLock lock(&lock_); + // TODO(tommi): Update PeerConnection::GetSctpTransport to hand out a proxy + // to the transport so that we can be sure that methods get called on the + // expected thread. Chromium currently calls this method from + // TransceiverStateSurfacer. + if (!owner_thread_->IsCurrent()) { + return owner_thread_->Invoke( + RTC_FROM_HERE, [this] { return Information(); }); + } + RTC_DCHECK_RUN_ON(owner_thread_); return info_; } @@ -71,103 +79,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