From b533dd834cf7156dffbb8f2db7971dd5f80ed4f3 Mon Sep 17 00:00:00 2001 From: Edwin Wu Date: Thu, 16 Apr 2026 17:02:55 -0700 Subject: [PATCH] Fix tsan error. PiperOrigin-RevId: 900978318 --- connections/implementation/analytics/BUILD | 2 +- .../analytics/packet_meta_data.h | 17 +- .../analytics/throughput_recorder.cc | 363 ++++++++++-------- .../analytics/throughput_recorder.h | 202 ++++++---- .../analytics/throughput_recorder_test.cc | 170 ++++---- .../implementation/endpoint_manager.cc | 7 +- connections/implementation/payload_manager.cc | 31 +- 7 files changed, 428 insertions(+), 364 deletions(-) diff --git a/connections/implementation/analytics/BUILD b/connections/implementation/analytics/BUILD index 903272ff..019db88e 100644 --- a/connections/implementation/analytics/BUILD +++ b/connections/implementation/analytics/BUILD @@ -43,9 +43,9 @@ cc_library( "//proto:connections_enums_cc_proto", "@com_google_absl//absl/algorithm:container", "@com_google_absl//absl/base:core_headers", + "@com_google_absl//absl/base:no_destructor", "@com_google_absl//absl/container:btree", "@com_google_absl//absl/container:flat_hash_map", - "@com_google_absl//absl/meta:type_traits", "@com_google_absl//absl/strings", "@com_google_absl//absl/strings:str_format", "@com_google_absl//absl/time", diff --git a/connections/implementation/analytics/packet_meta_data.h b/connections/implementation/analytics/packet_meta_data.h index 09d2a85a..ea29c856 100644 --- a/connections/implementation/analytics/packet_meta_data.h +++ b/connections/implementation/analytics/packet_meta_data.h @@ -19,7 +19,6 @@ #include "absl/time/time.h" #include "internal/platform/implementation/system_clock.h" -#include "internal/platform/system_clock.h" namespace nearby { namespace analytics { @@ -35,7 +34,7 @@ struct PacketMetaData { void Reset() { file_io_start_time = SystemClock::ElapsedRealtime(); - socket_io_start_time = SystemClock::ElapsedRealtime(); + encryption_start_time = SystemClock::ElapsedRealtime(); socket_io_start_time = SystemClock::ElapsedRealtime(); packet_size = 0; } @@ -44,7 +43,7 @@ struct PacketMetaData { this->packet_size = packet_size; } - int GetPacketSize() { + int GetPacketSize() const { return packet_size; } @@ -72,27 +71,27 @@ struct PacketMetaData { socket_io_end_time = SystemClock::ElapsedRealtime(); } - int64_t GetEncryptionTimeInMillis() { + int64_t GetEncryptionTimeInMillis() const { if (encryption_end_time > encryption_start_time) { return absl::ToInt64Milliseconds(encryption_end_time - encryption_start_time); } - return 0L; + return 0; } - int64_t GetFileIoTimeInMillis() { + 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 0L; + return 0; } - int64_t GetSocketIoTimeInMillis() { + 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 0L; + return 0; } }; diff --git a/connections/implementation/analytics/throughput_recorder.cc b/connections/implementation/analytics/throughput_recorder.cc index 7833d4ad..ab467d12 100644 --- a/connections/implementation/analytics/throughput_recorder.cc +++ b/connections/implementation/analytics/throughput_recorder.cc @@ -14,18 +14,17 @@ #include "connections/implementation/analytics/throughput_recorder.h" -#include - -#include -#include +#include +#include #include -#include #include +#include "absl/base/no_destructor.h" #include "absl/container/flat_hash_map.h" -#include "absl/meta/type_traits.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" @@ -33,61 +32,73 @@ 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 -ThroughputRecorder::ThroughputRecorder(int64_t payload_id) - : payload_id_(payload_id) {} - ThroughputRecorderContainer& ThroughputRecorderContainer::GetInstance() { - alignas(ThroughputRecorderContainer) static char - storage[sizeof(ThroughputRecorderContainer)]; - static ThroughputRecorderContainer* env = - new (&storage) ThroughputRecorderContainer(); - return *env; + static absl::NoDestructor instance; + return *instance; } -void ThroughputRecorder::Start(PayloadType payload_type, - PayloadDirection payload_direction) { - std::string direction = - (payload_direction == PayloadDirection::INCOMING_PAYLOAD) ? "; Receive" - : "; Send"; +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"; +} - VLOG(1) << "Start TP profiling for payload_id:" << payload_id_ << direction; - - if (payload_type == PayloadType::kUnknown) { - VLOG(1) << "Ignore ThroughputRecorder::start for Unknown Payload type"; - return; +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; } - MutexLock lock(&mutex_); start_timestamp_ = SystemClock::ElapsedRealtime(); - payload_type_ = payload_type; - payload_direction_ = payload_direction; - // Add packetLostAlarm later } -bool ThroughputRecorder::Stop() { - MutexLock lock(&mutex_); +bool ThroughputRecorderContainer::ThroughputRecorder::Stop() { VLOG(1) << "Stop TP profiling for payload_id:" << payload_id_; - if (payload_type_ == PayloadType::kUnknown) { - VLOG(1) << "Ignore ThroughputRecorder::stop as it never start"; - return false; - } { - // Add packetLostAlarm stop process later absl::Time stop_timestamp = SystemClock::ElapsedRealtime(); int64_t total_byte_size = 0; int medium_size = throughputs_.size(); - // The worse case is the socket/connect blocking the write request, never - // got return when writing a frame out, it would get a very good data rate - // for this case. e.g. use 60 seconds to send a file and failed, the counter - // only get the duration as 30 seconds because the last write request - // blocked. if (!success_) { if (!throughputs_.empty()) { for (auto& tp : throughputs_) { @@ -96,9 +107,8 @@ bool ThroughputRecorder::Stop() { } } - // calculate throughput by medium for (auto& tp : throughputs_) { - tp.second.dump(); + tp.second.dump(payload_direction_, payload_type_); total_byte_size += tp.second.GetTotalByteSize(); } @@ -107,17 +117,16 @@ bool ThroughputRecorder::Stop() { int64_t total_millis = absl::ToInt64Milliseconds(stop_timestamp - start_timestamp_); throughput_kbps_ = CalculateThroughputKBps(total_byte_size, total_millis); - int throughput_mbps = CalculateThroughputMBps(throughput_kbps_); + int64_t throughput_mbps = CalculateThroughputMBps(throughput_kbps_); - // calculate overall throughput if there are multiple mediums if (medium_size > 1) { if (throughput_kbps_ != kDefaultThroughoutKbps) { std::string dump_content = absl::StrFormat( - "%s %s data(%d bytes) %s, overall used %d milliseconds, " + "%s %s data(%lld bytes) %s, overall used %lld milliseconds, " "throughput " - "is %d MB/s (%d KB/s), File IO takes %d ms, %s takes %d " + "is %lld MB/s (%lld KB/s), File IO takes %lld ms, %s takes %lld " "ms, " - "Socket IO takes %d ms", + "Socket IO takes %lld ms", (payload_direction_ == PayloadDirection::INCOMING_PAYLOAD) ? "Received" : "Sent", @@ -135,93 +144,79 @@ bool ThroughputRecorder::Stop() { return true; } -void ThroughputRecorder::MarkAsSuccess() { - MutexLock lock(&mutex_); +void ThroughputRecorderContainer::ThroughputRecorder::MarkAsSuccess() { success_ = true; } -int ThroughputRecorder::CalculateThroughputKBps(int64_t total_byte_size, - int64_t total_millis) { - if (total_millis > 0) { - return (int)(total_byte_size * kSecInMs / kKbInBytes / total_millis); - } - return kDefaultThroughoutKbps; -} - -int ThroughputRecorder::CalculateThroughputMBps(int throughputKBps) { - return throughputKBps / kKbInBytes; -} - -void ThroughputRecorder::Throughput::Add(int frame_size, int64_t file_io_time, - int64_t encryption_time, - int64_t socket_io_time) { +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; - // reset the last timestamp last_timestamp_ = SystemClock::ElapsedRealtime(); file_io_time_ += file_io_time; encryption_time_ += encryption_time; socket_io_time_ += socket_io_time; } -bool ThroughputRecorder::Throughput::dump() { +bool ThroughputRecorderContainer::ThroughputRecorder::Throughput::dump( + PayloadDirection payload_direction, PayloadType payload_type) { int64_t total_millis = absl::ToInt64Milliseconds(last_timestamp_ - start_timestamp_); - int throughput_kbps = CalculateThroughputKBps(total_byte_size_, total_millis); + int64_t throughput_kbps = + CalculateThroughputKBps(total_byte_size_, total_millis); if (throughput_kbps == kDefaultThroughoutKbps) { return false; } - int throughpu_mbps = CalculateThroughputMBps(throughput_kbps); + 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(%ld bytes) via %s used %ld milliseconds, throughput is %d " - "MB/s (%d KB/s), File IO takes %ld ms, %s takes %ld ms, " - "Socket IO takes %ld ms, " - "Other takes %ld ms", - (payload_direction_ == PayloadDirection::INCOMING_PAYLOAD) ? "Received" - : "Sent", - ToString(payload_type_), total_byte_size_, + "%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, - throughpu_mbps, throughput_kbps, file_io_time_, - (payload_direction_ == PayloadDirection::INCOMING_PAYLOAD) ? "Decryption" - : "Encryption", + 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; } -ThroughputRecorder::Throughput& ThroughputRecorder::GetThroughput( +ThroughputRecorderContainer::ThroughputRecorder::Throughput& +ThroughputRecorderContainer::ThroughputRecorder::GetThroughput( Medium medium, int64_t duration_millis) { auto it = throughputs_.find(medium); if (it == throughputs_.end()) { - auto throughput = new Throughput( - medium, - SystemClock::ElapsedRealtime() - absl::Milliseconds(duration_millis), - payload_type_, payload_direction_); - throughputs_.emplace(medium, std::move(*throughput)); - delete throughput; + throughputs_.emplace( + medium, Throughput(medium, SystemClock::ElapsedRealtime() - + absl::Milliseconds(duration_millis))); return throughputs_.find(medium)->second; } return it->second; } -int ThroughputRecorder::GetThroughputsSize() { - MutexLock lock(&mutex_); +int ThroughputRecorderContainer::ThroughputRecorder::GetThroughputsSize() + const { return throughputs_.size(); } -int ThroughputRecorder::GetThroughputKbps() { return throughput_kbps_; } +int64_t ThroughputRecorderContainer::ThroughputRecorder::GetThroughputKbps() + const { + return throughput_kbps_; +} -int64_t ThroughputRecorder::GetDurationMillis() { return duration_millis_; } - -void ThroughputRecorder::OnFrameSent(Medium medium, - PacketMetaData& packetMetaData) { - MutexLock lock(&mutex_); - if (payload_type_ == PayloadType::kUnknown) { - VLOG(1) << "PayloadType is invalid, return"; - return; - } +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(); @@ -232,96 +227,69 @@ void ThroughputRecorder::OnFrameSent(Medium medium, CalculateDurationTimes(packetMetaData); } -void ThroughputRecorder::OnFrameReceived(Medium medium, - PacketMetaData& packetMetaData) { - MutexLock lock(&mutex_); - if (payload_type_ == PayloadType::kUnknown) { - VLOG(1) << "PayloadType is invalid, return"; - return; - } - - // Add packetLostAlarm process later - 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 ThroughputRecorder::CalculateDurationTimes(PacketMetaData packetMetaData) { +void ThroughputRecorderContainer::ThroughputRecorder::CalculateDurationTimes( + const PacketMetaData& packetMetaData) { encryption_time_ += packetMetaData.GetEncryptionTimeInMillis(); socket_io_time_ += packetMetaData.GetSocketIoTimeInMillis(); file_io_time_ += packetMetaData.GetFileIoTimeInMillis(); } -std::string ThroughputRecorder::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"); +// Implementation for ThroughputRecorderContainer + +void ThroughputRecorderContainer::Start(int64_t payload_id, + PayloadDirection payload_direction, + PayloadType payload_type) { + if (payload_type == PayloadType::kUnknown) { + return; } -} - -// Inplementation for ThroughputRecorderContainer - -void ThroughputRecorderContainer::Shutdown() { - MutexLock lock(&mutex_); - VLOG(1) << __func__ << ". Num of Instance:" << throughput_recorders_.size(); - for (auto& throughput_recorder : throughput_recorders_) { - VLOG(1) << "Stop instance: " << throughput_recorder.second; - throughput_recorder.second->Stop(); - delete throughput_recorder.second; - } - throughput_recorders_.clear(); -} - -ThroughputRecorder* ThroughputRecorderContainer::GetTPRecorder( - const 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()) { - auto instance = new ThroughputRecorder(payload_id); - std::string direction = - (payload_direction == PayloadDirection::INCOMING_PAYLOAD) ? "; Receive" - : "; Send"; - VLOG(1) << "Add ThroughputRecorder instance : " << instance - << " for payload_id:" << payload_id << direction; + auto instance = std::make_unique( + payload_id, payload_direction, payload_type); + instance->Start(); throughput_recorders_.emplace( std::pair(payload_id, payload_direction), - instance); - return instance; + std::move(instance)); + } else { + it->second->Start(); } - - return it->second; } -void ThroughputRecorderContainer::StopTPRecorder( - const int64_t payload_id, PayloadDirection payload_direction) { +void ThroughputRecorderContainer::UpdateFrameData( + int64_t payload_id, PayloadDirection payload_direction, Medium medium, + PacketMetaData& packet_meta_data) { MutexLock lock(&mutex_); - std::string direction = - (payload_direction == PayloadDirection::INCOMING_PAYLOAD) ? "; Receive" - : "; Send"; auto it = throughput_recorders_.find( std::pair(payload_id, payload_direction)); if (it != throughput_recorders_.end()) { - VLOG(1) << "Found and stop/delete ThroughputRecorder instance : " - << &(it->second) << " for payload_id:" << payload_id << direction; - it->second->Stop(); - delete it->second; - throughput_recorders_.erase( - std::pair(payload_id, payload_direction)); - return; + it->second->UpdateFrameData(medium, packet_meta_data); } - VLOG(1) << "No ThroughputRecorder found for :" << payload_id; +} + +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() { @@ -329,5 +297,66 @@ int ThroughputRecorderContainer::GetSize() { 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 index e6dcdc05..f8dd5d19 100644 --- a/connections/implementation/analytics/throughput_recorder.h +++ b/connections/implementation/analytics/throughput_recorder.h @@ -16,125 +16,159 @@ #define NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_ #include -#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" -#include "proto/connections_enums.pb.h" namespace nearby { namespace analytics { -// The following aliases are only for users' convenience. -using ::location::nearby::proto::connections::Medium; -using ::nearby::connections::PayloadType; -// Enum to represent if a payload is incoming or outgoing. -using ::nearby::connections::PayloadDirection; - -class ThroughputRecorder { +// 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: - explicit ThroughputRecorder(int64_t payload_id); - ~ThroughputRecorder() = default; + static ThroughputRecorderContainer& GetInstance(); - void Start(PayloadType payload_type, PayloadDirection payload_direction); - bool Stop() ABSL_LOCKS_EXCLUDED(mutex_); - static int CalculateThroughputKBps(int64_t total_byte_size, - int64_t total_millis); - static int CalculateThroughputMBps(int throughputKBps); + // 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_); - class Throughput { + // 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: - Throughput() = default; - ~Throughput() = default; - Throughput(Medium medium, absl::Time start_timestamp, - PayloadType payload_type, PayloadDirection payload_direction) - : medium_(medium), - start_timestamp_(start_timestamp), - payload_type_(payload_type), - payload_direction_(payload_direction) {} + ThroughputRecorder(int64_t payload_id, + connections::PayloadDirection payload_direction, + connections::PayloadType payload_type); + ~ThroughputRecorder() = default; - void Add(int frame_size, int64_t file_io_time, int64_t encryption_time, - int64_t socket_io_time); + void Start(); + bool Stop() ABSL_LOCKS_EXCLUDED(mutex_); - void SetLastTimestamp(absl::Time time_stamp) { - last_timestamp_ = time_stamp; - } + class Throughput { + public: + Throughput() = default; + ~Throughput() = default; + Throughput(location::nearby::proto::connections::Medium medium, + absl::Time start_timestamp) + : medium_(medium), start_timestamp_(start_timestamp) {} - int64_t GetTotalByteSize() { return total_byte_size_; } + void Add(int frame_size, int64_t file_io_time, int64_t encryption_time, + int64_t socket_io_time); - bool dump(); + 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: - Medium medium_; + void CalculateDurationTimes(const PacketMetaData& packetMetaData); + + const int64_t payload_id_; + const connections::PayloadDirection payload_direction_; + const connections::PayloadType payload_type_; absl::Time start_timestamp_; - PayloadType payload_type_; - int64_t total_byte_size_ = 0; - absl::Time last_timestamp_; - PayloadDirection payload_direction_ = PayloadDirection::INCOMING_PAYLOAD; + 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; }; - Throughput& GetThroughput(Medium medium, int64_t duration_millis); - int GetThroughputsSize(); - int GetThroughputKbps(); - int64_t GetDurationMillis(); - void OnFrameSent(Medium medium, PacketMetaData& packetMetaData); - void OnFrameReceived(Medium medium, PacketMetaData& packetMetaData); - void MarkAsSuccess(); - - private: - void CalculateDurationTimes(PacketMetaData packetMetaData); - static std::string ToString(PayloadType type); - - Mutex mutex_; - int64_t payload_id_ = 0; - absl::Time start_timestamp_; - PayloadType payload_type_ = PayloadType::kUnknown; - PayloadDirection payload_direction_ = PayloadDirection::INCOMING_PAYLOAD; - 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; - int throughput_kbps_ = 0; -}; - -class ThroughputRecorderContainer { - public: + ThroughputRecorderContainer() = default; ThroughputRecorderContainer(const ThroughputRecorderContainer&) = delete; ThroughputRecorderContainer& operator=(const ThroughputRecorderContainer&) = delete; - - static ThroughputRecorderContainer& GetInstance(); - void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_); - - ThroughputRecorder* GetTPRecorder(int64_t payload_id, - PayloadDirection payload_direction) - ABSL_LOCKS_EXCLUDED(mutex_); - void StopTPRecorder(int64_t payload_id, PayloadDirection payload_direction) - ABSL_LOCKS_EXCLUDED(mutex_); - int GetSize() ABSL_LOCKS_EXCLUDED(mutex_); - - private: - // This is a singleton object, for which destructor will never be called. - // Constructor will be invoked once from Instance() static method. - // Object is create in-place (with a placement new) to guarantee that - // destructor is not scheduled for execution at exit. - ThroughputRecorderContainer() = default; ~ThroughputRecorderContainer() = default; Mutex mutex_; // std::pair for - absl::flat_hash_map, ThroughputRecorder*> + absl::flat_hash_map, + std::unique_ptr> throughput_recorders_ ABSL_GUARDED_BY(mutex_); }; diff --git a/connections/implementation/analytics/throughput_recorder_test.cc b/connections/implementation/analytics/throughput_recorder_test.cc index 8b31add3..c28d8c8d 100644 --- a/connections/implementation/analytics/throughput_recorder_test.cc +++ b/connections/implementation/analytics/throughput_recorder_test.cc @@ -16,35 +16,29 @@ #include -#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 { -// TODO(b/246693797): Add unit tests coverage for throughput recorder code constexpr int64_t kPayloadIdA = 123456789; constexpr int64_t kPayloadIdB = 987654321; constexpr int kFrameSize = 10 * 64 * 1024; -constexpr int64_t kTotalByteSize1GB = 1024 * 1024 * 1024; -constexpr int64_t kTotalMillis10Sec = 10 * 1000; -constexpr int kTPResultKBPerSec = 1024 * 1024 / 10; -constexpr int kTPKBPerSec = 100 * 1024; -constexpr int kTPResultMBPerSec = 100; -// class ThroughputRecorderTest : public testing::Test { class ThroughputRecorderTest : public testing::TestWithParam { protected: ThroughputRecorderTest() = default; ~ThroughputRecorderTest() override { - ThroughputRecorderContainer::GetInstance().Shutdown(); + ThroughputRecorderContainer::GetInstance().ClearForTest(); } ThroughputRecorderContainer& tp_recorder_container_ = @@ -54,70 +48,71 @@ class ThroughputRecorderTest : public testing::TestWithParam { INSTANTIATE_TEST_SUITE_P(ParametrisedTestThroughputRecorderTest, ThroughputRecorderTest, testing::Values(true, false)); -TEST(ThroughputRecorder, CalculateThroughputKBps) { - EXPECT_EQ(ThroughputRecorder::CalculateThroughputKBps(kTotalByteSize1GB, - kTotalMillis10Sec), - kTPResultKBPerSec); - EXPECT_EQ(ThroughputRecorder::CalculateThroughputKBps(kTotalByteSize1GB, 0), - 0); -} - -TEST(ThroughputRecorder, CalculateThroughputMBps) { - EXPECT_EQ(ThroughputRecorder::CalculateThroughputMBps(kTPKBPerSec), - kTPResultMBPerSec); -} - TEST(ThroughputRecorderContainer, InstanceCreate_ContainerSize) { ThroughputRecorderContainer& TPRecorderContainer = ThroughputRecorderContainer::GetInstance(); - TPRecorderContainer.GetTPRecorder(kPayloadIdA, - PayloadDirection::OUTGOING_PAYLOAD); - TPRecorderContainer.GetTPRecorder(kPayloadIdB, - PayloadDirection::INCOMING_PAYLOAD); + 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().Shutdown(); + ThroughputRecorderContainer::GetInstance().ClearForTest(); EXPECT_EQ(ThroughputRecorderContainer::GetInstance().GetSize(), 0); } TEST_F(ThroughputRecorderTest, OnFrameSentSaveTransferredSize) { - auto TPRecorder = tp_recorder_container_.GetTPRecorder( - kPayloadIdA, PayloadDirection::OUTGOING_PAYLOAD); - TPRecorder->Start(PayloadType::kFile, PayloadDirection::OUTGOING_PAYLOAD); + tp_recorder_container_.Start(kPayloadIdA, + connections::PayloadDirection::OUTGOING_PAYLOAD, + connections::PayloadType::kFile); PacketMetaData packet_meta_data; packet_meta_data.SetPacketSize(kFrameSize); - TPRecorder->OnFrameSent(location::nearby::proto::connections::BLE, - packet_meta_data); - TPRecorder->OnFrameSent(location::nearby::proto::connections::BLE, - packet_meta_data); - TPRecorder->OnFrameSent(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); + tp_recorder_container_.UpdateFrameData( + kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, + location::nearby::proto::connections::BLE, packet_meta_data); - auto throughput = - TPRecorder->GetThroughput(location::nearby::proto::connections::BLE, 0); - EXPECT_EQ(throughput.GetTotalByteSize(), kFrameSize * 3); + EXPECT_EQ(tp_recorder_container_.GetTotalByteSizeForTesting( + kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, + location::nearby::proto::connections::BLE), + kFrameSize * 3); } TEST_F(ThroughputRecorderTest, OnIgnoreUnkownPaylaodType) { - auto TPRecorder = tp_recorder_container_.GetTPRecorder( - kPayloadIdA, PayloadDirection::OUTGOING_PAYLOAD); - TPRecorder->Start(PayloadType::kUnknown, PayloadDirection::OUTGOING_PAYLOAD); + tp_recorder_container_.Start(kPayloadIdA, + connections::PayloadDirection::OUTGOING_PAYLOAD, + connections::PayloadType::kUnknown); PacketMetaData packet_meta_data; - TPRecorder->OnFrameSent(location::nearby::proto::connections::BLE, - packet_meta_data); - EXPECT_EQ(TPRecorder->GetThroughputsSize(), 0); + 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); - TPRecorder->Start(PayloadType::kUnknown, PayloadDirection::INCOMING_PAYLOAD); - TPRecorder->OnFrameReceived(location::nearby::proto::connections::BLE, - packet_meta_data); - EXPECT_EQ(TPRecorder->GetThroughputsSize(), 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) { - auto TPRecorder = tp_recorder_container_.GetTPRecorder( - kPayloadIdA, PayloadDirection::OUTGOING_PAYLOAD); - TPRecorder->Start(PayloadType::kFile, PayloadDirection::OUTGOING_PAYLOAD); + tp_recorder_container_.Start(kPayloadIdA, + connections::PayloadDirection::OUTGOING_PAYLOAD, + connections::PayloadType::kFile); PacketMetaData packet_meta_data; packet_meta_data.SetPacketSize(kFrameSize); @@ -130,9 +125,11 @@ TEST_P(ThroughputRecorderTest, OnFrameSentStopAndDump) { packet_meta_data.StartSocketIo(); absl::SleepFor(absl::Milliseconds(7)); packet_meta_data.StopSocketIo(); - TPRecorder->OnFrameSent(location::nearby::proto::connections::BLE, - packet_meta_data); - EXPECT_EQ(TPRecorder->GetDurationMillis(), + 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()); @@ -147,21 +144,25 @@ TEST_P(ThroughputRecorderTest, OnFrameSentStopAndDump) { packet_meta_data.StartSocketIo(); absl::SleepFor(absl::Milliseconds(17)); packet_meta_data.StopSocketIo(); - TPRecorder->OnFrameSent(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); if (GetParam() == true) { LOG(INFO) << "MarkAsSuccess"; - TPRecorder->MarkAsSuccess(); + tp_recorder_container_.MarkAsSuccess( + kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD); } - EXPECT_TRUE(TPRecorder->Stop()); - EXPECT_NE(TPRecorder->GetThroughputKbps(), 0); + 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) { - auto TPRecorder = tp_recorder_container_.GetTPRecorder( - kPayloadIdA, PayloadDirection::OUTGOING_PAYLOAD); - TPRecorder->Start(PayloadType::kFile, PayloadDirection::OUTGOING_PAYLOAD); + tp_recorder_container_.Start(kPayloadIdA, + connections::PayloadDirection::OUTGOING_PAYLOAD, + connections::PayloadType::kFile); PacketMetaData packet_meta_data1; packet_meta_data1.SetPacketSize(kFrameSize); @@ -174,8 +175,9 @@ TEST_F(ThroughputRecorderTest, OnFrameSentStopAndDumpForMultiMeadium) { packet_meta_data1.StartSocketIo(); absl::SleepFor(absl::Milliseconds(7)); packet_meta_data1.StopSocketIo(); - TPRecorder->OnFrameSent(location::nearby::proto::connections::BLE, - packet_meta_data1); + 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); @@ -188,18 +190,21 @@ TEST_F(ThroughputRecorderTest, OnFrameSentStopAndDumpForMultiMeadium) { packet_meta_data2.StartSocketIo(); absl::SleepFor(absl::Milliseconds(17)); packet_meta_data2.StopSocketIo(); - TPRecorder->OnFrameSent(location::nearby::proto::connections::WIFI_LAN, - packet_meta_data2); + tp_recorder_container_.UpdateFrameData( + kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD, + location::nearby::proto::connections::WIFI_LAN, packet_meta_data2); - TPRecorder->MarkAsSuccess(); - EXPECT_TRUE(TPRecorder->Stop()); - EXPECT_NE(TPRecorder->GetThroughputKbps(), 0); + 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) { - auto TPRecorder = tp_recorder_container_.GetTPRecorder( - kPayloadIdA, PayloadDirection::INCOMING_PAYLOAD); - TPRecorder->Start(PayloadType::kFile, PayloadDirection::INCOMING_PAYLOAD); + tp_recorder_container_.Start(kPayloadIdA, + connections::PayloadDirection::INCOMING_PAYLOAD, + connections::PayloadType::kFile); PacketMetaData packet_meta_data; packet_meta_data.SetPacketSize(kFrameSize); @@ -212,20 +217,23 @@ TEST_F(ThroughputRecorderTest, OnFrameReceivedCheckDurationMillis) { packet_meta_data.StartSocketIo(); absl::SleepFor(absl::Milliseconds(7)); packet_meta_data.StopSocketIo(); - TPRecorder->OnFrameReceived(location::nearby::proto::connections::BLE, - packet_meta_data); - EXPECT_EQ(TPRecorder->GetDurationMillis(), + 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) { - auto TPRecorder = tp_recorder_container_.GetTPRecorder( - kPayloadIdA, PayloadDirection::OUTGOING_PAYLOAD); - auto throughput = - TPRecorder->GetThroughput(location::nearby::proto::connections::BLE, 0); - EXPECT_FALSE(throughput.dump()); + 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 diff --git a/connections/implementation/endpoint_manager.cc b/connections/implementation/endpoint_manager.cc index 1c950387..a93f4688 100644 --- a/connections/implementation/endpoint_manager.cc +++ b/connections/implementation/endpoint_manager.cc @@ -457,7 +457,6 @@ EndpointManager::~EndpointManager() { MutexLock lock(&mutex_); is_shutdown_ = true; } - analytics::ThroughputRecorderContainer::GetInstance().Shutdown(); CountDownLatch latch(1); RunOnEndpointManagerThread("bring-down-endpoints", [this, &latch]() { LOG(INFO) << "Bringing down endpoints"; @@ -961,9 +960,9 @@ std::vector EndpointManager::SendTransferFrameBytes( LOG(INFO) << "Failed to send packet; endpoint_id=" << endpoint_id; continue; } - analytics::ThroughputRecorderContainer::GetInstance() - .GetTPRecorder(payload_id, PayloadDirection::OUTGOING_PAYLOAD) - ->OnFrameSent(channel->GetMedium(), packet_meta_data); + analytics::ThroughputRecorderContainer::GetInstance().UpdateFrameData( + payload_id, PayloadDirection::OUTGOING_PAYLOAD, channel->GetMedium(), + packet_meta_data); } return failed_endpoint_ids; diff --git a/connections/implementation/payload_manager.cc b/connections/implementation/payload_manager.cc index 9dbf2521..5cd26943 100644 --- a/connections/implementation/payload_manager.cc +++ b/connections/implementation/payload_manager.cc @@ -209,10 +209,9 @@ bool PayloadManager::SendPayloadLoop( VLOG(1) << "Payload xfer done: payload_id=" << pending_payload.GetInternalPayload()->GetId() << "; size=" << next_chunk_offset; - ThroughputRecorderContainer::GetInstance() - .GetTPRecorder(pending_payload.GetInternalPayload()->GetId(), - PayloadDirection::OUTGOING_PAYLOAD) - ->MarkAsSuccess(); + ThroughputRecorderContainer::GetInstance().MarkAsSuccess( + pending_payload.GetInternalPayload()->GetId(), + PayloadDirection::OUTGOING_PAYLOAD); return false; } } @@ -364,7 +363,6 @@ void PayloadManager::DisconnectFromEndpointManager() { PayloadManager::~PayloadManager() { VLOG(1) << "PayloadManager: going down; self=" << this; - ThroughputRecorderContainer::GetInstance().Shutdown(); DisconnectFromEndpointManager(); CancelAllPayloads(); VLOG(1) << "PayloadManager: turn down payload executors; self=" << this; @@ -481,9 +479,8 @@ void PayloadManager::SendPayload(ClientProxy* client, std::int64_t next_chunk_offset = 0; int index = 0; - ThroughputRecorderContainer::GetInstance() - .GetTPRecorder(payload_id, PayloadDirection::OUTGOING_PAYLOAD) - ->Start(payload_type, PayloadDirection::OUTGOING_PAYLOAD); + ThroughputRecorderContainer::GetInstance().Start( + payload_id, PayloadDirection::OUTGOING_PAYLOAD, payload_type); while (should_continue && !shutdown_.Get()) { should_continue = SendPayloadLoop(client, *pending_payload, payload_header, @@ -1326,10 +1323,9 @@ void PayloadManager::ProcessDataPacket( Payload::Id payload_id = payload_header.id(); PendingPayloadHandle pending_payload; if (payload_chunk.offset() == 0) { - ThroughputRecorderContainer::GetInstance() - .GetTPRecorder(payload_id, PayloadDirection::INCOMING_PAYLOAD) - ->Start((PayloadType)payload_header.type(), - PayloadDirection::INCOMING_PAYLOAD); + 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, @@ -1439,13 +1435,12 @@ void PayloadManager::ProcessDataPacket( payload_chunk.flags(), payload_chunk.offset(), payload_body_size); - ThroughputRecorderContainer::GetInstance() - .GetTPRecorder(payload_header.id(), PayloadDirection::INCOMING_PAYLOAD) - ->OnFrameReceived(medium, packet_meta_data); + ThroughputRecorderContainer::GetInstance().UpdateFrameData( + payload_header.id(), PayloadDirection::INCOMING_PAYLOAD, medium, + packet_meta_data); if (is_last_chunk) { - ThroughputRecorderContainer::GetInstance() - .GetTPRecorder(payload_header.id(), PayloadDirection::INCOMING_PAYLOAD) - ->MarkAsSuccess(); + ThroughputRecorderContainer::GetInstance().MarkAsSuccess( + payload_header.id(), PayloadDirection::INCOMING_PAYLOAD); } }