// 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 "common/daemon.h"

// IWYU pragma: no_include <bthread/errno.h>

#include <errno.h> // IWYU pragma: keep

#include <gflags/gflags.h>

#include "runtime/memory/jemalloc_control.h"

#if !defined(__SANITIZE_ADDRESS__) && !defined(ADDRESS_SANITIZER) && !defined(LEAK_SANITIZER) && \

        !defined(THREAD_SANITIZER) && !defined(USE_JEMALLOC)

#include <gperftools/malloc_extension.h> // IWYU pragma: keep

#endif

// IWYU pragma: no_include <bits/std_abs.h>

#include <butil/iobuf.h>

#include <math.h>

#include <stdint.h>

#include <stdlib.h>

// IWYU pragma: no_include <bits/chrono.h>

#include <chrono> // IWYU pragma: keep

#include <map>

#include <ostream>

#include <string>

#include "cloud/config.h"

#include "common/config.h"

#include "common/logging.h"

#include "common/metrics/doris_metrics.h"

#include "common/metrics/metrics.h"

#include "common/metrics/system_metrics.h"

#include "common/status.h"

#include "load/memtable/memtable_memory_limiter.h"

#include "runtime/be_proc_monitor.h"

#include "runtime/exec_env.h"

#include "runtime/fragment_mgr.h"

#include "runtime/memory/global_memory_arbitrator.h"

#include "runtime/memory/mem_tracker_limiter.h"

#include "runtime/memory/memory_reclamation.h"

#include "runtime/process_profile.h"

#include "runtime/runtime_query_statistics_mgr.h"

#include "runtime/workload_group/workload_group_manager.h"

#include "storage/storage_engine.h"

#include "storage/tablet/tablet_manager.h"

#include "util/algorithm_util.h"

#include "util/mem_info.h"

#include "util/perf_counters.h"

#include "util/s3_rate_limiter_manager.h"

#include "util/time.h"

