From d8786fbb45e075f51b5131f216a1108928aa9bb2 Mon Sep 17 00:00:00 2001 From: Deniz Ariyan Date: Wed, 5 Aug 2026 21:59:40 +0200 Subject: [PATCH 1/5] [TEST] Add BatchSpanProcessor lost wakeup stress test --- sdk/test/trace/BUILD | 13 ++ sdk/test/trace/CMakeLists.txt | 1 + .../trace/batch_span_processor_test_stress.cc | 137 ++++++++++++++++++ 3 files changed, 151 insertions(+) create mode 100644 sdk/test/trace/batch_span_processor_test_stress.cc diff --git a/sdk/test/trace/BUILD b/sdk/test/trace/BUILD index 1a754e68e8..daf0e4db14 100644 --- a/sdk/test/trace/BUILD +++ b/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 = [ diff --git a/sdk/test/trace/CMakeLists.txt b/sdk/test/trace/CMakeLists.txt index 671a6666d8..eef9da364f 100644 --- a/sdk/test/trace/CMakeLists.txt +++ b/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( diff --git a/sdk/test/trace/batch_span_processor_test_stress.cc b/sdk/test/trace/batch_span_processor_test_stress.cc new file mode 100644 index 0000000000..6927895356 --- /dev/null +++ b/sdk/test/trace/batch_span_processor_test_stress.cc @@ -0,0 +1,137 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#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> exported_count) noexcept + : exported_count_(std::move(exported_count)) + {} + + std::unique_ptr MakeRecordable() noexcept override + { + return std::unique_ptr(new sdk::trace::SpanData); + } + + sdk::common::ExportResult Export( + const nostd::span> &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> 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 +bool CallWithWatchdog(const char *operation_name, + const char *stall_hint, + int round, + 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 << std::endl; + std::abort(); + } + return result.get(); +} + +template +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>(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( + std::make_unique(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; + for (volatile int s = 0; s < spin_iterations; ++s) + { + // busy-spin a scheduling-independent increasing amount to sweep the race offset + } + 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 From d2345a4abe45a02f720adf76581232d30e13d1a3 Mon Sep 17 00:00:00 2001 From: Deniz Ariyan Date: Sat, 8 Aug 2026 13:42:48 +0200 Subject: [PATCH 2/5] [SDK] Fix lost wakeups in BatchSpanProcessor --- CHANGELOG.md | 4 +++ sdk/src/trace/batch_span_processor.cc | 39 ++++++++++++++++----------- 2 files changed, 28 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b05b5b6d66..22e311f501 100644 --- a/CHANGELOG.md +++ b/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)) diff --git a/sdk/src/trace/batch_span_processor.cc b/sdk/src/trace/batch_span_processor.cc index c59c9866a2..15ee344261 100644 --- a/sdk/src/trace/batch_span_processor.cc +++ b/sdk/src/trace/batch_span_processor.cc @@ -98,7 +98,7 @@ void BatchSpanProcessor::OnEnd(std::unique_ptr &&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 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 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 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 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 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(); } From 305e961725ea66bd684cf0899f69fee11531fbc6 Mon Sep 17 00:00:00 2001 From: Deniz Ariyan Date: Sun, 9 Aug 2026 10:30:45 +0200 Subject: [PATCH 3/5] [TEST] Fix warnings in stress tests --- sdk/test/trace/batch_span_processor_test_stress.cc | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/sdk/test/trace/batch_span_processor_test_stress.cc b/sdk/test/trace/batch_span_processor_test_stress.cc index 6927895356..4b4d28d54a 100644 --- a/sdk/test/trace/batch_span_processor_test_stress.cc +++ b/sdk/test/trace/batch_span_processor_test_stress.cc @@ -63,13 +63,13 @@ template bool CallWithWatchdog(const char *operation_name, const char *stall_hint, int round, - Operation operation) + 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 << std::endl; + << "m at round " << round << ". " << stall_hint << '\n'; std::abort(); } return result.get(); @@ -95,10 +95,13 @@ void RunWorkerParkRace(const char *operation_name, const char *stall_hint, Opera // 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; - for (volatile int s = 0; s < spin_iterations; ++s) + 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()); From 217e5e19c754c6c237be5a22c49ea2dfb642d775 Mon Sep 17 00:00:00 2001 From: Deniz Ariyan Date: Sun, 9 Aug 2026 17:44:34 +0200 Subject: [PATCH 4/5] [SDK] Clarify intended notify without lock --- sdk/src/trace/batch_span_processor.cc | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sdk/src/trace/batch_span_processor.cc b/sdk/src/trace/batch_span_processor.cc index 15ee344261..ad4f9c8884 100644 --- a/sdk/src/trace/batch_span_processor.cc +++ b/sdk/src/trace/batch_span_processor.cc @@ -98,7 +98,8 @@ void BatchSpanProcessor::OnEnd(std::unique_ptr &&span) noexcept size_t buffer_size = buffer_.size(); if (buffer_size >= max_queue_size_ / 2 || buffer_size >= max_export_batch_size_) { - // Best effort wakeup for worker thread. + // Notified without lock to reduce contention for span end. If this notify is lost, + // the worker thread may wait until next schedule or until the next notify attempt. synchronization_data_->cv.notify_all(); } } From 32da17e2f9227e9f88935dd9affeb109e70ca3d3 Mon Sep 17 00:00:00 2001 From: Deniz Ariyan Date: Mon, 10 Aug 2026 23:12:54 +0200 Subject: [PATCH 5/5] [SDK] Clarify required lock ordering --- sdk/src/trace/batch_span_processor.cc | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/sdk/src/trace/batch_span_processor.cc b/sdk/src/trace/batch_span_processor.cc index ad4f9c8884..403fcb55af 100644 --- a/sdk/src/trace/batch_span_processor.cc +++ b/sdk/src/trace/batch_span_processor.cc @@ -190,8 +190,11 @@ void BatchSpanProcessor::DoBackgroundWork() } #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */ - // Wait for `timeout` milliseconds. + // 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 lk(synchronization_data_->cv_m); synchronization_data_->cv.wait_for(lk, timeout, [this] { if (synchronization_data_->is_force_wakeup_background_worker.load(