diff --git a/connections/implementation/BUILD b/connections/implementation/BUILD index 861b2d83..0121567f 100644 --- a/connections/implementation/BUILD +++ b/connections/implementation/BUILD @@ -262,6 +262,7 @@ cc_library( "//internal/platform/implementation:wifi_utils", "//proto:connections_enums_cc_proto", "@com_google_absl//absl/base:core_headers", + "@com_google_absl//absl/cleanup", "@com_google_absl//absl/container:btree", "@com_google_absl//absl/container:flat_hash_map", "@com_google_absl//absl/container:flat_hash_set", diff --git a/connections/implementation/base_pcp_handler.cc b/connections/implementation/base_pcp_handler.cc index 40eda845..d2f993be 100644 --- a/connections/implementation/base_pcp_handler.cc +++ b/connections/implementation/base_pcp_handler.cc @@ -25,6 +25,7 @@ #include "securegcm/ukey2_handshake.h" #include "absl/base/thread_annotations.h" +#include "absl/cleanup/cleanup.h" #include "absl/container/btree_map.h" #include "absl/container/flat_hash_set.h" #include "absl/strings/escaping.h" @@ -94,7 +95,6 @@ using ::location::nearby::connections::ConnectionResponseFrame; using ::location::nearby::connections::ConnectionsDevice; using ::location::nearby::connections::MediumMetadata; using ::location::nearby::connections::OfflineFrame; -using ::location::nearby::connections::OsInfo; using ::location::nearby::connections::PresenceDevice; using ::location::nearby::connections::V1Frame; using ::location::nearby::proto::connections::OperationResultCode; @@ -576,15 +576,16 @@ Status BasePcpHandler::WaitForResult(const std::string& method_name, return result.result(); } -void BasePcpHandler::RunOnPcpHandlerThread(const std::string& name, +bool BasePcpHandler::RunOnPcpHandlerThread(const std::string& name, Runnable runnable) { if (closed_.Get()) { LOG(WARNING) << "Skip to run PCP Handler task " << name << " due to PCP Handler is closed"; - return; + return false; } serial_executor_.Execute(name, std::move(runnable)); + return true; } EncryptionRunner::ResultListener BasePcpHandler::GetResultListener( @@ -1681,9 +1682,10 @@ void BasePcpHandler::OnIncomingFrame( OfflineFrame& frame, const std::string& endpoint_id, ClientProxy* client, location::nearby::proto::connections::Medium medium) { CountDownLatch latch(1); - RunOnPcpHandlerThread( + bool scheduled = RunOnPcpHandlerThread( "incoming-frame", [this, client, endpoint_id, frame, &latch]() RUN_ON_PCP_HANDLER_THREAD() { + absl::Cleanup release_caller = [&latch] { latch.CountDown(); }; LOG(INFO) << "OnConnectionResponse: endpoint_id=" << endpoint_id; if (client->HasRemoteEndpointResponded(endpoint_id)) { @@ -1741,9 +1743,10 @@ void BasePcpHandler::OnIncomingFrame( client->SetRemoteDeviceName( endpoint_id, connection_response.wifi_direct_device_name()); } - latch.CountDown(); }); - WaitForLatch("OnIncomingFrame()", &latch); + if (scheduled) { + WaitForLatch("OnIncomingFrame()", &latch); + } } void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client, diff --git a/connections/implementation/base_pcp_handler.h b/connections/implementation/base_pcp_handler.h index 49cf0b70..e5e6a1eb 100644 --- a/connections/implementation/base_pcp_handler.h +++ b/connections/implementation/base_pcp_handler.h @@ -278,7 +278,7 @@ class BasePcpHandler : public PcpHandler, }; void Shutdown(); - void RunOnPcpHandlerThread(const std::string& name, Runnable runnable); + bool RunOnPcpHandlerThread(const std::string& name, Runnable runnable); BluetoothDevice GetRemoteBluetoothDevice( MacAddress remote_bluetooth_mac_address); diff --git a/connections/implementation/base_pcp_handler_test.cc b/connections/implementation/base_pcp_handler_test.cc index be2af7bf..59e809d9 100644 --- a/connections/implementation/base_pcp_handler_test.cc +++ b/connections/implementation/base_pcp_handler_test.cc @@ -455,8 +455,7 @@ class BasePcpHandlerTest void TearDown() override { env_.Stop(); } std::unique_ptr CreateAnalyticsRecorder() { - auto recorder = - std::make_unique(); + auto recorder = std::make_unique(); mock_analytics_recorder_ptr_ = recorder.get(); return recorder; } @@ -576,10 +575,9 @@ class BasePcpHandlerTest [channel = channel_a.get()]() { return channel->DoRead(); }); EXPECT_CALL(*channel_a, Write(_)) .WillOnce(Return(Exception{Exception::kSuccess})) - .WillRepeatedly( - [channel = channel_a.get()](absl::string_view data) { - return channel->DoWrite(data); - }); + .WillRepeatedly([channel = channel_a.get()](absl::string_view data) { + return channel->DoWrite(data); + }); EXPECT_CALL(*channel_a, GetMedium).WillRepeatedly(Return(medium)); EXPECT_CALL(*channel_a, GetLastReadTimestamp) .WillRepeatedly(Return(absl::Now())); @@ -588,10 +586,9 @@ class BasePcpHandlerTest .WillRepeatedly( [channel = channel_b.get()]() { return channel->DoRead(); }); EXPECT_CALL(*channel_b, Write(_)) - .WillRepeatedly( - [channel = channel_b.get()](absl::string_view data) { - return channel->DoWrite(data); - }); + .WillRepeatedly([channel = channel_b.get()](absl::string_view data) { + return channel->DoWrite(data); + }); EXPECT_CALL(*channel_b, GetMedium).WillRepeatedly(Return(medium)); EXPECT_CALL(*channel_b, GetLastReadTimestamp) .WillRepeatedly(Return(absl::Now())); @@ -628,10 +625,9 @@ class BasePcpHandlerTest .WillRepeatedly( [channel = channel_b.get()]() { return channel->DoRead(); }); EXPECT_CALL(*channel_b, Write(_)) - .WillRepeatedly( - [channel = channel_b.get()](absl::string_view data) { - return channel->DoWrite(data); - }); + .WillRepeatedly([channel = channel_b.get()](absl::string_view data) { + return channel->DoWrite(data); + }); EXPECT_CALL(*channel_b, GetMedium).WillRepeatedly(Return(medium)); EXPECT_CALL(*channel_b, GetLastReadTimestamp) .WillRepeatedly(Return(absl::Now())); @@ -751,16 +747,15 @@ class BasePcpHandlerTest auto allowed_mediums = pcp_handler->GetDiscoveryMediums(client); EXPECT_CALL(*pcp_handler, ConnectImpl) - .WillRepeatedly( - [&channel_a, connect_medium]( - ClientProxy* client, - MockPcpHandler::DiscoveredEndpoint* endpoint) { - return MockPcpHandler::ConnectImplResult{ - .medium = connect_medium, - .status = {Status::kSuccess}, - .endpoint_channel = std::move(channel_a), - }; - }); + .WillRepeatedly([&channel_a, connect_medium]( + ClientProxy* client, + MockPcpHandler::DiscoveredEndpoint* endpoint) { + return MockPcpHandler::ConnectImplResult{ + .medium = connect_medium, + .status = {Status::kSuccess}, + .endpoint_channel = std::move(channel_a), + }; + }); for (const auto& discovered_medium : allowed_mediums) { pcp_handler->OnEndpointFound( @@ -1609,8 +1604,8 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) { Status{Status::kSuccess}); LOG(INFO) << "Simulating remote accept: id=" << endpoint_id; OsInfo os_info; - auto frame = parser::FromBytes(parser::ForConnectionResponse( - Status::kSuccess, os_info, "device_name")); + auto frame = parser::FromBytes( + parser::ForConnectionResponse(Status::kSuccess, os_info, "device_name")); EXPECT_CALL(mock_connection_listener_.bandwidth_changed_cb, Call).Times(1); pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, client_.get(), connect_medium); @@ -1621,6 +1616,48 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) { env_.Stop(); } +TEST_P(BasePcpHandlerTest, OnIncomingFrameDuplicateFrameDoesNotDeadlock) { + env_.Start(); + std::string endpoint_id{"1234"}; + Mediums m; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + BwuManager bwu(m, em, ecm, {}, {}); + MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); + StartDiscovery(client_.get(), &pcp_handler); + auto mediums = pcp_handler.GetDiscoveryMediums(client_.get()); + auto connect_medium = mediums[mediums.size() - 1]; + auto channel_pair = SetupConnection(connect_medium); + auto& channel_a = channel_pair.first; + std::shared_ptr channel_b = + std::move(channel_pair.second); + EXPECT_CALL(*channel_a, CloseImpl).Times(1); + EXPECT_CALL(*channel_b, CloseImpl).Times(1); + RequestConnection(endpoint_id, std::move(channel_a), channel_b, client_.get(), + &pcp_handler, connect_medium); + LOG(INFO) << "Attempting to accept connection: id=" << endpoint_id; + EXPECT_CALL(mock_connection_listener_.accepted_cb, Call).Times(1); + EXPECT_CALL(mock_connection_listener_.disconnected_cb, Call) + .Times(AtLeast(0)); + EXPECT_EQ(pcp_handler.AcceptConnection(client_.get(), endpoint_id, {}), + Status{Status::kSuccess}); + LOG(INFO) << "Simulating remote accept: id=" << endpoint_id; + OsInfo os_info; + auto frame = parser::FromBytes( + parser::ForConnectionResponse(Status::kSuccess, os_info, "device_name")); + EXPECT_CALL(mock_connection_listener_.bandwidth_changed_cb, Call).Times(1); + pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, client_.get(), + connect_medium); + LOG(INFO) << "Simulating duplicate remote accept: id=" << endpoint_id; + pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, client_.get(), + connect_medium); + LOG(INFO) << "Closing connection: id=" << endpoint_id; + channel_b->Close(); + bwu.Shutdown(); + pcp_handler.DisconnectFromEndpointManager(); + env_.Stop(); +} + TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) { env_.Start(); std::atomic_int destroyed_flag = 0; @@ -1798,8 +1835,8 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) { EXPECT_CALL(pcp_handler, InjectEndpointImpl(client_.get(), service_id, _)) .WillOnce([&pcp_handler, &endpoint_id]( - ClientProxy* client, const std::string& service_id, - const OutOfBandConnectionMetadata& metadata) { + ClientProxy* client, const std::string& service_id, + const OutOfBandConnectionMetadata& metadata) { pcp_handler.OnEndpointFound( client, std::make_shared(MockDiscoveredEndpoint{ @@ -1862,8 +1899,8 @@ TEST_F(BasePcpHandlerTest, ::testing::InSequence seq; EXPECT_CALL(mock_discovery_listener_.endpoint_found_cb, Call) .WillOnce([id = endpoint_id](const std::string& endpoint_id, - const ByteArray& endpoint_info, - const std::string& service_id) { + const ByteArray& endpoint_info, + const std::string& service_id) { EXPECT_EQ(endpoint_id, id); EXPECT_EQ(endpoint_info, ByteArray{"ABCD"}); }); @@ -1875,8 +1912,8 @@ TEST_F(BasePcpHandlerTest, EXPECT_CALL(mock_discovery_listener_.endpoint_found_cb, Call) .WillOnce([id = endpoint_id](const std::string& endpoint_id, - const ByteArray& endpoint_info, - const std::string& service_id) { + const ByteArray& endpoint_info, + const std::string& service_id) { EXPECT_EQ(endpoint_id, id); EXPECT_EQ(endpoint_info, ByteArray{"ABCDEF"}); }); @@ -1975,8 +2012,8 @@ TEST_F(BasePcpHandlerTest, TestStartStopEndpointLostAlarm) { EXPECT_CALL(pcp_handler, InjectEndpointImpl) .WillOnce([&pcp_handler, &endpoint_id]( - ClientProxy* client, const std::string& service_id, - const OutOfBandConnectionMetadata& metadata) { + ClientProxy* client, const std::string& service_id, + const OutOfBandConnectionMetadata& metadata) { pcp_handler.OnEndpointFound( client, std::make_shared(MockDiscoveredEndpoint{ @@ -2038,8 +2075,8 @@ TEST_F(BasePcpHandlerTest, TestStartEndpointLostByMediumAlarms) { EXPECT_CALL(pcp_handler, InjectEndpointImpl) .WillOnce([&pcp_handler, &endpoint_id]( - ClientProxy* client, const std::string& service_id, - const OutOfBandConnectionMetadata& metadata) { + ClientProxy* client, const std::string& service_id, + const OutOfBandConnectionMetadata& metadata) { pcp_handler.OnEndpointFound( client, std::make_shared(MockDiscoveredEndpoint{ @@ -2103,31 +2140,30 @@ TEST_F(BasePcpHandlerTest, TestEndpointFoundStopsAlarm) { bool first_call = true; EXPECT_CALL(pcp_handler, InjectEndpointImpl) .Times(2) - .WillRepeatedly( - [&pcp_handler, &endpoint_id, &first_call]( - ClientProxy* client, const std::string& service_id, - const OutOfBandConnectionMetadata& metadata) { - ByteArray endpoint_info; - if (first_call) { - endpoint_info = ByteArray("ABCD"); - } else { - endpoint_info = ByteArray("ABCDE"); - } - first_call = false; - pcp_handler.OnEndpointFound( - client, - std::make_shared(MockDiscoveredEndpoint{ - { - endpoint_id, - endpoint_info, - service_id, - Medium::BLUETOOTH, - WebRtcState::kUndefined, - }, - MockContext{nullptr}, - })); - return Status{Status::kSuccess}; - }); + .WillRepeatedly([&pcp_handler, &endpoint_id, &first_call]( + ClientProxy* client, const std::string& service_id, + const OutOfBandConnectionMetadata& metadata) { + ByteArray endpoint_info; + if (first_call) { + endpoint_info = ByteArray("ABCD"); + } else { + endpoint_info = ByteArray("ABCDE"); + } + first_call = false; + pcp_handler.OnEndpointFound( + client, + std::make_shared(MockDiscoveredEndpoint{ + { + endpoint_id, + endpoint_info, + service_id, + Medium::BLUETOOTH, + WebRtcState::kUndefined, + }, + MockContext{nullptr}, + })); + return Status{Status::kSuccess}; + }); pcp_handler.InjectEndpoint( client_.get(), service_id, OutOfBandConnectionMetadata{ diff --git a/connections/implementation/endpoint_manager.cc b/connections/implementation/endpoint_manager.cc index e4a214ef..29e3659a 100644 --- a/connections/implementation/endpoint_manager.cc +++ b/connections/implementation/endpoint_manager.cc @@ -297,8 +297,12 @@ ExceptionOr EndpointManager::HandleData( continue; } - LockedFrameProcessor frame_processor = GetFrameProcessor(frame_type); - if (!frame_processor) { + FrameProcessor* processor = nullptr; + { + LockedFrameProcessor frame_processor = GetFrameProcessor(frame_type); + processor = frame_processor.get(); + } + if (!processor) { // report messages without handlers, except KEEP_ALIVE, which has // no explicit handler. if (frame_type == V1Frame::KEEP_ALIVE) { @@ -333,8 +337,8 @@ ExceptionOr EndpointManager::HandleData( continue; } - frame_processor->OnIncomingFrame(frame, endpoint_id, client, - endpoint_channel->GetMedium()); + processor->OnIncomingFrame(frame, endpoint_id, client, + endpoint_channel->GetMedium()); } }