[SDK] Fix forceflush may wait for ever

sdk/include/opentelemetry/sdk/logs/batch_log_record_processor.h

@@ -115,20 +115,25 @@ class BatchLogRecordProcessor : public LogRecordProcessor
 
  /* Important boolean flags to handle the workflow of the processor */
  std::atomic<bool> is_force_wakeup_background_worker{false};
- std::atomic<bool> is_force_flush_pending{false};
- std::atomic<bool> is_force_flush_notified{false};
- std::atomic<std::chrono::microseconds::rep> force_flush_timeout_us{0};
  std::atomic<bool> is_shutdown{false};
+ std::atomic<uint64_t> force_flush_pending_sequence{0};
+ std::atomic<uint64_t> force_flush_notified_sequence{0};
+ std::atomic<std::chrono::microseconds::rep> force_flush_timeout_us{0};
+
+ // Do not use SynchronizationData() = default; here, some versions of GCC&Clang have BUGs
+ // and may not initialize the member correctly. See also
+ // https://stackoverflow.com/questions/53408962/try-to-understand-compiler-error-message-default-member-initializer-required-be
+ inline SynchronizationData() {}
  };
 
  /**
  * @brief Notify completion of shutdown and force flush. This may be called from the any thread at
  * any time
  *
- * @param notify_force_flush Flag to indicate whether to notify force flush completion.
+ * @param notify_force_flush Sequence to indicate whether to notify force flush completion.
  * @param synchronization_data Synchronization data to be notified.
  */
- static void NotifyCompletion(bool notify_force_flush,
+ static void NotifyCompletion(uint64_t notify_force_flush,
  const std::unique_ptr<LogRecordExporter> &exporter,
  const std::shared_ptr<SynchronizationData> &synchronization_data);
 

sdk/include/opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h

@@ -6,6 +6,7 @@
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
+#include <cstdint>
 #include <memory>
 #include <mutex>
 #include <thread>
@@ -50,9 +51,9 @@ class PeriodicExportingMetricReader : public MetricReader
  std::thread worker_thread_;
 
  /* Synchronization primitives */
- std::atomic<bool> is_force_flush_pending_{false};
  std::atomic<bool> is_force_wakeup_background_worker_{false};
- std::atomic<bool> is_force_flush_notified_{false};
+ std::atomic<uint64_t> force_flush_pending_sequence_{0};
+ std::atomic<uint64_t> force_flush_notified_sequence_{0};
  std::condition_variable cv_, force_flush_cv_;
  std::mutex cv_m_, force_flush_m_;
 };

sdk/include/opentelemetry/sdk/trace/batch_span_processor.h

@@ -5,6 +5,7 @@
 
 #include <atomic>
 #include <condition_variable>
+#include <cstdint>
 #include <memory>
 #include <mutex>
 #include <thread>
@@ -115,20 +116,25 @@ class BatchSpanProcessor : public SpanProcessor
 
  /* Important boolean flags to handle the workflow of the processor */
  std::atomic<bool> is_force_wakeup_background_worker{false};
- std::atomic<bool> is_force_flush_pending{false};
- std::atomic<bool> is_force_flush_notified{false};
- std::atomic<std::chrono::microseconds::rep> force_flush_timeout_us{0};
  std::atomic<bool> is_shutdown{false};
+ std::atomic<uint64_t> force_flush_pending_sequence{0};
+ std::atomic<uint64_t> force_flush_notified_sequence{0};
+ std::atomic<std::chrono::microseconds::rep> force_flush_timeout_us{0};
+
+ // Do not use SynchronizationData() = default; here, some versions of GCC&Clang have BUGs
+ // and may not initialize the member correctly. See also
+ // https://stackoverflow.com/questions/53408962/try-to-understand-compiler-error-message-default-member-initializer-required-be
+ inline SynchronizationData() {}
  };
 
  /**
  * @brief Notify completion of shutdown and force flush. This may be called from the any thread at
  * any time
  *
- * @param notify_force_flush Flag to indicate whether to notify force flush completion.
+ * @param notify_force_flush Sequence to indicate whether to notify force flush completion.
  * @param synchronization_data Synchronization data to be notified.
  */
