From 55aedfb235424b68407b7ec18cd935f0e81f66ec Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 25 Aug 2023 09:50:16 -0700 Subject: [PATCH] Implement Safe to Disconnect feature PiperOrigin-RevId: 560119556 --- connections/implementation/BUILD | 4 + .../implementation/base_endpoint_channel.h | 3 +- .../implementation/base_pcp_handler.cc | 56 +++- connections/implementation/base_pcp_handler.h | 7 +- .../implementation/base_pcp_handler_test.cc | 16 +- connections/implementation/bwu_manager.cc | 11 +- connections/implementation/bwu_manager.h | 3 +- .../implementation/bwu_manager_test.cc | 161 ++++++++--- connections/implementation/client_proxy.cc | 49 +++- connections/implementation/client_proxy.h | 17 ++ .../implementation/client_proxy_test.cc | 11 +- .../endpoint_channel_manager.cc | 170 +++++++++++- .../implementation/endpoint_channel_manager.h | 60 +++- .../endpoint_channel_manager_test.cc | 15 +- .../implementation/endpoint_manager.cc | 260 ++++++++++++++---- connections/implementation/endpoint_manager.h | 22 +- .../implementation/endpoint_manager_test.cc | 40 ++- .../flags/nearby_connections_feature_flags.h | 19 +- connections/implementation/fuzzers/BUILD | 1 + connections/implementation/offline_frames.cc | 17 +- connections/implementation/offline_frames.h | 4 +- .../implementation/offline_frames_test.cc | 20 ++ .../offline_service_controller_test.cc | 3 + connections/implementation/payload_manager.cc | 190 ++++++++++++- connections/implementation/payload_manager.h | 31 ++- connections/implementation/simulation_user.h | 26 +- internal/platform/feature_flags.h | 14 + 27 files changed, 1052 insertions(+), 178 deletions(-) diff --git a/connections/implementation/BUILD b/connections/implementation/BUILD index 6c06e435..649b57f2 100644 --- a/connections/implementation/BUILD +++ b/connections/implementation/BUILD @@ -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", diff --git a/connections/implementation/base_endpoint_channel.h b/connections/implementation/base_endpoint_channel.h index 5259748c..889cf4fa 100644 --- a/connections/implementation/base_endpoint_channel.h +++ b/connections/implementation/base_endpoint_channel.h @@ -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" diff --git a/connections/implementation/base_pcp_handler.cc b/connections/implementation/base_pcp_handler.cc index 0e8af5be..085121d8 100644 --- a/connections/implementation/base_pcp_handler.cc +++ b/connections/implementation/base_pcp_handler.cc @@ -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( "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_)); } diff --git a/connections/implementation/base_pcp_handler.h b/connections/implementation/base_pcp_handler.h index 38edf662..07496504 100644 --- a/connections/implementation/base_pcp_handler.h +++ b/connections/implementation/base_pcp_handler.h @@ -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* 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. diff --git a/connections/implementation/base_pcp_handler_test.cc b/connections/implementation/base_pcp_handler_test.cc index ddb38e7e..59cdded1 100644 --- a/connections/implementation/base_pcp_handler_test.cc +++ b/connections/implementation/base_pcp_handler_test.cc @@ -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 { 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(); }; diff --git a/connections/implementation/bwu_manager.cc b/connections/implementation/bwu_manager.cc index 97f597d5..78ab4992 100644 --- a/connections/implementation/bwu_manager.cc +++ b/connections/implementation/bwu_manager.cc @@ -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. diff --git a/connections/implementation/bwu_manager.h b/connections/implementation/bwu_manager.h index 354564e6..c13d3208 100644 --- a/connections/implementation/bwu_manager.h +++ b/connections/implementation/bwu_manager.h @@ -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(); diff --git a/connections/implementation/bwu_manager_test.cc b/connections/implementation/bwu_manager_test.cc index 037dccdd..5a5e2659 100644 --- a/connections/implementation/bwu_manager_test.cc +++ b/connections/implementation/bwu_manager_test.cc @@ -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( 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( 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( 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) { diff --git a/connections/implementation/client_proxy.cc b/connections/implementation/client_proxy.cc index c3a2e6fe..06768e72 100644 --- a/connections/implementation/client_proxy.cc +++ b/connections/implementation/client_proxy.cc @@ -14,6 +14,7 @@ #include "connections/implementation/client_proxy.h" +#include #include #include #include @@ -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 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(); diff --git a/connections/implementation/client_proxy.h b/connections/implementation/client_proxy.h index f19190e4..e39ae058 100644 --- a/connections/implementation/client_proxy.h +++ b/connections/implementation/client_proxy.h @@ -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 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 os_info; + std::int32_t safe_to_disconnect_version; }; using ConnectionPair = std::pair; @@ -427,6 +442,8 @@ class ClientProxy final { NearbyDeviceProvider* external_device_provider_ = nullptr; // For Nearby Connections' own device provider. std::unique_ptr connections_device_provider_; + bool supports_safe_to_disconnect_; + std::int32_t local_safe_to_disconnect_version_; }; } // namespace connections diff --git a/connections/implementation/client_proxy_test.cc b/connections/implementation/client_proxy_test.cc index 011cb8ec..3d53bc87 100644 --- a/connections/implementation/client_proxy_test.cc +++ b/connections/implementation/client_proxy_test.cc @@ -14,6 +14,7 @@ #include "connections/implementation/client_proxy.h" +#include #include #include #include @@ -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, diff --git a/connections/implementation/endpoint_channel_manager.cc b/connections/implementation/endpoint_channel_manager.cc index f8c66e21..cc3b5282 100644 --- a/connections/implementation/endpoint_channel_manager.cc +++ b/connections/implementation/endpoint_channel_manager.cc @@ -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 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; } diff --git a/connections/implementation/endpoint_channel_manager.h b/connections/implementation/endpoint_channel_manager.h index 980c53a7..16980164 100644 --- a/connections/implementation/endpoint_channel_manager.h +++ b/connections/implementation/endpoint_channel_manager.h @@ -17,15 +17,23 @@ #include #include +#include #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 channel; std::shared_ptr 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 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 diff --git a/connections/implementation/endpoint_channel_manager_test.cc b/connections/implementation/endpoint_channel_manager_test.cc index 466f67c0..eb0da0c2 100644 --- a/connections/implementation/endpoint_channel_manager_test.cc +++ b/connections/implementation/endpoint_channel_manager_test.cc @@ -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 diff --git a/connections/implementation/endpoint_manager.cc b/connections/implementation/endpoint_manager.cc index 4a9f5a32..1cccfc41 100644 --- a/connections/implementation/endpoint_manager.cc +++ b/connections/implementation/endpoint_manager.cc @@ -21,8 +21,11 @@ #include #include +#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 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 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 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 EndpointManager::HandleKeepAlive( return ExceptionOr(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 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 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 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 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 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. diff --git a/connections/implementation/endpoint_manager.h b/connections/implementation/endpoint_manager.h index 5bfd0fbf..02306ca2 100644 --- a/connections/implementation/endpoint_manager.h +++ b/connections/implementation/endpoint_manager.h @@ -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 SendTransferFrameBytes( const std::vector& endpoint_ids, diff --git a/connections/implementation/endpoint_manager_test.cc b/connections/implementation/endpoint_manager_test.cc index 49d87f6a..d2047130 100644 --- a/connections/implementation/endpoint_manager_test.cc +++ b/connections/implementation/endpoint_manager_test.cc @@ -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 client_ = std::make_unique(); 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(); - EXPECT_CALL(*endpoint_channel, Read()) - .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); +// auto endpoint_channel = std::make_unique(); +// EXPECT_CALL(*endpoint_channel, Read()) +// .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); RegisterEndpoint(std::make_unique()); // 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()); + // 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(); auto connect_request = std::make_unique(); @@ -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(); diff --git a/connections/implementation/flags/nearby_connections_feature_flags.h b/connections/implementation/flags/nearby_connections_feature_flags.h index 430be69a..d7f6a398 100644 --- a/connections/implementation/flags/nearby_connections_feature_flags.h +++ b/connections/implementation/flags/nearby_connections_feature_flags.h @@ -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 #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(kConfigPackage, "45401515", false); @@ -44,11 +44,18 @@ constexpr auto kEnableGattQueryInThread = constexpr auto kEnablePayloadManagerToSkipChunkUpdate = flags::Flag(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(kConfigPackage, "45425789", false); + +// When true, allows to enable payload-received-ack protocol. +constexpr auto kEnablePayloadReceivedAck = + flags::Flag(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(kConfigPackage, "45425841", 2); } // namespace nearby_connections_feature } // namespace config_package_nearby diff --git a/connections/implementation/fuzzers/BUILD b/connections/implementation/fuzzers/BUILD index 64223d93..54b9df9b 100644 --- a/connections/implementation/fuzzers/BUILD +++ b/connections/implementation/fuzzers/BUILD @@ -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", ], ) diff --git a/connections/implementation/offline_frames.cc b/connections/implementation/offline_frames.cc index fcebbddc..395d2a99 100644 --- a/connections/implementation/offline_frames.cc +++ b/connections/implementation/offline_frames.cc @@ -14,14 +14,17 @@ #include "connections/implementation/offline_frames.h" +#include #include #include #include #include +#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)); } diff --git a/connections/implementation/offline_frames.h b/connections/implementation/offline_frames.h index 6a66d9f7..fb6214ab 100644 --- a/connections/implementation/offline_frames.h +++ b/connections/implementation/offline_frames.h @@ -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); diff --git a/connections/implementation/offline_frames_test.cc b/connections/implementation/offline_frames_test.cc index a1b14320..55d1e8a1 100644 --- a/connections/implementation/offline_frames_test.cc +++ b/connections/implementation/offline_frames_test.cc @@ -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 diff --git a/connections/implementation/offline_service_controller_test.cc b/connections/implementation/offline_service_controller_test.cc index 5b4e0989..5173dc1f 100644 --- a/connections/implementation/offline_service_controller_test.cc +++ b/connections/implementation/offline_service_controller_test.cc @@ -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) { diff --git a/connections/implementation/payload_manager.cc b/connections/implementation/payload_manager.cc index 1eec8436..349b8cb2 100644 --- a/connections/implementation/payload_manager.cc +++ b/connections/implementation/payload_manager.cc @@ -15,7 +15,7 @@ #include "connections/implementation/payload_manager.h" #include -#include +#include #include #include #include @@ -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 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 diff --git a/connections/implementation/payload_manager.h b/connections/implementation/payload_manager.h index eeb46b93..768841d1 100644 --- a/connections/implementation/payload_manager.h +++ b/connections/implementation/payload_manager.h @@ -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::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( diff --git a/connections/implementation/simulation_user.h b/connections/implementation/simulation_user.h index ef68db3d..7a66e4b7 100644 --- a/connections/implementation/simulation_user.h +++ b/connections/implementation/simulation_user.h @@ -15,6 +15,7 @@ #ifndef CORE_INTERNAL_SIMULATION_USER_H_ #define CORE_INTERNAL_SIMULATION_USER_H_ +#include #include #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_}; diff --git a/internal/platform/feature_flags.h b/internal/platform/feature_flags.h index 58f8835d..d4202188 100644 --- a/internal/platform/feature_flags.h +++ b/internal/platform/feature_flags.h @@ -15,6 +15,8 @@ #ifndef PLATFORM_BASE_FEATURE_FLAGS_H_ #define PLATFORM_BASE_FEATURE_FLAGS_H_ +#include + #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() {