v3.0: move forwarding stage backed up warning to send-side (backport of #7986) (#8093)

move forwarding stage backed up warning to send-side (#7986)

(cherry picked from commit fff1a49)

Co-authored-by: Andrew Fitzgerald <[email protected]>

Author: mergify[bot]

core/src/forwarding_stage.rs

@@ -236,25 +236,12 @@ impl<VoteClient: ForwardingClient, NonVoteClient: ForwardingClient>
                 self.buffer_packet_batches(packet_batches, tpu_vote_batch, bank);
 
                 // Drain the channel up to timeout
-                let timed_out = loop {
-                    if now.elapsed() >= TIMEOUT {
-                        break true;
-                    }
+                while now.elapsed() < TIMEOUT {
                     match self.receiver.try_recv() {
                         Ok((packet_batches, tpu_vote_batch)) => {
                             self.buffer_packet_batches(packet_batches, tpu_vote_batch, bank)
                         }
-                        Err(_) => break false,
-                    }
-                };
-
-                // If timeout was reached, prevent backup by draining all
-                // packets in the channel.
-                if timed_out {
-                    warn!("ForwardingStage is backed up, dropping packets");
-                    while let Ok((packet_batch, _)) = self.receiver.try_recv() {
-                        self.metrics.dropped_on_timeout +=
-                            packet_batch.iter().map(|b| b.len()).sum::<usize>();
+                        Err(_) => break,
                     }
                 }
 
@@ -722,8 +709,6 @@ struct ForwardingStageMetrics {
     non_votes_dropped_on_data_budget: usize,
     non_votes_forwarded: usize,
     non_votes_dropped_on_send: usize,
-
-    dropped_on_timeout: usize,
 }
 
 impl ForwardingStageMetrics {
@@ -803,7 +788,6 @@ impl Default for ForwardingStageMetrics {
             non_votes_dropped_on_data_budget: 0,
             non_votes_forwarded: 0,
             non_votes_dropped_on_send: 0,
-            dropped_on_timeout: 0,
         }
     }
 }

core/src/sigverify.rs

@@ -13,7 +13,7 @@ use {
         sigverify_stage::{SigVerifier, SigVerifyServiceError},
     },
     agave_banking_stage_ingress_types::BankingPacketBatch,
-    crossbeam_channel::Sender,
+    crossbeam_channel::{Sender, TrySendError},
     solana_perf::{cuda_runtime::PinnedVec, packet::PacketBatch, recycler::Recycler, sigverify},
 };
 
@@ -61,7 +61,11 @@ impl SigVerifier for TransactionSigVerifier {
         if let Some(forward_stage_sender) = &self.forward_stage_sender {
             self.banking_stage_sender
                 .send(banking_packet_batch.clone())?;
-            let _ = forward_stage_sender.try_send((banking_packet_batch, self.reject_non_vote));
+            if let Err(TrySendError::Full(_)) =
+                forward_stage_sender.try_send((banking_packet_batch, self.reject_non_vote))
+            {
+                warn!("forwarding stage channel is full, dropping packets.");
+            }
         } else {
             self.banking_stage_sender.send(banking_packet_batch)?;
         }