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;

rust/sys/denali/src/ahci/controller.rs

@@ -1,3 +1,6 @@
+use core::ops::DerefMut;
+
+use alloc::boxed::Box;
 use alloc::sync::Arc;
 use mammoth::sync::Mutex;
 use mammoth::thread;
@@ -203,11 +206,11 @@ struct Command {
     #[allow(dead_code)] // We need to own this even if we never access it.
     memory_region: MemoryRegion,
 
-    callback: fn(&Self) -> (),
+    callback_internal: Box<dyn Fn(&Self) + Sync + Send + 'static>,
 }
 
 impl Command {
-    pub fn identify(callback: fn(&Self) -> ()) -> Result<Self, ZError> {
+    pub fn identify(callback: Box<dyn Fn(&Self) + Sync + Send + 'static>) -> Result<Self, ZError> {
         let (memory_region, paddr) = MemoryRegion::contiguous_physical(512)?;
 
         Ok(Self {
@@ -216,12 +219,12 @@ impl Command {
             sector_cnt: 1,
             paddr,
             memory_region,
-            callback,
+            callback_internal: callback,
         })
     }
 
     pub fn callback(&self) {
-        (self.callback)(self);
+        (self.callback_internal)(self);
     }
 }
 
@@ -235,6 +238,10 @@ struct PortController<'a> {
     ahci_port_hba: &'a mut AhciPortHba,
     command_slots: [Option<Arc<Command>>; 32],
     command_structures: MemoryRegion,
+
+    // FIXME: These should probably be something like a OnceCell (or OnceLock).
+    sector_size: Arc<Mutex<Option<u64>>>,
+    sector_cnt: Arc<Mutex<Option<u64>>>,
 }
 
 impl<'a> PortController<'a> {
@@ -263,6 +270,8 @@ impl<'a> PortController<'a> {
             ahci_port_hba,
             command_slots: [const { None }; 32],
             command_structures,
+            sector_size: Arc::new(Mutex::new(None)),
+            sector_cnt: Arc::new(Mutex::new(None)),
         };
 
         // This leaves space for 8 prdt entries.
@@ -276,10 +285,12 @@ impl<'a> PortController<'a> {
 
     pub fn identify(&mut self) -> Result<(), ZError> {
         if self.ahci_port_hba.signature == 0x101 {
-            let callback = |c: &Command| {
+            let sector_size = self.sector_size.clone();
+            let sector_cnt = self.sector_cnt.clone();
+            let callback = move |c: &Command| {
                 mammoth::debug!("TESTING");
                 let ident = c.memory_region.slice::<u16>();
-                let sector_size = if ident[106] & (1 << 12) != 0 {
+                let new_sector_size = if ident[106] & (1 << 12) != 0 {
                     ident[117] as u32 | ((ident[118] as u32) << 16)
                 } else {
                     512
@@ -291,12 +302,18 @@ impl<'a> PortController<'a> {
                         | (ident[102] as u64) << 32
                         | (ident[103] as u64) << 48
                 } else {
-                    (ident[60] as u64 | (ident[61] as u64) << 16) as u64
+                    ident[60] as u64 | (ident[61] as u64) << 16
                 };
-                mammoth::debug!("Sector size: {:#0x}", sector_size);
+                mammoth::debug!("Sector size: {:#0x}", new_sector_size);
                 mammoth::debug!("LBA Count: {:#0x}", lba_count);
+
+                let _ = sector_size
+                    .lock()
+                    .deref_mut()
+                    .insert(new_sector_size as u64);
+                let _ = sector_cnt.lock().deref_mut().insert(lba_count as u64);
             };
-            self.issue_command(Arc::from(Command::identify(callback)?))?;
+            self.issue_command(Arc::from(Command::identify(Box::new(callback))?))?;
         } else {
             let sig = self.ahci_port_hba.signature;
             mammoth::debug!("Skipping non-sata sig: {:#0x}", sig);

rust/usr/testbed/src/main.rs

@@ -7,6 +7,8 @@ use alloc::boxed::Box;
 use alloc::string::ToString;
 use mammoth::debug;
 use mammoth::define_entry;
+use mammoth::task::Executor;
+use mammoth::task::Task;
 use mammoth::thread;
 use mammoth::zion::z_err_t;
 use yellowstone_yunq::GetEndpointRequest;
@@ -17,6 +19,22 @@ pub fn testthread() {
     debug!("Testing 1, 8 ,9");
 }
 
+async fn inner_async() {
+    debug!("Inner Async executed");
+}
+
+async fn test_async() {
+    debug!("Async executed");
+
+    let i = inner_async();
+
+    debug!("inner async created");
+
+    i.await;
+
+    debug!("inner async returned");
+}
+
 #[no_mangle]
 pub extern "C" fn main() -> z_err_t {
     debug!("Testing!");
@@ -47,5 +65,11 @@ pub extern "C" fn main() -> z_err_t {
     let t = thread::spawn(|| debug!("Testing 4, 5, 6"));
     t.join().expect("Failed to wait.");
 
+    let mut executor = Executor::new();
+
+    executor.spawn(Task::new(test_async()));
+
+    executor.run();
+
     0
 }