turbo_tasks/

manager.rs

use std::{
    borrow::Borrow,
    cell::Cell,
    cmp::Reverse,
    fmt::{Debug, Display},
    future::Future,
    hash::{BuildHasher, BuildHasherDefault, Hash},
    mem::take,
    ops::Deref,
    panic::AssertUnwindSafe,
    pin::Pin,
    process::abort,
    sync::{
        Arc, Mutex, RwLock, Weak,
        atomic::{AtomicBool, AtomicUsize, Ordering},
    },
    task::{Context, Poll, Waker},
    time::{Duration, Instant},
};

use anyhow::{Result, anyhow};
use auto_hash_map::AutoMap;
use bincode::{Decode, Encode};
use either::Either;
use futures::FutureExt;
use rustc_hash::{FxBuildHasher, FxHasher};
use serde::{Deserialize, Serialize};
use smallvec::SmallVec;
use tokio::{select, sync::mpsc::Receiver, task_local};
use tracing::{Instrument, Span, instrument};
use turbo_tasks_hash::{DeterministicHash, hash_xxh3_hash128};

use crate::{
    CellId, Completion, InvalidationReason, InvalidationReasonSet, NonLocalValue, OperationValue,
    OperationVc, OutputContent, RawVc, ReadCellOptions, ReadOutcome, ReadOutputOptions, ResolvedVc,
    SharedReference, TaskId, TraitMethod, ValueTypeId, Vc, VcRead, VcValueTrait, VcValueType,
    backend::{
        Backend, CellContent, CellHash, TaskCollectiblesMap, TaskExecutionSpec, TransientTaskType,
        TurboTasksExecutionError, TypedCellContent, VerificationMode,
    },
    capture_future::CaptureFuture,
    dyn_task_inputs::DynTaskInputsStorage,
    event::{Event, EventListener},
    id::{ExecutionId, LocalTaskId, TraitTypeId},
    keyed::KeyedEq,
    local_task_tracker::LocalTaskTracker,
    macro_helpers::NativeFunction,
    message_queue::{CompilationEvent, CompilationEventQueue},
    priority_runner::{Claimable, Executor, PriorityRunner},
    registry,
    serialization_invalidation::SerializationInvalidator,
    task::local_task::{LocalTask, LocalTaskSpec, LocalTaskType},
    task_statistics::TaskStatisticsApi,
    util::{IdFactory, StaticOrArc},
};

/// Common base trait for [`TurboTasksApi`] and [`TurboTasks`]. Provides APIs for creating tasks
/// from function calls.
pub trait TurboTasksCallApi: Sync + Send {
    /// Calls a native function with arguments. Resolves arguments when needed
    /// with a wrapper task.
    ///
    /// `inputs_resolved` is `TaskInput::is_resolved(&args)` computed at the macro callsite on
    /// the concrete tuple type — when [`InputResolution::Resolved`], the fast path skips wrapper
    /// task creation.
    fn dynamic_call(
        &self,
        native_fn: &'static NativeFunction,
        this: Option<RawVc>,
        arg: &mut dyn DynTaskInputsStorage,
        inputs_resolved: InputResolution,
        persistence: TaskPersistence,
    ) -> RawVc;
    /// Call a native function with arguments.
    /// All inputs must be resolved.
    fn native_call(
        &self,
        native_fn: &'static NativeFunction,
        this: Option<RawVc>,
        arg: &mut dyn DynTaskInputsStorage,
        persistence: TaskPersistence,
    ) -> RawVc;
    /// Calls a trait method with arguments. First input is the `self` object.
    /// Uses a wrapper task to resolve.
    ///
    /// `inputs_resolved` is the macro-site `InputResolution` of the *exposed* tuple; when filtering
    /// is involved, the post-filter value is computed inside the filter functor and supersedes
    /// this argument.
    fn trait_call(
        &self,
        trait_method: &'static TraitMethod,
        this: RawVc,
        arg: &mut dyn DynTaskInputsStorage,
        inputs_resolved: InputResolution,
        persistence: TaskPersistence,
    ) -> RawVc;

    fn run(
        &self,
        future: Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>,
    ) -> Pin<Box<dyn Future<Output = Result<(), TurboTasksExecutionError>> + Send>>;
    fn run_once(
        &self,
        future: Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
    fn run_once_with_reason(
        &self,
        reason: StaticOrArc<dyn InvalidationReason>,
        future: Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
    fn start_once_process(&self, future: Pin<Box<dyn Future<Output = ()> + Send + 'static>>);

    /// Sends a compilation event to subscribers.
    fn send_compilation_event(&self, event: Arc<dyn CompilationEvent>);

    /// Returns a human-readable name for the given task.
    fn get_task_name(&self, task: TaskId) -> String;
}

/// A type-erased subset of [`TurboTasks`] stored inside a thread local when we're in a turbo task
/// context. Returned by the [`turbo_tasks`] helper function.
///
/// This trait is needed because thread locals cannot contain an unresolved [`Backend`] type
/// parameter.
pub trait TurboTasksApi: TurboTasksCallApi + Sync + Send {
    fn invalidate(&self, task: TaskId);
    fn invalidate_with_reason(&self, task: TaskId, reason: StaticOrArc<dyn InvalidationReason>);

    fn invalidate_serialization(&self, task: TaskId);

    fn try_read_task_output(
        &self,
        task: TaskId,
        options: ReadOutputOptions,
    ) -> Result<ReadOutcome<RawVc>>;

    fn try_read_task_cell(
        &self,
        task: TaskId,
        index: CellId,
        options: ReadCellOptions,
    ) -> Result<ReadOutcome<TypedCellContent>>;

    /// Reads a [`RawVc::LocalOutput`]. If the task has completed, returns the [`RawVc`] the local
    /// task points to.
    ///
    /// The returned [`RawVc`] may also be a [`RawVc::LocalOutput`], so this may need to be called
    /// recursively or in a loop.
    ///
    /// This does not accept a consistency argument, as you cannot control consistency of a read of
    /// an operation owned by your own task. Strongly consistent reads are only allowed on
    /// [`OperationVc`]s, which should never be local tasks.
    ///
    /// No dependency tracking will happen as a result of this function call, as it's a no-op for a
    /// task to depend on itself.
    ///
    /// [`OperationVc`]: crate::OperationVc
    fn try_read_local_output(
        &self,
        execution_id: ExecutionId,
        local_task_id: LocalTaskId,
    ) -> Result<Result<RawVc, EventListener>>;

    fn read_task_collectibles(&self, task: TaskId, trait_id: TraitTypeId) -> TaskCollectiblesMap;

    /// Executes a task that is scheduled but not started yet inline on the current thread, so that
    /// a read doesn't have to wait for a worker to pick the task up. Returns whether the task's
    /// execution completed.
    ///
    /// Used by the read paths; see `TurboTasks::try_execute_scheduled_task_inline` for the details.
    fn try_execute_scheduled_task_inline(&self, key: ScheduleKey) -> bool;

    /// Records that a read waited for a task that a worker was already executing, so it did not try
    /// to claim it. Diagnostics only, see `TurboTasks::inline_execution_stats`.
    #[cfg(feature = "inline_execution_stats")]
    fn note_waited_for_in_progress_task(&self);

    fn emit_collectible(&self, trait_type: TraitTypeId, collectible: RawVc);
    fn unemit_collectible(&self, trait_type: TraitTypeId, collectible: RawVc, count: u32);
    fn unemit_collectibles(&self, trait_type: TraitTypeId, collectibles: &TaskCollectiblesMap);

    /// INVALIDATION: Be careful with this, it will not track dependencies, so
    /// using it could break cache invalidation.
    fn try_read_own_task_cell(
        &self,
        current_task: TaskId,
        index: CellId,
    ) -> Result<TypedCellContent>;

    fn read_own_task_cell(&self, task: TaskId, index: CellId) -> Result<TypedCellContent>;
    fn update_own_task_cell(
        &self,
        task: TaskId,
        index: CellId,
        content: CellContent,
        updated_key_hashes: Option<SmallVec<[u64; 2]>>,
        content_hash: Option<CellHash>,
        verification_mode: VerificationMode,
    );
    fn mark_own_task_as_finished(&self, task: TaskId);

    /// Pin a task against garbage collection. Delegates to
    /// [`Backend::pin_task_for_gc`](crate::backend::Backend::pin_task_for_gc).
    fn pin_task_for_gc(&self, task: TaskId);

    /// Removes a pin added by [`pin_task_for_gc`](TurboTasksApi::pin_task_for_gc).
    fn unpin_task_for_gc(&self, task: TaskId);

    fn connect_task(&self, task: TaskId);

    /// Wraps the given future in the current task.
    ///
    /// Beware: this method is not safe to use in production code. It is only intended for use in
    /// tests and for debugging purposes.
    fn spawn_detached_for_testing(&self, f: Pin<Box<dyn Future<Output = ()> + Send + 'static>>);

    fn task_statistics(&self) -> &TaskStatisticsApi;

    fn stop_and_wait(&self) -> Pin<Box<dyn Future<Output = ()> + Send>>;

    fn subscribe_to_compilation_events(
        &self,
        event_types: Option<Vec<String>>,
    ) -> Receiver<Arc<dyn CompilationEvent>>;

    // Returns true if TurboTasks is configured to track dependencies.
    fn is_tracking_dependencies(&self) -> bool;
}

/// A wrapper around a value that is unused.
pub struct Unused<T> {
    inner: T,
}

impl<T> Unused<T> {
    /// Creates a new unused value.
    ///
    /// # Safety
    ///
    /// The wrapped value must not be used.
    pub unsafe fn new_unchecked(inner: T) -> Self {
        Self { inner }
    }

    /// Get the inner value, without consuming the `Unused` wrapper.
    ///
    /// # Safety
    ///
    /// The user need to make sure that the value stays unused.
    pub unsafe fn get_unchecked(&self) -> &T {
        &self.inner
    }

    /// Unwraps the value, consuming the `Unused` wrapper.
    pub fn into(self) -> T {
        self.inner
    }
}

#[allow(clippy::manual_non_exhaustive)]
pub struct UpdateInfo {
    pub duration: Duration,
    pub tasks: usize,
    pub reasons: InvalidationReasonSet,
    #[allow(dead_code)]
    placeholder_for_future_fields: (),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize, Encode, Decode)]
pub enum TaskPersistence {
    /// Tasks that may be persisted across sessions using serialization.
    Persistent,

    /// Tasks that will be persisted in memory for the life of this session, but won't persist
    /// between sessions.
    ///
    /// This is used for [root tasks][TurboTasks::spawn_root_task] and tasks with an argument of
    /// type [`TransientValue`][crate::value::TransientValue] or
    /// [`TransientInstance`][crate::value::TransientInstance].
    Transient,
}

impl Display for TaskPersistence {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            TaskPersistence::Persistent => write!(f, "persistent"),
            TaskPersistence::Transient => write!(f, "transient"),
        }
    }
}

/// Whether a task call's inputs are already resolved, decided on the concrete input tuple at the
/// call site. Travels alongside [`TaskPersistence`] through [`dynamic_call`] / [`trait_call`].
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
pub enum InputResolution {
    /// All inputs (and `this`, where applicable) are resolved — eligible for the synchronous fast
    /// path with no async resolution task.
    Resolved,
    /// At least one input is unresolved and must be resolved in a local task first.
    Unresolved,
}

impl InputResolution {
    #[inline]
    pub fn from_is_resolved(is_resolved: bool) -> Self {
        if is_resolved {
            Self::Resolved
        } else {
            Self::Unresolved
        }
    }

