Merge #534

534: Add futex syscalls r=mkroening a=joboet

[Futexes](https://man7.org/linux/man-pages/man2/futex.2.html) greatly simplify the creation of synchronization primitives (see [`std`'s mutex](https://github.com/rust-lang/rust/blob/master/library/std/src/sys/unix/locks/futex_mutex.rs)). They are here implemented by using a hashmap, in which the handles to waiting threads are stored, keyed with the address of the futex they are waiting on. 

I prioritised readability over performance in this PR, faster implementations would use finer grained locking and would not need to allocate the queues. It should be performant enough for most purposes however.

Co-authored-by: joboet <[email protected]>

Author: bors[bot]

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,107 @@
+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, 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) or the specified timeout elapses (returns -ETIMEDOUT).
+///
+/// The timeout is given in microseconds. If [`Flags::RELATIVE`] is given, it is interpreted as
+/// relative to the current time. Otherwise it is understood to be an absolute time
+/// (see `get_timer_ticks`).
+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.and_then(|t| get_timer_ticks().checked_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);
+
+	loop {
+		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();
+			}
+		};
+
+		if wakeup {
+			return 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.
+			return -ETIMEDOUT;
+		}
+
+		// A spurious wakeup occurred, sleep again.
+		scheduler.block_current_task(wakeup_time);
+	}
+}
+
+/// 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(),