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, BufRead, BufReader, Read, Stdout, Write},
     path::Path,
     process::{Child, Command, Stdio},
     sync::Arc,
-    thread::{self, JoinHandle},
 };
 
 use crate::{Error, ErrorKind, Object};
@@ -41,83 +40,46 @@ 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) fn reader_for_stderr_as_warnings(
+    child: &mut Child,
+) -> Option<BufReader<std::process::ChildStderr>> {
+    child.stderr.take().map(BufReader::new)
 }
 
-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();
+fn forward_all_stderr_as_warnings(child: &mut Child) {
+    if let Some(mut stderr) = reader_for_stderr_as_warnings(child) {
+        let mut line = Vec::new();
+        let stdout = io::stdout();
 
-            // read_until returns 0 on Eof
-            while stderr.read_until(b'\n', &mut line).unwrap() != 0 {
-                {
-                    let mut stdout = stdout.lock();
+        // read_until returns 0 on Eof
+        while stderr.read_until(b'\n', &mut line).unwrap_or_default() != 0 {
+            write_warning(&stdout, &line);
 
-                    stdout.write_all(b"cargo:warning=").unwrap();
-                    stdout.write_all(&line).unwrap();
-                    stdout.write_all(b"\n").unwrap();
-                }
-
-                // read_until does not clear the buffer
-                line.clear();
-            }
-        });
-
-        Ok(Self {
-            handle: Some(print),
-            pipe_writer: Some(pipe_writer),
-        })
-    }
-
-    /// # 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()
-    }
-
-    /// # 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)
-    }
-
-    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)
+            // read_until does not clear the buffer
+            line.clear();
+        }
     }
 }
 
-impl Drop for PrintThread {
-    fn drop(&mut self) {
-        // Drop pipe_writer first to avoid deadlock
-        self.pipe_writer.take();
+fn write_warning(stdout: &Stdout, line: &[u8]) {
+    let mut stdout = stdout.lock();
 
-        self.handle.take().unwrap().join().unwrap();
-    }
+    stdout.write_all(b"cargo:warning=").unwrap();
+    stdout.write_all(line).unwrap();
+    stdout.write_all(b"\n").unwrap();
 }
 
 fn wait_on_child(cmd: &Command, program: &str, child: &mut Child) -> Result<(), Error> {
+    forward_all_stderr_as_warnings(child);
     let status = match child.wait() {
         Ok(s) => s,
         Err(e) => {
@@ -193,20 +155,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 +171,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 +189,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 +204,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 => {
@@ -301,14 +248,41 @@ pub(crate) fn command_add_output_file(
     }
 }
 
+#[cfg(feature = "parallel")]
+fn forward_available_stderr_as_warnings(
+    child_stderr_reader: &mut Option<BufReader<std::process::ChildStderr>>,
+) {
+    if let Some(child_stderr_reader) = child_stderr_reader.as_mut() {
+        if let Ok(available) = child_stderr_reader.fill_buf() {
+            let stdout = io::stdout();
+            let mut consumed = 0;
+            for line in available.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(&stdout, line);
+                }
+            }
+            child_stderr_reader.consume(consumed);
+        }
+    }
+}
+
 #[cfg(feature = "parallel")]
 pub(crate) fn try_wait_on_child(
     cmd: &Command,
     program: &str,
     child: &mut Child,
     stdout: &mut dyn io::Write,
+    child_stderr_reader: &mut Option<BufReader<std::process::ChildStderr>>,
 ) -> Result<Option<()>, Error> {
-    match child.try_wait() {
+    // Check the child status THEN forward stderr messages, so that we don't race
+    // between the child printing messages and then exiting.
+    let wait_result = child.try_wait();
+
+    forward_available_stderr_as_warnings(child_stderr_reader);
+
+    match wait_result {
         Ok(Some(status)) => {
             let _ = writeln!(stdout, "{}", status);