    #[inline]
    pub fn is_resolved(self) -> bool {
        matches!(self, Self::Resolved)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Default)]
pub enum ReadConsistency {
    /// The default behavior for most APIs. Reads are faster, but may return stale values, which
    /// may later trigger re-computation.
    #[default]
    Eventual,
    /// Ensures all dependencies are fully resolved before returning the cell or output data, at
    /// the cost of slower reads.
    ///
    /// Top-level code that returns data to the user should use strongly consistent reads.
    Strong,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ReadCellTracking {
    /// Reads are tracked as dependencies of the current task.
    Tracked {
        /// The key used for the dependency
        key: Option<u64>,
    },
    /// The read is only tracked when there is an error, otherwise it is untracked.
    ///
    /// INVALIDATION: Be careful with this, it will not track dependencies, so
    /// using it could break cache invalidation.
    TrackOnlyError,
    /// The read is not tracked as a dependency of the current task.
    ///
    /// INVALIDATION: Be careful with this, it will not track dependencies, so
    /// using it could break cache invalidation.
    Untracked,
}

impl ReadCellTracking {
    pub fn should_track(&self, is_err: bool) -> bool {
        match self {
            ReadCellTracking::Tracked { .. } => true,
            ReadCellTracking::TrackOnlyError => is_err,
            ReadCellTracking::Untracked => false,
        }
    }

    pub fn key(&self) -> Option<u64> {
        match self {
            ReadCellTracking::Tracked { key } => *key,
            ReadCellTracking::TrackOnlyError => None,
            ReadCellTracking::Untracked => None,
        }
    }
}

impl Default for ReadCellTracking {
    fn default() -> Self {
        ReadCellTracking::Tracked { key: None }
    }
}

impl Display for ReadCellTracking {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ReadCellTracking::Tracked { key: None } => write!(f, "tracked"),
            ReadCellTracking::Tracked { key: Some(key) } => write!(f, "tracked with key {key}"),
            ReadCellTracking::TrackOnlyError => write!(f, "track only error"),
            ReadCellTracking::Untracked => write!(f, "untracked"),
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Default)]
pub enum ReadTracking {
    /// Reads are tracked as dependencies of the current task.
    #[default]
    Tracked,
    /// The read is only tracked when there is an error, otherwise it is untracked.
    ///
    /// INVALIDATION: Be careful with this, it will not track dependencies, so
    /// using it could break cache invalidation.
    TrackOnlyError,
    /// The read is not tracked as a dependency of the current task.
    ///
    /// INVALIDATION: Be careful with this, it will not track dependencies, so
    /// using it could break cache invalidation.
    Untracked,
}

impl ReadTracking {
    pub fn should_track(&self, is_err: bool) -> bool {
        match self {
            ReadTracking::Tracked => true,
            ReadTracking::TrackOnlyError => is_err,
            ReadTracking::Untracked => false,
        }
    }
}

impl Display for ReadTracking {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ReadTracking::Tracked => write!(f, "tracked"),
            ReadTracking::TrackOnlyError => write!(f, "track only error"),
            ReadTracking::Untracked => write!(f, "untracked"),
        }
    }
}

#[derive(Encode, Decode, Default, Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)]
pub enum TaskPriority {
    #[default]
    Initial,
    Invalidation {
        priority: Reverse<u32>,
    },
    Recomputation,
}

impl TaskPriority {
    pub fn invalidation(priority: u32) -> Self {
        Self::Invalidation {
            priority: Reverse(priority),
        }
    }

    pub fn initial() -> Self {
        Self::Initial
    }

    pub fn leaf() -> Self {
        Self::Invalidation {
            priority: Reverse(0),
        }
    }

    pub fn in_parent(&self, parent_priority: TaskPriority) -> Self {
        match self {
            TaskPriority::Initial => parent_priority,
            TaskPriority::Invalidation { priority } => {
                if let TaskPriority::Invalidation {
                    priority: parent_priority,
                } = parent_priority
                    && priority.0 < parent_priority.0
                {
                    Self::Invalidation {
                        priority: Reverse(parent_priority.0.saturating_add(1)),
                    }
                } else {
                    *self
                }
            }
            TaskPriority::Recomputation => TaskPriority::Recomputation,
        }
    }
}

impl Display for TaskPriority {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            TaskPriority::Initial => write!(f, "initial"),
            TaskPriority::Invalidation { priority } => write!(f, "invalidation({})", priority.0),
            TaskPriority::Recomputation => write!(f, "recomputation"),
        }
    }
}

enum ScheduledTask {
    Task {
        task_id: TaskId,
        span: Span,
    },
    LocalTask {
        ty: LocalTaskSpec,
        persistence: TaskPersistence,
        execution_id: ExecutionId,
        local_task_id: LocalTaskId,
        global_task_state: CurrentTaskStateHandle,
        span: Span,
    },
}

/// Identifies a scheduled task, so that a read which is about to wait for it can take it out of the
/// scheduler queue and execute it inline instead (see `PriorityRunner::claim` and
/// `execute_read_target_inline`).
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum ScheduleKey {
    /// A cached (non-local) task.
    Task(TaskId),
    /// A local task, which is only known within the execution that created it.
    LocalTask(ExecutionId, LocalTaskId),
}

impl Claimable for ScheduledTask {
    type Key = ScheduleKey;

    fn claim_key(&self) -> Option<ScheduleKey> {
        Some(match self {
            ScheduledTask::Task { task_id, .. } => ScheduleKey::Task(*task_id),
            ScheduledTask::LocalTask {
                execution_id,
                local_task_id,
                ..
            } => ScheduleKey::LocalTask(*execution_id, *local_task_id),
        })
    }
}

#[cfg(feature = "inline_execution_stats")]
use std::sync::atomic::AtomicU64;

/// Counters describing how reads and inline execution interacted, see
/// [`TurboTasks::inline_execution_stats`]. Diagnostics only.
#[cfg(feature = "inline_execution_stats")]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct InlineExecutionStats {
    /// Tasks that were put into the scheduler queue.
    pub queued: u64,
    /// Reads that tried to take a queued task out of the queue.
    pub claim_attempted: u64,
    /// Claims that succeeded and whose execution finished on the reading thread.
    pub claim_completed: u64,
    /// Claims that succeeded but whose execution yielded, so it was handed to the runtime.
    pub claim_yielded: u64,
    /// Claims that found nothing to take, because a worker had already picked the task up.
    pub claim_failed: u64,
    /// Reads that waited without attempting a claim, because the task was already being executed.
    pub waited_in_progress: u64,
}

/// The counters behind [`InlineExecutionStats`].
///
/// Without the `inline_execution_stats` feature this is zero-sized and every method is an empty
/// `#[inline]` no-op, so the counting compiles away: the counters sit on the read-miss path, and a
/// build that doesn't want the numbers shouldn't pay for them.
#[derive(Default)]
struct InlineExecutionCounters {
    #[cfg(feature = "inline_execution_stats")]
    claim_attempted: AtomicU64,
    #[cfg(feature = "inline_execution_stats")]
    claim_completed: AtomicU64,
    #[cfg(feature = "inline_execution_stats")]
    claim_yielded: AtomicU64,
    #[cfg(feature = "inline_execution_stats")]
    claim_failed: AtomicU64,
    #[cfg(feature = "inline_execution_stats")]
    waited_in_progress: AtomicU64,
}

impl InlineExecutionCounters {
    /// A read found its task queued and tried to take it over.
    #[inline]
    fn claim_attempted(&self) {
        #[cfg(feature = "inline_execution_stats")]
        self.claim_attempted.fetch_add(1, Ordering::Relaxed);
    }

    /// A claim succeeded and the execution finished on the reading thread.
    #[inline]
    fn claim_completed(&self) {
        #[cfg(feature = "inline_execution_stats")]
        self.claim_completed.fetch_add(1, Ordering::Relaxed);
    }

    /// A claim succeeded but the execution yielded, so it was handed to the runtime.
    #[inline]
    fn claim_yielded(&self) {
        #[cfg(feature = "inline_execution_stats")]
        self.claim_yielded.fetch_add(1, Ordering::Relaxed);
    }

    /// A claim found nothing to take, because a worker had already picked the task up.
    #[inline]
    fn claim_failed(&self) {
        #[cfg(feature = "inline_execution_stats")]
        self.claim_failed.fetch_add(1, Ordering::Relaxed);
    }

    /// A read waited without attempting a claim, because the task was already being executed.
    #[cfg(feature = "inline_execution_stats")]
    #[inline]
    fn waited_in_progress(&self) {
        self.waited_in_progress.fetch_add(1, Ordering::Relaxed);
    }
}

/// Whether a dump of [`InlineExecutionStats`] was requested via `TURBO_ENGINE_INLINE_STATS=1`.
#[cfg(feature = "inline_execution_stats")]
pub(crate) fn inline_stats_requested() -> bool {
    static REQUESTED: std::sync::LazyLock<bool> = std::sync::LazyLock::new(|| {
        std::env::var("TURBO_ENGINE_INLINE_STATS").is_ok_and(|value| value != "0")
    });
    *REQUESTED
}

/// Maximum number of task executions that may be nested inline on a single thread, to conserve
/// stack space. (The alternative would be growing the stack on demand, the way SWC does.)
const MAX_INLINE_EXECUTION_DEPTH: usize = 16;

thread_local! {
    /// How many task executions are currently nested inline on this thread.
    static INLINE_EXECUTION_DEPTH: Cell<usize> = const { Cell::new(0) };
}

/// Whether the current thread may execute another task inline, see [`MAX_INLINE_EXECUTION_DEPTH`].
fn inline_execution_allowed() -> bool {
    INLINE_EXECUTION_DEPTH.get() < MAX_INLINE_EXECUTION_DEPTH
}

/// Counts one level of inline task execution on this thread, see [`MAX_INLINE_EXECUTION_DEPTH`].
struct InlineExecutionDepthGuard;

impl InlineExecutionDepthGuard {
    fn enter() -> Self {
        INLINE_EXECUTION_DEPTH.set(INLINE_EXECUTION_DEPTH.get() + 1);
        Self
    }
}

impl Drop for InlineExecutionDepthGuard {
    fn drop(&mut self) {
        INLINE_EXECUTION_DEPTH.set(INLINE_EXECUTION_DEPTH.get() - 1);
    }
}

/// Polls `future` once inline and then spawns it if it doesn't complete so tokio drives it. Returns
/// whether it completed.
fn poll_once_or_spawn(future: impl Future<Output = ()> + Send + 'static) -> bool {
    let _depth_guard = InlineExecutionDepthGuard::enter();
    let span_slot = InlineExecutionSpanSlot::default();
    let mut future = Box::pin(INLINE_EXECUTION_SPAN.scope(span_slot.clone(), future));
    // A waker that never wakes anything is fine here: if this poll doesn't complete the future we
    // spawn it, and a spawned task is always polled at least once, which is the poll that registers
    // the real waker.
    match future
        .as_mut()
        .poll(&mut Context::from_waker(Waker::noop()))
    {
        Poll::Ready(()) => {
            span_slot.record("complete");
            true
        }
        Poll::Pending => {
            span_slot.record("partial");
            tokio::task::spawn(future);
            false
        }
    }
}

/// Executes the task inline if possible, returns true if it executed to completion.
pub(crate) fn execute_read_target_inline(
    turbo_tasks: &dyn TurboTasksApi,
    key: ScheduleKey,
) -> bool {
    if !inline_execution_allowed() {
        // Nested too deeply; a worker will pick the task up, as it always did.
        return false;
    }
    turbo_tasks.try_execute_scheduled_task_inline(key)
}

