gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-21 05:44:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Deprecate futures ticker (#6630)

Browse Source 
* deprecate futures-ticker

* Merge branch 'unstable' of github.com:sigp/lighthouse into deprecate-futures-timer

* Merge branch 'unstable' into deprecate-futures-timer

* making the linter happy

* remove unrequired #[allow(unused_imports)]

* fixing minor issues

* merge commit

* minor fix

* clippy changes
This commit is contained in:
hopinheimer
2024-12-03 03:56:43 -05:00
committed by GitHub
parent 1fd86f8b59
commit fec502db9f
5 changed files with 38 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
beacon_node/lighthouse_network/gossipsub/src/behaviour.rs
View File

@@ -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<D = IdentityTransform, F = AllowAllSubscriptionFilter> {
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<D = IdentityTransform, F = AllowAllSubscriptionFilter> {
outbound_peers: HashSet<PeerId>,
/// 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<PeerId, usize>,
@@ -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