Remove exit variable from respond [stage]

And drop the sender that feeds input to the responder.

src/ncp.rs

@@ -37,12 +37,8 @@ impl Ncp {
             request_sender,
         )?;
         let (response_sender, response_receiver) = channel();
-        let t_responder = streamer::responder(
-            gossip_send_socket,
-            exit.clone(),
-            blob_recycler.clone(),
-            response_receiver,
-        );
+        let t_responder =
+            streamer::responder(gossip_send_socket, blob_recycler.clone(), response_receiver);
         let t_listen = Crdt::listen(
             crdt.clone(),
             window,

src/rpu.rs

@@ -65,12 +65,7 @@ impl Rpu {
             blob_recycler.clone(),
         );
 
-        let t_responder = streamer::responder(
-            respond_socket,
-            exit.clone(),
-            blob_recycler.clone(),
-            blob_receiver,
-        );
+        let t_responder = streamer::responder(respond_socket, blob_recycler.clone(), blob_receiver);
 
         let mut thread_hdls = vec![t_receiver, t_responder];
         thread_hdls.extend(request_stage.thread_hdls().into_iter());

src/streamer.rs

@@ -101,17 +101,16 @@ pub fn recv_batch(recvr: &PacketReceiver) -> Result<(Vec<SharedPackets>, usize)>
     Ok((batch, len))
 }
 
-pub fn responder(
-    sock: UdpSocket,
-    exit: Arc<AtomicBool>,
-    recycler: BlobRecycler,
-    r: BlobReceiver,
-) -> JoinHandle<()> {
+pub fn responder(sock: UdpSocket, recycler: BlobRecycler, r: BlobReceiver) -> JoinHandle<()> {
     Builder::new()
         .name("solana-responder".to_string())
         .spawn(move || loop {
-            if recv_send(&sock, &recycler, &r).is_err() && exit.load(Ordering::Relaxed) {
-                break;
+            if let Err(e) = recv_send(&sock, &recycler, &r) {
+                match e {
+                    Error::RecvTimeoutError(RecvTimeoutError::Disconnected) => break,
+                    Error::RecvTimeoutError(RecvTimeoutError::Timeout) => (),
+                    _ => error!("{:?}", e),
+                }
             }
         })
         .unwrap()
@@ -844,8 +843,9 @@ mod test {
         let resp_recycler = BlobRecycler::default();
         let (s_reader, r_reader) = channel();
         let t_receiver = receiver(read, exit.clone(), pack_recycler.clone(), s_reader);
+        let t_responder = {
             let (s_responder, r_responder) = channel();
-        let t_responder = responder(send, exit.clone(), resp_recycler.clone(), r_responder);
+            let t_responder = responder(send, resp_recycler.clone(), r_responder);
             let mut msgs = VecDeque::new();
             for i in 0..10 {
                 let b = resp_recycler.allocate();
@@ -858,6 +858,9 @@ mod test {
                 msgs.push_back(b);
             }
             s_responder.send(msgs).expect("send");
+            t_responder
+        };
+
         let mut num = 0;
         get_msgs(r_reader, &mut num);
         assert_eq!(num, 10);
@@ -914,13 +917,10 @@ mod test {
             s_window,
             s_retransmit,
         );
+
+        let t_responder = {
             let (s_responder, r_responder) = channel();
-        let t_responder = responder(
-            tn.sockets.replicate,
-            exit.clone(),
-            resp_recycler.clone(),
-            r_responder,
-        );
+            let t_responder = responder(tn.sockets.replicate, resp_recycler.clone(), r_responder);
             let mut msgs = VecDeque::new();
             for v in 0..10 {
                 let i = 9 - v;
@@ -936,6 +936,9 @@ mod test {
                 msgs.push_back(b);
             }
             s_responder.send(msgs).expect("send");
+            t_responder
+        };
+
         let mut num = 0;
         get_blobs(r_window, &mut num);
         assert_eq!(num, 10);

src/tvu.rs

@@ -193,12 +193,8 @@ pub mod tests {
 
         // simulate leader sending messages
         let (s_responder, r_responder) = channel();
-        let t_responder = streamer::responder(
-            leader.sockets.requests,
-            exit.clone(),
-            resp_recycler.clone(),
-            r_responder,
-        );
+        let t_responder =
+            streamer::responder(leader.sockets.requests, resp_recycler.clone(), r_responder);
 
         let starting_balance = 10_000;
         let mint = Mint::new(starting_balance);
@@ -269,6 +265,7 @@ pub mod tests {
 
         // send the blobs into the socket
         s_responder.send(msgs).expect("send");
+        drop(s_responder);
 
         // receive retransmitted messages
         let timer = Duration::new(1, 0);