rollup merge of #19274: alexcrichton/rewrite-sync

This commit is a reimplementation of `std::sync` to be based on the
system-provided primitives wherever possible. The previous implementation was
fundamentally built on top of channels, and as part of the runtime reform it has
become clear that this is not the level of abstraction that the standard level
should be providing. This rewrite aims to provide as thin of a shim as possible
on top of the system primitives in order to make them safe.

The overall interface of the `std::sync` module has in general not changed, but
there are a few important distinctions, highlighted below:

* The condition variable type, `Condvar`, has been separated out of a `Mutex`.
  A condition variable is now an entirely separate type. This separation
  benefits users who only use one mutex, and provides a clearer distinction of
  who's responsible for managing condition variables (the application).

* All of `Condvar`, `Mutex`, and `RWLock` are now directly built on top of
  system primitives rather than using a custom implementation. The `Once`,
  `Barrier`, and `Semaphore` types are still built upon these abstractions of
  the system primitives.

* The `Condvar`, `Mutex`, and `RWLock` types all have a new static type and
  constant initializer corresponding to them. These are provided primarily for C
  FFI interoperation, but are often useful to otherwise simply have a global
  lock. The types, however, will leak memory unless `destroy()` is called on
  them, which is clearly documented.

* The fundamental architecture of this design is to provide two separate layers.
  The first layer is that exposed by `sys_common` which is a cross-platform
  bare-metal abstraction of the system synchronization primitives. No attempt is
  made at making this layer safe, and it is quite unsafe to use! It is currently
  not exported as part of the API of the standard library, but the stabilization
  of the `sys` module will ensure that these will be exposed in time. The
  purpose of this layer is to provide the core cross-platform abstractions if
  necessary to implementors.

  The second layer is the layer provided by `std::sync` which is intended to be
  the thinnest possible layer on top of `sys_common` which is entirely safe to
  use. There are a few concerns which need to be addressed when making these
  system primitives safe:

    * Once used, the OS primitives can never be **moved**. This means that they
      essentially need to have a stable address. The static primitives use
      `&'static self` to enforce this, and the non-static primitives all use a
      `Box` to provide this guarantee.

    * Poisoning is leveraged to ensure that invalid data is not accessible from
      other tasks after one has panicked.

  In addition to these overall blanket safety limitations, each primitive has a
  few restrictions of its own:

    * Mutexes and rwlocks can only be unlocked from the same thread that they
      were locked by. This is achieved through RAII lock guards which cannot be
      sent across threads.

    * Mutexes and rwlocks can only be unlocked if they were previously locked.
      This is achieved by not exposing an unlocking method.

    * A condition variable can only be waited on with a locked mutex. This is
      achieved by requiring a `MutexGuard` in the `wait()` method.

    * A condition variable cannot be used concurrently with more than one mutex.
      This is guaranteed by dynamically binding a condition variable to
      precisely one mutex for its entire lifecycle. This restriction may be able
      to be relaxed in the future (a mutex is unbound when no threads are
      waiting on the condvar), but for now it is sufficient to guarantee safety.

* Condvars support timeouts for their blocking operations. The
  implementation for these operations is provided by the system.

Due to the modification of the `Condvar` API, removal of the `std::sync::mutex`
API, and reimplementation, this is a breaking change. Most code should be fairly
easy to port using the examples in the documentation of these primitives.

[breaking-change]

Closes #17094
Closes #18003

Author: emberian

src/etc/licenseck.py

@@ -38,9 +38,8 @@
     "rt/isaac/randport.cpp", # public domain
     "rt/isaac/rand.h", # public domain
     "rt/isaac/standard.h", # public domain
-    "libstd/sync/mpsc_queue.rs", # BSD
-    "libstd/sync/spsc_queue.rs", # BSD
-    "libstd/sync/mpmc_bounded_queue.rs", # BSD
+    "libstd/comm/mpsc_queue.rs", # BSD
+    "libstd/comm/spsc_queue.rs", # BSD
     "test/bench/shootout-binarytrees.rs", # BSD
     "test/bench/shootout-chameneos-redux.rs", # BSD
     "test/bench/shootout-fannkuch-redux.rs", # BSD

src/libstd/comm/mod.rs

@@ -354,6 +354,8 @@ mod select;
 mod shared;
 mod stream;
 mod sync;
+mod mpsc_queue;
+mod spsc_queue;
 
 /// The receiving-half of Rust's channel type. This half can only be owned by
 /// one task
