Auto merge of #31618 - alexcrichton:no-thread-spawns, r=brson

Optimize some functions in std::process

* Be sure that `read_to_end` gets directed towards `read_to_end_uninitialized` for all handles
* When spawning a child that guaranteed doesn't need a stdin, don't actually create a stdin pipe for that process, instead just redirect it to /dev/null
* When calling `wait_with_output`, don't spawn threads to read out the pipes of the child. Instead drain all pipes on the calling thread and *then* wait on the process.

Functionally, it is intended that nothing changes as part of this PR

---

Note that this was the same as #31613, and even after that it turned out that fixing Windows was easier than I thought! To copy a comment from over there:

> As some rationale for this as well, it's always bothered me that we've spawned threads in the standard library for this (seems a bit overkill), and I've also been curious lately as to our why our build times for Windows are so much higher than Unix (on the buildbots we have). I have done basically 0 investigation into why, but I figured it can't help to try to optimize Command::output which I believe is called quite a few times during the test suite.

Author: bors

src/liblibc

@@ -22,7 +22,6 @@ use ffi::OsString;
 use io::{self, SeekFrom, Seek, Read, Write};
 use path::{Path, PathBuf};
 use sys::fs as fs_imp;
-use sys_common::io::read_to_end_uninitialized;
 use sys_common::{AsInnerMut, FromInner, AsInner, IntoInner};
 use vec::Vec;
 use time::SystemTime;
@@ -351,7 +350,7 @@ impl Read for File {
         self.inner.read(buf)
     }
     fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
-        unsafe { read_to_end_uninitialized(self, buf) }
+        self.inner.read_to_end(buf)
     }
 }
 #[stable(feature = "rust1", since = "1.0.0")]
@@ -372,6 +371,9 @@ impl<'a> Read for &'a File {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         self.inner.read(buf)
     }
+    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.inner.read_to_end(buf)
+    }
 }
 #[stable(feature = "rust1", since = "1.0.0")]
 impl<'a> Write for &'a File {

src/libstd/fs.rs

@@ -18,7 +18,6 @@ use io::lazy::Lazy;
 use io::{self, BufReader, LineWriter};
 use sync::{Arc, Mutex, MutexGuard};
 use sys::stdio;
-use sys_common::io::{read_to_end_uninitialized};
 use sys_common::remutex::{ReentrantMutex, ReentrantMutexGuard};
 use thread::LocalKeyState;
 
@@ -78,6 +77,9 @@ fn stderr_raw() -> io::Result<StderrRaw> { stdio::Stderr::new().map(StderrRaw) }
 
 impl Read for StdinRaw {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { self.0.read(buf) }
+    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.0.read_to_end(buf)
+    }
 }
 impl Write for StdoutRaw {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> { self.0.write(buf) }
@@ -116,6 +118,12 @@ impl<R: io::Read> io::Read for Maybe<R> {
             Maybe::Fake => Ok(0)
         }
     }
+    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        match *self {
+            Maybe::Real(ref mut r) => handle_ebadf(r.read_to_end(buf), 0),
+            Maybe::Fake => Ok(0)
+        }
+    }
 }
 
 fn handle_ebadf<T>(r: io::Result<T>, default: T) -> io::Result<T> {
@@ -294,7 +302,7 @@ impl<'a> Read for StdinLock<'a> {
         self.inner.read(buf)
     }
     fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
-        unsafe { read_to_end_uninitialized(self, buf) }
+        self.inner.read_to_end(buf)
     }
 }
 

src/libstd/io/stdio.rs

@@ -14,7 +14,6 @@ use io::prelude::*;
 use fmt;
 use io;
 use net::{ToSocketAddrs, SocketAddr, Shutdown};
-use sys_common::io::read_to_end_uninitialized;
 use sys_common::net as net_imp;
 use sys_common::{AsInner, FromInner, IntoInner};
 use time::Duration;
@@ -269,7 +268,7 @@ impl TcpStream {
 impl Read for TcpStream {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { self.0.read(buf) }
     fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
-        unsafe { read_to_end_uninitialized(self, buf) }
+        self.0.read_to_end(buf)
     }
 }
 #[stable(feature = "rust1", since = "1.0.0")]
@@ -281,7 +280,7 @@ impl Write for TcpStream {
 impl<'a> Read for &'a TcpStream {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { self.0.read(buf) }
     fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
-        unsafe { read_to_end_uninitialized(self, buf) }
+        self.0.read_to_end(buf)
     }
 }
 #[stable(feature = "rust1", since = "1.0.0")]

src/libstd/net/tcp.rs

@@ -20,10 +20,9 @@ use fmt;
 use io;
 use path::Path;
 use str;
-use sys::pipe::AnonPipe;
+use sys::pipe::{read2, AnonPipe};
 use sys::process as imp;
 use sys_common::{AsInner, AsInnerMut, FromInner, IntoInner};
