Rewrite std::comm

* Streams are now ~3x faster than before (fewer allocations and more optimized)
    * Based on a single-producer single-consumer lock-free queue that doesn't
      always have to allocate on every send.
    * Blocking via mutexes/cond vars outside the runtime
* Streams work in/out of the runtime seamlessly
* Select now works in/out of the runtime seamlessly
* Streams will now fail!() on send() if the other end has hung up
    * try_send() will not fail
* PortOne/ChanOne removed
* SharedPort removed
* MegaPipe removed
* Generic select removed (only one kind of port now)
* API redesign
    * try_recv == never block
    * recv_opt == block, don't fail
    * iter() == Iterator<T> for Port<T>
    * removed peek
    * Type::new
* Removed rt::comm

4 files changed, 423 insertions, 122 deletions

diff --git a/src/libstd/rt/mpmc_bounded_queue.rs b/src/libstd/rt/mpmc_bounded_queue.rs
index 7f607fcf12a..1e04e5eb78d 100644
--- a/src/libstd/rt/mpmc_bounded_queue.rs
+++ b/src/libstd/rt/mpmc_bounded_queue.rs
@@ -1,5 +1,4 @@
-/* Multi-producer/multi-consumer bounded queue
- * Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
+/* Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are met:
  *
@@ -163,7 +162,6 @@ mod tests {
     use prelude::*;
     use option::*;
     use task;
-    use comm;
     use super::Queue;
 
     #[test]
@@ -174,10 +172,9 @@ mod tests {
         assert_eq!(None, q.pop());
 
         for _ in range(0, nthreads) {
-            let (port, chan)  = comm::stream();
-            chan.send(q.clone());
+            let q = q.clone();
             do task::spawn_sched(task::SingleThreaded) {
-                let mut q = port.recv();
+                let mut q = q;
                 for i in range(0, nmsgs) {
                     assert!(q.push(i));
                 }
@@ -186,12 +183,11 @@ mod tests {
 
         let mut completion_ports = ~[];
         for _ in range(0, nthreads) {
-            let (completion_port, completion_chan) = comm::stream();
+            let (completion_port, completion_chan) = Chan::new();
             completion_ports.push(completion_port);
-            let (port, chan)  = comm::stream();
-            chan.send(q.clone());
+            let q = q.clone();
             do task::spawn_sched(task::SingleThreaded) {
-                let mut q = port.recv();
+                let mut q = q;
                 let mut i = 0u;
                 loop {
                     match q.pop() {
@@ -206,7 +202,7 @@ mod tests {
             }
         }
 
-        for completion_port in completion_ports.iter() {
+        for completion_port in completion_ports.mut_iter() {
             assert_eq!(nmsgs, completion_port.recv());
         }
     }
diff --git a/src/libstd/rt/mpsc_queue.rs b/src/libstd/rt/mpsc_queue.rs
index 4f39a1df4fa..d575028af70 100644
--- a/src/libstd/rt/mpsc_queue.rs
+++ b/src/libstd/rt/mpsc_queue.rs
@@ -1,5 +1,4 @@
-/* Multi-producer/single-consumer queue
- * Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
+/* Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are met:
  *
@@ -27,163 +26,177 @@
  */
 
 //! A mostly lock-free multi-producer, single consumer queue.
-// http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue
 
-use unstable::sync::UnsafeArc;
-use unstable::atomics::{AtomicPtr,Relaxed,Release,Acquire};
-use ptr::{mut_null, to_mut_unsafe_ptr};
+// http://www.1024cores.net/home/lock-free-algorithms
+//                         /queues/non-intrusive-mpsc-node-based-queue
+
 use cast;
-use option::*;
 use clone::Clone;
 use kinds::Send;
+use ops::Drop;
+use option::{Option, None, Some};
+use unstable::atomics::{AtomicPtr, Release, Acquire, AcqRel, Relaxed};
+use unstable::sync::UnsafeArc;
+
+pub enum PopResult<T> {
+    /// Some data has been popped
+    Data(T),
+    /// The queue is empty
+    Empty,
+    /// The queue is in an inconsistent state. Popping data should succeed, but
+    /// some pushers have yet to make enough progress in order allow a pop to
+    /// succeed. It is recommended that a pop() occur "in the near future" in
+    /// order to see if the sender has made progress or not
+    Inconsistent,
+}
 
 struct Node<T> {
     next: AtomicPtr<Node<T>>,
     value: Option<T>,
 }
 
