Tidy, add metric, debounce errors

sigp · Aug 12, 2020 · a6091dd · a6091dd
1 parent f299d8b
commit a6091dd
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 5 deletions.
diff --git a/beacon_node/network/src/metrics.rs b/beacon_node/network/src/metrics.rs
@@ -56,6 +56,10 @@ lazy_static! {
         "gossip_processor_workers_active_total",
         "Count of active workers in the gossip processing pool."
     );
+    pub static ref GOSSIP_PROCESSOR_EVENTS_TOTAL: Result<IntCounter> = try_create_int_counter(
+        "gossip_processor_events_total",
+        "Count of events processed by the gossip processor manager."
+    );
     pub static ref GOSSIP_PROCESSOR_EVENT_HANDLING_SECONDS: Result<Histogram> = try_create_histogram(
         "gossip_processor_event_handling_seconds",
         "Time spend handling a new message and allocating it to a queue or worker."

diff --git a/beacon_node/network/src/router/gossip_processor.rs b/beacon_node/network/src/router/gossip_processor.rs
@@ -188,8 +188,8 @@ impl<T: BeaconChainTypes> GossipProcessor<T> {
     /// error for an upstream caller to send a `WorkerIdle` message.
     pub fn spawn_manager(mut self) -> mpsc::Sender<Event<T::EthSpec>> {
         let (event_tx, mut event_rx) = mpsc::channel::<Event<T::EthSpec>>(MAX_WORK_QUEUE_LEN);
-        let mut attestation_queue = LifoQueue::new(MAX_UNAGGREGATED_ATTESTATION_QUEUE_LEN);
         let mut aggregate_queue = LifoQueue::new(MAX_AGGREGATED_ATTESTATION_QUEUE_LEN);
+        let mut attestation_queue = LifoQueue::new(MAX_UNAGGREGATED_ATTESTATION_QUEUE_LEN);
         let inner_event_tx = event_tx.clone();
         let executor = self.executor.clone();
 
@@ -198,12 +198,15 @@ impl<T: BeaconChainTypes> GossipProcessor<T> {
                 while let Some(event) = event_rx.recv().await {
                     let _event_timer =
                         metrics::start_timer(&metrics::GOSSIP_PROCESSOR_EVENT_HANDLING_SECONDS);
+                    metrics::inc_counter(&metrics::GOSSIP_PROCESSOR_EVENTS_TOTAL);
 
                     if event == Event::WorkerIdle {
                         self.current_workers = self.current_workers.saturating_sub(1);
                     }
 
                     let can_spawn = self.current_workers < self.max_workers;
+                    let initial_aggregate_queue_len = aggregate_queue.len();
+                    let initial_attestation_queue_len = attestation_queue.len();
 
                     match event {
                         Event::WorkerIdle => {
@@ -263,9 +266,9 @@ impl<T: BeaconChainTypes> GossipProcessor<T> {
                         aggregate_queue.len() as i64,
                     );
 
-                    // TODO: rate limit the logs below.
-
-                    if aggregate_queue.is_full() {
+                    if initial_aggregate_queue_len != aggregate_queue.len()
+                        && aggregate_queue.is_full()
+                    {
                         error!(
                             self.log,
                             "Aggregate attestation queue full";
@@ -274,7 +277,9 @@ impl<T: BeaconChainTypes> GossipProcessor<T> {
                         )
                     }
 
-                    if attestation_queue.is_full() {
+                    if initial_attestation_queue_len != attestation_queue.len()
+                        && attestation_queue.is_full()
+                    {
                         error!(
                             self.log,
                             "Attestation queue full";