- static void NotifyCompletion(bool notify_force_flush,
+ static void NotifyCompletion(uint64_t notify_force_flush,
  const std::unique_ptr<SpanExporter> &exporter,
  const std::shared_ptr<SynchronizationData> &synchronization_data);
 

sdk/src/logs/batch_log_record_processor.cc

@@ -65,7 +65,7 @@ void BatchLogRecordProcessor::OnEmit(std::unique_ptr<Recordable> &&record) noexc
  {
  // signal the worker thread
  synchronization_data_->is_force_wakeup_background_worker.store(true, std::memory_order_release);
- synchronization_data_->cv.notify_one();
+ synchronization_data_->cv.notify_all();
  }
 }
 
@@ -79,21 +79,25 @@ bool BatchLogRecordProcessor::ForceFlush(std::chrono::microseconds timeout) noex
  // Now wait for the worker thread to signal back from the Export method
  std::unique_lock<std::mutex> lk_cv(synchronization_data_->force_flush_cv_m);
 
- synchronization_data_->is_force_flush_pending.store(true, std::memory_order_release);
+ std::uint64_t current_sequence =
+ synchronization_data_->force_flush_pending_sequence.fetch_add(1, std::memory_order_release) +
+ 1;
  synchronization_data_->force_flush_timeout_us.store(timeout.count(), std::memory_order_release);
- auto break_condition = [this]() {
+ auto break_condition = [this, current_sequence]() {
  if (synchronization_data_->is_shutdown.load() == true)
  {
  return true;
  }
 
  // Wake up the worker thread once.
- if (synchronization_data_->is_force_flush_pending.load(std::memory_order_acquire))
+ if (synchronization_data_->force_flush_pending_sequence.load(std::memory_order_acquire) >
+ synchronization_data_->force_flush_notified_sequence.load(std::memory_order_acquire))
  {
- synchronization_data_->cv.notify_one();
+ synchronization_data_->cv.notify_all();
  }
 
- return synchronization_data_->is_force_flush_notified.load(std::memory_order_acquire);
+ return synchronization_data_->force_flush_notified_sequence.load(std::memory_order_acquire) >=
+ current_sequence;
  };
 
  // Fix timeout to meet requirement of wait_for
@@ -110,35 +114,22 @@ bool BatchLogRecordProcessor::ForceFlush(std::chrono::microseconds timeout) noex
  bool result = false;
  while (!result && timeout_steady > std::chrono::steady_clock::duration::zero())
  {
- // When is_force_flush_notified.store(true) and force_flush_cv.notify_all() is called
- // between is_force_flush_pending.load() and force_flush_cv.wait(). We must not wait
- // for ever
+ // When force_flush_notified_sequence.compare_exchange_strong(...) and
+ // force_flush_cv.notify_all() is called between force_flush_pending_sequence.load(...) and
+ // force_flush_cv.wait(). We must not wait for ever
  std::chrono::steady_clock::time_point start_timepoint = std::chrono::steady_clock::now();
- result = synchronization_data_->force_flush_cv.wait_for(lk_cv, scheduled_delay_millis_,
- break_condition);
- timeout_steady -= std::chrono::steady_clock::now() - start_timepoint;
- }
+ std::chrono::microseconds wait_timeout = scheduled_delay_millis_;
 
- // If it's already signaled, we must wait util notified.
- // We use a spin lock here
- if (false ==
- synchronization_data_->is_force_flush_pending.exchange(false, std::memory_order_acq_rel))
- {
- for (int retry_waiting_times = 0;
- false == synchronization_data_->is_force_flush_notified.load(std::memory_order_acquire);
- ++retry_waiting_times)
+ if (wait_timeout > timeout_steady)
  {
- opentelemetry::common::SpinLockMutex::fast_yield();
- if ((retry_waiting_times & 127) == 127)
- {
- std::this_thread::yield();
- }
+ wait_timeout = std::chrono::duration_cast<std::chrono::microseconds>(timeout_steady);
  }
+ result = synchronization_data_->force_flush_cv.wait_for(lk_cv, wait_timeout, break_condition);
+ timeout_steady -= std::chrono::steady_clock::now() - start_timepoint;
  }
 
- synchronization_data_->is_force_flush_notified.store(false, std::memory_order_release);
-
- return result;
+ return synchronization_data_->force_flush_notified_sequence.load(std::memory_order_acquire) >=
+ current_sequence;
 }
 
 void BatchLogRecordProcessor::DoBackgroundWork()
