From ae049e82f876e335b835edb674228090d84f811e Mon Sep 17 00:00:00 2001 From: Alex Crichton Date: Sat, 22 Mar 2014 00:50:19 -0700 Subject: [PATCH] sync: Rewrite the base primitives This commit rewrites the core primitives of the sync library: Mutex, RWLock, and Semaphore. These primitives now have updated, more modernized apis: * Guards are returned instead of locking with closures. All condition variables have moved inside the guards and extraneous methods have been removed. * Downgrading on an rwlock is now done through the guard instead of the rwlock itself. These types are meant to be general locks, not locks of an internal type (for external usage). New types will be introduced for locking shared data. --- src/libsync/{sync/mod.rs => raw.rs} | 1029 +++++++++++---------------- 1 file changed, 409 insertions(+), 620 deletions(-) rename src/libsync/{sync/mod.rs => raw.rs} (50%) diff --git a/src/libsync/sync/mod.rs b/src/libsync/raw.rs similarity index 50% rename from src/libsync/sync/mod.rs rename to src/libsync/raw.rs index 2217706d4f0..36f0748fe71 100644 --- a/src/libsync/sync/mod.rs +++ b/src/libsync/raw.rs @@ -8,42 +8,34 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -#[allow(missing_doc)]; - -/** - * The concurrency primitives you know and love. - * - * Maybe once we have a "core exports x only to std" mechanism, these can be - * in std. - */ +//! Raw concurrency primitives you know and love. +//! +//! These primitives are not recommended for general use, but are provided for +//! flavorful use-cases. It is recommended to use the types at the top of the +//! `sync` crate which wrap values directly and provide safer abstractions for +//! containing data. use std::cast; use std::comm; use std::kinds::marker; use std::mem::replace; -use std::sync::arc::UnsafeArc; use std::sync::atomics; use std::unstable::finally::Finally; -use arc::MutexArc; +use mutex; /**************************************************************************** * Internals ****************************************************************************/ -pub mod mutex; -pub mod one; -mod mpsc_intrusive; - // Each waiting task receives on one of these. -#[doc(hidden)] type WaitEnd = Receiver<()>; -#[doc(hidden)] type SignalEnd = Sender<()>; // A doubly-ended queue of waiting tasks. -#[doc(hidden)] -struct WaitQueue { head: Receiver, - tail: Sender } +struct WaitQueue { + head: Receiver, + tail: Sender, +} impl WaitQueue { fn new() -> WaitQueue { @@ -90,33 +82,49 @@ impl WaitQueue { } // The building-block used to make semaphores, mutexes, and rwlocks. -struct SemInner { +struct Sem { lock: mutex::Mutex, - count: int, - waiters: WaitQueue, - // Can be either unit or another waitqueue. Some sems shouldn't come with - // a condition variable attached, others should. - blocked: Q + // n.b, we need Sem to be `Share`, but the WaitQueue type is not send/share + // (for good reason). We have an internal invariant on this semaphore, + // however, that the queue is never accessed outside of a locked + // context. For this reason, we shove these behind a pointer which will + // be inferred to be `Share`. + // + // FIXME: this requires an extra allocation, which is bad. + inner: *() } -struct Sem(UnsafeArc>); +struct SemInner { + count: int, + waiters: WaitQueue, + // Can be either unit or another waitqueue. Some sems shouldn't come with + // a condition variable attached, others should. + blocked: Q, +} -#[doc(hidden)] -impl Sem { +#[must_use] +struct SemGuard<'a, Q> { + sem: &'a Sem, +} + +impl Sem { fn new(count: int, q: Q) -> Sem { - Sem(UnsafeArc::new(SemInner { - count: count, - waiters: WaitQueue::new(), - blocked: q, + let inner = unsafe { + cast::transmute(~SemInner { + waiters: WaitQueue::new(), + count: count, + blocked: q, + }) + }; + Sem { lock: mutex::Mutex::new(), - })) + inner: inner, + } } unsafe fn with(&self, f: |&mut SemInner|) { - let Sem(ref arc) = *self; - let state = arc.get(); - let _g = (*state).lock.lock(); - f(cast::transmute(state)); + let _g = self.lock.lock(); + f(&mut *(self.inner as *mut SemInner)) } pub fn acquire(&self) { @@ -130,7 +138,8 @@ impl Sem { waiter_nobe = Some(state.waiters.wait_end()); } }); - // Uncomment if you wish to test for sem races. Not valgrind-friendly. + // Uncomment if you wish to test for sem races. Not + // valgrind-friendly. /* for _ in range(0, 1000) { task::deschedule(); } */ // Need to wait outside the exclusive. if waiter_nobe.is_some() { @@ -150,24 +159,42 @@ impl Sem { } } - pub fn access(&self, blk: || -> U) -> U { - (|| { - self.acquire(); - blk() - }).finally(|| { - self.release(); - }) + pub fn access<'a>(&'a self) -> SemGuard<'a, Q> { + self.acquire(); + SemGuard { sem: self } } } -#[doc(hidden)] -impl Sem > { - fn new_and_signal(count: int, num_condvars: uint) - -> Sem > { +#[unsafe_destructor] +impl Drop for Sem { + fn drop(&mut self) { + let _waiters: ~SemInner = unsafe { cast::transmute(self.inner) }; + self.inner = 0 as *(); + } +} + +#[unsafe_destructor] +impl<'a, Q: Send> Drop for SemGuard<'a, Q> { + fn drop(&mut self) { + self.sem.release(); + } +} + +impl Sem> { + fn new_and_signal(count: int, num_condvars: uint) -> Sem> { let mut queues = Vec::new(); for _ in range(0, num_condvars) { queues.push(WaitQueue::new()); } Sem::new(count, queues) } + + // The only other places that condvars get built are rwlock.write_cond() + // and rwlock_write_mode. + pub fn access_cond<'a>(&'a self) -> SemCondGuard<'a> { + SemCondGuard { + guard: self.access(), + cvar: Condvar { sem: self, order: Nothing, nopod: marker::NoPod }, + } + } } // FIXME(#3598): Want to use an Option down below, but we need a custom enum @@ -195,27 +222,23 @@ pub struct Condvar<'a> { } impl<'a> Condvar<'a> { - /** - * Atomically drop the associated lock, and block until a signal is sent. - * - * # Failure - * A task which is killed (i.e., by linked failure with another task) - * while waiting on a condition variable will wake up, fail, and unlock - * the associated lock as it unwinds. - */ + /// Atomically drop the associated lock, and block until a signal is sent. + /// + /// # Failure + /// + /// A task which is killed while waiting on a condition variable will wake + /// up, fail, and unlock the associated lock as it unwinds. pub fn wait(&self) { self.wait_on(0) } - /** - * As wait(), but can specify which of multiple condition variables to - * wait on. Only a signal_on() or broadcast_on() with the same condvar_id - * will wake this thread. - * - * The associated lock must have been initialised with an appropriate - * number of condvars. The condvar_id must be between 0 and num_condvars-1 - * or else this call will fail. - * - * wait() is equivalent to wait_on(0). - */ + /// As wait(), but can specify which of multiple condition variables to + /// wait on. Only a signal_on() or broadcast_on() with the same condvar_id + /// will wake this thread. + /// + /// The associated lock must have been initialised with an appropriate + /// number of condvars. The condvar_id must be between 0 and num_condvars-1 + /// or else this call will fail. + /// + /// wait() is equivalent to wait_on(0). pub fn wait_on(&self, condvar_id: uint) { let mut wait_end = None; let mut out_of_bounds = None; @@ -248,7 +271,10 @@ impl<'a> Condvar<'a> { }).finally(|| { // Reacquire the condvar. match self.order { - Just(lock) => lock.access(|| self.sem.acquire()), + Just(lock) => { + let _g = lock.access(); + self.sem.acquire(); + } Nothing => self.sem.acquire(), } }) @@ -309,7 +335,6 @@ impl<'a> Condvar<'a> { // Checks whether a condvar ID was out of bounds, and fails if so, or does // something else next on success. #[inline] -#[doc(hidden)] fn check_cvar_bounds( out_of_bounds: Option, id: uint, @@ -325,19 +350,10 @@ fn check_cvar_bounds( } } -#[doc(hidden)] -impl Sem > { - // The only other places that condvars get built are rwlock.write_cond() - // and rwlock_write_mode. - pub fn access_cond(&self, blk: |c: &Condvar| -> U) -> U { - self.access(|| { - blk(&Condvar { - sem: self, - order: Nothing, - nopod: marker::NoPod - }) - }) - } +#[must_use] +struct SemCondGuard<'a> { + guard: SemGuard<'a, Vec>, + cvar: Condvar<'a>, } /**************************************************************************** @@ -345,15 +361,15 @@ impl Sem > { ****************************************************************************/ /// A counting, blocking, bounded-waiting semaphore. -pub struct Semaphore { priv sem: Sem<()> } +pub struct Semaphore { + priv sem: Sem<()>, +} - -impl Clone for Semaphore { - /// Create a new handle to the semaphore. - fn clone(&self) -> Semaphore { - let Sem(ref lock) = self.sem; - Semaphore { sem: Sem(lock.clone()) } - } +/// An RAII guard used to represent an acquired resource to a semaphore. When +/// dropped, this value will release the resource back to the semaphore. +#[must_use] +pub struct SemaphoreGuard<'a> { + priv guard: SemGuard<'a, ()>, } impl Semaphore { @@ -362,66 +378,64 @@ impl Semaphore { Semaphore { sem: Sem::new(count, ()) } } - /** - * Acquire a resource represented by the semaphore. Blocks if necessary - * until resource(s) become available. - */ - pub fn acquire(&self) { (&self.sem).acquire() } + /// Acquire a resource represented by the semaphore. Blocks if necessary + /// until resource(s) become available. + pub fn acquire(&self) { self.sem.acquire() } - /** - * Release a held resource represented by the semaphore. Wakes a blocked - * contending task, if any exist. Won't block the caller. - */ - pub fn release(&self) { (&self.sem).release() } + /// Release a held resource represented by the semaphore. Wakes a blocked + /// contending task, if any exist. Won't block the caller. + pub fn release(&self) { self.sem.release() } - /// Run a function with ownership of one of the semaphore's resources. - pub fn access(&self, blk: || -> U) -> U { (&self.sem).access(blk) } + /// Acquire a resource of this semaphore, returning an RAII guard which will + /// release the resource when dropped. + pub fn access<'a>(&'a self) -> SemaphoreGuard<'a> { + SemaphoreGuard { guard: self.sem.access() } + } } /**************************************************************************** * Mutexes ****************************************************************************/ -/** - * A blocking, bounded-waiting, mutual exclusion lock with an associated - * FIFO condition variable. - * - * # Failure - * A task which fails while holding a mutex will unlock the mutex as it - * unwinds. - */ +/// A blocking, bounded-waiting, mutual exclusion lock with an associated +/// FIFO condition variable. +/// +/// # Failure +/// A task which fails while holding a mutex will unlock the mutex as it +/// unwinds. +pub struct Mutex { + priv sem: Sem>, +} -pub struct Mutex { priv sem: Sem > } -impl Clone for Mutex { - /// Create a new handle to the mutex. - fn clone(&self) -> Mutex { - let Sem(ref queue) = self.sem; - Mutex { sem: Sem(queue.clone()) } } +/// An RAII structure which is used to gain access to a mutex's condition +/// variable. Additionally, when a value of this type is dropped, the +/// corresponding mutex is also unlocked. +#[must_use] +pub struct MutexGuard<'a> { + priv guard: SemGuard<'a, Vec>, + /// Inner condition variable which is connected to the outer mutex, and can + /// be used for atomic-unlock-and-deschedule. + cond: Condvar<'a>, } impl Mutex { /// Create a new mutex, with one associated condvar. pub fn new() -> Mutex { Mutex::new_with_condvars(1) } - /** - * Create a new mutex, with a specified number of associated condvars. This - * will allow calling wait_on/signal_on/broadcast_on with condvar IDs between - * 0 and num_condvars-1. (If num_condvars is 0, lock_cond will be allowed but - * any operations on the condvar will fail.) - */ + /// Create a new mutex, with a specified number of associated condvars. This + /// will allow calling wait_on/signal_on/broadcast_on with condvar IDs + /// between 0 and num_condvars-1. (If num_condvars is 0, lock_cond will be + /// allowed but any operations on the condvar will fail.) pub fn new_with_condvars(num_condvars: uint) -> Mutex { Mutex { sem: Sem::new_and_signal(1, num_condvars) } } - - /// Run a function with ownership of the mutex. - pub fn lock(&self, blk: || -> U) -> U { - (&self.sem).access(blk) - } - - /// Run a function with ownership of the mutex and a handle to a condvar. - pub fn lock_cond(&self, blk: |c: &Condvar| -> U) -> U { - (&self.sem).access_cond(blk) + /// Acquires ownership of this mutex, returning an RAII guard which will + /// unlock the mutex when dropped. The associated condition variable can + /// also be accessed through the returned guard. + pub fn lock<'a>(&'a self) -> MutexGuard<'a> { + let SemCondGuard { guard, cvar } = self.sem.access_cond(); + MutexGuard { guard: guard, cond: cvar } } } @@ -431,118 +445,95 @@ impl Mutex { // NB: Wikipedia - Readers-writers_problem#The_third_readers-writers_problem -#[doc(hidden)] -struct RWLockInner { - // You might ask, "Why don't you need to use an atomic for the mode flag?" - // This flag affects the behaviour of readers (for plain readers, they - // assert on it; for downgraders, they use it to decide which mode to - // unlock for). Consider that the flag is only unset when the very last - // reader exits; therefore, it can never be unset during a reader/reader - // (or reader/downgrader) race. - // By the way, if we didn't care about the assert in the read unlock path, - // we could instead store the mode flag in write_downgrade's stack frame, - // and have the downgrade tokens store a reference to it. - read_mode: bool, +/// A blocking, no-starvation, reader-writer lock with an associated condvar. +/// +/// # Failure +/// +/// A task which fails while holding an rwlock will unlock the rwlock as it +/// unwinds. +pub struct RWLock { + priv order_lock: Semaphore, + priv access_lock: Sem>, + // The only way the count flag is ever accessed is with xadd. Since it is // a read-modify-write operation, multiple xadds on different cores will // always be consistent with respect to each other, so a monotonic/relaxed // consistency ordering suffices (i.e., no extra barriers are needed). + // // FIXME(#6598): The atomics module has no relaxed ordering flag, so I use // acquire/release orderings superfluously. Change these someday. - read_count: atomics::AtomicUint, + priv read_count: atomics::AtomicUint, } -/** - * A blocking, no-starvation, reader-writer lock with an associated condvar. - * - * # Failure - * A task which fails while holding an rwlock will unlock the rwlock as it - * unwinds. - */ -pub struct RWLock { - priv order_lock: Semaphore, - priv access_lock: Sem >, - priv state: UnsafeArc, +/// An RAII helper which is created by acquiring a read lock on an RWLock. When +/// dropped, this will unlock the RWLock. +#[must_use] +pub struct RWLockReadGuard<'a> { + priv lock: &'a RWLock, +} + +/// An RAII helper which is created by acquiring a write lock on an RWLock. When +/// dropped, this will unlock the RWLock. +/// +/// A value of this type can also be consumed to downgrade to a read-only lock. +#[must_use] +pub struct RWLockWriteGuard<'a> { + priv lock: &'a RWLock, + /// Inner condition variable that is connected to the write-mode of the + /// outer rwlock. + cond: Condvar<'a>, } impl RWLock { /// Create a new rwlock, with one associated condvar. pub fn new() -> RWLock { RWLock::new_with_condvars(1) } - /** - * Create a new rwlock, with a specified number of associated condvars. - * Similar to mutex_with_condvars. - */ + /// Create a new rwlock, with a specified number of associated condvars. + /// Similar to mutex_with_condvars. pub fn new_with_condvars(num_condvars: uint) -> RWLock { - let state = UnsafeArc::new(RWLockInner { - read_mode: false, + RWLock { + order_lock: Semaphore::new(1), + access_lock: Sem::new_and_signal(1, num_condvars), read_count: atomics::AtomicUint::new(0), - }); - RWLock { order_lock: Semaphore::new(1), - access_lock: Sem::new_and_signal(1, num_condvars), - state: state, } - } - - /// Create a new handle to the rwlock. - pub fn clone(&self) -> RWLock { - let Sem(ref access_lock_queue) = self.access_lock; - RWLock { order_lock: (&(self.order_lock)).clone(), - access_lock: Sem(access_lock_queue.clone()), - state: self.state.clone() } - } - - /** - * Run a function with the rwlock in read mode. Calls to 'read' from other - * tasks may run concurrently with this one. - */ - pub fn read(&self, blk: || -> U) -> U { - unsafe { - (&self.order_lock).access(|| { - let state = &mut *self.state.get(); - let old_count = state.read_count.fetch_add(1, atomics::Acquire); - if old_count == 0 { - (&self.access_lock).acquire(); - state.read_mode = true; - } - }); - (|| { - blk() - }).finally(|| { - let state = &mut *self.state.get(); - assert!(state.read_mode); - let old_count = state.read_count.fetch_sub(1, atomics::Release); - assert!(old_count > 0); - if old_count == 1 { - state.read_mode = false; - // Note: this release used to be outside of a locked access - // to exclusive-protected state. If this code is ever - // converted back to such (instead of using atomic ops), - // this access MUST NOT go inside the exclusive access. - (&self.access_lock).release(); - } - }) } } - /** - * Run a function with the rwlock in write mode. No calls to 'read' or - * 'write' from other tasks will run concurrently with this one. - */ - pub fn write(&self, blk: || -> U) -> U { - (&self.order_lock).acquire(); - (&self.access_lock).access(|| { - (&self.order_lock).release(); - blk() - }) + /// Acquires a read-lock, returning an RAII guard that will unlock the lock + /// when dropped. Calls to 'read' from other tasks may run concurrently with + /// this one. + pub fn read<'a>(&'a self) -> RWLockReadGuard<'a> { + let _guard = self.order_lock.access(); + let old_count = self.read_count.fetch_add(1, atomics::Acquire); + if old_count == 0 { + self.access_lock.acquire(); + } + RWLockReadGuard { lock: self } } - /** - * As write(), but also with a handle to a condvar. Waiting on this - * condvar will allow readers and writers alike to take the rwlock before - * the waiting task is signalled. (Note: a writer that waited and then - * was signalled might reacquire the lock before other waiting writers.) - */ - pub fn write_cond(&self, blk: |c: &Condvar| -> U) -> U { + /// Acquire a write-lock, returning an RAII guard that will unlock the lock + /// when dropped. No calls to 'read' or 'write' from other tasks will run + /// concurrently with this one. + /// + /// You can also downgrade a write to a read by calling the `downgrade` + /// method on the returned guard. Additionally, the guard will contain a + /// `Condvar` attached to this lock. + /// + /// # Example + /// + /// ```rust + /// use sync::raw::RWLock; + /// + /// let lock = RWLock::new(); + /// let write = lock.write(); + /// // ... exclusive access ... + /// let read = write.downgrade(); + /// // ... shared access ... + /// drop(read); + /// ``` + pub fn write<'a>(&'a self) -> RWLockWriteGuard<'a> { + let _g = self.order_lock.access(); + self.access_lock.acquire(); + // It's important to thread our order lock into the condvar, so that // when a cond.wait() wakes up, it uses it while reacquiring the // access lock. If we permitted a waking-up writer to "cut in line", @@ -569,187 +560,59 @@ impl RWLock { // which can't happen until T2 finishes the downgrade-read entirely. // The astute reader will also note that making waking writers use the // order_lock is better for not starving readers. - (&self.order_lock).acquire(); - (&self.access_lock).access_cond(|cond| { - (&self.order_lock).release(); - let opt_lock = Just(&self.order_lock); - blk(&Condvar { sem: cond.sem, order: opt_lock, - nopod: marker::NoPod }) - }) - } - - /** - * As write(), but with the ability to atomically 'downgrade' the lock; - * i.e., to become a reader without letting other writers get the lock in - * the meantime (such as unlocking and then re-locking as a reader would - * do). The block takes a "write mode token" argument, which can be - * transformed into a "read mode token" by calling downgrade(). Example: - * - * # Example - * - * ```rust - * use sync::RWLock; - * - * let lock = RWLock::new(); - * lock.write_downgrade(|mut write_token| { - * write_token.write_cond(|condvar| { - * // ... exclusive access ... - * }); - * let read_token = lock.downgrade(write_token); - * read_token.read(|| { - * // ... shared access ... - * }) - * }) - * ``` - */ - pub fn write_downgrade(&self, blk: |v: RWLockWriteMode| -> U) -> U { - // Implementation slightly different from the slicker 'write's above. - // The exit path is conditional on whether the caller downgrades. - (&self.order_lock).acquire(); - (&self.access_lock).acquire(); - (&self.order_lock).release(); - (|| { - blk(RWLockWriteMode { lock: self, nopod: marker::NoPod }) - }).finally(|| { - let writer_or_last_reader; - // Check if we're releasing from read mode or from write mode. - let state = unsafe { &mut *self.state.get() }; - if state.read_mode { - // Releasing from read mode. - let old_count = state.read_count.fetch_sub(1, atomics::Release); - assert!(old_count > 0); - // Check if other readers remain. - if old_count == 1 { - // Case 1: Writer downgraded & was the last reader - writer_or_last_reader = true; - state.read_mode = false; - } else { - // Case 2: Writer downgraded & was not the last reader - writer_or_last_reader = false; - } - } else { - // Case 3: Writer did not downgrade - writer_or_last_reader = true; - } - if writer_or_last_reader { - // Nobody left inside; release the "reader cloud" lock. - (&self.access_lock).release(); - } - }) - } - - /// To be called inside of the write_downgrade block. - pub fn downgrade<'a>(&self, token: RWLockWriteMode<'a>) - -> RWLockReadMode<'a> { - if !((self as *RWLock) == (token.lock as *RWLock)) { - fail!("Can't downgrade() with a different rwlock's write_mode!"); - } - unsafe { - let state = &mut *self.state.get(); - assert!(!state.read_mode); - state.read_mode = true; - // If a reader attempts to enter at this point, both the - // downgrader and reader will set the mode flag. This is fine. - let old_count = state.read_count.fetch_add(1, atomics::Release); - // If another reader was already blocking, we need to hand-off - // the "reader cloud" access lock to them. - if old_count != 0 { - // Guaranteed not to let another writer in, because - // another reader was holding the order_lock. Hence they - // must be the one to get the access_lock (because all - // access_locks are acquired with order_lock held). See - // the comment in write_cond for more justification. - (&self.access_lock).release(); + RWLockWriteGuard { + lock: self, + cond: Condvar { + sem: &self.access_lock, + order: Just(&self.order_lock), + nopod: marker::NoPod, } } - RWLockReadMode { lock: token.lock, nopod: marker::NoPod } } } -/// The "write permission" token used for rwlock.write_downgrade(). +impl<'a> RWLockWriteGuard<'a> { + /// Consumes this write lock and converts it into a read lock. + pub fn downgrade(self) -> RWLockReadGuard<'a> { + let lock = self.lock; + // Don't run the destructor of the write guard, we're in charge of + // things from now on + unsafe { cast::forget(self) } -pub struct RWLockWriteMode<'a> { priv lock: &'a RWLock, priv nopod: marker::NoPod } -/// The "read permission" token used for rwlock.write_downgrade(). -pub struct RWLockReadMode<'a> { priv lock: &'a RWLock, - priv nopod: marker::NoPod } - -impl<'a> RWLockWriteMode<'a> { - /// Access the pre-downgrade rwlock in write mode. - pub fn write(&self, blk: || -> U) -> U { blk() } - /// Access the pre-downgrade rwlock in write mode with a condvar. - pub fn write_cond(&self, blk: |c: &Condvar| -> U) -> U { - // Need to make the condvar use the order lock when reacquiring the - // access lock. See comment in RWLock::write_cond for why. - blk(&Condvar { sem: &self.lock.access_lock, - order: Just(&self.lock.order_lock), - nopod: marker::NoPod }) - } -} - -impl<'a> RWLockReadMode<'a> { - /// Access the post-downgrade rwlock in read mode. - pub fn read(&self, blk: || -> U) -> U { blk() } -} - -/// A barrier enables multiple tasks to synchronize the beginning -/// of some computation. -/// -/// ```rust -/// use sync::Barrier; -/// -/// let barrier = Barrier::new(10); -/// for _ in range(0, 10) { -/// let c = barrier.clone(); -/// // The same messages will be printed together. -/// // You will NOT see any interleaving. -/// spawn(proc() { -/// println!("before wait"); -/// c.wait(); -/// println!("after wait"); -/// }); -/// } -/// ``` -#[deriving(Clone)] -pub struct Barrier { - priv arc: MutexArc, - priv num_tasks: uint, -} - -// The inner state of a double barrier -struct BarrierState { - count: uint, - generation_id: uint, -} - -impl Barrier { - /// Create a new barrier that can block a given number of tasks. - pub fn new(num_tasks: uint) -> Barrier { - Barrier { - arc: MutexArc::new(BarrierState { - count: 0, - generation_id: 0, - }), - num_tasks: num_tasks, + let old_count = lock.read_count.fetch_add(1, atomics::Release); + // If another reader was already blocking, we need to hand-off + // the "reader cloud" access lock to them. + if old_count != 0 { + // Guaranteed not to let another writer in, because + // another reader was holding the order_lock. Hence they + // must be the one to get the access_lock (because all + // access_locks are acquired with order_lock held). See + // the comment in write_cond for more justification. + lock.access_lock.release(); } + RWLockReadGuard { lock: lock } } +} - /// Block the current task until a certain number of tasks is waiting. - pub fn wait(&self) { - self.arc.access_cond(|state, cond| { - let local_gen = state.generation_id; - state.count += 1; - if state.count < self.num_tasks { - // We need a while loop to guard against spurious wakeups. - // http://en.wikipedia.org/wiki/Spurious_wakeup - while local_gen == state.generation_id && state.count < self.num_tasks { - cond.wait(); - } - } else { - state.count = 0; - state.generation_id += 1; - cond.broadcast(); - } - }); +#[unsafe_destructor] +impl<'a> Drop for RWLockWriteGuard<'a> { + fn drop(&mut self) { + self.lock.access_lock.release(); + } +} + +#[unsafe_destructor] +impl<'a> Drop for RWLockReadGuard<'a> { + fn drop(&mut self) { + let old_count = self.lock.read_count.fetch_sub(1, atomics::Release); + assert!(old_count > 0); + if old_count == 1 { + // Note: this release used to be outside of a locked access + // to exclusive-protected state. If this code is ever + // converted back to such (instead of using atomic ops), + // this access MUST NOT go inside the exclusive access. + self.lock.access_lock.release(); + } } } @@ -759,12 +622,12 @@ impl Barrier { #[cfg(test)] mod tests { - use sync::{Semaphore, Mutex, RWLock, Barrier, Condvar}; + use arc::Arc; + use super::{Semaphore, Mutex, RWLock, Condvar}; use std::cast; use std::result; use std::task; - use std::comm::Empty; /************************************************************************ * Semaphore tests @@ -779,26 +642,24 @@ mod tests { #[test] fn test_sem_basic() { let s = Semaphore::new(1); - s.access(|| { }) + let _g = s.access(); } #[test] fn test_sem_as_mutex() { - let s = Semaphore::new(1); + let s = Arc::new(Semaphore::new(1)); let s2 = s.clone(); task::spawn(proc() { - s2.access(|| { - for _ in range(0, 5) { task::deschedule(); } - }) - }); - s.access(|| { + let _g = s2.access(); for _ in range(0, 5) { task::deschedule(); } - }) + }); + let _g = s.access(); + for _ in range(0, 5) { task::deschedule(); } } #[test] fn test_sem_as_cvar() { /* Child waits and parent signals */ let (tx, rx) = channel(); - let s = Semaphore::new(0); + let s = Arc::new(Semaphore::new(0)); let s2 = s.clone(); task::spawn(proc() { s2.acquire(); @@ -810,7 +671,7 @@ mod tests { /* Parent waits and child signals */ let (tx, rx) = channel(); - let s = Semaphore::new(0); + let s = Arc::new(Semaphore::new(0)); let s2 = s.clone(); task::spawn(proc() { for _ in range(0, 5) { task::deschedule(); } @@ -824,40 +685,37 @@ mod tests { fn test_sem_multi_resource() { // Parent and child both get in the critical section at the same // time, and shake hands. - let s = Semaphore::new(2); + let s = Arc::new(Semaphore::new(2)); let s2 = s.clone(); let (tx1, rx1) = channel(); let (tx2, rx2) = channel(); task::spawn(proc() { - s2.access(|| { - let _ = rx2.recv(); - tx1.send(()); - }) + let _g = s2.access(); + let _ = rx2.recv(); + tx1.send(()); }); - s.access(|| { - tx2.send(()); - let _ = rx1.recv(); - }) + let _g = s.access(); + tx2.send(()); + let _ = rx1.recv(); } #[test] fn test_sem_runtime_friendly_blocking() { // Force the runtime to schedule two threads on the same sched_loop. // When one blocks, it should schedule the other one. - let s = Semaphore::new(1); + let s = Arc::new(Semaphore::new(1)); let s2 = s.clone(); let (tx, rx) = channel(); - let mut child_data = Some((s2, tx)); - s.access(|| { - let (s2, tx) = child_data.take_unwrap(); + { + let _g = s.access(); task::spawn(proc() { tx.send(()); - s2.access(|| { }); + drop(s2.access()); tx.send(()); }); - let _ = rx.recv(); // wait for child to come alive + rx.recv(); // wait for child to come alive for _ in range(0, 5) { task::deschedule(); } // let the child contend - }); - let _ = rx.recv(); // wait for child to be done + } + rx.recv(); // wait for child to be done } /************************************************************************ * Mutex tests @@ -867,93 +725,90 @@ mod tests { // Unsafely achieve shared state, and do the textbook // "load tmp = move ptr; inc tmp; store ptr <- tmp" dance. let (tx, rx) = channel(); - let m = Mutex::new(); + let m = Arc::new(Mutex::new()); let m2 = m.clone(); let mut sharedstate = ~0; { - let ptr: *int = &*sharedstate; + let ptr: *mut int = &mut *sharedstate; task::spawn(proc() { - let sharedstate: &mut int = - unsafe { cast::transmute(ptr) }; - access_shared(sharedstate, &m2, 10); + access_shared(ptr, &m2, 10); tx.send(()); }); } { - access_shared(sharedstate, &m, 10); + access_shared(&mut *sharedstate, &m, 10); let _ = rx.recv(); assert_eq!(*sharedstate, 20); } - fn access_shared(sharedstate: &mut int, m: &Mutex, n: uint) { + fn access_shared(sharedstate: *mut int, m: &Arc, n: uint) { for _ in range(0, n) { - m.lock(|| { - let oldval = *sharedstate; - task::deschedule(); - *sharedstate = oldval + 1; - }) + let _g = m.lock(); + let oldval = unsafe { *sharedstate }; + task::deschedule(); + unsafe { *sharedstate = oldval + 1; } } } } #[test] fn test_mutex_cond_wait() { - let m = Mutex::new(); + let m = Arc::new(Mutex::new()); // Child wakes up parent - m.lock_cond(|cond| { + { + let lock = m.lock(); let m2 = m.clone(); task::spawn(proc() { - m2.lock_cond(|cond| { - let woken = cond.signal(); - assert!(woken); - }) + let lock = m2.lock(); + let woken = lock.cond.signal(); + assert!(woken); }); - cond.wait(); - }); + lock.cond.wait(); + } // Parent wakes up child let (tx, rx) = channel(); let m3 = m.clone(); task::spawn(proc() { - m3.lock_cond(|cond| { - tx.send(()); - cond.wait(); - tx.send(()); - }) + let lock = m3.lock(); + tx.send(()); + lock.cond.wait(); + tx.send(()); }); - let _ = rx.recv(); // Wait until child gets in the mutex - m.lock_cond(|cond| { - let woken = cond.signal(); + rx.recv(); // Wait until child gets in the mutex + { + let lock = m.lock(); + let woken = lock.cond.signal(); assert!(woken); - }); - let _ = rx.recv(); // Wait until child wakes up + } + rx.recv(); // Wait until child wakes up } - #[cfg(test)] + fn test_mutex_cond_broadcast_helper(num_waiters: uint) { - let m = Mutex::new(); - let mut rxs = vec!(); + let m = Arc::new(Mutex::new()); + let mut rxs = Vec::new(); for _ in range(0, num_waiters) { let mi = m.clone(); let (tx, rx) = channel(); rxs.push(rx); task::spawn(proc() { - mi.lock_cond(|cond| { - tx.send(()); - cond.wait(); - tx.send(()); - }) + let lock = mi.lock(); + tx.send(()); + lock.cond.wait(); + tx.send(()); }); } // wait until all children get in the mutex - for rx in rxs.mut_iter() { let _ = rx.recv(); } - m.lock_cond(|cond| { - let num_woken = cond.broadcast(); + for rx in rxs.mut_iter() { rx.recv(); } + { + let lock = m.lock(); + let num_woken = lock.cond.broadcast(); assert_eq!(num_woken, num_waiters); - }); + } // wait until all children wake up - for rx in rxs.mut_iter() { let _ = rx.recv(); } + for rx in rxs.mut_iter() { rx.recv(); } } #[test] fn test_mutex_cond_broadcast() { @@ -965,61 +820,57 @@ mod tests { } #[test] fn test_mutex_cond_no_waiter() { - let m = Mutex::new(); + let m = Arc::new(Mutex::new()); let m2 = m.clone(); let _ = task::try(proc() { - m.lock_cond(|_x| { }) + drop(m.lock()); }); - m2.lock_cond(|cond| { - assert!(!cond.signal()); - }) + let lock = m2.lock(); + assert!(!lock.cond.signal()); } #[test] fn test_mutex_killed_simple() { use std::any::Any; // Mutex must get automatically unlocked if failed/killed within. - let m = Mutex::new(); + let m = Arc::new(Mutex::new()); let m2 = m.clone(); let result: result::Result<(), ~Any> = task::try(proc() { - m2.lock(|| { - fail!(); - }) + let _lock = m2.lock(); + fail!(); }); assert!(result.is_err()); // child task must have finished by the time try returns - m.lock(|| { }) + drop(m.lock()); } #[test] fn test_mutex_cond_signal_on_0() { // Tests that signal_on(0) is equivalent to signal(). - let m = Mutex::new(); - m.lock_cond(|cond| { - let m2 = m.clone(); - task::spawn(proc() { - m2.lock_cond(|cond| { - cond.signal_on(0); - }) - }); - cond.wait(); - }) + let m = Arc::new(Mutex::new()); + let lock = m.lock(); + let m2 = m.clone(); + task::spawn(proc() { + let lock = m2.lock(); + lock.cond.signal_on(0); + }); + lock.cond.wait(); } #[test] fn test_mutex_no_condvars() { let result = task::try(proc() { let m = Mutex::new_with_condvars(0); - m.lock_cond(|cond| { cond.wait(); }) + m.lock().cond.wait(); }); assert!(result.is_err()); let result = task::try(proc() { let m = Mutex::new_with_condvars(0); - m.lock_cond(|cond| { cond.signal(); }) + m.lock().cond.signal(); }); assert!(result.is_err()); let result = task::try(proc() { let m = Mutex::new_with_condvars(0); - m.lock_cond(|cond| { cond.broadcast(); }) + m.lock().cond.broadcast(); }); assert!(result.is_err()); } @@ -1029,23 +880,16 @@ mod tests { #[cfg(test)] pub enum RWLockMode { Read, Write, Downgrade, DowngradeRead } #[cfg(test)] - fn lock_rwlock_in_mode(x: &RWLock, mode: RWLockMode, blk: ||) { + fn lock_rwlock_in_mode(x: &Arc, mode: RWLockMode, blk: ||) { match mode { - Read => x.read(blk), - Write => x.write(blk), - Downgrade => - x.write_downgrade(|mode| { - mode.write(|| { blk() }); - }), - DowngradeRead => - x.write_downgrade(|mode| { - let mode = x.downgrade(mode); - mode.read(|| { blk() }); - }), + Read => { let _g = x.read(); blk() } + Write => { let _g = x.write(); blk() } + Downgrade => { let _g = x.write(); blk() } + DowngradeRead => { let _g = x.write().downgrade(); blk() } } } #[cfg(test)] - fn test_rwlock_exclusion(x: &RWLock, + fn test_rwlock_exclusion(x: Arc, mode1: RWLockMode, mode2: RWLockMode) { // Test mutual exclusion between readers and writers. Just like the @@ -1063,14 +907,14 @@ mod tests { }); } { - access_shared(sharedstate, x, mode2, 10); + access_shared(sharedstate, &x, mode2, 10); let _ = rx.recv(); assert_eq!(*sharedstate, 20); } - fn access_shared(sharedstate: &mut int, x: &RWLock, mode: RWLockMode, - n: uint) { + fn access_shared(sharedstate: &mut int, x: &Arc, + mode: RWLockMode, n: uint) { for _ in range(0, n) { lock_rwlock_in_mode(x, mode, || { let oldval = *sharedstate; @@ -1082,132 +926,127 @@ mod tests { } #[test] fn test_rwlock_readers_wont_modify_the_data() { - test_rwlock_exclusion(&RWLock::new(), Read, Write); - test_rwlock_exclusion(&RWLock::new(), Write, Read); - test_rwlock_exclusion(&RWLock::new(), Read, Downgrade); - test_rwlock_exclusion(&RWLock::new(), Downgrade, Read); + test_rwlock_exclusion(Arc::new(RWLock::new()), Read, Write); + test_rwlock_exclusion(Arc::new(RWLock::new()), Write, Read); + test_rwlock_exclusion(Arc::new(RWLock::new()), Read, Downgrade); + test_rwlock_exclusion(Arc::new(RWLock::new()), Downgrade, Read); + test_rwlock_exclusion(Arc::new(RWLock::new()), Write, DowngradeRead); + test_rwlock_exclusion(Arc::new(RWLock::new()), DowngradeRead, Write); } #[test] fn test_rwlock_writers_and_writers() { - test_rwlock_exclusion(&RWLock::new(), Write, Write); - test_rwlock_exclusion(&RWLock::new(), Write, Downgrade); - test_rwlock_exclusion(&RWLock::new(), Downgrade, Write); - test_rwlock_exclusion(&RWLock::new(), Downgrade, Downgrade); + test_rwlock_exclusion(Arc::new(RWLock::new()), Write, Write); + test_rwlock_exclusion(Arc::new(RWLock::new()), Write, Downgrade); + test_rwlock_exclusion(Arc::new(RWLock::new()), Downgrade, Write); + test_rwlock_exclusion(Arc::new(RWLock::new()), Downgrade, Downgrade); } #[cfg(test)] - fn test_rwlock_handshake(x: &RWLock, - mode1: RWLockMode, - mode2: RWLockMode, - make_mode2_go_first: bool) { + fn test_rwlock_handshake(x: Arc, + mode1: RWLockMode, + mode2: RWLockMode, + make_mode2_go_first: bool) { // Much like sem_multi_resource. let x2 = x.clone(); let (tx1, rx1) = channel(); let (tx2, rx2) = channel(); task::spawn(proc() { if !make_mode2_go_first { - let _ = rx2.recv(); // parent sends to us once it locks, or ... + rx2.recv(); // parent sends to us once it locks, or ... } lock_rwlock_in_mode(&x2, mode2, || { if make_mode2_go_first { tx1.send(()); // ... we send to it once we lock } - let _ = rx2.recv(); + rx2.recv(); tx1.send(()); }) }); if make_mode2_go_first { - let _ = rx1.recv(); // child sends to us once it locks, or ... + rx1.recv(); // child sends to us once it locks, or ... } - lock_rwlock_in_mode(x, mode1, || { + lock_rwlock_in_mode(&x, mode1, || { if !make_mode2_go_first { tx2.send(()); // ... we send to it once we lock } tx2.send(()); - let _ = rx1.recv(); + rx1.recv(); }) } #[test] fn test_rwlock_readers_and_readers() { - test_rwlock_handshake(&RWLock::new(), Read, Read, false); + test_rwlock_handshake(Arc::new(RWLock::new()), Read, Read, false); // The downgrader needs to get in before the reader gets in, otherwise // they cannot end up reading at the same time. - test_rwlock_handshake(&RWLock::new(), DowngradeRead, Read, false); - test_rwlock_handshake(&RWLock::new(), Read, DowngradeRead, true); + test_rwlock_handshake(Arc::new(RWLock::new()), DowngradeRead, Read, false); + test_rwlock_handshake(Arc::new(RWLock::new()), Read, DowngradeRead, true); // Two downgrade_reads can never both end up reading at the same time. } #[test] fn test_rwlock_downgrade_unlock() { // Tests that downgrade can unlock the lock in both modes - let x = RWLock::new(); + let x = Arc::new(RWLock::new()); lock_rwlock_in_mode(&x, Downgrade, || { }); - test_rwlock_handshake(&x, Read, Read, false); - let y = RWLock::new(); + test_rwlock_handshake(x, Read, Read, false); + let y = Arc::new(RWLock::new()); lock_rwlock_in_mode(&y, DowngradeRead, || { }); - test_rwlock_exclusion(&y, Write, Write); + test_rwlock_exclusion(y, Write, Write); } #[test] fn test_rwlock_read_recursive() { let x = RWLock::new(); - x.read(|| { x.read(|| { }) }) + let _g1 = x.read(); + let _g2 = x.read(); } #[test] fn test_rwlock_cond_wait() { // As test_mutex_cond_wait above. - let x = RWLock::new(); + let x = Arc::new(RWLock::new()); // Child wakes up parent - x.write_cond(|cond| { + { + let lock = x.write(); let x2 = x.clone(); task::spawn(proc() { - x2.write_cond(|cond| { - let woken = cond.signal(); - assert!(woken); - }) + let lock = x2.write(); + assert!(lock.cond.signal()); }); - cond.wait(); - }); + lock.cond.wait(); + } // Parent wakes up child let (tx, rx) = channel(); let x3 = x.clone(); task::spawn(proc() { - x3.write_cond(|cond| { - tx.send(()); - cond.wait(); - tx.send(()); - }) + let lock = x3.write(); + tx.send(()); + lock.cond.wait(); + tx.send(()); }); - let _ = rx.recv(); // Wait until child gets in the rwlock - x.read(|| { }); // Must be able to get in as a reader in the meantime - x.write_cond(|cond| { // Or as another writer - let woken = cond.signal(); - assert!(woken); - }); - let _ = rx.recv(); // Wait until child wakes up - x.read(|| { }); // Just for good measure + rx.recv(); // Wait until child gets in the rwlock + drop(x.read()); // Must be able to get in as a reader + { + let x = x.write(); + assert!(x.cond.signal()); + } + rx.recv(); // Wait until child wakes up + drop(x.read()); // Just for good measure } #[cfg(test)] - fn test_rwlock_cond_broadcast_helper(num_waiters: uint, - dg1: bool, - dg2: bool) { + fn test_rwlock_cond_broadcast_helper(num_waiters: uint) { // Much like the mutex broadcast test. Downgrade-enabled. - fn lock_cond(x: &RWLock, downgrade: bool, blk: |c: &Condvar|) { - if downgrade { - x.write_downgrade(|mode| { - mode.write_cond(|c| { blk(c) }); - }); - } else { - x.write_cond(|c| { blk(c) }); - } + fn lock_cond(x: &Arc, blk: |c: &Condvar|) { + let lock = x.write(); + blk(&lock.cond); } - let x = RWLock::new(); - let mut rxs = vec!(); + + let x = Arc::new(RWLock::new()); + let mut rxs = Vec::new(); for _ in range(0, num_waiters) { let xi = x.clone(); let (tx, rx) = channel(); rxs.push(rx); task::spawn(proc() { - lock_cond(&xi, dg1, |cond| { + lock_cond(&xi, |cond| { tx.send(()); cond.wait(); tx.send(()); @@ -1217,7 +1056,7 @@ mod tests { // wait until all children get in the mutex for rx in rxs.mut_iter() { let _ = rx.recv(); } - lock_cond(&x, dg2, |cond| { + lock_cond(&x, |cond| { let num_woken = cond.broadcast(); assert_eq!(num_woken, num_waiters); }); @@ -1226,21 +1065,15 @@ mod tests { } #[test] fn test_rwlock_cond_broadcast() { - test_rwlock_cond_broadcast_helper(0, true, true); - test_rwlock_cond_broadcast_helper(0, true, false); - test_rwlock_cond_broadcast_helper(0, false, true); - test_rwlock_cond_broadcast_helper(0, false, false); - test_rwlock_cond_broadcast_helper(12, true, true); - test_rwlock_cond_broadcast_helper(12, true, false); - test_rwlock_cond_broadcast_helper(12, false, true); - test_rwlock_cond_broadcast_helper(12, false, false); + test_rwlock_cond_broadcast_helper(0); + test_rwlock_cond_broadcast_helper(12); } #[cfg(test)] fn rwlock_kill_helper(mode1: RWLockMode, mode2: RWLockMode) { use std::any::Any; // Mutex must get automatically unlocked if failed/killed within. - let x = RWLock::new(); + let x = Arc::new(RWLock::new()); let x2 = x.clone(); let result: result::Result<(), ~Any> = task::try(proc() { @@ -1283,48 +1116,4 @@ mod tests { rwlock_kill_helper(Downgrade, DowngradeRead); rwlock_kill_helper(Downgrade, DowngradeRead); } - #[test] #[should_fail] - fn test_rwlock_downgrade_cant_swap() { - // Tests that you can't downgrade with a different rwlock's token. - let x = RWLock::new(); - let y = RWLock::new(); - x.write_downgrade(|xwrite| { - let mut xopt = Some(xwrite); - y.write_downgrade(|_ywrite| { - y.downgrade(xopt.take_unwrap()); - error!("oops, y.downgrade(x) should have failed!"); - }) - }) - } - - /************************************************************************ - * Barrier tests - ************************************************************************/ - #[test] - fn test_barrier() { - let barrier = Barrier::new(10); - let (tx, rx) = channel(); - - for _ in range(0, 9) { - let c = barrier.clone(); - let tx = tx.clone(); - spawn(proc() { - c.wait(); - tx.send(true); - }); - } - - // At this point, all spawned tasks should be blocked, - // so we shouldn't get anything from the port - assert!(match rx.try_recv() { - Empty => true, - _ => false, - }); - - barrier.wait(); - // Now, the barrier is cleared and we should get data. - for _ in range(0, 9) { - rx.recv(); - } - } }