[SDK] Avoid missing conditional variable update and simplify atomic b…

…ool (#2553)


Addresses two issues -
1. Fix the use of a conditional variable where a wait on the variable might not be in flight when a notify is called. This is fixed by ensuring that an associated lock is aquired before calling the notify.
2. Instead of relying on a lock an a boolean, replace the use wit a single atomic boolean.

---------

Co-authored-by: Ehsan Saei <71217171+esigo@users.noreply.github.com>
Co-authored-by: Marc Alff <marc.alff@oracle.com>
Co-authored-by: Marc Alff <marc.alff@free.fr>

sdk/include/opentelemetry/sdk/metrics/metric_reader.h

@@ -3,10 +3,10 @@
 
 #pragma once
 
+#include <atomic>
 #include <chrono>
 #include <memory>
 
-#include "opentelemetry/common/spin_lock_mutex.h"
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/sdk/metrics/data/metric_data.h"
 #include "opentelemetry/sdk/metrics/instruments.h"
@@ -72,8 +72,7 @@ class MetricReader
 protected:
 private:
   MetricProducer *metric_producer_;
-  mutable opentelemetry::common::SpinLockMutex lock_;
-  bool shutdown_;
+  std::atomic<bool> shutdown_{false};
 };
 }  // namespace metrics
 }  // namespace sdk

sdk/src/metrics/export/periodic_exporting_metric_reader.cc

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #include "opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h"
+#include "opentelemetry/common/spin_lock_mutex.h"
 #include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/sdk/metrics/push_metric_exporter.h"
 
@@ -101,6 +102,7 @@ bool PeriodicExportingMetricReader::CollectAndExportOnce()
   bool notify_force_flush = is_force_flush_pending_.exchange(false, std::memory_order_acq_rel);
   if (notify_force_flush)
   {
+    std::unique_lock<std::mutex> lk(force_flush_m_);
     is_force_flush_notified_.store(true, std::memory_order_release);
     force_flush_cv_.notify_one();
   }
@@ -191,7 +193,11 @@ bool PeriodicExportingMetricReader::OnShutDown(std::chrono::microseconds timeout
 {
   if (worker_thread_.joinable())
   {
-    cv_.notify_one();
+    {
+      // ensure that `cv_` is awaiting, and the update doesn't get lost
+      std::unique_lock<std::mutex> lk(cv_m_);
+      cv_.notify_all();
+    }
     worker_thread_.join();
   }
   return exporter_->Shutdown(timeout);

sdk/src/metrics/metric_reader.cc

@@ -48,10 +48,7 @@ bool MetricReader::Shutdown(std::chrono::microseconds timeout) noexcept
     OTEL_INTERNAL_LOG_WARN("MetricReader::Shutdown - Cannot invoke shutdown twice!");
   }
 
-  {
-    const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);
-    shutdown_ = true;
-  }
+  shutdown_.store(true, std::memory_order_release);
 
   if (!OnShutDown(timeout))
   {
@@ -65,7 +62,7 @@ bool MetricReader::Shutdown(std::chrono::microseconds timeout) noexcept
 bool MetricReader::ForceFlush(std::chrono::microseconds timeout) noexcept
 {
   bool status = true;
-  if (shutdown_)
+  if (IsShutdown())
   {
     OTEL_INTERNAL_LOG_WARN("MetricReader::Shutdown Cannot invoke Force flush on shutdown reader!");
   }
@@ -79,8 +76,7 @@ bool MetricReader::ForceFlush(std::chrono::microseconds timeout) noexcept
 
 bool MetricReader::IsShutdown() const noexcept
 {
-  const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);
-  return shutdown_;
+  return shutdown_.load(std::memory_order_acquire);
 }
 
 }  // namespace metrics