feat: use async-executor

Author: mkroening

Cargo.lock

@@ -110,6 +110,7 @@ virtio = { package = "virtio-spec", version = "0.3", optional = true, features =
 ahash = { version = "0.8", default-features = false }
 align-address = "0.3"
 anstyle = { version = "1", default-features = false }
+async-executor = { git = "https://github.com/hermit-os/async-executor.git", branch = "no_std", default-features = false, features = ["static"] }
 async-lock = { version = "3.4.1", default-features = false }
 async-trait = "0.1.86"
 bit_field = "0.10"

Cargo.toml

@@ -1,16 +1,16 @@
 use alloc::boxed::Box;
-use alloc::collections::vec_deque::VecDeque;
 use core::arch::asm;
-use core::cell::{Cell, RefCell, RefMut};
+use core::cell::Cell;
 use core::ptr;
 use core::sync::atomic::Ordering;
 
+use async_executor::StaticExecutor;
 #[cfg(feature = "smp")]
 use hermit_sync::InterruptTicketMutex;
+use hermit_sync::{RawRwSpinLock, RawSpinMutex};
 
 use super::CPU_ONLINE;
 use super::interrupts::{IRQ_COUNTERS, IrqStatistics};
-use crate::executor::task::AsyncTask;
 #[cfg(feature = "smp")]
 use crate::scheduler::SchedulerInput;
 use crate::scheduler::{CoreId, PerCoreScheduler};
@@ -23,8 +23,8 @@ pub(crate) struct CoreLocal {
 	scheduler: Cell<*mut PerCoreScheduler>,
 	/// Interface to the interrupt counters
 	irq_statistics: &'static IrqStatistics,
-	/// Queue of async tasks
-	async_tasks: RefCell<VecDeque<AsyncTask>>,
+	/// The core-local async executor.
+	ex: StaticExecutor<RawSpinMutex, RawRwSpinLock>,
 	/// Queues to handle incoming requests from the other cores
 	#[cfg(feature = "smp")]
 	pub scheduler_input: InterruptTicketMutex<SchedulerInput>,
@@ -46,7 +46,7 @@ impl CoreLocal {
 			core_id,
 			scheduler: Cell::new(ptr::null_mut()),
 			irq_statistics,
-			async_tasks: RefCell::new(VecDeque::new()),
+			ex: StaticExecutor::new(),
 			#[cfg(feature = "smp")]
 			scheduler_input: InterruptTicketMutex::new(SchedulerInput::new()),
 		};
@@ -96,8 +96,8 @@ pub(crate) fn core_scheduler() -> &'static mut PerCoreScheduler {
 	unsafe { CoreLocal::get().scheduler.get().as_mut().unwrap() }
 }
 
-pub(crate) fn async_tasks() -> RefMut<'static, VecDeque<AsyncTask>> {
-	CoreLocal::get().async_tasks.borrow_mut()
+pub(crate) fn ex() -> &'static StaticExecutor<RawSpinMutex, RawRwSpinLock> {
+	&CoreLocal::get().ex
 }
 
 pub(crate) fn set_core_scheduler(scheduler: *mut PerCoreScheduler) {

src/arch/aarch64/kernel/core_local.rs

@@ -1,15 +1,15 @@
 use alloc::boxed::Box;
-use alloc::collections::vec_deque::VecDeque;
 use core::arch::asm;
-use core::cell::{Cell, RefCell, RefMut};
+use core::cell::Cell;
 use core::ptr;
 use core::sync::atomic::Ordering;
 
+use async_executor::StaticExecutor;
 #[cfg(feature = "smp")]
 use hermit_sync::InterruptTicketMutex;
+use hermit_sync::{RawRwSpinLock, RawSpinMutex};
 
 use crate::arch::riscv64::kernel::CPU_ONLINE;
-use crate::executor::task::AsyncTask;
 #[cfg(feature = "smp")]
 use crate::scheduler::SchedulerInput;
 use crate::scheduler::{CoreId, PerCoreScheduler};
@@ -21,8 +21,8 @@ pub struct CoreLocal {
 	scheduler: Cell<*mut PerCoreScheduler>,
 	/// start address of the kernel stack
 	pub kernel_stack: Cell<u64>,
-	/// Queue of async tasks
-	async_tasks: RefCell<VecDeque<AsyncTask>>,
+	/// The core-local async executor.
+	ex: StaticExecutor<RawSpinMutex, RawRwSpinLock>,
 	/// Queues to handle incoming requests from the other cores
 	#[cfg(feature = "smp")]
 	pub scheduler_input: InterruptTicketMutex<SchedulerInput>,
@@ -41,7 +41,7 @@ impl CoreLocal {
 				core_id,
 				scheduler: Cell::new(ptr::null_mut()),
 				kernel_stack: Cell::new(0),
-				async_tasks: RefCell::new(VecDeque::new()),
+				ex: StaticExecutor::new(),
 				#[cfg(feature = "smp")]
 				scheduler_input: InterruptTicketMutex::new(SchedulerInput::new()),
 			};
@@ -84,6 +84,6 @@ pub fn set_core_scheduler(scheduler: *mut PerCoreScheduler) {
 	CoreLocal::get().scheduler.set(scheduler);
 }
 
-pub(crate) fn async_tasks() -> RefMut<'static, VecDeque<AsyncTask>> {
-	CoreLocal::get().async_tasks.borrow_mut()
+pub(crate) fn ex() -> &'static StaticExecutor<RawSpinMutex, RawRwSpinLock> {
+	&CoreLocal::get().ex
 }

src/arch/riscv64/kernel/core_local.rs

@@ -1,21 +1,21 @@
 use alloc::boxed::Box;
-use alloc::collections::vec_deque::VecDeque;
 use core::arch::asm;
-use core::cell::{Cell, RefCell, RefMut};
+use core::cell::Cell;
 #[cfg(feature = "smp")]
 use core::sync::atomic::AtomicBool;
 use core::sync::atomic::Ordering;
 use core::{mem, ptr};
 
+use async_executor::StaticExecutor;
 #[cfg(feature = "smp")]
 use hermit_sync::InterruptTicketMutex;
+use hermit_sync::{RawRwSpinLock, RawSpinMutex};
 use x86_64::VirtAddr;
 use x86_64::registers::model_specific::GsBase;
 use x86_64::structures::tss::TaskStateSegment;
 
 use super::CPU_ONLINE;
 use super::interrupts::{IRQ_COUNTERS, IrqStatistics};
-use crate::executor::task::AsyncTask;
 #[cfg(feature = "smp")]
 use crate::scheduler::SchedulerInput;
 use crate::scheduler::{CoreId, PerCoreScheduler};
@@ -32,8 +32,8 @@ pub(crate) struct CoreLocal {
 	pub kernel_stack: Cell<*mut u8>,
 	/// Interface to the interrupt counters
 	irq_statistics: &'static IrqStatistics,
-	/// Queue of async tasks
-	async_tasks: RefCell<VecDeque<AsyncTask>>,
+	/// The core-local async executor.
+	ex: StaticExecutor<RawSpinMutex, RawRwSpinLock>,
 	#[cfg(feature = "smp")]
 	pub hlt: AtomicBool,
 	/// Queues to handle incoming requests from the other cores
@@ -61,7 +61,7 @@ impl CoreLocal {
 			tss: Cell::new(ptr::null_mut()),
 			kernel_stack: Cell::new(ptr::null_mut()),
 			irq_statistics,
-			async_tasks: RefCell::new(VecDeque::new()),
+			ex: StaticExecutor::new(),
 			#[cfg(feature = "smp")]
 			hlt: AtomicBool::new(false),
 			#[cfg(feature = "smp")]
@@ -110,8 +110,8 @@ pub(crate) fn core_scheduler() -> &'static mut PerCoreScheduler {
 	unsafe { CoreLocal::get().scheduler.get().as_mut().unwrap() }
 }
 
-pub(crate) fn async_tasks() -> RefMut<'static, VecDeque<AsyncTask>> {
-	CoreLocal::get().async_tasks.borrow_mut()
+pub(crate) fn ex() -> &'static StaticExecutor<RawSpinMutex, RawRwSpinLock> {
+	&CoreLocal::get().ex
 }
 
 pub(crate) fn set_core_scheduler(scheduler: *mut PerCoreScheduler) {

src/arch/x86_64/kernel/core_local.rs

@@ -9,7 +9,7 @@ pub(crate) mod vsock;
 use alloc::sync::Arc;
 use alloc::task::Wake;
 use core::future::Future;
-use core::pin::{Pin, pin};
+use core::pin::pin;
 use core::sync::atomic::AtomicU32;
 use core::task::{Context, Poll, Waker};
 use core::time::Duration;
@@ -94,14 +94,13 @@ impl Wake for TaskNotify {
 }
 
 pub(crate) fn run() {
-	let mut cx = Context::from_waker(Waker::noop());
-
 	without_interrupts(|| {
-		// FIXME(mkroening): Not all tasks register wakers and never sleep
-		for _ in 0..({ core_local::async_tasks().len() }) {
-			let mut task = { core_local::async_tasks().pop_front().unwrap() };
-			if Pin::new(&mut task).poll(&mut cx).is_pending() {
-				core_local::async_tasks().push_back(task);
+		// FIXME: We currently have no more than 3 tasks at a time, so this is fine.
+		// Ideally, we would set this value to 200, but the network task currently immediately wakes up again.
+		// This would lead to the network task being polled 200 times back to back, slowing things down considerably.
+		for _ in 0..3 {
+			if !core_local::ex().try_tick() {
+				break;
 			}
 		}
 	});
@@ -116,7 +115,7 @@ pub(crate) fn spawn<F>(future: F)
 where
 	F: Future<Output = ()> + Send + 'static,
 {
-	without_interrupts(|| core_local::async_tasks().push_back(AsyncTask::new(future)));
+	core_local::ex().spawn(AsyncTask::new(future)).detach();
 }
 
 pub fn init() {