748 lines
29 KiB
Rust
748 lines
29 KiB
Rust
// Copyright 2012-2013 The Rust Project Developers. See the COPYRIGHT
|
|
// file at the top-level directory of this distribution and at
|
|
// http://rust-lang.org/COPYRIGHT.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
|
|
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
|
|
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
|
|
// option. This file may not be copied, modified, or distributed
|
|
// except according to those terms.
|
|
|
|
/*!**************************************************************************
|
|
* Spawning & linked failure
|
|
*
|
|
* Several data structures are involved in task management to allow properly
|
|
* propagating failure across linked/supervised tasks.
|
|
*
|
|
* (1) The "taskgroup_arc" is an unsafe::exclusive which contains a hashset of
|
|
* all tasks that are part of the group. Some tasks are 'members', which
|
|
* means if they fail, they will kill everybody else in the taskgroup.
|
|
* Other tasks are 'descendants', which means they will not kill tasks
|
|
* from this group, but can be killed by failing members.
|
|
*
|
|
* A new one of these is created each spawn_linked or spawn_supervised.
|
|
*
|
|
* (2) The "taskgroup" is a per-task control structure that tracks a task's
|
|
* spawn configuration. It contains a reference to its taskgroup_arc, a
|
|
* reference to its node in the ancestor list (below), and an optionally
|
|
* configured notification port. These are stored in TLS.
|
|
*
|
|
* (3) The "ancestor_list" is a cons-style list of unsafe::exclusives which
|
|
* tracks 'generations' of taskgroups -- a group's ancestors are groups
|
|
* which (directly or transitively) spawn_supervised-ed them. Each task
|
|
* is recorded in the 'descendants' of each of its ancestor groups.
|
|
*
|
|
* Spawning a supervised task is O(n) in the number of generations still
|
|
* alive, and exiting (by success or failure) that task is also O(n).
|
|
*
|
|
* This diagram depicts the references between these data structures:
|
|
*
|
|
* linked_________________________________
|
|
* ___/ _________ \___
|
|
* / \ | group X | / \
|
|
* ( A ) - - - - - - - > | {A,B} {}|< - - -( B )
|
|
* \___/ |_________| \___/
|
|
* unlinked
|
|
* | __ (nil)
|
|
* | //| The following code causes this:
|
|
* |__ // /\ _________
|
|
* / \ // || | group Y | fn taskA() {
|
|
* ( C )- - - ||- - - > |{C} {D,E}| spawn(taskB);
|
|
* \___/ / \=====> |_________| spawn_unlinked(taskC);
|
|
* supervise /gen \ ...
|
|
* | __ \ 00 / }
|
|
* | //| \__/ fn taskB() { ... }
|
|
* |__ // /\ _________ fn taskC() {
|
|
* / \/ || | group Z | spawn_supervised(taskD);
|
|
* ( D )- - - ||- - - > | {D} {E} | ...
|
|
* \___/ / \=====> |_________| }
|
|
* supervise /gen \ fn taskD() {
|
|
* | __ \ 01 / spawn_supervised(taskE);
|
|
* | //| \__/ ...
|
|
* |__ // _________ }
|
|
* / \/ | group W | fn taskE() { ... }
|
|
* ( E )- - - - - - - > | {E} {} |
|
|
* \___/ |_________|
|
|
*
|
|
* "tcb" "taskgroup_arc"
|
|
* "ancestor_list"
|
|
*
|
|
****************************************************************************/
|
|
|
|
#[doc(hidden)];
|
|
|
|
use prelude::*;
|
|
|
|
use cast::transmute;
|
|
use cast;
|
|
use cell::Cell;
|
|
use container::MutableMap;
|
|
use comm::{Chan, GenericChan, oneshot};
|
|
use hashmap::{HashSet, HashSetMoveIterator};
|
|
use local_data;
|
|
use task::{Failure, SingleThreaded};
|
|
use task::{Success, TaskOpts, TaskResult};
|
|
use task::unkillable;
|
|
use uint;
|
|
use util;
|
|
use unstable::sync::Exclusive;
|
|
use rt::in_green_task_context;
|
|
use rt::local::Local;
|
|
use rt::task::{Task, Sched};
|
|
use rt::kill::KillHandle;
|
|
use rt::sched::Scheduler;
|
|
use rt::uv::uvio::UvEventLoop;
|
|
use rt::thread::Thread;
|
|
use rt::work_queue::WorkQueue;
|
|
|
|
#[cfg(test)] use task::default_task_opts;
|
|
#[cfg(test)] use comm;
|
|
#[cfg(test)] use task;
|
|
|
|
struct TaskSet(HashSet<KillHandle>);
|
|
|
|
impl TaskSet {
|
|
#[inline]
|
|
fn new() -> TaskSet {
|
|
TaskSet(HashSet::new())
|
|
}
|
|
#[inline]
|
|
fn insert(&mut self, task: KillHandle) {
|
|
let didnt_overwrite = (**self).insert(task);
|
|
assert!(didnt_overwrite);
|
|
}
|
|
#[inline]
|
|
fn remove(&mut self, task: &KillHandle) {
|
|
let was_present = (**self).remove(task);
|
|
assert!(was_present);
|
|
}
|
|
#[inline]
|
|
fn move_iter(self) -> HashSetMoveIterator<KillHandle> {
|
|
(*self).move_iter()
|
|
}
|
|
}
|
|
|
|
// One of these per group of linked-failure tasks.
|
|
struct TaskGroupData {
|
|
// All tasks which might kill this group. When this is empty, the group
|
|
// can be "GC"ed (i.e., its link in the ancestor list can be removed).
|
|
members: TaskSet,
|
|
// All tasks unidirectionally supervised by (directly or transitively)
|
|
// tasks in this group.
|
|
descendants: TaskSet,
|
|
}
|
|
type TaskGroupArc = Exclusive<Option<TaskGroupData>>;
|
|
|
|
type TaskGroupInner<'self> = &'self mut Option<TaskGroupData>;
|
|
|
|
// A taskgroup is 'dead' when nothing can cause it to fail; only members can.
|
|
fn taskgroup_is_dead(tg: &TaskGroupData) -> bool {
|
|
tg.members.is_empty()
|
|
}
|
|
|
|
// A list-like structure by which taskgroups keep track of all ancestor groups
|
|
// which may kill them. Needed for tasks to be able to remove themselves from
|
|
// ancestor groups upon exit. The list has a node for each "generation", and
|
|
// ends either at the root taskgroup (which has no ancestors) or at a
|
|
// taskgroup which was spawned-unlinked. Tasks from intermediate generations
|
|
// have references to the middle of the list; when intermediate generations
|
|
// die, their node in the list will be collected at a descendant's spawn-time.
|
|
struct AncestorNode {
|
|
// Since the ancestor list is recursive, we end up with references to
|
|
// exclusives within other exclusives. This is dangerous business (if
|
|
// circular references arise, deadlock and memory leaks are imminent).
|
|
// Hence we assert that this counter monotonically decreases as we
|
|
// approach the tail of the list.
|
|
generation: uint,
|
|
// Handle to the tasks in the group of the current generation.
|
|
parent_group: TaskGroupArc,
|
|
// Recursive rest of the list.
|
|
ancestors: AncestorList,
|
|
}
|
|
|
|
struct AncestorList(Option<Exclusive<AncestorNode>>);
|
|
|
|
// Accessors for taskgroup arcs and ancestor arcs that wrap the unsafety.
|
|
#[inline]
|
|
fn access_group<U>(x: &TaskGroupArc, blk: &fn(TaskGroupInner) -> U) -> U {
|
|
unsafe {
|
|
x.with(blk)
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
fn access_ancestors<U>(x: &Exclusive<AncestorNode>,
|
|
blk: &fn(x: &mut AncestorNode) -> U) -> U {
|
|
unsafe {
|
|
x.with(blk)
|
|
}
|
|
}
|
|
|
|
#[inline] #[cfg(test)]
|
|
fn check_generation(younger: uint, older: uint) { assert!(younger > older); }
|
|
#[inline] #[cfg(not(test))]
|
|
fn check_generation(_younger: uint, _older: uint) { }
|
|
|
|
#[inline] #[cfg(test)]
|
|
fn incr_generation(ancestors: &AncestorList) -> uint {
|
|
ancestors.map_default(0, |arc| access_ancestors(arc, |a| a.generation+1))
|
|
}
|
|
#[inline] #[cfg(not(test))]
|
|
fn incr_generation(_ancestors: &AncestorList) -> uint { 0 }
|
|
|
|
// Iterates over an ancestor list.
|
|
// (1) Runs forward_blk on each ancestral taskgroup in the list
|
|
// (2) If forward_blk "break"s, runs optional bail_blk on all ancestral
|
|
// taskgroups that forward_blk already ran on successfully (Note: bail_blk
|
|
// is NOT called on the block that forward_blk broke on!).
|
|
// (3) As a bonus, coalesces away all 'dead' taskgroup nodes in the list.
|
|
fn each_ancestor(list: &mut AncestorList,
|
|
bail_blk: &fn(TaskGroupInner),
|
|
forward_blk: &fn(TaskGroupInner) -> bool)
|
|
-> bool {
|
|
// "Kickoff" call - there was no last generation.
|
|
return !coalesce(list, bail_blk, forward_blk, uint::max_value);
|
|
|
|
// Recursively iterates, and coalesces afterwards if needed. Returns
|
|
// whether or not unwinding is needed (i.e., !successful iteration).
|
|
fn coalesce(list: &mut AncestorList,
|
|
bail_blk: &fn(TaskGroupInner),
|
|
forward_blk: &fn(TaskGroupInner) -> bool,
|
|
last_generation: uint) -> bool {
|
|
let (coalesce_this, early_break) =
|
|
iterate(list, bail_blk, forward_blk, last_generation);
|
|
// What should our next ancestor end up being?
|
|
if coalesce_this.is_some() {
|
|
// Needed coalesce. Our next ancestor becomes our old
|
|
// ancestor's next ancestor. ("next = old_next->next;")
|
|
*list = coalesce_this.unwrap();
|
|
}
|
|
return early_break;
|
|
}
|
|
|
|
// Returns an optional list-to-coalesce and whether unwinding is needed.
|
|
// Option<ancestor_list>:
|
|
// Whether or not the ancestor taskgroup being iterated over is
|
|
// dead or not; i.e., it has no more tasks left in it, whether or not
|
|
// it has descendants. If dead, the caller shall coalesce it away.
|
|
// bool:
|
|
// True if the supplied block did 'break', here or in any recursive
|
|
// calls. If so, must call the unwinder on all previous nodes.
|
|
fn iterate(ancestors: &mut AncestorList,
|
|
bail_blk: &fn(TaskGroupInner),
|
|
forward_blk: &fn(TaskGroupInner) -> bool,
|
|
last_generation: uint)
|
|
-> (Option<AncestorList>, bool) {
|
|
// At each step of iteration, three booleans are at play which govern
|
|
// how the iteration should behave.
|
|
// 'nobe_is_dead' - Should the list should be coalesced at this point?
|
|
// Largely unrelated to the other two.
|
|
// 'need_unwind' - Should we run the bail_blk at this point? (i.e.,
|
|
// do_continue was false not here, but down the line)
|
|
// 'do_continue' - Did the forward_blk succeed at this point? (i.e.,
|
|
// should we recurse? or should our callers unwind?)
|
|
|
|
let forward_blk = Cell::new(forward_blk);
|
|
|
|
// The map defaults to None, because if ancestors is None, we're at
|
|
// the end of the list, which doesn't make sense to coalesce.
|
|
do ancestors.map_default((None,false)) |ancestor_arc| {
|
|
// NB: Takes a lock! (this ancestor node)
|
|
do access_ancestors(ancestor_arc) |nobe| {
|
|
// Argh, but we couldn't give it to coalesce() otherwise.
|
|
let forward_blk = forward_blk.take();
|
|
// Check monotonicity
|
|
check_generation(last_generation, nobe.generation);
|
|
/*##########################################################*
|
|
* Step 1: Look at this ancestor group (call iterator block).
|
|
*##########################################################*/
|
|
let mut nobe_is_dead = false;
|
|
let do_continue =
|
|
// NB: Takes a lock! (this ancestor node's parent group)
|
|
do access_group(&nobe.parent_group) |tg_opt| {
|
|
// Decide whether this group is dead. Note that the
|
|
// group being *dead* is disjoint from it *failing*.
|
|
nobe_is_dead = match *tg_opt {
|
|
Some(ref tg) => taskgroup_is_dead(tg),
|
|
None => nobe_is_dead
|
|
};
|
|
// Call iterator block. (If the group is dead, it's
|
|
// safe to skip it. This will leave our KillHandle
|
|
// hanging around in the group even after it's freed,
|
|
// but that's ok because, by virtue of the group being
|
|
// dead, nobody will ever kill-all (for) over it.)
|
|
if nobe_is_dead { true } else { forward_blk(tg_opt) }
|
|
};
|
|
/*##########################################################*
|
|
* Step 2: Recurse on the rest of the list; maybe coalescing.
|
|
*##########################################################*/
|
|
// 'need_unwind' is only set if blk returned true above, *and*
|
|
// the recursive call early-broke.
|
|
let mut need_unwind = false;
|
|
if do_continue {
|
|
// NB: Takes many locks! (ancestor nodes & parent groups)
|
|
need_unwind = coalesce(&mut nobe.ancestors, |tg| bail_blk(tg),
|
|
forward_blk, nobe.generation);
|
|
}
|
|
/*##########################################################*
|
|
* Step 3: Maybe unwind; compute return info for our caller.
|
|
*##########################################################*/
|
|
if need_unwind && !nobe_is_dead {
|
|
do access_group(&nobe.parent_group) |tg_opt| {
|
|
bail_blk(tg_opt)
|
|
}
|
|
}
|
|
// Decide whether our caller should unwind.
|
|
need_unwind = need_unwind || !do_continue;
|
|
// Tell caller whether or not to coalesce and/or unwind
|
|
if nobe_is_dead {
|
|
// Swap the list out here; the caller replaces us with it.
|
|
let rest = util::replace(&mut nobe.ancestors,
|
|
AncestorList(None));
|
|
(Some(rest), need_unwind)
|
|
} else {
|
|
(None, need_unwind)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// One of these per task.
|
|
pub struct Taskgroup {
|
|
// List of tasks with whose fates this one's is intertwined.
|
|
tasks: TaskGroupArc, // 'none' means the group has failed.
|
|
// Lists of tasks who will kill us if they fail, but whom we won't kill.
|
|
ancestors: AncestorList,
|
|
notifier: Option<AutoNotify>,
|
|
}
|
|
|
|
impl Drop for Taskgroup {
|
|
// Runs on task exit.
|
|
fn drop(&mut self) {
|
|
// If we are failing, the whole taskgroup needs to die.
|
|
do RuntimeGlue::with_task_handle_and_failing |me, failing| {
|
|
if failing {
|
|
for x in self.notifier.mut_iter() {
|
|
x.failed = true;
|
|
}
|
|
// Take everybody down with us. After this point, every
|
|
// other task in the group will see 'tg' as none, which
|
|
// indicates the whole taskgroup is failing (and forbids
|
|
// new spawns from succeeding).
|
|
let tg = do access_group(&self.tasks) |tg| { tg.take() };
|
|
// It's safe to send kill signals outside the lock, because
|
|
// we have a refcount on all kill-handles in the group.
|
|
kill_taskgroup(tg, me);
|
|
} else {
|
|
// Remove ourselves from the group(s).
|
|
do access_group(&self.tasks) |tg| {
|
|
leave_taskgroup(tg, me, true);
|
|
}
|
|
}
|
|
// It doesn't matter whether this happens before or after dealing
|
|
// with our own taskgroup, so long as both happen before we die.
|
|
// We remove ourself from every ancestor we can, so no cleanup; no
|
|
// break.
|
|
do each_ancestor(&mut self.ancestors, |_| {}) |ancestor_group| {
|
|
leave_taskgroup(ancestor_group, me, false);
|
|
true
|
|
};
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn Taskgroup(tasks: TaskGroupArc,
|
|
ancestors: AncestorList,
|
|
mut notifier: Option<AutoNotify>) -> Taskgroup {
|
|
for x in notifier.mut_iter() {
|
|
x.failed = false;
|
|
}
|
|
|
|
Taskgroup {
|
|
tasks: tasks,
|
|
ancestors: ancestors,
|
|
notifier: notifier
|
|
}
|
|
}
|
|
|
|
struct AutoNotify {
|
|
notify_chan: Chan<TaskResult>,
|
|
failed: bool,
|
|
}
|
|
|
|
impl Drop for AutoNotify {
|
|
fn drop(&mut self) {
|
|
let result = if self.failed { Failure } else { Success };
|
|
self.notify_chan.send(result);
|
|
}
|
|
}
|
|
|
|
fn AutoNotify(chan: Chan<TaskResult>) -> AutoNotify {
|
|
AutoNotify {
|
|
notify_chan: chan,
|
|
failed: true // Un-set above when taskgroup successfully made.
|
|
}
|
|
}
|
|
|
|
fn enlist_in_taskgroup(state: TaskGroupInner, me: KillHandle,
|
|
is_member: bool) -> bool {
|
|
let me = Cell::new(me); // :(
|
|
// If 'None', the group was failing. Can't enlist.
|
|
do state.map_mut_default(false) |group| {
|
|
(if is_member {
|
|
&mut group.members
|
|
} else {
|
|
&mut group.descendants
|
|
}).insert(me.take());
|
|
true
|
|
}
|
|
}
|
|
|
|
// NB: Runs in destructor/post-exit context. Can't 'fail'.
|
|
fn leave_taskgroup(state: TaskGroupInner, me: &KillHandle, is_member: bool) {
|
|
let me = Cell::new(me); // :(
|
|
// If 'None', already failing and we've already gotten a kill signal.
|
|
do state.map_mut |group| {
|
|
(if is_member {
|
|
&mut group.members
|
|
} else {
|
|
&mut group.descendants
|
|
}).remove(me.take());
|
|
};
|
|
}
|
|
|
|
// NB: Runs in destructor/post-exit context. Can't 'fail'.
|
|
fn kill_taskgroup(state: Option<TaskGroupData>, me: &KillHandle) {
|
|
// Might already be None, if somebody is failing simultaneously.
|
|
// That's ok; only one task needs to do the dirty work. (Might also
|
|
// see 'None' if somebody already failed and we got a kill signal.)
|
|
do state.map_move |TaskGroupData { members: members, descendants: descendants }| {
|
|
for sibling in members.move_iter() {
|
|
// Skip self - killing ourself won't do much good.
|
|
if &sibling != me {
|
|
RuntimeGlue::kill_task(sibling);
|
|
}
|
|
}
|
|
for child in descendants.move_iter() {
|
|
assert!(&child != me);
|
|
RuntimeGlue::kill_task(child);
|
|
}
|
|
};
|
|
// (note: multiple tasks may reach this point)
|
|
}
|
|
|
|
// FIXME (#2912): Work around core-vs-coretest function duplication. Can't use
|
|
// a proper closure because the #[test]s won't understand. Have to fake it.
|
|
fn taskgroup_key() -> local_data::Key<@@mut Taskgroup> {
|
|
unsafe { cast::transmute(-2) }
|
|
}
|
|
|
|
// Transitionary.
|
|
struct RuntimeGlue;
|
|
impl RuntimeGlue {
|
|
fn kill_task(mut handle: KillHandle) {
|
|
do handle.kill().map_move |killed_task| {
|
|
let killed_task = Cell::new(killed_task);
|
|
do Local::borrow |sched: &mut Scheduler| {
|
|
sched.enqueue_task(killed_task.take());
|
|
}
|
|
};
|
|
}
|
|
|
|
fn with_task_handle_and_failing(blk: &fn(&KillHandle, bool)) {
|
|
rtassert!(in_green_task_context());
|
|
unsafe {
|
|
// Can't use safe borrow, because the taskgroup destructor needs to
|
|
// access the scheduler again to send kill signals to other tasks.
|
|
let me: *mut Task = Local::unsafe_borrow();
|
|
blk((*me).death.kill_handle.get_ref(), (*me).unwinder.unwinding)
|
|
}
|
|
}
|
|
|
|
fn with_my_taskgroup<U>(blk: &fn(&Taskgroup) -> U) -> U {
|
|
rtassert!(in_green_task_context());
|
|
unsafe {
|
|
// Can't use safe borrow, because creating new hashmaps for the
|
|
// tasksets requires an rng, which needs to borrow the sched.
|
|
let me: *mut Task = Local::unsafe_borrow();
|
|
blk(match (*me).taskgroup {
|
|
None => {
|
|
// First task in its (unlinked/unsupervised) taskgroup.
|
|
// Lazily initialize.
|
|
let mut members = TaskSet::new();
|
|
let my_handle = (*me).death.kill_handle.get_ref().clone();
|
|
members.insert(my_handle);
|
|
let tasks = Exclusive::new(Some(TaskGroupData {
|
|
members: members,
|
|
descendants: TaskSet::new(),
|
|
}));
|
|
let group = Taskgroup(tasks, AncestorList(None), None);
|
|
(*me).taskgroup = Some(group);
|
|
(*me).taskgroup.get_ref()
|
|
}
|
|
Some(ref group) => group,
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
// Returns 'None' in the case where the child's TG should be lazily initialized.
|
|
fn gen_child_taskgroup(linked: bool, supervised: bool)
|
|
-> Option<(TaskGroupArc, AncestorList)> {
|
|
if linked || supervised {
|
|
// with_my_taskgroup will lazily initialize the parent's taskgroup if
|
|
// it doesn't yet exist. We don't want to call it in the unlinked case.
|
|
do RuntimeGlue::with_my_taskgroup |spawner_group| {
|
|
let ancestors = AncestorList(spawner_group.ancestors.map(|x| x.clone()));
|
|
if linked {
|
|
// Child is in the same group as spawner.
|
|
// Child's ancestors are spawner's ancestors.
|
|
Some((spawner_group.tasks.clone(), ancestors))
|
|
} else {
|
|
// Child is in a separate group from spawner.
|
|
let g = Exclusive::new(Some(TaskGroupData {
|
|
members: TaskSet::new(),
|
|
descendants: TaskSet::new(),
|
|
}));
|
|
let a = if supervised {
|
|
let new_generation = incr_generation(&ancestors);
|
|
assert!(new_generation < uint::max_value);
|
|
// Child's ancestors start with the spawner.
|
|
// Build a new node in the ancestor list.
|
|
AncestorList(Some(Exclusive::new(AncestorNode {
|
|
generation: new_generation,
|
|
parent_group: spawner_group.tasks.clone(),
|
|
ancestors: ancestors,
|
|
})))
|
|
} else {
|
|
// Child has no ancestors.
|
|
AncestorList(None)
|
|
};
|
|
Some((g, a))
|
|
}
|
|
}
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
// Set up membership in taskgroup and descendantship in all ancestor
|
|
// groups. If any enlistment fails, Some task was already failing, so
|
|
// don't let the child task run, and undo every successful enlistment.
|
|
fn enlist_many(child: &KillHandle, child_arc: &TaskGroupArc,
|
|
ancestors: &mut AncestorList) -> bool {
|
|
// Join this taskgroup.
|
|
let mut result = do access_group(child_arc) |child_tg| {
|
|
enlist_in_taskgroup(child_tg, child.clone(), true) // member
|
|
};
|
|
if result {
|
|
// Unwinding function in case any ancestral enlisting fails
|
|
let bail: &fn(TaskGroupInner) = |tg| { leave_taskgroup(tg, child, false) };
|
|
// Attempt to join every ancestor group.
|
|
result = do each_ancestor(ancestors, bail) |ancestor_tg| {
|
|
// Enlist as a descendant, not as an actual member.
|
|
// Descendants don't kill ancestor groups on failure.
|
|
enlist_in_taskgroup(ancestor_tg, child.clone(), false)
|
|
};
|
|
// If any ancestor group fails, need to exit this group too.
|
|
if !result {
|
|
do access_group(child_arc) |child_tg| {
|
|
leave_taskgroup(child_tg, child, true); // member
|
|
}
|
|
}
|
|
}
|
|
result
|
|
}
|
|
|
|
pub fn spawn_raw(mut opts: TaskOpts, f: ~fn()) {
|
|
use rt::sched::*;
|
|
|
|
rtassert!(in_green_task_context());
|
|
|
|
let child_data = Cell::new(gen_child_taskgroup(opts.linked, opts.supervised));
|
|
let indestructible = opts.indestructible;
|
|
|
|
let child_wrapper: ~fn() = || {
|
|
// Child task runs this code.
|
|
|
|
// If child data is 'None', the enlist is vacuously successful.
|
|
let enlist_success = do child_data.take().map_move_default(true) |child_data| {
|
|
let child_data = Cell::new(child_data); // :(
|
|
do Local::borrow |me: &mut Task| {
|
|
let (child_tg, ancestors) = child_data.take();
|
|
let mut ancestors = ancestors;
|
|
let handle = me.death.kill_handle.get_ref();
|
|
// Atomically try to get into all of our taskgroups.
|
|
if enlist_many(handle, &child_tg, &mut ancestors) {
|
|
// Got in. We can run the provided child body, and can also run
|
|
// the taskgroup's exit-time-destructor afterward.
|
|
me.taskgroup = Some(Taskgroup(child_tg, ancestors, None));
|
|
true
|
|
} else {
|
|
false
|
|
}
|
|
}
|
|
};
|
|
// Should be run after the local-borrowed task is returned.
|
|
if enlist_success {
|
|
if indestructible {
|
|
do unkillable { f() }
|
|
} else {
|
|
f()
|
|
}
|
|
}
|
|
};
|
|
|
|
let mut task = if opts.sched.mode != SingleThreaded {
|
|
if opts.watched {
|
|
Task::build_child(opts.stack_size, child_wrapper)
|
|
} else {
|
|
Task::build_root(opts.stack_size, child_wrapper)
|
|
}
|
|
} else {
|
|
unsafe {
|
|
// Creating a 1:1 task:thread ...
|
|
let sched: *mut Scheduler = Local::unsafe_borrow();
|
|
let sched_handle = (*sched).make_handle();
|
|
|
|
// Since this is a 1:1 scheduler we create a queue not in
|
|
// the stealee set. The run_anything flag is set false
|
|
// which will disable stealing.
|
|
let work_queue = WorkQueue::new();
|
|
|
|
// Create a new scheduler to hold the new task
|
|
let new_loop = ~UvEventLoop::new();
|
|
let mut new_sched = ~Scheduler::new_special(new_loop,
|
|
work_queue,
|
|
(*sched).work_queues.clone(),
|
|
(*sched).sleeper_list.clone(),
|
|
false,
|
|
Some(sched_handle));
|
|
let mut new_sched_handle = new_sched.make_handle();
|
|
|
|
// Allow the scheduler to exit when the pinned task exits
|
|
new_sched_handle.send(Shutdown);
|
|
|
|
// Pin the new task to the new scheduler
|
|
let new_task = if opts.watched {
|
|
Task::build_homed_child(opts.stack_size, child_wrapper, Sched(new_sched_handle))
|
|
} else {
|
|
Task::build_homed_root(opts.stack_size, child_wrapper, Sched(new_sched_handle))
|
|
};
|
|
|
|
// Create a task that will later be used to join with the new scheduler
|
|
// thread when it is ready to terminate
|
|
let (thread_port, thread_chan) = oneshot();
|
|
let thread_port_cell = Cell::new(thread_port);
|
|
let join_task = do Task::build_child(None) {
|
|
rtdebug!("running join task");
|
|
let thread_port = thread_port_cell.take();
|
|
let thread: Thread = thread_port.recv();
|
|
thread.join();
|
|
};
|
|
|
|
// Put the scheduler into another thread
|
|
let new_sched_cell = Cell::new(new_sched);
|
|
let orig_sched_handle_cell = Cell::new((*sched).make_handle());
|
|
let join_task_cell = Cell::new(join_task);
|
|
|
|
let thread = do Thread::start {
|
|
let mut new_sched = new_sched_cell.take();
|
|
let mut orig_sched_handle = orig_sched_handle_cell.take();
|
|
let join_task = join_task_cell.take();
|
|
|
|
let bootstrap_task = ~do Task::new_root(&mut new_sched.stack_pool, None) || {
|
|
rtdebug!("boostrapping a 1:1 scheduler");
|
|
};
|
|
new_sched.bootstrap(bootstrap_task);
|
|
|
|
rtdebug!("enqueing join_task");
|
|
// Now tell the original scheduler to join with this thread
|
|
// by scheduling a thread-joining task on the original scheduler
|
|
orig_sched_handle.send(TaskFromFriend(join_task));
|
|
|
|
// NB: We can't simply send a message from here to another task
|
|
// because this code isn't running in a task and message passing doesn't
|
|
// work outside of tasks. Hence we're sending a scheduler message
|
|
// to execute a new task directly to a scheduler.
|
|
};
|
|
|
|
// Give the thread handle to the join task
|
|
thread_chan.send(thread);
|
|
|
|
// When this task is enqueued on the current scheduler it will then get
|
|
// forwarded to the scheduler to which it is pinned
|
|
new_task
|
|
}
|
|
};
|
|
|
|
if opts.notify_chan.is_some() {
|
|
let notify_chan = opts.notify_chan.take_unwrap();
|
|
let notify_chan = Cell::new(notify_chan);
|
|
let on_exit: ~fn(bool) = |success| {
|
|
notify_chan.take().send(
|
|
if success { Success } else { Failure }
|
|
)
|
|
};
|
|
task.death.on_exit = Some(on_exit);
|
|
}
|
|
|
|
task.name = opts.name.take();
|
|
rtdebug!("spawn calling run_task");
|
|
Scheduler::run_task(task);
|
|
|
|
}
|
|
|
|
#[test]
|
|
fn test_spawn_raw_simple() {
|
|
let (po, ch) = stream();
|
|
do spawn_raw(default_task_opts()) {
|
|
ch.send(());
|
|
}
|
|
po.recv();
|
|
}
|
|
|
|
#[test]
|
|
fn test_spawn_raw_unsupervise() {
|
|
let opts = task::TaskOpts {
|
|
linked: false,
|
|
watched: false,
|
|
notify_chan: None,
|
|
.. default_task_opts()
|
|
};
|
|
do spawn_raw(opts) {
|
|
fail2!();
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_spawn_raw_notify_success() {
|
|
let (notify_po, notify_ch) = comm::stream();
|
|
|
|
let opts = task::TaskOpts {
|
|
notify_chan: Some(notify_ch),
|
|
.. default_task_opts()
|
|
};
|
|
do spawn_raw(opts) {
|
|
}
|
|
assert_eq!(notify_po.recv(), Success);
|
|
}
|
|
|
|
#[test]
|
|
fn test_spawn_raw_notify_failure() {
|
|
// New bindings for these
|
|
let (notify_po, notify_ch) = comm::stream();
|
|
|
|
let opts = task::TaskOpts {
|
|
linked: false,
|
|
watched: false,
|
|
notify_chan: Some(notify_ch),
|
|
.. default_task_opts()
|
|
};
|
|
do spawn_raw(opts) {
|
|
fail2!();
|
|
}
|
|
assert_eq!(notify_po.recv(), Failure);
|
|
}
|