hanzoai/next.js
2
0
Fork 0
mirror of https://github.com/hanzoai/next.js.git synced 2026-09-24 17:00:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

turbo-tasks: replace async resolve fns with custom Future types (ResolveRawVcFuture, ResolveVcFuture, ToResolvedVcFuture) (#91554)

Browse Source 
### What?

Replace the `async fn resolve()`, `async fn
resolve_strongly_consistent()`, and `async fn to_resolved()` methods on
`RawVc`, `Vc<T>`, and `OperationVc<T>` with hand-written custom `Future`
implementations, following the existing `ReadRawVcFuture` pattern.

New types:
- **`ResolveRawVcFuture`** (`raw_vc.rs`) — core implementation, replaces
`async fn resolve_inner()`
- **`ResolveVcFuture<T>`** (`vc/mod.rs`) — typed wrapper over
`ResolveRawVcFuture`, returned by `Vc::resolve()`
- **`ResolveOperationVcFuture<T>`** (`vc/operation.rs`) — typed wrapper,
returned by `OperationVc::resolve()`
- **`ToResolvedVcFuture<T>`** (`vc/mod.rs`) — typed wrapper, returned by
`Vc::to_resolved()`

All new future types expose a `.strongly_consistent()` builder method,
enabling `resolve_strongly_consistent()` to be replaced by
`.resolve().strongly_consistent()` at call sites.

`ReadRawVcFuture` is also updated to delegate its phase-1 resolve loop
to `ResolveRawVcFuture` instead of duplicating the logic.
`std::task::ready!` is used throughout to simplify poll implementations.

Also adds `#[inline(never)]` to `ReadRawVcFuture::poll` and
`ResolveRawVcFuture::poll` to avoid inlining large poll implementations
into every await site.

### Why?

Performance, binary size, and improved API ergonomics:

- The hand-written `Future` pattern (already used by `ReadRawVcFuture`)
gives the compiler more predictable, smaller code than the state
machines generated for `async fn`. The `#[inline(never)]` attributes on
`poll` prevent large poll bodies from being duplicated at every await
site, which the async desugaring otherwise allows.
- The new builder API (`.resolve().strongly_consistent()`) is more
composable and removes the need for separate `_strongly_consistent`
method variants, reducing the number of methods on
`RawVc`/`Vc`/`OperationVc`.
- Having `ReadRawVcFuture` delegate to `ResolveRawVcFuture` removes the
duplicated resolve loop and ensures both paths stay in sync.

### How?

- `ResolveRawVcFuture` stores `current: RawVc`, `read_output_options:
ReadOutputOptions`, `strongly_consistent: bool`, and `listener:
Option<EventListener>`. Its `poll` replicates the loop from the old
`resolve_inner` using `try_read_task_output` / `try_read_local_output`.
- On `Err(listener)` from a `try_*` call, the listener is stored in
`self.listener` and `Poll::Pending` is returned. At the top of the loop,
`ready!(poll_listener(...))` re-polls it and short-circuits if still
pending.
- Consistency is downgraded to `Eventual` after the first `TaskOutput`
hop, matching the previous behavior.
- `strongly_consistent: true` keeps the
`SUPPRESS_EVENTUAL_CONSISTENCY_TOP_LEVEL_TASK_CHECK` suppression across
all polls (same logic as `ReadRawVcFuture`).
- `ReadRawVcFuture` now holds a `ResolveRawVcFuture` for phase 1 and
drives it via `Pin::new(&mut self.resolve).poll(cx)` before proceeding
to the cell read in phase 2. This eliminates the duplicated loop that
previously existed in both types.
- Typed wrappers (`ResolveVcFuture<T>`, `ResolveOperationVcFuture<T>`,
`ToResolvedVcFuture<T>`) delegate `poll` to the inner
`ResolveRawVcFuture` and map the output to the appropriate typed result.
- `OperationVc::resolve_strongly_consistent()` is removed; 16 call sites
updated to `.resolve().strongly_consistent()`.
- All new types implement `Unpin` and are exported from `lib.rs`.
- `std::task::ready!` is used in all `poll` implementations to reduce
boilerplate.

No behavioral changes — this is a pure implementation refactor.

### Binary size impact

A release build (`pnpm swc-build-native --release`) was measured before
and after the branch changes on the same merge-base commit (`a41bef94`):

| | Size |
|---|---|
| Base (`a41bef94`, before branch) | 199,690,656 bytes (~190.4 MB) |
| Branch (`6f7846f9`, after changes) | 199,252,384 bytes (~190.0 MB) |
| **Difference** | **−438,272 bytes (−428 KB, −0.22%)** |

The branch produces a slightly smaller binary. The reduction comes
primarily from the `#[inline(never)]` attributes preventing large `poll`
bodies from being duplicated at every await site.

---------

Co-authored-by: Tobias Koppers <[email protected]>
Co-authored-by: Claude <[email protected]>
This commit is contained in:
Tobias Koppers
2026-04-04 08:11:57 +02:00
committed by GitHub
co-authored by Tobias Koppers Claude
parent f6663b75fd
commit f65b10a54d
25 changed files with 367 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/next-api/src/module_graph.rs
+3 -3
View File
@@ -76,7 +76,7 @@ impl NextDynamicGraphs {
// `get_global_information_for_endpoint_inner` calls `take_collectibles()` when needed
let result_op = Self::new_operation(graphs, is_single_page);
let result_vc = if !is_single_page {
let result_vc = result_op.resolve_strongly_consistent().await?;
let result_vc = result_op.resolve().strongly_consistent().await?;
result_op.drop_collectibles::<Box<dyn Issue>>();
*result_vc
} else {
@@ -274,7 +274,7 @@ impl ServerActionsGraphs {
// `get_global_information_for_endpoint_inner` calls `take_collectibles()` when needed
let result_op = Self::new_operation(graphs, is_single_page);
let result_vc = if !is_single_page {
let result_vc = result_op.resolve_strongly_consistent().await?;
let result_vc = result_op.resolve().strongly_consistent().await?;
result_op.drop_collectibles::<Box<dyn Issue>>();
*result_vc
} else {
@@ -452,7 +452,7 @@ impl ClientReferencesGraphs {
// `get_global_information_for_endpoint_inner` calls `take_collectibles()` when needed
let result_op = Self::new_operation(graphs, is_single_page);
let result_vc = if !is_single_page {
let result_vc = result_op.resolve_strongly_consistent().await?;
let result_vc = result_op.resolve().strongly_consistent().await?;
result_op.drop_collectibles::<Box<dyn Issue>>();
*result_vc
} else {
crates/next-api/src/operation.rs
+1 -1
View File
@@ -37,7 +37,7 @@ pub struct EntrypointsOperation {
async fn entrypoints_without_collectibles_operation(
entrypoints: OperationVc<Entrypoints>,
) -> Result<Vc<Entrypoints>> {
let _ = entrypoints.resolve_strongly_consistent().await?;
let _ = entrypoints.resolve().strongly_consistent().await?;
entrypoints.drop_collectibles::<Box<dyn Diagnostic>>();
entrypoints.drop_issues();
let _ = get_effects(entrypoints).await?;
crates/next-api/src/project.rs
+7 -4
View File
@@ -615,7 +615,8 @@ impl ProjectContainer {
container.connect().project()
}
let project = project_from_container_operation(this_op)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
let project_fs = project_fs_operation(project)
.read_strongly_consistent()
@@ -738,7 +739,8 @@ impl ProjectContainer {
let watch = new_options.watch;
let project = project_operation(self)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
let prev_project_fs = project_fs_operation(project)
.read_strongly_consistent()
@@ -756,7 +758,8 @@ impl ProjectContainer {
}
this.options_state.set(Some(new_options));
let project = project_operation(self)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
let project_fs = project_fs_operation(project)
.read_strongly_consistent()
@@ -1512,7 +1515,7 @@ impl Project {
} else {
// In development mode, we need to to take and drop the issues, otherwise every
// route will report all issues.
let vc = module_graphs_op.resolve_strongly_consistent().await?;
let vc = module_graphs_op.resolve().strongly_consistent().await?;
module_graphs_op.drop_issues();
*vc
};
crates/next-build-test/src/lib.rs
+2 -2
View File
@@ -41,7 +41,7 @@ pub async fn main_inner(
.run(async {
let container_op = ProjectContainer::new_operation(rcstr!("next.js"), options.dev);
ProjectContainer::initialize(container_op, options).await?;
container_op.resolve_strongly_consistent().await
container_op.resolve().strongly_consistent().await
})
.await?;
@@ -248,7 +248,7 @@ async fn endpoint_write_to_disk_with_effects(
endpoint: ResolvedVc<Box<dyn Endpoint>>,
) -> Result<Vc<EndpointOutputPaths>> {
let op = endpoint_write_to_disk_operation(endpoint);
let result = op.resolve_strongly_consistent().await?;
let result = op.resolve().strongly_consistent().await?;
get_effects(op).await?.apply().await?;
Ok(*result)
}
crates/next-napi-bindings/src/next_api/project.rs
+1 -1
View File
@@ -588,7 +588,7 @@ pub fn project_new(
.run(async move {
let container_op = ProjectContainer::new_operation(rcstr!("next.js"), is_dev);
ProjectContainer::initialize(container_op, options).await?;
container_op.resolve_strongly_consistent().await
container_op.resolve().strongly_consistent().await
})
.or_else(|e| turbopack_ctx.throw_turbopack_internal_result(&e.into()))
.await?;
turbopack/crates/turbo-tasks-backend/tests/emptied_cells.rs
+1 -1
View File
@@ -12,7 +12,7 @@ static REGISTRATION: Registration = register!();
async fn test_emptied_cells() {
run(&REGISTRATION, || async {
let input_op = get_state_operation();
let input_vc = input_op.resolve_strongly_consistent().await?;
let input_vc = input_op.resolve().strongly_consistent().await?;
let input = input_op.read_strongly_consistent().await?;
input.state.set(0);
turbopack/crates/turbo-tasks-backend/tests/emptied_cells_session_dependent.rs
+1 -1
View File
@@ -12,7 +12,7 @@ static REGISTRATION: Registration = register!();
async fn test_emptied_cells_session_dependent() {
run(&REGISTRATION, || async {
let input_op = get_state_operation();
let input_vc = input_op.resolve_strongly_consistent().await?;
let input_vc = input_op.resolve().strongly_consistent().await?;
let input = input_op.read_strongly_consistent().await?;
input.state.set(0);
turbopack/crates/turbo-tasks-backend/tests/invalidation.rs
+2 -2
View File
@@ -21,7 +21,7 @@ fn create_state_operation() -> Vc<Step> {
async fn test_invalidation_map() {
run(&REGISTRATION, || async {
let state_op = create_state_operation();
let state_vc = state_op.resolve_strongly_consistent().await?;
let state_vc = state_op.resolve().strongly_consistent().await?;
let state = state_op.read_strongly_consistent().await?;
state.set(1);
@@ -101,7 +101,7 @@ async fn get_value(map: OperationVc<Map>, key: String) -> Result<Vc<GetValueResu
async fn test_invalidation_set() {
run(&REGISTRATION, || async {
let state_op = create_state_operation();
let state_vc = state_op.resolve_strongly_consistent().await?;
let state_vc = state_op.resolve().strongly_consistent().await?;
let state = state_op.read_strongly_consistent().await?;
state.set(1);
turbopack/crates/turbo-tasks-backend/tests/random_change.rs
+1 -1
View File
@@ -13,7 +13,7 @@ static REGISTRATION: Registration = register!();
async fn test_random_change() {
run_once(&REGISTRATION, || async {
let state_op = make_state_operation();
let state_vc = state_op.resolve_strongly_consistent().await?;
let state_vc = state_op.resolve().strongly_consistent().await?;
let state = state_op.read_strongly_consistent().await?;
let mut rng = StdRng::from_seed(Default::default());
turbopack/crates/turbo-tasks-backend/tests/top_level_task_consistency.rs
+2 -1
View File
@@ -39,7 +39,8 @@ async fn test_eventual_read_in_top_level_task_fails() {
async fn test_cell_read_in_top_level_task_succeeds() {
run_once(&REGISTRATION, || async {
let cell = returns_value_operation()
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
let value = cell.await?;
assert_eq!(value.value, 42);
turbopack/crates/turbo-tasks-fs/src/lib.rs
+4 -2
View File
@@ -3162,7 +3162,8 @@ mod tests {
tt.run_once(async move {
let fs = disk_file_system_operation(root)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
let root_path = disk_file_system_root(fs);
@@ -3269,7 +3270,8 @@ mod tests {
tt.run_once(async move {
let fs = disk_file_system_operation(root)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
let root_path = disk_file_system_root(fs);
let symlinks_dir = root_path.join("_symlinks")?;
turbopack/crates/turbo-tasks-fuzz/src/fs_watcher.rs
+4 -2
View File
@@ -103,10 +103,12 @@ pub async fn run(args: FsWatcher) -> anyhow::Result<()> {
tt.run_once(async move {
let invalidations = TransientInstance::new(PathInvalidations::default());
let project_fs = disk_file_system_operation(RcStr::from(fs_root.to_str().unwrap()))
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
let project_root = disk_file_system_root_operation(project_fs)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?
.owned()
.await?;
turbopack/crates/turbo-tasks-fuzz/src/symlink_stress.rs
+4 -2
View File
@@ -60,10 +60,12 @@ pub async fn run(args: SymlinkStress) -> anyhow::Result<()> {
tt.run_once(async move {
let project_fs = disk_file_system_operation(RcStr::from(fs_root.to_str().unwrap()))
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
let project_root = disk_file_system_root_operation(project_fs)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?
.owned()
.await?;
turbopack/crates/turbo-tasks/src/lib.rs
+4 -3
View File
@@ -97,7 +97,7 @@ pub use crate::{
},
mapped_read_ref::MappedReadRef,
output::OutputContent,
raw_vc::{CellId, RawVc, ReadRawVcFuture},
raw_vc::{CellId, RawVc, ReadRawVcFuture, ResolveRawVcFuture},
read_options::{ReadCellOptions, ReadOutputOptions},
read_ref::ReadRef,
serialization_invalidation::SerializationInvalidator,
@@ -112,8 +112,9 @@ pub use crate::{
value::{TransientInstance, TransientValue},
value_type::{TraitMethod, TraitType, ValueType},
vc::{
Dynamic, NonLocalValue, OperationValue, OperationVc, OptionVcExt, ReadVcFuture, ResolvedVc,
Upcast, UpcastStrict, ValueDefault, Vc, VcCast, VcCellCompareMode, VcCellKeyedCompareMode,
Dynamic, NonLocalValue, OperationValue, OperationVc, OptionVcExt, ReadVcFuture,
ResolveOperationVcFuture, ResolveVcFuture, ResolvedVc, ToResolvedVcFuture, Upcast,
UpcastStrict, ValueDefault, Vc, VcCast, VcCellCompareMode, VcCellKeyedCompareMode,
VcCellNewMode, VcDefaultRead, VcRead, VcTransparentRead, VcValueTrait, VcValueTraitCast,
VcValueType, VcValueTypeCast,
},
turbopack/crates/turbo-tasks/src/raw_vc.rs
+171 -143
View File
@@ -3,7 +3,7 @@ use std::{
future::Future,
pin::Pin,
sync::Arc,
task::Poll,
task::{Poll, ready},
};
use anyhow::Result;
@@ -19,9 +19,9 @@ use crate::{
id::{ExecutionId, LocalTaskId},
manager::{
ReadCellTracking, ReadTracking, SUPPRESS_EVENTUAL_CONSISTENCY_TOP_LEVEL_TASK_CHECK,
TurboTasksApi, read_local_output, read_task_output, with_turbo_tasks,
TurboTasksApi, read_local_output, with_turbo_tasks,
},
registry::{self, get_value_type},
registry::get_value_type,
turbo_tasks,
};
@@ -33,12 +33,7 @@ pub struct CellId {
impl Display for CellId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{}#{}",
registry::get_value_type(self.type_id).ty.name,
self.index
)
write!(f, "{}#{}", get_value_type(self.type_id).ty.name, self.index)
}
}
@@ -153,44 +148,8 @@ impl RawVc {
}
/// See [`crate::Vc::to_resolved`].
pub(crate) async fn resolve(self) -> Result<RawVc> {
self.resolve_inner(ReadOutputOptions {
consistency: ReadConsistency::Eventual,
..Default::default()
})
.await
}
/// See [`crate::Vc::resolve_strongly_consistent`].
pub(crate) async fn resolve_strongly_consistent(self) -> Result<RawVc> {
SuppressTopLevelTaskCheckFuture {
inner: self.resolve_inner(ReadOutputOptions {
consistency: ReadConsistency::Strong,
..Default::default()
}),
}
.await
}
async fn resolve_inner(self, mut options: ReadOutputOptions) -> Result<RawVc> {
let tt = turbo_tasks();
let mut current = self;
loop {
match current {
RawVc::TaskOutput(task) => {
current = read_task_output(&*tt, task, options).await?;
// We no longer need to read strongly consistent, as any Vc returned
// from the first task will be inside of the scope of the first
// task. So it's already strongly consistent.
options.consistency = ReadConsistency::Eventual;
}
RawVc::TaskCell(_, _) => return Ok(current),
RawVc::LocalOutput(execution_id, local_task_id, ..) => {
debug_assert_eq!(options.consistency, ReadConsistency::Eventual);
current = read_local_output(&*tt, execution_id, local_task_id).await?;
}
}
}
pub(crate) fn resolve(self) -> ResolveRawVcFuture {
ResolveRawVcFuture::new(self)
}
/// Convert a potentially local `RawVc` into a non-local `RawVc`. This is a subset of resolution
@@ -297,106 +256,87 @@ impl CollectiblesSource for RawVc {
}
}
/// A future wrapper that suppresses the top-level task eventual consistency check
/// during each [`poll`][Future::poll] call. The suppression is applied via
/// [`sync_scope`][tokio::task_local!] so it is only active during the synchronous
/// execution of the inner future's `poll`, and is never held across await points.
struct SuppressTopLevelTaskCheckFuture<F> {
inner: F,
/// Polls a pending [`EventListener`] slot. Returns [`Poll::Pending`] if the event has not yet
/// fired. On [`Poll::Ready`], clears the slot so it is not polled again.
fn poll_listener(
listener: &mut Option<EventListener>,
cx: &mut std::task::Context<'_>,
) -> Poll<()> {
if let Some(l) = listener {
ready!(Pin::new(l).poll(cx));
*listener = None;
}
Poll::Ready(())
}
impl<F: Future> Future for SuppressTopLevelTaskCheckFuture<F> {
type Output = F::Output;
fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
// SAFETY: we are only projecting the pin to the inner field, not moving it
let inner = unsafe { self.map_unchecked_mut(|this| &mut this.inner) };
if cfg!(debug_assertions) {
SUPPRESS_EVENTUAL_CONSISTENCY_TOP_LEVEL_TASK_CHECK.sync_scope(true, || inner.poll(cx))
} else {
inner.poll(cx)
}
/// Wraps `f` in a scope that suppresses the eventual-consistency top-level task assertion,
/// but only when `strongly_consistent` is `true` and debug assertions are enabled.
///
/// This is needed because a strongly-consistent read of a `TaskOutput` is not a single atomic
/// operation — inner reads switch to eventual consistency after the first output is resolved —
/// which would otherwise trigger the assertion in top-level tasks.
fn suppress_top_level_task_check<R>(strongly_consistent: bool, f: impl FnOnce() -> R) -> R {
if cfg!(debug_assertions) && strongly_consistent {
// Temporarily suppress the top-level task check
SUPPRESS_EVENTUAL_CONSISTENCY_TOP_LEVEL_TASK_CHECK.sync_scope(true, f)
} else {
f()
}
}
#[must_use]
pub struct ReadRawVcFuture {
pub struct ResolveRawVcFuture {
current: RawVc,
read_output_options: ReadOutputOptions,
read_cell_options: ReadCellOptions,
is_serializable_cell_content_unknown: bool,
/// This flag redundant with `read_output_options`, but `read_output_options` is mutated during
/// the read. This flag indicates that the initial read was strongly consistent.
/// This flag is redundant with `read_output_options`, but `read_output_options` is mutated
/// during the resolve. This flag indicates that the initial read was strongly consistent.
strongly_consistent: bool,
listener: Option<EventListener>,
}
impl ReadRawVcFuture {
pub(crate) fn new(vc: RawVc, is_serializable_cell_content: Option<bool>) -> Self {
ReadRawVcFuture {
impl ResolveRawVcFuture {
fn new(vc: RawVc) -> Self {
ResolveRawVcFuture {
current: vc,
read_output_options: ReadOutputOptions::default(),
read_cell_options: ReadCellOptions {
is_serializable_cell_content: is_serializable_cell_content.unwrap_or(false),
..Default::default()
},
is_serializable_cell_content_unknown: is_serializable_cell_content.is_none(),
strongly_consistent: false,
listener: None,
}
}
/// Make reads strongly consistent.
pub fn strongly_consistent(mut self) -> Self {
self.read_output_options.consistency = ReadConsistency::Strong;
self.strongly_consistent = true;
self.read_output_options.consistency = ReadConsistency::Strong;
self
}
/// Track the value as a dependency with an key.
pub fn track_with_key(mut self, key: u64) -> Self {
/// Track task output reads with a specific key (forwarded from
/// [`ReadRawVcFuture::track_with_key`]).
pub(crate) fn track_with_key(mut self) -> Self {
self.read_output_options.tracking = ReadTracking::Tracked;
self.read_cell_options.tracking = ReadCellTracking::Tracked { key: Some(key) };
self
}
/// This will not track the value as dependency, but will still track the error as dependency,
/// if there is an error.
///
/// INVALIDATION: Be careful with this, it will not track dependencies, so
/// using it could break cache invalidation.
pub fn untracked(mut self) -> Self {
/// Do not track task output reads as dependencies (forwarded from
/// [`ReadRawVcFuture::untracked`]).
pub(crate) fn untracked(mut self) -> Self {
self.read_output_options.tracking = ReadTracking::TrackOnlyError;
self.read_cell_options.tracking = ReadCellTracking::TrackOnlyError;
self
}
/// Hint that this is the final read of the cell content.
pub fn final_read_hint(mut self) -> Self {
self.read_cell_options.final_read_hint = true;
self
}
}
impl Future for ReadRawVcFuture {
type Output = Result<TypedCellContent>;
impl Future for ResolveRawVcFuture {
type Output = Result<RawVc>;
#[inline(never)]
fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
// SAFETY: we are not moving self
let this = unsafe { self.get_unchecked_mut() };
// Extract the closure to avoid deep nesting
let poll_fn = |tt: &Arc<dyn TurboTasksApi>| -> Poll<Self::Output> {
'outer: loop {
if let Some(listener) = &mut this.listener {
// SAFETY: listener is from previous pinned this
let listener = unsafe { Pin::new_unchecked(listener) };
if listener.poll(cx).is_pending() {
return Poll::Pending;
}
this.listener = None;
}
let mut listener = match this.current {
ready!(poll_listener(&mut this.listener, cx));
let listener = match this.current {
RawVc::TaskOutput(task) => {
let read_result = tt.try_read_task_output(task, this.read_output_options);
match read_result {
@@ -417,29 +357,13 @@ impl Future for ReadRawVcFuture {
Err(err) => return Poll::Ready(Err(err)),
}
}
RawVc::TaskCell(task, index) => {
if this.is_serializable_cell_content_unknown {
let value_type = registry::get_value_type(index.type_id);
this.read_cell_options.is_serializable_cell_content =
value_type.bincode.is_some();
}
let read_result =
tt.try_read_task_cell(task, index, this.read_cell_options);
match read_result {
Ok(Ok(content)) => {
// SAFETY: Constructor ensures that T and U are binary identical
return Poll::Ready(Ok(content));
}
Ok(Err(listener)) => listener,
Err(err) => return Poll::Ready(Err(err)),
}
}
RawVc::LocalOutput(execution_id, local_output_id, ..) => {
RawVc::TaskCell(_, _) => return Poll::Ready(Ok(this.current)),
RawVc::LocalOutput(execution_id, local_task_id, ..) => {
debug_assert_eq!(
this.read_output_options.consistency,
ReadConsistency::Eventual
);
let read_result = tt.try_read_local_output(execution_id, local_output_id);
let read_result = tt.try_read_local_output(execution_id, local_task_id);
match read_result {
Ok(Ok(vc)) => {
this.current = vc;
@@ -450,26 +374,10 @@ impl Future for ReadRawVcFuture {
}
}
};
// SAFETY: listener is from previous pinned this
match unsafe { Pin::new_unchecked(&mut listener) }.poll(cx) {
Poll::Ready(_) => continue,
Poll::Pending => {
this.listener = Some(listener);
return Poll::Pending;
}
};
this.listener = Some(listener);
}
};
fn suppress_top_level_task_check<R>(strongly_consistent: bool, f: impl FnOnce() -> R) -> R {
if cfg!(debug_assertions) && strongly_consistent {
// Temporarily suppress the top-level task check
SUPPRESS_EVENTUAL_CONSISTENCY_TOP_LEVEL_TASK_CHECK.sync_scope(true, f)
} else {
f()
}
}
// HACK: Temporarily suppress top-level task check if doing strongly consistent read.
//
// This masks a bug: There's an unlikely TOCTOU race condition in `poll_fn`. Because the
@@ -480,4 +388,124 @@ impl Future for ReadRawVcFuture {
}
}
impl Unpin for ResolveRawVcFuture {}
#[must_use]
pub struct ReadRawVcFuture {
/// Phase 1: resolves the [`RawVc`] pointer chain to a [`RawVc::TaskCell`].
resolve: ResolveRawVcFuture,
/// Phase 2: options for the cell read once we have a [`RawVc::TaskCell`].
read_cell_options: ReadCellOptions,
/// If `true`, the `is_serializable_cell_content` flag in `read_cell_options` is unknown at
/// construction time and must be determined lazily from the type registry once we reach the
/// [`RawVc::TaskCell`].
is_serializable_cell_content_unknown: bool,
/// Phase 2: the resolved task and cell identity, set when phase 1 completes.
resolved: Option<(TaskId, CellId)>,
/// Phase 2: listener for the cell read wait.
listener: Option<EventListener>,
}
impl ReadRawVcFuture {
pub(crate) fn new(vc: RawVc, is_serializable_cell_content: Option<bool>) -> Self {
ReadRawVcFuture {
resolve: ResolveRawVcFuture::new(vc),
read_cell_options: ReadCellOptions {
is_serializable_cell_content: is_serializable_cell_content.unwrap_or(false),
..Default::default()
},
is_serializable_cell_content_unknown: is_serializable_cell_content.is_none(),
resolved: None,
listener: None,
}
}
/// Make reads strongly consistent.
pub fn strongly_consistent(mut self) -> Self {
self.resolve = self.resolve.strongly_consistent();
self
}
/// Track the value as a dependency with an key.
pub fn track_with_key(mut self, key: u64) -> Self {
self.resolve = self.resolve.track_with_key();
self.read_cell_options.tracking = ReadCellTracking::Tracked { key: Some(key) };
self
}
/// This will not track the value as dependency, but will still track the error as dependency,
/// if there is an error.
///
/// INVALIDATION: Be careful with this, it will not track dependencies, so
/// using it could break cache invalidation.
pub fn untracked(mut self) -> Self {
self.resolve = self.resolve.untracked();
self.read_cell_options.tracking = ReadCellTracking::TrackOnlyError;
self
}
/// Hint that this is the final read of the cell content.
pub fn final_read_hint(mut self) -> Self {
self.read_cell_options.final_read_hint = true;
self
}
}
impl Future for ReadRawVcFuture {
type Output = Result<TypedCellContent>;
#[inline(never)]
fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
// SAFETY: we are not moving self
let this = unsafe { self.get_unchecked_mut() };
// --- Phase 1: resolve the RawVc pointer chain to a TaskCell ---
//
// `ResolveRawVcFuture` is `Unpin`, so `Pin::new` is safe.
// It handles `with_turbo_tasks` and `suppress_top_level_task_check` internally.
if this.resolved.is_none() {
match ready!(Pin::new(&mut this.resolve).poll(cx)) {
Err(err) => return Poll::Ready(Err(err)),
Ok(RawVc::TaskCell(task, index)) => {
this.resolved = Some((task, index));
}
Ok(_) => unreachable!("ResolveRawVcFuture always resolves to a TaskCell"),
}
}
// --- Phase 2: read the cell content ---
//
// At this point `this.resolved` is `Some((task, index))`.
let (task, index) = this.resolved.unwrap();
// Lazily resolve `is_serializable_cell_content` from the type registry on the first
// entry into phase 2, then clear the flag so subsequent polls skip this lookup.
if this.is_serializable_cell_content_unknown {
this.read_cell_options.is_serializable_cell_content =
get_value_type(index.type_id).bincode.is_some();
this.is_serializable_cell_content_unknown = false;
}
let poll_fn = |tt: &Arc<dyn TurboTasksApi>| -> Poll<Self::Output> {
loop {
ready!(poll_listener(&mut this.listener, cx));
let listener = match tt.try_read_task_cell(task, index, this.read_cell_options) {
Ok(Ok(content)) => return Poll::Ready(Ok(content)),
Ok(Err(listener)) => listener,
Err(err) => return Poll::Ready(Err(err)),
};
this.listener = Some(listener);
}
};
// Phase 2 must also suppress the top-level task check when phase 1 was
// strongly-consistent. The suppression from `ResolveRawVcFuture::poll` only lasts for
// the duration of that individual `poll` call and does not carry over to subsequent calls
// or to this phase.
suppress_top_level_task_check(this.resolve.strongly_consistent, || {
with_turbo_tasks(poll_fn)
})
}
}
impl Unpin for ReadRawVcFuture {}
turbopack/crates/turbo-tasks/src/vc/mod.rs
+90 -18
View File
@@ -14,6 +14,8 @@ use std::{
hash::{Hash, Hasher},
marker::PhantomData,
ops::Deref,
pin::Pin,
task::{Context, Poll},
};
use anyhow::Result;
@@ -26,13 +28,13 @@ pub use self::{
cell_mode::{VcCellCompareMode, VcCellKeyedCompareMode, VcCellMode, VcCellNewMode},
default::ValueDefault,
local::NonLocalValue,
operation::{OperationValue, OperationVc},
operation::{OperationValue, OperationVc, ResolveOperationVcFuture},
read::{ReadOwnedVcFuture, ReadVcFuture, VcDefaultRead, VcRead, VcTransparentRead},
resolved::ResolvedVc,
traits::{Dynamic, Upcast, UpcastStrict, VcValueTrait, VcValueType},
};
use crate::{
CellId, RawVc,
CellId, RawVc, ResolveRawVcFuture,
debug::{ValueDebug, ValueDebugFormat, ValueDebugFormatString},
keyed::{KeyedAccess, KeyedEq},
registry,
@@ -40,6 +42,78 @@ use crate::{
vc::read::{ReadContainsKeyedVcFuture, ReadKeyedVcFuture},
};
/// A future returned by [`Vc::resolve`] that resolves a [`Vc<T>`] to a cell.
///
/// To opt into strong consistency, use [`OperationVc::resolve`] which returns a
/// [`ResolveOperationVcFuture`] with a
/// [`.strongly_consistent()`][ResolveOperationVcFuture::strongly_consistent] method.
#[must_use]
pub struct ResolveVcFuture<T>
where
T: ?Sized,
{
pub(crate) inner: ResolveRawVcFuture,
pub(crate) _t: PhantomData<T>,
}
impl<T: ?Sized> Future for ResolveVcFuture<T> {
type Output = Result<Vc<T>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// SAFETY: we are not moving self
let this = unsafe { self.get_unchecked_mut() };
// ResolveRawVcFuture: Unpin, so Pin::new is safe
Pin::new(&mut this.inner).poll(cx).map(|r| {
r.map(|node| Vc {
node,
_t: PhantomData,
})
})
}
}
impl<T: ?Sized> Unpin for ResolveVcFuture<T> {}
/// A future returned by [`Vc::to_resolved`] that resolves a [`Vc<T>`] to a [`ResolvedVc<T>`].
///
/// Use [`.strongly_consistent()`][Self::strongly_consistent] to opt into strong consistency.
#[must_use]
pub struct ToResolvedVcFuture<T>
where
T: ?Sized,
{
inner: ResolveRawVcFuture,
_t: PhantomData<T>,
}
impl<T: ?Sized> ToResolvedVcFuture<T> {
/// Make the resolution strongly consistent.
pub fn strongly_consistent(mut self) -> Self {
self.inner = self.inner.strongly_consistent();
self
}
}
impl<T: ?Sized> Future for ToResolvedVcFuture<T> {
type Output = Result<ResolvedVc<T>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// SAFETY: we are not moving self
let this = unsafe { self.get_unchecked_mut() };
// ResolveRawVcFuture: Unpin, so Pin::new is safe
Pin::new(&mut this.inner).poll(cx).map(|r| {
r.map(|node| ResolvedVc {
node: Vc {
node,
_t: PhantomData,
},
})
})
}
}
impl<T: ?Sized> Unpin for ToResolvedVcFuture<T> {}
type VcReadTarget<T> = <<T as VcValueType>::Read as VcRead<T>>::Target;
#[doc = include_str!("README.md")]
@@ -337,16 +411,23 @@ where
Ok(())
}
/// Do not use this: Use [`Vc::to_resolved`] instead. If you must have a resolved [`Vc`] type
/// and not a [`ResolvedVc`] type, simply deref the result of [`Vc::to_resolved`].
pub fn resolve(self) -> ResolveVcFuture<T> {
ResolveVcFuture {
inner: self.node.resolve(),
_t: PhantomData,
}
}
/// Resolve the reference until it points to a cell directly, and wrap the
/// result in a [`ResolvedVc`], which statically guarantees that the
/// [`Vc`] was resolved.
pub async fn to_resolved(self) -> Result<ResolvedVc<T>> {
Ok(ResolvedVc {
node: Vc {
node: self.node.resolve().await?,
_t: PhantomData,
},
})
pub fn to_resolved(self) -> ToResolvedVcFuture<T> {
ToResolvedVcFuture {
inner: self.node.resolve(),
_t: PhantomData,
}
}
/// Returns `true` if the reference is resolved, meaning the underlying [`RawVc`] uses the
@@ -369,15 +450,6 @@ where
pub fn is_local(self) -> bool {
self.node.is_local()
}
/// Do not use this: Use [`OperationVc::resolve_strongly_consistent`] instead.
#[cfg(feature = "non_operation_vc_strongly_consistent")]
pub async fn resolve_strongly_consistent(self) -> Result<Self> {
Ok(Self {
node: self.node.resolve_strongly_consistent().await?,
_t: PhantomData,
})
}
}
impl<T> From<RawVc> for Vc<T>
turbopack/crates/turbo-tasks/src/vc/operation.rs
+50 -12
View File
@@ -1,4 +1,10 @@
use std::{fmt::Debug, hash::Hash, marker::PhantomData};
use std::{
fmt::Debug,
future::Future,
hash::Hash,
pin::Pin,
task::{Context, Poll},
};
use anyhow::Result;
use auto_hash_map::AutoSet;
@@ -11,6 +17,41 @@ use crate::{
VcValueTraitCast, VcValueType, marker_trait::impl_auto_marker_trait, trace::TraceRawVcs,
};
/// A future returned by [`OperationVc::resolve`] that connects an [`OperationVc<T>`] and resolves
/// it to a [`ResolvedVc<T>`].
///
/// Use [`.strongly_consistent()`][Self::strongly_consistent] to opt into strong consistency.
#[must_use]
pub struct ResolveOperationVcFuture<T>
where
T: ?Sized,
{
inner: super::ResolveVcFuture<T>,
}
impl<T: ?Sized> ResolveOperationVcFuture<T> {
/// Make the resolution strongly consistent.
pub fn strongly_consistent(mut self) -> Self {
self.inner.inner = self.inner.inner.strongly_consistent();
self
}
}
impl<T: ?Sized> Future for ResolveOperationVcFuture<T> {
type Output = anyhow::Result<ResolvedVc<T>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// SAFETY: we are not moving self
let this = unsafe { self.get_unchecked_mut() };
// ResolveVcFuture: Unpin, so Pin::new is safe
Pin::new(&mut this.inner)
.poll(cx)
.map(|r| r.map(|node| ResolvedVc { node }))
}
}
impl<T: ?Sized> Unpin for ResolveOperationVcFuture<T> {}
/// A "subtype" (can be converted via [`.connect()`]) of [`Vc`] that
/// represents a specific call (with arguments) to [a task][macro@crate::function].
///
@@ -106,23 +147,20 @@ impl<T: ?Sized> OperationVc<T> {
}
}
/// [Connects the `OperationVc`][Self::connect] and [resolves][Vc::to_resolved] the reference
/// until it points to a cell directly in a [strongly
/// consistent][crate::ReadConsistency::Strong] way.
/// [Connects the `OperationVc`][Self::connect] and resolves the reference until it points to a
/// cell directly.
///
/// Resolving will wait for task execution to be finished, so that the returned [`ResolvedVc`]
/// points to a cell that stores a value.
///
/// Resolving is necessary to compare identities of [`Vc`]s.
///
/// This is async and will rethrow any fatal error that happened during task execution.
pub async fn resolve_strongly_consistent(self) -> Result<ResolvedVc<T>> {
Ok(ResolvedVc {
node: Vc {
node: self.connect().node.resolve_strongly_consistent().await?,
_t: PhantomData,
},
})
/// Use [`.strongly_consistent()`][ResolveOperationVcFuture::strongly_consistent] to opt into
/// strong consistency.
pub fn resolve(self) -> ResolveOperationVcFuture<T> {
ResolveOperationVcFuture {
inner: self.connect().resolve(),
}
}
/// [Connects the `OperationVc`][Self::connect] and returns a [strongly
turbopack/crates/turbopack-core/src/issue/mod.rs
+1 -1
View File
@@ -1134,7 +1134,7 @@ pub async fn handle_issues<T: Send>(
operation: Option<&str>,
) -> Result<()> {
let source_vc = source_op.connect();
let _ = source_op.resolve_strongly_consistent().await?;
let _ = source_op.resolve().strongly_consistent().await?;
let has_fatal = issue_reporter.report_issues(
TransientValue::new(Vc::into_raw(source_vc)),
turbopack/crates/turbopack-core/src/module_graph/mod.rs
+1 -1
View File
@@ -801,7 +801,7 @@ impl ModuleGraph {
// all issues such that they aren't emitted multiple times.
async move {
let result_op = compute_async_module_info(self.to_resolved().await?);
let result_vc = result_op.resolve_strongly_consistent().await?;
let result_vc = result_op.resolve().strongly_consistent().await?;
result_op.drop_collectibles::<Box<dyn Issue>>();
anyhow::Ok(*result_vc)
}
turbopack/crates/turbopack-dev-server/src/http.rs
+1 -1
View File
@@ -84,7 +84,7 @@ pub async fn process_request_with_content_source(
let original_path = request.uri().path().to_string();
let request = http_request_to_source_request(request).await?;
let result_op = get_from_source_operation(source, TransientInstance::new(request));
let resolved_result = result_op.resolve_strongly_consistent().await?;
let resolved_result = result_op.resolve().strongly_consistent().await?;
apply_effects(result_op).await?;
let side_effects: AutoSet<ResolvedVc<Box<dyn ContentSourceSideEffect>>> =
result_op.peek_collectibles();
turbopack/crates/turbopack-dev-server/src/lib.rs
+2 -2
View File
@@ -65,7 +65,7 @@ struct ContentSourceWithIssues {
async fn get_source_with_issues_operation(
source_op: OperationVc<Box<dyn ContentSource>>,
) -> Result<Vc<ContentSourceWithIssues>> {
let _ = source_op.resolve_strongly_consistent().await?;
let _ = source_op.resolve().strongly_consistent().await?;
let effects = get_effects(source_op).await?;
Ok(ContentSourceWithIssues { source_op, effects }.cell())
}
@@ -238,7 +238,7 @@ impl DevServerBuilder {
http::process_request_with_content_source(
// HACK: pass `source` here (instead of `resolved_source`
// because the underlying API wants to do it's own
// `resolve_strongly_consistent` call.
// `.resolve().strongly_consistent()` call.
//
// It's unlikely (the calls happen one-after-another), but this
// could cause inconsistency between the reported issues and
turbopack/crates/turbopack-dev-server/src/source/resolve.rs
+4 -2
View File
@@ -88,7 +88,8 @@ pub async fn resolve_source_request(
let mut request_overwrites = (*request).clone();
let mut response_header_overwrites = Vec::new();
let mut route_tree = content_source_get_routes_operation(source)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
'routes: loop {
let mut sources_op = route_tree_get_operation(route_tree, current_asset_path.clone());
@@ -140,7 +141,8 @@ pub async fn resolve_source_request(
request_overwrites.uri = new_uri;
current_asset_path = new_asset_path.into();
route_tree = content_source_get_routes_operation(*source)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
continue 'routes;
}
turbopack/crates/turbopack-tests/tests/execution.rs
+1 -1
View File
@@ -612,7 +612,7 @@ async fn snapshot_issues(
run_result_op: OperationVc<RunTestResult>,
) -> Result<Vc<()>> {
let PreparedTest { path, .. } = &*prepared_test.await?;
let _ = run_result_op.resolve_strongly_consistent().await;
let _ = run_result_op.resolve().strongly_consistent().await;
let plain_issues = run_result_op
.peek_issues()
turbopack/crates/turbopack-tests/tests/snapshot.rs
+7 -2
View File
@@ -224,7 +224,12 @@ async fn run(resource: PathBuf) -> Result<()> {
#[turbo_tasks::function(operation)]
async fn inner_operation(resource: RcStr) -> Result<Vc<()>> {
let out_op = run_test_operation(resource);
let out_vc = out_op.resolve_strongly_consistent().await?.owned().await?;
let out_vc = out_op
.resolve()
.strongly_consistent()
.await?
.owned()
.await?;
let plain_issues = out_op
.peek_issues()
@@ -239,7 +244,7 @@ async fn run(resource: PathBuf) -> Result<()> {
#[turbo_tasks::function(operation)]
async fn extract_effects(op: OperationVc<()>) -> Result<Vc<Effects>> {
let _ = op.resolve_strongly_consistent().await?;
let _ = op.resolve().strongly_consistent().await?;
Ok(get_effects(op).await?.cell())
}
turbopack/crates/turbopack-tracing/tests/node-file-trace.rs
+2 -1
View File
@@ -485,7 +485,8 @@ fn node_file_trace<B: Backend + 'static>(
input.clone(),
directory.clone(),
)
.resolve_strongly_consistent()
.resolve()
.strongly_consistent()
.await?;
let duration = before_start.elapsed();