move concurrent stuff from libextra to libsync

4 files changed, 2284 insertions, 0 deletions

diff --git a/src/libsync/sync/mod.rs b/src/libsync/sync/mod.rs
new file mode 100644
index 00000000000..cfff31e08d4
--- /dev/null
+++ b/src/libsync/sync/mod.rs
@@ -0,0 +1,1420 @@
+// Copyright 2012-2014 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// 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.
+ */
+
+use std::cast;
+use std::comm;
+use std::kinds::marker;
+use std::sync::arc::UnsafeArc;
+use std::sync::atomics;
+use std::unstable::finally::Finally;
+use std::util;
+
+use arc::MutexArc;
+
+/****************************************************************************
+ * Internals
+ ****************************************************************************/
+
+pub mod mutex;
+pub mod one;
+mod mpsc_intrusive;
+
+// Each waiting task receives on one of these.
+#[doc(hidden)]
+type WaitEnd = Port<()>;
+#[doc(hidden)]
+type SignalEnd = Chan<()>;
+// A doubly-ended queue of waiting tasks.
+#[doc(hidden)]
+struct WaitQueue { head: Port<SignalEnd>,
+                   tail: Chan<SignalEnd> }
+
+impl WaitQueue {
+    fn new() -> WaitQueue {
+        let (block_head, block_tail) = Chan::new();
+        WaitQueue { head: block_head, tail: block_tail }
+    }
+
+    // Signals one live task from the queue.
+    fn signal(&self) -> bool {
+        match self.head.try_recv() {
+            comm::Data(ch) => {
+                // Send a wakeup signal. If the waiter was killed, its port will
+                // have closed. Keep trying until we get a live task.
+                if ch.try_send(()) {
+                    true
+                } else {
+                    self.signal()
+                }
+            }
+            _ => false
+        }
+    }
+
+    fn broadcast(&self) -> uint {
+        let mut count = 0;
+        loop {
+            match self.head.try_recv() {
+                comm::Data(ch) => {
+                    if ch.try_send(()) {
+                        count += 1;
+                    }
+                }
+                _ => break
+            }
+        }
+        count
+    }
+
+    fn wait_end(&self) -> WaitEnd {
+        let (wait_end, signal_end) = Chan::new();
+        assert!(self.tail.try_send(signal_end));
+        wait_end
+    }
+}
+
+// The building-block used to make semaphores, mutexes, and rwlocks.
+struct SemInner<Q> {
+    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
+}
+
+struct Sem<Q>(UnsafeArc<SemInner<Q>>);
+
+#[doc(hidden)]
+impl<Q:Send> Sem<Q> {
+    fn new(count: int, q: Q) -> Sem<Q> {
+        Sem(UnsafeArc::new(SemInner {
+            count: count,
+            waiters: WaitQueue::new(),
+            blocked: q,
+            lock: mutex::Mutex::new(),
+        }))
+    }
+
+    unsafe fn with(&self, f: |&mut SemInner<Q>|) {
+        let Sem(ref arc) = *self;
+        let state = arc.get();
+        let _g = (*state).lock.lock();
+        f(cast::transmute(state));
+    }
+
+    pub fn acquire(&self) {
+        unsafe {
+            let mut waiter_nobe = None;
+            self.with(|state| {
+                state.count -= 1;
+                if state.count < 0 {
+                    // Create waiter nobe, enqueue ourself, and tell
+                    // outer scope we need to block.
+                    waiter_nobe = Some(state.waiters.wait_end());
+                }
+            });
+            // 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() {
+                let _ = waiter_nobe.unwrap().recv();
+            }
+        }
+    }
+
+    pub fn release(&self) {
+        unsafe {
+            self.with(|state| {
+                state.count += 1;
+                if state.count <= 0 {
+                    state.waiters.signal();
+                }
+            })
+        }
+    }
+
+    pub fn access<U>(&self, blk: || -> U) -> U {
+        (|| {
+            self.acquire();
+            blk()
+        }).finally(|| {
+            self.release();
+        })
+    }
+}
+
+#[doc(hidden)]
+impl Sem<~[WaitQueue]> {
+    fn new_and_signal(count: int, num_condvars: uint)
+        -> Sem<~[WaitQueue]> {
+        let mut queues = ~[];
+        for _ in range(0, num_condvars) { queues.push(WaitQueue::new()); }
+        Sem::new(count, queues)
+    }
+}
+
+// FIXME(#3598): Want to use an Option down below, but we need a custom enum
+// that's not polymorphic to get around the fact that lifetimes are invariant
+// inside of type parameters.
+enum ReacquireOrderLock<'a> {
+    Nothing, // c.c
+    Just(&'a Semaphore),
+}
+
+/// A mechanism for atomic-unlock-and-deschedule blocking and signalling.
+pub struct Condvar<'a> {
+    // The 'Sem' object associated with this condvar. This is the one that's
+    // atomically-unlocked-and-descheduled upon and reacquired during wakeup.
+    priv sem: &'a Sem<~[WaitQueue]>,
+    // This is (can be) an extra semaphore which is held around the reacquire
+    // operation on the first one. This is only used in cvars associated with
+    // rwlocks, and is needed to ensure that, when a downgrader is trying to
+    // hand off the access lock (which would be the first field, here), a 2nd
+    // writer waking up from a cvar wait can't race with a reader to steal it,
+    // See the comment in write_cond for more detail.
+    priv order: ReacquireOrderLock<'a>,
+    // Make sure condvars are non-copyable.
+    priv nopod: marker::NoPod,
+}
+
+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.
+     */
+    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).
+     */
+    pub fn wait_on(&self, condvar_id: uint) {
+        let mut WaitEnd = None;
+        let mut out_of_bounds = None;
+        // Release lock, 'atomically' enqueuing ourselves in so doing.
+        unsafe {
+            self.sem.with(|state| {
+                if condvar_id < state.blocked.len() {
+                    // Drop the lock.
+                    state.count += 1;
+                    if state.count <= 0 {
+                        state.waiters.signal();
+                    }
+                    // Create waiter nobe, and enqueue ourself to
+                    // be woken up by a signaller.
+                    WaitEnd = Some(state.blocked[condvar_id].wait_end());
+                } else {
+                    out_of_bounds = Some(state.blocked.len());
+                }
+            })
+        }
+
+        // If deschedule checks start getting inserted anywhere, we can be
+        // killed before or after enqueueing.
+        check_cvar_bounds(out_of_bounds, condvar_id, "cond.wait_on()", || {
+            // Unconditionally "block". (Might not actually block if a
+            // signaller already sent -- I mean 'unconditionally' in contrast
+            // with acquire().)
+            (|| {
+                let _ = WaitEnd.take_unwrap().recv();
+            }).finally(|| {
+                // Reacquire the condvar.
+                match self.order {
+                    Just(lock) => lock.access(|| self.sem.acquire()),
+                    Nothing => self.sem.acquire(),
+                }
+            })
+        })
+    }
+
+    /// Wake up a blocked task. Returns false if there was no blocked task.
+    pub fn signal(&self) -> bool { self.signal_on(0) }
+
+    /// As signal, but with a specified condvar_id. See wait_on.
+    pub fn signal_on(&self, condvar_id: uint) -> bool {
+        unsafe {
+            let mut out_of_bounds = None;
+            let mut result = false;
+            self.sem.with(|state| {
+                if condvar_id < state.blocked.len() {
+                    result = state.blocked[condvar_id].signal();
+                } else {
+                    out_of_bounds = Some(state.blocked.len());
+                }
+            });
+            check_cvar_bounds(out_of_bounds,
+                              condvar_id,
+                              "cond.signal_on()",
+                              || result)
+        }
+    }
+
+    /// Wake up all blocked tasks. Returns the number of tasks woken.
+    pub fn broadcast(&self) -> uint { self.broadcast_on(0) }
+
+    /// As broadcast, but with a specified condvar_id. See wait_on.
+    pub fn broadcast_on(&self, condvar_id: uint) -> uint {
+        let mut out_of_bounds = None;
+        let mut queue = None;
+        unsafe {
+            self.sem.with(|state| {
+                if condvar_id < state.blocked.len() {
+                    // To avoid :broadcast_heavy, we make a new waitqueue,
+                    // swap it out with the old one, and broadcast on the
+                    // old one outside of the little-lock.
+                    queue = Some(util::replace(&mut state.blocked[condvar_id],
+                                               WaitQueue::new()));
+                } else {
+                    out_of_bounds = Some(state.blocked.len());
+                }
+            });
+            check_cvar_bounds(out_of_bounds,
+                              condvar_id,
+                              "cond.signal_on()",
+                              || {
+                queue.take_unwrap().broadcast()
+            })
+        }
+    }
+}
+
+// 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<U>(
+                     out_of_bounds: Option<uint>,
+                     id: uint,
+                     act: &str,
+                     blk: || -> U)
+                     -> U {
+    match out_of_bounds {
+        Some(0) =>
+            fail!("{} with illegal ID {} - this lock has no condvars!", act, id),
+        Some(length) =>
+            fail!("{} with illegal ID {} - ID must be less than {}", act, id, length),
+        None => blk()
+    }
+}
+
+#[doc(hidden)]
+impl Sem<~[WaitQueue]> {
+    // The only other places that condvars get built are rwlock.write_cond()
+    // and rwlock_write_mode.
+    pub fn access_cond<U>(&self, blk: |c: &Condvar| -> U) -> U {
+        self.access(|| {
+            blk(&Condvar {
+                sem: self,
+                order: Nothing,
+                nopod: marker::NoPod
+            })
+        })
+    }
+}
+
+/****************************************************************************
+ * Semaphores
+ ****************************************************************************/
+
+/// A counting, blocking, bounded-waiting semaphore.
+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()) }
+    }
+}
+
+impl Semaphore {
+    /// Create a new semaphore with the specified count.
+    pub fn new(count: int) -> 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() }
+
+    /**
+     * 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<U>(&self, blk: || -> U) -> U { (&self.sem).access(blk) }
+}
+
+/****************************************************************************
+ * 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.
+ */
+
+pub struct Mutex { priv sem: Sem<~[WaitQueue]> }
+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()) } }
+}
+
+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.)
+    */
+    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<U>(&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<U>(&self, blk: |c: &Condvar| -> U) -> U {
+        (&self.sem).access_cond(blk)
+    }
+}
+
+/****************************************************************************
+ * Reader-writer locks
+ ****************************************************************************/
+
+// 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,
+    // 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,
+}
+
+/**
+ * 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<~[WaitQueue]>,
+    priv state:       UnsafeArc<RWLockInner>,
+}
+
+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.
+    */
+    pub fn new_with_condvars(num_condvars: uint) -> RWLock {
+        let state = UnsafeArc::new(RWLockInner {
+            read_mode:  false,
+            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<U>(&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<U>(&self, blk: || -> U) -> U {
+        (&self.order_lock).acquire();
+        (&self.access_lock).access(|| {
+            (&self.order_lock).release();
+            blk()
+        })
+    }
+
+    /**
+     * 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<U>(&self, blk: |c: &Condvar| -> U) -> U {
+        // 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",
+        // there could arise a subtle race when a downgrader attempts to hand
+        // off the reader cloud lock to a waiting reader. This race is tested
+        // in arc.rs (test_rw_write_cond_downgrade_read_race) and looks like:
+        // T1 (writer)              T2 (downgrader)             T3 (reader)
+        // [in cond.wait()]
+        //                          [locks for writing]
+        //                          [holds access_lock]
+        // [is signalled, perhaps by
+        //  downgrader or a 4th thread]
+        // tries to lock access(!)
+        //                                                      lock order_lock
+        //                                                      xadd read_count[0->1]
+        //                                                      tries to lock access
+        //                          [downgrade]
+        //                          xadd read_count[1->2]
+        //                          unlock access
+        // Since T1 contended on the access lock before T3 did, it will steal
+        // the lock handoff. Adding order_lock in the condvar reacquire path
+        // solves this because T1 will hold order_lock while waiting on access,
+        // which will cause T3 to have to wait until T1 finishes its write,
+        // 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<U>(&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();
+            }
+        }
+        RWLockReadMode { lock: token.lock, nopod: marker::NoPod }
+    }
+}
+
+/// The "write permission" token used for rwlock.write_downgrade().
+
+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<U>(&self, blk: || -> U) -> U { blk() }
+    /// Access the pre-downgrade rwlock in write mode with a condvar.
+    pub fn write_cond<U>(&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<U>(&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<BarrierState>,
+    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,
+        }
+    }
+
+    /// 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();
+            }
+        });
+    }
+}
+
+/****************************************************************************
+ * Tests
+ ****************************************************************************/
+
+#[cfg(test)]
+mod tests {
+    use sync::{Semaphore, Mutex, RWLock, Barrier, Condvar};
+
+    use std::cast;
+    use std::result;
+    use std::task;
+    use std::comm::{SharedChan, Empty};
+
+    /************************************************************************
+     * Semaphore tests
+     ************************************************************************/
+    #[test]
+    fn test_sem_acquire_release() {
+        let s = Semaphore::new(1);
+        s.acquire();
+        s.release();
+        s.acquire();
+    }
+    #[test]
+    fn test_sem_basic() {
+        let s = Semaphore::new(1);
+        s.access(|| { })
+    }
+    #[test]
+    fn test_sem_as_mutex() {
+        let s = Semaphore::new(1);
+        let s2 = s.clone();
+        task::spawn(proc() {
+            s2.access(|| {
+                for _ in range(0, 5) { task::deschedule(); }
+            })
+        });
+        s.access(|| {
+            for _ in range(0, 5) { task::deschedule(); }
+        })
+    }
+    #[test]
+    fn test_sem_as_cvar() {
+        /* Child waits and parent signals */
+        let (p, c) = Chan::new();
+        let s = Semaphore::new(0);
+        let s2 = s.clone();
+        task::spawn(proc() {
+            s2.acquire();
+            c.send(());
+        });
+        for _ in range(0, 5) { task::deschedule(); }
+        s.release();
+        let _ = p.recv();
+
+        /* Parent waits and child signals */
+        let (p, c) = Chan::new();
+        let s = Semaphore::new(0);
+        let s2 = s.clone();
+        task::spawn(proc() {
+            for _ in range(0, 5) { task::deschedule(); }
+            s2.release();
+            let _ = p.recv();
+        });
+        s.acquire();
+        c.send(());
+    }
+    #[test]
+    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 s2 = s.clone();
+        let (p1,c1) = Chan::new();
+        let (p2,c2) = Chan::new();
+        task::spawn(proc() {
+            s2.access(|| {
+                let _ = p2.recv();
+                c1.send(());
+            })
+        });
+        s.access(|| {
+            c2.send(());
+            let _ = p1.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 s2 = s.clone();
+        let (p, c) = Chan::new();
+        let mut child_data = Some((s2, c));
+        s.access(|| {
+            let (s2, c) = child_data.take_unwrap();
+            task::spawn(proc() {
+                c.send(());
+                s2.access(|| { });
+                c.send(());
+            });
+            let _ = p.recv(); // wait for child to come alive
+            for _ in range(0, 5) { task::deschedule(); } // let the child contend
+        });
+        let _ = p.recv(); // wait for child to be done
+    }
+    /************************************************************************
+     * Mutex tests
+     ************************************************************************/
+    #[test]
+    fn test_mutex_lock() {
+        // Unsafely achieve shared state, and do the textbook
+        // "load tmp = move ptr; inc tmp; store ptr <- tmp" dance.
+        let (p, c) = Chan::new();
+        let m = Mutex::new();
+        let m2 = m.clone();
+        let mut sharedstate = ~0;
+        {
+            let ptr: *int = &*sharedstate;
+            task::spawn(proc() {
+                let sharedstate: &mut int =
+                    unsafe { cast::transmute(ptr) };
+                access_shared(sharedstate, &m2, 10);
+                c.send(());
+            });
+        }
+        {
+            access_shared(sharedstate, &m, 10);
+            let _ = p.recv();
+
+            assert_eq!(*sharedstate, 20);
+        }
+
+        fn access_shared(sharedstate: &mut int, m: &Mutex, n: uint) {
+            for _ in range(0, n) {
+                m.lock(|| {
+                    let oldval = *sharedstate;
+                    task::deschedule();
+                    *sharedstate = oldval + 1;
+                })
+            }
+        }
+    }
+    #[test]
+    fn test_mutex_cond_wait() {
+        let m = Mutex::new();
+
+        // Child wakes up parent
+        m.lock_cond(|cond| {
+            let m2 = m.clone();
+            task::spawn(proc() {
+                m2.lock_cond(|cond| {
+                    let woken = cond.signal();
+                    assert!(woken);
+                })
+            });
+            cond.wait();
+        });
+        // Parent wakes up child
+        let (port,chan) = Chan::new();
+        let m3 = m.clone();
+        task::spawn(proc() {
+            m3.lock_cond(|cond| {
+                chan.send(());
+                cond.wait();
+                chan.send(());
+            })
+        });
+        let _ = port.recv(); // Wait until child gets in the mutex
+        m.lock_cond(|cond| {
+            let woken = cond.signal();
+            assert!(woken);
+        });
+        let _ = port.recv(); // Wait until child wakes up
+    }
+    #[cfg(test)]
+    fn test_mutex_cond_broadcast_helper(num_waiters: uint) {
+        let m = Mutex::new();
+        let mut ports = ~[];
+
+        for _ in range(0, num_waiters) {
+            let mi = m.clone();
+            let (port, chan) = Chan::new();
+            ports.push(port);
+            task::spawn(proc() {
+                mi.lock_cond(|cond| {
+                    chan.send(());
+                    cond.wait();
+                    chan.send(());
+                })
+            });
+        }
+
+        // wait until all children get in the mutex
+        for port in ports.mut_iter() { let _ = port.recv(); }
+        m.lock_cond(|cond| {
+            let num_woken = cond.broadcast();
+            assert_eq!(num_woken, num_waiters);
+        });
+        // wait until all children wake up
+        for port in ports.mut_iter() { let _ = port.recv(); }
+    }
+    #[test]
+    fn test_mutex_cond_broadcast() {
+        test_mutex_cond_broadcast_helper(12);
+    }
+    #[test]
+    fn test_mutex_cond_broadcast_none() {
+        test_mutex_cond_broadcast_helper(0);
+    }
+    #[test]
+    fn test_mutex_cond_no_waiter() {
+        let m = Mutex::new();
+        let m2 = m.clone();
+        let _ = task::try(proc() {
+            m.lock_cond(|_x| { })
+        });
+        m2.lock_cond(|cond| {
+            assert!(!cond.signal());
+        })
+    }
+    #[test]
+    fn test_mutex_killed_simple() {
+        // Mutex must get automatically unlocked if failed/killed within.
+        let m = Mutex::new();
+        let m2 = m.clone();
+
+        let result: result::Result<(), ~Any> = task::try(proc() {
+            m2.lock(|| {
+                fail!();
+            })
+        });
+        assert!(result.is_err());
+        // child task must have finished by the time try returns
+        m.lock(|| { })
+    }
+    #[ignore(reason = "linked failure")]
+    #[test]
+    fn test_mutex_killed_cond() {
+        // Getting killed during cond wait must not corrupt the mutex while
+        // unwinding (e.g. double unlock).
+        let m = Mutex::new();
+        let m2 = m.clone();
+
+        let result: result::Result<(), ~Any> = task::try(proc() {
+            let (p, c) = Chan::new();
+            task::spawn(proc() { // linked
+                let _ = p.recv(); // wait for sibling to get in the mutex
+                task::deschedule();
+                fail!();
+            });
+            m2.lock_cond(|cond| {
+                c.send(()); // tell sibling go ahead
+                cond.wait(); // block forever
+            })
+        });
+        assert!(result.is_err());
+        // child task must have finished by the time try returns
+        m.lock_cond(|cond| {
+            let woken = cond.signal();
+            assert!(!woken);
+        })
+    }
+    #[ignore(reason = "linked failure")]
+    #[test]
+    fn test_mutex_killed_broadcast() {
+        use std::unstable::finally::Finally;
+
+        let m = Mutex::new();
+        let m2 = m.clone();
+        let (p, c) = Chan::new();
+
+        let result: result::Result<(), ~Any> = task::try(proc() {
+            let mut sibling_convos = ~[];
+            for _ in range(0, 2) {
+                let (p, c) = Chan::new();
+                sibling_convos.push(p);
+                let mi = m2.clone();
+                // spawn sibling task
+                task::spawn(proc() { // linked
+                    mi.lock_cond(|cond| {
+                        c.send(()); // tell sibling to go ahead
+                        (|| {
+                            cond.wait(); // block forever
+                        }).finally(|| {
+                            error!("task unwinding and sending");
+                            c.send(());
+                            error!("task unwinding and done sending");
+                        })
+                    })
+                });
+            }
+            for p in sibling_convos.mut_iter() {
+                let _ = p.recv(); // wait for sibling to get in the mutex
+            }
+            m2.lock(|| { });
+            c.send(sibling_convos); // let parent wait on all children
+            fail!();
+        });
+        assert!(result.is_err());
+        // child task must have finished by the time try returns
+        let mut r = p.recv();
+        for p in r.mut_iter() { p.recv(); } // wait on all its siblings
+        m.lock_cond(|cond| {
+            let woken = cond.broadcast();
+            assert_eq!(woken, 0);
+        })
+    }
+    #[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();
+        })
+    }
+    #[test]
+    #[ignore(reason = "linked failure?")]
+    fn test_mutex_different_conds() {
+        let result = task::try(proc() {
+            let m = Mutex::new_with_condvars(2);
+            let m2 = m.clone();
+            let (p, c) = Chan::new();
+            task::spawn(proc() {
+                m2.lock_cond(|cond| {
+                    c.send(());
+                    cond.wait_on(1);
+                })
+            });
+            let _ = p.recv();
+            m.lock_cond(|cond| {
+                if !cond.signal_on(0) {
+                    fail!(); // success; punt sibling awake.
+                }
+            })
+        });
+        assert!(result.is_err());
+    }
+    #[test]
+    fn test_mutex_no_condvars() {
+        let result = task::try(proc() {
+            let m = Mutex::new_with_condvars(0);
+            m.lock_cond(|cond| { cond.wait(); })
+        });
+        assert!(result.is_err());
+        let result = task::try(proc() {
+            let m = Mutex::new_with_condvars(0);
+            m.lock_cond(|cond| { cond.signal(); })
+        });
+        assert!(result.is_err());
+        let result = task::try(proc() {
+            let m = Mutex::new_with_condvars(0);
+            m.lock_cond(|cond| { cond.broadcast(); })
+        });
+        assert!(result.is_err());
+    }
+    /************************************************************************
+     * Reader/writer lock tests
+     ************************************************************************/
+    #[cfg(test)]
+    pub enum RWLockMode { Read, Write, Downgrade, DowngradeRead }
+    #[cfg(test)]
+    fn lock_rwlock_in_mode(x: &RWLock, 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() });
+                }),
+        }
+    }
+    #[cfg(test)]
+    fn test_rwlock_exclusion(x: &RWLock,
+                                 mode1: RWLockMode,
+                                 mode2: RWLockMode) {
+        // Test mutual exclusion between readers and writers. Just like the
+        // mutex mutual exclusion test, a ways above.
+        let (p, c) = Chan::new();
+        let x2 = x.clone();
+        let mut sharedstate = ~0;
+        {
+            let ptr: *int = &*sharedstate;
+            task::spawn(proc() {
+                let sharedstate: &mut int =
+                    unsafe { cast::transmute(ptr) };
+                access_shared(sharedstate, &x2, mode1, 10);
+                c.send(());
+            });
+        }
+        {
+            access_shared(sharedstate, x, mode2, 10);
+            let _ = p.recv();
+
+            assert_eq!(*sharedstate, 20);
+        }
+
+        fn access_shared(sharedstate: &mut int, x: &RWLock, mode: RWLockMode,
+                         n: uint) {
+            for _ in range(0, n) {
+                lock_rwlock_in_mode(x, mode, || {
+                    let oldval = *sharedstate;
+                    task::deschedule();
+                    *sharedstate = oldval + 1;
+                })
+            }
+        }
+    }
+    #[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]
+    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);
+    }
+    #[cfg(test)]
+    fn test_rwlock_handshake(x: &RWLock,
+                                 mode1: RWLockMode,
+                                 mode2: RWLockMode,
+                                 make_mode2_go_first: bool) {
+        // Much like sem_multi_resource.
+        let x2 = x.clone();
+        let (p1, c1) = Chan::new();
+        let (p2, c2) = Chan::new();
+        task::spawn(proc() {
+            if !make_mode2_go_first {
+                let _ = p2.recv(); // parent sends to us once it locks, or ...
+            }
+            lock_rwlock_in_mode(&x2, mode2, || {
+                if make_mode2_go_first {
+                    c1.send(()); // ... we send to it once we lock
+                }
+                let _ = p2.recv();
+                c1.send(());
+            })
+        });
+        if make_mode2_go_first {
+            let _ = p1.recv(); // child sends to us once it locks, or ...
+        }
+        lock_rwlock_in_mode(x, mode1, || {
+            if !make_mode2_go_first {
+                c2.send(()); // ... we send to it once we lock
+            }
+            c2.send(());
+            let _ = p1.recv();
+        })
+    }
+    #[test]
+    fn test_rwlock_readers_and_readers() {
+        test_rwlock_handshake(&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);
+        // 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();
+        lock_rwlock_in_mode(&x, Downgrade, || { });
+        test_rwlock_handshake(&x, Read, Read, false);
+        let y = RWLock::new();
+        lock_rwlock_in_mode(&y, DowngradeRead, || { });
+        test_rwlock_exclusion(&y, Write, Write);
+    }
+    #[test]
+    fn test_rwlock_read_recursive() {
+        let x = RWLock::new();
+        x.read(|| { x.read(|| { }) })
+    }
+    #[test]
+    fn test_rwlock_cond_wait() {
+        // As test_mutex_cond_wait above.
+        let x = RWLock::new();
+
+        // Child wakes up parent
+        x.write_cond(|cond| {
+            let x2 = x.clone();
+            task::spawn(proc() {
+                x2.write_cond(|cond| {
+                    let woken = cond.signal();
+                    assert!(woken);
+                })
+            });
+            cond.wait();
+        });
+        // Parent wakes up child
+        let (port, chan) = Chan::new();
+        let x3 = x.clone();
+        task::spawn(proc() {
+            x3.write_cond(|cond| {
+                chan.send(());
+                cond.wait();
+                chan.send(());
+            })
+        });
+        let _ = port.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 _ = port.recv(); // Wait until child wakes up
+        x.read(|| { }); // Just for good measure
+    }
+    #[cfg(test)]
+    fn test_rwlock_cond_broadcast_helper(num_waiters: uint,
+                                             dg1: bool,
+                                             dg2: bool) {
+        // 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) });
+            }
+        }
+        let x = RWLock::new();
+        let mut ports = ~[];
+
+        for _ in range(0, num_waiters) {
+            let xi = x.clone();
+            let (port, chan) = Chan::new();
+            ports.push(port);
+            task::spawn(proc() {
+                lock_cond(&xi, dg1, |cond| {
+                    chan.send(());
+                    cond.wait();
+                    chan.send(());
+                })
+            });
+        }
+
+        // wait until all children get in the mutex
+        for port in ports.mut_iter() { let _ = port.recv(); }
+        lock_cond(&x, dg2, |cond| {
+            let num_woken = cond.broadcast();
+            assert_eq!(num_woken, num_waiters);
+        });
+        // wait until all children wake up
+        for port in ports.mut_iter() { let _ = port.recv(); }
+    }
+    #[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);
+    }
+    #[cfg(test)]
+    fn rwlock_kill_helper(mode1: RWLockMode, mode2: RWLockMode) {
+        // Mutex must get automatically unlocked if failed/killed within.
+        let x = RWLock::new();
+        let x2 = x.clone();
+
+        let result: result::Result<(), ~Any> = task::try(proc() {
+            lock_rwlock_in_mode(&x2, mode1, || {
+                fail!();
+            })
+        });
+        assert!(result.is_err());
+        // child task must have finished by the time try returns
+        lock_rwlock_in_mode(&x, mode2, || { })
+    }
+    #[test]
+    fn test_rwlock_reader_killed_writer() {
+        rwlock_kill_helper(Read, Write);
+    }
+    #[test]
+    fn test_rwlock_writer_killed_reader() {
+        rwlock_kill_helper(Write, Read);
+    }
+    #[test]
+    fn test_rwlock_reader_killed_reader() {
+        rwlock_kill_helper(Read, Read);
+    }
+    #[test]
+    fn test_rwlock_writer_killed_writer() {
+        rwlock_kill_helper(Write, Write);
+    }
+    #[test]
+    fn test_rwlock_kill_downgrader() {
+        rwlock_kill_helper(Downgrade, Read);
+        rwlock_kill_helper(Read, Downgrade);
+        rwlock_kill_helper(Downgrade, Write);
+        rwlock_kill_helper(Write, Downgrade);
+        rwlock_kill_helper(DowngradeRead, Read);
+        rwlock_kill_helper(Read, DowngradeRead);
+        rwlock_kill_helper(DowngradeRead, Write);
+        rwlock_kill_helper(Write, DowngradeRead);
+        rwlock_kill_helper(DowngradeRead, Downgrade);
+        rwlock_kill_helper(DowngradeRead, Downgrade);
+        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 (port, chan) = SharedChan::new();
+
+        for _ in range(0, 9) {
+            let c = barrier.clone();
+            let chan = chan.clone();
+            spawn(proc() {
+                c.wait();
+                chan.send(true);
+            });
+        }
+
+        // At this point, all spawned tasks should be blocked,
+        // so we shouldn't get anything from the port
+        assert!(match port.try_recv() {
+            Empty => true,
+            _ => false,
+        });
+
+        barrier.wait();
+        // Now, the barrier is cleared and we should get data.
+        for _ in range(0, 9) {
+            port.recv();
+        }
+    }
+}
diff --git a/src/libsync/sync/mpsc_intrusive.rs b/src/libsync/sync/mpsc_intrusive.rs
new file mode 100644
index 00000000000..0f13a4980d9
--- /dev/null
+++ b/src/libsync/sync/mpsc_intrusive.rs
@@ -0,0 +1,139 @@
+/* Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *    1. Redistributions of source code must retain the above copyright notice,
+ *       this list of conditions and the following disclaimer.
+ *
+ *    2. Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in the
+ *       documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED
+ * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
+ * EVENT SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
+ * OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
+ * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * The views and conclusions contained in the software and documentation are
+ * those of the authors and should not be interpreted as representing official
+ * policies, either expressed or implied, of Dmitry Vyukov.
+ */
+
+//! A mostly lock-free multi-producer, single consumer queue.
+//!
+//! This module implements an intrusive MPSC queue. This queue is incredibly
+//! unsafe (due to use of unsafe pointers for nodes), and hence is not public.
+
+// http://www.1024cores.net/home/lock-free-algorithms
+//                         /queues/intrusive-mpsc-node-based-queue
+
+use std::cast;
+use std::sync::atomics;
+
+// NB: all links are done as AtomicUint instead of AtomicPtr to allow for static
+// initialization.
+
+pub struct Node<T> {
+    next: atomics::AtomicUint,
+    data: T,
+}
+
+pub struct DummyNode {
+    next: atomics::AtomicUint,
+}
+
+pub struct Queue<T> {
+    head: atomics::AtomicUint,
+    tail: *mut Node<T>,
+    stub: DummyNode,
+}
+
+impl<T: Send> Queue<T> {
+    pub fn new() -> Queue<T> {
+        Queue {
+            head: atomics::AtomicUint::new(0),
+            tail: 0 as *mut Node<T>,
+            stub: DummyNode {
+                next: atomics::AtomicUint::new(0),
+            },
+        }
+    }
+
+    pub unsafe fn push(&mut self, node: *mut Node<T>) {
+        (*node).next.store(0, atomics::Release);
+        let prev = self.head.swap(node as uint, atomics::AcqRel);
+
+        // Note that this code is slightly modified to allow static
+        // initialization of these queues with rust's flavor of static
+        // initialization.
+        if prev == 0 {
+            self.stub.next.store(node as uint, atomics::Release);
+        } else {
+            let prev = prev as *mut Node<T>;
+            (*prev).next.store(node as uint, atomics::Release);
+        }
+    }
+
+    /// You'll note that the other MPSC queue in std::sync is non-intrusive and
+    /// returns a `PopResult` here to indicate when the queue is inconsistent.
+    /// An "inconsistent state" in the other queue means that a pusher has
+    /// pushed, but it hasn't finished linking the rest of the chain.
+    ///
+    /// This queue also suffers from this problem, but I currently haven't been
+    /// able to detangle when this actually happens. This code is translated
+    /// verbatim from the website above, and is more complicated than the
+    /// non-intrusive version.
+    ///
+    /// Right now consumers of this queue must be ready for this fact. Just
+    /// because `pop` returns `None` does not mean that there is not data
+    /// on the queue.
+    pub unsafe fn pop(&mut self) -> Option<*mut Node<T>> {
+        let tail = self.tail;
+        let mut tail = if !tail.is_null() {tail} else {
+            cast::transmute(&self.stub)
+        };
+        let mut next = (*tail).next(atomics::Relaxed);
+        if tail as uint == &self.stub as *DummyNode as uint {
+            if next.is_null() {
+                return None;
+            }
+            self.tail = next;
+            tail = next;
+            next = (*next).next(atomics::Relaxed);
+        }
+        if !next.is_null() {
+            self.tail = next;
+            return Some(tail);
+        }
+        let head = self.head.load(atomics::Acquire) as *mut Node<T>;
+        if tail != head {
+            return None;
+        }
+        let stub = cast::transmute(&self.stub);
+        self.push(stub);
+        next = (*tail).next(atomics::Relaxed);
+        if !next.is_null() {
+            self.tail = next;
+            return Some(tail);
+        }
+        return None
+    }
+}
+
+impl<T: Send> Node<T> {
+    pub fn new(t: T) -> Node<T> {
+        Node {
+            data: t,
+            next: atomics::AtomicUint::new(0),
+        }
+    }
+    pub unsafe fn next(&mut self, ord: atomics::Ordering) -> *mut Node<T> {
+        cast::transmute::<uint, *mut Node<T>>(self.next.load(ord))
+    }
+}
diff --git a/src/libsync/sync/mutex.rs b/src/libsync/sync/mutex.rs
new file mode 100644
index 00000000000..f1a81d65c1d
--- /dev/null
+++ b/src/libsync/sync/mutex.rs
@@ -0,0 +1,557 @@
+// Copyright 2014 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+//! A proper mutex implementation regardless of the "flavor of task" which is
+//! acquiring the lock.
+
+// # Implementation of Rust mutexes
+//
+// Most answers to the question of "how do I use a mutex" are "use pthreads",
+// but for Rust this isn't quite sufficient. Green threads cannot acquire an OS
+// mutex because they can context switch among many OS threads, leading to
+// deadlocks with other green threads.
+//
+// Another problem for green threads grabbing an OS mutex is that POSIX dictates
+// that unlocking a mutex on a different thread from where it was locked is
+// undefined behavior. Remember that green threads can migrate among OS threads,
+// so this would mean that we would have to pin green threads to OS threads,
+// which is less than ideal.
+//
+// ## Using deschedule/reawaken
+//
+// We already have primitives for descheduling/reawakening tasks, so they're the
+// first obvious choice when implementing a mutex. The idea would be to have a
+// concurrent queue that everyone is pushed on to, and then the owner of the
+// mutex is the one popping from the queue.
+//
+// Unfortunately, this is not very performant for native tasks. The suspected
+// reason for this is that each native thread is suspended on its own condition
+// variable, unique from all the other threads. In this situation, the kernel
+// has no idea what the scheduling semantics are of the user program, so all of
+// the threads are distributed among all cores on the system. This ends up
+// having very expensive wakeups of remote cores high up in the profile when
+// handing off the mutex among native tasks. On the other hand, when using an OS
+// mutex, the kernel knows that all native threads are contended on the same
+// mutex, so they're in theory all migrated to a single core (fast context
+// switching).
+//
+// ## Mixing implementations
+//
+// From that above information, we have two constraints. The first is that
+// green threads can't touch os mutexes, and the second is that native tasks
+// pretty much *must* touch an os mutex.
+//
+// As a compromise, the queueing implementation is used for green threads and
+// the os mutex is used for native threads (why not have both?). This ends up
+// leading to fairly decent performance for both native threads and green
+// threads on various workloads (uncontended and contended).
+//
+// The crux of this implementation is an atomic work which is CAS'd on many many
+// times in order to manage a few flags about who's blocking where and whether
+// it's locked or not.
+
+use std::rt::local::Local;
+use std::rt::task::{BlockedTask, Task};
+use std::rt::thread::Thread;
+use std::sync::atomics;
+use std::unstable::mutex;
+
+use q = sync::mpsc_intrusive;
+
+pub static LOCKED: uint = 1 << 0;
+pub static GREEN_BLOCKED: uint = 1 << 1;
+pub static NATIVE_BLOCKED: uint = 1 << 2;
+
+/// A mutual exclusion primitive useful for protecting shared data
+///
+/// This mutex is an implementation of a lock for all flavors of tasks which may
+/// be grabbing. A common problem with green threads is that they cannot grab
+/// locks (if they reschedule during the lock a contender could deadlock the
+/// system), but this mutex does *not* suffer this problem.
+///
+/// This mutex will properly block tasks waiting for the lock to become
+/// available. The mutex can also be statically initialized or created via a
+/// `new` constructor.
+///
+/// # Example
+///
+/// ```rust
+/// use sync::mutex::Mutex;
+///
+/// let mut m = Mutex::new();
+/// let guard = m.lock();
+/// // do some work
+/// drop(guard); // unlock the lock
+/// ```
+pub struct Mutex {
+    priv lock: StaticMutex,
+}
+
+#[deriving(Eq)]
+enum Flavor {
+    Unlocked,
+    TryLockAcquisition,
+    GreenAcquisition,
+    NativeAcquisition,
+}
+
+/// The static mutex type is provided to allow for static allocation of mutexes.
+///
+/// Note that this is a separate type because using a Mutex correctly means that
+/// it needs to have a destructor run. In Rust, statics are not allowed to have
+/// destructors. As a result, a `StaticMutex` has one extra method when compared
+/// to a `Mutex`, a `destroy` method. This method is unsafe to call, and
+/// documentation can be found directly on the method.
+///
+/// # Example
+///
+/// ```rust
+/// use sync::mutex::{StaticMutex, MUTEX_INIT};
+///
+/// static mut LOCK: StaticMutex = MUTEX_INIT;
+///
+/// unsafe {
+///     let _g = LOCK.lock();
+///     // do some productive work
+/// }
+/// // lock is unlocked here.
+/// ```
+pub struct StaticMutex {
+    /// Current set of flags on this mutex
+    priv state: atomics::AtomicUint,
+    /// Type of locking operation currently on this mutex
+    priv flavor: Flavor,
+    /// uint-cast of the green thread waiting for this mutex
+    priv green_blocker: uint,
+    /// uint-cast of the native thread waiting for this mutex
+    priv native_blocker: uint,
+    /// an OS mutex used by native threads
+    priv lock: mutex::Mutex,
+
+    /// A concurrent mpsc queue used by green threads, along with a count used
+    /// to figure out when to dequeue and enqueue.
+    priv q: q::Queue<uint>,
+    priv green_cnt: atomics::AtomicUint,
+}
+
+/// An RAII implementation of a "scoped lock" of a mutex. When this structure is
+/// dropped (falls out of scope), the lock will be unlocked.
+pub struct Guard<'a> {
+    priv lock: &'a mut StaticMutex,
+}
+
+/// Static initialization of a mutex. This constant can be used to initialize
+/// other mutex constants.
+pub static MUTEX_INIT: StaticMutex = StaticMutex {
+    lock: mutex::MUTEX_INIT,
+    state: atomics::INIT_ATOMIC_UINT,
+    flavor: Unlocked,
+    green_blocker: 0,
+    native_blocker: 0,
+    green_cnt: atomics::INIT_ATOMIC_UINT,
+    q: q::Queue {
+        head: atomics::INIT_ATOMIC_UINT,
+        tail: 0 as *mut q::Node<uint>,
+        stub: q::DummyNode {
+            next: atomics::INIT_ATOMIC_UINT,
+        }
+    }
+};
+
+impl StaticMutex {
+    /// Attempts to grab this lock, see `Mutex::try_lock`
+    pub fn try_lock<'a>(&'a mut self) -> Option<Guard<'a>> {
+        // Attempt to steal the mutex from an unlocked state.
+        //
+        // FIXME: this can mess up the fairness of the mutex, seems bad
+        match self.state.compare_and_swap(0, LOCKED, atomics::SeqCst) {
+            0 => {
+                assert!(self.flavor == Unlocked);
+                self.flavor = TryLockAcquisition;
+                Some(Guard::new(self))
+            }
+            _ => None
+        }
+    }
+
+    /// Acquires this lock, see `Mutex::lock`
+    pub fn lock<'a>(&'a mut self) -> Guard<'a> {
+        // First, attempt to steal the mutex from an unlocked state. The "fast
+        // path" needs to have as few atomic instructions as possible, and this
+        // one cmpxchg is already pretty expensive.
+        //
+        // FIXME: this can mess up the fairness of the mutex, seems bad
+        match self.state.compare_and_swap(0, LOCKED, atomics::SeqCst) {
+            0 => {
+                assert!(self.flavor == Unlocked);
+                self.flavor = TryLockAcquisition;
+                return Guard::new(self)
+            }
+            _ => {}
+        }
+
+        // After we've failed the fast path, then we delegate to the differnet
+        // locking protocols for green/native tasks. This will select two tasks
+        // to continue further (one native, one green).
+        let t: ~Task = Local::take();
+        let can_block = t.can_block();
+        let native_bit;
+        if can_block {
+            self.native_lock(t);
+            native_bit = NATIVE_BLOCKED;
+        } else {
+            self.green_lock(t);
+            native_bit = GREEN_BLOCKED;
+        }
+
+        // After we've arbitrated among task types, attempt to re-acquire the
+        // lock (avoids a deschedule). This is very important to do in order to
+        // allow threads coming out of the native_lock function to try their
+        // best to not hit a cvar in deschedule.
+        let mut old = match self.state.compare_and_swap(0, LOCKED,
+                                                        atomics::SeqCst) {
+            0 => {
+                self.flavor = if can_block {
+                    NativeAcquisition
+                } else {
+                    GreenAcquisition
+                };
+                return Guard::new(self)
+            }
+            old => old,
+        };
+
+        // Alright, everything else failed. We need to deschedule ourselves and
+        // flag ourselves as waiting. Note that this case should only happen
+        // regularly in native/green contention. Due to try_lock and the header
+        // of lock stealing the lock, it's also possible for native/native
+        // contention to hit this location, but as less common.
+        let t: ~Task = Local::take();
+        t.deschedule(1, |task| {
+            let task = unsafe { task.cast_to_uint() };
+            if can_block {
+                assert_eq!(self.native_blocker, 0);
+                self.native_blocker = task;
+            } else {
+                assert_eq!(self.green_blocker, 0);
+                self.green_blocker = task;
+            }
+
+            loop {
+                assert_eq!(old & native_bit, 0);
+                // If the old state was locked, then we need to flag ourselves
+                // as blocking in the state. If the old state was unlocked, then
+                // we attempt to acquire the mutex. Everything here is a CAS
+                // loop that'll eventually make progress.
+                if old & LOCKED != 0 {
+                    old = match self.state.compare_and_swap(old,
+                                                            old | native_bit,
+                                                            atomics::SeqCst) {
+                        n if n == old => return Ok(()),
+                        n => n
+                    };
+                } else {
+                    assert_eq!(old, 0);
+                    old = match self.state.compare_and_swap(old,
+                                                            old | LOCKED,
+                                                            atomics::SeqCst) {
+                        n if n == old => {
+                            assert_eq!(self.flavor, Unlocked);
+                            if can_block {
+                                self.native_blocker = 0;
+                                self.flavor = NativeAcquisition;
+                            } else {
+                                self.green_blocker = 0;
+                                self.flavor = GreenAcquisition;
+                            }
+                            return Err(unsafe {
+                                BlockedTask::cast_from_uint(task)
+                            })
+                        }
+                        n => n,
+                    };
+                }
+            }
+        });
+
+        Guard::new(self)
+    }
+
+    // Tasks which can block are super easy. These tasks just call the blocking
+    // `lock()` function on an OS mutex
+    fn native_lock(&mut self, t: ~Task) {
+        Local::put(t);
+        unsafe { self.lock.lock(); }
+    }
+
+    fn native_unlock(&mut self) {
+        unsafe { self.lock.unlock(); }
+    }
+
+    fn green_lock(&mut self, t: ~Task) {
+        // Green threads flag their presence with an atomic counter, and if they
+        // fail to be the first to the mutex, they enqueue themselves on a
+        // concurrent internal queue with a stack-allocated node.
+        //
+        // FIXME: There isn't a cancellation currently of an enqueue, forcing
+        //        the unlocker to spin for a bit.
+        if self.green_cnt.fetch_add(1, atomics::SeqCst) == 0 {
+            Local::put(t);
+            return
+        }
+
+        let mut node = q::Node::new(0);
+        t.deschedule(1, |task| {
+            unsafe {
+                node.data = task.cast_to_uint();
+                self.q.push(&mut node);
+            }
+            Ok(())
+        });
+    }
+
+    fn green_unlock(&mut self) {
+        // If we're the only green thread, then no need to check the queue,
+        // otherwise the fixme above forces us to spin for a bit.
+        if self.green_cnt.fetch_sub(1, atomics::SeqCst) == 1 { return }
+        let node;
+        loop {
+            match unsafe { self.q.pop() } {
+                Some(t) => { node = t; break; }
+                None => Thread::yield_now(),
+            }
+        }
+        let task = unsafe { BlockedTask::cast_from_uint((*node).data) };
+        task.wake().map(|t| t.reawaken());
+    }
+
+    fn unlock(&mut self) {
+        // Unlocking this mutex is a little tricky. We favor any task that is
+        // manually blocked (not in each of the separate locks) in order to help
+        // provide a little fairness (green threads will wake up the pending
+        // native thread and native threads will wake up the pending green
+        // thread).
+        //
+        // There's also the question of when we unlock the actual green/native
+        // locking halves as well. If we're waking up someone, then we can wait
+        // to unlock until we've acquired the task to wake up (we're guaranteed
+        // the mutex memory is still valid when there's contenders), but as soon
+        // as we don't find any contenders we must unlock the mutex, and *then*
+        // flag the mutex as unlocked.
+        //
+        // This flagging can fail, leading to another round of figuring out if a
+        // task needs to be woken, and in this case it's ok that the "mutex
+        // halves" are unlocked, we're just mainly dealing with the atomic state
+        // of the outer mutex.
+        let flavor = self.flavor;
+        self.flavor = Unlocked;
+
+        let mut state = self.state.load(atomics::SeqCst);
+        let mut unlocked = false;
+        let task;
+        loop {
+            assert!(state & LOCKED != 0);
+            if state & GREEN_BLOCKED != 0 {
+                self.unset(state, GREEN_BLOCKED);
+                task = unsafe {
+                    BlockedTask::cast_from_uint(self.green_blocker)
+                };
+                self.green_blocker = 0;
+                self.flavor = GreenAcquisition;
+                break;
+            } else if state & NATIVE_BLOCKED != 0 {
+                self.unset(state, NATIVE_BLOCKED);
+                task = unsafe {
+                    BlockedTask::cast_from_uint(self.native_blocker)
+                };
+                self.native_blocker = 0;
+                self.flavor = NativeAcquisition;
+                break;
+            } else {
+                assert_eq!(state, LOCKED);
+                if !unlocked {
+                    match flavor {
+                        GreenAcquisition => { self.green_unlock(); }
+                        NativeAcquisition => { self.native_unlock(); }
+                        TryLockAcquisition => {}
+                        Unlocked => unreachable!()
+                    }
+                    unlocked = true;
+                }
+                match self.state.compare_and_swap(LOCKED, 0, atomics::SeqCst) {
+                    LOCKED => return,
+                    n => { state = n; }
+                }
+            }
+        }
+        if !unlocked {
+            match flavor {
+                GreenAcquisition => { self.green_unlock(); }
+                NativeAcquisition => { self.native_unlock(); }
+                TryLockAcquisition => {}
+                Unlocked => unreachable!()
+            }
+        }
+
+        task.wake().map(|t| t.reawaken());
+    }
+
+    /// Loops around a CAS to unset the `bit` in `state`
+    fn unset(&mut self, mut state: uint, bit: uint) {
+        loop {
+            assert!(state & bit != 0);
+            let new = state ^ bit;
+            match self.state.compare_and_swap(state, new, atomics::SeqCst) {
+                n if n == state => break,
+                n => { state = n; }
+            }
+        }
+    }
+
+    /// Deallocates resources associated with this static mutex.
+    ///
+    /// This method is unsafe because it provides no guarantees that there are
+    /// no active users of this mutex, and safety is not guaranteed if there are
+    /// active users of this mutex.
+    ///
+    /// This method is required to ensure that there are no memory leaks on
+    /// *all* platforms. It may be the case that some platforms do not leak
+    /// memory if this method is not called, but this is not guaranteed to be
+    /// true on all platforms.
+    pub unsafe fn destroy(&mut self) {
+        self.lock.destroy()
+    }
+}
+
+impl Mutex {
+    /// Creates a new mutex in an unlocked state ready for use.
+    pub fn new() -> Mutex {
+        Mutex {
+            lock: StaticMutex {
+                state: atomics::AtomicUint::new(0),
+                flavor: Unlocked,
+                green_blocker: 0,
+                native_blocker: 0,
+                green_cnt: atomics::AtomicUint::new(0),
+                q: q::Queue::new(),
+                lock: unsafe { mutex::Mutex::new() },
+            }
+        }
+    }
+
+    /// Attempts to acquire this lock.
+    ///
+    /// If the lock could not be acquired at this time, then `None` is returned.
+    /// Otherwise, an RAII guard is returned. The lock will be unlocked when the
+    /// guard is dropped.
+    ///
+    /// This function does not block.
+    pub fn try_lock<'a>(&'a mut self) -> Option<Guard<'a>> {
+        self.lock.try_lock()
+    }
+
+    /// Acquires a mutex, blocking the current task until it is able to do so.
+    ///
+    /// This function will block the local task until it is availble to acquire
+    /// the mutex. Upon returning, the task is the only task with the mutex
+    /// held. An RAII guard is returned to allow scoped unlock of the lock. When
+    /// the guard goes out of scope, the mutex will be unlocked.
+    pub fn lock<'a>(&'a mut self) -> Guard<'a> { self.lock.lock() }
+}
+
+impl<'a> Guard<'a> {
+    fn new<'b>(lock: &'b mut StaticMutex) -> Guard<'b> {
+        if cfg!(debug) {
+            assert!(lock.flavor != Unlocked);
+            assert!(lock.state.load(atomics::SeqCst) & LOCKED != 0);
+        }
+        Guard { lock: lock }
+    }
+}
+
+#[unsafe_destructor]
+impl<'a> Drop for Guard<'a> {
+    #[inline]
+    fn drop(&mut self) {
+        self.lock.unlock();
+    }
+}
+
+impl Drop for Mutex {
+    fn drop(&mut self) {
+        // This is actually safe b/c we know that there is no further usage of
+        // this mutex (it's up to the user to arrange for a mutex to get
+        // dropped, that's not our job)
+        unsafe { self.lock.destroy() }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    extern mod native;
+    use super::{Mutex, StaticMutex, MUTEX_INIT};
+
+    #[test]
+    fn smoke() {
+        let mut m = Mutex::new();
+        drop(m.lock());
+        drop(m.lock());
+    }
+
+    #[test]
+    fn smoke_static() {
+        static mut m: StaticMutex = MUTEX_INIT;
+        unsafe {
+            drop(m.lock());
+            drop(m.lock());
+            m.destroy();
+        }
+    }
+
+    #[test]
+    fn lots_and_lots() {
+        static mut m: StaticMutex = MUTEX_INIT;
+        static mut CNT: uint = 0;
+        static M: uint = 1000;
+        static N: uint = 3;
+
+        fn inc() {
+            for _ in range(0, M) {
+                unsafe {
+                    let _g = m.lock();
+                    CNT += 1;
+                }
+            }
+        }
+
+        let (p, c) = SharedChan::new();
+        for _ in range(0, N) {
+            let c2 = c.clone();
+            native::task::spawn(proc() { inc(); c2.send(()); });
+            let c2 = c.clone();
+            spawn(proc() { inc(); c2.send(()); });
+        }
+
+        drop(c);
+        for _ in range(0, 2 * N) {
+            p.recv();
+        }
+        assert_eq!(unsafe {CNT}, M * N * 2);
+        unsafe {
+            m.destroy();
+        }
+    }
+
+    #[test]
+    fn trylock() {
+        let mut m = Mutex::new();
+        assert!(m.try_lock().is_some());
+    }
+}
diff --git a/src/libsync/sync/one.rs b/src/libsync/sync/one.rs
new file mode 100644
index 00000000000..93d818b704d
--- /dev/null
+++ b/src/libsync/sync/one.rs
@@ -0,0 +1,168 @@
+// Copyright 2014 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+//! A "once initialization" primitive
+//!
+//! This primitive is meant to be used to run one-time initialization. An
+//! example use case would be for initializing an FFI library.
+
+use std::int;
+use std::sync::atomics;
+use sync::mutex::{StaticMutex, MUTEX_INIT};
+
+/// A type which can be used to run a one-time global initialization. This type
+/// is *unsafe* to use because it is built on top of the `Mutex` in this module.
+/// It does not know whether the currently running task is in a green or native
+/// context, and a blocking mutex should *not* be used under normal
+/// circumstances on a green task.
+///
+/// Despite its unsafety, it is often useful to have a one-time initialization
+/// routine run for FFI bindings or related external functionality. This type
+/// can only be statically constructed with the `ONCE_INIT` value.
+///
+/// # Example
+///
+/// ```rust
+/// use sync::one::{Once, ONCE_INIT};
+///
+/// static mut START: Once = ONCE_INIT;
+/// unsafe {
+///     START.doit(|| {
+///         // run initialization here
+///     });
+/// }
+/// ```
+pub struct Once {
+    priv mutex: StaticMutex,
+    priv cnt: atomics::AtomicInt,
+    priv lock_cnt: atomics::AtomicInt,
+}
+
+/// Initialization value for static `Once` values.
+pub static ONCE_INIT: Once = Once {
+    mutex: MUTEX_INIT,
+    cnt: atomics::INIT_ATOMIC_INT,
+    lock_cnt: atomics::INIT_ATOMIC_INT,
+};
+
+impl Once {
+    /// Perform an initialization routine once and only once. The given closure
+    /// will be executed if this is the first time `doit` has been called, and
+    /// otherwise the routine will *not* be invoked.
+    ///
+    /// This method will block the calling *os thread* if another initialization
+    /// routine is currently running.
+    ///
+    /// When this function returns, it is guaranteed that some initialization
+    /// has run and completed (it may not be the closure specified).
+    pub fn doit(&mut self, f: ||) {
+        // Implementation-wise, this would seem like a fairly trivial primitive.
+        // The stickler part is where our mutexes currently require an
+        // allocation, and usage of a `Once` should't leak this allocation.
+        //
+        // This means that there must be a deterministic destroyer of the mutex
+        // contained within (because it's not needed after the initialization
+        // has run).
+        //
+        // The general scheme here is to gate all future threads once
+        // initialization has completed with a "very negative" count, and to
+        // allow through threads to lock the mutex if they see a non negative
+        // count. For all threads grabbing the mutex, exactly one of them should
+        // be responsible for unlocking the mutex, and this should only be done
+        // once everyone else is done with the mutex.
+        //
+        // This atomicity is achieved by swapping a very negative value into the
+        // shared count when the initialization routine has completed. This will
+        // read the number of threads which will at some point attempt to
+        // acquire the mutex. This count is then squirreled away in a separate
+        // variable, and the last person on the way out of the mutex is then
+        // responsible for destroying the mutex.
+        //
+        // It is crucial that the negative value is swapped in *after* the
+        // initialization routine has completed because otherwise new threads
+        // calling `doit` will return immediately before the initialization has
+        // completed.
+
+        let prev = self.cnt.fetch_add(1, atomics::SeqCst);
+        if prev < 0 {
+            // Make sure we never overflow, we'll never have int::MIN
+            // simultaneous calls to `doit` to make this value go back to 0
+            self.cnt.store(int::MIN, atomics::SeqCst);
+            return
+        }
+
+        // If the count is negative, then someone else finished the job,
+        // otherwise we run the job and record how many people will try to grab
+        // this lock
+        {
+            let _guard = self.mutex.lock();
+            if self.cnt.load(atomics::SeqCst) > 0 {
+                f();
+                let prev = self.cnt.swap(int::MIN, atomics::SeqCst);
+                self.lock_cnt.store(prev, atomics::SeqCst);
+            }
+        }
+
+        // Last one out cleans up after everyone else, no leaks!
+        if self.lock_cnt.fetch_add(-1, atomics::SeqCst) == 1 {
+            unsafe { self.mutex.destroy() }
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::{ONCE_INIT, Once};
+    use std::task;
+
+    #[test]
+    fn smoke_once() {
+        static mut o: Once = ONCE_INIT;
+        let mut a = 0;
+        unsafe { o.doit(|| a += 1); }
+        assert_eq!(a, 1);
+        unsafe { o.doit(|| a += 1); }
+        assert_eq!(a, 1);
+    }
+
+    #[test]
+    fn stampede_once() {
+        static mut o: Once = ONCE_INIT;
+        static mut run: bool = false;
+
+        let (p, c) = SharedChan::new();
+        for _ in range(0, 10) {
+            let c = c.clone();
+            spawn(proc() {
+                for _ in range(0, 4) { task::deschedule() }
+                unsafe {
+                    o.doit(|| {
+                        assert!(!run);
+                        run = true;
+                    });
+                    assert!(run);
+                }
+                c.send(());
+            });
+        }
+
+        unsafe {
+            o.doit(|| {
+                assert!(!run);
+                run = true;
+            });
+            assert!(run);
+        }
+
+        for _ in range(0, 10) {
+            p.recv();
+        }
+    }
+}