/// This service receives messages from a leader in the network and processes the transactions
/// on the accountant state.
/// # Arguments
/// * `obj` - The accountant state.
/// * `me` - my configuration
/// * `leader` - leader configuration
/// * `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, if erasure coding is insufficient
///     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
This service receives messages from a leader in the network and processes the transactions
on the accountant state.
This service receives messages from a leader in the network and processes the transactions on the accountant state .
pub fn replicate(
        obj: &SharedSkel,
        me: ReplicatedData,
        gossip: UdpSocket,
        serve: UdpSocket,
        replicate: UdpSocket,
        leader: ReplicatedData,
        exit: Arc<AtomicBool>,
    ) -> Result<Vec<JoinHandle<()>>> {
        //replicate pipeline
        let crdt = Arc::new(RwLock::new(Crdt::new(me)));
        crdt.write().unwrap().set_leader(leader.id);
        crdt.write().unwrap().insert(leader);
        let t_gossip = Crdt::gossip(crdt.clone(), exit.clone());
        let t_listen = Crdt::listen(crdt.clone(), gossip, exit.clone());
        // make sure we are on the same interface
        let mut local = replicate.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(),
            replicate,
            blob_sender.clone(),
        )?;
        let (window_sender, window_receiver) = channel();
        let (retransmit_sender, retransmit_receiver) = channel();
        let t_retransmit = streamer::retransmitter(
            write,
            exit.clone(),
            crdt.clone(),
            blob_recycler.clone(),
            retransmit_receiver,
        );
        //TODO
        //the packets coming 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(),
            crdt.clone(),
            blob_recycler.clone(),
            blob_receiver,
            window_sender,
            retransmit_sender,
        );
        let skel = obj.clone();
        let s_exit = exit.clone();
        let t_replicator = spawn(move || loop {
            let e = Self::replicate_state(&skel, &window_receiver, &blob_recycler);
            if e.is_err() && s_exit.load(Ordering::Relaxed) {
                break;
            }
        });
        //serve pipeline
        // make sure we are on the same interface
        let mut local = serve.local_addr()?;
        local.set_port(0);
        let respond_socket = UdpSocket::bind(local.clone())?;
        let packet_recycler = packet::PacketRecycler::default();
        let blob_recycler = packet::BlobRecycler::default();
        let (packet_sender, packet_receiver) = channel();
        let t_packet_receiver =
            streamer::receiver(serve, exit.clone(), packet_recycler.clone(), packet_sender)?;
        let (responder_sender, responder_receiver) = channel();
        let t_responder = streamer::responder(
            respond_socket,
            exit.clone(),
            blob_recycler.clone(),
            responder_receiver,
        );
        let (verified_sender, verified_receiver) = channel();
        let mut verify_threads = Vec::new();
        let shared_verified_sender = Arc::new(Mutex::new(verified_sender));
        let shared_packet_receiver = Arc::new(Mutex::new(packet_receiver));
        for _ in 0..4 {
            let exit_ = exit.clone();
            let recv = shared_packet_receiver.clone();
            let sender = shared_verified_sender.clone();
            let thread = spawn(move || loop {
                let e = Self::verifier(&recv, &sender);
                if e.is_err() && exit_.load(Ordering::Relaxed) {
                    break;
                }
            });
            verify_threads.push(thread);
        }
        let t_sync = Self::sync_no_broadcast_service(obj.clone(), exit.clone());
        let skel = obj.clone();
        let s_exit = exit.clone();
        let t_server = spawn(move || loop {
            let e = Self::process(
                &mut skel.clone(),
                &verified_receiver,
                &responder_sender,
                &packet_recycler,
                &blob_recycler,
            );
            if e.is_err() {
                if s_exit.load(Ordering::Relaxed) {
                    break;
                }
            }
        });
        let mut threads = vec![
            //replicate threads
            t_blob_receiver,
            t_retransmit,
            t_window,
            t_replicator,
            t_gossip,
            t_listen,
            //serve threads
            t_packet_receiver,
            t_responder,
            t_server,
            t_sync,
        ];
        threads.extend(verify_threads.into_iter());
        Ok(threads)
    }
