Fallout of rewriting std::comm

Author: alexcrichton

src/etc/licenseck.py

@@ -77,6 +77,7 @@
     "rt/isaac/rand.h", # public domain
     "rt/isaac/standard.h", # public domain
     "libstd/rt/mpsc_queue.rs", # BSD
+    "libstd/rt/spsc_queue.rs", # BSD
     "libstd/rt/mpmc_bounded_queue.rs", # BSD
 ]
 

src/libextra/arc.rs

@@ -597,15 +597,14 @@ mod tests {
 
     use arc::*;
 
-    use std::comm;
     use std::task;
 
     #[test]
     fn manually_share_arc() {
         let v = ~[1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
         let arc_v = Arc::new(v);
 
-        let (p, c) = comm::stream();
+        let (p, c) = Chan::new();
 
         do task::spawn {
             let arc_v: Arc<~[int]> = p.recv();
@@ -626,7 +625,7 @@ mod tests {
     fn test_mutex_arc_condvar() {
         let arc = ~MutexArc::new(false);
         let arc2 = ~arc.clone();
-        let (p,c) = comm::oneshot();
+        let (p,c) = Chan::new();
         do task::spawn {
             // wait until parent gets in
             p.recv();
@@ -638,7 +637,7 @@ mod tests {
 
         let mut c = Some(c);
         arc.access_cond(|state, cond| {
-            c.take_unwrap().send(());
+            c.take_unwrawp().send(());
             assert!(!*state);
             while !*state {
                 cond.wait();
@@ -650,7 +649,7 @@ mod tests {
     fn test_arc_condvar_poison() {
         let arc = ~MutexArc::new(1);
         let arc2 = ~arc.clone();
-        let (p, c) = comm::stream();
+        let (p, c) = Chan::new();
 
         do spawn {
             let _ = p.recv();
@@ -687,7 +686,7 @@ mod tests {
     pub fn test_mutex_arc_unwrap_poison() {
         let arc = MutexArc::new(1);
         let arc2 = ~(&arc).clone();
-        let (p, c) = comm::stream();
+        let (p, c) = Chan::new();
         do task::spawn {
             arc2.access(|one| {
                 c.send(());
@@ -804,7 +803,7 @@ mod tests {
     fn test_rw_arc() {
         let arc = RWArc::new(0);
         let arc2 = arc.clone();
-        let (p, c) = comm::stream();
+        let (p, c) = Chan::new();
 
         do task::spawn {
             arc2.write(|num| {
@@ -832,7 +831,7 @@ mod tests {
         });
 
         // Wait for children to pass their asserts
-        for r in children.iter() {
+        for r in children.mut_iter() {
             r.recv();
         }
 
@@ -855,7 +854,7 @@ mod tests {
         // Reader tasks
         let mut reader_convos = ~[];
         10.times(|| {
-            let ((rp1, rc1), (rp2, rc2)) = (comm::stream(), comm::stream());
+            let ((rp1, rc1), (rp2, rc2)) = (Chan::new(), Chan::new());
             reader_convos.push((rc1, rp2));
             let arcn = arc.clone();
             do task::spawn {
@@ -869,7 +868,7 @@ mod tests {
 
         // Writer task
         let arc2 = arc.clone();
-        let ((wp1, wc1), (wp2, wc2)) = (comm::stream(), comm::stream());
+        let ((wp1, wc1), (wp2, wc2)) = (Chan::new(), Chan::new());
         do task::spawn || {
             wp1.recv();
             arc2.write_cond(|state, cond| {
@@ -897,14 +896,14 @@ mod tests {
                 assert_eq!(*state, 42);
                 *state = 31337;
                 // send to other readers
-                for &(ref rc, _) in reader_convos.iter() {
+                for &(ref mut rc, _) in reader_convos.mut_iter() {
                     rc.send(())
                 }
             });
             let read_mode = arc.downgrade(write_mode);
             read_mode.read(|state| {
                 // complete handshake with other readers
-                for &(_, ref rp) in reader_convos.iter() {
+                for &(_, ref mut rp) in reader_convos.mut_iter() {
                     rp.recv()
                 }
                 wc1.send(()); // tell writer to try again
@@ -926,7 +925,7 @@ mod tests {
         //     "blk(&Condvar { order: opt_lock, ..*cond })"
         // with just "blk(cond)".
         let x = RWArc::new(true);
-        let (wp, wc) = comm::stream();
+        let (wp, wc) = Chan::new();
 
         // writer task
         let xw = x.clone();
@@ -951,7 +950,7 @@ mod tests {
             });
             // make a reader task to trigger the "reader cloud lock" handoff
             let xr = x.clone();
-            let (rp, rc) = comm::stream();
+            let (rp, rc) = Chan::new();
             do task::spawn {
                 rc.send(());
                 xr.read(|_state| { })

src/libextra/comm.rs

@@ -16,11 +16,6 @@ Higher level communication abstractions.
 
 #[allow(missing_doc)];
 
-
-use std::comm::{GenericChan, GenericSmartChan, GenericPort};
-use std::comm::{Chan, Port, Peekable};
-use std::comm;
-
 /// An extension of `pipes::stream` that allows both sending and receiving.
 pub struct DuplexStream<T, U> {
     priv chan: Chan<T>,
@@ -29,108 +24,73 @@ pub struct DuplexStream<T, U> {
 
 // Allow these methods to be used without import:
 impl<T:Send,U:Send> DuplexStream<T, U> {
+    /// Creates a bidirectional stream.
+    pub fn new() -> (DuplexStream<T, U>, DuplexStream<U, T>) {
+        let (p1, c2) = Chan::new();
+        let (p2, c1) = Chan::new();
+        (DuplexStream { chan: c1, port: p1 },
+         DuplexStream { chan: c2, port: p2 })
+    }
     pub fn send(&self, x: T) {
         self.chan.send(x)
     }
     pub fn try_send(&self, x: T) -> bool {
         self.chan.try_send(x)
     }
-    pub fn recv(&self, ) -> U {
+    pub fn recv(&self) -> U {
         self.port.recv()
     }
     pub fn try_recv(&self) -> Option<U> {
         self.port.try_recv()
     }
-    pub fn peek(&self) -> bool {
-        self.port.peek()
-    }
-}
-
-impl<T:Send,U:Send> GenericChan<T> for DuplexStream<T, U> {
-    fn send(&self, x: T) {
-        self.chan.send(x)
-    }
-}
-
-impl<T:Send,U:Send> GenericSmartChan<T> for DuplexStream<T, U> {
-    fn try_send(&self, x: T) -> bool {
-        self.chan.try_send(x)
-    }
-}
-
-impl<T:Send,U:Send> GenericPort<U> for DuplexStream<T, U> {
-    fn recv(&self) -> U {
-        self.port.recv()
-    }
-
-    fn try_recv(&self) -> Option<U> {
-        self.port.try_recv()
-    }
-}
-
-impl<T:Send,U:Send> Peekable<U> for DuplexStream<T, U> {
-    fn peek(&self) -> bool {
-        self.port.peek()
+    pub fn recv_opt(&self) -> Option<U> {
+        self.port.recv_opt()
     }
 }
 
-/// Creates a bidirectional stream.
-pub fn DuplexStream<T:Send,U:Send>()
-    -> (DuplexStream<T, U>, DuplexStream<U, T>)
-{
-    let (p1, c2) = comm::stream();
-    let (p2, c1) = comm::stream();
-    (DuplexStream {
-        chan: c1,
-        port: p1
-    },
-     DuplexStream {
-         chan: c2,
-         port: p2
-     })
-}
-
 /// An extension of `pipes::stream` that provides synchronous message sending.
 pub struct SyncChan<T> { priv duplex_stream: DuplexStream<T, ()> }
 /// An extension of `pipes::stream` that acknowledges each message received.
 pub struct SyncPort<T> { priv duplex_stream: DuplexStream<(), T> }
 
-impl<T: Send> GenericChan<T> for SyncChan<T> {
-    fn send(&self, val: T) {
+impl<T: Send> SyncChan<T> {
+    pub fn send(&self, val: T) {
         assert!(self.try_send(val), "SyncChan.send: receiving port closed");
     }
-}
 
-impl<T: Send> GenericSmartChan<T> for SyncChan<T> {
-    /// Sends a message, or report if the receiver has closed the connection before receiving.
-    fn try_send(&self, val: T) -> bool {
-        self.duplex_stream.try_send(val) && self.duplex_stream.try_recv().is_some()
+    /// Sends a message, or report if the receiver has closed the connection
+    /// before receiving.
+    pub fn try_send(&self, val: T) -> bool {
+        self.duplex_stream.try_send(val) && self.duplex_stream.recv_opt().is_some()
     }
 }
 
-impl<T: Send> GenericPort<T> for SyncPort<T> {
-    fn recv(&self) -> T {
-        self.try_recv().expect("SyncPort.recv: sending channel closed")
+impl<T: Send> SyncPort<T> {
+    pub fn recv(&self) -> T {
+        self.recv_opt().expect("SyncPort.recv: sending channel closed")
     }
 
-    fn try_recv(&self) -> Option<T> {
-        self.duplex_stream.try_recv().map(|val| {
+    pub fn recv_opt(&self) -> Option<T> {
+        self.duplex_stream.recv_opt().map(|val| {
             self.duplex_stream.try_send(());
             val
         })
     }
-}
 
-impl<T: Send> Peekable<T> for SyncPort<T> {
-    fn peek(&self) -> bool {
-        self.duplex_stream.peek()
+    pub fn try_recv(&self) -> Option<T> {
+        self.duplex_stream.try_recv().map(|val| {
+            self.duplex_stream.try_send(());
+            val
+        })
     }
 }
 
-/// Creates a stream whose channel, upon sending a message, blocks until the message is received.
+/// Creates a stream whose channel, upon sending a message, blocks until the
+/// message is received.
 pub fn rendezvous<T: Send>() -> (SyncPort<T>, SyncChan<T>) {
-    let (chan_stream, port_stream) = DuplexStream();
-    (SyncPort { duplex_stream: port_stream }, SyncChan { duplex_stream: chan_stream })
+    let (chan_stream, port_stream) = DuplexStream::new();
+    (SyncPort { duplex_stream: port_stream },
+     SyncChan { duplex_stream: chan_stream })
 }
 
 #[cfg(test)]
@@ -141,7 +101,7 @@ mod test {
 
     #[test]
     pub fn DuplexStream1() {
-        let (left, right) = DuplexStream();
+        let (mut left, mut right) = DuplexStream::new();
 
         left.send(~"abc");
         right.send(123);
@@ -152,9 +112,10 @@ mod test {
 
     #[test]
     pub fn basic_rendezvous_test() {
-        let (port, chan) = rendezvous();
+        let (mut port, chan) = rendezvous();
 
         do spawn {
+            let mut chan = chan;
             chan.send("abc");
         }
 
@@ -165,8 +126,9 @@ mod test {
     fn recv_a_lot() {
         // Rendezvous streams should be able to handle any number of messages being sent
         do run_in_uv_task {
-            let (port, chan) = rendezvous();
+            let (mut port, chan) = rendezvous();
             do spawn {
+                let mut chan = chan;
                 1000000.times(|| { chan.send(()) })
             }
             1000000.times(|| { port.recv() })
@@ -175,8 +137,9 @@ mod test {
 
     #[test]
     fn send_and_fail_and_try_recv() {
-        let (port, chan) = rendezvous();
+        let (mut port, chan) = rendezvous();
         do spawn {
+            let mut chan = chan;
             chan.duplex_stream.send(()); // Can't access this field outside this module
             fail!()
         }
@@ -185,8 +148,9 @@ mod test {
 
     #[test]
     fn try_send_and_recv_then_fail_before_ack() {
-        let (port, chan) = rendezvous();
+        let (port, mut chan) = rendezvous();
         do spawn {
+            let mut port = port;
             port.duplex_stream.recv();
             fail!()
         }
@@ -196,8 +160,9 @@ mod test {
     #[test]
     #[should_fail]
     fn send_and_recv_then_fail_before_ack() {
-        let (port, chan) = rendezvous();
+        let (port, mut chan) = rendezvous();
         do spawn {
+            let mut port = port;
             port.duplex_stream.recv();
             fail!()
         }

src/libextra/future.rs

@@ -25,7 +25,6 @@
 
 #[allow(missing_doc)];
 
-use std::comm::{PortOne, oneshot};
 use std::util::replace;
 
 /// A type encapsulating the result of a computation which may not be complete
@@ -104,7 +103,7 @@ impl<A> Future<A> {
 }
 
 impl<A:Send> Future<A> {
-    pub fn from_port(port: PortOne<A>) -> Future<A> {
+    pub fn from_port(port: Port<A>) -> Future<A> {
         /*!
          * Create a future from a port
          *
@@ -125,7 +124,7 @@ impl<A:Send> Future<A> {
          * value of the future.
          */
 
-        let (port, chan) = oneshot();
+        let (port, chan) = Chan::new();
 
         do spawn {
             chan.send(blk());
@@ -139,7 +138,6 @@ impl<A:Send> Future<A> {
 mod test {
     use future::Future;
 
-    use std::comm::oneshot;
     use std::task;
 
     #[test]
@@ -150,7 +148,7 @@ mod test {
 
     #[test]
     fn test_from_port() {
-        let (po, ch) = oneshot();
+        let (po, ch) = Chan::new();
         ch.send(~"whale");
         let mut f = Future::from_port(po);
         assert_eq!(f.get(), ~"whale");