feat(bench): split the beacon import's stf phase into steps, classes and resources

bin/ethlambda/src/benchmark/import/mod.rs

@@ -9,6 +9,7 @@ use eyre::WrapErr as _;
 
 pub(crate) mod corpus;
 pub(crate) mod fetch;
+pub(crate) mod process;
 pub(crate) mod replay;
 pub(crate) mod source;
 

bin/ethlambda/src/benchmark/import/process.rs

@@ -0,0 +1,117 @@
+//! Process-level resource readings for the import replay: CPU time and memory.
+//!
+//! Wall time says how long an import took; it does not say whether the time
+//! went to computation or to waiting, nor whether it ran on one core or
+//! several. CPU seconds beside it does: an import whose CPU time is well above
+//! its wall time was parallel (rayon in the merkle fold and the BLS batches),
+//! and one whose CPU time is well below it spent the difference blocked.
+//!
+//! Everything here reads the whole process, not the importing thread. That is
+//! deliberate: the work an import causes is not all on its own thread (the
+//! state writer, the merkle fold's worker pool, RocksDB's background
+//! compaction), and a per-thread reading would call a parallel import cheap.
+//! The same property is the caveat: anything else the process does during the
+//! import is in the number, which is why the replay runs nothing else.
+//!
+//! No dependency beyond `libc`, which the binary already carries for its file
+//! descriptor limit. `getrusage` is the one portable process-wide CPU clock
+//! the standard library does not wrap, so it is the one `unsafe` call here.
+
+/// User plus system CPU seconds this process has used so far, across all of
+/// its threads, or `None` if the kernel refuses the reading.
+pub(crate) fn cpu_seconds() -> Option<f64> {
+    let usage = rusage()?;
+    Some(seconds(usage.ru_utime) + seconds(usage.ru_stime))
+}
+
+/// The process's peak resident set size so far, in bytes.
+///
+/// A high-water mark, so it never goes down: between two blocks it moves only
+/// when a block set a new peak, and a replay's last value is the run's peak.
+pub(crate) fn peak_rss_bytes() -> Option<u64> {
+    let usage = rusage()?;
+    let maxrss = u64::try_from(usage.ru_maxrss).ok()?;
+    // `ru_maxrss` is bytes on macOS and kibibytes on Linux.
+    if cfg!(target_os = "macos") {
+        Some(maxrss)
+    } else {
+        maxrss.checked_mul(1024)
+    }
+}
+
+/// The process's resident set size right now, in bytes.
+///
+/// Read from `/proc/self/status`, so Linux only: macOS has no equivalent that
+/// needs no new dependency, and `None` there is honest where a stand-in would
+/// not be. A replay on a Mac still reports [`peak_rss_bytes`].
+pub(crate) fn rss_bytes() -> Option<u64> {
+    let status = std::fs::read_to_string("/proc/self/status").ok()?;
+    parse_vm_rss(&status)
+}
+
+/// The `VmRSS` line of a `/proc/<pid>/status` file, which the kernel reports in
+/// kibibytes.
+fn parse_vm_rss(status: &str) -> Option<u64> {
+    let line = status
+        .lines()
+        .find_map(|line| line.strip_prefix("VmRSS:"))?;
+    let kibibytes: u64 = line.trim().strip_suffix("kB")?.trim().parse().ok()?;
+    kibibytes.checked_mul(1024)
+}
+
+fn seconds(time: libc::timeval) -> f64 {
+    time.tv_sec as f64 + time.tv_usec as f64 / 1e6
+}
+
+fn rusage() -> Option<libc::rusage> {
+    let mut usage = std::mem::MaybeUninit::<libc::rusage>::uninit();
+    // SAFETY: `getrusage` writes a complete `rusage` through the pointer when
+    // it returns 0 and does not read it, and the pointer is to a live local;
+    // the value is only assumed initialized on that success path.
+    unsafe {
+        if libc::getrusage(libc::RUSAGE_SELF, usage.as_mut_ptr()) != 0 {
+            return None;
+        }
+        Some(usage.assume_init())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn cpu_time_advances_while_the_process_computes() {
+        let before = cpu_seconds().expect("getrusage works on the host");
+        // Enough work that the clock's resolution cannot hide it; `black_box`
+        // keeps the loop from being folded away.
+        let mut acc = 0u64;
+        let started = std::time::Instant::now();
+        while started.elapsed() < std::time::Duration::from_millis(30) {
+            for i in 0..10_000u64 {
+                acc = std::hint::black_box(acc.wrapping_add(i * i));
+            }
+        }
+        let after = cpu_seconds().expect("getrusage works on the host");
+        assert!(after > before, "{after} should exceed {before}");
+    }
+
+    #[test]
+    fn a_process_has_a_nonzero_peak_resident_set() {
+        assert!(peak_rss_bytes().expect("getrusage works on the host") > 0);
+    }
+
+    #[test]
+    fn vm_rss_is_read_in_bytes_from_the_kibibyte_line() {
+        let status = "Name:\tethlambda\nVmHWM:\t  999 kB\nVmRSS:\t  2048 kB\nThreads:\t4\n";
+        assert_eq!(parse_vm_rss(status), Some(2048 * 1024));
+    }
+
+    #[test]
+    fn a_status_file_without_vm_rss_gives_none() {
+        // A kernel thread's status has no `VmRSS`, and a malformed line must
+        // not become a number.
+        assert_eq!(parse_vm_rss("Name:\tkthreadd\n"), None);
+        assert_eq!(parse_vm_rss("VmRSS:\tlots\n"), None);
+    }
+}

bin/ethlambda/src/benchmark/import/replay.rs

@@ -16,6 +16,7 @@
 //! only thing allowed to hold post-states, and a stray clone here would pin
 //! an entry the LRU believes it evicted, raising the real ceiling silently.
 
+use std::collections::BTreeMap;
 use std::path::Path;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -24,6 +25,7 @@ use ethlambda_blockchain::metrics::{BLOCK_ARRIVAL_PHASES, BLOCK_IMPORT_PHASES};
 use ethlambda_blockchain::{BlockChainServer, ImportOutcome};
 use ethlambda_p2p::beacon::decode;
 use ethlambda_state_transition::beacon::fork_choice;
+use ethlambda_state_transition::metrics::{BEACON_STF_PHASES, STF_CONTEXT_IMPORT};
 use ethlambda_storage::Store;
 use ethlambda_storage::backend::RocksDBBackend;
 use ethlambda_types::ShortRoot;
@@ -34,15 +36,101 @@ use ethlambda_types::primitives::H256;
 
 use super::ReplayOptions;
 use super::corpus::{ANCHOR_BLOCK_FILE, ANCHOR_STATE_FILE, Manifest, block_path, hex_root};
-use crate::benchmark::PhaseTimer;
+use super::process;
 use crate::benchmark::report::common::{Environment, format_ms};
-use crate::benchmark::report::import::{Params, Report, Sample};
+use crate::benchmark::report::import::{BlockClass, Caches, LookupCounts, Params, Report, Sample};
+use crate::benchmark::{PhaseTimer, read_counter};
 
 /// The histogram `ImportTimings` writes to, read back by [`PhaseTimer`] the
 /// same way the synthetic workload reads
 /// `lean_block_proposal_attestation_build_phase_seconds`.
 const IMPORT_PHASE_HISTOGRAM: &str = "lean_block_import_phase_seconds";
 
+/// The beacon state transition's own step histogram, read back over its
+/// `import` context only.
+///
+/// Reading every series would charge the epoch precompute worker's
+/// `process_epoch` to whichever block was importing when it finished. A replay
+/// runs no such worker, but the filter is what makes that a property of the
+/// reading rather than of how this binary happens to be wired.
+const STF_PHASE_HISTOGRAM: &str = "lean_beacon_stf_phase_seconds";
+
+/// The counters of the caches the state transition leans on, each labeled by
+/// `result`. Read around the measured blocks, so the report's hit rates cover
+/// the run and not the warm-up.
+const EPOCH_PRECOMPUTE_COUNTER: &str = "lean_beacon_epoch_precompute_lookups_total";
+const COMMITTEE_CACHE_COUNTER: &str = "lean_beacon_committee_cache_lookups_total";
+const TOTAL_ACTIVE_BALANCE_COUNTER: &str = "lean_beacon_total_active_balance_lookups_total";
+const JUSTIFIED_BALANCES_COUNTER: &str = "lean_beacon_justified_balances_lookups_total";
+
+/// The four cache counters' per-`result` values at one instant.
+struct CacheReading {
+    epoch_precompute: BTreeMap<String, u64>,
+    committee_cache: BTreeMap<String, u64>,
+    total_active_balance: BTreeMap<String, u64>,
+    justified_balances: BTreeMap<String, u64>,
+}
+
+impl CacheReading {
+    fn now() -> Self {
+        Self {
+            epoch_precompute: read_counter(EPOCH_PRECOMPUTE_COUNTER),
+            committee_cache: read_counter(COMMITTEE_CACHE_COUNTER),
+            total_active_balance: read_counter(TOTAL_ACTIVE_BALANCE_COUNTER),
+            justified_balances: read_counter(JUSTIFIED_BALANCES_COUNTER),
+        }
+    }
+
+    /// What each cache served between `self` and now.
+    fn deltas_since_start(self) -> Caches {
+        let now = Self::now();
+        Caches {
+            epoch_precompute: LookupCounts::from_results(
+                &delta(&self.epoch_precompute, &now.epoch_precompute),
+                false,
+            ),
+            committee_cache: LookupCounts::from_results(
+                &delta(&self.committee_cache, &now.committee_cache),
+                true,
+            ),
+            total_active_balance: LookupCounts::from_results(
+                &delta(&self.total_active_balance, &now.total_active_balance),
+                true,
+            ),
+            justified_balances: LookupCounts::from_results(
+                &delta(&self.justified_balances, &now.justified_balances),
+                false,
+            ),
+        }
+    }
+}
+
+/// `after - before` per result. Counters only go up, so a result missing from
+/// `before` started at zero.
+fn delta(before: &BTreeMap<String, u64>, after: &BTreeMap<String, u64>) -> BTreeMap<String, u64> {
+    after
+        .iter()
+        .map(|(result, value)| {
+            let start = before.get(result).copied().unwrap_or(0);
+            (result.clone(), value.saturating_sub(start))
+        })
+        .collect()
+}
+
+/// The seconds the `stf` import phase spent outside every STF step, or `None`
+/// when `stf` did not run.
+///
+/// The steps are nested inside `stf`, so their sum cannot exceed it; the clamp
+/// only absorbs the few microseconds two separate clock readings can disagree
+/// by, rather than reporting a negative duration.
+fn stf_unattributed(
+    phases: &BTreeMap<String, f64>,
+    stf_phases: &BTreeMap<String, f64>,
+) -> Option<f64> {
+    let stf = *phases.get("stf")?;
+    Some((stf - stf_phases.values().sum::<f64>()).max(0.0))
+}
+
 /// The phases a sample records: every per-block section, plus the
 /// per-arrival sections that do not enclose the others.
 ///
@@ -124,6 +212,11 @@ pub(crate) async fn replay_corpus(dir: &Path, options: &ReplayOptions) -> eyre::
 
     let mut samples = Vec::with_capacity(manifest.slots.len());
 
+    // The slot of the block each import is applied to. A corpus is linear
+    // (`fetch` refuses a block whose parent is not the previous one), so that
+    // is always the block imported just before, and the first is the anchor.
+    let mut parent_slot = manifest.anchor_slot;
+
     let warmup_total = manifest.warmup_slots.len();
     for (position, &slot) in manifest.warmup_slots.iter().enumerate() {
         let (block, block_root) = load_block(dir, &config, &mut clock, slot)?;
@@ -133,30 +226,58 @@ pub(crate) async fn replay_corpus(dir: &Path, options: &ReplayOptions) -> eyre::
             position + 1,
             format_ms(wall_seconds)
         );
+        parent_slot = slot;
         pause_between_blocks(options.block_delay).await;
     }
 
