github-actions[bot] commented on code in PR #68264:
URL: https://github.com/apache/doris/pull/68264#discussion_r4056607789
##########
be/src/runtime/workload_group/workload_group.cpp:
##########
@@ -634,17 +634,7 @@ Status
WorkloadGroup::upsert_thread_pool_no_lock(WorkloadGroupInfo* wg_info,
LOG(INFO) << "[upsert wg thread pool] create " + pool_name + "
succ, gid=" << wg_id
<< ", max thread num=" << max_flush_thread_num
<< ", min thread num=" << min_flush_thread_num;
- // Register the new pool with adaptive thread controller
- if (config::enable_adaptive_flush_threads) {
- auto* controller =
-
ExecEnv::GetInstance()->storage_engine().adaptive_thread_controller();
- auto* flush_pool = _memtable_flush_pool.get();
- controller->add("flush_wg_" + std::to_string(_id),
{flush_pool},
-
AdaptiveThreadPoolController::make_flush_adjust_func(controller,
-
flush_pool),
- config::max_flush_thread_num_per_cpu,
- config::min_flush_thread_num_per_cpu);
- }
+ register_adaptive_flush_no_lock();
Review Comment:
[P2] Cancel registrations when partial WG setup unwinds
If an earlier task/scan scheduler `start()` fails,
`upsert_thread_pool_no_lock()` retains that error but continues and can
successfully register this flush pool. `create_internal_wg()` then returns the
error before inserting the group, so the local `WorkloadGroup` is destroyed
through its default destructor; that frees `_memtable_flush_pool` without
calling either new cancellation helper, while the timer still captures the raw
pool pointer. Please drain the registration on this error path (or bind it to
object lifetime) before the group can be destroyed, and cover the partial-start
case in a test.
##########
be/test/storage/adaptive_thread_pool_controller_test.cpp:
##########
@@ -68,6 +71,56 @@ class AdaptiveThreadPoolControllerTest : public
testing::Test {
if (_pool2) _pool2->shutdown();
}
+ void check_cancel_race(const std::string& point) {
+ config::enable_adaptive_flush_threads = true;
+ auto* sp = SyncPoint::get_instance();
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+ SyncPoint::CallbackGuard guard;
+ std::promise<void> entered;
+ std::promise<void> release;
+ std::promise<void> cancelling;
+ auto entered_future = entered.get_future();
+ auto release_future = release.get_future().share();
+ auto cancelling_future = cancelling.get_future();
+ sp->set_call_back(
+ point,
+ [&](auto&&) {
+ entered.set_value();
+ release_future.wait();
+ },
+ &guard);
+ sp->set_call_back(
+ "AdaptiveThreadPoolController::cancel_stopped",
Review Comment:
[P1] Keep both SyncPoint callbacks registered
`set_call_back(..., &guard)` move-assigns a new `CallbackGuard`, and that
assignment clears the point previously held by the same guard. Registering
`cancel_stopped` here therefore removes the `callback_entered`/`before_rearm`
callback installed just above. As a result, `entered` is never fulfilled and
both new tests wait five seconds and hit `FAIL()` instead of testing
cancellation. Please use a separate guard for each callback.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
