From 97f4ec2adbf614842e9b16badb69a8d5c642abe0 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Wed, 17 Jul 2024 16:59:35 +0200 Subject: [PATCH 01/16] Read cgroup memory usage in async metrics thread --- programs/keeper/Keeper.cpp | 26 ++-- programs/server/Server.cpp | 34 ++--- src/Common/AsynchronousMetrics.cpp | 13 +- src/Common/AsynchronousMetrics.h | 5 +- src/Common/CgroupsMemoryUsageObserver.cpp | 127 ++++-------------- src/Common/CgroupsMemoryUsageObserver.h | 19 +-- src/Common/MemoryTracker.cpp | 38 ++---- src/Common/MemoryTracker.h | 4 +- .../KeeperAsynchronousMetrics.cpp | 2 +- src/Interpreters/Context.cpp | 13 ++ src/Interpreters/Context.h | 4 + .../ServerAsynchronousMetrics.cpp | 2 +- 12 files changed, 112 insertions(+), 175 deletions(-) diff --git a/programs/keeper/Keeper.cpp b/programs/keeper/Keeper.cpp index 44c2daa33ad..3f6020ad48c 100644 --- a/programs/keeper/Keeper.cpp +++ b/programs/keeper/Keeper.cpp @@ -399,6 +399,18 @@ try registerDisks(/*global_skip_access_check=*/false); + auto cgroups_memory_observer_wait_time = config().getUInt64("keeper_server.cgroups_memory_observer_wait_time", 15); + try + { + auto cgroups_reader = createCgroupsReader(); + global_context->setCgroupsReader(createCgroupsReader()); + } + catch (...) + { + if (cgroups_memory_observer_wait_time != 0) + tryLogCurrentException(log, "Failed to create cgroups reader"); + } + /// This object will periodically calculate some metrics. KeeperAsynchronousMetrics async_metrics( global_context, @@ -622,21 +634,19 @@ try main_config_reloader->start(); std::optional cgroups_memory_usage_observer; - try + if (cgroups_memory_observer_wait_time != 0) { - auto wait_time = config().getUInt64("keeper_server.cgroups_memory_observer_wait_time", 15); - if (wait_time != 0) + auto cgroups_reader = global_context->getCgroupsReader(); + if (cgroups_reader) { - cgroups_memory_usage_observer.emplace(std::chrono::seconds(wait_time)); + cgroups_memory_usage_observer.emplace(std::chrono::seconds(cgroups_memory_observer_wait_time), global_context->getCgroupsReader()); /// Not calling cgroups_memory_usage_observer->setLimits() here (as for the normal ClickHouse server) because Keeper controls /// its memory usage by other means (via setting 'max_memory_usage_soft_limit'). cgroups_memory_usage_observer->setOnMemoryAmountAvailableChangedFn([&]() { main_config_reloader->reload(); }); cgroups_memory_usage_observer->startThread(); } - } - catch (Exception &) - { - tryLogCurrentException(log, "Disabling cgroup memory observer because of an error during initialization"); + else + LOG_ERROR(log, "Disabling cgroup memory observer because of an error during initialization of cgroups reader"); } diff --git a/programs/server/Server.cpp b/programs/server/Server.cpp index 053ddaf8d8b..c52b1e037ec 100644 --- a/programs/server/Server.cpp +++ b/programs/server/Server.cpp @@ -897,6 +897,17 @@ try LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds()); }); + try + { + auto cgroups_reader = createCgroupsReader(); + global_context->setCgroupsReader(createCgroupsReader()); + } + catch (...) + { + if (server_settings.cgroups_memory_usage_observer_wait_time != 0) + tryLogCurrentException(log, "Failed to create cgroups reader"); + } + /// This object will periodically calculate some metrics. ServerAsynchronousMetrics async_metrics( global_context, @@ -1456,15 +1467,13 @@ try } std::optional cgroups_memory_usage_observer; - try + if (auto wait_time = server_settings.cgroups_memory_usage_observer_wait_time; wait_time != 0) { - auto wait_time = server_settings.cgroups_memory_usage_observer_wait_time; - if (wait_time != 0) - cgroups_memory_usage_observer.emplace(std::chrono::seconds(wait_time)); - } - catch (Exception &) - { - tryLogCurrentException(log, "Disabling cgroup memory observer because of an error during initialization"); + auto cgroups_reader = global_context->getCgroupsReader(); + if (cgroups_reader) + cgroups_memory_usage_observer.emplace(std::chrono::seconds(wait_time), std::move(cgroups_reader)); + else + LOG_ERROR(log, "Disabling cgroup memory observer because of an error during initialization of cgroups reader"); } std::string cert_path = config().getString("openSSL.server.certificateFile", ""); @@ -1532,15 +1541,6 @@ try total_memory_tracker.setDescription("(total)"); total_memory_tracker.setMetric(CurrentMetrics::MemoryTracking); - if (cgroups_memory_usage_observer) - { - double hard_limit_ratio = new_server_settings.cgroup_memory_watcher_hard_limit_ratio; - double soft_limit_ratio = new_server_settings.cgroup_memory_watcher_soft_limit_ratio; - cgroups_memory_usage_observer->setMemoryUsageLimits( - static_cast(max_server_memory_usage * hard_limit_ratio), - static_cast(max_server_memory_usage * soft_limit_ratio)); - } - size_t merges_mutations_memory_usage_soft_limit = new_server_settings.merges_mutations_memory_usage_soft_limit; size_t default_merges_mutations_server_memory_usage = static_cast(current_physical_server_memory * new_server_settings.merges_mutations_memory_usage_to_ram_ratio); diff --git a/src/Common/AsynchronousMetrics.cpp b/src/Common/AsynchronousMetrics.cpp index 6309f6079f6..0953ad88697 100644 --- a/src/Common/AsynchronousMetrics.cpp +++ b/src/Common/AsynchronousMetrics.cpp @@ -57,10 +57,12 @@ static std::unique_ptr openFileIfExists(const std::stri AsynchronousMetrics::AsynchronousMetrics( unsigned update_period_seconds, - const ProtocolServerMetricsFunc & protocol_server_metrics_func_) + const ProtocolServerMetricsFunc & protocol_server_metrics_func_, + std::shared_ptr cgroups_reader_) : update_period(update_period_seconds) , log(getLogger("AsynchronousMetrics")) , protocol_server_metrics_func(protocol_server_metrics_func_) + , cgroups_reader(std::move(cgroups_reader_)) { #if defined(OS_LINUX) openFileIfExists("/proc/meminfo", meminfo); @@ -669,6 +671,13 @@ void AsynchronousMetrics::update(TimePoint update_time, bool force_update) free_memory_in_allocator_arenas = je_malloc_pdirty * getPageSize(); #endif + if (cgroups_reader != nullptr) + { + rss = cgroups_reader->readMemoryUsage(); + new_values["CgroupsMemoryUsage"] = { rss, + "The amount of physical memory used by the server process, reported by cgroups." }; + } + Int64 difference = rss - amount; /// Log only if difference is high. This is for convenience. The threshold is arbitrary. @@ -681,7 +690,7 @@ void AsynchronousMetrics::update(TimePoint update_time, bool force_update) ReadableSize(rss), ReadableSize(difference)); - MemoryTracker::setRSS(rss, free_memory_in_allocator_arenas); + MemoryTracker::setRSS(rss, /*has_free_memory_in_allocator_arenas_=*/free_memory_in_allocator_arenas > 0); } } diff --git a/src/Common/AsynchronousMetrics.h b/src/Common/AsynchronousMetrics.h index 10a972d2458..0b110f41fc3 100644 --- a/src/Common/AsynchronousMetrics.h +++ b/src/Common/AsynchronousMetrics.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include #include @@ -68,7 +69,8 @@ public: AsynchronousMetrics( unsigned update_period_seconds, - const ProtocolServerMetricsFunc & protocol_server_metrics_func_); + const ProtocolServerMetricsFunc & protocol_server_metrics_func_, + std::shared_ptr cgroups_reader_); virtual ~AsynchronousMetrics(); @@ -91,6 +93,7 @@ private: virtual void logImpl(AsynchronousMetricValues &) {} ProtocolServerMetricsFunc protocol_server_metrics_func; + std::shared_ptr cgroups_reader; std::unique_ptr thread; diff --git a/src/Common/CgroupsMemoryUsageObserver.cpp b/src/Common/CgroupsMemoryUsageObserver.cpp index 02bde0d80b7..b12845df098 100644 --- a/src/Common/CgroupsMemoryUsageObserver.cpp +++ b/src/Common/CgroupsMemoryUsageObserver.cpp @@ -17,13 +17,6 @@ #include #include -#include "config.h" -#if USE_JEMALLOC -# include -#define STRINGIFY_HELPER(x) #x -#define STRINGIFY(x) STRINGIFY_HELPER(x) -#endif - using namespace DB; namespace fs = std::filesystem; @@ -155,15 +148,21 @@ std::optional getCgroupsV1Path() return {default_cgroups_mount / "memory"}; } -std::pair getCgroupsPath() +enum class CgroupsVersion : uint8_t +{ + V1, + V2 +}; + +std::pair getCgroupsPath() { auto v2_path = getCgroupsV2Path(); if (v2_path.has_value()) - return {*v2_path, CgroupsMemoryUsageObserver::CgroupsVersion::V2}; + return {*v2_path, CgroupsVersion::V2}; auto v1_path = getCgroupsV1Path(); if (v1_path.has_value()) - return {*v1_path, CgroupsMemoryUsageObserver::CgroupsVersion::V1}; + return {*v1_path, CgroupsVersion::V1}; throw Exception(ErrorCodes::FILE_DOESNT_EXIST, "Cannot find cgroups v1 or v2 current memory file"); } @@ -173,22 +172,29 @@ std::pair getCgroupsPat namespace DB { -CgroupsMemoryUsageObserver::CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_) - : log(getLogger("CgroupsMemoryUsageObserver")), wait_time(wait_time_) +std::shared_ptr createCgroupsReader() { const auto [cgroup_path, version] = getCgroupsPath(); + LOG_INFO( + getLogger("CgroupsReader"), + "Will create cgroup reader from '{}' (cgroups version: {})", + cgroup_path, + (version == CgroupsVersion::V1) ? "v1" : "v2"); if (version == CgroupsVersion::V2) - cgroup_reader = std::make_unique(cgroup_path); + return std::make_shared(cgroup_path); else - cgroup_reader = std::make_unique(cgroup_path); + { + chassert(version == CgroupsVersion::V1); + return std::make_shared(cgroup_path); + } - LOG_INFO( - log, - "Will read the current memory usage from '{}' (cgroups version: {}), wait time is {} sec", - cgroup_path, - (version == CgroupsVersion::V1) ? "v1" : "v2", - wait_time.count()); +} + +CgroupsMemoryUsageObserver::CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_, std::shared_ptr cgroups_reader_) + : log(getLogger("CgroupsMemoryUsageObserver")), wait_time(wait_time_), cgroups_reader(std::move(cgroups_reader_)) +{ + cgroups_reader = createCgroupsReader(); } CgroupsMemoryUsageObserver::~CgroupsMemoryUsageObserver() @@ -196,58 +202,6 @@ CgroupsMemoryUsageObserver::~CgroupsMemoryUsageObserver() stopThread(); } -void CgroupsMemoryUsageObserver::setMemoryUsageLimits(uint64_t hard_limit_, uint64_t soft_limit_) -{ - std::lock_guard limit_lock(limit_mutex); - - if (hard_limit_ == hard_limit && soft_limit_ == soft_limit) - return; - - hard_limit = hard_limit_; - soft_limit = soft_limit_; - - on_hard_limit = [this, hard_limit_](bool up) - { - if (up) - { - LOG_WARNING(log, "Exceeded hard memory limit ({})", ReadableSize(hard_limit_)); - - /// Update current usage in memory tracker. Also reset free_memory_in_allocator_arenas to zero though we don't know if they are - /// really zero. Trying to avoid OOM ... - MemoryTracker::setRSS(hard_limit_, 0); - } - else - { - LOG_INFO(log, "Dropped below hard memory limit ({})", ReadableSize(hard_limit_)); - } - }; - - on_soft_limit = [this, soft_limit_](bool up) - { - if (up) - { - LOG_WARNING(log, "Exceeded soft memory limit ({})", ReadableSize(soft_limit_)); - -# if USE_JEMALLOC - LOG_INFO(log, "Purging jemalloc arenas"); - mallctl("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge", nullptr, nullptr, nullptr, 0); -# endif - /// Reset current usage in memory tracker. Expect zero for free_memory_in_allocator_arenas as we just purged them. - uint64_t memory_usage = cgroup_reader->readMemoryUsage(); - LOG_TRACE(log, "Read current memory usage {} bytes ({}) from cgroups", memory_usage, ReadableSize(memory_usage)); - MemoryTracker::setRSS(memory_usage, 0); - - LOG_INFO(log, "Purged jemalloc arenas. Current memory usage is {}", ReadableSize(memory_usage)); - } - else - { - LOG_INFO(log, "Dropped below soft memory limit ({})", ReadableSize(soft_limit_)); - } - }; - - LOG_INFO(log, "Set new limits, soft limit: {}, hard limit: {}", ReadableSize(soft_limit_), ReadableSize(hard_limit_)); -} - void CgroupsMemoryUsageObserver::setOnMemoryAmountAvailableChangedFn(OnMemoryAmountAvailableChangedFn on_memory_amount_available_changed_) { std::lock_guard memory_amount_available_changed_lock(memory_amount_available_changed_mutex); @@ -301,35 +255,6 @@ void CgroupsMemoryUsageObserver::runThread() std::lock_guard memory_amount_available_changed_lock(memory_amount_available_changed_mutex); on_memory_amount_available_changed(); } - - std::lock_guard limit_lock(limit_mutex); - if (soft_limit > 0 && hard_limit > 0) - { - uint64_t memory_usage = cgroup_reader->readMemoryUsage(); - LOG_TRACE(log, "Read current memory usage {} bytes ({}) from cgroups", memory_usage, ReadableSize(memory_usage)); - if (memory_usage > hard_limit) - { - if (last_memory_usage <= hard_limit) - on_hard_limit(true); - } - else - { - if (last_memory_usage > hard_limit) - on_hard_limit(false); - } - - if (memory_usage > soft_limit) - { - if (last_memory_usage <= soft_limit) - on_soft_limit(true); - } - else - { - if (last_memory_usage > soft_limit) - on_soft_limit(false); - } - last_memory_usage = memory_usage; - } } catch (...) { diff --git a/src/Common/CgroupsMemoryUsageObserver.h b/src/Common/CgroupsMemoryUsageObserver.h index b848a2bff3c..078307a6fa0 100644 --- a/src/Common/CgroupsMemoryUsageObserver.h +++ b/src/Common/CgroupsMemoryUsageObserver.h @@ -16,6 +16,8 @@ struct ICgroupsReader virtual uint64_t readMemoryUsage() = 0; }; +std::shared_ptr createCgroupsReader(); + /// Does two things: /// 1. Periodically reads the memory usage of the process from Linux cgroups. /// You can specify soft or hard memory limits: @@ -35,19 +37,11 @@ struct ICgroupsReader class CgroupsMemoryUsageObserver { public: - using OnMemoryLimitFn = std::function; using OnMemoryAmountAvailableChangedFn = std::function; - enum class CgroupsVersion : uint8_t - { - V1, - V2 - }; - - explicit CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_); + explicit CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_, std::shared_ptr cgroups_reader_); ~CgroupsMemoryUsageObserver(); - void setMemoryUsageLimits(uint64_t hard_limit_, uint64_t soft_limit_); void setOnMemoryAmountAvailableChangedFn(OnMemoryAmountAvailableChangedFn on_memory_amount_available_changed_); void startThread(); @@ -58,22 +52,17 @@ private: const std::chrono::seconds wait_time; std::mutex limit_mutex; - size_t hard_limit TSA_GUARDED_BY(limit_mutex) = 0; - size_t soft_limit TSA_GUARDED_BY(limit_mutex) = 0; - OnMemoryLimitFn on_hard_limit TSA_GUARDED_BY(limit_mutex); - OnMemoryLimitFn on_soft_limit TSA_GUARDED_BY(limit_mutex); std::mutex memory_amount_available_changed_mutex; OnMemoryAmountAvailableChangedFn on_memory_amount_available_changed TSA_GUARDED_BY(memory_amount_available_changed_mutex); - uint64_t last_memory_usage = 0; /// how much memory does the process use uint64_t last_available_memory_amount; /// how much memory can the process use void stopThread(); void runThread(); - std::unique_ptr cgroup_reader; + std::shared_ptr cgroups_reader; std::mutex thread_mutex; std::condition_variable cond; diff --git a/src/Common/MemoryTracker.cpp b/src/Common/MemoryTracker.cpp index 28cfa98666a..eaf34d87ec5 100644 --- a/src/Common/MemoryTracker.cpp +++ b/src/Common/MemoryTracker.cpp @@ -26,7 +26,6 @@ #endif #include -#include #include #include #include @@ -123,7 +122,7 @@ static constexpr size_t log_peak_memory_usage_every = 1ULL << 30; MemoryTracker total_memory_tracker(nullptr, VariableContext::Global); MemoryTracker background_memory_tracker(&total_memory_tracker, VariableContext::User, false); -std::atomic MemoryTracker::free_memory_in_allocator_arenas; +std::atomic MemoryTracker::has_free_memory_in_allocator_arenas; MemoryTracker::MemoryTracker(VariableContext level_) : parent(&total_memory_tracker), level(level_) {} MemoryTracker::MemoryTracker(MemoryTracker * parent_, VariableContext level_) : parent(parent_), level(level_) {} @@ -204,7 +203,7 @@ void MemoryTracker::debugLogBigAllocationWithoutCheck(Int64 size [[maybe_unused] LOG_TEST(getLogger("MemoryTracker"), "Too big allocation ({} bytes) without checking memory limits, " "it may lead to OOM. Stack trace: {}", size, StackTrace().toString()); #else - return; /// Avoid trash logging in release builds + /// Avoid trash logging in release builds #endif } @@ -294,33 +293,18 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed } } - Int64 limit_to_check = current_hard_limit; - #if USE_JEMALLOC - if (level == VariableContext::Global && allow_use_jemalloc_memory.load(std::memory_order_relaxed)) + if (level == VariableContext::Global && will_be > soft_limit.load(std::memory_order_relaxed) + && has_free_memory_in_allocator_arenas.exchange(false)) { - /// Jemalloc arenas may keep some extra memory. - /// This memory was substucted from RSS to decrease memory drift. - /// In case memory is close to limit, try to pugre the arenas. - /// This is needed to avoid OOM, because some allocations are directly done with mmap. - Int64 current_free_memory_in_allocator_arenas = free_memory_in_allocator_arenas.load(std::memory_order_relaxed); - - if (current_free_memory_in_allocator_arenas > 0 && current_hard_limit && current_free_memory_in_allocator_arenas + will_be > current_hard_limit) - { - if (free_memory_in_allocator_arenas.exchange(-current_free_memory_in_allocator_arenas) > 0) - { - Stopwatch watch; - mallctl("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge", nullptr, nullptr, nullptr, 0); - ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurge); - ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurgeTimeMicroseconds, watch.elapsedMicroseconds()); - } - } - - limit_to_check += abs(current_free_memory_in_allocator_arenas); + Stopwatch watch; + mallctl("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge", nullptr, nullptr, nullptr, 0); + ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurge); + ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurgeTimeMicroseconds, watch.elapsedMicroseconds()); } #endif - if (unlikely(current_hard_limit && will_be > limit_to_check)) + if (unlikely(current_hard_limit && will_be > current_hard_limit)) { if (memoryTrackerCanThrow(level, false) && throw_if_memory_exceeded) { @@ -526,11 +510,11 @@ void MemoryTracker::reset() } -void MemoryTracker::setRSS(Int64 rss_, Int64 free_memory_in_allocator_arenas_) +void MemoryTracker::setRSS(Int64 rss_, bool has_free_memory_in_allocator_arenas_) { Int64 new_amount = rss_; total_memory_tracker.amount.store(new_amount, std::memory_order_relaxed); - free_memory_in_allocator_arenas.store(free_memory_in_allocator_arenas_, std::memory_order_relaxed); + has_free_memory_in_allocator_arenas.store(has_free_memory_in_allocator_arenas_, std::memory_order_relaxed); auto metric_loaded = total_memory_tracker.metric.load(std::memory_order_relaxed); if (metric_loaded != CurrentMetrics::end()) diff --git a/src/Common/MemoryTracker.h b/src/Common/MemoryTracker.h index fd32b631774..48d02fd1fc6 100644 --- a/src/Common/MemoryTracker.h +++ b/src/Common/MemoryTracker.h @@ -59,7 +59,7 @@ private: std::atomic profiler_limit {0}; std::atomic_bool allow_use_jemalloc_memory {true}; - static std::atomic free_memory_in_allocator_arenas; + static std::atomic has_free_memory_in_allocator_arenas; Int64 profiler_step = 0; @@ -252,7 +252,7 @@ public: /// Reset current counter to an RSS value. /// Jemalloc may have pre-allocated arenas, they are accounted in RSS. /// We can free this arenas in case of exception to avoid OOM. - static void setRSS(Int64 rss_, Int64 free_memory_in_allocator_arenas_); + static void setRSS(Int64 rss_, bool has_free_memory_in_allocator_arenas_); /// Prints info about peak memory consumption into log. void logPeakMemoryUsage(); diff --git a/src/Coordination/KeeperAsynchronousMetrics.cpp b/src/Coordination/KeeperAsynchronousMetrics.cpp index 86166ffe31b..3e404b7152b 100644 --- a/src/Coordination/KeeperAsynchronousMetrics.cpp +++ b/src/Coordination/KeeperAsynchronousMetrics.cpp @@ -115,7 +115,7 @@ void updateKeeperInformation(KeeperDispatcher & keeper_dispatcher, AsynchronousM KeeperAsynchronousMetrics::KeeperAsynchronousMetrics( ContextPtr context_, unsigned update_period_seconds, const ProtocolServerMetricsFunc & protocol_server_metrics_func_) - : AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_), context(std::move(context_)) + : AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_, context_->getCgroupsReader()), context(std::move(context_)) { } diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp index 2602afd8b78..771b8e9e558 100644 --- a/src/Interpreters/Context.cpp +++ b/src/Interpreters/Context.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -405,6 +406,8 @@ struct ContextSharedPart : boost::noncopyable std::unique_ptr cluster_discovery TSA_GUARDED_BY(clusters_mutex); size_t clusters_version TSA_GUARDED_BY(clusters_mutex) = 0; + std::shared_ptr cgroups_reader; + /// No lock required for async_insert_queue modified only during initialization std::shared_ptr async_insert_queue; @@ -5635,6 +5638,16 @@ const ServerSettings & Context::getServerSettings() const return shared->server_settings; } +void Context::setCgroupsReader(std::shared_ptr cgroups_reader_) +{ + shared->cgroups_reader = std::move(cgroups_reader_); +} + +std::shared_ptr Context::getCgroupsReader() const +{ + return shared->cgroups_reader; +} + uint64_t HTTPContext::getMaxHstsAge() const { return context->getSettingsRef().hsts_max_age; diff --git a/src/Interpreters/Context.h b/src/Interpreters/Context.h index 284cac50769..f183a72e8e2 100644 --- a/src/Interpreters/Context.h +++ b/src/Interpreters/Context.h @@ -150,6 +150,7 @@ class ServerType; template class MergeTreeBackgroundExecutor; class AsyncLoader; +struct ICgroupsReader; struct TemporaryTableHolder; using TemporaryTablesMapping = std::map>; @@ -1344,6 +1345,9 @@ public: const ServerSettings & getServerSettings() const; + void setCgroupsReader(std::shared_ptr cgroups_reader_); + std::shared_ptr getCgroupsReader() const; + private: std::shared_ptr getSettingsConstraintsAndCurrentProfilesWithLock() const; diff --git a/src/Interpreters/ServerAsynchronousMetrics.cpp b/src/Interpreters/ServerAsynchronousMetrics.cpp index 872a9f864df..6ee0168bede 100644 --- a/src/Interpreters/ServerAsynchronousMetrics.cpp +++ b/src/Interpreters/ServerAsynchronousMetrics.cpp @@ -57,7 +57,7 @@ ServerAsynchronousMetrics::ServerAsynchronousMetrics( unsigned heavy_metrics_update_period_seconds, const ProtocolServerMetricsFunc & protocol_server_metrics_func_) : WithContext(global_context_) - , AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_) + , AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_, getContext()->getCgroupsReader()) , heavy_metric_update_period(heavy_metrics_update_period_seconds) { /// sanity check From 21009577d867d493a6ceda48987c060498774f94 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Thu, 18 Jul 2024 09:01:08 +0200 Subject: [PATCH 02/16] Dedicated memory background thread --- programs/server/Server.cpp | 39 ++++++++++++++++++++++++--- src/CMakeLists.txt | 2 +- src/Common/AsynchronousMetrics.cpp | 43 ------------------------------ src/Common/Jemalloc.cpp | 16 ----------- src/Common/Jemalloc.h | 29 ++++++++++++++++++++ src/Common/MemoryTracker.cpp | 33 +++++++++-------------- src/Common/MemoryTracker.h | 1 - 7 files changed, 78 insertions(+), 85 deletions(-) diff --git a/programs/server/Server.cpp b/programs/server/Server.cpp index c52b1e037ec..ca46338d1c1 100644 --- a/programs/server/Server.cpp +++ b/programs/server/Server.cpp @@ -447,9 +447,12 @@ void checkForUsersNotInMainConfig( } } +namespace +{ + /// Unused in other builds #if defined(OS_LINUX) -static String readLine(const String & path) +String readLine(const String & path) { ReadBufferFromFile in(path); String contents; @@ -457,7 +460,7 @@ static String readLine(const String & path) return contents; } -static int readNumber(const String & path) +int readNumber(const String & path) { ReadBufferFromFile in(path); int result; @@ -467,7 +470,7 @@ static int readNumber(const String & path) #endif -static void sanityChecks(Server & server) +void sanityChecks(Server & server) { std::string data_path = getCanonicalPath(server.config().getString("path", DBMS_DEFAULT_PATH)); std::string logs_path = server.config().getString("logger.log", ""); @@ -588,6 +591,31 @@ static void sanityChecks(Server & server) } } +[[noreturn]] void backgroundMemoryThread() +{ + std::mutex mutex; + std::condition_variable cv; + + std::unique_lock lock(mutex); + while (true) + { + cv.wait_for(lock, std::chrono::microseconds(200)); + uint64_t epoch = 0; + mallctl("epoch", nullptr, nullptr, &epoch, sizeof(epoch)); + auto maybe_resident = getJemallocValue("stats.resident"); + if (!maybe_resident.has_value()) + continue; + + Int64 resident = *maybe_resident; + //LOG_INFO(getLogger("JEmalloc"), "Resident {}", ReadableSize(resident)); + MemoryTracker::setRSS(resident, false); + if (resident > total_memory_tracker.getHardLimit()) + purgeJemallocArenas(); + } +} + +} + void loadStartupScripts(const Poco::Util::AbstractConfiguration & config, ContextMutablePtr context, Poco::Logger * log) { try @@ -877,6 +905,11 @@ try total_memory_tracker.setSampleMaxAllocationSize(server_settings.total_memory_profiler_sample_max_allocation_size); } + ThreadFromGlobalPool background_memory_thread([] + { + backgroundMemoryThread(); + }); + Poco::ThreadPool server_pool( /* minCapacity */3, /* maxCapacity */server_settings.max_connections, diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index d985595154c..bfa41eacea1 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -174,7 +174,7 @@ add_library (clickhouse_new_delete STATIC Common/new_delete.cpp) target_link_libraries (clickhouse_new_delete PRIVATE clickhouse_common_io) if (TARGET ch_contrib::jemalloc) target_link_libraries (clickhouse_new_delete PRIVATE ch_contrib::jemalloc) - target_link_libraries (clickhouse_common_io PRIVATE ch_contrib::jemalloc) + target_link_libraries (clickhouse_common_io PUBLIC ch_contrib::jemalloc) target_link_libraries (clickhouse_storages_system PRIVATE ch_contrib::jemalloc) endif() diff --git a/src/Common/AsynchronousMetrics.cpp b/src/Common/AsynchronousMetrics.cpp index 0953ad88697..b3e53c29d4a 100644 --- a/src/Common/AsynchronousMetrics.cpp +++ b/src/Common/AsynchronousMetrics.cpp @@ -649,49 +649,6 @@ void AsynchronousMetrics::update(TimePoint update_time, bool force_update) "The amount of virtual memory mapped for the use of stack and for the allocated memory, in bytes." " It is unspecified whether it includes the per-thread stacks and most of the allocated memory, that is allocated with the 'mmap' system call." " This metric exists only for completeness reasons. I recommend to use the `MemoryResident` metric for monitoring."}; - - /// We must update the value of total_memory_tracker periodically. - /// Otherwise it might be calculated incorrectly - it can include a "drift" of memory amount. - /// See https://github.com/ClickHouse/ClickHouse/issues/10293 - { - Int64 amount = total_memory_tracker.get(); - Int64 peak = total_memory_tracker.getPeak(); - Int64 rss = data.resident; - Int64 free_memory_in_allocator_arenas = 0; - -#if USE_JEMALLOC - /// According to jemalloc man, pdirty is: - /// - /// Number of pages within unused extents that are potentially - /// dirty, and for which madvise() or similar has not been called. - /// - /// So they will be subtracted from RSS to make accounting more - /// accurate, since those pages are not really RSS but a memory - /// that can be used at anytime via jemalloc. - free_memory_in_allocator_arenas = je_malloc_pdirty * getPageSize(); -#endif - - if (cgroups_reader != nullptr) - { - rss = cgroups_reader->readMemoryUsage(); - new_values["CgroupsMemoryUsage"] = { rss, - "The amount of physical memory used by the server process, reported by cgroups." }; - } - - Int64 difference = rss - amount; - - /// Log only if difference is high. This is for convenience. The threshold is arbitrary. - if (difference >= 1048576 || difference <= -1048576) - LOG_TRACE(log, - "MemoryTracking: was {}, peak {}, free memory in arenas {}, will set to {} (RSS), difference: {}", - ReadableSize(amount), - ReadableSize(peak), - ReadableSize(free_memory_in_allocator_arenas), - ReadableSize(rss), - ReadableSize(difference)); - - MemoryTracker::setRSS(rss, /*has_free_memory_in_allocator_arenas_=*/free_memory_in_allocator_arenas > 0); - } } { diff --git a/src/Common/Jemalloc.cpp b/src/Common/Jemalloc.cpp index d7cc246db6a..d8ff9268cca 100644 --- a/src/Common/Jemalloc.cpp +++ b/src/Common/Jemalloc.cpp @@ -5,7 +5,6 @@ #include #include #include -#include #define STRINGIFY_HELPER(x) #x #define STRINGIFY(x) STRINGIFY_HELPER(x) @@ -26,7 +25,6 @@ namespace ErrorCodes void purgeJemallocArenas() { - LOG_TRACE(getLogger("SystemJemalloc"), "Purging unused memory"); Stopwatch watch; mallctl("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge", nullptr, nullptr, nullptr, 0); ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurge); @@ -46,20 +44,6 @@ void checkJemallocProfilingEnabled() "set: MALLOC_CONF=background_thread:true,prof:true"); } -template -void setJemallocValue(const char * name, T value) -{ - T old_value; - size_t old_value_size = sizeof(T); - if (mallctl(name, &old_value, &old_value_size, reinterpret_cast(&value), sizeof(T))) - { - LOG_WARNING(getLogger("Jemalloc"), "mallctl for {} failed", name); - return; - } - - LOG_INFO(getLogger("Jemalloc"), "Value for {} set to {} (from {})", name, value, old_value); -} - void setJemallocProfileActive(bool value) { checkJemallocProfilingEnabled(); diff --git a/src/Common/Jemalloc.h b/src/Common/Jemalloc.h index 499a906fd3d..0c533711f78 100644 --- a/src/Common/Jemalloc.h +++ b/src/Common/Jemalloc.h @@ -5,6 +5,8 @@ #if USE_JEMALLOC #include +#include +#include namespace DB { @@ -21,6 +23,33 @@ void setJemallocBackgroundThreads(bool enabled); void setJemallocMaxBackgroundThreads(size_t max_threads); +template +void setJemallocValue(const char * name, T value) +{ + T old_value; + size_t old_value_size = sizeof(T); + if (mallctl(name, &old_value, &old_value_size, reinterpret_cast(&value), sizeof(T))) + { + LOG_WARNING(getLogger("Jemalloc"), "mallctl for {} failed", name); + return; + } + + LOG_INFO(getLogger("Jemalloc"), "Value for {} set to {} (from {})", name, value, old_value); +} + +template +std::optional getJemallocValue(const char * name) +{ + T value; + size_t value_size = sizeof(T); + if (mallctl(name, &value, &value_size, nullptr, 0)) + { + LOG_WARNING(getLogger("Jemalloc"), "mallctl for {} failed", name); + return std::nullopt; + } + return value; +} + } #endif diff --git a/src/Common/MemoryTracker.cpp b/src/Common/MemoryTracker.cpp index eaf34d87ec5..a541eeeb25f 100644 --- a/src/Common/MemoryTracker.cpp +++ b/src/Common/MemoryTracker.cpp @@ -20,9 +20,6 @@ #if USE_JEMALLOC # include -#define STRINGIFY_HELPER(x) #x -#define STRINGIFY(x) STRINGIFY_HELPER(x) - #endif #include @@ -126,11 +123,11 @@ std::atomic MemoryTracker::has_free_memory_in_allocator_arenas; MemoryTracker::MemoryTracker(VariableContext level_) : parent(&total_memory_tracker), level(level_) {} MemoryTracker::MemoryTracker(MemoryTracker * parent_, VariableContext level_) : parent(parent_), level(level_) {} + MemoryTracker::MemoryTracker(MemoryTracker * parent_, VariableContext level_, bool log_peak_memory_usage_in_destructor_) - : parent(parent_) - , log_peak_memory_usage_in_destructor(log_peak_memory_usage_in_destructor_) - , level(level_) -{} + : parent(parent_), log_peak_memory_usage_in_destructor(log_peak_memory_usage_in_destructor_), level(level_) +{ +} MemoryTracker::~MemoryTracker() { @@ -200,8 +197,12 @@ void MemoryTracker::debugLogBigAllocationWithoutCheck(Int64 size [[maybe_unused] return; MemoryTrackerBlockerInThread blocker(VariableContext::Global); - LOG_TEST(getLogger("MemoryTracker"), "Too big allocation ({} bytes) without checking memory limits, " - "it may lead to OOM. Stack trace: {}", size, StackTrace().toString()); + LOG_TEST( + getLogger("MemoryTracker"), + "Too big allocation ({} bytes) without checking memory limits, " + "it may lead to OOM. Stack trace: {}", + size, + StackTrace().toString()); #else /// Avoid trash logging in release builds #endif @@ -293,17 +294,6 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed } } -#if USE_JEMALLOC - if (level == VariableContext::Global && will_be > soft_limit.load(std::memory_order_relaxed) - && has_free_memory_in_allocator_arenas.exchange(false)) - { - Stopwatch watch; - mallctl("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge", nullptr, nullptr, nullptr, 0); - ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurge); - ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurgeTimeMicroseconds, watch.elapsedMicroseconds()); - } -#endif - if (unlikely(current_hard_limit && will_be > current_hard_limit)) { if (memoryTrackerCanThrow(level, false) && throw_if_memory_exceeded) @@ -513,7 +503,8 @@ void MemoryTracker::reset() void MemoryTracker::setRSS(Int64 rss_, bool has_free_memory_in_allocator_arenas_) { Int64 new_amount = rss_; - total_memory_tracker.amount.store(new_amount, std::memory_order_relaxed); + if (rss_) + total_memory_tracker.amount.store(new_amount, std::memory_order_relaxed); has_free_memory_in_allocator_arenas.store(has_free_memory_in_allocator_arenas_, std::memory_order_relaxed); auto metric_loaded = total_memory_tracker.metric.load(std::memory_order_relaxed); diff --git a/src/Common/MemoryTracker.h b/src/Common/MemoryTracker.h index 48d02fd1fc6..257ed7d0629 100644 --- a/src/Common/MemoryTracker.h +++ b/src/Common/MemoryTracker.h @@ -2,7 +2,6 @@ #include #include -#include #include #include #include From 9a43183eb39dc056186432e6478e050db5045ecf Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Thu, 18 Jul 2024 10:31:24 +0200 Subject: [PATCH 03/16] Finish background memory thread --- programs/keeper/Keeper.cpp | 30 ++++------ programs/server/Server.cpp | 59 ++++--------------- src/Common/AsynchronousMetrics.cpp | 19 ++---- src/Common/AsynchronousMetrics.h | 3 +- src/Common/CgroupsMemoryUsageObserver.cpp | 8 +-- src/Common/CgroupsMemoryUsageObserver.h | 24 +++----- src/Common/Jemalloc.h | 42 +++++++++---- src/Common/MemoryTracker.cpp | 10 +--- src/Common/MemoryTracker.h | 13 +--- src/Common/MemoryWorker.cpp | 49 +++++++++++++++ src/Common/MemoryWorker.h | 34 +++++++++++ .../KeeperAsynchronousMetrics.cpp | 2 +- src/Core/ServerSettings.h | 1 + src/Interpreters/Context.cpp | 12 ---- src/Interpreters/Context.h | 3 - .../ServerAsynchronousMetrics.cpp | 2 +- .../System/StorageSystemServerSettings.cpp | 1 - tests/integration/test_memory_limit/test.py | 1 - 18 files changed, 158 insertions(+), 155 deletions(-) create mode 100644 src/Common/MemoryWorker.cpp create mode 100644 src/Common/MemoryWorker.h diff --git a/programs/keeper/Keeper.cpp b/programs/keeper/Keeper.cpp index 3f6020ad48c..87f126a0046 100644 --- a/programs/keeper/Keeper.cpp +++ b/programs/keeper/Keeper.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include #include @@ -371,6 +372,8 @@ try LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds()); }); + MemoryWorker memory_worker(config().getUInt64("memory_worker_period_ms", 100)); + static ServerErrorHandler error_handler; Poco::ErrorHandler::set(&error_handler); @@ -399,18 +402,6 @@ try registerDisks(/*global_skip_access_check=*/false); - auto cgroups_memory_observer_wait_time = config().getUInt64("keeper_server.cgroups_memory_observer_wait_time", 15); - try - { - auto cgroups_reader = createCgroupsReader(); - global_context->setCgroupsReader(createCgroupsReader()); - } - catch (...) - { - if (cgroups_memory_observer_wait_time != 0) - tryLogCurrentException(log, "Failed to create cgroups reader"); - } - /// This object will periodically calculate some metrics. KeeperAsynchronousMetrics async_metrics( global_context, @@ -634,21 +625,22 @@ try main_config_reloader->start(); std::optional cgroups_memory_usage_observer; - if (cgroups_memory_observer_wait_time != 0) + try { - auto cgroups_reader = global_context->getCgroupsReader(); - if (cgroups_reader) + auto wait_time = config().getUInt64("keeper_server.cgroups_memory_observer_wait_time", 15); + if (wait_time != 0) { - cgroups_memory_usage_observer.emplace(std::chrono::seconds(cgroups_memory_observer_wait_time), global_context->getCgroupsReader()); + cgroups_memory_usage_observer.emplace(std::chrono::seconds(wait_time)); /// Not calling cgroups_memory_usage_observer->setLimits() here (as for the normal ClickHouse server) because Keeper controls /// its memory usage by other means (via setting 'max_memory_usage_soft_limit'). cgroups_memory_usage_observer->setOnMemoryAmountAvailableChangedFn([&]() { main_config_reloader->reload(); }); cgroups_memory_usage_observer->startThread(); } - else - LOG_ERROR(log, "Disabling cgroup memory observer because of an error during initialization of cgroups reader"); } - + catch (Exception &) + { + tryLogCurrentException(log, "Disabling cgroup memory observer because of an error during initialization"); + } LOG_INFO(log, "Ready for connections."); diff --git a/programs/server/Server.cpp b/programs/server/Server.cpp index ca46338d1c1..ae445477c3a 100644 --- a/programs/server/Server.cpp +++ b/programs/server/Server.cpp @@ -11,7 +11,6 @@ #include #include #include -#include #include #include #include @@ -24,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -110,6 +110,8 @@ #include #include +#include + #include "config.h" #include @@ -591,29 +593,6 @@ void sanityChecks(Server & server) } } -[[noreturn]] void backgroundMemoryThread() -{ - std::mutex mutex; - std::condition_variable cv; - - std::unique_lock lock(mutex); - while (true) - { - cv.wait_for(lock, std::chrono::microseconds(200)); - uint64_t epoch = 0; - mallctl("epoch", nullptr, nullptr, &epoch, sizeof(epoch)); - auto maybe_resident = getJemallocValue("stats.resident"); - if (!maybe_resident.has_value()) - continue; - - Int64 resident = *maybe_resident; - //LOG_INFO(getLogger("JEmalloc"), "Resident {}", ReadableSize(resident)); - MemoryTracker::setRSS(resident, false); - if (resident > total_memory_tracker.getHardLimit()) - purgeJemallocArenas(); - } -} - } void loadStartupScripts(const Poco::Util::AbstractConfiguration & config, ContextMutablePtr context, Poco::Logger * log) @@ -905,11 +884,6 @@ try total_memory_tracker.setSampleMaxAllocationSize(server_settings.total_memory_profiler_sample_max_allocation_size); } - ThreadFromGlobalPool background_memory_thread([] - { - backgroundMemoryThread(); - }); - Poco::ThreadPool server_pool( /* minCapacity */3, /* maxCapacity */server_settings.max_connections, @@ -930,16 +904,7 @@ try LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds()); }); - try - { - auto cgroups_reader = createCgroupsReader(); - global_context->setCgroupsReader(createCgroupsReader()); - } - catch (...) - { - if (server_settings.cgroups_memory_usage_observer_wait_time != 0) - tryLogCurrentException(log, "Failed to create cgroups reader"); - } + MemoryWorker memory_worker(global_context->getServerSettings().memory_worker_period_ms); /// This object will periodically calculate some metrics. ServerAsynchronousMetrics async_metrics( @@ -1500,13 +1465,15 @@ try } std::optional cgroups_memory_usage_observer; - if (auto wait_time = server_settings.cgroups_memory_usage_observer_wait_time; wait_time != 0) + try { - auto cgroups_reader = global_context->getCgroupsReader(); - if (cgroups_reader) - cgroups_memory_usage_observer.emplace(std::chrono::seconds(wait_time), std::move(cgroups_reader)); - else - LOG_ERROR(log, "Disabling cgroup memory observer because of an error during initialization of cgroups reader"); + auto wait_time = server_settings.cgroups_memory_usage_observer_wait_time; + if (wait_time != 0) + cgroups_memory_usage_observer.emplace(std::chrono::seconds(wait_time)); + } + catch (Exception &) + { + tryLogCurrentException(log, "Disabling cgroup memory observer because of an error during initialization"); } std::string cert_path = config().getString("openSSL.server.certificateFile", ""); @@ -1602,8 +1569,6 @@ try background_memory_tracker.setDescription("(background)"); background_memory_tracker.setMetric(CurrentMetrics::MergesMutationsMemoryTracking); - total_memory_tracker.setAllowUseJemallocMemory(new_server_settings.allow_use_jemalloc_memory); - auto * global_overcommit_tracker = global_context->getGlobalOvercommitTracker(); total_memory_tracker.setOvercommitTracker(global_overcommit_tracker); diff --git a/src/Common/AsynchronousMetrics.cpp b/src/Common/AsynchronousMetrics.cpp index b3e53c29d4a..a5c9875188b 100644 --- a/src/Common/AsynchronousMetrics.cpp +++ b/src/Common/AsynchronousMetrics.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -57,12 +58,10 @@ static std::unique_ptr openFileIfExists(const std::stri AsynchronousMetrics::AsynchronousMetrics( unsigned update_period_seconds, - const ProtocolServerMetricsFunc & protocol_server_metrics_func_, - std::shared_ptr cgroups_reader_) + const ProtocolServerMetricsFunc & protocol_server_metrics_func_) : update_period(update_period_seconds) , log(getLogger("AsynchronousMetrics")) , protocol_server_metrics_func(protocol_server_metrics_func_) - , cgroups_reader(std::move(cgroups_reader_)) { #if defined(OS_LINUX) openFileIfExists("/proc/meminfo", meminfo); @@ -378,23 +377,13 @@ void AsynchronousMetrics::run() namespace { -uint64_t updateJemallocEpoch() -{ - uint64_t value = 0; - size_t size = sizeof(value); - mallctl("epoch", &value, &size, &value, size); - return value; -} - template Value saveJemallocMetricImpl( AsynchronousMetricValues & values, const std::string & jemalloc_full_name, const std::string & clickhouse_full_name) { - Value value{}; - size_t size = sizeof(value); - mallctl(jemalloc_full_name.c_str(), &value, &size, nullptr, 0); + auto value = getJemallocValue(jemalloc_full_name.c_str()); values[clickhouse_full_name] = AsynchronousMetricValue(value, "An internal metric of the low-level memory allocator (jemalloc). See https://jemalloc.net/jemalloc.3.html"); return value; } @@ -604,7 +593,7 @@ void AsynchronousMetrics::update(TimePoint update_time, bool force_update) // 'epoch' is a special mallctl -- it updates the statistics. Without it, all // the following calls will return stale values. It increments and returns // the current epoch number, which might be useful to log as a sanity check. - auto epoch = updateJemallocEpoch(); + auto epoch = getJemallocValue("epoch"); new_values["jemalloc.epoch"] = { epoch, "An internal incremental update number of the statistics of jemalloc (Jason Evans' memory allocator), used in all other `jemalloc` metrics." }; // Collect the statistics themselves. diff --git a/src/Common/AsynchronousMetrics.h b/src/Common/AsynchronousMetrics.h index 0b110f41fc3..bc379d4e92b 100644 --- a/src/Common/AsynchronousMetrics.h +++ b/src/Common/AsynchronousMetrics.h @@ -69,8 +69,7 @@ public: AsynchronousMetrics( unsigned update_period_seconds, - const ProtocolServerMetricsFunc & protocol_server_metrics_func_, - std::shared_ptr cgroups_reader_); + const ProtocolServerMetricsFunc & protocol_server_metrics_func_); virtual ~AsynchronousMetrics(); diff --git a/src/Common/CgroupsMemoryUsageObserver.cpp b/src/Common/CgroupsMemoryUsageObserver.cpp index b12845df098..ab7ca69ca04 100644 --- a/src/Common/CgroupsMemoryUsageObserver.cpp +++ b/src/Common/CgroupsMemoryUsageObserver.cpp @@ -191,11 +191,9 @@ std::shared_ptr createCgroupsReader() } -CgroupsMemoryUsageObserver::CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_, std::shared_ptr cgroups_reader_) - : log(getLogger("CgroupsMemoryUsageObserver")), wait_time(wait_time_), cgroups_reader(std::move(cgroups_reader_)) -{ - cgroups_reader = createCgroupsReader(); -} +CgroupsMemoryUsageObserver::CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_) + : log(getLogger("CgroupsMemoryUsageObserver")), wait_time(wait_time_), cgroups_reader(createCgroupsReader()) +{} CgroupsMemoryUsageObserver::~CgroupsMemoryUsageObserver() { diff --git a/src/Common/CgroupsMemoryUsageObserver.h b/src/Common/CgroupsMemoryUsageObserver.h index 078307a6fa0..33e0f167a59 100644 --- a/src/Common/CgroupsMemoryUsageObserver.h +++ b/src/Common/CgroupsMemoryUsageObserver.h @@ -18,28 +18,20 @@ struct ICgroupsReader std::shared_ptr createCgroupsReader(); -/// Does two things: -/// 1. Periodically reads the memory usage of the process from Linux cgroups. -/// You can specify soft or hard memory limits: -/// - When the soft memory limit is hit, drop jemalloc cache. -/// - When the hard memory limit is hit, update MemoryTracking metric to throw memory exceptions faster. -/// The goal of this is to avoid that the process hits the maximum allowed memory limit at which there is a good -/// chance that the Limux OOM killer terminates it. All of this is done is because internal memory tracking in -/// ClickHouse can unfortunately under-estimate the actually used memory. -/// 2. Periodically reads the the maximum memory available to the process (which can change due to cgroups settings). -/// You can specify a callback to react on changes. The callback typically reloads the configuration, i.e. Server -/// or Keeper configuration file. This reloads settings 'max_server_memory_usage' (Server) and 'max_memory_usage_soft_limit' -/// (Keeper) from which various other internal limits are calculated, including the soft and hard limits for (1.). -/// The goal of this is to provide elasticity when the container is scaled-up/scaled-down. The mechanism (polling -/// cgroups) is quite implicit, unfortunately there is currently no better way to communicate memory threshold changes -/// to the database. +/// Periodically reads the the maximum memory available to the process (which can change due to cgroups settings). +/// You can specify a callback to react on changes. The callback typically reloads the configuration, i.e. Server +/// or Keeper configuration file. This reloads settings 'max_server_memory_usage' (Server) and 'max_memory_usage_soft_limit' +/// (Keeper) from which various other internal limits are calculated, including the soft and hard limits for (1.). +/// The goal of this is to provide elasticity when the container is scaled-up/scaled-down. The mechanism (polling +/// cgroups) is quite implicit, unfortunately there is currently no better way to communicate memory threshold changes +/// to the database. #if defined(OS_LINUX) class CgroupsMemoryUsageObserver { public: using OnMemoryAmountAvailableChangedFn = std::function; - explicit CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_, std::shared_ptr cgroups_reader_); + explicit CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_); ~CgroupsMemoryUsageObserver(); void setOnMemoryAmountAvailableChangedFn(OnMemoryAmountAvailableChangedFn on_memory_amount_available_changed_); diff --git a/src/Common/Jemalloc.h b/src/Common/Jemalloc.h index 0c533711f78..629d039b483 100644 --- a/src/Common/Jemalloc.h +++ b/src/Common/Jemalloc.h @@ -28,28 +28,46 @@ void setJemallocValue(const char * name, T value) { T old_value; size_t old_value_size = sizeof(T); - if (mallctl(name, &old_value, &old_value_size, reinterpret_cast(&value), sizeof(T))) - { - LOG_WARNING(getLogger("Jemalloc"), "mallctl for {} failed", name); - return; - } - + mallctl(name, &old_value, &old_value_size, reinterpret_cast(&value), sizeof(T)); LOG_INFO(getLogger("Jemalloc"), "Value for {} set to {} (from {})", name, value, old_value); } template -std::optional getJemallocValue(const char * name) +T getJemallocValue(const char * name) { T value; size_t value_size = sizeof(T); - if (mallctl(name, &value, &value_size, nullptr, 0)) - { - LOG_WARNING(getLogger("Jemalloc"), "mallctl for {} failed", name); - return std::nullopt; - } + mallctl(name, &value, &value_size, nullptr, 0); return value; } +template +struct JemallocMibCache +{ + explicit JemallocMibCache(const char * name) + { + mallctlnametomib(name, mib, &mib_length); + } + + void setValue(T value) + { + mallctlbymib(mib, mib_length, nullptr, nullptr, reinterpret_cast(&value), sizeof(T)); + } + + T getValue() + { + T value; + size_t value_size = sizeof(T); + mallctlbymib(mib, mib_length, &value, &value_size, nullptr, 0); + return value; + } + +private: + static constexpr size_t max_mib_length = 4; + size_t mib[max_mib_length]; + size_t mib_length = max_mib_length; +}; + } #endif diff --git a/src/Common/MemoryTracker.cpp b/src/Common/MemoryTracker.cpp index a541eeeb25f..e237c3a0d33 100644 --- a/src/Common/MemoryTracker.cpp +++ b/src/Common/MemoryTracker.cpp @@ -108,8 +108,6 @@ void AllocationTrace::onFreeImpl(void * ptr, size_t size) const namespace ProfileEvents { extern const Event QueryMemoryLimitExceeded; - extern const Event MemoryAllocatorPurge; - extern const Event MemoryAllocatorPurgeTimeMicroseconds; } using namespace std::chrono_literals; @@ -119,8 +117,6 @@ static constexpr size_t log_peak_memory_usage_every = 1ULL << 30; MemoryTracker total_memory_tracker(nullptr, VariableContext::Global); MemoryTracker background_memory_tracker(&total_memory_tracker, VariableContext::User, false); -std::atomic MemoryTracker::has_free_memory_in_allocator_arenas; - MemoryTracker::MemoryTracker(VariableContext level_) : parent(&total_memory_tracker), level(level_) {} MemoryTracker::MemoryTracker(MemoryTracker * parent_, VariableContext level_) : parent(parent_), level(level_) {} @@ -500,12 +496,10 @@ void MemoryTracker::reset() } -void MemoryTracker::setRSS(Int64 rss_, bool has_free_memory_in_allocator_arenas_) +void MemoryTracker::setRSS(Int64 rss_) { Int64 new_amount = rss_; - if (rss_) - total_memory_tracker.amount.store(new_amount, std::memory_order_relaxed); - has_free_memory_in_allocator_arenas.store(has_free_memory_in_allocator_arenas_, std::memory_order_relaxed); + total_memory_tracker.amount.store(new_amount, std::memory_order_relaxed); auto metric_loaded = total_memory_tracker.metric.load(std::memory_order_relaxed); if (metric_loaded != CurrentMetrics::end()) diff --git a/src/Common/MemoryTracker.h b/src/Common/MemoryTracker.h index 257ed7d0629..4085bb321ed 100644 --- a/src/Common/MemoryTracker.h +++ b/src/Common/MemoryTracker.h @@ -56,9 +56,6 @@ private: std::atomic soft_limit {0}; std::atomic hard_limit {0}; std::atomic profiler_limit {0}; - std::atomic_bool allow_use_jemalloc_memory {true}; - - static std::atomic has_free_memory_in_allocator_arenas; Int64 profiler_step = 0; @@ -153,14 +150,6 @@ public: { return soft_limit.load(std::memory_order_relaxed); } - void setAllowUseJemallocMemory(bool value) - { - allow_use_jemalloc_memory.store(value, std::memory_order_relaxed); - } - bool getAllowUseJemallocMmemory() const - { - return allow_use_jemalloc_memory.load(std::memory_order_relaxed); - } /** Set limit if it was not set. * Otherwise, set limit to new value, if new value is greater than previous limit. @@ -251,7 +240,7 @@ public: /// Reset current counter to an RSS value. /// Jemalloc may have pre-allocated arenas, they are accounted in RSS. /// We can free this arenas in case of exception to avoid OOM. - static void setRSS(Int64 rss_, bool has_free_memory_in_allocator_arenas_); + static void setRSS(Int64 rss_); /// Prints info about peak memory consumption into log. void logPeakMemoryUsage(); diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp new file mode 100644 index 00000000000..dce47b83667 --- /dev/null +++ b/src/Common/MemoryWorker.cpp @@ -0,0 +1,49 @@ +#include "Common/ThreadPool.h" +#include + +#include +#include + +namespace DB +{ + +#if USE_JEMALLOC +MemoryWorker::MemoryWorker(uint64_t period_ms_) + : period_ms(period_ms_) +{ + background_thread = ThreadFromGlobalPool([this] { backgroundThread(); }); +} + +MemoryWorker::~MemoryWorker() +{ + { + std::unique_lock lock(mutex); + shutdown = true; + } + cv.notify_all(); + + if (background_thread.joinable()) + background_thread.join(); +} + +void MemoryWorker::backgroundThread() +{ + JemallocMibCache epoch_mib("epoch"); + JemallocMibCache resident_mib("stats.resident"); + std::unique_lock lock(mutex); + while (true) + { + cv.wait_for(lock, period_ms, [this] { return shutdown; }); + if (shutdown) + return; + + epoch_mib.setValue(0); + Int64 resident = resident_mib.getValue(); + MemoryTracker::setRSS(resident); + if (resident > total_memory_tracker.getHardLimit()) + purgeJemallocArenas(); + } +} +#endif + +} diff --git a/src/Common/MemoryWorker.h b/src/Common/MemoryWorker.h new file mode 100644 index 00000000000..8048194d9cd --- /dev/null +++ b/src/Common/MemoryWorker.h @@ -0,0 +1,34 @@ +#pragma once + +#include + +#include "config.h" + +namespace DB +{ + +#if USE_JEMALLOC +class MemoryWorker +{ +public: + explicit MemoryWorker(uint64_t period_ms_); + + ~MemoryWorker(); +private: + void backgroundThread(); + + ThreadFromGlobalPool background_thread; + + std::mutex mutex; + std::condition_variable cv; + bool shutdown = false; + + std::chrono::milliseconds period_ms; +}; +#else +class MemoryWorker +{ +}; +#endif + +} diff --git a/src/Coordination/KeeperAsynchronousMetrics.cpp b/src/Coordination/KeeperAsynchronousMetrics.cpp index 3e404b7152b..86166ffe31b 100644 --- a/src/Coordination/KeeperAsynchronousMetrics.cpp +++ b/src/Coordination/KeeperAsynchronousMetrics.cpp @@ -115,7 +115,7 @@ void updateKeeperInformation(KeeperDispatcher & keeper_dispatcher, AsynchronousM KeeperAsynchronousMetrics::KeeperAsynchronousMetrics( ContextPtr context_, unsigned update_period_seconds, const ProtocolServerMetricsFunc & protocol_server_metrics_func_) - : AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_, context_->getCgroupsReader()), context(std::move(context_)) + : AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_), context(std::move(context_)) { } diff --git a/src/Core/ServerSettings.h b/src/Core/ServerSettings.h index 28b32a6e6a5..aaea0388239 100644 --- a/src/Core/ServerSettings.h +++ b/src/Core/ServerSettings.h @@ -157,6 +157,7 @@ namespace DB M(Bool, prepare_system_log_tables_on_startup, false, "If true, ClickHouse creates all configured `system.*_log` tables before the startup. It can be helpful if some startup scripts depend on these tables.", 0) \ M(Double, gwp_asan_force_sample_probability, 0.0003, "Probability that an allocation from specific places will be sampled by GWP Asan (i.e. PODArray allocations)", 0) \ M(UInt64, config_reload_interval_ms, 2000, "How often clickhouse will reload config and check for new changes", 0) \ + M(UInt64, memory_worker_period_ms, 100, "Period of background memory worker which corrects memory tracker memory usages and cleans up unused pages during higher memory usage.", 0) \ /// If you add a setting which can be updated at runtime, please update 'changeable_settings' map in StorageSystemServerSettings.cpp diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp index 6a24e049998..f70ccfd77be 100644 --- a/src/Interpreters/Context.cpp +++ b/src/Interpreters/Context.cpp @@ -406,8 +406,6 @@ struct ContextSharedPart : boost::noncopyable std::unique_ptr cluster_discovery TSA_GUARDED_BY(clusters_mutex); size_t clusters_version TSA_GUARDED_BY(clusters_mutex) = 0; - std::shared_ptr cgroups_reader; - /// No lock required for async_insert_queue modified only during initialization std::shared_ptr async_insert_queue; @@ -5631,16 +5629,6 @@ const ServerSettings & Context::getServerSettings() const return shared->server_settings; } -void Context::setCgroupsReader(std::shared_ptr cgroups_reader_) -{ - shared->cgroups_reader = std::move(cgroups_reader_); -} - -std::shared_ptr Context::getCgroupsReader() const -{ - return shared->cgroups_reader; -} - uint64_t HTTPContext::getMaxHstsAge() const { return context->getSettingsRef().hsts_max_age; diff --git a/src/Interpreters/Context.h b/src/Interpreters/Context.h index 33b742d20ad..0e3c0591c12 100644 --- a/src/Interpreters/Context.h +++ b/src/Interpreters/Context.h @@ -1344,9 +1344,6 @@ public: const ServerSettings & getServerSettings() const; - void setCgroupsReader(std::shared_ptr cgroups_reader_); - std::shared_ptr getCgroupsReader() const; - private: std::shared_ptr getSettingsConstraintsAndCurrentProfilesWithLock() const; diff --git a/src/Interpreters/ServerAsynchronousMetrics.cpp b/src/Interpreters/ServerAsynchronousMetrics.cpp index 6ee0168bede..872a9f864df 100644 --- a/src/Interpreters/ServerAsynchronousMetrics.cpp +++ b/src/Interpreters/ServerAsynchronousMetrics.cpp @@ -57,7 +57,7 @@ ServerAsynchronousMetrics::ServerAsynchronousMetrics( unsigned heavy_metrics_update_period_seconds, const ProtocolServerMetricsFunc & protocol_server_metrics_func_) : WithContext(global_context_) - , AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_, getContext()->getCgroupsReader()) + , AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_) , heavy_metric_update_period(heavy_metrics_update_period_seconds) { /// sanity check diff --git a/src/Storages/System/StorageSystemServerSettings.cpp b/src/Storages/System/StorageSystemServerSettings.cpp index d242b6de4ec..ee99c472620 100644 --- a/src/Storages/System/StorageSystemServerSettings.cpp +++ b/src/Storages/System/StorageSystemServerSettings.cpp @@ -63,7 +63,6 @@ void StorageSystemServerSettings::fillData(MutableColumns & res_columns, Context /// current setting values, one needs to ask the components directly. std::unordered_map> changeable_settings = { {"max_server_memory_usage", {std::to_string(total_memory_tracker.getHardLimit()), ChangeableWithoutRestart::Yes}}, - {"allow_use_jemalloc_memory", {std::to_string(total_memory_tracker.getAllowUseJemallocMmemory()), ChangeableWithoutRestart::Yes}}, {"max_table_size_to_drop", {std::to_string(context->getMaxTableSizeToDrop()), ChangeableWithoutRestart::Yes}}, {"max_partition_size_to_drop", {std::to_string(context->getMaxPartitionSizeToDrop()), ChangeableWithoutRestart::Yes}}, diff --git a/tests/integration/test_memory_limit/test.py b/tests/integration/test_memory_limit/test.py index 6d6745711da..db68a38c1b1 100644 --- a/tests/integration/test_memory_limit/test.py +++ b/tests/integration/test_memory_limit/test.py @@ -13,7 +13,6 @@ node = cluster.add_instance( "configs/async_metrics_no.xml", ], mem_limit="4g", - env_variables={"MALLOC_CONF": "dirty_decay_ms:0"}, ) From 9ec1fd1ab769b2f6c6ad713b4e747a89bde48b78 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Thu, 18 Jul 2024 11:29:03 +0200 Subject: [PATCH 04/16] Fix non-jemalloc builds --- src/Common/MemoryWorker.cpp | 2 +- src/Common/MemoryWorker.h | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index dce47b83667..e5ebbe3b979 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -1,8 +1,8 @@ -#include "Common/ThreadPool.h" #include #include #include +#include namespace DB { diff --git a/src/Common/MemoryWorker.h b/src/Common/MemoryWorker.h index 8048194d9cd..5f02fd0b1d0 100644 --- a/src/Common/MemoryWorker.h +++ b/src/Common/MemoryWorker.h @@ -28,6 +28,8 @@ private: #else class MemoryWorker { +public: + explicit MemoryWorker(uint64_t /*period_ms_*/) {} }; #endif From 05c7dc582a48d738bed82bcb24d3a7619fec8bc9 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Thu, 18 Jul 2024 13:40:03 +0200 Subject: [PATCH 05/16] Add some comments --- src/Common/Jemalloc.h | 3 +++ src/Common/MemoryWorker.h | 6 ++++++ 2 files changed, 9 insertions(+) diff --git a/src/Common/Jemalloc.h b/src/Common/Jemalloc.h index 629d039b483..dfa265c5e59 100644 --- a/src/Common/Jemalloc.h +++ b/src/Common/Jemalloc.h @@ -41,6 +41,9 @@ T getJemallocValue(const char * name) return value; } +/// Each mallctl call consists of string name lookup which can be expensive. +/// This can be avoided by translating name to "Management Information Base" (MIB) +/// and using it in mallctlbymib calls template struct JemallocMibCache { diff --git a/src/Common/MemoryWorker.h b/src/Common/MemoryWorker.h index 5f02fd0b1d0..6c0a578aa61 100644 --- a/src/Common/MemoryWorker.h +++ b/src/Common/MemoryWorker.h @@ -8,6 +8,12 @@ namespace DB { #if USE_JEMALLOC +/// Correct MemoryTracker based on stats.resident read from jemalloc. +/// This requires jemalloc built with --enable-stats which we use. +/// The worker spawns a background thread which moves the jemalloc epoch (updates internal stats), +/// and fetches the current stats.resident whose value is sent to global MemoryTracker. +/// Additionally, if the current memory usage is higher than global hard limit, +/// jemalloc's dirty pages are forcefully purged. class MemoryWorker { public: From 7d66f400b25a5bf0f2f43f06c6ed432d1c572fb6 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Thu, 18 Jul 2024 16:51:49 +0200 Subject: [PATCH 06/16] Better --- src/Common/Jemalloc.h | 5 +++++ src/Common/MemoryTracker.cpp | 13 ++++++++----- src/Common/MemoryTracker.h | 8 ++++---- src/Common/MemoryWorker.cpp | 27 +++++++++++++++++++++++++-- src/Common/ProfileEvents.cpp | 3 +++ 5 files changed, 45 insertions(+), 11 deletions(-) diff --git a/src/Common/Jemalloc.h b/src/Common/Jemalloc.h index dfa265c5e59..22a94a44eba 100644 --- a/src/Common/Jemalloc.h +++ b/src/Common/Jemalloc.h @@ -65,6 +65,11 @@ struct JemallocMibCache return value; } + void run() + { + mallctlbymib(mib, mib_length, nullptr, nullptr, nullptr, 0); + } + private: static constexpr size_t max_mib_length = 4; size_t mib[max_mib_length]; diff --git a/src/Common/MemoryTracker.cpp b/src/Common/MemoryTracker.cpp index e237c3a0d33..49a3a6ef7ef 100644 --- a/src/Common/MemoryTracker.cpp +++ b/src/Common/MemoryTracker.cpp @@ -242,6 +242,7 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed * So, we allow over-allocations. */ Int64 will_be = size ? size + amount.fetch_add(size, std::memory_order_relaxed) : amount.load(std::memory_order_relaxed); + Int64 will_be_rss = size + rss.load(std::memory_order_relaxed); auto metric_loaded = metric.load(std::memory_order_relaxed); if (metric_loaded != CurrentMetrics::end() && size) @@ -290,7 +291,7 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed } } - if (unlikely(current_hard_limit && will_be > current_hard_limit)) + if (unlikely(current_hard_limit && (will_be > current_hard_limit || will_be_rss > current_hard_limit))) { if (memoryTrackerCanThrow(level, false) && throw_if_memory_exceeded) { @@ -310,12 +311,13 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed throw DB::Exception( DB::ErrorCodes::MEMORY_LIMIT_EXCEEDED, "Memory limit{}{} exceeded: " - "would use {} (attempt to allocate chunk of {} bytes), maximum: {}." + "would use {} (attempt to allocate chunk of {} bytes), current RSS {}, maximum: {}." "{}{}", description ? " " : "", description ? description : "", formatReadableSizeWithBinarySuffix(will_be), size, + formatReadableSizeWithBinarySuffix(rss.load(std::memory_order_relaxed)), formatReadableSizeWithBinarySuffix(current_hard_limit), overcommit_result == OvercommitResult::NONE ? "" : " OvercommitTracker decision: ", toDescription(overcommit_result)); @@ -496,17 +498,18 @@ void MemoryTracker::reset() } -void MemoryTracker::setRSS(Int64 rss_) +void MemoryTracker::updateValues(Int64 rss_, Int64 allocated_) { - Int64 new_amount = rss_; + Int64 new_amount = allocated_; total_memory_tracker.amount.store(new_amount, std::memory_order_relaxed); + total_memory_tracker.rss.store(rss_, std::memory_order_relaxed); auto metric_loaded = total_memory_tracker.metric.load(std::memory_order_relaxed); if (metric_loaded != CurrentMetrics::end()) CurrentMetrics::set(metric_loaded, new_amount); bool log_memory_usage = true; - total_memory_tracker.updatePeak(rss_, log_memory_usage); + total_memory_tracker.updatePeak(new_amount, log_memory_usage); } diff --git a/src/Common/MemoryTracker.h b/src/Common/MemoryTracker.h index 4085bb321ed..add8bcb43d2 100644 --- a/src/Common/MemoryTracker.h +++ b/src/Common/MemoryTracker.h @@ -57,6 +57,8 @@ private: std::atomic hard_limit {0}; std::atomic profiler_limit {0}; + std::atomic rss{0}; + Int64 profiler_step = 0; /// To test exception safety of calling code, memory tracker throws an exception on each memory allocation with specified probability. @@ -237,10 +239,8 @@ public: /// Reset the accumulated data. void reset(); - /// Reset current counter to an RSS value. - /// Jemalloc may have pre-allocated arenas, they are accounted in RSS. - /// We can free this arenas in case of exception to avoid OOM. - static void setRSS(Int64 rss_); + /// update values based on external information (e.g. jemalloc's stat) + static void updateValues(Int64 rss_, Int64 allocated_); /// Prints info about peak memory consumption into log. void logPeakMemoryUsage(); diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index e5ebbe3b979..ae488a47b67 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -3,11 +3,23 @@ #include #include #include +#include + +namespace ProfileEvents +{ + extern const Event MemoryAllocatorPurge; + extern const Event MemoryAllocatorPurgeTimeMicroseconds; + extern const Event MemoryWorkerRun; + extern const Event MemoryWorkerRunElapsedMicroseconds; +} namespace DB { #if USE_JEMALLOC +#define STRINGIFY_HELPER(x) #x +#define STRINGIFY(x) STRINGIFY_HELPER(x) + MemoryWorker::MemoryWorker(uint64_t period_ms_) : period_ms(period_ms_) { @@ -30,6 +42,8 @@ void MemoryWorker::backgroundThread() { JemallocMibCache epoch_mib("epoch"); JemallocMibCache resident_mib("stats.resident"); + JemallocMibCache allocated_mib("stats.allocated"); + JemallocMibCache purge_mib("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge"); std::unique_lock lock(mutex); while (true) { @@ -37,11 +51,20 @@ void MemoryWorker::backgroundThread() if (shutdown) return; + Stopwatch total_watch; epoch_mib.setValue(0); Int64 resident = resident_mib.getValue(); - MemoryTracker::setRSS(resident); if (resident > total_memory_tracker.getHardLimit()) - purgeJemallocArenas(); + { + Stopwatch purge_watch; + purge_mib.run(); + ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurge); + ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurgeTimeMicroseconds, purge_watch.elapsedMicroseconds()); + } + + MemoryTracker::updateValues(resident, allocated_mib.getValue()); + ProfileEvents::increment(ProfileEvents::MemoryWorkerRun); + ProfileEvents::increment(ProfileEvents::MemoryWorkerRunElapsedMicroseconds, total_watch.elapsedMicroseconds()); } } #endif diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 871ba7cab8b..d85c21fcded 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -778,6 +778,9 @@ The server successfully detected this situation and will download merged part fr M(GWPAsanAllocateSuccess, "Number of successful allocations done by GWPAsan") \ M(GWPAsanAllocateFailed, "Number of failed allocations done by GWPAsan (i.e. filled pool)") \ M(GWPAsanFree, "Number of free operations done by GWPAsan") \ + \ + M(MemoryWorkerRun, "Number of runs done by MemoryWorker in background") \ + M(MemoryWorkerRunElapsedMicroseconds, "Total time spent by MemoryWorker for background work") \ #ifdef APPLY_FOR_EXTERNAL_EVENTS From 2147a96475717f0af53dd62f487c011a5b9b933a Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Sun, 21 Jul 2024 11:32:57 +0200 Subject: [PATCH 07/16] Better --- src/Common/AsynchronousMetrics.cpp | 4 +++ src/Common/MemoryTracker.cpp | 31 +++++++++++++++---- src/Common/MemoryTracker.h | 2 +- src/Common/MemoryWorker.cpp | 7 ++++- src/Coordination/KeeperDispatcher.cpp | 8 ++++- .../configs/keeper_config2.xml | 2 +- .../configs/keeper_config3.xml | 2 +- 7 files changed, 45 insertions(+), 11 deletions(-) diff --git a/src/Common/AsynchronousMetrics.cpp b/src/Common/AsynchronousMetrics.cpp index a5c9875188b..dc2f687004b 100644 --- a/src/Common/AsynchronousMetrics.cpp +++ b/src/Common/AsynchronousMetrics.cpp @@ -638,6 +638,10 @@ void AsynchronousMetrics::update(TimePoint update_time, bool force_update) "The amount of virtual memory mapped for the use of stack and for the allocated memory, in bytes." " It is unspecified whether it includes the per-thread stacks and most of the allocated memory, that is allocated with the 'mmap' system call." " This metric exists only for completeness reasons. I recommend to use the `MemoryResident` metric for monitoring."}; + +#if !USE_JEMALLOC + MemoryTracker::updateValues(data.resident, data.resident, /*force_update=*/true); +#endif } { diff --git a/src/Common/MemoryTracker.cpp b/src/Common/MemoryTracker.cpp index 49a3a6ef7ef..07d6ba98745 100644 --- a/src/Common/MemoryTracker.cpp +++ b/src/Common/MemoryTracker.cpp @@ -221,6 +221,7 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed { /// For global memory tracker always update memory usage. amount.fetch_add(size, std::memory_order_relaxed); + rss.fetch_add(size, std::memory_order_relaxed); auto metric_loaded = metric.load(std::memory_order_relaxed); if (metric_loaded != CurrentMetrics::end()) @@ -242,7 +243,7 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed * So, we allow over-allocations. */ Int64 will_be = size ? size + amount.fetch_add(size, std::memory_order_relaxed) : amount.load(std::memory_order_relaxed); - Int64 will_be_rss = size + rss.load(std::memory_order_relaxed); + Int64 will_be_rss = size ? size + rss.fetch_add(size, std::memory_order_relaxed) : rss.load(std::memory_order_relaxed); auto metric_loaded = metric.load(std::memory_order_relaxed); if (metric_loaded != CurrentMetrics::end() && size) @@ -269,6 +270,7 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed { /// Revert amount.fetch_sub(size, std::memory_order_relaxed); + rss.fetch_sub(size, std::memory_order_relaxed); /// Prevent recursion. Exception::ctor -> std::string -> new[] -> MemoryTracker::alloc MemoryTrackerBlockerInThread untrack_lock(VariableContext::Global); @@ -291,7 +293,8 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed } } - if (unlikely(current_hard_limit && (will_be > current_hard_limit || will_be_rss > current_hard_limit))) + if (unlikely( + current_hard_limit && (will_be > current_hard_limit || (level == VariableContext::Global && will_be_rss > current_hard_limit)))) { if (memoryTrackerCanThrow(level, false) && throw_if_memory_exceeded) { @@ -303,6 +306,7 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed { /// Revert amount.fetch_sub(size, std::memory_order_relaxed); + rss.fetch_sub(size, std::memory_order_relaxed); /// Prevent recursion. Exception::ctor -> std::string -> new[] -> MemoryTracker::alloc MemoryTrackerBlockerInThread untrack_lock(VariableContext::Global); @@ -411,6 +415,7 @@ AllocationTrace MemoryTracker::free(Int64 size, double _sample_probability) { /// For global memory tracker always update memory usage. amount.fetch_sub(size, std::memory_order_relaxed); + rss.fetch_sub(size, std::memory_order_relaxed); auto metric_loaded = metric.load(std::memory_order_relaxed); if (metric_loaded != CurrentMetrics::end()) CurrentMetrics::sub(metric_loaded, size); @@ -424,7 +429,12 @@ AllocationTrace MemoryTracker::free(Int64 size, double _sample_probability) } Int64 accounted_size = size; - if (level == VariableContext::Thread || level == VariableContext::Global) + if (level == VariableContext::Global) + { + amount.fetch_sub(accounted_size, std::memory_order_relaxed); + rss.fetch_sub(accounted_size, std::memory_order_relaxed); + } + else if (level == VariableContext::Thread) { /// Could become negative if memory allocated in this thread is freed in another one amount.fetch_sub(accounted_size, std::memory_order_relaxed); @@ -498,12 +508,21 @@ void MemoryTracker::reset() } -void MemoryTracker::updateValues(Int64 rss_, Int64 allocated_) +void MemoryTracker::updateValues(Int64 rss_, Int64 allocated_, bool force_update) { - Int64 new_amount = allocated_; - total_memory_tracker.amount.store(new_amount, std::memory_order_relaxed); total_memory_tracker.rss.store(rss_, std::memory_order_relaxed); + if (likely(!force_update && total_memory_tracker.amount.load(std::memory_order_relaxed) >= 0)) + return; + + Int64 new_amount = allocated_; + LOG_INFO( + getLogger("MemoryTracker"), + "Correcting the value of global memory tracker from {} to {}", + ReadableSize(total_memory_tracker.amount.load(std::memory_order_relaxed)), + ReadableSize(allocated_)); + total_memory_tracker.amount.store(new_amount, std::memory_order_relaxed); + auto metric_loaded = total_memory_tracker.metric.load(std::memory_order_relaxed); if (metric_loaded != CurrentMetrics::end()) CurrentMetrics::set(metric_loaded, new_amount); diff --git a/src/Common/MemoryTracker.h b/src/Common/MemoryTracker.h index add8bcb43d2..4913be9781f 100644 --- a/src/Common/MemoryTracker.h +++ b/src/Common/MemoryTracker.h @@ -240,7 +240,7 @@ public: void reset(); /// update values based on external information (e.g. jemalloc's stat) - static void updateValues(Int64 rss_, Int64 allocated_); + static void updateValues(Int64 rss_, Int64 allocated_, bool force_update); /// Prints info about peak memory consumption into log. void logPeakMemoryUsage(); diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index ae488a47b67..23cd90178ff 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -44,6 +44,7 @@ void MemoryWorker::backgroundThread() JemallocMibCache resident_mib("stats.resident"); JemallocMibCache allocated_mib("stats.allocated"); JemallocMibCache purge_mib("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge"); + bool first_run = false; std::unique_lock lock(mutex); while (true) { @@ -62,9 +63,13 @@ void MemoryWorker::backgroundThread() ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurgeTimeMicroseconds, purge_watch.elapsedMicroseconds()); } - MemoryTracker::updateValues(resident, allocated_mib.getValue()); + /// force update the allocated stat from jemalloc for the first run to cover the allocations we missed + /// during initialization + MemoryTracker::updateValues(resident, allocated_mib.getValue(), first_run); ProfileEvents::increment(ProfileEvents::MemoryWorkerRun); ProfileEvents::increment(ProfileEvents::MemoryWorkerRunElapsedMicroseconds, total_watch.elapsedMicroseconds()); + + first_run = false; } } #endif diff --git a/src/Coordination/KeeperDispatcher.cpp b/src/Coordination/KeeperDispatcher.cpp index 8c7e6405153..332662117d8 100644 --- a/src/Coordination/KeeperDispatcher.cpp +++ b/src/Coordination/KeeperDispatcher.cpp @@ -148,7 +148,13 @@ void KeeperDispatcher::requestThread() Int64 mem_soft_limit = keeper_context->getKeeperMemorySoftLimit(); if (configuration_and_settings->standalone_keeper && isExceedingMemorySoftLimit() && checkIfRequestIncreaseMem(request.request)) { - LOG_WARNING(log, "Processing requests refused because of max_memory_usage_soft_limit {}, the total used memory is {}, request type is {}", ReadableSize(mem_soft_limit), ReadableSize(total_memory_tracker.get()), request.request->getOpNum()); + LOG_WARNING( + log, + "Processing requests refused because of max_memory_usage_soft_limit {}, the total used memory is {}, request type " + "is {}", + ReadableSize(mem_soft_limit), + ReadableSize(total_memory_tracker.get()), + request.request->getOpNum()); addErrorResponses({request}, Coordination::Error::ZCONNECTIONLOSS); continue; } diff --git a/tests/integration/test_keeper_memory_soft_limit/configs/keeper_config2.xml b/tests/integration/test_keeper_memory_soft_limit/configs/keeper_config2.xml index 25ececea3e8..e71b93379d0 100644 --- a/tests/integration/test_keeper_memory_soft_limit/configs/keeper_config2.xml +++ b/tests/integration/test_keeper_memory_soft_limit/configs/keeper_config2.xml @@ -16,7 +16,7 @@ az-zoo2 1 - 20000000 + 200000000 10000 diff --git a/tests/integration/test_keeper_memory_soft_limit/configs/keeper_config3.xml b/tests/integration/test_keeper_memory_soft_limit/configs/keeper_config3.xml index 81e343b77c9..cf4a4686f2c 100644 --- a/tests/integration/test_keeper_memory_soft_limit/configs/keeper_config3.xml +++ b/tests/integration/test_keeper_memory_soft_limit/configs/keeper_config3.xml @@ -13,7 +13,7 @@ 2181 3 - 20000000 + 200000000 10000 From 1c3f7d0fd0fbf27692ff29d8309382eae7b7a598 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Mon, 22 Jul 2024 14:58:00 +0200 Subject: [PATCH 08/16] Small fix --- programs/server/Server.cpp | 4 ++-- src/Common/MemoryWorker.cpp | 15 ++++++++++----- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/programs/server/Server.cpp b/programs/server/Server.cpp index 04480f0bfe9..5691d82e216 100644 --- a/programs/server/Server.cpp +++ b/programs/server/Server.cpp @@ -904,8 +904,6 @@ try LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds()); }); - MemoryWorker memory_worker(global_context->getServerSettings().memory_worker_period_ms); - /// This object will periodically calculate some metrics. ServerAsynchronousMetrics async_metrics( global_context, @@ -1198,6 +1196,8 @@ try FailPointInjection::enableFromGlobalConfig(config()); + MemoryWorker memory_worker(global_context->getServerSettings().memory_worker_period_ms); + int default_oom_score = 0; #if !defined(NDEBUG) diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index 23cd90178ff..2b945a30d3d 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -2,8 +2,9 @@ #include #include -#include #include +#include +#include namespace ProfileEvents { @@ -23,6 +24,7 @@ namespace DB MemoryWorker::MemoryWorker(uint64_t period_ms_) : period_ms(period_ms_) { + LOG_INFO(getLogger("MemoryWorker"), "Starting background memory thread with period of {}ms", period_ms.count()); background_thread = ThreadFromGlobalPool([this] { backgroundThread(); }); } @@ -42,9 +44,10 @@ void MemoryWorker::backgroundThread() { JemallocMibCache epoch_mib("epoch"); JemallocMibCache resident_mib("stats.resident"); + JemallocMibCache active_mib("stats.active"); JemallocMibCache allocated_mib("stats.allocated"); JemallocMibCache purge_mib("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge"); - bool first_run = false; + bool first_run = true; std::unique_lock lock(mutex); while (true) { @@ -55,6 +58,11 @@ void MemoryWorker::backgroundThread() Stopwatch total_watch; epoch_mib.setValue(0); Int64 resident = resident_mib.getValue(); + + /// force update the allocated stat from jemalloc for the first run to cover the allocations we missed + /// during initialization + MemoryTracker::updateValues(resident, allocated_mib.getValue(), first_run); + if (resident > total_memory_tracker.getHardLimit()) { Stopwatch purge_watch; @@ -63,9 +71,6 @@ void MemoryWorker::backgroundThread() ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurgeTimeMicroseconds, purge_watch.elapsedMicroseconds()); } - /// force update the allocated stat from jemalloc for the first run to cover the allocations we missed - /// during initialization - MemoryTracker::updateValues(resident, allocated_mib.getValue(), first_run); ProfileEvents::increment(ProfileEvents::MemoryWorkerRun); ProfileEvents::increment(ProfileEvents::MemoryWorkerRunElapsedMicroseconds, total_watch.elapsedMicroseconds()); From d78cfd030fa8364456ac5283a6a1469703c53b40 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Mon, 22 Jul 2024 21:47:46 +0200 Subject: [PATCH 09/16] Use cgroups as source --- programs/keeper/Keeper.cpp | 8 +- programs/server/Server.cpp | 9 +- src/Common/AsynchronousMetrics.cpp | 32 +- src/Common/AsynchronousMetrics.h | 10 +- src/Common/CgroupsMemoryUsageObserver.cpp | 168 +---------- src/Common/CgroupsMemoryUsageObserver.h | 12 - src/Common/MemoryTracker.cpp | 9 +- src/Common/MemoryTracker.h | 3 +- src/Common/MemoryWorker.cpp | 280 ++++++++++++++++-- src/Common/MemoryWorker.h | 49 ++- .../KeeperAsynchronousMetrics.cpp | 9 +- src/Coordination/KeeperAsynchronousMetrics.h | 8 +- src/Core/ServerSettings.h | 2 +- .../ServerAsynchronousMetrics.cpp | 6 +- src/Interpreters/ServerAsynchronousMetrics.h | 5 +- 15 files changed, 367 insertions(+), 243 deletions(-) diff --git a/programs/keeper/Keeper.cpp b/programs/keeper/Keeper.cpp index b10d3f34623..d308e741311 100644 --- a/programs/keeper/Keeper.cpp +++ b/programs/keeper/Keeper.cpp @@ -376,7 +376,8 @@ try LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds()); }); - MemoryWorker memory_worker(config().getUInt64("memory_worker_period_ms", 100)); + MemoryWorker memory_worker(config().getUInt64("memory_worker_period_ms", 0)); + memory_worker.start(); static ServerErrorHandler error_handler; Poco::ErrorHandler::set(&error_handler); @@ -419,8 +420,9 @@ try for (const auto & server : *servers) metrics.emplace_back(ProtocolServerMetrics{server.getPortName(), server.currentThreads()}); return metrics; - } - ); + }, + /*update_jemalloc_epoch_=*/memory_worker.getSource() != MemoryWorker::MemoryUsageSource::Jemalloc, + /*update_rss_=*/memory_worker.getSource() == MemoryWorker::MemoryUsageSource::None); std::vector listen_hosts = DB::getMultipleValuesFromConfig(config(), "", "listen_host"); diff --git a/programs/server/Server.cpp b/programs/server/Server.cpp index 5691d82e216..1fc1df1494c 100644 --- a/programs/server/Server.cpp +++ b/programs/server/Server.cpp @@ -904,6 +904,8 @@ try LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds()); }); + MemoryWorker memory_worker(global_context->getServerSettings().memory_worker_period_ms); + /// This object will periodically calculate some metrics. ServerAsynchronousMetrics async_metrics( global_context, @@ -922,8 +924,9 @@ try for (const auto & server : servers) metrics.emplace_back(ProtocolServerMetrics{server.getPortName(), server.currentThreads()}); return metrics; - } - ); + }, + /*update_jemalloc_epoch_=*/memory_worker.getSource() != MemoryWorker::MemoryUsageSource::Jemalloc, + /*update_rss_=*/memory_worker.getSource() == MemoryWorker::MemoryUsageSource::None); /// NOTE: global context should be destroyed *before* GlobalThreadPool::shutdown() /// Otherwise GlobalThreadPool::shutdown() will hang, since Context holds some threads. @@ -1196,7 +1199,7 @@ try FailPointInjection::enableFromGlobalConfig(config()); - MemoryWorker memory_worker(global_context->getServerSettings().memory_worker_period_ms); + memory_worker.start(); int default_oom_score = 0; diff --git a/src/Common/AsynchronousMetrics.cpp b/src/Common/AsynchronousMetrics.cpp index dc2f687004b..53b8e13eaaa 100644 --- a/src/Common/AsynchronousMetrics.cpp +++ b/src/Common/AsynchronousMetrics.cpp @@ -58,10 +58,14 @@ static std::unique_ptr openFileIfExists(const std::stri AsynchronousMetrics::AsynchronousMetrics( unsigned update_period_seconds, - const ProtocolServerMetricsFunc & protocol_server_metrics_func_) + const ProtocolServerMetricsFunc & protocol_server_metrics_func_, + bool update_jemalloc_epoch_, + bool update_rss_) : update_period(update_period_seconds) , log(getLogger("AsynchronousMetrics")) , protocol_server_metrics_func(protocol_server_metrics_func_) + , update_jemalloc_epoch(update_jemalloc_epoch_) + , update_rss(update_rss_) { #if defined(OS_LINUX) openFileIfExists("/proc/meminfo", meminfo); @@ -377,6 +381,14 @@ void AsynchronousMetrics::run() namespace { +uint64_t updateJemallocEpoch() +{ + uint64_t value = 0; + size_t size = sizeof(value); + mallctl("epoch", &value, &size, &value, size); + return value; +} + template Value saveJemallocMetricImpl( AsynchronousMetricValues & values, @@ -593,8 +605,11 @@ void AsynchronousMetrics::update(TimePoint update_time, bool force_update) // 'epoch' is a special mallctl -- it updates the statistics. Without it, all // the following calls will return stale values. It increments and returns // the current epoch number, which might be useful to log as a sanity check. - auto epoch = getJemallocValue("epoch"); - new_values["jemalloc.epoch"] = { epoch, "An internal incremental update number of the statistics of jemalloc (Jason Evans' memory allocator), used in all other `jemalloc` metrics." }; + auto epoch = update_jemalloc_epoch ? updateJemallocEpoch() : getJemallocValue("epoch"); + new_values["jemalloc.epoch"] + = {epoch, + "An internal incremental update number of the statistics of jemalloc (Jason Evans' memory allocator), used in all other " + "`jemalloc` metrics."}; // Collect the statistics themselves. saveJemallocMetric(new_values, "allocated"); @@ -607,10 +622,10 @@ void AsynchronousMetrics::update(TimePoint update_time, bool force_update) saveJemallocMetric(new_values, "background_thread.num_threads"); saveJemallocMetric(new_values, "background_thread.num_runs"); saveJemallocMetric(new_values, "background_thread.run_intervals"); - saveJemallocProf(new_values, "active"); + saveJemallocProf(new_values, "active"); saveAllArenasMetric(new_values, "pactive"); - [[maybe_unused]] size_t je_malloc_pdirty = saveAllArenasMetric(new_values, "pdirty"); - [[maybe_unused]] size_t je_malloc_pmuzzy = saveAllArenasMetric(new_values, "pmuzzy"); + saveAllArenasMetric(new_values, "pdirty"); + saveAllArenasMetric(new_values, "pmuzzy"); saveAllArenasMetric(new_values, "dirty_purged"); saveAllArenasMetric(new_values, "muzzy_purged"); #endif @@ -639,9 +654,8 @@ void AsynchronousMetrics::update(TimePoint update_time, bool force_update) " It is unspecified whether it includes the per-thread stacks and most of the allocated memory, that is allocated with the 'mmap' system call." " This metric exists only for completeness reasons. I recommend to use the `MemoryResident` metric for monitoring."}; -#if !USE_JEMALLOC - MemoryTracker::updateValues(data.resident, data.resident, /*force_update=*/true); -#endif + if (update_rss) + MemoryTracker::updateRSS(data.resident); } { diff --git a/src/Common/AsynchronousMetrics.h b/src/Common/AsynchronousMetrics.h index bc379d4e92b..eb6ede7a558 100644 --- a/src/Common/AsynchronousMetrics.h +++ b/src/Common/AsynchronousMetrics.h @@ -7,10 +7,8 @@ #include #include -#include #include #include -#include #include #include #include @@ -69,7 +67,9 @@ public: AsynchronousMetrics( unsigned update_period_seconds, - const ProtocolServerMetricsFunc & protocol_server_metrics_func_); + const ProtocolServerMetricsFunc & protocol_server_metrics_func_, + bool update_jemalloc_epoch_, + bool update_rss_); virtual ~AsynchronousMetrics(); @@ -92,7 +92,6 @@ private: virtual void logImpl(AsynchronousMetricValues &) {} ProtocolServerMetricsFunc protocol_server_metrics_func; - std::shared_ptr cgroups_reader; std::unique_ptr thread; @@ -113,6 +112,9 @@ private: MemoryStatisticsOS memory_stat TSA_GUARDED_BY(data_mutex); #endif + const bool update_jemalloc_epoch; + const bool update_rss; + #if defined(OS_LINUX) std::optional meminfo TSA_GUARDED_BY(data_mutex); std::optional loadavg TSA_GUARDED_BY(data_mutex); diff --git a/src/Common/CgroupsMemoryUsageObserver.cpp b/src/Common/CgroupsMemoryUsageObserver.cpp index ab7ca69ca04..afeac1808b2 100644 --- a/src/Common/CgroupsMemoryUsageObserver.cpp +++ b/src/Common/CgroupsMemoryUsageObserver.cpp @@ -13,12 +13,8 @@ #include #include -#include -#include -#include using namespace DB; -namespace fs = std::filesystem; namespace DB { @@ -29,170 +25,8 @@ extern const int FILE_DOESNT_EXIST; extern const int INCORRECT_DATA; } -} - -namespace -{ - -/// Format is -/// kernel 5 -/// rss 15 -/// [...] -uint64_t readMetricFromStatFile(ReadBufferFromFile & buf, const std::string & key) -{ - while (!buf.eof()) - { - std::string current_key; - readStringUntilWhitespace(current_key, buf); - if (current_key != key) - { - std::string dummy; - readStringUntilNewlineInto(dummy, buf); - buf.ignore(); - continue; - } - - assertChar(' ', buf); - uint64_t value = 0; - readIntText(value, buf); - return value; - } - - throw Exception(ErrorCodes::INCORRECT_DATA, "Cannot find '{}' in '{}'", key, buf.getFileName()); -} - -struct CgroupsV1Reader : ICgroupsReader -{ - explicit CgroupsV1Reader(const fs::path & stat_file_dir) : buf(stat_file_dir / "memory.stat") { } - - uint64_t readMemoryUsage() override - { - std::lock_guard lock(mutex); - buf.rewind(); - return readMetricFromStatFile(buf, "rss"); - } - -private: - std::mutex mutex; - ReadBufferFromFile buf TSA_GUARDED_BY(mutex); -}; - -struct CgroupsV2Reader : ICgroupsReader -{ - explicit CgroupsV2Reader(const fs::path & stat_file_dir) - : current_buf(stat_file_dir / "memory.current"), stat_buf(stat_file_dir / "memory.stat") - { - } - - uint64_t readMemoryUsage() override - { - std::lock_guard lock(mutex); - current_buf.rewind(); - stat_buf.rewind(); - - int64_t mem_usage = 0; - /// memory.current contains a single number - /// the reason why we subtract it described here: https://github.com/ClickHouse/ClickHouse/issues/64652#issuecomment-2149630667 - readIntText(mem_usage, current_buf); - mem_usage -= readMetricFromStatFile(stat_buf, "inactive_file"); - chassert(mem_usage >= 0, "Negative memory usage"); - return mem_usage; - } - -private: - std::mutex mutex; - ReadBufferFromFile current_buf TSA_GUARDED_BY(mutex); - ReadBufferFromFile stat_buf TSA_GUARDED_BY(mutex); -}; - -/// Caveats: -/// - All of the logic in this file assumes that the current process is the only process in the -/// containing cgroup (or more precisely: the only process with significant memory consumption). -/// If this is not the case, then other processe's memory consumption may affect the internal -/// memory tracker ... -/// - Cgroups v1 and v2 allow nested cgroup hierarchies. As v1 is deprecated for over half a -/// decade and will go away at some point, hierarchical detection is only implemented for v2. -/// - I did not test what happens if a host has v1 and v2 simultaneously enabled. I believe such -/// systems existed only for a short transition period. - -std::optional getCgroupsV2Path() -{ - if (!cgroupsV2Enabled()) - return {}; - - if (!cgroupsV2MemoryControllerEnabled()) - return {}; - - fs::path current_cgroup = cgroupV2PathOfProcess(); - if (current_cgroup.empty()) - return {}; - - /// Return the bottom-most nested current memory file. If there is no such file at the current - /// level, try again at the parent level as memory settings are inherited. - while (current_cgroup != default_cgroups_mount.parent_path()) - { - const auto current_path = current_cgroup / "memory.current"; - const auto stat_path = current_cgroup / "memory.stat"; - if (fs::exists(current_path) && fs::exists(stat_path)) - return {current_cgroup}; - current_cgroup = current_cgroup.parent_path(); - } - return {}; -} - -std::optional getCgroupsV1Path() -{ - auto path = default_cgroups_mount / "memory/memory.stat"; - if (!fs::exists(path)) - return {}; - return {default_cgroups_mount / "memory"}; -} - -enum class CgroupsVersion : uint8_t -{ - V1, - V2 -}; - -std::pair getCgroupsPath() -{ - auto v2_path = getCgroupsV2Path(); - if (v2_path.has_value()) - return {*v2_path, CgroupsVersion::V2}; - - auto v1_path = getCgroupsV1Path(); - if (v1_path.has_value()) - return {*v1_path, CgroupsVersion::V1}; - - throw Exception(ErrorCodes::FILE_DOESNT_EXIST, "Cannot find cgroups v1 or v2 current memory file"); -} - -} - -namespace DB -{ - -std::shared_ptr createCgroupsReader() -{ - const auto [cgroup_path, version] = getCgroupsPath(); - LOG_INFO( - getLogger("CgroupsReader"), - "Will create cgroup reader from '{}' (cgroups version: {})", - cgroup_path, - (version == CgroupsVersion::V1) ? "v1" : "v2"); - - if (version == CgroupsVersion::V2) - return std::make_shared(cgroup_path); - else - { - chassert(version == CgroupsVersion::V1); - return std::make_shared(cgroup_path); - } - -} - CgroupsMemoryUsageObserver::CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_) - : log(getLogger("CgroupsMemoryUsageObserver")), wait_time(wait_time_), cgroups_reader(createCgroupsReader()) + : log(getLogger("CgroupsMemoryUsageObserver")), wait_time(wait_time_) {} CgroupsMemoryUsageObserver::~CgroupsMemoryUsageObserver() diff --git a/src/Common/CgroupsMemoryUsageObserver.h b/src/Common/CgroupsMemoryUsageObserver.h index 33e0f167a59..3de83d6b437 100644 --- a/src/Common/CgroupsMemoryUsageObserver.h +++ b/src/Common/CgroupsMemoryUsageObserver.h @@ -3,21 +3,11 @@ #include #include -#include #include namespace DB { -struct ICgroupsReader -{ - virtual ~ICgroupsReader() = default; - - virtual uint64_t readMemoryUsage() = 0; -}; - -std::shared_ptr createCgroupsReader(); - /// Periodically reads the the maximum memory available to the process (which can change due to cgroups settings). /// You can specify a callback to react on changes. The callback typically reloads the configuration, i.e. Server /// or Keeper configuration file. This reloads settings 'max_server_memory_usage' (Server) and 'max_memory_usage_soft_limit' @@ -54,8 +44,6 @@ private: void runThread(); - std::shared_ptr cgroups_reader; - std::mutex thread_mutex; std::condition_variable cond; ThreadFromGlobalPool thread; diff --git a/src/Common/MemoryTracker.cpp b/src/Common/MemoryTracker.cpp index 07d6ba98745..0ffae89ffa6 100644 --- a/src/Common/MemoryTracker.cpp +++ b/src/Common/MemoryTracker.cpp @@ -508,13 +508,13 @@ void MemoryTracker::reset() } -void MemoryTracker::updateValues(Int64 rss_, Int64 allocated_, bool force_update) +void MemoryTracker::updateRSS(Int64 rss_) { total_memory_tracker.rss.store(rss_, std::memory_order_relaxed); +} - if (likely(!force_update && total_memory_tracker.amount.load(std::memory_order_relaxed) >= 0)) - return; - +void MemoryTracker::updateAllocated(Int64 allocated_) +{ Int64 new_amount = allocated_; LOG_INFO( getLogger("MemoryTracker"), @@ -531,7 +531,6 @@ void MemoryTracker::updateValues(Int64 rss_, Int64 allocated_, bool force_update total_memory_tracker.updatePeak(new_amount, log_memory_usage); } - void MemoryTracker::setSoftLimit(Int64 value) { soft_limit.store(value, std::memory_order_relaxed); diff --git a/src/Common/MemoryTracker.h b/src/Common/MemoryTracker.h index 4913be9781f..d2db8489f19 100644 --- a/src/Common/MemoryTracker.h +++ b/src/Common/MemoryTracker.h @@ -240,7 +240,8 @@ public: void reset(); /// update values based on external information (e.g. jemalloc's stat) - static void updateValues(Int64 rss_, Int64 allocated_, bool force_update); + static void updateRSS(Int64 rss_); + static void updateAllocated(Int64 allocated_); /// Prints info about peak memory consumption into log. void logPeakMemoryUsage(); diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index 2b945a30d3d..42e797a80d6 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -1,11 +1,21 @@ #include +#include +#include +#include +#include #include #include #include #include #include +#include +#include +#include + +namespace fs = std::filesystem; + namespace ProfileEvents { extern const Event MemoryAllocatorPurge; @@ -17,14 +27,227 @@ namespace ProfileEvents namespace DB { -#if USE_JEMALLOC -#define STRINGIFY_HELPER(x) #x -#define STRINGIFY(x) STRINGIFY_HELPER(x) - -MemoryWorker::MemoryWorker(uint64_t period_ms_) - : period_ms(period_ms_) +namespace ErrorCodes { - LOG_INFO(getLogger("MemoryWorker"), "Starting background memory thread with period of {}ms", period_ms.count()); + extern const int FILE_DOESNT_EXIST; + extern const int INCORRECT_DATA; +} + +#if defined(OS_LINUX) +struct ICgroupsReader +{ + virtual ~ICgroupsReader() = default; + + virtual uint64_t readMemoryUsage() = 0; +}; + +namespace +{ + +/// Format is +/// kernel 5 +/// rss 15 +/// [...] +uint64_t readMetricFromStatFile(ReadBufferFromFile & buf, const std::string & key) +{ + while (!buf.eof()) + { + std::string current_key; + readStringUntilWhitespace(current_key, buf); + if (current_key != key) + { + std::string dummy; + readStringUntilNewlineInto(dummy, buf); + buf.ignore(); + continue; + } + + assertChar(' ', buf); + uint64_t value = 0; + readIntText(value, buf); + return value; + } + + LOG_ERROR(getLogger("CgroupsReader"), "Cannot find '{}' in '{}'", key, buf.getFileName()); + return 0; +} + +struct CgroupsV1Reader : ICgroupsReader +{ + explicit CgroupsV1Reader(const fs::path & stat_file_dir) : buf(stat_file_dir / "memory.stat") { } + + uint64_t readMemoryUsage() override + { + std::lock_guard lock(mutex); + buf.rewind(); + return readMetricFromStatFile(buf, "rss"); + } + +private: + std::mutex mutex; + ReadBufferFromFile buf TSA_GUARDED_BY(mutex); +}; + +struct CgroupsV2Reader : ICgroupsReader +{ + explicit CgroupsV2Reader(const fs::path & stat_file_dir) : stat_buf(stat_file_dir / "memory.stat") { } + + uint64_t readMemoryUsage() override + { + std::lock_guard lock(mutex); + stat_buf.rewind(); + return readMetricFromStatFile(stat_buf, "anon"); + } + +private: + std::mutex mutex; + ReadBufferFromFile stat_buf TSA_GUARDED_BY(mutex); +}; + +/// Caveats: +/// - All of the logic in this file assumes that the current process is the only process in the +/// containing cgroup (or more precisely: the only process with significant memory consumption). +/// If this is not the case, then other processe's memory consumption may affect the internal +/// memory tracker ... +/// - Cgroups v1 and v2 allow nested cgroup hierarchies. As v1 is deprecated for over half a +/// decade and will go away at some point, hierarchical detection is only implemented for v2. +/// - I did not test what happens if a host has v1 and v2 simultaneously enabled. I believe such +/// systems existed only for a short transition period. + +std::optional getCgroupsV2Path() +{ + if (!cgroupsV2Enabled()) + return {}; + + if (!cgroupsV2MemoryControllerEnabled()) + return {}; + + fs::path current_cgroup = cgroupV2PathOfProcess(); + if (current_cgroup.empty()) + return {}; + + /// Return the bottom-most nested current memory file. If there is no such file at the current + /// level, try again at the parent level as memory settings are inherited. + while (current_cgroup != default_cgroups_mount.parent_path()) + { + const auto current_path = current_cgroup / "memory.current"; + const auto stat_path = current_cgroup / "memory.stat"; + if (fs::exists(current_path) && fs::exists(stat_path)) + return {current_cgroup}; + current_cgroup = current_cgroup.parent_path(); + } + return {}; +} + +std::optional getCgroupsV1Path() +{ + auto path = default_cgroups_mount / "memory/memory.stat"; + if (!fs::exists(path)) + return {}; + return {default_cgroups_mount / "memory"}; +} + +enum class CgroupsVersion : uint8_t +{ + V1, + V2 +}; + +std::pair getCgroupsPath() +{ + auto v2_path = getCgroupsV2Path(); + if (v2_path.has_value()) + return {*v2_path, CgroupsVersion::V2}; + + auto v1_path = getCgroupsV1Path(); + if (v1_path.has_value()) + return {*v1_path, CgroupsVersion::V1}; + + throw Exception(ErrorCodes::FILE_DOESNT_EXIST, "Cannot find cgroups v1 or v2 current memory file"); +} + +std::shared_ptr createCgroupsReader() +{ + const auto [cgroup_path, version] = getCgroupsPath(); + LOG_INFO( + getLogger("CgroupsReader"), + "Will create cgroup reader from '{}' (cgroups version: {})", + cgroup_path, + (version == CgroupsVersion::V1) ? "v1" : "v2"); + + if (version == CgroupsVersion::V2) + return std::make_shared(cgroup_path); + else + { + chassert(version == CgroupsVersion::V1); + return std::make_shared(cgroup_path); + } + +} +#endif + +constexpr uint64_t cgroups_memory_usage_tick_ms{50}; +constexpr uint64_t jemalloc_memory_usage_tick_ms{100}; + +std::string_view sourceToString(MemoryWorker::MemoryUsageSource source) +{ + switch (source) + { + case MemoryWorker::MemoryUsageSource::Cgroups: return "Cgroups"; + case MemoryWorker::MemoryUsageSource::Jemalloc: return "Jemalloc"; + case MemoryWorker::MemoryUsageSource::None: return "None"; + } +} + +} + +/// We try to pick the best possible supported source for reading memory usage. +/// Supported sources in order of priority +/// - reading from cgroups' pseudo-files (fastest and most accurate) +/// - reading jemalloc's resident stat (doesn't take into account allocations that didn't use jemalloc) +/// Also, different tick rates are used because not all options are equally fast +MemoryWorker::MemoryWorker(uint64_t period_ms_) + : log(getLogger("MemoryWorker")) + , period_ms(period_ms_) +{ +#if defined(OS_LINUX) + try + { + cgroups_reader = createCgroupsReader(); + source = MemoryUsageSource::Cgroups; + if (period_ms == 0) + period_ms = cgroups_memory_usage_tick_ms; + + return; + } + catch (...) + { + tryLogCurrentException(log, "Cannot use cgroups reader"); + } +#endif + +#if USE_JEMALLOC + source = MemoryUsageSource::Jemalloc; + if (period_ms == 0) + period_ms = jemalloc_memory_usage_tick_ms; +#endif +} + +MemoryWorker::MemoryUsageSource MemoryWorker::getSource() +{ + return source; +} + +void MemoryWorker::start() +{ + if (source == MemoryUsageSource::None) + return; + + LOG_INFO( + getLogger("MemoryWorker"), + "Starting background memory thread with period of {}ms, using {} as source", + period_ms, + sourceToString(source)); background_thread = ThreadFromGlobalPool([this] { backgroundThread(); }); } @@ -40,29 +263,39 @@ MemoryWorker::~MemoryWorker() background_thread.join(); } +uint64_t MemoryWorker::getMemoryUsage() +{ + switch (source) + { + case MemoryUsageSource::Cgroups: + return cgroups_reader->readMemoryUsage(); + case MemoryUsageSource::Jemalloc: + return resident_mib.getValue(); + case MemoryUsageSource::None: + throw DB::Exception(ErrorCodes::LOGICAL_ERROR, "Trying to fetch memory usage while no memory source can be used"); + } +} + void MemoryWorker::backgroundThread() { - JemallocMibCache epoch_mib("epoch"); - JemallocMibCache resident_mib("stats.resident"); - JemallocMibCache active_mib("stats.active"); - JemallocMibCache allocated_mib("stats.allocated"); - JemallocMibCache purge_mib("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge"); + std::chrono::milliseconds chrono_period_ms{period_ms}; bool first_run = true; std::unique_lock lock(mutex); while (true) { - cv.wait_for(lock, period_ms, [this] { return shutdown; }); + cv.wait_for(lock, chrono_period_ms, [this] { return shutdown; }); if (shutdown) return; Stopwatch total_watch; - epoch_mib.setValue(0); - Int64 resident = resident_mib.getValue(); - /// force update the allocated stat from jemalloc for the first run to cover the allocations we missed - /// during initialization - MemoryTracker::updateValues(resident, allocated_mib.getValue(), first_run); + if (source == MemoryUsageSource::Jemalloc) + epoch_mib.setValue(0); + Int64 resident = getMemoryUsage(); + MemoryTracker::updateRSS(resident); + +#if USE_JEMALLOC if (resident > total_memory_tracker.getHardLimit()) { Stopwatch purge_watch; @@ -71,12 +304,19 @@ void MemoryWorker::backgroundThread() ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurgeTimeMicroseconds, purge_watch.elapsedMicroseconds()); } + if (unlikely(first_run || total_memory_tracker.get() < 0)) + { + if (source != MemoryUsageSource::Jemalloc) + epoch_mib.setValue(0); + + MemoryTracker::updateAllocated(allocated_mib.getValue()); + } +#endif + ProfileEvents::increment(ProfileEvents::MemoryWorkerRun); ProfileEvents::increment(ProfileEvents::MemoryWorkerRunElapsedMicroseconds, total_watch.elapsedMicroseconds()); - first_run = false; } } -#endif } diff --git a/src/Common/MemoryWorker.h b/src/Common/MemoryWorker.h index 6c0a578aa61..6fde93d63ad 100644 --- a/src/Common/MemoryWorker.h +++ b/src/Common/MemoryWorker.h @@ -1,13 +1,14 @@ #pragma once +#include #include - -#include "config.h" +#include namespace DB { -#if USE_JEMALLOC +struct ICgroupsReader; + /// Correct MemoryTracker based on stats.resident read from jemalloc. /// This requires jemalloc built with --enable-stats which we use. /// The worker spawns a background thread which moves the jemalloc epoch (updates internal stats), @@ -19,8 +20,21 @@ class MemoryWorker public: explicit MemoryWorker(uint64_t period_ms_); + enum class MemoryUsageSource : uint8_t + { + None, + Cgroups, + Jemalloc + }; + + MemoryUsageSource getSource(); + + void start(); + ~MemoryWorker(); private: + uint64_t getMemoryUsage(); + void backgroundThread(); ThreadFromGlobalPool background_thread; @@ -29,14 +43,27 @@ private: std::condition_variable cv; bool shutdown = false; - std::chrono::milliseconds period_ms; -}; -#else -class MemoryWorker -{ -public: - explicit MemoryWorker(uint64_t /*period_ms_*/) {} -}; + LoggerPtr log; + + uint64_t period_ms; + + MemoryUsageSource source{MemoryUsageSource::None}; + +#if defined(OS_LINUX) + std::shared_ptr cgroups_reader; #endif +#if USE_JEMALLOC + JemallocMibCache epoch_mib{"epoch"}; + JemallocMibCache resident_mib{"stats.resident"}; + JemallocMibCache allocated_mib{"stats.allocated"}; + +#define STRINGIFY_HELPER(x) #x +#define STRINGIFY(x) STRINGIFY_HELPER(x) + JemallocMibCache purge_mib{"arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge"}; +#undef STRINGIFY +#undef STRINGIFY_HELPER +#endif +}; + } diff --git a/src/Coordination/KeeperAsynchronousMetrics.cpp b/src/Coordination/KeeperAsynchronousMetrics.cpp index 86166ffe31b..157858f3c44 100644 --- a/src/Coordination/KeeperAsynchronousMetrics.cpp +++ b/src/Coordination/KeeperAsynchronousMetrics.cpp @@ -114,8 +114,13 @@ void updateKeeperInformation(KeeperDispatcher & keeper_dispatcher, AsynchronousM } KeeperAsynchronousMetrics::KeeperAsynchronousMetrics( - ContextPtr context_, unsigned update_period_seconds, const ProtocolServerMetricsFunc & protocol_server_metrics_func_) - : AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_), context(std::move(context_)) + ContextPtr context_, + unsigned update_period_seconds, + const ProtocolServerMetricsFunc & protocol_server_metrics_func_, + bool update_jemalloc_epoch_, + bool update_rss_) + : AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_, update_jemalloc_epoch_, update_rss_) + , context(std::move(context_)) { } diff --git a/src/Coordination/KeeperAsynchronousMetrics.h b/src/Coordination/KeeperAsynchronousMetrics.h index ec0e60cbb6e..a2ab7cab756 100644 --- a/src/Coordination/KeeperAsynchronousMetrics.h +++ b/src/Coordination/KeeperAsynchronousMetrics.h @@ -13,9 +13,13 @@ class KeeperAsynchronousMetrics : public AsynchronousMetrics { public: KeeperAsynchronousMetrics( - ContextPtr context_, unsigned update_period_seconds, const ProtocolServerMetricsFunc & protocol_server_metrics_func_); - ~KeeperAsynchronousMetrics() override; + ContextPtr context_, + unsigned update_period_seconds, + const ProtocolServerMetricsFunc & protocol_server_metrics_func_, + bool update_jemalloc_epoch_, + bool update_rss_); + ~KeeperAsynchronousMetrics() override; private: ContextPtr context; diff --git a/src/Core/ServerSettings.h b/src/Core/ServerSettings.h index aaea0388239..ea5a3f19638 100644 --- a/src/Core/ServerSettings.h +++ b/src/Core/ServerSettings.h @@ -157,7 +157,7 @@ namespace DB M(Bool, prepare_system_log_tables_on_startup, false, "If true, ClickHouse creates all configured `system.*_log` tables before the startup. It can be helpful if some startup scripts depend on these tables.", 0) \ M(Double, gwp_asan_force_sample_probability, 0.0003, "Probability that an allocation from specific places will be sampled by GWP Asan (i.e. PODArray allocations)", 0) \ M(UInt64, config_reload_interval_ms, 2000, "How often clickhouse will reload config and check for new changes", 0) \ - M(UInt64, memory_worker_period_ms, 100, "Period of background memory worker which corrects memory tracker memory usages and cleans up unused pages during higher memory usage.", 0) \ + M(UInt64, memory_worker_period_ms, 0, "Tick period of background memory worker which corrects memory tracker memory usages and cleans up unused pages during higher memory usage. If set to 0, default value will be used depending on the memory usage source", 0) \ /// If you add a setting which can be updated at runtime, please update 'changeable_settings' map in StorageSystemServerSettings.cpp diff --git a/src/Interpreters/ServerAsynchronousMetrics.cpp b/src/Interpreters/ServerAsynchronousMetrics.cpp index 872a9f864df..079029695c9 100644 --- a/src/Interpreters/ServerAsynchronousMetrics.cpp +++ b/src/Interpreters/ServerAsynchronousMetrics.cpp @@ -55,9 +55,11 @@ ServerAsynchronousMetrics::ServerAsynchronousMetrics( ContextPtr global_context_, unsigned update_period_seconds, unsigned heavy_metrics_update_period_seconds, - const ProtocolServerMetricsFunc & protocol_server_metrics_func_) + const ProtocolServerMetricsFunc & protocol_server_metrics_func_, + bool update_jemalloc_epoch_, + bool update_rss_) : WithContext(global_context_) - , AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_) + , AsynchronousMetrics(update_period_seconds, protocol_server_metrics_func_, update_jemalloc_epoch_, update_rss_) , heavy_metric_update_period(heavy_metrics_update_period_seconds) { /// sanity check diff --git a/src/Interpreters/ServerAsynchronousMetrics.h b/src/Interpreters/ServerAsynchronousMetrics.h index e3c83dc748e..5fab419a32b 100644 --- a/src/Interpreters/ServerAsynchronousMetrics.h +++ b/src/Interpreters/ServerAsynchronousMetrics.h @@ -14,7 +14,10 @@ public: ContextPtr global_context_, unsigned update_period_seconds, unsigned heavy_metrics_update_period_seconds, - const ProtocolServerMetricsFunc & protocol_server_metrics_func_); + const ProtocolServerMetricsFunc & protocol_server_metrics_func_, + bool update_jemalloc_epoch_, + bool update_rss_); + ~ServerAsynchronousMetrics() override; private: From 5a1b96ac8453f73de5e891e3a9f235fd96270b50 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Tue, 23 Jul 2024 10:52:14 +0200 Subject: [PATCH 10/16] Style fix --- src/Common/CgroupsMemoryUsageObserver.cpp | 6 ------ src/Common/MemoryWorker.cpp | 2 +- 2 files changed, 1 insertion(+), 7 deletions(-) diff --git a/src/Common/CgroupsMemoryUsageObserver.cpp b/src/Common/CgroupsMemoryUsageObserver.cpp index afeac1808b2..16d5d1cccde 100644 --- a/src/Common/CgroupsMemoryUsageObserver.cpp +++ b/src/Common/CgroupsMemoryUsageObserver.cpp @@ -19,12 +19,6 @@ using namespace DB; namespace DB { -namespace ErrorCodes -{ -extern const int FILE_DOESNT_EXIST; -extern const int INCORRECT_DATA; -} - CgroupsMemoryUsageObserver::CgroupsMemoryUsageObserver(std::chrono::seconds wait_time_) : log(getLogger("CgroupsMemoryUsageObserver")), wait_time(wait_time_) {} diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index 42e797a80d6..ddc3fd783f4 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -30,7 +30,7 @@ namespace DB namespace ErrorCodes { extern const int FILE_DOESNT_EXIST; - extern const int INCORRECT_DATA; + extern const int LOGICAL_ERROR; } #if defined(OS_LINUX) From 5b51a35e015336227dcc5b25445fefbc3da3059c Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Tue, 23 Jul 2024 11:31:34 +0200 Subject: [PATCH 11/16] Add unused variables --- src/Common/AsynchronousMetrics.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Common/AsynchronousMetrics.h b/src/Common/AsynchronousMetrics.h index eb6ede7a558..fedba4e55be 100644 --- a/src/Common/AsynchronousMetrics.h +++ b/src/Common/AsynchronousMetrics.h @@ -112,8 +112,8 @@ private: MemoryStatisticsOS memory_stat TSA_GUARDED_BY(data_mutex); #endif - const bool update_jemalloc_epoch; - const bool update_rss; + [[maybe_unused]] const bool update_jemalloc_epoch; + [[maybe_unused]] const bool update_rss; #if defined(OS_LINUX) std::optional meminfo TSA_GUARDED_BY(data_mutex); From ade79cfd7a8465bf6e332f3b2ca6143ba652251a Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Tue, 23 Jul 2024 11:50:56 +0200 Subject: [PATCH 12/16] More fixes --- src/Common/MemoryWorker.cpp | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index ddc3fd783f4..8169bc7d177 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -186,9 +186,6 @@ std::shared_ptr createCgroupsReader() } #endif -constexpr uint64_t cgroups_memory_usage_tick_ms{50}; -constexpr uint64_t jemalloc_memory_usage_tick_ms{100}; - std::string_view sourceToString(MemoryWorker::MemoryUsageSource source) { switch (source) @@ -213,6 +210,8 @@ MemoryWorker::MemoryWorker(uint64_t period_ms_) #if defined(OS_LINUX) try { + static constexpr uint64_t cgroups_memory_usage_tick_ms{50}; + cgroups_reader = createCgroupsReader(); source = MemoryUsageSource::Cgroups; if (period_ms == 0) @@ -227,6 +226,8 @@ MemoryWorker::MemoryWorker(uint64_t period_ms_) #endif #if USE_JEMALLOC + static constexpr uint64_t jemalloc_memory_usage_tick_ms{100}; + source = MemoryUsageSource::Jemalloc; if (period_ms == 0) period_ms = jemalloc_memory_usage_tick_ms; @@ -270,7 +271,11 @@ uint64_t MemoryWorker::getMemoryUsage() case MemoryUsageSource::Cgroups: return cgroups_reader->readMemoryUsage(); case MemoryUsageSource::Jemalloc: +#if USE_JEMALLOC return resident_mib.getValue(); +#else + return 0; +#endif case MemoryUsageSource::None: throw DB::Exception(ErrorCodes::LOGICAL_ERROR, "Trying to fetch memory usage while no memory source can be used"); } @@ -279,7 +284,7 @@ uint64_t MemoryWorker::getMemoryUsage() void MemoryWorker::backgroundThread() { std::chrono::milliseconds chrono_period_ms{period_ms}; - bool first_run = true; + [[maybe_unused]] bool first_run = true; std::unique_lock lock(mutex); while (true) { @@ -289,8 +294,10 @@ void MemoryWorker::backgroundThread() Stopwatch total_watch; +#if USE_JEMALLOC if (source == MemoryUsageSource::Jemalloc) epoch_mib.setValue(0); +#endif Int64 resident = getMemoryUsage(); MemoryTracker::updateRSS(resident); From 04d80ec2763a0677f32e9f6190cf2bcda0ebf8b7 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Tue, 23 Jul 2024 16:13:07 +0200 Subject: [PATCH 13/16] Fix non-linux build --- src/Common/MemoryWorker.cpp | 4 ++-- src/Common/MemoryWorker.h | 2 -- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index 8169bc7d177..c576772d303 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -33,7 +33,6 @@ namespace ErrorCodes extern const int LOGICAL_ERROR; } -#if defined(OS_LINUX) struct ICgroupsReader { virtual ~ICgroupsReader() = default; @@ -44,6 +43,7 @@ struct ICgroupsReader namespace { +#if defined(OS_LINUX) /// Format is /// kernel 5 /// rss 15 @@ -269,7 +269,7 @@ uint64_t MemoryWorker::getMemoryUsage() switch (source) { case MemoryUsageSource::Cgroups: - return cgroups_reader->readMemoryUsage(); + return cgroups_reader != nullptr ? cgroups_reader->readMemoryUsage() : 0; case MemoryUsageSource::Jemalloc: #if USE_JEMALLOC return resident_mib.getValue(); diff --git a/src/Common/MemoryWorker.h b/src/Common/MemoryWorker.h index 6fde93d63ad..b1b0495bf14 100644 --- a/src/Common/MemoryWorker.h +++ b/src/Common/MemoryWorker.h @@ -49,9 +49,7 @@ private: MemoryUsageSource source{MemoryUsageSource::None}; -#if defined(OS_LINUX) std::shared_ptr cgroups_reader; -#endif #if USE_JEMALLOC JemallocMibCache epoch_mib{"epoch"}; From ec5459a60d0be15e7bc100bd257e2b5c65baf594 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Wed, 24 Jul 2024 09:15:34 +0200 Subject: [PATCH 14/16] Update allocated with resident if no jemalloc --- src/Common/MemoryWorker.cpp | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index c576772d303..75d0e7c32d8 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -310,15 +310,19 @@ void MemoryWorker::backgroundThread() ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurge); ProfileEvents::increment(ProfileEvents::MemoryAllocatorPurgeTimeMicroseconds, purge_watch.elapsedMicroseconds()); } +#endif if (unlikely(first_run || total_memory_tracker.get() < 0)) { +#if USE_JEMALLOC if (source != MemoryUsageSource::Jemalloc) epoch_mib.setValue(0); MemoryTracker::updateAllocated(allocated_mib.getValue()); - } +#elif defined(OS_LINUX) + MemoryTracker::updateAllocated(resident); #endif + } ProfileEvents::increment(ProfileEvents::MemoryWorkerRun); ProfileEvents::increment(ProfileEvents::MemoryWorkerRunElapsedMicroseconds, total_watch.elapsedMicroseconds()); From e2e4c8ee0f8fcf1d6ef4d566c2fa4f9ee2123a56 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Wed, 24 Jul 2024 10:21:09 +0200 Subject: [PATCH 15/16] Better --- src/Common/MemoryWorker.cpp | 22 ++++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index 84ccffb8e90..e148f7f8f49 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -65,11 +65,25 @@ Metrics readAllMetricsFromStatFile(ReadBufferFromFile & buf) return metrics; } -uint64_t readMetricFromStatFile(ReadBufferFromFile & buf, const std::string & key) +uint64_t readMetricFromStatFile(ReadBufferFromFile & buf, std::string_view key) { - const auto all_metrics = readAllMetricsFromStatFile(buf); - if (const auto it = all_metrics.find(key); it != all_metrics.end()) - return it->second; + while (!buf.eof()) + { + std::string current_key; + readStringUntilWhitespace(current_key, buf); + if (current_key != key) + { + std::string dummy; + readStringUntilNewlineInto(dummy, buf); + buf.ignore(); + continue; + } + + assertChar(' ', buf); + uint64_t value = 0; + readIntText(value, buf); + return value; + } LOG_ERROR(getLogger("CgroupsReader"), "Cannot find '{}' in '{}'", key, buf.getFileName()); return 0; } From f449c2fea0487abbe262c10fc5af9a99df1bc822 Mon Sep 17 00:00:00 2001 From: Antonio Andelic Date: Thu, 25 Jul 2024 08:45:08 +0200 Subject: [PATCH 16/16] Fix --- src/Common/MemoryTracker.h | 5 +++++ src/Common/MemoryWorker.cpp | 8 +++----- src/Coordination/KeeperDispatcher.cpp | 3 ++- src/Coordination/KeeperServer.cpp | 2 +- 4 files changed, 11 insertions(+), 7 deletions(-) diff --git a/src/Common/MemoryTracker.h b/src/Common/MemoryTracker.h index d2db8489f19..f15465a20c1 100644 --- a/src/Common/MemoryTracker.h +++ b/src/Common/MemoryTracker.h @@ -120,6 +120,11 @@ public: return amount.load(std::memory_order_relaxed); } + Int64 getRSS() const + { + return rss.load(std::memory_order_relaxed); + } + // Merges and mutations may pass memory ownership to other threads thus in the end of execution // MemoryTracker for background task may have a non-zero counter. // This method is intended to fix the counter inside of background_memory_tracker. diff --git a/src/Common/MemoryWorker.cpp b/src/Common/MemoryWorker.cpp index e148f7f8f49..1b869ed9d6b 100644 --- a/src/Common/MemoryWorker.cpp +++ b/src/Common/MemoryWorker.cpp @@ -312,7 +312,7 @@ uint64_t MemoryWorker::getMemoryUsage() void MemoryWorker::backgroundThread() { std::chrono::milliseconds chrono_period_ms{period_ms}; - bool first_run = true; + [[maybe_unused]] bool first_run = true; std::unique_lock lock(mutex); while (true) { @@ -340,17 +340,15 @@ void MemoryWorker::backgroundThread() } #endif +#if USE_JEMALLOC if (unlikely(first_run || total_memory_tracker.get() < 0)) { -#if USE_JEMALLOC if (source != MemoryUsageSource::Jemalloc) epoch_mib.setValue(0); MemoryTracker::updateAllocated(allocated_mib.getValue()); -#elif defined(OS_LINUX) - MemoryTracker::updateAllocated(resident); -#endif } +#endif ProfileEvents::increment(ProfileEvents::MemoryWorkerRun); ProfileEvents::increment(ProfileEvents::MemoryWorkerRunElapsedMicroseconds, total_watch.elapsedMicroseconds()); diff --git a/src/Coordination/KeeperDispatcher.cpp b/src/Coordination/KeeperDispatcher.cpp index 4c2ccb8db64..893bb8e6082 100644 --- a/src/Coordination/KeeperDispatcher.cpp +++ b/src/Coordination/KeeperDispatcher.cpp @@ -150,10 +150,11 @@ void KeeperDispatcher::requestThread() { LOG_WARNING( log, - "Processing requests refused because of max_memory_usage_soft_limit {}, the total used memory is {}, request type " + "Processing requests refused because of max_memory_usage_soft_limit {}, the total allocated memory is {}, RSS is {}, request type " "is {}", ReadableSize(mem_soft_limit), ReadableSize(total_memory_tracker.get()), + ReadableSize(total_memory_tracker.getRSS()), request.request->getOpNum()); addErrorResponses({request}, Coordination::Error::ZCONNECTIONLOSS); continue; diff --git a/src/Coordination/KeeperServer.cpp b/src/Coordination/KeeperServer.cpp index d40e5ef2e50..ad9e8d32caa 100644 --- a/src/Coordination/KeeperServer.cpp +++ b/src/Coordination/KeeperServer.cpp @@ -599,7 +599,7 @@ bool KeeperServer::isLeaderAlive() const bool KeeperServer::isExceedingMemorySoftLimit() const { Int64 mem_soft_limit = keeper_context->getKeeperMemorySoftLimit(); - return mem_soft_limit > 0 && total_memory_tracker.get() >= mem_soft_limit; + return mem_soft_limit > 0 && std::max(total_memory_tracker.get(), total_memory_tracker.getRSS()) >= mem_soft_limit; } /// TODO test whether taking failed peer in count