[audio][inspect] Make BufferTracker thread-safe

The BufferTracker class can be accessed from multiple threads,
but it was not thread-safe. This could lead to data races when
buffers are submitted and completed concurrently.

This change introduces a mutex to protect the internal state of
`BufferTracker`, ensuring that all access to its member variables is
properly synchronized. Additionally, `ZX_ASSERT_MSG` calls were
replaced with `FDF_LOG` to prevent crashing the driver in case of an
unexpected state.

Test: audio-inspect-recorder-test
Bug: 462554546
Change-Id: I3dd3f1b88e4487e038fd9c95d8c5000991ebf348
Reviewed-on: https://fuchsia-review.googlesource.com/c/fuchsia/+/1429571
Reviewed-by: Martin Puryear <[email protected]>
Fuchsia-Auto-Submit: Puneetha Ramachandra <[email protected]>
Commit-Queue: Puneetha Ramachandra <[email protected]>

Author: Puneetha-Ramachandra

src/media/audio/drivers/lib/inspect/BUILD.bazel

@@ -17,6 +17,7 @@ cc_library(
     ],
     visibility = ["//visibility:public"],
     deps = [
+        "@fuchsia_sdk//pkg/driver_logging_cpp",
         "@fuchsia_sdk//pkg/inspect",
     ],
 )

src/media/audio/drivers/lib/inspect/BUILD.gn

