pub fn broadcast(
        me: &NodeInfo,
        broadcast_table: &[NodeInfo],
        window: &SharedWindow,
        s: &UdpSocket,
        transmit_index: &mut WindowIndex,
        received_index: u64,
    ) -> Result<()> {
        if broadcast_table.is_empty() {
            debug!("{}:not enough peers in cluster_info table", me.id);
            inc_new_counter_info!("cluster_info-broadcast-not_enough_peers_error", 1);
            Err(ClusterInfoError::NoPeers)?;
        }
        trace!(
            "{} transmit_index: {:?} received_index: {} broadcast_len: {}",
            me.id,
            *transmit_index,
            received_index,
            broadcast_table.len()
        );

        let old_transmit_index = transmit_index.data;

        // enumerate all the blobs in the window, those are the indices
        // transmit them to nodes, starting from a different node. Add one
        // to the capacity in case we want to send an extra blob notifying the
        // next leader about the blob right before leader rotation
        let mut orders = Vec::with_capacity((received_index - transmit_index.data + 1) as usize);
        let window_l = window.read().unwrap();

        let mut br_idx = transmit_index.data as usize % broadcast_table.len();

        for idx in transmit_index.data..received_index {
            let w_idx = idx as usize % window_l.len();

            trace!(
                "{} broadcast order data w_idx {} br_idx {}",
                me.id,
                w_idx,
                br_idx
            );

            orders.push((window_l[w_idx].data.clone(), &broadcast_table[br_idx]));
            br_idx += 1;
            br_idx %= broadcast_table.len();
        }

        for idx in transmit_index.coding..received_index {
            let w_idx = idx as usize % window_l.len();

            // skip over empty slots
            if window_l[w_idx].coding.is_none() {
                continue;
            }

            trace!(
                "{} broadcast order coding w_idx: {} br_idx  :{}",
                me.id,
                w_idx,
                br_idx,
            );

            orders.push((window_l[w_idx].coding.clone(), &broadcast_table[br_idx]));
            br_idx += 1;
            br_idx %= broadcast_table.len();
        }

        trace!("broadcast orders table {}", orders.len());
        let errs: Vec<_> = orders
            .into_iter()
            .map(|(b, v)| {
                // only leader should be broadcasting
                assert!(me.leader_id != v.id);
                let bl = b.unwrap();
                let blob = bl.read().unwrap();
                //TODO profile this, may need multiple sockets for par_iter
                trace!(
                    "{}: BROADCAST idx: {} sz: {} to {},{} coding: {}",
                    me.id,
                    blob.index().unwrap(),
                    blob.meta.size,
                    v.id,
                    v.contact_info.tvu,
                    blob.is_coding()
                );
                assert!(blob.meta.size <= BLOB_SIZE);
                let e = s.send_to(&blob.data[..blob.meta.size], &v.contact_info.tvu);
                trace!(
                    "{}: done broadcast {} to {} {}",
                    me.id,
                    blob.meta.size,
                    v.id,
                    v.contact_info.tvu
                );
                e
            }).collect();

        trace!("broadcast results {}", errs.len());
        for e in errs {
            if let Err(e) = &e {
                trace!("broadcast result {:?}", e);
            }
            e?;
            if transmit_index.data < received_index {
                transmit_index.data += 1;
            }
        }
        inc_new_counter_info!(
            "cluster_info-broadcast-max_idx",
            (transmit_index.data - old_transmit_index) as usize
        );
        transmit_index.coding = transmit_index.data;

        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(
        me: &NodeInfo,
        broadcast_table: &[NodeInfo],
        window: &SharedWindow,
        s: &UdpSocket,
        transmit_index: &mut WindowIndex,
        received_index: u64,
    ) -> Result<()> {
        if broadcast_table.is_empty() {
            debug!("{}:not enough peers in cluster_info table", me.id);
            inc_new_counter_info!("cluster_info-broadcast-not_enough_peers_error", 1);
            Err(ClusterInfoError::NoPeers)?;
        }
        trace!(
            "{} transmit_index: {:?} received_index: {} broadcast_len: {}",
            me.id,
            *transmit_index,
            received_index,
            broadcast_table.len()
        );
        let old_transmit_index = transmit_index.data;
        // enumerate all the blobs in the window, those are the indices
        // transmit them to nodes, starting from a different node. Add one
        // to the capacity in case we want to send an extra blob notifying the
        // next leader about the blob right before leader rotation
        let mut orders = Vec::with_capacity((received_index - transmit_index.data + 1) as usize);
        let window_l = window.read().unwrap();
        let mut br_idx = transmit_index.data as usize % broadcast_table.len();
        for idx in transmit_index.data..received_index {
            let w_idx = idx as usize % window_l.len();
            trace!(
                "{} broadcast order data w_idx {} br_idx {}",
                me.id,
                w_idx,
                br_idx
            );
            orders.push((window_l[w_idx].data.clone(), &broadcast_table[br_idx]));
            br_idx += 1;
            br_idx %= broadcast_table.len();
        }
        for idx in transmit_index.coding..received_index {
            let w_idx = idx as usize % window_l.len();
            // skip over empty slots
            if window_l[w_idx].coding.is_none() {
                continue;
            }
            trace!(
                "{} broadcast order coding w_idx: {} br_idx  :{}",
                me.id,
                w_idx,
                br_idx,
            );
            orders.push((window_l[w_idx].coding.clone(), &broadcast_table[br_idx]));
            br_idx += 1;
            br_idx %= broadcast_table.len();
        }
        trace!("broadcast orders table {}", orders.len());
        let errs: Vec<_> = orders
            .into_iter()
            .map(|(b, v)| {
                // only leader should be broadcasting
                assert!(me.leader_id != v.id);
                let bl = b.unwrap();
                let blob = bl.read().unwrap();
                //TODO profile this, may need multiple sockets for par_iter
                trace!(
                    "{}: BROADCAST idx: {} sz: {} to {},{} coding: {}",
                    me.id,
                    blob.index().unwrap(),
                    blob.meta.size,
                    v.id,
                    v.contact_info.tvu,
                    blob.is_coding()
                );
                assert!(blob.meta.size <= BLOB_SIZE);
                let e = s.send_to(&blob.data[..blob.meta.size], &v.contact_info.tvu);
                trace!(
                    "{}: done broadcast {} to {} {}",
                    me.id,
                    blob.meta.size,
                    v.id,
                    v.contact_info.tvu
                );
                e
            }).collect();
        trace!("broadcast results {}", errs.len());
        for e in errs {
            if let Err(e) = &e {
                trace!("broadcast result {:?}", e);
            }
            e?;
            if transmit_index.data < received_index {
                transmit_index.data += 1;
            }
        }
        inc_new_counter_info!(
            "cluster_info-broadcast-max_idx",
            (transmit_index.data - old_transmit_index) as usize
        );
        transmit_index.coding = transmit_index.data;
        Ok(())
    }
pub fn broadcast ( me : & NodeInfo , broadcast_table : & [ NodeInfo ] , window : & SharedWindow , s : & UdpSocket , transmit_index : & mut WindowIndex , received_index : u64 , ) -> Result < ( ) > { if broadcast_table . is_empty ( ) { debug ! ( " " me id ) ; inc_new_counter_info ! ( " " 1 ) ; Err ( ClusterInfoError :: NoPeers ) ? ; } trace ! ( " " me id transmit_index received_index broadcast_table len ( ) ) ; let old_transmit_index = transmit_index . data ; // enumerate all the blobs in the window, those are the indices // transmit them to nodes, starting from a different node. Add one // to the capacity in case we want to send an extra blob notifying the // next leader about the blob right before leader rotation let mut orders = Vec :: with_capacity ( ( received_index - transmit_index . data + 1 ) as usize ) ; let window_l = window . read ( ) . unwrap ( ) ; let mut br_idx = transmit_index . data as usize % broadcast_table . len ( ) ; for idx in transmit_index . data .. received_index { let w_idx = idx as usize % window_l . len ( ) ; trace ! ( " " me id w_idx br_idx ) ; orders . push ( ( window_l [ w_idx ] . data . clone ( ) , & broadcast_table [ br_idx ] ) ) ; br_idx += 1 ; br_idx %= broadcast_table . len ( ) ; } for idx in transmit_index . coding .. received_index { let w_idx = idx as usize % window_l . len ( ) ; // skip over empty slots if window_l [ w_idx ] . coding . is_none ( ) { continue ; } trace ! ( " " me id w_idx br_idx ) ; orders . push ( ( window_l [ w_idx ] . coding . clone ( ) , & broadcast_table [ br_idx ] ) ) ; br_idx += 1 ; br_idx %= broadcast_table . len ( ) ; } trace ! ( " " orders len ( ) ) ; let errs : Vec < _ > = orders . into_iter ( ) . map ( | ( b , v ) | { // only leader should be broadcasting assert ! ( me leader_id v id ) ; let bl = b . unwrap ( ) ; let blob = bl . read ( ) . unwrap ( ) ; //TODO profile this, may need multiple sockets for par_iter trace ! ( " " me id blob index ( ) unwrap ( ) blob meta size v id v contact_info tvu blob is_coding ( ) ) ; assert ! ( blob meta size BLOB_SIZE ) ; let e = s . send_to ( & blob . data [ .. blob . meta . size ] , & v . contact_info . tvu ) ; trace ! ( " " me id blob meta size v id v contact_info tvu ) ; e } ) . collect ( ) ; trace ! ( " " errs len ( ) ) ; for e in errs { if let Err ( e ) = & e { trace ! ( " " e ) ; } e ? ; if transmit_index . data < received_index { transmit_index . data += 1 ; } } inc_new_counter_info ! ( " " ( transmit_index data old_transmit_index ) as usize ) ; transmit_index . coding = transmit_index . data ; Ok ( ( ) ) }
broadcast messages from the leader to layer 1 nodes
broadcast messages from the leader to layer 1 nodes
me
&NodeInfo
broadcast_table
&[NodeInfo]
window
&SharedWindow
s
&UdpSocket
transmit_index
&mut WindowIndex
received_index
u64
me
&NodeInfo
broadcast_table
&[NodeInfo]
window
&SharedWindow
s
&UdpSocket
transmit_index
&mut WindowIndex
received_index
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 `