pub struct TurboTasks<B: Backend + 'static> {
    this: Weak<Self>,
    backend: B,
    execution_id_factory: IdFactory<ExecutionId>,
    stopped: AtomicBool,
    currently_scheduled_foreground_jobs: AtomicUsize,
    currently_scheduled_background_jobs: AtomicUsize,
    scheduled_tasks: AtomicUsize,
    /// Diagnostics for reads and inline execution, see `TurboTasks::inline_execution_stats`.
    /// Zero-sized without the `inline_execution_stats` feature.
    inline_counters: InlineExecutionCounters,
    priority_runner:
        Arc<PriorityRunner<TurboTasks<B>, ScheduledTask, TaskPriority, TurboTasksExecutor>>,
    start: Mutex<Option<Instant>>,
    aggregated_update: Mutex<(Option<(Duration, usize)>, InvalidationReasonSet)>,
    /// Event that is triggered when currently_scheduled_foreground_jobs becomes non-zero
    event_foreground_start: Event,
    /// Event that is triggered when all foreground jobs are done
    /// (currently_scheduled_foreground_jobs becomes zero)
    event_foreground_done: Event,
    /// Event that is triggered when all background jobs are done
    event_background_done: Event,
    compilation_events: CompilationEventQueue,
}

/// Information about a non-local task. A non-local task can contain multiple "local" tasks, which
/// all share the same non-local task state.
///
/// A non-local task is one that:
///
/// - Has a unique task id.
/// - Is potentially cached.
/// - The backend is aware of.
struct CurrentTaskState {
    task_id: Option<TaskId>,
    execution_id: ExecutionId,
    priority: TaskPriority,

    /// True if the current task has state in cells (interior mutability).
    /// Only tracked when verify_determinism feature is enabled.
    #[cfg(feature = "verify_determinism")]
    stateful: bool,

    /// True if the current task uses an external invalidator
    has_invalidator: bool,

    /// True if we're in a top-level task (e.g. `.run_once(...)` or `.run(...)`).
    /// Eventually consistent reads are not allowed in top-level tasks.
    in_top_level_task: bool,

    /// Tracks how many cells of each type has been allocated so far during this task execution.
    /// When a task is re-executed, the cell count may not match the existing cell vec length.
    ///
    /// This is taken (and becomes `None`) during teardown of a task.
    cell_counters: Option<AutoMap<ValueTypeId, u32, BuildHasherDefault<FxHasher>, 8>>,

    /// Tracks execution of Local tasks (and detached test futures) created during this global
    /// task's execution.
    local_tasks: LocalTaskTracker,
}

impl CurrentTaskState {
    fn new(
        task_id: TaskId,
        execution_id: ExecutionId,
        priority: TaskPriority,
        in_top_level_task: bool,
    ) -> Self {
        Self {
            task_id: Some(task_id),
            execution_id,
            priority,
            #[cfg(feature = "verify_determinism")]
            stateful: false,
            has_invalidator: false,
            in_top_level_task,
            cell_counters: Some(AutoMap::default()),
            local_tasks: LocalTaskTracker::new(),
        }
    }

    fn new_temporary(
        execution_id: ExecutionId,
        priority: TaskPriority,
        in_top_level_task: bool,
    ) -> Self {
        Self {
            task_id: None,
            execution_id,
            priority,
            #[cfg(feature = "verify_determinism")]
            stateful: false,
            has_invalidator: false,
            in_top_level_task,
            cell_counters: None,
            local_tasks: LocalTaskTracker::new(),
        }
    }

    fn assert_execution_id(&self, expected_execution_id: ExecutionId) {
        if self.execution_id != expected_execution_id {
            panic!(
                "Local tasks can only be scheduled/awaited within the same execution of the \
                 parent task that created them"
            );
        }
    }
}

/// A shareable current-task state handle with the immutable task ID cached
/// outside the lock. The rest of the state is mutated by global and local
/// tasks, but the task ID is fixed for the lifetime of an execution.
#[derive(Clone)]
struct CurrentTaskStateHandle {
    inner: Arc<CurrentTaskStateInner>,
}

struct CurrentTaskStateInner {
    current_task_id: Option<TaskId>,
    state: RwLock<CurrentTaskState>,
}

impl CurrentTaskStateHandle {
    fn new(state: CurrentTaskState) -> Self {
        Self {
            inner: Arc::new(CurrentTaskStateInner {
                current_task_id: state.task_id,
                state: RwLock::new(state),
            }),
        }
    }

    fn current_task_id(&self) -> Option<TaskId> {
        self.inner.current_task_id
    }
}

impl Deref for CurrentTaskStateHandle {
    type Target = RwLock<CurrentTaskState>;

    fn deref(&self) -> &Self::Target {
        &self.inner.state
    }
}

// TODO implement our own thread pool and make these thread locals instead
task_local! {
    /// The current TurboTasks instance
    static TURBO_TASKS: Arc<dyn TurboTasksApi>;

    static CURRENT_TASK_STATE: CurrentTaskStateHandle;

    /// Temporarily suppresses the eventual consistency check in top-level tasks.
    /// This is used by strongly consistent reads to allow them to succeed in top-level tasks.
    /// This is NOT shared across local tasks (unlike CURRENT_TASK_STATE), so it's safe
    /// to set/unset without race conditions.
    pub(crate) static SUPPRESS_EVENTUAL_CONSISTENCY_TOP_LEVEL_TASK_CHECK: bool;

    /// Set only while a reader polls a task it claimed, so that the outcome of that poll can be
    /// recorded on the *task's* span rather than the reader's, see [`InlineExecutionSpanSlot`].
    static INLINE_EXECUTION_SPAN: InlineExecutionSpanSlot;
}

/// Lets a reader that executes a claimed task inline record the outcome on the span of the task it
/// executed.
///
/// The reader only sees the outer execution future, whose instrumented span has already been exited
/// by the time its `poll` returns — `Span::current()` there is the reader's own span. So the
/// executor puts the span it is about to instrument the task body with in here, and the reader
/// records the outcome on it afterwards.
///
/// Only present while a claimed task is being polled inline: a task started by a worker doesn't
/// have this task-local set, and leaves the field unset.
#[derive(Clone, Default)]
struct InlineExecutionSpanSlot(Arc<Mutex<Option<Span>>>);

impl InlineExecutionSpanSlot {
    /// Called by the executor with the span it instruments the task body with.
    fn set(span: &Span) {
        let _ = INLINE_EXECUTION_SPAN.try_with(|slot| {
            *slot.0.lock().unwrap() = Some(span.clone());
        });
    }

    /// Records the outcome of the inline poll on the executed task's span, if the executor got far
    /// enough to register one (it doesn't when the task was already claimed by someone else and the
    /// execution turns into a no-op).
    fn record(&self, outcome: &'static str) {
        if let Some(span) = self.0.lock().unwrap().as_ref() {
            span.record("inline_execution", outcome);
        }
    }
}

