std: Improve pipe() functionality

* os::pipe() now returns IoResult<os::Pipe>
* os::pipe() is now unsafe because it does not arrange for deallocation of file
  descriptors
* os::Pipe fields are renamed from input to reader and out to write.
* PipeStream::pair() has been added. This is a safe method to get a pair of
  pipes.
* Dealing with pipes in native process bindings have been improved to be more
  robust in the face of failure and intermittent errors. This converts a few
  fail!() situations to Err situations.

Closes #9458
cc #13538
Closes #14724
[breaking-change]

2 files changed, 76 insertions, 26 deletions

diff --git a/src/libstd/io/pipe.rs b/src/libstd/io/pipe.rs
index 6e2009545aa..84d388c1136 100644
--- a/src/libstd/io/pipe.rs
+++ b/src/libstd/io/pipe.rs
@@ -16,8 +16,10 @@
 #![allow(missing_doc)]
 
 use prelude::*;
+
 use io::{IoResult, IoError};
 use libc;
+use os;
 use owned::Box;
 use rt::rtio::{RtioPipe, LocalIo};
 
@@ -27,6 +29,11 @@ pub struct PipeStream {
     obj: Box<RtioPipe + Send>,
 }
 
+pub struct PipePair {
+    pub reader: PipeStream,
+    pub writer: PipeStream,
+}
+
 impl PipeStream {
     /// Consumes a file descriptor to return a pipe stream that will have
     /// synchronous, but non-blocking reads/writes. This is useful if the file
@@ -58,6 +65,38 @@ impl PipeStream {
     pub fn new(inner: Box<RtioPipe + Send>) -> PipeStream {
         PipeStream { obj: inner }
     }
+
+    /// Creates a pair of in-memory OS pipes for a unidirectional communication
+    /// stream.
+    ///
+    /// The structure returned contains a reader and writer I/O object. Data
+    /// written to the writer can be read from the reader.
+    ///
+    /// # Errors
+    ///
+    /// This function can fail to succeed if the underlying OS has run out of
+    /// available resources to allocate a new pipe.
+    pub fn pair() -> IoResult<PipePair> {
+        struct Closer { fd: libc::c_int }
+
+        let os::Pipe { reader, writer } = try!(unsafe { os::pipe() });
+        let mut reader = Closer { fd: reader };
+        let mut writer = Closer { fd: writer };
+
+        let io_reader = try!(PipeStream::open(reader.fd));
+        reader.fd = -1;
+        let io_writer = try!(PipeStream::open(writer.fd));
+        writer.fd = -1;
+        return Ok(PipePair { reader: io_reader, writer: io_writer });
+
+        impl Drop for Closer {
+            fn drop(&mut self) {
+                if self.fd != -1 {
+                    let _ = unsafe { libc::close(self.fd) };
+                }
+            }
+        }
+    }
 }
 
 impl Clone for PipeStream {
@@ -84,9 +123,9 @@ mod test {
         use os;
         use io::pipe::PipeStream;
 
-        let os::Pipe { input, out } = os::pipe();
-        let out = PipeStream::open(out);
-        let mut input = PipeStream::open(input);
+        let os::Pipe { reader, writer } = unsafe { os::pipe().unwrap() };
+        let out = PipeStream::open(writer);
+        let mut input = PipeStream::open(reader);
         let (tx, rx) = channel();
         spawn(proc() {
             let mut out = out;
diff --git a/src/libstd/os.rs b/src/libstd/os.rs
index f6b1c04dd34..0747e7ccbe3 100644
--- a/src/libstd/os.rs
+++ b/src/libstd/os.rs
@@ -32,6 +32,7 @@
 use clone::Clone;
 use collections::Collection;
 use fmt;
+use io::{IoResult, IoError};
 use iter::Iterator;
 use libc::{c_void, c_int};
 use libc;
@@ -513,40 +514,50 @@ pub fn split_paths<T: BytesContainer>(unparsed: T) -> Vec<Path> {
 pub struct Pipe {
     /// A file descriptor representing the reading end of the pipe. Data written
     /// on the `out` file descriptor can be read from this file descriptor.
-    pub input: c_int,
+    pub reader: c_int,
     /// A file descriptor representing the write end of the pipe. Data written
     /// to this file descriptor can be read from the `input` file descriptor.
-    pub out: c_int,
+    pub writer: c_int,
 }
 
-/// Creates a new low-level OS in-memory pipe represented as a Pipe struct.
-#[cfg(unix)]
-pub fn pipe() -> Pipe {
-    unsafe {
-        let mut fds = Pipe {input: 0,
-                            out: 0};
-        assert_eq!(libc::pipe(&mut fds.input), 0);
-        return Pipe {input: fds.input, out: fds.out};
+/// Creates a new low-level OS in-memory pipe.
+///
+/// This function can fail to succeed if there are no more resources available
+/// to allocate a pipe.
+///
+/// This function is also unsafe as there is no destructor associated with the
+/// `Pipe` structure will return. If it is not arranged for the returned file
+/// descriptors to be closed, the file descriptors will leak. For safe handling
+/// of this scenario, use `std::io::PipeStream` instead.
+pub unsafe fn pipe() -> IoResult<Pipe> {
+    return _pipe();
+
+    #[cfg(unix)]
+    unsafe fn _pipe() -> IoResult<Pipe> {
+        let mut fds = [0, ..2];
+        match libc::pipe(fds.as_mut_ptr()) {
+            0 => Ok(Pipe { reader: fds[0], writer: fds[1] }),
+            _ => Err(IoError::last_error()),
+        }
     }
-}
 
-/// Creates a new low-level OS in-memory pipe represented as a Pipe struct.
-#[cfg(windows)]
-pub fn pipe() -> Pipe {
-    unsafe {
+    #[cfg(windows)]
+    unsafe fn _pipe() -> IoResult<Pipe> {
         // Windows pipes work subtly differently than unix pipes, and their
         // inheritance has to be handled in a different way that I do not
         // fully understand. Here we explicitly make the pipe non-inheritable,
         // which means to pass it to a subprocess they need to be duplicated
         // first, as in std::run.
-        let mut fds = Pipe {input: 0,
-                    out: 0};
-        let res = libc::pipe(&mut fds.input, 1024 as ::libc::c_uint,
-                             (libc::O_BINARY | libc::O_NOINHERIT) as c_int);
-        assert_eq!(res, 0);
-        assert!((fds.input != -1 && fds.input != 0 ));
-        assert!((fds.out != -1 && fds.input != 0));
-        return Pipe {input: fds.input, out: fds.out};
+        let mut fds = [0, ..2];
+        match libc::pipe(fds.as_mut_ptr(), 1024 as ::libc::c_uint,
+                         (libc::O_BINARY | libc::O_NOINHERIT) as c_int) {
+            0 => {
+                assert!(fds[0] != -1 && fds[0] != 0);
+                assert!(fds[1] != -1 && fds[1] != 0);
+                Ok(Pipe { reader: fds[0], writer: fds[1] })
+            }
+            _ => Err(IoError::last_error()),
+        }
     }
 }