Rewrite channels yet again for upgradeability

This, the Nth rewrite of channels, is not a rewrite of the core logic behind
channels, but rather their API usage. In the past, we had the distinction
between oneshot, stream, and shared channels, but the most recent rewrite
dropped oneshots in favor of streams and shared channels.

This distinction of stream vs shared has shown that it's not quite what we'd
like either, and this moves the `std::comm` module in the direction of "one
channel to rule them all". There now remains only one Chan and one Port.

This new channel is actually a hybrid oneshot/stream/shared channel under the
hood in order to optimize for the use cases in question. Additionally, this also
reduces the cognitive burden of having to choose between a Chan or a SharedChan
in an API.

My simple benchmarks show no reduction in efficiency over the existing channels
today, and a 3x improvement in the oneshot case. I sadly don't have a
pre-last-rewrite compiler to test out the old old oneshots, but I would imagine
that the performance is comparable, but slightly slower (due to atomic reference
counting).

This commit also brings the bonus bugfix to channels that the pending queue of
messages are all dropped when a Port disappears rather then when both the Port
and the Chan disappear.

src/doc/guide-tasks.md

@@ -232,7 +232,7 @@ Instead we can use a `SharedChan`, a type that allows a single
 ~~~
 # use std::task::spawn;
 
-let (port, chan) = SharedChan::new();
+let (port, chan) = Chan::new();
 
 for init_val in range(0u, 3) {
     // Create a new channel handle to distribute to the child task

src/libextra/test.rs

@@ -767,7 +767,7 @@ fn run_tests(opts: &TestOpts,
     remaining.reverse();
     let mut pending = 0;
 
-    let (p, ch) = SharedChan::new();
+    let (p, ch) = Chan::new();
 
     while pending > 0 || !remaining.is_empty() {
         while pending < concurrency && !remaining.is_empty() {
@@ -878,7 +878,7 @@ pub fn filter_tests(
 
 pub fn run_test(force_ignore: bool,
                 test: TestDescAndFn,
-                monitor_ch: SharedChan<MonitorMsg>) {
+                monitor_ch: Chan<MonitorMsg>) {
 
     let TestDescAndFn {desc, testfn} = test;
 
@@ -888,7 +888,7 @@ pub fn run_test(force_ignore: bool,
     }
 
     fn run_test_inner(desc: TestDesc,
-                      monitor_ch: SharedChan<MonitorMsg>,
+                      monitor_ch: Chan<MonitorMsg>,
                       testfn: proc()) {
         spawn(proc() {
             let mut task = task::task();
@@ -1260,7 +1260,7 @@ mod tests {
             },
             testfn: DynTestFn(proc() f()),
         };
-        let (p, ch) = SharedChan::new();
+        let (p, ch) = Chan::new();
         run_test(false, desc, ch);
         let (_, res) = p.recv();
         assert!(res != TrOk);
@@ -1277,7 +1277,7 @@ mod tests {
             },
             testfn: DynTestFn(proc() f()),
         };
-        let (p, ch) = SharedChan::new();
+        let (p, ch) = Chan::new();
         run_test(false, desc, ch);
         let (_, res) = p.recv();
         assert_eq!(res, TrIgnored);
@@ -1294,7 +1294,7 @@ mod tests {
             },
             testfn: DynTestFn(proc() f()),
         };
-        let (p, ch) = SharedChan::new();
+        let (p, ch) = Chan::new();
         run_test(false, desc, ch);
         let (_, res) = p.recv();
         assert_eq!(res, TrOk);
@@ -1311,7 +1311,7 @@ mod tests {
             },
             testfn: DynTestFn(proc() f()),
         };
-        let (p, ch) = SharedChan::new();
+        let (p, ch) = Chan::new();
         run_test(false, desc, ch);
         let (_, res) = p.recv();
         assert_eq!(res, TrFailed);

src/libgreen/lib.rs

@@ -315,7 +315,7 @@ pub struct SchedPool {
 #[deriving(Clone)]
 struct TaskState {
     cnt: UnsafeArc<AtomicUint>,
-    done: SharedChan<()>,
+    done: Chan<()>,
 }
 
 impl SchedPool {
@@ -469,7 +469,7 @@ impl SchedPool {
 
 impl TaskState {
     fn new() -> (Port<()>, TaskState) {
-        let (p, c) = SharedChan::new();
+        let (p, c) = Chan::new();
         (p, TaskState {
             cnt: UnsafeArc::new(AtomicUint::new(0)),
             done: c,

src/libnative/io/mod.rs

@@ -22,7 +22,6 @@
 //! that you would find on the respective platform.
 
 use std::c_str::CString;
-use std::comm::SharedChan;
 use std::io;
 use std::io::IoError;
 use std::io::net::ip::SocketAddr;
@@ -289,7 +288,7 @@ impl rtio::IoFactory for IoFactory {
             })
         }
     }
-    fn signal(&mut self, _signal: Signum, _channel: SharedChan<Signum>)
+    fn signal(&mut self, _signal: Signum, _channel: Chan<Signum>)
         -> IoResult<~RtioSignal> {
         Err(unimpl())
     }

src/libnative/io/timer_helper.rs

@@ -33,7 +33,7 @@ use task;
 // only torn down after everything else has exited. This means that these
 // variables are read-only during use (after initialization) and both of which
 // are safe to use concurrently.
-static mut HELPER_CHAN: *mut SharedChan<Req> = 0 as *mut SharedChan<Req>;
+static mut HELPER_CHAN: *mut Chan<Req> = 0 as *mut Chan<Req>;
 static mut HELPER_SIGNAL: imp::signal = 0 as imp::signal;
 
 pub fn boot(helper: fn(imp::signal, Port<Req>)) {
@@ -43,7 +43,9 @@ pub fn boot(helper: fn(imp::signal, Port<Req>)) {
     unsafe {
         LOCK.lock();
         if !INITIALIZED {
-            let (msgp, msgc) = SharedChan::new();
+            let (msgp, msgc) = Chan::new();
+            // promote this to a shared channel
+            drop(msgc.clone());
             HELPER_CHAN = cast::transmute(~msgc);
             let (receive, send) = imp::new();
             HELPER_SIGNAL = send;
@@ -84,8 +86,8 @@ fn shutdown() {
     // Clean up after ther helper thread
     unsafe {
         imp::close(HELPER_SIGNAL);
-        let _chan: ~SharedChan<Req> = cast::transmute(HELPER_CHAN);
-        HELPER_CHAN = 0 as *mut SharedChan<Req>;
+        let _chan: ~Chan<Req> = cast::transmute(HELPER_CHAN);
+        HELPER_CHAN = 0 as *mut Chan<Req>;
         HELPER_SIGNAL = 0 as imp::signal;
     }
 }

src/librustuv/signal.rs

@@ -10,7 +10,6 @@
 
 use std::libc::c_int;
 use std::io::signal::Signum;
-use std::comm::SharedChan;
 use std::rt::rtio::RtioSignal;
 
 use homing::{HomingIO, HomeHandle};
@@ -22,13 +21,13 @@ pub struct SignalWatcher {
     handle: *uvll::uv_signal_t,
     home: HomeHandle,
 
-    channel: SharedChan<Signum>,
+    channel: Chan<Signum>,
     signal: Signum,
 }
 
 impl SignalWatcher {
     pub fn new(io: &mut UvIoFactory, signum: Signum,
-               channel: SharedChan<Signum>) -> Result<~SignalWatcher, UvError> {
+               channel: Chan<Signum>) -> Result<~SignalWatcher, UvError> {
         let s = ~SignalWatcher {
             handle: UvHandle::alloc(None::<SignalWatcher>, uvll::UV_SIGNAL),
             home: io.make_handle(),
@@ -81,7 +80,7 @@ mod test {
     #[test]
     fn closing_channel_during_drop_doesnt_kill_everything() {
         // see issue #10375, relates to timers as well.
-        let (port, chan) = SharedChan::new();
+        let (port, chan) = Chan::new();
         let _signal = SignalWatcher::new(local_loop(), signal::Interrupt,
                                          chan);
 

src/librustuv/uvio.rs

@@ -10,7 +10,6 @@
 
 use std::c_str::CString;
 use std::cast;
-use std::comm::SharedChan;
 use std::io::IoError;
 use std::io::net::ip::SocketAddr;
 use std::io::process::ProcessConfig;
@@ -304,7 +303,7 @@ impl IoFactory for UvIoFactory {
         }
     }
 
-    fn signal(&mut self, signum: Signum, channel: SharedChan<Signum>)
+    fn signal(&mut self, signum: Signum, channel: Chan<Signum>)
         -> Result<~rtio::RtioSignal, IoError> {
         match SignalWatcher::new(self, signum, channel) {
             Ok(s) => Ok(s as ~rtio::RtioSignal),

src/libstd/comm/oneshot.rs

@@ -0,0 +1,382 @@
+// Copyright 2014 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.
+
+/// Oneshot channels/ports
+///
+/// This is the initial flavor of channels/ports used for comm module. This is
+/// an optimization for the one-use case of a channel. The major optimization of
+/// this type is to have one and exactly one allocation when the chan/port pair
+/// is created.
+///
+/// Another possible optimization would be to not use an UnsafeArc box because
+/// in theory we know when the shared packet can be deallocated (no real need
+/// for the atomic reference counting), but I was having trouble how to destroy
+/// the data early in a drop of a Port.
+///
+/// # Implementation
+///
+/// Oneshots are implemented around one atomic uint variable. This variable
+/// indicates both the state of the port/chan but also contains any tasks
+/// blocked on the port. All atomic operations happen on this one word.
+///
+/// In order to upgrade a oneshot channel, an upgrade is considered a disconnect
+/// on behalf of the channel side of things (it can be mentally thought of as
+/// consuming the port). This upgrade is then also stored in the shared packet.
+/// The one caveat to consider is that when a port sees a disconnected channel
+/// it must check for data because there is no "data plus upgrade" state.
+
+use comm::Port;
+use kinds::Send;
+use ops::Drop;
+use option::{Some, None, Option};
+use result::{Result, Ok, Err};
+use rt::local::Local;
+use rt::task::{Task, BlockedTask};
+use sync::atomics;
+use util;
+
+// Various states you can find a port in.
+static EMPTY: uint = 0;
+static DATA: uint = 1;
+static DISCONNECTED: uint = 2;
+
+pub struct Packet<T> {
+    // Internal state of the chan/port pair (stores the blocked task as well)
+    state: atomics::AtomicUint,
+    // One-shot data slot location
+    data: Option<T>,
+    // when used for the second time, a oneshot channel must be upgraded, and
+    // this contains the slot for the upgrade
+    upgrade: MyUpgrade<T>,
+}
+
+pub enum Failure<T> {
+    Empty,
+    Disconnected,
+    Upgraded(Port<T>),
+}
+
+pub enum UpgradeResult {
+    UpSuccess,
+    UpDisconnected,
+    UpWoke(BlockedTask),
+}
+
+pub enum SelectionResult<T> {
+    SelCanceled(BlockedTask),
+    SelUpgraded(BlockedTask, Port<T>),
+    SelSuccess,
+}
+
+enum MyUpgrade<T> {
+    NothingSent,
+    SendUsed,
+    GoUp(Port<T>),
+}
+
+impl<T: Send> Packet<T> {
+    pub fn new() -> Packet<T> {
+        Packet {
+            data: None,
+            upgrade: NothingSent,
+            state: atomics::AtomicUint::new(EMPTY),
+        }
+    }
+
+    pub fn send(&mut self, t: T) -> bool {
+        // Sanity check
+        match self.upgrade {
+            NothingSent => {}
+            _ => fail!("sending on a oneshot that's already sent on "),
+        }
+        assert!(self.data.is_none());
+        self.data = Some(t);
+        self.upgrade = SendUsed;
+
+        // This atomic swap uses a "Release" memory ordering to ensure that all
+        // our previous memory writes are visible to the other thread (notably
+        // the write of data/upgrade)
+        match self.state.swap(DATA, atomics::Release) {
+            // Sent the data, no one was waiting
+            EMPTY => true,
+
+            // Couldn't send the data, the port hung up first. We need to be
+            // sure to deallocate the sent data (to not leave it stuck in the
+            // queue)
+            DISCONNECTED => {
+                self.data.take_unwrap();
+                false
+            }
+
+            // Not possible, these are one-use channels
+            DATA => unreachable!(),
+
+            // Anything else means that there was a task waiting on the other
+            // end. We leave the 'DATA' state inside so it'll pick it up on the
+            // other end.
+            n => unsafe {
+                let t = BlockedTask::cast_from_uint(n);
+                t.wake().map(|t| t.reawaken());
+                true
+            }
+        }
+    }
+
+    // Just tests whether this channel has been sent on or not, this is only
+    // safe to use from the sender.
+    pub fn sent(&self) -> bool {
+        match self.upgrade {
+            NothingSent => false,
+            _ => true,
+        }
+    }
+
+    pub fn recv(&mut self) -> Result<T, Failure<T>> {
+        // Attempt to not block the task (it's a little expensive). If it looks
+        // like we're not empty, then immediately go through to `try_recv`.
+        //
+        // These atomics use an Acquire memory ordering in order to have all the
+        // previous writes of the releasing thread visible to us.
+        if self.state.load(atomics::Acquire) == EMPTY {
+            let t: ~Task = Local::take();
+            t.deschedule(1, |task| {
+                let n = unsafe { task.cast_to_uint() };
+                match self.state.compare_and_swap(EMPTY, n, atomics::Acquire) {
+                    // Nothing on the channel, we legitimately block
+                    EMPTY => Ok(()),
+
+                    // If there's data or it's a disconnected channel, then we
+                    // failed the cmpxchg, so we just wake ourselves back up
+                    DATA | DISCONNECTED => {
+                        unsafe { Err(BlockedTask::cast_from_uint(n)) }
+                    }
+
+                    // Only one thread is allowed to sleep on this port
+                    _ => unreachable!()
+                }
+            });
+        }
+
+        self.try_recv()
+    }
+
+    pub fn try_recv(&mut self) -> Result<T, Failure<T>> {
+        // see above for why Acquire is used.
+        match self.state.load(atomics::Acquire) {
+            EMPTY => Err(Empty),
+
+            // We saw some data on the channel, but the channel can be used
+            // again to send us an upgrade. As a result, we need to re-insert
+            // into the channel that there's no data available (otherwise we'll
+            // just see DATA next time). This is done as a cmpxchg because if
+            // the state changes under our feet we'd rather just see that state
+            // change.
+            DATA => {
+                self.state.compare_and_swap(DATA, EMPTY, atomics::Acquire);
+                match self.data.take() {
+                    Some(data) => Ok(data),
+                    None => unreachable!(),
+                }
+            }
+
+            // There's no guarantee that we receive before an upgrade happens,
+            // and an upgrade flags the channel as disconnected, so when we see
+            // this we first need to check if there's data available and *then*
+            // we go through and process the upgrade.
+            DISCONNECTED => {
+                match self.data.take() {
+                    Some(data) => Ok(data),
+                    None => {
+                        match util::replace(&mut self.upgrade, SendUsed) {
+                            SendUsed | NothingSent => Err(Disconnected),
+                            GoUp(upgrade) => Err(Upgraded(upgrade))
+                        }
+                    }
+                }
+            }
+            _ => unreachable!()
+        }
+    }
+
+    // Returns whether the upgrade was completed. If the upgrade wasn't
+    // completed, then the port couldn't get sent to the other half (it will
+    // never receive it).
+    pub fn upgrade(&mut self, up: Port<T>) -> UpgradeResult {
+        let prev = match self.upgrade {
+            NothingSent => NothingSent,
+            SendUsed => SendUsed,
+            _ => fail!("upgrading again"),
+        };
+        self.upgrade = GoUp(up);
+
+        // Use a Release memory ordering in order to make sure that our write to
+        // `upgrade` is visible to the other thread.
+        match self.state.swap(DISCONNECTED, atomics::Release) {
+            // If the channel is empty or has data on it, then we're good to go.
+            // Senders will check the data before the upgrade (in case we
+            // plastered over the DATA state).
+            DATA | EMPTY => UpSuccess,
+
+            // If the other end is already disconnected, then we failed the
+            // upgrade. Be sure to trash the port we were given.
+            DISCONNECTED => { self.upgrade = prev; UpDisconnected }
+
+            // If someone's waiting, we gotta wake them up
+            n => UpWoke(unsafe { BlockedTask::cast_from_uint(n) })
+        }
+    }
+
+    pub fn drop_chan(&mut self) {
+        match self.state.swap(DISCONNECTED, atomics::SeqCst) {
+            DATA | DISCONNECTED | EMPTY => {}
+
+            // If someone's waiting, we gotta wake them up
+            n => unsafe {
+                let t = BlockedTask::cast_from_uint(n);
+                t.wake().map(|t| t.reawaken());
+            }
+        }
+    }
+
+    pub fn drop_port(&mut self) {
+        // Use an Acquire memory ordering in order to see the data that the
+        // senders are sending.
+        match self.state.swap(DISCONNECTED, atomics::Acquire) {
+            // An empty channel has nothing to do, and a remotely disconnected
+            // channel also has nothing to do b/c we're about to run the drop
+            // glue
+            DISCONNECTED | EMPTY => {}
+
+            // There's data on the channel, so make sure we destroy it promptly.
+            // This is why not using an arc is a little difficult (need the box
+            // to stay valid while we take the data).
+            DATA => { self.data.take_unwrap(); }
+
+            // We're the only ones that can block on this port
+            _ => unreachable!()
+        }
+    }
+
+    ////////////////////////////////////////////////////////////////////////////
+    // select implementation
+    ////////////////////////////////////////////////////////////////////////////
+
+    // If Ok, the value is whether this port has data, if Err, then the upgraded
+    // port needs to be checked instead of this one.
+    pub fn can_recv(&mut self) -> Result<bool, Port<T>> {
+        // Use Acquire so we can see all previous memory writes
+        match self.state.load(atomics::Acquire) {
+            EMPTY => Ok(false), // Welp, we tried
+            DATA => Ok(true),   // we have some un-acquired data
+            DISCONNECTED if self.data.is_some() => Ok(true), // we have data
+            DISCONNECTED => {
+                match util::replace(&mut self.upgrade, SendUsed) {
+                    // The other end sent us an upgrade, so we need to
+                    // propagate upwards whether the upgrade can receive
+                    // data
+                    GoUp(upgrade) => Err(upgrade),
+
+                    // If the other end disconnected without sending an
+                    // upgrade, then we have data to receive (the channel is
+                    // disconnected).
+                    up => { self.upgrade = up; Ok(true) }
+                }
+            }
+            _ => unreachable!(), // we're the "one blocker"
+        }
+    }
+
+    // Attempts to start selection on this port. This can either succeed, fail
+    // because there is data, or fail because there is an upgrade pending.
+    pub fn start_selection(&mut self, task: BlockedTask) -> SelectionResult<T> {
+        let n = unsafe { task.cast_to_uint() };
+        match self.state.compare_and_swap(EMPTY, n, atomics::SeqCst) {
+            EMPTY => SelSuccess,
+            DATA => SelCanceled(unsafe { BlockedTask::cast_from_uint(n) }),
+            DISCONNECTED if self.data.is_some() => {
+                SelCanceled(unsafe { BlockedTask::cast_from_uint(n) })
+            }
+            DISCONNECTED => {
+                match util::replace(&mut self.upgrade, SendUsed) {
+                    // The other end sent us an upgrade, so we need to
+                    // propagate upwards whether the upgrade can receive
+                    // data
+                    GoUp(upgrade) => {
+                        SelUpgraded(unsafe { BlockedTask::cast_from_uint(n) },
+                                    upgrade)
+                    }
+
+                    // If the other end disconnected without sending an
+                    // upgrade, then we have data to receive (the channel is
+                    // disconnected).
+                    up => {
+                        self.upgrade = up;
+                        SelCanceled(unsafe { BlockedTask::cast_from_uint(n) })
+                    }
+                }
+            }
+            _ => unreachable!(), // we're the "one blocker"
+        }
+    }
+
+    // Remove a previous selecting task from this port. This ensures that the
+    // blocked task will no longer be visible to any other threads.
+    //
+    // The return value indicates whether there's data on this port.
+    pub fn abort_selection(&mut self) -> Result<bool, Port<T>> {
+        // use Acquire to make sure we see all previous memory writes
+        let state = match self.state.load(atomics::Acquire) {
+            // Each of these states means that no further activity will happen
+            // with regard to abortion selection
+            s @ EMPTY |
+            s @ DATA |
+            s @ DISCONNECTED => s,
+
+            // If we've got a blocked task, then use an atomic to gain ownership
+            // of it (may fail)
+            n => self.state.compare_and_swap(n, EMPTY, atomics::SeqCst)
+        };
+
+        // Now that we've got ownership of our state, figure out what to do
+        // about it.
+        match state {
+            EMPTY => unreachable!(),
+            // our task used for select was stolen
+            DATA => Ok(true),
+
+            // If the other end has hung up, then we have complete ownership
+            // of the port. We need to check to see if there was an upgrade
+            // requested, and if so, the other end needs to have its selection
+            // aborted.
+            DISCONNECTED => {
+                assert!(self.data.is_none());
+                match util::replace(&mut self.upgrade, SendUsed) {
+                    GoUp(port) => Err(port),
+                    _ => Ok(true),
+                }
+            }
+
+            // We woke ourselves up from select. Assert that the task should be
+            // trashed and returne that we don't have any data.
+            n => {
+                let t = unsafe { BlockedTask::cast_from_uint(n) };
+                t.trash();
+                Ok(false)
+            }
+        }
+    }
+}
+
+#[unsafe_destructor]
+impl<T: Send> Drop for Packet<T> {
+    fn drop(&mut self) {
+        assert_eq!(self.state.load(atomics::SeqCst), DISCONNECTED);
+    }
+}

src/libstd/comm/select.rs

@@ -45,19 +45,17 @@
 #[allow(dead_code)];
 
 use cast;
-use comm;
+use cell::Cell;
 use iter::Iterator;
 use kinds::marker;
 use kinds::Send;
 use ops::Drop;
 use option::{Some, None, Option};
 use ptr::RawPtr;
-use result::{Ok, Err};
+use result::{Ok, Err, Result};
 use rt::local::Local;
-use rt::task::Task;
-use super::{Packet, Port};
-use sync::atomics::{Relaxed, SeqCst};
-use task;
+use rt::task::{Task, BlockedTask};
+use super::Port;
 use uint;
 
 macro_rules! select {
@@ -67,8 +65,12 @@ macro_rules! select {
     ) => ({
         use std::comm::Select;
         let sel = Select::new();
-        let mut $port1 = sel.add(&mut $port1);
-        $( let mut $port = sel.add(&mut $port); )*
+        let mut $port1 = sel.handle(&$port1);
+        $( let mut $port = sel.handle(&$port); )*
+        unsafe {
+            $port1.add();
+            $( $port.add(); )*
+        }
         let ret = sel.wait();
         if ret == $port1.id { let $name1 = $port1.$meth1(); $code1 }
         $( else if ret == $port.id { let $name = $port.$meth(); $code } )*
@@ -79,9 +81,9 @@ macro_rules! select {
 /// The "port set" of the select interface. This structure is used to manage a
 /// set of ports which are being selected over.
 pub struct Select {
-    priv head: *mut Packet,
-    priv tail: *mut Packet,
-    priv next_id: uint,
+    priv head: *mut Handle<'static, ()>,
+    priv tail: *mut Handle<'static, ()>,
+    priv next_id: Cell<uint>,
     priv marker1: marker::NoSend,
     priv marker2: marker::NoFreeze,
 }
@@ -90,13 +92,28 @@ pub struct Select {
 /// This handle is used to keep the port in the set as well as interact with the
 /// underlying port.
 pub struct Handle<'port, T> {
-    /// A unique ID for this Handle.
+    /// The ID of this handle, used to compare against the return value of
+    /// `Select::wait()`
     id: uint,
     priv selector: &'port Select,
-    priv port: &'port mut Port<T>,
+    priv next: *mut Handle<'static, ()>,
+    priv prev: *mut Handle<'static, ()>,
+    priv added: bool,
+    priv packet: &'port Packet,
+
+    // due to our fun transmutes, we be sure to place this at the end. (nothing
+    // previous relies on T)
+    priv port: &'port Port<T>,
 }
 
-struct Packets { cur: *mut Packet }
+struct Packets { cur: *mut Handle<'static, ()> }
+
+#[doc(hidden)]
+pub trait Packet {
+    fn can_recv(&self) -> bool;
+    fn start_selection(&self, task: BlockedTask) -> Result<(), BlockedTask>;
+    fn abort_selection(&self) -> bool;
+}
 
 impl Select {
     /// Creates a new selection structure. This set is initially empty and
@@ -106,45 +123,29 @@ impl Select {
     /// rather much easier through the `select!` macro.
     pub fn new() -> Select {
         Select {
-            head: 0 as *mut Packet,
-            tail: 0 as *mut Packet,
-            next_id: 1,
             marker1: marker::NoSend,
             marker2: marker::NoFreeze,
+            head: 0 as *mut Handle<'static, ()>,
+            tail: 0 as *mut Handle<'static, ()>,
+            next_id: Cell::new(1),
         }
     }
 
-    /// Adds a new port to this set, returning a handle which is then used to
-    /// receive on the port.
-    ///
-    /// Note that this port parameter takes `&mut Port` instead of `&Port`. None
-    /// of the methods of receiving on a port require `&mut self`, but `&mut` is
-    /// used here in order to have the compiler guarantee that the same port is
-    /// not added to this set more than once.
-    ///
-    /// When the returned handle falls out of scope, the port will be removed
-    /// from this set. While the handle is in this set, usage of the port can be
-    /// done through the `Handle`'s receiving methods.
-    pub fn add<'a, T: Send>(&'a self, port: &'a mut Port<T>) -> Handle<'a, T> {
-        let this = unsafe { cast::transmute_mut(self) };
-        let id = this.next_id;
-        this.next_id += 1;
-        unsafe {
-            let packet = port.inner.packet();
-            assert!(!(*packet).selecting.load(Relaxed));
-            assert_eq!((*packet).selection_id, 0);
-            (*packet).selection_id = id;
-            if this.head.is_null() {
-                this.head = packet as *mut Packet;
-                this.tail = packet as *mut Packet;
-            } else {
-                (*packet).select_prev = this.tail;
-                assert!((*packet).select_next.is_null());
-                (*this.tail).select_next = packet as *mut Packet;
-                this.tail = packet as *mut Packet;
-            }
+    /// Creates a new handle into this port set for a new port. Note that this
+    /// does *not* add the port to the port set, for that you must call the
+    /// `add` method on the handle itself.
+    pub fn handle<'a, T: Send>(&'a self, port: &'a Port<T>) -> Handle<'a, T> {
+        let id = self.next_id.get();
+        self.next_id.set(id + 1);
+        Handle {
+            id: id,
+            selector: self,
+            next: 0 as *mut Handle<'static, ()>,
+            prev: 0 as *mut Handle<'static, ()>,
+            added: false,
+            port: port,
+            packet: port,
         }
-        Handle { id: id, selector: this, port: port }
     }
 
     /// Waits for an event on this port set. The returned valus is *not* and
@@ -177,10 +178,9 @@ impl Select {
         unsafe {
             let mut amt = 0;
             for p in self.iter() {
-                assert!(!(*p).selecting.load(Relaxed));
                 amt += 1;
-                if (*p).can_recv() {
-                    return (*p).selection_id;
+                if (*p).packet.can_recv() {
+                    return (*p).id;
                 }
             }
             assert!(amt > 0);
@@ -195,22 +195,14 @@ impl Select {
             let task: ~Task = Local::take();
             task.deschedule(amt, |task| {
                 // Prepare for the block
-                let (i, packet) = iter.next().unwrap();
-                assert!((*packet).to_wake.is_none());
-                (*packet).to_wake = Some(task);
-                (*packet).selecting.store(true, SeqCst);
-
-                if (*packet).decrement() {
-                    Ok(())
-                } else {
-                    // Empty to_wake first to avoid tripping an assertion in
-                    // abort_selection in the disconnected case.
-                    let task = (*packet).to_wake.take_unwrap();
-                    (*packet).abort_selection(false);
-                    (*packet).selecting.store(false, SeqCst);
-                    ready_index = i;
-                    ready_id = (*packet).selection_id;
-                    Err(task)
+                let (i, handle) = iter.next().unwrap();
+                match (*handle).packet.start_selection(task) {
+                    Ok(()) => Ok(()),
+                    Err(task) => {
+                        ready_index = i;
+                        ready_id = (*handle).id;
+                        Err(task)
+                    }
                 }
             });
 
@@ -235,45 +227,17 @@ impl Select {
             // A rewrite should focus on avoiding a yield loop, and for now this
             // implementation is tying us over to a more efficient "don't
             // iterate over everything every time" implementation.
-            for packet in self.iter().take(ready_index) {
-                if (*packet).abort_selection(true) {
-                    ready_id = (*packet).selection_id;
-                    while (*packet).selecting.load(Relaxed) {
-                        task::deschedule();
-                    }
+            for handle in self.iter().take(ready_index) {
+                if (*handle).packet.abort_selection() {
+                    ready_id = (*handle).id;
                 }
             }
 
-            // Sanity check for now to make sure that everyone is turned off.
-            for packet in self.iter() {
-                assert!(!(*packet).selecting.load(Relaxed));
-            }
-
             assert!(ready_id != uint::MAX);
             return ready_id;
         }
     }
 
-    unsafe fn remove(&self, packet: *mut Packet) {
-        let this = cast::transmute_mut(self);
-        assert!(!(*packet).selecting.load(Relaxed));
-        if (*packet).select_prev.is_null() {
-            assert_eq!(packet, this.head);
-            this.head = (*packet).select_next;
-        } else {
-            (*(*packet).select_prev).select_next = (*packet).select_next;
-        }
-        if (*packet).select_next.is_null() {
-            assert_eq!(packet, this.tail);
-            this.tail = (*packet).select_prev;
-        } else {
-            (*(*packet).select_next).select_prev = (*packet).select_prev;
-        }
-        (*packet).select_next = 0 as *mut Packet;
-        (*packet).select_prev = 0 as *mut Packet;
-        (*packet).selection_id = 0;
-    }
-
     fn iter(&self) -> Packets { Packets { cur: self.head } }
 }
 
@@ -285,10 +249,56 @@ impl<'port, T: Send> Handle<'port, T> {
     /// success or `None` if the channel disconnects. This function has the same
     /// semantics as `Port.recv_opt`
     pub fn recv_opt(&mut self) -> Option<T> { self.port.recv_opt() }
-    /// Immediately attempt to receive a value on a port, this function will
-    /// never block. Has the same semantics as `Port.try_recv`.
-    pub fn try_recv(&mut self) -> comm::TryRecvResult<T> {
-        self.port.try_recv()
+
+    /// Adds this handle to the port set that the handle was created from. This
+    /// method can be called multiple times, but it has no effect if `add` was
+    /// called previously.
+    ///
+    /// This method is unsafe because it requires that the `Handle` is not moved
+    /// while it is added to the `Select` set.
+    pub unsafe fn add(&mut self) {
+        if self.added { return }
+        let selector: &mut Select = cast::transmute(&*self.selector);
+        let me: *mut Handle<'static, ()> = cast::transmute(&*self);
+
+        if selector.head.is_null() {
+            selector.head = me;
+            selector.tail = me;
+        } else {
+            (*me).prev = selector.tail;
+            assert!((*me).next.is_null());
+            (*selector.tail).next = me;
+            selector.tail = me;
+        }
+        self.added = true;
+    }
+
+    /// Removes this handle from the `Select` set. This method is unsafe because
+    /// it has no guarantee that the `Handle` was not moved since `add` was
+    /// called.
+    pub unsafe fn remove(&mut self) {
+        if !self.added { return }
+
+        let selector: &mut Select = cast::transmute(&*self.selector);
+        let me: *mut Handle<'static, ()> = cast::transmute(&*self);
+
+        if self.prev.is_null() {
+            assert_eq!(selector.head, me);
+            selector.head = self.next;
+        } else {
+            (*self.prev).next = self.next;
+        }
+        if self.next.is_null() {
+            assert_eq!(selector.tail, me);
+            selector.tail = self.prev;
+        } else {
+            (*self.next).prev = self.prev;
+        }
+
+        self.next = 0 as *mut Handle<'static, ()>;
+        self.prev = 0 as *mut Handle<'static, ()>;
+
+        self.added = false;
     }
 }
 
@@ -303,17 +313,17 @@ impl Drop for Select {
 #[unsafe_destructor]
 impl<'port, T: Send> Drop for Handle<'port, T> {
     fn drop(&mut self) {
-        unsafe { self.selector.remove(self.port.inner.packet()) }
+        unsafe { self.remove() }
     }
 }
 
-impl Iterator<*mut Packet> for Packets {
-    fn next(&mut self) -> Option<*mut Packet> {
+impl Iterator<*mut Handle<'static, ()>> for Packets {
+    fn next(&mut self) -> Option<*mut Handle<'static, ()>> {
         if self.cur.is_null() {
             None
         } else {
             let ret = Some(self.cur);
-            unsafe { self.cur = (*self.cur).select_next; }
+            unsafe { self.cur = (*self.cur).next; }
             ret
         }
     }
@@ -326,8 +336,8 @@ mod test {
     use prelude::*;
 
     test!(fn smoke() {
-        let (mut p1, c1) = Chan::<int>::new();
-        let (mut p2, c2) = Chan::<int>::new();
+        let (p1, c1) = Chan::<int>::new();
+        let (p2, c2) = Chan::<int>::new();
         c1.send(1);
         select! (
             foo = p1.recv() => { assert_eq!(foo, 1); },
@@ -350,11 +360,11 @@ mod test {
     })
 
     test!(fn smoke2() {
-        let (mut p1, _c1) = Chan::<int>::new();
-        let (mut p2, _c2) = Chan::<int>::new();
-        let (mut p3, _c3) = Chan::<int>::new();
-        let (mut p4, _c4) = Chan::<int>::new();
-        let (mut p5, c5) = Chan::<int>::new();
+        let (p1, _c1) = Chan::<int>::new();
+        let (p2, _c2) = Chan::<int>::new();
+        let (p3, _c3) = Chan::<int>::new();
+        let (p4, _c4) = Chan::<int>::new();
+        let (p5, c5) = Chan::<int>::new();
         c5.send(4);
         select! (
             _foo = p1.recv() => { fail!("1") },
@@ -366,8 +376,8 @@ mod test {
     })
 
     test!(fn closed() {
-        let (mut p1, _c1) = Chan::<int>::new();
-        let (mut p2, c2) = Chan::<int>::new();
+        let (p1, _c1) = Chan::<int>::new();
+        let (p2, c2) = Chan::<int>::new();
         drop(c2);
 
         select! (
@@ -377,8 +387,8 @@ mod test {
     })
 
     test!(fn unblocks() {
-        let (mut p1, c1) = Chan::<int>::new();
-        let (mut p2, _c2) = Chan::<int>::new();
+        let (p1, c1) = Chan::<int>::new();
+        let (p2, _c2) = Chan::<int>::new();
         let (p3, c3) = Chan::<int>::new();
 
         spawn(proc() {
@@ -400,8 +410,8 @@ mod test {
     })
 
     test!(fn both_ready() {
-        let (mut p1, c1) = Chan::<int>::new();
-        let (mut p2, c2) = Chan::<int>::new();
+        let (p1, c1) = Chan::<int>::new();
+        let (p2, c2) = Chan::<int>::new();
         let (p3, c3) = Chan::<()>::new();
 
         spawn(proc() {
@@ -426,8 +436,8 @@ mod test {
 
     test!(fn stress() {
         static AMT: int = 10000;
-        let (mut p1, c1) = Chan::<int>::new();
-        let (mut p2, c2) = Chan::<int>::new();
+        let (p1, c1) = Chan::<int>::new();
+        let (p2, c2) = Chan::<int>::new();
         let (p3, c3) = Chan::<()>::new();
 
         spawn(proc() {
@@ -449,4 +459,66 @@ mod test {
             c3.send(());
         }
     })
+
+    test!(fn cloning() {
+        let (p1, c1) = Chan::<int>::new();
+        let (p2, _c2) = Chan::<int>::new();
+        let (p3, c3) = Chan::<()>::new();
+
+        spawn(proc() {
+            p3.recv();
+            c1.clone();
+            assert_eq!(p3.try_recv(), Empty);
+            c1.send(2);
+            p3.recv();
+        });
+
+        c3.send(());
+        select!(
+            _i1 = p1.recv() => {},
+            _i2 = p2.recv() => fail!()
+        )
+        c3.send(());
+    })
+
+    test!(fn cloning2() {
+        let (p1, c1) = Chan::<int>::new();
+        let (p2, _c2) = Chan::<int>::new();
+        let (p3, c3) = Chan::<()>::new();
+
+        spawn(proc() {
+            p3.recv();
+            c1.clone();
+            assert_eq!(p3.try_recv(), Empty);
+            c1.send(2);
+            p3.recv();
+        });
+
+        c3.send(());
+        select!(
+            _i1 = p1.recv() => {},
+            _i2 = p2.recv() => fail!()
+        )
+        c3.send(());
+    })
+
+    test!(fn cloning3() {
+        let (p1, c1) = Chan::<()>::new();
+        let (p2, c2) = Chan::<()>::new();
+        let (p, c) = Chan::new();
+        spawn(proc() {
+            let mut s = Select::new();
+            let mut h1 = s.handle(&p1);
+            let mut h2 = s.handle(&p2);
+            unsafe { h2.add(); }
+            unsafe { h1.add(); }
+            assert_eq!(s.wait(), h2.id);
+            c.send(());
+        });
+
+        for _ in range(0, 1000) { task::deschedule(); }
+        drop(c1.clone());
+        c2.send(());
+        p.recv();
+    })
 }

src/libstd/comm/shared.rs

@@ -0,0 +1,483 @@
+// Copyright 2014 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.
+
+/// Shared channels
+///
+/// This is the flavor of channels which are not necessarily optimized for any
+/// particular use case, but are the most general in how they are used. Shared
+/// channels are cloneable allowing for multiple senders.
+///
+/// High level implementation details can be found in the comment of the parent
+/// module. You'll also note that the implementation of the shared and stream
+/// channels are quite similar, and this is no coincidence!
+
+use int;
+use iter::Iterator;
+use kinds::Send;
+use ops::Drop;
+use option::{Some, None, Option};
+use result::{Ok, Err, Result};
+use rt::local::Local;
+use rt::task::{Task, BlockedTask};
+use rt::thread::Thread;
+use sync::atomics;
+use unstable::mutex::Mutex;
+use vec::OwnedVector;
+
+use mpsc = sync::mpsc_queue;
+
+static DISCONNECTED: int = int::MIN;
+static FUDGE: int = 1024;
+static MAX_STEALS: int = 1 << 20;
+
+pub struct Packet<T> {
+    queue: mpsc::Queue<T>,
+    cnt: atomics::AtomicInt, // How many items are on this channel
+    steals: int, // How many times has a port received without blocking?
+    to_wake: atomics::AtomicUint, // Task to wake up
+
+    // The number of channels which are currently using this packet.
+    channels: atomics::AtomicInt,
+
+    // See the discussion in Port::drop and the channel send methods for what
+    // these are used for
+    port_dropped: atomics::AtomicBool,
+    sender_drain: atomics::AtomicInt,
+
+    // this lock protects various portions of this implementation during
+    // select()
+    select_lock: Mutex,
+}
+
+pub enum Failure {
+    Empty,
+    Disconnected,
+}
+
+impl<T: Send> Packet<T> {
+    // Creation of a packet *must* be followed by a call to inherit_blocker
+    pub fn new() -> Packet<T> {
+        let mut p = Packet {
+            queue: mpsc::Queue::new(),
+            cnt: atomics::AtomicInt::new(0),
+            steals: 0,
+            to_wake: atomics::AtomicUint::new(0),
+            channels: atomics::AtomicInt::new(2),
+            port_dropped: atomics::AtomicBool::new(false),
+            sender_drain: atomics::AtomicInt::new(0),
+            select_lock: unsafe { Mutex::new() },
+        };
+        // see comments in inherit_blocker about why we grab this lock
+        unsafe { p.select_lock.lock() }
+        return p;
+    }
+
+    // This function is used at the creation of a shared packet to inherit a
+    // previously blocked task. This is done to prevent spurious wakeups of
+    // tasks in select().
+    //
+    // This can only be called at channel-creation time
+    pub fn inherit_blocker(&mut self, task: Option<BlockedTask>) {
+        match task {
+            Some(task) => {
+                assert_eq!(self.cnt.load(atomics::SeqCst), 0);
+                assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
+                self.to_wake.store(unsafe { task.cast_to_uint() },
+                                   atomics::SeqCst);
+                self.cnt.store(-1, atomics::SeqCst);
+
+                // This store is a little sketchy. What's happening here is
+                // that we're transferring a blocker from a oneshot or stream
+                // channel to this shared channel. In doing so, we never
+                // spuriously wake them up and rather only wake them up at the
+                // appropriate time. This implementation of shared channels
+                // assumes that any blocking recv() will undo the increment of
+                // steals performed in try_recv() once the recv is complete.
+                // This thread that we're inheriting, however, is not in the
+                // middle of recv. Hence, the first time we wake them up,
+                // they're going to wake up from their old port, move on to the
+                // upgraded port, and then call the block recv() function.
+                //
+                // When calling this function, they'll find there's data
+                // immediately available, counting it as a steal. This in fact
+                // wasn't a steal because we appropriately blocked them waiting
+                // for data.
+                //
+                // To offset this bad increment, we initially set the steal
+                // count to -1. You'll find some special code in
+                // abort_selection() as well to ensure that this -1 steal count
+                // doesn't escape too far.
+                self.steals = -1;
+            }
+            None => {}
+        }
+
+        // When the shared packet is constructed, we grabbed this lock. The
+        // purpose of this lock is to ensure that abort_selection() doesn't
+        // interfere with this method. After we unlock this lock, we're
+        // signifying that we're done modifying self.cnt and self.to_wake and
+        // the port is ready for the world to continue using it.
+        unsafe { self.select_lock.unlock() }
+    }
+
+    pub fn send(&mut self, t: T) -> bool {
+        // See Port::drop for what's going on
+        if self.port_dropped.load(atomics::SeqCst) { return false }
+
+        // Note that the multiple sender case is a little tricker
+        // semantically than the single sender case. The logic for
+        // incrementing is "add and if disconnected store disconnected".
+        // This could end up leading some senders to believe that there
+        // wasn't a disconnect if in fact there was a disconnect. This means
+        // that while one thread is attempting to re-store the disconnected
+        // states, other threads could walk through merrily incrementing
+        // this very-negative disconnected count. To prevent senders from
+        // spuriously attempting to send when the channels is actually
+        // disconnected, the count has a ranged check here.
+        //
+        // This is also done for another reason. Remember that the return
+        // value of this function is:
+        //
+        //  `true` == the data *may* be received, this essentially has no
+        //            meaning
+        //  `false` == the data will *never* be received, this has a lot of
+        //             meaning
+        //
+        // In the SPSC case, we have a check of 'queue.is_empty()' to see
+        // whether the data was actually received, but this same condition
+        // means nothing in a multi-producer context. As a result, this
+        // preflight check serves as the definitive "this will never be
+        // received". Once we get beyond this check, we have permanently
+        // entered the realm of "this may be received"
+        if self.cnt.load(atomics::SeqCst) < DISCONNECTED + FUDGE {
+            return false
+        }
+
+        self.queue.push(t);
+        match self.cnt.fetch_add(1, atomics::SeqCst) {
+            -1 => {
+                self.take_to_wake().wake().map(|t| t.reawaken());
+            }
+
+            // In this case, we have possibly failed to send our data, and
+            // we need to consider re-popping the data in order to fully
+            // destroy it. We must arbitrate among the multiple senders,
+            // however, because the queues that we're using are
+            // single-consumer queues. In order to do this, all exiting
+            // pushers will use an atomic count in order to count those
+            // flowing through. Pushers who see 0 are required to drain as
+            // much as possible, and then can only exit when they are the
+            // only pusher (otherwise they must try again).
+            n if n < DISCONNECTED + FUDGE => {
+                // see the comment in 'try' for a shared channel for why this
+                // window of "not disconnected" is ok.
+                self.cnt.store(DISCONNECTED, atomics::SeqCst);
+
+                if self.sender_drain.fetch_add(1, atomics::SeqCst) == 0 {
+                    loop {
+                        // drain the queue, for info on the thread yield see the
+                        // discussion in try_recv
+                        loop {
+                            match self.queue.pop() {
+                                mpsc::Data(..) => {}
+                                mpsc::Empty => break,
+                                mpsc::Inconsistent => Thread::yield_now(),
+                            }
+                        }
+                        // maybe we're done, if we're not the last ones
+                        // here, then we need to go try again.
+                        if self.sender_drain.fetch_sub(1, atomics::SeqCst) == 1 {
+                            break
+                        }
+                    }
+
+                    // At this point, there may still be data on the queue,
+                    // but only if the count hasn't been incremented and
+                    // some other sender hasn't finished pushing data just
+                    // yet. That sender in question will drain its own data.
+                }
+            }
+
+            // Can't make any assumptions about this case like in the SPSC case.
+            _ => {}
+        }
+
+        true
+    }
+
+    pub fn recv(&mut self) -> Result<T, Failure> {
+        // This code is essentially the exact same as that found in the stream
+        // case (see stream.rs)
+        match self.try_recv() {
+            Err(Empty) => {}
+            data => return data,
+        }
+
+        let task: ~Task = Local::take();
+        task.deschedule(1, |task| {
+            self.decrement(task)
+        });
+
+        match self.try_recv() {
+            data @ Ok(..) => { self.steals -= 1; data }
+            data => data,
+        }
+    }
+
+    // Essentially the exact same thing as the stream decrement function.
+    fn decrement(&mut self, task: BlockedTask) -> Result<(), BlockedTask> {
+        assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
+        let n = unsafe { task.cast_to_uint() };
+        self.to_wake.store(n, atomics::SeqCst);
+
+        let steals = self.steals;
+        self.steals = 0;
+
+        match self.cnt.fetch_sub(1 + steals, atomics::SeqCst) {
+            DISCONNECTED => { self.cnt.store(DISCONNECTED, atomics::SeqCst); }
+            // If we factor in our steals and notice that the channel has no
+            // data, we successfully sleep
+            n => {
+                assert!(n >= 0);
+                if n - steals <= 0 { return Ok(()) }
+            }
+        }
+
+        self.to_wake.store(0, atomics::SeqCst);
+        Err(unsafe { BlockedTask::cast_from_uint(n) })
+    }
+
+    pub fn try_recv(&mut self) -> Result<T, Failure> {
+        let ret = match self.queue.pop() {
+            mpsc::Data(t) => Some(t),
+            mpsc::Empty => None,
+
+            // This is a bit of an interesting case. The channel is
+            // reported as having data available, but our pop() has
+            // failed due to the queue being in an inconsistent state.
+            // This means that there is some pusher somewhere which has
+            // yet to complete, but we are guaranteed that a pop will
+            // eventually succeed. In this case, we spin in a yield loop
+            // because the remote sender should finish their enqueue
+            // operation "very quickly".
+            //
+            // Note that this yield loop does *not* attempt to do a green
+            // yield (regardless of the context), but *always* performs an
+            // OS-thread yield. The reasoning for this is that the pusher in
+            // question which is causing the inconsistent state is
+            // guaranteed to *not* be a blocked task (green tasks can't get
+            // pre-empted), so it must be on a different OS thread. Also,
+            // `try_recv` is normally a "guaranteed no rescheduling" context
+            // in a green-thread situation. By yielding control of the
+            // thread, we will hopefully allow time for the remote task on
+            // the other OS thread to make progress.
+            //
+            // Avoiding this yield loop would require a different queue
+            // abstraction which provides the guarantee that after M
+            // pushes have succeeded, at least M pops will succeed. The
+            // current queues guarantee that if there are N active
+            // pushes, you can pop N times once all N have finished.
+            mpsc::Inconsistent => {
+                let data;
+                loop {
+                    Thread::yield_now();
+                    match self.queue.pop() {
+                        mpsc::Data(t) => { data = t; break }
+                        mpsc::Empty => fail!("inconsistent => empty"),
+                        mpsc::Inconsistent => {}
+                    }
+                }
+                Some(data)
+            }
+        };
+        match ret {
+            // See the discussion in the stream implementation for why we we
+            // might decrement steals.
+            Some(data) => {
+                self.steals += 1;
+                if self.steals > MAX_STEALS {
+                    match self.cnt.swap(0, atomics::SeqCst) {
+                        DISCONNECTED => {
+                            self.cnt.store(DISCONNECTED, atomics::SeqCst);
+                        }
+                        n => { self.steals -= n; }
+                    }
+                    assert!(self.steals >= 0);
+                }
+                Ok(data)
+            }
+
+            // See the discussion in the stream implementation for why we try
+            // again.
+            None => {
+                match self.cnt.load(atomics::SeqCst) {
+                    n if n != DISCONNECTED => Err(Empty),
+                    _ => {
+                        match self.queue.pop() {
+                            mpsc::Data(t) => Ok(t),
+                            mpsc::Empty => Err(Disconnected),
+                            // with no senders, an inconsistency is impossible.
+                            mpsc::Inconsistent => unreachable!(),
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    // Prepares this shared packet for a channel clone, essentially just bumping
+    // a refcount.
+    pub fn clone_chan(&mut self) {
+        self.channels.fetch_add(1, atomics::SeqCst);
+    }
+
+    // Decrement the reference count on a channel. This is called whenever a
+    // Chan is dropped and may end up waking up a receiver. It's the receiver's
+    // responsibility on the other end to figure out that we've disconnected.
+    pub fn drop_chan(&mut self) {
+        match self.channels.fetch_sub(1, atomics::SeqCst) {
+            1 => {}
+            n if n > 1 => return,
+            n => fail!("bad number of channels left {}", n),
+        }
+
+        match self.cnt.swap(DISCONNECTED, atomics::SeqCst) {
+            -1 => { self.take_to_wake().wake().map(|t| t.reawaken()); }
+            DISCONNECTED => {}
+            n => { assert!(n >= 0); }
+        }
+    }
+
+    // See the long discussion inside of stream.rs for why the queue is drained,
+    // and why it is done in this fashion.
+    pub fn drop_port(&mut self) {
+        self.port_dropped.store(true, atomics::SeqCst);
+        let mut steals = self.steals;
+        while {
+            let cnt = self.cnt.compare_and_swap(
+                            steals, DISCONNECTED, atomics::SeqCst);
+            cnt != DISCONNECTED && cnt != steals
+        } {
+            // See the discussion in 'try_recv' for why we yield
+            // control of this thread.
+            loop {
+                match self.queue.pop() {
+                    mpsc::Data(..) => { steals += 1; }
+                    mpsc::Empty | mpsc::Inconsistent => break,
+                }
+            }
+        }
+    }
+
+    // Consumes ownership of the 'to_wake' field.
+    fn take_to_wake(&mut self) -> BlockedTask {
+        let task = self.to_wake.load(atomics::SeqCst);
+        self.to_wake.store(0, atomics::SeqCst);
+        assert!(task != 0);
+        unsafe { BlockedTask::cast_from_uint(task) }
+    }
+
+    ////////////////////////////////////////////////////////////////////////////
+    // select implementation
+    ////////////////////////////////////////////////////////////////////////////
+
+    // Helper function for select, tests whether this port can receive without
+    // blocking (obviously not an atomic decision).
+    //
+    // This is different than the stream version because there's no need to peek
+    // at the queue, we can just look at the local count.
+    pub fn can_recv(&mut self) -> bool {
+        let cnt = self.cnt.load(atomics::SeqCst);
+        cnt == DISCONNECTED || cnt - self.steals > 0
+    }
+
+    // Inserts the blocked task for selection on this port, returning it back if
+    // the port already has data on it.
+    //
+    // The code here is the same as in stream.rs, except that it doesn't need to
+    // peek at the channel to see if an upgrade is pending.
+    pub fn start_selection(&mut self,
+                           task: BlockedTask) -> Result<(), BlockedTask> {
+        match self.decrement(task) {
+            Ok(()) => Ok(()),
+            Err(task) => {
+                let prev = self.cnt.fetch_add(1, atomics::SeqCst);
+                assert!(prev >= 0);
+                return Err(task);
+            }
+        }
+    }
+
+    // Cancels a previous task waiting on this port, returning whether there's
+    // data on the port.
+    //
+    // This is similar to the stream implementation (hence fewer comments), but
+    // uses a different value for the "steals" variable.
+    pub fn abort_selection(&mut self, _was_upgrade: bool) -> bool {
+        // Before we do anything else, we bounce on this lock. The reason for
+        // doing this is to ensure that any upgrade-in-progress is gone and
+        // done with. Without this bounce, we can race with inherit_blocker
+        // about looking at and dealing with to_wake. Once we have acquired the
+        // lock, we are guaranteed that inherit_blocker is done.
+        unsafe {
+            self.select_lock.lock();
+            self.select_lock.unlock();
+        }
+
+        // Like the stream implementation, we want to make sure that the count
+        // on the channel goes non-negative. We don't know how negative the
+        // stream currently is, so instead of using a steal value of 1, we load
+        // the channel count and figure out what we should do to make it
+        // positive.
+        let steals = {
+            let cnt = self.cnt.load(atomics::SeqCst);
+            if cnt < 0 && cnt != DISCONNECTED {-cnt} else {0}
+        };
+        let prev = self.cnt.fetch_add(steals + 1, atomics::SeqCst);
+
+        if prev == DISCONNECTED {
+            assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
+            self.cnt.store(DISCONNECTED, atomics::SeqCst);
+            true
+        } else {
+            let cur = prev + steals + 1;
+            assert!(cur >= 0);
+            if prev < 0 {
+                self.take_to_wake().trash();
+            } else {
+                while self.to_wake.load(atomics::SeqCst) != 0 {
+                    Thread::yield_now();
+                }
+            }
+            // if the number of steals is -1, it was the pre-emptive -1 steal
+            // count from when we inherited a blocker. This is fine because
+            // we're just going to overwrite it with a real value.
+            assert!(self.steals == 0 || self.steals == -1);
+            self.steals = steals;
+            prev >= 0
+        }
+    }
+}
+
+#[unsafe_destructor]
+impl<T: Send> Drop for Packet<T> {
+    fn drop(&mut self) {
+        unsafe {
+            // Note that this load is not only an assert for correctness about
+            // disconnection, but also a proper fence before the read of
+            // `to_wake`, so this assert cannot be removed with also removing
+            // the `to_wake` assert.
+            assert_eq!(self.cnt.load(atomics::SeqCst), DISCONNECTED);
+            assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
+            assert_eq!(self.channels.load(atomics::SeqCst), 0);
+            self.select_lock.destroy();
+        }
+    }
+}

src/libstd/comm/stream.rs

@@ -0,0 +1,460 @@
+// Copyright 2014 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.
+
+/// Stream channels
+///
+/// This is the flavor of channels which are optimized for one sender and one
+/// receiver. The sender will be upgraded to a shared channel if the channel is
+/// cloned.
+///
+/// High level implementation details can be found in the comment of the parent
+/// module.
+
+use comm::Port;
+use int;
+use iter::Iterator;
+use kinds::Send;
+use ops::Drop;
+use option::{Some, None};
+use result::{Ok, Err, Result};
+use rt::local::Local;
+use rt::task::{Task, BlockedTask};
+use rt::thread::Thread;
+use spsc = sync::spsc_queue;
+use sync::atomics;
+use vec::OwnedVector;
+
+static DISCONNECTED: int = int::MIN;
+static MAX_STEALS: int = 1 << 20;
+
+pub struct Packet<T> {
+    queue: spsc::Queue<Message<T>>, // internal queue for all message
+
+    cnt: atomics::AtomicInt, // How many items are on this channel
+    steals: int, // How many times has a port received without blocking?
+    to_wake: atomics::AtomicUint, // Task to wake up
+
+    port_dropped: atomics::AtomicBool, // flag if the channel has been destroyed.
+}
+
+pub enum Failure<T> {
+    Empty,
+    Disconnected,
+    Upgraded(Port<T>),
+}
+
+pub enum UpgradeResult {
+    UpSuccess,
+    UpDisconnected,
+    UpWoke(BlockedTask),
+}
+
+pub enum SelectionResult<T> {
+    SelSuccess,
+    SelCanceled(BlockedTask),
+    SelUpgraded(BlockedTask, Port<T>),
+}
+
+// Any message could contain an "upgrade request" to a new shared port, so the
+// internal queue it's a queue of T, but rather Message<T>
+enum Message<T> {
+    Data(T),
+    GoUp(Port<T>),
+}
+
+impl<T: Send> Packet<T> {
+    pub fn new() -> Packet<T> {
+        Packet {
+            queue: spsc::Queue::new(128),
+
+            cnt: atomics::AtomicInt::new(0),
+            steals: 0,
+            to_wake: atomics::AtomicUint::new(0),
+
+            port_dropped: atomics::AtomicBool::new(false),
+        }
+    }
+
+
+    pub fn send(&mut self, t: T) -> bool {
+        match self.do_send(Data(t)) {
+            UpSuccess => true,
+            UpDisconnected => false,
+            UpWoke(task) => {
+                task.wake().map(|t| t.reawaken());
+                true
+            }
+        }
+    }
+    pub fn upgrade(&mut self, up: Port<T>) -> UpgradeResult {
+        self.do_send(GoUp(up))
+    }
+
+    fn do_send(&mut self, t: Message<T>) -> UpgradeResult {
+        // Use an acquire/release ordering to maintain the same position with
+        // respect to the atomic loads below
+        if self.port_dropped.load(atomics::SeqCst) { return UpDisconnected }
+
+        self.queue.push(t);
+        match self.cnt.fetch_add(1, atomics::SeqCst) {
+            // As described in the mod's doc comment, -1 == wakeup
+            -1 => UpWoke(self.take_to_wake()),
+            // As as described before, SPSC queues must be >= -2
+            -2 => UpSuccess,
+
+            // Be sure to preserve the disconnected state, and the return value
+            // in this case is going to be whether our data was received or not.
+            // This manifests itself on whether we have an empty queue or not.
+            //
+            // Primarily, are required to drain the queue here because the port
+            // will never remove this data. We can only have at most one item to
+            // drain (the port drains the rest).
+            DISCONNECTED => {
+                self.cnt.store(DISCONNECTED, atomics::SeqCst);
+                let first = self.queue.pop();
+                let second = self.queue.pop();
+                assert!(second.is_none());
+
+                match first {
+                    Some(..) => UpSuccess,  // we failed to send the data
+                    None => UpDisconnected, // we successfully sent data
+                }
+            }
+
+            // Otherwise we just sent some data on a non-waiting queue, so just
+            // make sure the world is sane and carry on!
+            n => { assert!(n >= 0); UpSuccess }
+        }
+    }
+
+    // Consumes ownership of the 'to_wake' field.
+    fn take_to_wake(&mut self) -> BlockedTask {
+        let task = self.to_wake.load(atomics::SeqCst);
+        self.to_wake.store(0, atomics::SeqCst);
+        assert!(task != 0);
+        unsafe { BlockedTask::cast_from_uint(task) }
+    }
+
+    // Decrements the count on the channel for a sleeper, returning the sleeper
+    // back if it shouldn't sleep. Note that this is the location where we take
+    // steals into account.
+    fn decrement(&mut self, task: BlockedTask) -> Result<(), BlockedTask> {
+        assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
+        let n = unsafe { task.cast_to_uint() };
+        self.to_wake.store(n, atomics::SeqCst);
+
+        let steals = self.steals;
+        self.steals = 0;
+
+        match self.cnt.fetch_sub(1 + steals, atomics::SeqCst) {
+            DISCONNECTED => { self.cnt.store(DISCONNECTED, atomics::SeqCst); }
+            // If we factor in our steals and notice that the channel has no
+            // data, we successfully sleep
+            n => {
+                assert!(n >= 0);
+                if n - steals <= 0 { return Ok(()) }
+            }
+        }
+
+        self.to_wake.store(0, atomics::SeqCst);
+        Err(unsafe { BlockedTask::cast_from_uint(n) })
+    }
+
+    pub fn recv(&mut self) -> Result<T, Failure<T>> {
+        // Optimistic preflight check (scheduling is expensive).
+        match self.try_recv() {
+            Err(Empty) => {}
+            data => return data,
+        }
+
+        // Welp, our channel has no data. Deschedule the current task and
+        // initiate the blocking protocol.
+        let task: ~Task = Local::take();
+        task.deschedule(1, |task| {
+            self.decrement(task)
+        });
+
+        match self.try_recv() {
+            // Messages which actually popped from the queue shouldn't count as
+            // a steal, so offset the decrement here (we already have our
+            // "steal" factored into the channel count above).
+            data @ Ok(..) |
+            data @ Err(Upgraded(..)) => {
+                self.steals -= 1;
+                data
+            }
+
+            data => data,
+        }
+    }
+
+    pub fn try_recv(&mut self) -> Result<T, Failure<T>> {
+        match self.queue.pop() {
+            // If we stole some data, record to that effect (this will be
+            // factored into cnt later on). Note that we don't allow steals to
+            // grow without bound in order to prevent eventual overflow of
+            // either steals or cnt as an overflow would have catastrophic
+            // results. Also note that we don't unconditionally set steals to 0
+            // because it can be true that steals > cnt.
+            Some(data) => {
+                self.steals += 1;
+                if self.steals > MAX_STEALS {
+                    match self.cnt.swap(0, atomics::SeqCst) {
+                        DISCONNECTED => {
+                            self.cnt.store(DISCONNECTED, atomics::SeqCst);
+                        }
+                        n => { self.steals -= n; }
+                    }
+                    assert!(self.steals >= 0);
+                }
+                match data {
+                    Data(t) => Ok(t),
+                    GoUp(up) => Err(Upgraded(up)),
+                }
+            }
+
+            None => {
+                match self.cnt.load(atomics::SeqCst) {
+                    n if n != DISCONNECTED => Err(Empty),
+
+                    // This is a little bit of a tricky case. We failed to pop
+                    // data above, and then we have viewed that the channel is
+                    // disconnected. In this window more data could have been
+                    // sent on the channel. It doesn't really make sense to
+                    // return that the channel is disconnected when there's
+                    // actually data on it, so be extra sure there's no data by
+                    // popping one more time.
+                    //
+                    // We can ignore steals because the other end is
+                    // disconnected and we'll never need to really factor in our
+                    // steals again.
+                    _ => {
+                        match self.queue.pop() {
+                            Some(Data(t)) => Ok(t),
+                            Some(GoUp(up)) => Err(Upgraded(up)),
+                            None => Err(Disconnected),
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    pub fn drop_chan(&mut self) {
+        // Dropping a channel is pretty simple, we just flag it as disconnected
+        // and then wakeup a blocker if there is one.
+        match self.cnt.swap(DISCONNECTED, atomics::SeqCst) {
+            -1 => { self.take_to_wake().wake().map(|t| t.reawaken()); }
+            DISCONNECTED => {}
+            n => { assert!(n >= 0); }
+        }
+    }
+
+    pub fn drop_port(&mut self) {
+        // Dropping a port seems like a fairly trivial thing. In theory all we
+        // need to do is flag that we're disconnected and then everything else
+        // can take over (we don't have anyone to wake up).
+        //
+        // The catch for Ports is that we want to drop the entire contents of
+        // the queue. There are multiple reasons for having this property, the
+        // largest of which is that if another chan is waiting in this channel
+        // (but not received yet), then waiting on that port will cause a
+        // deadlock.
+        //
+        // So if we accept that we must now destroy the entire contents of the
+        // queue, this code may make a bit more sense. The tricky part is that
+        // we can't let any in-flight sends go un-dropped, we have to make sure
+        // *everything* is dropped and nothing new will come onto the channel.
+
+        // The first thing we do is set a flag saying that we're done for. All
+        // sends are gated on this flag, so we're immediately guaranteed that
+        // there are a bounded number of active sends that we'll have to deal
+        // with.
+        self.port_dropped.store(true, atomics::SeqCst);
+
+        // Now that we're guaranteed to deal with a bounded number of senders,
+        // we need to drain the queue. This draining process happens atomically
+        // with respect to the "count" of the channel. If the count is nonzero
+        // (with steals taken into account), then there must be data on the
+        // channel. In this case we drain everything and then try again. We will
+        // continue to fail while active senders send data while we're dropping
+        // data, but eventually we're guaranteed to break out of this loop
+        // (because there is a bounded number of senders).
+        let mut steals = self.steals;
+        while {
+            let cnt = self.cnt.compare_and_swap(
+                            steals, DISCONNECTED, atomics::SeqCst);
+            cnt != DISCONNECTED && cnt != steals
+        } {
+            loop {
+                match self.queue.pop() {
+                    Some(..) => { steals += 1; }
+                    None => break
+                }
+            }
+        }
+
+        // At this point in time, we have gated all future senders from sending,
+        // and we have flagged the channel as being disconnected. The senders
+        // still have some responsibility, however, because some sends may not
+        // complete until after we flag the disconnection. There are more
+        // details in the sending methods that see DISCONNECTED
+    }
+
+    ////////////////////////////////////////////////////////////////////////////
+    // select implementation
+    ////////////////////////////////////////////////////////////////////////////
+
+    // Tests to see whether this port can receive without blocking. If Ok is
+    // returned, then that's the answer. If Err is returned, then the returned
+    // port needs to be queried instead (an upgrade happened)
+    pub fn can_recv(&mut self) -> Result<bool, Port<T>> {
+        // We peek at the queue to see if there's anything on it, and we use
+        // this return value to determine if we should pop from the queue and
+        // upgrade this channel immediately. If it looks like we've got an
+        // upgrade pending, then go through the whole recv rigamarole to update
+        // the internal state.
+        match self.queue.peek() {
+            Some(&GoUp(..)) => {
+                match self.recv() {
+                    Err(Upgraded(port)) => Err(port),
+                    _ => unreachable!(),
+                }
+            }
+            Some(..) => Ok(true),
+            None => Ok(false)
+        }
+    }
+
+    // Attempts to start selecting on this port. Like a oneshot, this can fail
+    // immediately because of an upgrade.
+    pub fn start_selection(&mut self, task: BlockedTask) -> SelectionResult<T> {
+        match self.decrement(task) {
+            Ok(()) => SelSuccess,
+            Err(task) => {
+                let ret = match self.queue.peek() {
+                    Some(&GoUp(..)) => {
+                        match self.queue.pop() {
+                            Some(GoUp(port)) => SelUpgraded(task, port),
+                            _ => unreachable!(),
+                        }
+                    }
+                    Some(..) => SelCanceled(task),
+                    None => SelCanceled(task),
+                };
+                // Undo our decrement above, and we should be guaranteed that the
+                // previous value is positive because we're not going to sleep
+                let prev = self.cnt.fetch_add(1, atomics::SeqCst);
+                assert!(prev >= 0);
+                return ret;
+            }
+        }
+    }
+
+    // Removes a previous task from being blocked in this port
+    pub fn abort_selection(&mut self,
+                           was_upgrade: bool) -> Result<bool, Port<T>> {
+        // If we're aborting selection after upgrading from a oneshot, then
+        // we're guarantee that no one is waiting. The only way that we could
+        // have seen the upgrade is if data was actually sent on the channel
+        // half again. For us, this means that there is guaranteed to be data on
+        // this channel. Furthermore, we're guaranteed that there was no
+        // start_selection previously, so there's no need to modify `self.cnt`
+        // at all.
+        //
+        // Hence, because of these invariants, we immediately return `Ok(true)`.
+        // Note that the data may not actually be sent on the channel just yet.
+        // The other end could have flagged the upgrade but not sent data to
+        // this end. This is fine because we know it's a small bounded windows
+        // of time until the data is actually sent.
+        if was_upgrade {
+            assert_eq!(self.steals, 0);
+            assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
+            return Ok(true)
+        }
+
+        // We want to make sure that the count on the channel goes non-negative,
+        // and in the stream case we can have at most one steal, so just assume
+        // that we had one steal.
+        let steals = 1;
+        let prev = self.cnt.fetch_add(steals + 1, atomics::SeqCst);
+
+        // If we were previously disconnected, then we know for sure that there
+        // is no task in to_wake, so just keep going
+        let has_data = if prev == DISCONNECTED {
+            assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
+            self.cnt.store(DISCONNECTED, atomics::SeqCst);
+            true // there is data, that data is that we're disconnected
+        } else {
+            let cur = prev + steals + 1;
+            assert!(cur >= 0);
+
+            // If the previous count was negative, then we just made things go
+            // positive, hence we passed the -1 boundary and we're responsible
+            // for removing the to_wake() field and trashing it.
+            //
+            // If the previous count was positive then we're in a tougher
+            // situation. A possible race is that a sender just incremented
+            // through -1 (meaning it's going to try to wake a task up), but it
+            // hasn't yet read the to_wake. In order to prevent a future recv()
+            // from waking up too early (this sender picking up the plastered
+            // over to_wake), we spin loop here waiting for to_wake to be 0.
+            // Note that this entire select() implementation needs an overhaul,
+            // and this is *not* the worst part of it, so this is not done as a
+            // final solution but rather out of necessity for now to get
+            // something working.
+            if prev < 0 {
+                self.take_to_wake().trash();
+            } else {
+                while self.to_wake.load(atomics::SeqCst) != 0 {
+                    Thread::yield_now();
+                }
+            }
+            assert_eq!(self.steals, 0);
+            self.steals = steals;
+
+            // if we were previously positive, then there's surely data to
+            // receive
+            prev >= 0
+        };
+
+        // Now that we've determined that this queue "has data", we peek at the
+        // queue to see if the data is an upgrade or not. If it's an upgrade,
+        // then we need to destroy this port and abort selection on the
+        // upgraded port.
+        if has_data {
+            match self.queue.peek() {
+                Some(&GoUp(..)) => {
+                    match self.queue.pop() {
+                        Some(GoUp(port)) => Err(port),
+                        _ => unreachable!(),
+                    }
+                }
+                _ => Ok(true),
+            }
+        } else {
+            Ok(false)
+        }
+    }
+}
+
+#[unsafe_destructor]
+impl<T: Send> Drop for Packet<T> {
+    fn drop(&mut self) {
+        unsafe {
+            // Note that this load is not only an assert for correctness about
+            // disconnection, but also a proper fence before the read of
+            // `to_wake`, so this assert cannot be removed with also removing
+            // the `to_wake` assert.
+            assert_eq!(self.cnt.load(atomics::SeqCst), DISCONNECTED);
+            assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
+        }
+    }
+}

src/libstd/io/signal.rs

@@ -21,7 +21,7 @@ definitions for a number of signals.
 
 use clone::Clone;
 use result::{Ok, Err};
-use comm::{Port, SharedChan};
+use comm::{Port, Chan};
 use container::{Map, MutableMap};
 use hashmap;
 use io;
@@ -81,7 +81,7 @@ pub struct Listener {
     priv handles: hashmap::HashMap<Signum, ~RtioSignal>,
     /// chan is where all the handles send signums, which are received by
     /// the clients from port.
-    priv chan: SharedChan<Signum>,
+    priv chan: Chan<Signum>,
 
     /// Clients of Listener can `recv()` from this port. This is exposed to
     /// allow selection over this port as well as manipulation of the port
@@ -93,7 +93,7 @@ impl Listener {
     /// Creates a new listener for signals. Once created, signals are bound via
     /// the `register` method (otherwise nothing will ever be received)
     pub fn new() -> Listener {
-        let (port, chan) = SharedChan::new();
+        let (port, chan) = Chan::new();
         Listener {
             chan: chan,
             port: port,

src/libstd/prelude.rs

@@ -80,7 +80,7 @@ pub use vec::{MutableVector, MutableTotalOrdVector};
 pub use vec::{Vector, VectorVector, CloneableVector, ImmutableVector};
 
 // Reexported runtime types
-pub use comm::{Port, Chan, SharedChan};
+pub use comm::{Port, Chan};
 pub use task::spawn;
 
 // Reexported statics

src/libstd/rt/mod.rs

@@ -73,6 +73,7 @@ pub use self::unwind::{begin_unwind, begin_unwind_raw, begin_unwind_fmt};
 // FIXME: these probably shouldn't be public...
 #[doc(hidden)]
 pub mod shouldnt_be_public {
+    #[cfg(not(test))]
     pub use super::local_ptr::native::maybe_tls_key;
     #[cfg(not(windows), not(target_os = "android"))]
     pub use super::local_ptr::compiled::RT_TLS_PTR;

src/libstd/rt/rtio.rs

@@ -10,7 +10,7 @@
 
 use c_str::CString;
 use cast;
-use comm::{SharedChan, Port};
+use comm::{Chan, Port};
 use libc::c_int;
 use libc;
 use ops::Drop;
@@ -181,7 +181,7 @@ pub trait IoFactory {
     fn pipe_open(&mut self, fd: c_int) -> Result<~RtioPipe, IoError>;
     fn tty_open(&mut self, fd: c_int, readable: bool)
             -> Result<~RtioTTY, IoError>;
-    fn signal(&mut self, signal: Signum, channel: SharedChan<Signum>)
+    fn signal(&mut self, signal: Signum, channel: Chan<Signum>)
         -> Result<~RtioSignal, IoError>;
 }
 

src/libstd/rt/task.rs

@@ -449,7 +449,7 @@ mod test {
 
     #[test]
     fn comm_shared_chan() {
-        let (port, chan) = SharedChan::new();
+        let (port, chan) = Chan::new();
         chan.send(10);
         assert!(port.recv() == 10);
     }

src/libstd/rt/unwind.rs

@@ -73,6 +73,7 @@ use unstable::intrinsics;
 
 use uw = self::libunwind;
 
+#[allow(dead_code)]
 mod libunwind {
     //! Unwind library interface
 

src/libstd/run.rs

@@ -13,7 +13,7 @@
 #[allow(missing_doc)];
 #[deny(unused_must_use)];
 
-use comm::SharedChan;
+use comm::Chan;
 use io::Reader;
 use io::process::ProcessExit;
 use io::process;
@@ -225,7 +225,7 @@ impl Process {
         // in parallel so we don't deadlock while blocking on one
         // or the other. FIXME (#2625): Surely there's a much more
         // clever way to do this.
-        let (p, ch) = SharedChan::new();
+        let (p, ch) = Chan::new();
         let ch_clone = ch.clone();
 
         spawn(proc() {

src/libstd/sync/mpmc_bounded_queue.rs

@@ -172,7 +172,7 @@ mod tests {
         let nmsgs = 1000u;
         let mut q = Queue::with_capacity(nthreads*nmsgs);
         assert_eq!(None, q.pop());
-        let (port, chan) = SharedChan::new();
+        let (port, chan) = Chan::new();
 
         for _ in range(0, nthreads) {
             let q = q.clone();

src/libstd/sync/mpsc_queue.rs

@@ -156,14 +156,15 @@ impl<T: Send> Drop for Queue<T> {
 mod tests {
     use prelude::*;
 
-    use super::{Queue, Data, Empty, Inconsistent};
     use native;
+    use super::{Queue, Data, Empty, Inconsistent};
+    use sync::arc::UnsafeArc;
 
     #[test]
     fn test_full() {
         let mut q = Queue::new();
-        p.push(~1);
-        p.push(~2);
+        q.push(~1);
+        q.push(~2);
     }
 
     #[test]
@@ -171,11 +172,11 @@ mod tests {
         let nthreads = 8u;
         let nmsgs = 1000u;
         let mut q = Queue::new();
-        match c.pop() {
+        match q.pop() {
             Empty => {}
             Inconsistent | Data(..) => fail!()
         }
-        let (port, chan) = SharedChan::new();
+        let (port, chan) = Chan::new();
         let q = UnsafeArc::new(q);
 
         for _ in range(0, nthreads) {

src/libstd/sync/spsc_queue.rs

@@ -194,14 +194,16 @@ impl<T: Send> Queue<T> {
         }
     }
 
-    /// Tests whether this queue is empty or not. Remember that there can only
-    /// be one tester/popper, and also keep in mind that the answer returned
-    /// from this is likely to change if it is `false`.
-    pub fn is_empty(&self) -> bool {
+    /// Attempts to peek at the head of the queue, returning `None` if the queue
+    /// has no data currently
+    pub fn peek<'a>(&'a mut self) -> Option<&'a mut T> {
+        // This is essentially the same as above with all the popping bits
+        // stripped out.
         unsafe {
             let tail = self.tail;
             let next = (*tail).next.load(Acquire);
-            return next.is_null();
+            if next.is_null() { return None }
+            return (*next).value.as_mut();
         }
     }
 }
@@ -223,8 +225,9 @@ impl<T: Send> Drop for Queue<T> {
 #[cfg(test)]
 mod test {
     use prelude::*;
-    use super::Queue;
     use native;
+    use super::Queue;
+    use sync::arc::UnsafeArc;
 
     #[test]
     fn smoke() {
@@ -272,7 +275,6 @@ mod test {
             let (a, b) = UnsafeArc::new2(Queue::new(bound));
             let (port, chan) = Chan::new();
             native::task::spawn(proc() {
-                let mut c = c;
                 for _ in range(0, 100000) {
                     loop {
                         match unsafe { (*b.get()).pop() } {

src/libstd/task.rs

@@ -65,7 +65,6 @@ use rt::task::Task;
 use str::{Str, SendStr, IntoMaybeOwned};
 
 #[cfg(test)] use any::{AnyOwnExt, AnyRefExt};
-#[cfg(test)] use comm::SharedChan;
 #[cfg(test)] use ptr;
 #[cfg(test)] use result;
 
@@ -474,9 +473,9 @@ fn test_try_fail() {
 fn test_spawn_sched() {
     use clone::Clone;
 
-    let (po, ch) = SharedChan::new();
+    let (po, ch) = Chan::new();
 
-    fn f(i: int, ch: SharedChan<()>) {
+    fn f(i: int, ch: Chan<()>) {
         let ch = ch.clone();
         spawn(proc() {
             if i == 0 {

src/libsync/sync/mod.rs

@@ -764,7 +764,7 @@ mod tests {
     use std::cast;
     use std::result;
     use std::task;
-    use std::comm::{SharedChan, Empty};
+    use std::comm::Empty;
 
     /************************************************************************
      * Semaphore tests
@@ -1393,7 +1393,7 @@ mod tests {
     #[test]
     fn test_barrier() {
         let barrier = Barrier::new(10);
-        let (port, chan) = SharedChan::new();
+        let (port, chan) = Chan::new();
 
         for _ in range(0, 9) {
             let c = barrier.clone();

src/libsync/sync/mutex.rs

@@ -531,7 +531,7 @@ mod test {
             }
         }
 
-        let (p, c) = SharedChan::new();
+        let (p, c) = Chan::new();
         for _ in range(0, N) {
             let c2 = c.clone();
             native::task::spawn(proc() { inc(); c2.send(()); });

src/libsync/sync/one.rs

@@ -137,7 +137,7 @@ mod test {
         static mut o: Once = ONCE_INIT;
         static mut run: bool = false;
 
-        let (p, c) = SharedChan::new();
+        let (p, c) = Chan::new();
         for _ in range(0, 10) {
             let c = c.clone();
             spawn(proc() {

src/test/bench/msgsend-pipes-shared.rs

@@ -53,7 +53,7 @@ fn server(requests: &Port<request>, responses: &Chan<uint>) {
 
 fn run(args: &[~str]) {
     let (from_child, to_parent) = Chan::new();
-    let (from_parent, to_child) = SharedChan::new();
+    let (from_parent, to_child) = Chan::new();
 
     let size = from_str::<uint>(args[1]).unwrap();
     let workers = from_str::<uint>(args[2]).unwrap();

src/test/bench/msgsend-pipes.rs

@@ -67,7 +67,7 @@ fn run(args: &[~str]) {
         });
         from_parent
     } else {
-        let (from_parent, to_child) = SharedChan::new();
+        let (from_parent, to_child) = Chan::new();
         for _ in range(0u, workers) {
             let to_child = to_child.clone();
             let mut builder = task::task();

src/test/bench/shootout-chameneos-redux.rs

@@ -100,8 +100,8 @@ fn creature(
     name: uint,
     color: color,
     from_rendezvous: Port<Option<CreatureInfo>>,
-    to_rendezvous: SharedChan<CreatureInfo>,
-    to_rendezvous_log: SharedChan<~str>
+    to_rendezvous: Chan<CreatureInfo>,
+    to_rendezvous_log: Chan<~str>
 ) {
     let mut color = color;
     let mut creatures_met = 0;
@@ -137,8 +137,8 @@ fn creature(
 fn rendezvous(nn: uint, set: ~[color]) {
 
     // these ports will allow us to hear from the creatures
-    let (from_creatures, to_rendezvous) = SharedChan::<CreatureInfo>::new();
-    let (from_creatures_log, to_rendezvous_log) = SharedChan::<~str>::new();
+    let (from_creatures, to_rendezvous) = Chan::<CreatureInfo>::new();
+    let (from_creatures_log, to_rendezvous_log) = Chan::<~str>::new();
 
     // these channels will be passed to the creatures so they can talk to us
 

src/test/bench/shootout-pfib.rs

@@ -28,13 +28,13 @@ use std::task;
 use std::uint;
 
 fn fib(n: int) -> int {
-    fn pfib(c: &SharedChan<int>, n: int) {
+    fn pfib(c: &Chan<int>, n: int) {
         if n == 0 {
             c.send(0);
         } else if n <= 2 {
             c.send(1);
         } else {
-            let (pp, cc) = SharedChan::new();
+            let (pp, cc) = Chan::new();
             let ch = cc.clone();
             task::spawn(proc() pfib(&ch, n - 1));
             let ch = cc.clone();
@@ -43,7 +43,7 @@ fn fib(n: int) -> int {
         }
     }
 
-    let (p, ch) = SharedChan::new();
+    let (p, ch) = Chan::new();
     let _t = task::spawn(proc() pfib(&ch, n) );
     p.recv()
 }

src/test/bench/task-perf-linked-failure.rs

@@ -33,7 +33,7 @@
 // Creates in the background 'num_tasks' tasks, all blocked forever.
 // Doesn't return until all such tasks are ready, but doesn't block forever itself.
 
-use std::comm::{stream, SharedChan};
+use std::comm::{stream, Chan};
 use std::os;
 use std::result;
 use std::task;
@@ -41,7 +41,7 @@ use std::uint;
 
 fn grandchild_group(num_tasks: uint) {
     let (po, ch) = stream();
-    let ch = SharedChan::new(ch);
+    let ch = Chan::new(ch);
 
     for _ in range(0, num_tasks) {
         let ch = ch.clone();

src/test/compile-fail/comm-not-freeze.rs

@@ -13,5 +13,5 @@ fn test<T: Freeze>() {}
 fn main() {
     test::<Chan<int>>();        //~ ERROR: does not fulfill `Freeze`
     test::<Port<int>>();        //~ ERROR: does not fulfill `Freeze`
-    test::<SharedChan<int>>();  //~ ERROR: does not fulfill `Freeze`
+    test::<Chan<int>>();  //~ ERROR: does not fulfill `Freeze`
 }

src/test/run-pass/hashmap-memory.rs

@@ -31,7 +31,7 @@ mod map_reduce {
 
     enum ctrl_proto { find_reducer(~[u8], Chan<int>), mapper_done, }
 
-    fn start_mappers(ctrl: SharedChan<ctrl_proto>, inputs: ~[~str]) {
+    fn start_mappers(ctrl: Chan<ctrl_proto>, inputs: ~[~str]) {
         for i in inputs.iter() {
             let ctrl = ctrl.clone();
             let i = i.clone();
@@ -39,11 +39,11 @@ mod map_reduce {
         }
     }
 
-    fn map_task(ctrl: SharedChan<ctrl_proto>, input: ~str) {
+    fn map_task(ctrl: Chan<ctrl_proto>, input: ~str) {
         let mut intermediates = HashMap::new();
 
         fn emit(im: &mut HashMap<~str, int>,
-                ctrl: SharedChan<ctrl_proto>, key: ~str,
+                ctrl: Chan<ctrl_proto>, key: ~str,
                 _val: ~str) {
             if im.contains_key(&key) {
                 return;
@@ -63,7 +63,7 @@ mod map_reduce {
     }
 
     pub fn map_reduce(inputs: ~[~str]) {
-        let (ctrl_port, ctrl_chan) = SharedChan::new();
+        let (ctrl_port, ctrl_chan) = Chan::new();
 
         // This task becomes the master control task. It spawns others
         // to do the rest.

src/test/run-pass/task-comm-14.rs

@@ -13,7 +13,7 @@
 use std::task;
 
 pub fn main() {
-    let (po, ch) = SharedChan::new();
+    let (po, ch) = Chan::new();
 
     // Spawn 10 tasks each sending us back one int.
     let mut i = 10;
@@ -37,7 +37,7 @@ pub fn main() {
     info!("main thread exiting");
 }
 
-fn child(x: int, ch: &SharedChan<int>) {
+fn child(x: int, ch: &Chan<int>) {
     info!("{}", x);
     ch.send(x);
 }

src/test/run-pass/task-comm-3.rs

@@ -16,7 +16,7 @@ use std::task;
 
 pub fn main() { info!("===== WITHOUT THREADS ====="); test00(); }
 
-fn test00_start(ch: &SharedChan<int>, message: int, count: int) {
+fn test00_start(ch: &Chan<int>, message: int, count: int) {
     info!("Starting test00_start");
     let mut i: int = 0;
     while i < count {
@@ -33,7 +33,7 @@ fn test00() {
 
     info!("Creating tasks");
 
-    let (po, ch) = SharedChan::new();
+    let (po, ch) = Chan::new();
 
     let mut i: int = 0;
 

src/test/run-pass/task-comm-6.rs

@@ -15,7 +15,7 @@ pub fn main() { test00(); }
 fn test00() {
     let mut r: int = 0;
     let mut sum: int = 0;
-    let (p, ch) = SharedChan::new();
+    let (p, ch) = Chan::new();
     let mut c0 = ch.clone();
     let mut c1 = ch.clone();
     let mut c2 = ch.clone();

src/test/run-pass/task-comm-7.rs

@@ -18,7 +18,7 @@ use std::task;
 
 pub fn main() { test00(); }
 
-fn test00_start(c: &SharedChan<int>, start: int,
+fn test00_start(c: &Chan<int>, start: int,
                 number_of_messages: int) {
     let mut i: int = 0;
     while i < number_of_messages { c.send(start + i); i += 1; }
@@ -27,7 +27,7 @@ fn test00_start(c: &SharedChan<int>, start: int,
 fn test00() {
     let mut r: int = 0;
     let mut sum: int = 0;
-    let (p, ch) = SharedChan::new();
+    let (p, ch) = Chan::new();
     let number_of_messages: int = 10;
 
     let c = ch.clone();

src/test/run-pass/unique-send-2.rs

@@ -10,12 +10,12 @@
 
 use std::task;
 
-fn child(c: &SharedChan<~uint>, i: uint) {
+fn child(c: &Chan<~uint>, i: uint) {
     c.send(~i);
 }
 
 pub fn main() {
-    let (p, ch) = SharedChan::new();
+    let (p, ch) = Chan::new();
     let n = 100u;
     let mut expected = 0u;
     for i in range(0u, n) {

src/test/run-pass/unwind-resource.rs

@@ -15,7 +15,7 @@ extern mod extra;
 use std::task;
 
 struct complainer {
-  c: SharedChan<bool>,
+  c: Chan<bool>,
 }
 
 impl Drop for complainer {
@@ -26,20 +26,20 @@ impl Drop for complainer {
     }
 }
 
-fn complainer(c: SharedChan<bool>) -> complainer {
+fn complainer(c: Chan<bool>) -> complainer {
     error!("Hello!");
     complainer {
         c: c
     }
 }
 
-fn f(c: SharedChan<bool>) {
+fn f(c: Chan<bool>) {
     let _c = complainer(c);
     fail!();
 }
 
 pub fn main() {
-    let (p, c) = SharedChan::new();
+    let (p, c) = Chan::new();
     task::spawn(proc() f(c.clone()));
     error!("hiiiiiiiii");
     assert!(p.recv());