impl<B: Backend + 'static> TurboTasks<B> {
    // TODO better lifetime management for turbo tasks
    // consider using unsafe for the task_local turbo tasks
    // that should be safe as long tasks can't outlife turbo task
    // so we probably want to make sure that all tasks are joined
    // when trying to drop turbo tasks
    pub fn new(backend: B) -> Arc<Self> {
        let execution_id_factory = IdFactory::new(ExecutionId::MIN, ExecutionId::MAX);
        let this = Arc::new_cyclic(|this| Self {
            this: this.clone(),
            backend,
            execution_id_factory,
            stopped: AtomicBool::new(false),
            currently_scheduled_foreground_jobs: AtomicUsize::new(0),
            currently_scheduled_background_jobs: AtomicUsize::new(0),
            scheduled_tasks: AtomicUsize::new(0),
            inline_counters: InlineExecutionCounters::default(),
            priority_runner: Arc::new(PriorityRunner::new(TurboTasksExecutor)),
            start: Default::default(),
            aggregated_update: Default::default(),
            event_foreground_done: Event::new(|| {
                || "TurboTasks::event_foreground_done".to_string()
            }),
            event_foreground_start: Event::new(|| {
                || "TurboTasks::event_foreground_start".to_string()
            }),
            event_background_done: Event::new(|| {
                || "TurboTasks::event_background_done".to_string()
            }),
            compilation_events: CompilationEventQueue::default(),
        });
        this.backend.startup(&*this);
        this
    }

    pub fn pin(&self) -> Arc<Self> {
        self.this.upgrade().unwrap()
    }

    /// Creates a new root task
    pub fn spawn_root_task<T, F, Fut>(&self, functor: F) -> TaskId
    where
        T: ?Sized,
        F: Fn() -> Fut + Send + Sync + Clone + 'static,
        Fut: Future<Output = Result<Vc<T>>> + Send,
    {
        let id = self.backend.create_transient_task(
            TransientTaskType::Root(Box::new(move || {
                let functor = functor.clone();
                Box::pin(async move {
                    mark_top_level_task();
                    let raw_vc = functor().await?.node;
                    raw_vc.to_non_local().await
                })
            })),
            self,
        );
        self.schedule(id, TaskPriority::initial());
        id
    }

    pub fn dispose_root_task(&self, task_id: TaskId) {
        self.backend.dispose_root_task(task_id, self);
    }

    /// Pins a task against garbage collection (a transient, session-only reference). Balanced by
    /// [`unpin_task_for_gc`](Self::unpin_task_for_gc). Used for references that escape the tracked
    /// task graph — e.g. a `DetachedVc` holding an `OperationVc` across the NAPI boundary.
    pub fn pin_task_for_gc(&self, task_id: TaskId) {
        self.backend.pin_task_for_gc(task_id, self);
    }

    /// Releases a pin added by [`pin_task_for_gc`](Self::pin_task_for_gc).
    pub fn unpin_task_for_gc(&self, task_id: TaskId) {
        self.backend.unpin_task_for_gc(task_id, self);
    }

    // TODO make sure that all dependencies settle before reading them
    /// Creates a new root task, that is only executed once.
    /// Dependencies will not invalidate the task.
    #[track_caller]
    fn spawn_once_task<T, Fut>(&self, future: Fut)
    where
        T: ?Sized,
        Fut: Future<Output = Result<Vc<T>>> + Send + 'static,
    {
        let id = self.backend.create_transient_task(
            TransientTaskType::Once(Box::pin(async move {
                mark_top_level_task();
                let raw_vc = future.await?.node;
                raw_vc.to_non_local().await
            })),
            self,
        );
        self.schedule(id, TaskPriority::initial());
    }

    /// Runs `future` as a top-level task and returns its result.
    ///
    /// # Returned values must not contain `Vc` or `ResolvedVc` or if they do they should be covered by a pin
    ///
    /// `T` is deliberately unconstrained for ergonomics, but a `Vc`/`ResolvedVc` that escapes this
    /// call is **not** protected against garbage collection. It can be collected, and
    /// dereferencing the stale handle then fails.
    ///    ///
    /// To hand a computation out of a top-level task, return the [`OperationVc`] and pin it:
    ///
    /// ```ignore
    /// let (resolved, op) = tt.run_once(async move {
    ///     let op = my_operation(args);
    ///     let resolved = op.resolve().strongly_consistent().await?;
    ///     // Return the operation too, so it can be pinned below.
    ///     Ok((resolved, op))
    /// }).await?;
    /// let _gc_root = GcRoot::pin(tt.clone(), op);  // unpins on drop
    /// ```
    ///
    /// Returning owned data -- [`ReadRef`](crate::ReadRef), `RcStr`, plain values -- is always
    /// fine.
    pub async fn run_once<T: Send + 'static>(
        &self,
        future: impl Future<Output = Result<T>> + Send + 'static,
    ) -> Result<T> {
        let (tx, rx) = tokio::sync::oneshot::channel();
        self.spawn_once_task(async move {
            mark_top_level_task();
            let result = future.await;
            tx.send(result)
                .map_err(|_| anyhow!("unable to send result"))?;
            Ok(Completion::new())
        });

        rx.await?
    }

    /// Runs `future` as a top-level task and returns its result.
    ///
    /// The same constraint as [`run_once`](Self::run_once) applies to `T`: a `Vc`/`ResolvedVc`
    /// returned from here is not anchored against garbage collection. See that method's docs for
    /// why, and for the pin-the-`OperationVc` idiom.
    #[tracing::instrument(level = "trace", skip_all, name = "turbo_tasks::run")]
    pub async fn run<T: Send + 'static>(
        &self,
        future: impl Future<Output = Result<T>> + Send + 'static,
    ) -> Result<T, TurboTasksExecutionError> {
        self.begin_foreground_job();
        // it's okay for execution ids to overflow and wrap, they're just used for an assert
        let execution_id = self.execution_id_factory.wrapping_get();
        let current_task_state = CurrentTaskStateHandle::new(CurrentTaskState::new_temporary(
            execution_id,
            TaskPriority::initial(),
            true, // in_top_level_task
        ));

        let result = TURBO_TASKS
            .scope(
                self.pin(),
                CURRENT_TASK_STATE.scope(current_task_state, async {
                    let result = CaptureFuture::new(future).await;

                    // wait for all spawned local tasks using `local` to finish
                    wait_for_local_tasks().await;

                    match result {
                        Ok(Ok(value)) => Ok(value),
                        Ok(Err(err)) => Err(err.into()),
                        Err(err) => Err(TurboTasksExecutionError::Panic(Arc::new(err))),
                    }
                }),
            )
            .await;
        self.finish_foreground_job();
        result
    }

    pub fn start_once_process(&self, future: impl Future<Output = ()> + Send + 'static) {
        let this = self.pin();
        tokio::spawn(async move {
            this.pin()
                .run_once(async move {
                    this.finish_foreground_job();
                    future.await;
                    this.begin_foreground_job();
                    Ok(())
                })
                .await
                .unwrap()
        });
    }

    pub(crate) fn native_call(
        &self,
        native_fn: &'static NativeFunction,
        this: Option<RawVc>,
        arg: &mut dyn DynTaskInputsStorage,
        persistence: TaskPersistence,
    ) -> RawVc {
        RawVc::task_output(self.backend.get_or_create_task(
            native_fn,
            this,
            arg,
            current_task_if_available("turbo_function calls"),
            persistence,
            self,
        ))
    }

    pub fn dynamic_call(
        &self,
        native_fn: &'static NativeFunction,
        this: Option<RawVc>,
        arg: &mut dyn DynTaskInputsStorage,
        inputs_resolved: InputResolution,
        persistence: TaskPersistence,
    ) -> RawVc {
        if inputs_resolved.is_resolved() && this.is_none_or(|this| this.is_resolved()) {
            return self.native_call(native_fn, this, arg, persistence);
        }
        // Need async resolution — must move the arg to the heap now
        let arg = arg.take_box();
        let task_type = LocalTaskSpec {
            task_type: LocalTaskType::ResolveNative { native_fn },
            this,
            arg,
        };
        self.schedule_local_task(task_type, persistence)
    }

    pub fn trait_call(
        &self,
        trait_method: &'static TraitMethod,
        this: RawVc,
        arg: &mut dyn DynTaskInputsStorage,
        inputs_resolved: InputResolution,
        persistence: TaskPersistence,
    ) -> RawVc {
        // avoid creating a wrapper task if self is already resolved
        // for resolved cells we already know the value type so we can lookup the
        // function
        if let Some((_, cell_id)) = this.as_task_cell() {
            match registry::get_value_type(cell_id.type_id()).get_trait_method(trait_method) {
                Some(native_fn) => {
                    if let Some(filter) = native_fn.arg_meta.filter_owned {
                        let (resolved, mut arg) = (filter)(arg);
                        return self.dynamic_call(
                            native_fn,
                            Some(this),
                            &mut arg,
                            resolved,
                            persistence,
                        );
                    } else {
                        return self.dynamic_call(
                            native_fn,
                            Some(this),
                            arg,
                            inputs_resolved,
                            persistence,
                        );
                    }
                }
                None => {
                    // We are destined to fail at this point, but we just retry resolution in the
                    // local task since we cannot report an error from here.
                    // TODO: A panic seems appropriate since the immediate caller is to blame
                }
            }
        }

        // create a wrapper task to resolve all inputs
        let task_type = LocalTaskSpec {
            task_type: LocalTaskType::ResolveTrait { trait_method },
            this: Some(this),
            arg: arg.take_box(),
        };

        self.schedule_local_task(task_type, persistence)
    }

    #[track_caller]
    pub fn schedule(&self, task_id: TaskId, priority: TaskPriority) {
        self.begin_foreground_job();
        self.scheduled_tasks.fetch_add(1, Ordering::AcqRel);

        let task = ScheduledTask::Task {
            task_id,
            span: Span::current(),
        };
        self.priority_runner.schedule(&self.pin(), task, priority);
    }

    fn schedule_local_task(
        &self,
        ty: LocalTaskSpec,
        // if this is a `LocalTaskType::Resolve*`, we may spawn another task with this persistence,
        persistence: TaskPersistence,
    ) -> RawVc {
        let task_type = ty.task_type;
        let (global_task_state, execution_id, priority, local_task_id) =
            CURRENT_TASK_STATE.with(|gts| {
                let mut gts_write = gts.write().unwrap();
                let local_task_id = gts_write.local_tasks.create(task_type);
                (
                    gts.clone(),
                    gts_write.execution_id,
                    gts_write.priority,
                    local_task_id,
                )
            });

        let task = ScheduledTask::LocalTask {
            ty,
            persistence,
            execution_id,
            local_task_id,
            global_task_state,
            span: Span::current(),
        };
        self.priority_runner.schedule(&self.pin(), task, priority);

        RawVc::local_output(execution_id, local_task_id, persistence)
    }

    /// Executes the task inline if possible, returns true if it executed to completion.
    fn try_execute_scheduled_task_inline(&self, key: ScheduleKey) -> bool {
        let this = self.pin();
        self.inline_counters.claim_attempted();
        if let Some(future) = self.priority_runner.claim(&this, &key) {
            let completed = poll_once_or_spawn(future);
            if completed {
                self.inline_counters.claim_completed();
            } else {
                self.inline_counters.claim_yielded();
            }
            return completed;
        }
        self.inline_counters.claim_failed();
        false
    }

    #[cfg(feature = "inline_execution_stats")]
    fn note_waited_for_in_progress_task(&self) {
        self.inline_counters.waited_in_progress();
    }

    fn begin_foreground_job(&self) {
        if self
            .currently_scheduled_foreground_jobs
            .fetch_add(1, Ordering::AcqRel)
            == 0
        {
            *self.start.lock().unwrap() = Some(Instant::now());
            self.event_foreground_start.notify(usize::MAX);
            self.backend.idle_end(self);
        }
    }

    fn finish_foreground_job(&self) {
        if self
            .currently_scheduled_foreground_jobs
            .fetch_sub(1, Ordering::AcqRel)
            == 1
        {
            self.backend.idle_start(self);
            // That's not super race-condition-safe, but it's only for
            // statistical reasons
            let total = self.scheduled_tasks.load(Ordering::Acquire);
            self.scheduled_tasks.store(0, Ordering::Release);
            if let Some(start) = *self.start.lock().unwrap() {
                let (update, _) = &mut *self.aggregated_update.lock().unwrap();
                if let Some(update) = update.as_mut() {
                    update.0 += start.elapsed();
                    update.1 += total;
                } else {
                    *update = Some((start.elapsed(), total));
                }
            }
            self.event_foreground_done.notify(usize::MAX);
        }
    }

    fn begin_background_job(&self) {
        self.currently_scheduled_background_jobs
            .fetch_add(1, Ordering::Relaxed);
    }

    fn finish_background_job(&self) {
        if self
            .currently_scheduled_background_jobs
            .fetch_sub(1, Ordering::Relaxed)
            == 1
        {
            self.event_background_done.notify(usize::MAX);
        }
    }

    pub fn get_in_progress_count(&self) -> usize {
        self.currently_scheduled_foreground_jobs
            .load(Ordering::Acquire)
    }

    /// Counters describing how reads and inline execution interacted. Diagnostics only; a dump of
    /// these can be requested with `TURBO_ENGINE_INLINE_STATS=1`.
    #[cfg(feature = "inline_execution_stats")]
    #[doc(hidden)]
    pub fn inline_execution_stats(&self) -> InlineExecutionStats {
        let counters = &self.inline_counters;
        InlineExecutionStats {
            queued: self.priority_runner.total_queued(),
            claim_attempted: counters.claim_attempted.load(Ordering::Relaxed),
            claim_completed: counters.claim_completed.load(Ordering::Relaxed),
            claim_yielded: counters.claim_yielded.load(Ordering::Relaxed),
            claim_failed: counters.claim_failed.load(Ordering::Relaxed),
            waited_in_progress: counters.waited_in_progress.load(Ordering::Relaxed),
        }
    }

    /// Waits for the given task to finish executing. This works by performing an untracked read,
    /// and discarding the value of the task output.
    ///
    /// [`ReadConsistency::Eventual`] means that this will return after the task executes, but
    /// before all dependencies have completely settled.
    ///
    /// [`ReadConsistency::Strong`] means that this will also wait for the task and all dependencies
    /// to fully settle before returning.
    ///
    /// As this function is typically called in top-level code that waits for results to be ready
    /// for the user to access, most callers should use [`ReadConsistency::Strong`].
    pub async fn wait_task_completion(
        &self,
        id: TaskId,
        consistency: ReadConsistency,
    ) -> Result<()> {
        read_task_output(
            self,
            id,
            ReadOutputOptions {
                // INVALIDATION: This doesn't return a value, only waits for it to be ready.
                tracking: ReadTracking::Untracked,
                consistency,
            },
        )
        .await?;
        Ok(())
    }

    /// Returns [UpdateInfo] with all updates aggregated over a given duration
    /// (`aggregation`). Will wait until an update happens.
    pub async fn get_or_wait_aggregated_update_info(&self, aggregation: Duration) -> UpdateInfo {
        self.aggregated_update_info(aggregation, Duration::MAX)
            .await
            .unwrap()
    }

    /// Returns [UpdateInfo] with all updates aggregated over a given duration
    /// (`aggregation`). Will only return None when the timeout is reached while
    /// waiting for the first update.
    pub async fn aggregated_update_info(
        &self,
        aggregation: Duration,
        timeout: Duration,
    ) -> Option<UpdateInfo> {
        let listener = self
            .event_foreground_done
            .listen_with_note(|| || "wait for update info".to_string());
        let wait_for_finish = {
            let (update, reason_set) = &mut *self.aggregated_update.lock().unwrap();
            if aggregation.is_zero() {
                if let Some((duration, tasks)) = update.take() {
                    return Some(UpdateInfo {
                        duration,
                        tasks,
                        reasons: take(reason_set),
                        placeholder_for_future_fields: (),
                    });
                } else {
                    true
                }
            } else {
                update.is_none()
            }
        };
        if wait_for_finish {
            if timeout == Duration::MAX {
                // wait for finish
                listener.await;
            } else {
                // wait for start, then wait for finish or timeout
                let start_listener = self
                    .event_foreground_start
                    .listen_with_note(|| || "wait for update info".to_string());
                if self
                    .currently_scheduled_foreground_jobs
                    .load(Ordering::Acquire)
                    == 0
                {
                    start_listener.await;
                } else {
                    drop(start_listener);
                }
                if timeout.is_zero() || tokio::time::timeout(timeout, listener).await.is_err() {
                    // Timeout
                    return None;
                }
            }
        }
        if !aggregation.is_zero() {
            loop {
                select! {
                    () = tokio::time::sleep(aggregation) => {
                        break;
                    }
                    () = self.event_foreground_done.listen_with_note(|| || "wait for update info".to_string()) => {
                        // Resets the sleep
                    }
                }
            }
        }
        let (update, reason_set) = &mut *self.aggregated_update.lock().unwrap();
        if let Some((duration, tasks)) = update.take() {
            Some(UpdateInfo {
                duration,
                tasks,
                reasons: take(reason_set),
                placeholder_for_future_fields: (),
            })
        } else {
            panic!("aggregated_update_info must not called concurrently")
        }
    }

    pub async fn wait_background_done(&self) {
        let listener = self.event_background_done.listen();
        if self
            .currently_scheduled_background_jobs
            .load(Ordering::Acquire)
            != 0
        {
            listener.await;
        }
    }

    pub async fn stop_and_wait(&self) {
        #[cfg(feature = "inline_execution_stats")]
        if inline_stats_requested() {
            // Requested with `TURBO_ENGINE_INLINE_STATS=1`; printed rather than traced so it shows
            // up without a tracing subscriber configured.
            eprintln!(
                "turbo-tasks inline execution stats: {:#?}",
                self.inline_execution_stats()
            );
        }
        turbo_tasks_future_scope(self.pin(), async move {
            self.backend.stopping(self);
            self.stopped.store(true, Ordering::Release);
            {
                let listener = self
                    .event_foreground_done
                    .listen_with_note(|| || "wait for stop".to_string());
                if self
                    .currently_scheduled_foreground_jobs
                    .load(Ordering::Acquire)
                    != 0
                {
                    listener.await;
                }
            }
            {
                let listener = self.event_background_done.listen();
                if self
                    .currently_scheduled_background_jobs
                    .load(Ordering::Acquire)
                    != 0
                {
                    listener.await;
                }
            }
            self.backend.stop(self);
            // Deliver compilation events sent during shutdown (e.g. the persistence trace span)
            // to subscribers before returning, then close the queue so subscriptions end after
            // draining.
            self.compilation_events.flush_and_close().await;
        })
        .await;
    }

    #[track_caller]
    pub(crate) fn schedule_background_job<T>(&self, func: T)
    where
        T: AsyncFnOnce(Arc<TurboTasks<B>>) -> Arc<TurboTasks<B>> + Send + 'static,
        T::CallOnceFuture: Send,
    {
        let mut this = self.pin();
        self.begin_background_job();
        tokio::spawn(
            TURBO_TASKS
                .scope(this.clone(), async move {
                    if !this.stopped.load(Ordering::Acquire) {
                        this = func(this).await;
                    }
                    this.finish_background_job();
                })
                .in_current_span(),
        );
    }

    fn finish_current_task_state(&self) -> FinishedTaskState {
        CURRENT_TASK_STATE.with(|cell| {
            let current_task_state = &*cell.write().unwrap();
            FinishedTaskState {
                #[cfg(feature = "verify_determinism")]
                stateful: current_task_state.stateful,
                has_invalidator: current_task_state.has_invalidator,
            }
        })
    }

    pub fn backend(&self) -> &B {
        &self.backend
    }

    pub fn get_current_task_priority(&self) -> TaskPriority {
        CURRENT_TASK_STATE
            .try_with(|task_state| task_state.read().unwrap().priority)
            .unwrap_or(TaskPriority::initial())
    }

    pub fn is_idle(&self) -> bool {
        self.currently_scheduled_foreground_jobs
            .load(Ordering::Acquire)
            == 0
    }

    #[track_caller]
    pub fn schedule_backend_background_job(&self, job: B::BackendJob) {
        self.schedule_background_job(async move |this| {
            this.backend.run_backend_job(job, &*this).await;
            this
        })
    }
}

