1

be/src/common/config.cpp

@@ -566,6 +566,7 @@ DEFINE_String(pprof_profile_dir, "${DORIS_HOME}/log");
 // for jeprofile in jemalloc
 DEFINE_mString(jeprofile_dir, "${DORIS_HOME}/log");
 DEFINE_mBool(enable_je_purge_dirty_pages, "true");
+DEFINE_mString(je_cache_limit_percent, "10%");
 
 // to forward compatibility, will be removed later
 DEFINE_mBool(enable_token_check, "true");

be/src/common/config.h

@@ -617,6 +617,7 @@ DECLARE_String(pprof_profile_dir);
 DECLARE_mString(jeprofile_dir);
 // Purge all unused dirty pages for all arenas.
 DECLARE_mBool(enable_je_purge_dirty_pages);
+DECLARE_mString(je_cache_limit_percent);
 
 // to forward compatibility, will be removed later
 DECLARE_mBool(enable_token_check);

be/src/common/daemon.cpp

@@ -225,6 +225,11 @@ void Daemon::memory_maintenance_thread() {
             // Refresh allocator memory metrics.
 #if !defined(ADDRESS_SANITIZER) && !defined(LEAK_SANITIZER) && !defined(THREAD_SANITIZER)
             doris::MemInfo::refresh_allocator_mem();
+#ifdef USE_JEMALLOC
+            if (doris::MemInfo::allocator_cache_mem() > doris::MemInfo::je_cache_limit()) {
+                doris::MemInfo::notify_je_purge_dirty_pages();
+            }
+#endif
             if (config::enable_system_metrics) {
                 DorisMetrics::instance()->system_metrics()->update_allocator_metrics();
             }
@@ -374,11 +379,13 @@ void Daemon::report_runtime_query_statistics_thread() {
 }
 
 void Daemon::je_purge_dirty_pages_thread() const {
+    int32_t interval_milliseconds = config::memory_maintenance_sleep_time_ms;
     do {
         std::unique_lock<std::mutex> l(doris::MemInfo::je_purge_dirty_pages_lock);
         while (_stop_background_threads_latch.count() != 0 &&
                !doris::MemInfo::je_purge_dirty_pages_notify.load(std::memory_order_relaxed)) {
-            doris::MemInfo::je_purge_dirty_pages_cv.wait_for(l, std::chrono::seconds(1));
+            doris::MemInfo::je_purge_dirty_pages_cv.wait_for(
+                    l, std::chrono::milliseconds(interval_milliseconds));
         }
         if (_stop_background_threads_latch.count() == 0) {
             break;

be/src/util/mem_info.cpp

@@ -52,6 +52,7 @@ std::atomic<int64_t> MemInfo::_s_mem_limit = std::numeric_limits<int64_t>::max()
 std::atomic<int64_t> MemInfo::_s_soft_mem_limit = std::numeric_limits<int64_t>::max();
 
 std::atomic<int64_t> MemInfo::_s_allocator_cache_mem = 0;
+std::atomic<int64_t> MemInfo::_s_je_cache_limit = std::numeric_limits<int64_t>::max();
 std::atomic<int64_t> MemInfo::_s_virtual_memory_used = 0;
 
 int64_t MemInfo::_s_cgroup_mem_limit = std::numeric_limits<int64_t>::max();
@@ -244,6 +245,8 @@ void MemInfo::refresh_proc_meminfo() {
                                                                  _s_mem_limit, &is_percent));
         _s_process_full_gc_size.store(ParseUtil::parse_mem_spec(config::process_full_gc_size, -1,
                                                                 _s_mem_limit, &is_percent));
+        _s_je_cache_limit.store(ParseUtil::parse_mem_spec(config::je_cache_limit_percent, -1,
+                                                          _s_mem_limit, &is_percent));
     }
 
     // 3. refresh process available memory

be/src/util/mem_info.h

@@ -75,6 +75,9 @@ class MemInfo {
 
     static void refresh_proc_meminfo();
 
+    static inline int64_t je_cache_limit() {
+        return _s_je_cache_limit.load(std::memory_order_relaxed);
+    }
     static inline int64_t sys_mem_available_low_water_mark() {
         return _s_sys_mem_available_low_water_mark;
     }
@@ -181,6 +184,7 @@ class MemInfo {
     static std::atomic<int64_t> _s_soft_mem_limit;
 
     static std::atomic<int64_t> _s_allocator_cache_mem;
+    static std::atomic<int64_t> _s_je_cache_limit;
     static std::atomic<int64_t> _s_virtual_memory_used;
 
     static int64_t _s_cgroup_mem_limit;