[wip] replicator

src/accountant_skel.rs

@@ -22,7 +22,7 @@ use std::io::Write;
 use std::net::{SocketAddr, UdpSocket};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc::{channel, Receiver, Sender};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, RwLock};
 use std::thread::{spawn, JoinHandle};
 use std::time::Duration;
 use streamer;
@@ -247,6 +247,7 @@ impl<W: Write + Send + 'static> AccountantSkel<W> {
     }
 
     /// Create a UDP microservice that forwards messages the given AccountantSkel.
+    /// This service is the network leader
     /// Set `exit` to shutdown its threads.
     pub fn serve(
         obj: &Arc<Mutex<AccountantSkel<W>>>,
@@ -271,15 +272,15 @@ impl<W: Write + Send + 'static> AccountantSkel<W> {
 
         let exit_ = exit.clone();
         let t_verifier = spawn(move || loop {
-            let e = Self::verifier(&packet_receiver, &verified_sender);
+            let e = Self::blob_verifier(&blob_receiver, &verified_sender);
             if e.is_err() && exit_.load(Ordering::Relaxed) {
                 break;
             }
         });
 
         let skel = obj.clone();
         let t_server = spawn(move || loop {
-            let e = AccountantSkel::process(
+            let e = Self::process(
                 &skel,
                 &verified_receiver,
                 &blob_sender,
@@ -292,6 +293,70 @@ impl<W: Write + Send + 'static> AccountantSkel<W> {
         });
         Ok(vec![t_receiver, t_responder, t_server, t_verifier])
     }
+
+    /// This service receives messages from a leader in the network and processes the transactions
+    /// on the accountant state.
+    /// # Arguments
+    /// * `obj` - The accoutnant state.
+    /// * `rsubs` - The subscribers.
+    /// * `exit` - The exit signal.
+    /// # Remarks
+    /// The pipeline is constructed as follows
+    /// 1. receive blobs from the network, these are out of order
+    /// 2. verify blobs, PoH, signatures (TODO)
+    /// 3. reconstruct contiguous window
+    ///     a. order the blobs
+    ///     b. use erasure coding to reconstruct missing blobs
+    ///     c. ask the network for missing blobs
+    ///     d. make sure that the blobs PoH sequences connect (TODO)
+    /// 4. process the transaction state machine
+    /// 5. respond with the hash of the state back to the leader
+    pub fn replicate(
+        obj: &Arc<Mutex<AccountantSkel<W>>>,
+        rsubs: Subscribers,
+        exit: Arc<AtomicBool>,
+    ) -> Result<Vec<JoinHandle<()>>> {
+        let read = UdpSocket::bind(rsubs.me.addr)?;
+        // make sure we are on the same interface
+        let mut local = read.local_addr()?;
+        local.set_port(0);
+        let write = UdpSocket::bind(local)?;
+
+        let blob_recycler = packet::BlobRecycler::default();
+        let (blob_sender, blob_receiver) = channel();
+        let t_blob_receiver =
+            streamer::blob_receiver(exit.clone(), blob_recycler.clone(), read, blob_sender)?;
+        let (window_sender, window_receiver) = channel();
+        let (retransmit_sender, retransmit_receiver) = channel();
+
+        let subs = Arc::new(RwLock::new(rsubs));
+        let t_retransmit = retransmitter(
+            write,
+            exit.clone(),
+            subs,
+            blob_recycler.clone(),
+            retransmit_receiver,
+        );
+        //TODO
+        //the packets comming out of blob_receiver need to be sent to the GPU and verified
+        //then sent to the window, which does the erasure coding reconstruction
+        let t_window = streamer::window(
+            exit.clone(),
+            blob_recycler.clone(),
+            blob_receiver,
+            window_sender,
+            retransmit_sender,
+        );
+
+        let skel = obj.clone();
+        let t_server = spawn(move || loop {
+            let e = Self::replicate_state(&skel, &window_receiver, &blob_sender, &blob_recycler);
+            if e.is_err() && exit.load(Ordering::Relaxed) {
+                break;
+            }
+        });
+        Ok(vec![t_blob_receiver, t_retransmit, t_window, t_server])
+    }
 }
 
 #[cfg(test)]

src/subscribers.rs

@@ -40,18 +40,19 @@ impl Node {
 
 pub struct Subscribers {
     data: Vec<Node>,
-    me: Node,
+    pub me: Node,
     pub leader: Node,
 }
 
 impl Subscribers {
-    pub fn new(me: Node, leader: Node) -> Subscribers {
+    pub fn new(me: Node, leader: Node, network: &[Node]) -> Subscribers {
         let mut h = Subscribers {
             data: vec![],
             me: me.clone(),
             leader: leader.clone(),
         };
         h.insert(&[me, leader]);
+        h.insert(network);
         h
     }