Remove dependency on pipe, unless parallel

Author: dpaoliello

Cargo.toml

@@ -19,13 +19,17 @@ exclude = ["/.github", "tests", "src/bin"]
 edition = "2018"
 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
+# Don't turn on the feature "std" for these dependencies, 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 }
+
+[dependencies]
+memchr = { version = "2.5", default-features = false, optional = true }
+
+[target.'cfg(unix)'.dependencies]
+libc = { version = "0.2.62", default-features = false, optional = true }
 
 [features]
-parallel = []
+parallel = ["libc", "memchr"]
 
 [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,55 @@ 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>,
-}
 
-impl PrintThread {
-    pub(crate) fn new() -> Result<Self, Error> {
-        let (pipe_reader, pipe_writer) = crate::os_pipe::pipe()?;
+    #[cfg(feature = "parallel")]
+    pub(crate) fn reader_for_stderr_as_warnings(
+        &self,
+        child: &mut Child,
+    ) -> Option<BufReader<std::process::ChildStderr>> {
+        self.warnings
+            .then(|| BufReader::with_capacity(4096, child.stderr.take().unwrap()))
+    }
 
-        // 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);
+    fn forward_stderr_as_warnings(&self, child: &mut Child) {
+        if self.warnings {
+            let mut stderr = BufReader::with_capacity(4096, child.stderr.as_mut().unwrap());
             let mut line = Vec::with_capacity(20);
             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();
-
-                    stdout.write_all(b"cargo:warning=").unwrap();
-                    stdout.write_all(&line).unwrap();
-                    stdout.write_all(b"\n").unwrap();
-                }
+            while stderr.read_until(b'\n', &mut line).unwrap_or_default() != 0 {
+                write_warning(&stdout, &line);
 
                 // 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)
+        }
     }
 }
 
-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> {
+fn wait_on_child(
+    cmd: &Command,
+    program: &str,
+    child: &mut Child,
+    cargo_output: &CargoOutput,
+) -> Result<(), Error> {
+    cargo_output.forward_stderr_as_warnings(child);
     let status = match child.wait() {
         Ok(s) => s,
         Err(e) => {
@@ -193,20 +164,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, cargo_output)
 }
 
 pub(crate) fn run_output(
@@ -216,12 +180,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
@@ -231,15 +190,15 @@ pub(crate) fn run_output(
         .read_to_end(&mut stdout)
         .unwrap();
 
-    wait_on_child(cmd, program, &mut child)?;
+    wait_on_child(cmd, program, &mut child, cargo_output)?;
 
     Ok(stdout)
 }
 
 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 +213,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,8 +263,27 @@ pub(crate) fn try_wait_on_child(
     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();
+
+    if let Some(child_stderr_reader) = child_stderr_reader.as_mut() {
+        if let Ok(mut available) = child_stderr_reader.fill_buf() {
+            let stdout = io::stdout();
+            let original_size = available.len();
+            while let Some(i) = memchr::memchr(b'\n', available) {
+                let (line, remainder) = available.split_at(i);
+                write_warning(&stdout, line);
+                available = &remainder[1..];
+            }
+            let consumed = original_size - available.len();
+            child_stderr_reader.consume(consumed);
+        }
+    }
+
+    match wait_result {
         Ok(Some(status)) => {
             let _ = writeln!(stdout, "{}", status);