auto merge of #8008 : bblum/rust/select, r=brson

Main logic in ```Implement select() for new runtime pipes.```. The guts of the ```PortOne::try_recv()``` implementation are now split up across several functions, ```optimistic_check```, ```block_on```, and ```recv_ready```.

There is one weird FIXME I left open here, in the "implement select" commit -- an assertion I couldn't get to work in the receive path, on an invariant that for some reason doesn't hold with ```SharedPort```. Still investigating this.

4 files changed, 557 insertions, 100 deletions

diff --git a/src/libstd/rt/comm.rs b/src/libstd/rt/comm.rs
index c0effdaa94c..6528835c52c 100644
--- a/src/libstd/rt/comm.rs
+++ b/src/libstd/rt/comm.rs
@@ -12,13 +12,13 @@
 
 use option::*;
 use cast;
-use util;
 use ops::Drop;
 use rt::kill::BlockedTask;
 use kinds::Send;
 use rt::sched::Scheduler;
 use rt::local::Local;
-use unstable::atomics::{AtomicUint, AtomicOption, SeqCst};
+use rt::select::{Select, SelectPort};
+use unstable::atomics::{AtomicUint, AtomicOption, Acquire, Release, SeqCst};
 use unstable::sync::UnsafeAtomicRcBox;
 use util::Void;
 use comm::{GenericChan, GenericSmartChan, GenericPort, Peekable};
