Created a basic async runtime.

rust/lib/mammoth/src/lib.rs

@@ -16,5 +16,6 @@ pub mod mem;
 pub mod port;
 pub mod sync;
 pub mod syscall;
+pub mod task;
 pub mod thread;
 pub mod zion;

rust/lib/mammoth/src/task/mod.rs

@@ -0,0 +1,125 @@
+use core::{
+    future::Future,
+    pin::Pin,
+    sync::atomic::{AtomicU64, Ordering},
+    task::{Context, Poll, Waker},
+};
+
+use alloc::{
+    boxed::Box,
+    collections::{BTreeMap, VecDeque},
+    sync::Arc,
+    task::Wake,
+};
+
+use crate::{sync::Mutex, syscall};
+
+#[derive(PartialEq, Eq, PartialOrd, Ord, Copy, Clone)]
+struct TaskId(u64);
+
+impl TaskId {
+    pub fn new() -> TaskId {
+        static NEXT_ID: AtomicU64 = AtomicU64::new(0);
+        TaskId(NEXT_ID.fetch_add(1, Ordering::Relaxed))
+    }
+}
+
+pub struct Task {
+    id: TaskId,
+    // FIXME: This only needs to be sync because of the CURRENT_EXECUTOR
+    // needing to be shared between threads.
+    future: Pin<Box<dyn Future<Output = ()> + Sync>>,
+}
+
+impl Task {
+    pub fn new(future: impl Future<Output = ()> + Sync + 'static) -> Task {
+        Task {
+            id: TaskId::new(),
+            future: Box::pin(future),
+        }
+    }
+
+    pub fn poll(&mut self, context: &mut Context) -> Poll<()> {
+        self.future.as_mut().poll(context)
+    }
+}
+
+struct TaskWaker {
+    task_id: TaskId,
+    task_queue: Arc<Mutex<VecDeque<TaskId>>>,
+}
+
+impl TaskWaker {
+    fn new(task_id: TaskId, task_queue: Arc<Mutex<VecDeque<TaskId>>>) -> Waker {
+        Waker::from(Arc::new(TaskWaker {
+            task_id,
+            task_queue,
+        }))
+    }
+    fn wake_task(&self) {
+        self.task_queue.lock().push_back(self.task_id);
+    }
+}
+
+impl Wake for TaskWaker {
+    fn wake(self: Arc<Self>) {
+        self.wake_task();
+    }
+
+    fn wake_by_ref(self: &Arc<Self>) {
+        self.wake_task();
+    }
+}
+
+pub struct Executor {
+    tasks: BTreeMap<TaskId, Task>,
+    // TODO: Consider a better datastructure for this.
+    task_queue: Arc<Mutex<VecDeque<TaskId>>>,
+    waker_cache: BTreeMap<TaskId, Waker>,
+}
+
+impl Executor {
+    pub fn new() -> Executor {
+        Executor {
+            tasks: BTreeMap::new(),
+            task_queue: Arc::new(Mutex::new(VecDeque::new())),
+            waker_cache: BTreeMap::new(),
+        }
+    }
+
+    pub fn spawn(&mut self, task: Task) {
+        let task_id = task.id;
+        if self.tasks.insert(task_id, task).is_some() {
+            panic!("Task is already existed in executor map");
+        }
+        self.task_queue.lock().push_back(task_id);
+    }
+
+    fn run_ready_tasks(&mut self) {
+        while let Some(task_id) = self.task_queue.lock().pop_front() {
+            let task = self.tasks.get_mut(&task_id).unwrap();
+            let waker = self
+                .waker_cache
+                .entry(task_id)
+                .or_insert_with(|| TaskWaker::new(task_id, self.task_queue.clone()));
+            let mut ctx = Context::from_waker(waker);
+            match task.poll(&mut ctx) {
+                Poll::Ready(()) => {
+                    self.tasks.remove(&task_id);
+                    self.waker_cache.remove(&task_id);
+                }
+                Poll::Pending => {}
+            };
+        }
+    }
+
+    pub fn run(&mut self) {
+        loop {
+            self.run_ready_tasks();
+            // TODO: We need some sort of semaphore wait here.
+            syscall::thread_sleep(50).unwrap();
+        }
+    }
+}
+
+static CURRENT_EXECUTOR: Option<Executor> = None;