From 7b4dcebde74ac2d0b2a444fc2e9c3a641971f51e Mon Sep 17 00:00:00 2001 From: Age Manning Date: Fri, 11 Feb 2022 10:34:06 +1100 Subject: [PATCH] Revert back to wasm_timer for interval --- protocols/gossipsub/Cargo.toml | 1 + protocols/gossipsub/src/behaviour.rs | 7 +- protocols/gossipsub/src/interval.rs | 209 --------------------------- protocols/gossipsub/src/lib.rs | 1 - 4 files changed, 5 insertions(+), 213 deletions(-) delete mode 100644 protocols/gossipsub/src/interval.rs diff --git a/protocols/gossipsub/Cargo.toml b/protocols/gossipsub/Cargo.toml index 9ffff384946..81f600e92f6 100644 --- a/protocols/gossipsub/Cargo.toml +++ b/protocols/gossipsub/Cargo.toml @@ -33,6 +33,7 @@ instant = "0.1.11" serde = { version = "1", optional = true, features = ["derive"] } # Metrics dependencies prometheus-client = "0.15.0" +wasm-timer = "0.2.5" [dev-dependencies] async-std = "1.6.3" diff --git a/protocols/gossipsub/src/behaviour.rs b/protocols/gossipsub/src/behaviour.rs index c354936ef2f..588d5a03827 100644 --- a/protocols/gossipsub/src/behaviour.rs +++ b/protocols/gossipsub/src/behaviour.rs @@ -46,6 +46,7 @@ use libp2p_swarm::{ IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters, }; +use crate::backoff::BackoffStorage; use crate::config::{GossipsubConfig, ValidationMode}; use crate::error::{PublishError, SubscriptionError, ValidationError}; use crate::gossip_promises::GossipPromises; @@ -63,9 +64,9 @@ use crate::types::{ GossipsubSubscriptionAction, MessageAcceptance, MessageId, PeerInfo, RawGossipsubMessage, }; use crate::types::{GossipsubRpc, PeerConnections, PeerKind}; -use crate::{backoff::BackoffStorage, interval::Interval}; use crate::{rpc_proto, TopicScoreParams}; use std::{cmp::Ordering::Equal, fmt::Debug}; +use wasm_timer::Interval; #[cfg(test)] mod tests; @@ -439,8 +440,8 @@ where config.backoff_slack(), ), mcache: MessageCache::new(config.history_gossip(), config.history_length()), - heartbeat: Interval::new_initial( - config.heartbeat_initial_delay(), + heartbeat: Interval::new_at( + Instant::now() + config.heartbeat_initial_delay(), config.heartbeat_interval(), ), heartbeat_ticks: 0, diff --git a/protocols/gossipsub/src/interval.rs b/protocols/gossipsub/src/interval.rs deleted file mode 100644 index 3080f541a55..00000000000 --- a/protocols/gossipsub/src/interval.rs +++ /dev/null @@ -1,209 +0,0 @@ -// Copyright 2021 Oliver Wangler -// Copyright 2019 Pierre Krieger -// Copyright (c) 2019 Tokio Contributors -// -// Permission is hereby granted, free of charge, to any person obtaining a copy of -// this software and associated documentation files (the "Software"), to deal in -// the Software without restriction, including without limitation the rights to -// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of -// the Software, and to permit persons to whom the Software is furnished to do so, -// subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in all -// copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS -// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS -// OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, -// WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN -// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. -// -// Initial version copied from -// https://github.com/tomaka/wasm-timer/blob/8964804eff980dd3eb115b711c57e481ba541708/src/timer/interval.rs -// and adapted. -use std::{ - pin::Pin, - task::{Context, Poll}, - time::Duration, -}; - -use futures::prelude::*; -use futures_timer::Delay; -use instant::Instant; -use pin_project::pin_project; - -/// A stream representing notifications at fixed interval -/// -/// Intervals are created through the `Interval::new` or -/// `Interval::new_intial` methods indicating when a first notification -/// should be triggered and when it will be repeated. -/// -/// Note that intervals are not intended for high resolution timers, but rather -/// they will likely fire some granularity after the exact instant that they're -/// otherwise indicated to fire at. -#[pin_project] -#[derive(Debug)] -pub struct Interval { - #[pin] - delay: Delay, - interval: Duration, - fires_at: Instant, -} - -impl Interval { - /// Creates a new interval which will fire at `dur` time into the future, - /// and will repeat every `dur` interval after - /// - /// The returned object will be bound to the default timer for this thread. - /// The default timer will be spun up in a helper thread on first use. - pub fn new(dur: Duration) -> Interval { - Interval::new_initial(dur, dur) - } - - /// Creates a new interval which will fire the first time after the specified `initial_delay`, - /// and then will repeat every `dur` interval after. - /// - /// The returned object will be bound to the default timer for this thread. - /// The default timer will be spun up in a helper thread on first use. - pub fn new_initial(initial_delay: Duration, dur: Duration) -> Interval { - let fires_at = Instant::now() + initial_delay; - Interval { - delay: Delay::new(initial_delay), - interval: dur, - fires_at, - } - } -} - -impl Stream for Interval { - type Item = (); - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - if self.as_mut().project().delay.poll(cx).is_pending() { - return Poll::Pending; - } - let next = next_interval(self.fires_at, Instant::now(), self.interval); - self.delay.reset(next); - self.fires_at += next; - Poll::Ready(Some(())) - } -} - -/// Converts Duration object to raw nanoseconds if possible -/// -/// This is useful to divide intervals. -/// -/// While technically for large duration it's impossible to represent any -/// duration as nanoseconds, the largest duration we can represent is about -/// 427_000 years. Large enough for any interval we would use or calculate in -/// tokio. -fn duration_to_nanos(dur: Duration) -> Option { - let v = dur.as_secs().checked_mul(1_000_000_000)?; - v.checked_add(dur.subsec_nanos() as u64) -} - -fn next_interval(prev: Instant, now: Instant, interval: Duration) -> Duration { - let new = prev + interval; - if new > now { - interval - } else { - let spent_ns = - duration_to_nanos(now.duration_since(prev)).expect("interval should be expired"); - let interval_ns = - duration_to_nanos(interval).expect("interval is less that 427 thousand years"); - let mult = spent_ns / interval_ns + 1; - assert!( - mult < (1 << 32), - "can't skip more than 4 billion intervals of {:?} \ - (trying to skip {})", - interval, - mult - ); - interval * mult as u32 - } -} - -#[cfg(test)] -mod test { - use super::next_interval; - use std::time::{Duration, Instant}; - - struct Timeline(Instant); - - impl Timeline { - fn new() -> Timeline { - Timeline(Instant::now()) - } - fn at(&self, millis: u64) -> Instant { - self.0 + Duration::from_millis(millis) - } - fn at_ns(&self, sec: u64, nanos: u32) -> Instant { - self.0 + Duration::new(sec, nanos) - } - } - - fn dur(millis: u64) -> Duration { - Duration::from_millis(millis) - } - - // The math around Instant/Duration isn't 100% precise due to rounding - // errors - fn almost_eq(a: Instant, b: Instant) -> bool { - let diff = match a.cmp(&b) { - std::cmp::Ordering::Less => b - a, - std::cmp::Ordering::Equal => return true, - std::cmp::Ordering::Greater => a - b, - }; - - diff < Duration::from_millis(1) - } - - #[test] - fn norm_next() { - let tm = Timeline::new(); - assert!(almost_eq( - tm.at(1) + next_interval(tm.at(1), tm.at(2), dur(10)), - tm.at(11) - )); - assert!(almost_eq( - tm.at(7777) + next_interval(tm.at(7777), tm.at(7788), dur(100)), - tm.at(7877) - )); - assert!(almost_eq( - tm.at(1) + next_interval(tm.at(1), tm.at(1000), dur(2100)), - tm.at(2101) - )); - } - - #[test] - fn fast_forward() { - let tm = Timeline::new(); - - assert!(almost_eq( - tm.at(1) + next_interval(tm.at(1), tm.at(1000), dur(10)), - tm.at(1001) - )); - assert!(almost_eq( - tm.at(7777) + next_interval(tm.at(7777), tm.at(8888), dur(100)), - tm.at(8977) - )); - assert!(almost_eq( - tm.at(1) + next_interval(tm.at(1), tm.at(10000), dur(2100)), - tm.at(10501) - )); - } - - /// TODO: this test actually should be successful, but since we can't - /// multiply Duration on anything larger than u32 easily we decided - /// to allow it to fail for now - #[test] - #[should_panic(expected = "can't skip more than 4 billion intervals")] - fn large_skip() { - let tm = Timeline::new(); - assert_eq!( - tm.0 + next_interval(tm.at_ns(0, 1), tm.at_ns(25, 0), Duration::new(0, 2)), - tm.at_ns(25, 1) - ); - } -} diff --git a/protocols/gossipsub/src/lib.rs b/protocols/gossipsub/src/lib.rs index 4088a71f128..0168a0a2842 100644 --- a/protocols/gossipsub/src/lib.rs +++ b/protocols/gossipsub/src/lib.rs @@ -142,7 +142,6 @@ mod behaviour; mod config; mod gossip_promises; mod handler; -mod interval; mod mcache; pub mod metrics; mod peer_score;