From a8e4c637dd58e919220f4346f4427e02733369e6 Mon Sep 17 00:00:00 2001 From: Greg Fitzgerald Date: Mon, 14 May 2018 12:43:38 -0600 Subject: [PATCH] Add historian to pipeline No longer intercept entries to register_entry_id(). Intead, register the ID in the Write stage. EventProcessor is now just being used as a place to store data. Fixes #216 --- src/event_processor.rs | 6 +++++- src/historian.rs | 18 +++++++----------- src/request_processor.rs | 11 +++++------ src/request_stage.rs | 13 +++++-------- src/rpu.rs | 10 ++++++++-- src/tvu.rs | 10 ++++++++-- 6 files changed, 38 insertions(+), 30 deletions(-) diff --git a/src/event_processor.rs b/src/event_processor.rs index 8d3b9cdafefa74..07fc66b10abfc3 100644 --- a/src/event_processor.rs +++ b/src/event_processor.rs @@ -14,6 +14,8 @@ pub struct EventProcessor { pub accountant: Arc, historian_input: Mutex>, historian: Mutex, + pub start_hash: Hash, + pub ms_per_tick: Option, } impl EventProcessor { @@ -25,6 +27,8 @@ impl EventProcessor { accountant: Arc::new(accountant), historian_input: Mutex::new(historian_input), historian: Mutex::new(historian), + start_hash: *start_hash, + ms_per_tick, } } @@ -37,7 +41,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.entry_receiver.lock().unwrap().recv()?; + let entry = historian.entry_receiver.recv()?; self.accountant.register_entry_id(&entry.id); Ok(entry) } diff --git a/src/historian.rs b/src/historian.rs index bed9bb8ac070c3..e7234ece56755f 100644 --- a/src/historian.rs +++ b/src/historian.rs @@ -4,13 +4,12 @@ use entry::Entry; use hash::Hash; use recorder::{ExitReason, Recorder, Signal}; -use std::sync::Mutex; use std::sync::mpsc::{channel, Receiver, Sender, TryRecvError}; use std::thread::{spawn, JoinHandle}; use std::time::Instant; pub struct Historian { - pub entry_receiver: Mutex>, + pub entry_receiver: Receiver, pub thread_hdl: JoinHandle, } @@ -24,7 +23,7 @@ impl Historian { let thread_hdl = Historian::create_recorder(*start_hash, ms_per_tick, event_receiver, entry_sender); Historian { - entry_receiver: Mutex::new(entry_receiver), + entry_receiver, thread_hdl, } } @@ -52,10 +51,7 @@ impl Historian { } pub fn receive(self: &Self) -> Result { - self.entry_receiver - .lock() - .expect("'entry_receiver' lock in pub fn receive") - .try_recv() + self.entry_receiver.try_recv() } } @@ -78,9 +74,9 @@ mod tests { sleep(Duration::new(0, 1_000_000)); input.send(Signal::Tick).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(); + let entry0 = hist.entry_receiver.recv().unwrap(); + let entry1 = hist.entry_receiver.recv().unwrap(); + let entry2 = hist.entry_receiver.recv().unwrap(); assert_eq!(entry0.num_hashes, 0); assert_eq!(entry1.num_hashes, 0); @@ -117,7 +113,7 @@ mod tests { sleep(Duration::from_millis(900)); input.send(Signal::Tick).unwrap(); drop(input); - let entries: Vec = hist.entry_receiver.lock().unwrap().iter().collect(); + let entries: Vec = hist.entry_receiver.iter().collect(); assert!(entries.len() > 1); // Ensure the ID is not the seed. diff --git a/src/request_processor.rs b/src/request_processor.rs index 4e6c815edfbc61..47692e1ef6f8de 100644 --- a/src/request_processor.rs +++ b/src/request_processor.rs @@ -2,12 +2,11 @@ use accountant::Accountant; use bincode::{deserialize, serialize}; -use entry::Entry; use event::Event; -use event_processor::EventProcessor; use packet; use packet::SharedPackets; use rayon::prelude::*; +use recorder::Signal; use request::{Request, Response}; use result::Result; use std::collections::VecDeque; @@ -140,9 +139,8 @@ impl RequestProcessor { pub fn process_request_packets( &self, - event_processor: &EventProcessor, verified_receiver: &Receiver)>>, - entry_sender: &Sender, + signal_sender: &Sender, blob_sender: &streamer::BlobSender, packet_recycler: &packet::PacketRecycler, blob_recycler: &packet::BlobRecycler, @@ -176,8 +174,9 @@ impl RequestProcessor { debug!("events: {} reqs: {}", events.len(), reqs.len()); debug!("process_events"); - let entry = event_processor.process_events(events)?; - entry_sender.send(entry)?; + let results = self.accountant.process_verified_events(events); + let events = results.into_iter().filter_map(|x| x.ok()).collect(); + signal_sender.send(Signal::Events(events))?; debug!("done process_events"); debug!("process_requests"); diff --git a/src/request_stage.rs b/src/request_stage.rs index 26ac76d779aac1..2f9935a2f6e278 100644 --- a/src/request_stage.rs +++ b/src/request_stage.rs @@ -1,9 +1,8 @@ //! The `request_stage` processes thin client Request messages. -use entry::Entry; -use event_processor::EventProcessor; use packet; use packet::SharedPackets; +use recorder::Signal; use request_processor::RequestProcessor; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -13,7 +12,7 @@ use streamer; pub struct RequestStage { pub thread_hdl: JoinHandle<()>, - pub entry_receiver: Receiver, + pub signal_receiver: Receiver, pub blob_receiver: streamer::BlobReceiver, pub request_processor: Arc, } @@ -21,7 +20,6 @@ pub struct RequestStage { impl RequestStage { pub fn new( request_processor: RequestProcessor, - event_processor: Arc, exit: Arc, verified_receiver: Receiver)>>, packet_recycler: packet::PacketRecycler, @@ -29,13 +27,12 @@ impl RequestStage { ) -> Self { let request_processor = Arc::new(request_processor); let request_processor_ = request_processor.clone(); - let (entry_sender, entry_receiver) = channel(); + let (signal_sender, signal_receiver) = 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, + &signal_sender, &blob_sender, &packet_recycler, &blob_recycler, @@ -48,7 +45,7 @@ impl RequestStage { }); RequestStage { thread_hdl, - entry_receiver, + signal_receiver, blob_receiver, request_processor, } diff --git a/src/rpu.rs b/src/rpu.rs index b8954c55426ae6..b5127c16035812 100644 --- a/src/rpu.rs +++ b/src/rpu.rs @@ -6,6 +6,7 @@ use crdt::{Crdt, ReplicatedData}; use entry::Entry; use entry_writer::EntryWriter; use event_processor::EventProcessor; +use historian::Historian; use packet; use request_processor::RequestProcessor; use request_stage::RequestStage; @@ -88,13 +89,18 @@ impl Rpu { let request_processor = RequestProcessor::new(self.event_processor.accountant.clone()); let request_stage = RequestStage::new( request_processor, - self.event_processor.clone(), exit.clone(), sig_verify_stage.verified_receiver, packet_recycler.clone(), blob_recycler.clone(), ); + let historian_stage = Historian::new( + request_stage.signal_receiver, + &self.event_processor.start_hash, + self.event_processor.ms_per_tick, + ); + let (broadcast_sender, broadcast_receiver) = channel(); let t_write = Self::write_service( self.event_processor.accountant.clone(), @@ -102,7 +108,7 @@ impl Rpu { broadcast_sender, blob_recycler.clone(), Mutex::new(writer), - request_stage.entry_receiver, + historian_stage.entry_receiver, ); let broadcast_socket = UdpSocket::bind(local)?; diff --git a/src/tvu.rs b/src/tvu.rs index 8deedbf23cbc07..39e0498d45d364 100644 --- a/src/tvu.rs +++ b/src/tvu.rs @@ -6,6 +6,7 @@ use crdt::{Crdt, ReplicatedData}; use entry::Entry; use entry_writer::EntryWriter; use event_processor::EventProcessor; +use historian::Historian; use ledger; use packet; use request_processor::RequestProcessor; @@ -173,17 +174,22 @@ impl Tvu { let request_processor = RequestProcessor::new(obj.event_processor.accountant.clone()); let request_stage = RequestStage::new( request_processor, - obj.event_processor.clone(), exit.clone(), sig_verify_stage.verified_receiver, packet_recycler.clone(), blob_recycler.clone(), ); + let historian_stage = Historian::new( + request_stage.signal_receiver, + &obj.event_processor.start_hash, + obj.event_processor.ms_per_tick, + ); + let t_write = Self::drain_service( obj.event_processor.accountant.clone(), exit.clone(), - request_stage.entry_receiver, + historian_stage.entry_receiver, ); let t_responder = streamer::responder(