From a00ffc5e0dcd69767296cf6b99b7f8ca04252661 Mon Sep 17 00:00:00 2001 From: hai007 Date: Wed, 11 Nov 2020 20:47:55 -0800 Subject: [PATCH] Roll forward to cl/341966255 Signed-off-by: hai007 --- cpp/core/internal/mediums/webrtc.cc | 45 ++++++++--------- cpp/core/internal/mediums/webrtc.h | 29 +++-------- cpp/core/internal/mediums/webrtc/BUILD | 1 + cpp/core/internal/mediums/webrtc_test.cc | 53 +++++++++++++------- cpp/core/internal/p2p_cluster_pcp_handler.cc | 29 ++--------- cpp/core/internal/p2p_cluster_pcp_handler.h | 2 +- cpp/core/internal/webrtc_bwu_handler.cc | 11 ++-- 7 files changed, 73 insertions(+), 97 deletions(-) diff --git a/cpp/core/internal/mediums/webrtc.cc b/cpp/core/internal/mediums/webrtc.cc index d24b206c..96cdec3d 100644 --- a/cpp/core/internal/mediums/webrtc.cc +++ b/cpp/core/internal/mediums/webrtc.cc @@ -48,14 +48,14 @@ const std::string WebRtc::GetDefaultCountryCode() { bool WebRtc::IsAvailable() { return medium_.IsValid(); } -bool WebRtc::IsAcceptingConnections() { +bool WebRtc::IsAcceptingConnections(const std::string& service_id) { MutexLock lock(&mutex_); + // TODO(hais): refractor the implementation with maps. return role_ == Role::kOfferer; } bool WebRtc::StartAcceptingConnections(const PeerId& self_id, const std::string& service_id, - const std::string& local_endpoint_id, const LocationHint& location_hint, AcceptedConnectionCallback callback) { if (!IsAvailable()) { @@ -66,10 +66,17 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id, return false; } + if (IsAcceptingConnections(service_id)) { + NEARBY_LOG(WARNING, "Already accepting WebRTC connections."); + return false; + } + { MutexLock lock(&mutex_); - if (self_id_.GetId() == self_id.GetId()) { - NEARBY_LOG(WARNING, "Already accepting WebRTC connections."); + if (role_ != Role::kNone) { + NEARBY_LOG(WARNING, + "Cannot start accepting WebRTC connections, current role %d", + role_); return false; } @@ -77,7 +84,8 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id, restart_receive_messages_alarm_ = CancelableAlarm( "restart_receiving_messages_webrtc", - std::bind(&WebRtc::RestartReceiveMessages, this, location_hint), + std::bind(&WebRtc::RestartReceiveMessages, this, location_hint, + service_id), kRestartReceiveMessagesDuration, &restart_receive_messages_executor_); SessionDescriptionWrapper offer = connection_flow_->CreateOffer(); @@ -91,8 +99,6 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id, // the actual transport can begin. ListenForWebRtcSocketFuture(connection_flow_->GetDataChannel(), std::move(callback)); - latest_service_id_ = service_id; - latest_local_endpoint_id_ = local_endpoint_id; NEARBY_LOG(INFO, "Started listening for WebRtc connections as %s", self_id.GetId().c_str()); } @@ -151,11 +157,12 @@ bool WebRtc::SetLocalSessionDescription(SessionDescriptionWrapper sdp) { return true; } -void WebRtc::StopAcceptingConnections() { - if (!IsAcceptingConnections()) { +void WebRtc::StopAcceptingConnections(const std::string& service_id) { + if (!IsAcceptingConnections(service_id)) { NEARBY_LOG(INFO, "Skipped StopAcceptingConnections since we are not currently " - "accepting WebRTC connections"); + "accepting WebRTC connections for %s", + service_id.c_str()); return; } @@ -166,19 +173,6 @@ void WebRtc::StopAcceptingConnections() { NEARBY_LOG(INFO, "Stopped accepting WebRTC connections"); } -void WebRtc::StopAcceptingConnection(const std::string& service_id, - const std::string& local_endpoint_id) { - MutexLock lock(&mutex_); - if (service_id == latest_service_id_ && - local_endpoint_id == latest_local_endpoint_id_) { - StopAcceptingConnections(); - } else { - NEARBY_LOG(INFO, - "Skipped StopAcceptingConnection since we are not the latest" - "ongoing connection."); - } -} - Future WebRtc::ListenForWebRtcSocketFuture( Future> data_channel_future, @@ -486,8 +480,9 @@ void WebRtc::OffloadFromSignalingThread(Runnable runnable) { single_thread_executor_.Execute(std::move(runnable)); } -void WebRtc::RestartReceiveMessages(const LocationHint& location_hint) { - if (!IsAcceptingConnections()) { +void WebRtc::RestartReceiveMessages(const LocationHint& location_hint, + const std::string& service_id) { + if (!IsAcceptingConnections(service_id)) { NEARBY_LOG(INFO, "Skipping restart since we are not accepting connections."); return; diff --git a/cpp/core/internal/mediums/webrtc.h b/cpp/core/internal/mediums/webrtc.h index 43d772d8..33f39008 100644 --- a/cpp/core/internal/mediums/webrtc.h +++ b/cpp/core/internal/mediums/webrtc.h @@ -53,14 +53,9 @@ class WebRtc { // Runs on @MainThread. bool IsAvailable(); - // Returns if the device is ready to accept connections from remote devices. + // Returns if the device is accepting connection with specific service id. // Runs on @MainThread. - bool IsAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_); - - // Returns if the device is accepting connection with specific service id and - // local endpoint id. Runs on @MainThread. - bool IsAcceptingConnection(const std::string& service_id, - const std::string& local_endpoint_id) + bool IsAcceptingConnections(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); // Prepares the device to accept incoming WebRtc connections. Returns a @@ -68,21 +63,13 @@ class WebRtc { // Runs on @MainThread. bool StartAcceptingConnections(const PeerId& self_id, const std::string& service_id, - const std::string& local_endpoint_id, const LocationHint& location_hint, AcceptedConnectionCallback callback) ABSL_LOCKS_EXCLUDED(mutex_); - // Prevents device from accepting future connections until - // StartAcceptingConnections() is called. - // Runs on @MainThread. - void StopAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_); - - // Try to stop (accepting) the specific connection with provided service id - // and local endpoint id. If the specific connection is not the latest one, - // then nothing will happen; if it's the latest one, - void StopAcceptingConnection(const std::string& service_id, - const std::string& local_endpoint_id) + // Try to stop (accepting) the specific connection with provided service id. + // Runs on @MainThread + void StopAcceptingConnections(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); // Initiates a WebRtc connection with peer device identified by |peer_id|. @@ -161,7 +148,8 @@ class WebRtc { void OffloadFromSignalingThread(Runnable runnable); // Runs on |restart_receive_messages_executor_|. - void RestartReceiveMessages(const LocationHint& location_hint) + void RestartReceiveMessages(const LocationHint& location_hint, + const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); Mutex mutex_; @@ -184,9 +172,6 @@ class WebRtc { // Restarts the signaling messenger for receiving messages. ScheduledExecutor restart_receive_messages_executor_; CancelableAlarm restart_receive_messages_alarm_; - - std::string latest_service_id_ ABSL_GUARDED_BY(mutex_); - std::string latest_local_endpoint_id_ ABSL_GUARDED_BY(mutex_); }; } // namespace mediums diff --git a/cpp/core/internal/mediums/webrtc/BUILD b/cpp/core/internal/mediums/webrtc/BUILD index cb614815..9cd3593d 100644 --- a/cpp/core/internal/mediums/webrtc/BUILD +++ b/cpp/core/internal/mediums/webrtc/BUILD @@ -46,6 +46,7 @@ cc_test( "signaling_frames_test.cc", "webrtc_socket_test.cc", ], + tags = ["notsan"], # NOTE(b/139734036): known data race in usrsctplib. deps = [ ":webrtc", "//platform/base", diff --git a/cpp/core/internal/mediums/webrtc_test.cc b/cpp/core/internal/mediums/webrtc_test.cc index 4874e105..ae5c599e 100644 --- a/cpp/core/internal/mediums/webrtc_test.cc +++ b/cpp/core/internal/mediums/webrtc_test.cc @@ -26,7 +26,7 @@ class WebRtcTest : public ::testing::Test { TEST_F(WebRtcTest, NotAcceptingConnections) { WebRtc webrtc; ASSERT_TRUE(webrtc.IsAvailable()); - EXPECT_FALSE(webrtc.IsAcceptingConnections()); + EXPECT_FALSE(webrtc.IsAcceptingConnections(std::string{})); } // Tests the flow when the device tries to accept connections twice. In this @@ -38,14 +38,17 @@ TEST_F(WebRtcTest, StartAcceptingConnectionTwice) { WebRtc webrtc; PeerId self_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint{}; ASSERT_TRUE(webrtc.IsAvailable()); ASSERT_TRUE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); EXPECT_FALSE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); - EXPECT_TRUE(webrtc.IsAcceptingConnections()); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); + EXPECT_TRUE(webrtc.IsAcceptingConnections(std::string{})); } // Tests the flow when the device tries to connect but the data channel times @@ -53,14 +56,15 @@ TEST_F(WebRtcTest, StartAcceptingConnectionTwice) { TEST_F(WebRtcTest, Connect_DataChannelTimeOut) { WebRtc webrtc; PeerId peer_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); WebRtcSocketWrapper wrapper_1 = webrtc.Connect(peer_id, location_hint); EXPECT_FALSE(wrapper_1.IsValid()); - EXPECT_TRUE(webrtc.StartAcceptingConnections(peer_id, location_hint, - AcceptedConnectionCallback())); + EXPECT_TRUE(webrtc.StartAcceptingConnections( + peer_id, service_id, location_hint, AcceptedConnectionCallback())); } // Tests the flow when the device calls Connect() after calling @@ -72,17 +76,20 @@ TEST_F(WebRtcTest, StartAcceptingConnection_ThenConnect) { WebRtc webrtc; PeerId self_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); ASSERT_TRUE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); WebRtcSocketWrapper wrapper = webrtc.Connect(PeerId("random_peer_id"), location_hint); - EXPECT_TRUE(webrtc.IsAcceptingConnections()); + EXPECT_TRUE(webrtc.IsAcceptingConnections(std::string{})); EXPECT_FALSE(wrapper.IsValid()); EXPECT_FALSE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); } // Tests the flow when the device calls StartAcceptingConnections but the medium @@ -94,13 +101,15 @@ TEST_F(WebRtcTest, StartAndStopAcceptingConnections) { WebRtc webrtc; PeerId self_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); ASSERT_TRUE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); - webrtc.StopAcceptingConnections(); - EXPECT_FALSE(webrtc.IsAcceptingConnections()); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); + webrtc.StopAcceptingConnections(service_id); + EXPECT_FALSE(webrtc.IsAcceptingConnections(std::string{})); } // Tests the flow when the device tries to connect to two different peers @@ -109,12 +118,13 @@ TEST_F(WebRtcTest, ConnectTwice) { WebRtc receiver, sender, device_c; WebRtcSocketWrapper receiver_socket, sender_socket; const PeerId self_id("self_id"), other_id("other_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; Future connected; ByteArray message("message xyz"); receiver.StartAcceptingConnections( - self_id, location_hint, + self_id, service_id, location_hint, {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { receiver_socket = wrapper; connected.Set(receiver_socket.IsValid()); @@ -123,7 +133,7 @@ TEST_F(WebRtcTest, ConnectTwice) { using MockAcceptedCallback = testing::MockFunction; testing::StrictMock mock_accepted_callback_; - device_c.StartAcceptingConnections(other_id, location_hint, + device_c.StartAcceptingConnections(other_id, service_id, location_hint, {mock_accepted_callback_.AsStdFunction()}); sender_socket = sender.Connect(self_id, location_hint); @@ -154,12 +164,13 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndAbort) { WebRtc receiver, sender; WebRtcSocketWrapper receiver_socket, sender_socket; const PeerId self_id("self_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; Future connected; ByteArray message("message xyz"); receiver.StartAcceptingConnections( - self_id, location_hint, + self_id, service_id, location_hint, {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { receiver_socket = wrapper; connected.Set(receiver_socket.IsValid()); @@ -181,12 +192,13 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndSendData) { WebRtc receiver, sender; WebRtcSocketWrapper receiver_socket, sender_socket; const PeerId self_id("self_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; Future connected; ByteArray message("message"); receiver.StartAcceptingConnections( - self_id, location_hint, + self_id, service_id, location_hint, {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { receiver_socket = wrapper; connected.Set(receiver_socket.IsValid()); @@ -214,12 +226,13 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { WebRtc receiver, sender; WebRtcSocketWrapper receiver_socket, sender_socket; const PeerId self_id("self_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; Future connected; ByteArray message("message xyz"); receiver.StartAcceptingConnections( - self_id, location_hint, + self_id, service_id, location_hint, {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { receiver_socket = wrapper; connected.Set(receiver_socket.IsValid()); @@ -233,7 +246,7 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { EXPECT_TRUE(devices_connected.result()); // Only shuts down signaling channel. - receiver.StopAcceptingConnections(); + receiver.StopAcceptingConnections(service_id); sender_socket.GetOutputStream().Write(message); ExceptionOr received_msg = @@ -252,11 +265,13 @@ TEST_F(WebRtcTest, StartAcceptingConnections_NullPeerConnection) { WebRtc webrtc; PeerId self_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); EXPECT_FALSE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); } TEST_F(WebRtcTest, Connect_NullPeerConnection) { diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.cc b/cpp/core/internal/p2p_cluster_pcp_handler.cc index 780940d3..17ba3bcd 100644 --- a/cpp/core/internal/p2p_cluster_pcp_handler.cc +++ b/cpp/core/internal/p2p_cluster_pcp_handler.cc @@ -78,17 +78,6 @@ BasePcpHandler::StartOperationResult P2pClusterPcpHandler::StartAdvertisingImpl( std::vector mediums_started_successfully; WebRtcState web_rtc_state{WebRtcState::kUnconnectable}; - if (options.allowed.web_rtc) { - proto::connections::Medium webrtc_medium = - StartListeningForWebRtcConnections( - client, service_id, local_endpoint_id, local_endpoint_info); - if (webrtc_medium != proto::connections::UNKNOWN_MEDIUM) { - NEARBY_LOG(INFO, - "P2pClusterPcpHandler::StartAdvertisingImpl: WebRtc added"); - mediums_started_successfully.push_back(webrtc_medium); - web_rtc_state = WebRtcState::kConnectable; - } - } if (options.allowed.wifi_lan) { const ByteArray wifi_lan_hash = @@ -149,9 +138,6 @@ Status P2pClusterPcpHandler::StopAdvertisingImpl(ClientProxy* client) { ble_medium_.StopAdvertising(client->GetAdvertisingServiceId()); ble_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId()); - webrtc_medium_.StopAcceptingConnection(client->GetAdvertisingServiceId(), - client->GetLocalEndpointId()); - wifi_lan_medium_.StopAdvertising(client->GetAdvertisingServiceId()); wifi_lan_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId()); @@ -193,8 +179,8 @@ bool P2pClusterPcpHandler::IsRecognizedBluetoothEndpoint( void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler( ClientProxy* client, const std::string& service_id, - BluetoothDevice& device) { - RunOnPcpHandlerThread([this, client, service_id, &device]() { + BluetoothDevice device) { + RunOnPcpHandlerThread([this, client, service_id, device]() { // Make sure we are still discovering before proceeding. if (!client->IsDiscovering()) { NEARBY_LOG(INFO, @@ -205,7 +191,7 @@ void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler( } // Parse the Bluetooth device name. - const std::string& device_name_string = device.GetName(); + const std::string device_name_string = device.GetName(); BluetoothDeviceName device_name(device_name_string); // Make sure the Bluetooth device name points to a valid @@ -690,10 +676,6 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::ConnectImpl( break; } case proto::connections::Medium::WEB_RTC: { - auto* webrtc_endpoint = down_cast(endpoint); - if (webrtc_endpoint) { - return WebRtcConnectImpl(client, webrtc_endpoint); - } break; } default: @@ -1155,13 +1137,12 @@ P2pClusterPcpHandler::StartListeningForWebRtcConnections( return proto::connections::UNKNOWN_MEDIUM; } - if (!webrtc_medium_.IsAcceptingConnections()) { + if (!webrtc_medium_.IsAcceptingConnections(service_id)) { mediums::PeerId self_id = CreatePeerIdFromAdvertisement( service_id, local_endpoint_id, local_endpoint_info); std::string empty_country_code; if (!webrtc_medium_.StartAcceptingConnections( - self_id, service_id, local_endpoint_id, - Utils::BuildLocationHint(empty_country_code), + self_id, service_id, Utils::BuildLocationHint(empty_country_code), {[this, client, local_endpoint_info](mediums::WebRtcSocketWrapper socket) { if (!socket.IsValid()) { diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.h b/cpp/core/internal/p2p_cluster_pcp_handler.h index 4922d6d0..58fe7d23 100644 --- a/cpp/core/internal/p2p_cluster_pcp_handler.h +++ b/cpp/core/internal/p2p_cluster_pcp_handler.h @@ -134,7 +134,7 @@ class P2pClusterPcpHandler : public BasePcpHandler { const BluetoothDeviceName& name) const; void BluetoothDeviceDiscoveredHandler(ClientProxy* client, const std::string& service_id, - BluetoothDevice& device); + BluetoothDevice device); void BluetoothDeviceLostHandler(ClientProxy* client, const std::string& service_id, BluetoothDevice& device); diff --git a/cpp/core/internal/webrtc_bwu_handler.cc b/cpp/core/internal/webrtc_bwu_handler.cc index d37f8921..4bd3e269 100644 --- a/cpp/core/internal/webrtc_bwu_handler.cc +++ b/cpp/core/internal/webrtc_bwu_handler.cc @@ -20,10 +20,10 @@ WebrtcBwuHandler::WebrtcBwuHandler(Mediums& mediums, mediums_(mediums) {} void WebrtcBwuHandler::Revert() { - if (!active_service_ids_.empty()) { - webrtc_.StopAcceptingConnections(); - active_service_ids_.clear(); + for (const auto& service_id : active_service_ids_) { + webrtc_.StopAcceptingConnections(service_id); } + active_service_ids_.clear(); NEARBY_LOG(INFO, "WebrtcBwuHandler successfully reverted state."); } @@ -61,10 +61,9 @@ ByteArray WebrtcBwuHandler::InitializeUpgradedMediumForEndpoint( Utils::BuildLocationHint(webrtc_.GetDefaultCountryCode()); mediums::PeerId self_id{mediums::PeerId::FromRandom()}; - if (!webrtc_.IsAcceptingConnections()) { + if (!webrtc_.IsAcceptingConnections(service_id)) { if (!webrtc_.StartAcceptingConnections( - self_id, upgrade_service_id, client->GetLocalEndpointId(), - location_hint, + self_id, upgrade_service_id, location_hint, { .accepted_cb = absl::bind_front( &WebrtcBwuHandler::OnIncomingWebrtcConnection, this, client,