add async/await support

- add Executor to spawn and run tasks
- add ScancodeStream to asynchronously handle keyboard input

Author: AlexJMohr

Cargo.lock

@@ -14,7 +14,7 @@ version = "0.6.0"
 panic = "abort" # disable stack unwinding on panic
 
 [dependencies]
-bootloader = {version = "0.9.8", features = ["map_physical_memory"]}# for BIOS bootloader.     
+bootloader = {version = "0.9.8", features = ["map_physical_memory"]}# for BIOS bootloader.        
 linked_list_allocator = "0.8.0" 
 pc-keyboard = "0.5.0" # interprets PS/2 keyboard scancode sets 1 and 2
 pic8259_simple = "0.2.0" # keyboard interrupts
@@ -23,6 +23,20 @@ uart_16550 = "0.2.0" # serial port for qemu
 volatile = "0.2.6" # to mark VGA buffer as volatile. TODO: try upgrading
 x86_64 = "0.13.2" 
 
+[dependencies.crossbeam-queue]
+default-features = false
+features = ["alloc"]
+version = "0.2.1"
+
+[dependencies.conquer-once]
+default-features = false
+version = "0.2.0"
+
+[dependencies.futures-util]
+default-features = false
+features = ["alloc"]
+version = "0.3.4"
+
 [dependencies.lazy_static]
 features = ["spin_no_std"]
 version = "1.0"

Cargo.toml

@@ -3,7 +3,7 @@ use pic8259_simple::ChainedPics;
 use spin;
 use x86_64::structures::idt::{InterruptDescriptorTable, InterruptStackFrame};
 
-use crate::{gdt, print, println};
+use crate::{gdt, println};
 
 pub fn init_idt() {
     IDT.load();
@@ -66,30 +66,12 @@ extern "x86-interrupt" fn timer_interrupt_handler(_stack_frame: &mut InterruptSt
 }
 
 extern "x86-interrupt" fn keyboard_interrupt_handler(_stack_frame: &mut InterruptStackFrame) {
-    use pc_keyboard::{layouts, DecodedKey, HandleControl, Keyboard, ScancodeSet1};
-    use spin::Mutex;
     use x86_64::instructions::port::Port;
 
     // Read the scancode. The keyboard controller won't send more interrupts until we do so.
-
-    lazy_static! {
-        static ref KEYBOARD: Mutex<Keyboard<layouts::Us104Key, ScancodeSet1>> = Mutex::new(
-            Keyboard::new(layouts::Us104Key, ScancodeSet1, HandleControl::Ignore)
-        );
-    }
-
-    let mut keyboard = KEYBOARD.lock();
     let mut port = Port::new(0x60);
-
     let scancode: u8 = unsafe { port.read() };
-    if let Ok(Some(key_event)) = keyboard.add_byte(scancode) {
-        if let Some(key) = keyboard.process_keyevent(key_event) {
-            match key {
-                DecodedKey::Unicode(character) => print!("{}", character),
-                DecodedKey::RawKey(key) => print!("{:?}", key),
-            }
-        }
-    }
+    crate::task::keyboard::add_scancode(scancode);
 
     unsafe {
         PICS.lock()

src/interrupts.rs

@@ -4,6 +4,7 @@
 #![feature(alloc_error_handler)]
 #![feature(const_mut_refs)]
 #![feature(custom_test_frameworks)]
+// #![feature(wake_trait)]
 #![test_runner(crate::test_runner)]
 #![reexport_test_harness_main = "test_main"]
 
@@ -19,6 +20,7 @@ pub mod gdt;
 pub mod interrupts;
 pub mod memory;
 pub mod serial;
+pub mod task;
 pub mod vga_buffer;
 
 #[alloc_error_handler]

src/lib.rs

@@ -7,7 +7,12 @@
 extern crate alloc;
 
 use alloc::{boxed::Box, rc::Rc, vec, vec::Vec};
-use blog_os::{allocator, memory::BootInfoFrameAllocator, println};
+use blog_os::{
+    allocator,
+    memory::BootInfoFrameAllocator,
+    println,
+    task::{executor::Executor, keyboard},
+};
 use bootloader::{entry_point, BootInfo};
 use core::panic::PanicInfo;
 
@@ -18,6 +23,8 @@ pub fn kernel_main(boot_info: &'static BootInfo) -> ! {
     use blog_os::memory;
     use x86_64::VirtAddr;
 
+    use blog_os::task::Task;
+
     println!("Hello World{}", "!");
     blog_os::init();
 
@@ -53,8 +60,10 @@ pub fn kernel_main(boot_info: &'static BootInfo) -> ! {
     #[cfg(test)]
     test_main();
 
-    println!("It did not crash!");
-    blog_os::hlt_loop();
+    let mut executor = Executor::new();
+    executor.spawn(Task::new(example_task()));
+    executor.spawn(Task::new(keyboard::print_keypresses()));
+    executor.run();
 }
 
 /// Panic handler. Prints to the VGA buffer
@@ -71,3 +80,12 @@ fn panic(info: &PanicInfo) -> ! {
 fn panic(info: &PanicInfo) -> ! {
     blog_os::test_panic_handler(info)
 }
+
+async fn async_number() -> u32 {
+    42
+}
+
+async fn example_task() {
+    let number = async_number().await;
+    println!("async number: {}", number);
+}

src/main.rs

@@ -0,0 +1,103 @@
+use super::{Task, TaskId};
+use alloc::{collections::BTreeMap, sync::Arc, task::Wake};
+use core::task::{Context, Poll, Waker};
+use crossbeam_queue::ArrayQueue;
+
+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!("task with same ID already in tasks");
+        }
+        self.task_queue.push(task_id).expect("queue full");
+    }
+
+    pub fn run(&mut self) -> ! {
+        loop {
+            self.run_ready_tasks();
+            self.sleep_if_idle();
+        }
+    }
+
+    fn run_ready_tasks(&mut self) {
+        // destructure `self` to avoid borrow checker errors
+        let Self {
+            tasks,
+            task_queue,
+            waker_cache,
+        } = self;
+
+        while let Ok(task_id) = task_queue.pop() {
+            let task = match tasks.get_mut(&task_id) {
+                Some(task) => task,
+                None => continue, // task no longer exists
+            };
+            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(()) => {
+                    // task done -> remove it and its cached waker
+                    tasks.remove(&task_id);
+                    waker_cache.remove(&task_id);
+                }
+                Poll::Pending => {}
+            }
+        }
+    }
+
+    fn sleep_if_idle(&self) {
+        use x86_64::instructions::interrupts::{self, enable_and_hlt};
+        // Sleep until an interrupt happens, so we aren't spinning the CPU
+        // Need to disable interrupts before hlt to prevent a race condition.
+        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 full");
+    }
+}
+
+impl Wake for TaskWaker {
+    fn wake(self: Arc<Self>) {
+        self.wake_task();
+    }
+
+    fn wake_by_ref(self: &Arc<Self>) {
+        self.wake_task();
+    }
+}