turbine: Fix and cleanup redundant metric (anza-xyz#9287)

The broadcast-process-shred-stats (and interrupted variant) have
duplicate fields for the number of data and coding shred. This is
probably a relic of when we rolled merkle shreds out. So, remove the
duplicate fields.

This also addresses a small issue where some shreds would be counted
in metrics for the wrong slot when we have an interrupted slot

Author: steviez

ledger/src/shred/stats.rs

@@ -74,8 +74,6 @@ impl ProcessShredsStats {
         &mut self,
         name: &'static str,
         slot: Slot,
-        num_data_shreds: u32,
-        num_coding_shreds: u32,
         slot_broadcast_time: Option<Duration>,
     ) {
         let slot_broadcast_time = slot_broadcast_time
@@ -90,14 +88,8 @@ impl ProcessShredsStats {
             ("slot", slot, i64),
             ("shredding_time", self.shredding_elapsed, i64),
             ("receive_time", self.receive_elapsed, i64),
-            ("num_data_shreds", num_data_shreds, i64),
-            ("num_coding_shreds", num_coding_shreds, i64),
-            ("num_merkle_data_shreds", self.num_merkle_data_shreds, i64),
-            (
-                "num_merkle_coding_shreds",
-                self.num_merkle_coding_shreds,
-                i64
-            ),
+            ("num_data_shreds", self.num_merkle_data_shreds, i64),
+            ("num_coding_shreds", self.num_merkle_coding_shreds, i64),
             ("slot_broadcast_time", slot_broadcast_time, i64),
             (
                 "get_leader_schedule_time",

turbine/src/broadcast_stage/standard_broadcast_run.rs

@@ -74,12 +74,7 @@ impl StandardBroadcastRun {
     // If the current slot has changed, generates an empty shred indicating
     // last shred in the previous slot, along with coding shreds for the data
     // shreds buffered.
-    fn finish_prev_slot(
-        &mut self,
-        keypair: &Keypair,
-        max_ticks_in_slot: u8,
-        stats: &mut ProcessShredsStats,
-    ) -> Vec<Shred> {
+    fn finish_prev_slot(&mut self, keypair: &Keypair, max_ticks_in_slot: u8) -> Vec<Shred> {
         if self.completed {
             return vec![];
         }
@@ -96,9 +91,11 @@ impl StandardBroadcastRun {
                     self.next_shred_index,
                     self.next_code_index,
                     &self.reed_solomon_cache,
-                    stats,
+                    &mut self.process_shreds_stats,
                 )
-                .inspect(|shred| stats.record_shred(shred))
+                // These shreds will finish the slot so no need to update
+                // self.next_shred_index and self.next_code_index
+                .inspect(|shred| self.process_shreds_stats.record_shred(shred))
                 .collect();
         if let Some(shred) = shreds.iter().max_by_key(|shred| shred.fec_set_index()) {
             self.chained_merkle_root = shred.merkle_root().unwrap();
@@ -208,8 +205,7 @@ impl StandardBroadcastRun {
         if self.slot != bank.slot() {
             // Finish previous slot if it was interrupted.
             if !self.completed {
-                let shreds =
-                    self.finish_prev_slot(keypair, bank.ticks_per_slot() as u8, process_stats);
+                let shreds = self.finish_prev_slot(keypair, bank.ticks_per_slot() as u8);
                 debug_assert!(shreds.iter().all(|shred| shred.slot() == self.slot));
                 // Broadcast shreds for the interrupted slot.
                 let batch_info = Some(BroadcastShredBatchInfo {
@@ -435,13 +431,8 @@ impl StandardBroadcastRun {
             )
         };
 
-        self.process_shreds_stats.submit(
-            name,
-            self.slot,
-            self.next_shred_index, // num_data_shreds
-            self.next_code_index,  // num_coding_shreds
-            slot_broadcast_time,
-        );
+        self.process_shreds_stats
+            .submit(name, self.slot, slot_broadcast_time);
     }
 }
 
@@ -580,11 +571,8 @@ mod test {
         run.slot_broadcast_start = Instant::now();
 
         // Slot 2 interrupted slot 1
-        let shreds = run.finish_prev_slot(
-            &keypair,
-            0, // max_ticks_in_slot
-            &mut ProcessShredsStats::default(),
-        );
+        let max_ticks_in_slot = 0;
+        let shreds = run.finish_prev_slot(&keypair, max_ticks_in_slot);
         assert!(run.completed);
         let shred = shreds
             .first()