From 6f52ec53dd26ba0b2436611d20d53d1b20017bab Mon Sep 17 00:00:00 2001 From: Francis Tsui Date: Thu, 30 Apr 2026 13:54:40 -0700 Subject: [PATCH] Remove unnecessary code. PiperOrigin-RevId: 908359047 --- connections/implementation/BUILD | 7 +- connections/implementation/analytics/BUILD | 4 - .../analytics/packet_meta_data.h | 101 ----- .../analytics/throughput_recorder.cc | 362 ------------------ .../analytics/throughput_recorder.h | 178 --------- .../analytics/throughput_recorder_test.cc | 241 ------------ .../implementation/base_endpoint_channel.cc | 22 -- .../implementation/base_endpoint_channel.h | 9 +- .../implementation/base_pcp_handler.cc | 3 +- connections/implementation/base_pcp_handler.h | 9 +- .../implementation/base_pcp_handler_test.cc | 4 +- connections/implementation/bwu_manager.cc | 5 +- connections/implementation/bwu_manager.h | 5 +- .../implementation/bwu_manager_test.cc | 23 +- ...nnections_authentication_transport_test.cc | 75 +--- .../implementation/encryption_runner_test.cc | 9 +- connections/implementation/endpoint_channel.h | 9 - .../implementation/endpoint_manager.cc | 31 +- connections/implementation/endpoint_manager.h | 10 +- .../implementation/endpoint_manager_test.cc | 91 +---- .../implementation/fake_endpoint_channel.h | 9 - .../implementation/mock_endpoint_channel.h | 77 ++++ connections/implementation/payload_manager.cc | 41 +- connections/implementation/payload_manager.h | 7 +- .../implementation/payload_manager_test.cc | 6 +- 25 files changed, 153 insertions(+), 1185 deletions(-) delete mode 100644 connections/implementation/analytics/packet_meta_data.h delete mode 100644 connections/implementation/analytics/throughput_recorder.cc delete mode 100644 connections/implementation/analytics/throughput_recorder.h delete mode 100644 connections/implementation/analytics/throughput_recorder_test.cc create mode 100644 connections/implementation/mock_endpoint_channel.h diff --git a/connections/implementation/BUILD b/connections/implementation/BUILD index cdb58602..edc8fd47 100644 --- a/connections/implementation/BUILD +++ b/connections/implementation/BUILD @@ -214,6 +214,7 @@ cc_library( "fake_bwu_handler.h", "fake_endpoint_channel.h", "mock_device.h", + "mock_endpoint_channel.h", "mock_service_controller.h", "mock_service_controller_router.h", "offline_simulation_user.h", @@ -440,8 +441,8 @@ cc_test( ], deps = [ ":internal", + ":internal_test", "//connections:core_types", - "//connections/implementation/analytics", "//connections/implementation/flags:connections_flags", "//internal/flags:nearby_flags", "//internal/platform:base", @@ -491,13 +492,11 @@ cc_test( ], deps = [ ":internal", - "//connections/implementation/analytics", + ":internal_test", "//internal/platform:base", "//internal/platform/implementation/g3", # build_cleaner: keep - "//proto:connections_enums_cc_proto", "@com_github_protobuf_matchers//protobuf-matchers", "@com_google_absl//absl/strings:string_view", - "@com_google_absl//absl/time", "@com_google_googletest//:gtest_main", ], ) diff --git a/connections/implementation/analytics/BUILD b/connections/implementation/analytics/BUILD index 019db88e..f84377b4 100644 --- a/connections/implementation/analytics/BUILD +++ b/connections/implementation/analytics/BUILD @@ -20,15 +20,12 @@ cc_library( name = "analytics", srcs = [ "analytics_recorder.cc", - "throughput_recorder.cc", ], hdrs = [ "advertising_metadata_params.h", "analytics_recorder.h", "connection_attempt_metadata_params.h", "discovery_metadata_params.h", - "packet_meta_data.h", - "throughput_recorder.h", ], copts = ["-DCORE_ADAPTER_DLL"], visibility = ["//connections:__subpackages__"], @@ -58,7 +55,6 @@ cc_test( size = "small", srcs = [ "analytics_recorder_test.cc", - "throughput_recorder_test.cc", ], shard_count = 16, deps = [ diff --git a/connections/implementation/analytics/packet_meta_data.h b/connections/implementation/analytics/packet_meta_data.h deleted file mode 100644 index ea29c856..00000000 --- a/connections/implementation/analytics/packet_meta_data.h +++ /dev/null @@ -1,101 +0,0 @@ -// Copyright 2022-2023 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_PACKET_META_DATA_H_ -#define NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_PACKET_META_DATA_H_ - -#include - -#include "absl/time/time.h" -#include "internal/platform/implementation/system_clock.h" - -namespace nearby { -namespace analytics { - -struct PacketMetaData { - int packet_size; - absl::Time file_io_start_time; - absl::Time file_io_end_time; - absl::Time encryption_start_time; - absl::Time encryption_end_time; - absl::Time socket_io_start_time; - absl::Time socket_io_end_time; - - void Reset() { - file_io_start_time = SystemClock::ElapsedRealtime(); - encryption_start_time = SystemClock::ElapsedRealtime(); - socket_io_start_time = SystemClock::ElapsedRealtime(); - packet_size = 0; - } - - void SetPacketSize(int packet_size) { - this->packet_size = packet_size; - } - - int GetPacketSize() const { - return packet_size; - } - - void StartFileIo() { - file_io_start_time = SystemClock::ElapsedRealtime(); - } - - void StopFileIo() { - file_io_end_time = SystemClock::ElapsedRealtime(); - } - - void StartEncryption() { - encryption_start_time = SystemClock::ElapsedRealtime(); - } - - void StopEncryption() { - encryption_end_time = SystemClock::ElapsedRealtime(); - } - - void StartSocketIo() { - socket_io_start_time = SystemClock::ElapsedRealtime(); - } - - void StopSocketIo() { - socket_io_end_time = SystemClock::ElapsedRealtime(); - } - - int64_t GetEncryptionTimeInMillis() const { - if (encryption_end_time > encryption_start_time) { - return absl::ToInt64Milliseconds(encryption_end_time - - encryption_start_time); - } - return 0; - } - - int64_t GetFileIoTimeInMillis() const { - if (file_io_end_time > file_io_start_time) { - return absl::ToInt64Milliseconds(file_io_end_time - file_io_start_time); - } - return 0; - } - - int64_t GetSocketIoTimeInMillis() const { - if (socket_io_end_time > socket_io_start_time) { - return absl::ToInt64Milliseconds(socket_io_end_time - - socket_io_start_time); - } - return 0; - } -}; - -} // namespace analytics -} // namespace nearby - -#endif // NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_PACKET_META_DATA_H_ diff --git a/connections/implementation/analytics/throughput_recorder.cc b/connections/implementation/analytics/throughput_recorder.cc deleted file mode 100644 index ab467d12..00000000 --- a/connections/implementation/analytics/throughput_recorder.cc +++ /dev/null @@ -1,362 +0,0 @@ -// Copyright 2022-2023 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#include "connections/implementation/analytics/throughput_recorder.h" - -#include -#include -#include -#include - -#include "absl/base/no_destructor.h" -#include "absl/container/flat_hash_map.h" -#include "absl/strings/str_format.h" -#include "absl/time/time.h" -#include "connections/implementation/analytics/packet_meta_data.h" -#include "connections/payload_type.h" -#include "internal/platform/implementation/system_clock.h" -#include "internal/platform/logging.h" -#include "internal/platform/mutex_lock.h" - -namespace nearby { -namespace analytics { - -using Medium = ::location::nearby::proto::connections::Medium; -using ::nearby::connections::PayloadDirection; -using ::nearby::connections::PayloadType; - -namespace { -constexpr int kDefaultThroughoutKbps = 0; -constexpr int kKbInBytes = 1024; -constexpr int kSecInMs = 1000; - -int64_t CalculateThroughputKBps(int64_t total_byte_size, int64_t total_millis) { - if (total_millis > 0) { - return total_byte_size * kSecInMs / kKbInBytes / total_millis; - } - return kDefaultThroughoutKbps; -} - -int64_t CalculateThroughputMBps(int64_t throughputKBps) { - return throughputKBps / kKbInBytes; -} - -std::string ToString(PayloadType type) { - switch (type) { - case PayloadType::kBytes: - return std::string("Bytes"); - case PayloadType::kStream: - return std::string("Stream"); - case PayloadType::kFile: - return std::string("File"); - case PayloadType::kUnknown: - return std::string("Unknown"); - } -} -} // namespace - -ThroughputRecorderContainer& ThroughputRecorderContainer::GetInstance() { - static absl::NoDestructor instance; - return *instance; -} - -ThroughputRecorderContainer::ThroughputRecorder::ThroughputRecorder( - int64_t payload_id, PayloadDirection payload_direction, - PayloadType payload_type) - : payload_id_(payload_id), - payload_direction_(payload_direction), - payload_type_(payload_type) { - LOG_IF(DFATAL, payload_type_ == PayloadType::kUnknown) - << "Invalid payload type"; -} - -void ThroughputRecorderContainer::ThroughputRecorder::Start() { - if (VLOG_IS_ON(1)) { - std::string direction = - (payload_direction_ == PayloadDirection::INCOMING_PAYLOAD) ? "; Receive" - : "; Send"; - VLOG(1) << "Start TP profiling for payload_id:" << payload_id_ << direction; - } - - start_timestamp_ = SystemClock::ElapsedRealtime(); -} - -bool ThroughputRecorderContainer::ThroughputRecorder::Stop() { - VLOG(1) << "Stop TP profiling for payload_id:" << payload_id_; - { - absl::Time stop_timestamp = SystemClock::ElapsedRealtime(); - int64_t total_byte_size = 0; - int medium_size = throughputs_.size(); - - if (!success_) { - if (!throughputs_.empty()) { - for (auto& tp : throughputs_) { - tp.second.SetLastTimestamp(stop_timestamp); - } - } - } - - for (auto& tp : throughputs_) { - tp.second.dump(payload_direction_, payload_type_); - total_byte_size += tp.second.GetTotalByteSize(); - } - - throughputs_.clear(); - - int64_t total_millis = - absl::ToInt64Milliseconds(stop_timestamp - start_timestamp_); - throughput_kbps_ = CalculateThroughputKBps(total_byte_size, total_millis); - int64_t throughput_mbps = CalculateThroughputMBps(throughput_kbps_); - - if (medium_size > 1) { - if (throughput_kbps_ != kDefaultThroughoutKbps) { - std::string dump_content = absl::StrFormat( - "%s %s data(%lld bytes) %s, overall used %lld milliseconds, " - "throughput " - "is %lld MB/s (%lld KB/s), File IO takes %lld ms, %s takes %lld " - "ms, " - "Socket IO takes %lld ms", - (payload_direction_ == PayloadDirection::INCOMING_PAYLOAD) - ? "Received" - : "Sent", - ToString(payload_type_), total_byte_size, - success_ ? "SUCCEEDED" : "FAILED", total_millis, throughput_mbps, - throughput_kbps_, file_io_time_, - (payload_direction_ == PayloadDirection::INCOMING_PAYLOAD) - ? "Decryption" - : "Encryption", - encryption_time_, socket_io_time_); - LOG(INFO) << dump_content; - } - } - } - return true; -} - -void ThroughputRecorderContainer::ThroughputRecorder::MarkAsSuccess() { - success_ = true; -} - -void ThroughputRecorderContainer::ThroughputRecorder::Throughput::Add( - int frame_size, int64_t file_io_time, int64_t encryption_time, - int64_t socket_io_time) { - total_byte_size_ += frame_size; - last_timestamp_ = SystemClock::ElapsedRealtime(); - file_io_time_ += file_io_time; - encryption_time_ += encryption_time; - socket_io_time_ += socket_io_time; -} - -bool ThroughputRecorderContainer::ThroughputRecorder::Throughput::dump( - PayloadDirection payload_direction, PayloadType payload_type) { - int64_t total_millis = - absl::ToInt64Milliseconds(last_timestamp_ - start_timestamp_); - int64_t throughput_kbps = - CalculateThroughputKBps(total_byte_size_, total_millis); - if (throughput_kbps == kDefaultThroughoutKbps) { - return false; - } - int64_t throughput_mbps = CalculateThroughputMBps(throughput_kbps); - int64_t other = - total_millis - file_io_time_ - encryption_time_ - socket_io_time_; - std::string dump_content = absl::StrFormat( - "%s %s data(%lld bytes) via %s used %lld ms, throughput is %lld " - "MB/s (%lld KB/s), File IO takes %lld ms, %s takes %lld ms, " - "Socket IO takes %lld ms, " - "Other takes %lld ms", - (payload_direction == PayloadDirection::INCOMING_PAYLOAD) ? "Received" - : "Sent", - ToString(payload_type), total_byte_size_, - location::nearby::proto::connections::Medium_Name(medium_), total_millis, - throughput_mbps, throughput_kbps, file_io_time_, - (payload_direction == PayloadDirection::INCOMING_PAYLOAD) ? "Decryption" - : "Encryption", - encryption_time_, socket_io_time_, other); - LOG(INFO) << dump_content; - return true; -} - -ThroughputRecorderContainer::ThroughputRecorder::Throughput& -ThroughputRecorderContainer::ThroughputRecorder::GetThroughput( - Medium medium, int64_t duration_millis) { - auto it = throughputs_.find(medium); - if (it == throughputs_.end()) { - throughputs_.emplace( - medium, Throughput(medium, SystemClock::ElapsedRealtime() - - absl::Milliseconds(duration_millis))); - return throughputs_.find(medium)->second; - } - return it->second; -} - -int ThroughputRecorderContainer::ThroughputRecorder::GetThroughputsSize() - const { - return throughputs_.size(); -} - -int64_t ThroughputRecorderContainer::ThroughputRecorder::GetThroughputKbps() - const { - return throughput_kbps_; -} - -int64_t ThroughputRecorderContainer::ThroughputRecorder::GetDurationMillis() - const { - return duration_millis_; -} - -void ThroughputRecorderContainer::ThroughputRecorder::UpdateFrameData( - Medium medium, PacketMetaData& packetMetaData) { - duration_millis_ = packetMetaData.GetEncryptionTimeInMillis() + - packetMetaData.GetFileIoTimeInMillis() + - packetMetaData.GetSocketIoTimeInMillis(); - GetThroughput(medium, duration_millis_) - .Add(packetMetaData.packet_size, packetMetaData.GetFileIoTimeInMillis(), - packetMetaData.GetEncryptionTimeInMillis(), - packetMetaData.GetSocketIoTimeInMillis()); - CalculateDurationTimes(packetMetaData); -} - -void ThroughputRecorderContainer::ThroughputRecorder::CalculateDurationTimes( - const PacketMetaData& packetMetaData) { - encryption_time_ += packetMetaData.GetEncryptionTimeInMillis(); - socket_io_time_ += packetMetaData.GetSocketIoTimeInMillis(); - file_io_time_ += packetMetaData.GetFileIoTimeInMillis(); -} - -// Implementation for ThroughputRecorderContainer - -void ThroughputRecorderContainer::Start(int64_t payload_id, - PayloadDirection payload_direction, - PayloadType payload_type) { - if (payload_type == PayloadType::kUnknown) { - return; - } - MutexLock lock(&mutex_); - auto it = throughput_recorders_.find( - std::pair(payload_id, payload_direction)); - if (it == throughput_recorders_.end()) { - auto instance = std::make_unique( - payload_id, payload_direction, payload_type); - instance->Start(); - throughput_recorders_.emplace( - std::pair(payload_id, payload_direction), - std::move(instance)); - } else { - it->second->Start(); - } -} - -void ThroughputRecorderContainer::UpdateFrameData( - int64_t payload_id, PayloadDirection payload_direction, Medium medium, - PacketMetaData& packet_meta_data) { - MutexLock lock(&mutex_); - auto it = throughput_recorders_.find( - std::pair(payload_id, payload_direction)); - if (it != throughput_recorders_.end()) { - it->second->UpdateFrameData(medium, packet_meta_data); - } -} - -void ThroughputRecorderContainer::MarkAsSuccess( - int64_t payload_id, PayloadDirection payload_direction) { - MutexLock lock(&mutex_); - auto it = throughput_recorders_.find( - std::pair(payload_id, payload_direction)); - if (it != throughput_recorders_.end()) { - it->second->MarkAsSuccess(); - } -} - -int64_t ThroughputRecorderContainer::StopTPRecorder( - int64_t payload_id, PayloadDirection payload_direction) { - MutexLock lock(&mutex_); - auto it = throughput_recorders_.find( - std::pair(payload_id, payload_direction)); - if (it != throughput_recorders_.end()) { - it->second->Stop(); - int64_t throughput_kbps = it->second->GetThroughputKbps(); - throughput_recorders_.erase(it); - return throughput_kbps; - } - return 0; -} - -int ThroughputRecorderContainer::GetSize() { - MutexLock lock(&mutex_); - return throughput_recorders_.size(); -} - -void ThroughputRecorderContainer::ClearForTest() { - MutexLock lock(&mutex_); - throughput_recorders_.clear(); -} - -int64_t ThroughputRecorderContainer::GetTotalByteSizeForTesting( - int64_t payload_id, PayloadDirection payload_direction, Medium medium) { - MutexLock lock(&mutex_); - auto it = throughput_recorders_.find( - std::pair(payload_id, payload_direction)); - if (it != throughput_recorders_.end()) { - return it->second->GetThroughput(medium, 0).GetTotalByteSize(); - } - return 0; -} - -int ThroughputRecorderContainer::GetThroughputsSizeForTesting( - int64_t payload_id, PayloadDirection payload_direction) { - MutexLock lock(&mutex_); - auto it = throughput_recorders_.find( - std::pair(payload_id, payload_direction)); - if (it != throughput_recorders_.end()) { - return it->second->GetThroughputsSize(); - } - return 0; -} - -int64_t ThroughputRecorderContainer::GetDurationMillisForTesting( - int64_t payload_id, PayloadDirection payload_direction) { - MutexLock lock(&mutex_); - auto it = throughput_recorders_.find( - std::pair(payload_id, payload_direction)); - if (it != throughput_recorders_.end()) { - return it->second->GetDurationMillis(); - } - return 0; -} - -int64_t ThroughputRecorderContainer::GetThroughputKbpsForTesting( - int64_t payload_id, PayloadDirection payload_direction) { - MutexLock lock(&mutex_); - auto it = throughput_recorders_.find( - std::pair(payload_id, payload_direction)); - if (it != throughput_recorders_.end()) { - return it->second->GetThroughputKbps(); - } - return 0; -} - -bool ThroughputRecorderContainer::DumpForTesting( - int64_t payload_id, PayloadDirection payload_direction, Medium medium) { - MutexLock lock(&mutex_); - auto it = throughput_recorders_.find( - std::pair(payload_id, payload_direction)); - if (it != throughput_recorders_.end()) { - return it->second->GetThroughput(medium, 0).dump( - payload_direction, it->second->GetPayloadType()); - } - return false; -} - -} // namespace analytics -} // namespace nearby diff --git a/connections/implementation/analytics/throughput_recorder.h b/connections/implementation/analytics/throughput_recorder.h deleted file mode 100644 index f8dd5d19..00000000 --- a/connections/implementation/analytics/throughput_recorder.h +++ /dev/null @@ -1,178 +0,0 @@ -// Copyright 2022-2023 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_ -#define NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_ - -#include -#include -#include - -#include "absl/base/no_destructor.h" -#include "absl/base/thread_annotations.h" -#include "absl/container/flat_hash_map.h" -#include "absl/time/time.h" -#include "connections/implementation/analytics/packet_meta_data.h" -#include "connections/payload_type.h" -#include "internal/platform/mutex.h" - -namespace nearby { -namespace analytics { - -// Container class to manage ThroughputRecorder instances. -// This class is a singleton and provides thread-safe proxy methods to record -// throughput for different payloads. -class ThroughputRecorderContainer { - public: - static ThroughputRecorderContainer& GetInstance(); - - // Records the start of a payload transfer. - void Start(int64_t payload_id, - connections::PayloadDirection payload_direction, - connections::PayloadType payload_type) ABSL_LOCKS_EXCLUDED(mutex_); - - // Records when a frame is sent or received. - void UpdateFrameData(int64_t payload_id, - connections::PayloadDirection payload_direction, - location::nearby::proto::connections::Medium medium, - PacketMetaData& packet_meta_data) - ABSL_LOCKS_EXCLUDED(mutex_); - - // Marks a payload transfer as successful. - void MarkAsSuccess(int64_t payload_id, - connections::PayloadDirection payload_direction) - ABSL_LOCKS_EXCLUDED(mutex_); - - // Stops and removes the throughput recorder for a given payload. - // This calculates and logs the final throughput statistics. - // Returns the throughput in KBps. - int64_t StopTPRecorder(int64_t payload_id, - connections::PayloadDirection payload_direction) - ABSL_LOCKS_EXCLUDED(mutex_); - - // Returns the number of active recorder instances. - int GetSize() ABSL_LOCKS_EXCLUDED(mutex_); - - // Clear all recorders. Used for testing. - void ClearForTest() ABSL_LOCKS_EXCLUDED(mutex_); - - // Testing proxy methods - int64_t GetTotalByteSizeForTesting( - int64_t payload_id, connections::PayloadDirection payload_direction, - location::nearby::proto::connections::Medium medium) - ABSL_LOCKS_EXCLUDED(mutex_); - int GetThroughputsSizeForTesting( - int64_t payload_id, connections::PayloadDirection payload_direction) - ABSL_LOCKS_EXCLUDED(mutex_); - int64_t GetDurationMillisForTesting( - int64_t payload_id, connections::PayloadDirection payload_direction) - ABSL_LOCKS_EXCLUDED(mutex_); - int64_t GetThroughputKbpsForTesting( - int64_t payload_id, connections::PayloadDirection payload_direction) - ABSL_LOCKS_EXCLUDED(mutex_); - bool DumpForTesting(int64_t payload_id, - connections::PayloadDirection payload_direction, - location::nearby::proto::connections::Medium medium) - ABSL_LOCKS_EXCLUDED(mutex_); - - private: - friend class absl::NoDestructor; - - class ThroughputRecorder { - public: - ThroughputRecorder(int64_t payload_id, - connections::PayloadDirection payload_direction, - connections::PayloadType payload_type); - ~ThroughputRecorder() = default; - - void Start(); - bool Stop() ABSL_LOCKS_EXCLUDED(mutex_); - - class Throughput { - public: - Throughput() = default; - ~Throughput() = default; - Throughput(location::nearby::proto::connections::Medium medium, - absl::Time start_timestamp) - : medium_(medium), start_timestamp_(start_timestamp) {} - - void Add(int frame_size, int64_t file_io_time, int64_t encryption_time, - int64_t socket_io_time); - - void SetLastTimestamp(absl::Time time_stamp) { - last_timestamp_ = time_stamp; - } - - int64_t GetTotalByteSize() const { return total_byte_size_; } - - bool dump(connections::PayloadDirection payload_direction, - connections::PayloadType payload_type); - - private: - const ::location::nearby::proto::connections::Medium medium_; - const absl::Time start_timestamp_; - int64_t total_byte_size_ = 0; - absl::Time last_timestamp_; - int64_t file_io_time_ = 0; - int64_t encryption_time_ = 0; - int64_t socket_io_time_ = 0; - }; - - Throughput& GetThroughput( - location::nearby::proto::connections::Medium medium, - int64_t duration_millis); - int GetThroughputsSize() const; - int64_t GetThroughputKbps() const; - int64_t GetDurationMillis() const; - void UpdateFrameData(location::nearby::proto::connections::Medium medium, - PacketMetaData& packetMetaData); - void MarkAsSuccess(); - connections::PayloadType GetPayloadType() const { return payload_type_; } - - private: - void CalculateDurationTimes(const PacketMetaData& packetMetaData); - - const int64_t payload_id_; - const connections::PayloadDirection payload_direction_; - const connections::PayloadType payload_type_; - absl::Time start_timestamp_; - absl::flat_hash_map - throughputs_; - bool success_ = false; - - int64_t file_io_time_ = 0; - int64_t encryption_time_ = 0; - int64_t socket_io_time_ = 0; - int64_t duration_millis_ = 0; - int64_t throughput_kbps_ = 0; - }; - - ThroughputRecorderContainer() = default; - ThroughputRecorderContainer(const ThroughputRecorderContainer&) = delete; - ThroughputRecorderContainer& operator=(const ThroughputRecorderContainer&) = - delete; - ~ThroughputRecorderContainer() = default; - - Mutex mutex_; - // std::pair for - absl::flat_hash_map, - std::unique_ptr> - throughput_recorders_ ABSL_GUARDED_BY(mutex_); -}; - -} // namespace analytics -} // namespace nearby - -#endif // NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_ diff --git a/connections/implementation/analytics/throughput_recorder_test.cc b/connections/implementation/analytics/throughput_recorder_test.cc deleted file mode 100644 index c28d8c8d..00000000 --- a/connections/implementation/analytics/throughput_recorder_test.cc +++ /dev/null @@ -1,241 +0,0 @@ -// Copyright 2022-2023 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#include "connections/implementation/analytics/throughput_recorder.h" - -#include - -#include - -#include "gtest/gtest.h" -#include "absl/time/clock.h" -#include "absl/time/time.h" -#include "connections/implementation/analytics/packet_meta_data.h" -#include "connections/payload_type.h" -#include "internal/platform/logging.h" -#include "proto/connections_enums.pb.h" - -namespace nearby { -namespace analytics { -namespace { - -constexpr int64_t kPayloadIdA = 123456789; -constexpr int64_t kPayloadIdB = 987654321; -constexpr int kFrameSize = 10 * 64 * 1024; - -class ThroughputRecorderTest : public testing::TestWithParam { - protected: - ThroughputRecorderTest() = default; - ~ThroughputRecorderTest() override { - ThroughputRecorderContainer::GetInstance().ClearForTest(); - } - - ThroughputRecorderContainer& tp_recorder_container_ = - ThroughputRecorderContainer::GetInstance(); -}; - -INSTANTIATE_TEST_SUITE_P(ParametrisedTestThroughputRecorderTest, - ThroughputRecorderTest, testing::Values(true, false)); - -TEST(ThroughputRecorderContainer, InstanceCreate_ContainerSize) { - ThroughputRecorderContainer& TPRecorderContainer = - ThroughputRecorderContainer::GetInstance(); - TPRecorderContainer.Start(kPayloadIdA, - connections::PayloadDirection::OUTGOING_PAYLOAD, - connections::PayloadType::kFile); - TPRecorderContainer.Start(kPayloadIdB, - connections::PayloadDirection::INCOMING_PAYLOAD, - connections::PayloadType::kFile); - EXPECT_EQ(ThroughputRecorderContainer::GetInstance().GetSize(), 2); - ThroughputRecorderContainer::GetInstance().ClearForTest(); - EXPECT_EQ(ThroughputRecorderContainer::GetInstance().GetSize(), 0); -} - -TEST_F(ThroughputRecorderTest, OnFrameSentSaveTransferredSize) { - tp_recorder_container_.Start(kPayloadIdA, - connections::PayloadDirection::OUTGOING_PAYLOAD, - connections::PayloadType::kFile); - - PacketMetaData packet_meta_data; - packet_meta_data.SetPacketSize(kFrameSize); - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::BLE, packet_meta_data); - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::BLE, packet_meta_data); - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::BLE, packet_meta_data); - - EXPECT_EQ(tp_recorder_container_.GetTotalByteSizeForTesting( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::BLE), - kFrameSize * 3); -} - -TEST_F(ThroughputRecorderTest, OnIgnoreUnkownPaylaodType) { - tp_recorder_container_.Start(kPayloadIdA, - connections::PayloadDirection::OUTGOING_PAYLOAD, - connections::PayloadType::kUnknown); - - PacketMetaData packet_meta_data; - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::BLE, packet_meta_data); - EXPECT_EQ(tp_recorder_container_.GetThroughputsSizeForTesting( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD), - 0); - - tp_recorder_container_.Start(kPayloadIdA, - connections::PayloadDirection::INCOMING_PAYLOAD, - connections::PayloadType::kUnknown); - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::INCOMING_PAYLOAD, - location::nearby::proto::connections::BLE, packet_meta_data); - EXPECT_EQ(tp_recorder_container_.GetThroughputsSizeForTesting( - kPayloadIdA, connections::PayloadDirection::INCOMING_PAYLOAD), - 0); -} - -TEST_P(ThroughputRecorderTest, OnFrameSentStopAndDump) { - tp_recorder_container_.Start(kPayloadIdA, - connections::PayloadDirection::OUTGOING_PAYLOAD, - connections::PayloadType::kFile); - - PacketMetaData packet_meta_data; - packet_meta_data.SetPacketSize(kFrameSize); - packet_meta_data.StartFileIo(); - absl::SleepFor(absl::Milliseconds(5)); - packet_meta_data.StopFileIo(); - packet_meta_data.StartEncryption(); - absl::SleepFor(absl::Milliseconds(6)); - packet_meta_data.StopEncryption(); - packet_meta_data.StartSocketIo(); - absl::SleepFor(absl::Milliseconds(7)); - packet_meta_data.StopSocketIo(); - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::BLE, packet_meta_data); - EXPECT_EQ(tp_recorder_container_.GetDurationMillisForTesting( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD), - packet_meta_data.GetEncryptionTimeInMillis() + - packet_meta_data.GetFileIoTimeInMillis() + - packet_meta_data.GetSocketIoTimeInMillis()); - - packet_meta_data.SetPacketSize(kFrameSize); - packet_meta_data.StartFileIo(); - absl::SleepFor(absl::Milliseconds(15)); - packet_meta_data.StopFileIo(); - packet_meta_data.StartEncryption(); - absl::SleepFor(absl::Milliseconds(16)); - packet_meta_data.StopEncryption(); - packet_meta_data.StartSocketIo(); - absl::SleepFor(absl::Milliseconds(17)); - packet_meta_data.StopSocketIo(); - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::BLE, packet_meta_data); - - if (GetParam() == true) { - LOG(INFO) << "MarkAsSuccess"; - tp_recorder_container_.MarkAsSuccess( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD); - } - int throughput_kbps = tp_recorder_container_.StopTPRecorder( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD); - EXPECT_NE(throughput_kbps, 0); - EXPECT_EQ(tp_recorder_container_.GetSize(), 0); -} - -TEST_F(ThroughputRecorderTest, OnFrameSentStopAndDumpForMultiMeadium) { - tp_recorder_container_.Start(kPayloadIdA, - connections::PayloadDirection::OUTGOING_PAYLOAD, - connections::PayloadType::kFile); - - PacketMetaData packet_meta_data1; - packet_meta_data1.SetPacketSize(kFrameSize); - packet_meta_data1.StartFileIo(); - absl::SleepFor(absl::Milliseconds(5)); - packet_meta_data1.StopFileIo(); - packet_meta_data1.StartEncryption(); - absl::SleepFor(absl::Milliseconds(6)); - packet_meta_data1.StopEncryption(); - packet_meta_data1.StartSocketIo(); - absl::SleepFor(absl::Milliseconds(7)); - packet_meta_data1.StopSocketIo(); - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::BLE, packet_meta_data1); - - PacketMetaData packet_meta_data2; - packet_meta_data2.SetPacketSize(kFrameSize); - packet_meta_data2.StartFileIo(); - absl::SleepFor(absl::Milliseconds(15)); - packet_meta_data2.StopFileIo(); - packet_meta_data2.StartEncryption(); - absl::SleepFor(absl::Milliseconds(16)); - packet_meta_data2.StopEncryption(); - packet_meta_data2.StartSocketIo(); - absl::SleepFor(absl::Milliseconds(17)); - packet_meta_data2.StopSocketIo(); - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::WIFI_LAN, packet_meta_data2); - - tp_recorder_container_.MarkAsSuccess( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD); - int throughput_kbps = tp_recorder_container_.StopTPRecorder( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD); - EXPECT_NE(throughput_kbps, 0); -} - -TEST_F(ThroughputRecorderTest, OnFrameReceivedCheckDurationMillis) { - tp_recorder_container_.Start(kPayloadIdA, - connections::PayloadDirection::INCOMING_PAYLOAD, - connections::PayloadType::kFile); - - PacketMetaData packet_meta_data; - packet_meta_data.SetPacketSize(kFrameSize); - packet_meta_data.StartFileIo(); - absl::SleepFor(absl::Milliseconds(5)); - packet_meta_data.StopFileIo(); - packet_meta_data.StartEncryption(); - absl::SleepFor(absl::Milliseconds(6)); - packet_meta_data.StopEncryption(); - packet_meta_data.StartSocketIo(); - absl::SleepFor(absl::Milliseconds(7)); - packet_meta_data.StopSocketIo(); - tp_recorder_container_.UpdateFrameData( - kPayloadIdA, connections::PayloadDirection::INCOMING_PAYLOAD, - location::nearby::proto::connections::BLE, packet_meta_data); - EXPECT_EQ(tp_recorder_container_.GetDurationMillisForTesting( - kPayloadIdA, connections::PayloadDirection::INCOMING_PAYLOAD), - packet_meta_data.GetEncryptionTimeInMillis() + - packet_meta_data.GetFileIoTimeInMillis() + - packet_meta_data.GetSocketIoTimeInMillis()); -} - -TEST_F(ThroughputRecorderTest, OnTPRecorderNotStarted) { - tp_recorder_container_.Start(kPayloadIdA, - connections::PayloadDirection::OUTGOING_PAYLOAD, - connections::PayloadType::kUnknown); - EXPECT_FALSE(tp_recorder_container_.DumpForTesting( - kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, - location::nearby::proto::connections::BLE)); -} - -} // namespace -} // namespace analytics -} // namespace nearby diff --git a/connections/implementation/base_endpoint_channel.cc b/connections/implementation/base_endpoint_channel.cc index ec45785c..16baec65 100644 --- a/connections/implementation/base_endpoint_channel.cc +++ b/connections/implementation/base_endpoint_channel.cc @@ -92,17 +92,10 @@ BaseEndpointChannel::BaseEndpointChannel( try_count_(try_count) {} ExceptionOr BaseEndpointChannel::Read() { - PacketMetaData packet_meta_data; - return Read(packet_meta_data); -} - -ExceptionOr BaseEndpointChannel::Read( - PacketMetaData& packet_meta_data) { ByteArray result; { MutexLock lock(&reader_mutex_); - packet_meta_data.StartSocketIo(); ExceptionOr read_int; if (NearbyFlags::GetInstance().GetBoolFlag( config_package_nearby::nearby_connections_feature:: @@ -133,8 +126,6 @@ ExceptionOr BaseEndpointChannel::Read( if (!read_bytes.ok()) { return read_bytes; } - packet_meta_data.StopSocketIo(); - packet_meta_data.SetPacketSize(read_int.result() + sizeof(std::int32_t)); result = std::move(read_bytes.result()); } @@ -144,7 +135,6 @@ ExceptionOr BaseEndpointChannel::Read( if (IsEncryptionEnabledLocked()) { // If encryption is enabled, decode the message. std::string input(std::move(result)); - packet_meta_data.StartEncryption(); std::unique_ptr decrypted_data = crypto_context_->DecodeMessageFromPeer(input); if (decrypted_data) { @@ -175,7 +165,6 @@ ExceptionOr BaseEndpointChannel::Read( << ": Unable to parse data as unencrypted message."; } } - packet_meta_data.StopEncryption(); if (result.Empty()) { LOG(WARNING) << __func__ << ": Unable to parse read result."; return ExceptionOr(message_exception); @@ -191,12 +180,6 @@ ExceptionOr BaseEndpointChannel::Read( } Exception BaseEndpointChannel::Write(absl::string_view data) { - PacketMetaData packet_meta_data; - return Write(data, packet_meta_data); -} - -Exception BaseEndpointChannel::Write(absl::string_view data, - PacketMetaData& packet_meta_data) { { MutexLock pause_lock(&is_paused_mutex_); if (is_paused_) { @@ -217,9 +200,7 @@ Exception BaseEndpointChannel::Write(absl::string_view data, MutexLock crypto_lock(&crypto_mutex_); if (IsEncryptionEnabledLocked()) { // If encryption is enabled, encode the message. - packet_meta_data.StartEncryption(); encrypted = crypto_context_->EncodeMessageToPeer(data); - packet_meta_data.StopEncryption(); if (!encrypted) { LOG(WARNING) << __func__ << ": Failed to encrypt data."; return {Exception::kIo}; @@ -235,7 +216,6 @@ Exception BaseEndpointChannel::Write(absl::string_view data, return {Exception::kIo}; } - packet_meta_data.StartSocketIo(); Exception write_exception; if (NearbyFlags::GetInstance().GetBoolFlag( config_package_nearby::nearby_connections_feature:: @@ -262,8 +242,6 @@ Exception BaseEndpointChannel::Write(absl::string_view data, << ": Failed to flush writer: " << flush_exception.value; return flush_exception; } - packet_meta_data.StopSocketIo(); - packet_meta_data.SetPacketSize(data_size + sizeof(std::uint32_t)); } { diff --git a/connections/implementation/base_endpoint_channel.h b/connections/implementation/base_endpoint_channel.h index 3aa770a4..c426c118 100644 --- a/connections/implementation/base_endpoint_channel.h +++ b/connections/implementation/base_endpoint_channel.h @@ -23,7 +23,6 @@ #include "absl/strings/string_view.h" #include "absl/time/time.h" #include "connections/implementation/analytics/analytics_recorder.h" -#include "connections/implementation/analytics/packet_meta_data.h" #include "connections/implementation/endpoint_channel.h" #include "internal/platform/byte_array.h" #include "internal/platform/condition_variable.h" @@ -35,8 +34,6 @@ namespace nearby { namespace connections { -using analytics::PacketMetaData; - class BaseEndpointChannel : public EndpointChannel { public: BaseEndpointChannel(const std::string& service_id, @@ -51,12 +48,10 @@ class BaseEndpointChannel : public EndpointChannel { ~BaseEndpointChannel() override = default; // EndpointChannel: - ExceptionOr Read() override; - ExceptionOr Read(PacketMetaData& packet_meta_data) + ExceptionOr Read() ABSL_LOCKS_EXCLUDED(reader_mutex_, crypto_mutex_, last_read_mutex_) override; - Exception Write(absl::string_view data) override; - Exception Write(absl::string_view data, PacketMetaData& packet_meta_data) + Exception Write(absl::string_view data) ABSL_LOCKS_EXCLUDED(writer_mutex_, crypto_mutex_) override; void Close() ABSL_LOCKS_EXCLUDED(is_paused_mutex_) override; void Close(location::nearby::proto::connections::DisconnectionReason reason) diff --git a/connections/implementation/base_pcp_handler.cc b/connections/implementation/base_pcp_handler.cc index 0d79ceac..59418df5 100644 --- a/connections/implementation/base_pcp_handler.cc +++ b/connections/implementation/base_pcp_handler.cc @@ -1666,8 +1666,7 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client, void BasePcpHandler::OnIncomingFrame( OfflineFrame& frame, const std::string& endpoint_id, ClientProxy* client, - location::nearby::proto::connections::Medium medium, - PacketMetaData& packet_meta_data) { + location::nearby::proto::connections::Medium medium) { CountDownLatch latch(1); RunOnPcpHandlerThread( "incoming-frame", diff --git a/connections/implementation/base_pcp_handler.h b/connections/implementation/base_pcp_handler.h index f14665f6..4dda6590 100644 --- a/connections/implementation/base_pcp_handler.h +++ b/connections/implementation/base_pcp_handler.h @@ -31,7 +31,6 @@ #include "connections/advertising_options.h" #include "connections/connection_options.h" #include "connections/discovery_options.h" -#include "connections/implementation/analytics/packet_meta_data.h" #include "connections/implementation/bwu_manager.h" #include "connections/implementation/client_proxy.h" #include "connections/implementation/encryption_runner.h" @@ -158,10 +157,10 @@ class BasePcpHandler : public PcpHandler, const std::string& endpoint_id) override; // @EndpointManagerReaderThread - void OnIncomingFrame(location::nearby::connections::OfflineFrame& frame, - const std::string& endpoint_id, ClientProxy* client, - location::nearby::proto::connections::Medium medium, - analytics::PacketMetaData& packet_meta_data) override; + void OnIncomingFrame( + location::nearby::connections::OfflineFrame& frame, + const std::string& endpoint_id, ClientProxy* client, + location::nearby::proto::connections::Medium medium) override; // Called when an endpoint disconnects while we're waiting for both sides to // approve/reject the connection. diff --git a/connections/implementation/base_pcp_handler_test.cc b/connections/implementation/base_pcp_handler_test.cc index e1fad699..4313a5e3 100644 --- a/connections/implementation/base_pcp_handler_test.cc +++ b/connections/implementation/base_pcp_handler_test.cc @@ -32,7 +32,6 @@ #include "connections/advertising_options.h" #include "connections/connection_options.h" #include "connections/discovery_options.h" -#include "connections/implementation/analytics/packet_meta_data.h" #include "connections/implementation/base_endpoint_channel.h" #include "connections/implementation/bwu_manager.h" #include "connections/implementation/client_proxy.h" @@ -1591,7 +1590,6 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) { EndpointManager em(&ecm); BwuManager bwu(m, em, ecm, {}, {}); MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); - analytics::PacketMetaData packet_meta_data; StartDiscovery(client_.get(), &pcp_handler); auto mediums = pcp_handler.GetDiscoveryMediums(client_.get()); auto connect_medium = mediums[mediums.size() - 1]; @@ -1615,7 +1613,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) { Status::kSuccess, os_info, /*multiplex_socket_bitmask=*/0)); EXPECT_CALL(mock_connection_listener_.bandwidth_changed_cb, Call).Times(1); pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, client_.get(), - connect_medium, packet_meta_data); + connect_medium); LOG(INFO) << "Closing connection: id=" << endpoint_id; channel_b->Close(); bwu.Shutdown(); diff --git a/connections/implementation/bwu_manager.cc b/connections/implementation/bwu_manager.cc index 53084af8..d43a2329 100644 --- a/connections/implementation/bwu_manager.cc +++ b/connections/implementation/bwu_manager.cc @@ -400,8 +400,7 @@ void BwuManager::InitiateBwuForEndpoint(ClientProxy* client, void BwuManager::OnIncomingFrame(OfflineFrame& frame, const std::string& endpoint_id, - ClientProxy* client, Medium medium, - PacketMetaData& packet_meta_data) { + ClientProxy* client, Medium medium) { V1Frame::FrameType frame_type = parser::GetFrameType(frame); if (frame_type != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION) return; @@ -548,7 +547,7 @@ BwuHandler* BwuManager::GetHandlerForMedium(Medium medium) const { } void BwuManager::OnBwuNegotiationFrame( - ClientProxy* client, const BandwidthUpgradeNegotiationFrame frame, + ClientProxy* client, const BandwidthUpgradeNegotiationFrame& frame, const std::string& endpoint_id) { LOG(INFO) << "OnBwuNegotiationFrame: processing incoming " << BandwidthUpgradeNegotiationFrame::EventType_Name( diff --git a/connections/implementation/bwu_manager.h b/connections/implementation/bwu_manager.h index 2c5c766c..d2f53913 100644 --- a/connections/implementation/bwu_manager.h +++ b/connections/implementation/bwu_manager.h @@ -92,8 +92,7 @@ class BwuManager : public EndpointManager::FrameProcessor { // @EndpointManagerReaderThread void OnIncomingFrame(location::nearby::connections::OfflineFrame& frame, const std::string& endpoint_id, ClientProxy* client, - Medium medium, - PacketMetaData& packet_meta_data) override; + Medium medium) override; // Cleans up in-progress upgrades after endpoint disconnection. // @EndpointManagerReaderThread @@ -144,7 +143,7 @@ class BwuManager : public EndpointManager::FrameProcessor { // upgrade. void OnBwuNegotiationFrame( ClientProxy* client, - const location::nearby::connections::BandwidthUpgradeNegotiationFrame + const location::nearby::connections::BandwidthUpgradeNegotiationFrame& frame, const string& endpoint_id); diff --git a/connections/implementation/bwu_manager_test.cc b/connections/implementation/bwu_manager_test.cc index 4862c2e1..6a6495c5 100644 --- a/connections/implementation/bwu_manager_test.cc +++ b/connections/implementation/bwu_manager_test.cc @@ -188,12 +188,12 @@ class BwuManagerTest : public ::testing::Test { parser::FromBytes(parser::ForBwuLastWrite()); bwu_manager_->OnIncomingFrame(last_write_frame.result(), std::string(endpoint_id), &client_, - initial_medium, packet_meta_data_); + initial_medium); ExceptionOr safe_to_close_frame = parser::FromBytes(parser::ForBwuSafeToClose()); bwu_manager_->OnIncomingFrame(safe_to_close_frame.result(), std::string(endpoint_id), &client_, - initial_medium, packet_meta_data_); + initial_medium); return upgraded_channel; } @@ -209,7 +209,6 @@ class BwuManagerTest : public ::testing::Test { FakeBwuHandler* fake_wifi_direct_bwu_handler_ = nullptr; FakeBwuHandler* fake_wifi_hotspot_bwu_handler_ = nullptr; std::unique_ptr bwu_manager_; - PacketMetaData packet_meta_data_; }; TEST(BwuManagerBaseTest, AllowToUpgradeMedium) { @@ -370,12 +369,12 @@ TEST_P(BwuManagerTestParam, InitiateBwu_Success) { parser::FromBytes(parser::ForBwuLastWrite()); bwu_manager_->OnIncomingFrame(last_write_frame.result(), std::string(kEndpointId1), &client_, - Medium::BLUETOOTH, packet_meta_data_); + Medium::BLUETOOTH); ExceptionOr safe_to_close_frame = parser::FromBytes(parser::ForBwuSafeToClose()); bwu_manager_->OnIncomingFrame(safe_to_close_frame.result(), std::string(kEndpointId1), &client_, - Medium::BLUETOOTH, packet_meta_data_); + Medium::BLUETOOTH); // Confirm that upgrade channel is resumed after initial channel is shut down. // Note: If we didn't grab the shared initial channel pointer above, this @@ -871,7 +870,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnUpgradeFailure_FlagEnabled) { parser::FromBytes(parser::ForBwuFailure(info)); bwu_manager_->OnIncomingFrame(upgrade_failure.result(), std::string(kEndpointId3), &client_, - Medium::WEB_RTC, packet_meta_data_); + Medium::WEB_RTC); // With the flag enabled, we can safely revert WebRTC just for service B // because service B has no active WebRTC endpoints. @@ -908,7 +907,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnUpgradeFailure_FlagDisabled) { parser::FromBytes(parser::ForBwuFailure(info)); bwu_manager_->OnIncomingFrame(upgrade_failure.result(), std::string(kEndpointId3), &client_, - Medium::WEB_RTC, packet_meta_data_); + Medium::WEB_RTC); // With the flag disabled, we don't revert if there are still connected // endpoints for _any_ service. We don't have service-level bookkeeping; we @@ -938,7 +937,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_WifiDirect) { upgrade_path_info = sub_frame->mutable_upgrade_path_info(); upgrade_path_info->set_supports_client_introduction_ack(false); bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId1), &client_, - Medium::BLUETOOTH, packet_meta_data_); + Medium::BLUETOOTH); CountDownLatch latch(1); bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA, std::string(kEndpointId1), latch, @@ -973,7 +972,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Hotspot) { upgrade_path_info->set_supports_client_introduction_ack(false); upgrade_path_info->set_supports_disabling_encryption(true); bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId1), &client_, - Medium::BLUETOOTH, packet_meta_data_); + Medium::BLUETOOTH); CountDownLatch latch(1); bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA, std::string(kEndpointId1), latch, @@ -1001,7 +1000,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Wlan) { upgrade_path_info->set_supports_client_introduction_ack(false); bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId1), &client_, - Medium::BLUETOOTH, packet_meta_data_); + Medium::BLUETOOTH); CountDownLatch latch(1); bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA, std::string(kEndpointId1), latch, @@ -1040,7 +1039,7 @@ TEST_F(BwuManagerTest, BlockBwuFrameBeforeAccept) { upgrade_path_info2->set_supports_client_introduction_ack(false); upgrade_path_info2->set_supports_disabling_encryption(true); bwu_manager_->OnIncomingFrame(frame2, std::string(kEndpointId2), &client_, - Medium::BLUETOOTH, packet_meta_data_); + Medium::BLUETOOTH); CountDownLatch latch2(1); // The BWU frame should be drop, so the inProgressUpgrades should be empty. ASSERT_EQ(bwu_manager_->IsUpgradeOngoing(std::string(kEndpointId2)), false); @@ -1084,7 +1083,7 @@ TEST_F(BwuManagerTest, BlockBwuFrameFromAdvertiser) { EXPECT_TRUE(client_.IsConnectedToEndpoint(std::string(kEndpointId2))); bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId2), &client_, - Medium::BLUETOOTH, packet_meta_data_); + Medium::BLUETOOTH); CountDownLatch latch2(1); // The BWU frame should be drop, so the IsUpgradeOngoing should be empty. ASSERT_EQ(bwu_manager_->IsUpgradeOngoing(std::string(kEndpointId2)), false); diff --git a/connections/implementation/connections_authentication_transport_test.cc b/connections/implementation/connections_authentication_transport_test.cc index 667fbab6..333a4da2 100644 --- a/connections/implementation/connections_authentication_transport_test.cc +++ b/connections/implementation/connections_authentication_transport_test.cc @@ -14,7 +14,6 @@ #include "connections/implementation/connections_authentication_transport.h" -#include #include #include #include @@ -23,12 +22,9 @@ #include "protobuf-matchers/protocol-buffer-matchers.h" #include "gtest/gtest.h" #include "absl/strings/string_view.h" -#include "absl/time/time.h" -#include "connections/implementation/analytics/analytics_recorder.h" -#include "connections/implementation/endpoint_channel.h" +#include "connections/implementation/mock_endpoint_channel.h" #include "internal/platform/byte_array.h" #include "internal/platform/exception.h" -#include "proto/connections_enums.pb.h" namespace nearby { namespace connections { @@ -36,86 +32,39 @@ namespace { using ::testing::_; -class MockEndpointChannel : public EndpointChannel { - public: - MOCK_METHOD(ExceptionOr, Read, (), (override)); - MOCK_METHOD(ExceptionOr, Read, (PacketMetaData&), (override)); - MOCK_METHOD(Exception, Write, (absl::string_view data), (override)); - MOCK_METHOD(Exception, Write, (absl::string_view data, PacketMetaData&), - (override)); - MOCK_METHOD(void, Close, (), (override)); - MOCK_METHOD( - void, Close, - (location::nearby::proto::connections::DisconnectionReason reason), - (override)); - MOCK_METHOD(void, Close, - (location::nearby::proto::connections::DisconnectionReason reason, - location::nearby::analytics::proto::ConnectionsLog:: - EstablishedConnection::SafeDisconnectionResult result), - (override)); - MOCK_METHOD(bool, IsClosed, (), (const, override)); - MOCK_METHOD(std::string, GetType, (), (const, override)); - MOCK_METHOD(std::string, GetServiceId, (), (const, override)); - MOCK_METHOD(std::string, GetName, (), (const, override)); - MOCK_METHOD(location::nearby::proto::connections::Medium, GetMedium, (), - (const, override)); - MOCK_METHOD(location::nearby::proto::connections::ConnectionTechnology, - GetTechnology, (), (const, override)); - MOCK_METHOD(location::nearby::proto::connections::ConnectionBand, GetBand, (), - (const, override)); - MOCK_METHOD(int, GetFrequency, (), (const, override)); - MOCK_METHOD(int, GetTryCount, (), (const, override)); - MOCK_METHOD(int, GetMaxTransmitPacketSize, (), (const, override)); - MOCK_METHOD(void, EnableEncryption, (std::shared_ptr), - (override)); - MOCK_METHOD(void, DisableEncryption, (), (override)); - MOCK_METHOD(bool, IsEncrypted, (), (override)); - MOCK_METHOD(ExceptionOr, TryDecrypt, (const ByteArray& data), - (override)); - MOCK_METHOD(bool, IsPaused, (), (const, override)); - MOCK_METHOD(void, Pause, (), (override)); - MOCK_METHOD(void, Resume, (), (override)); - MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const, override)); - MOCK_METHOD(absl::Time, GetLastWriteTimestamp, (), (const, override)); - MOCK_METHOD(uint32_t, GetNextKeepAliveSeqNo, (), (const, override)); - MOCK_METHOD(void, SetAnalyticsRecorder, - (analytics::AnalyticsRecorder*, const std::string&), (override)); - - std::vector messages_; -}; - TEST(ConnectionsAuthenticationTransportTest, TestWriteMessage) { + std::vector messages; auto channel = std::make_shared(); - auto* channel_ptr = channel.get(); ConnectionsAuthenticationTransport transport(channel); EXPECT_CALL(*channel, Write(_)) - .WillOnce([channel_ptr](absl::string_view data) { - channel_ptr->messages_.push_back(std::string(data)); + .WillOnce([&messages](absl::string_view data) { + messages.push_back(std::string(data)); return Exception{ .value = Exception::Value::kSuccess, }; }); transport.WriteMessage("hello world"); - EXPECT_THAT(channel_ptr->messages_, testing::ElementsAre("hello world")); + EXPECT_THAT(messages, testing::ElementsAre("hello world")); } TEST(ConnectionsAuthenticationTransportTest, TestReadMessage) { + std::vector messages; auto channel = std::make_shared(); - auto* channel_ptr = channel.get(); ConnectionsAuthenticationTransport transport(channel); - channel_ptr->messages_.push_back("hello world"); - EXPECT_CALL(*channel, Read()).WillOnce([channel_ptr]() { - std::string ret = channel_ptr->messages_[0]; - channel_ptr->messages_.erase(channel_ptr->messages_.begin()); + messages.push_back("hello world"); + EXPECT_CALL(*channel, Read()).WillOnce([&messages]() { + std::string ret = messages[0]; + messages.erase(messages.begin()); return ExceptionOr(ByteArray(ret)); }); EXPECT_EQ(transport.ReadMessage(), "hello world"); } TEST(ConnectionsAuthenticationTransportTest, TestReadMessageFail) { + std::vector messages; auto channel = std::make_shared(); ConnectionsAuthenticationTransport transport(channel); - channel->messages_.push_back("hello world"); + messages.push_back("hello world"); EXPECT_CALL(*channel, Read()).WillOnce([]() { return ExceptionOr(Exception::Value::kIo); }); diff --git a/connections/implementation/encryption_runner_test.cc b/connections/implementation/encryption_runner_test.cc index f3cd2bd0..ee9bbc1d 100644 --- a/connections/implementation/encryption_runner_test.cc +++ b/connections/implementation/encryption_runner_test.cc @@ -52,18 +52,11 @@ class FakeEndpointChannel : public EndpointChannel { read_timestamp_ = SystemClock::ElapsedRealtime(); return in_ ? in_->Read(kChunkSize) : ExceptionOr{Exception::kIo}; } - ExceptionOr Read(PacketMetaData& packet_meta_data) override { - read_timestamp_ = SystemClock::ElapsedRealtime(); - return in_ ? in_->Read(kChunkSize) : ExceptionOr{Exception::kIo}; - } + Exception Write(absl::string_view data) override { write_timestamp_ = SystemClock::ElapsedRealtime(); return out_ ? out_->Write(data) : Exception{Exception::kIo}; } - Exception Write(absl::string_view data, - PacketMetaData& packet_meta_data) override { - return Write(data); - } void Close() override { if (in_) in_->Close(); if (out_) out_->Close(); diff --git a/connections/implementation/endpoint_channel.h b/connections/implementation/endpoint_channel.h index abd20ef3..fc0d87cf 100644 --- a/connections/implementation/endpoint_channel.h +++ b/connections/implementation/endpoint_channel.h @@ -23,15 +23,12 @@ #include "absl/strings/string_view.h" #include "absl/time/time.h" #include "connections/implementation/analytics/analytics_recorder.h" -#include "connections/implementation/analytics/packet_meta_data.h" #include "internal/platform/byte_array.h" #include "internal/platform/exception.h" namespace nearby { namespace connections { -using analytics::PacketMetaData; - class EndpointChannel { public: virtual ~EndpointChannel() = default; @@ -41,15 +38,9 @@ class EndpointChannel { virtual ExceptionOr Read() = 0; // throws Exception::IO, Exception::INTERRUPTED - virtual ExceptionOr Read(PacketMetaData& packet_meta_data) = 0; - virtual Exception Write(absl::string_view data) = 0; // throws Exception::IO - virtual Exception Write( - absl::string_view data, - PacketMetaData& packet_meta_data) = 0; // throws Exception::IO // Closes this EndpointChannel, without tracking the closure in analytics. - virtual void Close() = 0; // Closes this EndpointChannel and records the closure with the given reason. diff --git a/connections/implementation/endpoint_manager.cc b/connections/implementation/endpoint_manager.cc index c0e94b8b..5070d390 100644 --- a/connections/implementation/endpoint_manager.cc +++ b/connections/implementation/endpoint_manager.cc @@ -24,8 +24,6 @@ #include "absl/functional/any_invocable.h" #include "absl/time/time.h" #include "connections/connection_options.h" -#include "connections/implementation/analytics/packet_meta_data.h" -#include "connections/implementation/analytics/throughput_recorder.h" #include "connections/implementation/client_proxy.h" #include "connections/implementation/endpoint_channel.h" #include "connections/implementation/endpoint_channel_manager.h" @@ -34,7 +32,6 @@ #include "connections/implementation/service_id_constants.h" #include "connections/listeners.h" #include "connections/medium_selector.h" -#include "connections/payload_type.h" #include "internal/platform/byte_array.h" #include "internal/platform/count_down_latch.h" #include "internal/platform/exception.h" @@ -58,7 +55,6 @@ using ::location::nearby::connections::OfflineFrame; using ::location::nearby::connections::PayloadTransferFrame; using ::location::nearby::connections::V1Frame; using ::location::nearby::proto::connections::DisconnectionReason; -using ::nearby::analytics::PacketMetaData; // We set this to 11s to provide sufficient time for an in-progress WebRTC // bandwidth upgrade to resolve. This is chosen to be slightly longer than the @@ -235,8 +231,7 @@ ExceptionOr EndpointManager::HandleData( // a replacement for this endpoint since we last checked with the // EndpointChannelManager. while (true) { - PacketMetaData packet_meta_data; - ExceptionOr bytes = endpoint_channel->Read(packet_meta_data); + ExceptionOr bytes = endpoint_channel->Read(); if (!bytes.ok()) { LOG(INFO) << "Stop reading on read-time exception: " << bytes.exception(); // Treat kNoData as kIo. @@ -317,8 +312,7 @@ ExceptionOr EndpointManager::HandleData( } frame_processor->OnIncomingFrame(frame, endpoint_id, client, - endpoint_channel->GetMedium(), - packet_meta_data); + endpoint_channel->GetMedium()); } } @@ -657,8 +651,7 @@ int EndpointManager::GetMaxTransmitPacketSize(const std::string& endpoint_id) { std::vector EndpointManager::SendPayloadChunk( const PayloadTransferFrame::PayloadHeader& payload_header, const PayloadTransferFrame::PayloadChunk& payload_chunk, - const std::vector& endpoint_ids, - PacketMetaData& packet_meta_data) { + const std::vector& endpoint_ids) { std::string bytes = parser::ForDataPayloadTransfer(payload_header, payload_chunk); @@ -666,8 +659,7 @@ std::vector EndpointManager::SendPayloadChunk( endpoint_ids, bytes, payload_header.id(), /*offset=*/payload_chunk.offset(), /*packet_type=*/ - PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::DATA), - packet_meta_data); + PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::DATA)); } // Designed to run asynchronously. It is called from IO thread pools, and @@ -735,14 +727,12 @@ std::vector EndpointManager::SendControlMessage( const PayloadTransferFrame::ControlMessage& control, const std::vector& endpoint_ids) { std::string bytes = parser::ForControlPayloadTransfer(header, control); - PacketMetaData packet_meta_data; return SendTransferFrameBytes( endpoint_ids, bytes, header.id(), /*offset=*/control.offset(), /*packet_type=*/ - PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::CONTROL), - packet_meta_data); + PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::CONTROL)); } // @EndpointManagerThread @@ -912,20 +902,18 @@ CountDownLatch EndpointManager::NotifyFrameProcessorsOnEndpointDisconnect( std::vector EndpointManager::SendPayloadAck( std::int64_t payload_id, const std::vector& endpoint_ids) { std::string bytes = parser::ForPayloadAckPayloadTransfer(payload_id); - PacketMetaData packet_meta_data; return SendTransferFrameBytes( endpoint_ids, bytes, payload_id, /* offset= */ -1, /*packet_type=*/ - PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::PAYLOAD_ACK), - packet_meta_data); + PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::PAYLOAD_ACK)); } std::vector EndpointManager::SendTransferFrameBytes( const std::vector& endpoint_ids, const std::string& bytes, std::int64_t payload_id, std::int64_t offset, - const std::string& packet_type, PacketMetaData& packet_meta_data) { + const std::string& packet_type) { std::vector failed_endpoint_ids; for (const std::string& endpoint_id : endpoint_ids) { std::shared_ptr channel = @@ -944,15 +932,12 @@ std::vector EndpointManager::SendTransferFrameBytes( continue; } - Exception write_exception = channel->Write(bytes, packet_meta_data); + Exception write_exception = channel->Write(bytes); if (!write_exception.Ok()) { failed_endpoint_ids.push_back(endpoint_id); LOG(INFO) << "Failed to send packet; endpoint_id=" << endpoint_id; continue; } - analytics::ThroughputRecorderContainer::GetInstance().UpdateFrameData( - payload_id, PayloadDirection::OUTGOING_PAYLOAD, channel->GetMedium(), - packet_meta_data); } return failed_endpoint_ids; diff --git a/connections/implementation/endpoint_manager.h b/connections/implementation/endpoint_manager.h index 4b47f9c5..2250a958 100644 --- a/connections/implementation/endpoint_manager.h +++ b/connections/implementation/endpoint_manager.h @@ -26,7 +26,6 @@ #include "absl/functional/any_invocable.h" #include "absl/time/time.h" #include "connections/connection_options.h" -#include "connections/implementation/analytics/packet_meta_data.h" #include "connections/implementation/client_proxy.h" #include "connections/implementation/endpoint_channel.h" #include "connections/implementation/endpoint_channel_manager.h" @@ -80,8 +79,7 @@ class EndpointManager { virtual void OnIncomingFrame( location::nearby::connections::OfflineFrame& offline_frame, const std::string& from_endpoint_id, ClientProxy* to_client, - location::nearby::proto::connections::Medium current_medium, - analytics::PacketMetaData& packet_meta_data) = 0; + location::nearby::proto::connections::Medium current_medium) = 0; // Implementations must call barrier.CountDown() once // they're done. This parallelizes the disconnection event across all frame @@ -134,8 +132,7 @@ class EndpointManager { payload_header, const location::nearby::connections::PayloadTransferFrame::PayloadChunk& payload_chunk, - const std::vector& endpoint_ids, - analytics::PacketMetaData& packet_meta_data); + const std::vector& endpoint_ids); std::vector SendControlMessage( const location::nearby::connections::PayloadTransferFrame::PayloadHeader& payload_header, @@ -285,8 +282,7 @@ class EndpointManager { std::vector SendTransferFrameBytes( const std::vector& endpoint_ids, const std::string& payload_transfer_frame_bytes, std::int64_t payload_id, - std::int64_t offset, const std::string& packet_type, - analytics::PacketMetaData& packet_meta_data); + std::int64_t offset, const std::string& packet_type); // Executes all jobs sequentially, on a serial_executor_. void RunOnEndpointManagerThread(const std::string& name, Runnable runnable); diff --git a/connections/implementation/endpoint_manager_test.cc b/connections/implementation/endpoint_manager_test.cc index aecd71ab..70a85949 100644 --- a/connections/implementation/endpoint_manager_test.cc +++ b/connections/implementation/endpoint_manager_test.cc @@ -29,11 +29,10 @@ #include "absl/time/clock.h" #include "absl/time/time.h" #include "connections/connection_options.h" -#include "connections/implementation/analytics/analytics_recorder.h" #include "connections/implementation/client_proxy.h" -#include "connections/implementation/endpoint_channel.h" #include "connections/implementation/endpoint_channel_manager.h" #include "connections/implementation/flags/nearby_connections_feature_flags.h" +#include "connections/implementation/mock_endpoint_channel.h" #include "connections/implementation/offline_frames.h" #include "connections/listeners.h" #include "connections/status.h" @@ -63,68 +62,12 @@ using ::testing::MockFunction; using ::testing::Return; using ::testing::StrictMock; -class MockEndpointChannel : public EndpointChannel { - public: - MOCK_METHOD(ExceptionOr, Read, (), (override)); - MOCK_METHOD(ExceptionOr, Read, (PacketMetaData & packet_meta_data), - (override)); - MOCK_METHOD(Exception, Write, (absl::string_view data), (override)); - MOCK_METHOD(Exception, Write, - (absl::string_view data, PacketMetaData& packet_meta_data), - (override)); - MOCK_METHOD(void, Close, (), (override)); - MOCK_METHOD(void, Close, (DisconnectionReason reason), (override)); - MOCK_METHOD(void, Close, - (DisconnectionReason reason, - location::nearby::analytics::proto::ConnectionsLog:: - EstablishedConnection::SafeDisconnectionResult result), - (override)); - MOCK_METHOD(location::nearby::proto::connections::ConnectionTechnology, - GetTechnology, (), (const, override)); - MOCK_METHOD(location::nearby::proto::connections::ConnectionBand, GetBand, (), - (const, override)); - MOCK_METHOD(int, GetFrequency, (), (const, override)); - MOCK_METHOD(int, GetTryCount, (), (const, override)); - MOCK_METHOD(std::string, GetType, (), (const, override)); - MOCK_METHOD(std::string, GetServiceId, (), (const, override)); - MOCK_METHOD(std::string, GetName, (), (const, override)); - MOCK_METHOD(Medium, GetMedium, (), (const, override)); - MOCK_METHOD(int, GetMaxTransmitPacketSize, (), (const, override)); - MOCK_METHOD(void, EnableEncryption, - (std::shared_ptr context), (override)); - MOCK_METHOD(void, DisableEncryption, (), (override)); - MOCK_METHOD(bool, IsPaused, (), (const, override)); - MOCK_METHOD(bool, IsEncrypted, (), (override)); - MOCK_METHOD(ExceptionOr, TryDecrypt, (const ByteArray& data), - (override)); - MOCK_METHOD(void, Pause, (), (override)); - MOCK_METHOD(void, Resume, (), (override)); - MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const, override)); - MOCK_METHOD(absl::Time, GetLastWriteTimestamp, (), (const, override)); - MOCK_METHOD(uint32_t, GetNextKeepAliveSeqNo, (), (const, override)); - MOCK_METHOD(void, SetAnalyticsRecorder, - (analytics::AnalyticsRecorder*, const std::string&), (override)); - - bool IsClosed() const override { - absl::MutexLock lock(mutex_); - return closed_; - } - void DoClose() { - absl::MutexLock lock(mutex_); - closed_ = true; - } - - private: - mutable absl::Mutex mutex_; - bool closed_ = false; -}; - class MockFrameProcessor : public EndpointManager::FrameProcessor { public: MOCK_METHOD(void, OnIncomingFrame, (OfflineFrame & offline_frame, const std::string& from_endpoint_id, ClientProxy* to_client, - Medium current_medium, PacketMetaData& packet_meta_data), + Medium current_medium), (override)); MOCK_METHOD(void, OnEndpointDisconnect, @@ -281,7 +224,7 @@ TEST_F(EndpointManagerTest, RegisterFrameProcessorWorks) { parser::ForConnectionRequestConnections({}, connection_info); EXPECT_CALL(*connect_request, OnIncomingFrame); EXPECT_CALL(*connect_request, OnEndpointDisconnect); - EXPECT_CALL(*endpoint_channel, Read(_)) + EXPECT_CALL(*endpoint_channel, Read()) .WillOnce(Return(ExceptionOr(ByteArray(read_data)))) .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); EXPECT_CALL(*endpoint_channel, Write(_)) @@ -319,6 +262,8 @@ TEST_F(EndpointManagerTest, UnregisterFrameProcessorWorks) { TEST_F(EndpointManagerTest, SendControlMessageAndPayloadAckWorks) { auto endpoint_channel = std::make_unique(); + absl::Mutex close_mutex; + bool closed = false; PayloadTransferFrame::PayloadHeader header; PayloadTransferFrame::ControlMessage control; header.set_id(12345); @@ -327,22 +272,24 @@ TEST_F(EndpointManagerTest, SendControlMessageAndPayloadAckWorks) { control.set_offset(150); control.set_event(PayloadTransferFrame::ControlMessage::PAYLOAD_CANCELED); - ON_CALL(*endpoint_channel, Read(_)) - .WillByDefault([channel = endpoint_channel.get()]() { - if (channel->IsClosed()) return ExceptionOr(Exception::kIo); + ON_CALL(*endpoint_channel, Read()) + .WillByDefault([&, channel = endpoint_channel.get()]() { + absl::MutexLock lock(close_mutex); + if (closed) return ExceptionOr(Exception::kIo); LOG(INFO) << "Simulate read delay: wait"; absl::SleepFor(absl::Milliseconds(100)); LOG(INFO) << "Simulate read delay: done"; - if (channel->IsClosed()) return ExceptionOr(Exception::kIo); + if (closed) return ExceptionOr(Exception::kIo); return ExceptionOr(ByteArray{}); }); ON_CALL(*endpoint_channel, Close(_)) .WillByDefault( - [channel = endpoint_channel.get()](DisconnectionReason reason) { - channel->DoClose(); + [&, channel = endpoint_channel.get()](DisconnectionReason reason) { + absl::MutexLock lock(close_mutex); + closed = true; LOG(INFO) << "Channel closed"; }); - EXPECT_CALL(*endpoint_channel, Write(_, _)) + EXPECT_CALL(*endpoint_channel, Write(_)) .WillRepeatedly(Return(Exception{Exception::kSuccess})); RegisterEndpoint(std::move(endpoint_channel), false); @@ -359,7 +306,7 @@ TEST_F(EndpointManagerTest, SendControlMessageAndPayloadAckWorks) { TEST_F(EndpointManagerTest, SingleReadOnReadError) { auto endpoint_channel = std::make_unique(); - EXPECT_CALL(*endpoint_channel, Read(_)) + EXPECT_CALL(*endpoint_channel, Read()) .WillOnce( Return(ExceptionOr(Exception::kInvalidProtocolBuffer))); EXPECT_CALL(*endpoint_channel, Write(_)) @@ -377,7 +324,7 @@ TEST_F(EndpointManagerTest, ReadInvalidUnencryptedPayloadIgnoresFrame) { CountDownLatch latch(1); const ByteArray payload("not a valid frame"); auto endpoint_channel = std::make_unique(); - EXPECT_CALL(*endpoint_channel, Read(_)) + EXPECT_CALL(*endpoint_channel, Read()) .WillOnce(Return(ExceptionOr(payload))) .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload))) @@ -401,7 +348,7 @@ class EndpointManagerFuzzTest // too. // 4. Invalid frame is ignored. No bad side effects. auto endpoint_channel = std::make_unique(); - EXPECT_CALL(*endpoint_channel, Read(_)) + EXPECT_CALL(*endpoint_channel, Read()) .WillOnce(Return(ExceptionOr(payload))) .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload))) @@ -417,7 +364,7 @@ class EndpointManagerFuzzTest // 2. EndpointManager receives an invalid encrypted frame. // 3. No calls to TryDecrypt. auto endpoint_channel = std::make_unique(); - EXPECT_CALL(*endpoint_channel, Read(_)) + EXPECT_CALL(*endpoint_channel, Read()) .WillOnce(Return(ExceptionOr(payload))) .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); EXPECT_CALL(*endpoint_channel, IsEncrypted()).WillRepeatedly(Return(true)); @@ -468,7 +415,7 @@ TEST_F(EndpointManagerTest, TryDecrypt) { parser::ForConnectionRequestConnections({}, connection_info); EXPECT_CALL(*connect_request, OnIncomingFrame); EXPECT_CALL(*connect_request, OnEndpointDisconnect); - EXPECT_CALL(*endpoint_channel, Read(_)) + EXPECT_CALL(*endpoint_channel, Read()) .WillOnce(Return(ExceptionOr(payload))) .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload))) diff --git a/connections/implementation/fake_endpoint_channel.h b/connections/implementation/fake_endpoint_channel.h index 9221147d..cb03659f 100644 --- a/connections/implementation/fake_endpoint_channel.h +++ b/connections/implementation/fake_endpoint_channel.h @@ -46,19 +46,10 @@ class FakeEndpointChannel : public EndpointChannel { read_timestamp_ = SystemClock::ElapsedRealtime(); return read_output_; } - ExceptionOr Read(PacketMetaData& packet_meta_data) override { - read_timestamp_ = SystemClock::ElapsedRealtime(); - return read_output_; - } Exception Write(absl::string_view data) override { write_timestamp_ = SystemClock::ElapsedRealtime(); return write_output_; } - Exception Write(absl::string_view data, - PacketMetaData& packet_meta_data) override { - write_timestamp_ = SystemClock::ElapsedRealtime(); - return write_output_; - } void Close() override { is_closed_ = true; } void Close(location::nearby::proto::connections::DisconnectionReason reason) override { diff --git a/connections/implementation/mock_endpoint_channel.h b/connections/implementation/mock_endpoint_channel.h new file mode 100644 index 00000000..87f51995 --- /dev/null +++ b/connections/implementation/mock_endpoint_channel.h @@ -0,0 +1,77 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef THIRD_PARTY_NEARBY_CONNECTIONS_IMPLEMENTATION_MOCK_ENDPOINT_CHANNEL_H_ +#define THIRD_PARTY_NEARBY_CONNECTIONS_IMPLEMENTATION_MOCK_ENDPOINT_CHANNEL_H_ + +#include +#include +#include +#include "gmock/gmock.h" +#include "absl/strings/string_view.h" +#include "absl/time/time.h" +#include "connections/implementation/analytics/analytics_recorder.h" +#include "connections/implementation/endpoint_channel.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" + +namespace nearby::connections { + +class MockEndpointChannel : public EndpointChannel { + public: + MOCK_METHOD(ExceptionOr, Read, (), (override)); + MOCK_METHOD(Exception, Write, (absl::string_view data), + (override)); + MOCK_METHOD(void, Close, (), (override)); + MOCK_METHOD( + void, Close, + (location::nearby::proto::connections::DisconnectionReason reason), + (override)); + MOCK_METHOD(void, Close, + (location::nearby::proto::connections::DisconnectionReason reason, + location::nearby::analytics::proto::ConnectionsLog:: + EstablishedConnection::SafeDisconnectionResult result), + (override)); + MOCK_METHOD(bool, IsClosed, (), (const, override)); + MOCK_METHOD(std::string, GetType, (), (const, override)); + MOCK_METHOD(std::string, GetServiceId, (), (const, override)); + MOCK_METHOD(std::string, GetName, (), (const, override)); + MOCK_METHOD(location::nearby::proto::connections::Medium, GetMedium, (), + (const, override)); + MOCK_METHOD(location::nearby::proto::connections::ConnectionTechnology, + GetTechnology, (), (const, override)); + MOCK_METHOD(location::nearby::proto::connections::ConnectionBand, GetBand, (), + (const, override)); + MOCK_METHOD(int, GetFrequency, (), (const, override)); + MOCK_METHOD(int, GetTryCount, (), (const, override)); + MOCK_METHOD(int, GetMaxTransmitPacketSize, (), (const, override)); + MOCK_METHOD(void, EnableEncryption, (std::shared_ptr), + (override)); + MOCK_METHOD(void, DisableEncryption, (), (override)); + MOCK_METHOD(bool, IsEncrypted, (), (override)); + MOCK_METHOD(ExceptionOr, TryDecrypt, (const ByteArray& data), + (override)); + MOCK_METHOD(bool, IsPaused, (), (const, override)); + MOCK_METHOD(void, Pause, (), (override)); + MOCK_METHOD(void, Resume, (), (override)); + MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const, override)); + MOCK_METHOD(absl::Time, GetLastWriteTimestamp, (), (const, override)); + MOCK_METHOD(uint32_t, GetNextKeepAliveSeqNo, (), (const, override)); + MOCK_METHOD(void, SetAnalyticsRecorder, + (analytics::AnalyticsRecorder*, const std::string&), (override)); +}; + +} // namespace nearby::connections + +#endif // THIRD_PARTY_NEARBY_CONNECTIONS_IMPLEMENTATION_MOCK_ENDPOINT_CHANNEL_H_ diff --git a/connections/implementation/payload_manager.cc b/connections/implementation/payload_manager.cc index 5cd26943..e9f8b57b 100644 --- a/connections/implementation/payload_manager.cc +++ b/connections/implementation/payload_manager.cc @@ -29,8 +29,6 @@ #include "absl/strings/str_cat.h" #include "absl/strings/str_format.h" #include "absl/time/time.h" -#include "connections/implementation/analytics/packet_meta_data.h" -#include "connections/implementation/analytics/throughput_recorder.h" #include "connections/implementation/client_proxy.h" #include "connections/implementation/endpoint_channel_manager.h" #include "connections/implementation/endpoint_manager.h" @@ -65,8 +63,6 @@ using ::location::nearby::connections::V1Frame; using ::location::nearby::proto::connections::Medium; using ::location::nearby::proto::connections::OperationResultCode; using ::location::nearby::proto::connections::PayloadStatus; -using PacketMetaData = ::nearby::analytics::PacketMetaData; -using ::nearby::analytics::ThroughputRecorderContainer; using PayloadDirection = ::nearby::connections::PayloadDirection; constexpr absl::Duration kMinTransferUpdateInterval = absl::Milliseconds(50); @@ -81,7 +77,6 @@ bool PayloadManager::SendPayloadLoop( const EndpointIds& available_endpoint_ids = EndpointsToEndpointIds(pair.first); const Endpoints& unavailable_endpoints = pair.second; - PacketMetaData packet_meta_data; // First, handle any non-available endpoints. for (const auto& endpoint : unavailable_endpoints) { @@ -143,10 +138,8 @@ bool PayloadManager::SendPayloadLoop( // This will block if there is no data to transfer. // It will resume when new data arrives, or if Close() is called. int chunk_size = GetOptimalChunkSize(available_endpoint_ids); - packet_meta_data.StartFileIo(); ByteArray next_chunk = pending_payload.GetInternalPayload()->DetachNextChunk(chunk_size); - packet_meta_data.StopFileIo(); if (shutdown_.Get()) return false; // Save chunk size. We'll need it after we move next_chunk. auto next_chunk_size = next_chunk.size(); @@ -169,7 +162,7 @@ bool PayloadManager::SendPayloadLoop( PayloadTransferFrame::PayloadChunk payload_chunk(CreatePayloadChunk( next_chunk_offset - resume_offset, std::move(next_chunk), index)); const EndpointIds& failed_endpoint_ids = endpoint_manager_->SendPayloadChunk( - payload_header, payload_chunk, available_endpoint_ids, packet_meta_data); + payload_header, payload_chunk, available_endpoint_ids); // Check whether at least one endpoint failed. if (!failed_endpoint_ids.empty()) { VLOG(1) << "Payload xfer: endpoints failed: payload_id=" @@ -209,9 +202,6 @@ bool PayloadManager::SendPayloadLoop( VLOG(1) << "Payload xfer done: payload_id=" << pending_payload.GetInternalPayload()->GetId() << "; size=" << next_chunk_offset; - ThroughputRecorderContainer::GetInstance().MarkAsSuccess( - pending_payload.GetInternalPayload()->GetId(), - PayloadDirection::OUTGOING_PAYLOAD); return false; } } @@ -479,8 +469,6 @@ void PayloadManager::SendPayload(ClientProxy* client, std::int64_t next_chunk_offset = 0; int index = 0; - ThroughputRecorderContainer::GetInstance().Start( - payload_id, PayloadDirection::OUTGOING_PAYLOAD, payload_type); while (should_continue && !shutdown_.Get()) { should_continue = SendPayloadLoop(client, *pending_payload, payload_header, @@ -528,8 +516,7 @@ Status PayloadManager::CancelPayload(ClientProxy* client, void PayloadManager::OnIncomingFrame(OfflineFrame& offline_frame, const std::string& from_endpoint_id, ClientProxy* to_client, - Medium current_medium, - PacketMetaData& packet_meta_data) { + Medium current_medium) { PayloadTransferFrame& frame = *offline_frame.mutable_v1()->mutable_payload_transfer(); @@ -560,8 +547,7 @@ void PayloadManager::OnIncomingFrame(OfflineFrame& offline_frame, ProcessControlPacket(to_client, from_endpoint_id, frame); break; case PayloadTransferFrame::DATA: - ProcessDataPacket(to_client, from_endpoint_id, frame, current_medium, - packet_meta_data); + ProcessDataPacket(to_client, from_endpoint_id, frame, current_medium); break; case PayloadTransferFrame::PAYLOAD_ACK: VLOG(1) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender " @@ -812,10 +798,6 @@ PayloadManager::CreateIncomingPayload(const PayloadTransferFrame& frame, void PayloadManager::OnPendingPayloadDestroy(const PendingPayload* payload) { VLOG(1) << "PayloadManager: destroying " << payload->ToString() << " self=" << this; - ThroughputRecorderContainer::GetInstance().StopTPRecorder( - payload->GetId(), payload->IsIncoming() - ? PayloadDirection::INCOMING_PAYLOAD - : PayloadDirection::OUTGOING_PAYLOAD); if (payload->IsIncoming()) return; RunOnStatusUpdateThread( "~PendingPayload", @@ -1301,8 +1283,7 @@ void PayloadManager::HandleSuccessfulIncomingChunk( // @EndpointManagerDataPool void PayloadManager::ProcessDataPacket( ClientProxy* to_client, const std::string& from_endpoint_id, - PayloadTransferFrame& payload_transfer_frame, Medium medium, - PacketMetaData& packet_meta_data) { + PayloadTransferFrame& payload_transfer_frame, Medium medium) { PayloadTransferFrame::PayloadHeader& payload_header = *payload_transfer_frame.mutable_payload_header(); PayloadTransferFrame::PayloadChunk& payload_chunk = @@ -1323,10 +1304,6 @@ void PayloadManager::ProcessDataPacket( Payload::Id payload_id = payload_header.id(); PendingPayloadHandle pending_payload; if (payload_chunk.offset() == 0) { - ThroughputRecorderContainer::GetInstance().Start( - payload_id, PayloadDirection::INCOMING_PAYLOAD, - (PayloadType)payload_header.type()); - packet_meta_data.Reset(); RunOnStatusUpdateThread( "process-data-packet", [to_client, from_endpoint_id, payload_header, this]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() { @@ -1413,7 +1390,6 @@ void PayloadManager::ProcessDataPacket( // Save size of packet before we move it. std::int64_t payload_body_size = payload_chunk.body().size(); - packet_meta_data.StartFileIo(); if (pending_payload->GetInternalPayload() ->AttachNextChunk(payload_chunk.body()) .Raised()) { @@ -1425,7 +1401,6 @@ void PayloadManager::ProcessDataPacket( PayloadStatus::LOCAL_ERROR, OperationResultCode::IO_FILE_WRITING_ERROR); return; } - packet_meta_data.StopFileIo(); bool is_last_chunk = (payload_chunk.flags() & PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0; SendPayloadReceivedAck(to_client, *pending_payload, from_endpoint_id, @@ -1434,14 +1409,6 @@ void PayloadManager::ProcessDataPacket( HandleSuccessfulIncomingChunk(to_client, from_endpoint_id, payload_header, payload_chunk.flags(), payload_chunk.offset(), payload_body_size); - - ThroughputRecorderContainer::GetInstance().UpdateFrameData( - payload_header.id(), PayloadDirection::INCOMING_PAYLOAD, medium, - packet_meta_data); - if (is_last_chunk) { - ThroughputRecorderContainer::GetInstance().MarkAsSuccess( - payload_header.id(), PayloadDirection::INCOMING_PAYLOAD); - } } // @EndpointManagerDataPool diff --git a/connections/implementation/payload_manager.h b/connections/implementation/payload_manager.h index 4ba438c5..533ee79d 100644 --- a/connections/implementation/payload_manager.h +++ b/connections/implementation/payload_manager.h @@ -26,7 +26,6 @@ #include "absl/container/flat_hash_map.h" #include "absl/functional/any_invocable.h" #include "absl/time/time.h" -#include "connections/implementation/analytics/packet_meta_data.h" #include "connections/implementation/client_proxy.h" #include "connections/implementation/endpoint_manager.h" #include "connections/implementation/internal_payload.h" @@ -67,8 +66,7 @@ class PayloadManager : public EndpointManager::FrameProcessor { void OnIncomingFrame( location::nearby::connections::OfflineFrame& offline_frame, const std::string& from_endpoint_id, ClientProxy* to_client, - location::nearby::proto::connections::Medium current_medium, - analytics::PacketMetaData& packet_meta_data) override; + location::nearby::proto::connections::Medium current_medium) override; // @EndpointManagerThread void OnEndpointDisconnect( @@ -412,8 +410,7 @@ class PayloadManager : public EndpointManager::FrameProcessor { const std::string& from_endpoint_id, location::nearby::connections::PayloadTransferFrame& payload_transfer_frame, - location::nearby::proto::connections::Medium medium, - analytics::PacketMetaData& packet_meta_data); + location::nearby::proto::connections::Medium medium); void ProcessControlPacket(ClientProxy* to_client, const std::string& from_endpoint_id, location::nearby::connections::PayloadTransferFrame& diff --git a/connections/implementation/payload_manager_test.cc b/connections/implementation/payload_manager_test.cc index 612d69df..bcffb773 100644 --- a/connections/implementation/payload_manager_test.cc +++ b/connections/implementation/payload_manager_test.cc @@ -21,7 +21,6 @@ #include "gtest/gtest.h" #include "absl/strings/string_view.h" #include "absl/time/time.h" -#include "connections/implementation/analytics/packet_meta_data.h" #include "connections/implementation/offline_frames.h" #include "connections/implementation/simulation_user.h" #include "connections/listeners.h" @@ -43,7 +42,6 @@ namespace { using ::location::nearby::connections::OfflineFrame; using ::location::nearby::connections::PayloadTransferFrame; using ::location::nearby::proto::connections::Medium; -using ::nearby::analytics::PacketMetaData; constexpr size_t kChunkSize = 64 * 1024; constexpr absl::string_view kServiceId = "service-id"; @@ -116,10 +114,8 @@ class PayloadSimulationUser : public SimulationUser { std::string bytes = parser::ForDataPayloadTransfer(header, chunk); offline_frame.ParseFromString(bytes); - PacketMetaData packet_meta_data; - pm_.OnIncomingFrame(offline_frame, from_payload_id, &client_, - Medium::WIFI_HOTSPOT, packet_meta_data); + Medium::WIFI_HOTSPOT); } Status CancelPayload() {