Remove dependency on pipe, unless parallel

Author: dpaoliello

Cargo.toml

@@ -22,10 +22,10 @@ rust-version = "1.53"
 [target.'cfg(unix)'.dependencies]
 # Don't turn on the feature "std" for this, see https://github.com/rust-lang/cargo/issues/4866
 # which is still an issue with `resolver = "1"`.
-libc = { version = "0.2.62", default-features = false }
+libc = { version = "0.2.62", default-features = false, optional = true }
 
 [features]
-parallel = []
+parallel = ["libc"]
 
 [dev-dependencies]
 tempfile = "3"

src/command_helpers.rs

@@ -4,13 +4,12 @@ use std::{
     collections::hash_map,
     ffi::OsString,
     fmt::Display,
-    fs::{self, File},
+    fs,
     hash::Hasher,
-    io::{self, BufRead, BufReader, Read, Write},
+    io::{self, Read, Write},
     path::Path,
-    process::{Child, Command, Stdio},
+    process::{Child, ChildStderr, Command, Stdio},
     sync::Arc,
-    thread::{self, JoinHandle},
 };
 
 use crate::{Error, ErrorKind, Object};
@@ -41,83 +40,175 @@ impl CargoOutput {
         }
     }
 
-    pub(crate) fn print_thread(&self) -> Result<Option<PrintThread>, Error> {
-        self.warnings.then(PrintThread::new).transpose()
+    fn stdio_for_warnings(&self) -> Stdio {
+        if self.warnings {
+            Stdio::piped()
+        } else {
+            Stdio::null()
+        }
     }
 }
 
