KillHandle, BlockedTask, exit code propagation, linked failure

src/libextra/arc.rs

@@ -50,9 +50,9 @@ use std::borrow;
 
 /// As sync::condvar, a mechanism for unlock-and-descheduling and signaling.
 pub struct Condvar<'self> {
-    is_mutex: bool,
-    failed: &'self mut bool,
-    cond: &'self sync::Condvar<'self>
+    priv is_mutex: bool,
+    priv failed: &'self mut bool,
+    priv cond: &'self sync::Condvar<'self>
 }
 
 impl<'self> Condvar<'self> {
@@ -108,7 +108,7 @@ impl<'self> Condvar<'self> {
  ****************************************************************************/
 
 /// An atomically reference counted wrapper for shared immutable state.
-pub struct ARC<T> { x: UnsafeAtomicRcBox<T> }
+pub struct ARC<T> { priv x: UnsafeAtomicRcBox<T> }
 
 /// Create an atomically reference counted wrapper.
 pub fn ARC<T:Freeze + Send>(data: T) -> ARC<T> {
@@ -123,6 +123,20 @@ impl<T:Freeze+Send> ARC<T> {
     pub fn get<'a>(&'a self) -> &'a T {
         unsafe { &*self.x.get_immut() }
     }
+
+    /**
+     * Retrieve the data back out of the ARC. This function blocks until the
+     * reference given to it is the last existing one, and then unwrap the data
+     * instead of destroying it.
+     *
+     * If multiple tasks call unwrap, all but the first will fail. Do not call
+     * unwrap from a task that holds another reference to the same ARC; it is
+     * guaranteed to deadlock.
+     */
+    pub fn unwrap(self) -> T {
+        let ARC { x: x } = self;
+        unsafe { x.unwrap() }
+    }
 }
 
 /**
@@ -143,9 +157,9 @@ impl<T:Freeze + Send> Clone for ARC<T> {
  ****************************************************************************/
 
 #[doc(hidden)]
-struct MutexARCInner<T> { lock: Mutex, failed: bool, data: T }
+struct MutexARCInner<T> { priv lock: Mutex, priv failed: bool, priv data: T }
 /// An ARC with mutable data protected by a blocking mutex.
-struct MutexARC<T> { x: UnsafeAtomicRcBox<MutexARCInner<T>> }
+struct MutexARC<T> { priv x: UnsafeAtomicRcBox<MutexARCInner<T>> }
 
 /// Create a mutex-protected ARC with the supplied data.
 pub fn MutexARC<T:Send>(user_data: T) -> MutexARC<T> {
@@ -225,6 +239,22 @@ impl<T:Send> MutexARC<T> {
                           cond: cond })
         }
     }
+
+    /**
+     * Retrieves the data, blocking until all other references are dropped,
+     * exactly as arc::unwrap.
+     *
+     * Will additionally fail if another task has failed while accessing the arc.
+     */
+    pub fn unwrap(self) -> T {
+        let MutexARC { x: x } = self;
+        let inner = unsafe { x.unwrap() };
+        let MutexARCInner { failed: failed, data: data, _ } = inner;
+        if failed {
+            fail!(~"Can't unwrap poisoned MutexARC - another task failed inside!");
+        }
+        data
+    }
 }
 
 // Common code for {mutex.access,rwlock.write}{,_cond}.
@@ -268,7 +298,7 @@ fn PoisonOnFail<'r>(failed: &'r mut bool) -> PoisonOnFail {
  ****************************************************************************/
 
 #[doc(hidden)]
-struct RWARCInner<T> { lock: RWlock, failed: bool, data: T }
+struct RWARCInner<T> { priv lock: RWlock, priv failed: bool, priv data: T }
 /**
  * A dual-mode ARC protected by a reader-writer lock. The data can be accessed
  * mutably or immutably, and immutably-accessing tasks may run concurrently.
@@ -278,7 +308,7 @@ struct RWARCInner<T> { lock: RWlock, failed: bool, data: T }
 #[mutable] // XXX remove after snap
 #[no_freeze]
 struct RWARC<T> {
-    x: UnsafeAtomicRcBox<RWARCInner<T>>,
+    priv x: UnsafeAtomicRcBox<RWARCInner<T>>,
 }
 
 /// Create a reader/writer ARC with the supplied data.
@@ -429,6 +459,23 @@ impl<T:Freeze + Send> RWARC<T> {
             }
         }
     }
+
+    /**
+     * Retrieves the data, blocking until all other references are dropped,
+     * exactly as arc::unwrap.
+     *
+     * Will additionally fail if another task has failed while accessing the arc
+     * in write mode.
+     */
+    pub fn unwrap(self) -> T {
+        let RWARC { x: x, _ } = self;
+        let inner = unsafe { x.unwrap() };
+        let RWARCInner { failed: failed, data: data, _ } = inner;
+        if failed {
+            fail!(~"Can't unwrap poisoned RWARC - another task failed inside!")
+        }
+        data
+    }
 }
 
 // Borrowck rightly complains about immutably aliasing the rwlock in order to