namespace doris {

namespace {

std::atomic<int64_t> last_print_proc_mem = 0;

std::atomic<int32_t> refresh_cache_capacity_sleep_time_ms = 0;

std::atomic<int32_t> memory_gc_sleep_time = 0;

#ifdef USE_JEMALLOC

std::atomic<int32_t> je_reset_dirty_decay_sleep_time_ms = 0;

#endif

void update_rowsets_and_segments_num_metrics() {

    if (config::is_cloud_mode()) {

        // TODO(plat1ko): CloudStorageEngine

    } else {

        StorageEngine& engine = ExecEnv::GetInstance()->storage_engine().to_local();

        auto* metrics = DorisMetrics::instance();

        metrics->all_rowsets_num->set_value(engine.tablet_manager()->get_rowset_nums());

        metrics->all_segments_num->set_value(engine.tablet_manager()->get_segment_nums());

    }

}

} // namespace

void Daemon::tcmalloc_gc_thread() {

    // TODO All cache GC wish to be supported

#if !defined(ADDRESS_SANITIZER) && !defined(LEAK_SANITIZER) && !defined(THREAD_SANITIZER) && \

        !defined(USE_JEMALLOC)

    // Limit size of tcmalloc cache via release_rate and max_cache_percent.

    // We adjust release_rate according to memory_pressure, which is usage percent of memory.

    int64_t max_cache_percent = 60;

    double release_rates[10] = {1.0, 1.0, 1.0, 5.0, 5.0, 20.0, 50.0, 100.0, 500.0, 2000.0};

    int64_t pressure_limit = 90;

    bool is_performance_mode = false;

    int64_t physical_limit_bytes =

            std::min(MemInfo::physical_mem() - MemInfo::sys_mem_available_low_water_mark(),

                     MemInfo::mem_limit());

    if (config::memory_mode == std::string("performance")) {

        max_cache_percent = 100;

        pressure_limit = 90;

        is_performance_mode = true;

        physical_limit_bytes = std::min(MemInfo::mem_limit(), MemInfo::physical_mem());

    } else if (config::memory_mode == std::string("compact")) {

        max_cache_percent = 20;

        pressure_limit = 80;

    }

    int last_ms = 0;

    const int kMaxLastMs = 30000;

    const int kIntervalMs = 10;

    size_t init_aggressive_decommit = 0;

    size_t current_aggressive_decommit = 0;

    size_t expected_aggressive_decommit = 0;

    int64_t last_memory_pressure = 0;

    MallocExtension::instance()->GetNumericProperty("tcmalloc.aggressive_memory_decommit",

                                                    &init_aggressive_decommit);

    current_aggressive_decommit = init_aggressive_decommit;

    while (!_stop_background_threads_latch.wait_for(std::chrono::milliseconds(kIntervalMs))) {

        size_t tc_used_bytes = 0;

        size_t tc_alloc_bytes = 0;

        size_t rss = PerfCounters::get_vm_rss();

        MallocExtension::instance()->GetNumericProperty("generic.total_physical_bytes",

                                                        &tc_alloc_bytes);

        MallocExtension::instance()->GetNumericProperty("generic.current_allocated_bytes",

                                                        &tc_used_bytes);

        int64_t tc_cached_bytes = (int64_t)tc_alloc_bytes - (int64_t)tc_used_bytes;

        int64_t to_free_bytes =

                (int64_t)tc_cached_bytes - ((int64_t)tc_used_bytes * max_cache_percent / 100);

        to_free_bytes = std::max(to_free_bytes, (int64_t)0);

        int64_t memory_pressure = 0;

        int64_t rss_pressure = 0;

        int64_t alloc_bytes = std::max(rss, tc_alloc_bytes);

        memory_pressure = alloc_bytes * 100 / physical_limit_bytes;

        rss_pressure = rss * 100 / physical_limit_bytes;

        expected_aggressive_decommit = init_aggressive_decommit;

        if (memory_pressure > pressure_limit) {

            // We are reaching oom, so release cache aggressively.

            // Ideally, we should reuse cache and not allocate from system any more,

            // however, it is hard to set limit on cache of tcmalloc and doris

            // use mmap in vectorized mode.

            // Limit cache capactiy is enough.

            if (rss_pressure > pressure_limit) {

                int64_t min_free_bytes = alloc_bytes - physical_limit_bytes * 9 / 10;

                to_free_bytes = std::max(to_free_bytes, min_free_bytes);

                to_free_bytes = std::max(to_free_bytes, tc_cached_bytes * 30 / 100);

                // We assure that we have at least 500M bytes in cache.

                to_free_bytes = std::min(to_free_bytes, tc_cached_bytes - 500 * 1024 * 1024);

                expected_aggressive_decommit = 1;

            }

            last_ms = kMaxLastMs;

        } else if (memory_pressure > (pressure_limit - 10)) {

            // In most cases, adjusting release rate is enough, if memory are consumed quickly

            // we should release manually.

            if (last_memory_pressure <= (pressure_limit - 10)) {

                to_free_bytes = std::max(to_free_bytes, tc_cached_bytes * 10 / 100);

            }

        }

        int release_rate_index = int(memory_pressure) / 10;

        double release_rate = 1.0;

        if (release_rate_index >= sizeof(release_rates) / sizeof(release_rates[0])) {

            release_rate = 2000.0;

        } else {

            release_rate = release_rates[release_rate_index];

        }

        MallocExtension::instance()->SetMemoryReleaseRate(release_rate);

        if ((current_aggressive_decommit != expected_aggressive_decommit) && !is_performance_mode) {

            MallocExtension::instance()->SetNumericProperty("tcmalloc.aggressive_memory_decommit",

                                                            expected_aggressive_decommit);

            current_aggressive_decommit = expected_aggressive_decommit;

        }

        last_memory_pressure = memory_pressure;

        // We release at least 2% bytes once, frequent releasing hurts performance.

        if (to_free_bytes > (physical_limit_bytes * 2 / 100)) {

            last_ms += kIntervalMs;

            if (last_ms >= kMaxLastMs) {

                LOG(INFO) << "generic.current_allocated_bytes " << tc_used_bytes

                          << ", generic.total_physical_bytes " << tc_alloc_bytes << ", rss " << rss

                          << ", max_cache_percent " << max_cache_percent << ", release_rate "

                          << release_rate << ", memory_pressure " << memory_pressure

                          << ", physical_limit_bytes " << physical_limit_bytes << ", to_free_bytes "

                          << to_free_bytes << ", current_aggressive_decommit "

                          << current_aggressive_decommit;

                MallocExtension::instance()->ReleaseToSystem(to_free_bytes);

                last_ms = 0;

            }

        } else {

            last_ms = 0;

        }

    }

#endif

}

void refresh_process_memory_metrics() {

    doris::PerfCounters::refresh_proc_status();

    doris::MemInfo::refresh_proc_meminfo();

    doris::GlobalMemoryArbitrator::reset_refresh_interval_memory_growth();

    ExecEnv::GetInstance()->brpc_iobuf_block_memory_tracker()->set_consumption(

            butil::IOBuf::block_memory());

}

void refresh_common_allocator_metrics() {

#if !defined(ADDRESS_SANITIZER) && !defined(LEAK_SANITIZER) && !defined(THREAD_SANITIZER)

    doris::JemallocControl::refresh_allocator_mem();

    if (config::enable_system_metrics) {

        DorisMetrics::instance()->system_metrics()->update_allocator_metrics();

    }

#endif

    doris::GlobalMemoryArbitrator::refresh_memory_bvar();

}

void refresh_memory_state_after_memory_change() {

    if (abs(last_print_proc_mem - PerfCounters::get_vm_rss()) > 268435456) {

        last_print_proc_mem = PerfCounters::get_vm_rss();

        doris::MemTrackerLimiter::clean_tracker_limiter_group();

        doris::ProcessProfile::instance()->memory_profile()->enable_print_log_process_usage();

        doris::ProcessProfile::instance()->memory_profile()->refresh_memory_overview_profile();

        doris::JemallocControl::notify_je_purge_dirty_pages();

        LOG(INFO) << doris::GlobalMemoryArbitrator::

                        process_mem_log_str(); // print mem log when memory state by 256M

    }

}

void refresh_cache_capacity() {

    if (doris::GlobalMemoryArbitrator::cache_adjust_capacity_notify.load(

                std::memory_order_relaxed)) {

        // the last cache capacity adjustment has not been completed.

        // if not return, last_periodic_refreshed_cache_capacity_adjust_weighted may be modified, but notify is ignored.

        return;

    }

    if (refresh_cache_capacity_sleep_time_ms <= 0) {

        auto cache_capacity_reduce_mem_limit = int64_t(

                doris::MemInfo::soft_mem_limit() * config::cache_capacity_reduce_mem_limit_frac);

        int64_t process_memory_usage = doris::GlobalMemoryArbitrator::process_memory_usage();

        // the rule is like this:

        // 1. if the process mem usage < soft memlimit * 0.6, then do not need adjust cache capacity.

        // 2. if the process mem usage > soft memlimit * 0.6 and process mem usage < soft memlimit, then it will be adjusted to a lower value.

        // 3. if the process mem usage > soft memlimit, then the capacity is adjusted to 0.

        double new_cache_capacity_adjust_weighted =

                AlgoUtil::descent_by_step(10, cache_capacity_reduce_mem_limit,

                                          doris::MemInfo::soft_mem_limit(), process_memory_usage);

        if (new_cache_capacity_adjust_weighted !=

            doris::GlobalMemoryArbitrator::last_periodic_refreshed_cache_capacity_adjust_weighted) {

            doris::GlobalMemoryArbitrator::last_periodic_refreshed_cache_capacity_adjust_weighted =

                    new_cache_capacity_adjust_weighted;

            doris::GlobalMemoryArbitrator::notify_cache_adjust_capacity();

            refresh_cache_capacity_sleep_time_ms = config::memory_gc_sleep_time_ms;

        } else {

            refresh_cache_capacity_sleep_time_ms = 0;

        }

    }

    refresh_cache_capacity_sleep_time_ms -= config::memory_maintenance_sleep_time_ms;

}

void je_reset_dirty_decay() {

#ifdef USE_JEMALLOC

    if (doris::JemallocControl::je_reset_dirty_decay_notify.load(std::memory_order_relaxed)) {

        // if not return, je_enable_dirty_page may be modified, but notify is ignored.

        return;

    }

    if (je_reset_dirty_decay_sleep_time_ms <= 0) {

        bool new_je_enable_dirty_page = true;

        if (doris::JemallocControl::je_enable_dirty_page) {

            // if Jemalloc dirty page is enabled and process memory exceed soft mem limit,

            // disable Jemalloc dirty page.

            new_je_enable_dirty_page = !GlobalMemoryArbitrator::is_exceed_soft_mem_limit();

        } else {

            // if Jemalloc dirty page is disabled and 10% free memory left to exceed soft mem limit,

            // enable Jemalloc dirty page, this is to avoid frequent changes

            // between enabling and disabling Jemalloc dirty pages, if the process memory does

            // not exceed the soft mem limit after turning off Jemalloc dirty pages,

            // but it will exceed soft mem limit after turning it on.

            new_je_enable_dirty_page = !GlobalMemoryArbitrator::is_exceed_soft_mem_limit(

                    int64_t(doris::MemInfo::soft_mem_limit() * 0.1));

        }

        if (doris::JemallocControl::je_enable_dirty_page != new_je_enable_dirty_page) {

            // `notify_je_reset_dirty_decay` only if `je_enable_dirty_page` changes.

            doris::JemallocControl::je_enable_dirty_page = new_je_enable_dirty_page;

            doris::JemallocControl::notify_je_reset_dirty_decay();

            je_reset_dirty_decay_sleep_time_ms = config::memory_gc_sleep_time_ms;

        } else {

            je_reset_dirty_decay_sleep_time_ms = 0;

        }

    }

    je_reset_dirty_decay_sleep_time_ms -= config::memory_maintenance_sleep_time_ms;

#endif

}

void memory_gc() {

    if (config::disable_memory_gc) {

        return;

    }

    if (memory_gc_sleep_time <= 0) {

        auto gc_func = [](const std::string& revoke_reason) {

            doris::ProcessProfile::instance()->memory_profile()->print_log_process_usage();

            if (doris::MemoryReclamation::revoke_process_memory(revoke_reason)) {

                // If there is not enough memory to be gc, the process memory usage will not be printed in the next continuous gc.

                doris::ProcessProfile::instance()

                        ->memory_profile()

                        ->enable_print_log_process_usage();

            }

        };

        if (doris::GlobalMemoryArbitrator::sys_mem_available() <

            doris::MemInfo::sys_mem_available_low_water_mark()) {

            gc_func("sys available memory less than low water mark");

        } else if (doris::GlobalMemoryArbitrator::process_memory_usage() >

                   doris::MemInfo::mem_limit()) {

            gc_func("process memory used exceed limit");

        }

        memory_gc_sleep_time = config::memory_gc_sleep_time_ms;

    }

    memory_gc_sleep_time -= config::memory_maintenance_sleep_time_ms;

}

void Daemon::memory_maintenance_thread() {

    while (!_stop_background_threads_latch.wait_for(

            std::chrono::milliseconds(config::memory_maintenance_sleep_time_ms))) {

        // step 1. Refresh process memory metrics.

        refresh_process_memory_metrics();

        // step 2. Refresh jemalloc/tcmalloc metrics.

        refresh_common_allocator_metrics();

        // step 3. Update and print memory stat when the memory changes by 256M.

        refresh_memory_state_after_memory_change();

        // step 4. Asyn Refresh cache capacity

        // TODO adjust cache capacity based on smoothstep (smooth gradient).

        refresh_cache_capacity();

        // step 5. Cancel top memory task when process memory exceed hard limit.

        memory_gc();

        // step 6. Refresh weighted memory ratio of workload groups.

        doris::ExecEnv::GetInstance()->workload_group_mgr()->do_sweep();

        doris::ExecEnv::GetInstance()->workload_group_mgr()->refresh_workload_group_memory_state();

        // step 7: handle paused queries(caused by memory insufficient)

        doris::ExecEnv::GetInstance()->workload_group_mgr()->handle_paused_queries();

        // step 8. Flush memtable

        doris::GlobalMemoryArbitrator::notify_memtable_memory_refresh();

        // TODO notify flush memtable

        // step 9. Reset Jemalloc dirty page decay.

        je_reset_dirty_decay();

    }

}

void Daemon::memtable_memory_refresh_thread() {

    // Refresh the memory statistics of the load channel tracker more frequently,

    // which helps to accurately control the memory of LoadChannelMgr.

    do {

        std::unique_lock<std::mutex> l(doris::GlobalMemoryArbitrator::memtable_memory_refresh_lock);

        while (_stop_background_threads_latch.count() != 0 &&

               !doris::GlobalMemoryArbitrator::memtable_memory_refresh_notify.load(

                       std::memory_order_relaxed)) {

            doris::GlobalMemoryArbitrator::memtable_memory_refresh_cv.wait_for(

                    l, std::chrono::milliseconds(100));

        }

        if (_stop_background_threads_latch.count() == 0) {

            break;

        }

        Defer defer {[&]() {

            doris::GlobalMemoryArbitrator::memtable_memory_refresh_notify.store(

                    false, std::memory_order_relaxed);

        }};

        doris::ExecEnv::GetInstance()->memtable_memory_limiter()->refresh_mem_tracker();

    } while (true);

}

/*

 * this thread will calculate some metrics at a fix interval(15 sec)

 * 1. push bytes per second

 * 2. scan bytes per second

 * 3. max io util of all disks

 * 4. max network send bytes rate

 * 5. max network receive bytes rate

 */

void Daemon::calculate_metrics_thread() {

    int64_t last_ts = -1L;

    int64_t lst_query_bytes = -1;

    std::map<std::string, int64_t> lst_disks_io_time;

    std::map<std::string, int64_t> lst_net_send_bytes;

    std::map<std::string, int64_t> lst_net_receive_bytes;

    do {

        DorisMetrics::instance()->metric_registry()->trigger_all_hooks(true);

        if (last_ts == -1L) {

            last_ts = GetMonoTimeMicros() / 1000;

            lst_query_bytes = DorisMetrics::instance()->query_scan_bytes->value();

            if (config::enable_system_metrics) {

                DorisMetrics::instance()->system_metrics()->get_disks_io_time(&lst_disks_io_time);

                DorisMetrics::instance()->system_metrics()->get_network_traffic(

                        &lst_net_send_bytes, &lst_net_receive_bytes);

            }

        } else {

            int64_t current_ts = GetMonoTimeMicros() / 1000;

            long interval = (current_ts - last_ts) / 1000;

            last_ts = current_ts;

            // 1. query bytes per second

            int64_t current_query_bytes = DorisMetrics::instance()->query_scan_bytes->value();

            int64_t qps = (current_query_bytes - lst_query_bytes) / (interval + 1);

            DorisMetrics::instance()->query_scan_bytes_per_second->set_value(qps < 0 ? 0 : qps);

            lst_query_bytes = current_query_bytes;

            if (config::enable_system_metrics) {

                // 2. max disk io util

                DorisMetrics::instance()->system_metrics()->update_max_disk_io_util_percent(

                        lst_disks_io_time, 15);

                // update lst map

                DorisMetrics::instance()->system_metrics()->get_disks_io_time(&lst_disks_io_time);

                // 3. max network traffic

                int64_t max_send = 0;

                int64_t max_receive = 0;

                DorisMetrics::instance()->system_metrics()->get_max_net_traffic(

                        lst_net_send_bytes, lst_net_receive_bytes, 15, &max_send, &max_receive);

                DorisMetrics::instance()->system_metrics()->update_max_network_send_bytes_rate(

                        max_send);

                DorisMetrics::instance()->system_metrics()->update_max_network_receive_bytes_rate(

                        max_receive);

                // update lst map

                DorisMetrics::instance()->system_metrics()->get_network_traffic(

                        &lst_net_send_bytes, &lst_net_receive_bytes);

                DorisMetrics::instance()->system_metrics()->update_be_avail_cpu_num();

            }

            update_rowsets_and_segments_num_metrics();

        }

    } while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(15)));

}

