From 1097288e544608b2c4fe96977efc4cd8a8062463 Mon Sep 17 00:00:00 2001 From: Francis Tsui Date: Thu, 16 May 2024 14:21:03 -0700 Subject: [PATCH] Fix threading issues in NearbySharingServiceImpl. PiperOrigin-RevId: 634527776 --- internal/test/fake_account_manager.cc | 8 +- internal/test/fake_task_runner.cc | 11 +- sharing/BUILD | 4 +- sharing/nearby_connections_manager_factory.cc | 8 +- sharing/nearby_connections_manager_factory.h | 7 +- sharing/nearby_connections_manager_impl.cc | 43 ++- sharing/nearby_connections_manager_impl.h | 6 +- .../nearby_connections_manager_impl_test.cc | 16 +- sharing/nearby_sharing_service_factory.cc | 8 +- sharing/nearby_sharing_service_impl.cc | 7 +- sharing/nearby_sharing_service_impl.h | 6 +- sharing/nearby_sharing_service_impl_test.cc | 247 ++++++++++-------- 12 files changed, 210 insertions(+), 161 deletions(-) diff --git a/internal/test/fake_account_manager.cc b/internal/test/fake_account_manager.cc index ef2fdb8d..45b3c47c 100644 --- a/internal/test/fake_account_manager.cc +++ b/internal/test/fake_account_manager.cc @@ -35,9 +35,11 @@ void FakeAccountManager::Login( absl::AnyInvocable login_success_callback, absl::AnyInvocable login_failure_callback) { if (account_.has_value()) { - login_success_callback(*account_); UpdateCurrentUser(account_->id); NotifyLogin(account_->id); + // Invoke callback after all operations have been performed since test cases + // may rely on the callback for synchronization. + login_success_callback(*account_); return; } @@ -49,8 +51,10 @@ void FakeAccountManager::Logout( if (is_logout_success_) { std::string account_id = account_->id; SetAccount(std::nullopt); - logout_callback(absl::OkStatus()); NotifyLogout(account_id); + // Invoke callback after all operations have been performed since test cases + // may rely on the callback for synchronization. + logout_callback(absl::OkStatus()); return; } diff --git a/internal/test/fake_task_runner.cc b/internal/test/fake_task_runner.cc index 4d237869..e88c7394 100644 --- a/internal/test/fake_task_runner.cc +++ b/internal/test/fake_task_runner.cc @@ -24,7 +24,6 @@ #include "absl/synchronization/notification.h" #include "absl/time/clock.h" #include "absl/time/time.h" -#include "internal/platform/count_down_latch.h" #include "internal/platform/timer.h" #include "internal/test/fake_timer.h" @@ -68,13 +67,9 @@ void FakeTaskRunner::Sync() { } bool FakeTaskRunner::SyncWithTimeout(absl::Duration timeout) { - CountDownLatch latch(count_); - for (int i = 0; i < count_; ++i) { - PostTask([&] { latch.CountDown(); }); - } - - auto result = latch.Await(timeout); - return result.ok() && result.result(); + absl::Notification notification; + PostTask([&] { notification.Notify(); }); + return notification.WaitForNotificationWithTimeout(timeout); } bool FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Duration timeout) { diff --git a/sharing/BUILD b/sharing/BUILD index 18d02346..7d574f8e 100644 --- a/sharing/BUILD +++ b/sharing/BUILD @@ -324,15 +324,14 @@ cc_test( "//base:casts", "//connections:core_types", "//internal/account", - "//internal/analytics:mock_event_logger", "//internal/base:files", "//internal/flags:nearby_flags", "//internal/network:types", "//internal/network:url", + "//internal/platform:types", "//internal/platform/implementation:account_manager", "//internal/platform/implementation/g3", # fixdeps: keep "//internal/test", - "//proto:sharing_enums_cc_proto", "//sharing/certificates", "//sharing/certificates:test_support", "//sharing/common", @@ -352,7 +351,6 @@ cc_test( "//sharing/proto:enums_cc_proto", "//sharing/proto:share_cc_proto", "//sharing/proto:wire_format_cc_proto", - "//sharing/proto/analytics:sharing_log_cc_proto", "@com_github_protobuf_matchers//protobuf-matchers", "@com_google_absl//absl/base:core_headers", "@com_google_absl//absl/container:flat_hash_map", diff --git a/sharing/nearby_connections_manager_factory.cc b/sharing/nearby_connections_manager_factory.cc index 821a1db8..30e1ab09 100644 --- a/sharing/nearby_connections_manager_factory.cc +++ b/sharing/nearby_connections_manager_factory.cc @@ -18,6 +18,7 @@ #include "internal/analytics/event_logger.h" #include "internal/platform/device_info.h" +#include "internal/platform/task_runner.h" #include "sharing/internal/public/context.h" #include "sharing/nearby_connections_manager.h" #include "sharing/nearby_connections_manager_impl.h" @@ -28,11 +29,12 @@ namespace sharing { std::unique_ptr NearbyConnectionsManagerFactory::CreateConnectionsManager( - LinkType link_type, Context* context, - nearby::DeviceInfo& device_info, + LinkType link_type, nearby::TaskRunner* connections_callback_task_runner, + Context* context, nearby::DeviceInfo& device_info, nearby::analytics::EventLogger* event_logger) { return std::make_unique( - context, *context->GetConnectivityManager(), device_info, + connections_callback_task_runner, context, + *context->GetConnectivityManager(), device_info, std::make_unique(event_logger)); } diff --git a/sharing/nearby_connections_manager_factory.h b/sharing/nearby_connections_manager_factory.h index 80117a0e..d4d2212b 100644 --- a/sharing/nearby_connections_manager_factory.h +++ b/sharing/nearby_connections_manager_factory.h @@ -19,6 +19,7 @@ #include "internal/analytics/event_logger.h" #include "internal/platform/device_info.h" +#include "internal/platform/task_runner.h" #include "sharing/internal/public/context.h" #include "sharing/nearby_connections_manager.h" #include "sharing/nearby_sharing_service_factory.h" @@ -31,8 +32,12 @@ class NearbyConnectionsManagerFactory { using LinkType = NearbySharingServiceFactory::LinkType; // Return a singleton instance of NearbyConnectionsManagerFactory. + // |task_runner| is the thread when callbacks from NearbyConnection destined + // for NearSharingServiec need to be scheduled. This is usually the thread + // that NearbySharingService is running on. static std::unique_ptr CreateConnectionsManager( - NearbySharingServiceFactory::LinkType link_type, Context* context, + NearbySharingServiceFactory::LinkType link_type, + nearby::TaskRunner* connections_callback_task_runner, Context* context, nearby::DeviceInfo& device_info, nearby::analytics::EventLogger* event_logger = nullptr); diff --git a/sharing/nearby_connections_manager_impl.cc b/sharing/nearby_connections_manager_impl.cc index 0988a5f8..001e6cde 100644 --- a/sharing/nearby_connections_manager_impl.cc +++ b/sharing/nearby_connections_manager_impl.cc @@ -35,6 +35,7 @@ #include "internal/flags/nearby_flags.h" #include "internal/platform/device_info.h" #include "internal/platform/mutex_lock.h" +#include "internal/platform/task_runner.h" #include "sharing/advertisement.h" #include "sharing/common/nearby_share_enums.h" #include "sharing/constants.h" @@ -136,10 +137,12 @@ std::string MediumSelectionToString(const MediumSelection& mediums) { } // namespace NearbyConnectionsManagerImpl::NearbyConnectionsManagerImpl( + TaskRunner* connections_callback_task_runner, Context* context, ConnectivityManager& connectivity_manager, nearby::DeviceInfo& device_info, std::unique_ptr nearby_connections_service) - : context_(context), + : connections_callback_task_runner_(connections_callback_task_runner), + context_(context), connectivity_manager_(connectivity_manager), device_info_(device_info), nearby_connections_service_(std::move(nearby_connections_service)) {} @@ -195,15 +198,21 @@ void NearbyConnectionsManagerImpl::StartAdvertising( const ConnectionInfo& connection_info) { OnConnectionInitiated(endpoint_id, connection_info); }; - connection_listener.accepted_cb = [&](absl::string_view endpoint_id) { - OnConnectionAccepted(endpoint_id); + connection_listener.accepted_cb = [this](absl::string_view endpoint_id) { + connections_callback_task_runner_->PostTask( + [this, endpoint_id = std::string(endpoint_id)]() { + OnConnectionAccepted(endpoint_id); + }); }; connection_listener.rejected_cb = [&](absl::string_view endpoint_id, Status status) { OnConnectionRejected(endpoint_id, status); }; - connection_listener.disconnected_cb = [&](absl::string_view endpoint_id) { - OnDisconnected(endpoint_id); + connection_listener.disconnected_cb = [this](absl::string_view endpoint_id) { + connections_callback_task_runner_->PostTask( + [this, endpoint_id = std::string(endpoint_id)]() { + OnDisconnected(endpoint_id); + }); }; connection_listener.bandwidth_changed_cb = [&](absl::string_view endpoint_id, Medium medium) { @@ -377,15 +386,21 @@ void NearbyConnectionsManagerImpl::Connect( const ConnectionInfo& connection_info) { OnConnectionInitiated(endpoint_id, connection_info); }; - connection_listener.accepted_cb = [&](absl::string_view endpoint_id) { - OnConnectionAccepted(endpoint_id); + connection_listener.accepted_cb = [this](absl::string_view endpoint_id) { + connections_callback_task_runner_->PostTask( + [this, endpoint_id = std::string(endpoint_id)]() { + OnConnectionAccepted(endpoint_id); + }); }; connection_listener.rejected_cb = [&](absl::string_view endpoint_id, Status status) { OnConnectionRejected(endpoint_id, status); }; - connection_listener.disconnected_cb = [&](absl::string_view endpoint_id) { - OnDisconnected(endpoint_id); + connection_listener.disconnected_cb = [this](absl::string_view endpoint_id) { + connections_callback_task_runner_->PostTask( + [this, endpoint_id = std::string(endpoint_id)]() { + OnDisconnected(endpoint_id); + }); }; connection_listener.bandwidth_changed_cb = [&](absl::string_view endpoint_id, Medium medium) { @@ -467,7 +482,7 @@ void NearbyConnectionsManagerImpl::Disconnect(absl::string_view endpoint_id) { << " attempted over Nearby Connections with result: " << ConnectionsStatusToString(status); - context_->GetTaskRunner()->PostTask([&, endpoint_id]() { + connections_callback_task_runner_->PostTask([this, endpoint_id]() { OnDisconnected(endpoint_id); { MutexLock lock(&mutex_); @@ -663,8 +678,12 @@ void NearbyConnectionsManagerImpl::OnConnectionInitiated( }; payload_listener.payload_progress_cb = - [&](absl::string_view endpoint_id, const PayloadTransferUpdate& update) { - OnPayloadTransferUpdate(endpoint_id, update); + [this](absl::string_view endpoint_id, + const PayloadTransferUpdate& update) { + connections_callback_task_runner_->PostTask( + [this, endpoint_id = std::string(endpoint_id), update = update]() { + OnPayloadTransferUpdate(endpoint_id, update); + }); }; nearby_connections_service_->AcceptConnection( diff --git a/sharing/nearby_connections_manager_impl.h b/sharing/nearby_connections_manager_impl.h index 459b7e8b..3944870b 100644 --- a/sharing/nearby_connections_manager_impl.h +++ b/sharing/nearby_connections_manager_impl.h @@ -29,6 +29,8 @@ #include "absl/strings/string_view.h" #include "internal/platform/device_info.h" #include "internal/platform/mutex.h" +#include "internal/platform/task_runner.h" +#include "internal/platform/timer.h" #include "sharing/common/nearby_share_enums.h" #include "sharing/internal/public/connectivity_manager.h" #include "sharing/internal/public/context.h" @@ -45,7 +47,8 @@ namespace sharing { class NearbyConnectionsManagerImpl : public NearbyConnectionsManager { public: explicit NearbyConnectionsManagerImpl( - Context* context, nearby::ConnectivityManager& connectivity_manager, + nearby::TaskRunner* connections_callback_task_runner, Context* context, + nearby::ConnectivityManager& connectivity_manager, nearby::DeviceInfo& device_info, std::unique_ptr nearby_connections_service); ~NearbyConnectionsManagerImpl() override; @@ -124,6 +127,7 @@ class NearbyConnectionsManagerImpl : public NearbyConnectionsManager { void SendWithoutDelay(absl::string_view endpoint_id, std::unique_ptr payload); + nearby::TaskRunner* const connections_callback_task_runner_; Context* const context_; nearby::ConnectivityManager& connectivity_manager_; nearby::DeviceInfo& device_info_; diff --git a/sharing/nearby_connections_manager_impl_test.cc b/sharing/nearby_connections_manager_impl_test.cc index 6d469cf7..75ceac5a 100644 --- a/sharing/nearby_connections_manager_impl_test.cc +++ b/sharing/nearby_connections_manager_impl_test.cc @@ -131,7 +131,6 @@ class MockPayloadStatusListener class NearbyConnectionsManagerImplTest : public testing::Test { public: void SetUp() override { - FakeTaskRunner::ResetPendingTasksCount(); NearbyFlags::GetInstance().OverrideBoolFlagValue( config_package_nearby::nearby_sharing_feature::kEnableMediumWebRtc, false); @@ -145,8 +144,8 @@ class NearbyConnectionsManagerImplTest : public testing::Test { nearby_connections_manager_ = std::make_unique( - &fake_context_, fake_connectivity_manager_, fake_device_info_, - std::move(nearby_connections_service)); + &fake_task_runner_, &fake_context_, fake_connectivity_manager_, + fake_device_info_, std::move(nearby_connections_service)); } void TearDown() override { @@ -157,7 +156,7 @@ class NearbyConnectionsManagerImplTest : public testing::Test { config_package_nearby::nearby_sharing_feature::kEnableMediumWifiLan, true); - FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Seconds(1)); + fake_task_runner_.SyncWithTimeout(absl::Seconds(1)); } void SetConnectionType(ConnectivityManager::ConnectionType connection_type) { @@ -169,8 +168,7 @@ class NearbyConnectionsManagerImplTest : public testing::Test { } void Sync() { - EXPECT_TRUE( - FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Seconds(1))); + EXPECT_TRUE(fake_task_runner_.SyncWithTimeout(absl::Seconds(1))); } protected: @@ -289,6 +287,7 @@ class NearbyConnectionsManagerImplTest : public testing::Test { TransportType::kHighQuality, [&](NearbyConnection* connection, Status status) { nearby_connection = connection; + notification.Notify(); }); EXPECT_TRUE(request_connection_notification.WaitForNotificationWithTimeout( @@ -304,7 +303,6 @@ class NearbyConnectionsManagerImplTest : public testing::Test { payload_listener_remote = std::move(listener); std::move(callback)(Status::kSuccess); - notification.Notify(); }); ConnectionInfo connection_info; @@ -440,6 +438,7 @@ class NearbyConnectionsManagerImplTest : public testing::Test { } nearby::FakeContext fake_context_; + FakeTaskRunner fake_task_runner_{fake_context_.fake_clock(), 1}; nearby::FakeDeviceInfo fake_device_info_; nearby::FakeConnectivityManager fake_connectivity_manager_; bool should_use_web_rtc_ = false; @@ -741,7 +740,7 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectDisconnected) { NearbyConnection* nearby_connection = Connect(connection_listener_remote, payload_listener_remote, ConnectionResponse::kDisconnected); - EXPECT_TRUE(FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Seconds(1))); + EXPECT_TRUE(fake_task_runner_.SyncWithTimeout(absl::Seconds(1))); EXPECT_FALSE(nearby_connection); EXPECT_FALSE(nearby_connections_manager_->GetRawAuthenticationToken( kRemoteEndpointId)); @@ -837,6 +836,7 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectReadAfterAppend) { absl::Notification read_notification; nearby_connection->Read([&](std::optional> bytes) { + ASSERT_TRUE(bytes.has_value()); EXPECT_EQ(bytes, byte_payload); read_notification.Notify(); }); diff --git a/sharing/nearby_sharing_service_factory.cc b/sharing/nearby_sharing_service_factory.cc index b90e94bd..0fe7dbd5 100644 --- a/sharing/nearby_sharing_service_factory.cc +++ b/sharing/nearby_sharing_service_factory.cc @@ -18,6 +18,7 @@ #include #include "internal/analytics/event_logger.h" +#include "internal/platform/task_runner.h" #include "sharing/internal/api/sharing_platform.h" #include "sharing/internal/public/context_impl.h" #include "sharing/nearby_connections_manager_factory.h" @@ -48,12 +49,15 @@ NearbySharingService* NearbySharingServiceFactory::CreateSharingService( std::make_unique(sharing_platform); event_logger_ = event_logger; decoder_ = std::make_unique(); + std::unique_ptr service_thread = + context_->CreateSequencedTaskRunner(); auto nearby_connections_manager = NearbyConnectionsManagerFactory::CreateConnectionsManager( - link_type, context_.get(), sharing_platform.GetDeviceInfo(), - event_logger_); + link_type, service_thread.get(), context_.get(), + sharing_platform.GetDeviceInfo(), event_logger_); nearby_sharing_service_ = std::make_unique( + std::move(service_thread), context_.get(), sharing_platform, decoder_.get(), std::move(nearby_connections_manager), event_logger_); diff --git a/sharing/nearby_sharing_service_impl.cc b/sharing/nearby_sharing_service_impl.cc index bd7cbc81..753b842b 100644 --- a/sharing/nearby_sharing_service_impl.cc +++ b/sharing/nearby_sharing_service_impl.cc @@ -47,6 +47,7 @@ #include "internal/platform/device_info.h" #include "internal/platform/implementation/account_manager.h" #include "internal/platform/implementation/device_info.h" +#include "internal/platform/task_runner.h" #include "proto/sharing_enums.pb.h" #include "sharing/advertisement.h" #include "sharing/analytics/analytics_information.h" @@ -152,11 +153,13 @@ constexpr absl::string_view kProfileRelativePath = "Google/Nearby/Sharing"; } // namespace NearbySharingServiceImpl::NearbySharingServiceImpl( + std::unique_ptr service_thread, Context* context, SharingPlatform& sharing_platform, NearbySharingDecoder* decoder, std::unique_ptr nearby_connections_manager, nearby::analytics::EventLogger* event_logger) - : context_(context), + : service_thread_(std::move(service_thread)), + context_(context), device_info_(sharing_platform.GetDeviceInfo()), preference_manager_(sharing_platform.GetPreferenceManager()), account_manager_(sharing_platform.GetAccountManager()), @@ -189,8 +192,6 @@ NearbySharingServiceImpl::NearbySharingServiceImpl( NL_DCHECK(decoder_); NL_DCHECK(nearby_connections_manager_); - service_thread_ = context_->CreateSequencedTaskRunner(); - certificate_download_during_discovery_timer_ = context_->CreateTimer(); on_network_changed_delay_timer_ = context_->CreateTimer(); mutual_acceptance_timeout_alarm_ = context_->CreateTimer(); diff --git a/sharing/nearby_sharing_service_impl.h b/sharing/nearby_sharing_service_impl.h index e943eba0..e2073df3 100644 --- a/sharing/nearby_sharing_service_impl.h +++ b/sharing/nearby_sharing_service_impl.h @@ -110,6 +110,7 @@ class NearbySharingServiceImpl public: NearbySharingServiceImpl( + std::unique_ptr service_thread, Context* context, nearby::sharing::api::SharingPlatform& sharing_platform, NearbySharingDecoder* decoder, std::unique_ptr nearby_connections_manager, @@ -483,6 +484,8 @@ class NearbySharingServiceImpl // Update file path for the file attachment. void UpdateFilePath(ShareTarget& share_target); + // Used to run nearby sharing service APIs. + std::unique_ptr service_thread_; Context* const context_; nearby::DeviceInfo& device_info_; nearby::sharing::api::PreferenceManager& preference_manager_; @@ -631,9 +634,6 @@ class NearbySharingServiceImpl // Called when cleanup for ARC is needed as part of the transfer. std::function arc_transfer_cleanup_callback_; - // Used to run nearby sharing service APIs. - std::unique_ptr service_thread_ = nullptr; - // Shouldn't schedule new task after shutting down, and skip task if the // object is null. std::shared_ptr is_shutting_down_ = nullptr; diff --git a/sharing/nearby_sharing_service_impl_test.cc b/sharing/nearby_sharing_service_impl_test.cc index 90554104..0d93b75f 100644 --- a/sharing/nearby_sharing_service_impl_test.cc +++ b/sharing/nearby_sharing_service_impl_test.cc @@ -42,7 +42,6 @@ #include "absl/types/span.h" #include "internal/account/account_manager_impl.h" #include "internal/flags/nearby_flags.h" -#include "internal/network/http_client_factory.h" #include "internal/test/fake_account_manager.h" #include "internal/test/fake_device_info.h" #include "internal/test/fake_task_runner.h" @@ -366,7 +365,6 @@ class NearbySharingServiceImplTest : public testing::Test { ~NearbySharingServiceImplTest() override = default; void SetUp() override { - FakeTaskRunner::ResetPendingTasksCount(); ON_CALL(mock_sharing_platform_, GetDeviceInfo) .WillByDefault(ReturnRef(fake_device_info_)); ON_CALL(mock_sharing_platform_, GetPreferenceManager) @@ -399,12 +397,15 @@ class NearbySharingServiceImplTest : public testing::Test { true); prefs::RegisterNearbySharingPrefs(preference_manager_); + auto fake_task_runner = + std::make_unique(fake_context_.fake_clock(), 1); + sharing_service_task_runner_ = fake_task_runner.get(); SetBluetoothIsPresent(true); SetBluetoothIsPowered(true); SetScreenLocked(false); SetConnectionType(ConnectionType::kWifi); - service_ = CreateService(); + service_ = CreateService(std::move(fake_task_runner)); } void TearDown() override { @@ -432,15 +433,14 @@ class NearbySharingServiceImplTest : public testing::Test { connectivity_manager->SetConnectionType(type); } - std::unique_ptr CreateService() { + std::unique_ptr CreateService( + std::unique_ptr task_runner) { preference_manager_.SetBoolean(prefs::kNearbySharingEnabledName, true); fake_nearby_connections_manager_ = new FakeNearbyConnectionsManager(); - auto service = std::make_unique( - &fake_context_, mock_sharing_platform_, &fake_decoder_, - absl::WrapUnique(fake_nearby_connections_manager_)); - - return service; + return std::make_unique( + std::move(task_runner), &fake_context_, mock_sharing_platform_, + &fake_decoder_, absl::WrapUnique(fake_nearby_connections_manager_)); } void SetVisibility(DeviceVisibility visibility) { @@ -854,7 +854,7 @@ class NearbySharingServiceImplTest : public testing::Test { Frame GetWrittenFrame() { EXPECT_TRUE( - FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Seconds(2))); + sharing_service_task_runner_->SyncWithTimeout(absl::Seconds(2))); std::vector data = connection_.GetWrittenData(); Frame frame; frame.ParseFromArray(data.data(), data.size()); @@ -1051,14 +1051,17 @@ class NearbySharingServiceImplTest : public testing::Test { ExpectTransferUpdates(transfer_callback, target, {TransferMetadata::Status::kComplete}, [] {}); - auto payload_transfer_update = std::make_unique( - info.payload_id, PayloadStatus::kSuccess, - /*total_bytes=*/strlen(kTextPayload), - /*bytes_transferred=*/strlen(kTextPayload)); - if (auto listener = info.listener.lock()) { - listener->OnStatusUpdate(std::move(payload_transfer_update), - /*upgraded_medium=*/std::nullopt); - } + sharing_service_task_runner_->PostTask([info = info]() { + auto payload_transfer_update = std::make_unique( + info.payload_id, PayloadStatus::kSuccess, + /*total_bytes=*/strlen(kTextPayload), + /*bytes_transferred=*/strlen(kTextPayload)); + if (auto listener = info.listener.lock()) { + listener->OnStatusUpdate(std::move(payload_transfer_update), + /*upgraded_medium=*/std::nullopt); + } + }); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kWaitTimeout)); } std::unique_ptr GetCurrentAdvertisement() { @@ -1138,15 +1141,17 @@ class NearbySharingServiceImplTest : public testing::Test { progress_notification.Notify(); })); - PayloadTransferUpdate payload = - PayloadTransferUpdate(id, PayloadStatus::kSuccess, - /*total_bytes=*/kPayloadSize, - /*bytes_transferred=*/kPayloadSize); - if (auto locked_listener = listener.lock()) { - locked_listener->OnStatusUpdate( - std::make_unique(payload), - /*upgraded_medium=*/std::nullopt); - } + sharing_service_task_runner_->PostTask([id, listener = listener]() { + PayloadTransferUpdate payload = + PayloadTransferUpdate(id, PayloadStatus::kSuccess, + /*total_bytes=*/kPayloadSize, + /*bytes_transferred=*/kPayloadSize); + if (auto locked_listener = listener.lock()) { + locked_listener->OnStatusUpdate( + std::make_unique(payload), + /*upgraded_medium=*/std::nullopt); + } + }); EXPECT_TRUE( progress_notification.WaitForNotificationWithTimeout(kWaitTimeout)); @@ -1176,19 +1181,21 @@ class NearbySharingServiceImplTest : public testing::Test { success_notification.Notify(); })); - std::weak_ptr listener = - fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( - kFilePayloadId); + sharing_service_task_runner_->PostTask([this]() { + std::weak_ptr listener = + fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( + kFilePayloadId); - PayloadTransferUpdate payload = - PayloadTransferUpdate(kFilePayloadId, PayloadStatus::kSuccess, - /*total_bytes=*/kPayloadSize, - /*bytes_transferred=*/kPayloadSize); - if (auto locked_listener = listener.lock()) { - locked_listener->OnStatusUpdate( - std::make_unique(payload), - /*upgraded_medium=*/std::nullopt); - } + PayloadTransferUpdate payload = + PayloadTransferUpdate(kFilePayloadId, PayloadStatus::kSuccess, + /*total_bytes=*/kPayloadSize, + /*bytes_transferred=*/kPayloadSize); + if (auto locked_listener = listener.lock()) { + locked_listener->OnStatusUpdate( + std::make_unique(payload), + /*upgraded_medium=*/std::nullopt); + } + }); EXPECT_TRUE( success_notification.WaitForNotificationWithTimeout(kWaitTimeout)); @@ -1206,7 +1213,8 @@ class NearbySharingServiceImplTest : public testing::Test { void FlushTesting() { absl::SleepFor(absl::Milliseconds(200)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Milliseconds(200)); + EXPECT_TRUE( + sharing_service_task_runner_->SyncWithTimeout(absl::Milliseconds(200))); } void SetDiskSpace(size_t size) { @@ -1258,6 +1266,7 @@ class NearbySharingServiceImplTest : public testing::Test { std::unique_ptr service_; int expect_transfer_updates_count_ = 0; std::function expect_transfer_updates_callback_; + FakeTaskRunner* sharing_service_task_runner_ = nullptr; }; struct ValidSendSurfaceTestData { @@ -1484,7 +1493,7 @@ TEST_F(NearbySharingServiceImplTest, FastInitiationScanning_StartAndStop) { // Trigger a call to StartFastInitiationScanning(). SetBluetoothIsPowered(true); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_EQ(fast_initiation->StartScanningCount(), 2); EXPECT_EQ(fast_initiation->StopScanningCount(), 1); } @@ -2786,11 +2795,6 @@ TEST_F(NearbySharingServiceImplTest, // Deliberately not calling SetIncomingPayload() for text payloads to check // for failure condition. - std::weak_ptr listener = - fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( - id); - ASSERT_FALSE(listener.expired()); - absl::Notification progress_notification; EXPECT_CALL(callback, OnTransferUpdate(testing::_, testing::_)) .WillOnce(testing::Invoke([&](const ShareTarget& share_target, @@ -2800,14 +2804,21 @@ TEST_F(NearbySharingServiceImplTest, progress_notification.Notify(); })); - auto payload = std::make_unique( - id, PayloadStatus::kSuccess, - /*total_bytes=*/kPayloadSize, - /*bytes_transferred=*/kPayloadSize); - if (auto locked_listener = listener.lock()) { - locked_listener->OnStatusUpdate(std::move(payload), - /*upgraded_medium=*/std::nullopt); - } + sharing_service_task_runner_->PostTask([this, id]() { + std::weak_ptr listener = + fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( + id); + ASSERT_FALSE(listener.expired()); + + auto payload = std::make_unique( + id, PayloadStatus::kSuccess, + /*total_bytes=*/kPayloadSize, + /*bytes_transferred=*/kPayloadSize); + if (auto locked_listener = listener.lock()) { + locked_listener->OnStatusUpdate(std::move(payload), + /*upgraded_medium=*/std::nullopt); + } + }); EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout)); FastForward(kMinProgressUpdateFrequency); } @@ -2826,19 +2837,21 @@ TEST_F(NearbySharingServiceImplTest, success_notification.Notify(); })); - std::weak_ptr listener = - fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( - kFilePayloadId); - ASSERT_FALSE(listener.expired()); + sharing_service_task_runner_->PostTask([this]() { + std::weak_ptr listener = + fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( + kFilePayloadId); + ASSERT_FALSE(listener.expired()); - auto payload = std::make_unique( - kFilePayloadId, PayloadStatus::kSuccess, - /*total_bytes=*/kPayloadSize, - /*bytes_transferred=*/kPayloadSize); - if (auto locked_listener = listener.lock()) { - locked_listener->OnStatusUpdate(std::move(payload), - /*upgraded_medium=*/std::nullopt); - } + auto payload = std::make_unique( + kFilePayloadId, PayloadStatus::kSuccess, + /*total_bytes=*/kPayloadSize, + /*bytes_transferred=*/kPayloadSize); + if (auto locked_listener = listener.lock()) { + locked_listener->OnStatusUpdate(std::move(payload), + /*upgraded_medium=*/std::nullopt); + } + }); EXPECT_TRUE( success_notification.WaitForNotificationWithTimeout(kWaitTimeout)); @@ -2848,7 +2861,7 @@ TEST_F(NearbySharingServiceImplTest, // File deletion runs in a ThreadPool. EXPECT_TRUE( - FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Milliseconds(200))); + sharing_service_task_runner_->SyncWithTimeout(absl::Milliseconds(200))); // To avoid UAF in OnIncomingTransferUpdate(). UnregisterReceiveSurface(&callback); @@ -2875,11 +2888,6 @@ TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadFailed) { EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout)); - std::weak_ptr listener = - fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( - kFilePayloadId); - ASSERT_FALSE(listener.expired()); - absl::Notification failure_notification; EXPECT_CALL(callback, OnTransferUpdate(testing::_, testing::_)) .WillOnce(testing::Invoke( @@ -2893,14 +2901,21 @@ TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadFailed) { failure_notification.Notify(); })); - auto payload = std::make_unique( - kFilePayloadId, PayloadStatus::kFailure, - /*total_bytes=*/kPayloadSize, - /*bytes_transferred=*/kPayloadSize); - if (auto locked_listener = listener.lock()) { - locked_listener->OnStatusUpdate(std::move(payload), - /*upgraded_medium=*/std::nullopt); - } + sharing_service_task_runner_->PostTask([this]() { + std::weak_ptr listener = + fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( + kFilePayloadId); + ASSERT_FALSE(listener.expired()); + + auto payload = std::make_unique( + kFilePayloadId, PayloadStatus::kFailure, + /*total_bytes=*/kPayloadSize, + /*bytes_transferred=*/kPayloadSize); + if (auto locked_listener = listener.lock()) { + locked_listener->OnStatusUpdate(std::move(payload), + /*upgraded_medium=*/std::nullopt); + } + }); EXPECT_TRUE( failure_notification.WaitForNotificationWithTimeout(kWaitTimeout)); @@ -2911,7 +2926,7 @@ TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadFailed) { // File deletion runs in a ThreadPool. EXPECT_TRUE( - FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Milliseconds(200))); + sharing_service_task_runner_->SyncWithTimeout(absl::Milliseconds(200))); // To avoid UAF in OnIncomingTransferUpdate(). UnregisterReceiveSurface(&callback); @@ -2938,11 +2953,6 @@ TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadCancelled) { EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout)); - std::weak_ptr listener = - fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( - kFilePayloadId); - ASSERT_FALSE(listener.expired()); - absl::Notification failure_notification; EXPECT_CALL(callback, OnTransferUpdate(testing::_, testing::_)) .WillOnce(testing::Invoke( @@ -2956,14 +2966,21 @@ TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadCancelled) { failure_notification.Notify(); })); - auto payload = std::make_unique( - kFilePayloadId, PayloadStatus::kCanceled, - /*total_bytes=*/kPayloadSize, - /*bytes_transferred=*/kPayloadSize); - if (auto locked_listener = listener.lock()) { - locked_listener->OnStatusUpdate(std::move(payload), - /*upgraded_medium=*/std::nullopt); - } + sharing_service_task_runner_->PostTask([this]() { + std::weak_ptr listener = + fake_nearby_connections_manager_->GetRegisteredPayloadStatusListener( + kFilePayloadId); + ASSERT_FALSE(listener.expired()); + + auto payload = std::make_unique( + kFilePayloadId, PayloadStatus::kCanceled, + /*total_bytes=*/kPayloadSize, + /*bytes_transferred=*/kPayloadSize); + if (auto locked_listener = listener.lock()) { + locked_listener->OnStatusUpdate(std::move(payload), + /*upgraded_medium=*/std::nullopt); + } + }); EXPECT_TRUE( failure_notification.WaitForNotificationWithTimeout(kWaitTimeout)); @@ -2973,7 +2990,7 @@ TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadCancelled) { // File deletion runs in a ThreadPool. EXPECT_TRUE( - FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Milliseconds(200))); + sharing_service_task_runner_->SyncWithTimeout(absl::Milliseconds(200))); // To avoid UAF in OnIncomingTransferUpdate(). UnregisterReceiveSurface(&callback); @@ -3164,7 +3181,7 @@ TEST_F(NearbySharingServiceImplTest, // Ensure that the messages sent by ProcessLatestPublicCertificateDecryption // are processed prior to checking if connection is closed. EXPECT_TRUE( - FakeTaskRunner::WaitForRunningTasksWithTimeout(absl::Milliseconds(200))); + sharing_service_task_runner_->SyncWithTimeout(absl::Milliseconds(200))); EXPECT_TRUE(connection_.IsClosed()); // To avoid UAF in OnIncomingTransferUpdate(). @@ -3937,13 +3954,13 @@ TEST_F(NearbySharingServiceImplTest, certificate_manager()->set_next_salt({0x00, 0x02}); certificate_manager()->NotifyPrivateCertificatesChanged(); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_TRUE(fake_nearby_connections_manager_->IsAdvertising()); auto endpoint_info_rotated = fake_nearby_connections_manager_->advertising_endpoint_info(); EXPECT_NE(endpoint_info_initial, endpoint_info_rotated); UnregisterReceiveSurface(&callback); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); } TEST_F(NearbySharingServiceImplTest, OrderedEndpointDiscoveryEvents) { @@ -4458,7 +4475,7 @@ TEST_F(NearbySharingServiceImplTest, ObserveAccountLoginAndLogout) { }); EXPECT_TRUE(logout_notification.WaitForNotificationWithTimeout(kWaitTimeout)); service_->GetAccountManager()->RemoveObserver(&account_observer); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); } TEST_F(NearbySharingServiceImplTest, LoginAndLogoutShouldResetSettings) { @@ -4481,7 +4498,7 @@ TEST_F(NearbySharingServiceImplTest, LoginAndLogoutShouldResetSettings) { }, [](absl::Status status) {}); ASSERT_TRUE(login_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_TRUE(service_->GetSettings()->GetIsAnalyticsEnabled()); ASSERT_TRUE(service_->GetAccountManager()->GetCurrentAccount().has_value()); EXPECT_EQ(service_->GetAccountManager()->GetCurrentAccount()->id, @@ -4497,7 +4514,7 @@ TEST_F(NearbySharingServiceImplTest, LoginAndLogoutShouldResetSettings) { absl::SleepFor(absl::Milliseconds(100)); EXPECT_FALSE(service_->GetSettings()->GetIsAnalyticsEnabled()); EXPECT_FALSE(service_->GetAccountManager()->GetCurrentAccount().has_value()); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); } TEST_F(NearbySharingServiceImplTest, @@ -4525,11 +4542,11 @@ TEST_F(NearbySharingServiceImplTest, }, [](absl::Status status) {}); ASSERT_TRUE(login_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_EQ(service_->GetSettings()->GetVisibility(), DeviceVisibility::DEVICE_VISIBILITY_ALL_CONTACTS); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); } TEST_F(NearbySharingServiceImplTest, LogoutShouldNotResetOnboarding) { @@ -4552,7 +4569,7 @@ TEST_F(NearbySharingServiceImplTest, LogoutShouldNotResetOnboarding) { }, [](absl::Status status) {}); ASSERT_TRUE(login_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_FALSE(service_->GetSettings()->IsOnboardingComplete()); // Logout user. @@ -4562,7 +4579,7 @@ TEST_F(NearbySharingServiceImplTest, LogoutShouldNotResetOnboarding) { logout_notification.Notify(); }); EXPECT_TRUE(logout_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_FALSE(service_->GetSettings()->IsOnboardingComplete()); // Complete onboarding. @@ -4578,7 +4595,7 @@ TEST_F(NearbySharingServiceImplTest, LogoutShouldNotResetOnboarding) { }, [](absl::Status status) {}); ASSERT_TRUE(login2_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_TRUE(service_->GetSettings()->IsOnboardingComplete()); // Logout user. @@ -4589,9 +4606,9 @@ TEST_F(NearbySharingServiceImplTest, LogoutShouldNotResetOnboarding) { }); EXPECT_TRUE( logout2_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_TRUE(service_->GetSettings()->IsOnboardingComplete()); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); } TEST_F(NearbySharingServiceImplTest, LogoutShouldSetValidVisibility) { @@ -4611,7 +4628,7 @@ TEST_F(NearbySharingServiceImplTest, LogoutShouldSetValidVisibility) { }, [](absl::Status status) {}); ASSERT_TRUE(login_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); // Set visibility. service_->GetSettings()->SetIsReceiving(true); @@ -4625,7 +4642,7 @@ TEST_F(NearbySharingServiceImplTest, LogoutShouldSetValidVisibility) { logout_notification.Notify(); }); EXPECT_TRUE(logout_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_FALSE(service_->GetSettings()->GetIsReceiving()); // Login user. @@ -4638,7 +4655,7 @@ TEST_F(NearbySharingServiceImplTest, LogoutShouldSetValidVisibility) { }, [](absl::Status status) {}); ASSERT_TRUE(login2_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); // Set visibility. service_->GetSettings()->SetIsReceiving(true); @@ -4653,11 +4670,11 @@ TEST_F(NearbySharingServiceImplTest, LogoutShouldSetValidVisibility) { }); EXPECT_TRUE( logout2_notification.WaitForNotificationWithTimeout(kWaitTimeout)); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_TRUE(service_->GetSettings()->GetIsReceiving()); EXPECT_EQ(service_->GetSettings()->GetVisibility(), DeviceVisibility::DEVICE_VISIBILITY_EVERYONE); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); } TEST_F(NearbySharingServiceImplTest, LoginAndLogoutNoStopRunningSurfaces) { @@ -4692,7 +4709,7 @@ TEST_F(NearbySharingServiceImplTest, LoginAndLogoutNoStopRunningSurfaces) { }); EXPECT_TRUE(logout_notification.WaitForNotificationWithTimeout(kWaitTimeout)); UnregisterSendSurface(&transfer_callback, &discovery_callback); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); } TEST_F(NearbySharingServiceImplTest, @@ -4725,7 +4742,7 @@ TEST_F(NearbySharingServiceImplTest, service_->GetSettings()->SetIsReceiving(true); NearbySharingService::StatusCodes result = RegisterReceiveSurface( &callback, NearbySharingService::ReceiveSurfaceState::kBackground); - FakeTaskRunner::WaitForRunningTasksWithTimeout(kTaskWaitTimeout); + EXPECT_TRUE(sharing_service_task_runner_->SyncWithTimeout(kTaskWaitTimeout)); EXPECT_EQ(result, NearbySharingService::StatusCodes::kOk); EXPECT_TRUE(fake_nearby_connections_manager_->IsAdvertising()); UnregisterReceiveSurface(&callback);