Improve performance of almost fresh builds

src/cargo/core/compiler/custom_build.rs

@@ -128,7 +128,7 @@ fn emit_build_output(
     output: &BuildOutput,
     out_dir: &Path,
     package_id: PackageId,
-) {
+) -> CargoResult<()> {
     let library_paths = output
         .library_paths
         .iter()
@@ -144,7 +144,8 @@ fn emit_build_output(
         out_dir,
     }
     .to_json_string();
-    state.stdout(msg);
+    state.stdout(msg)?;
+    Ok(())
 }
 
 fn build_work(cx: &mut Context<'_, '_>, unit: &Unit) -> CargoResult<Job> {
@@ -353,13 +354,13 @@ fn build_work(cx: &mut Context<'_, '_>, unit: &Unit) -> CargoResult<Job> {
                         warnings_in_case_of_panic.push(warning.to_owned());
                     }
                     if extra_verbose {
-                        state.stdout(format!("{}{}", prefix, stdout));
+                        state.stdout(format!("{}{}", prefix, stdout))?;
                     }
                     Ok(())
                 },
                 &mut |stderr| {
                     if extra_verbose {
-                        state.stderr(format!("{}{}", prefix, stderr));
+                        state.stderr(format!("{}{}", prefix, stderr))?;
                     }
                     Ok(())
                 },
@@ -396,7 +397,7 @@ fn build_work(cx: &mut Context<'_, '_>, unit: &Unit) -> CargoResult<Job> {
             BuildOutput::parse(&output.stdout, &pkg_name, &script_out_dir, &script_out_dir)?;
 
         if json_messages {
-            emit_build_output(state, &parsed_output, script_out_dir.as_path(), id);
+            emit_build_output(state, &parsed_output, script_out_dir.as_path(), id)?;
         }
         build_script_outputs
             .lock()
@@ -421,7 +422,7 @@ fn build_work(cx: &mut Context<'_, '_>, unit: &Unit) -> CargoResult<Job> {
         };
 
         if json_messages {
-            emit_build_output(state, &output, script_out_dir.as_path(), id);
+            emit_build_output(state, &output, script_out_dir.as_path(), id)?;
         }
 
         build_script_outputs
@@ -432,7 +433,7 @@ fn build_work(cx: &mut Context<'_, '_>, unit: &Unit) -> CargoResult<Job> {
     });
 
     let mut job = if cx.bcx.build_config.build_plan {
-        Job::new(Work::noop(), Freshness::Dirty)
+        Job::new_dirty(Work::noop())
     } else {
         fingerprint::prepare_target(cx, unit, false)?
     };

src/cargo/core/compiler/fingerprint.rs

@@ -337,10 +337,7 @@ use crate::util::paths;
 use crate::util::{internal, profile, ProcessBuilder};
 
 use super::custom_build::BuildDeps;
-use super::job::{
-    Freshness::{Dirty, Fresh},
-    Job, Work,
-};
+use super::job::{Job, Work};
 use super::{BuildContext, Context, FileFlavor, Unit};
 
 /// Determines if a `unit` is up-to-date, and if not prepares necessary work to
@@ -396,7 +393,7 @@ pub fn prepare_target(cx: &mut Context<'_, '_>, unit: &Unit, force: bool) -> Car
     }
 
     if compare.is_ok() && !force {
-        return Ok(Job::new(Work::noop(), Fresh));
+        return Ok(Job::new_fresh());
     }
 
     // Clear out the old fingerprint file if it exists. This protects when
@@ -469,7 +466,7 @@ pub fn prepare_target(cx: &mut Context<'_, '_>, unit: &Unit, force: bool) -> Car
         Work::new(move |_| write_fingerprint(&loc, &fingerprint))
     };
 
-    Ok(Job::new(write_fingerprint, Dirty))
+    Ok(Job::new_dirty(write_fingerprint))
 }
 
 /// Dependency edge information for fingerprints. This is generated for each

src/cargo/core/compiler/job.rs

