systemmemory: flush memory profiling

Author: steve.sterling

src/init.cpp

@@ -81,7 +81,7 @@
 #include <util/check.h>
 #include <util/fs.h>
 #include <util/fs_helpers.h>
-#include <util/mempressure.h>
+#include <util/systemmemory.h>
 #include <util/moneystr.h>
 #include <util/result.h>
 #include <util/signalinterrupt.h>
@@ -403,6 +403,7 @@ void Shutdown(NodeContext& node)
     node.chainman.reset();
     node.validation_signals.reset();
     node.scheduler.reset();
+    g_memory_profiler.reset();
     node.ecc_context.reset();
     node.kernel.reset();
 
@@ -1605,11 +1606,15 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
         }
     }, std::chrono::minutes{5});
 
-    // Check system memory pressure every 10 seconds.
+    // Initialize memory profiler for tracking flush overhead
+    assert(!g_memory_profiler);
+    g_memory_profiler = std::make_unique<MemoryProfiler>();
+
+    // Check system memory pressure every 5 seconds.
     // Updates g_system_needs_memory_released flag used by FlushStateToDisk.
     scheduler.scheduleEvery([]{
         CheckMemoryPressure();
-    }, std::chrono::seconds{10});
+    }, std::chrono::seconds{5});
 
     if (args.GetBoolArg("-logratelimit", BCLog::DEFAULT_LOGRATELIMIT)) {
         LogInstance().SetRateLimiting(BCLog::LogRateLimiter::Create(

src/test/validation_chainstate_tests.cpp

@@ -14,7 +14,7 @@
 #include <test/util/setup_common.h>
 #include <uint256.h>
 #include <util/check.h>
-#include <util/mempressure.h>
+#include <util/systemmemory.h>
 #include <validation.h>
 
 #include <vector>

src/util/CMakeLists.txt

@@ -15,7 +15,8 @@ add_library(bitcoin_util STATIC EXCLUDE_FROM_ALL
   fs_helpers.cpp
   hasher.cpp
   ioprio.cpp
-  mempressure.cpp
+  systemmemory.cpp
+  memory_profiler.cpp
   moneystr.cpp
   rbf.cpp
   readwritefile.cpp

src/util/memory_profiler.cpp

@@ -0,0 +1,193 @@
+// Copyright (c) 2025 The Bitcoin Knots developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <bitcoin-build-config.h> // IWYU pragma: keep
+
+#include <util/memory_profiler.h>
+
+#include <algorithm>
+
+#ifdef WIN32
+#include <windows.h>
+#include <psapi.h>
+#elif defined(__linux__)
+#include <fstream>
+#include <sstream>
+#include <string>
+#elif defined(__APPLE__)
+#include <sys/resource.h>
+#include <mach/mach.h>
+#endif
+
+std::unique_ptr<MemoryProfiler> g_memory_profiler;
+
+MemoryProfiler::MemoryProfiler()
+{
+    m_profiler_thread = std::make_unique<std::thread>(&MemoryProfiler::ProfilerThread, this);
+}
+
+MemoryProfiler::~MemoryProfiler()
+{
+    m_shutdown = true;
+    {
+        std::lock_guard<std::mutex> cv_lock(m_cv_mutex);
+        m_cv.notify_all();
+    }
+    if (m_profiler_thread && m_profiler_thread->joinable()) {
+        m_profiler_thread->join();
+    }
+}
+
+size_t MemoryProfiler::GetProcessRSS()
+{
+#ifdef WIN32
+    PROCESS_MEMORY_COUNTERS_EX pmc;
+    if (GetProcessMemoryInfo(GetCurrentProcess(), (PROCESS_MEMORY_COUNTERS*)&pmc, sizeof(pmc))) {
+        return static_cast<size_t>(pmc.PrivateUsage);  // Current commit charge
+    }
+#elif defined(__linux__)
+    std::ifstream status("/proc/self/status");
+    std::string line;
+    while (std::getline(status, line)) {
+        if (line.compare(0, 6, "VmRSS:") == 0) {
+            size_t pos = line.find_first_of("0123456789");
+            if (pos != std::string::npos) {
+                size_t kb = std::stoull(line.substr(pos));
+                return kb * 1024;
+            }
+        }
+    }
+#elif defined(__APPLE__)
+    struct mach_task_basic_info info;
+    mach_msg_type_number_t count = MACH_TASK_BASIC_INFO_COUNT;
+    if (task_info(mach_task_self(), MACH_TASK_BASIC_INFO, (task_info_t)&info, &count) == KERN_SUCCESS) {
+        return static_cast<size_t>(info.resident_size);
+    }
+#endif
+    return 0;
+}
+
+void MemoryProfiler::StartFlushProfiling()
+{
+    // Check if already profiling and stop previous profile if needed
+    if (m_profiling.load()) {
+        StopFlushProfiling();
+    }
+
+    LOCK(m_mutex);
+
+    m_current_profile = std::make_unique<FlushProfile>();
+    m_current_profile->start_time = std::chrono::steady_clock::now();
+
+    // Capture initial memory state (lock already held)
+    auto initial_sample = CaptureMemorySampleLocked("pre_flush", true);
+    m_current_profile->pre_flush_memory_mb = initial_sample.process_rss_mb;
+    m_current_profile->peak_memory_mb = initial_sample.process_rss_mb;
+    m_current_profile->samples.push_back(initial_sample);
+
+    m_profiling = true;
+
+    // Notify profiler thread
+    {
+        std::lock_guard<std::mutex> cv_lock(m_cv_mutex);
+        m_cv.notify_one();
+    }
+}
+
+std::unique_ptr<MemoryProfiler::FlushProfile> MemoryProfiler::StopFlushProfiling()
+{
+    LOCK(m_mutex);
+
+    if (!m_current_profile) {
+        return nullptr;
+    }
+
+    m_profiling = false;
+
+    // Capture final memory state (lock already held)
+    auto final_sample = CaptureMemorySampleLocked("post_flush", true);
+    m_current_profile->samples.push_back(final_sample);
+    m_current_profile->post_flush_memory_mb = final_sample.process_rss_mb;
+    m_current_profile->end_time = std::chrono::steady_clock::now();
+
+    // Calculate peak from all samples
+    for (const auto& sample : m_current_profile->samples) {
+        m_current_profile->peak_memory_mb = std::max(m_current_profile->peak_memory_mb, sample.process_rss_mb);
+    }
+
+    // Store in history
+    m_flush_history.push_back(*m_current_profile);
+    if (m_flush_history.size() > MAX_HISTORY_SIZE) {
+        m_flush_history.pop_front();
+    }
+
+    return std::move(m_current_profile);
+}
+
+std::vector<MemoryProfiler::FlushProfile> MemoryProfiler::GetRecentProfiles(size_t count) const
+{
+    LOCK(m_mutex);
+
+    std::vector<FlushProfile> result;
+    size_t start_idx = m_flush_history.size() > count ? m_flush_history.size() - count : 0;
+
+    for (size_t i = start_idx; i < m_flush_history.size(); ++i) {
+        result.push_back(m_flush_history[i]);
+    }
+
+    return result;
+}
+
+void MemoryProfiler::ProfilerThread()
+{
+    while (!m_shutdown) {
+        // Use cv_mutex for condition variable
+        std::unique_lock<std::mutex> cv_lock(m_cv_mutex);
+
+        // Wait for either profiling to start or shutdown
+        m_cv.wait_for(cv_lock, std::chrono::milliseconds(100), [this] {
+            return m_profiling.load() || m_shutdown.load();
+        });
+
+        cv_lock.unlock();
+
+        if (m_shutdown) break;
+
+        if (m_profiling) {
+            LOCK(m_mutex);
+            if (m_current_profile) {
+                // Need to capture sample without holding m_mutex to avoid deadlock
+                auto profile_copy = *m_current_profile; // Copy current profile state
+                LEAVE_CRITICAL_SECTION(m_mutex);
+
+                auto sample = CaptureMemorySample("profiling");
+
+                ENTER_CRITICAL_SECTION(m_mutex);
+                // Double-check we still have a profile after reacquiring lock
+                if (m_current_profile && m_profiling) {
+                    m_current_profile->samples.push_back(sample);
+                    m_current_profile->peak_memory_mb = std::max(m_current_profile->peak_memory_mb, sample.process_rss_mb);
+                }
+            }
+        }
+    }
+}
+
+MemoryProfiler::MemorySample MemoryProfiler::CaptureMemorySample(const std::string& phase) const
+{
+    return CaptureMemorySampleLocked(phase, false);
+}
+
+MemoryProfiler::MemorySample MemoryProfiler::CaptureMemorySampleLocked(const std::string& phase, bool lock_already_held) const
+{
+    MemorySample sample;
+    sample.timestamp = std::chrono::steady_clock::now();
+    sample.phase = phase;
+
+    // Just get RSS - nothing else
+    size_t rss_bytes = GetProcessRSS();
+    sample.process_rss_mb = rss_bytes > 0 ? rss_bytes / (1 << 20) : 0;
+
+    return sample;
+}

src/util/memory_profiler.h

@@ -0,0 +1,90 @@
+// Copyright (c) 2025 The Bitcoin Knots developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_UTIL_MEMORY_PROFILER_H
+#define BITCOIN_UTIL_MEMORY_PROFILER_H
+
+#if defined(HAVE_CONFIG_H)
+#include <config/bitcoin-config.h>
+#endif
+
+#include <sync.h>
+#include <util/time.h>
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <deque>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+/**
+ * Memory profiler for tracking memory usage during database operations
+ * Particularly focused on understanding memory overhead during LevelDB flushes
+ */
+class MemoryProfiler
+{
+public:
+    struct MemorySample {
+        std::chrono::steady_clock::time_point timestamp;
+        size_t process_rss_mb;          // Process resident set size
+        std::string phase;              // Current operation phase
+    };
+
+    struct FlushProfile {
+        std::chrono::steady_clock::time_point start_time;
+        std::chrono::steady_clock::time_point end_time;
+        size_t pre_flush_memory_mb;
+        size_t peak_memory_mb;
+        size_t post_flush_memory_mb;
+        std::vector<MemorySample> samples;
+
+        int64_t GetDurationMs() const {
+            return std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time).count();
+        }
+    };
+
+    MemoryProfiler();
+    ~MemoryProfiler();
+
+    // Start profiling a flush operation
+    void StartFlushProfiling() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+
+    // Stop profiling and return the profile
+    std::unique_ptr<FlushProfile> StopFlushProfiling() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+
+    // Get historical flush profiles
+    std::vector<FlushProfile> GetRecentProfiles(size_t count = 10) const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+
+private:
+    void ProfilerThread() NO_THREAD_SAFETY_ANALYSIS;
+    MemorySample CaptureMemorySample(const std::string& phase) const NO_THREAD_SAFETY_ANALYSIS;
+    MemorySample CaptureMemorySampleLocked(const std::string& phase, bool lock_already_held) const NO_THREAD_SAFETY_ANALYSIS;
+
+    // Platform-specific memory tracking functions
+    static size_t GetProcessRSS();
+
+    mutable Mutex m_mutex;
+
+    // Current profiling state
+    std::atomic<bool> m_profiling{false};
+    std::unique_ptr<FlushProfile> m_current_profile GUARDED_BY(m_mutex);
+
+    // Historical data
+    std::deque<FlushProfile> m_flush_history GUARDED_BY(m_mutex);
+    static constexpr size_t MAX_HISTORY_SIZE = 100;
+
+    // Profiler thread
+    std::unique_ptr<std::thread> m_profiler_thread;
+    std::atomic<bool> m_shutdown{false};
+    mutable std::mutex m_cv_mutex;
+    mutable std::condition_variable m_cv;
+};
+
+// Global instance (initialized in init.cpp if profiling is enabled)
+extern std::unique_ptr<MemoryProfiler> g_memory_profiler;
+
+#endif // BITCOIN_UTIL_MEMORY_PROFILER_H