github-actions[bot] commented on code in PR #65542:
URL: https://github.com/apache/doris/pull/65542#discussion_r3575651836
##########
be/src/runtime/workload_group/workload_group.cpp:
##########
@@ -154,38 +154,29 @@ void WorkloadGroup::check_and_update(const
WorkloadGroupInfo& wg_info) {
if (UNLIKELY(wg_info.id != _id)) {
return;
}
- {
- std::shared_lock<std::shared_mutex> rl {_mutex};
- if (LIKELY(wg_info.version <= _version)) {
- return;
- }
- }
- {
- std::lock_guard<std::shared_mutex> wl {_mutex};
- if (wg_info.version > _version) {
- _name = wg_info.name;
- _version = wg_info.version;
- _min_cpu_percent = wg_info.min_cpu_percent;
- _max_cpu_percent = wg_info.max_cpu_percent;
- _memory_limit = wg_info.memory_limit;
- _min_memory_percent = wg_info.min_memory_percent;
- _max_memory_percent = wg_info.max_memory_percent;
- _scan_thread_num = wg_info.scan_thread_num;
- _max_remote_scan_thread_num = wg_info.max_remote_scan_thread_num;
- _min_remote_scan_thread_num = wg_info.min_remote_scan_thread_num;
- _memory_low_watermark = wg_info.memory_low_watermark;
- _memory_high_watermark = wg_info.memory_high_watermark;
- _scan_bytes_per_second = wg_info.read_bytes_per_second;
- _remote_scan_bytes_per_second =
wg_info.remote_read_bytes_per_second;
- _total_query_slot_count = wg_info.total_query_slot_count;
- _slot_mem_policy = wg_info.slot_mem_policy;
- if (_max_memory_percent > 0) {
- _min_memory_limit = static_cast<int64_t>(
- static_cast<double>(_memory_limit *
_min_memory_percent) /
- _max_memory_percent);
- }
- } else {
- return;
+ std::lock_guard<std::shared_mutex> wl {_mutex};
+ // In serverless mode, user may modify cgroup's memory limit directly and
workload group's config
+ // is not changed. So that we should update workload group's config ignore
version.
+ if (wg_info.version > _version || _memory_limit != wg_info.memory_limit) {
Review Comment:
This now lets stale lower-version topic payloads roll the group back. FE
publishes workload-group topics through per-BE workers and only waits for a
timeout; it does not cancel a worker that is still queued or running, so an
older publish can arrive after a newer version was already applied. On the BE
side, `parse_topic_info()` recomputes `wg_info.memory_limit` from the current
local `MemInfo::mem_limit()` and the topic's `max_memory_percent`, so a stale
payload can satisfy `_memory_limit != wg_info.memory_limit` after a
cgroup/process memory refresh or simply because the old version carried a
different memory percent, even though `wg_info.version < _version`. This block
then assigns `_version = wg_info.version` and overwrites the CPU, watermark,
thread, IO, slot, and memory-percent fields from the stale payload; the
listener also continues to call scheduler and IO-throttle upserts with that
same stale info after this method returns. Please keep version ordering
monotonic for the
whole update path, e.g. reject `< _version`, update all fields/side effects
for `> _version`, and for `== _version` refresh only the BE-derived
memory-limit/min-limit values needed for the serverless cgroup case.
--
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]
