Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mike-nugent-mongodb wants to merge 12 commits into
base: main
Choose a base branch
from
+39 −8
Open

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

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
e35bb01
[BUG] Prevent lost condition-variable wakeups during OTLP file, perio…
mike-nugent-mongodb Aug 5, 2026
6551f60
Revert changes to batch_span_processor.cc
mike-nugent-mongodb Aug 19, 2026
b28f264
Update changelog and fix additional races
mike-nugent-mongodb Aug 19, 2026
255f810
Merge remote-tracking branch 'fork/main' into exporter-shutdown-race
mike-nugent-mongodb Aug 19, 2026
c2e05e4
Add PR to changelog
mike-nugent-mongodb Aug 19, 2026
6b7a1db
Break after AfterWait
mike-nugent-mongodb Aug 20, 2026
18f3bd4
Merge remote-tracking branch 'fork/main' into exporter-shutdown-race
mike-nugent-mongodb Aug 20, 2026
f39c117
Update comment around is_shutdown
mike-nugent-mongodb Aug 24, 2026
ad6e820
Merge branch 'main' into exporter-shutdown-race
marcalff Sep 1, 2026
ca42402
Merge branch 'main' into exporter-shutdown-race
dbarker Sep 2, 2026
b9c7dd5
Merge branch 'main' into exporter-shutdown-race
marcalff Sep 3, 2026
be2890a
Merge branch 'main' into exporter-shutdown-race
marcalff Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,10 @@ Increment the:

## [Unreleased]

