diff --git a/crates/engine/tree/src/tree/payload_processor/multiproof.rs b/crates/engine/tree/src/tree/payload_processor/multiproof.rs index a000e7a5adf..ca3bd380d4d 100644 --- a/crates/engine/tree/src/tree/payload_processor/multiproof.rs +++ b/crates/engine/tree/src/tree/payload_processor/multiproof.rs @@ -9,7 +9,7 @@ use alloy_primitives::{ use crossbeam_channel::{unbounded, Receiver as CrossbeamReceiver, Sender as CrossbeamSender}; use dashmap::DashMap; use derive_more::derive::Deref; -use metrics::Histogram; +use metrics::{Gauge, Histogram}; use reth_metrics::Metrics; use reth_revm::state::EvmState; use reth_trie::{ @@ -319,8 +319,6 @@ impl MultiproofInput { /// `ProofSequencer`. #[derive(Debug)] pub struct MultiproofManager { - /// Currently running calculations. - inflight: usize, /// Handle to the proof worker pools (storage and account). proof_worker_handle: ProofWorkerHandle, /// Cached storage proof roots for missed leaves; this maps @@ -349,8 +347,11 @@ impl MultiproofManager { proof_worker_handle: ProofWorkerHandle, proof_result_tx: CrossbeamSender, ) -> Self { + // Initialize the max worker gauges with the worker pool sizes + metrics.max_storage_workers.set(proof_worker_handle.total_storage_workers() as f64); + metrics.max_account_workers.set(proof_worker_handle.total_account_workers() as f64); + Self { - inflight: 0, metrics, proof_worker_handle, missed_leaves_storage_roots: Default::default(), @@ -359,7 +360,7 @@ impl MultiproofManager { } /// Dispatches a new multiproof calculation to worker pools. - fn dispatch(&mut self, input: PendingMultiproofTask) { + fn dispatch(&self, input: PendingMultiproofTask) { // If there are no proof targets, we can just send an empty multiproof back immediately if input.proof_targets_is_empty() { debug!( @@ -381,7 +382,7 @@ impl MultiproofManager { } /// Dispatches a single storage proof calculation to worker pool. - fn dispatch_storage_proof(&mut self, storage_multiproof_input: StorageMultiproofInput) { + fn dispatch_storage_proof(&self, storage_multiproof_input: StorageMultiproofInput) { let StorageMultiproofInput { hashed_state_update, hashed_address, @@ -432,8 +433,12 @@ impl MultiproofManager { return; } - self.inflight += 1; - self.metrics.inflight_multiproofs_histogram.record(self.inflight as f64); + self.metrics + .active_storage_workers_histogram + .record(self.proof_worker_handle.active_storage_workers() as f64); + self.metrics + .active_account_workers_histogram + .record(self.proof_worker_handle.active_account_workers() as f64); self.metrics .pending_storage_multiproofs_histogram .record(self.proof_worker_handle.pending_storage_tasks() as f64); @@ -443,9 +448,13 @@ impl MultiproofManager { } /// Signals that a multiproof calculation has finished. - fn on_calculation_complete(&mut self) { - self.inflight = self.inflight.saturating_sub(1); - self.metrics.inflight_multiproofs_histogram.record(self.inflight as f64); + fn on_calculation_complete(&self) { + self.metrics + .active_storage_workers_histogram + .record(self.proof_worker_handle.active_storage_workers() as f64); + self.metrics + .active_account_workers_histogram + .record(self.proof_worker_handle.active_account_workers() as f64); self.metrics .pending_storage_multiproofs_histogram .record(self.proof_worker_handle.pending_storage_tasks() as f64); @@ -455,7 +464,7 @@ impl MultiproofManager { } /// Dispatches a single multiproof calculation to worker pool. - fn dispatch_multiproof(&mut self, multiproof_input: MultiproofInput) { + fn dispatch_multiproof(&self, multiproof_input: MultiproofInput) { let MultiproofInput { source, hashed_state_update, @@ -506,8 +515,12 @@ impl MultiproofManager { return; } - self.inflight += 1; - self.metrics.inflight_multiproofs_histogram.record(self.inflight as f64); + self.metrics + .active_storage_workers_histogram + .record(self.proof_worker_handle.active_storage_workers() as f64); + self.metrics + .active_account_workers_histogram + .record(self.proof_worker_handle.active_account_workers() as f64); self.metrics .pending_storage_multiproofs_histogram .record(self.proof_worker_handle.pending_storage_tasks() as f64); @@ -520,8 +533,14 @@ impl MultiproofManager { #[derive(Metrics, Clone)] #[metrics(scope = "tree.root")] pub(crate) struct MultiProofTaskMetrics { - /// Histogram of inflight multiproofs. - pub inflight_multiproofs_histogram: Histogram, + /// Histogram of active storage workers processing proofs. + pub active_storage_workers_histogram: Histogram, + /// Histogram of active account workers processing proofs. + pub active_account_workers_histogram: Histogram, + /// Gauge for the maximum number of storage workers in the pool. + pub max_storage_workers: Gauge, + /// Gauge for the maximum number of account workers in the pool. + pub max_account_workers: Gauge, /// Histogram of pending storage multiproofs in the queue. pub pending_storage_multiproofs_histogram: Histogram, /// Histogram of pending account multiproofs in the queue. @@ -583,7 +602,6 @@ pub(crate) struct MultiProofTaskMetrics { /// ▼ │ /// ┌──────────────────────────────────────────────────────────────┐ │ /// │ MultiproofManager │ │ -/// │ - Tracks inflight calculations │ │ /// │ - Deduplicates against fetched_proof_targets │ │ /// │ - Routes to appropriate worker pool │ │ /// └──┬───────────────────────────────────────────────────────────┘ │ @@ -624,7 +642,6 @@ pub(crate) struct MultiProofTaskMetrics { /// /// - **[`MultiproofManager`]**: Calculation orchestrator /// - Decides between fast path ([`EmptyProof`]) and worker dispatch -/// - Tracks inflight calculations /// - Routes storage-only vs full multiproofs to appropriate workers /// - Records metrics for monitoring /// diff --git a/crates/trie/parallel/src/proof_task.rs b/crates/trie/parallel/src/proof_task.rs index 7e453cbc7c3..caca8687534 100644 --- a/crates/trie/parallel/src/proof_task.rs +++ b/crates/trie/parallel/src/proof_task.rs @@ -1017,6 +1017,10 @@ pub struct ProofWorkerHandle { /// Counter tracking available account workers. Workers decrement when starting work, /// increment when finishing. Used to determine whether to chunk multiproofs. account_available_workers: Arc, + /// Total number of storage workers spawned + storage_worker_count: usize, + /// Total number of account workers spawned + account_worker_count: usize, } impl ProofWorkerHandle { @@ -1118,6 +1122,8 @@ impl ProofWorkerHandle { account_work_tx, storage_available_workers, account_available_workers, + storage_worker_count, + account_worker_count, } } @@ -1141,6 +1147,32 @@ impl ProofWorkerHandle { self.account_work_tx.len() } + /// Returns the total number of storage workers in the pool. + pub const fn total_storage_workers(&self) -> usize { + self.storage_worker_count + } + + /// Returns the total number of account workers in the pool. + pub const fn total_account_workers(&self) -> usize { + self.account_worker_count + } + + /// Returns the number of storage workers currently processing tasks. + /// + /// This is calculated as total workers minus available workers. + pub fn active_storage_workers(&self) -> usize { + self.storage_worker_count + .saturating_sub(self.storage_available_workers.load(Ordering::Relaxed)) + } + + /// Returns the number of account workers currently processing tasks. + /// + /// This is calculated as total workers minus available workers. + pub fn active_account_workers(&self) -> usize { + self.account_worker_count + .saturating_sub(self.account_available_workers.load(Ordering::Relaxed)) + } + /// Dispatch a storage proof computation to storage worker pool /// /// The result will be sent via the `proof_result_sender` channel. diff --git a/etc/grafana/dashboards/overview.json b/etc/grafana/dashboards/overview.json index 46a465ca4a4..aba888dba07 100644 --- a/etc/grafana/dashboards/overview.json +++ b/etc/grafana/dashboards/overview.json @@ -4308,7 +4308,38 @@ }, "unit": "none" }, - "overrides": [] + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "Max storage workers" + }, + "properties": [ + { + "id": "custom.lineStyle", + "value": { + "dash": [10, 10], + "fill": "dash" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "Max account workers" + }, + "properties": [ + { + "id": "custom.lineStyle", + "value": { + "dash": [10, 10], + "fill": "dash" + } + } + ] + } + ] }, "gridPos": { "h": 8, @@ -4316,6 +4347,7 @@ "x": 12, "y": 104 }, + "description": "The max metrics (Max storage workers and Max account workers) are displayed as dotted lines to highlight the configured upper limits.", "id": 256, "options": { "legend": { @@ -4338,14 +4370,50 @@ "uid": "${DS_PROMETHEUS}" }, "editorMode": "code", - "expr": "reth_tree_root_inflight_multiproofs_histogram{$instance_label=\"$instance\",quantile=~\"(0|0.5|0.9|0.95|1)\"}", + "expr": "reth_tree_root_active_storage_workers_histogram{$instance_label=\"$instance\",quantile=~\"(0|0.5|0.9|0.95|1)\"}", "instant": false, - "legendFormat": "{{quantile}} percentile", + "legendFormat": "Storage workers {{quantile}} percentile", "range": true, - "refId": "Branch Nodes" + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "reth_tree_root_active_account_workers_histogram{$instance_label=\"$instance\",quantile=~\"(0|0.5|0.9|0.95|1)\"}", + "instant": false, + "legendFormat": "Account workers {{quantile}} percentile", + "range": true, + "refId": "B" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "reth_tree_root_max_storage_workers{$instance_label=\"$instance\"}", + "instant": false, + "legendFormat": "Max storage workers", + "range": true, + "refId": "C" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "reth_tree_root_max_account_workers{$instance_label=\"$instance\"}", + "instant": false, + "legendFormat": "Max account workers", + "range": true, + "refId": "D" } ], - "title": "In-flight MultiProof requests", + "title": "Active MultiProof Workers", "type": "timeseries" }, {