diff --git a/src/libstd/rt/comm.rs b/src/libstd/rt/comm.rs index dd27c03ff5164..7608bc89e021f 100644 --- a/src/libstd/rt/comm.rs +++ b/src/libstd/rt/comm.rs @@ -20,7 +20,8 @@ use cast; use util; use ops::Drop; use kinds::Owned; -use rt::sched::{Scheduler, Coroutine}; +use rt::sched::{Scheduler}; +use rt::task::Task; use rt::local::Local; use unstable::atomics::{AtomicUint, AtomicOption, SeqCst}; use unstable::sync::UnsafeAtomicRcBox; @@ -136,7 +137,7 @@ impl ChanOne { } task_as_state => { // Port is blocked. Wake it up. - let recvr: ~Coroutine = cast::transmute(task_as_state); + let recvr: ~Task = cast::transmute(task_as_state); let mut sched = Local::take::(); rtdebug!("rendezvous send"); sched.metrics.rendezvous_sends += 1; @@ -192,7 +193,7 @@ impl PortOne { // NB: We have to drop back into the scheduler event loop here // instead of switching immediately back or we could end up // triggering infinite recursion on the scheduler's stack. - let task: ~Coroutine = cast::transmute(task_as_state); + let task: ~Task = cast::transmute(task_as_state); sched.enqueue_task(task); } _ => util::unreachable() @@ -257,7 +258,7 @@ impl Drop for ChanOneHack { task_as_state => { // The port is blocked waiting for a message we will never send. Wake it. assert!((*this.packet()).payload.is_none()); - let recvr: ~Coroutine = cast::transmute(task_as_state); + let recvr: ~Task = cast::transmute(task_as_state); let sched = Local::take::(); sched.schedule_task(recvr); } @@ -554,6 +555,8 @@ mod test { { let _c = chan; } port.recv(); }; + // What is our res? + rtdebug!("res is: %?", res.is_err()); assert!(res.is_err()); } } @@ -905,4 +908,5 @@ mod test { } } } + } diff --git a/src/libstd/rt/join_latch.rs b/src/libstd/rt/join_latch.rs index ad5cf2eb378c5..79c0d5da9a4f5 100644 --- a/src/libstd/rt/join_latch.rs +++ b/src/libstd/rt/join_latch.rs @@ -643,3 +643,4 @@ mod test { } } } + diff --git a/src/libstd/rt/local.rs b/src/libstd/rt/local.rs index 6df1ffaa453f3..374933ab281b8 100644 --- a/src/libstd/rt/local.rs +++ b/src/libstd/rt/local.rs @@ -13,6 +13,7 @@ use rt::sched::Scheduler; use rt::task::Task; use rt::local_ptr; use rt::rtio::{EventLoop, IoFactoryObject}; +//use borrow::to_uint; pub trait Local { fn put(value: ~Self); @@ -32,6 +33,7 @@ impl Local for Scheduler { let res_ptr: *mut Option = &mut res; unsafe { do local_ptr::borrow |sched| { +// rtdebug!("successfully unsafe borrowed sched pointer"); let result = f(sched); *res_ptr = Some(result); } @@ -51,9 +53,12 @@ impl Local for Task { fn exists() -> bool { rtabort!("unimpl") } fn borrow(f: &fn(&mut Task) -> T) -> T { do Local::borrow:: |sched| { +// rtdebug!("sched about to grab current_task"); match sched.current_task { Some(~ref mut task) => { - f(&mut *task.task) +// rtdebug!("current task pointer: %x", to_uint(task)); +// rtdebug!("current task heap pointer: %x", to_uint(&task.heap)); + f(task) } None => { rtabort!("no scheduler") @@ -64,7 +69,7 @@ impl Local for Task { unsafe fn unsafe_borrow() -> *mut Task { match (*Local::unsafe_borrow::()).current_task { Some(~ref mut task) => { - let s: *mut Task = &mut *task.task; + let s: *mut Task = &mut *task; return s; } None => { diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs index 97a0285ea196e..aae194ae54838 100644 --- a/src/libstd/rt/mod.rs +++ b/src/libstd/rt/mod.rs @@ -67,7 +67,7 @@ use iter::Times; use iterator::IteratorUtil; use option::Some; use ptr::RawPtr; -use rt::sched::{Scheduler, Coroutine, Shutdown}; +use rt::sched::{Scheduler, Shutdown}; use rt::sleeper_list::SleeperList; use rt::task::Task; use rt::thread::Thread; @@ -267,10 +267,10 @@ pub fn run(main: ~fn()) -> int { }; // Create and enqueue the main task. - let mut new_task = ~Task::new_root(); - new_task.on_exit = Some(on_exit); - let main_task = ~Coroutine::with_task(&mut scheds[0].stack_pool, - new_task, main); + let main_cell = Cell::new(main); + let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, + main_cell.take()); + main_task.on_exit = Some(on_exit); scheds[0].enqueue_task(main_task); // Run each scheduler in a thread. @@ -347,7 +347,7 @@ pub fn context() -> RuntimeContext { #[test] fn test_context() { use unstable::run_in_bare_thread; - use self::sched::{Scheduler, Coroutine}; + use self::sched::{Scheduler}; use rt::local::Local; use rt::test::new_test_uv_sched; @@ -355,7 +355,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_root(&mut sched.stack_pool) { + let task = ~do Task::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 26956c49a629a..7d8c673636e1b 100644 --- a/src/libstd/rt/sched.rs +++ b/src/libstd/rt/sched.rs @@ -11,24 +11,19 @@ use option::*; use sys; use cast::transmute; -use cell::Cell; use clone::Clone; use super::sleeper_list::SleeperList; use super::work_queue::WorkQueue; -use super::stack::{StackPool, StackSegment}; +use super::stack::{StackPool}; use super::rtio::{EventLoop, EventLoopObject, RemoteCallbackObject}; use super::context::Context; -use super::task::Task; +use super::task::{Task, AnySched, Sched}; use super::message_queue::MessageQueue; use rt::local_ptr; use rt::local::Local; use rt::rtio::RemoteCallback; use rt::metrics::SchedMetrics; - -//use to_str::ToStr; - -/// To allow for using pointers as scheduler ids use borrow::{to_uint}; /// The Scheduler is responsible for coordinating execution of Coroutines @@ -41,7 +36,7 @@ use borrow::{to_uint}; pub struct Scheduler { /// A queue of available work. Under a work-stealing policy there /// is one per Scheduler. - priv work_queue: WorkQueue<~Coroutine>, + priv work_queue: 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. @@ -66,7 +61,7 @@ pub struct Scheduler { /// Always valid when a task is executing, otherwise not priv saved_context: Context, /// The currently executing task - current_task: Option<~Coroutine>, + current_task: Option<~Task>, /// An action performed after a context switch on behalf of the /// code running before the context switch priv cleanup_job: Option, @@ -81,33 +76,15 @@ pub struct SchedHandle { sched_id: uint } -pub struct Coroutine { - /// The segment of stack on which the task is currently running or, - /// if the task is blocked, on which the task will resume execution - priv current_stack_segment: StackSegment, - /// These are always valid when the task is not running, unless - /// the task is dead - priv saved_context: Context, - /// The heap, GC, unwinding, local storage, logging - task: ~Task, -} - -// A scheduler home is either a handle to the home scheduler, or an -// explicit "AnySched". -pub enum SchedHome { - AnySched, - Sched(SchedHandle) -} - pub enum SchedMessage { Wake, Shutdown, - PinnedTask(~Coroutine) + PinnedTask(~Task) } enum CleanupJob { DoNothing, - GiveTask(~Coroutine, UnsafeTaskReceiver) + GiveTask(~Task, UnsafeTaskReceiver) } impl Scheduler { @@ -116,7 +93,7 @@ impl Scheduler { pub fn sched_id(&self) -> uint { to_uint(self) } pub fn new(event_loop: ~EventLoopObject, - work_queue: WorkQueue<~Coroutine>, + work_queue: WorkQueue<~Task>, sleeper_list: SleeperList) -> Scheduler { @@ -125,7 +102,7 @@ impl Scheduler { } pub fn new_special(event_loop: ~EventLoopObject, - work_queue: WorkQueue<~Coroutine>, + work_queue: WorkQueue<~Task>, sleeper_list: SleeperList, run_anything: bool) -> Scheduler { @@ -253,7 +230,7 @@ impl Scheduler { /// Pushes the task onto the work stealing queue and tells the /// event loop to run it later. Always use this instead of pushing /// to the work queue directly. - pub fn enqueue_task(&mut self, task: ~Coroutine) { + pub fn enqueue_task(&mut self, task: ~Task) { // We don't want to queue tasks that belong on other threads, // so we send them home at enqueue time. @@ -307,7 +284,7 @@ impl Scheduler { rtdebug!("recv BiasedTask message in sched: %u", this.sched_id()); let mut task = task; - task.task.home = Some(Sched(this.make_handle())); + task.home = Some(Sched(this.make_handle())); this.resume_task_immediately(task); return true; } @@ -349,9 +326,9 @@ impl Scheduler { } /// Given an input Coroutine sends it back to its home scheduler. - fn send_task_home(task: ~Coroutine) { + fn send_task_home(task: ~Task) { let mut task = task; - let mut home = task.task.home.swap_unwrap(); + let mut home = task.home.swap_unwrap(); match home { Sched(ref mut home_handle) => { home_handle.send(PinnedTask(task)); @@ -377,7 +354,7 @@ impl Scheduler { match this.work_queue.pop() { Some(task) => { let action_id = { - let home = &task.task.home; + let home = &task.home; match home { &Some(Sched(ref home_handle)) if home_handle.sched_id != this.sched_id() => { @@ -440,14 +417,15 @@ impl Scheduler { rtdebug!("ending running task"); do self.deschedule_running_task_and_then |sched, dead_task| { - let dead_task = Cell::new(dead_task); - dead_task.take().recycle(&mut sched.stack_pool); + let mut dead_task = dead_task; + let coroutine = dead_task.coroutine.swap_unwrap(); + coroutine.recycle(&mut sched.stack_pool); } rtabort!("control reached end of task"); } - pub fn schedule_task(~self, task: ~Coroutine) { + pub fn schedule_task(~self, task: ~Task) { assert!(self.in_task_context()); // is the task home? @@ -477,7 +455,7 @@ impl Scheduler { // Core scheduling ops - pub fn resume_task_immediately(~self, task: ~Coroutine) { + pub fn resume_task_immediately(~self, task: ~Task) { let mut this = self; assert!(!this.in_task_context()); @@ -520,7 +498,7 @@ impl Scheduler { /// This passes a Scheduler pointer to the fn after the context switch /// in order to prevent that fn from performing further scheduling operations. /// Doing further scheduling could easily result in infinite recursion. - pub fn deschedule_running_task_and_then(~self, f: &fn(&mut Scheduler, ~Coroutine)) { + pub fn deschedule_running_task_and_then(~self, f: &fn(&mut Scheduler, ~Task)) { let mut this = self; assert!(this.in_task_context()); @@ -529,8 +507,8 @@ impl Scheduler { unsafe { let blocked_task = this.current_task.swap_unwrap(); - let f_fake_region = transmute::<&fn(&mut Scheduler, ~Coroutine), - &fn(&mut Scheduler, ~Coroutine)>(f); + let f_fake_region = transmute::<&fn(&mut Scheduler, ~Task), + &fn(&mut Scheduler, ~Task)>(f); let f_opaque = ClosureConverter::from_fn(f_fake_region); this.enqueue_cleanup_job(GiveTask(blocked_task, f_opaque)); } @@ -552,8 +530,8 @@ impl Scheduler { /// Switch directly to another task, without going through the scheduler. /// You would want to think hard about doing this, e.g. if there are /// pending I/O events it would be a bad idea. - pub fn switch_running_tasks_and_then(~self, next_task: ~Coroutine, - f: &fn(&mut Scheduler, ~Coroutine)) { + pub fn switch_running_tasks_and_then(~self, next_task: ~Task, + f: &fn(&mut Scheduler, ~Task)) { let mut this = self; assert!(this.in_task_context()); @@ -562,8 +540,8 @@ impl Scheduler { let old_running_task = this.current_task.swap_unwrap(); let f_fake_region = unsafe { - transmute::<&fn(&mut Scheduler, ~Coroutine), - &fn(&mut Scheduler, ~Coroutine)>(f) + transmute::<&fn(&mut Scheduler, ~Task), + &fn(&mut Scheduler, ~Task)>(f) }; let f_opaque = ClosureConverter::from_fn(f_fake_region); this.enqueue_cleanup_job(GiveTask(old_running_task, f_opaque)); @@ -630,12 +608,22 @@ impl Scheduler { // because borrowck thinks the three patterns are conflicting // borrows unsafe { - let last_task = transmute::, Option<&mut Coroutine>>(last_task); + let last_task = transmute::, Option<&mut Task>>(last_task); let last_task_context = match last_task { - Some(t) => Some(&mut t.saved_context), None => None + Some(t) => { + Some(&mut t.coroutine.get_mut_ref().saved_context) + } + None => { + None + } }; let next_task_context = match self.current_task { - Some(ref mut t) => Some(&mut t.saved_context), None => None + Some(ref mut t) => { + Some(&mut t.coroutine.get_mut_ref().saved_context) + } + None => { + None + } }; // XXX: These transmutes can be removed after snapshot return (transmute(&mut self.saved_context), @@ -660,186 +648,34 @@ impl SchedHandle { } } -impl Coroutine { - - /// This function checks that a coroutine is running "home". - pub fn is_home(&self) -> bool { - rtdebug!("checking if coroutine is home"); - do Local::borrow:: |sched| { - match self.task.home { - Some(AnySched) => { false } - Some(Sched(SchedHandle { sched_id: ref id, _ })) => { - *id == sched.sched_id() - } - None => { rtabort!("error: homeless task!"); } - } - } - } - - /// Without access to self, but with access to the "expected home - /// id", see if we are home. - fn is_home_using_id(id: uint) -> bool { - rtdebug!("checking if coroutine is home using id"); - do Local::borrow:: |sched| { - if sched.sched_id() == id { - true - } else { - false - } - } - } - - /// Check if this coroutine has a home - fn homed(&self) -> bool { - rtdebug!("checking if this coroutine has a home"); - match self.task.home { - Some(AnySched) => { false } - Some(Sched(_)) => { true } - None => { rtabort!("error: homeless task!"); - } - } - } - - /// A version of is_home that does not need to use TLS, it instead - /// takes local scheduler as a parameter. - fn is_home_no_tls(&self, sched: &~Scheduler) -> bool { - rtdebug!("checking if coroutine is home without tls"); - match self.task.home { - Some(AnySched) => { true } - Some(Sched(SchedHandle { sched_id: ref id, _})) => { - *id == sched.sched_id() - } - None => { rtabort!("error: homeless task!"); } - } - } - - /// Check TLS for the scheduler to see if we are on a special - /// scheduler. - pub fn on_special() -> bool { - rtdebug!("checking if coroutine is executing on special sched"); - do Local::borrow::() |sched| { - !sched.run_anything - } - } - - // Created new variants of "new" that takes a home scheduler - // parameter. The original with_task now calls with_task_homed - // using the AnySched paramter. - - pub fn new_homed(stack_pool: &mut StackPool, home: SchedHome, start: ~fn()) -> Coroutine { - Coroutine::with_task_homed(stack_pool, ~Task::new_root(), start, home) - } - - pub fn new_root(stack_pool: &mut StackPool, start: ~fn()) -> Coroutine { - Coroutine::with_task(stack_pool, ~Task::new_root(), start) - } - - pub fn with_task_homed(stack_pool: &mut StackPool, - task: ~Task, - start: ~fn(), - home: SchedHome) -> Coroutine { - - static MIN_STACK_SIZE: uint = 1000000; // XXX: Too much stack - - let start = Coroutine::build_start_wrapper(start); - let mut stack = stack_pool.take_segment(MIN_STACK_SIZE); - // NB: Context holds a pointer to that ~fn - let initial_context = Context::new(start, &mut stack); - let mut crt = Coroutine { - current_stack_segment: stack, - saved_context: initial_context, - task: task, - }; - crt.task.home = Some(home); - return crt; - } - - pub fn with_task(stack_pool: &mut StackPool, - task: ~Task, - start: ~fn()) -> Coroutine { - Coroutine::with_task_homed(stack_pool, - task, - start, - AnySched) - } - - fn build_start_wrapper(start: ~fn()) -> ~fn() { - // XXX: The old code didn't have this extra allocation - let start_cell = Cell::new(start); - let wrapper: ~fn() = || { - // This is the first code to execute after the initial - // context switch to the task. The previous context may - // have asked us to do some cleanup. - unsafe { - let sched = Local::unsafe_borrow::(); - (*sched).run_cleanup_job(); - - let sched = Local::unsafe_borrow::(); - let task = (*sched).current_task.get_mut_ref(); - // FIXME #6141: shouldn't neet to put `start()` in - // another closure - let start_cell = Cell::new(start_cell.take()); - do task.task.run { - // N.B. Removing `start` from the start wrapper - // closure by emptying a cell is critical for - // correctness. The ~Task pointer, and in turn the - // closure used to initialize the first call - // frame, is destroyed in scheduler context, not - // task context. So any captured closures must - // not contain user-definable dtors that expect to - // be in task context. By moving `start` out of - // the closure, all the user code goes out of - // scope while the task is still running. - let start = start_cell.take(); - start(); - }; - } - - let sched = Local::take::(); - sched.terminate_current_task(); - }; - return wrapper; - } - - /// Destroy the task and try to reuse its components - pub fn recycle(~self, stack_pool: &mut StackPool) { - match self { - ~Coroutine {current_stack_segment, _} => { - stack_pool.give_segment(current_stack_segment); - } - } - } -} - // XXX: Some hacks to put a &fn in Scheduler without borrowck // complaining type UnsafeTaskReceiver = sys::Closure; trait ClosureConverter { - fn from_fn(&fn(&mut Scheduler, ~Coroutine)) -> Self; - fn to_fn(self) -> &fn(&mut Scheduler, ~Coroutine); + fn from_fn(&fn(&mut Scheduler, ~Task)) -> Self; + fn to_fn(self) -> &fn(&mut Scheduler, ~Task); } impl ClosureConverter for UnsafeTaskReceiver { - fn from_fn(f: &fn(&mut Scheduler, ~Coroutine)) -> UnsafeTaskReceiver { unsafe { transmute(f) } } - fn to_fn(self) -> &fn(&mut Scheduler, ~Coroutine) { unsafe { transmute(self) } } + fn from_fn(f: &fn(&mut Scheduler, ~Task)) -> UnsafeTaskReceiver { unsafe { transmute(f) } } + fn to_fn(self) -> &fn(&mut Scheduler, ~Task) { unsafe { transmute(self) } } } + #[cfg(test)] mod test { use int; use cell::Cell; - use iterator::IteratorUtil; use unstable::run_in_bare_thread; use task::spawn; use rt::local::Local; use rt::test::*; use super::*; use rt::thread::Thread; - use ptr::to_uint; - use vec::MutableVector; + use borrow::to_uint; + use rt::task::{Task,Sched}; // Confirm that a sched_id actually is the uint form of the // pointer to the scheduler struct. - #[test] fn simple_sched_id_test() { do run_in_bare_thread { @@ -850,7 +686,6 @@ mod test { // Compare two scheduler ids that are different, this should never // fail but may catch a mistake someday. - #[test] fn compare_sched_id_test() { do run_in_bare_thread { @@ -862,7 +697,6 @@ mod test { // A simple test to check if a homed task run on a single // scheduler ends up executing while home. - #[test] fn test_home_sched() { do run_in_bare_thread { @@ -873,8 +707,8 @@ mod test { let sched_handle = sched.make_handle(); let sched_id = sched.sched_id(); - let task = ~do Coroutine::new_homed(&mut sched.stack_pool, - Sched(sched_handle)) { + let task = ~do Task::new_root_homed(&mut sched.stack_pool, + Sched(sched_handle)) { unsafe { *task_ran_ptr = true }; let sched = Local::take::(); assert!(sched.sched_id() == sched_id); @@ -887,7 +721,6 @@ mod test { } // A test for each state of schedule_task - #[test] fn test_schedule_home_states() { @@ -897,7 +730,6 @@ mod test { use rt::work_queue::WorkQueue; do run_in_bare_thread { -// let nthreads = 2; let sleepers = SleeperList::new(); let work_queue = WorkQueue::new(); @@ -923,33 +755,33 @@ mod test { let t1_handle = special_sched.make_handle(); let t4_handle = special_sched.make_handle(); - let t1f = ~do Coroutine::new_homed(&mut special_sched.stack_pool, - Sched(t1_handle)) { - let is_home = Coroutine::is_home_using_id(special_id); + let t1f = ~do Task::new_root_homed(&mut special_sched.stack_pool, + Sched(t1_handle)) || { + let is_home = Task::is_home_using_id(special_id); rtdebug!("t1 should be home: %b", is_home); assert!(is_home); }; let t1f = Cell::new(t1f); - let t2f = ~do Coroutine::new_root(&mut normal_sched.stack_pool) { - let on_special = Coroutine::on_special(); + let t2f = ~do Task::new_root(&mut normal_sched.stack_pool) { + let on_special = Task::on_special(); rtdebug!("t2 should not be on special: %b", on_special); assert!(!on_special); }; let t2f = Cell::new(t2f); - let t3f = ~do Coroutine::new_root(&mut normal_sched.stack_pool) { + let t3f = ~do Task::new_root(&mut normal_sched.stack_pool) { // not on special - let on_special = Coroutine::on_special(); + let on_special = Task::on_special(); rtdebug!("t3 should not be on special: %b", on_special); assert!(!on_special); }; let t3f = Cell::new(t3f); - let t4f = ~do Coroutine::new_homed(&mut special_sched.stack_pool, - Sched(t4_handle)) { + let t4f = ~do Task::new_root_homed(&mut special_sched.stack_pool, + Sched(t4_handle)) { // is home - let home = Coroutine::is_home_using_id(special_id); + let home = Task::is_home_using_id(special_id); rtdebug!("t4 should be home: %b", home); assert!(home); }; @@ -987,7 +819,7 @@ mod test { let t4 = Cell::new(t4); // build a main task that runs our four tests - let main_task = ~do Coroutine::new_root(&mut normal_sched.stack_pool) { + let main_task = ~do Task::new_root(&mut normal_sched.stack_pool) { // the two tasks that require a normal start location t2.take()(); t4.take()(); @@ -996,7 +828,7 @@ mod test { }; // task to run the two "special start" tests - let special_task = ~do Coroutine::new_homed( + let special_task = ~do Task::new_root_homed( &mut special_sched.stack_pool, Sched(special_handle2.take())) { t1.take()(); @@ -1026,91 +858,7 @@ mod test { } } - // The following test is a bit of a mess, but it trys to do - // something tricky so I'm not sure how to get around this in the - // short term. - - // A number of schedulers are created, and then a task is created - // and assigned a home scheduler. It is then "started" on a - // different scheduler. The scheduler it is started on should - // observe that the task is not home, and send it home. - - // This test is light in that it does very little. - - #[test] - fn test_transfer_task_home() { - - use rt::uv::uvio::UvEventLoop; - use rt::sched::Shutdown; - use rt::sleeper_list::SleeperList; - use rt::work_queue::WorkQueue; - use uint; - use container::Container; - use vec::OwnedVector; - - do run_in_bare_thread { - - static N: uint = 8; - - let sleepers = SleeperList::new(); - let work_queue = WorkQueue::new(); - - let mut handles = ~[]; - let mut scheds = ~[]; - - for uint::range(0, N) |_| { - let loop_ = ~UvEventLoop::new(); - let mut sched = ~Scheduler::new(loop_, - work_queue.clone(), - sleepers.clone()); - let handle = sched.make_handle(); - rtdebug!("sched id: %u", handle.sched_id); - handles.push(handle); - scheds.push(sched); - }; - - let handles = Cell::new(handles); - - let home_handle = scheds[6].make_handle(); - let home_id = home_handle.sched_id; - let home = Sched(home_handle); - - let main_task = ~do Coroutine::new_homed(&mut scheds[1].stack_pool, home) { - - // Here we check if the task is running on its home. - let sched = Local::take::(); - rtdebug!("run location scheduler id: %u, home: %u", - sched.sched_id(), - home_id); - assert!(sched.sched_id() == home_id); - Local::put::(sched); - - let mut handles = handles.take(); - for handles.mut_iter().advance |handle| { - handle.send(Shutdown); - } - }; - - scheds[0].enqueue_task(main_task); - - let mut threads = ~[]; - - while !scheds.is_empty() { - let sched = scheds.pop(); - let sched_cell = Cell::new(sched); - let thread = do Thread::start { - let sched = sched_cell.take(); - sched.run(); - }; - threads.push(thread); - } - - let _threads = threads; - } - } - // Do it a lot - #[test] fn test_stress_schedule_task_states() { let n = stress_factor() * 120; @@ -1119,21 +867,6 @@ mod test { } } - // The goal is that this is the high-stress test for making sure - // homing is working. It allocates RUST_RT_STRESS tasks that - // do nothing but assert that they are home at execution - // time. These tasks are queued to random schedulers, so sometimes - // they are home and sometimes not. It also runs RUST_RT_STRESS - // times. - - #[test] - fn test_stress_homed_tasks() { - let n = stress_factor(); - for int::range(0,n as int) |_| { - run_in_mt_newsched_task_random_homed(); - } - } - #[test] fn test_simple_scheduling() { do run_in_bare_thread { @@ -1141,7 +874,7 @@ mod test { let task_ran_ptr: *mut bool = &mut task_ran; let mut sched = ~new_test_uv_sched(); - let task = ~do Coroutine::new_root(&mut sched.stack_pool) { + let task = ~do Task::new_root(&mut sched.stack_pool) { unsafe { *task_ran_ptr = true; } }; sched.enqueue_task(task); @@ -1159,7 +892,7 @@ mod test { let mut sched = ~new_test_uv_sched(); for int::range(0, total) |_| { - let task = ~do Coroutine::new_root(&mut sched.stack_pool) { + let task = ~do Task::new_root(&mut sched.stack_pool) { unsafe { *task_count_ptr = *task_count_ptr + 1; } }; sched.enqueue_task(task); @@ -1176,10 +909,10 @@ mod test { let count_ptr: *mut int = &mut count; let mut sched = ~new_test_uv_sched(); - let task1 = ~do Coroutine::new_root(&mut sched.stack_pool) { + let task1 = ~do Task::new_root(&mut sched.stack_pool) { unsafe { *count_ptr = *count_ptr + 1; } let mut sched = Local::take::(); - let task2 = ~do Coroutine::new_root(&mut sched.stack_pool) { + let task2 = ~do Task::new_root(&mut sched.stack_pool) { unsafe { *count_ptr = *count_ptr + 1; } }; // Context switch directly to the new task @@ -1204,7 +937,7 @@ mod test { let mut sched = ~new_test_uv_sched(); - let start_task = ~do Coroutine::new_root(&mut sched.stack_pool) { + let start_task = ~do Task::new_root(&mut sched.stack_pool) { run_task(count_ptr); }; sched.enqueue_task(start_task); @@ -1214,7 +947,7 @@ mod test { fn run_task(count_ptr: *mut int) { do Local::borrow:: |sched| { - let task = ~do Coroutine::new_root(&mut sched.stack_pool) { + let task = ~do Task::new_root(&mut sched.stack_pool) { unsafe { *count_ptr = *count_ptr + 1; if *count_ptr != MAX { @@ -1232,7 +965,7 @@ mod test { fn test_block_task() { do run_in_bare_thread { let mut sched = ~new_test_uv_sched(); - let task = ~do Coroutine::new_root(&mut sched.stack_pool) { + let task = ~do Task::new_root(&mut sched.stack_pool) { let sched = Local::take::(); assert!(sched.in_task_context()); do sched.deschedule_running_task_and_then() |sched, task| { @@ -1279,13 +1012,13 @@ mod test { let mut sched1 = ~new_test_uv_sched(); let handle1 = sched1.make_handle(); let handle1_cell = Cell::new(handle1); - let task1 = ~do Coroutine::new_root(&mut sched1.stack_pool) { + let task1 = ~do Task::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_root(&mut sched2.stack_pool) { + let task2 = ~do Task::new_root(&mut sched2.stack_pool) { port_cell.take().recv(); // Release the other scheduler's handle so it can exit handle1_cell.take(); @@ -1382,7 +1115,6 @@ mod test { } } } - } #[test] @@ -1407,5 +1139,4 @@ mod test { } } } - } diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs index 97c3b6a749bc9..b2e4f0d4716ff 100644 --- a/src/libstd/rt/task.rs +++ b/src/libstd/rt/task.rs @@ -23,8 +23,11 @@ use option::{Option, Some, None}; use rt::local::Local; use rt::logging::StdErrLogger; use super::local_heap::LocalHeap; -use rt::sched::{SchedHome, AnySched}; +use rt::sched::{Scheduler, SchedHandle}; use rt::join_latch::JoinLatch; +use rt::stack::{StackSegment, StackPool}; +use rt::context::Context; +use cell::Cell; pub struct Task { heap: LocalHeap, @@ -35,7 +38,22 @@ pub struct Task { home: Option, join_latch: Option<~JoinLatch>, on_exit: Option<~fn(bool)>, - destroyed: bool + destroyed: bool, + coroutine: Option<~Coroutine> +} + +pub struct Coroutine { + /// The segment of stack on which the task is currently running or + /// if the task is blocked, on which the task will resume + /// execution. + priv current_stack_segment: StackSegment, + /// Always valid if the task is alive and not running. + saved_context: Context +} + +pub enum SchedHome { + AnySched, + Sched(SchedHandle) } pub struct GarbageCollector; @@ -46,31 +64,50 @@ pub struct Unwinder { } impl Task { - pub fn new_root() -> Task { + + pub fn new_root(stack_pool: &mut StackPool, + start: ~fn()) -> Task { + Task::new_root_homed(stack_pool, AnySched, start) + } + + pub fn new_child(&mut self, + stack_pool: &mut StackPool, + start: ~fn()) -> Task { + self.new_child_homed(stack_pool, AnySched, start) + } + + pub fn new_root_homed(stack_pool: &mut StackPool, + home: SchedHome, + start: ~fn()) -> Task { Task { heap: LocalHeap::new(), gc: GarbageCollector, storage: LocalStorage(ptr::null(), None), logger: StdErrLogger, unwinder: Unwinder { unwinding: false }, - home: Some(AnySched), + home: Some(home), join_latch: Some(JoinLatch::new_root()), on_exit: None, - destroyed: false + destroyed: false, + coroutine: Some(~Coroutine::new(stack_pool, start)) } } - pub fn new_child(&mut self) -> Task { + pub fn new_child_homed(&mut self, + stack_pool: &mut StackPool, + home: SchedHome, + start: ~fn()) -> Task { Task { heap: LocalHeap::new(), gc: GarbageCollector, storage: LocalStorage(ptr::null(), None), logger: StdErrLogger, - home: Some(AnySched), + home: Some(home), unwinder: Unwinder { unwinding: false }, join_latch: Some(self.join_latch.get_mut_ref().new_child()), on_exit: None, - destroyed: false + destroyed: false, + coroutine: Some(~Coroutine::new(stack_pool, start)) } } @@ -108,11 +145,11 @@ impl Task { /// called unsafely, without removing Task from /// thread-local-storage. fn destroy(&mut self) { - // This is just an assertion that `destroy` was called unsafely - // and this instance of Task is still accessible. + do Local::borrow:: |task| { assert!(borrow::ref_eq(task, self)); } + match self.storage { LocalStorage(ptr, Some(ref dtor)) => { (*dtor)(ptr) @@ -125,12 +162,129 @@ impl Task { self.destroyed = true; } + + /// Check if *task* is currently home. + pub fn is_home(&self) -> bool { + do Local::borrow:: |sched| { + match self.home { + Some(AnySched) => { false } + Some(Sched(SchedHandle { sched_id: ref id, _ })) => { + *id == sched.sched_id() + } + None => { rtabort!("task home of None") } + } + } + } + + pub fn is_home_no_tls(&self, sched: &~Scheduler) -> bool { + match self.home { + Some(AnySched) => { false } + Some(Sched(SchedHandle { sched_id: ref id, _ })) => { + *id == sched.sched_id() + } + None => {rtabort!("task home of None") } + } + } + + pub fn is_home_using_id(sched_id: uint) -> bool { + do Local::borrow:: |task| { + match task.home { + Some(Sched(SchedHandle { sched_id: ref id, _ })) => { + *id == sched_id + } + Some(AnySched) => { false } + None => { rtabort!("task home of None") } + } + } + } + + /// Check if this *task* has a home. + pub fn homed(&self) -> bool { + match self.home { + Some(AnySched) => { false } + Some(Sched(_)) => { true } + None => { + rtabort!("task home of None") + } + } + } + + /// On a special scheduler? + pub fn on_special() -> bool { + do Local::borrow:: |sched| { + !sched.run_anything + } + } + } impl Drop for Task { fn finalize(&self) { assert!(self.destroyed) } } +// Coroutines represent nothing more than a context and a stack +// segment. + +impl Coroutine { + + pub fn new(stack_pool: &mut StackPool, start: ~fn()) -> Coroutine { + static MIN_STACK_SIZE: uint = 100000; // XXX: Too much stack + + let start = Coroutine::build_start_wrapper(start); + let mut stack = stack_pool.take_segment(MIN_STACK_SIZE); + let initial_context = Context::new(start, &mut stack); + Coroutine { + current_stack_segment: stack, + saved_context: initial_context + } + } + + fn build_start_wrapper(start: ~fn()) -> ~fn() { + let start_cell = Cell::new(start); + let wrapper: ~fn() = || { + // First code after swap to this new context. Run our + // cleanup job. + unsafe { + let sched = Local::unsafe_borrow::(); + (*sched).run_cleanup_job(); + + let sched = Local::unsafe_borrow::(); + let task = (*sched).current_task.get_mut_ref(); + + do task.run { + // N.B. Removing `start` from the start wrapper + // closure by emptying a cell is critical for + // correctness. The ~Task pointer, and in turn the + // closure used to initialize the first call + // frame, is destroyed in the scheduler context, + // not task context. So any captured closures must + // not contain user-definable dtors that expect to + // be in task context. By moving `start` out of + // the closure, all the user code goes our of + // scope while the task is still running. + let start = start_cell.take(); + start(); + }; + } + + let sched = Local::take::(); + sched.terminate_current_task(); + }; + return wrapper; + } + + /// Destroy coroutine and try to reuse stack segment. + pub fn recycle(~self, stack_pool: &mut StackPool) { + match self { + ~Coroutine { current_stack_segment, _ } => { + stack_pool.give_segment(current_stack_segment); + } + } + } + +} + + // Just a sanity check to make sure we are catching a Rust-thrown exception static UNWIND_TOKEN: uintptr_t = 839147; @@ -209,8 +363,10 @@ mod test { fn unwind() { do run_in_newsched_task() { let result = spawntask_try(||()); + rtdebug!("trying first assert"); assert!(result.is_ok()); let result = spawntask_try(|| fail!()); + rtdebug!("trying second assert"); assert!(result.is_err()); } } diff --git a/src/libstd/rt/test.rs b/src/libstd/rt/test.rs index b0e4968401474..659c7eb498573 100644 --- a/src/libstd/rt/test.rs +++ b/src/libstd/rt/test.rs @@ -8,30 +8,27 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +use cell::Cell; use uint; use option::{Some, None}; -use cell::Cell; -use clone::Clone; -use container::Container; -use iterator::IteratorUtil; -use vec::{OwnedVector, MutableVector}; -use result::{Result, Ok, Err}; -use unstable::run_in_bare_thread; +use rt::sched::Scheduler; use super::io::net::ip::{IpAddr, Ipv4}; -use rt::comm::oneshot; -use rt::task::Task; -use rt::thread::Thread; use rt::local::Local; -use rt::sched::{Scheduler, Coroutine}; -use rt::sleeper_list::SleeperList; +use unstable::run_in_bare_thread; +use rt::thread::Thread; +use rt::task::Task; +use rt::uv::uvio::UvEventLoop; use rt::work_queue::WorkQueue; +use rt::sleeper_list::SleeperList; +use rt::task::{Sched}; +use rt::comm::oneshot; +use result::{Result, Ok, Err}; pub fn new_test_uv_sched() -> Scheduler { - use rt::uv::uvio::UvEventLoop; - use rt::work_queue::WorkQueue; - use rt::sleeper_list::SleeperList; - let mut sched = Scheduler::new(~UvEventLoop::new(), WorkQueue::new(), SleeperList::new()); + let mut sched = Scheduler::new(~UvEventLoop::new(), + WorkQueue::new(), + SleeperList::new()); // Don't wait for the Shutdown message sched.no_sleep = true; return sched; @@ -41,19 +38,15 @@ pub fn new_test_uv_sched() -> Scheduler { /// then waits for the scheduler to exit. Failure of the task /// will abort the process. pub fn run_in_newsched_task(f: ~fn()) { - use super::sched::*; - use unstable::run_in_bare_thread; - let f = Cell::new(f); do run_in_bare_thread { let mut sched = ~new_test_uv_sched(); - let mut new_task = ~Task::new_root(); let on_exit: ~fn(bool) = |exit_status| rtassert!(exit_status); - new_task.on_exit = Some(on_exit); - let task = ~Coroutine::with_task(&mut sched.stack_pool, - new_task, - f.take()); + let mut task = ~Task::new_root(&mut sched.stack_pool, + f.take()); + rtdebug!("newsched_task: %x", to_uint(task)); + task.on_exit = Some(on_exit); sched.enqueue_task(task); sched.run(); } @@ -65,7 +58,6 @@ pub fn run_in_newsched_task(f: ~fn()) { pub fn run_in_mt_newsched_task(f: ~fn()) { use os; use from_str::FromStr; - use rt::uv::uvio::UvEventLoop; use rt::sched::Shutdown; use rt::util; @@ -90,7 +82,9 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { for uint::range(0, nthreads) |_| { let loop_ = ~UvEventLoop::new(); - let mut sched = ~Scheduler::new(loop_, work_queue.clone(), sleepers.clone()); + let mut sched = ~Scheduler::new(loop_, + work_queue.clone(), + sleepers.clone()); let handle = sched.make_handle(); handles.push(handle); @@ -99,9 +93,7 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { let f_cell = Cell::new(f_cell.take()); let handles = Cell::new(handles); - let mut new_task = ~Task::new_root(); let on_exit: ~fn(bool) = |exit_status| { - let mut handles = handles.take(); // Tell schedulers to exit for handles.mut_iter().advance |handle| { @@ -110,9 +102,9 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { rtassert!(exit_status); }; - new_task.on_exit = Some(on_exit); - let main_task = ~Coroutine::with_task(&mut scheds[0].stack_pool, - new_task, f_cell.take()); + let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, + f_cell.take()); + main_task.on_exit = Some(on_exit); scheds[0].enqueue_task(main_task); let mut threads = ~[]; @@ -134,144 +126,44 @@ pub fn run_in_mt_newsched_task(f: ~fn()) { } -// THIS IS AWFUL. Copy-pasted the above initialization function but -// with a number of hacks to make it spawn tasks on a variety of -// schedulers with a variety of homes using the new spawn. - -pub fn run_in_mt_newsched_task_random_homed() { - use libc; - use os; - use from_str::FromStr; - use rt::uv::uvio::UvEventLoop; - use rt::sched::Shutdown; - - do run_in_bare_thread { - let nthreads = match os::getenv("RUST_TEST_THREADS") { - Some(nstr) => FromStr::from_str(nstr).get(), - None => unsafe { - // Using more threads than cores in test code to force - // the OS to preempt them frequently. Assuming that - // this help stress test concurrent types. - rust_get_num_cpus() * 2 - } - }; - - let sleepers = SleeperList::new(); - let work_queue = WorkQueue::new(); - - let mut handles = ~[]; - let mut scheds = ~[]; - - // create a few special schedulers, those with even indicies - // will be pinned-only - for uint::range(0, nthreads) |i| { - let special = (i % 2) == 0; - let loop_ = ~UvEventLoop::new(); - let mut sched = ~Scheduler::new_special( - loop_, work_queue.clone(), sleepers.clone(), special); - let handle = sched.make_handle(); - handles.push(handle); - scheds.push(sched); - } - - // Schedule a pile o tasks - let n = 5*stress_factor(); - for uint::range(0,n) |_i| { - rtdebug!("creating task: %u", _i); - let hf: ~fn() = || { assert!(true) }; - spawntask_homed(&mut scheds, hf); - } - - // Now we want another pile o tasks that do not ever run on a - // special scheduler, because they are normal tasks. Because - // we can we put these in the "main" task. - - let n = 5*stress_factor(); - - let f: ~fn() = || { - for uint::range(0,n) |_| { - let f: ~fn() = || { - // Borrow the scheduler we run on and check if it is - // privileged. - do Local::borrow:: |sched| { - assert!(sched.run_anything); - }; - }; - spawntask_random(f); - }; - }; - - let f_cell = Cell::new(f); - let handles = Cell::new(handles); - - rtdebug!("creating main task"); - - let main_task = ~do Coroutine::new_root(&mut scheds[0].stack_pool) { - f_cell.take()(); - let mut handles = handles.take(); - // Tell schedulers to exit - for handles.mut_iter().advance |handle| { - handle.send(Shutdown); - } - }; - - rtdebug!("queuing main task") - - scheds[0].enqueue_task(main_task); - - let mut threads = ~[]; - - while !scheds.is_empty() { - let sched = scheds.pop(); - let sched_cell = Cell::new(sched); - let thread = do Thread::start { - let sched = sched_cell.take(); - rtdebug!("running sched: %u", sched.sched_id()); - sched.run(); - }; - - threads.push(thread); - } - - rtdebug!("waiting on scheduler threads"); - - // Wait for schedulers - let _threads = threads; - } - - extern { - fn rust_get_num_cpus() -> libc::uintptr_t; - } -} - - /// Test tasks will abort on failure instead of unwinding pub fn spawntask(f: ~fn()) { use super::sched::*; + let f = Cell::new(f); - rtdebug!("spawntask taking the scheduler from TLS") - let task = do Local::borrow::() |running_task| { - ~running_task.new_child() + let task = unsafe { + let sched = Local::unsafe_borrow::(); + rtdebug!("spawntask taking the scheduler from TLS"); + + + do Local::borrow::() |running_task| { + ~running_task.new_child(&mut (*sched).stack_pool, f.take()) + } }; - let mut sched = Local::take::(); - let task = ~Coroutine::with_task(&mut sched.stack_pool, - task, f); + rtdebug!("new task pointer: %x", to_uint(task)); + + let sched = Local::take::(); rtdebug!("spawntask scheduling the new task"); sched.schedule_task(task); } + /// Create a new task and run it right now. Aborts on failure pub fn spawntask_immediately(f: ~fn()) { use super::sched::*; - let task = do Local::borrow::() |running_task| { - ~running_task.new_child() + let f = Cell::new(f); + + let task = unsafe { + let sched = Local::unsafe_borrow::(); + do Local::borrow::() |running_task| { + ~running_task.new_child(&mut (*sched).stack_pool, + f.take()) + } }; - let mut sched = Local::take::(); - let task = ~Coroutine::with_task(&mut sched.stack_pool, - task, f); + let sched = Local::take::(); do sched.switch_running_tasks_and_then(task) |sched, task| { sched.enqueue_task(task); } @@ -280,15 +172,16 @@ pub fn spawntask_immediately(f: ~fn()) { /// Create a new task and run it right now. Aborts on failure pub fn spawntask_later(f: ~fn()) { use super::sched::*; + let f = Cell::new(f); - let task = do Local::borrow::() |running_task| { - ~running_task.new_child() + let task = unsafe { + let sched = Local::unsafe_borrow::(); + do Local::borrow::() |running_task| { + ~running_task.new_child(&mut (*sched).stack_pool, f.take()) + } }; let mut sched = Local::take::(); - let task = ~Coroutine::with_task(&mut sched.stack_pool, - task, f); - sched.enqueue_task(task); Local::put(sched); } @@ -298,13 +191,18 @@ pub fn spawntask_random(f: ~fn()) { use super::sched::*; use rand::{Rand, rng}; - let task = do Local::borrow::() |running_task| { - ~running_task.new_child() + let f = Cell::new(f); + + let task = unsafe { + let sched = Local::unsafe_borrow::(); + do Local::borrow::() |running_task| { + ~running_task.new_child(&mut (*sched).stack_pool, + f.take()) + + } }; let mut sched = Local::take::(); - let task = ~Coroutine::with_task(&mut sched.stack_pool, - task, f); let mut rng = rng(); let run_now: bool = Rand::rand(&mut rng); @@ -343,33 +241,49 @@ pub fn spawntask_homed(scheds: &mut ~[~Scheduler], f: ~fn()) { f() }; - ~Coroutine::with_task_homed(&mut sched.stack_pool, - ~Task::new_root(), - af, - Sched(handle)) + ~Task::new_root_homed(&mut sched.stack_pool, + Sched(handle), + af) }; let dest_sched = &mut scheds[rng.gen_int_range(0,scheds.len() as int)]; // enqueue it for future execution dest_sched.enqueue_task(task); } -/// Spawn a task and wait for it to finish, returning whether it completed successfully or failed +/// Spawn a task and wait for it to finish, returning whether it +/// completed successfully or failed pub fn spawntask_try(f: ~fn()) -> Result<(), ()> { use cell::Cell; use super::sched::*; + let f = Cell::new(f); + let (port, chan) = oneshot(); let chan = Cell::new(chan); - let mut new_task = ~Task::new_root(); let on_exit: ~fn(bool) = |exit_status| chan.take().send(exit_status); + let mut new_task = unsafe { + let sched = Local::unsafe_borrow::(); + do Local::borrow:: |_running_task| { + + // I don't understand why using a child task here fails. I + // think the fail status is propogating back up the task + // tree and triggering a fail for the parent, which we + // aren't correctly expecting. + + // ~running_task.new_child(&mut (*sched).stack_pool, + ~Task::new_root(&mut (*sched).stack_pool, + f.take()) + } + }; new_task.on_exit = Some(on_exit); - let mut sched = Local::take::(); - let new_task = ~Coroutine::with_task(&mut sched.stack_pool, - new_task, f); + + let sched = Local::take::(); do sched.switch_running_tasks_and_then(new_task) |sched, old_task| { sched.enqueue_task(old_task); } + rtdebug!("enqueued the new task, now waiting on exit_status"); + let exit_status = port.recv(); if exit_status { Ok(()) } else { Err(()) } } @@ -378,23 +292,27 @@ pub fn spawntask_try(f: ~fn()) -> Result<(), ()> { pub fn spawntask_thread(f: ~fn()) -> Thread { use rt::sched::*; - let task = do Local::borrow::() |running_task| { - ~running_task.new_child() + let f = Cell::new(f); + + let task = unsafe { + let sched = Local::unsafe_borrow::(); + do Local::borrow::() |running_task| { + ~running_task.new_child(&mut (*sched).stack_pool, + f.take()) + } }; let task = Cell::new(task); - let f = Cell::new(f); + let thread = do Thread::start { let mut sched = ~new_test_uv_sched(); - let task = ~Coroutine::with_task(&mut sched.stack_pool, - task.take(), - f.take()); - sched.enqueue_task(task); + sched.enqueue_task(task.take()); sched.run(); }; return thread; } + /// Get a port number, starting at 9600, for use in tests pub fn next_test_port() -> u16 { unsafe { @@ -410,7 +328,8 @@ pub fn next_test_ip4() -> IpAddr { Ipv4(127, 0, 0, 1, next_test_port()) } -/// Get a constant that represents the number of times to repeat stress tests. Default 1. +/// Get a constant that represents the number of times to repeat +/// stress tests. Default 1. pub fn stress_factor() -> uint { use os::getenv; diff --git a/src/libstd/rt/tube.rs b/src/libstd/rt/tube.rs index 89f3d10b5e4cf..013eb438c3657 100644 --- a/src/libstd/rt/tube.rs +++ b/src/libstd/rt/tube.rs @@ -16,14 +16,15 @@ use option::*; use clone::Clone; use super::rc::RC; -use rt::sched::{Scheduler, Coroutine}; +use rt::sched::Scheduler; use rt::{context, TaskContext, SchedulerContext}; use rt::local::Local; +use rt::task::Task; use vec::OwnedVector; use container::Container; struct TubeState { - blocked_task: Option<~Coroutine>, + blocked_task: Option<~Task>, buf: ~[T] } diff --git a/src/libstd/rt/uv/uvio.rs b/src/libstd/rt/uv/uvio.rs index 298277b3df05b..b1708e70733ea 100644 --- a/src/libstd/rt/uv/uvio.rs +++ b/src/libstd/rt/uv/uvio.rs @@ -29,7 +29,10 @@ use unstable::sync::{Exclusive, exclusive}; #[cfg(test)] use container::Container; #[cfg(test)] use uint; #[cfg(test)] use unstable::run_in_bare_thread; -#[cfg(test)] use rt::test::*; +#[cfg(test)] use rt::test::{spawntask_immediately, + next_test_ip4, + run_in_newsched_task}; + pub struct UvEventLoop { uvio: UvIoFactory diff --git a/src/libstd/task/mod.rs b/src/libstd/task/mod.rs index 99858feab224c..b0fc6b2884f01 100644 --- a/src/libstd/task/mod.rs +++ b/src/libstd/task/mod.rs @@ -1182,3 +1182,4 @@ fn test_simple_newsched_spawn() { spawn(||()) } } + diff --git a/src/libstd/task/spawn.rs b/src/libstd/task/spawn.rs index 63eb768d1c9cd..aea8cda6a2102 100644 --- a/src/libstd/task/spawn.rs +++ b/src/libstd/task/spawn.rs @@ -581,13 +581,20 @@ pub fn spawn_raw(opts: TaskOpts, f: ~fn()) { fn spawn_raw_newsched(mut opts: TaskOpts, f: ~fn()) { use rt::sched::*; - let mut task = if opts.linked { - do Local::borrow::() |running_task| { - ~running_task.new_child() + let f = Cell::new(f); + + let mut task = unsafe { + let sched = Local::unsafe_borrow::(); + rtdebug!("unsafe borrowed sched"); + + if opts.linked { + do Local::borrow::() |running_task| { + ~running_task.new_child(&mut (*sched).stack_pool, f.take()) + } + } else { + // An unlinked task is a new root in the task tree + ~Task::new_root(&mut (*sched).stack_pool, f.take()) } - } else { - // An unlinked task is a new root in the task tree - ~Task::new_root() }; if opts.notify_chan.is_some() { @@ -601,9 +608,13 @@ fn spawn_raw_newsched(mut opts: TaskOpts, f: ~fn()) { task.on_exit = Some(on_exit); } + rtdebug!("spawn about to take scheduler"); + let mut sched = Local::take::(); - let task = ~Coroutine::with_task(&mut sched.stack_pool, - task, f); + rtdebug!("took sched in spawn"); +// let task = ~Coroutine::with_task(&mut sched.stack_pool, +// task, f); +// let task = ~Task::new_root(&mut sched.stack_pool, f); sched.schedule_task(task); } diff --git a/src/libstd/unstable/lang.rs b/src/libstd/unstable/lang.rs index f750b31a466b9..3a071af5d4cfd 100644 --- a/src/libstd/unstable/lang.rs +++ b/src/libstd/unstable/lang.rs @@ -23,6 +23,7 @@ use option::{Option, Some, None}; use io; use rt::global_heap; use rt::borrowck; +use borrow::to_uint; #[allow(non_camel_case_types)] pub type rust_task = c_void; @@ -90,6 +91,9 @@ pub unsafe fn local_malloc(td: *c_char, size: uintptr_t) -> *c_char { _ => { let mut alloc = ::ptr::null(); do Local::borrow:: |task| { + rtdebug!("task pointer: %x, heap pointer: %x", + to_uint(task), + to_uint(&task.heap)); alloc = task.heap.alloc(td as *c_void, size as uint) as *c_char; } return alloc;