b545751597
It turns out that the uv implementation would cause use-after-free if the idle callback was used after the call to `close`, and additionally nothing would ever really work that well if `start()` were called twice. To change this, the `start` and `close` methods were removed in favor of specifying the callback at creation, and allowing destruction to take care of closing the watcher.
1379 lines
47 KiB
Rust
1379 lines
47 KiB
Rust
// 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 <LICENSE-APACHE or
|
|
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
|
|
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
|
|
// option. This file may not be copied, modified, or distributed
|
|
// except according to those terms.
|
|
|
|
use either::{Left, Right};
|
|
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};
|
|
use super::rtio::EventLoop;
|
|
use super::context::Context;
|
|
use super::task::{Task, AnySched, Sched};
|
|
use super::message_queue::MessageQueue;
|
|
use rt::kill::BlockedTask;
|
|
use rt::local_ptr;
|
|
use rt::local::Local;
|
|
use rt::rtio::{RemoteCallback, PausibleIdleCallback, Callback};
|
|
use borrow::{to_uint};
|
|
use cell::Cell;
|
|
use rand::{XorShiftRng, Rng, Rand};
|
|
use iter::range;
|
|
use vec::{OwnedVector};
|
|
|
|
/// A scheduler is responsible for coordinating the execution of Tasks
|
|
/// on a single thread. The scheduler runs inside a slightly modified
|
|
/// Rust Task. When not running this task is stored in the scheduler
|
|
/// struct. The scheduler struct acts like a baton, all scheduling
|
|
/// actions are transfers of the baton.
|
|
///
|
|
/// XXX: This creates too many callbacks to run_sched_once, resulting
|
|
/// in too much allocation and too many events.
|
|
pub struct Scheduler {
|
|
/// 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.
|
|
priv message_queue: MessageQueue<SchedMessage>,
|
|
/// A shared list of sleeping schedulers. We'll use this to wake
|
|
/// up schedulers when pushing work onto the work queue.
|
|
sleeper_list: SleeperList,
|
|
/// Indicates that we have previously pushed a handle onto the
|
|
/// SleeperList but have not yet received the Wake message.
|
|
/// Being `true` does not necessarily mean that the scheduler is
|
|
/// not active since there are multiple event sources that may
|
|
/// wake the scheduler. It just prevents the scheduler from pushing
|
|
/// multiple handles onto the sleeper list.
|
|
priv sleepy: bool,
|
|
/// A flag to indicate we've received the shutdown message and should
|
|
/// no longer try to go to sleep, but exit instead.
|
|
no_sleep: bool,
|
|
stack_pool: StackPool,
|
|
/// The scheduler runs on a special task. When it is not running
|
|
/// it is stored here instead of the work queue.
|
|
priv sched_task: Option<~Task>,
|
|
/// An action performed after a context switch on behalf of the
|
|
/// code running before the context switch
|
|
priv cleanup_job: Option<CleanupJob>,
|
|
/// Should this scheduler run any task, or only pinned tasks?
|
|
run_anything: bool,
|
|
/// If the scheduler shouldn't run some tasks, a friend to send
|
|
/// them to.
|
|
priv friend_handle: Option<SchedHandle>,
|
|
/// A fast XorShift rng for scheduler use
|
|
rng: XorShiftRng,
|
|
/// A toggleable idle callback
|
|
priv idle_callback: Option<~PausibleIdleCallback>,
|
|
/// A countdown that starts at a random value and is decremented
|
|
/// every time a yield check is performed. When it hits 0 a task
|
|
/// will yield.
|
|
priv yield_check_count: uint,
|
|
/// A flag to tell the scheduler loop it needs to do some stealing
|
|
/// in order to introduce randomness as part of a yield
|
|
priv steal_for_yield: bool,
|
|
|
|
// n.b. currently destructors of an object are run in top-to-bottom in order
|
|
// of field declaration. Due to its nature, the pausible idle callback
|
|
// must have some sort of handle to the event loop, so it needs to get
|
|
// destroyed before the event loop itself. For this reason, we destroy
|
|
// the event loop last to ensure that any unsafe references to it are
|
|
// destroyed before it's actually destroyed.
|
|
|
|
/// The event loop used to drive the scheduler and perform I/O
|
|
event_loop: ~EventLoop,
|
|
}
|
|
|
|
/// An indication of how hard to work on a given operation, the difference
|
|
/// mainly being whether memory is synchronized or not
|
|
#[deriving(Eq)]
|
|
enum EffortLevel {
|
|
DontTryTooHard,
|
|
GiveItYourBest
|
|
}
|
|
|
|
static MAX_YIELD_CHECKS: uint = 200;
|
|
|
|
fn reset_yield_check(rng: &mut XorShiftRng) -> uint {
|
|
let r: uint = Rand::rand(rng);
|
|
r % MAX_YIELD_CHECKS + 1
|
|
}
|
|
|
|
impl Scheduler {
|
|
|
|
// * Initialization Functions
|
|
|
|
pub fn new(event_loop: ~EventLoop,
|
|
work_queue: WorkQueue<~Task>,
|
|
work_queues: ~[WorkQueue<~Task>],
|
|
sleeper_list: SleeperList)
|
|
-> Scheduler {
|
|
|
|
Scheduler::new_special(event_loop, work_queue,
|
|
work_queues,
|
|
sleeper_list, true, None)
|
|
|
|
}
|
|
|
|
pub fn new_special(event_loop: ~EventLoop,
|
|
work_queue: WorkQueue<~Task>,
|
|
work_queues: ~[WorkQueue<~Task>],
|
|
sleeper_list: SleeperList,
|
|
run_anything: bool,
|
|
friend: Option<SchedHandle>)
|
|
-> Scheduler {
|
|
|
|
let mut sched = Scheduler {
|
|
sleeper_list: sleeper_list,
|
|
message_queue: MessageQueue::new(),
|
|
sleepy: false,
|
|
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,
|
|
run_anything: run_anything,
|
|
friend_handle: friend,
|
|
rng: new_sched_rng(),
|
|
idle_callback: None,
|
|
yield_check_count: 0,
|
|
steal_for_yield: false
|
|
};
|
|
|
|
sched.yield_check_count = reset_yield_check(&mut sched.rng);
|
|
|
|
return sched;
|
|
}
|
|
|
|
// XXX: This may eventually need to be refactored so that
|
|
// the scheduler itself doesn't have to call event_loop.run.
|
|
// That will be important for embedding the runtime into external
|
|
// event loops.
|
|
|
|
// Take a main task to run, and a scheduler to run it in. Create a
|
|
// scheduler task and bootstrap into it.
|
|
pub fn bootstrap(mut ~self, task: ~Task) {
|
|
|
|
// Build an Idle callback.
|
|
let cb = ~SchedRunner as ~Callback;
|
|
self.idle_callback = Some(self.event_loop.pausible_idle_callback(cb));
|
|
|
|
// Initialize the TLS key.
|
|
local_ptr::init_tls_key();
|
|
|
|
// Create a task for the scheduler with an empty context.
|
|
let sched_task = ~Task::new_sched_task();
|
|
|
|
// Now that we have an empty task struct for the scheduler
|
|
// task, put it in TLS.
|
|
Local::put(sched_task);
|
|
|
|
// Before starting our first task, make sure the idle callback
|
|
// is active. As we do not start in the sleep state this is
|
|
// important.
|
|
self.idle_callback.get_mut_ref().resume();
|
|
|
|
// Now, as far as all the scheduler state is concerned, we are
|
|
// inside the "scheduler" context. So we can act like the
|
|
// scheduler and resume the provided task.
|
|
self.resume_task_immediately(task);
|
|
|
|
// Now we are back in the scheduler context, having
|
|
// successfully run the input task. Start by running the
|
|
// scheduler. Grab it out of TLS - performing the scheduler
|
|
// action will have given it away.
|
|
let sched: ~Scheduler = Local::take();
|
|
|
|
rtdebug!("starting scheduler {}", sched.sched_id());
|
|
sched.run();
|
|
|
|
// Close the idle callback.
|
|
let mut sched: ~Scheduler = Local::take();
|
|
sched.idle_callback.take();
|
|
// Make one go through the loop to run the close callback.
|
|
sched.run();
|
|
|
|
// Now that we are done with the scheduler, clean up the
|
|
// scheduler task. Do so by removing it from TLS and manually
|
|
// cleaning up the memory it uses. As we didn't actually call
|
|
// task.run() on the scheduler task we never get through all
|
|
// the cleanup code it runs.
|
|
let mut stask: ~Task = Local::take();
|
|
|
|
rtdebug!("stopping scheduler {}", stask.sched.get_ref().sched_id());
|
|
|
|
// Should not have any messages
|
|
let message = stask.sched.get_mut_ref().message_queue.pop();
|
|
rtassert!(message.is_none());
|
|
|
|
stask.destroyed = true;
|
|
}
|
|
|
|
// This does not return a scheduler, as the scheduler is placed
|
|
// inside the task.
|
|
pub fn run(mut ~self) {
|
|
|
|
// This is unsafe because we need to place the scheduler, with
|
|
// the event_loop inside, inside our task. But we still need a
|
|
// mutable reference to the event_loop to give it the "run"
|
|
// command.
|
|
unsafe {
|
|
let event_loop: *mut ~EventLoop = &mut self.event_loop;
|
|
|
|
// Our scheduler must be in the task before the event loop
|
|
// is started.
|
|
let self_sched = Cell::new(self);
|
|
do Local::borrow |stask: &mut Task| {
|
|
stask.sched = Some(self_sched.take());
|
|
};
|
|
|
|
(*event_loop).run();
|
|
}
|
|
}
|
|
|
|
// * Execution Functions - Core Loop Logic
|
|
|
|
// The model for this function is that you continue through it
|
|
// until you either use the scheduler while performing a schedule
|
|
// action, in which case you give it away and return early, or
|
|
// you reach the end and sleep. In the case that a scheduler
|
|
// action is performed the loop is evented such that this function
|
|
// is called again.
|
|
fn run_sched_once() {
|
|
|
|
// When we reach the scheduler context via the event loop we
|
|
// already have a scheduler stored in our local task, so we
|
|
// start off by taking it. This is the only path through the
|
|
// scheduler where we get the scheduler this way.
|
|
let mut sched: ~Scheduler = Local::take();
|
|
|
|
// Assume that we need to continue idling unless we reach the
|
|
// end of this function without performing an action.
|
|
sched.idle_callback.get_mut_ref().resume();
|
|
|
|
// First we check for scheduler messages, these are higher
|
|
// priority than regular tasks.
|
|
let sched = match sched.interpret_message_queue(DontTryTooHard) {
|
|
Some(sched) => sched,
|
|
None => return
|
|
};
|
|
|
|
// This helper will use a randomized work-stealing algorithm
|
|
// to find work.
|
|
let sched = match sched.do_work() {
|
|
Some(sched) => sched,
|
|
None => return
|
|
};
|
|
|
|
// Now, before sleeping we need to find out if there really
|
|
// were any messages. Give it your best!
|
|
let mut sched = match sched.interpret_message_queue(GiveItYourBest) {
|
|
Some(sched) => sched,
|
|
None => return
|
|
};
|
|
|
|
// If we got here then there was no work to do.
|
|
// Generate a SchedHandle and push it to the sleeper list so
|
|
// somebody can wake us up later.
|
|
if !sched.sleepy && !sched.no_sleep {
|
|
rtdebug!("scheduler has no work to do, going to sleep");
|
|
sched.sleepy = true;
|
|
let handle = sched.make_handle();
|
|
sched.sleeper_list.push(handle);
|
|
// Since we are sleeping, deactivate the idle callback.
|
|
sched.idle_callback.get_mut_ref().pause();
|
|
} else {
|
|
rtdebug!("not sleeping, already doing so or no_sleep set");
|
|
// We may not be sleeping, but we still need to deactivate
|
|
// the idle callback.
|
|
sched.idle_callback.get_mut_ref().pause();
|
|
}
|
|
|
|
// Finished a cycle without using the Scheduler. Place it back
|
|
// in TLS.
|
|
Local::put(sched);
|
|
}
|
|
|
|
// This function returns None if the scheduler is "used", or it
|
|
// returns the still-available scheduler. At this point all
|
|
// message-handling will count as a turn of work, and as a result
|
|
// return None.
|
|
fn interpret_message_queue(mut ~self, effort: EffortLevel) -> Option<~Scheduler> {
|
|
|
|
let msg = if effort == DontTryTooHard {
|
|
// Do a cheap check that may miss messages
|
|
self.message_queue.casual_pop()
|
|
} else {
|
|
self.message_queue.pop()
|
|
};
|
|
|
|
match msg {
|
|
Some(PinnedTask(task)) => {
|
|
let mut task = task;
|
|
task.give_home(Sched(self.make_handle()));
|
|
self.resume_task_immediately(task);
|
|
return None;
|
|
}
|
|
Some(TaskFromFriend(task)) => {
|
|
rtdebug!("got a task from a friend. lovely!");
|
|
self.process_task(task, Scheduler::resume_task_immediately_cl);
|
|
return None;
|
|
}
|
|
Some(RunOnce(task)) => {
|
|
// bypass the process_task logic to force running this task once
|
|
// on this home scheduler. This is often used for I/O (homing).
|
|
Scheduler::resume_task_immediately_cl(self, task);
|
|
return None;
|
|
}
|
|
Some(Wake) => {
|
|
self.sleepy = false;
|
|
Local::put(self);
|
|
return None;
|
|
}
|
|
Some(Shutdown) => {
|
|
rtdebug!("shutting down");
|
|
if self.sleepy {
|
|
// There may be an outstanding handle on the
|
|
// sleeper list. Pop them all to make sure that's
|
|
// not the case.
|
|
loop {
|
|
match self.sleeper_list.pop() {
|
|
Some(handle) => {
|
|
let mut handle = handle;
|
|
handle.send(Wake);
|
|
}
|
|
None => break
|
|
}
|
|
}
|
|
}
|
|
// No more sleeping. After there are no outstanding
|
|
// event loop references we will shut down.
|
|
self.no_sleep = true;
|
|
self.sleepy = false;
|
|
Local::put(self);
|
|
return None;
|
|
}
|
|
None => {
|
|
return Some(self);
|
|
}
|
|
}
|
|
}
|
|
|
|
fn do_work(mut ~self) -> Option<~Scheduler> {
|
|
rtdebug!("scheduler calling do work");
|
|
match self.find_work() {
|
|
Some(task) => {
|
|
rtdebug!("found some work! processing the task");
|
|
self.process_task(task, Scheduler::resume_task_immediately_cl);
|
|
return None;
|
|
}
|
|
None => {
|
|
rtdebug!("no work was found, returning the scheduler struct");
|
|
return Some(self);
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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");
|
|
if !self.steal_for_yield {
|
|
match self.work_queue.pop() {
|
|
Some(task) => {
|
|
rtdebug!("found a task locally");
|
|
return Some(task)
|
|
}
|
|
None => {
|
|
rtdebug!("scheduler trying to steal");
|
|
return self.try_steals();
|
|
}
|
|
}
|
|
} else {
|
|
// During execution of the last task, it performed a 'yield',
|
|
// so we're doing some work stealing in order to introduce some
|
|
// scheduling randomness. Otherwise we would just end up popping
|
|
// that same task again. This is pretty lame and is to work around
|
|
// the problem that work stealing is not designed for 'non-strict'
|
|
// (non-fork-join) task parallelism.
|
|
self.steal_for_yield = false;
|
|
match self.try_steals() {
|
|
Some(task) => {
|
|
rtdebug!("stole a task after yielding");
|
|
return Some(task);
|
|
}
|
|
None => {
|
|
rtdebug!("did not steal a task after yielding");
|
|
// Back to business
|
|
return self.find_work();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Try stealing from all queues the scheduler knows about. This
|
|
// naive implementation can steal from our own queue or from other
|
|
// special schedulers.
|
|
fn try_steals(&mut self) -> Option<~Task> {
|
|
let work_queues = &mut self.work_queues;
|
|
let len = work_queues.len();
|
|
let start_index = self.rng.gen_range(0, len);
|
|
for index in range(0, len).map(|i| (i + start_index) % len) {
|
|
match work_queues[index].steal() {
|
|
Some(task) => {
|
|
rtdebug!("found task by stealing");
|
|
return Some(task)
|
|
}
|
|
None => ()
|
|
}
|
|
};
|
|
rtdebug!("giving up on stealing");
|
|
return None;
|
|
}
|
|
|
|
// * Task Routing Functions - Make sure tasks send up in the right
|
|
// place.
|
|
|
|
fn process_task(mut ~self, mut task: ~Task, schedule_fn: SchedulingFn) {
|
|
rtdebug!("processing a task");
|
|
|
|
let home = task.take_unwrap_home();
|
|
match home {
|
|
Sched(home_handle) => {
|
|
if home_handle.sched_id != self.sched_id() {
|
|
rtdebug!("sending task home");
|
|
task.give_home(Sched(home_handle));
|
|
Scheduler::send_task_home(task);
|
|
Local::put(self);
|
|
} else {
|
|
rtdebug!("running task here");
|
|
task.give_home(Sched(home_handle));
|
|
schedule_fn(self, task);
|
|
}
|
|
}
|
|
AnySched if self.run_anything => {
|
|
rtdebug!("running anysched task here");
|
|
task.give_home(AnySched);
|
|
schedule_fn(self, task);
|
|
}
|
|
AnySched => {
|
|
rtdebug!("sending task to friend");
|
|
task.give_home(AnySched);
|
|
self.send_to_friend(task);
|
|
Local::put(self);
|
|
}
|
|
}
|
|
}
|
|
|
|
fn send_task_home(task: ~Task) {
|
|
let mut task = task;
|
|
let mut home = task.take_unwrap_home();
|
|
match home {
|
|
Sched(ref mut home_handle) => {
|
|
home_handle.send(PinnedTask(task));
|
|
}
|
|
AnySched => {
|
|
rtabort!("error: cannot send anysched task home");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Take a non-homed task we aren't allowed to run here and send
|
|
/// it to the designated friend scheduler to execute.
|
|
fn send_to_friend(&mut self, task: ~Task) {
|
|
rtdebug!("sending a task to friend");
|
|
match self.friend_handle {
|
|
Some(ref mut handle) => {
|
|
handle.send(TaskFromFriend(task));
|
|
}
|
|
None => {
|
|
rtabort!("tried to send task to a friend but scheduler has no friends");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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.
|
|
pub fn enqueue_task(&mut self, task: ~Task) {
|
|
|
|
// We push the task onto our local queue clone.
|
|
self.work_queue.push(task);
|
|
self.idle_callback.get_mut_ref().resume();
|
|
|
|
// We've made work available. Notify a
|
|
// sleeping scheduler.
|
|
|
|
match self.sleeper_list.casual_pop() {
|
|
Some(handle) => {
|
|
let mut handle = handle;
|
|
handle.send(Wake)
|
|
}
|
|
None => { (/* pass */) }
|
|
};
|
|
}
|
|
|
|
/// As enqueue_task, but with the possibility for the blocked task to
|
|
/// already have been killed.
|
|
pub fn enqueue_blocked_task(&mut self, blocked_task: BlockedTask) {
|
|
do blocked_task.wake().map |task| {
|
|
self.enqueue_task(task);
|
|
};
|
|
}
|
|
|
|
// * Core Context Switching Functions
|
|
|
|
// The primary function for changing contexts. In the current
|
|
// design the scheduler is just a slightly modified GreenTask, so
|
|
// all context swaps are from Task to Task. The only difference
|
|
// between the various cases is where the inputs come from, and
|
|
// what is done with the resulting task. That is specified by the
|
|
// cleanup function f, which takes the scheduler and the
|
|
// old task as inputs.
|
|
|
|
pub fn change_task_context(mut ~self,
|
|
next_task: ~Task,
|
|
f: &fn(&mut Scheduler, ~Task)) {
|
|
// The current task is grabbed from TLS, not taken as an input.
|
|
// Doing an unsafe_take to avoid writing back a null pointer -
|
|
// We're going to call `put` later to do that.
|
|
let current_task: ~Task = unsafe { Local::unsafe_take() };
|
|
|
|
// Check that the task is not in an atomically() section (e.g.,
|
|
// holding a pthread mutex, which could deadlock the scheduler).
|
|
current_task.death.assert_may_sleep();
|
|
|
|
// These transmutes do something fishy with a closure.
|
|
let f_fake_region = unsafe {
|
|
transmute::<&fn(&mut Scheduler, ~Task),
|
|
&fn(&mut Scheduler, ~Task)>(f)
|
|
};
|
|
let f_opaque = ClosureConverter::from_fn(f_fake_region);
|
|
|
|
// The current task is placed inside an enum with the cleanup
|
|
// function. This enum is then placed inside the scheduler.
|
|
self.cleanup_job = Some(CleanupJob::new(current_task, f_opaque));
|
|
|
|
// The scheduler is then placed inside the next task.
|
|
let mut next_task = next_task;
|
|
next_task.sched = Some(self);
|
|
|
|
// However we still need an internal mutable pointer to the
|
|
// original task. The strategy here was "arrange memory, then
|
|
// get pointers", so we crawl back up the chain using
|
|
// transmute to eliminate borrowck errors.
|
|
unsafe {
|
|
|
|
let sched: &mut Scheduler =
|
|
transmute_mut_region(*next_task.sched.get_mut_ref());
|
|
|
|
let current_task: &mut Task = match sched.cleanup_job {
|
|
Some(CleanupJob { task: ref task, _ }) => {
|
|
let task_ptr: *~Task = task;
|
|
transmute_mut_region(*transmute_mut_unsafe(task_ptr))
|
|
}
|
|
None => {
|
|
rtabort!("no cleanup job");
|
|
}
|
|
};
|
|
|
|
let (current_task_context, next_task_context) =
|
|
Scheduler::get_contexts(current_task, next_task);
|
|
|
|
// Done with everything - put the next task in TLS. This
|
|
// works because due to transmute the borrow checker
|
|
// believes that we have no internal pointers to
|
|
// next_task.
|
|
Local::put(next_task);
|
|
|
|
// The raw context swap operation. The next action taken
|
|
// will be running the cleanup job from the context of the
|
|
// next task.
|
|
Context::swap(current_task_context, next_task_context);
|
|
}
|
|
|
|
// When the context swaps back to this task we immediately
|
|
// run the cleanup job, as expected by the previously called
|
|
// swap_contexts function.
|
|
unsafe {
|
|
let task: *mut Task = Local::unsafe_borrow();
|
|
(*task).sched.get_mut_ref().run_cleanup_job();
|
|
|
|
// Must happen after running the cleanup job (of course).
|
|
(*task).death.check_killed((*task).unwinder.unwinding);
|
|
}
|
|
}
|
|
|
|
// Returns a mutable reference to both contexts involved in this
|
|
// swap. This is unsafe - we are getting mutable internal
|
|
// references to keep even when we don't own the tasks. It looks
|
|
// kinda safe because we are doing transmutes before passing in
|
|
// the arguments.
|
|
pub fn get_contexts<'a>(current_task: &mut Task, next_task: &mut Task) ->
|
|
(&'a mut Context, &'a mut Context) {
|
|
let current_task_context =
|
|
&mut current_task.coroutine.get_mut_ref().saved_context;
|
|
let next_task_context =
|
|
&mut next_task.coroutine.get_mut_ref().saved_context;
|
|
unsafe {
|
|
(transmute_mut_region(current_task_context),
|
|
transmute_mut_region(next_task_context))
|
|
}
|
|
}
|
|
|
|
// * Context Swapping Helpers - Here be ugliness!
|
|
|
|
pub fn resume_task_immediately(~self, task: ~Task) {
|
|
do self.change_task_context(task) |sched, stask| {
|
|
sched.sched_task = Some(stask);
|
|
}
|
|
}
|
|
|
|
fn resume_task_immediately_cl(sched: ~Scheduler,
|
|
task: ~Task) {
|
|
sched.resume_task_immediately(task)
|
|
}
|
|
|
|
|
|
pub fn resume_blocked_task_immediately(~self, blocked_task: BlockedTask) {
|
|
match blocked_task.wake() {
|
|
Some(task) => { self.resume_task_immediately(task); }
|
|
None => Local::put(self)
|
|
};
|
|
}
|
|
|
|
/// Block a running task, context switch to the scheduler, then pass the
|
|
/// blocked task to a closure.
|
|
///
|
|
/// # Safety note
|
|
///
|
|
/// The closure here is a *stack* closure that lives in the
|
|
/// running task. It gets transmuted to the scheduler's lifetime
|
|
/// and called while the task is blocked.
|
|
///
|
|
/// 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(mut ~self,
|
|
f: &fn(&mut Scheduler, BlockedTask)) {
|
|
// Trickier - we need to get the scheduler task out of self
|
|
// and use it as the destination.
|
|
let stask = self.sched_task.take_unwrap();
|
|
// Otherwise this is the same as below.
|
|
self.switch_running_tasks_and_then(stask, f);
|
|
}
|
|
|
|
pub fn switch_running_tasks_and_then(~self, next_task: ~Task,
|
|
f: &fn(&mut Scheduler, BlockedTask)) {
|
|
// This is where we convert the BlockedTask-taking closure into one
|
|
// that takes just a Task, and is aware of the block-or-killed protocol.
|
|
do self.change_task_context(next_task) |sched, task| {
|
|
// Task might need to receive a kill signal instead of blocking.
|
|
// We can call the "and_then" only if it blocks successfully.
|
|
match BlockedTask::try_block(task) {
|
|
Left(killed_task) => sched.enqueue_task(killed_task),
|
|
Right(blocked_task) => f(sched, blocked_task),
|
|
}
|
|
}
|
|
}
|
|
|
|
fn switch_task(sched: ~Scheduler, task: ~Task) {
|
|
do sched.switch_running_tasks_and_then(task) |sched, last_task| {
|
|
sched.enqueue_blocked_task(last_task);
|
|
};
|
|
}
|
|
|
|
// * Task Context Helpers
|
|
|
|
/// Called by a running task to end execution, after which it will
|
|
/// be recycled by the scheduler for reuse in a new task.
|
|
pub fn terminate_current_task(mut ~self) {
|
|
// Similar to deschedule running task and then, but cannot go through
|
|
// the task-blocking path. The task is already dying.
|
|
let stask = self.sched_task.take_unwrap();
|
|
do self.change_task_context(stask) |sched, mut dead_task| {
|
|
let coroutine = dead_task.coroutine.take_unwrap();
|
|
coroutine.recycle(&mut sched.stack_pool);
|
|
}
|
|
}
|
|
|
|
pub fn run_task(task: ~Task) {
|
|
let sched: ~Scheduler = Local::take();
|
|
sched.process_task(task, Scheduler::switch_task);
|
|
}
|
|
|
|
pub fn run_task_later(next_task: ~Task) {
|
|
let next_task = Cell::new(next_task);
|
|
do Local::borrow |sched: &mut Scheduler| {
|
|
sched.enqueue_task(next_task.take());
|
|
};
|
|
}
|
|
|
|
/// Yield control to the scheduler, executing another task. This is guaranteed
|
|
/// to introduce some amount of randomness to the scheduler. Currently the
|
|
/// randomness is a result of performing a round of work stealing (which
|
|
/// may end up stealing from the current scheduler).
|
|
pub fn yield_now(mut ~self) {
|
|
self.yield_check_count = reset_yield_check(&mut self.rng);
|
|
// Tell the scheduler to start stealing on the next iteration
|
|
self.steal_for_yield = true;
|
|
do self.deschedule_running_task_and_then |sched, task| {
|
|
sched.enqueue_blocked_task(task);
|
|
}
|
|
}
|
|
|
|
pub fn maybe_yield(mut ~self) {
|
|
// The number of times to do the yield check before yielding, chosen arbitrarily.
|
|
rtassert!(self.yield_check_count > 0);
|
|
self.yield_check_count -= 1;
|
|
if self.yield_check_count == 0 {
|
|
self.yield_now();
|
|
} else {
|
|
Local::put(self);
|
|
}
|
|
}
|
|
|
|
|
|
// * Utility Functions
|
|
|
|
pub fn sched_id(&self) -> uint { to_uint(self) }
|
|
|
|
pub fn run_cleanup_job(&mut self) {
|
|
let cleanup_job = self.cleanup_job.take_unwrap();
|
|
cleanup_job.run(self);
|
|
}
|
|
|
|
pub fn make_handle(&mut self) -> SchedHandle {
|
|
let remote = self.event_loop.remote_callback(~SchedRunner as ~Callback);
|
|
|
|
return SchedHandle {
|
|
remote: remote,
|
|
queue: self.message_queue.clone(),
|
|
sched_id: self.sched_id()
|
|
};
|
|
}
|
|
}
|
|
|
|
// Supporting types
|
|
|
|
type SchedulingFn = extern "Rust" fn (~Scheduler, ~Task);
|
|
|
|
pub enum SchedMessage {
|
|
Wake,
|
|
Shutdown,
|
|
PinnedTask(~Task),
|
|
TaskFromFriend(~Task),
|
|
RunOnce(~Task),
|
|
}
|
|
|
|
pub struct SchedHandle {
|
|
priv remote: ~RemoteCallback,
|
|
priv queue: MessageQueue<SchedMessage>,
|
|
sched_id: uint
|
|
}
|
|
|
|
impl SchedHandle {
|
|
pub fn send(&mut self, msg: SchedMessage) {
|
|
self.queue.push(msg);
|
|
self.remote.fire();
|
|
}
|
|
}
|
|
|
|
struct SchedRunner;
|
|
|
|
impl Callback for SchedRunner {
|
|
fn call(&mut self) {
|
|
Scheduler::run_sched_once();
|
|
}
|
|
}
|
|
|
|
struct CleanupJob {
|
|
task: ~Task,
|
|
f: UnsafeTaskReceiver
|
|
}
|
|
|
|
impl CleanupJob {
|
|
pub fn new(task: ~Task, f: UnsafeTaskReceiver) -> CleanupJob {
|
|
CleanupJob {
|
|
task: task,
|
|
f: f
|
|
}
|
|
}
|
|
|
|
pub fn run(self, sched: &mut Scheduler) {
|
|
let CleanupJob { task: task, f: f } = self;
|
|
f.to_fn()(sched, task)
|
|
}
|
|
}
|
|
|
|
// XXX: Some hacks to put a &fn in Scheduler without borrowck
|
|
// complaining
|
|
type UnsafeTaskReceiver = raw::Closure;
|
|
trait ClosureConverter {
|
|
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, ~Task)) -> UnsafeTaskReceiver {
|
|
unsafe { transmute(f) }
|
|
}
|
|
fn to_fn(self) -> &fn(&mut Scheduler, ~Task) { unsafe { transmute(self) } }
|
|
}
|
|
|
|
// On unix, we read randomness straight from /dev/urandom, but the
|
|
// default constructor of an XorShiftRng does this via io::fs, which
|
|
// relies on the scheduler existing, so we have to manually load
|
|
// randomness. Windows has its own C API for this, so we don't need to
|
|
// worry there.
|
|
#[cfg(windows)]
|
|
fn new_sched_rng() -> XorShiftRng {
|
|
XorShiftRng::new()
|
|
}
|
|
#[cfg(unix)]
|
|
#[fixed_stack_segment] #[inline(never)]
|
|
fn new_sched_rng() -> XorShiftRng {
|
|
use libc;
|
|
use mem;
|
|
use c_str::ToCStr;
|
|
use vec::MutableVector;
|
|
use iter::Iterator;
|
|
use rand::SeedableRng;
|
|
|
|
let fd = do "/dev/urandom".with_c_str |name| {
|
|
unsafe { libc::open(name, libc::O_RDONLY, 0) }
|
|
};
|
|
if fd == -1 {
|
|
rtabort!("could not open /dev/urandom for reading.")
|
|
}
|
|
|
|
let mut seeds = [0u32, .. 4];
|
|
let size = mem::size_of_val(&seeds);
|
|
loop {
|
|
let nbytes = do seeds.as_mut_buf |buf, _| {
|
|
unsafe {
|
|
libc::read(fd,
|
|
buf as *mut libc::c_void,
|
|
size as libc::size_t)
|
|
}
|
|
};
|
|
rtassert!(nbytes as uint == size);
|
|
|
|
if !seeds.iter().all(|x| *x == 0) {
|
|
break;
|
|
}
|
|
}
|
|
|
|
unsafe {libc::close(fd);}
|
|
|
|
SeedableRng::from_seed(seeds)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod test {
|
|
extern mod extra;
|
|
|
|
use prelude::*;
|
|
use rt::test::*;
|
|
use unstable::run_in_bare_thread;
|
|
use borrow::to_uint;
|
|
use rt::sched::{Scheduler};
|
|
use cell::Cell;
|
|
use rt::thread::Thread;
|
|
use rt::task::{Task, Sched};
|
|
use rt::basic;
|
|
use rt::util;
|
|
use option::{Some};
|
|
use rt::task::UnwindResult;
|
|
|
|
#[test]
|
|
fn trivial_run_in_newsched_task_test() {
|
|
let mut task_ran = false;
|
|
let task_ran_ptr: *mut bool = &mut task_ran;
|
|
do run_in_newsched_task || {
|
|
unsafe { *task_ran_ptr = true };
|
|
rtdebug!("executed from the new scheduler")
|
|
}
|
|
assert!(task_ran);
|
|
}
|
|
|
|
#[test]
|
|
fn multiple_task_test() {
|
|
let total = 10;
|
|
let mut task_run_count = 0;
|
|
let task_run_count_ptr: *mut uint = &mut task_run_count;
|
|
do run_in_newsched_task || {
|
|
for _ in range(0u, total) {
|
|
do spawntask || {
|
|
unsafe { *task_run_count_ptr = *task_run_count_ptr + 1};
|
|
}
|
|
}
|
|
}
|
|
assert!(task_run_count == total);
|
|
}
|
|
|
|
#[test]
|
|
fn multiple_task_nested_test() {
|
|
let mut task_run_count = 0;
|
|
let task_run_count_ptr: *mut uint = &mut task_run_count;
|
|
do run_in_newsched_task || {
|
|
do spawntask || {
|
|
unsafe { *task_run_count_ptr = *task_run_count_ptr + 1 };
|
|
do spawntask || {
|
|
unsafe { *task_run_count_ptr = *task_run_count_ptr + 1 };
|
|
do spawntask || {
|
|
unsafe { *task_run_count_ptr = *task_run_count_ptr + 1 };
|
|
}
|
|
}
|
|
}
|
|
}
|
|
assert!(task_run_count == 3);
|
|
}
|
|
|
|
// 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 very simple test that confirms that a task executing on the
|
|
// home scheduler notices that it is 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 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());
|
|
};
|
|
|
|
let on_exit: ~fn(UnwindResult) = |exit_status| rtassert!(exit_status.is_success());
|
|
task.death.on_exit = Some(on_exit);
|
|
|
|
sched.bootstrap(task);
|
|
}
|
|
}
|
|
|
|
// An advanced test that checks all four possible states that a
|
|
// (task,sched) can be in regarding homes.
|
|
|
|
#[test]
|
|
fn test_schedule_home_states() {
|
|
use rt::sleeper_list::SleeperList;
|
|
use rt::work_queue::WorkQueue;
|
|
use rt::sched::Shutdown;
|
|
use borrow;
|
|
use rt::comm::*;
|
|
|
|
do run_in_bare_thread {
|
|
|
|
let sleepers = SleeperList::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(
|
|
basic::event_loop(),
|
|
normal_queue,
|
|
queues.clone(),
|
|
sleepers.clone());
|
|
|
|
let normal_handle = Cell::new(normal_sched.make_handle());
|
|
|
|
let friend_handle = normal_sched.make_handle();
|
|
|
|
// Our special scheduler
|
|
let mut special_sched = ~Scheduler::new_special(
|
|
basic::event_loop(),
|
|
special_queue.clone(),
|
|
queues.clone(),
|
|
sleepers.clone(),
|
|
false,
|
|
Some(friend_handle));
|
|
|
|
let special_handle = Cell::new(special_sched.make_handle());
|
|
|
|
let t1_handle = special_sched.make_handle();
|
|
let t4_handle = special_sched.make_handle();
|
|
|
|
// Four test tasks:
|
|
// 1) task is home on special
|
|
// 2) task not homed, sched doesn't care
|
|
// 3) task not homed, sched requeues
|
|
// 4) task not home, send home
|
|
|
|
let task1 = ~do Task::new_root_homed(&mut special_sched.stack_pool, None,
|
|
Sched(t1_handle)) || {
|
|
rtassert!(Task::on_appropriate_sched());
|
|
};
|
|
rtdebug!("task1 id: **{}**", borrow::to_uint(task1));
|
|
|
|
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, None) {
|
|
rtassert!(Task::on_appropriate_sched());
|
|
};
|
|
|
|
let task4 = ~do Task::new_root_homed(&mut special_sched.stack_pool, None,
|
|
Sched(t4_handle)) {
|
|
rtassert!(Task::on_appropriate_sched());
|
|
};
|
|
rtdebug!("task4 id: **{}**", borrow::to_uint(task4));
|
|
|
|
let task1 = Cell::new(task1);
|
|
let task2 = Cell::new(task2);
|
|
let task3 = Cell::new(task3);
|
|
let task4 = Cell::new(task4);
|
|
|
|
// Signal from the special task that we are done.
|
|
let (port, chan) = oneshot::<()>();
|
|
let port = Cell::new(port);
|
|
let chan = Cell::new(chan);
|
|
|
|
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*");
|
|
Scheduler::run_task(task4.take());
|
|
rtdebug!("*normal_task done*");
|
|
port.take().recv();
|
|
let mut nh = normal_handle.take();
|
|
nh.send(Shutdown);
|
|
let mut sh = special_handle.take();
|
|
sh.send(Shutdown);
|
|
};
|
|
|
|
rtdebug!("normal task: {}", borrow::to_uint(normal_task));
|
|
|
|
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*");
|
|
Scheduler::run_task(task3.take());
|
|
rtdebug!("*done with special_task*");
|
|
chan.take().send(());
|
|
};
|
|
|
|
rtdebug!("special task: {}", borrow::to_uint(special_task));
|
|
|
|
let special_sched = Cell::new(special_sched);
|
|
let normal_sched = Cell::new(normal_sched);
|
|
let special_task = Cell::new(special_task);
|
|
let normal_task = Cell::new(normal_task);
|
|
|
|
let normal_thread = do Thread::start {
|
|
normal_sched.take().bootstrap(normal_task.take());
|
|
rtdebug!("finished with normal_thread");
|
|
};
|
|
|
|
let special_thread = do Thread::start {
|
|
special_sched.take().bootstrap(special_task.take());
|
|
rtdebug!("finished with special_sched");
|
|
};
|
|
|
|
normal_thread.join();
|
|
special_thread.join();
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_stress_schedule_task_states() {
|
|
if util::limit_thread_creation_due_to_osx_and_valgrind() { return; }
|
|
let n = stress_factor() * 120;
|
|
for _ in range(0, n as int) {
|
|
test_schedule_home_states();
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_io_callback() {
|
|
use rt::io::timer;
|
|
|
|
// This is a regression test that when there are no schedulable tasks
|
|
// in the work queue, but we are performing I/O, that once we do put
|
|
// something in the work queue again the scheduler picks it up and doesn't
|
|
// exit before emptying the work queue
|
|
do run_in_uv_task {
|
|
do spawntask {
|
|
timer::sleep(10);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn handle() {
|
|
use rt::comm::*;
|
|
|
|
do run_in_bare_thread {
|
|
let (port, chan) = oneshot::<()>();
|
|
let port = Cell::new(port);
|
|
let chan = Cell::new(chan);
|
|
|
|
let thread_one = do Thread::start {
|
|
let chan = Cell::new(chan.take());
|
|
do run_in_newsched_task_core {
|
|
chan.take().send(());
|
|
}
|
|
};
|
|
|
|
let thread_two = do Thread::start {
|
|
let port = Cell::new(port.take());
|
|
do run_in_newsched_task_core {
|
|
port.take().recv();
|
|
}
|
|
};
|
|
|
|
thread_two.join();
|
|
thread_one.join();
|
|
}
|
|
}
|
|
|
|
// A regression test that the final message is always handled.
|
|
// Used to deadlock because Shutdown was never recvd.
|
|
#[test]
|
|
fn no_missed_messages() {
|
|
use rt::work_queue::WorkQueue;
|
|
use rt::sleeper_list::SleeperList;
|
|
use rt::stack::StackPool;
|
|
use rt::sched::{Shutdown, TaskFromFriend};
|
|
use util;
|
|
|
|
do run_in_bare_thread {
|
|
do stress_factor().times {
|
|
let sleepers = SleeperList::new();
|
|
let queue = WorkQueue::new();
|
|
let queues = ~[queue.clone()];
|
|
|
|
let mut sched = ~Scheduler::new(
|
|
basic::event_loop(),
|
|
queue,
|
|
queues.clone(),
|
|
sleepers.clone());
|
|
|
|
let mut handle = sched.make_handle();
|
|
|
|
let sched = Cell::new(sched);
|
|
|
|
let thread = do Thread::start {
|
|
let mut sched = sched.take();
|
|
let bootstrap_task = ~Task::new_root(&mut sched.stack_pool, None, ||());
|
|
sched.bootstrap(bootstrap_task);
|
|
};
|
|
|
|
let mut stack_pool = StackPool::new();
|
|
let task = ~Task::new_root(&mut stack_pool, None, ||());
|
|
handle.send(TaskFromFriend(task));
|
|
|
|
handle.send(Shutdown);
|
|
util::ignore(handle);
|
|
|
|
thread.join();
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn multithreading() {
|
|
use rt::comm::*;
|
|
use num::Times;
|
|
use vec::OwnedVector;
|
|
use container::Container;
|
|
|
|
do run_in_mt_newsched_task {
|
|
let mut ports = ~[];
|
|
do 10.times {
|
|
let (port, chan) = oneshot();
|
|
let chan_cell = Cell::new(chan);
|
|
do spawntask_later {
|
|
chan_cell.take().send(());
|
|
}
|
|
ports.push(port);
|
|
}
|
|
|
|
while !ports.is_empty() {
|
|
ports.pop().recv();
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn thread_ring() {
|
|
use rt::comm::*;
|
|
use comm::{GenericPort, GenericChan};
|
|
|
|
do run_in_mt_newsched_task {
|
|
let (end_port, end_chan) = oneshot();
|
|
|
|
let n_tasks = 10;
|
|
let token = 2000;
|
|
|
|
let (p, ch1) = stream();
|
|
let mut p = p;
|
|
ch1.send((token, end_chan));
|
|
let mut i = 2;
|
|
while i <= n_tasks {
|
|
let (next_p, ch) = stream();
|
|
let imm_i = i;
|
|
let imm_p = p;
|
|
do spawntask_random {
|
|
roundtrip(imm_i, n_tasks, &imm_p, &ch);
|
|
};
|
|
p = next_p;
|
|
i += 1;
|
|
}
|
|
let imm_p = p;
|
|
let imm_ch = ch1;
|
|
do spawntask_random {
|
|
roundtrip(1, n_tasks, &imm_p, &imm_ch);
|
|
}
|
|
|
|
end_port.recv();
|
|
}
|
|
|
|
fn roundtrip(id: int, n_tasks: int,
|
|
p: &Port<(int, ChanOne<()>)>, ch: &Chan<(int, ChanOne<()>)>) {
|
|
while (true) {
|
|
match p.recv() {
|
|
(1, end_chan) => {
|
|
debug!("{}\n", id);
|
|
end_chan.send(());
|
|
return;
|
|
}
|
|
(token, end_chan) => {
|
|
debug!("thread: {} got token: {}", id, token);
|
|
ch.send((token - 1, end_chan));
|
|
if token <= n_tasks {
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn start_closure_dtor() {
|
|
use ops::Drop;
|
|
|
|
// Regression test that the `start` task entrypoint can
|
|
// contain dtors that use task resources
|
|
do run_in_newsched_task {
|
|
struct S { field: () }
|
|
|
|
impl Drop for S {
|
|
fn drop(&mut self) {
|
|
let _foo = @0;
|
|
}
|
|
}
|
|
|
|
let s = S { field: () };
|
|
|
|
do spawntask {
|
|
let _ss = &s;
|
|
}
|
|
}
|
|
}
|
|
|
|
// FIXME: #9407: xfail-test
|
|
fn dont_starve_1() {
|
|
use rt::comm::oneshot;
|
|
|
|
do stress_factor().times {
|
|
do run_in_mt_newsched_task {
|
|
let (port, chan) = oneshot();
|
|
|
|
// This task should not be able to starve the sender;
|
|
// The sender should get stolen to another thread.
|
|
do spawntask {
|
|
while !port.peek() { }
|
|
}
|
|
|
|
chan.send(());
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn dont_starve_2() {
|
|
use rt::comm::oneshot;
|
|
|
|
do stress_factor().times {
|
|
do run_in_newsched_task {
|
|
let (port, chan) = oneshot();
|
|
let (_port2, chan2) = stream();
|
|
|
|
// This task should not be able to starve the other task.
|
|
// The sends should eventually yield.
|
|
do spawntask {
|
|
while !port.peek() {
|
|
chan2.send(());
|
|
}
|
|
}
|
|
|
|
chan.send(());
|
|
}
|
|
}
|
|
}
|
|
|
|
// Regression test for a logic bug that would cause single-threaded schedulers
|
|
// to sleep forever after yielding and stealing another task.
|
|
#[test]
|
|
fn single_threaded_yield() {
|
|
use task::{spawn, spawn_sched, SingleThreaded, deschedule};
|
|
use num::Times;
|
|
|
|
do spawn_sched(SingleThreaded) {
|
|
do 5.times { deschedule(); }
|
|
}
|
|
do spawn { }
|
|
do spawn { }
|
|
}
|
|
}
|