From fae44ed114c320f36ca0713bb62c6c650971f741 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 19 Feb 2021 11:22:22 -0800 Subject: [PATCH 1/8] 357261873 Close channels as UNFINISHED on dupe BwuAvailable event. --- cpp/core/internal/bwu_manager.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc index 6cfeab48..01e4c456 100644 --- a/cpp/core/internal/bwu_manager.cc +++ b/cpp/core/internal/bwu_manager.cc @@ -412,7 +412,7 @@ void BwuManager::ProcessBwuPathAvailableEvent( std::shared_ptr previous_endpoint_channel = item.mapped(); if (previous_endpoint_channel) { - previous_endpoint_channel->Close(DisconnectionReason::IO_ERROR); + previous_endpoint_channel->Close(DisconnectionReason::UNFINISHED); } } std::shared_ptr new_channel = @@ -421,7 +421,7 @@ void BwuManager::ProcessBwuPathAvailableEvent( // The upgraded channel never finished upgrading, and therefore is still // paused. new_channel->Resume(); - new_channel->Close(DisconnectionReason::IO_ERROR); + new_channel->Close(DisconnectionReason::UNFINISHED); } return; From 570258dd48fa47edb247cb212cdbdae1c7c8c2b4 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 19 Feb 2021 11:25:19 -0800 Subject: [PATCH 2/8] 358283607 Do not log names of discovered Bluetooth devices. --- cpp/core/internal/bluetooth_device_name.cc | 5 ----- cpp/platform/public/ble.cc | 3 --- 2 files changed, 8 deletions(-) diff --git a/cpp/core/internal/bluetooth_device_name.cc b/cpp/core/internal/bluetooth_device_name.cc index 850c5031..6bd177ba 100644 --- a/cpp/core/internal/bluetooth_device_name.cc +++ b/cpp/core/internal/bluetooth_device_name.cc @@ -48,12 +48,7 @@ BluetoothDeviceName::BluetoothDeviceName( absl::string_view bluetooth_device_name_string) { ByteArray bluetooth_device_name_bytes = Base64Utils::Decode(bluetooth_device_name_string); - if (bluetooth_device_name_bytes.Empty()) { - NEARBY_LOG( - INFO, - "Cannot deserialize BluetoothDeviceName: failed Base64 decoding of %s", - std::string(bluetooth_device_name_string).c_str()); return; } diff --git a/cpp/platform/public/ble.cc b/cpp/platform/public/ble.cc index 2ebe0c95..230554d0 100644 --- a/cpp/platform/public/ble.cc +++ b/cpp/platform/public/ble.cc @@ -39,9 +39,6 @@ bool BleMedium::StartScanning( auto& context = *pair.first->second; if (pair.second) { context.peripheral = BlePeripheral(&peripheral); - NEARBY_LOG(INFO, - "Discovered peripheral '%s'", - peripheral.GetName().c_str()); discovered_peripheral_callback_.peripheral_discovered_cb( context.peripheral, service_id, context.peripheral.GetAdvertisementBytes(service_id), From 81808a2aabb36381d79b8c2ba7748506ffd36b79 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 19 Feb 2021 11:26:17 -0800 Subject: [PATCH 3/8] 358285146 2021-02-18 15:58 -08:00 [Nearby] Continue reading frames during bandwidth upgrade. --- cpp/core/internal/BUILD | 1 + cpp/core/internal/bwu_manager.cc | 20 +++++++++++++------- cpp/core/internal/bwu_manager.h | 2 +- cpp/core/internal/bwu_manager_test.cc | 18 ++++++++++++++++++ cpp/platform/base/feature_flags.h | 1 + 5 files changed, 34 insertions(+), 8 deletions(-) diff --git a/cpp/core/internal/BUILD b/cpp/core/internal/BUILD index f6829072..e593c89f 100644 --- a/cpp/core/internal/BUILD +++ b/cpp/core/internal/BUILD @@ -171,6 +171,7 @@ cc_test( ":internal_test", "//core:core_types", "//core/internal/mediums", + "//core/internal/mediums:utils", "//proto/connections:offline_wire_formats_portable_proto", "//platform/base", "//platform/base:test_util", diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc index 01e4c456..42271740 100644 --- a/cpp/core/internal/bwu_manager.cc +++ b/cpp/core/internal/bwu_manager.cc @@ -191,12 +191,18 @@ void BwuManager::OnIncomingFrame(OfflineFrame& frame, if (parser::GetFrameType(frame) != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION) return; auto bwu_frame = frame.v1().bandwidth_upgrade_negotiation(); - CountDownLatch latch(1); - RunOnBwuManagerThread([this, client, endpoint_id, &bwu_frame, &latch]() { - OnBwuNegotiationFrame(client, bwu_frame, endpoint_id); - latch.CountDown(); - }); - latch.Await(); + if (FeatureFlags::GetInstance().GetFlags().enable_async_bandwidth_upgrade) { + RunOnBwuManagerThread([this, client, endpoint_id, bwu_frame]() { + OnBwuNegotiationFrame(client, bwu_frame, endpoint_id); + }); + } else { + CountDownLatch latch(1); + RunOnBwuManagerThread([this, client, endpoint_id, bwu_frame, &latch]() { + OnBwuNegotiationFrame(client, bwu_frame, endpoint_id); + latch.CountDown(); + }); + latch.Await(); + } } void BwuManager::OnEndpointDisconnect(ClientProxy* client, @@ -260,7 +266,7 @@ void BwuManager::Revert() { } void BwuManager::OnBwuNegotiationFrame(ClientProxy* client, - const BwuNegotiationFrame& frame, + const BwuNegotiationFrame frame, const string& endpoint_id) { NEARBY_LOG(INFO, "OnBwuNegotiationFrame for endpoint %s", endpoint_id.c_str()); diff --git a/cpp/core/internal/bwu_manager.h b/cpp/core/internal/bwu_manager.h index a1f6c480..3895faf8 100644 --- a/cpp/core/internal/bwu_manager.h +++ b/cpp/core/internal/bwu_manager.h @@ -100,7 +100,7 @@ class BwuManager : public EndpointManager::FrameProcessor { // Processes the BwuNegotiationFrames that come over the // EndpointChannel on both initiator and responder side of the upgrade. void OnBwuNegotiationFrame(ClientProxy* client, - const BwuNegotiationFrame& frame, + const BwuNegotiationFrame frame, const string& endpoint_id); // Called to revert any state changed by the Initiator or Responder in the diff --git a/cpp/core/internal/bwu_manager_test.cc b/cpp/core/internal/bwu_manager_test.cc index e3fb271a..231b0cfd 100644 --- a/cpp/core/internal/bwu_manager_test.cc +++ b/cpp/core/internal/bwu_manager_test.cc @@ -6,6 +6,7 @@ #include "core/internal/endpoint_channel_manager.h" #include "core/internal/endpoint_manager.h" #include "core/internal/mediums/mediums.h" +#include "core/internal/mediums/utils.h" #include "platform/public/system_clock.h" #include "gmock/gmock.h" #include "gtest/gtest.h" @@ -42,6 +43,23 @@ TEST(BwuManagerTest, CanInitiateBwu) { bwu_manager.Shutdown(); } +TEST(BwuManagerTest, CanProcessPathAvailableFrame) { + ClientProxy client; + std::string endpoint_id("EP_A"); + Mediums mediums; + EndpointChannelManager ecm; + EndpointManager em{&ecm}; + BwuManager bwu_manager{mediums, em, ecm, {}, {}}; + + LocationHint location_hint = Utils::BuildLocationHint("US"); + ExceptionOr wrapped_frame = parser::FromBytes( + parser::ForBwuWebrtcPathAvailable("my_id", location_hint)); + + bwu_manager.OnIncomingFrame(wrapped_frame.result(), endpoint_id, &client, + Medium::WEB_RTC); + bwu_manager.Shutdown(); +} + } // namespace } // namespace connections } // namespace nearby diff --git a/cpp/platform/base/feature_flags.h b/cpp/platform/base/feature_flags.h index a6c90dc0..e5b58747 100644 --- a/cpp/platform/base/feature_flags.h +++ b/cpp/platform/base/feature_flags.h @@ -21,6 +21,7 @@ class FeatureFlags { // Ignore subsequent BWU Available events when we're still processing the // first one. bool disallow_out_of_order_bwu_avail_event = true; + bool enable_async_bandwidth_upgrade = true; }; static const FeatureFlags& GetInstance() { From de83b242c527c5ac8957a8dd8c75aa2bad3891e7 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 19 Feb 2021 11:28:26 -0800 Subject: [PATCH 4/8] 358287309 Cleanup PeerConnectionObserver so it can't touch a dead ConnectionFlow --- .../mediums/webrtc/connection_flow.cc | 4 ++++ .../webrtc/peer_connection_observer_impl.cc | 23 +++++++++++++++---- .../webrtc/peer_connection_observer_impl.h | 6 +++-- 3 files changed, 27 insertions(+), 6 deletions(-) diff --git a/cpp/core/internal/mediums/webrtc/connection_flow.cc b/cpp/core/internal/mediums/webrtc/connection_flow.cc index 58c70649..995aac9b 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow.cc +++ b/cpp/core/internal/mediums/webrtc/connection_flow.cc @@ -329,11 +329,15 @@ bool ConnectionFlow::TransitionState(State current_state, State new_state) { } bool ConnectionFlow::CloseLocked() { + NEARBY_LOG(INFO, "Closing WebRTC connection."); if (state_ == State::kEnded) { return false; } state_ = State::kEnded; + single_threaded_signaling_offloader_.Shutdown(); + peer_connection_observer_.Shutdown(); + if (peer_connection_) peer_connection_->Close(); data_channel_observer_.reset(); diff --git a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc index 789f1c51..8d715284 100644 --- a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc +++ b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc @@ -14,6 +14,10 @@ PeerConnectionObserverImpl::PeerConnectionObserverImpl( : connection_flow_(connection_flow), local_ice_candidate_listener_(std::move(local_ice_candidate_listener)) {} +PeerConnectionObserverImpl::~PeerConnectionObserverImpl() { + Shutdown(); +} + void PeerConnectionObserverImpl::OnIceCandidate( const webrtc::IceCandidateInterface* candidate) { local_ice_candidate_listener_.local_ice_candidate_found_cb(candidate); @@ -24,8 +28,10 @@ void PeerConnectionObserverImpl::OnSignalingChange( NEARBY_LOG(INFO, "OnSignalingChange: %d", new_state); OffloadFromSignalingThread([this, new_state]() { - if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable) + if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable && + connection_flow_) { connection_flow_->OnSignalingStable(); + } }); } @@ -33,8 +39,10 @@ void PeerConnectionObserverImpl::OnDataChannel( rtc::scoped_refptr data_channel) { NEARBY_LOG(INFO, "OnDataChannel"); - data_channel->RegisterObserver( - connection_flow_->CreateDataChannelObserver(data_channel)); + if (connection_flow_) { + data_channel->RegisterObserver( + connection_flow_->CreateDataChannelObserver(data_channel)); + } } void PeerConnectionObserverImpl::OnIceGatheringChange( @@ -47,7 +55,9 @@ void PeerConnectionObserverImpl::OnConnectionChange( NEARBY_LOG(INFO, "OnConnectionChange: %d", new_state); OffloadFromSignalingThread([this, new_state]() { - connection_flow_->ProcessOnPeerConnectionChange(new_state); + if (connection_flow_) { + connection_flow_->ProcessOnPeerConnectionChange(new_state); + } }); } @@ -55,6 +65,11 @@ void PeerConnectionObserverImpl ::OnRenegotiationNeeded() { NEARBY_LOG(INFO, "OnRenegotiationNeeded"); } +void PeerConnectionObserverImpl::Shutdown() { + single_threaded_signaling_offloader_.Shutdown(); + connection_flow_ = nullptr; +} + void PeerConnectionObserverImpl::OffloadFromSignalingThread(Runnable runnable) { single_threaded_signaling_offloader_.Execute(std::move(runnable)); } diff --git a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h index a8aee854..a88ed15e 100644 --- a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h +++ b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h @@ -14,10 +14,10 @@ class ConnectionFlow; class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver { public: - ~PeerConnectionObserverImpl() override = default; PeerConnectionObserverImpl( ConnectionFlow* connection_flow, LocalIceCandidateListener local_ice_candidate_listener); + ~PeerConnectionObserverImpl() override; // webrtc::PeerConnectionObserver: void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override; @@ -31,10 +31,12 @@ class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver { webrtc::PeerConnectionInterface::PeerConnectionState new_state) override; void OnRenegotiationNeeded() override; + void Shutdown(); + private: void OffloadFromSignalingThread(Runnable runnable); - ConnectionFlow* connection_flow_; + ConnectionFlow* volatile connection_flow_; LocalIceCandidateListener local_ice_candidate_listener_; SingleThreadExecutor single_threaded_signaling_offloader_; }; From 102f1e6fc3a230960b850d478874896303d4b7f0 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 19 Feb 2021 11:29:29 -0800 Subject: [PATCH 5/8] 358291589 Fix deadlock in BluetoothServerSocket::Accept. --- cpp/platform/impl/g3/bluetooth_classic.cc | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/cpp/platform/impl/g3/bluetooth_classic.cc b/cpp/platform/impl/g3/bluetooth_classic.cc index 331031e6..ffd66ee8 100644 --- a/cpp/platform/impl/g3/bluetooth_classic.cc +++ b/cpp/platform/impl/g3/bluetooth_classic.cc @@ -92,9 +92,8 @@ BluetoothDevice* BluetoothSocket::GetRemoteDevice() { std::unique_ptr BluetoothServerSocket::Accept() { absl::MutexLock lock(&mutex_); - while (pending_sockets_.empty()) { + while (!closed_ && pending_sockets_.empty()) { cond_.Wait(&mutex_); - if (closed_) break; } // whether or not we were running in the wait loop, return early if closed. if (closed_) return {}; From a745acb0d6330ac25b0a2989ce24484b1c13c8f3 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 19 Feb 2021 11:30:34 -0800 Subject: [PATCH 6/8] 358300997 Cleanup Qualifies std::string --- cpp/core/internal/base_endpoint_channel_test.cc | 16 ++++++++++------ cpp/core/internal/base_pcp_handler_test.cc | 2 +- cpp/core/internal/encryption_runner_test.cc | 14 ++++++++------ 3 files changed, 19 insertions(+), 13 deletions(-) diff --git a/cpp/core/internal/base_endpoint_channel_test.cc b/cpp/core/internal/base_endpoint_channel_test.cc index e8f16533..d25f80ee 100644 --- a/cpp/core/internal/base_endpoint_channel_test.cc +++ b/cpp/core/internal/base_endpoint_channel_test.cc @@ -93,9 +93,10 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a, { .on_success_cb = [&latch, &context_a]( - const string& endpoint_id, + const std::string& endpoint_id, std::unique_ptr ukey2, - const string& auth_token, const ByteArray& raw_auth_token) { + const std::string& auth_token, + const ByteArray& raw_auth_token) { NEARBY_LOG(INFO, "client-A side key negotiation done"); EXPECT_TRUE(ukey2->VerifyHandshake()); auto context = ukey2->ToConnectionContext(); @@ -104,7 +105,8 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a, latch.CountDown(); }, .on_failure_cb = - [&latch](const string& endpoint_id, EndpointChannel* channel) { + [&latch](const std::string& endpoint_id, + EndpointChannel* channel) { NEARBY_LOG(INFO, "client-A side key negotiation failed"); latch.CountDown(); }, @@ -114,9 +116,10 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a, { .on_success_cb = [&latch, &context_b]( - const string& endpoint_id, + const std::string& endpoint_id, std::unique_ptr ukey2, - const string& auth_token, const ByteArray& raw_auth_token) { + const std::string& auth_token, + const ByteArray& raw_auth_token) { NEARBY_LOG(INFO, "client-B side key negotiation done"); EXPECT_TRUE(ukey2->VerifyHandshake()); auto context = ukey2->ToConnectionContext(); @@ -125,7 +128,8 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a, latch.CountDown(); }, .on_failure_cb = - [&latch](const string& endpoint_id, EndpointChannel* channel) { + [&latch](const std::string& endpoint_id, + EndpointChannel* channel) { NEARBY_LOG(INFO, "client-B side key negotiation failed"); latch.CountDown(); }, diff --git a/cpp/core/internal/base_pcp_handler_test.cc b/cpp/core/internal/base_pcp_handler_test.cc index bbd48478..d4393214 100644 --- a/cpp/core/internal/base_pcp_handler_test.cc +++ b/cpp/core/internal/base_pcp_handler_test.cc @@ -116,7 +116,7 @@ class MockPcpHandler : public BasePcpHandler { (override)); MOCK_METHOD(Status, StopAdvertisingImpl, (ClientProxy * client), (override)); MOCK_METHOD(StartOperationResult, StartDiscoveryImpl, - (ClientProxy * client, const string& service_id, + (ClientProxy * client, const std::string& service_id, const ConnectionOptions& options), (override)); MOCK_METHOD(Status, StopDiscoveryImpl, (ClientProxy * client), (override)); diff --git a/cpp/core/internal/encryption_runner_test.cc b/cpp/core/internal/encryption_runner_test.cc index acb023c0..48802bcf 100644 --- a/cpp/core/internal/encryption_runner_test.cc +++ b/cpp/core/internal/encryption_runner_test.cc @@ -88,15 +88,16 @@ TEST(EncryptionRunnerTest, ReadWrite) { &user_a.client, "endpoint_id", &user_a.channel, { .on_success_cb = - [&response](const string& endpoint_id, + [&response](const std::string& endpoint_id, std::unique_ptr ukey2, - const string& auth_token, + const std::string& auth_token, const ByteArray& raw_auth_token) { response.server_status = Response::Status::kDone; response.latch.CountDown(); }, .on_failure_cb = - [&response](const string& endpoint_id, EndpointChannel* channel) { + [&response](const std::string& endpoint_id, + EndpointChannel* channel) { response.server_status = Response::Status::kFailed; response.latch.CountDown(); }, @@ -105,15 +106,16 @@ TEST(EncryptionRunnerTest, ReadWrite) { &user_b.client, "endpoint_id", &user_b.channel, { .on_success_cb = - [&response](const string& endpoint_id, + [&response](const std::string& endpoint_id, std::unique_ptr ukey2, - const string& auth_token, + const std::string& auth_token, const ByteArray& raw_auth_token) { response.client_status = Response::Status::kDone; response.latch.CountDown(); }, .on_failure_cb = - [&response](const string& endpoint_id, EndpointChannel* channel) { + [&response](const std::string& endpoint_id, + EndpointChannel* channel) { response.client_status = Response::Status::kFailed; response.latch.CountDown(); }, From da0de8973b63b64fb625868d1571cc2556f35e02 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 19 Feb 2021 11:31:40 -0800 Subject: [PATCH 7/8] 358303772 endpoint manager clean up to reduce chance of countDownLatch crash --- cpp/core/internal/base_pcp_handler_test.cc | 28 +++++++++++++++++ cpp/core/internal/endpoint_manager.cc | 36 +++++++++++++++------- cpp/platform/base/feature_flags.h | 3 ++ 3 files changed, 56 insertions(+), 11 deletions(-) diff --git a/cpp/core/internal/base_pcp_handler_test.cc b/cpp/core/internal/base_pcp_handler_test.cc index d4393214..6c02dc07 100644 --- a/cpp/core/internal/base_pcp_handler_test.cc +++ b/cpp/core/internal/base_pcp_handler_test.cc @@ -15,6 +15,7 @@ #include "proto/connections/offline_wire_formats.pb.h" #include "platform/base/byte_array.h" #include "platform/base/exception.h" +#include "platform/base/medium_environment.h" #include "platform/public/count_down_latch.h" #include "platform/public/pipe.h" #include "proto/connections_enums.pb.h" @@ -385,9 +386,11 @@ class BasePcpHandlerTest .endpoint_distance_changed_cb = mock_discovery_listener_.endpoint_distance_changed_cb.AsStdFunction(), }; + MediumEnvironment& env_ = MediumEnvironment::Instance(); }; TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) { + env_.Start(); Mediums m; EndpointChannelManager ecm; EndpointManager em(&ecm); @@ -395,9 +398,11 @@ TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) { MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); SUCCEED(); bwu.Shutdown(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) { + env_.Start(); ClientProxy client; Mediums m; EndpointChannelManager ecm; @@ -406,9 +411,11 @@ TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) { MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); StartAdvertising(&client, &pcp_handler); bwu.Shutdown(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) { + env_.Start(); ClientProxy client; Mediums m; EndpointChannelManager ecm; @@ -421,9 +428,11 @@ TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) { pcp_handler.StopAdvertising(&client); EXPECT_FALSE(client.IsAdvertising()); bwu.Shutdown(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) { + env_.Start(); ClientProxy client; Mediums m; EndpointChannelManager ecm; @@ -432,9 +441,11 @@ TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) { MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); StartDiscovery(&client, &pcp_handler); bwu.Shutdown(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) { + env_.Start(); ClientProxy client; Mediums m; EndpointChannelManager ecm; @@ -447,9 +458,11 @@ TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) { pcp_handler.StopDiscovery(&client); EXPECT_FALSE(client.IsDiscovering()); bwu.Shutdown(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) { + env_.Start(); std::string endpoint_id{"1234"}; ClientProxy client; Mediums m; @@ -472,9 +485,11 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) { channel_b->Close(); bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, IoError_RequestConnectionFails) { + env_.Start(); std::string endpoint_id{"1234"}; ClientProxy client; Mediums m; @@ -499,9 +514,11 @@ TEST_P(BasePcpHandlerTest, IoError_RequestConnectionFails) { channel_b->Close(); bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) { + env_.Start(); std::string endpoint_id{"1234"}; ClientProxy client; Mediums m; @@ -528,9 +545,11 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) { channel_b->Close(); bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) { + env_.Start(); std::string endpoint_id{"1234"}; ClientProxy client; Mediums m; @@ -553,9 +572,11 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) { channel_b->Close(); bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) { + env_.Start(); std::string endpoint_id{"1234"}; ClientProxy client; Mediums m; @@ -588,9 +609,11 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) { channel_b->Close(); bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); + env_.Stop(); } TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) { + env_.Start(); std::atomic_int destroyed_flag = 0; int mediums_count = 0; { @@ -623,9 +646,11 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) { pcp_handler.DisconnectFromEndpointManager(); } EXPECT_EQ(destroyed_flag.load(), mediums_count); + env_.Stop(); } TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) { + env_.Start(); BooleanMediumSelector allowed = GetParam(); if (allowed.Count(true) < 2) { // Ignore single-medium test cases, and implicit "all mediums" case. @@ -670,12 +695,14 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) { pcp_handler.DisconnectFromEndpointManager(); } EXPECT_EQ(destroyed_flag.load(), mediums_count); + env_.Stop(); } INSTANTIATE_TEST_SUITE_P(ParameterizedBasePcpHandlerTest, BasePcpHandlerTest, ::testing::ValuesIn(kTestCases)); TEST_F(BasePcpHandlerTest, InjectEndpoint) { + env_.Start(); std::string service_id{"service"}; std::string endpoint_id{"ABCD"}; ClientProxy client; @@ -727,6 +754,7 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) { .remote_bluetooth_mac_address = ByteArray(kFakeMacAddress), }); bwu.Shutdown(); + env_.Stop(); } } // namespace diff --git a/cpp/core/internal/endpoint_manager.cc b/cpp/core/internal/endpoint_manager.cc index eb994719..35e16ed4 100644 --- a/cpp/core/internal/endpoint_manager.cc +++ b/cpp/core/internal/endpoint_manager.cc @@ -6,10 +6,10 @@ #include "core/internal/endpoint_channel.h" #include "core/internal/offline_frames.h" #include "platform/base/exception.h" +#include "platform/base/feature_flags.h" #include "platform/public/count_down_latch.h" #include "platform/public/logging.h" #include "proto/connections_enums.pb.h" - namespace location { namespace nearby { namespace connections { @@ -287,21 +287,19 @@ void EndpointManager::UnregisterFrameProcessor(V1Frame::FrameType frame_type, EndpointManager::FrameProcessor* EndpointManager::GetFrameProcessor( V1Frame::FrameType frame_type) { EndpointManager::FrameProcessor* processor = nullptr; - CountDownLatch latch(1); - RunOnEndpointManagerThread([this, frame_type, &processor, &latch]() { - auto it = frame_processors_.find(frame_type); - if (it != frame_processors_.end()) { - processor = it->second; - } - latch.CountDown(); - }); - latch.Await(); + auto it = frame_processors_.find(frame_type); + if (it != frame_processors_.end()) { + processor = it->second; + } return processor; } void EndpointManager::EnsureWorkersTerminated(const std::string& endpoint_id) { + NEARBY_LOG(ERROR, "EnsureWorkersTerminated for endpoint %s", + endpoint_id.c_str()); auto item = endpoints_.find(endpoint_id); if (item != endpoints_.end()) { + NEARBY_LOGS(INFO) << "EndpointState found for id: " << endpoint_id; // If another instance of data and keep-alive handlers is running, it will // terminate soon; we should block until it happens. EndpointState& endpoint_state = item->second; @@ -332,13 +330,21 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client, RunOnEndpointManagerThread([this, client, channel = channel.release(), &endpoint_id, &info, &options, &listener, &latch]() { + if (endpoints_.contains(endpoint_id)) { + NEARBY_LOG(WARNING, "Registing duplicate endpoint %s", + endpoint_id.c_str()); + if (!FeatureFlags::GetInstance() + .GetFlags() + .endpoint_manager_ensure_workers_terminated_inside_remove) { + EnsureWorkersTerminated(endpoint_id); + } + } // Pass ownership of channel to EndpointChannelManager NEARBY_LOG(INFO, "Registering endpoint with channel manager: id=%s", endpoint_id.c_str()); channel_manager_->RegisterChannelForEndpoint( client, endpoint_id, std::unique_ptr(channel)); - EnsureWorkersTerminated(endpoint_id); EndpointState& endpoint_state = endpoints_.emplace(endpoint_id, EndpointState()).first->second; endpoint_state.client = client; @@ -392,6 +398,7 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client, void EndpointManager::UnregisterEndpoint(ClientProxy* client, const std::string& endpoint_id) { + NEARBY_LOG(ERROR, "UnregisterEndpoint for endpoint %s", endpoint_id.c_str()); CountDownLatch latch(1); RunOnEndpointManagerThread([this, client, endpoint_id, &latch]() { RemoveEndpoint(client, endpoint_id, @@ -428,6 +435,7 @@ std::vector EndpointManager::SendPayloadChunk( // allow synchronous behavior here it will cause a live lock. void EndpointManager::DiscardEndpoint(ClientProxy* client, const std::string& endpoint_id) { + NEARBY_LOG(ERROR, "DiscardEndpoint for endpoint %s", endpoint_id.c_str()); RunOnEndpointManagerThread([this, client, endpoint_id]() { RemoveEndpoint(client, endpoint_id, /*notify=*/ @@ -450,6 +458,7 @@ std::vector EndpointManager::SendControlMessage( void EndpointManager::RemoveEndpoint(ClientProxy* client, const std::string& endpoint_id, bool notify) { + NEARBY_LOG(ERROR, "RemoveEndpoint for endpoint %s", endpoint_id.c_str()); // Unregistering from channel_manager_ will also serve to terminate // the dedicated handler and KeepAlive threads we started when we registered // this endpoint. @@ -464,6 +473,11 @@ void EndpointManager::RemoveEndpoint(ClientProxy* client, client->OnDisconnected(endpoint_id, notify); NEARBY_LOG(INFO, "Removed endpoint; id=%s", endpoint_id.c_str()); } + if (FeatureFlags::GetInstance() + .GetFlags() + .endpoint_manager_ensure_workers_terminated_inside_remove) { + EnsureWorkersTerminated(endpoint_id); + } } // @EndpointManagerThread diff --git a/cpp/platform/base/feature_flags.h b/cpp/platform/base/feature_flags.h index e5b58747..6880c1b7 100644 --- a/cpp/platform/base/feature_flags.h +++ b/cpp/platform/base/feature_flags.h @@ -22,6 +22,9 @@ class FeatureFlags { // first one. bool disallow_out_of_order_bwu_avail_event = true; bool enable_async_bandwidth_upgrade = true; + // Let endpoint_manager erase deleted endpoint from endpoints_ inside + // function RemoveEndpoint. + bool endpoint_manager_ensure_workers_terminated_inside_remove = true; }; static const FeatureFlags& GetInstance() { From d8f4d4963bbd9a469ae75abceb2851c53a923332 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 19 Feb 2021 11:32:44 -0800 Subject: [PATCH 8/8] 358328147 Synchronous scheduled task cancel --- cpp/platform/base/feature_flags.h | 3 + cpp/platform/public/BUILD | 2 + cpp/platform/public/cancelable.h | 14 ++++- cpp/platform/public/cancellable_task.h | 56 ++++++++++++++++++ cpp/platform/public/scheduled_executor.h | 10 +++- .../public/scheduled_executor_test.cc | 57 +++++++++++++++++++ 6 files changed, 137 insertions(+), 5 deletions(-) create mode 100644 cpp/platform/public/cancellable_task.h diff --git a/cpp/platform/base/feature_flags.h b/cpp/platform/base/feature_flags.h index 6880c1b7..e85342e2 100644 --- a/cpp/platform/base/feature_flags.h +++ b/cpp/platform/base/feature_flags.h @@ -25,6 +25,9 @@ class FeatureFlags { // Let endpoint_manager erase deleted endpoint from endpoints_ inside // function RemoveEndpoint. bool endpoint_manager_ensure_workers_terminated_inside_remove = true; + // If a scheduled runnable is already running, Cancel() will synchronously + // wait for the task to complete. + bool cancel_waits_for_running_tasks = true; }; static const FeatureFlags& GetInstance() { diff --git a/cpp/platform/public/BUILD b/cpp/platform/public/BUILD index 3c1514bb..400afc55 100644 --- a/cpp/platform/public/BUILD +++ b/cpp/platform/public/BUILD @@ -8,6 +8,7 @@ cc_library( "atomic_reference.h", "cancelable.h", "cancelable_alarm.h", + "cancellable_task.h", "condition_variable.h", "count_down_latch.h", "crypto.h", @@ -93,6 +94,7 @@ cc_library( cc_test( name = "public_test", size = "small", + timeout = "moderate", srcs = [ "atomic_boolean_test.cc", "atomic_reference_test.cc", diff --git a/cpp/platform/public/cancelable.h b/cpp/platform/public/cancelable.h index a565e669..ec0273a1 100644 --- a/cpp/platform/public/cancelable.h +++ b/cpp/platform/public/cancelable.h @@ -5,6 +5,7 @@ #include #include "platform/api/cancelable.h" +#include "platform/public/cancellable_task.h" namespace location { namespace nearby { @@ -21,14 +22,21 @@ class Cancelable final { // This constructor is used internally only, // by other classes in "//platform/public/". - explicit Cancelable(std::shared_ptr impl) - : impl_(std::move(impl)) {} + explicit Cancelable(std::shared_ptr task, + std::shared_ptr impl) + : task_{task}, impl_(std::move(impl)) {} - bool Cancel() { return impl_ ? impl_->Cancel() : false; } + bool Cancel() { + if (!impl_) return false; + bool result = impl_->Cancel(); + task_->CancelAndWaitIfStarted(); + return result; + } bool IsValid() { return impl_ != nullptr; } private: + std::shared_ptr task_; std::shared_ptr impl_; }; diff --git a/cpp/platform/public/cancellable_task.h b/cpp/platform/public/cancellable_task.h new file mode 100644 index 00000000..70c6e013 --- /dev/null +++ b/cpp/platform/public/cancellable_task.h @@ -0,0 +1,56 @@ +#ifndef PLATFORM_PUBLIC_CANCELLABLE_TASK_H_ +#define PLATFORM_PUBLIC_CANCELLABLE_TASK_H_ + +#include + +#include "platform/base/feature_flags.h" +#include "platform/base/runnable.h" +#include "platform/public/atomic_boolean.h" +#include "platform/public/future.h" + +namespace location { +namespace nearby { + +/** + * Runnable wrapper that allows one to wait for the task + * to complete if it is already running. + */ +class CancellableTask { + public: + explicit CancellableTask(Runnable&& runnable) + : runnable_{std::move(runnable)} {} + + /** + * Try to cancel the task and wait until completion if the task is already + * running. + */ + void CancelAndWaitIfStarted() { + if (started_or_cancelled_.Set(true)) { + if (FeatureFlags::GetInstance() + .GetFlags() + .cancel_waits_for_running_tasks) { + // task could still be running, wait until finish + finished_.Get(); + } + } else { + // mark as finished to support multiple calls to this method + finished_.Set(true); + } + } + + void operator()() { + if (started_or_cancelled_.Set(true)) return; + runnable_(); + finished_.Set(true); + } + + private: + AtomicBoolean started_or_cancelled_; + Future finished_; + Runnable runnable_; +}; + +} // namespace nearby +} // namespace location + +#endif // PLATFORM_PUBLIC_CANCELLABLE_TASK_H_ diff --git a/cpp/platform/public/scheduled_executor.h b/cpp/platform/public/scheduled_executor.h index 6dea9e55..e250b44f 100644 --- a/cpp/platform/public/scheduled_executor.h +++ b/cpp/platform/public/scheduled_executor.h @@ -9,6 +9,7 @@ #include "platform/api/scheduled_executor.h" #include "platform/base/runnable.h" #include "platform/public/cancelable.h" +#include "platform/public/cancellable_task.h" #include "platform/public/mutex.h" #include "platform/public/mutex_lock.h" #include "absl/time/time.h" @@ -59,8 +60,13 @@ class ScheduledExecutor final { Cancelable Schedule(Runnable&& runnable, absl::Duration duration) ABSL_LOCKS_EXCLUDED(mutex_) { MutexLock lock(&mutex_); - return impl_ ? Cancelable(impl_->Schedule(std::move(runnable), duration)) - : Cancelable(); + if (impl_) { + auto task = std::make_shared(std::move(runnable)); + return Cancelable(task, + impl_->Schedule([task]() { (*task)(); }, duration)); + } else { + return Cancelable(); + } } private: diff --git a/cpp/platform/public/scheduled_executor_test.cc b/cpp/platform/public/scheduled_executor_test.cc index 95720137..49155c0d 100644 --- a/cpp/platform/public/scheduled_executor_test.cc +++ b/cpp/platform/public/scheduled_executor_test.cc @@ -4,6 +4,7 @@ #include #include "platform/base/exception.h" +#include "platform/public/count_down_latch.h" #include "gtest/gtest.h" #include "absl/synchronization/mutex.h" #include "absl/time/clock.h" @@ -81,6 +82,20 @@ TEST(ScheduledExecutorTest, CanCancel) { EXPECT_EQ(value, 0); } +TEST(ScheduledExecutorTest, CanCancelTwice) { + ScheduledExecutor executor; + std::atomic_int value = 0; + Cancelable cancelable = + executor.Schedule([&value]() { value += 1; }, kShortDelay); + EXPECT_EQ(value, 0); + + cancelable.Cancel(); + cancelable.Cancel(); + + absl::SleepFor(kLongDelay); + EXPECT_EQ(value, 0); +} + TEST(ScheduledExecutorTest, FailToCancel) { absl::Mutex mutex; absl::CondVar cond; @@ -104,5 +119,47 @@ TEST(ScheduledExecutorTest, FailToCancel) { EXPECT_EQ(value, 1); } +TEST(ScheduledExecutorTest, + CancelWhileRunning_TaskCompletesBeforeCancelReturns) { + CountDownLatch start_latch(1); + ScheduledExecutor executor; + std::atomic_int value = 0; + // A task that takes a little bit of time to complete + Cancelable cancelable = executor.Schedule( + [&start_latch, &value]() { + start_latch.CountDown(); + absl::SleepFor(kLongDelay); + value += 1; + }, + absl::ZeroDuration()); + + start_latch.Await(); + cancelable.Cancel(); + + EXPECT_EQ(value, 1); +} + +TEST(ScheduledExecutorTest, + CancelTwiceWhileRunning_TaskCompletesBeforeCancelReturns) { + CountDownLatch start_latch(1); + ScheduledExecutor executor; + std::atomic_int value = 0; + // A task that takes a little bit of time to complete + Cancelable cancelable = executor.Schedule( + [&start_latch, &value]() { + start_latch.CountDown(); + absl::SleepFor(kLongDelay); + value += 1; + }, + absl::ZeroDuration()); + + start_latch.Await(); + + cancelable.Cancel(); + cancelable.Cancel(); + + EXPECT_EQ(value, 1); +} + } // namespace nearby } // namespace location