Throughput profiling

PiperOrigin-RevId: 477752265

Package.swift

@@ -426,6 +426,7 @@ let package = Package(
         "connections/implementation/service_controller_router_test.cc",
         "connections/implementation/wifi_hotspot_test.cc",
         "connections/implementation/analytics/analytics_recorder_test.cc",
+        "connections/implementation/analytics/throughput_recorder_test.cc",
         "connections/implementation/mediums/ble_v2_test.cc",
         "connections/implementation/mediums/ble_v2/bloom_filter_test.cc",
         "connections/implementation/mediums/ble_v2/ble_packet_test.cc",

connections/implementation/analytics/BUILD

@@ -17,10 +17,13 @@ cc_library(
     name = "analytics",
     srcs = [
         "analytics_recorder.cc",
+        "throughput_recorder.cc",
     ],
     hdrs = [
         "analytics_recorder.h",
         "connection_attempt_metadata_params.h",
+        "packet_meta_data.h",
+        "throughput_recorder.h",
     ],
     copts = ["-DCORE_ADAPTER_DLL"],
     visibility = [
@@ -38,6 +41,7 @@ cc_library(
         "//proto:connections_enums_cc_proto",
         "//proto/errorcode:error_code_enums_cc_proto",
         "@com_google_absl//absl/container:btree",
+        "@com_google_absl//absl/container:flat_hash_map",
         "@com_google_absl//absl/time",
     ],
 )
@@ -47,6 +51,7 @@ cc_test(
     size = "small",
     srcs = [
         "analytics_recorder_test.cc",
+        "throughput_recorder_test.cc",
     ],
     shard_count = 16,
     deps = [

connections/implementation/analytics/packet_meta_data.h

@@ -0,0 +1,104 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_PACKET_META_DATA_H_
+#define NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_PACKET_META_DATA_H_
+
+#include <cstdint>
+#include <string>
+
+#include "absl/time/time.h"
+#include "internal/platform/system_clock.h"
+
+namespace location {
+namespace nearby {
+namespace analytics {
+
+struct PacketMetaData {
+  int packet_size;
+  absl::Time file_io_start_time;
+  absl::Time file_io_end_time;
+  absl::Time encryption_start_time;
+  absl::Time encryption_end_time;
+  absl::Time socket_io_start_time;
+  absl::Time socket_io_end_time;
+
+  void Reset() {
+    file_io_start_time = SystemClock::ElapsedRealtime();
+    socket_io_start_time = SystemClock::ElapsedRealtime();
+    socket_io_start_time = SystemClock::ElapsedRealtime();
+    packet_size = 0;
+  }
+
+  void SetPacketSize(int packet_size) {
+    this->packet_size = packet_size;
+  }
+
+  int GetPacketSize() {
+    return packet_size;
+  }
+
+  void StartFileIo() {
+    file_io_start_time = SystemClock::ElapsedRealtime();
+  }
+
+  void StopFileIo() {
+    file_io_end_time = SystemClock::ElapsedRealtime();
+  }
+
+  void StartEncryption() {
+    encryption_start_time = SystemClock::ElapsedRealtime();
+  }
+
+  void StopEncryption() {
+    encryption_end_time = SystemClock::ElapsedRealtime();
+  }
+
+  void StartSocketIo() {
+    socket_io_start_time = SystemClock::ElapsedRealtime();
+  }
+
+  void StopSocketIo() {
+    socket_io_end_time = SystemClock::ElapsedRealtime();
+  }
+
+  int64_t GetEncryptionTimeInMillis() {
+    if (encryption_end_time > encryption_start_time) {
+      return absl::ToInt64Milliseconds(encryption_end_time -
+                                       encryption_start_time);
+    }
+    return 0L;
+  }
+
+  int64_t GetFileIoTimeInMillis() {
+    if (file_io_end_time > file_io_start_time) {
+      return absl::ToInt64Milliseconds(file_io_end_time - file_io_start_time);
+    }
+    return 0L;
+  }
+
+  int64_t GetSocketIoTimeInMillis() {
+    if (socket_io_end_time > socket_io_start_time) {
+      return absl::ToInt64Milliseconds(socket_io_end_time -
+                                       socket_io_start_time);
+    }
+    return 0L;
+  }
+};
+
+}  // namespace analytics
+}  // namespace nearby
+}  // namespace location
+
+#endif  // NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_PACKET_META_DATA_H_

connections/implementation/analytics/throughput_recorder.cc

@@ -0,0 +1,303 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "connections/implementation/analytics/throughput_recorder.h"
+
+#include <string>
+#include <utility>
+
+#include "internal/platform/logging.h"
+#include "internal/platform/mutex_lock.h"
+namespace location {
+namespace nearby {
+namespace analytics {
+
+namespace {
+constexpr int kDefaultThroughoutKbps = 0;
+constexpr int kKbInBytes = 1024;
+constexpr int kSecInMs = 1000;
+}  // namespace
+
+ThroughputRecorder::ThroughputRecorder(int64_t payload_id)
+    : payload_id_(payload_id) {}
+
+ThroughputRecorderContainer& ThroughputRecorderContainer::GetInstance() {
+  static std::aligned_storage_t<sizeof(ThroughputRecorderContainer),
+                                alignof(ThroughputRecorderContainer)>
+      storage;
+  static ThroughputRecorderContainer* env =
+      new (&storage) ThroughputRecorderContainer();
+  return *env;
+}
+
+void ThroughputRecorder::Start(PayloadType payload_type,
+                               const bool is_incoming) {
+  NEARBY_LOGS(INFO) << "Start TP profiling for payload_id:" << payload_id_;
+
+  // MutexLock lock(&mutex_);
+  if (payload_type == PayloadType::kUnknown) {
+    NEARBY_LOGS(INFO)
+        << "Ignore ThroughputRecorder::start for Unknown Payload type";
+    return;
+  }
+
+  start_timestamp_ = SystemClock::ElapsedRealtime();
+  payload_type_ = payload_type;
+  is_incoming_ = is_incoming;
+  // Add packetLostAlarm later
+}
+
+bool ThroughputRecorder::Stop() {
+  NEARBY_LOGS(INFO) << "Stop TP profiling for payload_id:" << payload_id_;
+  if (payload_type_ == PayloadType::kUnknown) {
+    NEARBY_LOGS(INFO) << "Ignore ThroughputRecorder::stop as it never start";
+    return false;
+  }
+  {
+    MutexLock lock(&mutex_);
+    // 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_) {
+          tp.second.SetLastTimestamp(stop_timestamp);
+        }
+      }
+    }
+
+    // calculate throughput by medium
+    for (auto& tp : throughputs_) {
+      tp.second.dump();
+      total_byte_size += tp.second.GetTotalByteSize();
+    }
+
+    throughputs_.clear();
+
+    int64_t total_millis =
+        absl::ToInt64Milliseconds(stop_timestamp - start_timestamp_);
+    throughput_kbps_ = CalculateThroughputKBps(total_byte_size, total_millis);
+    int 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, "
+            "throughput "
+            "is %d MB/s (%d KB/s), File IO takes %d ms, %s takes %d "
+            "ms, "
+            "Socket IO takes %d ms",
+            is_incoming_ ? "Received" : "Sent", ToString(payload_type_),
+            total_byte_size, success_ ? "SUCCEEDED" : "FAILED", total_millis,
+            throughput_mbps, throughput_kbps_, file_io_time_,
+            is_incoming_ ? "Decryption" : "Encryption", encryption_time_,
+            socket_io_time_);
+        NEARBY_LOGS(INFO) << dump_content;
+      }
+    }
+  }
+  return 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) {
+  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() {
+  int64_t total_millis =
+      absl::ToInt64Milliseconds(last_timestamp_ - start_timestamp_);
+  int throughput_kbps = CalculateThroughputKBps(total_byte_size_, total_millis);
+  if (throughput_kbps == kDefaultThroughoutKbps) {
+    return false;
+  }
+  int throughpu_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",
+      is_incoming_ ? "Received" : "Sent", ToString(payload_type_),
+      total_byte_size_, proto::connections::Medium_Name(medium_), total_millis,
+      throughpu_mbps, throughput_kbps, file_io_time_,
+      is_incoming_ ? "Decryption" : "Encryption", encryption_time_,
+      socket_io_time_, other);
+  NEARBY_LOGS(INFO) << dump_content;
+  return true;
+}
+
+ThroughputRecorder::Throughput& ThroughputRecorder::GetThroughput(
+    Medium medium, int64_t duration_millis) {
+  MutexLock lock(&mutex_);
+  auto it = throughputs_.find(medium);
+  if (it == throughputs_.end()) {
+    auto throughput = new Throughput(
+        medium,
+        SystemClock::ElapsedRealtime() - absl::Milliseconds(duration_millis),
+        payload_type_, is_incoming_);
+    throughputs_.emplace(medium, std::move(*throughput));
+    delete throughput;
+    return throughputs_.find(medium)->second;
+  }
+  return it->second;
+}
+
+int ThroughputRecorder::GetThroughputsSize() {
+  MutexLock lock(&mutex_);
+  return throughputs_.size();
+}
+
+int ThroughputRecorder::GetThroughputKbps() { return throughput_kbps_; }
+
+int64_t ThroughputRecorder::GetDurationMillis() {
+  return duration_millis_;
+}
+
+void ThroughputRecorder::OnFrameSent(Medium medium,
+                                     PacketMetaData& packetMetaData) {
+  if (payload_type_ == PayloadType::kUnknown) {
+    NEARBY_LOGS(INFO) << "PayloadType is invalid, return";
+    return;
+  }
+
+  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::OnFrameReceived(Medium medium,
+                                         PacketMetaData& packetMetaData) {
+  if (payload_type_ == PayloadType::kUnknown) {
+    NEARBY_LOGS(INFO) << "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) {
+  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");
+  }
+}
+
+// Inplementation for ThroughputRecorderContainer
+
+void ThroughputRecorderContainer::Shutdown() {
+  MutexLock lock(&mutex_);
+  NEARBY_LOGS(INFO) << __func__
+                    << ".  Num of Instance:" << throughput_recorders_.size();
+  for (auto& throughput_recorder : throughput_recorders_) {
+    NEARBY_LOGS(INFO) << "Stop instance: " << throughput_recorder.second;
+    throughput_recorder.second->Stop();
+    delete throughput_recorder.second;
+  }
+  throughput_recorders_.clear();
+}
+
+ThroughputRecorder* ThroughputRecorderContainer::GetTPRecorder(
+    const int64_t payload_id) {
+  MutexLock lock(&mutex_);
+  auto it = throughput_recorders_.find(payload_id);
+  if (it == throughput_recorders_.end()) {
+    auto instance = new ThroughputRecorder(payload_id);
+    NEARBY_LOGS(INFO) << "Add ThroughputRecorder instance : " << instance
+                      << " for payload_id:" << payload_id;
+    throughput_recorders_.emplace(payload_id, instance);
+    return instance;
+  }
+
+  return it->second;
+}
+
+void ThroughputRecorderContainer::StopTPRecorder(
+    const int64_t payload_id) {
+  MutexLock lock(&mutex_);
+  auto it = throughput_recorders_.find(payload_id);
+  if (it != throughput_recorders_.end()) {
+    NEARBY_LOGS(INFO) << "Found and stop/delete ThroughputRecorder instance : "
+                      << &(it->second) << " for payload_id:" << payload_id;
+    it->second->Stop();
+    delete it->second;
+    throughput_recorders_.erase(payload_id);
+    return;
+  }
+  NEARBY_LOGS(INFO) << "No ThroughputRecorder found for :" << payload_id;
+}
+
+int ThroughputRecorderContainer::GetSize() {
+  MutexLock lock(&mutex_);
+  return throughput_recorders_.size();
+}
+
+}  // namespace analytics
+}  // namespace nearby
+}  // namespace location

