diff --git a/gossip/src/cluster_info.rs b/gossip/src/cluster_info.rs
index ded61d664c0..9f76919ee40 100644
--- a/gossip/src/cluster_info.rs
+++ b/gossip/src/cluster_info.rs
@@ -1844,20 +1844,16 @@ impl ClusterInfo {
) -> (usize, usize, usize) {
let len = crds_values.len();
let mut pull_stats = ProcessPullStats::default();
- let (filtered_pulls, filtered_pulls_expired_timeout, failed_inserts) = {
+ let (filtered_pulls, filtered_pulls_expired_timeout) = {
let _st = ScopedTimer::from(&self.stats.filter_pull_response);
self.gossip
.filter_pull_responses(timeouts, crds_values, timestamp(), &mut pull_stats)
};
- if !filtered_pulls.is_empty()
- || !filtered_pulls_expired_timeout.is_empty()
- || !failed_inserts.is_empty()
- {
+ if !filtered_pulls.is_empty() || !filtered_pulls_expired_timeout.is_empty() {
let _st = ScopedTimer::from(&self.stats.process_pull_response);
self.gossip.process_pull_responses(
filtered_pulls,
filtered_pulls_expired_timeout,
- failed_inserts,
timestamp(),
&mut pull_stats,
);
diff --git a/gossip/src/crds.rs b/gossip/src/crds.rs
index 32b062661d6..c551a9e2a25 100644
--- a/gossip/src/crds.rs
+++ b/gossip/src/crds.rs
@@ -35,6 +35,7 @@ use {
crds_gossip_pull::CrdsTimeouts,
crds_shards::CrdsShards,
crds_value::{CrdsValue, CrdsValueLabel},
+ recent_hashes::RecentHashes,
},
assert_matches::debug_assert_matches,
indexmap::{
@@ -49,13 +50,14 @@ use {
solana_pubkey::Pubkey,
std::{
cmp::Ordering,
- collections::{BTreeMap, HashMap, HashSet, VecDeque, hash_map},
+ collections::{BTreeMap, HashMap, HashSet, hash_map},
ops::{Bound, Index, IndexMut},
sync::Mutex,
},
};
pub(crate) const CRDS_SHARDS_BITS: u32 = 12;
+const MAX_PURGED_HASHES: usize = crate::crds_gossip_pull::MIN_NUM_BLOOM_ITEMS;
// Number of vote slots to track in an lru-cache for metrics.
const VOTE_SLOTS_METRICS_CAP: usize = 100;
// Required number of leading zero bits for crds signature to get reported to influx
@@ -82,7 +84,7 @@ pub struct Crds {
// Indices of all entries keyed by insert order.
entries: BTreeMap,
// Hash of recently purged values.
- purged: VecDeque<(Hash, u64 /*timestamp*/)>,
+ purged: RecentHashes,
stats: Mutex,
// Optional channel that receives a snapshot of every accepted contact
// info update. When `None` (the default), no work is done on the hot
@@ -183,7 +185,7 @@ impl Default for Crds {
duplicate_shreds: BTreeMap::default(),
records: HashMap::default(),
entries: BTreeMap::default(),
- purged: VecDeque::default(),
+ purged: RecentHashes::new(MAX_PURGED_HASHES),
stats: Mutex::::default(),
contact_info_sender: None,
}
@@ -333,7 +335,7 @@ impl Crds {
// does not need to be updated.
debug_assert_eq!(entry.get().value.pubkey(), pubkey);
self.cursor.consume(value.ordinal);
- self.purged.push_back((*entry.get().value.hash(), now));
+ self.purged.insert(*entry.get().value.hash(), now);
entry.insert(value);
Ok(())
}
@@ -347,7 +349,7 @@ impl Crds {
// Identify if the message is outdated (as opposed to
// duplicate) by comparing value hashes.
if entry.get().value.hash() != value.value.hash() {
- self.purged.push_back((*value.value.hash(), now));
+ self.purged.insert(*value.value.hash(), now);
Err(CrdsError::InsertFailed)
} else if matches!(route, GossipRoute::PushMessage(_)) {
let entry = entry.get_mut();
@@ -488,17 +490,12 @@ impl Crds {
}
pub(crate) fn purged(&self) -> impl IndexedParallelIterator- + '_ {
- self.purged.par_iter().map(|(hash, _)| *hash)
+ self.purged.par_iter()
}
/// Drops purged value hashes with timestamp less than the given one.
pub(crate) fn trim_purged(&mut self, timestamp: u64) {
- let count = self
- .purged
- .iter()
- .take_while(|(_, ts)| *ts < timestamp)
- .count();
- self.purged.drain(..count);
+ self.purged.purge(timestamp);
}
/// Returns all crds values which the first 'mask_bits'
@@ -593,7 +590,7 @@ impl Crds {
let Some((index, _ /*label*/, value)) = self.table.swap_remove_full(key) else {
return;
};
- self.purged.push_back((*value.value.hash(), now));
+ self.purged.insert(*value.value.hash(), now);
self.shards.remove(index, &value);
match value.value.data() {
CrdsData::ContactInfo(node) => {
diff --git a/gossip/src/crds_gossip.rs b/gossip/src/crds_gossip.rs
index ef8e753bd4d..1d99af1b41b 100644
--- a/gossip/src/crds_gossip.rs
+++ b/gossip/src/crds_gossip.rs
@@ -21,7 +21,6 @@ use {
rand::{CryptoRng, Rng},
rayon::ThreadPool,
solana_clock::Slot,
- solana_hash::Hash,
solana_keypair::{Address, Keypair},
solana_ledger::shred::Shred,
solana_net_utils::SocketAddrSpace,
@@ -258,7 +257,6 @@ impl CrdsGossip {
) -> (
Vec, // valid responses.
Vec, // responses with expired timestamps.
- Vec, // hash of outdated values.
) {
self.pull
.filter_pull_responses(&self.crds, timeouts, response, now, process_pull_stats)
@@ -269,7 +267,6 @@ impl CrdsGossip {
&self,
responses: Vec,
responses_expired_timeout: Vec,
- failed_inserts: Vec,
now: u64,
process_pull_stats: &mut ProcessPullStats,
) {
@@ -277,7 +274,6 @@ impl CrdsGossip {
&self.crds,
responses,
responses_expired_timeout,
- failed_inserts,
now,
process_pull_stats,
);
diff --git a/gossip/src/crds_gossip_pull.rs b/gossip/src/crds_gossip_pull.rs
index 0a1b0b54493..6f4a2b96208 100644
--- a/gossip/src/crds_gossip_pull.rs
+++ b/gossip/src/crds_gossip_pull.rs
@@ -20,6 +20,7 @@ use {
crds_gossip_error::CrdsGossipError,
crds_value::CrdsValue,
protocol::{Ping, PingCache},
+ recent_hashes::RecentHashes,
},
itertools::Itertools,
rand::{
@@ -37,9 +38,9 @@ use {
solana_pubkey::Pubkey,
solana_signer::Signer,
std::{
- collections::{HashMap, HashSet, VecDeque},
+ collections::{HashMap, HashSet},
convert::TryInto,
- iter::{repeat, repeat_with},
+ iter::repeat_with,
net::SocketAddr,
ops::Index,
sync::{
@@ -66,6 +67,7 @@ pub struct CrdsFilter {
}
pub(crate) const MIN_NUM_BLOOM_ITEMS: usize = 65_536;
+const MAX_FAILED_INSERTS: usize = MIN_NUM_BLOOM_ITEMS;
// Loosest mask_bits floor accepted for incoming pull requests.
// `PACKET_DATA_SIZE` avoids rejecting honest smaller bloom filters.
@@ -258,7 +260,7 @@ pub struct CrdsGossipPull {
// inserted in crds table; Preserved to stop the sender to send back the
// same outdated payload again by adding them to the filter for the next
// pull request.
- failed_inserts: RwLock>,
+ failed_inserts: RwLock,
pub crds_timeout: u64,
pub num_pulls: AtomicUsize,
}
@@ -266,7 +268,7 @@ pub struct CrdsGossipPull {
impl Default for CrdsGossipPull {
fn default() -> Self {
Self {
- failed_inserts: RwLock::default(),
+ failed_inserts: RwLock::new(RecentHashes::new(MAX_FAILED_INSERTS)),
crds_timeout: CRDS_GOSSIP_PULL_CRDS_TIMEOUT_MS,
num_pulls: AtomicUsize::default(),
}
@@ -359,11 +361,10 @@ impl CrdsGossipPull {
}
// Checks if responses should be inserted and
- // returns those responses converted to VersionedCrdsValue
- // Separated in three vecs as:
+ // returns those responses converted to VersionedCrdsValue.
+ // Separated in two vecs as:
// .0 => responses that update the owner timestamp
// .1 => responses that do not update the owner timestamp
- // .2 => hash value of outdated values which will fail to insert.
pub(crate) fn filter_pull_responses(
&self,
crds: &RwLock,
@@ -371,11 +372,20 @@ impl CrdsGossipPull {
responses: Vec,
now: u64,
stats: &mut ProcessPullStats,
- ) -> (Vec, Vec, Vec) {
+ ) -> (Vec, Vec) {
let mut active_values = vec![];
let mut expired_values = vec![];
let crds = crds.read().unwrap();
- let upsert = |response: CrdsValue| {
+ let mut failed_inserts = None;
+ let mut cache_failed_insert = |hash| {
+ let failed_inserts = failed_inserts.get_or_insert_with(|| {
+ let mut failed_inserts = self.failed_inserts.write().unwrap();
+ failed_inserts.purge(now.saturating_sub(FAILED_INSERTS_RETENTION_MS));
+ failed_inserts
+ });
+ failed_inserts.insert(hash, now);
+ };
+ for response in responses {
let owner = response.label().pubkey();
// Check if the crds value is older than the msg_timeout
let timeout = timeouts[&owner];
@@ -383,26 +393,21 @@ impl CrdsGossipPull {
// owner exists in the table. If it doesn't, that implies that this
// value can be discarded
if !crds.upserts(&response) {
- Some(response)
+ stats.failed_insert += 1;
+ cache_failed_insert(*response.hash());
} else if now <= response.wallclock().saturating_add(timeout) {
active_values.push(response);
- None
} else if crds.get::<&ContactInfo>(owner).is_some() {
// Silently insert this old value without bumping record
// timestamps
expired_values.push(response);
- None
} else {
stats.failed_timeout += 1;
- Some(response)
+ stats.failed_insert += 1;
+ cache_failed_insert(*response.hash());
}
- };
- let failed_inserts = responses
- .into_iter()
- .filter_map(upsert)
- .map(|resp| *resp.hash())
- .collect();
- (active_values, expired_values, failed_inserts)
+ }
+ (active_values, expired_values)
}
/// Process a vec of pull responses
@@ -411,7 +416,6 @@ impl CrdsGossipPull {
crds: &RwLock,
responses: Vec,
responses_expired_timeout: Vec,
- failed_inserts: Vec,
now: u64,
stats: &mut ProcessPullStats,
) {
@@ -433,23 +437,13 @@ impl CrdsGossipPull {
for owner in owners {
crds.update_record_timestamp(&owner, now);
}
- drop(crds);
- stats.failed_insert += failed_inserts.len();
- self.purge_failed_inserts(now);
- let failed_inserts = failed_inserts.into_iter().zip(repeat(now));
- self.failed_inserts.write().unwrap().extend(failed_inserts);
}
pub(crate) fn purge_failed_inserts(&self, now: u64) {
- if FAILED_INSERTS_RETENTION_MS < now {
- let cutoff = now - FAILED_INSERTS_RETENTION_MS;
- let mut failed_inserts = self.failed_inserts.write().unwrap();
- let outdated = failed_inserts
- .iter()
- .take_while(|(_, ts)| *ts < cutoff)
- .count();
- failed_inserts.drain(..outdated);
- }
+ self.failed_inserts
+ .write()
+ .unwrap()
+ .purge(now.saturating_sub(FAILED_INSERTS_RETENTION_MS));
}
pub(crate) fn failed_inserts_size(&self) -> usize {
@@ -465,9 +459,8 @@ impl CrdsGossipPull {
bloom_size: usize,
) -> Vec {
const PAR_MIN_LENGTH: usize = 512;
- let failed_inserts = self.failed_inserts.read().unwrap();
- // crds should be locked last after self.failed_inserts.
let crds = crds.read().unwrap();
+ let failed_inserts = self.failed_inserts.read().unwrap();
let num_items = crds.len() + crds.num_purged() + failed_inserts.len();
let num_items = MIN_NUM_BLOOM_ITEMS.max(num_items);
let filters = CrdsFilterSet::new(&mut rand::rng(), num_items, bloom_size);
@@ -476,16 +469,11 @@ impl CrdsGossipPull {
.with_min_len(PAR_MIN_LENGTH)
.map(|v| *v.value.hash())
.chain(crds.purged().with_min_len(PAR_MIN_LENGTH))
- .chain(
- failed_inserts
- .par_iter()
- .with_min_len(PAR_MIN_LENGTH)
- .map(|(v, _)| *v),
- )
+ .chain(failed_inserts.par_iter().with_min_len(PAR_MIN_LENGTH))
.for_each(|v| filters.add(v));
});
- drop(crds);
drop(failed_inserts);
+ drop(crds);
filters.into()
}
@@ -1191,17 +1179,15 @@ pub(crate) mod tests {
let stakes = HashMap::new();
let timeouts = node.make_timeouts(node_pubkey, &stakes, Duration::default());
let mut stats = ProcessPullStats::default();
- let (responses, responses_expired_timeout, failed_inserts) =
+ let (responses, responses_expired_timeout) =
node.filter_pull_responses(&node_crds, &timeouts, vec![new.clone()], 1, &mut stats);
assert_eq!(responses, vec![new.clone()]);
assert!(responses_expired_timeout.is_empty());
- assert!(failed_inserts.is_empty());
node.process_pull_responses(
&node_crds,
responses,
responses_expired_timeout,
- failed_inserts,
1,
&mut stats,
);
@@ -1214,6 +1200,38 @@ pub(crate) mod tests {
assert_eq!(entry.local_timestamp, 1);
}
+ #[test]
+ fn test_failed_inserts_deduplication_and_expiration() {
+ let keypair = Keypair::new();
+ let current = CrdsValue::new_unsigned(CrdsData::from(ContactInfo::new_localhost(
+ &keypair.pubkey(),
+ 2,
+ )));
+ let duplicate = current.clone();
+ let mut crds = Crds::default();
+ crds.insert(current, 2, GossipRoute::LocalMessage).unwrap();
+ let crds = RwLock::new(crds);
+ let node = CrdsGossipPull::default();
+ let stakes = HashMap::new();
+ let timeouts = node.make_timeouts(Pubkey::new_unique(), &stakes, Duration::default());
+ let mut stats = ProcessPullStats::default();
+
+ let (responses, expired) = node.filter_pull_responses(
+ &crds,
+ &timeouts,
+ vec![duplicate.clone(), duplicate],
+ 2,
+ &mut stats,
+ );
+ assert!(responses.is_empty());
+ assert!(expired.is_empty());
+ assert_eq!(stats.failed_insert, 2);
+ assert_eq!(node.failed_inserts_size(), 1);
+
+ node.purge_failed_inserts(FAILED_INSERTS_RETENTION_MS + 3);
+ assert_eq!(node.failed_inserts_size(), 0);
+ }
+
#[test]
fn test_gossip_purge() {
let thread_pool = ThreadPoolBuilder::new().build().unwrap();
diff --git a/gossip/src/lib.rs b/gossip/src/lib.rs
index 62c66c17727..ca4f6bd605d 100644
--- a/gossip/src/lib.rs
+++ b/gossip/src/lib.rs
@@ -30,6 +30,7 @@ pub mod ping_pong;
mod protocol;
mod push_active_set;
mod received_cache;
+mod recent_hashes;
pub mod restart_crds_values;
mod sigverify_cache;
pub mod weighted_shuffle;
diff --git a/gossip/src/recent_hashes.rs b/gossip/src/recent_hashes.rs
new file mode 100644
index 00000000000..0f91df26df4
--- /dev/null
+++ b/gossip/src/recent_hashes.rs
@@ -0,0 +1,121 @@
+use {
+ rayon::prelude::*,
+ solana_hash::Hash,
+ std::collections::{HashSet, VecDeque},
+};
+
+/// A bounded set of hashes ordered by insertion time.
+///
+/// Duplicate inserts do not refresh an entry. Once full, a new hash evicts
+/// the oldest one. Removing entries retains the backing allocations so that
+/// steady-state churn does not repeatedly allocate and deallocate memory.
+pub(crate) struct RecentHashes {
+ entries: VecDeque<(Hash, /*timestamp:*/ u64)>,
+ members: HashSet,
+ capacity: usize,
+}
+
+impl RecentHashes {
+ pub(crate) fn new(capacity: usize) -> Self {
+ assert_ne!(capacity, 0);
+ Self {
+ entries: VecDeque::new(),
+ members: HashSet::new(),
+ capacity,
+ }
+ }
+
+ /// Inserts a hash if it is not already present.
+ pub(crate) fn insert(&mut self, hash: Hash, timestamp: u64) {
+ if self.members.contains(&hash) {
+ return;
+ }
+ if self.entries.len() == self.capacity {
+ let (hash, _) = self.entries.pop_front().unwrap();
+ assert!(self.members.remove(&hash));
+ }
+ self.entries.push_back((hash, timestamp));
+ assert!(self.members.insert(hash));
+ }
+
+ /// Removes hashes inserted before `cutoff`.
+ pub(crate) fn purge(&mut self, cutoff: u64) {
+ while self
+ .entries
+ .front()
+ .is_some_and(|(_, timestamp)| *timestamp < cutoff)
+ {
+ let (hash, _) = self.entries.pop_front().unwrap();
+ assert!(self.members.remove(&hash));
+ }
+ }
+
+ pub(crate) fn len(&self) -> usize {
+ debug_assert_eq!(self.entries.len(), self.members.len());
+ self.entries.len()
+ }
+
+ pub(crate) fn par_iter(&self) -> impl IndexedParallelIterator
- + '_ {
+ self.entries.par_iter().map(|(hash, _)| *hash)
+ }
+
+ #[cfg(test)]
+ pub(crate) fn is_empty(&self) -> bool {
+ self.entries.is_empty()
+ }
+
+ #[cfg(test)]
+ pub(crate) fn back(&self) -> Option<&(Hash, u64)> {
+ self.entries.back()
+ }
+
+ #[cfg(test)]
+ pub(crate) fn iter(&self) -> impl Iterator
- {
+ self.entries.iter()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_recent_hashes_capacity_and_deduplication() {
+ let hashes: Vec<_> = (0..4)
+ .map(|index| Hash::new_from_array([index; 32]))
+ .collect();
+ let mut recent = RecentHashes::new(3);
+
+ recent.insert(hashes[0], 0);
+ recent.insert(hashes[1], 1);
+ recent.insert(hashes[0], 2);
+ assert_eq!(recent.len(), 2);
+
+ recent.insert(hashes[2], 3);
+ recent.insert(hashes[3], 4);
+ assert_eq!(recent.len(), 3);
+ assert!(!recent.members.contains(&hashes[0]));
+ assert!(
+ recent
+ .members
+ .is_superset(&HashSet::from([hashes[1], hashes[2], hashes[3],]))
+ );
+ }
+
+ #[test]
+ fn test_recent_hashes_purge() {
+ let hashes: Vec<_> = (0..3)
+ .map(|index| Hash::new_from_array([index; 32]))
+ .collect();
+ let mut recent = RecentHashes::new(3);
+ for (timestamp, hash) in hashes.iter().copied().enumerate() {
+ recent.insert(hash, timestamp as u64);
+ }
+
+ // A duplicate does not refresh the original insertion timestamp.
+ recent.insert(hashes[0], 10);
+ recent.purge(2);
+ assert_eq!(recent.len(), 1);
+ assert!(recent.members.contains(&hashes[2]));
+ }
+}
diff --git a/gossip/tests/crds_gossip.rs b/gossip/tests/crds_gossip.rs
index 36d9f83e352..909fc401519 100644
--- a/gossip/tests/crds_gossip.rs
+++ b/gossip/tests/crds_gossip.rs
@@ -614,16 +614,11 @@ fn network_run_pull(
Duration::from_secs(48 * 3600), // epoch_duration
&stakes,
);
- let (vers, vers_expired_timeout, failed_inserts) = node
+ let (vers, vers_expired_timeout) = node
.gossip
.filter_pull_responses(&timeouts, rsp, now, &mut stats);
- node.gossip.process_pull_responses(
- vers,
- vers_expired_timeout,
- failed_inserts,
- now,
- &mut stats,
- );
+ node.gossip
+ .process_pull_responses(vers, vers_expired_timeout, now, &mut stats);
overhead += stats.failed_insert;
overhead += stats.failed_timeout;
}