@@ -40,9 +40,20 @@ impl Work {
 }
 
 impl Job {
+    /// Creates a new job that does nothing.
+    pub fn new_fresh() -> Job {
+        Job {
+            work: Work::noop(),
+            fresh: Freshness::Fresh,
+        }
+    }
+
     /// Creates a new job representing a unit of work.
-    pub fn new(work: Work, fresh: Freshness) -> Job {
-        Job { work, fresh }
+    pub fn new_dirty(work: Work) -> Job {
+        Job {
+            work,
+            fresh: Freshness::Dirty,
+        }
     }
 
     /// Consumes this job by running it, returning the result of the

src/cargo/core/compiler/job_queue.rs

@@ -162,10 +162,30 @@ impl std::fmt::Display for JobId {
     }
 }
 
+/// A `JobState` is constructed by `JobQueue::run` and passed to `Job::run`. It includes everything
+/// necessary to communicate between the main thread and the execution of the job.
+///
+/// The job may execute on either a dedicated thread or the main thread. If the job executes on the
+/// main thread, the `output` field must be set to prevent a deadlock.
 pub struct JobState<'a> {
     /// Channel back to the main thread to coordinate messages and such.
+    ///
+    /// When the `output` field is `Some`, care must be taken to avoid calling `push_bounded` on
+    /// the message queue to prevent a deadlock.
     messages: Arc<Queue<Message>>,
 
+    /// Normally output is sent to the job queue with backpressure. When the job is fresh
+    /// however we need to immediately display the output to prevent a deadlock as the
+    /// output messages are processed on the same thread as they are sent from. `output`
+    /// defines where to output in this case.
+    ///
+    /// Currently the `Shell` inside `Config` is wrapped in a `RefCell` and thus can't be passed
+    /// between threads. This means that it isn't possible for multiple output messages to be
+    /// interleaved. In the future, it may be wrapped in a `Mutex` instead. In this case
+    /// interleaving is still prevented as the lock would be held for the whole printing of an
+    /// output message.
+    output: Option<&'a Config>,
+
     /// The job id that this state is associated with, used when sending
     /// messages back to the main thread.
     id: JobId,
@@ -231,12 +251,24 @@ impl<'a> JobState<'a> {
             .push(Message::BuildPlanMsg(module_name, cmd, filenames));
     }
 
-    pub fn stdout(&self, stdout: String) {
-        self.messages.push_bounded(Message::Stdout(stdout));
+    pub fn stdout(&self, stdout: String) -> CargoResult<()> {
+        if let Some(config) = self.output {
+            writeln!(config.shell().out(), "{}", stdout)?;
+        } else {
+            self.messages.push_bounded(Message::Stdout(stdout));
+        }
+        Ok(())
     }
 
-    pub fn stderr(&self, stderr: String) {
-        self.messages.push_bounded(Message::Stderr(stderr));
+    pub fn stderr(&self, stderr: String) -> CargoResult<()> {
+        if let Some(config) = self.output {
+            let mut shell = config.shell();
+            shell.print_ansi(stderr.as_bytes())?;
+            shell.err().write_all(b"\n")?;
+        } else {
+            self.messages.push_bounded(Message::Stderr(stderr));
+        }
+        Ok(())
     }
 
     /// A method used to signal to the coordinator thread that the rmeta file
@@ -826,16 +858,9 @@ impl<'cfg> DrainState<'cfg> {
             self.note_working_on(cx.bcx.config, unit, fresh)?;
         }
 
-        let doit = move || {
-            let state = JobState {
-                id,
-                messages: messages.clone(),
-                rmeta_required: Cell::new(rmeta_required),
-                _marker: marker::PhantomData,
-            };
-
+        let doit = move |state: JobState<'_>| {
             let mut sender = FinishOnDrop {
-                messages: &messages,
+                messages: &state.messages,
                 id,
                 result: None,
             };
@@ -854,7 +879,9 @@ impl<'cfg> DrainState<'cfg> {
             // we need to make sure that the metadata is flagged as produced so
             // send a synthetic message here.
             if state.rmeta_required.get() && sender.result.as_ref().unwrap().is_ok() {
-                messages.push(Message::Finish(id, Artifact::Metadata, Ok(())));
+                state
+                    .messages
+                    .push(Message::Finish(state.id, Artifact::Metadata, Ok(())));
             }
 
             // Use a helper struct with a `Drop` implementation to guarantee
@@ -880,10 +907,31 @@ impl<'cfg> DrainState<'cfg> {
         };
 
         match fresh {
-            Freshness::Fresh => self.timings.add_fresh(),
-            Freshness::Dirty => self.timings.add_dirty(),
+            Freshness::Fresh => {
+                self.timings.add_fresh();
+                // Running a fresh job on the same thread is often much faster than spawning a new
+                // thread to run the job.
+                doit(JobState {
+                    id,
+                    messages: messages.clone(),
+                    output: Some(cx.bcx.config),
+                    rmeta_required: Cell::new(rmeta_required),
+                    _marker: marker::PhantomData,
+                });
+            }
+            Freshness::Dirty => {
+                self.timings.add_dirty();
+                scope.spawn(move |_| {
+                    doit(JobState {
+                        id,
+                        messages: messages.clone(),
+                        output: None,
+                        rmeta_required: Cell::new(rmeta_required),
+                        _marker: marker::PhantomData,
+                    })
+                });
+            }
         }
-        scope.spawn(move |_| doit());
 
         Ok(())
     }

src/cargo/core/compiler/mod.rs

@@ -130,9 +130,9 @@ fn compile<'cfg>(
         custom_build::prepare(cx, unit)?
     } else if unit.mode.is_doc_test() {
         // We run these targets later, so this is just a no-op for now.
-        Job::new(Work::noop(), Freshness::Fresh)
+        Job::new_fresh()
     } else if build_plan {
-        Job::new(rustc(cx, unit, &exec.clone())?, Freshness::Dirty)
+        Job::new_dirty(rustc(cx, unit, &exec.clone())?)
     } else {
         let force = exec.force_rebuild(unit) || force_rebuild;
         let mut job = fingerprint::prepare_target(cx, unit, force)?;
@@ -448,7 +448,7 @@ fn link_targets(cx: &mut Context<'_, '_>, unit: &Unit, fresh: bool) -> CargoResu
                 fresh,
             }
             .to_json_string();
-            state.stdout(msg);
+            state.stdout(msg)?;
         }
         Ok(())
     }))
