[SDK] Avoid missing conditional variable update and simplify atomic bool

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