@@ -182,8 +173,8 @@ void BatchLogRecordProcessor::Export()
  {
  std::vector<std::unique_ptr<Recordable>> records_arr;
  size_t num_records_to_export;
- bool notify_force_flush =
- synchronization_data_->is_force_flush_pending.exchange(false, std::memory_order_acq_rel);
+ std::uint64_t notify_force_flush =
+ synchronization_data_->force_flush_pending_sequence.load(std::memory_order_acquire);
  if (notify_force_flush)
  {
  num_records_to_export = buffer_.size();
@@ -217,7 +208,7 @@ void BatchLogRecordProcessor::Export()
 }
 
 void BatchLogRecordProcessor::NotifyCompletion(
- bool notify_force_flush,
+ std::uint64_t notify_force_flush,
  const std::unique_ptr<LogRecordExporter> &exporter,
  const std::shared_ptr<SynchronizationData> &synchronization_data)
 {
@@ -226,7 +217,8 @@ void BatchLogRecordProcessor::NotifyCompletion(
  return;
  }
 
- if (notify_force_flush)
+ if (notify_force_flush >
+ synchronization_data->force_flush_notified_sequence.load(std::memory_order_acquire))
  {
  if (exporter)
  {
@@ -236,8 +228,15 @@ void BatchLogRecordProcessor::NotifyCompletion(
  std::chrono::microseconds::zero());
  exporter->ForceFlush(timeout);
  }
- synchronization_data->is_force_flush_notified.store(true, std::memory_order_release);
- synchronization_data->force_flush_cv.notify_one();
+
+ std::uint64_t notified_sequence =
+ synchronization_data->force_flush_notified_sequence.load(std::memory_order_acquire);
+ while (notify_force_flush > notified_sequence)
+ {
+ synchronization_data->force_flush_notified_sequence.compare_exchange_strong(
+ notified_sequence, notify_force_flush, std::memory_order_acq_rel);
+ synchronization_data->force_flush_cv.notify_all();
+ }
  }
 }
 
@@ -246,7 +245,8 @@ void BatchLogRecordProcessor::DrainQueue()
  while (true)
  {
  if (buffer_.empty() &&
- false == synchronization_data_->is_force_flush_pending.load(std::memory_order_acquire))
+ synchronization_data_->force_flush_pending_sequence.load(std::memory_order_acquire) <=
+ synchronization_data_->force_flush_notified_sequence.load(std::memory_order_acquire))
  {
  break;
  }
@@ -285,7 +285,7 @@ bool BatchLogRecordProcessor::Shutdown(std::chrono::microseconds timeout) noexce
  if (worker_thread_.joinable())
  {
  synchronization_data_->is_force_wakeup_background_worker.store(true, std::memory_order_release);
- synchronization_data_->cv.notify_one();
+ synchronization_data_->cv.notify_all();
  worker_thread_.join();
  }
 

sdk/src/metrics/export/periodic_exporting_metric_reader.cc

@@ -90,6 +90,7 @@ bool PeriodicExportingMetricReader::CollectAndExportOnce()
  });
 
  std::future_status status;
+ std::uint64_t notify_force_flush = force_flush_pending_sequence_.load(std::memory_order_acquire);
  do
  {
  status = future_receive.wait_for(std::chrono::milliseconds(export_timeout_millis_));
@@ -99,12 +100,13 @@ bool PeriodicExportingMetricReader::CollectAndExportOnce()
  break;
  }
  } while (status != std::future_status::ready);
- bool notify_force_flush = is_force_flush_pending_.exchange(false, std::memory_order_acq_rel);
- if (notify_force_flush)
+
+ std::uint64_t notified_sequence = force_flush_notified_sequence_.load(std::memory_order_acquire);
+ while (notify_force_flush > notified_sequence)
  {
- std::unique_lock<std::mutex> lk(force_flush_m_);
- is_force_flush_notified_.store(true, std::memory_order_release);
- force_flush_cv_.notify_one();
+ force_flush_notified_sequence_.compare_exchange_strong(notified_sequence, notify_force_flush,
+  std::memory_order_acq_rel);
+ force_flush_cv_.notify_all();
  }
 
  return true;