struct TurboTasksExecutor;

/// Run a future and abort the process if a panic is reported
///
/// Turbtasks catches panics from user code and propagates throught the task tree, but if it happens
/// as part of state management we have to abort
async fn abort_on_panic<F: Future>(f: F) -> F::Output {
    match AssertUnwindSafe(f).catch_unwind().await {
        Ok(r) => r,
        Err(_) => {
            eprintln!(
                "\nturbo-tasks: an internal panic occurred outside the per-task panic \
                 boundary. This is a bug in turbo-tasks/Turbopack — please report it at \
                 https://github.com/vercel/next.js/discussions and include the panic message \
                 and stack trace above.\n\nAborting."
            );
            abort();
        }
    }
}

impl<B: Backend> Executor<TurboTasks<B>, ScheduledTask, TaskPriority> for TurboTasksExecutor {
    type Future = impl Future<Output = ()> + Send + 'static;

    fn execute(
        &self,
        this: &Arc<TurboTasks<B>>,
        scheduled_task: ScheduledTask,
        priority: TaskPriority,
    ) -> Self::Future {
        match scheduled_task {
            ScheduledTask::Task { task_id, span } => {
                let this2 = this.clone();
                let this = this.clone();
                let future = async move {
                    abort_on_panic(async {
                        // it's okay for execution ids to overflow and wrap, they're just used
                        // for an assert
                        let execution_id = this.execution_id_factory.wrapping_get();
                        let current_task_state =
                            CurrentTaskStateHandle::new(CurrentTaskState::new(
                                task_id,
                                execution_id,
                                priority,
                                false, // in_top_level_task
                            ));
                        let single_execution_future = async {
                            if this.stopped.load(Ordering::Acquire) {
                                this.backend.task_execution_canceled(task_id, &*this);
                                return None;
                            }

                            let TaskExecutionSpec { future, span } = this
                                .backend
                                .try_start_task_execution(task_id, priority, &*this)?;

                            // When a reader claimed this task and is polling it inline, let it
                            // record the outcome on this span rather than its own.
                            InlineExecutionSpanSlot::set(&span);

                            async {
                                let result = CaptureFuture::new(future).await;

                                // wait for all spawned local tasks using `local` to finish
                                wait_for_local_tasks().await;

                                let result = match result {
                                    Ok(Ok(raw_vc)) => {
                                        // This is safe because we waited for all local tasks to
                                        // complete above
                                        raw_vc
                                            .to_non_local_unchecked_sync(&*this)
                                            .map_err(|err| err.into())
                                    }
                                    Ok(Err(err)) => Err(err.into()),
                                    Err(err) => Err(TurboTasksExecutionError::Panic(Arc::new(err))),
                                };

                                let finished_state = this.finish_current_task_state();
                                let cell_counters = CURRENT_TASK_STATE
                                    .with(|ts| ts.write().unwrap().cell_counters.take().unwrap());
                                this.backend.task_execution_completed(
                                    task_id,
                                    result,
                                    &cell_counters,
                                    #[cfg(feature = "verify_determinism")]
                                    finished_state.stateful,
                                    finished_state.has_invalidator,
                                    &*this,
                                )
                            }
                            .instrument(span)
                            .await
                        };
                        if let Some(stale_priority) = CURRENT_TASK_STATE
                            .scope(current_task_state, single_execution_future)
                            .await
                        {
                            // Task was stale; re-schedule at the correct invalidation priority so
                            // other tasks can run in the right priority order.
                            this.schedule(task_id, stale_priority);
                        }
                        this.finish_foreground_job();
                    })
                    .await
                };

                Either::Left(TURBO_TASKS.scope(this2, future).instrument(span))
            }
            ScheduledTask::LocalTask {
                ty,
                persistence,
                execution_id: _,
                local_task_id,
                global_task_state,
                span,
            } => {
                let this2 = this.clone();
                let this = this.clone();
                let task_type = ty.task_type;
                let future = async move {
                    let span = match &ty.task_type {
                        LocalTaskType::ResolveNative { native_fn } => {
                            native_fn.resolve_span(priority)
                        }
                        LocalTaskType::ResolveTrait { trait_method } => {
                            trait_method.resolve_span(priority)
                        }
                    };
                    // See the cached-task arm: lets a reader that claimed this local task record
                    // the outcome of its inline poll on this span.
                    InlineExecutionSpanSlot::set(&span);
                    abort_on_panic(
                        async move {
                            let result = match ty.task_type {
                                LocalTaskType::ResolveNative { native_fn } => {
                                    LocalTaskType::run_resolve_native(
                                        native_fn,
                                        ty.this,
                                        &*ty.arg,
                                        persistence,
                                        this,
                                    )
                                    .await
                                }
                                LocalTaskType::ResolveTrait { trait_method } => {
                                    LocalTaskType::run_resolve_trait(
                                        trait_method,
                                        ty.this.unwrap(),
                                        &*ty.arg,
                                        persistence,
                                        this,
                                    )
                                    .await
                                }
                            };

                            let output = match result {
                                Ok(raw_vc) => OutputContent::Link(raw_vc),
                                Err(err) => OutputContent::Error(
                                    TurboTasksExecutionError::from(err)
                                        .with_local_task_context(task_type.to_string()),
                                ),
                            };

                            CURRENT_TASK_STATE.with(move |gts| {
                                gts.write()
                                    .unwrap()
                                    .local_tasks
                                    .complete(local_task_id, output);
                            });
                        }
                        .instrument(span),
                    )
                    .await
                };
                let future = CURRENT_TASK_STATE.scope(global_task_state, future);

                Either::Right(TURBO_TASKS.scope(this2, future).instrument(span))
            }
        }
    }
}

struct FinishedTaskState {
    /// True if the task has state in cells (interior mutability).
    /// Only tracked when verify_determinism feature is enabled.
    #[cfg(feature = "verify_determinism")]
    stateful: bool,

    /// True if the task uses an external invalidator
    has_invalidator: bool,
}

impl<B: Backend + 'static> TurboTasksCallApi for TurboTasks<B> {
    fn dynamic_call(
        &self,
        native_fn: &'static NativeFunction,
        this: Option<RawVc>,
        arg: &mut dyn DynTaskInputsStorage,
        inputs_resolved: InputResolution,
        persistence: TaskPersistence,
    ) -> RawVc {
        self.dynamic_call(native_fn, this, arg, inputs_resolved, persistence)
    }
    fn native_call(
        &self,
        native_fn: &'static NativeFunction,
        this: Option<RawVc>,
        arg: &mut dyn DynTaskInputsStorage,
        persistence: TaskPersistence,
    ) -> RawVc {
        self.native_call(native_fn, this, arg, persistence)
    }
    fn trait_call(
        &self,
        trait_method: &'static TraitMethod,
        this: RawVc,
        arg: &mut dyn DynTaskInputsStorage,
        inputs_resolved: InputResolution,
        persistence: TaskPersistence,
    ) -> RawVc {
        self.trait_call(trait_method, this, arg, inputs_resolved, persistence)
    }

    #[track_caller]
    fn run(
        &self,
        future: Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>,
    ) -> Pin<Box<dyn Future<Output = Result<(), TurboTasksExecutionError>> + Send>> {
        let this = self.pin();
        Box::pin(async move { this.run(future).await })
    }

    #[track_caller]
    fn run_once(
        &self,
        future: Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
        let this = self.pin();
        Box::pin(async move { this.run_once(future).await })
    }

    #[track_caller]
    fn run_once_with_reason(
        &self,
        reason: StaticOrArc<dyn InvalidationReason>,
        future: Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
        {
            let (_, reason_set) = &mut *self.aggregated_update.lock().unwrap();
            reason_set.insert(reason);
        }
        let this = self.pin();
        Box::pin(async move { this.run_once(future).await })
    }

    #[track_caller]
    fn start_once_process(&self, future: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
        self.start_once_process(future)
    }

    fn send_compilation_event(&self, event: Arc<dyn CompilationEvent>) {
        if let Err(e) = self.compilation_events.send(event) {
            tracing::warn!("Failed to send compilation event: {e}");
        }
    }

    fn get_task_name(&self, task: TaskId) -> String {
        self.backend.get_task_name(task, self)
    }
}

