From 04f4bb429075221c951fb53a56b6455afb869917 Mon Sep 17 00:00:00 2001 From: Jonas Hietala Date: Tue, 16 Sep 2014 10:10:56 +0200 Subject: Rename std::io::net::unix to std::io::net::pipe. Renamed as we may support pipes for other platforms. Closes #12093 [breaking-change] --- src/libstd/io/net/mod.rs | 3 +- src/libstd/io/net/pipe.rs | 802 ++++++++++++++++++++++++++++++++++++++++++++++ src/libstd/io/net/unix.rs | 802 ---------------------------------------------- src/libstd/io/test.rs | 2 +- 4 files changed, 804 insertions(+), 805 deletions(-) create mode 100644 src/libstd/io/net/pipe.rs delete mode 100644 src/libstd/io/net/unix.rs (limited to 'src/libstd/io') diff --git a/src/libstd/io/net/mod.rs b/src/libstd/io/net/mod.rs index 54af83462ee..4bd998d5ed9 100644 --- a/src/libstd/io/net/mod.rs +++ b/src/libstd/io/net/mod.rs @@ -19,8 +19,7 @@ pub mod addrinfo; pub mod tcp; pub mod udp; pub mod ip; -// FIXME(#12093) - this should not be called unix -pub mod unix; +pub mod pipe; fn to_rtio(ip: IpAddr) -> rtio::IpAddr { match ip { diff --git a/src/libstd/io/net/pipe.rs b/src/libstd/io/net/pipe.rs new file mode 100644 index 00000000000..bdd58b07d86 --- /dev/null +++ b/src/libstd/io/net/pipe.rs @@ -0,0 +1,802 @@ +// Copyright 2013 The Rust Project Developers. See the COPYRIGHT +// file at the top-level directory of this distribution and at +// http://rust-lang.org/COPYRIGHT. +// +// Licensed under the Apache License, Version 2.0 or the MIT license +// , at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +/*! + +Named pipes + +This module contains the ability to communicate over named pipes with +synchronous I/O. On windows, this corresponds to talking over a Named Pipe, +while on Unix it corresponds to UNIX domain sockets. + +These pipes are similar to TCP in the sense that you can have both a stream to a +server and a server itself. The server provided accepts other `UnixStream` +instances as clients. + +*/ + +#![allow(missing_doc)] + +use prelude::*; + +use io::{Listener, Acceptor, IoResult, IoError, TimedOut, standard_error}; +use rt::rtio::{IoFactory, LocalIo, RtioUnixListener}; +use rt::rtio::{RtioUnixAcceptor, RtioPipe}; +use time::Duration; + +/// A stream which communicates over a named pipe. +pub struct UnixStream { + obj: Box, +} + +impl UnixStream { + /// Connect to a pipe named by `path`. This will attempt to open a + /// connection to the underlying socket. + /// + /// The returned stream will be closed when the object falls out of scope. + /// + /// # Example + /// + /// ```rust + /// # #![allow(unused_must_use)] + /// use std::io::net::pipe::UnixStream; + /// + /// let server = Path::new("path/to/my/socket"); + /// let mut stream = UnixStream::connect(&server); + /// stream.write([1, 2, 3]); + /// ``` + pub fn connect(path: &P) -> IoResult { + LocalIo::maybe_raise(|io| { + io.unix_connect(&path.to_c_str(), None).map(|p| UnixStream { obj: p }) + }).map_err(IoError::from_rtio_error) + } + + /// Connect to a pipe named by `path`, timing out if the specified number of + /// milliseconds. + /// + /// This function is similar to `connect`, except that if `timeout` + /// elapses the function will return an error of kind `TimedOut`. + /// + /// If a `timeout` with zero or negative duration is specified then + /// the function returns `Err`, with the error kind set to `TimedOut`. + #[experimental = "the timeout argument is likely to change types"] + pub fn connect_timeout(path: &P, + timeout: Duration) -> IoResult { + if timeout <= Duration::milliseconds(0) { + return Err(standard_error(TimedOut)); + } + + LocalIo::maybe_raise(|io| { + let s = io.unix_connect(&path.to_c_str(), Some(timeout.num_milliseconds() as u64)); + s.map(|p| UnixStream { obj: p }) + }).map_err(IoError::from_rtio_error) + } + + + /// Closes the reading half of this connection. + /// + /// This method will close the reading portion of this connection, causing + /// all pending and future reads to immediately return with an error. + /// + /// Note that this method affects all cloned handles associated with this + /// stream, not just this one handle. + pub fn close_read(&mut self) -> IoResult<()> { + self.obj.close_read().map_err(IoError::from_rtio_error) + } + + /// Closes the writing half of this connection. + /// + /// This method will close the writing portion of this connection, causing + /// all pending and future writes to immediately return with an error. + /// + /// Note that this method affects all cloned handles associated with this + /// stream, not just this one handle. + pub fn close_write(&mut self) -> IoResult<()> { + self.obj.close_write().map_err(IoError::from_rtio_error) + } + + /// Sets the read/write timeout for this socket. + /// + /// For more information, see `TcpStream::set_timeout` + #[experimental = "the timeout argument may change in type and value"] + pub fn set_timeout(&mut self, timeout_ms: Option) { + self.obj.set_timeout(timeout_ms) + } + + /// Sets the read timeout for this socket. + /// + /// For more information, see `TcpStream::set_timeout` + #[experimental = "the timeout argument may change in type and value"] + pub fn set_read_timeout(&mut self, timeout_ms: Option) { + self.obj.set_read_timeout(timeout_ms) + } + + /// Sets the write timeout for this socket. + /// + /// For more information, see `TcpStream::set_timeout` + #[experimental = "the timeout argument may change in type and value"] + pub fn set_write_timeout(&mut self, timeout_ms: Option) { + self.obj.set_write_timeout(timeout_ms) + } +} + +impl Clone for UnixStream { + fn clone(&self) -> UnixStream { + UnixStream { obj: self.obj.clone() } + } +} + +impl Reader for UnixStream { + fn read(&mut self, buf: &mut [u8]) -> IoResult { + self.obj.read(buf).map_err(IoError::from_rtio_error) + } +} + +impl Writer for UnixStream { + fn write(&mut self, buf: &[u8]) -> IoResult<()> { + self.obj.write(buf).map_err(IoError::from_rtio_error) + } +} + +/// A value that can listen for incoming named pipe connection requests. +pub struct UnixListener { + /// The internal, opaque runtime Unix listener. + obj: Box, +} + +impl UnixListener { + + /// Creates a new listener, ready to receive incoming connections on the + /// specified socket. The server will be named by `path`. + /// + /// This listener will be closed when it falls out of scope. + /// + /// # Example + /// + /// ``` + /// # fn main() {} + /// # fn foo() { + /// # #![allow(unused_must_use)] + /// use std::io::net::pipe::UnixListener; + /// use std::io::{Listener, Acceptor}; + /// + /// let server = Path::new("/path/to/my/socket"); + /// let stream = UnixListener::bind(&server); + /// for mut client in stream.listen().incoming() { + /// client.write([1, 2, 3, 4]); + /// } + /// # } + /// ``` + pub fn bind(path: &P) -> IoResult { + LocalIo::maybe_raise(|io| { + io.unix_bind(&path.to_c_str()).map(|s| UnixListener { obj: s }) + }).map_err(IoError::from_rtio_error) + } +} + +impl Listener for UnixListener { + fn listen(self) -> IoResult { + self.obj.listen().map(|obj| { + UnixAcceptor { obj: obj } + }).map_err(IoError::from_rtio_error) + } +} + +/// A value that can accept named pipe connections, returned from `listen()`. +pub struct UnixAcceptor { + /// The internal, opaque runtime Unix acceptor. + obj: Box, +} + +impl UnixAcceptor { + /// Sets a timeout for this acceptor, after which accept() will no longer + /// block indefinitely. + /// + /// The argument specified is the amount of time, in milliseconds, into the + /// future after which all invocations of accept() will not block (and any + /// pending invocation will return). A value of `None` will clear any + /// existing timeout. + /// + /// When using this method, it is likely necessary to reset the timeout as + /// appropriate, the timeout specified is specific to this object, not + /// specific to the next request. + #[experimental = "the name and arguments to this function are likely \ + to change"] + pub fn set_timeout(&mut self, timeout_ms: Option) { + self.obj.set_timeout(timeout_ms) + } + + /// Closes the accepting capabilities of this acceptor. + /// + /// This function has the same semantics as `TcpAcceptor::close_accept`, and + /// more information can be found in that documentation. + #[experimental] + pub fn close_accept(&mut self) -> IoResult<()> { + self.obj.close_accept().map_err(IoError::from_rtio_error) + } +} + +impl Acceptor for UnixAcceptor { + fn accept(&mut self) -> IoResult { + self.obj.accept().map(|s| { + UnixStream { obj: s } + }).map_err(IoError::from_rtio_error) + } +} + +impl Clone for UnixAcceptor { + /// Creates a new handle to this unix acceptor, allowing for simultaneous + /// accepts. + /// + /// The underlying unix acceptor will not be closed until all handles to the + /// acceptor have been deallocated. Incoming connections will be received on + /// at most once acceptor, the same connection will not be accepted twice. + /// + /// The `close_accept` method will shut down *all* acceptors cloned from the + /// same original acceptor, whereas the `set_timeout` method only affects + /// the selector that it is called on. + /// + /// This function is useful for creating a handle to invoke `close_accept` + /// on to wake up any other task blocked in `accept`. + fn clone(&self) -> UnixAcceptor { + UnixAcceptor { obj: self.obj.clone() } + } +} + +#[cfg(test)] +#[allow(experimental)] +mod tests { + use prelude::*; + use super::*; + use io::*; + use io::test::*; + + pub fn smalltest(server: proc(UnixStream):Send, client: proc(UnixStream):Send) { + let path1 = next_test_unix(); + let path2 = path1.clone(); + + let mut acceptor = UnixListener::bind(&path1).listen(); + + spawn(proc() { + match UnixStream::connect(&path2) { + Ok(c) => client(c), + Err(e) => fail!("failed connect: {}", e), + } + }); + + match acceptor.accept() { + Ok(c) => server(c), + Err(e) => fail!("failed accept: {}", e), + } + } + + iotest!(fn bind_error() { + let path = "path/to/nowhere"; + match UnixListener::bind(&path) { + Ok(..) => fail!(), + Err(e) => { + assert!(e.kind == PermissionDenied || e.kind == FileNotFound || + e.kind == InvalidInput); + } + } + }) + + iotest!(fn connect_error() { + let path = if cfg!(windows) { + r"\\.\pipe\this_should_not_exist_ever" + } else { + "path/to/nowhere" + }; + match UnixStream::connect(&path) { + Ok(..) => fail!(), + Err(e) => { + assert!(e.kind == FileNotFound || e.kind == OtherIoError); + } + } + }) + + iotest!(fn smoke() { + smalltest(proc(mut server) { + let mut buf = [0]; + server.read(buf).unwrap(); + assert!(buf[0] == 99); + }, proc(mut client) { + client.write([99]).unwrap(); + }) + }) + + iotest!(fn read_eof() { + smalltest(proc(mut server) { + let mut buf = [0]; + assert!(server.read(buf).is_err()); + assert!(server.read(buf).is_err()); + }, proc(_client) { + // drop the client + }) + } #[ignore(cfg(windows))]) // FIXME(#12516) + + iotest!(fn write_begone() { + smalltest(proc(mut server) { + let buf = [0]; + loop { + match server.write(buf) { + Ok(..) => {} + Err(e) => { + assert!(e.kind == BrokenPipe || + e.kind == NotConnected || + e.kind == ConnectionReset, + "unknown error {:?}", e); + break; + } + } + } + }, proc(_client) { + // drop the client + }) + }) + + iotest!(fn accept_lots() { + let times = 10; + let path1 = next_test_unix(); + let path2 = path1.clone(); + + let mut acceptor = match UnixListener::bind(&path1).listen() { + Ok(a) => a, + Err(e) => fail!("failed listen: {}", e), + }; + + spawn(proc() { + for _ in range(0u, times) { + let mut stream = UnixStream::connect(&path2); + match stream.write([100]) { + Ok(..) => {} + Err(e) => fail!("failed write: {}", e) + } + } + }); + + for _ in range(0, times) { + let mut client = acceptor.accept(); + let mut buf = [0]; + match client.read(buf) { + Ok(..) => {} + Err(e) => fail!("failed read/accept: {}", e), + } + assert_eq!(buf[0], 100); + } + }) + + #[cfg(unix)] + iotest!(fn path_exists() { + let path = next_test_unix(); + let _acceptor = UnixListener::bind(&path).listen(); + assert!(path.exists()); + }) + + iotest!(fn unix_clone_smoke() { + let addr = next_test_unix(); + let mut acceptor = UnixListener::bind(&addr).listen(); + + spawn(proc() { + let mut s = UnixStream::connect(&addr); + let mut buf = [0, 0]; + debug!("client reading"); + assert_eq!(s.read(buf), Ok(1)); + assert_eq!(buf[0], 1); + debug!("client writing"); + s.write([2]).unwrap(); + debug!("client dropping"); + }); + + let mut s1 = acceptor.accept().unwrap(); + let s2 = s1.clone(); + + let (tx1, rx1) = channel(); + let (tx2, rx2) = channel(); + spawn(proc() { + let mut s2 = s2; + rx1.recv(); + debug!("writer writing"); + s2.write([1]).unwrap(); + debug!("writer done"); + tx2.send(()); + }); + tx1.send(()); + let mut buf = [0, 0]; + debug!("reader reading"); + assert_eq!(s1.read(buf), Ok(1)); + debug!("reader done"); + rx2.recv(); + }) + + iotest!(fn unix_clone_two_read() { + let addr = next_test_unix(); + let mut acceptor = UnixListener::bind(&addr).listen(); + let (tx1, rx) = channel(); + let tx2 = tx1.clone(); + + spawn(proc() { + let mut s = UnixStream::connect(&addr); + s.write([1]).unwrap(); + rx.recv(); + s.write([2]).unwrap(); + rx.recv(); + }); + + let mut s1 = acceptor.accept().unwrap(); + let s2 = s1.clone(); + + let (done, rx) = channel(); + spawn(proc() { + let mut s2 = s2; + let mut buf = [0, 0]; + s2.read(buf).unwrap(); + tx2.send(()); + done.send(()); + }); + let mut buf = [0, 0]; + s1.read(buf).unwrap(); + tx1.send(()); + + rx.recv(); + }) + + iotest!(fn unix_clone_two_write() { + let addr = next_test_unix(); + let mut acceptor = UnixListener::bind(&addr).listen(); + + spawn(proc() { + let mut s = UnixStream::connect(&addr); + let mut buf = [0, 1]; + s.read(buf).unwrap(); + s.read(buf).unwrap(); + }); + + let mut s1 = acceptor.accept().unwrap(); + let s2 = s1.clone(); + + let (tx, rx) = channel(); + spawn(proc() { + let mut s2 = s2; + s2.write([1]).unwrap(); + tx.send(()); + }); + s1.write([2]).unwrap(); + + rx.recv(); + }) + + iotest!(fn drop_removes_listener_path() { + let path = next_test_unix(); + let l = UnixListener::bind(&path).unwrap(); + assert!(path.exists()); + drop(l); + assert!(!path.exists()); + } #[cfg(not(windows))]) + + iotest!(fn drop_removes_acceptor_path() { + let path = next_test_unix(); + let l = UnixListener::bind(&path).unwrap(); + assert!(path.exists()); + drop(l.listen().unwrap()); + assert!(!path.exists()); + } #[cfg(not(windows))]) + + iotest!(fn accept_timeout() { + let addr = next_test_unix(); + let mut a = UnixListener::bind(&addr).unwrap().listen().unwrap(); + + a.set_timeout(Some(10)); + + // Make sure we time out once and future invocations also time out + let err = a.accept().err().unwrap(); + assert_eq!(err.kind, TimedOut); + let err = a.accept().err().unwrap(); + assert_eq!(err.kind, TimedOut); + + // Also make sure that even though the timeout is expired that we will + // continue to receive any pending connections. + let (tx, rx) = channel(); + let addr2 = addr.clone(); + spawn(proc() { + tx.send(UnixStream::connect(&addr2).unwrap()); + }); + let l = rx.recv(); + for i in range(0u, 1001) { + match a.accept() { + Ok(..) => break, + Err(ref e) if e.kind == TimedOut => {} + Err(e) => fail!("error: {}", e), + } + ::task::deschedule(); + if i == 1000 { fail!("should have a pending connection") } + } + drop(l); + + // Unset the timeout and make sure that this always blocks. + a.set_timeout(None); + let addr2 = addr.clone(); + spawn(proc() { + drop(UnixStream::connect(&addr2).unwrap()); + }); + a.accept().unwrap(); + }) + + iotest!(fn connect_timeout_error() { + let addr = next_test_unix(); + assert!(UnixStream::connect_timeout(&addr, Duration::milliseconds(100)).is_err()); + }) + + iotest!(fn connect_timeout_success() { + let addr = next_test_unix(); + let _a = UnixListener::bind(&addr).unwrap().listen().unwrap(); + assert!(UnixStream::connect_timeout(&addr, Duration::milliseconds(100)).is_ok()); + }) + + iotest!(fn connect_timeout_zero() { + let addr = next_test_unix(); + let _a = UnixListener::bind(&addr).unwrap().listen().unwrap(); + assert!(UnixStream::connect_timeout(&addr, Duration::milliseconds(0)).is_err()); + }) + + iotest!(fn connect_timeout_negative() { + let addr = next_test_unix(); + let _a = UnixListener::bind(&addr).unwrap().listen().unwrap(); + assert!(UnixStream::connect_timeout(&addr, Duration::milliseconds(-1)).is_err()); + }) + + iotest!(fn close_readwrite_smoke() { + let addr = next_test_unix(); + let a = UnixListener::bind(&addr).listen().unwrap(); + let (_tx, rx) = channel::<()>(); + spawn(proc() { + let mut a = a; + let _s = a.accept().unwrap(); + let _ = rx.recv_opt(); + }); + + let mut b = [0]; + let mut s = UnixStream::connect(&addr).unwrap(); + let mut s2 = s.clone(); + + // closing should prevent reads/writes + s.close_write().unwrap(); + assert!(s.write([0]).is_err()); + s.close_read().unwrap(); + assert!(s.read(b).is_err()); + + // closing should affect previous handles + assert!(s2.write([0]).is_err()); + assert!(s2.read(b).is_err()); + + // closing should affect new handles + let mut s3 = s.clone(); + assert!(s3.write([0]).is_err()); + assert!(s3.read(b).is_err()); + + // make sure these don't die + let _ = s2.close_read(); + let _ = s2.close_write(); + let _ = s3.close_read(); + let _ = s3.close_write(); + }) + + iotest!(fn close_read_wakes_up() { + let addr = next_test_unix(); + let a = UnixListener::bind(&addr).listen().unwrap(); + let (_tx, rx) = channel::<()>(); + spawn(proc() { + let mut a = a; + let _s = a.accept().unwrap(); + let _ = rx.recv_opt(); + }); + + let mut s = UnixStream::connect(&addr).unwrap(); + let s2 = s.clone(); + let (tx, rx) = channel(); + spawn(proc() { + let mut s2 = s2; + assert!(s2.read([0]).is_err()); + tx.send(()); + }); + // 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(); + }) + + iotest!(fn readwrite_timeouts() { + let addr = next_test_unix(); + let mut a = UnixListener::bind(&addr).listen().unwrap(); + let (tx, rx) = channel::<()>(); + spawn(proc() { + let mut s = UnixStream::connect(&addr).unwrap(); + rx.recv(); + assert!(s.write([0]).is_ok()); + let _ = rx.recv_opt(); + }); + + let mut s = a.accept().unwrap(); + s.set_timeout(Some(20)); + assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); + assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); + + s.set_timeout(Some(20)); + for i in range(0u, 1001) { + match s.write([0, .. 128 * 1024]) { + Ok(()) | Err(IoError { kind: ShortWrite(..), .. }) => {}, + Err(IoError { kind: TimedOut, .. }) => break, + Err(e) => fail!("{}", e), + } + if i == 1000 { fail!("should have filled up?!"); } + } + + // I'm not sure as to why, but apparently the write on windows always + // succeeds after the previous timeout. Who knows? + if !cfg!(windows) { + assert_eq!(s.write([0]).err().unwrap().kind, TimedOut); + } + + tx.send(()); + s.set_timeout(None); + assert_eq!(s.read([0, 0]), Ok(1)); + }) + + iotest!(fn read_timeouts() { + let addr = next_test_unix(); + let mut a = UnixListener::bind(&addr).listen().unwrap(); + let (tx, rx) = channel::<()>(); + spawn(proc() { + let mut s = UnixStream::connect(&addr).unwrap(); + rx.recv(); + let mut amt = 0; + while amt < 100 * 128 * 1024 { + match s.read([0, ..128 * 1024]) { + Ok(n) => { amt += n; } + Err(e) => fail!("{}", e), + } + } + let _ = rx.recv_opt(); + }); + + let mut s = a.accept().unwrap(); + s.set_read_timeout(Some(20)); + assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); + assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); + + tx.send(()); + for _ in range(0u, 100) { + assert!(s.write([0, ..128 * 1024]).is_ok()); + } + }) + + iotest!(fn write_timeouts() { + let addr = next_test_unix(); + let mut a = UnixListener::bind(&addr).listen().unwrap(); + let (tx, rx) = channel::<()>(); + spawn(proc() { + let mut s = UnixStream::connect(&addr).unwrap(); + rx.recv(); + assert!(s.write([0]).is_ok()); + let _ = rx.recv_opt(); + }); + + let mut s = a.accept().unwrap(); + s.set_write_timeout(Some(20)); + for i in range(0u, 1001) { + match s.write([0, .. 128 * 1024]) { + Ok(()) | Err(IoError { kind: ShortWrite(..), .. }) => {}, + Err(IoError { kind: TimedOut, .. }) => break, + Err(e) => fail!("{}", e), + } + if i == 1000 { fail!("should have filled up?!"); } + } + + tx.send(()); + assert!(s.read([0]).is_ok()); + }) + + iotest!(fn timeout_concurrent_read() { + let addr = next_test_unix(); + let mut a = UnixListener::bind(&addr).listen().unwrap(); + let (tx, rx) = channel::<()>(); + spawn(proc() { + let mut s = UnixStream::connect(&addr).unwrap(); + rx.recv(); + assert!(s.write([0]).is_ok()); + let _ = rx.recv_opt(); + }); + + let mut s = a.accept().unwrap(); + let s2 = s.clone(); + let (tx2, rx2) = channel(); + spawn(proc() { + let mut s2 = s2; + assert!(s2.read([0]).is_ok()); + tx2.send(()); + }); + + s.set_read_timeout(Some(20)); + assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); + tx.send(()); + + rx2.recv(); + }) + + #[cfg(not(windows))] + iotest!(fn clone_accept_smoke() { + let addr = next_test_unix(); + let l = UnixListener::bind(&addr); + let mut a = l.listen().unwrap(); + let mut a2 = a.clone(); + + let addr2 = addr.clone(); + spawn(proc() { + let _ = UnixStream::connect(&addr2); + }); + spawn(proc() { + let _ = UnixStream::connect(&addr); + }); + + assert!(a.accept().is_ok()); + drop(a); + assert!(a2.accept().is_ok()); + }) + + iotest!(fn clone_accept_concurrent() { + let addr = next_test_unix(); + let l = UnixListener::bind(&addr); + let a = l.listen().unwrap(); + let a2 = a.clone(); + + let (tx, rx) = channel(); + let tx2 = tx.clone(); + + spawn(proc() { let mut a = a; tx.send(a.accept()) }); + spawn(proc() { let mut a = a2; tx2.send(a.accept()) }); + + let addr2 = addr.clone(); + spawn(proc() { + let _ = UnixStream::connect(&addr2); + }); + spawn(proc() { + let _ = UnixStream::connect(&addr); + }); + + assert!(rx.recv().is_ok()); + assert!(rx.recv().is_ok()); + }) + + iotest!(fn close_accept_smoke() { + let addr = next_test_unix(); + let l = UnixListener::bind(&addr); + let mut a = l.listen().unwrap(); + + a.close_accept().unwrap(); + assert_eq!(a.accept().err().unwrap().kind, EndOfFile); + }) + + iotest!(fn close_accept_concurrent() { + let addr = next_test_unix(); + let l = UnixListener::bind(&addr); + let a = l.listen().unwrap(); + let mut a2 = a.clone(); + + let (tx, rx) = channel(); + spawn(proc() { + let mut a = a; + tx.send(a.accept()); + }); + a2.close_accept().unwrap(); + + assert_eq!(rx.recv().err().unwrap().kind, EndOfFile); + }) +} diff --git a/src/libstd/io/net/unix.rs b/src/libstd/io/net/unix.rs deleted file mode 100644 index ea851d44531..00000000000 --- a/src/libstd/io/net/unix.rs +++ /dev/null @@ -1,802 +0,0 @@ -// Copyright 2013 The Rust Project Developers. See the COPYRIGHT -// file at the top-level directory of this distribution and at -// http://rust-lang.org/COPYRIGHT. -// -// Licensed under the Apache License, Version 2.0 or the MIT license -// , at your -// option. This file may not be copied, modified, or distributed -// except according to those terms. - -/*! - -Named pipes - -This module contains the ability to communicate over named pipes with -synchronous I/O. On windows, this corresponds to talking over a Named Pipe, -while on Unix it corresponds to UNIX domain sockets. - -These pipes are similar to TCP in the sense that you can have both a stream to a -server and a server itself. The server provided accepts other `UnixStream` -instances as clients. - -*/ - -#![allow(missing_doc)] - -use prelude::*; - -use io::{Listener, Acceptor, IoResult, IoError, TimedOut, standard_error}; -use rt::rtio::{IoFactory, LocalIo, RtioUnixListener}; -use rt::rtio::{RtioUnixAcceptor, RtioPipe}; -use time::Duration; - -/// A stream which communicates over a named pipe. -pub struct UnixStream { - obj: Box, -} - -impl UnixStream { - /// Connect to a pipe named by `path`. This will attempt to open a - /// connection to the underlying socket. - /// - /// The returned stream will be closed when the object falls out of scope. - /// - /// # Example - /// - /// ```rust - /// # #![allow(unused_must_use)] - /// use std::io::net::unix::UnixStream; - /// - /// let server = Path::new("path/to/my/socket"); - /// let mut stream = UnixStream::connect(&server); - /// stream.write([1, 2, 3]); - /// ``` - pub fn connect(path: &P) -> IoResult { - LocalIo::maybe_raise(|io| { - io.unix_connect(&path.to_c_str(), None).map(|p| UnixStream { obj: p }) - }).map_err(IoError::from_rtio_error) - } - - /// Connect to a pipe named by `path`, timing out if the specified number of - /// milliseconds. - /// - /// This function is similar to `connect`, except that if `timeout` - /// elapses the function will return an error of kind `TimedOut`. - /// - /// If a `timeout` with zero or negative duration is specified then - /// the function returns `Err`, with the error kind set to `TimedOut`. - #[experimental = "the timeout argument is likely to change types"] - pub fn connect_timeout(path: &P, - timeout: Duration) -> IoResult { - if timeout <= Duration::milliseconds(0) { - return Err(standard_error(TimedOut)); - } - - LocalIo::maybe_raise(|io| { - let s = io.unix_connect(&path.to_c_str(), Some(timeout.num_milliseconds() as u64)); - s.map(|p| UnixStream { obj: p }) - }).map_err(IoError::from_rtio_error) - } - - - /// Closes the reading half of this connection. - /// - /// This method will close the reading portion of this connection, causing - /// all pending and future reads to immediately return with an error. - /// - /// Note that this method affects all cloned handles associated with this - /// stream, not just this one handle. - pub fn close_read(&mut self) -> IoResult<()> { - self.obj.close_read().map_err(IoError::from_rtio_error) - } - - /// Closes the writing half of this connection. - /// - /// This method will close the writing portion of this connection, causing - /// all pending and future writes to immediately return with an error. - /// - /// Note that this method affects all cloned handles associated with this - /// stream, not just this one handle. - pub fn close_write(&mut self) -> IoResult<()> { - self.obj.close_write().map_err(IoError::from_rtio_error) - } - - /// Sets the read/write timeout for this socket. - /// - /// For more information, see `TcpStream::set_timeout` - #[experimental = "the timeout argument may change in type and value"] - pub fn set_timeout(&mut self, timeout_ms: Option) { - self.obj.set_timeout(timeout_ms) - } - - /// Sets the read timeout for this socket. - /// - /// For more information, see `TcpStream::set_timeout` - #[experimental = "the timeout argument may change in type and value"] - pub fn set_read_timeout(&mut self, timeout_ms: Option) { - self.obj.set_read_timeout(timeout_ms) - } - - /// Sets the write timeout for this socket. - /// - /// For more information, see `TcpStream::set_timeout` - #[experimental = "the timeout argument may change in type and value"] - pub fn set_write_timeout(&mut self, timeout_ms: Option) { - self.obj.set_write_timeout(timeout_ms) - } -} - -impl Clone for UnixStream { - fn clone(&self) -> UnixStream { - UnixStream { obj: self.obj.clone() } - } -} - -impl Reader for UnixStream { - fn read(&mut self, buf: &mut [u8]) -> IoResult { - self.obj.read(buf).map_err(IoError::from_rtio_error) - } -} - -impl Writer for UnixStream { - fn write(&mut self, buf: &[u8]) -> IoResult<()> { - self.obj.write(buf).map_err(IoError::from_rtio_error) - } -} - -/// A value that can listen for incoming named pipe connection requests. -pub struct UnixListener { - /// The internal, opaque runtime Unix listener. - obj: Box, -} - -impl UnixListener { - - /// Creates a new listener, ready to receive incoming connections on the - /// specified socket. The server will be named by `path`. - /// - /// This listener will be closed when it falls out of scope. - /// - /// # Example - /// - /// ``` - /// # fn main() {} - /// # fn foo() { - /// # #![allow(unused_must_use)] - /// use std::io::net::unix::UnixListener; - /// use std::io::{Listener, Acceptor}; - /// - /// let server = Path::new("/path/to/my/socket"); - /// let stream = UnixListener::bind(&server); - /// for mut client in stream.listen().incoming() { - /// client.write([1, 2, 3, 4]); - /// } - /// # } - /// ``` - pub fn bind(path: &P) -> IoResult { - LocalIo::maybe_raise(|io| { - io.unix_bind(&path.to_c_str()).map(|s| UnixListener { obj: s }) - }).map_err(IoError::from_rtio_error) - } -} - -impl Listener for UnixListener { - fn listen(self) -> IoResult { - self.obj.listen().map(|obj| { - UnixAcceptor { obj: obj } - }).map_err(IoError::from_rtio_error) - } -} - -/// A value that can accept named pipe connections, returned from `listen()`. -pub struct UnixAcceptor { - /// The internal, opaque runtime Unix acceptor. - obj: Box, -} - -impl UnixAcceptor { - /// Sets a timeout for this acceptor, after which accept() will no longer - /// block indefinitely. - /// - /// The argument specified is the amount of time, in milliseconds, into the - /// future after which all invocations of accept() will not block (and any - /// pending invocation will return). A value of `None` will clear any - /// existing timeout. - /// - /// When using this method, it is likely necessary to reset the timeout as - /// appropriate, the timeout specified is specific to this object, not - /// specific to the next request. - #[experimental = "the name and arguments to this function are likely \ - to change"] - pub fn set_timeout(&mut self, timeout_ms: Option) { - self.obj.set_timeout(timeout_ms) - } - - /// Closes the accepting capabilities of this acceptor. - /// - /// This function has the same semantics as `TcpAcceptor::close_accept`, and - /// more information can be found in that documentation. - #[experimental] - pub fn close_accept(&mut self) -> IoResult<()> { - self.obj.close_accept().map_err(IoError::from_rtio_error) - } -} - -impl Acceptor for UnixAcceptor { - fn accept(&mut self) -> IoResult { - self.obj.accept().map(|s| { - UnixStream { obj: s } - }).map_err(IoError::from_rtio_error) - } -} - -impl Clone for UnixAcceptor { - /// Creates a new handle to this unix acceptor, allowing for simultaneous - /// accepts. - /// - /// The underlying unix acceptor will not be closed until all handles to the - /// acceptor have been deallocated. Incoming connections will be received on - /// at most once acceptor, the same connection will not be accepted twice. - /// - /// The `close_accept` method will shut down *all* acceptors cloned from the - /// same original acceptor, whereas the `set_timeout` method only affects - /// the selector that it is called on. - /// - /// This function is useful for creating a handle to invoke `close_accept` - /// on to wake up any other task blocked in `accept`. - fn clone(&self) -> UnixAcceptor { - UnixAcceptor { obj: self.obj.clone() } - } -} - -#[cfg(test)] -#[allow(experimental)] -mod tests { - use prelude::*; - use super::*; - use io::*; - use io::test::*; - - pub fn smalltest(server: proc(UnixStream):Send, client: proc(UnixStream):Send) { - let path1 = next_test_unix(); - let path2 = path1.clone(); - - let mut acceptor = UnixListener::bind(&path1).listen(); - - spawn(proc() { - match UnixStream::connect(&path2) { - Ok(c) => client(c), - Err(e) => fail!("failed connect: {}", e), - } - }); - - match acceptor.accept() { - Ok(c) => server(c), - Err(e) => fail!("failed accept: {}", e), - } - } - - iotest!(fn bind_error() { - let path = "path/to/nowhere"; - match UnixListener::bind(&path) { - Ok(..) => fail!(), - Err(e) => { - assert!(e.kind == PermissionDenied || e.kind == FileNotFound || - e.kind == InvalidInput); - } - } - }) - - iotest!(fn connect_error() { - let path = if cfg!(windows) { - r"\\.\pipe\this_should_not_exist_ever" - } else { - "path/to/nowhere" - }; - match UnixStream::connect(&path) { - Ok(..) => fail!(), - Err(e) => { - assert!(e.kind == FileNotFound || e.kind == OtherIoError); - } - } - }) - - iotest!(fn smoke() { - smalltest(proc(mut server) { - let mut buf = [0]; - server.read(buf).unwrap(); - assert!(buf[0] == 99); - }, proc(mut client) { - client.write([99]).unwrap(); - }) - }) - - iotest!(fn read_eof() { - smalltest(proc(mut server) { - let mut buf = [0]; - assert!(server.read(buf).is_err()); - assert!(server.read(buf).is_err()); - }, proc(_client) { - // drop the client - }) - } #[ignore(cfg(windows))]) // FIXME(#12516) - - iotest!(fn write_begone() { - smalltest(proc(mut server) { - let buf = [0]; - loop { - match server.write(buf) { - Ok(..) => {} - Err(e) => { - assert!(e.kind == BrokenPipe || - e.kind == NotConnected || - e.kind == ConnectionReset, - "unknown error {:?}", e); - break; - } - } - } - }, proc(_client) { - // drop the client - }) - }) - - iotest!(fn accept_lots() { - let times = 10; - let path1 = next_test_unix(); - let path2 = path1.clone(); - - let mut acceptor = match UnixListener::bind(&path1).listen() { - Ok(a) => a, - Err(e) => fail!("failed listen: {}", e), - }; - - spawn(proc() { - for _ in range(0u, times) { - let mut stream = UnixStream::connect(&path2); - match stream.write([100]) { - Ok(..) => {} - Err(e) => fail!("failed write: {}", e) - } - } - }); - - for _ in range(0, times) { - let mut client = acceptor.accept(); - let mut buf = [0]; - match client.read(buf) { - Ok(..) => {} - Err(e) => fail!("failed read/accept: {}", e), - } - assert_eq!(buf[0], 100); - } - }) - - #[cfg(unix)] - iotest!(fn path_exists() { - let path = next_test_unix(); - let _acceptor = UnixListener::bind(&path).listen(); - assert!(path.exists()); - }) - - iotest!(fn unix_clone_smoke() { - let addr = next_test_unix(); - let mut acceptor = UnixListener::bind(&addr).listen(); - - spawn(proc() { - let mut s = UnixStream::connect(&addr); - let mut buf = [0, 0]; - debug!("client reading"); - assert_eq!(s.read(buf), Ok(1)); - assert_eq!(buf[0], 1); - debug!("client writing"); - s.write([2]).unwrap(); - debug!("client dropping"); - }); - - let mut s1 = acceptor.accept().unwrap(); - let s2 = s1.clone(); - - let (tx1, rx1) = channel(); - let (tx2, rx2) = channel(); - spawn(proc() { - let mut s2 = s2; - rx1.recv(); - debug!("writer writing"); - s2.write([1]).unwrap(); - debug!("writer done"); - tx2.send(()); - }); - tx1.send(()); - let mut buf = [0, 0]; - debug!("reader reading"); - assert_eq!(s1.read(buf), Ok(1)); - debug!("reader done"); - rx2.recv(); - }) - - iotest!(fn unix_clone_two_read() { - let addr = next_test_unix(); - let mut acceptor = UnixListener::bind(&addr).listen(); - let (tx1, rx) = channel(); - let tx2 = tx1.clone(); - - spawn(proc() { - let mut s = UnixStream::connect(&addr); - s.write([1]).unwrap(); - rx.recv(); - s.write([2]).unwrap(); - rx.recv(); - }); - - let mut s1 = acceptor.accept().unwrap(); - let s2 = s1.clone(); - - let (done, rx) = channel(); - spawn(proc() { - let mut s2 = s2; - let mut buf = [0, 0]; - s2.read(buf).unwrap(); - tx2.send(()); - done.send(()); - }); - let mut buf = [0, 0]; - s1.read(buf).unwrap(); - tx1.send(()); - - rx.recv(); - }) - - iotest!(fn unix_clone_two_write() { - let addr = next_test_unix(); - let mut acceptor = UnixListener::bind(&addr).listen(); - - spawn(proc() { - let mut s = UnixStream::connect(&addr); - let mut buf = [0, 1]; - s.read(buf).unwrap(); - s.read(buf).unwrap(); - }); - - let mut s1 = acceptor.accept().unwrap(); - let s2 = s1.clone(); - - let (tx, rx) = channel(); - spawn(proc() { - let mut s2 = s2; - s2.write([1]).unwrap(); - tx.send(()); - }); - s1.write([2]).unwrap(); - - rx.recv(); - }) - - iotest!(fn drop_removes_listener_path() { - let path = next_test_unix(); - let l = UnixListener::bind(&path).unwrap(); - assert!(path.exists()); - drop(l); - assert!(!path.exists()); - } #[cfg(not(windows))]) - - iotest!(fn drop_removes_acceptor_path() { - let path = next_test_unix(); - let l = UnixListener::bind(&path).unwrap(); - assert!(path.exists()); - drop(l.listen().unwrap()); - assert!(!path.exists()); - } #[cfg(not(windows))]) - - iotest!(fn accept_timeout() { - let addr = next_test_unix(); - let mut a = UnixListener::bind(&addr).unwrap().listen().unwrap(); - - a.set_timeout(Some(10)); - - // Make sure we time out once and future invocations also time out - let err = a.accept().err().unwrap(); - assert_eq!(err.kind, TimedOut); - let err = a.accept().err().unwrap(); - assert_eq!(err.kind, TimedOut); - - // Also make sure that even though the timeout is expired that we will - // continue to receive any pending connections. - let (tx, rx) = channel(); - let addr2 = addr.clone(); - spawn(proc() { - tx.send(UnixStream::connect(&addr2).unwrap()); - }); - let l = rx.recv(); - for i in range(0u, 1001) { - match a.accept() { - Ok(..) => break, - Err(ref e) if e.kind == TimedOut => {} - Err(e) => fail!("error: {}", e), - } - ::task::deschedule(); - if i == 1000 { fail!("should have a pending connection") } - } - drop(l); - - // Unset the timeout and make sure that this always blocks. - a.set_timeout(None); - let addr2 = addr.clone(); - spawn(proc() { - drop(UnixStream::connect(&addr2).unwrap()); - }); - a.accept().unwrap(); - }) - - iotest!(fn connect_timeout_error() { - let addr = next_test_unix(); - assert!(UnixStream::connect_timeout(&addr, Duration::milliseconds(100)).is_err()); - }) - - iotest!(fn connect_timeout_success() { - let addr = next_test_unix(); - let _a = UnixListener::bind(&addr).unwrap().listen().unwrap(); - assert!(UnixStream::connect_timeout(&addr, Duration::milliseconds(100)).is_ok()); - }) - - iotest!(fn connect_timeout_zero() { - let addr = next_test_unix(); - let _a = UnixListener::bind(&addr).unwrap().listen().unwrap(); - assert!(UnixStream::connect_timeout(&addr, Duration::milliseconds(0)).is_err()); - }) - - iotest!(fn connect_timeout_negative() { - let addr = next_test_unix(); - let _a = UnixListener::bind(&addr).unwrap().listen().unwrap(); - assert!(UnixStream::connect_timeout(&addr, Duration::milliseconds(-1)).is_err()); - }) - - iotest!(fn close_readwrite_smoke() { - let addr = next_test_unix(); - let a = UnixListener::bind(&addr).listen().unwrap(); - let (_tx, rx) = channel::<()>(); - spawn(proc() { - let mut a = a; - let _s = a.accept().unwrap(); - let _ = rx.recv_opt(); - }); - - let mut b = [0]; - let mut s = UnixStream::connect(&addr).unwrap(); - let mut s2 = s.clone(); - - // closing should prevent reads/writes - s.close_write().unwrap(); - assert!(s.write([0]).is_err()); - s.close_read().unwrap(); - assert!(s.read(b).is_err()); - - // closing should affect previous handles - assert!(s2.write([0]).is_err()); - assert!(s2.read(b).is_err()); - - // closing should affect new handles - let mut s3 = s.clone(); - assert!(s3.write([0]).is_err()); - assert!(s3.read(b).is_err()); - - // make sure these don't die - let _ = s2.close_read(); - let _ = s2.close_write(); - let _ = s3.close_read(); - let _ = s3.close_write(); - }) - - iotest!(fn close_read_wakes_up() { - let addr = next_test_unix(); - let a = UnixListener::bind(&addr).listen().unwrap(); - let (_tx, rx) = channel::<()>(); - spawn(proc() { - let mut a = a; - let _s = a.accept().unwrap(); - let _ = rx.recv_opt(); - }); - - let mut s = UnixStream::connect(&addr).unwrap(); - let s2 = s.clone(); - let (tx, rx) = channel(); - spawn(proc() { - let mut s2 = s2; - assert!(s2.read([0]).is_err()); - tx.send(()); - }); - // 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(); - }) - - iotest!(fn readwrite_timeouts() { - let addr = next_test_unix(); - let mut a = UnixListener::bind(&addr).listen().unwrap(); - let (tx, rx) = channel::<()>(); - spawn(proc() { - let mut s = UnixStream::connect(&addr).unwrap(); - rx.recv(); - assert!(s.write([0]).is_ok()); - let _ = rx.recv_opt(); - }); - - let mut s = a.accept().unwrap(); - s.set_timeout(Some(20)); - assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); - assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); - - s.set_timeout(Some(20)); - for i in range(0u, 1001) { - match s.write([0, .. 128 * 1024]) { - Ok(()) | Err(IoError { kind: ShortWrite(..), .. }) => {}, - Err(IoError { kind: TimedOut, .. }) => break, - Err(e) => fail!("{}", e), - } - if i == 1000 { fail!("should have filled up?!"); } - } - - // I'm not sure as to why, but apparently the write on windows always - // succeeds after the previous timeout. Who knows? - if !cfg!(windows) { - assert_eq!(s.write([0]).err().unwrap().kind, TimedOut); - } - - tx.send(()); - s.set_timeout(None); - assert_eq!(s.read([0, 0]), Ok(1)); - }) - - iotest!(fn read_timeouts() { - let addr = next_test_unix(); - let mut a = UnixListener::bind(&addr).listen().unwrap(); - let (tx, rx) = channel::<()>(); - spawn(proc() { - let mut s = UnixStream::connect(&addr).unwrap(); - rx.recv(); - let mut amt = 0; - while amt < 100 * 128 * 1024 { - match s.read([0, ..128 * 1024]) { - Ok(n) => { amt += n; } - Err(e) => fail!("{}", e), - } - } - let _ = rx.recv_opt(); - }); - - let mut s = a.accept().unwrap(); - s.set_read_timeout(Some(20)); - assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); - assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); - - tx.send(()); - for _ in range(0u, 100) { - assert!(s.write([0, ..128 * 1024]).is_ok()); - } - }) - - iotest!(fn write_timeouts() { - let addr = next_test_unix(); - let mut a = UnixListener::bind(&addr).listen().unwrap(); - let (tx, rx) = channel::<()>(); - spawn(proc() { - let mut s = UnixStream::connect(&addr).unwrap(); - rx.recv(); - assert!(s.write([0]).is_ok()); - let _ = rx.recv_opt(); - }); - - let mut s = a.accept().unwrap(); - s.set_write_timeout(Some(20)); - for i in range(0u, 1001) { - match s.write([0, .. 128 * 1024]) { - Ok(()) | Err(IoError { kind: ShortWrite(..), .. }) => {}, - Err(IoError { kind: TimedOut, .. }) => break, - Err(e) => fail!("{}", e), - } - if i == 1000 { fail!("should have filled up?!"); } - } - - tx.send(()); - assert!(s.read([0]).is_ok()); - }) - - iotest!(fn timeout_concurrent_read() { - let addr = next_test_unix(); - let mut a = UnixListener::bind(&addr).listen().unwrap(); - let (tx, rx) = channel::<()>(); - spawn(proc() { - let mut s = UnixStream::connect(&addr).unwrap(); - rx.recv(); - assert!(s.write([0]).is_ok()); - let _ = rx.recv_opt(); - }); - - let mut s = a.accept().unwrap(); - let s2 = s.clone(); - let (tx2, rx2) = channel(); - spawn(proc() { - let mut s2 = s2; - assert!(s2.read([0]).is_ok()); - tx2.send(()); - }); - - s.set_read_timeout(Some(20)); - assert_eq!(s.read([0]).err().unwrap().kind, TimedOut); - tx.send(()); - - rx2.recv(); - }) - - #[cfg(not(windows))] - iotest!(fn clone_accept_smoke() { - let addr = next_test_unix(); - let l = UnixListener::bind(&addr); - let mut a = l.listen().unwrap(); - let mut a2 = a.clone(); - - let addr2 = addr.clone(); - spawn(proc() { - let _ = UnixStream::connect(&addr2); - }); - spawn(proc() { - let _ = UnixStream::connect(&addr); - }); - - assert!(a.accept().is_ok()); - drop(a); - assert!(a2.accept().is_ok()); - }) - - iotest!(fn clone_accept_concurrent() { - let addr = next_test_unix(); - let l = UnixListener::bind(&addr); - let a = l.listen().unwrap(); - let a2 = a.clone(); - - let (tx, rx) = channel(); - let tx2 = tx.clone(); - - spawn(proc() { let mut a = a; tx.send(a.accept()) }); - spawn(proc() { let mut a = a2; tx2.send(a.accept()) }); - - let addr2 = addr.clone(); - spawn(proc() { - let _ = UnixStream::connect(&addr2); - }); - spawn(proc() { - let _ = UnixStream::connect(&addr); - }); - - assert!(rx.recv().is_ok()); - assert!(rx.recv().is_ok()); - }) - - iotest!(fn close_accept_smoke() { - let addr = next_test_unix(); - let l = UnixListener::bind(&addr); - let mut a = l.listen().unwrap(); - - a.close_accept().unwrap(); - assert_eq!(a.accept().err().unwrap().kind, EndOfFile); - }) - - iotest!(fn close_accept_concurrent() { - let addr = next_test_unix(); - let l = UnixListener::bind(&addr); - let a = l.listen().unwrap(); - let mut a2 = a.clone(); - - let (tx, rx) = channel(); - spawn(proc() { - let mut a = a; - tx.send(a.accept()); - }); - a2.close_accept().unwrap(); - - assert_eq!(rx.recv().err().unwrap().kind, EndOfFile); - }) -} diff --git a/src/libstd/io/test.rs b/src/libstd/io/test.rs index 769ad2a8468..6c94eb4af58 100644 --- a/src/libstd/io/test.rs +++ b/src/libstd/io/test.rs @@ -34,7 +34,7 @@ macro_rules! iotest ( use io::net::ip::*; use io::net::udp::*; #[cfg(unix)] - use io::net::unix::*; + use io::net::pipe::*; use io::timer::*; use io::process::*; use rt::running_on_valgrind; -- cgit 1.4.1-3-g733a5