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

wip

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2022-10-04 09:22:41 +01:00
parent f4ea183c1e
commit dc9daea2f6
15 changed files with 2116 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+46 -4
View File
@@ -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"
clients/client-core/Cargo.toml
+6 -2
View File
@@ -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"]
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+38 -9
View File
@@ -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<Box<time::Sleep>>,
#[cfg(target_arch = "wasm32")]
next_delay: Pin<Box<wasm_timer::Delay>>,
/// 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;
}
clients/client-core/src/client/received_buffer.rs
+1 -1
View File
@@ -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;
clients/client-core/src/lib.rs
+12 -5
View File
@@ -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<F>(future: F)
where
F: Future<Output = ()> + '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<F>(future: F)
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
tokio::spawn(future);
}
clients/webassembly/js-example/index.js
+36 -19
View File
@@ -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);
}
clients/webassembly/src/client/mod.rs
+69 -17
View File
@@ -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<SURBEncryptionKey>,
/// Channel used for transforming 'raw' messages into sphinx packets and sending them
/// through the mix network.
input_tx: Option<InputMessageSender>,
// TODO: only temporary
topology: Option<NymTopology>,
@@ -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::<Vec<_>>();
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();
common/nonexhaustive-delayqueue/Cargo.toml
+10
View File
@@ -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"
common/nonexhaustive-delayqueue/src/lib.rs
+20 -3
View File
@@ -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<T> = tokio_util::time::DelayQueue<T>;
#[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<T> = crate::wasm_delay_queue::DelayQueue<T>;
#[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<T> {
common/nonexhaustive-delayqueue/src/wasm_delay_queue/delay_queue.rs
+1240
View File
File diff suppressed because it is too large Load Diff
common/nonexhaustive-delayqueue/src/wasm_delay_queue/mod.rs
+42
View File
@@ -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)
}
common/nonexhaustive-delayqueue/src/wasm_delay_queue/wheel/level.rs
+253
View File
@@ -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<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
@@ -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<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
@@ -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<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;
}
explorer-api/Cargo.toml
-4
View File
@@ -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" }