be/src/agent/cgroup_cpu_ctl.cpp

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "agent/cgroup_cpu_ctl.h"

#include <fmt/format.h>
#include <sys/stat.h>
#include <unistd.h>

#include <filesystem>

#include "util/cgroup_util.h"
#include "util/defer_op.h"

namespace doris {

#include "common/compile_check_begin.h"

bool CgroupCpuCtl::is_a_valid_cgroup_path(std::string cg_path) {
    if (!cg_path.empty()) {
        if (cg_path.back() != '/') {
            cg_path = cg_path + "/";
        }
        if (_is_enable_cgroup_v2_in_env) {
            std::string query_path_cg_type = cg_path + "cgroup.type";
            std::string query_path_ctl = cg_path + "cgroup.subtree_control";
            std::string query_path_procs = cg_path + "cgroup.procs";
            if (access(query_path_cg_type.c_str(), F_OK) != 0 ||
                access(query_path_ctl.c_str(), F_OK) != 0 ||
                access(query_path_procs.c_str(), F_OK) != 0) {
                LOG(WARNING) << "[cgroup_init_path]invalid cgroup v2 path, access neccessary file "
                                "failed";
            } else {
                return true;
            }
        } else if (_is_enable_cgroup_v1_in_env) {
            std::string query_path_tasks = cg_path + "tasks";
            std::string query_path_cpu_shares = cg_path + "cpu.shares";
            std::string query_path_quota = cg_path + "cpu.cfs_quota_us";
            if (access(query_path_tasks.c_str(), F_OK) != 0 ||
                access(query_path_cpu_shares.c_str(), F_OK) != 0 ||
                access(query_path_quota.c_str(), F_OK) != 0) {
                LOG(WARNING) << "[cgroup_init_path]invalid cgroup v1 path, access neccessary file "
                                "failed";
            } else {
                return true;
            }
        }
    }
    return false;
}

void CgroupCpuCtl::init_doris_cgroup_path() {
    std::string conf_path = config::doris_cgroup_cpu_path;
    if (conf_path.empty()) {
        LOG(INFO) << "[cgroup_init_path]doris cgroup home path is not specify, if you not use "
                     "workload group, you can ignore this log.";
        return;
    }

    if (access(conf_path.c_str(), F_OK) != 0) {
        LOG(INFO) << "[cgroup_init_path]doris cgroup home path not exists, path=" << conf_path;
        return;
    }

    if (conf_path.back() != '/') {
        conf_path = conf_path + "/";
    }

    // check whether current user specified path is a valid cgroup path
    std::string cg_msg = "not set cgroup in env";
    if (CGroupUtil::cgroupsv2_enable()) {
        _is_enable_cgroup_v2_in_env = true;
        cg_msg = "cgroup v2 is enabled in env";
    } else if (CGroupUtil::cgroupsv1_enable()) {
        _is_enable_cgroup_v1_in_env = true;
        cg_msg = "cgroup v1 is enabled in env";
    }
    bool is_cgroup_path_valid = CgroupCpuCtl::is_a_valid_cgroup_path(conf_path);

    std::string tmp_query_path = conf_path + "query";
    if (is_cgroup_path_valid) {
        if (access(tmp_query_path.c_str(), F_OK) != 0) {
            int ret = mkdir(tmp_query_path.c_str(), S_IRWXU);
            if (ret != 0) {
                LOG(ERROR) << "[cgroup_init_path]cgroup mkdir query failed, path="
                           << tmp_query_path;
            }
        }
        _is_cgroup_query_path_valid = CgroupCpuCtl::is_a_valid_cgroup_path(tmp_query_path);
    }

    _doris_cgroup_cpu_path = conf_path;
    _doris_cgroup_cpu_query_path = tmp_query_path;
    std::string query_path_msg = _is_cgroup_query_path_valid ? "cgroup query path is valid"
                                                             : "cgroup query path is not valid";
    _cpu_core_num = CpuInfo::num_cores();

    std::string init_cg_v2_msg = "";
    if (_is_enable_cgroup_v2_in_env && _is_cgroup_query_path_valid) {
        Status ret = init_cgroup_v2_query_path_public_file(_doris_cgroup_cpu_path,
                                                           _doris_cgroup_cpu_query_path);
        if (!ret.ok()) {
            init_cg_v2_msg = " write cgroup v2 file failed, err=" + ret.to_string_no_stack() + ". ";
        } else {
            init_cg_v2_msg = "write cgroup v2 public file succ.";
        }
    }

    LOG(INFO) << "[cgroup_init_path]init cgroup home path finish, home path="
              << _doris_cgroup_cpu_path << ", query path=" << _doris_cgroup_cpu_query_path << ", "
              << cg_msg << ", " << query_path_msg << ", core_num=" << _cpu_core_num << ". "
              << init_cg_v2_msg;
}

Status CgroupCpuCtl::init_cgroup_v2_query_path_public_file(std::string home_path,
                                                           std::string query_path) {
    // 1 enable cpu controller for home path's child
    _doris_cgroup_cpu_path_subtree_ctl_file = home_path + "cgroup.subtree_control";
    if (access(_doris_cgroup_cpu_path_subtree_ctl_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v2 doris home's subtree control file");
    }
    RETURN_IF_ERROR(CgroupCpuCtl::write_cg_sys_file(_doris_cgroup_cpu_path_subtree_ctl_file, "+cpu",
                                                    "set cpu controller", false));

    // 2 enable cpu controller for query path's child
    _cgroup_v2_query_path_subtree_ctl_file = query_path + "/cgroup.subtree_control";
    if (access(_cgroup_v2_query_path_subtree_ctl_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v2 query path's subtree control file");
    }
    RETURN_IF_ERROR(CgroupCpuCtl::write_cg_sys_file(_cgroup_v2_query_path_subtree_ctl_file, "+cpu",
                                                    "set cpu controller", false));

    // 3 write cgroup.procs
    _doris_cg_v2_procs_file = query_path + "/cgroup.procs";
    if (access(_doris_cg_v2_procs_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v2 cgroup.procs file");
    }
    RETURN_IF_ERROR(CgroupCpuCtl::write_cg_sys_file(_doris_cg_v2_procs_file,
                                                    std::to_string(getpid()),
                                                    "set pid to cg v2 procs file", false));
    return Status::OK();
}

uint64_t CgroupCpuCtl::cpu_soft_limit_default_value() {
    return _is_enable_cgroup_v2_in_env ? 100 : 1024;
}

std::shared_ptr<CgroupCpuCtl> CgroupCpuCtl::create_cgroup_cpu_ctl(uint64_t wg_id) {
    if (_is_enable_cgroup_v2_in_env) {
        return std::make_shared<CgroupV2CpuCtl>(wg_id);
    } else if (_is_enable_cgroup_v1_in_env) {
        return std::make_shared<CgroupV1CpuCtl>(wg_id);
    }
    return nullptr;
}

void CgroupCpuCtl::get_cgroup_cpu_info(uint64_t* cpu_shares, int* cpu_hard_limit) {
    std::lock_guard<std::shared_mutex> w_lock(_lock_mutex);
    *cpu_shares = this->_cpu_shares;
    *cpu_hard_limit = this->_cpu_hard_limit;
}

void CgroupCpuCtl::update_cpu_hard_limit(int max_cpu_percent) {
    if (!_init_succ) {
        return;
    }
    if (max_cpu_percent < 0 || max_cpu_percent > 100) {
        LOG(WARNING) << "invalid max_cpu_percent: " << max_cpu_percent;
        return;
    }
    std::lock_guard<std::shared_mutex> w_lock(_lock_mutex);
    if (_cpu_hard_limit != max_cpu_percent) {
        Status ret = modify_cg_cpu_hard_limit_no_lock(max_cpu_percent);
        if (ret.ok()) {
            _cpu_hard_limit = max_cpu_percent;
        } else {
            LOG(WARNING) << "update cpu hard limit failed, cpu hard limit: " << max_cpu_percent
                         << ", error: " << ret;
        }
    }
}

void CgroupCpuCtl::update_cpu_soft_limit(int min_cpu_percent) {
    if (!_init_succ) {
        return;
    }
    if (min_cpu_percent < 0 || min_cpu_percent > 100) {
        LOG(WARNING) << "min_cpu_percent is invalid, min_cpu_percent: " << min_cpu_percent;
        return;
    }
    // cgroup v1: cpu_shares is a value between 2 and 262144
    // cgroup v2: cpu_shares is a value between 1 and 10000
    // so mapping the cpu percent to 64 to 6400
    // if cpu percent == 0, then set it to 2, it is a min value for cgroup v1 and v2
    int cpu_shares = min_cpu_percent == 0 ? 2 : min_cpu_percent * 64;
    std::lock_guard<std::shared_mutex> w_lock(_lock_mutex);
    if (_cpu_shares != cpu_shares) {
        Status ret = modify_cg_cpu_soft_limit_no_lock(cpu_shares);
        if (ret.ok()) {
            _cpu_shares = cpu_shares;
        }
    }
}

Status CgroupCpuCtl::write_cg_sys_file(std::string file_path, std::string value, std::string msg,
                                       bool is_append) {
    int fd = open(file_path.c_str(), is_append ? O_RDWR | O_APPEND : O_RDWR);
    if (fd == -1) {
        LOG(ERROR) << "open path failed, path=" << file_path;
        return Status::InternalError<false>("open path failed, path={}", file_path);
    }

    Defer defer {[&]() {
        if (-1 == ::close(fd)) {
            LOG(INFO) << "close file fd failed";
        }
    }};

    auto str = fmt::format("{}\n", value);
    ssize_t ret = write(fd, str.c_str(), str.size());
    if (ret == -1) {
        LOG(ERROR) << msg << " write sys file failed, file_path=" << file_path;
        return Status::InternalError<false>("{} write sys file failed, file_path={}", msg,
                                            file_path);
    }
    LOG(INFO) << msg << " success, file path: " << file_path;
    return Status::OK();
}

Status CgroupCpuCtl::add_thread_to_cgroup(std::string task_path) {
    if (!_init_succ) {
        return Status::OK();
    }
#if defined(__APPLE__)
    //unsupported now
    return Status::OK();
#else
    int tid = static_cast<int>(syscall(SYS_gettid));
    std::string msg =
            "add thread " + std::to_string(tid) + " to group" + " " + std::to_string(_wg_id);
    std::lock_guard<std::shared_mutex> w_lock(_lock_mutex);
    return CgroupCpuCtl::write_cg_sys_file(task_path, std::to_string(tid), msg, true);
#endif
}

Status CgroupCpuCtl::delete_unused_cgroup_path(std::set<uint64_t>& used_wg_ids) {
    if (!_is_cgroup_query_path_valid) {
        return Status::InternalError<false>("not find a valid cgroup query path");
    }
    // 1 get unused wg id
    std::set<std::string> unused_wg_ids;
    for (const auto& entry : std::filesystem::directory_iterator(_doris_cgroup_cpu_query_path)) {
        const std::string dir_name = entry.path().string();
        struct stat st;
        // == 0 means exists
        if (stat(dir_name.c_str(), &st) == 0 && (st.st_mode & S_IFDIR)) {
            auto pos = dir_name.rfind("/");
            std::string wg_dir_name = dir_name.substr(pos + 1, dir_name.length());
            if (wg_dir_name.empty()) {
                return Status::InternalError<false>("find an empty workload group path, path={}",
                                                    dir_name);
            }
            if (std::all_of(wg_dir_name.begin(), wg_dir_name.end(), ::isdigit)) {
                uint64_t id_in_path = std::stoll(wg_dir_name);
                if (used_wg_ids.find(id_in_path) == used_wg_ids.end()) {
                    unused_wg_ids.insert(wg_dir_name);
                }
            }
        }
    }

    // 2 delete unused cgroup path
    int failed_count = 0;
    std::string query_path = _doris_cgroup_cpu_query_path.back() != '/'
                                     ? _doris_cgroup_cpu_query_path + "/"
                                     : _doris_cgroup_cpu_query_path;
    for (const std::string& unused_wg_id : unused_wg_ids) {
        std::string wg_path = query_path + unused_wg_id;
        int ret = rmdir(wg_path.c_str());
        if (ret < 0) {
            LOG(WARNING) << "remove cgroup path failed, path=" << wg_path << ", error=" << ret;
            failed_count++;
        }
    }
    if (failed_count != 0) {
        return Status::InternalError<false>("error happens when delete unused path, count={}",
                                            failed_count);
    }
    return Status::OK();
}

Status CgroupV1CpuCtl::init() {
    if (!_is_cgroup_query_path_valid) {
        return Status::InternalError<false>("cgroup query path is not valid");
    }

    if (_wg_id <= 0) {
        return Status::InternalError<false>("find an invalid wg_id {}", _wg_id);
    }

    // workload group path
    _cgroup_v1_cpu_tg_path = _doris_cgroup_cpu_query_path + "/" + std::to_string(_wg_id);
    if (access(_cgroup_v1_cpu_tg_path.c_str(), F_OK) != 0) {
        int ret = mkdir(_cgroup_v1_cpu_tg_path.c_str(), S_IRWXU);
        if (ret != 0) {
            LOG(WARNING) << "cgroup v1 make workload group dir failed, path="
                         << _cgroup_v1_cpu_tg_path << ", error=" << ret;
            return Status::InternalError<false>("cgroup v1 mkdir workload group failed, path={}",
                                                _cgroup_v1_cpu_tg_path);
        }
    }

    _cgroup_v1_cpu_tg_quota_file = _cgroup_v1_cpu_tg_path + "/cpu.cfs_quota_us";
    if (access(_cgroup_v1_cpu_tg_quota_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v1 cpu.cfs_quota_us file");
    }
    _cgroup_v1_cpu_tg_shares_file = _cgroup_v1_cpu_tg_path + "/cpu.shares";
    if (access(_cgroup_v1_cpu_tg_shares_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v1 cpu.shares file");
    }
    _cgroup_v1_cpu_tg_task_file = _cgroup_v1_cpu_tg_path + "/tasks";
    if (access(_cgroup_v1_cpu_tg_task_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v1 cpu.shares file");
    }
    LOG(INFO) << "cgroup v1 cpu path init success"
              << ", query tg path=" << _cgroup_v1_cpu_tg_path
              << ", query wg quota file path=" << _cgroup_v1_cpu_tg_quota_file
              << ", query wg share file path=" << _cgroup_v1_cpu_tg_shares_file
              << ", query wg tasks file path=" << _cgroup_v1_cpu_tg_task_file
              << ", core num=" << _cpu_core_num;
    _init_succ = true;
    return Status::OK();
}

Status CgroupV1CpuCtl::modify_cg_cpu_soft_limit_no_lock(int cpu_shares) {
    std::string cpu_share_str = std::to_string(cpu_shares);
    std::string msg = "modify cpu shares to " + cpu_share_str;
    return CgroupCpuCtl::write_cg_sys_file(_cgroup_v1_cpu_tg_shares_file, cpu_share_str, msg,
                                           false);
}

Status CgroupV1CpuCtl::modify_cg_cpu_hard_limit_no_lock(int cpu_hard_limit) {
    // If cpu_hard_limit == 0, we do not know the actual behavior of CGroup
    // just set it to 1% of the cpu core.
    uint64_t val = _cpu_cfs_period_us / 100;
    if (cpu_hard_limit > 0) {
        val = _cpu_cfs_period_us * _cpu_core_num * cpu_hard_limit / 100;
    }
    std::string str_val = std::to_string(val);
    std::string msg = "modify cpu quota value to " + str_val;
    return CgroupCpuCtl::write_cg_sys_file(_cgroup_v1_cpu_tg_quota_file, str_val, msg, false);
}

Status CgroupV1CpuCtl::add_thread_to_cgroup() {
    return CgroupCpuCtl::add_thread_to_cgroup(_cgroup_v1_cpu_tg_task_file);
}

Status CgroupV2CpuCtl::init() {
    if (!_is_cgroup_query_path_valid) {
        return Status::InternalError<false>(" cgroup query path is empty");
    }

    if (_wg_id <= 0) {
        return Status::InternalError<false>("find an invalid wg_id {}", _wg_id);
    }

    // wg path
    _cgroup_v2_query_wg_path = _doris_cgroup_cpu_query_path + "/" + std::to_string(_wg_id);
    if (access(_cgroup_v2_query_wg_path.c_str(), F_OK) != 0) {
        int ret = mkdir(_cgroup_v2_query_wg_path.c_str(), S_IRWXU);
        if (ret != 0) {
            LOG(WARNING) << "cgroup v2 make workload group dir failed, path="
                         << _cgroup_v2_query_wg_path << ", error=" << ret;
            return Status::InternalError<false>("cgroup v2 mkdir wg failed, path={}",
                                                _cgroup_v2_query_wg_path);
        }
    }

    _cgroup_v2_query_wg_cpu_max_file = _cgroup_v2_query_wg_path + "/cpu.max";
    if (access(_cgroup_v2_query_wg_cpu_max_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v2 wg cpu.max file");
    }

    _cgroup_v2_query_wg_cpu_weight_file = _cgroup_v2_query_wg_path + "/cpu.weight";
    if (access(_cgroup_v2_query_wg_cpu_weight_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v2 wg cpu.weight file");
    }

    _cgroup_v2_query_wg_thread_file = _cgroup_v2_query_wg_path + "/cgroup.threads";
    if (access(_cgroup_v2_query_wg_thread_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v2 wg cgroup.threads file");
    }

    _cgroup_v2_query_wg_type_file = _cgroup_v2_query_wg_path + "/cgroup.type";
    if (access(_cgroup_v2_query_wg_type_file.c_str(), F_OK) != 0) {
        return Status::InternalError<false>("not find cgroup v2 wg cgroup.type file");
    }
    RETURN_IF_ERROR(CgroupCpuCtl::write_cg_sys_file(_cgroup_v2_query_wg_type_file, "threaded",
                                                    "set cgroup type", false));

    LOG(INFO) << "cgroup v2 cpu path init success"
              << ", query wg path=" << _cgroup_v2_query_wg_path
              << ", cpu.max file = " << _cgroup_v2_query_wg_cpu_max_file
              << ", cgroup.threads file = " << _cgroup_v2_query_wg_thread_file
              << ", core num=" << _cpu_core_num;
    _init_succ = true;
    return Status::OK();
}

Status CgroupV2CpuCtl::modify_cg_cpu_hard_limit_no_lock(int cpu_hard_limit) {
    // If cpu_hard_limit is 0, we set the cpu.max to 1000 100000.
    // Means 1% of one cpu core.
    uint64_t int_val = _cpu_cfs_period_us / 100;
    if (cpu_hard_limit > 0) {
        int_val = _cpu_cfs_period_us * _cpu_core_num * cpu_hard_limit / 100;
    }
    std::string value = fmt::format("{} {}", int_val, _cpu_cfs_period_us);
    std::string msg = fmt::format("modify cpu.max to [{}]", value);
    return CgroupCpuCtl::write_cg_sys_file(_cgroup_v2_query_wg_cpu_max_file, value, msg, false);
}

Status CgroupV2CpuCtl::modify_cg_cpu_soft_limit_no_lock(int cpu_weight) {
    std::string cpu_weight_str = std::to_string(cpu_weight);
    std::string msg = "modify cpu.weight to " + cpu_weight_str;
    return CgroupCpuCtl::write_cg_sys_file(_cgroup_v2_query_wg_cpu_weight_file, cpu_weight_str, msg,
                                           false);
}

Status CgroupV2CpuCtl::add_thread_to_cgroup() {
    return CgroupCpuCtl::add_thread_to_cgroup(_cgroup_v2_query_wg_thread_file);
}

#include "common/compile_check_end.h"

} // namespace doris

Coverage Report

Created: 2026-03-15 22:14

Line	Count	Source
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		#include "agent/cgroup_cpu_ctl.h"
19
20		#include <fmt/format.h>
21		#include <sys/stat.h>
22		#include <unistd.h>
23
24		#include <filesystem>
25
26		#include "util/cgroup_util.h"
27		#include "util/defer_op.h"
28
29		namespace doris {
30
31		#include "common/compile_check_begin.h"
32
33	0	bool CgroupCpuCtl::is_a_valid_cgroup_path(std::string cg_path) {
34	0	if (!cg_path.empty()) {
35	0	if (cg_path.back() != '/') {
36	0	cg_path = cg_path + "/";
37	0	}
38	0	if (_is_enable_cgroup_v2_in_env) {
39	0	std::string query_path_cg_type = cg_path + "cgroup.type";
40	0	std::string query_path_ctl = cg_path + "cgroup.subtree_control";
41	0	std::string query_path_procs = cg_path + "cgroup.procs";
42	0	if (access(query_path_cg_type.c_str(), F_OK) != 0 \|\|
43	0	access(query_path_ctl.c_str(), F_OK) != 0 \|\|
44	0	access(query_path_procs.c_str(), F_OK) != 0) {
45	0	LOG(WARNING) << "[cgroup_init_path]invalid cgroup v2 path, access neccessary file "
46	0	"failed";
47	0	} else {
48	0	return true;
49	0	}
50	0	} else if (_is_enable_cgroup_v1_in_env) {
51	0	std::string query_path_tasks = cg_path + "tasks";
52	0	std::string query_path_cpu_shares = cg_path + "cpu.shares";
53	0	std::string query_path_quota = cg_path + "cpu.cfs_quota_us";
54	0	if (access(query_path_tasks.c_str(), F_OK) != 0 \|\|
55	0	access(query_path_cpu_shares.c_str(), F_OK) != 0 \|\|
56	0	access(query_path_quota.c_str(), F_OK) != 0) {
57	0	LOG(WARNING) << "[cgroup_init_path]invalid cgroup v1 path, access neccessary file "
58	0	"failed";
59	0	} else {
60	0	return true;
61	0	}
62	0	}
63	0	}
64	0	return false;
65	0	}
66
67	0	void CgroupCpuCtl::init_doris_cgroup_path() {
68	0	std::string conf_path = config::doris_cgroup_cpu_path;
69	0	if (conf_path.empty()) {
70	0	LOG(INFO) << "[cgroup_init_path]doris cgroup home path is not specify, if you not use "
71	0	"workload group, you can ignore this log.";
72	0	return;
73	0	}
74
75	0	if (access(conf_path.c_str(), F_OK) != 0) {
76	0	LOG(INFO) << "[cgroup_init_path]doris cgroup home path not exists, path=" << conf_path;
77	0	return;
78	0	}
79
80	0	if (conf_path.back() != '/') {
81	0	conf_path = conf_path + "/";
82	0	}
83
84		// check whether current user specified path is a valid cgroup path
85	0	std::string cg_msg = "not set cgroup in env";
86	0	if (CGroupUtil::cgroupsv2_enable()) {
87	0	_is_enable_cgroup_v2_in_env = true;
88	0	cg_msg = "cgroup v2 is enabled in env";
89	0	} else if (CGroupUtil::cgroupsv1_enable()) {
90	0	_is_enable_cgroup_v1_in_env = true;
91	0	cg_msg = "cgroup v1 is enabled in env";
92	0	}
93	0	bool is_cgroup_path_valid = CgroupCpuCtl::is_a_valid_cgroup_path(conf_path);
94
95	0	std::string tmp_query_path = conf_path + "query";
96	0	if (is_cgroup_path_valid) {
97	0	if (access(tmp_query_path.c_str(), F_OK) != 0) {
98	0	int ret = mkdir(tmp_query_path.c_str(), S_IRWXU);
99	0	if (ret != 0) {
100	0	LOG(ERROR) << "[cgroup_init_path]cgroup mkdir query failed, path="
101	0	<< tmp_query_path;
102	0	}
103	0	}
104	0	_is_cgroup_query_path_valid = CgroupCpuCtl::is_a_valid_cgroup_path(tmp_query_path);
105	0	}
106
107	0	_doris_cgroup_cpu_path = conf_path;
108	0	_doris_cgroup_cpu_query_path = tmp_query_path;
109	0	std::string query_path_msg = _is_cgroup_query_path_valid ? "cgroup query path is valid"
110	0	: "cgroup query path is not valid";
111	0	_cpu_core_num = CpuInfo::num_cores();
112
113	0	std::string init_cg_v2_msg = "";
114	0	if (_is_enable_cgroup_v2_in_env && _is_cgroup_query_path_valid) {
115	0	Status ret = init_cgroup_v2_query_path_public_file(_doris_cgroup_cpu_path,
116	0	_doris_cgroup_cpu_query_path);
117	0	if (!ret.ok()) {
118	0	init_cg_v2_msg = " write cgroup v2 file failed, err=" + ret.to_string_no_stack() + ". ";
119	0	} else {
120	0	init_cg_v2_msg = "write cgroup v2 public file succ.";
121	0	}
122	0	}
123
124	0	LOG(INFO) << "[cgroup_init_path]init cgroup home path finish, home path="
125	0	<< _doris_cgroup_cpu_path << ", query path=" << _doris_cgroup_cpu_query_path << ", "
126	0	<< cg_msg << ", " << query_path_msg << ", core_num=" << _cpu_core_num << ". "
127	0	<< init_cg_v2_msg;
128	0	}
129
130		Status CgroupCpuCtl::init_cgroup_v2_query_path_public_file(std::string home_path,
131	0	std::string query_path) {
132		// 1 enable cpu controller for home path's child
133	0	_doris_cgroup_cpu_path_subtree_ctl_file = home_path + "cgroup.subtree_control";
134	0	if (access(_doris_cgroup_cpu_path_subtree_ctl_file.c_str(), F_OK) != 0) {
135	0	return Status::InternalError<false>("not find cgroup v2 doris home's subtree control file");
136	0	}
137	0	RETURN_IF_ERROR(CgroupCpuCtl::write_cg_sys_file(_doris_cgroup_cpu_path_subtree_ctl_file, "+cpu",
138	0	"set cpu controller", false));
139
140		// 2 enable cpu controller for query path's child
141	0	_cgroup_v2_query_path_subtree_ctl_file = query_path + "/cgroup.subtree_control";
142	0	if (access(_cgroup_v2_query_path_subtree_ctl_file.c_str(), F_OK) != 0) {
143	0	return Status::InternalError<false>("not find cgroup v2 query path's subtree control file");
144	0	}
145	0	RETURN_IF_ERROR(CgroupCpuCtl::write_cg_sys_file(_cgroup_v2_query_path_subtree_ctl_file, "+cpu",
146	0	"set cpu controller", false));
147
148		// 3 write cgroup.procs
149	0	_doris_cg_v2_procs_file = query_path + "/cgroup.procs";
150	0	if (access(_doris_cg_v2_procs_file.c_str(), F_OK) != 0) {
151	0	return Status::InternalError<false>("not find cgroup v2 cgroup.procs file");
152	0	}
153	0	RETURN_IF_ERROR(CgroupCpuCtl::write_cg_sys_file(_doris_cg_v2_procs_file,
154	0	std::to_string(getpid()),
155	0	"set pid to cg v2 procs file", false));
156	0	return Status::OK();
157	0	}
158
159	0	uint64_t CgroupCpuCtl::cpu_soft_limit_default_value() {
160	0	return _is_enable_cgroup_v2_in_env ? 100 : 1024;
161	0	}
162
163	0	std::shared_ptr<CgroupCpuCtl> CgroupCpuCtl::create_cgroup_cpu_ctl(uint64_t wg_id) {
164	0	if (_is_enable_cgroup_v2_in_env) {
165	0	return std::make_shared<CgroupV2CpuCtl>(wg_id);
166	0	} else if (_is_enable_cgroup_v1_in_env) {
167	0	return std::make_shared<CgroupV1CpuCtl>(wg_id);
168	0	}
169	0	return nullptr;
170	0	}
171
172	0	void CgroupCpuCtl::get_cgroup_cpu_info(uint64_t* cpu_shares, int* cpu_hard_limit) {
173	0	std::lock_guard<std::shared_mutex> w_lock(_lock_mutex);
174	0	*cpu_shares = this->_cpu_shares;
175	0	*cpu_hard_limit = this->_cpu_hard_limit;
176	0	}
177
178	0	void CgroupCpuCtl::update_cpu_hard_limit(int max_cpu_percent) {
179	0	if (!_init_succ) {
180	0	return;
181	0	}
182	0	if (max_cpu_percent < 0 \|\| max_cpu_percent > 100) {
183	0	LOG(WARNING) << "invalid max_cpu_percent: " << max_cpu_percent;
184	0	return;
185	0	}
186	0	std::lock_guard<std::shared_mutex> w_lock(_lock_mutex);
187	0	if (_cpu_hard_limit != max_cpu_percent) {
188	0	Status ret = modify_cg_cpu_hard_limit_no_lock(max_cpu_percent);
189	0	if (ret.ok()) {
190	0	_cpu_hard_limit = max_cpu_percent;
191	0	} else {
192	0	LOG(WARNING) << "update cpu hard limit failed, cpu hard limit: " << max_cpu_percent
193	0	<< ", error: " << ret;
194	0	}
195	0	}
196	0	}
197
198	0	void CgroupCpuCtl::update_cpu_soft_limit(int min_cpu_percent) {
199	0	if (!_init_succ) {
200	0	return;
201	0	}
202	0	if (min_cpu_percent < 0 \|\| min_cpu_percent > 100) {
203	0	LOG(WARNING) << "min_cpu_percent is invalid, min_cpu_percent: " << min_cpu_percent;
204	0	return;
205	0	}
206		// cgroup v1: cpu_shares is a value between 2 and 262144
207		// cgroup v2: cpu_shares is a value between 1 and 10000
208		// so mapping the cpu percent to 64 to 6400
209		// if cpu percent == 0, then set it to 2, it is a min value for cgroup v1 and v2
210	0	int cpu_shares = min_cpu_percent == 0 ? 2 : min_cpu_percent * 64;
211	0	std::lock_guard<std::shared_mutex> w_lock(_lock_mutex);
212	0	if (_cpu_shares != cpu_shares) {
213	0	Status ret = modify_cg_cpu_soft_limit_no_lock(cpu_shares);
214	0	if (ret.ok()) {
215	0	_cpu_shares = cpu_shares;
216	0	}
217	0	}
218	0	}
219
220		Status CgroupCpuCtl::write_cg_sys_file(std::string file_path, std::string value, std::string msg,
221	0	bool is_append) {
222	0	int fd = open(file_path.c_str(), is_append ? O_RDWR \| O_APPEND : O_RDWR);
223	0	if (fd == -1) {
224	0	LOG(ERROR) << "open path failed, path=" << file_path;
225	0	return Status::InternalError<false>("open path failed, path={}", file_path);
226	0	}
227
228	0	Defer defer {[&]() {
229	0	if (-1 == ::close(fd)) {
230	0	LOG(INFO) << "close file fd failed";
231	0	}
232	0	}};
233
234	0	auto str = fmt::format("{}\n", value);
235	0	ssize_t ret = write(fd, str.c_str(), str.size());
236	0	if (ret == -1) {
237	0	LOG(ERROR) << msg << " write sys file failed, file_path=" << file_path;
238	0	return Status::InternalError<false>("{} write sys file failed, file_path={}", msg,
239	0	file_path);
240	0	}
241	0	LOG(INFO) << msg << " success, file path: " << file_path;
242	0	return Status::OK();
243	0	}
244
245	0	Status CgroupCpuCtl::add_thread_to_cgroup(std::string task_path) {
246	0	if (!_init_succ) {
247	0	return Status::OK();
248	0	}
249		#if defined(__APPLE__)
250		//unsupported now
251		return Status::OK();
252		#else
253	0	int tid = static_cast<int>(syscall(SYS_gettid));
254	0	std::string msg =
255	0	"add thread " + std::to_string(tid) + " to group" + " " + std::to_string(_wg_id);
256	0	std::lock_guard<std::shared_mutex> w_lock(_lock_mutex);
257	0	return CgroupCpuCtl::write_cg_sys_file(task_path, std::to_string(tid), msg, true);
258	0	#endif
259	0	}
260
261	0	Status CgroupCpuCtl::delete_unused_cgroup_path(std::set<uint64_t>& used_wg_ids) {
262	0	if (!_is_cgroup_query_path_valid) {
263	0	return Status::InternalError<false>("not find a valid cgroup query path");
264	0	}
265		// 1 get unused wg id
266	0	std::set<std::string> unused_wg_ids;
267	0	for (const auto& entry : std::filesystem::directory_iterator(_doris_cgroup_cpu_query_path)) {
268	0	const std::string dir_name = entry.path().string();
269	0	struct stat st;
270		// == 0 means exists
271	0	if (stat(dir_name.c_str(), &st) == 0 && (st.st_mode & S_IFDIR)) {
272	0	auto pos = dir_name.rfind("/");
273	0	std::string wg_dir_name = dir_name.substr(pos + 1, dir_name.length());
274	0	if (wg_dir_name.empty()) {
275	0	return Status::InternalError<false>("find an empty workload group path, path={}",
276	0	dir_name);
277	0	}
278	0	if (std::all_of(wg_dir_name.begin(), wg_dir_name.end(), ::isdigit)) {
279	0	uint64_t id_in_path = std::stoll(wg_dir_name);
280	0	if (used_wg_ids.find(id_in_path) == used_wg_ids.end()) {
281	0	unused_wg_ids.insert(wg_dir_name);
282	0	}
283	0	}
284	0	}
285	0	}
286
287		// 2 delete unused cgroup path
288	0	int failed_count = 0;
289	0	std::string query_path = _doris_cgroup_cpu_query_path.back() != '/'
290	0	? _doris_cgroup_cpu_query_path + "/"
291	0	: _doris_cgroup_cpu_query_path;
292	0	for (const std::string& unused_wg_id : unused_wg_ids) {
293	0	std::string wg_path = query_path + unused_wg_id;
294	0	int ret = rmdir(wg_path.c_str());
295	0	if (ret < 0) {
296	0	LOG(WARNING) << "remove cgroup path failed, path=" << wg_path << ", error=" << ret;
297	0	failed_count++;
298	0	}
299	0	}
300	0	if (failed_count != 0) {
301	0	return Status::InternalError<false>("error happens when delete unused path, count={}",
302	0	failed_count);
303	0	}
304	0	return Status::OK();
305	0	}
306
307	0	Status CgroupV1CpuCtl::init() {
308	0	if (!_is_cgroup_query_path_valid) {
309	0	return Status::InternalError<false>("cgroup query path is not valid");
310	0	}
311
312	0	if (_wg_id <= 0) {
313	0	return Status::InternalError<false>("find an invalid wg_id {}", _wg_id);
314	0	}
315
316		// workload group path
317	0	_cgroup_v1_cpu_tg_path = _doris_cgroup_cpu_query_path + "/" + std::to_string(_wg_id);
318	0	if (access(_cgroup_v1_cpu_tg_path.c_str(), F_OK) != 0) {
319	0	int ret = mkdir(_cgroup_v1_cpu_tg_path.c_str(), S_IRWXU);
320	0	if (ret != 0) {
321	0	LOG(WARNING) << "cgroup v1 make workload group dir failed, path="
322	0	<< _cgroup_v1_cpu_tg_path << ", error=" << ret;
323	0	return Status::InternalError<false>("cgroup v1 mkdir workload group failed, path={}",
324	0	_cgroup_v1_cpu_tg_path);
325	0	}
326	0	}
327
328	0	_cgroup_v1_cpu_tg_quota_file = _cgroup_v1_cpu_tg_path + "/cpu.cfs_quota_us";
329	0	if (access(_cgroup_v1_cpu_tg_quota_file.c_str(), F_OK) != 0) {
330	0	return Status::InternalError<false>("not find cgroup v1 cpu.cfs_quota_us file");
331	0	}
332	0	_cgroup_v1_cpu_tg_shares_file = _cgroup_v1_cpu_tg_path + "/cpu.shares";
333	0	if (access(_cgroup_v1_cpu_tg_shares_file.c_str(), F_OK) != 0) {
334	0	return Status::InternalError<false>("not find cgroup v1 cpu.shares file");
335	0	}
336	0	_cgroup_v1_cpu_tg_task_file = _cgroup_v1_cpu_tg_path + "/tasks";
337	0	if (access(_cgroup_v1_cpu_tg_task_file.c_str(), F_OK) != 0) {
338	0	return Status::InternalError<false>("not find cgroup v1 cpu.shares file");
339	0	}
340	0	LOG(INFO) << "cgroup v1 cpu path init success"
341	0	<< ", query tg path=" << _cgroup_v1_cpu_tg_path
342	0	<< ", query wg quota file path=" << _cgroup_v1_cpu_tg_quota_file
343	0	<< ", query wg share file path=" << _cgroup_v1_cpu_tg_shares_file
344	0	<< ", query wg tasks file path=" << _cgroup_v1_cpu_tg_task_file
345	0	<< ", core num=" << _cpu_core_num;
346	0	_init_succ = true;
347	0	return Status::OK();
348	0	}
349
350	0	Status CgroupV1CpuCtl::modify_cg_cpu_soft_limit_no_lock(int cpu_shares) {
351	0	std::string cpu_share_str = std::to_string(cpu_shares);
352	0	std::string msg = "modify cpu shares to " + cpu_share_str;
353	0	return CgroupCpuCtl::write_cg_sys_file(_cgroup_v1_cpu_tg_shares_file, cpu_share_str, msg,
354	0	false);
355	0	}
356
357	0	Status CgroupV1CpuCtl::modify_cg_cpu_hard_limit_no_lock(int cpu_hard_limit) {
358		// If cpu_hard_limit == 0, we do not know the actual behavior of CGroup
359		// just set it to 1% of the cpu core.
360	0	uint64_t val = _cpu_cfs_period_us / 100;
361	0	if (cpu_hard_limit > 0) {
362	0	val = _cpu_cfs_period_us * _cpu_core_num * cpu_hard_limit / 100;
363	0	}
364	0	std::string str_val = std::to_string(val);
365	0	std::string msg = "modify cpu quota value to " + str_val;
366	0	return CgroupCpuCtl::write_cg_sys_file(_cgroup_v1_cpu_tg_quota_file, str_val, msg, false);
367	0	}
368
369	0	Status CgroupV1CpuCtl::add_thread_to_cgroup() {
370	0	return CgroupCpuCtl::add_thread_to_cgroup(_cgroup_v1_cpu_tg_task_file);
371	0	}
372
373	0	Status CgroupV2CpuCtl::init() {
374	0	if (!_is_cgroup_query_path_valid) {
375	0	return Status::InternalError<false>(" cgroup query path is empty");
376	0	}
377
378	0	if (_wg_id <= 0) {
379	0	return Status::InternalError<false>("find an invalid wg_id {}", _wg_id);
380	0	}
381
382		// wg path
383	0	_cgroup_v2_query_wg_path = _doris_cgroup_cpu_query_path + "/" + std::to_string(_wg_id);
384	0	if (access(_cgroup_v2_query_wg_path.c_str(), F_OK) != 0) {
385	0	int ret = mkdir(_cgroup_v2_query_wg_path.c_str(), S_IRWXU);
386	0	if (ret != 0) {
387	0	LOG(WARNING) << "cgroup v2 make workload group dir failed, path="
388	0	<< _cgroup_v2_query_wg_path << ", error=" << ret;
389	0	return Status::InternalError<false>("cgroup v2 mkdir wg failed, path={}",
390	0	_cgroup_v2_query_wg_path);
391	0	}
392	0	}
393
394	0	_cgroup_v2_query_wg_cpu_max_file = _cgroup_v2_query_wg_path + "/cpu.max";
395	0	if (access(_cgroup_v2_query_wg_cpu_max_file.c_str(), F_OK) != 0) {
396	0	return Status::InternalError<false>("not find cgroup v2 wg cpu.max file");
397	0	}
398
399	0	_cgroup_v2_query_wg_cpu_weight_file = _cgroup_v2_query_wg_path + "/cpu.weight";
400	0	if (access(_cgroup_v2_query_wg_cpu_weight_file.c_str(), F_OK) != 0) {
401	0	return Status::InternalError<false>("not find cgroup v2 wg cpu.weight file");
402	0	}
403
404	0	_cgroup_v2_query_wg_thread_file = _cgroup_v2_query_wg_path + "/cgroup.threads";
405	0	if (access(_cgroup_v2_query_wg_thread_file.c_str(), F_OK) != 0) {
406	0	return Status::InternalError<false>("not find cgroup v2 wg cgroup.threads file");
407	0	}
408
409	0	_cgroup_v2_query_wg_type_file = _cgroup_v2_query_wg_path + "/cgroup.type";
410	0	if (access(_cgroup_v2_query_wg_type_file.c_str(), F_OK) != 0) {
411	0	return Status::InternalError<false>("not find cgroup v2 wg cgroup.type file");
412	0	}
413	0	RETURN_IF_ERROR(CgroupCpuCtl::write_cg_sys_file(_cgroup_v2_query_wg_type_file, "threaded",
414	0	"set cgroup type", false));
415
416	0	LOG(INFO) << "cgroup v2 cpu path init success"
417	0	<< ", query wg path=" << _cgroup_v2_query_wg_path
418	0	<< ", cpu.max file = " << _cgroup_v2_query_wg_cpu_max_file
419	0	<< ", cgroup.threads file = " << _cgroup_v2_query_wg_thread_file
420	0	<< ", core num=" << _cpu_core_num;
421	0	_init_succ = true;
422	0	return Status::OK();
423	0	}
424
425	0	Status CgroupV2CpuCtl::modify_cg_cpu_hard_limit_no_lock(int cpu_hard_limit) {
426		// If cpu_hard_limit is 0, we set the cpu.max to 1000 100000.
427		// Means 1% of one cpu core.
428	0	uint64_t int_val = _cpu_cfs_period_us / 100;
429	0	if (cpu_hard_limit > 0) {
430	0	int_val = _cpu_cfs_period_us * _cpu_core_num * cpu_hard_limit / 100;
431	0	}
432	0	std::string value = fmt::format("{} {}", int_val, _cpu_cfs_period_us);
433	0	std::string msg = fmt::format("modify cpu.max to [{}]", value);
434	0	return CgroupCpuCtl::write_cg_sys_file(_cgroup_v2_query_wg_cpu_max_file, value, msg, false);
435	0	}
436
437	0	Status CgroupV2CpuCtl::modify_cg_cpu_soft_limit_no_lock(int cpu_weight) {
438	0	std::string cpu_weight_str = std::to_string(cpu_weight);
439	0	std::string msg = "modify cpu.weight to " + cpu_weight_str;
440	0	return CgroupCpuCtl::write_cg_sys_file(_cgroup_v2_query_wg_cpu_weight_file, cpu_weight_str, msg,
441	0	false);
442	0	}
443
444	0	Status CgroupV2CpuCtl::add_thread_to_cgroup() {
445	0	return CgroupCpuCtl::add_thread_to_cgroup(_cgroup_v2_query_wg_thread_file);
446	0	}
447
448		#include "common/compile_check_end.h"
449
450		} // namespace doris