This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push:
new a62127d68 [util] correct invariant in ThreadPoolToken::Transition()
a62127d68 is described below
commit a62127d6808c40a298127ef371ae194f4185d1a6
Author: Alexey Serbin <[email protected]>
AuthorDate: Wed Mar 19 10:38:27 2025 -0700
[util] correct invariant in ThreadPoolToken::Transition()
This patch corrects the invariant check when transitioning from
State::RUNNING to State::GRACEFUL_QUIESCING in ThreadPoolToken::Close().
Without this fix, a couple of test scenarios in threadpool-test
would fail with output like below:
F20250319 10:15:25.060393 1859138 threadpool.cc:344] Check failed:
active_threads_ > 0 (0 vs. 0)
I also added extra checks for into ThreadPoolToken::Transition() and
updated the inline documentation for ThreadPoolToken::State.
This is a follow-up to 4e5cd00b62e6be94e410abf98dda6fb066c6e1bb.
Change-Id: I6cfcdff743ad0074446ab9664dda43e4a9cb23ea
Reviewed-on: http://gerrit.cloudera.org:8080/22642
Tested-by: Alexey Serbin <[email protected]>
Reviewed-by: Abhishek Chennaka <[email protected]>
---
src/kudu/util/threadpool.cc | 42 ++++++++++++++++++++++++++++--------------
src/kudu/util/threadpool.h | 15 +++++++++------
2 files changed, 37 insertions(+), 20 deletions(-)
diff --git a/src/kudu/util/threadpool.cc b/src/kudu/util/threadpool.cc
index b3c57d630..de60d1d87 100644
--- a/src/kudu/util/threadpool.cc
+++ b/src/kudu/util/threadpool.cc
@@ -203,9 +203,9 @@ void ThreadPoolToken::Close() {
switch (state_) {
case State::IDLE:
- // If there aren't any outstanding tasks, quiesce the token immediately.
- // Otherwise, transition from IDLE into GRACEFUL_QUIESCING state.
- Transition(entries_.empty() ? State::QUIESCED :
State::GRACEFUL_QUIESCING);
+ // There aren't any outstanding tasks and no active worker threads,
+ // so quiesce the token immediately.
+ Transition(State::QUIESCED);
return;
case State::RUNNING:
// Unconditionally transition the token into GRACEFUL_QUIESCING state.
@@ -322,10 +322,8 @@ void ThreadPoolToken::Transition(State new_state) {
switch (state_) {
case State::IDLE:
CHECK(new_state == State::RUNNING ||
- new_state == State::GRACEFUL_QUIESCING ||
new_state == State::QUIESCED);
- if (new_state == State::RUNNING ||
- new_state == State::GRACEFUL_QUIESCING) {
+ if (new_state == State::RUNNING) {
CHECK(!entries_.empty());
} else {
CHECK(entries_.empty());
@@ -337,19 +335,32 @@ void ThreadPoolToken::Transition(State new_state) {
new_state == State::GRACEFUL_QUIESCING ||
new_state == State::QUIESCING ||
new_state == State::QUIESCED);
- if (new_state != State::GRACEFUL_QUIESCING) {
+ if (new_state == State::QUIESCING) {
CHECK(entries_.empty());
- }
- if (new_state == State::GRACEFUL_QUIESCING || new_state ==
State::QUIESCING) {
CHECK_GT(active_threads_, 0);
}
+ if (new_state == State::QUIESCED || new_state == State::IDLE) {
+ CHECK(entries_.empty());
+ CHECK_EQ(active_threads_, 0);
+ }
+ if (new_state == State::GRACEFUL_QUIESCING) {
+ CHECK(active_threads_ > 0 || !entries_.empty());
+ }
break;
case State::GRACEFUL_QUIESCING:
CHECK(new_state == State::QUIESCING ||
new_state == State::QUIESCED);
+ CHECK(entries_.empty());
+ if (new_state == State::QUIESCING) {
+ CHECK_GT(active_threads_, 0);
+ }
+ if (new_state == State::QUIESCED) {
+ CHECK_EQ(active_threads_, 0);
+ }
break;
case State::QUIESCING:
CHECK(new_state == State::QUIESCED);
+ CHECK(entries_.empty());
CHECK_EQ(active_threads_, 0);
break;
case State::QUIESCED:
@@ -498,9 +509,8 @@ void ThreadPool::Shutdown() {
// (i.e. there are no active threads), the tasks will have been removed
// above and we can quiesce immediately. Otherwise, we need to wait for
// the threads to finish.
- t->Transition(t->active_threads_ > 0 ?
- ThreadPoolToken::State::QUIESCING :
- ThreadPoolToken::State::QUIESCED);
+ t->Transition(t->active_threads_ > 0 ?
ThreadPoolToken::State::QUIESCING
+ :
ThreadPoolToken::State::QUIESCED);
break;
default:
break;
@@ -849,8 +859,12 @@ void ThreadPool::DispatchThread() {
// Possible states:
// 1. The token was shut down while we ran its task. Transition to
QUIESCED.
- // 2. The token has no more queued tasks. Transition back to IDLE.
- // 3. The token has more tasks. Requeue it and transition back to RUNNABLE.
+ // 2. The token has no more queued tasks. Transition to IDLE.
+ // 3. The token has more tasks. Requeue it and keep the state RUNNING
+ // or GRACEFUL_QUIESCING.
+ // 4. The token was gracefully quiesced while its task was running, and
+ // (a) the token's queue has more elements: transition to
GRACEFUL_QUIESCING
+ // (b) the token's queue is empty: transition to QUIESCED
const ThreadPoolToken::State state = token->state();
DCHECK(state == ThreadPoolToken::State::RUNNING ||
state == ThreadPoolToken::State::GRACEFUL_QUIESCING ||
diff --git a/src/kudu/util/threadpool.h b/src/kudu/util/threadpool.h
index b855e0e3b..136f4805f 100644
--- a/src/kudu/util/threadpool.h
+++ b/src/kudu/util/threadpool.h
@@ -714,16 +714,19 @@ class ThreadPoolToken {
// worker thread finishes executing a task belonging
// to a shut down token or pool
enum class State {
- // Token has no queued tasks.
+ // Token has no queued tasks, and there aren't any active worker threads
+ // running the token's tasks.
IDLE,
- // A worker thread is running one of the token's previously queued tasks.
+ // A worker thread is running one of the token's previously queued tasks,
+ // or there aren't yet any active worker threads running the token's tasks
+ // at this particular moment, but the token's queue isn't empty.
RUNNING,
- // No new tasks may be submitted to the token. A worker thread is still
- // running one of the token's previously queued tasks, and will continue
- // running all the rest of the scheduled tasks if any are still present
- // in the queue unless the token is shut down in the process of doing so.
+ // No new tasks may be submitted to the token, but otherwise the same
+ // set of conditions apply in RUNNING state. Worker threads will continue
+ // running the scheduled tasks if any are still present in the queue
+ // unless the token is shut down in the process.
GRACEFUL_QUIESCING,
// No new tasks may be submitted to the token. A worker thread is still
