kernel: use lock-free queues for scheduler work submission

Author: Qix-

Cargo.lock

@@ -69,6 +69,7 @@ volatile-register = "0.2.2"
 buddy_system_allocator = "0.11.0"
 hashbrown = { version = "0.15.2", default-features = false, features = ["nightly", "inline-more", "allocator-api2"] }
 foldhash = { version = "0.1.4", default-features = false }
+nolock = { version = "0.4.1", default-features = false, features = ["queues"] }
 
 bindgen = { git = "https://github.com/oro-os/dep.rust-bindgen.git" }
 syn = { version = "2.0.60", features = ["full", "printing"] }

Cargo.toml

@@ -183,9 +183,11 @@ pub unsafe fn initialize_primary(lapic: Lapic) {
 				.expect("failed to create root ring instance");
 
 			// Create a thread for the entry point.
-			// TODO(qix-): Allow stack size to be passed in via module command line.
-			let _thread = Thread::new(&instance, entry_point)
+			let thread = Thread::new(&instance, entry_point)
 				.expect("failed to create root ring instance thread");
+
+			// Spawn it.
+			Thread::spawn(thread);
 		}
 	}
 }

oro-arch-x86_64/src/init.rs

@@ -38,6 +38,7 @@ oro-dbgutil.workspace = true
 
 hashbrown.workspace = true
 foldhash = { workspace = true, default-features = false }
+nolock = { workspace = true, default-features = false, features = ["queues"] }
 
 [lints]
 workspace = true

oro-kernel/Cargo.toml

@@ -52,17 +52,20 @@ use core::{
 	sync::atomic::{AtomicBool, Ordering::SeqCst},
 };
 
-use interface::RingInterface;
+use nolock::queues::{
+	DequeueError,
+	mpmc::bounded::scq::{Receiver, Sender},
+};
 use oro_macro::assert;
 use oro_mem::{
-	alloc::vec::Vec,
 	global_alloc::GlobalPfa,
 	mapper::{AddressSegment, AddressSpace as _, MapError},
 	pfa::Alloc,
 };
-use oro_sync::{Lock, TicketMutex};
+use oro_sync::TicketMutex;
+use tab::Tab;
 
-use self::{arch::Arch, scheduler::Scheduler};
+use self::{arch::Arch, interface::RingInterface, scheduler::Scheduler, thread::Thread};
 
 /// Core-local instance of the Oro kernel.
 ///
