Tune GNU malloc

Cargo.toml

@@ -32,6 +32,7 @@ members = [
     "common/lockfile",
     "common/logging",
     "common/lru_cache",
+    "common/malloc_ctl",
     "common/remote_signer_consumer",
     "common/slot_clock",
     "common/task_executor",

beacon_node/http_api/Cargo.toml

@@ -26,6 +26,7 @@ lighthouse_metrics = { path = "../../common/lighthouse_metrics" }
 lazy_static = "1.4.0"
 warp_utils = { path = "../../common/warp_utils" }
 slot_clock = { path = "../../common/slot_clock" }
+malloc_ctl = { path = "../../common/malloc_ctl" }
 eth2_ssz = { path = "../../consensus/ssz" }
 bs58 = "0.4.0"
 futures = "0.3.8"

beacon_node/http_api/src/lib.rs

@@ -22,6 +22,7 @@ use block_id::BlockId;
 use eth2::types::{self as api_types, ValidatorId};
 use eth2_libp2p::{types::SyncState, EnrExt, NetworkGlobals, PeerId, PubsubMessage};
 use lighthouse_version::version_with_platform;
+use malloc_ctl::{eprintln_malloc_stats, malloc_trim, DEFAULT_TRIM};
 use network::NetworkMessage;
 use serde::{Deserialize, Serialize};
 use slog::{crit, debug, error, info, warn, Logger};
@@ -2127,6 +2128,32 @@ pub fn serve<T: BeaconChainTypes>(
             })
         });
 
+    // GET lighthouse/malloc_stats
+    let get_lighthouse_malloc_stats = warp::path("lighthouse")
+        .and(warp::path("malloc_stats"))
+        .and(warp::path::end())
+        .and_then(|| {
+            blocking_json_task(move || {
+                eprintln_malloc_stats();
+                Ok::<_, warp::reject::Rejection>(())
+            })
+        });
+
+    // GET lighthouse/malloc_trim
+    let get_lighthouse_malloc_trim = warp::path("lighthouse")
+        .and(warp::path("malloc_trim"))
+        .and(warp::path::end())
+        .and_then(|| {
+            blocking_json_task(move || {
+                malloc_trim(DEFAULT_TRIM).map_err(|e| {
+                    warp_utils::reject::custom_server_error(format!(
+                        "malloc_trim failed with code {}",
+                        e
+                    ))
+                })
+            })
+        });
+
     let get_events = eth1_v1
         .and(warp::path("events"))
         .and(warp::path::end())
@@ -2233,6 +2260,8 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(get_lighthouse_eth1_deposit_cache.boxed())
                 .or(get_lighthouse_beacon_states_ssz.boxed())
                 .or(get_lighthouse_staking.boxed())