@@ -611,6 +658,23 @@ mod tests {
         }
     }
     #[test] #[should_fail] #[ignore(cfg(windows))]
+    pub fn test_mutex_arc_unwrap_poison() {
+        let arc = MutexARC(1);
+        let arc2 = ~(&arc).clone();
+        let (p, c) = comm::stream();
+        do task::spawn {
+            unsafe {
+                do arc2.access |one| {
+                    c.send(());
+                    assert!(*one == 2);
+                }
+            }
+        }
+        let _ = p.recv();
+        let one = arc.unwrap();
+        assert!(one == 1);
+    }
+    #[test] #[should_fail] #[ignore(cfg(windows))]
     fn test_rw_arc_poison_wr() {
         let arc = ~RWARC(1);
         let arc2 = (*arc).clone();

src/libstd/either.rs

@@ -18,6 +18,7 @@ use kinds::Copy;
 use iterator::IteratorUtil;
 use result::Result;
 use result;
+use str::StrSlice;
 use vec;
 use vec::{OwnedVector, ImmutableVector};
 
@@ -121,24 +122,37 @@ pub fn is_right<T, U>(eith: &Either<T, U>) -> bool {
     }
 }
 
-/// Retrieves the value in the left branch. Fails if the either is Right.
+/// Retrieves the value in the left branch.
+/// Fails with a specified reason if the either is Right.
 #[inline]
