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

CHANGELOG.md

@@ -15,6 +15,10 @@ Increment the:
 
 ## [Unreleased]
 
+* [SDK] Fix lost-wakeups in BatchSpanProcessor to prevent stalls during
+  shutdown and force flush.
+  [#4382](https://github.com/open-telemetry/opentelemetry-cpp/pull/4382)
+
 * [CONFIGURATION] Add support for the composite sampler configuration
   (programmatic and from yaml)
   ([#4366](https://github.com/open-telemetry/opentelemetry-cpp/pull/4366))

sdk/src/trace/batch_span_processor.cc

@@ -98,7 +98,7 @@ void BatchSpanProcessor::OnEnd(std::unique_ptr<Recordable> &&span) noexcept
   size_t buffer_size = buffer_.size();
   if (buffer_size >= max_queue_size_ / 2 || buffer_size >= max_export_batch_size_)
   {
-    // signal the worker thread
+    // Best effort wakeup for worker thread.
     synchronization_data_->cv.notify_all();
   }
 }
@@ -127,6 +127,7 @@ bool BatchSpanProcessor::ForceFlush(std::chrono::microseconds timeout) noexcept
     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();
@@ -188,18 +189,21 @@ void BatchSpanProcessor::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);
+    // 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)
@@ -309,6 +313,7 @@ void BatchSpanProcessor::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)
@@ -376,8 +381,12 @@ bool BatchSpanProcessor::InternalShutdown(std::chrono::microseconds timeout) noe
 
   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/trace/BUILD

@@ -83,6 +83,19 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "batch_span_processor_test_stress",
+    srcs = glob(["*_test_stress.cc"]),
+    tags = [
+        "test",
+        "trace",
+    ],
+    deps = [
+        "//sdk/src/trace",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "tracer_test",
     srcs = [

sdk/test/trace/CMakeLists.txt

@@ -15,6 +15,7 @@ foreach(
   trace_id_ratio_sampler_test
   composable_sampler_test
   batch_span_processor_test
+  batch_span_processor_test_stress
   tracer_config_test)
   add_executable(${testname} "${testname}.cc")
   target_link_libraries(

sdk/test/trace/batch_span_processor_test_stress.cc

@@ -0,0 +1,140 @@
+// 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/trace/batch_span_processor.h"
+#include "opentelemetry/sdk/trace/batch_span_processor_options.h"
+#include "opentelemetry/sdk/trace/exporter.h"
+#include "opentelemetry/sdk/trace/recordable.h"
+#include "opentelemetry/sdk/trace/span_data.h"
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+
+namespace
+{
+
+class CountingSpanExporter final : public sdk::trace::SpanExporter
+{
+public:
+  explicit CountingSpanExporter(std::shared_ptr<std::atomic<std::size_t>> exported_count) noexcept
+      : exported_count_(std::move(exported_count))
+  {}
+
+  std::unique_ptr<sdk::trace::Recordable> MakeRecordable() noexcept override
+  {
+    return std::unique_ptr<sdk::trace::Recordable>(new sdk::trace::SpanData);
+  }
+
+  sdk::common::ExportResult Export(
+      const nostd::span<std::unique_ptr<sdk::trace::Recordable>> &recordables) noexcept override
+  {
+    exported_count_->fetch_add(recordables.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::trace::BatchSpanProcessorOptions options;
+    options.schedule_delay_millis = kParkScheduleDelay;
+    options.max_queue_size        = 4096;
+    options.max_export_batch_size = 512;
+
+    auto processor = std::make_shared<sdk::trace::BatchSpanProcessor>(
+        std::make_unique<CountingSpanExporter>(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->OnEnd(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(BatchSpanProcessorStress, ShutdownRacesWorkerPark)
+{
+  RunWorkerParkRace("ShutdownRacesWorkerPark: Shutdown()",
+                    "possible lost shutdown wakeup stall during worker join()",
+                    [](sdk::trace::BatchSpanProcessor &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(BatchSpanProcessorStress, ForceFlushRacesWorkerPark)
+{
+  RunWorkerParkRace(
+      "ForceFlushRacesWorkerPark: ForceFlush()", "possible lost force-flush wakeup",
+      [](sdk::trace::BatchSpanProcessor &processor) { return processor.ForceFlush(); });
+}
+
+}  // namespace
+
+OPENTELEMETRY_END_NAMESPACE