Introducing the hottest library crate in town (libk!)

libk/src/scheduling/task.rs

@@ -0,0 +1,133 @@
+extern crate alloc;
+use alloc::{boxed::Box, collections::BTreeMap, sync::Arc, task::Wake};
+use core::{
+    future::Future,
+    pin::Pin,
+    sync::atomic::{AtomicU64, Ordering},
+    task::{Context, Poll, Waker},
+};
+use crossbeam::queue::ArrayQueue;
+
+pub struct Task {
+    id: TaskId,
+    future: Pin<Box<dyn Future<Output = ()>>>,
+}
+
+impl Task {
+    pub fn new(future: impl Future<Output = ()> + 'static) -> Task {
+        Task {
+            id: TaskId::new(),
+            future: Box::pin(future),
+        }
+    }
+
+    fn poll(&mut self, context: &mut Context) -> Poll<()> {
+        self.future.as_mut().poll(context)
+    }
+}
+
+pub struct Executor {
+    tasks: BTreeMap<TaskId, Task>,
+    task_queue: Arc<ArrayQueue<TaskId>>,
+    waker_cache: BTreeMap<TaskId, Waker>,
+}
+
+impl Executor {
+    pub fn new() -> Self {
+        Executor {
+            tasks: BTreeMap::new(),
+            task_queue: Arc::new(ArrayQueue::new(100)),
+            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!("a task with this id has already been allocated");
+        }
+        self.task_queue.push(task_id).expect("task queue is full");
+    }
+
+    fn run_ready_tasks(&mut self) {
+        let Self {
+            tasks,
+            task_queue,
+            waker_cache,
+        } = self;
+
+        while let Some(task_id) = task_queue.pop() {
+            let task = match tasks.get_mut(&task_id) {
+                Some(task) => task,
+                None => continue,
+            };
+            let waker = waker_cache
+                .entry(task_id)
+                .or_insert_with(|| TaskWaker::new(task_id, task_queue.clone()));
+            let mut context = Context::from_waker(waker);
+            match task.poll(&mut context) {
+                Poll::Ready(()) => {
+                    tasks.remove(&task_id);
+                    waker_cache.remove(&task_id);
+                }
+                Poll::Pending => {}
+            }
+        }
+    }
+
+    pub fn try_run(&mut self) {
+        self.run_ready_tasks();
+        self.sleep_if_idle();
+    }
+
+    fn sleep_if_idle(&self) {
+        use x86_64::instructions::interrupts::{self, enable_and_hlt};
+
+        interrupts::disable();
+        if self.task_queue.is_empty() {
+            enable_and_hlt();
+        } else {
+            interrupts::enable();
+        }
+    }
+}
+
+struct TaskWaker {
+    task_id: TaskId,
+    task_queue: Arc<ArrayQueue<TaskId>>,
+}
+
+impl TaskWaker {
+    fn new(task_id: TaskId, task_queue: Arc<ArrayQueue<TaskId>>) -> Waker {
+        Waker::from(Arc::new(TaskWaker {
+            task_id,
+            task_queue,
+        }))
+    }
+
+    fn wake_task(&self) {
+        self.task_queue
+            .push(self.task_id)
+            .expect("task queue is full!");
+    }
+}
+
+impl Wake for TaskWaker {
+    fn wake(self: Arc<Self>) {
+        self.wake_task();
+    }
+
+    fn wake_by_ref(self: &Arc<Self>) {
+        self.wake_task();
+    }
+}
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
+struct TaskId(u64);
+
+impl TaskId {
+    fn new() -> Self {
+        static NEXT: AtomicU64 = AtomicU64::new(0);
+        TaskId(NEXT.fetch_add(1, Ordering::Relaxed))
+    }
+}