diff --git a/crates/blockchain/src/epoch_precompute.rs b/crates/blockchain/src/epoch_precompute.rs index 9c77e137a..be592baa5 100644 --- a/crates/blockchain/src/epoch_precompute.rs +++ b/crates/blockchain/src/epoch_precompute.rs @@ -221,6 +221,7 @@ impl BlockChainServer { let config = self.store.config(); let actor = ctx.actor_ref(); tokio::task::spawn_blocking(move || { + let _running = crate::metrics::EpochPrecomputeRunning::start(); let result = { let _timing = stf_metrics::time_epoch_precompute(); fork_choice::advance_to_epoch_start(&head_state, key.epoch, &config) diff --git a/crates/blockchain/src/metrics.rs b/crates/blockchain/src/metrics.rs index 34c67398f..b87362a82 100644 --- a/crates/blockchain/src/metrics.rs +++ b/crates/blockchain/src/metrics.rs @@ -118,6 +118,35 @@ impl SlotPosition { // --- Gauges --- +static LEAN_BEACON_EPOCH_PRECOMPUTE_RUNNING: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_int_gauge!( + "lean_beacon_epoch_precompute_running", + "1 while an epoch precompute worker is running its advance, 0 otherwise" + ) + .unwrap() + }); + +/// Marks an epoch precompute worker as running in +/// `lean_beacon_epoch_precompute_running` until dropped. A guard rather than a +/// set/unset pair so a panicking worker still clears the gauge. At most one +/// worker runs at a time (`epoch_precompute_in_flight`), so a plain 0/1 gauge +/// is exact. +pub struct EpochPrecomputeRunning; + +impl EpochPrecomputeRunning { + pub fn start() -> Self { + LEAN_BEACON_EPOCH_PRECOMPUTE_RUNNING.set(1); + Self + } +} + +impl Drop for EpochPrecomputeRunning { + fn drop(&mut self) { + LEAN_BEACON_EPOCH_PRECOMPUTE_RUNNING.set(0); + } +} + static LEAN_HEAD_SLOT: std::sync::LazyLock = std::sync::LazyLock::new(|| { register_int_gauge!("lean_head_slot", "Latest slot of the lean chain").unwrap() }); diff --git a/crates/net/p2p/src/beacon/column_checks.rs b/crates/net/p2p/src/beacon/column_checks.rs index 91fbd1efe..48025e040 100644 --- a/crates/net/p2p/src/beacon/column_checks.rs +++ b/crates/net/p2p/src/beacon/column_checks.rs @@ -46,12 +46,18 @@ pub(crate) fn check_and_forward(server: &P2PServer, sidecars: Vec bool { @@ -103,15 +103,21 @@ pub(crate) fn check_and_forward_then( for envelope in envelopes { // The semaphore is never closed, so this only returns once a // permit is free. + let waiting_since = std::time::Instant::now(); let Ok(permit) = permits.clone().acquire_owned().await else { if let Some(done) = done { done(); } return; }; + metrics::observe_column_check_permit_wait("envelope_check", waiting_since.elapsed()); + let in_use = metrics::PoolInUse::new("column_check"); + let task = metrics::BlockingTask::new("envelope_check"); let store = store.clone(); checks.push(tokio::task::spawn_blocking(move || { let _permit = permit; + let _in_use = in_use; + let _run = task.start(); let outcome = judge(&store, &envelope); (envelope, outcome) })); diff --git a/crates/net/p2p/src/beacon/verdict.rs b/crates/net/p2p/src/beacon/verdict.rs index ab5651c26..ad44a84c9 100644 --- a/crates/net/p2p/src/beacon/verdict.rs +++ b/crates/net/p2p/src/beacon/verdict.rs @@ -601,20 +601,26 @@ fn permits_for<'a>( server: &'a P2PServer, object: &Validated, ) -> &'a std::sync::Arc { + match pool_of(object) { + "gossip" => &server.gossip_validation_permits, + "attestation" => &server.attestation_validation_permits, + "builder" => &server.builder_validation_permits, + _ => &server.sync_validation_permits, + } +} + +/// The label (see [`metrics::VALIDATION_POOLS`]) of the permit pool +/// [`permits_for`] draws `object`'s permit from. The one `match` on the +/// object kind, so the pool metrics and the pool actually used cannot differ. +fn pool_of(object: &Validated) -> &'static str { match object { - Validated::Block { .. } | Validated::Column(_) | Validated::Envelope(_) => { - &server.gossip_validation_permits - } + Validated::Block { .. } | Validated::Column(_) | Validated::Envelope(_) => "gossip", Validated::Aggregate { .. } | Validated::Attestation { .. } | Validated::PayloadAttestation(_) - | Validated::Operation(_) => &server.attestation_validation_permits, - Validated::ExecutionPayloadBid { .. } | Validated::ProposerPreferences(_) => { - &server.builder_validation_permits - } - Validated::SyncCommitteeMessage { .. } | Validated::SyncContribution(_) => { - &server.sync_validation_permits - } + | Validated::Operation(_) => "attestation", + Validated::ExecutionPayloadBid { .. } | Validated::ProposerPreferences(_) => "builder", + Validated::SyncCommitteeMessage { .. } | Validated::SyncContribution(_) => "sync", } } @@ -643,7 +649,9 @@ pub(crate) fn spawn_stateful_checks( id: GossipId, object: Validated, ) { + let pool = pool_of(&object); let Ok(permit) = permits_for(server, &object).clone().try_acquire_owned() else { + metrics::inc_validation_permit_exhausted(pool); let received_at = id.received_at; let outcome = Outcome::Ignore(IgnoreReason::Overloaded); report(server, id, outcome); @@ -652,8 +660,12 @@ pub(crate) fn spawn_stateful_checks( }; let store = server.store.clone(); let actor = ctx.actor_ref(); + let in_use = metrics::PoolInUse::new(pool); + let task = metrics::BlockingTask::new(id.kind); tokio::task::spawn_blocking(move || { let _permit = permit; + let _in_use = in_use; + let _run = task.start(); let mut object = object; let outcome = guarded(|| object.stateful_checks(&store)); let _ = actor diff --git a/crates/net/p2p/src/metrics.rs b/crates/net/p2p/src/metrics.rs index 643300967..ab74833c0 100644 --- a/crates/net/p2p/src/metrics.rs +++ b/crates/net/p2p/src/metrics.rs @@ -450,6 +450,10 @@ pub fn init() { for &reason in DATA_COLUMN_REJECT_REASONS { LEAN_DATA_COLUMNS_REJECTED_TOTAL.with_label_values(&[reason]); } + for &pool in VALIDATION_POOLS { + LEAN_BEACON_VALIDATION_PERMITS_IN_USE.with_label_values(&[pool]); + LEAN_BEACON_VALIDATION_PERMIT_EXHAUSTED_TOTAL.with_label_values(&[pool]); + } } /// Count one beacon gossip verdict and how long it took from arrival. @@ -467,6 +471,151 @@ pub fn observe_beacon_gossip_verdict( .observe(elapsed.as_secs_f64()); } +/// Labels of the permit pools that bound blocking-thread validation work. +/// `column_check` is [`P2PServer::column_check_permits`](crate::P2PServer), +/// shared by the fetched-column and fetched-envelope checks; the rest are the +/// four gossip pools. +pub(crate) const VALIDATION_POOLS: &[&str] = + &["gossip", "attestation", "builder", "sync", "column_check"]; + +static LEAN_BEACON_VALIDATION_PERMITS_IN_USE: LazyLock = LazyLock::new(|| { + register_int_gauge_vec!( + "lean_beacon_validation_permits_in_use", + "Permits held right now, per validation permit pool (one per blocking task queued or running)", + &["pool"] + ) + .unwrap() +}); + +static LEAN_BEACON_VALIDATION_PERMIT_EXHAUSTED_TOTAL: LazyLock = LazyLock::new( + || { + register_int_counter_vec!( + "lean_beacon_validation_permit_exhausted_total", + "Gossip objects reported Ignore(Overloaded) because their permit pool had no free permit", + &["pool"] + ) + .unwrap() + }, +); + +/// Buckets for blocking-pool queue delay and closure run time: from a +/// scheduler hop to a verdict that has eaten a whole gossip deadline. +const VALIDATION_TASK_BUCKETS: &[f64] = &[ + 0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 4.0, +]; + +static LEAN_BEACON_VALIDATION_QUEUE_SECONDS: LazyLock = LazyLock::new(|| { + register_histogram_vec!( + "lean_beacon_validation_queue_seconds", + "Time from spawn_blocking being called for a validation task to its closure starting", + &["kind"], + VALIDATION_TASK_BUCKETS.to_vec() + ) + .unwrap() +}); + +static LEAN_BEACON_VALIDATION_RUN_SECONDS: LazyLock = LazyLock::new(|| { + register_histogram_vec!( + "lean_beacon_validation_run_seconds", + "Run time of a validation task's blocking closure", + &["kind"], + VALIDATION_TASK_BUCKETS.to_vec() + ) + .unwrap() +}); + +static LEAN_BEACON_COLUMN_CHECK_PERMIT_WAIT_SECONDS: LazyLock = LazyLock::new(|| { + register_histogram_vec!( + "lean_beacon_column_check_permit_wait_seconds", + "Time a fetched sidecar or envelope waited for a column_check permit", + &["kind"], + VALIDATION_TASK_BUCKETS.to_vec() + ) + .unwrap() +}); + +/// A held permit of the pool `pool`, counted in +/// `lean_beacon_validation_permits_in_use` for as long as the guard lives. +/// Move it into the blocking closure next to the permit itself, so the gauge +/// falls when the permit is released (including on a panic's unwind). +pub(crate) struct PoolInUse(&'static str); + +impl PoolInUse { + pub(crate) fn new(pool: &'static str) -> Self { + LEAN_BEACON_VALIDATION_PERMITS_IN_USE + .with_label_values(&[pool]) + .inc(); + Self(pool) + } +} + +impl Drop for PoolInUse { + fn drop(&mut self) { + LEAN_BEACON_VALIDATION_PERMITS_IN_USE + .with_label_values(&[self.0]) + .dec(); + } +} + +/// Count one object refused for want of a permit in `pool`. +pub(crate) fn inc_validation_permit_exhausted(pool: &str) { + LEAN_BEACON_VALIDATION_PERMIT_EXHAUSTED_TOTAL + .with_label_values(&[pool]) + .inc(); +} + +/// Record how long a fetched sidecar or envelope (`kind`) awaited a +/// `column_check` permit. +pub(crate) fn observe_column_check_permit_wait(kind: &str, waited: std::time::Duration) { + LEAN_BEACON_COLUMN_CHECK_PERMIT_WAIT_SECONDS + .with_label_values(&[kind]) + .observe(waited.as_secs_f64()); +} + +/// Times one `spawn_blocking` validation task. Create it immediately before +/// the `spawn_blocking` call, move it into the closure and call +/// [`Self::start`] first thing there: the gap is the tokio blocking pool's +/// queueing, and the returned guard observes the closure's run time when it +/// drops, a panic's unwind included. +pub(crate) struct BlockingTask { + kind: &'static str, + spawned_at: std::time::Instant, +} + +impl BlockingTask { + pub(crate) fn new(kind: &'static str) -> Self { + Self { + kind, + spawned_at: std::time::Instant::now(), + } + } + + pub(crate) fn start(self) -> BlockingRun { + let started_at = std::time::Instant::now(); + LEAN_BEACON_VALIDATION_QUEUE_SECONDS + .with_label_values(&[self.kind]) + .observe(started_at.duration_since(self.spawned_at).as_secs_f64()); + BlockingRun { + kind: self.kind, + started_at, + } + } +} + +/// Observes a blocking validation closure's run time on drop. +pub(crate) struct BlockingRun { + kind: &'static str, + started_at: std::time::Instant, +} + +impl Drop for BlockingRun { + fn drop(&mut self) { + LEAN_BEACON_VALIDATION_RUN_SECONDS + .with_label_values(&[self.kind]) + .observe(self.started_at.elapsed().as_secs_f64()); + } +} + /// Count one verdict gossipsub could no longer act on. pub fn inc_beacon_gossip_verdict_expired(kind: &str) { LEAN_BEACON_GOSSIP_VERDICT_EXPIRED_TOTAL @@ -685,4 +834,41 @@ mod tests { assert_eq!(gauge.with_label_values(&["bbbbbbbb"]).get(), 1); assert_eq!(gauge.with_label_values(&["aaaaaaaa"]).get(), 0); } + + #[test] + fn a_held_permit_is_counted_until_its_guard_drops() { + // A pool label no real pool uses, so parallel tests cannot move it. + let gauge = || { + LEAN_BEACON_VALIDATION_PERMITS_IN_USE + .with_label_values(&["test_pool"]) + .get() + }; + let first = PoolInUse::new("test_pool"); + let second = PoolInUse::new("test_pool"); + assert_eq!(gauge(), 2); + drop(first); + assert_eq!(gauge(), 1); + drop(second); + assert_eq!(gauge(), 0); + } + + #[test] + fn a_blocking_task_observes_queue_delay_and_run_time_once() { + let queue = || { + LEAN_BEACON_VALIDATION_QUEUE_SECONDS + .with_label_values(&["test_kind"]) + .get_sample_count() + }; + let run = || { + LEAN_BEACON_VALIDATION_RUN_SECONDS + .with_label_values(&["test_kind"]) + .get_sample_count() + }; + let task = BlockingTask::new("test_kind"); + assert_eq!((queue(), run()), (0, 0)); + let running = task.start(); + assert_eq!((queue(), run()), (1, 0)); + drop(running); + assert_eq!((queue(), run()), (1, 1)); + } } diff --git a/docs/beacon_wire.md b/docs/beacon_wire.md index c2138785b..b1d3484ec 100644 --- a/docs/beacon_wire.md +++ b/docs/beacon_wire.md @@ -183,7 +183,10 @@ operation topics. A mainnet slot carries up to `MAX_COMMITTEES_PER_SLOT * TARGET_AGGREGATORS_PER_COMMITTEE` aggregates alone, arriving every slot rather than only during a range sync, which is why that traffic needs a pool of its -own rather than sharing the block and column one. Accept propagates the +own rather than sharing the block and column one. Each pool's occupancy, +refusals and blocking-pool queue delay are metrics, with the queries that tell +whether validation competes with block import: see "Validation contention" in +[`metrics.md`](metrics.md). Accept propagates the message; a message whose dependency is not ready yet is IGNOREd. See [Aggregate attestations](#aggregate-attestations) for the two topics with their own section. diff --git a/docs/metrics.md b/docs/metrics.md index 9ac92fd4f..6000f4aa1 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -439,9 +439,67 @@ payload votes, and `builder_validation_permits` for bids and proposer preferences. They are deliberately separate: a mainnet slot's worth of aggregates and backbone attestations arrives every slot, not only during a range sync, and sharing one pool would let that burst answer `Ignore(Overloaded)` for a block or a column -instead. Neither pool has a metric of its own yet; a permit exhausted on -either shows up as `outcome="ignore",reason="overloaded"` on -`lean_beacon_gossip_validation_total`, for the topics that draw from it. +instead. `sync_validation_permits` (sync committee messages and contributions) +is a fourth, and `column_check_permits` bounds the checks on fetched columns and +envelopes, which await a permit instead of failing. A permit exhausted on a +gossip pool shows up as `outcome="ignore",reason="overloaded"` on +`lean_beacon_gossip_validation_total` and, per pool, on +`lean_beacon_validation_permit_exhausted_total` below. + +#### Validation contention + +These answer whether validation competes with block import and epoch precompute +for CPU. They only measure: no pool size, schedule or verdict depends on them. +ethlambda-specific, not part of the leanMetrics spec. + +| Name | Type | Usage | Sample collection event | Labels | Buckets | +|------|------|-------|-------------------------|--------|---------| +| `lean_beacon_validation_permits_in_use` | Gauge | Permits held per pool: blocking tasks queued or running | A permit is acquired / released | pool=gossip,attestation,builder,sync,column_check | | +| `lean_beacon_validation_permit_exhausted_total` | Counter | Objects reported `Ignore(Overloaded)` because their pool had no free permit | On each failed `try_acquire` | pool=gossip,attestation,builder,sync | | +| `lean_beacon_validation_queue_seconds` | Histogram | Time from `spawn_blocking` being called to the closure starting (tokio blocking-pool queueing) | When the closure starts | kind | 0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 4 | +| `lean_beacon_validation_run_seconds` | Histogram | Run time of the closure itself | When the closure ends (panic included) | kind | same as above | +| `lean_beacon_column_check_permit_wait_seconds` | Histogram | Time a fetched sidecar or envelope waited for a `column_check` permit | When the permit is granted | kind=column_check,envelope_check | same as above | +| `lean_beacon_epoch_precompute_running` | Gauge | 1 while an epoch precompute worker runs its advance | Worker start / end | | | + +`kind` on the queue and run histograms is the gossip topic kind above for gossip +validation. `column_check` and `envelope_check` are the tasks of +`check_and_forward` (fetched and replayed columns) and of the envelope checks: +their queue delay is measured from `spawn_blocking`, after the permit wait, which +has its own histogram, so the two add up. The in-use gauge counts a task from its +permit being taken, so one waiting in the blocking pool counts too; it is a count +of admitted work, not of busy threads. Only +`lean_beacon_validation_queue_seconds` isolates time spent waiting for a thread. + +`lean_beacon_epoch_precompute_running` is set when the worker's closure starts +and cleared when it ends, so the interval matches +`lean_beacon_epoch_precompute_seconds` (which gives a duration, not an overlap +with other series). At most one worker runs at a time. + +Block import needs no new series: `lean_block_import_phase_seconds{phase="queue"}` +is already the wait in the chain actor's mailbox before the import starts (see +Block Import Timing). + +**Does validation compete with import?** Plot these on one panel: + +```promql +# Validation work admitted, waiting for a thread, and refused +sum(lean_beacon_validation_permits_in_use) +histogram_quantile(0.99, sum by (le) (rate(lean_beacon_validation_queue_seconds_bucket[5m]))) +sum by (pool) (rate(lean_beacon_validation_permit_exhausted_total[5m])) + +# Against import's STF phase +histogram_quantile(0.99, sum by (le) (rate(lean_block_import_phase_seconds_bucket{phase="stf"}[5m]))) + +# Precompute overlap: fraction of the window a worker ran +avg_over_time(lean_beacon_epoch_precompute_running[5m]) +``` + +Reading it: if STF p99 rises in the windows where queue-delay p99 or in-use +rises, the two share threads (the tokio blocking pool, or cores). If queue delay +stays near zero and STF p99 still moves, suspect cores instead: validation's CPU +share is roughly `sum(rate(lean_beacon_validation_run_seconds_sum[5m]))`, in +cores. To see whether precompute hurts import, compare STF p99 in windows where +`lean_beacon_epoch_precompute_running` is 1 against windows where it is 0. ### Beacon Peer Scoring