diff --git a/connections/implementation/mediums/ble/BUILD b/connections/implementation/mediums/ble/BUILD index 32655111..3a1234f5 100644 --- a/connections/implementation/mediums/ble/BUILD +++ b/connections/implementation/mediums/ble/BUILD @@ -176,7 +176,6 @@ cc_test( ":ble_advertisement_header", ":bloom_filter", "//connections/implementation:types", - "//connections/implementation/flags:connections_flags", "//connections/implementation/mediums:utils", "//connections/implementation/mediums/advertisements:dct_advertisement", "//internal/flags:nearby_flags", @@ -193,6 +192,7 @@ cc_test( "@com_github_protobuf_matchers//protobuf-matchers", "@com_google_absl//absl/base:core_headers", "@com_google_absl//absl/hash:hash_testing", + "@com_google_absl//absl/memory", "@com_google_absl//absl/status", "@com_google_absl//absl/status:statusor", "@com_google_absl//absl/strings", diff --git a/connections/implementation/mediums/ble/discovered_peripheral_tracker.cc b/connections/implementation/mediums/ble/discovered_peripheral_tracker.cc index 59549ce8..c1412a18 100644 --- a/connections/implementation/mediums/ble/discovered_peripheral_tracker.cc +++ b/connections/implementation/mediums/ble/discovered_peripheral_tracker.cc @@ -63,18 +63,30 @@ constexpr absl::Duration kExtendedAdvertisementHeaderDelay = absl::Seconds(3); constexpr absl::Duration kAdvertisementHeaderExpiry = absl::Seconds(15); } // namespace +// Private c'tor for testing. +DiscoveredPeripheralTracker::DiscoveredPeripheralTracker( + bool is_extended_advertisement_available, bool start_fetch_executor) + : is_extended_advertisement_available_( + is_extended_advertisement_available), + start_fetch_executor_(start_fetch_executor) {} + DiscoveredPeripheralTracker::DiscoveredPeripheralTracker( bool is_extended_advertisement_available) - : is_extended_advertisement_available_( - is_extended_advertisement_available) { - executor_ = std::make_unique(kGattThreadCount); - executor_->Execute([this]() { GattFetchingLoop(); }); -} + : DiscoveredPeripheralTracker(is_extended_advertisement_available, + /*start_fetch_executor=*/true) {} DiscoveredPeripheralTracker::~DiscoveredPeripheralTracker() { Shutdown(); } +void DiscoveredPeripheralTracker::StartFetchExecutorIfNeeded() { + if (executor_ != nullptr) { + return; + } + executor_ = std::make_unique(kGattThreadCount); + executor_->Execute([this]() { GattFetchingLoop(); }); +} + void DiscoveredPeripheralTracker::Shutdown() { if (shutting_down_.Set(true)) { return; @@ -101,6 +113,9 @@ void DiscoveredPeripheralTracker::StartTracking( const Uuid& fast_advertisement_service_uuid) { MutexLock lock(&mutex_); + if (start_fetch_executor_) { + StartFetchExecutorIfNeeded(); + } ServiceIdInfo service_id_info = { .discovered_peripheral_callback = std::move(discovered_peripheral_callback), @@ -318,9 +333,13 @@ bool DiscoveredPeripheralTracker::IsSkippableGattAdvertisement( ExtractAdvertisementHeaderBytes(advertisement_data)); if (!advertisement_header.IsValid()) { + // Don't skip any advertisement if the header is not valid. It may be one + // of the legacy advertisement formats that is dealt with later. return false; } + // Delay processing GATT advertisements that have corresponding extended + // advertisements for 3s. After that all GATT advertisements are processed. if (advertisement_header.IsSupportExtendedAdvertisement() && (SystemClock::ElapsedRealtime() - medium_start_scanning_time_) < kExtendedAdvertisementHeaderDelay) { @@ -740,16 +759,11 @@ void DiscoveredPeripheralTracker::HandleAdvertisementHeader( } // Determine whether or not we need to read a fresh GATT advertisement. - VLOG(1) << ": Handle GATT advertisement header with hash " + VLOG(1) << "Handle GATT advertisement header with hash " << absl::BytesToHexString( advertisement_header.GetAdvertisementHash().AsStringView()) << " in thread"; - if (executor_ == nullptr) { - // The situation happens when flag value changed - executor_ = std::make_unique(kGattThreadCount); - } - if (!ShouldReadRawAdvertisementFromServer(advertisement_header)) { UpdateCommonStateForFoundBleAdvertisement(advertisement_header); return; @@ -831,7 +845,6 @@ bool DiscoveredPeripheralTracker::ShouldReadRawAdvertisementFromServer( const BleAdvertisementHeader& advertisement_header) { // Check if we have never seen this header. New headers should always be // read. - ByteArray advertisement_header_bytes(advertisement_header); const auto it = advertisement_read_results_.find(advertisement_header); if (it == advertisement_read_results_.end()) { LOG(INFO) << "Received advertisement header with hash " @@ -965,7 +978,8 @@ void DiscoveredPeripheralTracker::FetchRawAdvertisementsInThread( void DiscoveredPeripheralTracker::GattFetchingLoop() { while (true) { - GattFetchTask task; + GattFetchTask* task = nullptr; + bool is_extended_advertisement = false; { MutexLock lock(&task_mutex_); if (shutting_down_) { @@ -986,27 +1000,40 @@ void DiscoveredPeripheralTracker::GattFetchingLoop() { // Non-extended advertisements are prioritized over extended // advertisements. if (!gatt_fetch_tasks_.empty()) { - task = std::move(gatt_fetch_tasks_.front()); - gatt_fetch_tasks_.pop_front(); + task = &gatt_fetch_tasks_.front(); } else if (!gatt_extended_fetch_tasks_.empty()) { - task = std::move(gatt_extended_fetch_tasks_.front()); - gatt_extended_fetch_tasks_.pop_front(); + is_extended_advertisement = true; + task = &gatt_extended_fetch_tasks_.front(); } } + if (task == nullptr) { + LOG(WARNING) << "No task found, skip to fetch raw advertisement."; + continue;; + } // Check if the task is expired. - if (SystemClock::ElapsedRealtime() - task.scheduled_time > + if (SystemClock::ElapsedRealtime() - task->scheduled_time > kAdvertisementHeaderExpiry) { VLOG(1) << "GATT advertisement with hash: " << absl::BytesToHexString( - task.advertisement_header.GetAdvertisementHash() + task->advertisement_header.GetAdvertisementHash() .AsStringView()) << " is expired, skip to fetch raw advertisement."; - continue; + } else { + FetchRawAdvertisementsInThread(task->peripheral, + task->advertisement_header, + std::move(task->advertisement_fetcher)); + } + // Remove task from queue after fetching is done. This allows newly scanned + // advertisements to be deduped against a running fetch task. + { + MutexLock lock(&task_mutex_); + if (is_extended_advertisement) { + gatt_extended_fetch_tasks_.pop_front(); + } else { + gatt_fetch_tasks_.pop_front(); + } } - - FetchRawAdvertisementsInThread(task.peripheral, task.advertisement_header, - std::move(task.advertisement_fetcher)); } } diff --git a/connections/implementation/mediums/ble/discovered_peripheral_tracker.h b/connections/implementation/mediums/ble/discovered_peripheral_tracker.h index 4b13d568..724a9376 100644 --- a/connections/implementation/mediums/ble/discovered_peripheral_tracker.h +++ b/connections/implementation/mediums/ble/discovered_peripheral_tracker.h @@ -70,7 +70,7 @@ class DiscoveredPeripheralTracker { mediums::AdvertisementReadResult& advertisement_read_result)>; explicit DiscoveredPeripheralTracker( - bool is_extended_advertisement_available = false); + bool is_extended_advertisement_available); ~DiscoveredPeripheralTracker(); @@ -112,6 +112,11 @@ class DiscoveredPeripheralTracker { // Shuts down the GATT fetching thread. void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_); + void StartFetchExecutorForTesting() { + MutexLock lock(&mutex_); + StartFetchExecutorIfNeeded(); + } + private: using BleAdvertisementSet = absl::flat_hash_set; @@ -161,6 +166,16 @@ class DiscoveredPeripheralTracker { absl::Time scheduled_time; }; + friend class DiscoveredPeripheralTrackerTest; + + // C'tor used for tests only. + DiscoveredPeripheralTracker( + bool is_extended_advertisement_available, bool start_fetch_executor); + + // Initiatializes the executor used for handling GATT fetch tasks if not + // already started. + void StartFetchExecutorIfNeeded() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + // Clears stale data from any previous sessions. void ClearDataForServiceId(const std::string& service_id) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); @@ -306,7 +321,8 @@ class DiscoveredPeripheralTracker { ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); Mutex mutex_; - bool is_extended_advertisement_available_; + const bool is_extended_advertisement_available_; + const bool start_fetch_executor_; absl::Time medium_start_scanning_time_ ABSL_GUARDED_BY(mutex_) = absl::InfiniteFuture(); diff --git a/connections/implementation/mediums/ble/discovered_peripheral_tracker_test.cc b/connections/implementation/mediums/ble/discovered_peripheral_tracker_test.cc index 1f8c69f0..79fd120f 100644 --- a/connections/implementation/mediums/ble/discovered_peripheral_tracker_test.cc +++ b/connections/implementation/mediums/ble/discovered_peripheral_tracker_test.cc @@ -27,10 +27,10 @@ #include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.h" #include "absl/base/thread_annotations.h" +#include "absl/memory/memory.h" #include "absl/strings/string_view.h" #include "absl/time/clock.h" #include "absl/time/time.h" -#include "connections/implementation/flags/nearby_connections_feature_flags.h" #include "connections/implementation/mediums/advertisements/dct_advertisement.h" #include "connections/implementation/mediums/ble/advertisement_read_result.h" #include "connections/implementation/mediums/ble/ble_advertisement.h" @@ -187,6 +187,8 @@ class MockDiscoveredPeripheralCallback : public DiscoveredPeripheralCallback { MOCK_METHOD(void, OnLegacyDeviceDiscovered, (), ()); }; +} // namespace + class DiscoveredPeripheralTrackerTest : public testing::TestWithParam< std::tuple> { @@ -198,9 +200,11 @@ class DiscoveredPeripheralTrackerTest .use_simulated_clock = true, }; MediumEnvironment::Instance().Start(config); - discovered_peripheral_tracker_ = - std::make_unique( - is_extended_advertisement_available); + discovered_peripheral_tracker_ = absl::WrapUnique( + new DiscoveredPeripheralTracker(is_extended_advertisement_available, + /*start_fetch_executor=*/false)); + // Do not start the fetch executor since we will manually run the GATT + // fetch loop in the test. adapter_peripheral_ = std::make_unique(); adapter_central_ = std::make_unique(); ble_peripheral_ = std::make_unique(*adapter_peripheral_); @@ -350,6 +354,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, Uuid(kFastAdvertisementServiceUuid)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!fast_advertisement_bytes.Empty()) { @@ -385,6 +390,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, DctAdvertisementPeripheralDiscovered) { }, }, Uuid(kFastAdvertisementServiceUuid)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; advertisement_data.service_data.insert( @@ -424,6 +430,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, .legacy_device_discovered_cb = [&legacy_found_latch]() { legacy_found_latch.CountDown(); }}, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -464,6 +471,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, Uuid(kFastAdvertisementServiceUuid)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!fast_advertisement_bytes.Empty()) { @@ -535,6 +543,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, Uuid(kFastAdvertisementServiceUuid)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!fast_advertisement_bytes.Empty()) { @@ -579,6 +588,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, Uuid("FE3C")); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!fast_advertisement_bytes.Empty()) { @@ -622,6 +632,8 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, Uuid(kFastAdvertisementServiceUuid)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); + discovered_peripheral_tracker_->StartTracking( std::string(kServiceIdB), false, Pcp::kP2pPointToPoint, { @@ -681,6 +693,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -717,6 +730,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, bool fast_advertisement) { found_latch.CountDown(); }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -763,6 +777,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -809,6 +824,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -852,6 +868,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, bool fast_advertisement) { found_latch.CountDown(); }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -901,6 +918,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, Uuid(kFastAdvertisementServiceUuid)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -961,6 +979,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, bool fast_advertisement) { lost_latch.CountDown(); }, }, Uuid(kFastAdvertisementServiceUuid)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -1018,6 +1037,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, LostPeripheralForAdvertisementLost) { bool fast_advertisement) { lost_latch.CountDown(); }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -1096,6 +1116,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, bool fast_advertisement) { lost_latch_b.CountDown(); }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -1158,6 +1179,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, bool fast_advertisement) { lost_latch.CountDown(); }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -1212,6 +1234,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, InstantLostPeripheralForInstantOnLost) { bool fast_advertisement) { lost_latch.CountDown(); }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -1284,6 +1307,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -1360,6 +1384,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, bool fast_advertisement) { lost_latch.CountDown(); }, }, Uuid(kFastAdvertisementServiceUuid)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!fast_advertisement_bytes.Empty()) { @@ -1416,6 +1441,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, HandleDummyAdvertisement) { }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); FindAdvertisement(advertising_data, {}, fetch_latch); @@ -1452,6 +1478,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, SkipDummyAdvertisement) { }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); FindAdvertisement(advertising_data, {}, fetch_latch); @@ -1483,6 +1510,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, FetchGattAdvertisementInThread) { }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -1524,6 +1552,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -1571,6 +1600,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, {}); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { @@ -1625,6 +1655,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, }, bleutils::kCopresenceServiceUuid); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); // 1. First receive a GATT advertisement data, it should be skipped. api::ble::BleAdvertisementData advertisement_data{}; @@ -1675,7 +1706,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, SkipExpiredGattAdvertisement) { }, bleutils::kCopresenceServiceUuid); - // 1. First receive a GATT advertisement data, it should be skipped. + // 1. First receive a GATT advertisement data, it will expire after 15s. api::ble::BleAdvertisementData advertisement_data{}; if (!advertisement_header_bytes.Empty()) { advertisement_data.service_data.insert( @@ -1685,11 +1716,12 @@ TEST_P(DiscoveredPeripheralTrackerTest, SkipExpiredGattAdvertisement) { FindAdvertisementWithDelay(advertisement_data, {advertisement_bytes}, fetch_latch, kDefaultGattFetchDelay); - // 2. The GATT advertisement will be expired after 20 seconds. + // 2. The GATT advertisement is already queued and will be skipped. FindAdvertisement(advertisement_data, {advertisement_bytes}, fetch_latch); (*fake_clock)->FastForward(absl::Seconds(20)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); - // We should receive a client callback of a peripheral discovery. + // We should not receive a client callback of a peripheral discovery. fetch_latch.Await(kWaitDuration); EXPECT_FALSE(found_latch.Await(kWaitDuration).result()); EXPECT_EQ(GetFetchAdvertisementCallbackCount(), 0); @@ -1725,6 +1757,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, bleutils::kCopresenceServiceUuid); (*fake_clock)->FastForward(absl::Seconds(4)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); // 1. Received extended advertisement. api::ble::BleAdvertisementData extended_advertisement_data{}; @@ -1791,6 +1824,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, }, bleutils::kCopresenceServiceUuid); (*fake_clock)->FastForward(absl::Seconds(4)); + discovered_peripheral_tracker_->StartFetchExecutorForTesting(); // 1. Find peripheral A with GATT advertisement. api::ble::BleAdvertisementData advertisement_data_a{}; @@ -1824,8 +1858,6 @@ INSTANTIATE_TEST_SUITE_P( ::testing::Combine( /*is_extended_advertisement_available=*/testing::Bool())); -} // namespace - } // namespace mediums } // namespace connections } // namespace nearby