1 // Copyright 2012-2013 The Rust Project Developers. See the COPYRIGHT
2 // file at the top-level directory of this distribution and at
3 // http://rust-lang.org/COPYRIGHT.
5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
8 // option. This file may not be copied, modified, or distributed
9 // except according to those terms.
11 /*!**************************************************************************
12 * Spawning & linked failure
14 * Several data structures are involved in task management to allow properly
15 * propagating failure across linked/supervised tasks.
17 * (1) The "taskgroup_arc" is an unsafe::exclusive which contains a hashset of
18 * all tasks that are part of the group. Some tasks are 'members', which
19 * means if they fail, they will kill everybody else in the taskgroup.
20 * Other tasks are 'descendants', which means they will not kill tasks
21 * from this group, but can be killed by failing members.
23 * A new one of these is created each spawn_linked or spawn_supervised.
25 * (2) The "taskgroup" is a per-task control structure that tracks a task's
26 * spawn configuration. It contains a reference to its taskgroup_arc, a
27 * reference to its node in the ancestor list (below), and an optionally
28 * configured notification port. These are stored in TLS.
30 * (3) The "ancestor_list" is a cons-style list of unsafe::exclusives which
31 * tracks 'generations' of taskgroups -- a group's ancestors are groups
32 * which (directly or transitively) spawn_supervised-ed them. Each task
33 * is recorded in the 'descendants' of each of its ancestor groups.
35 * Spawning a supervised task is O(n) in the number of generations still
36 * alive, and exiting (by success or failure) that task is also O(n).
38 * This diagram depicts the references between these data structures:
40 * linked_________________________________
43 * ( A ) - - - - - - - > | {A,B} {}|< - - -( B )
44 * \___/ |_________| \___/
47 * | //| The following code causes this:
49 * / \ // || | group Y | fn taskA() {
50 * ( C )- - - ||- - - > |{C} {D,E}| spawn(taskB);
51 * \___/ / \=====> |_________| spawn_unlinked(taskC);
52 * supervise /gen \ ...
54 * | //| \__/ fn taskB() { ... }
55 * |__ // /\ _________ fn taskC() {
56 * / \/ || | group Z | spawn_supervised(taskD);
57 * ( D )- - - ||- - - > | {D} {E} | ...
58 * \___/ / \=====> |_________| }
59 * supervise /gen \ fn taskD() {
60 * | __ \ 01 / spawn_supervised(taskE);
63 * / \/ | group W | fn taskE() { ... }
64 * ( E )- - - - - - - > | {E} {} |
67 * "tcb" "taskgroup_arc"
70 ****************************************************************************/
79 use container::MutableMap;
80 use comm::{Chan, GenericChan, oneshot};
81 use hashmap::{HashSet, HashSetMoveIterator};
83 use task::{Failure, SingleThreaded};
84 use task::{Success, TaskOpts, TaskResult};
88 use unstable::sync::Exclusive;
89 use rt::in_green_task_context;
91 use rt::task::{Task, Sched};
92 use rt::kill::KillHandle;
93 use rt::sched::Scheduler;
94 use rt::uv::uvio::UvEventLoop;
95 use rt::thread::Thread;
96 use rt::work_queue::WorkQueue;
98 #[cfg(test)] use task::default_task_opts;
99 #[cfg(test)] use comm;
100 #[cfg(test)] use task;
102 struct TaskSet(HashSet<KillHandle>);
106 fn new() -> TaskSet {
107 TaskSet(HashSet::new())
110 fn insert(&mut self, task: KillHandle) {
111 let didnt_overwrite = (**self).insert(task);
112 assert!(didnt_overwrite);
115 fn remove(&mut self, task: &KillHandle) {
116 let was_present = (**self).remove(task);
117 assert!(was_present);
120 fn move_iter(self) -> HashSetMoveIterator<KillHandle> {
125 // One of these per group of linked-failure tasks.
126 struct TaskGroupData {
127 // All tasks which might kill this group. When this is empty, the group
128 // can be "GC"ed (i.e., its link in the ancestor list can be removed).
130 // All tasks unidirectionally supervised by (directly or transitively)
131 // tasks in this group.
132 descendants: TaskSet,
134 type TaskGroupArc = Exclusive<Option<TaskGroupData>>;
136 type TaskGroupInner<'self> = &'self mut Option<TaskGroupData>;
138 // A taskgroup is 'dead' when nothing can cause it to fail; only members can.
139 fn taskgroup_is_dead(tg: &TaskGroupData) -> bool {
140 tg.members.is_empty()
143 // A list-like structure by which taskgroups keep track of all ancestor groups
144 // which may kill them. Needed for tasks to be able to remove themselves from
145 // ancestor groups upon exit. The list has a node for each "generation", and
146 // ends either at the root taskgroup (which has no ancestors) or at a
147 // taskgroup which was spawned-unlinked. Tasks from intermediate generations
148 // have references to the middle of the list; when intermediate generations
149 // die, their node in the list will be collected at a descendant's spawn-time.
150 struct AncestorNode {
151 // Since the ancestor list is recursive, we end up with references to
152 // exclusives within other exclusives. This is dangerous business (if
153 // circular references arise, deadlock and memory leaks are imminent).
154 // Hence we assert that this counter monotonically decreases as we
155 // approach the tail of the list.
157 // Handle to the tasks in the group of the current generation.
158 parent_group: TaskGroupArc,
159 // Recursive rest of the list.
160 ancestors: AncestorList,
163 struct AncestorList(Option<Exclusive<AncestorNode>>);
165 // Accessors for taskgroup arcs and ancestor arcs that wrap the unsafety.
167 fn access_group<U>(x: &TaskGroupArc, blk: &fn(TaskGroupInner) -> U) -> U {
174 fn access_ancestors<U>(x: &Exclusive<AncestorNode>,
175 blk: &fn(x: &mut AncestorNode) -> U) -> U {
181 #[inline] #[cfg(test)]
182 fn check_generation(younger: uint, older: uint) { assert!(younger > older); }
183 #[inline] #[cfg(not(test))]
184 fn check_generation(_younger: uint, _older: uint) { }
186 #[inline] #[cfg(test)]
187 fn incr_generation(ancestors: &AncestorList) -> uint {
188 ancestors.map_default(0, |arc| access_ancestors(arc, |a| a.generation+1))
190 #[inline] #[cfg(not(test))]
191 fn incr_generation(_ancestors: &AncestorList) -> uint { 0 }
193 // Iterates over an ancestor list.
194 // (1) Runs forward_blk on each ancestral taskgroup in the list
195 // (2) If forward_blk "break"s, runs optional bail_blk on all ancestral
196 // taskgroups that forward_blk already ran on successfully (Note: bail_blk
197 // is NOT called on the block that forward_blk broke on!).
198 // (3) As a bonus, coalesces away all 'dead' taskgroup nodes in the list.
199 fn each_ancestor(list: &mut AncestorList,
200 bail_blk: &fn(TaskGroupInner),
201 forward_blk: &fn(TaskGroupInner) -> bool)
203 // "Kickoff" call - there was no last generation.
204 return !coalesce(list, bail_blk, forward_blk, uint::max_value);
206 // Recursively iterates, and coalesces afterwards if needed. Returns
207 // whether or not unwinding is needed (i.e., !successful iteration).
208 fn coalesce(list: &mut AncestorList,
209 bail_blk: &fn(TaskGroupInner),
210 forward_blk: &fn(TaskGroupInner) -> bool,
211 last_generation: uint) -> bool {
212 let (coalesce_this, early_break) =
213 iterate(list, bail_blk, forward_blk, last_generation);
214 // What should our next ancestor end up being?
215 if coalesce_this.is_some() {
216 // Needed coalesce. Our next ancestor becomes our old
217 // ancestor's next ancestor. ("next = old_next->next;")
218 *list = coalesce_this.unwrap();
223 // Returns an optional list-to-coalesce and whether unwinding is needed.
224 // Option<ancestor_list>:
225 // Whether or not the ancestor taskgroup being iterated over is
226 // dead or not; i.e., it has no more tasks left in it, whether or not
227 // it has descendants. If dead, the caller shall coalesce it away.
229 // True if the supplied block did 'break', here or in any recursive
230 // calls. If so, must call the unwinder on all previous nodes.
231 fn iterate(ancestors: &mut AncestorList,
232 bail_blk: &fn(TaskGroupInner),
233 forward_blk: &fn(TaskGroupInner) -> bool,
234 last_generation: uint)
235 -> (Option<AncestorList>, bool) {
236 // At each step of iteration, three booleans are at play which govern
237 // how the iteration should behave.
238 // 'nobe_is_dead' - Should the list should be coalesced at this point?
239 // Largely unrelated to the other two.
240 // 'need_unwind' - Should we run the bail_blk at this point? (i.e.,
241 // do_continue was false not here, but down the line)
242 // 'do_continue' - Did the forward_blk succeed at this point? (i.e.,
243 // should we recurse? or should our callers unwind?)
245 let forward_blk = Cell::new(forward_blk);
247 // The map defaults to None, because if ancestors is None, we're at
248 // the end of the list, which doesn't make sense to coalesce.
249 do ancestors.map_default((None,false)) |ancestor_arc| {
250 // NB: Takes a lock! (this ancestor node)
251 do access_ancestors(ancestor_arc) |nobe| {
252 // Argh, but we couldn't give it to coalesce() otherwise.
253 let forward_blk = forward_blk.take();
254 // Check monotonicity
255 check_generation(last_generation, nobe.generation);
256 /*##########################################################*
257 * Step 1: Look at this ancestor group (call iterator block).
258 *##########################################################*/
259 let mut nobe_is_dead = false;
261 // NB: Takes a lock! (this ancestor node's parent group)
262 do access_group(&nobe.parent_group) |tg_opt| {
263 // Decide whether this group is dead. Note that the
264 // group being *dead* is disjoint from it *failing*.
265 nobe_is_dead = match *tg_opt {
266 Some(ref tg) => taskgroup_is_dead(tg),
269 // Call iterator block. (If the group is dead, it's
270 // safe to skip it. This will leave our KillHandle
271 // hanging around in the group even after it's freed,
272 // but that's ok because, by virtue of the group being
273 // dead, nobody will ever kill-all (for) over it.)
274 if nobe_is_dead { true } else { forward_blk(tg_opt) }
276 /*##########################################################*
277 * Step 2: Recurse on the rest of the list; maybe coalescing.
278 *##########################################################*/
279 // 'need_unwind' is only set if blk returned true above, *and*
280 // the recursive call early-broke.
281 let mut need_unwind = false;
283 // NB: Takes many locks! (ancestor nodes & parent groups)
284 need_unwind = coalesce(&mut nobe.ancestors, |tg| bail_blk(tg),
285 forward_blk, nobe.generation);
287 /*##########################################################*
288 * Step 3: Maybe unwind; compute return info for our caller.
289 *##########################################################*/
290 if need_unwind && !nobe_is_dead {
291 do access_group(&nobe.parent_group) |tg_opt| {
295 // Decide whether our caller should unwind.
296 need_unwind = need_unwind || !do_continue;
297 // Tell caller whether or not to coalesce and/or unwind
299 // Swap the list out here; the caller replaces us with it.
300 let rest = util::replace(&mut nobe.ancestors,
302 (Some(rest), need_unwind)
311 // One of these per task.
312 pub struct Taskgroup {
313 // List of tasks with whose fates this one's is intertwined.
314 tasks: TaskGroupArc, // 'none' means the group has failed.
315 // Lists of tasks who will kill us if they fail, but whom we won't kill.
316 ancestors: AncestorList,
317 notifier: Option<AutoNotify>,
320 impl Drop for Taskgroup {
321 // Runs on task exit.
324 // FIXME(#4330) Need self by value to get mutability.
325 let this: &mut Taskgroup = transmute(self);
327 // If we are failing, the whole taskgroup needs to die.
328 do RuntimeGlue::with_task_handle_and_failing |me, failing| {
330 for x in this.notifier.mut_iter() {
333 // Take everybody down with us. After this point, every
334 // other task in the group will see 'tg' as none, which
335 // indicates the whole taskgroup is failing (and forbids
336 // new spawns from succeeding).
337 let tg = do access_group(&self.tasks) |tg| { tg.take() };
338 // It's safe to send kill signals outside the lock, because
339 // we have a refcount on all kill-handles in the group.
340 kill_taskgroup(tg, me);
342 // Remove ourselves from the group(s).
343 do access_group(&self.tasks) |tg| {
344 leave_taskgroup(tg, me, true);
347 // It doesn't matter whether this happens before or after dealing
348 // with our own taskgroup, so long as both happen before we die.
349 // We remove ourself from every ancestor we can, so no cleanup; no
351 do each_ancestor(&mut this.ancestors, |_| {}) |ancestor_group| {
352 leave_taskgroup(ancestor_group, me, false);
360 pub fn Taskgroup(tasks: TaskGroupArc,
361 ancestors: AncestorList,
362 mut notifier: Option<AutoNotify>) -> Taskgroup {
363 for x in notifier.mut_iter() {
369 ancestors: ancestors,
375 notify_chan: Chan<TaskResult>,
379 impl Drop for AutoNotify {
381 let result = if self.failed { Failure } else { Success };
382 self.notify_chan.send(result);
386 fn AutoNotify(chan: Chan<TaskResult>) -> AutoNotify {
389 failed: true // Un-set above when taskgroup successfully made.
393 fn enlist_in_taskgroup(state: TaskGroupInner, me: KillHandle,
394 is_member: bool) -> bool {
395 let me = Cell::new(me); // :(
396 // If 'None', the group was failing. Can't enlist.
397 do state.map_mut_default(false) |group| {
401 &mut group.descendants
402 }).insert(me.take());
407 // NB: Runs in destructor/post-exit context. Can't 'fail'.
408 fn leave_taskgroup(state: TaskGroupInner, me: &KillHandle, is_member: bool) {
409 let me = Cell::new(me); // :(
410 // If 'None', already failing and we've already gotten a kill signal.
411 do state.map_mut |group| {
415 &mut group.descendants
416 }).remove(me.take());
420 // NB: Runs in destructor/post-exit context. Can't 'fail'.
421 fn kill_taskgroup(state: Option<TaskGroupData>, me: &KillHandle) {
422 // Might already be None, if somebody is failing simultaneously.
423 // That's ok; only one task needs to do the dirty work. (Might also
424 // see 'None' if somebody already failed and we got a kill signal.)
425 do state.map_move |TaskGroupData { members: members, descendants: descendants }| {
426 for sibling in members.move_iter() {
427 // Skip self - killing ourself won't do much good.
429 RuntimeGlue::kill_task(sibling);
432 for child in descendants.move_iter() {
433 assert!(&child != me);
434 RuntimeGlue::kill_task(child);
437 // (note: multiple tasks may reach this point)
440 // FIXME (#2912): Work around core-vs-coretest function duplication. Can't use
441 // a proper closure because the #[test]s won't understand. Have to fake it.
442 fn taskgroup_key() -> local_data::Key<@@mut Taskgroup> {
443 unsafe { cast::transmute(-2) }
449 fn kill_task(mut handle: KillHandle) {
450 do handle.kill().map_move |killed_task| {
451 let killed_task = Cell::new(killed_task);
452 do Local::borrow |sched: &mut Scheduler| {
453 sched.enqueue_task(killed_task.take());
458 fn with_task_handle_and_failing(blk: &fn(&KillHandle, bool)) {
459 rtassert!(in_green_task_context());
461 // Can't use safe borrow, because the taskgroup destructor needs to
462 // access the scheduler again to send kill signals to other tasks.
463 let me: *mut Task = Local::unsafe_borrow();
464 blk((*me).death.kill_handle.get_ref(), (*me).unwinder.unwinding)
468 fn with_my_taskgroup<U>(blk: &fn(&Taskgroup) -> U) -> U {
469 rtassert!(in_green_task_context());
471 // Can't use safe borrow, because creating new hashmaps for the
472 // tasksets requires an rng, which needs to borrow the sched.
473 let me: *mut Task = Local::unsafe_borrow();
474 blk(match (*me).taskgroup {
476 // First task in its (unlinked/unsupervised) taskgroup.
477 // Lazily initialize.
478 let mut members = TaskSet::new();
479 let my_handle = (*me).death.kill_handle.get_ref().clone();
480 members.insert(my_handle);
481 let tasks = Exclusive::new(Some(TaskGroupData {
483 descendants: TaskSet::new(),
485 let group = Taskgroup(tasks, AncestorList(None), None);
486 (*me).taskgroup = Some(group);
487 (*me).taskgroup.get_ref()
489 Some(ref group) => group,
495 // Returns 'None' in the case where the child's TG should be lazily initialized.
496 fn gen_child_taskgroup(linked: bool, supervised: bool)
497 -> Option<(TaskGroupArc, AncestorList)> {
498 if linked || supervised {
499 // with_my_taskgroup will lazily initialize the parent's taskgroup if
500 // it doesn't yet exist. We don't want to call it in the unlinked case.
501 do RuntimeGlue::with_my_taskgroup |spawner_group| {
502 let ancestors = AncestorList(spawner_group.ancestors.map(|x| x.clone()));
504 // Child is in the same group as spawner.
505 // Child's ancestors are spawner's ancestors.
506 Some((spawner_group.tasks.clone(), ancestors))
508 // Child is in a separate group from spawner.
509 let g = Exclusive::new(Some(TaskGroupData {
510 members: TaskSet::new(),
511 descendants: TaskSet::new(),
513 let a = if supervised {
514 let new_generation = incr_generation(&ancestors);
515 assert!(new_generation < uint::max_value);
516 // Child's ancestors start with the spawner.
517 // Build a new node in the ancestor list.
518 AncestorList(Some(Exclusive::new(AncestorNode {
519 generation: new_generation,
520 parent_group: spawner_group.tasks.clone(),
521 ancestors: ancestors,
524 // Child has no ancestors.
535 // Set up membership in taskgroup and descendantship in all ancestor
536 // groups. If any enlistment fails, Some task was already failing, so
537 // don't let the child task run, and undo every successful enlistment.
538 fn enlist_many(child: &KillHandle, child_arc: &TaskGroupArc,
539 ancestors: &mut AncestorList) -> bool {
540 // Join this taskgroup.
541 let mut result = do access_group(child_arc) |child_tg| {
542 enlist_in_taskgroup(child_tg, child.clone(), true) // member
545 // Unwinding function in case any ancestral enlisting fails
546 let bail: &fn(TaskGroupInner) = |tg| { leave_taskgroup(tg, child, false) };
547 // Attempt to join every ancestor group.
548 result = do each_ancestor(ancestors, bail) |ancestor_tg| {
549 // Enlist as a descendant, not as an actual member.
550 // Descendants don't kill ancestor groups on failure.
551 enlist_in_taskgroup(ancestor_tg, child.clone(), false)
553 // If any ancestor group fails, need to exit this group too.
555 do access_group(child_arc) |child_tg| {
556 leave_taskgroup(child_tg, child, true); // member
563 pub fn spawn_raw(mut opts: TaskOpts, f: ~fn()) {
566 rtassert!(in_green_task_context());
568 let child_data = Cell::new(gen_child_taskgroup(opts.linked, opts.supervised));
569 let indestructible = opts.indestructible;
571 let child_wrapper: ~fn() = || {
572 // Child task runs this code.
574 // If child data is 'None', the enlist is vacuously successful.
575 let enlist_success = do child_data.take().map_move_default(true) |child_data| {
576 let child_data = Cell::new(child_data); // :(
577 do Local::borrow |me: &mut Task| {
578 let (child_tg, ancestors) = child_data.take();
579 let mut ancestors = ancestors;
580 let handle = me.death.kill_handle.get_ref();
581 // Atomically try to get into all of our taskgroups.
582 if enlist_many(handle, &child_tg, &mut ancestors) {
583 // Got in. We can run the provided child body, and can also run
584 // the taskgroup's exit-time-destructor afterward.
585 me.taskgroup = Some(Taskgroup(child_tg, ancestors, None));
592 // Should be run after the local-borrowed task is returned.
595 do unkillable { f() }
602 let mut task = if opts.sched.mode != SingleThreaded {
604 Task::build_child(opts.stack_size, child_wrapper)
606 Task::build_root(opts.stack_size, child_wrapper)
610 // Creating a 1:1 task:thread ...
611 let sched: *mut Scheduler = Local::unsafe_borrow();
612 let sched_handle = (*sched).make_handle();
614 // Since this is a 1:1 scheduler we create a queue not in
615 // the stealee set. The run_anything flag is set false
616 // which will disable stealing.
617 let work_queue = WorkQueue::new();
619 // Create a new scheduler to hold the new task
620 let new_loop = ~UvEventLoop::new();
621 let mut new_sched = ~Scheduler::new_special(new_loop,
623 (*sched).work_queues.clone(),
624 (*sched).sleeper_list.clone(),
627 let mut new_sched_handle = new_sched.make_handle();
629 // Allow the scheduler to exit when the pinned task exits
630 new_sched_handle.send(Shutdown);
632 // Pin the new task to the new scheduler
633 let new_task = if opts.watched {
634 Task::build_homed_child(opts.stack_size, child_wrapper, Sched(new_sched_handle))
636 Task::build_homed_root(opts.stack_size, child_wrapper, Sched(new_sched_handle))
639 // Create a task that will later be used to join with the new scheduler
640 // thread when it is ready to terminate
641 let (thread_port, thread_chan) = oneshot();
642 let thread_port_cell = Cell::new(thread_port);
643 let join_task = do Task::build_child(None) {
644 rtdebug!("running join task");
645 let thread_port = thread_port_cell.take();
646 let thread: Thread = thread_port.recv();
650 // Put the scheduler into another thread
651 let new_sched_cell = Cell::new(new_sched);
652 let orig_sched_handle_cell = Cell::new((*sched).make_handle());
653 let join_task_cell = Cell::new(join_task);
655 let thread = do Thread::start {
656 let mut new_sched = new_sched_cell.take();
657 let mut orig_sched_handle = orig_sched_handle_cell.take();
658 let join_task = join_task_cell.take();
660 let bootstrap_task = ~do Task::new_root(&mut new_sched.stack_pool, None) || {
661 rtdebug!("boostrapping a 1:1 scheduler");
663 new_sched.bootstrap(bootstrap_task);
665 rtdebug!("enqueing join_task");
666 // Now tell the original scheduler to join with this thread
667 // by scheduling a thread-joining task on the original scheduler
668 orig_sched_handle.send(TaskFromFriend(join_task));
670 // NB: We can't simply send a message from here to another task
671 // because this code isn't running in a task and message passing doesn't
672 // work outside of tasks. Hence we're sending a scheduler message
673 // to execute a new task directly to a scheduler.
676 // Give the thread handle to the join task
677 thread_chan.send(thread);
679 // When this task is enqueued on the current scheduler it will then get
680 // forwarded to the scheduler to which it is pinned
685 if opts.notify_chan.is_some() {
686 let notify_chan = opts.notify_chan.take_unwrap();
687 let notify_chan = Cell::new(notify_chan);
688 let on_exit: ~fn(bool) = |success| {
689 notify_chan.take().send(
690 if success { Success } else { Failure }
693 task.death.on_exit = Some(on_exit);
696 task.name = opts.name.take();
697 rtdebug!("spawn calling run_task");
698 Scheduler::run_task(task);
703 fn test_spawn_raw_simple() {
704 let (po, ch) = stream();
705 do spawn_raw(default_task_opts()) {
712 fn test_spawn_raw_unsupervise() {
713 let opts = task::TaskOpts {
717 .. default_task_opts()
725 fn test_spawn_raw_notify_success() {
726 let (notify_po, notify_ch) = comm::stream();
728 let opts = task::TaskOpts {
729 notify_chan: Some(notify_ch),
730 .. default_task_opts()
734 assert_eq!(notify_po.recv(), Success);
738 fn test_spawn_raw_notify_failure() {
739 // New bindings for these
740 let (notify_po, notify_ch) = comm::stream();
742 let opts = task::TaskOpts {
745 notify_chan: Some(notify_ch),
746 .. default_task_opts()
751 assert_eq!(notify_po.recv(), Failure);