+    // Read once the warm-up is done, so a cache's warm-up misses (every cache
+    // starts cold) are not charged to the measured blocks' hit rates.
+    let caches_before = CacheReading::now();
+
     let total = manifest.slots.len();
     for (iteration, &slot) in manifest.slots.iter().enumerate() {
         let (block, block_root) = load_block(dir, &config, &mut clock, slot)?;
+        let class = BlockClass::classify(parent_slot, slot, preset::SLOTS_PER_EPOCH);
 
         let timer = PhaseTimer::start(IMPORT_PHASE_HISTOGRAM);
+        let stf_timer =
+            PhaseTimer::start_filtered(STF_PHASE_HISTOGRAM, ("context", STF_CONTEXT_IMPORT));
+        let cpu_before = process::cpu_seconds();
         let wall_seconds = import(&mut server, block, slot, block_root).await?;
+        // The CPU reading and the RSS go first, before the histogram reads,
+        // which gather the whole registry and would otherwise be in the CPU
+        // delta of the block they measure.
+        let cpu_seconds = cpu_before
+            .zip(process::cpu_seconds())
+            .map(|(before, after)| (after - before).max(0.0));
+        let rss_bytes = process::rss_bytes();
+        let peak_rss_bytes = process::peak_rss_bytes();
         let phases = timer.finish_at_most_once(sampled_phases())?;
+        let stf_phases = stf_timer.finish_summed(BEACON_STF_PHASES.iter().copied());
+        let stf_unattributed_seconds = stf_unattributed(&phases, &stf_phases);
 
         eprintln!(
-            "block {}/{total}: slot {slot} imported in {}",
+            "block {}/{total}: slot {slot} ({}) imported in {}",
             iteration + 1,
+            class.label(),
             format_ms(wall_seconds)
         );
         samples.push(Sample {
             iteration: iteration as u64,
             slot,
+            parent_slot,
+            class,
             block_root: hex_root(block_root),
             wall_seconds,
             phases,
+            stf_phases,
+            stf_unattributed_seconds,
+            cpu_seconds,
+            rss_bytes,
+            peak_rss_bytes,
             outcome: "imported",
         });
+        parent_slot = slot;
         // After `finish_at_most_once` and the sample are recorded, so the
         // sleep is in no phase delta and no `wall_seconds`.
         pause_between_blocks(options.block_delay).await;
@@ -166,6 +287,7 @@ pub(crate) async fn replay_corpus(dir: &Path, options: &ReplayOptions) -> eyre::
         Environment::collect(),
         params_from(&manifest, dir, options.block_delay),
         samples,
+        caches_before.deltas_since_start(),
     ))
 }
 
