Internal change

PiperOrigin-RevId: 368765308

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<ConnectionFlow> 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 =

cpp/core/internal/mediums/webrtc/connection_flow.cc

@@ -18,6 +18,7 @@
 #include <memory>
 
 #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<SessionDescriptionWrapper>* settable_future)
-      : settable_future_(settable_future) {}
-  ~CreateSessionDescriptionObserverImpl() override = default;
+  CreateSessionDescriptionObserverImpl(
+      ConnectionFlow* connection_flow,
+      Future<SessionDescriptionWrapper> 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<Future<SessionDescriptionWrapper>> settable_future_;
+  ConnectionFlow* connection_flow_;
+  Future<SessionDescriptionWrapper> settable_future_;
+  ConnectionFlow::State expected_entry_state_;
+  ConnectionFlow::State exit_state_;
+};
+
+class SetDescriptionObserverBase {
+ public:
+  ExceptionOr<bool> 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<bool> settable_future_;
 };
 
 class SetLocalDescriptionObserver
-    : public webrtc::SetLocalDescriptionObserverInterface {
+    : public webrtc::SetLocalDescriptionObserverInterface,
+      public SetDescriptionObserverBase {
  public:
-  explicit SetLocalDescriptionObserver(Future<bool>* 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<Future<bool>> settable_future_;
 };
 
 class SetRemoteDescriptionObserver
-    : public webrtc::SetRemoteDescriptionObserverInterface {
+    : public webrtc::SetRemoteDescriptionObserverInterface,
+      public SetDescriptionObserverBase {
  public:
-  explicit SetRemoteDescriptionObserver(Future<bool>* 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<Future<bool>> settable_future_;
 };
 
 using PeerConnectionState =
     webrtc::PeerConnectionInterface::PeerConnectionState;
 
-}  // namespace
-
 std::unique_ptr<ConnectionFlow> 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<SessionDescriptionWrapper> 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<webrtc::DataChannelInterface> data_channel =
-      peer_connection_->CreateDataChannel(kDataChannelName, &data_channel_init);
-  RegisterDataChannelObserver(data_channel);
-
-  auto success_future = new Future<SessionDescriptionWrapper>();
-  webrtc::PeerConnectionInterface::RTCOfferAnswerOptions options;
-  rtc::scoped_refptr<CreateSessionDescriptionObserverImpl> observer =
-      new rtc::RefCountedObject<CreateSessionDescriptionObserverImpl>(
-          success_future);
-  peer_connection_->CreateOffer(observer, options);
-
-  ExceptionOr<SessionDescriptionWrapper> result = success_future->Get(kTimeout);
-  if (result.ok() &&
-      TransitionState(State::kCreatingOffer, State::kWaitingForAnswer)) {
+  ExceptionOr<SessionDescriptionWrapper> 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<SessionDescriptionWrapper> 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<webrtc::DataChannelInterface> data_channel =
+      pc->CreateDataChannel(kDataChannelName, &data_channel_init);
+  RegisterDataChannelObserver(data_channel);
 
-  auto success_future = new Future<SessionDescriptionWrapper>();
   webrtc::PeerConnectionInterface::RTCOfferAnswerOptions options;
   rtc::scoped_refptr<CreateSessionDescriptionObserverImpl> observer =
       new rtc::RefCountedObject<CreateSessionDescriptionObserverImpl>(
-          success_future);
-  peer_connection_->CreateAnswer(observer, options);
+          this, success_future, State::kCreatingOffer,
+          State::kWaitingForAnswer);
+  pc->CreateOffer(observer, options);
+}
 
-  ExceptionOr<SessionDescriptionWrapper> result = success_future->Get(kTimeout);
-  if (result.ok() &&
-      TransitionState(State::kCreatingAnswer, State::kWaitingToConnect)) {
+SessionDescriptionWrapper ConnectionFlow::CreateAnswer() {
+  CHECK(!IsRunningOnSignalingThread());
+  Future<SessionDescriptionWrapper> success_future;
+  if (!RunOnSignalingThread([this, success_future] {
+        CreateAnswerOnSignalingThread(success_future);
+      })) {
+    NEARBY_LOG(ERROR, "Failed to create answer");
+    return SessionDescriptionWrapper();
+  }
+  ExceptionOr<SessionDescriptionWrapper> 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<SessionDescriptionWrapper> success_future) {
+  if (!TransitionState(State::kReceivedOffer, State::kCreatingAnswer)) {
+    success_future.SetException({Exception::kFailed});
+    return;
+  }
+  webrtc::PeerConnectionInterface::RTCOfferAnswerOptions options;
+  rtc::scoped_refptr<CreateSessionDescriptionObserverImpl> observer =
+      new rtc::RefCountedObject<CreateSessionDescriptionObserverImpl>(
+          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<bool>();
   rtc::scoped_refptr<SetLocalDescriptionObserver> observer =
-      new rtc::RefCountedObject<SetLocalDescriptionObserver>(
-          success_future);
+      new rtc::RefCountedObject<SetLocalDescriptionObserver>();
 
-  peer_connection_->SetLocalDescription(
-      std::unique_ptr<webrtc::SessionDescriptionInterface>(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<bool> result = success_future->Get(kTimeout);
+        pc->SetLocalDescription(
+            std::unique_ptr<webrtc::SessionDescriptionInterface>(sdp.Release()),
+            observer);
+      })) {
+    return false;
+  }
+
+  ExceptionOr<bool> 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<bool>();
   rtc::scoped_refptr<SetRemoteDescriptionObserver> observer =
-      new rtc::RefCountedObject<SetRemoteDescriptionObserver>(
-          success_future);
+      new rtc::RefCountedObject<SetRemoteDescriptionObserver>();
 
-  peer_connection_->SetRemoteDescription(
-      std::unique_ptr<webrtc::SessionDescriptionInterface>(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<bool> result = success_future->Get(kTimeout);
+        pc->SetRemoteDescription(
+            std::unique_ptr<webrtc::SessionDescriptionInterface>(sdp.Release()),
+            observer);
+      })) {
+    return false;
+  }
+
+  ExceptionOr<bool> 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<std::unique_ptr<webrtc::IceCandidateInterface>>
         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<void>(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<std::unique_ptr<webrtc::IceCandidateInterface>>
+        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<bool> 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<webrtc::DataChannelInterface> 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<webrtc::DataChannelInterface> 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<webrtc::DataChannelInterface> 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<DataChannelObserverImpl>(
@@ -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<void>(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<webrtc::PeerConnectionInterface>
+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<webrtc::PeerConnectionInterface>
+ConnectionFlow::GetAndResetPeerConnection() {
+  MutexLock lock(&mutex_);
+  return std::move(peer_connection_);
+}
 }  // namespace mediums
 }  // namespace connections
 }  // namespace nearby

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<ConnectionFlow> 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<std::unique_ptr<webrtc::IceCandidateInterface>>
           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<webrtc::PeerConnectionInterface> GetPeerConnection();
 
  private:
   ConnectionFlow(LocalIceCandidateListener local_ice_candidate_listener,
                  DataChannelListener data_channel_listener);
 
+  // Resets peer connection reference. Returns old value.
+  rtc::scoped_refptr<webrtc::PeerConnectionInterface>
+  GetAndResetPeerConnection();
+  void CreateOfferOnSignalingThread(
+      Future<SessionDescriptionWrapper> success_future);
+  void CreateAnswerOnSignalingThread(
+      Future<SessionDescriptionWrapper> success_future);
+  void AddIceCandidatesOnSignalingThread(
+      std::vector<std::unique_ptr<webrtc::IceCandidateInterface>>
+          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<webrtc::DataChannelInterface>)
       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<DataChannelObserverImpl> data_channel_observer_;
 
   LocalIceCandidateListener local_ice_candidate_listener_;
-  rtc::scoped_refptr<webrtc::PeerConnectionInterface> 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<webrtc::PeerConnectionInterface> peer_connection_
+      ABSL_GUARDED_BY(mutex_);
 
   std::vector<std::unique_ptr<webrtc::IceCandidateInterface>>
-      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<void> can_run_tasks_ = std::make_shared<int>();
 
-  SingleThreadExecutor single_threaded_signaling_offloader_;
+  friend class CreateSessionDescriptionObserverImpl;
 };
 
 }  // namespace mediums

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";