impl<B: Backend + 'static> TurboTasksApi for TurboTasks<B> {
    #[instrument(level = "info", skip_all, name = "invalidate")]
    fn invalidate(&self, task: TaskId) {
        self.backend.invalidate_task(task, self);
    }

    #[instrument(level = "info", skip_all, name = "invalidate", fields(name = display(&reason)))]
    fn invalidate_with_reason(&self, task: TaskId, reason: StaticOrArc<dyn InvalidationReason>) {
        {
            let (_, reason_set) = &mut *self.aggregated_update.lock().unwrap();
            reason_set.insert(reason);
        }
        self.backend.invalidate_task(task, self);
    }

    fn invalidate_serialization(&self, task: TaskId) {
        self.backend.invalidate_serialization(task, self);
    }

    #[track_caller]
    fn try_read_task_output(
        &self,
        task: TaskId,
        options: ReadOutputOptions,
    ) -> Result<ReadOutcome<RawVc>> {
        if options.consistency == ReadConsistency::Eventual {
            debug_assert_not_in_top_level_task("read_task_output");
        }
        self.backend.try_read_task_output(
            task,
            current_task_if_available("reading Vcs"),
            options,
            self,
        )
    }

    #[track_caller]
    fn try_read_task_cell(
        &self,
        task: TaskId,
        index: CellId,
        options: ReadCellOptions,
    ) -> Result<ReadOutcome<TypedCellContent>> {
        let reader = current_task_if_available("reading Vcs");
        self.backend
            .try_read_task_cell(task, index, reader, options, self)
    }

    fn try_read_own_task_cell(
        &self,
        current_task: TaskId,
        index: CellId,
    ) -> Result<TypedCellContent> {
        self.backend
            .try_read_own_task_cell(current_task, index, self)
    }

    #[track_caller]
    fn try_read_local_output(
        &self,
        execution_id: ExecutionId,
        local_task_id: LocalTaskId,
    ) -> Result<Result<RawVc, EventListener>> {
        debug_assert_not_in_top_level_task("read_local_output");
        CURRENT_TASK_STATE.with(|gts| {
            let gts_read = gts.read().unwrap();

            // Local Vcs are local to their parent task's current execution, and do not exist
            // outside of it. This is weakly enforced at compile time using the `NonLocalValue`
            // marker trait. This assertion exists to handle any potential escapes that the
            // compile-time checks cannot capture.
            gts_read.assert_execution_id(execution_id);

            match gts_read.local_tasks.get(local_task_id) {
                LocalTask::Scheduled { done_event } => Ok(Err(done_event.listen())),
                LocalTask::Done { output } => Ok(Ok(output.as_read_result()?)),
            }
        })
    }

    fn read_task_collectibles(&self, task: TaskId, trait_id: TraitTypeId) -> TaskCollectiblesMap {
        // TODO: Add assert_not_in_top_level_task("read_task_collectibles") check here.
        // Collectible reads are eventually consistent.
        self.backend.read_task_collectibles(
            task,
            trait_id,
            current_task_if_available("reading collectibles"),
            self,
        )
    }

    fn try_execute_scheduled_task_inline(&self, key: ScheduleKey) -> bool {
        self.try_execute_scheduled_task_inline(key)
    }

    #[cfg(feature = "inline_execution_stats")]
    fn note_waited_for_in_progress_task(&self) {
        self.note_waited_for_in_progress_task()
    }

    fn emit_collectible(&self, trait_type: TraitTypeId, collectible: RawVc) {
        self.backend.emit_collectible(
            trait_type,
            collectible,
            current_task("emitting collectible"),
            self,
        );
    }

    fn unemit_collectible(&self, trait_type: TraitTypeId, collectible: RawVc, count: u32) {
        self.backend.unemit_collectible(
            trait_type,
            collectible,
            count,
            current_task("emitting collectible"),
            self,
        );
    }

    fn unemit_collectibles(&self, trait_type: TraitTypeId, collectibles: &TaskCollectiblesMap) {
        for (&collectible, &count) in collectibles {
            if count > 0 {
                self.backend.unemit_collectible(
                    trait_type,
                    collectible,
                    count as u32,
                    current_task("emitting collectible"),
                    self,
                );
            }
        }
    }

    fn read_own_task_cell(&self, task: TaskId, index: CellId) -> Result<TypedCellContent> {
        self.try_read_own_task_cell(task, index)
    }

    fn update_own_task_cell(
        &self,
        task: TaskId,
        index: CellId,
        content: CellContent,
        updated_key_hashes: Option<SmallVec<[u64; 2]>>,
        content_hash: Option<CellHash>,
        verification_mode: VerificationMode,
    ) {
        self.backend.update_task_cell(
            task,
            index,
            content,
            updated_key_hashes,
            content_hash,
            verification_mode,
            self,
        );
    }

    fn connect_task(&self, task: TaskId) {
        self.backend
            .connect_task(task, current_task_if_available("connecting task"), self);
    }

    fn mark_own_task_as_finished(&self, task: TaskId) {
        self.backend.mark_own_task_as_finished(task, self);
    }

    fn pin_task_for_gc(&self, task: TaskId) {
        self.backend.pin_task_for_gc(task, self);
    }

    fn unpin_task_for_gc(&self, task: TaskId) {
        self.backend.unpin_task_for_gc(task, self);
    }

    /// Creates a future that inherits the current task id and task state. The current global task
    /// will wait for this future to be dropped before exiting.
    fn spawn_detached_for_testing(&self, fut: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
        // this is similar to what happens for a local task, except that we keep the local task's
        // state as well.
        let global_task_state = CURRENT_TASK_STATE.with(|ts| ts.clone());
        global_task_state
            .write()
            .unwrap()
            .local_tasks
            .register_detached();
        let wrapped = async move {
            // use a drop guard for panic safety
            struct DropGuard;
            impl Drop for DropGuard {
                fn drop(&mut self) {
                    CURRENT_TASK_STATE
                        .with(|ts| ts.write().unwrap().local_tasks.decrement_in_flight());
                }
            }
            let _guard = DropGuard;
            fut.await;
        };
        tokio::spawn(TURBO_TASKS.scope(
            turbo_tasks(),
            CURRENT_TASK_STATE.scope(global_task_state, wrapped),
        ));
    }

    fn task_statistics(&self) -> &TaskStatisticsApi {
        self.backend.task_statistics()
    }

    fn stop_and_wait(&self) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
        let this = self.pin();
        Box::pin(async move {
            this.stop_and_wait().await;
        })
    }

    fn subscribe_to_compilation_events(
        &self,
        event_types: Option<Vec<String>>,
    ) -> Receiver<Arc<dyn CompilationEvent>> {
        self.compilation_events.subscribe(event_types)
    }

    fn is_tracking_dependencies(&self) -> bool {
        self.backend.is_tracking_dependencies()
    }
}

async fn wait_for_local_tasks() {
    let listener =
        CURRENT_TASK_STATE.with(|ts| ts.read().unwrap().local_tasks.listen_for_in_flight());
    let Some(listener) = listener else {
        return;
    };
    listener.await;
}

pub(crate) fn current_task_if_available(from: &str) -> Option<TaskId> {
    match CURRENT_TASK_STATE.try_with(|ts| ts.current_task_id()) {
        Ok(id) => id,
        Err(_) => panic!(
            "{from} can only be used in the context of a turbo_tasks task execution or \
             turbo_tasks run"
        ),
    }
}

pub(crate) fn current_task(from: &str) -> TaskId {
    match CURRENT_TASK_STATE.try_with(|ts| ts.current_task_id()) {
        Ok(Some(id)) => id,
        Ok(None) | Err(_) => {
            panic!("{from} can only be used in the context of a turbo_tasks task execution")
        }
    }
}

/// Panics if we're not in a top-level task (e.g. [`run_once`]). Some function calls should only
/// happen in a top-level task (e.g. [`Effects::apply`][crate::Effects::apply]).
#[track_caller]
pub(crate) fn debug_assert_in_top_level_task(message: &str) {
    if !cfg!(debug_assertions) {
        return;
    }

    let in_top_level = CURRENT_TASK_STATE
        .try_with(|ts| ts.read().unwrap().in_top_level_task)
        .unwrap_or(true);
    if !in_top_level {
        panic!("{message}");
    }
}

#[track_caller]
pub(crate) fn debug_assert_not_in_top_level_task(operation: &str) {
    if !cfg!(debug_assertions) {
        return;
    }

    // HACK: We set this inside of `ReadRawVcFuture` to suppress warnings about an internal
    // consistency bug
    let suppressed = SUPPRESS_EVENTUAL_CONSISTENCY_TOP_LEVEL_TASK_CHECK
        .try_with(|&suppressed| suppressed)
        .unwrap_or(false);
    if suppressed {
        return;
    }

    let in_top_level = CURRENT_TASK_STATE
        .try_with(|ts| ts.read().unwrap().in_top_level_task)
        .unwrap_or(false);
    if in_top_level {
        panic!(
            "Eventually consistent read ({operation}) cannot be performed from a top-level task. \
             Top-level tasks (e.g. code inside `.run_once(...)`) must use strongly consistent \
             reads to avoid leaking inconsistent return values."
        );
    }
}

pub async fn run<T: Send + 'static>(
    tt: Arc<dyn TurboTasksApi>,
    future: impl Future<Output = Result<T>> + Send + 'static,
) -> Result<T> {
    let (tx, rx) = tokio::sync::oneshot::channel();

    tt.run(Box::pin(async move {
        let result = future.await?;
        tx.send(result)
            .map_err(|_| anyhow!("unable to send result"))?;
        Ok(())
    }))
    .await?;

    Ok(rx.await?)
}

pub async fn run_once<T: Send + 'static>(
    tt: Arc<dyn TurboTasksApi>,
    future: impl Future<Output = Result<T>> + Send + 'static,
) -> Result<T> {
    let (tx, rx) = tokio::sync::oneshot::channel();

    tt.run_once(Box::pin(async move {
        let result = future.await?;
        tx.send(result)
            .map_err(|_| anyhow!("unable to send result"))?;
        Ok(())
    }))
    .await?;

    Ok(rx.await?)
}

pub async fn run_once_with_reason<T: Send + 'static>(
    tt: Arc<dyn TurboTasksApi>,
    reason: impl InvalidationReason,
    future: impl Future<Output = Result<T>> + Send + 'static,
) -> Result<T> {
    let (tx, rx) = tokio::sync::oneshot::channel();

    tt.run_once_with_reason(
        (Arc::new(reason) as Arc<dyn InvalidationReason>).into(),
        Box::pin(async move {
            let result = future.await?;
            tx.send(result)
                .map_err(|_| anyhow!("unable to send result"))?;
            Ok(())
        }),
    )
    .await?;

    Ok(rx.await?)
}

/// Calls [`TurboTasks::dynamic_call`] for the current turbo tasks instance.
pub fn dynamic_call(
    func: &'static NativeFunction,
    this: Option<RawVc>,
    arg: &mut dyn DynTaskInputsStorage,
    inputs_resolved: InputResolution,
    persistence: TaskPersistence,
) -> RawVc {
    with_turbo_tasks(|tt| tt.dynamic_call(func, this, arg, inputs_resolved, persistence))
}

/// Calls [`TurboTasks::trait_call`] for the current turbo tasks instance.
pub fn trait_call(
    trait_method: &'static TraitMethod,
    this: RawVc,
    arg: &mut dyn DynTaskInputsStorage,
    inputs_resolved: InputResolution,
    persistence: TaskPersistence,
) -> RawVc {
    with_turbo_tasks(|tt| tt.trait_call(trait_method, this, arg, inputs_resolved, persistence))
}

pub fn turbo_tasks() -> Arc<dyn TurboTasksApi> {
    TURBO_TASKS.with(|arc| arc.clone())
}

pub fn turbo_tasks_weak() -> Weak<dyn TurboTasksApi> {
    TURBO_TASKS.with(Arc::downgrade)
}

pub fn try_turbo_tasks() -> Option<Arc<dyn TurboTasksApi>> {
    TURBO_TASKS.try_with(|arc| arc.clone()).ok()
}

pub fn with_turbo_tasks<T>(func: impl FnOnce(&Arc<dyn TurboTasksApi>) -> T) -> T {
    TURBO_TASKS.with(|arc| func(arc))
}

pub fn turbo_tasks_scope<T>(tt: Arc<dyn TurboTasksApi>, f: impl FnOnce() -> T) -> T {
    TURBO_TASKS.sync_scope(tt, f)
}

