From 35fc2a11456b0a8f2ee16ebcc02afd680902ef3e Mon Sep 17 00:00:00 2001 From: hai007 Date: Thu, 15 Apr 2021 19:38:08 -0700 Subject: [PATCH] Internal change PiperOrigin-RevId: 368765308 --- cpp/core/internal/mediums/webrtc.cc | 12 +- .../mediums/webrtc/connection_flow.cc | 421 +++++++++++------- .../internal/mediums/webrtc/connection_flow.h | 83 +++- .../mediums/webrtc/connection_flow_test.cc | 23 +- 4 files changed, 348 insertions(+), 191 deletions(-) diff --git a/cpp/core/internal/mediums/webrtc.cc b/cpp/core/internal/mediums/webrtc.cc index 78416c62..047e1af8 100644 --- a/cpp/core/internal/mediums/webrtc.cc +++ b/cpp/core/internal/mediums/webrtc.cc @@ -188,11 +188,10 @@ void WebRtc::StopAcceptingConnections(const std::string& service_id) { // Skip fully connected connections in this step. If the connection was // formed while we were accepting connections, then it will stay alive until // it's explicitly closed. - if (entry->second->GetState() == ConnectionFlow::State::kConnected) { + if (!entry->second->CloseIfNotConnected()) { continue; } - entry->second->Close(); connection_flows_.erase(peer_id); } @@ -260,7 +259,6 @@ WebRtcSocketWrapper WebRtc::AttemptToConnect( "Cannot connect to WebRTC peer %s because we failed to create a " "SignalingMessenger.", remote_peer_id.GetId().c_str()); - connection_flow->Close(); return WebRtcSocketWrapper(); } @@ -281,7 +279,6 @@ WebRtcSocketWrapper WebRtc::AttemptToConnect( "receiving messages over Tachyon.", remote_peer_id.GetId().c_str()); info.signaling_messenger.reset(); - connection_flow->Close(); return WebRtcSocketWrapper(); } @@ -294,7 +291,6 @@ WebRtcSocketWrapper WebRtc::AttemptToConnect( "the peer over Tachyon.", remote_peer_id.GetId().c_str()); info.signaling_messenger.reset(); - connection_flow->Close(); return WebRtcSocketWrapper(); } @@ -786,14 +782,10 @@ std::unique_ptr WebRtc::CreateConnectionFlow( } void WebRtc::RemoveConnectionFlow(const PeerId& remote_peer_id) { - const auto& entry = connection_flows_.find(remote_peer_id.GetId()); - if (entry == connection_flows_.end()) { + if (!connection_flows_.erase(remote_peer_id.GetId())) { return; } - entry->second->Close(); - connection_flows_.erase(remote_peer_id.GetId()); - // If we had an outgoing connection request w/ this peer, report the failure // to the future that's being waited on. const auto& connection_request_entry = diff --git a/cpp/core/internal/mediums/webrtc/connection_flow.cc b/cpp/core/internal/mediums/webrtc/connection_flow.cc index 6a7cdfec..ba02cc6d 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow.cc +++ b/cpp/core/internal/mediums/webrtc/connection_flow.cc @@ -18,6 +18,7 @@ #include #include "core/internal/mediums/webrtc/session_description_wrapper.h" +#include "platform/public/count_down_latch.h" #include "platform/public/logging.h" #include "platform/public/mutex_lock.h" #include "platform/public/webrtc.h" @@ -34,80 +35,85 @@ namespace mediums { constexpr absl::Duration ConnectionFlow::kTimeout; constexpr absl::Duration ConnectionFlow::kPeerConnectionTimeout; -namespace { // This is the same as the nearby data channel name. -const char kDataChannelName[] = "dataChannel"; +constexpr char kDataChannelName[] = "dataChannel"; class CreateSessionDescriptionObserverImpl : public webrtc::CreateSessionDescriptionObserver { public: - explicit CreateSessionDescriptionObserverImpl( - Future* settable_future) - : settable_future_(settable_future) {} - ~CreateSessionDescriptionObserverImpl() override = default; + CreateSessionDescriptionObserverImpl( + ConnectionFlow* connection_flow, + Future settable_future, + ConnectionFlow::State expected_entry_state, + ConnectionFlow::State exit_state) + : connection_flow_{connection_flow}, + settable_future_{settable_future}, + expected_entry_state_{expected_entry_state}, + exit_state_{exit_state} {} // webrtc::CreateSessionDescriptionObserver void OnSuccess(webrtc::SessionDescriptionInterface* desc) override { - settable_future_->Set(SessionDescriptionWrapper{desc}); + if (connection_flow_->TransitionState(expected_entry_state_, exit_state_)) { + settable_future_.Set(SessionDescriptionWrapper{desc}); + } else { + settable_future_.SetException({Exception::kFailed}); + } } void OnFailure(webrtc::RTCError error) override { NEARBY_LOG(ERROR, "Error when creating session description: %s", error.message()); - settable_future_->SetException({Exception::kFailed}); + settable_future_.SetException({Exception::kFailed}); } private: - std::unique_ptr> settable_future_; + ConnectionFlow* connection_flow_; + Future settable_future_; + ConnectionFlow::State expected_entry_state_; + ConnectionFlow::State exit_state_; +}; + +class SetDescriptionObserverBase { + public: + ExceptionOr GetResult(absl::Duration timeout) { + return settable_future_.Get(timeout); + } + + protected: + void OnSetDescriptionComplete(webrtc::RTCError error) { + // On success, |error.ok()| is true. + if (error.ok()) { + settable_future_.Set(true); + return; + } + settable_future_.SetException({Exception::kFailed}); + } + + private: + Future settable_future_; }; class SetLocalDescriptionObserver - : public webrtc::SetLocalDescriptionObserverInterface { + : public webrtc::SetLocalDescriptionObserverInterface, + public SetDescriptionObserverBase { public: - explicit SetLocalDescriptionObserver(Future* settable_future) - : settable_future_(settable_future) {} - void OnSetLocalDescriptionComplete(webrtc::RTCError error) override { - // On success, |error.ok()| is true. - if (error.ok()) { - settable_future_->Set(true); - return; - } - - NEARBY_LOG(ERROR, "Error when setting local session description: %s", - error.message()); - settable_future_->SetException({Exception::kFailed}); + OnSetDescriptionComplete(error); } - private: - std::unique_ptr> settable_future_; }; class SetRemoteDescriptionObserver - : public webrtc::SetRemoteDescriptionObserverInterface { + : public webrtc::SetRemoteDescriptionObserverInterface, + public SetDescriptionObserverBase { public: - explicit SetRemoteDescriptionObserver(Future* settable_future) - : settable_future_(settable_future) {} - void OnSetRemoteDescriptionComplete(webrtc::RTCError error) override { - // On success, |error.ok()| is true. - if (error.ok()) { - settable_future_->Set(true); - return; - } - - NEARBY_LOG(ERROR, "Error when setting remote session description: %s", - error.message()); - settable_future_->SetException({Exception::kFailed}); + OnSetDescriptionComplete(error); } - private: - std::unique_ptr> settable_future_; }; using PeerConnectionState = webrtc::PeerConnectionInterface::PeerConnectionState; -} // namespace - std::unique_ptr ConnectionFlow::Create( LocalIceCandidateListener local_ice_candidate_listener, DataChannelListener data_channel_listener, WebRtcMedium& webrtc_medium) { @@ -127,83 +133,110 @@ ConnectionFlow::ConnectionFlow( : data_channel_listener_(std::move(data_channel_listener)), local_ice_candidate_listener_(std::move(local_ice_candidate_listener)) {} -ConnectionFlow::~ConnectionFlow() { Close(); } - -ConnectionFlow::State ConnectionFlow::GetState() { - MutexLock lock(&mutex_); - return state_; +ConnectionFlow::~ConnectionFlow() { + NEARBY_LOG(INFO, "~ConnectionFlow"); + CountDownLatch latch(1); + if (RunOnSignalingThread([this, latch]() mutable { + CloseOnSignalingThread(); + latch.CountDown(); + })) { + latch.Await(); + } } SessionDescriptionWrapper ConnectionFlow::CreateOffer() { - MutexLock lock(&mutex_); - - if (!TransitionState(State::kInitialized, State::kCreatingOffer)) { + CHECK(!IsRunningOnSignalingThread()); + Future success_future; + if (!RunOnSignalingThread([this, success_future] { + CreateOfferOnSignalingThread(success_future); + })) { + NEARBY_LOG(ERROR, "Failed to create offer"); return SessionDescriptionWrapper(); } - - webrtc::DataChannelInit data_channel_init; - data_channel_init.reliable = true; - rtc::scoped_refptr data_channel = - peer_connection_->CreateDataChannel(kDataChannelName, &data_channel_init); - RegisterDataChannelObserver(data_channel); - - auto success_future = new Future(); - webrtc::PeerConnectionInterface::RTCOfferAnswerOptions options; - rtc::scoped_refptr observer = - new rtc::RefCountedObject( - success_future); - peer_connection_->CreateOffer(observer, options); - - ExceptionOr result = success_future->Get(kTimeout); - if (result.ok() && - TransitionState(State::kCreatingOffer, State::kWaitingForAnswer)) { + ExceptionOr result = success_future.Get(kTimeout); + if (result.ok()) { return std::move(result.result()); } - NEARBY_LOG(ERROR, "Failed to create offer: %d", result.exception()); return SessionDescriptionWrapper(); } -SessionDescriptionWrapper ConnectionFlow::CreateAnswer() { - MutexLock lock(&mutex_); - - if (!TransitionState(State::kReceivedOffer, State::kCreatingAnswer)) { - return SessionDescriptionWrapper(); +void ConnectionFlow::CreateOfferOnSignalingThread( + Future success_future) { + if (!TransitionState(State::kInitialized, State::kCreatingOffer)) { + success_future.SetException({Exception::kFailed}); + return; } + webrtc::DataChannelInit data_channel_init; + data_channel_init.reliable = true; + auto pc = GetPeerConnection(); + rtc::scoped_refptr data_channel = + pc->CreateDataChannel(kDataChannelName, &data_channel_init); + RegisterDataChannelObserver(data_channel); - auto success_future = new Future(); webrtc::PeerConnectionInterface::RTCOfferAnswerOptions options; rtc::scoped_refptr observer = new rtc::RefCountedObject( - success_future); - peer_connection_->CreateAnswer(observer, options); + this, success_future, State::kCreatingOffer, + State::kWaitingForAnswer); + pc->CreateOffer(observer, options); +} - ExceptionOr result = success_future->Get(kTimeout); - if (result.ok() && - TransitionState(State::kCreatingAnswer, State::kWaitingToConnect)) { +SessionDescriptionWrapper ConnectionFlow::CreateAnswer() { + CHECK(!IsRunningOnSignalingThread()); + Future success_future; + if (!RunOnSignalingThread([this, success_future] { + CreateAnswerOnSignalingThread(success_future); + })) { + NEARBY_LOG(ERROR, "Failed to create answer"); + return SessionDescriptionWrapper(); + } + ExceptionOr result = success_future.Get(kTimeout); + if (result.ok()) { return std::move(result.result()); } - NEARBY_LOG(ERROR, "Failed to create answer: %d", result.exception()); return SessionDescriptionWrapper(); } -bool ConnectionFlow::SetLocalSessionDescription(SessionDescriptionWrapper sdp) { - MutexLock lock(&mutex_); +void ConnectionFlow::CreateAnswerOnSignalingThread( + Future success_future) { + if (!TransitionState(State::kReceivedOffer, State::kCreatingAnswer)) { + success_future.SetException({Exception::kFailed}); + return; + } + webrtc::PeerConnectionInterface::RTCOfferAnswerOptions options; + rtc::scoped_refptr observer = + new rtc::RefCountedObject( + this, success_future, State::kCreatingAnswer, + State::kWaitingToConnect); + auto pc = GetPeerConnection(); + pc->CreateAnswer(observer, options); +} - if (state_ == State::kEnded) return false; +bool ConnectionFlow::SetLocalSessionDescription(SessionDescriptionWrapper sdp) { + CHECK(!IsRunningOnSignalingThread()); if (!sdp.IsValid()) return false; - auto success_future = new Future(); rtc::scoped_refptr observer = - new rtc::RefCountedObject( - success_future); + new rtc::RefCountedObject(); - peer_connection_->SetLocalDescription( - std::unique_ptr(sdp.Release()), - observer); + if (!RunOnSignalingThread([this, observer, sdp = std::move(sdp)]() mutable { + if (state_ == State::kEnded) { + observer->OnSetLocalDescriptionComplete( + webrtc::RTCError(webrtc::RTCErrorType::INVALID_STATE)); + return; + } + auto pc = GetPeerConnection(); - ExceptionOr result = success_future->Get(kTimeout); + pc->SetLocalDescription( + std::unique_ptr(sdp.Release()), + observer); + })) { + return false; + } + + ExceptionOr result = observer->GetResult(kTimeout); bool success = result.ok() && result.result(); if (!success) { NEARBY_LOG(ERROR, "Failed to set local session description: %d", @@ -212,20 +245,31 @@ bool ConnectionFlow::SetLocalSessionDescription(SessionDescriptionWrapper sdp) { return success; } -bool ConnectionFlow::SetRemoteSessionDescription( - SessionDescriptionWrapper sdp) { +bool ConnectionFlow::SetRemoteSessionDescription(SessionDescriptionWrapper sdp, + State expected_entry_state, + State exit_state) { if (!sdp.IsValid()) return false; - auto success_future = new Future(); rtc::scoped_refptr observer = - new rtc::RefCountedObject( - success_future); + new rtc::RefCountedObject(); - peer_connection_->SetRemoteDescription( - std::unique_ptr(sdp.Release()), - observer); + if (!RunOnSignalingThread([this, observer, sdp = std::move(sdp), + expected_entry_state, exit_state]() mutable { + if (!TransitionState(expected_entry_state, exit_state)) { + observer->OnSetRemoteDescriptionComplete( + webrtc::RTCError(webrtc::RTCErrorType::INVALID_STATE)); + return; + } + auto pc = GetPeerConnection(); - ExceptionOr result = success_future->Get(kTimeout); + pc->SetRemoteDescription( + std::unique_ptr(sdp.Release()), + observer); + })) { + return false; + } + + ExceptionOr result = observer->GetResult(kTimeout); bool success = result.ok() && result.result(); if (!success) { NEARBY_LOG(ERROR, "Failed to set remote description: %d", @@ -235,52 +279,79 @@ bool ConnectionFlow::SetRemoteSessionDescription( } bool ConnectionFlow::OnOfferReceived(SessionDescriptionWrapper offer) { - MutexLock lock(&mutex_); - - if (!TransitionState(State::kInitialized, State::kReceivedOffer)) { - return false; - } - return SetRemoteSessionDescription(std::move(offer)); + CHECK(!IsRunningOnSignalingThread()); + return SetRemoteSessionDescription(std::move(offer), State::kInitialized, + State::kReceivedOffer); } bool ConnectionFlow::OnAnswerReceived(SessionDescriptionWrapper answer) { - MutexLock lock(&mutex_); - - if (!TransitionState(State::kWaitingForAnswer, State::kWaitingToConnect)) { - return false; - } - return SetRemoteSessionDescription(std::move(answer)); + CHECK(!IsRunningOnSignalingThread()); + return SetRemoteSessionDescription( + std::move(answer), State::kWaitingForAnswer, State::kWaitingToConnect); } bool ConnectionFlow::OnRemoteIceCandidatesReceived( std::vector> ice_candidates) { - MutexLock lock(&mutex_); + CHECK(!IsRunningOnSignalingThread()); + // We can't call RunOnSignalingThread because C++ wants to copy ice_candidates + // if we try. unique_ptr is not CopyConstructible and compilation fails. + auto pc = GetPeerConnection(); + if (!pc) { + return false; + } + pc->signaling_thread()->PostTask( + RTC_FROM_HERE, [this, can_run_tasks = std::weak_ptr(can_run_tasks_), + candidates = std::move(ice_candidates)]() mutable { + // don't run the task if the weak_ptr is no longer valid. + if (!can_run_tasks.lock()) { + return; + } + AddIceCandidatesOnSignalingThread(std::move(candidates)); + }); + return true; +} + +void ConnectionFlow::AddIceCandidatesOnSignalingThread( + std::vector> + ice_candidates) { + CHECK(IsRunningOnSignalingThread()); if (state_ == State::kEnded) { NEARBY_LOG(WARNING, "You cannot add ice candidates to a disconnected session."); - return false; + return; } - if (state_ != State::kWaitingToConnect && state_ != State::kConnected) { cached_remote_ice_candidates_.insert( cached_remote_ice_candidates_.end(), std::make_move_iterator(ice_candidates.begin()), std::make_move_iterator(ice_candidates.end())); - return true; + return; } - + auto pc = GetPeerConnection(); for (auto&& ice_candidate : ice_candidates) { - if (!peer_connection_->AddIceCandidate(ice_candidate.get())) { + if (!pc->AddIceCandidate(ice_candidate.get())) { NEARBY_LOG(WARNING, "Unable to add remote ice candidate."); } } - return true; } -bool ConnectionFlow::Close() { - return Close(/* close_peer_connection= */ true); +bool ConnectionFlow::CloseIfNotConnected() { + CHECK(!IsRunningOnSignalingThread()); + Future closed; + if (RunOnSignalingThread([this, closed]() mutable { + if (state_ == State::kConnected) { + closed.Set(false); + } else { + CloseOnSignalingThread(); + closed.Set(true); + } + })) { + auto result = closed.Get(); + return result.ok() && result.result(); + } + return true; } bool ConnectionFlow::InitPeerConnection(WebRtcMedium& webrtc_medium) { @@ -303,7 +374,10 @@ bool ConnectionFlow::InitPeerConnection(WebRtcMedium& webrtc_medium) { // 1) this is the 2nd call of this callback (and this is a bug), or // 2) Get(timeout) has set the future value as exception already. if (success_future.IsSet()) return; + MutexLock lock(&mutex_); peer_connection_ = peer_connection; + signaling_thread_for_dcheck_only_ = + peer_connection_->signaling_thread(); success_future.Set(true); }); @@ -317,29 +391,26 @@ bool ConnectionFlow::InitPeerConnection(WebRtcMedium& webrtc_medium) { } void ConnectionFlow::OnSignalingStable() { - OffloadFromSignalingThread([this] { - MutexLock lock(&mutex_); - - if (state_ != State::kWaitingToConnect && state_ != State::kConnected) - return; - - for (auto&& ice_candidate : cached_remote_ice_candidates_) { - if (!peer_connection_->AddIceCandidate(ice_candidate.get())) { - NEARBY_LOG(WARNING, "Unable to add remote ice candidate."); - } + if (state_ != State::kWaitingToConnect && state_ != State::kConnected) return; + auto pc = GetPeerConnection(); + for (auto&& ice_candidate : cached_remote_ice_candidates_) { + if (!pc->AddIceCandidate(ice_candidate.get())) { + NEARBY_LOG(WARNING, "Unable to add remote ice candidate."); } - cached_remote_ice_candidates_.clear(); - }); + } + cached_remote_ice_candidates_.clear(); } void ConnectionFlow::OnIceCandidate( const webrtc::IceCandidateInterface* candidate) { + CHECK(IsRunningOnSignalingThread()); local_ice_candidate_listener_.local_ice_candidate_found_cb(candidate); } void ConnectionFlow::OnSignalingChange( webrtc::PeerConnectionInterface::SignalingState new_state) { NEARBY_LOG(INFO, "OnSignalingChange: %d", new_state); + CHECK(IsRunningOnSignalingThread()); if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable) { OnSignalingStable(); } @@ -348,38 +419,38 @@ void ConnectionFlow::OnSignalingChange( void ConnectionFlow::OnDataChannel( rtc::scoped_refptr data_channel) { NEARBY_LOG(INFO, "OnDataChannel"); + CHECK(IsRunningOnSignalingThread()); RegisterDataChannelObserver(std::move(data_channel)); } void ConnectionFlow::OnIceGatheringChange( webrtc::PeerConnectionInterface::IceGatheringState new_state) { NEARBY_LOG(INFO, "OnIceGatheringChange: %d", new_state); + CHECK(IsRunningOnSignalingThread()); } void ConnectionFlow::OnConnectionChange( webrtc::PeerConnectionInterface::PeerConnectionState new_state) { NEARBY_LOG(INFO, "OnConnectionChange: %d", new_state); + CHECK(IsRunningOnSignalingThread()); if (new_state == PeerConnectionState::kClosed || new_state == PeerConnectionState::kFailed || new_state == PeerConnectionState::kDisconnected) { - // kClosed means that PeerConnection is already closed or - // is closing right now - PeerConnection::Close() triggered - // PeerConnectionObserver::OnConnectionChange(kClosed). - // We must not call PeerConnection::Close() again on that code path NEARBY_LOG(INFO, "Closing due to peer connection state change: %d", new_state); - Close(new_state != PeerConnectionState::kClosed); + CloseOnSignalingThread(); } } void ConnectionFlow::OnRenegotiationNeeded() { NEARBY_LOG(INFO, "OnRenegotiationNeeded"); + CHECK(IsRunningOnSignalingThread()); } -void ConnectionFlow::ProcessDataChannelConnected( +void ConnectionFlow::ProcessDataChannelConnectedOnSignalingThread( rtc::scoped_refptr data_channel) { - MutexLock lock(&mutex_); NEARBY_LOG(INFO, "Data channel state changed to connected."); + CHECK(IsRunningOnSignalingThread()); if (!TransitionState(State::kWaitingToConnect, State::kConnected)) { data_channel->Close(); return; @@ -390,14 +461,12 @@ void ConnectionFlow::ProcessDataChannelConnected( void ConnectionFlow::RegisterDataChannelObserver( rtc::scoped_refptr data_channel) { + CHECK(IsRunningOnSignalingThread()); if (!data_channel_observer_) { auto state_change_callback = [this, data_channel]() { if (data_channel->state() == webrtc::DataChannelInterface::DataState::kOpen) { - OffloadFromSignalingThread( - [this, data_channel{std::move(data_channel)}]() { - ProcessDataChannelConnected(std::move(data_channel)); - }); + ProcessDataChannelConnectedOnSignalingThread(std::move(data_channel)); } }; data_channel_observer_ = absl::make_unique( @@ -410,6 +479,7 @@ void ConnectionFlow::RegisterDataChannelObserver( } bool ConnectionFlow::TransitionState(State current_state, State new_state) { + CHECK(IsRunningOnSignalingThread()); if (current_state != state_) { NEARBY_LOG( WARNING, @@ -417,33 +487,72 @@ bool ConnectionFlow::TransitionState(State current_state, State new_state) { new_state, state_, current_state); return false; } + NEARBY_LOG(INFO, "Transition: %d -> %d", state_, new_state); state_ = new_state; return true; } -bool ConnectionFlow::Close(bool close_peer_connection) { - { - MutexLock lock(&mutex_); - if (state_ == State::kEnded) { - return false; - } - state_ = State::kEnded; +bool ConnectionFlow::CloseOnSignalingThread() { + if (state_ == State::kEnded) { + return false; } + state_ = State::kEnded; + auto pc = GetAndResetPeerConnection(); + NEARBY_LOG(INFO, "Closing WebRTC connection."); - - single_threaded_signaling_offloader_.Shutdown(); - - if (peer_connection_ && close_peer_connection) peer_connection_->Close(); - + if (pc) pc->Close(); data_channel_observer_.reset(); + can_run_tasks_.reset(); NEARBY_LOG(INFO, "Closed WebRTC connection."); return true; } -void ConnectionFlow::OffloadFromSignalingThread(Runnable runnable) { - single_threaded_signaling_offloader_.Execute(std::move(runnable)); +bool ConnectionFlow::RunOnSignalingThread(Runnable&& runnable) { + CHECK(!IsRunningOnSignalingThread()); + auto pc = GetPeerConnection(); + if (!pc) { + NEARBY_LOG(WARNING, + "Peer connection not available. Cannot schedule tasks."); + return false; + } + // We are off signaling thread, so we can't use peer connection's methods + // but we can access the signaling thread handle. + pc->signaling_thread()->PostTask( + RTC_FROM_HERE, [can_run_tasks = std::weak_ptr(can_run_tasks_), + task = std::move(runnable)] { + // don't run the task if the weak_ptr is no longer valid. + // shared_ptr |can_run_tasks_| is destroyed on the same thread + // (signaling thread). This guarantees that if the weak_ptr is valid + // when this task starts, it will stay valid until the task ends. + if (!can_run_tasks.lock()) { + NEARBY_LOG(VERBOSE, + "Peer connection already closed. Cannot run tasks."); + return; + } + task(); + }); + return true; } +bool ConnectionFlow::IsRunningOnSignalingThread() { + return signaling_thread_for_dcheck_only_ == rtc::Thread::Current(); +} + +rtc::scoped_refptr +ConnectionFlow::GetPeerConnection() { + // We must use a mutex to ensure that peer connection is + // fully initialized. + // We increase the peer_connection_'s refcount to keep it + // alive while we use it. + MutexLock lock(&mutex_); + return peer_connection_; +} + +rtc::scoped_refptr +ConnectionFlow::GetAndResetPeerConnection() { + MutexLock lock(&mutex_); + return std::move(peer_connection_); +} } // namespace mediums } // namespace connections } // namespace nearby diff --git a/cpp/core/internal/mediums/webrtc/connection_flow.h b/cpp/core/internal/mediums/webrtc/connection_flow.h index 101da1c5..bf8b7f3e 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow.h +++ b/cpp/core/internal/mediums/webrtc/connection_flow.h @@ -79,43 +79,49 @@ class ConnectionFlow : public webrtc::PeerConnectionObserver { }; // This method blocks on the creation of the peer connection object. + // Can be called on any thread but never called on signaling thread. static std::unique_ptr Create( LocalIceCandidateListener local_ice_candidate_listener, DataChannelListener data_channel_listener, WebRtcMedium& webrtc_medium); ~ConnectionFlow() override; - // Returns the current state of the ConnectionFlow. - State GetState() ABSL_LOCKS_EXCLUDED(mutex_); - // Create the offer that will be sent to the remote. Mirrors the behaviour of // PeerConnectionInterface::CreateOffer. + // Can be called on any thread but never called on signaling thread. SessionDescriptionWrapper CreateOffer() ABSL_LOCKS_EXCLUDED(mutex_); // Create the answer that will be sent to the remote. Mirrors the behaviour of // PeerConnectionInterface::CreateAnswer. + // Can be called on any thread but never called on signaling thread. SessionDescriptionWrapper CreateAnswer() ABSL_LOCKS_EXCLUDED(mutex_); // Set the local session description. |sdp| was created via CreateOffer() // or CreateAnswer(). + // Can be called on any thread but never called on signaling thread. bool SetLocalSessionDescription(SessionDescriptionWrapper sdp) ABSL_LOCKS_EXCLUDED(mutex_); // Invoked when an offer was received from a remote; this will set the remote // session description on the peer connection. Returns true if the offer was // successfully set as remote session description. + // Can be called on any thread but never called on signaling thread. bool OnOfferReceived(SessionDescriptionWrapper offer) ABSL_LOCKS_EXCLUDED(mutex_); // Invoked when an answer was received from a remote; this will set the remote // session description on the peer connection. Returns true if the offer was // successfully set as remote session description. + // Can be called on any thread but never called on signaling thread. bool OnAnswerReceived(SessionDescriptionWrapper answer) ABSL_LOCKS_EXCLUDED(mutex_); // Invoked when an ice candidate was received from a remote; this will add the // ice candidate to the peer connection if ready or cache it otherwise. + // Can be called on any thread but never called on signaling thread. bool OnRemoteIceCandidatesReceived( std::vector> ice_candidates) ABSL_LOCKS_EXCLUDED(mutex_); - // Close the peer connection and data channel. - bool Close() ABSL_LOCKS_EXCLUDED(mutex_); + // Close the peer connection and data channel if not connected. + // Can be called on any thread but never called on signaling thread. + bool CloseIfNotConnected() ABSL_LOCKS_EXCLUDED(mutex_); // webrtc::PeerConnectionObserver: + // All methods called only on signaling thread. void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override; void OnSignalingChange( webrtc::PeerConnectionInterface::SignalingState new_state) override; @@ -127,15 +133,23 @@ class ConnectionFlow : public webrtc::PeerConnectionObserver { webrtc::PeerConnectionInterface::PeerConnectionState new_state) override; void OnRenegotiationNeeded() override; - // For tests only - webrtc::PeerConnectionInterface* GetPeerConnection() { - return peer_connection_.get(); - } + // Public because it's used in tests too. + rtc::scoped_refptr GetPeerConnection(); private: ConnectionFlow(LocalIceCandidateListener local_ice_candidate_listener, DataChannelListener data_channel_listener); + // Resets peer connection reference. Returns old value. + rtc::scoped_refptr + GetAndResetPeerConnection(); + void CreateOfferOnSignalingThread( + Future success_future); + void CreateAnswerOnSignalingThread( + Future success_future); + void AddIceCandidatesOnSignalingThread( + std::vector> + ice_candidates); // Invoked when the peer connection indicates that signaling is stable. void OnSignalingStable() ABSL_LOCKS_EXCLUDED(mutex_); void RegisterDataChannelObserver( @@ -148,34 +162,63 @@ class ConnectionFlow : public webrtc::PeerConnectionObserver { bool InitPeerConnection(WebRtcMedium& webrtc_medium); - bool TransitionState(State current_state, State new_state) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + bool TransitionState(State current_state, State new_state); - bool SetRemoteSessionDescription(SessionDescriptionWrapper sdp); + bool SetRemoteSessionDescription(SessionDescriptionWrapper sdp, + State expected_entry_state, + State exit_state); - void ProcessDataChannelConnected( + void ProcessDataChannelConnectedOnSignalingThread( rtc::scoped_refptr) ABSL_LOCKS_EXCLUDED(mutex_); - void CloseAndNotifyLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - bool Close(bool close_peer_connection) ABSL_LOCKS_EXCLUDED(mutex_); + bool CloseOnSignalingThread() ABSL_LOCKS_EXCLUDED(mutex_); - void OffloadFromSignalingThread(Runnable runnable); + bool RunOnSignalingThread(Runnable&& runnable); + bool IsRunningOnSignalingThread(); Mutex mutex_; - State state_ ABSL_GUARDED_BY(mutex_) = State::kInitialized; + // State is used on signaling thread only. + State state_ = State::kInitialized; DataChannelListener data_channel_listener_; std::unique_ptr data_channel_observer_; LocalIceCandidateListener local_ice_candidate_listener_; - rtc::scoped_refptr peer_connection_; + // Peer connection can be used only on signaling thread. The only exception + // is accessing the signaling thread handle. Tasks posted on the + // signaling thread may outlive both |peer_connection_| and |this| objects. + // A mutex is required to access peer connection reference because peer + // connection object and the reference can be initialized on different + // threads - the reference could be initialized before peer connection's + // constructor has finished. + // |peer_connection_| is actually implemented by PeerConnectionProxy, which + // runs the real PeerConnection's methods on the correct thread (signaling or + // worker). If a proxy method is called on the correct thread, then the real + // method is called directly. Otherwise, a task is posted on the correct + // thread and the current thread is blocked until that task finishes. We + // choose to explicitly use |peer_connection_| on the signaling thread, + // because it allows us to do state management on the signaling thread too, + // simplifies locking, and we don't have to block the current thread for every + // peer connection call. + rtc::scoped_refptr peer_connection_ + ABSL_GUARDED_BY(mutex_); std::vector> - cached_remote_ice_candidates_ ABSL_GUARDED_BY(mutex_); + cached_remote_ice_candidates_; + // This pointer is only for DCHECK() assertions. + // It allows us to check if we are running on signaling thread even + // after destroying |peer_connection_|. + const void* signaling_thread_for_dcheck_only_ = nullptr; + // This shared_ptr is reset on the signaling thread when ConnectionFlow is + // closed. This prevents us from running tasks on the signaling thread when + // peer connection is closed. The value stored in |can_run_tasks_| is not + // used. We are using std::shared_ptr instead of rtc::WeakPtrFactory because + // the former is thread-safe. + std::shared_ptr can_run_tasks_ = std::make_shared(); - SingleThreadExecutor single_threaded_signaling_offloader_; + friend class CreateSessionDescriptionObserverImpl; }; } // namespace mediums diff --git a/cpp/core/internal/mediums/webrtc/connection_flow_test.cc b/cpp/core/internal/mediums/webrtc/connection_flow_test.cc index 45e71632..4db8cb9f 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow_test.cc +++ b/cpp/core/internal/mediums/webrtc/connection_flow_test.cc @@ -20,6 +20,7 @@ #include "core/internal/mediums/webrtc/session_description_wrapper.h" #include "platform/base/byte_array.h" #include "platform/base/medium_environment.h" +#include "platform/public/count_down_latch.h" #include "platform/public/webrtc.h" #include "gmock/gmock.h" #include "gtest/gtest.h" @@ -173,7 +174,7 @@ TEST_F(ConnectionFlowTest, CannotCreateOfferAfterClose) { LocalIceCandidateListener(), DataChannelListener(), webrtc_medium); ASSERT_NE(offerer, nullptr); - EXPECT_TRUE(offerer->Close()); + EXPECT_TRUE(offerer->CloseIfNotConnected()); EXPECT_FALSE(offerer->CreateOffer().IsValid()); } @@ -188,7 +189,7 @@ TEST_F(ConnectionFlowTest, CannotSetSessionDescriptionAfterClose) { SessionDescriptionWrapper offer = offerer->CreateOffer(); EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer); - EXPECT_TRUE(offerer->Close()); + EXPECT_TRUE(offerer->CloseIfNotConnected()); EXPECT_FALSE(offerer->SetLocalSessionDescription(offer)); } @@ -205,7 +206,7 @@ TEST_F(ConnectionFlowTest, CannotReceiveOfferAfterClose) { webrtc_medium_answerer); ASSERT_NE(answerer, nullptr); - EXPECT_TRUE(answerer->Close()); + EXPECT_TRUE(answerer->CloseIfNotConnected()); SessionDescriptionWrapper offer = offerer->CreateOffer(); EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer); @@ -311,7 +312,13 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) { answerer_channel = answerer_data_channel_future.Get(absl::Seconds(1)); EXPECT_TRUE(answerer_channel.ok()); - answerer->GetPeerConnection()->Close(); + CountDownLatch latch(1); + auto pc = answerer->GetPeerConnection(); + pc->signaling_thread()->PostTask(RTC_FROM_HERE, [pc, latch]() mutable { + pc->Close(); + latch.CountDown(); + }); + latch.Await(); // Send message on data channel const char message[] = "Test"; @@ -393,7 +400,13 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) { answerer_channel = answerer_data_channel_future.Get(absl::Seconds(1)); EXPECT_TRUE(answerer_channel.ok()); - offerer->GetPeerConnection()->Close(); + CountDownLatch latch(1); + auto pc = offerer->GetPeerConnection(); + pc->signaling_thread()->PostTask(RTC_FROM_HERE, [pc, latch]() mutable { + pc->Close(); + latch.CountDown(); + }); + latch.Await(); // Send message on data channel const char message[] = "Test";