diff --git a/Cargo.lock b/Cargo.lock index d6b3314c..b0c8efea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1919,6 +1919,7 @@ dependencies = [ "clap", "ethlambda-blockchain", "ethlambda-crypto", + "ethlambda-metrics", "ethlambda-network-api", "ethlambda-p2p", "ethlambda-rpc", diff --git a/Makefile b/Makefile index d404100a..ee6c28dc 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: help fmt lint docker-build shadow-build shadow-docker-build run-devnet test docs docs-deps docs-serve +.PHONY: help fmt lint bench docker-build shadow-build shadow-docker-build run-devnet test docs docs-deps docs-serve help: ## ๐Ÿ“š Show help for each of the Makefile recipes @grep -E '^[a-zA-Z0-9_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}' @@ -14,6 +14,11 @@ test: leanSpec/fixtures ## ๐Ÿงช Run all tests # signature verification/aggregation, without paying for LTO on every rebuild cargo test --workspace --profile release-fast +BENCH_ARGS ?= synthetic --mock-crypto + +bench: ## ๐Ÿ Benchmark block building offline (override BENCH_ARGS to customize) + cargo run --release --bin ethlambda -- benchmark $(BENCH_ARGS) + GIT_COMMIT=$(shell git rev-parse HEAD) GIT_BRANCH=$(shell git rev-parse --abbrev-ref HEAD) DOCKER_TAG?=local diff --git a/bin/ethlambda/Cargo.toml b/bin/ethlambda/Cargo.toml index 94913342..e5853842 100644 --- a/bin/ethlambda/Cargo.toml +++ b/bin/ethlambda/Cargo.toml @@ -21,6 +21,7 @@ shadow-integration = ["ethlambda-crypto/shadow-integration"] [dependencies] ethlambda-blockchain.workspace = true ethlambda-crypto.workspace = true +ethlambda-metrics.workspace = true ethlambda-network-api.workspace = true ethlambda-p2p.workspace = true ethlambda-types.workspace = true diff --git a/bin/ethlambda/src/benchmark/corpus.rs b/bin/ethlambda/src/benchmark/corpus.rs new file mode 100644 index 00000000..8c28cbc6 --- /dev/null +++ b/bin/ethlambda/src/benchmark/corpus.rs @@ -0,0 +1,163 @@ +//! Synthetic benchmark corpus: deterministic validators, a genesis store, and +//! per-slot attestation-pool seeding. + +use std::sync::Arc; + +use ethlambda_blockchain::store::produce_attestation_data; +use ethlambda_storage::{Store, backend::InMemoryBackend}; +use ethlambda_types::{ + attestation::{AggregationBits, HashedAttestationData}, + block::SingleMessageAggregate, + state::{State, Validator, ValidatorPubkeyBytes}, +}; + +/// Fixed genesis time for synthetic runs. The harness derives every tick +/// timestamp from slot numbers relative to this value and never reads the wall +/// clock, so runs are reproducible at any time of day. +const GENESIS_TIME: u64 = 1_700_000_000; + +pub(crate) struct SyntheticCorpus { + num_validators: u64, + proofs_per_data: u64, +} + +impl SyntheticCorpus { + pub(crate) fn new(num_validators: u64, proofs_per_data: u64) -> Self { + Self { + num_validators, + proofs_per_data, + } + } + + /// Build a genesis store over an in-memory backend with `num_validators` + /// seed-derived validators. + /// + /// Pubkeys are deterministic placeholder bytes: in mock-crypto mode no code + /// path decodes them (signature verification is skipped and best-proof + /// compaction never resolves pubkeys). + pub(crate) fn genesis_store(&self, seed: u64) -> Store { + let mut rng_state = seed; + let validators = (0..self.num_validators) + .map(|index| Validator { + attestation_pubkey: synthetic_pubkey(&mut rng_state), + proposal_pubkey: synthetic_pubkey(&mut rng_state), + index, + }) + .collect(); + let genesis_state = State::from_genesis(GENESIS_TIME, validators); + Store::from_anchor_state(Arc::new(InMemoryBackend::new()), genesis_state) + } + + /// Seed the pending ("new") pool with the full validator set's attestations + /// for `attestation_slot`, split into `proofs_per_data` disjoint aggregates. + /// + /// Mirrors what committee aggregators gossip during a slot: several + /// aggregates for the same `AttestationData`, each covering a validator + /// subset. The proposal tick then promotes them to the known pool, exactly + /// as on a live node. Entries are inserted in a fixed order because pool + /// insertion order pins within-entry proof choice during selection. + /// + /// Returns the total number of pool entries the next build will see, across + /// both pools. + pub(crate) fn seed_pool( + &self, + store: &mut Store, + attestation_slot: u64, + ) -> eyre::Result { + let data = produce_attestation_data(store, attestation_slot); + let entries = participant_groups(self.num_validators, self.proofs_per_data) + .into_iter() + .map(|participants| { + ( + HashedAttestationData::new(data.clone()), + SingleMessageAggregate::empty(participants), + ) + }) + .collect(); + store.insert_new_aggregated_payloads_batch(entries); + + // The pending pool evicts whole data-root entries FIFO once its proof + // cap is exceeded, so an over-cap batch seeds nothing and every + // measured block would come out empty. + let pending = store.new_aggregated_payloads_count(); + eyre::ensure!( + pending > 0, + "attestations seeded for slot {attestation_slot} were evicted from the pending pool; \ + the measured workload would not match the requested parameters" + ); + Ok(pending + store.known_aggregated_payloads_count()) + } +} + +/// Partition validators 0..num_validators into `groups` disjoint bitfields, +/// assigning validator `i` to group `i % groups`. Every group is non-empty +/// (groups is capped at the validator count) and the union covers every +/// validator exactly once. +fn participant_groups(num_validators: u64, groups: u64) -> Vec { + let groups = groups.clamp(1, num_validators); + (0..groups) + .map(|group| { + let mut bits = AggregationBits::with_length(num_validators as usize) + .expect("validator count is within the bitlist limit"); + for index in (group..num_validators).step_by(groups as usize) { + bits.set(index as usize, true) + .expect("index is within the bitlist length"); + } + bits + }) + .collect() +} + +/// splitmix64: tiny deterministic generator for placeholder pubkey bytes, +/// avoiding a rand dependency. +fn splitmix64(state: &mut u64) -> u64 { + *state = state.wrapping_add(0x9e37_79b9_7f4a_7c15); + let mut z = *state; + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + z ^ (z >> 31) +} + +fn synthetic_pubkey(rng_state: &mut u64) -> ValidatorPubkeyBytes { + let mut bytes = [0u8; 52]; + for chunk in bytes.chunks_mut(8) { + let word = splitmix64(rng_state).to_le_bytes(); + chunk.copy_from_slice(&word[..chunk.len()]); + } + bytes +} + +#[cfg(test)] +mod tests { + use super::*; + use ethlambda_types::attestation::validator_indices; + + #[test] + fn participant_groups_partition_all_validators() { + for (validators, groups) in [(8u64, 2u64), (8, 3), (5, 8), (1, 1), (4096, 4)] { + let partition = participant_groups(validators, groups); + assert_eq!(partition.len() as u64, groups.min(validators)); + let mut seen = vec![0u32; validators as usize]; + for bits in &partition { + let indices: Vec = validator_indices(bits).collect(); + assert!(!indices.is_empty(), "every group must be non-empty"); + for index in indices { + seen[index as usize] += 1; + } + } + assert!( + seen.iter().all(|&count| count == 1), + "every validator must appear in exactly one group: {seen:?}" + ); + } + } + + #[test] + fn synthetic_pubkeys_are_deterministic() { + let mut a = 42u64; + let mut b = 42u64; + assert_eq!(synthetic_pubkey(&mut a), synthetic_pubkey(&mut b)); + let mut c = 43u64; + assert_ne!(synthetic_pubkey(&mut a), synthetic_pubkey(&mut c)); + } +} diff --git a/bin/ethlambda/src/benchmark/mod.rs b/bin/ethlambda/src/benchmark/mod.rs new file mode 100644 index 00000000..8d6da8f4 --- /dev/null +++ b/bin/ethlambda/src/benchmark/mod.rs @@ -0,0 +1,313 @@ +//! Offline block-building benchmark (`ethlambda benchmark`). +//! +//! Drives the exact production proposer path โ€” `produce_block_with_signatures`, +//! the same entry `BlockChainServer::propose_block` uses โ€” against a synthetic +//! in-memory chain, and reports per-phase timing distributions. Gossip publish +//! and the slot-alignment sleep are outside the measured span, matching the +//! node's own `lean_block_building_time_seconds` boundary. +//! +//! See docs/plans/block-building-benchmark.md for the design and roadmap +//! (real-crypto pools and replay-from-datadir land in later milestones). + +mod corpus; +mod report; + +use std::collections::{BTreeMap, HashMap}; +use std::time::Instant; + +use ethlambda_blockchain::block_builder::ProposerConfig; +use ethlambda_blockchain::metrics::BLOCK_PROPOSAL_ATTESTATION_BUILD_PHASES; +use ethlambda_blockchain::store::{on_block_without_verification, produce_block_with_signatures}; +use ethlambda_storage::{NEW_PAYLOAD_CAP, Store}; +use ethlambda_types::block::{MultiMessageAggregate, SignedBlock}; +use ethlambda_types::primitives::HashTreeRoot as _; +use eyre::WrapErr as _; + +use report::{Environment, Params, Report, Sample}; + +#[derive(Debug, clap::Args)] +pub(crate) struct BenchmarkOptions { + #[command(subcommand)] + workload: Workload, +} + +#[derive(Debug, clap::Subcommand)] +enum Workload { + /// Benchmark block building on a synthetic in-memory chain. + Synthetic(SyntheticOptions), +} + +#[derive(Debug, clap::Args)] +struct SyntheticOptions { + /// Number of validators in the synthetic genesis. + #[arg(long, default_value = "8", value_parser = clap::value_parser!(u64).range(1..=4096))] + num_validators: u64, + /// Unmeasured chain-advancement slots before measuring. Builds and imports + /// one block per slot so the measured builds run on a state with + /// representative historical roots and justifications, and warms the state + /// cache. + #[arg(long, default_value = "8")] + warmup_slots: u64, + /// Aggregate proofs seeded per AttestationData, mimicking committee + /// aggregators covering disjoint validator subsets. The default of 1 (one + /// full-coverage proof per data) keeps justification/finalization + /// advancing every slot. Higher values exercise multi-proof selection and + /// same-data collapse, but without --enable-proposer-aggregation the block + /// then carries only the best partial proof (< 2/3 coverage), so + /// justification stalls โ€” the real coverage cost of disabling proposer + /// aggregation. + #[arg(long, default_value = "1", value_parser = clap::value_parser!(u64).range(1..))] + proofs_per_data: u64, + /// Deterministic seed for the synthetic validator set. Two runs with the + /// same seed and parameters produce identical per-iteration block roots. + #[arg(long, default_value = "42")] + seed: u64, + #[command(flatten)] + common: CommonOptions, +} + +impl SyntheticOptions { + fn validate(&self) -> eyre::Result<()> { + eyre::ensure!( + self.common.mock_crypto, + "real-crypto benchmarking is not implemented yet; rerun with --mock-crypto" + ); + // The pending pool evicts whole data-root entries FIFO once its proof + // cap is exceeded, so a single slot's batch larger than the cap would + // silently seed nothing and every measured block would be empty. + eyre::ensure!( + self.proofs_per_data as usize <= NEW_PAYLOAD_CAP, + "--proofs-per-data {} exceeds the pending-pool capacity ({NEW_PAYLOAD_CAP}); \ + one slot's batch would be evicted whole and every measured block would be empty", + self.proofs_per_data + ); + Ok(()) + } +} + +impl From<&SyntheticOptions> for Params { + fn from(options: &SyntheticOptions) -> Self { + Self { + mode: "synthetic", + mock_crypto: options.common.mock_crypto, + num_validators: options.num_validators, + warmup_slots: options.warmup_slots, + proofs_per_data: options.proofs_per_data, + seed: options.seed, + iterations: options.common.iterations, + enable_proposer_aggregation: options.common.enable_proposer_aggregation, + max_attestations_per_block: options.common.max_attestations_per_block, + } + } +} + +#[derive(Debug, clap::Args)] +struct CommonOptions { + /// Measured iterations (one built block each), after warmup. + #[arg(long, default_value = "10", value_parser = clap::value_parser!(u64).range(1..))] + iterations: u64, + /// Seed pools with empty placeholder proofs instead of real XMSS/leanVM + /// crypto. Measures selection + best-proof compaction + state transition + /// only; runs in seconds. Conflicts with --enable-proposer-aggregation, + /// whose recursive aggregation needs real proof bytes. + #[arg(long, conflicts_with = "enable_proposer_aggregation")] + mock_crypto: bool, + /// Mirrors the node flag: collapse same-data proofs via recursive leanVM + /// aggregation instead of keeping the single best-coverage proof. + #[arg(long)] + enable_proposer_aggregation: bool, + /// Mirrors the node flag: distinct AttestationData cap per built block. + #[arg(long, default_value = "3")] + max_attestations_per_block: usize, +} + +pub(crate) fn run(options: BenchmarkOptions) -> eyre::Result<()> { + let Workload::Synthetic(synthetic) = options.workload; + run_synthetic(synthetic) +} + +fn run_synthetic(options: SyntheticOptions) -> eyre::Result<()> { + options.validate()?; + let common = &options.common; + + let proposer_config = ProposerConfig { + enable_proposer_aggregation: common.enable_proposer_aggregation, + max_attestations_per_block: common.max_attestations_per_block, + }; + let corpus = corpus::SyntheticCorpus::new(options.num_validators, options.proofs_per_data); + let mut store = corpus.genesis_store(options.seed); + + let total_slots = options + .warmup_slots + .checked_add(common.iterations) + .ok_or_else(|| eyre::eyre!("--warmup-slots plus --iterations overflows u64"))?; + + let mut samples = Vec::with_capacity(common.iterations as usize); + for slot in 1..=total_slots { + let sample = build_one_slot( + &corpus, + &mut store, + slot, + options.num_validators, + proposer_config, + )?; + let measured = slot > options.warmup_slots; + log_progress(slot, total_slots, measured, &sample); + if measured { + samples.push(Sample { + iteration: slot - options.warmup_slots, + ..sample + }); + } + } + eyre::ensure!( + samples.len() as u64 == common.iterations, + "collected {} samples but expected {}; the measured-slot accounting drifted", + samples.len(), + common.iterations + ); + + let report = Report::new(Environment::collect(), Params::from(&options), samples); + println!("{}", report.human_table()); + + Ok(()) +} + +/// One line per built slot, on stderr so the report keeps stdout to itself. +fn log_progress(slot: u64, total_slots: u64, measured: bool, sample: &Sample) { + let label = if measured { "measured" } else { "warmup" }; + eprintln!( + "[{slot}/{total_slots}] {label}: built block in {:.3}ms \ + (attestations={}, pool_entries={})", + sample.wall_seconds * 1e3, + sample.attestations_packed, + sample.pool_entries, + ); +} + +/// Seed the pool, build one block the way the proposer does, and import it. +/// +/// The returned sample carries `iteration: 0`; the caller sets it for the slots +/// it keeps. Warmup and measured slots do exactly the same work โ€” only whether +/// the sample is kept differs โ€” so there is one code path for both. +fn build_one_slot( + corpus: &corpus::SyntheticCorpus, + store: &mut Store, + slot: u64, + num_validators: u64, + proposer_config: ProposerConfig, +) -> eyre::Result { + // Seed the pending pool with the previous slot's attestations, exactly + // where gossip aggregates would sit before the proposal tick promotes them + // to the known pool. Entries from earlier slots stay in the known pool, as + // they would on a live node. + let pool_entries = corpus.seed_pool(store, slot - 1)?; + + // Round-robin proposer, matching `is_proposer`. + let proposer = slot % num_validators; + + let phases = PhaseTimer::start(); + let build_start = Instant::now(); + let (block, aggregates, _checkpoints) = + produce_block_with_signatures(store, slot, proposer, proposer_config) + .wrap_err_with(|| format!("block build failed at slot {slot}"))?; + let wall_seconds = build_start.elapsed().as_secs_f64(); + let phases = phases.finish()?; + + let block_root = block.hash_tree_root(); + let attestations_packed = block.body.attestations.len(); + let aggregates_count = aggregates.len(); + + // Import the built block (outside the measured span) so the next iteration + // builds one slot ahead of head, like a live proposer; building repeatedly + // on a fixed head would make `process_slots` cost grow with the iteration + // index. + let signed_block = SignedBlock { + message: block, + proof: MultiMessageAggregate::default(), + }; + on_block_without_verification(store, signed_block) + .wrap_err_with(|| format!("importing the built block failed at slot {slot}"))?; + + // Clamped: the unattributed preamble makes the remainder positive in + // practice, but summing many small phase values can round just above the + // wall measurement, and a negative overhead would read as an accounting bug. + let overhead_seconds = (wall_seconds - phases.values().sum::()).max(0.0); + + Ok(Sample { + iteration: 0, + slot, + proposer, + block_root: format!("0x{}", hex::encode(block_root.0)), + wall_seconds, + phases, + overhead_seconds, + attestations_packed, + aggregates: aggregates_count, + pool_entries, + }) +} + +const PHASE_HISTOGRAM: &str = "lean_block_proposal_attestation_build_phase_seconds"; + +/// Exact per-phase durations for one block build, taken from the block-proposal +/// phase histogram in the default prometheus registry. +/// +/// Histogram sums accumulate the raw f64 seconds of every observation, so the +/// difference between two readings IS the build's phase time โ€” bucket +/// boundaries play no role, and the hot path needs no extra instrumentation. +struct PhaseTimer { + /// Per-phase (sample_sum, sample_count) before the build. + before: HashMap, +} + +impl PhaseTimer { + fn start() -> Self { + Self { before: read() } + } + + /// Per-phase durations since [`PhaseTimer::start`]. + /// + /// Each phase must have been observed exactly once โ€” one `build_block` in + /// this single-threaded process โ€” so anything else means the accounting + /// drifted and attribution would be wrong. That is a hard error, not a + /// warning: a silently mis-attributed report is worse than no report. + fn finish(self) -> eyre::Result> { + let after = read(); + let mut phases = BTreeMap::new(); + for &phase in BLOCK_PROPOSAL_ATTESTATION_BUILD_PHASES { + let (sum_before, count_before) = self.before.get(phase).copied().unwrap_or((0.0, 0)); + let (sum_after, count_after) = after.get(phase).copied().unwrap_or((0.0, 0)); + let observations = count_after.saturating_sub(count_before); + eyre::ensure!( + observations == 1, + "phase '{phase}' was observed {observations} times during one build \ + (expected 1); phase attribution would be wrong" + ); + phases.insert(phase.to_string(), sum_after - sum_before); + } + Ok(phases) + } +} + +/// Current (sample_sum, sample_count) per phase label. +fn read() -> HashMap { + ethlambda_metrics::gather() + .iter() + .filter(|family| family.name() == PHASE_HISTOGRAM) + .flat_map(|family| family.get_metric()) + .filter_map(|metric| { + let phase = metric + .get_label() + .iter() + .find(|label| label.name() == "phase")? + .value() + .to_string(); + let histogram = metric.get_histogram(); + Some(( + phase, + (histogram.get_sample_sum(), histogram.get_sample_count()), + )) + }) + .collect() +} diff --git a/bin/ethlambda/src/benchmark/report.rs b/bin/ethlambda/src/benchmark/report.rs new file mode 100644 index 00000000..2fce8cc4 --- /dev/null +++ b/bin/ethlambda/src/benchmark/report.rs @@ -0,0 +1,150 @@ +//! Report emission for the block-building benchmark. +//! +//! Every measured iteration is reported on its own row: outliers are never +//! discarded (XMSS signing and OTS window advancement produce legitimate heavy +//! tails worth inspecting), and the per-iteration block roots let a +//! baseline-vs-optimized diff prove an optimization changed only speed, not +//! which attestations get selected. + +use std::collections::BTreeMap; +use std::fmt::Write as _; + +use serde::Serialize; + +use crate::version; + +#[derive(Debug, Serialize)] +pub(crate) struct Sample { + pub iteration: u64, + pub slot: u64, + pub proposer: u64, + /// Determinism checksum: same seed + params must reproduce the same roots. + pub block_root: String, + pub wall_seconds: f64, + /// Per-phase seconds from histogram sum deltas. + pub phases: BTreeMap, + /// Wall time not attributed to any phase: the `produce_block_with_signatures` + /// preamble (tick advance, pool promotion, fork-choice head update, pool + /// deep-clone, block-roots scan) plus measurement slack. + pub overhead_seconds: f64, + pub attestations_packed: usize, + pub aggregates: usize, + /// Pool entries (new + known) visible to this build; reported so pool + /// growth across iterations is visible in the samples. + pub pool_entries: usize, +} + +#[derive(Debug, Serialize)] +pub(crate) struct Environment { + pub client_version: &'static str, + pub os: &'static str, + pub arch: &'static str, + pub available_parallelism: usize, +} + +impl Environment { + pub(crate) fn collect() -> Self { + Self { + client_version: version::CLIENT_VERSION, + os: std::env::consts::OS, + arch: std::env::consts::ARCH, + available_parallelism: std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(0), + } + } +} + +#[derive(Debug, Serialize)] +pub(crate) struct Params { + pub mode: &'static str, + pub mock_crypto: bool, + pub num_validators: u64, + pub warmup_slots: u64, + pub proofs_per_data: u64, + pub seed: u64, + pub iterations: u64, + pub enable_proposer_aggregation: bool, + pub max_attestations_per_block: usize, +} + +#[derive(Debug, Serialize)] +pub(crate) struct Report { + pub environment: Environment, + pub params: Params, + pub samples: Vec, +} + +impl Report { + pub(crate) fn new(environment: Environment, params: Params, samples: Vec) -> Self { + Self { + environment, + params, + samples, + } + } + + pub(crate) fn human_table(&self) -> String { + let mut out = String::new(); + let params = &self.params; + let env = &self.environment; + let crypto = if params.mock_crypto { "mock" } else { "real" }; + let _ = writeln!( + out, + "Block-building benchmark โ€” {} workload ({crypto} crypto)", + params.mode + ); + let _ = writeln!( + out, + " validators={} warmup_slots={} iterations={} proofs_per_data={} seed={}", + params.num_validators, + params.warmup_slots, + params.iterations, + params.proofs_per_data, + params.seed + ); + let _ = writeln!( + out, + " enable_proposer_aggregation={} max_attestations_per_block={}", + params.enable_proposer_aggregation, params.max_attestations_per_block + ); + let _ = writeln!( + out, + " {} os={} arch={} threads={}", + env.client_version, env.os, env.arch, env.available_parallelism + ); + let _ = writeln!(out); + + // Phase columns come from the first sample: every build observes the + // same phases, and `run_synthetic` asserts each advanced exactly once. + let phases: Vec<&String> = match self.samples.first() { + Some(sample) => sample.phases.keys().collect(), + None => return out, + }; + let _ = write!(out, " {:<5}", "iter"); + for phase in &phases { + let _ = write!(out, " {phase:>16}"); + } + let _ = writeln!(out, " {:>10} {:>10} {:>12}", "overhead", "wall", "root"); + + for sample in &self.samples { + let _ = write!(out, " {:<5}", sample.iteration); + for phase in &phases { + let seconds = sample.phases.get(*phase).copied().unwrap_or(0.0); + let _ = write!(out, " {:>16}", format_ms(seconds)); + } + let _ = writeln!( + out, + " {:>10} {:>10} {:>12}", + format_ms(sample.overhead_seconds), + format_ms(sample.wall_seconds), + &sample.block_root[..10], + ); + } + out + } +} + +fn format_ms(seconds: f64) -> String { + format!("{:.3}ms", seconds * 1e3) +} diff --git a/bin/ethlambda/src/cli.rs b/bin/ethlambda/src/cli.rs index d8bb6963..9fa7d581 100644 --- a/bin/ethlambda/src/cli.rs +++ b/bin/ethlambda/src/cli.rs @@ -242,8 +242,10 @@ mod tests { "ethlambda_0", ]; argv.extend_from_slice(extra); - let Command::Node(options) = try_parse_from(argv).expect("node options parse"); - options + match try_parse_from(argv).expect("node options parse") { + Command::Node(options) => *options, + other => panic!("expected a node invocation, got {other:?}"), + } } /// `--discovery.enable` on its own has to work: a default that is never diff --git a/bin/ethlambda/src/command.rs b/bin/ethlambda/src/command.rs index 8209bc9a..034646a6 100644 --- a/bin/ethlambda/src/command.rs +++ b/bin/ethlambda/src/command.rs @@ -1,8 +1,8 @@ //! Sub-command definition and dispatch. //! -//! `node` is an ordinary clap sub-command, so clap owns its help, usage lines -//! and error messages. The one thing clap cannot express is a *default* -//! sub-command, and the node needs one: the Dockerfile, +//! `node` and `benchmark` are ordinary clap sub-commands, so clap owns their +//! help, usage lines and error messages. The one thing clap cannot express is a +//! *default* sub-command, and the node needs one: the Dockerfile, //! lean-quickstart, the hive shim and the devnet skills all invoke the binary as //! a bare list of node flags, from before there was anything else to run. That //! form keeps working because a missing sub-command is filled in as `node` @@ -12,14 +12,16 @@ use std::ffi::OsString; use clap::Parser; +use crate::benchmark::BenchmarkOptions; use crate::cli::NodeOptions; use crate::version; /// Tokens that already say what to run, so no default is inserted ahead of /// them. `help` is clap's own generated sub-command (`ethlambda help node`). -const EXPLICIT: &[&str] = &[NODE, "help", "-h", "--help", "-V", "--version"]; +const EXPLICIT: &[&str] = &[NODE, BENCHMARK, "help", "-h", "--help", "-V", "--version"]; const NODE: &str = "node"; +const BENCHMARK: &str = "benchmark"; #[derive(Debug, clap::Parser)] #[command( @@ -50,8 +52,18 @@ pub(crate) enum Command { // printing `ethlambda ` after node flags, as it did when the node // flags were the whole command line. It is still listed and invoked as // `node`. + // Boxed because the node options dwarf every other variant's payload + // (~312 bytes against ~80), which `clippy::large_enum_variant` rightly + // flags: one allocation per process is cheaper than carrying that size in + // every value of this enum. #[command(display_name = "ethlambda")] - Node(NodeOptions), + // Boxed because the node options dwarf the other variant's payload (~312 + // bytes against ~80), which `clippy::large_enum_variant` rightly flags: one + // allocation per process is cheaper than carrying that size in every value + // of this enum. + Node(Box), + /// Benchmark block building offline against a controlled workload. + Benchmark(BenchmarkOptions), } /// Parse the process arguments, exiting the way clap does on a parse error, @@ -123,9 +135,10 @@ mod tests { } fn node_options(args: &[&str]) -> NodeOptions { - let Command::Node(options) = - try_parse_from(args.iter().map(OsString::from)).expect("invocation parses"); - options + match try_parse_from(args.iter().map(OsString::from)).expect("invocation parses") { + Command::Node(options) => *options, + other => panic!("expected a node invocation, got {other:?}"), + } } #[test] diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index a32ac594..8314bfc0 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -1,3 +1,4 @@ +mod benchmark; mod checkpoint_sync; mod cli; mod command; @@ -32,8 +33,8 @@ use std::{ }; use tokio_util::sync::CancellationToken; +use cli::NodeOptions; use command::Command; - use ethlambda_blockchain::MILLISECONDS_PER_SLOT; use ethlambda_blockchain::block_builder::ProposerConfig; use ethlambda_blockchain::key_manager::ValidatorKeyPair; @@ -68,21 +69,54 @@ const ASCII_ART: &str = r#" \___|\__|_| |_|_|\__,_|_| |_| |_|_.__/ \__,_|\__,_| "#; -// Shadow single-steps execution in a discrete-event simulation, so the default -// multi-threaded runtime's worker threads add only scheduling noise, never -// parallelism. Use a single-threaded runtime under Shadow. This is an -// optimization, not a correctness requirement. -#[cfg_attr(not(feature = "shadow-integration"), tokio::main)] -#[cfg_attr(feature = "shadow-integration", tokio::main(flavor = "current_thread"))] -async fn main() -> eyre::Result<()> { +fn main() -> eyre::Result<()> { + match command::parse() { + Command::Node(options) => { + init_node_logging()?; + run_node(*options) + } + // The benchmark is synchronous, CPU-bound work, so it runs on this + // thread and the tokio runtime is never started โ€” rather than parking + // a worker thread for the whole run. + Command::Benchmark(options) => { + init_benchmark_logging()?; + benchmark::run(options) + } + } +} + +/// Node logging: INFO and above, on stdout. +fn init_node_logging() -> eyre::Result<()> { let filter = EnvFilter::builder() .with_default_directive(tracing::Level::INFO.into()) .from_env_lossy(); let subscriber = Registry::default().with(tracing_subscriber::fmt::layer().with_filter(filter)); tracing::subscriber::set_global_default(subscriber) - .wrap_err("failed to set global tracing subscriber")?; + .wrap_err("failed to set global tracing subscriber") +} - let Command::Node(options) = command::parse(); +/// Benchmark logging: WARN and above, on stderr, so that the report owns stdout +/// and stays pipe-clean for `--format json | jq`. +fn init_benchmark_logging() -> eyre::Result<()> { + let filter = EnvFilter::builder() + .with_default_directive(tracing::Level::WARN.into()) + .from_env_lossy(); + let subscriber = Registry::default().with( + tracing_subscriber::fmt::layer() + .with_writer(std::io::stderr) + .with_filter(filter), + ); + tracing::subscriber::set_global_default(subscriber) + .wrap_err("failed to set global tracing subscriber") +} + +// Shadow single-steps execution in a discrete-event simulation, so the default +// multi-threaded runtime's worker threads add only scheduling noise, never +// parallelism. Use a single-threaded runtime under Shadow. This is an +// optimization, not a correctness requirement. +#[cfg_attr(not(feature = "shadow-integration"), tokio::main)] +#[cfg_attr(feature = "shadow-integration", tokio::main(flavor = "current_thread"))] +async fn run_node(options: NodeOptions) -> eyre::Result<()> { options.validate_discovery()?; #[cfg(feature = "shadow-integration")] diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index ffd10301..95da4df6 100644 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -8,4 +8,6 @@ pub use api::{ALL_TABLES, StorageBackend, StorageReadView, StorageWriteBatch, Ta /// Error type returned by the fallible [`Store`] operations, exported so /// callers can match on it (e.g. to distinguish [`Error::GenesisMismatch`]). pub use error::Error; -pub use store::{ForkCheckpoints, GetForkchoiceStoreError, MAX_RESUMABLE_DB_STATE_AGE, Store}; +pub use store::{ + ForkCheckpoints, GetForkchoiceStoreError, MAX_RESUMABLE_DB_STATE_AGE, NEW_PAYLOAD_CAP, Store, +}; diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 05621f8a..914b2cd1 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -125,7 +125,9 @@ const AGGREGATED_PAYLOAD_CAP: usize = 512; /// Hard cap for the new (pending) aggregated payload buffer. /// Smaller than known since new payloads are drained every interval (~4s). -const NEW_PAYLOAD_CAP: usize = 64; +/// Public so pool-seeding callers (the block-building benchmark) can reject +/// workloads that a single insertion batch would silently evict. +pub const NEW_PAYLOAD_CAP: usize = 64; /// Hard cap for the gossip signature buffer (individual signatures, not distinct data_roots). /// With 4 validators and 4-second slots, 2048 signatures covers ~512 slots (~34 min).