Add metrics for shard broadcast + decoding

Author: clabby

consensus/src/marshal/ingress/coding/application.rs

@@ -10,9 +10,10 @@ use crate::{
 use commonware_coding::{Config as CodingConfig, Scheme};
 use commonware_cryptography::{bls12381::primitives::variant::Variant, Committable, PublicKey};
 use commonware_runtime::{Clock, Metrics, Spawner};
-use futures::channel::oneshot;
+use futures::{channel::oneshot, lock::Mutex};
+use prometheus_client::metrics::gauge::Gauge;
 use rand::Rng;
-use std::sync::{Arc, Mutex};
+use std::{sync::Arc, time::Instant};
 use tracing::{debug, info, warn};
 
 /// An [Application] adapter that handles erasure coding and shard verification for consensus.
@@ -34,6 +35,10 @@ where
     identity: P,
     supervisor: Z,
     last_built: Arc<Mutex<Option<(View, CodedBlock<B, S>)>>>,
+
+    parent_fetch_duration: Gauge,
+    build_duration: Gauge,
+    erasure_code_duration: Gauge,
 }
 
 impl<E, A, V, B, S, P, Z> CodingAdapter<E, A, V, B, S, P, Z>
@@ -53,13 +58,38 @@ where
         identity: P,
         supervisor: Z,
     ) -> Self {
+        let parent_fetch_duration = Gauge::default();
+        context.register(
+            "parent_fetch_duration",
+            "Time taken to fetch a parent block from marshal to build on top of, in milliseconds",
+            parent_fetch_duration.clone(),
+        );
+
+        let build_duration = Gauge::default();
+        context.register(
+            "build_duration",
+            "Time taken for the application to build a new block, in milliseconds",
+            build_duration.clone(),
+        );
+
+        let erasure_code_duration = Gauge::default();
+        context.register(
+            "erasure_code_duration",
+            "Time taken to erasure code a built block, in milliseconds",
+            erasure_code_duration.clone(),
+        );
+
         Self {
             context,
             application,
             marshal,
             identity,
             supervisor,
             last_built: Arc::new(Mutex::new(None)),
+
+            parent_fetch_duration,
+            build_duration,
+            erasure_code_duration,
         }
     }
 }
@@ -100,10 +130,16 @@ where
         let n_participants = participants.len() as u16;
         let coding_config = coding_config_for_participants(n_participants);
 