connections/implementation/analytics/throughput_recorder.h

@@ -0,0 +1,139 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_
+#define NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_
+
+#include <cstdint>
+#include <string>
+
+#include "absl/container/flat_hash_map.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 location {
+namespace nearby {
+namespace analytics {
+
+using Medium = location::nearby::proto::connections::Medium;
+using PayloadType = connections::PayloadType;
+
+class ThroughputRecorder {
+ public:
+  explicit ThroughputRecorder(int64_t payload_id);
+  ~ThroughputRecorder() = default;
+
+  void Start(PayloadType payload_type, const bool is_incoming);
+  bool Stop() ABSL_LOCKS_EXCLUDED(mutex_);
+  static int CalculateThroughputKBps(int64_t total_byte_size,
+                                     int64_t total_millis);
+  static int CalculateThroughputMBps(int throughputKBps);
+
+  class Throughput {
+   public:
+    Throughput() = default;
+    ~Throughput() = default;
+    Throughput(Medium medium, absl::Time start_timestamp,
+               PayloadType payload_type, bool is_incoming)
+        : medium_(medium),
+          start_timestamp_(start_timestamp),
+          payload_type_(payload_type),
+          is_incoming_(is_incoming) {}
+
+    void Add(int frame_size, int64_t file_io_time,
+             int64_t encryption_time, int64_t socket_io_time);
+
+    void SetLastTimestamp(absl::Time time_stamp) {
+      last_timestamp_ = time_stamp;
+    }
+
+    int64_t GetTotalByteSize() { return total_byte_size_; }
+
+    bool dump();
+
+   private:
+    Medium medium_;
+    absl::Time start_timestamp_;
+    PayloadType payload_type_;
+    int64_t total_byte_size_ = 0;
+    absl::Time last_timestamp_;
+    bool is_incoming_ = false;
+    int64_t file_io_time_ = 0;
+    int64_t encryption_time_ = 0;
+    int64_t socket_io_time_ = 0;
+  };
+
+  Throughput& GetThroughput(Medium medium, int64_t duration_millis)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  int GetThroughputsSize() ABSL_LOCKS_EXCLUDED(mutex_);
+  int GetThroughputKbps();
+  int64_t GetDurationMillis();
+  void OnFrameSent(Medium medium, PacketMetaData& packetMetaData);
+  void OnFrameReceived(Medium medium, PacketMetaData& packetMetaData);
+  void MarkAsSuccess() { success_ = true; }
+
+ private:
+  void CalculateDurationTimes(PacketMetaData packetMetaData);
+  static std::string ToString(PayloadType type);
+
+  Mutex mutex_;
+  int64_t payload_id_;
+  absl::Time start_timestamp_;
+  PayloadType payload_type_ = PayloadType::kUnknown;
+  bool is_incoming_;
+  absl::flat_hash_map<Medium, Throughput> throughputs_ ABSL_GUARDED_BY(mutex_);
+  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_;
+};
+
+class ThroughputRecorderContainer {
+ public:
+  ThroughputRecorderContainer(const ThroughputRecorderContainer&) = delete;
+  ThroughputRecorderContainer& operator=(const ThroughputRecorderContainer&) =
+      delete;
+
+  static ThroughputRecorderContainer& GetInstance();
+  void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  ThroughputRecorder* GetTPRecorder(const int64_t payload_id)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  void StopTPRecorder(const int64_t payload_id)
+      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_;
+  absl::flat_hash_map<int64_t, ThroughputRecorder*> throughput_recorders_
+      ABSL_GUARDED_BY(mutex_);
+};
+
+}  // namespace analytics
+}  // namespace nearby
+}  // namespace location
+
+#endif  // NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_

