feat(timings): derive concurrency dat from unit data

Added a post-processing that derives concurrency information
(active/waiting/inactive units over time)
from completed unit timing data.

This enables eliminating runtime concurrency tracking.

Author: weihanglo

src/cargo/core/compiler/job_queue/mod.rs

@@ -887,11 +887,6 @@ impl<'gctx> DrainState<'gctx> {
         // Record some timing information if `--timings` is enabled, and
         // this'll end up being a noop if we're not recording this
         // information.
-        self.timings.mark_concurrency(
-            self.active.len(),
-            self.pending_queue.len(),
-            self.queue.len(),
-        );
         self.timings.record_cpu();
 
         let active_names = self

src/cargo/core/compiler/timings/mod.rs

@@ -62,9 +62,6 @@ pub struct Timings<'gctx> {
     /// Units that are in the process of being built.
     /// When they finished, they are moved to `unit_times`.
     active: HashMap<JobId, UnitTime>,
-    /// Concurrency-tracking information. This is periodically updated while
-    /// compilation progresses.
-    concurrency: Vec<Concurrency>,
     /// Last recorded state of the system's CPUs and when it happened
     last_cpu_state: Option<State>,
     last_cpu_recording: Instant,
@@ -161,7 +158,6 @@ impl<'gctx> Timings<'gctx> {
                 unit_to_index: HashMap::new(),
                 unit_times: Vec::new(),
                 active: HashMap::new(),
-                concurrency: Vec::new(),
                 last_cpu_state: None,
                 last_cpu_recording: Instant::now(),
                 cpu_usage: Vec::new(),
@@ -214,7 +210,6 @@ impl<'gctx> Timings<'gctx> {
             unit_to_index,
             unit_times: Vec::new(),
             active: HashMap::new(),
-            concurrency: Vec::new(),
             last_cpu_state,
             last_cpu_recording: Instant::now(),
             cpu_usage: Vec::new(),
@@ -384,20 +379,6 @@ impl<'gctx> Timings<'gctx> {
         }
     }
 
-    /// This is called periodically to mark the concurrency of internal structures.
-    pub fn mark_concurrency(&mut self, active: usize, waiting: usize, inactive: usize) {
-        if !self.enabled {
-            return;
-        }
-        let c = Concurrency {
-            t: self.start.elapsed().as_secs_f64(),
-            active,
-            waiting,
-            inactive,
-        };
-        self.concurrency.push(c);
-    }
-
     /// Mark that a fresh unit was encountered. (No re-compile needed)
     pub fn add_fresh(&mut self) {
         self.total_fresh += 1;
@@ -444,7 +425,6 @@ impl<'gctx> Timings<'gctx> {
         if !self.enabled {
             return Ok(());
         }
-        self.mark_concurrency(0, 0, 0);
         self.unit_times
             .sort_unstable_by(|a, b| a.start.partial_cmp(&b.start).unwrap());
         if self.report_html {
@@ -473,6 +453,7 @@ impl<'gctx> Timings<'gctx> {
                 .collect::<Vec<_>>();
 
             let unit_data = report::to_unit_data(&self.unit_times, &self.unit_to_index);
+            let concurrency = report::compute_concurrency(&unit_data);
 
             let ctx = report::RenderContext {
                 start: self.start,
@@ -482,7 +463,7 @@ impl<'gctx> Timings<'gctx> {
                 total_fresh: self.total_fresh,
                 total_dirty: self.total_dirty,
                 unit_data,
-                concurrency: &self.concurrency,
+                concurrency,
                 cpu_usage: &self.cpu_usage,
                 rustc_version,
                 host: &build_runner.bcx.rustc().host,

src/cargo/core/compiler/timings/report.rs

@@ -2,6 +2,7 @@
 
 use std::borrow::Cow;
 use std::collections::HashMap;
+use std::collections::HashSet;
 use std::io::Write;
 use std::time::Instant;
 
@@ -100,7 +101,7 @@ pub struct RenderContext<'a> {
     pub unit_data: Vec<UnitData>,
     /// Concurrency-tracking information. This is periodically updated while
     /// compilation progresses.
-    pub concurrency: &'a [Concurrency],
+    pub concurrency: Vec<Concurrency>,
     /// Recorded CPU states, stored as tuples. First element is when the
     /// recording was taken and second element is percentage usage of the
     /// system.
@@ -391,6 +392,131 @@ pub(super) fn to_unit_data(
         .collect()
 }
 
+/// Derives concurrency information from unit timing data.
+pub(super) fn compute_concurrency(unit_data: &[UnitData]) -> Vec<Concurrency> {
+    if unit_data.is_empty() {
+        return Vec::new();
+    }
+
+    let unit_by_index: HashMap<_, _> = unit_data.iter().map(|u| (u.i, u)).collect();
+
+    enum UnblockedBy {
+        Rmeta(u64),
+        Full(u64),
+    }
+
+    // unit_id -> unit that unblocks it.
+    let mut unblocked_by: HashMap<_, _> = HashMap::new();
+    for unit in unit_data {
+        for id in unit.unblocked_rmeta_units.iter() {
+            assert!(
+                unblocked_by
+                    .insert(*id, UnblockedBy::Rmeta(unit.i))
+                    .is_none()
+            );
+        }
+
+        for id in unit.unblocked_units.iter() {
+            assert!(
+                unblocked_by
+                    .insert(*id, UnblockedBy::Full(unit.i))
+                    .is_none()
+            );
+        }
+    }
+
+    let ready_time = |unit: &UnitData| -> Option<f64> {
+        let dep = unblocked_by.get(&unit.i)?;
+        match dep {
+            UnblockedBy::Rmeta(id) => {
+                let dep = unit_by_index.get(id)?;
+                let duration = dep.sections.iter().flatten().find_map(|(name, section)| {
+                    matches!(name, SectionName::Frontend).then_some(section.end)
+                });
+
+                Some(dep.start + duration.unwrap_or(dep.duration))
+            }
+            UnblockedBy::Full(id) => {
+                let dep = unit_by_index.get(id)?;
+                Some(dep.start + dep.duration)
+            }
+        }
+    };
+
+    #[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
+    enum State {
+        Ready,
+        Start,
+        End,
+    }
+
+    let mut events: Vec<_> = unit_data
+        .iter()
+        .flat_map(|unit| {
+            // Adding rounded numbers may cause ready > start,
+            // so cap with unit.start here to be defensive.
+            let ready = ready_time(unit).unwrap_or(unit.start).min(unit.start);
+
+            [
+                (ready, State::Ready, unit.i),
+                (unit.start, State::Start, unit.i),
+                (unit.start + unit.duration, State::End, unit.i),
+            ]
+        })
+        .collect();
+
+    events.sort_by(|a, b| {
+        a.0.partial_cmp(&b.0)
+            .unwrap()
+            .then_with(|| a.1.cmp(&b.1))
+            .then_with(|| a.2.cmp(&b.2))
+    });
+
+    let mut concurrency: Vec<Concurrency> = Vec::new();
+    let mut inactive: HashSet<u64> = unit_data.iter().map(|unit| unit.i).collect();
+    let mut waiting: HashSet<u64> = HashSet::new();
+    let mut active: HashSet<u64> = HashSet::new();
+
+    for (t, state, unit_id) in events {
+        match state {
+            State::Ready => {
+                inactive.remove(&unit_id);
+                waiting.insert(unit_id);
+                active.remove(&unit_id);
+            }
+            State::Start => {
+                inactive.remove(&unit_id);
+                waiting.remove(&unit_id);
+                active.insert(unit_id);
+            }
+            State::End => {
+                inactive.remove(&unit_id);
+                waiting.remove(&unit_id);
+                active.remove(&unit_id);
+            }
+        }
+
+        let record = Concurrency {
+            t,
+            active: active.len(),
+            waiting: waiting.len(),
+            inactive: inactive.len(),
+        };
+
+        if let Some(last) = concurrency.last_mut()
+            && last.t == t
+        {
+            // We don't want to draw long vertical lines at the same timestamp,
+            // so we keep only the latest state.
+            *last = record;
+        } else {
+            concurrency.push(record);
+        }
+    }
+
+    concurrency
+}
+
 /// Aggregates section timing information from individual compilation sections.
 ///
 /// We can have a bunch of situations here.