@@ -628,24 +630,26 @@ impl<T: Send> Sender<T> {
 #[unstable]
 impl<T: Send> Clone for Sender<T> {
     fn clone(&self) -> Sender<T> {
-        let (packet, sleeper) = match *unsafe { self.inner() } {
+        let (packet, sleeper, guard) = match *unsafe { self.inner() } {
             Oneshot(ref p) => {
                 let a = Arc::new(UnsafeCell::new(shared::Packet::new()));
                 unsafe {
-                    (*a.get()).postinit_lock();
+                    let guard = (*a.get()).postinit_lock();
                     match (*p.get()).upgrade(Receiver::new(Shared(a.clone()))) {
-                        oneshot::UpSuccess | oneshot::UpDisconnected => (a, None),
-                        oneshot::UpWoke(task) => (a, Some(task))
+                        oneshot::UpSuccess |
+                        oneshot::UpDisconnected => (a, None, guard),
+                        oneshot::UpWoke(task) => (a, Some(task), guard)
                     }
                 }
             }
             Stream(ref p) => {
                 let a = Arc::new(UnsafeCell::new(shared::Packet::new()));
                 unsafe {
-                    (*a.get()).postinit_lock();
+                    let guard = (*a.get()).postinit_lock();
                     match (*p.get()).upgrade(Receiver::new(Shared(a.clone()))) {
-                        stream::UpSuccess | stream::UpDisconnected => (a, None),
-                        stream::UpWoke(task) => (a, Some(task)),
+                        stream::UpSuccess |
+                        stream::UpDisconnected => (a, None, guard),
+                        stream::UpWoke(task) => (a, Some(task), guard),
                     }
                 }
             }
@@ -657,7 +661,7 @@ impl<T: Send> Clone for Sender<T> {
         };
 
         unsafe {
-            (*packet.get()).inherit_blocker(sleeper);
+            (*packet.get()).inherit_blocker(sleeper, guard);
 
             let tmp = Sender::new(Shared(packet.clone()));
             mem::swap(self.inner_mut(), tmp.inner_mut());

src/libstd/sync/mpsc_queue.rs renamed to src/libstd/comm/mpsc_queue.rs

@@ -132,15 +132,6 @@ impl<T: Send> Queue<T> {
             if self.head.load(Acquire) == tail {Empty} else {Inconsistent}
         }
     }
-
-    /// Attempts to pop data from this queue, but doesn't attempt too hard. This
-    /// will canonicalize inconsistent states to a `None` value.
-    pub fn casual_pop(&self) -> Option<T> {
-        match self.pop() {
-            Data(t) => Some(t),
-            Empty | Inconsistent => None,
-        }
-    }
 }
 
 #[unsafe_destructor]

src/libstd/comm/shared.rs

@@ -26,12 +26,11 @@ use alloc::boxed::Box;
 use core::cmp;
 use core::int;
 use rustrt::local::Local;
-use rustrt::mutex::NativeMutex;
 use rustrt::task::{Task, BlockedTask};
 use rustrt::thread::Thread;
 
-use sync::atomic;
-use sync::mpsc_queue as mpsc;
+use sync::{atomic, Mutex, MutexGuard};
+use comm::mpsc_queue as mpsc;
 
 const DISCONNECTED: int = int::MIN;
 const FUDGE: int = 1024;
@@ -56,7 +55,7 @@ pub struct Packet<T> {
 
     // this lock protects various portions of this implementation during
     // select()
-    select_lock: NativeMutex,
+    select_lock: Mutex<()>,
 }
 
 pub enum Failure {
@@ -76,7 +75,7 @@ impl<T: Send> Packet<T> {
             channels: atomic::AtomicInt::new(2),
             port_dropped: atomic::AtomicBool::new(false),
             sender_drain: atomic::AtomicInt::new(0),
-            select_lock: unsafe { NativeMutex::new() },
+            select_lock: Mutex::new(()),
         };
         return p;
     }
@@ -86,16 +85,18 @@ impl<T: Send> Packet<T> {
     // In other case mutex data will be duplicated while cloning
     // and that could cause problems on platforms where it is
     // represented by opaque data structure
-    pub fn postinit_lock(&mut self) {
-        unsafe { self.select_lock.lock_noguard() }
+    pub fn postinit_lock(&self) -> MutexGuard<()> {
+        self.select_lock.lock()
     }
 
     // This function is used at the creation of a shared packet to inherit a
     // previously blocked task. This is done to prevent spurious wakeups of
     // tasks in select().
     //
     // This can only be called at channel-creation time
-    pub fn inherit_blocker(&mut self, task: Option<BlockedTask>) {
+    pub fn inherit_blocker(&mut self,
+                           task: Option<BlockedTask>,
+                           guard: MutexGuard<()>) {
         match task {
             Some(task) => {
                 assert_eq!(self.cnt.load(atomic::SeqCst), 0);
@@ -135,7 +136,7 @@ impl<T: Send> Packet<T> {
         // interfere with this method. After we unlock this lock, we're
         // signifying that we're done modifying self.cnt and self.to_wake and
         // the port is ready for the world to continue using it.
-        unsafe { self.select_lock.unlock_noguard() }
+        drop(guard);
     }
 
     pub fn send(&mut self, t: T) -> Result<(), T> {
@@ -441,7 +442,7 @@ impl<T: Send> Packet<T> {
         // done with. Without this bounce, we can race with inherit_blocker
         // about looking at and dealing with to_wake. Once we have acquired the
         // lock, we are guaranteed that inherit_blocker is done.
-        unsafe {
+        {
             let _guard = self.select_lock.lock();
         }
 

src/libstd/sync/spsc_queue.rs renamed to src/libstd/comm/spsc_queue.rs

@@ -40,7 +40,6 @@ use core::prelude::*;
 use alloc::boxed::Box;
 use core::mem;
 use core::cell::UnsafeCell;
-use alloc::arc::Arc;
 
 use sync::atomic::{AtomicPtr, Relaxed, AtomicUint, Acquire, Release};
 
@@ -74,39 +73,6 @@ pub struct Queue<T> {
     cache_subtractions: AtomicUint,
 }
 
-/// A safe abstraction for the consumer in a single-producer single-consumer
-/// queue.
-pub struct Consumer<T> {
-    inner: Arc<Queue<T>>
-}
-
-impl<T: Send> Consumer<T> {
-    /// Attempts to pop the value from the head of the queue, returning `None`
-    /// if the queue is empty.
-    pub fn pop(&mut self) -> Option<T> {
-        self.inner.pop()
-    }
-
-    /// Attempts to peek at the head of the queue, returning `None` if the queue
-    /// is empty.
-    pub fn peek<'a>(&'a mut self) -> Option<&'a mut T> {
-        self.inner.peek()
-    }
-}
-
-/// A safe abstraction for the producer in a single-producer single-consumer
-/// queue.
-pub struct Producer<T> {
-    inner: Arc<Queue<T>>
-}
-
-impl<T: Send> Producer<T> {
-    /// Pushes a new value onto the queue.
-    pub fn push(&mut self, t: T) {
-        self.inner.push(t)
-    }
-}
-
 impl<T: Send> Node<T> {
     fn new() -> *mut Node<T> {
         unsafe {
@@ -118,30 +84,6 @@ impl<T: Send> Node<T> {
     }
 }
 
-/// Creates a new queue with a consumer-producer pair.
-///
-/// The producer returned is connected to the consumer to push all data to
-/// the consumer.
-///
-/// # Arguments
-///
-///   * `bound` - This queue implementation is implemented with a linked
-///               list, and this means that a push is always a malloc. In
-///               order to amortize this cost, an internal cache of nodes is
-///               maintained to prevent a malloc from always being
-///               necessary. This bound is the limit on the size of the
-///               cache (if desired). If the value is 0, then the cache has
-///               no bound. Otherwise, the cache will never grow larger than
-///               `bound` (although the queue itself could be much larger.
-pub fn queue<T: Send>(bound: uint) -> (Consumer<T>, Producer<T>) {
-    let q = unsafe { Queue::new(bound) };
-    let arc = Arc::new(q);
-    let consumer = Consumer { inner: arc.clone() };
-    let producer = Producer { inner: arc };
-
-    (consumer, producer)
-}
-
 impl<T: Send> Queue<T> {
     /// Creates a new queue.
     ///
@@ -296,78 +238,88 @@ impl<T: Send> Drop for Queue<T> {
 mod test {
     use prelude::*;
 
-    use super::{queue};
+    use sync::Arc;
+    use super::Queue;
 
     #[test]
     fn smoke() {
-        let (mut consumer, mut producer) = queue(0);
-        producer.push(1i);
-        producer.push(2);
-        assert_eq!(consumer.pop(), Some(1i));
-        assert_eq!(consumer.pop(), Some(2));
-        assert_eq!(consumer.pop(), None);
-        producer.push(3);
-        producer.push(4);
-        assert_eq!(consumer.pop(), Some(3));
-        assert_eq!(consumer.pop(), Some(4));
-        assert_eq!(consumer.pop(), None);
+        unsafe {
+            let queue = Queue::new(0);
+            queue.push(1i);
+            queue.push(2);
+            assert_eq!(queue.pop(), Some(1i));
+            assert_eq!(queue.pop(), Some(2));
+            assert_eq!(queue.pop(), None);
+            queue.push(3);
+            queue.push(4);
+            assert_eq!(queue.pop(), Some(3));
+            assert_eq!(queue.pop(), Some(4));
+            assert_eq!(queue.pop(), None);
+        }
     }
 
     #[test]
     fn peek() {
-        let (mut consumer, mut producer) = queue(0);
-        producer.push(vec![1i]);
+        unsafe {
+            let queue = Queue::new(0);
+            queue.push(vec![1i]);
+
+            // Ensure the borrowchecker works
+            match queue.peek() {
+                Some(vec) => match vec.as_slice() {
+                    // Note that `pop` is not allowed here due to borrow
+                    [1] => {}
+                    _ => return
+                },
+                None => unreachable!()
+            }
 
-        // Ensure the borrowchecker works
-        match consumer.peek() {
-            Some(vec) => match vec.as_slice() {
-                // Note that `pop` is not allowed here due to borrow
-                [1] => {}
-                _ => return
-            },
-            None => unreachable!()
+            queue.pop();
         }
-
-        consumer.pop();
     }
 
     #[test]
     fn drop_full() {
-        let (_, mut producer) = queue(0);
-        producer.push(box 1i);
-        producer.push(box 2i);
+        unsafe {
+            let q = Queue::new(0);
+            q.push(box 1i);
+            q.push(box 2i);
+        }
     }
 
     #[test]
     fn smoke_bound() {
-        let (mut consumer, mut producer) = queue(1);
-        producer.push(1i);
-        producer.push(2);
-        assert_eq!(consumer.pop(), Some(1));
-        assert_eq!(consumer.pop(), Some(2));
-        assert_eq!(consumer.pop(), None);
-        producer.push(3);
-        producer.push(4);
-        assert_eq!(consumer.pop(), Some(3));
-        assert_eq!(consumer.pop(), Some(4));
-        assert_eq!(consumer.pop(), None);
+        unsafe {
+            let q = Queue::new(0);
+            q.push(1i);
+            q.push(2);
+            assert_eq!(q.pop(), Some(1));
+            assert_eq!(q.pop(), Some(2));
+            assert_eq!(q.pop(), None);
+            q.push(3);
+            q.push(4);
+            assert_eq!(q.pop(), Some(3));
+            assert_eq!(q.pop(), Some(4));
+            assert_eq!(q.pop(), None);
+        }
     }
 
     #[test]
     fn stress() {
-        stress_bound(0);
-        stress_bound(1);
+        unsafe {
+            stress_bound(0);
+            stress_bound(1);
+        }
 
-        fn stress_bound(bound: uint) {
-            let (consumer, mut producer) = queue(bound);
+        unsafe fn stress_bound(bound: uint) {
+            let q = Arc::new(Queue::new(bound));
 
             let (tx, rx) = channel();
+            let q2 = q.clone();
             spawn(proc() {
-                // Move the consumer to a local mutable slot
-                let mut consumer = consumer;
                 for _ in range(0u, 100000) {
                     loop {
-                        match consumer.pop() {
+                        match q2.pop() {
                             Some(1i) => break,
                             Some(_) => panic!(),
                             None => {}
@@ -377,7 +329,7 @@ mod test {
                 tx.send(());
             });
             for _ in range(0i, 100000) {
-                producer.push(1);
+                q.push(1);
             }
             rx.recv();
         }

src/libstd/comm/stream.rs

@@ -32,7 +32,7 @@ use rustrt::task::{Task, BlockedTask};
 use rustrt::thread::Thread;
 
 use sync::atomic;
-use sync::spsc_queue as spsc;
+use comm::spsc_queue as spsc;
 use comm::Receiver;
 
 const DISCONNECTED: int = int::MIN;

src/libstd/dynamic_lib.rs

@@ -225,8 +225,8 @@ pub mod dl {
     }
 
     pub fn check_for_errors_in<T>(f: || -> T) -> Result<T, String> {
-        use rustrt::mutex::{StaticNativeMutex, NATIVE_MUTEX_INIT};
-        static LOCK: StaticNativeMutex = NATIVE_MUTEX_INIT;
+        use sync::{StaticMutex, MUTEX_INIT};
+        static LOCK: StaticMutex = MUTEX_INIT;
         unsafe {
             // dlerror isn't thread safe, so we need to lock around this entire
             // sequence