2018-11-15 13:23:26 -08:00
|
|
|
//! This module implements Cluster Replicated Data Store for
|
|
|
|
//! asynchronous updates in a distributed network.
|
|
|
|
//!
|
|
|
|
//! Data is stored in the CrdsValue type, each type has a specific
|
|
|
|
//! CrdsValueLabel. Labels are semantically grouped into a single record
|
|
|
|
//! that is identified by a Pubkey.
|
|
|
|
//! * 1 Pubkey maps many CrdsValueLabels
|
|
|
|
//! * 1 CrdsValueLabel maps to 1 CrdsValue
|
|
|
|
//! The Label, the record Pubkey, and all the record labels can be derived
|
|
|
|
//! from a single CrdsValue.
|
|
|
|
//!
|
|
|
|
//! The actual data is stored in a single map of
|
|
|
|
//! `CrdsValueLabel(Pubkey) -> CrdsValue` This allows for partial record
|
|
|
|
//! updates to be propagated through the network.
|
|
|
|
//!
|
|
|
|
//! This means that full `Record` updates are not atomic.
|
|
|
|
//!
|
|
|
|
//! Additional labels can be added by appending them to the CrdsValueLabel,
|
|
|
|
//! CrdsValue enums.
|
|
|
|
//!
|
|
|
|
//! Merge strategy is implemented in:
|
|
|
|
//! impl PartialOrd for VersionedCrdsValue
|
|
|
|
//!
|
|
|
|
//! A value is updated to a new version if the labels match, and the value
|
|
|
|
//! wallclock is later, or the value hash is greater.
|
|
|
|
|
2020-11-04 11:15:58 -08:00
|
|
|
use crate::contact_info::ContactInfo;
|
2020-09-17 07:05:16 -07:00
|
|
|
use crate::crds_shards::CrdsShards;
|
2020-11-15 08:38:04 -08:00
|
|
|
use crate::crds_value::{CrdsData, CrdsValue, CrdsValueLabel};
|
2018-11-15 13:23:26 -08:00
|
|
|
use bincode::serialize;
|
2020-09-17 07:05:16 -07:00
|
|
|
use indexmap::map::{Entry, IndexMap};
|
2020-11-15 08:38:04 -08:00
|
|
|
use indexmap::set::IndexSet;
|
2020-10-23 07:17:37 -07:00
|
|
|
use rayon::{prelude::*, ThreadPool};
|
2018-11-16 08:04:46 -08:00
|
|
|
use solana_sdk::hash::{hash, Hash};
|
2018-11-15 13:23:26 -08:00
|
|
|
use solana_sdk::pubkey::Pubkey;
|
2020-11-15 08:38:04 -08:00
|
|
|
use solana_sdk::signature::Keypair;
|
|
|
|
use solana_sdk::timing::timestamp;
|
2018-11-15 13:23:26 -08:00
|
|
|
use std::cmp;
|
2019-11-20 11:25:18 -08:00
|
|
|
use std::collections::HashMap;
|
2020-09-17 07:05:16 -07:00
|
|
|
use std::ops::Index;
|
|
|
|
|
|
|
|
const CRDS_SHARDS_BITS: u32 = 8;
|
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 Crds {
|
|
|
|
/// Stores the map of labels and values
|
|
|
|
pub table: IndexMap<CrdsValueLabel, VersionedCrdsValue>,
|
2020-06-13 22:03:38 -07:00
|
|
|
pub num_inserts: usize,
|
2020-09-17 07:05:16 -07:00
|
|
|
pub shards: CrdsShards,
|
2020-11-15 08:38:04 -08:00
|
|
|
// Indices of all crds values which are node ContactInfo.
|
|
|
|
nodes: IndexSet<usize>,
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(PartialEq, Debug)]
|
|
|
|
pub enum CrdsError {
|
|
|
|
InsertFailed,
|
|
|
|
}
|
|
|
|
|
2018-12-01 12:00:30 -08:00
|
|
|
/// This structure stores some local metadata associated with the CrdsValue
|
2018-11-15 13:23:26 -08:00
|
|
|
/// The implementation of PartialOrd ensures that the "highest" version is always picked to be
|
|
|
|
/// stored in the Crds
|
2019-02-11 16:20:31 -08:00
|
|
|
#[derive(PartialEq, Debug, Clone)]
|
2018-11-15 13:23:26 -08:00
|
|
|
pub struct VersionedCrdsValue {
|
|
|
|
pub value: CrdsValue,
|
|
|
|
/// local time when inserted
|
|
|
|
pub insert_timestamp: u64,
|
|
|
|
/// local time when updated
|
|
|
|
pub local_timestamp: u64,
|
|
|
|
/// value hash
|
|
|
|
pub value_hash: Hash,
|
|
|
|
}
|
|
|
|
|
|
|
|
impl PartialOrd for VersionedCrdsValue {
|
|
|
|
fn partial_cmp(&self, other: &VersionedCrdsValue) -> Option<cmp::Ordering> {
|
|
|
|
if self.value.label() != other.value.label() {
|
|
|
|
None
|
|
|
|
} else if self.value.wallclock() == other.value.wallclock() {
|
|
|
|
Some(self.value_hash.cmp(&other.value_hash))
|
|
|
|
} else {
|
|
|
|
Some(self.value.wallclock().cmp(&other.value.wallclock()))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
impl VersionedCrdsValue {
|
|
|
|
pub fn new(local_timestamp: u64, value: CrdsValue) -> Self {
|
|
|
|
let value_hash = hash(&serialize(&value).unwrap());
|
|
|
|
VersionedCrdsValue {
|
|
|
|
value,
|
|
|
|
insert_timestamp: local_timestamp,
|
|
|
|
local_timestamp,
|
|
|
|
value_hash,
|
|
|
|
}
|
|
|
|
}
|
2020-11-15 08:38:04 -08:00
|
|
|
|
|
|
|
/// New random VersionedCrdsValue for tests and simulations.
|
|
|
|
pub fn new_rand<R: rand::Rng>(rng: &mut R, keypair: Option<&Keypair>) -> Self {
|
|
|
|
let delay = 10 * 60 * 1000; // 10 minutes
|
|
|
|
let now = timestamp() - delay + rng.gen_range(0, 2 * delay);
|
|
|
|
Self::new(now, CrdsValue::new_rand(rng, keypair))
|
|
|
|
}
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
impl Default for Crds {
|
|
|
|
fn default() -> Self {
|
|
|
|
Crds {
|
2020-11-15 08:38:04 -08:00
|
|
|
table: IndexMap::default(),
|
2020-06-13 22:03:38 -07:00
|
|
|
num_inserts: 0,
|
2020-09-17 07:05:16 -07:00
|
|
|
shards: CrdsShards::new(CRDS_SHARDS_BITS),
|
2020-11-15 08:38:04 -08:00
|
|
|
nodes: IndexSet::default(),
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Crds {
|
|
|
|
/// must be called atomically with `insert_versioned`
|
|
|
|
pub fn new_versioned(&self, local_timestamp: u64, value: CrdsValue) -> VersionedCrdsValue {
|
|
|
|
VersionedCrdsValue::new(local_timestamp, value)
|
|
|
|
}
|
2020-06-09 17:08:13 -07:00
|
|
|
pub fn would_insert(
|
|
|
|
&self,
|
|
|
|
value: CrdsValue,
|
|
|
|
local_timestamp: u64,
|
2020-09-30 17:39:22 -07:00
|
|
|
) -> (bool, VersionedCrdsValue) {
|
2020-06-09 17:08:13 -07:00
|
|
|
let new_value = self.new_versioned(local_timestamp, value);
|
|
|
|
let label = new_value.value.label();
|
2020-09-30 17:39:22 -07:00
|
|
|
// New value is outdated and fails to insert, if it already exists in
|
|
|
|
// the table with a more recent wallclock.
|
|
|
|
let outdated = matches!(self.table.get(&label), Some(current) if new_value <= *current);
|
|
|
|
(!outdated, new_value)
|
2020-06-09 17:08:13 -07:00
|
|
|
}
|
2018-11-15 13:23:26 -08:00
|
|
|
/// insert the new value, returns the old value if insert succeeds
|
|
|
|
pub fn insert_versioned(
|
|
|
|
&mut self,
|
|
|
|
new_value: VersionedCrdsValue,
|
|
|
|
) -> Result<Option<VersionedCrdsValue>, CrdsError> {
|
|
|
|
let label = new_value.value.label();
|
2020-09-17 07:05:16 -07:00
|
|
|
match self.table.entry(label) {
|
|
|
|
Entry::Vacant(entry) => {
|
2020-11-15 08:38:04 -08:00
|
|
|
let entry_index = entry.index();
|
|
|
|
assert!(self.shards.insert(entry_index, &new_value));
|
|
|
|
if let CrdsData::ContactInfo(_) = new_value.value.data {
|
|
|
|
assert!(self.nodes.insert(entry_index));
|
|
|
|
}
|
2020-09-17 07:05:16 -07:00
|
|
|
entry.insert(new_value);
|
|
|
|
self.num_inserts += 1;
|
|
|
|
Ok(None)
|
|
|
|
}
|
|
|
|
Entry::Occupied(mut entry) if *entry.get() < new_value => {
|
|
|
|
let index = entry.index();
|
|
|
|
assert!(self.shards.remove(index, entry.get()));
|
|
|
|
assert!(self.shards.insert(index, &new_value));
|
|
|
|
self.num_inserts += 1;
|
|
|
|
Ok(Some(entry.insert(new_value)))
|
|
|
|
}
|
|
|
|
_ => {
|
|
|
|
trace!(
|
|
|
|
"INSERT FAILED data: {} new.wallclock: {}",
|
|
|
|
new_value.value.label(),
|
|
|
|
new_value.value.wallclock(),
|
|
|
|
);
|
|
|
|
Err(CrdsError::InsertFailed)
|
|
|
|
}
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
pub fn insert(
|
|
|
|
&mut self,
|
|
|
|
value: CrdsValue,
|
|
|
|
local_timestamp: u64,
|
|
|
|
) -> Result<Option<VersionedCrdsValue>, CrdsError> {
|
|
|
|
let new_value = self.new_versioned(local_timestamp, value);
|
|
|
|
self.insert_versioned(new_value)
|
|
|
|
}
|
|
|
|
pub fn lookup(&self, label: &CrdsValueLabel) -> Option<&CrdsValue> {
|
|
|
|
self.table.get(label).map(|x| &x.value)
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn lookup_versioned(&self, label: &CrdsValueLabel) -> Option<&VersionedCrdsValue> {
|
|
|
|
self.table.get(label)
|
|
|
|
}
|
|
|
|
|
2020-11-04 11:15:58 -08:00
|
|
|
pub fn get_contact_info(&self, pubkey: &Pubkey) -> Option<&ContactInfo> {
|
|
|
|
let label = CrdsValueLabel::ContactInfo(*pubkey);
|
|
|
|
self.table.get(&label)?.value.contact_info()
|
|
|
|
}
|
|
|
|
|
2020-11-15 08:38:04 -08:00
|
|
|
/// Returns all entries which are ContactInfo.
|
|
|
|
pub fn get_nodes(&self) -> impl Iterator<Item = &VersionedCrdsValue> {
|
|
|
|
self.nodes.iter().map(move |i| self.table.index(*i))
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Returns ContactInfo of all known nodes.
|
|
|
|
pub fn get_nodes_contact_info(&self) -> impl Iterator<Item = &ContactInfo> {
|
|
|
|
self.get_nodes().map(|v| match &v.value.data {
|
|
|
|
CrdsData::ContactInfo(info) => info,
|
|
|
|
_ => panic!("this should not happen!"),
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2018-11-15 13:23:26 -08:00
|
|
|
fn update_label_timestamp(&mut self, id: &CrdsValueLabel, now: u64) {
|
|
|
|
if let Some(e) = self.table.get_mut(id) {
|
|
|
|
e.local_timestamp = cmp::max(e.local_timestamp, now);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-06-17 20:54:52 -07:00
|
|
|
/// Update the timestamp's of all the labels that are associated with Pubkey
|
2019-03-09 19:28:43 -08:00
|
|
|
pub fn update_record_timestamp(&mut self, pubkey: &Pubkey, now: u64) {
|
2018-11-15 13:23:26 -08:00
|
|
|
for label in &CrdsValue::record_labels(pubkey) {
|
|
|
|
self.update_label_timestamp(label, now);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-11-20 11:25:18 -08:00
|
|
|
/// Find all the keys that are older or equal to the timeout.
|
|
|
|
/// * timeouts - Pubkey specific timeouts with Pubkey::default() as the default timeout.
|
|
|
|
pub fn find_old_labels(
|
|
|
|
&self,
|
2020-10-23 07:17:37 -07:00
|
|
|
thread_pool: &ThreadPool,
|
2019-11-20 11:25:18 -08:00
|
|
|
now: u64,
|
|
|
|
timeouts: &HashMap<Pubkey, u64>,
|
|
|
|
) -> Vec<CrdsValueLabel> {
|
2020-09-29 02:04:40 -07:00
|
|
|
let default_timeout = *timeouts
|
2019-11-20 11:25:18 -08:00
|
|
|
.get(&Pubkey::default())
|
|
|
|
.expect("must have default timeout");
|
2020-10-23 07:17:37 -07:00
|
|
|
thread_pool.install(|| {
|
|
|
|
self.table
|
|
|
|
.par_iter()
|
|
|
|
.with_min_len(1024)
|
|
|
|
.filter_map(|(k, v)| {
|
|
|
|
let timeout = timeouts.get(&k.pubkey()).unwrap_or(&default_timeout);
|
|
|
|
if v.local_timestamp.saturating_add(*timeout) <= now {
|
|
|
|
Some(k.clone())
|
|
|
|
} else {
|
|
|
|
None
|
|
|
|
}
|
|
|
|
})
|
|
|
|
.collect()
|
|
|
|
})
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
|
2020-10-23 07:17:37 -07:00
|
|
|
pub fn remove(&mut self, key: &CrdsValueLabel) -> Option<VersionedCrdsValue> {
|
|
|
|
let (index, _, value) = self.table.swap_remove_full(key)?;
|
|
|
|
assert!(self.shards.remove(index, &value));
|
2020-11-15 08:38:04 -08:00
|
|
|
if let CrdsData::ContactInfo(_) = value.value.data {
|
|
|
|
assert!(self.nodes.swap_remove(&index));
|
|
|
|
}
|
|
|
|
// If index == self.table.len(), then the removed entry was the last
|
|
|
|
// entry in the table, in which case no other keys were modified.
|
|
|
|
// Otherwise, the previously last element in the table is now moved to
|
|
|
|
// the 'index' position; and so shards and nodes need to be updated
|
|
|
|
// accordingly.
|
|
|
|
let size = self.table.len();
|
|
|
|
if index < size {
|
2020-10-23 07:17:37 -07:00
|
|
|
let value = self.table.index(index);
|
2020-11-15 08:38:04 -08:00
|
|
|
assert!(self.shards.remove(size, value));
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(self.shards.insert(index, value));
|
2020-11-15 08:38:04 -08:00
|
|
|
if let CrdsData::ContactInfo(_) = value.value.data {
|
|
|
|
assert!(self.nodes.swap_remove(&size));
|
|
|
|
assert!(self.nodes.insert(index));
|
|
|
|
}
|
2020-09-17 07:05:16 -07:00
|
|
|
}
|
2020-10-23 07:17:37 -07:00
|
|
|
Some(value)
|
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;
|
2020-09-17 07:05:16 -07:00
|
|
|
use rand::{thread_rng, Rng};
|
2020-10-23 07:17:37 -07:00
|
|
|
use rayon::ThreadPoolBuilder;
|
2018-11-15 13:23:26 -08:00
|
|
|
|
|
|
|
#[test]
|
|
|
|
fn test_insert() {
|
|
|
|
let mut crds = Crds::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.insert(val.clone(), 0).ok(), Some(None));
|
|
|
|
assert_eq!(crds.table.len(), 1);
|
|
|
|
assert!(crds.table.contains_key(&val.label()));
|
|
|
|
assert_eq!(crds.table[&val.label()].local_timestamp, 0);
|
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_update_old() {
|
|
|
|
let mut crds = Crds::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.insert(val.clone(), 0), Ok(None));
|
|
|
|
assert_eq!(crds.insert(val.clone(), 1), Err(CrdsError::InsertFailed));
|
|
|
|
assert_eq!(crds.table[&val.label()].local_timestamp, 0);
|
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_update_new() {
|
|
|
|
let mut crds = Crds::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let original = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::default(),
|
|
|
|
0,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_matches!(crds.insert(original.clone(), 0), Ok(_));
|
2019-11-03 10:07:51 -08:00
|
|
|
let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::default(),
|
|
|
|
1,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(
|
|
|
|
crds.insert(val.clone(), 1).unwrap().unwrap().value,
|
|
|
|
original
|
|
|
|
);
|
|
|
|
assert_eq!(crds.table[&val.label()].local_timestamp, 1);
|
|
|
|
}
|
|
|
|
#[test]
|
2019-03-08 19:28:19 -08:00
|
|
|
fn test_update_timestamp() {
|
2018-11-15 13:23:26 -08:00
|
|
|
let mut crds = Crds::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::default(),
|
|
|
|
0,
|
|
|
|
)));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.insert(val.clone(), 0), Ok(None));
|
|
|
|
|
|
|
|
crds.update_label_timestamp(&val.label(), 1);
|
|
|
|
assert_eq!(crds.table[&val.label()].local_timestamp, 1);
|
|
|
|
assert_eq!(crds.table[&val.label()].insert_timestamp, 0);
|
|
|
|
|
2019-11-03 10:07:51 -08:00
|
|
|
let val2 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(val2.label().pubkey(), val.label().pubkey());
|
2019-03-08 19:28:19 -08:00
|
|
|
assert_matches!(crds.insert(val2.clone(), 0), Ok(Some(_)));
|
2018-11-15 13:23:26 -08:00
|
|
|
|
2019-03-09 19:28:43 -08:00
|
|
|
crds.update_record_timestamp(&val.label().pubkey(), 2);
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.table[&val.label()].local_timestamp, 2);
|
|
|
|
assert_eq!(crds.table[&val.label()].insert_timestamp, 0);
|
|
|
|
assert_eq!(crds.table[&val2.label()].local_timestamp, 2);
|
|
|
|
assert_eq!(crds.table[&val2.label()].insert_timestamp, 0);
|
|
|
|
|
2019-03-09 19:28:43 -08:00
|
|
|
crds.update_record_timestamp(&val.label().pubkey(), 1);
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.table[&val.label()].local_timestamp, 2);
|
|
|
|
assert_eq!(crds.table[&val.label()].insert_timestamp, 0);
|
|
|
|
|
|
|
|
let mut ci = ContactInfo::default();
|
|
|
|
ci.wallclock += 1;
|
2019-11-03 10:07:51 -08:00
|
|
|
let val3 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci));
|
2020-05-15 09:35:43 -07:00
|
|
|
assert_matches!(crds.insert(val3, 3), Ok(Some(_)));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.table[&val2.label()].local_timestamp, 3);
|
|
|
|
assert_eq!(crds.table[&val2.label()].insert_timestamp, 3);
|
|
|
|
}
|
|
|
|
#[test]
|
2019-11-20 11:25:18 -08:00
|
|
|
fn test_find_old_records_default() {
|
2020-10-23 07:17:37 -07:00
|
|
|
let thread_pool = ThreadPoolBuilder::new().build().unwrap();
|
2019-11-20 11:25:18 -08:00
|
|
|
let mut crds = Crds::default();
|
|
|
|
let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
|
|
|
|
assert_eq!(crds.insert(val.clone(), 1), Ok(None));
|
|
|
|
let mut set = HashMap::new();
|
|
|
|
set.insert(Pubkey::default(), 0);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(crds.find_old_labels(&thread_pool, 0, &set).is_empty());
|
2019-11-20 11:25:18 -08:00
|
|
|
set.insert(Pubkey::default(), 1);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert_eq!(
|
|
|
|
crds.find_old_labels(&thread_pool, 2, &set),
|
|
|
|
vec![val.label()]
|
|
|
|
);
|
2019-11-20 11:25:18 -08:00
|
|
|
set.insert(Pubkey::default(), 2);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert_eq!(
|
|
|
|
crds.find_old_labels(&thread_pool, 4, &set),
|
|
|
|
vec![val.label()]
|
|
|
|
);
|
2019-11-20 11:25:18 -08:00
|
|
|
}
|
|
|
|
#[test]
|
2020-09-29 02:04:40 -07:00
|
|
|
fn test_find_old_records_with_override() {
|
2020-10-23 07:17:37 -07:00
|
|
|
let thread_pool = ThreadPoolBuilder::new().build().unwrap();
|
2020-09-29 02:04:40 -07:00
|
|
|
let mut rng = thread_rng();
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut timeouts = HashMap::new();
|
2020-11-15 08:38:04 -08:00
|
|
|
let val = CrdsValue::new_rand(&mut rng, None);
|
2020-09-29 02:04:40 -07:00
|
|
|
timeouts.insert(Pubkey::default(), 3);
|
|
|
|
assert_eq!(crds.insert(val.clone(), 0), Ok(None));
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(crds.find_old_labels(&thread_pool, 2, &timeouts).is_empty());
|
2020-09-29 02:04:40 -07:00
|
|
|
timeouts.insert(val.pubkey(), 1);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert_eq!(
|
|
|
|
crds.find_old_labels(&thread_pool, 2, &timeouts),
|
|
|
|
vec![val.label()]
|
|
|
|
);
|
2020-09-29 02:04:40 -07:00
|
|
|
timeouts.insert(val.pubkey(), u64::MAX);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(crds.find_old_labels(&thread_pool, 2, &timeouts).is_empty());
|
2020-09-29 02:04:40 -07:00
|
|
|
timeouts.insert(Pubkey::default(), 1);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(crds.find_old_labels(&thread_pool, 2, &timeouts).is_empty());
|
2020-09-29 02:04:40 -07:00
|
|
|
timeouts.remove(&val.pubkey());
|
2020-10-23 07:17:37 -07:00
|
|
|
assert_eq!(
|
|
|
|
crds.find_old_labels(&thread_pool, 2, &timeouts),
|
|
|
|
vec![val.label()]
|
|
|
|
);
|
2020-09-29 02:04:40 -07:00
|
|
|
}
|
|
|
|
#[test]
|
2019-11-20 11:25:18 -08:00
|
|
|
fn test_remove_default() {
|
2020-10-23 07:17:37 -07:00
|
|
|
let thread_pool = ThreadPoolBuilder::new().build().unwrap();
|
2019-11-20 11:25:18 -08:00
|
|
|
let mut crds = Crds::default();
|
|
|
|
let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
|
|
|
|
assert_matches!(crds.insert(val.clone(), 1), Ok(_));
|
|
|
|
let mut set = HashMap::new();
|
|
|
|
set.insert(Pubkey::default(), 1);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert_eq!(
|
|
|
|
crds.find_old_labels(&thread_pool, 2, &set),
|
|
|
|
vec![val.label()]
|
|
|
|
);
|
2019-11-20 11:25:18 -08:00
|
|
|
crds.remove(&val.label());
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(crds.find_old_labels(&thread_pool, 2, &set).is_empty());
|
2019-11-20 11:25:18 -08:00
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_find_old_records_staked() {
|
2020-10-23 07:17:37 -07:00
|
|
|
let thread_pool = ThreadPoolBuilder::new().build().unwrap();
|
2018-11-15 13:23:26 -08:00
|
|
|
let mut crds = Crds::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_eq!(crds.insert(val.clone(), 1), Ok(None));
|
2019-11-20 11:25:18 -08:00
|
|
|
let mut set = HashMap::new();
|
|
|
|
//now < timestamp
|
|
|
|
set.insert(Pubkey::default(), 0);
|
|
|
|
set.insert(val.pubkey(), 0);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(crds.find_old_labels(&thread_pool, 0, &set).is_empty());
|
2018-11-15 13:23:26 -08:00
|
|
|
|
2019-11-20 11:25:18 -08:00
|
|
|
//pubkey shouldn't expire since its timeout is MAX
|
|
|
|
set.insert(val.pubkey(), std::u64::MAX);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(crds.find_old_labels(&thread_pool, 2, &set).is_empty());
|
2019-11-20 11:25:18 -08:00
|
|
|
|
|
|
|
//default has max timeout, but pubkey should still expire
|
|
|
|
set.insert(Pubkey::default(), std::u64::MAX);
|
|
|
|
set.insert(val.pubkey(), 1);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert_eq!(
|
|
|
|
crds.find_old_labels(&thread_pool, 2, &set),
|
|
|
|
vec![val.label()]
|
|
|
|
);
|
2019-11-20 11:25:18 -08:00
|
|
|
|
|
|
|
set.insert(val.pubkey(), 2);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(crds.find_old_labels(&thread_pool, 2, &set).is_empty());
|
|
|
|
assert_eq!(
|
|
|
|
crds.find_old_labels(&thread_pool, 3, &set),
|
|
|
|
vec![val.label()]
|
|
|
|
);
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
2019-11-20 11:25:18 -08:00
|
|
|
|
2020-09-17 07:05:16 -07:00
|
|
|
#[test]
|
|
|
|
fn test_crds_shards() {
|
|
|
|
fn check_crds_shards(crds: &Crds) {
|
|
|
|
crds.shards
|
|
|
|
.check(&crds.table.values().cloned().collect::<Vec<_>>());
|
|
|
|
}
|
|
|
|
|
|
|
|
let mut crds = Crds::default();
|
2020-11-15 08:38:04 -08:00
|
|
|
let keypairs: Vec<_> = std::iter::repeat_with(Keypair::new).take(256).collect();
|
2020-09-17 07:05:16 -07:00
|
|
|
let mut rng = thread_rng();
|
|
|
|
let mut num_inserts = 0;
|
2020-11-15 08:38:04 -08:00
|
|
|
let mut num_overrides = 0;
|
2020-09-17 07:05:16 -07:00
|
|
|
for _ in 0..4096 {
|
2020-11-15 08:38:04 -08:00
|
|
|
let keypair = &keypairs[rng.gen_range(0, keypairs.len())];
|
|
|
|
let value = VersionedCrdsValue::new_rand(&mut rng, Some(keypair));
|
|
|
|
match crds.insert_versioned(value) {
|
|
|
|
Ok(None) => {
|
|
|
|
num_inserts += 1;
|
|
|
|
check_crds_shards(&crds);
|
|
|
|
}
|
|
|
|
Ok(Some(_)) => {
|
|
|
|
num_inserts += 1;
|
|
|
|
num_overrides += 1;
|
|
|
|
check_crds_shards(&crds);
|
|
|
|
}
|
|
|
|
Err(_) => (),
|
2020-09-17 07:05:16 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
assert_eq!(num_inserts, crds.num_inserts);
|
|
|
|
assert!(num_inserts > 700);
|
2020-11-15 08:38:04 -08:00
|
|
|
assert!(num_overrides > 500);
|
2020-09-17 07:05:16 -07:00
|
|
|
assert!(crds.table.len() > 200);
|
|
|
|
assert!(num_inserts > crds.table.len());
|
|
|
|
check_crds_shards(&crds);
|
|
|
|
// Remove values one by one and assert that shards stay valid.
|
|
|
|
while !crds.table.is_empty() {
|
|
|
|
let index = rng.gen_range(0, crds.table.len());
|
|
|
|
let key = crds.table.get_index(index).unwrap().0.clone();
|
|
|
|
crds.remove(&key);
|
|
|
|
check_crds_shards(&crds);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-11-15 08:38:04 -08:00
|
|
|
#[test]
|
|
|
|
fn test_crds_nodes() {
|
|
|
|
fn check_crds_nodes(crds: &Crds) -> usize {
|
|
|
|
let num_nodes = crds
|
|
|
|
.table
|
|
|
|
.values()
|
|
|
|
.filter(|value| matches!(value.value.data, CrdsData::ContactInfo(_)))
|
|
|
|
.count();
|
|
|
|
assert_eq!(num_nodes, crds.get_nodes_contact_info().count());
|
|
|
|
num_nodes
|
|
|
|
}
|
|
|
|
let mut rng = thread_rng();
|
|
|
|
let keypairs: Vec<_> = std::iter::repeat_with(Keypair::new).take(256).collect();
|
|
|
|
let mut crds = Crds::default();
|
|
|
|
let mut num_inserts = 0;
|
|
|
|
let mut num_overrides = 0;
|
|
|
|
for _ in 0..4096 {
|
|
|
|
let keypair = &keypairs[rng.gen_range(0, keypairs.len())];
|
|
|
|
let value = VersionedCrdsValue::new_rand(&mut rng, Some(keypair));
|
|
|
|
match crds.insert_versioned(value) {
|
|
|
|
Ok(None) => {
|
|
|
|
num_inserts += 1;
|
|
|
|
check_crds_nodes(&crds);
|
|
|
|
}
|
|
|
|
Ok(Some(_)) => {
|
|
|
|
num_inserts += 1;
|
|
|
|
num_overrides += 1;
|
|
|
|
check_crds_nodes(&crds);
|
|
|
|
}
|
|
|
|
Err(_) => (),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
assert_eq!(num_inserts, crds.num_inserts);
|
|
|
|
assert!(num_inserts > 700);
|
|
|
|
assert!(num_overrides > 500);
|
|
|
|
assert!(crds.table.len() > 200);
|
|
|
|
assert!(num_inserts > crds.table.len());
|
|
|
|
let num_nodes = check_crds_nodes(&crds);
|
|
|
|
assert!(num_nodes * 3 < crds.table.len());
|
|
|
|
assert!(num_nodes > 150);
|
|
|
|
// Remove values one by one and assert that nodes indices stay valid.
|
|
|
|
while !crds.table.is_empty() {
|
|
|
|
let index = rng.gen_range(0, crds.table.len());
|
|
|
|
let key = crds.table.get_index(index).unwrap().0.clone();
|
|
|
|
crds.remove(&key);
|
|
|
|
check_crds_nodes(&crds);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-11-15 13:23:26 -08:00
|
|
|
#[test]
|
2019-11-20 11:25:18 -08:00
|
|
|
fn test_remove_staked() {
|
2020-10-23 07:17:37 -07:00
|
|
|
let thread_pool = ThreadPoolBuilder::new().build().unwrap();
|
2018-11-15 13:23:26 -08:00
|
|
|
let mut crds = Crds::default();
|
2019-11-03 10:07:51 -08:00
|
|
|
let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
|
2018-11-15 13:23:26 -08:00
|
|
|
assert_matches!(crds.insert(val.clone(), 1), Ok(_));
|
2019-11-20 11:25:18 -08:00
|
|
|
let mut set = HashMap::new();
|
2018-11-15 13:23:26 -08:00
|
|
|
|
2019-11-20 11:25:18 -08:00
|
|
|
//default has max timeout, but pubkey should still expire
|
|
|
|
set.insert(Pubkey::default(), std::u64::MAX);
|
|
|
|
set.insert(val.pubkey(), 1);
|
2020-10-23 07:17:37 -07:00
|
|
|
assert_eq!(
|
|
|
|
crds.find_old_labels(&thread_pool, 2, &set),
|
|
|
|
vec![val.label()]
|
|
|
|
);
|
2018-11-15 13:23:26 -08:00
|
|
|
crds.remove(&val.label());
|
2020-10-23 07:17:37 -07:00
|
|
|
assert!(crds.find_old_labels(&thread_pool, 2, &set).is_empty());
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
2019-11-20 11:25:18 -08:00
|
|
|
|
2018-11-15 13:23:26 -08:00
|
|
|
#[test]
|
2020-05-15 09:35:43 -07:00
|
|
|
#[allow(clippy::neg_cmp_op_on_partial_ord)]
|
2018-11-15 13:23:26 -08:00
|
|
|
fn test_equal() {
|
2019-11-03 10:07:51 -08:00
|
|
|
let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
|
2019-03-08 19:28:19 -08:00
|
|
|
let v1 = VersionedCrdsValue::new(1, val.clone());
|
|
|
|
let v2 = VersionedCrdsValue::new(1, val);
|
|
|
|
assert_eq!(v1, v2);
|
2018-11-15 13:23:26 -08:00
|
|
|
assert!(!(v1 != v2));
|
2020-05-15 09:35:43 -07:00
|
|
|
assert_eq!(v1.partial_cmp(&v2), Some(cmp::Ordering::Equal));
|
|
|
|
assert_eq!(v2.partial_cmp(&v1), Some(cmp::Ordering::Equal));
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
#[test]
|
2020-05-15 09:35:43 -07:00
|
|
|
#[allow(clippy::neg_cmp_op_on_partial_ord)]
|
2018-11-15 13:23:26 -08:00
|
|
|
fn test_hash_order() {
|
|
|
|
let v1 = VersionedCrdsValue::new(
|
|
|
|
1,
|
2019-11-03 10:07:51 -08:00
|
|
|
CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::default(),
|
|
|
|
0,
|
|
|
|
))),
|
2018-11-15 13:23:26 -08:00
|
|
|
);
|
2019-03-08 19:28:19 -08:00
|
|
|
let v2 = VersionedCrdsValue::new(1, {
|
2019-03-09 19:28:43 -08:00
|
|
|
let mut contact_info = ContactInfo::new_localhost(&Pubkey::default(), 0);
|
2019-03-08 19:28:19 -08:00
|
|
|
contact_info.rpc = socketaddr!("0.0.0.0:0");
|
2019-11-03 10:07:51 -08:00
|
|
|
CrdsValue::new_unsigned(CrdsData::ContactInfo(contact_info))
|
2019-03-08 19:28:19 -08:00
|
|
|
});
|
|
|
|
|
|
|
|
assert_eq!(v1.value.label(), v2.value.label());
|
|
|
|
assert_eq!(v1.value.wallclock(), v2.value.wallclock());
|
|
|
|
assert_ne!(v1.value_hash, v2.value_hash);
|
2018-11-15 13:23:26 -08:00
|
|
|
assert!(v1 != v2);
|
|
|
|
assert!(!(v1 == v2));
|
|
|
|
if v1 > v2 {
|
2019-03-08 19:28:19 -08:00
|
|
|
assert!(v1 > v2);
|
|
|
|
assert!(v2 < v1);
|
2020-05-15 09:35:43 -07:00
|
|
|
assert_eq!(v1.partial_cmp(&v2), Some(cmp::Ordering::Greater));
|
|
|
|
assert_eq!(v2.partial_cmp(&v1), Some(cmp::Ordering::Less));
|
2019-03-08 19:28:19 -08:00
|
|
|
} else if v2 > v1 {
|
|
|
|
assert!(v1 < v2);
|
|
|
|
assert!(v2 > v1);
|
2020-05-15 09:35:43 -07:00
|
|
|
assert_eq!(v1.partial_cmp(&v2), Some(cmp::Ordering::Less));
|
|
|
|
assert_eq!(v2.partial_cmp(&v1), Some(cmp::Ordering::Greater));
|
2018-11-15 13:23:26 -08:00
|
|
|
} else {
|
2019-03-08 19:28:19 -08:00
|
|
|
panic!("bad PartialOrd implementation?");
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
#[test]
|
2020-05-15 09:35:43 -07:00
|
|
|
#[allow(clippy::neg_cmp_op_on_partial_ord)]
|
2018-11-15 13:23:26 -08:00
|
|
|
fn test_wallclock_order() {
|
|
|
|
let v1 = VersionedCrdsValue::new(
|
|
|
|
1,
|
2019-11-03 10:07:51 -08:00
|
|
|
CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::default(),
|
|
|
|
1,
|
|
|
|
))),
|
2018-11-15 13:23:26 -08:00
|
|
|
);
|
|
|
|
let v2 = VersionedCrdsValue::new(
|
|
|
|
1,
|
2019-11-03 10:07:51 -08:00
|
|
|
CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
|
|
|
&Pubkey::default(),
|
|
|
|
0,
|
|
|
|
))),
|
2018-11-15 13:23:26 -08:00
|
|
|
);
|
2019-03-08 19:28:19 -08:00
|
|
|
assert_eq!(v1.value.label(), v2.value.label());
|
2018-11-15 13:23:26 -08:00
|
|
|
assert!(v1 > v2);
|
|
|
|
assert!(!(v1 < v2));
|
|
|
|
assert!(v1 != v2);
|
|
|
|
assert!(!(v1 == v2));
|
2020-05-15 09:35:43 -07:00
|
|
|
assert_eq!(v1.partial_cmp(&v2), Some(cmp::Ordering::Greater));
|
|
|
|
assert_eq!(v2.partial_cmp(&v1), Some(cmp::Ordering::Less));
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
#[test]
|
2020-05-15 09:35:43 -07:00
|
|
|
#[allow(clippy::neg_cmp_op_on_partial_ord)]
|
2018-11-15 13:23:26 -08:00
|
|
|
fn test_label_order() {
|
|
|
|
let v1 = VersionedCrdsValue::new(
|
|
|
|
1,
|
2019-11-03 10:07:51 -08:00
|
|
|
CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
2020-10-19 12:12:08 -07:00
|
|
|
&solana_sdk::pubkey::new_rand(),
|
2019-11-03 10:07:51 -08:00
|
|
|
0,
|
|
|
|
))),
|
2018-11-15 13:23:26 -08:00
|
|
|
);
|
|
|
|
let v2 = VersionedCrdsValue::new(
|
|
|
|
1,
|
2019-11-03 10:07:51 -08:00
|
|
|
CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
|
2020-10-19 12:12:08 -07:00
|
|
|
&solana_sdk::pubkey::new_rand(),
|
2019-11-03 10:07:51 -08:00
|
|
|
0,
|
|
|
|
))),
|
2018-11-15 13:23:26 -08:00
|
|
|
);
|
2019-03-08 19:28:19 -08:00
|
|
|
assert_ne!(v1, v2);
|
2018-11-15 13:23:26 -08:00
|
|
|
assert!(!(v1 == v2));
|
|
|
|
assert!(!(v1 < v2));
|
|
|
|
assert!(!(v1 > v2));
|
|
|
|
assert!(!(v2 < v1));
|
|
|
|
assert!(!(v2 > v1));
|
2020-05-15 09:35:43 -07:00
|
|
|
assert_eq!(v1.partial_cmp(&v2), None);
|
|
|
|
assert_eq!(v2.partial_cmp(&v1), None);
|
2018-11-15 13:23:26 -08:00
|
|
|
}
|
|
|
|
}
|