diff --git a/src/event_processor.rs b/src/event_processor.rs index 19f69c9d6..8d3b9cdaf 100644 --- a/src/event_processor.rs +++ b/src/event_processor.rs @@ -37,7 +37,7 @@ impl EventProcessor { sender.send(Signal::Events(events))?; // Wait for the historian to tag our Events with an ID and then register it. - let entry = historian.output.lock().unwrap().recv()?; + let entry = historian.entry_receiver.lock().unwrap().recv()?; self.accountant.register_entry_id(&entry.id); Ok(entry) } diff --git a/src/historian.rs b/src/historian.rs index 351fca9c0..b44fabece 100644 --- a/src/historian.rs +++ b/src/historian.rs @@ -10,7 +10,7 @@ use std::thread::{spawn, JoinHandle}; use std::time::Instant; pub struct Historian { - pub output: Mutex>, + pub entry_receiver: Mutex>, pub thread_hdl: JoinHandle, } @@ -20,11 +20,11 @@ impl Historian { start_hash: &Hash, ms_per_tick: Option, ) -> Self { - let (entry_sender, output) = channel(); + let (entry_sender, entry_receiver) = channel(); let thread_hdl = Historian::create_recorder(*start_hash, ms_per_tick, event_receiver, entry_sender); Historian { - output: Mutex::new(output), + entry_receiver: Mutex::new(entry_receiver), thread_hdl, } } @@ -52,9 +52,9 @@ impl Historian { } pub fn receive(self: &Self) -> Result { - self.output + self.entry_receiver .lock() - .expect("'output' lock in pub fn receive") + .expect("'entry_receiver' lock in pub fn receive") .try_recv() } } @@ -78,9 +78,9 @@ mod tests { sleep(Duration::new(0, 1_000_000)); input.send(Signal::Tick).unwrap(); - let entry0 = hist.output.lock().unwrap().recv().unwrap(); - let entry1 = hist.output.lock().unwrap().recv().unwrap(); - let entry2 = hist.output.lock().unwrap().recv().unwrap(); + let entry0 = hist.entry_receiver.lock().unwrap().recv().unwrap(); + let entry1 = hist.entry_receiver.lock().unwrap().recv().unwrap(); + let entry2 = hist.entry_receiver.lock().unwrap().recv().unwrap(); assert_eq!(entry0.num_hashes, 0); assert_eq!(entry1.num_hashes, 0); @@ -100,7 +100,7 @@ mod tests { let (input, event_receiver) = channel(); let zero = Hash::default(); let hist = Historian::new(event_receiver, &zero, None); - drop(hist.output); + drop(hist.entry_receiver); input.send(Signal::Tick).unwrap(); assert_eq!( hist.thread_hdl.join().unwrap(), @@ -116,7 +116,7 @@ mod tests { sleep(Duration::from_millis(300)); input.send(Signal::Tick).unwrap(); drop(input); - let entries: Vec = hist.output.lock().unwrap().iter().collect(); + let entries: Vec = hist.entry_receiver.lock().unwrap().iter().collect(); assert!(entries.len() > 1); // Ensure the ID is not the seed. diff --git a/src/request_stage.rs b/src/request_stage.rs index 068ce3e81..6b9a693cd 100644 --- a/src/request_stage.rs +++ b/src/request_stage.rs @@ -208,7 +208,7 @@ impl RequestProcessor { event_processor: &EventProcessor, verified_receiver: &Receiver)>>, entry_sender: &Sender, - responder_sender: &streamer::BlobSender, + blob_sender: &streamer::BlobSender, packet_recycler: &packet::PacketRecycler, blob_recycler: &packet::BlobRecycler, ) -> Result<()> { @@ -253,7 +253,7 @@ impl RequestProcessor { if !blobs.is_empty() { info!("process: sending blobs: {}", blobs.len()); //don't wake up the other side if there is nothing - responder_sender.send(blobs)?; + blob_sender.send(blobs)?; } packet_recycler.recycle(msgs); } @@ -274,7 +274,7 @@ impl RequestProcessor { pub struct RequestStage { pub thread_hdl: JoinHandle<()>, pub entry_receiver: Receiver, - pub output: streamer::BlobReceiver, + pub blob_receiver: streamer::BlobReceiver, pub request_processor: Arc, } @@ -290,13 +290,13 @@ impl RequestStage { let request_processor = Arc::new(request_processor); let request_processor_ = request_processor.clone(); let (entry_sender, entry_receiver) = channel(); - let (responder_sender, output) = channel(); + let (blob_sender, blob_receiver) = channel(); let thread_hdl = spawn(move || loop { let e = request_processor_.process_request_packets( &event_processor, &verified_receiver, &entry_sender, - &responder_sender, + &blob_sender, &packet_recycler, &blob_recycler, ); @@ -309,7 +309,7 @@ impl RequestStage { RequestStage { thread_hdl, entry_receiver, - output, + blob_receiver, request_processor, } } diff --git a/src/rpu.rs b/src/rpu.rs index e4e275eee..fe7a2e5eb 100644 --- a/src/rpu.rs +++ b/src/rpu.rs @@ -89,7 +89,7 @@ impl Rpu { request_processor, self.event_processor.clone(), exit.clone(), - sig_verify_stage.output, + sig_verify_stage.verified_receiver, packet_recycler.clone(), blob_recycler.clone(), ); @@ -119,7 +119,7 @@ impl Rpu { respond_socket, exit.clone(), blob_recycler.clone(), - request_stage.output, + request_stage.blob_receiver, ); let mut threads = vec![ diff --git a/src/sig_verify_stage.rs b/src/sig_verify_stage.rs index db6c6dbab..6528537c6 100644 --- a/src/sig_verify_stage.rs +++ b/src/sig_verify_stage.rs @@ -13,17 +13,17 @@ use streamer; use timing; pub struct SigVerifyStage { - pub output: Receiver)>>, + pub verified_receiver: Receiver)>>, pub thread_hdls: Vec>, } impl SigVerifyStage { pub fn new(exit: Arc, packets_receiver: Receiver) -> Self { - let (verified_sender, output) = channel(); + let (verified_sender, verified_receiver) = channel(); let thread_hdls = Self::verifier_services(exit, packets_receiver, verified_sender); SigVerifyStage { thread_hdls, - output, + verified_receiver, } } diff --git a/src/tvu.rs b/src/tvu.rs index 869a9ba4e..59bb599a3 100644 --- a/src/tvu.rs +++ b/src/tvu.rs @@ -174,7 +174,7 @@ impl Tvu { request_processor, obj.event_processor.clone(), exit.clone(), - sig_verify_stage.output, + sig_verify_stage.verified_receiver, packet_recycler.clone(), blob_recycler.clone(), ); @@ -190,7 +190,7 @@ impl Tvu { respond_socket, exit.clone(), blob_recycler.clone(), - request_stage.output, + request_stage.blob_receiver, ); let mut threads = vec![