xinyiZzz commented on code in PR #47462:
URL: https://github.com/apache/doris/pull/47462#discussion_r2003212388
##########
be/src/runtime/workload_group/workload_group_manager.cpp:
##########
@@ -287,6 +257,642 @@ void WorkloadGroupMgr::refresh_workload_group_metrics() {
}
}
+void WorkloadGroupMgr::add_paused_query(const std::shared_ptr<QueryContext>&
query_ctx,
+ int64_t reserve_size, const Status&
status) {
+ DCHECK(query_ctx != nullptr);
+ query_ctx->update_paused_reason(status);
+ query_ctx->set_low_memory_mode();
+ query_ctx->set_memory_sufficient(false);
+ std::lock_guard<std::mutex> lock(_paused_queries_lock);
+ auto wg = query_ctx->workload_group();
+ auto&& [it, inserted] = _paused_queries_list[wg].emplace(
+ query_ctx,
doris::GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted,
+ doris::GlobalMemoryArbitrator::any_workload_group_exceed_limit,
reserve_size);
+ // Check if this is an invalid reserve, for example, if the reserve size
is too large, larger than the query limit
+ // if hard limit is enabled, then not need enable other queries hard limit.
+ if (inserted) {
+ LOG(INFO) << "Insert one new paused query: " <<
query_ctx->debug_string()
+ << ", workload group: " << wg->debug_string();
+ }
+}
+
+/**
+ * Strategy 1: A revocable query should not have any running
task(PipelineTask).
+ * strategy 2: If the workload group has any task exceed workload group
memlimit, then set all queryctx's memlimit
+ * strategy 3: If any query exceed process memlimit, then should clear all
caches.
+ * strategy 4: If any query exceed query's memlimit, then do spill disk or
cancel it.
+ * strategy 5: If any query exceed process's memlimit and cache is zero, then
do following:
+ */
+void WorkloadGroupMgr::handle_paused_queries() {
+ {
+ std::shared_lock<std::shared_mutex> r_lock(_group_mutex);
+ for (auto& [wg_id, wg] : _workload_groups) {
+ std::unique_lock<std::mutex> lock(_paused_queries_lock);
+ if (_paused_queries_list[wg].empty()) {
+ // Add an empty set to wg that not contains paused queries.
+ }
+ }
+ }
+
+ std::unique_lock<std::mutex> lock(_paused_queries_lock);
+ bool has_revoked_from_other_group = false;
+ bool has_query_exceed_process_memlimit = false;
+ for (auto it = _paused_queries_list.begin(); it !=
_paused_queries_list.end();) {
+ auto& queries_list = it->second;
+ auto query_count = queries_list.size();
+ const auto& wg = it->first;
+
+ if (query_count != 0) {
+ LOG_EVERY_T(INFO, 1) << "Paused queries count of wg " <<
wg->name() << ": "
+ << query_count;
+ }
+
+ bool has_changed_hard_limit = false;
+ int64_t flushed_memtable_bytes = 0;
+ // If the query is paused because its limit exceed the query itself's
memlimit, then just spill disk.
+ // The query's memlimit is set using slot mechanism and its value is
set using the user settings, not
+ // by weighted value. So if reserve failed, then it is actually exceed
limit.
+ for (auto query_it = queries_list.begin(); query_it !=
queries_list.end();) {
+ auto query_ctx = query_it->query_ctx_.lock();
+ // The query is finished during in paused list.
+ if (query_ctx == nullptr) {
+ LOG(INFO) << "Query: " << query_it->query_id() << " is
nullptr, erase it.";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ if (query_ctx->is_cancelled()) {
+ LOG(INFO) << "Query: " << print_id(query_ctx->query_id())
+ << " was canceled, remove from paused list";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+
+ if
(query_ctx->paused_reason().is<ErrorCode::QUERY_MEMORY_EXCEEDED>()) {
+ // Streamload, kafka load, group commit will never have query
memory exceeded error because
+ // their query limit is very large.
+ bool spill_res =
+ handle_single_query_(query_ctx,
query_it->reserve_size_,
+ query_it->elapsed_time(),
query_ctx->paused_reason());
+ if (!spill_res) {
+ ++query_it;
+ continue;
+ } else {
+ VLOG_DEBUG << "Query: " << print_id(query_ctx->query_id())
+ << " remove from paused list";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ } else if
(query_ctx->paused_reason().is<ErrorCode::WORKLOAD_GROUP_MEMORY_EXCEEDED>()) {
+ // Only deal with non overcommit workload group.
+ if (wg->enable_memory_overcommit()) {
+ // Soft limit wg will only reserve failed when process
limit exceed. But in some corner case,
+ // when reserve, the wg is hard limit, the query reserve
failed, but when this loop run
+ // the wg is converted to soft limit.
+ // So that should resume the query.
+ LOG(WARNING)
+ << "Query: " << print_id(query_ctx->query_id())
+ << " reserve memory failed because exceed workload
group memlimit, it "
+ "should not happen, resume it again. paused
reason: "
+ << query_ctx->paused_reason();
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ // check if the reserve is too large, if it is too large,
+ // should set the query's limit only.
+ // Check the query's reserve with expected limit.
+ if (query_ctx->adjusted_mem_limit() <
+ query_ctx->get_mem_tracker()->consumption() +
query_it->reserve_size_) {
+ query_ctx->set_mem_limit(query_ctx->adjusted_mem_limit());
+ query_ctx->set_memory_sufficient(true);
+ LOG(INFO) << "Workload group memory reserve failed because
"
+ << query_ctx->debug_string() << " reserve size "
+ <<
PrettyPrinter::print_bytes(query_it->reserve_size_)
+ << " is too large, set hard limit to "
+ <<
PrettyPrinter::print_bytes(query_ctx->adjusted_mem_limit())
+ << " and resume running.";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ if (flushed_memtable_bytes <= 0) {
+ flushed_memtable_bytes =
+ flush_memtable_from_current_group_(wg,
query_it->reserve_size_);
+ }
+ if (flushed_memtable_bytes > 0) {
+ // Flushed some memtable, just wait flush finished and not
do anything more.
+ wg->enable_write_buffer_limit(true);
+ ++query_it;
+ continue;
+ }
+ if (!has_changed_hard_limit) {
+ update_queries_limit_(wg, true);
+ has_changed_hard_limit = true;
+ LOG(INFO) << "Query: " << print_id(query_ctx->query_id())
<< " reserve memory("
+ <<
PrettyPrinter::print_bytes(query_it->reserve_size_)
+ << ") failed due to workload group memory
exceed, "
+ "should set the workload group work in memory
insufficent mode, "
+ "so that other query will reduce their
memory."
+ << " Query mem limit: "
+ <<
PrettyPrinter::print_bytes(query_ctx->get_mem_limit())
+ << " mem usage: "
+ << PrettyPrinter::print_bytes(
+
query_ctx->get_mem_tracker()->consumption())
+ << ", wg: " << wg->debug_string();
+ }
+ if (wg->slot_memory_policy() == TWgSlotMemoryPolicy::NONE) {
+ // If not enable slot memory policy, then should spill
directly
+ // Maybe there are another query that use too much memory,
but we
+ // not encourage not enable slot memory.
+ // TODO should kill the query that exceed limit.
+ bool spill_res = handle_single_query_(query_ctx,
query_it->reserve_size_,
+
query_it->elapsed_time(),
+
query_ctx->paused_reason());
+ if (!spill_res) {
+ ++query_it;
+ continue;
+ } else {
+ VLOG_DEBUG << "Query: " <<
print_id(query_ctx->query_id())
+ << " remove from paused list";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ } else {
+ // Should not put the query back to task scheduler
immediately, because when wg's memory not sufficient,
+ // and then set wg's flag, other query may not free memory
very quickly.
+ if (query_it->elapsed_time() >
config::spill_in_paused_queue_timeout_ms) {
+ // set wg's memory to insufficent, then add it back to
task scheduler to run.
+ LOG(INFO) << "Query: " <<
print_id(query_ctx->query_id())
+ << " will be resume.";
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ } else {
+ ++query_it;
+ continue;
+ }
+ }
+ } else {
+ has_query_exceed_process_memlimit = true;
+ // If wg's memlimit not exceed, but process memory exceed, it
means cache or other metadata
+ // used too much memory. Should clean all cache here.
+ // 1. Check cache used, if cache is larger than > 0, then just
return and wait for it to 0 to release some memory.
+ if
(doris::GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted >
+ 0.05 &&
+
doris::GlobalMemoryArbitrator::last_wg_trigger_cache_capacity_adjust_weighted >
+ 0.05) {
+
doris::GlobalMemoryArbitrator::last_wg_trigger_cache_capacity_adjust_weighted =
+ 0.04;
+
doris::GlobalMemoryArbitrator::notify_cache_adjust_capacity();
+ LOG(INFO) << "There are some queries need process memory,
so that set cache "
+ "capacity "
+ "to 0 now";
+ }
+ // need to check config::disable_memory_gc here, if not, when
config::disable_memory_gc == true,
+ // cache is not adjusted, query_it->cache_ratio_ will always
be 1, and this if branch will nenver
+ // execute, this query will never be resumed, and will
deadlock here
+ if ((!config::disable_memory_gc && query_it->cache_ratio_ <
0.05) ||
+ config::disable_memory_gc) {
+ // 1. Check if could revoke some memory from memtable
+ if (flushed_memtable_bytes <= 0) {
+ flushed_memtable_bytes =
+ flush_memtable_from_current_group_(wg,
query_it->reserve_size_);
+ }
+ if (flushed_memtable_bytes > 0) {
+ // Flushed some memtable, just wait flush finished and
not do anything more.
+ ++query_it;
+ continue;
+ }
+ // TODO should wait here to check if the process has
release revoked_size memory and then continue.
+ if (!has_revoked_from_other_group) {
+ int64_t revoked_size = revoke_memory_from_other_group_(
+ query_ctx, wg->enable_memory_overcommit(),
query_it->reserve_size_);
+ if (revoked_size > 0) {
+ has_revoked_from_other_group = true;
+ query_ctx->set_memory_sufficient(true);
+ VLOG_DEBUG << "Query: " <<
print_id(query_ctx->query_id())
+ << " is resumed after revoke memory
from other group.";
+ query_it = queries_list.erase(query_it);
+ // Do not care if the revoked_size > reserve size,
and try to run again.
+ continue;
+ } else {
+ bool spill_res = handle_single_query_(
+ query_ctx, query_it->reserve_size_,
query_it->elapsed_time(),
+ query_ctx->paused_reason());
+ if (spill_res) {
+ VLOG_DEBUG << "Query: " <<
print_id(query_ctx->query_id())
+ << " remove from paused list";
+ query_it = queries_list.erase(query_it);
+ continue;
+ } else {
+ ++query_it;
+ continue;
+ }
+ }
+ } else {
Review Comment:
实际的行为，至少 cancel 第一个 query 的reserve size，实际可能只cancel一个query
或不足够所有query跑，但依然把所有query唤醒，大不了所有query跑起来后再次进入 paused queries。
--
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: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