-pub fn unwrap_left<T,U>(eith: Either<T,U>) -> T {
+pub fn expect_left<T,U>(eith: Either<T,U>, reason: &str) -> T {
     match eith {
         Left(x) => x,
-        Right(_) => fail!("either::unwrap_left Right")
+        Right(_) => fail!(reason.to_owned())
     }
 }
 
-/// Retrieves the value in the right branch. Fails if the either is Left.
+/// Retrieves the value in the left branch. Fails if the either is Right.
 #[inline]
-pub fn unwrap_right<T,U>(eith: Either<T,U>) -> U {
+pub fn unwrap_left<T,U>(eith: Either<T,U>) -> T {
+    expect_left(eith, "either::unwrap_left Right")
+}
+
+/// Retrieves the value in the right branch.
+/// Fails with a specified reason if the either is Left.
+#[inline]
+pub fn expect_right<T,U>(eith: Either<T,U>, reason: &str) -> U {
     match eith {
         Right(x) => x,
-        Left(_) => fail!("either::unwrap_right Left")
+        Left(_) => fail!(reason.to_owned())
     }
 }
 
+/// Retrieves the value in the right branch. Fails if the either is Left.
+pub fn unwrap_right<T,U>(eith: Either<T,U>) -> U {
+    expect_right(eith, "either::unwrap_right Left")
+}
+
 impl<T, U> Either<T, U> {
     #[inline]
     pub fn either<V>(&self, f_left: &fn(&T) -> V, f_right: &fn(&U) -> V) -> V {
@@ -157,9 +171,15 @@ impl<T, U> Either<T, U> {
     #[inline]
     pub fn is_right(&self) -> bool { is_right(self) }
 
+    #[inline]
+    pub fn expect_left(self, reason: &str) -> T { expect_left(self, reason) }
+
     #[inline]
     pub fn unwrap_left(self) -> T { unwrap_left(self) }
 
+    #[inline]
+    pub fn expect_right(self, reason: &str) -> U { expect_right(self, reason) }
+
     #[inline]
     pub fn unwrap_right(self) -> U { unwrap_right(self) }
 }

src/libstd/option.rs

@@ -180,6 +180,13 @@ impl<T> Option<T> {
         match *self { Some(ref mut x) => Some(f(x)), None => None }
     }
 
+    /// Maps a `some` value from one type to another by a mutable reference,
+    /// or returns a default value.
+    #[inline]
+    pub fn map_mut_default<'a, U>(&'a mut self, def: U, f: &fn(&'a mut T) -> U) -> U {
+        match *self { Some(ref mut x) => f(x), None => def }
+    }
+
     /// As `map`, but consumes the option and gives `f` ownership to avoid
     /// copying.
     #[inline]
@@ -200,6 +207,26 @@ impl<T> Option<T> {
         match self { None => def, Some(v) => f(v) }
     }
 
+    /// Take the value out of the option, leaving a `None` in its place.
+    #[inline]
+    pub fn take(&mut self) -> Option<T> {
+        util::replace(self, None)
+    }
+
+    /// As `map_consume`, but swaps a None into the original option rather
+    /// than consuming it by-value.
+    #[inline]
+    pub fn take_map<U>(&mut self, blk: &fn(T) -> U) -> Option<U> {
+        self.take().map_consume(blk)
+    }
+
+    /// As `map_consume_default`, but swaps a None into the original option
+    /// rather than consuming it by-value.
+    #[inline]
+    pub fn take_map_default<U> (&mut self, def: U, blk: &fn(T) -> U) -> U {
+        self.take().map_consume_default(def, blk)
+    }
+
     /// Apply a function to the contained value or do nothing
     pub fn mutate(&mut self, f: &fn(T) -> T) {
         if self.is_some() {
@@ -295,7 +322,7 @@ impl<T> Option<T> {
     #[inline]
     pub fn swap_unwrap(&mut self) -> T {
         if self.is_none() { fail!("option::swap_unwrap none") }
-        util::replace(self, None).unwrap()
+        self.take().unwrap()
     }
 
     /**

src/libstd/rt/comm.rs

@@ -19,7 +19,7 @@ use option::*;
 use cast;
 use util;
 use ops::Drop;
-use rt::task::Task;
+use rt::kill::BlockedTask;
 use kinds::Send;
 use rt::sched::Scheduler;
 use rt::local::Local;
@@ -30,13 +30,13 @@ use comm::{GenericChan, GenericSmartChan, GenericPort, Peekable};
 use cell::Cell;
 use clone::Clone;
 
-/// A combined refcount / ~Task pointer.
+/// A combined refcount / BlockedTask-as-uint pointer.
 ///
 /// Can be equal to the following values:
 ///
 /// * 2 - both endpoints are alive
 /// * 1 - either the sender or the receiver is dead, determined by context
-/// * <ptr> - A pointer to a blocked Task that can be transmuted to ~Task
+/// * <ptr> - A pointer to a blocked Task (see BlockedTask::cast_{to,from}_uint)
 type State = uint;
 
 static STATE_BOTH: State = 2;
@@ -137,11 +137,13 @@ impl<T> ChanOne<T> {
                 }
                 task_as_state => {
                     // Port is blocked. Wake it up.
-                    let recvr: ~Task = cast::transmute(task_as_state);
-                    let mut sched = Local::take::<Scheduler>();
-                    rtdebug!("rendezvous send");
-                    sched.metrics.rendezvous_sends += 1;
-                    sched.schedule_task(recvr);
+                    let recvr = BlockedTask::cast_from_uint(task_as_state);
+                    do recvr.wake().map_consume |woken_task| {
+                        let mut sched = Local::take::<Scheduler>();
+                        rtdebug!("rendezvous send");
+                        sched.metrics.rendezvous_sends += 1;
+                        sched.schedule_task(woken_task);
+                    };
                 }
             }
         }
@@ -177,7 +179,7 @@ impl<T> PortOne<T> {
                 // an acquire barrier to prevent reordering of the subsequent read
                 // of the payload. Also issues a release barrier to prevent reordering
                 // of any previous writes to the task structure.
-                let task_as_state: State = cast::transmute(task);
+                let task_as_state = task.cast_to_uint();
                 let oldstate = (*packet).state.swap(task_as_state, SeqCst);
                 match oldstate {
                     STATE_BOTH => {
@@ -193,8 +195,8 @@ impl<T> PortOne<T> {
                         // NB: We have to drop back into the scheduler event loop here
                         // instead of switching immediately back or we could end up
                         // triggering infinite recursion on the scheduler's stack.
-                        let task: ~Task = cast::transmute(task_as_state);
-                        sched.enqueue_task(task);
+                        let recvr = BlockedTask::cast_from_uint(task_as_state);
+                        sched.enqueue_blocked_task(recvr);
                     }
                     _ => util::unreachable()
                 }
@@ -258,9 +260,11 @@ impl<T> Drop for ChanOneHack<T> {
                 task_as_state => {
                     // The port is blocked waiting for a message we will never send. Wake it.
                     assert!((*this.packet()).payload.is_none());
-                    let recvr: ~Task = cast::transmute(task_as_state);
-                    let sched = Local::take::<Scheduler>();
-                    sched.schedule_task(recvr);
+                    let recvr = BlockedTask::cast_from_uint(task_as_state);
+                    do recvr.wake().map_consume |woken_task| {
+                        let sched = Local::take::<Scheduler>();
+                        sched.schedule_task(woken_task);
+                    };
                 }
             }
         }
@@ -283,7 +287,10 @@ impl<T> Drop for PortOneHack<T> {
                     let _packet: ~Packet<T> = cast::transmute(this.void_packet);
                 }
                 _ => {
-                    util::unreachable()
+                    // This case occurs during unwinding, when the blocked
+                    // receiver was killed awake. The send-end will free it.
+                    // It sure would be nice if we could assert the task is
+                    // failing, but that would need a fragile local borrow.
                 }
             }
         }