connections/implementation/analytics/throughput_recorder_test.cc

@@ -0,0 +1,212 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "connections/implementation/analytics/throughput_recorder.h"
+
+#include "gmock/gmock.h"
+#include "protobuf-matchers/protocol-buffer-matchers.h"
+#include "gtest/gtest.h"
+#include "absl/time/clock.h"
+#include "internal/platform/logging.h"
+#include "proto/connections_enums.proto.h"
+
+namespace location {
+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& tp_recorder_container_ =
+      ThroughputRecorderContainer::GetInstance();
+};
+
+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);
+  TPRecorderContainer.GetTPRecorder(kPayloadIdB);
+  EXPECT_EQ(ThroughputRecorderContainer::GetInstance().GetSize(), 2);
+  ThroughputRecorderContainer::GetInstance().Shutdown();
+  EXPECT_EQ(ThroughputRecorderContainer::GetInstance().GetSize(), 0);
+}
+
+TEST_F(ThroughputRecorderTest, OnFrameSentSaveTransferredSize) {
+  auto TPRecorder = tp_recorder_container_.GetTPRecorder(kPayloadIdA);
+  TPRecorder->Start(PayloadType::kFile, /*isIncoming=*/false);
+
+  PacketMetaData packet_meta_data;
+  packet_meta_data.SetPacketSize(kFrameSize);
+  TPRecorder->OnFrameSent(proto::connections::BLE, packet_meta_data);
+  TPRecorder->OnFrameSent(proto::connections::BLE, packet_meta_data);
+  TPRecorder->OnFrameSent(proto::connections::BLE, packet_meta_data);
+
+  auto throughput = TPRecorder->GetThroughput(proto::connections::BLE, 0);
+  EXPECT_EQ(throughput.GetTotalByteSize(), kFrameSize * 3);
+}
+
+TEST_F(ThroughputRecorderTest, OnIgnoreUnkownPaylaodType) {
+  auto TPRecorder = tp_recorder_container_.GetTPRecorder(kPayloadIdA);
+  TPRecorder->Start(PayloadType::kUnknown, /*isIncoming=*/false);
+
+  PacketMetaData packet_meta_data;
+  TPRecorder->OnFrameSent(proto::connections::BLE, packet_meta_data);
+  EXPECT_EQ(TPRecorder->GetThroughputsSize(), 0);
+
+  TPRecorder->Start(PayloadType::kUnknown, /*isIncoming=*/true);
+  TPRecorder->OnFrameReceived(proto::connections::BLE, packet_meta_data);
+  EXPECT_EQ(TPRecorder->GetThroughputsSize(), 0);
+}
+
+TEST_P(ThroughputRecorderTest, OnFrameSentStopAndDump) {
+  auto TPRecorder = tp_recorder_container_.GetTPRecorder(kPayloadIdA);
+  TPRecorder->Start(PayloadType::kFile, /*isIncoming=*/false);
+
+  PacketMetaData packet_meta_data;
+  packet_meta_data.SetPacketSize(kFrameSize);
+  packet_meta_data.StartFileIo();
+  absl::SleepFor(absl::Milliseconds(5));
+  packet_meta_data.StopFileIo();
+  packet_meta_data.StartEncryption();
+  absl::SleepFor(absl::Milliseconds(6));
+  packet_meta_data.StopEncryption();
+  packet_meta_data.StartSocketIo();
+  absl::SleepFor(absl::Milliseconds(7));
+  packet_meta_data.StopSocketIo();
+  TPRecorder->OnFrameSent(proto::connections::BLE, packet_meta_data);
+  EXPECT_EQ(TPRecorder->GetDurationMillis(),
+            packet_meta_data.GetEncryptionTimeInMillis() +
+                packet_meta_data.GetFileIoTimeInMillis() +
+                packet_meta_data.GetSocketIoTimeInMillis());
+
+  packet_meta_data.SetPacketSize(kFrameSize);
+  packet_meta_data.StartFileIo();
+  absl::SleepFor(absl::Milliseconds(15));
+  packet_meta_data.StopFileIo();
+  packet_meta_data.StartEncryption();
+  absl::SleepFor(absl::Milliseconds(16));
+  packet_meta_data.StopEncryption();
+  packet_meta_data.StartSocketIo();
+  absl::SleepFor(absl::Milliseconds(17));
+  packet_meta_data.StopSocketIo();
+  TPRecorder->OnFrameSent(proto::connections::BLE, packet_meta_data);
+
+  if (GetParam() == true) {
+    NEARBY_LOGS(INFO) << "MarkAsSuccess";
+    TPRecorder->MarkAsSuccess();
+  }
+  EXPECT_TRUE(TPRecorder->Stop());
+  EXPECT_NE(TPRecorder->GetThroughputKbps(), 0);
+}
+
+TEST_F(ThroughputRecorderTest, OnFrameSentStopAndDumpForMultiMeadium) {
+  auto TPRecorder = tp_recorder_container_.GetTPRecorder(kPayloadIdA);
+  TPRecorder->Start(PayloadType::kFile, /*isIncoming=*/false);
+
+  PacketMetaData packet_meta_data1;
+  packet_meta_data1.SetPacketSize(kFrameSize);
+  packet_meta_data1.StartFileIo();
+  absl::SleepFor(absl::Milliseconds(5));
+  packet_meta_data1.StopFileIo();
+  packet_meta_data1.StartEncryption();
+  absl::SleepFor(absl::Milliseconds(6));
+  packet_meta_data1.StopEncryption();
+  packet_meta_data1.StartSocketIo();
+  absl::SleepFor(absl::Milliseconds(7));
+  packet_meta_data1.StopSocketIo();
+  TPRecorder->OnFrameSent(proto::connections::BLE, packet_meta_data1);
+
+  PacketMetaData packet_meta_data2;
+  packet_meta_data2.SetPacketSize(kFrameSize);
+  packet_meta_data2.StartFileIo();
+  absl::SleepFor(absl::Milliseconds(15));
+  packet_meta_data2.StopFileIo();
+  packet_meta_data2.StartEncryption();
+  absl::SleepFor(absl::Milliseconds(16));
+  packet_meta_data2.StopEncryption();
+  packet_meta_data2.StartSocketIo();
+  absl::SleepFor(absl::Milliseconds(17));
+  packet_meta_data2.StopSocketIo();
+  TPRecorder->OnFrameSent(proto::connections::WIFI_LAN, packet_meta_data2);
+
+  TPRecorder->MarkAsSuccess();
+  EXPECT_TRUE(TPRecorder->Stop());
+  EXPECT_NE(TPRecorder->GetThroughputKbps(), 0);
+}
+
+
+TEST_F(ThroughputRecorderTest, OnFrameReceivedCheckDurationMillis) {
+  auto TPRecorder = tp_recorder_container_.GetTPRecorder(kPayloadIdA);
+  TPRecorder->Start(PayloadType::kFile, /*isIncoming=*/true);
+
+  PacketMetaData packet_meta_data;
+  packet_meta_data.SetPacketSize(kFrameSize);
+  packet_meta_data.StartFileIo();
+  absl::SleepFor(absl::Milliseconds(5));
+  packet_meta_data.StopFileIo();
+  packet_meta_data.StartEncryption();
+  absl::SleepFor(absl::Milliseconds(6));
+  packet_meta_data.StopEncryption();
+  packet_meta_data.StartSocketIo();
+  absl::SleepFor(absl::Milliseconds(7));
+  packet_meta_data.StopSocketIo();
+  TPRecorder->OnFrameReceived(proto::connections::BLE, packet_meta_data);
+  EXPECT_EQ(TPRecorder->GetDurationMillis(),
+            packet_meta_data.GetEncryptionTimeInMillis() +
+                packet_meta_data.GetFileIoTimeInMillis() +
+                packet_meta_data.GetSocketIoTimeInMillis());
+}
+
+TEST_F(ThroughputRecorderTest, OnTPRecorderNotStarted) {
+  auto TPRecorder = tp_recorder_container_.GetTPRecorder(kPayloadIdA);
+  auto throughput = TPRecorder->GetThroughput(proto::connections::BLE, 0);
+  EXPECT_FALSE(throughput.dump());
+}
+
+}  // namespace
+}  // namespace analytics
+}  // namespace nearby
+}  // namespace location

