perf: Parallelize task hash computation across topological waves (#11969)

## Summary

Moves task hash computation out of the serial dispatch loop and into a
parallel pre-computation phase using rayon. Tasks are grouped into
topological waves by dependency depth, and each wave is hashed
concurrently. Also parallelizes external dependency hash precomputation.

Previously, every task's hash and execution environment were computed
one-at-a-time inside the `while let` visitor dispatch loop. With ~1700
tasks at ~160μs each, this was ~270ms of serial work on the critical
path. Now, the same work completes in ~53ms using rayon's thread pool.

Also cleans up `Engine::dependencies()`/`dependents()` to return `Vec`
instead of `HashSet`, and `calculate_dependency_hashes` to take a
`&[&TaskNode]` slice, since all callers only iterate — no set operations
are performed.

## Benchmarks

Profiled with `--profile` across three monorepos of varying size (full
cache hit scenario, median of 5 warm runs):

| Repo | Tasks | Baseline | After | Delta |
|------|-------|----------|-------|-------|
| Small (~5 packages) | 15 | 170ms | 169ms | ~flat |
| Medium (~125 packages) | 700 | 876ms | 876ms | ~flat |
| Large (~1000 packages) | 1700 | 867ms | 775ms | **-92ms (-10.6%)** |

`queue_task` self-time in the large repo dropped from **268ms to 78ms**
(71% reduction). The improvement scales with task count.

## Changes

- **`crates/turborepo-lib/src/task_graph/visitor/mod.rs`** — Added
`precompute_task_hashes()` which groups tasks into topological waves and
hashes each wave in parallel via rayon. The dispatch loop now looks up
pre-computed hashes instead of computing them inline.
- **`crates/turborepo-engine/src/lib.rs`** — `dependencies()`,
`dependents()`, and `neighbors()` return `Vec` instead of `HashSet`.
Updated `EngineInfo` impl's associated type accordingly.
- **`crates/turborepo-task-hash/src/lib.rs`** — `calculate_task_hash`
takes `&[&TaskNode]` instead of `HashSet<&TaskNode>`.
`calculate_dependency_hashes` takes a slice.
`precompute_external_deps_hashes` uses `par_iter` instead of a
sequential loop.

Author: anthonyshew

crates/turborepo-engine/src/lib.rs

@@ -417,11 +417,11 @@ impl<T: TaskDefinitionInfo + Clone> Engine<Built, T> {
         }
     }
 
