avoid using recv in sigverify

core/src/sigverify_stage.rs

@@ -293,6 +293,9 @@ impl SigVerifyStage {
         stats: &mut SigVerifierStats,
     ) -> Result<(), T::SendType> {
         let (mut batches, num_packets, recv_duration) = streamer::recv_packet_batches(recvr)?;
+        if num_packets == 0 {
+            return Ok(());
+        }
 
         let batches_len = batches.len();
         debug!(
@@ -408,9 +411,6 @@ impl SigVerifyStage {
                             SigVerifyServiceError::Streamer(StreamerError::RecvTimeout(
                                 RecvTimeoutError::Disconnected,
                             )) => break,
-                            SigVerifyServiceError::Streamer(StreamerError::RecvTimeout(
-                                RecvTimeoutError::Timeout,
-                            )) => (),
                             SigVerifyServiceError::Send(_) => {
                                 break;
                             }

streamer/src/streamer.rs

@@ -9,7 +9,9 @@ use {
         },
         sendmmsg::{batch_send, SendPktsError},
     },
-    crossbeam_channel::{Receiver, RecvTimeoutError, SendError, Sender, TrySendError},
+    crossbeam_channel::{
+        Receiver, RecvTimeoutError, SendError, Sender, TryRecvError, TrySendError,
+    },
     histogram::Histogram,
     solana_net_utils::{
         multihomed_sockets::{
@@ -494,16 +496,34 @@ fn recv_send(
 pub fn recv_packet_batches(
     recvr: &PacketBatchReceiver,
 ) -> Result<(Vec<PacketBatch>, usize, Duration)> {
+    const MAX_RECV_ATTEMPTS: usize = 1_000;
     let recv_start = Instant::now();
-    let timer = Duration::new(1, 0);
-    let packet_batch = recvr.recv_timeout(timer)?;
-    trace!("got packets");
-    let mut num_packets = packet_batch.len();
-    let mut packet_batches = vec![packet_batch];
-    while let Ok(packet_batch) = recvr.try_recv() {
-        trace!("got more packets");
-        num_packets += packet_batch.len();
-        packet_batches.push(packet_batch);
+
+    let mut num_packets = 0;
+    let mut packet_batches = Vec::new();
+    let mut num_attempts = 0;
+
+    while num_attempts < MAX_RECV_ATTEMPTS {
+        loop {
+            match recvr.try_recv() {
+                Ok(packet_batch) => {
+                    trace!("got more packets");
+                    num_packets += packet_batch.len();
+                    packet_batches.push(packet_batch);
+                }
+                Err(TryRecvError::Empty) => {
+                    break;
+                }
+                Err(TryRecvError::Disconnected) => {
+                    return Err(StreamerError::RecvTimeout(RecvTimeoutError::Disconnected));
+                }
+            }
+        }
+        if num_packets > 0 {
+            break;
+        }
+        sleep(Duration::from_millis(1));
+        num_attempts += 1;
     }
     let recv_duration = recv_start.elapsed();
     trace!(