GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

removed dependency on wasm-timer fork (#3733)

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2023-08-03 11:01:03 +01:00
committed by GitHub
parent 5dacf0c8f8
commit b66c4cf0bb
19 changed files with 102 additions and 2027 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+17 -22
View File
@@ -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"
Cargo.toml
+3
View File
@@ -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"
clients/webassembly/Cargo.lock
Generated
+29 -24
View File
@@ -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"
clients/webassembly/Cargo.toml
+1 -1
View File
@@ -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" }
clients/webassembly/src/tester/ephemeral_receiver.rs
+2 -1
View File
@@ -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! {
common/client-core/Cargo.toml
+3 -3
View File
@@ -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"
common/client-core/src/client/cover_traffic_stream.rs
+8 -29
View File
@@ -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<R>
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<Box<time::Sleep>>,
#[cfg(target_arch = "wasm32")]
next_delay: Pin<Box<wasm_timer::Delay>>,
next_delay: Pin<Box<Sleep>>,
/// 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<OsRng> {
) -> 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<OsRng> {
}
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;
}
common/client-core/src/client/helpers/wasm.rs
+2 -3
View File
@@ -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 {
common/client-core/src/client/real_messages_control/real_traffic_stream.rs
+8 -24
View File
@@ -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<Pin<Box<time::Sleep>>>,
#[cfg(target_arch = "wasm32")]
next_delay: Option<Pin<Box<wasm_timer::Delay>>>,
next_delay: Option<Pin<Box<Sleep>>>,
// 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
common/client-core/src/init/helpers.rs
+7 -9
View File
@@ -27,13 +27,17 @@ use tokio_tungstenite::connect_async;
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
#[cfg(not(target_arch = "wasm32"))]
type WsConn = WebSocketStream<MaybeTlsStream<TcpStream>>;
#[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<GatewayWithLatency,
Ok::<(), ClientCoreError>(())
};
// 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...")
common/client-libs/gateway-client/Cargo.toml
+3 -3
View File
@@ -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.
common/client-libs/gateway-client/src/client.rs
+6 -19
View File
@@ -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<C, St> GatewayClient<C, St> {
}
#[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<C, St> GatewayClient<C, St> {
_ => 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() => {
common/nonexhaustive-delayqueue/Cargo.toml
+9 -9
View File
@@ -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"
common/nonexhaustive-delayqueue/src/lib.rs
+4 -8
View File
@@ -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<T> = tokio_util::time::DelayQueue<T>;
@@ -23,13 +19,13 @@ pub type QueueKey = tokio_util::time::delay_queue::Key;
use tokio::time::Instant;
#[cfg(target_arch = "wasm32")]
type DelayQueue<T> = crate::wasm_delay_queue::DelayQueue<T>;
type DelayQueue<T> = wasmtimer::tokio_util::DelayQueue<T>;
#[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<T> {
common/nonexhaustive-delayqueue/src/wasm_delay_queue/delay_queue.rs
-1234
View File
File diff suppressed because it is too large Load Diff
common/nonexhaustive-delayqueue/src/wasm_delay_queue/mod.rs
-42
View File
@@ -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)
}
common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/level.rs
-253
View File
@@ -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<T> {
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<T: Stack> Level<T> {
pub(crate) fn new(level: usize) -> Level<T> {
// 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<Expiration> {
// 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<usize> {
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<T::Owned> {
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<T> fmt::Debug for Level<T> {
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));
}
}
}
}
common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/mod.rs
-315
View File
@@ -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<T> {
/// 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<Level<T>>,
}
/// 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<T> Wheel<T>
where
T: Stack,
{
/// Create a new timing wheel
pub(crate) fn new() -> Wheel<T> {
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<u64> {
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<T::Owned> {
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<Expiration> {
// 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<T::Owned> {
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<T::Owned> {
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
);
}
}
}
}
}
common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/stack.rs
-28
View File
@@ -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<Self::Borrowed>;
/// 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<Self::Owned>;
fn remove(&mut self, item: &Self::Borrowed, store: &mut Self::Store);
fn when(item: &Self::Borrowed, store: &Self::Store) -> u64;
}