* [BUG] Prevent lost condition-variable wakeups in OTLP file exporter and periodic
metric exporter
[#4365](https://github.com/open-telemetry/opentelemetry-cpp/pull/4365)

* [CONFIGURATION] Build the configured resource detectors in SdkBuilder, apply
the `detection.attributes` include/exclude filter to the detected attributes,
and merge the resource per the resource SDK specification.
Expand Down
21 changes: 15 additions & 6 deletions exporters/otlp/src/otlp_file_client.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

@lalitb lalitb Aug 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we apply this same guarded store plus notify in OtlpFileSystemBackend::Shutdown() too? When there is nothing pending, ForceFlush() returns before notifying the worker, so Shutdown() can return while the background thread remains parked until flush_interval. The destructor now handles this correctly, but callers that keep the client alive after Shutdown() still retain the delayed worker.

@mike-nugent-mongodb mike-nugent-mongodb Aug 19, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

}
file_->background_thread_waker_cv.notify_all();
std::unique_ptr<std::thread> background_flush_thread;
{
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -1482,11 +1489,6 @@ 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)
{
Expand All @@ -1496,6 +1498,13 @@ class OPENTELEMETRY_LOCAL_SYMBOL OtlpFileSystemBackend : public OtlpFileAppender

{
std::unique_lock<std::mutex> lk(concurrency_file->background_thread_waker_lock);
// 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.
if (concurrency_file->is_shutdown.load(std::memory_order_acquire))

@mateenali66 mateenali66 Aug 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

with the early check gone, shutdown is observed only after one more flush pass, intended? fine if it's deliberate flush-on-shutdown, just checking

@mike-nugent-mongodb mike-nugent-mongodb Aug 19, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The check on is_shutdown occurs at almost the same time as previously, the only difference is that the check now occurs while holding concurrency_file->background_thread_waker_lock instead of prior to acquiring it. In both cases, if a flush is occurring while is_shutdown is set then the code will break prior to the next wait_for. Does that make sense or am I misunderstanding?

While looking into this I do notice there is an issue where a ForceFlush may not cause an additional flush and make take the full flush interval if file_->background_thread_waker_cv.notify_all(); happens just before the background thread blocks on concurrency_file->background_thread_waker_cv.wait_for(lk, flush_interval);. Fixing this would require a more involved change, so I'd prefer that to be separate. Some details on that: if you just add a variable to cause background_thread_waker_cv.wait_for to exit when it's set (e.g. is_force_background_thread_wake), the problem is that ForceFlush waits on background_thread_waiter_cv , so it could miss the notification from the background thread unless there is some additional synchronization (e.g., acquire the waiter lock before releasing the waker lock).

@mateenali66 mateenali66 Aug 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Re-read the loop against the head commit, the only thing between the old check position and the new one is the BeforeWait() hook, so no extra flush pass. Agreed.

That hook is the one thing the move does change. On the shutdown path BeforeWait() now fires and AfterWait() never does, because the break is inside the lock scope. Before this PR the shutdown break came ahead of BeforeWait(), so the pair stayed balanced. thread_instrumentation.h documents the two as bracketing a blocking wait, so an app that flips thread state in BeforeWait and restores it in AfterWait leaves it set at thread exit. Hoisting the break out of the lock scope keeps both the fence and the pairing:

bool shutdown_requested = false;
{
  std::unique_lock<std::mutex> lk(concurrency_file->background_thread_waker_lock);
  shutdown_requested = concurrency_file->is_shutdown.load(std::memory_order_acquire);
  if (!shutdown_requested)
  {
    concurrency_file->background_thread_waker_cv.wait_for(lk, flush_interval);
  }
}

then break after the AfterWait() block. Agreed the ForceFlush / wait_for race is a separate change.

@owent owent Aug 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

  std::unique_lock<std::mutex> lk(concurrency_file->background_thread_waker_lock);
  concurrency_file->background_thread_waker_cv.wait_for(lk, flush_interval, [concurrency_file]() {
    return concurrency_file->is_shutdown.load(std::memory_order_acquire);
  });

Is these codes more clear? We also use conditional wait_for in other components.

@mike-nugent-mongodb mike-nugent-mongodb Aug 20, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch, I've moved the break to after AfterWait.

As far as using the conditional wait_for, I don't think it would be a lot clearer right now because is_shutdown has to kept outside the function, so the code would be more like

  std::unique_lock<std::mutex> lk(concurrency_file->background_thread_waker_lock);
  concurrency_file->background_thread_waker_cv.wait_for(lk, flush_interval, [concurrency_file, &is_shutdown]() {
    return is_shutdown = concurrency_file->is_shutdown.load(std::memory_order_acquire);
  });

but let me know if this is preferred.

{
break;
}

@owent owent Aug 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In my understanding, When concurrency_file->is_shutdown is true, std::fflush should still need to be called one more time.So Shutdown will still flush all pending records.

@mike-nugent-mongodb mike-nugent-mongodb Aug 20, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is a preexisting issue, as previously is_shutdown did not cause an additional fflush, so I'd prefer to leave off fixing this.

@owent owent Aug 23, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the existing codes. Shutdown() will calls ForceFlush onece, which will wake up the background thread and calls std::fflush.

@mike-nugent-mongodb mike-nugent-mongodb Aug 24, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is no significant behavior change here because is_shutdown is set prior to wait_for, so both in the old code and new if Shutdown() (and hence ForceFlush()) is called while the thread is in wait_for an additional std::fflush will occur. It would be a problem if is_shutdown was set in a potential condition of wait_for (as suggested in the other comment) so I updated the code comment to mention this.

The new code does slightly expand the window for which a shutdown call can fail to fflush, but in a minor way: previously, Shutdown() called just after BeforeWait would cause fflush but now it does not (Shutdown() called after the std::unique_lock<std::mutex> creation still causes fflush because Shutdown() now blocks on acquiring the lock).

@owent owent Aug 27, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The behaviour before just follow the other OTLP exporters. Which mean when Shutdown is called, the pending records will be try to be exported once, but the new codes will break and skip std::fflush in the codes below( https://github.com/open-telemetry/opentelemetry-cpp/pull/4365/changes#diff-0a8555076f1a604583ef92e84a10c80eb2806d28624516bea8d71f911cae8d62R1521 ). In my understanding it also breaks the spec ( https://opentelemetry.io/docs/specs/otel/trace/sdk/#shutdown-1 , "Shutdown MUST include the effects of ForceFlush." ) ""

@mike-nugent-mongodb mike-nugent-mongodb Aug 27, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When the background writer thread is blocked on background_thread_waker_cv.wait_for and Shutdown is called, the if on line 1521 returns false because is_shutdown was set on line 1507 prior to when Shutdown was called and is not updated again until the next iteration of the loop. If the statement was

if(concurrency_file->is_shutdown.load(std::memory_order_acquire))
{
  break;
}

then it would be a problem.

@owent owent Sep 1, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, I misunderstand before, it uses if (is_shutdown), not if (concurrency_file->is_shutdown). LGTM after conflicts are resolved.

concurrency_file->background_thread_waker_cv.wait_for(lk, flush_interval);
}

Expand Down
12 changes: 11 additions & 1 deletion sdk/src/metrics/export/periodic_exporting_metric_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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_};

@mateenali66 mateenali66 Aug 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fence is right, the wait predicate checks IsShutdown() under cv_m_. the force flush path below has the same store-then-notify shape (the "must not wait for ever" workaround), in scope here or follow-up?

@mike-nugent-mongodb mike-nugent-mongodb Aug 19, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since it's in the same vein, I've added the change to ForceFlush as well.

@mateenali66 mateenali66 Aug 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ForceFlush change looks right, and no lock-order problem with it: this path takes force_flush_m_ then cv_m_ inside the predicate, and the worker's cv_m_ lock is scoped to the do-while body so it is released before CollectAndExportOnce runs.

One direction is still open though. This fixes the waker side (cv_). The completion side is unchanged: CollectAndExportOnce CASes force_flush_notified_sequence_ and calls force_flush_cv_.notify_all() holding nothing, so an OnForceFlush caller can evaluate the predicate, miss the notify, then wait. That is what the "must not wait for ever" chunked wait is compensating for, so the chunked loop has to stay as long as that store-then-notify is unguarded. Same bucket as the wait_for race you described, fine as a follow-up, just noting why that workaround can't come out yet.

}
cv_.notify_all();
worker_thread_.join();
}
Expand Down
Loading