[BUG] Prevent lost condition-variable wakeups during OTLP file, periodic metric, and batch span processor shutdown

CHANGELOG.md

@@ -290,6 +290,10 @@ Increment the:
   for process entity
   [#4437](https://github.com/open-telemetry/opentelemetry-cpp/pull/4437)
 
+* [BUG] Prevent lost condition-variable wakeups in OTLP file exporter and periodic
+  metric exporter
+  [#4365](https://github.com/open-telemetry/opentelemetry-cpp/pull/4365)
+
 Important changes:
 
 * [API] Never set a null global provider or propagator

exporters/otlp/src/otlp_file_client.cc

@@ -994,6 +994,10 @@ class OPENTELEMETRY_LOCAL_SYMBOL OtlpFileSystemBackend : public OtlpFileAppender
   {
     if (file_)
     {
+      {
+        std::lock_guard<std::mutex> 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<std::thread> 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<std::mutex> waker_guard{file_->background_thread_waker_lock};
+      file_->is_shutdown.store(true, std::memory_order_release);
+    }
 
     bool result = ForceFlush(timeout);
     return result;
@@ -1482,21 +1489,26 @@ 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)
           {
             thread_instrumentation->BeforeWait();
           }
 #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
 
+          bool is_shutdown = false;
           {
             std::unique_lock<std::mutex> 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);

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<std::mutex> 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<std::mutex> cv_guard{cv_m_};
+    }
     cv_.notify_all();
     worker_thread_.join();
   }