add futex syscalls

Author: joboet

Cargo.lock

@@ -67,8 +67,10 @@ dhcpv4 = [
 ]
 
 [dependencies]
+ahash = { version = "0.8", default-features = false }
 bitflags = "1.3"
 crossbeam-utils = { version = "0.8", default-features = false }
+hashbrown = { version = "0.12", default-features = false }
 hermit-entry = { version = "0.9", features = ["kernel"] }
 include-transformed = { version = "0.2", optional = true }
 log = { version = "0.4", default-features = false }

Cargo.toml

@@ -9,6 +9,7 @@
 #![allow(incomplete_features)]
 #![feature(abi_x86_interrupt)]
 #![feature(allocator_api)]
+#![feature(atomic_mut_ptr)]
 #![feature(asm_const)]
 #![feature(asm_sym)]
 #![feature(const_btree_new)]
@@ -23,6 +24,8 @@
 #![feature(alloc_error_handler)]
 #![feature(vec_into_raw_parts)]
 #![feature(drain_filter)]
+#![feature(strict_provenance)]
+#![feature(is_some_with)]
 #![no_std]
 #![cfg_attr(target_os = "none", feature(custom_test_frameworks))]
 #![cfg_attr(target_os = "none", cfg_attr(test, test_runner(crate::test_runner)))]

src/lib.rs

