Add own abstraction layer over tokio::time::Interval

Cargo.toml

@@ -46,6 +46,7 @@ members = [
   "test-log",
   "test-log/proc-macro",
   "test-utils",
+  "time",
   "tools",
   "transaction-builder",
   "utils",
@@ -148,7 +149,6 @@ opt-level = 2
 ark-ec = { git = "https://github.com/paberr/algebra", branch = "pb/0.4" }
 ark-ff = { git = "https://github.com/paberr/algebra", branch = "pb/0.4" }
 ark-r1cs-std = { git = "https://github.com/paberr/r1cs-std", branch = "pb/fix-pedersen" }
-wasm-timer = { git = "https://github.com/sisou/wasm-timer.git" }
 
 [workspace.package]
 version = "0.1.0"
@@ -214,6 +214,7 @@ nimiq-tendermint = { path = "tendermint", default-features = false }
 nimiq-test-log = { path = "test-log", default-features = false }
 nimiq-test-log-proc-macro = { path = "test-log/proc-macro", default-features = false }
 nimiq-test-utils = { path = "test-utils", default-features = false }
+nimiq-time = { path = "time", default-features = false }
 nimiq-transaction = { path = "primitives/transaction", default-features = false }
 nimiq-transaction-builder = { path = "transaction-builder", default-features = false }
 nimiq-trie = { path = "primitives/trie", default-features = false }

client/src/main.rs

@@ -3,11 +3,10 @@ use std::time::Duration;
 use log::info;
 use nimiq::prover::prover_main;
 pub use nimiq::{
-    client::{Client, Consensus},
+    client::Client,
     config::{command_line::CommandLine, config::ClientConfig, config_file::ConfigFile},
     error::Error,
     extras::{
-        deadlock::initialize_deadlock_detection,
         logging::{initialize_logging, log_error_cause_chain},
         metrics_server::NimiqTaskMonitor,
         panic::initialize_panic_reporting,

consensus/Cargo.toml

@@ -32,7 +32,6 @@ serde = "1.0"
 thiserror = "1.0"
 tokio = { version = "1.37", features = ["rt", "sync", "time"] }
 tokio-stream = { version = "0.1", features = ["sync"] }
-wasm-timer = "0.2"
 
 nimiq-account = { workspace = true, default-features = false }
 nimiq-block = { workspace = true }
@@ -47,6 +46,7 @@ nimiq-mmr = { workspace = true }
 nimiq-network-interface = { workspace = true }
 nimiq-primitives = { workspace = true, features = ["policy", "trie"] }
 nimiq-serde = { workspace = true }
+nimiq-time = { workspace = true }
 nimiq-transaction = { workspace = true }
 nimiq-utils = { workspace = true, features = [
     "math",

consensus/src/consensus/mod.rs

@@ -25,8 +25,6 @@ use tokio::sync::{
     },
     oneshot::{error::RecvError, Sender as OneshotSender},
 };
-#[cfg(not(target_family = "wasm"))]
-use tokio::time::{sleep, Sleep};
 use tokio_stream::wrappers::BroadcastStream;
 
 use self::consensus_proxy::ConsensusProxy;
@@ -128,10 +126,6 @@ pub struct Consensus<N: Network> {
 
     pub sync: SyncerProxy<N>,
 
-    /// A Delay which exists purely for the waker on its poll to reactivate the task running Consensus::poll
-    /// FIXME Remove this
-    #[cfg(not(target_family = "wasm"))]
-    next_execution_timer: Option<Pin<Box<Sleep>>>,
     events: BroadcastSender<ConsensusEvent>,
     established_flag: Arc<AtomicBool>,
     head_requests: Option<HeadRequests<N>>,
@@ -168,13 +162,6 @@ impl<N: Network> Consensus<N> {
     /// established state and to advance the chain.
     const HEAD_REQUESTS_TIMEOUT: Duration = Duration::from_secs(5);
 
-    /// Timeout after which the consensus is polled after it ran last
-    ///
-    /// TODO: Set appropriate duration
-    /// FIXME Remove this
-    #[cfg(not(target_family = "wasm"))]
-    const CONSENSUS_POLL_TIMER: Duration = Duration::from_secs(1);
-
     pub fn from_network(
         blockchain: BlockchainProxy,
         network: Arc<N>,
@@ -210,16 +197,11 @@ impl<N: Network> Consensus<N> {
 
         let established_flag = Arc::new(AtomicBool::new(false));
 
-        #[cfg(not(target_family = "wasm"))]
-        let timer = Box::pin(sleep(Self::CONSENSUS_POLL_TIMER));
-
         Consensus {
             blockchain,
             network,
             sync: syncer,
             events: tx,
-            #[cfg(not(target_family = "wasm"))]
-            next_execution_timer: Some(timer),
             established_flag,
             head_requests: None,
             head_requests_time: None,
@@ -428,7 +410,7 @@ impl<N: Network> Future for Consensus<N> {
     type Output = ();
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        // 1. Poll and advance block queue
+        // Poll and advance block queue
         while let Poll::Ready(Some(event)) = self.sync.poll_next_unpin(cx) {
             match event {
                 LiveSyncPushEvent::AcceptedAnnouncedBlock(_) => {
@@ -473,7 +455,7 @@ impl<N: Network> Future for Consensus<N> {
             self.events.send(event).ok();
         }
 
-        // 2. Poll any head requests if active.
+        // Poll any head requests if active.
         if let Some(ref mut head_requests) = self.head_requests {
             if let Poll::Ready(mut result) = head_requests.poll_unpin(cx) {
                 // Reset head requests.
@@ -491,27 +473,14 @@ impl<N: Network> Future for Consensus<N> {
             }
         }
 
-        // 3. Check if a ConsensusRequest was received
+        // Check if a ConsensusRequest was received
         while let Poll::Ready(Some(request)) = self.requests.1.poll_recv(cx) {
             match request {
                 ConsensusRequest::ResolveBlock(request) => self.resolve_block(request),
             }
         }
 
-        // 4. Update timer and poll it so the task gets woken when the timer runs out (at the latest)
-        // The timer itself running out (producing an Instant) is of no interest to the execution. This poll method
-        // was potentially awoken by the delays waker, but even then all there is to do is set up a new timer such
-        // that it will wake this task again after another time frame has elapsed. No interval was used as that
-        // would periodically wake the task even though it might have just executed
-        #[cfg(not(target_family = "wasm"))]
-        {
-            let mut timer = Box::pin(sleep(Self::CONSENSUS_POLL_TIMER));
-            // If the sleep wasn't pending anymore, it didn't register us with the waker, but we need that.
-            assert!(timer.poll_unpin(cx) == Poll::Pending);
-            self.next_execution_timer = Some(timer);
-        }
-
-        // 5. Advance consensus and catch-up through head requests.
+        // Advance consensus and catch-up through head requests.
         self.request_heads();
 
         Poll::Pending

consensus/src/sync/live/diff_queue/diff_request_component.rs

@@ -3,8 +3,9 @@ use std::{sync::Arc, time::Duration};
 use futures::future::BoxFuture;
 use nimiq_network_interface::network::{Network, PubsubId};
 use nimiq_primitives::{trie::trie_diff::TrieDiff, TreeProof};
+use nimiq_time::sleep;
 use parking_lot::RwLock;
-use tokio::{sync::Semaphore, time};
+use tokio::sync::Semaphore;
 
 use super::{RequestTrieDiff, ResponseTrieDiff};
 use crate::sync::{
@@ -72,7 +73,7 @@ impl<N: Network> DiffRequestComponent<N> {
                         Some(peer_id) => peer_id,
                         None => {
                             error!("couldn't fetch diff: no peers");
-                            time::sleep(Duration::from_secs(5)).await;
+                            sleep(Duration::from_secs(5)).await;
                             continue;
                         }
                     };
@@ -112,7 +113,7 @@ impl<N: Network> DiffRequestComponent<N> {
                     if num_tries >= max_tries {
                         error!(%num_tries, %max_tries, ?backoff_delay, "couldn't fetch diff: maximum tries reached");
 
-                        time::sleep(backoff_delay).await;
+                        sleep(backoff_delay).await;
                         backoff_delay = Duration::min(backoff_delay.mul_f32(2_f32), max_backoff);
                         num_tries = 0;
                     }

consensus/src/sync/syncer.rs

@@ -13,7 +13,7 @@ use nimiq_blockchain_interface::AbstractBlockchain;
 use nimiq_blockchain_proxy::BlockchainProxy;
 use nimiq_hash::Blake2bHash;
 use nimiq_network_interface::network::{CloseReason, Network};
-use wasm_timer::Interval;
+use nimiq_time::{interval, Interval};
 
 use crate::{consensus::ResolveBlockRequest, messages::RequestHead};
 
@@ -158,7 +158,7 @@ impl<N: Network, M: MacroSync<N::PeerId>, L: LiveSync<N>> Syncer<N, M, L> {
             network,
             outdated_peers: Default::default(),
             incompatible_peers: Default::default(),
-            check_interval: Interval::new(Self::CHECK_INTERVAL),
+            check_interval: interval(Self::CHECK_INTERVAL),
             pending_checks: Default::default(),
             accepted_announcements: 0,
         }

consensus/tests/history_sync.rs

@@ -1,5 +1,3 @@
-mod sync_utils;
-
 use std::{sync::Arc, time::Duration};
 
 use futures::StreamExt;
@@ -29,12 +27,15 @@ use nimiq_test_utils::{
     node::TESTING_BLS_CACHE_MAX_CAPACITY,
     test_network::TestNetwork,
 };
+use nimiq_time::sleep;
 use nimiq_utils::time::OffsetTime;
 use nimiq_zkp_component::ZKPComponent;
 use parking_lot::{Mutex, RwLock};
 
 use crate::sync_utils::{sync_two_peers, SyncMode};
 
+mod sync_utils;
+
 #[test(tokio::test)]
 async fn two_peers_can_sync_empty_chain() {
     sync_two_peers(0, 0, SyncMode::History).await
@@ -272,7 +273,7 @@ async fn sync_ingredients() {
     Network::connect_networks(&networks, 3u64).await;
     // Then wait for connection to be established.
     let _ = stream.next().await.unwrap();
-    tokio::time::sleep(Duration::from_secs(1)).await; // FIXME, Prof. Berrang told me to do this
+    sleep(Duration::from_secs(1)).await; // FIXME, Prof. Berrang told me to do this
 
     // Test ingredients:
     // Request macro chain, first request must return all epochs and one checkpoint.

consensus/tests/request_component.rs

@@ -1,5 +1,6 @@
 use std::{sync::Arc, time::Duration};
 
+use futures::StreamExt;
 use nimiq_blockchain::BlockProducer;
 use nimiq_blockchain_interface::AbstractBlockchain;
 use nimiq_bls::KeyPair as BLSKeyPair;
@@ -14,6 +15,7 @@ use nimiq_test_utils::{
     node::Node,
     validator::seeded_rng,
 };
+use nimiq_time::{interval, sleep};
 
 #[test(tokio::test(flavor = "multi_thread", worker_threads = 4))]
 #[ignore]
@@ -56,13 +58,13 @@ async fn test_request_component() {
         tokio::spawn(async move {
             loop {
                 produce_macro_blocks(&producer1, &prod_blockchain, 1);
-                tokio::time::sleep(Duration::from_secs(5)).await;
+                sleep(Duration::from_secs(5)).await;
             }
         });
     }
 
     let mut connected = false;
-    let mut interval = tokio::time::interval(Duration::from_secs(1));
+    let mut interval = interval(Duration::from_secs(1));
     loop {
         if node1.blockchain.read().block_number() > 200 + Policy::genesis_block_number()
             && !connected
@@ -83,6 +85,6 @@ async fn test_request_component() {
             node2.blockchain.read().head_hash()
         );
 
-        interval.tick().await;
+        interval.next().await;
     }
 }

handel/Cargo.toml

@@ -17,6 +17,7 @@ workspace = true
 async-trait = "0.1"
 futures = { workspace = true }
 log = { workspace = true }
+instant = { version = "0.1", features = ["wasm-bindgen"] }
 parking_lot = "0.12"
 rand = "0.8"
 serde = "1.0"
@@ -28,13 +29,13 @@ nimiq-bls = { workspace = true }
 nimiq-collections = { workspace = true }
 nimiq-hash = { workspace = true }
 nimiq-serde = { workspace = true }
-nimiq-utils = { workspace = true, features = [
-    "math",
-] }
+nimiq-time = { workspace = true }
+nimiq-utils = { workspace = true, features = ["math"] }
 
 [dev-dependencies]
 nimiq-network-interface = { workspace = true }
 nimiq-network-mock = { workspace = true }
 nimiq-test-log = { workspace = true }
 
 tokio = { version = "1.37", features = ["rt", "time", "macros"] }
+nimiq-utils = { workspace = true, features = ["math"] }

handel/src/aggregation.rs

@@ -9,8 +9,7 @@ use futures::{
     ready, select,
     stream::{BoxStream, Stream, StreamExt},
 };
-use tokio::time::{interval_at, Instant};
-use tokio_stream::wrappers::IntervalStream;
+use nimiq_time::{interval, Interval};
 
 use crate::{
     config::Config,
@@ -57,10 +56,10 @@ struct NextAggregation<
     sender: LevelUpdateSender<N>,
 
     /// Interval for starting the next level regardless of previous levels completion
-    start_level_interval: IntervalStream,
+    start_level_interval: Interval,
 
     /// Interval for sending level updates to the corresponding peers regardless of progression
-    periodic_update_interval: IntervalStream,
+    periodic_update_interval: Interval,
 
     /// the level which needs activation next
     next_level_timeout: usize,
@@ -90,15 +89,11 @@ impl<
 
         // Regardless of level completion consecutive levels need to be activated at some point. Activate Levels every time this interval ticks,
         // if the level has not already been activated due to level completion
-        let start_level_interval =
-            IntervalStream::new(interval_at(Instant::now() + config.timeout, config.timeout));
+        let start_level_interval = interval(config.timeout);
 
         // Every `config.update_interval` send Level updates to corresponding peers no matter the aggregations progression
         // (makes sure other peers can catch up).
-        let periodic_update_interval = IntervalStream::new(interval_at(
-            Instant::now() + config.update_interval,
-            config.update_interval,
-        ));
+        let periodic_update_interval = interval(config.update_interval);
 
         // Create the NextAggregation struct
         Self {

handel/src/network.rs

@@ -159,14 +159,13 @@ impl<TNetwork: Network + Unpin> Stream for LevelUpdateSender<TNetwork> {
 
 #[cfg(test)]
 mod test {
-    use std::{sync::Arc, task::Context};
+    use std::{sync::Arc, task::Context, time::Duration};
 
     use futures::{FutureExt, StreamExt};
     use nimiq_collections::BitSet;
     use nimiq_test_log::test;
     use parking_lot::Mutex;
     use serde::{Deserialize, Serialize};
-    use tokio::time;
 
     use crate::{
         contribution::{AggregatableContribution, ContributionError},
@@ -199,7 +198,7 @@ mod test {
         ) -> futures::future::BoxFuture<'static, ()> {
             self.0.lock().push(msg);
 
-            async move { time::sleep(time::Duration::from_millis(100)).await }.boxed()
+            async move { nimiq_time::sleep(Duration::from_millis(100)).await }.boxed()
         }
     }
 
@@ -247,7 +246,7 @@ mod test {
         // Clear the buffer so test starts from scratch
         t.lock().clear();
         // Needed because the send also sleeps
-        time::sleep(time::Duration::from_millis(110)).await;
+        nimiq_time::sleep(Duration::from_millis(110)).await;
 
         assert_eq!(0, t.lock().len());
         send(&mut sender, 0);
@@ -271,7 +270,7 @@ mod test {
         assert_eq!(10, t.lock().len());
 
         // Wait for the futures to resolve, imitating a delay
-        time::sleep(time::Duration::from_millis(150)).await;
+        nimiq_time::sleep(Duration::from_millis(150)).await;
         // Send some more
         send(&mut sender, 9); // Not a Duplicate, this should be accepted
         send(&mut sender, 8); // Not a Duplicate, this should be accepted