auto merge of #8282 : brson/rust/more-newsched-fixes, r=brson

7 files changed, 74 insertions, 211 deletions

diff --git a/src/libstd/comm.rs b/src/libstd/comm.rs
index a0731dc3494..4356f1143da 100644
--- a/src/libstd/comm.rs
+++ b/src/libstd/comm.rs
@@ -14,7 +14,6 @@ Message passing
 
 #[allow(missing_doc)];
 
-use cast::transmute;
 use either::{Either, Left, Right};
 use kinds::Send;
 use option::{Option, Some};
@@ -23,12 +22,6 @@ pub use rt::comm::SendDeferred;
 use rtcomm = rt::comm;
 use rt;
 
-use pipes::{wait_many, PacketHeader};
-
-// FIXME #5160: Making this public exposes some plumbing from
-// pipes. Needs some refactoring
-pub use pipes::Selectable;
-
 /// A trait for things that can send multiple messages.
 pub trait GenericChan<T> {
     /// Sends a message.
@@ -146,15 +139,6 @@ impl<T: Send> Peekable<T> for Port<T> {
     }
 }
 
-impl<T: Send> Selectable for Port<T> {
-    fn header(&mut self) -> *mut PacketHeader {
-        match self.inner {
-            Left(ref mut port) => port.header(),
-            Right(_) => fail!("can't select on newsched ports")
-        }
-    }
-}
-
 /// A channel that can be shared between many senders.
 pub struct SharedChan<T> {
     inner: Either<Exclusive<pipesy::Chan<T>>, rtcomm::SharedChan<T>>
@@ -318,8 +302,8 @@ mod pipesy {
 
     use kinds::Send;
     use option::{Option, Some, None};
-    use pipes::{recv, try_recv, peek, PacketHeader};
-    use super::{GenericChan, GenericSmartChan, GenericPort, Peekable, Selectable};
+    use pipes::{recv, try_recv, peek};
+    use super::{GenericChan, GenericSmartChan, GenericPort, Peekable};
     use cast::transmute_mut;
 
     /*proto! oneshot (
@@ -651,80 +635,6 @@ mod pipesy {
         }
     }
 
-    impl<T: Send> Selectable for Port<T> {
-        fn header(&mut self) -> *mut PacketHeader {
-            match self.endp {
-                Some(ref mut endp) => endp.header(),
-                None => fail!("peeking empty stream")
-            }
-    }
-}
-
-}
-
-/// Returns the index of an endpoint that is ready to receive.
-pub fn selecti<T: Selectable>(endpoints: &mut [T]) -> uint {
-    wait_many(endpoints)
-}
-
-/// Returns 0 or 1 depending on which endpoint is ready to receive
-pub fn select2i<A:Selectable, B:Selectable>(a: &mut A, b: &mut B)
-                                            -> Either<(), ()> {
-    let mut endpoints = [ a.header(), b.header() ];
-    match wait_many(endpoints) {
-        0 => Left(()),
-        1 => Right(()),
-        _ => fail!("wait returned unexpected index"),
-    }
-}
-
-/// Receive a message from one of two endpoints.
-pub trait Select2<T: Send, U: Send> {
-    /// Receive a message or return `None` if a connection closes.
-    fn try_select(&mut self) -> Either<Option<T>, Option<U>>;
-    /// Receive a message or fail if a connection closes.
-    fn select(&mut self) -> Either<T, U>;
-}
-
-impl<T:Send,
-     U:Send,
-     Left:Selectable + GenericPort<T>,
-     Right:Selectable + GenericPort<U>>
-     Select2<T, U>
-     for (Left, Right) {
-    fn select(&mut self) -> Either<T, U> {
-        // XXX: Bad borrow check workaround.
-        unsafe {
-            let this: &(Left, Right) = transmute(self);
-            match *this {
-                (ref lp, ref rp) => {
-                    let lp: &mut Left = transmute(lp);
-                    let rp: &mut Right = transmute(rp);
-                    match select2i(lp, rp) {
-                        Left(()) => Left(lp.recv()),
-                        Right(()) => Right(rp.recv()),
-                    }
-                }
-            }
-        }
-    }
-
-    fn try_select(&mut self) -> Either<Option<T>, Option<U>> {
-        // XXX: Bad borrow check workaround.
-        unsafe {
-            let this: &(Left, Right) = transmute(self);
-            match *this {
-                (ref lp, ref rp) => {
-                    let lp: &mut Left = transmute(lp);
-                    let rp: &mut Right = transmute(rp);
-                    match select2i(lp, rp) {
-                        Left(()) => Left (lp.try_recv()),
-                        Right(()) => Right(rp.try_recv()),
-                    }
-                }
-            }
-        }
-    }
 }
 
 #[cfg(test)]
@@ -733,22 +643,6 @@ mod test {
     use super::{Chan, Port, oneshot, stream};
 
     #[test]
-    fn test_select2() {
-        let (p1, c1) = stream();
-        let (p2, c2) = stream();
-
-        c1.send(~"abc");
-
-        let mut tuple = (p1, p2);
-        match tuple.select() {
-            Right(_) => fail!(),
-            _ => (),
-        }
-
-        c2.send(123);
-    }
-
-    #[test]
     fn test_oneshot() {
         let (p, c) = oneshot();
 
diff --git a/src/libstd/pipes.rs b/src/libstd/pipes.rs
index 1fd534825a5..78f937e058a 100644
--- a/src/libstd/pipes.rs
+++ b/src/libstd/pipes.rs
@@ -868,47 +868,3 @@ pub mod rt {
     pub fn make_some<T>(val: T) -> Option<T> { Some(val) }
     pub fn make_none<T>() -> Option<T> { None }
 }
-
-#[cfg(test)]
-mod test {
-    use either::Right;
-    use comm::{Chan, Port, oneshot, recv_one, stream, Select2,
-               GenericChan, Peekable};
-
-    #[test]
-    fn test_select2() {
-        let (p1, c1) = stream();
-        let (p2, c2) = stream();
-
-        c1.send(~"abc");
-
-        let mut tuple = (p1, p2);
-        match tuple.select() {
-            Right(_) => fail!(),
-            _ => (),
-        }
-
-        c2.send(123);
-    }
-
-    #[test]
-    fn test_oneshot() {
-        let (p, c) = oneshot();
-
-        c.send(());
-
-        recv_one(p)
-    }
-
-    #[test]
-    fn test_peek_terminated() {
-        let (port, chan): (Port<int>, Chan<int>) = stream();
-
-        {
-            // Destroy the channel
-            let _chan = chan;
-        }
-
-        assert!(!port.peek());
-    }
-}
diff --git a/src/libstd/rt/local.rs b/src/libstd/rt/local.rs
index 7ab63233cff..131507196b1 100644
--- a/src/libstd/rt/local.rs
+++ b/src/libstd/rt/local.rs
@@ -126,6 +126,7 @@ impl Local for IoFactoryObject {
 
 #[cfg(test)]
 mod test {
+    use unstable::run_in_bare_thread;
     use rt::test::*;
     use super::*;
     use rt::task::Task;
@@ -133,56 +134,64 @@ mod test {
 
     #[test]
     fn thread_local_task_smoke_test() {
-        local_ptr::init_tls_key();
-        let mut sched = ~new_test_uv_sched();
-        let task = ~Task::new_root(&mut sched.stack_pool, || {});
-        Local::put(task);
-        let task: ~Task = Local::take();
-        cleanup_task(task);
+        do run_in_bare_thread {
+            local_ptr::init_tls_key();
+            let mut sched = ~new_test_uv_sched();
+            let task = ~Task::new_root(&mut sched.stack_pool, || {});
+            Local::put(task);
+            let task: ~Task = Local::take();
+            cleanup_task(task);
+        }
     }
 
     #[test]
     fn thread_local_task_two_instances() {
-        local_ptr::init_tls_key();
-        let mut sched = ~new_test_uv_sched();
-        let task = ~Task::new_root(&mut sched.stack_pool, || {});
-        Local::put(task);
-        let task: ~Task = Local::take();
-        cleanup_task(task);
-        let task = ~Task::new_root(&mut sched.stack_pool, || {});
-        Local::put(task);
-        let task: ~Task = Local::take();
-        cleanup_task(task);
+        do run_in_bare_thread {
+            local_ptr::init_tls_key();
+            let mut sched = ~new_test_uv_sched();
+            let task = ~Task::new_root(&mut sched.stack_pool, || {});
+            Local::put(task);
+            let task: ~Task = Local::take();
+            cleanup_task(task);
+            let task = ~Task::new_root(&mut sched.stack_pool, || {});
+            Local::put(task);
+            let task: ~Task = Local::take();
+            cleanup_task(task);
+        }
 
     }
 
     #[test]
     fn borrow_smoke_test() {
-        local_ptr::init_tls_key();
-        let mut sched = ~new_test_uv_sched();
-        let task = ~Task::new_root(&mut sched.stack_pool, || {});
-        Local::put(task);
-
-        unsafe {
-            let _task: *mut Task = Local::unsafe_borrow();
+        do run_in_bare_thread {
+            local_ptr::init_tls_key();
+            let mut sched = ~new_test_uv_sched();
+            let task = ~Task::new_root(&mut sched.stack_pool, || {});
+            Local::put(task);
+
+            unsafe {
+                let _task: *mut Task = Local::unsafe_borrow();
+            }
+            let task: ~Task = Local::take();
+            cleanup_task(task);
         }
-        let task: ~Task = Local::take();
-        cleanup_task(task);
     }
 
     #[test]
     fn borrow_with_return() {
-        local_ptr::init_tls_key();
-        let mut sched = ~new_test_uv_sched();
-        let task = ~Task::new_root(&mut sched.stack_pool, || {});
-        Local::put(task);
-
-        let res = do Local::borrow::<Task,bool> |_task| {
-            true
-        };
-        assert!(res)
-        let task: ~Task = Local::take();
-        cleanup_task(task);
+        do run_in_bare_thread {
+            local_ptr::init_tls_key();
+            let mut sched = ~new_test_uv_sched();
+            let task = ~Task::new_root(&mut sched.stack_pool, || {});
+            Local::put(task);
+
+            let res = do Local::borrow::<Task,bool> |_task| {
+                true
+            };
+            assert!(res)
+                let task: ~Task = Local::take();
+            cleanup_task(task);
+        }
     }
 
 }
diff --git a/src/libstd/rt/local_ptr.rs b/src/libstd/rt/local_ptr.rs
index cd7c5daa444..652e39b05c7 100644
--- a/src/libstd/rt/local_ptr.rs
+++ b/src/libstd/rt/local_ptr.rs
@@ -52,7 +52,9 @@ pub unsafe fn put<T>(sched: ~T) {
 pub unsafe fn take<T>() -> ~T {
     let key = tls_key();
     let void_ptr: *mut c_void = tls::get(key);
-    rtassert!(void_ptr.is_not_null());
+    if void_ptr.is_null() {
+        rtabort!("thread-local pointer is null. bogus!");
+    }
     let ptr: ~T = cast::transmute(void_ptr);
     tls::set(key, ptr::mut_null());
     return ptr;
@@ -68,8 +70,8 @@ pub fn exists() -> bool {
     }
 }
 
-/// Borrow the thread-local scheduler from thread-local storage.
-/// While the scheduler is borrowed it is not available in TLS.
+/// Borrow the thread-local value from thread-local storage.
+/// While the value is borrowed it is not available in TLS.
 ///
 /// # Safety note
 ///
@@ -88,21 +90,23 @@ pub unsafe fn borrow<T>(f: &fn(&mut T)) {
     }
 }
 
-/// Borrow a mutable reference to the thread-local Scheduler
+/// Borrow a mutable reference to the thread-local value
 ///
 /// # Safety Note
 ///
-/// Because this leaves the Scheduler in thread-local storage it is possible
+/// Because this leaves the value in thread-local storage it is possible
 /// For the Scheduler pointer to be aliased
 pub unsafe fn unsafe_borrow<T>() -> *mut T {
     let key = tls_key();
-    let mut void_sched: *mut c_void = tls::get(key);
-    rtassert!(void_sched.is_not_null());
+    let mut void_ptr: *mut c_void = tls::get(key);
+    if void_ptr.is_null() {
+        rtabort!("thread-local pointer is null. bogus!");
+    }
     {
-        let sched: *mut *mut c_void = &mut void_sched;
-        let sched: *mut ~T = sched as *mut ~T;
-        let sched: *mut T = &mut **sched;
-        return sched;
+        let ptr: *mut *mut c_void = &mut void_ptr;
+        let ptr: *mut ~T = ptr as *mut ~T;
+        let ptr: *mut T = &mut **ptr;
+        return ptr;
     }
 }
 
diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs
index 4cfe0efacfe..33e83fd9040 100644
--- a/src/libstd/rt/mod.rs
+++ b/src/libstd/rt/mod.rs
@@ -432,13 +432,3 @@ pub fn context() -> RuntimeContext {
         pub fn rust_try_get_task() -> *rust_task;
     }
 }
-
-#[test]
-fn test_context() {
-    use unstable::run_in_bare_thread;
-
-    assert_eq!(context(), OldTaskContext);
-    do run_in_bare_thread {
-        assert_eq!(context(), GlobalContext);
-    }
-}
diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index 18cfeade157..a5c8abc2a6c 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -172,6 +172,10 @@ impl Scheduler {
 
         rtdebug!("stopping scheduler %u", stask.sched.get_ref().sched_id());
 
+        // Should not have any messages
+        let message = stask.sched.get_mut_ref().message_queue.pop();
+        assert!(message.is_none());
+
         stask.destroyed = true;
     }
 
@@ -336,11 +340,14 @@ impl Scheduler {
         match this.message_queue.pop() {
             Some(PinnedTask(task)) => {
                 let mut task = task;
+                this.event_loop.callback(Scheduler::run_sched_once);
                 task.give_home(Sched(this.make_handle()));
                 this.resume_task_immediately(task);
                 return None;
             }
             Some(TaskFromFriend(task)) => {
+                this.event_loop.callback(Scheduler::run_sched_once);
+                rtdebug!("got a task from a friend. lovely!");
                 return this.sched_schedule_task(task);
             }
             Some(Wake) => {
@@ -395,6 +402,7 @@ impl Scheduler {
     /// Take a non-homed task we aren't allowed to run here and send
     /// it to the designated friend scheduler to execute.
     fn send_to_friend(&mut self, task: ~Task) {
+        rtdebug!("sending a task to friend");
         match self.friend_handle {
             Some(ref mut handle) => {
                 handle.send(TaskFromFriend(task));
@@ -426,12 +434,14 @@ impl Scheduler {
                             Scheduler::send_task_home(task);
                             return Some(this);
                         } else {
+                            this.event_loop.callback(Scheduler::run_sched_once);
                             task.give_home(Sched(home_handle));
                             this.resume_task_immediately(task);
                             return None;
                         }
                     }
                     AnySched if this.run_anything => {
+                        this.event_loop.callback(Scheduler::run_sched_once);
                         task.give_home(AnySched);
                         this.resume_task_immediately(task);
                         return None;
diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs
index abafe1cf209..ffe9f118b47 100644
--- a/src/libstd/rt/task.rs
+++ b/src/libstd/rt/task.rs
@@ -303,7 +303,7 @@ impl Task {
 impl Drop for Task {
     fn drop(&self) {
         rtdebug!("called drop for a task: %u", borrow::to_uint(self));
-        assert!(self.destroyed)
+        rtassert!(self.destroyed)
     }
 }
 
@@ -313,7 +313,7 @@ impl Drop for Task {
 impl Coroutine {
 
     pub fn new(stack_pool: &mut StackPool, start: ~fn()) -> Coroutine {
-        static MIN_STACK_SIZE: uint = 2000000; // XXX: Too much stack
+        static MIN_STACK_SIZE: uint = 3000000; // XXX: Too much stack
 
         let start = Coroutine::build_start_wrapper(start);
         let mut stack = stack_pool.take_segment(MIN_STACK_SIZE);