/// retransmit 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`
retransmit messages from the leader to layer 1 nodes
retransmit messages from the leader to layer 1 nodes
pub fn retransmit(obj: &Arc<RwLock<Self>>, blob: &SharedBlob, s: &UdpSocket) -> Result<()> {
        let (me, table): (ReplicatedData, Vec<ReplicatedData>) = {
            // copy to avoid locking durring IO
            let s = obj.read().unwrap();
            (s.table[&s.me].clone(), s.table.values().cloned().collect())
        };
        let rblob = blob.read().unwrap();
        let daddr = "0.0.0.0:0".parse().unwrap();
        let orders: Vec<_> = table
            .iter()
            .filter(|v| {
                if me.id == v.id {
                    false
                } else if me.current_leader_id == v.id {
                    trace!("skip retransmit to leader {:?}", v.id);
                    false
                } else if v.replicate_addr == daddr {
                    trace!("skip nodes that are not listening {:?}", v.id);
                    false
                } else {
                    true
                }
            })
            .collect();
        let errs: Vec<_> = orders
            .par_iter()
            .map(|v| {
                trace!("retransmit blob to {}", v.replicate_addr);
                //TODO profile this, may need multiple sockets for par_iter
                s.send_to(&rblob.data[..rblob.meta.size], &v.replicate_addr)
            })
            .collect();
        for e in errs {
            trace!("retransmit result {:?}", e);
            match e {
                Err(e) => return Err(Error::IO(e)),
                _ => (),
            }
        }
        Ok(())
    }
pub fn retransmit ( obj : & Arc < RwLock < Self > > , blob : & SharedBlob , s : & UdpSocket ) -> Result < ( ) > { let ( me , table ) : ( ReplicatedData , Vec < ReplicatedData > ) = { // copy to avoid locking durring IO let s = obj . read ( ) . unwrap ( ) ; ( s . table [ & s . me ] . clone ( ) , s . table . values ( ) . cloned ( ) . collect ( ) ) } ; let rblob = blob . read ( ) . unwrap ( ) ; let daddr = " " . parse ( ) . unwrap ( ) ; let orders : Vec < _ > = table . iter ( ) . filter ( | v | { if me . id == v . id { false } else if me . current_leader_id == v . id { trace ! ( " " v id ) ; false } else if v . replicate_addr == daddr { trace ! ( " " v id ) ; false } else { true } } ) . collect ( ) ; let errs : Vec < _ > = orders . par_iter ( ) . map ( | v | { trace ! ( " " v replicate_addr ) ; //TODO profile this, may need multiple sockets for par_iter s . send_to ( & rblob . data [ .. rblob . meta . size ] , & v . replicate_addr ) } ) . collect ( ) ; for e in errs { trace ! ( " " e ) ; match e { Err ( e ) => return Err ( Error :: IO ( e ) ) , _ => ( ) , } } Ok ( ( ) ) }
retransmit messages from the leader to layer 1 nodes
retransmit messages from the leader to layer 1 nodes
obj
&Arc<RwLock<Self>>
blob
&SharedBlob
s
&UdpSocket
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 `