From abc3a8aa1e76f3ecc3930e20453a52681843cec0 Mon Sep 17 00:00:00 2001 From: Brian Anderson Date: Sun, 2 Jun 2013 01:55:22 -0700 Subject: std::rt: Add JoinLatch This is supposed to be an efficient way to link the lifetimes of tasks into a tree. JoinLatches form a tree and when `release` is called they wait on children then signal the parent. This structure creates zombie tasks which currently keep the entire task allocated. Zombie tasks are supposed to be tombstoned but that code does not work correctly. --- src/libstd/rt/mod.rs | 3 +++ 1 file changed, 3 insertions(+) (limited to 'src/libstd/rt/mod.rs') diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs index caf3e15e535..2008c4a180f 100644 --- a/src/libstd/rt/mod.rs +++ b/src/libstd/rt/mod.rs @@ -133,6 +133,9 @@ pub mod local_ptr; /// Bindings to pthread/windows thread-local storage. pub mod thread_local_storage; +/// A concurrent data structure with which parent tasks wait on child tasks. +pub mod join_latch; + pub mod metrics; -- cgit 1.4.1-3-g733a5 From fd148cd3e2d08ce15272f0690f6e41d2e85ee721 Mon Sep 17 00:00:00 2001 From: Brian Anderson Date: Thu, 13 Jun 2013 22:43:20 -0700 Subject: std::rt: Change the Task constructors to reflect a tree --- src/libstd/rt/mod.rs | 4 ++-- src/libstd/rt/sched.rs | 22 +++++++++++----------- src/libstd/rt/task.rs | 26 ++++++++++++++++++++++++-- src/libstd/rt/test.rs | 46 ++++++++++++++++++++++++++++++++++++---------- src/libstd/task/spawn.rs | 9 ++++++++- 5 files changed, 81 insertions(+), 26 deletions(-) (limited to 'src/libstd/rt/mod.rs') diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs index 2008c4a180f..a65b07fdbcf 100644 --- a/src/libstd/rt/mod.rs +++ b/src/libstd/rt/mod.rs @@ -167,7 +167,7 @@ pub fn start(_argc: int, _argv: **u8, crate_map: *u8, main: ~fn()) -> int { let sleepers = SleeperList::new(); let mut sched = ~Scheduler::new(loop_, work_queue, sleepers); sched.no_sleep = true; - let main_task = ~Coroutine::new(&mut sched.stack_pool, main); + let main_task = ~Coroutine::new_root(&mut sched.stack_pool, main); sched.enqueue_task(main_task); sched.run(); @@ -241,7 +241,7 @@ fn test_context() { do run_in_bare_thread { assert_eq!(context(), GlobalContext); let mut sched = ~new_test_uv_sched(); - let task = ~do Coroutine::new(&mut sched.stack_pool) { + let task = ~do Coroutine::new_root(&mut sched.stack_pool) { assert_eq!(context(), TaskContext); let sched = Local::take::(); do sched.deschedule_running_task_and_then() |sched, task| { diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs index 104eb4b8bae..9abcc2ec3cc 100644 --- a/src/libstd/rt/sched.rs +++ b/src/libstd/rt/sched.rs @@ -518,8 +518,8 @@ impl SchedHandle { } pub impl Coroutine { - fn new(stack_pool: &mut StackPool, start: ~fn()) -> Coroutine { - Coroutine::with_task(stack_pool, ~Task::new(), start) + fn new_root(stack_pool: &mut StackPool, start: ~fn()) -> Coroutine { + Coroutine::with_task(stack_pool, ~Task::new_root(), start) } fn with_task(stack_pool: &mut StackPool, @@ -614,7 +614,7 @@ mod test { let task_ran_ptr: *mut bool = &mut task_ran; let mut sched = ~new_test_uv_sched(); - let task = ~do Coroutine::new(&mut sched.stack_pool) { + let task = ~do Coroutine::new_root(&mut sched.stack_pool) { unsafe { *task_ran_ptr = true; } }; sched.enqueue_task(task); @@ -632,7 +632,7 @@ mod test { let mut sched = ~new_test_uv_sched(); for int::range(0, total) |_| { - let task = ~do Coroutine::new(&mut sched.stack_pool) { + let task = ~do Coroutine::new_root(&mut sched.stack_pool) { unsafe { *task_count_ptr = *task_count_ptr + 1; } }; sched.enqueue_task(task); @@ -649,10 +649,10 @@ mod test { let count_ptr: *mut int = &mut count; let mut sched = ~new_test_uv_sched(); - let task1 = ~do Coroutine::new(&mut sched.stack_pool) { + let task1 = ~do Coroutine::new_root(&mut sched.stack_pool) { unsafe { *count_ptr = *count_ptr + 1; } let mut sched = Local::take::(); - let task2 = ~do Coroutine::new(&mut sched.stack_pool) { + let task2 = ~do Coroutine::new_root(&mut sched.stack_pool) { unsafe { *count_ptr = *count_ptr + 1; } }; // Context switch directly to the new task @@ -677,7 +677,7 @@ mod test { let mut sched = ~new_test_uv_sched(); - let start_task = ~do Coroutine::new(&mut sched.stack_pool) { + let start_task = ~do Coroutine::new_root(&mut sched.stack_pool) { run_task(count_ptr); }; sched.enqueue_task(start_task); @@ -687,7 +687,7 @@ mod test { fn run_task(count_ptr: *mut int) { do Local::borrow:: |sched| { - let task = ~do Coroutine::new(&mut sched.stack_pool) { + let task = ~do Coroutine::new_root(&mut sched.stack_pool) { unsafe { *count_ptr = *count_ptr + 1; if *count_ptr != MAX { @@ -705,7 +705,7 @@ mod test { fn test_block_task() { do run_in_bare_thread { let mut sched = ~new_test_uv_sched(); - let task = ~do Coroutine::new(&mut sched.stack_pool) { + let task = ~do Coroutine::new_root(&mut sched.stack_pool) { let sched = Local::take::(); assert!(sched.in_task_context()); do sched.deschedule_running_task_and_then() |sched, task| { @@ -752,13 +752,13 @@ mod test { let mut sched1 = ~new_test_uv_sched(); let handle1 = sched1.make_handle(); let handle1_cell = Cell(handle1); - let task1 = ~do Coroutine::new(&mut sched1.stack_pool) { + let task1 = ~do Coroutine::new_root(&mut sched1.stack_pool) { chan_cell.take().send(()); }; sched1.enqueue_task(task1); let mut sched2 = ~new_test_uv_sched(); - let task2 = ~do Coroutine::new(&mut sched2.stack_pool) { + let task2 = ~do Coroutine::new_root(&mut sched2.stack_pool) { port_cell.take().recv(); // Release the other scheduler's handle so it can exit handle1_cell.take(); diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs index cf4967b12b3..10b4672df05 100644 --- a/src/libstd/rt/task.rs +++ b/src/libstd/rt/task.rs @@ -37,7 +37,7 @@ pub struct Unwinder { } impl Task { - pub fn new() -> Task { + pub fn new_root() -> Task { Task { heap: LocalHeap::new(), gc: GarbageCollector, @@ -48,7 +48,29 @@ impl Task { } } - pub fn without_unwinding() -> Task { + pub fn new_root_without_unwinding() -> Task { + Task { + heap: LocalHeap::new(), + gc: GarbageCollector, + storage: LocalStorage(ptr::null(), None), + logger: StdErrLogger, + unwinder: None, + destroyed: false + } + } + + pub fn new_child(&mut self) -> Task { + Task { + heap: LocalHeap::new(), + gc: GarbageCollector, + storage: LocalStorage(ptr::null(), None), + logger: StdErrLogger, + unwinder: Some(Unwinder { unwinding: false }), + destroyed: false + } + } + + pub fn new_child_without_unwinding(&mut self) -> Task { Task { heap: LocalHeap::new(), gc: GarbageCollector, diff --git a/src/libstd/rt/test.rs b/src/libstd/rt/test.rs index c8df3a61203..4a4d498a26e 100644 --- a/src/libstd/rt/test.rs +++ b/src/libstd/rt/test.rs @@ -48,7 +48,7 @@ pub fn run_in_newsched_task(f: ~fn()) { do run_in_bare_thread { let mut sched = ~new_test_uv_sched(); let task = ~Coroutine::with_task(&mut sched.stack_pool, - ~Task::without_unwinding(), + ~Task::new_root_without_unwinding(), f.take()); sched.enqueue_task(task); sched.run(); @@ -94,7 +94,7 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { let f_cell = Cell(f_cell.take()); let handles = Cell(handles); - let main_task = ~do Coroutine::new(&mut scheds[0].stack_pool) { + let main_task = ~do Coroutine::new_root(&mut scheds[0].stack_pool) { f_cell.take()(); let mut handles = handles.take(); @@ -132,9 +132,14 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { pub fn spawntask(f: ~fn()) { use super::sched::*; + let mut task = None; + do Local::borrow::() |running_task| { + task = Some(~running_task.new_child_without_unwinding()); + } + let mut sched = Local::take::(); let task = ~Coroutine::with_task(&mut sched.stack_pool, - ~Task::without_unwinding(), + task.swap_unwrap(), f); sched.schedule_new_task(task); } @@ -143,9 +148,14 @@ pub fn spawntask(f: ~fn()) { pub fn spawntask_immediately(f: ~fn()) { use super::sched::*; + let mut task = None; + do Local::borrow::() |running_task| { + task = Some(~running_task.new_child_without_unwinding()); + } + let mut sched = Local::take::(); let task = ~Coroutine::with_task(&mut sched.stack_pool, - ~Task::without_unwinding(), + task.swap_unwrap(), f); do sched.switch_running_tasks_and_then(task) |sched, task| { sched.enqueue_task(task); @@ -156,9 +166,14 @@ pub fn spawntask_immediately(f: ~fn()) { pub fn spawntask_later(f: ~fn()) { use super::sched::*; + let mut task = None; + do Local::borrow::() |running_task| { + task = Some(~running_task.new_child_without_unwinding()); + } + let mut sched = Local::take::(); let task = ~Coroutine::with_task(&mut sched.stack_pool, - ~Task::without_unwinding(), + task.swap_unwrap(), f); sched.enqueue_task(task); @@ -170,14 +185,19 @@ pub fn spawntask_random(f: ~fn()) { use super::sched::*; use rand::{Rand, rng}; - let mut rng = rng(); - let run_now: bool = Rand::rand(&mut rng); + let mut task = None; + do Local::borrow::() |running_task| { + task = Some(~running_task.new_child_without_unwinding()); + } let mut sched = Local::take::(); let task = ~Coroutine::with_task(&mut sched.stack_pool, - ~Task::without_unwinding(), + task.swap_unwrap(), f); + let mut rng = rng(); + let run_now: bool = Rand::rand(&mut rng); + if run_now { do sched.switch_running_tasks_and_then(task) |sched, task| { sched.enqueue_task(task); @@ -206,7 +226,7 @@ pub fn spawntask_try(f: ~fn()) -> Result<(), ()> { do sched.deschedule_running_task_and_then() |sched, old_task| { let old_task = Cell(old_task); let f = f.take(); - let new_task = ~do Coroutine::new(&mut sched.stack_pool) { + let new_task = ~do Coroutine::new_root(&mut sched.stack_pool) { do (|| { (f.take())() }).finally { @@ -229,11 +249,17 @@ pub fn spawntask_try(f: ~fn()) -> Result<(), ()> { pub fn spawntask_thread(f: ~fn()) -> Thread { use rt::sched::*; + let mut task = None; + do Local::borrow::() |running_task| { + task = Some(~running_task.new_child_without_unwinding()); + } + + let task = Cell(task.swap_unwrap()); let f = Cell(f); let thread = do Thread::start { let mut sched = ~new_test_uv_sched(); let task = ~Coroutine::with_task(&mut sched.stack_pool, - ~Task::without_unwinding(), + task.take(), f.take()); sched.enqueue_task(task); sched.run(); diff --git a/src/libstd/task/spawn.rs b/src/libstd/task/spawn.rs index 5941221821a..a4fbec11d72 100644 --- a/src/libstd/task/spawn.rs +++ b/src/libstd/task/spawn.rs @@ -91,6 +91,7 @@ use uint; use util; use unstable::sync::{Exclusive, exclusive}; use rt::local::Local; +use rt::task::Task; #[cfg(test)] use task::default_task_opts; @@ -576,8 +577,14 @@ pub fn spawn_raw(opts: TaskOpts, f: ~fn()) { fn spawn_raw_newsched(_opts: TaskOpts, f: ~fn()) { use rt::sched::*; + let mut task = None; + do Local::borrow::() |running_task| { + task = Some(~running_task.new_child_without_unwinding()); + } + let mut sched = Local::take::(); - let task = ~Coroutine::new(&mut sched.stack_pool, f); + let task = ~Coroutine::with_task(&mut sched.stack_pool, + task.swap_unwrap(), f); sched.schedule_new_task(task); } -- cgit 1.4.1-3-g733a5