-pub(crate) struct PrintThread {
-    handle: Option<JoinHandle<()>>,
-    pipe_writer: Option<File>,
+pub(crate) struct StderrForwarder {
+    inner: Option<(ChildStderr, Vec<u8>)>,
+    #[cfg(feature = "parallel")]
+    is_non_blocking: bool,
 }
 
-impl PrintThread {
-    pub(crate) fn new() -> Result<Self, Error> {
-        let (pipe_reader, pipe_writer) = crate::os_pipe::pipe()?;
-
-        // Capture the standard error coming from compilation, and write it out
-        // with cargo:warning= prefixes. Note that this is a bit wonky to avoid
-        // requiring the output to be UTF-8, we instead just ship bytes from one
-        // location to another.
-        let print = thread::spawn(move || {
-            let mut stderr = BufReader::with_capacity(4096, pipe_reader);
-            let mut line = Vec::with_capacity(20);
-            let stdout = io::stdout();
+#[cfg(all(feature = "parallel", not(unix), not(windows)))]
+compile_error!("Only unix and windows support non-blocking pipes! For other OSes, disable the parallel feature.");
+
+const MIN_BUFFER_CAPACITY: usize = 100;
+
+impl StderrForwarder {
+    pub(crate) fn new(child: &mut Child) -> Self {
+        Self {
+            inner: child
+                .stderr
+                .take()
+                .map(|stderr| (stderr, Vec::with_capacity(MIN_BUFFER_CAPACITY))),
+            #[cfg(feature = "parallel")]
+            is_non_blocking: false,
+        }
+    }
 
-            // read_until returns 0 on Eof
-            while stderr.read_until(b'\n', &mut line).unwrap() != 0 {
-                {
-                    let mut stdout = stdout.lock();
+    fn forward_available(&mut self) -> bool {
+        if let Some((stderr, buffer)) = self.inner.as_mut() {
+            let stdout = io::stdout();
+            let write_warning = move |line: &[u8]| {
+                let mut stdout = stdout.lock();
+                stdout.write_all(b"cargo:warning=").unwrap();
+                stdout.write_all(line).unwrap();
+                stdout.write_all(b"\n").unwrap();
+            };
 
-                    stdout.write_all(b"cargo:warning=").unwrap();
-                    stdout.write_all(&line).unwrap();
-                    stdout.write_all(b"\n").unwrap();
+            #[cfg(all(windows, feature = "parallel"))]
+            let is_non_blocking = self.is_non_blocking;
+            let mut read_stderr = move |buf: &mut [u8]| -> Result<usize, io::Error> {
+                // On Unix, the pipe is non-blocking, so we can just read.
+                // On Windows, take a peek at the pipe to see if there's data.
+                #[cfg(all(windows, feature = "parallel"))]
+                if is_non_blocking {
+                    use crate::windows::windows_sys::PeekNamedPipe;
+                    use std::os::windows::io::AsRawHandle;
+                    use std::ptr::null_mut;
+                    let mut bytes_available = 0;
+                    unsafe {
+                        if PeekNamedPipe(
+                            stderr.as_raw_handle(),
+                            null_mut(),
+                            0,
+                            null_mut(),
+                            &mut bytes_available,
+                            null_mut(),
+                        ) == 0
+                        {
+                            return Err(io::Error::last_os_error());
+                        }
+                    }
+                    if bytes_available == 0 {
+                        return Err(io::Error::new(
+                            io::ErrorKind::WouldBlock,
+                            "The pipe is empty",
+                        ));
+                    }
                 }
 
-                // read_until does not clear the buffer
-                line.clear();
-            }
-        });
+                stderr.read(buf)
+            };
 
-        Ok(Self {
-            handle: Some(print),
-            pipe_writer: Some(pipe_writer),
-        })
+            loop {
+                buffer.reserve(MIN_BUFFER_CAPACITY);
+
+                let old_data_end = buffer.len();
+                buffer.resize(buffer.capacity(), 0);
+                match read_stderr(&mut buffer[old_data_end..]) {
+                    Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
+                        // No data currently, yield back.
+                        buffer.truncate(old_data_end);
+                        return false;
+                    }
+                    Err(err) if err.kind() == std::io::ErrorKind::Interrupted => {
+                        // Interrupted, try again.
+                        buffer.truncate(old_data_end);
+                    }
+                    Ok(0) | Err(_) => {
+                        // End of stream: flush remaining data and bail.
+                        if old_data_end > 0 {
+                            write_warning(&buffer[..old_data_end]);
+                        }
+                        return true;
+                    }
+                    Ok(bytes_read) => {
+                        buffer.truncate(old_data_end + bytes_read);
+                        let mut consumed = 0;
+                        for line in buffer.split_inclusive(|&b| b == b'\n') {
+                            // Only forward complete lines, leave the rest in the buffer.
+                            if let Some((b'\n', line)) = line.split_last() {
+                                consumed += line.len() + 1;
+                                write_warning(line);
+                            }
+                        }
+                        buffer.drain(..consumed);
+                    }
+                }
+            }
+        } else {
+            true
+        }
     }
 
-    /// # Panics
-    ///
-    /// Will panic if the pipe writer has already been taken.
-    pub(crate) fn take_pipe_writer(&mut self) -> File {
-        self.pipe_writer.take().unwrap()
-    }
+    #[cfg(feature = "parallel")]
+    pub(crate) fn set_non_blocking(&mut self) -> Result<(), Error> {
+        assert!(!self.is_non_blocking);
+
+        // On Unix, switch the pipe to non-blocking mode.
+        // On Windows, we have a different way to be non-blocking.
+        #[cfg(unix)]
+        if let Some((stderr, _)) = self.inner.as_mut() {
+            use std::os::unix::io::AsRawFd;
+            let fd = stderr.as_raw_fd();
+            let flags = unsafe { libc::fcntl(fd, libc::F_GETFL, 0) };
+            if flags < 0 {
+                return Err(Error::new(
+                    ErrorKind::IOError,
+                    format!(
+                        "Failed to get flags for child stderr: {}",
+                        io::Error::last_os_error()
+                    ),
+                ));
+            }
 
-    /// # Panics
-    ///
-    /// Will panic if the pipe writer has already been taken.
-    pub(crate) fn clone_pipe_writer(&self) -> Result<File, Error> {
-        self.try_clone_pipe_writer().map(Option::unwrap)
-    }
+            if unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } != 0 {
+                return Err(Error::new(
+                    ErrorKind::IOError,
+                    format!(
+                        "Failed to set flags for child stderr: {}",
+                        io::Error::last_os_error()
+                    ),
+                ));
+            }
+        }
 
-    pub(crate) fn try_clone_pipe_writer(&self) -> Result<Option<File>, Error> {
-        self.pipe_writer
-            .as_ref()
-            .map(File::try_clone)
-            .transpose()
-            .map_err(From::from)
+        self.is_non_blocking = true;
+        Ok(())
     }
-}
 
-impl Drop for PrintThread {
-    fn drop(&mut self) {
-        // Drop pipe_writer first to avoid deadlock
-        self.pipe_writer.take();
+    #[cfg(feature = "parallel")]
+    fn forward_all(&mut self) {
+        while !self.forward_available() {}
+    }
 
-        self.handle.take().unwrap().join().unwrap();
+    #[cfg(not(feature = "parallel"))]
+    fn forward_all(&mut self) {
+        let forward_result = self.forward_available();
+        assert!(forward_result, "Should have consumed all data");
     }
 }
 
 fn wait_on_child(cmd: &Command, program: &str, child: &mut Child) -> Result<(), Error> {
+    StderrForwarder::new(child).forward_all();
+
     let status = match child.wait() {
         Ok(s) => s,
         Err(e) => {
@@ -193,20 +284,13 @@ pub(crate) fn objects_from_files(files: &[Arc<Path>], dst: &Path) -> Result<Vec<
     Ok(objects)
 }
 
-fn run_inner(cmd: &mut Command, program: &str, pipe_writer: Option<File>) -> Result<(), Error> {
-    let mut child = spawn(cmd, program, pipe_writer)?;
-    wait_on_child(cmd, program, &mut child)
-}
-
 pub(crate) fn run(
     cmd: &mut Command,
     program: &str,
-    print: Option<&PrintThread>,
+    cargo_output: &CargoOutput,
 ) -> Result<(), Error> {
-    let pipe_writer = print.map(PrintThread::clone_pipe_writer).transpose()?;
-    run_inner(cmd, program, pipe_writer)?;
-
-    Ok(())
+    let mut child = spawn(cmd, program, cargo_output)?;
+    wait_on_child(cmd, program, &mut child)
 }
 
 pub(crate) fn run_output(
@@ -216,12 +300,7 @@ pub(crate) fn run_output(
 ) -> Result<Vec<u8>, Error> {
     cmd.stdout(Stdio::piped());
 
-    let mut print = cargo_output.print_thread()?;
-    let mut child = spawn(
-        cmd,
-        program,
-        print.as_mut().map(PrintThread::take_pipe_writer),
-    )?;
+    let mut child = spawn(cmd, program, cargo_output)?;
 
     let mut stdout = vec![];
     child
@@ -239,7 +318,7 @@ pub(crate) fn run_output(
 pub(crate) fn spawn(
     cmd: &mut Command,
     program: &str,
-    pipe_writer: Option<File>,
+    cargo_output: &CargoOutput,
 ) -> Result<Child, Error> {
     struct ResetStderr<'cmd>(&'cmd mut Command);
 
@@ -254,10 +333,7 @@ pub(crate) fn spawn(
     println!("running: {:?}", cmd);
 
     let cmd = ResetStderr(cmd);
-    let child = cmd
-        .0
-        .stderr(pipe_writer.map_or_else(Stdio::null, Stdio::from))
-        .spawn();
+    let child = cmd.0.stderr(cargo_output.stdio_for_warnings()).spawn();
     match child {
         Ok(child) => Ok(child),
         Err(ref e) if e.kind() == io::ErrorKind::NotFound => {
@@ -307,9 +383,14 @@ pub(crate) fn try_wait_on_child(
     program: &str,
     child: &mut Child,
     stdout: &mut dyn io::Write,
+    stderr_forwarder: &mut StderrForwarder,
 ) -> Result<Option<()>, Error> {
+    stderr_forwarder.forward_available();
+
     match child.try_wait() {
         Ok(Some(status)) => {
+            stderr_forwarder.forward_all();
+
             let _ = writeln!(stdout, "{}", status);
 
             if status.success() {
@@ -325,12 +406,15 @@ pub(crate) fn try_wait_on_child(
             }
         }
         Ok(None) => Ok(None),
-        Err(e) => Err(Error::new(
-            ErrorKind::ToolExecError,
-            format!(
-                "Failed to wait on spawned child process, command {:?} with args {:?}: {}.",
-                cmd, program, e
-            ),
-        )),
+        Err(e) => {
+            stderr_forwarder.forward_all();
+            Err(Error::new(
+                ErrorKind::ToolExecError,
+                format!(
+                    "Failed to wait on spawned child process, command {:?} with args {:?}: {}.",
+                    cmd, program, e
+                ),
+            ))
+        }
     }
 }