void Daemon::report_runtime_query_statistics_thread() {

    while (!_stop_background_threads_latch.wait_for(

            std::chrono::milliseconds(config::report_query_statistics_interval_ms))) {

        ExecEnv::GetInstance()->runtime_query_statistics_mgr()->report_runtime_query_statistics();

    }

}

void Daemon::je_reset_dirty_decay_thread() const {

    do {

        std::unique_lock<std::mutex> l(doris::JemallocControl::je_reset_dirty_decay_lock);

        while (_stop_background_threads_latch.count() != 0 &&

               !doris::JemallocControl::je_reset_dirty_decay_notify.load(

                       std::memory_order_relaxed)) {

            doris::JemallocControl::je_reset_dirty_decay_cv.wait_for(

                    l, std::chrono::milliseconds(100));

        }

        if (_stop_background_threads_latch.count() == 0) {

            break;

        }

        Defer defer {[&]() {

            doris::JemallocControl::je_reset_dirty_decay_notify.store(false,

                                                                      std::memory_order_relaxed);

        }};

#ifdef USE_JEMALLOC

        if (config::disable_memory_gc || !config::enable_je_purge_dirty_pages) {

            continue;

        }

        // There is a significant difference only when dirty_decay_ms is equal to 0 or not.

        //

        // 1. When dirty_decay_ms is not equal to 0, the free memory will be cached in the Jemalloc

        // dirty page first. even if dirty_decay_ms is equal to 1, the Jemalloc dirty page will not

        // be released to the system exactly after 1ms, it will be released according to the decay rule.

        // The Jemalloc document specifies that dirty_decay_ms is an approximate time.

        //

        // 2. It has been observed in an actual cluster that even if dirty_decay_ms is changed

        // from th default 5000 to 1, Jemalloc dirty page will still cache a large amount of memory, everything

        // seems to be the same as `dirty_decay_ms:5000`. only when dirty_decay_ms is changed to 0,

        // jemalloc dirty page will stop caching and free memory will be released to the system immediately.

        // of course, performance will be affected.

        //

        // 3. After reducing dirty_decay_ms, manually calling `decay_all_arena_dirty_pages` may release dirty pages

        // as soon as possible, but no relevant experimental data can be found, so it is simple and safe

        // to adjust dirty_decay_ms only between zero and non-zero.

        if (doris::JemallocControl::je_enable_dirty_page) {

            doris::JemallocControl::je_reset_all_arena_dirty_decay_ms(config::je_dirty_decay_ms);

        } else {

            doris::JemallocControl::je_reset_all_arena_dirty_decay_ms(0);

        }

#endif

    } while (true);

}