pub fn turbo_tasks_future_scope<T>(
    tt: Arc<dyn TurboTasksApi>,
    f: impl Future<Output = T>,
) -> impl Future<Output = T> {
    TURBO_TASKS.scope(tt, f)
}

/// Spawns the given future within the context of the current task.
///
/// Beware: this method is not safe to use in production code. It is only
/// intended for use in tests and for debugging purposes.
pub fn spawn_detached_for_testing(f: impl Future<Output = ()> + Send + 'static) {
    turbo_tasks().spawn_detached_for_testing(Box::pin(f));
}

/// Marks the current task as finished. This excludes it from waiting for
/// strongly consistency.
pub fn mark_finished() {
    with_turbo_tasks(|tt| {
        tt.mark_own_task_as_finished(current_task("turbo_tasks::mark_finished()"))
    });
}

/// Returns a [`SerializationInvalidator`] that can be used to invalidate the
/// serialization of the current task cells.
///
/// Also marks the current task as stateful when the `verify_determinism` feature is enabled,
/// since State allocation implies interior mutability.
pub fn get_serialization_invalidator() -> SerializationInvalidator {
    CURRENT_TASK_STATE.with(|cell| {
        let CurrentTaskState {
            task_id,
            #[cfg(feature = "verify_determinism")]
            stateful,
            ..
        } = &mut *cell.write().unwrap();
        #[cfg(feature = "verify_determinism")]
        {
            *stateful = true;
        }
        let Some(task_id) = *task_id else {
            panic!(
                "get_serialization_invalidator() can only be used in the context of a turbo_tasks \
                 task execution"
            );
        };
        SerializationInvalidator::new(task_id)
    })
}

pub fn mark_invalidator() {
    CURRENT_TASK_STATE.with(|cell| {
        let CurrentTaskState {
            has_invalidator, ..
        } = &mut *cell.write().unwrap();
        *has_invalidator = true;
    })
}

/// Marks the current task as stateful. This is used to indicate that the task
/// has interior mutability (e.g., via [`State`][crate::State]), which means
/// the task may produce different outputs even with the same inputs.
///
/// Only has an effect when the `verify_determinism` feature is enabled.
pub fn mark_stateful() {
    #[cfg(feature = "verify_determinism")]
    {
        CURRENT_TASK_STATE.with(|cell| {
            let CurrentTaskState { stateful, .. } = &mut *cell.write().unwrap();
            *stateful = true;
        })
    }
    // No-op when verify_determinism is not enabled
}

/// Marks the current task context as being in a top-level task. When in a top-level task,
/// eventually consistent reads will panic. It is almost always a mistake to perform an eventually
/// consistent read at the top-level of the application.
pub fn mark_top_level_task() {
    if cfg!(debug_assertions) {
        CURRENT_TASK_STATE.with(|cell| {
            cell.write().unwrap().in_top_level_task = true;
        })
    }
}

/// Unmarks the current task context as being in a top-level task. The opposite of
/// [`mark_top_level_task`].
///
/// This utility can be okay in unit tests, where we're observing the internal behavior of
/// turbo-tasks, but otherwise, it is probably a mistake to call this function.
///
/// Calling this will allow eventually-consistent reads at the top-level, potentially exposing
/// incomplete computations and internal errors caused by eventual consistency that would've been
/// caught when the function was re-run. A strongly-consistent read re-runs parts of a task until
/// all of the dependencies have settled.
pub fn unmark_top_level_task_may_leak_eventually_consistent_state() {
    if cfg!(debug_assertions) {
        CURRENT_TASK_STATE.with(|cell| {
            cell.write().unwrap().in_top_level_task = false;
        })
    }
}

/// Pins the current task against garbage collection for the rest of the session, keeping it (and,
/// via the reachability it anchors, the values it produced) alive even if it becomes disconnected
/// from the live task graph. Use this when a value escapes the tracked graph — e.g. a `Vc` sent out
/// of a `spawn_detached` future across a channel, or handed across the NAPI boundary — so no
/// persistent parent lists it as a child and it would otherwise be collected.
///
/// No-op outside a task context, and on backends without garbage collection.
pub fn prevent_gc() {
    if let Some(task) = current_task_if_available("prevent_gc") {
        with_turbo_tasks(|tt| tt.pin_task_for_gc(task));
    }
}

/// An RAII guard that pins an [`OperationVc`]'s task against garbage collection.
pub struct GcRoot<T: ?Sized> {
    tt: Arc<dyn TurboTasksApi>,
    vc: OperationVc<T>,
}

impl<T: ?Sized> GcRoot<T> {
    /// Pins `vc`'s task, returning a guard that unpins it on drop.
    pub fn pin(tt: Arc<dyn TurboTasksApi>, vc: OperationVc<T>) -> Self {
        tt.pin_task_for_gc(vc.task_id());
        Self { tt, vc }
    }
}

/// A guard derefs to the operation it pins, so [`OperationVc`]'s own methods can be called on it
/// directly and `*guard` recovers the operation itself.
impl<T: ?Sized> Deref for GcRoot<T> {
    type Target = OperationVc<T>;

    fn deref(&self) -> &Self::Target {
        &self.vc
    }
}

impl<T: ?Sized> Clone for GcRoot<T> {
    fn clone(&self) -> Self {
        Self::pin(self.tt.clone(), self.vc)
    }
}

impl<T: ?Sized> Drop for GcRoot<T> {
    fn drop(&mut self) {
        self.tt.unpin_task_for_gc(self.vc.task_id());
    }
}

impl<T: ?Sized> Debug for GcRoot<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("GcRoot").field("vc", &self.vc).finish()
    }
}

impl<T: ?Sized> PartialEq for GcRoot<T> {
    /// Compares the pinned operation only. Two guards for the same operation are interchangeable
    /// as far as reachability is concerned, even though each holds its own pin.
    fn eq(&self, other: &Self) -> bool {
        self.vc == other.vc
    }
}

impl<T: ?Sized> Eq for GcRoot<T> {}

impl<T: ?Sized> Hash for GcRoot<T> {
    /// Hashes the pinned operation, consistently with [`PartialEq`], so a guard can be looked up
    /// in a set by the [`OperationVc`] it pins (see the [`Borrow`] impl).
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.vc.hash(state);
    }
}

/// Lets a collection keyed on guards be queried with the bare operation: `Borrow` plus the
/// matching [`Hash`]/[`Eq`] impls give `OperationVc<T>: Equivalent<GcRoot<T>>`, so e.g.
/// `IndexSet<GcRoot<T>>::swap_remove` accepts an `&OperationVc<T>`.
impl<T: ?Sized> Borrow<OperationVc<T>> for GcRoot<T> {
    fn borrow(&self) -> &OperationVc<T> {
        &self.vc
    }
}

/// Safety: a `GcRoot` contains exactly one [`OperationVc`] and no [`Vc`] or [`ResolvedVc`], which
/// is what [`OperationValue`] asserts.
unsafe impl<T: ?Sized + Send> OperationValue for GcRoot<T> {}

/// Safety: mirrors the [`OperationVc`] impl — a `GcRoot` holds no task-local data beyond the
/// operation it pins.
unsafe impl<T: NonLocalValue + ?Sized> NonLocalValue for GcRoot<T> {}

pub fn emit<T: VcValueTrait + ?Sized>(collectible: ResolvedVc<T>) {
    with_turbo_tasks(|tt| {
        let raw_vc = collectible.node.node;
        tt.emit_collectible(T::get_trait_type_id(), raw_vc)
    })
}

pub(crate) async fn read_task_output(
    this: &dyn TurboTasksApi,
    id: TaskId,
    options: ReadOutputOptions,
) -> Result<RawVc> {
    loop {
        match this.try_read_task_output(id, options)? {
            ReadOutcome::Value(result) => return Ok(result),
            ReadOutcome::Scheduled(listener) => {
                // Nobody has started it yet, so take it over instead of waiting for a worker.
                if execute_read_target_inline(this, ScheduleKey::Task(id)) {
                    continue;
                }
                listener.await
            }
            ReadOutcome::InProgress(listener) => {
                // A worker is on it — there is nothing to take over, so don't touch the queue.
                #[cfg(feature = "inline_execution_stats")]
                this.note_waited_for_in_progress_task();
                listener.await
            }
        }
    }
}

/// A reference to a task's cell with methods that allow updating the contents
/// of the cell.
///
/// Mutations should not outside of the task that that owns this cell. Doing so
/// is a logic error, and may lead to incorrect caching behavior.
#[derive(Clone, Copy)]
pub struct CurrentCellRef {
    current_task: TaskId,
    index: CellId,
}

type VcReadTarget<T> = <<T as VcValueType>::Read as VcRead<T>>::Target;

/// What a conditional cell update returns: the new content, the key hashes that changed, and an
/// optional hash of the value.
type CellUpdate = (
    SharedReference,
    Option<SmallVec<[u64; 2]>>,
    Option<CellHash>,
);

/// The callback [`CurrentCellRef::conditional_update_with_shared_reference`] takes. It is a `dyn`
/// trait object so that the function's body is compiled once rather than once per cell type.
type CellUpdateFn<'l> = dyn FnMut(Option<&SharedReference>) -> Option<CellUpdate> + 'l;

impl CurrentCellRef {
    /// Updates the cell if the given `functor` returns a value.
    fn conditional_update<T>(
        &self,
        functor: impl FnOnce(Option<&T>) -> Option<(T, Option<SmallVec<[u64; 2]>>, Option<CellHash>)>,
    ) where
        T: VcValueType,
    {
        // `FnMut` cannot move out of its captures, and the callee calls this at most once, so
        // the `FnOnce` is handed over through an `Option`.
        let mut functor = Some(functor);
        self.conditional_update_with_shared_reference(&mut |old_shared_reference| {
            let functor = functor.take().expect("functor is called at most once");
            let old_ref = old_shared_reference.and_then(|sr| sr.0.downcast_ref::<T>());
            let (new_value, updated_key_hashes, content_hash) = functor(old_ref)?;
            Some((
                SharedReference::new(triomphe::Arc::new(new_value)),
                updated_key_hashes,
                content_hash,
            ))
        })
    }

    /// Updates the cell if the given `functor` returns a `SharedReference`.
    ///
    /// `functor` is a `dyn` trait object rather than a generic parameter on purpose. This body is
    /// identical for every cell type, so making it generic monomorphized it once per
    /// `VcValueType` in the dependency graph — over a thousand copies of the same code. The
    /// indirect call it costs instead is negligible next to the cell read and update it wraps.
    fn conditional_update_with_shared_reference(&self, functor: &mut CellUpdateFn<'_>) {
        let tt = turbo_tasks();
        let cell_content = tt.read_own_task_cell(self.current_task, self.index).ok();
        let update = functor(cell_content.as_ref().and_then(|cc| cc.1.0.as_ref()));
        if let Some((update, updated_key_hashes, content_hash)) = update {
            tt.update_own_task_cell(
                self.current_task,
                self.index,
                CellContent(Some(update)),
                updated_key_hashes,
                content_hash,
                VerificationMode::EqualityCheck,
            )
        }
    }

