pub fn broadcast(
        obj: &Arc<RwLock<Self>>,
        blobs: &Vec<SharedBlob>,
        s: &UdpSocket,
        transmit_index: &mut u64,
    ) -> Result<()> {
        let (me, table): (ReplicatedData, Vec<ReplicatedData>) = {
            // copy to avoid locking durring IO
            let robj = obj.read().unwrap();
            let cloned_table: Vec<ReplicatedData> = robj.table.values().cloned().collect();
            (robj.table[&robj.me].clone(), cloned_table)
        };
        let daddr = "0.0.0.0:0".parse().unwrap();
        let items: Vec<(usize, &ReplicatedData)> = table
            .iter()
            .filter(|v| {
                if me.id == v.id {
                    //filter myself
                    false
                } else if v.replicate_addr == daddr {
                    //filter nodes that are not listening
                    false
                } else {
                    true
                }
            })
            .enumerate()
            .collect();
        let orders: Vec<_> = items.into_iter().cycle().zip(blobs.iter()).collect();
        let errs: Vec<_> = orders
            .into_par_iter()
            .map(|((i, v), b)| {
                // only leader should be broadcasting
                assert!(me.current_leader_id != v.id);
                let mut blob = b.write().unwrap();
                blob.set_id(me.id).expect("set_id");
                blob.set_index(*transmit_index + i as u64)
                    .expect("set_index");
                //TODO profile this, may need multiple sockets for par_iter
                s.send_to(&blob.data[..blob.meta.size], &v.replicate_addr)
            })
            .collect();
        for e in errs {
            trace!("retransmit result {:?}", e);
            match e {
                Err(e) => return Err(Error::IO(e)),
                _ => (),
            }
            *transmit_index += 1;
        }
        Ok(())
    }
/// broadcast messages from the leader to layer 1 nodes
/// # Remarks
/// We need to avoid having obj locked while doing any io, such as the `send_to`
broadcast messages from the leader to layer 1 nodes
broadcast messages from the leader to layer 1 nodes
pub fn broadcast(
        obj: &Arc<RwLock<Self>>,
        blobs: &Vec<SharedBlob>,
        s: &UdpSocket,
        transmit_index: &mut u64,
    ) -> Result<()> {
        let (me, table): (ReplicatedData, Vec<ReplicatedData>) = {
            // copy to avoid locking durring IO
            let robj = obj.read().unwrap();
            let cloned_table: Vec<ReplicatedData> = robj.table.values().cloned().collect();
            (robj.table[&robj.me].clone(), cloned_table)
        };
        let daddr = "0.0.0.0:0".parse().unwrap();
        let items: Vec<(usize, &ReplicatedData)> = table
            .iter()
            .filter(|v| {
                if me.id == v.id {
                    //filter myself
                    false
                } else if v.replicate_addr == daddr {
                    //filter nodes that are not listening
                    false
                } else {
                    true
                }
            })
            .enumerate()
            .collect();
        let orders: Vec<_> = items.into_iter().cycle().zip(blobs.iter()).collect();
        let errs: Vec<_> = orders
            .into_par_iter()
            .map(|((i, v), b)| {
                // only leader should be broadcasting
                assert!(me.current_leader_id != v.id);
                let mut blob = b.write().unwrap();
                blob.set_id(me.id).expect("set_id");
                blob.set_index(*transmit_index + i as u64)
                    .expect("set_index");
                //TODO profile this, may need multiple sockets for par_iter
                s.send_to(&blob.data[..blob.meta.size], &v.replicate_addr)
            })
            .collect();
        for e in errs {
            trace!("retransmit result {:?}", e);
            match e {
                Err(e) => return Err(Error::IO(e)),
                _ => (),
            }
            *transmit_index += 1;
        }
        Ok(())
    }
pub fn broadcast ( obj : & Arc < RwLock < Self > > , blobs : & Vec < SharedBlob > , s : & UdpSocket , transmit_index : & mut u64 , ) -> Result < ( ) > { let ( me , table ) : ( ReplicatedData , Vec < ReplicatedData > ) = { // copy to avoid locking durring IO let robj = obj . read ( ) . unwrap ( ) ; let cloned_table : Vec < ReplicatedData > = robj . table . values ( ) . cloned ( ) . collect ( ) ; ( robj . table [ & robj . me ] . clone ( ) , cloned_table ) } ; let daddr = " " . parse ( ) . unwrap ( ) ; let items : Vec < ( usize , & ReplicatedData ) > = table . iter ( ) . filter ( | v | { if me . id == v . id { //filter myself false } else if v . replicate_addr == daddr { //filter nodes that are not listening false } else { true } } ) . enumerate ( ) . collect ( ) ; let orders : Vec < _ > = items . into_iter ( ) . cycle ( ) . zip ( blobs . iter ( ) ) . collect ( ) ; let errs : Vec < _ > = orders . into_par_iter ( ) . map ( | ( ( i , v ) , b ) | { // only leader should be broadcasting assert ! ( me current_leader_id v id ) ; let mut blob = b . write ( ) . unwrap ( ) ; blob . set_id ( me . id ) . expect ( " " ) ; blob . set_index ( * transmit_index + i as u64 ) . expect ( " " ) ; //TODO profile this, may need multiple sockets for par_iter s . send_to ( & blob . data [ .. blob . meta . size ] , & v . replicate_addr ) } ) . collect ( ) ; for e in errs { trace ! ( " " e ) ; match e { Err ( e ) => return Err ( Error :: IO ( e ) ) , _ => ( ) , } * transmit_index += 1 ; } Ok ( ( ) ) }
broadcast messages from the leader to layer 1 nodes
broadcast messages from the leader to layer 1 nodes
obj
&Arc<RwLock<Self>>
blobs
&Vec<SharedBlob>
s
&UdpSocket
transmit_index
&mut u64
obj
&Arc<RwLock<Self>>
blobs
&Vec<SharedBlob>
s
&UdpSocket
transmit_index
&mut u64
remarks
We need to avoid having obj locked while doing any io, such as the `send_to`
We need to avoid having obj locked while doing any io such as the ` send_to `