void Daemon::cache_adjust_capacity_thread() {

    do {

        std::unique_lock<std::mutex> l(doris::GlobalMemoryArbitrator::cache_adjust_capacity_lock);

        while (_stop_background_threads_latch.count() != 0 &&

               !doris::GlobalMemoryArbitrator::cache_adjust_capacity_notify.load(

                       std::memory_order_relaxed)) {

            doris::GlobalMemoryArbitrator::cache_adjust_capacity_cv.wait_for(

                    l, std::chrono::milliseconds(100));

        }

        double adjust_weighted = std::min<double>(

                GlobalMemoryArbitrator::last_periodic_refreshed_cache_capacity_adjust_weighted,

                GlobalMemoryArbitrator::last_memory_exceeded_cache_capacity_adjust_weighted);

        if (_stop_background_threads_latch.count() == 0) {

            break;

        }

        Defer defer {[&]() {

            doris::GlobalMemoryArbitrator::cache_adjust_capacity_notify.store(

                    false, std::memory_order_relaxed);

        }};

        if (config::disable_memory_gc) {

            continue;

        }

        if (GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted ==

            adjust_weighted) {

            LOG(INFO) << fmt::format(

                    "[MemoryGC] adjust cache capacity end, adjust_weighted {} has not been "

                    "modified.",

                    adjust_weighted);

            continue;

        }

        std::unique_ptr<RuntimeProfile> profile = std::make_unique<RuntimeProfile>("");

        auto freed_mem = CacheManager::instance()->for_each_cache_refresh_capacity(adjust_weighted,

                                                                                   profile.get());

        std::stringstream ss;

        profile->pretty_print(&ss);

        LOG(INFO) << fmt::format(

                "[MemoryGC] adjust cache capacity end, free memory {}, details: {}",

                PrettyPrinter::print(freed_mem, TUnit::BYTES), ss.str());

        GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted = adjust_weighted;

    } while (true);

}

