/// 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 fn new_push_messages(&mut self, crds: &Crds, now: u64) -> HashMap<Pubkey, Vec<CrdsValue>> {
        let max = self.active_set.len();
        let mut nodes: Vec<_> = (0..max).collect();
        nodes.shuffle(&mut rand::thread_rng());
        let peers: Vec<Pubkey> = nodes
            .into_iter()
            .filter_map(|n| self.active_set.get_index(n))
            .take(self.push_fanout)
            .map(|n| *n.0)
            .collect();
        let mut total_bytes: usize = 0;
        let mut values = vec![];
        let mut push_messages: HashMap<Pubkey, Vec<CrdsValue>> = HashMap::new();
        for (label, hash) in &self.push_messages {
            let res = crds.lookup_versioned(label);
            if res.is_none() {
                continue;
            }
            let version = res.unwrap();
            if version.value_hash != *hash {
                continue;
            }
            let value = &version.value;
            if value.wallclock() > now || value.wallclock() + self.msg_timeout < now {
                continue;
            }
            total_bytes += serialized_size(value).unwrap() as usize;
            if total_bytes > self.max_bytes {
                break;
            }
            values.push(value.clone());
        }
        for v in values {
            for p in peers.iter() {
                let filter = self.active_set.get_mut(p);
                if filter.is_some() && !filter.unwrap().contains(&v.label().pubkey()) {
                    push_messages.entry(*p).or_default().push(v.clone());
                }
            }
            self.push_messages.remove(&v.label());
        }
        push_messages
    }
pub fn new_push_messages ( & mut self , crds : & Crds , now : u64 ) -> HashMap < Pubkey , Vec < CrdsValue > > { let max = self . active_set . len ( ) ; let mut nodes : Vec < _ > = ( 0 .. max ) . collect ( ) ; nodes . shuffle ( & mut rand :: thread_rng ( ) ) ; let peers : Vec < Pubkey > = nodes . into_iter ( ) . filter_map ( | n | self . active_set . get_index ( n ) ) . take ( self . push_fanout ) . map ( | n | * n . 0 ) . collect ( ) ; let mut total_bytes : usize = 0 ; let mut values = vec ! [ ] ; let mut push_messages : HashMap < Pubkey , Vec < CrdsValue > > = HashMap :: new ( ) ; for ( label , hash ) in & self . push_messages { let res = crds . lookup_versioned ( label ) ; if res . is_none ( ) { continue ; } let version = res . unwrap ( ) ; if version . value_hash != * hash { continue ; } let value = & version . value ; if value . wallclock ( ) > now || value . wallclock ( ) + self . msg_timeout < now { continue ; } total_bytes += serialized_size ( value ) . unwrap ( ) as usize ; if total_bytes > self . max_bytes { break ; } values . push ( value . clone ( ) ) ; } for v in values { for p in peers . iter ( ) { let filter = self . active_set . get_mut ( p ) ; if filter . is_some ( ) && ! filter . unwrap ( ) . contains ( & v . label ( ) . pubkey ( ) ) { push_messages . entry ( * p ) . or_default ( ) . push ( v . clone ( ) ) ; } } self . push_messages . remove ( & v . label ( ) ) ; } push_messages }
New push message to broadcast to peers.
New push message to broadcast to peers .
&mut self
crds
&Crds
now
u64