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

Turbopack: Define Effect as a trait instead of a closure (#89080)

Browse Source 
This switches effects (special `Collectible`s used to defer file writes) from using a closure representation to using a trait representation.

This makes constructing effects a little more annoying, but it ensures that we'll be able to correctly implement `TraceRawVcs` on these in a subsequent PR.

Changes to `turbo-tasks/src/effect.rs` were done by hand, an LLM was used for updating the two callsites.
This commit is contained in:
Benjamin Woodruff
2026-03-24 14:34:40 -07:00
committed by GitHub
parent 43f9017719
commit 468a40fa32
3 changed files with 408 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
turbopack/crates/turbo-tasks-fs/src/lib.rs
+337 -263
View File
@@ -36,6 +36,7 @@ use std::{
borrow::Cow,
cmp::{Ordering, min},
env,
error::Error as StdError,
fmt::{self, Debug, Formatter},
fs::FileType,
future::Future,
@@ -63,9 +64,9 @@ use tokio::{
use tracing::Instrument;
use turbo_rcstr::{RcStr, rcstr};
use turbo_tasks::{
ApplyEffectsContext, Completion, InvalidationReason, Invalidator, NonLocalValue, ReadRef,
ResolvedVc, TaskInput, TurboTasksApi, ValueToString, ValueToStringRef, Vc,
debug::ValueDebugFormat, effect, mark_session_dependent, parallel, trace::TraceRawVcs,
ApplyEffectsContext, Completion, Effect, InvalidationReason, Invalidator, NonLocalValue,
ReadRef, ResolvedVc, TaskInput, TurboTasksApi, ValueToString, ValueToStringRef, Vc,
debug::ValueDebugFormat, emit_effect, mark_session_dependent, parallel, trace::TraceRawVcs,
turbo_tasks_weak, turbobail, turbofmt,
};
use turbo_tasks_hash::{
@@ -975,114 +976,136 @@ impl FileSystem for DiskFileSystem {
let inner = self.inner.clone();
let invalidator = turbo_tasks::get_invalidator();
effect(async move {
let full_path = validate_path_length(&full_path)?;
struct WriteEffect {
full_path: PathBuf,
inner: Arc<DiskFileSystemInner>,
invalidator: Option<Invalidator>,
content: ReadRef<FileContent>,
}
let _lock = inner.lock_path(&full_path).await;
impl Effect for WriteEffect {
type Error = AnyhowWrapper;
// Track the file, so that we will rewrite it if it ever changes.
let old_invalidators = invalidator
.map(|invalidator| {
inner.register_write_invalidator(
&full_path,
invalidator,
WriteContent::File(content.clone()),
)
})
.transpose()?
.unwrap_or_default();
async fn apply(self) -> Result<(), Self::Error> {
let full_path = validate_path_length(&self.full_path)?;
// We perform an untracked comparison here, so that this write is not dependent
// on a read's Vc<FileContent> (and the memory it holds). Our untracked read can
// be freed immediately. Given this is an output file, it's unlikely any Turbo
// code will need to read the file from disk into a Vc<FileContent>, so we're
// not wasting cycles.
let compare = content
.streaming_compare(&full_path)
.instrument(tracing::info_span!("read file before write", name = ?full_path))
.concurrency_limited(&inner.read_semaphore)
.await?;
if compare == FileComparison::Equal {
if !old_invalidators.is_empty() {
for (invalidator, write_content) in old_invalidators {
inner.invalidator_map.insert(
full_path.clone().into_owned(),
let _lock = self.inner.lock_path(&full_path).await;
// Track the file, so that we will rewrite it if it ever changes.
let old_invalidators = self
.invalidator
.map(|invalidator| {
self.inner.register_write_invalidator(
&full_path,
invalidator,
write_content,
);
WriteContent::File(self.content.clone()),
)
})
.transpose()?
.unwrap_or_default();
// We perform an untracked comparison here, so that this write is not dependent
// on a read's Vc<FileContent> (and the memory it holds). Our untracked read can
// be freed immediately. Given this is an output file, it's unlikely any Turbo
// code will need to read the file from disk into a Vc<FileContent>, so we're
// not wasting cycles.
let compare = self
.content
.streaming_compare(&full_path)
.instrument(tracing::info_span!("read file before write", name = ?full_path))
.concurrency_limited(&self.inner.read_semaphore)
.await?;
if compare == FileComparison::Equal {
if !old_invalidators.is_empty() {
for (invalidator, write_content) in old_invalidators {
self.inner.invalidator_map.insert(
full_path.clone().into_owned(),
invalidator,
write_content,
);
}
}
return Ok(());
}
return Ok(());
}
match &*content {
FileContent::Content(..) => {
let create_directory = compare == FileComparison::Create;
if create_directory && let Some(parent) = full_path.parent() {
inner.create_directory(parent).await.with_context(|| {
format!(
"failed to create directory {parent:?} for write to {full_path:?}",
)
})?;
}
match &*self.content {
FileContent::Content(..) => {
let create_directory = compare == FileComparison::Create;
if create_directory && let Some(parent) = full_path.parent() {
self.inner.create_directory(parent).await.with_context(|| {
format!(
"failed to create directory {parent:?} for write to \
{full_path:?}",
)
})?;
}
let content = content.clone();
retry_blocking(|| {
let mut f = std::fs::File::create(&full_path)?;
let FileContent::Content(file) = &*content else {
unreachable!()
};
std::io::copy(&mut file.read(), &mut f)?;
#[cfg(unix)]
f.set_permissions(file.meta.permissions.into())?;
f.flush()?;
static WRITE_VERSION: LazyLock<bool> = LazyLock::new(|| {
std::env::var_os("TURBO_ENGINE_WRITE_VERSION")
.is_some_and(|v| v == "1" || v == "true")
});
if *WRITE_VERSION {
let mut full_path = full_path.clone().into_owned();
let hash = hash_xxh3_hash64(file);
let ext = full_path.extension();
let ext = if let Some(ext) = ext {
format!("{:016x}.{}", hash, ext.to_string_lossy())
} else {
format!("{hash:016x}")
};
full_path.set_extension(ext);
let content = self.content.clone();
retry_blocking(|| {
let mut f = std::fs::File::create(&full_path)?;
let FileContent::Content(file) = &*content else {
unreachable!()
};
std::io::copy(&mut file.read(), &mut f)?;
#[cfg(unix)]
f.set_permissions(file.meta.permissions.into())?;
f.flush()?;
}
Ok::<(), io::Error>(())
})
.instrument(tracing::info_span!("write file", name = ?full_path))
.concurrency_limited(&inner.write_semaphore)
.await
.with_context(|| format!("failed to write to {full_path:?}"))?;
}
FileContent::NotFound => {
retry_blocking(|| std::fs::remove_file(&full_path))
.instrument(tracing::info_span!("remove file", name = ?full_path))
.concurrency_limited(&inner.write_semaphore)
.await
.or_else(|err| {
if err.kind() == ErrorKind::NotFound {
Ok(())
} else {
Err(err)
static WRITE_VERSION: LazyLock<bool> = LazyLock::new(|| {
std::env::var_os("TURBO_ENGINE_WRITE_VERSION")
.is_some_and(|v| v == "1" || v == "true")
});
if *WRITE_VERSION {
let mut full_path = full_path.clone().into_owned();
let hash = hash_xxh3_hash64(file);
let ext = full_path.extension();
let ext = if let Some(ext) = ext {
format!("{:016x}.{}", hash, ext.to_string_lossy())
} else {
format!("{hash:016x}")
};
full_path.set_extension(ext);
let mut f = std::fs::File::create(&full_path)?;
std::io::copy(&mut file.read(), &mut f)?;
#[cfg(unix)]
f.set_permissions(file.meta.permissions.into())?;
f.flush()?;
}
Ok::<(), io::Error>(())
})
.with_context(|| format!("removing {full_path:?} failed"))?;
.instrument(tracing::info_span!("write file", name = ?full_path))
.concurrency_limited(&self.inner.write_semaphore)
.await
.with_context(|| format!("failed to write to {full_path:?}"))?;
}
FileContent::NotFound => {
retry_blocking(|| std::fs::remove_file(&full_path))
.instrument(tracing::info_span!("remove file", name = ?full_path))
.concurrency_limited(&self.inner.write_semaphore)
.await
.or_else(|err| {
if err.kind() == ErrorKind::NotFound {
Ok(())
} else {
Err(err)
}
})
.with_context(|| format!("removing {full_path:?} failed"))?;
}
}
self.inner
.invalidate_from_write(&full_path, old_invalidators);
Ok(())
}
}
inner.invalidate_from_write(&full_path, old_invalidators);
Ok(())
emit_effect(WriteEffect {
full_path,
inner,
invalidator,
content,
});
Ok(())
@@ -1105,196 +1128,219 @@ impl FileSystem for DiskFileSystem {
let inner = self.inner.clone();
let invalidator = turbo_tasks::get_invalidator();
effect(async move {
let full_path = validate_path_length(&full_path)?;
struct WriteLinkEffect {
full_path: PathBuf,
inner: Arc<DiskFileSystemInner>,
invalidator: Option<Invalidator>,
content: ReadRef<LinkContent>,
}
let _lock = inner.lock_path(&full_path).await;
impl Effect for WriteLinkEffect {
type Error = AnyhowWrapper;
let old_invalidators = invalidator
.map(|invalidator| {
inner.register_write_invalidator(
&full_path,
invalidator,
WriteContent::Link(content.clone()),
)
})
.transpose()?
.unwrap_or_default();
async fn apply(self) -> Result<(), Self::Error> {
let full_path = validate_path_length(&self.full_path)?;
enum OsSpecificLinkContent {
Link {
#[cfg(windows)]
is_directory: bool,
target: PathBuf,
},
NotFound,
Invalid,
}
let _lock = self.inner.lock_path(&full_path).await;
let os_specific_link_content = match &*content {
LinkContent::Link { target, link_type } => {
let is_directory = link_type.contains(LinkType::DIRECTORY);
let target_path = if link_type.contains(LinkType::ABSOLUTE) {
Path::new(&inner.root).join(unix_to_sys(target).as_ref())
} else {
let relative_target = PathBuf::from(unix_to_sys(target).as_ref());
if cfg!(windows) && is_directory {
// Windows junction points must always be stored as absolute
full_path
.parent()
.unwrap_or(&full_path)
.join(relative_target)
let old_invalidators = self
.invalidator
.map(|invalidator| {
self.inner.register_write_invalidator(
&full_path,
invalidator,
WriteContent::Link(self.content.clone()),
)
})
.transpose()?
.unwrap_or_default();
enum OsSpecificLinkContent {
Link {
#[cfg(windows)]
is_directory: bool,
target: PathBuf,
},
NotFound,
Invalid,
}
let os_specific_link_content = match &*self.content {
LinkContent::Link { target, link_type } => {
let is_directory = link_type.contains(LinkType::DIRECTORY);
let target_path = if link_type.contains(LinkType::ABSOLUTE) {
Path::new(&self.inner.root).join(unix_to_sys(target).as_ref())
} else {
relative_target
let relative_target = PathBuf::from(unix_to_sys(target).as_ref());
if cfg!(windows) && is_directory {
// Windows junction points must always be stored as absolute
full_path
.parent()
.unwrap_or(&full_path)
.join(relative_target)
} else {
relative_target
}
};
OsSpecificLinkContent::Link {
#[cfg(windows)]
is_directory,
target: target_path,
}
};
}
LinkContent::Invalid => OsSpecificLinkContent::Invalid,
LinkContent::NotFound => OsSpecificLinkContent::NotFound,
};
let old_content = match retry_blocking(|| std::fs::read_link(&full_path))
.instrument(tracing::info_span!("read symlink before write", name = ?full_path))
.concurrency_limited(&self.inner.read_semaphore)
.await
{
Ok(res) => Some((res.is_absolute(), res)),
Err(_) => None,
};
let is_equal = match (&os_specific_link_content, &old_content) {
(
OsSpecificLinkContent::Link { target, .. },
Some((old_is_absolute, old_target)),
) => target == old_target && target.is_absolute() == *old_is_absolute,
(OsSpecificLinkContent::NotFound, None) => true,
_ => false,
};
if is_equal {
if !old_invalidators.is_empty() {
for (invalidator, write_content) in old_invalidators {
self.inner.invalidator_map.insert(
full_path.clone().into_owned(),
invalidator,
write_content,
);
}
}
return Ok(());
}
match os_specific_link_content {
OsSpecificLinkContent::Link {
target,
#[cfg(windows)]
is_directory,
target: target_path,
}
}
LinkContent::Invalid => OsSpecificLinkContent::Invalid,
LinkContent::NotFound => OsSpecificLinkContent::NotFound,
};
..
} => {
let full_path = full_path.into_owned();
let old_content = match retry_blocking(|| std::fs::read_link(&full_path))
.instrument(tracing::info_span!("read symlink before write", name = ?full_path))
.concurrency_limited(&inner.read_semaphore)
.await
{
Ok(res) => Some((res.is_absolute(), res)),
Err(_) => None,
};
let is_equal = match (&os_specific_link_content, &old_content) {
(
OsSpecificLinkContent::Link { target, .. },
Some((old_is_absolute, old_target)),
) => target == old_target && target.is_absolute() == *old_is_absolute,
(OsSpecificLinkContent::NotFound, None) => true,
_ => false,
};
if is_equal {
if !old_invalidators.is_empty() {
for (invalidator, write_content) in old_invalidators {
inner.invalidator_map.insert(
full_path.clone().into_owned(),
invalidator,
write_content,
);
}
}
return Ok(());
}
let create_directory = old_content.is_none();
if create_directory && let Some(parent) = full_path.parent() {
self.inner.create_directory(parent).await.with_context(|| {
format!(
"failed to create directory {parent:?} for write link to \
{full_path:?}",
)
})?;
}
match os_specific_link_content {
OsSpecificLinkContent::Link {
target,
#[cfg(windows)]
is_directory,
..
} => {
let full_path = full_path.into_owned();
#[derive(thiserror::Error, Debug)]
#[error("{msg}: {source}")]
struct SymlinkCreationError {
msg: &'static str,
#[source]
source: io::Error,
}
let create_directory = old_content.is_none();
if create_directory && let Some(parent) = full_path.parent() {
inner.create_directory(parent).await.with_context(|| {
format!(
"failed to create directory {parent:?} for write link to \
{full_path:?}",
)
})?;
}
#[derive(thiserror::Error, Debug)]
#[error("{msg}: {source}")]
struct SymlinkCreationError {
msg: &'static str,
#[source]
source: io::Error,
}
let mut has_old_content = old_content.is_some();
let try_create_link = || {
if has_old_content {
// Remove existing symlink before creating a new one. On Unix,
// symlink(2) fails with EEXIST if the link already exists instead of
// overwriting it. Windows has similar behavior with junction points.
remove_symbolic_link_dir_helper(&full_path).map_err(|err| {
let mut has_old_content = old_content.is_some();
let try_create_link = || {
if has_old_content {
// Remove existing symlink before creating a new one. On Unix,
// symlink(2) fails with EEXIST if the link already exists instead
// of overwriting it. Windows has similar behavior with junction
// points.
remove_symbolic_link_dir_helper(&full_path).map_err(|err| {
SymlinkCreationError {
msg: "removal of existing symbolic link or junction point \
failed",
source: err,
}
})?;
has_old_content = false;
}
#[cfg(not(windows))]
let io_result = std::os::unix::fs::symlink(&target, &full_path);
#[cfg(windows)]
let io_result = if is_directory {
std::os::windows::fs::junction_point(&target, &full_path)
} else {
std::os::windows::fs::symlink_file(&target, &full_path)
};
io_result.map_err(|err| {
if err.kind() == ErrorKind::AlreadyExists {
// try to remove the symlink on the next iteration of the loop
has_old_content = true;
}
SymlinkCreationError {
msg: "removal of existing symbolic link or junction point \
failed",
msg: "creation of a new symbolic link or junction point failed",
source: err,
}
})?;
has_old_content = false;
})
};
fn can_retry_link(err: &SymlinkCreationError) -> bool {
err.source.kind() == ErrorKind::AlreadyExists || can_retry(&err.source)
}
#[cfg(not(windows))]
let io_result = std::os::unix::fs::symlink(&target, &full_path);
#[cfg(windows)]
let io_result = if is_directory {
std::os::windows::fs::junction_point(&target, &full_path)
} else {
std::os::windows::fs::symlink_file(&target, &full_path)
let err_context = || {
#[cfg(not(windows))]
let message = format!(
"failed to create symlink at {full_path:?} pointing to {target:?}"
);
#[cfg(windows)]
let message = if is_directory {
format!(
"failed to create junction point at {full_path:?} pointing to \
{target:?}"
)
} else {
format!(
"failed to create symlink at {full_path:?} pointing to \
{target:?}\n\
(Note: creating file symlinks on Windows require developer \
mode or admin permissions: \
https://learn.microsoft.com/en-us/windows/advanced-settings/developer-mode)",
)
};
message
};
io_result.map_err(|err| {
if err.kind() == ErrorKind::AlreadyExists {
// try to remove the symlink on the next iteration of the loop
has_old_content = true;
}
SymlinkCreationError {
msg: "creation of a new symbolic link or junction point failed",
source: err,
}
})
};
fn can_retry_link(err: &SymlinkCreationError) -> bool {
err.source.kind() == ErrorKind::AlreadyExists || can_retry(&err.source)
retry_blocking_custom(try_create_link, can_retry_link)
.instrument(tracing::info_span!(
"write symlink",
name = ?full_path,
target = ?target,
))
.concurrency_limited(&self.inner.write_semaphore)
.await
.with_context(err_context)?;
}
OsSpecificLinkContent::Invalid => {
return Err(AnyhowWrapper(anyhow!(
"invalid symlink target: {full_path:?}"
)));
}
OsSpecificLinkContent::NotFound => {
retry_blocking(|| remove_symbolic_link_dir_helper(&full_path))
.instrument(tracing::info_span!("remove symlink", name = ?full_path))
.concurrency_limited(&self.inner.write_semaphore)
.await
.with_context(|| format!("removing {full_path:?} failed"))?;
}
let err_context = || {
#[cfg(not(windows))]
let message = format!(
"failed to create symlink at {full_path:?} pointing to {target:?}"
);
#[cfg(windows)]
let message = if is_directory {
format!(
"failed to create junction point at {full_path:?} pointing to \
{target:?}"
)
} else {
format!(
"failed to create symlink at {full_path:?} pointing to {target:?}\n\
(Note: creating file symlinks on Windows require developer mode or \
admin permissions: \
https://learn.microsoft.com/en-us/windows/advanced-settings/developer-mode)",
)
};
message
};
retry_blocking_custom(try_create_link, can_retry_link)
.instrument(tracing::info_span!(
"write symlink",
name = ?full_path,
target = ?target,
))
.concurrency_limited(&inner.write_semaphore)
.await
.with_context(err_context)?;
}
OsSpecificLinkContent::Invalid => {
bail!("invalid symlink target: {full_path:?}")
}
OsSpecificLinkContent::NotFound => {
retry_blocking(|| remove_symbolic_link_dir_helper(&full_path))
.instrument(tracing::info_span!("remove symlink", name = ?full_path))
.concurrency_limited(&inner.write_semaphore)
.await
.with_context(|| format!("removing {full_path:?} failed"))?;
}
}
Ok(())
Ok(())
}
}
emit_effect(WriteLinkEffect {
full_path,
inner,
invalidator,
content,
});
Ok(())
}
@@ -2815,6 +2861,34 @@ async fn realpath_with_links(path: FileSystemPath) -> Result<Vc<RealPathResult>>
.cell())
}
/// Wrapper to convert `anyhow::Error` to `impl std::error::Error` for use in `Effect::apply`.
// TODO(bgw): use a structured error type instead of anyhow for write/write_link
struct AnyhowWrapper(anyhow::Error);
impl std::fmt::Display for AnyhowWrapper {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
std::fmt::Display::fmt(&self.0, f)
}
}
impl std::fmt::Debug for AnyhowWrapper {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
std::fmt::Debug::fmt(&self.0, f)
}
}
impl StdError for AnyhowWrapper {
fn source(&self) -> Option<&(dyn StdError + 'static)> {
self.0.source()
}
}
impl From<anyhow::Error> for AnyhowWrapper {
fn from(err: anyhow::Error) -> Self {
AnyhowWrapper(err)
}
}
#[cfg(test)]
mod tests {
use turbo_rcstr::rcstr;
turbopack/crates/turbo-tasks/src/effect.rs
+70 -33
View File
@@ -1,5 +1,6 @@
use std::{
any::{Any, TypeId},
error::Error as StdError,
future::Future,
mem::replace,
panic,
@@ -19,31 +20,66 @@ use crate::{
self as turbo_tasks, CollectiblesSource, ReadRef, ResolvedVc, TryJoinIterExt, emit,
event::{Event, EventListener},
spawn,
util::SharedError,
};
const APPLY_EFFECTS_CONCURRENCY_LIMIT: usize = 1024;
/// A trait to emit a task effect as collectible. This trait only has one
/// implementation, `EffectInstance` and no other implementation is allowed.
/// The trait is private to this module so that no other implementation can be
/// added.
#[turbo_tasks::value_trait]
trait Effect {}
pub trait Effect: Send + Sync + 'static {
type Error: StdError + Send + Sync + 'static;
/// A future that represents the effect of a task. The future is executed when
/// the effect is applied.
type EffectFuture = Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'static>>;
/// The inner state of an effect instance if it has not been applied yet.
struct EffectInner {
future: EffectFuture,
/// A function that is called once at the top level of the program's execution after everything
/// has "settled".
///
/// This function is executed outside of the turbo-tasks context, and therefore cannot read any
/// `Vc`s or call any turbo-task functions. The effect can store [`ResolvedVc`]s (or any other
/// `Vc` type), but should not read or resolve their contents.
fn apply(self) -> impl Future<Output = Result<(), Self::Error>> + Send;
}
/// The error type that an effect can return. We use `dyn std::error::Error` (instead of
/// [`anyhow::Error`] or [`SharedError`]) to encourage use of structured error types that can
/// potentially be transformed into `Issue`s.
///
/// We can't require that the returned error implements `Issue`:
/// - `Issue` uses `FileSystemPath`
/// - `turbo-tasks-fs` returns effect errors that should be transformed into `Issue`s.
/// - It logically doesn't make sense to define `Issue` in `turbo-tasks-fs`, `Issue` can't be
/// defined in a base crate either because it would form a circular crate dependency.
///
/// So instead, we leave it up to the caller to figure out how to downcast these errors themselves.
///
/// [`SharedError`]: crate::util::SharedError
type EffectError = Arc<dyn StdError + Send + Sync + 'static>;
// Private wrapper trait to allow dynamic dispatch of an `Effect`. This is similar to the pattern
// that the dynosaur crate uses: https://github.com/spastorino/dynosaur
trait DynEffect: Send + Sync + 'static {
fn dyn_apply(self: Box<Self>) -> Pin<Box<dyn Future<Output = Result<(), EffectError>> + Send>>;
}
impl<T> DynEffect for T
where
T: Effect,
{
fn dyn_apply(self: Box<Self>) -> Pin<Box<dyn Future<Output = Result<(), EffectError>> + Send>> {
Box::pin(async move {
self.apply()
.await
.map_err(|err| Arc::new(err) as EffectError)
})
}
}
/// A trait to emit a task effect as collectible. This trait only has one implementation,
/// `EffectInstance` and no other implementation is allowed. The trait is private to this module so
/// that no other implementation can be added.
#[turbo_tasks::value_trait]
trait EffectCollectible {}
enum EffectState {
NotStarted(EffectInner),
NotStarted(Box<dyn DynEffect>),
Started(Event),
Finished(Result<(), SharedError>),
Finished(Result<(), EffectError>),
}
/// The Effect instance collectible that is emitted for effects.
@@ -54,11 +90,11 @@ struct EffectInstance {
}
impl EffectInstance {
fn new(future: impl Future<Output = Result<()>> + Send + Sync + 'static) -> Self {
fn new(effect: impl Effect) -> Self {
Self {
inner: Mutex::new(EffectState::NotStarted(EffectInner {
future: Box::pin(future),
})),
inner: Mutex::new(EffectState::NotStarted(
Box::new(effect) as Box<dyn DynEffect>
)),
}
}
@@ -66,7 +102,7 @@ impl EffectInstance {
loop {
enum State {
Started(EventListener),
NotStarted(EffectInner),
NotStarted(Box<dyn DynEffect>),
}
let state = {
let mut guard = self.inner.lock();
@@ -93,10 +129,11 @@ impl EffectInstance {
State::Started(listener) => {
listener.await;
}
State::NotStarted(EffectInner { future }) => {
let join_handle = spawn(ApplyEffectsContext::in_current_scope(future));
State::NotStarted(effect) => {
let join_handle =
spawn(ApplyEffectsContext::in_current_scope(effect.dyn_apply()));
let result = match join_handle.await {
Err(err) => Err(SharedError::new(err)),
Err(err) => Err(err),
Ok(()) => Ok(()),
};
let event = {
@@ -117,19 +154,19 @@ impl EffectInstance {
}
#[turbo_tasks::value_impl]
impl Effect for EffectInstance {}
impl EffectCollectible for EffectInstance {}
/// Schedules an effect to be applied. The passed future is executed once `apply_effects` is called.
/// Emits an effect to be applied. The effect is executed once `apply_effects` is called.
///
/// The effect will only executed once. The passed future is executed outside of the current task
/// and can't read any Vcs. These need to be read before. ReadRefs can be passed into the future.
/// The effect will only executed once. The effect is executed outside of the current task
/// and can't read any Vcs. These need to be read before. ReadRefs can be passed into the effect.
///
/// Effects are executed in parallel, so they might need to use async locking to avoid problems.
/// Order of execution of multiple effects is not defined. You must not use multiple conflicting
/// effects to avoid non-deterministic behavior.
pub fn effect(future: impl Future<Output = Result<()>> + Send + Sync + 'static) {
emit::<Box<dyn Effect>>(ResolvedVc::upcast(
EffectInstance::new(future).resolved_cell(),
pub fn emit_effect(effect: impl Effect) {
emit::<Box<dyn EffectCollectible>>(ResolvedVc::upcast(
EffectInstance::new(effect).resolved_cell(),
));
}
@@ -151,7 +188,7 @@ pub fn effect(future: impl Future<Output = Result<()>> + Send + Sync + 'static)
/// apply_effects(operation).await?;
/// ```
pub async fn apply_effects(source: impl CollectiblesSource) -> Result<()> {
let effects: AutoSet<ResolvedVc<Box<dyn Effect>>> = source.take_collectibles();
let effects: AutoSet<ResolvedVc<Box<dyn EffectCollectible>>> = source.take_collectibles();
if effects.is_empty() {
return Ok(());
}
@@ -194,7 +231,7 @@ pub async fn apply_effects(source: impl CollectiblesSource) -> Result<()> {
/// result_with_effects.effects.apply().await?;
/// ```
pub async fn get_effects(source: impl CollectiblesSource) -> Result<Effects> {
let effects: AutoSet<ResolvedVc<Box<dyn Effect>>> = source.take_collectibles();
let effects: AutoSet<ResolvedVc<Box<dyn EffectCollectible>>> = source.take_collectibles();
let effects = effects
.into_iter()
.map(|effect| async move {
turbopack/crates/turbo-tasks/src/lib.rs
+1 -1
View File
@@ -78,7 +78,7 @@ pub use crate::{
collectibles::CollectiblesSource,
completion::{Completion, Completions},
display::{ValueToString, ValueToStringRef},
effect::{ApplyEffectsContext, Effects, apply_effects, effect, get_effects},
effect::{ApplyEffectsContext, Effect, Effects, apply_effects, emit_effect, get_effects},
error::PrettyPrintError,
id::{ExecutionId, LocalTaskId, TRANSIENT_TASK_BIT, TaskId, TraitTypeId, ValueTypeId},
invalidation::{