[opt](memory) Purge all unused Jemalloc dirty pages for all arenas when exceed limit

be/src/common/config.cpp

@@ -568,6 +568,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_dirty_pages_mem_limit_percent, "5%");
 
 // to forward compatibility, will be removed later
 DEFINE_mBool(enable_token_check, "true");

be/src/common/config.h

@@ -619,6 +619,8 @@ DECLARE_String(pprof_profile_dir);
 DECLARE_mString(jeprofile_dir);
 // Purge all unused dirty pages for all arenas.
 DECLARE_mBool(enable_je_purge_dirty_pages);
+// Purge all unused Jemalloc dirty pages for all arenas when exceed je_dirty_pages_mem_limit and process exceed soft limit.
+DECLARE_mString(je_dirty_pages_mem_limit_percent);
 
 // to forward compatibility, will be removed later
 DECLARE_mBool(enable_token_check);

be/src/common/daemon.cpp

@@ -212,26 +212,29 @@ void Daemon::memory_maintenance_thread() {
         doris::PerfCounters::refresh_proc_status();
         doris::MemInfo::refresh_proc_meminfo();
         doris::GlobalMemoryArbitrator::reset_refresh_interval_memory_growth();
+        ExecEnv::GetInstance()->brpc_iobuf_block_memory_tracker()->set_consumption(
+                butil::IOBuf::block_memory());
+        // 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::je_dirty_pages_mem() > doris::MemInfo::je_dirty_pages_mem_limit() &&
+            GlobalMemoryArbitrator::is_exceed_soft_mem_limit()) {
+            doris::MemInfo::notify_je_purge_dirty_pages();
+        }
+#endif
+        if (config::enable_system_metrics) {
+            DorisMetrics::instance()->system_metrics()->update_allocator_metrics();
+        }
+#endif
 
         // Update and print memory stat when the memory changes by 256M.
         if (abs(last_print_proc_mem - PerfCounters::get_vm_rss()) > 268435456) {
             last_print_proc_mem = PerfCounters::get_vm_rss();
             doris::MemTrackerLimiter::clean_tracker_limiter_group();
             doris::MemTrackerLimiter::enable_print_log_process_usage();
-
             // Refresh mem tracker each type counter.
             doris::MemTrackerLimiter::refresh_global_counter();
-
-            // Refresh allocator memory metrics.
-#if !defined(ADDRESS_SANITIZER) && !defined(LEAK_SANITIZER) && !defined(THREAD_SANITIZER)
-            doris::MemInfo::refresh_allocator_mem();
-            if (config::enable_system_metrics) {
-                DorisMetrics::instance()->system_metrics()->update_allocator_metrics();
-            }
-#endif
-
-            ExecEnv::GetInstance()->brpc_iobuf_block_memory_tracker()->set_consumption(
-                    butil::IOBuf::block_memory());
             LOG(INFO) << doris::GlobalMemoryArbitrator::
                             process_mem_log_str(); // print mem log when memory state by 256M
         }

be/src/util/mem_info.cpp

@@ -52,6 +52,8 @@ 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_dirty_pages_mem = std::numeric_limits<int64_t>::min();
+std::atomic<int64_t> MemInfo::_s_je_dirty_pages_mem_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();
@@ -86,6 +88,8 @@ void MemInfo::refresh_allocator_mem() {
                                          get_je_metrics("stats.metadata") +
                                          get_je_all_arena_metrics("pdirty") * get_page_size(),
                                  std::memory_order_relaxed);
+    _s_je_dirty_pages_mem.store(get_je_all_arena_metrics("pdirty") * get_page_size(),
+                                std::memory_order_relaxed);
     _s_virtual_memory_used.store(get_je_metrics("stats.mapped"), std::memory_order_relaxed);
 #else
     _s_allocator_cache_mem.store(get_tc_metrics("tcmalloc.pageheap_free_bytes") +
@@ -244,6 +248,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_dirty_pages_mem_limit.store(ParseUtil::parse_mem_spec(
+                config::je_dirty_pages_mem_limit_percent, -1, _s_mem_limit, &is_percent));
     }
 
     // 3. refresh process available memory

be/src/util/mem_info.h

@@ -147,6 +147,12 @@ class MemInfo {
     static inline size_t allocator_cache_mem() {
         return _s_allocator_cache_mem.load(std::memory_order_relaxed);
     }
+    static inline int64_t je_dirty_pages_mem() {
+        return _s_je_dirty_pages_mem.load(std::memory_order_relaxed);
+    }
+    static inline int64_t je_dirty_pages_mem_limit() {
+        return _s_je_dirty_pages_mem_limit.load(std::memory_order_relaxed);
+    }
 
     // Tcmalloc property `generic.total_physical_bytes` records the total length of the virtual memory
     // obtained by the process malloc, not the physical memory actually used by the process in the OS.
@@ -181,6 +187,8 @@ 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_dirty_pages_mem;
+    static std::atomic<int64_t> _s_je_dirty_pages_mem_limit;
     static std::atomic<int64_t> _s_virtual_memory_used;
 
     static int64_t _s_cgroup_mem_limit;