@@ -45,23 +45,12 @@ struct Packet<T> {
 
 /// A one-shot channel.
 pub struct ChanOne<T> {
-    // XXX: Hack extra allocation to make by-val self work
-    inner: ~ChanOneHack<T>
-}
-
-
-/// A one-shot port.
-pub struct PortOne<T> {
-    // XXX: Hack extra allocation to make by-val self work
-    inner: ~PortOneHack<T>
-}
-
-pub struct ChanOneHack<T> {
     void_packet: *mut Void,
     suppress_finalize: bool
 }
 
-pub struct PortOneHack<T> {
+/// A one-shot port.
+pub struct PortOne<T> {
     void_packet: *mut Void,
     suppress_finalize: bool
 }
@@ -75,22 +64,26 @@ pub fn oneshot<T: Send>() -> (PortOne<T>, ChanOne<T>) {
     unsafe {
         let packet: *mut Void = cast::transmute(packet);
         let port = PortOne {
-            inner: ~PortOneHack {
-                void_packet: packet,
-                suppress_finalize: false
-            }
+            void_packet: packet,
+            suppress_finalize: false
         };
         let chan = ChanOne {
-            inner: ~ChanOneHack {
-                void_packet: packet,
-                suppress_finalize: false
-            }
+            void_packet: packet,
+            suppress_finalize: false
         };
         return (port, chan);
     }
 }
 
 impl<T> ChanOne<T> {
+    #[inline]
+    fn packet(&self) -> *mut Packet<T> {
+        unsafe {
+            let p: *mut ~Packet<T> = cast::transmute(&self.void_packet);
+            let p: *mut Packet<T> = &mut **p;
+            return p;
+        }
+    }
 
     pub fn send(self, val: T) {
         self.try_send(val);
@@ -99,7 +92,7 @@ impl<T> ChanOne<T> {
     pub fn try_send(self, val: T) -> bool {
         let mut this = self;
         let mut recvr_active = true;
-        let packet = this.inner.packet();
+        let packet = this.packet();
 
         unsafe {
 
@@ -127,7 +120,7 @@ impl<T> ChanOne<T> {
                         sched.metrics.rendezvous_sends += 1;
                     }
                     // Port has closed. Need to clean up.
-                    let _packet: ~Packet<T> = cast::transmute(this.inner.void_packet);
+                    let _packet: ~Packet<T> = cast::transmute(this.void_packet);
                     recvr_active = false;
                 }
                 task_as_state => {
@@ -144,13 +137,20 @@ impl<T> ChanOne<T> {
         }
 
         // Suppress the synchronizing actions in the finalizer. We're done with the packet.
-        this.inner.suppress_finalize = true;
+        this.suppress_finalize = true;
         return recvr_active;
     }
 }
 
-
 impl<T> PortOne<T> {
+    fn packet(&self) -> *mut Packet<T> {
+        unsafe {
+            let p: *mut ~Packet<T> = cast::transmute(&self.void_packet);
+            let p: *mut Packet<T> = &mut **p;
+            return p;
+        }
+    }
+
     pub fn recv(self) -> T {
         match self.try_recv() {
             Some(val) => val,
@@ -162,43 +162,129 @@ impl<T> PortOne<T> {
 
     pub fn try_recv(self) -> Option<T> {
         let mut this = self;
-        let packet = this.inner.packet();
-
-        // XXX: Optimize this to not require the two context switches when data is available
-
-        // Switch to the scheduler to put the ~Task into the Packet state.
-        let sched = Local::take::<Scheduler>();
-        do sched.deschedule_running_task_and_then |sched, task| {
-            unsafe {
-                // Atomically swap the task pointer into the Packet state, issuing
-                // an acquire barrier to prevent reordering of the subsequent read
-                // of the payload. Also issues a release barrier to prevent reordering
-                // of any previous writes to the task structure.
-                let task_as_state = task.cast_to_uint();
-                let oldstate = (*packet).state.swap(task_as_state, SeqCst);
-                match oldstate {
-                    STATE_BOTH => {
-                        // Data has not been sent. Now we're blocked.
-                        rtdebug!("non-rendezvous recv");
-                        sched.metrics.non_rendezvous_recvs += 1;
-                    }
-                    STATE_ONE => {
-                        rtdebug!("rendezvous recv");
-                        sched.metrics.rendezvous_recvs += 1;
-
-                        // Channel is closed. Switch back and check the data.
-                        // NB: We have to drop back into the scheduler event loop here
-                        // instead of switching immediately back or we could end up
-                        // triggering infinite recursion on the scheduler's stack.
-                        let recvr = BlockedTask::cast_from_uint(task_as_state);
-                        sched.enqueue_blocked_task(recvr);
+
+        // Optimistic check. If data was sent already, we don't even need to block.
+        // No release barrier needed here; we're not handing off our task pointer yet.
+        if !this.optimistic_check() {
+            // No data available yet.
+            // Switch to the scheduler to put the ~Task into the Packet state.
+            let sched = Local::take::<Scheduler>();
+            do sched.deschedule_running_task_and_then |sched, task| {
+                this.block_on(sched, task);
+            }
+        }
+
+        // Task resumes.
+        this.recv_ready()
+    }
+}
+
+impl<T> Select for PortOne<T> {
+    #[inline] #[cfg(not(test))]
+    fn optimistic_check(&mut self) -> bool {
+        unsafe { (*self.packet()).state.load(Acquire) == STATE_ONE }
+    }
+
+    #[inline] #[cfg(test)]
+    fn optimistic_check(&mut self) -> bool {
+        // The optimistic check is never necessary for correctness. For testing
+        // purposes, making it randomly return false simulates a racing sender.
+        use rand::{Rand, rng};
+        let mut rng = rng();
+        let actually_check = Rand::rand(&mut rng);
+        if actually_check {
+            unsafe { (*self.packet()).state.load(Acquire) == STATE_ONE }
+        } else {
+            false
+        }
+    }
+
+    fn block_on(&mut self, sched: &mut Scheduler, task: BlockedTask) -> bool {
+        unsafe {
+            // Atomically swap the task pointer into the Packet state, issuing
+            // an acquire barrier to prevent reordering of the subsequent read
+            // of the payload. Also issues a release barrier to prevent
+            // reordering of any previous writes to the task structure.
+            let task_as_state = task.cast_to_uint();
+            let oldstate = (*self.packet()).state.swap(task_as_state, SeqCst);
+            match oldstate {
+                STATE_BOTH => {
+                    // Data has not been sent. Now we're blocked.
+                    rtdebug!("non-rendezvous recv");
+                    sched.metrics.non_rendezvous_recvs += 1;
+                    false
+                }
+                STATE_ONE => {
+                    // Re-record that we are the only owner of the packet.
+                    // Release barrier needed in case the task gets reawoken
+                    // on a different core (this is analogous to writing a
+                    // payload; a barrier in enqueueing the task protects it).
+                    // NB(#8132). This *must* occur before the enqueue below.
+                    // FIXME(#6842, #8130) This is usually only needed for the
+                    // assertion in recv_ready, except in the case of select().
+                    // This won't actually ever have cacheline contention, but
+                    // maybe should be optimized out with a cfg(test) anyway?
+                    (*self.packet()).state.store(STATE_ONE, Release);
+
+                    rtdebug!("rendezvous recv");
+                    sched.metrics.rendezvous_recvs += 1;
+
+                    // Channel is closed. Switch back and check the data.
+                    // NB: We have to drop back into the scheduler event loop here
+                    // instead of switching immediately back or we could end up
+                    // triggering infinite recursion on the scheduler's stack.
+                    let recvr = BlockedTask::cast_from_uint(task_as_state);
+                    sched.enqueue_blocked_task(recvr);
+                    true
+                }
+                _ => rtabort!("can't block_on; a task is already blocked")
+            }
+        }
+    }
+
+    // This is the only select trait function that's not also used in recv.
+    fn unblock_from(&mut self) -> bool {
+        let packet = self.packet();
+        unsafe {
+            // In case the data is available, the acquire barrier here matches
+            // the release barrier the sender used to release the payload.
+            match (*packet).state.load(Acquire) {
+                // Impossible. We removed STATE_BOTH when blocking on it, and
+                // no self-respecting sender would put it back.
+                STATE_BOTH    => rtabort!("refcount already 2 in unblock_from"),
+                // Here, a sender already tried to wake us up. Perhaps they
+                // even succeeded! Data is available.
+                STATE_ONE     => true,
+                // Still registered as blocked. Need to "unblock" the pointer.
+                task_as_state => {
+                    // In the window between the load and the CAS, a sender
+                    // might take the pointer and set the refcount to ONE. If
+                    // that happens, we shouldn't clobber that with BOTH!
+                    // Acquire barrier again for the same reason as above.
+                    match (*packet).state.compare_and_swap(task_as_state, STATE_BOTH,
+                                                           Acquire) {
+                        STATE_BOTH => rtabort!("refcount became 2 in unblock_from"),
+                        STATE_ONE  => true, // Lost the race. Data available.
+                        same_ptr   => {
+                            // We successfully unblocked our task pointer.
+                            assert!(task_as_state == same_ptr);
+                            let handle = BlockedTask::cast_from_uint(task_as_state);
+                            // Because we are already awake, the handle we
+                            // gave to this port shall already be empty.
+                            handle.assert_already_awake();
+                            false
+                        }
                     }
-                    _ => util::unreachable()
                 }
             }
         }
+    }
+}
 
-        // Task resumes.
+impl<T> SelectPort<T> for PortOne<T> {
+    fn recv_ready(self) -> Option<T> {
+        let mut this = self;
+        let packet = this.packet();
 
         // No further memory barrier is needed here to access the
         // payload. Some scenarios:
@@ -210,14 +296,17 @@ impl<T> PortOne<T> {
         // 3) We encountered STATE_BOTH above and blocked, but the receiving task (this task)
         //    is pinned to some other scheduler, so the sending task had to give us to
         //    a different scheduler for resuming. That send synchronized memory.
-
         unsafe {
-            let payload = util::replace(&mut (*packet).payload, None);
+            // See corresponding store() above in block_on for rationale.
+            // FIXME(#8130) This can happen only in test builds.
+            assert!((*packet).state.load(Acquire) == STATE_ONE);
+
+            let payload = (*packet).payload.take();
 
             // The sender has closed up shop. Drop the packet.
-            let _packet: ~Packet<T> = cast::transmute(this.inner.void_packet);
+            let _packet: ~Packet<T> = cast::transmute(this.void_packet);
             // Suppress the synchronizing actions in the finalizer. We're done with the packet.
-            this.inner.suppress_finalize = true;
+            this.suppress_finalize = true;
             return payload;
         }
     }
@@ -226,19 +315,19 @@ impl<T> PortOne<T> {
 impl<T> Peekable<T> for PortOne<T> {
     fn peek(&self) -> bool {
         unsafe {
-            let packet: *mut Packet<T> = self.inner.packet();
+            let packet: *mut Packet<T> = self.packet();
             let oldstate = (*packet).state.load(SeqCst);
             match oldstate {
                 STATE_BOTH => false,
                 STATE_ONE => (*packet).payload.is_some(),
-                _ => util::unreachable()
+                _ => rtabort!("peeked on a blocked task")
             }
         }
     }
 }
 
 #[unsafe_destructor]
-impl<T> Drop for ChanOneHack<T> {
+impl<T> Drop for ChanOne<T> {
     fn drop(&self) {
         if self.suppress_finalize { return }
 
@@ -267,7 +356,7 @@ impl<T> Drop for ChanOneHack<T> {
 }
 
 #[unsafe_destructor]
-impl<T> Drop for PortOneHack<T> {
+impl<T> Drop for PortOne<T> {
     fn drop(&self) {
         if self.suppress_finalize { return }
 
@@ -295,26 +384,6 @@ impl<T> Drop for PortOneHack<T> {
     }
 }
 
-impl<T> ChanOneHack<T> {
-    fn packet(&self) -> *mut Packet<T> {
-        unsafe {
-            let p: *mut ~Packet<T> = cast::transmute(&self.void_packet);
-            let p: *mut Packet<T> = &mut **p;
-            return p;
-        }
-    }
-}
-
-impl<T> PortOneHack<T> {
-    fn packet(&self) -> *mut Packet<T> {
-        unsafe {
-            let p: *mut ~Packet<T> = cast::transmute(&self.void_packet);
-            let p: *mut Packet<T> = &mut **p;
-            return p;
-        }
-    }
-}
-
 struct StreamPayload<T> {
     val: T,
     next: PortOne<StreamPayload<T>>
@@ -385,6 +454,36 @@ impl<T> Peekable<T> for Port<T> {
     }
 }
 
+impl<T> Select for Port<T> {
+    #[inline]
+    fn optimistic_check(&mut self) -> bool {
+        do self.next.with_mut_ref |pone| { pone.optimistic_check() }
+    }
+
+    #[inline]
+    fn block_on(&mut self, sched: &mut Scheduler, task: BlockedTask) -> bool {
+        let task = Cell::new(task);
+        do self.next.with_mut_ref |pone| { pone.block_on(sched, task.take()) }
+    }
+
+    #[inline]
+    fn unblock_from(&mut self) -> bool {
+        do self.next.with_mut_ref |pone| { pone.unblock_from() }
+    }
+}
+
+impl<T> SelectPort<(T, Port<T>)> for Port<T> {
+    fn recv_ready(self) -> Option<(T, Port<T>)> {
+        match self.next.take().recv_ready() {
+            Some(StreamPayload { val, next }) => {
+                self.next.put_back(next);
+                Some((val, self))
+            }
+            None => None
+        }
+    }
+}
+
 pub struct SharedChan<T> {
     // Just like Chan, but a shared AtomicOption instead of Cell
     priv next: UnsafeAtomicRcBox<AtomicOption<StreamChanOne<T>>>
diff --git a/src/libstd/rt/kill.rs b/src/libstd/rt/kill.rs
index 2bf4543df50..e691bf51ea5 100644
--- a/src/libstd/rt/kill.rs
+++ b/src/libstd/rt/kill.rs
@@ -106,8 +106,14 @@ impl Drop for KillFlag {
 // blocked task handle. So unblocking a task must restore that spare.
 unsafe fn revive_task_ptr(task_ptr: uint, spare_flag: Option<KillFlagHandle>) -> ~Task {
     let mut task: ~Task = cast::transmute(task_ptr);
-    rtassert!(task.death.spare_kill_flag.is_none());
-    task.death.spare_kill_flag = spare_flag;
+    if task.death.spare_kill_flag.is_none() {
+        task.death.spare_kill_flag = spare_flag;
+    } else {
+        // A task's spare kill flag is not used for blocking in one case:
+        // when an unkillable task blocks on select. In this case, a separate
+        // one was created, which we now discard.
+        rtassert!(task.death.unkillable > 0);
+    }
     task
 }
 
@@ -119,7 +125,7 @@ impl BlockedTask {
             Killable(flag_arc) => {
                 let flag = unsafe { &mut **flag_arc.get() };
                 match flag.swap(KILL_RUNNING, SeqCst) {
-                    KILL_RUNNING => rtabort!("tried to wake an already-running task"),
+                    KILL_RUNNING => None, // woken from select(), perhaps
                     KILL_KILLED  => None, // a killer stole it already
                     task_ptr     =>
                         Some(unsafe { revive_task_ptr(task_ptr, Some(flag_arc)) })
@@ -162,6 +168,27 @@ impl BlockedTask {
         }
     }
 
+    /// Converts one blocked task handle to a list of many handles to the same.
+    pub fn make_selectable(self, num_handles: uint) -> ~[BlockedTask] {
+        let handles = match self {
+            Unkillable(task) => {
+                let flag = unsafe { KillFlag(AtomicUint::new(cast::transmute(task))) };
+                UnsafeAtomicRcBox::newN(flag, num_handles)
+            }
+            Killable(flag_arc) => flag_arc.cloneN(num_handles),
+        };
+        // Even if the task was unkillable before, we use 'Killable' because
+        // multiple pipes will have handles. It does not really mean killable.
+        handles.consume_iter().transform(|x| Killable(x)).collect()
+    }
+
+    // This assertion has two flavours because the wake involves an atomic op.
+    // In the faster version, destructors will fail dramatically instead.
+    #[inline] #[cfg(not(test))]
+    pub fn assert_already_awake(self) { }
+    #[inline] #[cfg(test)]
+    pub fn assert_already_awake(self) { assert!(self.wake().is_none()); }
+
     /// Convert to an unsafe uint value. Useful for storing in a pipe's state flag.
     #[inline]
     pub unsafe fn cast_to_uint(self) -> uint {
@@ -301,7 +328,7 @@ impl KillHandle {
         }
 
         // Try to see if all our children are gone already.
-        match unsafe { self.try_unwrap() } {
+        match self.try_unwrap() {
             // Couldn't unwrap; children still alive. Reparent entire handle as
             // our own tombstone, to be unwrapped later.
             Left(this) => {
@@ -313,7 +340,7 @@ impl KillHandle {
                         // Prefer to check tombstones that were there first,
                         // being "more fair" at the expense of tail-recursion.
                         others.take().map_consume_default(true, |f| f()) && {
-                            let mut inner = unsafe { this.take().unwrap() };
+                            let mut inner = this.take().unwrap();
                             (!inner.any_child_failed) &&
                                 inner.child_tombstones.take_map_default(true, |f| f())
                         }
@@ -402,7 +429,7 @@ impl Death {
         do self.on_exit.take_map |on_exit| {
             if success {
                 // We succeeded, but our children might not. Need to wait for them.
-                let mut inner = unsafe { self.kill_handle.take_unwrap().unwrap() };
+                let mut inner = self.kill_handle.take_unwrap().unwrap();
                 if inner.any_child_failed {
                     success = false;
                 } else {
@@ -528,7 +555,7 @@ mod test {
 
             // Without another handle to child, the try unwrap should succeed.
             child.reparent_children_to(&mut parent);
-            let mut parent_inner = unsafe { parent.unwrap() };
+            let mut parent_inner = parent.unwrap();
             assert!(parent_inner.child_tombstones.is_none());
             assert!(parent_inner.any_child_failed == false);
         }
@@ -543,7 +570,7 @@ mod test {
             child.notify_immediate_failure();
             // Without another handle to child, the try unwrap should succeed.
             child.reparent_children_to(&mut parent);
-            let mut parent_inner = unsafe { parent.unwrap() };
+            let mut parent_inner = parent.unwrap();
             assert!(parent_inner.child_tombstones.is_none());
             // Immediate failure should have been propagated.
             assert!(parent_inner.any_child_failed);
@@ -565,7 +592,7 @@ mod test {
             // Otherwise, due to 'link', it would try to tombstone.
             child2.reparent_children_to(&mut parent);
             // Should successfully unwrap even though 'link' is still alive.
-            let mut parent_inner = unsafe { parent.unwrap() };
+            let mut parent_inner = parent.unwrap();
             assert!(parent_inner.child_tombstones.is_none());
             // Immediate failure should have been propagated by first child.
             assert!(parent_inner.any_child_failed);
@@ -584,7 +611,7 @@ mod test {
             // Let parent collect tombstones.
             util::ignore(link);
             // Must have created a tombstone
-            let mut parent_inner = unsafe { parent.unwrap() };
+            let mut parent_inner = parent.unwrap();
             assert!(parent_inner.child_tombstones.take_unwrap()());
             assert!(parent_inner.any_child_failed == false);
         }
@@ -603,7 +630,7 @@ mod test {
             // Let parent collect tombstones.
             util::ignore(link);
             // Must have created a tombstone
-            let mut parent_inner = unsafe { parent.unwrap() };
+            let mut parent_inner = parent.unwrap();
             // Failure must be seen in the tombstone.
             assert!(parent_inner.child_tombstones.take_unwrap()() == false);
             assert!(parent_inner.any_child_failed == false);
@@ -623,7 +650,7 @@ mod test {
             // Let parent collect tombstones.
             util::ignore(link);
             // Must have created a tombstone
-            let mut parent_inner = unsafe { parent.unwrap() };
+            let mut parent_inner = parent.unwrap();
             assert!(parent_inner.child_tombstones.take_unwrap()());
             assert!(parent_inner.any_child_failed == false);
         }
@@ -644,7 +671,7 @@ mod test {
             // Let parent collect tombstones.
             util::ignore(link);
             // Must have created a tombstone
-            let mut parent_inner = unsafe { parent.unwrap() };
+            let mut parent_inner = parent.unwrap();
             // Failure must be seen in the tombstone.
             assert!(parent_inner.child_tombstones.take_unwrap()() == false);
             assert!(parent_inner.any_child_failed == false);
diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs
index 808d07ce77d..2ca7d01da49 100644
--- a/src/libstd/rt/mod.rs
+++ b/src/libstd/rt/mod.rs
@@ -142,6 +142,9 @@ pub mod tube;
 /// Simple reimplementation of core::comm
 pub mod comm;
 
+/// Routines for select()ing on pipes.
+pub mod select;
+
 // FIXME #5248 shouldn't be pub
 /// The runtime needs to be able to put a pointer into thread-local storage.
 pub mod local_ptr;
diff --git a/src/libstd/rt/select.rs b/src/libstd/rt/select.rs
new file mode 100644
index 00000000000..bc9e265c8d9
--- /dev/null
+++ b/src/libstd/rt/select.rs
@@ -0,0 +1,328 @@
+// Copyright 2013 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use option::*;
+// use either::{Either, Left, Right};
+use rt::kill::BlockedTask;
+use rt::sched::Scheduler;
+use rt::local::Local;
+
+/// Trait for message-passing primitives that can be select()ed on.
+pub trait Select {
+    // Returns true if data was available.
+    fn optimistic_check(&mut self) -> bool;
+    // Returns true if data was available. If so, shall also wake() the task.
+    fn block_on(&mut self, &mut Scheduler, BlockedTask) -> bool;
+    // Returns true if data was available.
+    fn unblock_from(&mut self) -> bool;
+}
+
+/// Trait for message-passing primitives that can use the select2() convenience wrapper.
+// (This is separate from the above trait to enable heterogeneous lists of ports
+// that implement Select on different types to use select().)
+pub trait SelectPort<T> : Select {
+    fn recv_ready(self) -> Option<T>;
+}
+
+/// Receive a message from any one of many ports at once.
+pub fn select<A: Select>(ports: &mut [A]) -> uint {
+    if ports.is_empty() {
+        fail!("can't select on an empty list");
+    }
+
+    for ports.mut_iter().enumerate().advance |(index, port)| {
+        if port.optimistic_check() {
+            return index;
+        }
+    }
+
+    // If one of the ports already contains data when we go to block on it, we
+    // don't bother enqueueing on the rest of them, so we shouldn't bother
+    // unblocking from it either. This is just for efficiency, not correctness.
+    // (If not, we need to unblock from all of them. Length is a placeholder.)
+    let mut ready_index = ports.len();
+
+    let sched = Local::take::<Scheduler>();
+    do sched.deschedule_running_task_and_then |sched, task| {
+        let task_handles = task.make_selectable(ports.len());
+
+        for ports.mut_iter().zip(task_handles.consume_iter()).enumerate().advance
+                |(index, (port, task_handle))| {
+            // If one of the ports has data by now, it will wake the handle.
+            if port.block_on(sched, task_handle) {
+                ready_index = index;
+                break;
+            }
+        }
+    }
+
+    // Task resumes. Now unblock ourselves from all the ports we blocked on.
+    // If the success index wasn't reset, 'take' will just take all of them.
+    // Iterate in reverse so the 'earliest' index that's ready gets returned.
+    for ports.mut_slice(0, ready_index).mut_rev_iter().enumerate().advance |(index, port)| {
+        if port.unblock_from() {
+            ready_index = index;
+        }
+    }
+
+    assert!(ready_index < ports.len());
+    return ready_index;
+}
+
+/* FIXME(#5121, #7914) This all should be legal, but rust is not clever enough yet.
+
+impl <'self> Select for &'self mut Select {
+    fn optimistic_check(&mut self) -> bool { self.optimistic_check() }
+    fn block_on(&mut self, sched: &mut Scheduler, task: BlockedTask) -> bool {
+        self.block_on(sched, task)
+    }
+    fn unblock_from(&mut self) -> bool { self.unblock_from() }
+}
+
+pub fn select2<TA, A: SelectPort<TA>, TB, B: SelectPort<TB>>(mut a: A, mut b: B)
+        -> Either<(Option<TA>, B), (A, Option<TB>)> {
+    let result = {
+        let mut ports = [&mut a as &mut Select, &mut b as &mut Select];
+        select(ports)
+    };
+    match result {
+        0 => Left ((a.recv_ready(), b)),
+        1 => Right((a, b.recv_ready())),
+        x => fail!("impossible case in select2: %?", x)
+    }
+}
+
+*/
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use option::*;
+    use rt::comm::*;
+    use rt::test::*;
+    use vec::*;
+    use comm::GenericChan;
+    use task;
+    use cell::Cell;
+
+    #[test] #[ignore(cfg(windows))] #[should_fail]
+    fn select_doesnt_get_trolled() {
+        select::<PortOne<()>>([]);
+    }
+
+    /* non-blocking select tests */
+
+    #[cfg(test)]
+    fn select_helper(num_ports: uint, send_on_chans: &[uint]) {
+        // Unfortunately this does not actually test the block_on early-break
+        // codepath in select -- racing between the sender and the receiver in
+        // separate tasks is necessary to get around the optimistic check.
+        let (ports, chans) = unzip(from_fn(num_ports, |_| oneshot::<()>()));
+        let mut dead_chans = ~[];
+        let mut ports = ports;
+        for chans.consume_iter().enumerate().advance |(i, chan)| {
+            if send_on_chans.contains(&i) {
+                chan.send(());
+            } else {
+                dead_chans.push(chan);
+            }
+        }
+        let ready_index = select(ports);
+        assert!(send_on_chans.contains(&ready_index));
+        assert!(ports.swap_remove(ready_index).recv_ready().is_some());
+        let _ = dead_chans;
+
+        // Same thing with streams instead.
+        // FIXME(#7971): This should be in a macro but borrowck isn't smart enough.
+        let (ports, chans) = unzip(from_fn(num_ports, |_| stream::<()>()));
+        let mut dead_chans = ~[];
+        let mut ports = ports;
+        for chans.consume_iter().enumerate().advance |(i, chan)| {
+            if send_on_chans.contains(&i) {
+                chan.send(());
+            } else {
+                dead_chans.push(chan);
+            }
+        }
+        let ready_index = select(ports);
+        assert!(send_on_chans.contains(&ready_index));
+        assert!(ports.swap_remove(ready_index).recv_ready().is_some());
+        let _ = dead_chans;
+    }
+
+    #[test]
+    fn select_one() {
+        do run_in_newsched_task { select_helper(1, [0]) }
+    }
+
+    #[test]
+    fn select_two() {
+        // NB. I would like to have a test that tests the first one that is
+        // ready is the one that's returned, but that can't be reliably tested
+        // with the randomized behaviour of optimistic_check.
+        do run_in_newsched_task { select_helper(2, [1]) }
+        do run_in_newsched_task { select_helper(2, [0]) }
+        do run_in_newsched_task { select_helper(2, [1,0]) }
+    }
+
+    #[test]
+    fn select_a_lot() {
+        do run_in_newsched_task { select_helper(12, [7,8,9]) }
+    }
+
+    #[test]
+    fn select_stream() {
+        use util;
+        use comm::GenericChan;
+
+        // Sends 10 buffered packets, and uses select to retrieve them all.
+        // Puts the port in a different spot in the vector each time.
+        do run_in_newsched_task {
+            let (ports, _) = unzip(from_fn(10, |_| stream()));
+            let (port, chan) = stream();
+            for 10.times { chan.send(31337); }
+            let mut ports = ports;
+            let mut port = Some(port);
+            let order = [5u,0,4,3,2,6,9,8,7,1];
+            for order.iter().advance |&index| {
+                // put the port in the vector at any index
+                util::swap(port.get_mut_ref(), &mut ports[index]);
+                assert!(select(ports) == index);
+                // get it back out
+                util::swap(port.get_mut_ref(), &mut ports[index]);
+                // NB. Not recv(), because optimistic_check randomly fails.
+                let (data, new_port) = port.take_unwrap().recv_ready().unwrap();
+                assert!(data == 31337);
+                port = Some(new_port);
+            }
+        }
+    }
+
+    #[test]
+    fn select_unkillable() {
+        do run_in_newsched_task {
+            do task::unkillable { select_helper(2, [1]) }
+        }
+    }
+
+    /* blocking select tests */
+
+    #[test]
+    fn select_blocking() {
+        select_blocking_helper(true);
+        select_blocking_helper(false);
+
+        fn select_blocking_helper(killable: bool) {
+            do run_in_newsched_task {
+                let (p1,_c) = oneshot();
+                let (p2,c2) = oneshot();
+                let mut ports = [p1,p2];
+
+                let (p3,c3) = oneshot();
+                let (p4,c4) = oneshot();
+
+                let x = Cell::new((c2, p3, c4));
+                do task::spawn {
+                    let (c2, p3, c4) = x.take();
+                    p3.recv();   // handshake parent
+                    c4.send(()); // normal receive
+                    task::yield();
+                    c2.send(()); // select receive
+                }
+
+                // Try to block before child sends on c2.
+                c3.send(());
+                p4.recv();
+                if killable {
+                    assert!(select(ports) == 1);
+                } else {
+                    do task::unkillable { assert!(select(ports) == 1); }
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn select_racing_senders() {
+        static NUM_CHANS: uint = 10;
+
+        select_racing_senders_helper(true,  ~[0,1,2,3,4,5,6,7,8,9]);
+        select_racing_senders_helper(false, ~[0,1,2,3,4,5,6,7,8,9]);
+        select_racing_senders_helper(true,  ~[0,1,2]);
+        select_racing_senders_helper(false, ~[0,1,2]);
+        select_racing_senders_helper(true,  ~[3,4,5,6]);
+        select_racing_senders_helper(false, ~[3,4,5,6]);
+        select_racing_senders_helper(true,  ~[7,8,9]);
+        select_racing_senders_helper(false, ~[7,8,9]);
+
+        fn select_racing_senders_helper(killable: bool, send_on_chans: ~[uint]) {
+            use uint;
+            use rt::test::spawntask_random;
+
+            do run_in_newsched_task {
+                // A bit of stress, since ordinarily this is just smoke and mirrors.
+                for 4.times {
+                    let send_on_chans = send_on_chans.clone();
+                    do task::spawn {
+                        let mut ports = ~[];
+                        for uint::range(0, NUM_CHANS) |i| {
+                            let (p,c) = oneshot();
+                            ports.push(p);
+                            if send_on_chans.contains(&i) {
+                                let c = Cell::new(c);
+                                do spawntask_random {
+                                    task::yield();
+                                    c.take().send(());
+                                }
+                            }
+                        }
+                        // nondeterministic result, but should succeed
+                        if killable {
+                            select(ports);
+                        } else {
+                            do task::unkillable { select(ports); }
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    #[test] #[ignore(cfg(windows))]
+    fn select_killed() {
+        do run_in_newsched_task {
+            let (success_p, success_c) = oneshot::<bool>();
+            let success_c = Cell::new(success_c);
+            do task::try {
+                let success_c = Cell::new(success_c.take());
+                do task::unkillable {
+                    let (p,c) = oneshot();
+                    let c = Cell::new(c);
+                    do task::spawn {
+                        let (dead_ps, dead_cs) = unzip(from_fn(5, |_| oneshot::<()>()));
+                        let mut ports = dead_ps;
+                        select(ports); // should get killed; nothing should leak
+                        c.take().send(()); // must not happen
+                        // Make sure dead_cs doesn't get closed until after select.
+                        let _ = dead_cs;
+                    }
+                    do task::spawn {
+                        fail!(); // should kill sibling awake
+                    }
+
+                    // wait for killed selector to close (NOT send on) its c.
+                    // hope to send 'true'.
+                    success_c.take().send(p.try_recv().is_none());
+                }
+            };
+            assert!(success_p.recv());
+        }
+    }
+}