pub(crate) fn new_push_messages(
        &self,
        crds: &RwLock<Crds>,
        now: u64,
    ) -> HashMap<Pubkey, Vec<CrdsValue>> {
        let active_set = self.active_set.read().unwrap();
        let active_set_len = active_set.len();
        let push_fanout = self.push_fanout.min(active_set_len);
        if push_fanout == 0 {
            return HashMap::default();
        }
        let mut num_pushes = 0;
        let mut num_values = 0;
        let mut total_bytes: usize = 0;
        let mut push_messages: HashMap<Pubkey, Vec<CrdsValue>> = HashMap::new();
        let wallclock_window = self.wallclock_window(now);
        let mut crds_cursor = self.crds_cursor.lock().unwrap();
        // crds should be locked last after self.{active_set,crds_cursor}.
        let crds = crds.read().unwrap();
        let entries = crds
            .get_entries(crds_cursor.deref_mut())
            .map(|entry| &entry.value)
            .filter(|value| wallclock_window.contains(&value.wallclock()));
        for value in entries {
            let serialized_size = serialized_size(&value).unwrap();
            total_bytes = total_bytes.saturating_add(serialized_size as usize);
            if total_bytes > self.max_bytes {
                break;
            }
            num_values += 1;
            let origin = value.pubkey();
            // Use a consistent index for the same origin so the active set
            // learns the MST for that origin.
            let offset = origin.as_ref()[0] as usize;
            for i in offset..offset + push_fanout {
                let index = i % active_set_len;
                let (peer, filter) = active_set.get_index(index).unwrap();
                if !filter.contains(&origin) || value.should_force_push(peer) {
                    trace!("new_push_messages insert {} {:?}", *peer, value);
                    push_messages.entry(*peer).or_default().push(value.clone());
                    num_pushes += 1;
                }
            }
        }
        drop(crds);
        drop(crds_cursor);
        drop(active_set);
        self.num_pushes.fetch_add(num_pushes, Ordering::Relaxed);
        trace!("new_push_messages {} {}", num_values, active_set_len);
        let mut last_pushed_to = self.last_pushed_to.write().unwrap();
        for target_pubkey in push_messages.keys().copied() {
            last_pushed_to.put(target_pubkey, now);
        }
        push_messages
    }
/// New push message to broadcast to peers.
///
/// Returns a list of Pubkeys for the selected peers and a list of values to send to all the
/// peers.
/// The list of push messages is created such that all the randomly selected peers have not
/// pruned the source addresses.
New push message to broadcast to peers.
Returns a list of Pubkeys for the selected peers and a list of values to send to all the
peers.
The list of push messages is created such that all the randomly selected peers have not
pruned the source addresses.
New push message to broadcast to peers . Returns a list of Pubkeys for the selected peers and a list of values to send to all the peers . The list of push messages is created such that all the randomly selected peers have not pruned the source addresses .
pub(crate) fn new_push_messages(
        &self,
        crds: &RwLock<Crds>,
        now: u64,
    ) -> HashMap<Pubkey, Vec<CrdsValue>> {
        let active_set = self.active_set.read().unwrap();
        let active_set_len = active_set.len();
        let push_fanout = self.push_fanout.min(active_set_len);
        if push_fanout == 0 {
            return HashMap::default();
        }
        let mut num_pushes = 0;
        let mut num_values = 0;
        let mut total_bytes: usize = 0;
        let mut push_messages: HashMap<Pubkey, Vec<CrdsValue>> = HashMap::new();
        let wallclock_window = self.wallclock_window(now);
        let mut crds_cursor = self.crds_cursor.lock().unwrap();
        // crds should be locked last after self.{active_set,crds_cursor}.
        let crds = crds.read().unwrap();
        let entries = crds
            .get_entries(crds_cursor.deref_mut())
            .map(|entry| &entry.value)
            .filter(|value| wallclock_window.contains(&value.wallclock()));
        for value in entries {
            let serialized_size = serialized_size(&value).unwrap();
            total_bytes = total_bytes.saturating_add(serialized_size as usize);
            if total_bytes > self.max_bytes {
                break;
            }
            num_values += 1;
            let origin = value.pubkey();
            // Use a consistent index for the same origin so the active set
            // learns the MST for that origin.
            let offset = origin.as_ref()[0] as usize;
            for i in offset..offset + push_fanout {
                let index = i % active_set_len;
                let (peer, filter) = active_set.get_index(index).unwrap();
                if !filter.contains(&origin) || value.should_force_push(peer) {
                    trace!("new_push_messages insert {} {:?}", *peer, value);
                    push_messages.entry(*peer).or_default().push(value.clone());
                    num_pushes += 1;
                }
            }
        }
        drop(crds);
        drop(crds_cursor);
        drop(active_set);
        self.num_pushes.fetch_add(num_pushes, Ordering::Relaxed);
        trace!("new_push_messages {} {}", num_values, active_set_len);
        let mut last_pushed_to = self.last_pushed_to.write().unwrap();
        for target_pubkey in push_messages.keys().copied() {
            last_pushed_to.put(target_pubkey, now);
        }
        push_messages
    }
pub ( crate ) fn new_push_messages ( & self , crds : & RwLock < Crds > , now : u64 , ) -> HashMap < Pubkey , Vec < CrdsValue > > { let active_set = self . active_set . read ( ) . unwrap ( ) ; let active_set_len = active_set . len ( ) ; let push_fanout = self . push_fanout . min ( active_set_len ) ; if push_fanout == 0 { return HashMap :: default ( ) ; } let mut num_pushes = 0 ; let mut num_values = 0 ; let mut total_bytes : usize = 0 ; let mut push_messages : HashMap < Pubkey , Vec < CrdsValue > > = HashMap :: new ( ) ; let wallclock_window = self . wallclock_window ( now ) ; let mut crds_cursor = self . crds_cursor . lock ( ) . unwrap ( ) ; // crds should be locked last after self.{active_set,crds_cursor}. let crds = crds . read ( ) . unwrap ( ) ; let entries = crds . get_entries ( crds_cursor . deref_mut ( ) ) . map ( | entry | & entry . value ) . filter ( | value | wallclock_window . contains ( & value . wallclock ( ) ) ) ; for value in entries { let serialized_size = serialized_size ( & value ) . unwrap ( ) ; total_bytes = total_bytes . saturating_add ( serialized_size as usize ) ; if total_bytes > self . max_bytes { break ; } num_values += 1 ; let origin = value . pubkey ( ) ; // Use a consistent index for the same origin so the active set // learns the MST for that origin. let offset = origin . as_ref ( ) [ 0 ] as usize ; for i in offset .. offset + push_fanout { let index = i % active_set_len ; let ( peer , filter ) = active_set . get_index ( index ) . unwrap ( ) ; if ! filter . contains ( & origin ) || value . should_force_push ( peer ) { trace ! ( " " peer value ) ; push_messages . entry ( * peer ) . or_default ( ) . push ( value . clone ( ) ) ; num_pushes += 1 ; } } } drop ( crds ) ; drop ( crds_cursor ) ; drop ( active_set ) ; self . num_pushes . fetch_add ( num_pushes , Ordering :: Relaxed ) ; trace ! ( " " num_values active_set_len ) ; let mut last_pushed_to = self . last_pushed_to . write ( ) . unwrap ( ) ; for target_pubkey in push_messages . keys ( ) . copied ( ) { last_pushed_to . put ( target_pubkey , now ) ; } push_messages }
New push message to broadcast to peers.
New push message to broadcast to peers .
&self
crds
&RwLock<Crds>
now
u64