[SDK] Fix lost wakeup in BatchLogProcessor shutdown/force-flush notify

CHANGELOG.md

@@ -274,6 +274,10 @@ Increment the:
   shutdown and force flush.
   [#4382](https://github.com/open-telemetry/opentelemetry-cpp/pull/4382)
 
+* [SDK] Fix lost-wakeups in BatchLogRecordProcessor to prevent stalls during
+  shutdown and force flush.
+  [#4400](https://github.com/open-telemetry/opentelemetry-cpp/issues/4400)
+
 * [METRICS SDK] Support `OTEL_METRICS_EXEMPLAR_FILTER` when metrics exemplars
   are enabled. The default exemplar filter changes from `always_off` to
   `trace_based` to match the specification.

sdk/src/logs/batch_log_record_processor.cc

@@ -108,7 +108,8 @@ void BatchLogRecordProcessor::OnEmit(std::unique_ptr<Recordable> &&record) noexc
   size_t buffer_size = buffer_.size();
   if (buffer_size >= max_queue_size_ / 2 || buffer_size >= max_export_batch_size_)
   {
-    // signal the worker thread
+    // Notified without lock to reduce contention for log emit. If this notify is lost,
+    // the worker thread may wait until next schedule or until the next notify attempt.
     synchronization_data_->is_force_wakeup_background_worker.store(true, std::memory_order_release);
     synchronization_data_->cv.notify_all();
   }
@@ -138,6 +139,7 @@ bool BatchLogRecordProcessor::ForceFlush(std::chrono::microseconds timeout) noex
     if (synchronization_data_->force_flush_pending_sequence.load(std::memory_order_acquire) >
         synchronization_data_->force_flush_notified_sequence.load(std::memory_order_acquire))
     {
+      std::lock_guard<std::mutex> cv_lock(synchronization_data_->cv_m);
       synchronization_data_->is_force_wakeup_background_worker.store(true,
                                                                      std::memory_order_release);
       synchronization_data_->cv.notify_all();
@@ -199,18 +201,24 @@ void BatchLogRecordProcessor::DoBackgroundWork()
     }
 #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
 
-    // Wait for `timeout` milliseconds
-    std::unique_lock<std::mutex> lk(synchronization_data_->cv_m);
-    synchronization_data_->cv.wait_for(lk, timeout, [this] {
-      if (synchronization_data_->is_force_wakeup_background_worker.load(std::memory_order_acquire))
-      {
-        return true;
-      }
-
-      return !buffer_.empty();
-    });
-    synchronization_data_->is_force_wakeup_background_worker.store(false,
-                                                                   std::memory_order_release);
+    // This scope is important! `cv_m` must be released before acquiring `force_flush_cv_m`.
+    // Since `Export()` calls `NotifyCompletion()` which takes `force_flush_cv_m`,
+    // holding `cv_m` while calling `Export()` can lead to a ABBA deadlock.
+    {
+      // Wait for `timeout` milliseconds.
+      std::unique_lock<std::mutex> lk(synchronization_data_->cv_m);
+      synchronization_data_->cv.wait_for(lk, timeout, [this] {
+        if (synchronization_data_->is_force_wakeup_background_worker.load(
+                std::memory_order_acquire))
+        {
+          return true;
+        }
+
+        return !buffer_.empty();
+      });
+      synchronization_data_->is_force_wakeup_background_worker.store(false,
+                                                                     std::memory_order_release);
+    }
 
 #ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW
     if (worker_thread_instrumentation_ != nullptr)
@@ -320,6 +328,7 @@ void BatchLogRecordProcessor::NotifyCompletion(
       exporter->ForceFlush(timeout);
     }
 
+    std::lock_guard<std::mutex> lock(synchronization_data->force_flush_cv_m);
     std::uint64_t notified_sequence =
         synchronization_data->force_flush_notified_sequence.load(std::memory_order_acquire);
     while (notify_force_flush > notified_sequence)
@@ -388,8 +397,12 @@ bool BatchLogRecordProcessor::InternalShutdown(std::chrono::microseconds timeout
 
   if (worker_thread_.joinable())
   {
-    synchronization_data_->is_force_wakeup_background_worker.store(true, std::memory_order_release);
-    synchronization_data_->cv.notify_all();
+    {
+      std::lock_guard<std::mutex> cv_lock(synchronization_data_->cv_m);
+      synchronization_data_->is_force_wakeup_background_worker.store(true,
+                                                                     std::memory_order_release);
+      synchronization_data_->cv.notify_all();
+    }
     worker_thread_.join();
   }
 

sdk/test/logs/BUILD

@@ -127,6 +127,19 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "batch_log_record_processor_test_stress",
+    srcs = glob(["*_test_stress.cc"]),
+    tags = [
+        "logs",
+        "test",
+    ],
+    deps = [
+        "//sdk/src/logs",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "logger_config_test",
     srcs = [

sdk/test/logs/CMakeLists.txt

@@ -11,6 +11,7 @@ foreach(
   simple_log_record_processor_test
   multi_log_record_processor_test
   batch_log_record_processor_test
+  batch_log_record_processor_test_stress
   logger_config_test)
   add_executable(${testname} "${testname}.cc")
   target_link_libraries(

sdk/test/logs/batch_log_record_processor_test_stress.cc

@@ -0,0 +1,142 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include <gtest/gtest.h>
+#include <atomic>
+#include <chrono>
+#include <cstdlib>
+#include <future>
+#include <iostream>
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "opentelemetry/nostd/span.h"
+#include "opentelemetry/sdk/common/exporter_utils.h"
+#include "opentelemetry/sdk/logs/batch_log_record_processor.h"
+#include "opentelemetry/sdk/logs/batch_log_record_processor_options.h"
+#include "opentelemetry/sdk/logs/exporter.h"
+#include "opentelemetry/sdk/logs/read_write_log_record.h"
+#include "opentelemetry/sdk/logs/recordable.h"
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+
+namespace
+{
+
+class CountingLogRecordExporter final : public sdk::logs::LogRecordExporter
+{
+public:
+  explicit CountingLogRecordExporter(
+      std::shared_ptr<std::atomic<std::size_t>> exported_count) noexcept
+      : exported_count_(std::move(exported_count))
+  {}
+
+  std::unique_ptr<sdk::logs::Recordable> MakeRecordable() noexcept override
+  {
+    return std::unique_ptr<sdk::logs::Recordable>(new sdk::logs::ReadWriteLogRecord);
+  }
+
+  sdk::common::ExportResult Export(
+      const nostd::span<std::unique_ptr<sdk::logs::Recordable>> &records) noexcept override
+  {
+    exported_count_->fetch_add(records.size(), std::memory_order_relaxed);
+    return sdk::common::ExportResult::kSuccess;
+  }
+
+  bool ForceFlush(std::chrono::microseconds /*timeout*/) noexcept override { return true; }
+
+  bool Shutdown(std::chrono::microseconds /*timeout*/) noexcept override { return true; }
+
+private:
+  std::shared_ptr<std::atomic<std::size_t>> exported_count_;
+};
+
+// A lost wakeup results in the worker being parked for the entire schedule delay,
+// so the watchdog only has to separate "instant" from "parked for the entire delay"
+// while being generous enough to avoid false positives on slow CI runners.
+constexpr std::chrono::minutes kParkScheduleDelay{10};
+constexpr std::chrono::minutes kWakeupWatchdog{1};
+
+// Runs `operation` on another thread and aborts the binary if it does not return in time.
+template <typename Operation>
+bool CallWithWatchdog(const char *operation_name,
+                      const char *stall_hint,
+                      int round,
+                      const Operation &operation)
+{
+  auto result = std::async(std::launch::async, operation);
+  if (result.wait_for(kWakeupWatchdog) == std::future_status::timeout)
+  {
+    std::cerr << operation_name << " did not return within " << kWakeupWatchdog.count()
+              << "m at round " << round << ". " << stall_hint << '\n';
+    std::abort();
+  }
+  return result.get();
+}
+
+template <typename Operation>
+void RunWorkerParkRace(const char *operation_name, const char *stall_hint, Operation operation)
+{
+  constexpr int kRounds    = 2000;
+  constexpr int kSpinSweep = 50;
+
+  for (int round = 0; round < kRounds; ++round)
+  {
+    auto exported_count = std::make_shared<std::atomic<std::size_t>>(0);
+
+    sdk::logs::BatchLogRecordProcessorOptions options;
+    options.schedule_delay_millis = kParkScheduleDelay;
+    options.max_queue_size        = 4096;
+    options.max_export_batch_size = 512;
+
+    auto processor = std::make_shared<sdk::logs::BatchLogRecordProcessor>(
+        std::make_unique<CountingLogRecordExporter>(exported_count), options);
+
+    // Vary the offset across a sweep so that over the whole set we have a better chance of hitting
+    // the race window.
+    int spin_iterations    = round * kSpinSweep;
+    volatile int spin_sink = 0;
+    for (int s = 0; s < spin_iterations; ++s)
+    {
+      // busy-spin a scheduling-independent increasing amount to sweep the race offset
+      int next  = spin_sink;
+      spin_sink = next + 1;
+    }
+    processor->OnEmit(processor->MakeRecordable());
+
+    EXPECT_TRUE(CallWithWatchdog(operation_name, stall_hint, round,
+                                 [operation, processor] { return operation(*processor); }));
+    EXPECT_EQ(exported_count->load(std::memory_order_relaxed), 1u);
+
+    // Shutdown() already joined the worker; ForceFlush() left it running. Join it either way
+    // before the next round.
+    EXPECT_TRUE(CallWithWatchdog("teardown Shutdown()",
+                                 "possible lost shutdown wakeup stall during worker join()", round,
+                                 [processor] { return processor->Shutdown(); }));
+  }
+}
+
+// Catch a lost cv wakeup during Shutdown(). A lost wakeup parks the worker for the whole schedule
+// delay, so the untimed join() inside Shutdown() blocks for that long.
+TEST(BatchLogRecordProcessorStress, ShutdownRacesWorkerPark)
+{
+  RunWorkerParkRace(
+      "ShutdownRacesWorkerPark: Shutdown()",
+      "possible lost shutdown wakeup stall during worker join()",
+      [](sdk::logs::BatchLogRecordProcessor &processor) { return processor.Shutdown(); });
+}
+
+// Catch a lost cv wakeup during ForceFlush(). A lost wakeup parks the worker for the whole
+// schedule delay before it services the flush, so ForceFlush() blocks for that long.
+TEST(BatchLogRecordProcessorStress, ForceFlushRacesWorkerPark)
+{
+  RunWorkerParkRace(
+      "ForceFlushRacesWorkerPark: ForceFlush()", "possible lost force-flush wakeup",
+      [](sdk::logs::BatchLogRecordProcessor &processor) { return processor.ForceFlush(); });
+}
+
+}  // namespace
+
+OPENTELEMETRY_END_NAMESPACE