@@ -1139,7 +1139,7 @@ fn on_stdout_line(
     _package_id: PackageId,
     _target: &Target,
 ) -> CargoResult<()> {
-    state.stdout(line.to_string());
+    state.stdout(line.to_string())?;
     Ok(())
 }
 
@@ -1177,7 +1177,7 @@ fn on_stderr_line_inner(
     // something like that), so skip over everything that doesn't look like a
     // JSON message.
     if !line.starts_with('{') {
-        state.stderr(line.to_string());
+        state.stderr(line.to_string())?;
         return Ok(true);
     }
 
@@ -1189,7 +1189,7 @@ fn on_stderr_line_inner(
         // to stderr.
         Err(e) => {
             debug!("failed to parse json: {:?}", e);
-            state.stderr(line.to_string());
+            state.stderr(line.to_string())?;
             return Ok(true);
         }
     };
@@ -1225,7 +1225,7 @@ fn on_stderr_line_inner(
                         .map(|v| String::from_utf8(v).expect("utf8"))
                         .expect("strip should never fail")
                 };
-                state.stderr(rendered);
+                state.stderr(rendered)?;
                 return Ok(true);
             }
         }
@@ -1316,7 +1316,7 @@ fn on_stderr_line_inner(
     // Switch json lines from rustc/rustdoc that appear on stderr to stdout
     // instead. We want the stdout of Cargo to always be machine parseable as
     // stderr has our colorized human-readable messages.
-    state.stdout(msg);
+    state.stdout(msg)?;
     Ok(true)
 }