[turbopack] Enforce root attribute for strongly consistent reads and collectibles (#93114)

### What?

Replaces the silent "make root node" promotion in the turbo-tasks
backend with a panic that enforces tasks to already have the `root`
attribute before performing strongly consistent reads, reading task
collectibles, or removing collectibles.

### Why?

Previously, the backend would silently promote any non-root task to a
root node (aggregation number `u32::MAX`) when these operations were
requested. This masked incorrect task configuration — tasks that needed
root-level aggregation weren't explicitly declared as such, making it
harder to reason about the aggregation graph and hiding potential
performance issues from implicit promotions.

### How?

**Backend enforcement (3 locations):**
- **Strongly consistent read** (`backend/mod.rs`): Checks
`NativeFunction.is_root` on the target task. If it's a persistent task
without `root`, panics with both the target and reader task
descriptions.
- **Read task collectibles** (`backend/mod.rs`): Same check when reading
collectibles from a task.
- **Remove collectibles** (`operation/update_collectible.rs`): Same
check when removing collectibles (count < 0).

All three locations drop the task guard before panicking to avoid
deadlocking with `debug_get_task_description`.

**Macro support for `root` on methods:**
- `value_impl_macro.rs`: Now reads the `root` attribute from
`#[turbo_tasks::function(root)]` on inherent impl and trait impl methods
(previously hardcoded to `false`).
- `value_trait_macro.rs`: Same for trait default methods.

**`root` attribute additions:**
- All `#[turbo_tasks::function(operation)]` in test files, benchmarks,
and fuzz code
- Production operations in `crates/next-api/` that are read with strong
consistency
- Regular functions and methods in tests that use
`.strongly_consistent()`
- Trait methods in `value_impl` and `value_trait` blocks used with
strong consistency

**New test:**
- `non_root_task_panic.rs`: Verifies the panic fires when attempting a
strongly consistent read on a non-root operation task. Captures the
panic from the worker thread via a panic hook since the panic propagates
as a channel error to the test thread.

<!-- NEXT_JS_LLM_PR -->

---------

Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]>
This commit is contained in:
Tobias Koppers
2026-05-12 18:56:02 +00:00
committed by GitHub
co-authored by Claude Opus 4.7
parent 4bad5e74fe
commit 5d0dab5fc6
79 changed files with 362 additions and 295 deletions
+6 -6
View File
@@ -50,7 +50,7 @@ pub struct NextDynamicGraphs(Vec<ResolvedVc<NextDynamicGraph>>);
#[turbo_tasks::value_impl]
impl NextDynamicGraphs {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn new_operation(
graphs: ResolvedVc<ModuleGraph>,
is_single_page: bool,
@@ -71,7 +71,7 @@ impl NextDynamicGraphs {
Ok(Self(next_dynamic).cell())
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
pub async fn new(graphs: ResolvedVc<ModuleGraph>, is_single_page: bool) -> Result<Vc<Self>> {
// TODO get rid of this function once everything inside of
// `get_global_information_for_endpoint_inner` calls `take_collectibles()` when needed
@@ -248,7 +248,7 @@ pub struct ServerActionsGraphs(Vec<ResolvedVc<ServerActionsGraph>>);
#[turbo_tasks::value_impl]
impl ServerActionsGraphs {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn new_operation(
graphs: ResolvedVc<ModuleGraph>,
is_single_page: bool,
@@ -269,7 +269,7 @@ impl ServerActionsGraphs {
Ok(Self(server_actions).cell())
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
pub async fn new(graphs: ResolvedVc<ModuleGraph>, is_single_page: bool) -> Result<Vc<Self>> {
// TODO get rid of this function once everything inside of
// `get_global_information_for_endpoint_inner` calls `take_collectibles()` when needed
@@ -426,7 +426,7 @@ pub struct ClientReferencesGraphs(Vec<ResolvedVc<ClientReferencesGraph>>);
#[turbo_tasks::value_impl]
impl ClientReferencesGraphs {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn new_operation(
graphs: ResolvedVc<ModuleGraph>,
is_single_page: bool,
@@ -447,7 +447,7 @@ impl ClientReferencesGraphs {
Ok(Self(client_references).cell())
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
pub async fn new(graphs: ResolvedVc<ModuleGraph>, is_single_page: bool) -> Result<Vc<Self>> {
// TODO get rid of this function once everything inside of
// `get_global_information_for_endpoint_inner` calls `take_collectibles()` when needed
+3 -3
View File
@@ -33,7 +33,7 @@ pub struct EntrypointsOperation {
/// Removes issues and effects from the top-level `entrypoints` operation so that they're not
/// duplicated across many different individual entrypoints or routes.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn entrypoints_without_collectibles_operation(
entrypoints: OperationVc<Entrypoints>,
) -> Result<Vc<Entrypoints>> {
@@ -45,7 +45,7 @@ async fn entrypoints_without_collectibles_operation(
#[turbo_tasks::value_impl]
impl EntrypointsOperation {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn new(entrypoints: OperationVc<Entrypoints>) -> Result<Vc<Self>> {
let e = entrypoints.connect().await?;
let entrypoints = entrypoints_without_collectibles_operation(entrypoints);
@@ -143,7 +143,7 @@ pub struct OptionEndpoint(Option<ResolvedVc<Box<dyn Endpoint>>>);
/// Given a selector and the `Entrypoints` operation that it comes from, connect the operation and
/// return an `OperationVc` containing the selected value. The returned operation will keep the
/// entire `Entrypoints` operation alive.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn pick_endpoint(
op: OperationVc<Entrypoints>,
selector: EndpointSelector,
+9 -9
View File
@@ -464,7 +464,7 @@ pub struct ProjectContainer {
#[turbo_tasks::value_impl]
impl ProjectContainer {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub fn new_operation(name: RcStr, dev: bool) -> Result<Vc<Self>> {
Ok(ProjectContainer {
name,
@@ -481,17 +481,17 @@ impl ProjectContainer {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_operation(project: ResolvedVc<ProjectContainer>) -> Vc<Project> {
project.project()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_fs_operation(project: ResolvedVc<Project>) -> Vc<DiskFileSystem> {
project.project_fs()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn output_fs_operation(project: ResolvedVc<Project>) -> Vc<DiskFileSystem> {
project.project_fs()
}
@@ -606,7 +606,7 @@ impl ProjectContainer {
}
this.options_state.set(Some(options));
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_from_container_operation(
container: OperationVc<ProjectContainer>,
) -> Vc<Project> {
@@ -654,7 +654,7 @@ impl ProjectContainer {
// to upgrade the `ResolvedVc` to an `OperationVc`. This is mostly okay
// because we can assume the `ProjectContainer` was originally resolved with
// strong consistency, and is rarely updated.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_container_operation_hack(
container: ResolvedVc<ProjectContainer>,
) -> Vc<ProjectContainer> {
@@ -1508,7 +1508,7 @@ impl Project {
/// Computes the whole app module graph, dropping issues in development mode so that
/// individual routes don't each report every issue from the shared graph.
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
pub async fn whole_app_module_graphs(
self: ResolvedVc<Self>,
) -> Result<Vc<BaseAndFullModuleGraph>> {
@@ -2408,7 +2408,7 @@ impl Project {
// sessions.
let _ = session;
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn hmr_version_operation(
this: ResolvedVc<Project>,
chunk_name: RcStr,
@@ -2726,7 +2726,7 @@ async fn any_output_changed(
Ok(Vc::<Completions>::cell(completions).completed())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn all_assets_from_entries_operation(
operation: OperationVc<OutputAssets>,
) -> Result<Vc<ExpandedOutputAssets>> {
@@ -133,7 +133,7 @@ pub async fn expand_outputs(
#[turbo_tasks::value_impl]
impl Asset for AssetHashesManifestAsset {
#[turbo_tasks::function(root)]
#[turbo_tasks::function]
async fn content(&self) -> Result<Vc<AssetContent>> {
let output_assets = expand_outputs(*self.project, self.asset_root.clone()).await?;
+3 -3
View File
@@ -220,7 +220,7 @@ async fn endpoint_output_assets_operation(
Ok(*output.connect().await?.output_assets)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn endpoint_write_to_disk_operation(
endpoint: OperationVc<OptionEndpoint>,
) -> Result<Vc<EndpointOutputPaths>> {
@@ -231,7 +231,7 @@ pub async fn endpoint_write_to_disk_operation(
})
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn endpoint_server_changed_operation(
endpoint: OperationVc<OptionEndpoint>,
) -> Result<Vc<Completion>> {
@@ -242,7 +242,7 @@ pub async fn endpoint_server_changed_operation(
})
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn endpoint_client_changed_operation(
endpoint: OperationVc<OptionEndpoint>,
) -> Result<Vc<Completion>> {
+2 -2
View File
@@ -267,7 +267,7 @@ type GetEntriesResultT = Vec<(FileSystemPath, ResolvedVc<Box<dyn OutputAsset>>)>
#[turbo_tasks::value(transparent)]
struct GetEntriesResult(GetEntriesResultT);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_entries(assets: OperationVc<ExpandedOutputAssets>) -> Result<Vc<GetEntriesResult>> {
let assets_ref = assets.connect().await?;
let entries = assets_ref
@@ -281,7 +281,7 @@ async fn get_entries(assets: OperationVc<ExpandedOutputAssets>) -> Result<Vc<Get
Ok(Vc::cell(entries))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn compute_entry_operation(
map: ResolvedVc<VersionedContentMap>,
assets_operation: OperationVc<ExpandedOutputAssets>,
+4 -4
View File
@@ -49,7 +49,7 @@ pub async fn main_inner(
tracing::info!("collecting endpoints");
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_entrypoints_operation(project: ResolvedVc<ProjectContainer>) -> Vc<Entrypoints> {
project.entrypoints()
}
@@ -248,7 +248,7 @@ pub async fn render_routes(
async fn endpoint_write_to_disk_with_apply(
endpoint: ResolvedVc<Box<dyn Endpoint>>,
) -> Result<ReadRef<EndpointOutputPaths>> {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn inner_operation(endpoint: ResolvedVc<Box<dyn Endpoint>>) -> Vc<EndpointOutputPaths> {
// we must wrap this in an operation so we can get the Effects collectibles
endpoint_write_to_disk(*endpoint)
@@ -260,7 +260,7 @@ async fn endpoint_write_to_disk_with_apply(
effects: Effects,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn inner_operation_with_effects(
endpoint: ResolvedVc<Box<dyn Endpoint>>,
) -> Result<Vc<WithEffects>> {
@@ -291,7 +291,7 @@ async fn hmr(
tracing::info!("HMR...");
let session = TransientInstance::new(());
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_hmr_chunk_names_operation(project: ResolvedVc<ProjectContainer>) -> Vc<Vec<RcStr>> {
project.hmr_chunk_names(HmrTarget::Client)
}
@@ -20,7 +20,7 @@ pub struct WriteAnalyzeResult {
pub effects: Arc<Effects>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn write_analyze_data_with_issues_operation(
project: ResolvedVc<ProjectContainer>,
app_dir_only: bool,
@@ -34,7 +34,7 @@ pub async fn write_analyze_data_with_issues_operation(
Ok(WriteAnalyzeResult { issues, effects }.cell())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn write_analyze_data_with_issues_operation_inner(
project: ResolvedVc<ProjectContainer>,
app_dir_only: bool,
@@ -119,7 +119,7 @@ struct WrittenEndpointWithIssues {
effects: Arc<Effects>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_written_endpoint_with_issues_operation(
endpoint_op: OperationVc<OptionEndpoint>,
) -> Result<Vc<WrittenEndpointWithIssues>> {
@@ -222,7 +222,7 @@ impl PartialEq for EndpointIssuesAndDiags {
impl Eq for EndpointIssuesAndDiags {}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn subscribe_issues_and_diags_operation(
endpoint_op: OperationVc<OptionEndpoint>,
should_include_issues: bool,
@@ -614,7 +614,7 @@ pub fn project_new(
let result = tt
.clone()
.run(async move {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_node_root_path_operation(
container: ResolvedVc<ProjectContainer>,
) -> Vc<FileSystemPath> {
@@ -995,7 +995,7 @@ struct EntrypointsWithIssues {
effects: Arc<Effects>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_entrypoints_with_issues_operation(
container: ResolvedVc<ProjectContainer>,
) -> Result<Vc<EntrypointsWithIssues>> {
@@ -1012,7 +1012,7 @@ async fn get_entrypoints_with_issues_operation(
.cell())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_container_entrypoints_operation(
// the container is a long-lived object with internally mutable state, there's no risk of it
// becoming stale
@@ -1207,7 +1207,7 @@ async fn invalidate_deferred_entry_source_dirs_after_callback(
#[turbo_tasks::value(cell = "new", eq = "manual")]
struct ProjectInfo(Option<FileSystemPath>, DiskFileSystem);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn project_info_operation(
container: ResolvedVc<ProjectContainer>,
) -> Result<Vc<ProjectInfo>> {
@@ -1326,7 +1326,7 @@ pub async fn project_write_all_entrypoints_to_disk(
let container = project.container;
let tt = ctx.turbo_tasks();
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn has_deferred_entrypoints_operation(
container: ResolvedVc<ProjectContainer>,
) -> Result<Vc<bool>> {
@@ -1366,7 +1366,7 @@ pub async fn project_write_all_entrypoints_to_disk(
#[turbo_tasks::value]
struct DeferredEntrypointInfo(ReadRef<Entrypoints>, ReadRef<Vec<RcStr>>);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn deferred_entrypoint_info_operation(
container: ResolvedVc<ProjectContainer>,
) -> Result<Vc<DeferredEntrypointInfo>> {
@@ -1546,7 +1546,7 @@ pub async fn project_write_all_entrypoints_to_disk(
})
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_all_written_entrypoints_with_issues_operation(
container: ResolvedVc<ProjectContainer>,
app_dir_only: bool,
@@ -1568,7 +1568,7 @@ async fn get_all_written_entrypoints_with_issues_operation(
.cell())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn all_entrypoints_write_to_disk_operation(
project: ResolvedVc<ProjectContainer>,
app_dir_only: bool,
@@ -1612,7 +1612,7 @@ async fn output_assets_for_single_emit_operation(
Ok(Vc::cell(merged_output_assets.into_iter().collect()))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn emit_all_output_assets_once_operation(
container: ResolvedVc<ProjectContainer>,
app_dir_only: bool,
@@ -1629,7 +1629,7 @@ async fn emit_all_output_assets_once_operation(
Ok(container.entrypoints())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn emit_all_output_assets_once_with_issues_operation(
container: ResolvedVc<ProjectContainer>,
app_dir_only: bool,
@@ -1803,7 +1803,7 @@ struct HmrUpdateWithIssues {
effects: Arc<Effects>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_hmr_update_operation(
project: ResolvedVc<Project>,
chunk_name: RcStr,
@@ -1813,7 +1813,7 @@ fn project_hmr_update_operation(
project.hmr_update(chunk_name, target, *state)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn hmr_update_with_issues_operation(
project: ResolvedVc<Project>,
chunk_name: RcStr,
@@ -1943,7 +1943,7 @@ struct HmrChunkNamesWithIssues {
effects: Arc<Effects>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn project_hmr_chunk_names_operation(
container: ResolvedVc<ProjectContainer>,
target: HmrTarget,
@@ -1951,7 +1951,7 @@ fn project_hmr_chunk_names_operation(
container.hmr_chunk_names(target)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_hmr_chunk_names_with_issues_operation(
container: ResolvedVc<ProjectContainer>,
target: HmrTarget,
@@ -2257,7 +2257,7 @@ pub async fn get_source_map_rope(
Ok(map)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub fn get_source_map_rope_operation(
container: ResolvedVc<ProjectContainer>,
file_path: RcStr,
@@ -2265,7 +2265,7 @@ pub fn get_source_map_rope_operation(
get_source_map_rope(*container, file_path)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn project_trace_source_operation(
container: ResolvedVc<ProjectContainer>,
frame: StackFrame,
@@ -2392,7 +2392,7 @@ pub async fn project_get_source_for_asset(
let ctx = &project.turbopack_ctx;
ctx.turbo_tasks()
.run(async move {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn source_content_operation(
container: ResolvedVc<ProjectContainer>,
file_path: RcStr,
@@ -2480,7 +2480,7 @@ pub async fn project_write_analyze_data(
})
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_all_compilation_issues_inner_operation(
container: ResolvedVc<ProjectContainer>,
) -> Result<Vc<()>> {
@@ -2497,7 +2497,7 @@ async fn get_all_compilation_issues_inner_operation(
Ok(Vc::cell(()))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_all_compilation_issues_operation(
container: ResolvedVc<ProjectContainer>,
) -> Result<Vc<OperationResult>> {
@@ -2523,7 +2523,7 @@ pub async fn project_feature_usage(
.turbopack_ctx
.turbo_tasks()
.run_once(async move {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn project_feature_usage_operation(
container: ResolvedVc<ProjectContainer>,
) -> Result<Vc<ProjectFeatureUsageSummary>> {
@@ -71,7 +71,7 @@ pub fn scope_stress(c: &mut Criterion) {
/// This fills a rectagle from (0, 0) to (a, b) by
/// first filling (0, 0) to (a - 1, b) and then (0, 0) to (a, b - 1) recursively
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn rectangle_operation(a: u32, b: u32) -> Result<Vc<Completion>> {
if a > 0 {
rectangle_operation(a - 1, b).connect().await?;
@@ -169,7 +169,7 @@ fn create_state() -> Vc<Iteration> {
Vc::cell(State::new(0))
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn run_task_chain(
spec: Arc<Vec<TaskSpec>>,
iteration: Vc<Iteration>,
@@ -186,7 +186,7 @@ async fn run_task_chain(
Ok(Vc::cell(()))
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn run_task(
spec: Arc<Vec<TaskSpec>>,
iteration: Vc<Iteration>,
@@ -61,8 +61,8 @@ use crate::{
AggregationUpdateJob, AggregationUpdateQueue, ChildExecuteContext,
CleanupOldEdgesOperation, ConnectChildOperation, ExecuteContext, ExecuteContextImpl,
LeafDistanceUpdateQueue, Operation, OutdatedEdge, TaskGuard, TaskType,
connect_children, get_aggregation_number, get_uppers, is_root_node,
make_task_dirty_internal, prepare_new_children,
connect_children, get_aggregation_number, get_uppers, make_task_dirty_internal,
prepare_new_children,
},
snapshot_coordinator::{OperationGuard, SnapshotCoordinator},
storage::Storage,
@@ -519,36 +519,21 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
}
if matches!(options.consistency, ReadConsistency::Strong) {
// Ensure it's an root node
loop {
let aggregation_number = get_aggregation_number(&task);
if is_root_node(aggregation_number) {
break;
}
if task
.get_persistent_task_type()
.is_some_and(|t| !t.native_fn.is_root)
{
drop(task);
drop(reader_task);
{
let _span = tracing::trace_span!(
"make root node for strongly consistent read",
task = self.debug_get_task_description(task_id)
panic!(
"Strongly consistent read of non-root task {} (reader: {}). The `root` \
attribute is missing on the task.",
self.debug_get_task_description(task_id),
reader.map_or_else(
|| "unknown".to_string(),
|r| self.debug_get_task_description(r)
)
.entered();
AggregationUpdateQueue::run(
AggregationUpdateJob::UpdateAggregationNumber {
task_id,
base_aggregation_number: u32::MAX,
distance: None,
},
&mut ctx,
);
}
(task, reader_task) = if let Some(reader_id) = need_reader_task {
// TODO(sokra): see comment above
let (task, reader) = ctx.task_pair(task_id, reader_id, TaskDataCategory::All);
(task, Some(reader))
} else {
(ctx.task(task_id, TaskDataCategory::All), None)
}
);
}
let is_dirty = task.is_dirty();
@@ -1973,6 +1958,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
immutable = tracing::field::Empty,
new_output = tracing::field::Empty,
output_dependents = tracing::field::Empty,
aggregation_number = tracing::field::Empty,
stale = tracing::field::Empty,
)
.entered();
@@ -2217,7 +2203,11 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
if has_children {
// Prepare all new children
prepare_new_children(task_id, &mut task, &new_children, &mut queue);
let _aggregation_number =
prepare_new_children(task_id, &mut task, &new_children, &mut queue);
#[cfg(feature = "trace_task_details")]
span.record("aggregation_number", _aggregation_number);
// Filter actual new children
old_edges.extend(
@@ -2948,22 +2938,20 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
let mut collectibles = AutoMap::default();
{
let mut task = ctx.task(task_id, TaskDataCategory::All);
// Ensure it's an root node
loop {
let aggregation_number = get_aggregation_number(&task);
if is_root_node(aggregation_number) {
break;
}
if task
.get_persistent_task_type()
.is_some_and(|t| !t.native_fn.is_root)
{
drop(task);
AggregationUpdateQueue::run(
AggregationUpdateJob::UpdateAggregationNumber {
task_id,
base_aggregation_number: u32::MAX,
distance: None,
},
&mut ctx,
panic!(
"Reading collectibles of non-root task {} (reader: {}). The `root` attribute \
is missing on the task.",
self.debug_get_task_description(task_id),
reader_id.map_or_else(
|| "unknown".to_string(),
|r| self.debug_get_task_description(r)
)
);
task = ctx.task(task_id, TaskDataCategory::All);
}
for (collectible, count) in task.iter_aggregated_collectibles() {
if *count > 0 && collectible.collectible_type == collectible_type {
@@ -14,7 +14,7 @@ pub fn prepare_new_children(
parent_task: &mut impl TaskGuard,
new_children: &FxHashSet<TaskId>,
queue: &mut AggregationUpdateQueue,
) {
) -> u32 {
debug_assert!(!new_children.is_empty());
let children_count = new_children.len();
@@ -52,4 +52,6 @@ pub fn prepare_new_children(
});
}
};
future_parent_aggregation
}
@@ -8,7 +8,7 @@ use crate::{
TaskDataCategory,
operation::{
AggregatedDataUpdate, AggregationUpdateJob, AggregationUpdateQueue, ExecuteContext,
Operation, get_aggregation_number, is_root_node,
Operation,
},
storage_schema::TaskStorageAccessors,
},
@@ -25,31 +25,18 @@ impl UpdateCollectibleOperation {
mut ctx: impl ExecuteContext<'_>,
) {
let mut task = ctx.task(task_id, TaskDataCategory::All);
if count < 0 {
// Ensure it's an root node
loop {
let aggregation_number = get_aggregation_number(&task);
if is_root_node(aggregation_number) {
break;
}
drop(task);
{
let _span = tracing::trace_span!(
"make root node for removing collectible",
task = ctx.debug_get_task_description(task_id)
)
.entered();
AggregationUpdateQueue::run(
AggregationUpdateJob::UpdateAggregationNumber {
task_id,
base_aggregation_number: u32::MAX,
distance: None,
},
&mut ctx,
);
}
task = ctx.task(task_id, TaskDataCategory::All);
}
if count < 0
&& task
.get_persistent_task_type()
.is_some_and(|t| !t.native_fn.is_root)
{
drop(task);
panic!(
"Removing collectibles from non-root task {} (emitting task: {}). The `root` \
attribute is missing on the task.",
ctx.debug_get_task_description(task_id),
ctx.debug_get_task_description(task_id)
);
}
let mut queue = AggregationUpdateQueue::new();
let outdated = task.get_outdated_collectibles(&collectible).copied();
@@ -25,7 +25,7 @@ async fn test_all_in_one() {
.unwrap()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_all_in_one_operation(nonce: u32) -> Result<Vc<()>> {
let _ = nonce; // ensure the nonce is part of our cache key
@@ -20,7 +20,7 @@ async fn test_basic() {
.unwrap()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_basic_operation(nonce: u32) -> Result<Vc<()>> {
let _ = nonce; // ensure the nonce is part of our cache key
@@ -38,7 +38,7 @@ async fn test_graph_bug() {
.unwrap()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_graph_bug_operation(nonce: u32) -> Result<Vc<()>> {
let _ = nonce; // ensure the nonce is part of our cache key
@@ -190,7 +190,7 @@ async fn test_graph_bug_operation(nonce: u32) -> Result<Vc<()>> {
Ok(Vc::cell(()))
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn run_task(spec: Vc<TasksSpec>, task: u16) -> Result<Vc<()>> {
let spec_ref = spec.await?;
let task = &spec_ref[task as usize];
@@ -49,7 +49,7 @@ async fn test_graph_bug() {
.unwrap()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_graph_bug_operation(nonce: u32) -> Result<Vc<()>> {
let _ = nonce; // ensure the nonce is part of our cache key
@@ -98,7 +98,7 @@ fn create_iteration() -> Vc<Iteration> {
Vc::cell(State::new(0))
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn run_task_chain(
spec: Arc<Vec<TaskSpec>>,
iteration: Vc<Iteration>,
@@ -116,7 +116,7 @@ async fn run_task_chain(
Ok(Vc::cell(()))
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn run_task(
spec: Arc<Vec<TaskSpec>>,
iteration: Vc<Iteration>,
@@ -24,7 +24,7 @@ async fn test_functions() {
.unwrap()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_functions_operation(nonce: u32) -> Result<Vc<()>> {
let _ = nonce; // ensure the nonce is part of our cache key
@@ -77,7 +77,7 @@ async fn test_methods() {
.unwrap()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_methods_operation() -> Result<Vc<()>> {
assert_eq!(*Value::static_method().await?, 42);
assert_eq!(*Value::async_static_method().await?, 42);
@@ -138,7 +138,7 @@ async fn test_trait_methods() {
.unwrap()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_trait_methods_operation() -> Result<Vc<()>> {
assert_eq!(*Value::static_trait_method().await?, 42);
assert_eq!(*Value::async_static_trait_method().await?, 42);
@@ -165,7 +165,7 @@ async fn taking_collectibles_with_resolve() {
#[turbo_tasks::value(transparent)]
struct Collectibles(AutoSet<ResolvedVc<Box<dyn ValueToString>>>);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn my_collecting_function() -> Result<Vc<Thing>> {
let result_op = my_transitive_emitting_function(rcstr!(""), rcstr!(""));
let result_vc = result_op.connect();
@@ -174,7 +174,7 @@ async fn my_collecting_function() -> Result<Vc<Thing>> {
Ok(result_vc)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn my_collecting_function_indirect() -> Result<Vc<Thing>> {
let result_op = my_collecting_function();
let result_vc = result_op.connect();
@@ -186,7 +186,7 @@ async fn my_collecting_function_indirect() -> Result<Vc<Thing>> {
Ok(result_vc)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn my_multi_emitting_function() -> Result<Vc<Thing>> {
my_transitive_emitting_function(rcstr!(""), rcstr!("a"))
.connect()
@@ -198,14 +198,14 @@ async fn my_multi_emitting_function() -> Result<Vc<Thing>> {
Ok(Thing::cell(Thing(0)))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn my_transitive_emitting_function(key: RcStr, key2: RcStr) -> Result<Vc<Thing>> {
let _ = key2;
my_emitting_function(key).await?;
Ok(Thing::cell(Thing(0)))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn my_transitive_emitting_function_collectibles(
key: RcStr,
key2: RcStr,
@@ -216,7 +216,7 @@ fn my_transitive_emitting_function_collectibles(
))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn my_transitive_emitting_function_with_child_scope(
key: RcStr,
key2: RcStr,
@@ -248,7 +248,7 @@ fn my_transitive_emitting_function_with_thing(key: RcStr, _thing: Vc<Thing>) ->
Ok(())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn my_transitive_emitting_function_with_resolve(key: RcStr) -> Result<()> {
let _ = my_transitive_emitting_function_with_thing(key, get_thing(0));
Ok(())
@@ -10,7 +10,7 @@ use turbo_tasks_testing::{Registration, register, run_once};
static REGISTRATION: Registration = register!();
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn dbg_operation(value: ResolvedVc<Box<dyn ValueDebug>>) -> anyhow::Result<Vc<RcStr>> {
let trait_ref = value.into_trait_ref().await?;
let s = trait_ref.dbg().await?;
@@ -10,7 +10,7 @@ use turbo_tasks_testing::{Registration, register, run_once};
static REGISTRATION: Registration = register!();
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn to_string_operation(value: ResolvedVc<Box<dyn ValueToString>>) -> Vc<RcStr> {
value.to_string()
}
@@ -68,7 +68,7 @@ impl<T: TraceRawVcs> TraceRawVcs for WatchSenderTaskInput<T> {
}
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn spawns_detached(
notify: TransientInstance<NotifyTaskInput>,
sender: TransientInstance<WatchSenderTaskInput<Option<Vc<u32>>>>,
@@ -138,7 +138,7 @@ struct ChangingInput {
state: State<u32>,
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn spawns_detached_changing(
sender: TransientInstance<WatchSenderTaskInput<Option<Vc<u32>>>>,
changing_input_detached: Vc<ChangingInput>,
@@ -70,7 +70,7 @@ impl ValueToString for Collectible {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn inner_compute(input: ResolvedVc<ChangingInput>) -> Result<Vc<u32>> {
println!("start inner_compute");
let value = *input.await?.state.get();
@@ -88,7 +88,7 @@ async fn inner_compute(input: ResolvedVc<ChangingInput>) -> Result<Vc<u32>> {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn compute_operation(input: ResolvedVc<ChangingInput>) -> Result<Vc<Output>> {
println!("start compute");
let operation = inner_compute(input);
@@ -45,7 +45,7 @@ async fn test_emptied_cells() {
.unwrap();
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn get_state_operation() -> Vc<ChangingInput> {
ChangingInput {
state: State::new(0),
@@ -58,7 +58,7 @@ struct ChangingInput {
state: State<u32>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn compute_operation(input: ResolvedVc<ChangingInput>) -> Result<Vc<u32>> {
println!("compute_operation()");
let value = *inner_compute(*input).await?;
@@ -45,7 +45,7 @@ async fn test_emptied_cells_session_dependent() {
.unwrap();
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn get_state_operation() -> Vc<ChangingInput> {
ChangingInput {
state: State::new(0),
@@ -58,7 +58,7 @@ struct ChangingInput {
state: State<u32>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn compute_operation(input: ResolvedVc<ChangingInput>) -> Result<Vc<u32>> {
println!("compute_operation()");
let value = *inner_compute(*input).await?;
@@ -230,7 +230,7 @@ async fn eviction_dependency_chain() {
#[turbo_tasks::value(transparent)]
struct Step(State<u32>);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn create_state(initial: u32) -> Vc<Step> {
Step(State::new(initial)).cell()
}
@@ -241,7 +241,7 @@ struct Output {
random: u32,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn compute(input: ResolvedVc<Step>) -> Result<Vc<Output>> {
let value = *input.await?.get();
Ok(Output {
@@ -259,7 +259,7 @@ async fn double(input: ResolvedVc<Step>) -> Result<Vc<u32>> {
}
/// Outer function that depends on `double`
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn compute_chain(input: ResolvedVc<Step>) -> Result<Vc<Output>> {
let doubled = double(input);
let value = *doubled.connect().await?;
@@ -274,25 +274,25 @@ async fn compute_chain(input: ResolvedVc<Step>) -> Result<Vc<Output>> {
// Deep chain helpers — each layer reads the previous layer's output
// =========================================================================
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn add_one(input: ResolvedVc<Step>) -> Result<Vc<u32>> {
let value = *input.await?.get();
Ok(Vc::cell(value + 1))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn times_three(input: ResolvedVc<u32>) -> Result<Vc<u32>> {
let value = *input.await?;
Ok(Vc::cell(value * 3))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn plus_ten(input: ResolvedVc<u32>) -> Result<Vc<u32>> {
let value = *input.await?;
Ok(Vc::cell(value + 10))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn deep_chain(input: ResolvedVc<Step>) -> Result<Vc<Output>> {
// input → add_one → times_three → plus_ten → Output
// For input=10: (10+1)*3+10 = 43
@@ -324,7 +324,7 @@ struct SessionCounter {
/// Because this task is only resolved (not directly read) by the top-level
/// transient task, it has no transient dependents and is eligible for eviction
/// consideration — but should be blocked by the session-stateful flag.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn create_session_counter(initial: u32) -> Vc<SessionCounter> {
SessionCounter { count: initial }.cell()
}
@@ -333,7 +333,7 @@ fn create_session_counter(initial: u32) -> Vc<SessionCounter> {
/// doesn't need to resolve it directly (which would add a transient dependent
/// edge to create_session_counter, preventing us from testing the
/// session-stateful eviction gate).
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn read_session_counter(initial: u32) -> Result<Vc<Output>> {
let counter = create_session_counter(initial)
.resolve()
@@ -465,14 +465,14 @@ async fn eviction_transient_reader_invalidated() {
/// Adds an offset to a value — the offset parameter makes each call a unique
/// memoized task, creating truly independent intermediate tasks for fan-out.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn add_offset(input: ResolvedVc<Step>, offset: u32) -> Result<Vc<u32>> {
let value = *input.await?.get();
Ok(Vc::cell(value.wrapping_add(offset)))
}
/// Multiplies by a factor — unique per factor argument.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn multiply(input: ResolvedVc<u32>, factor: u32) -> Result<Vc<u32>> {
let value = *input.await?;
Ok(Vc::cell(value.wrapping_mul(factor)))
@@ -482,7 +482,7 @@ async fn multiply(input: ResolvedVc<u32>, factor: u32) -> Result<Vc<u32>> {
/// single state. Each chain uses unique arguments (offset/factor) so they
/// produce distinct memoized tasks — `width * 2` intermediate persistent tasks
/// that are candidates for eviction.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn fan_out(input: ResolvedVc<Step>, width: u32) -> Result<Vc<u32>> {
let mut total = 0u32;
for i in 0..width {
@@ -626,7 +626,7 @@ fn create_session_alive() -> Vc<SessionAlive> {
/// run_once — and is eligible for the eviction sweep. The `Step` input
/// gives us a knob to invalidate this reader (forcing re-read of the
/// writer's cell) without invalidating `create_session_alive` itself.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn read_session_alive_id(state: ResolvedVc<Step>) -> Result<Vc<AlivePtr>> {
let _state = *state.await?.get();
let v = create_session_alive().resolve().await?;
@@ -45,14 +45,14 @@ struct ConsumeResult {
random: u32,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn create_state_operation() -> Vc<Step> {
Step(State::new(0)).cell()
}
/// Produces a HashedValue from a state. The noise field changes each execution
/// but does not affect hash or equality.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn produce_hashed(input: ResolvedVc<Step>) -> Result<Vc<HashedValue>> {
let value = *input.await?.get();
let noise = EXECUTION_COUNTER.fetch_add(1, Ordering::Relaxed) as u64;
@@ -60,7 +60,7 @@ async fn produce_hashed(input: ResolvedVc<Step>) -> Result<Vc<HashedValue>> {
}
/// Consumes the HashedValue and records a random number to detect re-execution.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn consume_hashed(input: ResolvedVc<Step>) -> Result<Vc<ConsumeResult>> {
let hashed = produce_hashed(input).connect();
let v = hashed.await?;
@@ -68,7 +68,7 @@ async fn create_input() -> Result<Vc<ChangingInput>> {
.cell())
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn compute(input: Vc<ChangingInput>) -> Result<Vc<Value>> {
println!("compute()");
let input = input.await?;
@@ -76,14 +76,14 @@ async fn compute(input: Vc<ChangingInput>) -> Result<Vc<Value>> {
Ok(Value { value: *value }.cell())
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn read_input(input: Vc<Value>) -> Result<Vc<u32>> {
println!("read_input()");
let value = input.await?;
Ok(Vc::cell(value.value))
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn immutable_fn(input: Vc<Value>) -> Vc<u32> {
let _ = input;
println!("immutable_fn()");
@@ -92,13 +92,13 @@ fn immutable_fn(input: Vc<Value>) -> Vc<u32> {
#[turbo_tasks::value_impl]
impl Value {
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn read_self(&self) -> Vc<u32> {
println!("read_self()");
Vc::cell(self.value)
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn immutable_self_fn(self: Vc<Value>) -> Vc<u32> {
let _ = self;
println!("immutable_self_fn()");
@@ -12,7 +12,7 @@ static REGISTRATION: Registration = register!();
#[turbo_tasks::value(transparent)]
struct Step(State<u32>);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn create_state_operation() -> Vc<Step> {
Step(State::new(0)).cell()
}
@@ -70,7 +70,7 @@ async fn test_invalidation_map() {
#[turbo_tasks::value(transparent, cell = "keyed")]
struct Map(FxHashMap<String, u32>);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn create_map(step: ResolvedVc<Step>) -> Result<Vc<Map>> {
let step = step.await?;
let step_value = step.get();
@@ -89,7 +89,7 @@ struct GetValueResult {
random: u32,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_value(map: OperationVc<Map>, key: String) -> Result<Vc<GetValueResult>> {
let map = map.connect();
let value = map.get(&key).await?.as_deref().copied();
@@ -151,7 +151,7 @@ async fn test_invalidation_set() {
#[turbo_tasks::value(transparent, cell = "keyed")]
struct Set(FxHashSet<String>);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn create_set(step: ResolvedVc<Step>) -> Result<Vc<Set>> {
let step = step.await?;
let step_value = step.get();
@@ -170,7 +170,7 @@ struct HasValueResult {
random: u32,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn has_value(set: OperationVc<Set>, key: String) -> Result<Vc<HasValueResult>> {
let set = set.connect();
let value = set.contains_key(&key).await?;
@@ -0,0 +1,65 @@
#![feature(arbitrary_self_types)]
#![feature(arbitrary_self_types_pointers)]
#![allow(clippy::needless_return)]
use anyhow::Result;
use turbo_tasks::Vc;
use turbo_tasks_testing::{Registration, register, run_once_without_cache_check};
static REGISTRATION: Registration = register!();
#[turbo_tasks::value]
#[derive(Clone, Debug)]
struct Value {
value: u32,
}
// NOT marked with `root` — this is the key
#[turbo_tasks::function(operation)]
async fn non_root_operation() -> Result<Vc<Value>> {
Ok(Value { value: 42 }.cell())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_strongly_consistent_read_of_non_root_task_panics() {
// The panic happens on a worker thread. Capture the message via a panic hook.
let panic_message = std::sync::Arc::new(std::sync::Mutex::new(None::<String>));
let panic_message_clone = panic_message.clone();
let prev_hook = std::panic::take_hook();
std::panic::set_hook(Box::new(move |info| {
let msg = if let Some(s) = info.payload().downcast_ref::<String>() {
s.clone()
} else if let Some(s) = info.payload().downcast_ref::<&str>() {
s.to_string()
} else {
format!("{info}")
};
if msg.contains("root") {
*panic_message_clone.lock().unwrap() = Some(msg);
}
}));
// Spawn to catch the unwinding panic from the channel close on the test task.
let handle = tokio::task::spawn(async move {
run_once_without_cache_check(&REGISTRATION, async move {
non_root_operation().read_strongly_consistent().await
})
.await
});
// The spawned task will fail because the worker thread panics and the channel closes.
let _result = handle.await;
std::panic::set_hook(prev_hook);
let msg = panic_message.lock().unwrap().take();
assert!(
msg.is_some(),
"Expected a panic about missing `root` attribute on the worker thread"
);
let msg = msg.unwrap();
assert!(
msg.contains("root"),
"Panic message should mention 'root', got: {msg}"
);
}
@@ -7,18 +7,18 @@ use turbo_tasks_testing::{Registration, register, run};
static REGISTRATION: Registration = register!();
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn bare_op_fn() -> Vc<i32> {
Vc::cell(21)
}
// operations can take `ResolvedVc`s too (anything that's a `NonLocalValue`).
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn multiply(value: OperationVc<i32>, coefficient: ResolvedVc<i32>) -> Result<Vc<i32>> {
Ok(Vc::cell((*value.connect().await?) * (*coefficient.await?)))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn use_operations() -> Vc<i32> {
let twenty_one: OperationVc<i32> = bare_op_fn();
let forty_two: OperationVc<i32> = multiply(twenty_one, ResolvedVc::cell(2));
@@ -48,7 +48,7 @@ struct ValueContainer {
state: State<i32>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn make_state_operation() -> Vc<ValueContainer> {
ValueContainer {
state: State::new(0),
@@ -56,7 +56,7 @@ fn make_state_operation() -> Vc<ValueContainer> {
.cell()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn func2_operation(input: ResolvedVc<ValueContainer>) -> Result<Vc<Value>> {
let state = input.await?;
let value = state.state.get();
@@ -64,7 +64,7 @@ async fn func2_operation(input: ResolvedVc<ValueContainer>) -> Result<Vc<Value>>
Ok(func(*input, -*value))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn func_operation(input: ResolvedVc<ValueContainer>) -> Vc<Value> {
func(*input, 0)
}
@@ -77,7 +77,7 @@ impl Counter {
#[turbo_tasks::value_impl]
impl Counter {
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn get_value(&self) -> Result<Vc<CounterValue>> {
let mut lock = self.value.lock().unwrap();
lock.1.insert(get_invalidator().unwrap());
@@ -87,7 +87,7 @@ impl Counter {
#[turbo_tasks::value_impl]
impl CounterValue {
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn get_value(self: Vc<Self>) -> Vc<Self> {
self
}
@@ -103,7 +103,7 @@ struct VcHolder {
#[turbo_tasks::value_impl]
impl VcHolder {
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn compute(&self) -> Vc<Output> {
compute(*self.vc, *self.vc)
}
@@ -116,7 +116,7 @@ struct Output {
random_value: u32,
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn compute(input: Vc<ChangingInput>, input2: Vc<ChangingInput>) -> Result<Vc<Output>> {
let state_value = *input.await?.state.get();
let state_value2 = if state_value < 5 {
@@ -227,7 +227,7 @@ async fn inner_compute(input: Vc<ChangingInput>) -> Result<Vc<u32>> {
}
/// Outer task - depends on inner_compute
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn outer_compute(input: Vc<ChangingInput>) -> Result<Vc<DependencyOutput>> {
println!("outer_compute()");
let inner_result = *inner_compute(input).await?;
@@ -80,7 +80,7 @@ impl ValueToString for Collectible {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn inner_compute(
input: ResolvedVc<ChangingInput>,
input2: ResolvedVc<ChangingInput>,
@@ -103,7 +103,7 @@ async fn inner_compute2(input: Vc<ChangingInput>, innerness: u32) -> Result<Vc<u
Ok(Vc::cell(42))
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn compute(
input: ResolvedVc<ChangingInput>,
input2: ResolvedVc<ChangingInput>,
@@ -30,7 +30,7 @@ async fn test_conversion() -> Result<()> {
run_once(&REGISTRATION, move || {
nonce += 1;
async move {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_operation(nonce: u32) -> Result<Vc<()>> {
let _ = nonce; // ensure the nonce is part of our cache key
let unresolved: Vc<u32> = Vc::cell(42);
@@ -54,7 +54,7 @@ async fn test_cell_construction() -> Result<()> {
run_once(&REGISTRATION, move || {
nonce += 1;
async move {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_operation(nonce: u32) -> Result<Vc<()>> {
let _ = nonce;
let a: ResolvedVc<u32> = ResolvedVc::cell(42);
@@ -75,7 +75,7 @@ async fn test_resolved_vc_as_arg() -> Result<()> {
run_once(&REGISTRATION, move || {
nonce += 1;
async move {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_operation(nonce: u32) -> Result<Vc<()>> {
dbg!(nonce);
let _ = nonce;
@@ -97,7 +97,7 @@ async fn test_into_future() -> Result<()> {
run_once(&REGISTRATION, move || {
nonce += 1;
async move {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_operation(nonce: u32) -> Result<Vc<()>> {
let _ = nonce;
let mut resolved = ResolvedVc::cell(42);
@@ -35,7 +35,7 @@ async fn rectangle_stress() -> Result<()> {
/// This fills a rectagle from (0, 0) to (a, b) by
/// first filling (0, 0) to (a - 1, b) and then (0, 0) to (a, b - 1) recursively
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
async fn rectangle(a: u32, b: u32) -> Result<Vc<Completion>> {
if a > 0 {
rectangle(a - 1, b).await?;
@@ -14,7 +14,7 @@ struct Wrapper(Vec<u32>);
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_shrink_to_fit() -> Result<()> {
run_once(&REGISTRATION, || async {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn capacity_operation(wrapper: ResolvedVc<Wrapper>) -> Result<Vc<usize>> {
Ok(Vc::cell(wrapper.await?.capacity()))
}
@@ -16,7 +16,7 @@ struct Value {
value: u32,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn returns_value_operation() -> Result<Vc<Value>> {
Ok(Value { value: 42 }.cell())
}
@@ -72,7 +72,7 @@ async fn test_manual_mark_unmark_top_level_task() {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[should_panic]
async fn test_manual_mark_top_level_task_causes_error() {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn operation() -> Result<Vc<Value>> {
// Manually mark as top-level task
mark_top_level_task();
@@ -35,7 +35,7 @@ async fn test_trace_transient() {
assert!(message.contains(&EXPECTED_TRACE.to_string()));
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_trace_transient_operation(
arg1: ResolvedVc<()>,
arg2: ResolvedVc<u16>,
@@ -86,7 +86,7 @@ impl Counter {
#[turbo_tasks::value_trait]
trait CounterTrait {
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn get_value(&self) -> Vc<CounterValue>;
fn get_value_sync(&self) -> CounterValue;
@@ -94,7 +94,7 @@ trait CounterTrait {
#[turbo_tasks::value_impl]
impl CounterTrait for Counter {
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn get_value(&self) -> Result<Vc<CounterValue>> {
let mut lock = self.value.lock().unwrap();
lock.1.insert(get_invalidator().unwrap());
@@ -108,13 +108,13 @@ impl CounterTrait for Counter {
#[turbo_tasks::value_trait]
trait CounterValueTrait {
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn get_value(&self) -> Vc<CounterValue>;
}
#[turbo_tasks::value_impl]
impl CounterValueTrait for CounterValue {
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
fn get_value(self: Vc<Self>) -> Vc<Self> {
self
}
@@ -24,7 +24,7 @@ async fn test_transient_emit_from_persistent() {
assert!(message.contains(&EXPECTED_MSG.to_string()));
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn emit_incorrect_task_input_operation(value: IncorrectTaskInput) {
turbo_tasks::emit(ResolvedVc::upcast::<Box<dyn Number>>(value.0));
}
@@ -17,7 +17,7 @@ async fn test_transient_vc() -> Result<()> {
.await
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_transient_operation(transient_arg: TransientValue<i32>) -> Result<()> {
let called_with_transient = has_transient_arg(transient_arg);
let called_with_persistent = has_persistent_arg(123);
@@ -16,13 +16,13 @@ struct FmtTest {
count: u32,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn turbofmt_operation(value: ResolvedVc<FmtTest>) -> anyhow::Result<Vc<RcStr>> {
let s: RcStr = turbofmt!("prefix {} vc {}", 42u32, value).await?;
Ok(Vc::cell(s))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn turbobail_operation(value: ResolvedVc<FmtTest>) -> anyhow::Result<Vc<RcStr>> {
turbobail!("error: {} with {}", 42u32, value)
}
@@ -41,7 +41,7 @@ async fn basic_get() {
#[turbo_tasks::value]
struct FetchOutput(u16, RcStr);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn fetch_operation(url: RcStr) -> Result<Vc<FetchOutput>> {
let client_vc = FetchClientConfig::default().cell();
let response = &*client_vc
@@ -88,7 +88,7 @@ async fn sends_user_agent() {
#[turbo_tasks::value]
struct FetchOutput(u16, RcStr);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn fetch_operation(url: RcStr) -> Result<Vc<FetchOutput>> {
let client_vc = FetchClientConfig::default().cell();
let response = &*client_vc
@@ -137,7 +137,7 @@ async fn invalidation_does_not_invalidate() {
#[turbo_tasks::value]
struct FetchOutput(u16, RcStr, u16, RcStr);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn fetch_operation(url: RcStr) -> Result<Vc<FetchOutput>> {
let client_vc = FetchClientConfig::default().cell();
let response = &*client_vc
@@ -195,7 +195,7 @@ async fn errors_on_failed_connection() {
StyledString,
);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn fetch_operation(url: RcStr) -> Result<Vc<FetchOutput>> {
let client_vc = FetchClientConfig::default().cell();
let response_vc = client_vc.fetch(url.clone(), None);
@@ -259,7 +259,7 @@ async fn errors_on_404() {
StyledString,
);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn fetch_operation(url: RcStr) -> Result<Vc<FetchOutput>> {
let client_vc = FetchClientConfig::default().cell();
let response_vc = client_vc.fetch(url.clone(), None);
@@ -315,7 +315,7 @@ async fn client_cache() {
let server_url = RcStr::from(server.url());
// a simple fetch that should always succeed
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn simple_fetch_operation(server_url: RcStr, path: RcStr) -> anyhow::Result<()> {
let url = RcStr::from(format!("{}{}", server_url, path));
let response = match &*FetchClientConfig::default()
+11 -11
View File
@@ -2952,7 +2952,7 @@ mod tests {
use super::*;
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn extract_effects_operation(op: OperationVc<()>) -> anyhow::Result<Vc<Effects>> {
let _ = op.resolve().strongly_consistent().await?;
Ok(take_effects(op).await?.cell())
@@ -3060,7 +3060,7 @@ mod tests {
.unwrap();
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn assert_try_from_sys_path_operation(sys_root: RcStr) -> anyhow::Result<()> {
let sys_root = Path::new(sys_root.as_str());
let fs_vc = DiskFileSystem::new(
@@ -3163,7 +3163,7 @@ mod tests {
use super::extract_effects_operation;
use crate::{DiskFileSystem, FileSystem, FileSystemPath, LinkContent, LinkType};
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_write_link_effect_operation(
fs: ResolvedVc<DiskFileSystem>,
path: FileSystemPath,
@@ -3273,7 +3273,7 @@ mod tests {
const STRESS_TARGET_COUNT: usize = 20;
const STRESS_SYMLINK_COUNT: usize = 16;
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn disk_file_system_operation(fs_root: RcStr) -> Vc<DiskFileSystem> {
DiskFileSystem::new(rcstr!("test"), Vc::cell(fs_root))
}
@@ -3285,7 +3285,7 @@ mod tests {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn write_symlink_stress_batch(
fs: ResolvedVc<DiskFileSystem>,
symlinks_dir: FileSystemPath,
@@ -3450,7 +3450,7 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_denied_path_read() {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_operation(root: RcStr, denied_path: RcStr) -> anyhow::Result<()> {
let fs = DiskFileSystem::new_with_denied_paths(
rcstr!("test"),
@@ -3513,7 +3513,7 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_denied_path_read_dir() {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_operation(root: RcStr, denied_path: RcStr) -> anyhow::Result<()> {
let fs = DiskFileSystem::new_with_denied_paths(
rcstr!("test"),
@@ -3575,7 +3575,7 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_denied_path_read_glob() {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_operation(root: RcStr, denied_path: RcStr) -> anyhow::Result<()> {
let fs = DiskFileSystem::new_with_denied_paths(
rcstr!("test"),
@@ -3641,7 +3641,7 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_denied_path_write() {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn write_file_operation(
path: FileSystemPath,
contents: RcStr,
@@ -3656,7 +3656,7 @@ mod tests {
/// Writes the allowed file and captures effects to be applied at
/// the top level.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn write_allowed_file_operation(
root: RcStr,
denied_path: RcStr,
@@ -3675,7 +3675,7 @@ mod tests {
Ok(take_effects(write_op).await?.cell())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn test_denied_writes_operation(
root: RcStr,
denied_path: RcStr,
@@ -263,7 +263,7 @@ pub mod tests {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn assert_read_glob_basic_operation(path: RcStr) -> anyhow::Result<()> {
let fs = DiskFileSystem::new(rcstr!("temp"), Vc::cell(path));
let root = fs.root().await?;
@@ -306,7 +306,7 @@ pub mod tests {
Ok(())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn assert_read_glob_symlinks_operation(path: RcStr) -> anyhow::Result<()> {
let fs = DiskFileSystem::new(rcstr!("temp"), Vc::cell(path));
let root = fs.root().await?;
@@ -357,7 +357,7 @@ pub mod tests {
Ok(())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn assert_dead_symlink_read_glob_operation(path: RcStr) -> anyhow::Result<()> {
let fs =
Vc::upcast::<Box<dyn FileSystem>>(DiskFileSystem::new(rcstr!("temp"), Vc::cell(path)));
@@ -460,13 +460,13 @@ pub mod tests {
.unwrap();
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn delete(path: FileSystemPath) -> anyhow::Result<()> {
path.write(FileContent::NotFound.cell()).await?;
Ok(())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn write(path: FileSystemPath, contents: RcStr) -> anyhow::Result<()> {
path.write(
FileContent::Content(crate::File::from_bytes(contents.to_string().into_bytes())).cell(),
@@ -475,25 +475,25 @@ pub mod tests {
Ok(())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub fn track_star_star_glob(path: FileSystemPath) -> Vc<Completion> {
path.track_glob(Glob::new(rcstr!("**"), GlobOptions::default()), false)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn disk_file_system_root_operation(path: RcStr) -> Vc<FileSystemPath> {
let fs =
Vc::upcast::<Box<dyn FileSystem>>(DiskFileSystem::new(rcstr!("temp"), Vc::cell(path)));
fs.root()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn extract_effects_operation(op: OperationVc<()>) -> anyhow::Result<Vc<Effects>> {
let _ = op.resolve().strongly_consistent().await?;
Ok(take_effects(op).await?.cell())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn track_glob_operation(path: RcStr, glob: RcStr) -> anyhow::Result<()> {
let root = disk_file_system_root_operation(path)
.read_strongly_consistent()
@@ -503,7 +503,7 @@ pub mod tests {
Ok(())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn read_glob_operation(path: RcStr, glob: RcStr) -> anyhow::Result<()> {
let root = disk_file_system_root_operation(path)
.read_strongly_consistent()
@@ -89,7 +89,7 @@ impl SymlinkMode {
#[derive(Default, NonLocalValue, TraceRawVcs)]
struct PathInvalidations(#[turbo_tasks(trace_ignore)] Arc<Mutex<FxHashSet<RcStr>>>);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn extract_effects_operation(op: OperationVc<()>) -> anyhow::Result<Vc<Effects>> {
let _ = op.resolve().strongly_consistent().await?;
Ok(take_effects(op).await?.cell())
@@ -30,7 +30,7 @@ pub struct SymlinkStress {
duration_secs: u64,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn extract_effects_operation(op: OperationVc<()>) -> anyhow::Result<Vc<Effects>> {
let _ = op.resolve().strongly_consistent().await?;
Ok(take_effects(op).await?.cell())
@@ -22,7 +22,7 @@ fn one_unnamed_field(input: OneUnnamedField) -> Vc<Completion> {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn tests() {
run_once(&REGISTRATION, || async {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn equality_operation() -> Result<Vc<bool>> {
Ok(Vc::cell(ReadRef::ptr_eq(
&one_unnamed_field(OneUnnamedField(42)).await?,
@@ -22,7 +22,7 @@ async fn ignored_indexes() {
);
run_once(&REGISTRATION, || async {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn value_debug_format_operation() -> Result<Vc<RcStr>> {
let input = IgnoredIndexes(-1, 2, -3);
let debug = input.value_debug_format(usize::MAX).try_to_string().await?;
@@ -101,6 +101,7 @@ pub fn value_impl(args: TokenStream, input: TokenStream) -> TokenStream {
}
};
let is_self_used = func_args.operation.is_some() || is_self_used(block);
let is_root = func_args.root.is_some();
let Some(turbo_fn) = TurboFn::new(
sig,
@@ -112,7 +113,6 @@ pub fn value_impl(args: TokenStream, input: TokenStream) -> TokenStream {
// An error occurred while parsing the function signature.
};
};
let inline_function_ident = turbo_fn.inline_ident();
let (inline_signature, inline_block) = turbo_fn.inline_signature_and_block(block);
let inline_attrs = filter_inline_attributes(attrs.iter().copied());
@@ -123,7 +123,7 @@ pub fn value_impl(args: TokenStream, input: TokenStream) -> TokenStream {
is_method: turbo_fn.is_method(),
is_self_used,
filter_trait_call_args: None, // not a trait method
is_root: false,
is_root,
};
let native_function_ident = get_inherent_impl_function_ident(ty_ident, ident);
@@ -206,6 +206,7 @@ pub fn value_impl(args: TokenStream, input: TokenStream) -> TokenStream {
};
// operations are not currently compatible with methods
let is_self_used = func_args.operation.is_some() || is_self_used(block);
let is_root = func_args.root.is_some();
let Some(turbo_fn) = TurboFn::new(
sig,
@@ -238,7 +239,7 @@ pub fn value_impl(args: TokenStream, input: TokenStream) -> TokenStream {
is_method: turbo_fn.is_method(),
is_self_used,
filter_trait_call_args: turbo_fn.filter_trait_call_args(),
is_root: false,
is_root,
};
let native_function_ident =
@@ -181,6 +181,7 @@ pub fn value_trait(args: TokenStream, input: TokenStream) -> TokenStream {
}
let is_self_used = default.as_ref().map(is_self_used).unwrap_or(false);
let is_root = func_args.root.is_some();
let Some(turbo_fn) = TurboFn::new(
sig,
DefinitionContext::ValueTrait,
@@ -216,7 +217,7 @@ pub fn value_trait(args: TokenStream, input: TokenStream) -> TokenStream {
is_method: turbo_fn.is_method(),
is_self_used,
filter_trait_call_args: turbo_fn.filter_trait_call_args(),
is_root: false,
is_root,
};
let native_function_ident = get_trait_default_impl_function_ident(trait_ident, ident);
@@ -232,6 +233,7 @@ pub fn value_trait(args: TokenStream, input: TokenStream) -> TokenStream {
method_name: #method_name_str,
default_method: Some(&#native_function_ident),
index: #index,
is_root: #is_root,
},
});
default_methods.push(quote! { Some(&#native_function_ident) });
@@ -269,6 +271,7 @@ pub fn value_trait(args: TokenStream, input: TokenStream) -> TokenStream {
method_name: #method_name_str,
default_method: None,
index: #index,
is_root: #is_root,
},
});
default_methods.push(quote! { None });
@@ -271,6 +271,24 @@ turbo_registry!("Value", ValueType);
// Single-threaded during Lazy init.
pub(crate) fn register_all_trait_methods(_: &[&'static ValueType]) {
for entry in inventory::iter::<CollectableTraitMethods> {
for (i, impl_method) in entry.methods.iter().enumerate() {
let trait_method = &entry.trait_type.methods[i];
if trait_method.is_root != impl_method.is_root {
let attr = if trait_method.is_root {
"the trait method has `#[turbo_tasks::function(root)]` but the impl does not"
} else {
"the impl has `#[turbo_tasks::function(root)]` but the trait method does not"
};
panic!(
"`root` attribute mismatch on `{}::{}` for `{}`: {}. The `root` attribute \
must match between trait and impl methods.",
trait_method.trait_name,
trait_method.method_name,
entry.value_type.ty.name,
attr,
);
}
}
entry
.value_type
.register_trait(entry.trait_type, entry.methods);
@@ -285,6 +303,9 @@ pub struct TraitMethod {
pub trait_name: &'static str,
pub method_name: &'static str,
pub default_method: Option<&'static NativeFunction>,
/// Whether the trait method declared `#[turbo_tasks::function(root)]`. All impls of a trait
/// method must agree on this attribute (enforced at registration time).
pub is_root: bool,
}
impl Hash for TraitMethod {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
@@ -55,7 +55,7 @@ async fn split_chunk() {
}
.resolved_cell();
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn split_parts_operation(asset: ResolvedVc<TestAsset>) -> Vc<ChunkParts> {
split_output_asset_into_parts(Vc::upcast(*asset))
}
@@ -96,7 +96,7 @@ async fn split_chunk() {
}]
);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn compressed_size_operation(
parts: OperationVc<ChunkParts>,
index: usize,
@@ -180,13 +180,13 @@ impl TurbopackBuildBuilder {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn extract_effects_operation(op: OperationVc<()>) -> Result<Vc<Effects>> {
let _ = op.resolve().strongly_consistent().await?;
Ok(take_effects(op).await?.cell())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn build_internal(
project_dir: RcStr,
root_dir: RcStr,
@@ -252,7 +252,7 @@ impl TurbopackDevServerBuilder {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn source(
root_dir: RcStr,
project_dir: RcStr,
+1 -1
View File
@@ -429,7 +429,7 @@ pub mod tests {
noop_backing_storage(),
));
tt.run_once(async move {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn output_name_operation() -> anyhow::Result<Vc<RcStr>> {
let fs = VirtualFileSystem::new_with_name(rcstr!("test"));
let root = fs.root().owned().await?;
@@ -18,7 +18,7 @@ impl AsyncModulesInfo {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn compute_async_module_info(
graphs: ResolvedVc<ModuleGraph>,
) -> Result<Vc<AsyncModulesInfo>> {
@@ -32,7 +32,7 @@ pub async fn compute_async_module_info(
.connect())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn compute_async_module_info_single(
graph: OperationVc<ModuleGraphLayer>,
parent_async_modules: Option<OperationVc<AsyncModulesInfo>>,
@@ -714,7 +714,7 @@ impl ModuleGraph {
Ok(ReadRef::cell(graph))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn from_graphs_inner(
graphs: Vec<OperationVc<SingleModuleGraph>>,
binding_usage: Option<OperationVc<BindingUsageInfo>>,
@@ -763,7 +763,7 @@ impl ModuleGraph {
compute_style_groups(self, chunking_context, &config).await
}
#[turbo_tasks::function]
#[turbo_tasks::function(root)]
pub async fn async_module_info(self: Vc<Self>) -> Result<Vc<AsyncModulesInfo>> {
// `compute_async_module_info` calls `module.is_self_async()`, so we need to again ignore
// all issues such that they aren't emitted multiple times.
@@ -833,7 +833,7 @@ pub struct ModuleGraphLayer {
#[turbo_tasks::value_impl]
impl ModuleGraphLayer {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn new(
graph: OperationVc<SingleModuleGraph>,
graph_idx: u32,
@@ -2108,7 +2108,7 @@ pub mod tests {
reverse_from_b: Vec<RcStr>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn reverse_traversal_results_operation() -> Result<Vc<ReverseTraversalResults>> {
let fs = VirtualFileSystem::new_with_name(rcstr!("test"));
let root = fs.root().await?;
@@ -2281,7 +2281,7 @@ pub mod tests {
iter_modules_single: Vec<Vec<RcStr>>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn reverse_traversal_results_operation() -> Result<Vc<Results>> {
let fs = VirtualFileSystem::new_with_name(rcstr!("test"));
let root = fs.root().await?;
@@ -2567,7 +2567,7 @@ pub mod tests {
module_to_name: FxHashMap<ResolvedVc<Box<dyn Module>>, RcStr>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn setup_graph(
entries: Vec<RcStr>,
graph_entries: Vec<(RcStr, Vec<RcStr>)>,
@@ -3761,7 +3761,7 @@ mod tests {
#[turbo_tasks::value(transparent)]
struct ResolveRelativeRequestOutput(Vec<(String, String)>);
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn resolve_relative_request_operation(
path: RcStr,
pattern: Pattern,
@@ -3911,7 +3911,7 @@ mod tests {
BackendOptions::default(),
noop_backing_storage(),
));
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn run_test() -> Result<Vc<DupCheckResult>> {
let m_a = make_module(rcstr!("a.js")).to_resolved().await?;
let m_b = make_module(rcstr!("b.js")).to_resolved().await?;
@@ -3948,7 +3948,7 @@ mod tests {
BackendOptions::default(),
noop_backing_storage(),
));
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn run_test() -> Result<Vc<DupCheckResult>> {
let m = make_module(rcstr!("a.js")).to_resolved().await?;
@@ -3981,7 +3981,7 @@ mod tests {
BackendOptions::default(),
noop_backing_storage(),
));
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn run_test() -> Result<Vc<DupCheckResult>> {
let m = make_module(rcstr!("a.js")).to_resolved().await?;
@@ -4024,7 +4024,7 @@ mod tests {
BackendOptions::default(),
noop_backing_storage(),
));
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn run_test() -> Result<Vc<DupCheckResult>> {
let m_a = make_module(rcstr!("a.js")).to_resolved().await?;
let m_b = make_module(rcstr!("b.js")).to_resolved().await?;
@@ -2666,7 +2666,7 @@ mod tests {
subpath_ordering: Vec<String>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn read_matches_operation() -> anyhow::Result<Vc<ReadMatchesOutput>> {
let root = DiskFileSystem::new(
rcstr!("test"),
@@ -363,7 +363,7 @@ mod tests {
rooted_sources: Vec<Option<String>>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn resolve_source_map_sources_operation()
-> anyhow::Result<Vc<SourceMapSourcesOutput>> {
let sys_root = if cfg!(windows) {
@@ -42,7 +42,7 @@ enum GetFromSourceResult {
/// Resolves a [SourceRequest] within a [super::ContentSource], returning the
/// corresponding content as a
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_from_source_operation(
source: OperationVc<Box<dyn ContentSource>>,
request: TransientInstance<SourceRequest>,
@@ -78,7 +78,7 @@ struct GetFromSourceResultWithCollectibles {
content_source_side_effects: AutoSet<ResolvedVc<Box<dyn ContentSourceSideEffect>>>,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_from_source_with_collectibles_operation(
source_op: OperationVc<Box<dyn ContentSource>>,
request: TransientInstance<SourceRequest>,
@@ -61,7 +61,7 @@ struct ContentSourceWithIssues {
effects: Effects,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_source_with_issues_operation(
source_op: OperationVc<Box<dyn ContentSource>>,
) -> Result<Vc<ContentSourceWithIssues>> {
@@ -193,7 +193,7 @@ fn get_sub_paths(sub_path: &str) -> ([RcStr; 3], usize) {
(sub_paths_buffer, n)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn all_assets_map_operation(source: ResolvedVc<AssetGraphContentSource>) -> Vc<OutputAssetsMap> {
source.all_assets_map()
}
@@ -30,14 +30,14 @@ pub enum ResolveSourceRequestResult {
HttpProxy(OperationVc<ProxyResult>),
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn content_source_get_routes_operation(
source: OperationVc<Box<dyn ContentSource>>,
) -> Vc<RouteTree> {
source.connect().get_routes()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn route_tree_get_operation(
route_tree: ResolvedVc<RouteTree>,
asset_path: RcStr,
@@ -45,14 +45,14 @@ fn route_tree_get_operation(
route_tree.get(asset_path)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn get_content_source_content_vary_operation(
get_content: ResolvedVc<Box<dyn GetContentSourceContent>>,
) -> Vc<ContentSourceDataVary> {
get_content.vary()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn get_content_source_content_get_operation(
get_content: ResolvedVc<Box<dyn GetContentSourceContent>>,
path: RcStr,
@@ -77,7 +77,7 @@ fn get_content_source_content_get_operation(
/// TODO: The callers of this function now read this operation using strong consistency. This may
/// have re-introduced performance issues that were solved in
/// <https://github.com/vercel/turborepo/pull/5360>.
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn resolve_source_request(
source: OperationVc<Box<dyn ContentSource>>,
request: TransientInstance<SourceRequest>,
@@ -39,7 +39,7 @@ impl WrappedGetContentSourceContent {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn wrap_sources_operation(
sources: OperationVc<GetContentSourceContents>,
processor: ResolvedVc<Box<dyn ContentSourceProcessor>>,
@@ -103,7 +103,7 @@ fn extend_issues(issues: &mut Vec<ReadRef<PlainIssue>>, new_issues: Vec<ReadRef<
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
fn versioned_content_update_operation(
content: ResolvedVc<Box<dyn VersionedContent>>,
from: ResolvedVc<Box<dyn Version>>,
@@ -111,7 +111,7 @@ fn versioned_content_update_operation(
content.update(*from)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn get_update_stream_item_operation(
resource: RcStr,
from: ResolvedVc<VersionState>,
@@ -413,7 +413,7 @@ pub mod test {
use super::*;
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub fn noop_operation() -> Vc<ResolveSourceRequestResult> {
ResolveSourceRequestResult::NotFound.cell()
}
@@ -140,7 +140,7 @@ struct EmittedEvaluatePoolAssets {
entrypoint: FileSystemPath,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn emit_evaluate_pool_assets_operation(
entries: ResolvedVc<EvaluateEntries>,
chunking_context: ResolvedVc<Box<dyn ChunkingContext>>,
@@ -184,7 +184,7 @@ async fn emit_evaluate_pool_assets_operation(
.cell())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn create_evaluate_pool_assets_operation(
entries: ResolvedVc<EvaluateEntries>,
chunking_context: ResolvedVc<Box<dyn ChunkingContext>>,
@@ -215,7 +215,7 @@ pub enum EnvVarTracking {
Untracked,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
/// Pass the file you cared as `runtime_entries` to invalidate and reload the
/// evaluated result automatically.
pub async fn get_evaluate_pool(
@@ -219,7 +219,7 @@ async fn run(resource: PathBuf, snapshot_mode: IssueSnapshotMode) -> Result<JsRe
effects: Effects,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn run_inner_operation(
resource: RcStr,
snapshot_mode: IssueSnapshotMode,
@@ -237,7 +237,7 @@ async fn run(resource: PathBuf, snapshot_mode: IssueSnapshotMode) -> Result<JsRe
/// Wrapper operation that collects all effects (including snapshot issue file writes) from
/// [`run_inner_operation`].
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn run_inner_operation_with_effects(
resource: RcStr,
snapshot_mode: IssueSnapshotMode,
@@ -358,7 +358,7 @@ async fn prepare_test(resource: RcStr) -> Result<Vc<PreparedTest>> {
.cell())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn run_test_operation(prepared_test: ResolvedVc<PreparedTest>) -> Result<Vc<RunTestResult>> {
let PreparedTest {
path,
@@ -222,7 +222,7 @@ async fn run(resource: PathBuf) -> Result<()> {
noop_backing_storage(),
));
tt.run_once(async move {
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn inner_operation(resource: RcStr) -> Result<Vc<()>> {
let out_op = run_test_operation(resource);
let out_vc = out_op
@@ -243,7 +243,7 @@ async fn run(resource: PathBuf) -> Result<()> {
Ok(Vc::cell(()))
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn extract_effects(op: OperationVc<()>) -> Result<Vc<Effects>> {
let _ = op.resolve().strongly_consistent().await?;
Ok(take_effects(op).await?.cell())
@@ -262,7 +262,7 @@ async fn run(resource: PathBuf) -> Result<()> {
Ok(())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn run_test_operation(resource: RcStr) -> Result<Vc<FileSystemPath>> {
let test_path = canonicalize(&resource)?;
assert!(test_path.exists(), "{resource} does not exist");
@@ -22,7 +22,7 @@ use turbopack_core::{
};
use turbopack_resolve::resolve_options_context::ResolveOptionsContext;
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn extract_effects_operation(op: OperationVc<()>) -> anyhow::Result<Vc<Effects>> {
let _ = op.resolve().strongly_consistent().await?;
Ok(take_effects(op).await?.cell())
@@ -352,7 +352,7 @@ struct NodeFileTraceResult {
effects: Effects,
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn node_file_trace_operation(
package_root: RcStr,
input: RcStr,
@@ -768,12 +768,12 @@ impl std::str::FromStr for CaseInput {
}
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn asset_path_operation(asset: ResolvedVc<Box<dyn OutputAsset>>) -> Vc<FileSystemPath> {
asset.path()
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn print_graph_operation(asset: ResolvedVc<Box<dyn OutputAsset>>) -> Result<()> {
let mut visited = HashSet::new();
let mut queue = Vec::new();
@@ -197,7 +197,7 @@ fn unit_test(#[case] input: &str) -> Result<()> {
node_file_trace(input)
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
async fn node_file_trace_operation(package_root: RcStr, input: RcStr) -> Result<Vc<Vec<RcStr>>> {
let workspace_fs: Vc<Box<dyn FileSystem>> = Vc::upcast(DiskFileSystem::new(
rcstr!("workspace"),
+1 -1
View File
@@ -1218,7 +1218,7 @@ pub async fn emit_assets_into_dir(
Ok(())
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn emit_assets_into_dir_operation(
assets: ResolvedVc<ExpandedOutputAssets>,
output_dir: FileSystemPath,
@@ -485,7 +485,7 @@ pub mod tests {
.unwrap();
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn run_leaves_test_operation() -> Result<()> {
let fs = VirtualFileSystem::new();
let virtual_path = fs.root().await?.join("foo.js")?;
@@ -622,7 +622,7 @@ pub mod tests {
.unwrap();
}
#[turbo_tasks::function(operation)]
#[turbo_tasks::function(operation, root)]
pub async fn run_rule_condition_tree_test_operation() -> Result<()> {
let fs = VirtualFileSystem::new();
let virtual_path = fs.root().await?.join("foo.js")?;