Removed NextID

- replace i with UUID
- remove the need to increment everytime
- depend on key

Author: StaticallyTypedAnxiety

src/engine.rs

@@ -1,12 +1,14 @@
+use crate::Timer;
 use crate::{bindings::wasi::io::poll::Pollable, poll_tasks::PollTasks};
 use crate::{io::timer::TIMERS, poll_tasks::EventWithWaker};
+use dashmap::DashMap;
 use futures::pin_mut;
 use lazy_static::lazy_static;
 use std::{
     future::Future,
     pin::Pin,
     sync::{
-        atomic::{AtomicBool, AtomicU32, Ordering},
+        atomic::{AtomicBool, Ordering},
         Arc, Mutex,
     },
     task::{Context, Wake, Waker},
@@ -16,8 +18,6 @@ lazy_static! {
     pub static ref REACTOR: Mutex<Reactor<'static>> = Mutex::new(Reactor::default());
 
 }
-
-pub(crate) static NEXT_ID: AtomicU32 = AtomicU32::new(0);
 /// The async engine instance
 pub struct WasmRuntimeAsyncEngine;
 
@@ -37,6 +37,7 @@ impl<'a> Task<'a> {
 pub struct Reactor<'a> {
     events: PollTasks,
     future_tasks: Vec<Task<'a>>, //right now the engine holds the tasks but depending
+    timers: DashMap<String, EventWithWaker<Timer>>,
 }
 
 impl<'a> Reactor<'a> {
@@ -49,8 +50,13 @@ impl<'a> Reactor<'a> {
         self.events.contains(key)
     }
 
+    //checks if timer is pollable
+    pub fn is_timer_pollable(&self, key: &str) -> bool {
+        self.timers.contains_key(key)
+    }
+
     //polls event queue to see if any of the events are readycar
-    pub fn wait(&mut self) {
+    pub fn wait_for_io(&mut self) {
         self.events.wait_for_pollables();
     }
 
@@ -62,6 +68,26 @@ impl<'a> Reactor<'a> {
     pub fn is_empty(&self) -> bool {
         self.events.is_empty() && self.future_tasks.is_empty()
     }
+
+    pub(crate) fn update_timers(&self) {
+        self.timers.iter_mut().for_each(|mut cell| {
+            cell.0.update_elapsed();
+            if cell.0.elapsed() {
+                cell.1.wake_by_ref();
+            }
+        });
+    }
+
+    pub(crate) fn timer_has_elapsed(&self, timer_key: &str) -> bool {
+        self.timers
+            .get(timer_key)
+            .map(|s| s.0.elapsed())
+            .unwrap_or_default()
+    }
+
+    pub(crate) fn register_timer(&mut self, timer_name: String, event: EventWithWaker<Timer>) {
+        self.timers.insert(timer_name, event);
+    }
 }
 
 impl WasmRuntimeAsyncEngine {
@@ -75,14 +101,9 @@ impl WasmRuntimeAsyncEngine {
         let mut future_tasks = Vec::new();
         future_tasks.push(task);
         loop {
-            TIMERS.iter_mut().for_each(|mut cell| {
-                cell.0.update_elapsed();
-                if cell.0.elapsed() {
-                    cell.1.wake_by_ref();
-                }
-            });
             let mut reactor = REACTOR.lock().unwrap();
-            reactor.wait();
+            reactor.update_timers();
+            reactor.wait_for_io();
             reactor.future_tasks.retain_mut(|task_info| {
                 if task_info.waker.should_wake() {
                     task_info.waker.reset();

src/io/net.rs

@@ -1,3 +1,5 @@
+use uuid::Uuid;
+
 use crate::{
     bindings::wasi::{
         io::{
@@ -11,7 +13,7 @@ use crate::{
             tcp_create_socket::{create_tcp_socket, ErrorCode},
         },
     },
-    engine::{NEXT_ID, REACTOR},
+    engine::REACTOR,
 };
 use std::io::ErrorKind;
 use std::net::IpAddr;
@@ -59,10 +61,7 @@ impl TcpStream {
     pub async fn connect<T: Into<IpAddress>>(&mut self, address: T, port: u16) -> IOResult<()> {
         let connect_future = ConnectionFuture {
             stream: self,
-            async_key: format!(
-                "socket-connection={}",
-                NEXT_ID.load(std::sync::atomic::Ordering::Relaxed)
-            ),
+            async_key: format!("socket-connection={}", Uuid::new_v4()),
             address: address.into(),
             port,
         };
@@ -160,7 +159,6 @@ impl<'a> Future for ConnectionFuture<'a> {
         let this = self.get_mut();
         if !REACTOR.lock().unwrap().is_pollable(&this.async_key) {
             this.stream.start_connect(this.address, this.port)?;
-            NEXT_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
             REACTOR.lock().unwrap().register(
                 this.async_key.clone(),
                 (this.stream.pollable.clone(), cx.waker().clone()),

src/io/timer.rs

@@ -1,11 +1,11 @@
+use crate::engine::REACTOR;
+use crate::poll_tasks::EventWithWaker;
 use dashmap::DashMap;
 use lazy_static::lazy_static;
 use std::future::Future;
 use std::time::Duration;
 use std::time::Instant;
-
-use crate::engine::NEXT_ID;
-use crate::poll_tasks::EventWithWaker;
+use uuid::Uuid;
 lazy_static! {
     pub static ref TIMERS: DashMap<u32, EventWithWaker<Timer>> = DashMap::new();
 }
@@ -21,7 +21,7 @@ impl Timer {
     /// create a timer that resolves once it elapses
     pub async fn sleep(until: std::time::Duration) {
         let timeout = TimeoutFuture {
-            id: None,
+            timer_key: format!("sleep-{}", Uuid::new_v4()),
             timer: Self {
                 at: Instant::now(),
                 until,
@@ -36,7 +36,6 @@ impl Timer {
             .checked_duration_since(self.at)
             .map(|s| s > self.until)
             .unwrap_or_default();
-        println!("elapsed {elapsed}");
         self.elapsed = elapsed;
     }
 
@@ -46,7 +45,7 @@ impl Timer {
 }
 
 struct TimeoutFuture {
-    id: Option<u32>,
+    timer_key: String,
     timer: Timer,
 }
 
@@ -57,19 +56,17 @@ impl Future for TimeoutFuture {
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Self::Output> {
         let this = self.get_mut();
-        match this.id {
-            Some(id) => {
-                let has_elapsed = TIMERS.get(&id).map(|s| s.0.elapsed()).unwrap_or_default();
-                if has_elapsed {
-                    return std::task::Poll::Ready(());
-                }
-            }
-            None => {
-                let id = NEXT_ID.load(std::sync::atomic::Ordering::Relaxed);
-                TIMERS.insert(id, (this.timer.clone(), cx.waker().clone()));
-                NEXT_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-                let _ = this.id.insert(id);
+        let mut reactor = REACTOR.lock().unwrap();
+        if reactor.is_timer_pollable(&this.timer_key) {
+            let has_elapsed = reactor.timer_has_elapsed(&this.timer_key);
+            if has_elapsed {
+                return std::task::Poll::Ready(());
             }
+        } else {
+            reactor.register_timer(
+                this.timer_key.clone(),
+                (this.timer.clone(), cx.waker().clone()),
+            );
         }
 
         std::task::Poll::Pending