+        // Metrics
+        let parent_fetch_duration = self.parent_fetch_duration.clone();
+        let build_duration = self.build_duration.clone();
+        let erasure_code_duration = self.erasure_code_duration.clone();
+
         let (tx, rx) = oneshot::channel();
         self.context
             .with_label("propose")
             .spawn(move |r_ctx| async move {
+                let start = Instant::now();
                 let parent_block = if parent_commitment == genesis.commitment() {
                     genesis
                 } else {
@@ -122,16 +158,25 @@ where
                         return;
                     }
                 };
+                parent_fetch_duration.set(start.elapsed().as_millis() as i64);
 
+                let start = Instant::now();
                 let built_block = application
                     .build(r_ctx.with_label("build"), parent_commitment, parent_block)
                     .await;
+                build_duration.set(start.elapsed().as_millis() as i64);
+
+                let start = Instant::now();
                 let coded_block = CodedBlock::new(built_block, coding_config);
+                erasure_code_duration.set(start.elapsed().as_millis() as i64);
+
                 let commitment = coded_block.commitment();
 
                 // Update the latest built block.
-                let mut lock = last_built.lock().expect("failed to lock last_built mutex");
-                *lock = Some((context.view(), coded_block));
+                {
+                    let mut lock = last_built.lock().await;
+                    *lock = Some((context.view(), coded_block));
+                }
 
                 let result = tx.send(commitment);
                 info!(
@@ -197,7 +242,7 @@ where
     type Digest = B::Commitment;
 
     async fn broadcast(&mut self, _commitment: Self::Digest) {
-        let Some((round, block)) = self.last_built.lock().unwrap().clone() else {
+        let Some((round, block)) = self.last_built.lock().await.clone() else {
             warn!("missing block to broadcast");
             return;
         };

consensus/src/marshal/ingress/coding/mailbox.rs

@@ -10,10 +10,13 @@ use commonware_codec::{Decode, Error as CodecError};
 use commonware_coding::Scheme;
 use commonware_cryptography::{Hasher, PublicKey};
 use commonware_p2p::Recipients;
+use commonware_runtime::Metrics;
 use futures::channel::oneshot;
+use prometheus_client::metrics::gauge::Gauge;
 use std::{
     collections::{btree_map::Entry, BTreeMap},
     ops::Deref,
+    time::Instant,
 };
 use thiserror::Error;
 use tracing::debug;
@@ -61,6 +64,10 @@ where
     ///
     /// These blocks are evicted by marshal after they are delivered to the application.
     reconstructed_blocks: BTreeMap<CodingCommitment, CodedBlock<B, S>>,
+
+    block_broadcast_duration: Gauge,
+    shard_broadcast_duration: Gauge,
+    erasure_decode_duration: Gauge,
 }
 
 impl<S, H, B, P> ShardMailbox<S, H, B, P>
@@ -70,12 +77,40 @@ where
     B: Block<Commitment = CodingCommitment>,
     P: PublicKey,
 {
-    pub fn new(mailbox: buffered::Mailbox<P, Shard<S, H>>, block_codec_cfg: B::Cfg) -> Self {
+    pub fn new(
+        context: impl Metrics,
+        mailbox: buffered::Mailbox<P, Shard<S, H>>,
+        block_codec_cfg: B::Cfg,
+    ) -> Self {
+        let block_broadcast_duration = Gauge::default();
+        context.register(
+            "block_broadcast_duration",
+            "Duration of proposer broadcast in milliseconds",
+            block_broadcast_duration.clone(),
+        );
+
+        let shard_broadcast_duration = Gauge::default();
+        context.register(
+            "shard_broadcast_duration",
+            "Duration of individual shard broadcast in milliseconds",
+            shard_broadcast_duration.clone(),
+        );
+
+        let erasure_decode_duration = Gauge::default();
+        context.register(
+            "erasure_decode_duration",
+            "Duration of erasure decoding in milliseconds",
+            erasure_decode_duration.clone(),
+        );
+
         Self {
             mailbox,
             block_codec_cfg,
             block_subscriptions: BTreeMap::new(),
             reconstructed_blocks: BTreeMap::new(),
+            block_broadcast_duration,
+            shard_broadcast_duration,
+            erasure_decode_duration,
         }
     }
 
@@ -91,12 +126,15 @@ where
             "number of participants must equal number of shards"
         );
 
+        let start = Instant::now();
         for (index, peer) in participants.into_iter().enumerate() {
             let message = block
                 .shard(index)
                 .expect("peer index impossibly out of bounds");
             let _peers = self.mailbox.broadcast(Recipients::One(peer), message).await;
         }
+        self.block_broadcast_duration
+            .set(start.elapsed().as_millis() as i64);
     }
 
     /// Broadcasts a local [Shard] of a block to all peers, if the [Shard] is present
@@ -133,7 +171,11 @@ where
 
             // Broadcast the weak shard to all peers for reconstruction.
             let reshard = Shard::new(commitment, index, DistributionShard::Weak(reshard));
+
+            let start = Instant::now();
             let _peers = self.mailbox.broadcast(Recipients::All, reshard).await;
+            self.shard_broadcast_duration
+                .set(start.elapsed().as_millis() as i64);
 
             debug!(%commitment, index, "broadcasted local shard to all peers");
         } else {
@@ -218,13 +260,16 @@ where
         }
 
         // Attempt to reconstruct the encoded blob
+        let start = Instant::now();
         let decoded = S::decode(
             &config,
             &commitment.inner(),
             checking_data.clone(),
             checked_shards.as_slice(),
         )
         .map_err(ReconstructionError::CodingRecovery)?;
+        self.erasure_decode_duration
+            .set(start.elapsed().as_millis() as i64);
 
         // Attempt to decode the block from the encoded blob
         let block = CodedBlock::<B, S>::decode_cfg(decoded.as_slice(), &self.block_codec_cfg)?;
@@ -418,7 +463,8 @@ mod test {
                 PublicKey,
                 Shard<ReedSolomon<Sha256>, Sha256>,
             >::new(context.clone(), config);
-            let shard_mailbox = SMailbox::new(engine_mailbox, ());
+            let shard_mailbox =
+                SMailbox::new(context.with_label("shard_mailbox"), engine_mailbox, ());
             mailboxes.insert(peer.clone(), shard_mailbox);
 
             engine.start(network);

consensus/src/marshal/mod.rs

@@ -206,7 +206,8 @@ mod tests {
         let network = oracle.register(secret.public_key(), 2).await.unwrap();
         broadcast_engine.start(network);
 
-        let shard_mailbox = ShardMailbox::<_, H, _, _>::new(buffer, ());
+        let shard_mailbox =
+            ShardMailbox::<_, H, _, _>::new(context.with_label("shard_mailbox"), buffer, ());
 
         let (actor, mailbox) = actor::Actor::init(context.clone(), config).await;
         let application = Application::<B>::default();