void Daemon::cache_prune_stale_thread() {

    int32_t interval = config::cache_periodic_prune_stale_sweep_sec;

    while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(interval))) {

        if (config::cache_periodic_prune_stale_sweep_sec <= 0) {

            LOG(WARNING) << "config of cache clean interval is: [" << interval

                         << "], it means the cache prune stale thread is disabled, will wait 3s "

                            "and check again.";

            interval = 3;

            continue;

        }

        if (config::disable_memory_gc) {

            continue;

        }

        CacheManager::instance()->for_each_cache_prune_stale();

        interval = config::cache_periodic_prune_stale_sweep_sec;

    }

}

void Daemon::be_proc_monitor_thread() {

    while (!_stop_background_threads_latch.wait_for(

            std::chrono::milliseconds(config::be_proc_monitor_interval_ms))) {

        LOG(INFO) << "log be thread num, " << BeProcMonitor::get_be_thread_info();

    }

}

void Daemon::calculate_workload_group_metrics_thread() {

    while (!_stop_background_threads_latch.wait_for(

            std::chrono::milliseconds(config::workload_group_metrics_interval_ms))) {

        ExecEnv::GetInstance()->workload_group_mgr()->refresh_workload_group_metrics();

    }

}

void Daemon::s3_rate_limiter_refresh_thread() {

    // Single trigger for dynamic rate limiter changes: picks up both mutable

    // s3_{get,put}_* config updates and cgroup CPU quota changes (serverless BEs can

    // be resized in place). refresh() is idempotent and compares against the buckets'

    // own parameters, so quiet iterations are cheap no-ops.

    while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(10))) {

        if (config::enable_s3_rate_limiter) {

            S3RateLimiterManager::instance().refresh();

        }

    }

}

