From e44f7eac6199a9b31011815eb9bf45182196baa7 Mon Sep 17 00:00:00 2001 From: Janusz Sobczak Date: Tue, 29 Nov 2022 16:57:41 -0800 Subject: [PATCH] Aynchronous start and stop broadcast PiperOrigin-RevId: 491772964 --- presence/implementation/BUILD | 1 + .../implementation/service_controller_impl.cc | 108 +++++++++++++----- .../implementation/service_controller_impl.h | 39 ++++++- .../service_controller_impl_test.cc | 56 ++++++--- 4 files changed, 151 insertions(+), 53 deletions(-) diff --git a/presence/implementation/BUILD b/presence/implementation/BUILD index 7e08eff9..55932954 100644 --- a/presence/implementation/BUILD +++ b/presence/implementation/BUILD @@ -71,6 +71,7 @@ cc_library( "//presence:__subpackages__", ], deps = [ + "@com_google_absl//absl/base:core_headers", "@com_google_absl//absl/container:flat_hash_map", "@com_google_absl//absl/container:flat_hash_set", "@com_google_absl//absl/log:die_if_null", diff --git a/presence/implementation/service_controller_impl.cc b/presence/implementation/service_controller_impl.cc index 4d8c84ca..86067ec6 100644 --- a/presence/implementation/service_controller_impl.cc +++ b/presence/implementation/service_controller_impl.cc @@ -61,47 +61,95 @@ absl::StatusOr ServiceControllerImpl::StartBroadcast( callback.start_broadcast_cb(Status{Status::Value::kError}); return request.status(); } - absl::StatusOr advertisement = - AdvertisementFactory(&credential_manager_).CreateAdvertisement(*request); - if (!advertisement.ok()) { - NEARBY_LOGS(WARNING) << "Can't create advertisement, reason: " - << advertisement.status(); - callback.start_broadcast_cb(Status{Status::Value::kError}); - return advertisement.status(); - } - std::unique_ptr session = - mediums_.GetBle().StartAdvertising( - *advertisement, broadcast_request.power_mode, - AdvertisingCallback{.start_advertising_result = - [callback](BleOperationStatus status) { - callback.start_broadcast_cb( - ConvertBleStatus(status)); - }}); - if (!session) { - callback.start_broadcast_cb(Status{Status::Value::kError}); - return absl::UnavailableError("Failed to start broadcasting"); - } - BroadcastSessionId id = GenerateBroadcastSessionId(); - sessions_.insert({id, Session{.advertising_session = std::move(session)}}); + RunOnServiceControllerThread( + "start-broadcast", + [this, id, power_mode = broadcast_request.power_mode, request = *request, + broadcast_callback = + std::move(callback)]() ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_) { + sessions_.insert({id, BroadcastSessionState(broadcast_callback)}); + absl::StatusOr advertisement = + AdvertisementFactory(&credential_manager_) + .CreateAdvertisement(request); + if (!advertisement.ok()) { + NEARBY_LOGS(WARNING) << "Can't create advertisement, reason: " + << advertisement.status(); + NotifyStartCallbackStatus(id, Status{Status::Value::kError}); + return; + } + std::unique_ptr session = + mediums_.GetBle().StartAdvertising( + *advertisement, power_mode, + AdvertisingCallback{.start_advertising_result = + [this, id](BleOperationStatus status) { + NotifyStartCallbackStatus( + id, ConvertBleStatus(status)); + }}); + if (!session) { + NotifyStartCallbackStatus(id, Status{Status::Value::kError}); + return; + } + sessions_.at(id).SetAdvertisingSession(std::move(session)); + }); return id; } +void ServiceControllerImpl::NotifyStartCallbackStatus(BroadcastSessionId id, + Status status) { + RunOnServiceControllerThread("started-broadcast-cb", + [this, id, status]() + ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_) { + auto it = sessions_.find(id); + if (it == sessions_.end()) { + return; + } + it->second.CallStartedCallback(status); + if (!status.Ok()) { + // Delete failed session. + sessions_.erase(it); + } + }); +} + void ServiceControllerImpl::StopBroadcast(BroadcastSessionId id) { - auto it = sessions_.find(id); - if (it != sessions_.end()) { - it->second.advertising_session->stop_advertising(); - sessions_.erase(it); - } else { - NEARBY_LOGS(WARNING) << absl::StrFormat( - "BroadcastSessionId(0x%x) not found", id); - } + RunOnServiceControllerThread( + "stop-broadcast", [this, id]() ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_) { + auto it = sessions_.find(id); + if (it == sessions_.end()) { + NEARBY_LOGS(VERBOSE) + << absl::StrFormat("BroadcastSession(0x%x) not found", id); + return; + } + it->second.StopAdvertising(); + sessions_.erase(it); + }); } BroadcastSessionId ServiceControllerImpl::GenerateBroadcastSessionId() { return absl::Uniform(bit_gen_); } +void ServiceControllerImpl::BroadcastSessionState::SetAdvertisingSession( + std::unique_ptr session) { + advertising_session_ = std::move(session); +} + +void ServiceControllerImpl::BroadcastSessionState::CallStartedCallback( + Status status) { + BroadcastCallback callback = std::move(broadcast_callback_); + if (callback.start_broadcast_cb) { + callback.start_broadcast_cb(status); + } +} + +void ServiceControllerImpl::BroadcastSessionState::StopAdvertising() { + std::unique_ptr advertising_session = + std::move(advertising_session_); + if (advertising_session) { + advertising_session->stop_advertising(); + } +} + } // namespace presence } // namespace nearby diff --git a/presence/implementation/service_controller_impl.h b/presence/implementation/service_controller_impl.h index 908cbb02..874b3652 100644 --- a/presence/implementation/service_controller_impl.h +++ b/presence/implementation/service_controller_impl.h @@ -16,8 +16,12 @@ #define THIRD_PARTY_NEARBY_PRESENCE_IMPLEMENTATION_SERVICE_CONTROLLER_IMPL_H_ #include +#include +#include +#include "absl/base/thread_annotations.h" #include "absl/random/random.h" +#include "internal/platform/single_thread_executor.h" #include "presence/broadcast_request.h" #include "presence/data_types.h" #include "presence/implementation/credential_manager_impl.h" @@ -25,6 +29,7 @@ #include "presence/implementation/scan_manager.h" #include "presence/implementation/service_controller.h" #include "presence/scan_request.h" + /* * This class implements {@code ServiceController} functions. Owns mediums and * other managers instances. @@ -34,30 +39,54 @@ namespace presence { class ServiceControllerImpl : public ServiceController { public: + using SingleThreadExecutor = ::location::nearby::SingleThreadExecutor; using AdvertisingSession = - location::nearby::api::ble_v2::BleMedium::AdvertisingSession; + ::location::nearby::api::ble_v2::BleMedium::AdvertisingSession; + using Runnable = ::location::nearby::Runnable; + ServiceControllerImpl() = default; + ~ServiceControllerImpl() override { executor_.Shutdown(); } + std::unique_ptr StartScan(ScanRequest scan_request, ScanCallback callback) override; absl::StatusOr StartBroadcast( BroadcastRequest broadcast_request, BroadcastCallback callback) override; void StopBroadcast(BroadcastSessionId) override; + SingleThreadExecutor& GetBackgroundExecutor() { return executor_; } + // Gives tests access to mediums. Mediums& GetMediums() { return mediums_; } private: - struct Session { - std::unique_ptr advertising_session; - }; + class BroadcastSessionState { + public: + explicit BroadcastSessionState(BroadcastCallback broadcast_callback) + : broadcast_callback_(broadcast_callback) {} + void SetAdvertisingSession(std::unique_ptr session); + + void CallStartedCallback(Status status); + + void StopAdvertising(); + + private: + BroadcastCallback broadcast_callback_; + std::unique_ptr advertising_session_; + }; + SingleThreadExecutor executor_; BroadcastSessionId GenerateBroadcastSessionId(); + void NotifyStartCallbackStatus(BroadcastSessionId id, Status status); + void RunOnServiceControllerThread(absl::string_view name, Runnable runnable) { + executor_.Execute(std::string(name), std::move(runnable)); + } Mediums mediums_; // NOLINT: further impl will use it. CredentialManagerImpl credential_manager_; // NOLINT: further impl will use it. ScanManager scan_manager_{ mediums_, credential_manager_}; // NOLINT: further impl will use it. - absl::flat_hash_map sessions_; + absl::flat_hash_map sessions_ + ABSL_GUARDED_BY(executor_); absl::BitGen bit_gen_; }; diff --git a/presence/implementation/service_controller_impl_test.cc b/presence/implementation/service_controller_impl_test.cc index d9e65206..e8053508 100644 --- a/presence/implementation/service_controller_impl_test.cc +++ b/presence/implementation/service_controller_impl_test.cc @@ -20,6 +20,7 @@ #include "gmock/gmock.h" #include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.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" @@ -31,6 +32,7 @@ namespace { using FeatureFlags = ::location::nearby::FeatureFlags::Flags; using internal::IdentityType; +using ::location::nearby::CountDownLatch; using ::location::nearby::MediumEnvironment; using ::testing::status::StatusIs; @@ -61,15 +63,37 @@ class MediumEnvironmentStarter { class ServiceControllerImplTest : public testing::TestWithParam { protected: + void TearDown() override { MediumEnvironment::Instance().Sync(); } + bool IsAdvertising() { + WaitForServiceControllerTasks(); + MediumEnvironment::Instance().Sync(); + return MediumEnvironment::Instance() + .GetBleV2MediumStatus( + *service_controller_.GetMediums().GetBle().GetImpl()) + ->is_advertising; + } + BroadcastCallback CreateBroadcastCallback() { + return BroadcastCallback{.start_broadcast_cb = [this](Status status) { + start_broadcast_status_.Set(status); + }}; + } + + void WaitForServiceControllerTasks() { + CountDownLatch latch(1); + service_controller_.GetBackgroundExecutor().Execute( + [&]() { latch.CountDown(); }); + latch.Await(); + } + // The medium environment must be initialized (started) before the service // controller. MediumEnvironmentStarter env_; - ServiceControllerImpl service_controller_; location::nearby::Future start_broadcast_status_; BroadcastCallback broadcast_callback_{ .start_broadcast_cb = [this](Status status) { start_broadcast_status_.Set(status); }}; + ServiceControllerImpl service_controller_; }; INSTANTIATE_TEST_SUITE_P(ParametrisedServiceControllerImplTest, @@ -80,40 +104,34 @@ TEST_P(ServiceControllerImplTest, StartBroadcastPublicIdentity) { absl::StatusOr session = service_controller_.StartBroadcast( CreateBroadcastRequest(internal::IDENTITY_TYPE_PUBLIC), - broadcast_callback_); + CreateBroadcastCallback()); EXPECT_OK(session); EXPECT_TRUE(start_broadcast_status_.Get().ok()); EXPECT_EQ(start_broadcast_status_.Get().GetResult(), Status{Status::Value::kSuccess}); - EXPECT_TRUE(MediumEnvironment::Instance() - .GetBleV2MediumStatus( - *service_controller_.GetMediums().GetBle().GetImpl()) - ->is_advertising); + EXPECT_TRUE(IsAdvertising()); } TEST_P(ServiceControllerImplTest, StartAndStopBroadcast) { absl::StatusOr session = service_controller_.StartBroadcast( CreateBroadcastRequest(internal::IDENTITY_TYPE_PUBLIC), - broadcast_callback_); - + CreateBroadcastCallback()); ASSERT_OK(session); + EXPECT_TRUE(IsAdvertising()); + service_controller_.StopBroadcast(*session); - MediumEnvironment::Instance().Sync(); - EXPECT_FALSE(MediumEnvironment::Instance() - .GetBleV2MediumStatus( - *service_controller_.GetMediums().GetBle().GetImpl()) - ->is_advertising); + EXPECT_FALSE(IsAdvertising()); } TEST_P(ServiceControllerImplTest, StopBroadcastTwiceNoSideEffects) { absl::StatusOr session = service_controller_.StartBroadcast( CreateBroadcastRequest(internal::IDENTITY_TYPE_PUBLIC), - broadcast_callback_); + CreateBroadcastCallback()); ASSERT_OK(session); - service_controller_.StopBroadcast(*session); + EXPECT_TRUE(IsAdvertising()); service_controller_.StopBroadcast(*session); } @@ -125,12 +143,13 @@ TEST_P(ServiceControllerImplTest, StopBroadcastInvalidSessionNoSideEffects) { TEST_P(ServiceControllerImplTest, StartBroadcastInvalidRequestFails) { absl::StatusOr session = service_controller_.StartBroadcast(BroadcastRequest{}, - broadcast_callback_); + CreateBroadcastCallback()); EXPECT_THAT(session, StatusIs(absl::StatusCode::kInvalidArgument)); EXPECT_TRUE(start_broadcast_status_.Get().ok()); EXPECT_EQ(start_broadcast_status_.Get().GetResult(), Status{Status::Value::kError}); + EXPECT_FALSE(IsAdvertising()); } TEST_P(ServiceControllerImplTest, StartBroadcastPrivateIdentityFails) { @@ -138,12 +157,13 @@ TEST_P(ServiceControllerImplTest, StartBroadcastPrivateIdentityFails) { absl::StatusOr session = service_controller_.StartBroadcast( CreateBroadcastRequest(internal::IDENTITY_TYPE_PRIVATE), - broadcast_callback_); + CreateBroadcastCallback()); - EXPECT_THAT(session, StatusIs(absl::StatusCode::kUnavailable)); + ASSERT_OK(session); EXPECT_TRUE(start_broadcast_status_.Get().ok()); EXPECT_EQ(start_broadcast_status_.Get().GetResult(), Status{Status::Value::kError}); + EXPECT_FALSE(IsAdvertising()); } } // namespace