feature(collator): remove cumulative stats recalculation

Author: drmick

collator/src/collator/messages_reader/internals_reader.rs

@@ -356,7 +356,6 @@ impl<V: InternalMessageValue> InternalsPartitionReader<V> {
 
             ranges.push(QueueShardBoundedRange {
                 shard_ident: *shard_id,
-                // TODO: may be we should pass `from` here because we should load stats for the full range
                 from: Bound::Excluded(shard_reader_state.current_position),
                 to: Bound::Included(shard_range_to),
             });

collator/src/collator/types.rs

@@ -24,7 +24,6 @@ use tycho_block_util::state::{RefMcStateHandle, ShardStateStuff};
 use tycho_core::global_config::MempoolGlobalConfig;
 use tycho_executor::{AccountMeta, PublicLibraryChange, TransactionMeta};
 use tycho_network::PeerId;
-use tycho_util::metrics::HistogramGuard;
 use tycho_util::{DashMapEntry, FastDashMap, FastHashMap, FastHashSet};
 
 use super::do_collate::work_units::PrepareMsgGroupsWu;
@@ -1290,9 +1289,6 @@ pub struct CumulativeStatistics {
 
     /// The final aggregated statistics (across all shards) by partitions.
     result: FastHashMap<QueuePartitionIdx, QueueStatisticsWithRemaning>,
-
-    /// A flag indicating that data has changed, and we need to recalculate before returning `result`.
-    dirty: bool,
 }
 
 impl CumulativeStatistics {
@@ -1306,7 +1302,6 @@ impl CumulativeStatistics {
             all_shards_processed_to_by_partitions,
             shards_stats_by_partitions: Default::default(),
             result: Default::default(),
-            dirty: false,
         }
     }
 
@@ -1345,7 +1340,6 @@ impl CumulativeStatistics {
         partitions: &FastHashSet<QueuePartitionIdx>,
         ranges: Vec<QueueShardBoundedRange>,
     ) -> Result<()> {
-        self.dirty = true;
         tracing::trace!(
             target: tracing_targets::COLLATOR,
             "cumulative_stats_partial_ranges: {:?}",
@@ -1405,9 +1399,6 @@ impl CumulativeStatistics {
                 by_partitions.retain(|_part, diffs| !diffs.is_empty());
                 !by_partitions.is_empty()
             });
-
-        self.all_shards_processed_to_by_partitions = new_pt;
-        self.dirty = true;
     }
 
     fn compute_cumulative_stats_ranges(
@@ -1469,15 +1460,19 @@ impl CumulativeStatistics {
             }
         }
 
+        if !diff_partition_stats.is_empty() {
+            let part_stats = self.result.entry(partition).or_default();
+            part_stats.initial_stats.append(&diff_partition_stats);
+            part_stats.remaning_stats.append(&diff_partition_stats);
+        }
+
         // finally add weeded stats
         self.add_diff_partition_stats(
             partition,
             diff_shard,
             diff_max_message,
             diff_partition_stats,
         );
-
-        self.dirty = true;
     }
 
     fn add_diff_partition_stats(
@@ -1541,6 +1536,9 @@ impl CumulativeStatistics {
                                         cumulative_stats
                                             .initial_stats
                                             .decrement_for_account(dst_acc.clone(), *count);
+                                        cumulative_stats
+                                            .remaning_stats
+                                            .decrement_for_account(dst_acc.clone(), *count);
                                         false
                                     } else {
                                         true
@@ -1571,15 +1569,12 @@ impl CumulativeStatistics {
     /// Returns  a reference to the aggregated stats by partitions.
     /// If the data is marked as dirty, it triggers a lazy recalculation first.
     pub fn result(&mut self) -> &FastHashMap<QueuePartitionIdx, QueueStatisticsWithRemaning> {
-        self.ensure_finalized();
         &self.result
     }
 
     /// Calc aggregated stats among all partitions.
     /// If the data is marked as dirty, it triggers a lazy recalculation first.
     pub fn get_aggregated_result(&mut self) -> QueueStatistics {
-        self.ensure_finalized();
-
         let mut res: Option<QueueStatistics> = None;
         for stats in self.result.values() {
             if let Some(aggregated) = res.as_mut() {
@@ -1590,42 +1585,6 @@ impl CumulativeStatistics {
         }
         res.unwrap_or_default()
     }
-
-    /// A helper function to trigger a recalculation if `dirty` is set.
-    fn ensure_finalized(&mut self) {
-        if self.dirty {
-            self.recalculate();
-        }
-    }
-
-    /// Clears the existing result and aggregates all data from `shards_statistics`.
-    fn recalculate(&mut self) {
-        let _histogram = HistogramGuard::begin("tycho_do_collate_recalculate_statistics_time");
-        self.result.clear();
-
-        for shard_stats_by_partitions in self.shards_stats_by_partitions.values() {
-            for (&partition, diffs) in shard_stats_by_partitions {
-                let mut partition_stats = AccountStatistics::new();
-                for diff_stats in diffs.values() {
-                    for (account, &count) in diff_stats {
-                        *partition_stats.entry(account.clone()).or_default() += count;
-                    }
-                }
-                self.result
-                    .entry(partition)
-                    .and_modify(|stats| {
-                        stats.initial_stats.append(&partition_stats);
-                        stats.remaning_stats.append(&partition_stats);
-                    })
-                    .or_insert(QueueStatisticsWithRemaning {
-                        initial_stats: QueueStatistics::with_statistics(partition_stats.clone()),
-                        remaning_stats: ConcurrentQueueStatistics::with_statistics(partition_stats),
-                    });
-            }
-        }
-
-        self.dirty = false;
-    }
 }
 
 #[derive(Debug, Default, Clone)]
@@ -1634,12 +1593,6 @@ pub struct ConcurrentQueueStatistics {
 }
 
 impl ConcurrentQueueStatistics {
-    pub fn with_statistics(statistics: AccountStatistics) -> Self {
-        Self {
-            statistics: Arc::new(statistics.into_iter().collect()),
-        }
-    }
-
     pub fn statistics(&self) -> &FastDashMap<IntAddr, u64> {
         &self.statistics
     }
@@ -1694,7 +1647,6 @@ impl fmt::Debug for CumulativeStatistics {
             )
             .field("shards_stats_by_partitions", &shards_summary)
             .field("result", &format!("{} partitions", self.result.len()))
-            .field("dirty", &self.dirty)
             .finish()
     }
 }

scripts/gen-dashboard.py

@@ -1710,10 +1710,6 @@ def collator_core_operations_metrics() -> RowPanel:
             "Handle block candidate",
             labels=['workchain=~"$workchain"'],
         ),
-        create_heatmap_panel(
-            "tycho_do_collate_recalculate_statistics_time",
-            "Recalculate cumulative statistics"
-        ),
     ]
     return create_row("collator: Core Operations Metrics", metrics)