diff options
| author | Alex Crichton <alex@alexcrichton.com> | 2014-12-23 11:53:35 -0800 |
|---|---|---|
| committer | Alex Crichton <alex@alexcrichton.com> | 2014-12-29 12:16:49 -0800 |
| commit | bc83a009f655dd3896be4a7cd33cac8032a605f2 (patch) | |
| tree | 3acc8533031219690fe14fa56f4427cfa9297296 | |
| parent | bb8f4fc3b73918abd19d67be702f78e8f73d1874 (diff) | |
| download | rust-bc83a009f655dd3896be4a7cd33cac8032a605f2.tar.gz rust-bc83a009f655dd3896be4a7cd33cac8032a605f2.zip | |
std: Second pass stabilization for `comm`
This commit is a second pass stabilization for the `std::comm` module,
performing the following actions:
* The entire `std::comm` module was moved under `std::sync::mpsc`. This movement
reflects that channels are just yet another synchronization primitive, and
they don't necessarily deserve a special place outside of the other
concurrency primitives that the standard library offers.
* The `send` and `recv` methods have all been removed.
* The `send_opt` and `recv_opt` methods have been renamed to `send` and `recv`.
This means that all send/receive operations return a `Result` now indicating
whether the operation was successful or not.
* The error type of `send` is now a `SendError` to implement a custom error
message and allow for `unwrap()`. The error type contains an `into_inner`
method to extract the value.
* The error type of `recv` is now `RecvError` for the same reasons as `send`.
* The `TryRecvError` and `TrySendError` types have had public reexports removed
of their variants and the variant names have been tweaked with enum
namespacing rules.
* The `Messages` iterator is renamed to `Iter`
This functionality is now all `#[stable]`:
* `Sender`
* `SyncSender`
* `Receiver`
* `std::sync::mpsc`
* `channel`
* `sync_channel`
* `Iter`
* `Sender::send`
* `Sender::clone`
* `SyncSender::send`
* `SyncSender::try_send`
* `SyncSender::clone`
* `Receiver::recv`
* `Receiver::try_recv`
* `Receiver::iter`
* `SendError`
* `RecvError`
* `TrySendError::{mod, Full, Disconnected}`
* `TryRecvError::{mod, Empty, Disconnected}`
* `SendError::into_inner`
* `TrySendError::into_inner`
This is a breaking change due to the modification of where this module is
located, as well as the changing of the semantics of `send` and `recv`. Most
programs just need to rename imports of `std::comm` to `std::sync::mpsc` and
add calls to `unwrap` after a send or a receive operation.
[breaking-change]
109 files changed, 1175 insertions, 1206 deletions
diff --git a/src/etc/licenseck.py b/src/etc/licenseck.py index 7669df36b04..9ac0acc38a7 100644 --- a/src/etc/licenseck.py +++ b/src/etc/licenseck.py @@ -38,8 +38,8 @@ exceptions = [ "rt/isaac/randport.cpp", # public domain "rt/isaac/rand.h", # public domain "rt/isaac/standard.h", # public domain - "libstd/comm/mpsc_queue.rs", # BSD - "libstd/comm/spsc_queue.rs", # BSD + "libstd/sync/mpsc/mpsc_queue.rs", # BSD + "libstd/sync/mpsc/spsc_queue.rs", # BSD "test/bench/shootout-binarytrees.rs", # BSD "test/bench/shootout-chameneos-redux.rs", # BSD "test/bench/shootout-fannkuch-redux.rs", # BSD diff --git a/src/liballoc/arc.rs b/src/liballoc/arc.rs index 3e235caab18..b8c7bc74132 100644 --- a/src/liballoc/arc.rs +++ b/src/liballoc/arc.rs @@ -592,7 +592,7 @@ impl<T: Default + Sync + Send> Default for Arc<T> { #[allow(experimental)] mod tests { use std::clone::Clone; - use std::comm::channel; + use std::sync::mpsc::channel; use std::mem::drop; use std::ops::Drop; use std::option::Option; @@ -628,11 +628,11 @@ mod tests { let (tx, rx) = channel(); task::spawn(move || { - let arc_v: Arc<Vec<int>> = rx.recv(); + let arc_v: Arc<Vec<int>> = rx.recv().unwrap(); assert_eq!((*arc_v)[3], 4); }); - tx.send(arc_v.clone()); + tx.send(arc_v.clone()).unwrap(); assert_eq!((*arc_v)[2], 3); assert_eq!((*arc_v)[4], 5); diff --git a/src/libcollections/bit.rs b/src/libcollections/bit.rs index e8bd5e76eb7..fe3267b6c35 100644 --- a/src/libcollections/bit.rs +++ b/src/libcollections/bit.rs @@ -2452,7 +2452,7 @@ mod tests { #[cfg(test)] mod bitv_bench { - use std::prelude::*; + use std::prelude::v1::*; use std::rand; use std::rand::Rng; use std::u32; @@ -2947,7 +2947,7 @@ mod bitv_set_test { #[cfg(test)] mod bitv_set_bench { - use std::prelude::*; + use std::prelude::v1::*; use std::rand; use std::rand::Rng; use std::u32; diff --git a/src/librustc_driver/lib.rs b/src/librustc_driver/lib.rs index 07a4bec4bf8..8b716a8431d 100644 --- a/src/librustc_driver/lib.rs +++ b/src/librustc_driver/lib.rs @@ -55,10 +55,10 @@ use rustc::DIAGNOSTICS; use std::any::AnyRefExt; use std::cmp::Ordering::Equal; -use std::comm::channel; use std::io; use std::iter::repeat; use std::os; +use std::sync::mpsc::channel; use std::thread; use rustc::session::early_error; diff --git a/src/librustc_trans/back/write.rs b/src/librustc_trans/back/write.rs index 3a40430beae..55e6dfaebcd 100644 --- a/src/librustc_trans/back/write.rs +++ b/src/librustc_trans/back/write.rs @@ -23,7 +23,6 @@ use syntax::diagnostic; use syntax::diagnostic::{Emitter, Handler, Level, mk_handler}; use std::c_str::{ToCStr, CString}; -use std::comm::channel; use std::io::Command; use std::io::fs; use std::iter::Unfold; @@ -31,6 +30,7 @@ use std::ptr; use std::str; use std::mem; use std::sync::{Arc, Mutex}; +use std::sync::mpsc::channel; use std::thread; use libc::{c_uint, c_int, c_void}; @@ -929,13 +929,13 @@ fn run_work_multithreaded(sess: &Session, } } - tx.take().unwrap().send(()); + tx.take().unwrap().send(()).unwrap(); }).detach(); } let mut panicked = false; for rx in futures.into_iter() { - match rx.recv_opt() { + match rx.recv() { Ok(()) => {}, Err(_) => { panicked = true; diff --git a/src/librustc_trans/save/mod.rs b/src/librustc_trans/save/mod.rs index f491bc84b62..e105a1f6a95 100644 --- a/src/librustc_trans/save/mod.rs +++ b/src/librustc_trans/save/mod.rs @@ -1203,8 +1203,7 @@ impl<'l, 'tcx, 'v> Visitor<'v> for DxrVisitor<'l, 'tcx> { let glob_map = &self.analysis.glob_map; let glob_map = glob_map.as_ref().unwrap(); if glob_map.contains_key(&id) { - let names = glob_map.index(&id); - for n in names.iter() { + for n in glob_map[id].iter() { if name_string.len() > 0 { name_string.push_str(", "); } diff --git a/src/librustdoc/test.rs b/src/librustdoc/test.rs index a9f30ee5ef9..a50bfbde0fe 100644 --- a/src/librustdoc/test.rs +++ b/src/librustdoc/test.rs @@ -9,7 +9,7 @@ // except according to those terms. use std::cell::RefCell; -use std::comm::channel; +use std::sync::mpsc::channel; use std::dynamic_lib::DynamicLibrary; use std::io::{Command, TempDir}; use std::io; diff --git a/src/libstd/bitflags.rs b/src/libstd/bitflags.rs index 1826ad3dc31..4ec329de600 100644 --- a/src/libstd/bitflags.rs +++ b/src/libstd/bitflags.rs @@ -273,7 +273,6 @@ macro_rules! bitflags { mod tests { use hash; use option::Option::{Some, None}; - use ops::{BitOr, BitAnd, BitXor, Sub, Not}; bitflags! { #[doc = "> The first principle is that you must not fool yourself — and"] diff --git a/src/libstd/c_str.rs b/src/libstd/c_str.rs index fa0d8fa9bd5..d096903f9c4 100644 --- a/src/libstd/c_str.rs +++ b/src/libstd/c_str.rs @@ -621,7 +621,7 @@ mod tests { #[test] fn test_unwrap() { let c_str = "hello".to_c_str(); - unsafe { libc::free(c_str.unwrap() as *mut libc::c_void) } + unsafe { libc::free(c_str.into_inner() as *mut libc::c_void) } } #[test] diff --git a/src/libstd/c_vec.rs b/src/libstd/c_vec.rs index d0293aa4a1c..a80659ed937 100644 --- a/src/libstd/c_vec.rs +++ b/src/libstd/c_vec.rs @@ -228,7 +228,7 @@ mod tests { let cv = CVec::new_with_dtor(1 as *mut int, 0, move|:| panic!("Don't run this destructor!")); - let p = cv.unwrap(); + let p = cv.into_inner(); assert_eq!(p, 1 as *mut int); } } diff --git a/src/libstd/collections/hash/map.rs b/src/libstd/collections/hash/map.rs index e7918d605cb..8585b4ecf52 100644 --- a/src/libstd/collections/hash/map.rs +++ b/src/libstd/collections/hash/map.rs @@ -1485,6 +1485,7 @@ mod test_map { struct KindaIntLike(int); + #[allow(deprecated)] impl Equiv<int> for KindaIntLike { fn equiv(&self, other: &int) -> bool { let KindaIntLike(this) = *self; @@ -1812,7 +1813,7 @@ mod test_map { } #[test] - #[allow(experimental)] + #[allow(deprecated)] fn test_pop_equiv() { let mut m = HashMap::new(); m.insert(1i, 2i); diff --git a/src/libstd/io/buffered.rs b/src/libstd/io/buffered.rs index 7bf45915273..1679d2e552f 100644 --- a/src/libstd/io/buffered.rs +++ b/src/libstd/io/buffered.rs @@ -533,7 +533,7 @@ mod test { w.write(&[0, 1]).unwrap(); let a: &[_] = &[]; assert_eq!(a, w.get_ref()[]); - let w = w.unwrap(); + let w = w.into_inner(); let a: &[_] = &[0, 1]; assert_eq!(a, w[]); } diff --git a/src/libstd/io/comm_adapters.rs b/src/libstd/io/comm_adapters.rs index be1dc0e9c34..7b8513ce423 100644 --- a/src/libstd/io/comm_adapters.rs +++ b/src/libstd/io/comm_adapters.rs @@ -10,7 +10,7 @@ use clone::Clone; use cmp; -use comm::{Sender, Receiver}; +use sync::mpsc::{Sender, Receiver}; use io; use option::Option::{None, Some}; use result::Result::{Ok, Err}; @@ -23,7 +23,7 @@ use vec::Vec; /// # Example /// /// ``` -/// use std::comm::channel; +/// use std::sync::mpsc::channel; /// use std::io::ChanReader; /// /// let (tx, rx) = channel(); @@ -59,11 +59,11 @@ impl Buffer for ChanReader { fn fill_buf<'a>(&'a mut self) -> IoResult<&'a [u8]> { if self.pos >= self.buf.len() { self.pos = 0; - match self.rx.recv_opt() { + match self.rx.recv() { Ok(bytes) => { self.buf = bytes; }, - Err(()) => { + Err(..) => { self.closed = true; self.buf = Vec::new(); } @@ -115,7 +115,7 @@ impl Reader for ChanReader { /// /// ``` /// # #![allow(unused_must_use)] -/// use std::comm::channel; +/// use std::sync::mpsc::channel; /// use std::io::ChanWriter; /// /// let (tx, rx) = channel(); @@ -143,7 +143,7 @@ impl Clone for ChanWriter { impl Writer for ChanWriter { fn write(&mut self, buf: &[u8]) -> IoResult<()> { - self.tx.send_opt(buf.to_vec()).map_err(|_| { + self.tx.send(buf.to_vec()).map_err(|_| { io::IoError { kind: io::BrokenPipe, desc: "Pipe closed", @@ -158,7 +158,7 @@ impl Writer for ChanWriter { mod test { use prelude::v1::*; - use comm::channel; + use sync::mpsc::channel; use super::*; use io; use thread::Thread; @@ -167,11 +167,11 @@ mod test { fn test_rx_reader() { let (tx, rx) = channel(); Thread::spawn(move|| { - tx.send(vec![1u8, 2u8]); - tx.send(vec![]); - tx.send(vec![3u8, 4u8]); - tx.send(vec![5u8, 6u8]); - tx.send(vec![7u8, 8u8]); + tx.send(vec![1u8, 2u8]).unwrap(); + tx.send(vec![]).unwrap(); + tx.send(vec![3u8, 4u8]).unwrap(); + tx.send(vec![5u8, 6u8]).unwrap(); + tx.send(vec![7u8, 8u8]).unwrap(); }).detach(); let mut reader = ChanReader::new(rx); @@ -209,12 +209,12 @@ mod test { fn test_rx_buffer() { let (tx, rx) = channel(); Thread::spawn(move|| { - tx.send(b"he".to_vec()); - tx.send(b"llo wo".to_vec()); - tx.send(b"".to_vec()); - tx.send(b"rld\nhow ".to_vec()); - tx.send(b"are you?".to_vec()); - tx.send(b"".to_vec()); + tx.send(b"he".to_vec()).unwrap(); + tx.send(b"llo wo".to_vec()).unwrap(); + tx.send(b"".to_vec()).unwrap(); + tx.send(b"rld\nhow ".to_vec()).unwrap(); + tx.send(b"are you?".to_vec()).unwrap(); + tx.send(b"".to_vec()).unwrap(); }).detach(); let mut reader = ChanReader::new(rx); @@ -234,7 +234,7 @@ mod test { writer.write_be_u32(42).unwrap(); let wanted = vec![0u8, 0u8, 0u8, 42u8]; - let got = match Thread::spawn(move|| { rx.recv() }).join() { + let got = match Thread::spawn(move|| { rx.recv().unwrap() }).join() { Ok(got) => got, Err(_) => panic!(), }; diff --git a/src/libstd/io/mem.rs b/src/libstd/io/mem.rs index 4341666a27e..71c42273c22 100644 --- a/src/libstd/io/mem.rs +++ b/src/libstd/io/mem.rs @@ -402,8 +402,8 @@ mod test { use prelude::v1::*; use super::*; - use io::*; use io; + use io::{SeekSet, SeekCur, SeekEnd}; use self::test_crate::Bencher; #[test] diff --git a/src/libstd/io/net/pipe.rs b/src/libstd/io/net/pipe.rs index f5edf8955d8..68f3a8e1836 100644 --- a/src/libstd/io/net/pipe.rs +++ b/src/libstd/io/net/pipe.rs @@ -267,11 +267,13 @@ impl sys_common::AsInner<UnixAcceptorImp> for UnixAcceptor { mod tests { use prelude::v1::*; - use comm::channel; - use io::*; use io::fs::PathExtensions; + use io::{EndOfFile, TimedOut, ShortWrite, IoError, ConnectionReset}; + use io::{NotConnected, BrokenPipe, FileNotFound, InvalidInput, OtherIoError}; + use io::{PermissionDenied, Acceptor, Listener}; use io::test::*; use super::*; + use sync::mpsc::channel; use thread::Thread; use time::Duration; @@ -431,18 +433,18 @@ mod tests { let (tx2, rx2) = channel(); let _t = Thread::spawn(move|| { let mut s2 = s2; - rx1.recv(); + rx1.recv().unwrap(); debug!("writer writing"); s2.write(&[1]).unwrap(); debug!("writer done"); - tx2.send(()); + tx2.send(()).unwrap(); }); - tx1.send(()); + tx1.send(()).unwrap(); let mut buf = [0, 0]; debug!("reader reading"); assert_eq!(s1.read(&mut buf), Ok(1)); debug!("reader done"); - rx2.recv(); + rx2.recv().unwrap(); } #[test] @@ -455,9 +457,9 @@ mod tests { let _t = Thread::spawn(move|| { let mut s = UnixStream::connect(&addr); s.write(&[1]).unwrap(); - rx.recv(); + rx.recv().unwrap(); s.write(&[2]).unwrap(); - rx.recv(); + rx.recv().unwrap(); }); let mut s1 = acceptor.accept().unwrap(); @@ -468,14 +470,14 @@ mod tests { let mut s2 = s2; let mut buf = [0, 0]; s2.read(&mut buf).unwrap(); - tx2.send(()); - done.send(()); + tx2.send(()).unwrap(); + done.send(()).unwrap(); }); let mut buf = [0, 0]; s1.read(&mut buf).unwrap(); - tx1.send(()); + tx1.send(()).unwrap(); - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -497,11 +499,11 @@ mod tests { let _t = Thread::spawn(move|| { let mut s2 = s2; s2.write(&[1]).unwrap(); - tx.send(()); + tx.send(()).unwrap(); }); s1.write(&[2]).unwrap(); - rx.recv(); + rx.recv().unwrap(); } #[cfg(not(windows))] @@ -542,9 +544,9 @@ mod tests { let (tx, rx) = channel(); let addr2 = addr.clone(); let _t = Thread::spawn(move|| { - tx.send(UnixStream::connect(&addr2).unwrap()); + tx.send(UnixStream::connect(&addr2).unwrap()).unwrap(); }); - let l = rx.recv(); + let l = rx.recv().unwrap(); for i in range(0u, 1001) { match a.accept() { Ok(..) => break, @@ -600,7 +602,7 @@ mod tests { Thread::spawn(move|| { let mut a = a; let _s = a.accept().unwrap(); - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut b = [0]; @@ -637,7 +639,7 @@ mod tests { Thread::spawn(move|| { let mut a = a; let _s = a.accept().unwrap(); - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut s = UnixStream::connect(&addr).unwrap(); @@ -646,13 +648,13 @@ mod tests { let _t = Thread::spawn(move|| { let mut s2 = s2; assert!(s2.read(&mut [0]).is_err()); - tx.send(()); + tx.send(()).unwrap(); }); // this should wake up the child task s.close_read().unwrap(); // this test will never finish if the child doesn't wake up - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -662,9 +664,9 @@ mod tests { let (tx, rx) = channel::<()>(); Thread::spawn(move|| { let mut s = UnixStream::connect(&addr).unwrap(); - rx.recv(); + rx.recv().unwrap(); assert!(s.write(&[0]).is_ok()); - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut s = a.accept().unwrap(); @@ -688,7 +690,7 @@ mod tests { assert_eq!(s.write(&[0]).err().unwrap().kind, TimedOut); } - tx.send(()); + tx.send(()).unwrap(); s.set_timeout(None); assert_eq!(s.read(&mut [0, 0]), Ok(1)); } @@ -700,7 +702,7 @@ mod tests { let (tx, rx) = channel::<()>(); Thread::spawn(move|| { let mut s = UnixStream::connect(&addr).unwrap(); - rx.recv(); + rx.recv().unwrap(); let mut amt = 0; while amt < 100 * 128 * 1024 { match s.read(&mut [0, ..128 * 1024]) { @@ -708,7 +710,7 @@ mod tests { Err(e) => panic!("{}", e), } } - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut s = a.accept().unwrap(); @@ -716,7 +718,7 @@ mod tests { assert_eq!(s.read(&mut [0]).err().unwrap().kind, TimedOut); assert_eq!(s.read(&mut [0]).err().unwrap().kind, TimedOut); - tx.send(()); + tx.send(()).unwrap(); for _ in range(0u, 100) { assert!(s.write(&[0, ..128 * 1024]).is_ok()); } @@ -729,9 +731,9 @@ mod tests { let (tx, rx) = channel::<()>(); Thread::spawn(move|| { let mut s = UnixStream::connect(&addr).unwrap(); - rx.recv(); + rx.recv().unwrap(); assert!(s.write(&[0]).is_ok()); - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut s = a.accept().unwrap(); @@ -745,7 +747,7 @@ mod tests { if i == 1000 { panic!("should have filled up?!"); } } - tx.send(()); + tx.send(()).unwrap(); assert!(s.read(&mut [0]).is_ok()); } @@ -756,9 +758,9 @@ mod tests { let (tx, rx) = channel::<()>(); Thread::spawn(move|| { let mut s = UnixStream::connect(&addr).unwrap(); - rx.recv(); + rx.recv().unwrap(); assert!(s.write(&[0]).is_ok()); - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut s = a.accept().unwrap(); @@ -767,14 +769,14 @@ mod tests { let _t = Thread::spawn(move|| { let mut s2 = s2; assert!(s2.read(&mut [0]).is_ok()); - tx2.send(()); + tx2.send(()).unwrap(); }); s.set_read_timeout(Some(20)); assert_eq!(s.read(&mut [0]).err().unwrap().kind, TimedOut); - tx.send(()); + tx.send(()).unwrap(); - rx2.recv(); + rx2.recv().unwrap(); } #[cfg(not(windows))] @@ -809,8 +811,14 @@ mod tests { let (tx, rx) = channel(); let tx2 = tx.clone(); - let _t = Thread::spawn(move|| { let mut a = a; tx.send(a.accept()) }); - let _t = Thread::spawn(move|| { let mut a = a2; tx2.send(a.accept()) }); + let _t = Thread::spawn(move|| { + let mut a = a; + tx.send(a.accept()).unwrap() + }); + let _t = Thread::spawn(move|| { + let mut a = a2; + tx2.send(a.accept()).unwrap() + }); let addr2 = addr.clone(); let _t = Thread::spawn(move|| { @@ -820,8 +828,8 @@ mod tests { let _ = UnixStream::connect(&addr); }); - assert!(rx.recv().is_ok()); - assert!(rx.recv().is_ok()); + assert!(rx.recv().unwrap().is_ok()); + assert!(rx.recv().unwrap().is_ok()); } #[test] @@ -844,10 +852,10 @@ mod tests { let (tx, rx) = channel(); let _t = Thread::spawn(move|| { let mut a = a; - tx.send(a.accept()); + tx.send(a.accept()).unwrap(); }); a2.close_accept().unwrap(); - assert_eq!(rx.recv().err().unwrap().kind, EndOfFile); + assert_eq!(rx.recv().unwrap().err().unwrap().kind, EndOfFile); } } diff --git a/src/libstd/io/net/tcp.rs b/src/libstd/io/net/tcp.rs index 67aaf7dda8e..57ffcfaad30 100644 --- a/src/libstd/io/net/tcp.rs +++ b/src/libstd/io/net/tcp.rs @@ -484,12 +484,15 @@ impl sys_common::AsInner<TcpAcceptorImp> for TcpAcceptor { mod test { use prelude::v1::*; - use comm::channel; + use sync::mpsc::channel; use thread::Thread; use io::net::tcp::*; use io::net::ip::*; - use io::*; use io::test::*; + use io::{EndOfFile, TimedOut, ShortWrite, IoError}; + use io::{ConnectionRefused, BrokenPipe, ConnectionAborted}; + use io::{ConnectionReset, NotConnected, PermissionDenied, OtherIoError}; + use io::{Acceptor, Listener}; // FIXME #11530 this fails on android because tests are run as root #[cfg_attr(any(windows, target_os = "android"), ignore)] @@ -694,11 +697,11 @@ mod test { let (tx, rx) = channel(); let _t = Thread::spawn(move|| { drop(TcpStream::connect(addr)); - tx.send(()); + tx.send(()).unwrap(); }); let mut stream = acceptor.accept(); - rx.recv(); + rx.recv().unwrap(); let buf = [0]; match stream.write(&buf) { Ok(..) => {} @@ -719,11 +722,11 @@ mod test { let (tx, rx) = channel(); let _t = Thread::spawn(move|| { drop(TcpStream::connect(addr)); - tx.send(()); + tx.send(()).unwrap(); }); let mut stream = acceptor.accept(); - rx.recv(); + rx.recv().unwrap(); let buf = [0]; match stream.write(&buf) { Ok(..) => {} @@ -969,20 +972,20 @@ mod test { let (tx, rx) = channel(); let _t = Thread::spawn(move|| { let mut srv = TcpListener::bind(addr).listen().unwrap(); - tx.send(()); + tx.send(()).unwrap(); let mut cl = srv.accept().unwrap(); cl.write(&[10]).unwrap(); let mut b = [0]; cl.read(&mut b).unwrap(); - tx.send(()); + tx.send(()).unwrap(); }); - rx.recv(); + rx.recv().unwrap(); let mut c = TcpStream::connect(addr).unwrap(); let mut b = [0, ..10]; assert_eq!(c.read(&mut b), Ok(1)); c.write(&[1]).unwrap(); - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -1005,19 +1008,19 @@ mod test { let (tx, rx) = channel(); let _t = Thread::spawn(move|| { - rx.recv(); + rx.recv().unwrap(); let _stream = TcpStream::connect(addr).unwrap(); // Close - rx.recv(); + rx.recv().unwrap(); }); { let mut acceptor = TcpListener::bind(addr).listen(); - tx.send(()); + tx.send(()).unwrap(); { let _stream = acceptor.accept().unwrap(); // Close client - tx.send(()); + tx.send(()).unwrap(); } // Close listener } @@ -1044,14 +1047,14 @@ mod test { let (tx2, rx2) = channel(); let _t = Thread::spawn(move|| { let mut s2 = s2; - rx1.recv(); + rx1.recv().unwrap(); s2.write(&[1]).unwrap(); - tx2.send(()); + tx2.send(()).unwrap(); }); - tx1.send(()); + tx1.send(()).unwrap(); let mut buf = [0, 0]; assert_eq!(s1.read(&mut buf), Ok(1)); - rx2.recv(); + rx2.recv().unwrap(); } #[test] @@ -1064,9 +1067,9 @@ mod test { let _t = Thread::spawn(move|| { let mut s = TcpStream::connect(addr); s.write(&[1]).unwrap(); - rx.recv(); + rx.recv().unwrap(); s.write(&[2]).unwrap(); - rx.recv(); + rx.recv().unwrap(); }); let mut s1 = acceptor.accept().unwrap(); @@ -1077,14 +1080,14 @@ mod test { let mut s2 = s2; let mut buf = [0, 0]; s2.read(&mut buf).unwrap(); - tx2.send(()); - done.send(()); + tx2.send(()).unwrap(); + done.send(()).unwrap(); }); let mut buf = [0, 0]; s1.read(&mut buf).unwrap(); - tx1.send(()); + tx1.send(()).unwrap(); - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -1106,11 +1109,11 @@ mod test { let _t = Thread::spawn(move|| { let mut s2 = s2; s2.write(&[1]).unwrap(); - done.send(()); + done.send(()).unwrap(); }); s1.write(&[2]).unwrap(); - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -1152,9 +1155,9 @@ mod test { if !cfg!(target_os = "freebsd") { let (tx, rx) = channel(); let _t = Thread::spawn(move|| { - tx.send(TcpStream::connect(addr).unwrap()); + tx.send(TcpStream::connect(addr).unwrap()).unwrap(); }); - let _l = rx.recv(); + let _l = rx.recv().unwrap(); for i in range(0i, 1001) { match a.accept() { Ok(..) => break, @@ -1182,7 +1185,7 @@ mod test { Thread::spawn(move|| { let mut a = a; let _s = a.accept().unwrap(); - let _ = rx.recv_opt(); + let _ = rx.recv().unwrap(); }).detach(); let mut b = [0]; @@ -1219,7 +1222,7 @@ mod test { Thread::spawn(move|| { let mut a = a; let _s = a.accept().unwrap(); - let _ = rx.recv_opt(); + let _ = rx.recv().unwrap(); }).detach(); let mut s = TcpStream::connect(addr).unwrap(); @@ -1228,13 +1231,13 @@ mod test { let _t = Thread::spawn(move|| { let mut s2 = s2; assert!(s2.read(&mut [0]).is_err()); - tx.send(()); + tx.send(()).unwrap(); }); // this should wake up the child task s.close_read().unwrap(); // this test will never finish if the child doesn't wake up - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -1244,9 +1247,9 @@ mod test { let (tx, rx) = channel::<()>(); Thread::spawn(move|| { let mut s = TcpStream::connect(addr).unwrap(); - rx.recv(); + rx.recv().unwrap(); assert!(s.write(&[0]).is_ok()); - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut s = a.accept().unwrap(); @@ -1265,7 +1268,7 @@ mod test { } assert_eq!(s.write(&[0]).err().unwrap().kind, TimedOut); - tx.send(()); + tx.send(()).unwrap(); s.set_timeout(None); assert_eq!(s.read(&mut [0, 0]), Ok(1)); } @@ -1277,7 +1280,7 @@ mod test { let (tx, rx) = channel::<()>(); Thread::spawn(move|| { let mut s = TcpStream::connect(addr).unwrap(); - rx.recv(); + rx.recv().unwrap(); let mut amt = 0; while amt < 100 * 128 * 1024 { match s.read(&mut [0, ..128 * 1024]) { @@ -1285,7 +1288,7 @@ mod test { Err(e) => panic!("{}", e), } } - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut s = a.accept().unwrap(); @@ -1293,7 +1296,7 @@ mod test { assert_eq!(s.read(&mut [0]).err().unwrap().kind, TimedOut); assert_eq!(s.read(&mut [0]).err().unwrap().kind, TimedOut); - tx.send(()); + tx.send(()).unwrap(); for _ in range(0i, 100) { assert!(s.write(&[0, ..128 * 1024]).is_ok()); } @@ -1306,9 +1309,9 @@ mod test { let (tx, rx) = channel::<()>(); Thread::spawn(move|| { let mut s = TcpStream::connect(addr).unwrap(); - rx.recv(); + rx.recv().unwrap(); assert!(s.write(&[0]).is_ok()); - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut s = a.accept().unwrap(); @@ -1323,7 +1326,7 @@ mod test { } assert_eq!(s.write(&[0]).err().unwrap().kind, TimedOut); - tx.send(()); + tx.send(()).unwrap(); assert!(s.read(&mut [0]).is_ok()); } @@ -1334,9 +1337,9 @@ mod test { let (tx, rx) = channel::<()>(); Thread::spawn(move|| { let mut s = TcpStream::connect(addr).unwrap(); - rx.recv(); + rx.recv().unwrap(); assert_eq!(s.write(&[0]), Ok(())); - let _ = rx.recv_opt(); + let _ = rx.recv(); }).detach(); let mut s = a.accept().unwrap(); @@ -1345,14 +1348,14 @@ mod test { let _t = Thread::spawn(move|| { let mut s2 = s2; assert_eq!(s2.read(&mut [0]), Ok(1)); - tx2.send(()); + tx2.send(()).unwrap(); }); s.set_read_timeout(Some(20)); assert_eq!(s.read(&mut [0]).err().unwrap().kind, TimedOut); - tx.send(()); + tx.send(()).unwrap(); - rx2.recv(); + rx2.recv().unwrap(); } #[test] @@ -1367,9 +1370,9 @@ mod test { let txdone2 = txdone.clone(); let _t = Thread::spawn(move|| { let mut tcp = TcpStream::connect(addr).unwrap(); - rx.recv(); + rx.recv().unwrap(); tcp.write_u8(0).unwrap(); - txdone2.send(()); + txdone2.send(()).unwrap(); }); // Spawn off a reading clone @@ -1379,7 +1382,7 @@ mod test { let _t = Thread::spawn(move|| { let mut tcp2 = tcp2; tcp2.read_u8().unwrap(); - txdone3.send(()); + txdone3.send(()).unwrap(); }); // Try to ensure that the reading clone is indeed reading @@ -1390,9 +1393,9 @@ mod test { // clone the handle again while it's reading, then let it finish the // read. let _ = tcp.clone(); - tx.send(()); - rxdone.recv(); - rxdone.recv(); + tx.send(()).unwrap(); + rxdone.recv().unwrap(); + rxdone.recv().unwrap(); } #[test] @@ -1423,8 +1426,14 @@ mod test { let (tx, rx) = channel(); let tx2 = tx.clone(); - let _t = Thread::spawn(move|| { let mut a = a; tx.send(a.accept()) }); - let _t = Thread::spawn(move|| { let mut a = a2; tx2.send(a.accept()) }); + let _t = Thread::spawn(move|| { + let mut a = a; + tx.send(a.accept()).unwrap(); + }); + let _t = Thread::spawn(move|| { + let mut a = a2; + tx2.send(a.accept()).unwrap(); + }); let _t = Thread::spawn(move|| { let _ = TcpStream::connect(addr); @@ -1433,8 +1442,8 @@ mod test { let _ = TcpStream::connect(addr); }); - assert!(rx.recv().is_ok()); - assert!(rx.recv().is_ok()); + assert!(rx.recv().unwrap().is_ok()); + assert!(rx.recv().unwrap().is_ok()); } #[test] @@ -1457,10 +1466,10 @@ mod test { let (tx, rx) = channel(); let _t = Thread::spawn(move|| { let mut a = a; - tx.send(a.accept()); + tx.send(a.accept()).unwrap(); }); a2.close_accept().unwrap(); - assert_eq!(rx.recv().err().unwrap().kind, EndOfFile); + assert_eq!(rx.recv().unwrap().err().unwrap().kind, EndOfFile); } } diff --git a/src/libstd/io/net/udp.rs b/src/libstd/io/net/udp.rs index 84269eb1ed9..d6c379cc9f7 100644 --- a/src/libstd/io/net/udp.rs +++ b/src/libstd/io/net/udp.rs @@ -250,10 +250,10 @@ impl Writer for UdpStream { mod test { use prelude::v1::*; - use comm::channel; - use io::*; + use sync::mpsc::channel; use io::net::ip::*; use io::test::*; + use io::{IoError, TimedOut, PermissionDenied, ShortWrite}; use super::*; use thread::Thread; @@ -278,17 +278,17 @@ mod test { let _t = Thread::spawn(move|| { match UdpSocket::bind(client_ip) { Ok(ref mut client) => { - rx1.recv(); + rx1.recv().unwrap(); client.send_to(&[99], server_ip).unwrap() } Err(..) => panic!() } - tx2.send(()); + tx2.send(()).unwrap(); }); match UdpSocket::bind(server_ip) { Ok(ref mut server) => { - tx1.send(()); + tx1.send(()).unwrap(); let mut buf = [0]; match server.recv_from(&mut buf) { Ok((nread, src)) => { @@ -301,7 +301,7 @@ mod test { } Err(..) => panic!() } - rx2.recv(); + rx2.recv().unwrap(); } #[test] @@ -313,7 +313,7 @@ mod test { let _t = Thread::spawn(move|| { match UdpSocket::bind(client_ip) { Ok(ref mut client) => { - rx.recv(); + rx.recv().unwrap(); client.send_to(&[99], server_ip).unwrap() } Err(..) => panic!() @@ -322,7 +322,7 @@ mod test { match UdpSocket::bind(server_ip) { Ok(ref mut server) => { - tx.send(()); + tx.send(()).unwrap(); let mut buf = [0]; match server.recv_from(&mut buf) { Ok((nread, src)) => { @@ -357,17 +357,17 @@ mod test { Err(..) => panic!() } }; - rx1.recv(); + rx1.recv().unwrap(); send_as(dummy_ip, &[98]); send_as(client_ip, &[99]); - tx2.send(()); + tx2.send(()).unwrap(); }); match UdpSocket::bind(server_ip) { Ok(server) => { let server = box server; let mut stream = server.connect(client_ip); - tx1.send(()); + tx1.send(()).unwrap(); let mut buf = [0]; match stream.read(&mut buf) { Ok(nread) => { @@ -379,7 +379,7 @@ mod test { } Err(..) => panic!() } - rx2.recv(); + rx2.recv().unwrap(); } #[test] @@ -395,19 +395,19 @@ mod test { Ok(client) => { let client = box client; let mut stream = client.connect(server_ip); - rx1.recv(); + rx1.recv().unwrap(); stream.write(&[99]).unwrap(); } Err(..) => panic!() } - tx2.send(()); + tx2.send(()).unwrap(); }); match UdpSocket::bind(server_ip) { Ok(server) => { let server = box server; let mut stream = server.connect(client_ip); - tx1.send(()); + tx1.send(()).unwrap(); let mut buf = [0]; match stream.read(&mut buf) { Ok(nread) => { @@ -419,7 +419,7 @@ mod test { } Err(..) => panic!() } - rx2.recv(); + rx2.recv().unwrap(); } pub fn socket_name(addr: SocketAddr) { @@ -466,14 +466,14 @@ mod test { let (tx2, rx2) = channel(); let _t = Thread::spawn(move|| { let mut sock3 = sock3; - rx1.recv(); + rx1.recv().unwrap(); sock3.send_to(&[1], addr2).unwrap(); - tx2.send(()); + tx2.send(()).unwrap(); }); - tx1.send(()); + tx1.send(()).unwrap(); let mut buf = [0, 0]; assert_eq!(sock1.recv_from(&mut buf), Ok((1, addr2))); - rx2.recv(); + rx2.recv().unwrap(); } #[test] @@ -488,9 +488,9 @@ mod test { let _t = Thread::spawn(move|| { let mut sock2 = sock2; sock2.send_to(&[1], addr1).unwrap(); - rx.recv(); + rx.recv().unwrap(); sock2.send_to(&[2], addr1).unwrap(); - rx.recv(); + rx.recv().unwrap(); }); let sock3 = sock1.clone(); @@ -500,14 +500,14 @@ mod test { let mut sock3 = sock3; let mut buf = [0, 0]; sock3.recv_from(&mut buf).unwrap(); - tx2.send(()); - done.send(()); + tx2.send(()).unwrap(); + done.send(()).unwrap(); }); let mut buf = [0, 0]; sock1.recv_from(&mut buf).unwrap(); - tx1.send(()); + tx1.send(()).unwrap(); - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -524,12 +524,12 @@ mod test { let mut sock2 = sock2; let mut buf = [0, 1]; - rx.recv(); + rx.recv().unwrap(); match sock2.recv_from(&mut buf) { Ok(..) => {} Err(e) => panic!("failed receive: {}", e), } - serv_tx.send(()); + serv_tx.send(()).unwrap(); }); let sock3 = sock1.clone(); @@ -539,19 +539,19 @@ mod test { let _t = Thread::spawn(move|| { let mut sock3 = sock3; match sock3.send_to(&[1], addr2) { - Ok(..) => { let _ = tx2.send_opt(()); } + Ok(..) => { let _ = tx2.send(()); } Err(..) => {} } - done.send(()); + done.send(()).unwrap(); }); match sock1.send_to(&[2], addr2) { - Ok(..) => { let _ = tx.send_opt(()); } + Ok(..) => { let _ = tx.send(()); } Err(..) => {} } drop(tx); - rx.recv(); - serv_rx.recv(); + rx.recv().unwrap(); + serv_rx.recv().unwrap(); } #[cfg(not(windows))] // FIXME #17553 @@ -568,10 +568,10 @@ mod test { let mut a = a2; assert_eq!(a.recv_from(&mut [0]), Ok((1, addr1))); assert_eq!(a.send_to(&[0], addr1), Ok(())); - rx.recv(); + rx.recv().unwrap(); assert_eq!(a.send_to(&[0], addr1), Ok(())); - tx2.send(()); + tx2.send(()).unwrap(); }); // Make sure that reads time out, but writes can continue @@ -586,11 +586,11 @@ mod test { // Clearing the timeout should allow for receiving a.set_timeout(None); - tx.send(()); + tx.send(()).unwrap(); assert_eq!(a2.recv_from(&mut [0]), Ok((1, addr2))); // Make sure the child didn't die - rx2.recv(); + rx2.recv().unwrap(); } #[test] diff --git a/src/libstd/io/pipe.rs b/src/libstd/io/pipe.rs index 40ae4922823..ee376658283 100644 --- a/src/libstd/io/pipe.rs +++ b/src/libstd/io/pipe.rs @@ -114,7 +114,7 @@ impl Writer for PipeStream { mod test { use prelude::v1::*; - use comm::channel; + use sync::mpsc::channel; use thread::Thread; #[test] @@ -129,11 +129,11 @@ mod test { let _t = Thread::spawn(move|| { let mut out = out; out.write(&[10]).unwrap(); - rx.recv(); // don't close the pipe until the other read has finished + rx.recv().unwrap(); // don't close the pipe until the other read has finished }); let mut buf = [0, ..10]; input.read(&mut buf).unwrap(); - tx.send(()); + tx.send(()).unwrap(); } } diff --git a/src/libstd/io/process.rs b/src/libstd/io/process.rs index d1e9e2c4ea1..1e008287a31 100644 --- a/src/libstd/io/process.rs +++ b/src/libstd/io/process.rs @@ -20,7 +20,6 @@ use prelude::v1::*; use c_str::{CString, ToCStr}; use collections::HashMap; -use comm::{channel, Receiver}; use fmt; use hash::Hash; use io::pipe::{PipeStream, PipePair}; @@ -29,6 +28,7 @@ use io; use libc; use os; use path::BytesContainer; +use sync::mpsc::{channel, Receiver}; use sys::fs::FileDesc; use sys::process::Process as ProcessImp; use sys; @@ -693,10 +693,10 @@ impl Process { Some(stream) => { Thread::spawn(move |:| { let mut stream = stream; - tx.send(stream.read_to_end()) + tx.send(stream.read_to_end()).unwrap(); }).detach(); } - None => tx.send(Ok(Vec::new())) + None => tx.send(Ok(Vec::new())).unwrap() } rx } @@ -707,8 +707,8 @@ impl Process { Ok(ProcessOutput { status: status, - output: stdout.recv().ok().unwrap_or(Vec::new()), - error: stderr.recv().ok().unwrap_or(Vec::new()), + output: stdout.recv().unwrap().unwrap_or(Vec::new()), + error: stderr.recv().unwrap().unwrap_or(Vec::new()), }) } @@ -743,13 +743,15 @@ impl Drop for Process { mod tests { use prelude::v1::*; - use comm::channel; - use io::*; use io::fs::PathExtensions; + use io::process; use io::timer::*; + use io::{Truncate, Write, TimedOut, timer, FileNotFound}; use rt::running_on_valgrind; use str; - use super::*; + use super::{CreatePipe}; + use super::{InheritFd, Process, PleaseExitSignal, Command, ProcessOutput}; + use sync::mpsc::channel; use thread::Thread; use time::Duration; @@ -1160,17 +1162,17 @@ mod tests { p.set_timeout(Some(10)); assert_eq!(p.wait().err().unwrap().kind, TimedOut); p.signal_kill().unwrap(); - tx.send(()); + tx.send(()).unwrap(); }); let _t = Thread::spawn(move|| { let mut p = sleeper(); p.set_timeout(Some(10)); assert_eq!(p.wait().err().unwrap().kind, TimedOut); p.signal_kill().unwrap(); - tx2.send(()); + tx2.send(()).unwrap(); }); - rx.recv(); - rx.recv(); + rx.recv().unwrap(); + rx.recv().unwrap(); } #[test] diff --git a/src/libstd/io/stdio.rs b/src/libstd/io/stdio.rs index c378b95ff33..74b0930a145 100644 --- a/src/libstd/io/stdio.rs +++ b/src/libstd/io/stdio.rs @@ -543,7 +543,7 @@ mod tests { use prelude::v1::*; use super::*; - use comm::channel; + use sync::mpsc::channel; use thread::Thread; #[test] diff --git a/src/libstd/io/timer.rs b/src/libstd/io/timer.rs index 2067a97fdac..e073f76af82 100644 --- a/src/libstd/io/timer.rs +++ b/src/libstd/io/timer.rs @@ -15,7 +15,7 @@ // FIXME: These functions take Durations but only pass ms to the backend impls. -use comm::{Receiver, Sender, channel}; +use sync::mpsc::{Receiver, Sender, channel}; use time::Duration; use io::IoResult; use sys::timer::Callback; @@ -40,11 +40,11 @@ use sys::timer::Timer as TimerImp; /// /// let timeout = timer.oneshot(Duration::milliseconds(10)); /// // do some work -/// timeout.recv(); // wait for the timeout to expire +/// timeout.recv().unwrap(); // wait for the timeout to expire /// /// let periodic = timer.periodic(Duration::milliseconds(10)); /// loop { -/// periodic.recv(); +/// periodic.recv().unwrap(); /// // this loop is only executed once every 10ms /// } /// # } @@ -126,7 +126,7 @@ impl Timer { /// for _ in range(0u, 100) { /* do work */ } /// /// // blocks until 10 ms after the `oneshot` call - /// ten_milliseconds.recv(); + /// ten_milliseconds.recv().unwrap(); /// ``` /// /// ```rust @@ -136,7 +136,7 @@ impl Timer { /// // Incorrect, method chaining-style: /// let mut five_ms = Timer::new().unwrap().oneshot(Duration::milliseconds(5)); /// // The timer object was destroyed, so this will always fail: - /// // five_ms.recv() + /// // five_ms.recv().unwrap() /// ``` /// /// When provided a zero or negative `duration`, the message will @@ -147,7 +147,7 @@ impl Timer { if in_ms_u64(duration) != 0 { self.inner.oneshot(in_ms_u64(duration), box TimerCallback { tx: tx }); } else { - tx.send(()); + tx.send(()).unwrap(); } return rx } @@ -178,13 +178,13 @@ impl Timer { /// for _ in range(0u, 100) { /* do work */ } /// /// // blocks until 10 ms after the `periodic` call - /// ten_milliseconds.recv(); + /// ten_milliseconds.recv().unwrap(); /// /// for _ in range(0u, 100) { /* do work */ } /// /// // blocks until 20 ms after the `periodic` call (*not* 10ms after the /// // previous `recv`) - /// ten_milliseconds.recv(); + /// ten_milliseconds.recv().unwrap(); /// ``` /// /// ```rust @@ -194,7 +194,7 @@ impl Timer { /// // Incorrect, method chaining-style. /// let mut five_ms = Timer::new().unwrap().periodic(Duration::milliseconds(5)); /// // The timer object was destroyed, so this will always fail: - /// // five_ms.recv() + /// // five_ms.recv().unwrap() /// ``` /// /// When provided a zero or negative `duration`, the messages will @@ -213,7 +213,7 @@ impl Timer { impl Callback for TimerCallback { fn call(&mut self) { - let _ = self.tx.send_opt(()); + let _ = self.tx.send(()); } } @@ -225,8 +225,6 @@ fn in_ms_u64(d: Duration) -> u64 { #[cfg(test)] mod test { - use prelude::v1::*; - use super::Timer; use thread::Thread; use time::Duration; @@ -240,7 +238,7 @@ mod test { #[test] fn test_io_timer_sleep_oneshot() { let mut timer = Timer::new().unwrap(); - timer.oneshot(Duration::milliseconds(1)).recv(); + timer.oneshot(Duration::milliseconds(1)).recv().unwrap(); } #[test] @@ -254,8 +252,8 @@ mod test { let mut timer = Timer::new().unwrap(); let rx1 = timer.oneshot(Duration::milliseconds(10000)); let rx = timer.oneshot(Duration::milliseconds(1)); - rx.recv(); - assert_eq!(rx1.recv_opt(), Err(())); + rx.recv().unwrap(); + assert!(rx1.recv().is_err()); } #[test] @@ -264,16 +262,16 @@ mod test { let rx = timer.oneshot(Duration::milliseconds(100000000)); timer.sleep(Duration::milliseconds(1)); // this should invalidate rx - assert_eq!(rx.recv_opt(), Err(())); + assert!(rx.recv().is_err()); } #[test] fn test_io_timer_sleep_periodic() { let mut timer = Timer::new().unwrap(); let rx = timer.periodic(Duration::milliseconds(1)); - rx.recv(); - rx.recv(); - rx.recv(); + rx.recv().unwrap(); + rx.recv().unwrap(); + rx.recv().unwrap(); } #[test] @@ -292,12 +290,12 @@ mod test { let mut timer = Timer::new().unwrap(); let rx = timer.oneshot(Duration::milliseconds(1)); - rx.recv(); - assert!(rx.recv_opt().is_err()); + rx.recv().unwrap(); + assert!(rx.recv().is_err()); let rx = timer.oneshot(Duration::milliseconds(1)); - rx.recv(); - assert!(rx.recv_opt().is_err()); + rx.recv().unwrap(); + assert!(rx.recv().is_err()); } #[test] @@ -306,20 +304,20 @@ mod test { let orx = timer.oneshot(Duration::milliseconds(100)); let prx = timer.periodic(Duration::milliseconds(100)); timer.sleep(Duration::milliseconds(1)); - assert_eq!(orx.recv_opt(), Err(())); - assert_eq!(prx.recv_opt(), Err(())); - timer.oneshot(Duration::milliseconds(1)).recv(); + assert!(orx.recv().is_err()); + assert!(prx.recv().is_err()); + timer.oneshot(Duration::milliseconds(1)).recv().unwrap(); } #[test] fn period() { let mut timer = Timer::new().unwrap(); let rx = timer.periodic(Duration::milliseconds(1)); - rx.recv(); - rx.recv(); + rx.recv().unwrap(); + rx.recv().unwrap(); let rx2 = timer.periodic(Duration::milliseconds(1)); - rx2.recv(); - rx2.recv(); + rx2.recv().unwrap(); + rx2.recv().unwrap(); } #[test] @@ -359,7 +357,7 @@ mod test { let timer_rx = timer.periodic(Duration::milliseconds(1000)); Thread::spawn(move|| { - let _ = timer_rx.recv_opt(); + let _ = timer_rx.recv(); }).detach(); // when we drop the TimerWatcher we're going to destroy the channel, @@ -373,7 +371,7 @@ mod test { let timer_rx = timer.periodic(Duration::milliseconds(1000)); Thread::spawn(move|| { - let _ = timer_rx.recv_opt(); + let _ = timer_rx.recv(); }).detach(); timer.oneshot(Duration::milliseconds(1)); @@ -386,7 +384,7 @@ mod test { let timer_rx = timer.periodic(Duration::milliseconds(1000)); Thread::spawn(move|| { - let _ = timer_rx.recv_opt(); + let _ = timer_rx.recv(); }).detach(); timer.sleep(Duration::milliseconds(1)); @@ -398,7 +396,7 @@ mod test { let mut timer = Timer::new().unwrap(); timer.oneshot(Duration::milliseconds(1000)) }; - assert_eq!(rx.recv_opt(), Err(())); + assert!(rx.recv().is_err()); } #[test] @@ -407,7 +405,7 @@ mod test { let mut timer = Timer::new().unwrap(); timer.periodic(Duration::milliseconds(1000)) }; - assert_eq!(rx.recv_opt(), Err(())); + assert!(rx.recv().is_err()); } #[test] @@ -446,34 +444,34 @@ mod test { fn oneshot_zero() { let mut timer = Timer::new().unwrap(); let rx = timer.oneshot(Duration::milliseconds(0)); - rx.recv(); + rx.recv().unwrap(); } #[test] fn oneshot_negative() { let mut timer = Timer::new().unwrap(); let rx = timer.oneshot(Duration::milliseconds(-1000000)); - rx.recv(); + rx.recv().unwrap(); } #[test] fn periodic_zero() { let mut timer = Timer::new().unwrap(); let rx = timer.periodic(Duration::milliseconds(0)); - rx.recv(); - rx.recv(); - rx.recv(); - rx.recv(); + rx.recv().unwrap(); + rx.recv().unwrap(); + rx.recv().unwrap(); + rx.recv().unwrap(); } #[test] fn periodic_negative() { let mut timer = Timer::new().unwrap(); let rx = timer.periodic(Duration::milliseconds(-1000000)); - rx.recv(); - rx.recv(); - rx.recv(); - rx.recv(); + rx.recv().unwrap(); + rx.recv().unwrap(); + rx.recv().unwrap(); + rx.recv().unwrap(); } } diff --git a/src/libstd/io/util.rs b/src/libstd/io/util.rs index e4888f3dd97..b22090e0f16 100644 --- a/src/libstd/io/util.rs +++ b/src/libstd/io/util.rs @@ -384,7 +384,7 @@ mod test { let mut r = TeeReader::new(MemReader::new(vec!(0, 1, 2)), Vec::new()); assert_eq!(vec!(0, 1, 2), r.read_to_end().unwrap()); - let (_, w) = r.unwrap(); + let (_, w) = r.into_inner(); assert_eq!(vec!(0, 1, 2), w); } diff --git a/src/libstd/lib.rs b/src/libstd/lib.rs index 56d906c8b69..aa1a0add068 100644 --- a/src/libstd/lib.rs +++ b/src/libstd/lib.rs @@ -226,7 +226,6 @@ pub mod hash; pub mod task; pub mod thread; pub mod sync; -pub mod comm; #[cfg(unix)] #[path = "sys/unix/mod.rs"] mod sys; @@ -254,7 +253,7 @@ mod std { pub use cmp; pub use hash; - pub use comm; // used for select!() + pub use sync; // used for select!() pub use error; // used for try!() pub use fmt; // used for any formatting strings pub use io; // used for println!() diff --git a/src/libstd/macros.rs b/src/libstd/macros.rs index 3606f5df5ca..51a0853687e 100644 --- a/src/libstd/macros.rs +++ b/src/libstd/macros.rs @@ -474,7 +474,7 @@ macro_rules! vec { /// /// ``` /// use std::thread::Thread; -/// use std::comm::channel; +/// use std::sync::mpsc::channel; /// /// let (tx1, rx1) = channel(); /// let (tx2, rx2) = channel(); @@ -485,21 +485,21 @@ macro_rules! vec { /// Thread::spawn(move|| { tx2.send(calculate_the_answer()) }).detach(); /// /// select! ( -/// () = rx1.recv() => println!("the long running task finished first"), +/// _ = rx1.recv() => println!("the long running task finished first"), /// answer = rx2.recv() => { -/// println!("the answer was: {}", answer); +/// println!("the answer was: {}", answer.unwrap()); /// } /// ) /// ``` /// -/// For more information about select, see the `std::comm::Select` structure. +/// For more information about select, see the `std::sync::mpsc::Select` structure. #[macro_export] #[experimental] macro_rules! select { ( $($name:pat = $rx:ident.$meth:ident() => $code:expr),+ ) => ({ - use std::comm::Select; + use std::sync::mpsc::Select; let sel = Select::new(); $( let mut $rx = sel.handle(&$rx); )+ unsafe { diff --git a/src/libstd/num/f32.rs b/src/libstd/num/f32.rs index 6de49c38b73..f2a0419e391 100644 --- a/src/libstd/num/f32.rs +++ b/src/libstd/num/f32.rs @@ -496,23 +496,25 @@ mod tests { #[test] fn test_real_consts() { - let pi: f32 = Float::pi(); - let two_pi: f32 = Float::two_pi(); - let frac_pi_2: f32 = Float::frac_pi_2(); - let frac_pi_3: f32 = Float::frac_pi_3(); - let frac_pi_4: f32 = Float::frac_pi_4(); - let frac_pi_6: f32 = Float::frac_pi_6(); - let frac_pi_8: f32 = Float::frac_pi_8(); - let frac_1_pi: f32 = Float::frac_1_pi(); - let frac_2_pi: f32 = Float::frac_2_pi(); - let frac_2_sqrtpi: f32 = Float::frac_2_sqrtpi(); - let sqrt2: f32 = Float::sqrt2(); - let frac_1_sqrt2: f32 = Float::frac_1_sqrt2(); - let e: f32 = Float::e(); - let log2_e: f32 = Float::log2_e(); - let log10_e: f32 = Float::log10_e(); - let ln_2: f32 = Float::ln_2(); - let ln_10: f32 = Float::ln_10(); + use super::consts; + + let pi: f32 = consts::PI; + let two_pi: f32 = consts::PI_2; + let frac_pi_2: f32 = consts::FRAC_PI_2; + let frac_pi_3: f32 = consts::FRAC_PI_3; + let frac_pi_4: f32 = consts::FRAC_PI_4; + let frac_pi_6: f32 = consts::FRAC_PI_6; + let frac_pi_8: f32 = consts::FRAC_PI_8; + let frac_1_pi: f32 = consts::FRAC_1_PI; + let frac_2_pi: f32 = consts::FRAC_2_PI; + let frac_2_sqrtpi: f32 = consts::FRAC_2_SQRTPI; + let sqrt2: f32 = consts::SQRT2; + let frac_1_sqrt2: f32 = consts::FRAC_1_SQRT2; + let e: f32 = consts::E; + let log2_e: f32 = consts::LOG2_E; + let log10_e: f32 = consts::LOG10_E; + let ln_2: f32 = consts::LN_2; + let ln_10: f32 = consts::LN_10; assert_approx_eq!(two_pi, 2f32 * pi); assert_approx_eq!(frac_pi_2, pi / 2f32); diff --git a/src/libstd/num/f64.rs b/src/libstd/num/f64.rs index 50855d21b4a..105a8a23bd1 100644 --- a/src/libstd/num/f64.rs +++ b/src/libstd/num/f64.rs @@ -499,23 +499,24 @@ mod tests { #[test] fn test_real_consts() { - let pi: f64 = Float::pi(); - let two_pi: f64 = Float::two_pi(); - let frac_pi_2: f64 = Float::frac_pi_2(); - let frac_pi_3: f64 = Float::frac_pi_3(); - let frac_pi_4: f64 = Float::frac_pi_4(); - let frac_pi_6: f64 = Float::frac_pi_6(); - let frac_pi_8: f64 = Float::frac_pi_8(); - let frac_1_pi: f64 = Float::frac_1_pi(); - let frac_2_pi: f64 = Float::frac_2_pi(); - let frac_2_sqrtpi: f64 = Float::frac_2_sqrtpi(); - let sqrt2: f64 = Float::sqrt2(); - let frac_1_sqrt2: f64 = Float::frac_1_sqrt2(); - let e: f64 = Float::e(); - let log2_e: f64 = Float::log2_e(); - let log10_e: f64 = Float::log10_e(); - let ln_2: f64 = Float::ln_2(); - let ln_10: f64 = Float::ln_10(); + use super::consts; + let pi: f64 = consts::PI; + let two_pi: f64 = consts::PI_2; + let frac_pi_2: f64 = consts::FRAC_PI_2; + let frac_pi_3: f64 = consts::FRAC_PI_3; + let frac_pi_4: f64 = consts::FRAC_PI_4; + let frac_pi_6: f64 = consts::FRAC_PI_6; + let frac_pi_8: f64 = consts::FRAC_PI_8; + let frac_1_pi: f64 = consts::FRAC_1_PI; + let frac_2_pi: f64 = consts::FRAC_2_PI; + let frac_2_sqrtpi: f64 = consts::FRAC_2_SQRTPI; + let sqrt2: f64 = consts::SQRT2; + let frac_1_sqrt2: f64 = consts::FRAC_1_SQRT2; + let e: f64 = consts::E; + let log2_e: f64 = consts::LOG2_E; + let log10_e: f64 = consts::LOG10_E; + let ln_2: f64 = consts::LN_2; + let ln_10: f64 = consts::LN_10; assert_approx_eq!(two_pi, 2.0 * pi); assert_approx_eq!(frac_pi_2, pi / 2f64); diff --git a/src/libstd/os.rs b/src/libstd/os.rs index 61ac89239f9..754b13886af 100644 --- a/src/libstd/os.rs +++ b/src/libstd/os.rs @@ -1425,7 +1425,8 @@ mod arch_consts { #[cfg(test)] mod tests { use prelude::v1::*; - use option; + + use iter::repeat; use os::{env, getcwd, getenv, make_absolute}; use os::{split_paths, join_paths, setenv, unsetenv}; use os; @@ -1454,7 +1455,7 @@ mod tests { fn test_setenv() { let n = make_rand_name(); setenv(n.as_slice(), "VALUE"); - assert_eq!(getenv(n.as_slice()), option::Option::Some("VALUE".to_string())); + assert_eq!(getenv(n.as_slice()), Some("VALUE".to_string())); } #[test] @@ -1462,7 +1463,7 @@ mod tests { let n = make_rand_name(); setenv(n.as_slice(), "VALUE"); unsetenv(n.as_slice()); - assert_eq!(getenv(n.as_slice()), option::Option::None); + assert_eq!(getenv(n.as_slice()), None); } #[test] @@ -1471,9 +1472,9 @@ mod tests { let n = make_rand_name(); setenv(n.as_slice(), "1"); setenv(n.as_slice(), "2"); - assert_eq!(getenv(n.as_slice()), option::Option::Some("2".to_string())); + assert_eq!(getenv(n.as_slice()), Some("2".to_string())); setenv(n.as_slice(), ""); - assert_eq!(getenv(n.as_slice()), option::Option::Some("".to_string())); + assert_eq!(getenv(n.as_slice()), Some("".to_string())); } // Windows GetEnvironmentVariable requires some extra work to make sure @@ -1490,7 +1491,7 @@ mod tests { let n = make_rand_name(); setenv(n.as_slice(), s.as_slice()); debug!("{}", s.clone()); - assert_eq!(getenv(n.as_slice()), option::Option::Some(s)); + assert_eq!(getenv(n.as_slice()), Some(s)); } #[test] @@ -1527,14 +1528,14 @@ mod tests { // MingW seems to set some funky environment variables like // "=C:=C:\MinGW\msys\1.0\bin" and "!::=::\" that are returned // from env() but not visible from getenv(). - assert!(v2.is_none() || v2 == option::Option::Some(v)); + assert!(v2.is_none() || v2 == Some(v)); } } #[test] fn test_env_set_get_huge() { let n = make_rand_name(); - let s = "x".repeat(10000).to_string(); + let s = repeat("x").take(10000).collect::<String>(); setenv(n.as_slice(), s.as_slice()); assert_eq!(getenv(n.as_slice()), Some(s)); unsetenv(n.as_slice()); @@ -1656,8 +1657,8 @@ mod tests { path.push("mmap_file.tmp"); let size = MemoryMap::granularity() * 2; let mut file = File::open_mode(&path, Open, ReadWrite).unwrap(); - file.seek(size as i64, SeekSet); - file.write_u8(0); + file.seek(size as i64, SeekSet).unwrap(); + file.write_u8(0).unwrap(); let chunk = MemoryMap::new(size / 2, &[ MapOption::MapReadable, diff --git a/src/libstd/path/posix.rs b/src/libstd/path/posix.rs index da4576ca36d..5c415239c5e 100644 --- a/src/libstd/path/posix.rs +++ b/src/libstd/path/posix.rs @@ -449,7 +449,6 @@ static dot_dot_static: &'static [u8] = b".."; mod tests { use prelude::v1::*; use str; - use super::*; macro_rules! t { (s: $path:expr, $exp:expr) => ( diff --git a/src/libstd/path/windows.rs b/src/libstd/path/windows.rs index e6fea07b3aa..9117827ffc2 100644 --- a/src/libstd/path/windows.rs +++ b/src/libstd/path/windows.rs @@ -1119,10 +1119,13 @@ fn prefix_len(p: Option<PathPrefix>) -> uint { #[cfg(test)] mod tests { - use prelude::v1::*; - use super::*; + use prelude::v1::Option::{mod, Some, None}; + use prelude::v1::{Vec, Clone, AsSlice, SliceExt, CloneSliceExt, IteratorExt}; + use prelude::v1::{DoubleEndedIteratorExt, Str, ToString, GenericPath}; + use super::PathPrefix::*; use super::parse_prefix; + use super::*; macro_rules! t { (s: $path:expr, $exp:expr) => ( diff --git a/src/libstd/rand/os.rs b/src/libstd/rand/os.rs index 95238c0dad9..0a7b9d07636 100644 --- a/src/libstd/rand/os.rs +++ b/src/libstd/rand/os.rs @@ -337,7 +337,7 @@ mod imp { mod test { use prelude::v1::*; - use comm::channel; + use sync::mpsc::channel; use rand::Rng; use super::OsRng; use thread::Thread; @@ -363,7 +363,7 @@ mod test { Thread::spawn(move|| { // wait until all the tasks are ready to go. - rx.recv(); + rx.recv().unwrap(); // deschedule to attempt to interleave things as much // as possible (XXX: is this a good test?) @@ -384,7 +384,7 @@ mod test { // start all the tasks for tx in txs.iter() { - tx.send(()) + tx.send(()).unwrap(); } } } diff --git a/src/libstd/sync/barrier.rs b/src/libstd/sync/barrier.rs index 15a682ad3b8..55d50af3b83 100644 --- a/src/libstd/sync/barrier.rs +++ b/src/libstd/sync/barrier.rs @@ -92,7 +92,7 @@ mod tests { use prelude::v1::*; use sync::{Arc, Barrier}; - use comm::{channel, Empty}; + use sync::mpsc::{channel, TryRecvError}; use thread::Thread; #[test] @@ -105,21 +105,21 @@ mod tests { let tx = tx.clone(); Thread::spawn(move|| { c.wait(); - tx.send(true); + tx.send(true).unwrap(); }).detach(); } // At this point, all spawned tasks should be blocked, // so we shouldn't get anything from the port assert!(match rx.try_recv() { - Err(Empty) => true, + Err(TryRecvError::Empty) => true, _ => false, }); barrier.wait(); // Now, the barrier is cleared and we should get data. for _ in range(0u, 9) { - rx.recv(); + rx.recv().unwrap(); } } } diff --git a/src/libstd/sync/condvar.rs b/src/libstd/sync/condvar.rs index 984b895b31d..28960c1574e 100644 --- a/src/libstd/sync/condvar.rs +++ b/src/libstd/sync/condvar.rs @@ -264,8 +264,8 @@ impl StaticCondvar { mod tests { use prelude::v1::*; - use comm::channel; use super::{StaticCondvar, CONDVAR_INIT}; + use sync::mpsc::channel; use sync::{StaticMutex, MUTEX_INIT, Condvar, Mutex, Arc}; use thread::Thread; use time::Duration; @@ -314,25 +314,25 @@ mod tests { let mut cnt = lock.lock(); *cnt += 1; if *cnt == N { - tx.send(()); + tx.send(()).unwrap(); } while *cnt != 0 { cond.wait(&cnt); } - tx.send(()); + tx.send(()).unwrap(); }).detach(); } drop(tx); let &(ref lock, ref cond) = &*data; - rx.recv(); + rx.recv().unwrap(); let mut cnt = lock.lock(); *cnt = 0; cond.notify_all(); drop(cnt); for _ in range(0, N) { - rx.recv(); + rx.recv().unwrap(); } } diff --git a/src/libstd/sync/future.rs b/src/libstd/sync/future.rs index a0e7236b8d2..e3620617d57 100644 --- a/src/libstd/sync/future.rs +++ b/src/libstd/sync/future.rs @@ -28,7 +28,7 @@ use core::prelude::*; use core::mem::replace; use self::FutureState::*; -use comm::{Receiver, channel}; +use sync::mpsc::{Receiver, channel}; use thunk::{Thunk}; use thread::Thread; @@ -122,8 +122,8 @@ impl<A:Send> Future<A> { * waiting for the result to be received on the port. */ - Future::from_fn(move|:| { - rx.recv() + Future::from_fn(move |:| { + rx.recv().unwrap() }) } @@ -141,7 +141,7 @@ impl<A:Send> Future<A> { Thread::spawn(move |:| { // Don't panic if the other end has hung up - let _ = tx.send_opt(blk()); + let _ = tx.send(blk()); }).detach(); Future::from_receiver(rx) @@ -151,7 +151,7 @@ impl<A:Send> Future<A> { #[cfg(test)] mod test { use prelude::v1::*; - use comm::channel; + use sync::mpsc::channel; use sync::Future; use thread::Thread; @@ -164,7 +164,7 @@ mod test { #[test] fn test_from_receiver() { let (tx, rx) = channel(); - tx.send("whale".to_string()); + tx.send("whale".to_string()).unwrap(); let mut f = Future::from_receiver(rx); assert_eq!(f.get(), "whale"); } @@ -184,7 +184,7 @@ mod test { #[test] fn test_interface_unwrap() { let f = Future::from_value("fail".to_string()); - assert_eq!(f.unwrap(), "fail"); + assert_eq!(f.into_inner(), "fail"); } #[test] @@ -213,8 +213,8 @@ mod test { let f = Future::spawn(move|| { expected }); let _t = Thread::spawn(move|| { let mut f = f; - tx.send(f.get()); + tx.send(f.get()).unwrap(); }); - assert_eq!(rx.recv(), expected); + assert_eq!(rx.recv().unwrap(), expected); } } diff --git a/src/libstd/sync/mod.rs b/src/libstd/sync/mod.rs index 7605a6a96a0..2b7311e4e98 100644 --- a/src/libstd/sync/mod.rs +++ b/src/libstd/sync/mod.rs @@ -32,6 +32,8 @@ pub use self::future::Future; pub use self::task_pool::TaskPool; pub mod atomic; +pub mod mpsc; + mod barrier; mod condvar; mod future; diff --git a/src/libstd/comm/blocking.rs b/src/libstd/sync/mpsc/blocking.rs index 412b7161305..412b7161305 100644 --- a/src/libstd/comm/blocking.rs +++ b/src/libstd/sync/mpsc/blocking.rs diff --git a/src/libstd/comm/mod.rs b/src/libstd/sync/mpsc/mod.rs index de7f3d00478..e2294906229 100644 --- a/src/libstd/comm/mod.rs +++ b/src/libstd/sync/mpsc/mod.rs @@ -8,12 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -//! Communication primitives for concurrent tasks -//! -//! Rust makes it very difficult to share data among tasks to prevent race -//! conditions and to improve parallelism, but there is often a need for -//! communication between concurrent tasks. The primitives defined in this -//! module are the building blocks for synchronization in rust. +//! Multi-producer, single-consumer communication primitives threads //! //! This module provides message-based communication over channels, concretely //! defined among three types: @@ -23,12 +18,10 @@ //! * `Receiver` //! //! A `Sender` or `SyncSender` is used to send data to a `Receiver`. Both -//! senders are clone-able such that many tasks can send simultaneously to one -//! receiver. These channels are *task blocking*, not *thread blocking*. This -//! means that if one task is blocked on a channel, other tasks can continue to -//! make progress. +//! senders are clone-able (multi-producer) such that many threads can send +//! simultaneously to one receiver (single-consumer). These channels are //! -//! Rust channels come in one of two flavors: +//! These channels come in two flavors: //! //! 1. An asynchronous, infinitely buffered channel. The `channel()` function //! will return a `(Sender, Receiver)` tuple where all sends will be @@ -43,38 +36,39 @@ //! "rendezvous" channel where each sender atomically hands off a message to //! a receiver. //! -//! ## Panic Propagation +//! ## Disconnection //! -//! In addition to being a core primitive for communicating in rust, channels -//! are the points at which panics are propagated among tasks. Whenever the one -//! half of channel is closed, the other half will have its next operation -//! `panic!`. The purpose of this is to allow propagation of panics among tasks -//! that are linked to one another via channels. +//! The send and receive operations on channels will all return a `Result` +//! indicating whether the operation succeeded or not. An unsuccessful operation +//! is normally indicative of the other half of a channel having "hung up" by +//! being dropped in its corresponding thread. //! -//! There are methods on both of senders and receivers to perform their -//! respective operations without panicking, however. +//! Once half of a channel has been deallocated, most operations can no longer +//! continue to make progress, so `Err` will be returned. Many applications will +//! continue to `unwrap()` the results returned from this module, instigating a +//! propagation of failure among threads if one unexpectedly dies. //! -//! # Example +//! # Examples //! //! Simple usage: //! //! ``` //! use std::thread::Thread; -//! use std::comm::channel; +//! use std::sync::mpsc::channel; //! //! // Create a simple streaming channel //! let (tx, rx) = channel(); //! Thread::spawn(move|| { -//! tx.send(10i); +//! tx.send(10i).unwrap(); //! }).detach(); -//! assert_eq!(rx.recv(), 10i); +//! assert_eq!(rx.recv().unwrap(), 10i); //! ``` //! //! Shared usage: //! //! ``` //! use std::thread::Thread; -//! use std::comm::channel; +//! use std::sync::mpsc::channel; //! //! // Create a shared channel that can be sent along from many threads //! // where tx is the sending half (tx for transmission), and rx is the receiving @@ -83,40 +77,40 @@ //! for i in range(0i, 10i) { //! let tx = tx.clone(); //! Thread::spawn(move|| { -//! tx.send(i); +//! tx.send(i).unwrap(); //! }).detach() //! } //! //! for _ in range(0i, 10i) { -//! let j = rx.recv(); +//! let j = rx.recv().unwrap(); //! assert!(0 <= j && j < 10); //! } //! ``` //! //! Propagating panics: //! -//! ```should_fail -//! use std::comm::channel; +//! ``` +//! use std::sync::mpsc::channel; //! -//! // The call to recv() will panic!() because the channel has already hung -//! // up (or been deallocated) +//! // The call to recv() will return an error because the channel has already +//! // hung up (or been deallocated) //! let (tx, rx) = channel::<int>(); //! drop(tx); -//! rx.recv(); +//! assert!(rx.recv().is_err()); //! ``` //! //! Synchronous channels: //! //! ``` //! use std::thread::Thread; -//! use std::comm::sync_channel; +//! use std::sync::mpsc::sync_channel; //! //! let (tx, rx) = sync_channel::<int>(0); //! Thread::spawn(move|| { //! // This will wait for the parent task to start receiving -//! tx.send(53); +//! tx.send(53).unwrap(); //! }).detach(); -//! rx.recv(); +//! rx.recv().unwrap(); //! ``` //! //! Reading from a channel with a timeout requires to use a Timer together @@ -125,7 +119,7 @@ //! after 10 seconds no matter what: //! //! ```no_run -//! use std::comm::channel; +//! use std::sync::mpsc::channel; //! use std::io::timer::Timer; //! use std::time::Duration; //! @@ -135,8 +129,8 @@ //! //! loop { //! select! { -//! val = rx.recv() => println!("Received {}", val), -//! () = timeout.recv() => { +//! val = rx.recv() => println!("Received {}", val.unwrap()), +//! _ = timeout.recv() => { //! println!("timed out, total time was more than 10 seconds"); //! break; //! } @@ -149,7 +143,7 @@ //! has been inactive for 5 seconds: //! //! ```no_run -//! use std::comm::channel; +//! use std::sync::mpsc::channel; //! use std::io::timer::Timer; //! use std::time::Duration; //! @@ -160,8 +154,8 @@ //! let timeout = timer.oneshot(Duration::seconds(5)); //! //! select! { -//! val = rx.recv() => println!("Received {}", val), -//! () = timeout.recv() => { +//! val = rx.recv() => println!("Received {}", val.unwrap()), +//! _ = timeout.recv() => { //! println!("timed out, no message received in 5 seconds"); //! break; //! } @@ -319,17 +313,13 @@ // And now that you've seen all the races that I found and attempted to fix, // here's the code for you to find some more! -use core::prelude::*; +use prelude::v1::*; -pub use self::TryRecvError::*; -pub use self::TrySendError::*; -use self::Flavor::*; - -use alloc::arc::Arc; -use core::kinds; -use core::kinds::marker; -use core::mem; -use core::cell::UnsafeCell; +use sync::Arc; +use fmt; +use kinds::marker; +use mem; +use cell::UnsafeCell; pub use self::select::{Select, Handle}; use self::select::StartResult; @@ -347,7 +337,7 @@ mod spsc_queue; /// The receiving-half of Rust's channel type. This half can only be owned by /// one task -#[unstable] +#[stable] pub struct Receiver<T> { inner: UnsafeCell<Flavor<T>>, } @@ -359,14 +349,14 @@ unsafe impl<T:Send> Send for Receiver<T> { } /// An iterator over messages on a receiver, this iterator will block /// whenever `next` is called, waiting for a new message, and `None` will be /// returned when the corresponding channel has hung up. -#[unstable] -pub struct Messages<'a, T:'a> { +#[stable] +pub struct Iter<'a, T:'a> { rx: &'a Receiver<T> } /// The sending-half of Rust's asynchronous channel type. This half can only be /// owned by one task, but it can be cloned to send to other tasks. -#[unstable] +#[stable] pub struct Sender<T> { inner: UnsafeCell<Flavor<T>>, } @@ -377,30 +367,50 @@ unsafe impl<T:Send> Send for Sender<T> { } /// The sending-half of Rust's synchronous channel type. This half can only be /// owned by one task, but it can be cloned to send to other tasks. -#[unstable = "this type may be renamed, but it will always exist"] +#[stable] pub struct SyncSender<T> { inner: Arc<RacyCell<sync::Packet<T>>>, // can't share in an arc _marker: marker::NoSync, } +/// An error returned from the `send` function on channels. +/// +/// A `send` operation can only fail if the receiving end of a channel is +/// disconnected, implying that the data could never be received. The error +/// contains the data being sent as a payload so it can be recovered. +#[deriving(PartialEq, Eq)] +#[stable] +pub struct SendError<T>(pub T); + +/// An error returned from the `recv` function on a `Receiver`. +/// +/// The `recv` operation can only fail if the sending half of a channel is +/// disconnected, implying that no further messages will ever be received. +#[deriving(PartialEq, Eq, Clone, Copy)] +#[stable] +pub struct RecvError; + /// This enumeration is the list of the possible reasons that try_recv could not /// return data when called. -#[deriving(PartialEq, Clone, Copy, Show)] -#[experimental = "this is likely to be removed in changing try_recv()"] +#[deriving(PartialEq, Clone, Copy)] +#[stable] pub enum TryRecvError { /// This channel is currently empty, but the sender(s) have not yet /// disconnected, so data may yet become available. + #[stable] Empty, + /// This channel's sending half has become disconnected, and there will /// never be any more data received on this channel + #[stable] Disconnected, } /// This enumeration is the list of the possible error outcomes for the /// `SyncSender::try_send` method. -#[deriving(PartialEq, Clone, Show)] -#[experimental = "this is likely to be removed in changing try_send()"] +#[deriving(PartialEq, Clone)] +#[stable] pub enum TrySendError<T> { /// The data could not be sent on the channel because it would require that /// the callee block to send the data. @@ -408,10 +418,13 @@ pub enum TrySendError<T> { /// If this is a buffered channel, then the buffer is full at this time. If /// this is not a buffered channel, then there is no receiver available to /// acquire the data. + #[stable] Full(T), + /// This channel's receiving half has disconnected, so the data could not be /// sent. The data is returned back to the callee in this case. - RecvDisconnected(T), + #[stable] + Disconnected(T), } enum Flavor<T> { @@ -450,7 +463,7 @@ impl<T> UnsafeFlavor<T> for Receiver<T> { /// # Example /// /// ``` -/// use std::comm::channel; +/// use std::sync::mpsc::channel; /// use std::thread::Thread; /// /// // tx is is the sending half (tx for transmission), and rx is the receiving @@ -460,18 +473,18 @@ impl<T> UnsafeFlavor<T> for Receiver<T> { /// // Spawn off an expensive computation /// Thread::spawn(move|| { /// # fn expensive_computation() {} -/// tx.send(expensive_computation()); +/// tx.send(expensive_computation()).unwrap(); /// }).detach(); /// /// // Do some useful work for awhile /// /// // Let's see what that answer was -/// println!("{}", rx.recv()); +/// println!("{}", rx.recv().unwrap()); /// ``` -#[unstable] +#[stable] pub fn channel<T: Send>() -> (Sender<T>, Receiver<T>) { let a = Arc::new(RacyCell::new(oneshot::Packet::new())); - (Sender::new(Oneshot(a.clone())), Receiver::new(Oneshot(a))) + (Sender::new(Flavor::Oneshot(a.clone())), Receiver::new(Flavor::Oneshot(a))) } /// Creates a new synchronous, bounded channel. @@ -492,27 +505,26 @@ pub fn channel<T: Send>() -> (Sender<T>, Receiver<T>) { /// # Example /// /// ``` -/// use std::comm::sync_channel; +/// use std::sync::mpsc::sync_channel; /// use std::thread::Thread; /// /// let (tx, rx) = sync_channel(1); /// /// // this returns immediately -/// tx.send(1i); +/// tx.send(1i).unwrap(); /// /// Thread::spawn(move|| { /// // this will block until the previous message has been received -/// tx.send(2i); +/// tx.send(2i).unwrap(); /// }).detach(); /// -/// assert_eq!(rx.recv(), 1i); -/// assert_eq!(rx.recv(), 2i); +/// assert_eq!(rx.recv().unwrap(), 1i); +/// assert_eq!(rx.recv().unwrap(), 2i); /// ``` -#[unstable = "this function may be renamed to more accurately reflect the type \ - of channel that is is creating"] +#[stable] pub fn sync_channel<T: Send>(bound: uint) -> (SyncSender<T>, Receiver<T>) { let a = Arc::new(RacyCell::new(sync::Packet::new(bound))); - (SyncSender::new(a.clone()), Receiver::new(Sync(a))) + (SyncSender::new(a.clone()), Receiver::new(Flavor::Sync(a))) } //////////////////////////////////////////////////////////////////////////////// @@ -526,33 +538,6 @@ impl<T: Send> Sender<T> { } } - /// Sends a value along this channel to be received by the corresponding - /// receiver. - /// - /// Rust channels are infinitely buffered so this method will never block. - /// - /// # Panics - /// - /// This function will panic if the other end of the channel has hung up. - /// This means that if the corresponding receiver has fallen out of scope, - /// this function will trigger a panic message saying that a message is - /// being sent on a closed channel. - /// - /// Note that if this function does *not* panic, it does not mean that the - /// data will be successfully received. All sends are placed into a queue, - /// so it is possible for a send to succeed (the other end is alive), but - /// then the other end could immediately disconnect. - /// - /// The purpose of this functionality is to propagate panics among tasks. - /// If a panic is not desired, then consider using the `send_opt` method - #[experimental = "this function is being considered candidate for removal \ - to adhere to the general guidelines of rust"] - pub fn send(&self, t: T) { - if self.send_opt(t).is_err() { - panic!("sending on a closed channel"); - } - } - /// Attempts to send a value on this channel, returning it back if it could /// not be sent. /// @@ -564,39 +549,34 @@ impl<T: Send> Sender<T> { /// will be received. It is possible for the corresponding receiver to /// hang up immediately after this function returns `Ok`. /// - /// Like `send`, this method will never block. - /// - /// # Panics - /// - /// This method will never panic, it will return the message back to the - /// caller if the other end is disconnected + /// This method will never block the current thread. /// /// # Example /// /// ``` - /// use std::comm::channel; + /// use std::sync::mpsc::channel; /// /// let (tx, rx) = channel(); /// /// // This send is always successful - /// assert_eq!(tx.send_opt(1i), Ok(())); + /// tx.send(1i).unwrap(); /// /// // This send will fail because the receiver is gone /// drop(rx); - /// assert_eq!(tx.send_opt(1i), Err(1)); + /// assert_eq!(tx.send(1i).err().unwrap().0, 1); /// ``` - #[unstable = "this function may be renamed to send() in the future"] - pub fn send_opt(&self, t: T) -> Result<(), T> { + pub fn send(&self, t: T) -> Result<(), SendError<T>> { let (new_inner, ret) = match *unsafe { self.inner() } { - Oneshot(ref p) => { + Flavor::Oneshot(ref p) => { unsafe { let p = p.get(); if !(*p).sent() { - return (*p).send(t); + return (*p).send(t).map_err(SendError); } else { let a = Arc::new(RacyCell::new(stream::Packet::new())); - match (*p).upgrade(Receiver::new(Stream(a.clone()))) { + let rx = Receiver::new(Flavor::Stream(a.clone())); + match (*p).upgrade(rx) { oneshot::UpSuccess => { let ret = (*a.get()).send(t); (a, ret) @@ -614,16 +594,20 @@ impl<T: Send> Sender<T> { } } } - Stream(ref p) => return unsafe { (*p.get()).send(t) }, - Shared(ref p) => return unsafe { (*p.get()).send(t) }, - Sync(..) => unreachable!(), + Flavor::Stream(ref p) => return unsafe { + (*p.get()).send(t).map_err(SendError) + }, + Flavor::Shared(ref p) => return unsafe { + (*p.get()).send(t).map_err(SendError) + }, + Flavor::Sync(..) => unreachable!(), }; unsafe { - let tmp = Sender::new(Stream(new_inner)); + let tmp = Sender::new(Flavor::Stream(new_inner)); mem::swap(self.inner_mut(), tmp.inner_mut()); } - return ret; + ret.map_err(SendError) } } @@ -631,42 +615,44 @@ impl<T: Send> Sender<T> { impl<T: Send> Clone for Sender<T> { fn clone(&self) -> Sender<T> { let (packet, sleeper, guard) = match *unsafe { self.inner() } { - Oneshot(ref p) => { + Flavor::Oneshot(ref p) => { let a = Arc::new(RacyCell::new(shared::Packet::new())); unsafe { let guard = (*a.get()).postinit_lock(); - match (*p.get()).upgrade(Receiver::new(Shared(a.clone()))) { + let rx = Receiver::new(Flavor::Shared(a.clone())); + match (*p.get()).upgrade(rx) { oneshot::UpSuccess | oneshot::UpDisconnected => (a, None, guard), oneshot::UpWoke(task) => (a, Some(task), guard) } } } - Stream(ref p) => { + Flavor::Stream(ref p) => { let a = Arc::new(RacyCell::new(shared::Packet::new())); unsafe { let guard = (*a.get()).postinit_lock(); - match (*p.get()).upgrade(Receiver::new(Shared(a.clone()))) { + let rx = Receiver::new(Flavor::Shared(a.clone())); + match (*p.get()).upgrade(rx) { stream::UpSuccess | stream::UpDisconnected => (a, None, guard), stream::UpWoke(task) => (a, Some(task), guard), } } } - Shared(ref p) => { + Flavor::Shared(ref p) => { unsafe { (*p.get()).clone_chan(); } - return Sender::new(Shared(p.clone())); + return Sender::new(Flavor::Shared(p.clone())); } - Sync(..) => unreachable!(), + Flavor::Sync(..) => unreachable!(), }; unsafe { (*packet.get()).inherit_blocker(sleeper, guard); - let tmp = Sender::new(Shared(packet.clone())); + let tmp = Sender::new(Flavor::Shared(packet.clone())); mem::swap(self.inner_mut(), tmp.inner_mut()); } - Sender::new(Shared(packet)) + Sender::new(Flavor::Shared(packet)) } } @@ -674,10 +660,10 @@ impl<T: Send> Clone for Sender<T> { impl<T: Send> Drop for Sender<T> { fn drop(&mut self) { match *unsafe { self.inner_mut() } { - Oneshot(ref mut p) => unsafe { (*p.get()).drop_chan(); }, - Stream(ref mut p) => unsafe { (*p.get()).drop_chan(); }, - Shared(ref mut p) => unsafe { (*p.get()).drop_chan(); }, - Sync(..) => unreachable!(), + Flavor::Oneshot(ref mut p) => unsafe { (*p.get()).drop_chan(); }, + Flavor::Stream(ref mut p) => unsafe { (*p.get()).drop_chan(); }, + Flavor::Shared(ref mut p) => unsafe { (*p.get()).drop_chan(); }, + Flavor::Sync(..) => unreachable!(), } } } @@ -697,59 +683,29 @@ impl<T: Send> SyncSender<T> { /// available or a receiver is available to hand off the message to. /// /// Note that a successful send does *not* guarantee that the receiver will - /// ever see the data if there is a buffer on this channel. Messages may be + /// ever see the data if there is a buffer on this channel. Items may be /// enqueued in the internal buffer for the receiver to receive at a later /// time. If the buffer size is 0, however, it can be guaranteed that the /// receiver has indeed received the data if this function returns success. /// - /// # Panics - /// - /// Similarly to `Sender::send`, this function will panic if the - /// corresponding `Receiver` for this channel has disconnected. This - /// behavior is used to propagate panics among tasks. - /// - /// If a panic is not desired, you can achieve the same semantics with the - /// `SyncSender::send_opt` method which will not panic if the receiver - /// disconnects. - #[experimental = "this function is being considered candidate for removal \ - to adhere to the general guidelines of rust"] - pub fn send(&self, t: T) { - if self.send_opt(t).is_err() { - panic!("sending on a closed channel"); - } - } - - /// Send a value on a channel, returning it back if the receiver - /// disconnected - /// - /// This method will *block* to send the value `t` on the channel, but if - /// the value could not be sent due to the receiver disconnecting, the value - /// is returned back to the callee. This function is similar to `try_send`, - /// except that it will block if the channel is currently full. - /// - /// # Panics - /// - /// This function cannot panic. - #[unstable = "this function may be renamed to send() in the future"] - pub fn send_opt(&self, t: T) -> Result<(), T> { - unsafe { (*self.inner.get()).send(t) } + /// This function will never panic, but it may return `Err` if the + /// `Receiver` has disconnected and is no longer able to receive + /// information. + #[stable] + pub fn send(&self, t: T) -> Result<(), SendError<T>> { + unsafe { (*self.inner.get()).send(t).map_err(SendError) } } /// Attempts to send a value on this channel without blocking. /// - /// This method differs from `send_opt` by returning immediately if the + /// This method differs from `send` by returning immediately if the /// channel's buffer is full or no receiver is waiting to acquire some - /// data. Compared with `send_opt`, this function has two failure cases + /// data. Compared with `send`, this function has two failure cases /// instead of one (one for disconnection, one for a full buffer). /// /// See `SyncSender::send` for notes about guarantees of whether the /// receiver has received the data or not if this function is successful. - /// - /// # Panics - /// - /// This function cannot panic - #[unstable = "the return type of this function is candidate for \ - modification"] + #[stable] pub fn try_send(&self, t: T) -> Result<(), TrySendError<T>> { unsafe { (*self.inner.get()).try_send(t) } } @@ -779,34 +735,6 @@ impl<T: Send> Receiver<T> { Receiver { inner: UnsafeCell::new(inner) } } - /// Blocks waiting for a value on this receiver - /// - /// This function will block if necessary to wait for a corresponding send - /// on the channel from its paired `Sender` structure. This receiver will - /// be woken up when data is ready, and the data will be returned. - /// - /// # Panics - /// - /// Similar to channels, this method will trigger a task panic if the - /// other end of the channel has hung up (been deallocated). The purpose of - /// this is to propagate panics among tasks. - /// - /// If a panic is not desired, then there are two options: - /// - /// * If blocking is still desired, the `recv_opt` method will return `None` - /// when the other end hangs up - /// - /// * If blocking is not desired, then the `try_recv` method will attempt to - /// peek at a value on this receiver. - #[experimental = "this function is being considered candidate for removal \ - to adhere to the general guidelines of rust"] - pub fn recv(&self) -> T { - match self.recv_opt() { - Ok(t) => t, - Err(()) => panic!("receiving on a closed channel"), - } - } - /// Attempts to return a pending value on this receiver without blocking /// /// This method will never block the caller in order to wait for data to @@ -815,42 +743,46 @@ impl<T: Send> Receiver<T> { /// /// This is useful for a flavor of "optimistic check" before deciding to /// block on a receiver. - /// - /// # Panics - /// - /// This function cannot panic. - #[unstable = "the return type of this function may be altered"] + #[stable] pub fn try_recv(&self) -> Result<T, TryRecvError> { loop { let new_port = match *unsafe { self.inner() } { - Oneshot(ref p) => { + Flavor::Oneshot(ref p) => { match unsafe { (*p.get()).try_recv() } { Ok(t) => return Ok(t), - Err(oneshot::Empty) => return Err(Empty), - Err(oneshot::Disconnected) => return Err(Disconnected), + Err(oneshot::Empty) => return Err(TryRecvError::Empty), + Err(oneshot::Disconnected) => { + return Err(TryRecvError::Disconnected) + } Err(oneshot::Upgraded(rx)) => rx, } } - Stream(ref p) => { + Flavor::Stream(ref p) => { match unsafe { (*p.get()).try_recv() } { Ok(t) => return Ok(t), - Err(stream::Empty) => return Err(Empty), - Err(stream::Disconnected) => return Err(Disconnected), + Err(stream::Empty) => return Err(TryRecvError::Empty), + Err(stream::Disconnected) => { + return Err(TryRecvError::Disconnected) + } Err(stream::Upgraded(rx)) => rx, } } - Shared(ref p) => { + Flavor::Shared(ref p) => { match unsafe { (*p.get()).try_recv() } { Ok(t) => return Ok(t), - Err(shared::Empty) => return Err(Empty), - Err(shared::Disconnected) => return Err(Disconnected), + Err(shared::Empty) => return Err(TryRecvError::Empty), + Err(shared::Disconnected) => { + return Err(TryRecvError::Disconnected) + } } } - Sync(ref p) => { + Flavor::Sync(ref p) => { match unsafe { (*p.get()).try_recv() } { Ok(t) => return Ok(t), - Err(sync::Empty) => return Err(Empty), - Err(sync::Disconnected) => return Err(Disconnected), + Err(sync::Empty) => return Err(TryRecvError::Empty), + Err(sync::Disconnected) => { + return Err(TryRecvError::Disconnected) + } } } }; @@ -861,46 +793,47 @@ impl<T: Send> Receiver<T> { } } - /// Attempt to wait for a value on this receiver, but does not panic if the + /// Attempt to wait for a value on this receiver, returning an error if the /// corresponding channel has hung up. /// - /// This implementation of iterators for ports will always block if there is - /// not data available on the receiver, but it will not panic in the case - /// that the channel has been deallocated. - /// - /// In other words, this function has the same semantics as the `recv` - /// method except for the panic aspect. + /// This function will always block the current thread if there is no data + /// available and it's possible for more data to be sent. Once a message is + /// sent to the corresponding `Sender`, then this receiver will wake up and + /// return that message. /// - /// If the channel has hung up, then `Err` is returned. Otherwise `Ok` of - /// the value found on the receiver is returned. - #[unstable = "this function may be renamed to recv()"] - pub fn recv_opt(&self) -> Result<T, ()> { + /// If the corresponding `Sender` has disconnected, or it disconnects while + /// this call is blocking, this call will wake up and return `Err` to + /// indicate that no more messages can ever be received on this channel. + #[stable] + pub fn recv(&self) -> Result<T, RecvError> { loop { let new_port = match *unsafe { self.inner() } { - Oneshot(ref p) => { + Flavor::Oneshot(ref p) => { match unsafe { (*p.get()).recv() } { Ok(t) => return Ok(t), Err(oneshot::Empty) => return unreachable!(), - Err(oneshot::Disconnected) => return Err(()), + Err(oneshot::Disconnected) => return Err(RecvError), Err(oneshot::Upgraded(rx)) => rx, } } - Stream(ref p) => { + Flavor::Stream(ref p) => { match unsafe { (*p.get()).recv() } { Ok(t) => return Ok(t), Err(stream::Empty) => return unreachable!(), - Err(stream::Disconnected) => return Err(()), + Err(stream::Disconnected) => return Err(RecvError), Err(stream::Upgraded(rx)) => rx, } } - Shared(ref p) => { + Flavor::Shared(ref p) => { match unsafe { (*p.get()).recv() } { Ok(t) => return Ok(t), Err(shared::Empty) => return unreachable!(), - Err(shared::Disconnected) => return Err(()), + Err(shared::Disconnected) => return Err(RecvError), } } - Sync(ref p) => return unsafe { (*p.get()).recv() } + Flavor::Sync(ref p) => return unsafe { + (*p.get()).recv().map_err(|()| RecvError) + } }; unsafe { mem::swap(self.inner_mut(), new_port.inner_mut()); @@ -910,9 +843,9 @@ impl<T: Send> Receiver<T> { /// Returns an iterator that will block waiting for messages, but never /// `panic!`. It will return `None` when the channel has hung up. - #[unstable] - pub fn iter<'a>(&'a self) -> Messages<'a, T> { - Messages { rx: self } + #[stable] + pub fn iter(&self) -> Iter<T> { + Iter { rx: self } } } @@ -920,22 +853,22 @@ impl<T: Send> select::Packet for Receiver<T> { fn can_recv(&self) -> bool { loop { let new_port = match *unsafe { self.inner() } { - Oneshot(ref p) => { + Flavor::Oneshot(ref p) => { match unsafe { (*p.get()).can_recv() } { Ok(ret) => return ret, Err(upgrade) => upgrade, } } - Stream(ref p) => { + Flavor::Stream(ref p) => { match unsafe { (*p.get()).can_recv() } { Ok(ret) => return ret, Err(upgrade) => upgrade, } } - Shared(ref p) => { + Flavor::Shared(ref p) => { return unsafe { (*p.get()).can_recv() }; } - Sync(ref p) => { + Flavor::Sync(ref p) => { return unsafe { (*p.get()).can_recv() }; } }; @@ -949,24 +882,24 @@ impl<T: Send> select::Packet for Receiver<T> { fn start_selection(&self, mut token: SignalToken) -> StartResult { loop { let (t, new_port) = match *unsafe { self.inner() } { - Oneshot(ref p) => { + Flavor::Oneshot(ref p) => { match unsafe { (*p.get()).start_selection(token) } { oneshot::SelSuccess => return Installed, oneshot::SelCanceled => return Abort, oneshot::SelUpgraded(t, rx) => (t, rx), } } - Stream(ref p) => { + Flavor::Stream(ref p) => { match unsafe { (*p.get()).start_selection(token) } { stream::SelSuccess => return Installed, stream::SelCanceled => return Abort, stream::SelUpgraded(t, rx) => (t, rx), } } - Shared(ref p) => { + Flavor::Shared(ref p) => { return unsafe { (*p.get()).start_selection(token) }; } - Sync(ref p) => { + Flavor::Sync(ref p) => { return unsafe { (*p.get()).start_selection(token) }; } }; @@ -981,14 +914,14 @@ impl<T: Send> select::Packet for Receiver<T> { let mut was_upgrade = false; loop { let result = match *unsafe { self.inner() } { - Oneshot(ref p) => unsafe { (*p.get()).abort_selection() }, - Stream(ref p) => unsafe { + Flavor::Oneshot(ref p) => unsafe { (*p.get()).abort_selection() }, + Flavor::Stream(ref p) => unsafe { (*p.get()).abort_selection(was_upgrade) }, - Shared(ref p) => return unsafe { + Flavor::Shared(ref p) => return unsafe { (*p.get()).abort_selection(was_upgrade) }, - Sync(ref p) => return unsafe { + Flavor::Sync(ref p) => return unsafe { (*p.get()).abort_selection() }, }; @@ -1003,18 +936,18 @@ impl<T: Send> select::Packet for Receiver<T> { } #[unstable] -impl<'a, T: Send> Iterator<T> for Messages<'a, T> { - fn next(&mut self) -> Option<T> { self.rx.recv_opt().ok() } +impl<'a, T: Send> Iterator<T> for Iter<'a, T> { + fn next(&mut self) -> Option<T> { self.rx.recv().ok() } } #[unsafe_destructor] impl<T: Send> Drop for Receiver<T> { fn drop(&mut self) { match *unsafe { self.inner_mut() } { - Oneshot(ref mut p) => unsafe { (*p.get()).drop_port(); }, - Stream(ref mut p) => unsafe { (*p.get()).drop_port(); }, - Shared(ref mut p) => unsafe { (*p.get()).drop_port(); }, - Sync(ref mut p) => unsafe { (*p.get()).drop_port(); }, + Flavor::Oneshot(ref mut p) => unsafe { (*p.get()).drop_port(); }, + Flavor::Stream(ref mut p) => unsafe { (*p.get()).drop_port(); }, + Flavor::Shared(ref mut p) => unsafe { (*p.get()).drop_port(); }, + Flavor::Sync(ref mut p) => unsafe { (*p.get()).drop_port(); }, } } } @@ -1037,8 +970,45 @@ impl<T> RacyCell<T> { unsafe impl<T:Send> Send for RacyCell<T> { } -unsafe impl<T> kinds::Sync for RacyCell<T> { } // Oh dear +unsafe impl<T> Sync for RacyCell<T> { } // Oh dear + +impl<T> fmt::Show for SendError<T> { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + "sending on a closed channel".fmt(f) + } +} + +impl<T> fmt::Show for TrySendError<T> { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + match *self { + TrySendError::Full(..) => { + "sending on a full channel".fmt(f) + } + TrySendError::Disconnected(..) => { + "sending on a closed channel".fmt(f) + } + } + } +} + +impl fmt::Show for RecvError { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + "receiving on a closed channel".fmt(f) + } +} +impl fmt::Show for TryRecvError { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + match *self { + TryRecvError::Empty => { + "receiving on an empty channel".fmt(f) + } + TryRecvError::Disconnected => { + "receiving on a closed channel".fmt(f) + } + } + } +} #[cfg(test)] mod test { @@ -1047,11 +1017,10 @@ mod test { use os; use super::*; use thread::Thread; - use str::from_str; pub fn stress_factor() -> uint { match os::getenv("RUST_TEST_STRESS") { - Some(val) => from_str::<uint>(val.as_slice()).unwrap(), + Some(val) => val.parse().unwrap(), None => 1, } } @@ -1059,14 +1028,14 @@ mod test { #[test] fn smoke() { let (tx, rx) = channel::<int>(); - tx.send(1); - assert_eq!(rx.recv(), 1); + tx.send(1).unwrap(); + assert_eq!(rx.recv().unwrap(), 1); } #[test] fn drop_full() { let (tx, _rx) = channel(); - tx.send(box 1i); + tx.send(box 1i).unwrap(); } #[test] @@ -1074,115 +1043,104 @@ mod test { let (tx, _rx) = channel(); drop(tx.clone()); drop(tx.clone()); - tx.send(box 1i); + tx.send(box 1i).unwrap(); } #[test] fn smoke_shared() { let (tx, rx) = channel::<int>(); - tx.send(1); - assert_eq!(rx.recv(), 1); + tx.send(1).unwrap(); + assert_eq!(rx.recv().unwrap(), 1); let tx = tx.clone(); - tx.send(1); - assert_eq!(rx.recv(), 1); + tx.send(1).unwrap(); + assert_eq!(rx.recv().unwrap(), 1); } #[test] fn smoke_threads() { let (tx, rx) = channel::<int>(); let _t = Thread::spawn(move|| { - tx.send(1); + tx.send(1).unwrap(); }); - assert_eq!(rx.recv(), 1); + assert_eq!(rx.recv().unwrap(), 1); } #[test] - #[should_fail] fn smoke_port_gone() { let (tx, rx) = channel::<int>(); drop(rx); - tx.send(1); + assert!(tx.send(1).is_err()); } #[test] - #[should_fail] fn smoke_shared_port_gone() { let (tx, rx) = channel::<int>(); drop(rx); - tx.send(1); + assert!(tx.send(1).is_err()) } #[test] - #[should_fail] fn smoke_shared_port_gone2() { let (tx, rx) = channel::<int>(); drop(rx); let tx2 = tx.clone(); drop(tx); - tx2.send(1); + assert!(tx2.send(1).is_err()); } #[test] - #[should_fail] fn port_gone_concurrent() { let (tx, rx) = channel::<int>(); - Thread::spawn(move|| { - rx.recv(); - }).detach(); - loop { tx.send(1) } + let _t = Thread::spawn(move|| { + rx.recv().unwrap(); + }); + while tx.send(1).is_ok() {} } #[test] - #[should_fail] fn port_gone_concurrent_shared() { let (tx, rx) = channel::<int>(); let tx2 = tx.clone(); - Thread::spawn(move|| { - rx.recv(); - }).detach(); - loop { - tx.send(1); - tx2.send(1); - } + let _t = Thread::spawn(move|| { + rx.recv().unwrap(); + }); + while tx.send(1).is_ok() && tx2.send(1).is_ok() {} } #[test] - #[should_fail] fn smoke_chan_gone() { let (tx, rx) = channel::<int>(); drop(tx); - rx.recv(); + assert!(rx.recv().is_err()); } #[test] - #[should_fail] fn smoke_chan_gone_shared() { let (tx, rx) = channel::<()>(); let tx2 = tx.clone(); drop(tx); drop(tx2); - rx.recv(); + assert!(rx.recv().is_err()); } #[test] - #[should_fail] fn chan_gone_concurrent() { let (tx, rx) = channel::<int>(); - Thread::spawn(move|| { - tx.send(1); - tx.send(1); - }).detach(); - loop { rx.recv(); } + let _t = Thread::spawn(move|| { + tx.send(1).unwrap(); + tx.send(1).unwrap(); + }); + while rx.recv().is_ok() {} } #[test] fn stress() { let (tx, rx) = channel::<int>(); let t = Thread::spawn(move|| { - for _ in range(0u, 10000) { tx.send(1i); } + for _ in range(0u, 10000) { tx.send(1i).unwrap(); } }); for _ in range(0u, 10000) { - assert_eq!(rx.recv(), 1); + assert_eq!(rx.recv().unwrap(), 1); } t.join().ok().unwrap(); } @@ -1195,7 +1153,7 @@ mod test { let t = Thread::spawn(move|| { for _ in range(0, AMT * NTHREADS) { - assert_eq!(rx.recv(), 1); + assert_eq!(rx.recv().unwrap(), 1); } match rx.try_recv() { Ok(..) => panic!(), @@ -1206,7 +1164,7 @@ mod test { for _ in range(0, NTHREADS) { let tx = tx.clone(); Thread::spawn(move|| { - for _ in range(0, AMT) { tx.send(1); } + for _ in range(0, AMT) { tx.send(1).unwrap(); } }).detach(); } drop(tx); @@ -1218,15 +1176,15 @@ mod test { let (tx1, rx1) = channel::<()>(); let (tx2, rx2) = channel::<int>(); let t1 = Thread::spawn(move|| { - tx1.send(()); + tx1.send(()).unwrap(); for _ in range(0i, 40) { - assert_eq!(rx2.recv(), 1); + assert_eq!(rx2.recv().unwrap(), 1); } }); - rx1.recv(); + rx1.recv().unwrap(); let t2 = Thread::spawn(move|| { for _ in range(0i, 40) { - tx2.send(1); + tx2.send(1).unwrap(); } }); t1.join().ok().unwrap(); @@ -1238,11 +1196,11 @@ mod test { let (tx, rx) = channel::<int>(); let t = Thread::spawn(move|| { for _ in range(0i, 40) { - assert_eq!(rx.recv(), 1); + assert_eq!(rx.recv().unwrap(), 1); } }); for _ in range(0u, 40) { - tx.send(1); + tx.send(1).unwrap(); } t.join().ok().unwrap(); } @@ -1252,12 +1210,12 @@ mod test { let (tx1, rx1) = channel::<int>(); let (tx2, rx2) = channel::<int>(); let t1 = Thread::spawn(move|| { - assert_eq!(rx1.recv(), 1); - tx2.send(2); + assert_eq!(rx1.recv().unwrap(), 1); + tx2.send(2).unwrap(); }); let t2 = Thread::spawn(move|| { - tx1.send(1); - assert_eq!(rx2.recv(), 2); + tx1.send(1).unwrap(); + assert_eq!(rx2.recv().unwrap(), 2); }); t1.join().ok().unwrap(); t2.join().ok().unwrap(); @@ -1278,12 +1236,11 @@ mod test { } #[test] - #[should_fail] fn oneshot_single_thread_send_port_close() { // Testing that the sender cleans up the payload if receiver is closed let (tx, rx) = channel::<Box<int>>(); drop(rx); - tx.send(box 0); + assert!(tx.send(box 0).is_err()); } #[test] @@ -1292,7 +1249,7 @@ mod test { let res = Thread::spawn(move|| { let (tx, rx) = channel::<int>(); drop(tx); - rx.recv(); + rx.recv().unwrap(); }).join(); // What is our res? assert!(res.is_err()); @@ -1301,43 +1258,43 @@ mod test { #[test] fn oneshot_single_thread_send_then_recv() { let (tx, rx) = channel::<Box<int>>(); - tx.send(box 10); - assert!(rx.recv() == box 10); + tx.send(box 10).unwrap(); + assert!(rx.recv().unwrap() == box 10); } #[test] fn oneshot_single_thread_try_send_open() { let (tx, rx) = channel::<int>(); - assert!(tx.send_opt(10).is_ok()); - assert!(rx.recv() == 10); + assert!(tx.send(10).is_ok()); + assert!(rx.recv().unwrap() == 10); } #[test] fn oneshot_single_thread_try_send_closed() { let (tx, rx) = channel::<int>(); drop(rx); - assert!(tx.send_opt(10).is_err()); + assert!(tx.send(10).is_err()); } #[test] fn oneshot_single_thread_try_recv_open() { let (tx, rx) = channel::<int>(); - tx.send(10); - assert!(rx.recv_opt() == Ok(10)); + tx.send(10).unwrap(); + assert!(rx.recv() == Ok(10)); } #[test] fn oneshot_single_thread_try_recv_closed() { let (tx, rx) = channel::<int>(); drop(tx); - assert!(rx.recv_opt() == Err(())); + assert!(rx.recv().is_err()); } #[test] fn oneshot_single_thread_peek_data() { let (tx, rx) = channel::<int>(); - assert_eq!(rx.try_recv(), Err(Empty)); - tx.send(10); + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); + tx.send(10).unwrap(); assert_eq!(rx.try_recv(), Ok(10)); } @@ -1345,24 +1302,24 @@ mod test { fn oneshot_single_thread_peek_close() { let (tx, rx) = channel::<int>(); drop(tx); - assert_eq!(rx.try_recv(), Err(Disconnected)); - assert_eq!(rx.try_recv(), Err(Disconnected)); + assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); + assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); } #[test] fn oneshot_single_thread_peek_open() { let (_tx, rx) = channel::<int>(); - assert_eq!(rx.try_recv(), Err(Empty)); + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); } #[test] fn oneshot_multi_task_recv_then_send() { let (tx, rx) = channel::<Box<int>>(); let _t = Thread::spawn(move|| { - assert!(rx.recv() == box 10); + assert!(rx.recv().unwrap() == box 10); }); - tx.send(box 10); + tx.send(box 10).unwrap(); } #[test] @@ -1372,7 +1329,7 @@ mod test { drop(tx); }); let res = Thread::spawn(move|| { - assert!(rx.recv() == box 10); + assert!(rx.recv().unwrap() == box 10); }).join(); assert!(res.is_err()); } @@ -1396,7 +1353,7 @@ mod test { drop(rx); }); let _ = Thread::spawn(move|| { - tx.send(1); + tx.send(1).unwrap(); }).join(); } } @@ -1407,7 +1364,7 @@ mod test { let (tx, rx) = channel::<int>(); Thread::spawn(move|| { let res = Thread::spawn(move|| { - rx.recv(); + rx.recv().unwrap(); }).join(); assert!(res.is_err()); }).detach(); @@ -1424,9 +1381,9 @@ mod test { for _ in range(0, stress_factor()) { let (tx, rx) = channel(); let _t = Thread::spawn(move|| { - tx.send(box 10i); + tx.send(box 10i).unwrap(); }); - assert!(rx.recv() == box 10i); + assert!(rx.recv().unwrap() == box 10i); } } @@ -1442,7 +1399,7 @@ mod test { if i == 10 { return } Thread::spawn(move|| { - tx.send(box i); + tx.send(box i).unwrap(); send(tx, i + 1); }).detach(); } @@ -1451,7 +1408,7 @@ mod test { if i == 10 { return } Thread::spawn(move|| { - assert!(rx.recv() == box i); + assert!(rx.recv().unwrap() == box i); recv(rx, i + 1); }).detach(); } @@ -1462,8 +1419,8 @@ mod test { fn recv_a_lot() { // Regression test that we don't run out of stack in scheduler context let (tx, rx) = channel(); - for _ in range(0i, 10000) { tx.send(()); } - for _ in range(0i, 10000) { rx.recv(); } + for _ in range(0i, 10000) { tx.send(()).unwrap(); } + for _ in range(0i, 10000) { rx.recv().unwrap(); } } #[test] @@ -1473,12 +1430,12 @@ mod test { for _ in range(0, total) { let tx = tx.clone(); Thread::spawn(move|| { - tx.send(()); + tx.send(()).unwrap(); }).detach(); } for _ in range(0, total) { - rx.recv(); + rx.recv().unwrap(); } } @@ -1492,14 +1449,14 @@ mod test { for x in rx.iter() { acc += x; } - total_tx.send(acc); + total_tx.send(acc).unwrap(); }); - tx.send(3); - tx.send(1); - tx.send(2); + tx.send(3).unwrap(); + tx.send(1).unwrap(); + tx.send(2).unwrap(); drop(tx); - assert_eq!(total_rx.recv(), 6); + assert_eq!(total_rx.recv().unwrap(), 6); } #[test] @@ -1516,15 +1473,15 @@ mod test { count += x; } } - count_tx.send(count); + count_tx.send(count).unwrap(); }); - tx.send(2); - tx.send(2); - tx.send(2); - let _ = tx.send_opt(2); + tx.send(2).unwrap(); + tx.send(2).unwrap(); + tx.send(2).unwrap(); + let _ = tx.send(2); drop(tx); - assert_eq!(count_rx.recv(), 4); + assert_eq!(count_rx.recv().unwrap(), 4); } #[test] @@ -1533,22 +1490,22 @@ mod test { let (tx2, rx2) = channel::<()>(); let (tx3, rx3) = channel::<()>(); let _t = Thread::spawn(move|| { - rx2.recv(); - tx1.send(1); - tx3.send(()); - rx2.recv(); + rx2.recv().unwrap(); + tx1.send(1).unwrap(); + tx3.send(()).unwrap(); + rx2.recv().unwrap(); drop(tx1); - tx3.send(()); + tx3.send(()).unwrap(); }); - assert_eq!(rx1.try_recv(), Err(Empty)); - tx2.send(()); - rx3.recv(); + assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty)); + tx2.send(()).unwrap(); + rx3.recv().unwrap(); assert_eq!(rx1.try_recv(), Ok(1)); - assert_eq!(rx1.try_recv(), Err(Empty)); - tx2.send(()); - rx3.recv(); - assert_eq!(rx1.try_recv(), Err(Disconnected)); + assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty)); + tx2.send(()).unwrap(); + rx3.recv().unwrap(); + assert_eq!(rx1.try_recv(), Err(TryRecvError::Disconnected)); } // This bug used to end up in a livelock inside of the Receiver destructor @@ -1558,9 +1515,9 @@ mod test { let (tx, rx) = channel(); let (tx2, rx2) = channel(); let _t = Thread::spawn(move|| { - rx.recv(); // wait on a oneshot + rx.recv().unwrap(); // wait on a oneshot drop(rx); // destroy a shared - tx2.send(()); + tx2.send(()).unwrap(); }); // make sure the other task has gone to sleep for _ in range(0u, 5000) { Thread::yield_now(); } @@ -1568,24 +1525,24 @@ mod test { // upgrade to a shared chan and send a message let t = tx.clone(); drop(tx); - t.send(()); + t.send(()).unwrap(); // wait for the child task to exit before we exit - rx2.recv(); + rx2.recv().unwrap(); } } #[cfg(test)] mod sync_tests { use prelude::v1::*; + use os; use thread::Thread; use super::*; - use str::from_str; pub fn stress_factor() -> uint { match os::getenv("RUST_TEST_STRESS") { - Some(val) => from_str::<uint>(val.as_slice()).unwrap(), + Some(val) => val.parse().unwrap(), None => 1, } } @@ -1593,114 +1550,104 @@ mod sync_tests { #[test] fn smoke() { let (tx, rx) = sync_channel::<int>(1); - tx.send(1); - assert_eq!(rx.recv(), 1); + tx.send(1).unwrap(); + assert_eq!(rx.recv().unwrap(), 1); } #[test] fn drop_full() { let (tx, _rx) = sync_channel(1); - tx.send(box 1i); + tx.send(box 1i).unwrap(); } #[test] fn smoke_shared() { let (tx, rx) = sync_channel::<int>(1); - tx.send(1); - assert_eq!(rx.recv(), 1); + tx.send(1).unwrap(); + assert_eq!(rx.recv().unwrap(), 1); let tx = tx.clone(); - tx.send(1); - assert_eq!(rx.recv(), 1); + tx.send(1).unwrap(); + assert_eq!(rx.recv().unwrap(), 1); } #[test] fn smoke_threads() { let (tx, rx) = sync_channel::<int>(0); let _t = Thread::spawn(move|| { - tx.send(1); + tx.send(1).unwrap(); }); - assert_eq!(rx.recv(), 1); + assert_eq!(rx.recv().unwrap(), 1); } #[test] - #[should_fail] fn smoke_port_gone() { let (tx, rx) = sync_channel::<int>(0); drop(rx); - tx.send(1); + assert!(tx.send(1).is_err()); } #[test] - #[should_fail] fn smoke_shared_port_gone2() { let (tx, rx) = sync_channel::<int>(0); drop(rx); let tx2 = tx.clone(); drop(tx); - tx2.send(1); + assert!(tx2.send(1).is_err()); } #[test] - #[should_fail] fn port_gone_concurrent() { let (tx, rx) = sync_channel::<int>(0); - Thread::spawn(move|| { - rx.recv(); - }).detach(); - loop { tx.send(1) } + let _t = Thread::spawn(move|| { + rx.recv().unwrap(); + }); + while tx.send(1).is_ok() {} } #[test] - #[should_fail] fn port_gone_concurrent_shared() { let (tx, rx) = sync_channel::<int>(0); let tx2 = tx.clone(); - Thread::spawn(move|| { - rx.recv(); - }).detach(); - loop { - tx.send(1); - tx2.send(1); - } + let _t = Thread::spawn(move|| { + rx.recv().unwrap(); + }); + while tx.send(1).is_ok() && tx2.send(1).is_ok() {} } #[test] - #[should_fail] fn smoke_chan_gone() { let (tx, rx) = sync_channel::<int>(0); drop(tx); - rx.recv(); + assert!(rx.recv().is_err()); } #[test] - #[should_fail] fn smoke_chan_gone_shared() { let (tx, rx) = sync_channel::<()>(0); let tx2 = tx.clone(); drop(tx); drop(tx2); - rx.recv(); + assert!(rx.recv().is_err()); } #[test] - #[should_fail] fn chan_gone_concurrent() { let (tx, rx) = sync_channel::<int>(0); Thread::spawn(move|| { - tx.send(1); - tx.send(1); + tx.send(1).unwrap(); + tx.send(1).unwrap(); }).detach(); - loop { rx.recv(); } + while rx.recv().is_ok() {} } #[test] fn stress() { let (tx, rx) = sync_channel::<int>(0); Thread::spawn(move|| { - for _ in range(0u, 10000) { tx.send(1); } + for _ in range(0u, 10000) { tx.send(1).unwrap(); } }).detach(); for _ in range(0u, 10000) { - assert_eq!(rx.recv(), 1); + assert_eq!(rx.recv().unwrap(), 1); } } @@ -1713,23 +1660,23 @@ mod sync_tests { Thread::spawn(move|| { for _ in range(0, AMT * NTHREADS) { - assert_eq!(rx.recv(), 1); + assert_eq!(rx.recv().unwrap(), 1); } match rx.try_recv() { Ok(..) => panic!(), _ => {} } - dtx.send(()); + dtx.send(()).unwrap(); }).detach(); for _ in range(0, NTHREADS) { let tx = tx.clone(); Thread::spawn(move|| { - for _ in range(0, AMT) { tx.send(1); } + for _ in range(0, AMT) { tx.send(1).unwrap(); } }).detach(); } drop(tx); - drx.recv(); + drx.recv().unwrap(); } #[test] @@ -1747,12 +1694,11 @@ mod sync_tests { } #[test] - #[should_fail] fn oneshot_single_thread_send_port_close() { // Testing that the sender cleans up the payload if receiver is closed let (tx, rx) = sync_channel::<Box<int>>(0); drop(rx); - tx.send(box 0); + assert!(tx.send(box 0).is_err()); } #[test] @@ -1761,7 +1707,7 @@ mod sync_tests { let res = Thread::spawn(move|| { let (tx, rx) = sync_channel::<int>(0); drop(tx); - rx.recv(); + rx.recv().unwrap(); }).join(); // What is our res? assert!(res.is_err()); @@ -1770,49 +1716,49 @@ mod sync_tests { #[test] fn oneshot_single_thread_send_then_recv() { let (tx, rx) = sync_channel::<Box<int>>(1); - tx.send(box 10); - assert!(rx.recv() == box 10); + tx.send(box 10).unwrap(); + assert!(rx.recv().unwrap() == box 10); } #[test] fn oneshot_single_thread_try_send_open() { let (tx, rx) = sync_channel::<int>(1); assert_eq!(tx.try_send(10), Ok(())); - assert!(rx.recv() == 10); + assert!(rx.recv().unwrap() == 10); } #[test] fn oneshot_single_thread_try_send_closed() { let (tx, rx) = sync_channel::<int>(0); drop(rx); - assert_eq!(tx.try_send(10), Err(RecvDisconnected(10))); + assert_eq!(tx.try_send(10), Err(TrySendError::Disconnected(10))); } #[test] fn oneshot_single_thread_try_send_closed2() { let (tx, _rx) = sync_channel::<int>(0); - assert_eq!(tx.try_send(10), Err(Full(10))); + assert_eq!(tx.try_send(10), Err(TrySendError::Full(10))); } #[test] fn oneshot_single_thread_try_recv_open() { let (tx, rx) = sync_channel::<int>(1); - tx.send(10); - assert!(rx.recv_opt() == Ok(10)); + tx.send(10).unwrap(); + assert!(rx.recv() == Ok(10)); } #[test] fn oneshot_single_thread_try_recv_closed() { let (tx, rx) = sync_channel::<int>(0); drop(tx); - assert!(rx.recv_opt() == Err(())); + assert!(rx.recv().is_err()); } #[test] fn oneshot_single_thread_peek_data() { let (tx, rx) = sync_channel::<int>(1); - assert_eq!(rx.try_recv(), Err(Empty)); - tx.send(10); + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); + tx.send(10).unwrap(); assert_eq!(rx.try_recv(), Ok(10)); } @@ -1820,24 +1766,24 @@ mod sync_tests { fn oneshot_single_thread_peek_close() { let (tx, rx) = sync_channel::<int>(0); drop(tx); - assert_eq!(rx.try_recv(), Err(Disconnected)); - assert_eq!(rx.try_recv(), Err(Disconnected)); + assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); + assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); } #[test] fn oneshot_single_thread_peek_open() { let (_tx, rx) = sync_channel::<int>(0); - assert_eq!(rx.try_recv(), Err(Empty)); + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); } #[test] fn oneshot_multi_task_recv_then_send() { let (tx, rx) = sync_channel::<Box<int>>(0); let _t = Thread::spawn(move|| { - assert!(rx.recv() == box 10); + assert!(rx.recv().unwrap() == box 10); }); - tx.send(box 10); + tx.send(box 10).unwrap(); } #[test] @@ -1847,7 +1793,7 @@ mod sync_tests { drop(tx); }); let res = Thread::spawn(move|| { - assert!(rx.recv() == box 10); + assert!(rx.recv().unwrap() == box 10); }).join(); assert!(res.is_err()); } @@ -1871,7 +1817,7 @@ mod sync_tests { drop(rx); }); let _ = Thread::spawn(move || { - tx.send(1); + tx.send(1).unwrap(); }).join(); } } @@ -1882,7 +1828,7 @@ mod sync_tests { let (tx, rx) = sync_channel::<int>(0); let _t = Thread::spawn(move|| { let res = Thread::spawn(move|| { - rx.recv(); + rx.recv().unwrap(); }).join(); assert!(res.is_err()); }); @@ -1899,9 +1845,9 @@ mod sync_tests { for _ in range(0, stress_factor()) { let (tx, rx) = sync_channel::<Box<int>>(0); let _t = Thread::spawn(move|| { - tx.send(box 10i); + tx.send(box 10i).unwrap(); }); - assert!(rx.recv() == box 10i); + assert!(rx.recv().unwrap() == box 10i); } } @@ -1917,7 +1863,7 @@ mod sync_tests { if i == 10 { return } Thread::spawn(move|| { - tx.send(box i); + tx.send(box i).unwrap(); send(tx, i + 1); }).detach(); } @@ -1926,7 +1872,7 @@ mod sync_tests { if i == 10 { return } Thread::spawn(move|| { - assert!(rx.recv() == box i); + assert!(rx.recv().unwrap() == box i); recv(rx, i + 1); }).detach(); } @@ -1937,8 +1883,8 @@ mod sync_tests { fn recv_a_lot() { // Regression test that we don't run out of stack in scheduler context let (tx, rx) = sync_channel(10000); - for _ in range(0u, 10000) { tx.send(()); } - for _ in range(0u, 10000) { rx.recv(); } + for _ in range(0u, 10000) { tx.send(()).unwrap(); } + for _ in range(0u, 10000) { rx.recv().unwrap(); } } #[test] @@ -1948,12 +1894,12 @@ mod sync_tests { for _ in range(0, total) { let tx = tx.clone(); Thread::spawn(move|| { - tx.send(()); + tx.send(()).unwrap(); }).detach(); } for _ in range(0, total) { - rx.recv(); + rx.recv().unwrap(); } } @@ -1967,14 +1913,14 @@ mod sync_tests { for x in rx.iter() { acc += x; } - total_tx.send(acc); + total_tx.send(acc).unwrap(); }); - tx.send(3); - tx.send(1); - tx.send(2); + tx.send(3).unwrap(); + tx.send(1).unwrap(); + tx.send(2).unwrap(); drop(tx); - assert_eq!(total_rx.recv(), 6); + assert_eq!(total_rx.recv().unwrap(), 6); } #[test] @@ -1991,15 +1937,15 @@ mod sync_tests { count += x; } } - count_tx.send(count); + count_tx.send(count).unwrap(); }); - tx.send(2); - tx.send(2); - tx.send(2); + tx.send(2).unwrap(); + tx.send(2).unwrap(); + tx.send(2).unwrap(); let _ = tx.try_send(2); drop(tx); - assert_eq!(count_rx.recv(), 4); + assert_eq!(count_rx.recv().unwrap(), 4); } #[test] @@ -2008,22 +1954,22 @@ mod sync_tests { let (tx2, rx2) = sync_channel::<()>(1); let (tx3, rx3) = sync_channel::<()>(1); let _t = Thread::spawn(move|| { - rx2.recv(); - tx1.send(1); - tx3.send(()); - rx2.recv(); + rx2.recv().unwrap(); + tx1.send(1).unwrap(); + tx3.send(()).unwrap(); + rx2.recv().unwrap(); drop(tx1); - tx3.send(()); + tx3.send(()).unwrap(); }); - assert_eq!(rx1.try_recv(), Err(Empty)); - tx2.send(()); - rx3.recv(); + assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty)); + tx2.send(()).unwrap(); + rx3.recv().unwrap(); assert_eq!(rx1.try_recv(), Ok(1)); - assert_eq!(rx1.try_recv(), Err(Empty)); - tx2.send(()); - rx3.recv(); - assert_eq!(rx1.try_recv(), Err(Disconnected)); + assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty)); + tx2.send(()).unwrap(); + rx3.recv().unwrap(); + assert_eq!(rx1.try_recv(), Err(TryRecvError::Disconnected)); } // This bug used to end up in a livelock inside of the Receiver destructor @@ -2033,9 +1979,9 @@ mod sync_tests { let (tx, rx) = sync_channel::<()>(0); let (tx2, rx2) = sync_channel::<()>(0); let _t = Thread::spawn(move|| { - rx.recv(); // wait on a oneshot + rx.recv().unwrap(); // wait on a oneshot drop(rx); // destroy a shared - tx2.send(()); + tx2.send(()).unwrap(); }); // make sure the other task has gone to sleep for _ in range(0u, 5000) { Thread::yield_now(); } @@ -2043,64 +1989,64 @@ mod sync_tests { // upgrade to a shared chan and send a message let t = tx.clone(); drop(tx); - t.send(()); + t.send(()).unwrap(); // wait for the child task to exit before we exit - rx2.recv(); + rx2.recv().unwrap(); } #[test] - fn send_opt1() { + fn send1() { let (tx, rx) = sync_channel::<int>(0); - let _t = Thread::spawn(move|| { rx.recv(); }); - assert_eq!(tx.send_opt(1), Ok(())); + let _t = Thread::spawn(move|| { rx.recv().unwrap(); }); + assert_eq!(tx.send(1), Ok(())); } #[test] - fn send_opt2() { + fn send2() { let (tx, rx) = sync_channel::<int>(0); let _t = Thread::spawn(move|| { drop(rx); }); - assert_eq!(tx.send_opt(1), Err(1)); + assert!(tx.send(1).is_err()); } #[test] - fn send_opt3() { + fn send3() { let (tx, rx) = sync_channel::<int>(1); - assert_eq!(tx.send_opt(1), Ok(())); - let _t = Thread::spawn(move|| { drop(rx); }); - assert_eq!(tx.send_opt(1), Err(1)); + assert_eq!(tx.send(1), Ok(())); + let _t =Thread::spawn(move|| { drop(rx); }); + assert!(tx.send(1).is_err()); } #[test] - fn send_opt4() { + fn send4() { let (tx, rx) = sync_channel::<int>(0); let tx2 = tx.clone(); let (done, donerx) = channel(); let done2 = done.clone(); let _t = Thread::spawn(move|| { - assert_eq!(tx.send_opt(1), Err(1)); - done.send(()); + assert!(tx.send(1).is_err()); + done.send(()).unwrap(); }); let _t = Thread::spawn(move|| { - assert_eq!(tx2.send_opt(2), Err(2)); - done2.send(()); + assert!(tx2.send(2).is_err()); + done2.send(()).unwrap(); }); drop(rx); - donerx.recv(); - donerx.recv(); + donerx.recv().unwrap(); + donerx.recv().unwrap(); } #[test] fn try_send1() { let (tx, _rx) = sync_channel::<int>(0); - assert_eq!(tx.try_send(1), Err(Full(1))); + assert_eq!(tx.try_send(1), Err(TrySendError::Full(1))); } #[test] fn try_send2() { let (tx, _rx) = sync_channel::<int>(1); assert_eq!(tx.try_send(1), Ok(())); - assert_eq!(tx.try_send(1), Err(Full(1))); + assert_eq!(tx.try_send(1), Err(TrySendError::Full(1))); } #[test] @@ -2108,7 +2054,7 @@ mod sync_tests { let (tx, rx) = sync_channel::<int>(1); assert_eq!(tx.try_send(1), Ok(())); drop(rx); - assert_eq!(tx.try_send(1), Err(RecvDisconnected(1))); + assert_eq!(tx.try_send(1), Err(TrySendError::Disconnected(1))); } #[test] @@ -2118,12 +2064,12 @@ mod sync_tests { let (tx2, rx2) = sync_channel::<()>(3); let _t = Thread::spawn(move|| { - rx1.recv(); + rx1.recv().unwrap(); tx2.try_send(()).unwrap(); }); tx1.try_send(()).unwrap(); - rx2.recv(); + rx2.recv().unwrap(); } for _ in range(0u, 100) { diff --git a/src/libstd/comm/mpsc_queue.rs b/src/libstd/sync/mpsc/mpsc_queue.rs index d1b6d0d697c..8945233dac9 100644 --- a/src/libstd/comm/mpsc_queue.rs +++ b/src/libstd/sync/mpsc/mpsc_queue.rs @@ -155,7 +155,7 @@ impl<T: Send> Drop for Queue<T> { mod tests { use prelude::v1::*; - use comm::channel; + use sync::mpsc::channel; use super::{Queue, Data, Empty, Inconsistent}; use sync::Arc; use thread::Thread; @@ -186,7 +186,7 @@ mod tests { for i in range(0, nmsgs) { q.push(i); } - tx.send(()); + tx.send(()).unwrap(); }).detach(); } @@ -199,7 +199,7 @@ mod tests { } drop(tx); for _ in range(0, nthreads) { - rx.recv(); + rx.recv().unwrap(); } } } diff --git a/src/libstd/comm/oneshot.rs b/src/libstd/sync/mpsc/oneshot.rs index 9c5a6518845..5f599752a46 100644 --- a/src/libstd/comm/oneshot.rs +++ b/src/libstd/sync/mpsc/oneshot.rs @@ -39,8 +39,8 @@ use self::MyUpgrade::*; use core::prelude::*; -use comm::Receiver; -use comm::blocking::{mod, SignalToken}; +use sync::mpsc::Receiver; +use sync::mpsc::blocking::{mod, SignalToken}; use core::mem; use sync::atomic; diff --git a/src/libstd/comm/select.rs b/src/libstd/sync/mpsc/select.rs index 5c476775bdb..fc1e0b34977 100644 --- a/src/libstd/comm/select.rs +++ b/src/libstd/sync/mpsc/select.rs @@ -27,20 +27,20 @@ //! # Example //! //! ```rust -//! use std::comm::channel; +//! use std::sync::mpsc::channel; //! //! let (tx1, rx1) = channel(); //! let (tx2, rx2) = channel(); //! -//! tx1.send(1i); -//! tx2.send(2i); +//! tx1.send(1i).unwrap(); +//! tx2.send(2i).unwrap(); //! //! select! { //! val = rx1.recv() => { -//! assert_eq!(val, 1i); +//! assert_eq!(val.unwrap(), 1i); //! }, //! val = rx2.recv() => { -//! assert_eq!(val, 2i); +//! assert_eq!(val.unwrap(), 2i); //! } //! } //! ``` @@ -61,8 +61,8 @@ use core::kinds::marker; use core::mem; use core::uint; -use comm::Receiver; -use comm::blocking::{mod, SignalToken}; +use sync::mpsc::{Receiver, RecvError}; +use sync::mpsc::blocking::{mod, SignalToken}; /// The "receiver set" of the select interface. This structure is used to manage /// a set of receivers which are being selected over. @@ -247,13 +247,10 @@ impl<'rx, T: Send> Handle<'rx, T> { #[inline] pub fn id(&self) -> uint { self.id } - /// Receive a value on the underlying receiver. Has the same semantics as - /// `Receiver.recv` - pub fn recv(&mut self) -> T { self.rx.recv() } /// Block to receive a value on the underlying receiver, returning `Some` on /// success or `None` if the channel disconnects. This function has the same - /// semantics as `Receiver.recv_opt` - pub fn recv_opt(&mut self) -> Result<T, ()> { self.rx.recv_opt() } + /// semantics as `Receiver.recv` + pub fn recv(&mut self) -> Result<T, RecvError> { self.rx.recv() } /// Adds this handle to the receiver set that the handle was created from. This /// method can be called multiple times, but it has no effect if `add` was @@ -339,9 +336,9 @@ impl Iterator<*mut Handle<'static, ()>> for Packets { mod test { use prelude::v1::*; - use super::*; - use comm::*; use thread::Thread; + use super::*; + use sync::mpsc::*; // Don't use the libstd version so we can pull in the right Select structure // (std::comm points at the wrong one) @@ -349,7 +346,6 @@ mod test { ( $($name:pat = $rx:ident.$meth:ident() => $code:expr),+ ) => ({ - use comm::Select; let sel = Select::new(); $( let mut $rx = sel.handle(&$rx); )+ unsafe { @@ -365,24 +361,24 @@ mod test { fn smoke() { let (tx1, rx1) = channel::<int>(); let (tx2, rx2) = channel::<int>(); - tx1.send(1); + tx1.send(1).unwrap(); select! { - foo = rx1.recv() => { assert_eq!(foo, 1); }, + foo = rx1.recv() => { assert_eq!(foo.unwrap(), 1); }, _bar = rx2.recv() => { panic!() } } - tx2.send(2); + tx2.send(2).unwrap(); select! { _foo = rx1.recv() => { panic!() }, - bar = rx2.recv() => { assert_eq!(bar, 2) } + bar = rx2.recv() => { assert_eq!(bar.unwrap(), 2) } } drop(tx1); select! { - foo = rx1.recv_opt() => { assert_eq!(foo, Err(())); }, + foo = rx1.recv() => { assert!(foo.is_err()); }, _bar = rx2.recv() => { panic!() } } drop(tx2); select! { - bar = rx2.recv_opt() => { assert_eq!(bar, Err(())); } + bar = rx2.recv() => { assert!(bar.is_err()); } } } @@ -393,13 +389,13 @@ mod test { let (_tx3, rx3) = channel::<int>(); let (_tx4, rx4) = channel::<int>(); let (tx5, rx5) = channel::<int>(); - tx5.send(4); + tx5.send(4).unwrap(); select! { _foo = rx1.recv() => { panic!("1") }, _foo = rx2.recv() => { panic!("2") }, _foo = rx3.recv() => { panic!("3") }, _foo = rx4.recv() => { panic!("4") }, - foo = rx5.recv() => { assert_eq!(foo, 4); } + foo = rx5.recv() => { assert_eq!(foo.unwrap(), 4); } } } @@ -410,8 +406,8 @@ mod test { drop(tx2); select! { - _a1 = rx1.recv_opt() => { panic!() }, - a2 = rx2.recv_opt() => { assert_eq!(a2, Err(())); } + _a1 = rx1.recv() => { panic!() }, + a2 = rx2.recv() => { assert!(a2.is_err()); } } } @@ -423,18 +419,18 @@ mod test { let _t = Thread::spawn(move|| { for _ in range(0u, 20) { Thread::yield_now(); } - tx1.send(1); - rx3.recv(); + tx1.send(1).unwrap(); + rx3.recv().unwrap(); for _ in range(0u, 20) { Thread::yield_now(); } }); select! { - a = rx1.recv() => { assert_eq!(a, 1); }, + a = rx1.recv() => { assert_eq!(a.unwrap(), 1); }, _b = rx2.recv() => { panic!() } } - tx3.send(1); + tx3.send(1).unwrap(); select! { - a = rx1.recv_opt() => { assert_eq!(a, Err(())); }, + a = rx1.recv() => { assert!(a.is_err()) }, _b = rx2.recv() => { panic!() } } } @@ -447,22 +443,22 @@ mod test { let _t = Thread::spawn(move|| { for _ in range(0u, 20) { Thread::yield_now(); } - tx1.send(1); - tx2.send(2); - rx3.recv(); + tx1.send(1).unwrap(); + tx2.send(2).unwrap(); + rx3.recv().unwrap(); }); select! { - a = rx1.recv() => { assert_eq!(a, 1); }, - a = rx2.recv() => { assert_eq!(a, 2); } + a = rx1.recv() => { assert_eq!(a.unwrap(), 1); }, + a = rx2.recv() => { assert_eq!(a.unwrap(), 2); } } select! { - a = rx1.recv() => { assert_eq!(a, 1); }, - a = rx2.recv() => { assert_eq!(a, 2); } + a = rx1.recv() => { assert_eq!(a.unwrap(), 1); }, + a = rx2.recv() => { assert_eq!(a.unwrap(), 2); } } - assert_eq!(rx1.try_recv(), Err(Empty)); - assert_eq!(rx2.try_recv(), Err(Empty)); - tx3.send(()); + assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty)); + assert_eq!(rx2.try_recv(), Err(TryRecvError::Empty)); + tx3.send(()).unwrap(); } #[test] @@ -475,20 +471,20 @@ mod test { let _t = Thread::spawn(move|| { for i in range(0, AMT) { if i % 2 == 0 { - tx1.send(i); + tx1.send(i).unwrap(); } else { - tx2.send(i); + tx2.send(i).unwrap(); } - rx3.recv(); + rx3.recv().unwrap(); } }); for i in range(0, AMT) { select! { - i1 = rx1.recv() => { assert!(i % 2 == 0 && i == i1); }, - i2 = rx2.recv() => { assert!(i % 2 == 1 && i == i2); } + i1 = rx1.recv() => { assert!(i % 2 == 0 && i == i1.unwrap()); }, + i2 = rx2.recv() => { assert!(i % 2 == 1 && i == i2.unwrap()); } } - tx3.send(()); + tx3.send(()).unwrap(); } } @@ -499,19 +495,19 @@ mod test { let (tx3, rx3) = channel::<()>(); let _t = Thread::spawn(move|| { - rx3.recv(); + rx3.recv().unwrap(); tx1.clone(); - assert_eq!(rx3.try_recv(), Err(Empty)); - tx1.send(2); - rx3.recv(); + assert_eq!(rx3.try_recv(), Err(TryRecvError::Empty)); + tx1.send(2).unwrap(); + rx3.recv().unwrap(); }); - tx3.send(()); + tx3.send(()).unwrap(); select! { _i1 = rx1.recv() => {}, _i2 = rx2.recv() => panic!() } - tx3.send(()); + tx3.send(()).unwrap(); } #[test] @@ -521,19 +517,19 @@ mod test { let (tx3, rx3) = channel::<()>(); let _t = Thread::spawn(move|| { - rx3.recv(); + rx3.recv().unwrap(); tx1.clone(); - assert_eq!(rx3.try_recv(), Err(Empty)); - tx1.send(2); - rx3.recv(); + assert_eq!(rx3.try_recv(), Err(TryRecvError::Empty)); + tx1.send(2).unwrap(); + rx3.recv().unwrap(); }); - tx3.send(()); + tx3.send(()).unwrap(); select! { _i1 = rx1.recv() => {}, _i2 = rx2.recv() => panic!() } - tx3.send(()); + tx3.send(()).unwrap(); } #[test] @@ -548,31 +544,31 @@ mod test { unsafe { h2.add(); } unsafe { h1.add(); } assert_eq!(s.wait(), h2.id); - tx3.send(()); + tx3.send(()).unwrap(); }); for _ in range(0u, 1000) { Thread::yield_now(); } drop(tx1.clone()); - tx2.send(()); - rx3.recv(); + tx2.send(()).unwrap(); + rx3.recv().unwrap(); } #[test] fn preflight1() { let (tx, rx) = channel(); - tx.send(()); + tx.send(()).unwrap(); select! { - () = rx.recv() => {} + _n = rx.recv() => {} } } #[test] fn preflight2() { let (tx, rx) = channel(); - tx.send(()); - tx.send(()); + tx.send(()).unwrap(); + tx.send(()).unwrap(); select! { - () = rx.recv() => {} + _n = rx.recv() => {} } } @@ -580,16 +576,16 @@ mod test { fn preflight3() { let (tx, rx) = channel(); drop(tx.clone()); - tx.send(()); + tx.send(()).unwrap(); select! { - () = rx.recv() => {} + _n = rx.recv() => {} } } #[test] fn preflight4() { let (tx, rx) = channel(); - tx.send(()); + tx.send(()).unwrap(); let s = Select::new(); let mut h = s.handle(&rx); unsafe { h.add(); } @@ -599,8 +595,8 @@ mod test { #[test] fn preflight5() { let (tx, rx) = channel(); - tx.send(()); - tx.send(()); + tx.send(()).unwrap(); + tx.send(()).unwrap(); let s = Select::new(); let mut h = s.handle(&rx); unsafe { h.add(); } @@ -611,7 +607,7 @@ mod test { fn preflight6() { let (tx, rx) = channel(); drop(tx.clone()); - tx.send(()); + tx.send(()).unwrap(); let s = Select::new(); let mut h = s.handle(&rx); unsafe { h.add(); } @@ -631,9 +627,9 @@ mod test { #[test] fn preflight8() { let (tx, rx) = channel(); - tx.send(()); + tx.send(()).unwrap(); drop(tx); - rx.recv(); + rx.recv().unwrap(); let s = Select::new(); let mut h = s.handle(&rx); unsafe { h.add(); } @@ -644,9 +640,9 @@ mod test { fn preflight9() { let (tx, rx) = channel(); drop(tx.clone()); - tx.send(()); + tx.send(()).unwrap(); drop(tx); - rx.recv(); + rx.recv().unwrap(); let s = Select::new(); let mut h = s.handle(&rx); unsafe { h.add(); } @@ -659,34 +655,34 @@ mod test { let (tx2, rx2) = channel(); let _t = Thread::spawn(move|| { select! { - () = rx1.recv() => {} + _n = rx1.recv() => {} } - tx2.send(()); + tx2.send(()).unwrap(); }); for _ in range(0u, 100) { Thread::yield_now() } - tx1.send(()); - rx2.recv(); + tx1.send(()).unwrap(); + rx2.recv().unwrap(); } #[test] fn stream_data_waiting() { let (tx1, rx1) = channel(); let (tx2, rx2) = channel(); - tx1.send(()); - tx1.send(()); - rx1.recv(); - rx1.recv(); + tx1.send(()).unwrap(); + tx1.send(()).unwrap(); + rx1.recv().unwrap(); + rx1.recv().unwrap(); let _t = Thread::spawn(move|| { select! { - () = rx1.recv() => {} + _n = rx1.recv() => {} } - tx2.send(()); + tx2.send(()).unwrap(); }); for _ in range(0u, 100) { Thread::yield_now() } - tx1.send(()); - rx2.recv(); + tx1.send(()).unwrap(); + rx2.recv().unwrap(); } #[test] @@ -694,26 +690,26 @@ mod test { let (tx1, rx1) = channel(); let (tx2, rx2) = channel(); drop(tx1.clone()); - tx1.send(()); - rx1.recv(); + tx1.send(()).unwrap(); + rx1.recv().unwrap(); let _t = Thread::spawn(move|| { select! { - () = rx1.recv() => {} + _n = rx1.recv() => {} } - tx2.send(()); + tx2.send(()).unwrap(); }); for _ in range(0u, 100) { Thread::yield_now() } - tx1.send(()); - rx2.recv(); + tx1.send(()).unwrap(); + rx2.recv().unwrap(); } #[test] fn sync1() { let (tx, rx) = sync_channel::<int>(1); - tx.send(1); + tx.send(1).unwrap(); select! { - n = rx.recv() => { assert_eq!(n, 1); } + n = rx.recv() => { assert_eq!(n.unwrap(), 1); } } } @@ -722,10 +718,10 @@ mod test { let (tx, rx) = sync_channel::<int>(0); let _t = Thread::spawn(move|| { for _ in range(0u, 100) { Thread::yield_now() } - tx.send(1); + tx.send(1).unwrap(); }); select! { - n = rx.recv() => { assert_eq!(n, 1); } + n = rx.recv() => { assert_eq!(n.unwrap(), 1); } } } @@ -733,16 +729,18 @@ mod test { fn sync3() { let (tx1, rx1) = sync_channel::<int>(0); let (tx2, rx2): (Sender<int>, Receiver<int>) = channel(); - let _t = Thread::spawn(move|| { tx1.send(1); }); - let _t = Thread::spawn(move|| { tx2.send(2); }); + let _t = Thread::spawn(move|| { tx1.send(1).unwrap(); }); + let _t = Thread::spawn(move|| { tx2.send(2).unwrap(); }); select! { n = rx1.recv() => { + let n = n.unwrap(); assert_eq!(n, 1); - assert_eq!(rx2.recv(), 2); + assert_eq!(rx2.recv().unwrap(), 2); }, n = rx2.recv() => { + let n = n.unwrap(); assert_eq!(n, 2); - assert_eq!(rx1.recv(), 1); + assert_eq!(rx1.recv().unwrap(), 1); } } } diff --git a/src/libstd/comm/shared.rs b/src/libstd/sync/mpsc/shared.rs index 1022694e634..e1606cb4317 100644 --- a/src/libstd/comm/shared.rs +++ b/src/libstd/sync/mpsc/shared.rs @@ -26,10 +26,10 @@ use core::cmp; use core::int; use sync::{atomic, Mutex, MutexGuard}; -use comm::mpsc_queue as mpsc; -use comm::blocking::{mod, SignalToken}; -use comm::select::StartResult; -use comm::select::StartResult::*; +use sync::mpsc::mpsc_queue as mpsc; +use sync::mpsc::blocking::{mod, SignalToken}; +use sync::mpsc::select::StartResult; +use sync::mpsc::select::StartResult::*; use thread::Thread; const DISCONNECTED: int = int::MIN; diff --git a/src/libstd/comm/spsc_queue.rs b/src/libstd/sync/mpsc/spsc_queue.rs index 1e2f5222d8b..15624601157 100644 --- a/src/libstd/comm/spsc_queue.rs +++ b/src/libstd/sync/mpsc/spsc_queue.rs @@ -245,7 +245,7 @@ mod test { use sync::Arc; use super::Queue; use thread::Thread; - use comm::channel; + use sync::mpsc::channel; #[test] fn smoke() { @@ -332,12 +332,12 @@ mod test { } } } - tx.send(()); + tx.send(()).unwrap(); }); for _ in range(0i, 100000) { q.push(1); } - rx.recv(); + rx.recv().unwrap(); } } } diff --git a/src/libstd/comm/stream.rs b/src/libstd/sync/mpsc/stream.rs index b68f626060e..01b799283ee 100644 --- a/src/libstd/comm/stream.rs +++ b/src/libstd/sync/mpsc/stream.rs @@ -28,10 +28,10 @@ use core::cmp; use core::int; use thread::Thread; +use sync::mpsc::blocking::{mod, SignalToken}; +use sync::mpsc::spsc_queue as spsc; +use sync::mpsc::Receiver; use sync::atomic; -use comm::spsc_queue as spsc; -use comm::Receiver; -use comm::blocking::{mod, SignalToken}; const DISCONNECTED: int = int::MIN; #[cfg(test)] diff --git a/src/libstd/comm/sync.rs b/src/libstd/sync/mpsc/sync.rs index 88338849965..28005831d4f 100644 --- a/src/libstd/comm/sync.rs +++ b/src/libstd/sync/mpsc/sync.rs @@ -42,8 +42,8 @@ use vec::Vec; use core::mem; use sync::{atomic, Mutex, MutexGuard}; -use comm::blocking::{mod, WaitToken, SignalToken}; -use comm::select::StartResult::{mod, Installed, Abort}; +use sync::mpsc::blocking::{mod, WaitToken, SignalToken}; +use sync::mpsc::select::StartResult::{mod, Installed, Abort}; pub struct Packet<T> { /// Only field outside of the mutex. Just done for kicks, but mainly because @@ -204,14 +204,14 @@ impl<T: Send> Packet<T> { pub fn try_send(&self, t: T) -> Result<(), super::TrySendError<T>> { let mut guard = self.lock.lock(); if guard.disconnected { - Err(super::RecvDisconnected(t)) + Err(super::TrySendError::Disconnected(t)) } else if guard.buf.size() == guard.buf.cap() { - Err(super::Full(t)) + Err(super::TrySendError::Full(t)) } else if guard.cap == 0 { // With capacity 0, even though we have buffer space we can't // transfer the data unless there's a receiver waiting. match mem::replace(&mut guard.blocker, NoneBlocked) { - NoneBlocked => Err(super::Full(t)), + NoneBlocked => Err(super::TrySendError::Full(t)), BlockedSender(..) => unreachable!(), BlockedReceiver(token) => { guard.buf.enqueue(t); diff --git a/src/libstd/sync/mutex.rs b/src/libstd/sync/mutex.rs index 87a02bd4ef5..1562031499f 100644 --- a/src/libstd/sync/mutex.rs +++ b/src/libstd/sync/mutex.rs @@ -37,7 +37,7 @@ use sys_common::mutex as sys; /// ```rust /// use std::sync::{Arc, Mutex}; /// use std::thread::Thread; -/// use std::comm::channel; +/// use std::sync::mpsc::channel; /// /// const N: uint = 10; /// @@ -58,13 +58,13 @@ use sys_common::mutex as sys; /// let mut data = data.lock(); /// *data += 1; /// if *data == N { -/// tx.send(()); +/// tx.send(()).unwrap(); /// } /// // the lock is unlocked here when `data` goes out of scope. /// }).detach(); /// } /// -/// rx.recv(); +/// rx.recv().unwrap(); /// ``` pub struct Mutex<T> { // Note that this static mutex is in a *box*, not inlined into the struct @@ -284,7 +284,7 @@ impl Drop for StaticMutexGuard { mod test { use prelude::v1::*; - use comm::channel; + use sync::mpsc::channel; use sync::{Arc, Mutex, StaticMutex, MUTEX_INIT, Condvar}; use thread::Thread; @@ -329,14 +329,14 @@ mod test { let (tx, rx) = channel(); for _ in range(0, K) { let tx2 = tx.clone(); - Thread::spawn(move|| { inc(); tx2.send(()); }).detach(); + Thread::spawn(move|| { inc(); tx2.send(()).unwrap(); }).detach(); let tx2 = tx.clone(); - Thread::spawn(move|| { inc(); tx2.send(()); }).detach(); + Thread::spawn(move|| { inc(); tx2.send(()).unwrap(); }).detach(); } drop(tx); for _ in range(0, 2 * K) { - rx.recv(); + rx.recv().unwrap(); } assert_eq!(unsafe {CNT}, J * K * 2); unsafe { @@ -357,7 +357,7 @@ mod test { let (tx, rx) = channel(); let _t = Thread::spawn(move|| { // wait until parent gets in - rx.recv(); + rx.recv().unwrap(); let &(ref lock, ref cvar) = &*packet2.0; let mut lock = lock.lock(); *lock = true; @@ -366,7 +366,7 @@ mod test { let &(ref lock, ref cvar) = &*packet.0; let lock = lock.lock(); - tx.send(()); + tx.send(()).unwrap(); assert!(!*lock); while !*lock { cvar.wait(&lock); @@ -381,7 +381,7 @@ mod test { let (tx, rx) = channel(); let _t = Thread::spawn(move || -> () { - rx.recv(); + rx.recv().unwrap(); let &(ref lock, ref cvar) = &*packet2.0; let _g = lock.lock(); cvar.notify_one(); @@ -391,7 +391,7 @@ mod test { let &(ref lock, ref cvar) = &*packet.0; let lock = lock.lock(); - tx.send(()); + tx.send(()).unwrap(); while *lock == 1 { cvar.wait(&lock); } @@ -421,9 +421,9 @@ mod test { let lock = arc2.lock(); let lock2 = lock.lock(); assert_eq!(*lock2, 1); - tx.send(()); + tx.send(()).unwrap(); }); - rx.recv(); + rx.recv().unwrap(); } #[test] diff --git a/src/libstd/sync/once.rs b/src/libstd/sync/once.rs index fe25eca03d7..17b7b70c301 100644 --- a/src/libstd/sync/once.rs +++ b/src/libstd/sync/once.rs @@ -126,7 +126,7 @@ mod test { use thread::Thread; use super::{ONCE_INIT, Once}; - use comm::channel; + use sync::mpsc::channel; #[test] fn smoke_once() { @@ -155,7 +155,7 @@ mod test { }); assert!(run); } - tx.send(()); + tx.send(()).unwrap(); }).detach(); } @@ -168,7 +168,7 @@ mod test { } for _ in range(0u, 10) { - rx.recv(); + rx.recv().unwrap(); } } } diff --git a/src/libstd/sync/rwlock.rs b/src/libstd/sync/rwlock.rs index b316cd908b6..3c4283c72e2 100644 --- a/src/libstd/sync/rwlock.rs +++ b/src/libstd/sync/rwlock.rs @@ -363,7 +363,7 @@ mod tests { use prelude::v1::*; use rand::{mod, Rng}; - use comm::channel; + use sync::mpsc::channel; use thread::Thread; use sync::{Arc, RWLock, StaticRWLock, RWLOCK_INIT}; @@ -408,7 +408,7 @@ mod tests { }).detach(); } drop(tx); - let _ = rx.recv_opt(); + let _ = rx.recv(); unsafe { R.destroy(); } } @@ -475,7 +475,7 @@ mod tests { Thread::yield_now(); *lock = tmp + 1; } - tx.send(()); + tx.send(()).unwrap(); }).detach(); // Readers try to catch the writer in the act @@ -494,7 +494,7 @@ mod tests { } // Wait for writer to finish - rx.recv(); + rx.recv().unwrap(); let lock = arc.read(); assert_eq!(*lock, 10); } diff --git a/src/libstd/sync/semaphore.rs b/src/libstd/sync/semaphore.rs index 382caa2cf4a..784b173b99e 100644 --- a/src/libstd/sync/semaphore.rs +++ b/src/libstd/sync/semaphore.rs @@ -108,7 +108,7 @@ mod tests { use sync::Arc; use super::Semaphore; - use comm::channel; + use sync::mpsc::channel; use thread::Thread; #[test] @@ -143,7 +143,7 @@ mod tests { let s2 = s.clone(); let _t = Thread::spawn(move|| { s2.acquire(); - tx.send(()); + tx.send(()).unwrap(); }); s.release(); let _ = rx.recv(); @@ -157,7 +157,7 @@ mod tests { let _ = rx.recv(); }); s.acquire(); - tx.send(()); + tx.send(()).unwrap(); } #[test] @@ -171,11 +171,11 @@ mod tests { let _t = Thread::spawn(move|| { let _g = s2.access(); let _ = rx2.recv(); - tx1.send(()); + tx1.send(()).unwrap(); }); let _g = s.access(); - tx2.send(()); - let _ = rx1.recv(); + tx2.send(()).unwrap(); + rx1.recv().unwrap(); } #[test] @@ -186,12 +186,12 @@ mod tests { { let _g = s.access(); Thread::spawn(move|| { - tx.send(()); + tx.send(()).unwrap(); drop(s2.access()); - tx.send(()); + tx.send(()).unwrap(); }).detach(); - rx.recv(); // wait for child to come alive + rx.recv().unwrap(); // wait for child to come alive } - rx.recv(); // wait for child to be done + rx.recv().unwrap(); // wait for child to be done } } diff --git a/src/libstd/sync/task_pool.rs b/src/libstd/sync/task_pool.rs index ce7e883e803..b0325998358 100644 --- a/src/libstd/sync/task_pool.rs +++ b/src/libstd/sync/task_pool.rs @@ -12,9 +12,9 @@ use core::prelude::*; -use thread::Thread; -use comm::{channel, Sender, Receiver}; use sync::{Arc, Mutex}; +use sync::mpsc::{channel, Sender, Receiver}; +use thread::Thread; use thunk::Thunk; struct Sentinel<'a> { @@ -55,7 +55,7 @@ impl<'a> Drop for Sentinel<'a> { /// ```rust /// use std::sync::TaskPool; /// use std::iter::AdditiveIterator; -/// use std::comm::channel; +/// use std::sync::mpsc::channel; /// /// let pool = TaskPool::new(4u); /// @@ -63,7 +63,7 @@ impl<'a> Drop for Sentinel<'a> { /// for _ in range(0, 8u) { /// let tx = tx.clone(); /// pool.execute(move|| { -/// tx.send(1u); +/// tx.send(1u).unwrap(); /// }); /// } /// @@ -101,7 +101,7 @@ impl TaskPool { pub fn execute<F>(&self, job: F) where F : FnOnce(), F : Send { - self.jobs.send(Thunk::new(job)); + self.jobs.send(Thunk::new(job)).unwrap(); } } @@ -115,7 +115,7 @@ fn spawn_in_pool(jobs: Arc<Mutex<Receiver<Thunk>>>) { // Only lock jobs for the time it takes // to get a job, not run it. let lock = jobs.lock(); - lock.recv_opt() + lock.recv() }; match message { @@ -134,7 +134,7 @@ fn spawn_in_pool(jobs: Arc<Mutex<Receiver<Thunk>>>) { mod test { use prelude::v1::*; use super::*; - use comm::channel; + use sync::mpsc::channel; const TEST_TASKS: uint = 4u; @@ -148,7 +148,7 @@ mod test { for _ in range(0, TEST_TASKS) { let tx = tx.clone(); pool.execute(move|| { - tx.send(1u); + tx.send(1u).unwrap(); }); } @@ -177,7 +177,7 @@ mod test { for _ in range(0, TEST_TASKS) { let tx = tx.clone(); pool.execute(move|| { - tx.send(1u); + tx.send(1u).unwrap(); }); } diff --git a/src/libstd/sys/common/helper_thread.rs b/src/libstd/sys/common/helper_thread.rs index 5d94e8efeb8..b9dc5f0e398 100644 --- a/src/libstd/sys/common/helper_thread.rs +++ b/src/libstd/sys/common/helper_thread.rs @@ -23,10 +23,10 @@ use prelude::v1::*; use cell::UnsafeCell; -use comm::{channel, Sender, Receiver}; use mem; use rt; use sync::{StaticMutex, StaticCondvar}; +use sync::mpsc::{channel, Sender, Receiver}; use sys::helper_signal; use thread::Thread; @@ -118,7 +118,7 @@ impl<M: Send> Helper<M> { // message. Otherwise it could wake up and go to sleep before we // send the message. assert!(!self.chan.get().is_null()); - (**self.chan.get()).send(msg); + (**self.chan.get()).send(msg).unwrap(); helper_signal::signal(*self.signal.get() as helper_signal::signal); } } diff --git a/src/libstd/sys/unix/process.rs b/src/libstd/sys/unix/process.rs index 9824c9c7364..a11fe3487a8 100644 --- a/src/libstd/sys/unix/process.rs +++ b/src/libstd/sys/unix/process.rs @@ -13,7 +13,6 @@ use self::Req::*; use c_str::{CString, ToCStr}; use collections; -use comm::{channel, Sender, Receiver}; use hash::Hash; use io::process::{ProcessExit, ExitStatus, ExitSignal}; use io::{mod, IoResult, IoError, EndOfFile}; @@ -22,6 +21,7 @@ use mem; use os; use path::BytesContainer; use ptr; +use sync::mpsc::{channel, Sender, Receiver}; use sys::fs::FileDesc; use sys::{mod, retry, c, wouldblock, set_nonblocking, ms_to_timeval}; use sys_common::helper_thread::Helper; @@ -277,8 +277,8 @@ impl Process { } pub fn wait(&self, deadline: u64) -> IoResult<ProcessExit> { - use std::cmp; - use std::comm; + use cmp; + use sync::mpsc::TryRecvError; static mut WRITE_FD: libc::c_int = 0; @@ -337,9 +337,9 @@ impl Process { let (tx, rx) = channel(); unsafe { HELPER.send(NewChild(self.pid, tx, deadline)); } - return match rx.recv_opt() { + return match rx.recv() { Ok(e) => Ok(e), - Err(()) => Err(timeout("wait timed out")), + Err(..) => Err(timeout("wait timed out")), }; // Register a new SIGCHLD handler, returning the reading half of the @@ -420,11 +420,11 @@ impl Process { Ok(NewChild(pid, tx, deadline)) => { active.push((pid, tx, deadline)); } - Err(comm::Disconnected) => { + Err(TryRecvError::Disconnected) => { assert!(active.len() == 0); break 'outer; } - Err(comm::Empty) => break, + Err(TryRecvError::Empty) => break, } } } @@ -460,7 +460,7 @@ impl Process { active.retain(|&(pid, ref tx, _)| { let pr = Process { pid: pid }; match pr.try_wait() { - Some(msg) => { tx.send(msg); false } + Some(msg) => { tx.send(msg).unwrap(); false } None => true, } }); diff --git a/src/libstd/sys/unix/timer.rs b/src/libstd/sys/unix/timer.rs index 903c4954f6f..4afe13d8735 100644 --- a/src/libstd/sys/unix/timer.rs +++ b/src/libstd/sys/unix/timer.rs @@ -49,13 +49,13 @@ use prelude::v1::*; use self::Req::*; -use comm::{mod, channel, Sender, Receiver}; use io::IoResult; use libc; use mem; use os; use ptr; use sync::atomic; +use sync::mpsc::{channel, Sender, Receiver, TryRecvError}; use sys::c; use sys::fs::FileDesc; use sys_common::helper_thread::Helper; @@ -168,7 +168,7 @@ fn helper(input: libc::c_int, messages: Receiver<Req>, _: ()) { 1 => { loop { match messages.try_recv() { - Err(comm::Disconnected) => { + Err(TryRecvError::Disconnected) => { assert!(active.len() == 0); break 'outer; } @@ -179,7 +179,7 @@ fn helper(input: libc::c_int, messages: Receiver<Req>, _: ()) { match dead.iter().position(|&(i, _)| id == i) { Some(i) => { let (_, i) = dead.remove(i).unwrap(); - ack.send(i); + ack.send(i).unwrap(); continue } None => {} @@ -187,7 +187,7 @@ fn helper(input: libc::c_int, messages: Receiver<Req>, _: ()) { let i = active.iter().position(|i| i.id == id); let i = i.expect("no timer found"); let t = active.remove(i).unwrap(); - ack.send(t); + ack.send(t).unwrap(); } Err(..) => break } @@ -271,7 +271,7 @@ impl Timer { None => { let (tx, rx) = channel(); HELPER.send(RemoveTimer(self.id, tx)); - rx.recv() + rx.recv().unwrap() } } } diff --git a/src/libstd/thread.rs b/src/libstd/thread.rs index fdf2648faa1..5c9e6153afb 100644 --- a/src/libstd/thread.rs +++ b/src/libstd/thread.rs @@ -443,7 +443,7 @@ mod test { use any::{Any, AnyRefExt}; use boxed::BoxAny; - use comm::{channel, Sender}; + use sync::mpsc::{channel, Sender}; use result; use std::io::{ChanReader, ChanWriter}; use super::{Thread, Builder}; @@ -470,9 +470,9 @@ mod test { fn test_run_basic() { let (tx, rx) = channel(); Thread::spawn(move|| { - tx.send(()); + tx.send(()).unwrap(); }).detach(); - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -505,7 +505,7 @@ mod test { let tx = tx.clone(); Thread::spawn(move|| { if i == 0 { - tx.send(()); + tx.send(()).unwrap(); } else { f(i - 1, tx); } @@ -513,7 +513,7 @@ mod test { } f(10, tx); - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -522,11 +522,11 @@ mod test { Thread::spawn(move|| { Thread::spawn(move|| { - tx.send(()); + tx.send(()).unwrap(); }).detach(); }).detach(); - rx.recv(); + rx.recv().unwrap(); } fn avoid_copying_the_body<F>(spawnfn: F) where F: FnOnce(Thunk) { @@ -537,10 +537,10 @@ mod test { spawnfn(Thunk::new(move|| { let x_in_child = (&*x) as *const int as uint; - tx.send(x_in_child); + tx.send(x_in_child).unwrap(); })); - let x_in_child = rx.recv(); + let x_in_child = rx.recv().unwrap(); assert_eq!(x_in_parent, x_in_child); } diff --git a/src/libstd/thread_local/mod.rs b/src/libstd/thread_local/mod.rs index 674cb5a9805..6cd26a366ae 100644 --- a/src/libstd/thread_local/mod.rs +++ b/src/libstd/thread_local/mod.rs @@ -471,7 +471,7 @@ mod imp { mod tests { use prelude::v1::*; - use comm::{channel, Sender}; + use sync::mpsc::{channel, Sender}; use cell::UnsafeCell; use thread::Thread; @@ -480,7 +480,7 @@ mod tests { impl Drop for Foo { fn drop(&mut self) { let Foo(ref s) = *self; - s.send(()); + s.send(()).unwrap(); } } @@ -497,9 +497,9 @@ mod tests { FOO.with(|f| unsafe { assert_eq!(*f.get(), 1); }); - tx.send(()); + tx.send(()).unwrap(); }); - rx.recv(); + rx.recv().unwrap(); FOO.with(|f| unsafe { assert_eq!(*f.get(), 2); @@ -519,7 +519,7 @@ mod tests { *f.get() = Some(Foo(tx.take().unwrap())); }); }); - rx.recv(); + rx.recv().unwrap(); } #[test] @@ -610,7 +610,7 @@ mod tests { let mut tx = Some(tx); K1.with(|s| *s.get() = Some(S1(tx.take().unwrap()))); }); - rx.recv(); + rx.recv().unwrap(); } } diff --git a/src/libtest/lib.rs b/src/libtest/lib.rs index fc9826a913c..c70aa41c569 100644 --- a/src/libtest/lib.rs +++ b/src/libtest/lib.rs @@ -59,7 +59,6 @@ use term::color::{Color, RED, YELLOW, GREEN, CYAN}; use std::any::{Any, AnyRefExt}; use std::cmp; use std::collections::BTreeMap; -use std::comm::{channel, Sender}; use std::f64; use std::fmt::Show; use std::fmt; @@ -71,6 +70,7 @@ use std::iter::repeat; use std::num::{Float, FloatMath, Int}; use std::os; use std::str::{FromStr, from_str}; +use std::sync::mpsc::{channel, Sender}; use std::thread::{mod, Thread}; use std::thunk::{Thunk, Invoke}; use std::time::Duration; @@ -1021,7 +1021,7 @@ fn run_tests<F>(opts: &TestOpts, pending += 1; } - let (desc, result, stdout) = rx.recv(); + let (desc, result, stdout) = rx.recv().unwrap(); if concurrency != 1 { try!(callback(TeWait(desc.clone(), PadNone))); } @@ -1034,7 +1034,7 @@ fn run_tests<F>(opts: &TestOpts, for b in filtered_benchs_and_metrics.into_iter() { try!(callback(TeWait(b.desc.clone(), b.testfn.padding()))); run_test(opts, !opts.run_benchmarks, b, tx.clone()); - let (test, result, stdout) = rx.recv(); + let (test, result, stdout) = rx.recv().unwrap(); try!(callback(TeResult(test, result, stdout))); } Ok(()) @@ -1111,7 +1111,7 @@ pub fn run_test(opts: &TestOpts, let TestDescAndFn {desc, testfn} = test; if force_ignore || desc.ignore { - monitor_ch.send((desc, TrIgnored, Vec::new())); + monitor_ch.send((desc, TrIgnored, Vec::new())).unwrap(); return; } @@ -1138,31 +1138,31 @@ pub fn run_test(opts: &TestOpts, let result_guard = cfg.spawn(move || { testfn.invoke(()) }); let stdout = reader.read_to_end().unwrap().into_iter().collect(); let test_result = calc_result(&desc, result_guard.join()); - monitor_ch.send((desc.clone(), test_result, stdout)); + monitor_ch.send((desc.clone(), test_result, stdout)).unwrap(); }).detach(); } match testfn { DynBenchFn(bencher) => { let bs = ::bench::benchmark(|harness| bencher.run(harness)); - monitor_ch.send((desc, TrBench(bs), Vec::new())); + monitor_ch.send((desc, TrBench(bs), Vec::new())).unwrap(); return; } StaticBenchFn(benchfn) => { let bs = ::bench::benchmark(|harness| (benchfn.clone())(harness)); - monitor_ch.send((desc, TrBench(bs), Vec::new())); + monitor_ch.send((desc, TrBench(bs), Vec::new())).unwrap(); return; } DynMetricFn(f) => { let mut mm = MetricMap::new(); f.invoke(&mut mm); - monitor_ch.send((desc, TrMetrics(mm), Vec::new())); + monitor_ch.send((desc, TrMetrics(mm), Vec::new())).unwrap(); return; } StaticMetricFn(f) => { let mut mm = MetricMap::new(); f(&mut mm); - monitor_ch.send((desc, TrMetrics(mm), Vec::new())); + monitor_ch.send((desc, TrMetrics(mm), Vec::new())).unwrap(); return; } DynTestFn(f) => run_test_inner(desc, monitor_ch, opts.nocapture, f), @@ -1466,7 +1466,7 @@ mod tests { StaticTestName, DynTestName, DynTestFn, ShouldFail}; use std::io::TempDir; use std::thunk::Thunk; - use std::comm::channel; + use std::sync::mpsc::channel; #[test] pub fn do_not_run_ignored_tests() { @@ -1481,7 +1481,7 @@ mod tests { }; let (tx, rx) = channel(); run_test(&TestOpts::new(), false, desc, tx); - let (_, res, _) = rx.recv(); + let (_, res, _) = rx.recv().unwrap(); assert!(res != TrOk); } @@ -1498,7 +1498,7 @@ mod tests { }; let (tx, rx) = channel(); run_test(&TestOpts::new(), false, desc, tx); - let (_, res, _) = rx.recv(); + let (_, res, _) = rx.recv().unwrap(); assert!(res == TrIgnored); } @@ -1515,7 +1515,7 @@ mod tests { }; let (tx, rx) = channel(); run_test(&TestOpts::new(), false, desc, tx); - let (_, res, _) = rx.recv(); + let (_, res, _) = rx.recv().unwrap(); assert!(res == TrOk); } @@ -1532,7 +1532,7 @@ mod tests { }; let (tx, rx) = channel(); run_test(&TestOpts::new(), false, desc, tx); - let (_, res, _) = rx.recv(); + let (_, res, _) = rx.recv().unwrap(); assert!(res == TrOk); } @@ -1549,7 +1549,7 @@ mod tests { }; let (tx, rx) = channel(); run_test(&TestOpts::new(), false, desc, tx); - let (_, res, _) = rx.recv(); + let (_, res, _) = rx.recv().unwrap(); assert!(res == TrFailed); } @@ -1566,7 +1566,7 @@ mod tests { }; let (tx, rx) = channel(); run_test(&TestOpts::new(), false, desc, tx); - let (_, res, _) = rx.recv(); + let (_, res, _) = rx.recv().unwrap(); assert!(res == TrFailed); } diff --git a/src/test/auxiliary/cci_capture_clause.rs b/src/test/auxiliary/cci_capture_clause.rs index e3dbe3c7e22..ec470ddc213 100644 --- a/src/test/auxiliary/cci_capture_clause.rs +++ b/src/test/auxiliary/cci_capture_clause.rs @@ -9,7 +9,7 @@ // except according to those terms. use std::task; -use std::comm::{Receiver, channel}; +use std::sync::mpsc::{Receiver, channel}; pub fn foo<T:Send + Clone>(x: T) -> Receiver<T> { let (tx, rx) = channel(); diff --git a/src/test/compile-fail/bind-by-move-no-guards.rs b/src/test/compile-fail/bind-by-move-no-guards.rs index 90d5072f412..bb6060f2543 100644 --- a/src/test/compile-fail/bind-by-move-no-guards.rs +++ b/src/test/compile-fail/bind-by-move-no-guards.rs @@ -8,15 +8,16 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::channel; +use std::sync::mpsc::channel; fn main() { let (tx, rx) = channel(); let x = Some(rx); tx.send(false); match x { - Some(z) if z.recv() => { panic!() }, //~ ERROR cannot bind by-move into a pattern guard - Some(z) => { assert!(!z.recv()); }, + Some(z) if z.recv().unwrap() => { panic!() }, + //~^ ERROR cannot bind by-move into a pattern guard + Some(z) => { assert!(!z.recv().unwrap()); }, None => panic!() } } diff --git a/src/test/compile-fail/builtin-superkinds-self-type.rs b/src/test/compile-fail/builtin-superkinds-self-type.rs index 6228b924729..9826a5a0126 100644 --- a/src/test/compile-fail/builtin-superkinds-self-type.rs +++ b/src/test/compile-fail/builtin-superkinds-self-type.rs @@ -11,7 +11,7 @@ // Tests (negatively) the ability for the Self type in default methods // to use capabilities granted by builtin kinds as supertraits. -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; trait Foo : Sync+'static { fn foo(self, mut chan: Sender<Self>) { } diff --git a/src/test/compile-fail/comm-not-freeze-receiver.rs b/src/test/compile-fail/comm-not-freeze-receiver.rs index 83e329db0da..2e535b39509 100644 --- a/src/test/compile-fail/comm-not-freeze-receiver.rs +++ b/src/test/compile-fail/comm-not-freeze-receiver.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::Receiver; +use std::sync::mpsc::Receiver; fn test<T: Sync>() {} diff --git a/src/test/compile-fail/comm-not-freeze.rs b/src/test/compile-fail/comm-not-freeze.rs index 296b4187a1f..1b1c43e4793 100644 --- a/src/test/compile-fail/comm-not-freeze.rs +++ b/src/test/compile-fail/comm-not-freeze.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::Sender; +use std::sync::mpsc::Sender; fn test<T: Sync>() {} diff --git a/src/test/compile-fail/issue-12041.rs b/src/test/compile-fail/issue-12041.rs index f42e1a1bc15..094f6d64edc 100644 --- a/src/test/compile-fail/issue-12041.rs +++ b/src/test/compile-fail/issue-12041.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::channel; +use std::sync::mpsc::channel; use std::thread::Thread; fn main() { diff --git a/src/test/compile-fail/unsendable-class.rs b/src/test/compile-fail/unsendable-class.rs index 312f26394b2..993df8e59f3 100644 --- a/src/test/compile-fail/unsendable-class.rs +++ b/src/test/compile-fail/unsendable-class.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::channel; +use std::sync::mpsc::channel; // Test that a class with an unsendable field can't be // sent diff --git a/src/test/run-pass/bool.rs b/src/test/run-pass/bool.rs index 45710408172..b3c4802530e 100644 --- a/src/test/run-pass/bool.rs +++ b/src/test/run-pass/bool.rs @@ -10,7 +10,7 @@ // Basic boolean tests -use std::cmp::{Equal, Greater, Less}; +use std::cmp::Ordering::{Equal, Greater, Less}; use std::ops::{BitAnd, BitOr, BitXor}; fn main() { diff --git a/src/test/run-pass/builtin-superkinds-capabilities-transitive.rs b/src/test/run-pass/builtin-superkinds-capabilities-transitive.rs index 83c1696f7b2..365670db6f9 100644 --- a/src/test/run-pass/builtin-superkinds-capabilities-transitive.rs +++ b/src/test/run-pass/builtin-superkinds-capabilities-transitive.rs @@ -14,7 +14,7 @@ // a Send. Basically this just makes sure rustc is using // each_bound_trait_and_supertraits in type_contents correctly. -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; trait Bar : Send { } trait Foo : Bar { } @@ -23,11 +23,11 @@ impl <T: Send> Foo for T { } impl <T: Send> Bar for T { } fn foo<T: Foo>(val: T, chan: Sender<T>) { - chan.send(val); + chan.send(val).unwrap(); } pub fn main() { let (tx, rx) = channel(); foo(31337i, tx); - assert!(rx.recv() == 31337i); + assert!(rx.recv().unwrap() == 31337i); } diff --git a/src/test/run-pass/builtin-superkinds-capabilities-xc.rs b/src/test/run-pass/builtin-superkinds-capabilities-xc.rs index 1f002ec79c6..1f42076f6d6 100644 --- a/src/test/run-pass/builtin-superkinds-capabilities-xc.rs +++ b/src/test/run-pass/builtin-superkinds-capabilities-xc.rs @@ -16,7 +16,7 @@ extern crate trait_superkinds_in_metadata; -use std::comm::{channel, Sender, Receiver}; +use std::sync::mpsc::{channel, Sender, Receiver}; use trait_superkinds_in_metadata::{RequiresRequiresShareAndSend, RequiresShare}; #[deriving(PartialEq)] @@ -26,11 +26,11 @@ impl <T: Sync> RequiresShare for X<T> { } impl <T: Sync+Send> RequiresRequiresShareAndSend for X<T> { } fn foo<T: RequiresRequiresShareAndSend>(val: T, chan: Sender<T>) { - chan.send(val); + chan.send(val).unwrap(); } pub fn main() { let (tx, rx): (Sender<X<int>>, Receiver<X<int>>) = channel(); foo(X(31337i), tx); - assert!(rx.recv() == X(31337i)); + assert!(rx.recv().unwrap() == X(31337i)); } diff --git a/src/test/run-pass/builtin-superkinds-capabilities.rs b/src/test/run-pass/builtin-superkinds-capabilities.rs index 46ad3f93b0b..7f4a2398f54 100644 --- a/src/test/run-pass/builtin-superkinds-capabilities.rs +++ b/src/test/run-pass/builtin-superkinds-capabilities.rs @@ -12,18 +12,18 @@ // builtin-kinds, e.g., if a trait requires Send to implement, then // at usage site of that trait, we know we have the Send capability. -use std::comm::{channel, Sender, Receiver}; +use std::sync::mpsc::{channel, Sender, Receiver}; trait Foo : Send { } impl <T: Send> Foo for T { } fn foo<T: Foo>(val: T, chan: Sender<T>) { - chan.send(val); + chan.send(val).unwrap(); } pub fn main() { let (tx, rx): (Sender<int>, Receiver<int>) = channel(); foo(31337i, tx); - assert!(rx.recv() == 31337i); + assert!(rx.recv().unwrap() == 31337i); } diff --git a/src/test/run-pass/builtin-superkinds-self-type.rs b/src/test/run-pass/builtin-superkinds-self-type.rs index 27a7cd909fe..bf06bf8b8c6 100644 --- a/src/test/run-pass/builtin-superkinds-self-type.rs +++ b/src/test/run-pass/builtin-superkinds-self-type.rs @@ -11,11 +11,11 @@ // Tests the ability for the Self type in default methods to use // capabilities granted by builtin kinds as supertraits. -use std::comm::{Sender, channel}; +use std::sync::mpsc::{Sender, channel}; trait Foo : Send { fn foo(self, tx: Sender<Self>) { - tx.send(self); + tx.send(self).unwrap(); } } @@ -24,5 +24,5 @@ impl <T: Send> Foo for T { } pub fn main() { let (tx, rx) = channel(); 1193182i.foo(tx); - assert!(rx.recv() == 1193182i); + assert!(rx.recv().unwrap() == 1193182i); } diff --git a/src/test/run-pass/capturing-logging.rs b/src/test/run-pass/capturing-logging.rs index 4ed444f92ac..3f6d6a02c79 100644 --- a/src/test/run-pass/capturing-logging.rs +++ b/src/test/run-pass/capturing-logging.rs @@ -17,7 +17,7 @@ extern crate log; use log::{set_logger, Logger, LogRecord}; -use std::comm::channel; +use std::sync::mpsc::channel; use std::fmt; use std::io::{ChanReader, ChanWriter}; use std::thread::Thread; diff --git a/src/test/run-pass/cci_capture_clause.rs b/src/test/run-pass/cci_capture_clause.rs index c4bf8131506..8b2947ba3ee 100644 --- a/src/test/run-pass/cci_capture_clause.rs +++ b/src/test/run-pass/cci_capture_clause.rs @@ -16,5 +16,5 @@ extern crate cci_capture_clause; pub fn main() { - cci_capture_clause::foo(()).recv() + cci_capture_clause::foo(()).recv().unwrap(); } diff --git a/src/test/run-pass/closure-bounds-can-capture-chan.rs b/src/test/run-pass/closure-bounds-can-capture-chan.rs index e1dc5802162..816b28c3a9a 100644 --- a/src/test/run-pass/closure-bounds-can-capture-chan.rs +++ b/src/test/run-pass/closure-bounds-can-capture-chan.rs @@ -10,7 +10,7 @@ #![feature(unboxed_closures)] -use std::comm::channel; +use std::sync::mpsc::channel; fn foo<F:FnOnce()+Send>(blk: F) { blk(); @@ -19,7 +19,7 @@ fn foo<F:FnOnce()+Send>(blk: F) { pub fn main() { let (tx, rx) = channel(); foo(move || { - tx.send(()); + tx.send(()).unwrap(); }); - rx.recv(); + rx.recv().unwrap(); } diff --git a/src/test/run-pass/comm.rs b/src/test/run-pass/comm.rs index 119bdaff98e..5cfc692aae4 100644 --- a/src/test/run-pass/comm.rs +++ b/src/test/run-pass/comm.rs @@ -9,12 +9,12 @@ // except according to those terms. use std::task; -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; pub fn main() { let (tx, rx) = channel(); let _t = task::spawn(move|| { child(&tx) }); - let y = rx.recv(); + let y = rx.recv().unwrap(); println!("received"); println!("{}", y); assert_eq!(y, 10); @@ -22,6 +22,6 @@ pub fn main() { fn child(c: &Sender<int>) { println!("sending"); - c.send(10); + c.send(10).unwrap(); println!("value sent"); } diff --git a/src/test/run-pass/core-run-destroy.rs b/src/test/run-pass/core-run-destroy.rs index 198915fe02e..c1db8a6eb13 100644 --- a/src/test/run-pass/core-run-destroy.rs +++ b/src/test/run-pass/core-run-destroy.rs @@ -23,7 +23,7 @@ extern crate libc; use std::io::{Process, Command, timer}; use std::time::Duration; use std::str; -use std::comm::channel; +use std::sync::mpsc::channel; use std::thread::Thread; macro_rules! succeed( ($e:expr) => ( @@ -88,13 +88,13 @@ pub fn test_destroy_actually_kills(force: bool) { let rx2 = t.oneshot(Duration::milliseconds(1000)); Thread::spawn(move|| { select! { - () = rx2.recv() => unsafe { libc::exit(1) }, - () = rx1.recv() => {} + _ = rx2.recv() => unsafe { libc::exit(1) }, + _ = rx1.recv() => {} } }).detach(); match p.wait().unwrap() { ExitStatus(..) => panic!("expected a signal"), - ExitSignal(..) => tx.send(()), + ExitSignal(..) => tx.send(()).unwrap(), } } diff --git a/src/test/run-pass/drop-trait-enum.rs b/src/test/run-pass/drop-trait-enum.rs index bca61ab1bd3..1ad0e87b645 100644 --- a/src/test/run-pass/drop-trait-enum.rs +++ b/src/test/run-pass/drop-trait-enum.rs @@ -9,7 +9,7 @@ // except according to those terms. use std::task; -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; #[deriving(PartialEq, Show)] enum Message { @@ -23,7 +23,7 @@ struct SendOnDrop { impl Drop for SendOnDrop { fn drop(&mut self) { - self.sender.send(Message::Dropped); + self.sender.send(Message::Dropped).unwrap(); } } @@ -37,10 +37,10 @@ impl Drop for Foo { fn drop(&mut self) { match self { &Foo::SimpleVariant(ref mut sender) => { - sender.send(Message::DestructorRan); + sender.send(Message::DestructorRan).unwrap(); } &Foo::NestedVariant(_, _, ref mut sender) => { - sender.send(Message::DestructorRan); + sender.send(Message::DestructorRan).unwrap(); } &Foo::FailingVariant { .. } => { panic!("Failed"); @@ -54,23 +54,23 @@ pub fn main() { { let v = Foo::SimpleVariant(sender); } - assert_eq!(receiver.recv(), Message::DestructorRan); - assert_eq!(receiver.recv_opt().ok(), None); + assert_eq!(receiver.recv().unwrap(), Message::DestructorRan); + assert_eq!(receiver.recv().ok(), None); let (sender, receiver) = channel(); { let v = Foo::NestedVariant(box 42u, SendOnDrop { sender: sender.clone() }, sender); } - assert_eq!(receiver.recv(), Message::DestructorRan); - assert_eq!(receiver.recv(), Message::Dropped); - assert_eq!(receiver.recv_opt().ok(), None); + assert_eq!(receiver.recv().unwrap(), Message::DestructorRan); + assert_eq!(receiver.recv().unwrap(), Message::Dropped); + assert_eq!(receiver.recv().ok(), None); let (sender, receiver) = channel(); task::spawn(move|| { let v = Foo::FailingVariant { on_drop: SendOnDrop { sender: sender } }; }); - assert_eq!(receiver.recv(), Message::Dropped); - assert_eq!(receiver.recv_opt().ok(), None); + assert_eq!(receiver.recv().unwrap(), Message::Dropped); + assert_eq!(receiver.recv().ok(), None); let (sender, receiver) = channel(); { @@ -83,11 +83,11 @@ pub fn main() { v = Foo::FailingVariant { on_drop: SendOnDrop { sender: sender } }; }); } - assert_eq!(receiver.recv(), Message::DestructorRan); - assert_eq!(receiver.recv(), Message::Dropped); - assert_eq!(receiver.recv(), Message::DestructorRan); - assert_eq!(receiver.recv(), Message::Dropped); - assert_eq!(receiver.recv(), Message::DestructorRan); - assert_eq!(receiver.recv(), Message::Dropped); - assert_eq!(receiver.recv_opt().ok(), None); + assert_eq!(receiver.recv().unwrap(), Message::DestructorRan); + assert_eq!(receiver.recv().unwrap(), Message::Dropped); + assert_eq!(receiver.recv().unwrap(), Message::DestructorRan); + assert_eq!(receiver.recv().unwrap(), Message::Dropped); + assert_eq!(receiver.recv().unwrap(), Message::DestructorRan); + assert_eq!(receiver.recv().unwrap(), Message::Dropped); + assert_eq!(receiver.recv().ok(), None); } diff --git a/src/test/run-pass/hashmap-memory.rs b/src/test/run-pass/hashmap-memory.rs index 6155ca63572..162d7f10255 100644 --- a/src/test/run-pass/hashmap-memory.rs +++ b/src/test/run-pass/hashmap-memory.rs @@ -21,7 +21,7 @@ pub fn map(filename: String, emit: map_reduce::putter) { mod map_reduce { use std::collections::HashMap; - use std::comm::{channel, Sender}; + use std::sync::mpsc::{channel, Sender}; use std::str; use std::task; @@ -50,16 +50,16 @@ mod map_reduce { } let (tx, rx) = channel(); println!("sending find_reducer"); - ctrl.send(ctrl_proto::find_reducer(key.as_bytes().to_vec(), tx)); + ctrl.send(ctrl_proto::find_reducer(key.as_bytes().to_vec(), tx)).unwrap(); println!("receiving"); - let c = rx.recv(); + let c = rx.recv().unwrap(); println!("{}", c); im.insert(key, c); } let ctrl_clone = ctrl.clone(); ::map(input, |a,b| emit(&mut intermediates, ctrl.clone(), a, b) ); - ctrl_clone.send(ctrl_proto::mapper_done); + ctrl_clone.send(ctrl_proto::mapper_done).unwrap(); } pub fn map_reduce(inputs: Vec<String>) { @@ -77,7 +77,7 @@ mod map_reduce { let mut num_mappers = inputs.len() as int; while num_mappers > 0 { - match rx.recv() { + match rx.recv().unwrap() { ctrl_proto::mapper_done => { num_mappers -= 1; } ctrl_proto::find_reducer(k, cc) => { let mut c; @@ -86,7 +86,7 @@ mod map_reduce { Some(&_c) => { c = _c; } None => { c = 0; } } - cc.send(c); + cc.send(c).unwrap(); } } } diff --git a/src/test/run-pass/issue-13494.rs b/src/test/run-pass/issue-13494.rs index e6da859c110..b9339c1cc0d 100644 --- a/src/test/run-pass/issue-13494.rs +++ b/src/test/run-pass/issue-13494.rs @@ -11,12 +11,12 @@ // This test may not always fail, but it can be flaky if the race it used to // expose is still present. -use std::comm::{channel, Sender, Receiver}; +use std::sync::mpsc::{channel, Sender, Receiver}; use std::thread::Thread; fn helper(rx: Receiver<Sender<()>>) { for tx in rx.iter() { - let _ = tx.send_opt(()); + let _ = tx.send(()); } } @@ -25,11 +25,11 @@ fn main() { let _t = Thread::spawn(move|| { helper(rx) }).detach(); let (snd, rcv) = channel::<int>(); for _ in range(1i, 100000i) { - snd.send(1i); + snd.send(1i).unwrap(); let (tx2, rx2) = channel(); - tx.send(tx2); + tx.send(tx2).unwrap(); select! { - () = rx2.recv() => (), + _ = rx2.recv() => (), _ = rcv.recv() => () } } diff --git a/src/test/run-pass/issue-3609.rs b/src/test/run-pass/issue-3609.rs index df2a9e6bfa9..a02dbb6035b 100644 --- a/src/test/run-pass/issue-3609.rs +++ b/src/test/run-pass/issue-3609.rs @@ -11,7 +11,7 @@ #![feature(default_type_params)] use std::task; -use std::comm::Sender; +use std::sync::mpsc::Sender; use std::thunk::Invoke; type RingBuffer = Vec<f64> ; diff --git a/src/test/run-pass/issue-4446.rs b/src/test/run-pass/issue-4446.rs index 871e52e3639..30e1a14ecff 100644 --- a/src/test/run-pass/issue-4446.rs +++ b/src/test/run-pass/issue-4446.rs @@ -9,15 +9,15 @@ // except according to those terms. use std::io::println; -use std::comm::channel; +use std::sync::mpsc::channel; use std::thread::Thread; pub fn main() { let (tx, rx) = channel(); - tx.send("hello, world"); + tx.send("hello, world").unwrap(); Thread::spawn(move|| { - println(rx.recv()); + println(rx.recv().unwrap()); }).join().ok().unwrap(); } diff --git a/src/test/run-pass/issue-4448.rs b/src/test/run-pass/issue-4448.rs index 6692988f6da..7e53722726f 100644 --- a/src/test/run-pass/issue-4448.rs +++ b/src/test/run-pass/issue-4448.rs @@ -8,15 +8,16 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::task; -use std::comm::channel; +use std::sync::mpsc::channel; +use std::thread::Thread; pub fn main() { let (tx, rx) = channel::<&'static str>(); - task::spawn(move|| { - assert_eq!(rx.recv(), "hello, world"); + let t = Thread::spawn(move|| { + assert_eq!(rx.recv().unwrap(), "hello, world"); }); - tx.send("hello, world"); + tx.send("hello, world").unwrap(); + t.join().ok().unwrap(); } diff --git a/src/test/run-pass/issue-8827.rs b/src/test/run-pass/issue-8827.rs index 4163ebf4573..39695a8339f 100644 --- a/src/test/run-pass/issue-8827.rs +++ b/src/test/run-pass/issue-8827.rs @@ -9,19 +9,19 @@ // except according to those terms. use std::thread::Thread; -use std::comm::{channel, Receiver}; +use std::sync::mpsc::{channel, Receiver}; fn periodical(n: int) -> Receiver<bool> { let (chan, port) = channel(); Thread::spawn(move|| { loop { for _ in range(1, n) { - match chan.send_opt(false) { + match chan.send(false) { Ok(()) => {} Err(..) => break, } } - match chan.send_opt(true) { + match chan.send(true) { Ok(()) => {} Err(..) => break } @@ -35,7 +35,7 @@ fn integers() -> Receiver<int> { Thread::spawn(move|| { let mut i = 1; loop { - match chan.send_opt(i) { + match chan.send(i) { Ok(()) => {} Err(..) => break, } @@ -50,7 +50,7 @@ fn main() { let threes = periodical(3); let fives = periodical(5); for _ in range(1i, 100i) { - match (ints.recv(), threes.recv(), fives.recv()) { + match (ints.recv().unwrap(), threes.recv().unwrap(), fives.recv().unwrap()) { (_, true, true) => println!("FizzBuzz"), (_, true, false) => println!("Fizz"), (_, false, true) => println!("Buzz"), diff --git a/src/test/run-pass/issue-9396.rs b/src/test/run-pass/issue-9396.rs index d3cbd556776..34bb50c5cf6 100644 --- a/src/test/run-pass/issue-9396.rs +++ b/src/test/run-pass/issue-9396.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::{mod, channel}; +use std::sync::mpsc::{TryRecvError, channel}; use std::io::timer::Timer; use std::thread::Thread; use std::time::Duration; @@ -18,13 +18,13 @@ pub fn main() { let _t = Thread::spawn(move||{ let mut timer = Timer::new().unwrap(); timer.sleep(Duration::milliseconds(10)); - tx.send(()); + tx.send(()).unwrap(); }); loop { match rx.try_recv() { Ok(()) => break, - Err(comm::Empty) => {} - Err(comm::Disconnected) => unreachable!() + Err(TryRecvError::Empty) => {} + Err(TryRecvError::Disconnected) => unreachable!() } } } diff --git a/src/test/run-pass/ivec-tag.rs b/src/test/run-pass/ivec-tag.rs index 2e6127541a1..a9a50b9ef25 100644 --- a/src/test/run-pass/ivec-tag.rs +++ b/src/test/run-pass/ivec-tag.rs @@ -9,12 +9,12 @@ // except according to those terms. use std::task; -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; fn producer(tx: &Sender<Vec<u8>>) { tx.send( vec!(1u8, 2u8, 3u8, 4u8, 5u8, 6u8, 7u8, 8u8, 9u8, 10u8, 11u8, 12u8, - 13u8)); + 13u8)).unwrap(); } pub fn main() { @@ -23,5 +23,5 @@ pub fn main() { producer(&tx) }); - let _data: Vec<u8> = rx.recv(); + let _data: Vec<u8> = rx.recv().unwrap(); } diff --git a/src/test/run-pass/rust-log-filter.rs b/src/test/run-pass/rust-log-filter.rs index 8e547527f72..2612483ded4 100644 --- a/src/test/run-pass/rust-log-filter.rs +++ b/src/test/run-pass/rust-log-filter.rs @@ -14,7 +14,7 @@ #[phase(plugin,link)] extern crate log; -use std::comm::{channel, Sender, Receiver}; +use std::sync::mpsc::{channel, Sender, Receiver}; use std::thread::Thread; pub struct ChannelLogger { @@ -30,7 +30,7 @@ impl ChannelLogger { impl log::Logger for ChannelLogger { fn log(&mut self, record: &log::LogRecord) { - self.tx.send(format!("{}", record.args)); + self.tx.send(format!("{}", record.args)).unwrap(); } } @@ -49,9 +49,9 @@ pub fn main() { info!("f1o"); }); - assert_eq!(rx.recv().as_slice(), "foo"); - assert_eq!(rx.recv().as_slice(), "foo bar"); - assert_eq!(rx.recv().as_slice(), "bar foo"); - assert_eq!(rx.recv().as_slice(), "f1o"); - assert!(rx.recv_opt().is_err()); + assert_eq!(rx.recv().unwrap().as_slice(), "foo"); + assert_eq!(rx.recv().unwrap().as_slice(), "foo bar"); + assert_eq!(rx.recv().unwrap().as_slice(), "bar foo"); + assert_eq!(rx.recv().unwrap().as_slice(), "f1o"); + assert!(rx.recv().is_err()); } diff --git a/src/test/run-pass/send-resource.rs b/src/test/run-pass/send-resource.rs index 71620d05e01..7fd9706bd0f 100644 --- a/src/test/run-pass/send-resource.rs +++ b/src/test/run-pass/send-resource.rs @@ -9,7 +9,7 @@ // except according to those terms. use std::task; -use std::comm::channel; +use std::sync::mpsc::channel; struct test { f: int, @@ -30,10 +30,10 @@ pub fn main() { task::spawn(move|| { let (tx2, rx2) = channel(); - tx.send(tx2); + tx.send(tx2).unwrap(); - let _r = rx2.recv(); + let _r = rx2.recv().unwrap(); }); - rx.recv().send(test(42)); + rx.recv().unwrap().send(test(42)).unwrap(); } diff --git a/src/test/run-pass/send-type-inference.rs b/src/test/run-pass/send-type-inference.rs index 181874705da..ae992a0a358 100644 --- a/src/test/run-pass/send-type-inference.rs +++ b/src/test/run-pass/send-type-inference.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; // tests that ctrl's type gets inferred properly struct Command<K, V> { diff --git a/src/test/run-pass/sendable-class.rs b/src/test/run-pass/sendable-class.rs index 3d5f0af0e3f..8691d5e875b 100644 --- a/src/test/run-pass/sendable-class.rs +++ b/src/test/run-pass/sendable-class.rs @@ -10,7 +10,7 @@ // Test that a class with only sendable fields can be sent -use std::comm::channel; +use std::sync::mpsc::channel; struct foo { i: int, diff --git a/src/test/run-pass/spawn-types.rs b/src/test/run-pass/spawn-types.rs index e3b2a2903bc..1c86e3e6ea2 100644 --- a/src/test/run-pass/spawn-types.rs +++ b/src/test/run-pass/spawn-types.rs @@ -15,7 +15,7 @@ */ use std::task; -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; type ctx = Sender<int>; diff --git a/src/test/run-pass/task-comm-0.rs b/src/test/run-pass/task-comm-0.rs index 7c664b21fad..de077ffd190 100644 --- a/src/test/run-pass/task-comm-0.rs +++ b/src/test/run-pass/task-comm-0.rs @@ -9,27 +9,27 @@ // except according to those terms. use std::task; -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; pub fn main() { test05(); } fn test05_start(tx : &Sender<int>) { - tx.send(10); + tx.send(10).unwrap(); println!("sent 10"); - tx.send(20); + tx.send(20).unwrap(); println!("sent 20"); - tx.send(30); + tx.send(30).unwrap(); println!("sent 30"); } fn test05() { let (tx, rx) = channel(); task::spawn(move|| { test05_start(&tx) }); - let mut value: int = rx.recv(); + let mut value: int = rx.recv().unwrap(); println!("{}", value); - value = rx.recv(); + value = rx.recv().unwrap(); println!("{}", value); - value = rx.recv(); + value = rx.recv().unwrap(); println!("{}", value); assert_eq!(value, 30); } diff --git a/src/test/run-pass/task-comm-10.rs b/src/test/run-pass/task-comm-10.rs index c35e2f2f37a..93dca923b6b 100644 --- a/src/test/run-pass/task-comm-10.rs +++ b/src/test/run-pass/task-comm-10.rs @@ -9,18 +9,18 @@ // except according to those terms. use std::task; -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; fn start(tx: &Sender<Sender<String>>) { let (tx2, rx) = channel(); - tx.send(tx2); + tx.send(tx2).unwrap(); let mut a; let mut b; - a = rx.recv(); + a = rx.recv().unwrap(); assert!(a == "A".to_string()); println!("{}", a); - b = rx.recv(); + b = rx.recv().unwrap(); assert!(b == "B".to_string()); println!("{}", b); } @@ -29,8 +29,8 @@ pub fn main() { let (tx, rx) = channel(); let _child = task::spawn(move|| { start(&tx) }); - let mut c = rx.recv(); - c.send("A".to_string()); - c.send("B".to_string()); + let mut c = rx.recv().unwrap(); + c.send("A".to_string()).unwrap(); + c.send("B".to_string()).unwrap(); task::deschedule(); } diff --git a/src/test/run-pass/task-comm-11.rs b/src/test/run-pass/task-comm-11.rs index 10b4e516311..8168e84e426 100644 --- a/src/test/run-pass/task-comm-11.rs +++ b/src/test/run-pass/task-comm-11.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; use std::task; fn start(tx: &Sender<Sender<int>>) { diff --git a/src/test/run-pass/task-comm-13.rs b/src/test/run-pass/task-comm-13.rs index 31da1168bf7..bb92ef38728 100644 --- a/src/test/run-pass/task-comm-13.rs +++ b/src/test/run-pass/task-comm-13.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; use std::task; fn start(tx: &Sender<int>, start: int, number_of_messages: int) { diff --git a/src/test/run-pass/task-comm-14.rs b/src/test/run-pass/task-comm-14.rs index 8ec589d5db5..d63cbd5c8ba 100644 --- a/src/test/run-pass/task-comm-14.rs +++ b/src/test/run-pass/task-comm-14.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; use std::task; pub fn main() { diff --git a/src/test/run-pass/task-comm-15.rs b/src/test/run-pass/task-comm-15.rs index b69644244ab..7c652ddc406 100644 --- a/src/test/run-pass/task-comm-15.rs +++ b/src/test/run-pass/task-comm-15.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; use std::task; fn start(tx: &Sender<int>, i0: int) { diff --git a/src/test/run-pass/task-comm-16.rs b/src/test/run-pass/task-comm-16.rs index 5e71ed3ba06..b7098eb30a3 100644 --- a/src/test/run-pass/task-comm-16.rs +++ b/src/test/run-pass/task-comm-16.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::channel; +use std::sync::mpsc::channel; use std::cmp; // Tests of ports and channels on various types @@ -17,9 +17,9 @@ fn test_rec() { let (tx, rx) = channel(); let r0: R = R {val0: 0, val1: 1u8, val2: '2'}; - tx.send(r0); + tx.send(r0).unwrap(); let mut r1: R; - r1 = rx.recv(); + r1 = rx.recv().unwrap(); assert_eq!(r1.val0, 0); assert_eq!(r1.val1, 1u8); assert_eq!(r1.val2, '2'); @@ -28,8 +28,8 @@ fn test_rec() { fn test_vec() { let (tx, rx) = channel(); let v0: Vec<int> = vec!(0, 1, 2); - tx.send(v0); - let v1 = rx.recv(); + tx.send(v0).unwrap(); + let v1 = rx.recv().unwrap(); assert_eq!(v1[0], 0); assert_eq!(v1[1], 1); assert_eq!(v1[2], 2); @@ -38,8 +38,8 @@ fn test_vec() { fn test_str() { let (tx, rx) = channel(); let s0 = "test".to_string(); - tx.send(s0); - let s1 = rx.recv(); + tx.send(s0).unwrap(); + let s1 = rx.recv().unwrap(); assert_eq!(s1.as_bytes()[0], 't' as u8); assert_eq!(s1.as_bytes()[1], 'e' as u8); assert_eq!(s1.as_bytes()[2], 's' as u8); @@ -82,28 +82,28 @@ impl cmp::PartialEq for t { fn test_tag() { let (tx, rx) = channel(); - tx.send(t::tag1); - tx.send(t::tag2(10)); - tx.send(t::tag3(10, 11u8, 'A')); + tx.send(t::tag1).unwrap(); + tx.send(t::tag2(10)).unwrap(); + tx.send(t::tag3(10, 11u8, 'A')).unwrap(); let mut t1: t; - t1 = rx.recv(); + t1 = rx.recv().unwrap(); assert_eq!(t1, t::tag1); - t1 = rx.recv(); + t1 = rx.recv().unwrap(); assert_eq!(t1, t::tag2(10)); - t1 = rx.recv(); + t1 = rx.recv().unwrap(); assert_eq!(t1, t::tag3(10, 11u8, 'A')); } fn test_chan() { let (tx1, rx1) = channel(); let (tx2, rx2) = channel(); - tx1.send(tx2); - let tx2 = rx1.recv(); + tx1.send(tx2).unwrap(); + let tx2 = rx1.recv().unwrap(); // Does the transmitted channel still work? - tx2.send(10); + tx2.send(10).unwrap(); let mut i: int; - i = rx2.recv(); + i = rx2.recv().unwrap(); assert_eq!(i, 10); } diff --git a/src/test/run-pass/task-comm-3.rs b/src/test/run-pass/task-comm-3.rs index c7a6da6a794..a002a597481 100644 --- a/src/test/run-pass/task-comm-3.rs +++ b/src/test/run-pass/task-comm-3.rs @@ -11,7 +11,7 @@ // no-pretty-expanded FIXME #15189 use std::thread::Thread; -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; pub fn main() { println!("===== WITHOUT THREADS ====="); test00(); } @@ -20,7 +20,7 @@ fn test00_start(ch: &Sender<int>, message: int, count: int) { let mut i: int = 0; while i < count { println!("Sending Message"); - ch.send(message + 0); + ch.send(message + 0).unwrap(); i = i + 1; } println!("Ending test00_start"); @@ -54,7 +54,7 @@ fn test00() { for _r in results.iter() { i = 0; while i < number_of_messages { - let value = rx.recv(); + let value = rx.recv().unwrap(); sum += value; i = i + 1; } diff --git a/src/test/run-pass/task-comm-4.rs b/src/test/run-pass/task-comm-4.rs index 0e75479046b..1f1b750aa57 100644 --- a/src/test/run-pass/task-comm-4.rs +++ b/src/test/run-pass/task-comm-4.rs @@ -10,7 +10,7 @@ #![allow(dead_assignment)] -use std::comm::channel; +use std::sync::mpsc::channel; pub fn main() { test00(); } @@ -18,36 +18,36 @@ fn test00() { let mut r: int = 0; let mut sum: int = 0; let (tx, rx) = channel(); - tx.send(1); - tx.send(2); - tx.send(3); - tx.send(4); - r = rx.recv(); + tx.send(1).unwrap(); + tx.send(2).unwrap(); + tx.send(3).unwrap(); + tx.send(4).unwrap(); + r = rx.recv().unwrap(); sum += r; println!("{}", r); - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; println!("{}", r); - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; println!("{}", r); - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; println!("{}", r); - tx.send(5); - tx.send(6); - tx.send(7); - tx.send(8); - r = rx.recv(); + tx.send(5).unwrap(); + tx.send(6).unwrap(); + tx.send(7).unwrap(); + tx.send(8).unwrap(); + r = rx.recv().unwrap(); sum += r; println!("{}", r); - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; println!("{}", r); - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; println!("{}", r); - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; println!("{}", r); assert_eq!(sum, 1 + 2 + 3 + 4 + 5 + 6 + 7 + 8); diff --git a/src/test/run-pass/task-comm-5.rs b/src/test/run-pass/task-comm-5.rs index d4edbf85cd8..039308d5cfe 100644 --- a/src/test/run-pass/task-comm-5.rs +++ b/src/test/run-pass/task-comm-5.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::channel; +use std::sync::mpsc::channel; pub fn main() { test00(); } @@ -18,8 +18,8 @@ fn test00() { let (tx, rx) = channel(); let number_of_messages: int = 1000; let mut i: int = 0; - while i < number_of_messages { tx.send(i + 0); i += 1; } + while i < number_of_messages { tx.send(i + 0).unwrap(); i += 1; } i = 0; - while i < number_of_messages { sum += rx.recv(); i += 1; } + while i < number_of_messages { sum += rx.recv().unwrap(); i += 1; } assert_eq!(sum, number_of_messages * (number_of_messages - 1) / 2); } diff --git a/src/test/run-pass/task-comm-6.rs b/src/test/run-pass/task-comm-6.rs index 4c18e5dc313..7cdfddcdeb1 100644 --- a/src/test/run-pass/task-comm-6.rs +++ b/src/test/run-pass/task-comm-6.rs @@ -10,7 +10,7 @@ #![allow(dead_assignment)] -use std::comm::channel; +use std::sync::mpsc::channel; pub fn main() { test00(); } @@ -25,21 +25,21 @@ fn test00() { let number_of_messages: int = 1000; let mut i: int = 0; while i < number_of_messages { - tx0.send(i + 0); - tx1.send(i + 0); - tx2.send(i + 0); - tx3.send(i + 0); + tx0.send(i + 0).unwrap(); + tx1.send(i + 0).unwrap(); + tx2.send(i + 0).unwrap(); + tx3.send(i + 0).unwrap(); i += 1; } i = 0; while i < number_of_messages { - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; i += 1; } diff --git a/src/test/run-pass/task-comm-7.rs b/src/test/run-pass/task-comm-7.rs index e7e5c04e195..054090eca39 100644 --- a/src/test/run-pass/task-comm-7.rs +++ b/src/test/run-pass/task-comm-7.rs @@ -10,7 +10,7 @@ #![allow(dead_assignment)] -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; use std::task; pub fn main() { test00(); } @@ -18,7 +18,7 @@ pub fn main() { test00(); } fn test00_start(c: &Sender<int>, start: int, number_of_messages: int) { let mut i: int = 0; - while i < number_of_messages { c.send(start + i); i += 1; } + while i < number_of_messages { c.send(start + i).unwrap(); i += 1; } } fn test00() { @@ -46,13 +46,13 @@ fn test00() { let mut i: int = 0; while i < number_of_messages { - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; - r = rx.recv(); + r = rx.recv().unwrap(); sum += r; i += 1; } diff --git a/src/test/run-pass/task-comm-9.rs b/src/test/run-pass/task-comm-9.rs index 5a271f9dc05..d9faf6ee4e4 100644 --- a/src/test/run-pass/task-comm-9.rs +++ b/src/test/run-pass/task-comm-9.rs @@ -9,13 +9,13 @@ // except according to those terms. use std::thread::Thread; -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; pub fn main() { test00(); } fn test00_start(c: &Sender<int>, number_of_messages: int) { let mut i: int = 0; - while i < number_of_messages { c.send(i + 0); i += 1; } + while i < number_of_messages { c.send(i + 0).unwrap(); i += 1; } } fn test00() { @@ -30,7 +30,7 @@ fn test00() { let mut i: int = 0; while i < number_of_messages { - sum += rx.recv(); + sum += rx.recv().unwrap(); println!("{}", r); i += 1; } diff --git a/src/test/run-pass/task-comm-chan-nil.rs b/src/test/run-pass/task-comm-chan-nil.rs index 368cac1d27d..78a42632001 100644 --- a/src/test/run-pass/task-comm-chan-nil.rs +++ b/src/test/run-pass/task-comm-chan-nil.rs @@ -8,14 +8,14 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::channel; +use std::sync::mpsc::channel; // rustboot can't transmit nils across channels because they don't have // any size, but rustc currently can because they do have size. Whether // or not this is desirable I don't know, but here's a regression test. pub fn main() { let (tx, rx) = channel(); - tx.send(()); - let n: () = rx.recv(); + tx.send(()).unwrap(); + let n: () = rx.recv().unwrap(); assert_eq!(n, ()); } diff --git a/src/test/run-pass/task-spawn-move-and-copy.rs b/src/test/run-pass/task-spawn-move-and-copy.rs index 1e10a4186fb..623a30eda1a 100644 --- a/src/test/run-pass/task-spawn-move-and-copy.rs +++ b/src/test/run-pass/task-spawn-move-and-copy.rs @@ -9,7 +9,7 @@ // except according to those terms. use std::task; -use std::comm::channel; +use std::sync::mpsc::channel; pub fn main() { let (tx, rx) = channel::<uint>(); @@ -19,9 +19,9 @@ pub fn main() { task::spawn(move || { let x_in_child = &(*x) as *const int as uint; - tx.send(x_in_child); + tx.send(x_in_child).unwrap(); }); - let x_in_child = rx.recv(); + let x_in_child = rx.recv().unwrap(); assert_eq!(x_in_parent, x_in_child); } diff --git a/src/test/run-pass/task-stderr.rs b/src/test/run-pass/task-stderr.rs index 048d7b2be72..a7eabe0edb3 100644 --- a/src/test/run-pass/task-stderr.rs +++ b/src/test/run-pass/task-stderr.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::channel; +use std::sync::mpsc::channel; use std::io::{ChanReader, ChanWriter}; use std::thread; diff --git a/src/test/run-pass/tcp-accept-stress.rs b/src/test/run-pass/tcp-accept-stress.rs index 8f5a45cd1dd..07f71fe580e 100644 --- a/src/test/run-pass/tcp-accept-stress.rs +++ b/src/test/run-pass/tcp-accept-stress.rs @@ -13,7 +13,7 @@ // quite quickly and it takes a few seconds for the sockets to get // recycled. -use std::comm::channel; +use std::sync::mpsc::channel; use std::io::{TcpListener, Listener, Acceptor, EndOfFile, TcpStream}; use std::sync::{atomic, Arc}; use std::thread::Thread; diff --git a/src/test/run-pass/tcp-connect-timeouts.rs b/src/test/run-pass/tcp-connect-timeouts.rs index e098e7e96ec..c33bdcf8698 100644 --- a/src/test/run-pass/tcp-connect-timeouts.rs +++ b/src/test/run-pass/tcp-connect-timeouts.rs @@ -27,7 +27,7 @@ use std::io::net::tcp::*; use std::io::test::*; use std::io; use std::time::Duration; -use std::comm::channel; +use std::sync::mpsc::channel; #[cfg_attr(target_os = "freebsd", ignore)] fn eventual_timeout() { @@ -37,10 +37,10 @@ fn eventual_timeout() { let (_tx2, rx2) = channel::<()>(); std::task::spawn(move|| { let _l = TcpListener::bind(addr).unwrap().listen(); - tx1.send(()); - let _ = rx2.recv_opt(); + tx1.send(()).unwrap(); + let _ = rx2.recv(); }); - rx1.recv(); + rx1.recv().unwrap(); let mut v = Vec::new(); for _ in range(0u, 10000) { diff --git a/src/test/run-pass/tempfile.rs b/src/test/run-pass/tempfile.rs index 4fcdf49d971..9e67095bb30 100644 --- a/src/test/run-pass/tempfile.rs +++ b/src/test/run-pass/tempfile.rs @@ -23,7 +23,7 @@ use std::io::{fs, TempDir}; use std::io; use std::os; use std::task; -use std::comm::channel; +use std::sync::mpsc::channel; fn test_tempdir() { let path = { @@ -39,11 +39,11 @@ fn test_rm_tempdir() { let (tx, rx) = channel(); let f = move|:| -> () { let tmp = TempDir::new("test_rm_tempdir").unwrap(); - tx.send(tmp.path().clone()); + tx.send(tmp.path().clone()).unwrap(); panic!("panic to unwind past `tmp`"); }; task::try(f); - let path = rx.recv(); + let path = rx.recv().unwrap(); assert!(!path.exists()); let tmp = TempDir::new("test_rm_tempdir").unwrap(); @@ -80,12 +80,12 @@ fn test_rm_tempdir_close() { let (tx, rx) = channel(); let f = move|:| -> () { let tmp = TempDir::new("test_rm_tempdir").unwrap(); - tx.send(tmp.path().clone()); + tx.send(tmp.path().clone()).unwrap(); tmp.close(); panic!("panic when unwinding past `tmp`"); }; task::try(f); - let path = rx.recv(); + let path = rx.recv().unwrap(); assert!(!path.exists()); let tmp = TempDir::new("test_rm_tempdir").unwrap(); diff --git a/src/test/run-pass/trait-bounds-in-arc.rs b/src/test/run-pass/trait-bounds-in-arc.rs index dd16514d83a..0d2cb60c213 100644 --- a/src/test/run-pass/trait-bounds-in-arc.rs +++ b/src/test/run-pass/trait-bounds-in-arc.rs @@ -12,7 +12,7 @@ // and shared between tasks as long as all types fulfill Send. use std::sync::Arc; -use std::comm::channel; +use std::sync::mpsc::channel; use std::task; trait Pet { diff --git a/src/test/run-pass/trivial-message.rs b/src/test/run-pass/trivial-message.rs index 1ff3ba3106b..6bece8265c0 100644 --- a/src/test/run-pass/trivial-message.rs +++ b/src/test/run-pass/trivial-message.rs @@ -13,7 +13,7 @@ message. */ -use std::comm::channel; +use std::sync::mpsc::channel; pub fn main() { let (tx, rx) = channel(); diff --git a/src/test/run-pass/unique-send-2.rs b/src/test/run-pass/unique-send-2.rs index e66c8730898..f88c458f2ed 100644 --- a/src/test/run-pass/unique-send-2.rs +++ b/src/test/run-pass/unique-send-2.rs @@ -9,10 +9,10 @@ // except according to those terms. use std::task; -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; fn child(tx: &Sender<Box<uint>>, i: uint) { - tx.send(box i); + tx.send(box i).unwrap(); } pub fn main() { @@ -29,7 +29,7 @@ pub fn main() { let mut actual = 0u; for _ in range(0u, n) { - let j = rx.recv(); + let j = rx.recv().unwrap(); actual += *j; } diff --git a/src/test/run-pass/unique-send.rs b/src/test/run-pass/unique-send.rs index 0a661a51bf1..afafb204c1c 100644 --- a/src/test/run-pass/unique-send.rs +++ b/src/test/run-pass/unique-send.rs @@ -8,11 +8,11 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::channel; +use std::sync::mpsc::channel; pub fn main() { let (tx, rx) = channel(); - tx.send(box 100i); - let v = rx.recv(); + tx.send(box 100i).unwrap(); + let v = rx.recv().unwrap(); assert_eq!(v, box 100i); } diff --git a/src/test/run-pass/unwind-resource.rs b/src/test/run-pass/unwind-resource.rs index 1284c938d55..943b2d3edd1 100644 --- a/src/test/run-pass/unwind-resource.rs +++ b/src/test/run-pass/unwind-resource.rs @@ -8,7 +8,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::comm::{channel, Sender}; +use std::sync::mpsc::{channel, Sender}; use std::task; struct complainer { @@ -18,7 +18,7 @@ struct complainer { impl Drop for complainer { fn drop(&mut self) { println!("About to send!"); - self.tx.send(true); + self.tx.send(true).unwrap(); println!("Sent!"); } } @@ -39,5 +39,5 @@ pub fn main() { let (tx, rx) = channel(); task::spawn(move|| f(tx.clone())); println!("hiiiiiiiii"); - assert!(rx.recv()); + assert!(rx.recv().unwrap()); } |