Measure heap usage while processing the ledger at validator startup (bp #7667) (#7669)

automerge

Cargo.lock

@@ -3342,8 +3342,6 @@ dependencies = [
  "hex-literal 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "indexmap 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "itertools 0.8.2 (registry+https://github.com/rust-lang/crates.io-index)",
- "jemalloc-ctl 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
- "jemallocator 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 14.0.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-derive 14.0.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-http-server 14.0.3 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -3702,6 +3700,10 @@ dependencies = [
 name = "solana-measure"
 version = "0.21.6"
 dependencies = [
+ "jemalloc-ctl 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "jemallocator 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
+ "solana-metrics 0.21.6",
  "solana-sdk 0.21.6",
 ]
 

core/Cargo.toml

@@ -71,10 +71,6 @@ untrusted = "0.7.0"
 solana-rayon-threadlimit = { path = "../rayon-threadlimit", version = "0.21.6" }
 reed-solomon-erasure = { package = "solana-reed-solomon-erasure", version = "4.0.1-3", features = ["simd-accel"] }
 
-[target."cfg(unix)".dependencies]
-jemallocator = "0.3.2"
-jemalloc-ctl = "0.3.2"
-
 [dev-dependencies]
 hex-literal = "0.2.1"
 matches = "0.1.6"

core/src/banking_stage.rs

@@ -7,7 +7,6 @@ use crate::{
     poh_recorder::{PohRecorder, PohRecorderError, WorkingBankEntry},
     poh_service::PohService,
     result::{Error, Result},
-    thread_mem_usage,
 };
 use crossbeam_channel::{Receiver as CrossbeamReceiver, RecvTimeoutError};
 use itertools::Itertools;
@@ -17,7 +16,7 @@ use solana_ledger::{
     entry::hash_transactions,
     leader_schedule_cache::LeaderScheduleCache,
 };
-use solana_measure::measure::Measure;
+use solana_measure::{measure::Measure, thread_mem_usage};
 use solana_metrics::{inc_new_counter_debug, inc_new_counter_info, inc_new_counter_warn};
 use solana_perf::{cuda_runtime::PinnedVec, perf_libs};
 use solana_runtime::{accounts_db::ErrorCounters, bank::Bank, transaction_batch::TransactionBatch};

core/src/cluster_info.rs

@@ -24,7 +24,6 @@ use crate::{
     repair_service::RepairType,
     result::{Error, Result},
     sendmmsg::{multicast, send_mmsg},
-    thread_mem_usage,
     weighted_shuffle::{weighted_best, weighted_shuffle},
 };
 use bincode::{serialize, serialized_size};
@@ -32,6 +31,7 @@ use core::cmp;
 use itertools::Itertools;
 use rand::{thread_rng, Rng};
 use solana_ledger::{bank_forks::BankForks, blocktree::Blocktree, staking_utils};
+use solana_measure::thread_mem_usage;
 use solana_metrics::{datapoint_debug, inc_new_counter_debug, inc_new_counter_error};
 use solana_net_utils::{
     bind_common, bind_common_in_range, bind_in_range, find_available_port_in_range,

core/src/fetch_stage.rs

@@ -5,7 +5,7 @@ use crate::packet::PacketsRecycler;
 use crate::poh_recorder::PohRecorder;
 use crate::result::{Error, Result};
 use crate::streamer::{self, PacketReceiver, PacketSender};
-use crate::thread_mem_usage;
+use solana_measure::thread_mem_usage;
 use solana_metrics::{inc_new_counter_debug, inc_new_counter_info};
 use solana_perf::recycler::Recycler;
 use solana_sdk::clock::DEFAULT_TICKS_PER_SLOT;

core/src/lib.rs

@@ -12,7 +12,6 @@ pub mod chacha_cuda;
 pub mod cluster_info_vote_listener;
 pub mod commitment;
 pub mod shred_fetch_stage;
-pub mod thread_mem_usage;
 #[macro_use]
 pub mod contact_info;
 pub mod archiver;
@@ -84,10 +83,3 @@ extern crate solana_metrics;
 #[cfg(test)]
 #[macro_use]
 extern crate matches;
-
-#[cfg(unix)]
-extern crate jemallocator;
-
-#[cfg(unix)]
-#[global_allocator]
-static ALLOC: jemallocator::Jemalloc = jemallocator::Jemalloc;

core/src/replay_stage.rs

@@ -7,7 +7,6 @@ use crate::{
     poh_recorder::PohRecorder,
     result::{Error, Result},
     rpc_subscriptions::RpcSubscriptions,
-    thread_mem_usage,
 };
 use solana_ledger::{
     bank_forks::BankForks,
@@ -18,7 +17,7 @@ use solana_ledger::{
     leader_schedule_cache::LeaderScheduleCache,
     snapshot_package::SnapshotPackageSender,
 };
-use solana_measure::measure::Measure;
+use solana_measure::{measure::Measure, thread_mem_usage};
 use solana_metrics::inc_new_counter_info;
 use solana_runtime::bank::Bank;
 use solana_sdk::{

core/src/streamer.rs

@@ -4,7 +4,7 @@
 use crate::packet::{self, send_to, Packets, PacketsRecycler, PACKETS_PER_BATCH};
 use crate::recvmmsg::NUM_RCVMMSGS;
 use crate::result::{Error, Result};
-use crate::thread_mem_usage;
+use solana_measure::thread_mem_usage;
 use solana_sdk::timing::duration_as_ms;
 use std::net::UdpSocket;
 use std::sync::atomic::{AtomicBool, Ordering};

ledger/src/blocktree_processor.rs

@@ -11,6 +11,7 @@ use itertools::Itertools;
 use log::*;
 use rand::{seq::SliceRandom, thread_rng};
 use rayon::{prelude::*, ThreadPool};
+use solana_measure::thread_mem_usage;
 use solana_metrics::{datapoint, datapoint_error, inc_new_counter_debug};
 use solana_rayon_threadlimit::get_thread_count;
 use solana_runtime::{
@@ -282,6 +283,9 @@ pub fn process_blocktree_from_root(
     opts: &ProcessOptions,
 ) -> result::Result<(BankForks, Vec<BankForksInfo>, LeaderScheduleCache), BlocktreeProcessorError> {
     info!("processing ledger from root slot {}...", bank.slot());
+    let allocated = thread_mem_usage::Allocatedp::default();
+    let initial_allocation = allocated.get();
+
     // Starting slot must be a root, and thus has no parents
     assert!(bank.parent().is_none());
     let start_slot = bank.slot();
@@ -334,8 +338,9 @@ pub fn process_blocktree_from_root(
     };
 
     info!(
-        "ledger processed in {}ms. {} fork{} at {}",
+        "ledger processed in {}ms. {} MB allocated. {} fork{} at {}",
         duration_as_ms(&now.elapsed()),
+        allocated.since(initial_allocation) / 1_000_000,
         bank_forks_info.len(),
         if bank_forks_info.len() > 1 { "s" } else { "" },
         bank_forks_info
@@ -451,6 +456,9 @@ fn process_next_slots(
         // Only process full slots in blocktree_processor, replay_stage
         // handles any partials
         if next_meta.is_full() {
+            let allocated = thread_mem_usage::Allocatedp::default();
+            let initial_allocation = allocated.get();
+
             let next_bank = Arc::new(Bank::new_from_parent(
                 &bank,
                 &leader_schedule_cache
@@ -458,7 +466,12 @@ fn process_next_slots(
                     .unwrap(),
                 *next_slot,
             ));
-            trace!("Add child bank {} of slot={}", next_slot, bank.slot());
+            trace!(
+                "New bank for slot {}, parent slot is {}. {} bytes allocated",
+                next_slot,
+                bank.slot(),
+                allocated.since(initial_allocation)
+            );
             pending_slots.push((*next_slot, next_meta, next_bank, bank.last_blockhash()));
         } else {
             let bfi = BankForksInfo {
@@ -486,6 +499,7 @@ fn process_pending_slots(
     let mut fork_info = vec![];
     let mut last_status_report = Instant::now();
     let mut pending_slots = vec![];
+    let mut last_root_slot = root_bank.slot();
     process_next_slots(
         root_bank,
         root_meta,
@@ -500,7 +514,10 @@ fn process_pending_slots(
         let (slot, meta, bank, last_entry_hash) = pending_slots.pop().unwrap();
 
         if last_status_report.elapsed() > Duration::from_secs(2) {
-            info!("processing ledger...slot {}", slot);
+            info!(
+                "processing ledger: slot={}, last root slot={}",
+                slot, last_root_slot
+            );
             last_status_report = Instant::now();
         }
 
@@ -509,6 +526,9 @@ fn process_pending_slots(
             continue;
         }
 
+        let allocated = thread_mem_usage::Allocatedp::default();
+        let initial_allocation = allocated.get();
+
         // Fetch all entries for this slot
         let entries = blocktree.get_slot_entries(slot, 0, None).map_err(|err| {
             warn!("Failed to load entries for slot {}: {:?}", slot, err);
@@ -531,8 +551,16 @@ fn process_pending_slots(
             bank.squash();
             pending_slots.clear();
             fork_info.clear();
+            last_root_slot = slot;
         }
 
+        trace!(
+            "Bank for {}slot {} is complete. {} bytes allocated",
+            if last_root_slot == slot { "root " } else { "" },
+            slot,
+            allocated.since(initial_allocation)
+        );
+
         if slot >= dev_halt_at_slot {
             let bfi = BankForksInfo { bank_slot: slot };
             fork_info.push((bank, bfi));

measure/Cargo.toml

@@ -11,4 +11,10 @@ license = "Apache-2.0"
 edition = "2018"
 
 [dependencies]
+log = "0.4.8"
 solana-sdk = { path = "../sdk", version = "0.21.6" }
+solana-metrics = { path = "../metrics", version = "0.21.6" }
+
+[target."cfg(unix)".dependencies]
+jemallocator = "0.3.2"
+jemalloc-ctl = "0.3.2"

measure/src/lib.rs

@@ -1 +1,9 @@
 pub mod measure;
+pub mod thread_mem_usage;
+
+#[cfg(unix)]
+extern crate jemallocator;
+
+#[cfg(unix)]
+#[global_allocator]
+static ALLOC: jemallocator::Jemalloc = jemallocator::Jemalloc;

measure/src/thread_mem_usage.rs

@@ -28,6 +28,7 @@ impl Allocatedp {
         Self {}
     }
 
+    /// Return current thread heap usage
     pub fn get(&self) -> u64 {
         #[cfg(unix)]
         {
@@ -36,4 +37,9 @@ impl Allocatedp {
         #[cfg(not(unix))]
         0
     }
+
+    /// Return the difference in thread heap usage since a previous `get()`
+    pub fn since(&self, previous: u64) -> i64 {
+        self.get() as i64 - previous as i64
+    }
 }