vercel/next.js · #100465
turbo-tasks-backend: restore invalidated tasks with batched optional reads
turbopack/crates/turbo-tasks-backend/src/backend/operation/invalidate.rs16 + / 9 −
@@ -5,7 +5,6 @@ use turbo_tasks::{TaskExecutionReason, TaskId, TaskPriority, event::EventDescrip use crate::{ backend::{- TaskDataCategory, operation::{ ExecuteContext, TaskGuard, aggregation_update::{@@ -17,31 +16,39 @@ use crate::{ data::{Dirtyness, InProgressState, InProgressStateInner}, }; +/// Invalidates `task_ids`. Callers must not pass duplicate ids (the bulk task access expects+/// unique ids, and deduplicating here would cost every invalidation). pub fn invalidate( task_ids: SmallVec<[TaskId; 4]>, #[cfg(feature = "task_dirty_cause")] cause: TaskDirtyCause, mut ctx: ExecuteContext<'_>, ) {+ debug_assert!(+ task_ids.len() <= 1+ || task_ids.iter().collect::<rustc_hash::FxHashSet<_>>().len() == task_ids.len(),+ "invalidate() must not be called with duplicate task ids: {task_ids:?}"+ ); let mut queue = AggregationUpdateQueue::new(); // The tasks to schedule are collected and handed to the scheduler as one batch, which takes // the scheduler's queue lock once instead of once per task. let mut to_schedule: SmallVec<[_; 4]> = SmallVec::with_capacity(task_ids.len());- for task_id in task_ids {- let Some(mut task) = ctx.try_task(task_id, TaskDataCategory::All) else {- // The task no longer exists, nothing to invalidate.- continue;- };+ // Tasks that no longer exist are skipped, there is nothing to invalidate. Tasks that are not+ // resident are restored from persistent storage in batches.+ ctx.for_each_try_task_all(task_ids, "invalidate", |mut task, ctx| { if let Some(parent_priority) = make_task_dirty_without_scheduling( &mut task, true, #[cfg(feature = "task_dirty_cause")] cause.clone(), &mut queue,- &mut ctx,+ ctx, ) {- to_schedule.push((task_id, ctx.schedule_task_priority(&task, parent_priority)));+ to_schedule.push((+ task.id(),+ ctx.schedule_task_priority(&task, parent_priority),+ )); }- }+ }); ctx.schedule_tasks(&to_schedule); queue.execute(&mut ctx); }turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs401 + / 29 −
@@ -243,6 +243,8 @@ impl<'e> ExecuteContext<'e> { self.backend.should_restore(), "restore_task_data called when should_restore() is false" );+ #[cfg(test)]+ restore_lookup_counts::record(false); self.backend .backing_storage .lookup_data(task_id, category)@@ -261,6 +263,8 @@ impl<'e> ExecuteContext<'e> { self.backend.should_restore(), "restore_task_data_batch called when should_restore() is false" );+ #[cfg(test)]+ restore_lookup_counts::record(true); self.backend .backing_storage .batch_lookup_data(task_ids, category)@@ -380,19 +384,24 @@ impl<'e> ExecuteContext<'e> { Ok((task, outcome)) } - /// [`Self::restore_task`] on a freshly acquired guard, as a `MustExist` open.+ /// [`Self::restore_task`] on a freshly acquired guard. fn restore_existing_task( &self, task_id: TaskId, category: TaskDataCategory,+ access: TaskAccess, ) -> Result<(TaskEntryGuard<'e>, RestoreOutcome)> { let task = self.backend.storage.access_entry_mut(task_id);- self.restore_task(task_id, category, task, TaskAccess::MustExist)+ self.restore_task(task_id, category, task, access) } - /// Restores a batch of tasks, then hands each to `prepared_task_callback`. Like- /// [`ExecuteContext::task`], every task must have the requested category on disk if it is not- /// resident; a missing key panics.+ /// Restores a batch of tasks, then hands each to `prepared_task_callback`.+ ///+ /// With [`TaskAccess::MustExist`], like [`ExecuteContext::task`], every task must have the+ /// requested category on disk if it is not resident; a missing key panics. With+ /// [`TaskAccess::AllowMissing`], like [`ExecuteContext::try_task`], a task that is soft-deleted+ /// or missing in memory and on disk is skipped instead (a missing one is discarded), and the+ /// callback is not called for it. `MaybeCreate` is not supported. /// /// `prepared_task_callback` must not panic. A missing key is reported only after every task in /// the batch has been handed off, so that panic leaves nothing behind. A panicking callback is@@ -402,6 +411,7 @@ impl<'e> ExecuteContext<'e> { fn prepare_tasks_with_callback( &mut self, task_ids: impl IntoIterator<Item = (TaskId, TaskDataCategory)>,+ access: TaskAccess, reason: &'static str, mut prepared_task_callback: impl FnMut( &mut Self,@@ -412,18 +422,35 @@ impl<'e> ExecuteContext<'e> { ) { #[cfg(feature = "trace_prepare_tasks")] let _span = trace_span!("prepare_tasks_with_callback", reason).entered();+ debug_assert!(+ access != TaskAccess::MaybeCreate,+ "prepare_tasks_with_callback does not create tasks"+ );+ // Whether `task` is handed to the callback, under `AllowMissing` like `open_task`: a+ // soft-deleted task is treated as missing.+ let skip_deleted =+ |task: &TaskStorage| access == TaskAccess::AllowMissing && task.flags.deleted(); // Fast path: no backing storage to restore from, so every task that exists is already // restored. if !self.backend.should_restore() { for (task_id, category) in task_ids { self.task_lock_counter.acquire();- let task = self.backend.storage.access_mut(task_id);+ let task = self.backend.storage.access_entry_mut(task_id);+ if skip_deleted(&task) {+ self.task_lock_counter.release();+ continue;+ } if !task.flags.is_restored(category) {- panic_missing_task(task_id, reason);+ if access == TaskAccess::MustExist {+ panic_missing_task(task_id, reason);+ }+ handle_missing_task(task, task_id, access, reason);+ self.task_lock_counter.release();+ continue; } self.task_lock_counter.release();- prepared_task_callback(self, task_id, category, task);+ prepared_task_callback(self, task_id, category, task.into_write_guard()); } return; }@@ -435,14 +462,23 @@ impl<'e> ExecuteContext<'e> { .into_iter() .filter(|&(task_id, category)| { self.task_lock_counter.acquire();- let task = self.backend.storage.access_mut(task_id);+ let task = self.backend.storage.access_entry_mut(task_id);+ if skip_deleted(&task) {+ self.task_lock_counter.release();+ return false;+ } // Transient tasks are restored from creation and never evicted. if task_id.is_transient() || task.flags.is_restored(category) { if !task.flags.is_restored(category) {- panic_missing_task(task_id, reason);+ if access == TaskAccess::MustExist {+ panic_missing_task(task_id, reason);+ }+ handle_missing_task(task, task_id, access, reason);+ self.task_lock_counter.release();+ return false; } self.task_lock_counter.release();- prepared_task_callback(self, task_id, category, task);+ prepared_task_callback(self, task_id, category, task.into_write_guard()); return false; } self.task_lock_counter.release();@@ -508,7 +544,7 @@ impl<'e> ExecuteContext<'e> { } self.task_lock_counter.release(); // A peer may have completed the restore since filtering.- if ready {+ if ready && !skip_deleted(&task) { prepared_task_callback(self, task_id, category, task); } }@@ -621,7 +657,9 @@ impl<'e> ExecuteContext<'e> { task.flags.set_meta_restoring(false); } entry.failed = true;- missing_tasks.push(task_id);+ if access == TaskAccess::MustExist {+ missing_tasks.push(task_id);+ } // Discards the entry, or clears its restored flags if another thread still holds // it. handle_missing_task(task, task_id, TaskAccess::AllowMissing, reason);@@ -689,13 +727,22 @@ impl<'e> ExecuteContext<'e> { if !entry.wait_data && !entry.wait_meta { // Phase 1c already checked existence. This restores the category again if // eviction won the handoff.- let task = match self.restore_existing_task(entry.task_id, entry.category) {- Ok((task, _)) => task,+ let task = match self.restore_existing_task(entry.task_id, entry.category, access) {+ Ok((task, outcome)) => {+ if access == TaskAccess::AllowMissing && outcome.missing_on_disk {+ handle_missing_task(task, entry.task_id, access, reason);+ continue;+ }+ task+ } Err(e) => { restore_errors.push((entry.task_id, category_name(entry.category), e)); continue; } };+ if skip_deleted(&task) {+ continue;+ } prepared_task_callback( self, entry.task_id,@@ -715,7 +762,7 @@ impl<'e> ExecuteContext<'e> { } self.task_lock_counter.acquire(); let (task, outcome) =- match self.restore_existing_task(entry.task_id, entry.category) {+ match self.restore_existing_task(entry.task_id, entry.category, access) { Ok(restored) => restored, Err(e) => { restore_errors.push((entry.task_id, category_name(entry.category), e));@@ -725,12 +772,17 @@ impl<'e> ExecuteContext<'e> { }; if outcome.missing_on_disk { // Keep handing off the remaining tasks; report the failure at the end.- missing_tasks.push(entry.task_id);+ if access == TaskAccess::MustExist {+ missing_tasks.push(entry.task_id);+ } handle_missing_task(task, entry.task_id, TaskAccess::AllowMissing, reason); self.task_lock_counter.release(); continue; } self.task_lock_counter.release();+ if skip_deleted(&task) {+ continue;+ } prepared_task_callback( self, entry.task_id,@@ -938,7 +990,7 @@ impl<'e> ExecuteContext<'e> { task_ids: impl IntoIterator<Item = (TaskId, TaskDataCategory)>, reason: &'static str, ) {- self.prepare_tasks_with_callback(task_ids, reason, |_, _, _, _| {});+ self.prepare_tasks_with_callback(task_ids, TaskAccess::MustExist, reason, |_, _, _, _| {}); } /// Opens each task like [`Self::task`] (so every task must already exist), batching the reads@@ -947,18 +999,53 @@ impl<'e> ExecuteContext<'e> { &mut self, task_ids: impl IntoIterator<Item = (TaskId, TaskDataCategory)>, reason: &'static str,+ func: impl FnMut(TaskGuard<'e>, &mut Self),+ ) {+ self.for_each_task_with_access(task_ids, TaskAccess::MustExist, reason, func);+ }++ /// Opens each task like [`Self::try_task`] with [`TaskDataCategory::All`], batching the reads+ /// from persistent storage. Tasks that `try_task` would not return (soft-deleted, or missing+ /// in memory and on disk) are skipped: `func` is not called for them. Persisted tasks that are+ /// not resident are restored and handed to `func`.+ ///+ /// Like [`Self::prepare_tasks`], the ids should not contain duplicates.+ pub fn for_each_try_task_all(+ &mut self,+ task_ids: impl IntoIterator<Item = TaskId>,+ reason: &'static str,+ func: impl FnMut(TaskGuard<'e>, &mut Self),+ ) {+ self.for_each_task_with_access(+ task_ids.into_iter().map(|id| (id, TaskDataCategory::All)),+ TaskAccess::AllowMissing,+ reason,+ func,+ );+ }++ fn for_each_task_with_access(+ &mut self,+ task_ids: impl IntoIterator<Item = (TaskId, TaskDataCategory)>,+ access: TaskAccess,+ reason: &'static str, mut func: impl FnMut(TaskGuard<'e>, &mut Self), ) { let task_lock_counter = self.task_lock_counter.clone();- self.prepare_tasks_with_callback(task_ids, reason, |this, task_id, _category, task| {- // prepare_tasks_with_callback releases the counter before calling this callback,- // so the counter is 0 here. Acquire for the TaskGuard that will release on- // Drop.- task_lock_counter.acquire();-- let guard = TaskGuard::new(task, task_id, _category, task_lock_counter.clone());- func(guard, this);- });+ self.prepare_tasks_with_callback(+ task_ids,+ access,+ reason,+ |this, task_id, category, task| {+ // prepare_tasks_with_callback releases the counter before calling this callback,+ // so the counter is 0 here. Acquire for the TaskGuard that will release on+ // Drop.+ task_lock_counter.acquire();++ let guard = TaskGuard::new(task, task_id, category, task_lock_counter.clone());+ func(guard, this);+ },+ ); } pub fn for_each_task_meta(@@ -1883,6 +1970,15 @@ mod must_exist_tests { } fn persisted_tasks(ids: &[TaskId]) -> (Arc<TurboTasks<TurboTasksBackend>>, tempfile::TempDir) {+ persisted_tasks_with(ids, &[])+ }++ /// A fresh backend over a store holding both categories for `ids` and only the data category+ /// for `data_only_ids`. None of the tasks is resident.+ fn persisted_tasks_with(+ ids: &[TaskId],+ data_only_ids: &[TaskId],+ ) -> (Arc<TurboTasks<TurboTasksBackend>>, tempfile::TempDir) { use turbo_bincode::{TurboBincodeBuffer, new_turbo_bincode_encoder}; use crate::{BackingStorageOptions, GitVersionInfo, backing_storage::SnapshotItem};@@ -1918,10 +2014,12 @@ mod must_exist_tests { .unwrap(); let items = ids .iter()- .map(|&task_id| SnapshotItem::Put {+ .map(|&task_id| (task_id, true))+ .chain(data_only_ids.iter().map(|&task_id| (task_id, false)))+ .map(|(task_id, with_meta)| SnapshotItem::Put { task_id, data: Some(encoded_data.clone()),- meta: Some(encoded_meta.clone()),+ meta: with_meta.then(|| encoded_meta.clone()), task_type_hash: None, #[cfg(feature = "print_cache_item_size")] stats: Default::default(),@@ -2104,6 +2202,280 @@ mod must_exist_tests { assert!(!tt.backend().storage.access_mut(id).flags.meta_restored()); } }++ /// Runs [`ExecuteContext::for_each_try_task_all`] and returns the ids handed to the callback.+ fn try_tasks(tt: &TurboTasks<TurboTasksBackend>, ids: &[TaskId]) -> Vec<TaskId> {+ let mut ctx = ExecuteContext::new(tt.backend(), tt);+ let mut seen = Vec::new();+ ctx.for_each_try_task_all(ids.iter().copied(), "try tasks", |guard, _| {+ seen.push(guard.id())+ });+ seen+ }++ fn has_entry(tt: &TurboTasks<TurboTasksBackend>, id: TaskId) -> bool {+ tt.backend().storage.with_task(id, |_| ()).is_some()+ }++ fn assert_unpinned(tt: &TurboTasks<TurboTasksBackend>, id: TaskId) {+ assert_eq!(+ tt.backend()+ .storage+ .with_task(id, |t| t.gc_transient_ref_count()),+ Some(0)+ );+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_resident_handed_off() {+ let tt = backend();+ let present = persistent(1);+ let transient = TaskId::new(TRANSIENT_TASK_BIT | 2).unwrap();+ resident(&tt, present, TaskDataCategory::All);+ resident(&tt, transient, TaskDataCategory::All);+ restore_lookup_counts::reset();+ assert_eq!(try_tasks(&tt, &[present, transient]), [present, transient]);+ assert_eq!(restore_lookup_counts::get(), (0, 0));+ assert_unpinned(&tt, present);+ assert_unpinned(&tt, transient);+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_paged_out_restored_with_batched_reads() {+ let ids = [persistent(1), persistent(2), persistent(3)];+ let (tt, _dir) = persisted_tasks(&ids);+ restore_lookup_counts::reset();+ assert_eq!(try_tasks(&tt, &ids), ids);+ // One batched read per category for the whole batch, no per-task reads.+ assert_eq!(restore_lookup_counts::get(), (0, 2));+ for id in ids {+ let task = tt.backend().storage.access_mut(id);+ assert!(task.flags.is_restored(TaskDataCategory::All));+ assert_eq!(task.gc_transient_ref_count(), 0);+ }+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_missing_persistent_skipped() {+ let present = persistent(1);+ let missing = persistent(2);+ let (tt, _dir) = persisted_tasks(&[present]);+ assert_eq!(try_tasks(&tt, &[missing]), []);+ assert!(!has_entry(&tt, missing));+ // Also with the noop storage, where nothing exists on disk.+ let tt = backend();+ assert_eq!(try_tasks(&tt, &[missing]), []);+ assert!(!has_entry(&tt, missing));+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_missing_transient_skipped() {+ let tt = backend();+ let missing = TaskId::new(TRANSIENT_TASK_BIT | 2).unwrap();+ restore_lookup_counts::reset();+ assert_eq!(try_tasks(&tt, &[missing]), []);+ // Transient tasks are never looked up on disk.+ assert_eq!(restore_lookup_counts::get(), (0, 0));+ assert!(!has_entry(&tt, missing));+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_deleted_skipped() {+ let tt = backend();+ let deleted = persistent(1);+ resident(&tt, deleted, TaskDataCategory::All);+ tt.backend()+ .storage+ .access_mut(deleted)+ .flags+ .set_deleted(true);+ assert_eq!(try_tasks(&tt, &[deleted]), []);+ // Like `try_task`, a soft-deleted task is left in place.+ assert!(has_entry(&tt, deleted));+ let mut ctx = ExecuteContext::new(tt.backend(), &tt);+ assert!(ctx.try_task(deleted, TaskDataCategory::All).is_none());+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_partial_category_skipped() {+ let present = persistent(1);+ let data_only = persistent(2);+ let (tt, _dir) = persisted_tasks_with(&[present], &[data_only]);+ assert_eq!(try_tasks(&tt, &[data_only, present]), [present]);+ assert!(!has_entry(&tt, data_only));+ // Matches `try_task`.+ let mut ctx = ExecuteContext::new(tt.backend(), &tt);+ assert!(ctx.try_task(data_only, TaskDataCategory::All).is_none());+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_no_backing_storage_skips_nonresident() {+ let tt = TurboTasks::new(TurboTasksBackend::new(+ BackendOptions {+ storage_mode: None,+ num_workers: Some(1),+ small_preallocation: true,+ ..Default::default()+ },+ noop_backing_storage(),+ ));+ let present = persistent(1);+ let missing = persistent(2);+ resident(&tt, present, TaskDataCategory::All);+ assert_eq!(try_tasks(&tt, &[missing, present]), [present]);+ assert!(!has_entry(&tt, missing));+ assert_unpinned(&tt, present);+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_missing_task_does_not_wait_for_peer_restore() {+ use std::{sync::mpsc, time::Duration};++ let tt = backend();+ let id = persistent(4);+ tt.backend()+ .storage+ .access_mut(id)+ .flags+ .set_meta_restoring(true);+ let (sender, receiver) = mpsc::sync_channel(1);+ let worker_tt = tt.clone();+ let worker = std::thread::spawn(move || {+ sender.send(try_tasks(&worker_tt, &[id])).unwrap();+ });++ // Whether it timed out or not, release the synthetic peer before joining the worker.+ let completed_without_peer = receiver.recv_timeout(Duration::from_secs(2));+ tt.backend()+ .storage+ .access_mut(id)+ .flags+ .set_meta_restoring(false);+ tt.backend().storage.restored.notify(usize::MAX);+ worker.join().unwrap();+ assert_eq!(+ completed_without_peer.expect("missing key must be skipped without waiting for Meta"),+ []+ );+ let task = tt.backend().storage.access_mut(id);+ assert!(!task.flags.data_restoring() && !task.flags.meta_restoring());+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_mixed_batch() {+ let on_disk = persistent(1);+ let resident_task = persistent(2);+ let missing = persistent(3);+ let deleted = persistent(4);+ let missing_transient = TaskId::new(TRANSIENT_TASK_BIT | 5).unwrap();+ let (tt, _dir) = persisted_tasks(&[on_disk]);+ resident(&tt, resident_task, TaskDataCategory::All);+ resident(&tt, deleted, TaskDataCategory::All);+ tt.backend()+ .storage+ .access_mut(deleted)+ .flags+ .set_deleted(true);+ let mut seen = try_tasks(+ &tt,+ &[missing, on_disk, resident_task, missing_transient, deleted],+ );+ seen.sort();+ assert_eq!(seen, [on_disk, resident_task]);+ assert!(!has_entry(&tt, missing));+ assert!(!has_entry(&tt, missing_transient));+ assert_unpinned(&tt, on_disk);+ assert_unpinned(&tt, resident_task);+ }++ /// Claims both restores of `id` as a synthetic peer, runs [`try_tasks`] for `id` on another+ /// thread, checks it blocks on the peer (Phase 3), then lets the peer give up without+ /// restoring. Returns what the worker handed off.+ fn try_task_after_peer_gives_up(+ tt: &Arc<TurboTasks<TurboTasksBackend>>,+ id: TaskId,+ ) -> Vec<TaskId> {+ use std::{+ sync::mpsc::{self, RecvTimeoutError},+ time::Duration,+ };++ let set_peer_claims = |claimed: bool| {+ let mut task = tt.backend().storage.access_mut(id);+ task.flags.set_data_restoring(claimed);+ task.flags.set_meta_restoring(claimed);+ };+ set_peer_claims(true);+ let (sender, receiver) = mpsc::sync_channel(1);+ let worker_tt = tt.clone();+ let worker = std::thread::spawn(move || {+ sender.send(try_tasks(&worker_tt, &[id])).unwrap();+ });+ let blocked = receiver.recv_timeout(Duration::from_millis(200));+ // Release the synthetic peer before asserting, so the worker can always be joined.+ set_peer_claims(false);+ tt.backend().storage.restored.notify(usize::MAX);+ assert!(+ matches!(blocked, Err(RecvTimeoutError::Timeout)),+ "the worker must wait for the peer's restore"+ );+ let seen = receiver+ .recv_timeout(Duration::from_secs(10))+ .expect("the worker must finish once the peer is done");+ worker.join().unwrap();+ seen+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_waits_for_peer_then_restores() {+ let id = persistent(1);+ let (tt, _dir) = persisted_tasks(&[id]);+ assert_eq!(try_task_after_peer_gives_up(&tt, id), [id]);+ let task = tt.backend().storage.access_mut(id);+ assert!(task.flags.is_restored(TaskDataCategory::All));+ assert!(!task.flags.data_restoring() && !task.flags.meta_restoring());+ }++ #[tokio::test(flavor = "multi_thread")]+ async fn try_tasks_waits_for_peer_then_skips_missing() {+ let present = persistent(1);+ let missing = persistent(2);+ let (tt, _dir) = persisted_tasks(&[present]);+ assert_eq!(try_task_after_peer_gives_up(&tt, missing), []);+ assert!(!has_entry(&tt, missing));+ }+}++/// Counts the persistent storage lookups made by restores on the current thread, so tests can+/// check that a batch of tasks is restored with batched reads.+#[cfg(test)]+mod restore_lookup_counts {+ use std::cell::Cell;++ thread_local! {+ /// (single-task lookups, batched lookups)+ static COUNTS: Cell<(usize, usize)> = const { Cell::new((0, 0)) };+ }++ pub(super) fn record(batch: bool) {+ COUNTS.with(|counts| {+ let (single, batched) = counts.get();+ counts.set(if batch {+ (single, batched + 1)+ } else {+ (single + 1, batched)+ });+ });+ }++ pub(super) fn reset() {+ COUNTS.with(|counts| counts.set((0, 0)));+ }++ /// (single-task lookups, batched lookups) since the last [`reset`].+ pub(super) fn get() -> (usize, usize) {+ COUNTS.with(Cell::get)+ } } #[cfg(test)]turbopack/crates/turbo-tasks-backend/tests/invalidate_tasks.rsadded65 + / 0 −
@@ -0,0 +1,65 @@+#![feature(arbitrary_self_types)]+#![allow(clippy::needless_return)] // tokio macro-generated code doesn't respect this++mod util;++use std::sync::atomic::{AtomicU32, Ordering};++use turbo_tasks::{TRANSIENT_TASK_BIT, TaskId, Vc, backend::Backend};+use turbo_tasks_backend::TestSnapshotOutcome;++use crate::util::create_tt;++static EXECUTIONS: [AtomicU32; 2] = [AtomicU32::new(0), AtomicU32::new(0)];++#[turbo_tasks::function(operation, root)]+fn leaf(index: u32) -> Vc<u32> {+ // Tasks without an invalidator or dependencies are immutable and must not be invalidated.+ let _ = turbo_tasks::get_invalidator();+ let executions = EXECUTIONS[index as usize].fetch_add(1, Ordering::SeqCst) + 1;+ Vc::cell(executions)+}++/// `Backend::invalidate_tasks` with a batch containing ids of tasks that don't exist (persistent+/// and transient) invalidates the existing tasks once each and skips the rest.+///+/// Restoring non-resident tasks in a batch is covered by the `try_tasks_*` unit tests in+/// `backend/operation/mod.rs`; eviction here is best-effort, as tasks held by this session may+/// stay resident.+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]+async fn invalidate_tasks_mixed_batch() {+ let (tt, _persistence_dir) = create_tt("invalidate_tasks_mixed_batch");+ let tt2 = tt.clone();++ let result = turbo_tasks::run_once(tt.clone(), async move {+ let leaves = [leaf(0), leaf(1)];+ for leaf in leaves {+ assert_eq!(*leaf.read_strongly_consistent().await?, 1);+ }++ // Persist the tasks, so the missing persistent id is looked up in a real store, and evict+ // what can be evicted.+ let TestSnapshotOutcome { had_new_data, .. } =+ tt2.backend().snapshot_and_evict_for_testing(&tt2);+ assert!(had_new_data, "snapshot should have persisted data");++ let [a, b] = leaves.map(|leaf| leaf.task_id());+ let missing_persistent = TaskId::new(0x00ff_ffff).unwrap();+ let missing_transient = TaskId::new(TRANSIENT_TASK_BIT | 0x00ff_ffff).unwrap();+ tt2.backend()+ .invalidate_tasks(&[missing_persistent, a, b, missing_transient], &tt2);++ for leaf in leaves {+ assert_eq!(+ *leaf.read_strongly_consistent().await?,+ 2,+ "every existing task re-executes exactly once"+ );+ }++ anyhow::Ok(())+ })+ .await;+ tt.stop_and_wait().await;+ result.unwrap();+}turbopack/crates/turbo-tasks/src/backend.rs1 + / 0 −
@@ -608,6 +608,7 @@ pub trait Backend: Sized + Sync + Send { fn invalidate_task(&self, task: TaskId, turbo_tasks: &TurboTasks<Self>); + /// Invalidates all `tasks`. `tasks` must not contain duplicates. fn invalidate_tasks(&self, tasks: &[TaskId], turbo_tasks: &TurboTasks<Self>); fn invalidate_tasks_set(&self, tasks: &TaskIdSet, turbo_tasks: &TurboTasks<Self>);