-use thread::{self, JoinHandle};
 
 /// Representation of a running or exited child process.
 ///
@@ -134,6 +133,9 @@ impl Read for ChildStdout {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         self.inner.read(buf)
     }
+    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.inner.read_to_end(buf)
+    }
 }
 
 impl AsInner<AnonPipe> for ChildStdout {
@@ -161,6 +163,9 @@ impl Read for ChildStderr {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         self.inner.read(buf)
     }
+    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.inner.read_to_end(buf)
+    }
 }
 
 impl AsInner<AnonPipe> for ChildStderr {
@@ -289,7 +294,7 @@ impl Command {
     /// By default, stdin, stdout and stderr are inherited from the parent.
     #[stable(feature = "process", since = "1.0.0")]
     pub fn spawn(&mut self) -> io::Result<Child> {
-        self.inner.spawn(imp::Stdio::Inherit).map(Child::from_inner)
+        self.inner.spawn(imp::Stdio::Inherit, true).map(Child::from_inner)
     }
 
     /// Executes the command as a child process, waiting for it to finish and
@@ -312,7 +317,7 @@ impl Command {
     /// ```
     #[stable(feature = "process", since = "1.0.0")]
     pub fn output(&mut self) -> io::Result<Output> {
-        self.inner.spawn(imp::Stdio::MakePipe).map(Child::from_inner)
+        self.inner.spawn(imp::Stdio::MakePipe, false).map(Child::from_inner)
             .and_then(|p| p.wait_with_output())
     }
 
@@ -334,7 +339,8 @@ impl Command {
     /// ```
     #[stable(feature = "process", since = "1.0.0")]
     pub fn status(&mut self) -> io::Result<ExitStatus> {
-        self.spawn().and_then(|mut p| p.wait())
+        self.inner.spawn(imp::Stdio::Inherit, false).map(Child::from_inner)
+                  .and_then(|mut p| p.wait())
     }
 }
 
@@ -496,24 +502,29 @@ impl Child {
     #[stable(feature = "process", since = "1.0.0")]
     pub fn wait_with_output(mut self) -> io::Result<Output> {
         drop(self.stdin.take());
-        fn read<R>(mut input: R) -> JoinHandle<io::Result<Vec<u8>>>
-            where R: Read + Send + 'static
-        {
-            thread::spawn(move || {
-                let mut ret = Vec::new();
-                input.read_to_end(&mut ret).map(|_| ret)
-            })
+
+        let (mut stdout, mut stderr) = (Vec::new(), Vec::new());
+        match (self.stdout.take(), self.stderr.take()) {
+            (None, None) => {}
+            (Some(mut out), None) => {
+                let res = out.read_to_end(&mut stdout);
+                res.unwrap();
+            }
+            (None, Some(mut err)) => {
+                let res = err.read_to_end(&mut stderr);
+                res.unwrap();
+            }
+            (Some(out), Some(err)) => {
+                let res = read2(out.inner, &mut stdout, err.inner, &mut stderr);
+                res.unwrap();
+            }
         }
-        let stdout = self.stdout.take().map(read);
-        let stderr = self.stderr.take().map(read);
-        let status = try!(self.wait());
-        let stdout = stdout.and_then(|t| t.join().unwrap().ok());
-        let stderr = stderr.and_then(|t| t.join().unwrap().ok());
 
+        let status = try!(self.wait());
         Ok(Output {
             status: status,
-            stdout: stdout.unwrap_or(Vec::new()),
-            stderr: stderr.unwrap_or(Vec::new()),
+            stdout: stdout,
+            stderr: stderr,
         })
     }
 }

src/libstd/process.rs

@@ -225,6 +225,10 @@ impl TcpStream {
         self.inner.read(buf)
     }
 
+    pub fn read_to_end(&self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.inner.read_to_end(buf)
+    }
+
     pub fn write(&self, buf: &[u8]) -> io::Result<usize> {
         let len = cmp::min(buf.len(), <wrlen_t>::max_value() as usize) as wrlen_t;
         let ret = try!(cvt(unsafe {

src/libstd/sys/common/net.rs

@@ -8,12 +8,17 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use io;
+#![unstable(reason = "not public", issue = "0", feature = "fd")]
+
+use prelude::v1::*;
+
+use io::{self, Read};
 use libc::{self, c_int, size_t, c_void};
 use mem;
+use sync::atomic::{AtomicBool, Ordering};
 use sys::cvt;
 use sys_common::AsInner;
-use sync::atomic::{AtomicBool, Ordering};
+use sys_common::io::read_to_end_uninitialized;
 
 pub struct FileDesc {
     fd: c_int,
@@ -42,6 +47,11 @@ impl FileDesc {
         Ok(ret as usize)
     }
 
+    pub fn read_to_end(&self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        let mut me = self;
+        (&mut me).read_to_end(buf)
+    }
+
     pub fn write(&self, buf: &[u8]) -> io::Result<usize> {
         let ret = try!(cvt(unsafe {
             libc::write(self.fd,
@@ -67,6 +77,20 @@ impl FileDesc {
         }
     }
 
+    pub fn set_nonblocking(&self, nonblocking: bool) {
+        unsafe {
+            let previous = libc::fcntl(self.fd, libc::F_GETFL);
+            debug_assert!(previous != -1);
+            let new = if nonblocking {
+                previous | libc::O_NONBLOCK
+            } else {
+                previous & !libc::O_NONBLOCK
+            };
+            let ret = libc::fcntl(self.fd, libc::F_SETFL, new);
+            debug_assert!(ret != -1);
+        }
+    }
+
     pub fn duplicate(&self) -> io::Result<FileDesc> {
         // We want to atomically duplicate this file descriptor and set the
         // CLOEXEC flag, and currently that's done via F_DUPFD_CLOEXEC. This
@@ -118,6 +142,16 @@ impl FileDesc {
     }
 }
 
+impl<'a> Read for &'a FileDesc {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        (**self).read(buf)
+    }
+
+    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        unsafe { read_to_end_uninitialized(self, buf) }
+    }
+}
+
 impl AsInner<c_int> for FileDesc {
     fn as_inner(&self) -> &c_int { &self.fd }
 }

src/libstd/sys/unix/fd.rs

@@ -486,6 +486,10 @@ impl File {
         self.0.read(buf)
     }
 
+    pub fn read_to_end(&self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.0.read_to_end(buf)
+    }
+
     pub fn write(&self, buf: &[u8]) -> io::Result<usize> {
         self.0.write(buf)
     }

src/libstd/sys/unix/fs.rs

@@ -116,6 +116,10 @@ impl Socket {
         self.0.read(buf)
     }
 
+    pub fn read_to_end(&self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.0.read_to_end(buf)
+    }
+
     pub fn set_timeout(&self, dur: Option<Duration>, kind: libc::c_int) -> io::Result<()> {
         let timeout = match dur {
             Some(dur) => {

src/libstd/sys/unix/net.rs

@@ -8,8 +8,12 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use prelude::v1::*;
+
+use cmp;
 use io;
 use libc::{self, c_int};
+use mem;
 use sys::cvt_r;
 use sys::fd::FileDesc;
 
@@ -57,10 +61,65 @@ impl AnonPipe {
         self.0.read(buf)
     }
 
+    pub fn read_to_end(&self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.0.read_to_end(buf)
+    }
+
     pub fn write(&self, buf: &[u8]) -> io::Result<usize> {
         self.0.write(buf)
     }
 
     pub fn fd(&self) -> &FileDesc { &self.0 }
     pub fn into_fd(self) -> FileDesc { self.0 }
 }
+
+pub fn read2(p1: AnonPipe,
+             v1: &mut Vec<u8>,
+             p2: AnonPipe,
+             v2: &mut Vec<u8>) -> io::Result<()> {
+    // Set both pipes into nonblocking mode as we're gonna be reading from both
+    // in the `select` loop below, and we wouldn't want one to block the other!
+    let p1 = p1.into_fd();
+    let p2 = p2.into_fd();
+    p1.set_nonblocking(true);
+    p2.set_nonblocking(true);
+
+    let max = cmp::max(p1.raw(), p2.raw());
+    loop {
+        // wait for either pipe to become readable using `select`
+        try!(cvt_r(|| unsafe {
+            let mut read: libc::fd_set = mem::zeroed();
+            libc::FD_SET(p1.raw(), &mut read);
+            libc::FD_SET(p2.raw(), &mut read);
+            libc::select(max + 1, &mut read, 0 as *mut _, 0 as *mut _,
+                         0 as *mut _)
+        }));
+
+        // Read as much as we can from each pipe, ignoring EWOULDBLOCK or
+        // EAGAIN. If we hit EOF, then this will happen because the underlying
+        // reader will return Ok(0), in which case we'll see `Ok` ourselves. In
+        // this case we flip the other fd back into blocking mode and read
+        // whatever's leftover on that file descriptor.
+        let read = |fd: &FileDesc, dst: &mut Vec<u8>| {
+            match fd.read_to_end(dst) {
+                Ok(_) => Ok(true),
+                Err(e) => {
+                    if e.raw_os_error() == Some(libc::EWOULDBLOCK) ||
+                       e.raw_os_error() == Some(libc::EAGAIN) {
+                        Ok(false)
+                    } else {
+                        Err(e)
+                    }
+                }
+            }
+        };
+        if try!(read(&p1, v1)) {
+            p2.set_nonblocking(false);
+            return p2.read_to_end(v2).map(|_| ());
+        }
+        if try!(read(&p2, v2)) {
+            p1.set_nonblocking(false);
+            return p1.read_to_end(v1).map(|_| ());
+        }
+    }
+}