diff --git a/Cargo.lock b/Cargo.lock index 1ddeecf7116..5cea2d2ec5d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3375,22 +3375,15 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988" -[[package]] -name = "futures-ticker" -version = "0.0.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9763058047f713632a52e916cc7f6a4b3fc6e9fc1ff8c5b1dc49e5a89041682e" -dependencies = [ - "futures", - "futures-timer", - "instant", -] - [[package]] name = "futures-timer" version = "3.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f288b0a4f20f9a56b5d1da57e2227c661b7b16168e2f72365f57b63326e29b24" +dependencies = [ + "gloo-timers", + "send_wrapper 0.4.0", +] [[package]] name = "futures-util" @@ -3506,6 +3499,18 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b" +[[package]] +name = "gloo-timers" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b995a66bb87bebce9a0f4a95aed01daca4872c050bfcb21653361c03bc35e5c" +dependencies = [ + "futures-channel", + "futures-core", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "gossipsub" version = "0.5.0" @@ -3518,7 +3523,6 @@ dependencies = [ "either", "fnv", "futures", - "futures-ticker", "futures-timer", "getrandom", "hashlink 0.9.1", @@ -7736,6 +7740,12 @@ dependencies = [ "pest", ] +[[package]] +name = "send_wrapper" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f638d531eccd6e23b980caf34876660d38e265409d8e99b397ab71eb3612fad0" + [[package]] name = "send_wrapper" version = "0.6.0" @@ -10349,7 +10359,7 @@ dependencies = [ "log", "pharos", "rustc_version 0.4.1", - "send_wrapper", + "send_wrapper 0.6.0", "thiserror 1.0.69", "wasm-bindgen", "wasm-bindgen-futures", diff --git a/beacon_node/lighthouse_network/gossipsub/Cargo.toml b/beacon_node/lighthouse_network/gossipsub/Cargo.toml index a01d60dae99..6cbe6d3a1cc 100644 --- a/beacon_node/lighthouse_network/gossipsub/Cargo.toml +++ b/beacon_node/lighthouse_network/gossipsub/Cargo.toml @@ -10,7 +10,7 @@ keywords = ["peer-to-peer", "libp2p", "networking"] categories = ["network-programming", "asynchronous"] [features] -wasm-bindgen = ["getrandom/js"] +wasm-bindgen = ["getrandom/js", "futures-timer/wasm-bindgen"] rsa = [] [dependencies] @@ -22,7 +22,6 @@ bytes = "1.5" either = "1.9" fnv = "1.0.7" futures = "0.3.30" -futures-ticker = "0.0.3" futures-timer = "3.0.2" getrandom = "0.2.12" hashlink.workspace = true diff --git a/beacon_node/lighthouse_network/gossipsub/src/behaviour.rs b/beacon_node/lighthouse_network/gossipsub/src/behaviour.rs index 5ead0c06a0a..aafd869beed 100644 --- a/beacon_node/lighthouse_network/gossipsub/src/behaviour.rs +++ b/beacon_node/lighthouse_network/gossipsub/src/behaviour.rs @@ -29,8 +29,7 @@ use std::{ time::Duration, }; -use futures::StreamExt; -use futures_ticker::Ticker; +use futures::FutureExt; use hashlink::LinkedHashMap; use prometheus_client::registry::Registry; use rand::{seq::SliceRandom, thread_rng}; @@ -74,6 +73,7 @@ use super::{ types::RpcOut, }; use super::{PublishError, SubscriptionError, TopicScoreParams, ValidationError}; +use futures_timer::Delay; use quick_protobuf::{MessageWrite, Writer}; use std::{cmp::Ordering::Equal, fmt::Debug}; @@ -301,7 +301,7 @@ pub struct Behaviour { mcache: MessageCache, /// Heartbeat interval stream. - heartbeat: Ticker, + heartbeat: Delay, /// Number of heartbeats since the beginning of time; this allows us to amortize some resource /// clean up -- eg backoff clean up. @@ -318,7 +318,7 @@ pub struct Behaviour { outbound_peers: HashSet, /// Stores optional peer score data together with thresholds and decay interval. - peer_score: Option<(PeerScore, PeerScoreThresholds, Ticker)>, + peer_score: Option<(PeerScore, PeerScoreThresholds, Delay)>, /// Counts the number of `IHAVE` received from each peer since the last heartbeat. count_received_ihave: HashMap, @@ -466,10 +466,7 @@ where config.backoff_slack(), ), mcache: MessageCache::new(config.history_gossip(), config.history_length()), - heartbeat: Ticker::new_with_next( - config.heartbeat_interval(), - config.heartbeat_initial_delay(), - ), + heartbeat: Delay::new(config.heartbeat_interval() + config.heartbeat_initial_delay()), heartbeat_ticks: 0, px_peers: HashSet::new(), outbound_peers: HashSet::new(), @@ -938,7 +935,7 @@ where return Err("Peer score set twice".into()); } - let interval = Ticker::new(params.decay_interval); + let interval = Delay::new(params.decay_interval); let peer_score = PeerScore::new_with_message_delivery_time_callback(params, callback); self.peer_score = Some((peer_score, threshold, interval)); Ok(()) @@ -1208,7 +1205,7 @@ where } fn score_below_threshold_from_scores( - peer_score: &Option<(PeerScore, PeerScoreThresholds, Ticker)>, + peer_score: &Option<(PeerScore, PeerScoreThresholds, Delay)>, peer_id: &PeerId, threshold: impl Fn(&PeerScoreThresholds) -> f64, ) -> (bool, f64) { @@ -3427,14 +3424,16 @@ where } // update scores - if let Some((peer_score, _, interval)) = &mut self.peer_score { - while let Poll::Ready(Some(_)) = interval.poll_next_unpin(cx) { + if let Some((peer_score, _, delay)) = &mut self.peer_score { + if delay.poll_unpin(cx).is_ready() { peer_score.refresh_scores(); + delay.reset(peer_score.params.decay_interval); } } - while let Poll::Ready(Some(_)) = self.heartbeat.poll_next_unpin(cx) { + if self.heartbeat.poll_unpin(cx).is_ready() { self.heartbeat(); + self.heartbeat.reset(self.config.heartbeat_interval()); } Poll::Pending diff --git a/beacon_node/lighthouse_network/gossipsub/src/behaviour/tests.rs b/beacon_node/lighthouse_network/gossipsub/src/behaviour/tests.rs index 713fe1f2668..90b8fe43fb5 100644 --- a/beacon_node/lighthouse_network/gossipsub/src/behaviour/tests.rs +++ b/beacon_node/lighthouse_network/gossipsub/src/behaviour/tests.rs @@ -25,6 +25,7 @@ use crate::subscription_filter::WhitelistSubscriptionFilter; use crate::types::RpcReceiver; use crate::{config::ConfigBuilder, types::Rpc, IdentTopic as Topic}; use byteorder::{BigEndian, ByteOrder}; +use futures::StreamExt; use libp2p::core::ConnectedPoint; use rand::Rng; use std::net::Ipv4Addr; diff --git a/beacon_node/lighthouse_network/gossipsub/src/peer_score.rs b/beacon_node/lighthouse_network/gossipsub/src/peer_score.rs index fa02f06f69d..ec6fe7bdb6e 100644 --- a/beacon_node/lighthouse_network/gossipsub/src/peer_score.rs +++ b/beacon_node/lighthouse_network/gossipsub/src/peer_score.rs @@ -44,7 +44,7 @@ mod tests; const TIME_CACHE_DURATION: u64 = 120; pub(crate) struct PeerScore { - params: PeerScoreParams, + pub(crate) params: PeerScoreParams, /// The score parameters. peer_stats: HashMap, /// Tracking peers per IP.