From 066ca17eaae239a666579fc80fe9047aaac0599d Mon Sep 17 00:00:00 2001
From: toddaaro <github@opprobrio.us>
Date: Thu, 15 Aug 2013 11:13:41 -0700
Subject: [PATCH 1/6] an attempt at a singleton pausible idle callback for each
 scheduler. suffers from nondeterministic deadlock and also pending scheduler
 messages on scheduler shutdown.

---
 src/libstd/rt/sched.rs   | 88 +++++++++++++++++++++++++++++++---------
 src/libstd/rt/uv/idle.rs | 14 +++++++
 2 files changed, 83 insertions(+), 19 deletions(-)

diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index ce4e64c47d2ef..58e1676b42427 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -30,6 +30,7 @@ use cell::Cell;
 use rand::{XorShiftRng, RngUtil};
 use iterator::{range};
 use vec::{OwnedVector};
+use rt::uv::idle::IdleWatcher;
 
 /// The Scheduler is responsible for coordinating execution of Coroutines
 /// on a single thread. When the scheduler is running it is owned by
@@ -76,8 +77,11 @@ pub struct Scheduler {
     /// them to.
     friend_handle: Option<SchedHandle>,
     /// A fast XorShift rng for scheduler use
-    rng: XorShiftRng
-
+    rng: XorShiftRng,
+    /// An IdleWatcher
+    idle_watcher: IdleWatcher,
+    /// A flag to indicate whether or not the idle callback is active.
+    idle_flag: bool
 }
 
 pub struct SchedHandle {
@@ -124,6 +128,9 @@ impl Scheduler {
                        friend: Option<SchedHandle>)
         -> Scheduler {
 
+        let mut event_loop = event_loop;
+        let idle_watcher = IdleWatcher::new(event_loop.uvio.uv_loop());
+
         Scheduler {
             sleeper_list: sleeper_list,
             message_queue: MessageQueue::new(),
@@ -138,7 +145,9 @@ impl Scheduler {
             metrics: SchedMetrics::new(),
             run_anything: run_anything,
             friend_handle: friend,
-            rng: XorShiftRng::new()
+            rng: XorShiftRng::new(),
+            idle_watcher: idle_watcher,
+            idle_flag: true
         }
     }
 
@@ -151,6 +160,8 @@ impl Scheduler {
     // scheduler task and bootstrap into it.
     pub fn bootstrap(~self, task: ~Task) {
 
+        let mut this = self;
+
         // Initialize the TLS key.
         local_ptr::init_tls_key();
 
@@ -161,10 +172,17 @@ impl 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.
+        do this.idle_watcher.start |_idle_watcher, _status| {
+            Scheduler::run_sched_once();
+        }
+
         // 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);
+        this.resume_task_immediately(task);
 
         // Now we are back in the scheduler context, having
         // successfully run the input task. Start by running the
@@ -201,7 +219,7 @@ impl Scheduler {
         // Always run through the scheduler loop at least once so that
         // we enter the sleep state and can then be woken up by other
         // schedulers.
-        self_sched.event_loop.callback(Scheduler::run_sched_once);
+//        self_sched.event_loop.callback(Scheduler::run_sched_once);
 
         // This is unsafe because we need to place the scheduler, with
         // the event_loop inside, inside our task. But we still need a
@@ -235,7 +253,11 @@ impl Scheduler {
         // 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 sched = Local::take::<Scheduler>();
+        let mut sched = Local::take::<Scheduler>();
+
+        // Assume that we need to continue idling unless we reach the
+        // end of this function without performing an action.
+        sched.activate_idle();
 
         // Our first task is to read mail to see if we have important
         // messages.
@@ -282,8 +304,13 @@ impl Scheduler {
             sched.sleepy = true;
             let handle = sched.make_handle();
             sched.sleeper_list.push(handle);
+            // Since we are sleeping, deactivate the idle callback.
+            sched.pause_idle();
         } 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.pause_idle();
         }
 
         // Finished a cycle without using the Scheduler. Place it back
@@ -291,6 +318,26 @@ impl Scheduler {
         Local::put(sched);
     }
 
+    fn activate_idle(&mut self) {        
+        if self.idle_flag {
+            rtdebug!("idle flag already set, not reactivating idle watcher");
+        } else {
+            rtdebug!("idle flag was false, reactivating idle watcher");
+            self.idle_flag = true;
+            self.idle_watcher.restart();
+        }            
+    }
+
+    fn pause_idle(&mut self) {
+        if !self.idle_flag {
+            rtdebug!("idle flag false, not stopping idle watcher");
+        } else {
+            rtdebug!("idle flag true, stopping idle watcher");
+            self.idle_flag = false;
+            self.idle_watcher.stop();
+        }
+    }
+
     pub fn make_handle(&mut self) -> SchedHandle {
         let remote = self.event_loop.remote_callback(Scheduler::run_sched_once);
 
@@ -312,7 +359,7 @@ impl Scheduler {
 
         // We push the task onto our local queue clone.
         this.work_queue.push(task);
-        this.event_loop.callback(Scheduler::run_sched_once);
+//        this.event_loop.callback(Scheduler::run_sched_once);
 
         // We've made work available. Notify a
         // sleeping scheduler.
@@ -346,30 +393,34 @@ impl Scheduler {
     // * Scheduler-context operations
 
     // This function returns None if the scheduler is "used", or it
-    // returns the still-available scheduler.
+    // returns the still-available scheduler. Note: currently
+    // considers *any* message receive a use and returns None.
     fn interpret_message_queue(~self) -> Option<~Scheduler> {
 
         let mut this = self;
         match this.message_queue.pop() {
             Some(PinnedTask(task)) => {
-                this.event_loop.callback(Scheduler::run_sched_once);
+//                this.event_loop.callback(Scheduler::run_sched_once);
                 let mut task = task;
                 task.give_home(Sched(this.make_handle()));
                 this.resume_task_immediately(task);
                 return None;
             }
             Some(TaskFromFriend(task)) => {
-                this.event_loop.callback(Scheduler::run_sched_once);
+//                this.event_loop.callback(Scheduler::run_sched_once);
                 rtdebug!("got a task from a friend. lovely!");
-                return this.sched_schedule_task(task);
+                this.sched_schedule_task(task).map_move(Local::put);
+                return None;
             }
             Some(Wake) => {
-                this.event_loop.callback(Scheduler::run_sched_once);
+//                this.event_loop.callback(Scheduler::run_sched_once);
                 this.sleepy = false;
-                return Some(this);
+                Local::put(this);
+                return None;
+//                return Some(this);
             }
             Some(Shutdown) => {
-                this.event_loop.callback(Scheduler::run_sched_once);
+//                this.event_loop.callback(Scheduler::run_sched_once);
                 if this.sleepy {
                     // There may be an outstanding handle on the
                     // sleeper list.  Pop them all to make sure that's
@@ -388,11 +439,10 @@ impl Scheduler {
                 // event loop references we will shut down.
                 this.no_sleep = true;
                 this.sleepy = false;
-                // YYY: Does a shutdown count as a "use" of the
-                // scheduler? This seems to work - so I'm leaving it
-                // this way despite not having a solid rational for
-                // why I should return the scheduler here.
-                return Some(this);
+
+                Local::put(this);
+                return None;
+//                return Some(this);
             }
             None => {
                 return Some(this);
diff --git a/src/libstd/rt/uv/idle.rs b/src/libstd/rt/uv/idle.rs
index b73be9f7250db..a21146620ca82 100644
--- a/src/libstd/rt/uv/idle.rs
+++ b/src/libstd/rt/uv/idle.rs
@@ -48,6 +48,20 @@ impl IdleWatcher {
         }
     }
 
+    pub fn restart(&mut self) {
+        unsafe {
+            assert!(0 == uvll::idle_start(self.native_handle(), idle_cb))
+        };
+
+        extern fn idle_cb(handle: *uvll::uv_idle_t, status: c_int) {
+            let mut idle_watcher: IdleWatcher = NativeHandle::from_native_handle(handle);
+            let data = idle_watcher.get_watcher_data();
+            let cb: &IdleCallback = data.idle_cb.get_ref();
+            let status = status_to_maybe_uv_error(idle_watcher, status);
+            (*cb)(idle_watcher, status);
+        }
+    }
+
     pub fn stop(&mut self) {
         // NB: Not resetting the Rust idle_cb to None here because `stop` is
         // likely called from *within* the idle callback, causing a use after

From bd382ee6434e995153302135816685cc9cc80441 Mon Sep 17 00:00:00 2001
From: Brian Anderson <banderson@mozilla.com>
Date: Thu, 15 Aug 2013 19:35:39 -0700
Subject: [PATCH 2/6] std::rt: Touch up idle logic

---
 src/libstd/rt/sched.rs | 52 +++++++++++++++++++++++++++++-------------
 1 file changed, 36 insertions(+), 16 deletions(-)

diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index 58e1676b42427..c0481d06493dd 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -79,7 +79,7 @@ pub struct Scheduler {
     /// A fast XorShift rng for scheduler use
     rng: XorShiftRng,
     /// An IdleWatcher
-    idle_watcher: IdleWatcher,
+    idle_watcher: Option<IdleWatcher>,
     /// A flag to indicate whether or not the idle callback is active.
     idle_flag: bool
 }
@@ -128,9 +128,6 @@ impl Scheduler {
                        friend: Option<SchedHandle>)
         -> Scheduler {
 
-        let mut event_loop = event_loop;
-        let idle_watcher = IdleWatcher::new(event_loop.uvio.uv_loop());
-
         Scheduler {
             sleeper_list: sleeper_list,
             message_queue: MessageQueue::new(),
@@ -146,8 +143,8 @@ impl Scheduler {
             run_anything: run_anything,
             friend_handle: friend,
             rng: XorShiftRng::new(),
-            idle_watcher: idle_watcher,
-            idle_flag: true
+            idle_watcher: None,
+            idle_flag: false
         }
     }
 
@@ -175,9 +172,7 @@ impl Scheduler {
         // Before starting our first task, make sure the idle callback
         // is active. As we do not start in the sleep state this is
         // important.
-        do this.idle_watcher.start |_idle_watcher, _status| {
-            Scheduler::run_sched_once();
-        }
+        this.activate_idle();
 
         // Now, as far as all the scheduler state is concerned, we are
         // inside the "scheduler" context. So we can act like the
@@ -318,23 +313,43 @@ impl Scheduler {
         Local::put(sched);
     }
 
-    fn activate_idle(&mut self) {        
+    fn activate_idle(&mut self) {
+        rtdebug!("activating idle");
         if self.idle_flag {
+            rtassert!(self.idle_watcher.is_some());
             rtdebug!("idle flag already set, not reactivating idle watcher");
         } else {
             rtdebug!("idle flag was false, reactivating idle watcher");
             self.idle_flag = true;
-            self.idle_watcher.restart();
+            if self.idle_watcher.is_none() {
+                // There's no idle handle yet. Create one
+                let mut idle_watcher = IdleWatcher::new(self.event_loop.uvio.uv_loop());
+                do idle_watcher.start |_idle_watcher, _status| {
+                    Scheduler::run_sched_once();
+                }
+                self.idle_watcher = Some(idle_watcher);
+            } else {
+                self.idle_watcher.get_mut_ref().restart();
+            }
         }            
     }
 
     fn pause_idle(&mut self) {
-        if !self.idle_flag {
-            rtdebug!("idle flag false, not stopping idle watcher");
+        rtassert!(self.idle_watcher.is_some());
+        rtassert!(self.idle_flag);
+
+        rtdebug!("stopping idle watcher");
+
+        self.idle_flag = false;
+        if !self.no_sleep {
+            self.idle_watcher.get_mut_ref().stop();
         } else {
-            rtdebug!("idle flag true, stopping idle watcher");
-            self.idle_flag = false;
-            self.idle_watcher.stop();
+            rtdebug!("closing idle watcher");
+            // The scheduler is trying to exit. Destroy the idle watcher
+            // to drop the reference to the event loop.
+            let mut idle_watcher = self.idle_watcher.take_unwrap();
+            idle_watcher.stop();
+            idle_watcher.close(||());
         }
     }
 
@@ -357,10 +372,15 @@ impl Scheduler {
 
         let this = self;
 
+        rtdebug!("enqueuing task");
+
         // We push the task onto our local queue clone.
         this.work_queue.push(task);
 //        this.event_loop.callback(Scheduler::run_sched_once);
 
+        // There is definitely work to be done later. Make sure we wake up for it.
+        this.activate_idle();
+
         // We've made work available. Notify a
         // sleeping scheduler.
 

From 88d8baa76b0c03f6aceeb1f92120af664047ac02 Mon Sep 17 00:00:00 2001
From: Brian Anderson <banderson@mozilla.com>
Date: Thu, 15 Aug 2013 22:48:35 -0700
Subject: [PATCH 3/6] std::rt: Fix a race in UvRemoteCallback's dtor that
 misses callbacks

Full description in comments.
---
 src/libstd/rt/sched.rs   | 46 ++++++++++++++++++++++++++++++++++++++++
 src/libstd/rt/uv/uvio.rs | 35 ++++++++++++++++++++++++------
 2 files changed, 74 insertions(+), 7 deletions(-)

diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index c0481d06493dd..b07591183b740 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -440,6 +440,7 @@ impl Scheduler {
 //                return Some(this);
             }
             Some(Shutdown) => {
+                rtdebug!("shutting down");
 //                this.event_loop.callback(Scheduler::run_sched_once);
                 if this.sleepy {
                     // There may be an outstanding handle on the
@@ -1166,6 +1167,51 @@ mod test {
         }
     }
 
+    // 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::uv::uvio::UvEventLoop;
+        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(
+                    ~UvEventLoop::new(),
+                    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::*;
diff --git a/src/libstd/rt/uv/uvio.rs b/src/libstd/rt/uv/uvio.rs
index a26b8a3ad594d..1ad295641a74d 100644
--- a/src/libstd/rt/uv/uvio.rs
+++ b/src/libstd/rt/uv/uvio.rs
@@ -162,14 +162,32 @@ impl UvRemoteCallback {
         let exit_flag_clone = exit_flag.clone();
         let async = do AsyncWatcher::new(loop_) |watcher, status| {
             assert!(status.is_none());
+
+            // The synchronization logic here is subtle. To review, the uv async handle
+            // type promises that, after it is triggered the remote callback is definitely
+            // called at least once. UvRemoteCallback needs to maintain those semantics
+            // while also shutting down cleanly from the dtor. In our case that means that,
+            // when the UvRemoteCallback dtor calls `async.send()`, here `f` is always called
+            // later.
+
+            // In the dtor both the exit flag is set and the async callback fired under a lock.
+            // Here, before calling `f`, we take the lock and check the flag. Because we are
+            // checking the flag before calling `f`, and the flag is set under the same lock
+            // as the send, then if the flag is set then we're guaranteed to call `f` after
+            // the final send.
+
+            // If the check was done after `f()` then there would be a period between that call
+            // and the check where the dtor could be called in the other thread, missing the
+            // final callback while still destroying the handle.
+
+            let should_exit = unsafe { exit_flag_clone.with_imm(|&should_exit| should_exit) };
+
             f();
-            unsafe {
-                do exit_flag_clone.with_imm |&should_exit| {
-                    if should_exit {
-                        watcher.close(||());
-                    }
-                }
+
+            if should_exit {
+                watcher.close(||());
             }
+
         };
         UvRemoteCallback {
             async: async,
@@ -218,7 +236,10 @@ mod test_remote {
                 let tube_clone = tube_clone.clone();
                 let tube_clone_cell = Cell::new(tube_clone);
                 let remote = do sched.event_loop.remote_callback {
-                    tube_clone_cell.take().send(1);
+                    // This could be called multiple times
+                    if !tube_clone_cell.is_empty() {
+                        tube_clone_cell.take().send(1);
+                    }
                 };
                 remote_cell.put_back(remote);
             }

From f83835b0e7680188b46db5869f49672fef33ff40 Mon Sep 17 00:00:00 2001
From: toddaaro <github@opprobrio.us>
Date: Fri, 16 Aug 2013 13:41:30 -0700
Subject: [PATCH 4/6] Moved the logic for a pausible idle callback into a new
 type - PausibleIdleCallback and placed the appropriate signatures in rtio and
 implementation into uvio.

---
 src/libstd/rt/rtio.rs    | 13 +++---
 src/libstd/rt/sched.rs   | 96 ++++++++++------------------------------
 src/libstd/rt/uv/uvio.rs | 90 +++++++++++++++++++++++++++++--------
 3 files changed, 103 insertions(+), 96 deletions(-)

diff --git a/src/libstd/rt/rtio.rs b/src/libstd/rt/rtio.rs
index a7c794fb5f142..e29c30ba0334a 100644
--- a/src/libstd/rt/rtio.rs
+++ b/src/libstd/rt/rtio.rs
@@ -24,10 +24,12 @@ pub type RtioTcpStreamObject = uvio::UvTcpStream;
 pub type RtioTcpListenerObject = uvio::UvTcpListener;
 pub type RtioUdpSocketObject = uvio::UvUdpSocket;
 pub type RtioTimerObject = uvio::UvTimer;
+pub type PausibleIdleCallback = uvio::UvPausibleIdleCallback;
 
 pub trait EventLoop {
     fn run(&mut self);
     fn callback(&mut self, ~fn());
+    fn pausible_idle_callback(&mut self) -> ~PausibleIdleCallback;
     fn callback_ms(&mut self, ms: u64, ~fn());
     fn remote_callback(&mut self, ~fn()) -> ~RemoteCallbackObject;
     /// The asynchronous I/O services. Not all event loops may provide one
@@ -35,11 +37,12 @@ pub trait EventLoop {
 }
 
 pub trait RemoteCallback {
-    /// Trigger the remote callback. Note that the number of times the callback
-    /// is run is not guaranteed. All that is guaranteed is that, after calling 'fire',
-    /// the callback will be called at least once, but multiple callbacks may be coalesced
-    /// and callbacks may be called more often requested. Destruction also triggers the
-    /// callback.
+    /// Trigger the remote callback. Note that the number of times the
+    /// callback is run is not guaranteed. All that is guaranteed is
+    /// that, after calling 'fire', the callback will be called at
+    /// least once, but multiple callbacks may be coalesced and
+    /// callbacks may be called more often requested. Destruction also
+    /// triggers the callback.
     fn fire(&mut self);
 }
 
diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index b07591183b740..77e3a913e2f5a 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -23,14 +23,13 @@ use super::message_queue::MessageQueue;
 use rt::kill::BlockedTask;
 use rt::local_ptr;
 use rt::local::Local;
-use rt::rtio::RemoteCallback;
+use rt::rtio::{RemoteCallback, PausibleIdleCallback};
 use rt::metrics::SchedMetrics;
 use borrow::{to_uint};
 use cell::Cell;
 use rand::{XorShiftRng, RngUtil};
 use iterator::{range};
 use vec::{OwnedVector};
-use rt::uv::idle::IdleWatcher;
 
 /// The Scheduler is responsible for coordinating execution of Coroutines
 /// on a single thread. When the scheduler is running it is owned by
@@ -78,10 +77,13 @@ pub struct Scheduler {
     friend_handle: Option<SchedHandle>,
     /// A fast XorShift rng for scheduler use
     rng: XorShiftRng,
-    /// An IdleWatcher
-    idle_watcher: Option<IdleWatcher>,
-    /// A flag to indicate whether or not the idle callback is active.
-    idle_flag: bool
+    /// A toggleable idle callback
+    idle_callback: ~PausibleIdleCallback
+}
+
+enum CleanupJob {
+    DoNothing,
+    GiveTask(~Task, UnsafeTaskReceiver)
 }
 
 pub struct SchedHandle {
@@ -97,11 +99,6 @@ pub enum SchedMessage {
     TaskFromFriend(~Task)
 }
 
-enum CleanupJob {
-    DoNothing,
-    GiveTask(~Task, UnsafeTaskReceiver)
-}
-
 impl Scheduler {
 
     pub fn sched_id(&self) -> uint { to_uint(self) }
@@ -126,7 +123,10 @@ impl Scheduler {
                        sleeper_list: SleeperList,
                        run_anything: bool,
                        friend: Option<SchedHandle>)
-        -> Scheduler {
+        -> Scheduler {        
+
+        let mut event_loop = event_loop;
+        let idle_callback = event_loop.pausible_idle_callback();
 
         Scheduler {
             sleeper_list: sleeper_list,
@@ -142,9 +142,8 @@ impl Scheduler {
             metrics: SchedMetrics::new(),
             run_anything: run_anything,
             friend_handle: friend,
-            rng: XorShiftRng::new(),
-            idle_watcher: None,
-            idle_flag: false
+            rng: XorShiftRng::new(),            
+            idle_callback: idle_callback
         }
     }
 
@@ -172,7 +171,7 @@ impl Scheduler {
         // Before starting our first task, make sure the idle callback
         // is active. As we do not start in the sleep state this is
         // important.
-        this.activate_idle();
+        this.idle_callback.start(Scheduler::run_sched_once);
 
         // Now, as far as all the scheduler state is concerned, we are
         // inside the "scheduler" context. So we can act like the
@@ -194,7 +193,7 @@ impl Scheduler {
         // 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 = Local::take::<Task>();
+        let mut stask = Local::take::<Task>();        
 
         rtdebug!("stopping scheduler %u", stask.sched.get_ref().sched_id());
 
@@ -202,6 +201,9 @@ impl Scheduler {
         let message = stask.sched.get_mut_ref().message_queue.pop();
         assert!(message.is_none());
 
+        // Close the idle callback.
+        stask.sched.get_mut_ref().idle_callback.close();
+
         stask.destroyed = true;
     }
 
@@ -211,11 +213,6 @@ impl Scheduler {
 
         let mut self_sched = self;
 
-        // Always run through the scheduler loop at least once so that
-        // we enter the sleep state and can then be woken up by other
-        // schedulers.
-//        self_sched.event_loop.callback(Scheduler::run_sched_once);
-
         // 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"
@@ -252,7 +249,7 @@ impl Scheduler {
 
         // Assume that we need to continue idling unless we reach the
         // end of this function without performing an action.
-        sched.activate_idle();
+        sched.idle_callback.resume();
 
         // Our first task is to read mail to see if we have important
         // messages.
@@ -300,12 +297,12 @@ impl Scheduler {
             let handle = sched.make_handle();
             sched.sleeper_list.push(handle);
             // Since we are sleeping, deactivate the idle callback.
-            sched.pause_idle();
+            sched.idle_callback.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.pause_idle();
+            sched.idle_callback.pause();
         }
 
         // Finished a cycle without using the Scheduler. Place it back
@@ -313,46 +310,6 @@ impl Scheduler {
         Local::put(sched);
     }
 
-    fn activate_idle(&mut self) {
-        rtdebug!("activating idle");
-        if self.idle_flag {
-            rtassert!(self.idle_watcher.is_some());
-            rtdebug!("idle flag already set, not reactivating idle watcher");
-        } else {
-            rtdebug!("idle flag was false, reactivating idle watcher");
-            self.idle_flag = true;
-            if self.idle_watcher.is_none() {
-                // There's no idle handle yet. Create one
-                let mut idle_watcher = IdleWatcher::new(self.event_loop.uvio.uv_loop());
-                do idle_watcher.start |_idle_watcher, _status| {
-                    Scheduler::run_sched_once();
-                }
-                self.idle_watcher = Some(idle_watcher);
-            } else {
-                self.idle_watcher.get_mut_ref().restart();
-            }
-        }            
-    }
-
-    fn pause_idle(&mut self) {
-        rtassert!(self.idle_watcher.is_some());
-        rtassert!(self.idle_flag);
-
-        rtdebug!("stopping idle watcher");
-
-        self.idle_flag = false;
-        if !self.no_sleep {
-            self.idle_watcher.get_mut_ref().stop();
-        } else {
-            rtdebug!("closing idle watcher");
-            // The scheduler is trying to exit. Destroy the idle watcher
-            // to drop the reference to the event loop.
-            let mut idle_watcher = self.idle_watcher.take_unwrap();
-            idle_watcher.stop();
-            idle_watcher.close(||());
-        }
-    }
-
     pub fn make_handle(&mut self) -> SchedHandle {
         let remote = self.event_loop.remote_callback(Scheduler::run_sched_once);
 
@@ -376,10 +333,9 @@ impl Scheduler {
 
         // We push the task onto our local queue clone.
         this.work_queue.push(task);
-//        this.event_loop.callback(Scheduler::run_sched_once);
 
         // There is definitely work to be done later. Make sure we wake up for it.
-        this.activate_idle();
+        this.idle_callback.resume();
 
         // We've made work available. Notify a
         // sleeping scheduler.
@@ -420,28 +376,23 @@ impl Scheduler {
         let mut this = self;
         match this.message_queue.pop() {
             Some(PinnedTask(task)) => {
-//                this.event_loop.callback(Scheduler::run_sched_once);
                 let mut task = task;
                 task.give_home(Sched(this.make_handle()));
                 this.resume_task_immediately(task);
                 return None;
             }
             Some(TaskFromFriend(task)) => {
-//                this.event_loop.callback(Scheduler::run_sched_once);
                 rtdebug!("got a task from a friend. lovely!");
                 this.sched_schedule_task(task).map_move(Local::put);
                 return None;
             }
             Some(Wake) => {
-//                this.event_loop.callback(Scheduler::run_sched_once);
                 this.sleepy = false;
                 Local::put(this);
                 return None;
-//                return Some(this);
             }
             Some(Shutdown) => {
                 rtdebug!("shutting down");
-//                this.event_loop.callback(Scheduler::run_sched_once);
                 if this.sleepy {
                     // There may be an outstanding handle on the
                     // sleeper list.  Pop them all to make sure that's
@@ -463,7 +414,6 @@ impl Scheduler {
 
                 Local::put(this);
                 return None;
-//                return Some(this);
             }
             None => {
                 return Some(this);
diff --git a/src/libstd/rt/uv/uvio.rs b/src/libstd/rt/uv/uvio.rs
index 1ad295641a74d..12a6f6f37f76b 100644
--- a/src/libstd/rt/uv/uvio.rs
+++ b/src/libstd/rt/uv/uvio.rs
@@ -116,6 +116,15 @@ impl EventLoop for UvEventLoop {
         }
     }
 
+    fn pausible_idle_callback(&mut self) -> ~PausibleIdleCallback {
+        let idle_watcher = IdleWatcher::new(self.uvio.uv_loop());
+        return ~UvPausibleIdleCallback {
+            watcher: idle_watcher,
+            idle_flag: false,
+            closed: false
+        };
+    }
+
     fn callback_ms(&mut self, ms: u64, f: ~fn()) {
         let mut timer =  TimerWatcher::new(self.uvio.uv_loop());
         do timer.start(ms, 0) |timer, status| {
@@ -134,6 +143,44 @@ impl EventLoop for UvEventLoop {
     }
 }
 
+pub struct UvPausibleIdleCallback {
+    watcher: IdleWatcher,
+    idle_flag: bool,
+    closed: bool
+}
+
+impl UvPausibleIdleCallback {
+    #[inline]
+    pub fn start(&mut self, f: ~fn()) {
+        do self.watcher.start |_idle_watcher, _status| {
+            f();
+        };
+        self.idle_flag = true;
+    }
+    #[inline]
+    pub fn pause(&mut self) {
+        if self.idle_flag == true {
+            self.watcher.stop();
+            self.idle_flag = false;
+        }
+    }
+    #[inline]
+    pub fn resume(&mut self) {
+        if self.idle_flag == false {
+            self.watcher.restart();
+            self.idle_flag = true;
+        }
+    }
+    #[inline]
+    pub fn close(&mut self) {
+        self.pause();
+        if !self.closed {
+            self.closed = true;
+            self.watcher.close(||());
+        }
+    }                
+}
+
 #[test]
 fn test_callback_run_once() {
     do run_in_bare_thread {
@@ -163,24 +210,31 @@ impl UvRemoteCallback {
         let async = do AsyncWatcher::new(loop_) |watcher, status| {
             assert!(status.is_none());
 
-            // The synchronization logic here is subtle. To review, the uv async handle
-            // type promises that, after it is triggered the remote callback is definitely
-            // called at least once. UvRemoteCallback needs to maintain those semantics
-            // while also shutting down cleanly from the dtor. In our case that means that,
-            // when the UvRemoteCallback dtor calls `async.send()`, here `f` is always called
-            // later.
-
-            // In the dtor both the exit flag is set and the async callback fired under a lock.
-            // Here, before calling `f`, we take the lock and check the flag. Because we are
-            // checking the flag before calling `f`, and the flag is set under the same lock
-            // as the send, then if the flag is set then we're guaranteed to call `f` after
-            // the final send.
-
-            // If the check was done after `f()` then there would be a period between that call
-            // and the check where the dtor could be called in the other thread, missing the
-            // final callback while still destroying the handle.
-
-            let should_exit = unsafe { exit_flag_clone.with_imm(|&should_exit| should_exit) };
+            // The synchronization logic here is subtle. To review,
+            // the uv async handle type promises that, after it is
+            // triggered the remote callback is definitely called at
+            // least once. UvRemoteCallback needs to maintain those
+            // semantics while also shutting down cleanly from the
+            // dtor. In our case that means that, when the
+            // UvRemoteCallback dtor calls `async.send()`, here `f` is
+            // always called later.
+
+            // In the dtor both the exit flag is set and the async
+            // callback fired under a lock.  Here, before calling `f`,
+            // we take the lock and check the flag. Because we are
+            // checking the flag before calling `f`, and the flag is
+            // set under the same lock as the send, then if the flag
+            // is set then we're guaranteed to call `f` after the
+            // final send.
+
+            // If the check was done after `f()` then there would be a
+            // period between that call and the check where the dtor
+            // could be called in the other thread, missing the final
+            // callback while still destroying the handle.
+
+            let should_exit = unsafe { 
+                exit_flag_clone.with_imm(|&should_exit| should_exit) 
+            };
 
             f();
 

From 20213fcca420f6b6a0b8990a4647edad39148c07 Mon Sep 17 00:00:00 2001
From: toddaaro <github@opprobrio.us>
Date: Thu, 15 Aug 2013 19:46:23 -0700
Subject: [PATCH 5/6] A round of code cleaning for the primary scheduler code.
 Comments have been updated, a minor amount of support type restructing has
 happened, methods have been reordered, and some duplicate code has been
 purged.

---
 src/libstd/rt/sched.rs | 471 +++++++++++++++++------------------------
 src/libstd/rt/task.rs  |   2 +-
 src/libstd/rt/util.rs  |   3 +-
 3 files changed, 201 insertions(+), 275 deletions(-)

diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index 77e3a913e2f5a..6a624b1c13582 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -31,10 +31,11 @@ use rand::{XorShiftRng, RngUtil};
 use iterator::{range};
 use vec::{OwnedVector};
 
-/// The Scheduler is responsible for coordinating execution of Coroutines
-/// on a single thread. When the scheduler is running it is owned by
-/// thread local storage and the running task is owned by the
-/// scheduler.
+/// 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.
@@ -64,11 +65,12 @@ pub struct Scheduler {
     stack_pool: StackPool,
     /// The event loop used to drive the scheduler and perform I/O
     event_loop: ~EventLoopObject,
-    /// The scheduler runs on a special task.
+    /// The scheduler runs on a special task. When it is not running
+    /// it is stored here instead of the work queue.
     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>,
+    cleanup_job: Option<CleanupJob>,
     metrics: SchedMetrics,
     /// Should this scheduler run any task, or only pinned tasks?
     run_anything: bool,
@@ -81,27 +83,9 @@ pub struct Scheduler {
     idle_callback: ~PausibleIdleCallback
 }
 
-enum CleanupJob {
-    DoNothing,
-    GiveTask(~Task, UnsafeTaskReceiver)
-}
-
-pub struct SchedHandle {
-    priv remote: ~RemoteCallbackObject,
-    priv queue: MessageQueue<SchedMessage>,
-    sched_id: uint
-}
-
-pub enum SchedMessage {
-    Wake,
-    Shutdown,
-    PinnedTask(~Task),
-    TaskFromFriend(~Task)
-}
-
 impl Scheduler {
 
-    pub fn sched_id(&self) -> uint { to_uint(self) }
+    // * Initialization Functions
 
     pub fn new(event_loop: ~EventLoopObject,
                work_queue: WorkQueue<~Task>,
@@ -115,8 +99,6 @@ impl Scheduler {
 
     }
 
-    // When you create a scheduler it isn't yet "in" a task, so the
-    // task field is None.
     pub fn new_special(event_loop: ~EventLoopObject,
                        work_queue: WorkQueue<~Task>,
                        work_queues: ~[WorkQueue<~Task>],
@@ -185,7 +167,6 @@ impl Scheduler {
         let sched = Local::take::<Scheduler>();
 
         rtdebug!("starting scheduler %u", sched.sched_id());
-
         sched.run();
 
         // Now that we are done with the scheduler, clean up the
@@ -231,11 +212,11 @@ impl Scheduler {
         }
     }
 
-    // One iteration of the scheduler loop, always run at least once.
+    // * 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 do not return, or
+    // 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.
@@ -251,39 +232,18 @@ impl Scheduler {
         // end of this function without performing an action.
         sched.idle_callback.resume();
 
-        // Our first task is to read mail to see if we have important
-        // messages.
-
-        // 1) A wake message is easy, mutate sched struct and return
-        //    it.
-        // 2) A shutdown is also easy, shutdown.
-        // 3) A pinned task - we resume immediately and do not return
-        //    here.
-        // 4) A message from another scheduler with a non-homed task
-        //    to run here.
-
-        let result = sched.interpret_message_queue();
-        let sched = match result {
-            Some(sched) => {
-                // We did not resume a task, so we returned.
-                sched
-            }
-            None => {
-                return;
-            }
+        // First we check for scheduler messages, these are higher
+        // priority than regular tasks.
+        let sched = match sched.interpret_message_queue() {
+            Some(sched) => sched,
+            None => return
         };
 
-        // Second activity is to try resuming a task from the queue.
-
-        let result = sched.do_work();
-        let mut sched = match result {
-            Some(sched) => {
-                // Failed to dequeue a task, so we return.
-                sched
-            }
-            None => {
-                return;
-            }
+        // This helper will use a randomized work-stealing algorithm
+        // to find work.
+        let mut sched = match sched.do_work() {
+            Some(sched) => sched,
+            None => return
         };
 
         // If we got here then there was no work to do.
@@ -310,67 +270,10 @@ impl Scheduler {
         Local::put(sched);
     }
 
-    pub 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(),
-            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.
-    pub fn enqueue_task(&mut self, task: ~Task) {
-
-        let this = self;
-
-        rtdebug!("enqueuing task");
-
-        // We push the task onto our local queue clone.
-        this.work_queue.push(task);
-
-        // There is definitely work to be done later. Make sure we wake up for it.
-        this.idle_callback.resume();
-
-        // 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 */) }
-        };
-    }
-
-    /// 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_move |task| {
-            self.enqueue_task(task);
-        };
-    }
-
-    // * Scheduler-context operations
-
     // This function returns None if the scheduler is "used", or it
-    // returns the still-available scheduler. Note: currently
-    // considers *any* message receive a use and returns None.
+    // 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(~self) -> Option<~Scheduler> {
 
         let mut this = self;
@@ -383,7 +286,8 @@ impl Scheduler {
             }
             Some(TaskFromFriend(task)) => {
                 rtdebug!("got a task from a friend. lovely!");
-                this.sched_schedule_task(task).map_move(Local::put);
+                this.process_task(task,
+                                  Scheduler::resume_task_immediately_cl).map_move(Local::put);
                 return None;
             }
             Some(Wake) => {
@@ -411,7 +315,6 @@ impl Scheduler {
                 // event loop references we will shut down.
                 this.no_sleep = true;
                 this.sleepy = false;
-
                 Local::put(this);
                 return None;
             }
@@ -421,30 +324,19 @@ impl Scheduler {
         }
     }
 
-    /// Given an input Coroutine sends it back to its home scheduler.
-    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");
-            }
-        }
-    }
+    fn do_work(~self) -> Option<~Scheduler> {
+        let mut this = self;
 
-    /// 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));
+        rtdebug!("scheduler calling do work");
+        match this.find_work() {
+            Some(task) => {
+                rtdebug!("found some work! processing the task");
+                return this.process_task(task,
+                                         Scheduler::resume_task_immediately_cl);
             }
             None => {
-                rtabort!("tried to send task to a friend but scheduler has no friends");
+                rtdebug!("no work was found, returning the scheduler struct");
+                return Some(this);
             }
         }
     }
@@ -468,8 +360,8 @@ impl Scheduler {
             None => {
                 // Our naive stealing, try kinda hard.
                 rtdebug!("scheduler trying to steal");
-                let _len = self.work_queues.len();
-                return self.try_steals(2);
+                let len = self.work_queues.len();
+                return self.try_steals(len/2);
             }
         }
     }
@@ -483,7 +375,8 @@ impl Scheduler {
             let work_queues = &mut self.work_queues;
             match work_queues[index].steal() {
                 Some(task) => {
-                    rtdebug!("found task by stealing"); return Some(task)
+                    rtdebug!("found task by stealing");
+                    return Some(task)
                 }
                 None => ()
             }
@@ -492,8 +385,11 @@ impl Scheduler {
         return None;
     }
 
-    // Given a task, execute it correctly.
-    fn process_task(~self, task: ~Task) -> Option<~Scheduler> {
+    // * Task Routing Functions - Make sure tasks send up in the right
+    // place.
+
+    fn process_task(~self, task: ~Task,
+                    schedule_fn: SchedulingFn) -> Option<~Scheduler> {
         let mut this = self;
         let mut task = task;
 
@@ -510,15 +406,13 @@ impl Scheduler {
                 } else {
                     rtdebug!("running task here");
                     task.give_home(Sched(home_handle));
-                    this.resume_task_immediately(task);
-                    return None;
+                    return schedule_fn(this, task);
                 }
             }
             AnySched if this.run_anything => {
                 rtdebug!("running anysched task here");
                 task.give_home(AnySched);
-                this.resume_task_immediately(task);
-                return None;
+                return schedule_fn(this, task);
             }
             AnySched => {
                 rtdebug!("sending task to friend");
@@ -529,98 +423,71 @@ impl Scheduler {
         }
     }
 
-    // Bundle the helpers together.
-    fn do_work(~self) -> Option<~Scheduler> {
-        let mut this = self;
-
-        rtdebug!("scheduler calling do work");
-        match this.find_work() {
-            Some(task) => {
-                rtdebug!("found some work! processing the task");
-                return this.process_task(task);
+    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));
             }
-            None => {
-                rtdebug!("no work was found, returning the scheduler struct");
-                return Some(this);
+            AnySched => {
+                        rtabort!("error: cannot send anysched task home");
             }
         }
     }
 
-    /// 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(~self) {
-        // Similar to deschedule running task and then, but cannot go through
-        // the task-blocking path. The task is already dying.
-        let mut this = self;
-        let stask = this.sched_task.take_unwrap();
-        do this.change_task_context(stask) |sched, mut dead_task| {
-            let coroutine = dead_task.coroutine.take_unwrap();
-            coroutine.recycle(&mut sched.stack_pool);
+    /// 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");
+            }
         }
     }
 
-    // Scheduling a task requires a few checks to make sure the task
-    // ends up in the appropriate location. The run_anything flag on
-    // the scheduler and the home on the task need to be checked. This
-    // helper performs that check. It takes a function that specifies
-    // how to queue the the provided task if that is the correct
-    // action. This is a "core" function that requires handling the
-    // returned Option correctly.
-
-    pub fn schedule_task(~self, task: ~Task,
-                         schedule_fn: ~fn(sched: ~Scheduler, task: ~Task))
-        -> Option<~Scheduler> {
-
-        // is the task home?
-        let is_home = task.is_home_no_tls(&self);
+    /// 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) {
 
-        // does the task have a home?
-        let homed = task.homed();
+        let this = self;
 
-        let mut this = self;
+        // We push the task onto our local queue clone.
+        this.work_queue.push(task);
+        this.idle_callback.resume();
 
-        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
-            rtdebug!("task: %u is on ok sched, executing", to_uint(task));
-            schedule_fn(this, task);
-            return None;
-        } else if !homed && !this.run_anything {
-            // the task isn't homed, but it can't be run here
-            this.send_to_friend(task);
-            return Some(this);
-        } else {
-            // task isn't home, so don't run it here, send it home
-            Scheduler::send_task_home(task);
-            return Some(this);
-        }
-    }
+        // We've made work available. Notify a
+        // sleeping scheduler.
 
-    // There are two contexts in which schedule_task can be called:
-    // inside the scheduler, and inside a task. These contexts handle
-    // executing the task slightly differently. In the scheduler
-    // context case we want to receive the scheduler as an input, and
-    // manually deal with the option. In the task context case we want
-    // to use TLS to find the scheduler, and deal with the option
-    // inside the helper.
-
-    pub fn sched_schedule_task(~self, task: ~Task) -> Option<~Scheduler> {
-        do self.schedule_task(task) |sched, next_task| {
-            sched.resume_task_immediately(next_task);
-        }
+        // XXX: perf. Check for a sleeper without
+        // synchronizing memory.  It's not critical
+        // that we always find it.
+        match this.sleeper_list.pop() {
+            Some(handle) => {
+                        let mut handle = handle;
+                handle.send(Wake)
+            }
+            None => { (/* pass */) }
+        };
     }
 
-    // Task context case - use TLS.
-    pub fn run_task(task: ~Task) {
-        let sched = Local::take::<Scheduler>();
-        let opt = do sched.schedule_task(task) |sched, next_task| {
-            do sched.switch_running_tasks_and_then(next_task) |sched, last_task| {
-                sched.enqueue_blocked_task(last_task);
-            }
+    /// 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_move |task| {
+            self.enqueue_task(task);
         };
-        opt.map_move(Local::put);
     }
 
+    // * 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
@@ -650,7 +517,7 @@ impl Scheduler {
 
         // The current task is placed inside an enum with the cleanup
         // function. This enum is then placed inside the scheduler.
-        this.enqueue_cleanup_job(GiveTask(current_task, f_opaque));
+        this.cleanup_job = Some(CleanupJob::new(current_task, f_opaque));
 
         // The scheduler is then placed inside the next task.
         let mut next_task = next_task;
@@ -666,12 +533,9 @@ impl Scheduler {
                 transmute_mut_region(*next_task.sched.get_mut_ref());
 
             let current_task: &mut Task = match sched.cleanup_job {
-                Some(GiveTask(ref task, _)) => {
+                Some(CleanupJob { task: ref task, _ }) => {
                     transmute_mut_region(*transmute_mut_unsafe(task))
                 }
-                Some(DoNothing) => {
-                    rtabort!("no next task");
-                }
                 None => {
                     rtabort!("no cleanup job");
                 }
@@ -705,19 +569,42 @@ impl Scheduler {
         }
     }
 
-    // Old API for task manipulation implemented over the new core
-    // function.
+    // 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) {
+    pub fn resume_task_immediately(~self, task: ~Task) -> Option<~Scheduler> {
         do self.change_task_context(task) |sched, stask| {
             sched.sched_task = Some(stask);
         }
+        return None;
+    }
+
+    fn resume_task_immediately_cl(sched: ~Scheduler,
+                                  task: ~Task) -> Option<~Scheduler> {
+        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),
+            Some(task) => { self.resume_task_immediately(task); }
+            None => Local::put(self)
         };
     }
 
@@ -756,54 +643,75 @@ impl Scheduler {
         }
     }
 
-    // A helper that looks up the scheduler and runs a task later by
-    // enqueuing it.
+    fn switch_task(sched: ~Scheduler, task: ~Task) -> Option<~Scheduler> {
+        do sched.switch_running_tasks_and_then(task) |sched, last_task| {
+            sched.enqueue_blocked_task(last_task);
+        };
+        return None;
+    }
+
+    // * 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(~self) {
+        // Similar to deschedule running task and then, but cannot go through
+        // the task-blocking path. The task is already dying.
+        let mut this = self;
+        let stask = this.sched_task.take_unwrap();
+        do this.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 = Local::take::<Scheduler>();
+        sched.process_task(task, Scheduler::switch_task).map_move(Local::put);
+    }
+
     pub fn run_task_later(next_task: ~Task) {
-        // We aren't performing a scheduler operation, so we want to
-        // put the Scheduler back when we finish.
         let next_task = Cell::new(next_task);
         do Local::borrow::<Scheduler,()> |sched| {
             sched.enqueue_task(next_task.take());
         };
     }
 
-    // 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))
-        }
-    }
+    // * Utility Functions
 
-    pub fn enqueue_cleanup_job(&mut self, job: CleanupJob) {
-        self.cleanup_job = Some(job);
-    }
+    pub fn sched_id(&self) -> uint { to_uint(self) }
 
     pub fn run_cleanup_job(&mut self) {
-        rtdebug!("running cleanup job");
         let cleanup_job = self.cleanup_job.take_unwrap();
-        match cleanup_job {
-            DoNothing => { }
-            GiveTask(task, f) => f.to_fn()(self, task)
-        }
+        cleanup_job.run(self);
+    }
+
+    pub 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(),
+            sched_id: self.sched_id()
+        };
     }
 }
 
-// The cases for the below function.
-enum ResumeAction {
-    SendHome,
-    Requeue,
-    ResumeNow,
-    Homeless
+// Supporting types
+
+type SchedulingFn = ~fn(~Scheduler, ~Task) -> Option<~Scheduler>;
+
+pub enum SchedMessage {
+    Wake,
+    Shutdown,
+    PinnedTask(~Task),
+    TaskFromFriend(~Task)
+}
+
+pub struct SchedHandle {
+    priv remote: ~RemoteCallbackObject,
+    priv queue: MessageQueue<SchedMessage>,
+    sched_id: uint
 }
 
 impl SchedHandle {
@@ -813,6 +721,25 @@ impl SchedHandle {
     }
 }
 
+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;
diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs
index 708166518bb89..698c59805a4cf 100644
--- a/src/libstd/rt/task.rs
+++ b/src/libstd/rt/task.rs
@@ -360,7 +360,7 @@ impl Coroutine {
 
                 // Again - might work while safe, or it might not.
                 do Local::borrow::<Scheduler,()> |sched| {
-                    (sched).run_cleanup_job();
+                    sched.run_cleanup_job();
                 }
 
                 // To call the run method on a task we need a direct
diff --git a/src/libstd/rt/util.rs b/src/libstd/rt/util.rs
index 6280b64ecf51c..8a2541e73b384 100644
--- a/src/libstd/rt/util.rs
+++ b/src/libstd/rt/util.rs
@@ -38,8 +38,7 @@ pub fn default_sched_threads() -> uint {
 pub fn dumb_println(s: &str) {
     use io::WriterUtil;
     let dbg = ::libc::STDERR_FILENO as ::io::fd_t;
-    dbg.write_str(s);
-    dbg.write_str("\n");
+    dbg.write_str(s + "\n");
 }
 
 pub fn abort(msg: &str) -> ! {

From e5ccf13668ed7b66d6efd9a1a03926e98546705d Mon Sep 17 00:00:00 2001
From: toddaaro <github@opprobrio.us>
Date: Mon, 19 Aug 2013 12:01:13 -0700
Subject: [PATCH 6/6] clean whitespace :/

---
 src/libstd/rt/sched.rs   | 6 +++---
 src/libstd/rt/uv/uvio.rs | 6 +++---
 2 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index 6a624b1c13582..a50618ba0ad6d 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -105,7 +105,7 @@ impl Scheduler {
                        sleeper_list: SleeperList,
                        run_anything: bool,
                        friend: Option<SchedHandle>)
-        -> Scheduler {        
+        -> Scheduler {
 
         let mut event_loop = event_loop;
         let idle_callback = event_loop.pausible_idle_callback();
@@ -124,7 +124,7 @@ impl Scheduler {
             metrics: SchedMetrics::new(),
             run_anything: run_anything,
             friend_handle: friend,
-            rng: XorShiftRng::new(),            
+            rng: XorShiftRng::new(),
             idle_callback: idle_callback
         }
     }
@@ -174,7 +174,7 @@ impl Scheduler {
         // 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 = Local::take::<Task>();        
+        let mut stask = Local::take::<Task>();
 
         rtdebug!("stopping scheduler %u", stask.sched.get_ref().sched_id());
 
diff --git a/src/libstd/rt/uv/uvio.rs b/src/libstd/rt/uv/uvio.rs
index 12a6f6f37f76b..290e57942f47a 100644
--- a/src/libstd/rt/uv/uvio.rs
+++ b/src/libstd/rt/uv/uvio.rs
@@ -178,7 +178,7 @@ impl UvPausibleIdleCallback {
             self.closed = true;
             self.watcher.close(||());
         }
-    }                
+    }
 }
 
 #[test]
@@ -232,8 +232,8 @@ impl UvRemoteCallback {
             // could be called in the other thread, missing the final
             // callback while still destroying the handle.
 
-            let should_exit = unsafe { 
-                exit_flag_clone.with_imm(|&should_exit| should_exit) 
+            let should_exit = unsafe {
+                exit_flag_clone.with_imm(|&should_exit| should_exit)
             };
 
             f();