Fix tsan error.

PiperOrigin-RevId: 900978318

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",

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;
   }
 };
 

connections/implementation/analytics/throughput_recorder.cc

@@ -14,18 +14,17 @@
 
 #include "connections/implementation/analytics/throughput_recorder.h"
 
-#include <stdint.h>
-
-#include <new>
-#include <ostream>
+#include <cstdint>
+#include <memory>
 #include <string>
-#include <type_traits>
 #include <utility>
 
+#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<ThroughputRecorderContainer> 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<int64_t, PayloadDirection>(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<ThroughputRecorder>(
+        payload_id, payload_direction, payload_type);
+    instance->Start();
     throughput_recorders_.emplace(
         std::pair<int64_t, PayloadDirection>(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<int64_t, PayloadDirection>(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<int64_t, PayloadDirection>(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<int64_t, PayloadDirection>(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<int64_t, PayloadDirection>(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<int64_t, PayloadDirection>(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<int64_t, PayloadDirection>(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<int64_t, PayloadDirection>(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<int64_t, PayloadDirection>(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<int64_t, PayloadDirection>(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

connections/implementation/analytics/throughput_recorder.h

@@ -16,125 +16,159 @@
 #define NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_
 
 #include <cstdint>
-#include <string>
+#include <memory>
 #include <utility>
 
+#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<ThroughputRecorderContainer>;
+
+  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<location::nearby::proto::connections::Medium,
+                        Throughput>
+        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<Medium, Throughput> 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<int64_t, PayloadDirection> for <payload id, payload direction>
-  absl::flat_hash_map<std::pair<int64_t, PayloadDirection>, ThroughputRecorder*>
+  absl::flat_hash_map<std::pair<int64_t, connections::PayloadDirection>,
+                      std::unique_ptr<ThroughputRecorder>>
       throughput_recorders_ ABSL_GUARDED_BY(mutex_);
 };
 

connections/implementation/analytics/throughput_recorder_test.cc

@@ -16,35 +16,29 @@
 
 #include <stdint.h>
 
-#include <ostream>
 #include <string>
 
 #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<bool> {
  protected:
   ThroughputRecorderTest() = default;
   ~ThroughputRecorderTest() override {
-    ThroughputRecorderContainer::GetInstance().Shutdown();
+    ThroughputRecorderContainer::GetInstance().ClearForTest();
   }
 
   ThroughputRecorderContainer& tp_recorder_container_ =
@@ -54,70 +48,71 @@ class ThroughputRecorderTest : public testing::TestWithParam<bool> {
 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

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<std::string> 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;

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);
   }
 }