Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 130 additions & 39 deletions crates/turborepo-lib/src/task_graph/visitor/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ use turborepo_types::{EnvMode, ResolvedLogOrder, ResolvedLogPrefix};
use turborepo_ui::{sender::UISender, ColorConfig, ColorSelector};

use crate::{
engine::{Engine, ExecutionOptions},
engine::{Engine, ExecutionOptions, TaskNode},
microfrontends::MicrofrontendsConfigs,
opts::RunOpts,
run::{task_access::TaskAccess, RunCache},
Expand Down Expand Up @@ -216,15 +216,132 @@ impl<'a> Visitor<'a> {
/// 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_ready_task_hash(
&self,
engine: &Engine,
telemetry: &GenericEventBuilder,
task_id: &TaskId<'static>,
) -> Result<PrecomputedTask, Error> {
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(PrecomputedTask::Ready {
task_hash,
execution_env,
})
}

fn dependency_hashes_available(
&self,
engine: &Engine,
task_id: &TaskId<'static>,
) -> Result<bool, Error> {
let dependency_set = engine
.dependencies(task_id)
.ok_or(Error::MissingDefinition)?;
let task_hash_tracker = self.task_hasher.task_hash_tracker();

Ok(dependency_set.iter().all(|dependency| {
let TaskNode::Task(dependency_task_id) = dependency else {
return true;
};

task_hash_tracker.hash(dependency_task_id).is_some()
}))
}

fn precompute_unblocked_deferred_hashes(
&self,
engine: &Engine,
telemetry: &GenericEventBuilder,
precomputed: &mut HashMap<TaskId<'static>, PrecomputedTask>,
) -> Result<(), Error> {
use rayon::prelude::*;

loop {
let ready_to_hash = precomputed
.iter()
.filter_map(|(task_id, precomputed_task)| {
if !matches!(precomputed_task, PrecomputedTask::Deferred) {
return None;
}

let Some(task_definition) = engine.task_definition(task_id) else {
return Some(Err(Error::MissingDefinition));
};
if task_definition.inputs.has_jit_inputs() {
return None;
}

match self.dependency_hashes_available(engine, task_id) {
Ok(true) => Some(Ok(task_id.clone())),
Ok(false) => None,
Err(err) => Some(Err(err)),
}
})
.collect::<Result<Vec<_>, _>>()?;

if ready_to_hash.is_empty() {
return Ok(());
}

type HashResult = Result<(TaskId<'static>, PrecomputedTask), Error>;
let hash_results: Vec<HashResult> = ready_to_hash
.par_iter()
.map(|task_id| {
self.precompute_ready_task_hash(engine, telemetry, task_id)
.map(|precomputed_task| (task_id.clone(), precomputed_task))
})
.collect();

for result in hash_results {
let (task_id, precomputed_task) = result?;
precomputed.insert(task_id, precomputed_task);
}
}
}

fn precompute_task_hashes(
&self,
engine: &Engine,
telemetry: &GenericEventBuilder,
) -> Result<HashMap<TaskId<'static>, PrecomputedTask>, 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
Expand Down Expand Up @@ -273,19 +390,9 @@ impl<'a> Visitor<'a> {
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
Expand Down Expand Up @@ -316,32 +423,8 @@ impl<'a> Visitor<'a> {
return Ok(Some((task_id.clone(), PrecomputedTask::Deferred)));
}

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(),
PrecomputedTask::Ready {
task_hash,
execution_env,
},
)))
self.precompute_ready_task_hash(engine, telemetry, task_id)
.map(|precomputed_task| Some((task_id.clone(), precomputed_task)))
})
.collect();

Expand Down Expand Up @@ -528,6 +611,14 @@ impl<'a> Visitor<'a> {
break;
}
};
if let Err(err) = self.precompute_unblocked_deferred_hashes(
&engine,
telemetry,
&mut precomputed,
) {
dispatch_error = Some(err);
break;
}
(task_hash, execution_env)
}
}
Expand Down
2 changes: 2 additions & 0 deletions crates/turborepo/ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -194,6 +194,8 @@ The task graph visitor handles task execution:
engine dispatches the task, after its dependencies have completed and restored
any cached outputs. Tasks that depend on deferred tasks are also deferred so
their dependency hashes are available before their own hash is calculated.
Once a deferred task has a real hash, the visitor precomputes any unblocked
non-JIT descendants instead of waiting for each descendant to be dispatched.
- Creates `ExecContext` for each task
- Manages UI output and progress tracking
- Collects errors and execution information
Expand Down
66 changes: 66 additions & 0 deletions crates/turborepo/tests/run_caching.rs
Original file line number Diff line number Diff line change
Expand Up @@ -367,6 +367,72 @@ fn test_jit_dependency_defers_dependent_hashing() {
);
}

#[test]
fn test_jit_descendant_hashes_after_jit_hash_is_available() {
let tempdir = tempfile::tempdir().unwrap();
setup::setup_integration_test(tempdir.path(), "basic_monorepo", "npm@10.5.0", true).unwrap();

fs::write(tempdir.path().join("apps/my-app/marker.txt"), "before\n").unwrap();
fs::write(tempdir.path().join("apps/my-app/jit-input.txt"), "stable\n").unwrap();
fs::write(
tempdir.path().join("apps/my-app/package.json"),
r#"{
"name": "my-app",
"scripts": {
"generate": "node -e \"const fs = require('fs'); fs.mkdirSync('.generated', { recursive: true }); fs.writeFileSync('.generated/done.txt', 'done\\n'); fs.writeFileSync('marker.txt', 'after\\n')\"",
"build": "node -e \"const fs = require('fs'); fs.mkdirSync('.output', { recursive: true }); fs.writeFileSync('.output/marker.txt', fs.readFileSync('marker.txt'))\""
},
"dependencies": {
"util": "*"
}
}
"#,
)
.unwrap();

fs::write(
tempdir.path().join("turbo.json"),
r#"{
"$schema": "https://turborepo.dev/schema.json",
"tasks": {
"generate": {
"inputs": ["$TURBO_JIT$/jit-input.txt"],
"outputs": [".generated/**"]
},
"build": {
"dependsOn": ["generate"],
"inputs": ["marker.txt"],
"outputs": [".output/**"]
}
}
}
"#,
)
.unwrap();

let output = run_turbo(
tempdir.path(),
&["run", "build", "--filter=my-app", "--output-logs=none"],
);
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(
stdout.contains("0 cached, 2 total"),
"expected first run to execute both tasks, got:\n{stdout}"
);

fs::write(tempdir.path().join("apps/my-app/marker.txt"), "before\n").unwrap();

let output = run_turbo(
tempdir.path(),
&["run", "build", "--filter=my-app", "--output-logs=none"],
);
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(
stdout.contains("2 cached, 2 total"),
"expected JIT descendants to be hashed before the dependency command runs, got:\n{stdout}"
);
}

#[test]
fn test_gitignored_output_deletion_restores_from_cache() {
let tempdir = tempfile::tempdir().unwrap();
Expand Down
Loading