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

Conflicts:
	src/libstd/rt/uvio.rs

29 files changed, 1770 insertions, 364 deletions

diff --git a/src/libstd/rt/comm.rs b/src/libstd/rt/comm.rs
index b00df78f433..82e6d44fe62 100644
--- a/src/libstd/rt/comm.rs
+++ b/src/libstd/rt/comm.rs
@@ -120,13 +120,13 @@ impl<T> ChanOne<T> {
             match oldstate {
                 STATE_BOTH => {
                     // Port is not waiting yet. Nothing to do
-                    do Local::borrow::<Scheduler> |sched| {
+                    do Local::borrow::<Scheduler, ()> |sched| {
                         rtdebug!("non-rendezvous send");
                         sched.metrics.non_rendezvous_sends += 1;
                     }
                 }
                 STATE_ONE => {
-                    do Local::borrow::<Scheduler> |sched| {
+                    do Local::borrow::<Scheduler, ()> |sched| {
                         rtdebug!("rendezvous send");
                         sched.metrics.rendezvous_sends += 1;
                     }
@@ -331,8 +331,8 @@ pub struct Port<T> {
 
 pub fn stream<T: Owned>() -> (Port<T>, Chan<T>) {
     let (pone, cone) = oneshot();
-    let port = Port { next: Cell(pone) };
-    let chan = Chan { next: Cell(cone) };
+    let port = Port { next: Cell::new(pone) };
+    let chan = Chan { next: Cell::new(cone) };
     return (port, chan);
 }
 
@@ -647,7 +647,7 @@ mod test {
     fn oneshot_multi_task_recv_then_send() {
         do run_in_newsched_task {
             let (port, chan) = oneshot::<~int>();
-            let port_cell = Cell(port);
+            let port_cell = Cell::new(port);
             do spawntask_immediately {
                 assert!(port_cell.take().recv() == ~10);
             }
@@ -660,8 +660,8 @@ mod test {
     fn oneshot_multi_task_recv_then_close() {
         do run_in_newsched_task {
             let (port, chan) = oneshot::<~int>();
-            let port_cell = Cell(port);
-            let chan_cell = Cell(chan);
+            let port_cell = Cell::new(port);
+            let chan_cell = Cell::new(chan);
             do spawntask_later {
                 let _cell = chan_cell.take();
             }
@@ -677,7 +677,7 @@ mod test {
         for stress_factor().times {
             do run_in_newsched_task {
                 let (port, chan) = oneshot::<int>();
-                let port_cell = Cell(port);
+                let port_cell = Cell::new(port);
                 let _thread = do spawntask_thread {
                     let _p = port_cell.take();
                 };
@@ -691,8 +691,8 @@ mod test {
         for stress_factor().times {
             do run_in_newsched_task {
                 let (port, chan) = oneshot::<int>();
-                let chan_cell = Cell(chan);
-                let port_cell = Cell(port);
+                let chan_cell = Cell::new(chan);
+                let port_cell = Cell::new(port);
                 let _thread1 = do spawntask_thread {
                     let _p = port_cell.take();
                 };
@@ -709,17 +709,17 @@ mod test {
         for stress_factor().times {
             do run_in_newsched_task {
                 let (port, chan) = oneshot::<int>();
-                let chan_cell = Cell(chan);
-                let port_cell = Cell(port);
+                let chan_cell = Cell::new(chan);
+                let port_cell = Cell::new(port);
                 let _thread1 = do spawntask_thread {
-                    let port_cell = Cell(port_cell.take());
+                    let port_cell = Cell::new(port_cell.take());
                     let res = do spawntask_try {
                         port_cell.take().recv();
                     };
                     assert!(res.is_err());
                 };
                 let _thread2 = do spawntask_thread {
-                    let chan_cell = Cell(chan_cell.take());
+                    let chan_cell = Cell::new(chan_cell.take());
                     do spawntask {
                         chan_cell.take();
                     }
@@ -733,8 +733,8 @@ mod test {
         for stress_factor().times {
             do run_in_newsched_task {
                 let (port, chan) = oneshot::<~int>();
-                let chan_cell = Cell(chan);
-                let port_cell = Cell(port);
+                let chan_cell = Cell::new(chan);
+                let port_cell = Cell::new(port);
                 let _thread1 = do spawntask_thread {
                     chan_cell.take().send(~10);
                 };
@@ -757,7 +757,7 @@ mod test {
                 fn send(chan: Chan<~int>, i: int) {
                     if i == 10 { return }
 
-                    let chan_cell = Cell(chan);
+                    let chan_cell = Cell::new(chan);
                     do spawntask_random {
                         let chan = chan_cell.take();
                         chan.send(~i);
@@ -768,7 +768,7 @@ mod test {
                 fn recv(port: Port<~int>, i: int) {
                     if i == 10 { return }
 
-                    let port_cell = Cell(port);
+                    let port_cell = Cell::new(port);
                     do spawntask_random {
                         let port = port_cell.take();
                         assert!(port.recv() == ~i);
@@ -918,4 +918,3 @@ mod test {
         }
     }
 }
-
diff --git a/src/libstd/rt/context.rs b/src/libstd/rt/context.rs
index 0d011ce42ba..d5ca8473cee 100644
--- a/src/libstd/rt/context.rs
+++ b/src/libstd/rt/context.rs
@@ -27,8 +27,8 @@ pub struct Context {
     regs: ~Registers
 }
 
-pub impl Context {
-    fn empty() -> Context {
+impl Context {
+    pub fn empty() -> Context {
         Context {
             start: None,
             regs: new_regs()
@@ -36,7 +36,7 @@ pub impl Context {
     }
 
     /// Create a new context that will resume execution by running ~fn()
-    fn new(start: ~fn(), stack: &mut StackSegment) -> Context {
+    pub fn new(start: ~fn(), stack: &mut StackSegment) -> Context {
         // XXX: Putting main into a ~ so it's a thin pointer and can
         // be passed to the spawn function.  Another unfortunate
         // allocation
@@ -71,7 +71,7 @@ pub impl Context {
     saving the registers values of the executing thread to a Context
     then loading the registers from a previously saved Context.
     */
-    fn swap(out_context: &mut Context, in_context: &Context) {
+    pub fn swap(out_context: &mut Context, in_context: &Context) {
         let out_regs: &mut Registers = match out_context {
             &Context { regs: ~ref mut r, _ } => r
         };
diff --git a/src/libstd/rt/io/extensions.rs b/src/libstd/rt/io/extensions.rs
index ad9658e48ba..c7c3eadbe21 100644
--- a/src/libstd/rt/io/extensions.rs
+++ b/src/libstd/rt/io/extensions.rs
@@ -342,7 +342,7 @@ impl<T: Reader> ReaderByteConversions for T {
     fn read_le_uint_n(&mut self, nbytes: uint) -> u64 {
         assert!(nbytes > 0 && nbytes <= 8);
 
-        let mut val = 0u64, pos = 0, i = nbytes;
+        let mut (val, pos, i) = (0u64, 0, nbytes);
         while i > 0 {
             val += (self.read_u8() as u64) << pos;
             pos += 8;
@@ -358,7 +358,7 @@ impl<T: Reader> ReaderByteConversions for T {
     fn read_be_uint_n(&mut self, nbytes: uint) -> u64 {
         assert!(nbytes > 0 && nbytes <= 8);
 
-        let mut val = 0u64, i = nbytes;
+        let mut (val, i) = (0u64, nbytes);
         while i > 0 {
             i -= 1;
             val += (self.read_u8() as u64) << i * 8;
@@ -604,7 +604,7 @@ mod test {
     #[test]
     fn read_byte_0_bytes() {
         let mut reader = MockReader::new();
-        let count = Cell(0);
+        let count = Cell::new(0);
         reader.read = |buf| {
             do count.with_mut_ref |count| {
                 if *count == 0 {
@@ -652,7 +652,7 @@ mod test {
     #[test]
     fn read_bytes_partial() {
         let mut reader = MockReader::new();
-        let count = Cell(0);
+        let count = Cell::new(0);
         reader.read = |buf| {
             do count.with_mut_ref |count| {
                 if *count == 0 {
@@ -691,7 +691,7 @@ mod test {
     #[test]
     fn push_bytes_partial() {
         let mut reader = MockReader::new();
-        let count = Cell(0);
+        let count = Cell::new(0);
         reader.read = |buf| {
             do count.with_mut_ref |count| {
                 if *count == 0 {
@@ -725,7 +725,7 @@ mod test {
     #[test]
     fn push_bytes_error() {
         let mut reader = MockReader::new();
-        let count = Cell(0);
+        let count = Cell::new(0);
         reader.read = |buf| {
             do count.with_mut_ref |count| {
                 if *count == 0 {
@@ -752,7 +752,7 @@ mod test {
         // push_bytes unsafely sets the vector length. This is testing that
         // upon failure the length is reset correctly.
         let mut reader = MockReader::new();
-        let count = Cell(0);
+        let count = Cell::new(0);
         reader.read = |buf| {
             do count.with_mut_ref |count| {
                 if *count == 0 {
@@ -778,7 +778,7 @@ mod test {
     #[test]
     fn read_to_end() {
         let mut reader = MockReader::new();
-        let count = Cell(0);
+        let count = Cell::new(0);
         reader.read = |buf| {
             do count.with_mut_ref |count| {
                 if *count == 0 {
@@ -805,7 +805,7 @@ mod test {
     #[ignore(cfg(windows))]
     fn read_to_end_error() {
         let mut reader = MockReader::new();
-        let count = Cell(0);
+        let count = Cell::new(0);
         reader.read = |buf| {
             do count.with_mut_ref |count| {
                 if *count == 0 {
diff --git a/src/libstd/rt/io/file.rs b/src/libstd/rt/io/file.rs
index 1f61cf25fbd..a99f5da032c 100644
--- a/src/libstd/rt/io/file.rs
+++ b/src/libstd/rt/io/file.rs
@@ -75,5 +75,5 @@ fn super_simple_smoke_test_lets_go_read_some_files_and_have_a_good_time() {
     let message = "it's alright. have a good time";
     let filename = &Path("test.txt");
     let mut outstream = FileStream::open(filename, Create, Read).unwrap();
-    outstream.write(message.to_bytes());
+    outstream.write(message.as_bytes());
 }
diff --git a/src/libstd/rt/io/flate.rs b/src/libstd/rt/io/flate.rs
index db2683dc85d..e57b80658ee 100644
--- a/src/libstd/rt/io/flate.rs
+++ b/src/libstd/rt/io/flate.rs
@@ -100,13 +100,15 @@ mod test {
     use super::super::mem::*;
     use super::super::Decorator;
 
+    use str;
+
     #[test]
     #[ignore]
     fn smoke_test() {
         let mem_writer = MemWriter::new();
         let mut deflate_writer = DeflateWriter::new(mem_writer);
         let in_msg = "test";
-        let in_bytes = in_msg.to_bytes();
+        let in_bytes = in_msg.as_bytes();
         deflate_writer.write(in_bytes);
         deflate_writer.flush();
         let buf = deflate_writer.inner().inner();
diff --git a/src/libstd/rt/io/mem.rs b/src/libstd/rt/io/mem.rs
index b2701c1fdc3..bd9cff76e57 100644
--- a/src/libstd/rt/io/mem.rs
+++ b/src/libstd/rt/io/mem.rs
@@ -15,9 +15,10 @@
 //! * Should probably have something like this for strings.
 //! * Should they implement Closable? Would take extra state.
 
+use cmp::min;
 use prelude::*;
 use super::*;
-use cmp::min;
+use vec;
 
 /// Writes to an owned, growable byte vector
 pub struct MemWriter {
diff --git a/src/libstd/rt/io/net/tcp.rs b/src/libstd/rt/io/net/tcp.rs
index f7c03c13a58..3607f781da3 100644
--- a/src/libstd/rt/io/net/tcp.rs
+++ b/src/libstd/rt/io/net/tcp.rs
@@ -287,7 +287,7 @@ mod test {
             do spawntask_immediately {
                 let mut listener = TcpListener::bind(addr);
                 for int::range(0, MAX) |i| {
-                    let stream = Cell(listener.accept());
+                    let stream = Cell::new(listener.accept());
                     rtdebug!("accepted");
                     // Start another task to handle the connection
                     do spawntask_immediately {
@@ -326,7 +326,7 @@ mod test {
             do spawntask_immediately {
                 let mut listener = TcpListener::bind(addr);
                 for int::range(0, MAX) |_| {
-                    let stream = Cell(listener.accept());
+                    let stream = Cell::new(listener.accept());
                     rtdebug!("accepted");
                     // Start another task to handle the connection
                     do spawntask_later {
diff --git a/src/libstd/rt/io/stdio.rs b/src/libstd/rt/io/stdio.rs
index 247fe954408..57bec79563f 100644
--- a/src/libstd/rt/io/stdio.rs
+++ b/src/libstd/rt/io/stdio.rs
@@ -50,4 +50,3 @@ impl Writer for StdWriter {
 
     fn flush(&mut self) { fail!() }
 }
-
diff --git a/src/libstd/rt/join_latch.rs b/src/libstd/rt/join_latch.rs
new file mode 100644
index 00000000000..ad5cf2eb378
--- /dev/null
+++ b/src/libstd/rt/join_latch.rs
@@ -0,0 +1,645 @@
+// 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.
+
+//! The JoinLatch is a concurrent type that establishes the task
+//! tree and propagates failure.
+//!
+//! Each task gets a JoinLatch that is derived from the JoinLatch
+//! of its parent task. Every latch must be released by either calling
+//! the non-blocking `release` method or the task-blocking `wait` method.
+//! Releasing a latch does not complete until all of its child latches
+//! complete.
+//!
+//! Latches carry a `success` flag that is set to `false` during task
+//! failure and is propagated both from children to parents and parents
+//! to children. The status af this flag may be queried for the purposes
+//! of linked failure.
+//!
+//! In addition to failure propagation the task tree serves to keep the
+//! default task schedulers alive. The runtime only sends the shutdown
+//! message to schedulers once the root task exits.
+//!
+//! Under this scheme tasks that terminate before their children become
+//! 'zombies' since they may not exit until their children do. Zombie
+//! tasks are 'tombstoned' as `Tombstone(~JoinLatch)` and the tasks
+//! themselves allowed to terminate.
+//!
+//! XXX: Propagate flag from parents to children.
+//! XXX: Tombstoning actually doesn't work.
+//! XXX: This could probably be done in a way that doesn't leak tombstones
+//!      longer than the life of the child tasks.
+
+use comm::{GenericPort, Peekable, GenericSmartChan};
+use clone::Clone;
+use container::Container;
+use option::{Option, Some, None};
+use ops::Drop;
+use rt::comm::{SharedChan, Port, stream};
+use rt::local::Local;
+use rt::sched::Scheduler;
+use unstable::atomics::{AtomicUint, SeqCst};
+use util;
+use vec::OwnedVector;
+
+// FIXME #7026: Would prefer this to be an enum
+pub struct JoinLatch {
+    priv parent: Option<ParentLink>,
+    priv child: Option<ChildLink>,
+    closed: bool,
+}
+
+// Shared between parents and all their children.
+struct SharedState {
+    /// Reference count, held by a parent and all children.
+    count: AtomicUint,
+    success: bool
+}
+
+struct ParentLink {
+    shared: *mut SharedState,
+    // For communicating with the parent.
+    chan: SharedChan<Message>
+}
+
+struct ChildLink {
+    shared: ~SharedState,
+    // For receiving from children.
+    port: Port<Message>,
+    chan: SharedChan<Message>,
+    // Prevents dropping the child SharedState reference counts multiple times.
+    dropped_child: bool
+}
+
+// Messages from child latches to parent.
+enum Message {
+    Tombstone(~JoinLatch),
+    ChildrenTerminated
+}
+
+impl JoinLatch {
+    pub fn new_root() -> ~JoinLatch {
+        let this = ~JoinLatch {
+            parent: None,
+            child: None,
+            closed: false
+        };
+        rtdebug!("new root latch %x", this.id());
+        return this;
+    }
+
+    fn id(&self) -> uint {
+        unsafe { ::cast::transmute(&*self) }
+    }
+
+    pub fn new_child(&mut self) -> ~JoinLatch {
+        rtassert!(!self.closed);
+
+        if self.child.is_none() {
+            // This is the first time spawning a child
+            let shared = ~SharedState {
+                count: AtomicUint::new(1),
+                success: true
+            };
+            let (port, chan) = stream();
+            let chan = SharedChan::new(chan);
+            let child = ChildLink {
+                shared: shared,
+                port: port,
+                chan: chan,
+                dropped_child: false
+            };
+            self.child = Some(child);
+        }
+
+        let child_link: &mut ChildLink = self.child.get_mut_ref();
+        let shared_state: *mut SharedState = &mut *child_link.shared;
+
+        child_link.shared.count.fetch_add(1, SeqCst);
+
+        let child = ~JoinLatch {
+            parent: Some(ParentLink {
+                shared: shared_state,
+                chan: child_link.chan.clone()
+            }),
+            child: None,
+            closed: false
+        };
+        rtdebug!("NEW child latch %x", child.id());
+        return child;
+    }
+
+    pub fn release(~self, local_success: bool) {
+        // XXX: This should not block, but there's a bug in the below
+        // code that I can't figure out.
+        self.wait(local_success);
+    }
+
+    // XXX: Should not require ~self
+    fn release_broken(~self, local_success: bool) {
+        rtassert!(!self.closed);
+
+        rtdebug!("releasing %x", self.id());
+
+        let id = self.id();
+        let _ = id; // XXX: `id` is only used in debug statements so appears unused
+        let mut this = self;
+        let mut child_success = true;
+        let mut children_done = false;
+
+        if this.child.is_some() {
+            rtdebug!("releasing children");
+            let child_link: &mut ChildLink = this.child.get_mut_ref();
+            let shared: &mut SharedState = &mut *child_link.shared;
+
+            if !child_link.dropped_child {
+                let last_count = shared.count.fetch_sub(1, SeqCst);
+                rtdebug!("child count before sub %u %x", last_count, id);
+                if last_count == 1 {
+                    assert!(child_link.chan.try_send(ChildrenTerminated));
+                }
+                child_link.dropped_child = true;
+            }
+
+            // Wait for messages from children
+            let mut tombstones = ~[];
+            loop {
+                if child_link.port.peek() {
+                    match child_link.port.recv() {
+                        Tombstone(t) => {
+                            tombstones.push(t);
+                        },
+                        ChildrenTerminated => {
+                            children_done = true;
+                            break;
+                        }
+                    }
+                } else {
+                    break
+                }
+            }
+
+            rtdebug!("releasing %u tombstones %x", tombstones.len(), id);
+
+            // Try to release the tombstones. Those that still have
+            // outstanding will be re-enqueued.  When this task's
+            // parents release their latch we'll end up back here
+            // trying them again.
+            while !tombstones.is_empty() {
+                tombstones.pop().release(true);
+            }
+
+            if children_done {
+                let count = shared.count.load(SeqCst);
+                assert!(count == 0);
+                // self_count is the acquire-read barrier
+                child_success = shared.success;
+            }
+        } else {
+            children_done = true;
+        }
+
+        let total_success = local_success && child_success;
+
+        rtassert!(this.parent.is_some());
+
+        unsafe {
+            {
+                let parent_link: &mut ParentLink = this.parent.get_mut_ref();
+                let shared: *mut SharedState = parent_link.shared;
+
+                if !total_success {
+                    // parent_count is the write-wait barrier
+                    (*shared).success = false;
+                }
+            }
+
+            if children_done {
+                rtdebug!("children done");
+                do Local::borrow::<Scheduler, ()> |sched| {
+                    sched.metrics.release_tombstone += 1;
+                }
+                {
+                    rtdebug!("RELEASING parent %x", id);
+                    let parent_link: &mut ParentLink = this.parent.get_mut_ref();
+                    let shared: *mut SharedState = parent_link.shared;
+                    let last_count = (*shared).count.fetch_sub(1, SeqCst);
+                    rtdebug!("count before parent sub %u %x", last_count, id);
+                    if last_count == 1 {
+                        assert!(parent_link.chan.try_send(ChildrenTerminated));
+                    }
+                }
+                this.closed = true;
+                util::ignore(this);
+            } else {
+                rtdebug!("children not done");
+                rtdebug!("TOMBSTONING %x", id);
+                do Local::borrow::<Scheduler, ()> |sched| {
+                    sched.metrics.release_no_tombstone += 1;
+                }
+                let chan = {
+                    let parent_link: &mut ParentLink = this.parent.get_mut_ref();
+                    parent_link.chan.clone()
+                };
+                assert!(chan.try_send(Tombstone(this)));
+            }
+        }
+    }
+
+    // XXX: Should not require ~self
+    pub fn wait(~self, local_success: bool) -> bool {
+        rtassert!(!self.closed);
+
+        rtdebug!("WAITING %x", self.id());
+
+        let mut this = self;
+        let mut child_success = true;
+
+        if this.child.is_some() {
+            rtdebug!("waiting for children");
+            let child_link: &mut ChildLink = this.child.get_mut_ref();
+            let shared: &mut SharedState = &mut *child_link.shared;
+
+            if !child_link.dropped_child {
+                let last_count = shared.count.fetch_sub(1, SeqCst);
+                rtdebug!("child count before sub %u", last_count);
+                if last_count == 1 {
+                    assert!(child_link.chan.try_send(ChildrenTerminated));
+                }
+                child_link.dropped_child = true;
+            }
+
+            // Wait for messages from children
+            loop {
+                match child_link.port.recv() {
+                    Tombstone(t) => {
+                        t.wait(true);
+                    }
+                    ChildrenTerminated => break
+                }
+            }
+
+            let count = shared.count.load(SeqCst);
+            if count != 0 { ::io::println(fmt!("%u", count)); }
+            assert!(count == 0);
+            // self_count is the acquire-read barrier
+            child_success = shared.success;
+        }
+
+        let total_success = local_success && child_success;
+
+        if this.parent.is_some() {
+            rtdebug!("releasing parent");
+            unsafe {
+                let parent_link: &mut ParentLink = this.parent.get_mut_ref();
+                let shared: *mut SharedState = parent_link.shared;
+
+                if !total_success {
+                    // parent_count is the write-wait barrier
+                    (*shared).success = false;
+                }
+
+                let last_count = (*shared).count.fetch_sub(1, SeqCst);
+                rtdebug!("count before parent sub %u", last_count);
+                if last_count == 1 {
+                    assert!(parent_link.chan.try_send(ChildrenTerminated));
+                }
+            }
+        }
+
+        this.closed = true;
+        util::ignore(this);
+
+        return total_success;
+    }
+}
+
+impl Drop for JoinLatch {
+    fn finalize(&self) {
+        rtdebug!("DESTROYING %x", self.id());
+        rtassert!(self.closed);
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use cell::Cell;
+    use container::Container;
+    use iter::Times;
+    use old_iter::BaseIter;
+    use rt::test::*;
+    use rand;
+    use rand::RngUtil;
+    use vec::{CopyableVector, ImmutableVector};
+
+    #[test]
+    fn success_immediately() {
+        do run_in_newsched_task {
+            let mut latch = JoinLatch::new_root();
+
+            let child_latch = latch.new_child();
+            let child_latch = Cell::new(child_latch);
+            do spawntask_immediately {
+                let child_latch = child_latch.take();
+                assert!(child_latch.wait(true));
+            }
+
+            assert!(latch.wait(true));
+        }
+    }
+
+    #[test]
+    fn success_later() {
+        do run_in_newsched_task {
+            let mut latch = JoinLatch::new_root();
+
+            let child_latch = latch.new_child();
+            let child_latch = Cell::new(child_latch);
+            do spawntask_later {
+                let child_latch = child_latch.take();
+                assert!(child_latch.wait(true));
+            }
+
+            assert!(latch.wait(true));
+        }
+    }
+
+    #[test]
+    fn mt_success() {
+        do run_in_mt_newsched_task {
+            let mut latch = JoinLatch::new_root();
+
+            for 10.times {
+                let child_latch = latch.new_child();
+                let child_latch = Cell::new(child_latch);
+                do spawntask_random {
+                    let child_latch = child_latch.take();
+                    assert!(child_latch.wait(true));
+                }
+            }
+
+            assert!(latch.wait(true));
+        }
+    }
+
+    #[test]
+    fn mt_failure() {
+        do run_in_mt_newsched_task {
+            let mut latch = JoinLatch::new_root();
+
+            let spawn = |status| {
+                let child_latch = latch.new_child();
+                let child_latch = Cell::new(child_latch);
+                do spawntask_random {
+                    let child_latch = child_latch.take();
+                    child_latch.wait(status);
+                }
+            };
+
+            for 10.times { spawn(true) }
+            spawn(false);
+            for 10.times { spawn(true) }
+
+            assert!(!latch.wait(true));
+        }
+    }
+
+    #[test]
+    fn mt_multi_level_success() {
+        do run_in_mt_newsched_task {
+            let mut latch = JoinLatch::new_root();
+
+            fn child(latch: &mut JoinLatch, i: int) {
+                let child_latch = latch.new_child();
+                let child_latch = Cell::new(child_latch);
+                do spawntask_random {
+                    let mut child_latch = child_latch.take();
+                    if i != 0 {
+                        child(&mut *child_latch, i - 1);
+                        child_latch.wait(true);
+                    } else {
+                        child_latch.wait(true);
+                    }
+                }
+            }
+
+            child(&mut *latch, 10);
+
+            assert!(latch.wait(true));
+        }
+    }
+
+    #[test]
+    fn mt_multi_level_failure() {
+        do run_in_mt_newsched_task {
+            let mut latch = JoinLatch::new_root();
+
+            fn child(latch: &mut JoinLatch, i: int) {
+                let child_latch = latch.new_child();
+                let child_latch = Cell::new(child_latch);
+                do spawntask_random {
+                    let mut child_latch = child_latch.take();
+                    if i != 0 {
+                        child(&mut *child_latch, i - 1);
+                        child_latch.wait(false);
+                    } else {
+                        child_latch.wait(true);
+                    }
+                }
+            }
+
+            child(&mut *latch, 10);
+
+            assert!(!latch.wait(true));
+        }
+    }
+
+    #[test]
+    fn release_child() {
+        do run_in_newsched_task {
+            let mut latch = JoinLatch::new_root();
+            let child_latch = latch.new_child();
+            let child_latch = Cell::new(child_latch);
+
+            do spawntask_immediately {
+                let latch = child_latch.take();
+                latch.release(false);
+            }
+
+            assert!(!latch.wait(true));
+        }
+    }
+
+    #[test]
+    fn release_child_tombstone() {
+        do run_in_newsched_task {
+            let mut latch = JoinLatch::new_root();
+            let child_latch = latch.new_child();
+            let child_latch = Cell::new(child_latch);
+
+            do spawntask_immediately {
+                let mut latch = child_latch.take();
+                let child_latch = latch.new_child();
+                let child_latch = Cell::new(child_latch);
+                do spawntask_later {
+                    let latch = child_latch.take();
+                    latch.release(false);
+                }
+                latch.release(true);
+            }
+
+            assert!(!latch.wait(true));
+        }
+    }
+
+    #[test]
+    fn release_child_no_tombstone() {
+        do run_in_newsched_task {
+            let mut latch = JoinLatch::new_root();
+            let child_latch = latch.new_child();
+            let child_latch = Cell::new(child_latch);
+
+            do spawntask_later {
+                let mut latch = child_latch.take();
+                let child_latch = latch.new_child();
+                let child_latch = Cell::new(child_latch);
+                do spawntask_immediately {
+                    let latch = child_latch.take();
+                    latch.release(false);
+                }
+                latch.release(true);
+            }
+
+            assert!(!latch.wait(true));
+        }
+    }
+
+    #[test]
+    fn release_child_tombstone_stress() {
+        fn rand_orders() -> ~[bool] {
+            let mut v = ~[false,.. 5];
+            v[0] = true;
+            let mut rng = rand::rng();
+            return rng.shuffle(v);
+        }
+
+        fn split_orders(orders: &[bool]) -> (~[bool], ~[bool]) {
+            if orders.is_empty() {
+                return (~[], ~[]);
+            } else if orders.len() <= 2 {
+                return (orders.to_owned(), ~[]);
+            }
+            let mut rng = rand::rng();
+            let n = rng.gen_uint_range(1, orders.len());
+            let first = orders.slice(0, n).to_owned();
+            let last = orders.slice(n, orders.len()).to_owned();
+            assert!(first.len() + last.len() == orders.len());
+            return (first, last);
+        }
+
+        for stress_factor().times {
+            do run_in_newsched_task {
+                fn doit(latch: &mut JoinLatch, orders: ~[bool], depth: uint) {
+                    let (my_orders, remaining_orders) = split_orders(orders);
+                    rtdebug!("(my_orders, remaining): %?", (&my_orders, &remaining_orders));
+                    rtdebug!("depth: %u", depth);
+                    let mut remaining_orders = remaining_orders;
+                    let mut num = 0;
+                    for my_orders.each |&order| {
+                        let child_latch = latch.new_child();
+                        let child_latch = Cell::new(child_latch);
+                        let (child_orders, remaining) = split_orders(remaining_orders);
+                        rtdebug!("(child_orders, remaining): %?", (&child_orders, &remaining));
+                        remaining_orders = remaining;
+                        let child_orders = Cell::new(child_orders);
+                        let child_num = num;
+                        let _ = child_num; // XXX unused except in rtdebug!
+                        do spawntask_random {
+                            rtdebug!("depth %u num %u", depth, child_num);
+                            let mut child_latch = child_latch.take();
+                            let child_orders = child_orders.take();
+                            doit(&mut *child_latch, child_orders, depth + 1);
+                            child_latch.release(order);
+                        }
+
+                        num += 1;
+                    }
+                }
+
+                let mut latch = JoinLatch::new_root();
+                let orders = rand_orders();
+                rtdebug!("orders: %?", orders);
+
+                doit(&mut *latch, orders, 0);
+
+                assert!(!latch.wait(true));
+            }
+        }
+    }
+
+    #[test]
+    fn whateverman() {
+        struct Order {
+            immediate: bool,
+            succeed: bool,
+            orders: ~[Order]
+        }
+        fn next(latch: &mut JoinLatch, orders: ~[Order]) {
+            for orders.each |order| {
+                let suborders = copy order.orders;
+                let child_latch = Cell::new(latch.new_child());
+                let succeed = order.succeed;
+                if order.immediate {
+                    do spawntask_immediately {
+                        let mut child_latch = child_latch.take();
+                        next(&mut *child_latch, copy suborders);
+                        rtdebug!("immediate releasing");
+                        child_latch.release(succeed);
+                    }
+                } else {
+                    do spawntask_later {
+                        let mut child_latch = child_latch.take();
+                        next(&mut *child_latch, copy suborders);
+                        rtdebug!("later releasing");
+                        child_latch.release(succeed);
+                    }
+                }
+            }
+        }
+
+        do run_in_newsched_task {
+            let mut latch = JoinLatch::new_root();
+            let orders = ~[ Order { // 0 0
+                immediate: true,
+                succeed: true,
+                orders: ~[ Order { // 1 0
+                    immediate: true,
+                    succeed: false,
+                    orders: ~[ Order { // 2 0
+                        immediate: false,
+                        succeed: false,
+                        orders: ~[ Order { // 3 0
+                            immediate: true,
+                            succeed: false,
+                            orders: ~[]
+                        }, Order { // 3 1
+                            immediate: false,
+                            succeed: false,
+                            orders: ~[]
+                        }]
+                    }]
+                }]
+            }];
+
+            next(&mut *latch, orders);
+            assert!(!latch.wait(true));
+        }
+    }
+}
diff --git a/src/libstd/rt/local.rs b/src/libstd/rt/local.rs
index e6988c53888..6e0fbda5ec9 100644
--- a/src/libstd/rt/local.rs
+++ b/src/libstd/rt/local.rs
@@ -18,7 +18,7 @@ pub trait Local {
     fn put(value: ~Self);
     fn take() -> ~Self;
     fn exists() -> bool;
-    fn borrow(f: &fn(&mut Self));
+    fn borrow<T>(f: &fn(&mut Self) -> T) -> T;
     unsafe fn unsafe_borrow() -> *mut Self;
     unsafe fn try_unsafe_borrow() -> Option<*mut Self>;
 }
@@ -27,7 +27,20 @@ impl Local for Scheduler {
     fn put(value: ~Scheduler) { unsafe { local_ptr::put(value) }}
     fn take() -> ~Scheduler { unsafe { local_ptr::take() } }
     fn exists() -> bool { local_ptr::exists() }
-    fn borrow(f: &fn(&mut Scheduler)) { unsafe { local_ptr::borrow(f) } }
+    fn borrow<T>(f: &fn(&mut Scheduler) -> T) -> T {
+        let mut res: Option<T> = None;
+        let res_ptr: *mut Option<T> = &mut res;
+        unsafe {
+            do local_ptr::borrow |sched| {
+                let result = f(sched);
+                *res_ptr = Some(result);
+            }
+        }
+        match res {
+            Some(r) => { r }
+            None => abort!("function failed!")
+        }
+    }
     unsafe fn unsafe_borrow() -> *mut Scheduler { local_ptr::unsafe_borrow() }
     unsafe fn try_unsafe_borrow() -> Option<*mut Scheduler> { abort!("unimpl") }
 }
@@ -36,8 +49,8 @@ impl Local for Task {
     fn put(_value: ~Task) { abort!("unimpl") }
     fn take() -> ~Task { abort!("unimpl") }
     fn exists() -> bool { abort!("unimpl") }
-    fn borrow(f: &fn(&mut Task)) {
-        do Local::borrow::<Scheduler> |sched| {
+    fn borrow<T>(f: &fn(&mut Task) -> T) -> T {
+        do Local::borrow::<Scheduler, T> |sched| {
             match sched.current_task {
                 Some(~ref mut task) => {
                     f(&mut *task.task)
@@ -74,7 +87,7 @@ impl Local for IoFactoryObject {
     fn put(_value: ~IoFactoryObject) { abort!("unimpl") }
     fn take() -> ~IoFactoryObject { abort!("unimpl") }
     fn exists() -> bool { abort!("unimpl") }
-    fn borrow(_f: &fn(&mut IoFactoryObject)) { abort!("unimpl") }
+    fn borrow<T>(_f: &fn(&mut IoFactoryObject) -> T) -> T { abort!("unimpl") }
     unsafe fn unsafe_borrow() -> *mut IoFactoryObject {
         let sched = Local::unsafe_borrow::<Scheduler>();
         let io: *mut IoFactoryObject = (*sched).event_loop.io().unwrap();
@@ -115,4 +128,16 @@ mod test {
         }
         let _scheduler: ~Scheduler = Local::take();
     }
+
+    #[test]
+    fn borrow_with_return() {
+        let scheduler = ~new_test_uv_sched();
+        Local::put(scheduler);
+        let res = do Local::borrow::<Scheduler,bool> |_sched| {
+            true
+        };
+        assert!(res)
+        let _scheduler: ~Scheduler = Local::take();
+    }
+
 }
diff --git a/src/libstd/rt/local_ptr.rs b/src/libstd/rt/local_ptr.rs
index 80d797e8c65..0db903f81ee 100644
--- a/src/libstd/rt/local_ptr.rs
+++ b/src/libstd/rt/local_ptr.rs
@@ -79,7 +79,7 @@ pub unsafe fn borrow<T>(f: &fn(&mut T)) {
 
     // XXX: Need a different abstraction from 'finally' here to avoid unsafety
     let unsafe_ptr = cast::transmute_mut_region(&mut *value);
-    let value_cell = Cell(value);
+    let value_cell = Cell::new(value);
 
     do (|| {
         f(unsafe_ptr);
diff --git a/src/libstd/rt/message_queue.rs b/src/libstd/rt/message_queue.rs
index 21711bbe84c..734be808797 100644
--- a/src/libstd/rt/message_queue.rs
+++ b/src/libstd/rt/message_queue.rs
@@ -32,16 +32,20 @@ impl<T: Owned> MessageQueue<T> {
     }
 
     pub fn push(&mut self, value: T) {
-        let value = Cell(value);
-        self.queue.with(|q| q.push(value.take()) );
+        unsafe {
+            let value = Cell::new(value);
+            self.queue.with(|q| q.push(value.take()) );
+        }
     }
 
     pub fn pop(&mut self) -> Option<T> {
-        do self.queue.with |q| {
-            if !q.is_empty() {
-                Some(q.shift())
-            } else {
-                None
+        unsafe {
+            do self.queue.with |q| {
+                if !q.is_empty() {
+                    Some(q.shift())
+                } else {
+                    None
+                }
             }
         }
     }
diff --git a/src/libstd/rt/metrics.rs b/src/libstd/rt/metrics.rs
index 70e347fdfb6..b0c0fa5d708 100644
--- a/src/libstd/rt/metrics.rs
+++ b/src/libstd/rt/metrics.rs
@@ -34,7 +34,11 @@ pub struct SchedMetrics {
     // Message receives that do not block the receiver
     rendezvous_recvs: uint,
     // Message receives that block the receiver
-    non_rendezvous_recvs: uint
+    non_rendezvous_recvs: uint,
+    // JoinLatch releases that create tombstones
+    release_tombstone: uint,
+    // JoinLatch releases that do not create tombstones
+    release_no_tombstone: uint,
 }
 
 impl SchedMetrics {
@@ -51,7 +55,9 @@ impl SchedMetrics {
             rendezvous_sends: 0,
             non_rendezvous_sends: 0,
             rendezvous_recvs: 0,
-            non_rendezvous_recvs: 0
+            non_rendezvous_recvs: 0,
+            release_tombstone: 0,
+            release_no_tombstone: 0
         }
     }
 }
@@ -70,6 +76,8 @@ impl ToStr for SchedMetrics {
               non_rendezvous_sends: %u\n\
               rendezvous_recvs: %u\n\
               non_rendezvous_recvs: %u\n\
+              release_tombstone: %u\n\
+              release_no_tombstone: %u\n\
               ",
              self.turns,
              self.messages_received,
@@ -82,7 +90,9 @@ impl ToStr for SchedMetrics {
              self.rendezvous_sends,
              self.non_rendezvous_sends,
              self.rendezvous_recvs,
-             self.non_rendezvous_recvs
+             self.non_rendezvous_recvs,
+             self.release_tombstone,
+             self.release_no_tombstone
         )
     }
 }
\ No newline at end of file
diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs
index caf3e15e535..1724361cabc 100644
--- a/src/libstd/rt/mod.rs
+++ b/src/libstd/rt/mod.rs
@@ -59,7 +59,8 @@ Several modules in `core` are clients of `rt`:
 #[deny(unused_mut)];
 #[deny(unused_variable)];
 
-use ptr::Ptr;
+use cell::Cell;
+use ptr::RawPtr;
 
 /// The global (exchange) heap.
 pub mod global_heap;
@@ -133,6 +134,9 @@ pub mod local_ptr;
 /// Bindings to pthread/windows thread-local storage.
 pub mod thread_local_storage;
 
+/// A concurrent data structure with which parent tasks wait on child tasks.
+pub mod join_latch;
+
 pub mod metrics;
 
 
@@ -164,7 +168,7 @@ pub fn start(_argc: int, _argv: **u8, crate_map: *u8, main: ~fn()) -> int {
     let sleepers = SleeperList::new();
     let mut sched = ~Scheduler::new(loop_, work_queue, sleepers);
     sched.no_sleep = true;
-    let main_task = ~Coroutine::new(&mut sched.stack_pool, main);
+    let main_task = ~Coroutine::new_root(&mut sched.stack_pool, main);
 
     sched.enqueue_task(main_task);
     sched.run();
@@ -207,8 +211,8 @@ pub fn context() -> RuntimeContext {
         return OldTaskContext;
     } else {
         if Local::exists::<Scheduler>() {
-            let context = ::cell::empty_cell();
-            do Local::borrow::<Scheduler> |sched| {
+            let context = Cell::new_empty();
+            do Local::borrow::<Scheduler, ()> |sched| {
                 if sched.in_task_context() {
                     context.put_back(TaskContext);
                 } else {
@@ -238,7 +242,7 @@ fn test_context() {
     do run_in_bare_thread {
         assert_eq!(context(), GlobalContext);
         let mut sched = ~new_test_uv_sched();
-        let task = ~do Coroutine::new(&mut sched.stack_pool) {
+        let task = ~do Coroutine::new_root(&mut sched.stack_pool) {
             assert_eq!(context(), TaskContext);
             let sched = Local::take::<Scheduler>();
             do sched.deschedule_running_task_and_then() |sched, task| {
diff --git a/src/libstd/rt/rc.rs b/src/libstd/rt/rc.rs
index 1c0c8c14fdf..2977d081508 100644
--- a/src/libstd/rt/rc.rs
+++ b/src/libstd/rt/rc.rs
@@ -78,7 +78,7 @@ impl<T> Drop for RC<T> {
         assert!(self.refcount() > 0);
 
         unsafe {
-            // XXX: Mutable finalizer
+            // FIXME(#4330) Need self by value to get mutability.
             let this: &mut RC<T> = cast::transmute_mut(self);
 
             match *this.get_mut_state() {
diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index df231f6d88a..be57247d514 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -26,6 +26,11 @@ use rt::local::Local;
 use rt::rtio::RemoteCallback;
 use rt::metrics::SchedMetrics;
 
+//use to_str::ToStr;
+
+/// To allow for using pointers as scheduler ids
+use borrow::{to_uint};
+
 /// The Scheduler is responsible for coordinating execution of Coroutines
 /// on a single thread. When the scheduler is running it is owned by
 /// thread local storage and the running task is owned by the
@@ -65,12 +70,15 @@ pub struct Scheduler {
     /// An action performed after a context switch on behalf of the
     /// code running before the context switch
     priv cleanup_job: Option<CleanupJob>,
-    metrics: SchedMetrics
+    metrics: SchedMetrics,
+    /// Should this scheduler run any task, or only pinned tasks?
+    run_anything: bool
 }
 
 pub struct SchedHandle {
     priv remote: ~RemoteCallbackObject,
-    priv queue: MessageQueue<SchedMessage>
+    priv queue: MessageQueue<SchedMessage>,
+    sched_id: uint
 }
 
 pub struct Coroutine {
@@ -81,12 +89,20 @@ pub struct Coroutine {
     /// the task is dead
     priv saved_context: Context,
     /// The heap, GC, unwinding, local storage, logging
-    task: ~Task
+    task: ~Task,
+}
+
+// A scheduler home is either a handle to the home scheduler, or an
+// explicit "AnySched".
+pub enum SchedHome {
+    AnySched,
+    Sched(SchedHandle)
 }
 
 pub enum SchedMessage {
     Wake,
-    Shutdown
+    Shutdown,
+    PinnedTask(~Coroutine)
 }
 
 enum CleanupJob {
@@ -94,13 +110,24 @@ enum CleanupJob {
     GiveTask(~Coroutine, UnsafeTaskReceiver)
 }
 
-pub impl Scheduler {
+impl Scheduler {
+    pub fn in_task_context(&self) -> bool { self.current_task.is_some() }
+
+    pub fn sched_id(&self) -> uint { to_uint(self) }
+
+    pub fn new(event_loop: ~EventLoopObject,
+               work_queue: WorkQueue<~Coroutine>,
+               sleeper_list: SleeperList)
+        -> Scheduler {
+
+        Scheduler::new_special(event_loop, work_queue, sleeper_list, true)
 
-    fn in_task_context(&self) -> bool { self.current_task.is_some() }
+    }
 
-    fn new(event_loop: ~EventLoopObject,
-           work_queue: WorkQueue<~Coroutine>,
-           sleeper_list: SleeperList)
+    pub fn new_special(event_loop: ~EventLoopObject,
+                       work_queue: WorkQueue<~Coroutine>,
+                       sleeper_list: SleeperList,
+                       run_anything: bool)
         -> Scheduler {
 
         // Lazily initialize the runtime TLS key
@@ -117,7 +144,8 @@ pub impl Scheduler {
             saved_context: Context::empty(),
             current_task: None,
             cleanup_job: None,
-            metrics: SchedMetrics::new()
+            metrics: SchedMetrics::new(),
+            run_anything: run_anything
         }
     }
 
@@ -125,7 +153,7 @@ pub impl Scheduler {
     // the scheduler itself doesn't have to call event_loop.run.
     // That will be important for embedding the runtime into external
     // event loops.
-    fn run(~self) -> ~Scheduler {
+    pub fn run(~self) -> ~Scheduler {
         assert!(!self.in_task_context());
 
         let mut self_sched = self;
@@ -147,11 +175,15 @@ pub impl Scheduler {
             (*event_loop).run();
         }
 
+        rtdebug!("run taking sched");
         let sched = Local::take::<Scheduler>();
         // XXX: Reenable this once we're using a per-task queue. With a shared
         // queue this is not true
         //assert!(sched.work_queue.is_empty());
-        rtdebug!("scheduler metrics: %s\n", sched.metrics.to_str());
+        rtdebug!("scheduler metrics: %s\n", {
+            use to_str::ToStr;
+            sched.metrics.to_str()
+        });
         return sched;
     }
 
@@ -167,6 +199,7 @@ pub impl Scheduler {
         if sched.interpret_message_queue() {
             // We performed a scheduling action. There may be other work
             // to do yet, so let's try again later.
+            rtdebug!("run_sched_once, interpret_message_queue taking sched");
             let mut sched = Local::take::<Scheduler>();
             sched.metrics.messages_received += 1;
             sched.event_loop.callback(Scheduler::run_sched_once);
@@ -175,6 +208,7 @@ pub impl Scheduler {
         }
 
         // Now, look in the work queue for tasks to run
+        rtdebug!("run_sched_once taking");
         let sched = Local::take::<Scheduler>();
         if sched.resume_task_from_queue() {
             // We performed a scheduling action. There may be other work
@@ -204,38 +238,60 @@ pub impl Scheduler {
         Local::put(sched);
     }
 
-    fn make_handle(&mut self) -> SchedHandle {
+    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()
+            queue: self.message_queue.clone(),
+            sched_id: self.sched_id()
         };
     }
 
     /// Schedule a task to be executed later.
     ///
-    /// Pushes the task onto the work stealing queue and tells the event loop
-    /// to run it later. Always use this instead of pushing to the work queue
-    /// directly.
-    fn enqueue_task(&mut self, task: ~Coroutine) {
-        self.work_queue.push(task);
-        self.event_loop.callback(Scheduler::run_sched_once);
-
-        // We've made work available. Notify a sleeping scheduler.
-        // XXX: perf. Check for a sleeper without synchronizing memory.
-        // It's not critical that we always find it.
-        // XXX: perf. If there's a sleeper then we might as well just send
-        // it the task directly instead of pushing it to the
-        // queue. That is essentially the intent here and it is less
-        // work.
-        match self.sleeper_list.pop() {
+    /// Pushes the task onto the work stealing queue and tells the
+    /// event loop to run it later. Always use this instead of pushing
+    /// to the work queue directly.
+    pub fn enqueue_task(&mut self, task: ~Coroutine) {
+
+        // We don't want to queue tasks that belong on other threads,
+        // so we send them home at enqueue time.
+
+        // The borrow checker doesn't like our disassembly of the
+        // Coroutine struct and partial use and mutation of the
+        // fields. So completely disassemble here and stop using?
+
+        // XXX perf: I think we might be able to shuffle this code to
+        // only destruct when we need to.
+
+        rtdebug!("a task was queued on: %u", self.sched_id());
+
+        let this = self;
+
+        // We push the task onto our local queue clone.
+        this.work_queue.push(task);
+        this.event_loop.callback(Scheduler::run_sched_once);
+
+        // We've made work available. Notify a
+        // sleeping scheduler.
+
+        // XXX: perf. Check for a sleeper without
+        // synchronizing memory.  It's not critical
+        // that we always find it.
+
+        // XXX: perf. If there's a sleeper then we
+        // might as well just send it the task
+        // directly instead of pushing it to the
+        // queue. That is essentially the intent here
+        // and it is less work.
+        match this.sleeper_list.pop() {
             Some(handle) => {
                 let mut handle = handle;
                 handle.send(Wake)
             }
-            None => (/* pass */)
-        }
+            None => { (/* pass */) }
+        };
     }
 
     // * Scheduler-context operations
@@ -247,6 +303,15 @@ pub impl Scheduler {
 
         let mut this = self;
         match this.message_queue.pop() {
+            Some(PinnedTask(task)) => {
+                rtdebug!("recv BiasedTask message in sched: %u",
+                         this.sched_id());
+                let mut task = task;
+                task.task.home = Some(Sched(this.make_handle()));
+                this.resume_task_immediately(task);
+                return true;
+            }
+
             Some(Wake) => {
                 rtdebug!("recv Wake message");
                 this.sleepy = false;
@@ -256,8 +321,9 @@ pub impl Scheduler {
             Some(Shutdown) => {
                 rtdebug!("recv Shutdown message");
                 if this.sleepy {
-                    // There may be an outstanding handle on the sleeper list.
-                    // Pop them all to make sure that's not the case.
+                    // There may be an outstanding handle on the
+                    // sleeper list.  Pop them all to make sure that's
+                    // not the case.
                     loop {
                         match this.sleeper_list.pop() {
                             Some(handle) => {
@@ -268,8 +334,8 @@ pub impl Scheduler {
                         }
                     }
                 }
-                // No more sleeping. After there are no outstanding event loop
-                // references we will shut down.
+                // No more sleeping. After there are no outstanding
+                // event loop references we will shut down.
                 this.no_sleep = true;
                 this.sleepy = false;
                 Local::put(this);
@@ -282,23 +348,93 @@ pub impl Scheduler {
         }
     }
 
+    /// Given an input Coroutine sends it back to its home scheduler.
+    fn send_task_home(task: ~Coroutine) {
+        let mut task = task;
+        let mut home = task.task.home.swap_unwrap();
+        match home {
+            Sched(ref mut home_handle) => {
+                home_handle.send(PinnedTask(task));
+            }
+            AnySched => {
+                abort!("error: cannot send anysched task home");
+            }
+        }
+    }
+
+    // Resume a task from the queue - but also take into account that
+    // it might not belong here.
     fn resume_task_from_queue(~self) -> bool {
         assert!(!self.in_task_context());
 
         rtdebug!("looking in work queue for task to schedule");
-
         let mut this = self;
+
+        // The borrow checker imposes the possibly absurd requirement
+        // that we split this into two match expressions. This is due
+        // to the inspection of the internal bits of task, as that
+        // can't be in scope when we act on task.
         match this.work_queue.pop() {
             Some(task) => {
-                rtdebug!("resuming task from work queue");
-                this.resume_task_immediately(task);
-                return true;
+                let action_id = {
+                    let home = &task.task.home;
+                    match home {
+                        &Some(Sched(ref home_handle))
+                        if home_handle.sched_id != this.sched_id() => {
+                            0
+                        }
+                        &Some(AnySched) if this.run_anything => {
+                            1
+                        }
+                        &Some(AnySched) => {
+                            2
+                        }
+                        &Some(Sched(_)) => {
+                            3
+                        }
+                        &None => {
+                            4
+                        }
+                    }
+                };
+
+                match action_id {
+                    0 => {
+                        rtdebug!("sending task home");
+                        Scheduler::send_task_home(task);
+                        Local::put(this);
+                        return false;
+                    }
+                    1 => {
+                        rtdebug!("resuming now");
+                        this.resume_task_immediately(task);
+                        return true;
+                    }
+                    2 => {
+                        rtdebug!("re-queueing")
+                        this.enqueue_task(task);
+                        Local::put(this);
+                        return false;
+                    }
+                    3 => {
+                        rtdebug!("resuming now");
+                        this.resume_task_immediately(task);
+                        return true;
+                    }
+                    4 => {
+                        abort!("task home was None!");
+                    }
+                    _ => {
+                        abort!("literally, you should not be here");
+                    }
+                }
             }
+
             None => {
-                rtdebug!("no tasks in queue");
-                Local::put(this);
-                return false;
-            }
+               rtdebug!("no tasks in queue");
+               Local::put(this);
+               return false;
+           }
         }
     }
 
@@ -306,40 +442,51 @@ pub impl Scheduler {
 
     /// Called by a running task to end execution, after which it will
     /// be recycled by the scheduler for reuse in a new task.
-    fn terminate_current_task(~self) {
+    pub fn terminate_current_task(~self) {
         assert!(self.in_task_context());
 
         rtdebug!("ending running task");
 
         do self.deschedule_running_task_and_then |sched, dead_task| {
-            let dead_task = Cell(dead_task);
+            let dead_task = Cell::new(dead_task);
             dead_task.take().recycle(&mut sched.stack_pool);
         }
 
         abort!("control reached end of task");
     }
 
-    fn schedule_new_task(~self, task: ~Coroutine) {
+    pub fn schedule_task(~self, task: ~Coroutine) {
         assert!(self.in_task_context());
 
-        do self.switch_running_tasks_and_then(task) |sched, last_task| {
-            let last_task = Cell(last_task);
-            sched.enqueue_task(last_task.take());
-        }
-    }
+        // is the task home?
+        let is_home = task.is_home_no_tls(&self);
 
-    fn schedule_task(~self, task: ~Coroutine) {
-        assert!(self.in_task_context());
+        // does the task have a home?
+        let homed = task.homed();
+
+        let mut this = self;
 
-        do self.switch_running_tasks_and_then(task) |sched, last_task| {
-            let last_task = Cell(last_task);
-            sched.enqueue_task(last_task.take());
+        if is_home || (!homed && this.run_anything) {
+            // here we know we are home, execute now OR we know we
+            // aren't homed, and that this sched doesn't care
+            do this.switch_running_tasks_and_then(task) |sched, last_task| {
+                let last_task = Cell::new(last_task);
+                sched.enqueue_task(last_task.take());
+            }
+        } else if !homed && !this.run_anything {
+            // the task isn't homed, but it can't be run here
+            this.enqueue_task(task);
+            Local::put(this);
+        } else {
+            // task isn't home, so don't run it here, send it home
+            Scheduler::send_task_home(task);
+            Local::put(this);
         }
     }
 
     // Core scheduling ops
 
-    fn resume_task_immediately(~self, task: ~Coroutine) {
+    pub fn resume_task_immediately(~self, task: ~Coroutine) {
         let mut this = self;
         assert!(!this.in_task_context());
 
@@ -382,7 +529,7 @@ pub impl Scheduler {
     /// This passes a Scheduler pointer to the fn after the context switch
     /// in order to prevent that fn from performing further scheduling operations.
     /// Doing further scheduling could easily result in infinite recursion.
-    fn deschedule_running_task_and_then(~self, f: &fn(&mut Scheduler, ~Coroutine)) {
+    pub fn deschedule_running_task_and_then(~self, f: &fn(&mut Scheduler, ~Coroutine)) {
         let mut this = self;
         assert!(this.in_task_context());
 
@@ -414,8 +561,8 @@ 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(&mut Scheduler, ~Coroutine)) {
+    pub fn switch_running_tasks_and_then(~self, next_task: ~Coroutine,
+                                         f: &fn(&mut Scheduler, ~Coroutine)) {
         let mut this = self;
         assert!(this.in_task_context());
 
@@ -450,12 +597,12 @@ pub impl Scheduler {
 
     // * Other stuff
 
-    fn enqueue_cleanup_job(&mut self, job: CleanupJob) {
+    pub fn enqueue_cleanup_job(&mut self, job: CleanupJob) {
         assert!(self.cleanup_job.is_none());
         self.cleanup_job = Some(job);
     }
 
-    fn run_cleanup_job(&mut self) {
+    pub fn run_cleanup_job(&mut self) {
         rtdebug!("running cleanup job");
 
         assert!(self.cleanup_job.is_some());
@@ -476,9 +623,9 @@ pub impl Scheduler {
     /// callers should first arrange for that task to be located in the
     /// Scheduler's current_task slot and set up the
     /// post-context-switch cleanup job.
-    fn get_contexts<'a>(&'a mut self) -> (&'a mut Context,
-                                          Option<&'a mut Context>,
-                                          Option<&'a mut Context>) {
+    pub fn get_contexts<'a>(&'a mut self) -> (&'a mut Context,
+                                              Option<&'a mut Context>,
+                                              Option<&'a mut Context>) {
         let last_task = match self.cleanup_job {
             Some(GiveTask(~ref task, _)) => {
                 Some(task)
@@ -514,31 +661,112 @@ impl SchedHandle {
     }
 }
 
-pub impl Coroutine {
-    fn new(stack_pool: &mut StackPool, start: ~fn()) -> Coroutine {
-        Coroutine::with_task(stack_pool, ~Task::new(), start)
+impl Coroutine {
+
+    /// This function checks that a coroutine is running "home".
+    pub fn is_home(&self) -> bool {
+        rtdebug!("checking if coroutine is home");
+        do Local::borrow::<Scheduler,bool> |sched| {
+            match self.task.home {
+                Some(AnySched) => { false }
+                Some(Sched(SchedHandle { sched_id: ref id, _ })) => {
+                    *id == sched.sched_id()
+                }
+                None => { abort!("error: homeless task!"); }
+            }
+        }
     }
 
-    fn with_task(stack_pool: &mut StackPool,
-                  task: ~Task,
-                  start: ~fn()) -> Coroutine {
+    /// Without access to self, but with access to the "expected home
+    /// id", see if we are home.
+    fn is_home_using_id(id: uint) -> bool {
+        rtdebug!("checking if coroutine is home using id");
+        do Local::borrow::<Scheduler,bool> |sched| {
+            if sched.sched_id() == id {
+                true
+            } else {
+                false
+            }
+        }
+    }
 
-        static MIN_STACK_SIZE: uint = 10000000; // XXX: Too much stack
+    /// Check if this coroutine has a home
+    fn homed(&self) -> bool {
+        rtdebug!("checking if this coroutine has a home");
+        match self.task.home {
+            Some(AnySched) => { false }
+            Some(Sched(_)) => { true }
+            None => { abort!("error: homeless task!");
+                    }
+        }
+    }
+
+    /// A version of is_home that does not need to use TLS, it instead
+    /// takes local scheduler as a parameter.
+    fn is_home_no_tls(&self, sched: &~Scheduler) -> bool {
+        rtdebug!("checking if coroutine is home without tls");
+        match self.task.home {
+            Some(AnySched) => { true }
+            Some(Sched(SchedHandle { sched_id: ref id, _})) => {
+                *id == sched.sched_id()
+            }
+            None => { abort!("error: homeless task!"); }
+        }
+    }
+
+    /// Check TLS for the scheduler to see if we are on a special
+    /// scheduler.
+    pub fn on_special() -> bool {
+        rtdebug!("checking if coroutine is executing on special sched");
+        do Local::borrow::<Scheduler,bool>() |sched| {
+            !sched.run_anything
+        }
+    }
+
+    // Created new variants of "new" that takes a home scheduler
+    // parameter. The original with_task now calls with_task_homed
+    // using the AnySched paramter.
+
+    pub fn new_homed(stack_pool: &mut StackPool, home: SchedHome, start: ~fn()) -> Coroutine {
+        Coroutine::with_task_homed(stack_pool, ~Task::new_root(), start, home)
+    }
+
+    pub fn new_root(stack_pool: &mut StackPool, start: ~fn()) -> Coroutine {
+        Coroutine::with_task(stack_pool, ~Task::new_root(), start)
+    }
+
+    pub fn with_task_homed(stack_pool: &mut StackPool,
+                           task: ~Task,
+                           start: ~fn(),
+                           home: SchedHome) -> Coroutine {
+
+        static MIN_STACK_SIZE: uint = 1000000; // XXX: Too much stack
 
         let start = Coroutine::build_start_wrapper(start);
         let mut stack = stack_pool.take_segment(MIN_STACK_SIZE);
         // NB: Context holds a pointer to that ~fn
         let initial_context = Context::new(start, &mut stack);
-        return Coroutine {
+        let mut crt = Coroutine {
             current_stack_segment: stack,
             saved_context: initial_context,
-            task: task
+            task: task,
         };
+        crt.task.home = Some(home);
+        return crt;
     }
 
-    priv fn build_start_wrapper(start: ~fn()) -> ~fn() {
+    pub fn with_task(stack_pool: &mut StackPool,
+                 task: ~Task,
+                 start: ~fn()) -> Coroutine {
+        Coroutine::with_task_homed(stack_pool,
+                                   task,
+                                   start,
+                                   AnySched)
+    }
+
+    fn build_start_wrapper(start: ~fn()) -> ~fn() {
         // XXX: The old code didn't have this extra allocation
-        let start_cell = Cell(start);
+        let start_cell = Cell::new(start);
         let wrapper: ~fn() = || {
             // This is the first code to execute after the initial
             // context switch to the task. The previous context may
@@ -549,17 +777,20 @@ pub impl Coroutine {
 
                 let sched = Local::unsafe_borrow::<Scheduler>();
                 let task = (*sched).current_task.get_mut_ref();
-                // FIXME #6141: shouldn't neet to put `start()` in another closure
-                let start_cell = Cell(start_cell.take());
+                // FIXME #6141: shouldn't neet to put `start()` in
+                // another closure
+                let start_cell = Cell::new(start_cell.take());
                 do task.task.run {
-                    // N.B. Removing `start` from the start wrapper closure
-                    // by emptying a cell is critical for correctness. The ~Task
-                    // pointer, and in turn the closure used to initialize the first
-                    // call frame, is destroyed in scheduler context, not task context.
-                    // So any captured closures must not contain user-definable dtors
-                    // that expect to be in task context. By moving `start` out of
-                    // the closure, all the user code goes out of scope while
-                    // the task is still running.
+                    // N.B. Removing `start` from the start wrapper
+                    // closure by emptying a cell is critical for
+                    // correctness. The ~Task pointer, and in turn the
+                    // closure used to initialize the first call
+                    // frame, is destroyed in scheduler context, not
+                    // task context.  So any captured closures must
+                    // not contain user-definable dtors that expect to
+                    // be in task context. By moving `start` out of
+                    // the closure, all the user code goes out of
+                    // scope while the task is still running.
                     let start = start_cell.take();
                     start();
                 };
@@ -572,7 +803,7 @@ pub impl Coroutine {
     }
 
     /// Destroy the task and try to reuse its components
-    fn recycle(~self, stack_pool: &mut StackPool) {
+    pub fn recycle(~self, stack_pool: &mut StackPool) {
         match self {
             ~Coroutine {current_stack_segment, _} => {
                 stack_pool.give_segment(current_stack_segment);
@@ -597,12 +828,312 @@ impl ClosureConverter for UnsafeTaskReceiver {
 mod test {
     use int;
     use cell::Cell;
+    use iterator::IteratorUtil;
     use unstable::run_in_bare_thread;
     use task::spawn;
     use rt::local::Local;
     use rt::test::*;
     use super::*;
     use rt::thread::Thread;
+    use ptr::to_uint;
+    use vec::MutableVector;
+
+    // Confirm that a sched_id actually is the uint form of the
+    // pointer to the scheduler struct.
+
+    #[test]
+    fn simple_sched_id_test() {
+        do run_in_bare_thread {
+            let sched = ~new_test_uv_sched();
+            assert!(to_uint(sched) == sched.sched_id());
+        }
+    }
+
+    // Compare two scheduler ids that are different, this should never
+    // fail but may catch a mistake someday.
+
+    #[test]
+    fn compare_sched_id_test() {
+        do run_in_bare_thread {
+            let sched_one = ~new_test_uv_sched();
+            let sched_two = ~new_test_uv_sched();
+            assert!(sched_one.sched_id() != sched_two.sched_id());
+        }
+    }
+
+    // A simple test to check if a homed task run on a single
+    // scheduler ends up executing while home.
+
+    #[test]
+    fn test_home_sched() {
+        do run_in_bare_thread {
+            let mut task_ran = false;
+            let task_ran_ptr: *mut bool = &mut task_ran;
+            let mut sched = ~new_test_uv_sched();
+
+            let sched_handle = sched.make_handle();
+            let sched_id = sched.sched_id();
+
+            let task = ~do Coroutine::new_homed(&mut sched.stack_pool,
+                                                Sched(sched_handle)) {
+                unsafe { *task_ran_ptr = true };
+                let sched = Local::take::<Scheduler>();
+                assert!(sched.sched_id() == sched_id);
+                Local::put::<Scheduler>(sched);
+            };
+            sched.enqueue_task(task);
+            sched.run();
+            assert!(task_ran);
+        }
+    }
+
+    // A test for each state of schedule_task
+
+    #[test]
+    fn test_schedule_home_states() {
+
+        use rt::uv::uvio::UvEventLoop;
+        use rt::sched::Shutdown;
+        use rt::sleeper_list::SleeperList;
+        use rt::work_queue::WorkQueue;
+
+        do run_in_bare_thread {
+//            let nthreads = 2;
+
+            let sleepers = SleeperList::new();
+            let work_queue = WorkQueue::new();
+
+            // our normal scheduler
+            let mut normal_sched = ~Scheduler::new(
+                ~UvEventLoop::new(),
+                work_queue.clone(),
+                sleepers.clone());
+
+            let normal_handle = Cell::new(normal_sched.make_handle());
+
+            // our special scheduler
+            let mut special_sched = ~Scheduler::new_special(
+                ~UvEventLoop::new(),
+                work_queue.clone(),
+                sleepers.clone(),
+                true);
+
+            let special_handle = Cell::new(special_sched.make_handle());
+            let special_handle2 = Cell::new(special_sched.make_handle());
+            let special_id = special_sched.sched_id();
+            let t1_handle = special_sched.make_handle();
+            let t4_handle = special_sched.make_handle();
+
+            let t1f = ~do Coroutine::new_homed(&mut special_sched.stack_pool,
+                                            Sched(t1_handle)) {
+                let is_home = Coroutine::is_home_using_id(special_id);
+                rtdebug!("t1 should be home: %b", is_home);
+                assert!(is_home);
+            };
+            let t1f = Cell::new(t1f);
+
+            let t2f = ~do Coroutine::new_root(&mut normal_sched.stack_pool) {
+                let on_special = Coroutine::on_special();
+                rtdebug!("t2 should not be on special: %b", on_special);
+                assert!(!on_special);
+            };
+            let t2f = Cell::new(t2f);
+
+            let t3f = ~do Coroutine::new_root(&mut normal_sched.stack_pool) {
+                // not on special
+                let on_special = Coroutine::on_special();
+                rtdebug!("t3 should not be on special: %b", on_special);
+                assert!(!on_special);
+            };
+            let t3f = Cell::new(t3f);
+
+            let t4f = ~do Coroutine::new_homed(&mut special_sched.stack_pool,
+                                            Sched(t4_handle)) {
+                // is home
+                let home = Coroutine::is_home_using_id(special_id);
+                rtdebug!("t4 should be home: %b", home);
+                assert!(home);
+            };
+            let t4f = Cell::new(t4f);
+
+            // we have four tests, make them as closures
+            let t1: ~fn() = || {
+                // task is home on special
+                let task = t1f.take();
+                let sched = Local::take::<Scheduler>();
+                sched.schedule_task(task);
+            };
+            let t2: ~fn() = || {
+                // not homed, task doesn't care
+                let task = t2f.take();
+                let sched = Local::take::<Scheduler>();
+                sched.schedule_task(task);
+            };
+            let t3: ~fn() = || {
+                // task not homed, must leave
+                let task = t3f.take();
+                let sched = Local::take::<Scheduler>();
+                sched.schedule_task(task);
+            };
+            let t4: ~fn() = || {
+                // task not home, send home
+                let task = t4f.take();
+                let sched = Local::take::<Scheduler>();
+                sched.schedule_task(task);
+            };
+
+            let t1 = Cell::new(t1);
+            let t2 = Cell::new(t2);
+            let t3 = Cell::new(t3);
+            let t4 = Cell::new(t4);
+
+            // build a main task that runs our four tests
+            let main_task = ~do Coroutine::new_root(&mut normal_sched.stack_pool) {
+                // the two tasks that require a normal start location
+                t2.take()();
+                t4.take()();
+                normal_handle.take().send(Shutdown);
+                special_handle.take().send(Shutdown);
+            };
+
+            // task to run the two "special start" tests
+            let special_task = ~do Coroutine::new_homed(
+                &mut special_sched.stack_pool,
+                Sched(special_handle2.take())) {
+                t1.take()();
+                t3.take()();
+            };
+
+            // enqueue the main tasks
+            normal_sched.enqueue_task(special_task);
+            normal_sched.enqueue_task(main_task);
+
+            let nsched_cell = Cell::new(normal_sched);
+            let normal_thread = do Thread::start {
+                let sched = nsched_cell.take();
+                sched.run();
+            };
+
+            let ssched_cell = Cell::new(special_sched);
+            let special_thread = do Thread::start {
+                let sched = ssched_cell.take();
+                sched.run();
+            };
+
+            // wait for the end
+            let _thread1 = normal_thread;
+            let _thread2 = special_thread;
+
+        }
+    }
+
+    // The following test is a bit of a mess, but it trys to do
+    // something tricky so I'm not sure how to get around this in the
+    // short term.
+
+    // A number of schedulers are created, and then a task is created
+    // and assigned a home scheduler. It is then "started" on a
+    // different scheduler. The scheduler it is started on should
+    // observe that the task is not home, and send it home.
+
+    // This test is light in that it does very little.
+
+    #[test]
+    fn test_transfer_task_home() {
+
+        use rt::uv::uvio::UvEventLoop;
+        use rt::sched::Shutdown;
+        use rt::sleeper_list::SleeperList;
+        use rt::work_queue::WorkQueue;
+        use uint;
+        use container::Container;
+        use vec::OwnedVector;
+
+        do run_in_bare_thread {
+
+            static N: uint = 8;
+
+            let sleepers = SleeperList::new();
+            let work_queue = WorkQueue::new();
+
+            let mut handles = ~[];
+            let mut scheds = ~[];
+
+            for uint::range(0, N) |_| {
+                let loop_ = ~UvEventLoop::new();
+                let mut sched = ~Scheduler::new(loop_,
+                                                work_queue.clone(),
+                                                sleepers.clone());
+                let handle = sched.make_handle();
+                rtdebug!("sched id: %u", handle.sched_id);
+                handles.push(handle);
+                scheds.push(sched);
+            };
+
+            let handles = Cell::new(handles);
+
+            let home_handle = scheds[6].make_handle();
+            let home_id = home_handle.sched_id;
+            let home = Sched(home_handle);
+
+            let main_task = ~do Coroutine::new_homed(&mut scheds[1].stack_pool, home) {
+
+                // Here we check if the task is running on its home.
+                let sched = Local::take::<Scheduler>();
+                rtdebug!("run location scheduler id: %u, home: %u",
+                         sched.sched_id(),
+                         home_id);
+                assert!(sched.sched_id() == home_id);
+                Local::put::<Scheduler>(sched);
+
+                let mut handles = handles.take();
+                for handles.mut_iter().advance |handle| {
+                    handle.send(Shutdown);
+                }
+            };
+
+            scheds[0].enqueue_task(main_task);
+
+            let mut threads = ~[];
+
+            while !scheds.is_empty() {
+                let sched = scheds.pop();
+                let sched_cell = Cell::new(sched);
+                let thread = do Thread::start {
+                    let sched = sched_cell.take();
+                    sched.run();
+                };
+                threads.push(thread);
+            }
+
+            let _threads = threads;
+        }
+    }
+
+    // Do it a lot
+
+    #[test]
+    fn test_stress_schedule_task_states() {
+        let n = stress_factor() * 120;
+        for int::range(0,n as int) |_| {
+            test_schedule_home_states();
+        }
+    }
+
+    // The goal is that this is the high-stress test for making sure
+    // homing is working. It allocates RUST_RT_STRESS tasks that
+    // do nothing but assert that they are home at execution
+    // time. These tasks are queued to random schedulers, so sometimes
+    // they are home and sometimes not. It also runs RUST_RT_STRESS
+    // times.
+
+    #[test]
+    fn test_stress_homed_tasks() {
+        let n = stress_factor();
+        for int::range(0,n as int) |_| {
+            run_in_mt_newsched_task_random_homed();
+        }
+    }
 
     #[test]
     fn test_simple_scheduling() {
@@ -611,7 +1142,7 @@ mod test {
             let task_ran_ptr: *mut bool = &mut task_ran;
 
             let mut sched = ~new_test_uv_sched();
-            let task = ~do Coroutine::new(&mut sched.stack_pool) {
+            let task = ~do Coroutine::new_root(&mut sched.stack_pool) {
                 unsafe { *task_ran_ptr = true; }
             };
             sched.enqueue_task(task);
@@ -629,7 +1160,7 @@ mod test {
 
             let mut sched = ~new_test_uv_sched();
             for int::range(0, total) |_| {
-                let task = ~do Coroutine::new(&mut sched.stack_pool) {
+                let task = ~do Coroutine::new_root(&mut sched.stack_pool) {
                     unsafe { *task_count_ptr = *task_count_ptr + 1; }
                 };
                 sched.enqueue_task(task);
@@ -646,15 +1177,15 @@ mod test {
             let count_ptr: *mut int = &mut count;
 
             let mut sched = ~new_test_uv_sched();
-            let task1 = ~do Coroutine::new(&mut sched.stack_pool) {
+            let task1 = ~do Coroutine::new_root(&mut sched.stack_pool) {
                 unsafe { *count_ptr = *count_ptr + 1; }
                 let mut sched = Local::take::<Scheduler>();
-                let task2 = ~do Coroutine::new(&mut sched.stack_pool) {
+                let task2 = ~do Coroutine::new_root(&mut sched.stack_pool) {
                     unsafe { *count_ptr = *count_ptr + 1; }
                 };
                 // Context switch directly to the new task
                 do sched.switch_running_tasks_and_then(task2) |sched, task1| {
-                    let task1 = Cell(task1);
+                    let task1 = Cell::new(task1);
                     sched.enqueue_task(task1.take());
                 }
                 unsafe { *count_ptr = *count_ptr + 1; }
@@ -674,7 +1205,7 @@ mod test {
 
             let mut sched = ~new_test_uv_sched();
 
-            let start_task = ~do Coroutine::new(&mut sched.stack_pool) {
+            let start_task = ~do Coroutine::new_root(&mut sched.stack_pool) {
                 run_task(count_ptr);
             };
             sched.enqueue_task(start_task);
@@ -683,8 +1214,8 @@ mod test {
             assert_eq!(count, MAX);
 
             fn run_task(count_ptr: *mut int) {
-                do Local::borrow::<Scheduler> |sched| {
-                    let task = ~do Coroutine::new(&mut sched.stack_pool) {
+                do Local::borrow::<Scheduler, ()> |sched| {
+                    let task = ~do Coroutine::new_root(&mut sched.stack_pool) {
                         unsafe {
                             *count_ptr = *count_ptr + 1;
                             if *count_ptr != MAX {
@@ -702,11 +1233,11 @@ mod test {
     fn test_block_task() {
         do run_in_bare_thread {
             let mut sched = ~new_test_uv_sched();
-            let task = ~do Coroutine::new(&mut sched.stack_pool) {
+            let task = ~do Coroutine::new_root(&mut sched.stack_pool) {
                 let sched = Local::take::<Scheduler>();
                 assert!(sched.in_task_context());
                 do sched.deschedule_running_task_and_then() |sched, task| {
-                    let task = Cell(task);
+                    let task = Cell::new(task);
                     assert!(!sched.in_task_context());
                     sched.enqueue_task(task.take());
                 }
@@ -726,7 +1257,7 @@ mod test {
             do spawn {
                 let sched = Local::take::<Scheduler>();
                 do sched.deschedule_running_task_and_then |sched, task| {
-                    let task = Cell(task);
+                    let task = Cell::new(task);
                     do sched.event_loop.callback_ms(10) {
                         rtdebug!("in callback");
                         let mut sched = Local::take::<Scheduler>();
@@ -744,31 +1275,31 @@ mod test {
 
         do run_in_bare_thread {
             let (port, chan) = oneshot::<()>();
-            let port_cell = Cell(port);
-            let chan_cell = Cell(chan);
+            let port_cell = Cell::new(port);
+            let chan_cell = Cell::new(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) {
+            let handle1_cell = Cell::new(handle1);
+            let task1 = ~do Coroutine::new_root(&mut sched1.stack_pool) {
                 chan_cell.take().send(());
             };
             sched1.enqueue_task(task1);
 
             let mut sched2 = ~new_test_uv_sched();
-            let task2 = ~do Coroutine::new(&mut sched2.stack_pool) {
+            let task2 = ~do Coroutine::new_root(&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 sched1_cell = Cell::new(sched1);
             let _thread1 = do Thread::start {
                 let sched1 = sched1_cell.take();
                 sched1.run();
             };
 
-            let sched2_cell = Cell(sched2);
+            let sched2_cell = Cell::new(sched2);
             let _thread2 = do Thread::start {
                 let sched2 = sched2_cell.take();
                 sched2.run();
@@ -787,7 +1318,7 @@ mod test {
             let mut ports = ~[];
             for 10.times {
                 let (port, chan) = oneshot();
-                let chan_cell = Cell(chan);
+                let chan_cell = Cell::new(chan);
                 do spawntask_later {
                     chan_cell.take().send(());
                 }
@@ -859,8 +1390,8 @@ mod test {
     fn start_closure_dtor() {
         use ops::Drop;
 
-        // Regression test that the `start` task entrypoint can contain dtors
-        // that use task resources
+        // Regression test that the `start` task entrypoint can
+        // contain dtors that use task resources
         do run_in_newsched_task {
             struct S { field: () }
 
@@ -875,6 +1406,7 @@ mod test {
             do spawntask {
                 let _ss = &s;
             }
-        }        
+        }
     }
+
 }
diff --git a/src/libstd/rt/sleeper_list.rs b/src/libstd/rt/sleeper_list.rs
index e2873e78d80..3d6e9ef5635 100644
--- a/src/libstd/rt/sleeper_list.rs
+++ b/src/libstd/rt/sleeper_list.rs
@@ -31,16 +31,20 @@ impl SleeperList {
     }
 
     pub fn push(&mut self, handle: SchedHandle) {
-        let handle = Cell(handle);
-        self.stack.with(|s| s.push(handle.take()));
+        let handle = Cell::new(handle);
+        unsafe {
+            self.stack.with(|s| s.push(handle.take()));
+        }
     }
 
     pub fn pop(&mut self) -> Option<SchedHandle> {
-        do self.stack.with |s| {
-            if !s.is_empty() {
-                Some(s.pop())
-            } else {
-                None
+        unsafe {
+            do self.stack.with |s| {
+                if !s.is_empty() {
+                    Some(s.pop())
+                } else {
+                    None
+                }
             }
         }
     }
diff --git a/src/libstd/rt/stack.rs b/src/libstd/rt/stack.rs
index ec56e65931c..b0e87a62c8b 100644
--- a/src/libstd/rt/stack.rs
+++ b/src/libstd/rt/stack.rs
@@ -9,7 +9,7 @@
 // except according to those terms.
 
 use container::Container;
-use ptr::Ptr;
+use ptr::RawPtr;
 use vec;
 use ops::Drop;
 use libc::{c_uint, uintptr_t};
@@ -19,8 +19,8 @@ pub struct StackSegment {
     valgrind_id: c_uint
 }
 
-pub impl StackSegment {
-    fn new(size: uint) -> StackSegment {
+impl StackSegment {
+    pub fn new(size: uint) -> StackSegment {
         unsafe {
             // Crate a block of uninitialized values
             let mut stack = vec::with_capacity(size);
@@ -38,12 +38,12 @@ pub impl StackSegment {
     }
 
     /// Point to the low end of the allocated stack
-    fn start(&self) -> *uint {
-      vec::raw::to_ptr(self.buf) as *uint
+    pub fn start(&self) -> *uint {
+        vec::raw::to_ptr(self.buf) as *uint
     }
 
     /// Point one word beyond the high end of the allocated stack
-    fn end(&self) -> *uint {
+    pub fn end(&self) -> *uint {
         vec::raw::to_ptr(self.buf).offset(self.buf.len()) as *uint
     }
 }
diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs
index cf4967b12b3..e7f87906fe5 100644
--- a/src/libstd/rt/task.rs
+++ b/src/libstd/rt/task.rs
@@ -13,19 +13,27 @@
 //! local storage, and logging. Even a 'freestanding' Rust would likely want
 //! to implement this.
 
-use prelude::*;
-use libc::{c_void, uintptr_t};
+use borrow;
 use cast::transmute;
+use libc::{c_void, uintptr_t};
+use ptr;
+use prelude::*;
+use option::{Option, Some, None};
 use rt::local::Local;
-use super::local_heap::LocalHeap;
 use rt::logging::StdErrLogger;
+use super::local_heap::LocalHeap;
+use rt::sched::{SchedHome, AnySched};
+use rt::join_latch::JoinLatch;
 
 pub struct Task {
     heap: LocalHeap,
     gc: GarbageCollector,
     storage: LocalStorage,
     logger: StdErrLogger,
-    unwinder: Option<Unwinder>,
+    unwinder: Unwinder,
+    home: Option<SchedHome>,
+    join_latch: Option<~JoinLatch>,
+    on_exit: Option<~fn(bool)>,
     destroyed: bool
 }
 
@@ -37,49 +45,63 @@ pub struct Unwinder {
 }
 
 impl Task {
-    pub fn new() -> Task {
+    pub fn new_root() -> Task {
         Task {
             heap: LocalHeap::new(),
             gc: GarbageCollector,
             storage: LocalStorage(ptr::null(), None),
             logger: StdErrLogger,
-            unwinder: Some(Unwinder { unwinding: false }),
+            unwinder: Unwinder { unwinding: false },
+            home: Some(AnySched),
+            join_latch: Some(JoinLatch::new_root()),
+            on_exit: None,
             destroyed: false
         }
     }
 
-    pub fn without_unwinding() -> Task {
+    pub fn new_child(&mut self) -> Task {
         Task {
             heap: LocalHeap::new(),
             gc: GarbageCollector,
             storage: LocalStorage(ptr::null(), None),
             logger: StdErrLogger,
-            unwinder: None,
+            home: Some(AnySched),
+            unwinder: Unwinder { unwinding: false },
+            join_latch: Some(self.join_latch.get_mut_ref().new_child()),
+            on_exit: None,
             destroyed: false
         }
     }
 
+    pub fn give_home(&mut self, new_home: SchedHome) {
+        self.home = Some(new_home);
+    }
+
     pub fn run(&mut self, f: &fn()) {
         // This is just an assertion that `run` was called unsafely
         // and this instance of Task is still accessible.
-        do Local::borrow::<Task> |task| {
-            assert!(ptr::ref_eq(task, self));
+        do Local::borrow::<Task, ()> |task| {
+            assert!(borrow::ref_eq(task, self));
         }
 
-        match self.unwinder {
-            Some(ref mut unwinder) => {
-                // If there's an unwinder then set up the catch block
-                unwinder.try(f);
+        self.unwinder.try(f);
+        self.destroy();
+
+        // Wait for children. Possibly report the exit status.
+        let local_success = !self.unwinder.unwinding;
+        let join_latch = self.join_latch.swap_unwrap();
+        match self.on_exit {
+            Some(ref on_exit) => {
+                let success = join_latch.wait(local_success);
+                (*on_exit)(success);
             }
             None => {
-                // Otherwise, just run the body
-                f()
+                join_latch.release(local_success);
             }
         }
-        self.destroy();
     }
 
-    /// Must be called manually before finalization to clean up
+    /// must be called manually before finalization to clean up
     /// thread-local resources. Some of the routines here expect
     /// Task to be available recursively so this must be
     /// called unsafely, without removing Task from
@@ -87,8 +109,8 @@ impl Task {
     fn destroy(&mut self) {
         // This is just an assertion that `destroy` was called unsafely
         // and this instance of Task is still accessible.
-        do Local::borrow::<Task> |task| {
-            assert!(ptr::ref_eq(task, self));
+        do Local::borrow::<Task, ()> |task| {
+            assert!(borrow::ref_eq(task, self));
         }
         match self.storage {
             LocalStorage(ptr, Some(ref dtor)) => {
@@ -225,5 +247,14 @@ mod test {
             assert!(port.recv() == 10);
         }
     }
-}
 
+    #[test]
+    fn linked_failure() {
+        do run_in_newsched_task() {
+            let res = do spawntask_try {
+                spawntask_random(|| fail!());
+            };
+            assert!(res.is_err());
+        }
+    }
+}
diff --git a/src/libstd/rt/test.rs b/src/libstd/rt/test.rs
index c8df3a61203..6e4fb9b1d94 100644
--- a/src/libstd/rt/test.rs
+++ b/src/libstd/rt/test.rs
@@ -13,11 +13,12 @@ use option::{Some, None};
 use cell::Cell;
 use clone::Clone;
 use container::Container;
-use old_iter::MutableIter;
-use vec::OwnedVector;
+use iterator::IteratorUtil;
+use vec::{OwnedVector, MutableVector};
 use result::{Result, Ok, Err};
 use unstable::run_in_bare_thread;
 use super::io::net::ip::{IpAddr, Ipv4};
+use rt::comm::oneshot;
 use rt::task::Task;
 use rt::thread::Thread;
 use rt::local::Local;
@@ -43,12 +44,15 @@ pub fn run_in_newsched_task(f: ~fn()) {
     use super::sched::*;
     use unstable::run_in_bare_thread;
 
-    let f = Cell(f);
+    let f = Cell::new(f);
 
     do run_in_bare_thread {
         let mut sched = ~new_test_uv_sched();
+        let mut new_task = ~Task::new_root();
+        let on_exit: ~fn(bool) = |exit_status| rtassert!(exit_status);
+        new_task.on_exit = Some(on_exit);
         let task = ~Coroutine::with_task(&mut sched.stack_pool,
-                                         ~Task::without_unwinding(),
+                                         new_task,
                                          f.take());
         sched.enqueue_task(task);
         sched.run();
@@ -65,7 +69,7 @@ pub fn run_in_mt_newsched_task(f: ~fn()) {
     use rt::uv::uvio::UvEventLoop;
     use rt::sched::Shutdown;
 
-    let f_cell = Cell(f);
+    let f_cell = Cell::new(f);
 
     do run_in_bare_thread {
         let nthreads = match os::getenv("RUST_TEST_THREADS") {
@@ -88,37 +92,152 @@ pub fn run_in_mt_newsched_task(f: ~fn()) {
             let loop_ = ~UvEventLoop::new();
             let mut sched = ~Scheduler::new(loop_, work_queue.clone(), sleepers.clone());
             let handle = sched.make_handle();
+
             handles.push(handle);
             scheds.push(sched);
         }
 
-        let f_cell = Cell(f_cell.take());
-        let handles = Cell(handles);
-        let main_task = ~do Coroutine::new(&mut scheds[0].stack_pool) {
-            f_cell.take()();
+        let f_cell = Cell::new(f_cell.take());
+        let handles = Cell::new(handles);
+        let mut new_task = ~Task::new_root();
+        let on_exit: ~fn(bool) = |exit_status| {
 
             let mut handles = handles.take();
             // Tell schedulers to exit
-            for handles.each_mut |handle| {
+            for handles.mut_iter().advance |handle| {
                 handle.send(Shutdown);
             }
+
+            rtassert!(exit_status);
+        };
+        new_task.on_exit = Some(on_exit);
+        let main_task = ~Coroutine::with_task(&mut scheds[0].stack_pool,
+                                              new_task, f_cell.take());
+        scheds[0].enqueue_task(main_task);
+
+        let mut threads = ~[];
+
+        while !scheds.is_empty() {
+            let sched = scheds.pop();
+            let sched_cell = Cell::new(sched);
+            let thread = do Thread::start {
+                let sched = sched_cell.take();
+                sched.run();
+            };
+
+            threads.push(thread);
+        }
+
+        // Wait for schedulers
+        let _threads = threads;
+    }
+
+    extern {
+        fn rust_get_num_cpus() -> libc::uintptr_t;
+    }
+}
+
+// THIS IS AWFUL. Copy-pasted the above initialization function but
+// with a number of hacks to make it spawn tasks on a variety of
+// schedulers with a variety of homes using the new spawn.
+
+pub fn run_in_mt_newsched_task_random_homed() {
+    use libc;
+    use os;
+    use from_str::FromStr;
+    use rt::uv::uvio::UvEventLoop;
+    use rt::sched::Shutdown;
+
+    do run_in_bare_thread {
+        let nthreads = match os::getenv("RUST_TEST_THREADS") {
+            Some(nstr) => FromStr::from_str(nstr).get(),
+            None => unsafe {
+                // Using more threads than cores in test code to force
+                // the OS to preempt them frequently.  Assuming that
+                // this help stress test concurrent types.
+                rust_get_num_cpus() * 2
+            }
         };
 
+        let sleepers = SleeperList::new();
+        let work_queue = WorkQueue::new();
+
+        let mut handles = ~[];
+        let mut scheds = ~[];
+
+        // create a few special schedulers, those with even indicies
+        // will be pinned-only
+        for uint::range(0, nthreads) |i| {
+            let special = (i % 2) == 0;
+            let loop_ = ~UvEventLoop::new();
+            let mut sched = ~Scheduler::new_special(
+                loop_, work_queue.clone(), sleepers.clone(), special);
+            let handle = sched.make_handle();
+            handles.push(handle);
+            scheds.push(sched);
+        }
+
+        // Schedule a pile o tasks
+        let n = 5*stress_factor();
+        for uint::range(0,n) |_i| {
+                rtdebug!("creating task: %u", _i);
+                let hf: ~fn() = || { assert!(true) };
+                spawntask_homed(&mut scheds, hf);
+            }
+
+        // Now we want another pile o tasks that do not ever run on a
+        // special scheduler, because they are normal tasks. Because
+        // we can we put these in the "main" task.
+
+        let n = 5*stress_factor();
+
+        let f: ~fn() = || {
+            for uint::range(0,n) |_| {
+                let f: ~fn()  = || {
+                    // Borrow the scheduler we run on and check if it is
+                    // privileged.
+                    do Local::borrow::<Scheduler,()> |sched| {
+                        assert!(sched.run_anything);
+                    };
+                };
+                spawntask_random(f);
+            };
+        };
+
+        let f_cell = Cell::new(f);
+        let handles = Cell::new(handles);
+
+        rtdebug!("creating main task");
+
+        let main_task = ~do Coroutine::new_root(&mut scheds[0].stack_pool) {
+            f_cell.take()();
+            let mut handles = handles.take();
+            // Tell schedulers to exit
+            for handles.mut_iter().advance |handle| {
+                handle.send(Shutdown);
+            }
+        };
+
+        rtdebug!("queuing main task")
+
         scheds[0].enqueue_task(main_task);
 
         let mut threads = ~[];
 
         while !scheds.is_empty() {
             let sched = scheds.pop();
-            let sched_cell = Cell(sched);
+            let sched_cell = Cell::new(sched);
             let thread = do Thread::start {
                 let sched = sched_cell.take();
+                rtdebug!("running sched: %u", sched.sched_id());
                 sched.run();
             };
 
             threads.push(thread);
         }
 
+        rtdebug!("waiting on scheduler threads");
+
         // Wait for schedulers
         let _threads = threads;
     }
@@ -128,25 +247,34 @@ pub fn run_in_mt_newsched_task(f: ~fn()) {
     }
 }
 
+
 /// Test tasks will abort on failure instead of unwinding
 pub fn spawntask(f: ~fn()) {
     use super::sched::*;
 
+    rtdebug!("spawntask taking the scheduler from TLS")
+    let task = do Local::borrow::<Task, ~Task>() |running_task| {
+        ~running_task.new_child()
+    };
+
     let mut sched = Local::take::<Scheduler>();
     let task = ~Coroutine::with_task(&mut sched.stack_pool,
-                                     ~Task::without_unwinding(),
-                                     f);
-    sched.schedule_new_task(task);
+                                     task, f);
+    rtdebug!("spawntask scheduling the new task");
+    sched.schedule_task(task);
 }
 
 /// Create a new task and run it right now. Aborts on failure
 pub fn spawntask_immediately(f: ~fn()) {
     use super::sched::*;
 
+    let task = do Local::borrow::<Task, ~Task>() |running_task| {
+        ~running_task.new_child()
+    };
+
     let mut sched = Local::take::<Scheduler>();
     let task = ~Coroutine::with_task(&mut sched.stack_pool,
-                                     ~Task::without_unwinding(),
-                                     f);
+                                     task, f);
     do sched.switch_running_tasks_and_then(task) |sched, task| {
         sched.enqueue_task(task);
     }
@@ -156,10 +284,13 @@ pub fn spawntask_immediately(f: ~fn()) {
 pub fn spawntask_later(f: ~fn()) {
     use super::sched::*;
 
+    let task = do Local::borrow::<Task, ~Task>() |running_task| {
+        ~running_task.new_child()
+    };
+
     let mut sched = Local::take::<Scheduler>();
     let task = ~Coroutine::with_task(&mut sched.stack_pool,
-                                     ~Task::without_unwinding(),
-                                     f);
+                                     task, f);
 
     sched.enqueue_task(task);
     Local::put(sched);
@@ -170,13 +301,16 @@ pub fn spawntask_random(f: ~fn()) {
     use super::sched::*;
     use rand::{Rand, rng};
 
-    let mut rng = rng();
-    let run_now: bool = Rand::rand(&mut rng);
+    let task = do Local::borrow::<Task, ~Task>() |running_task| {
+        ~running_task.new_child()
+    };
 
     let mut sched = Local::take::<Scheduler>();
     let task = ~Coroutine::with_task(&mut sched.stack_pool,
-                                     ~Task::without_unwinding(),
-                                     f);
+                                     task, f);
+
+    let mut rng = rng();
+    let run_now: bool = Rand::rand(&mut rng);
 
     if run_now {
         do sched.switch_running_tasks_and_then(task) |sched, task| {
@@ -188,52 +322,75 @@ pub fn spawntask_random(f: ~fn()) {
     }
 }
 
+/// Spawn a task, with the current scheduler as home, and queue it to
+/// run later.
+pub fn spawntask_homed(scheds: &mut ~[~Scheduler], f: ~fn()) {
+    use super::sched::*;
+    use rand::{rng, RngUtil};
+    let mut rng = rng();
+
+    let task = {
+        let sched = &mut scheds[rng.gen_int_range(0,scheds.len() as int)];
+        let handle = sched.make_handle();
+        let home_id = handle.sched_id;
+
+        // now that we know where this is going, build a new function
+        // that can assert it is in the right place
+        let af: ~fn() = || {
+            do Local::borrow::<Scheduler,()>() |sched| {
+                rtdebug!("home_id: %u, runtime loc: %u",
+                         home_id,
+                         sched.sched_id());
+                assert!(home_id == sched.sched_id());
+            };
+            f()
+        };
+
+        ~Coroutine::with_task_homed(&mut sched.stack_pool,
+                                    ~Task::new_root(),
+                                    af,
+                                    Sched(handle))
+    };
+    let dest_sched = &mut scheds[rng.gen_int_range(0,scheds.len() as int)];
+    // enqueue it for future execution
+    dest_sched.enqueue_task(task);
+}
 
 /// Spawn a task and wait for it to finish, returning whether it completed successfully or failed
 pub fn spawntask_try(f: ~fn()) -> Result<(), ()> {
     use cell::Cell;
     use super::sched::*;
-    use task;
-    use unstable::finally::Finally;
-
-    // Our status variables will be filled in from the scheduler context
-    let mut failed = false;
-    let failed_ptr: *mut bool = &mut failed;
-
-    // Switch to the scheduler
-    let f = Cell(Cell(f));
-    let sched = Local::take::<Scheduler>();
-    do sched.deschedule_running_task_and_then() |sched, old_task| {
-        let old_task = Cell(old_task);
-        let f = f.take();
-        let new_task = ~do Coroutine::new(&mut sched.stack_pool) {
-            do (|| {
-                (f.take())()
-            }).finally {
-                // Check for failure then resume the parent task
-                unsafe { *failed_ptr = task::failing(); }
-                let sched = Local::take::<Scheduler>();
-                do sched.switch_running_tasks_and_then(old_task.take()) |sched, new_task| {
-                    sched.enqueue_task(new_task);
-                }
-            }
-        };
 
-        sched.enqueue_task(new_task);
+    let (port, chan) = oneshot();
+    let chan = Cell::new(chan);
+    let mut new_task = ~Task::new_root();
+    let on_exit: ~fn(bool) = |exit_status| chan.take().send(exit_status);
+    new_task.on_exit = Some(on_exit);
+    let mut sched = Local::take::<Scheduler>();
+    let new_task = ~Coroutine::with_task(&mut sched.stack_pool,
+                                         new_task, f);
+    do sched.switch_running_tasks_and_then(new_task) |sched, old_task| {
+        sched.enqueue_task(old_task);
     }
 
-    if !failed { Ok(()) } else { Err(()) }
+    let exit_status = port.recv();
+    if exit_status { Ok(()) } else { Err(()) }
 }
 
 // Spawn a new task in a new scheduler and return a thread handle.
 pub fn spawntask_thread(f: ~fn()) -> Thread {
     use rt::sched::*;
 
-    let f = Cell(f);
+    let task = do Local::borrow::<Task, ~Task>() |running_task| {
+        ~running_task.new_child()
+    };
+
+    let task = Cell::new(task);
+    let f = Cell::new(f);
     let thread = do Thread::start {
         let mut sched = ~new_test_uv_sched();
         let task = ~Coroutine::with_task(&mut sched.stack_pool,
-                                         ~Task::without_unwinding(),
+                                         task.take(),
                                          f.take());
         sched.enqueue_task(task);
         sched.run();
@@ -265,4 +422,3 @@ pub fn stress_factor() -> uint {
         None => 1
     }
 }
-
diff --git a/src/libstd/rt/thread.rs b/src/libstd/rt/thread.rs
index 0f1ae09bd94..bc290191310 100644
--- a/src/libstd/rt/thread.rs
+++ b/src/libstd/rt/thread.rs
@@ -19,8 +19,8 @@ pub struct Thread {
     raw_thread: *raw_thread
 }
 
-pub impl Thread {
-    fn start(main: ~fn()) -> Thread {
+impl Thread {
+    pub fn start(main: ~fn()) -> Thread {
         fn substart(main: &~fn()) -> *raw_thread {
             unsafe { rust_raw_thread_start(main) }
         }
diff --git a/src/libstd/rt/tube.rs b/src/libstd/rt/tube.rs
index 4482a92d916..89f3d10b5e4 100644
--- a/src/libstd/rt/tube.rs
+++ b/src/libstd/rt/tube.rs
@@ -105,7 +105,7 @@ mod test {
         do run_in_newsched_task {
             let mut tube: Tube<int> = Tube::new();
             let tube_clone = tube.clone();
-            let tube_clone_cell = Cell(tube_clone);
+            let tube_clone_cell = Cell::new(tube_clone);
             let sched = Local::take::<Scheduler>();
             do sched.deschedule_running_task_and_then |sched, task| {
                 let mut tube_clone = tube_clone_cell.take();
@@ -122,10 +122,10 @@ mod test {
         do run_in_newsched_task {
             let mut tube: Tube<int> = Tube::new();
             let tube_clone = tube.clone();
-            let tube_clone = Cell(tube_clone);
+            let tube_clone = Cell::new(tube_clone);
             let sched = Local::take::<Scheduler>();
             do sched.deschedule_running_task_and_then |sched, task| {
-                let tube_clone = Cell(tube_clone.take());
+                let tube_clone = Cell::new(tube_clone.take());
                 do sched.event_loop.callback {
                     let mut tube_clone = tube_clone.take();
                     // The task should be blocked on this now and
@@ -146,7 +146,7 @@ mod test {
         do run_in_newsched_task {
             let mut tube: Tube<int> = Tube::new();
             let tube_clone = tube.clone();
-            let tube_clone = Cell(tube_clone);
+            let tube_clone = Cell::new(tube_clone);
             let sched = Local::take::<Scheduler>();
             do sched.deschedule_running_task_and_then |sched, task| {
                 callback_send(tube_clone.take(), 0);
@@ -154,8 +154,8 @@ mod test {
                 fn callback_send(tube: Tube<int>, i: int) {
                     if i == 100 { return; }
 
-                    let tube = Cell(Cell(tube));
-                    do Local::borrow::<Scheduler> |sched| {
+                    let tube = Cell::new(Cell::new(tube));
+                    do Local::borrow::<Scheduler, ()> |sched| {
                         let tube = tube.take();
                         do sched.event_loop.callback {
                             let mut tube = tube.take();
diff --git a/src/libstd/rt/uv/async.rs b/src/libstd/rt/uv/async.rs
index 6ed06cc10b7..f3d1024024f 100644
--- a/src/libstd/rt/uv/async.rs
+++ b/src/libstd/rt/uv/async.rs
@@ -93,7 +93,7 @@ mod test {
         do run_in_bare_thread {
             let mut loop_ = Loop::new();
             let watcher = AsyncWatcher::new(&mut loop_, |w, _| w.close(||()) );
-            let watcher_cell = Cell(watcher);
+            let watcher_cell = Cell::new(watcher);
             let _thread = do Thread::start {
                 let mut watcher = watcher_cell.take();
                 watcher.send();
diff --git a/src/libstd/rt/uv/idle.rs b/src/libstd/rt/uv/idle.rs
index a81ab48696a..a3630c9b9bf 100644
--- a/src/libstd/rt/uv/idle.rs
+++ b/src/libstd/rt/uv/idle.rs
@@ -17,8 +17,8 @@ use rt::uv::status_to_maybe_uv_error;
 pub struct IdleWatcher(*uvll::uv_idle_t);
 impl Watcher for IdleWatcher { }
 
-pub impl IdleWatcher {
-    fn new(loop_: &mut Loop) -> IdleWatcher {
+impl IdleWatcher {
+    pub fn new(loop_: &mut Loop) -> IdleWatcher {
         unsafe {
             let handle = uvll::idle_new();
             assert!(handle.is_not_null());
@@ -29,7 +29,7 @@ pub impl IdleWatcher {
         }
     }
 
-    fn start(&mut self, cb: IdleCallback) {
+    pub fn start(&mut self, cb: IdleCallback) {
         {
             let data = self.get_watcher_data();
             data.idle_cb = Some(cb);
@@ -48,16 +48,17 @@ pub impl IdleWatcher {
         }
     }
 
-    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 free
+    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
+        // free
 
         unsafe {
             assert!(0 == uvll::idle_stop(self.native_handle()));
         }
     }
 
-    fn close(self, cb: NullCallback) {
+    pub fn close(self, cb: NullCallback) {
         {
             let mut this = self;
             let data = this.get_watcher_data();
diff --git a/src/libstd/rt/uv/mod.rs b/src/libstd/rt/uv/mod.rs
index f7cc5c6cc8b..a6a8edafe60 100644
--- a/src/libstd/rt/uv/mod.rs
+++ b/src/libstd/rt/uv/mod.rs
@@ -38,11 +38,9 @@ use container::Container;
 use option::*;
 use str::raw::from_c_str;
 use to_str::ToStr;
-use ptr::Ptr;
-use libc;
+use ptr::RawPtr;
 use vec;
 use ptr;
-use cast;
 use str;
 use libc::{c_void, c_int, size_t, malloc, free};
 use cast::transmute;
@@ -94,18 +92,18 @@ pub trait NativeHandle<T> {
     pub fn native_handle(&self) -> T;
 }
 
-pub impl Loop {
-    fn new() -> Loop {
+impl Loop {
+    pub fn new() -> Loop {
         let handle = unsafe { uvll::loop_new() };
         assert!(handle.is_not_null());
         NativeHandle::from_native_handle(handle)
     }
 
-    fn run(&mut self) {
+    pub fn run(&mut self) {
         unsafe { uvll::run(self.native_handle()) };
     }
 
-    fn close(&mut self) {
+    pub fn close(&mut self) {
         unsafe { uvll::loop_delete(self.native_handle()) };
     }
 }
@@ -204,9 +202,8 @@ impl<H, W: Watcher + NativeHandle<*H>> WatcherInterop for W {
 
 pub struct UvError(uvll::uv_err_t);
 
-pub impl UvError {
-
-    fn name(&self) -> ~str {
+impl UvError {
+    pub fn name(&self) -> ~str {
         unsafe {
             let inner = match self { &UvError(ref a) => a };
             let name_str = uvll::err_name(inner);
@@ -215,7 +212,7 @@ pub impl UvError {
         }
     }
 
-    fn desc(&self) -> ~str {
+    pub fn desc(&self) -> ~str {
         unsafe {
             let inner = match self { &UvError(ref a) => a };
             let desc_str = uvll::strerror(inner);
@@ -224,7 +221,7 @@ pub impl UvError {
         }
     }
 
-    fn is_eof(&self) -> bool {
+    pub fn is_eof(&self) -> bool {
         self.code == uvll::EOF
     }
 }
@@ -250,20 +247,6 @@ pub fn last_uv_error<H, W: Watcher + NativeHandle<*H>>(watcher: &W) -> UvError {
 }
 
 pub fn uv_error_to_io_error(uverr: UvError) -> IoError {
-
-    // XXX: Could go in str::raw
-    unsafe fn c_str_to_static_slice(s: *libc::c_char) -> &'static str {
-        let s = s as *u8;
-        let mut curr = s, len = 0u;
-        while *curr != 0u8 {
-            len += 1u;
-            curr = ptr::offset(s, len);
-        }
-
-        str::raw::buf_as_slice(s, len, |d| cast::transmute(d))
-    }
-
-
     unsafe {
         // Importing error constants
         use rt::uv::uvll::*;
@@ -271,7 +254,7 @@ pub fn uv_error_to_io_error(uverr: UvError) -> IoError {
 
         // uv error descriptions are static
         let c_desc = uvll::strerror(&*uverr);
-        let desc = c_str_to_static_slice(c_desc);
+        let desc = str::raw::c_str_to_static_slice(c_desc);
 
         let kind = match uverr.code {
             UNKNOWN => OtherIoError,
diff --git a/src/libstd/rt/uv/net.rs b/src/libstd/rt/uv/net.rs
index 0b77bd83958..5491b82b725 100644
--- a/src/libstd/rt/uv/net.rs
+++ b/src/libstd/rt/uv/net.rs
@@ -51,9 +51,8 @@ pub fn uv_ip4_to_ip4(addr: *sockaddr_in) -> IpAddr {
 pub struct StreamWatcher(*uvll::uv_stream_t);
 impl Watcher for StreamWatcher { }
 
-pub impl StreamWatcher {
-
-    fn read_start(&mut self, alloc: AllocCallback, cb: ReadCallback) {
+impl StreamWatcher {
+    pub fn read_start(&mut self, alloc: AllocCallback, cb: ReadCallback) {
         {
             let data = self.get_watcher_data();
             data.alloc_cb = Some(alloc);
@@ -81,7 +80,7 @@ pub impl StreamWatcher {
         }
     }
 
-    fn read_stop(&mut self) {
+    pub fn read_stop(&mut self) {
         // It would be nice to drop the alloc and read callbacks here,
         // but read_stop may be called from inside one of them and we
         // would end up freeing the in-use environment
@@ -89,7 +88,7 @@ pub impl StreamWatcher {
         unsafe { uvll::read_stop(handle); }
     }
 
-    fn write(&mut self, buf: Buf, cb: ConnectionCallback) {
+    pub fn write(&mut self, buf: Buf, cb: ConnectionCallback) {
         {
             let data = self.get_watcher_data();
             assert!(data.write_cb.is_none());
@@ -118,7 +117,7 @@ pub impl StreamWatcher {
         }
     }
 
-    fn accept(&mut self, stream: StreamWatcher) {
+    pub fn accept(&mut self, stream: StreamWatcher) {
         let self_handle = self.native_handle() as *c_void;
         let stream_handle = stream.native_handle() as *c_void;
         unsafe {
@@ -126,7 +125,7 @@ pub impl StreamWatcher {
         }
     }
 
-    fn close(self, cb: NullCallback) {
+    pub fn close(self, cb: NullCallback) {
         {
             let mut this = self;
             let data = this.get_watcher_data();
@@ -161,8 +160,8 @@ impl NativeHandle<*uvll::uv_stream_t> for StreamWatcher {
 pub struct TcpWatcher(*uvll::uv_tcp_t);
 impl Watcher for TcpWatcher { }
 
-pub impl TcpWatcher {
-    fn new(loop_: &mut Loop) -> TcpWatcher {
+impl TcpWatcher {
+    pub fn new(loop_: &mut Loop) -> TcpWatcher {
         unsafe {
             let handle = malloc_handle(UV_TCP);
             assert!(handle.is_not_null());
@@ -173,7 +172,7 @@ pub impl TcpWatcher {
         }
     }
 
-    fn bind(&mut self, address: IpAddr) -> Result<(), UvError> {
+    pub fn bind(&mut self, address: IpAddr) -> Result<(), UvError> {
         match address {
             Ipv4(*) => {
                 do ip4_as_uv_ip4(address) |addr| {
@@ -191,7 +190,7 @@ pub impl TcpWatcher {
         }
     }
 
-    fn connect(&mut self, address: IpAddr, cb: ConnectionCallback) {
+    pub fn connect(&mut self, address: IpAddr, cb: ConnectionCallback) {
         unsafe {
             assert!(self.get_watcher_data().connect_cb.is_none());
             self.get_watcher_data().connect_cb = Some(cb);
@@ -224,7 +223,7 @@ pub impl TcpWatcher {
         }
     }
 
-    fn listen(&mut self, cb: ConnectionCallback) {
+    pub fn listen(&mut self, cb: ConnectionCallback) {
         {
             let data = self.get_watcher_data();
             assert!(data.connect_cb.is_none());
@@ -248,7 +247,7 @@ pub impl TcpWatcher {
         }
     }
 
-    fn as_stream(&self) -> StreamWatcher {
+    pub fn as_stream(&self) -> StreamWatcher {
         NativeHandle::from_native_handle(self.native_handle() as *uvll::uv_stream_t)
     }
 }
@@ -439,9 +438,8 @@ pub struct WriteRequest(*uvll::uv_write_t);
 
 impl Request for WriteRequest { }
 
-pub impl WriteRequest {
-
-    fn new() -> WriteRequest {
+impl WriteRequest {
+    pub fn new() -> WriteRequest {
         let write_handle = unsafe {
             malloc_req(UV_WRITE)
         };
@@ -450,14 +448,14 @@ pub impl WriteRequest {
         WriteRequest(write_handle)
     }
 
-    fn stream(&self) -> StreamWatcher {
+    pub fn stream(&self) -> StreamWatcher {
         unsafe {
             let stream_handle = uvll::get_stream_handle_from_write_req(self.native_handle());
             NativeHandle::from_native_handle(stream_handle)
         }
     }
 
-    fn delete(self) {
+    pub fn delete(self) {
         unsafe { free_req(self.native_handle() as *c_void) }
     }
 }
@@ -554,7 +552,7 @@ mod test {
                 let client_tcp_watcher = TcpWatcher::new(&mut loop_);
                 let mut client_tcp_watcher = client_tcp_watcher.as_stream();
                 server_stream_watcher.accept(client_tcp_watcher);
-                let count_cell = Cell(0);
+                let count_cell = Cell::new(0);
                 let server_stream_watcher = server_stream_watcher;
                 rtdebug!("starting read");
                 let alloc: AllocCallback = |size| {
@@ -594,11 +592,11 @@ mod test {
                     let mut stream_watcher = stream_watcher;
                     let msg = ~[0, 1, 2, 3, 4, 5, 6 ,7 ,8, 9];
                     let buf = slice_to_uv_buf(msg);
-                    let msg_cell = Cell(msg);
+                    let msg_cell = Cell::new(msg);
                     do stream_watcher.write(buf) |stream_watcher, status| {
                         rtdebug!("writing");
                         assert!(status.is_none());
-                        let msg_cell = Cell(msg_cell.take());
+                        let msg_cell = Cell::new(msg_cell.take());
                         stream_watcher.close(||ignore(msg_cell.take()));
                     }
                 }
diff --git a/src/libstd/rt/uv/uvio.rs b/src/libstd/rt/uv/uvio.rs
index 1274dbc3220..eba84e537f9 100644
--- a/src/libstd/rt/uv/uvio.rs
+++ b/src/libstd/rt/uv/uvio.rs
@@ -11,7 +11,7 @@
 use option::*;
 use result::*;
 use ops::Drop;
-use cell::{Cell, empty_cell};
+use cell::Cell;
 use cast;
 use cast::transmute;
 use clone::Clone;
@@ -35,8 +35,8 @@ pub struct UvEventLoop {
     uvio: UvIoFactory
 }
 
-pub impl UvEventLoop {
-    fn new() -> UvEventLoop {
+impl UvEventLoop {
+    pub fn new() -> UvEventLoop {
         UvEventLoop {
             uvio: UvIoFactory(Loop::new())
         }
@@ -54,7 +54,6 @@ impl Drop for UvEventLoop {
 }
 
 impl EventLoop for UvEventLoop {
-
     fn run(&mut self) {
         self.uvio.uv_loop().run();
     }
@@ -117,9 +116,11 @@ impl UvRemoteCallback {
         let async = do AsyncWatcher::new(loop_) |watcher, status| {
             assert!(status.is_none());
             f();
-            do exit_flag_clone.with_imm |&should_exit| {
-                if should_exit {
-                    watcher.close(||());
+            unsafe {
+                do exit_flag_clone.with_imm |&should_exit| {
+                    if should_exit {
+                        watcher.close(||());
+                    }
                 }
             }
         };
@@ -152,7 +153,6 @@ impl Drop for UvRemoteCallback {
 
 #[cfg(test)]
 mod test_remote {
-    use cell;
     use cell::Cell;
     use rt::test::*;
     use rt::thread::Thread;
@@ -166,10 +166,10 @@ mod test_remote {
         do run_in_newsched_task {
             let mut tube = Tube::new();
             let tube_clone = tube.clone();
-            let remote_cell = cell::empty_cell();
-            do Local::borrow::<Scheduler>() |sched| {
+            let remote_cell = Cell::new_empty();
+            do Local::borrow::<Scheduler, ()>() |sched| {
                 let tube_clone = tube_clone.clone();
-                let tube_clone_cell = Cell(tube_clone);
+                let tube_clone_cell = Cell::new(tube_clone);
                 let remote = do sched.event_loop.remote_callback {
                     tube_clone_cell.take().send(1);
                 };
@@ -186,8 +186,8 @@ mod test_remote {
 
 pub struct UvIoFactory(Loop);
 
-pub impl UvIoFactory {
-    fn uv_loop<'a>(&'a mut self) -> &'a mut Loop {
+impl UvIoFactory {
+    pub fn uv_loop<'a>(&'a mut self) -> &'a mut Loop {
         match self { &UvIoFactory(ref mut ptr) => ptr }
     }
 }
@@ -199,7 +199,7 @@ impl IoFactory for UvIoFactory {
     fn tcp_connect(&mut self, addr: IpAddr) -> Result<~RtioTcpStreamObject, IoError> {
         // Create a cell in the task to hold the result. We will fill
         // the cell before resuming the task.
-        let result_cell = empty_cell();
+        let result_cell = Cell::new_empty();
         let result_cell_ptr: *Cell<Result<~RtioTcpStreamObject, IoError>> = &result_cell;
 
         let scheduler = Local::take::<Scheduler>();
@@ -211,7 +211,7 @@ impl IoFactory for UvIoFactory {
             rtdebug!("connect: entered scheduler context");
             assert!(!sched.in_task_context());
             let mut tcp_watcher = TcpWatcher::new(self.uv_loop());
-            let task_cell = Cell(task);
+            let task_cell = Cell::new(task);
 
             // Wait for a connection
             do tcp_watcher.connect(addr) |stream_watcher, status| {
@@ -228,7 +228,7 @@ impl IoFactory for UvIoFactory {
                     scheduler.resume_task_immediately(task_cell.take());
                 } else {
                     rtdebug!("status is some");
-                    let task_cell = Cell(task_cell.take());
+                    let task_cell = Cell::new(task_cell.take());
                     do stream_watcher.close {
                         let res = Err(uv_error_to_io_error(status.get()));
                         unsafe { (*result_cell_ptr).put_back(res); }
@@ -250,7 +250,7 @@ impl IoFactory for UvIoFactory {
             Err(uverr) => {
                 let scheduler = Local::take::<Scheduler>();
                 do scheduler.deschedule_running_task_and_then |_, task| {
-                    let task_cell = Cell(task);
+                    let task_cell = Cell::new(task);
                     do watcher.as_stream().close {
                         let scheduler = Local::take::<Scheduler>();
                         scheduler.resume_task_immediately(task_cell.take());
@@ -286,7 +286,7 @@ impl Drop for UvTcpListener {
         let watcher = self.watcher();
         let scheduler = Local::take::<Scheduler>();
         do scheduler.deschedule_running_task_and_then |_, task| {
-            let task_cell = Cell(task);
+            let task_cell = Cell::new(task);
             do watcher.as_stream().close {
                 let scheduler = Local::take::<Scheduler>();
                 scheduler.resume_task_immediately(task_cell.take());
@@ -307,9 +307,9 @@ impl RtioTcpListener for UvTcpListener {
         self.listening = true;
 
         let server_tcp_watcher = self.watcher();
-        let incoming_streams_cell = Cell(self.incoming_streams.clone());
+        let incoming_streams_cell = Cell::new(self.incoming_streams.clone());
 
-        let incoming_streams_cell = Cell(incoming_streams_cell.take());
+        let incoming_streams_cell = Cell::new(incoming_streams_cell.take());
         let mut server_tcp_watcher = server_tcp_watcher;
         do server_tcp_watcher.listen |server_stream_watcher, status| {
             let maybe_stream = if status.is_none() {
@@ -348,7 +348,7 @@ impl Drop for UvTcpStream {
         let watcher = self.watcher();
         let scheduler = Local::take::<Scheduler>();
         do scheduler.deschedule_running_task_and_then |_, task| {
-            let task_cell = Cell(task);
+            let task_cell = Cell::new(task);
             do watcher.close {
                 let scheduler = Local::take::<Scheduler>();
                 scheduler.resume_task_immediately(task_cell.take());
@@ -359,7 +359,7 @@ impl Drop for UvTcpStream {
 
 impl RtioTcpStream for UvTcpStream {
     fn read(&mut self, buf: &mut [u8]) -> Result<uint, IoError> {
-        let result_cell = empty_cell();
+        let result_cell = Cell::new_empty();
         let result_cell_ptr: *Cell<Result<uint, IoError>> = &result_cell;
 
         let scheduler = Local::take::<Scheduler>();
@@ -370,7 +370,7 @@ impl RtioTcpStream for UvTcpStream {
             rtdebug!("read: entered scheduler context");
             assert!(!sched.in_task_context());
             let mut watcher = watcher;
-            let task_cell = Cell(task);
+            let task_cell = Cell::new(task);
             // XXX: We shouldn't reallocate these callbacks every
             // call to read
             let alloc: AllocCallback = |_| unsafe {
@@ -404,7 +404,7 @@ impl RtioTcpStream for UvTcpStream {
     }
 
     fn write(&mut self, buf: &[u8]) -> Result<(), IoError> {
-        let result_cell = empty_cell();
+        let result_cell = Cell::new_empty();
         let result_cell_ptr: *Cell<Result<(), IoError>> = &result_cell;
         let scheduler = Local::take::<Scheduler>();
         assert!(scheduler.in_task_context());
@@ -412,7 +412,7 @@ impl RtioTcpStream for UvTcpStream {
         let buf_ptr: *&[u8] = &buf;
         do scheduler.deschedule_running_task_and_then |_, task| {
             let mut watcher = watcher;
-            let task_cell = Cell(task);
+            let task_cell = Cell::new(task);
             let buf = unsafe { slice_to_uv_buf(*buf_ptr) };
             do watcher.write(buf) |_watcher, status| {
                 let result = if status.is_none() {
@@ -585,7 +585,7 @@ fn test_read_and_block() {
                 // will trigger a read callback while we are
                 // not ready for it
                 do scheduler.deschedule_running_task_and_then |sched, task| {
-                    let task = Cell(task);
+                    let task = Cell::new(task);
                     sched.enqueue_task(task.take());
                 }
             }
diff --git a/src/libstd/rt/uv/uvll.rs b/src/libstd/rt/uv/uvll.rs
index 06315c4cb38..b73db77f3bb 100644
--- a/src/libstd/rt/uv/uvll.rs
+++ b/src/libstd/rt/uv/uvll.rs
@@ -31,7 +31,11 @@
 
 use libc::{size_t, c_int, c_uint, c_void, c_char, uintptr_t};
 use libc::{malloc, free};
+use libc;
 use prelude::*;
+use ptr;
+use str;
+use vec;
 
 pub static UNKNOWN: c_int = -1;
 pub static OK: c_int = 0;
diff --git a/src/libstd/rt/work_queue.rs b/src/libstd/rt/work_queue.rs
index e9eb663392b..cfffc55a58c 100644
--- a/src/libstd/rt/work_queue.rs
+++ b/src/libstd/rt/work_queue.rs
@@ -21,40 +21,48 @@ pub struct WorkQueue<T> {
     priv queue: ~Exclusive<~[T]>
 }
 
-pub impl<T: Owned> WorkQueue<T> {
-    fn new() -> WorkQueue<T> {
+impl<T: Owned> WorkQueue<T> {
+    pub fn new() -> WorkQueue<T> {
         WorkQueue {
             queue: ~exclusive(~[])
         }
     }
 
-    fn push(&mut self, value: T) {
-        let value = Cell(value);
-        self.queue.with(|q| q.unshift(value.take()) );
+    pub fn push(&mut self, value: T) {
+        unsafe {
+            let value = Cell::new(value);
+            self.queue.with(|q| q.unshift(value.take()) );
+        }
     }
 
-    fn pop(&mut self) -> Option<T> {
-        do self.queue.with |q| {
-            if !q.is_empty() {
-                Some(q.shift())
-            } else {
-                None
+    pub fn pop(&mut self) -> Option<T> {
+        unsafe {
+            do self.queue.with |q| {
+                if !q.is_empty() {
+                    Some(q.shift())
+                } else {
+                    None
+                }
             }
         }
     }
 
-    fn steal(&mut self) -> Option<T> {
-        do self.queue.with |q| {
-            if !q.is_empty() {
-                Some(q.pop())
-            } else {
-                None
+    pub fn steal(&mut self) -> Option<T> {
+        unsafe {
+            do self.queue.with |q| {
+                if !q.is_empty() {
+                    Some(q.pop())
+                } else {
+                    None
+                }
             }
         }
     }
 
-    fn is_empty(&self) -> bool {
-        self.queue.with_imm(|q| q.is_empty() )
+    pub fn is_empty(&self) -> bool {
+        unsafe {
+            self.queue.with_imm(|q| q.is_empty() )
+        }
     }
 }