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]

1 files changed, 42 insertions, 3 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;