diff --git a/CHANGELOG.md b/CHANGELOG.md index b7c48e22d2..feb3c5f9ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,9 @@ Increment the: ## [Unreleased] +* [BUG] Prevent lost condition-variable wakeups in OTLP file exporter and periodic + metric exporter + [#4365](https://github.com/open-telemetry/opentelemetry-cpp/pull/4365) * [CONFIGURATION] Add a configuration builder for the host resource detector [#4451](https://github.com/open-telemetry/opentelemetry-cpp/issues/4451) * [CONFIGURATION] Build the configured resource detectors in SdkBuilder, apply diff --git a/exporters/otlp/src/otlp_file_client.cc b/exporters/otlp/src/otlp_file_client.cc index 5bcbedfbe5..d881352ae8 100644 --- a/exporters/otlp/src/otlp_file_client.cc +++ b/exporters/otlp/src/otlp_file_client.cc @@ -994,6 +994,10 @@ class OPENTELEMETRY_LOCAL_SYMBOL OtlpFileSystemBackend : public OtlpFileAppender { if (file_) { + { + std::lock_guard waker_guard{file_->background_thread_waker_lock}; + file_->is_shutdown.store(true, std::memory_order_release); + } file_->background_thread_waker_cv.notify_all(); std::unique_ptr background_flush_thread; { @@ -1133,7 +1137,10 @@ class OPENTELEMETRY_LOCAL_SYMBOL OtlpFileSystemBackend : public OtlpFileAppender bool Shutdown(std::chrono::microseconds timeout) noexcept override { - file_->is_shutdown.store(true, std::memory_order_release); + { + std::lock_guard waker_guard{file_->background_thread_waker_lock}; + file_->is_shutdown.store(true, std::memory_order_release); + } bool result = ForceFlush(timeout); return result; @@ -1482,11 +1489,6 @@ class OPENTELEMETRY_LOCAL_SYMBOL OtlpFileSystemBackend : public OtlpFileAppender break; } - if (concurrency_file->is_shutdown.load(std::memory_order_acquire)) - { - break; - } - #ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW if (thread_instrumentation != nullptr) { @@ -1494,9 +1496,19 @@ class OPENTELEMETRY_LOCAL_SYMBOL OtlpFileSystemBackend : public OtlpFileAppender } #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */ + bool is_shutdown = false; { std::unique_lock lk(concurrency_file->background_thread_waker_lock); - concurrency_file->background_thread_waker_cv.wait_for(lk, flush_interval); + // Even though is_shutdown is atomic, the lock guarantees that either a change to + // is_shutdown will be observed, or background_thread_waker_cv will see the notification + // at shutdown. It is important to set is_shutdown prior to `wait_for` rather than + // as part of a condition in `wait_for` so that a shutdown while the thread is in + // `wait_for` will still call `std::fflush` below. + is_shutdown = concurrency_file->is_shutdown.load(std::memory_order_acquire); + if (!is_shutdown) + { + concurrency_file->background_thread_waker_cv.wait_for(lk, flush_interval); + } } #ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW @@ -1506,6 +1518,11 @@ class OPENTELEMETRY_LOCAL_SYMBOL OtlpFileSystemBackend : public OtlpFileAppender } #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */ + if (is_shutdown) + { + break; + } + { std::size_t current_record_count = concurrency_file->record_count.load(std::memory_order_acquire); diff --git a/sdk/src/metrics/export/periodic_exporting_metric_reader.cc b/sdk/src/metrics/export/periodic_exporting_metric_reader.cc index 86206ad84e..32b7d3d0d0 100644 --- a/sdk/src/metrics/export/periodic_exporting_metric_reader.cc +++ b/sdk/src/metrics/export/periodic_exporting_metric_reader.cc @@ -221,7 +221,12 @@ bool PeriodicExportingMetricReader::OnForceFlush(std::chrono::microseconds timeo if (force_flush_pending_sequence_.load(std::memory_order_acquire) > force_flush_notified_sequence_.load(std::memory_order_acquire)) { - is_force_wakeup_background_worker_.store(true, std::memory_order_release); + { + // Acquiring cv_m_ guarantees that the worker thread either is not currently waiting on cv_, + // or the notify below will cause it to re-check the wait condition. + std::lock_guard cv_guard{cv_m_}; + is_force_wakeup_background_worker_.store(true, std::memory_order_release); + } cv_.notify_all(); } return force_flush_notified_sequence_.load(std::memory_order_acquire) >= current_sequence; @@ -283,6 +288,11 @@ bool PeriodicExportingMetricReader::OnShutDown(std::chrono::microseconds timeout { if (worker_thread_.joinable()) { + { + // Acquiring cv_m_ guarantees that the next time the worker thread checks the wait condition + // on cv_ (either from notify below or any other reason) it will see IsShutdown() return true. + std::lock_guard cv_guard{cv_m_}; + } cv_.notify_all(); worker_thread_.join(); }