void Daemon::start() {

    Status st;

    st = Thread::create(

            "Daemon", "tcmalloc_gc_thread", [this]() { this->tcmalloc_gc_thread(); },

            &_threads.emplace_back());

    CHECK(st.ok()) << st;

    st = Thread::create(

            "Daemon", "memory_maintenance_thread", [this]() { this->memory_maintenance_thread(); },

            &_threads.emplace_back());

    CHECK(st.ok()) << st;

    st = Thread::create(

            "Daemon", "memtable_memory_refresh_thread",

            [this]() { this->memtable_memory_refresh_thread(); }, &_threads.emplace_back());

    CHECK(st.ok()) << st;

    if (config::enable_metric_calculator) {

        st = Thread::create(

                "Daemon", "calculate_metrics_thread",

                [this]() { this->calculate_metrics_thread(); }, &_threads.emplace_back());

        CHECK(st.ok()) << st;

    }

    st = Thread::create(

            "Daemon", "s3_rate_limiter_refresh_thread",

            [this]() { this->s3_rate_limiter_refresh_thread(); }, &_threads.emplace_back());

    CHECK(st.ok()) << st;

    st = Thread::create(

            "Daemon", "je_reset_dirty_decay_thread",

            [this]() { this->je_reset_dirty_decay_thread(); }, &_threads.emplace_back());

    CHECK(st.ok()) << st;

    st = Thread::create(

            "Daemon", "cache_adjust_capacity_thread",

            [this]() { this->cache_adjust_capacity_thread(); }, &_threads.emplace_back());

    CHECK(st.ok()) << st;

    st = Thread::create(

            "Daemon", "cache_prune_stale_thread", [this]() { this->cache_prune_stale_thread(); },

            &_threads.emplace_back());

    CHECK(st.ok()) << st;

    st = Thread::create(

            "Daemon", "query_runtime_statistics_thread",

            [this]() { this->report_runtime_query_statistics_thread(); }, &_threads.emplace_back());

    CHECK(st.ok()) << st;

    if (config::enable_be_proc_monitor) {

        st = Thread::create(

                "Daemon", "be_proc_monitor_thread", [this]() { this->be_proc_monitor_thread(); },

                &_threads.emplace_back());

    }

    CHECK(st.ok()) << st;

    st = Thread::create(

            "Daemon", "workload_group_metrics",

            [this]() { this->calculate_workload_group_metrics_thread(); },

            &_threads.emplace_back());

    CHECK(st.ok()) << st;

}

void Daemon::stop() {

    LOG(INFO) << "Doris daemon is stopping.";

    if (_stop_background_threads_latch.count() == 0) {

        LOG(INFO) << "Doris daemon stop returned since no bg threads latch.";

        return;

    }

    _stop_background_threads_latch.count_down();

    for (auto&& t : _threads) {

        if (t) {

            t->join();

        }

    }

    LOG(INFO) << "Doris daemon stopped after background threads are joined.";

}

} // namespace doris