2018-11-15 13:23:26 -08:00
|
|
|
//! Crds Gossip Push overlay
|
|
|
|
//! This module is used to propagate recently created CrdsValues across the network
|
|
|
|
//! Eager push strategy is based on Plumtree
|
|
|
|
//! http://asc.di.fct.unl.pt/~jleitao/pdf/srds07-leitao.pdf
|
|
|
|
//!
|
|
|
|
//! Main differences are:
|
|
|
|
//! 1. There is no `max hop`. Messages are signed with a local wallclock. If they are outside of
|
|
|
|
//! the local nodes wallclock window they are drooped silently.
|
|
|
|
//! 2. The prune set is stored in a Bloom filter.
|
|
|
|
|
2018-12-07 19:16:27 -08:00
|
|
|
use crate::contact_info::ContactInfo;
|
|
|
|
use crate::crds::{Crds, VersionedCrdsValue};
|
2019-08-13 18:04:14 -07:00
|
|
|
use crate::crds_gossip::{get_stake, get_weight, CRDS_GOSSIP_DEFAULT_BLOOM_ITEMS};
|
2018-12-07 19:16:27 -08:00
|
|
|
use crate::crds_gossip_error::CrdsGossipError;
|
|
|
|
use crate::crds_value::{CrdsValue, CrdsValueLabel};
|
2019-06-01 07:55:43 -07:00
|
|
|
use crate::weighted_shuffle::weighted_shuffle;
|
2018-11-15 13:23:26 -08:00
|
|
|
use bincode::serialized_size;
|
|
|
|
use indexmap::map::IndexMap;
|
2019-06-01 07:55:43 -07:00
|
|
|
use itertools::Itertools;
|
2018-12-05 14:12:10 -08:00
|
|
|
use rand;
|
|
|
|
use rand::seq::SliceRandom;
|
2019-10-29 21:02:11 -07:00
|
|
|
use rand::{thread_rng, RngCore};
|
2019-02-18 22:26:22 -08:00
|
|
|
use solana_runtime::bloom::Bloom;
|
2018-11-16 08:04:46 -08:00
|
|
|
use solana_sdk::hash::Hash;
|
2019-11-14 10:24:53 -08:00
|
|
|
use solana_sdk::packet::PACKET_DATA_SIZE;
|
2018-11-15 13:23:26 -08:00
|
|
|
use solana_sdk::pubkey::Pubkey;
|
2019-02-20 17:08:56 -08:00
|
|
|
use solana_sdk::timing::timestamp;
|
2018-11-15 13:23:26 -08:00
|
|
|
use std::cmp;
|
2019-06-26 00:30:16 -07:00
|
|
|
use std::collections::{HashMap, HashSet};
|
2018-11-15 13:23:26 -08:00
|
|
|
|
|
|
|
pub const CRDS_GOSSIP_NUM_ACTIVE: usize = 30;
|
|
|
|
pub const CRDS_GOSSIP_PUSH_FANOUT: usize = 6;
|
2020-02-07 12:38:24 -08:00
|
|
|
// With a fanout of 6, a 1000 node cluster should only take ~4 hops to converge.
|
|
|
|
// However since pushes are stake weighed, some trailing nodes
|
|
|
|
// might need more time to receive values. 30 seconds should be plenty.
|
|
|
|
pub const CRDS_GOSSIP_PUSH_MSG_TIMEOUT_MS: u64 = 30000;
|
2018-12-01 12:00:30 -08:00
|
|
|
pub const CRDS_GOSSIP_PRUNE_MSG_TIMEOUT_MS: u64 = 500;
|
2019-06-26 00:30:16 -07:00
|
|
|
pub const CRDS_GOSSIP_PRUNE_STAKE_THRESHOLD_PCT: f64 = 0.15;
|
2018-11-15 13:23:26 -08:00
|
|
|
|
2019-02-11 16:20:31 -08:00
|
|
|
#[derive(Clone)]
|
2018-11-15 13:23:26 -08:00
|
|
|
pub struct CrdsGossipPush {
|
|
|
|
/// max bytes per message
|
|
|
|
pub max_bytes: usize,
|
|
|
|
/// active set of validators for push
|
|
|
|
active_set: IndexMap<Pubkey, Bloom<Pubkey>>,
|
|
|
|
/// push message queue
|
|
|
|
push_messages: HashMap<CrdsValueLabel, Hash>,
|
2019-06-26 00:30:16 -07:00
|
|
|
/// cache that tracks which validators a message was received from
|
|
|
|
received_cache: HashMap<Hash, (u64, HashSet<Pubkey>)>,
|
2018-11-15 13:23:26 -08:00
|
|
|
pub num_active: usize,
|
|
|
|
pub push_fanout: usize,
|
|
|
|
pub msg_timeout: u64,
|
2018-12-01 12:00:30 -08:00
|
|
|
pub prune_timeout: u64,
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
impl Default for CrdsGossipPush {
|
|
|
|
fn default() -> Self {
|
|
|
|
Self {
|
2019-11-14 10:24:53 -08:00
|
|
|
// Allow upto 64 Crds Values per PUSH
|
|
|
|
max_bytes: PACKET_DATA_SIZE * 64,
|
2018-11-15 13:23:26 -08:00
|
|
|
active_set: IndexMap::new(),
|
|
|
|
push_messages: HashMap::new(),
|
2019-06-26 00:30:16 -07:00
|
|
|
received_cache: HashMap::new(),
|
2018-11-15 13:23:26 -08:00
|
|
|
num_active: CRDS_GOSSIP_NUM_ACTIVE,
|
|
|
|
push_fanout: CRDS_GOSSIP_PUSH_FANOUT,
|
|
|
|
msg_timeout: CRDS_GOSSIP_PUSH_MSG_TIMEOUT_MS,
|
2018-12-01 12:00:30 -08:00
|
|
|
prune_timeout: CRDS_GOSSIP_PRUNE_MSG_TIMEOUT_MS,
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
impl CrdsGossipPush {
|
|
|
|
pub fn num_pending(&self) -> usize {
|
|
|
|
self.push_messages.len()
|
|
|
|
}
|
2019-06-26 00:30:16 -07:00
|
|
|
|
|
|
|
fn prune_stake_threshold(self_stake: u64, origin_stake: u64) -> u64 {
|
|
|
|
let min_path_stake = self_stake.min(origin_stake);
|
2019-07-16 10:20:03 -07:00
|
|
|
((CRDS_GOSSIP_PRUNE_STAKE_THRESHOLD_PCT * min_path_stake as f64).round() as u64).max(1)
|
2019-06-26 00:30:16 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
pub fn prune_received_cache(
|
|
|
|
&mut self,
|
|
|
|
self_pubkey: &Pubkey,
|
|
|
|
origin: &Pubkey,
|
|
|
|
hash: Hash,
|
|
|
|
stakes: &HashMap<Pubkey, u64>,
|
|
|
|
) -> Vec<Pubkey> {
|
|
|
|
let origin_stake = stakes.get(origin).unwrap_or(&0);
|
|
|
|
let self_stake = stakes.get(self_pubkey).unwrap_or(&0);
|
|
|
|
let cache = self.received_cache.get(&hash);
|
|
|
|
if cache.is_none() {
|
|
|
|
return Vec::new();
|
|
|
|
}
|
|
|
|
|
|
|
|
let peers = &cache.unwrap().1;
|
|
|
|
let peer_stake_total: u64 = peers.iter().map(|p| stakes.get(p).unwrap_or(&0)).sum();
|
|
|
|
let prune_stake_threshold = Self::prune_stake_threshold(*self_stake, *origin_stake);
|
|
|
|
if peer_stake_total < prune_stake_threshold {
|
|
|
|
return Vec::new();
|
|
|
|
}
|
|
|
|
|
|
|
|
let staked_peers: Vec<(Pubkey, u64)> = peers
|
|
|
|
.iter()
|
|
|
|
.filter_map(|p| stakes.get(p).map(|s| (*p, *s)))
|
|
|
|
.filter(|(_, s)| *s > 0)
|
|
|
|
.collect();
|
|
|
|
|
|
|
|
let mut seed = [0; 32];
|
|
|
|
seed[0..8].copy_from_slice(&thread_rng().next_u64().to_le_bytes());
|
|
|
|
let shuffle = weighted_shuffle(
|
|
|
|
staked_peers.iter().map(|(_, stake)| *stake).collect_vec(),
|
2019-10-29 21:02:11 -07:00
|
|
|
seed,
|
2019-06-26 00:30:16 -07:00
|
|
|
);
|
|
|
|
|
|
|
|
let mut keep = HashSet::new();
|
|
|
|
let mut peer_stake_sum = 0;
|
|
|
|
for next in shuffle {
|
|
|
|
let (next_peer, next_stake) = staked_peers[next];
|
|
|
|
keep.insert(next_peer);
|
|
|
|
peer_stake_sum += next_stake;
|
|
|
|
if peer_stake_sum >= prune_stake_threshold {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
peers
|
|
|
|
.iter()
|
|
|
|
.filter(|p| !keep.contains(p))
|
|
|
|
.cloned()
|
|
|
|
.collect()
|
|
|
|
}
|
|
|
|
|
2018-11-15 13:23:26 -08:00
|
|
|
/// process a push message to the network
|
|
|
|
pub fn process_push_message(
|
|
|
|
&mut self,
|
|
|
|
crds: &mut Crds,
|
2019-06-26 00:30:16 -07:00
|
|
|
from: &Pubkey,
|
2018-11-15 13:23:26 -08:00
|
|
|
value: CrdsValue,
|
|
|
|
now: u64,
|
|
|
|
) -> Result<Option<VersionedCrdsValue>, CrdsGossipError> {
|
2020-02-07 12:38:24 -08:00
|
|
|
if now
|
|
|
|
> value
|
|
|
|
.wallclock()
|
|
|
|
.checked_add(self.msg_timeout)
|
|
|
|
.unwrap_or_else(|| 0)
|
|
|
|
{
|
2018-11-15 13:23:26 -08:00
|
|
|
return Err(CrdsGossipError::PushMessageTimeout);
|
|
|
|
}
|
|
|
|
if now + self.msg_timeout < value.wallclock() {
|
|
|
|
return Err(CrdsGossipError::PushMessageTimeout);
|
|
|
|
}
|
|
|
|
let label = value.label();
|
|
|
|
let new_value = crds.new_versioned(now, value);
|
|
|
|
let value_hash = new_value.value_hash;
|
2019-06-26 00:30:16 -07:00
|
|
|
if let Some((_, ref mut received_set)) = self.received_cache.get_mut(&value_hash) {
|
|
|
|
received_set.insert(from.clone());
|
|
|
|
return Err(CrdsGossipError::PushMessageAlreadyReceived);
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
let old = crds.insert_versioned(new_value);
|
|
|
|
if old.is_err() {
|
|
|
|
return Err(CrdsGossipError::PushMessageOldVersion);
|
|
|
|
}
|
2019-06-26 00:30:16 -07:00
|
|
|
let mut received_set = HashSet::new();
|
|
|
|
received_set.insert(from.clone());
|
2018-11-15 13:23:26 -08:00
|
|
|
self.push_messages.insert(label, value_hash);
|
2019-06-26 00:30:16 -07:00
|
|
|
self.received_cache.insert(value_hash, (now, received_set));
|
2018-11-15 13:23:26 -08:00
|
|
|
Ok(old.ok().and_then(|opt| opt))
|
|
|
|
}
|
|
|
|
|
|
|
|
/// 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.
|
2019-05-28 18:39:40 -07:00
|
|
|
pub fn new_push_messages(&mut self, crds: &Crds, now: u64) -> HashMap<Pubkey, Vec<CrdsValue>> {
|
2018-11-15 13:23:26 -08:00
|
|
|
let max = self.active_set.len();
|
|
|
|
let mut nodes: Vec<_> = (0..max).collect();
|
2018-12-05 14:12:10 -08:00
|
|
|
nodes.shuffle(&mut rand::thread_rng());
|
2018-11-15 13:23:26 -08:00
|
|
|
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![];
|
2019-05-28 18:39:40 -07:00
|
|
|
let mut push_messages: HashMap<Pubkey, Vec<CrdsValue>> = HashMap::new();
|
2018-11-15 13:23:26 -08:00
|
|
|
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());
|
|
|
|
}
|
2019-05-28 18:39:40 -07:00
|
|
|
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());
|
|
|
|
}
|
|
|
|
}
|
2018-11-15 13:23:26 -08:00
|
|
|
self.push_messages.remove(&v.label());
|
|
|
|
}
|
2019-05-28 18:39:40 -07:00
|
|
|
push_messages
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
/// add the `from` to the peer's filter of nodes
|
2019-03-09 19:28:43 -08:00
|
|
|
pub fn process_prune_msg(&mut self, peer: &Pubkey, origins: &[Pubkey]) {
|
2018-11-15 13:23:26 -08:00
|
|
|
for origin in origins {
|
2019-03-09 19:28:43 -08:00
|
|
|
if let Some(p) = self.active_set.get_mut(peer) {
|
2018-11-15 13:23:26 -08:00
|
|
|
p.add(origin)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
fn compute_need(num_active: usize, active_set_len: usize, ratio: usize) -> usize {
|
|
|
|
let num = active_set_len / ratio;
|
|
|
|
cmp::min(num_active, (num_active - active_set_len) + num)
|
|
|
|
}
|
|
|
|
|
|
|
|
/// refresh the push active set
|
|
|
|
/// * ratio - active_set.len()/ratio is the number of actives to rotate
|
|
|
|
pub fn refresh_push_active_set(
|
|
|
|
&mut self,
|
|
|
|
crds: &Crds,
|
2019-02-20 20:02:47 -08:00
|
|
|
stakes: &HashMap<Pubkey, u64>,
|
2019-03-09 19:28:43 -08:00
|
|
|
self_id: &Pubkey,
|
2018-11-15 13:23:26 -08:00
|
|
|
network_size: usize,
|
|
|
|
ratio: usize,
|
|
|
|
) {
|
|
|
|
let need = Self::compute_need(self.num_active, self.active_set.len(), ratio);
|
|
|
|
let mut new_items = HashMap::new();
|
|
|
|
|
2019-06-01 07:55:43 -07:00
|
|
|
let options: Vec<_> = self.push_options(crds, &self_id, stakes);
|
2019-02-20 17:08:56 -08:00
|
|
|
if options.is_empty() {
|
|
|
|
return;
|
|
|
|
}
|
2019-06-01 07:55:43 -07:00
|
|
|
|
|
|
|
let mut seed = [0; 32];
|
|
|
|
seed[0..8].copy_from_slice(&thread_rng().next_u64().to_le_bytes());
|
|
|
|
let mut shuffle = weighted_shuffle(
|
|
|
|
options.iter().map(|weighted| weighted.0).collect_vec(),
|
2019-10-29 21:02:11 -07:00
|
|
|
seed,
|
2019-06-01 07:55:43 -07:00
|
|
|
)
|
|
|
|
.into_iter();
|
|
|
|
|
2019-02-20 17:08:56 -08:00
|
|
|
while new_items.len() < need {
|
2019-06-01 07:55:43 -07:00
|
|
|
match shuffle.next() {
|
|
|
|
Some(index) => {
|
|
|
|
let item = options[index].1;
|
|
|
|
if self.active_set.get(&item.id).is_some() {
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
if new_items.get(&item.id).is_some() {
|
|
|
|
continue;
|
|
|
|
}
|
2019-08-13 18:04:14 -07:00
|
|
|
let size = cmp::max(CRDS_GOSSIP_DEFAULT_BLOOM_ITEMS, network_size);
|
2019-06-01 07:55:43 -07:00
|
|
|
let mut bloom = Bloom::random(size, 0.1, 1024 * 8 * 4);
|
|
|
|
bloom.add(&item.id);
|
|
|
|
new_items.insert(item.id, bloom);
|
|
|
|
}
|
|
|
|
_ => break,
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
let mut keys: Vec<Pubkey> = self.active_set.keys().cloned().collect();
|
2018-12-05 14:12:10 -08:00
|
|
|
keys.shuffle(&mut rand::thread_rng());
|
2018-11-15 13:23:26 -08:00
|
|
|
let num = keys.len() / ratio;
|
|
|
|
for k in &keys[..num] {
|
2019-10-17 14:19:27 -07:00
|
|
|
self.active_set.swap_remove(k);
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
for (k, v) in new_items {
|
|
|
|
self.active_set.insert(k, v);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-02-26 11:45:10 -08:00
|
|
|
fn push_options<'a>(
|
|
|
|
&self,
|
|
|
|
crds: &'a Crds,
|
|
|
|
self_id: &Pubkey,
|
|
|
|
stakes: &HashMap<Pubkey, u64>,
|
|
|
|
) -> Vec<(f32, &'a ContactInfo)> {
|
|
|
|
crds.table
|
|
|
|
.values()
|
|
|
|
.filter(|v| v.value.contact_info().is_some())
|
|
|
|
.map(|v| (v.value.contact_info().unwrap(), v))
|
|
|
|
.filter(|(info, _)| info.id != *self_id && ContactInfo::is_valid_address(&info.gossip))
|
|
|
|
.map(|(info, value)| {
|
|
|
|
let max_weight = f32::from(u16::max_value()) - 1.0;
|
|
|
|
let last_updated: u64 = value.local_timestamp;
|
|
|
|
let since = ((timestamp() - last_updated) / 1024) as u32;
|
|
|
|
let stake = get_stake(&info.id, stakes);
|
|
|
|
let weight = get_weight(max_weight, since, stake);
|
|
|
|
(weight, info)
|
|
|
|
})
|
|
|
|
.collect()
|
|
|
|
}
|
|
|
|
|
2018-11-15 13:23:26 -08:00
|
|
|
/// purge old pending push messages
|
|
|
|
pub fn purge_old_pending_push_messages(&mut self, crds: &Crds, min_time: u64) {
|
|
|
|
let old_msgs: Vec<CrdsValueLabel> = self
|
|
|
|
.push_messages
|
|
|
|
.iter()
|
|
|
|
.filter_map(|(k, hash)| {
|
|
|
|
if let Some(versioned) = crds.lookup_versioned(k) {
|
|
|
|
if versioned.value.wallclock() < min_time || versioned.value_hash != *hash {
|
|
|
|
Some(k)
|
|
|
|
} else {
|
|
|
|
None
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
Some(k)
|
|
|
|
}
|
2018-12-07 19:01:28 -08:00
|
|
|
})
|
|
|
|
.cloned()
|
2018-11-15 13:23:26 -08:00
|
|
|
.collect();
|
|
|
|
for k in old_msgs {
|
|
|
|
self.push_messages.remove(&k);
|
|
|
|
}
|
|
|
|
}
|
2019-06-26 00:30:16 -07:00
|
|
|
|
|
|
|
/// purge received push message cache
|
|
|
|
pub fn purge_old_received_cache(&mut self, min_time: u64) {
|
2018-11-15 13:23:26 -08:00
|
|
|
let old_msgs: Vec<Hash> = self
|
2019-06-26 00:30:16 -07:00
|
|
|
.received_cache
|
2018-11-15 13:23:26 -08:00
|
|
|
.iter()
|
2019-06-26 00:30:16 -07:00
|
|
|
.filter_map(|(k, (rcvd_time, _))| if *rcvd_time < min_time { Some(k) } else { None })
|
2018-11-15 13:23:26 -08:00
|
|
|
.cloned()
|
|
|
|
.collect();
|
|
|
|
for k in old_msgs {
|
2019-06-26 00:30:16 -07:00
|
|
|
self.received_cache.remove(&k);
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[cfg(test)]
|
|
|
|
mod test {
|
|
|
|
use super::*;
|
2018-12-07 19:16:27 -08:00
|
|
|
use crate::contact_info::ContactInfo;
|
2019-11-03 10:07:51 -08:00
|
|
|
use crate::crds_value::CrdsData;
|
2019-02-20 17:08:56 -08:00
|
|
|
|
2019-06-26 00:30:16 -07:00
|
|
|
#[test]
|
|
|
|
fn test_prune() {
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
|
|
|
let mut stakes = HashMap::new();
|
|
|
|
|
|
|
|
let self_id = Pubkey::new_rand();
|
|
|
|
let origin = Pubkey::new_rand();
|
|
|
|
stakes.insert(self_id, 100);
|
|
|
|
stakes.insert(origin, 100);
|
|
|
|
|
2019-11-03 10:07:51 -08:00
|
|
|
let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&origin, 0,
|
|
|
|
)));
|
2019-06-26 00:30:16 -07:00
|
|
|
let label = value.label();
|
|
|
|
let low_staked_peers = (0..10).map(|_| Pubkey::new_rand());
|
|
|
|
let mut low_staked_set = HashSet::new();
|
|
|
|
low_staked_peers.for_each(|p| {
|
|
|
|
let _ = push.process_push_message(&mut crds, &p, value.clone(), 0);
|
|
|
|
low_staked_set.insert(p);
|
|
|
|
stakes.insert(p, 1);
|
|
|
|
});
|
|
|
|
|
|
|
|
let versioned = crds
|
|
|
|
.lookup_versioned(&label)
|
|
|
|
.expect("versioned value should exist");
|
|
|
|
let hash = versioned.value_hash;
|
|
|
|
let pruned = push.prune_received_cache(&self_id, &origin, hash, &stakes);
|
|
|
|
assert!(
|
|
|
|
pruned.is_empty(),
|
|
|
|
"should not prune if min threshold has not been reached"
|
|
|
|
);
|
|
|
|
|
|
|
|
let high_staked_peer = Pubkey::new_rand();
|
|
|
|
let high_stake = CrdsGossipPush::prune_stake_threshold(100, 100) + 10;
|
|
|
|
stakes.insert(high_staked_peer, high_stake);
|
|
|
|
let _ = push.process_push_message(&mut crds, &high_staked_peer, value.clone(), 0);
|
|
|
|
|
|
|
|
let pruned = push.prune_received_cache(&self_id, &origin, hash, &stakes);
|
|
|
|
assert!(
|
|
|
|
pruned.len() < low_staked_set.len() + 1,
|
|
|
|
"should not prune all peers"
|
|
|
|
);
|
|
|
|
pruned.iter().for_each(|p| {
|
|
|
|
assert!(
|
|
|
|
low_staked_set.contains(p),
|
|
|
|
"only low staked peers should be pruned"
|
|
|
|
);
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
2018-11-15 13:23:26 -08:00
|
|
|
#[test]
|
|
|
|
fn test_process_push() {
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
let label = value.label();
|
|
|
|
// push a new message
|
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value.clone(), 0),
|
2018-11-15 13:23:26 -08:00
|
|
|
Ok(None)
|
|
|
|
);
|
|
|
|
assert_eq!(crds.lookup(&label), Some(&value));
|
|
|
|
|
|
|
|
// push it again
|
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value.clone(), 0),
|
|
|
|
Err(CrdsGossipError::PushMessageAlreadyReceived)
|
2018-11-15 13:23:26 -08:00
|
|
|
);
|
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_process_push_old_version() {
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
2019-03-30 20:37:33 -07:00
|
|
|
let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
|
2018-11-15 13:23:26 -08:00
|
|
|
ci.wallclock = 1;
|
2019-11-03 10:07:51 -08:00
|
|
|
let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
|
2018-11-15 13:23:26 -08:00
|
|
|
|
|
|
|
// push a new message
|
2019-06-26 00:30:16 -07:00
|
|
|
assert_eq!(
|
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value, 0),
|
|
|
|
Ok(None)
|
|
|
|
);
|
2018-11-15 13:23:26 -08:00
|
|
|
|
|
|
|
// push an old version
|
|
|
|
ci.wallclock = 0;
|
2019-11-03 10:07:51 -08:00
|
|
|
let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value, 0),
|
2018-11-15 13:23:26 -08:00
|
|
|
Err(CrdsGossipError::PushMessageOldVersion)
|
|
|
|
);
|
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_process_push_timeout() {
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
|
|
|
let timeout = push.msg_timeout;
|
2019-03-30 20:37:33 -07:00
|
|
|
let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
|
2018-11-15 13:23:26 -08:00
|
|
|
|
|
|
|
// push a version to far in the future
|
|
|
|
ci.wallclock = timeout + 1;
|
2019-11-03 10:07:51 -08:00
|
|
|
let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value, 0),
|
2018-11-15 13:23:26 -08:00
|
|
|
Err(CrdsGossipError::PushMessageTimeout)
|
|
|
|
);
|
|
|
|
|
|
|
|
// push a version to far in the past
|
|
|
|
ci.wallclock = 0;
|
2019-11-03 10:07:51 -08:00
|
|
|
let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value, timeout + 1),
|
2018-11-15 13:23:26 -08:00
|
|
|
Err(CrdsGossipError::PushMessageTimeout)
|
|
|
|
);
|
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_process_push_update() {
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
2019-03-30 20:37:33 -07:00
|
|
|
let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
|
2018-11-15 13:23:26 -08:00
|
|
|
ci.wallclock = 0;
|
2019-11-03 10:07:51 -08:00
|
|
|
let value_old = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
|
2018-11-15 13:23:26 -08:00
|
|
|
|
|
|
|
// push a new message
|
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value_old.clone(), 0),
|
2018-11-15 13:23:26 -08:00
|
|
|
Ok(None)
|
|
|
|
);
|
|
|
|
|
|
|
|
// push an old version
|
|
|
|
ci.wallclock = 1;
|
2019-11-03 10:07:51 -08:00
|
|
|
let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value, 0)
|
2018-11-15 13:23:26 -08:00
|
|
|
.unwrap()
|
|
|
|
.unwrap()
|
|
|
|
.value,
|
|
|
|
value_old
|
|
|
|
);
|
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_compute_need() {
|
|
|
|
assert_eq!(CrdsGossipPush::compute_need(30, 0, 10), 30);
|
|
|
|
assert_eq!(CrdsGossipPush::compute_need(30, 1, 10), 29);
|
|
|
|
assert_eq!(CrdsGossipPush::compute_need(30, 30, 10), 3);
|
|
|
|
assert_eq!(CrdsGossipPush::compute_need(30, 29, 10), 3);
|
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_refresh_active_set() {
|
2018-12-14 12:36:50 -08:00
|
|
|
solana_logger::setup();
|
2018-11-15 13:23:26 -08:00
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let value1 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
|
|
|
|
assert_eq!(crds.insert(value1.clone(), 0), Ok(None));
|
2019-03-09 19:28:43 -08:00
|
|
|
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
|
2018-11-15 13:23:26 -08:00
|
|
|
|
|
|
|
assert!(push.active_set.get(&value1.label().pubkey()).is_some());
|
2019-11-03 10:07:51 -08:00
|
|
|
let value2 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert!(push.active_set.get(&value2.label().pubkey()).is_none());
|
|
|
|
assert_eq!(crds.insert(value2.clone(), 0), Ok(None));
|
|
|
|
for _ in 0..30 {
|
2019-03-09 19:28:43 -08:00
|
|
|
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
|
2018-11-15 13:23:26 -08:00
|
|
|
if push.active_set.get(&value2.label().pubkey()).is_some() {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
assert!(push.active_set.get(&value2.label().pubkey()).is_some());
|
|
|
|
|
|
|
|
for _ in 0..push.num_active {
|
2019-11-03 10:07:51 -08:00
|
|
|
let value2 = CrdsValue::new_unsigned(CrdsData::ContactInfo(
|
|
|
|
ContactInfo::new_localhost(&Pubkey::new_rand(), 0),
|
|
|
|
));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.insert(value2.clone(), 0), Ok(None));
|
|
|
|
}
|
2019-03-09 19:28:43 -08:00
|
|
|
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(push.active_set.len(), push.num_active);
|
|
|
|
}
|
|
|
|
#[test]
|
2019-02-20 17:08:56 -08:00
|
|
|
fn test_active_set_refresh_with_bank() {
|
|
|
|
let time = timestamp() - 1024; //make sure there's at least a 1 second delay
|
|
|
|
let mut crds = Crds::default();
|
2019-02-26 11:45:10 -08:00
|
|
|
let push = CrdsGossipPush::default();
|
2019-02-20 20:02:47 -08:00
|
|
|
let mut stakes = HashMap::new();
|
2019-02-20 17:08:56 -08:00
|
|
|
for i in 1..=100 {
|
2019-11-03 10:07:51 -08:00
|
|
|
let peer = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
time,
|
|
|
|
)));
|
2019-02-20 17:08:56 -08:00
|
|
|
let id = peer.label().pubkey();
|
|
|
|
crds.insert(peer.clone(), time).unwrap();
|
2019-02-20 20:02:47 -08:00
|
|
|
stakes.insert(id, i * 100);
|
2019-02-20 17:08:56 -08:00
|
|
|
}
|
2019-02-26 11:45:10 -08:00
|
|
|
let mut options = push.push_options(&crds, &Pubkey::default(), &stakes);
|
|
|
|
assert!(!options.is_empty());
|
|
|
|
options.sort_by(|(weight_l, _), (weight_r, _)| weight_r.partial_cmp(weight_l).unwrap());
|
|
|
|
// check that the highest stake holder is also the heaviest weighted.
|
|
|
|
assert_eq!(
|
|
|
|
*stakes.get(&options.get(0).unwrap().1.id).unwrap(),
|
|
|
|
10_000_u64
|
2019-02-20 17:08:56 -08:00
|
|
|
);
|
|
|
|
}
|
|
|
|
#[test]
|
2018-11-15 13:23:26 -08:00
|
|
|
fn test_new_push_messages() {
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let peer = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.insert(peer.clone(), 0), Ok(None));
|
2019-03-09 19:28:43 -08:00
|
|
|
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
|
2018-11-15 13:23:26 -08:00
|
|
|
|
2019-11-03 10:07:51 -08:00
|
|
|
let new_msg = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2019-05-28 18:39:40 -07:00
|
|
|
let mut expected = HashMap::new();
|
|
|
|
expected.insert(peer.label().pubkey(), vec![new_msg.clone()]);
|
2019-06-26 00:30:16 -07:00
|
|
|
assert_eq!(
|
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), new_msg, 0),
|
|
|
|
Ok(None)
|
|
|
|
);
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(push.active_set.len(), 1);
|
2019-05-28 18:39:40 -07:00
|
|
|
assert_eq!(push.new_push_messages(&crds, 0), expected);
|
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_personalized_push_messages() {
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let peer_1 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2019-05-28 18:39:40 -07:00
|
|
|
assert_eq!(crds.insert(peer_1.clone(), 0), Ok(None));
|
2019-11-03 10:07:51 -08:00
|
|
|
let peer_2 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2019-05-28 18:39:40 -07:00
|
|
|
assert_eq!(crds.insert(peer_2.clone(), 0), Ok(None));
|
2019-11-03 10:07:51 -08:00
|
|
|
let peer_3 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), peer_3.clone(), 0),
|
2019-05-28 18:39:40 -07:00
|
|
|
Ok(None)
|
2018-11-15 13:23:26 -08:00
|
|
|
);
|
2019-05-28 18:39:40 -07:00
|
|
|
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
|
|
|
|
|
|
|
|
// push 3's contact info to 1 and 2 and 3
|
2019-11-03 10:07:51 -08:00
|
|
|
let new_msg = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&peer_3.pubkey(),
|
|
|
|
0,
|
|
|
|
)));
|
2019-05-28 18:39:40 -07:00
|
|
|
let mut expected = HashMap::new();
|
|
|
|
expected.insert(peer_1.pubkey(), vec![new_msg.clone()]);
|
|
|
|
expected.insert(peer_2.pubkey(), vec![new_msg.clone()]);
|
|
|
|
assert_eq!(push.active_set.len(), 3);
|
|
|
|
assert_eq!(push.new_push_messages(&crds, 0), expected);
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_process_prune() {
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let peer = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.insert(peer.clone(), 0), Ok(None));
|
2019-03-09 19:28:43 -08:00
|
|
|
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
|
2018-11-15 13:23:26 -08:00
|
|
|
|
2019-11-03 10:07:51 -08:00
|
|
|
let new_msg = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2019-05-28 18:39:40 -07:00
|
|
|
let expected = HashMap::new();
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), new_msg.clone(), 0),
|
2018-11-15 13:23:26 -08:00
|
|
|
Ok(None)
|
|
|
|
);
|
2019-03-09 19:28:43 -08:00
|
|
|
push.process_prune_msg(&peer.label().pubkey(), &[new_msg.label().pubkey()]);
|
2019-05-28 18:39:40 -07:00
|
|
|
assert_eq!(push.new_push_messages(&crds, 0), expected);
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_purge_old_pending_push_messages() {
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let peer = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::new_rand(),
|
|
|
|
0,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.insert(peer.clone(), 0), Ok(None));
|
2019-03-09 19:28:43 -08:00
|
|
|
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
|
2018-11-15 13:23:26 -08:00
|
|
|
|
2019-03-30 20:37:33 -07:00
|
|
|
let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
|
2018-11-15 13:23:26 -08:00
|
|
|
ci.wallclock = 1;
|
2019-11-03 10:07:51 -08:00
|
|
|
let new_msg = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
|
2019-05-28 18:39:40 -07:00
|
|
|
let expected = HashMap::new();
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), new_msg.clone(), 1),
|
2018-11-15 13:23:26 -08:00
|
|
|
Ok(None)
|
|
|
|
);
|
|
|
|
push.purge_old_pending_push_messages(&crds, 0);
|
2019-05-28 18:39:40 -07:00
|
|
|
assert_eq!(push.new_push_messages(&crds, 0), expected);
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
#[test]
|
2019-06-26 00:30:16 -07:00
|
|
|
fn test_purge_old_received_cache() {
|
2018-11-15 13:23:26 -08:00
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut push = CrdsGossipPush::default();
|
2019-03-30 20:37:33 -07:00
|
|
|
let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
|
2018-11-15 13:23:26 -08:00
|
|
|
ci.wallclock = 0;
|
2019-11-03 10:07:51 -08:00
|
|
|
let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
|
2018-11-15 13:23:26 -08:00
|
|
|
let label = value.label();
|
|
|
|
// push a new message
|
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value.clone(), 0),
|
2018-11-15 13:23:26 -08:00
|
|
|
Ok(None)
|
|
|
|
);
|
|
|
|
assert_eq!(crds.lookup(&label), Some(&value));
|
|
|
|
|
|
|
|
// push it again
|
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value.clone(), 0),
|
|
|
|
Err(CrdsGossipError::PushMessageAlreadyReceived)
|
2018-11-15 13:23:26 -08:00
|
|
|
);
|
|
|
|
|
|
|
|
// purge the old pushed
|
2019-06-26 00:30:16 -07:00
|
|
|
push.purge_old_received_cache(1);
|
2018-11-15 13:23:26 -08:00
|
|
|
|
|
|
|
// push it again
|
|
|
|
assert_eq!(
|
2019-06-26 00:30:16 -07:00
|
|
|
push.process_push_message(&mut crds, &Pubkey::default(), value.clone(), 0),
|
2018-11-15 13:23:26 -08:00
|
|
|
Err(CrdsGossipError::PushMessageOldVersion)
|
|
|
|
);
|
|
|
|
}
|
|
|
|
}
|