diff --git a/opentelemetry-sdk/src/metrics/meter_provider.rs b/opentelemetry-sdk/src/metrics/meter_provider.rs index 4c363dd181..011de1f41c 100644 --- a/opentelemetry-sdk/src/metrics/meter_provider.rs +++ b/opentelemetry-sdk/src/metrics/meter_provider.rs @@ -38,7 +38,7 @@ pub struct SdkMeterProvider { struct SdkMeterProviderInner { pipes: Arc, meters: Mutex>>, - is_shutdown: AtomicBool, + shutdown_invoked: AtomicBool, } impl Default for SdkMeterProvider { @@ -119,20 +119,29 @@ impl SdkMeterProvider { impl SdkMeterProviderInner { fn force_flush(&self) -> MetricResult<()> { - self.pipes.force_flush() + if self + .shutdown_invoked + .load(std::sync::atomic::Ordering::Relaxed) + { + Err(MetricError::Other( + "Cannot perform flush as MeterProvider shutdown already invoked.".into(), + )) + } else { + self.pipes.force_flush() + } } fn shutdown(&self) -> MetricResult<()> { if self - .is_shutdown - .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) - .is_ok() + .shutdown_invoked + .swap(true, std::sync::atomic::Ordering::SeqCst) { - self.pipes.shutdown() - } else { + // If the previous value was true, shutdown was already invoked. Err(MetricError::Other( - "metrics provider already shut down".into(), + "MeterProvider shutdown already invoked.".into(), )) + } else { + self.pipes.shutdown() } } } @@ -141,7 +150,7 @@ impl Drop for SdkMeterProviderInner { fn drop(&mut self) { // If user has already shutdown the provider manually by calling // shutdown(), then we don't need to call shutdown again. - if self.is_shutdown.load(Ordering::Relaxed) { + if self.shutdown_invoked.load(Ordering::Relaxed) { otel_debug!( name: "MeterProvider.Drop.AlreadyShutdown", message = "MeterProvider was already shut down; drop will not attempt shutdown again." @@ -173,7 +182,7 @@ impl MeterProvider for SdkMeterProvider { } fn meter_with_scope(&self, scope: InstrumentationScope) -> Meter { - if self.inner.is_shutdown.load(Ordering::Relaxed) { + if self.inner.shutdown_invoked.load(Ordering::Relaxed) { otel_debug!( name: "MeterProvider.NoOpMeterReturned", meter_name = scope.name(), @@ -270,7 +279,7 @@ impl MeterProviderBuilder { self.views, )), meters: Default::default(), - is_shutdown: AtomicBool::new(false), + shutdown_invoked: AtomicBool::new(false), }), }; diff --git a/opentelemetry-sdk/src/metrics/periodic_reader.rs b/opentelemetry-sdk/src/metrics/periodic_reader.rs index ff96e037b9..292ab0b8f2 100644 --- a/opentelemetry-sdk/src/metrics/periodic_reader.rs +++ b/opentelemetry-sdk/src/metrics/periodic_reader.rs @@ -1,7 +1,6 @@ use std::{ env, fmt, sync::{ - atomic::AtomicBool, mpsc::{self, Receiver, Sender}, Arc, Mutex, Weak, }, @@ -158,7 +157,6 @@ impl PeriodicReader { let reader = PeriodicReader { inner: Arc::new(PeriodicReaderInner { message_sender: Arc::new(message_sender), - shutdown_invoked: AtomicBool::new(false), producer: Mutex::new(None), exporter: Arc::new(exporter), }), @@ -300,7 +298,6 @@ struct PeriodicReaderInner { exporter: Arc, message_sender: Arc>, producer: Mutex>>, - shutdown_invoked: AtomicBool, } impl PeriodicReaderInner { @@ -374,15 +371,6 @@ impl PeriodicReaderInner { } fn force_flush(&self) -> MetricResult<()> { - if self - .shutdown_invoked - .load(std::sync::atomic::Ordering::Relaxed) - { - return Err(MetricError::Other( - "Cannot perform flush as PeriodicReader shutdown already invoked.".into(), - )); - } - // TODO: Better message for this scenario. // Flush and Shutdown called from 2 threads Flush check shutdown // flag before shutdown thread sets it. Both threads attempt to send @@ -414,15 +402,6 @@ impl PeriodicReaderInner { } fn shutdown(&self) -> MetricResult<()> { - if self - .shutdown_invoked - .swap(true, std::sync::atomic::Ordering::Relaxed) - { - return Err(MetricError::Other( - "PeriodicReader shutdown already invoked.".into(), - )); - } - // TODO: See if this is better to be created upfront. let (response_tx, response_rx) = mpsc::channel(); self.message_sender