diff --git a/Cargo.lock b/Cargo.lock index 8587f0d54fb..4f597eafd58 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1148,17 +1148,6 @@ dependencies = [ "cipher", ] -[[package]] -name = "cuckoofilter" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b810a8449931679f64cd7eef1bbd0fa315801b6d5d9cdc1ace2804d6529eee18" -dependencies = [ - "byteorder", - "fnv", - "rand 0.7.3", -] - [[package]] name = "curve25519-dalek" version = "4.1.3" @@ -1855,17 +1844,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "getrandom" -version = "0.1.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fc3cb4d91f53b50155bdcfd23f6a4c39ae1969c2ae85982b135750cccaf5fce" -dependencies = [ - "cfg-if", - "libc", - "wasi 0.9.0+wasi-snapshot-preview1", -] - [[package]] name = "getrandom" version = "0.2.15" @@ -3096,9 +3074,9 @@ version = "0.48.0" dependencies = [ "asynchronous-codec", "bytes", - "cuckoofilter", "fnv", "futures", + "hashlink", "libp2p-core", "libp2p-identity", "libp2p-swarm", @@ -5151,19 +5129,6 @@ dependencies = [ "ptr_meta", ] -[[package]] -name = "rand" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03" -dependencies = [ - "getrandom 0.1.16", - "libc", - "rand_chacha 0.2.2", - "rand_core 0.5.1", - "rand_hc", -] - [[package]] name = "rand" version = "0.8.6" @@ -5196,16 +5161,6 @@ dependencies = [ "rand_core 0.10.0", ] -[[package]] -name = "rand_chacha" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f4c8ed856279c9737206bf725bf36935d8666ead7aa69b52be55af369d193402" -dependencies = [ - "ppv-lite86", - "rand_core 0.5.1", -] - [[package]] name = "rand_chacha" version = "0.3.1" @@ -5226,15 +5181,6 @@ dependencies = [ "rand_core 0.9.3", ] -[[package]] -name = "rand_core" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19" -dependencies = [ - "getrandom 0.1.16", -] - [[package]] name = "rand_core" version = "0.6.4" @@ -5259,15 +5205,6 @@ version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c8d0fd677905edcbeedbf2edb6494d676f0e98d54d5cf9bda0b061cb8fb8aba" -[[package]] -name = "rand_hc" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c" -dependencies = [ - "rand_core 0.5.1", -] - [[package]] name = "rayon" version = "1.10.0" @@ -7124,12 +7061,6 @@ dependencies = [ "try-lock", ] -[[package]] -name = "wasi" -version = "0.9.0+wasi-snapshot-preview1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cccddf32554fecc6acb585f82a32a72e28b48f8c4c1883ddfeeeaa96f7d8e519" - [[package]] name = "wasi" version = "0.11.0+wasi-snapshot-preview1" diff --git a/protocols/floodsub/CHANGELOG.md b/protocols/floodsub/CHANGELOG.md index 68cdd4308a0..268e279d081 100644 --- a/protocols/floodsub/CHANGELOG.md +++ b/protocols/floodsub/CHANGELOG.md @@ -1,5 +1,8 @@ ## 0.48.0 +- Replace the duplicate-message cuckoo filter with a bounded cache, removing the `rand` 0.7.3 dependency. + See [issue 6419](https://github.com/libp2p/rust-libp2p/issues/6419). + - Raise MSRV to 1.88.0. See [PR 6273](https://github.com/libp2p/rust-libp2p/pull/6273). diff --git a/protocols/floodsub/Cargo.toml b/protocols/floodsub/Cargo.toml index f2196405ad0..3259e5e3012 100644 --- a/protocols/floodsub/Cargo.toml +++ b/protocols/floodsub/Cargo.toml @@ -12,10 +12,10 @@ categories = ["network-programming", "asynchronous"] [dependencies] asynchronous-codec = { workspace = true } -cuckoofilter = "0.5.0" -fnv = "1.0" bytes.workspace = true +fnv = "1.0" futures = { workspace = true } +hashlink = { workspace = true } libp2p-core = { workspace = true } libp2p-swarm = { workspace = true } libp2p-identity = { workspace = true } diff --git a/protocols/floodsub/src/layer.rs b/protocols/floodsub/src/layer.rs index bd991700f9a..73c44768724 100644 --- a/protocols/floodsub/src/layer.rs +++ b/protocols/floodsub/src/layer.rs @@ -23,13 +23,14 @@ use std::{ VecDeque, hash_map::{DefaultHasher, HashMap}, }, + hash::{Hash, Hasher}, iter, task::{Context, Poll}, }; use bytes::Bytes; -use cuckoofilter::{CuckooError, CuckooFilter}; use fnv::FnvHashSet; +use hashlink::LinkedHashSet; use libp2p_core::{Endpoint, Multiaddr, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ @@ -49,6 +50,42 @@ use crate::{ topic::Topic, }; +// Match the approximate maximum entry count of the previous cuckoo filter. +const RECEIVED_CACHE_CAPACITY: usize = 1 << 20; + +struct ReceivedCache { + message_hashes: LinkedHashSet, + capacity: usize, +} + +impl ReceivedCache { + fn new() -> Self { + Self::with_capacity(RECEIVED_CACHE_CAPACITY) + } + + fn with_capacity(capacity: usize) -> Self { + debug_assert!(capacity > 0); + + Self { + message_hashes: LinkedHashSet::new(), + capacity, + } + } + + /// Returns `true` if the value was not present in the cache. + fn insert(&mut self, value: &T) -> bool { + let mut hasher = DefaultHasher::new(); + value.hash(&mut hasher); + + let is_new = self.message_hashes.insert(hasher.finish()); + if self.message_hashes.len() > self.capacity { + self.message_hashes.pop_front(); + } + + is_new + } +} + #[deprecated = "Use `Behaviour` instead."] pub type Floodsub = Behaviour; @@ -73,7 +110,7 @@ pub struct Behaviour { // We keep track of the messages we received (in the format `hash(source ID, seq_no)`) so that // we don't dispatch the same message twice if we receive it twice on the network. - received: CuckooFilter, + received: ReceivedCache, } impl Behaviour { @@ -90,7 +127,7 @@ impl Behaviour { target_peers: FnvHashSet::default(), connected_peers: HashMap::new(), subscribed_topics: SmallVec::new(), - received: CuckooFilter::new(), + received: ReceivedCache::new(), } } @@ -235,13 +272,7 @@ impl Behaviour { .iter() .any(|t| message.topics.iter().any(|u| t == u)); if self_subscribed { - if let Err(e @ CuckooError::NotEnoughSpace) = self.received.add(&message) { - tracing::warn!( - "Message was added to 'received' Cuckoofilter but some \ - other message was removed as a consequence: {}", - e, - ); - } + self.received.insert(&message); if self.config.subscribe_local_messages { self.events .push_back(ToSwarm::GenerateEvent(Event::Message(message.clone()))); @@ -420,18 +451,8 @@ impl NetworkBehaviour for Behaviour { for message in event.messages { // Use `self.received` to skip the messages that we have already received in the past. - // Note that this can result in false positives. - match self.received.test_and_add(&message) { - Ok(true) => {} // Message was added. - Ok(false) => continue, // Message already existed. - Err(e @ CuckooError::NotEnoughSpace) => { - // Message added, but some other removed. - tracing::warn!( - "Message was added to 'received' Cuckoofilter but some \ - other message was removed as a consequence: {}", - e, - ); - } + if !self.received.insert(&message) { + continue; } // Add the message to be dispatched to the user. @@ -556,3 +577,22 @@ pub enum Event { topic: Topic, }, } + +#[cfg(test)] +mod tests { + use super::ReceivedCache; + + #[test] + fn received_cache_evicts_least_recently_seen_hash() { + let mut cache = ReceivedCache::with_capacity(2); + + assert!(cache.insert(&1)); + assert!(cache.insert(&2)); + assert!(!cache.insert(&1)); + + assert!(cache.insert(&3)); + assert_eq!(cache.message_hashes.len(), 2); + assert!(!cache.insert(&1)); + assert!(cache.insert(&2)); + } +}