| // 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/status.h" |
| #include "olap/memtable_memory_limiter.h" |
| #include "olap/storage_engine.h" |
| #include "olap/tablet_manager.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 "util/algorithm_util.h" |
| #include "util/doris_metrics.h" |
| #include "util/mem_info.h" |
| #include "util/metrics.h" |
| #include "util/perf_counters.h" |
| #include "util/system_metrics.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::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", "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 |
