Merge remote-tracking branch 'brson/io' into incoming

Conflicts:
	src/libstd/rt/sched.rs

1 files changed, 266 insertions, 63 deletions

diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index 2d9cdaddc84..089c95cd7cd 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -12,21 +12,44 @@ 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::rtio::{EventLoop, EventLoopObject};
+use super::rtio::{EventLoop, EventLoopObject, RemoteCallbackObject};
 use super::context::Context;
 use super::task::Task;
+use super::message_queue::MessageQueue;
 use rt::local_ptr;
 use rt::local::Local;
+use rt::rtio::{IoFactoryObject, RemoteCallback};
 
 /// 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
 /// scheduler.
 pub struct Scheduler {
+    /// A queue of available work. Under a work-stealing policy there
+    /// is one per Scheduler.
     priv work_queue: WorkQueue<~Coroutine>,
+    /// 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.
+    priv 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 event loop used to drive the scheduler and perform I/O
     event_loop: ~EventLoopObject,
@@ -40,16 +63,25 @@ pub struct Scheduler {
     priv cleanup_job: Option<CleanupJob>
 }
 
-// XXX: Some hacks to put a &fn in Scheduler without borrowck
-// complaining
-type UnsafeTaskReceiver = sys::Closure;
-trait ClosureConverter {
-    fn from_fn(&fn(~Coroutine)) -> Self;
-    fn to_fn(self) -> &fn(~Coroutine);
+pub struct SchedHandle {
+    priv remote: ~RemoteCallbackObject,
+    priv queue: MessageQueue<SchedMessage>
 }
-impl ClosureConverter for UnsafeTaskReceiver {
-    fn from_fn(f: &fn(~Coroutine)) -> UnsafeTaskReceiver { unsafe { transmute(f) } }
-    fn to_fn(self) -> &fn(~Coroutine) { unsafe { transmute(self) } }
+
+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
+}
+
+pub enum SchedMessage {
+    Wake,
+    Shutdown
 }
 
 enum CleanupJob {
@@ -61,18 +93,25 @@ pub impl Scheduler {
 
     fn in_task_context(&self) -> bool { self.current_task.is_some() }
 
-    fn new(event_loop: ~EventLoopObject) -> Scheduler {
+    fn new(event_loop: ~EventLoopObject,
+           work_queue: WorkQueue<~Coroutine>,
+           sleeper_list: SleeperList)
+        -> Scheduler {
 
         // Lazily initialize the runtime TLS key
         local_ptr::init_tls_key();
 
         Scheduler {
+            sleeper_list: sleeper_list,
+            message_queue: MessageQueue::new(),
+            sleepy: false,
+            no_sleep: false,
             event_loop: event_loop,
-            work_queue: WorkQueue::new(),
+            work_queue: work_queue,
             stack_pool: StackPool::new(),
             saved_context: Context::empty(),
             current_task: None,
-            cleanup_job: None
+            cleanup_job: None,
         }
     }
 
@@ -102,6 +141,53 @@ pub impl Scheduler {
         return sched;
     }
 
+    fn run_sched_once() {
+
+        let sched = Local::take::<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.
+            let mut sched = Local::take::<Scheduler>();
+            sched.event_loop.callback(Scheduler::run_sched_once);
+            Local::put(sched);
+            return;
+        }
+
+        let sched = Local::take::<Scheduler>();
+        if sched.resume_task_from_queue() {
+            // We performed a scheduling action. There may be other work
+            // to do yet, so let's try again later.
+            let mut sched = Local::take::<Scheduler>();
+            sched.event_loop.callback(Scheduler::run_sched_once);
+            Local::put(sched);
+            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.
+        rtdebug!("no work to do");
+        let mut sched = Local::take::<Scheduler>();
+        if !sched.sleepy && !sched.no_sleep {
+            rtdebug!("sleeping");
+            sched.sleepy = true;
+            let handle = sched.make_handle();
+            sched.sleeper_list.push(handle);
+        } else {
+            rtdebug!("not sleeping");
+        }
+        Local::put(sched);
+    }
+
+    fn make_handle(&mut self) -> SchedHandle {
+        let remote = self.event_loop.remote_callback(Scheduler::run_sched_once);
+
+        return SchedHandle {
+            remote: remote,
+            queue: self.message_queue.clone()
+        };
+    }
+
     /// Schedule a task to be executed later.
     ///
     /// Pushes the task onto the work stealing queue and tells the event loop
@@ -109,17 +195,63 @@ pub impl Scheduler {
     /// directly.
     fn enqueue_task(&mut self, task: ~Coroutine) {
         self.work_queue.push(task);
-        self.event_loop.callback(resume_task_from_queue);
+        self.event_loop.callback(Scheduler::run_sched_once);
 
-        fn resume_task_from_queue() {
-            let scheduler = Local::take::<Scheduler>();
-            scheduler.resume_task_from_queue();
+        // We've made work available. Notify a sleeping scheduler.
+        match self.sleeper_list.pop() {
+            Some(handle) => {
+                let mut handle = handle;
+                handle.send(Wake)
+            }
+            None => (/* pass */)
         }
     }
 
     // * Scheduler-context operations
 
-    fn resume_task_from_queue(~self) {
+    fn interpret_message_queue(~self) -> bool {
+        assert!(!self.in_task_context());
+
+        rtdebug!("looking for scheduler messages");
+
+        let mut this = self;
+        match this.message_queue.pop() {
+            Some(Wake) => {
+                rtdebug!("recv Wake message");
+                this.sleepy = false;
+                Local::put(this);
+                return true;
+            }
+            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.
+                    loop {
+                        match this.sleeper_list.pop() {
+                            Some(handle) => {
+                                let mut handle = handle;
+                                handle.send(Wake);
+                            }
+                            None => (/* pass */)
+                        }
+                    }
+                }
+                // 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);
+                return true;
+            }
+            None => {
+                Local::put(this);
+                return false;
+            }
+        }
+    }
+
+    fn resume_task_from_queue(~self) -> bool {
         assert!(!self.in_task_context());
 
         rtdebug!("looking in work queue for task to schedule");
@@ -129,10 +261,12 @@ pub impl Scheduler {
             Some(task) => {
                 rtdebug!("resuming task from work queue");
                 this.resume_task_immediately(task);
+                return true;
             }
             None => {
                 rtdebug!("no tasks in queue");
                 Local::put(this);
+                return false;
             }
         }
     }
@@ -146,11 +280,9 @@ pub impl Scheduler {
 
         rtdebug!("ending running task");
 
-        do self.deschedule_running_task_and_then |dead_task| {
+        do self.deschedule_running_task_and_then |sched, dead_task| {
             let dead_task = Cell(dead_task);
-            do Local::borrow::<Scheduler> |sched| {
-                dead_task.take().recycle(&mut sched.stack_pool);
-            }
+            dead_task.take().recycle(&mut sched.stack_pool);
         }
 
         abort!("control reached end of task");
@@ -159,22 +291,18 @@ pub impl Scheduler {
     fn schedule_new_task(~self, task: ~Coroutine) {
         assert!(self.in_task_context());
 
-        do self.switch_running_tasks_and_then(task) |last_task| {
+        do self.switch_running_tasks_and_then(task) |sched, last_task| {
             let last_task = Cell(last_task);
-            do Local::borrow::<Scheduler> |sched| {
-                sched.enqueue_task(last_task.take());
-            }
+            sched.enqueue_task(last_task.take());
         }
     }
 
     fn schedule_task(~self, task: ~Coroutine) {
         assert!(self.in_task_context());
 
-        do self.switch_running_tasks_and_then(task) |last_task| {
+        do self.switch_running_tasks_and_then(task) |sched, last_task| {
             let last_task = Cell(last_task);
-            do Local::borrow::<Scheduler> |sched| {
-                sched.enqueue_task(last_task.take());
-            }
+            sched.enqueue_task(last_task.take());
         }
     }
 
@@ -218,7 +346,11 @@ pub impl Scheduler {
     /// 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.
-    fn deschedule_running_task_and_then(~self, f: &fn(~Coroutine)) {
+    ///
+    /// 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.
+    fn deschedule_running_task_and_then(~self, f: &fn(&mut Scheduler, ~Coroutine)) {
         let mut this = self;
         assert!(this.in_task_context());
 
@@ -226,7 +358,8 @@ pub impl Scheduler {
 
         unsafe {
             let blocked_task = this.current_task.swap_unwrap();
-            let f_fake_region = transmute::<&fn(~Coroutine), &fn(~Coroutine)>(f);
+            let f_fake_region = transmute::<&fn(&mut Scheduler, ~Coroutine),
+                                            &fn(&mut Scheduler, ~Coroutine)>(f);
             let f_opaque = ClosureConverter::from_fn(f_fake_region);
             this.enqueue_cleanup_job(GiveTask(blocked_task, f_opaque));
         }
@@ -248,14 +381,18 @@ pub 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.
-    fn switch_running_tasks_and_then(~self, next_task: ~Coroutine, f: &fn(~Coroutine)) {
+    fn switch_running_tasks_and_then(~self, next_task: ~Coroutine,
+                                     f: &fn(&mut Scheduler, ~Coroutine)) {
         let mut this = self;
         assert!(this.in_task_context());
 
         rtdebug!("switching tasks");
 
         let old_running_task = this.current_task.swap_unwrap();
-        let f_fake_region = unsafe { transmute::<&fn(~Coroutine), &fn(~Coroutine)>(f) };
+        let f_fake_region = unsafe {
+            transmute::<&fn(&mut Scheduler, ~Coroutine),
+                        &fn(&mut Scheduler, ~Coroutine)>(f)
+        };
         let f_opaque = ClosureConverter::from_fn(f_fake_region);
         this.enqueue_cleanup_job(GiveTask(old_running_task, f_opaque));
         this.current_task = Some(next_task);
@@ -292,7 +429,7 @@ pub impl Scheduler {
         let cleanup_job = self.cleanup_job.swap_unwrap();
         match cleanup_job {
             DoNothing => { }
-            GiveTask(task, f) => (f.to_fn())(task)
+            GiveTask(task, f) => (f.to_fn())(self, task)
         }
     }
 
@@ -336,17 +473,11 @@ pub impl Scheduler {
     }
 }
 
-static MIN_STACK_SIZE: uint = 10000000; // XXX: Too much stack
-
-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
+impl SchedHandle {
+    pub fn send(&mut self, msg: SchedMessage) {
+        self.queue.push(msg);
+        self.remote.fire();
+    }
 }
 
 pub impl Coroutine {
@@ -357,6 +488,9 @@ pub impl Coroutine {
     fn with_task(stack_pool: &mut StackPool,
                   task: ~Task,
                   start: ~fn()) -> Coroutine {
+
+        static MIN_STACK_SIZE: uint = 10000000; // 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
@@ -400,6 +534,18 @@ pub impl Coroutine {
     }
 }
 
+// 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);
+}
+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) } }
+}
+
 #[cfg(test)]
 mod test {
     use int;
@@ -410,6 +556,7 @@ mod test {
     use rt::local::Local;
     use rt::test::*;
     use super::*;
+    use rt::thread::Thread;
 
     #[test]
     fn test_simple_scheduling() {
@@ -417,7 +564,7 @@ mod test {
             let mut task_ran = false;
             let task_ran_ptr: *mut bool = &mut task_ran;
 
-            let mut sched = ~UvEventLoop::new_scheduler();
+            let mut sched = ~new_test_uv_sched();
             let task = ~do Coroutine::new(&mut sched.stack_pool) {
                 unsafe { *task_ran_ptr = true; }
             };
@@ -434,7 +581,7 @@ mod test {
             let mut task_count = 0;
             let task_count_ptr: *mut int = &mut task_count;
 
-            let mut sched = ~UvEventLoop::new_scheduler();
+            let mut sched = ~new_test_uv_sched();
             for int::range(0, total) |_| {
                 let task = ~do Coroutine::new(&mut sched.stack_pool) {
                     unsafe { *task_count_ptr = *task_count_ptr + 1; }
@@ -452,7 +599,7 @@ mod test {
             let mut count = 0;
             let count_ptr: *mut int = &mut count;
 
-            let mut sched = ~UvEventLoop::new_scheduler();
+            let mut sched = ~new_test_uv_sched();
             let task1 = ~do Coroutine::new(&mut sched.stack_pool) {
                 unsafe { *count_ptr = *count_ptr + 1; }
                 let mut sched = Local::take::<Scheduler>();
@@ -460,11 +607,9 @@ mod test {
                     unsafe { *count_ptr = *count_ptr + 1; }
                 };
                 // Context switch directly to the new task
-                do sched.switch_running_tasks_and_then(task2) |task1| {
+                do sched.switch_running_tasks_and_then(task2) |sched, task1| {
                     let task1 = Cell(task1);
-                    do Local::borrow::<Scheduler> |sched| {
-                        sched.enqueue_task(task1.take());
-                    }
+                    sched.enqueue_task(task1.take());
                 }
                 unsafe { *count_ptr = *count_ptr + 1; }
             };
@@ -481,7 +626,7 @@ mod test {
             let mut count = 0;
             let count_ptr: *mut int = &mut count;
 
-            let mut sched = ~UvEventLoop::new_scheduler();
+            let mut sched = ~new_test_uv_sched();
 
             let start_task = ~do Coroutine::new(&mut sched.stack_pool) {
                 run_task(count_ptr);
@@ -510,16 +655,14 @@ mod test {
     #[test]
     fn test_block_task() {
         do run_in_bare_thread {
-            let mut sched = ~UvEventLoop::new_scheduler();
+            let mut sched = ~new_test_uv_sched();
             let task = ~do Coroutine::new(&mut sched.stack_pool) {
                 let sched = Local::take::<Scheduler>();
                 assert!(sched.in_task_context());
-                do sched.deschedule_running_task_and_then() |task| {
+                do sched.deschedule_running_task_and_then() |sched, task| {
                     let task = Cell(task);
-                    do Local::borrow::<Scheduler> |sched| {
-                        assert!(!sched.in_task_context());
-                        sched.enqueue_task(task.take());
-                    }
+                    assert!(!sched.in_task_context());
+                    sched.enqueue_task(task.take());
                 }
             };
             sched.enqueue_task(task);
@@ -536,8 +679,7 @@ mod test {
         do run_in_newsched_task {
             do spawn {
                 let sched = Local::take::<Scheduler>();
-                do sched.deschedule_running_task_and_then |task| {
-                    let mut sched = Local::take::<Scheduler>();
+                do sched.deschedule_running_task_and_then |sched, task| {
                     let task = Cell(task);
                     do sched.event_loop.callback_ms(10) {
                         rtdebug!("in callback");
@@ -545,9 +687,70 @@ mod test {
                         sched.enqueue_task(task.take());
                         Local::put(sched);
                     }
-                    Local::put(sched);
                 }
             }
         }
     }
+
+    #[test]
+    fn handle() {
+        use rt::comm::*;
+
+        do run_in_bare_thread {
+            let (port, chan) = oneshot::<()>();
+            let port_cell = Cell(port);
+            let chan_cell = Cell(chan);
+            let mut sched1 = ~new_test_uv_sched();
+            let handle1 = sched1.make_handle();
+            let handle1_cell = Cell(handle1);
+            let task1 = ~do Coroutine::new(&mut sched1.stack_pool) {
+                chan_cell.take().send(());
+            };
+            sched1.enqueue_task(task1);
+
+            let mut sched2 = ~new_test_uv_sched();
+            let task2 = ~do Coroutine::new(&mut sched2.stack_pool) {
+                port_cell.take().recv();
+                // Release the other scheduler's handle so it can exit
+                handle1_cell.take();
+            };
+            sched2.enqueue_task(task2);
+
+            let sched1_cell = Cell(sched1);
+            let _thread1 = do Thread::start {
+                let mut sched1 = sched1_cell.take();
+                sched1.run();
+            };
+
+            let sched2_cell = Cell(sched2);
+            let _thread2 = do Thread::start {
+                let mut sched2 = sched2_cell.take();
+                sched2.run();
+            };
+        }
+    }
+
+    #[test]
+    fn multithreading() {
+        use rt::comm::*;
+        use iter::Times;
+        use vec::OwnedVector;
+        use container::Container;
+
+        do run_in_mt_newsched_task {
+            let mut ports = ~[];
+            for 10.times {
+                let (port, chan) = oneshot();
+                let chan_cell = Cell(chan);
+                do spawntask_later {
+                    chan_cell.take().send(());
+                }
+                ports.push(port);
+            }
+
+            while !ports.is_empty() {
+                ports.pop().recv();
+            }
+        }
+    }
 }