diff --git a/libkernel/src/sync/mutex.rs b/libkernel/src/sync/mutex.rs index 68e4be60..e9e8948f 100644 --- a/libkernel/src/sync/mutex.rs +++ b/libkernel/src/sync/mutex.rs @@ -30,6 +30,11 @@ pub struct Mutex { /// /// When an `AsyncMutexGuard` is dropped, it automatically releases the lock and /// wakes up the next task. +/// +/// The guard is `Send` when `T: Send`, and `Sync` when `T: Sync`, independently +/// of the CPU marker type. Unlike the mutex itself, sharing a guard exposes +/// shared references to the protected value, so a guard over `Cell` cannot be +/// shared between threads. #[must_use = "if unused, the Mutex will immediately unlock"] pub struct AsyncMutexGuard<'a, T: ?Sized, CPU: CpuOps> { mutex: &'a Mutex, @@ -134,46 +139,12 @@ impl DerefMut for AsyncMutexGuard<'_, T, CPU> { unsafe impl Send for Mutex {} unsafe impl Sync for Mutex {} -impl Mutex<(), CPU> { - /// Acquires the mutex lock without caring about the data. - pub(crate) fn acquire(&self) -> MutexAcquireFuture<'_, CPU> { - MutexAcquireFuture { mutex: self } - } - - /// Releases the mutex lock without caring about the data. - /// - /// # Safety - /// The caller must ensure that they have previously called [`Self::acquire()`]. - pub(crate) unsafe fn release(&self) { - let mut state = self.state.lock_save_irq(); - - if let Some(next_waker) = state.waiters.pop_front() { - next_waker.wake(); - } - - state.is_locked = false; - } -} - -/// A future that resolves to a locked mutex -pub struct MutexAcquireFuture<'a, CPU: CpuOps> { - mutex: &'a Mutex<(), CPU>, -} - -impl Future for MutexAcquireFuture<'_, CPU> { - type Output = (); - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let mut state = self.mutex.state.lock_save_irq(); +// SAFETY: Moving the guard transfers exclusive access to T. The guard holds no +// saved interrupt state; its drop acquires and releases the state spinlock on +// the current CPU, so it is not tied to the CPU that acquired the mutex. +unsafe impl Send for AsyncMutexGuard<'_, T, CPU> {} - if !state.is_locked { - state.is_locked = true; - Poll::Ready(()) - } else { - if state.waiters.iter().all(|w| !w.will_wake(cx.waker())) { - state.waiters.push_back(cx.waker().clone()); - } - Poll::Pending - } - } -} +// SAFETY: Shared access to a guard exposes only &T, which may cross threads when +// T is Sync. Mutable access and dropping the guard still require ownership. +// This explicit impl prevents inheriting the mutex's weaker T: Send Sync bound. +unsafe impl Sync for AsyncMutexGuard<'_, T, CPU> {} diff --git a/libkernel/src/sync/rwlock.rs b/libkernel/src/sync/rwlock.rs index 2b6759ae..4263b983 100644 --- a/libkernel/src/sync/rwlock.rs +++ b/libkernel/src/sync/rwlock.rs @@ -1,27 +1,165 @@ //! Async-aware readers–writer lock. -use super::spinlock::SpinLockIrq; +use super::{spinlock::SpinLockIrq, waker_set::WakerSet}; use crate::CpuOps; -use crate::sync::mutex::Mutex; use core::cell::UnsafeCell; +use core::future::Future; use core::ops::{Deref, DerefMut}; +use core::pin::Pin; +use core::task::{Context, Poll, Waker}; -struct RwlockState { - num_readers: SpinLockIrq, - writer_lock: Mutex<(), CPU>, +struct RwlockState { + num_readers: usize, + writer_active: bool, + waiting_writers: usize, + read_waiters: WakerSet, + write_waiters: WakerSet, +} + +// Wakers may re-enter the lock, so extract notifications under the spinlock +// but run their callbacks only after releasing it and restoring interrupts. +enum Wakeup { + None, + Writer(Waker), + Readers(WakerSet), +} + +impl Wakeup { + fn wake(self) { + match self { + Self::None => {} + Self::Writer(waker) => waker.wake(), + Self::Readers(mut waiters) => waiters.wake_all(), + } + } +} + +impl RwlockState { + fn take_waiters(&mut self) -> Wakeup { + if self.writer_active { + return Wakeup::None; + } + + if self.waiting_writers != 0 { + if self.num_readers == 0 + && let Some((waker, ())) = self.write_waiters.take_first() + { + return Wakeup::Writer(waker); + } + Wakeup::None + } else { + Wakeup::Readers(self.read_waiters.take_all()) + } + } +} + +// Pending acquisitions own only a waiter registration, never a lock or a +// reader slot. A writer remains counted even after its waker has been removed +// by a notification, until it acquires the lock or is cancelled. +struct AcquireFuture<'a, T: ?Sized, CPU: CpuOps> { + rwlock: &'a Rwlock, + write: bool, + token: Option, +} + +impl Future for AcquireFuture<'_, T, CPU> { + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + let this = self.get_mut(); + let waker = cx.waker().clone(); + let mut state = this.rwlock.state.lock_save_irq(); + let can_acquire = !state.writer_active + && if this.write { + state.num_readers == 0 + } else { + state.waiting_writers == 0 + }; + + if can_acquire { + let retired = this.token.take().and_then(|token| { + if this.write { + state.waiting_writers -= 1; + state.write_waiters.take(token) + } else { + state.read_waiters.take(token) + } + }); + + if this.write { + state.writer_active = true; + } else { + state.num_readers = state + .num_readers + .checked_add(1) + .expect("rwlock reader count overflow"); + } + drop(state); + drop(retired); + return Poll::Ready(()); + } + + // Refresh the waker on every pending poll. Notifications remove + // registrations, and another task may take the lock before we run. + let retired = if this.write { + if let Some(token) = this.token { + state.write_waiters.take(token) + } else { + state.waiting_writers = state + .waiting_writers + .checked_add(1) + .expect("rwlock waiting writer count overflow"); + None + } + } else { + this.token.and_then(|token| state.read_waiters.take(token)) + }; + let waiters = if this.write { + &mut state.write_waiters + } else { + &mut state.read_waiters + }; + this.token = Some(waiters.insert(waker, ())); + drop(state); + drop(retired); + + Poll::Pending + } +} + +impl Drop for AcquireFuture<'_, T, CPU> { + fn drop(&mut self) { + if let Some(token) = self.token { + let mut state = self.rwlock.state.lock_save_irq(); + let retired = if self.write { + state.waiting_writers -= 1; + state.write_waiters.take(token) + } else { + state.read_waiters.take(token) + }; + // Cancellation of a queued or already-notified writer must not + // leave other waiters asleep while the lock is available. + let wakeup = state.take_waiters(); + drop(state); + drop(retired); + wakeup.wake(); + } + } } /// An asynchronous, rwlock primitive. /// /// This rwlock can be used to protect shared data across asynchronous tasks. -/// `lock()` returns a future that resolves to a guard. When the guard is -/// dropped, the lock is released. +/// [`read()`](Self::read) and [`write()`](Self::write) return futures that resolve +/// to guards. When a guard is dropped, its lock is released. +/// +/// Waiting writers take precedence over new readers. pub struct Rwlock { - state: RwlockState, + state: SpinLockIrq, data: UnsafeCell, } -/// A guard that provides read-only access to the data in an `AsyncRwlock`. +/// A guard that provides read-only access to the data in a [`Rwlock`]. /// /// When an `AsyncRwlockReadGuard` is dropped, it automatically decreases the /// read count and wakes up the next task if necessary. @@ -30,10 +168,10 @@ pub struct AsyncRwlockReadGuard<'a, T: ?Sized, CPU: CpuOps> { rwlock: &'a Rwlock, } -/// A guard that provides exclusive access to the data in an `AsyncRwlock`. +/// A guard that provides exclusive access to the data in a [`Rwlock`]. /// -/// When an `AsyncRwlockWriteGuard` is dropped, it automatically releases the lock and -/// wakes up the next task. +/// When an `AsyncRwlockWriteGuard` is dropped, it automatically releases the lock +/// and wakes up the next task. #[must_use = "if unused, the Rwlock will immediately unlock"] pub struct AsyncRwlockWriteGuard<'a, T: ?Sized, CPU: CpuOps> { rwlock: &'a Rwlock, @@ -43,10 +181,13 @@ impl Rwlock { /// Creates a new asynchronous rwlock in an unlocked state. pub fn new(data: T) -> Self { Self { - state: RwlockState { - num_readers: SpinLockIrq::new(0), - writer_lock: Mutex::new(()), - }, + state: SpinLockIrq::new(RwlockState { + num_readers: 0, + writer_active: false, + waiting_writers: 0, + read_waiters: WakerSet::new(), + write_waiters: WakerSet::new(), + }), data: UnsafeCell::new(data), } } @@ -66,11 +207,12 @@ impl Rwlock { /// Returns a guard asynchronously. The guard is released when the /// returned [`AsyncRwlockReadGuard`] is dropped. pub async fn read(&self) -> AsyncRwlockReadGuard<'_, T, CPU> { - let mut num_readers = self.state.num_readers.lock_save_irq(); - *num_readers += 1; - if *num_readers == 1 { - self.state.writer_lock.acquire().await; + AcquireFuture { + rwlock: self, + write: false, + token: None, } + .await; AsyncRwlockReadGuard { rwlock: self } } @@ -79,52 +221,61 @@ impl Rwlock { /// Returns a guard asynchronously. The guard is released when the /// returned [`AsyncRwlockWriteGuard`] is dropped. pub async fn write(&self) -> AsyncRwlockWriteGuard<'_, T, CPU> { - self.state.writer_lock.acquire().await; + AcquireFuture { + rwlock: self, + write: true, + token: None, + } + .await; AsyncRwlockWriteGuard { rwlock: self } } } impl Drop for AsyncRwlockReadGuard<'_, T, CPU> { fn drop(&mut self) { - let mut num_readers = self.rwlock.state.num_readers.lock_save_irq(); - *num_readers -= 1; - if *num_readers == 0 { - unsafe { self.rwlock.state.writer_lock.release() }; - } + let mut state = self.rwlock.state.lock_save_irq(); + state.num_readers -= 1; + let wakeup = state.take_waiters(); + drop(state); + wakeup.wake(); } } impl Deref for AsyncRwlockReadGuard<'_, T, CPU> { type Target = T; fn deref(&self) -> &T { - // SAFETY: This is safe because the existence of this guard guarantees - // we have read access to the data without any writers. + // SAFETY: The existence of this guard guarantees read access without + // any writers. Concurrent read access requires T: Sync to share a lock. unsafe { &*self.rwlock.data.get() } } } impl Drop for AsyncRwlockWriteGuard<'_, T, CPU> { fn drop(&mut self) { - unsafe { self.rwlock.state.writer_lock.release() }; + let mut state = self.rwlock.state.lock_save_irq(); + state.writer_active = false; + let wakeup = state.take_waiters(); + drop(state); + wakeup.wake(); } } impl Deref for AsyncRwlockWriteGuard<'_, T, CPU> { type Target = T; fn deref(&self) -> &T { - // SAFETY: This is safe because the existence of this guard guarantees - // we have exclusive access to the data. + // SAFETY: The existence of this guard guarantees exclusive access. unsafe { &*self.rwlock.data.get() } } } impl DerefMut for AsyncRwlockWriteGuard<'_, T, CPU> { fn deref_mut(&mut self) -> &mut T { - // SAFETY: This is safe because the existence of this guard guarantees - // we have exclusive access to the data. + // SAFETY: The existence of this guard guarantees exclusive access. unsafe { &mut *self.rwlock.data.get() } } } +// SAFETY: Moving an owned lock transfers ownership of T. Sharing a lock also +// permits concurrent readers to access &T, so T must be Sync as well as Send. unsafe impl Send for Rwlock {} -unsafe impl Sync for Rwlock {} +unsafe impl Sync for Rwlock {} diff --git a/libkernel/src/sync/waker_set.rs b/libkernel/src/sync/waker_set.rs index db20d7a9..b1a5c4dd 100644 --- a/libkernel/src/sync/waker_set.rs +++ b/libkernel/src/sync/waker_set.rs @@ -52,6 +52,20 @@ impl WakerSet { self.waiters.remove(&token); } + /// Removes and returns a registration using its token, without waking it. + pub fn take(&mut self, token: u64) -> Option<(Waker, T)> { + self.waiters.remove(&token) + } + + /// Extracts all registrations without waking them or resetting the token + /// sequence. This allows callers to wake them after releasing a state lock. + pub fn take_all(&mut self) -> Self { + Self { + waiters: core::mem::take(&mut self.waiters), + next_id: self.next_id, + } + } + /// Wakes one waiting task, if any. /// /// Returns `true` if a waker was awoken, `false` otherwise. @@ -264,6 +278,22 @@ mod waker_set_tests { assert!(set.is_empty()); } + #[test] + fn take_all_preserves_token_sequence() { + let mut set = set_with_data(&[1, 2]); + let mut taken = set.take_all(); + assert!(set.is_empty()); + assert_eq!(taken.take(0).unwrap().1, 1); + assert_eq!(taken.take_first().unwrap().1, 2); + assert!(taken.is_empty()); + + let token = set.insert(Waker::noop().clone(), 3); + assert_eq!(token, 2); + assert!(!set.contains_token(0)); + assert_eq!(set.take(token).unwrap().1, 3); + assert!(set.take(token).is_none()); + } + #[test] fn insert_then_remove_by_token() { let mut set = WakerSet::new();