Add kill::Death for task death services and use it in Task.

5 files changed, 79 insertions, 22 deletions

diff --git a/src/libstd/rt/kill.rs b/src/libstd/rt/kill.rs
index afd2d3b5a1a..a18c8cac215 100644
--- a/src/libstd/rt/kill.rs
+++ b/src/libstd/rt/kill.rs
@@ -37,6 +37,18 @@ struct KillHandleInner {
 #[deriving(Clone)]
 pub struct KillHandle(UnsafeAtomicRcBox<KillHandleInner>);
 
+/// Per-task state related to task death, killing, failure, etc.
+pub struct Death {
+    // Shared among this task, its watched children, and any linked tasks who
+    // might kill it. This is optional so we can take it by-value at exit time.
+    kill_handle:     Option<KillHandle>,
+    // Handle to a watching parent, if we have one, for exit code propagation.
+    watching_parent: Option<KillHandle>,
+    // Action to be done with the exit code. If set, also makes the task wait
+    // until all its watched children exit before collecting the status.
+    on_exit:         Option<~fn(bool)>,
+}
+
 impl KillHandle {
     pub fn new() -> KillHandle {
         KillHandle(UnsafeAtomicRcBox::new(KillHandleInner {
@@ -126,3 +138,58 @@ impl KillHandle {
     }
 }
 
+impl Death {
+    pub fn new() -> Death {
+        Death {
+            kill_handle:     Some(KillHandle::new()),
+            watching_parent: None,
+            on_exit:         None,
+        }
+    }
+
+    pub fn new_child(&self) -> Death {
+        // FIXME(#7327)
+        Death {
+            kill_handle:     Some(KillHandle::new()),
+            watching_parent: self.kill_handle.clone(),
+            on_exit:         None,
+        }
+    }
+
+    /// Collect failure exit codes from children and propagate them to a parent.
+    pub fn collect_failure(&mut self, mut success: bool) {
+        // Step 1. Decide if we need to collect child failures synchronously.
+        do self.on_exit.take_map |on_exit| {
+            if success {
+                // We succeeded, but our children might not. Need to wait for them.
+                let mut inner = unsafe { self.kill_handle.take_unwrap().unwrap() };
+                if inner.any_child_failed {
+                    success = false;
+                } else {
+                    // Lockless access to tombstones protected by unwrap barrier.
+                    success = inner.child_tombstones.take_map_default(true, |f| f());
+                }
+            }
+            on_exit(success);
+        };
+
+        // Step 2. Possibly alert possibly-watching parent to failure status.
+        // Note that as soon as parent_handle goes out of scope, the parent
+        // can successfully unwrap its handle and collect our reported status.
+        do self.watching_parent.take_map |mut parent_handle| {
+            if success {
+                // Our handle might be None if we had an exit callback, and
+                // already unwrapped it. But 'success' being true means no
+                // child failed, so there's nothing to do (see below case).
+                do self.kill_handle.take_map |own_handle| {
+                    own_handle.reparent_children_to(&mut parent_handle);
+                };
+            } else {
+                // Can inform watching parent immediately that we failed.
+                // (Note the importance of non-failing tasks NOT writing
+                // 'false', which could obscure another task's failure.)
+                parent_handle.notify_immediate_failure();
+            }
+        };
+    }
+}
diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs
index 1c3411b6247..f01ef31b196 100644
--- a/src/libstd/rt/mod.rs
+++ b/src/libstd/rt/mod.rs
@@ -280,7 +280,7 @@ pub fn run(main: ~fn()) -> int {
     let main_cell = Cell::new(main);
     let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool,
                                     main_cell.take());
-    main_task.on_exit = Some(on_exit);
+    main_task.death.on_exit = Some(on_exit);
     scheds[0].enqueue_task(main_task);
 
     // Run each scheduler in a thread.
diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs
index 449438b9205..d4a51afd506 100644
--- a/src/libstd/rt/task.rs
+++ b/src/libstd/rt/task.rs
@@ -20,6 +20,7 @@ use libc::{c_void, uintptr_t};
 use ptr;
 use prelude::*;
 use option::{Option, Some, None};
+use rt::kill::Death;
 use rt::local::Local;
 use rt::logging::StdErrLogger;
 use super::local_heap::LocalHeap;
@@ -36,8 +37,8 @@ pub struct Task {
     logger: StdErrLogger,
     unwinder: Unwinder,
     home: Option<SchedHome>,
-    join_latch: Option<~JoinLatch>,
-    on_exit: Option<~fn(bool)>,
+    death: Death,
+    join_latch: Option<~JoinLatch>, // FIXME(#7544) remove
     destroyed: bool,
     coroutine: Option<~Coroutine>
 }
@@ -86,8 +87,8 @@ impl Task {
             logger: StdErrLogger,
             unwinder: Unwinder { unwinding: false },
             home: Some(home),
+            death: Death::new(),
             join_latch: Some(JoinLatch::new_root()),
-            on_exit: None,
             destroyed: false,
             coroutine: Some(~Coroutine::new(stack_pool, start))
         }
@@ -104,8 +105,9 @@ impl Task {
             logger: StdErrLogger,
             home: Some(home),
             unwinder: Unwinder { unwinding: false },
+            // FIXME(#7544) make watching optional
+            death: self.death.new_child(),
             join_latch: Some(self.join_latch.get_mut_ref().new_child()),
-            on_exit: None,
             destroyed: false,
             coroutine: Some(~Coroutine::new(stack_pool, start))
         }
@@ -123,20 +125,8 @@ impl Task {
         }
 
         self.unwinder.try(f);
+        self.death.collect_failure(!self.unwinder.unwinding);
         self.destroy();
-
-        // Wait for children. Possibly report the exit status.
-        let local_success = !self.unwinder.unwinding;
-        let join_latch = self.join_latch.take_unwrap();
-        match self.on_exit {
-            Some(ref on_exit) => {
-                let success = join_latch.wait(local_success);
-                (*on_exit)(success);
-            }
-            None => {
-                join_latch.release(local_success);
-            }
-        }
     }
 
     /// must be called manually before finalization to clean up
diff --git a/src/libstd/rt/test.rs b/src/libstd/rt/test.rs
index e5393c84a08..a4242d83ecd 100644
--- a/src/libstd/rt/test.rs
+++ b/src/libstd/rt/test.rs
@@ -51,7 +51,7 @@ pub fn run_in_newsched_task(f: ~fn()) {
         let mut task = ~Task::new_root(&mut sched.stack_pool,
                                        f.take());
         rtdebug!("newsched_task: %x", to_uint(task));
-        task.on_exit = Some(on_exit);
+        task.death.on_exit = Some(on_exit);
         sched.enqueue_task(task);
         sched.run();
     }
@@ -109,7 +109,7 @@ pub fn run_in_mt_newsched_task(f: ~fn()) {
         };
         let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool,
                                         f_cell.take());
-        main_task.on_exit = Some(on_exit);
+        main_task.death.on_exit = Some(on_exit);
         scheds[0].enqueue_task(main_task);
 
         let mut threads = ~[];
@@ -280,7 +280,7 @@ pub fn spawntask_try(f: ~fn()) -> Result<(), ()> {
                            f.take())
         }
     };
-    new_task.on_exit = Some(on_exit);
+    new_task.death.on_exit = Some(on_exit);
 
     let sched = Local::take::<Scheduler>();
     do sched.switch_running_tasks_and_then(new_task) |sched, old_task| {
diff --git a/src/libstd/task/spawn.rs b/src/libstd/task/spawn.rs
index a08214ea40c..518b52a19fb 100644
--- a/src/libstd/task/spawn.rs
+++ b/src/libstd/task/spawn.rs
@@ -616,7 +616,7 @@ fn spawn_raw_newsched(mut opts: TaskOpts, f: ~fn()) {
                 if success { Success } else { Failure }
             )
         };
-        task.on_exit = Some(on_exit);
+        task.death.on_exit = Some(on_exit);
     }
 
     rtdebug!("spawn about to take scheduler");