-    pub fn dependencies(&self, task_id: &TaskId) -> Option<HashSet<&TaskNode>> {
+    pub fn dependencies(&self, task_id: &TaskId) -> Option<Vec<&TaskNode>> {
         self.neighbors(task_id, petgraph::Direction::Outgoing)
     }
 
-    pub fn dependents(&self, task_id: &TaskId) -> Option<HashSet<&TaskNode>> {
+    pub fn dependents(&self, task_id: &TaskId) -> Option<Vec<&TaskNode>> {
         self.neighbors(task_id, petgraph::Direction::Incoming)
     }
 
@@ -468,7 +468,7 @@ impl<T: TaskDefinitionInfo + Clone> Engine<Built, T> {
         &self,
         task_id: &TaskId,
         direction: petgraph::Direction,
-    ) -> Option<HashSet<&TaskNode>> {
+    ) -> Option<Vec<&TaskNode>> {
         let index = self.task_lookup.get(task_id)?;
         Some(
             self.task_graph
@@ -521,7 +521,7 @@ impl<T: TaskDefinitionInfo + Clone> Engine<Built, T> {
 // definitions and dependency information needed for run summaries.
 impl EngineInfo for Engine<Built, TaskDefinition> {
     type TaskIter<'a> = std::iter::FilterMap<
-        std::collections::hash_set::IntoIter<&'a TaskNode>,
+        std::vec::IntoIter<&'a TaskNode>,
         fn(&'a TaskNode) -> Option<&'a TaskId<'static>>,
     >;
 

crates/turborepo-lib/src/task_graph/visitor/mod.rs

@@ -3,7 +3,7 @@ mod exec;
 
 use std::{
     borrow::Cow,
-    collections::HashSet,
+    collections::{HashMap, HashSet},
     io::Write,
     sync::{Arc, Mutex},
 };
@@ -184,6 +184,124 @@ impl<'a> Visitor<'a> {
         }
     }
 
+    /// Pre-compute task hashes and execution environments for all tasks in
+    /// parallel. Tasks are processed in topological waves so dependency
+    /// hashes are always available when needed. Returns a map from TaskId
+    /// to (hash, execution_env).
+    fn precompute_task_hashes(
+        &self,
+        engine: &Engine,
+        telemetry: &GenericEventBuilder,
+    ) -> Result<HashMap<TaskId<'static>, (String, EnvironmentVariableMap)>, Error> {
+        use petgraph::algo::toposort;
+        use rayon::prelude::*;
+        use turborepo_engine::TaskNode;
+
+        let graph = engine.task_graph();
+        let mut sorted = toposort(graph, None).map_err(|_| Error::MissingDefinition)?;
+        // toposort returns dependents before dependencies (edges point
+        // dependent→dependency via Outgoing). Reverse so dependencies
+        // come first.
+        sorted.reverse();
+
+        // Compute depth (topological level) for each node so we can process
+        // independent tasks in parallel within each wave. Dependencies
+        // (Outgoing neighbors) must have lower depth.
+        let mut depth: HashMap<petgraph::graph::NodeIndex, usize> = HashMap::new();
+        for &node_idx in &sorted {
+            let max_dep_depth = graph
+                .neighbors_directed(node_idx, petgraph::Direction::Outgoing)
+                .filter_map(|dep| depth.get(&dep))
+                .max()
+                .copied();
+            let d = match max_dep_depth {
+                Some(dd) => dd + 1,
+                None => 0,
+            };
+            depth.insert(node_idx, d);
+        }
+
+        let max_depth = depth.values().max().copied().unwrap_or(0);
+
+        // Group task nodes by depth level.
+        let mut waves: Vec<Vec<petgraph::graph::NodeIndex>> = vec![Vec::new(); max_depth + 1];
+        for &node_idx in &sorted {
+            let d = depth[&node_idx];
+            waves[d].push(node_idx);
+        }
+
+        let results: Arc<Mutex<HashMap<TaskId<'static>, (String, EnvironmentVariableMap)>>> =
+            Arc::new(Mutex::new(HashMap::with_capacity(sorted.len())));
+
+        // Process each wave in parallel. Within a wave, all dependencies
+        // have already been hashed in earlier waves.
+        for wave in &waves {
+            type HashResult =
+                Result<Option<(TaskId<'static>, String, EnvironmentVariableMap)>, Error>;
+            let wave_results: Vec<HashResult> = wave
+                .par_iter()
+                .map(|&node_idx| {
+                    let node = &graph[node_idx];
+                    let TaskNode::Task(task_id) = node else {
+                        return Ok(None);
+                    };
+
+                    let package_name = PackageName::from(task_id.package());
+                    let workspace_info = self
+                        .package_graph
+                        .package_info(&package_name)
+                        .ok_or_else(|| Error::MissingPackage {
+                            package_name: package_name.clone(),
+                            task_id: task_id.clone(),
+                        })?;
+
+                    let task_definition = engine
+                        .task_definition(task_id)
+                        .ok_or(Error::MissingDefinition)?;
+
+                    let task_env_mode = task_definition.env_mode.unwrap_or(self.global_env_mode);
+
+                    let dependency_set = engine
+                        .dependencies(task_id)
+                        .ok_or(Error::MissingDefinition)?;
+
+                    let package_task_event =
+                        PackageTaskEventBuilder::new(task_id.package(), task_id.task())
+                            .with_parent(telemetry);
+                    package_task_event.track_env_mode(&task_env_mode.to_string());
+
+                    let task_hash_telemetry = package_task_event.child();
+                    let task_hash = self.task_hasher.calculate_task_hash(
+                        task_id,
+                        task_definition,
+                        task_env_mode,
+                        workspace_info,
+                        &dependency_set,
+                        task_hash_telemetry,
+                    )?;
+
+                    let execution_env =
+                        self.task_hasher
+                            .env(task_id, task_env_mode, task_definition)?;
+
+                    Ok(Some((task_id.clone(), task_hash, execution_env)))
+                })
+                .collect();
+
+            let mut map = results.lock().expect("precompute lock poisoned");
+            for result in wave_results {
+                if let Some((task_id, hash, env)) = result? {
+                    map.insert(task_id, (hash, env));
+                }
+            }
+        }
+
+        Ok(Arc::try_unwrap(results)
+            .expect("all wave references dropped")
+            .into_inner()
+            .expect("mutex not poisoned"))
+    }
+
     #[tracing::instrument(skip_all)]
     pub async fn visit(
         &self,
@@ -194,6 +312,16 @@ impl<'a> Visitor<'a> {
             self.color_cache.color_for_key(&task.to_string());
         }
 
+        // Pre-compute all task hashes and execution envs in parallel using
+        // rayon. Tasks are grouped into topological waves so that each
+        // task's dependency hashes are available before it is hashed.
+        // This replaces the per-task serial hashing that was inside the
+        // dispatch loop.
+        let precomputed = {
+            let _span = tracing::info_span!("precompute_task_hashes").entered();
+            self.precompute_task_hashes(&engine, telemetry)?
+        };
+
         let concurrency = self.run_opts.concurrency as usize;
         let (node_sender, mut node_stream) = mpsc::channel(concurrency);
 
@@ -222,12 +350,13 @@ impl<'a> Visitor<'a> {
                         task_id: info.clone(),
                     })?;
 
-            let package_task_event =
-                PackageTaskEventBuilder::new(info.package(), info.task()).with_parent(telemetry);
             let command = workspace_info.package_json.scripts.get(info.task());
 
             match command {
                 Some(cmd) if info.package() == ROOT_PKG_NAME && turbo_regex().is_match(cmd) => {
+                    let package_task_event =
+                        PackageTaskEventBuilder::new(info.package(), info.task())
+                            .with_parent(telemetry);
                     package_task_event.track_error(TrackedErrors::RecursiveError);
                     let (span, text) = cmd.span_and_text("package.json");
 
@@ -245,20 +374,11 @@ impl<'a> Visitor<'a> {
                 .task_definition(&info)
                 .ok_or(Error::MissingDefinition)?;
 
-            let task_env_mode = task_definition.env_mode.unwrap_or(self.global_env_mode);
-            package_task_event.track_env_mode(&task_env_mode.to_string());
-
-            let dependency_set = engine.dependencies(&info).ok_or(Error::MissingDefinition)?;
-
-            let task_hash_telemetry = package_task_event.child();
-            let task_hash = self.task_hasher.calculate_task_hash(
-                &info,
-                task_definition,
-                task_env_mode,
-                workspace_info,
-                dependency_set,
-                task_hash_telemetry,
-            )?;
+            // Look up pre-computed hash and env instead of computing them here.
+            let (task_hash, execution_env) = precomputed
+                .get(&info)
+                .ok_or(Error::MissingDefinition)?
+                .clone();
 
             debug!("task {} hash is {}", info, task_hash);
 
@@ -284,13 +404,6 @@ impl<'a> Visitor<'a> {
                     }));
                 }
                 false => {
-                    // Compute execution env only when we actually need it (not
-                    // during dry runs). The task_hasher is !Send so this must
-                    // happen in the dispatch loop rather than inside the spawned task.
-                    let execution_env =
-                        self.task_hasher
-                            .env(&info, task_env_mode, task_definition)?;
-
                     let takes_input = task_definition.interactive || task_definition.persistent;
                     let Some(mut exec_context) = factory.exec_context(
                         info.clone(),
@@ -301,11 +414,6 @@ impl<'a> Visitor<'a> {
                         self.task_access.clone(),
                     )?
                     else {
-                        // TODO(gsoltis): if/when we fix https://github.com/vercel/turborepo/issues/937
-                        // the following block should never get hit. In the meantime, keep it after
-                        // hashing so that downstream tasks can count on the hash existing
-                        //
-                        // bail if the script doesn't exist or is empty
                         continue;
                     };
 
@@ -319,6 +427,9 @@ impl<'a> Visitor<'a> {
 
                     let tracker = self.run_tracker.track_task(info.clone().into_owned());
                     let parent_span = Span::current();
+                    let package_task_event =
+                        PackageTaskEventBuilder::new(info.package(), info.task())
+                            .with_parent(telemetry);
                     let execution_telemetry = package_task_event.child();
 
                     tasks.push(tokio::spawn(async move {

crates/turborepo-task-hash/src/lib.rs

@@ -297,11 +297,14 @@ impl<'a, R: RunOptsHashInfo> TaskHasher<'a, R> {
         if self.run_opts.single_package() {
             return;
         }
-        for (name, info) in workspaces {
-            let hash = get_external_deps_hash(&info.transitive_dependencies);
-            self.external_deps_hash_cache
-                .insert(name.as_str().to_owned(), hash);
-        }
+        let ws: Vec<_> = workspaces.collect();
+        self.external_deps_hash_cache = ws
+            .par_iter()
+            .map(|(name, info)| {
+                let hash = get_external_deps_hash(&info.transitive_dependencies);
+                (name.as_str().to_owned(), hash)
+            })
+            .collect();
     }
 
     #[tracing::instrument(skip(self, task_definition, task_env_mode, workspace, dependency_set))]
@@ -311,7 +314,7 @@ impl<'a, R: RunOptsHashInfo> TaskHasher<'a, R> {
         task_definition: &T,
         task_env_mode: EnvMode,
         workspace: &PackageInfo,
-        dependency_set: HashSet<&TaskNode>,
+        dependency_set: &[&TaskNode],
         telemetry: PackageTaskEventBuilder,
     ) -> Result<String, Error> {
         let do_framework_inference = self.run_opts.framework_inference();
@@ -453,7 +456,7 @@ impl<'a, R: RunOptsHashInfo> TaskHasher<'a, R> {
     /// returns: Result<Vec<String, Global>, Error>
     fn calculate_dependency_hashes(
         &self,
-        dependency_set: HashSet<&TaskNode>,
+        dependency_set: &[&TaskNode],
     ) -> Result<Vec<Arc<str>>, Error> {
         let state = self
             .task_hash_tracker
@@ -462,7 +465,7 @@ impl<'a, R: RunOptsHashInfo> TaskHasher<'a, R> {
             .expect("hash tracker rwlock poisoned");
 
         let mut dependency_hash_list: Vec<Arc<str>> = Vec::with_capacity(dependency_set.len());
-        for dependency_task in &dependency_set {
+        for dependency_task in dependency_set {
             let TaskNode::Task(dependency_task_id) = dependency_task else {
                 continue;
             };