This is an automated email from the ASF dual-hosted git repository.
HappenLee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f2f2a5701f0 [fix](be) Fix waiter accounting in blocking queue (#65827)
f2f2a5701f0 is described below
commit f2f2a5701f0de5c140debe6812019a03f07549d1
Author: TengJianPing <[email protected]>
AuthorDate: Thu Jul 23 11:13:26 2026 +0800
[fix](be) Fix waiter accounting in blocking queue (#65827)
### What problem does this PR solve?
Issue Number: None
Problem Summary:
BlockingQueue split ownership of waiter counters between waiting threads
and notifying threads. Around a timeout race, both sides could decrement
the same registration, while a spurious wakeup could leave a stale
registration behind. Once the counters drifted, a real waiter could be
represented by zero and miss a notification, leaving routine load queue
operations blocked until the one-hour fallback timeout. This is a missed
wakeup and potentially long stall, rather than a permanent deadlock. In
addition, try_put read the shutdown flag and queue size before acquiring
the mutex, which introduced a data race.
The following sequence shows how lock contention after a timeout
corrupts the consumer waiter count:
| Time | Consumer threads | Producer threads | get_waiting | Result |
| --- | --- | --- | ---: | --- |
| T0 | C increments the counter and calls wait_for, which releases the
mutex. | - | 0 -> 1 | C is registered as a condition-variable waiter. |
| T1 | C's timeout expires. C leaves the condition-variable wait set,
but wait_for must reacquire the mutex before returning. | P acquires the
mutex before C can reacquire it. | 1 | C's registration is still visible
because C cannot update it without the mutex. |
| T2 | C remains blocked while trying to reacquire the mutex. | P pushes
an item, decrements the counter from 1 to 0, unlocks, and calls
notify_one. | 1 -> 0 | The notification cannot wake C because its timed
wait has already expired. |
| T3 | C reacquires the mutex. wait_for returns timeout, so the old
timeout branch decrements the same registration again. C then consumes
the item. | - | 0 -> SIZE_MAX | The unsigned waiter count wraps around.
|
| T4 | After the queue becomes empty, a new consumer C2 increments the
counter and starts waiting. | - | SIZE_MAX -> 0 | A real waiter is now
represented by zero. |
| T5 | Absent a spurious wakeup, shutdown, or another notification, C2
remains asleep until its next timeout, which is one hour by default. |
P2 pushes an item, observes zero, and skips notify_one. | 0 | C2 suffers
a missed wakeup and a potentially one-hour stall. |
Make each waiting thread register immediately before wait_for and always
unregister after wait_for returns, regardless of the wakeup reason.
Notifiers now inspect waiter counters under the mutex without consuming
registrations, then release the mutex before notifying. Remove the
unsynchronized try_put fast path and add deterministic tests for both
consumer and producer timeout races.
### Release note
Fix potential long stalls in backend blocking queues under concurrent
timeout and notification.
### Check List (For Author)
- Test: Unit Test
- Added BlockingQueueWaiterTest.TimedGetNotificationRace and
BlockingQueueWaiterTest.TimedPutNotificationRace.
- `./run-be-ut.sh --run --filter='BlockingQueueWaiterTest.*' -j 32`
- Behavior changed: Yes. Waiter registrations are now released by the
waiting thread, and try_put checks queue state only while holding the
mutex.
- Does this need documentation: No
---
be/src/util/blocking_queue.hpp | 69 +++++---
be/test/util/blocking_queue_test.cpp | 316 +++++++++++++++++++++++++++++++++++
2 files changed, 363 insertions(+), 22 deletions(-)
diff --git a/be/src/util/blocking_queue.hpp b/be/src/util/blocking_queue.hpp
index dbf59c341b1..0052f8dae24 100644
--- a/be/src/util/blocking_queue.hpp
+++ b/be/src/util/blocking_queue.hpp
@@ -30,6 +30,10 @@
#include "common/logging.h"
#include "util/stopwatch.hpp"
+#ifdef BE_TEST
+#include "cpp/sync_point.h"
+#endif
+
namespace doris {
// Fixed capacity FIFO queue, where both BlockingGet and BlockingPut
operations block
// if the queue is empty or full, respectively.
@@ -56,21 +60,26 @@ public:
MonotonicStopWatch timer;
timer.start();
std::unique_lock<std::mutex> unique_lock(_lock);
+#ifdef BE_TEST
+ TEST_SYNC_POINT("BlockingQueue::controlled_blocking_get::after_lock");
+#endif
while (!(_shutdown || !_list.empty())) {
++_get_waiting;
- if (_get_cv.wait_for(unique_lock,
std::chrono::milliseconds(cv_wait_timeout_ms)) ==
- std::cv_status::timeout) {
- _get_waiting--;
- }
+#ifdef BE_TEST
+
TEST_SYNC_POINT("BlockingQueue::controlled_blocking_get::before_wait");
+#endif
+ _get_cv.wait_for(unique_lock,
std::chrono::milliseconds(cv_wait_timeout_ms));
+ DCHECK_GT(_get_waiting, 0);
+ --_get_waiting;
}
_total_get_wait_time += timer.elapsed_time();
if (!_list.empty()) {
*out = _list.front();
_list.pop_front();
- if (_put_waiting > 0) {
- --_put_waiting;
- unique_lock.unlock();
+ const bool has_put_waiter = _put_waiting > 0;
+ unique_lock.unlock();
+ if (has_put_waiter) {
_put_cv.notify_one();
}
return true;
@@ -93,10 +102,12 @@ public:
std::unique_lock<std::mutex> unique_lock(_lock);
while (!(_shutdown || _list.size() < _max_elements)) {
++_put_waiting;
- if (_put_cv.wait_for(unique_lock,
std::chrono::milliseconds(cv_wait_timeout_ms)) ==
- std::cv_status::timeout) {
- _put_waiting--;
- }
+#ifdef BE_TEST
+
TEST_SYNC_POINT("BlockingQueue::controlled_blocking_put::before_wait");
+#endif
+ _put_cv.wait_for(unique_lock,
std::chrono::milliseconds(cv_wait_timeout_ms));
+ DCHECK_GT(_put_waiting, 0);
+ --_put_waiting;
}
_total_put_wait_time += timer.elapsed_time();
@@ -105,9 +116,9 @@ public:
}
_list.push_back(val);
- if (_get_waiting > 0) {
- --_get_waiting;
- unique_lock.unlock();
+ const bool has_get_waiter = _get_waiting > 0;
+ unique_lock.unlock();
+ if (has_get_waiter) {
_get_cv.notify_one();
}
return true;
@@ -115,13 +126,12 @@ public:
// Return false if queue full or has been shutdown.
bool try_put(const T& val) {
- if (_shutdown || _list.size() >= _max_elements) {
- return false;
- }
-
MonotonicStopWatch timer;
timer.start();
std::unique_lock<std::mutex> unique_lock(_lock);
+#ifdef BE_TEST
+ TEST_SYNC_POINT("BlockingQueue::try_put::after_lock");
+#endif
_total_put_wait_time += timer.elapsed_time();
if (_shutdown || _list.size() >= _max_elements) {
@@ -129,9 +139,9 @@ public:
}
_list.push_back(val);
- if (_get_waiting > 0) {
- --_get_waiting;
- unique_lock.unlock();
+ const bool has_get_waiter = _get_waiting > 0;
+ unique_lock.unlock();
+ if (has_get_waiter) {
_get_cv.notify_one();
}
return true;
@@ -155,6 +165,18 @@ public:
uint32_t get_capacity() const { return _max_elements; }
+#ifdef BE_TEST
+ size_t get_waiting_count_for_test() const {
+ std::lock_guard<std::mutex> guard(_lock);
+ return _get_waiting;
+ }
+
+ size_t put_waiting_count_for_test() const {
+ std::lock_guard<std::mutex> guard(_lock);
+ return _put_waiting;
+ }
+#endif
+
// Returns the total amount of time threads have blocked in BlockingGet.
uint64_t total_get_wait_time() const { return _total_get_wait_time; }
@@ -167,11 +189,14 @@ private:
const int _max_elements;
std::condition_variable _get_cv; // 'get' callers wait on this
std::condition_variable _put_cv; // 'put' callers wait on this
- // _lock guards access to _list, total_get_wait_time, and
total_put_wait_time
+ // _lock guards access to _shutdown, _get_waiting, _put_waiting, _list,
total_get_wait_time, and total_put_wait_time
mutable std::mutex _lock;
std::list<T> _list;
std::atomic<uint64_t> _total_get_wait_time;
std::atomic<uint64_t> _total_put_wait_time;
+ // Number of threads currently inside the corresponding wait_for() call.
The waiter that
+ // increments a counter is also responsible for decrementing it after
every wakeup reason.
+ // Notifiers only inspect these counters and never consume a waiter's
registration.
size_t _get_waiting;
size_t _put_waiting;
};
diff --git a/be/test/util/blocking_queue_test.cpp
b/be/test/util/blocking_queue_test.cpp
index 419ffe49ca1..00cf891fdba 100644
--- a/be/test/util/blocking_queue_test.cpp
+++ b/be/test/util/blocking_queue_test.cpp
@@ -21,8 +21,15 @@
#include <gtest/gtest.h>
#include <unistd.h>
+#include <chrono>
+#include <condition_variable>
+#include <future>
#include <mutex>
#include <thread>
+#include <vector>
+
+#include "cpp/sync_point.h"
+#include "util/defer_op.h"
namespace doris {
@@ -142,4 +149,313 @@ TEST(BlockingQueueTest, TestMultipleThreads) {
test.Run();
}
+// Coordinates the queue threads through callbacks installed at the sync
points below. The
+// waiter callback runs while the queue mutex is still held, immediately
before wait_for(). The
+// operation callback runs after the peer operation has acquired that same
mutex. Holding the
+// first peer operation there past the wait timeout deterministically creates
this ordering:
+//
+// waiter times out -> peer changes the queue and unlocks -> waiter
reacquires the mutex
+//
+// This is the ordering that made the old notifier and the timed waiter
decrement the same waiter
+// registration. Later peer operations are not blocked, so the test can verify
whether a real
+// waiter receives the notification that follows the corrupted registration.
+class GetPutRaceCoordinator {
+public:
+ // The first callback is the waiter used to create the counter corruption.
The second callback
+ // is the waiter used to verify that the next notification is not lost.
+ void get_waiter_registered() {
+ std::lock_guard lock(_get_lock);
+ ++_get_waiter_count;
+ _get_cv.notify_all();
+ }
+
+ void put_waiter_registered() {
+ std::lock_guard lock(_put_lock);
+ ++_put_waiter_count;
+ _put_cv.notify_all();
+ }
+
+ void block_first_get() {
+ std::unique_lock lock(_get_lock);
+ ++_get_operation_count;
+ // Only the first get must hold the queue mutex across the put
timeout. The later get must
+ // proceed normally to exercise BlockingQueue's put notification
decision.
+ if (_get_operation_count != 1) {
+ return;
+ }
+ _first_get_has_lock = true;
+ _get_cv.notify_all();
+ _get_cv.wait(lock, [&] { return _release_get; });
+ }
+
+ void block_first_put() {
+ std::unique_lock lock(_put_lock);
+ ++_put_operation_count;
+ // Only the first put must hold the queue mutex across the get
timeout. The later put must
+ // proceed normally to exercise BlockingQueue's get notification
decision.
+ if (_put_operation_count != 1) {
+ return;
+ }
+ _first_put_has_lock = true;
+ _put_cv.notify_all();
+ _put_cv.wait(lock, [&] { return _release_put; });
+ }
+
+ template <typename Rep, typename Period>
+ bool wait_for_get_waiter_count(int expected,
+ const std::chrono::duration<Rep, Period>&
timeout) {
+ std::unique_lock lock(_get_lock);
+ return _get_cv.wait_for(lock, timeout, [&] { return _get_waiter_count
>= expected; });
+ }
+
+ template <typename Rep, typename Period>
+ bool wait_for_put_waiter_count(int expected,
+ const std::chrono::duration<Rep, Period>&
timeout) {
+ std::unique_lock lock(_put_lock);
+ return _put_cv.wait_for(lock, timeout, [&] { return _put_waiter_count
>= expected; });
+ }
+
+ template <typename Rep, typename Period>
+ bool wait_for_first_get_lock(const std::chrono::duration<Rep, Period>&
timeout) {
+ std::unique_lock lock(_get_lock);
+ return _get_cv.wait_for(lock, timeout, [&] { return
_first_get_has_lock; });
+ }
+
+ template <typename Rep, typename Period>
+ bool wait_for_first_put_lock(const std::chrono::duration<Rep, Period>&
timeout) {
+ std::unique_lock lock(_put_lock);
+ return _put_cv.wait_for(lock, timeout, [&] { return
_first_put_has_lock; });
+ }
+
+ void release_blocked_get() {
+ std::lock_guard lock(_get_lock);
+ _release_get = true;
+ _get_cv.notify_all();
+ }
+
+ void release_blocked_put() {
+ std::lock_guard lock(_put_lock);
+ _release_put = true;
+ _put_cv.notify_all();
+ }
+
+private:
+ // Coordinates get waiter registration and a get operation that holds the
queue mutex. This
+ // state is independent from the put path so each callback changes only
get-related state.
+ std::mutex _get_lock;
+ std::condition_variable _get_cv;
+ int _get_waiter_count = 0;
+ int _get_operation_count = 0;
+ bool _first_get_has_lock = false;
+ bool _release_get = false;
+
+ // Coordinates put waiter registration and a put operation that holds the
queue mutex.
+ std::mutex _put_lock;
+ std::condition_variable _put_cv;
+ int _put_waiter_count = 0;
+ int _put_operation_count = 0;
+ bool _first_put_has_lock = false;
+ bool _release_put = false;
+};
+
+class BlockingQueueWaiterTest : public testing::Test {
+protected:
+ void SetUp() override {
+ _sync_point = SyncPoint::get_instance();
+ _sync_point->clear_all_call_backs();
+ _sync_point->clear_trace();
+ _sync_point->enable_processing();
+ }
+
+ void TearDown() override {
+ _sync_point->disable_processing();
+ _sync_point->clear_all_call_backs();
+ _sync_point->clear_trace();
+ }
+
+ SyncPoint* _sync_point = nullptr;
+};
+
+// Verifies that a timed get remains the sole owner of its _get_waiting
registration. The first
+// phase forces try_put to hold the queue mutex after the get timeout expires;
the old code made
+// try_put and the timed get both decrement the same registration, wrapping
the counter to
+// SIZE_MAX. The second phase registers another get and verifies that a
following try_put observes
+// that waiter and wakes it immediately instead of leaving it blocked until
the timeout fallback.
+TEST_F(BlockingQueueWaiterTest, TimedGetNotificationRace) {
+ // The first wait is deliberately short so the producer can hold the queue
mutex past its
+ // deadline. The second wait has a 30-second fallback, while the assertion
waits only 5
+ // seconds; therefore, a ready future proves that notify_one() woke it
rather than its timeout.
+ constexpr int64_t kTimedGetWaitTimeoutMs = 100;
+ constexpr int64_t kNotifiedGetWaitTimeoutMs = 30 * 1000;
+ constexpr auto kCoordinationTimeout = std::chrono::seconds(5);
+ constexpr auto kGetNotificationDeadline = std::chrono::seconds(5);
+
+ BlockingQueue<int32_t> get_queue(1);
+ GetPutRaceCoordinator get_race;
+
_sync_point->set_call_back("BlockingQueue::controlled_blocking_get::before_wait",
+ [&](auto&&) { get_race.get_waiter_registered();
});
+ _sync_point->set_call_back("BlockingQueue::try_put::after_lock",
+ [&](auto&&) { get_race.block_first_put(); });
+
+ std::promise<int32_t> timed_get_promise;
+ auto timed_get_future = timed_get_promise.get_future();
+ std::promise<bool> try_put_promise;
+ auto try_put_future = try_put_promise.get_future();
+ std::promise<int32_t> notified_get_promise;
+ auto notified_get_future = notified_get_promise.get_future();
+ std::vector<std::thread> get_threads;
+ // Keep this guard after every promise and future declaration. On a fatal
assertion it runs
+ // first, releases either blocked callback, shuts down the queue, and
joins all workers before
+ // their referenced asynchronous state is destroyed.
+ Defer cleanup {[&] {
+ get_race.release_blocked_put();
+ get_queue.shutdown();
+ for (auto& thread : get_threads) {
+ thread.join();
+ }
+ }};
+
+ get_threads.emplace_back([&] {
+ int32_t get_value = -1;
+ get_queue.controlled_blocking_get(&get_value, kTimedGetWaitTimeoutMs);
+ timed_get_promise.set_value(get_value);
+ });
+ ASSERT_TRUE(get_race.wait_for_get_waiter_count(1, kCoordinationTimeout));
+ EXPECT_EQ(get_queue.get_waiting_count_for_test(), 1);
+ EXPECT_EQ(get_queue.put_waiting_count_for_test(), 0);
+
+ // Phase 1: timed_get has registered _get_waiting=1 and released the queue
mutex in
+ // wait_for(). try_put then acquires that mutex and is stopped by the
callback. Keeping try_put
+ // there for twice the get timeout guarantees that timed_get expires while
it is unable to
+ // reacquire the mutex.
+ get_threads.emplace_back([&] {
try_put_promise.set_value(get_queue.try_put(1)); });
+ ASSERT_TRUE(get_race.wait_for_first_put_lock(kCoordinationTimeout));
+
+ std::this_thread::sleep_for(std::chrono::milliseconds(2 *
kTimedGetWaitTimeoutMs));
+ get_race.release_blocked_put();
+ // In the old implementation try_put decremented _get_waiting from 1 to 0
before notifying,
+ // then timed_get reacquired the mutex and decremented the same
registration again because
+ // wait_for() returned timeout. The size_t counter consequently wrapped
from 0 to SIZE_MAX. In
+ // the fixed implementation only timed_get owns the registration, so the
counter returns to
+ // zero.
+ ASSERT_EQ(try_put_future.wait_for(kCoordinationTimeout),
std::future_status::ready);
+ EXPECT_TRUE(try_put_future.get());
+ ASSERT_EQ(timed_get_future.wait_for(kCoordinationTimeout),
std::future_status::ready);
+ EXPECT_EQ(timed_get_future.get(), 1);
+ EXPECT_EQ(get_queue.get_waiting_count_for_test(), 0);
+ EXPECT_EQ(get_queue.put_waiting_count_for_test(), 0);
+
+ // Phase 2: in the old implementation notified_get changed _get_waiting
from SIZE_MAX to 0. The
+ // following try_put therefore observed no get waiter and skipped
notify_one(), leaving
+ // notified_get asleep until its 30-second fallback. With correct
accounting notified_get
+ // changes the counter from 0 to 1 and try_put wakes it within the
5-second notification
+ // deadline.
+ get_threads.emplace_back([&] {
+ int32_t get_value = -1;
+ get_queue.controlled_blocking_get(&get_value,
kNotifiedGetWaitTimeoutMs);
+ notified_get_promise.set_value(get_value);
+ });
+ ASSERT_TRUE(get_race.wait_for_get_waiter_count(2, kCoordinationTimeout));
+ EXPECT_EQ(get_queue.get_waiting_count_for_test(), 1);
+ EXPECT_EQ(get_queue.put_waiting_count_for_test(), 0);
+
+ EXPECT_TRUE(get_queue.try_put(2));
+ ASSERT_EQ(notified_get_future.wait_for(kGetNotificationDeadline),
std::future_status::ready);
+ EXPECT_EQ(notified_get_future.get(), 2);
+ EXPECT_EQ(get_queue.get_waiting_count_for_test(), 0);
+ EXPECT_EQ(get_queue.put_waiting_count_for_test(), 0);
+}
+
+// Verifies the symmetric rule for a timed put and its _put_waiting
registration. The first phase
+// forces blocking_get to hold the queue mutex after the put timeout expires;
the old code made the
+// get and timed put both decrement the registration and wrap the counter. The
second phase
+// registers another put, removes an item with blocking_get, and verifies that
the put is notified
+// immediately and can add its item without waiting for the timeout fallback.
+TEST_F(BlockingQueueWaiterTest, TimedPutNotificationRace) {
+ // As in TimedGetNotificationRace, the 30-second second wait and 5-second
assertion deadline
+ // distinguish a real notification from the periodic timeout fallback.
+ constexpr int64_t kTimedPutWaitTimeoutMs = 100;
+ constexpr int64_t kNotifiedPutWaitTimeoutMs = 30 * 1000;
+ constexpr auto kCoordinationTimeout = std::chrono::seconds(5);
+ constexpr auto kPutNotificationDeadline = std::chrono::seconds(5);
+
+ BlockingQueue<int32_t> put_queue(1);
+ ASSERT_TRUE(put_queue.try_put(0));
+
+ GetPutRaceCoordinator put_race;
+
_sync_point->set_call_back("BlockingQueue::controlled_blocking_put::before_wait",
+ [&](auto&&) { put_race.put_waiter_registered();
});
+
_sync_point->set_call_back("BlockingQueue::controlled_blocking_get::after_lock",
+ [&](auto&&) { put_race.block_first_get(); });
+
+ std::promise<bool> timed_put_promise;
+ auto timed_put_future = timed_put_promise.get_future();
+ std::promise<int32_t> blocking_get_promise;
+ auto blocking_get_future = blocking_get_promise.get_future();
+ std::promise<bool> notified_put_promise;
+ auto notified_put_future = notified_put_promise.get_future();
+ std::vector<std::thread> put_threads;
+ // The guard is declared last so every failure path joins the workers
before their promises and
+ // futures are destroyed.
+ Defer cleanup {[&] {
+ put_race.release_blocked_get();
+ put_queue.shutdown();
+ for (auto& thread : put_threads) {
+ thread.join();
+ }
+ }};
+
+ put_threads.emplace_back([&] {
+ timed_put_promise.set_value(put_queue.controlled_blocking_put(1,
kTimedPutWaitTimeoutMs));
+ });
+ ASSERT_TRUE(put_race.wait_for_put_waiter_count(1, kCoordinationTimeout));
+ EXPECT_EQ(put_queue.get_waiting_count_for_test(), 0);
+ EXPECT_EQ(put_queue.put_waiting_count_for_test(), 1);
+
+ // Phase 1 is the put-side mirror of the get race. timed_put registers
_put_waiting=1 and waits
+ // because the queue contains 0. blocking_get then acquires and holds the
queue mutex until the
+ // put timeout has expired, preventing timed_put from returning from
wait_for().
+ put_threads.emplace_back([&] {
+ int32_t get_value = -1;
+ put_queue.blocking_get(&get_value);
+ blocking_get_promise.set_value(get_value);
+ });
+ ASSERT_TRUE(put_race.wait_for_first_get_lock(kCoordinationTimeout));
+
+ std::this_thread::sleep_for(std::chrono::milliseconds(2 *
kTimedPutWaitTimeoutMs));
+ put_race.release_blocked_get();
+ // Previously blocking_get decremented _put_waiting when it removed 0, and
timed_put decremented
+ // it again after reacquiring the mutex with a timeout result. The counter
wrapped to SIZE_MAX
+ // before timed_put put 1. With single-owner accounting timed_put performs
the only decrement
+ // and leaves the counter at zero.
+ ASSERT_EQ(blocking_get_future.wait_for(kCoordinationTimeout),
std::future_status::ready);
+ EXPECT_EQ(blocking_get_future.get(), 0);
+ ASSERT_EQ(timed_put_future.wait_for(kCoordinationTimeout),
std::future_status::ready);
+ EXPECT_TRUE(timed_put_future.get());
+ EXPECT_EQ(put_queue.get_waiting_count_for_test(), 0);
+ EXPECT_EQ(put_queue.put_waiting_count_for_test(), 0);
+
+ // Phase 2: the old SIZE_MAX counter wrapped to zero when notified_put
registered. The following
+ // blocking_get therefore skipped notify_one(), so notified_put could not
put 2 before the
+ // 5-second deadline. The fixed counter is one, causing blocking_get to
wake notified_put
+ // immediately and allowing the final value check to finish.
+ put_threads.emplace_back([&] {
+ notified_put_promise.set_value(
+ put_queue.controlled_blocking_put(2,
kNotifiedPutWaitTimeoutMs));
+ });
+ ASSERT_TRUE(put_race.wait_for_put_waiter_count(2, kCoordinationTimeout));
+ EXPECT_EQ(put_queue.get_waiting_count_for_test(), 0);
+ EXPECT_EQ(put_queue.put_waiting_count_for_test(), 1);
+
+ int32_t get_value = -1;
+ EXPECT_TRUE(put_queue.blocking_get(&get_value));
+ EXPECT_EQ(get_value, 1);
+ ASSERT_EQ(notified_put_future.wait_for(kPutNotificationDeadline),
std::future_status::ready);
+ EXPECT_TRUE(notified_put_future.get());
+ EXPECT_EQ(put_queue.get_waiting_count_for_test(), 0);
+ EXPECT_EQ(put_queue.put_waiting_count_for_test(), 0);
+ EXPECT_TRUE(put_queue.blocking_get(&get_value));
+ EXPECT_EQ(get_value, 2);
+}
+
} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
