mikros/rust
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'toddaaro/io' into io

Browse Source
This commit is contained in:
Brian Anderson 2013-06-15 17:00:36 -07:00
commit b08c446798
12 changed files with 806 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
src/libstd/macros.rs
View File

@ -38,16 +38,18 @@ macro_rules! rtassert (
} )
)
// The do_abort function was originally inside the abort macro, but
// this was ICEing the compiler so it has been moved outside. Now this
// seems to work?
pub fn do_abort() -> ! {
unsafe { ::libc::abort(); }
}
macro_rules! abort(
($( $msg:expr),+) => ( {
rtdebug!($($msg),+);
do_abort();
// NB: This is in a fn to avoid putting the `unsafe` block in a macro,
// which causes spurious 'unnecessary unsafe block' warnings.
fn do_abort() -> ! {
unsafe { ::libc::abort(); }
}
::macros::do_abort();
} )
)

4
src/libstd/rt/comm.rs
View File

@ -120,13 +120,13 @@ impl<T> ChanOne<T> {
match oldstate {
STATE_BOTH => {
// Port is not waiting yet. Nothing to do
do Local::borrow::<Scheduler> |sched| {
do Local::borrow::<Scheduler, ()> |sched| {
rtdebug!("non-rendezvous send");
sched.metrics.non_rendezvous_sends += 1;
}
}
STATE_ONE => {
do Local::borrow::<Scheduler> |sched| {
do Local::borrow::<Scheduler, ()> |sched| {
rtdebug!("rendezvous send");
sched.metrics.rendezvous_sends += 1;
}

35
src/libstd/rt/local.rs
View File

@ -18,7 +18,7 @@ pub trait Local {
fn put(value: ~Self);
fn take() -> ~Self;
fn exists() -> bool;
fn borrow(f: &fn(&mut Self));
fn borrow<T>(f: &fn(&mut Self) -> T) -> T;
unsafe fn unsafe_borrow() -> *mut Self;
unsafe fn try_unsafe_borrow() -> Option<*mut Self>;
}
@ -27,7 +27,20 @@ impl Local for Scheduler {
fn put(value: ~Scheduler) { unsafe { local_ptr::put(value) }}
fn take() -> ~Scheduler { unsafe { local_ptr::take() } }
fn exists() -> bool { local_ptr::exists() }
fn borrow(f: &fn(&mut Scheduler)) { unsafe { local_ptr::borrow(f) } }
fn borrow<T>(f: &fn(&mut Scheduler) -> T) -> T {
let mut res: Option<T> = None;
let res_ptr: *mut Option<T> = &mut res;
unsafe {
do local_ptr::borrow |sched| {
let result = f(sched);
*res_ptr = Some(result);
}
}
match res {
Some(r) => { r }
None => abort!("function failed!")
}
}
unsafe fn unsafe_borrow() -> *mut Scheduler { local_ptr::unsafe_borrow() }
unsafe fn try_unsafe_borrow() -> Option<*mut Scheduler> { abort!("unimpl") }
}
@ -36,8 +49,8 @@ impl Local for Task {
fn put(_value: ~Task) { abort!("unimpl") }
fn take() -> ~Task { abort!("unimpl") }
fn exists() -> bool { abort!("unimpl") }
fn borrow(f: &fn(&mut Task)) {
do Local::borrow::<Scheduler> |sched| {
fn borrow<T>(f: &fn(&mut Task) -> T) -> T {
do Local::borrow::<Scheduler, T> |sched| {
match sched.current_task {
Some(~ref mut task) => {
f(&mut *task.task)
@ -74,7 +87,7 @@ impl Local for IoFactoryObject {
fn put(_value: ~IoFactoryObject) { abort!("unimpl") }
fn take() -> ~IoFactoryObject { abort!("unimpl") }
fn exists() -> bool { abort!("unimpl") }
fn borrow(_f: &fn(&mut IoFactoryObject)) { abort!("unimpl") }
fn borrow<T>(_f: &fn(&mut IoFactoryObject) -> T) -> T { abort!("unimpl") }
unsafe fn unsafe_borrow() -> *mut IoFactoryObject {
let sched = Local::unsafe_borrow::<Scheduler>();
let io: *mut IoFactoryObject = (*sched).event_loop.io().unwrap();
@ -115,4 +128,16 @@ mod test {
}
let _scheduler: ~Scheduler = Local::take();
}
#[test]
fn borrow_with_return() {
let scheduler = ~new_test_uv_sched();
Local::put(scheduler);
let res = do Local::borrow::<Scheduler,bool> |_sched| {
true
};
assert!(res)
let _scheduler: ~Scheduler = Local::take();
}
}

2
src/libstd/rt/mod.rs
View File

@ -208,7 +208,7 @@ pub fn context() -> RuntimeContext {
} else {
if Local::exists::<Scheduler>() {
let context = ::cell::empty_cell();
do Local::borrow::<Scheduler> |sched| {
do Local::borrow::<Scheduler, ()> |sched| {
if sched.in_task_context() {
context.put_back(TaskContext);
} else {

659
src/libstd/rt/sched.rs
View File

@ -26,6 +26,11 @@ 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 ptr::{to_uint};
/// The Scheduler is responsible for coordinating execution of Coroutines
/// on a single thread. When the scheduler is running it is owned by
/// thread local storage and the running task is owned by the
@ -65,12 +70,15 @@ pub struct Scheduler {
/// An action performed after a context switch on behalf of the
/// code running before the context switch
priv cleanup_job: Option<CleanupJob>,
metrics: SchedMetrics
metrics: SchedMetrics,
/// Should this scheduler run any task, or only pinned tasks?
run_anything: bool
}
pub struct SchedHandle {
priv remote: ~RemoteCallbackObject,
priv queue: MessageQueue<SchedMessage>
priv queue: MessageQueue<SchedMessage>,
sched_id: uint
}
pub struct Coroutine {
@ -81,12 +89,20 @@ pub struct Coroutine {
/// the task is dead
priv saved_context: Context,
/// The heap, GC, unwinding, local storage, logging
task: ~Task
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
Shutdown,
PinnedTask(~Coroutine)
}
enum CleanupJob {
@ -96,6 +112,8 @@ enum CleanupJob {
pub impl Scheduler {
pub fn sched_id(&self) -> uint { to_uint(self) }
fn in_task_context(&self) -> bool { self.current_task.is_some() }
fn new(event_loop: ~EventLoopObject,
@ -103,6 +121,16 @@ pub impl Scheduler {
sleeper_list: SleeperList)
-> Scheduler {
Scheduler::new_special(event_loop, work_queue, sleeper_list, true)
}
fn new_special(event_loop: ~EventLoopObject,
work_queue: WorkQueue<~Coroutine>,
sleeper_list: SleeperList,
run_anything: bool)
-> Scheduler {
// Lazily initialize the runtime TLS key
local_ptr::init_tls_key();
@ -117,7 +145,8 @@ pub impl Scheduler {
saved_context: Context::empty(),
current_task: None,
cleanup_job: None,
metrics: SchedMetrics::new()
metrics: SchedMetrics::new(),
run_anything: run_anything
}
}
@ -147,11 +176,13 @@ pub impl Scheduler {
(*event_loop).run();
}
rtdebug!("run taking sched");
let sched = Local::take::<Scheduler>();
// XXX: Reenable this once we're using a per-task queue. With a shared
// queue this is not true
//assert!(sched.work_queue.is_empty());
rtdebug!("scheduler metrics: %s\n", sched.metrics.to_str());
// let out = sched.metrics.to_str();
// rtdebug!("scheduler metrics: %s\n", out);
return sched;
}
@ -167,6 +198,7 @@ pub impl Scheduler {
if sched.interpret_message_queue() {
// We performed a scheduling action. There may be other work
// to do yet, so let's try again later.
rtdebug!("run_sched_once, interpret_message_queue taking sched");
let mut sched = Local::take::<Scheduler>();
sched.metrics.messages_received += 1;
sched.event_loop.callback(Scheduler::run_sched_once);
@ -175,6 +207,7 @@ pub impl Scheduler {
}
// Now, look in the work queue for tasks to run
rtdebug!("run_sched_once taking");
let sched = Local::take::<Scheduler>();
if sched.resume_task_from_queue() {
// We performed a scheduling action. There may be other work
@ -209,33 +242,56 @@ pub impl Scheduler {
return SchedHandle {
remote: remote,
queue: self.message_queue.clone()
queue: self.message_queue.clone(),
sched_id: self.sched_id()
};
}
/// Schedule a task to be executed later.
///
/// 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.
fn enqueue_task(&mut self, task: ~Coroutine) {
self.work_queue.push(task);
self.event_loop.callback(Scheduler::run_sched_once);
/// 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.
// We've made work available. Notify a sleeping scheduler.
// XXX: perf. Check for a sleeper without synchronizing memory.
// It's not critical that we always find it.
// XXX: perf. If there's a sleeper then we might as well just send
// it the task directly instead of pushing it to the
// queue. That is essentially the intent here and it is less
// work.
match self.sleeper_list.pop() {
fn enqueue_task(&mut self, mut task: ~Coroutine) {
// We don't want to queue tasks that belong on other threads,
// so we send them home at enqueue time.
// The borrow checker doesn't like our disassembly of the
// Coroutine struct and partial use and mutation of the
// fields. So completely disassemble here and stop using?
// XXX perf: I think we might be able to shuffle this code to
// only destruct when we need to.
rtdebug!("a task was queued on: %u", self.sched_id());
let this = self;
// We push the task onto our local queue clone.
this.work_queue.push(task);
this.event_loop.callback(Scheduler::run_sched_once);
// We've made work available. Notify a
// sleeping scheduler.
// XXX: perf. Check for a sleeper without
// synchronizing memory. It's not critical
// that we always find it.
// XXX: perf. If there's a sleeper then we
// might as well just send it the task
// directly instead of pushing it to the
// queue. That is essentially the intent here
// and it is less work.
match this.sleeper_list.pop() {
Some(handle) => {
let mut handle = handle;
handle.send(Wake)
}
None => (/* pass */)
}
None => { (/* pass */) }
};
}
// * Scheduler-context operations
@ -247,6 +303,15 @@ pub impl Scheduler {
let mut this = self;
match this.message_queue.pop() {
Some(PinnedTask(task)) => {
rtdebug!("recv BiasedTask message in sched: %u",
this.sched_id());
let mut task = task;
task.task.home = Some(Sched(this.make_handle()));
this.resume_task_immediately(task);
return true;
}
Some(Wake) => {
rtdebug!("recv Wake message");
this.sleepy = false;
@ -256,8 +321,9 @@ pub impl Scheduler {
Some(Shutdown) => {
rtdebug!("recv Shutdown message");
if this.sleepy {
// There may be an outstanding handle on the sleeper list.
// Pop them all to make sure that's not the case.
// There may be an outstanding handle on the
// sleeper list. Pop them all to make sure that's
// not the case.
loop {
match this.sleeper_list.pop() {
Some(handle) => {
@ -268,8 +334,8 @@ pub impl Scheduler {
}
}
}
// No more sleeping. After there are no outstanding event loop
// references we will shut down.
// No more sleeping. After there are no outstanding
// event loop references we will shut down.
this.no_sleep = true;
this.sleepy = false;
Local::put(this);
@ -282,23 +348,93 @@ pub impl Scheduler {
}
}
/// Given an input Coroutine sends it back to its home scheduler.
fn send_task_home(task: ~Coroutine) {
let mut task = task;
let mut home = task.task.home.swap_unwrap();
match home {
Sched(ref mut home_handle) => {
home_handle.send(PinnedTask(task));
}
AnySched => {
abort!("error: cannot send anysched task home");
}
}
}
// Resume a task from the queue - but also take into account that
// it might not belong here.
fn resume_task_from_queue(~self) -> bool {
assert!(!self.in_task_context());
rtdebug!("looking in work queue for task to schedule");
let mut this = self;
// The borrow checker imposes the possibly absurd requirement
// that we split this into two match expressions. This is due
// to the inspection of the internal bits of task, as that
// can't be in scope when we act on task.
match this.work_queue.pop() {
Some(task) => {
rtdebug!("resuming task from work queue");
this.resume_task_immediately(task);
return true;
let action_id = {
let home = &task.task.home;
match home {
&Some(Sched(ref home_handle))
if home_handle.sched_id != this.sched_id() => {
0
}
&Some(AnySched) if this.run_anything => {
1
}
&Some(AnySched) => {
2
}
&Some(Sched(_)) => {
3
}
&None => {
4
}
}
};
match action_id {
0 => {
rtdebug!("sending task home");
Scheduler::send_task_home(task);
Local::put(this);
return false;
}
1 => {
rtdebug!("resuming now");
this.resume_task_immediately(task);
return true;
}
2 => {
rtdebug!("re-queueing")
this.enqueue_task(task);
Local::put(this);
return false;
}
3 => {
rtdebug!("resuming now");
this.resume_task_immediately(task);
return true;
}
4 => {
abort!("task home was None!");
}
_ => {
abort!("literally, you should not be here");
}
}
}
None => {
rtdebug!("no tasks in queue");
Local::put(this);
return false;
}
rtdebug!("no tasks in queue");
Local::put(this);
return false;
}
}
}
@ -319,21 +455,32 @@ pub impl Scheduler {
abort!("control reached end of task");
}
fn schedule_new_task(~self, task: ~Coroutine) {
pub fn schedule_task(~self, task: ~Coroutine) {
assert!(self.in_task_context());
do self.switch_running_tasks_and_then(task) |sched, last_task| {
let last_task = Cell(last_task);
sched.enqueue_task(last_task.take());
}
}
// is the task home?
let is_home = task.is_home_no_tls(&self);
fn schedule_task(~self, task: ~Coroutine) {
assert!(self.in_task_context());
// does the task have a home?
let homed = task.homed();
do self.switch_running_tasks_and_then(task) |sched, last_task| {
let last_task = Cell(last_task);
sched.enqueue_task(last_task.take());
let mut this = self;
if is_home || (!homed && this.run_anything) {
// here we know we are home, execute now OR we know we
// aren't homed, and that this sched doesn't care
do this.switch_running_tasks_and_then(task) |sched, last_task| {
let last_task = Cell(last_task);
sched.enqueue_task(last_task.take());
}
} else if !homed && !this.run_anything {
// the task isn't homed, but it can't be run here
this.enqueue_task(task);
Local::put(this);
} else {
// task isn't home, so don't run it here, send it home
Scheduler::send_task_home(task);
Local::put(this);
}
}
@ -515,13 +662,83 @@ impl SchedHandle {
}
pub impl Coroutine {
/// This function checks that a coroutine is running "home".
fn is_home(&self) -> bool {
rtdebug!("checking if coroutine is home");
do Local::borrow::<Scheduler,bool> |sched| {
match self.task.home {
Some(AnySched) => { false }
Some(Sched(SchedHandle { sched_id: ref id, _ })) => {
*id == sched.sched_id()
}
None => { abort!("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::<Scheduler,bool> |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 => { abort!("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 => { abort!("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::<Scheduler,bool>() |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.
fn new_homed(stack_pool: &mut StackPool, home: SchedHome, start: ~fn()) -> Coroutine {
Coroutine::with_task_homed(stack_pool, ~Task::new(), start, home)
}
fn new(stack_pool: &mut StackPool, start: ~fn()) -> Coroutine {
Coroutine::with_task(stack_pool, ~Task::new(), start)
}
fn with_task(stack_pool: &mut StackPool,
task: ~Task,
start: ~fn()) -> Coroutine {
fn with_task_homed(stack_pool: &mut StackPool,
task: ~Task,
start: ~fn(),
home: SchedHome) -> Coroutine {
static MIN_STACK_SIZE: uint = 10000000; // XXX: Too much stack
@ -529,11 +746,22 @@ pub impl Coroutine {
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);
return Coroutine {
let mut crt = Coroutine {
current_stack_segment: stack,
saved_context: initial_context,
task: task
task: task,
};
crt.task.home = Some(home);
return crt;
}
fn with_task(stack_pool: &mut StackPool,
task: ~Task,
start: ~fn()) -> Coroutine {
Coroutine::with_task_homed(stack_pool,
task,
start,
AnySched)
}
priv fn build_start_wrapper(start: ~fn()) -> ~fn() {
@ -549,17 +777,20 @@ pub impl Coroutine {
let sched = Local::unsafe_borrow::<Scheduler>();
let task = (*sched).current_task.get_mut_ref();
// FIXME #6141: shouldn't neet to put `start()` in another closure
// FIXME #6141: shouldn't neet to put `start()` in
// another closure
let start_cell = Cell(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.
// 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();
};
@ -603,6 +834,305 @@ mod test {
use rt::test::*;
use super::*;
use rt::thread::Thread;
use ptr::to_uint;
// 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 {
let sched = ~new_test_uv_sched();
assert!(to_uint(sched) == sched.sched_id());
}
}
// 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 {
let sched_one = ~new_test_uv_sched();
let sched_two = ~new_test_uv_sched();
assert!(sched_one.sched_id() != sched_two.sched_id());
}
}
// 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 {
let mut task_ran = false;
let task_ran_ptr: *mut bool = &mut task_ran;
let mut sched = ~new_test_uv_sched();
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)) {
unsafe { *task_ran_ptr = true };
let sched = Local::take::<Scheduler>();
assert!(sched.sched_id() == sched_id);
Local::put::<Scheduler>(sched);
};
sched.enqueue_task(task);
sched.run();
assert!(task_ran);
}
}
// A test for each state of schedule_task
#[test]
fn test_schedule_home_states() {
use rt::uv::uvio::UvEventLoop;
use rt::sched::Shutdown;
use rt::sleeper_list::SleeperList;
use rt::work_queue::WorkQueue;
do run_in_bare_thread {
// let nthreads = 2;
let sleepers = SleeperList::new();
let work_queue = WorkQueue::new();
// our normal scheduler
let mut normal_sched = ~Scheduler::new(
~UvEventLoop::new(),
work_queue.clone(),
sleepers.clone());
let normal_handle = Cell(normal_sched.make_handle());
// our special scheduler
let mut special_sched = ~Scheduler::new_special(
~UvEventLoop::new(),
work_queue.clone(),
sleepers.clone(),
true);
let special_handle = Cell(special_sched.make_handle());
let special_handle2 = Cell(special_sched.make_handle());
let special_id = special_sched.sched_id();
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);
rtdebug!("t1 should be home: %b", is_home);
assert!(is_home);
};
let t1f = Cell(t1f);
let t2f = ~do Coroutine::new(&mut normal_sched.stack_pool) {
let on_special = Coroutine::on_special();
rtdebug!("t2 should not be on special: %b", on_special);
assert!(!on_special);
};
let t2f = Cell(t2f);
let t3f = ~do Coroutine::new(&mut normal_sched.stack_pool) {
// not on special
let on_special = Coroutine::on_special();
rtdebug!("t3 should not be on special: %b", on_special);
assert!(!on_special);
};
let t3f = Cell(t3f);
let t4f = ~do Coroutine::new_homed(&mut special_sched.stack_pool,
Sched(t4_handle)) {
// is home
let home = Coroutine::is_home_using_id(special_id);
rtdebug!("t4 should be home: %b", home);
assert!(home);
};
let t4f = Cell(t4f);
// we have four tests, make them as closures
let t1: ~fn() = || {
// task is home on special
let task = t1f.take();
let sched = Local::take::<Scheduler>();
sched.schedule_task(task);
};
let t2: ~fn() = || {
// not homed, task doesn't care
let task = t2f.take();
let sched = Local::take::<Scheduler>();
sched.schedule_task(task);
};
let t3: ~fn() = || {
// task not homed, must leave
let task = t3f.take();
let sched = Local::take::<Scheduler>();
sched.schedule_task(task);
};
let t4: ~fn() = || {
// task not home, send home
let task = t4f.take();
let sched = Local::take::<Scheduler>();
sched.schedule_task(task);
};
let t1 = Cell(t1);
let t2 = Cell(t2);
let t3 = Cell(t3);
let t4 = Cell(t4);
// build a main task that runs our four tests
let main_task = ~do Coroutine::new(&mut normal_sched.stack_pool) {
// the two tasks that require a normal start location
t2.take()();
t4.take()();
normal_handle.take().send(Shutdown);
special_handle.take().send(Shutdown);
};
// task to run the two "special start" tests
let special_task = ~do Coroutine::new_homed(
&mut special_sched.stack_pool,
Sched(special_handle2.take())) {
t1.take()();
t3.take()();
};
// enqueue the main tasks
normal_sched.enqueue_task(special_task);
normal_sched.enqueue_task(main_task);
let nsched_cell = Cell(normal_sched);
let normal_thread = do Thread::start {
let sched = nsched_cell.take();
sched.run();
};
let ssched_cell = Cell(special_sched);
let special_thread = do Thread::start {
let sched = ssched_cell.take();
sched.run();
};
// wait for the end
let _thread1 = normal_thread;
let _thread2 = special_thread;
}
}
// 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 old_iter::MutableIter;
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(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::<Scheduler>();
rtdebug!("run location scheduler id: %u, home: %u",
sched.sched_id(),
home_id);
assert!(sched.sched_id() == home_id);
Local::put::<Scheduler>(sched);
let mut handles = handles.take();
for handles.each_mut |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(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;
for int::range(0,n as int) |_| {
test_schedule_home_states();
}
}
// 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() {
@ -683,7 +1213,7 @@ mod test {
assert_eq!(count, MAX);
fn run_task(count_ptr: *mut int) {
do Local::borrow::<Scheduler> |sched| {
do Local::borrow::<Scheduler, ()> |sched| {
let task = ~do Coroutine::new(&mut sched.stack_pool) {
unsafe {
*count_ptr = *count_ptr + 1;
@ -859,8 +1389,8 @@ mod test {
fn start_closure_dtor() {
use ops::Drop;
// Regression test that the `start` task entrypoint can contain dtors
// that use task resources
// Regression test that the `start` task entrypoint can
// contain dtors that use task resources
do run_in_newsched_task {
struct S { field: () }
@ -875,6 +1405,7 @@ mod test {
do spawntask {
let _ss = &s;
}
}
}
}
}

18
src/libstd/rt/task.rs
View File

@ -19,6 +19,7 @@ use cast::transmute;
use rt::local::Local;
use super::local_heap::LocalHeap;
use rt::logging::StdErrLogger;
use rt::sched::{SchedHome, AnySched};
pub struct Task {
heap: LocalHeap,
@ -26,7 +27,8 @@ pub struct Task {
storage: LocalStorage,
logger: StdErrLogger,
unwinder: Option<Unwinder>,
destroyed: bool
destroyed: bool,
home: Option<SchedHome>
}
pub struct GarbageCollector;
@ -44,7 +46,8 @@ impl Task {
storage: LocalStorage(ptr::null(), None),
logger: StdErrLogger,
unwinder: Some(Unwinder { unwinding: false }),
destroyed: false
destroyed: false,
home: Some(AnySched)
}
}
@ -55,14 +58,19 @@ impl Task {
storage: LocalStorage(ptr::null(), None),
logger: StdErrLogger,
unwinder: None,
destroyed: false
destroyed: false,
home: Some(AnySched)
}
}
pub fn give_home(&mut self, new_home: SchedHome) {
self.home = Some(new_home);
}
pub fn run(&mut self, f: &fn()) {
// This is just an assertion that `run` was called unsafely
// and this instance of Task is still accessible.
do Local::borrow::<Task> |task| {
do Local::borrow::<Task, ()> |task| {
assert!(ptr::ref_eq(task, self));
}
@ -87,7 +95,7 @@ impl Task {
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> |task| {
do Local::borrow::<Task, ()> |task| {
assert!(ptr::ref_eq(task, self));
}
match self.storage {

150
src/libstd/rt/test.rs
View File

@ -88,6 +88,7 @@ pub fn run_in_mt_newsched_task(f: ~fn()) {
let loop_ = ~UvEventLoop::new();
let mut sched = ~Scheduler::new(loop_, work_queue.clone(), sleepers.clone());
let handle = sched.make_handle();
handles.push(handle);
scheds.push(sched);
}
@ -128,15 +129,128 @@ 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::<Scheduler,()> |sched| {
assert!(sched.run_anything);
};
};
spawntask_random(f);
};
};
let f_cell = Cell(f);
let handles = Cell(handles);
rtdebug!("creating main task");
let main_task = ~do Coroutine::new(&mut scheds[0].stack_pool) {
f_cell.take()();
let mut handles = handles.take();
// Tell schedulers to exit
for handles.each_mut |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(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::*;
rtdebug!("spawntask taking the scheduler from TLS")
let mut sched = Local::take::<Scheduler>();
let task = ~Coroutine::with_task(&mut sched.stack_pool,
~Task::without_unwinding(),
f);
sched.schedule_new_task(task);
rtdebug!("spawntask scheduling the new task");
sched.schedule_task(task);
}
/// Create a new task and run it right now. Aborts on failure
@ -188,6 +302,39 @@ pub fn spawntask_random(f: ~fn()) {
}
}
/// Spawn a task, with the current scheduler as home, and queue it to
/// run later.
pub fn spawntask_homed(scheds: &mut ~[~Scheduler], f: ~fn()) {
use super::sched::*;
use rand::{rng, RngUtil};
let mut rng = rng();
let task = {
let sched = &mut scheds[rng.gen_int_range(0,scheds.len() as int)];
let handle = sched.make_handle();
let home_id = handle.sched_id;
// now that we know where this is going, build a new function
// that can assert it is in the right place
let af: ~fn() = || {
do Local::borrow::<Scheduler,()>() |sched| {
rtdebug!("home_id: %u, runtime loc: %u",
home_id,
sched.sched_id());
assert!(home_id == sched.sched_id());
};
f()
};
~Coroutine::with_task_homed(&mut sched.stack_pool,
~Task::without_unwinding(),
af,
Sched(handle))
};
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
pub fn spawntask_try(f: ~fn()) -> Result<(), ()> {
@ -266,3 +413,4 @@ pub fn stress_factor() -> uint {
}
}

2
src/libstd/rt/tube.rs
View File

@ -155,7 +155,7 @@ mod test {
if i == 100 { return; }
let tube = Cell(Cell(tube));
do Local::borrow::<Scheduler> |sched| {
do Local::borrow::<Scheduler, ()> |sched| {
let tube = tube.take();
do sched.event_loop.callback {
let mut tube = tube.take();

2
src/libstd/rt/uv/uvio.rs
View File

@ -167,7 +167,7 @@ mod test_remote {
let mut tube = Tube::new();
let tube_clone = tube.clone();
let remote_cell = cell::empty_cell();
do Local::borrow::<Scheduler>() |sched| {
do Local::borrow::<Scheduler, ()>() |sched| {
let tube_clone = tube_clone.clone();
let tube_clone_cell = Cell(tube_clone);
let remote = do sched.event_loop.remote_callback {

2
src/libstd/task/mod.rs
View File

@ -514,7 +514,7 @@ pub fn failing() -> bool {
}
_ => {
let mut unwinding = false;
do Local::borrow::<Task> |local| {
do Local::borrow::<Task, ()> |local| {
unwinding = match local.unwinder {
Some(unwinder) => {
unwinder.unwinding

2
src/libstd/task/spawn.rs
View File

@ -578,7 +578,7 @@ fn spawn_raw_newsched(_opts: TaskOpts, f: ~fn()) {
let mut sched = Local::take::<Scheduler>();
let task = ~Coroutine::new(&mut sched.stack_pool, f);
sched.schedule_new_task(task);
sched.schedule_task(task);
}
fn spawn_raw_oldsched(mut opts: TaskOpts, f: ~fn()) {

4
src/libstd/unstable/lang.rs
View File

@ -244,7 +244,7 @@ pub unsafe fn local_malloc(td: *c_char, size: uintptr_t) -> *c_char {
}
_ => {
let mut alloc = ::ptr::null();
do Local::borrow::<Task> |task| {
do Local::borrow::<Task,()> |task| {
alloc = task.heap.alloc(td as *c_void, size as uint) as *c_char;
}
return alloc;
@ -262,7 +262,7 @@ pub unsafe fn local_free(ptr: *c_char) {
rustrt::rust_upcall_free_noswitch(ptr);
}
_ => {
do Local::borrow::<Task> |task| {
do Local::borrow::<Task,()> |task| {
task.heap.free(ptr as *c_void);
}
}