@@ -239,8 +242,10 @@ impl<A: Arch> Kernel<A> {
 /// Global state shared by all [`Kernel`] instances across
 /// core boot/powerdown/bringup cycles.
 pub struct KernelState<A: Arch> {
-	/// List of all threads.
-	threads: TicketMutex<Vec<tab::Tab<thread::Thread<A>>>>,
+	/// Unclaimed thread deque sender.
+	thread_tx: Sender<Tab<Thread<A>>>,
+	/// Unclaimed thread deque receiver.
+	thread_rx: Receiver<Tab<Thread<A>>>,
 	/// The root ring.
 	root_ring: tab::Tab<ring::Ring<A>>,
 	/// Whether or not the root ring has been initialized.
@@ -271,9 +276,12 @@ impl<A: Arch> KernelState<A> {
 
 		let root_ring = ring::Ring::<A>::new_root()?;
 
+		let (thread_rx, thread_tx) = nolock::queues::mpmc::bounded::scq::queue(128);
+
 		this.write(Self {
+			thread_tx,
+			thread_rx,
 			root_ring,
-			threads: TicketMutex::default(),
 			has_initialized_root: AtomicBool::new(false),
 		});
 
@@ -285,9 +293,37 @@ impl<A: Arch> KernelState<A> {
 		&self.root_ring
 	}
 
-	/// Returns a reference to the mutex-guarded list of threads.
-	pub fn threads(&self) -> &impl Lock<Target = Vec<tab::Tab<thread::Thread<A>>>> {
-		&self.threads
+	/// Submits a thread to the kernel state to be claimed by
+	/// the next 'free' scheduler.
+	///
+	/// # Panics
+	/// This function panics if the unclaimed thread queue is full.
+	///
+	/// For now, this is acceptable, but will be relaxed in the future
+	/// as the scheduler is fleshed out a bit more.
+	pub fn submit_unclaimed_thread(&self, thread: Tab<Thread<A>>) {
+		// Tell the thread it's been deallocated.
+		unsafe {
+			Thread::<A>::deallocate(&thread);
+		}
+
+		match self.thread_tx.try_enqueue(thread) {
+			Ok(t) => t,
+			Err((err, _)) => panic!("thread queue full or disconnected: {err:?}"),
+		};
+	}
+
+	/// Tries to take the next unclaimed thread.
+	#[expect(clippy::missing_panics_doc)]
+	pub fn try_claim_thread(&self) -> Option<Tab<Thread<A>>> {
+		match self.thread_rx.try_dequeue() {
+			Ok(thread) => Some(thread),
+			Err(DequeueError::Closed) => {
+				// NOTE(qix-): Should never happen.
+				panic!("thread queue disconnected");
+			}
+			Err(DequeueError::Empty) => None,
+		}
 	}
 }
 

oro-kernel/src/lib.rs

@@ -1,14 +1,17 @@
 //! Houses types, traits and functionality for the Oro kernel scheduler.
 
-use oro_mem::alloc::vec::Vec;
-use oro_sync::Lock;
+use nolock::queues::{
+	DequeueError,
+	spsc::unbounded::{UnboundedReceiver as Receiver, UnboundedSender as Sender},
+};
+use oro_debug::dbg_warn;
 
 use crate::{
 	Kernel,
 	arch::{Arch, CoreHandle},
 	syscall::{InFlightSystemCall, InterfaceResponse, SystemCallRequest, SystemCallResponse},
 	tab::Tab,
-	thread::{RunState, ScheduleAction, ScheduleError, Thread},
+	thread::{RunState, ScheduleAction, Thread},
 };
 
 /// Main scheduler state machine.
@@ -22,11 +25,13 @@ use crate::{
 /// itself.
 pub struct Scheduler<A: Arch> {
 	/// A reference to the kernel instance.
-	kernel:     &'static Kernel<A>,
+	kernel:    &'static Kernel<A>,
 	/// The current thread, if there is one being executed.
-	current:    Option<Tab<Thread<A>>>,
-	/// The index of the next thread to execute.
-	next_index: usize,
+	current:   Option<Tab<Thread<A>>>,
+	/// The thread queue transfer side.
+	thread_tx: Sender<Tab<Thread<A>>>,
+	/// The thread queue receive side.
+	thread_rx: Receiver<Tab<Thread<A>>>,
 }
 
 // XXX(qix-): Temporary workaround to make things compile
@@ -37,10 +42,13 @@ unsafe impl<A: Arch> Sync for Scheduler<A> {}
 impl<A: Arch> Scheduler<A> {
 	/// Creates a new scheduler instance.
 	pub(crate) fn new(kernel: &'static Kernel<A>) -> Self {
+		let (thread_rx, thread_tx) = nolock::queues::spsc::unbounded::queue();
+
 		Self {
 			kernel,
 			current: None,
-			next_index: 0,
+			thread_tx,
+			thread_rx,
 		}
 	}
 
@@ -72,48 +80,39 @@ impl<A: Arch> Scheduler<A> {
 			thread
 				.with_mut(|t| t.try_pause(self.kernel.id()))
 				.expect("thread pause failed");
-		}
-
-		// XXX(qix-): This is a terrible design but gets the job done for now.
-		// XXX(qix-): Every single core will be competing for a list of the same threads
-		// XXX(qix-): until a thread migration system is implemented.
-		let mut thread_list = self.kernel.state().threads().lock();
-
-		// XXX(qix-): An even worse design, but it's a temporary workaround
-		// XXX(qix-): to deal with a deadlock. The scheduling algorithm will
-		// XXX(qix-): get a huge refactor at some point in the near future.
-		let mut dead_threads = Vec::new();
 
-		while self.next_index < thread_list.len() {
-			let thread = &thread_list[self.next_index];
-			self.next_index += 1;
-
-			match thread.with_mut(|t| t.try_schedule(self.kernel.id())) {
-				Ok(action) => {
-					// Select it for execution.
-					self.current = Some(thread.clone());
-					return Some((thread.clone(), action));
-				}
-				Err(ScheduleError::Terminated) => {
-					dead_threads.push(self.next_index - 1);
-				}
-				_ => {}
+			if let Err((thread, err)) = self.thread_tx.enqueue(thread) {
+				// We failed to schedule the thread.
+				dbg_warn!(
+					"scheduler {} failed to requeue thread {:#016X}: {err:?}",
+					self.kernel.id(),
+					thread.id()
+				);
+				self.kernel.state().submit_unclaimed_thread(thread);
 			}
 		}
 
-		// For any dead threads we found, remove them from the thread list.
-		for thread_idx in dead_threads {
-			// XXX(qix-): This will inevitably screw up the schedule order
-			// XXX(qix-): for all schedulers in the system but it works for
-			// XXX(qix-): now. The scheduler will get a huge refactor at some
-			// XXX(qix-): point in the near future.
-			thread_list.swap_remove(thread_idx);
-		}
-
-		drop(thread_list);
+		loop {
+			let selected = match self.kernel.state().try_claim_thread() {
+				Some(thread) => thread,
+				None => {
+					match self.thread_rx.try_dequeue() {
+						Ok(thread) => thread,
+						Err(DequeueError::Closed) => panic!("thread queue closed"),
+						Err(DequeueError::Empty) => {
+							return None;
+						}
+					}
+				}
+			};
 
-		self.next_index = 0;
-		None
+			// Take action if needed, otherwise skip the thread; it'll be re-queued when
+			// it needs to be.
+			if let Ok(action) = selected.with_mut(|t| t.try_schedule(self.kernel.id())) {
+				self.current = Some(selected.clone());
+				return Some((selected, action));
+			}
+		}
 	}
 
 	/// Called whenever the architecture has reached a codepath
@@ -246,6 +245,15 @@ impl<A: Arch> Scheduler<A> {
 	}
 }
 
+impl<A: Arch> Drop for Scheduler<A> {
+	fn drop(&mut self) {
+		// Drain the thread queue.
+		while let Ok(thread) = self.thread_rx.try_dequeue() {
+			self.kernel.state().submit_unclaimed_thread(thread);
+		}
+	}
+}
+
 /// Indicates the type of context switch to be taken by an event caller
 /// (typically, the architecture).
 ///

oro-kernel/src/scheduler.rs

@@ -15,11 +15,10 @@ use oro_mem::{
 	mapper::{AddressSegment, AddressSpace as _, MapError, UnmapError},
 	pfa::Alloc,
 };
-use oro_sync::Lock;
 use oro_sysabi::{key, syscall::Error as SysError};
 
 use crate::{
-	AddressSpace, Kernel, UserHandle,
+	AddressSpace, UserHandle,
 	arch::{Arch, ThreadHandle},
 	instance::Instance,
 	ring::Ring,
@@ -197,12 +196,6 @@ impl<A: Arch> Thread<A> {
 
 		instance.with_mut(|instance| instance.threads.insert(tab.id(), tab.clone()));
 
-		Kernel::<A>::get()
-			.state()
-			.threads()
-			.lock()
-			.push(tab.clone());
-
 		Ok(tab)
 	}
 
@@ -422,6 +415,25 @@ impl<A: Arch> Thread<A> {
 		}
 	}
 
+	/// Spawns the thread. If the thread has already been spawned,
+	/// this function does nothing.
+	#[inline]
+	pub fn spawn(this: Tab<Thread<A>>) {
+		if this.with(|t| matches!(t.state, State::Unallocated)) {
+			crate::Kernel::<A>::get()
+				.state()
+				.submit_unclaimed_thread(this);
+		}
+	}
+
+	/// Tells the thread it's been deallocated by a scheduler.
+	///
+	/// # Safety
+	/// The caller must ensure that the thread is not actively running on any core.
+	pub(crate) unsafe fn deallocate(this: &Tab<Thread<A>>) {
+		this.with_mut(|t| t.state = State::Unallocated);
+	}
+
 	/// Returns a reference to the thread's data table.
 	#[inline]
 	pub fn data(&self) -> &TypeTable {