TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-16 15:36:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

fix tsan error in NC when accessing discovered_endpoints

Browse Source 
PiperOrigin-RevId: 546060069
This commit is contained in:
Anay Wadhera
2023-07-06 12:22:55 -07:00
committed by Copybara-Service
parent 5bce94d8ee
commit 5369c33112
3 changed files with 109 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/base_pcp_handler.cc
+29 -19
View File
@@ -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::DiscoveredEndpoint*>
BasePcpHandler::GetDiscoveredEndpoints(const std::string& endpoint_id) {
std::vector<BasePcpHandler::DiscoveredEndpoint*> 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::DiscoveredEndpoint*>
BasePcpHandler::GetDiscoveredEndpoints(
const location::nearby::proto::connections::Medium medium) {
std::vector<BasePcpHandler::DiscoveredEndpoint*> 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);
connections/implementation/base_pcp_handler.h
+16 -8
View File
@@ -258,10 +258,12 @@ class BasePcpHandler : public PcpHandler,
void OnEndpointFound(ClientProxy* client,
std::shared_ptr<DiscoveredEndpoint> 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<BasePcpHandler::DiscoveredEndpoint*> 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<BasePcpHandler::DiscoveredEndpoint*> 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<std::string, PendingConnectionInfo> pending_connections_;
// A map of endpoint id -> DiscoveredEndpoint.
absl::btree_multimap<std::string, std::shared_ptr<DiscoveredEndpoint>>
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
connections/implementation/base_pcp_handler_test.cc
+64
View File
@@ -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>(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"};