+                .or(get_lighthouse_malloc_stats.boxed())
+                .or(get_lighthouse_malloc_trim.boxed())
                 .or(get_events.boxed()),
         )
         .or(warp::post().and(

common/malloc_ctl/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "malloc_ctl"
+version = "0.1.0"
+authors = ["Paul Hauner <[email protected]>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+parking_lot = "0.11.0"
+lazy_static = "1.4.0"

common/malloc_ctl/src/lib.rs

@@ -0,0 +1,119 @@
+#![feature(backtrace)]
+#![feature(backtrace_frames)]
+
+pub mod profiling_allocator;
+pub mod trimmer_thread;
+
+pub use std::os::raw::{c_int, c_ulong};
+
+/// A default value to be provided to `malloc_trim`.
+///
+/// Value sourced from:
+///
+/// - https://man7.org/linux/man-pages/man3/mallopt.3.html
+pub const DEFAULT_TRIM: c_ulong = 1_024 * 128;
+
+/// A default value to be provided to `malloc_mmap_threshold`.
+///
+/// Value chosen so that it will store the values of the validators tree hash cache.
+pub const DEFAULT_MMAP_THRESHOLD: c_int = 2 * 1_024 * 1_024;
+
+/// Constants used to configure malloc internals.
+///
+/// Source:
+///
+/// https://github.com/lattera/glibc/blob/895ef79e04a953cac1493863bcae29ad85657ee1/malloc/malloc.h#L115-L123
+const M_MMAP_THRESHOLD: c_int = -4;
+const M_ARENA_MAX: c_int = -8;
+
+mod ffi {
+    /// See: https://man7.org/linux/man-pages/man3/malloc_trim.3.html
+    extern "C" {
+        pub fn malloc_trim(__pad: std::os::raw::c_ulong) -> ::std::os::raw::c_int;
+    }
+
+    /// See: https://man7.org/linux/man-pages/man3/malloc_stats.3.html
+    extern "C" {
+        pub fn malloc_stats();
+    }
+
+    /// See: https://man7.org/linux/man-pages/man3/mallopt.3.html
+    extern "C" {
+        pub fn mallopt(
+            __param: ::std::os::raw::c_int,
+            __val: ::std::os::raw::c_int,
+        ) -> ::std::os::raw::c_int;
+    }
+}
+
+fn into_result(result: c_int) -> Result<(), c_int> {
+    if result == 1 {
+        Ok(())
+    } else {
+        Err(result)
+    }
+}
+
+/// Uses `mallopt` to set the `M_ARENA_MAX` value, specifying the number of memory arenas to be
+/// created by malloc.
+///
+/// Generally speaking, a smaller arena count reduces memory fragmentation at the cost of memory contention
+/// between threads.
+///
+/// ## Resources
+///
+/// - https://man7.org/linux/man-pages/man3/mallopt.3.html
+pub fn malloc_arena_max(num_arenas: c_int) -> Result<(), c_int> {
+    unsafe { into_result(ffi::mallopt(M_ARENA_MAX, num_arenas)) }
+}
+
+/// Uses `mallopt` to set the `M_MMAP_THRESHOLD` value, specifying the threshold where objects of this
+/// size or larger are allocated via an `mmap`.
+///
+/// ## Resources
+///
+/// - https://man7.org/linux/man-pages/man3/mallopt.3.html
+pub fn malloc_mmap_threshold(num_arenas: c_int) -> Result<(), c_int> {
+    unsafe { into_result(ffi::mallopt(M_MMAP_THRESHOLD, num_arenas)) }
+}
+
+/// Calls `malloc_trim(0)`, freeing up available memory at the expense of CPU time and arena
+/// locking.
+///
+/// ## Resources
+///
+/// - https://man7.org/linux/man-pages/man3/malloc_trim.3.html
+pub fn malloc_trim(pad: c_ulong) -> Result<(), c_int> {
+    unsafe { into_result(ffi::malloc_trim(pad)) }
+}
+
+/// Calls `malloc_stats`, printing the output to stderr.
+///
+/// ## Resources
+///
+/// - https://man7.org/linux/man-pages/man3/malloc_stats.3.html
+pub fn eprintln_malloc_stats() {
+    unsafe { ffi::malloc_stats() }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn malloc_arena_max_does_not_panic() {
+        malloc_arena_max(1).unwrap()
+    }
+
+    #[test]
+    fn malloc_default_trim_does_not_panic() {
+        malloc_trim(DEFAULT_TRIM).unwrap()
+    }
+
+    /// Unfortunately this test will print into the test results, even on success. I don't know any
+    /// way to avoid this.
+    #[test]
+    fn eprintln_malloc_stats_does_not_panic() {
+        eprintln_malloc_stats()
+    }
+}

common/malloc_ctl/src/profiling_allocator.rs

@@ -0,0 +1,47 @@
+use crate::DEFAULT_MMAP_THRESHOLD;
+use parking_lot::RwLock;
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::backtrace::Backtrace;
+use std::collections::HashMap;
+use std::fs;
+use std::io::{self, Write};
+use std::path::PathBuf;
+use std::process;
+use lazy_static::lazy_static;
+
+lazy_static! {
+    pub static ref IN_USE: RwLock<bool> = <_>::default();
+}
+
+pub struct ProfilingAllocator;
+
+unsafe impl GlobalAlloc for ProfilingAllocator {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        let ret = System.alloc(layout);
+
+
+        if layout.size() >= DEFAULT_MMAP_THRESHOLD as usize {
+            let mut in_use = IN_USE.write();
+
+            if !*in_use {
+                *in_use = true;
+                drop(in_use);
+
+                let backtrace = Backtrace::capture().to_string();
+                eprintln!("alloc {}, {}, {}", layout.size(), *ret, backtrace);
+
+                *IN_USE.write() = false;
+            }
+        }
+
+        return ret;
+    }
+
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        if layout.size() >= DEFAULT_MMAP_THRESHOLD as usize {
+            eprintln!("dealloc {}, {}", layout.size(), *ptr);
+        }
+
+        System.dealloc(ptr, layout);
+    }
+}

common/malloc_ctl/src/trimmer_thread.rs

@@ -0,0 +1,20 @@
+use crate::{c_ulong, malloc_trim};
+use std::thread;
+use std::time::Duration;
+
+pub const DEFAULT_TRIM_INTERVAL: Duration = Duration::from_secs(60);
+
+/// Spawns a thread which will call `crate::malloc_trim(trim)`, sleeping `interval` between each
+/// call.
+///
+/// The function will not call `malloc_trim` on start, the first call will happen after `interval`
+/// has elapsed.
+pub fn spawn_trimmer_thread(interval: Duration, trim: c_ulong) -> thread::JoinHandle<()> {
+    thread::spawn(move || loop {
+        thread::sleep(interval);
+
+        if let Err(e) = malloc_trim(trim) {
+            eprintln!("malloc_trim failed with {}", e);
+        }
+    })
+}

lighthouse/Cargo.toml

@@ -43,6 +43,7 @@ account_utils = { path = "../common/account_utils" }
 remote_signer = { "path" = "../remote_signer" }
 lighthouse_metrics = { path = "../common/lighthouse_metrics" }
 lazy_static = "1.4.0"
+malloc_ctl = { path = "../common/malloc_ctl" }
 
 [dev-dependencies]
 tempfile = "3.1.0"

lighthouse/src/main.rs

@@ -1,3 +1,5 @@
+#![recursion_limit="256"]
+
 mod metrics;
 
 use beacon_node::{get_eth2_network_config, ProductionBeaconNode};
@@ -6,12 +8,21 @@ use env_logger::{Builder, Env};
 use environment::EnvironmentBuilder;
 use eth2_network_config::{Eth2NetworkConfig, DEFAULT_HARDCODED_NETWORK};
 use lighthouse_version::VERSION;
+use malloc_ctl::{
+    malloc_arena_max, malloc_mmap_threshold,
+    trimmer_thread::{spawn_trimmer_thread, DEFAULT_TRIM_INTERVAL},
+    DEFAULT_MMAP_THRESHOLD, DEFAULT_TRIM,
+    profiling_allocator::ProfilingAllocator
+};
 use slog::{crit, info, warn};
 use std::path::PathBuf;
 use std::process::exit;
 use types::{EthSpec, EthSpecId};
 use validator_client::ProductionValidatorClient;
 
+#[global_allocator]
+static A: ProfilingAllocator = ProfilingAllocator;
+
 pub const ETH2_CONFIG_FILENAME: &str = "eth2-spec.toml";
 
 fn bls_library_name() -> &'static str {
@@ -27,6 +38,24 @@ fn bls_library_name() -> &'static str {
 }
 
 fn main() {
+    // Configure malloc as the first thing we do, before it has the change to use the default
+    // values for anything.
+    //
+    // TODO: check for env variable so we don't overwrite it.
+    if let Err(e) = malloc_arena_max(1) {
+        eprintln!("Failed (code {}) to set malloc max arena count", e);
+        exit(1)
+    }
+
+    if let Err(e) = malloc_mmap_threshold(DEFAULT_MMAP_THRESHOLD) {
+        eprintln!("Failed (code {}) to set malloc mmap threshold", e);
+        exit(1)
+    }
+
+    // Spawn a thread which will periodically force malloc to "trim" itself, returning fragmented
+    // memory to the OS.
+    spawn_trimmer_thread(DEFAULT_TRIM_INTERVAL, DEFAULT_TRIM);
+
     // Parse the CLI parameters.
     let matches = App::new("Lighthouse")
         .version(VERSION.replace("Lighthouse/", "").as_str())