[SDK] BatchSpanProcessor: wait for a full batch instead of draining partial batches - #4466
[SDK] BatchSpanProcessor: wait for a full batch instead of draining partial batches#4466yswdqz wants to merge 6 commits into
Conversation
…artial batches - Change the worker wait predicate from '!buffer_.empty()' to 'buffer_.size() >= max_export_batch_size_'. - Export() now only drains the entire buffer when a force flush is pending or the processor is shutting down; on normal wakeups it exports at most one batch of max_export_batch_size spans. This prevents the processor from waking up and draining partial trailing batches every time a span arrives, reducing CPU usage and gRPC request count while preserving ForceFlush/Shutdown drain semantics.
c49d86c to
3d02bbf
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #4466 +/- ##
==========================================
+ Coverage 82.62% 82.67% +0.05%
==========================================
Files 512 515 +3
Lines 20138 20194 +56
==========================================
+ Hits 16637 16693 +56
Misses 3501 3501
🚀 New features to boost your workflow:
|
|
Thanks for the fix. Please see clang-format errors, either run clang-format or apply this manually: |
| bool should_drain = notify_force_flush != 0 || | ||
| synchronization_data_->is_shutdown.load(std::memory_order_acquire); |
There was a problem hiding this comment.
Export() now only drains the entire buffer when a force flush is pending
or the processor is shutting down; on normal wakeups it exports at most one
batch of max_export_batch_size spans.
notify_force_flush (i.e., synchronization_data_->force_flush_pending_sequence) is a monotonically increasing counter that increases on every call to ForceFlush. This would mean this condition would be permanently true after the first time one calls BatchSpanProcessor::ForceFlush.
Hence while (should_drain) below never becomes while (false), so Export() keeps draining to empty on every call which means the one-batch-per-wakeup behavior this PR adds never takes effect after the first call to BatchSpanProcessor::ForceFlush.
I think something like this would actually solve that problem.
| bool should_drain = notify_force_flush != 0 || | |
| synchronization_data_->is_shutdown.load(std::memory_order_acquire); | |
| bool should_drain = | |
| notify_force_flush > | |
| synchronization_data_->force_flush_notified_sequence.load(std::memory_order_acquire) || | |
| synchronization_data_->is_shutdown.load(std::memory_order_acquire); |
Maybe you could add some tests to confirm/check?
This issue could be tested with a test exporter that records how many spans each Export() call receives and pauses inside the first call, so the worker is held mid-export and you control what's in the buffer when it resumes. You could then add a couple spans while it's paused mid export to see that it will still drain less than the max batch size immediately instead of returning after exporting one batch as the PR suggests IF the BatchSpanProcessor::ForceFlush was ever called.
There was a problem hiding this comment.
Good catch, thank you for the review~ Updated the condition to compare against force_flush_notified_sequence and added a regression test covering the scenario you described.
| std::uint64_t notify_force_flush = | ||
| synchronization_data_->force_flush_pending_sequence.load(std::memory_order_acquire); | ||
| if (notify_force_flush) | ||
| if (should_drain) |
There was a problem hiding this comment.
should_drain is doing two orthogonal jobs here: how many records to take, and whether to keep looping. Because is_shutdown alone now sets it, it hands the entire buffer to a single Export() call, bypassing max_export_batch_size. Before this PR, a program that never called ForceFlush had notify_force_flush == 0, so DrainQueue() still chunked; now a backlog of 3000 with max_export_batch_size = 2048 goes out as one 3000-span request instead of 2048 + 952. If the receiver rejects the oversized message (max_receive_message_length defaults to 4 MiB), the spans are already consumed and the result is discarded at the line - and at process exit there is no retry.
Suggest keeping the cap unconditional and letting should_drain control only the loop.
There was a problem hiding this comment.
You're right, thank you for the review~ The cap is now unconditional and should_drain only controls the loop. Added a separate test for shutdown draining as well.
…export_batch_size
adb1e93 to
0e7cbe4
Compare
|
Apologies for my oversight and insufficient testing. Thanks for the reviews. I've addressed all the feedback: fixed the clang-format issue, corrected the ForceFlush condition, kept the batch size cap unconditional, and added regression tests for both issues. |
|
Some CI jobs failed because the new BlockingMockSpanExporter had a memory leak and a data race. Pushed a fix and verified locally with ASan and TSan — all batch_span_processor_test tests pass now. |
fix #4449
Problem
BatchSpanProcessorcurrently wakes up whenever the buffer is non-empty andthen drains the entire buffer in a tight loop. Under steady load this produces
exports that are much smaller than
max_export_batch_size, causing:Changes
!buffer_.empty()tobuffer_.size() >= max_export_batch_size_.Export()now only drains the entire buffer when a force flush is pendingor the processor is shutting down; on normal wakeups it exports at most one
batch of
max_export_batch_sizespans.This preserves
ForceFlush/Shutdownsemantics while making the normalexport path strictly batch-oriented.
Performance
I constructed a test where one span is finished every 50 µs, and the above metrics capture the performance difference between the two implementations under that steady load. In real-world scenarios, spans tend to be completed in batches, so the performance gap would be much less pronounced under typical conditions.
Checklist
CHANGELOG.mdupdated for non-trivial changesbatch_span_processor_testandbatch_span_processor_test_stressall pass)