From b66c4cf0bb9b29b0b326ce7a5dd748fe724ea3ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 3 Aug 2023 11:01:03 +0100 Subject: [PATCH] removed dependency on wasm-timer fork (#3733) --- Cargo.lock | 39 +- Cargo.toml | 3 + clients/webassembly/Cargo.lock | 53 +- clients/webassembly/Cargo.toml | 2 +- .../src/tester/ephemeral_receiver.rs | 3 +- common/client-core/Cargo.toml | 6 +- .../src/client/cover_traffic_stream.rs | 37 +- common/client-core/src/client/helpers/wasm.rs | 5 +- .../real_traffic_stream.rs | 32 +- common/client-core/src/init/helpers.rs | 16 +- common/client-libs/gateway-client/Cargo.toml | 6 +- .../client-libs/gateway-client/src/client.rs | 25 +- common/nonexhaustive-delayqueue/Cargo.toml | 18 +- common/nonexhaustive-delayqueue/src/lib.rs | 12 +- .../src/wasm_delay_queue/delay_queue.rs | 1234 ----------------- .../src/wasm_delay_queue/mod.rs | 42 - .../src/wasm_delay_queue/wheel/level.rs | 253 ---- .../src/wasm_delay_queue/wheel/mod.rs | 315 ----- .../src/wasm_delay_queue/wheel/stack.rs | 28 - 19 files changed, 102 insertions(+), 2027 deletions(-) delete mode 100644 common/nonexhaustive-delayqueue/src/wasm_delay_queue/delay_queue.rs delete mode 100644 common/nonexhaustive-delayqueue/src/wasm_delay_queue/mod.rs delete mode 100644 common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/level.rs delete mode 100644 common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/mod.rs delete mode 100644 common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/stack.rs diff --git a/Cargo.lock b/Cargo.lock index d576e05c0c..a823625624 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2924,9 +2924,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7a5bbe824c507c5da5956355e86a746d82e0e1464f65d862cc5e71da70e94b2c" dependencies = [ "cfg-if", - "js-sys", - "wasm-bindgen", - "web-sys", ] [[package]] @@ -3808,8 +3805,8 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-timer", "wasm-utils", + "wasmtimer", "zeroize", ] @@ -4082,8 +4079,8 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-timer", "wasm-utils", + "wasmtimer", ] [[package]] @@ -4361,12 +4358,10 @@ dependencies = [ name = "nym-nonexhaustive-delayqueue" version = "0.1.0" dependencies = [ - "futures-core", - "slab", "tokio", "tokio-stream", "tokio-util", - "wasm-timer", + "wasmtimer", ] [[package]] @@ -8131,20 +8126,6 @@ version = "0.2.83" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c38c045535d93ec4f0b4defec448e4291638ee608530863b1e2ba115d4fff7f" -[[package]] -name = "wasm-timer" -version = "0.2.5" -source = "git+https://github.com/mmsinclair/wasm-timer?rev=b9d1a54ad514c2f230a026afe0dde341e98cd7b6#b9d1a54ad514c2f230a026afe0dde341e98cd7b6" -dependencies = [ - "futures", - "js-sys", - "parking_lot 0.11.2", - "pin-utils", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "wasm-utils" version = "0.1.0" @@ -8163,6 +8144,20 @@ dependencies = [ "web-sys", ] +[[package]] +name = "wasmtimer" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5f656cd8858a5164932d8a90f936700860976ec21eb00e0fe2aa8cab13f6b4cf" +dependencies = [ + "futures", + "js-sys", + "parking_lot 0.12.1", + "pin-utils", + "slab", + "wasm-bindgen", +] + [[package]] name = "web-sys" version = "0.3.60" diff --git a/Cargo.toml b/Cargo.toml index fd137c6bdf..22b4cb99ac 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -146,3 +146,6 @@ thiserror = "1.0.38" tokio = "1.24.1" url = "2.2" zeroize = "1.6.0" + +# wasm-related dependencies +wasmtimer = "0.2.0" \ No newline at end of file diff --git a/clients/webassembly/Cargo.lock b/clients/webassembly/Cargo.lock index f0733eac1b..330f31c4d5 100644 --- a/clients/webassembly/Cargo.lock +++ b/clients/webassembly/Cargo.lock @@ -1539,7 +1539,7 @@ checksum = "a604f7a68fbf8103337523b1fadc8ade7361ee3f112f7c680ad179651616aed5" dependencies = [ "futures-core", "lock_api", - "parking_lot", + "parking_lot 0.11.2", ] [[package]] @@ -2114,9 +2114,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7a5bbe824c507c5da5956355e86a746d82e0e1464f65d862cc5e71da70e94b2c" dependencies = [ "cfg-if 1.0.0", - "js-sys", - "wasm-bindgen", - "web-sys", ] [[package]] @@ -2501,8 +2498,8 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-timer", "wasm-utils", + "wasmtimer", "zeroize", ] @@ -2539,8 +2536,8 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "wasm-bindgen-test", - "wasm-timer", "wasm-utils", + "wasmtimer", "wee_alloc", "zeroize", ] @@ -2719,8 +2716,8 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-timer", "wasm-utils", + "wasmtimer", ] [[package]] @@ -2836,12 +2833,10 @@ dependencies = [ name = "nym-nonexhaustive-delayqueue" version = "0.1.0" dependencies = [ - "futures-core", - "slab", "tokio", "tokio-stream", "tokio-util", - "wasm-timer", + "wasmtimer", ] [[package]] @@ -3252,6 +3247,16 @@ dependencies = [ "parking_lot_core 0.8.6", ] +[[package]] +name = "parking_lot" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f" +dependencies = [ + "lock_api", + "parking_lot_core 0.9.7", +] + [[package]] name = "parking_lot_core" version = "0.8.6" @@ -5233,20 +5238,6 @@ dependencies = [ "quote", ] -[[package]] -name = "wasm-timer" -version = "0.2.5" -source = "git+https://github.com/mmsinclair/wasm-timer?rev=b9d1a54ad514c2f230a026afe0dde341e98cd7b6#b9d1a54ad514c2f230a026afe0dde341e98cd7b6" -dependencies = [ - "futures", - "js-sys", - "parking_lot", - "pin-utils", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "wasm-utils" version = "0.1.0" @@ -5265,6 +5256,20 @@ dependencies = [ "web-sys", ] +[[package]] +name = "wasmtimer" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5f656cd8858a5164932d8a90f936700860976ec21eb00e0fe2aa8cab13f6b4cf" +dependencies = [ + "futures", + "js-sys", + "parking_lot 0.12.1", + "pin-utils", + "slab", + "wasm-bindgen", +] + [[package]] name = "web-sys" version = "0.3.60" diff --git a/clients/webassembly/Cargo.toml b/clients/webassembly/Cargo.toml index 11d1cb131a..7405002cea 100644 --- a/clients/webassembly/Cargo.toml +++ b/clients/webassembly/Cargo.toml @@ -33,7 +33,7 @@ wasm-bindgen-futures = "0.4" thiserror = "1.0.40" zeroize = "1.6.0" -wasm-timer = { git = "https://github.com/mmsinclair/wasm-timer", rev = "b9d1a54ad514c2f230a026afe0dde341e98cd7b6"} +wasmtimer = { version = "0.2.0", features = ["tokio"] } # internal nym-node-tester-utils = { path = "../../common/node-tester-utils" } diff --git a/clients/webassembly/src/tester/ephemeral_receiver.rs b/clients/webassembly/src/tester/ephemeral_receiver.rs index 4f2ce3cd8a..826bfa07b2 100644 --- a/clients/webassembly/src/tester/ephemeral_receiver.rs +++ b/clients/webassembly/src/tester/ephemeral_receiver.rs @@ -10,6 +10,7 @@ use std::collections::HashSet; use std::time::Duration; use tokio::sync::MutexGuard as AsyncMutexGuard; use wasm_utils::{console_error, console_log, console_warn}; +use wasmtimer::tokio::sleep; pub(crate) struct EphemeralTestReceiver<'a> { sent_packets: u32, @@ -90,7 +91,7 @@ impl<'a> EphemeralTestReceiver<'a> { } pub(crate) async fn perform_test(mut self) -> NodeTestResult { - let mut timeout_fut = wasm_timer::Delay::new(self.timeout_duration); + let mut timeout_fut = sleep(self.timeout_duration); loop { tokio::select! { diff --git a/common/client-core/Cargo.toml b/common/client-core/Cargo.toml index d5ae32dbd2..ab1c9b8c04 100644 --- a/common/client-core/Cargo.toml +++ b/common/client-core/Cargo.toml @@ -69,9 +69,9 @@ version = "0.4" [target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen] version = "0.2.83" -[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-timer] -git = "https://github.com/mmsinclair/wasm-timer" -rev = "b9d1a54ad514c2f230a026afe0dde341e98cd7b6" +[target."cfg(target_arch = \"wasm32\")".dependencies.wasmtimer] +workspace = true +features = ["tokio"] [target."cfg(target_arch = \"wasm32\")".dependencies.gloo-timers] version = "0.2.4" diff --git a/common/client-core/src/client/cover_traffic_stream.rs b/common/client-core/src/client/cover_traffic_stream.rs index 8784bfb7ba..c603926933 100644 --- a/common/client-core/src/client/cover_traffic_stream.rs +++ b/common/client-core/src/client/cover_traffic_stream.rs @@ -19,10 +19,10 @@ use std::time::Duration; use tokio::sync::mpsc::error::TrySendError; #[cfg(not(target_arch = "wasm32"))] -use tokio::time; +use tokio::time::{sleep, Sleep}; #[cfg(target_arch = "wasm32")] -use wasm_timer; +use wasmtimer::tokio::{sleep, Sleep}; pub struct LoopCoverTrafficStream where @@ -39,11 +39,7 @@ where /// Internal state, determined by `average_message_sending_delay`, /// used to keep track of when a next packet should be sent out. - #[cfg(not(target_arch = "wasm32"))] - next_delay: Pin>, - - #[cfg(target_arch = "wasm32")] - next_delay: Pin>, + next_delay: Pin>, /// Channel used for sending prepared nym packets to `MixTrafficController` that sends them /// out to the network without any further delays. @@ -90,17 +86,9 @@ where // The next interval value is `next_poisson_delay` after the one that just // yielded. - #[cfg(not(target_arch = "wasm32"))] - { - let now = self.next_delay.deadline(); - let next = now + next_poisson_delay; - self.next_delay.as_mut().reset(next); - } - - #[cfg(target_arch = "wasm32")] - { - self.next_delay.as_mut().reset(next_poisson_delay); - } + let now = self.next_delay.deadline(); + let next = now + next_poisson_delay; + self.next_delay.as_mut().reset(next); Poll::Ready(Some(())) } @@ -120,11 +108,7 @@ impl LoopCoverTrafficStream { ) -> Self { let rng = OsRng; - #[cfg(not(target_arch = "wasm32"))] - let next_delay = Box::pin(time::sleep(Default::default())); - - #[cfg(target_arch = "wasm32")] - let next_delay = Box::pin(wasm_timer::Delay::new(Default::default())); + let next_delay = Box::pin(sleep(Default::default())); LoopCoverTrafficStream { ack_key, @@ -142,12 +126,7 @@ impl LoopCoverTrafficStream { } fn set_next_delay(&mut self, amount: Duration) { - #[cfg(not(target_arch = "wasm32"))] - let next_delay = Box::pin(time::sleep(amount)); - - #[cfg(target_arch = "wasm32")] - let next_delay = Box::pin(wasm_timer::Delay::new(amount)); - + let next_delay = Box::pin(sleep(amount)); self.next_delay = next_delay; } diff --git a/common/client-core/src/client/helpers/wasm.rs b/common/client-core/src/client/helpers/wasm.rs index f04dd64028..5046463ee5 100644 --- a/common/client-core/src/client/helpers/wasm.rs +++ b/common/client-core/src/client/helpers/wasm.rs @@ -2,13 +2,12 @@ // SPDX-License-Identifier: Apache-2.0 use std::time::Duration; -use wasm_timer; -pub use wasm_timer::*; +pub use wasmtimer::{std::Instant, tokio::*}; pub type IntervalStream = gloo_timers::future::IntervalStream; pub(crate) fn get_time_now() -> Instant { - wasm_timer::Instant::now() + Instant::now() } pub(crate) fn new_interval_stream(polling_rate: Duration) -> IntervalStream { diff --git a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs index 44c9810e70..a160a51894 100644 --- a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -27,9 +27,10 @@ use std::sync::Arc; use std::time::Duration; #[cfg(not(target_arch = "wasm32"))] -use tokio::time; +use tokio::time::{sleep, Sleep}; + #[cfg(target_arch = "wasm32")] -use wasm_timer; +use wasmtimer::tokio::{sleep, Sleep}; mod sending_delay_controller; @@ -82,11 +83,7 @@ where /// Internal state, determined by `average_message_sending_delay`, /// used to keep track of when a next packet should be sent out. - #[cfg(not(target_arch = "wasm32"))] - next_delay: Option>>, - - #[cfg(target_arch = "wasm32")] - next_delay: Option>>, + next_delay: Option>>, // To make sure we don't overload the mix_tx channel, we limit the rate we are pushing // messages. @@ -373,17 +370,9 @@ where // The next interval value is `next_poisson_delay` after the one that just // yielded. - #[cfg(not(target_arch = "wasm32"))] - { - let now = next_delay.deadline(); - let next = now + next_poisson_delay; - next_delay.as_mut().reset(next); - } - - #[cfg(target_arch = "wasm32")] - { - next_delay.as_mut().reset(next_poisson_delay); - } + let now = next_delay.deadline(); + let next = now + next_poisson_delay; + next_delay.as_mut().reset(next); // On every iteration we get new messages from upstream. Given that these come bunched // in `Vec`, this ensures that on average we will fetch messages faster than we can @@ -421,12 +410,7 @@ where self.config.traffic.message_sending_average_delay, ); - #[cfg(not(target_arch = "wasm32"))] - let next_delay = Box::pin(time::sleep(sampled)); - - #[cfg(target_arch = "wasm32")] - let next_delay = Box::pin(wasm_timer::Delay::new(sampled)); - + let next_delay = Box::pin(sleep(sampled)); self.next_delay = Some(next_delay); Poll::Pending diff --git a/common/client-core/src/init/helpers.rs b/common/client-core/src/init/helpers.rs index 1c23d562f9..723e9d02a2 100644 --- a/common/client-core/src/init/helpers.rs +++ b/common/client-core/src/init/helpers.rs @@ -27,13 +27,17 @@ use tokio_tungstenite::connect_async; use tokio_tungstenite::{MaybeTlsStream, WebSocketStream}; #[cfg(not(target_arch = "wasm32"))] type WsConn = WebSocketStream>; +#[cfg(not(target_arch = "wasm32"))] +use tokio::time::sleep; #[cfg(target_arch = "wasm32")] use nym_bandwidth_controller::wasm_mockups::DirectSigningNyxdClient; #[cfg(target_arch = "wasm32")] -use wasm_timer::Instant; -#[cfg(target_arch = "wasm32")] use wasm_utils::websocket::JSWebsocket; +#[cfg(target_arch = "wasm32")] +use wasmtimer::std::Instant; +#[cfg(target_arch = "wasm32")] +use wasmtimer::tokio::sleep; #[cfg(target_arch = "wasm32")] type WsConn = JSWebsocket; @@ -125,15 +129,9 @@ async fn measure_latency(gateway: &gateway::Node) -> Result(()) }; - // thanks to wasm we can't use tokio::time::timeout : ( - #[cfg(not(target_arch = "wasm32"))] - let timeout = tokio::time::sleep(PING_TIMEOUT); - #[cfg(not(target_arch = "wasm32"))] + let timeout = sleep(PING_TIMEOUT); tokio::pin!(timeout); - #[cfg(target_arch = "wasm32")] - let mut timeout = wasm_timer::Delay::new(PING_TIMEOUT); - tokio::select! { _ = &mut timeout => { warn!("timed out while trying to perform measurement...") diff --git a/common/client-libs/gateway-client/Cargo.toml b/common/client-libs/gateway-client/Cargo.toml index 20d5c958b0..197acdadef 100644 --- a/common/client-libs/gateway-client/Cargo.toml +++ b/common/client-libs/gateway-client/Cargo.toml @@ -58,9 +58,9 @@ path = "../../wasm-utils" features = ["websocket"] # only import it in wasm. Prefer proper tokio timer in non-wasm -[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-timer] -git = "https://github.com/mmsinclair/wasm-timer" -rev = "b9d1a54ad514c2f230a026afe0dde341e98cd7b6" +[target."cfg(target_arch = \"wasm32\")".dependencies.wasmtimer] +workspace = true +features = ["tokio"] # this is due to tungstenite using `rand` 0.8 and associated changes in `getrandom` crate # which now does not support wasm32-unknown-unknown target by default. diff --git a/common/client-libs/gateway-client/src/client.rs b/common/client-libs/gateway-client/src/client.rs index a6ef774d8f..bdc2c626df 100644 --- a/common/client-libs/gateway-client/src/client.rs +++ b/common/client-libs/gateway-client/src/client.rs @@ -31,14 +31,16 @@ use tungstenite::protocol::Message; #[cfg(not(target_arch = "wasm32"))] use nym_validator_client::nyxd::traits::DkgQueryClient; #[cfg(not(target_arch = "wasm32"))] +use tokio::time::sleep; +#[cfg(not(target_arch = "wasm32"))] use tokio_tungstenite::connect_async; #[cfg(target_arch = "wasm32")] use nym_bandwidth_controller::wasm_mockups::DkgQueryClient; #[cfg(target_arch = "wasm32")] -use wasm_timer; -#[cfg(target_arch = "wasm32")] use wasm_utils::websocket::JSWebsocket; +#[cfg(target_arch = "wasm32")] +use wasmtimer::tokio::sleep; const DEFAULT_RECONNECTION_ATTEMPTS: usize = 10; const DEFAULT_RECONNECTION_BACKOFF: Duration = Duration::from_secs(5); @@ -198,15 +200,7 @@ impl GatewayClient { } #[cfg(not(target_arch = "wasm32"))] - tokio::time::sleep(self.reconnection_backoff).await; - - #[cfg(target_arch = "wasm32")] - if let Err(err) = wasm_timer::Delay::new(self.reconnection_backoff).await { - error!( - "the timer has gone away while in reconnection backoff! - {}", - err - ); - } + sleep(self.reconnection_backoff).await; } // final attempt (done separately to be able to return a proper error) @@ -235,16 +229,9 @@ impl GatewayClient { _ => return Err(GatewayClientError::ConnectionInInvalidState), }; - #[cfg(not(target_arch = "wasm32"))] - let timeout = tokio::time::sleep(self.response_timeout_duration); - #[cfg(not(target_arch = "wasm32"))] + let timeout = sleep(self.response_timeout_duration); tokio::pin!(timeout); - // technically the `wasm_timer` also works outside wasm, but unless required, - // I really prefer to just stick to tokio - #[cfg(target_arch = "wasm32")] - let mut timeout = wasm_timer::Delay::new(self.response_timeout_duration); - loop { tokio::select! { _ = self.shutdown.recv() => { diff --git a/common/nonexhaustive-delayqueue/Cargo.toml b/common/nonexhaustive-delayqueue/Cargo.toml index 975eb631fe..f79284dd50 100644 --- a/common/nonexhaustive-delayqueue/Cargo.toml +++ b/common/nonexhaustive-delayqueue/Cargo.toml @@ -7,16 +7,16 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -tokio = { version = "1.24.1", features = [] } tokio-stream = "0.1.11" # this one seems to be a thing until `Stream` trait is stabilised in stdlib -tokio-util = { version = "0.7.4", features = ["time"] } -[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-timer] -git = "https://github.com/mmsinclair/wasm-timer" -rev = "b9d1a54ad514c2f230a026afe0dde341e98cd7b6" +[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio] +workspace = true -[target."cfg(target_arch = \"wasm32\")".dependencies.slab] -version = "0.4.4" +[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio-util] +version = "0.7.4" +features = ["time"] + +[target."cfg(target_arch = \"wasm32\")".dependencies.wasmtimer] +workspace = true +features = ["tokio", "tokio-util"] -[target."cfg(target_arch = \"wasm32\")".dependencies.futures-core] -version = "0.3.0" diff --git a/common/nonexhaustive-delayqueue/src/lib.rs b/common/nonexhaustive-delayqueue/src/lib.rs index 4e2329db2b..ee8c9f214c 100644 --- a/common/nonexhaustive-delayqueue/src/lib.rs +++ b/common/nonexhaustive-delayqueue/src/lib.rs @@ -6,12 +6,8 @@ use std::task::{Context, Poll, Waker}; use std::time::Duration; use tokio_stream::Stream; -pub use tokio::time::error::Error as TimerError; - // this is a copy of tokio-util delay_queue with `Sleep` and `Instant` being replaced with // `wasm_timer` equivalents -#[cfg(target_arch = "wasm32")] -mod wasm_delay_queue; #[cfg(not(target_arch = "wasm32"))] type DelayQueue = tokio_util::time::DelayQueue; @@ -23,13 +19,13 @@ pub type QueueKey = tokio_util::time::delay_queue::Key; use tokio::time::Instant; #[cfg(target_arch = "wasm32")] -type DelayQueue = crate::wasm_delay_queue::DelayQueue; +type DelayQueue = wasmtimer::tokio_util::DelayQueue; #[cfg(target_arch = "wasm32")] -pub use crate::wasm_delay_queue::delay_queue::Expired; +pub use wasmtimer::tokio_util::delay_queue::Expired; #[cfg(target_arch = "wasm32")] -pub type QueueKey = crate::wasm_delay_queue::delay_queue::Key; +pub type QueueKey = wasmtimer::tokio_util::delay_queue::Key; #[cfg(target_arch = "wasm32")] -use wasm_timer::Instant; +use wasmtimer::std::Instant; /// A variant of tokio's `DelayQueue`, such that its `Stream` implementation will never return a 'None'. pub struct NonExhaustiveDelayQueue { diff --git a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/delay_queue.rs b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/delay_queue.rs deleted file mode 100644 index 6d18b7b12d..0000000000 --- a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/delay_queue.rs +++ /dev/null @@ -1,1234 +0,0 @@ -//! A queue of delayed elements. -//! -//! See [`DelayQueue`] for more details. -//! -//! [`DelayQueue`]: struct@DelayQueue - -use crate::wasm_delay_queue::sleep_until; -use crate::wasm_delay_queue::wheel::{self, Wheel}; -use core::ops::{Index, IndexMut}; -use futures_core::{ready, Future}; -use slab::Slab; -use std::cmp; -use std::collections::HashMap; -use std::convert::From; -use std::fmt; -use std::fmt::Debug; -use std::marker::PhantomData; -use std::pin::Pin; -use std::task::{self, Poll, Waker}; -use std::time::Duration; -use wasm_timer::{Delay, Instant}; - -/// A queue of delayed elements. -/// -/// Once an element is inserted into the `DelayQueue`, it is yielded once the -/// specified deadline has been reached. -/// -/// # Usage -/// -/// Elements are inserted into `DelayQueue` using the [`insert`] or -/// [`insert_at`] methods. A deadline is provided with the item and a [`Key`] is -/// returned. The key is used to remove the entry or to change the deadline at -/// which it should be yielded back. -/// -/// Once delays have been configured, the `DelayQueue` is used via its -/// [`Stream`] implementation. [`poll_expired`] is called. If an entry has reached its -/// deadline, it is returned. If not, `Poll::Pending` is returned indicating that the -/// current task will be notified once the deadline has been reached. -/// -/// # `Stream` implementation -/// -/// Items are retrieved from the queue via [`DelayQueue::poll_expired`]. If no delays have -/// expired, no items are returned. In this case, `Poll::Pending` is returned and the -/// current task is registered to be notified once the next item's delay has -/// expired. -/// -/// If no items are in the queue, i.e. `is_empty()` returns `true`, then `poll` -/// returns `Poll::Ready(None)`. This indicates that the stream has reached an end. -/// However, if a new item is inserted *after*, `poll` will once again start -/// returning items or `Poll::Pending`. -/// -/// Items are returned ordered by their expirations. Items that are configured -/// to expire first will be returned first. There are no ordering guarantees -/// for items configured to expire at the same instant. Also note that delays are -/// rounded to the closest millisecond. -/// -/// # Implementation -/// -/// The [`DelayQueue`] is backed by a separate instance of a timer wheel similar to that used internally -/// by Tokio's standalone timer utilities such as [`sleep`]. Because of this, it offers the same -/// performance and scalability benefits. -/// -/// State associated with each entry is stored in a [`slab`]. This amortizes the cost of allocation, -/// and allows reuse of the memory allocated for expired entires. -/// -/// Capacity can be checked using [`capacity`] and allocated preemptively by using -/// the [`reserve`] method. -/// -/// # Usage -/// -/// Using `DelayQueue` to manage cache entries. -/// -/// ```rust,no_run -/// use tokio_util::time::{DelayQueue, delay_queue}; -/// -/// use futures::ready; -/// use std::collections::HashMap; -/// use std::task::{Context, Poll}; -/// use std::time::Duration; -/// # type CacheKey = String; -/// # type Value = String; -/// -/// struct Cache { -/// entries: HashMap, -/// expirations: DelayQueue, -/// } -/// -/// const TTL_SECS: u64 = 30; -/// -/// impl Cache { -/// fn insert(&mut self, key: CacheKey, value: Value) { -/// let delay = self.expirations -/// .insert(key.clone(), Duration::from_secs(TTL_SECS)); -/// -/// self.entries.insert(key, (value, delay)); -/// } -/// -/// fn get(&self, key: &CacheKey) -> Option<&Value> { -/// self.entries.get(key) -/// .map(|&(ref v, _)| v) -/// } -/// -/// fn remove(&mut self, key: &CacheKey) { -/// if let Some((_, cache_key)) = self.entries.remove(key) { -/// self.expirations.remove(&cache_key); -/// } -/// } -/// -/// fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll<()> { -/// while let Some(entry) = ready!(self.expirations.poll_expired(cx)) { -/// self.entries.remove(entry.get_ref()); -/// } -/// -/// Poll::Ready(()) -/// } -/// } -/// ``` -/// -/// [`insert`]: method@Self::insert -/// [`insert_at`]: method@Self::insert_at -/// [`Key`]: struct@Key -/// [`Stream`]: https://docs.rs/futures/0.1/futures/stream/trait.Stream.html -/// [`poll_expired`]: method@Self::poll_expired -/// [`Stream::poll_expired`]: method@Self::poll_expired -/// [`DelayQueue`]: struct@DelayQueue -/// [`sleep`]: fn@tokio::time::sleep -/// [`slab`]: slab -/// [`capacity`]: method@Self::capacity -/// [`reserve`]: method@Self::reserve -#[derive(Debug)] -pub struct DelayQueue { - /// Stores data associated with entries - slab: SlabStorage, - - /// Lookup structure tracking all delays in the queue - wheel: Wheel>, - - /// Delays that were inserted when already expired. These cannot be stored - /// in the wheel - expired: Stack, - - /// Delay expiring when the *first* item in the queue expires - delay: Option>>, - - /// Wheel polling state - wheel_now: u64, - - /// Instant at which the timer starts - start: Instant, - - /// Waker that is invoked when we potentially need to reset the timer. - /// Because we lazily create the timer when the first entry is created, we - /// need to awaken any poller that polled us before that point. - waker: Option, -} - -#[derive(Default)] -struct SlabStorage { - inner: Slab>, - - // A `compact` call requires a re-mapping of the `Key`s that were changed - // during the `compact` call of the `slab`. Since the keys that were given out - // cannot be changed retroactively we need to keep track of these re-mappings. - // The keys of `key_map` correspond to the old keys that were given out and - // the values to the `Key`s that were re-mapped by the `compact` call. - key_map: HashMap, - - // Index used to create new keys to hand out. - next_key_index: usize, - - // Whether `compact` has been called, necessary in order to decide whether - // to include keys in `key_map`. - compact_called: bool, -} - -impl SlabStorage { - pub(crate) fn with_capacity(capacity: usize) -> SlabStorage { - SlabStorage { - inner: Slab::with_capacity(capacity), - key_map: HashMap::new(), - next_key_index: 0, - compact_called: false, - } - } - - // Inserts data into the inner slab and re-maps keys if necessary - pub(crate) fn insert(&mut self, val: Data) -> Key { - let mut key = KeyInternal::new(self.inner.insert(val)); - let key_contained = self.key_map.contains_key(&key.into()); - - if key_contained { - // It's possible that a `compact` call creates capacity in `self.inner` in - // such a way that a `self.inner.insert` call creates a `key` which was - // previously given out during an `insert` call prior to the `compact` call. - // If `key` is contained in `self.key_map`, we have encountered this exact situation, - // We need to create a new key `key_to_give_out` and include the relation - // `key_to_give_out` -> `key` in `self.key_map`. - let key_to_give_out = self.create_new_key(); - assert!(!self.key_map.contains_key(&key_to_give_out.into())); - self.key_map.insert(key_to_give_out.into(), key); - key = key_to_give_out; - } else if self.compact_called { - // Include an identity mapping in `self.key_map` in order to allow us to - // panic if a key that was handed out is removed more than once. - self.key_map.insert(key.into(), key); - } - - key.into() - } - - // Re-map the key in case compact was previously called. - // Note: Since we include identity mappings in key_map after compact was called, - // we have information about all keys that were handed out. In the case in which - // compact was called and we try to remove a Key that was previously removed - // we can detect invalid keys if no key is found in `key_map`. This is necessary - // in order to prevent situations in which a previously removed key - // corresponds to a re-mapped key internally and which would then be incorrectly - // removed from the slab. - // - // Example to illuminate this problem: - // - // Let's assume our `key_map` is {1 -> 2, 2 -> 1} and we call remove(1). If we - // were to remove 1 again, we would not find it inside `key_map` anymore. - // If we were to imply from this that no re-mapping was necessary, we would - // incorrectly remove 1 from `self.slab.inner`, which corresponds to the - // handed-out key 2. - pub(crate) fn remove(&mut self, key: &Key) -> Data { - let remapped_key = if self.compact_called { - match self.key_map.remove(key) { - Some(key_internal) => key_internal, - None => panic!("invalid key"), - } - } else { - (*key).into() - }; - - self.inner.remove(remapped_key.index) - } - - pub(crate) fn shrink_to_fit(&mut self) { - self.inner.shrink_to_fit(); - self.key_map.shrink_to_fit(); - } - - pub(crate) fn compact(&mut self) { - if !self.compact_called { - for (key, _) in self.inner.iter() { - self.key_map.insert(Key::new(key), KeyInternal::new(key)); - } - } - - let mut remapping = HashMap::new(); - self.inner.compact(|_, from, to| { - remapping.insert(from, to); - true - }); - - // At this point `key_map` contains a mapping for every element. - for internal_key in self.key_map.values_mut() { - if let Some(new_internal_key) = remapping.get(&internal_key.index) { - *internal_key = KeyInternal::new(*new_internal_key); - } - } - - if self.key_map.capacity() > 2 * self.key_map.len() { - self.key_map.shrink_to_fit(); - } - - self.compact_called = true; - } - - // Tries to re-map a `Key` that was given out to the user to its - // corresponding internal key. - fn remap_key(&self, key: &Key) -> Option { - let key_map = &self.key_map; - if self.compact_called { - key_map.get(&*key).copied() - } else { - Some((*key).into()) - } - } - - fn create_new_key(&mut self) -> KeyInternal { - while self.key_map.contains_key(&Key::new(self.next_key_index)) { - self.next_key_index = self.next_key_index.wrapping_add(1); - } - - KeyInternal::new(self.next_key_index) - } - - pub(crate) fn len(&self) -> usize { - self.inner.len() - } - - pub(crate) fn capacity(&self) -> usize { - self.inner.capacity() - } - - pub(crate) fn clear(&mut self) { - self.inner.clear(); - self.key_map.clear(); - self.compact_called = false; - } - - pub(crate) fn reserve(&mut self, additional: usize) { - self.inner.reserve(additional); - - if self.compact_called { - self.key_map.reserve(additional); - } - } - - pub(crate) fn is_empty(&self) -> bool { - self.inner.is_empty() - } - - pub(crate) fn contains(&self, key: &Key) -> bool { - let remapped_key = self.remap_key(key); - - match remapped_key { - Some(internal_key) => self.inner.contains(internal_key.index), - None => false, - } - } -} - -impl fmt::Debug for SlabStorage -where - T: fmt::Debug, -{ - fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { - if fmt.alternate() { - fmt.debug_map().entries(self.inner.iter()).finish() - } else { - fmt.debug_struct("Slab") - .field("len", &self.len()) - .field("cap", &self.capacity()) - .finish() - } - } -} - -impl Index for SlabStorage { - type Output = Data; - - fn index(&self, key: Key) -> &Self::Output { - let remapped_key = self.remap_key(&key); - - match remapped_key { - Some(internal_key) => &self.inner[internal_key.index], - None => panic!("Invalid index {}", key.index), - } - } -} - -impl IndexMut for SlabStorage { - fn index_mut(&mut self, key: Key) -> &mut Data { - let remapped_key = self.remap_key(&key); - - match remapped_key { - Some(internal_key) => &mut self.inner[internal_key.index], - None => panic!("Invalid index {}", key.index), - } - } -} - -/// An entry in `DelayQueue` that has expired and been removed. -/// -/// Values are returned by [`DelayQueue::poll_expired`]. -/// -/// [`DelayQueue::poll_expired`]: method@DelayQueue::poll_expired -#[derive(Debug)] -pub struct Expired { - /// The data stored in the queue - data: T, - - /// The expiration time - deadline: Instant, - - /// The key associated with the entry - key: Key, -} - -/// Token to a value stored in a `DelayQueue`. -/// -/// Instances of `Key` are returned by [`DelayQueue::insert`]. See [`DelayQueue`] -/// documentation for more details. -/// -/// [`DelayQueue`]: struct@DelayQueue -/// [`DelayQueue::insert`]: method@DelayQueue::insert -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct Key { - index: usize, -} - -// Whereas `Key` is given out to users that use `DelayQueue`, internally we use -// `KeyInternal` as the key type in order to make the logic of mapping between keys -// as a result of `compact` calls clearer. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -struct KeyInternal { - index: usize, -} - -#[derive(Debug)] -struct Stack { - /// Head of the stack - head: Option, - _p: PhantomData T>, -} - -#[derive(Debug)] -struct Data { - /// The data being stored in the queue and will be returned at the requested - /// instant. - inner: T, - - /// The instant at which the item is returned. - when: u64, - - /// Set to true when stored in the `expired` queue - expired: bool, - - /// Next entry in the stack - next: Option, - - /// Previous entry in the stack - prev: Option, -} - -/// Maximum number of entries the queue can handle -const MAX_ENTRIES: usize = (1 << 30) - 1; - -impl DelayQueue { - /// Creates a new, empty, `DelayQueue`. - /// - /// The queue will not allocate storage until items are inserted into it. - /// - /// # Examples - /// - /// ```rust - /// # use tokio_util::time::DelayQueue; - /// let delay_queue: DelayQueue = DelayQueue::new(); - /// ``` - pub fn new() -> DelayQueue { - DelayQueue::with_capacity(0) - } - - /// Creates a new, empty, `DelayQueue` with the specified capacity. - /// - /// The queue will be able to hold at least `capacity` elements without - /// reallocating. If `capacity` is 0, the queue will not allocate for - /// storage. - /// - /// # Examples - /// - /// ```rust - /// # use tokio_util::time::DelayQueue; - /// # use std::time::Duration; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::with_capacity(10); - /// - /// // These insertions are done without further allocation - /// for i in 0..10 { - /// delay_queue.insert(i, Duration::from_secs(i)); - /// } - /// - /// // This will make the queue allocate additional storage - /// delay_queue.insert(11, Duration::from_secs(11)); - /// # } - /// ``` - pub fn with_capacity(capacity: usize) -> DelayQueue { - DelayQueue { - wheel: Wheel::new(), - slab: SlabStorage::with_capacity(capacity), - expired: Stack::default(), - delay: None, - wheel_now: 0, - start: Instant::now(), - waker: None, - } - } - - /// Inserts `value` into the queue set to expire at a specific instant in - /// time. - /// - /// This function is identical to `insert`, but takes an `Instant` instead - /// of a `Duration`. - /// - /// `value` is stored in the queue until `when` is reached. At which point, - /// `value` will be returned from [`poll_expired`]. If `when` has already been - /// reached, then `value` is immediately made available to poll. - /// - /// The return value represents the insertion and is used as an argument to - /// [`remove`] and [`reset`]. Note that [`Key`] is a token and is reused once - /// `value` is removed from the queue either by calling [`poll_expired`] after - /// `when` is reached or by calling [`remove`]. At this point, the caller - /// must take care to not use the returned [`Key`] again as it may reference - /// a different item in the queue. - /// - /// See [type] level documentation for more details. - /// - /// # Panics - /// - /// This function panics if `when` is too far in the future. - /// - /// # Examples - /// - /// Basic usage - /// - /// ```rust - /// use tokio::time::{Duration, Instant}; - /// use tokio_util::time::DelayQueue; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::new(); - /// let key = delay_queue.insert_at( - /// "foo", Instant::now() + Duration::from_secs(5)); - /// - /// // Remove the entry - /// let item = delay_queue.remove(&key); - /// assert_eq!(*item.get_ref(), "foo"); - /// # } - /// ``` - /// - /// [`poll_expired`]: method@Self::poll_expired - /// [`remove`]: method@Self::remove - /// [`reset`]: method@Self::reset - /// [`Key`]: struct@Key - /// [type]: # - #[track_caller] - pub fn insert_at(&mut self, value: T, when: Instant) -> Key { - assert!(self.slab.len() < MAX_ENTRIES, "max entries exceeded"); - - // Normalize the deadline. Values cannot be set to expire in the past. - let when = self.normalize_deadline(when); - - // Insert the value in the store - let key = self.slab.insert(Data { - inner: value, - when, - expired: false, - next: None, - prev: None, - }); - - self.insert_idx(when, key); - - // Set a new delay if the current's deadline is later than the one of the new item - let should_set_delay = if let Some(ref delay) = self.delay { - let current_exp = self.normalize_deadline(delay.deadline()); - current_exp > when - } else { - true - }; - - if should_set_delay { - if let Some(waker) = self.waker.take() { - waker.wake(); - } - - let delay_time = self.start + Duration::from_millis(when); - if let Some(ref mut delay) = &mut self.delay { - delay.as_mut().reset_at(delay_time); - } else { - self.delay = Some(Box::pin(sleep_until(delay_time))); - } - } - - key - } - - /// Attempts to pull out the next value of the delay queue, registering the - /// current task for wakeup if the value is not yet available, and returning - /// `None` if the queue is exhausted. - pub fn poll_expired(&mut self, cx: &mut task::Context<'_>) -> Poll>> { - if !self - .waker - .as_ref() - .map(|w| w.will_wake(cx.waker())) - .unwrap_or(false) - { - self.waker = Some(cx.waker().clone()); - } - - let item = ready!(self.poll_idx(cx)); - Poll::Ready(item.map(|key| { - let data = self.slab.remove(&key); - debug_assert!(data.next.is_none()); - debug_assert!(data.prev.is_none()); - - Expired { - key, - data: data.inner, - deadline: self.start + Duration::from_millis(data.when), - } - })) - } - - /// Inserts `value` into the queue set to expire after the requested duration - /// elapses. - /// - /// This function is identical to `insert_at`, but takes a `Duration` - /// instead of an `Instant`. - /// - /// `value` is stored in the queue until `timeout` duration has - /// elapsed after `insert` was called. At that point, `value` will - /// be returned from [`poll_expired`]. If `timeout` is a `Duration` of - /// zero, then `value` is immediately made available to poll. - /// - /// The return value represents the insertion and is used as an - /// argument to [`remove`] and [`reset`]. Note that [`Key`] is a - /// token and is reused once `value` is removed from the queue - /// either by calling [`poll_expired`] after `timeout` has elapsed - /// or by calling [`remove`]. At this point, the caller must not - /// use the returned [`Key`] again as it may reference a different - /// item in the queue. - /// - /// See [type] level documentation for more details. - /// - /// # Panics - /// - /// This function panics if `timeout` is greater than the maximum - /// duration supported by the timer in the current `Runtime`. - /// - /// # Examples - /// - /// Basic usage - /// - /// ```rust - /// use tokio_util::time::DelayQueue; - /// use std::time::Duration; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::new(); - /// let key = delay_queue.insert("foo", Duration::from_secs(5)); - /// - /// // Remove the entry - /// let item = delay_queue.remove(&key); - /// assert_eq!(*item.get_ref(), "foo"); - /// # } - /// ``` - /// - /// [`poll_expired`]: method@Self::poll_expired - /// [`remove`]: method@Self::remove - /// [`reset`]: method@Self::reset - /// [`Key`]: struct@Key - /// [type]: # - #[track_caller] - pub fn insert(&mut self, value: T, timeout: Duration) -> Key { - self.insert_at(value, Instant::now() + timeout) - } - - #[track_caller] - fn insert_idx(&mut self, when: u64, key: Key) { - use self::wheel::{InsertError, Stack}; - - // Register the deadline with the timer wheel - match self.wheel.insert(when, key, &mut self.slab) { - Ok(_) => {} - Err((_, InsertError::Elapsed)) => { - self.slab[key].expired = true; - // The delay is already expired, store it in the expired queue - self.expired.push(key, &mut self.slab); - } - Err((_, err)) => panic!("invalid deadline; err={:?}", err), - } - } - - /// Removes the key from the expired queue or the timer wheel - /// depending on its expiration status. - /// - /// # Panics - /// - /// Panics if the key is not contained in the expired queue or the wheel. - #[track_caller] - fn remove_key(&mut self, key: &Key) { - use crate::wasm_delay_queue::wheel::Stack; - - // Special case the `expired` queue - if self.slab[*key].expired { - self.expired.remove(key, &mut self.slab); - } else { - self.wheel.remove(key, &mut self.slab); - } - } - - /// Removes the item associated with `key` from the queue. - /// - /// There must be an item associated with `key`. The function returns the - /// removed item as well as the `Instant` at which it will the delay will - /// have expired. - /// - /// # Panics - /// - /// The function panics if `key` is not contained by the queue. - /// - /// # Examples - /// - /// Basic usage - /// - /// ```rust - /// use tokio_util::time::DelayQueue; - /// use std::time::Duration; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::new(); - /// let key = delay_queue.insert("foo", Duration::from_secs(5)); - /// - /// // Remove the entry - /// let item = delay_queue.remove(&key); - /// assert_eq!(*item.get_ref(), "foo"); - /// # } - /// ``` - #[track_caller] - pub fn remove(&mut self, key: &Key) -> Expired { - let prev_deadline = self.next_deadline(); - - self.remove_key(key); - let data = self.slab.remove(key); - - let next_deadline = self.next_deadline(); - if prev_deadline != next_deadline { - match (next_deadline, &mut self.delay) { - (None, _) => self.delay = None, - (Some(deadline), Some(delay)) => delay.as_mut().reset_at(deadline), - (Some(deadline), None) => self.delay = Some(Box::pin(sleep_until(deadline))), - } - } - - Expired { - key: Key::new(key.index), - data: data.inner, - deadline: self.start + Duration::from_millis(data.when), - } - } - - /// Sets the delay of the item associated with `key` to expire at `when`. - /// - /// This function is identical to `reset` but takes an `Instant` instead of - /// a `Duration`. - /// - /// The item remains in the queue but the delay is set to expire at `when`. - /// If `when` is in the past, then the item is immediately made available to - /// the caller. - /// - /// # Panics - /// - /// This function panics if `when` is too far in the future or if `key` is - /// not contained by the queue. - /// - /// # Examples - /// - /// Basic usage - /// - /// ```rust - /// use tokio::time::{Duration, Instant}; - /// use tokio_util::time::DelayQueue; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::new(); - /// let key = delay_queue.insert("foo", Duration::from_secs(5)); - /// - /// // "foo" is scheduled to be returned in 5 seconds - /// - /// delay_queue.reset_at(&key, Instant::now() + Duration::from_secs(10)); - /// - /// // "foo" is now scheduled to be returned in 10 seconds - /// # } - /// ``` - #[track_caller] - pub fn reset_at(&mut self, key: &Key, when: Instant) { - self.remove_key(key); - - // Normalize the deadline. Values cannot be set to expire in the past. - let when = self.normalize_deadline(when); - - self.slab[*key].when = when; - self.slab[*key].expired = false; - - self.insert_idx(when, *key); - - let next_deadline = self.next_deadline(); - if let (Some(ref mut delay), Some(deadline)) = (&mut self.delay, next_deadline) { - // This should awaken us if necessary (ie, if already expired) - delay.as_mut().reset_at(deadline); - } - } - - /// Shrink the capacity of the slab, which `DelayQueue` uses internally for storage allocation. - /// This function is not guaranteed to, and in most cases, won't decrease the capacity of the slab - /// to the number of elements still contained in it, because elements cannot be moved to a different - /// index. To decrease the capacity to the size of the slab use [`compact`]. - /// - /// This function can take O(n) time even when the capacity cannot be reduced or the allocation is - /// shrunk in place. Repeated calls run in O(1) though. - /// - /// [`compact`]: method@Self::compact - pub fn shrink_to_fit(&mut self) { - self.slab.shrink_to_fit(); - } - - /// Shrink the capacity of the slab, which `DelayQueue` uses internally for storage allocation, - /// to the number of elements that are contained in it. - /// - /// This methods runs in O(n). - /// - /// # Examples - /// - /// Basic usage - /// - /// ```rust - /// use tokio_util::time::DelayQueue; - /// use std::time::Duration; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::with_capacity(10); - /// - /// let key1 = delay_queue.insert(5, Duration::from_secs(5)); - /// let key2 = delay_queue.insert(10, Duration::from_secs(10)); - /// let key3 = delay_queue.insert(15, Duration::from_secs(15)); - /// - /// delay_queue.remove(&key2); - /// - /// delay_queue.compact(); - /// assert_eq!(delay_queue.capacity(), 2); - /// # } - /// ``` - pub fn compact(&mut self) { - self.slab.compact(); - } - - /// Returns the next time to poll as determined by the wheel - fn next_deadline(&mut self) -> Option { - self.wheel - .poll_at() - .map(|poll_at| self.start + Duration::from_millis(poll_at)) - } - - /// Sets the delay of the item associated with `key` to expire after - /// `timeout`. - /// - /// This function is identical to `reset_at` but takes a `Duration` instead - /// of an `Instant`. - /// - /// The item remains in the queue but the delay is set to expire after - /// `timeout`. If `timeout` is zero, then the item is immediately made - /// available to the caller. - /// - /// # Panics - /// - /// This function panics if `timeout` is greater than the maximum supported - /// duration or if `key` is not contained by the queue. - /// - /// # Examples - /// - /// Basic usage - /// - /// ```rust - /// use tokio_util::time::DelayQueue; - /// use std::time::Duration; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::new(); - /// let key = delay_queue.insert("foo", Duration::from_secs(5)); - /// - /// // "foo" is scheduled to be returned in 5 seconds - /// - /// delay_queue.reset(&key, Duration::from_secs(10)); - /// - /// // "foo"is now scheduled to be returned in 10 seconds - /// # } - /// ``` - #[track_caller] - pub fn reset(&mut self, key: &Key, timeout: Duration) { - self.reset_at(key, Instant::now() + timeout); - } - - /// Clears the queue, removing all items. - /// - /// After calling `clear`, [`poll_expired`] will return `Ok(Ready(None))`. - /// - /// Note that this method has no effect on the allocated capacity. - /// - /// [`poll_expired`]: method@Self::poll_expired - /// - /// # Examples - /// - /// ```rust - /// use tokio_util::time::DelayQueue; - /// use std::time::Duration; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::new(); - /// - /// delay_queue.insert("foo", Duration::from_secs(5)); - /// - /// assert!(!delay_queue.is_empty()); - /// - /// delay_queue.clear(); - /// - /// assert!(delay_queue.is_empty()); - /// # } - /// ``` - pub fn clear(&mut self) { - self.slab.clear(); - self.expired = Stack::default(); - self.wheel = Wheel::new(); - self.delay = None; - } - - /// Returns the number of elements the queue can hold without reallocating. - /// - /// # Examples - /// - /// ```rust - /// use tokio_util::time::DelayQueue; - /// - /// let delay_queue: DelayQueue = DelayQueue::with_capacity(10); - /// assert_eq!(delay_queue.capacity(), 10); - /// ``` - pub fn capacity(&self) -> usize { - self.slab.capacity() - } - - /// Returns the number of elements currently in the queue. - /// - /// # Examples - /// - /// ```rust - /// use tokio_util::time::DelayQueue; - /// use std::time::Duration; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue: DelayQueue = DelayQueue::with_capacity(10); - /// assert_eq!(delay_queue.len(), 0); - /// delay_queue.insert(3, Duration::from_secs(5)); - /// assert_eq!(delay_queue.len(), 1); - /// # } - /// ``` - pub fn len(&self) -> usize { - self.slab.len() - } - - /// Reserves capacity for at least `additional` more items to be queued - /// without allocating. - /// - /// `reserve` does nothing if the queue already has sufficient capacity for - /// `additional` more values. If more capacity is required, a new segment of - /// memory will be allocated and all existing values will be copied into it. - /// As such, if the queue is already very large, a call to `reserve` can end - /// up being expensive. - /// - /// The queue may reserve more than `additional` extra space in order to - /// avoid frequent reallocations. - /// - /// # Panics - /// - /// Panics if the new capacity exceeds the maximum number of entries the - /// queue can contain. - /// - /// # Examples - /// - /// ``` - /// use tokio_util::time::DelayQueue; - /// use std::time::Duration; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::new(); - /// - /// delay_queue.insert("hello", Duration::from_secs(10)); - /// delay_queue.reserve(10); - /// - /// assert!(delay_queue.capacity() >= 11); - /// # } - /// ``` - #[track_caller] - pub fn reserve(&mut self, additional: usize) { - assert!( - self.slab.capacity() + additional <= MAX_ENTRIES, - "max queue capacity exceeded" - ); - self.slab.reserve(additional); - } - - /// Returns `true` if there are no items in the queue. - /// - /// Note that this function returns `false` even if all items have not yet - /// expired and a call to `poll` will return `Poll::Pending`. - /// - /// # Examples - /// - /// ``` - /// use tokio_util::time::DelayQueue; - /// use std::time::Duration; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let mut delay_queue = DelayQueue::new(); - /// assert!(delay_queue.is_empty()); - /// - /// delay_queue.insert("hello", Duration::from_secs(5)); - /// assert!(!delay_queue.is_empty()); - /// # } - /// ``` - pub fn is_empty(&self) -> bool { - self.slab.is_empty() - } - - /// Polls the queue, returning the index of the next slot in the slab that - /// should be returned. - /// - /// A slot should be returned when the associated deadline has been reached. - fn poll_idx(&mut self, cx: &mut task::Context<'_>) -> Poll> { - use self::wheel::Stack; - - let expired = self.expired.pop(&mut self.slab); - - if expired.is_some() { - return Poll::Ready(expired); - } - - loop { - if let Some(ref mut delay) = self.delay { - ready!(Pin::new(&mut *delay).poll(cx)).ok(); - - let now = crate::wasm_delay_queue::ms( - delay.deadline() - self.start, - crate::wasm_delay_queue::Round::Down, - ); - - self.wheel_now = now; - } - - // We poll the wheel to get the next value out before finding the next deadline. - let wheel_idx = self.wheel.poll(self.wheel_now, &mut self.slab); - - self.delay = self.next_deadline().map(|when| Box::pin(sleep_until(when))); - - if let Some(idx) = wheel_idx { - return Poll::Ready(Some(idx)); - } - - if self.delay.is_none() { - return Poll::Ready(None); - } - } - } - - fn normalize_deadline(&self, when: Instant) -> u64 { - let when = if when < self.start { - 0 - } else { - crate::wasm_delay_queue::ms(when - self.start, crate::wasm_delay_queue::Round::Up) - }; - - cmp::max(when, self.wheel.elapsed()) - } -} - -// We never put `T` in a `Pin`... -impl Unpin for DelayQueue {} - -impl Default for DelayQueue { - fn default() -> DelayQueue { - DelayQueue::new() - } -} - -impl futures_core::Stream for DelayQueue { - // DelayQueue seems much more specific, where a user may care that it - // has reached capacity, so return those errors instead of panicking. - type Item = Expired; - - fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - DelayQueue::poll_expired(self.get_mut(), cx) - } -} - -impl wheel::Stack for Stack { - type Owned = Key; - type Borrowed = Key; - type Store = SlabStorage; - - fn is_empty(&self) -> bool { - self.head.is_none() - } - - fn push(&mut self, item: Self::Owned, store: &mut Self::Store) { - // Ensure the entry is not already in a stack. - debug_assert!(store[item].next.is_none()); - debug_assert!(store[item].prev.is_none()); - - // Remove the old head entry - let old = self.head.take(); - - if let Some(idx) = old { - store[idx].prev = Some(item); - } - - store[item].next = old; - self.head = Some(item); - } - - fn pop(&mut self, store: &mut Self::Store) -> Option { - if let Some(key) = self.head { - self.head = store[key].next; - - if let Some(idx) = self.head { - store[idx].prev = None; - } - - store[key].next = None; - debug_assert!(store[key].prev.is_none()); - - Some(key) - } else { - None - } - } - - #[track_caller] - fn remove(&mut self, item: &Self::Borrowed, store: &mut Self::Store) { - let key = *item; - assert!(store.contains(item)); - - // Ensure that the entry is in fact contained by the stack - debug_assert!({ - // This walks the full linked list even if an entry is found. - let mut next = self.head; - let mut contains = false; - - while let Some(idx) = next { - let data = &store[idx]; - - if idx == *item { - debug_assert!(!contains); - contains = true; - } - - next = data.next; - } - - contains - }); - - if let Some(next) = store[key].next { - store[next].prev = store[key].prev; - } - - if let Some(prev) = store[key].prev { - store[prev].next = store[key].next; - } else { - self.head = store[key].next; - } - - store[key].next = None; - store[key].prev = None; - } - - fn when(item: &Self::Borrowed, store: &Self::Store) -> u64 { - store[*item].when - } -} - -impl Default for Stack { - fn default() -> Stack { - Stack { - head: None, - _p: PhantomData, - } - } -} - -impl Key { - pub(crate) fn new(index: usize) -> Key { - Key { index } - } -} - -impl KeyInternal { - pub(crate) fn new(index: usize) -> KeyInternal { - KeyInternal { index } - } -} - -impl From for KeyInternal { - fn from(item: Key) -> Self { - KeyInternal::new(item.index) - } -} - -impl From for Key { - fn from(item: KeyInternal) -> Self { - Key::new(item.index) - } -} - -impl Expired { - /// Returns a reference to the inner value. - pub fn get_ref(&self) -> &T { - &self.data - } - - /// Returns a mutable reference to the inner value. - pub fn get_mut(&mut self) -> &mut T { - &mut self.data - } - - /// Consumes `self` and returns the inner value. - pub fn into_inner(self) -> T { - self.data - } - - /// Returns the deadline that the expiration was set to. - pub fn deadline(&self) -> Instant { - self.deadline - } - - /// Returns the key that the expiration is indexed by. - pub fn key(&self) -> Key { - self.key - } -} diff --git a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/mod.rs b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/mod.rs deleted file mode 100644 index fa8b62f561..0000000000 --- a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/mod.rs +++ /dev/null @@ -1,42 +0,0 @@ -use std::time::Duration; - -mod wheel; - -pub mod delay_queue; - -#[doc(inline)] -pub use delay_queue::DelayQueue; - -// ===== Internal utils ===== - -enum Round { - Up, - Down, -} - -/// Convert a `Duration` to milliseconds, rounding up and saturating at -/// `u64::MAX`. -/// -/// The saturating is fine because `u64::MAX` milliseconds are still many -/// million years. -#[inline] -fn ms(duration: Duration, round: Round) -> u64 { - const NANOS_PER_MILLI: u32 = 1_000_000; - const MILLIS_PER_SEC: u64 = 1_000; - - // Round up. - let millis = match round { - Round::Up => (duration.subsec_nanos() + NANOS_PER_MILLI - 1) / NANOS_PER_MILLI, - Round::Down => duration.subsec_millis(), - }; - - duration - .as_secs() - .saturating_mul(MILLIS_PER_SEC) - .saturating_add(u64::from(millis)) -} - -#[inline] -fn sleep_until(deadline: wasm_timer::Instant) -> wasm_timer::Delay { - wasm_timer::Delay::new_at(deadline) -} diff --git a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/level.rs b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/level.rs deleted file mode 100644 index f84d80e265..0000000000 --- a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/level.rs +++ /dev/null @@ -1,253 +0,0 @@ -use crate::wasm_delay_queue::wheel::Stack; - -use std::fmt; - -/// Wheel for a single level in the timer. This wheel contains 64 slots. -pub(crate) struct Level { - level: usize, - - /// Bit field tracking which slots currently contain entries. - /// - /// Using a bit field to track slots that contain entries allows avoiding a - /// scan to find entries. This field is updated when entries are added or - /// removed from a slot. - /// - /// The least-significant bit represents slot zero. - occupied: u64, - - /// Slots - slot: [T; LEVEL_MULT], -} - -/// Indicates when a slot must be processed next. -#[derive(Debug)] -pub(crate) struct Expiration { - /// The level containing the slot. - pub(crate) level: usize, - - /// The slot index. - pub(crate) slot: usize, - - /// The instant at which the slot needs to be processed. - pub(crate) deadline: u64, -} - -/// Level multiplier. -/// -/// Being a power of 2 is very important. -const LEVEL_MULT: usize = 64; - -impl Level { - pub(crate) fn new(level: usize) -> Level { - // Rust's derived implementations for arrays require that the value - // contained by the array be `Copy`. So, here we have to manually - // initialize every single slot. - macro_rules! s { - () => { - T::default() - }; - } - - Level { - level, - occupied: 0, - slot: [ - // It does not look like the necessary traits are - // derived for [T; 64]. - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - s!(), - ], - } - } - - /// Finds the slot that needs to be processed next and returns the slot and - /// `Instant` at which this slot must be processed. - pub(crate) fn next_expiration(&self, now: u64) -> Option { - // Use the `occupied` bit field to get the index of the next slot that - // needs to be processed. - let slot = match self.next_occupied_slot(now) { - Some(slot) => slot, - None => return None, - }; - - // From the slot index, calculate the `Instant` at which it needs to be - // processed. This value *must* be in the future with respect to `now`. - - let level_range = level_range(self.level); - let slot_range = slot_range(self.level); - - // TODO: This can probably be simplified w/ power of 2 math - let level_start = now - (now % level_range); - let deadline = level_start + slot as u64 * slot_range; - - debug_assert!( - deadline >= now, - "deadline={}; now={}; level={}; slot={}; occupied={:b}", - deadline, - now, - self.level, - slot, - self.occupied - ); - - Some(Expiration { - level: self.level, - slot, - deadline, - }) - } - - fn next_occupied_slot(&self, now: u64) -> Option { - if self.occupied == 0 { - return None; - } - - // Get the slot for now using Maths - let now_slot = (now / slot_range(self.level)) as usize; - let occupied = self.occupied.rotate_right(now_slot as u32); - let zeros = occupied.trailing_zeros() as usize; - let slot = (zeros + now_slot) % 64; - - Some(slot) - } - - pub(crate) fn add_entry(&mut self, when: u64, item: T::Owned, store: &mut T::Store) { - let slot = slot_for(when, self.level); - - self.slot[slot].push(item, store); - self.occupied |= occupied_bit(slot); - } - - pub(crate) fn remove_entry(&mut self, when: u64, item: &T::Borrowed, store: &mut T::Store) { - let slot = slot_for(when, self.level); - - self.slot[slot].remove(item, store); - - if self.slot[slot].is_empty() { - // The bit is currently set - debug_assert!(self.occupied & occupied_bit(slot) != 0); - - // Unset the bit - self.occupied ^= occupied_bit(slot); - } - } - - pub(crate) fn pop_entry_slot(&mut self, slot: usize, store: &mut T::Store) -> Option { - let ret = self.slot[slot].pop(store); - - if ret.is_some() && self.slot[slot].is_empty() { - // The bit is currently set - debug_assert!(self.occupied & occupied_bit(slot) != 0); - - self.occupied ^= occupied_bit(slot); - } - - ret - } -} - -impl fmt::Debug for Level { - fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { - fmt.debug_struct("Level") - .field("occupied", &self.occupied) - .finish() - } -} - -fn occupied_bit(slot: usize) -> u64 { - 1 << slot -} - -fn slot_range(level: usize) -> u64 { - LEVEL_MULT.pow(level as u32) as u64 -} - -fn level_range(level: usize) -> u64 { - LEVEL_MULT as u64 * slot_range(level) -} - -/// Convert a duration (milliseconds) and a level to a slot position -fn slot_for(duration: u64, level: usize) -> usize { - ((duration >> (level * 6)) % LEVEL_MULT as u64) as usize -} - -#[cfg(all(test, not(loom)))] -mod test { - use super::*; - - #[test] - fn test_slot_for() { - for pos in 0..64 { - assert_eq!(pos as usize, slot_for(pos, 0)); - } - - for level in 1..5 { - for pos in level..64 { - let a = pos * 64_usize.pow(level as u32); - assert_eq!(pos as usize, slot_for(a as u64, level)); - } - } - } -} diff --git a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/mod.rs b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/mod.rs deleted file mode 100644 index ffa05ab71b..0000000000 --- a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/mod.rs +++ /dev/null @@ -1,315 +0,0 @@ -mod level; -pub(crate) use self::level::Expiration; -use self::level::Level; - -mod stack; -pub(crate) use self::stack::Stack; - -use std::borrow::Borrow; -use std::fmt::Debug; -use std::usize; - -/// Timing wheel implementation. -/// -/// This type provides the hashed timing wheel implementation that backs `Timer` -/// and `DelayQueue`. -/// -/// The structure is generic over `T: Stack`. This allows handling timeout data -/// being stored on the heap or in a slab. In order to support the latter case, -/// the slab must be passed into each function allowing the implementation to -/// lookup timer entries. -/// -/// See `Timer` documentation for some implementation notes. -#[derive(Debug)] -pub(crate) struct Wheel { - /// The number of milliseconds elapsed since the wheel started. - elapsed: u64, - - /// Timer wheel. - /// - /// Levels: - /// - /// * 1 ms slots / 64 ms range - /// * 64 ms slots / ~ 4 sec range - /// * ~ 4 sec slots / ~ 4 min range - /// * ~ 4 min slots / ~ 4 hr range - /// * ~ 4 hr slots / ~ 12 day range - /// * ~ 12 day slots / ~ 2 yr range - levels: Vec>, -} - -/// Number of levels. Each level has 64 slots. By using 6 levels with 64 slots -/// each, the timer is able to track time up to 2 years into the future with a -/// precision of 1 millisecond. -const NUM_LEVELS: usize = 6; - -/// The maximum duration of a delay -const MAX_DURATION: u64 = (1 << (6 * NUM_LEVELS)) - 1; - -#[derive(Debug)] -pub(crate) enum InsertError { - Elapsed, - Invalid, -} - -impl Wheel -where - T: Stack, -{ - /// Create a new timing wheel - pub(crate) fn new() -> Wheel { - let levels = (0..NUM_LEVELS).map(Level::new).collect(); - - Wheel { elapsed: 0, levels } - } - - /// Return the number of milliseconds that have elapsed since the timing - /// wheel's creation. - pub(crate) fn elapsed(&self) -> u64 { - self.elapsed - } - - /// Insert an entry into the timing wheel. - /// - /// # Arguments - /// - /// * `when`: is the instant at which the entry should be fired. It is - /// represented as the number of milliseconds since the creation - /// of the timing wheel. - /// - /// * `item`: The item to insert into the wheel. - /// - /// * `store`: The slab or `()` when using heap storage. - /// - /// # Return - /// - /// Returns `Ok` when the item is successfully inserted, `Err` otherwise. - /// - /// `Err(Elapsed)` indicates that `when` represents an instant that has - /// already passed. In this case, the caller should fire the timeout - /// immediately. - /// - /// `Err(Invalid)` indicates an invalid `when` argument as been supplied. - pub(crate) fn insert( - &mut self, - when: u64, - item: T::Owned, - store: &mut T::Store, - ) -> Result<(), (T::Owned, InsertError)> { - if when <= self.elapsed { - return Err((item, InsertError::Elapsed)); - } else if when - self.elapsed > MAX_DURATION { - return Err((item, InsertError::Invalid)); - } - - // Get the level at which the entry should be stored - let level = self.level_for(when); - - self.levels[level].add_entry(when, item, store); - - debug_assert!({ - self.levels[level] - .next_expiration(self.elapsed) - .map(|e| e.deadline >= self.elapsed) - .unwrap_or(true) - }); - - Ok(()) - } - - /// Remove `item` from the timing wheel. - #[track_caller] - pub(crate) fn remove(&mut self, item: &T::Borrowed, store: &mut T::Store) { - let when = T::when(item, store); - - assert!( - self.elapsed <= when, - "elapsed={}; when={}", - self.elapsed, - when - ); - - let level = self.level_for(when); - - self.levels[level].remove_entry(when, item, store); - } - - /// Instant at which to poll - pub(crate) fn poll_at(&self) -> Option { - self.next_expiration().map(|expiration| expiration.deadline) - } - - /// Advances the timer up to the instant represented by `now`. - pub(crate) fn poll(&mut self, now: u64, store: &mut T::Store) -> Option { - loop { - let expiration = self.next_expiration().and_then(|expiration| { - if expiration.deadline > now { - None - } else { - Some(expiration) - } - }); - - match expiration { - Some(ref expiration) => { - if let Some(item) = self.poll_expiration(expiration, store) { - return Some(item); - } - - self.set_elapsed(expiration.deadline); - } - None => { - // in this case the poll did not indicate an expiration - // _and_ we were not able to find a next expiration in - // the current list of timers. advance to the poll's - // current time and do nothing else. - self.set_elapsed(now); - return None; - } - } - } - } - - /// Returns the instant at which the next timeout expires. - fn next_expiration(&self) -> Option { - // Check all levels - for level in 0..NUM_LEVELS { - if let Some(expiration) = self.levels[level].next_expiration(self.elapsed) { - // There cannot be any expirations at a higher level that happen - // before this one. - debug_assert!(self.no_expirations_before(level + 1, expiration.deadline)); - - return Some(expiration); - } - } - - None - } - - /// Used for debug assertions - fn no_expirations_before(&self, start_level: usize, before: u64) -> bool { - let mut res = true; - - for l2 in start_level..NUM_LEVELS { - if let Some(e2) = self.levels[l2].next_expiration(self.elapsed) { - if e2.deadline < before { - res = false; - } - } - } - - res - } - - /// iteratively find entries that are between the wheel's current - /// time and the expiration time. for each in that population either - /// return it for notification (in the case of the last level) or tier - /// it down to the next level (in all other cases). - pub(crate) fn poll_expiration( - &mut self, - expiration: &Expiration, - store: &mut T::Store, - ) -> Option { - while let Some(item) = self.pop_entry(expiration, store) { - if expiration.level == 0 { - debug_assert_eq!(T::when(item.borrow(), store), expiration.deadline); - - return Some(item); - } else { - let when = T::when(item.borrow(), store); - - let next_level = expiration.level - 1; - - self.levels[next_level].add_entry(when, item, store); - } - } - - None - } - - fn set_elapsed(&mut self, when: u64) { - assert!( - self.elapsed <= when, - "elapsed={:?}; when={:?}", - self.elapsed, - when - ); - - if when > self.elapsed { - self.elapsed = when; - } - } - - fn pop_entry(&mut self, expiration: &Expiration, store: &mut T::Store) -> Option { - self.levels[expiration.level].pop_entry_slot(expiration.slot, store) - } - - fn level_for(&self, when: u64) -> usize { - level_for(self.elapsed, when) - } -} - -fn level_for(elapsed: u64, when: u64) -> usize { - const SLOT_MASK: u64 = (1 << 6) - 1; - - // Mask in the trailing bits ignored by the level calculation in order to cap - // the possible leading zeros - let masked = elapsed ^ when | SLOT_MASK; - - let leading_zeros = masked.leading_zeros() as usize; - let significant = 63 - leading_zeros; - significant / 6 -} - -#[cfg(all(test, not(loom)))] -mod test { - use super::*; - - #[test] - fn test_level_for() { - for pos in 0..64 { - assert_eq!( - 0, - level_for(0, pos), - "level_for({}) -- binary = {:b}", - pos, - pos - ); - } - - for level in 1..5 { - for pos in level..64 { - let a = pos * 64_usize.pow(level as u32); - assert_eq!( - level, - level_for(0, a as u64), - "level_for({}) -- binary = {:b}", - a, - a - ); - - if pos > level { - let a = a - 1; - assert_eq!( - level, - level_for(0, a as u64), - "level_for({}) -- binary = {:b}", - a, - a - ); - } - - if pos < 64 { - let a = a + 1; - assert_eq!( - level, - level_for(0, a as u64), - "level_for({}) -- binary = {:b}", - a, - a - ); - } - } - } - } -} diff --git a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/stack.rs b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/stack.rs deleted file mode 100644 index c87adcafda..0000000000 --- a/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/stack.rs +++ /dev/null @@ -1,28 +0,0 @@ -use std::borrow::Borrow; -use std::cmp::Eq; -use std::hash::Hash; - -/// Abstracts the stack operations needed to track timeouts. -pub(crate) trait Stack: Default { - /// Type of the item stored in the stack - type Owned: Borrow; - - /// Borrowed item - type Borrowed: Eq + Hash; - - /// Item storage, this allows a slab to be used instead of just the heap - type Store; - - /// Returns `true` if the stack is empty - fn is_empty(&self) -> bool; - - /// Push an item onto the stack - fn push(&mut self, item: Self::Owned, store: &mut Self::Store); - - /// Pop an item from the stack - fn pop(&mut self, store: &mut Self::Store) -> Option; - - fn remove(&mut self, item: &Self::Borrowed, store: &mut Self::Store); - - fn when(item: &Self::Borrowed, store: &Self::Store) -> u64; -}