@@ -22,6 +22,7 @@ test("recorder_test") {
     "task-metrics.h",
   ]
   deps = [
+    "//sdk/lib/driver/logging/cpp",
     "//sdk/lib/inspect/component/cpp",
     "//sdk/lib/inspect/testing/cpp",
     "//src/lib/fxl/test:gtest_main",

src/media/audio/drivers/lib/inspect/buffer-tracker.cc

@@ -3,6 +3,8 @@
 
 #include "src/media/audio/drivers/lib/inspect/buffer-tracker.h"
 
+#include <lib/driver/logging/cpp/logger.h>
+
 namespace audio {
 
 BufferTracker::BufferTracker(inspect::Node node, std::optional<uint32_t> max_buffer_count,
@@ -13,6 +15,7 @@ BufferTracker::BufferTracker(inspect::Node node, std::optional<uint32_t> max_buf
       max_buffer_count_(max_buffer_count) {
   avg_processing_time_us_ = node_.CreateLazyValues(
       "avg_processing_time_us", [this]() -> fpromise::promise<inspect::Inspector> {
+        std::lock_guard<std::mutex> lock(mutex_);
         inspect::Inspector inspector;
         inspector.GetRoot().CreateUint("avg_processing_time_us",
                                        total_buffers_processed_ == 0
@@ -27,6 +30,7 @@ BufferTracker::BufferTracker(inspect::Node node, std::optional<uint32_t> max_buf
   max_empty_buffer_duration_us_ = node_.CreateUint("max_empty_buffer_duration_us", 0);
   avg_outstanding_buffer_count_ = node_.CreateLazyValues(
       "avg_outstanding_buffer_count", [this]() -> fpromise::promise<inspect::Inspector> {
+        std::lock_guard<std::mutex> lock(mutex_);
         inspect::Inspector inspector;
         inspector.GetRoot().CreateUint(
             "avg_outstanding_buffer_count",
@@ -45,6 +49,7 @@ BufferTracker::BufferTracker(inspect::Node node, std::optional<uint32_t> max_buf
   if (per_buffer_duration_.has_value()) {
     total_buffers_processed_duration_us_ = node_.CreateLazyValues(
         "total_buffers_processed_duration_us", [this]() -> fpromise::promise<inspect::Inspector> {
+          std::lock_guard<std::mutex> lock(mutex_);
           inspect::Inspector inspector;
           inspector.GetRoot().CreateUint(
               "total_buffers_processed_duration_us",
@@ -55,9 +60,12 @@ BufferTracker::BufferTracker(inspect::Node node, std::optional<uint32_t> max_buf
 }
 
 void BufferTracker::RecordSubmission() {
-  ZX_ASSERT_MSG(submission_times_.size() < max_buffer_count_.value_or(UINT_MAX),
-                "Submission count (%lu) is more than or equal to max buffer count (%d).",
-                submission_times_.size(), max_buffer_count_.value_or(UINT_MAX));
+  std::lock_guard<std::mutex> lock(mutex_);
+  if (submission_times_.size() >= max_buffer_count_.value_or(UINT_MAX)) {
+    FDF_LOG(ERROR, "Submission count (%lu) is more than or equal to max buffer count (%d).",
+            submission_times_.size(), max_buffer_count_.value_or(UINT_MAX));
+    return;
+  }
   auto submission_time = zx::clock::get_monotonic();
   if (submission_times_.empty()) {
     if (empty_buffer_start_time_.get() != 0) {
@@ -80,11 +88,12 @@ void BufferTracker::RecordSubmission() {
 }
 
 void BufferTracker::RecordCompletion() {
+  std::lock_guard<std::mutex> lock(mutex_);
   auto completion_time = zx::clock::get_monotonic();
-  ZX_ASSERT_MSG(!submission_times_.empty(), "We have completed more buffers than submitted.");
-  ZX_ASSERT_MSG(submission_times_.size() <= max_buffer_count_.value_or(UINT_MAX),
-                "Submission count (%lu) is more than the max buffer count (%d).",
-                submission_times_.size(), max_buffer_count_.value_or(UINT_MAX));
+  if (submission_times_.empty()) {
+    FDF_LOG(ERROR, "No buffers submitted to this tracker yet.");
+    return;
+  }
   if (max_buffer_count_.has_value() && submission_times_.size() == max_buffer_count_.value()) {
     if (full_buffer_start_time_.get() != 0) {
       const zx::duration duration = completion_time - full_buffer_start_time_;

src/media/audio/drivers/lib/inspect/buffer-tracker.h

@@ -8,8 +8,10 @@
 #include <lib/inspect/cpp/vmo/types.h>
 #include <lib/zx/clock.h>
 #include <lib/zx/time.h>
+#include <zircon/compiler.h>
 
 #include <format>
+#include <mutex>
 #include <queue>
 
 namespace audio {
@@ -28,33 +30,34 @@ class BufferTracker {
   // Processing time metrics.
   inspect::LazyNode avg_processing_time_us_;
   inspect::UintProperty max_processing_time_us_;
-  uint64_t total_processing_time_us_ = 0;
-  zx::duration max_processing_time_ = zx::duration::infinite_past();
+  uint64_t total_processing_time_us_ __TA_GUARDED(mutex_) = 0;
+  zx::duration max_processing_time_ __TA_GUARDED(mutex_) = zx::duration::infinite_past();
 
   // Empty buffer metrics.
   inspect::UintProperty total_empty_buffer_duration_us_;
   inspect::UintProperty empty_buffer_episode_count_;
   inspect::UintProperty max_empty_buffer_duration_us_;
-  zx::duration max_empty_buffer_duration_ = zx::duration::infinite_past();
+  zx::duration max_empty_buffer_duration_ __TA_GUARDED(mutex_) = zx::duration::infinite_past();
 
   // Full buffer metrics.
   std::optional<inspect::UintProperty> total_full_buffer_duration_us_;
   std::optional<inspect::UintProperty> full_buffer_episode_count_;
   std::optional<inspect::UintProperty> max_full_buffer_duration_us_;
-  zx::duration max_full_buffer_duration_ = zx::duration::infinite_past();
+  zx::duration max_full_buffer_duration_ __TA_GUARDED(mutex_) = zx::duration::infinite_past();
 
   // Outstanding buffer metrics.
   inspect::LazyNode avg_outstanding_buffer_count_;
   inspect::UintProperty total_buffers_processed_count_;
-  uint64_t cumulative_outstanding_buffer_count_ = 0;
-  uint64_t total_buffers_processed_ = 0;
+  uint64_t cumulative_outstanding_buffer_count_ __TA_GUARDED(mutex_) = 0;
+  uint64_t total_buffers_processed_ __TA_GUARDED(mutex_) = 0;
   std::optional<inspect::LazyNode> total_buffers_processed_duration_us_;
   std::optional<zx::duration> per_buffer_duration_;
 
-  std::queue<zx::time> submission_times_;
-  zx::time empty_buffer_start_time_ = zx::time(0);
-  zx::time full_buffer_start_time_ = zx::time(0);
+  std::queue<zx::time> submission_times_ __TA_GUARDED(mutex_);
+  zx::time empty_buffer_start_time_ __TA_GUARDED(mutex_) = zx::time(0);
+  zx::time full_buffer_start_time_ __TA_GUARDED(mutex_) = zx::time(0);
   std::optional<uint32_t> max_buffer_count_;
+  std::mutex mutex_;
 };
 
 }  // namespace audio

src/media/audio/drivers/lib/inspect/recorder.cc

@@ -193,7 +193,8 @@ void AggregateRecords::RecordTaskMetrics(const Subtask::Metrics& metrics,
 void AggregateRecords::SetupBufferTracker(inspect::Node& node, const std::string& name,
                                           std::optional<uint32_t> max_buffer_count,
                                           std::optional<zx::duration> per_buffer_duration) {
-  buffer_tracker_.emplace(node.CreateChild(name), max_buffer_count, per_buffer_duration);
+  buffer_tracker_ = std::make_unique<BufferTracker>(node.CreateChild(name), max_buffer_count,
+                                                    per_buffer_duration);
 }
 
 void AggregateRecords::RecordBufferSubmission() {

src/media/audio/drivers/lib/inspect/recorder.h

@@ -184,7 +184,7 @@ class AggregateRecords {
   int64_t worst_overrun_frames_ = 0;
   size_t total_task_count_ = 0;
   size_t total_thread_metrics_count_ = 0;
-  std::optional<BufferTracker> buffer_tracker_;
+  std::unique_ptr<BufferTracker> buffer_tracker_;
   std::optional<zx::duration> task_schedule_interval_;
 };