| //! Efficient read-write locking without `pthread_rwlock_t`. | |
| //! | |
| //! The readers-writer lock provided by the `pthread` library has a number of problems which make it | |
| //! a suboptimal choice for `std`: | |
| //! | |
| //! * It is non-movable, so it needs to be allocated (lazily, to make the constructor `const`). | |
| //! * `pthread` is an external library, meaning the fast path of acquiring an uncontended lock | |
| //! cannot be inlined. | |
| //! * Some platforms (at least glibc before version 2.25) have buggy implementations that can easily | |
| //! lead to undefined behaviour in safe Rust code when not properly guarded against. | |
| //! * On some platforms (e.g. macOS), the lock is very slow. | |
| //! | |
| //! Therefore, we implement our own [`RwLock`]! Naively, one might reach for a spinlock, but those | |
| //! can be quite [problematic] when the lock is contended. | |
| //! | |
| //! Instead, this [`RwLock`] copies its implementation strategy from the Windows [SRWLOCK] and the | |
| //! [usync] library implementations. | |
| //! | |
| //! Spinning is still used for the fast path, but it is bounded: after spinning fails, threads will | |
| //! locklessly add an information structure ([`Node`]) containing a [`Thread`] handle into a queue | |
| //! of waiters associated with the lock. The lock owner, upon releasing the lock, will scan through | |
| //! the queue and wake up threads as appropriate, and the newly-awoken threads will then try to | |
| //! acquire the lock themselves. | |
| //! | |
| //! The resulting [`RwLock`] is: | |
| //! | |
| //! * adaptive, since it spins before doing any heavyweight parking operations | |
| //! * allocation-free, modulo the per-thread [`Thread`] handle, which is allocated anyways when | |
| //! using threads created by `std` | |
| //! * writer-preferring, even if some readers may still slip through | |
| //! * unfair, which reduces context-switching and thus drastically improves performance | |
| //! | |
| //! and also quite fast in most cases. | |
| //! | |
| //! [problematic]: https://matklad.github.io/2020/01/02/spinlocks-considered-harmful.html | |
| //! [SRWLOCK]: https://learn.microsoft.com/en-us/windows/win32/sync/slim-reader-writer--srw--locks | |
| //! [usync]: https://crates.io/crates/usync | |
| //! | |
| //! # Implementation | |
| //! | |
| //! ## State | |
| //! | |
| //! A single [`AtomicPtr`] is used as state variable. The lowest four bits are used to indicate the | |
| //! meaning of the remaining bits: | |
| //! | |
| //! | [`LOCKED`] | [`QUEUED`] | [`QUEUE_LOCKED`] | [`DOWNGRADED`] | Remaining | | | |
| //! |------------|:-----------|:-----------------|:---------------|:-------------|:----------------------------------------------------------------------------------------------------------------------------| | |
| //! | 0 | 0 | 0 | 0 | 0 | The lock is unlocked, no threads are waiting | | |
| //! | 1 | 0 | 0 | 0 | 0 | The lock is write-locked, no threads waiting | | |
| //! | 1 | 0 | 0 | 0 | n > 0 | The lock is read-locked with n readers | | |
| //! | 0 | 1 | * | 0 | `*mut Node` | The lock is unlocked, but some threads are waiting. Only writers may lock the lock | | |
| //! | 1 | 1 | * | * | `*mut Node` | The lock is locked, but some threads are waiting. If the lock is read-locked, the last queue node contains the reader count | | |
| //! | |
| //! ## Waiter Queue | |
| //! | |
| //! When threads are waiting on the lock (the `QUEUE` bit is set), the lock state points to a queue | |
| //! of waiters, which is implemented as a linked list of nodes stored on the stack to avoid memory | |
| //! allocation. | |
| //! | |
| //! To enable lock-free enqueuing of new nodes to the queue, the linked list is singly-linked upon | |
| //! creation. | |
| //! | |
| //! When the lock is read-locked, the lock count (number of readers) is stored in the last link of | |
| //! the queue. Threads have to traverse the queue to find the last element upon releasing the lock. | |
| //! To avoid having to traverse the entire list every time we want to access the reader count, a | |
| //! pointer to the found tail is cached in the (current) first element of the queue. | |
| //! | |
| //! Also, while the lock is unfair for performance reasons, it is still best to wake the tail node | |
| //! first (FIFO ordering). Since we always pop nodes off the tail of the queue, we must store | |
| //! backlinks to previous nodes so that we can update the `tail` field of the (current) first | |
| //! element of the queue. Adding backlinks is done at the same time as finding the tail (via the | |
| //! function [`find_tail_and_add_backlinks`]), and thus encountering a set tail field on a node | |
| //! indicates that all following nodes in the queue are initialized. | |
| //! | |
| //! TLDR: Here's a diagram of what the queue looks like: | |
| //! | |
| //! ```text | |
| //! state | |
| //! │ | |
| //! ▼ | |
| //! ╭───────╮ next ╭───────╮ next ╭───────╮ next ╭───────╮ | |
| //! │ ├─────►│ ├─────►│ ├─────►│ count │ | |
| //! │ │ │ │ │ │ │ │ | |
| //! │ │ │ │◄─────┤ │◄─────┤ │ | |
| //! ╰───────╯ ╰───────╯ prev ╰───────╯ prev ╰───────╯ | |
| //! │ ▲ | |
| //! └───────────────────────────┘ | |
| //! tail | |
| //! ``` | |
| //! | |
| //! Invariants: | |
| //! 1. At least one node must contain a non-null, current `tail` field. | |
| //! 2. The first non-null `tail` field must be valid and current. | |
| //! 3. All nodes preceding this node must have a correct, non-null `next` field. | |
| //! 4. All nodes following this node must have a correct, non-null `prev` field. | |
| //! | |
| //! Access to the queue is controlled by the `QUEUE_LOCKED` bit. Threads will try to set this bit | |
| //! in two cases: one is when a thread enqueues itself and eagerly adds backlinks to the queue | |
| //! (which drastically improves performance), and the other is after a thread unlocks the lock to | |
| //! wake up the next waiter(s). | |
| //! | |
| //! `QUEUE_LOCKED` is set atomically at the same time as the enqueuing/unlocking operations. The | |
| //! thread releasing the `QUEUE_LOCKED` bit will check the state of the lock (in particular, whether | |
| //! a downgrade was requested using the [`DOWNGRADED`] bit) and wake up waiters as appropriate. This | |
| //! guarantees forward progress even if the unlocking thread could not acquire the queue lock. | |
| //! | |
| //! ## Memory Orderings | |
| //! | |
| //! To properly synchronize changes to the data protected by the lock, the lock is acquired and | |
| //! released with [`Acquire`] and [`Release`] ordering, respectively. To propagate the | |
| //! initialization of nodes, changes to the queue lock are also performed using these orderings. | |
| #![forbid(unsafe_op_in_unsafe_fn)] | |
| use crate::cell::OnceCell; | |
| use crate::hint::spin_loop; | |
| use crate::mem; | |
| use crate::ptr::{self, NonNull, null_mut, without_provenance_mut}; | |
| use crate::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release}; | |
| use crate::sync::atomic::{Atomic, AtomicBool, AtomicPtr}; | |
| use crate::thread::{self, Thread}; | |
| /// The atomic lock state. | |
| type AtomicState = Atomic<State>; | |
| /// The inner lock state. | |
| type State = *mut (); | |
| const UNLOCKED: State = without_provenance_mut(0); | |
| const LOCKED: usize = 1 << 0; | |
| const QUEUED: usize = 1 << 1; | |
| const QUEUE_LOCKED: usize = 1 << 2; | |
| const DOWNGRADED: usize = 1 << 3; | |
| const SINGLE: usize = 1 << 4; | |
| const STATE: usize = DOWNGRADED | QUEUE_LOCKED | QUEUED | LOCKED; | |
| const NODE_MASK: usize = !STATE; | |
| /// Locking uses exponential backoff. `SPIN_COUNT` indicates how many times the locking operation | |
| /// will be retried. | |
| /// | |
| /// In other words, `spin_loop` will be called `2.pow(SPIN_COUNT) - 1` times. | |
| const SPIN_COUNT: usize = 7; | |
| /// Marks the state as write-locked, if possible. | |
| #[inline] | |
| fn write_lock(state: State) -> Option<State> { | |
| if state.addr() & LOCKED == 0 { Some(state.map_addr(|addr| addr | LOCKED)) } else { None } | |
| } | |
| /// Marks the state as read-locked, if possible. | |
| #[inline] | |
| fn read_lock(state: State) -> Option<State> { | |
| if state.addr() & QUEUED == 0 && state.addr() != LOCKED { | |
| Some(without_provenance_mut(state.addr().checked_add(SINGLE)? | LOCKED)) | |
| } else { | |
| None | |
| } | |
| } | |
| /// Converts a `State` into a `Node` by masking out the bottom bits of the state, assuming that the | |
| /// state points to a queue node. | |
| /// | |
| /// # Safety | |
| /// | |
| /// The state must contain a valid pointer to a queue node. | |
| #[inline] | |
| unsafe fn to_node(state: State) -> NonNull<Node> { | |
| unsafe { NonNull::new_unchecked(state.mask(NODE_MASK)).cast() } | |
| } | |
| /// The representation of a thread waiting on the lock queue. | |
| /// | |
| /// We initialize these `Node`s on thread execution stacks to avoid allocation. | |
| /// | |
| /// Note that we need an alignment of 16 to ensure that the last 4 bits of any | |
| /// pointers to `Node`s are always zeroed (for the bit flags described in the | |
| /// module-level documentation). | |
| #[repr(align(16))] | |
| struct Node { | |
| next: AtomicLink, | |
| prev: AtomicLink, | |
| tail: AtomicLink, | |
| write: bool, | |
| thread: OnceCell<Thread>, | |
| completed: Atomic<bool>, | |
| } | |
| /// An atomic node pointer with relaxed operations. | |
| struct AtomicLink(Atomic<*mut Node>); | |
| impl AtomicLink { | |
| fn new(v: Option<NonNull<Node>>) -> AtomicLink { | |
| AtomicLink(AtomicPtr::new(v.map_or(null_mut(), NonNull::as_ptr))) | |
| } | |
| fn get(&self) -> Option<NonNull<Node>> { | |
| NonNull::new(self.0.load(Relaxed)) | |
| } | |
| fn set(&self, v: Option<NonNull<Node>>) { | |
| self.0.store(v.map_or(null_mut(), NonNull::as_ptr), Relaxed); | |
| } | |
| } | |
| impl Node { | |
| /// Creates a new queue node. | |
| fn new(write: bool) -> Node { | |
| Node { | |
| next: AtomicLink::new(None), | |
| prev: AtomicLink::new(None), | |
| tail: AtomicLink::new(None), | |
| write, | |
| thread: OnceCell::new(), | |
| completed: AtomicBool::new(false), | |
| } | |
| } | |
| /// Prepare this node for waiting. | |
| fn prepare(&mut self) { | |
| // Fall back to creating an unnamed `Thread` handle to allow locking in TLS destructors. | |
| self.thread.get_or_init(thread::current_or_unnamed); | |
| self.completed = AtomicBool::new(false); | |
| } | |
| /// Wait until this node is marked as [`complete`](Node::complete)d by another thread. | |
| /// | |
| /// # Safety | |
| /// | |
| /// May only be called from the thread that created the node. | |
| unsafe fn wait(&self) { | |
| while !self.completed.load(Acquire) { | |
| unsafe { | |
| self.thread.get().unwrap().park(); | |
| } | |
| } | |
| } | |
| /// Atomically mark this node as completed. | |
| /// | |
| /// # Safety | |
| /// | |
| /// `node` must point to a valid `Node`, and the node may not outlive this call. | |
| unsafe fn complete(node: NonNull<Node>) { | |
| // Since the node may be destroyed immediately after the completed flag is set, clone the | |
| // thread handle before that. | |
| let thread = unsafe { node.as_ref().thread.get().unwrap().clone() }; | |
| unsafe { | |
| node.as_ref().completed.store(true, Release); | |
| } | |
| thread.unpark(); | |
| } | |
| } | |
| /// Traverse the queue and find the tail, adding backlinks to the queue while traversing. | |
| /// | |
| /// This may be called from multiple threads at the same time as long as the queue is not being | |
| /// modified (this happens when unlocking multiple readers). | |
| /// | |
| /// # Safety | |
| /// | |
| /// * `head` must point to a node in a valid queue. | |
| /// * `head` must be in front of the previous head node that was used to perform the last removal. | |
| /// * The part of the queue starting with `head` must not be modified during this call. | |
| unsafe fn find_tail_and_add_backlinks(head: NonNull<Node>) -> NonNull<Node> { | |
| let mut current = head; | |
| // Traverse the queue until we find a node that has a set `tail`. | |
| let tail = loop { | |
| let c = unsafe { current.as_ref() }; | |
| if let Some(tail) = c.tail.get() { | |
| break tail; | |
| } | |
| // SAFETY: All `next` fields before the first node with a set `tail` are non-null and valid | |
| // (by Invariant 3). | |
| unsafe { | |
| let next = c.next.get().unwrap_unchecked(); | |
| next.as_ref().prev.set(Some(current)); | |
| current = next; | |
| } | |
| }; | |
| unsafe { | |
| head.as_ref().tail.set(Some(tail)); | |
| tail | |
| } | |
| } | |
| /// [`complete`](Node::complete)s all threads in the queue ending with `tail`. | |
| /// | |
| /// # Safety | |
| /// | |
| /// * `tail` must be a valid tail of a fully linked queue. | |
| /// * The current thread must have exclusive access to that queue. | |
| unsafe fn complete_all(tail: NonNull<Node>) { | |
| let mut current = tail; | |
| // Traverse backwards through the queue (FIFO) and `complete` all of the nodes. | |
| loop { | |
| let prev = unsafe { current.as_ref().prev.get() }; | |
| unsafe { | |
| Node::complete(current); | |
| } | |
| match prev { | |
| Some(prev) => current = prev, | |
| None => return, | |
| } | |
| } | |
| } | |
| /// A type to guard against the unwinds of stacks that nodes are located on due to panics. | |
| struct PanicGuard; | |
| impl Drop for PanicGuard { | |
| fn drop(&mut self) { | |
| rtabort!("tried to drop node in intrusive list."); | |
| } | |
| } | |
| /// The public inner `RwLock` type. | |
| pub struct RwLock { | |
| state: AtomicState, | |
| } | |
| impl RwLock { | |
| #[inline] | |
| pub const fn new() -> RwLock { | |
| RwLock { state: AtomicPtr::new(UNLOCKED) } | |
| } | |
| #[inline] | |
| pub fn try_read(&self) -> bool { | |
| self.state.try_update(Acquire, Relaxed, read_lock).is_ok() | |
| } | |
| #[inline] | |
| pub fn read(&self) { | |
| if !self.try_read() { | |
| self.lock_contended(false) | |
| } | |
| } | |
| #[inline] | |
| pub fn try_write(&self) -> bool { | |
| // Atomically set the `LOCKED` bit. This is lowered to a single atomic instruction on most | |
| // modern processors (e.g. "lock bts" on x86 and "ldseta" on modern AArch64), and therefore | |
| // is more efficient than `try_update(lock(true))`, which can spuriously fail if a new | |
| // node is appended to the queue. | |
| self.state.fetch_or(LOCKED, Acquire).addr() & LOCKED == 0 | |
| } | |
| #[inline] | |
| pub fn write(&self) { | |
| if !self.try_write() { | |
| self.lock_contended(true) | |
| } | |
| } | |
| #[cold] | |
| fn lock_contended(&self, write: bool) { | |
| let mut node = Node::new(write); | |
| let mut state = self.state.load(Relaxed); | |
| let mut count = 0; | |
| let update_fn = if write { write_lock } else { read_lock }; | |
| loop { | |
| // Optimistically update the state. | |
| if let Some(next) = update_fn(state) { | |
| // The lock is available, try locking it. | |
| match self.state.compare_exchange_weak(state, next, Acquire, Relaxed) { | |
| Ok(_) => return, | |
| Err(new) => state = new, | |
| } | |
| continue; | |
| } else if state.addr() & QUEUED == 0 && count < SPIN_COUNT { | |
| // If the lock is not available and no threads are queued, optimistically spin for a | |
| // while, using exponential backoff to decrease cache contention. | |
| for _ in 0..(1 << count) { | |
| spin_loop(); | |
| } | |
| state = self.state.load(Relaxed); | |
| count += 1; | |
| continue; | |
| } | |
| // The optimistic paths did not succeed, so fall back to parking the thread. | |
| // First, prepare the node. | |
| node.prepare(); | |
| // If there are threads queued, this will set the `next` field to be a pointer to the | |
| // first node in the queue. | |
| // If the state is read-locked, this will set `next` to the lock count. | |
| // If it is write-locked, it will set `next` to zero. | |
| node.next.0 = AtomicPtr::new(state.mask(NODE_MASK).cast()); | |
| node.prev = AtomicLink::new(None); | |
| // Set the `QUEUED` bit and preserve the `LOCKED` and `DOWNGRADED` bit. | |
| let mut next = ptr::from_ref(&node) | |
| .map_addr(|addr| addr | QUEUED | (state.addr() & (DOWNGRADED | LOCKED))) | |
| as State; | |
| let mut is_queue_locked = false; | |
| if state.addr() & QUEUED == 0 { | |
| // If this is the first node in the queue, set the `tail` field to the node itself | |
| // to ensure there is a valid `tail` field in the queue (Invariants 1 & 2). | |
| // This needs to use `set` to avoid invalidating the new pointer. | |
| node.tail.set(Some(NonNull::from(&node))); | |
| } else { | |
| // Otherwise, the tail of the queue is not known. | |
| node.tail.set(None); | |
| // Try locking the queue to eagerly add backlinks. | |
| next = next.map_addr(|addr| addr | QUEUE_LOCKED); | |
| // Track if we changed the `QUEUE_LOCKED` bit from off to on. | |
| is_queue_locked = state.addr() & QUEUE_LOCKED == 0; | |
| } | |
| // Register the node, using release ordering to propagate our changes to the waking | |
| // thread. | |
| if let Err(new) = self.state.compare_exchange_weak(state, next, AcqRel, Relaxed) { | |
| // The state has changed, just try again. | |
| state = new; | |
| continue; | |
| } | |
| // The node has been registered, so the structure must not be mutably accessed or | |
| // destroyed while other threads may be accessing it. | |
| // Guard against unwinds using a `PanicGuard` that aborts when dropped. | |
| let guard = PanicGuard; | |
| // If the current thread locked the queue, unlock it to eagerly adding backlinks. | |
| if is_queue_locked { | |
| // SAFETY: This thread set the `QUEUE_LOCKED` bit above. | |
| unsafe { | |
| self.unlock_queue(next); | |
| } | |
| } | |
| // Wait until the node is removed from the queue. | |
| // SAFETY: the node was created by the current thread. | |
| unsafe { | |
| node.wait(); | |
| } | |
| // The node was removed from the queue, disarm the guard. | |
| mem::forget(guard); | |
| // Reload the state and try again. | |
| state = self.state.load(Relaxed); | |
| count = 0; | |
| } | |
| } | |
| #[inline] | |
| pub unsafe fn read_unlock(&self) { | |
| match self.state.try_update(Release, Acquire, |state| { | |
| if state.addr() & QUEUED == 0 { | |
| // If there are no threads queued, simply decrement the reader count. | |
| let count = state.addr() - (SINGLE | LOCKED); | |
| Some(if count > 0 { without_provenance_mut(count | LOCKED) } else { UNLOCKED }) | |
| } else if state.addr() & DOWNGRADED != 0 { | |
| // This thread used to have exclusive access, but requested a downgrade. This has | |
| // not been completed yet, so we still have exclusive access. | |
| // Retract the downgrade request and unlock, but leave waking up new threads to the | |
| // thread that already holds the queue lock. | |
| Some(state.mask(!(DOWNGRADED | LOCKED))) | |
| } else { | |
| None | |
| } | |
| }) { | |
| Ok(_) => {} | |
| // There are waiters queued and the lock count was moved to the tail of the queue. | |
| Err(state) => unsafe { self.read_unlock_contended(state) }, | |
| } | |
| } | |
| /// # Safety | |
| /// | |
| /// * There must be threads queued on the lock. | |
| /// * `state` must be a pointer to a node in a valid queue. | |
| /// * There cannot be a `downgrade` in progress. | |
| #[cold] | |
| unsafe fn read_unlock_contended(&self, state: State) { | |
| // SAFETY: | |
| // The state was observed with acquire ordering above, so the current thread will have | |
| // observed all node initializations. | |
| // We also know that no threads can be modifying the queue starting at `state`: because new | |
| // read-locks cannot be acquired while there are any threads queued on the lock, all | |
| // queue-lock owners will observe a set `LOCKED` bit in `self.state` and will not modify | |
| // the queue. The other case that a thread could modify the queue is if a downgrade is in | |
| // progress (removal of the entire queue), but since that is part of this function's safety | |
| // contract, we can guarantee that no other threads can modify the queue. | |
| let tail = unsafe { find_tail_and_add_backlinks(to_node(state)).as_ref() }; | |
| // The lock count is stored in the `next` field of `tail`. | |
| // Decrement it, making sure to observe all changes made to the queue by the other lock | |
| // owners by using acquire-release ordering. | |
| let was_last = tail.next.0.fetch_byte_sub(SINGLE, AcqRel).addr() - SINGLE == 0; | |
| if was_last { | |
| // SAFETY: Other threads cannot read-lock while threads are queued. Also, the `LOCKED` | |
| // bit is still set, so there are no writers. Thus the current thread exclusively owns | |
| // this lock, even though it is a reader. | |
| unsafe { self.unlock_contended(state) } | |
| } | |
| } | |
| #[inline] | |
| pub unsafe fn write_unlock(&self) { | |
| if let Err(state) = | |
| self.state.compare_exchange(without_provenance_mut(LOCKED), UNLOCKED, Release, Relaxed) | |
| { | |
| // SAFETY: Since other threads cannot acquire the lock, the state can only have changed | |
| // because there are threads queued on the lock. | |
| unsafe { self.unlock_contended(state) } | |
| } | |
| } | |
| /// # Safety | |
| /// | |
| /// * The lock must be exclusively owned by this thread. | |
| /// * There must be threads queued on the lock. | |
| /// * `state` must be a pointer to a node in a valid queue. | |
| /// * There cannot be a `downgrade` in progress. | |
| #[cold] | |
| unsafe fn unlock_contended(&self, state: State) { | |
| debug_assert_eq!(state.addr() & (DOWNGRADED | QUEUED | LOCKED), QUEUED | LOCKED); | |
| let mut current = state; | |
| // We want to atomically release the lock and try to acquire the queue lock. | |
| loop { | |
| // First check if the queue lock is already held. | |
| if current.addr() & QUEUE_LOCKED != 0 { | |
| // Another thread holds the queue lock, so let them wake up waiters for us. | |
| let next = current.mask(!LOCKED); | |
| match self.state.compare_exchange_weak(current, next, Release, Relaxed) { | |
| Ok(_) => return, | |
| Err(new) => { | |
| current = new; | |
| continue; | |
| } | |
| } | |
| } | |
| // Atomically release the lock and try to acquire the queue lock. | |
| let next = current.map_addr(|addr| (addr & !LOCKED) | QUEUE_LOCKED); | |
| match self.state.compare_exchange_weak(current, next, AcqRel, Relaxed) { | |
| // Now that we have the queue lock, we can wake up the next waiter. | |
| Ok(_) => { | |
| // SAFETY: This thread just acquired the queue lock, and this function's safety | |
| // contract requires that there are threads already queued on the lock. | |
| unsafe { self.unlock_queue(next) }; | |
| return; | |
| } | |
| Err(new) => current = new, | |
| } | |
| } | |
| } | |
| /// # Safety | |
| /// | |
| /// * The lock must be write-locked by this thread. | |
| #[inline] | |
| pub unsafe fn downgrade(&self) { | |
| // Optimistically change the state from write-locked with a single writer and no waiters to | |
| // read-locked with a single reader and no waiters. | |
| if let Err(state) = self.state.compare_exchange( | |
| without_provenance_mut(LOCKED), | |
| without_provenance_mut(SINGLE | LOCKED), | |
| Release, | |
| Relaxed, | |
| ) { | |
| // SAFETY: The only way the state can have changed is if there are threads queued. | |
| // Wake all of them up. | |
| unsafe { self.downgrade_slow(state) } | |
| } | |
| } | |
| /// Downgrades the lock from write-locked to read-locked in the case that there are threads | |
| /// waiting on the wait queue. | |
| /// | |
| /// This function will either wake up all of the waiters on the wait queue or designate the | |
| /// current holder of the queue lock to wake up all of the waiters instead. Once the waiters | |
| /// wake up, they will continue in the execution loop of `lock_contended`. | |
| /// | |
| /// # Safety | |
| /// | |
| /// * The lock must be write-locked by this thread. | |
| /// * `state` must be a pointer to a node in a valid queue. | |
| /// * There must be threads queued on the lock. | |
| #[cold] | |
| unsafe fn downgrade_slow(&self, mut state: State) { | |
| debug_assert_eq!(state.addr() & (DOWNGRADED | QUEUED | LOCKED), QUEUED | LOCKED); | |
| // Attempt to wake up all waiters by taking ownership of the entire waiter queue. | |
| loop { | |
| if state.addr() & QUEUE_LOCKED != 0 { | |
| // Another thread already holds the queue lock. Tell it to wake up all waiters. | |
| // If the other thread succeeds in waking up waiters before we release our lock, the | |
| // effect will be just the same as if we had changed the state below. | |
| // Otherwise, the `DOWNGRADED` bit will still be set, meaning that when this thread | |
| // calls `read_unlock` later (because it holds a read lock and must unlock | |
| // eventually), it will realize that the lock is still exclusively locked and act | |
| // accordingly. | |
| let next = state.map_addr(|addr| addr | DOWNGRADED); | |
| match self.state.compare_exchange_weak(state, next, Release, Relaxed) { | |
| Ok(_) => return, | |
| Err(new) => state = new, | |
| } | |
| } else { | |
| // Grab the entire queue by swapping the `state` with a single reader. | |
| let next = ptr::without_provenance_mut(SINGLE | LOCKED); | |
| if let Err(new) = self.state.compare_exchange_weak(state, next, AcqRel, Relaxed) { | |
| state = new; | |
| continue; | |
| } | |
| // SAFETY: We have full ownership of this queue now, so nobody else can modify it. | |
| let tail = unsafe { find_tail_and_add_backlinks(to_node(state)) }; | |
| // Wake up all waiters. | |
| // SAFETY: `tail` was just computed, meaning the whole queue is linked, and we have | |
| // full ownership of the queue, so we have exclusive access. | |
| unsafe { complete_all(tail) }; | |
| return; | |
| } | |
| } | |
| } | |
| /// Unlocks the queue. Wakes up all threads if a downgrade was requested, otherwise wakes up the | |
| /// next eligible thread(s) if the lock is unlocked. | |
| /// | |
| /// # Safety | |
| /// | |
| /// * The queue lock must be held by the current thread. | |
| /// * `state` must be a pointer to a node in a valid queue. | |
| /// * There must be threads queued on the lock. | |
| unsafe fn unlock_queue(&self, mut state: State) { | |
| debug_assert_eq!(state.addr() & (QUEUED | QUEUE_LOCKED), QUEUED | QUEUE_LOCKED); | |
| loop { | |
| // SAFETY: Since we have the queue lock, nobody else can be modifying the queue. | |
| let tail = unsafe { find_tail_and_add_backlinks(to_node(state)) }; | |
| if state.addr() & (DOWNGRADED | LOCKED) == LOCKED { | |
| // Another thread has locked the lock and no downgrade was requested. | |
| // Leave waking up waiters to them by releasing the queue lock. | |
| match self.state.compare_exchange_weak( | |
| state, | |
| state.mask(!QUEUE_LOCKED), | |
| Release, | |
| Acquire, | |
| ) { | |
| Ok(_) => return, | |
| Err(new) => { | |
| state = new; | |
| continue; | |
| } | |
| } | |
| } | |
| // Since we hold the queue lock and downgrades cannot be requested if the lock is | |
| // already read-locked, we have exclusive control over the queue here and can make | |
| // modifications. | |
| let downgrade = state.addr() & DOWNGRADED != 0; | |
| let is_writer = unsafe { tail.as_ref().write }; | |
| if !downgrade | |
| && is_writer | |
| && let Some(prev) = unsafe { tail.as_ref().prev.get() } | |
| { | |
| // If we are not downgrading and the next thread is a writer, only wake up that | |
| // writing thread. | |
| // Split off `tail`. | |
| // There are no set `tail` links before the node pointed to by `state`, so the first | |
| // non-null tail field will be current (Invariant 2). | |
| // We also fulfill Invariant 4 since `find_tail` was called on this node, which | |
| // ensures all backlinks are set. | |
| unsafe { | |
| to_node(state).as_ref().tail.set(Some(prev)); | |
| } | |
| // Try to release the queue lock. We need to check the state again since another | |
| // thread might have acquired the lock and requested a downgrade. | |
| let next = state.mask(!QUEUE_LOCKED); | |
| if let Err(new) = self.state.compare_exchange_weak(state, next, Release, Acquire) { | |
| // Undo the tail modification above, so that we can find the tail again above. | |
| // As mentioned above, we have exclusive control over the queue, so no other | |
| // thread could have noticed the change. | |
| unsafe { | |
| to_node(state).as_ref().tail.set(Some(tail)); | |
| } | |
| state = new; | |
| continue; | |
| } | |
| // The tail was split off and the lock was released. Mark the node as completed. | |
| unsafe { | |
| return Node::complete(tail); | |
| } | |
| } else { | |
| // We are either downgrading, the next waiter is a reader, or the queue only | |
| // consists of one waiter. In any case, just wake all threads. | |
| // Clear the queue. | |
| let next = | |
| if downgrade { ptr::without_provenance_mut(SINGLE | LOCKED) } else { UNLOCKED }; | |
| if let Err(new) = self.state.compare_exchange_weak(state, next, Release, Acquire) { | |
| state = new; | |
| continue; | |
| } | |
| // SAFETY: we computed `tail` above, and no new nodes can have been added since | |
| // (otherwise the CAS above would have failed). | |
| // Thus we have complete control over the whole queue. | |
| unsafe { | |
| return complete_all(tail); | |
| } | |
| } | |
| } | |
| } | |
| } |