diff --git a/connections/implementation/base_pcp_handler.cc b/connections/implementation/base_pcp_handler.cc index 7d0e4b1a..6aa13dc6 100644 --- a/connections/implementation/base_pcp_handler.cc +++ b/connections/implementation/base_pcp_handler.cc @@ -351,24 +351,29 @@ Status BasePcpHandler::StartDiscovery(ClientProxy* client, << GetStringValueOfSupportedMediums( stripped_discovery_options); RunOnPcpHandlerThread( - "start-discovery", [this, client, service_id, stripped_discovery_options, - &listener, &response]() RUN_ON_PCP_HANDLER_THREAD() { - // Ask the implementation to attempt to start discovery. - auto result = - StartDiscoveryImpl(client, service_id, stripped_discovery_options); - if (!result.status.Ok()) { - response.Set(result.status); - return; - } + "start-discovery", + [this, client, service_id, stripped_discovery_options, &listener, + &response]() RUN_ON_PCP_HANDLER_THREAD() + ABSL_LOCKS_EXCLUDED(discovered_endpoint_mutex_) { + // Ask the implementation to attempt to start discovery. + auto result = StartDiscoveryImpl(client, service_id, + stripped_discovery_options); + if (!result.status.Ok()) { + response.Set(result.status); + return; + } - // Now that we've succeeded, mark the client as discovering and clear - // out any old endpoints we had discovered. - discovered_endpoints_.clear(); - client->StartedDiscovery(service_id, GetStrategy(), listener, - absl::MakeSpan(result.mediums), - stripped_discovery_options); - response.Set({Status::kSuccess}); - }); + // Now that we've succeeded, mark the client as discovering and + // clear out any old endpoints we had discovered. + { + MutexLock lock(&discovered_endpoint_mutex_); + discovered_endpoints_.clear(); + } + client->StartedDiscovery(service_id, GetStrategy(), listener, + absl::MakeSpan(result.mediums), + stripped_discovery_options); + response.Set({Status::kSuccess}); + }); return WaitForResult(absl::StrCat("StartDiscovery(", service_id, ")"), client->GetClientId(), &response); } @@ -830,6 +835,7 @@ void BasePcpHandler::StripOutUnavailableMediums( // Get any single discovered endpoint for a given endpoint_id. BasePcpHandler::DiscoveredEndpoint* BasePcpHandler::GetDiscoveredEndpoint( const std::string& endpoint_id) { + MutexLock lock(&discovered_endpoint_mutex_); auto it = discovered_endpoints_.find(endpoint_id); if (it == discovered_endpoints_.end()) { return nullptr; @@ -840,6 +846,7 @@ BasePcpHandler::DiscoveredEndpoint* BasePcpHandler::GetDiscoveredEndpoint( std::vector BasePcpHandler::GetDiscoveredEndpoints(const std::string& endpoint_id) { std::vector result; + MutexLock lock(&discovered_endpoint_mutex_); auto it = discovered_endpoints_.equal_range(endpoint_id); for (auto item = it.first; item != it.second; item++) { result.push_back(item->second.get()); @@ -856,6 +863,7 @@ std::vector BasePcpHandler::GetDiscoveredEndpoints( const location::nearby::proto::connections::Medium medium) { std::vector result; + MutexLock lock(&discovered_endpoint_mutex_); for (const auto& item : discovered_endpoints_) { if (item.second->medium == medium) { result.push_back(item.second.get()); @@ -1181,7 +1189,7 @@ void BasePcpHandler::OnEndpointFound( // Check if we've seen this endpoint ID before. std::string& endpoint_id = endpoint->endpoint_id; NEARBY_LOGS(INFO) << "OnEndpointFound: id=" << endpoint_id << " [enter]"; - + MutexLock lock(&discovered_endpoint_mutex_); auto range = discovered_endpoints_.equal_range(endpoint->endpoint_id); bool is_range_empty = range.first == range.second; DiscoveredEndpoint* owned_endpoint = nullptr; @@ -1234,7 +1242,7 @@ void BasePcpHandler::OnEndpointLost( ClientProxy* client, const BasePcpHandler::DiscoveredEndpoint& endpoint) { // Look up the DiscoveredEndpoint we have in our cache. NEARBY_LOGS(INFO) << "OnEndpointLost: id=" << endpoint.endpoint_id; - + MutexLock lock(&discovered_endpoint_mutex_); auto range = discovered_endpoints_.equal_range(endpoint.endpoint_id); bool is_range_empty = range.first == range.second; if (is_range_empty) { @@ -1601,6 +1609,7 @@ bool BasePcpHandler::AppendRemoteBluetoothMacAddressEndpoint( if (!local_discovery_options.allowed.bluetooth) { return false; } + MutexLock lock(&discovered_endpoint_mutex_); auto it = discovered_endpoints_.equal_range(endpoint_id); if (it.first == it.second) { @@ -1646,6 +1655,7 @@ bool BasePcpHandler::AppendWebRTCEndpoint( if (!local_discovery_options.allowed.web_rtc) { return false; } + MutexLock lock(&discovered_endpoint_mutex_); bool should_connect_web_rtc = false; auto it = discovered_endpoints_.equal_range(endpoint_id); diff --git a/connections/implementation/base_pcp_handler.h b/connections/implementation/base_pcp_handler.h index 9d9d77aa..ada2c1c8 100644 --- a/connections/implementation/base_pcp_handler.h +++ b/connections/implementation/base_pcp_handler.h @@ -258,10 +258,12 @@ class BasePcpHandler : public PcpHandler, void OnEndpointFound(ClientProxy* client, std::shared_ptr endpoint) - RUN_ON_PCP_HANDLER_THREAD(); + RUN_ON_PCP_HANDLER_THREAD() + ABSL_LOCKS_EXCLUDED(discovered_endpoint_mutex_); void OnEndpointLost(ClientProxy* client, const DiscoveredEndpoint& endpoint) - RUN_ON_PCP_HANDLER_THREAD(); + RUN_ON_PCP_HANDLER_THREAD() + ABSL_LOCKS_EXCLUDED(discovered_endpoint_mutex_); Exception OnIncomingConnection( ClientProxy* client, const ByteArray& remote_endpoint_info, @@ -339,16 +341,19 @@ class BasePcpHandler : public PcpHandler, GetDefaultUpgradeMedium() = 0; // Returns the first discovered endpoint for the given endpoint_id. - DiscoveredEndpoint* GetDiscoveredEndpoint(const std::string& endpoint_id); + DiscoveredEndpoint* GetDiscoveredEndpoint(const std::string& endpoint_id) + ABSL_LOCKS_EXCLUDED(discovered_endpoint_mutex_); // Returns a vector of discovered endpoints, sorted in order of decreasing // preference. std::vector GetDiscoveredEndpoints( - const std::string& endpoint_id); + const std::string& endpoint_id) + ABSL_LOCKS_EXCLUDED(discovered_endpoint_mutex_); // Returns a vector of discovered endpoints that share a given Medium. std::vector GetDiscoveredEndpoints( - const location::nearby::proto::connections::Medium medium); + const location::nearby::proto::connections::Medium medium) + ABSL_LOCKS_EXCLUDED(discovered_endpoint_mutex_); // Start alarms for endpoints lost by their mediums. Used when updating // discovery options. @@ -484,12 +489,14 @@ class BasePcpHandler : public PcpHandler, bool AppendRemoteBluetoothMacAddressEndpoint( const std::string& endpoint_id, const std::string& remote_bluetooth_mac_address, - const DiscoveryOptions& local_discovery_options); + const DiscoveryOptions& local_discovery_options) + ABSL_LOCKS_EXCLUDED(discovered_endpoint_mutex_); // Returns true if the webrtc endpoint is created and appended into // discovered_endpoints_ with key endpoint_id. bool AppendWebRTCEndpoint(const std::string& endpoint_id, - const DiscoveryOptions& local_discovery_options); + const DiscoveryOptions& local_discovery_options) + ABSL_LOCKS_EXCLUDED(discovered_endpoint_mutex_); void ProcessPreConnectionInitiationFailure( ClientProxy* client, Medium medium, const std::string& endpoint_id, @@ -567,6 +574,7 @@ class BasePcpHandler : public PcpHandler, ScheduledExecutor alarm_executor_; SingleThreadExecutor serial_executor_; + Mutex discovered_endpoint_mutex_; // A map of endpoint id -> PendingConnectionInfo. Entries in this map imply // that there is an active connection to the endpoint and we're waiting for @@ -576,7 +584,7 @@ class BasePcpHandler : public PcpHandler, absl::flat_hash_map pending_connections_; // A map of endpoint id -> DiscoveredEndpoint. absl::btree_multimap> - discovered_endpoints_; + discovered_endpoints_ ABSL_GUARDED_BY(discovered_endpoint_mutex_); // A map of endpoint id -> alarm. These alarms delay closing the // EndpointChannel to give the other side enough time to read the rejection // message. It's expected that the other side will close the connection diff --git a/connections/implementation/base_pcp_handler_test.cc b/connections/implementation/base_pcp_handler_test.cc index 8c9168e4..38d365e0 100644 --- a/connections/implementation/base_pcp_handler_test.cc +++ b/connections/implementation/base_pcp_handler_test.cc @@ -394,6 +394,11 @@ class BasePcpHandlerTest discovery_listener_), Status{Status::kSuccess}); EXPECT_TRUE(client->IsDiscovering()); + for (const auto& discovered_medium : + pcp_handler->GetDiscoveryMediums(client)) { + EXPECT_TRUE( + pcp_handler->GetDiscoveredEndpoints(discovered_medium).empty()); + } } void UpdateDiscoveryOptions(ClientProxy* client, MockPcpHandler* pcp_handler, @@ -680,6 +685,34 @@ TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) { env_.Stop(); } +TEST_P(BasePcpHandlerTest, StartDiscoveryFails) { + env_.Start(); + ClientProxy client; + Mediums m; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + BwuManager bwu(m, em, ecm, {}, {}); + MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); + DiscoveryOptions discovery_options{ + {}, + true, // auto_upgrade_bandwidth + true, // enforce_topology_constraints + false, // is_out_of_band_connection, + "", // fast_advertisement_service_uuid + true, // low_power + }; + EXPECT_CALL(pcp_handler, StartDiscoveryImpl) + .WillOnce(Return(MockPcpHandler::StartOperationResult{ + .status = {Status::kError}, + .mediums = {}, + })); + EXPECT_EQ(pcp_handler.StartDiscovery(&client, "service", discovery_options, + discovery_listener_), + Status{Status::kError}); + bwu.Shutdown(); + env_.Stop(); +} + TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) { env_.Start(); ClientProxy client; @@ -697,6 +730,37 @@ TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) { env_.Stop(); } +TEST_P(BasePcpHandlerTest, StartStopStartDiscoveryClearsEndpoints) { + env_.Start(); + ClientProxy client; + Mediums m; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + BwuManager bwu(m, em, ecm, {}, {}); + MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); + StartDiscovery(&client, &pcp_handler); + auto mediums = pcp_handler.GetDiscoveryMediums(&client); + auto connect_medium = mediums[mediums.size() - 1]; + EXPECT_CALL(mock_discovery_listener_.endpoint_found_cb, Call); + pcp_handler.OnEndpointFound( + &client, std::make_shared(MockDiscoveredEndpoint{ + { + "DEFG", + ByteArray("1"), + "service", + connect_medium, + WebRtcState::kUndefined, + }, + MockContext{nullptr}, + })); + EXPECT_CALL(pcp_handler, StopDiscoveryImpl(&client)).Times(1); + pcp_handler.StopDiscovery(&client); + EXPECT_FALSE(client.IsDiscovering()); + StartDiscovery(&client, &pcp_handler); + bwu.Shutdown(); + env_.Stop(); +} + TEST_F(BasePcpHandlerTest, WifiMediumFailFallBackToBT) { env_.Start(); std::string service_id{"service"};