@@ -288,3 +410,55 @@ fn prepare_data_dir(data_dir: &Path, force: bool) -> eyre::Result<()> {
     std::fs::create_dir_all(data_dir)?;
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn map(entries: &[(&str, u64)]) -> BTreeMap<String, u64> {
+        entries
+            .iter()
+            .map(|(result, value)| ((*result).to_string(), *value))
+            .collect()
+    }
+
+    #[test]
+    fn a_counter_delta_excludes_what_was_counted_before_and_starts_new_results_at_zero() {
+        let before = map(&[("hit", 10), ("miss", 4)]);
+        let after = map(&[("hit", 15), ("miss", 4), ("unkeyable", 2)]);
+
+        let delta = delta(&before, &after);
+
+        assert_eq!(delta, map(&[("hit", 5), ("miss", 0), ("unkeyable", 2)]));
+    }
+
+    #[test]
+    fn the_unattributed_stf_time_is_what_the_steps_leave_of_the_stf_phase() {
+        let phases: BTreeMap<String, f64> = [("stf".to_string(), 1.0)].into_iter().collect();
+        let steps: BTreeMap<String, f64> = [
+            ("process_epoch".to_string(), 0.6),
+            ("state_root".to_string(), 0.3),
+        ]
+        .into_iter()
+        .collect();
+
+        let unattributed = stf_unattributed(&phases, &steps).expect("stf ran");
+        assert!((unattributed - 0.1).abs() < 1e-9);
+    }
+
+    #[test]
+    fn a_block_whose_stf_did_not_run_has_no_unattributed_time() {
+        // Absent, not zero: nothing ran, so nothing was left over.
+        assert_eq!(stf_unattributed(&BTreeMap::new(), &BTreeMap::new()), None);
+    }
+
+    #[test]
+    fn steps_that_overshoot_the_stf_phase_by_clock_skew_clamp_to_zero() {
+        let phases: BTreeMap<String, f64> = [("stf".to_string(), 1.0)].into_iter().collect();
+        let steps: BTreeMap<String, f64> = [("state_root".to_string(), 1.000_001)]
+            .into_iter()
+            .collect();
+
+        assert_eq!(stf_unattributed(&phases, &steps), Some(0.0));
+    }
+}