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.

3 files changed, 16 insertions, 13 deletions

diff --git a/src/libstd/sync/mpmc_bounded_queue.rs b/src/libstd/sync/mpmc_bounded_queue.rs
index 74f3a6f6918..44825a1ef94 100644
--- a/src/libstd/sync/mpmc_bounded_queue.rs
+++ b/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();
diff --git a/src/libstd/sync/mpsc_queue.rs b/src/libstd/sync/mpsc_queue.rs
index 258162069d9..b5a55f3f8c9 100644
--- a/src/libstd/sync/mpsc_queue.rs
+++ b/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) {
diff --git a/src/libstd/sync/spsc_queue.rs b/src/libstd/sync/spsc_queue.rs
index d1fde759cc1..a2c61a2b135 100644
--- a/src/libstd/sync/spsc_queue.rs
+++ b/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() } {