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

Implement Safe to Disconnect feature

Browse Source 
PiperOrigin-RevId: 560119556
This commit is contained in:
hai007
2023-08-25 09:51:44 -07:00
committed by Copybara-Service
parent 965cc0ecbc
commit 55aedfb235
27 changed files with 1052 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/BUILD
+4
View File
@@ -131,6 +131,7 @@ cc_library(
"//internal/platform/implementation:comm",
"//internal/platform/implementation:platform",
"//internal/platform/implementation/shared:file",
"//internal/proto/analytics:connections_log_cc_proto",
"//proto:connections_enums_cc_proto",
"@com_google_absl//absl/base:core_headers",
"@com_google_absl//absl/container:btree",
@@ -182,7 +183,9 @@ cc_library(
deps = [
":internal",
"//connections:core_types",
"//connections/implementation/flags:connections_flags",
"//connections/v3:v3_types",
"//internal/flags:nearby_flags",
"//internal/platform:base",
"//internal/platform:test_util",
"//internal/platform:types",
@@ -240,6 +243,7 @@ cc_test(
"//internal/platform:test_util",
"//internal/platform:types",
"//internal/platform/implementation/g3", # build_cleaner: keep
"//internal/proto/analytics:connections_log_cc_proto",
"//internal/test",
"//proto:connections_enums_cc_proto",
"@com_github_protobuf_matchers//protobuf-matchers",
connections/implementation/base_endpoint_channel.h
+2 -1
View File
@@ -21,9 +21,10 @@
#include "absl/base/thread_annotations.h"
#include "connections/implementation/analytics/analytics_recorder.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/implementation/endpoint_channel.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/condition_variable.h"
#include "internal/platform/exception.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/mutex.h"
#include "internal/platform/output_stream.h"
connections/implementation/base_pcp_handler.cc
+43 -13
View File
@@ -28,6 +28,8 @@
#include "absl/types/span.h"
#include "connections/advertising_options.h"
#include "connections/connection_options.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/mediums/utils.h"
#include "connections/implementation/offline_frames.h"
@@ -1012,9 +1014,12 @@ void BasePcpHandler::ProcessPreConnectionInitiationFailure(
}
void BasePcpHandler::ProcessPreConnectionResultFailure(
ClientProxy* client, const std::string& endpoint_id) {
ClientProxy* client, const std::string& endpoint_id,
bool should_call_disconnect_endpoint, const DisconnectionReason& reason) {
auto item = pending_connections_.extract(endpoint_id);
endpoint_manager_->DiscardEndpoint(client, endpoint_id);
if (should_call_disconnect_endpoint) {
endpoint_manager_->DiscardEndpoint(client, endpoint_id, reason);
}
client->OnConnectionRejected(endpoint_id, {Status::kError});
}
@@ -1048,7 +1053,9 @@ Status BasePcpHandler::AcceptConnection(ClientProxy* client,
NEARBY_LOGS(ERROR) << "Channel destroyed before Accept; bring down "
"connection: endpoint_id="
<< endpoint_id;
ProcessPreConnectionResultFailure(client, endpoint_id);
ProcessPreConnectionResultFailure(
client, endpoint_id, /* should_call_disconnect_endpoint= */ true,
DisconnectionReason::IO_ERROR);
response.Set({Status::kEndpointUnknown});
return;
}
@@ -1060,7 +1067,9 @@ Status BasePcpHandler::AcceptConnection(ClientProxy* client,
NEARBY_LOGS(INFO)
<< "AcceptConnection: failed to send response: endpoint_id="
<< endpoint_id;
ProcessPreConnectionResultFailure(client, endpoint_id);
ProcessPreConnectionResultFailure(
client, endpoint_id, /* should_call_disconnect_endpoint= */ true,
DisconnectionReason::IO_ERROR);
response.Set({Status::kEndpointIoError});
return;
}
@@ -1106,7 +1115,9 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client,
<< "Channel destroyed before Reject; bring down connection: "
"endpoint_id="
<< endpoint_id;
ProcessPreConnectionResultFailure(client, endpoint_id);
ProcessPreConnectionResultFailure(
client, endpoint_id, /* should_call_disconnect_endpoint= */ true,
DisconnectionReason::IO_ERROR);
response.Set({Status::kEndpointUnknown});
return;
}
@@ -1118,7 +1129,9 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client,
NEARBY_LOGS(INFO)
<< "RejectConnection: failed to send response: endpoint_id="
<< endpoint_id;
ProcessPreConnectionResultFailure(client, endpoint_id);
ProcessPreConnectionResultFailure(
client, endpoint_id, /* should_call_disconnect_endpoint= */ true,
DisconnectionReason::IO_ERROR);
response.Set({Status::kEndpointIoError});
return;
}
@@ -1182,6 +1195,16 @@ void BasePcpHandler::OnIncomingFrame(
client->SetRemoteOsInfo(endpoint_id, connection_response.os_info());
}
if (connection_response.has_safe_to_disconnect_version()) {
NEARBY_LOGS(INFO)
<< "[safe-to-disconnect]: endpoint_id=" << endpoint_id
<< "; Version = "
<< connection_response.safe_to_disconnect_version();
client->SetRemoteSafeToDisconnectVersion(
endpoint_id, connection_response.safe_to_disconnect_version());
}
channel_manager_->UpdateSafeToDisconnectForEndpoint(endpoint_id,
client->IsSafeToDisconnectEnabled(endpoint_id));
EvaluateConnectionResult(client, endpoint_id,
/* can_close_immediately= */ true);
@@ -1193,13 +1216,14 @@ void BasePcpHandler::OnIncomingFrame(
void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client,
const std::string& service_id,
const std::string& endpoint_id,
CountDownLatch barrier) {
CountDownLatch barrier,
DisconnectionReason reason) {
if (stop_.Get()) {
barrier.CountDown();
return;
}
RunOnPcpHandlerThread("on-endpoint-disconnect",
[this, client, endpoint_id, barrier]()
[this, client, endpoint_id, barrier, reason]()
RUN_ON_PCP_HANDLER_THREAD() mutable {
auto item = pending_alarms_.find(endpoint_id);
if (item != pending_alarms_.end()) {
@@ -1207,8 +1231,10 @@ void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client,
alarm->Cancel();
pending_alarms_.erase(item);
}
ProcessPreConnectionResultFailure(client,
endpoint_id);
ProcessPreConnectionResultFailure(
client, endpoint_id,
/* should_call_disconnect_endpoint= */ false,
reason);
barrier.CountDown();
});
}
@@ -1651,7 +1677,9 @@ void BasePcpHandler::ProcessTieBreakLoss(
client, info->channel->GetMedium(), endpoint_id, info->channel.get(),
info->is_incoming, info->start_time, {Status::kEndpointIoError},
info->result.lock().get());
ProcessPreConnectionResultFailure(client, endpoint_id);
ProcessPreConnectionResultFailure(client, endpoint_id,
/* should_call_disconnect_endpoint= */ true,
DisconnectionReason::IO_ERROR);
}
bool BasePcpHandler::AppendRemoteBluetoothMacAddressEndpoint(
@@ -1798,14 +1826,16 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client,
// Clean up the channel in EndpointManager if it's no longer required.
if (can_close_immediately) {
endpoint_manager_->DiscardEndpoint(client, endpoint_id);
endpoint_manager_->DiscardEndpoint(client, endpoint_id,
DisconnectionReason::UNFINISHED);
} else {
pending_alarms_.emplace(
endpoint_id,
std::make_unique<CancelableAlarm>(
"BasePcpHandler.evaluateConnectionResult() delayed close",
[this, client, endpoint_id]() {
endpoint_manager_->DiscardEndpoint(client, endpoint_id);
endpoint_manager_->DiscardEndpoint(
client, endpoint_id, DisconnectionReason::UNFINISHED);
},
kRejectedConnectionCloseDelay, &alarm_executor_));
}
connections/implementation/base_pcp_handler.h
+5 -2
View File
@@ -149,7 +149,8 @@ class BasePcpHandler : public PcpHandler,
// @EndpointManagerThread
void OnEndpointDisconnect(ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id,
CountDownLatch barrier) override;
CountDownLatch barrier,
DisconnectionReason reason) override;
Status UpdateAdvertisingOptions(
ClientProxy* client, absl::string_view service_id,
@@ -507,7 +508,9 @@ class BasePcpHandler : public PcpHandler,
EndpointChannel* channel, bool is_incoming, absl::Time start_time,
Status status, Future<Status>* result);
void ProcessPreConnectionResultFailure(ClientProxy* client,
const std::string& endpoint_id);
const std::string& endpoint_id,
bool should_call_disconnect_endpoint,
const DisconnectionReason& reason);
// Called when either side accepts/rejects the connection, but only takes
// effect after both have accepted or one side has rejected.
connections/implementation/base_pcp_handler_test.cc
+13 -3
View File
@@ -37,6 +37,7 @@
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/encryption_runner.h"
#include "connections/implementation/endpoint_manager.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/mediums/mediums.h"
#include "connections/implementation/offline_frames.h"
#include "connections/implementation/pcp.h"
@@ -48,19 +49,17 @@
#include "connections/status.h"
#include "connections/strategy.h"
#include "connections/v3/connection_listening_options.h"
#include "internal/flags/nearby_flags.h"
#include "internal/interop/device.h"
#include "internal/interop/device_provider.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/future.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/logging.h"
#include "internal/platform/medium_environment.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/pipe.h"
#include "proto/connections_enums.pb.h"
#include "proto/connections_enums.proto.h"
namespace nearby {
namespace connections {
@@ -358,6 +357,16 @@ struct MockDiscoveredEndpoint : public MockPcpHandler::DiscoveredEndpoint {
MockContext context;
};
class SetSafeToDisconnect {
public:
explicit SetSafeToDisconnect(bool safe_to_disconnect) {
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::
kEnableSafeToDisconnect,
safe_to_disconnect);
}
};
class BasePcpHandlerTest
: public ::testing::TestWithParam<BooleanMediumSelector> {
protected:
@@ -693,6 +702,7 @@ class BasePcpHandlerTest
.endpoint_distance_changed_cb =
mock_discovery_listener_.endpoint_distance_changed_cb.AsStdFunction(),
};
SetSafeToDisconnect set_safe_to_disconnect_{true};
MediumEnvironment& env_ = MediumEnvironment::Instance();
};
connections/implementation/bwu_manager.cc
+9 -2
View File
@@ -24,6 +24,8 @@
#include "absl/time/time.h"
#include "connections/implementation/bluetooth_bwu_handler.h"
#include "connections/implementation/bwu_handler.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/offline_frames.h"
#include "connections/implementation/service_id_constants.h"
#ifdef NO_WEBRTC
@@ -343,7 +345,8 @@ void BwuManager::OnIncomingFrame(OfflineFrame& frame,
void BwuManager::OnEndpointDisconnect(ClientProxy* client,
const std::string& service_id,
const std::string& endpoint_id,
CountDownLatch barrier) {
CountDownLatch barrier,
DisconnectionReason reason) {
NEARBY_LOGS(INFO)
<< "BwuManager has processed endpoint disconnection for endpoint "
<< endpoint_id;
@@ -1112,7 +1115,11 @@ void BwuManager::ProcessSafeToClosePriorChannelEvent(
// circumstances so it is necessary to send it unencrypted. This way the
// serial crypto context does not increment here.
previous_endpoint_channel->DisableEncryption();
previous_endpoint_channel->Write(parser::ForDisconnection());
NEARBY_LOGS(INFO) << "[safe-to-disconnect] Sending "
"DISCONNECTION frame with request 0, ack 0";
previous_endpoint_channel->Write(
parser::ForDisconnection(/* request_safe_to_disconnect */ false,
/* ack_safe_to_disconnect */ false));
// Attempt to read the disconnect message from the previous channel. We don't
// care whether we successfully read it or whether we get an exception here.
connections/implementation/bwu_manager.h
+2 -1
View File
@@ -94,7 +94,8 @@ class BwuManager : public EndpointManager::FrameProcessor {
void OnEndpointDisconnect(ClientProxy* client_proxy,
const std::string& service_id,
const std::string& endpoint_id,
CountDownLatch barrier) override;
CountDownLatch barrier,
DisconnectionReason reason) override;
void Shutdown();
connections/implementation/bwu_manager_test.cc
+119 -42
View File
@@ -30,15 +30,19 @@
#include "connections/implementation/offline_frames.h"
#include "connections/implementation/service_id_constants.h"
#include "internal/platform/exception.h"
#include "internal/proto/analytics/connections_log.pb.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
namespace {
using ::location::nearby::analytics::proto::ConnectionsLog;
using ::location::nearby::connections::BandwidthUpgradeNegotiationFrame;
using ::location::nearby::connections::
BandwidthUpgradeNegotiationFrame_UpgradePathInfo;
using ::location::nearby::connections::OfflineFrame;
using ::location::nearby::connections::V1Frame;
using ::location::nearby::proto::connections::DisconnectionReason;
constexpr absl::string_view kServiceIdA = "ServiceA";
constexpr absl::string_view kServiceIdB = "ServiceB";
@@ -95,6 +99,11 @@ class BwuManagerTest : public ::testing::Test {
std::move(channel));
return channel_raw;
}
void UnRegisterChannelForEndpoint(absl::string_view endpoint_id) {
ecm_.UnregisterChannelForEndpoint(
std::string(endpoint_id), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
}
// Upgrade from |initial_medium| to |upgrade_medium|, close down the BLUETOOTH
// channel, return the upgraded endpoint channel. This logic is tested in
@@ -175,6 +184,9 @@ TEST(BwuManagerBaseTest, AllowToUpgradeMedium) {
bwu_manager->InitiateBwuForEndpoint(&client, std::string(kEndpointId1),
Medium::WIFI_LAN);
EXPECT_TRUE(bwu_manager->IsUpgradeOngoing(std::string(kEndpointId1)));
ecm.UnregisterChannelForEndpoint(
std::string(kEndpointId1), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
auto channel2 = std::make_unique<FakeEndpointChannel>(
Medium::BLUETOOTH, std::string(kServiceIdA));
@@ -183,6 +195,9 @@ TEST(BwuManagerBaseTest, AllowToUpgradeMedium) {
bwu_manager->InitiateBwuForEndpoint(&client, std::string(kEndpointId2),
Medium::WIFI_HOTSPOT);
EXPECT_TRUE(bwu_manager->IsUpgradeOngoing(std::string(kEndpointId2)));
ecm.UnregisterChannelForEndpoint(
std::string(kEndpointId2), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
auto channel3 = std::make_unique<FakeEndpointChannel>(
Medium::BLUETOOTH, std::string(kServiceIdA));
@@ -191,6 +206,9 @@ TEST(BwuManagerBaseTest, AllowToUpgradeMedium) {
bwu_manager->InitiateBwuForEndpoint(&client, std::string(kEndpointId3),
Medium::WIFI_DIRECT);
EXPECT_TRUE(bwu_manager->IsUpgradeOngoing(std::string(kEndpointId3)));
ecm.UnregisterChannelForEndpoint(
std::string(kEndpointId3), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
auto channel4 = std::make_unique<FakeEndpointChannel>(
Medium::WEB_RTC, std::string(kServiceIdA));
@@ -199,6 +217,9 @@ TEST(BwuManagerBaseTest, AllowToUpgradeMedium) {
bwu_manager->InitiateBwuForEndpoint(&client, std::string(kEndpointId4),
Medium::BLUETOOTH);
EXPECT_FALSE(bwu_manager->IsUpgradeOngoing(std::string(kEndpointId4)));
ecm.UnregisterChannelForEndpoint(
std::string(kEndpointId4), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
bwu_manager->Shutdown();
}
@@ -272,6 +293,7 @@ TEST_P(BwuManagerTestParam, InitiateBwu_Success) {
EXPECT_TRUE(old_channel->is_closed());
EXPECT_EQ(location::nearby::proto::connections::DisconnectionReason::UPGRADED,
old_channel->disconnection_reason());
UnRegisterChannelForEndpoint(kEndpointId1);
}
TEST_P(BwuManagerTestParam,
@@ -285,6 +307,7 @@ TEST_P(BwuManagerTestParam,
bwu_manager_->InitiateBwuForEndpoint(&client_, std::string(kEndpointId1),
Medium::WEB_RTC);
EXPECT_EQ(1u, fake_web_rtc_bwu_handler_->handle_initialize_calls().size());
UnRegisterChannelForEndpoint(kEndpointId1);
}
TEST_P(BwuManagerTestParam,
@@ -296,6 +319,7 @@ TEST_P(BwuManagerTestParam,
Medium::WIFI_HOTSPOT);
EXPECT_TRUE(
fake_wifi_hotspot_bwu_handler_->handle_initialize_calls().empty());
UnRegisterChannelForEndpoint(kEndpointId1);
}
TEST_P(BwuManagerTestParam, InitiateBwu_Error_NoInitialMedium) {
@@ -327,6 +351,7 @@ TEST_P(BwuManagerTestParam, InitiateBwu_Error_UpgradeAlreadyInProgress) {
EXPECT_TRUE(
fake_wifi_hotspot_bwu_handler_->handle_initialize_calls().empty());
EXPECT_TRUE(fake_wifi_direct_bwu_handler_->handle_initialize_calls().empty());
UnRegisterChannelForEndpoint(kEndpointId1);
}
TEST_P(BwuManagerTestParam,
@@ -358,6 +383,7 @@ TEST_P(BwuManagerTestParam,
ecm_.GetChannelForEndpoint(std::string(kEndpointId1)).get());
EXPECT_EQ(initial_channel,
ecm_.GetChannelForEndpoint(std::string(kEndpointId1)).get());
UnRegisterChannelForEndpoint(kEndpointId1);
}
TEST_F(BwuManagerTest,
@@ -380,9 +406,12 @@ TEST_F(BwuManagerTest,
// Disconnect the first WebRTC endpoint. We don't expect a revert until the
// last WebRTC endpoint for the service is disconnected.
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId1));
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id,
std::string(kEndpointId1), latch);
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId1), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id, std::string(kEndpointId1), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
ASSERT_EQ(1u, fake_web_rtc_bwu_handler_->disconnect_calls().size());
EXPECT_EQ(kEndpointId1,
fake_web_rtc_bwu_handler_->disconnect_calls()[0].endpoint_id);
@@ -391,9 +420,12 @@ TEST_F(BwuManagerTest,
{
// Disconnect the second WebRTC endpoint. We expect a revert.
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId2));
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id,
std::string(kEndpointId2), latch);
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId2), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id, std::string(kEndpointId2), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
ASSERT_EQ(2u, fake_web_rtc_bwu_handler_->disconnect_calls().size());
EXPECT_EQ(kEndpointId2,
fake_web_rtc_bwu_handler_->disconnect_calls()[1].endpoint_id);
@@ -423,9 +455,12 @@ TEST_F(BwuManagerTest,
{
// Disconnect the first WebRTC endpoint.
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId1));
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id,
std::string(kEndpointId1), latch);
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId1), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id, std::string(kEndpointId1), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
ASSERT_EQ(1u, fake_web_rtc_bwu_handler_->disconnect_calls().size());
EXPECT_EQ(kEndpointId1,
fake_web_rtc_bwu_handler_->disconnect_calls()[0].endpoint_id);
@@ -440,9 +475,12 @@ TEST_F(BwuManagerTest,
{
// Disconnect the second WebRTC endpoint.
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId2));
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id,
std::string(kEndpointId2), latch);
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId2), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id, std::string(kEndpointId2), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
// Note(nohle): There appears to be an off-by-one error in the
// existing/flag-disabled code. Revert is called when there are "<= 1"
@@ -474,10 +512,13 @@ TEST_F(BwuManagerTest,
{
CountDownLatch latch(1);
EXPECT_EQ(2u, ecm_.GetConnectedEndpointsCount());
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId1));
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId1), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
EXPECT_EQ(1u, ecm_.GetConnectedEndpointsCount());
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id_A,
std::string(kEndpointId1), latch);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id_A, std::string(kEndpointId1), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
ASSERT_EQ(1u, fake_wifi_lan_bwu_handler_->disconnect_calls().size());
EXPECT_EQ(kEndpointId1,
fake_wifi_lan_bwu_handler_->disconnect_calls()[0].endpoint_id);
@@ -491,10 +532,13 @@ TEST_F(BwuManagerTest,
}
{
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId2));
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId2), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
EXPECT_EQ(0u, ecm_.GetConnectedEndpointsCount());
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id_B,
std::string(kEndpointId2), latch);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id_B, std::string(kEndpointId2), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
ASSERT_EQ(2u, fake_wifi_lan_bwu_handler_->disconnect_calls().size());
EXPECT_EQ(kEndpointId2,
fake_wifi_lan_bwu_handler_->disconnect_calls()[1].endpoint_id);
@@ -523,10 +567,13 @@ TEST_F(BwuManagerTest,
{
CountDownLatch latch(1);
EXPECT_EQ(2u, ecm_.GetConnectedEndpointsCount());
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId1));
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId1), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
EXPECT_EQ(1u, ecm_.GetConnectedEndpointsCount());
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id_A,
std::string(kEndpointId1), latch);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id_A, std::string(kEndpointId1), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
ASSERT_EQ(1u, fake_wifi_lan_bwu_handler_->disconnect_calls().size());
EXPECT_EQ(kEndpointId1,
fake_wifi_lan_bwu_handler_->disconnect_calls()[0].endpoint_id);
@@ -540,10 +587,13 @@ TEST_F(BwuManagerTest,
}
{
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId2));
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId2), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
EXPECT_EQ(0u, ecm_.GetConnectedEndpointsCount());
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id_B,
std::string(kEndpointId2), latch);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id_B, std::string(kEndpointId2), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
// Note(nohle): There appears to be an off-by-one error in the
// existing/flag-disabled code. Revert is called when there are "<= 1"
// (instead of "== 0") connected endpoints.
@@ -593,9 +643,12 @@ TEST_F(
EXPECT_TRUE(fake_wifi_direct_bwu_handler_->handle_revert_calls().empty());
{
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId1));
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id_A,
std::string(kEndpointId1), latch);
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId1), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id_A, std::string(kEndpointId1), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
// No more WebRTC channels for service A; expect revert call.
ASSERT_EQ(1u, fake_web_rtc_bwu_handler_->disconnect_calls().size());
@@ -615,9 +668,12 @@ TEST_F(
}
{
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId2));
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id_A,
std::string(kEndpointId2), latch);
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId2), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id_A, std::string(kEndpointId2), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
// We reverted a WLAN channel; no additional WebRTC calls expected.
EXPECT_EQ(1u, fake_web_rtc_bwu_handler_->disconnect_calls().size());
@@ -633,9 +689,12 @@ TEST_F(
}
{
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId3));
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id_B,
std::string(kEndpointId3), latch);
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId3), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id_B, std::string(kEndpointId3), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
// We reverted a WLAN channel; no additional WebRTC calls expected.
EXPECT_EQ(1u, fake_web_rtc_bwu_handler_->disconnect_calls().size());
@@ -651,9 +710,12 @@ TEST_F(
}
{
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId4));
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id_B,
std::string(kEndpointId4), latch);
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId4), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id_B, std::string(kEndpointId4), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
// We reverted a Hotspot channel; no additional WebRTC calls expected.
EXPECT_EQ(1u, fake_web_rtc_bwu_handler_->disconnect_calls().size());
@@ -671,9 +733,12 @@ TEST_F(
}
{
CountDownLatch latch(1);
ecm_.UnregisterChannelForEndpoint(std::string(kEndpointId5));
bwu_manager_->OnEndpointDisconnect(&client_, upgrade_service_id_B,
std::string(kEndpointId5), latch);
ecm_.UnregisterChannelForEndpoint(
std::string(kEndpointId5), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION);
bwu_manager_->OnEndpointDisconnect(
&client_, upgrade_service_id_B, std::string(kEndpointId5), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
// We reverted a WifiDirect channel; no additional WebRTC calls expected.
EXPECT_EQ(1u, fake_web_rtc_bwu_handler_->disconnect_calls().size());
@@ -722,6 +787,9 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnUpgradeFailure_FlagEnabled) {
ASSERT_EQ(1u, fake_web_rtc_bwu_handler_->handle_revert_calls().size());
EXPECT_EQ(WrapInitiatorUpgradeServiceId(kServiceIdB),
fake_web_rtc_bwu_handler_->handle_revert_calls()[0].service_id);
UnRegisterChannelForEndpoint(kEndpointId1);
UnRegisterChannelForEndpoint(kEndpointId2);
UnRegisterChannelForEndpoint(kEndpointId3);
}
TEST_F(BwuManagerTest, InitiateBwu_Revert_OnUpgradeFailure_FlagDisabled) {
@@ -756,6 +824,9 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnUpgradeFailure_FlagDisabled) {
// endpoints for _any_ service. We don't have service-level bookkeeping; we
// only know that there is some active WebRTC endpoint.
EXPECT_TRUE(fake_web_rtc_bwu_handler_->handle_revert_calls().empty());
UnRegisterChannelForEndpoint(kEndpointId1);
UnRegisterChannelForEndpoint(kEndpointId2);
UnRegisterChannelForEndpoint(kEndpointId3);
}
TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_WifiDirect) {
@@ -779,7 +850,8 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_WifiDirect) {
Medium::BLUETOOTH, packet_meta_data_);
CountDownLatch latch(1);
bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA,
std::string(kEndpointId1), latch);
std::string(kEndpointId1), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
ASSERT_EQ(fake_wifi_direct_bwu_handler_->disconnect_calls().size(), 1u);
EXPECT_EQ(kEndpointId1,
@@ -787,6 +859,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_WifiDirect) {
// This is called by the RESPONDER--call RevertInitiatorState only when
// BWU Medium is Hotspot or WifiDirect.
ASSERT_EQ(fake_wifi_direct_bwu_handler_->handle_revert_calls().size(), 1u);
UnRegisterChannelForEndpoint(kEndpointId1);
}
TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Hotspot) {
@@ -811,9 +884,11 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Hotspot) {
Medium::BLUETOOTH, packet_meta_data_);
CountDownLatch latch(1);
bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA,
std::string(kEndpointId1), latch);
std::string(kEndpointId1), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
ASSERT_EQ(fake_wifi_hotspot_bwu_handler_->handle_revert_calls().size(), 1u);
UnRegisterChannelForEndpoint(kEndpointId1);
}
TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Wlan) {
@@ -837,9 +912,11 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Wlan) {
Medium::BLUETOOTH, packet_meta_data_);
CountDownLatch latch(1);
bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA,
std::string(kEndpointId1), latch);
std::string(kEndpointId1), latch,
DisconnectionReason::LOCAL_DISCONNECTION);
ASSERT_EQ(fake_wifi_lan_bwu_handler_->handle_revert_calls().size(), 0u);
UnRegisterChannelForEndpoint(kEndpointId1);
}
TEST_F(BwuManagerTest, OnReceiveBwuEvent) {
connections/implementation/client_proxy.cc
+48 -1
View File
@@ -14,6 +14,7 @@
#include "connections/implementation/client_proxy.h"
#include <cstdint>
#include <cstdlib>
#include <functional>
#include <ios>
@@ -28,11 +29,13 @@
#include "absl/container/flat_hash_set.h"
#include "absl/functional/any_invocable.h"
#include "absl/strings/escaping.h"
#include "absl/strings/str_format.h"
#include "absl/strings/string_view.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/v3/bandwidth_info.h"
#include "connections/v3/connection_listening_options.h"
#include "connections/v3/connections_device_provider.h"
#include "internal/analytics/event_logger.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/error_code_recorder.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/implementation/platform.h"
@@ -67,6 +70,12 @@ ClientProxy::ClientProxy(::nearby::analytics::EventLogger* event_logger)
});
local_os_info_.set_type(
OSNameToOsInfoType(api::ImplementationPlatform::GetCurrentOS()));
supports_safe_to_disconnect_ = NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::
kEnableSafeToDisconnect);
local_safe_to_disconnect_version_ = NearbyFlags::GetInstance().GetInt64Flag(
config_package_nearby::nearby_connections_feature::
kSafeToDisconnectVersion);
}
ClientProxy::~ClientProxy() { Reset(); }
@@ -807,6 +816,44 @@ void ClientProxy::SetRemoteOsInfo(absl::string_view endpoint_id,
item->first.os_info.emplace(remote_os_info);
}
}
std::optional<std::int32_t> ClientProxy::GetRemoteSafeToDisconnectVersion(
absl::string_view endpoint_id) const {
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->first.safe_to_disconnect_version;
}
return std::nullopt;
}
void ClientProxy::SetRemoteSafeToDisconnectVersion(
absl::string_view endpoint_id,
const std::int32_t& safe_to_disconnect_version) {
ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
item->first.safe_to_disconnect_version = safe_to_disconnect_version;
}
}
bool ClientProxy::IsSafeToDisconnectEnabled(absl::string_view endpoint_id) {
return IsSupportSafeToDisconnect() &&
GetRemoteSafeToDisconnectVersion(endpoint_id).has_value() &&
(GetRemoteSafeToDisconnectVersion(endpoint_id) >=
FeatureFlags::GetInstance()
.GetFlags()
.min_nc_version_supports_safe_to_disconnect);
}
bool ClientProxy::IsPayloadReceivedAckEnabled(absl::string_view endpoint_id) {
return IsSupportSafeToDisconnect() &&
GetRemoteSafeToDisconnectVersion(endpoint_id).has_value() &&
(GetRemoteSafeToDisconnectVersion(endpoint_id) >=
FeatureFlags::GetInstance()
.GetFlags()
.min_nc_version_supports_payload_received_ack);
}
void ClientProxy::CancelAllEndpoints() {
for (const auto& item : cancellation_flags_) {
CancellationFlag* cancellation_flag = item.second.get();
connections/implementation/client_proxy.h
+17
View File
@@ -267,6 +267,20 @@ class ClientProxy final {
connections_device_provider_ = std::move(provider);
}
const bool& IsSupportSafeToDisconnect() const {
return supports_safe_to_disconnect_;
}
const std::int32_t& GetLocalSafeToDisconnectVersion() const {
return local_safe_to_disconnect_version_;
}
std::optional<std::int32_t> GetRemoteSafeToDisconnectVersion(
absl::string_view endpoint_id) const;
void SetRemoteSafeToDisconnectVersion(
absl::string_view endpoint_id,
const std::int32_t& safe_to_disconnect_version);
bool IsSafeToDisconnectEnabled(absl::string_view endpoint_id);
bool IsPayloadReceivedAckEnabled(absl::string_view endpoint_id);
private:
struct Connection {
// Status: may be either:
@@ -296,6 +310,7 @@ class ClientProxy final {
AdvertisingOptions advertising_options;
std::string connection_token;
std::optional<location::nearby::connections::OsInfo> os_info;
std::int32_t safe_to_disconnect_version;
};
using ConnectionPair = std::pair<Connection, PayloadListener>;
@@ -427,6 +442,8 @@ class ClientProxy final {
NearbyDeviceProvider* external_device_provider_ = nullptr;
// For Nearby Connections' own device provider.
std::unique_ptr<v3::ConnectionsDeviceProvider> connections_device_provider_;
bool supports_safe_to_disconnect_;
std::int32_t local_safe_to_disconnect_version_;
};
} // namespace connections
connections/implementation/client_proxy_test.cc
+10 -1
View File
@@ -14,6 +14,7 @@
#include "connections/implementation/client_proxy.h"
#include <cstdint>
#include <cstdio>
#include <memory>
#include <optional>
@@ -40,7 +41,7 @@
#include "internal/platform/count_down_latch.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/medium_environment.h"
#include "proto/connections_enums.proto.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
@@ -1023,6 +1024,9 @@ TEST_F(ClientProxyTest, GetRemoteInfoNullWithoutConnections) {
StartAdvertising(&client1_, advertising_connection_listener_);
EXPECT_FALSE(client1_.GetRemoteOsInfo(advertising_endpoint.id).has_value());
EXPECT_FALSE(
client1_.GetRemoteSafeToDisconnectVersion(advertising_endpoint.id)
.has_value());
}
TEST_F(ClientProxyTest, SetRemoteInfoCorrect) {
@@ -1032,11 +1036,16 @@ TEST_F(ClientProxyTest, SetRemoteInfoCorrect) {
OsInfo os_info;
os_info.set_type(OsInfo::ANDROID);
std::int32_t nearby_connections_version = 2;
client1_.SetRemoteOsInfo(advertising_endpoint.id, os_info);
client1_.SetRemoteSafeToDisconnectVersion(advertising_endpoint.id,
nearby_connections_version);
ASSERT_TRUE(client1_.GetRemoteOsInfo(advertising_endpoint.id).has_value());
EXPECT_EQ(client1_.GetRemoteOsInfo(advertising_endpoint.id).value().type(),
OsInfo::ANDROID);
EXPECT_EQ(client1_.GetRemoteSafeToDisconnectVersion(advertising_endpoint.id),
nearby_connections_version);
}
// Test ClientProxy::AddCancellationFlag, where if a flag is already in the map,
connections/implementation/endpoint_channel_manager.cc
+159 -11
View File
@@ -20,12 +20,16 @@
#include "absl/time/time.h"
#include "connections/implementation/offline_frames.h"
#include "internal/platform/condition_variable.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex.h"
#include "internal/platform/mutex_lock.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
using ::location::nearby::analytics::proto::ConnectionsLog;
namespace {
const absl::Duration kDataTransferDelay = absl::Milliseconds(500);
@@ -97,7 +101,8 @@ void EndpointChannelManager::SetActiveEndpointChannel(
// crypto context is present.
channel->SetAnalyticsRecorder(&client->GetAnalyticsRecorder(), endpoint_id);
channel_state_.UpdateChannelForEndpoint(endpoint_id, std::move(channel));
channel_state_.UpdateSafeToDisconnectForEndpoint(
endpoint_id, client->IsSafeToDisconnectEnabled(endpoint_id));
auto* endpoint = channel_state_.LookupEndpointData(endpoint_id);
if (endpoint->IsEncrypted() && enable_encryption)
channel_state_.EncryptChannel(endpoint);
@@ -113,6 +118,37 @@ bool EndpointChannelManager::isWifiLanConnected() const {
return channel_state_.isWifiLanConnected();
}
void EndpointChannelManager::UpdateSafeToDisconnectForEndpoint(
const std::string& endpoint_id,
bool safe_to_disconnect_enabled) {
MutexLock lock(&mutex_);
channel_state_.UpdateSafeToDisconnectForEndpoint(endpoint_id,
safe_to_disconnect_enabled);
}
void EndpointChannelManager::MarkEndpointStopWaitToDisconnect(
const std::string& endpoint_id, bool is_safe_to_disconnect,
bool notify_stop_waiting) {
MutexLock lock(&mutex_);
channel_state_.MarkEndpointStopWaitToDisconnect(
endpoint_id, is_safe_to_disconnect, notify_stop_waiting);
}
bool EndpointChannelManager::CreateNewTimeoutDisconnectedState(
const std::string& endpoint_id) {
return channel_state_.CreateNewTimeoutDisconnectedState(endpoint_id);
}
bool EndpointChannelManager::IsSafeToDisconnect(
const std::string& endpoint_id) {
return channel_state_.IsSafeToDisconnect(endpoint_id);
}
void EndpointChannelManager::RemoveTimeoutDisconnectedState(
const std::string& endpoint_id) {
MutexLock lock(&mutex_);
channel_state_.RemoveTimeoutDisconnectedState(endpoint_id);
}
///////////////////////////////// ChannelState /////////////////////////////////
// endpoint - channel endpoint to encrypt
@@ -133,6 +169,15 @@ EndpointChannelManager::ChannelState::LookupEndpointData(
return item != endpoints_.end() ? &item->second : nullptr;
}
void EndpointChannelManager::ChannelState::DestroyAll() {
for (auto& item : endpoints_) {
RemoveEndpoint(item.first, DisconnectionReason::SHUTDOWN,
/* safe_to_disconnect_enabled */ false,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
}
endpoints_.clear();
}
void EndpointChannelManager::ChannelState::UpdateChannelForEndpoint(
const std::string& endpoint_id, std::unique_ptr<EndpointChannel> channel) {
// Create EndpointData instance, if necessary, and populate channel.
@@ -146,24 +191,54 @@ void EndpointChannelManager::ChannelState::UpdateEncryptionContextForEndpoint(
endpoints_[endpoint_id].context = std::move(context);
}
bool EndpointChannelManager::ChannelState::RemoveEndpoint(
void EndpointChannelManager::ChannelState::UpdateSafeToDisconnectForEndpoint(
const std::string& endpoint_id,
location::nearby::proto::connections::DisconnectionReason reason) {
bool safe_to_disconnect_enabled) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect] "
"UpdateSafeToDisconnectForEndpoint for: "
<< endpoint_id << " " << safe_to_disconnect_enabled;
endpoints_[endpoint_id].safe_to_disconnect_enabled =
safe_to_disconnect_enabled;
}
bool EndpointChannelManager::ChannelState::GetSafeToDisconnectForEndpoint(
const std::string& endpoint_id) {
auto item = endpoints_.find(endpoint_id);
if (item == endpoints_.end()) return false;
NEARBY_LOGS(INFO) << "[safe-to-disconnect] GetSafeToDisconnectForEndpoint: "
<< item->second.safe_to_disconnect_enabled;
return item->second.safe_to_disconnect_enabled;
}
bool EndpointChannelManager::ChannelState::RemoveEndpoint(
const std::string& endpoint_id, DisconnectionReason reason,
bool safe_to_disconnect_enabled, SafeDisconnectionResult result) {
auto item = endpoints_.find(endpoint_id);
if (item == endpoints_.end()) return false;
MarkEndpointStopWaitToDisconnect(endpoint_id,
/* is_safe_to_disconnect */ true,
/* notify_stop_waiting */ true);
item->second.disconnect_reason = reason;
auto channel = item->second.channel;
if (channel) {
if (channel && !safe_to_disconnect_enabled) {
// If the channel was paused (i.e. during a bandwidth upgrade negotiation)
// we resume to ensure the thread won't hang when trying to write to it.
channel->Resume();
channel->Write(parser::ForDisconnection());
NEARBY_LOGS(INFO) << "[safe-to-disconnect] Sending DISCONNECTION frame"
" with request 0, ack 0";
channel->Write(
parser::ForDisconnection(/* request_safe_to_disconnect */ false,
/* ack_safe_to_disconnect */ false));
NEARBY_LOGS(INFO)
<< "EndpointChannelManager reported the disconnection to endpoint "
<< endpoint_id;
SystemClock::Sleep(kDataTransferDelay);
}
NEARBY_LOGS(INFO) << "Remove Endpoint: " << endpoint_id;
endpoints_.erase(item);
return true;
}
@@ -183,20 +258,93 @@ bool EndpointChannelManager::ChannelState::isWifiLanConnected() const {
return false;
}
bool EndpointChannelManager::UnregisterChannelForEndpoint(
void EndpointChannelManager::ChannelState::MarkEndpointStopWaitToDisconnect(
const std::string& endpoint_id, bool is_safe_to_disconnect,
bool notify_stop_waiting) {
auto item = endpoints_.find(endpoint_id);
if (item == endpoints_.end()) return;
NEARBY_LOGS(INFO) << "[safe-to-disconnect] is_safe_to_disconnect= "
<< is_safe_to_disconnect
<< ", notify_stop_waiting= " << notify_stop_waiting
<< " for endpoint: " << endpoint_id;
{
MutexLock lock(&item->second.timeout_to_disconnected_mutex);
item->second.is_safe_to_disconnect = is_safe_to_disconnect;
if (!item->second.timeout_to_disconnected_enabled) return;
if (notify_stop_waiting) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect] Notify stop "
"waiting before timeout.";
item->second.timeout_to_disconnected.Notify();
item->second.timeout_to_disconnected_notified = true;
}
}
}
bool EndpointChannelManager::ChannelState::CreateNewTimeoutDisconnectedState(
const std::string& endpoint_id) {
auto item = endpoints_.find(endpoint_id);
if (item == endpoints_.end()) return false;
NEARBY_LOGS(INFO) << "[safe-to-disconnect] "
"Create TimeoutDisconnectedState for endpoint: "
<< endpoint_id;
{
MutexLock lock(&item->second.timeout_to_disconnected_mutex);
item->second.timeout_to_disconnected_enabled = true;
item->second.timeout_to_disconnected_notified = false;
item->second.timeout_to_disconnected.Wait(FeatureFlags::GetInstance()
.GetFlags()
.safe_to_disconnect_ack_delay_millis);
NEARBY_LOGS(INFO) << "[safe-to-disconnect] Wait is done with "
<< (item->second.timeout_to_disconnected_notified
? "notification"
: "timeout");
if (!item->second.timeout_to_disconnected_notified)
item->second.is_safe_to_disconnect = true;
item->second.timeout_to_disconnected_notified = false;
item->second.timeout_to_disconnected_enabled = false;
}
return true;
}
bool EndpointChannelManager::ChannelState::IsSafeToDisconnect(
const std::string& endpoint_id) {
auto item = endpoints_.find(endpoint_id);
if (item == endpoints_.end()) return true;
{
MutexLock lock(&item->second.timeout_to_disconnected_mutex);
NEARBY_LOGS(INFO)
<< "[safe-to-disconnect] Get SafeToDisconnect status for endpoint: "
<< endpoint_id << ": " << item->second.is_safe_to_disconnect;
return (item->second.is_safe_to_disconnect);
}
}
void EndpointChannelManager::ChannelState::RemoveTimeoutDisconnectedState(
const std::string& endpoint_id) {
auto item = endpoints_.find(endpoint_id);
if (item == endpoints_.end()) return;
{
MutexLock lock(&item->second.timeout_to_disconnected_mutex);
item->second.timeout_to_disconnected_notified = false;
item->second.timeout_to_disconnected_enabled = false;
}
}
bool EndpointChannelManager::UnregisterChannelForEndpoint(
const std::string& endpoint_id, DisconnectionReason reason,
SafeDisconnectionResult result) {
MutexLock lock(&mutex_);
if (!channel_state_.RemoveEndpoint(
endpoint_id, location::nearby::proto::connections::
DisconnectionReason::LOCAL_DISCONNECTION)) {
auto safe_to_disconnect_enabled =
channel_state_.GetSafeToDisconnectForEndpoint(endpoint_id);
if (!channel_state_.RemoveEndpoint(endpoint_id, reason,
safe_to_disconnect_enabled, result)) {
return false;
}
NEARBY_LOGS(INFO)
<< "EndpointChannelManager unregistered channel for endpoint "
<< endpoint_id;
return true;
}
connections/implementation/endpoint_channel_manager.h
+51 -9
View File
@@ -17,15 +17,23 @@
#include <memory>
#include <string>
#include <utility>
#include "securegcm/d2d_connection_context_v1.h"
#include "absl/base/thread_annotations.h"
#include "absl/container/flat_hash_map.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/mutex.h"
#include "internal/proto/analytics/connections_log.pb.h"
namespace nearby {
namespace connections {
using DisconnectionReason =
::location::nearby::proto::connections::DisconnectionReason;
using SafeDisconnectionResult = ::location::nearby::analytics::proto::
ConnectionsLog::EstablishedConnection::SafeDisconnectionResult;
// NOTE(std::string):
// All the strings in internal class public interfaces should be exchanged as
@@ -90,13 +98,28 @@ class EndpointChannelManager final {
// Returns true if 'endpoint_id' actually had a registered EndpointChannel.
// IOW, a return of false signifies a no-op.
bool UnregisterChannelForEndpoint(const std::string& endpoint_id)
bool UnregisterChannelForEndpoint(const std::string& endpoint_id,
DisconnectionReason reason,
SafeDisconnectionResult result)
ABSL_LOCKS_EXCLUDED(mutex_);
int GetConnectedEndpointsCount() const ABSL_LOCKS_EXCLUDED(mutex_);
// Check if any endpoint uses WLAN Medium
bool isWifiLanConnected() const ABSL_LOCKS_EXCLUDED(mutex_);
void UpdateSafeToDisconnectForEndpoint(const std::string& endpoint_id,
bool safe_to_disconnect_enabled)
ABSL_LOCKS_EXCLUDED(mutex_);
void MarkEndpointStopWaitToDisconnect(const std::string& endpoint_id,
bool is_safe_to_disconnect,
bool notify_stop_waiting)
ABSL_LOCKS_EXCLUDED(mutex_);
bool CreateNewTimeoutDisconnectedState(const std::string& endpoint_id)
ABSL_LOCKS_EXCLUDED(mutex_);
bool IsSafeToDisconnect(const std::string& endpoint_id)
ABSL_LOCKS_EXCLUDED(mutex_);
void RemoveTimeoutDisconnectedState(const std::string& endpoint_id)
ABSL_LOCKS_EXCLUDED(mutex_);
private:
// Tracks channel state for all endpoints. This includes what EndpointChannel
@@ -119,9 +142,17 @@ class EndpointChannelManager final {
std::shared_ptr<EndpointChannel> channel;
std::shared_ptr<EncryptionContext> context;
location::nearby::proto::connections::DisconnectionReason
disconnect_reason = location::nearby::proto::connections::
DisconnectionReason::UNKNOWN_DISCONNECTION_REASON;
DisconnectionReason disconnect_reason =
DisconnectionReason::UNKNOWN_DISCONNECTION_REASON;
bool safe_to_disconnect_enabled = false;
mutable Mutex timeout_to_disconnected_mutex;
ConditionVariable timeout_to_disconnected{&timeout_to_disconnected_mutex};
bool timeout_to_disconnected_enabled
ABSL_GUARDED_BY(timeout_to_disconnected_mutex) = false;
bool timeout_to_disconnected_notified
ABSL_GUARDED_BY(timeout_to_disconnected_mutex) = false;
bool is_safe_to_disconnect
ABSL_GUARDED_BY(timeout_to_disconnected_mutex) = false;
};
ChannelState() = default;
@@ -130,7 +161,7 @@ class EndpointChannelManager final {
ChannelState& operator=(ChannelState&&) = default;
// Provides a way to destroy contents of a container, while holding a lock.
void DestroyAll() { endpoints_.clear(); }
void DestroyAll();
// Return pointer to endpoint data, or nullptr, it not found.
EndpointData* LookupEndpointData(const std::string& endpoint_id);
@@ -145,15 +176,26 @@ class EndpointChannelManager final {
const std::string& endpoint_id,
std::unique_ptr<EncryptionContext> context);
void UpdateSafeToDisconnectForEndpoint(const std::string& endpoint_id,
bool safe_to_disconnect_enabled);
bool GetSafeToDisconnectForEndpoint(const std::string& endpoint_id);
// Removes all knowledge of this endpoint, cleaning up as necessary.
// Returns false if the endpoint was not found.
bool RemoveEndpoint(
const std::string& endpoint_id,
location::nearby::proto::connections::DisconnectionReason reason);
bool RemoveEndpoint(const std::string& endpoint_id,
DisconnectionReason reason,
bool safe_to_disconnect_enabled,
SafeDisconnectionResult result);
bool EncryptChannel(EndpointData* endpoint);
int GetConnectedEndpointsCount() const { return endpoints_.size(); }
bool isWifiLanConnected() const;
void MarkEndpointStopWaitToDisconnect(const std::string& endpoint_id,
bool is_safe_to_disconnect,
bool notify_stop_waiting);
bool CreateNewTimeoutDisconnectedState(const std::string& endpoint_id);
bool IsSafeToDisconnect(const std::string& endpoint_id);
void RemoveTimeoutDisconnectedState(const std::string& endpoint_id);
private:
// Endpoint ID -> EndpointData. Contains everything we know about the
@@ -168,7 +210,7 @@ class EndpointChannelManager final {
ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
mutable Mutex mutex_;
ChannelState channel_state_ ABSL_GUARDED_BY(mutex_);
ChannelState channel_state_;
};
} // namespace connections
connections/implementation/endpoint_channel_manager_test.cc
+14 -1
View File
@@ -39,12 +39,14 @@
#include "internal/platform/multi_thread_executor.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/pipe.h"
#include "internal/proto/analytics/connections_log.pb.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
namespace {
using ::location::nearby::analytics::proto::ConnectionsLog;
using ::location::nearby::proto::connections::DisconnectionReason;
using ::location::nearby::proto::connections::Medium;
using EncryptionContext = BaseEndpointChannel::EncryptionContext;
@@ -240,6 +242,12 @@ TEST(BaseEndpointChannelManagerTest, RegisterChannelEncryptedReadwrite) {
// Shutdown test environment.
channel_a_raw->Close(DisconnectionReason::LOCAL_DISCONNECTION);
channel_b_raw->Close(DisconnectionReason::REMOTE_DISCONNECTION);
ecm_a.UnregisterChannelForEndpoint(
std::string(kEndpointId), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
ecm_b.UnregisterChannelForEndpoint(
std::string(kEndpointId), DisconnectionReason::REMOTE_DISCONNECTION,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
}
TEST(BaseEndpointChannelManagerTest, ReplaceChannelNoEncrypted) {
@@ -302,7 +310,12 @@ TEST(BaseEndpointChannelManagerTest, ReplaceChannelNoEncrypted) {
// Shutdown test environment.
channel_a_raw->Close(DisconnectionReason::LOCAL_DISCONNECTION);
channel_b_raw->Close(DisconnectionReason::REMOTE_DISCONNECTION);
}
ecm_a.UnregisterChannelForEndpoint(
std::string(kEndpointId), DisconnectionReason::LOCAL_DISCONNECTION,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
ecm_b.UnregisterChannelForEndpoint(
std::string(kEndpointId), DisconnectionReason::REMOTE_DISCONNECTION,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);}
} // namespace
} // namespace connections
connections/implementation/endpoint_manager.cc
+210 -50
View File
@@ -21,8 +21,11 @@
#include <utility>
#include <vector>
#include "absl/time/time.h"
#include "connections/implementation/analytics/throughput_recorder.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/offline_frames.h"
#include "connections/implementation/payload_manager.h"
#include "connections/implementation/proto/offline_wire_formats.pb.h"
@@ -30,12 +33,16 @@
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex.h"
#include "internal/platform/mutex_lock.h"
#include "internal/proto/analytics/connections_log.pb.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
namespace {
using ::location::nearby::analytics::proto::ConnectionsLog;
using ::location::nearby::connections::OfflineFrame;
using ::location::nearby::connections::V1Frame;
using ::nearby::analytics::PacketMetaData;
@@ -162,6 +169,11 @@ void EndpointManager::EndpointChannelLoopRunnable(
NEARBY_LOGS(INFO) << "Dropping current channel: last medium="
<< location::nearby::proto::connections::Medium_Name(
last_failed_medium);
if (client->IsSafeToDisconnectEnabled(endpoint_id)) {
channel_manager_->MarkEndpointStopWaitToDisconnect(
endpoint_id, /* is_safe_to_disconnect */ false,
/* notify_stop_waiting */ true);
}
break;
}
}
@@ -171,7 +183,7 @@ void EndpointManager::EndpointChannelLoopRunnable(
<< "; endpoint_id=" << endpoint_id;
// Always clear out all state related to this endpoint before terminating
// this thread.
DiscardEndpoint(client, endpoint_id);
DiscardEndpoint(client, endpoint_id, DisconnectionReason::IO_ERROR);
NEARBY_LOGS(INFO) << "Worker done; worker name=" << runnable_name
<< "; endpoint_id=" << endpoint_id;
}
@@ -260,7 +272,7 @@ ExceptionOr<bool> EndpointManager::HandleData(
} else if (frame_type == V1Frame::DISCONNECTION) {
NEARBY_LOG(INFO, "Disconnect message for endpoint %s",
endpoint_id.c_str());
endpoint_channel->Close();
ProcessDisconnectionFrame(client, endpoint_id, endpoint_channel, frame);
} else {
NEARBY_LOGS(ERROR) << "Unhandled message: endpoint_id=" << endpoint_id
<< ", frame type="
@@ -275,6 +287,62 @@ ExceptionOr<bool> EndpointManager::HandleData(
}
}
void EndpointManager::ProcessDisconnectionFrame(
ClientProxy* client, const std::string& endpoint_id,
EndpointChannel* endpoint_channel, OfflineFrame& frame) {
if (!client->IsSafeToDisconnectEnabled(endpoint_id)) {
NEARBY_LOGS(INFO)
<< "EndpointManager received a DISCONNECTION frame from endpoint "
<< endpoint_id << " on channel " << endpoint_channel->GetType()
<< ", disconnecting...";
endpoint_channel->Close(DisconnectionReason::REMOTE_DISCONNECTION);
return;
}
if (!frame.v1().has_disconnection() ||
!frame.v1().disconnection().has_request_safe_to_disconnect() ||
!frame.v1().disconnection().request_safe_to_disconnect()) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect] no need to apply "
"safe-to-disconnect protocol for endpoint "
<< endpoint_id << " on channel "
<< endpoint_channel->GetType() << ", disconnecting...";
endpoint_channel->Close(DisconnectionReason::REMOTE_DISCONNECTION);
return;
}
NEARBY_LOGS(INFO)
<< "[safe-to-disconnect] received a "
"DISCONNECTION frame with request safe to disconnect = true and ack = "
<< frame.v1().disconnection().ack_safe_to_disconnect()
<< " from endpoint " << endpoint_id << " on channel "
<< endpoint_channel->GetType()
<< ", disconnecting with safe-to-disconnect protocol ...";
if (frame.v1().disconnection().ack_safe_to_disconnect()) {
channel_manager_->MarkEndpointStopWaitToDisconnect(
endpoint_id, /* is_safe_to_disconnect */ true,
/* notify_stop_waiting */ true);
} else {
channel_manager_->MarkEndpointStopWaitToDisconnect(
endpoint_id, /* is_safe_to_disconnect */ true,
/* notify_stop_waiting */ false);
RunOnEndpointManagerThread(
"safe-to-disconnect", [this, client, &endpoint_id]() {
RemoveEndpoint(client, endpoint_id, /*notify=*/true,
DisconnectionReason::REMOTE_DISCONNECTION);
});
endpoint_channel->Resume();
NEARBY_LOGS(INFO) << "[safe-to-disconnect] Sending "
"DISCONNECTION frame with request 1, ack 1";
Exception write_exception = endpoint_channel->Write(
parser::ForDisconnection(/* request_safe_to_disconnect= */ true,
/* ack_safe_to_disconnect= */ true));
if (!write_exception.Ok()) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect] Failed to send "
"DISCONNECTION frame with ack to endpoint"
<< endpoint_id;
}
}
}
ExceptionOr<bool> EndpointManager::HandleKeepAlive(
EndpointChannel* endpoint_channel, absl::Duration keep_alive_interval,
absl::Duration keep_alive_timeout, Mutex* keep_alive_waiter_mutex,
@@ -290,10 +358,10 @@ ExceptionOr<bool> EndpointManager::HandleKeepAlive(
return ExceptionOr<bool>(false);
}
// If we haven't written anything to the endpoint for a while, attempt to send
// the KeepAlive frame over the endpoint channel. If the write fails, our
// super class will loop back around and try our luck again in case there's
// been a replacement for this endpoint.
// If we haven't written anything to the endpoint for a while, attempt to
// send the KeepAlive frame over the endpoint channel. If the write fails,
// our super class will loop back around and try our luck again in case
// there's been a replacement for this endpoint.
absl::Time last_write_time = endpoint_channel->GetLastWriteTimestamp();
absl::Duration duration_until_write_keep_alive =
last_write_time == kInvalidTimestamp
@@ -323,15 +391,15 @@ ExceptionOr<bool> EndpointManager::HandleKeepAlive(
bool operator==(const EndpointManager::FrameProcessor& lhs,
const EndpointManager::FrameProcessor& rhs) {
// We're comparing addresses because these objects are callbacks which need to
// be matched by exact instances.
// We're comparing addresses because these objects are callbacks which need
// to be matched by exact instances.
return &lhs == &rhs;
}
bool operator<(const EndpointManager::FrameProcessor& lhs,
const EndpointManager::FrameProcessor& rhs) {
// We're comparing addresses because these objects are callbacks which need to
// be matched by exact instances.
// We're comparing addresses because these objects are callbacks which need
// to be matched by exact instances.
return &lhs < &rhs;
}
@@ -444,10 +512,11 @@ void EndpointManager::RegisterEndpoint(
// NOTE (unique_ptr<> capture):
// std::unique_ptr<> is not copyable, so we can not pass it to
// lambda capture, because lambda eventually is converted to std::function<>.
// Instead, we release() a pointer, and pass a raw pointer, which is copyalbe.
// We ignore the risk of job not scheduled (and an associated risk of memory
// leak), because this may only happen during service shutdown.
// lambda capture, because lambda eventually is converted to
// std::function<>. Instead, we release() a pointer, and pass a raw pointer,
// which is copyalbe. We ignore the risk of job not scheduled (and an
// associated risk of memory leak), because this may only happen during
// service shutdown.
RunOnEndpointManagerThread("register-endpoint", [this, client,
channel = channel.release(),
&endpoint_id, &info,
@@ -456,8 +525,8 @@ void EndpointManager::RegisterEndpoint(
&latch]() {
if (endpoints_.contains(endpoint_id)) {
NEARBY_LOGS(WARNING) << "Registering duplicate endpoint " << endpoint_id;
// We must remove old endpoint state before registering a new one for the
// same endpoint_id.
// We must remove old endpoint state before registering a new one
// for the same endpoint_id.
RemoveEndpointState(endpoint_id);
}
@@ -487,8 +556,8 @@ void EndpointManager::RegisterEndpoint(
// For every endpoint, there's normally only one Read handler instance
// running on a dedicated thread. This instance reads data from the
// endpoint and delegates incoming frames to various FrameProcessors.
// Once the frame has been properly handled, it starts reading again for
// the next frame. If the handler fails its read and no other
// Once the frame has been properly handled, it starts reading again
// for the next frame. If the handler fails its read and no other
// EndpointChannels are available for this endpoint, a disconnection
// will be initiated.
endpoint_state.StartEndpointReader([this, client, endpoint_id]() {
@@ -499,17 +568,18 @@ void EndpointManager::RegisterEndpoint(
});
});
// For every endpoint, there's only one KeepAliveManager instance running on
// a dedicated thread. This instance will periodically send out a ping* to
// the endpoint while listening for an incoming pong**. If it fails to send
// the ping, or if no pong is heard within keep_alive_timeout, it initiates
// a disconnection.
// For every endpoint, there's only one KeepAliveManager instance
// running on a dedicated thread. This instance will periodically send
// out a ping* to the endpoint while listening for an incoming pong**.
// If it fails to send the ping, or if no pong is heard within
// keep_alive_timeout, it initiates a disconnection.
//
// (*) Bluetooth requires a constant outgoing stream of messages. If
// there's silence, Android will break the socket. This is why we ping.
// (**) Wifi Hotspots can fail to notice a connection has been lost, and
// they will happily keep writing to /dev/null. This is why we listen
// for the pong.
// there's silence, Android will break the socket. This is why we
// ping.
// (**) Wifi Hotspots can fail to notice a connection has been lost,
// and they will happily keep writing to /dev/null. This is why we
// listen for the pong.
NEARBY_LOGS(VERBOSE) << "EndpointManager enabling KeepAlive for endpoint "
<< endpoint_id;
endpoint_state.StartEndpointKeepAliveManager(
@@ -545,7 +615,8 @@ void EndpointManager::UnregisterEndpoint(ClientProxy* client,
RunOnEndpointManagerThread(
"unregister-endpoint", [this, client, endpoint_id, &latch]() {
RemoveEndpoint(client, endpoint_id,
/*notify=*/client->IsConnectedToEndpoint(endpoint_id));
/*notify=*/client->IsConnectedToEndpoint(endpoint_id),
DisconnectionReason::LOCAL_DISCONNECTION);
latch.CountDown();
});
latch.Await();
@@ -578,12 +649,19 @@ std::vector<std::string> EndpointManager::SendPayloadChunk(
}
// Designed to run asynchronously. It is called from IO thread pools, and
// jobs in these pools may be waited for from the EndpointManager thread. If we
// allow synchronous behavior here it will cause a live lock.
// jobs in these pools may be waited for from the EndpointManager thread. If
// we allow synchronous behavior here it will cause a live lock.
void EndpointManager::DiscardEndpoint(ClientProxy* client,
const std::string& endpoint_id) {
NEARBY_LOGS(VERBOSE) << "DiscardEndpoint for endpoint " << endpoint_id;
RunOnEndpointManagerThread("discard-endpoint", [this, client, endpoint_id]() {
const std::string& endpoint_id,
DisconnectionReason reason) {
NEARBY_LOGS(INFO) << "DiscardEndpoint for endpoint " << endpoint_id;
if (reason == DisconnectionReason::IO_ERROR) {
channel_manager_->MarkEndpointStopWaitToDisconnect(
endpoint_id, /* is_safe_to_disconnect */ false,
/* notify_stop_waiting */ true);
}
RunOnEndpointManagerThread("discard-endpoint", [this, client, endpoint_id,
reason]() {
// `ClientProxy` is destroyed before `EndpointManager` in
// `~NearbyConnections`, which means "discard-endpoint" needs to check
// if this task is being executing during `~EndpointManager` to
@@ -625,7 +703,8 @@ void EndpointManager::DiscardEndpoint(ClientProxy* client,
}
RemoveEndpoint(client, endpoint_id,
/*notify=*/client->IsConnectedToEndpoint(endpoint_id));
/* notify */client->IsConnectedToEndpoint(endpoint_id),
reason);
});
}
@@ -647,8 +726,12 @@ std::vector<std::string> EndpointManager::SendControlMessage(
// @EndpointManagerThread
void EndpointManager::RemoveEndpoint(ClientProxy* client,
const std::string& endpoint_id,
bool notify) {
NEARBY_LOGS(INFO) << "RemoveEndpoint for endpoint " << endpoint_id;
bool notify, DisconnectionReason reason) {
NEARBY_LOGS(INFO) << "RemoveEndpoint for endpoint: " << endpoint_id
<< ", reason: " << reason;
SafeDisconnectionResult safe_disconnect_result =
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION;
// Grab the service ID before we destroy the channel.
EndpointChannel* channel =
@@ -656,16 +739,35 @@ void EndpointManager::RemoveEndpoint(ClientProxy* client,
std::string service_id =
channel ? channel->GetServiceId() : std::string(kUnknownServiceId);
if (client->IsSafeToDisconnectEnabled(endpoint_id)) {
if (channel != nullptr) {
bool is_safe_disconnection =
ApplySafeToDisconnect(endpoint_id, channel, reason);
safe_disconnect_result =
is_safe_disconnection
? ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION
: ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION;
NEARBY_LOGS(INFO) << "[safe-to-disconnect] safe_disconnect_result:"
<< (safe_disconnect_result? "true" : "false");
}
}
if (safe_disconnect_result ==
ConnectionsLog::EstablishedConnection::UNSAFE_DISCONNECTION) {
// TODO(b/297259496): Autoreconnect
}
// Unregistering from channel_manager_ will also serve to terminate
// the dedicated handler and KeepAlive threads we started when we registered
// this endpoint.
if (channel_manager_->UnregisterChannelForEndpoint(endpoint_id)) {
if (channel_manager_->UnregisterChannelForEndpoint(endpoint_id, reason,
safe_disconnect_result)) {
// Notify all frame processors of the disconnection immediately and wait
// for them to clean up state. Only once all processors are done cleaning
// up, we can remove the endpoint from ClientProxy after which there
// should be no further interactions with the endpoint.
// (See b/37352254 for history)
WaitForEndpointDisconnectionProcessing(client, service_id, endpoint_id);
WaitForEndpointDisconnectionProcessing(client, service_id, endpoint_id,
reason);
client->OnDisconnected(endpoint_id, notify);
NEARBY_LOGS(INFO) << "Removed endpoint for endpoint " << endpoint_id;
@@ -673,15 +775,68 @@ void EndpointManager::RemoveEndpoint(ClientProxy* client,
RemoveEndpointState(endpoint_id);
}
bool EndpointManager::ApplySafeToDisconnect(const std::string& endpoint_id,
EndpointChannel* endpoint_channel,
DisconnectionReason reason) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect] ApplySafeToDisconnect reason: "
<< reason;
bool is_safe_disconnection = false;
bool send_disconnection_frame = true;
switch (reason) {
case DisconnectionReason::UPGRADED:
case DisconnectionReason::SHUTDOWN:
case DisconnectionReason::UNFINISHED:
return true; // safe disconnection
case DisconnectionReason::IO_ERROR:
return false; // unsafe disconnection
case DisconnectionReason::LOCAL_DISCONNECTION:
is_safe_disconnection = true;
send_disconnection_frame = true;
break;
case DisconnectionReason::REMOTE_DISCONNECTION:
is_safe_disconnection = true;
send_disconnection_frame = false;
break;
default:
is_safe_disconnection = false;
send_disconnection_frame = true;
}
if (send_disconnection_frame) {
// If the channel was paused (i.e. during a bandwidth upgrade negotiation)
// we resume to ensure the thread won't hang when trying to write to it.
endpoint_channel->Resume();
NEARBY_LOGS(INFO) << "[safe-to-disconnect] Sending "
"DISCONNECTION frame with request 1, ack 0";
Exception write_exception = endpoint_channel->Write(
parser::ForDisconnection(/* request_safe_to_disconnect= */ true,
/* ack_safe_to_disconnect= */ false));
if (!write_exception.Ok()) {
NEARBY_LOGS(WARNING) << "[safe-to-disconnect] Failed to send "
"DISCONNECTION frame to endpoint"
<< endpoint_id << " for reason: " << reason;
return is_safe_disconnection;
}
}
bool state =
channel_manager_->CreateNewTimeoutDisconnectedState(endpoint_id);
if (!state) return is_safe_disconnection;
return is_safe_disconnection ||
channel_manager_->IsSafeToDisconnect(endpoint_id);
}
// @EndpointManagerThread
void EndpointManager::WaitForEndpointDisconnectionProcessing(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id) {
const std::string& endpoint_id, DisconnectionReason reason) {
NEARBY_LOGS(INFO) << "Wait: client=" << client
<< "; service_id=" << service_id
<< "; endpoint_id=" << endpoint_id;
CountDownLatch barrier = NotifyFrameProcessorsOnEndpointDisconnect(
client, service_id, endpoint_id);
client, service_id, endpoint_id, reason);
NEARBY_LOGS(INFO)
<< "Waiting for frame processors to disconnect from endpoint "
@@ -690,15 +845,15 @@ void EndpointManager::WaitForEndpointDisconnectionProcessing(
NEARBY_LOGS(INFO) << "Failed to disconnect frame processors from endpoint "
<< endpoint_id;
} else {
NEARBY_LOGS(INFO)
<< "Finished waiting for frame processors to disconnect from endpoint "
<< endpoint_id;
NEARBY_LOGS(INFO) << "Finished waiting for frame processors to "
"disconnect from endpoint "
<< endpoint_id;
}
}
CountDownLatch EndpointManager::NotifyFrameProcessorsOnEndpointDisconnect(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id) {
const std::string& endpoint_id, DisconnectionReason reason) {
NEARBY_LOGS(INFO) << "NotifyFrameProcessorsOnEndpointDisconnect: client="
<< client << "; service_id=" << service_id
<< "; endpoint_id=" << endpoint_id;
@@ -714,7 +869,8 @@ CountDownLatch EndpointManager::NotifyFrameProcessorsOnEndpointDisconnect(
<< "; frame type=" << V1Frame::FrameType_Name(item.first);
if (processor) {
valid++;
processor->OnEndpointDisconnect(client, service_id, endpoint_id, barrier);
processor->OnEndpointDisconnect(client, service_id, endpoint_id, barrier,
reason);
} else {
barrier.CountDown();
}
@@ -739,7 +895,8 @@ std::vector<std::string> EndpointManager::SendTransferFrameBytes(
if (channel == nullptr) {
// We no longer know about this endpoint (it was either explicitly
// unregistered, or a read/write error made us unregister it internally).
// unregistered, or a read/write error made us unregister it
// internally).
NEARBY_LOGS(ERROR) << "EndpointManager failed to find EndpointChannel "
"over which to write "
<< packet_type << " at offset " << offset
@@ -766,12 +923,15 @@ std::vector<std::string> EndpointManager::SendTransferFrameBytes(
EndpointManager::EndpointState::~EndpointState() {
// We must unregister the endpoint first to signal the runnables that they
// should exit their loops. SingleThreadExecutor destructors will wait for the
// workers to finish. |channel_manager_| is null after moved from this object
// (in move constructor) which prevents unregistering the channel prematurely.
// should exit their loops. SingleThreadExecutor destructors will wait for
// the workers to finish. |channel_manager_| is null after moved from this
// object (in move constructor) which prevents unregistering the channel
// prematurely.
if (channel_manager_) {
NEARBY_LOG(VERBOSE, "EndpointState destructor %s", endpoint_id_.c_str());
channel_manager_->UnregisterChannelForEndpoint(endpoint_id_);
channel_manager_->UnregisterChannelForEndpoint(
endpoint_id_, DisconnectionReason::SHUTDOWN,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
}
// Make sure the KeepAlive thread isn't blocking shutdown.
connections/implementation/endpoint_manager.h
+15 -7
View File
@@ -90,7 +90,8 @@ class EndpointManager {
virtual void OnEndpointDisconnect(ClientProxy* client,
const std::string& service_id,
const std::string& endpoint_id,
CountDownLatch barrier) = 0;
CountDownLatch barrier,
DisconnectionReason reason) = 0;
};
explicit EndpointManager(EndpointChannelManager* manager);
@@ -154,7 +155,8 @@ class EndpointManager {
// ask everyone who's registered an FrameProcessor to
// processEndpointDisconnection() while the caller of DiscardEndpoint() is
// blocked here.
void DiscardEndpoint(ClientProxy* client, const std::string& endpoint_id);
void DiscardEndpoint(ClientProxy* client, const std::string& endpoint_id,
DisconnectionReason reason);
protected:
// For unit tests only to control executing tasks on the executor.
@@ -260,15 +262,21 @@ class EndpointManager {
// this method is idempotent.
// @EndpointManagerThread
void RemoveEndpoint(ClientProxy* client, const std::string& endpoint_id,
bool notify);
bool notify, DisconnectionReason reason);
bool ApplySafeToDisconnect(const std::string& endpoint_id,
EndpointChannel* endpoint_channel,
DisconnectionReason reason);
void WaitForEndpointDisconnectionProcessing(ClientProxy* client,
const std::string& service_id,
const std::string& endpoint_id);
const std::string& endpoint_id,
DisconnectionReason reason);
void ProcessDisconnectionFrame(
ClientProxy* client, const std::string& endpoint_id,
EndpointChannel* endpoint_channel,
location::nearby::connections::OfflineFrame& frame);
CountDownLatch NotifyFrameProcessorsOnEndpointDisconnect(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id);
const std::string& endpoint_id, DisconnectionReason reason);
std::vector<std::string> SendTransferFrameBytes(
const std::vector<std::string>& endpoint_ids,
connections/implementation/endpoint_manager_test.cc
+34 -6
View File
@@ -29,10 +29,13 @@
#include "connections/connection_options.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/offline_frames.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
// #include "internal/platform/feature_flags.h"
#include "internal/platform/logging.h"
#include "internal/test/fake_single_thread_executor.h"
#include "proto/connections_enums.pb.h"
@@ -112,10 +115,21 @@ class MockFrameProcessor : public EndpointManager::FrameProcessor {
MOCK_METHOD(void, OnEndpointDisconnect,
(ClientProxy * client, const std::string& service_id,
const std::string& endpoint_id, CountDownLatch barrier),
const std::string& endpoint_id, CountDownLatch barrier,
DisconnectionReason reason),
(override));
};
class SetSafeToDisconnect {
public:
explicit SetSafeToDisconnect(bool safe_to_disconnect) {
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::
kEnableSafeToDisconnect,
safe_to_disconnect);
}
};
class TestEndpointManager : public EndpointManager {
public:
TestEndpointManager(EndpointChannelManager* manager,
@@ -146,7 +160,7 @@ class EndpointManagerTest : public ::testing::Test {
EXPECT_TRUE(done.Await(absl::Milliseconds(1000)).result());
}
}
SetSafeToDisconnect set_safe_to_disconnect_{true};
std::unique_ptr<ClientProxy> client_ = std::make_unique<ClientProxy>();
ConnectionOptions connection_options_{
.keep_alive_interval_millis = 5000,
@@ -199,9 +213,9 @@ TEST_F(EndpointManagerTest, RegisterEndpointCallsOnConnectionInitiated) {
}
TEST_F(EndpointManagerTest, UnregisterEndpointCallsOnDisconnected) {
auto endpoint_channel = std::make_unique<MockEndpointChannel>();
EXPECT_CALL(*endpoint_channel, Read())
.WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
// auto endpoint_channel = std::make_unique<MockEndpointChannel>();
// EXPECT_CALL(*endpoint_channel, Read())
// .WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
RegisterEndpoint(std::make_unique<MockEndpointChannel>());
// NOTE: disconnect_cb is not called, because we did not reach fully connected
// state. On top of that, UnregisterEndpoint is suppressing this notification.
@@ -211,6 +225,19 @@ TEST_F(EndpointManagerTest, UnregisterEndpointCallsOnDisconnected) {
em_.UnregisterEndpoint(client_.get(), endpoint_id_);
}
TEST_F(EndpointManagerTest,
UnregisterEndpointCallsOnDisconnectedSafeToDisconnect) {
RegisterEndpoint(std::make_unique<MockEndpointChannel>());
// NOTE: disconnect_cb is not called, because we did not reach fully connected
// state. On top of that, UnregisterEndpoint is suppressing this notification.
// (IMO, it should be called as long as any connection callback was called
// before. (in this case initiated_cb is called)).
// Test captures current protocol behavior.
client_->SetRemoteSafeToDisconnectVersion(endpoint_id_, 2);
ecm_.UpdateSafeToDisconnectForEndpoint(endpoint_id_, true);
em_.UnregisterEndpoint(client_.get(), endpoint_id_);
}
TEST_F(EndpointManagerTest, RegisterFrameProcessorWorks) {
auto endpoint_channel = std::make_unique<MockEndpointChannel>();
auto connect_request = std::make_unique<MockFrameProcessor>();
@@ -434,7 +461,8 @@ TEST_F(EndpointManagerTest, DisconnectEndpointDuringDestruction) {
// immediately.
fake_serial_executor->SetRunExecutablesImmediately(
/*run_executables_immediately=*/false);
endpoint_manager->DiscardEndpoint(client_.get(), endpoint_id_);
endpoint_manager->DiscardEndpoint(client_.get(), endpoint_id_,
DisconnectionReason::IO_ERROR);
// Simulate Core destruction of ClientProxy by destroying `client_`.
client_.reset();
connections/implementation/flags/nearby_connections_feature_flags.h
+13 -6
View File
@@ -15,6 +15,7 @@
#ifndef THIRD_PARTY_NEARBY_CONNECTIONS_IMPLEMENTATION_FLAGS_NEARBY_CONNECTIONS_FEATURE_FLAGS_H_
#define THIRD_PARTY_NEARBY_CONNECTIONS_IMPLEMENTATION_FLAGS_NEARBY_CONNECTIONS_FEATURE_FLAGS_H_
#include <cstdint>
#include "absl/strings/string_view.h"
#include "internal/flags/flag.h"
@@ -27,7 +28,6 @@ constexpr absl::string_view kConfigPackage = "nearby";
// The Nearby Connections features.
namespace nearby_connections_feature {
// LINT.IfChanged
// Disable/Enable BLE v2 in Nearby Connections SDK.
constexpr auto kEnableBleV2 =
flags::Flag<bool>(kConfigPackage, "45401515", false);
@@ -44,11 +44,18 @@ constexpr auto kEnableGattQueryInThread =
constexpr auto kEnablePayloadManagerToSkipChunkUpdate =
flags::Flag<bool>(kConfigPackage, "45415729", false);
// LINT.ThenChange(
// //depot/google3/location/nearby/cpp/sharing/clients/cpp/nearby_sharing_service_adapter_dart.h,
// //depot/google3/location/nearby/cpp/sharing/clients/cpp/nearby_sharing_service_adapter_dart.cc,
// //depot/google3/location/nearby/cpp/sharing/clients/dart/platform/lib/types/models.dart
// )
// Enable/Disable safe-to-disconnect feature.
constexpr auto kEnableSafeToDisconnect =
flags::Flag<bool>(kConfigPackage, "45425789", false);
// When true, allows to enable payload-received-ack protocol.
constexpr auto kEnablePayloadReceivedAck =
flags::Flag<bool>(kConfigPackage, "45425840", false);
// Support 1. safe-to-disconnect 2. reserved 3. auto-reconnect
// 4. auto-resume for dev device 5. payload_ack
constexpr auto kSafeToDisconnectVersion =
flags::Flag<int64_t>(kConfigPackage, "45425841", 2);
} // namespace nearby_connections_feature
} // namespace config_package_nearby
connections/implementation/fuzzers/BUILD
+1
View File
@@ -23,6 +23,7 @@ cc_fuzz_target(
deps = [
"//connections/implementation:internal",
"//internal/platform:base",
"//internal/platform/implementation/g3",
"//security/fuzzing/blaze:default_init_google_for_cc_fuzz_target",
],
)
connections/implementation/offline_frames.cc
+14 -3
View File
@@ -14,14 +14,17 @@
#include "connections/implementation/offline_frames.h"
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
#include <vector>
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/offline_frames_validator.h"
#include "connections/implementation/proto/offline_wire_formats.pb.h"
#include "connections/status.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/byte_array.h"
namespace nearby {
@@ -165,7 +168,8 @@ ByteArray ForConnectionRequestPresence(
return ToBytes(std::move(frame));
}
ByteArray ForConnectionResponse(std::int32_t status, const OsInfo& os_info) {
ByteArray ForConnectionResponse(
std::int32_t status, const OsInfo& os_info) {
OfflineFrame frame;
frame.set_version(OfflineFrame::V1);
@@ -181,6 +185,10 @@ ByteArray ForConnectionResponse(std::int32_t status, const OsInfo& os_info) {
? ConnectionResponseFrame::ACCEPT
: ConnectionResponseFrame::REJECT);
*sub_frame->mutable_os_info() = os_info;
sub_frame->set_safe_to_disconnect_version(
NearbyFlags::GetInstance().GetInt64Flag(
config_package_nearby::nearby_connections_feature::
kSafeToDisconnectVersion));
return ToBytes(std::move(frame));
}
@@ -445,13 +453,16 @@ ByteArray ForKeepAlive() {
return ToBytes(std::move(frame));
}
ByteArray ForDisconnection() {
ByteArray ForDisconnection(bool request_safe_to_disconnect,
bool ack_safe_to_disconnect) {
OfflineFrame frame;
frame.set_version(OfflineFrame::V1);
auto* v1_frame = frame.mutable_v1();
v1_frame->set_type(V1Frame::DISCONNECTION);
v1_frame->mutable_disconnection();
auto* disconnection = v1_frame->mutable_disconnection();
disconnection->set_request_safe_to_disconnect(request_safe_to_disconnect);
disconnection->set_ack_safe_to_disconnect(ack_safe_to_disconnect);
return ToBytes(std::move(frame));
}
connections/implementation/offline_frames.h
+2 -2
View File
@@ -98,8 +98,8 @@ ByteArray ForBwuLastWrite();
ByteArray ForBwuSafeToClose();
ByteArray ForKeepAlive();
ByteArray ForDisconnection();
ByteArray ForDisconnection(bool request_safe_to_disconnect,
bool ack_safe_to_disconnect);
UpgradePathInfo::Medium MediumToUpgradePathInfoMedium(Medium medium);
Medium UpgradePathInfoMediumToMedium(UpgradePathInfo::Medium medium);
connections/implementation/offline_frames_test.cc
+20
View File
@@ -269,6 +269,7 @@ TEST(OfflineFramesTest, CanGenerateConnectionResponse) {
status: 1
response: REJECT
os_info { type: LINUX }
safe_to_disconnect_version: 2
>
>)pb";
@@ -538,6 +539,25 @@ TEST(OfflineFramesTest, CanGenerateKeepAlive) {
EXPECT_THAT(message, EqualsProto(kExpected));
}
TEST(OfflineFramesTest, CanGenerateDisconnection) {
constexpr absl::string_view kExpected =
R"pb(
version: V1
v1: <
type: DISCONNECTION
disconnection: <
request_safe_to_disconnect: true
ack_safe_to_disconnect: true
>
>)pb";
ByteArray bytes = ForDisconnection(/* request_safe_to_disconnect */ true,
/* ack_safe_to_disconnect */ true);
auto response = FromBytes(bytes);
ASSERT_TRUE(response.ok());
OfflineFrame message = response.result();
EXPECT_THAT(message, EqualsProto(kExpected));
}
} // namespace
} // namespace parser
} // namespace connections
connections/implementation/offline_service_controller_test.cc
+3
View File
@@ -94,6 +94,9 @@ class OfflineServiceControllerTest
void SetUp() override {
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableBleV2, true);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::
kEnableSafeToDisconnect, false);
}
bool SetupConnection(OfflineSimulationUser& user_a,
OfflineSimulationUser& user_b) {
connections/implementation/payload_manager.cc
+175 -15
View File
@@ -15,7 +15,7 @@
#include "connections/implementation/payload_manager.h"
#include <algorithm>
#include <functional>
#include <cstdint>
#include <limits>
#include <memory>
#include <string>
@@ -24,24 +24,30 @@
#include "absl/functional/any_invocable.h"
#include "absl/functional/bind_front.h"
#include "absl/memory/memory.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/str_format.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/throughput_recorder.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/internal_payload_factory.h"
#include "connections/payload_type.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex_lock.h"
#include "internal/platform/single_thread_executor.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
using ::location::nearby::connections::OfflineFrame;
using ::location::nearby::connections::V1Frame;
using ::location::nearby::proto::connections::PayloadStatus;
using ::nearby::analytics::PacketMetaData;
using ::nearby::analytics::ThroughputRecorderContainer;
using ::nearby::connections::PayloadDirection;
@@ -158,7 +164,7 @@ bool PayloadManager::SendPayloadLoop(
location::nearby::proto::connections::
PayloadStatus::ENDPOINT_IO_ERROR);
}
bool is_last_chunk = IsLastChunk(payload_chunk);
// Check whether at least one endpoint succeeded -- if they all failed,
// we'll just go right back to the top of the loop and break out when
// availableEndpointIds is re-synced and found to be empty at that point.
@@ -166,6 +172,12 @@ bool PayloadManager::SendPayloadLoop(
for (const auto& endpoint_id : available_endpoint_ids) {
if (std::find(failed_endpoint_ids.begin(), failed_endpoint_ids.end(),
endpoint_id) == failed_endpoint_ids.end()) {
if (!WaitForReceivedAck(client, endpoint_id, pending_payload,
payload_header, next_chunk_offset,
is_last_chunk)) {
continue;
}
HandleSuccessfulOutgoingChunk(
client, endpoint_id, payload_header, payload_chunk.flags(),
payload_chunk.offset(), payload_chunk.body().size());
@@ -524,7 +536,8 @@ void PayloadManager::OnIncomingFrame(
void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
const std::string& service_id,
const std::string& endpoint_id,
CountDownLatch barrier) {
CountDownLatch barrier,
DisconnectionReason reason) {
if (shutdown_.Get()) {
barrier.CountDown();
return;
@@ -532,7 +545,7 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
RunOnStatusUpdateThread(
"payload-manager-on-disconnect",
[this, client, endpoint_id,
barrier]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() mutable {
barrier, reason]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() mutable {
// Iterate through all our payloads and look for payloads associated
// with this endpoint.
MutexLock lock(&mutex_);
@@ -561,14 +574,27 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
// Send a client notification of a payload transfer failure.
client->OnPayloadProgress(endpoint_id, update);
PayloadStatus payload_status;
switch (reason) {
case DisconnectionReason::LOCAL_DISCONNECTION:
payload_status = PayloadStatus::LOCAL_CLIENT_DISCONNECTION;
break;
case DisconnectionReason::REMOTE_DISCONNECTION:
payload_status = PayloadStatus::REMOTE_CLIENT_DISCONNECTION;
break;
case DisconnectionReason::IO_ERROR:
default:
payload_status = PayloadStatus::ENDPOINT_IO_ERROR;
break;
}
if (pending_payload->IsIncoming()) {
client->GetAnalyticsRecorder().OnIncomingPayloadDone(
endpoint_id, pending_payload->GetId(),
location::nearby::proto::connections::ENDPOINT_IO_ERROR);
endpoint_id, pending_payload->GetId(), payload_status);
} else {
client->GetAnalyticsRecorder().OnOutgoingPayloadDone(
endpoint_id, pending_payload->GetId(),
location::nearby::proto::connections::ENDPOINT_IO_ERROR);
endpoint_id, pending_payload->GetId(), payload_status);
}
});
@@ -803,6 +829,99 @@ void PayloadManager::SendControlMessage(
endpoint_ids);
}
void PayloadManager::SendPayloadReceivedAck(
ClientProxy* client, PendingPayload& pending_payload,
const std::string& endpoint_id,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t chunk_size, bool is_last_chunk) {
if (!is_last_chunk ||
!IsPayloadReceivedAckEnabled(client, endpoint_id, pending_payload)) {
return;
}
// Send the PAYLOAD_RECEIVED_ACK to the remote endpoint for the sender asap.
NEARBY_LOGS(INFO)
<< "[PAYLOAD_RECEIVED_ACK] isLastChunk, receiver send ack to "
<< endpoint_id;
SendControlMessage(
{endpoint_id}, payload_header, chunk_size,
PayloadTransferFrame::ControlMessage::PAYLOAD_RECEIVED_ACK);
}
bool PayloadManager::WaitForReceivedAck(
ClientProxy* client, const std::string& endpoint_id,
PendingPayload& pending_payload,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t payload_chunk_offset, bool is_last_chunk) {
if (!is_last_chunk ||
!IsPayloadReceivedAckEnabled(client, endpoint_id, pending_payload)) {
return true;
}
NEARBY_LOGS(INFO) << "[safe-to-disconnect] Last Chunk, sender wait for "
"PAYLOAD_RECEIVED_ACK frame from: "
<< endpoint_id;
while (true) {
PendingPayloadHandle latest_pending_payload =
GetPayload(payload_header.id());
// Make sure we're still tracking this payload and its associated endpoint.
if (!latest_pending_payload) {
return false;
}
auto* endpoint_info = latest_pending_payload->GetEndpoint(endpoint_id);
if (endpoint_info == nullptr) {
return false;
}
// Local payload cancellation
if (latest_pending_payload->IsLocallyCanceled()) {
HandleFinishedOutgoingPayload(client, {endpoint_id}, payload_header,
payload_chunk_offset,
location::nearby::proto::connections::
PayloadStatus::LOCAL_CANCELLATION);
return false;
}
// Remote payload cancellation, etc
if (!endpoint_info->IsEndpointAvailable(client,
endpoint_info->status.Get())) {
HandleFinishedOutgoingPayload(
client, {endpoint_id}, payload_header, payload_chunk_offset,
EndpointInfoStatusToPayloadStatus(endpoint_info->status.Get()));
return false;
}
{
MutexLock lock(&endpoint_info->payload_received_ack_mutex);
if (endpoint_info->is_payload_received_ack) {
endpoint_info->is_payload_received_ack = false;
return true;
}
Exception wait_exception = endpoint_info->payload_received_ack_cond.Wait(
FeatureFlags::GetInstance()
.GetFlags()
.wait_payload_received_ack_millis);
endpoint_info->is_payload_received_ack = false;
if (!wait_exception.Ok()) {
return false;
}
return true;
}
}
return true;
}
bool PayloadManager::IsPayloadReceivedAckEnabled(
ClientProxy* client, const std::string& endpoint_id,
PendingPayload& pending_payload) {
return NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::
kEnablePayloadReceivedAck) &&
client->IsPayloadReceivedAckEnabled(endpoint_id) &&
(pending_payload.GetInternalPayload()->GetType() !=
nearby::connections::PayloadTransferFrame::PayloadTransferFrame::
PayloadHeader::BYTES);
}
void PayloadManager::HandleFinishedOutgoingPayload(
ClientProxy* client, const EndpointIds& finished_endpoint_ids,
const PayloadTransferFrame::PayloadHeader& payload_header,
@@ -833,7 +952,8 @@ void PayloadManager::HandleFinishedOutgoingPayload(
// Unregister these endpoints, since we had an IO error on the physical
// connection.
for (const auto& endpoint_id : finished_endpoint_ids) {
endpoint_manager_->DiscardEndpoint(client, endpoint_id);
endpoint_manager_->DiscardEndpoint(client, endpoint_id,
DisconnectionReason::IO_ERROR);
}
break;
case location::nearby::proto::connections::PayloadStatus::REMOTE_ERROR:
@@ -1124,6 +1244,7 @@ void PayloadManager::ProcessDataPacket(
// Save size of packet before we move it.
std::int64_t payload_body_size = payload_chunk.body().size();
packet_meta_data.StartFileIo();
if (pending_payload->GetInternalPayload()
->AttachNextChunk(ByteArray(std::move(*payload_chunk.mutable_body())))
@@ -1137,6 +1258,11 @@ void PayloadManager::ProcessDataPacket(
return;
}
packet_meta_data.StopFileIo();
bool is_last_chunk = (payload_chunk.flags() &
PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0;
SendPayloadReceivedAck(
to_client, *pending_payload, from_endpoint_id, payload_header,
payload_chunk.offset() + payload_body_size, is_last_chunk);
HandleSuccessfulIncomingChunk(to_client, from_endpoint_id, payload_header,
payload_chunk.flags(), payload_chunk.offset(),
@@ -1145,8 +1271,6 @@ void PayloadManager::ProcessDataPacket(
ThroughputRecorderContainer::GetInstance()
.GetTPRecorder(payload_header.id(), PayloadDirection::INCOMING_PAYLOAD)
->OnFrameReceived(medium, packet_meta_data);
bool is_last_chunk = (payload_chunk.flags() &
PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0;
if (is_last_chunk) {
ThroughputRecorderContainer::GetInstance()
.GetTPRecorder(payload_header.id(), PayloadDirection::INCOMING_PAYLOAD)
@@ -1206,6 +1330,17 @@ void PayloadManager::ProcessControlPacket(
control_message);
}
break;
case PayloadTransferFrame::ControlMessage::PAYLOAD_RECEIVED_ACK:
if (!pending_payload->IsIncoming() &&
IsPayloadReceivedAckEnabled(to_client, from_endpoint_id,
*pending_payload)) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect]Sender received "
"PAYLOAD_RECEIVED_ACK frame with id:"
<< pending_payload->GetInternalPayload()->GetId()
<< " from endpoint_id=" << from_endpoint_id;
pending_payload->MarkReceivedAckFromEndpoint(from_endpoint_id);
}
break;
default:
NEARBY_LOGS(INFO) << "Unhandled control message "
<< control_message.event() << " for payload_id="
@@ -1290,8 +1425,26 @@ void PayloadManager::EndpointInfo::SetStatusFromControlMessage(
<< " based on OOB ControlMessage";
}
//////////////////////////////// PendingPayload
///////////////////////////////////
void PayloadManager::EndpointInfo::MarkReceivedAckFromEndpoint() {
MutexLock lock(&payload_received_ack_mutex);
is_payload_received_ack = true;
payload_received_ack_cond.Notify();
}
bool PayloadManager::EndpointInfo::IsEndpointAvailable(
ClientProxy* clientProxy, EndpointInfo::Status status) {
// Pending endpointIds would be removed from the payload after
// onPayloadTransferUpdate, but there is the racing problem that gets the
// available endpoints before update. Here force to remove those endpoints
// (b/227419433).
bool is_pending_endpoint = false;
if (clientProxy->HasPendingConnectionToEndpoint(id)) {
is_pending_endpoint = true;
}
return (status == EndpointInfo::Status::kAvailable) && !is_pending_endpoint;
}
//////////////////////////////// PendingPayload ////////////////////////////////
PayloadManager::PendingPayload::PendingPayload(
std::unique_ptr<InternalPayload> internal_payload,
@@ -1305,7 +1458,7 @@ PayloadManager::PendingPayload::PendingPayload(
// failures. Any of these situations will cause endpoint to be marked as
// unavailable.
for (const auto& id : endpoint_ids) {
EndpointInfo endpoint_info{};
EndpointInfo endpoint_info;
endpoint_info.id = id;
endpoint_info.status.Set(EndpointInfo::Status::kAvailable);
@@ -1329,6 +1482,13 @@ void PayloadManager::PendingPayload::MarkLocallyCanceled() {
is_locally_canceled_.Set(true);
}
void PayloadManager::PendingPayload::MarkReceivedAckFromEndpoint(
const std::string& from_endpoint_id) {
auto info = GetEndpoint(from_endpoint_id);
if (!info) return;
info->MarkReceivedAckFromEndpoint();
}
bool PayloadManager::PendingPayload::IsIncoming() const { return is_incoming_; }
std::vector<const PayloadManager::EndpointInfo*>
connections/implementation/payload_manager.h
+30 -1
View File
@@ -35,6 +35,7 @@
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/atomic_reference.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/condition_variable.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/mutex.h"
@@ -70,7 +71,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
// @EndpointManagerThread
void OnEndpointDisconnect(ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id,
CountDownLatch barrier) override;
CountDownLatch barrier,
DisconnectionReason reason) override;
void DisconnectFromEndpointManager();
@@ -92,10 +94,17 @@ class PayloadManager : public EndpointManager::FrameProcessor {
static Status ControlMessageEventToEndpointInfoStatus(
PayloadTransferFrame::ControlMessage::EventType event);
void MarkReceivedAckFromEndpoint();
bool IsEndpointAvailable(ClientProxy* clientProxy,
EndpointInfo::Status status);
std::string id;
AtomicReference<Status> status{Status::kUnknown};
std::int64_t offset = 0;
mutable Mutex payload_received_ack_mutex;
ConditionVariable payload_received_ack_cond{&payload_received_ack_mutex};
bool is_payload_received_ack ABSL_GUARDED_BY(payload_received_ack_mutex) =
false;
};
// Tracks state for an InternalPayload and the endpoints associated with it.
@@ -121,6 +130,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
bool IsLocallyCanceled() const;
void MarkLocallyCanceled();
void MarkReceivedAckFromEndpoint(const std::string& from_endpoint_id);
bool IsIncoming() const;
// Gets the EndpointInfo objects for the endpoints (still) associated with
@@ -289,6 +299,10 @@ class PayloadManager : public EndpointManager::FrameProcessor {
PayloadTransferFrame::PayloadChunk CreatePayloadChunk(std::int64_t offset,
ByteArray body);
bool IsLastChunk(PayloadTransferFrame::PayloadChunk payload_chunk) {
return ((payload_chunk.flags() &
PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0);
}
PendingPayloadHandle CreateIncomingPayload(const PayloadTransferFrame& frame,
const std::string& endpoint_id)
@@ -315,6 +329,21 @@ class PayloadManager : public EndpointManager::FrameProcessor {
std::int64_t num_bytes_successfully_transferred,
PayloadTransferFrame::ControlMessage::EventType event_type);
void SendPayloadReceivedAck(
ClientProxy* client, PendingPayload& pending_payload,
const std::string& endpoint_id,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t chunk_size, bool is_last_chunk);
bool WaitForReceivedAck(
ClientProxy* client, const std::string& endpoint_id,
PendingPayload& pending_payload,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t payload_chunk_offset, bool is_last_chunk);
bool IsPayloadReceivedAckEnabled(ClientProxy* client,
const std::string& endpoint_id,
PendingPayload& pending_payload);
// Handles a finished outgoing payload for the given endpointIds. All
// statuses except for SUCCESS are handled here.
void HandleFinishedOutgoingPayload(
connections/implementation/simulation_user.h
+25 -1
View File
@@ -15,6 +15,7 @@
#ifndef CORE_INTERNAL_SIMULATION_USER_H_
#define CORE_INTERNAL_SIMULATION_USER_H_
#include <cstdint>
#include <string>
#include "gtest/gtest.h"
@@ -22,13 +23,15 @@
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/endpoint_manager.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/injected_bluetooth_device_store.h"
#include "connections/implementation/payload_manager.h"
#include "connections/implementation/pcp_manager.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/condition_variable.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/future.h"
#include "internal/platform/medium_environment.h"
// Test-only class to help run end-to-end simulations for nearby connections
// protocol.
@@ -39,6 +42,26 @@
namespace nearby {
namespace connections {
class SetSafeToDisconnect {
public:
explicit SetSafeToDisconnect(bool safe_to_disconnect,
bool payload_received_ack,
std::int32_t safe_to_disconnect_version) {
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::
kEnableSafeToDisconnect,
safe_to_disconnect);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::
kEnablePayloadReceivedAck,
payload_received_ack);
NearbyFlags::GetInstance().OverrideInt64FlagValue(
config_package_nearby::nearby_connections_feature::
kSafeToDisconnectVersion,
safe_to_disconnect_version);
}
};
class SimulationUser {
public:
struct DiscoveredInfo {
@@ -176,6 +199,7 @@ class SimulationUser {
AdvertisingOptions advertising_options_;
ConnectionOptions connection_options_;
DiscoveryOptions discovery_options_;
SetSafeToDisconnect set_safe_to_disconnect_{true, true, 2};
ClientProxy client_;
EndpointChannelManager ecm_;
EndpointManager em_{&ecm_};
internal/platform/feature_flags.h
+14
View File
@@ -15,6 +15,8 @@
#ifndef PLATFORM_BASE_FEATURE_FLAGS_H_
#define PLATFORM_BASE_FEATURE_FLAGS_H_
#include <cstdint>
#include "absl/synchronization/mutex.h"
#include "absl/time/time.h"
@@ -60,6 +62,18 @@ class FeatureFlags {
// requested service id before attempting to connect over rfcomm. SDP fails
// on Windows when connecting to FP service id but the rfcomm is successful.
bool skip_service_discovery_before_connecting_to_rfcomm = false;
std::int32_t min_nc_version_supports_safe_to_disconnect = 1;
// Android code won't be able to launch "payload_received_ack" feature for
// in near future, so change "payload_received_ack" version from "2" to "5"
// after auto-reconnect and auto-resume.
std::int32_t min_nc_version_supports_payload_received_ack = 5;
// If the other part doesn't ack the safe_to_disconnect request, the
// initiator will end the connection in 30s.
absl::Duration safe_to_disconnect_ack_delay_millis =
absl::Milliseconds(30000);
// If the receiver doesn't ack with payload_received_ack frame in 1s, the
// sender will timeout the waiting.
absl::Duration wait_payload_received_ack_millis = absl::Milliseconds(1000);
};
static const FeatureFlags& GetInstance() {