    /// Replace the current cell's content with `new_value` if the current content is not equal by
    /// value with the existing content.
    ///
    /// The comparison happens using the value itself, not the [`VcRead::Target`] of that value.
    ///
    /// Take this example of a custom equality implementation on a transparent wrapper type:
    ///
    /// ```
    /// #[turbo_tasks::value(transparent, eq = "manual")]
    /// #[derive(Clone)]
    /// struct Wrapper(Vec<u32>);
    ///
    /// impl PartialEq for Wrapper {
    ///     fn eq(&self, other: &Wrapper) -> bool {
    ///         // Example: order doesn't matter for equality
    ///         let (mut this, mut other) = (self.0.clone(), other.0.clone());
    ///         this.sort_unstable();
    ///         other.sort_unstable();
    ///         this == other
    ///     }
    /// }
    ///
    /// impl Eq for Wrapper {}
    /// ```
    ///
    /// Comparisons of [`Vc<Wrapper>`] used when updating the cell will use `Wrapper`'s custom
    /// equality implementation, rather than the one provided by the target ([`Vec<u32>`]) type.
    ///
    /// However, in most cases, the default derived implementation of [`PartialEq`] is used which
    /// just forwards to the inner value's [`PartialEq`].
    ///
    /// If you already have a `SharedReference`, consider calling
    /// [`Self::compare_and_update_with_shared_reference`] which can re-use the [`SharedReference`]
    /// object.
    pub fn compare_and_update<T>(&self, new_value: T)
    where
        T: PartialEq + VcValueType,
    {
        self.conditional_update(|old_value| {
            if let Some(old_value) = old_value
                && old_value == &new_value
            {
                return None;
            }
            Some((new_value, None, None))
        });
    }

    /// Replace the current cell's content with `new_shared_reference` if the current content is not
    /// equal by value with the existing content.
    ///
    /// If you already have a `SharedReference`, this is a faster version of
    /// [`CurrentCellRef::compare_and_update`].
    ///
    /// The value should be stored in [`SharedReference`] using the type `T`.
    pub fn compare_and_update_with_shared_reference<T>(&self, new_shared_reference: SharedReference)
    where
        T: VcValueType + PartialEq,
    {
        let mut new_shared_reference = Some(new_shared_reference);
        self.conditional_update_with_shared_reference(&mut |old_sr| {
            let new_shared_reference = new_shared_reference
                .take()
                .expect("functor is called at most once");
            if let Some(old_sr) = old_sr {
                let old_value = extract_sr_value::<T>(old_sr);
                let new_value = extract_sr_value::<T>(&new_shared_reference);
                if old_value == new_value {
                    return None;
                }
            }
            Some((new_shared_reference, None, None))
        });
    }

    /// Replace the current cell's content if the new value is different.
    ///
    /// Like [`Self::compare_and_update`], but also computes and stores a hash of the value.
    /// When the cell's transient data is evicted, the stored hash enables the backend to detect
    /// whether the value actually changed without re-comparing values—avoiding unnecessary
    /// downstream invalidation.
    ///
    /// Requires `T: DeterministicHash` in addition to `T: PartialEq`.
    pub fn hashed_compare_and_update<T>(&self, new_value: T)
    where
        T: PartialEq + DeterministicHash + VcValueType,
    {
        self.conditional_update(|old_value| {
            if let Some(old_value) = old_value
                && old_value == &new_value
            {
                return None;
            }
            let content_hash = hash_xxh3_hash128(&new_value).to_le_bytes();

            Some((new_value, None, Some(content_hash)))
        });
    }

    /// Replace the current cell's content if the new value (from a pre-existing
    /// [`SharedReference`]) is different.
    ///
    /// Like [`Self::compare_and_update_with_shared_reference`], but also passes a hash
    /// for hash-based change detection when transient data has been evicted.
    pub fn hashed_compare_and_update_with_shared_reference<T>(
        &self,
        new_shared_reference: SharedReference,
    ) where
        T: VcValueType + PartialEq + DeterministicHash,
    {
        let mut new_shared_reference = Some(new_shared_reference);
        self.conditional_update_with_shared_reference(&mut move |old_sr| {
            let new_shared_reference = new_shared_reference
                .take()
                .expect("functor is called at most once");
            if let Some(old_sr) = old_sr {
                let old_value = extract_sr_value::<T>(old_sr);
                let new_value = extract_sr_value::<T>(&new_shared_reference);
                if old_value == new_value {
                    return None;
                }
            }
            let content_hash =
                hash_xxh3_hash128(extract_sr_value::<T>(&new_shared_reference)).to_le_bytes();
            Some((new_shared_reference, None, Some(content_hash)))
        });
    }

    /// See [`Self::compare_and_update`], but selectively update individual keys.
    pub fn keyed_compare_and_update<T>(&self, new_value: T)
    where
        T: PartialEq + VcValueType,
        VcReadTarget<T>: KeyedEq,
        <VcReadTarget<T> as KeyedEq>::Key: std::hash::Hash,
    {
        self.conditional_update(|old_value| {
            let Some(old_value) = old_value else {
                return Some((new_value, None, None));
            };
            let old_value = <T as VcValueType>::Read::value_to_target_ref(old_value);
            let new_value_ref = <T as VcValueType>::Read::value_to_target_ref(&new_value);
            let updated_keys = old_value.different_keys(new_value_ref);
            if updated_keys.is_empty() {
                return None;
            }
            // Duplicates are very unlikely, but ok since the backend is deduplicating them
            let updated_key_hashes = updated_keys
                .into_iter()
                .map(|key| FxBuildHasher.hash_one(key))
                .collect();
            Some((new_value, Some(updated_key_hashes), None))
        });
    }

    /// See [`Self::compare_and_update_with_shared_reference`], but selectively update individual
    /// keys.
    pub fn keyed_compare_and_update_with_shared_reference<T>(
        &self,
        new_shared_reference: SharedReference,
    ) where
        T: VcValueType + PartialEq,
        VcReadTarget<T>: KeyedEq,
        <VcReadTarget<T> as KeyedEq>::Key: std::hash::Hash,
    {
        let mut new_shared_reference = Some(new_shared_reference);
        self.conditional_update_with_shared_reference(&mut |old_sr| {
            let new_shared_reference = new_shared_reference
                .take()
                .expect("functor is called at most once");
            let Some(old_sr) = old_sr else {
                return Some((new_shared_reference, None, None));
            };
            let old_value = extract_sr_value::<T>(old_sr);
            let old_value = <T as VcValueType>::Read::value_to_target_ref(old_value);
            let new_value = extract_sr_value::<T>(&new_shared_reference);
            let new_value = <T as VcValueType>::Read::value_to_target_ref(new_value);
            let updated_keys = old_value.different_keys(new_value);
            if updated_keys.is_empty() {
                return None;
            }
            // Duplicates are very unlikely, but ok since the backend is deduplicating them
            let updated_key_hashes = updated_keys
                .into_iter()
                .map(|key| FxBuildHasher.hash_one(key))
                .collect();
            Some((new_shared_reference, Some(updated_key_hashes), None))
        });
    }

    /// Unconditionally updates the content of the cell.
    pub fn update<T>(&self, new_value: T, verification_mode: VerificationMode)
    where
        T: VcValueType,
    {
        let tt = turbo_tasks();
        tt.update_own_task_cell(
            self.current_task,
            self.index,
            CellContent(Some(SharedReference::new(triomphe::Arc::new(new_value)))),
            None,
            None,
            verification_mode,
        )
    }

    /// A faster version of [`Self::update`] if you already have a
    /// [`SharedReference`].
    ///
    /// If the passed-in [`SharedReference`] is the same as the existing cell's
    /// by identity, no update is performed.
    ///
    /// The value should be stored in [`SharedReference`] using the type `T`.
    pub fn update_with_shared_reference(
        &self,
        shared_ref: SharedReference,
        verification_mode: VerificationMode,
    ) {
        let tt = turbo_tasks();
        let update = if matches!(verification_mode, VerificationMode::EqualityCheck) {
            let content = tt.read_own_task_cell(self.current_task, self.index).ok();
            if let Some(TypedCellContent(_, CellContent(Some(shared_ref_exp)))) = content {
                // pointer equality (not value equality)
                shared_ref_exp != shared_ref
            } else {
                true
            }
        } else {
            true
        };
        if update {
            tt.update_own_task_cell(
                self.current_task,
                self.index,
                CellContent(Some(shared_ref)),
                None,
                None,
                verification_mode,
            )
        }
    }
}

impl From<CurrentCellRef> for RawVc {
    fn from(cell: CurrentCellRef) -> Self {
        RawVc::task_cell(cell.current_task, cell.index)
    }
}

fn extract_sr_value<T: VcValueType>(sr: &SharedReference) -> &T {
    sr.0.downcast_ref::<T>()
        .expect("cannot update SharedReference of different type")
}

pub fn find_cell_by_type<T: VcValueType>() -> CurrentCellRef {
    find_cell_by_id(T::get_value_type_id())
}

pub fn find_cell_by_id(ty: ValueTypeId) -> CurrentCellRef {
    CURRENT_TASK_STATE.with(|ts| {
        let current_task = current_task("celling turbo_tasks values");
        let mut ts = ts.write().unwrap();
        let map = ts.cell_counters.as_mut().unwrap();
        let current_index = map.entry(ty).or_default();
        let index = *current_index;
        assert!(
            index <= CellId::MAX_CELL_INDEX,
            "task allocated more than {} cells of a single type",
            CellId::MAX_CELL_INDEX as u64 + 1,
        );
        *current_index += 1;
        CurrentCellRef {
            current_task,
            index: CellId::new(ty, index),
        }
    })
}

pub(crate) async fn read_local_output(
    this: &dyn TurboTasksApi,
    execution_id: ExecutionId,
    local_task_id: LocalTaskId,
) -> Result<RawVc> {
    loop {
        match this.try_read_local_output(execution_id, local_task_id)? {
            Ok(raw_vc) => return Ok(raw_vc),
            Err(event_listener) => {
                // The local task is not done yet. If it is only scheduled, execute it right here
                // instead of waiting for a worker to pick it up.
                if execute_read_target_inline(
                    this,
                    ScheduleKey::LocalTask(execution_id, local_task_id),
                ) {
                    continue;
                }
                event_listener.await
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_inline_execution_depth_guard_restores_depth() {
        assert_eq!(INLINE_EXECUTION_DEPTH.get(), 0);
        {
            let _outer = InlineExecutionDepthGuard::enter();
            {
                let _inner = InlineExecutionDepthGuard::enter();
                assert_eq!(INLINE_EXECUTION_DEPTH.get(), 2);
            }
            assert_eq!(INLINE_EXECUTION_DEPTH.get(), 1);
        }
        assert_eq!(INLINE_EXECUTION_DEPTH.get(), 0);
    }

    #[test]
    fn test_inline_depth_cap() {
        assert!(inline_execution_allowed(), "nothing is nested yet");
        let mut guards = (0..MAX_INLINE_EXECUTION_DEPTH)
            .map(|_| InlineExecutionDepthGuard::enter())
            .collect::<Vec<_>>();
        assert_eq!(INLINE_EXECUTION_DEPTH.get(), MAX_INLINE_EXECUTION_DEPTH);
        assert!(
            !inline_execution_allowed(),
            "at the nesting cap reads wait for a worker instead of executing inline"
        );

        // One level below the cap inline execution is allowed again.
        guards.pop();
        assert!(inline_execution_allowed());
    }

    #[tokio::test]
    async fn test_poll_once_or_spawn_completed_execution() {
        assert!(
            poll_once_or_spawn(async {}),
            "a future that completes on the first poll is executed inline"
        );
        assert_eq!(INLINE_EXECUTION_DEPTH.get(), 0);
    }

    #[tokio::test]
    async fn test_poll_once_or_spawn_pending_execution() {
        let (tx, rx) = tokio::sync::oneshot::channel();
        let done = Arc::new(AtomicBool::new(false));
        let done_in_task = done.clone();
        assert!(
            !poll_once_or_spawn(async move {
                // Yields on the first poll, so it cannot be executed inline.
                tokio::task::yield_now().await;
                done_in_task.store(true, Ordering::SeqCst);
                let _ = tx.send(());
            }),
            "a future that yields is not completed inline"
        );
        assert_eq!(INLINE_EXECUTION_DEPTH.get(), 0);

        // ...but it was spawned, so it still runs to completion.
        rx.await.unwrap();
        assert!(done.load(Ordering::SeqCst));
    }
}