TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-14 14:46:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix threading in NearbyFileHandler.

Browse Source 
PiperOrigin-RevId: 633678194
This commit is contained in:
Francis Tsui
2024-05-14 12:57:35 -07:00
committed by Copybara-Service
parent 1c8395eede
commit 7840f93181
11 changed files with 12 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sharing/fake_nearby_connections_manager.cc
-29
View File
@@ -139,22 +139,6 @@ void FakeNearbyConnectionsManager::RegisterPayloadStatusListener(
payload_status_listeners_[payload_id] = listener;
}
void FakeNearbyConnectionsManager::RegisterPayloadPath(
int64_t payload_id, const std::filesystem::path& file_path,
ConnectionsCallback callback) {
NL_DCHECK(!is_shutdown());
registered_payload_paths_[payload_id] = file_path;
auto it = payload_path_status_.find(payload_id);
if (it == payload_path_status_.end()) {
std::move(callback)(nearby::sharing::Status::kPayloadUnknown);
return;
}
std::move(callback)(it->second);
}
Payload* FakeNearbyConnectionsManager::GetIncomingPayload(int64_t payload_id) {
NL_DCHECK(!is_shutdown());
absl::MutexLock lock(&incoming_payloads_mutex_);
@@ -248,11 +232,6 @@ bool FakeNearbyConnectionsManager::DidUpgradeBandwidth(
upgrade_bandwidth_endpoint_ids_.end();
}
void FakeNearbyConnectionsManager::SetPayloadPathStatus(
int64_t payload_id, ConnectionsStatus status) {
payload_path_status_[payload_id] = status;
}
std::weak_ptr<FakeNearbyConnectionsManager::PayloadStatusListener>
FakeNearbyConnectionsManager::GetRegisteredPayloadStatusListener(
int64_t payload_id) {
@@ -273,14 +252,6 @@ bool FakeNearbyConnectionsManager::WasPayloadCanceled(
return absl::c_linear_search(canceled_payload_ids_, payload_id);
}
std::optional<std::filesystem::path>
FakeNearbyConnectionsManager::GetRegisteredPayloadPath(int64_t payload_id) {
auto it = registered_payload_paths_.find(payload_id);
if (it == registered_payload_paths_.end()) return std::nullopt;
return it->second;
}
void FakeNearbyConnectionsManager::CleanupForProcessStopped() {
absl::MutexLock lock(&listener_mutex_);
advertising_listener_ = nullptr;
sharing/fake_nearby_connections_manager.h
-8
View File
@@ -65,9 +65,6 @@ class FakeNearbyConnectionsManager : public NearbyConnectionsManager {
void RegisterPayloadStatusListener(
int64_t payload_id,
std::weak_ptr<PayloadStatusListener> listener) override;
void RegisterPayloadPath(int64_t payload_id,
const std::filesystem::path& file_path,
ConnectionsCallback callback) override;
Payload* GetIncomingPayload(int64_t payload_id) override;
void Cancel(int64_t payload_id) override;
void ClearIncomingPayloads() override;
@@ -92,12 +89,9 @@ class FakeNearbyConnectionsManager : public NearbyConnectionsManager {
bool IsAdvertising() const;
bool IsDiscovering() const;
bool DidUpgradeBandwidth(absl::string_view endpoint_id) const;
void SetPayloadPathStatus(int64_t payload_id, ConnectionsStatus status);
std::weak_ptr<PayloadStatusListener> GetRegisteredPayloadStatusListener(
int64_t payload_id);
void SetIncomingPayload(int64_t payload_id, std::unique_ptr<Payload> payload);
std::optional<std::filesystem::path> GetRegisteredPayloadPath(
int64_t payload_id);
bool WasPayloadCanceled(int64_t payload_id) const;
void CleanupForProcessStopped();
ConnectionsCallback GetStartAdvertisingCallback();
@@ -175,13 +169,11 @@ class FakeNearbyConnectionsManager : public NearbyConnectionsManager {
// Maps endpoint_id to endpoint_info.
std::map<std::string, std::vector<uint8_t>> connection_endpoint_infos_;
std::map<int64_t, ConnectionsStatus> payload_path_status_;
std::map<int64_t, std::weak_ptr<PayloadStatusListener>>
payload_status_listeners_;
absl::Mutex incoming_payloads_mutex_;
std::map<int64_t, std::unique_ptr<Payload>> incoming_payloads_
ABSL_GUARDED_BY(incoming_payloads_mutex_);
std::map<int64_t, std::filesystem::path> registered_payload_paths_;
absl::flat_hash_set<std::filesystem::path> file_paths_to_delete_;
std::string Dump() const override;
};
sharing/nearby_connections_manager.h
-5
View File
@@ -139,11 +139,6 @@ class NearbyConnectionsManager {
virtual void RegisterPayloadStatusListener(
int64_t payload_id, std::weak_ptr<PayloadStatusListener> listener) = 0;
// Register a `file_path` for receiving incoming payload with `payload_id`.
virtual void RegisterPayloadPath(int64_t payload_id,
const std::filesystem::path& file_path,
ConnectionsCallback callback) = 0;
// Gets the payload associated with `payload_id` if available.
virtual Payload* GetIncomingPayload(int64_t payload_id) = 0;
sharing/nearby_connections_manager_impl.cc
-10
View File
@@ -526,16 +526,6 @@ void NearbyConnectionsManagerImpl::RegisterPayloadStatusListener(
payload_status_listeners_.insert_or_assign(payload_id, listener);
}
void NearbyConnectionsManagerImpl::RegisterPayloadPath(
int64_t payload_id, const std::filesystem::path& file_path,
ConnectionsCallback callback) {
NL_DCHECK(!file_path.empty());
// Create file is put into Nearby Connections, don't need to create file in
// Nearby Sharing.
callback(Status::kSuccess);
}
Payload* NearbyConnectionsManagerImpl::GetIncomingPayload(int64_t payload_id) {
MutexLock lock(&mutex_);
auto it = incoming_payloads_.find(payload_id);
sharing/nearby_connections_manager_impl.h
-3
View File
@@ -74,9 +74,6 @@ class NearbyConnectionsManagerImpl : public NearbyConnectionsManager {
void RegisterPayloadStatusListener(
int64_t payload_id,
std::weak_ptr<PayloadStatusListener> listener) override;
void RegisterPayloadPath(int64_t payload_id,
const std::filesystem::path& file_path,
ConnectionsCallback callback) override;
Payload* GetIncomingPayload(int64_t payload_id) override;
void Cancel(int64_t payload_id) override;
void ClearIncomingPayloads() override;
sharing/nearby_file_handler.cc
-17
View File
@@ -50,14 +50,6 @@ std::vector<NearbyFileHandler::FileInfo> DoOpenFiles(
return files;
}
std::filesystem::path GenerateUniquePath(const std::filesystem::path& path) {
NL_DCHECK(!path.empty());
// Nearby Share is not responsible for generating unique paths, any more.
// Nearby Connections contains the logic to ensure there is no conflict.
// Just return the original file path, here.
return path;
}
} // namespace
NearbyFileHandler::NearbyFileHandler() {
@@ -75,15 +67,6 @@ void NearbyFileHandler::OpenFiles(std::vector<std::filesystem::path> file_paths,
});
}
void NearbyFileHandler::GetUniquePath(const std::filesystem::path& file_path,
GetUniquePathCallback callback) {
sequenced_task_runner_->PostTask(
[callback = std::move(callback), file_path]() {
std::filesystem::path unique_path = GenerateUniquePath(file_path);
callback(unique_path);
});
}
void NearbyFileHandler::DeleteFilesFromDisk(
std::vector<std::filesystem::path> file_paths,
DeleteFilesFromDiskCallback callback) {
sharing/nearby_file_handler.h
+1 -6
View File
@@ -37,7 +37,6 @@ class NearbyFileHandler {
};
using OpenFilesCallback = std::function<void(std::vector<FileInfo>)>;
using GetUniquePathCallback = std::function<void(std::filesystem::path)>;
using DeleteFilesFromDiskCallback = std::function<void()>;
NearbyFileHandler();
@@ -51,12 +50,8 @@ class NearbyFileHandler {
void DeleteFilesFromDisk(std::vector<std::filesystem::path> file_paths,
DeleteFilesFromDiskCallback callback);
// Finds a unique path name for |file_path| and runs |callback| with the same.
void GetUniquePath(const std::filesystem::path& file_path,
GetUniquePathCallback callback);
private:
std::unique_ptr<TaskRunner> sequenced_task_runner_ = nullptr;
std::unique_ptr<TaskRunner> sequenced_task_runner_;
};
} // namespace sharing
sharing/nearby_file_handler_test.cc
-23
View File
@@ -38,29 +38,6 @@ bool CreateFile(std::filesystem::path file_path) {
return true;
}
TEST(NearbyFileHandler, GetUniquePath) {
NearbyFileHandler nearby_file_handler;
std::filesystem::path unique_path;
absl::Notification notification;
std::filesystem::path test_file =
std::filesystem::temp_directory_path() / "nearby_nfh_test_abc.jpg";
std::filesystem::path expected_file =
std::filesystem::temp_directory_path() / "nearby_nfh_test_abc.jpg";
ASSERT_TRUE(CreateFile(test_file));
ASSERT_TRUE(RemoveFile(expected_file));
nearby_file_handler.GetUniquePath(
test_file, [&notification, &unique_path](std::filesystem::path path) {
unique_path = path;
notification.Notify();
});
notification.WaitForNotificationWithTimeout(absl::Seconds(1));
EXPECT_EQ(unique_path, expected_file);
}
TEST(NearbyFileHandler, OpenFiles) {
NearbyFileHandler nearby_file_handler;
absl::Notification notification;
sharing/nearby_sharing_service_impl.cc
+10 -67
View File
@@ -2464,7 +2464,6 @@ void NearbySharingServiceImpl::ReceivePayloads(
std::filesystem::u8path(settings_->GetCustomSavePath());
// Register payload path for all valid file payloads.
absl::flat_hash_map<int64_t, std::filesystem::path> valid_file_payloads;
for (auto& file : share_target.file_attachments) {
std::optional<int64_t> payload_id = GetAttachmentPayloadId(file.id());
if (!payload_id) {
@@ -2478,38 +2477,9 @@ void NearbySharingServiceImpl::ReceivePayloads(
std::filesystem::path file_path =
download_path / std::filesystem::u8path(file.file_name().cbegin(),
file.file_name().cend());
valid_file_payloads.emplace(file.id(), std::move(file_path));
}
auto aggregated_success = std::make_unique<bool>(true);
if (valid_file_payloads.empty()) {
OnPayloadPathsRegistered(share_target, std::move(aggregated_success),
std::move(status_codes_callback));
return;
}
path_registration_status_.share_target = share_target;
path_registration_status_.expected_count = valid_file_payloads.size();
path_registration_status_.current_count = 0;
path_registration_status_.status_codes_callback =
std::move(status_codes_callback);
path_registration_status_.status = true;
for (const auto& payload : valid_file_payloads) {
std::optional<int64_t> payload_id = GetAttachmentPayloadId(payload.first);
NL_DCHECK(payload_id);
file_handler_.GetUniquePath(
payload.second,
[this, attachment_id = payload.first,
payload_id = *payload_id](std::filesystem::path unique_path) {
OnUniquePathFetched(
attachment_id, payload_id,
[this](Status status) { OnPayloadPathRegistered(status); },
unique_path);
});
attachment_info_map_[file.id()].file_path = std::move(file_path);
}
OnPayloadPathsRegistered(share_target, std::move(status_codes_callback));
}
NearbySharingService::StatusCodes NearbySharingServiceImpl::SendPayloads(
@@ -2540,41 +2510,9 @@ NearbySharingService::StatusCodes NearbySharingServiceImpl::SendPayloads(
return StatusCodes::kOk;
}
void NearbySharingServiceImpl::OnUniquePathFetched(
int64_t attachment_id, int64_t payload_id,
std::function<void(Status)> callback, std::filesystem::path file_path) {
attachment_info_map_[attachment_id].file_path = file_path;
nearby_connections_manager_->RegisterPayloadPath(payload_id, file_path,
std::move(callback));
}
void NearbySharingServiceImpl::OnPayloadPathRegistered(Status status) {
if (status != Status::kSuccess) {
path_registration_status_.status = false;
}
path_registration_status_.current_count += 1;
if (path_registration_status_.current_count ==
path_registration_status_.expected_count) {
OnPayloadPathsRegistered(
path_registration_status_.share_target,
std::make_unique<bool>(path_registration_status_.status),
std::move(path_registration_status_.status_codes_callback));
}
}
void NearbySharingServiceImpl::OnPayloadPathsRegistered(
const ShareTarget& share_target, std::unique_ptr<bool> aggregated_success,
const ShareTarget& share_target,
std::function<void(StatusCodes status_codes)> status_codes_callback) {
NL_DCHECK(aggregated_success);
if (!*aggregated_success) {
NL_LOG(WARNING)
<< __func__
<< ": Not all payload paths could be registered successfully.";
std::move(status_codes_callback)(StatusCodes::kError);
return;
}
ShareTargetInfo* info = GetShareTargetInfo(share_target.id);
if (!info || !info->connection()) {
NL_LOG(WARNING) << __func__ << ": Accept invoked for unknown share target";
@@ -2864,8 +2802,13 @@ void NearbySharingServiceImpl::CreatePayloads(
[this, share_target = std::move(share_target),
callback = std::move(callback)](
std::vector<NearbyFileHandler::FileInfo> file_infos) {
OnOpenFiles(std::move(share_target), std::move(callback),
std::move(file_infos));
RunOnNearbySharingServiceThread(
"open_files", [this, share_target = std::move(share_target),
callback = std::move(callback),
file_infos = std::move(file_infos)]() {
OnOpenFiles(std::move(share_target), std::move(callback),
std::move(file_infos));
});
});
}
sharing/nearby_sharing_service_impl.h
+1 -16
View File
@@ -307,12 +307,8 @@ class NearbySharingServiceImpl
ShareTarget share_target,
std::function<void(StatusCodes status_codes)> status_codes_callback);
StatusCodes SendPayloads(const ShareTarget& share_target);
void OnUniquePathFetched(int64_t attachment_id, int64_t payload_id,
std::function<void(Status)> callback,
std::filesystem::path path);
void OnPayloadPathRegistered(Status status);
void OnPayloadPathsRegistered(
const ShareTarget& share_target, std::unique_ptr<bool> aggregated_success,
const ShareTarget& share_target,
std::function<void(StatusCodes status_codes)> status_codes_callback);
void OnOutgoingConnection(const ShareTarget& share_target,
@@ -642,17 +638,6 @@ class NearbySharingServiceImpl
// object is null.
std::shared_ptr<bool> is_shutting_down_ = nullptr;
// Tracks the path registration.
struct PathRegistrationStatus {
ShareTarget share_target;
uint32_t expected_count;
uint32_t current_count;
std::function<void(StatusCodes status_codes)> status_codes_callback;
bool status;
};
PathRegistrationStatus path_registration_status_;
// Used to identify current scanning session.
int64_t scanning_session_id_ = 0;
sharing/nearby_sharing_service_impl_test.cc
-72
View File
@@ -1083,11 +1083,6 @@ class NearbySharingServiceImplTest : public testing::Test {
// This method sets up an incoming connection and performs the steps
// required to simulate a successful incoming transfer.
void SuccessfullyReceiveTransfer() {
for (int64_t payload_id : GetValidIntroductionFramePayloadIds()) {
fake_nearby_connections_manager_->SetPayloadPathStatus(payload_id,
Status::kSuccess);
}
NiceMock<MockTransferUpdateCallback> callback;
ShareTarget share_target = SetUpIncomingConnection(callback);
@@ -2717,49 +2712,7 @@ TEST_F(NearbySharingServiceImplTest, AcceptInvalidShareTarget) {
EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout));
}
TEST_F(NearbySharingServiceImplTest,
AcceptValidShareTargetRegisterPayloadError) {
fake_nearby_connections_manager_->SetPayloadPathStatus(kFilePayloadId,
Status::kError);
NiceMock<MockTransferUpdateCallback> callback;
ShareTarget share_target = SetUpIncomingConnection(callback);
absl::Notification notification;
service_->Accept(
share_target.id, [&](NearbySharingServiceImpl::StatusCodes status_code) {
EXPECT_EQ(NearbySharingServiceImpl::StatusCodes::kError, status_code);
notification.Notify();
});
EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout));
EXPECT_TRUE(
fake_nearby_connections_manager_->DidUpgradeBandwidth(kEndpointId));
// Check data written to connection_.
EXPECT_TRUE(ExpectPairedKeyEncryptionFrame());
EXPECT_TRUE(ExpectPairedKeyResultFrame());
EXPECT_FALSE(connection_.IsClosed());
{
std::optional<std::filesystem::path> path =
fake_nearby_connections_manager_->GetRegisteredPayloadPath(
kFilePayloadId);
EXPECT_TRUE(path.has_value());
std::filesystem::remove(*path);
}
// To avoid UAF in OnIncomingTransferUpdate().
UnregisterReceiveSurface(&callback);
}
TEST_F(NearbySharingServiceImplTest, AcceptValidShareTarget) {
for (int64_t payload_id : GetValidIntroductionFramePayloadIds()) {
fake_nearby_connections_manager_->SetPayloadPathStatus(payload_id,
Status::kSuccess);
}
NiceMock<MockTransferUpdateCallback> callback;
ShareTarget share_target = SetUpIncomingConnection(callback);
@@ -2801,11 +2754,6 @@ TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadSuccessful) {
TEST_F(NearbySharingServiceImplTest,
AcceptValidShareTargetPayloadSuccessfulIncomingPayloadNotFound) {
for (int64_t payload_id : GetValidIntroductionFramePayloadIds()) {
fake_nearby_connections_manager_->SetPayloadPathStatus(payload_id,
Status::kSuccess);
}
NiceMock<MockTransferUpdateCallback> callback;
ShareTarget share_target = SetUpIncomingConnection(callback);
@@ -2905,11 +2853,6 @@ TEST_F(NearbySharingServiceImplTest,
}
TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadFailed) {
for (int64_t payload_id : GetValidIntroductionFramePayloadIds()) {
fake_nearby_connections_manager_->SetPayloadPathStatus(payload_id,
Status::kSuccess);
}
NiceMock<MockTransferUpdateCallback> callback;
ShareTarget share_target = SetUpIncomingConnection(callback);
@@ -2973,11 +2916,6 @@ TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadFailed) {
}
TEST_F(NearbySharingServiceImplTest, AcceptValidShareTargetPayloadCancelled) {
for (int64_t payload_id : GetValidIntroductionFramePayloadIds()) {
fake_nearby_connections_manager_->SetPayloadPathStatus(payload_id,
Status::kSuccess);
}
NiceMock<MockTransferUpdateCallback> callback;
ShareTarget share_target = SetUpIncomingConnection(callback);
@@ -4439,11 +4377,6 @@ TEST_F(NearbySharingServiceImplTest, CreateShareTarget) {
}
TEST_F(NearbySharingServiceImplTest, SelfShareAutoAccept) {
for (int64_t payload_id : GetValidIntroductionFramePayloadIds()) {
fake_nearby_connections_manager_->SetPayloadPathStatus(payload_id,
Status::kSuccess);
}
// We create an incoming connection corresponding to a certificate where the
// |for_self_share| field is set to 'true'. This value will be propagated to
// the ShareTarget, which will be used as a signal for the service to
@@ -4476,11 +4409,6 @@ TEST_F(NearbySharingServiceImplTest, SelfShareAutoAccept) {
}
TEST_F(NearbySharingServiceImplTest, SelfShareNoAutoAcceptInForeground) {
for (int64_t payload_id : GetValidIntroductionFramePayloadIds()) {
fake_nearby_connections_manager_->SetPayloadPathStatus(payload_id,
Status::kSuccess);
}
NiceMock<MockTransferUpdateCallback> callback;
ShareTarget share_target = SetUpIncomingConnection(
callback, /*is_foreground=*/true, /*for_self_share=*/true);