pub fn retransmit(obj: &Arc<RwLock<Self>>, blob: &SharedBlob, s: &UdpSocket) -> Result<()> {
        let (me, table): (NodeInfo, Vec<NodeInfo>) = {
            // copy to avoid locking during IO
            let s = obj.read().expect("'obj' read lock in pub fn retransmit");
            (s.my_data().clone(), s.table.values().cloned().collect())
        };
        blob.write()
            .unwrap()
            .set_id(&me.id)
            .expect("set_id in pub fn retransmit");
        let rblob = blob.read().unwrap();
        let orders: Vec<_> = table
            .iter()
            .filter(|v| {
                if me.id == v.id {
                    trace!("skip retransmit to self {:?}", v.id);
                    false
                } else if me.leader_id == v.id {
                    trace!("skip retransmit to leader {:?}", v.id);
                    false
                } else if !(Self::is_valid_address(&v.contact_info.tvu)) {
                    trace!(
                        "skip nodes that are not listening {:?} {}",
                        v.id,
                        v.contact_info.tvu
                    );
                    false
                } else {
                    true
                }
            }).collect();
        trace!("retransmit orders {}", orders.len());
        let errs: Vec<_> = orders
            .par_iter()
            .map(|v| {
                debug!(
                    "{}: retransmit blob {} to {} {}",
                    me.id,
                    rblob.index().unwrap(),
                    v.id,
                    v.contact_info.tvu,
                );
                //TODO profile this, may need multiple sockets for par_iter
                assert!(rblob.meta.size <= BLOB_SIZE);
                s.send_to(&rblob.data[..rblob.meta.size], &v.contact_info.tvu)
            }).collect();
        for e in errs {
            if let Err(e) = &e {
                inc_new_counter_info!("cluster_info-retransmit-send_to_error", 1, 1);
                error!("retransmit result {:?}", e);
            }
            e?;
        }
        Ok(())
    }
/// 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): (NodeInfo, Vec<NodeInfo>) = {
            // copy to avoid locking during IO
            let s = obj.read().expect("'obj' read lock in pub fn retransmit");
            (s.my_data().clone(), s.table.values().cloned().collect())
        };
        blob.write()
            .unwrap()
            .set_id(&me.id)
            .expect("set_id in pub fn retransmit");
        let rblob = blob.read().unwrap();
        let orders: Vec<_> = table
            .iter()
            .filter(|v| {
                if me.id == v.id {
                    trace!("skip retransmit to self {:?}", v.id);
                    false
                } else if me.leader_id == v.id {
                    trace!("skip retransmit to leader {:?}", v.id);
                    false
                } else if !(Self::is_valid_address(&v.contact_info.tvu)) {
                    trace!(
                        "skip nodes that are not listening {:?} {}",
                        v.id,
                        v.contact_info.tvu
                    );
                    false
                } else {
                    true
                }
            }).collect();
        trace!("retransmit orders {}", orders.len());
        let errs: Vec<_> = orders
            .par_iter()
            .map(|v| {
                debug!(
                    "{}: retransmit blob {} to {} {}",
                    me.id,
                    rblob.index().unwrap(),
                    v.id,
                    v.contact_info.tvu,
                );
                //TODO profile this, may need multiple sockets for par_iter
                assert!(rblob.meta.size <= BLOB_SIZE);
                s.send_to(&rblob.data[..rblob.meta.size], &v.contact_info.tvu)
            }).collect();
        for e in errs {
            if let Err(e) = &e {
                inc_new_counter_info!("cluster_info-retransmit-send_to_error", 1, 1);
                error!("retransmit result {:?}", e);
            }
            e?;
        }
        Ok(())
    }
pub fn retransmit ( obj : & Arc < RwLock < Self > > , blob : & SharedBlob , s : & UdpSocket ) -> Result < ( ) > { let ( me , table ) : ( NodeInfo , Vec < NodeInfo > ) = { // copy to avoid locking during IO let s = obj . read ( ) . expect ( " " ) ; ( s . my_data ( ) . clone ( ) , s . table . values ( ) . cloned ( ) . collect ( ) ) } ; blob . write ( ) . unwrap ( ) . set_id ( & me . id ) . expect ( " " ) ; let rblob = blob . read ( ) . unwrap ( ) ; let orders : Vec < _ > = table . iter ( ) . filter ( | v | { if me . id == v . id { trace ! ( " " v id ) ; false } else if me . leader_id == v . id { trace ! ( " " v id ) ; false } else if ! ( Self :: is_valid_address ( & v . contact_info . tvu ) ) { trace ! ( " " v id v contact_info tvu ) ; false } else { true } } ) . collect ( ) ; trace ! ( " " orders len ( ) ) ; let errs : Vec < _ > = orders . par_iter ( ) . map ( | v | { debug ! ( " " me id rblob index ( ) unwrap ( ) v id v contact_info tvu ) ; //TODO profile this, may need multiple sockets for par_iter assert ! ( rblob meta size BLOB_SIZE ) ; s . send_to ( & rblob . data [ .. rblob . meta . size ] , & v . contact_info . tvu ) } ) . collect ( ) ; for e in errs { if let Err ( e ) = & e { inc_new_counter_info ! ( " " 1 1 ) ; error ! ( " " e ) ; } 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
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 `