Stream build script output to the console

This commit alters Cargo's behavior when the `-vv` option is passed (two verbose
flags) to stream output of all build scripts to the console. Cargo makes not
attempt to prevent interleaving or indicate *which* build script is producing
output, rather it simply forwards all output to one to the console.

Cargo still acts as a middle-man, capturing the output, to parse build script
output and interpret the results. The parsing is still deferred to completion
but the stream output happens while the build script is running.

On Unix this is implemented via `select` and on Windows this is implemented via
IOCP.

Closes rust-lang#1106

Cargo.toml

@@ -33,6 +33,7 @@ glob = "0.2"
 kernel32-sys = "0.2"
 libc = "0.2"
 log = "0.3"
+miow = "0.1"
 num_cpus = "0.2"
 regex = "0.1"
 rustc-serialize = "0.3"

src/cargo/ops/cargo_rustc/custom_build.rs

@@ -3,13 +3,16 @@ use std::fs;
 use std::path::{PathBuf, Path};
 use std::str;
 use std::sync::{Mutex, Arc};
+use std::process::{Stdio, Output};
 
 use core::PackageId;
 use util::{CargoResult, Human};
 use util::{internal, ChainError, profile, paths};
-use util::Freshness;
+use util::{Freshness, ProcessBuilder, read2};
+use util::errors::{process_error, ProcessError};
 
 use super::job::Work;
+use super::job_queue::JobState;
 use super::{fingerprint, Kind, Context, Unit};
 use super::CommandType;
 
@@ -159,14 +162,12 @@ fn build_work<'a, 'cfg>(cx: &mut Context<'a, 'cfg>, unit: &Unit<'a>)
     try!(fs::create_dir_all(&cx.layout(unit.pkg, Kind::Host).build(unit.pkg)));
     try!(fs::create_dir_all(&cx.layout(unit.pkg, unit.kind).build(unit.pkg)));
 
-    let exec_engine = cx.exec_engine.clone();
-
     // Prepare the unit of "dirty work" which will actually run the custom build
     // command.
     //
     // Note that this has to do some extra work just before running the command
     // to determine extra environment variables and such.
-    let dirty = Work::new(move |desc_tx| {
+    let dirty = Work::new(move |state| {
         // Make sure that OUT_DIR exists.
         //
         // If we have an old build directory, then just move it into place,
@@ -203,8 +204,9 @@ fn build_work<'a, 'cfg>(cx: &mut Context<'a, 'cfg>, unit: &Unit<'a>)
         }
 
         // And now finally, run the build command itself!
-        desc_tx.send(p.to_string()).ok();
-        let output = try!(exec_engine.exec_with_output(p).map_err(|mut e| {
+        state.running(&p);
+        let cmd = p.into_process_builder();
+        let output = try!(stream_output(state, &cmd).map_err(|mut e| {
             e.desc = format!("failed to run custom build command for `{}`\n{}",
                              pkg_name, e.desc);
             Human(e)
@@ -438,3 +440,55 @@ pub fn build_map<'b, 'cfg>(cx: &mut Context<'b, 'cfg>,
         }
     }
 }
+
+fn stream_output(state: &JobState, cmd: &ProcessBuilder)
+                 -> Result<Output, ProcessError> {
+    let mut stdout = Vec::new();
+    let mut stderr = Vec::new();
+
+    let status = try!((|| {
+        let mut cmd = cmd.build_command();
+        cmd.stdout(Stdio::piped())
+           .stderr(Stdio::piped())
+           .stdin(Stdio::null());
+        let mut child = try!(cmd.spawn());
+        let out = child.stdout.take().unwrap();
+        let err = child.stderr.take().unwrap();
+
+        try!(read2(out, err, &mut |is_out, data, eof| {
+            let idx = if eof {
+                data.len()
+            } else {
+                match data.iter().rposition(|b| *b == b'\n') {
+                    Some(i) => i + 1,
+                    None => return,
+                }
+            };
+            let data = data.drain(..idx);
+            let dst = if is_out {&mut stdout} else {&mut stderr};
+            let start = dst.len();
+            dst.extend(data);
+            let s = String::from_utf8_lossy(&dst[start..]);
+            if is_out {
+                state.stdout(&s);
+            } else {
+                state.stderr(&s);
+            }
+        }));
+        child.wait()
+    })().map_err(|e| {
+        let msg = format!("could not exeute process {}", cmd);
+        process_error(&msg, Some(e), None, None)
+    }));
+    let output = Output {
+        stdout: stdout,
+        stderr: stderr,
+        status: status,
+    };
+    if !output.status.success() {
+        let msg = format!("process didn't exit successfully: {}", cmd);
+        Err(process_error(&msg, None, Some(&status), Some(&output)))
+    } else {
+        Ok(output)
+    }
+}

src/cargo/ops/cargo_rustc/job.rs

@@ -1,14 +1,14 @@
-use std::sync::mpsc::Sender;
 use std::fmt;
 
 use util::{CargoResult, Fresh, Dirty, Freshness};
+use super::job_queue::JobState;
 
 pub struct Job { dirty: Work, fresh: Work }
 
 /// Each proc should send its description before starting.
 /// It should send either once or close immediately.
 pub struct Work {
-    inner: Box<FnBox<Sender<String>, CargoResult<()>> + Send>,
+    inner: Box<for <'a, 'b> FnBox<&'a JobState<'b>, CargoResult<()>> + Send>,
 }
 
 trait FnBox<A, R> {
@@ -23,7 +23,7 @@ impl<A, R, F: FnOnce(A) -> R> FnBox<A, R> for F {
 
 impl Work {
     pub fn new<F>(f: F) -> Work
-        where F: FnOnce(Sender<String>) -> CargoResult<()> + Send + 'static
+        where F: FnOnce(&JobState) -> CargoResult<()> + Send + 'static
     {
         Work { inner: Box::new(f) }
     }
@@ -32,14 +32,14 @@ impl Work {
         Work::new(|_| Ok(()))
     }
 
-    pub fn call(self, tx: Sender<String>) -> CargoResult<()> {
+    pub fn call(self, tx: &JobState) -> CargoResult<()> {
         self.inner.call_box(tx)
     }
 
     pub fn then(self, next: Work) -> Work {
-        Work::new(move |tx| {
-            try!(self.call(tx.clone()));
-            next.call(tx)
+        Work::new(move |state| {
+            try!(self.call(state));
+            next.call(state)
         })
     }
 }
@@ -52,10 +52,10 @@ impl Job {
 
     /// Consumes this job by running it, returning the result of the
     /// computation.
-    pub fn run(self, fresh: Freshness, tx: Sender<String>) -> CargoResult<()> {
+    pub fn run(self, fresh: Freshness, state: &JobState) -> CargoResult<()> {
         match fresh {
-            Fresh => self.fresh.call(tx),
-            Dirty => self.dirty.call(tx),
+            Fresh => self.fresh.call(state),
+            Dirty => self.dirty.call(state),
         }
     }
 }