connections/implementation/base_endpoint_channel.cc

@@ -15,9 +15,10 @@
 #include "connections/implementation/base_endpoint_channel.h"
 
 #include <cassert>
+#include <memory>
 #include <string>
+#include <utility>
 
-#include "absl/strings/escaping.h"
 #include "absl/strings/str_cat.h"
 #include "connections/implementation/offline_frames.h"
 #include "internal/platform/byte_array.h"
@@ -122,10 +123,17 @@ BaseEndpointChannel::BaseEndpointChannel(
       try_count_(try_count) {}
 
 ExceptionOr<ByteArray> BaseEndpointChannel::Read() {
+  PacketMetaData packet_meta_data;
+  return Read(packet_meta_data);
+}
+
+ExceptionOr<ByteArray> BaseEndpointChannel::Read(
+    PacketMetaData& packet_meta_data) {
   ByteArray result;
   {
     MutexLock lock(&reader_mutex_);
 
+    packet_meta_data.StartSocketIo();
     ExceptionOr<std::int32_t> read_int = ReadInt(reader_);
     if (!read_int.ok()) {
       return ExceptionOr<ByteArray>(read_int.exception());
@@ -141,6 +149,8 @@ ExceptionOr<ByteArray> BaseEndpointChannel::Read() {
     if (!read_bytes.ok()) {
       return read_bytes;
     }
+    packet_meta_data.StopSocketIo();
+    packet_meta_data.SetPacketSize(read_int.result() + sizeof(std::int32_t));
     result = std::move(read_bytes.result());
   }
 
@@ -149,6 +159,7 @@ ExceptionOr<ByteArray> BaseEndpointChannel::Read() {
     if (IsEncryptionEnabledLocked()) {
       // If encryption is enabled, decode the message.
       std::string input(std::move(result));
+      packet_meta_data.StartEncryption();
       std::unique_ptr<std::string> decrypted_data =
           crypto_context_->DecodeMessageFromPeer(input);
       if (decrypted_data) {
@@ -178,6 +189,7 @@ ExceptionOr<ByteArray> BaseEndpointChannel::Read() {
               << __func__ << ": Unable to parse data as unencrypted message.";
         }
       }
+      packet_meta_data.StopEncryption();
       if (result.Empty()) {
         NEARBY_LOGS(WARNING) << __func__ << ": Unable to parse read result.";
         return ExceptionOr<ByteArray>(Exception::kInvalidProtocolBuffer);
@@ -193,6 +205,12 @@ ExceptionOr<ByteArray> BaseEndpointChannel::Read() {
 }
 
 Exception BaseEndpointChannel::Write(const ByteArray& data) {
+  PacketMetaData packet_meta_data;
+  return Write(data, packet_meta_data);
+}
+
+Exception BaseEndpointChannel::Write(const ByteArray& data,
+                                     PacketMetaData& packet_meta_data) {
   {
     MutexLock pause_lock(&is_paused_mutex_);
     if (is_paused_) {
@@ -212,8 +230,10 @@ Exception BaseEndpointChannel::Write(const ByteArray& data) {
       MutexLock crypto_lock(&crypto_mutex_);
       if (IsEncryptionEnabledLocked()) {
         // If encryption is enabled, encode the message.
+        packet_meta_data.StartEncryption();
         std::unique_ptr<std::string> encrypted =
             crypto_context_->EncodeMessageToPeer(std::string(data));
+        packet_meta_data.StopEncryption();
         if (!encrypted) {
           NEARBY_LOGS(WARNING) << __func__ << ": Failed to encrypt data.";
           return {Exception::kIo};
@@ -230,6 +250,7 @@ Exception BaseEndpointChannel::Write(const ByteArray& data) {
       return {Exception::kIo};
     }
 
+    packet_meta_data.StartSocketIo();
     Exception write_exception =
         WriteInt(writer_, static_cast<std::int32_t>(data_size));
     if (write_exception.Raised()) {
@@ -249,6 +270,8 @@ Exception BaseEndpointChannel::Write(const ByteArray& data) {
                            << flush_exception.value;
       return flush_exception;
     }
+    packet_meta_data.StopSocketIo();
+    packet_meta_data.SetPacketSize(data_size + sizeof(std::uint32_t));
   }
 
   {

connections/implementation/base_endpoint_channel.h

@@ -19,22 +19,21 @@
 #include <memory>
 #include <string>
 
-#include "securegcm/d2d_connection_context_v1.h"
 #include "absl/base/thread_annotations.h"
 #include "connections/implementation/analytics/analytics_recorder.h"
 #include "connections/implementation/endpoint_channel.h"
-#include "internal/platform/atomic_reference.h"
 #include "internal/platform/byte_array.h"
 #include "internal/platform/condition_variable.h"
 #include "internal/platform/input_stream.h"
 #include "internal/platform/mutex.h"
 #include "internal/platform/output_stream.h"
-#include "internal/platform/system_clock.h"
 
 namespace location {
 namespace nearby {
 namespace connections {
 
+using analytics::PacketMetaData;
+
 class BaseEndpointChannel : public EndpointChannel {
  public:
   BaseEndpointChannel(const std::string& service_id,
@@ -49,10 +48,12 @@ class BaseEndpointChannel : public EndpointChannel {
   ~BaseEndpointChannel() override = default;
 
   // EndpointChannel:
-  ExceptionOr<ByteArray> Read()
+  ExceptionOr<ByteArray> Read() override;
+  ExceptionOr<ByteArray> Read(PacketMetaData& packet_meta_data)
       ABSL_LOCKS_EXCLUDED(reader_mutex_, crypto_mutex_,
                           last_read_mutex_) override;
-  Exception Write(const ByteArray& data)
+  Exception Write(const ByteArray& data) override;
+  Exception Write(const ByteArray& data, PacketMetaData& packet_meta_data)
       ABSL_LOCKS_EXCLUDED(writer_mutex_, crypto_mutex_) override;
   void Close() ABSL_LOCKS_EXCLUDED(is_paused_mutex_) override;
   void Close(proto::connections::DisconnectionReason reason) override;

connections/implementation/base_pcp_handler.cc

@@ -905,7 +905,8 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client,
 void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame,
                                      const std::string& endpoint_id,
                                      ClientProxy* client,
-                                     proto::connections::Medium medium) {
+                                     proto::connections::Medium medium,
+                                     PacketMetaData& packet_meta_data) {
   CountDownLatch latch(1);
   RunOnPcpHandlerThread(
       "incoming-frame",

connections/implementation/base_pcp_handler.h

@@ -134,7 +134,8 @@ class BasePcpHandler : public PcpHandler,
   // @EndpointManagerReaderThread
   void OnIncomingFrame(OfflineFrame& frame, const std::string& endpoint_id,
                        ClientProxy* client,
-                       proto::connections::Medium medium) override;
+                       proto::connections::Medium medium,
+                       analytics::PacketMetaData& packet_meta_data) override;
 
   // Called when an endpoint disconnects while we're waiting for both sides to
   // approve/reject the connection.

connections/implementation/base_pcp_handler_test.cc

@@ -627,6 +627,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
   EndpointManager em(&ecm);
   BwuManager bwu(m, em, ecm, {}, {});
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
+  analytics::PacketMetaData packet_meta_data;
   StartDiscovery(&client, &pcp_handler);
   auto mediums = pcp_handler.GetDiscoveryMediums(&client);
   auto connect_medium = mediums[mediums.size() - 1];
@@ -647,7 +648,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
   auto frame =
       parser::FromBytes(parser::ForConnectionResponse(Status::kSuccess));
   pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, &client,
-                              connect_medium);
+                              connect_medium, packet_meta_data);
   NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
   channel_b->Close();
   bwu.Shutdown();

connections/implementation/bwu_manager.cc

@@ -291,7 +291,8 @@ void BwuManager::InitiateBwuForEndpoint(ClientProxy* client,
 
 void BwuManager::OnIncomingFrame(OfflineFrame& frame,
                                  const std::string& endpoint_id,
-                                 ClientProxy* client, Medium medium) {
+                                 ClientProxy* client, Medium medium,
+                                 PacketMetaData& packet_meta_data) {
   V1Frame::FrameType frame_type = parser::GetFrameType(frame);
   if (frame_type != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION) return;
 

connections/implementation/bwu_manager.h

@@ -86,7 +86,8 @@ class BwuManager : public EndpointManager::FrameProcessor {
   // inbound BWU protocol.
   // @EndpointManagerReaderThread
   void OnIncomingFrame(OfflineFrame& frame, const std::string& endpoint_id,
-                       ClientProxy* client, Medium medium) override;
+                       ClientProxy* client, Medium medium,
+                       PacketMetaData& packet_meta_data) override;
 
   // Cleans up in-progress upgrades after endpoint disconnection.
   // @EndpointManagerReaderThread

connections/implementation/bwu_manager_test.cc

@@ -14,6 +14,7 @@
 
 #include "connections/implementation/bwu_manager.h"
 
+#include <memory>
 #include <string>
 #include <utility>
 
@@ -119,12 +120,12 @@ class BwuManagerTest : public ::testing::Test {
         parser::FromBytes(parser::ForBwuLastWrite());
     bwu_manager_->OnIncomingFrame(last_write_frame.result(),
                                   std::string(endpoint_id), &client_,
-                                  initial_medium);
+                                  initial_medium, packet_meta_data_);
     ExceptionOr<OfflineFrame> safe_to_close_frame =
         parser::FromBytes(parser::ForBwuSafeToClose());
     bwu_manager_->OnIncomingFrame(safe_to_close_frame.result(),
                                   std::string(endpoint_id), &client_,
-                                  initial_medium);
+                                  initial_medium, packet_meta_data_);
 
     return upgraded_channel;
   }
@@ -139,6 +140,7 @@ class BwuManagerTest : public ::testing::Test {
   FakeBwuHandler* fake_wifi_lan_bwu_handler_ = nullptr;
   FakeBwuHandler* fake_wifi_hotspot_bwu_handler_ = nullptr;
   std::unique_ptr<BwuManager> bwu_manager_;
+  PacketMetaData packet_meta_data_;
 };
 
 class BwuManagerTestParam : public BwuManagerTest,
@@ -193,12 +195,12 @@ TEST_P(BwuManagerTestParam, InitiateBwu_Success) {
       parser::FromBytes(parser::ForBwuLastWrite());
   bwu_manager_->OnIncomingFrame(last_write_frame.result(),
                                 std::string(kEndpointId1), &client_,
-                                Medium::BLUETOOTH);
+                                Medium::BLUETOOTH, packet_meta_data_);
   ExceptionOr<OfflineFrame> safe_to_close_frame =
       parser::FromBytes(parser::ForBwuSafeToClose());
   bwu_manager_->OnIncomingFrame(safe_to_close_frame.result(),
                                 std::string(kEndpointId1), &client_,
-                                Medium::BLUETOOTH);
+                                Medium::BLUETOOTH, packet_meta_data_);
 
   // Confirm that upgrade channel is resumed after initial channel is shut down.
   // Note: If we didn't grab the shared initial channel pointer above, this
@@ -626,7 +628,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnUpgradeFailure_FlagEnabled) {
       parser::FromBytes(parser::ForBwuFailure(info));
   bwu_manager_->OnIncomingFrame(upgrade_failure.result(),
                                 std::string(kEndpointId3), &client_,
-                                Medium::WEB_RTC);
+                                Medium::WEB_RTC, packet_meta_data_);
 
     // With the flag enabled, we can safely revert WebRTC just for service B
     // because service B has no active WebRTC endpoints.
@@ -661,7 +663,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnUpgradeFailure_FlagDisabled) {
       parser::FromBytes(parser::ForBwuFailure(info));
   bwu_manager_->OnIncomingFrame(upgrade_failure.result(),
                                 std::string(kEndpointId3), &client_,
-                                Medium::WEB_RTC);
+                                Medium::WEB_RTC, packet_meta_data_);
 
   // With the flag disabled, we don't revert if there are still connected
   // endpoints for _any_ service. We don't have service-level bookkeeping; we
@@ -688,7 +690,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Hotspot) {
   upgrade_path_info->set_supports_client_introduction_ack(false);
   upgrade_path_info->set_supports_disabling_encryption(true);
   bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId1), &client_,
-                                Medium::BLUETOOTH);
+                                Medium::BLUETOOTH, packet_meta_data_);
   CountDownLatch latch(1);
   bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA,
                                      std::string(kEndpointId1), latch);
@@ -714,7 +716,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Wlan) {
   upgrade_path_info->set_supports_client_introduction_ack(false);
 
   bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId1), &client_,
-                                Medium::BLUETOOTH);
+                                Medium::BLUETOOTH, packet_meta_data_);
   CountDownLatch latch(1);
   bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA,
                                      std::string(kEndpointId1), latch);

connections/implementation/encryption_runner_test.cc

@@ -42,10 +42,20 @@ class FakeEndpointChannel : public EndpointChannel {
     return in_ ? in_->Read(Pipe::kChunkSize)
                : ExceptionOr<ByteArray>{Exception::kIo};
   }
+  ExceptionOr<ByteArray> Read(PacketMetaData& packet_meta_data) override {
+    read_timestamp_ = SystemClock::ElapsedRealtime();
+    return in_ ? in_->Read(Pipe::kChunkSize)
+               : ExceptionOr<ByteArray>{Exception::kIo};
+  }
   Exception Write(const ByteArray& data) override {
     write_timestamp_ = SystemClock::ElapsedRealtime();
     return out_ ? out_->Write(data) : Exception{Exception::kIo};
   }
+  Exception Write(const ByteArray& data,
+                  PacketMetaData& packet_meta_data) override {
+    write_timestamp_ = SystemClock::ElapsedRealtime();
+    return out_ ? out_->Write(data) : Exception{Exception::kIo};
+  }
   void Close() override {
     if (in_) in_->Close();
     if (out_) out_->Close();

connections/implementation/endpoint_channel.h

@@ -19,17 +19,18 @@
 #include <string>
 
 #include "securegcm/d2d_connection_context_v1.h"
-#include "absl/time/clock.h"
 #include "connections/implementation/analytics/analytics_recorder.h"
+#include "connections/implementation/analytics/packet_meta_data.h"
 #include "internal/platform/byte_array.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/mutex.h"
-#include "proto/connections_enums.pb.h"
 
 namespace location {
 namespace nearby {
 namespace connections {
 
+using analytics::PacketMetaData;
+
 class EndpointChannel {
  public:
   virtual ~EndpointChannel() = default;
@@ -39,9 +40,15 @@ class EndpointChannel {
   virtual ExceptionOr<ByteArray>
   Read() = 0;  // throws Exception::IO, Exception::INTERRUPTED
 
+  virtual ExceptionOr<ByteArray> Read(PacketMetaData& packet_meta_data) = 0;
+
   virtual Exception Write(const ByteArray& data) = 0;  // throws Exception::IO
 
+  virtual Exception Write(
+      const ByteArray& data,
+      PacketMetaData& packet_meta_data) = 0;  // throws Exception::IO
   // Closes this EndpointChannel, without tracking the closure in analytics.
+
   virtual void Close() = 0;
 
   // Closes this EndpointChannel and records the closure with the given reason.

connections/implementation/endpoint_manager.cc

@@ -19,7 +19,9 @@
 #include <memory>
 #include <string>
 #include <utility>
+#include <vector>
 
+#include "connections/implementation/analytics/throughput_recorder.h"
 #include "connections/implementation/endpoint_channel.h"
 #include "connections/implementation/offline_frames.h"
 #include "connections/implementation/proto/offline_wire_formats.pb.h"
@@ -167,7 +169,8 @@ ExceptionOr<bool> EndpointManager::HandleData(
   // a replacement for this endpoint since we last checked with the
   // EndpointChannelManager.
   while (true) {
-    ExceptionOr<ByteArray> bytes = endpoint_channel->Read();
+    PacketMetaData packet_meta_data;
+    ExceptionOr<ByteArray> bytes = endpoint_channel->Read(packet_meta_data);
     if (!bytes.ok()) {
       NEARBY_LOG(INFO, "Stop reading on read-time exception: %d",
                  bytes.exception());
@@ -210,7 +213,8 @@ ExceptionOr<bool> EndpointManager::HandleData(
     }
 
     frame_processor->OnIncomingFrame(frame, endpoint_id, client,
-                                     endpoint_channel->GetMedium());
+                                     endpoint_channel->GetMedium(),
+                                     packet_meta_data);
   }
 }
 
@@ -279,6 +283,7 @@ EndpointManager::EndpointManager(EndpointChannelManager* manager)
 
 EndpointManager::~EndpointManager() {
   NEARBY_LOG(INFO, "Initiating shutdown of EndpointManager.");
+  analytics::ThroughputRecorderContainer::GetInstance().Shutdown();
   CountDownLatch latch(1);
   RunOnEndpointManagerThread("bring-down-endpoints", [this, &latch]() {
     NEARBY_LOG(INFO, "Bringing down endpoints");
@@ -493,7 +498,8 @@ int EndpointManager::GetMaxTransmitPacketSize(const std::string& endpoint_id) {
 std::vector<std::string> EndpointManager::SendPayloadChunk(
     const PayloadTransferFrame::PayloadHeader& payload_header,
     const PayloadTransferFrame::PayloadChunk& payload_chunk,
-    const std::vector<std::string>& endpoint_ids) {
+    const std::vector<std::string>& endpoint_ids,
+    PacketMetaData& packet_meta_data) {
   ByteArray bytes =
       parser::ForDataPayloadTransfer(payload_header, payload_chunk);
 
@@ -501,7 +507,8 @@ std::vector<std::string> EndpointManager::SendPayloadChunk(
       endpoint_ids, bytes, payload_header.id(),
       /*offset=*/payload_chunk.offset(),
       /*packet_type=*/
-      PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::DATA));
+      PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::DATA),
+      packet_meta_data);
 }
 
 // Designed to run asynchronously. It is called from IO thread pools, and
@@ -521,12 +528,14 @@ std::vector<std::string> EndpointManager::SendControlMessage(
     const PayloadTransferFrame::ControlMessage& control,
     const std::vector<std::string>& endpoint_ids) {
   ByteArray bytes = parser::ForControlPayloadTransfer(header, control);
+  PacketMetaData packet_meta_data;
 
   return SendTransferFrameBytes(
       endpoint_ids, bytes, header.id(),
       /*offset=*/control.offset(),
       /*packet_type=*/
-      PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::CONTROL));
+      PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::CONTROL),
+      packet_meta_data);
 }
 
 // @EndpointManagerThread
@@ -616,7 +625,7 @@ CountDownLatch EndpointManager::NotifyFrameProcessorsOnEndpointDisconnect(
 std::vector<std::string> EndpointManager::SendTransferFrameBytes(
     const std::vector<std::string>& endpoint_ids, const ByteArray& bytes,
     std::int64_t payload_id, std::int64_t offset,
-    const std::string& packet_type) {
+    const std::string& packet_type, PacketMetaData& packet_meta_data) {
   std::vector<std::string> failed_endpoint_ids;
   for (const std::string& endpoint_id : endpoint_ids) {
     std::shared_ptr<EndpointChannel> channel =
@@ -635,12 +644,15 @@ std::vector<std::string> EndpointManager::SendTransferFrameBytes(
       continue;
     }
 
-    Exception write_exception = channel->Write(bytes);
+    Exception write_exception = channel->Write(bytes, packet_meta_data);
     if (!write_exception.Ok()) {
       failed_endpoint_ids.push_back(endpoint_id);
       NEARBY_LOGS(INFO) << "Failed to send packet; endpoint_id=" << endpoint_id;
       continue;
     }
+    analytics::ThroughputRecorderContainer::GetInstance()
+        .GetTPRecorder(payload_id)
+        ->OnFrameSent(channel->GetMedium(), packet_meta_data);
   }
 
   return failed_endpoint_ids;

connections/implementation/endpoint_manager.h

@@ -20,7 +20,9 @@
 #include <memory>
 #include <utility>
 #include <string>
+#include <vector>
 
+#include "connections/implementation/analytics/packet_meta_data.h"
 #include "absl/base/thread_annotations.h"
 #include "absl/container/flat_hash_map.h"
 #include "absl/container/flat_hash_set.h"
@@ -33,10 +35,8 @@
 #include "internal/platform/byte_array.h"
 #include "internal/platform/condition_variable.h"
 #include "internal/platform/count_down_latch.h"
-#include "internal/platform/multi_thread_executor.h"
 #include "internal/platform/runnable.h"
 #include "internal/platform/single_thread_executor.h"
-#include "internal/platform/system_clock.h"
 
 namespace location {
 namespace nearby {
@@ -60,6 +60,8 @@ namespace connections {
 // to PayloadManager::ProcessFrame() (still running on that
 // same dedicated reader thread).
 
+using analytics::PacketMetaData;
+
 class EndpointManager {
  public:
   class FrameProcessor {
@@ -77,7 +79,8 @@ class EndpointManager {
     virtual void OnIncomingFrame(OfflineFrame& offline_frame,
                                  const std::string& from_endpoint_id,
                                  ClientProxy* to_client,
-                                 proto::connections::Medium current_medium) = 0;
+                                 proto::connections::Medium current_medium,
+                                 PacketMetaData& packet_meta_data) = 0;
 
     // Implementations must call barrier.CountDown() once
     // they're done. This parallelizes the disconnection event across all frame
@@ -125,7 +128,8 @@ class EndpointManager {
   std::vector<std::string> SendPayloadChunk(
       const PayloadTransferFrame::PayloadHeader& payload_header,
       const PayloadTransferFrame::PayloadChunk& payload_chunk,
-      const std::vector<std::string>& endpoint_ids);
+      const std::vector<std::string>& endpoint_ids,
+      PacketMetaData& packet_meta_data);
   std::vector<std::string> SendControlMessage(
       const PayloadTransferFrame::PayloadHeader& payload_header,
       const PayloadTransferFrame::ControlMessage& control_message,
@@ -261,7 +265,8 @@ class EndpointManager {
   std::vector<std::string> SendTransferFrameBytes(
       const std::vector<std::string>& endpoint_ids,
       const ByteArray& payload_transfer_frame_bytes, std::int64_t payload_id,
-      std::int64_t offset, const std::string& packet_type);
+      std::int64_t offset, const std::string& packet_type,
+      PacketMetaData& packet_meta_data);
 
   // Executes all jobs sequentially, on a serial_executor_.
   void RunOnEndpointManagerThread(const std::string& name, Runnable runnable);

connections/implementation/endpoint_manager_test.cc

@@ -18,6 +18,7 @@
 #include <memory>
 #include <string>
 #include <utility>
+#include <vector>
 
 #include "gmock/gmock.h"
 #include "protobuf-matchers/protocol-buffer-matchers.h"
@@ -51,7 +52,12 @@ using ::testing::StrictMock;
 class MockEndpointChannel : public EndpointChannel {
  public:
   MOCK_METHOD(ExceptionOr<ByteArray>, Read, (), (override));
+  MOCK_METHOD(ExceptionOr<ByteArray>, Read, (PacketMetaData & packet_meta_data),
+              (override));
   MOCK_METHOD(Exception, Write, (const ByteArray& data), (override));
+  MOCK_METHOD(Exception, Write,
+              (const ByteArray& data, PacketMetaData& packet_meta_data),
+              (override));
   MOCK_METHOD(void, Close, (), (override));
   MOCK_METHOD(void, Close, (DisconnectionReason reason), (override));
   MOCK_METHOD(proto::connections::ConnectionTechnology, GetTechnology, (),
@@ -95,7 +101,7 @@ class MockFrameProcessor : public EndpointManager::FrameProcessor {
   MOCK_METHOD(void, OnIncomingFrame,
               (OfflineFrame & offline_frame,
                const std::string& from_endpoint_id, ClientProxy* to_client,
-               Medium current_medium),
+               Medium current_medium, PacketMetaData& packet_meta_data),
               (override));
 
   MOCK_METHOD(void, OnEndpointDisconnect,
@@ -210,7 +216,7 @@ TEST_F(EndpointManagerTest, RegisterFrameProcessorWorks) {
   auto read_data = parser::ForConnectionRequest(connection_info);
   EXPECT_CALL(*connect_request, OnIncomingFrame);
   EXPECT_CALL(*connect_request, OnEndpointDisconnect);
-  EXPECT_CALL(*endpoint_channel, Read())
+  EXPECT_CALL(*endpoint_channel, Read(_))
       .WillOnce(Return(ExceptionOr<ByteArray>(read_data)))
       .WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
   EXPECT_CALL(*endpoint_channel, Write(_))
@@ -256,7 +262,7 @@ TEST_F(EndpointManagerTest, SendControlMessageWorks) {
   control.set_offset(150);
   control.set_event(PayloadTransferFrame::ControlMessage::PAYLOAD_CANCELED);
 
-  ON_CALL(*endpoint_channel, Read())
+  ON_CALL(*endpoint_channel, Read(_))
       .WillByDefault([channel = endpoint_channel.get()]() {
         if (channel->IsClosed()) return ExceptionOr<ByteArray>(Exception::kIo);
         NEARBY_LOG(INFO, "Simulate read delay: wait");
@@ -271,7 +277,7 @@ TEST_F(EndpointManagerTest, SendControlMessageWorks) {
             channel->DoClose();
             NEARBY_LOG(INFO, "Channel closed");
           });
-  EXPECT_CALL(*endpoint_channel, Write(_))
+  EXPECT_CALL(*endpoint_channel, Write(_, _))
       .WillRepeatedly(Return(Exception{Exception::kSuccess}));
 
   RegisterEndpoint(std::move(endpoint_channel), false);
@@ -285,7 +291,7 @@ TEST_F(EndpointManagerTest, SendControlMessageWorks) {
 
 TEST_F(EndpointManagerTest, SingleReadOnInvalidPayload) {
   auto endpoint_channel = std::make_unique<MockEndpointChannel>();
-  EXPECT_CALL(*endpoint_channel, Read())
+  EXPECT_CALL(*endpoint_channel, Read(_))
       .WillOnce(
           Return(ExceptionOr<ByteArray>(Exception::kInvalidProtocolBuffer)));
   EXPECT_CALL(*endpoint_channel, Write(_))

connections/implementation/fake_endpoint_channel.h

@@ -40,10 +40,19 @@ class FakeEndpointChannel : public EndpointChannel {
     read_timestamp_ = SystemClock::ElapsedRealtime();
     return read_output_;
   }
+  ExceptionOr<ByteArray> Read(PacketMetaData& packet_meta_data) override {
+    read_timestamp_ = SystemClock::ElapsedRealtime();
+    return read_output_;
+  }
   Exception Write(const ByteArray& data) override {
     write_timestamp_ = SystemClock::ElapsedRealtime();
     return write_output_;
   }
+  Exception Write(const ByteArray& data,
+                  PacketMetaData& packet_meta_data) override {
+    write_timestamp_ = SystemClock::ElapsedRealtime();
+    return write_output_;
+  }
   void Close() override { is_closed_ = true; }
   void Close(proto::connections::DisconnectionReason reason) override {
     is_closed_ = true;

connections/implementation/payload_manager.cc

@@ -23,16 +23,20 @@
 #include "absl/memory/memory.h"
 #include "absl/strings/str_cat.h"
 #include "absl/time/time.h"
+#include "connections/implementation/analytics/throughput_recorder.h"
 #include "connections/implementation/internal_payload_factory.h"
 #include "internal/platform/count_down_latch.h"
+#include "internal/platform/logging.h"
 #include "internal/platform/mutex_lock.h"
 #include "internal/platform/single_thread_executor.h"
-#include "internal/platform/system_clock.h"
 
 namespace location {
 namespace nearby {
 namespace connections {
 
+using analytics::PacketMetaData;
+using analytics::ThroughputRecorderContainer;
+
 // C++14 requires to declare this.
 // TODO(apolyudov): remove when migration to c++17 is possible.
 constexpr const absl::Duration PayloadManager::kWaitCloseTimeout;
@@ -46,6 +50,7 @@ bool PayloadManager::SendPayloadLoop(
   const EndpointIds& available_endpoint_ids =
       EndpointsToEndpointIds(pair.first);
   const Endpoints& unavailable_endpoints = pair.second;
+  PacketMetaData packet_meta_data;
 
   // First, handle any non-available endpoints.
   for (const auto& endpoint : unavailable_endpoints) {
@@ -106,8 +111,10 @@ bool PayloadManager::SendPayloadLoop(
   // This will block if there is no data to transfer.
   // It will resume when new data arrives, or if Close() is called.
   int chunk_size = GetOptimalChunkSize(available_endpoint_ids);
+  packet_meta_data.StartFileIo();
   ByteArray next_chunk =
       pending_payload.GetInternalPayload()->DetachNextChunk(chunk_size);
+  packet_meta_data.StopFileIo();
   if (shutdown_.Get()) return false;
   // Save chunk size. We'll need it after we move next_chunk.
   auto next_chunk_size = next_chunk.size();
@@ -130,7 +137,7 @@ bool PayloadManager::SendPayloadLoop(
   PayloadTransferFrame::PayloadChunk payload_chunk(CreatePayloadChunk(
       next_chunk_offset - resume_offset, std::move(next_chunk)));
   const EndpointIds& failed_endpoint_ids = endpoint_manager_->SendPayloadChunk(
-      payload_header, payload_chunk, available_endpoint_ids);
+      payload_header, payload_chunk, available_endpoint_ids, packet_meta_data);
   // Check whether at least one endpoint failed.
   if (!failed_endpoint_ids.empty()) {
     NEARBY_LOGS(INFO) << "Payload xfer: endpoints failed: payload_id="
@@ -163,6 +170,9 @@ bool PayloadManager::SendPayloadLoop(
       NEARBY_LOGS(INFO) << "Payload xfer done: payload_id="
                         << pending_payload.GetInternalPayload()->GetId()
                         << "; size=" << next_chunk_offset;
+      ThroughputRecorderContainer::GetInstance()
+          .GetTPRecorder(pending_payload.GetInternalPayload()->GetId())
+          ->MarkAsSuccess();
       return false;
     }
   }
@@ -305,6 +315,7 @@ void PayloadManager::DisconnectFromEndpointManager() {
 
 PayloadManager::~PayloadManager() {
   NEARBY_LOG(INFO, "PayloadManager: going down; self=%p", this);
+  ThroughputRecorderContainer::GetInstance().Shutdown();
   DisconnectFromEndpointManager();
   CancelAllPayloads();
   NEARBY_LOG(INFO, "PayloadManager: turn down payload executors; self=%p",
@@ -424,11 +435,17 @@ void PayloadManager::SendPayload(ClientProxy* client,
 
         bool should_continue = true;
         std::int64_t next_chunk_offset = 0;
+
+        ThroughputRecorderContainer::GetInstance()
+            .GetTPRecorder(payload_id)
+            ->Start(payload_type, /*isIncoming=*/false);
         while (should_continue && !shutdown_.Get()) {
           should_continue =
               SendPayloadLoop(client, *pending_payload, payload_header,
                               next_chunk_offset, resume_offset);
         }
+
+        ThroughputRecorderContainer::GetInstance().StopTPRecorder(payload_id);
         RunOnStatusUpdateThread("destroy-payload",
                                 [this, payload_id]()
                                     RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() {
@@ -468,9 +485,11 @@ Status PayloadManager::CancelPayload(ClientProxy* client,
 }
 
 // @EndpointManagerDataPool
-void PayloadManager::OnIncomingFrame(
-    OfflineFrame& offline_frame, const std::string& from_endpoint_id,
-    ClientProxy* to_client, proto::connections::Medium current_medium) {
+void PayloadManager::OnIncomingFrame(OfflineFrame& offline_frame,
+                                     const std::string& from_endpoint_id,
+                                     ClientProxy* to_client,
+                                     proto::connections::Medium current_medium,
+                                     PacketMetaData& packet_meta_data) {
   PayloadTransferFrame& frame =
       *offline_frame.mutable_v1()->mutable_payload_transfer();
 
@@ -481,7 +500,8 @@ void PayloadManager::OnIncomingFrame(
       ProcessControlPacket(to_client, from_endpoint_id, frame);
       break;
     case PayloadTransferFrame::DATA:
-      ProcessDataPacket(to_client, from_endpoint_id, frame);
+      ProcessDataPacket(to_client, from_endpoint_id, frame, current_medium,
+                        packet_meta_data);
       break;
     default:
       NEARBY_LOGS(WARNING)
@@ -805,6 +825,8 @@ void PayloadManager::HandleFinishedIncomingPayload(
     ClientProxy* client, const std::string& endpoint_id,
     const PayloadTransferFrame::PayloadHeader& payload_header,
     std::int64_t offset_bytes, proto::connections::PayloadStatus status) {
+  ThroughputRecorderContainer::GetInstance().StopTPRecorder(
+      payload_header.id());
   SendClientCallbacksForFinishedIncomingPayload(
       client, endpoint_id, payload_header, offset_bytes, status);
 
@@ -941,7 +963,8 @@ void PayloadManager::HandleSuccessfulIncomingChunk(
 // @EndpointManagerDataPool
 void PayloadManager::ProcessDataPacket(
     ClientProxy* to_client, const std::string& from_endpoint_id,
-    PayloadTransferFrame& payload_transfer_frame) {
+    PayloadTransferFrame& payload_transfer_frame, Medium medium,
+    PacketMetaData& packet_meta_data) {
   PayloadTransferFrame::PayloadHeader& payload_header =
       *payload_transfer_frame.mutable_payload_header();
   PayloadTransferFrame::PayloadChunk& payload_chunk =
@@ -953,6 +976,10 @@ void PayloadManager::ProcessDataPacket(
 
   PendingPayload* pending_payload;
   if (payload_chunk.offset() == 0) {
+    ThroughputRecorderContainer::GetInstance()
+        .GetTPRecorder(payload_header.id())
+        ->Start((PayloadType)payload_header.type(), /*isIncoming=*/true);
+    packet_meta_data.Reset();
     RunOnStatusUpdateThread(
         "process-data-packet", [to_client, from_endpoint_id, payload_header,
                                 this]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() {
@@ -1024,6 +1051,7 @@ void PayloadManager::ProcessDataPacket(
 
   // Save size of packet before we move it.
   std::int64_t payload_body_size = payload_chunk.body().size();
+  packet_meta_data.StartFileIo();
   if (pending_payload->GetInternalPayload()
           ->AttachNextChunk(ByteArray(std::move(*payload_chunk.mutable_body())))
           .Raised()) {
@@ -1035,10 +1063,25 @@ void PayloadManager::ProcessDataPacket(
         proto::connections::PayloadStatus::LOCAL_ERROR);
     return;
   }
+  packet_meta_data.StopFileIo();
 
   HandleSuccessfulIncomingChunk(to_client, from_endpoint_id, payload_header,
                                 payload_chunk.flags(), payload_chunk.offset(),
                                 payload_body_size);
+
+  ThroughputRecorderContainer::GetInstance()
+      .GetTPRecorder(payload_header.id())
+      ->OnFrameReceived(medium, packet_meta_data);
+  bool is_last_chunk = (payload_chunk.flags() &
+                        PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0;
+  if (is_last_chunk) {
+    ThroughputRecorderContainer::GetInstance()
+        .GetTPRecorder(payload_header.id())
+        ->MarkAsSuccess();
+
+    ThroughputRecorderContainer::GetInstance().StopTPRecorder(
+        payload_header.id());
+  }
 }
 
 // @EndpointManagerDataPool

connections/implementation/payload_manager.h

@@ -16,8 +16,10 @@
 #define CORE_INTERNAL_PAYLOAD_MANAGER_H_
 
 #include <cstdint>
+#include <functional>
 #include <memory>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "absl/container/flat_hash_map.h"
@@ -37,6 +39,8 @@ namespace location {
 namespace nearby {
 namespace connections {
 
+using analytics::PacketMetaData;
+
 // Annotations for methods that need to run on PayloadStatusUpdateThread.
 // Use only in PayloadManager
 #define RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() \
@@ -59,7 +63,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
   void OnIncomingFrame(OfflineFrame& offline_frame,
                        const std::string& from_endpoint_id,
                        ClientProxy* to_client,
-                       proto::connections::Medium current_medium) override;
+                       proto::connections::Medium current_medium,
+                       PacketMetaData& packet_meta_data) override;
 
   // @EndpointManagerThread
   void OnEndpointDisconnect(ClientProxy* client, const std::string& service_id,
@@ -268,7 +273,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
 
   void ProcessDataPacket(ClientProxy* to_client,
                          const std::string& from_endpoint_id,
-                         PayloadTransferFrame& payload_transfer_frame);
+                         PayloadTransferFrame& payload_transfer_frame,
+                         Medium medium, PacketMetaData& packet_meta_data);
   void ProcessControlPacket(ClientProxy* to_client,
                             const std::string& from_endpoint_id,
                             PayloadTransferFrame& payload_transfer_frame);