[consensus] verify proposals in parallel, add counters (#12209)

### Description

For blocks with many batches, using par_iter significantly improves proposal processing time.

The counters were used to find this issue in prep for previewnet.

### Test Plan

Observe counters changed in forge. Existing forge tests.

consensus/consensus-types/src/common.rs

@@ -285,9 +285,11 @@ impl Payload {
         match (quorum_store_enabled, self) {
             (false, Payload::DirectMempool(_)) => Ok(()),
             (true, Payload::InQuorumStore(proof_with_status)) => {
-                for proof in proof_with_status.proofs.iter() {
-                    proof.verify(validator)?;
-                }
+                proof_with_status
+                    .proofs
+                    .par_iter()
+                    .with_min_len(4)
+                    .try_for_each(|proof| proof.verify(validator))?;
                 Ok(())
             },
             (true, Payload::InQuorumStoreWithLimit(proof_with_status)) => {

consensus/src/counters.rs

@@ -636,6 +636,21 @@ pub static WAIT_DURATION_S: Lazy<DurationHistogram> = Lazy::new(|| {
     CONSENSUS_WAIT_DURATION_BUCKETS.to_vec()).unwrap())
 });
 
+const VERIFY_BUCKETS: &[f64] = &[
+    0.0001, 0.00025, 0.0005, 0.001, 0.0015, 0.002, 0.0025, 0.003, 0.0035, 0.004, 0.005, 0.006,
+    0.007, 0.008, 0.009, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0,
+];
+
+pub static VERIFY_MSG: Lazy<HistogramVec> = Lazy::new(|| {
+    register_histogram_vec!(
+        "aptos_consensus_verify_msg",
+        "Histogram of the time it takes to verify a message",
+        &["msg"],
+        VERIFY_BUCKETS.to_vec()
+    )
+    .unwrap()
+});
+
 ///////////////////
 // CHANNEL COUNTERS
 ///////////////////

consensus/src/round_manager.rs

@@ -85,17 +85,24 @@ impl UnverifiedEvent {
         max_num_batches: usize,
         max_batch_expiry_gap_usecs: u64,
     ) -> Result<VerifiedEvent, VerifyError> {
+        let start_time = Instant::now();
         Ok(match self {
             //TODO: no need to sign and verify the proposal
             UnverifiedEvent::ProposalMsg(p) => {
                 if !self_message {
                     p.verify(validator, quorum_store_enabled)?;
+                    counters::VERIFY_MSG
+                        .with_label_values(&["proposal"])
+                        .observe(start_time.elapsed().as_secs_f64());
                 }
                 VerifiedEvent::ProposalMsg(p)
             },
             UnverifiedEvent::VoteMsg(v) => {
                 if !self_message {
                     v.verify(validator)?;
+                    counters::VERIFY_MSG
+                        .with_label_values(&["vote"])
+                        .observe(start_time.elapsed().as_secs_f64());
                 }
                 VerifiedEvent::VoteMsg(v)
             },
@@ -104,6 +111,9 @@ impl UnverifiedEvent {
             UnverifiedEvent::BatchMsg(b) => {
                 if !self_message {
                     b.verify(peer_id, max_num_batches)?;
+                    counters::VERIFY_MSG
+                        .with_label_values(&["batch"])
+                        .observe(start_time.elapsed().as_secs_f64());
                 }
                 VerifiedEvent::BatchMsg(b)
             },
@@ -115,12 +125,18 @@ impl UnverifiedEvent {
                         max_batch_expiry_gap_usecs,
                         validator,
                     )?;
+                    counters::VERIFY_MSG
+                        .with_label_values(&["signed_batch"])
+                        .observe(start_time.elapsed().as_secs_f64());
                 }
                 VerifiedEvent::SignedBatchInfo(sd)
             },
             UnverifiedEvent::ProofOfStoreMsg(p) => {
                 if !self_message {
                     p.verify(max_num_batches, validator)?;
+                    counters::VERIFY_MSG
+                        .with_label_values(&["proof_of_store"])
+                        .observe(start_time.elapsed().as_secs_f64());
                 }
                 VerifiedEvent::ProofOfStoreMsg(p)
             },