@@ -113,24 +115,27 @@ bool PeriodicExportingMetricReader::CollectAndExportOnce()
 bool PeriodicExportingMetricReader::OnForceFlush(std::chrono::microseconds timeout) noexcept
 {
  std::unique_lock<std::mutex> lk_cv(force_flush_m_);
- is_force_flush_pending_.store(true, std::memory_order_release);
- auto break_condition = [this]() {
+ std::uint64_t current_sequence =
+ force_flush_pending_sequence_.fetch_add(1, std::memory_order_release) + 1;
+ auto break_condition = [this, current_sequence]() {
  if (IsShutdown())
  {
  return true;
  }
 
- // Wake up the worker thread once.
- if (is_force_flush_pending_.load(std::memory_order_acquire))
+ // Wake up the worker thread.
+ 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);
- cv_.notify_one();
+ cv_.notify_all();
  }
- return is_force_flush_notified_.load(std::memory_order_acquire);
+ return force_flush_notified_sequence_.load(std::memory_order_acquire) >= current_sequence;
  };
 
- auto wait_timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout(
- timeout, std::chrono::microseconds::zero());
+ std::chrono::microseconds wait_timeout =
+ opentelemetry::common::DurationUtil::AdjustWaitForTimeout(timeout,
+ std::chrono::microseconds::zero());
  std::chrono::steady_clock::duration timeout_steady =
  std::chrono::duration_cast<std::chrono::steady_clock::duration>(wait_timeout);
  if (timeout_steady <= std::chrono::steady_clock::duration::zero())
@@ -141,29 +146,19 @@ bool PeriodicExportingMetricReader::OnForceFlush(std::chrono::microseconds timeo
  bool result = false;
  while (!result && timeout_steady > std::chrono::steady_clock::duration::zero())
  {
- // When is_force_flush_notified_.store(true) and force_flush_cv_.notify_all() is called
- // between is_force_flush_pending_.load() and force_flush_cv_.wait(). We must not wait
- // for ever
+ // When force_flush_notified_sequence_.compare_exchange_strong(...) and
+ // force_flush_cv_.notify_all() is called between force_flush_pending_sequence_.load(...) and
+ // force_flush_cv_.wait(). We must not wait for ever
  std::chrono::steady_clock::time_point start_timepoint = std::chrono::steady_clock::now();
- result = force_flush_cv_.wait_for(lk_cv, export_interval_millis_, break_condition);
- timeout_steady -= std::chrono::steady_clock::now() - start_timepoint;
- }
 
- // If it will be already signaled, we must wait until notified.
- // We use a spin lock here
- if (false == is_force_flush_pending_.exchange(false, std::memory_order_acq_rel))
- {
- for (int retry_waiting_times = 0;
- false == is_force_flush_notified_.load(std::memory_order_acquire); ++retry_waiting_times)
+ wait_timeout = export_interval_millis_;
+ if (wait_timeout > timeout_steady)
  {
- opentelemetry::common::SpinLockMutex::fast_yield();
- if ((retry_waiting_times & 127) == 127)
- {
- std::this_thread::yield();
- }
+ wait_timeout = std::chrono::duration_cast<std::chrono::microseconds>(timeout_steady);
  }
+ result = force_flush_cv_.wait_for(lk_cv, wait_timeout, break_condition);
+ timeout_steady -= std::chrono::steady_clock::now() - start_timepoint;
  }
- is_force_flush_notified_.store(false, std::memory_order_release);
 
  if (result)
  {
@@ -186,18 +181,15 @@ bool PeriodicExportingMetricReader::OnForceFlush(std::chrono::microseconds timeo
  result = false;
  }
  }
- return result;
+ return result &&
+ force_flush_notified_sequence_.load(std::memory_order_acquire) >= current_sequence;
 }
 
 bool PeriodicExportingMetricReader::OnShutDown(std::chrono::microseconds timeout) noexcept
 {
  if (worker_thread_.joinable())
  {
- {
- // ensure that `cv_` is awaiting, and the update doesn't get lost
- std::unique_lock<std::mutex> lk(cv_m_);
- cv_.notify_all();
- }
+ cv_.notify_all();
  worker_thread_.join();
  }
  return exporter_->Shutdown(timeout);