diff --git a/Cargo.lock b/Cargo.lock index c65c78fbd9..e215e6270c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -651,6 +651,7 @@ dependencies = [ "config", "crypto", "dirs", + "fluvio-wasm-timer", "futures", "gateway-client", "gateway-requests", @@ -663,11 +664,14 @@ dependencies = [ "serde", "sled", "tap", + "task", "tempfile", "thiserror", + "tokio", "topology", "url", "validator-client", + "wasm-bindgen-futures", ] [[package]] @@ -1719,6 +1723,7 @@ dependencies = [ "isocountry", "itertools", "log", + "maxminddb", "mixnet-contract-common", "network-defaults", "okapi", @@ -2708,6 +2713,15 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879d54834c8c76457ef4293a689b2a8c59b076067ad77b15efafbb05f92a592b" +[[package]] +name = "ipnetwork" +version = "0.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4088d739b183546b239688ddbc79891831df421773df95e236daf7867866d355" +dependencies = [ + "serde", +] + [[package]] name = "ipnetwork" version = "0.20.0" @@ -2933,6 +2947,18 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "73cbba799671b762df5a175adf59ce145165747bb891505c43d09aefbbf38beb" +[[package]] +name = "maxminddb" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe2ba61113f9f7a9f0e87c519682d39c43a6f3f79c2cc42c3ba3dda83b1fa334" +dependencies = [ + "ipnetwork 0.18.0", + "log", + "memchr", + "serde", +] + [[package]] name = "maybe-uninit" version = "2.0.0" @@ -3117,9 +3143,12 @@ dependencies = [ name = "nonexhaustive-delayqueue" version = "0.1.0" dependencies = [ + "futures-core", + "slab", "tokio", "tokio-stream", "tokio-util 0.7.4", + "wasm-timer", ] [[package]] @@ -3360,7 +3389,7 @@ dependencies = [ "clap 3.2.21", "dirs", "futures", - "ipnetwork", + "ipnetwork 0.20.0", "log", "network-defaults", "nymsphinx", @@ -6014,9 +6043,9 @@ dependencies = [ [[package]] name = "tokio" -version = "1.21.1" +version = "1.21.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0020c875007ad96677dcc890298f4b942882c5d4eb7cc8f439fc3bf813dc9c95" +checksum = "a9e03c497dc955702ba729190dc4aac6f2a0ce97f913e5b1b5912fc5039d9099" dependencies = [ "autocfg 1.1.0", "bytes", @@ -6024,7 +6053,6 @@ dependencies = [ "memchr", "mio", "num_cpus", - "once_cell", "parking_lot 0.12.1", "pin-project-lite", "signal-hook-registry", @@ -6814,6 +6842,20 @@ version = "0.2.78" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0237232789cf037d5480773fe568aac745bfe2afbc11a863e97901780a6b47cc" +[[package]] +name = "wasm-timer" +version = "0.2.5" +source = "git+https://github.com/jstuczyn/wasm-timer?rev=2859241251dd00c54a9c99b74727b99bd7feffb4#2859241251dd00c54a9c99b74727b99bd7feffb4" +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" diff --git a/clients/client-core/Cargo.toml b/clients/client-core/Cargo.toml index db1d03ba45..125e6e5fdc 100644 --- a/clients/client-core/Cargo.toml +++ b/clients/client-core/Cargo.toml @@ -25,7 +25,6 @@ gateway-requests = { path = "../../gateway/gateway-requests" } nonexhaustive-delayqueue = { path = "../../common/nonexhaustive-delayqueue" } nymsphinx = { path = "../../common/nymsphinx" } pemstore = { path = "../../common/pemstore" } -#task = { path = "../../common/task" } topology = { path = "../../common/topology" } validator-client = { path = "../../common/client-libs/validator-client", default-features = false } tap = "1.0.1" @@ -35,6 +34,11 @@ tokio = { version = "1.21.2", features = ["time", "macros"]} [target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen-futures] version = "0.4" +[target."cfg(target_arch = \"wasm32\")".dependencies.fluvio-wasm-timer] +version = "0.2.5" + +[target."cfg(not(target_arch = \"wasm32\"))".dependencies.task] +path = "../../common/task" [dev-dependencies] tempfile = "3.1.0" @@ -43,5 +47,5 @@ tempfile = "3.1.0" #wasm = ["gateway-client/wasm"] default = ["reply-surb"] wasm = [] -#coconut = ["gateway-client/coconut", "gateway-requests/coconut"] +coconut = ["gateway-client/coconut", "gateway-requests/coconut"] reply-surb = ["sled"] \ No newline at end of file diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs index 72c787bbef..01e7ddf3ac 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -19,8 +19,13 @@ use std::collections::VecDeque; use std::pin::Pin; use std::sync::Arc; use std::time::Duration; + +#[cfg(not(target_arch = "wasm32"))] use tokio::time; +#[cfg(target_arch = "wasm32")] +use fluvio_wasm_timer as wasm_timer; + #[cfg(not(target_arch = "wasm32"))] use task::ShutdownListener; @@ -65,8 +70,12 @@ 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>, + /// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them /// out to the network without any further delays. mix_tx: BatchMixMessageSender, @@ -131,13 +140,22 @@ where // we know it's time to send a message, so let's prepare delay for the next one // Get the `now` by looking at the current `delay` deadline let avg_delay = self.config.average_message_sending_delay; - let now = self.next_delay.deadline(); + let next_poisson_delay = sample_poisson_duration(&mut self.rng, avg_delay); // The next interval value is `next_poisson_delay` after the one that just // yielded. - let next = now + next_poisson_delay; - self.next_delay.as_mut().reset(next); + #[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); + } // check if we have anything immediately available if let Some(real_available) = self.received_buffer.pop_front() { @@ -183,11 +201,17 @@ where topology_access: TopologyAccessor, #[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener, ) -> Self { + #[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())); + OutQueueControl { config, ack_key, sent_notifier, - next_delay: Box::pin(time::sleep(Default::default())), + next_delay, mix_tx, real_receiver, our_full_destination, @@ -274,13 +298,18 @@ where // Send messages at certain rate and if no real traffic is available, send cover message. async fn run_normal_out_queue(&mut self) { // we should set initial delay only when we actually start the stream - self.next_delay = Box::pin(time::sleep(sample_poisson_duration( - &mut self.rng, - self.config.average_message_sending_delay, - ))); + let sampled = + sample_poisson_duration(&mut self.rng, self.config.average_message_sending_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)); + + self.next_delay = next_delay; // TODO: fix it for non-wasm - while let Some(next_message) = self.next().await { self.on_message(next_message).await; } diff --git a/clients/client-core/src/client/received_buffer.rs b/clients/client-core/src/client/received_buffer.rs index f6a064112a..04135ed15d 100644 --- a/clients/client-core/src/client/received_buffer.rs +++ b/clients/client-core/src/client/received_buffer.rs @@ -18,7 +18,7 @@ use std::collections::HashSet; use std::sync::atomic::Ordering; use std::sync::Arc; -#[cfg(not(target_arch = "wasm32"))] +#[cfg(feature = "reply-surb")] use crate::client::reply_key_storage::ReplyKeyStorage; #[cfg(not(target_arch = "wasm32"))] use task::ShutdownListener; diff --git a/clients/client-core/src/lib.rs b/clients/client-core/src/lib.rs index e1a2f10345..9d23410a50 100644 --- a/clients/client-core/src/lib.rs +++ b/clients/client-core/src/lib.rs @@ -5,14 +5,21 @@ pub mod config; pub mod error; pub mod init; -// for now we're losing the output but we never really cared about it anyway +// TODO: move those to separate lower modules and conditionally re-export them accordingly to make intellij happier about name clash + +#[cfg(target_arch = "wasm32")] pub(crate) fn spawn_future(future: F) where F: Future + 'static, { - #[cfg(not(target_arch = "wasm32"))] - tokio::spawn(future); - - #[cfg(target_arch = "wasm32")] wasm_bindgen_futures::spawn_local(future); } + +#[cfg(not(target_arch = "wasm32"))] +pub(crate) fn spawn_future(future: F) +where + F: Future + Send + 'static, + F::Output: Send + 'static, +{ + tokio::spawn(future); +} diff --git a/clients/webassembly/js-example/index.js b/clients/webassembly/js-example/index.js index 3aa18f2148..50a060b27a 100644 --- a/clients/webassembly/js-example/index.js +++ b/clients/webassembly/js-example/index.js @@ -12,12 +12,38 @@ // See the License for the specific language governing permissions and // limitations under the License. -import { - NymClient, - set_panic_hook -} from "@nymproject/nym-client-wasm" +import {NymClient, set_panic_hook} from "@nymproject/nym-client-wasm" -// current limitation of rust-wasm for async stuff : ( +class ClientWrapper { + constructor(validator) { + this.rustClient = new NymClient(validator); + this.rustClient.set_on_message(this.on_message); + this.rustClient.set_on_gateway_connect(this.on_connect); + } + + selfAddress = () => { + const self_address = this.rustClient.self_address(); + console.log(self_address) + } + + on_message = (msg) => displayReceived(msg); + on_connect = () => { + console.log("Established (and authenticated) gateway connection!"); + } + + start = async () => { + // this is current limitation of wasm in rust - for async methods you can't take self my reference... + // I'm trying to figure out if I can somehow hack my way around it, but for time being you have to re-assign + // the object (it's the same one) + this.rustClient = await this.rustClient.start() + } + + sendMessage = async (recipient, message) => { + this.rustClient = await this.rustClient.send_message(recipient, message) + } +} + +// // current limitation of rust-wasm for async stuff : ( let client = null async function main() { @@ -27,20 +53,10 @@ async function main() { // validator server we will use to get topology from const validator = "https://validator.nymtech.net/api"; //"http://localhost:8081"; - client = new NymClient(validator); + client = new ClientWrapper(validator); + await client.start(); - const on_message = (msg) => displayReceived(msg); - const on_connect = () => console.log("Established (and authenticated) gateway connection!"); - - client.set_on_message(on_message); - client.set_on_gateway_connect(on_connect); - - // this is current limitation of wasm in rust - for async methods you can't take self my reference... - // I'm trying to figure out if I can somehow hack my way around it, but for time being you have to re-assign - // the object (it's the same one) - client = await client.start(); - - const self_address = client.self_address(); + const self_address = client.rustClient.self_address(); displaySenderAddress(self_address); const sendButton = document.querySelector('#send-button'); @@ -59,7 +75,8 @@ async function main() { async function sendMessageTo() { const message = document.getElementById("message").value; const recipient = document.getElementById("recipient").value; - client = await client.send_message(message, recipient); + + await client.sendMessage(message, recipient); displaySend(message); } diff --git a/clients/webassembly/src/client/mod.rs b/clients/webassembly/src/client/mod.rs index f49d7c1d07..a3ba06e3ce 100644 --- a/clients/webassembly/src/client/mod.rs +++ b/clients/webassembly/src/client/mod.rs @@ -4,12 +4,14 @@ use client_core::client::cover_traffic_stream::LoopCoverTrafficStream; use client_core::client::inbound_messages::InputMessage; use client_core::client::inbound_messages::InputMessageReceiver; +use client_core::client::inbound_messages::InputMessageSender; use client_core::client::key_manager::KeyManager; use client_core::client::mix_traffic::BatchMixMessageReceiver; use client_core::client::mix_traffic::BatchMixMessageSender; use client_core::client::mix_traffic::MixTrafficController; use client_core::client::real_messages_control; use client_core::client::real_messages_control::RealMessagesController; +use client_core::client::received_buffer::ReceivedBufferMessage; use client_core::client::received_buffer::ReceivedBufferRequestReceiver; use client_core::client::received_buffer::ReceivedMessagesBufferController; use client_core::client::topology_control::TopologyAccessor; @@ -17,6 +19,7 @@ use client_core::client::topology_control::TopologyRefresher; use client_core::client::topology_control::TopologyRefresherConfig; use crypto::asymmetric::{encryption, identity}; use futures::channel::mpsc; +use futures::StreamExt; use gateway_client::AcknowledgementReceiver; use gateway_client::AcknowledgementSender; use gateway_client::GatewayClient; @@ -41,7 +44,7 @@ pub(crate) mod received_processor; const ACK_WAIT_MULTIPLIER: f64 = 1.5; const ACK_WAIT_ADDITION: Duration = Duration::from_millis(1_500); const LOOP_COVER_STREAM_AVERAGE_DELAY: Duration = Duration::from_millis(200); -const MESSAGE_STREAM_AVERAGE_DELAY: Duration = Duration::from_millis(20); +const MESSAGE_STREAM_AVERAGE_DELAY: Duration = Duration::from_millis(10); const AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(50); const AVERAGE_ACK_DELAY: Duration = Duration::from_millis(50); const TOPOLOGY_REFRESH_RATE: Duration = Duration::from_secs(5 * 60); @@ -62,6 +65,9 @@ pub struct NymClient { // TODO: this should be stored somewhere persistently // received_keys: HashSet, + /// Channel used for transforming 'raw' messages into sphinx packets and sending them + /// through the mix network. + input_tx: Option, // TODO: only temporary topology: Option, @@ -95,6 +101,7 @@ impl NymClient { on_message: None, on_gateway_connect: None, disabled_credentials_mode: true, + input_tx: None, } } @@ -125,8 +132,8 @@ impl NymClient { } pub fn self_address(&self) -> String { - return "foomp".into(); - // self.as_mix_recipient().to_string() + // return "foomp".into(); + self.as_mix_recipient().to_string() } // future constantly pumping loop cover traffic at some specified average rate @@ -138,16 +145,16 @@ impl NymClient { ) { console_log!("Starting loop cover traffic stream..."); - LoopCoverTrafficStream::new( - self.key_manager.ack_key(), - AVERAGE_ACK_DELAY, - AVERAGE_PACKET_DELAY, - LOOP_COVER_STREAM_AVERAGE_DELAY, - mix_tx, - self.as_mix_recipient(), - topology_accessor, - ) - .start(); + // LoopCoverTrafficStream::new( + // self.key_manager.ack_key(), + // AVERAGE_ACK_DELAY, + // AVERAGE_PACKET_DELAY, + // LOOP_COVER_STREAM_AVERAGE_DELAY, + // mix_tx, + // self.as_mix_recipient(), + // topology_accessor, + // ) + // .start(); } fn start_real_traffic_controller( @@ -232,7 +239,7 @@ impl NymClient { self.key_manager.identity_keypair(), gateway_identity, gateway_owner, - Some(self.key_manager.gateway_shared_key()), + None, mixnet_message_sender, ack_sender, GATEWAY_RESPONSE_TIMEOUT, @@ -241,10 +248,11 @@ impl NymClient { gateway_client.set_disabled_credentials_mode(self.disabled_credentials_mode); - gateway_client + let shared_keys = gateway_client .authenticate_and_start() .await .expect("could not authenticate and start up the gateway connection"); + self.key_manager.insert_gateway_shared_key(shared_keys); match self.on_gateway_connect.as_ref() { Some(callback) => { @@ -284,7 +292,7 @@ impl NymClient { } console_log!("Starting topology refresher..."); - topology_refresher.start(); + // topology_refresher.start(); } // controller for sending sphinx packets to mixnet (either real traffic or cover traffic) @@ -351,6 +359,29 @@ impl NymClient { ); self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender); + + self.input_tx = Some(input_sender); + + // TEMP + spawn_local(async move { + let (reconstructed_sender, mut reconstructed_receiver) = mpsc::unbounded(); + + // tell the buffer to start sending stuff to us + received_buffer_request_sender + .unbounded_send(ReceivedBufferMessage::ReceiverAnnounce( + reconstructed_sender, + )) + .expect("the buffer request failed!"); + + // self.receive_tx = Some(reconstructed_receiver); + // self.input_tx = Some(input_sender); + + while let Some(new) = reconstructed_receiver.next().await { + console_log!("received: len {:?}", new[0].message.len()); + console_log!("received: {:?}", new); + } + }); + self } @@ -424,7 +455,28 @@ impl NymClient { pub async fn send_message(mut self, message: String, recipient: String) -> Self { console_log!("Sending {} to {}", message, recipient); - todo!() + let message_bytes = message.into_bytes(); + + let new_message = std::iter::repeat(message_bytes) + .flatten() + .take(1000000) + .collect::>(); + + let message_bytes = new_message; + + let recipient = Recipient::try_from_base58_string(recipient).unwrap(); + + let input_msg = InputMessage::new_fresh(recipient, message_bytes, false); + + self.input_tx + .as_ref() + .expect("start method was not called before!") + .unbounded_send(input_msg) + .unwrap(); + + self + + // todo!() // let message_bytes = message.into_bytes(); // let recipient = Recipient::try_from_base58_string(recipient).unwrap(); diff --git a/common/nonexhaustive-delayqueue/Cargo.toml b/common/nonexhaustive-delayqueue/Cargo.toml index 9fcfedde36..3880a0eda2 100644 --- a/common/nonexhaustive-delayqueue/Cargo.toml +++ b/common/nonexhaustive-delayqueue/Cargo.toml @@ -10,3 +10,13 @@ edition = "2021" tokio = { version = "1.21.2", features = [] } tokio-stream = "0.1.9" # this one seems to be a thing until `Stream` trait is stabilised in stdlib tokio-util = { version = "0.7.3", features = ["time"] } + +[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-timer] +git = "https://github.com/jstuczyn/wasm-timer" +rev = "2859241251dd00c54a9c99b74727b99bd7feffb4" + +[target."cfg(target_arch = \"wasm32\")".dependencies.slab] +version = "0.4.4" + +[target."cfg(target_arch = \"wasm32\")".dependencies.futures-core] +version = "0.3.0" \ No newline at end of file diff --git a/common/nonexhaustive-delayqueue/src/lib.rs b/common/nonexhaustive-delayqueue/src/lib.rs index 11c1d608c3..4d40aef609 100644 --- a/common/nonexhaustive-delayqueue/src/lib.rs +++ b/common/nonexhaustive-delayqueue/src/lib.rs @@ -4,13 +4,30 @@ use std::pin::Pin; use std::task::{Context, Poll, Waker}; use std::time::Duration; -use tokio::time::Instant; use tokio_stream::Stream; -use tokio_util::time::{delay_queue, DelayQueue}; pub use tokio::time::error::Error as TimerError; + +#[cfg(target_arch = "wasm32")] +mod wasm_delay_queue; + +#[cfg(not(target_arch = "wasm32"))] +type DelayQueue = tokio_util::time::DelayQueue; +#[cfg(not(target_arch = "wasm32"))] pub use tokio_util::time::delay_queue::Expired; -pub type QueueKey = delay_queue::Key; +#[cfg(not(target_arch = "wasm32"))] +pub type QueueKey = tokio_util::time::delay_queue::Key; +#[cfg(not(target_arch = "wasm32"))] +use tokio::time::Instant; + +#[cfg(target_arch = "wasm32")] +type DelayQueue = crate::wasm_delay_queue::DelayQueue; +#[cfg(target_arch = "wasm32")] +pub use crate::wasm_delay_queue::delay_queue::Expired; +#[cfg(target_arch = "wasm32")] +pub type QueueKey = crate::wasm_delay_queue::delay_queue::Key; +#[cfg(target_arch = "wasm32")] +use wasm_timer::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 new file mode 100644 index 0000000000..1101fd1de7 --- /dev/null +++ b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/delay_queue.rs @@ -0,0 +1,1240 @@ +//! A queue of delayed elements. +//! +//! See [`DelayQueue`] for more details. +//! +//! [`DelayQueue`]: struct@DelayQueue + +use crate::wasm_delay_queue::wheel::{self, Wheel}; + +use futures_core::ready; +// use tokio::time::{sleep_until, Duration, Instant, Sleep}; +use crate::wasm_delay_queue::sleep_until; +use core::ops::{Index, IndexMut}; +use slab::Slab; +use std::cmp; +use std::collections::HashMap; +use std::convert::From; +use std::fmt; +use std::fmt::Debug; +use std::future::Future; +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 { + // TODO: i dont like that + // if !delay.is_elapsed() { + // ready!(Pin::new(&mut *delay).poll(cx)); + // } + + 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 new file mode 100644 index 0000000000..fa8b62f561 --- /dev/null +++ b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/mod.rs @@ -0,0 +1,42 @@ +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 new file mode 100644 index 0000000000..f84d80e265 --- /dev/null +++ b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/level.rs @@ -0,0 +1,253 @@ +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 new file mode 100644 index 0000000000..ffa05ab71b --- /dev/null +++ b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/mod.rs @@ -0,0 +1,315 @@ +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 new file mode 100644 index 0000000000..c87adcafda --- /dev/null +++ b/common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/stack.rs @@ -0,0 +1,28 @@ +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; +} diff --git a/explorer-api/Cargo.toml b/explorer-api/Cargo.toml index 690e4704bd..296ffab936 100644 --- a/explorer-api/Cargo.toml +++ b/explorer-api/Cargo.toml @@ -22,12 +22,8 @@ schemars = { version = "0.8", features = ["preserve_order"] } serde = "1.0.126" serde_json = "1.0.66" thiserror = "1.0.29" -<<<<<<< Updated upstream -tokio = {version = "1.19.1", features = ["full"] } -======= tokio = {version = "1.21.2", features = ["full"] } maxminddb = "0.23.0" ->>>>>>> Stashed changes mixnet-contract-common = { path = "../common/cosmwasm-smart-contracts/mixnet-contract" } network-defaults = { path = "../common/network-defaults" }