rollup merge of rust-lang#19661: alexcrichton/mutex-result

All of the current std::sync primitives have poisoning enable which means that
when a task fails inside of a write-access lock then all future attempts to
acquire the lock will fail. This strategy ensures that stale data whose
invariants are possibly not upheld are never viewed by other tasks to help
propagate unexpected panics (bugs in a program) among tasks.

Currently there is no way to test whether a mutex or rwlock is poisoned. One
method would be to duplicate all the methods with a sister foo_catch function,
for example. This pattern is, however, against our [error guidelines][errors].
As a result, this commit exposes the fact that a task has failed internally
through the return value of a `Result`.

[errors]: https://github.com/rust-lang/rfcs/blob/master/text/0236-error-conventions.md#do-not-provide-both-result-and-fail-variants

All methods now return a `LockResult<T>` or a `TryLockResult<T>` which
communicates whether the lock was poisoned or not. In a `LockResult`, both the
`Ok` and `Err` variants contains the `MutexGuard<T>` that is being returned in
order to allow access to the data if poisoning is not desired. This also means
that the lock is *always* held upon returning from `.lock()`.

A new type, `PoisonError`, was added with one method `into_guard` which can
consume the assertion that a lock is poisoned to gain access to the underlying
data.

This is a breaking change because the signatures of these methods have changed,
often incompatible ways. One major difference is that the `wait` methods on a
condition variable now consume the guard and return it in as a `LockResult` to
indicate whether the lock was poisoned while waiting. Most code can be updated
by calling `.unwrap()` on the return value of `.lock()`.

[breaking-change]

Author: alexcrichton

src/doc/intro.md