pub fn replicate ( obj : & SharedSkel , me : ReplicatedData , gossip : UdpSocket , serve : UdpSocket , replicate : UdpSocket , leader : ReplicatedData , exit : Arc < AtomicBool > , ) -> Result < Vec < JoinHandle < ( ) > > > { //replicate pipeline let crdt = Arc :: new ( RwLock :: new ( Crdt :: new ( me ) ) ) ; crdt . write ( ) . unwrap ( ) . set_leader ( leader . id ) ; crdt . write ( ) . unwrap ( ) . insert ( leader ) ; let t_gossip = Crdt :: gossip ( crdt . clone ( ) , exit . clone ( ) ) ; let t_listen = Crdt :: listen ( crdt . clone ( ) , gossip , exit . clone ( ) ) ; // make sure we are on the same interface let mut local = replicate . 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 ( ) , replicate , blob_sender . clone ( ) , ) ? ; let ( window_sender , window_receiver ) = channel ( ) ; let ( retransmit_sender , retransmit_receiver ) = channel ( ) ; let t_retransmit = streamer :: retransmitter ( write , exit . clone ( ) , crdt . clone ( ) , blob_recycler . clone ( ) , retransmit_receiver , ) ; //TODO //the packets coming 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 ( ) , crdt . clone ( ) , blob_recycler . clone ( ) , blob_receiver , window_sender , retransmit_sender , ) ; let skel = obj . clone ( ) ; let s_exit = exit . clone ( ) ; let t_replicator = spawn ( move | | loop { let e = Self :: replicate_state ( & skel , & window_receiver , & blob_recycler ) ; if e . is_err ( ) && s_exit . load ( Ordering :: Relaxed ) { break ; } } ) ; //serve pipeline // make sure we are on the same interface let mut local = serve . local_addr ( ) ? ; local . set_port ( 0 ) ; let respond_socket = UdpSocket :: bind ( local . clone ( ) ) ? ; let packet_recycler = packet :: PacketRecycler :: default ( ) ; let blob_recycler = packet :: BlobRecycler :: default ( ) ; let ( packet_sender , packet_receiver ) = channel ( ) ; let t_packet_receiver = streamer :: receiver ( serve , exit . clone ( ) , packet_recycler . clone ( ) , packet_sender ) ? ; let ( responder_sender , responder_receiver ) = channel ( ) ; let t_responder = streamer :: responder ( respond_socket , exit . clone ( ) , blob_recycler . clone ( ) , responder_receiver , ) ; let ( verified_sender , verified_receiver ) = channel ( ) ; let mut verify_threads = Vec :: new ( ) ; let shared_verified_sender = Arc :: new ( Mutex :: new ( verified_sender ) ) ; let shared_packet_receiver = Arc :: new ( Mutex :: new ( packet_receiver ) ) ; for _ in 0 .. 4 { let exit_ = exit . clone ( ) ; let recv = shared_packet_receiver . clone ( ) ; let sender = shared_verified_sender . clone ( ) ; let thread = spawn ( move | | loop { let e = Self :: verifier ( & recv , & sender ) ; if e . is_err ( ) && exit_ . load ( Ordering :: Relaxed ) { break ; } } ) ; verify_threads . push ( thread ) ; } let t_sync = Self :: sync_no_broadcast_service ( obj . clone ( ) , exit . clone ( ) ) ; let skel = obj . clone ( ) ; let s_exit = exit . clone ( ) ; let t_server = spawn ( move | | loop { let e = Self :: process ( & mut skel . clone ( ) , & verified_receiver , & responder_sender , & packet_recycler , & blob_recycler , ) ; if e . is_err ( ) { if s_exit . load ( Ordering :: Relaxed ) { break ; } } } ) ; let mut threads = vec ! [ //replicate threads t_blob_receiver t_retransmit t_window t_replicator t_gossip t_listen //serve threads t_packet_receiver t_responder t_server t_sync ] ; threads . extend ( verify_threads . into_iter ( ) ) ; Ok ( threads ) }
This service receives messages from a leader in the network and processes the transactions
on the accountant state.
This service receives messages from a leader in the network and processes the transactions on the accountant state .
obj
&SharedSkel
me
ReplicatedData
gossip
UdpSocket
serve
UdpSocket
replicate
UdpSocket
leader
ReplicatedData
exit
Arc<AtomicBool>
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, if erasure coding is insufficient
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
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 if erasure coding is insufficient 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