From 1e490813b017f99cb4385fe846d645efe5d62b62 Mon Sep 17 00:00:00 2001 From: Erick Tryzelaar Date: Sun, 4 Aug 2013 14:59:36 -0700 Subject: core: option.map_consume -> option.map_move --- src/libstd/task/spawn.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'src/libstd/task') diff --git a/src/libstd/task/spawn.rs b/src/libstd/task/spawn.rs index 527b20b0e90..7486a78837c 100644 --- a/src/libstd/task/spawn.rs +++ b/src/libstd/task/spawn.rs @@ -500,7 +500,7 @@ impl RuntimeGlue { OldTask(ptr) => rt::rust_task_kill_other(ptr), NewTask(handle) => { let mut handle = handle; - do handle.kill().map_consume |killed_task| { + do handle.kill().map_move |killed_task| { let killed_task = Cell::new(killed_task); do Local::borrow:: |sched| { sched.enqueue_task(killed_task.take()); @@ -682,7 +682,7 @@ fn spawn_raw_newsched(mut opts: TaskOpts, f: ~fn()) { // Child task runs this code. // If child data is 'None', the enlist is vacuously successful. - let enlist_success = do child_data.take().map_consume_default(true) |child_data| { + let enlist_success = do child_data.take().map_move_default(true) |child_data| { let child_data = Cell::new(child_data); // :( do Local::borrow:: |me| { let (child_tg, ancestors, is_main) = child_data.take(); @@ -854,7 +854,7 @@ fn spawn_raw_oldsched(mut opts: TaskOpts, f: ~fn()) { // Even if the below code fails to kick the child off, we must // send Something on the notify channel. - let notifier = notify_chan.map_consume(|c| AutoNotify(c)); + let notifier = notify_chan.map_move(|c| AutoNotify(c)); if enlist_many(OldTask(child), &child_arc, &mut ancestors) { let group = @@mut Taskgroup(child_arc, ancestors, is_main, notifier); -- cgit 1.4.1-3-g733a5 From ae1ed4fd78273502153083f1514a1fcd7c929886 Mon Sep 17 00:00:00 2001 From: Brian Anderson Date: Mon, 5 Aug 2013 13:10:08 -0700 Subject: std: Allow spawners to specify stack size --- src/libstd/rt/local.rs | 11 ++++++----- src/libstd/rt/mod.rs | 7 +++---- src/libstd/rt/sched.rs | 14 +++++++------- src/libstd/rt/task.rs | 37 +++++++++++++++++++++++-------------- src/libstd/rt/test.rs | 15 +++++++-------- src/libstd/task/mod.rs | 12 ++++++++---- src/libstd/task/spawn.rs | 14 +++++++------- 7 files changed, 61 insertions(+), 49 deletions(-) (limited to 'src/libstd/task') diff --git a/src/libstd/rt/local.rs b/src/libstd/rt/local.rs index 131507196b1..7154066e7b7 100644 --- a/src/libstd/rt/local.rs +++ b/src/libstd/rt/local.rs @@ -126,6 +126,7 @@ impl Local for IoFactoryObject { #[cfg(test)] mod test { + use option::None; use unstable::run_in_bare_thread; use rt::test::*; use super::*; @@ -137,7 +138,7 @@ mod test { do run_in_bare_thread { local_ptr::init_tls_key(); let mut sched = ~new_test_uv_sched(); - let task = ~Task::new_root(&mut sched.stack_pool, || {}); + let task = ~Task::new_root(&mut sched.stack_pool, None, || {}); Local::put(task); let task: ~Task = Local::take(); cleanup_task(task); @@ -149,11 +150,11 @@ mod test { do run_in_bare_thread { local_ptr::init_tls_key(); let mut sched = ~new_test_uv_sched(); - let task = ~Task::new_root(&mut sched.stack_pool, || {}); + let task = ~Task::new_root(&mut sched.stack_pool, None, || {}); Local::put(task); let task: ~Task = Local::take(); cleanup_task(task); - let task = ~Task::new_root(&mut sched.stack_pool, || {}); + let task = ~Task::new_root(&mut sched.stack_pool, None, || {}); Local::put(task); let task: ~Task = Local::take(); cleanup_task(task); @@ -166,7 +167,7 @@ mod test { do run_in_bare_thread { local_ptr::init_tls_key(); let mut sched = ~new_test_uv_sched(); - let task = ~Task::new_root(&mut sched.stack_pool, || {}); + let task = ~Task::new_root(&mut sched.stack_pool, None, || {}); Local::put(task); unsafe { @@ -182,7 +183,7 @@ mod test { do run_in_bare_thread { local_ptr::init_tls_key(); let mut sched = ~new_test_uv_sched(); - let task = ~Task::new_root(&mut sched.stack_pool, || {}); + let task = ~Task::new_root(&mut sched.stack_pool, None, || {}); Local::put(task); let res = do Local::borrow:: |_task| { diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs index 5b22eace56f..147c75e5c41 100644 --- a/src/libstd/rt/mod.rs +++ b/src/libstd/rt/mod.rs @@ -331,8 +331,7 @@ fn run_(main: ~fn(), use_main_sched: bool) -> int { // In the case where we do not use a main_thread scheduler we // run the main task in one of our threads. - let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, - main.take()); + let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, None, main.take()); main_task.death.on_exit = Some(on_exit.take()); let main_task_cell = Cell::new(main_task); @@ -352,7 +351,7 @@ fn run_(main: ~fn(), use_main_sched: bool) -> int { let sched_cell = Cell::new(sched); let thread = do Thread::start { let mut sched = sched_cell.take(); - let bootstrap_task = ~do Task::new_root(&mut sched.stack_pool) || { + let bootstrap_task = ~do Task::new_root(&mut sched.stack_pool, None) || { rtdebug!("boostraping a non-primary scheduler"); }; sched.bootstrap(bootstrap_task); @@ -369,7 +368,7 @@ fn run_(main: ~fn(), use_main_sched: bool) -> int { let mut main_sched = main_sched.unwrap(); let home = Sched(main_sched.make_handle()); - let mut main_task = ~Task::new_root_homed(&mut main_sched.stack_pool, + let mut main_task = ~Task::new_root_homed(&mut main_sched.stack_pool, None, home, main.take()); main_task.death.on_exit = Some(on_exit.take()); rtdebug!("boostrapping main_task"); diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs index c2c12c6e3c0..990e1a4a3de 100644 --- a/src/libstd/rt/sched.rs +++ b/src/libstd/rt/sched.rs @@ -833,7 +833,7 @@ mod test { let mut sched = ~new_test_uv_sched(); let sched_handle = sched.make_handle(); - let mut task = ~do Task::new_root_homed(&mut sched.stack_pool, + let mut task = ~do Task::new_root_homed(&mut sched.stack_pool, None, Sched(sched_handle)) { unsafe { *task_ran_ptr = true }; assert!(Task::on_appropriate_sched()); @@ -893,21 +893,21 @@ mod test { // 3) task not homed, sched requeues // 4) task not home, send home - let task1 = ~do Task::new_root_homed(&mut special_sched.stack_pool, + let task1 = ~do Task::new_root_homed(&mut special_sched.stack_pool, None, Sched(t1_handle)) || { rtassert!(Task::on_appropriate_sched()); }; rtdebug!("task1 id: **%u**", borrow::to_uint(task1)); - let task2 = ~do Task::new_root(&mut normal_sched.stack_pool) { + let task2 = ~do Task::new_root(&mut normal_sched.stack_pool, None) { rtassert!(Task::on_appropriate_sched()); }; - let task3 = ~do Task::new_root(&mut normal_sched.stack_pool) { + let task3 = ~do Task::new_root(&mut normal_sched.stack_pool, None) { rtassert!(Task::on_appropriate_sched()); }; - let task4 = ~do Task::new_root_homed(&mut special_sched.stack_pool, + let task4 = ~do Task::new_root_homed(&mut special_sched.stack_pool, None, Sched(t4_handle)) { rtassert!(Task::on_appropriate_sched()); }; @@ -923,7 +923,7 @@ mod test { let port = Cell::new(port); let chan = Cell::new(chan); - let normal_task = ~do Task::new_root(&mut normal_sched.stack_pool) { + let normal_task = ~do Task::new_root(&mut normal_sched.stack_pool, None) { rtdebug!("*about to submit task2*"); Scheduler::run_task(task2.take()); rtdebug!("*about to submit task4*"); @@ -938,7 +938,7 @@ mod test { rtdebug!("normal task: %u", borrow::to_uint(normal_task)); - let special_task = ~do Task::new_root(&mut special_sched.stack_pool) { + let special_task = ~do Task::new_root(&mut special_sched.stack_pool, None) { rtdebug!("*about to submit task1*"); Scheduler::run_task(task1.take()); rtdebug!("*about to submit task3*"); diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs index aa6d51a480b..364439a4526 100644 --- a/src/libstd/rt/task.rs +++ b/src/libstd/rt/task.rs @@ -86,12 +86,13 @@ impl Task { // A helper to build a new task using the dynamically found // scheduler and task. Only works in GreenTask context. - pub fn build_homed_child(f: ~fn(), home: SchedHome) -> ~Task { + pub fn build_homed_child(stack_size: Option, f: ~fn(), home: SchedHome) -> ~Task { let f = Cell::new(f); let home = Cell::new(home); do Local::borrow:: |running_task| { let mut sched = running_task.sched.take_unwrap(); let new_task = ~running_task.new_child_homed(&mut sched.stack_pool, + stack_size, home.take(), f.take()); running_task.sched = Some(sched); @@ -99,25 +100,26 @@ impl Task { } } - pub fn build_child(f: ~fn()) -> ~Task { - Task::build_homed_child(f, AnySched) + pub fn build_child(stack_size: Option, f: ~fn()) -> ~Task { + Task::build_homed_child(stack_size, f, AnySched) } - pub fn build_homed_root(f: ~fn(), home: SchedHome) -> ~Task { + pub fn build_homed_root(stack_size: Option, f: ~fn(), home: SchedHome) -> ~Task { let f = Cell::new(f); let home = Cell::new(home); do Local::borrow:: |running_task| { let mut sched = running_task.sched.take_unwrap(); let new_task = ~Task::new_root_homed(&mut sched.stack_pool, - home.take(), - f.take()); + stack_size, + home.take(), + f.take()); running_task.sched = Some(sched); new_task } } - pub fn build_root(f: ~fn()) -> ~Task { - Task::build_homed_root(f, AnySched) + pub fn build_root(stack_size: Option, f: ~fn()) -> ~Task { + Task::build_homed_root(stack_size, f, AnySched) } pub fn new_sched_task() -> Task { @@ -138,17 +140,20 @@ impl Task { } pub fn new_root(stack_pool: &mut StackPool, + stack_size: Option, start: ~fn()) -> Task { - Task::new_root_homed(stack_pool, AnySched, start) + Task::new_root_homed(stack_pool, stack_size, AnySched, start) } pub fn new_child(&mut self, stack_pool: &mut StackPool, + stack_size: Option, start: ~fn()) -> Task { - self.new_child_homed(stack_pool, AnySched, start) + self.new_child_homed(stack_pool, stack_size, AnySched, start) } pub fn new_root_homed(stack_pool: &mut StackPool, + stack_size: Option, home: SchedHome, start: ~fn()) -> Task { Task { @@ -161,7 +166,7 @@ impl Task { death: Death::new(), destroyed: false, name: None, - coroutine: Some(Coroutine::new(stack_pool, start)), + coroutine: Some(Coroutine::new(stack_pool, stack_size, start)), sched: None, task_type: GreenTask(Some(~home)) } @@ -169,6 +174,7 @@ impl Task { pub fn new_child_homed(&mut self, stack_pool: &mut StackPool, + stack_size: Option, home: SchedHome, start: ~fn()) -> Task { Task { @@ -182,7 +188,7 @@ impl Task { death: self.death.new_child(), destroyed: false, name: None, - coroutine: Some(Coroutine::new(stack_pool, start)), + coroutine: Some(Coroutine::new(stack_pool, stack_size, start)), sched: None, task_type: GreenTask(Some(~home)) } @@ -326,8 +332,11 @@ impl Drop for Task { impl Coroutine { - pub fn new(stack_pool: &mut StackPool, start: ~fn()) -> Coroutine { - let stack_size = env::min_stack(); + pub fn new(stack_pool: &mut StackPool, stack_size: Option, start: ~fn()) -> Coroutine { + let stack_size = match stack_size { + Some(size) => size, + None => env::min_stack() + }; let start = Coroutine::build_start_wrapper(start); let mut stack = stack_pool.take_segment(stack_size); let initial_context = Context::new(start, &mut stack); diff --git a/src/libstd/rt/test.rs b/src/libstd/rt/test.rs index 8b5215ae969..792ea5eb33f 100644 --- a/src/libstd/rt/test.rs +++ b/src/libstd/rt/test.rs @@ -57,7 +57,7 @@ pub fn run_in_newsched_task_core(f: ~fn()) { exit_handle.take().send(Shutdown); rtassert!(exit_status); }; - let mut task = ~Task::new_root(&mut sched.stack_pool, f); + let mut task = ~Task::new_root(&mut sched.stack_pool, None, f); task.death.on_exit = Some(on_exit); sched.bootstrap(task); @@ -190,8 +190,7 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { rtassert!(exit_status); }; - let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, - f.take()); + let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, None, f.take()); main_task.death.on_exit = Some(on_exit); let mut threads = ~[]; @@ -209,7 +208,7 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { while !scheds.is_empty() { let mut sched = scheds.pop(); - let bootstrap_task = ~do Task::new_root(&mut sched.stack_pool) || { + let bootstrap_task = ~do Task::new_root(&mut sched.stack_pool, None) || { rtdebug!("bootstrapping non-primary scheduler"); }; let bootstrap_task_cell = Cell::new(bootstrap_task); @@ -232,12 +231,12 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { /// Test tasks will abort on failure instead of unwinding pub fn spawntask(f: ~fn()) { - Scheduler::run_task(Task::build_child(f)); + Scheduler::run_task(Task::build_child(None, f)); } /// Create a new task and run it right now. Aborts on failure pub fn spawntask_later(f: ~fn()) { - Scheduler::run_task_later(Task::build_child(f)); + Scheduler::run_task_later(Task::build_child(None, f)); } pub fn spawntask_random(f: ~fn()) { @@ -259,7 +258,7 @@ pub fn spawntask_try(f: ~fn()) -> Result<(),()> { let chan = Cell::new(chan); let on_exit: ~fn(bool) = |exit_status| chan.take().send(exit_status); - let mut new_task = Task::build_root(f); + let mut new_task = Task::build_root(None, f); new_task.death.on_exit = Some(on_exit); Scheduler::run_task(new_task); @@ -285,7 +284,7 @@ pub fn spawntask_thread(f: ~fn()) -> Thread { pub fn with_test_task(blk: ~fn(~Task) -> ~Task) { do run_in_bare_thread { let mut sched = ~new_test_uv_sched(); - let task = blk(~Task::new_root(&mut sched.stack_pool, ||{})); + let task = blk(~Task::new_root(&mut sched.stack_pool, None, ||{})); cleanup_task(task); } } diff --git a/src/libstd/task/mod.rs b/src/libstd/task/mod.rs index 225a4b8cfd2..4b5543b8186 100644 --- a/src/libstd/task/mod.rs +++ b/src/libstd/task/mod.rs @@ -142,7 +142,8 @@ pub struct TaskOpts { indestructible: bool, notify_chan: Option>, name: Option<~str>, - sched: SchedOpts + sched: SchedOpts, + stack_size: Option } /** @@ -197,7 +198,8 @@ impl TaskBuilder { indestructible: self.opts.indestructible, notify_chan: notify_chan, name: name, - sched: self.opts.sched + sched: self.opts.sched, + stack_size: self.opts.stack_size }, gen_body: gen_body, can_not_copy: None, @@ -351,7 +353,8 @@ impl TaskBuilder { indestructible: x.opts.indestructible, notify_chan: notify_chan, name: name, - sched: x.opts.sched + sched: x.opts.sched, + stack_size: x.opts.stack_size }; let f = match gen_body { Some(gen) => { @@ -422,7 +425,8 @@ pub fn default_task_opts() -> TaskOpts { name: None, sched: SchedOpts { mode: DefaultScheduler, - } + }, + stack_size: None } } diff --git a/src/libstd/task/spawn.rs b/src/libstd/task/spawn.rs index 7486a78837c..2d0a2d98e9f 100644 --- a/src/libstd/task/spawn.rs +++ b/src/libstd/task/spawn.rs @@ -713,9 +713,9 @@ fn spawn_raw_newsched(mut opts: TaskOpts, f: ~fn()) { let mut task = unsafe { if opts.sched.mode != SingleThreaded { if opts.watched { - Task::build_child(child_wrapper) + Task::build_child(opts.stack_size, child_wrapper) } else { - Task::build_root(child_wrapper) + Task::build_root(opts.stack_size, child_wrapper) } } else { // Creating a 1:1 task:thread ... @@ -736,16 +736,16 @@ fn spawn_raw_newsched(mut opts: TaskOpts, f: ~fn()) { // Pin the new task to the new scheduler let new_task = if opts.watched { - Task::build_homed_child(child_wrapper, Sched(new_sched_handle)) + Task::build_homed_child(opts.stack_size, child_wrapper, Sched(new_sched_handle)) } else { - Task::build_homed_root(child_wrapper, Sched(new_sched_handle)) + 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() { + 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(); @@ -762,8 +762,8 @@ fn spawn_raw_newsched(mut opts: TaskOpts, f: ~fn()) { 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) || { - rtdebug!("bootstrapping a 1:1 scheduler"); + let bootstrap_task = ~do Task::new_root(&mut new_sched.stack_pool, None) || { + rtdebug!("boostrapping a 1:1 scheduler"); }; new_sched.bootstrap(bootstrap_task); -- cgit 1.4.1-3-g733a5 From ce95b01014391f29a655d165d9e6d31449ceb835 Mon Sep 17 00:00:00 2001 From: Brian Anderson Date: Tue, 6 Aug 2013 14:32:30 -0700 Subject: Disable linked failure tests The implementation currently contains a race that leads to segfaults. --- doc/tutorial-tasks.md | 16 ++++++++-------- src/libextra/arc.rs | 1 + src/libextra/sync.rs | 2 ++ src/libstd/rt/kill.rs | 6 ++++++ src/libstd/task/mod.rs | 19 +++++++++++++++++++ src/test/run-fail/extern-fail.rs | 1 + src/test/run-fail/linked-failure.rs | 1 + src/test/run-fail/linked-failure2.rs | 1 + src/test/run-fail/linked-failure3.rs | 1 + src/test/run-fail/linked-failure4.rs | 1 + src/test/run-fail/spawnfail.rs | 1 + src/test/run-fail/task-comm-recv-block.rs | 1 + src/test/run-pass/issue-3168.rs | 1 + src/test/run-pass/lots-a-fail.rs | 1 + src/test/run-pass/send-iloop.rs | 1 + src/test/run-pass/task-killjoin-rsrc.rs | 1 + src/test/run-pass/task-killjoin.rs | 1 + 17 files changed, 48 insertions(+), 8 deletions(-) (limited to 'src/libstd/task') diff --git a/doc/tutorial-tasks.md b/doc/tutorial-tasks.md index d9e4b9b399d..d190c332e66 100644 --- a/doc/tutorial-tasks.md +++ b/doc/tutorial-tasks.md @@ -424,7 +424,7 @@ there is no way to "catch" the exception. All tasks are, by default, _linked_ to each other. That means that the fates of all tasks are intertwined: if one fails, so do all the others. -~~~ +~~~{.xfail-test .linked-failure} # use std::task::spawn; # use std::task; # fn do_some_work() { loop { task::yield() } } @@ -447,7 +447,7 @@ pattern-match on a result to check whether it's an `Ok` result with an `int` field (representing a successful result) or an `Err` result (representing termination with an error). -~~~ +~~~{.xfail-test .linked-failure} # use std::task; # fn some_condition() -> bool { false } # fn calculate_result() -> int { 0 } @@ -490,7 +490,7 @@ proceed). Hence, you will need different _linked failure modes_. By default, task failure is _bidirectionally linked_, which means that if either task fails, it kills the other one. -~~~ +~~~{.xfail-test .linked-failure} # use std::task; # use std::comm::oneshot; # fn sleep_forever() { loop { let (p, c) = oneshot::<()>(); p.recv(); } } @@ -512,7 +512,7 @@ function `task::try`, which we saw previously, uses `spawn_supervised` internally, with additional logic to wait for the child task to finish before returning. Hence: -~~~ +~~~{.xfail-test .linked-failure} # use std::comm::{stream, Chan, Port}; # use std::comm::oneshot; # use std::task::{spawn, try}; @@ -543,7 +543,7 @@ also fail. Supervised task failure propagates across multiple generations even if an intermediate generation has already exited: -~~~ +~~~{.xfail-test .linked-failure} # use std::task; # use std::comm::oneshot; # fn sleep_forever() { loop { let (p, c) = oneshot::<()>(); p.recv(); } } @@ -563,7 +563,7 @@ fail!(); // Will kill grandchild even if child has already exited Finally, tasks can be configured to not propagate failure to each other at all, using `task::spawn_unlinked` for _isolated failure_. -~~~ +~~~{.xfail-test .linked-failure} # use std::task; # fn random() -> uint { 100 } # fn sleep_for(i: uint) { for _ in range(0, i) { task::yield() } } @@ -591,7 +591,7 @@ that repeatedly receives a `uint` message, converts it to a string, and sends the string in response. The child terminates when it receives `0`. Here is the function that implements the child task: -~~~~ +~~~{.xfail-test .linked-failure} # use extra::comm::DuplexStream; # use std::uint; fn stringifier(channel: &DuplexStream<~str, uint>) { @@ -614,7 +614,7 @@ response itself is simply the stringified version of the received value, Here is the code for the parent task: -~~~~ +~~~{.xfail-test .linked-failure} # use std::task::spawn; # use std::uint; # use extra::comm::DuplexStream; diff --git a/src/libextra/arc.rs b/src/libextra/arc.rs index cb4468f48ec..17f4cbbd152 100644 --- a/src/libextra/arc.rs +++ b/src/libextra/arc.rs @@ -611,6 +611,7 @@ mod tests { } } } + #[test] #[should_fail] #[ignore(cfg(windows))] fn test_arc_condvar_poison() { unsafe { diff --git a/src/libextra/sync.rs b/src/libextra/sync.rs index 63e371899a9..4172c715adb 100644 --- a/src/libextra/sync.rs +++ b/src/libextra/sync.rs @@ -935,6 +935,7 @@ mod tests { // child task must have finished by the time try returns do m.lock { } } + #[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_mutex_killed_cond() { // Getting killed during cond wait must not corrupt the mutex while @@ -961,6 +962,7 @@ mod tests { assert!(!woken); } } + #[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_mutex_killed_broadcast() { use std::unstable::finally::Finally; diff --git a/src/libstd/rt/kill.rs b/src/libstd/rt/kill.rs index 789c7531eca..fbc9d1d2445 100644 --- a/src/libstd/rt/kill.rs +++ b/src/libstd/rt/kill.rs @@ -614,6 +614,7 @@ mod test { // Test cases don't care about the spare killed flag. fn make_kill_handle() -> KillHandle { let (h,_) = KillHandle::new(); h } + #[ignore(reason = "linked failure")] #[test] fn no_tombstone_success() { do run_in_newsched_task { @@ -819,6 +820,7 @@ mod test { } } + #[ignore(reason = "linked failure")] #[test] fn block_and_get_killed() { do with_test_task |mut task| { @@ -830,6 +832,7 @@ mod test { } } + #[ignore(reason = "linked failure")] #[test] fn block_already_killed() { do with_test_task |mut task| { @@ -839,6 +842,7 @@ mod test { } } + #[ignore(reason = "linked failure")] #[test] fn block_unkillably_and_get_killed() { do with_test_task |mut task| { @@ -856,6 +860,7 @@ mod test { } } + #[ignore(reason = "linked failure")] #[test] fn block_on_pipe() { // Tests the "killable" path of casting to/from uint. @@ -869,6 +874,7 @@ mod test { } } + #[ignore(reason = "linked failure")] #[test] fn block_unkillably_on_pipe() { // Tests the "indestructible" path of casting to/from uint. diff --git a/src/libstd/task/mod.rs b/src/libstd/task/mod.rs index 4b5543b8186..2e0c9c1d1ad 100644 --- a/src/libstd/task/mod.rs +++ b/src/libstd/task/mod.rs @@ -659,6 +659,7 @@ pub unsafe fn rekillable(f: &fn() -> U) -> U { } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_kill_unkillable_task() { use rt::test::*; @@ -679,6 +680,7 @@ fn test_kill_unkillable_task() { } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_kill_rekillable_task() { use rt::test::*; @@ -720,6 +722,7 @@ fn test_cant_dup_task_builder() { #[cfg(test)] fn block_forever() { let (po, _ch) = stream::<()>(); po.recv(); } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_unlinked_unsup_no_fail_down() { // grandchild sends on a port use rt::test::run_in_newsched_task; @@ -738,6 +741,7 @@ fn test_spawn_unlinked_unsup_no_fail_down() { // grandchild sends on a port po.recv(); } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_unlinked_unsup_no_fail_up() { // child unlinked fails use rt::test::run_in_newsched_task; @@ -745,6 +749,7 @@ fn test_spawn_unlinked_unsup_no_fail_up() { // child unlinked fails do spawn_unlinked { fail!(); } } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_unlinked_sup_no_fail_up() { // child unlinked fails use rt::test::run_in_newsched_task; @@ -754,6 +759,7 @@ fn test_spawn_unlinked_sup_no_fail_up() { // child unlinked fails do 16.times { task::yield(); } } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_unlinked_sup_fail_down() { use rt::test::run_in_newsched_task; @@ -766,6 +772,7 @@ fn test_spawn_unlinked_sup_fail_down() { } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_linked_sup_fail_up() { // child fails; parent fails use rt::test::run_in_newsched_task; @@ -786,6 +793,7 @@ fn test_spawn_linked_sup_fail_up() { // child fails; parent fails assert!(result.is_err()); } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_linked_sup_fail_down() { // parent fails; child fails use rt::test::run_in_newsched_task; @@ -802,6 +810,7 @@ fn test_spawn_linked_sup_fail_down() { // parent fails; child fails assert!(result.is_err()); } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_linked_unsup_fail_up() { // child fails; parent fails use rt::test::run_in_newsched_task; @@ -814,6 +823,7 @@ fn test_spawn_linked_unsup_fail_up() { // child fails; parent fails assert!(result.is_err()); } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_linked_unsup_fail_down() { // parent fails; child fails use rt::test::run_in_newsched_task; @@ -826,6 +836,7 @@ fn test_spawn_linked_unsup_fail_down() { // parent fails; child fails assert!(result.is_err()); } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_linked_unsup_default_opts() { // parent fails; child fails use rt::test::run_in_newsched_task; @@ -844,6 +855,7 @@ fn test_spawn_linked_unsup_default_opts() { // parent fails; child fails // A couple bonus linked failure tests - testing for failure propagation even // when the middle task exits successfully early before kill signals are sent. +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_failure_propagate_grandchild() { use rt::test::run_in_newsched_task; @@ -860,6 +872,7 @@ fn test_spawn_failure_propagate_grandchild() { } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_failure_propagate_secondborn() { use rt::test::run_in_newsched_task; @@ -876,6 +889,7 @@ fn test_spawn_failure_propagate_secondborn() { } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_failure_propagate_nephew_or_niece() { use rt::test::run_in_newsched_task; @@ -892,6 +906,7 @@ fn test_spawn_failure_propagate_nephew_or_niece() { } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_linked_sup_propagate_sibling() { use rt::test::run_in_newsched_task; @@ -1195,6 +1210,7 @@ fn test_avoid_copying_the_body_unlinked() { } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] #[should_fail] @@ -1230,6 +1246,7 @@ fn test_unkillable() { po.recv(); } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] #[should_fail] @@ -1296,6 +1313,7 @@ fn test_simple_newsched_spawn() { } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_spawn_watched() { use rt::test::run_in_newsched_task; @@ -1318,6 +1336,7 @@ fn test_spawn_watched() { } } +#[ignore(reason = "linked failure")] #[test] #[ignore(cfg(windows))] fn test_indestructible() { use rt::test::run_in_newsched_task; diff --git a/src/test/run-fail/extern-fail.rs b/src/test/run-fail/extern-fail.rs index a281e986364..a65db3ee515 100644 --- a/src/test/run-fail/extern-fail.rs +++ b/src/test/run-fail/extern-fail.rs @@ -8,6 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// xfail-test linked failure // error-pattern:explicit failure // Testing that runtime failure doesn't cause callbacks to abort abnormally. // Instead the failure will be delivered after the callbacks return. diff --git a/src/test/run-fail/linked-failure.rs b/src/test/run-fail/linked-failure.rs index 41a9d7ddcea..52dfb8aef13 100644 --- a/src/test/run-fail/linked-failure.rs +++ b/src/test/run-fail/linked-failure.rs @@ -10,6 +10,7 @@ // except according to those terms. +// xfail-test linked failure // error-pattern:1 == 2 extern mod extra; diff --git a/src/test/run-fail/linked-failure2.rs b/src/test/run-fail/linked-failure2.rs index 0269e395986..d4049f6753e 100644 --- a/src/test/run-fail/linked-failure2.rs +++ b/src/test/run-fail/linked-failure2.rs @@ -10,6 +10,7 @@ // except according to those terms. +// xfail-test linked failure // error-pattern:fail use std::comm; diff --git a/src/test/run-fail/linked-failure3.rs b/src/test/run-fail/linked-failure3.rs index 1203f74322f..f40eae20bc0 100644 --- a/src/test/run-fail/linked-failure3.rs +++ b/src/test/run-fail/linked-failure3.rs @@ -10,6 +10,7 @@ // except according to those terms. +// xfail-test linked failure // error-pattern:fail use std::comm; diff --git a/src/test/run-fail/linked-failure4.rs b/src/test/run-fail/linked-failure4.rs index 766b43f211f..94e41f1ae68 100644 --- a/src/test/run-fail/linked-failure4.rs +++ b/src/test/run-fail/linked-failure4.rs @@ -9,6 +9,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// xfail-test linked failure // error-pattern:1 == 2 use std::comm; diff --git a/src/test/run-fail/spawnfail.rs b/src/test/run-fail/spawnfail.rs index de085a6f3ad..12dab8e25b7 100644 --- a/src/test/run-fail/spawnfail.rs +++ b/src/test/run-fail/spawnfail.rs @@ -8,6 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// xfail-test linked failure // xfail-win32 // error-pattern:explicit extern mod extra; diff --git a/src/test/run-fail/task-comm-recv-block.rs b/src/test/run-fail/task-comm-recv-block.rs index 8302b96ca3e..bd51ce38ec0 100644 --- a/src/test/run-fail/task-comm-recv-block.rs +++ b/src/test/run-fail/task-comm-recv-block.rs @@ -8,6 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// xfail-test linked failure // error-pattern:goodfail use std::comm; diff --git a/src/test/run-pass/issue-3168.rs b/src/test/run-pass/issue-3168.rs index 609849bffb4..f4e2a9f36a0 100644 --- a/src/test/run-pass/issue-3168.rs +++ b/src/test/run-pass/issue-3168.rs @@ -8,6 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// xfail-test linked failure // xfail-fast // xfail-win32 #7999 diff --git a/src/test/run-pass/lots-a-fail.rs b/src/test/run-pass/lots-a-fail.rs index cec0a7a756c..13296131236 100644 --- a/src/test/run-pass/lots-a-fail.rs +++ b/src/test/run-pass/lots-a-fail.rs @@ -8,6 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// xfail-test linked failure // xfail-win32 leaks extern mod extra; diff --git a/src/test/run-pass/send-iloop.rs b/src/test/run-pass/send-iloop.rs index e27f35d1851..a647e5849a8 100644 --- a/src/test/run-pass/send-iloop.rs +++ b/src/test/run-pass/send-iloop.rs @@ -8,6 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// xfail-test linked failure // xfail-win32 extern mod extra; diff --git a/src/test/run-pass/task-killjoin-rsrc.rs b/src/test/run-pass/task-killjoin-rsrc.rs index c811e548f3f..b8a1aa433a3 100644 --- a/src/test/run-pass/task-killjoin-rsrc.rs +++ b/src/test/run-pass/task-killjoin-rsrc.rs @@ -8,6 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// xfail-test linked failure // xfail-win32 // A port of task-killjoin to use a class with a dtor to manage diff --git a/src/test/run-pass/task-killjoin.rs b/src/test/run-pass/task-killjoin.rs index c94e00251d2..5382ac77671 100644 --- a/src/test/run-pass/task-killjoin.rs +++ b/src/test/run-pass/task-killjoin.rs @@ -8,6 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// xfail-test linked failure // xfail-win32 // Create a task that is supervised by another task, join the supervised task -- cgit 1.4.1-3-g733a5 From af2e03998d4d06f2781ca72ec005f6913148f8bb Mon Sep 17 00:00:00 2001 From: toddaaro Date: Mon, 5 Aug 2013 13:06:24 -0700 Subject: Enabled workstealing in the scheduler. Previously we had one global work queue shared by each scheduler. Now there is a separate work queue for each scheduler, and work is "stolen" from other queues when it is exhausted locally. --- src/libstd/rt/comm.rs | 7 +- src/libstd/rt/mod.rs | 32 +++++-- src/libstd/rt/sched.rs | 159 ++++++++++++++++++++++--------- src/libstd/rt/select.rs | 2 + src/libstd/rt/test.rs | 18 +++- src/libstd/task/spawn.rs | 9 +- src/test/bench/rt-messaging-ping-pong.rs | 82 ++++++++++++++++ src/test/bench/rt-parfib.rs | 49 ++++++++++ src/test/bench/rt-spawn-rate.rs | 33 +++++++ 9 files changed, 328 insertions(+), 63 deletions(-) create mode 100644 src/test/bench/rt-messaging-ping-pong.rs create mode 100644 src/test/bench/rt-parfib.rs create mode 100644 src/test/bench/rt-spawn-rate.rs (limited to 'src/libstd/task') diff --git a/src/libstd/rt/comm.rs b/src/libstd/rt/comm.rs index 0cf223f3029..33b4b307af8 100644 --- a/src/libstd/rt/comm.rs +++ b/src/libstd/rt/comm.rs @@ -225,9 +225,10 @@ impl Select for PortOne { fn optimistic_check(&mut self) -> bool { // The optimistic check is never necessary for correctness. For testing // purposes, making it randomly return false simulates a racing sender. - use rand::{Rand, rng}; - let mut rng = rng(); - let actually_check = Rand::rand(&mut rng); + use rand::{Rand}; + let actually_check = do Local::borrow:: |sched| { + Rand::rand(&mut sched.rng) + }; if actually_check { unsafe { (*self.packet()).state.load(Acquire) == STATE_ONE } } else { diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs index 147c75e5c41..01a52892f63 100644 --- a/src/libstd/rt/mod.rs +++ b/src/libstd/rt/mod.rs @@ -63,8 +63,7 @@ Several modules in `core` are clients of `rt`: use cell::Cell; use clone::Clone; use container::Container; -use iter::Times; -use iterator::{Iterator, IteratorUtil}; +use iterator::{Iterator, IteratorUtil, range}; use option::{Some, None}; use ptr::RawPtr; use rt::local::Local; @@ -247,11 +246,16 @@ fn run_(main: ~fn(), use_main_sched: bool) -> int { let main = Cell::new(main); - // The shared list of sleeping schedulers. Schedulers wake each other - // occassionally to do new work. + // The shared list of sleeping schedulers. let sleepers = SleeperList::new(); - // The shared work queue. Temporary until work stealing is implemented. - let work_queue = WorkQueue::new(); + + // Create a work queue for each scheduler, ntimes. Create an extra + // for the main thread if that flag is set. We won't steal from it. + let mut work_queues = ~[]; + for _ in range(0u, nscheds) { + let work_queue: WorkQueue<~Task> = WorkQueue::new(); + work_queues.push(work_queue); + } // The schedulers. let mut scheds = ~[]; @@ -259,12 +263,15 @@ fn run_(main: ~fn(), use_main_sched: bool) -> int { // sent the Shutdown message to terminate the schedulers. let mut handles = ~[]; - do nscheds.times { + for i in range(0u, nscheds) { rtdebug!("inserting a regular scheduler"); // Every scheduler is driven by an I/O event loop. let loop_ = ~UvEventLoop::new(); - let mut sched = ~Scheduler::new(loop_, work_queue.clone(), sleepers.clone()); + let mut sched = ~Scheduler::new(loop_, + work_queues[i].clone(), + work_queues.clone(), + sleepers.clone()); let handle = sched.make_handle(); scheds.push(sched); @@ -280,9 +287,14 @@ fn run_(main: ~fn(), use_main_sched: bool) -> int { let friend_handle = friend_sched.make_handle(); scheds.push(friend_sched); + // This scheduler needs a queue that isn't part of the stealee + // set. + let work_queue = WorkQueue::new(); + let main_loop = ~UvEventLoop::new(); let mut main_sched = ~Scheduler::new_special(main_loop, - work_queue.clone(), + work_queue, + work_queues.clone(), sleepers.clone(), false, Some(friend_handle)); @@ -371,7 +383,7 @@ fn run_(main: ~fn(), use_main_sched: bool) -> int { let mut main_task = ~Task::new_root_homed(&mut main_sched.stack_pool, None, home, main.take()); main_task.death.on_exit = Some(on_exit.take()); - rtdebug!("boostrapping main_task"); + rtdebug!("bootstrapping main_task"); main_sched.bootstrap(main_task); } diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs index 990e1a4a3de..ce4e64c47d2 100644 --- a/src/libstd/rt/sched.rs +++ b/src/libstd/rt/sched.rs @@ -13,7 +13,6 @@ use option::{Option, Some, None}; use cast::{transmute, transmute_mut_region, transmute_mut_unsafe}; use clone::Clone; use unstable::raw; - use super::sleeper_list::SleeperList; use super::work_queue::WorkQueue; use super::stack::{StackPool}; @@ -28,6 +27,9 @@ use rt::rtio::RemoteCallback; use rt::metrics::SchedMetrics; use borrow::{to_uint}; use cell::Cell; +use rand::{XorShiftRng, RngUtil}; +use iterator::{range}; +use vec::{OwnedVector}; /// The Scheduler is responsible for coordinating execution of Coroutines /// on a single thread. When the scheduler is running it is owned by @@ -37,9 +39,11 @@ use cell::Cell; /// XXX: This creates too many callbacks to run_sched_once, resulting /// in too much allocation and too many events. pub struct Scheduler { - /// A queue of available work. Under a work-stealing policy there - /// is one per Scheduler. - work_queue: WorkQueue<~Task>, + /// There are N work queues, one per scheduler. + priv work_queue: WorkQueue<~Task>, + /// Work queues for the other schedulers. These are created by + /// cloning the core work queues. + work_queues: ~[WorkQueue<~Task>], /// The queue of incoming messages from other schedulers. /// These are enqueued by SchedHandles after which a remote callback /// is triggered to handle the message. @@ -70,7 +74,10 @@ pub struct Scheduler { run_anything: bool, /// If the scheduler shouldn't run some tasks, a friend to send /// them to. - friend_handle: Option + friend_handle: Option, + /// A fast XorShift rng for scheduler use + rng: XorShiftRng + } pub struct SchedHandle { @@ -97,10 +104,13 @@ impl Scheduler { pub fn new(event_loop: ~EventLoopObject, work_queue: WorkQueue<~Task>, + work_queues: ~[WorkQueue<~Task>], sleeper_list: SleeperList) -> Scheduler { - Scheduler::new_special(event_loop, work_queue, sleeper_list, true, None) + Scheduler::new_special(event_loop, work_queue, + work_queues, + sleeper_list, true, None) } @@ -108,6 +118,7 @@ impl Scheduler { // task field is None. pub fn new_special(event_loop: ~EventLoopObject, work_queue: WorkQueue<~Task>, + work_queues: ~[WorkQueue<~Task>], sleeper_list: SleeperList, run_anything: bool, friend: Option) @@ -120,12 +131,14 @@ impl Scheduler { no_sleep: false, event_loop: event_loop, work_queue: work_queue, + work_queues: work_queues, stack_pool: StackPool::new(), sched_task: None, cleanup_job: None, metrics: SchedMetrics::new(), run_anything: run_anything, - friend_handle: friend + friend_handle: friend, + rng: XorShiftRng::new() } } @@ -248,7 +261,7 @@ impl Scheduler { // Second activity is to try resuming a task from the queue. - let result = sched.resume_task_from_queue(); + let result = sched.do_work(); let mut sched = match result { Some(sched) => { // Failed to dequeue a task, so we return. @@ -415,47 +428,98 @@ impl Scheduler { } } - // Resume a task from the queue - but also take into account that - // it might not belong here. + // Workstealing: In this iteration of the runtime each scheduler + // thread has a distinct work queue. When no work is available + // locally, make a few attempts to steal work from the queues of + // other scheduler threads. If a few steals fail we end up in the + // old "no work" path which is fine. + + // First step in the process is to find a task. This function does + // that by first checking the local queue, and if there is no work + // there, trying to steal from the remote work queues. + fn find_work(&mut self) -> Option<~Task> { + rtdebug!("scheduler looking for work"); + match self.work_queue.pop() { + Some(task) => { + rtdebug!("found a task locally"); + return Some(task) + } + None => { + // Our naive stealing, try kinda hard. + rtdebug!("scheduler trying to steal"); + let _len = self.work_queues.len(); + return self.try_steals(2); + } + } + } + + // With no backoff try stealing n times from the queues the + // scheduler knows about. This naive implementation can steal from + // our own queue or from other special schedulers. + fn try_steals(&mut self, n: uint) -> Option<~Task> { + for _ in range(0, n) { + let index = self.rng.gen_uint_range(0, self.work_queues.len()); + let work_queues = &mut self.work_queues; + match work_queues[index].steal() { + Some(task) => { + rtdebug!("found task by stealing"); return Some(task) + } + None => () + } + }; + rtdebug!("giving up on stealing"); + return None; + } - // If we perform a scheduler action we give away the scheduler ~ - // pointer, if it is still available we return it. + // Given a task, execute it correctly. + fn process_task(~self, task: ~Task) -> Option<~Scheduler> { + let mut this = self; + let mut task = task; - fn resume_task_from_queue(~self) -> Option<~Scheduler> { + rtdebug!("processing a task"); + let home = task.take_unwrap_home(); + match home { + Sched(home_handle) => { + if home_handle.sched_id != this.sched_id() { + rtdebug!("sending task home"); + task.give_home(Sched(home_handle)); + Scheduler::send_task_home(task); + return Some(this); + } else { + rtdebug!("running task here"); + task.give_home(Sched(home_handle)); + this.resume_task_immediately(task); + return None; + } + } + AnySched if this.run_anything => { + rtdebug!("running anysched task here"); + task.give_home(AnySched); + this.resume_task_immediately(task); + return None; + } + AnySched => { + rtdebug!("sending task to friend"); + task.give_home(AnySched); + this.send_to_friend(task); + return Some(this); + } + } + } + + // Bundle the helpers together. + fn do_work(~self) -> Option<~Scheduler> { let mut this = self; - match this.work_queue.pop() { + rtdebug!("scheduler calling do work"); + match this.find_work() { Some(task) => { - let mut task = task; - let home = task.take_unwrap_home(); - match home { - Sched(home_handle) => { - if home_handle.sched_id != this.sched_id() { - task.give_home(Sched(home_handle)); - Scheduler::send_task_home(task); - return Some(this); - } else { - this.event_loop.callback(Scheduler::run_sched_once); - task.give_home(Sched(home_handle)); - this.resume_task_immediately(task); - return None; - } - } - AnySched if this.run_anything => { - this.event_loop.callback(Scheduler::run_sched_once); - task.give_home(AnySched); - this.resume_task_immediately(task); - return None; - } - AnySched => { - task.give_home(AnySched); - this.send_to_friend(task); - return Some(this); - } - } + rtdebug!("found some work! processing the task"); + return this.process_task(task); } None => { + rtdebug!("no work was found, returning the scheduler struct"); return Some(this); } } @@ -711,7 +775,6 @@ impl Scheduler { GiveTask(task, f) => f.to_fn()(self, task) } } - } // The cases for the below function. @@ -745,6 +808,8 @@ impl ClosureConverter for UnsafeTaskReceiver { #[cfg(test)] mod test { + extern mod extra; + use prelude::*; use rt::test::*; use unstable::run_in_bare_thread; @@ -862,12 +927,15 @@ mod test { do run_in_bare_thread { let sleepers = SleeperList::new(); - let work_queue = WorkQueue::new(); + let normal_queue = WorkQueue::new(); + let special_queue = WorkQueue::new(); + let queues = ~[normal_queue.clone(), special_queue.clone()]; // Our normal scheduler let mut normal_sched = ~Scheduler::new( ~UvEventLoop::new(), - work_queue.clone(), + normal_queue, + queues.clone(), sleepers.clone()); let normal_handle = Cell::new(normal_sched.make_handle()); @@ -877,7 +945,8 @@ mod test { // Our special scheduler let mut special_sched = ~Scheduler::new_special( ~UvEventLoop::new(), - work_queue.clone(), + special_queue.clone(), + queues.clone(), sleepers.clone(), false, Some(friend_handle)); diff --git a/src/libstd/rt/select.rs b/src/libstd/rt/select.rs index 006b777b71b..07f8ca77d9d 100644 --- a/src/libstd/rt/select.rs +++ b/src/libstd/rt/select.rs @@ -182,6 +182,7 @@ mod test { fn select_stream() { use util; use comm::GenericChan; + use iter::Times; // Sends 10 buffered packets, and uses select to retrieve them all. // Puts the port in a different spot in the vector each time. @@ -265,6 +266,7 @@ mod test { fn select_racing_senders_helper(killable: bool, send_on_chans: ~[uint]) { use rt::test::spawntask_random; + use iter::Times; do run_in_newsched_task { // A bit of stress, since ordinarily this is just smoke and mirrors. diff --git a/src/libstd/rt/test.rs b/src/libstd/rt/test.rs index 792ea5eb33f..92366d5187f 100644 --- a/src/libstd/rt/test.rs +++ b/src/libstd/rt/test.rs @@ -15,8 +15,8 @@ use cell::Cell; use clone::Clone; use container::Container; use iterator::{Iterator, range}; -use vec::{OwnedVector, MutableVector}; use super::io::net::ip::{SocketAddr, Ipv4Addr, Ipv6Addr}; +use vec::{OwnedVector, MutableVector, ImmutableVector}; use rt::sched::Scheduler; use unstable::run_in_bare_thread; use rt::thread::Thread; @@ -29,8 +29,12 @@ use result::{Result, Ok, Err}; pub fn new_test_uv_sched() -> Scheduler { + let queue = WorkQueue::new(); + let queues = ~[queue.clone()]; + let mut sched = Scheduler::new(~UvEventLoop::new(), - WorkQueue::new(), + queue, + queues, SleeperList::new()); // Don't wait for the Shutdown message @@ -164,15 +168,21 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { }; let sleepers = SleeperList::new(); - let work_queue = WorkQueue::new(); let mut handles = ~[]; let mut scheds = ~[]; + let mut work_queues = ~[]; for _ in range(0u, nthreads) { + let work_queue = WorkQueue::new(); + work_queues.push(work_queue); + } + + for i in range(0u, nthreads) { let loop_ = ~UvEventLoop::new(); let mut sched = ~Scheduler::new(loop_, - work_queue.clone(), + work_queues[i].clone(), + work_queues.clone(), sleepers.clone()); let handle = sched.make_handle(); diff --git a/src/libstd/task/spawn.rs b/src/libstd/task/spawn.rs index 2d0a2d98e9f..05a17f8539c 100644 --- a/src/libstd/task/spawn.rs +++ b/src/libstd/task/spawn.rs @@ -98,6 +98,7 @@ 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; @@ -722,10 +723,16 @@ fn spawn_raw_newsched(mut opts: TaskOpts, f: ~fn()) { let sched = 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, - (*sched).work_queue.clone(), + work_queue, + (*sched).work_queues.clone(), (*sched).sleeper_list.clone(), false, Some(sched_handle)); diff --git a/src/test/bench/rt-messaging-ping-pong.rs b/src/test/bench/rt-messaging-ping-pong.rs new file mode 100644 index 00000000000..3d38d61bc2e --- /dev/null +++ b/src/test/bench/rt-messaging-ping-pong.rs @@ -0,0 +1,82 @@ +// Copyright 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 or the MIT license +// , at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +extern mod extra; + +use std::task::spawn; +use std::os; +use std::uint; +use std::rt::test::spawntask_later; +use std::cell::Cell; + +// This is a simple bench that creates M pairs of of tasks. These +// tasks ping-pong back and forth over a pair of streams. This is a +// cannonical message-passing benchmark as it heavily strains message +// passing and almost nothing else. + +fn ping_pong_bench(n: uint, m: uint) { + + // Create pairs of tasks that pingpong back and forth. + fn run_pair(n: uint) { + // Create a stream A->B + let (pa,ca) = stream::<()>(); + // Create a stream B->A + let (pb,cb) = stream::<()>(); + + let pa = Cell::new(pa); + let ca = Cell::new(ca); + let pb = Cell::new(pb); + let cb = Cell::new(cb); + + do spawntask_later() || { + let chan = ca.take(); + let port = pb.take(); + do n.times { + chan.send(()); + port.recv(); + } + } + + do spawntask_later() || { + let chan = cb.take(); + let port = pa.take(); + do n.times { + port.recv(); + chan.send(()); + } + } + } + + do m.times { + run_pair(n) + } + +} + + + +fn main() { + + let args = os::args(); + let n = if args.len() == 3 { + uint::from_str(args[1]).unwrap() + } else { + 10000 + }; + + let m = if args.len() == 3 { + uint::from_str(args[2]).unwrap() + } else { + 4 + }; + + ping_pong_bench(n, m); + +} diff --git a/src/test/bench/rt-parfib.rs b/src/test/bench/rt-parfib.rs new file mode 100644 index 00000000000..6669342f511 --- /dev/null +++ b/src/test/bench/rt-parfib.rs @@ -0,0 +1,49 @@ +// Copyright 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 or the MIT license +// , at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +extern mod extra; + +use std::task::spawn; +use std::os; +use std::uint; +use std::rt::test::spawntask_later; +use std::cell::Cell; +use std::comm::*; + +// A simple implementation of parfib. One subtree is found in a new +// task and communicated over a oneshot pipe, the other is found +// locally. There is no sequential-mode threshold. + +fn parfib(n: uint) -> uint { + if(n == 0 || n == 1) { + return 1; + } + + let (port,chan) = oneshot::(); + let chan = Cell::new(chan); + do spawntask_later { + chan.take().send(parfib(n-1)); + }; + let m2 = parfib(n-2); + return (port.recv() + m2); +} + +fn main() { + + let args = os::args(); + let n = if args.len() == 2 { + uint::from_str(args[1]).unwrap() + } else { + 10 + }; + + parfib(n); + +} diff --git a/src/test/bench/rt-spawn-rate.rs b/src/test/bench/rt-spawn-rate.rs new file mode 100644 index 00000000000..ff578ed70b9 --- /dev/null +++ b/src/test/bench/rt-spawn-rate.rs @@ -0,0 +1,33 @@ +// Copyright 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 or the MIT license +// , at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +extern mod extra; + +use std::task::spawn; +use std::os; +use std::uint; + +// Very simple spawn rate test. Spawn N tasks that do nothing and +// return. + +fn main() { + + let args = os::args(); + let n = if args.len() == 2 { + uint::from_str(args[1]).unwrap() + } else { + 100000 + }; + + do n.times { + do spawn || {}; + } + +} -- cgit 1.4.1-3-g733a5