@@ -140,6 +140,7 @@ impl PartialEq for TaskHandle {
 impl Eq for TaskHandle {}
 
 /// Realize a priority queue for task handles
+#[derive(Default)]
 pub struct TaskHandlePriorityQueue {
 	queues: [Option<VecDeque<TaskHandle>>; NO_PRIORITIES],
 	prio_bitmap: u64,
@@ -158,6 +159,11 @@ impl TaskHandlePriorityQueue {
 		}
 	}
 
+	/// Checks if the queue is empty.
+	pub fn is_empty(&self) -> bool {
+		self.prio_bitmap == 0
+	}
+
 	/// Add a task handle by its priority to the queue
 	pub fn push(&mut self, task: TaskHandle) {
 		let i = task.priority.into() as usize;
@@ -196,16 +202,19 @@ impl TaskHandlePriorityQueue {
 		None
 	}
 
-	/// Remove a specific task handle from the priority queue.
-	pub fn remove(&mut self, task: TaskHandle) {
+	/// Remove a specific task handle from the priority queue. Returns `true` if
+	/// the handle was in the queue.
+	pub fn remove(&mut self, task: TaskHandle) -> bool {
 		let queue_index = task.priority.into() as usize;
 		//assert!(queue_index < NO_PRIORITIES, "Priority {} is too high", queue_index);
 
+		let mut success = false;
 		if let Some(queue) = &mut self.queues[queue_index] {
 			let mut i = 0;
 			while i != queue.len() {
 				if queue[i].id == task.id {
 					queue.remove(i);
+					success = true;
 				} else {
 					i += 1;
 				}
@@ -215,6 +224,8 @@ impl TaskHandlePriorityQueue {
 				self.prio_bitmap &= !(1 << queue_index as u64);
 			}
 		}
+
+		success
 	}
 }
 

src/scheduler/task.rs

@@ -0,0 +1,106 @@
+use ahash::RandomState;
+use core::sync::atomic::{AtomicU32, Ordering::SeqCst};
+use hashbrown::{hash_map::Entry, HashMap};
+
+use crate::{
+	arch::kernel::{percore::core_scheduler, processor::get_timer_ticks},
+	errno::{EAGAIN, EINTR, EINVAL, ETIMEDOUT},
+	scheduler::task::TaskHandlePriorityQueue,
+};
+
+use super::spinlock::SpinlockIrqSave;
+
+// TODO: Replace with a concurrent hashmap.
+static PARKING_LOT: SpinlockIrqSave<HashMap<usize, TaskHandlePriorityQueue, RandomState>> =
+	SpinlockIrqSave::new(HashMap::with_hasher(RandomState::with_seeds(0, 0, 0, 0)));
+
+bitflags! {
+	pub struct Flags: u32 {
+		/// Use a relative timeout
+		const RELATIVE = 0b01;
+	}
+}
+
+/// If the value at address matches the expected value, park the current thread until it is either
+/// woken up with `futex_wake` (returns 0), the specified timeout elapses (returns -ETIMEDOUT) or
+/// a spurious wakeup occurs (returns -EINTR).
+///
+/// The timeout uses processor ticks. If [`Flags::RELATIVE`] is given, it is interpreted as relative
+/// to the current time.
+pub fn futex_wait(address: &AtomicU32, expected: u32, timeout: Option<u64>, flags: Flags) -> i32 {
+	let mut parking_lot = PARKING_LOT.lock();
+	// Check the futex value after locking the parking lot so that all changes are observed.
+	if address.load(SeqCst) != expected {
+		return -EAGAIN;
+	}
+
+	let wakeup_time = if flags.contains(Flags::RELATIVE) {
+		timeout.map(|t| get_timer_ticks().wrapping_add(t))
+	} else {
+		timeout
+	};
+
+	let scheduler = core_scheduler();
+	scheduler.block_current_task(wakeup_time);
+	let handle = scheduler.get_current_task_handle();
+	parking_lot
+		.entry(address.as_mut_ptr().addr())
+		.or_default()
+		.push(handle);
+	drop(parking_lot);
+
+	scheduler.reschedule();
+
+	// Try to remove ourselves from the waiting queue.
+	let mut parking_lot = PARKING_LOT.lock();
+	let mut wakeup = true;
+	if let Entry::Occupied(mut queue) = parking_lot.entry(address.as_mut_ptr().addr()) {
+		// If we are not in the waking queue, this must have been a wakeup.
+		wakeup = !queue.get_mut().remove(handle);
+		if queue.get().is_empty() {
+			queue.remove();
+		}
+	};
+	drop(parking_lot);
+
+	if wakeup {
+		0
+	} else if wakeup_time.is_some_and(|&t| t <= get_timer_ticks()) {
+		// If the current time is past the wakeup time, the operation timed out.
+		-ETIMEDOUT
+	} else {
+		// Otherwise the wakeup must have been spurious.
+		-EINTR
+	}
+}
+
+/// Wake `count` threads waiting on the futex at address. Returns the number of threads
+/// woken up (saturates to `i32::MAX`). If `count` is `i32::MAX`, wake up all matching
+/// waiting threads. If `count` is negative, returns -EINVAL.
+pub fn futex_wake(address: &AtomicU32, count: i32) -> i32 {
+	if count < 0 {
+		return -EINVAL;
+	}
+
+	let mut parking_lot = PARKING_LOT.lock();
+	let mut queue = match parking_lot.entry(address.as_mut_ptr().addr()) {
+		Entry::Occupied(entry) => entry,
+		Entry::Vacant(_) => return 0,
+	};
+
+	let scheduler = core_scheduler();
+	let mut woken = 0;
+	while woken != count || count == i32::MAX {
+		match queue.get_mut().pop() {
+			Some(handle) => scheduler.custom_wakeup(handle),
+			None => break,
+		}
+		woken = woken.saturating_add(1);
+	}
+
+	if queue.get().is_empty() {
+		queue.remove();
+	}
+
+	woken
+}

src/synch/futex.rs

@@ -1,5 +1,6 @@
 //! Synchronization primitives
 
+pub mod futex;
 pub mod recmutex;
 pub mod semaphore;
 pub mod spinlock;

src/synch/mod.rs

@@ -0,0 +1,67 @@
+use core::sync::atomic::AtomicU32;
+
+use crate::{
+	errno::EINVAL,
+	synch::futex::{self as synch, Flags},
+	timespec, timespec_to_microseconds,
+};
+
+/// Like `synch::futex_wait`, but does extra sanity checks and takes a `timespec`.
+///
+/// Returns -EINVAL if
+/// * `address` is null
+/// * `timeout` is negative
+/// * `flags` contains unknown flags
+extern "C" fn __sys_futex_wait(
+	address: *mut u32,
+	expected: u32,
+	timeout: *const timespec,
+	flags: u32,
+) -> i32 {
+	if address.is_null() {
+		return -EINVAL;
+	}
+
+	let address = unsafe { &*(address as *const AtomicU32) };
+	let timeout = if timeout.is_null() {
+		None
+	} else {
+		match timespec_to_microseconds(unsafe { timeout.read() }) {
+			t @ Some(_) => t,
+			None => return -EINVAL,
+		}
+	};
+	let flags = match Flags::from_bits(flags) {
+		Some(flags) => flags,
+		None => return -EINVAL,
+	};
+
+	synch::futex_wait(address, expected, timeout, flags)
+}
+
+#[no_mangle]
+pub extern "C" fn sys_futex_wait(
+	address: *mut u32,
+	expected: u32,
+	timeout: *const timespec,
+	flags: u32,
+) -> i32 {
+	kernel_function!(__sys_futex_wait(address, expected, timeout, flags))
+}
+
+/// Like `synch::futex_wake`, but does extra sanity checks.
+///
+/// Returns -EINVAL if `address` is null.
+extern "C" fn __sys_futex_wake(address: *mut u32, count: i32) -> i32 {
+	if address.is_null() {
+		return -EINVAL;
+	}
+
+	let address = unsafe { &*(address as *const AtomicU32) };
+	synch::futex_wake(address, count)
+}
+
+#[no_mangle]
+pub extern "C" fn sys_futex_wake(address: *mut u32, count: i32) -> i32 {
+	kernel_function!(__sys_futex_wake(address, count))
+}

src/syscalls/futex.rs

@@ -8,6 +8,7 @@ use crate::syscalls::interfaces::SyscallInterface;
 use crate::{__sys_free, __sys_malloc, __sys_realloc};
 
 pub use self::condvar::*;
+pub use self::futex::*;
 pub use self::processor::*;
 pub use self::random::*;
 pub use self::recmutex::*;
@@ -19,6 +20,7 @@ pub use self::timer::*;
 
 mod condvar;
 pub(crate) mod fs;
+mod futex;
 mod interfaces;
 #[cfg(feature = "newlib")]
 mod lwip;

src/syscalls/mod.rs

@@ -39,6 +39,13 @@ fn microseconds_to_timeval(microseconds: u64, result: &mut timeval) {
 	result.tv_usec = (microseconds % 1_000_000) as i64;
 }
 
+pub(crate) fn timespec_to_microseconds(time: timespec) -> Option<u64> {
+	u64::try_from(time.tv_sec)
+		.ok()
+		.and_then(|secs| secs.checked_mul(1_000_000))
+		.and_then(|millions| millions.checked_add(u64::try_from(time.tv_nsec).ok()? / 1000))
+}
+
 extern "C" fn __sys_clock_getres(clock_id: u64, res: *mut timespec) -> i32 {
 	assert!(
 		!res.is_null(),