turbo-tasks: execute scheduled tasks inline when they are read

@@ -77,6 +77,9 @@ thiserror = { workspace = true } turbo-tasks-malloc = { workspace = true , features = ["custom_allocator"]} rstest = { workspace = true } turbo-tasks-testing = { workspace = true }+# The read-outcome tests assert on the counters, so they need them compiled in. Only for tests —+# normal builds of this crate get the default (disabled) turbo-tasks.+turbo-tasks = { workspace = true, features = ["inline_execution_stats"] }   [[bench]]

@@ -31,9 +31,9 @@ use tracing::{Span, field::display, trace_span}; use turbo_bincode::{TurboBincodeBuffer, new_turbo_bincode_decoder, new_turbo_bincode_encoder}; use turbo_tasks::{     CellId, DynTaskInputsStorage, RawVc, RawVcUnpacked, ReadCellOptions, ReadCellTracking,-    ReadConsistency, ReadOutputOptions, ReadTracking, SharedReference, TRANSIENT_TASK_BIT,-    TaskExecutionReason, TaskId, TaskPersistence, TaskPriority, TraitTypeId, TurboTasks,-    TurboTasksCallApi, TurboTasksPanic, ValueTypeId,+    ReadConsistency, ReadOutcome, ReadOutputOptions, ReadTracking, SharedReference,+    TRANSIENT_TASK_BIT, TaskExecutionReason, TaskId, TaskPersistence, TaskPriority, TraitTypeId,+    TurboTasks, TurboTasksCallApi, TurboTasksPanic, ValueTypeId,     backend::{         Backend, CachedTaskType, CachedTaskTypeArc, CellContent, CellHash, TaskExecutionSpec,         TransientTaskType, TurboTaskContextError, TurboTaskLocalContextError, TurboTasksError,@@ -486,7 +486,7 @@ impl TurboTasksBackend {         reader: Option<TaskId>,         options: ReadOutputOptions,         turbo_tasks: &TurboTasks<TurboTasksBackend>,-    ) -> Result<Result<RawVc, EventListener>> {+    ) -> Result<ReadOutcome<RawVc>> {         self.assert_not_persistent_calling_transient(reader, task_id, /* cell_id */ None);          let mut ctx = self.execute_context(turbo_tasks);@@ -518,19 +518,25 @@ impl TurboTasksBackend {             })         } -        fn check_in_progress(+        /// Reports whether the task is already being computed, and — crucially for the reader —+        /// whether a worker has actually started it. A task that is only `Scheduled` can be taken+        /// over and executed by the reader; one that is `InProgress` can only be waited for.+        fn check_in_progress<T>(             task: &impl TaskGuard,             reader_description: Option<EventDescription>,             tracking: ReadTracking,-        ) -> Option<std::result::Result<std::result::Result<RawVc, EventListener>, anyhow::Error>>-        {+        ) -> Option<Result<ReadOutcome<T>>> {             match task.get_in_progress() {-                Some(InProgressState::Scheduled { done_event, .. }) => Some(Ok(Err(-                    listen_to_done_event(reader_description, tracking, done_event),-                ))),+                Some(InProgressState::Scheduled { done_event, .. }) => {+                    Some(Ok(ReadOutcome::Scheduled(listen_to_done_event(+                        reader_description,+                        tracking,+                        done_event,+                    ))))+                }                 Some(InProgressState::InProgress(box InProgressStateInner {                     done_event, ..-                })) => Some(Ok(Err(listen_to_done_event(+                })) => Some(Ok(ReadOutcome::InProgress(listen_to_done_event(                     reader_description,                     tracking,                     done_event,@@ -704,7 +710,7 @@ impl TurboTasksBackend {                     queue.execute(&mut ctx);                 } -                return Ok(Err(listener));+                return Ok(ReadOutcome::InProgress(listener));             }         } @@ -723,8 +729,8 @@ impl TurboTasksBackend {          if let Some(output) = task.get_output() {             let result = match output {-                OutputValue::Cell(cell) => Ok(Ok(RawVc::task_cell(cell.task, cell.cell))),-                OutputValue::Output(task) => Ok(Ok(RawVc::task_output(*task))),+                OutputValue::Cell(cell) => Ok(RawVc::task_cell(cell.task, cell.cell)),+                OutputValue::Output(task) => Ok(RawVc::task_output(*task)),                 OutputValue::Error(error) => Err(error.clone()),             };             if let Some(mut reader_task) = reader_task.take()@@ -770,7 +776,7 @@ impl TurboTasksBackend {                 drop(task);             } -            return result.map_err(|error| {+            return result.map(ReadOutcome::Value).map_err(|error| {                 self.task_error_to_turbo_tasks_execution_error(&error, &mut ctx)                     .with_task_context(task_id, turbo_tasks.pin())                     .into()@@ -801,7 +807,7 @@ impl TurboTasksBackend {         debug_assert!(old.is_none(), "InProgress already exists");         ctx.schedule_task(task, TaskPriority::Recomputation); -        Ok(Err(listener))+        Ok(ReadOutcome::Scheduled(listener))     }      fn try_read_task_cell(@@ -811,7 +817,7 @@ impl TurboTasksBackend {         cell: CellId,         options: ReadCellOptions,         turbo_tasks: &TurboTasks<TurboTasksBackend>,-    ) -> Result<Result<TypedCellContent, EventListener>> {+    ) -> Result<ReadOutcome<TypedCellContent>> {         self.assert_not_persistent_calling_transient(reader, task_id, Some(cell));          fn add_cell_dependency(@@ -876,7 +882,7 @@ impl TurboTasksBackend {             if tracking.should_track(false) {                 add_cell_dependency(task_id, task, reader, reader_task, cell, tracking.key());             }-            return Ok(Ok(TypedCellContent(+            return Ok(ReadOutcome::Value(TypedCellContent(                 cell.type_id(),                 CellContent(Some(content)),             )));@@ -887,9 +893,17 @@ impl TurboTasksBackend {             in_progress,             Some(InProgressState::InProgress(..) | InProgressState::Scheduled { .. })         ) {-            return Ok(Err(self+            // Tell the reader whether the task is merely queued (it may take it over and execute it+            // itself) or already being executed by a worker (it can only wait).+            let started = matches!(in_progress, Some(InProgressState::InProgress(..)));+            let listener = self                 .listen_to_cell(&mut task, task_id, need_reader_task, &reader_task, cell)-                .0));+                .0;+            return Ok(if started {+                ReadOutcome::InProgress(listener)+            } else {+                ReadOutcome::Scheduled(listener)+            });         }         let is_cancelled = matches!(in_progress, Some(InProgressState::Canceled)); @@ -926,7 +940,8 @@ impl TurboTasksBackend {             self.listen_to_cell(&mut task, task_id, need_reader_task, &reader_task, cell);         drop(reader_task);         if !new_listener {-            return Ok(Err(listener));+            // Somebody else is already waiting for this cell, so the task is being taken care of.+            return Ok(ReadOutcome::InProgress(listener));         }          let _span = tracing::trace_span!(@@ -942,7 +957,7 @@ impl TurboTasksBackend {         );         ctx.schedule_task(task, TaskPriority::Recomputation); -        Ok(Err(listener))+        Ok(ReadOutcome::Scheduled(listener))     }      fn listen_to_cell(@@ -1982,7 +1997,12 @@ impl TurboTasksBackend {                 )             }             TaskType::Transient(task_type) => {-                let span = tracing::trace_span!("turbo_tasks::root_task");+                // `inline_execution` is recorded when a read executed this task on its own thread,+                // see `NativeFunction::span`.+                let span = tracing::trace_span!(+                    "turbo_tasks::root_task",+                    inline_execution = tracing::field::Empty+                );                 let future = match &*task_type {                     TransientTask::Root(f) => f(),                     TransientTask::Once(future_mutex) => take(&mut *future_mutex.lock())?,@@ -3662,7 +3682,7 @@ impl Backend for TurboTasksBackend {         reader: Option<TaskId>,         options: ReadOutputOptions,         turbo_tasks: &TurboTasks<Self>,-    ) -> Result<Result<RawVc, EventListener>> {+    ) -> Result<ReadOutcome<RawVc>> {         self.try_read_task_output(task_id, reader, options, turbo_tasks)     } @@ -3673,7 +3693,7 @@ impl Backend for TurboTasksBackend {         reader: Option<TaskId>,         options: ReadCellOptions,         turbo_tasks: &TurboTasks<Self>,-    ) -> Result<Result<TypedCellContent, EventListener>> {+    ) -> Result<ReadOutcome<TypedCellContent>> {         self.try_read_task_cell(task_id, reader, cell, options, turbo_tasks)     } 

@@ -796,6 +796,22 @@ fn restored_from_disk(result: &Option<Result<Option<TaskStorage>>>) -> bool {     matches!(result, Some(Ok(Some(_)))) } +/// The priority a task is scheduled with: an already computed task is a re-computation of a+/// (possibly deep) dependency, everything else starts at the initial priority.+fn schedule_priority(task: &impl TaskGuard, parent_priority: TaskPriority) -> TaskPriority {+    let priority = if task.has_output() {+        TaskPriority::invalidation(+            task.get_leaf_distance()+                .copied()+                .unwrap_or_default()+                .distance,+        )+    } else {+        TaskPriority::initial()+    };+    priority.in_parent(parent_priority)+}+ /// Combines per-category booleans into a single `TaskDataCategory` for waiting. fn wait_category(wait_data: bool, wait_meta: bool) -> Option<TaskDataCategory> {     match (wait_data, wait_meta) {@@ -1095,18 +1111,8 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> {     }      fn schedule_task(&self, task: Self::TaskGuardImpl, parent_priority: TaskPriority) {-        let priority = if task.has_output() {-            TaskPriority::invalidation(-                task.get_leaf_distance()-                    .copied()-                    .unwrap_or_default()-                    .distance,-            )-        } else {-            TaskPriority::initial()-        };-        self.turbo_tasks-            .schedule(task.id(), priority.in_parent(parent_priority));+        let priority = schedule_priority(&task, parent_priority);+        self.turbo_tasks.schedule(task.id(), priority);     }      fn get_current_task_priority(&self) -> TaskPriority {

@@ -0,0 +1,411 @@+#![feature(arbitrary_self_types)]+#![feature(arbitrary_self_types_pointers)]+#![allow(clippy::needless_return)] // tokio macro-generated code doesn't respect this++//! Reading a task that is scheduled but not started yet must execute that task *inline* on the+//! reading thread, so the read can complete without ever returning `Poll::Pending`.+//!+//! Most tests run on a single tokio worker on purpose: with `worker_threads = 1` the+//! `PriorityRunner`'s target worker count is 1, so a task scheduled from inside another task's+//! execution is always queued (never immediately spawned). That makes "the task is still in the+//! priority runner when it is read" deterministic. The tests that are about contention+//! (`test_read_of_running_task_is_not_executed_again`, `test_read_outcome_counters`) use more.++use std::{+    future::{Future, IntoFuture},+    pin::pin,+    sync::atomic::{AtomicUsize, Ordering},+    task::{Context, Poll, Waker},+};++use anyhow::Result;+use turbo_tasks::{ReadRef, ResolvedVc, State, Vc};+use turbo_tasks_testing::{Registration, register, run_once};++static REGISTRATION: Registration = register!();++/// Polls `fut` exactly once with a non-waking waker.+fn poll_once<F: Future>(fut: F) -> Poll<F::Output> {+    let mut fut = pin!(fut);+    let waker = Waker::noop();+    let mut cx = Context::from_waker(waker);+    fut.as_mut().poll(&mut cx)+}++/// A read of a not-yet-started task completes in a single poll, because the reader picks the+/// task up from the priority runner and executes it inline.+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]+async fn test_read_of_scheduled_task_is_inline() {+    let mut nonce = 0;+    run_once(&REGISTRATION, move || {+        nonce += 1;+        async move {+            read_scheduled_task_inline(nonce)+                .read_strongly_consistent()+                .await+        }+    })+    .await+    .unwrap();+}++#[turbo_tasks::function(operation, root)]+async fn read_scheduled_task_inline(nonce: u32) -> Result<Vc<()>> {+    // `leaf` has never been computed and completes without awaiting anything, so the read must+    // resolve in the very first poll.+    let leaf_vc = leaf(nonce);+    let Poll::Ready(result) = poll_once(leaf_vc.into_future()) else {+        panic!("reading a scheduled-but-not-started task did not complete inline");+    };+    assert_eq!(result?.value, 42);++    // A second read hits the cache and obviously stays inline.+    let Poll::Ready(result) = poll_once(leaf(nonce).into_future()) else {+        panic!("reading a completed task did not complete inline");+    };+    assert_eq!(result?.value, 42);++    Ok(Vc::cell(()))+}++/// When the inline execution yields, the read parks as usual and the execution is completed+/// elsewhere — the value must still arrive.+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]+async fn test_read_of_yielding_task_still_completes() {+    let mut nonce = 0;+    read_yielding_task(&mut nonce).await;+}++async fn read_yielding_task(nonce: &mut u32) {+    run_once(&REGISTRATION, {+        *nonce += 1;+        let nonce = *nonce;+        move || async move {+            read_yielding_task_operation(nonce)+                .read_strongly_consistent()+                .await+        }+    })+    .await+    .unwrap();+}++#[turbo_tasks::function(operation, root)]+async fn read_yielding_task_operation(nonce: u32) -> Result<Vc<()>> {+    // `yielding_leaf` yields during its execution, so the inline poll cannot finish it. The read+    // has to park — and the value must still be produced.+    let vc = yielding_leaf(nonce);+    assert!(+        poll_once(vc.into_future()).is_pending(),+        "a task that yields must not resolve in the first poll"+    );+    assert_eq!(vc.await?.value, 7);++    Ok(Vc::cell(()))+}++/// A call with unresolved arguments creates a *local* task. Reading its output must execute the+/// local task (and the global task it resolves to) inline as well.+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]+async fn test_read_of_local_task_is_inline() {+    let mut nonce = 0;+    run_once(&REGISTRATION, move || {+        nonce += 1;+        async move {+            read_local_task_inline(nonce)+                .read_strongly_consistent()+                .await+        }+    })+    .await+    .unwrap();+}++#[turbo_tasks::function(operation, root)]+async fn read_local_task_inline(nonce: u32) -> Result<Vc<()>> {+    // An unresolved `Vc` argument: `identity` is called with the not-yet-resolved output of+    // `leaf`, which creates a local resolve task.+    let leaf_vc = leaf(nonce);+    assert_eq!(leaf_vc.await?.value, 42);++    let local_vc = identity(leaf_vc);+    let Poll::Ready(result) = poll_once(local_vc.into_future()) else {+        panic!("reading a scheduled local task did not complete inline");+    };+    assert_eq!(result?.value, 42);++    Ok(Vc::cell(()))+}++/// Reports how reads and inline execution interacted, and checks the invariants of the counters.+///+/// Deliberately does *not* assert a particular mix: what a read finds depends on how loaded the+/// scheduler is. On a small, idle instance like this one, `connect_child` schedules each fresh task+/// eagerly and the worker it spawns wins the race against the reader, so claims mostly fail; in a+/// saturated build the tasks stay queued and the reader takes them over instead. Both are correct,+/// and the printed histogram is the point of this test.+///+/// In particular `waited_in_progress` cannot be asserted here: a worker pops a task from the queue+/// *before* `try_start_task_execution` flips the state to `InProgress`, so a read that lands in+/// that window still sees `Scheduled` and attempts a claim that cannot succeed. Whether a read sees+/// `Scheduled` or `InProgress` is therefore a matter of timing by design — it is a hint, and acting+/// on a stale one only costs a failed claim.+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]+async fn test_read_outcome_counters() {+    let tt = create_test_turbo_tasks("test_read_outcome_counters", true);++    let before = tt.inline_execution_stats();+    turbo_tasks::run_once(tt.clone(), async move {+        read_many_leaves(1, 21).read_strongly_consistent().await?;+        Ok(())+    })+    .await+    .unwrap();+    let after = tt.inline_execution_stats();++    println!("stats before: {before:#?}\nstats after: {after:#?}");+    assert!(+        after.claim_attempted > before.claim_attempted,+        "reads that find a task queued must try to take it over"+    );+    assert_eq!(+        after.claim_attempted - before.claim_attempted,+        (after.claim_completed - before.claim_completed)+            + (after.claim_yielded - before.claim_yielded)+            + (after.claim_failed - before.claim_failed),+        "every claim attempt has exactly one outcome"+    );+}++/// A task that is invalidated *while it is being executed* is stale and gets scheduled again, so a+/// read can go around the retry more than once. It has to terminate, and it must not grow the+/// native stack per retry (which is why the retry is a loop and not recursion).+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]+async fn test_stale_during_execution_terminates() {+    let result = tokio::time::timeout(std::time::Duration::from_secs(60), async {+        run_once(&REGISTRATION, || async {+            let input = ReadRef::resolved_cell(ReadRef::new_owned(ChangingInput {+                state: State::new(0),+            }));+            let output = self_invalidating(input);+            // The task bumps the state it depends on the first few times it runs, invalidating+            // itself, so this read has to survive several rescheduled executions.+            assert!(output.read_strongly_consistent().await?.value >= 3);+            Ok(())+        })+        .await+        .unwrap();+    })+    .await;+    assert!(+        result.is_ok(),+        "a task that keeps invalidating itself must still settle"+    );+}++#[turbo_tasks::value]+struct ChangingInput {+    state: State<u32>,+}++#[turbo_tasks::function(operation, root)]+async fn self_invalidating(input: ResolvedVc<ChangingInput>) -> Result<Vc<Value>> {+    let value = *input.await?.state.get();+    if value < 3 {+        // Invalidate ourselves: the execution that is running right now becomes stale and is+        // scheduled again.+        input.await?.state.set(value + 1);+    }+    Ok(Value { value: value + 1 }.cell())+}++/// Restoring from a persistent cache in a fresh instance — what an incremental build does — must+/// recompute what it cannot reuse, produce the right values, and not panic.+///+/// Worth keeping even though it looks mundane: reads behave differently here than in a cold build+/// (tasks come back dirty and are recomputed rather than found in the queue), and every other test+/// in this file — like every cold build — never exercises that.+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]+async fn test_restore_from_persistent_cache_recomputes_and_does_not_panic() {+    let name = "test_restore_from_persistent_cache_recomputes";++    // First instance: compute the tasks and flush them to the persistent cache.+    let first = create_test_turbo_tasks(name, true);+    turbo_tasks::run_once(first.clone(), async move {+        read_session_leaves(21).read_strongly_consistent().await?;+        Ok(())+    })+    .await+    .unwrap();+    first.stop_and_wait().await;++    // Second instance on the same cache: the session-dependent results cannot be reused, so they+    // have to be recomputed.+    let second = create_test_turbo_tasks(name, false);+    let before = second.inline_execution_stats();+    turbo_tasks::run_once(second.clone(), async move {+        read_session_leaves(21).read_strongly_consistent().await?;+        Ok(())+    })+    .await+    .unwrap();+    let after = second.inline_execution_stats();+    second.stop_and_wait().await;++    println!("stats before: {before:#?}\nstats after: {after:#?}");+    assert!(+        after.queued > before.queued,+        "the session-dependent tasks must be recomputed after the restore"+    );+}++/// Builds a `TurboTasks` on a per-name persistence directory. `initial` wipes that directory first;+/// passing `false` reuses what a previous instance flushed, which is how an incremental build+/// starts up.+fn create_test_turbo_tasks(+    name: &str,+    initial: bool,+) -> std::sync::Arc<turbo_tasks::TurboTasks<turbo_tasks_backend::TurboTasksBackend>> {+    let inner = include!(concat!(+        env!("CARGO_MANIFEST_DIR"),+        "/tests/test_config.trs"+    ));+    (inner)(name, initial)+}++#[turbo_tasks::function(operation, root)]+async fn read_session_leaves(leaves: u32) -> Result<Vc<()>> {+    for i in 0..leaves {+        assert_eq!(session_leaf(i).await?.value, i + 1);+    }+    Ok(Vc::cell(()))+}++/// Session-dependent, so its result is not reused across `TurboTasks` instances: a second instance+/// restoring from the same persistent cache has to recompute it, and the read is what schedules it.+#[turbo_tasks::function(session_dependent)]+async fn session_leaf(index: u32) -> Result<Vc<Value>> {+    Ok(Value { value: index + 1 }.cell())+}++#[turbo_tasks::function(operation, root)]+async fn read_many_leaves(nonce: u32, leaves: u32) -> Result<Vc<()>> {+    for i in 0..leaves {+        // Every leaf is a distinct, never-computed task.+        assert_eq!(leaf(nonce * 1000 + i).await?.value, 42);+    }+    Ok(Vc::cell(()))+}++#[turbo_tasks::value]+#[derive(Clone, Debug)]+struct Value {+    value: u32,+}++/// A task whose execution is already in progress is not executed a second time, and reading it+/// works as it always did.+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]+async fn test_read_of_running_task_is_not_executed_again() {+    let mut nonce = 0;+    run_once(&REGISTRATION, move || {+        nonce += 1;+        async move {+            read_running_task(nonce)+                .read_strongly_consistent()+                .await+                .map(|_| ())+        }+    })+    .await+    .unwrap();+    assert_eq!(+        COUNTED_EXECUTIONS.load(Ordering::SeqCst),+        1,+        "the task must be executed exactly once, no matter how many readers race for it"+    );+}++/// Only `read_running_task` may use this: it is process-global, and that operation resets it at the+/// start of each execution.+static COUNTED_EXECUTIONS: AtomicUsize = AtomicUsize::new(0);++#[turbo_tasks::function(operation, root)]+async fn read_running_task(nonce: u32) -> Result<Vc<()>> {+    COUNTED_EXECUTIONS.store(0, Ordering::SeqCst);+    // Eight readers of the same task, spread over the worker threads: at most one of them can+    // execute it (inline or on a worker), the others have to wait for it.+    let mut values = Vec::new();+    for _ in 0..8 {+        values.push(counted_leaf(nonce));+    }+    for value in values {+        assert_eq!(value.await?.value, 5);+    }+    Ok(Vc::cell(()))+}++#[turbo_tasks::function]+async fn counted_leaf(nonce: u32) -> Result<Vc<Value>> {+    // Keeps `nonce` in the cache key: `#[turbo_tasks::function]` filters out arguments the body+    // never uses, and then every run would reuse the first run's cached value.+    let _ = nonce;+    COUNTED_EXECUTIONS.fetch_add(1, Ordering::SeqCst);+    // Yield so the other readers reach the task while it is in progress.+    tokio::task::yield_now().await;+    Ok(Value { value: 5 }.cell())+}++/// Inline execution nests: reading the deepest task of a chain of uncomputed tasks executes them+/// one inside the other. The nesting cap keeps that from growing the stack without bounds.+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]+async fn test_deep_dependency_chain() {+    let mut nonce = 0;+    run_once(&REGISTRATION, move || {+        nonce += 1;+        async move {+            deep_chain_operation(nonce)+                .read_strongly_consistent()+                .await+                .map(|_| ())+        }+    })+    .await+    .unwrap();+}++#[turbo_tasks::function(operation, root)]+async fn deep_chain_operation(nonce: u32) -> Result<Vc<()>> {+    // Reading the top of the chain requires computing all 500 links, each of which reads the next+    // one as the first thing it does.+    assert_eq!(chain_link(nonce, 500).await?.value, 500);+    Ok(Vc::cell(()))+}++#[turbo_tasks::function]+async fn chain_link(nonce: u32, depth: u32) -> Result<Vc<Value>> {+    if depth == 0 {+        return Ok(Value { value: 0 }.cell());+    }+    let inner = chain_link(nonce, depth - 1).await?.value;+    Ok(Value { value: inner + 1 }.cell())+}++#[turbo_tasks::function]+fn leaf(nonce: u32) -> Result<Vc<Value>> {+    let _ = nonce; // keeps `nonce` in the cache key, see `counted_leaf`+    Ok(Value { value: 42 }.cell())+}++#[turbo_tasks::function]+async fn yielding_leaf(nonce: u32) -> Result<Vc<Value>> {+    let _ = nonce; // keeps `nonce` in the cache key, see `counted_leaf`+    tokio::task::yield_now().await;+    Ok(Value { value: 7 }.cell())+}++#[turbo_tasks::function]+async fn identity(input: Vc<Value>) -> Result<Vc<Value>> {+    let value = input.await?.value;+    Ok(Value { value }.cell())+}

@@ -15,6 +15,14 @@ hanging_detection = [] task_id_details = [] task_dirty_cause = [] verify_determinism = []+# Counts what reads do when they can't get a value straight away: queue pushes, claim attempts and+# their outcomes, and waits for a task a worker is already running. Reported by+# `TurboTasks::inline_execution_stats()`, and printed on shutdown with+# `TURBO_ENGINE_INLINE_STATS=1`.+#+# Off by default: the counters sit on the read-miss path, so a build that doesn't want the numbers+# shouldn't pay for them. With the feature off there are no counters and no atomics at all.+inline_execution_stats = []  # TODO(bgw): This feature is here to unblock turning on local tasks by default. It's only turned on # in unit tests. This will be removed very soon.

@@ -29,11 +29,10 @@ use turbo_rcstr::RcStr; use turbo_tasks_hash::DeterministicHasher;  use crate::{-    CellId, RawVc, ReadCellOptions, ReadOutputOptions, ReadRef, SharedReference, TaskId, TaskIdSet,-    TaskPriority, TraitRef, TraitTypeId, TurboTasksCallApi, TurboTasksPanic, ValueTypeId,-    ValueTypePersistence, VcValueTrait, VcValueType,+    CellId, RawVc, ReadCellOptions, ReadOutcome, ReadOutputOptions, ReadRef, SharedReference,+    TaskId, TaskIdSet, TaskPriority, TraitRef, TraitTypeId, TurboTasksCallApi, TurboTasksPanic,+    ValueTypeId, ValueTypePersistence, VcValueTrait, VcValueType,     dyn_task_inputs::{DynTaskInputs, DynTaskInputsStorage},-    event::EventListener,     macro_helpers::NativeFunction,     manager::{TaskPersistence, TurboTasks},     registry,@@ -654,7 +653,7 @@ pub trait Backend: Sized + Sync + Send {         reader: Option<TaskId>,         options: ReadOutputOptions,         turbo_tasks: &TurboTasks<Self>,-    ) -> Result<Result<RawVc, EventListener>>;+    ) -> Result<ReadOutcome<RawVc>>;      /// INVALIDATION: Be careful with this, when reader is None, it will not track dependencies, so     /// using it could break cache invalidation.@@ -665,7 +664,7 @@ pub trait Backend: Sized + Sync + Send {         reader: Option<TaskId>,         options: ReadCellOptions,         turbo_tasks: &TurboTasks<Self>,-    ) -> Result<Result<TypedCellContent, EventListener>>;+    ) -> Result<ReadOutcome<TypedCellContent>>;      /// INVALIDATION: Be careful with this, it will not track dependencies, so     /// using it could break cache invalidation.

@@ -73,6 +73,8 @@ use rustc_hash::FxHasher; pub use shrink_to_fit::ShrinkToFit; pub use turbo_tasks_macros::{DeterministicHash, turbobail, turbofmt}; +#[cfg(feature = "inline_execution_stats")]+pub use crate::manager::InlineExecutionStats; #[cfg(feature = "task_dirty_cause")] pub use crate::task_dirty_cause::TaskDirtyCause; pub use crate::{@@ -99,15 +101,15 @@ pub use crate::{     join_iter_ext::{JoinIterExt, TryFlatJoinIterExt, TryJoinIterExt},     manager::{         CurrentCellRef, InputResolution, ReadCellTracking, ReadConsistency, ReadTracking,-        TaskPersistence, TaskPriority, TurboTasks, TurboTasksApi, TurboTasksCallApi, Unused,-        UpdateInfo, dynamic_call, emit, get_serialization_invalidator, mark_finished,+        ScheduleKey, TaskPersistence, TaskPriority, TurboTasks, TurboTasksApi, TurboTasksCallApi,+        Unused, UpdateInfo, dynamic_call, emit, get_serialization_invalidator, mark_finished,         mark_stateful, mark_top_level_task, prevent_gc, run, run_once, run_once_with_reason,         trait_call, turbo_tasks, turbo_tasks_scope, turbo_tasks_weak,         unmark_top_level_task_may_leak_eventually_consistent_state, with_turbo_tasks,     },     mapped_read_ref::MappedReadRef,     output::OutputContent,-    read_options::{ReadCellOptions, ReadOutputOptions},+    read_options::{ReadCellOptions, ReadOutcome, ReadOutputOptions},     read_ref::ReadRef,     serialization_invalidation::SerializationInvalidator,     spawn::{JoinHandle, block_for_future, block_in_place, spawn, spawn_blocking, spawn_thread},

@@ -1,4 +1,5 @@ use std::{+    cell::Cell,     cmp::Reverse,     fmt::{Debug, Display},     future::Future,@@ -12,6 +13,7 @@ use std::{         Arc, Mutex, RwLock, Weak,         atomic::{AtomicBool, AtomicUsize, Ordering},     },+    task::{Context, Poll, Waker},     time::{Duration, Instant}, }; @@ -29,8 +31,8 @@ use turbo_tasks_hash::{DeterministicHash, hash_xxh3_hash128};  use crate::{     CellId, Completion, InvalidationReason, InvalidationReasonSet, OutputContent, RawVc,-    ReadCellOptions, ReadOutputOptions, ResolvedVc, SharedReference, TaskId, TraitMethod,-    ValueTypeId, Vc, VcRead, VcValueTrait, VcValueType,+    ReadCellOptions, ReadOutcome, ReadOutputOptions, ResolvedVc, SharedReference, TaskId,+    TraitMethod, ValueTypeId, Vc, VcRead, VcValueTrait, VcValueType,     backend::{         Backend, CellContent, CellHash, TaskCollectiblesMap, TaskExecutionSpec, TransientTaskType,         TurboTasksExecutionError, TypedCellContent, VerificationMode,@@ -43,7 +45,7 @@ use crate::{     local_task_tracker::LocalTaskTracker,     macro_helpers::NativeFunction,     message_queue::{CompilationEvent, CompilationEventQueue},-    priority_runner::{Executor, PriorityRunner},+    priority_runner::{Claimable, Executor, PriorityRunner},     registry,     serialization_invalidation::SerializationInvalidator,     task::local_task::{LocalTask, LocalTaskSpec, LocalTaskType},@@ -130,14 +132,14 @@ pub trait TurboTasksApi: TurboTasksCallApi + Sync + Send {         &self,         task: TaskId,         options: ReadOutputOptions,-    ) -> Result<Result<RawVc, EventListener>>;+    ) -> Result<ReadOutcome<RawVc>>;      fn try_read_task_cell(         &self,         task: TaskId,         index: CellId,         options: ReadCellOptions,-    ) -> Result<Result<TypedCellContent, EventListener>>;+    ) -> Result<ReadOutcome<TypedCellContent>>;      /// Reads a [`RawVc::LocalOutput`]. If the task has completed, returns the [`RawVc`] the local     /// task points to.@@ -161,6 +163,18 @@ pub trait TurboTasksApi: TurboTasksCallApi + Sync + Send {      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);@@ -465,12 +479,193 @@ enum ScheduledTask {     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,@@ -479,6 +674,9 @@ pub struct TurboTasks<B: Backend + 'static> {     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>>,@@ -625,6 +823,41 @@ task_local! {     /// 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> {@@ -643,6 +876,7 @@ impl<B: Backend + 'static> TurboTasks<B> {             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(),@@ -871,14 +1105,11 @@ impl<B: Backend + 'static> TurboTasks<B> {         self.begin_foreground_job();         self.scheduled_tasks.fetch_add(1, Ordering::AcqRel); -        self.priority_runner.schedule(-            &self.pin(),-            ScheduledTask::Task {-                task_id,-                span: Span::current(),-            },-            priority,-        );+        let task = ScheduledTask::Task {+            task_id,+            span: Span::current(),+        };+        self.priority_runner.schedule(&self.pin(), task, priority);     }      fn schedule_local_task(@@ -900,21 +1131,41 @@ impl<B: Backend + 'static> TurboTasks<B> {                 )             }); -        self.priority_runner.schedule(-            &self.pin(),-            ScheduledTask::LocalTask {-                ty,-                persistence,-                local_task_id,-                global_task_state,-                span: Span::current(),-            },-            priority,-        );+        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@@ -971,6 +1222,22 @@ impl<B: Backend + 'static> TurboTasks<B> {             .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.     ///@@ -1097,6 +1364,15 @@ impl<B: Backend + 'static> TurboTasks<B> {     }      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);@@ -1239,6 +1515,10 @@ impl<B: Backend> Executor<TurboTasks<B>, ScheduledTask, TaskPriority> for TurboT                                 .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; @@ -1291,6 +1571,7 @@ impl<B: Backend> Executor<TurboTasks<B>, ScheduledTask, TaskPriority> for TurboT             ScheduledTask::LocalTask {                 ty,                 persistence,+                execution_id: _,                 local_task_id,                 global_task_state,                 span,@@ -1307,6 +1588,9 @@ impl<B: Backend> Executor<TurboTasks<B>, ScheduledTask, TaskPriority> for TurboT                             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 {@@ -1472,7 +1756,7 @@ impl<B: Backend + 'static> TurboTasksApi for TurboTasks<B> {         &self,         task: TaskId,         options: ReadOutputOptions,-    ) -> Result<Result<RawVc, EventListener>> {+    ) -> Result<ReadOutcome<RawVc>> {         if options.consistency == ReadConsistency::Eventual {             debug_assert_not_in_top_level_task("read_task_output");         }@@ -1490,7 +1774,7 @@ impl<B: Backend + 'static> TurboTasksApi for TurboTasks<B> {         task: TaskId,         index: CellId,         options: ReadCellOptions,-    ) -> Result<Result<TypedCellContent, EventListener>> {+    ) -> Result<ReadOutcome<TypedCellContent>> {         let reader = current_task_if_available("reading Vcs");         self.backend             .try_read_task_cell(task, index, reader, options, self)@@ -1539,6 +1823,15 @@ impl<B: Backend + 'static> TurboTasksApi for TurboTasks<B> {         )     } +    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,@@ -1947,8 +2240,20 @@ pub(crate) async fn read_task_output( ) -> Result<RawVc> {     loop {         match this.try_read_task_output(id, options)? {-            Ok(result) => return Ok(result),-            Err(listener) => listener.await,+            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+            }         }     } }@@ -2284,7 +2589,83 @@ pub(crate) async fn read_local_output(     loop {         match this.try_read_local_output(execution_id, local_task_id)? {             Ok(raw_vc) => return Ok(raw_vc),-            Err(event_listener) => event_listener.await,+            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));     } }

@@ -289,6 +289,9 @@ impl NativeFunction {             TaskPersistence::Persistent => "",             TaskPersistence::Transient => "transient",         };+        // `inline_execution` is recorded when a read executed this task on its own thread instead+        // of waiting for a worker: "complete" if it finished there, "partial" if it yielded+        // and was handed to the runtime. It stays unset for a task a worker executed.         #[cfg(feature = "task_dirty_cause")]         {             tracing::trace_span!(@@ -298,6 +301,7 @@ impl NativeFunction {                 flags = flags,                 reason = reason.as_str(),                 cause = cause.map(tracing::field::display),+                inline_execution = tracing::field::Empty,             )         }         #[cfg(not(feature = "task_dirty_cause"))]@@ -308,11 +312,17 @@ impl NativeFunction {                 priority = %priority,                 flags = flags,                 reason = reason.as_str(),+                inline_execution = tracing::field::Empty,             )         }     }      pub fn resolve_span(&'static self, priority: TaskPriority) -> Span {-        tracing::trace_span!("turbo_tasks::resolve_call", name = self.ty.name, priority = %priority)+        tracing::trace_span!(+            "turbo_tasks::resolve_call",+            name = self.ty.name,+            priority = %priority,+            inline_execution = tracing::field::Empty,+        )     } }

@@ -2,6 +2,7 @@ use std::{     collections::BinaryHeap,     fmt::Debug,     future::Future,+    hash::Hash,     pin::Pin,     ptr::drop_in_place,     sync::{@@ -14,13 +15,29 @@ use std::{  use parking_lot::Mutex; use pin_project_lite::pin_project;+use rustc_hash::FxHashMap;  pub trait Executor<C, T, P>: Send + Sync {     type Future: Future<Output = ()> + Send;      fn execute(&self, execute_context: &Arc<C>, task: T, priority: P) -> Self::Future; } +/// A queued item that can be claimed by key before a worker starts executing it.+///+/// Claiming is how a reader takes over work it is about to wait for: instead of parking until some+/// worker gets around to the queued item, the reader removes it from the queue (see+/// [`PriorityRunner::claim`]) and drives it itself.+pub trait Claimable {+    type Key: Eq + Hash + Copy + Debug + Send + Sync;++    /// The key this item can be claimed by, or `None` when it must not be claimable.+    ///+    /// When multiple queued items share a key, only the most recently queued one is claimable; the+    /// others stay in the queue and are executed by workers as usual.+    fn claim_key(&self) -> Option<Self::Key>;+}+ impl<C, T, P, F, Fut> Executor<C, T, P> for F where     F: Fn(&Arc<C>, T, P) -> Fut + Send + Sync,@@ -33,42 +50,158 @@ where     } } -struct HeapItem<P, T> {+struct HeapItem<P> {     priority: P,-    task: T,+    /// Index into [`Queue::slots`]. The slot holds the queued item, or `None` when it was claimed+    /// (see [`Queue::claim`]).+    slot: usize, } -impl<P: Eq, T> PartialEq for HeapItem<P, T> {+impl<P: Eq> PartialEq for HeapItem<P> {     fn eq(&self, other: &Self) -> bool {         self.priority == other.priority     } } -impl<P: Eq, T> Eq for HeapItem<P, T> {}+impl<P: Eq> Eq for HeapItem<P> {} -impl<P: Ord, T> Ord for HeapItem<P, T> {+impl<P: Ord> Ord for HeapItem<P> {     fn cmp(&self, other: &Self) -> std::cmp::Ordering {         self.priority.cmp(&other.priority)     } } -impl<P: Ord, T> PartialOrd for HeapItem<P, T> {+impl<P: Ord> PartialOrd for HeapItem<P> {     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {         Some(self.cmp(other))     } } +/// The queue of items that are not scheduled yet.+///+/// Items are ordered by priority in a [`BinaryHeap`], but they are stored out-of-line in `slots` so+/// that a single item can be removed by key without disturbing the heap (a binary heap has no+/// keyed removal). Claiming an item takes the value out of its slot and leaves the heap entry+/// behind as a tombstone, which is skipped (and its slot recycled) when a worker pops it.+struct Queue<P, T: Claimable> {+    heap: BinaryHeap<HeapItem<P>>,+    /// The queued items with their priority. A slot is `Some` while the item is queued, `None` if+    /// it was claimed. The slot itself is only recycled once its (tombstone) heap entry has+    /// been popped, so a slot index is never reused while it is still referenced by the heap.+    ///+    /// The priority is stored here as well as in the heap entry because [`Queue::claim`] finds an+    /// item by key and never touches the heap, so unlike [`Queue::pop`] it has no heap entry to+    /// read it from — and [`Executor::execute`] needs the priority.+    slots: Vec<Option<(P, T)>>,+    /// Recycled indices into `slots`.+    free_slots: Vec<usize>,+    /// Slot index of the claimable item for each key.+    claimable: FxHashMap<T::Key, usize>,+    /// How many items were ever pushed. Diagnostics only, see [`PriorityRunner::total_queued`].+    #[cfg(feature = "inline_execution_stats")]+    pushes: u64,+}++impl<P: Clone + Ord, T: Claimable> Queue<P, T> {+    fn new() -> Self {+        Self {+            heap: BinaryHeap::new(),+            slots: Vec::new(),+            free_slots: Vec::new(),+            claimable: FxHashMap::default(),+            #[cfg(feature = "inline_execution_stats")]+            pushes: 0,+        }+    }++    /// Whether there is any heap entry left. This can be `true` while all remaining entries are+    /// tombstones of claimed items; popping is what cleans those up.+    fn is_empty(&self) -> bool {+        self.heap.is_empty()+    }++    fn push(&mut self, priority: P, task: T) {+        #[cfg(feature = "inline_execution_stats")]+        {+            self.pushes += 1;+        }+        let key = task.claim_key();+        let heap_priority = priority.clone();+        let slot = if let Some(slot) = self.free_slots.pop() {+            self.slots[slot] = Some((priority, task));+            slot+        } else {+            self.slots.push(Some((priority, task)));+            self.slots.len() - 1+        };+        if let Some(key) = key {+            // If this key is already queued, the older item stops being claimable. It stays in the+            // queue and is executed by a worker as usual.+            self.claimable.insert(key, slot);+        }+        self.heap.push(HeapItem {+            priority: heap_priority,+            slot,+        });+    }++    /// Pops the highest priority item, skipping tombstones of claimed items.+    fn pop(&mut self) -> Option<(P, T)> {+        while let Some(HeapItem { slot, .. }) = self.heap.pop() {+            let entry = self.slots[slot].take();+            self.free_slots.push(slot);+            if let Some((priority, task)) = entry {+                if let Some(key) = task.claim_key() {+                    // Only remove the mapping when it still points at this item. A newer item with+                    // the same key must stay claimable.+                    if self.claimable.get(&key) == Some(&slot) {+                        self.claimable.remove(&key);+                    }+                }+                self.shrink_amortized();+                return Some((priority, task));+            }+        }+        self.shrink_amortized();+        None+    }++    /// Removes the queued item with the given key, if it is still queued and claimable.+    fn claim(&mut self, key: &T::Key) -> Option<(P, T)> {+        let slot = self.claimable.remove(key)?;+        // The slot is intentionally not recycled here: its heap entry is still around as a+        // tombstone and must not start pointing at a different item.+        self.slots.get_mut(slot).and_then(|slot| slot.take())+    }++    /// Amortized shrinking of the queue, but with a lower threshold to avoid+    /// frequent reallocations when the queue is small.+    fn shrink_amortized(&mut self) {+        if self.heap.capacity() > self.heap.len() * 3 && self.heap.capacity() > 128 {+            let new_capacity = self.heap.len().next_power_of_two().max(128);+            self.heap.shrink_to(new_capacity);+        }+        if self.heap.is_empty() && self.claimable.is_empty() && self.slots.capacity() > 128 {+            // Nothing references any slot anymore.+            self.slots.clear();+            self.slots.shrink_to(128);+            self.free_slots.clear();+            self.free_slots.shrink_to(128);+        }+    }+}+ pub struct PriorityRunner<     C: Send + Sync + 'static,-    T: Send + 'static,-    P: Ord + Send + 'static,+    T: Claimable + Send + 'static,+    P: Clone + Ord + Send + 'static,     E: Executor<C, T, P> + 'static, > {     executor: E,     /// The target number of workers to spawn.     target_workers: usize,     /// The queue of tasks to execute. These tasks are not scheduled yet.-    queue: Mutex<BinaryHeap<HeapItem<P, T>>>,+    queue: Mutex<Queue<P, T>>,     /// The number of active workers currently polling tasks.     /// Workers that responded with Poll::Pending are not counted until they are polled again.     active_workers: AtomicUsize,@@ -77,28 +210,47 @@ pub struct PriorityRunner<  impl<     C: Send + Sync + 'static,-    T: Send + 'static,-    P: Debug + Ord + Send + 'static,+    T: Claimable + Send + 'static,+    P: Clone + Debug + Ord + Send + 'static,     E: Executor<C, T, P> + 'static, > PriorityRunner<C, T, P, E> {     pub fn new(executor: E) -> Self {+        Self::with_target_workers(+            executor,+            tokio::runtime::Handle::current().metrics().num_workers(),+        )+    }++    fn with_target_workers(executor: E, target_workers: usize) -> Self {         Self {             executor,-            target_workers: tokio::runtime::Handle::current().metrics().num_workers(),-            queue: Mutex::new(BinaryHeap::new()),+            target_workers,+            queue: Mutex::new(Queue::new()),             active_workers: AtomicUsize::new(0),             phantom: std::marker::PhantomData,         }     } +    /// How many tasks were ever put into the queue, as opposed to being executed without ever being+    /// queued. Diagnostics only — it lets a test assert that a task never took the detour through+    /// the queue.+    #[cfg(feature = "inline_execution_stats")]+    pub fn total_queued(&self) -> u64 {+        self.queue.lock().pushes+    }+     pub fn schedule(self: &Arc<Self>, execute_context: &Arc<C>, task: T, priority: P) {         let mut queue = self.queue.lock();         if !queue.is_empty() {             // If there is already work in the queue, we don't have any             // free capacity so we can just push the task to the queue.             // It will be picked up by existing workers.-            queue.push(HeapItem { priority, task });+            //+            // A worker only stops when it finds the queue empty, so a non-empty queue always has a+            // worker that will drain it. [`claim`](Self::claim) does take work out of the queue+            // without being a worker, but that only ever makes the queue shorter.+            queue.push(priority, task);             return;         }         // The queue is empty, so we might have free capacity to spawn a new worker.@@ -111,14 +263,25 @@ impl<             WorkerFuture::spawn(future, execute_context.clone(), self.clone());         } else {             // No free capacity, push the task to the queue.-            queue.push(HeapItem { priority, task });+            queue.push(priority, task);             drop(queue);              // Undo the added active worker since we didn't spawn a new worker.             self.decrease_active_workers(execute_context);         }     } +    /// Takes the queued task with the given key out of the queue and returns its execution future,+    /// or `None` when there is no such task in the queue (it was never scheduled, a worker already+    /// picked it up, or it was claimed before).+    ///+    /// The caller takes over the responsibility to drive the returned future to completion; the+    /// task left the queue, so no worker will do it.+    pub fn claim(&self, execute_context: &Arc<C>, key: &T::Key) -> Option<E::Future> {+        let (priority, task) = self.queue.lock().claim(key)?;+        Some(self.executor.execute(execute_context, task, priority))+    }+     /// Tries to decrease the active worker count by 1.     /// If there is work available in the queue, a new worker is spawned instead.     fn reuse_or_decrease_active_workers(self: &Arc<Self>, execute_context: &Arc<C>) {@@ -148,31 +311,18 @@ impl<     }      fn pop_future_from_worker(&self, execute_context: &Arc<C>) -> Option<E::Future> {-        let mut queue = self.queue.lock();-        if let Some(heap_item) = queue.pop() {-            shrink_amortized(&mut queue);-            drop(queue);-            Some(-                self.executor-                    .execute(execute_context, heap_item.task, heap_item.priority),-            )-        } else {-            None-        }+        let popped = self.queue.lock().pop();+        popped.map(|(priority, task)| self.executor.execute(execute_context, task, priority))     }      fn spawn_worker_if_work_available(         self: &Arc<Self>,         execute_context: &Arc<C>,         unused_active_count: bool,     ) -> bool {-        let mut queue = self.queue.lock();-        if let Some(heap_item) = queue.pop() {-            shrink_amortized(&mut queue);-            drop(queue);-            let new_future =-                self.executor-                    .execute(execute_context, heap_item.task, heap_item.priority);+        let popped = self.queue.lock().pop();+        if let Some((priority, task)) = popped {+            let new_future = self.executor.execute(execute_context, task, priority);              if !unused_active_count {                 self.active_workers.fetch_add(1, Ordering::Relaxed);@@ -185,15 +335,6 @@ impl<     } } -fn shrink_amortized<P, T>(queue: &mut BinaryHeap<HeapItem<P, T>>) {-    // Amortized shrinking of the queue, but with a lower threshold to avoid-    // frequent reallocations when the queue is small.-    if queue.capacity() > queue.len() * 3 && queue.capacity() > 128 {-        let new_capacity = queue.len().next_power_of_two().max(128);-        queue.shrink_to(new_capacity);-    }-}- #[derive(Debug)] enum WorkerState {     UnfinishedFuture,@@ -209,8 +350,10 @@ pin_project! {         C: Send,         C: Sync,         C: 'static,+        T: Claimable,         T: Send,         T: 'static,+        P: Clone,         P: Ord,         P: Send,         P: 'static,@@ -228,8 +371,8 @@ pin_project! {  impl<     C: Send + Sync + 'static,-    T: Send + 'static,-    P: Debug + Ord + Send + 'static,+    T: Claimable + Send + 'static,+    P: Clone + Debug + Ord + Send + 'static,     E: Executor<C, T, P> + 'static, > WorkerFuture<C, T, P, E> {@@ -245,8 +388,8 @@ impl<  impl<     C: Send + Sync + 'static,-    T: Send + 'static,-    P: Debug + Ord + Send + 'static,+    T: Claimable + Send + 'static,+    P: Clone + Debug + Ord + Send + 'static,     E: Executor<C, T, P> + 'static, > Future for WorkerFuture<C, T, P, E> {@@ -336,6 +479,224 @@ mod tests {      use super::*; +    impl Claimable for u32 {+        type Key = u32;++        fn claim_key(&self) -> Option<u32> {+            Some(*self)+        }+    }++    impl Claimable for (u32, bool) {+        type Key = u32;++        fn claim_key(&self) -> Option<u32> {+            Some(self.0)+        }+    }++    /// An item that is never claimable, to check that `None` keys are queued and executed as usual.+    #[derive(Clone, Copy, Debug, PartialEq, Eq)]+    struct Unkeyed(u32);++    impl Claimable for Unkeyed {+        type Key = u32;++        fn claim_key(&self) -> Option<u32> {+            None+        }+    }++    /// An executor that records which items it was asked to execute, in order, and whose futures+    /// complete immediately. Lets the queue be driven without a tokio runtime.+    struct RecordingExecutor;++    impl<T: Claimable + Copy + Send + Sync + Debug + 'static> Executor<Mutex<Vec<T>>, T, u32>+        for RecordingExecutor+    {+        type Future = std::future::Ready<()>;++        fn execute(+            &self,+            execute_context: &Arc<Mutex<Vec<T>>>,+            task: T,+            _priority: u32,+        ) -> Self::Future {+            execute_context.lock().push(task);+            std::future::ready(())+        }+    }++    /// The recorded executions of a test runner, in execution order.+    type Executions<T> = Arc<Mutex<Vec<T>>>;+    /// A test runner over items of type `T`.+    type TestRunner<T> = Arc<PriorityRunner<Mutex<Vec<T>>, T, u32, RecordingExecutor>>;++    /// A runner that queues every scheduled item (`target_workers == 0`, so no worker is ever+    /// spawned) and therefore needs no tokio runtime. `pop_future_from_worker` stands in for what a+    /// worker would do.+    fn queueing_runner<T: Claimable + Copy + Send + Sync + Debug + 'static>()+    -> (TestRunner<T>, Executions<T>) {+        (+            Arc::new(PriorityRunner::with_target_workers(RecordingExecutor, 0)),+            Arc::new(Mutex::new(Vec::new())),+        )+    }++    /// Drains the queue the way workers would and returns the items in execution order.+    fn drain<T: Claimable + Copy + Send + Sync + Debug + 'static>(+        runner: &TestRunner<T>,+        executed: &Executions<T>,+    ) -> Vec<T> {+        while runner.pop_future_from_worker(executed).is_some() {}+        let items = executed.lock().clone();+        executed.lock().clear();+        items+    }++    #[test]+    fn test_claim_queued_entry_by_key() {+        let (runner, executed) = queueing_runner::<u32>();+        for task in 0..4 {+            runner.schedule(&executed, task, task);+        }++        // Claiming builds the execution future, which the recording executor counts as executed.+        assert!(runner.claim(&executed, &2).is_some());+        assert_eq!(*executed.lock(), vec![2]);+        executed.lock().clear();++        // The claimed entry is gone from the queue; everything else still runs, highest priority+        // first.+        assert_eq!(drain(&runner, &executed), vec![3, 1, 0]);+    }++    #[test]+    fn test_claim_unknown_key_returns_none() {+        let (runner, executed) = queueing_runner::<u32>();+        runner.schedule(&executed, 1, 1);++        // Never scheduled.+        assert!(runner.claim(&executed, &42).is_none());+        // Already executed by a "worker".+        assert_eq!(drain(&runner, &executed), vec![1]);+        assert!(runner.claim(&executed, &1).is_none());+        assert!(executed.lock().is_empty());+    }++    #[test]+    fn test_claim_twice_returns_none() {+        let (runner, executed) = queueing_runner::<u32>();+        runner.schedule(&executed, 7, 7);++        assert!(runner.claim(&executed, &7).is_some());+        assert!(runner.claim(&executed, &7).is_none());+        assert_eq!(*executed.lock(), vec![7]);+        executed.lock().clear();++        // Only a tombstone is left.+        assert!(drain(&runner, &executed).is_empty());+    }++    #[test]+    fn test_claimed_entry_is_executed_exactly_once() {+        let (runner, executed) = queueing_runner::<u32>();+        for task in 0..10 {+            runner.schedule(&executed, task, task);+        }+        for task in [0, 5, 9] {+            assert!(runner.claim(&executed, &task).is_some());+        }+        let mut all = drain(&runner, &executed);+        all.sort_unstable();+        // Every scheduled item was executed exactly once: three by the claimer, the rest by+        // "workers".+        assert_eq!(all, (0..10).collect::<Vec<_>>());+    }++    #[test]+    fn test_claim_preserves_priority_order() {+        let (runner, executed) = queueing_runner::<u32>();+        for task in 0..6 {+            runner.schedule(&executed, task, task);+        }+        assert!(runner.claim(&executed, &4).is_some());+        executed.lock().clear();++        assert_eq!(drain(&runner, &executed), vec![5, 3, 2, 1, 0]);+    }++    #[test]+    fn test_duplicate_keys() {+        let (runner, executed) = queueing_runner::<(u32, bool)>();+        // Both items share the claim key `1`.+        runner.schedule(&executed, (1, false), 1);+        runner.schedule(&executed, (1, true), 2);++        // The most recently queued item is the claimable one.+        assert!(runner.claim(&executed, &1).is_some());+        assert_eq!(*executed.lock(), vec![(1, true)]);+        executed.lock().clear();+        // The other one is not claimable anymore, but it is not lost either.+        assert!(runner.claim(&executed, &1).is_none());+        assert_eq!(drain(&runner, &executed), vec![(1, false)]);+    }++    #[test]+    fn test_unkeyed_entries_are_not_claimable() {+        let (runner, executed) = queueing_runner::<Unkeyed>();+        runner.schedule(&executed, Unkeyed(1), 1);+        runner.schedule(&executed, Unkeyed(2), 2);++        assert!(runner.claim(&executed, &1).is_none());+        assert_eq!(+            drain(&runner, &executed),+            vec![Unkeyed(2), Unkeyed(1)],+            "unkeyed items are queued and executed as usual"+        );+    }++    #[test]+    fn test_slots_are_recycled() {+        let (runner, executed) = queueing_runner::<u32>();+        for _ in 0..100 {+            for task in 0..8 {+                runner.schedule(&executed, task, task);+            }+            // Claim one of them each round, so tombstones are part of the cycle.+            assert!(runner.claim(&executed, &3).is_some());+            drain(&runner, &executed);+            let queue = runner.queue.lock();+            assert!(queue.is_empty());+            assert!(+                queue.slots.len() <= 8,+                "slots should be recycled, got {}",+                queue.slots.len()+            );+            assert!(+                queue.claimable.is_empty(),+                "claimable index should be empty when the queue is empty"+            );+        }+    }++    /// Every push into the queue is counted, so a test can assert that a task was executed without+    /// ever being queued.+    #[cfg(feature = "inline_execution_stats")]+    #[test]+    fn test_total_queued_counts_pushes() {+        let (runner, executed) = queueing_runner::<u32>();+        assert_eq!(runner.total_queued(), 0);+        for task in 0..3 {+            runner.schedule(&executed, task, task);+        }+        assert_eq!(runner.total_queued(), 3);+        // Claiming and draining do not change how many pushes happened.+        assert!(runner.claim(&executed, &1).is_some());+        drain(&runner, &executed);+        assert_eq!(runner.total_queued(), 3);+    }+     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]     async fn test_cpu_bound_tasks() {         struct ExecutorImpl;

@@ -1,4 +1,25 @@-use crate::{ReadConsistency, ReadTracking, manager::ReadCellTracking};+use crate::{ReadConsistency, ReadTracking, event::EventListener, manager::ReadCellTracking};++/// What a read of a task's output or cell found.+///+/// The two "not available yet" cases are distinguished because the reader can act on them+/// differently: a task that is only *scheduled* can be taken over and executed by the reader, while+/// one that some worker is already executing can only be waited for — and finding that out without+/// taking the scheduler's queue lock is the point of telling them apart.+///+/// The state is a **hint**: it is a snapshot, and a task can be picked up by a worker right after+/// the read looked (a worker even pops a task off the queue before it marks it as started). Acting+/// on a stale hint costs at most a failed claim, never correctness.+pub enum ReadOutcome<T> {+    /// The value is available.+    Value(T),+    /// The task is queued but no worker has started it, so the reader may take it out of the queue+    /// and execute it itself. The listener fires when the task is done.+    Scheduled(EventListener),+    /// A worker is already executing the task; there is nothing to take over. The listener fires+    /// when the task is done.+    InProgress(EventListener),+}  #[derive(Clone, Copy, Debug, Default)] pub struct ReadCellOptions {

@@ -19,8 +19,10 @@ use crate::{     id::{ExecutionId, LocalTaskId, TASK_ID_MAX},     manager::{         ReadCellTracking, ReadTracking, SUPPRESS_EVENTUAL_CONSISTENCY_TOP_LEVEL_TASK_CHECK,-        TurboTasksApi, read_local_output, with_turbo_tasks,+        ScheduleKey, TurboTasksApi, execute_read_target_inline, read_local_output,+        with_turbo_tasks,     },+    read_options::ReadOutcome,     registry::get_value_type,     turbo_tasks, };@@ -502,6 +504,15 @@ fn suppress_top_level_task_check<R>(strongly_consistent: bool, f: impl FnOnce()     } } +/// Executes the task a read is waiting for when it is only scheduled, so the read can continue+/// without waiting for a worker.+///+/// Must be called *outside* [`with_turbo_tasks`]: executing a task enters a task-local scope of its+/// own, which panics while the read borrows `TURBO_TASKS`.+fn execute_inline(key: ScheduleKey) {+    execute_read_target_inline(&*turbo_tasks(), key);+}+ #[must_use] pub struct ResolveRawVcFuture {     current: RawVc,@@ -551,14 +562,20 @@ impl Future for ResolveRawVcFuture {         // SAFETY: we are not moving self         let this = unsafe { self.get_unchecked_mut() }; -        let poll_fn = |tt: &Arc<dyn TurboTasksApi>| -> Poll<Self::Output> {+        let strongly_consistent = this.strongly_consistent;+        // `execute_inline` is the task to execute inline, reported out-of-band so that the value+        // path stays exactly as cheap as it was. It is only set for tasks that are merely+        // scheduled; one that a worker is already executing cannot be taken over.+        let mut poll_fn = |tt: &Arc<dyn TurboTasksApi>,+                           execute_inline: &mut Option<ScheduleKey>|+         -> Poll<Self::Output> {             'outer: loop {                 ready!(poll_listener(&mut this.listener, cx));-                let listener = match this.current.unpack() {+                let (listener, key) = match this.current.unpack() {                     RawVcUnpacked::TaskOutput(task) => {                         let read_result = tt.try_read_task_output(task, this.read_output_options);                         match read_result {-                            Ok(Ok(vc)) => {+                            Ok(ReadOutcome::Value(vc)) => {                                 // turbo-tasks-backend doesn't currently have any sort of                                 // "transaction" or global lock mechanism to group together chains                                 // of `TaskOutput`/`TaskCell` reads.@@ -571,7 +588,19 @@ impl Future for ResolveRawVcFuture {                                 this.current = vc;                                 continue 'outer;                             }-                            Ok(Err(listener)) => listener,+                            // Nobody has started the task yet, so the caller may take it over.+                            Ok(ReadOutcome::Scheduled(listener)) => {+                                (listener, Some(ScheduleKey::Task(task)))+                            }+                            Ok(ReadOutcome::InProgress(listener)) => {+                                // A worker is on it; nothing to take over. Loop back so+                                // `poll_listener` registers the waker on this listener — returning+                                // `Pending` here would sleep through the event.+                                #[cfg(feature = "inline_execution_stats")]+                                tt.note_waited_for_in_progress_task();+                                this.listener = Some(listener);+                                continue 'outer;+                            }                             Err(err) => return Poll::Ready(Err(err)),                         }                     }@@ -587,22 +616,40 @@ impl Future for ResolveRawVcFuture {                                 this.current = vc;                                 continue 'outer;                             }-                            Ok(Err(listener)) => listener,+                            Ok(Err(listener)) => (+                                listener,+                                Some(ScheduleKey::LocalTask(execution_id, local_task_id)),+                            ),                             Err(err) => return Poll::Ready(Err(err)),                         }                     }                 };+                // The task is not done yet, so we have to wait for it — unless it is merely+                // scheduled, in which case our caller executes it and we read again.                 this.listener = Some(listener);+                *execute_inline = key;+                return Poll::Pending;             }         }; -        // HACK: Temporarily suppress top-level task check if doing strongly consistent read.-        //-        // This masks a bug: There's an unlikely TOCTOU race condition in `poll_fn`. Because the-        // strongly consistent read isn't a single atomic operation, any inner `TaskOutput` or-        // `TaskCell` could get mutated after the strongly consistent read of the outer-        // `TaskOutput`.-        suppress_top_level_task_check(this.strongly_consistent, || with_turbo_tasks(poll_fn))+        loop {+            let mut execute_inline_key = None;+            // HACK: Temporarily suppress top-level task check if doing strongly consistent read.+            //+            // This masks a bug: There's an unlikely TOCTOU race condition in `poll_fn`. Because the+            // strongly consistent read isn't a single atomic operation, any inner `TaskOutput` or+            // `TaskCell` could get mutated after the strongly consistent read of the outer+            // `TaskOutput`.+            let result = suppress_top_level_task_check(strongly_consistent, || {+                with_turbo_tasks(|tt| poll_fn(tt, &mut execute_inline_key))+            });+            if let Some(key) = execute_inline_key {+                // Not inside `with_turbo_tasks`, see `execute_inline`.+                execute_inline(key);+                continue;+            }+            return result;+        }     } } @@ -724,23 +771,53 @@ impl Future for ReadRawVcFuture {         let index = *index;         let read_cell_options = this.read_cell_options; -        let poll_fn = |tt: &Arc<dyn TurboTasksApi>| -> Poll<Self::Output> {+        let strongly_consistent = *strongly_consistent;++        let mut poll_fn = |tt: &Arc<dyn TurboTasksApi>,+                           execute_inline: &mut Option<ScheduleKey>|+         -> Poll<Self::Output> {             loop {                 ready!(poll_listener(listener, cx));-                let new_listener = match tt.try_read_task_cell(task, index, read_cell_options) {-                    Ok(Ok(content)) => return Poll::Ready(Ok(content)),-                    Ok(Err(l)) => l,-                    Err(err) => return Poll::Ready(Err(err)),-                };+                let (new_listener, key) =+                    match tt.try_read_task_cell(task, index, read_cell_options) {+                        Ok(ReadOutcome::Value(content)) => return Poll::Ready(Ok(content)),+                        Ok(ReadOutcome::Scheduled(l)) => (l, Some(ScheduleKey::Task(task))),+                        Ok(ReadOutcome::InProgress(l)) => {+                            // A worker is already filling the cell; nothing to take over. Loop back+                            // so `poll_listener` registers the waker on this listener — returning+                            // `Pending` here would sleep through the event.+                            #[cfg(feature = "inline_execution_stats")]+                            tt.note_waited_for_in_progress_task();+                            *listener = Some(l);+                            continue;+                        }+                        Err(err) => return Poll::Ready(Err(err)),+                    };+                // The cell isn't available yet, so we have to wait for the task that fills it —+                // unless that task is merely scheduled, in which case our caller executes it and we+                // read again.                 *listener = Some(new_listener);+                *execute_inline = key;+                return Poll::Pending;             }         }; -        // Phase 2 must also suppress the top-level task check when phase 1 was-        // strongly-consistent. The suppression from `ResolveRawVcFuture::poll` only lasts for-        // the duration of that individual `poll` call and does not carry over to subsequent calls-        // or to this phase.-        suppress_top_level_task_check(*strongly_consistent, || with_turbo_tasks(poll_fn))+        loop {+            let mut execute_inline_key = None;+            // Phase 2 must also suppress the top-level task check when phase 1 was+            // strongly-consistent. The suppression from `ResolveRawVcFuture::poll` only lasts for+            // the duration of that individual `poll` call and does not carry over to subsequent+            // calls or to this phase.+            let result = suppress_top_level_task_check(strongly_consistent, || {+                with_turbo_tasks(|tt| poll_fn(tt, &mut execute_inline_key))+            });+            if let Some(key) = execute_inline_key {+                // Not inside `with_turbo_tasks`, see `execute_inline`.+                execute_inline(key);+                continue;+            }+            return result;+        }     } }