@@ -483,7 +483,7 @@ fn main() {
     for i in range(0u, 3u) {
         let number = numbers.clone();
         Thread::spawn(move || {
-            let mut array = number.lock();
+            let mut array = number.lock().unwrap();
 
             (*array)[i] += 1;
 

src/liballoc/arc.rs

@@ -58,7 +58,7 @@
 //!     let five = five.clone();
 //!
 //!     Thread::spawn(move || {
-//!         let mut number = five.lock();
+//!         let mut number = five.lock().unwrap();
 //!
 //!         *number += 1;
 //!
@@ -722,7 +722,7 @@ mod tests {
 
         let a = Arc::new(Cycle { x: Mutex::new(None) });
         let b = a.clone().downgrade();
-        *a.x.lock() = Some(b);
+        *a.x.lock().unwrap() = Some(b);
 
         // hopefully we don't double-free (or leak)...
     }

src/librustc_trans/back/write.rs

@@ -98,7 +98,7 @@ impl SharedEmitter {
     }
 
     fn dump(&mut self, handler: &Handler) {
-        let mut buffer = self.buffer.lock();
+        let mut buffer = self.buffer.lock().unwrap();
         for diag in buffer.iter() {
             match diag.code {
                 Some(ref code) => {
@@ -123,7 +123,7 @@ impl Emitter for SharedEmitter {
             msg: &str, code: Option<&str>, lvl: Level) {
         assert!(cmsp.is_none(), "SharedEmitter doesn't support spans");
 
-        self.buffer.lock().push(Diagnostic {
+        self.buffer.lock().unwrap().push(Diagnostic {
             msg: msg.to_string(),
             code: code.map(|s| s.to_string()),
             lvl: lvl,
@@ -915,7 +915,7 @@ fn run_work_multithreaded(sess: &Session,
 
             loop {
                 // Avoid holding the lock for the entire duration of the match.
-                let maybe_work = work_items_arc.lock().pop();
+                let maybe_work = work_items_arc.lock().unwrap().pop();
                 match maybe_work {
                     Some(work) => {
                         execute_work_item(&cgcx, work);

src/libstd/comm/shared.rs

@@ -86,7 +86,7 @@ impl<T: Send> Packet<T> {
     // and that could cause problems on platforms where it is
     // represented by opaque data structure
     pub fn postinit_lock(&self) -> MutexGuard<()> {
-        self.select_lock.lock()
+        self.select_lock.lock().unwrap()
     }
 
     // This function is used at the creation of a shared packet to inherit a
@@ -435,7 +435,7 @@ impl<T: Send> Packet<T> {
         // about looking at and dealing with to_wake. Once we have acquired the
         // lock, we are guaranteed that inherit_blocker is done.
         {
-            let _guard = self.select_lock.lock();
+            let _guard = self.select_lock.lock().unwrap();
         }
 
         // Like the stream implementation, we want to make sure that the count

src/libstd/comm/sync.rs

@@ -121,9 +121,9 @@ fn wait<'a, 'b, T: Send>(lock: &'a Mutex<State<T>>,
         NoneBlocked => {}
         _ => unreachable!(),
     }
-    drop(guard);        // unlock
-    wait_token.wait();  // block
-    lock.lock()         // relock
+    drop(guard);         // unlock
+    wait_token.wait();   // block
+    lock.lock().unwrap() // relock
 }
 
 /// Wakes up a thread, dropping the lock at the correct time
@@ -161,7 +161,7 @@ impl<T: Send> Packet<T> {
     fn acquire_send_slot(&self) -> MutexGuard<State<T>> {
         let mut node = Node { token: None, next: 0 as *mut Node };
         loop {
-            let mut guard = self.lock.lock();
+            let mut guard = self.lock.lock().unwrap();
             // are we ready to go?
             if guard.disconnected || guard.buf.size() < guard.buf.cap() {
                 return guard;
@@ -202,7 +202,7 @@ impl<T: Send> Packet<T> {
     }
 
     pub fn try_send(&self, t: T) -> Result<(), super::TrySendError<T>> {
-        let mut guard = self.lock.lock();
+        let mut guard = self.lock.lock().unwrap();
         if guard.disconnected {
             Err(super::RecvDisconnected(t))
         } else if guard.buf.size() == guard.buf.cap() {
@@ -239,7 +239,7 @@ impl<T: Send> Packet<T> {
     // When reading this, remember that there can only ever be one receiver at
     // time.
     pub fn recv(&self) -> Result<T, ()> {
-        let mut guard = self.lock.lock();
+        let mut guard = self.lock.lock().unwrap();
 
         // Wait for the buffer to have something in it. No need for a while loop
         // because we're the only receiver.
@@ -258,7 +258,7 @@ impl<T: Send> Packet<T> {
     }
 
     pub fn try_recv(&self) -> Result<T, Failure> {
-        let mut guard = self.lock.lock();
+        let mut guard = self.lock.lock().unwrap();
 
         // Easy cases first
         if guard.disconnected { return Err(Disconnected) }
@@ -315,7 +315,7 @@ impl<T: Send> Packet<T> {
         }
 
         // Not much to do other than wake up a receiver if one's there
-        let mut guard = self.lock.lock();
+        let mut guard = self.lock.lock().unwrap();
         if guard.disconnected { return }
         guard.disconnected = true;
         match mem::replace(&mut guard.blocker, NoneBlocked) {
@@ -326,7 +326,7 @@ impl<T: Send> Packet<T> {
     }
 
     pub fn drop_port(&self) {
-        let mut guard = self.lock.lock();
+        let mut guard = self.lock.lock().unwrap();
 
         if guard.disconnected { return }
         guard.disconnected = true;
@@ -372,14 +372,14 @@ impl<T: Send> Packet<T> {
     // If Ok, the value is whether this port has data, if Err, then the upgraded
     // port needs to be checked instead of this one.
     pub fn can_recv(&self) -> bool {
-        let guard = self.lock.lock();
+        let guard = self.lock.lock().unwrap();
         guard.disconnected || guard.buf.size() > 0
     }
 
     // Attempts to start selection on this port. This can either succeed or fail
     // because there is data waiting.
     pub fn start_selection(&self, token: SignalToken) -> StartResult {
-        let mut guard = self.lock.lock();
+        let mut guard = self.lock.lock().unwrap();
         if guard.disconnected || guard.buf.size() > 0 {
             Abort
         } else {
@@ -397,7 +397,7 @@ impl<T: Send> Packet<T> {
     //
     // The return value indicates whether there's data on this port.
     pub fn abort_selection(&self) -> bool {
-        let mut guard = self.lock.lock();
+        let mut guard = self.lock.lock().unwrap();
         match mem::replace(&mut guard.blocker, NoneBlocked) {
             NoneBlocked => true,
             BlockedSender(token) => {
@@ -413,7 +413,7 @@ impl<T: Send> Packet<T> {
 impl<T: Send> Drop for Packet<T> {
     fn drop(&mut self) {
         assert_eq!(self.channels.load(atomic::SeqCst), 0);
-        let mut guard = self.lock.lock();
+        let mut guard = self.lock.lock().unwrap();
         assert!(guard.queue.dequeue().is_none());
         assert!(guard.canceled.is_none());
     }

src/libstd/io/stdio.rs

@@ -146,7 +146,7 @@ impl StdinReader {
     /// ```
     pub fn lock<'a>(&'a mut self) -> StdinReaderGuard<'a> {
         StdinReaderGuard {
-            inner: self.inner.lock()
+            inner: self.inner.lock().unwrap()
         }
     }
 
@@ -155,53 +155,53 @@ impl StdinReader {
     /// The read is performed atomically - concurrent read calls in other
     /// threads will not interleave with this one.
     pub fn read_line(&mut self) -> IoResult<String> {
-        self.inner.lock().0.read_line()
+        self.inner.lock().unwrap().0.read_line()
     }
 
     /// Like `Buffer::read_until`.
     ///
     /// The read is performed atomically - concurrent read calls in other
     /// threads will not interleave with this one.
     pub fn read_until(&mut self, byte: u8) -> IoResult<Vec<u8>> {
-        self.inner.lock().0.read_until(byte)
+        self.inner.lock().unwrap().0.read_until(byte)
     }
 
     /// Like `Buffer::read_char`.
     ///
     /// The read is performed atomically - concurrent read calls in other
     /// threads will not interleave with this one.
     pub fn read_char(&mut self) -> IoResult<char> {
-        self.inner.lock().0.read_char()
+        self.inner.lock().unwrap().0.read_char()
     }
 }
 
 impl Reader for StdinReader {
     fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
-        self.inner.lock().0.read(buf)
+        self.inner.lock().unwrap().0.read(buf)
     }
 
     // We have to manually delegate all of these because the default impls call
     // read more than once and we don't want those calls to interleave (or
     // incur the costs of repeated locking).
 
     fn read_at_least(&mut self, min: uint, buf: &mut [u8]) -> IoResult<uint> {
-        self.inner.lock().0.read_at_least(min, buf)
+        self.inner.lock().unwrap().0.read_at_least(min, buf)
     }
 
     fn push_at_least(&mut self, min: uint, len: uint, buf: &mut Vec<u8>) -> IoResult<uint> {
-        self.inner.lock().0.push_at_least(min, len, buf)
+        self.inner.lock().unwrap().0.push_at_least(min, len, buf)
     }
 
     fn read_to_end(&mut self) -> IoResult<Vec<u8>> {
-        self.inner.lock().0.read_to_end()
+        self.inner.lock().unwrap().0.read_to_end()
     }
 
     fn read_le_uint_n(&mut self, nbytes: uint) -> IoResult<u64> {
-        self.inner.lock().0.read_le_uint_n(nbytes)
+        self.inner.lock().unwrap().0.read_le_uint_n(nbytes)
     }
 
     fn read_be_uint_n(&mut self, nbytes: uint) -> IoResult<u64> {
-        self.inner.lock().0.read_be_uint_n(nbytes)
+        self.inner.lock().unwrap().0.read_be_uint_n(nbytes)
     }
 }
 

src/libstd/sync/barrier.rs

@@ -69,15 +69,15 @@ impl Barrier {
     /// Barriers are re-usable after all threads have rendezvoused once, and can
     /// be used continuously.
     pub fn wait(&self) {
-        let mut lock = self.lock.lock();
+        let mut lock = self.lock.lock().unwrap();
         let local_gen = lock.generation_id;
         lock.count += 1;
         if lock.count < self.num_threads {
             // We need a while loop to guard against spurious wakeups.
             // http://en.wikipedia.org/wiki/Spurious_wakeup
             while local_gen == lock.generation_id &&
                   lock.count < self.num_threads {
-                self.cvar.wait(&lock);
+                lock = self.cvar.wait(lock).unwrap();
             }
         } else {
             lock.count = 0;