-impl<T> Node<T> {
-    fn empty() -> Node<T> {
-        Node{next: AtomicPtr::new(mut_null()), value: None}
-    }
-
-    fn with_value(value: T) -> Node<T> {
-        Node{next: AtomicPtr::new(mut_null()), value: Some(value)}
-    }
-}
-
-struct State<T> {
-    pad0: [u8, ..64],
+struct State<T, P> {
     head: AtomicPtr<Node<T>>,
-    pad1: [u8, ..64],
-    stub: Node<T>,
-    pad2: [u8, ..64],
     tail: *mut Node<T>,
-    pad3: [u8, ..64],
+    packet: P,
 }
 
-struct Queue<T> {
-    priv state: UnsafeArc<State<T>>,
+pub struct Consumer<T, P> {
+    priv state: UnsafeArc<State<T, P>>,
 }
 
-impl<T: Send> Clone for Queue<T> {
-    fn clone(&self) -> Queue<T> {
-        Queue {
-            state: self.state.clone()
-        }
-    }
+pub struct Producer<T, P> {
+    priv state: UnsafeArc<State<T, P>>,
 }
 
-impl<T: Send> State<T> {
-    pub fn new() -> State<T> {
-        State{
-            pad0: [0, ..64],
-            head: AtomicPtr::new(mut_null()),
-            pad1: [0, ..64],
-            stub: Node::<T>::empty(),
-            pad2: [0, ..64],
-            tail: mut_null(),
-            pad3: [0, ..64],
-        }
+impl<T: Send, P: Send> Clone for Producer<T, P> {
+    fn clone(&self) -> Producer<T, P> {
+        Producer { state: self.state.clone() }
     }
+}
 
-    fn init(&mut self) {
-        let stub = self.get_stub_unsafe();
-        self.head.store(stub, Relaxed);
-        self.tail = stub;
+pub fn queue<T: Send, P: Send>(p: P) -> (Consumer<T, P>, Producer<T, P>) {
+    unsafe {
+        let (a, b) = UnsafeArc::new2(State::new(p));
+        (Consumer { state: a }, Producer { state: b })
     }
+}
 
-    fn get_stub_unsafe(&mut self) -> *mut Node<T> {
-        to_mut_unsafe_ptr(&mut self.stub)
+impl<T> Node<T> {
+    unsafe fn new(v: Option<T>) -> *mut Node<T> {
+        cast::transmute(~Node {
+            next: AtomicPtr::new(0 as *mut Node<T>),
+            value: v,
+        })
     }
+}
 
-    fn push(&mut self, value: T) {
-        unsafe {
-            let node = cast::transmute(~Node::with_value(value));
-            self.push_node(node);
+impl<T: Send, P: Send> State<T, P> {
+    pub unsafe fn new(p: P) -> State<T, P> {
+        let stub = Node::new(None);
+        State {
+            head: AtomicPtr::new(stub),
+            tail: stub,
+            packet: p,
         }
     }
 
-    fn push_node(&mut self, node: *mut Node<T>) {
-        unsafe {
-            (*node).next.store(mut_null(), Release);
-            let prev = self.head.swap(node, Relaxed);
-            (*prev).next.store(node, Release);
-        }
+    unsafe fn push(&mut self, t: T) {
+        let n = Node::new(Some(t));
+        let prev = self.head.swap(n, AcqRel);
+        (*prev).next.store(n, Release);
     }
 
-    fn pop(&mut self) -> Option<T> {
-        unsafe {
-            let mut tail = self.tail;
-            let mut next = (*tail).next.load(Acquire);
-            let stub = self.get_stub_unsafe();
-            if tail == stub {
-                if mut_null() == next {
-                    return None
-                }
-                self.tail = next;
-                tail = next;
-                next = (*next).next.load(Acquire);
-            }
-            if next != mut_null() {
-                let tail: ~Node<T> = cast::transmute(tail);
-                self.tail = next;
-                return tail.value
-            }
-            let head = self.head.load(Relaxed);
-            if tail != head {
-                return None
-            }
-            self.push_node(stub);
-            next = (*tail).next.load(Acquire);
-            if next != mut_null() {
-                let tail: ~Node<T> = cast::transmute(tail);
-                self.tail = next;
-                return tail.value
-            }
+    unsafe fn pop(&mut self) -> PopResult<T> {
+        let tail = self.tail;
+        let next = (*tail).next.load(Acquire);
+
+        if !next.is_null() {
+            self.tail = next;
+            assert!((*tail).value.is_none());
+            assert!((*next).value.is_some());
+            let ret = (*next).value.take_unwrap();
+            let _: ~Node<T> = cast::transmute(tail);
+            return Data(ret);
         }
-        None
+
+        if self.head.load(Acquire) == tail {Empty} else {Inconsistent}
+    }
+
+    unsafe fn is_empty(&mut self) -> bool {
+        return (*self.tail).next.load(Acquire).is_null();
     }
 }
 
-impl<T: Send> Queue<T> {
-    pub fn new() -> Queue<T> {
+#[unsafe_destructor]
+impl<T: Send, P: Send> Drop for State<T, P> {
+    fn drop(&mut self) {
         unsafe {
-            let q = Queue{state: UnsafeArc::new(State::new())};
-            (*q.state.get()).init();
-            q
+            let mut cur = self.tail;
+            while !cur.is_null() {
+                let next = (*cur).next.load(Relaxed);
+                let _: ~Node<T> = cast::transmute(cur);
+                cur = next;
+            }
         }
     }
+}
 
+impl<T: Send, P: Send> Producer<T, P> {
     pub fn push(&mut self, value: T) {
         unsafe { (*self.state.get()).push(value) }
     }
+    pub fn is_empty(&self) -> bool {
+        unsafe{ (*self.state.get()).is_empty() }
+    }
+    pub unsafe fn packet(&self) -> *mut P {
+        &mut (*self.state.get()).packet as *mut P
+    }
+}
 
-    pub fn pop(&mut self) -> Option<T> {
-        unsafe{ (*self.state.get()).pop() }
+impl<T: Send, P: Send> Consumer<T, P> {
+    pub fn pop(&mut self) -> PopResult<T> {
+        unsafe { (*self.state.get()).pop() }
+    }
+    pub fn casual_pop(&mut self) -> Option<T> {
+        match self.pop() {
+            Data(t) => Some(t),
+            Empty | Inconsistent => None,
+        }
+    }
+    pub unsafe fn packet(&self) -> *mut P {
+        &mut (*self.state.get()).packet as *mut P
     }
 }
 
 #[cfg(test)]
 mod tests {
     use prelude::*;
-    use option::*;
+
     use task;
-    use comm;
-    use super::Queue;
+    use super::{queue, Data, Empty, Inconsistent};
+
+    #[test]
+    fn test_full() {
+        let (_, mut p) = queue(());
+        p.push(~1);
+        p.push(~2);
+    }
 
     #[test]
     fn test() {
         let nthreads = 8u;
         let nmsgs = 1000u;
-        let mut q = Queue::new();
-        assert_eq!(None, q.pop());
+        let (mut c, p) = queue(());
+        match c.pop() {
+            Empty => {}
+            Inconsistent | Data(..) => fail!()
+        }
 
         for _ in range(0, nthreads) {
-            let (port, chan)  = comm::stream();
-            chan.send(q.clone());
+            let q = p.clone();
             do task::spawn_sched(task::SingleThreaded) {
-                let mut q = port.recv();
+                let mut q = q;
                 for i in range(0, nmsgs) {
                     q.push(i);
                 }
@@ -191,13 +204,10 @@ mod tests {
         }
 
         let mut i = 0u;
-        loop {
-            match q.pop() {
-                None => {},
-                Some(_) => {
-                    i += 1;
-                    if i == nthreads*nmsgs { break }
-                }
+        while i < nthreads * nmsgs {
+            match c.pop() {
+                Empty | Inconsistent => {},
+                Data(_) => { i += 1 }
             }
         }
     }
diff --git a/src/libstd/rt/spsc_queue.rs b/src/libstd/rt/spsc_queue.rs
new file mode 100644
index 00000000000..f14533d726a
--- /dev/null
+++ b/src/libstd/rt/spsc_queue.rs
@@ -0,0 +1,296 @@
+/* Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *    1. Redistributions of source code must retain the above copyright notice,
+ *       this list of conditions and the following disclaimer.
+ *
+ *    2. Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in the
+ *       documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED
+ * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT
+ * SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
+ * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
+ * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * The views and conclusions contained in the software and documentation are
+ * those of the authors and should not be interpreted as representing official
+ * policies, either expressed or implied, of Dmitry Vyukov.
+ */
+
+// http://www.1024cores.net/home/lock-free-algorithms/queues/unbounded-spsc-queue
+use cast;
+use kinds::Send;
+use ops::Drop;
+use option::{Some, None, Option};
+use unstable::atomics::{AtomicPtr, Relaxed, AtomicUint, Acquire, Release};
+use unstable::sync::UnsafeArc;
+
+// Node within the linked list queue of messages to send
+struct Node<T> {
+    // XXX: this could be an uninitialized T if we're careful enough, and
+    //      that would reduce memory usage (and be a bit faster).
+    //      is it worth it?
+    value: Option<T>,           // nullable for re-use of nodes
+    next: AtomicPtr<Node<T>>,   // next node in the queue
+}
+
+// The producer/consumer halves both need access to the `tail` field, and if
+// they both have access to that we may as well just give them both access
+// to this whole structure.
+struct State<T, P> {
+    // consumer fields
+    tail: *mut Node<T>, // where to pop from
+    tail_prev: AtomicPtr<Node<T>>, // where to pop from
+
+    // producer fields
+    head: *mut Node<T>,      // where to push to
+    first: *mut Node<T>,     // where to get new nodes from
+    tail_copy: *mut Node<T>, // between first/tail
+
+    // Cache maintenance fields. Additions and subtractions are stored
+    // separately in order to allow them to use nonatomic addition/subtraction.
+    cache_bound: uint,
+    cache_additions: AtomicUint,
+    cache_subtractions: AtomicUint,
+
+    packet: P,
+}
+
+pub struct Producer<T, P> {
+    priv state: UnsafeArc<State<T, P>>,
+}
+
+pub struct Consumer<T, P> {
+    priv state: UnsafeArc<State<T, P>>,
+}
+
+pub fn queue<T: Send, P: Send>(bound: uint,
+                               p: P) -> (Consumer<T, P>, Producer<T, P>)
+{
+    let n1 = Node::new();
+    let n2 = Node::new();
+    unsafe { (*n1).next.store(n2, Relaxed) }
+    let state = State {
+        tail: n2,
+        tail_prev: AtomicPtr::new(n1),
+        head: n2,
+        first: n1,
+        tail_copy: n1,
+        cache_bound: bound,
+        cache_additions: AtomicUint::new(0),
+        cache_subtractions: AtomicUint::new(0),
+        packet: p,
+    };
+    let (arc1, arc2) = UnsafeArc::new2(state);
+    (Consumer { state: arc1 }, Producer { state: arc2 })
+}
+
+impl<T: Send> Node<T> {
+    fn new() -> *mut Node<T> {
+        unsafe {
+            cast::transmute(~Node {
+                value: None,
+                next: AtomicPtr::new(0 as *mut Node<T>),
+            })
+        }
+    }
+}
+
+impl<T: Send, P: Send> Producer<T, P> {
+    pub fn push(&mut self, t: T) {
+        unsafe { (*self.state.get()).push(t) }
+    }
+    pub fn is_empty(&self) -> bool {
+        unsafe { (*self.state.get()).is_empty() }
+    }
+    pub unsafe fn packet(&self) -> *mut P {
+        &mut (*self.state.get()).packet as *mut P
+    }
+}
+
+impl<T: Send, P: Send> Consumer<T, P> {
+    pub fn pop(&mut self) -> Option<T> {
+        unsafe { (*self.state.get()).pop() }
+    }
+    pub unsafe fn packet(&self) -> *mut P {
+        &mut (*self.state.get()).packet as *mut P
+    }
+}
+
+impl<T: Send, P: Send> State<T, P> {
+    // remember that there is only one thread executing `push` (and only one
+    // thread executing `pop`)
+    unsafe fn push(&mut self, t: T) {
+        // Acquire a node (which either uses a cached one or allocates a new
+        // one), and then append this to the 'head' node.
+        let n = self.alloc();
+        assert!((*n).value.is_none());
+        (*n).value = Some(t);
+        (*n).next.store(0 as *mut Node<T>, Relaxed);
+        (*self.head).next.store(n, Release);
+        self.head = n;
+    }
+
+    unsafe fn alloc(&mut self) -> *mut Node<T> {
+        // First try to see if we can consume the 'first' node for our uses.
+        // We try to avoid as many atomic instructions as possible here, so
+        // the addition to cache_subtractions is not atomic (plus we're the
+        // only one subtracting from the cache).
+        if self.first != self.tail_copy {
+            if self.cache_bound > 0 {
+                let b = self.cache_subtractions.load(Relaxed);
+                self.cache_subtractions.store(b + 1, Relaxed);
+            }
+            let ret = self.first;
+            self.first = (*ret).next.load(Relaxed);
+            return ret;
+        }
+        // If the above fails, then update our copy of the tail and try
+        // again.
+        self.tail_copy = self.tail_prev.load(Acquire);
+        if self.first != self.tail_copy {
+            if self.cache_bound > 0 {
+                let b = self.cache_subtractions.load(Relaxed);
+                self.cache_subtractions.store(b + 1, Relaxed);
+            }
+            let ret = self.first;
+            self.first = (*ret).next.load(Relaxed);
+            return ret;
+        }
+        // If all of that fails, then we have to allocate a new node
+        // (there's nothing in the node cache).
+        Node::new()
+    }
+
+    // remember that there is only one thread executing `pop` (and only one
+    // thread executing `push`)
+    unsafe fn pop(&mut self) -> Option<T> {
+        // The `tail` node is not actually a used node, but rather a
+        // sentinel from where we should start popping from. Hence, look at
+        // tail's next field and see if we can use it. If we do a pop, then
+        // the current tail node is a candidate for going into the cache.
+        let tail = self.tail;
+        let next = (*tail).next.load(Acquire);
+        if next.is_null() { return None }
+        assert!((*next).value.is_some());
+        let ret = (*next).value.take();
+
+        self.tail = next;
+        if self.cache_bound == 0 {
+            self.tail_prev.store(tail, Release);
+        } else {
+            // XXX: this is dubious with overflow.
+            let additions = self.cache_additions.load(Relaxed);
+            let subtractions = self.cache_subtractions.load(Relaxed);
+            let size = additions - subtractions;
+
+            if size < self.cache_bound {
+                self.tail_prev.store(tail, Release);
+                self.cache_additions.store(additions + 1, Relaxed);
+            } else {
+                (*self.tail_prev.load(Relaxed)).next.store(next, Relaxed);
+                // We have successfully erased all references to 'tail', so
+                // now we can safely drop it.
+                let _: ~Node<T> = cast::transmute(tail);
+            }
+        }
+        return ret;
+    }
+
+    unsafe fn is_empty(&self) -> bool {
+        let tail = self.tail;
+        let next = (*tail).next.load(Acquire);
+        return next.is_null();
+    }
+}
+
+#[unsafe_destructor]
+impl<T: Send, P: Send> Drop for State<T, P> {
+    fn drop(&mut self) {
+        unsafe {
+            let mut cur = self.first;
+            while !cur.is_null() {
+                let next = (*cur).next.load(Relaxed);
+                let _n: ~Node<T> = cast::transmute(cur);
+                cur = next;
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use prelude::*;
+    use super::queue;
+    use task;
+
+    #[test]
+    fn smoke() {
+        let (mut c, mut p) = queue(0, ());
+        p.push(1);
+        p.push(2);
+        assert_eq!(c.pop(), Some(1));
+        assert_eq!(c.pop(), Some(2));
+        assert_eq!(c.pop(), None);
+        p.push(3);
+        p.push(4);
+        assert_eq!(c.pop(), Some(3));
+        assert_eq!(c.pop(), Some(4));
+        assert_eq!(c.pop(), None);
+    }
+
+    #[test]
+    fn drop_full() {
+        let (_, mut p) = queue(0, ());
+        p.push(~1);
+        p.push(~2);
+    }
+
+    #[test]
+    fn smoke_bound() {
+        let (mut c, mut p) = queue(1, ());
+        p.push(1);
+        p.push(2);
+        assert_eq!(c.pop(), Some(1));
+        assert_eq!(c.pop(), Some(2));
+        assert_eq!(c.pop(), None);
+        p.push(3);
+        p.push(4);
+        assert_eq!(c.pop(), Some(3));
+        assert_eq!(c.pop(), Some(4));
+        assert_eq!(c.pop(), None);
+    }
+
+    #[test]
+    fn stress() {
+        stress_bound(0);
+        stress_bound(1);
+
+        fn stress_bound(bound: uint) {
+            let (c, mut p) = queue(bound, ());
+            do task::spawn_sched(task::SingleThreaded) {
+                let mut c = c;
+                for _ in range(0, 100000) {
+                    loop {
+                        match c.pop() {
+                            Some(1) => break,
+                            Some(_) => fail!(),
+                            None => {}
+                        }
+                    }
+                }
+            }
+            for _ in range(0, 100000) {
+                p.push(1);
+            }
+        }
+    }
+}
diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs
index 86cc895eb27..2adc32f33fb 100644
--- a/src/libstd/rt/task.rs
+++ b/src/libstd/rt/task.rs
@@ -26,7 +26,6 @@ use option::{Option, Some, None};
 use rt::borrowck::BorrowRecord;
 use rt::borrowck;
 use rt::context::Context;
-use rt::context;
 use rt::env;
 use io::Writer;
 use rt::kill::Death;