diff --git a/packages/next/src/build/index.ts b/packages/next/src/build/index.ts index b18ee907abde..7fbc77f2df5a 100644 --- a/packages/next/src/build/index.ts +++ b/packages/next/src/build/index.ts @@ -1208,7 +1208,7 @@ export default async function build( if (experimentalBuildMode === 'generate-env') { if (bundler === Bundler.Turbopack) { Log.warn('generate-env is not needed with turbopack') - process.exit(0) + return } Log.info('Inlining static env ...') await nextBuildSpan @@ -1221,9 +1221,7 @@ export default async function build( }) Log.info('Complete') - flushAllTraces() - teardownTraceSubscriber() - process.exit(0) + return } // when using compile mode static env isn't inlined so we @@ -1793,11 +1791,7 @@ export default async function build( shutdownPromise: p, warnings, ...rest - } = await turbopackBuild( - process.env.NEXT_TURBOPACK_USE_WORKER === undefined || - process.env.NEXT_TURBOPACK_USE_WORKER !== '0', - telemetry - ) + } = await turbopackBuild(telemetry) shutdownPromise = p deferredTurbopackWarnings = warnings traceMemoryUsage('Finished build', nextBuildSpan) diff --git a/packages/next/src/build/turbopack-build/impl.ts b/packages/next/src/build/turbopack-build/impl.ts index bb106175be5d..a6aee31e8d61 100644 --- a/packages/next/src/build/turbopack-build/impl.ts +++ b/packages/next/src/build/turbopack-build/impl.ts @@ -1,11 +1,8 @@ -// Import cpu-profile first to start profiling early if enabled -import { saveCpuProfile } from '../../server/lib/cpu-profile' import path from 'path' import { validateTurboNextConfig } from '../../lib/turbopack-warning' import { seedTurbopackCacheIfNeeded } from '../../lib/turbopack-cache-seed' import { NextBuildContext } from '../build-context' import { createDefineEnv, getBindingsSync } from '../swc' -import { installBindings } from '../swc/install-bindings' import { handleRouteType, rawEntrypointsToEntrypoints, @@ -14,16 +11,8 @@ import { TurbopackManifestLoader } from '../../shared/lib/turbopack/manifest-loa import { promises as fs } from 'fs' import { PHASE_PRODUCTION_BUILD } from '../../shared/lib/constants' import loadConfig from '../../server/config' -import { hasCustomExportOutput } from '../../export/utils' -import { Telemetry } from '../../telemetry/storage' +import type { Telemetry } from '../../telemetry/storage' import { eventBuildFeatureUsageFromTurbopack } from '../../telemetry/events/build' -import { - setGlobal, - trace, - initializeTraceState, - getTraceEvents, -} from '../../trace' -import type { TraceState } from '../../trace' import { isCI } from '../../server/ci-info' import { backgroundLogCompilationEvents } from '../../shared/lib/turbopack/compilation-events' import { getSupportedBrowsers } from '../get-supported-browsers' @@ -157,15 +146,23 @@ export async function turbopackBuild(telemetry: Telemetry): Promise<{ } : undefined ) - const buildEventsSpan = trace('turbopack-build-events') - // Stop immediately: this span is only used as a parent for - // manualTraceChild calls which carry their own timestamps. - buildEventsSpan.stop() const shutdownController = new AbortController() const compilationEvents = backgroundLogCompilationEvents(project, { - parentSpan: buildEventsSpan, + // Compilation events carry their own timestamps, so they hang directly off + // the build rather than a synthetic grouping span. + parentSpan: NextBuildContext.nextBuildSpan, signal: shutdownController.signal, }) + const runShutdown = async () => { + // Shutdown may trigger final compilation events (e.g. persistence, + // compaction trace spans). This is the last chance to capture them. + // After shutdown resolves we abort the signal to close the iterator + // and drain any remaining buffered events. + + await project.shutdown() + shutdownController.abort() + await compilationEvents + } try { // Write an empty file in a known location to signal this was built with Turbopack @@ -295,95 +292,15 @@ export async function turbopackBuild(telemetry: Telemetry): Promise<{ await project.writeAnalyzeData(appDirOnly) } - // Shutdown may trigger final compilation events (e.g. persistence, - // compaction trace spans). This is the last chance to capture them. - // After shutdown resolves we abort the signal to close the iterator - // and drain any remaining buffered events. - const shutdownPromise = project.shutdown().then(() => { - shutdownController.abort() - return compilationEvents.catch(() => {}) - }) - const time = process.hrtime(startTime) return { duration: time[0] + time[1] / 1e9, buildTraceContext: undefined, - shutdownPromise, + shutdownPromise: runShutdown(), warnings, } } catch (err) { - await project.shutdown() - shutdownController.abort() - await compilationEvents.catch(() => {}) + await runShutdown() throw err } } - -let shutdownPromise: Promise | undefined -export async function workerMain(workerData: { - buildContext: typeof NextBuildContext - traceState: TraceState & { shouldSaveTraceEvents: boolean } -}): Promise< - Omit>, 'shutdownPromise'> -> { - // setup new build context from the serialized data passed from the parent - Object.assign(NextBuildContext, workerData.buildContext) - initializeTraceState(workerData.traceState) - - /// load the config because it's not serializable - const config = await loadConfig( - PHASE_PRODUCTION_BUILD, - NextBuildContext.dir!, - { - debugPrerender: NextBuildContext.debugPrerender, - reactProductionProfiling: NextBuildContext.reactProductionProfiling, - bundler: Bundler.Turbopack, - } - ) - NextBuildContext.config = config - // Matches handling in build/index.ts - // https://github.com/vercel/next.js/blob/84f347fc86f4efc4ec9f13615c215e4b9fb6f8f0/packages/next/src/build/index.ts#L815-L818 - // Ensures the `config.distDir` option is matched. - if (hasCustomExportOutput(NextBuildContext.config)) { - NextBuildContext.config.distDir = '.next' - } - - // Clone the telemetry for worker - const telemetry = new Telemetry({ - distDir: NextBuildContext.config.distDir, - }) - setGlobal('telemetry', telemetry) - // Install bindings early so we can access synchronously later - await installBindings(config.experimental?.useWasmBinary) - - try { - const { - shutdownPromise: resultShutdownPromise, - buildTraceContext, - duration, - warnings, - } = await turbopackBuild(telemetry) - shutdownPromise = resultShutdownPromise - return { - buildTraceContext, - duration, - warnings, - } - } finally { - // Always flush telemetry before worker exits (waits for async operations like setTimeout in debug mode) - await telemetry.flush() - // Save CPU profile before worker exits - await saveCpuProfile() - } -} - -export async function waitForShutdown(): Promise<{ - debugTraceEvents?: ReturnType -}> { - if (shutdownPromise) { - await shutdownPromise - } - // Collect trace events after shutdown completes so that all compilation - // events (e.g. persistence trace spans) have been processed. - return { debugTraceEvents: getTraceEvents() } -} diff --git a/packages/next/src/build/turbopack-build/index.ts b/packages/next/src/build/turbopack-build/index.ts index 063528458d7a..a1e073164f17 100644 --- a/packages/next/src/build/turbopack-build/index.ts +++ b/packages/next/src/build/turbopack-build/index.ts @@ -1,95 +1,12 @@ -import path from 'path' - -import { Worker } from '../../lib/worker' import { NextBuildContext } from '../build-context' -import { exportTraceState, recordTraceEvents } from '../../trace' import type { Telemetry } from '../../telemetry/storage' - -async function turbopackBuildWithWorker(): ReturnType< - typeof import('./impl').turbopackBuild -> { - const nextBuildSpan = NextBuildContext.nextBuildSpan! - try { - const worker = new Worker(path.join(__dirname, 'impl.js'), { - exposedMethods: ['workerMain', 'waitForShutdown'], - enableWorkerThreads: true, - debuggerPortOffset: -1, - isolatedMemory: false, - numWorkers: 1, - maxRetries: 0, - forkOptions: { - env: { - NEXT_PRIVATE_BUILD_WORKER: '1', - ...(process.env.NEXT_CPU_PROF - ? { - NEXT_CPU_PROF: '1', - NEXT_CPU_PROF_DIR: process.env.NEXT_CPU_PROF_DIR, - __NEXT_PRIVATE_CPU_PROFILE: 'build-turbopack', - } - : undefined), - }, - }, - }) as Worker & typeof import('./impl') - const { - nextBuildSpan: _nextBuildSpan, - // Config is not serializable and is loaded in the worker. - config: _config, - ...prunedBuildContext - } = NextBuildContext - const { buildTraceContext, duration, warnings } = await worker.workerMain({ - buildContext: prunedBuildContext, - traceState: { - ...exportTraceState(), - defaultParentSpanId: nextBuildSpan.getId(), - shouldSaveTraceEvents: true, - }, - }) - - return { - // destroy worker when Turbopack has shutdown so it's not sticking around using memory - // We need to wait for shutdown to make sure filesystem cache is flushed - shutdownPromise: worker.waitForShutdown().then(({ debugTraceEvents }) => { - if (debugTraceEvents) { - recordTraceEvents(debugTraceEvents) - } - worker.end() - }), - buildTraceContext, - duration, - warnings, - } - } catch (err: any) { - // When the error is a serialized `Error` object we need to recreate the `Error` instance - // in order to keep the consistent error reporting behavior. - if (err.type === 'Error') { - const error = new Error(err.message) - if (err.name) { - error.name = err.name - } - if (err.cause) { - error.cause = err.cause - } - error.message = err.message - error.stack = err.stack - throw error - } - throw err - } -} +import { turbopackBuild as turbopackBuildImpl } from './impl' export function turbopackBuild( - withWorker: boolean, telemetry: Telemetry -): ReturnType { +): ReturnType { const nextBuildSpan = NextBuildContext.nextBuildSpan! - return nextBuildSpan.traceChild('run-turbopack').traceAsyncFn(async () => { - if (withWorker) { - // Worker creates its own Telemetry instance; no need to forward. - return await turbopackBuildWithWorker() - } else { - const build = (require('./impl') as typeof import('./impl')) - .turbopackBuild - return await build(telemetry) - } - }) + return nextBuildSpan + .traceChild('run-turbopack') + .traceAsyncFn(() => turbopackBuildImpl(telemetry)) } diff --git a/packages/next/src/shared/lib/turbopack/compilation-events.ts b/packages/next/src/shared/lib/turbopack/compilation-events.ts index 8dbbddec6ece..c108ca79159a 100644 --- a/packages/next/src/shared/lib/turbopack/compilation-events.ts +++ b/packages/next/src/shared/lib/turbopack/compilation-events.ts @@ -30,6 +30,10 @@ export function backgroundLogCompilationEvents( ): Promise { const iterator = project.compilationEventsSubscribe(eventTypes) + // If there is no signal assume there will be no clean shutdown, + // to ensure trace spans aren't lost just flush after each one. + const flushEachTraceEvent = signal === undefined + // Close the iterator as soon as the signal fires so the for-await loop // exits without waiting for the next compilation event. signal?.addEventListener('abort', () => iterator.return?.(undefined as any), { @@ -49,11 +53,9 @@ export function backgroundLogCompilationEvents( Object.fromEntries(data.attributes ?? []) ) traceMemoryUsage(data.name, parentSpan) - // We flush after each event to make sure it makes it to disk. These events are rare and - // tend to happen at the very end of a build so to make sure they are logged we need to - // flush. - // NOTE: in a `next build` environment where we are reporting events to the parent thread, this is a no-op. - flushAllTraces() + if (flushEachTraceEvent) { + flushAllTraces() + } } catch {} continue // don't log these events, they just go to the trace file } @@ -83,6 +85,5 @@ export function backgroundLogCompilationEvents( } })() // Prevent unhandled rejection if the subscription errors after the project shuts down. - promise.catch(() => {}) - return promise + return promise.catch(() => {}) } diff --git a/packages/next/src/trace/report/to-json-build.ts b/packages/next/src/trace/report/to-json-build.ts index dfa16b0cc6c6..a80d9fb3dbe7 100644 --- a/packages/next/src/trace/report/to-json-build.ts +++ b/packages/next/src/trace/report/to-json-build.ts @@ -15,7 +15,6 @@ const allowlistedEvents = new Set([ 'adapter-handle-build-complete', 'output-standalone', 'telemetry-flush', - 'turbopack-build-events', 'turbopack-persistence', 'turbopack-compaction', ]) diff --git a/test/e2e/app-dir/trace-build-file/trace-build-file.test.ts b/test/e2e/app-dir/trace-build-file/trace-build-file.test.ts index 6edc50892e56..806f2da3c004 100644 --- a/test/e2e/app-dir/trace-build-file/trace-build-file.test.ts +++ b/test/e2e/app-dir/trace-build-file/trace-build-file.test.ts @@ -78,6 +78,9 @@ describe('trace-build-file', () => { } if (process.env.IS_TURBOPACK_TEST) { + // Compaction only runs when it is due, so it may or may not appear. + foundEvents.delete('turbopack-compaction') + expect([...foundEvents].sort()).toMatchInlineSnapshot(` [ "next-build", @@ -86,7 +89,6 @@ describe('trace-build-file', () => { "static-check", "static-generation", "telemetry-flush", - "turbopack-build-events", "turbopack-persistence", ] `) diff --git a/test/e2e/cpu-profiling/cpu-profiling-build.test.ts b/test/e2e/cpu-profiling/cpu-profiling-build.test.ts index 9b85c53c397d..d10e4672e035 100644 --- a/test/e2e/cpu-profiling/cpu-profiling-build.test.ts +++ b/test/e2e/cpu-profiling/cpu-profiling-build.test.ts @@ -44,11 +44,9 @@ describe('CPU Profiling - next build', () => { expect(cpuProfiles.some((f) => f.startsWith('build-main-'))).toBe(true) if (isTurbopack) { - // Turbopack mode generates: build-main, build-turbopack - expect(cpuProfiles.length).toBe(2) - expect(cpuProfiles.some((f) => f.startsWith('build-turbopack-'))).toBe( - true - ) + // Turbopack builds in the main build process, so `build-main` is the only + // profile. + expect(cpuProfiles.length).toBe(1) } else { // Webpack mode generates: build-main, build-webpack-client, build-webpack-server, build-webpack-edge-server expect(cpuProfiles.length).toBe(4) diff --git a/test/production/prerender-worker-threads/prerender-worker-threads.test.ts b/test/production/prerender-worker-threads/prerender-worker-threads.test.ts index 945b9598a319..fd4d4a82cb64 100644 --- a/test/production/prerender-worker-threads/prerender-worker-threads.test.ts +++ b/test/production/prerender-worker-threads/prerender-worker-threads.test.ts @@ -14,7 +14,7 @@ module.exports = {} // flag defaults to false (PR #9199) and why the static export worker was fixed to // respect it instead of hardcoding threads on (PR #25063). describe('prerender worker threads', () => { - const { next, isTurbopack } = nextTestSetup({ + const { next } = nextTestSetup({ files: __dirname, skipStart: true, // `bindings` is a direct dependency rather than one of the addon. The addon @@ -52,28 +52,21 @@ module.exports = { experimental: { workerThreads: true } } expect(exitCode).not.toBe(0) }) - // This documents a bug rather than intended behavior. `next build` runs Turbopack - // in a worker thread (`enableWorkerThreads: true` is hardcoded in - // packages/next/src/build/turbopack-build/index.ts) and that worker re-evaluates - // `next.config.js`, so requiring a non-context-aware addon from the config breaks - // the build even though `experimental.workerThreads` is off. It is the same class - // of failure PR #9199 and PR #25063 fixed, in a worker those PRs did not touch. + // Requiring the addon from `next.config.js` without an `isMainThread` guard is + // safe for both bundlers, because neither re-evaluates the config on a worker + // thread of the build process. Webpack's build worker is a forked child process, + // and Turbopack builds in the main process. // - // Webpack is unaffected, because its build worker is a forked child process. - // - // When Turbopack stops evaluating the config on a worker thread, this test will - // fail: drop the branch and assert the webpack outcome for both bundlers. - it('should currently fail under Turbopack when the config loads the addon', async () => { + // Turbopack used to run its build in a worker thread that re-evaluated the + // config, which broke this case even with `experimental.workerThreads` off -- + // the same class of failure PR #9199 and PR #25063 fixed, in a worker those PRs + // did not touch. + it('should prerender when the config loads the addon unguarded', async () => { await next.patchFile('next.config.js', UNGUARDED_CONFIG) const { exitCode, cliOutput } = await next.build() - if (isTurbopack) { - expect(cliOutput).toContain('Module did not self-register') - expect(exitCode).not.toBe(0) - } else { - expect(cliOutput).not.toContain('Module did not self-register') - expect(exitCode).toBe(0) - } + expect(cliOutput).not.toContain('Module did not self-register') + expect(exitCode).toBe(0) }) }) diff --git a/turbopack/crates/turbo-tasks-backend/benches/gc.rs b/turbopack/crates/turbo-tasks-backend/benches/gc.rs new file mode 100644 index 000000000000..ae06bae937e2 --- /dev/null +++ b/turbopack/crates/turbo-tasks-backend/benches/gc.rs @@ -0,0 +1,161 @@ +//! Benchmarks for the `parent_count` garbage-collection pass ([`TurboTasksBackend::gc_collect`], +//! driven here via the `gc_for_testing` hook). +//! +//! Gated behind `TURBOPACK_BENCH_GC` because the per-iteration setup is expensive. Run with: +//! +//! ```bash +//! TURBOPACK_BENCH_GC=1 cargo bench -p turbo-tasks-backend --bench mod -- gc +//! ``` + +use std::{sync::Arc, time::Duration}; + +use anyhow::Result; +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput}; +use tokio::runtime::Runtime; +use turbo_tasks::{ + ResolvedVc, State, TurboTasks, Vc, unmark_top_level_task_may_leak_eventually_consistent_state, +}; +use turbo_tasks_backend::{ + BackendOptions, BackingStorageOptions, EvictionMode, GitVersionInfo, StorageMode, + TurboTasksBackend, +}; + +fn enabled() -> bool { + !matches!( + std::env::var("TURBOPACK_BENCH_GC").ok().as_deref(), + None | Some("") | Some("no") | Some("false") + ) +} + +/// A persistent backend with GC-relevant options +fn create_tt() -> (Arc>, tempfile::TempDir) { + let parent = std::path::PathBuf::from(format!("{}/.cache", env!("CARGO_TARGET_TMPDIR"))); + std::fs::create_dir_all(&parent).unwrap(); + let dir = tempfile::Builder::new() + .prefix("gc-bench-") + .tempdir_in(&parent) + .unwrap(); + let tt = TurboTasks::new(TurboTasksBackend::new( + BackendOptions { + num_workers: None, + small_preallocation: false, + storage_mode: Some(StorageMode::ReadWriteOnShutdown), + eviction_mode: EvictionMode::Full, + gc: Some(true), + ..Default::default() + }, + turbo_tasks_backend::turbo_backing_storage( + dir.path(), + &GitVersionInfo { + describe: "bench-unversioned", + dirty: false, + }, + BackingStorageOptions { + is_ci: false, + is_short_session: true, + skip_compaction: true, + }, + ) + .unwrap() + .0, + )); + (tt, dir) +} + +#[turbo_tasks::value(transparent)] +struct Generation(State); + +#[turbo_tasks::function(operation, root)] +fn create_generation() -> Vc { + Generation(State::new(0)).cell() +} + +// WIDE shape: root -> `width` intermediates, each -> a leaf (2 tasks per index). Bumping the +// generation disconnects the whole previous generation, so a collect tears down ~2*width tasks and +// exercises the per-task fan-out plus a one-level cascade (intermediate -> its leaf). + +#[turbo_tasks::function] +fn wide_leaf(generation: u32, index: u32) -> Vc { + Vc::cell(generation.wrapping_mul(1_000_003).wrapping_add(index)) +} + +#[turbo_tasks::function] +async fn wide_intermediate(generation: u32, index: u32) -> Result> { + Ok(Vc::cell(1 + *wide_leaf(generation, index).await?)) +} + +#[turbo_tasks::function(operation, root)] +async fn wide_root(generation: ResolvedVc, width: u32) -> Result> { + let generation = *generation.await?.get(); + let mut sum = 0u32; + for index in 0..width { + sum = sum.wrapping_add(*wide_intermediate(generation, index).await?); + } + Ok(Vc::cell(sum)) +} + +/// Build generation 0 of the WIDE graph then bump to generation 1, leaving generation 0 fully +/// disconnected (garbage) and resident. Returns the backend ready for a timed collect. +fn setup_wide_garbage( + rt: &Runtime, + width: u32, +) -> (Arc>, tempfile::TempDir) { + rt.block_on(async move { + // `TurboTasks::new` builds a `PriorityRunner` that calls `Handle::current()`, so the + // backend must be constructed inside the runtime context. + let (tt, dir) = create_tt(); + turbo_tasks::run_once(tt.clone(), async move { + unmark_top_level_task_may_leak_eventually_consistent_state(); + let generation_op = create_generation(); + let generation_vc = generation_op.resolve().strongly_consistent().await?; + let generation = generation_op.read_strongly_consistent().await?; + wide_root(generation_vc, width) + .read_strongly_consistent() + .await?; + generation.set(1); + wide_root(generation_vc, width) + .read_strongly_consistent() + .await?; + anyhow::Ok(()) + }) + .await + .unwrap(); + (tt, dir) + }) +} + +pub fn gc(c: &mut Criterion) { + if !enabled() { + return; + } + + let mut group = c.benchmark_group("turbo_tasks_backend_gc"); + group.measurement_time(Duration::from_secs(20)); + group.sample_size(10); + + for width in [5_000u32, 25_000, 100_000] { + // ~2 tasks per index become garbage (intermediate + leaf). + let garbage = (2 * width) as u64; + group.throughput(Throughput::Elements(garbage)); + group.bench_with_input(BenchmarkId::new("wide", garbage), &width, |b, &width| { + // Must match `create_tt`'s `num_workers`. + let rt = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .worker_threads(std::thread::available_parallelism().map_or(4, |n| n.get())) + .build() + .unwrap(); + // Each collect consumes its garbage, so every iteration needs a freshly + // built+disconnected graph. Setup is not timed. + b.iter_batched( + || setup_wide_garbage(&rt, width), + |(tt, _dir)| { + let collected = rt.block_on(async { tt.backend().gc_for_testing(&tt) }); + (collected, tt, _dir) + }, + BatchSize::PerIteration, + ); + }); + } + + group.finish(); +} diff --git a/turbopack/crates/turbo-tasks-backend/benches/mod.rs b/turbopack/crates/turbo-tasks-backend/benches/mod.rs index 537e93e49478..2d90aaa7799f 100644 --- a/turbopack/crates/turbo-tasks-backend/benches/mod.rs +++ b/turbopack/crates/turbo-tasks-backend/benches/mod.rs @@ -3,6 +3,7 @@ use criterion::{Criterion, criterion_group, criterion_main}; +pub(crate) mod gc; pub(crate) mod overhead; pub(crate) mod scope_stress; pub(crate) mod stress; @@ -10,6 +11,6 @@ pub(crate) mod stress; criterion_group!( name = turbo_tasks_backend_stress; config = Criterion::default(); - targets = stress::fibonacci, scope_stress::scope_stress, overhead::overhead + targets = stress::fibonacci, scope_stress::scope_stress, overhead::overhead, gc::gc ); criterion_main!(turbo_tasks_backend_stress); diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/gc.rs b/turbopack/crates/turbo-tasks-backend/src/backend/gc.rs new file mode 100644 index 000000000000..2c549c82732d --- /dev/null +++ b/turbopack/crates/turbo-tasks-backend/src/backend/gc.rs @@ -0,0 +1,191 @@ +//! Garbage collection for the persistent backend. +//! +//! GC identifies and tears down tasks that have no reverse references using the `parent_count` and +//! `transient_ref_count`. Tasks are marked `deleted` and then have their outgoing edges teared down +//! recursively. +//! +//! A collected task also has its cell data released immediately to deliver immediate memory wins. +//! +//! The pass runs under the coordinator's exclusion phase (see +//! [`SnapshotCoordinator::begin_exclusion`](crate::backend::snapshot_coordinator)), which excludes +//! normal operations. That exclusion is what lets a pass edit the graph without racing a mutation +//! that could resurrect a task mid-collect, and hand its decisions straight to persistence: the +//! same guard stays held across the snapshot that writes the tombstones. +//! +//! TODO: find a way to collect GC roots that go away between sessions. Right now they are +//! persisted forever and if a later session doesn't read them it is never deleted. + +use std::{fmt::Display, ops::ControlFlow, sync::atomic::Ordering}; + +use turbo_tasks::{TaskId, TurboTasks, scope_unbounded::scope_unbounded_with}; + +use crate::backend::{ + AnyOperation, TurboTasksBackend, + operation::{ + AggregationUpdateQueue, CleanupOldEdgesOperation, ExecuteContext, ExecuteContextImpl, + TaskGuard, capture_all_outgoing_edges, + }, + snapshot_coordinator::SnapshotPhase, + storage::{SpecificTaskDataCategory, TaskDataCategory}, + storage_schema::TaskStorageAccessors, +}; + +/// One unit of GC work. +enum GcJob { + /// Scan one shard of the resident map (by index) and enqueue its candidates as + /// [`GcJob::Collect`]. + ScanShard(usize), + /// Collect a single task + Collect(TaskId), +} + +/// Observability counters for one [`TurboTasksBackend::gc_collect`] pass. +#[derive(Default)] +pub(crate) struct GcStats { + /// Tasks collected (marked soft-deleted). + pub collected: usize, + /// Edges torn down across all collected tasks (children + forward-dependency reverse edges). + pub edges_deleted: usize, +} + +impl Display for GcStats { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "collected: {collected}, edges_deleted: {edges_deleted}", + collected = self.collected, + edges_deleted = self.edges_deleted + ) + } +} + +impl GcStats { + fn merge(mut self, other: Self) -> Self { + self.collected += other.collected; + self.edges_deleted += other.edges_deleted; + self + } +} + +impl TurboTasksBackend { + /// Collect all garbage from the task-cache + /// + /// `phase` is the held exclusion; it is the caller's proof that no operation is running, which + /// is what makes it safe to mutate the graph here. + /// + /// Returns [`GcStats`] for the pass. + pub(crate) fn gc_collect( + &self, + turbo_tasks: &TurboTasks, + phase: &SnapshotPhase<'_, AnyOperation>, + ) -> GcStats { + // TODO(perf): recycle the task ids of collected tasks. + scope_unbounded_with( + (0..self.storage.shard_count()).map(GcJob::ScanShard), + GcStats::default, + |spawner, job, stats| { + let collector = |task_id| spawner.spawn(GcJob::Collect(task_id)); + let task_id = match job { + GcJob::ScanShard(index) => { + self.storage.gc_scan_shard(index, collector); + return ControlFlow::Continue(()); + } + GcJob::Collect(task_id) => task_id, + }; + let mut ctx = ExecuteContextImpl::new_for_gc(self, turbo_tasks, phase, &collector); + // `All` restores Data so the edge capture below can read the Data-category dep + // sets. + let mut task = ctx.task(task_id, TaskDataCategory::All); + // Recheck under the guard, and note that this is the **authoritative** check: + // the shard scan that produced this candidate only had Meta, so it could not see + // dependency edges (see `TaskStorage::gc_maybe_collectible`). With `All` open the + // same predicate is exact. A racing teardown can also add uppers/followers that + // temporarily remove collectibility; such a task is re-enqueued by a later pass. + if !task.is_gc_collectible() { + return ControlFlow::Continue(()); + } + + let old_edges = capture_all_outgoing_edges(&task); + // Clear `immutable` defensively so `resurrect_deleted` can mark the task dirty if + // it needs to + task.set_immutable(false); + // Drop the whole cell payload. This recovers most of the RAM while persistence + // writes the tombstone. + drop(task.take_cell_data()); + task.set_deleted(true); + if task.new_task() { + task.discard_modifications_for_gc_new_task(); + } else { + // Persisted ensure it is marked modified so the next snapshot tombstones it. + // It is almost certainly already marked modified, so this is mostly a no-op. + let _ = task.track_modification(SpecificTaskDataCategory::Meta, "gc_deleted"); + } + drop(task); // drop the lock so CleanupOldEdgesOperation can run + stats.collected += 1; + stats.edges_deleted += old_edges.len(); + CleanupOldEdgesOperation::run( + task_id, + old_edges, + AggregationUpdateQueue::new(), + &mut ctx, + ); + ControlFlow::Continue(()) + }, + GcStats::merge, + ) + } + + pub(super) fn pin_task_for_gc( + &self, + task: TaskId, + turbo_tasks: &TurboTasks, + ) { + self.gc_update_pin(task, 1, "pin_task_for_gc", turbo_tasks); + } + + pub(super) fn unpin_task_for_gc( + &self, + task: TaskId, + turbo_tasks: &TurboTasks, + ) { + self.gc_update_pin(task, -1, "unpin_task_for_gc", turbo_tasks); + } + + /// Applies `delta` to a task's `transient_ref_count` + fn gc_update_pin( + &self, + task: TaskId, + delta: i32, + op: &'static str, + turbo_tasks: &TurboTasks, + ) { + // Once stopping, GC bookkeeping is irrelevant. This also keeps handles finalized during + // shutdown (after the map is dropped) from underflowing the count. + if self.stopping.load(Ordering::Acquire) { + return; + } + let mut ctx = self.execute_context(turbo_tasks); + // Technically we only need to manipulate transient data so meta is overkill. But the task + // must be resident if we are adding a pin so this isn't wasteful + let mut task = ctx.task(task, TaskDataCategory::Meta); + task.assert_not_deleted(op); + task.update_and_get_transient_ref_count(delta); + } + + /// Runs a full GC pass under the GC phase and returns the number of tasks collected. + #[doc(hidden)] + pub fn gc_for_testing(&self, turbo_tasks: &TurboTasks) -> usize { + // A pass sets `deleted` flags, and the persist path only knows how to tombstone those when + // GC is enabled. Running a pass on a GC-disabled backend would leave soft-deleted tasks + // that persistence refuses to handle, so require the backend to be configured for GC + // (`BackendOptions::gc` or `TURBO_ENGINE_GC`) rather than silently diverging from + // production. + assert!( + self.gc_enabled, + "gc_for_testing requires a GC-enabled backend: set `BackendOptions::gc = Some(true)`" + ); + let _serialize = self.snapshot_in_progress.lock(); + let phase = self.snapshot_coord.begin_snapshot(); + self.gc_collect(turbo_tasks, &phase).collected + } +} diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs b/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs index f761b9354760..55e7e04dc444 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs @@ -1,6 +1,7 @@ mod cell_data; mod counter_map; mod eviction; +mod gc; mod operation; mod snapshot_coordinator; mod storage; @@ -108,6 +109,7 @@ fn compute_stale_priority(task: &impl TaskGuard) -> TaskPriority { .in_parent(task.is_dirty().unwrap_or(TaskPriority::leaf())) } +#[derive(PartialEq, Eq)] pub enum StorageMode { /// Queries the storage for cache entries that don't exist locally. ReadOnly, @@ -147,6 +149,10 @@ pub struct BackendOptions { /// This reclaims memory by clearing persisted data that can be re-loaded from disk on demand. /// This is an EXPERIMENTAL FEATURE under development pub eviction_mode: EvictionMode, + + /// Overrides whether the reference-counting GC runs for this backend. `None` (default) derives + /// it from the `TURBO_ENGINE_GC` env var; + pub gc: Option, } impl Default for BackendOptions { @@ -158,6 +164,7 @@ impl Default for BackendOptions { num_workers: None, small_preallocation: false, eviction_mode: EvictionMode::Off, + gc: None, } } } @@ -205,15 +212,18 @@ pub struct TurboTasksBackend { storage: Storage, - /// Coordinates the operation/snapshot interleaving protocol. See + /// Coordinates the operation/snapshot/GC interleaving protocol. See /// [`SnapshotCoordinator`] for details. snapshot_coord: SnapshotCoordinator, - /// Serializes calls to `snapshot_and_persist`. The coordinator's - /// `begin_snapshot` asserts that snapshots don't overlap; this mutex - /// enforces that contract for our two callers (background loop and + /// Serializes calls to `snapshot_and_persist` (and `gc_for_testing`). The coordinator's + /// `begin_exclusion` asserts that exclusive phases don't overlap; this mutex + /// enforces that contract for our callers (background loop and /// `stop_and_wait`). snapshot_in_progress: Mutex<()>, + /// Experimental feature to enable dead tasks to be deleted from storage and ram. + gc_enabled: bool, + stopping: AtomicBool, stopping_event: Event, idle_start_event: Event, @@ -248,8 +258,26 @@ impl TurboTasksBackend { let next_task_id = backing_storage .next_free_task_id() .expect("Failed to get task id"); + + let mut gc_enabled = options.gc.unwrap_or_else(|| { + std::env::var_os("TURBO_ENGINE_GC") + .is_some_and(|v| matches!(v.to_str(), Some("1" | "true" | "yes"))) + }); + if gc_enabled + && options.storage_mode == Some(StorageMode::ReadWrite) + && options.eviction_mode == EvictionMode::Off + { + eprintln!( + "warning: GC is enabled but eviction is disabled on a ReadWrite backend; GC would \ + leave collected tasks resident forever. Forcing GC off. Enable eviction \ + ('auto'/'full') to use GC in this mode." + ); + gc_enabled = false; + } + Self { options, + gc_enabled, start_time: Instant::now(), persisted_task_id_factory: IdFactoryWithReuse::new( next_task_id, @@ -331,6 +359,14 @@ impl TurboTasksBackend { (had_new_data, counts) } + /// The number of persistent (non-transient) tasks resident in the map. Test-only hook; see + /// [`Storage::resident_persistent_task_count_for_testing`] for why the metric excludes + /// transient tasks. + #[doc(hidden)] + pub fn resident_persistent_task_count_for_testing(&self) -> usize { + self.storage.resident_persistent_task_count_for_testing() + } + /// The persistent `parent_count` of a resident task (0 if absent or not resident). Test-only /// hook for verifying incremental refcount maintenance. #[doc(hidden)] @@ -349,9 +385,8 @@ impl TurboTasksBackend { } /// Opens `task` with the must-exist [`ExecuteContext::task`] and drops the guard. Test-only - /// hook to exercise the non-fabricating existence guarantee: this panics (debug builds) if - /// `task` exists in neither memory nor persistent storage (rather than fabricating a - /// blank). + /// hook to exercise the non-fabricating existence guarantee: this panics if `task` exists in + /// neither memory nor persistent storage (rather than fabricating a blank). #[doc(hidden)] pub fn assert_task_exists_for_testing( &self, @@ -516,6 +551,7 @@ impl TurboTasksBackend { }); let (mut task, mut reader_task) = lock_task_and_optional_reader(&mut ctx, task_id, need_reader_task); + task.assert_not_deleted("read_task_output"); fn listen_to_done_event( reader_description: Option, @@ -823,7 +859,7 @@ impl TurboTasksBackend { // done: true } it must have Output and would early return. let old = task.set_in_progress(in_progress_state); debug_assert!(old.is_none(), "InProgress already exists"); - ctx.schedule_task(task, TaskPriority::Recomputation); + ctx.schedule_task(&task, TaskPriority::Recomputation); Ok(ReadOutcome::Scheduled(listener)) } @@ -890,6 +926,7 @@ impl TurboTasksBackend { }); let (mut task, reader_task) = lock_task_and_optional_reader(&mut ctx, task_id, need_reader_task); + task.assert_not_deleted("read_task_cell"); let content = if final_read_hint { task.remove_cell_data(&cell, &get_value_type(cell.type_id()).persistence) @@ -973,7 +1010,7 @@ impl TurboTasksBackend { TaskExecutionReason::CellNotAvailable, EventDescription::new(|| task.get_task_desc_fn()), ); - ctx.schedule_task(task, TaskPriority::Recomputation); + ctx.schedule_task(&task, TaskPriority::Recomputation); Ok(ReadOutcome::Scheduled(listener)) } @@ -1023,18 +1060,33 @@ impl TurboTasksBackend { // request bit, suspended_operations) assumes only one snapshot runs at // a time. Held for the entire snapshot lifecycle. let _snapshot_in_progress = self.snapshot_in_progress.lock(); + + // One exclusion covers the GC pass and the snapshot that follows it, so the collected + // tasks' tombstones (derived from the `deleted` flag) ride this same commit and no + // operation can resurrect a collected task in between. let start = Instant::now(); // SystemTime for wall-clock timestamps in trace events (milliseconds // since epoch). Instant is monotonic but has no defined epoch, so it // can't be used for cross-process trace correlation. let wall_start = SystemTime::now(); + let mut snapshot_phase = self.snapshot_coord.begin_snapshot(); + let gc_elapsed = if self.gc_enabled { + let gc_span = tracing::info_span!( + parent: parent_span.clone(), + "gc", + stats = tracing::field::Empty, + edges_deleted = tracing::field::Empty, + ) + .entered(); + let stats = self.gc_collect(turbo_tasks, &snapshot_phase); + gc_span.record("stats", display(stats)); + Some(start.elapsed()) + } else { + None + }; + debug_assert!(self.should_persist()); - let mut snapshot_phase = { - let _span = tracing::info_span!("blocking").entered(); - self.snapshot_coord.begin_snapshot() - }; - // Enter snapshot mode, which atomically reads and resets the modified count. // Checking after start_snapshot ensures no concurrent increments can race. let (snapshot_guard, has_modifications) = self.storage.start_snapshot(); @@ -1254,6 +1306,36 @@ impl TurboTasksBackend { unreachable!("transient task_ids should never be enqueued to be persisted"); } + if self.gc_enabled { + if inner.flags.deleted() { + debug_assert!( + !inner.flags.new_task(), + "a scanned GC-deleted task must be persisted; new tasks are discarded by \ + GC" + ); + let task_type_hash = compute_task_type_hash( + inner + .get_persistent_task_type() + .expect("a GC-deleted task must have a task type"), + ); + return SnapshotItem::Delete { + task_id, + task_type_hash, + }; + } else { + debug_assert!( + !inner.gc_maybe_collectible(), + "tasks scheduled for persistent must not be collectible, this implies a \ + missed task during GC" + ); + } + } else { + debug_assert!( + !inner.flags.deleted(), + "Deleted flags should only be set by GC and it is disabled" + ) + } + let encode_meta = inner.flags.meta_modified(); let encode_data = inner.flags.data_modified(); @@ -1420,11 +1502,23 @@ impl TurboTasksBackend { // as_millis_f64 is not stable yet .as_secs_f64() * 1000.0; - let wall_end_ms = wall_start_ms + elapsed.as_secs_f64() * 1000.0; + let wall_end_ms: f64 = wall_start_ms + elapsed.as_secs_f64() * 1000.0; + let (persist_start_ms, persist_end_ms) = if let Some(gc_elapsed) = gc_elapsed { + let persist_begin_ms = wall_start_ms + gc_elapsed.as_secs_f64() * 1000.0; + turbo_tasks.send_compilation_event(Arc::new(TraceEvent::new( + "turbopack-gc", + wall_start_ms, + persist_begin_ms, + serde_json::json!([]), + ))); + (persist_begin_ms, wall_end_ms) + } else { + (wall_start_ms, wall_end_ms) + }; turbo_tasks.send_compilation_event(Arc::new(TraceEvent::new( "turbopack-persistence", - wall_start_ms, - wall_end_ms, + persist_start_ms, + persist_end_ms, serde_json::json!([ ["reason", reason.as_str()], [ @@ -1485,15 +1579,11 @@ impl TurboTasksBackend { } // eagerly drop the task cache before persisting self.storage.drop_task_cache(); - if self.should_persist() { - // The task_cache is a pure perf cache backed by the DB and isn't read during the - // stop snapshot (no task creation runs concurrently with stop). Drop it before - // persisting to lower peak memory during the serialization/write. - if let Err(err) = + if self.should_persist() + && let Err(err) = self.snapshot_and_persist(Span::current().into(), SnapshotReason::Stop, turbo_tasks) - { - eprintln!("Persisting failed during shutdown: {err:?}"); - } + { + eprintln!("Persisting failed during shutdown: {err:?}"); } self.storage.drop_contents(); if let Err(err) = self.backing_storage.shutdown() { @@ -1932,6 +2022,7 @@ impl TurboTasksBackend { { let mut ctx = self.execute_context(turbo_tasks); let mut task = ctx.task(task_id, TaskDataCategory::All); + task.assert_not_deleted("try_start_task_execution"); task_type = task.get_task_type().to_owned(); let once_task = matches!(task_type, TaskType::Transient(ref tt) if matches!(&**tt, TransientTask::Once(_))); if let Some(tasks) = task.prefetch() { @@ -2529,7 +2620,7 @@ impl TurboTasksBackend { ) .entered(); let mut make_stale = true; - let dependent = ctx.task(dependent_task_id, TaskDataCategory::All); + let mut dependent = ctx.task(dependent_task_id, TaskDataCategory::All); let transient_task_type = dependent.get_transient_task_type(); if transient_task_type.is_some_and(|tt| matches!(&**tt, TransientTask::Once(_))) { // once tasks are never invalidated @@ -2553,8 +2644,7 @@ impl TurboTasksBackend { return; } make_task_dirty_internal( - dependent, - dependent_task_id, + &mut dependent, make_stale, #[cfg(feature = "task_dirty_cause")] cause.clone(), @@ -3118,6 +3208,7 @@ impl TurboTasksBackend { ) -> Result { let mut ctx = self.execute_context(turbo_tasks); let task = ctx.task(task_id, TaskDataCategory::Data); + task.assert_not_deleted("try_read_own_task_cell"); if let Some(content) = task.get_cell_data(&cell).cloned() { Ok(CellContent(Some(content)).into_typed(cell.type_id())) } else { @@ -3136,6 +3227,7 @@ impl TurboTasksBackend { let mut collectibles = AutoMap::default(); { let mut task = ctx.task(task_id, TaskDataCategory::All); + task.assert_not_deleted("read_task_collectibles"); if task .get_persistent_task_type() .is_some_and(|t| !t.native_fn.is_root) @@ -3762,6 +3854,14 @@ impl Backend for TurboTasksBackend { self.mark_own_task_as_finished(task_id, turbo_tasks); } + fn pin_task_for_gc(&self, task: TaskId, turbo_tasks: &TurboTasks) { + self.pin_task_for_gc(task, turbo_tasks); + } + + fn unpin_task_for_gc(&self, task: TaskId, turbo_tasks: &TurboTasks) { + self.unpin_task_for_gc(task, turbo_tasks); + } + fn connect_task( &self, task: TaskId, diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/aggregation_update.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/aggregation_update.rs index 733de17d9e54..0b786533c641 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/operation/aggregation_update.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/aggregation_update.rs @@ -1,3 +1,15 @@ +//! Maintenance of the aggregation tree (`upper` / `followers` edges). +//! +//! # GC invariant: no aggregation edge may point at a GC-deleted task +//! +//! [`crate::backend::gc`] soft-deletes a task (`deleted` flag) and then removes all its outgoing +//! edges. For this to be correct all 'incoming edges' must be gone. However GC is a highly +//! concurrent process and an AggregationUpdateQueue may 'suspend' while GC is running. To assist +//! with GC aggregation updates need to be defensive about running on deleted tasks and also assist +//! GC by telling the context whenever an aggregation update queue operation may make a task +//! collectible (via [`ExecuteContext::note_maybe_collectible`] on `1->0` transitions), and also +//! as a safety check we [`TaskGuard::assert_not_deleted`] on every +1 transition + use std::{ cmp::max, collections::{VecDeque, hash_map::Entry as HashMapEntry}, @@ -29,7 +41,10 @@ use turbo_tasks::{FxIndexMap, TaskExecutionReason, TaskId, TaskPriority, event:: use crate::{ backend::{ TaskDataCategory, - operation::{ExecuteContext, Operation, TaskGuard, invalidate::make_task_dirty}, + operation::{ + ExecuteContext, Operation, TaskGuard, connect_child::resurrect_deleted, + invalidate::make_task_dirty, + }, storage_schema::TaskStorageAccessors, }, data::{ActivenessState, AggregationNumber, CollectibleRef}, @@ -1457,18 +1472,18 @@ impl AggregationUpdateQueue { } } AggregationUpdateJob::AdjustParentCount { task_ids, delta } => { - ctx.for_each_task_meta(task_ids, "AdjustParentCount", |mut task, _ctx| { - task.update_and_get_parent_count(delta); + ctx.for_each_task_meta(task_ids, "AdjustParentCount", |mut task, ctx| { + if task.update_and_get_parent_count(delta) == 0 { + ctx.note_maybe_collectible(&task); + } }); } AggregationUpdateJob::AdjustTransientRefCount { task_ids, delta } => { - ctx.for_each_task_meta( - task_ids, - "AdjustTransientRefCount", - |mut task, _ctx| { - task.update_and_get_transient_ref_count(delta); - }, - ); + ctx.for_each_task_meta(task_ids, "AdjustTransientRefCount", |mut task, ctx| { + if task.update_and_get_transient_ref_count(delta) == 0 { + ctx.note_maybe_collectible(&task); + } + }); } AggregationUpdateJob::DecreaseActiveCount { task } => { self.decrease_active_count(ctx, task); @@ -1618,7 +1633,7 @@ impl AggregationUpdateQueue { "schedule tasks", |task, ctx| { let parent_priority = self.scheduled_tasks[&task.id()]; - ctx.schedule_task(task, parent_priority); + ctx.schedule_task(&task, parent_priority); }, ); self.scheduled_tasks.clear(); @@ -1666,6 +1681,10 @@ impl AggregationUpdateQueue { let upper_ids = get_uppers(&upper); // Add the same amount of upper edges + // Probe: neither endpoint of a new aggregation edge may be GC-deleted. + // `balance_edge` is the one site holding both guards, so both are checked. + upper.assert_not_deleted("balance_edge add upper (upper endpoint)"); + task.assert_not_deleted("balance_edge add upper (inner endpoint)"); if task.update_upper_count(upper_id, count) { if task.upper_len().is_power_of_two() { self.push_optimize_task(&mut task); @@ -1729,6 +1748,8 @@ impl AggregationUpdateQueue { let upper_ids = get_uppers(&upper); // Add the same amount of follower edges + upper.assert_not_deleted("balance_edge add follower (upper endpoint)"); + task.assert_not_deleted("balance_edge add follower (follower endpoint)"); if upper.update_followers_count(task_id, count) { // May optimize the task if upper.followers_len().is_power_of_two() { @@ -1935,6 +1956,10 @@ impl AggregationUpdateQueue { if removed_upper { let data = AggregatedDataUpdate::from_task(&mut follower).invert(); let followers = get_followers(&follower); + // if uppers became empty, it might be collectible, check now. + if follower.is_upper_empty() { + ctx.note_maybe_collectible(&follower); + } drop(follower); // STEP 5 @@ -2011,6 +2036,9 @@ impl AggregationUpdateQueue { let has_active_count = ctx.should_track_activeness() && upper.get_activeness().is_some_and(|a| a.active_counter > 0); let upper_ids = get_uppers(&upper); + if upper.is_followers_empty() { + ctx.note_maybe_collectible(&upper); + } drop(upper); // STEP 14 @@ -2109,6 +2137,9 @@ impl AggregationUpdateQueue { if !removed_uppers.is_empty() { let data = AggregatedDataUpdate::from_task(&mut follower).invert(); let followers = get_followers(&follower); + if follower.is_upper_empty() { + ctx.note_maybe_collectible(&follower); + } drop(follower); // STEP 5 @@ -2189,6 +2220,9 @@ impl AggregationUpdateQueue { let has_active_count = ctx.should_track_activeness() && upper.get_activeness().is_some_and(|a| a.active_counter > 0); let upper_ids = get_uppers(&upper); + if upper.is_followers_empty() { + ctx.note_maybe_collectible(&upper); + } drop(upper); // STEP 14 @@ -2297,6 +2331,9 @@ impl AggregationUpdateQueue { if remove_upper { let data = AggregatedDataUpdate::from_task(&mut follower).invert(); let followers = get_followers(&follower); + if follower.is_upper_empty() { + ctx.note_maybe_collectible(&follower); + } drop(follower); // STEP 5 @@ -2378,6 +2415,9 @@ impl AggregationUpdateQueue { let has_active_count = ctx.should_track_activeness() && upper.get_activeness().is_some_and(|a| a.active_counter > 0); let upper_ids = get_uppers(&upper); + if upper.is_followers_empty() { + ctx.note_maybe_collectible(&upper); + } drop(upper); // STEP 14 @@ -2492,6 +2532,7 @@ impl AggregationUpdateQueue { { // STEP 3a // It's a follower of the upper node + upper.assert_not_deleted("inner_of_uppers_has_new_follower add follower"); if upper.update_followers_count(new_follower_id, count) { // STEP 3b // May optimize the task @@ -2573,6 +2614,7 @@ impl AggregationUpdateQueue { } // STEP 6a + new_follower.assert_not_deleted("inner_of_uppers_has_new_follower add upper"); if new_follower.update_upper_count(upper_id, count) { // It's a new upper // STEP 6b @@ -2752,6 +2794,9 @@ impl AggregationUpdateQueue { // STEP 3a // It's a follower of the upper node + upper.assert_not_deleted( + "inner_of_upper_has_new_followers add follower", + ); if upper.update_followers_count(*follower_id, *count) { // STEP 3b // May optimize the task @@ -2846,6 +2891,7 @@ impl AggregationUpdateQueue { } // STEP 6a + new_follower.assert_not_deleted("inner_of_upper_has_new_followers add upper"); if new_follower.update_upper_count(upper_id, count) { // STEP 6b if new_follower.upper_len().is_power_of_two() { @@ -2992,6 +3038,7 @@ impl AggregationUpdateQueue { // STEP 3a // It's a follower of the upper node + upper.assert_not_deleted("inner_of_upper_has_new_follower add follower"); if upper.update_followers_count(new_follower_id, count) { // STEP 3b // May optimize the task @@ -3058,6 +3105,7 @@ impl AggregationUpdateQueue { let _span = trace_span!("new inner").entered(); // STEP 6a + new_follower.assert_not_deleted("inner_of_upper_has_new_follower add upper"); if new_follower.update_upper_count(upper_id, count) { // STEP 6b if new_follower.upper_len().is_power_of_two() { @@ -3172,12 +3220,14 @@ impl AggregationUpdateQueue { #[cfg(feature = "trace_aggregation_update")] let _span = trace_span!("increase active count").entered(); - let mut task = ctx.task( + let task = ctx.task( task_id, // For performance reasons this should stay Meta and not All. // persistent_task_type is now set eagerly in initialize_new_task. AGGREGATION_UPDATE_CATEGORY, ); + // Revive the task if GC soft-deleted it. + let mut task = resurrect_deleted(task, task_id, self, ctx); self.check_optimization_pending(&task); let state = task.get_activeness_mut_or_insert_with(|| ActivenessState::new(task_id)); let is_new = state.is_empty(); @@ -3226,6 +3276,12 @@ impl AggregationUpdateQueue { // For performance reasons this should stay `Meta` and not `All` AGGREGATION_UPDATE_CATEGORY, ); + // Skip deleted tasks, GC can run during suspend points of aggregation updates + // This just means there is no point in rebalancing the subgraph here since this 'root' is + // dead. + if task.deleted() { + return; + } self.check_optimization_pending(&task); let current = task.get_aggregation_number().copied().unwrap_or_default(); let old = current.effective; @@ -3329,6 +3385,11 @@ impl AggregationUpdateQueue { // free and lets us elide the write entirely. task.set_optimization_pending(false); } + // A racing GC pass may have deleted this task, just return + // still clear the `optimization_pending` flag just in case we get resurrected. + if task.deleted() { + return; + } let aggregation_number = task.get_aggregation_number().copied().unwrap_or_default(); if is_root_node(aggregation_number.effective) { return; diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/cleanup_old_edges.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/cleanup_old_edges.rs index 23ba912b6b35..35cba7767a2a 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/operation/cleanup_old_edges.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/cleanup_old_edges.rs @@ -54,6 +54,39 @@ pub enum OutdatedEdge { CollectiblesDependency(CollectiblesRef), } +/// Captures *all* of a task's outgoing edges as [`OutdatedEdge`]s +pub fn capture_all_outgoing_edges(task: &impl TaskStorageAccessors) -> Vec { + let mut old_edges: Vec = Vec::new(); + old_edges.extend(task.iter_children().map(OutdatedEdge::Child)); + old_edges.extend( + task.iter_output_dependencies() + .map(OutdatedEdge::OutputDependency), + ); + old_edges.extend( + task.iter_cell_dependencies() + .map(OutdatedEdge::CellDependency), + ); + old_edges.extend( + task.iter_cell_dependencies_hashed() + .map(|(r, k)| OutdatedEdge::HashedCellDependency(r, k)), + ); + old_edges.extend( + task.iter_collectibles_dependencies() + .map(OutdatedEdge::CollectiblesDependency), + ); + old_edges +} + +/// The category to open a dependency *target* with when scrubbing its incoming edge. +fn dependent_scrub_category<'e, C: ExecuteContext<'e>>(ctx: &C) -> TaskDataCategory { + if ctx.collects_gc_candidates() { + // Under GC we need to query meta fields so be sure to recover Meta also + TaskDataCategory::All + } else { + TaskDataCategory::Data + } +} + #[cfg(feature = "trace_aggregation_update_stats")] type Stats = super::aggregation_update::AggregationUpdateQueueStats; #[cfg(not(feature = "trace_aggregation_update_stats"))] @@ -194,11 +227,15 @@ impl CleanupOldEdgesOperation { cell, } = forward; { - let mut task = ctx.task(cell_task_id, TaskDataCategory::Data); - task.remove_cell_dependents(&CellRef { + let category = dependent_scrub_category(ctx); + let mut task = ctx.task(cell_task_id, category); + let removed = task.remove_cell_dependents(&CellRef { task: task_id, cell, }); + if removed && task.is_cell_dependents_empty() { + ctx.note_maybe_collectible(&task); + } } { let mut task = ctx.task(task_id, TaskDataCategory::Data); @@ -212,14 +249,19 @@ impl CleanupOldEdgesOperation { cell, } = forward; { - let mut task = ctx.task(cell_task_id, TaskDataCategory::Data); - task.remove_cell_dependents_hashed(&( + let category = dependent_scrub_category(ctx); + let mut task = ctx.task(cell_task_id, category); + let removed = task.remove_cell_dependents_hashed(&( CellRef { task: task_id, cell, }, key, )); + + if removed && task.is_cell_dependents_hashed_empty() { + ctx.note_maybe_collectible(&task); + } } { let mut task = ctx.task(task_id, TaskDataCategory::Data); @@ -235,8 +277,12 @@ impl CleanupOldEdgesOperation { ) .entered(); { - let mut task = ctx.task(output_task_id, TaskDataCategory::Data); - task.remove_output_dependent(&task_id); + let category = dependent_scrub_category(ctx); + let mut task = ctx.task(output_task_id, category); + let removed = task.remove_output_dependent(&task_id); + if removed && task.is_output_dependent_empty() { + ctx.note_maybe_collectible(&task); + } } { let mut task = ctx.task(task_id, TaskDataCategory::Data); @@ -248,12 +294,15 @@ impl CleanupOldEdgesOperation { task: dependent_task_id, }) => { { - let mut task = - ctx.task(dependent_task_id, TaskDataCategory::Data); - task.remove_collectibles_dependents(&( + let category = dependent_scrub_category(ctx); + let mut task = ctx.task(dependent_task_id, category); + let removed = task.remove_collectibles_dependents(&( collectible_type, task_id, )); + if removed && task.collectibles_dependents_len() == 0 { + ctx.note_maybe_collectible(&task); + } } { let mut task = ctx.task(task_id, TaskDataCategory::Data); diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_child.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_child.rs index d098e6ac64f3..d362b037bbd4 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_child.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_child.rs @@ -9,12 +9,55 @@ use crate::{ aggregation_update::{ AggregationUpdateJob, AggregationUpdateQueue, get_aggregation_number, is_root_node, }, + invalidate::make_task_dirty_internal, }, storage_schema::TaskStorageAccessors, }, data::{InProgressState, InProgressStateInner}, }; +/// Revive `task_id` if it was GC-soft-deleted, given a guard the caller already holds during the +/// connect handshake. +/// +/// GC destructively mutates tasks so mark resurrected tasks as dirty to get them re-scheduled. +pub(super) fn resurrect_deleted<'e, C: ExecuteContext<'e>>( + guard: C::TaskGuardImpl, + task_id: TaskId, + queue: &mut AggregationUpdateQueue, + ctx: &mut C, +) -> C::TaskGuardImpl { + if !guard.deleted() { + return guard; + } + #[cfg(debug_assertions)] + let category = guard.access(); + drop(guard); + + let mut task = ctx.task(task_id, TaskDataCategory::All); + // Double-check under the re-acquired guard: a concurrent connect may have already done this + if task.deleted() { + task.set_deleted(false); + // Mark dirty so it is rescheduled, GC has already dropped its edges and data, so we need to + // re-execute them to bring it back + // NOTE: recovering from disk is technically sometimes possible but doesn't work for new + // tasks, and the snapshot may have already persisted a tombstone. So it would at + // best be an optimistic way to recover data that is in the process of being deleted. It + // shouldn't matter for resolving this rare race condition. + make_task_dirty_internal( + &mut task, + /* make_stale */ true, + #[cfg(feature = "task_dirty_cause")] + turbo_tasks::TaskDirtyCause::Resurrected, + queue, + ctx, + ); + } + // Conditionally downgrade from All->category so we don't hide incorrect access patterns. + #[cfg(debug_assertions)] + task.downgrade_access(category); + task +} + #[derive(Encode, Decode, Clone, Default)] #[allow(clippy::large_enum_variant)] pub enum ConnectChildOperation { @@ -61,7 +104,7 @@ impl ConnectChildOperation { let mut queue = AggregationUpdateQueue::new(); - // Handle the transient to persistent boundary by making the persistent task a root task + // Handle the transient to persistent boundary by making the persistent task a root task. let should_make_root = parent_task_id.is_none_or(|id| id.is_transient() && !child_task_id.is_transient()); @@ -79,8 +122,12 @@ impl ConnectChildOperation { } else { // First connect of this child: its id is minted but the storage entry may not exist // yet, and concurrent connects race to be the one that first touches it. - let mut child_task = - ctx.open_or_create_task_storage(child_task_id, TaskDataCategory::Meta); + let child_task = ctx.open_or_create_task_storage(child_task_id, TaskDataCategory::Meta); + + // Revive the child if GC soft-deleted it. This can happen in a rare race between a + // cache hit on a task and snapshotting actually performing the delete. + let mut child_task = resurrect_deleted(child_task, child_task_id, &mut queue, &mut ctx); + let has_output = child_task.has_output(); // An already constructed top-level task was made a root when it was first connected. // It may still be dirty and need to run; this only avoids repeating the idempotent @@ -103,7 +150,7 @@ impl ConnectChildOperation { EventDescription::new(|| child_task.get_task_desc_fn()), ) { - ctx.schedule_task(child_task, ctx.get_current_task_priority()); + ctx.schedule_task(&child_task, ctx.get_current_task_priority()); } } diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_children.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_children.rs index bfcc2c81c012..7198a2dd4435 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_children.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_children.rs @@ -79,10 +79,8 @@ pub fn connect_children( } } if !child.has_output() { - let child_id = child.id(); make_task_dirty_internal( - child, - child_id, + &mut child, false, #[cfg(feature = "task_dirty_cause")] TaskDirtyCause::InitialDirty, diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/invalidate.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/invalidate.rs index 39908bf4a686..d5d40991fd90 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/operation/invalidate.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/invalidate.rs @@ -13,6 +13,7 @@ use crate::{ AggregationUpdateJob, AggregationUpdateQueue, ComputeDirtyAndCleanUpdate, }, }, + storage_schema::TaskStorageAccessors, }, data::{Dirtyness, InProgressState, InProgressStateInner}, }; @@ -93,10 +94,9 @@ pub fn make_task_dirty( queue: &mut AggregationUpdateQueue, ctx: &mut impl ExecuteContext<'_>, ) { - let task = ctx.task(task_id, TaskDataCategory::All); + let mut task = ctx.task(task_id, TaskDataCategory::All); make_task_dirty_internal( - task, - task_id, + &mut task, true, #[cfg(feature = "task_dirty_cause")] cause, @@ -105,13 +105,13 @@ pub fn make_task_dirty( ); } -pub fn make_task_dirty_internal( - mut task: impl TaskGuard, - task_id: TaskId, +/// Requires the guard to be allocated with [TaskDataCategory::All] +pub fn make_task_dirty_internal<'e, E: ExecuteContext<'e>>( + task: &mut E::TaskGuardImpl, make_stale: bool, #[cfg(feature = "task_dirty_cause")] cause: TaskDirtyCause, queue: &mut AggregationUpdateQueue, - ctx: &mut impl ExecuteContext<'_>, + ctx: &mut E, ) { // There must be no way to invalidate immutable tasks. If there would be a way the task is not // immutable. @@ -252,11 +252,8 @@ pub fn make_task_dirty_internal( } .compute(); - if let Some(aggregated_update) = result.aggregated_update(task_id) { - queue.extend(AggregationUpdateJob::data_update( - &mut task, - aggregated_update, - )); + if let Some(aggregated_update) = result.aggregated_update(task.id()) { + queue.extend(AggregationUpdateJob::data_update(task, aggregated_update)); } let should_schedule = !ctx.should_track_activeness() || task.has_activeness(); @@ -264,9 +261,7 @@ pub fn make_task_dirty_internal( if should_schedule { let description = EventDescription::new(|| task.get_task_desc_fn()); if task.add_scheduled(TaskExecutionReason::Invalidated, description) { - drop(task); - let task = ctx.task(task_id, TaskDataCategory::All); - ctx.schedule_task(task, parent_priority); + ctx.schedule_task(&*task, parent_priority); } } } diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs index c570b75e454e..a08ac1926761 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs @@ -27,7 +27,8 @@ pub use self::aggregation_update::ComputeDirtyAndCleanUpdate; use crate::{ backend::{ EventDescription, TaskDataCategory, TurboTasksBackend, - snapshot_coordinator::OperationGuard, + cell_data::CellData, + snapshot_coordinator::{OperationGuard, SnapshotPhase}, storage::{SpecificTaskDataCategory, StorageWriteGuard, TrackOutcome}, storage_schema::{TaskStorage, TaskStorageAccessors}, }, @@ -54,6 +55,9 @@ enum TaskAccess { MustExist, } +// TODO: consider removing this trait (and `TaskGuard`) in favor of the concrete types. Each has +// exactly one implementation (`ExecuteContextImpl` / `TaskGuardImpl`), so the abstraction buys +// nothing and just adds declaration overhead and extra generic plumbing. pub trait ExecuteContext<'e>: Sized { type TaskGuardImpl: TaskGuard + 'e; fn child_context<'l, 'r>(&'r self) -> impl ChildExecuteContext<'l> + use<'e, 'l, Self> @@ -127,11 +131,23 @@ pub trait ExecuteContext<'e>: Sized { task_id2: TaskId, category: TaskDataCategory, ) -> (Self::TaskGuardImpl, Self::TaskGuardImpl); - fn schedule_task(&self, task: Self::TaskGuardImpl, parent_priority: TaskPriority); + fn schedule_task(&self, task: &Self::TaskGuardImpl, parent_priority: TaskPriority); fn get_current_task_priority(&self) -> TaskPriority; fn operation_suspend_point(&mut self, op: &T) where T: Clone + Into; + /// Record `task` as a GC candidate **if it is in fact collectible**. + /// + /// Call this wherever an operation removes the last reference of some kind to a task. This may + /// be a transition to collectibility. + /// + /// Only effective in a gc context see [`Self::collects_gc_candidates`]. + fn note_maybe_collectible(&mut self, task: &impl TaskGuard); + /// Whether [`Self::note_maybe_collectible`] does anything, i.e. this is a GC context. + /// + /// Lets a caller skip work that only exists to feed the collector — in particular opening a + /// task with a wider [`TaskDataCategory`] than it would otherwise need. + fn collects_gc_candidates(&self) -> bool; fn should_track_dependencies(&self) -> bool; fn should_track_activeness(&self) -> bool; fn turbo_tasks(&self) -> Arc; @@ -203,10 +219,18 @@ impl TaskLockCounter { } } +enum ExecutePhase<'e> { + Normal { + _guard: OperationGuard<'e, AnyOperation>, + }, + Child, + Gc(&'e dyn Fn(TaskId)), +} + pub struct ExecuteContextImpl<'e> { backend: &'e TurboTasksBackend, turbo_tasks: &'e TurboTasks, - _operation_guard: Option>, + phase: ExecutePhase<'e>, task_lock_counter: TaskLockCounter, } @@ -218,7 +242,29 @@ impl<'e> ExecuteContextImpl<'e> { Self { backend, turbo_tasks, - _operation_guard: Some(backend.start_operation()), + phase: ExecutePhase::Normal { + _guard: backend.start_operation(), + }, + task_lock_counter: TaskLockCounter::new(), + } + } + + /// Constructs a context that does NOT take an operation guard, for use by the garbage + /// collector while it holds the coordinator's exclusion phase. + /// + /// The exclusion excludes all concurrent operations and task execution, so taking an operation + /// guard here would deadlock. Requiring `&ExclusionPhase` makes that a type-level obligation: + /// the caller cannot construct this context without actually holding the exclusion. + pub(super) fn new_for_gc( + backend: &'e TurboTasksBackend, + turbo_tasks: &'e TurboTasks, + _phase: &'e SnapshotPhase<'_, AnyOperation>, + gc_collectible: &'e dyn Fn(TaskId), + ) -> Self { + Self { + backend, + turbo_tasks, + phase: ExecutePhase::Gc(gc_collectible), task_lock_counter: TaskLockCounter::new(), } } @@ -1110,8 +1156,8 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { ) } - fn schedule_task(&self, task: Self::TaskGuardImpl, parent_priority: TaskPriority) { - let priority = schedule_priority(&task, parent_priority); + fn schedule_task(&self, task: &Self::TaskGuardImpl, parent_priority: TaskPriority) { + let priority = schedule_priority(task, parent_priority); self.turbo_tasks.schedule(task.id(), priority); } @@ -1120,9 +1166,25 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { } fn operation_suspend_point>(&mut self, op: &T) { + // suspend guards become no-ops under GC + if matches!(self.phase, ExecutePhase::Gc(_)) { + return; + } self.backend.operation_suspend_point(|| op.clone().into()); } + fn note_maybe_collectible(&mut self, task: &impl TaskGuard) { + if let ExecutePhase::Gc(collector) = self.phase + && task.is_gc_collectible() + { + collector(task.id()); + } + } + + fn collects_gc_candidates(&self) -> bool { + matches!(self.phase, ExecutePhase::Gc(_)) + } + fn should_track_dependencies(&self) -> bool { self.backend.should_track_dependencies() } @@ -1180,7 +1242,7 @@ impl<'e> ChildExecuteContext<'e> for ChildExecuteContextImpl<'e> { ExecuteContextImpl { backend: self.backend, turbo_tasks: self.turbo_tasks, - _operation_guard: None, + phase: ExecutePhase::Child, task_lock_counter: TaskLockCounter::new(), } } @@ -1227,6 +1289,26 @@ impl Display for TaskType { pub trait TaskGuard: Debug + TaskStorageAccessors { fn id(&self) -> TaskId; + #[cfg(debug_assertions)] + fn access(&self) -> TaskDataCategory; + #[cfg(debug_assertions)] + fn downgrade_access(&mut self, access: TaskDataCategory); + + /// Asserts this task has not been GC-collected. + #[track_caller] + #[inline] + fn assert_not_deleted(&self, operation: &str) { + debug_assert!( + !self.deleted(), + "{operation} on GC-deleted task {} — a resurrection path was missed", + self.id() + ); + } + + /// Clears all modified/new flags for a GC-collected task that was **never persisted** + /// (`new_task`). + fn discard_modifications_for_gc_new_task(&mut self); + /// Get mutable reference to the activeness state, inserting a new one if not present fn get_activeness_mut_or_insert_with(&mut self, f: F) -> &mut ActivenessState where @@ -1294,6 +1376,18 @@ pub trait TaskGuard: Debug + TaskStorageAccessors { new_value } + /// Whether a GC pass may collect this task: it is non-transient and nothing references it. + /// + /// How much this proves depends on the guard's category — with only `Meta` open it is a sound + /// pre-filter that cannot see dependency edges, and with `All` open it is authoritative. See + /// [`TaskStorage::gc_maybe_collectible`] for the full contract. + fn is_gc_collectible(&self) -> bool { + // Transient-ness is a property of the id, not the storage; transient tasks are never + // collected. + self.check_access(SpecificTaskDataCategory::Meta); + !self.id().is_transient() && self.typed().gc_maybe_collectible() + } + fn invalidate_serialization(&mut self); /// Determine which tasks to prefetch for a task. /// Only returns Some once per task. @@ -1448,6 +1542,16 @@ pub trait TaskGuard: Debug + TaskStorageAccessors { self.typed_mut().cell_data_mut().insert(cell, value) } + fn take_cell_data(&mut self) -> Option { + self.check_access(SpecificTaskDataCategory::Data); + let undoable = self.track_modification(SpecificTaskDataCategory::Data, "cell_data"); + let prev = self.typed_mut().take_cell_data(); + if prev.is_none() { + // very unlikely + self.undo_track_modification(undoable); + } + prev + } /// Remove cell data, returning the old value if present. fn remove_cell_data( &mut self, @@ -1540,6 +1644,7 @@ pub trait TaskGuard: Debug + TaskStorageAccessors { pub struct TaskGuardImpl<'a> { task_id: TaskId, task: StorageWriteGuard<'a>, + // None means no categories are accessible other than transient data. #[cfg(debug_assertions)] category: TaskDataCategory, task_lock_counter: TaskLockCounter, @@ -1556,13 +1661,15 @@ impl TaskGuardImpl<'_> { /// before accessing the data. #[inline] #[track_caller] - fn check_access(&self, category: crate::backend::storage::SpecificTaskDataCategory) { + fn check_access(&self, category: SpecificTaskDataCategory) { match category { SpecificTaskDataCategory::Data => { #[cfg(debug_assertions)] debug_assert!( - self.category == TaskDataCategory::Data - || self.category == TaskDataCategory::All, + matches!( + self.category, + TaskDataCategory::Data | TaskDataCategory::All + ), "To read data of {:?} the task need to be accessed with this category (It's \ accessed with {:?})", category, @@ -1572,8 +1679,10 @@ impl TaskGuardImpl<'_> { SpecificTaskDataCategory::Meta => { #[cfg(debug_assertions)] debug_assert!( - self.category == TaskDataCategory::Meta - || self.category == TaskDataCategory::All, + matches!( + self.category, + TaskDataCategory::Meta | TaskDataCategory::All + ), "To read data of {:?} the task need to be accessed with this category (It's \ accessed with {:?})", category, @@ -1598,6 +1707,24 @@ impl TaskGuard for TaskGuardImpl<'_> { self.task_id } + #[cfg(debug_assertions)] + fn access(&self) -> TaskDataCategory { + self.category + } + #[cfg(debug_assertions)] + fn downgrade_access(&mut self, access: TaskDataCategory) { + assert!( + self.category >= access, + "Cannot downgrade {:?} to {access:?}", + self.category + ); + self.category = access; + } + + fn discard_modifications_for_gc_new_task(&mut self) { + self.task.discard_modifications_for_gc_new_task(); + } + fn invalidate_serialization(&mut self) { // TODO this causes race conditions, since we never know when a value is changed. We can't // "snapshot" the value correctly. @@ -1664,7 +1791,7 @@ impl TaskStorageAccessors for TaskGuardImpl<'_> { #[inline(always)] fn track_modification( &mut self, - category: crate::backend::storage::SpecificTaskDataCategory, + category: SpecificTaskDataCategory, name: &str, ) -> TrackOutcome { if self.task_id.is_transient() { @@ -1682,7 +1809,7 @@ impl TaskStorageAccessors for TaskGuardImpl<'_> { } #[track_caller] - fn check_access(&self, category: crate::backend::storage::SpecificTaskDataCategory) { + fn check_access(&self, category: SpecificTaskDataCategory) { self.check_access(category); } } @@ -1751,7 +1878,7 @@ pub use self::{ AggregatedDataUpdate, AggregationUpdateJob, get_aggregation_number, get_uppers, is_aggregating_node, is_root_node, }, - cleanup_old_edges::OutdatedEdge, + cleanup_old_edges::{OutdatedEdge, capture_all_outgoing_edges}, connect_children::connect_children, invalidate::make_task_dirty_internal, prepare_new_children::prepare_new_children, diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/update_cell.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/update_cell.rs index 53d6b7ceba31..7ccd46de6103 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/operation/update_cell.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/update_cell.rs @@ -295,7 +295,7 @@ impl Operation for UpdateCellOperation { } => { if let Some((dependent_task_id, keys)) = dependent_tasks.pop() { let mut make_stale = false; - let dependent = ctx.task(dependent_task_id, TaskDataCategory::All); + let mut dependent = ctx.task(dependent_task_id, TaskDataCategory::All); for key in keys.iter().copied() { let (in_outdated, in_current) = if let Some(k) = key { let in_outdated = dependent @@ -327,8 +327,7 @@ impl Operation for UpdateCellOperation { } } make_task_dirty_internal( - dependent, - dependent_task_id, + &mut dependent, make_stale, #[cfg(feature = "task_dirty_cause")] TaskDirtyCause::CellChange { diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/snapshot_coordinator.rs b/turbopack/crates/turbo-tasks-backend/src/backend/snapshot_coordinator.rs index 7fe6bbb46953..aedc486a95f2 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/snapshot_coordinator.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/snapshot_coordinator.rs @@ -1,15 +1,21 @@ -//! Coordinator that gates concurrent operations against snapshotting. +//! Coordinator that gates concurrent operations against snapshotting and garbage collection. //! -//! Backend operations and snapshot work share a single [`SnapshotCoordinator`] -//! that enforces the protocol: +//! Backend operations, snapshot work, and garbage collection share a single +//! [`SnapshotCoordinator`] that enforces the protocol: //! -//! - When no snapshot is in flight, [`begin_operation`](SnapshotCoordinator::begin_operation) is a -//! single uncontended atomic increment. -//! - When a snapshot is requested, new operations block until the snapshot finishes, and operations -//! already in flight either complete or call -//! [`suspend_point`](SnapshotCoordinator::suspend_point) to suspend. -//! - The snapshotter waits for every in-flight operation to drain or suspend, takes its snapshot, -//! then wakes everyone. +//! - When no exclusive phase is in flight, +//! [`begin_operation`](SnapshotCoordinator::begin_operation) is a single uncontended atomic +//! increment. +//! - When a phase is requested, new operations block until it finishes, and operations already in +//! flight either complete or call [`suspend_point`](SnapshotCoordinator::suspend_point) to +//! suspend. +//! - The phase holder waits for every in-flight operation to drain or suspend, does its work, then +//! wakes everyone. +//! +//! Snapshotting and GC are the same kind of exclusion and share one [`ExclusionPhase`]: both need +//! every operation stopped, neither can overlap the other, and GC hands its work straight to the +//! snapshot that persists it. A single [`EXCLUSION_REQUESTED_BIT`] covers both, which is also what +//! lets a GC pass and the snapshot that commits it run under one uninterrupted guard. use std::sync::{ Arc, @@ -18,6 +24,7 @@ use std::sync::{ use parking_lot::{Condvar, Mutex}; use rustc_hash::FxHashSet; +use tracing::info_span; use crate::{backend::AnyOperation, utils::ptr_eq_arc::PtrEqArc}; @@ -36,7 +43,7 @@ struct State { suspended_operations: FxHashSet>, } -/// Coordinates operation/snapshot interleaving. +/// Coordinates operation/snapshot/GC interleaving. /// /// Generic over the operation type the caller wants to suspend. The /// coordinator only requires `O: Send + Sync + 'static`; it never inspects @@ -106,8 +113,10 @@ impl SnapshotCoordinator { if prev - 1 == SNAPSHOT_REQUESTED_BIT { this.operations_drained.notify_all(); } - this.snapshot_completed - .wait_while(&mut state, |s| s.snapshot_requested); + tokio::task::block_in_place(|| { + this.snapshot_completed + .wait_while(&mut state, |s| s.snapshot_requested); + }); // Re-add now that the snapshot is done. Bit is cleared because // we just observed `snapshot_requested == false` under the // mutex. @@ -130,13 +139,13 @@ impl SnapshotCoordinator { } #[cold] fn suspend_point_cold(this: &SnapshotCoordinator, suspend: impl FnOnce() -> O) { - let op = Arc::new(suspend()); let mut state = this.state.lock(); if !state.snapshot_requested { // Race: snapshot finished between the `snapshot_pending` check // and acquiring the mutex. Nothing to do. return; } + let op = Arc::new(suspend()); state .suspended_operations .insert(PtrEqArc::from(op.clone())); @@ -153,8 +162,11 @@ impl SnapshotCoordinator { this.operations_drained.notify_all(); } // Wait for the snapshot to finish. - this.snapshot_completed - .wait_while(&mut state, |s| s.snapshot_requested); + tokio::task::block_in_place(|| { + this.snapshot_completed + .wait_while(&mut state, |s| s.snapshot_requested); + }); + // Resume: re-increment and remove ourselves from the suspended set. this.in_progress_operations.fetch_add(1, Ordering::AcqRel); state.suspended_operations.remove(&PtrEqArc::from(op)); @@ -190,11 +202,19 @@ impl SnapshotCoordinator { "snapshot bit was already set when begin_snapshot ran: {active:#x}" ); if (active & !SNAPSHOT_REQUESTED_BIT) != 0 { - // Some operations are in flight. Wait for them to drain or - // suspend. The predicate is Acquire-loaded so we synchronize - // with the AcqRel decrement that woke us. - self.operations_drained.wait_while(&mut state, |_| { - self.in_progress_operations.load(Ordering::Acquire) != SNAPSHOT_REQUESTED_BIT + // The predicate is Acquire-loaded so we synchronize with the AcqRel decrement that woke + // us. This can block for a while under load (until every in-flight operation reaches a + // suspend point or finishes), so it gets its own span for latency attribution. + let num_operations = active & !SNAPSHOT_REQUESTED_BIT; + let _span = info_span!("await operations settle", num_operations).entered(); + // Release our worker thread so pending operations have it available. This preserves + // liveness: this runs as a `tokio::spawn` background job, so parking here without + // handing the worker back would starve the very operations we are waiting on. + tokio::task::block_in_place(|| { + self.operations_drained.wait_while(&mut state, |_| { + (self.in_progress_operations.load(Ordering::Acquire) & !SNAPSHOT_REQUESTED_BIT) + != 0 + }); }); } // Snapshot ranges that follow can read the suspended_operations @@ -254,7 +274,7 @@ impl Drop for OperationGuard<'_, O> { // (not under the user mutex), so a notifier that has never synchronized // with the user mutex can racily observe stale null and drop the notify. // - // It is generally a best practice to only notify under the loc + // It is generally a best practice to only notify under the lock let _g = coord.state.lock(); coord.operations_drained.notify_all(); } @@ -293,7 +313,7 @@ impl Drop for SnapshotPhase<'_, O> { let prev = self .coord .in_progress_operations - .fetch_sub(SNAPSHOT_REQUESTED_BIT, Ordering::AcqRel); + .fetch_and(!SNAPSHOT_REQUESTED_BIT, Ordering::AcqRel); assert!( (prev & SNAPSHOT_REQUESTED_BIT) != 0, "SnapshotPhase::drop: snapshot bit was already cleared (prev={prev:#x})" @@ -412,11 +432,19 @@ mod tests { while arrived.load(Ordering::Acquire) == 0 { thread::yield_now(); } - assert_eq!(started_op.load(Ordering::Acquire), 0); + assert_eq!( + started_op.load(Ordering::Acquire), + 0, + "a new operation must block while the exclusion is held" + ); drop(phase); op_thread.join().unwrap(); - assert_eq!(started_op.load(Ordering::Acquire), 1); + assert_eq!( + started_op.load(Ordering::Acquire), + 1, + "the operation must run once the exclusion ends" + ); } #[test] @@ -431,7 +459,11 @@ mod tests { let snapshotter_done = snapshotter_done.clone(); move || { let phase = coord_snap.begin_snapshot(); - assert_eq!(phase.suspended_operations().len(), 1); + assert_eq!( + phase.suspended_operations().len(), + 1, + "must record the suspended operation for replay" + ); snapshotter_done.store(1, Ordering::Release); // Hold the snapshot for a moment so the suspend_point thread // observes `snapshot_requested == true` after waking. @@ -440,11 +472,26 @@ mod tests { }); wait_for_snapshot_pending(&coord); - // Snapshotter is now waiting for our operation to drain. Calling - // suspend_point should let it proceed. - coord.suspend_point(|| 42u32); - // suspend_point returns once the snapshot is finished. - assert_eq!(snapshotter_done.load(Ordering::Acquire), 1); + // The phase is waiting for our operation to drain; suspending should let it proceed. + let recorded = Arc::new(AtomicUsize::new(0)); + coord.suspend_point({ + let recorded = recorded.clone(); + move || { + recorded.fetch_add(1, Ordering::Release); + 42u32 + } + }); + // `suspend_point` returns only once the phase has finished. + assert_eq!( + snapshotter_done.load(Ordering::Acquire), + 1, + "suspend_point must not return before the phase completes" + ); + assert_eq!( + recorded.load(Ordering::Acquire), + 1, + "the suspend closure must run so the operation is recorded for a possible snapshot" + ); snap_thread.join().unwrap(); drop(g); @@ -595,4 +642,12 @@ mod tests { "in_progress_operations should be 0 after all ops and snapshots done" ); } + + #[test] + #[should_panic(expected = "already in flight")] + fn overlapping_exclusions_panic() { + let coord = SnapshotCoordinator::::new(); + let _first = coord.begin_snapshot(); + let _second = coord.begin_snapshot(); + } } diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs b/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs index 456ddda81fe2..9d2227eb9064 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs @@ -9,6 +9,7 @@ use std::{ }, }; +use crossbeam_utils::CachePadded; use hashbrown::hash_table; use thread_local::ThreadLocal; use tracing::span::Id; @@ -34,6 +35,20 @@ pub enum TaskDataCategory { Data, All, } +impl PartialOrd for TaskDataCategory { + /// `All` is greater than both `Meta` and `Data`; `Meta` and `Data` are unordered. + fn partial_cmp(&self, other: &Self) -> Option { + use std::cmp::Ordering::*; + + use TaskDataCategory::All; + match (self, other) { + _ if self == other => Some(Equal), + (All, _) => Some(Greater), + (_, All) => Some(Less), + _ => None, + } + } +} /// Counts of tasks evicted at each level. #[derive(Debug, Default)] @@ -161,7 +176,7 @@ pub struct Storage { /// that `shard_modified_counts.len()==map.shards().len()` /// /// Should only be modified while holding the corresponding dashmap shard lock. - shard_modified_counts: Box<[AtomicU64]>, + shard_modified_counts: Box<[CachePadded]>, /// Stores snapshots of task state for tasks accessed during snapshot mode. /// - `Some(snapshot)`: Task was modified before snapshot mode and accessed again during it. /// Contains a copy of the pre-snapshot state that needs to be persisted. @@ -218,7 +233,7 @@ impl Storage { shard_amount, ); let shard_modified_counts = (0..shard_amount) - .map(|_| AtomicU64::new(0)) + .map(|_| CachePadded::new(AtomicU64::new(0))) .collect::>() .into_boxed_slice(); Self { @@ -512,6 +527,41 @@ impl Storage { Some(f(task.value())) } + /// The number of **persistent** (non-transient) tasks resident in the map. Use this to assert + /// GC returns to a flat baseline across re-rooting: GC never collects transient tasks (e.g. + /// `run_once`/Once roots), so their count is not expected to settle. + #[doc(hidden)] + pub fn resident_persistent_task_count_for_testing(&self) -> usize { + let mut persistent = 0; + for shard in self.map.shards() { + let shard = shard.read(); + for (task_id, _) in shard.iter() { + if !task_id.is_transient() { + persistent += 1; + } + } + } + persistent + } + + /// The number of shards in the resident map. GC seeds one `ScanShard` job per index; the slice + /// returned by `map.shards()` is fixed for the map's lifetime, so an index is a stable handle + /// to one shard. + pub fn shard_count(&self) -> usize { + self.map.shards().len() + } + + /// Scans a **single** shard by index, invoking `on_candidate` for each resident, non-transient + /// task whose storage passes the cheap [`TaskStorage::gc_maybe_collectible`] pre-filter. + pub fn gc_scan_shard(&self, index: usize, mut on_candidate: impl FnMut(TaskId)) { + let shard = self.map.shards()[index].read(); + for (task_id, task) in shard.iter() { + if !task_id.is_transient() && task.gc_maybe_collectible() { + on_candidate(*task_id); + } + } + } + pub fn access_pair_mut( &self, key1: TaskId, @@ -573,11 +623,44 @@ impl Storage { // avoid a lock cycle with get_or_create_persistent_task, which takes task_cache // before map. Allocated lazily on first conflict. let mut deferred_task_cache_removals: Vec = Vec::new(); + // Remove a task type from `task_cache`, deferring on contention. Shared by the + // GC-deleted path below and the ordinary key eviction. + let remove_from_task_cache = + |evicted: &mut EvictionCounts, + deferred: &mut Vec, + task_type: &CachedTaskTypeArc| { + match try_lock_and_remove(&self.task_cache, task_type.as_ref()) { + TryLockAndRemove::Removed => { + evicted.key_evictions += 1; + } + TryLockAndRemove::NotFound => { + // Generally this should be rare, it more or less implies something + // else is concurrently holding the Arc + } + TryLockAndRemove::WouldBlock => { + // Contention, to avoid a deadlock just defer + deferred.push(task_type.clone()); + } + } + }; shard.retain(|(task_id, task)| { if task_id.is_transient() { evicted.unevictable_reasons[UnevictableReason::Transient.index()] += 1; return true; } + // GC'd tasks were tombstoned during the snapshot so we can drop them fully now. + if task.flags.deleted() { + let task_type = task + .get_persistent_task_type() + .expect("GC deleted tasks must have a task type"); + remove_from_task_cache( + &mut evicted, + &mut deferred_task_cache_removals, + task_type, + ); + evicted.full += 1; + return false; + } let (key_evictability, value_evictability) = task.evictability(); match key_evictability { KeyEvictability::Evictable => { @@ -589,19 +672,11 @@ impl Storage { // Because `get_or_create_task` acquires 'task_cache' then `storage.map` and // we do the opposite we need to be defensive here. Attempting here is just // an optimization to avoid pushing into `deferred_task_cache_removals` - match try_lock_and_remove(&self.task_cache, task_type.as_ref()) { - TryLockAndRemove::Removed => { - evicted.key_evictions += 1; - } - TryLockAndRemove::NotFound => { - // Generally this should be rare, it more or less implies something - // else is concurrently holding the Arc - } - TryLockAndRemove::WouldBlock => { - // Contention, to avoid a deadlock just defer - deferred_task_cache_removals.push(task_type.clone()); - } - } + remove_from_task_cache( + &mut evicted, + &mut deferred_task_cache_removals, + task_type, + ); } KeyEvictability::AlreadyEvicted | KeyEvictability::Unevictable => {} } @@ -810,6 +885,26 @@ impl StorageWriteGuard<'_> { } } } + + /// Clears all modified/new flags for a GC-collected task that was **never persisted** + /// (`new_task`). + pub fn discard_modifications_for_gc_new_task(&mut self) { + debug_assert!( + !self.storage.snapshot_mode(), + "discard_modifications_for_gc_new_task must run before the snapshot starts" + ); + debug_assert!( + self.inner.flags.new_task(), + "only a never-persisted (new_task) collected task may be discarded this way" + ); + if self.inner.flags.any_modified() { + let shard_idx = self.storage.shard_index(self.inner.key()); + self.storage.shard_modified_counts[shard_idx].fetch_sub(1, Ordering::Relaxed); + } + self.inner.flags.set_meta_modified(false); + self.inner.flags.set_data_modified(false); + self.inner.flags.set_new_task(false); + } } impl Deref for StorageWriteGuard<'_> { diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/storage_schema.rs b/turbopack/crates/turbo-tasks-backend/src/backend/storage_schema.rs index a3eb64e1b225..065f1a2813ef 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/storage_schema.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/storage_schema.rs @@ -230,6 +230,10 @@ struct TaskStorageSchema { #[field(storage = "flag", category = "transient")] pub new_task: bool, + /// GC soft-deletion marker. Set by the garbage collector when a task is marked for deletion. + #[field(storage = "flag", category = "transient")] + deleted: bool, + // ========================================================================= // CHILDREN & AGGREGATION (meta) // ========================================================================= @@ -268,7 +272,7 @@ struct TaskStorageSchema { shrink_on_completion, drop_on_completion_if_immutable )] - cell_dependencies: AutoSet, + cell_dependencies: AutoSet, /// Cells this task depends on, narrowed to a hashed sub-value (`CellDependency::Hash`). Rare. #[field( @@ -296,7 +300,7 @@ struct TaskStorageSchema { /// Outdated keyless cell dependencies to be cleaned up (transient). #[field(storage = "auto_set", category = "transient", shrink_on_completion)] - outdated_cell_dependencies: AutoSet, + outdated_cell_dependencies: AutoSet, /// Outdated hashed cell dependencies to be cleaned up (transient). #[field(storage = "auto_set", category = "transient", shrink_on_completion)] @@ -318,7 +322,7 @@ struct TaskStorageSchema { filter_transient, drop_on_completion_if_immutable )] - cell_dependents: AutoSet, + cell_dependents: AutoSet, /// Tasks that depend on a hashed sub-value of this task's cells. Reverse of /// `cell_dependencies_hashed`. @@ -853,6 +857,47 @@ impl TaskStorage { pub fn gc_transient_ref_count(&self) -> u32 { self.get_transient_ref_count().copied().unwrap_or(0) } + + /// Whether a GC pass may collect this task: nothing references it, via parents, transient + /// pins, aggregation edges, or dependency edges. + /// + /// Precision depends on what the caller restored. Meta alone cannot see the three Data-category + /// dependent sets, so the answer is a sound *pre-filter*: a `false` is definitive, a `true` may + /// still have dependents. With Meta + Data it is the full predicate. That one-directional + /// conservatism lets the cheap Meta-only shard scan and the authoritative under-guard recheck + /// (which opens `TaskDataCategory::All`, and is what actually gates collection) share this + /// single predicate. + /// + /// Refusing to collect a task with dependents is what makes task-id reuse safe: a hard-deleted + /// id can be handed out again, so a surviving dependent edge would silently resolve to an + /// unrelated live task instead of tripping `MustExist`. + pub fn gc_maybe_collectible(&self) -> bool { + // None of the predicates below are correct without this. + self.flags.is_restored(TaskDataCategory::Meta) + // Already collected this session (soft-deleted, awaiting tombstone + hard-delete): + // don't re-select it, or a second pass would collect it again while it is still + // resident. + && !self.flags.deleted() + && self.gc_parent_count() == 0 + && self.gc_transient_ref_count() == 0 + && self.get_activeness().is_none() + && self.get_in_progress().is_none() + // It is rare for upper/followers to be present when the ref counts are 0 but it can happen transiently during a concurrent GC pass as uppers are moved around during the cascade. + && self.upper().is_empty() + && self.followers().is_none_or(|f| f.is_empty()) + // `collectibles_dependents` is Meta, so it is always checkable here. + && self + .collectibles_dependents() + .is_none_or(|d| d.is_empty()) + // The remaining dependent sets are Data; skipped (leaving this a pre-filter) when Data + // is not restored. + && (!self.flags.is_restored(TaskDataCategory::Data) + || (self.output_dependent().is_empty() + && self.cell_dependents().is_none_or(|d| d.is_empty()) + && self + .cell_dependents_hashed() + .is_none_or(|d| d.is_empty()))) + } } /// Counts for aggregation tree and collectibles fields. diff --git a/turbopack/crates/turbo-tasks-backend/src/backing_storage.rs b/turbopack/crates/turbo-tasks-backend/src/backing_storage.rs index 655915be55a0..546ea46d0870 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backing_storage.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backing_storage.rs @@ -22,9 +22,6 @@ pub enum SnapshotItem { /// Task type for new tasks that need to be added to the task cache task_type_hash: Option, }, - // Constructed by the GC pass that emits `Delete` for soft-deleted tasks, which lands in a - // later PR in the stack. - #[allow(dead_code)] Delete { task_id: TaskId, /// The deleted task's `TaskCache` key. Always present: only persistent tasks are diff --git a/turbopack/crates/turbo-tasks-backend/tests/debug.rs b/turbopack/crates/turbo-tasks-backend/tests/debug.rs index 4810c52474bb..77e25e1b195b 100644 --- a/turbopack/crates/turbo-tasks-backend/tests/debug.rs +++ b/turbopack/crates/turbo-tasks-backend/tests/debug.rs @@ -1,3 +1,5 @@ +// `ValueDebug` only carries `dbg()` under `debug_assertions` +#![cfg(debug_assertions)] #![feature(arbitrary_self_types)] #![feature(arbitrary_self_types_pointers)] #![allow(clippy::needless_return)] // tokio macro-generated code doesn't respect this diff --git a/turbopack/crates/turbo-tasks-backend/tests/gc_collection.rs b/turbopack/crates/turbo-tasks-backend/tests/gc_collection.rs new file mode 100644 index 000000000000..692a46f9919d --- /dev/null +++ b/turbopack/crates/turbo-tasks-backend/tests/gc_collection.rs @@ -0,0 +1,202 @@ +#![feature(arbitrary_self_types)] +#![feature(arbitrary_self_types_pointers)] +#![allow(clippy::needless_return)] // tokio macro-generated code doesn't respect this + +mod util; + +use anyhow::Result; +use turbo_tasks::{ResolvedVc, State, TaskId, Vc, prevent_gc}; + +use crate::util::create_tt; + +/// The `TaskId` backing a resolved `Vc` (its `TaskOutput` node). +fn task_id_of(vc: Vc) -> TaskId { + Vc::into_raw(vc) + .try_get_task_id() + .expect("a resolved Vc should be backed by a task") +} + +#[turbo_tasks::value(transparent)] +struct Selector(State); + +#[turbo_tasks::function(operation, root)] +fn create_selector(initial: bool) -> Vc { + Selector(State::new(initial)).cell() +} + +#[turbo_tasks::function] +fn leaf(n: u32) -> Vc { + Vc::cell(n) +} + +/// Resolves `leaf(n)` and returns the raw `TaskId` backing it, as a `u32`. +/// +/// This runs inside a tracked task rather than at the top level of `run_once`, so the plain +/// (eventually consistent) `.resolve()` read below is legal and deterministic — the read is +/// ordered by the task graph instead of racing whatever the session happens to be doing. +#[turbo_tasks::function(operation, root)] +async fn leaf_task_id(n: u32) -> Result> { + Ok(Vc::cell(*task_id_of(leaf(n).resolve().await?))) +} + +#[turbo_tasks::function] +async fn branch_a() -> Result> { + Ok(Vc::cell(1 + *leaf(10).await?)) +} + +#[turbo_tasks::function] +async fn branch_b() -> Result> { + Ok(Vc::cell(2 + *leaf(20).await?)) +} + +/// A task that pins itself against GC while executing. Once pinned it must survive collection even +/// after it is disconnected. +#[turbo_tasks::function] +async fn pinned_branch() -> Result> { + prevent_gc(); + Ok(Vc::cell(99)) +} + +/// Reads exactly one branch depending on the selector; flipping it re-executes and disconnects the +/// previously-read branch (and its subtree), which should drop that branch's `parent_count` to 0. +#[turbo_tasks::function(operation, root)] +async fn select(selector: ResolvedVc) -> Result> { + let use_b = *selector.await?.get(); + let value = if use_b { + *branch_b().await? + } else { + *branch_a().await? + }; + Ok(Vc::cell(value)) +} + +/// Like `select`, but reads `pinned_branch` instead of `branch_a` when the selector is false. +#[turbo_tasks::function(operation, root)] +async fn select_pinned(selector: ResolvedVc) -> Result> { + let use_b = *selector.await?.get(); + let value = if use_b { + *branch_b().await? + } else { + *pinned_branch().await? + }; + Ok(Vc::cell(value)) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn gc_collects_disconnected_subtree() { + let (tt, _persistence_dir) = create_tt("gc_collects_disconnected_subtree"); + let tt2 = tt.clone(); + + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let selector = selector_op.read_strongly_consistent().await?; + + let output = select(selector_vc); + assert_eq!(*output.read_strongly_consistent().await?, 11); + + // Flip: select drops branch_a; branch_a (parent_count 0) becomes a candidate. + selector.set(true); + assert_eq!(*output.read_strongly_consistent().await?, 22); + + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + let collected = tt2.backend().gc_for_testing(&tt2); + assert_eq!( + collected, 2, + "branch_a and its cascaded child leaf(10) should both be collected" + ); + assert_eq!( + tt2.backend().gc_for_testing(&tt2), + 0, + "a second GC pass must collect nothing" + ); + + // Flipping back must recompute branch_a fresh, since it was collected. + let tt3 = tt.clone(); + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(true); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let selector = selector_op.read_strongly_consistent().await?; + let output = select(selector_vc); + assert_eq!(*output.read_strongly_consistent().await?, 22); + selector.set(false); + assert_eq!(*output.read_strongly_consistent().await?, 11); + let _ = &tt3; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + tt.stop_and_wait().await; +} + +/// A task that pins itself via `prevent_gc()` must survive collection even after it is disconnected +/// from the live graph, because the pin makes it a GC root. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn gc_does_not_collect_pinned_task() { + let (tt, _persistence_dir) = create_tt("gc_does_not_collect_pinned_task"); + let tt2 = tt.clone(); + + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let selector = selector_op.read_strongly_consistent().await?; + + let output = select_pinned(selector_vc); + assert_eq!(*output.read_strongly_consistent().await?, 99); + + // Flip: select_pinned re-executes, reads branch_b, and disconnects pinned_branch. + selector.set(true); + assert_eq!(*output.read_strongly_consistent().await?, 22); + + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + // GC runs after the run has released activeness, so pinned_branch is disconnected + // (parent_count 0) and otherwise collectible. + let collected = tt2.backend().gc_for_testing(&tt2); + assert_eq!( + collected, 0, + "a pinned task must not be collected even when disconnected" + ); + + // A snapshot + evict must not lose the (transient) pin. A pinned task is not forced fully + // resident — its Meta/Data may be partially evicted — but the session-only + // `transient_ref_count` is retained as residue (the map entry is kept), so the task stays + // uncollectible and a subsequent GC still collects nothing. + tt2.backend().snapshot_and_evict_for_testing(&tt2); + assert_eq!( + tt2.backend().gc_for_testing(&tt2), + 0, + "pinned task must survive eviction and not be collected" + ); + + tt.stop_and_wait().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unpin_after_stop_does_not_panic() { + let (tt, _persistence_dir) = create_tt("unpin_after_stop_does_not_panic"); + + // Pin a real task inside a session (as `prevent_gc` / `DetachedVc::new` would). + let tt2 = tt.clone(); + let leaf_id = turbo_tasks::run_once(tt.clone(), async move { + let id = TaskId::try_from(*leaf_task_id(7).read_strongly_consistent().await?)?; + tt2.pin_task_for_gc(id); + anyhow::Ok(id) + }) + .await + .unwrap(); + + // Stop the backend — this drops the in-memory task map, so the pinned task is no longer + // resident (exactly as at `next build` shutdown). + tt.stop_and_wait().await; + + tt.unpin_task_for_gc(leaf_id); +} diff --git a/turbopack/crates/turbo-tasks-backend/tests/gc_resurrection.rs b/turbopack/crates/turbo-tasks-backend/tests/gc_resurrection.rs new file mode 100644 index 000000000000..ebd7a6be517e --- /dev/null +++ b/turbopack/crates/turbo-tasks-backend/tests/gc_resurrection.rs @@ -0,0 +1,462 @@ +#![feature(arbitrary_self_types)] +#![feature(arbitrary_self_types_pointers)] +#![allow(clippy::needless_return)] // tokio macro-generated code doesn't respect this + +//! GC teardown of a cleanly-disconnected subtree whose tasks hold real forward-dependency edges on +//! each other. +//! +//! The central invariant: when a task `S` that is the `upper` of its children is collected, GC +//! must **rebalance the aggregation graph** — remove `S` from each child's `upper` set — so the +//! children (now both parentless and upper-less) become collectible and cascade in the same pass. +//! +//! Fixture constraints: only **mutable** tasks record dependency edges (see +//! `add_cell_dependency`), so the leaves read a long-lived `State` whose value never changes. The +//! subtree must also be disconnected *cleanly* — drop the parent's reference, don't invalidate, +//! since invalidation runs `cleanup_old_edges` and strips the outgoing deps first. + +mod util; + +use std::sync::atomic::{AtomicU32, Ordering}; + +use anyhow::Result; +use turbo_tasks::{ResolvedVc, State, Vc}; + +use crate::util::create_tt; + +#[turbo_tasks::value(transparent)] +struct Selector(State); + +#[turbo_tasks::function(operation, root)] +fn create_selector(initial: bool) -> Vc { + Selector(State::new(initial)).cell() +} + +/// A long-lived State whose *value never changes*, read by the leaves purely to make each leaf +/// **mutable**, so a reader records a real dependency edge on it. The state task is a root and +/// stays alive; the leaves reach it via a dependency edge, not a child edge, so disconnecting the +/// leaves as children still lets them lose activeness. +#[turbo_tasks::value(transparent)] +struct Constant(State); + +#[turbo_tasks::function(operation, root)] +fn create_constant() -> Vc { + Constant(State::new(0)).cell() +} + +/// The forward-dependency *target*: mutable because it reads `constant`'s State. `FANOUT` distinct +/// leaves per reader give many chances for the racing interleaving. +#[turbo_tasks::function] +async fn sd_leaf(constant: ResolvedVc, index: u32) -> Result> { + let base = *constant.await?.get(); + Ok(Vc::cell(base.wrapping_add(index))) +} + +const FANOUT: u32 = 400; + +/// The *reader* `S`: reads every `sd_leaf`, so it (a) connects each leaf as a child and (b) records +/// an outgoing dependency on each. When `reader` is collected while still holding those deps, +/// `Collect(reader)` spawns scrub jobs referencing all leaves — and every leaf is itself +/// collectible (cascaded from `reader`'s decrement), so each is a resurrection candidate. +#[turbo_tasks::function] +async fn reader(constant: ResolvedVc) -> Result> { + let mut sum = 0u32; + for index in 0..FANOUT { + sum = sum.wrapping_add(*sd_leaf(*constant, index).await?); + } + Ok(Vc::cell(sum)) +} + +/// A *sibling* forward-dependency target for the diamond fixture (`B`). +#[turbo_tasks::function] +async fn diamond_target(constant: ResolvedVc, index: u32) -> Result> { + let base = *constant.await?.get(); + Ok(Vc::cell(base.wrapping_add(index).wrapping_mul(7))) +} + +/// A diamond *reader* (`A`): reads the cell of a `diamond_target` (`B`) that is **passed in as an +/// already-resolved `Vc`** — so `A` records a forward (cell) dependency on `B` WITHOUT connecting +/// `B` as `A`'s child (a child edge is only created by *calling* a task; here `B` was called by +/// `diamond_root`). This decoupling is the crux: `B`'s only parent is `diamond_root`, so when the +/// root is collected BOTH `A` and `B` reach `parent_count 0` at the same time and cascade-collect +/// concurrently — while `A` still holds a forward-dep on `B` to scrub. +#[turbo_tasks::function] +async fn diamond_reader(target: ResolvedVc) -> Result> { + Ok(Vc::cell(1 + *target.await?)) +} + +/// The diamond root: parents both `A` and `B` as siblings, so collecting the root cascades a +/// `Collect` for every `A` and `B` at once — a `B` can therefore be collected before the +/// `Collect(A)` whose `CleanupOldEdges` opens it. +#[turbo_tasks::function] +async fn diamond_root(constant: ResolvedVc) -> Result> { + let mut sum = 0u32; + for index in 0..FANOUT { + let target = diamond_target(*constant, index).to_resolved().await?; + sum = sum.wrapping_add(*target.await?); + sum = sum.wrapping_add(*diamond_reader(*target).await?); + } + Ok(Vc::cell(sum)) +} + +/// A root that reads `reader()` only while the selector is `false`. Flipping the selector to `true` +/// drops `reader` (and, transitively, all `sd_leaf`s) from the live graph **without invalidating +/// them** — so they stay clean and retain their outgoing dependency edges, then all become +/// `parent_count 0` and collectible in a single pass. +#[turbo_tasks::function(operation, root)] +async fn select_reader( + selector: ResolvedVc, + constant: ResolvedVc, +) -> Result> { + let use_reader = !*selector.await?.get(); + let value = if use_reader { + *reader(*constant).await? + } else { + 0u32 + }; + Ok(Vc::cell(value)) +} + +/// Selector-gated root for the diamond fixture: reads `diamond_root` only while the selector is +/// `false`, so flipping to `true` disconnects the whole diamond subtree cleanly. +#[turbo_tasks::function(operation, root)] +async fn select_diamond( + selector: ResolvedVc, + constant: ResolvedVc, +) -> Result> { + let use_diamond = !*selector.await?.get(); + let value = if use_diamond { + *diamond_root(*constant).await? + } else { + 0u32 + }; + Ok(Vc::cell(value)) +} + +/// The **aggregation-graph rebalance** in GC: when the `reader` subtree is disconnected cleanly and +/// collected, GC must remove `reader` from each `sd_leaf`'s `upper` set so the leaves — now +/// parentless *and* upper-less — cascade-collect in the same pass. Without the rebalance a leaf +/// keeps a dangling `upper` edge to the deleted `reader`, fails `gc_maybe_collectible`, and leaks +/// until eviction hides it. +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn gc_rebalances_aggregation_and_cascades_in_one_pass() { + let (tt, _persistence_dir) = create_tt("gc_rebalances_aggregation_and_cascades_in_one_pass"); + let tt2 = tt.clone(); + + // Build the graph (selector=false: select_reader -> reader -> FANOUT sd_leaves). + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let selector = selector_op.read_strongly_consistent().await?; + + let constant_op = create_constant(); + let constant_vc = constant_op.resolve().strongly_consistent().await?; + + let output = select_reader(selector_vc, constant_vc); + output.read_strongly_consistent().await?; + + // Disconnect reader + all sd_leaves in one shot (no invalidation of the children). + selector.set(true); + output.read_strongly_consistent().await?; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + // Baseline resident count with the reader subtree disconnected but not yet collected. + let baseline = tt2.backend().resident_persistent_task_count_for_testing(); + + let collected = tt2.backend().gc_for_testing(&tt2); + tt2.backend().snapshot_and_evict_for_testing(&tt2); + let after = tt2.backend().resident_persistent_task_count_for_testing(); + + assert_eq!( + collected, + FANOUT as usize + 1, + "reader and all {FANOUT} sd_leaves should be collected in one pass" + ); + assert_eq!( + after, + baseline - (FANOUT as usize + 1), + "resident count must drop by exactly the collected subtree" + ); + + tt.stop_and_wait().await; +} + +/// The **residual resurrection** race that soft-deletion exists to close. Collecting `diamond_root` +/// cascades a concurrent `Collect` for every reader `A` and target `B`; `CleanupOldEdges(A)` opens +/// `B` via `ctx.task(B, Data)` to scrub the reverse edge. If `B` had already been removed from the +/// map, `ctx.task` would restore it from disk as a zombie and memory/disk would diverge. Because a +/// collected task stays resident until the tombstoning snapshot commits, `ctx.task` always finds a +/// resident entry. +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn gc_diamond_forward_dep_no_resurrection() { + let (tt, _persistence_dir) = create_tt("gc_diamond_forward_dep_no_resurrection"); + let tt2 = tt.clone(); + + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let selector = selector_op.read_strongly_consistent().await?; + + let constant_op = create_constant(); + let constant_vc = constant_op.resolve().strongly_consistent().await?; + + let output = select_diamond(selector_vc, constant_vc); + output.read_strongly_consistent().await?; + + // Disconnect the whole diamond subtree cleanly (no invalidation). + selector.set(true); + output.read_strongly_consistent().await?; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + let baseline = tt2.backend().resident_persistent_task_count_for_testing(); + + let collected = tt2.backend().gc_for_testing(&tt2); + tt2.backend().snapshot_and_evict_for_testing(&tt2); + let after = tt2.backend().resident_persistent_task_count_for_testing(); + + assert_eq!( + collected, + 2 * FANOUT as usize + 1, + "diamond_root + {FANOUT} readers + {FANOUT} targets should all be collected in one pass" + ); + assert_eq!( + after, + baseline - (2 * FANOUT as usize + 1), + "resident count must drop by exactly the collected subtree — a higher count means a \ + CleanupOldEdges scrub resurrected an already-collected diamond_target" + ); + + tt.stop_and_wait().await; +} + +/// Resurrection-on-connect: a task marked `deleted` by GC but reconnected **before** the +/// tombstoning snapshot must come back to life (marker cleared, made dirty, re-executed) rather +/// than being tombstoned/hard-deleted. +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn gc_resurrect_on_reconnect() { + let (tt, _persistence_dir) = create_tt("gc_resurrect_on_reconnect"); + let tt2 = tt.clone(); + let expected: u32 = (0..FANOUT).fold(0u32, |a, b| a.wrapping_add(b)); + + // Build (reader connected), then disconnect the reader subtree cleanly. + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let selector = selector_op.read_strongly_consistent().await?; + let constant_op = create_constant(); + let constant_vc = constant_op.resolve().strongly_consistent().await?; + + let output = select_reader(selector_vc, constant_vc); + assert_eq!(*output.read_strongly_consistent().await?, expected); + + selector.set(true); + output.read_strongly_consistent().await?; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + // Mark the disconnected subtree `deleted` (still resident — no snapshot yet). + let collected = tt2.backend().gc_for_testing(&tt2); + assert_eq!( + collected, + FANOUT as usize + 1, + "reader + {FANOUT} leaves should be marked collected" + ); + + // Reconnect the subtree (selector back to false) BEFORE any snapshot. Reading `reader` again + // connects it, which must resurrect it (and its leaves, as it re-reads them). + let tt3 = tt.clone(); + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let selector = selector_op.read_strongly_consistent().await?; + let constant_op = create_constant(); + let constant_vc = constant_op.resolve().strongly_consistent().await?; + selector.set(false); + let output = select_reader(selector_vc, constant_vc); + assert_eq!( + *output.read_strongly_consistent().await?, + expected, + "resurrected reader must recompute the correct value" + ); + let _ = &tt3; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + // A snapshot+evict now must NOT have tombstoned/hard-deleted the resurrected subtree. + tt2.backend().snapshot_and_evict_for_testing(&tt2); + let tt4 = tt.clone(); + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let constant_op = create_constant(); + let constant_vc = constant_op.resolve().strongly_consistent().await?; + let output = select_reader(selector_vc, constant_vc); + assert_eq!(*output.read_strongly_consistent().await?, expected); + let _ = &tt4; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + tt.stop_and_wait().await; +} + +static IMM_LEAF_EXECUTIONS: AtomicU32 = AtomicU32::new(0); + +/// An immutable leaf: no invalidator, no dependencies, not session dependent. +#[turbo_tasks::function] +fn imm_leaf(n: u32) -> Vc { + IMM_LEAF_EXECUTIONS.fetch_add(1, Ordering::Relaxed); + Vc::cell(n * 3) +} + +/// Reads `imm_leaf(n)` from inside a tracked task. +/// +/// The read runs here rather than at the top level of `run_once`, so the plain (eventually +/// consistent) `.await` is legal and deterministic: it is ordered by the task graph instead of +/// racing whatever the session happens to be doing. +#[turbo_tasks::function(operation, root)] +async fn read_imm_leaf(n: u32) -> Result> { + Ok(Vc::cell(*imm_leaf(n).await?)) +} + +#[turbo_tasks::function] +async fn imm_reader() -> Result> { + let mut sum = 0u32; + for index in 0..IMM_FANOUT { + sum = sum.wrapping_add(*imm_leaf(index).await?); + } + Ok(Vc::cell(sum)) +} + +const IMM_FANOUT: u32 = 8; + +/// Selector-gated root: reads `imm_reader` only while the selector is `false`, so flipping to +/// `true` disconnects the immutable subtree cleanly (no invalidation) and lets it reach +/// `parent_count 0`. +#[turbo_tasks::function(operation, root)] +async fn select_imm_reader(selector: ResolvedVc) -> Result> { + let use_reader = !*selector.await?.get(); + let value = if use_reader { + *imm_reader().await? + } else { + 0u32 + }; + Ok(Vc::cell(value)) +} + +/// An **immutable** task that is collected and reconnected before any snapshot must come back dirty +/// and re-execute, producing the correct value. Normally we don't allow re-execution of immutable +/// tasks, but resurrection is special +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn gc_resurrect_immutable_recomputes() { + let (tt, _persistence_dir) = create_tt("gc_resurrect_immutable_recomputes"); + let tt2 = tt.clone(); + let expected: u32 = (0..IMM_FANOUT).fold(0u32, |a, b| a.wrapping_add(b * 3)); + + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let selector = selector_op.read_strongly_consistent().await?; + + let output = select_imm_reader(selector_vc); + assert_eq!(*output.read_strongly_consistent().await?, expected); + + // Disconnect the immutable subtree without invalidating it. + selector.set(true); + assert_eq!(*output.read_strongly_consistent().await?, 0); + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + // Relative, not absolute: the counter is process-global and `imm_leaf` is shared with the other + // test in this file, which may have already run against a different backend instance. + let executions_before = IMM_LEAF_EXECUTIONS.load(Ordering::Relaxed); + assert!( + executions_before >= IMM_FANOUT, + "each immutable leaf should have executed at least once during the build, got \ + {executions_before}" + ); + + // Collect the disconnected subtree. The entries stay resident (no snapshot yet), so the tasks + // are soft-deleted rather than gone. + let collected = tt2.backend().gc_for_testing(&tt2); + assert_eq!( + collected, + IMM_FANOUT as usize + 1, + "imm_reader and all {IMM_FANOUT} immutable leaves should be collected" + ); + + // First route back to a collected leaf: read one *directly*, not through the selector root. + // This must recompute it rather than serve a stale value. Done before the reconnect below, + // while the subtree is still collected — afterwards the leaves are live again and a read would + // legitimately hit a fresh cell, proving nothing. + let tt_direct = tt.clone(); + let result = turbo_tasks::run_once(tt.clone(), async move { + assert_eq!(*read_imm_leaf(0).read_strongly_consistent().await?, 0); + let _ = &tt_direct; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + let executions_after_direct = IMM_LEAF_EXECUTIONS.load(Ordering::Relaxed); + assert!( + executions_after_direct > executions_before, + "reading a collected immutable leaf directly must re-execute it, but the execution count \ + did not move past {executions_before}" + ); + + // Reconnect BEFORE any snapshot. These tasks were never persisted, so there is nothing on disk + // to restore — the only way back to a correct value is re-execution. + let tt3 = tt.clone(); + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let selector = selector_op.read_strongly_consistent().await?; + selector.set(false); + let output = select_imm_reader(selector_vc); + assert_eq!( + *output.read_strongly_consistent().await?, + expected, + "a resurrected immutable task must recompute the correct value" + ); + let _ = &tt3; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + assert!( + IMM_LEAF_EXECUTIONS.load(Ordering::Relaxed) > executions_after_direct, + "the resurrected leaves must have re-executed, but the execution count did not move past \ + {executions_after_direct} — compared against the post-direct-read value so that read's \ + own re-execution cannot satisfy this" + ); + + // A snapshot + evict must not have tombstoned the resurrected subtree, and the restored data + // must survive the round trip. + tt2.backend().snapshot_and_evict_for_testing(&tt2); + let tt4 = tt.clone(); + let result = turbo_tasks::run_once(tt.clone(), async move { + let selector_op = create_selector(false); + let selector_vc = selector_op.resolve().strongly_consistent().await?; + let output = select_imm_reader(selector_vc); + assert_eq!(*output.read_strongly_consistent().await?, expected); + let _ = &tt4; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + tt.stop_and_wait().await; +} diff --git a/turbopack/crates/turbo-tasks-backend/tests/gc_stress.rs b/turbopack/crates/turbo-tasks-backend/tests/gc_stress.rs new file mode 100644 index 000000000000..bb251bf37748 --- /dev/null +++ b/turbopack/crates/turbo-tasks-backend/tests/gc_stress.rs @@ -0,0 +1,153 @@ +#![feature(arbitrary_self_types)] +#![feature(arbitrary_self_types_pointers)] +#![allow(clippy::needless_return)] // tokio macro-generated code doesn't respect this + +mod util; + +use std::sync::Arc; + +use anyhow::Result; +use turbo_tasks::{ResolvedVc, State, TurboTasks, Vc}; +use turbo_tasks_backend::TurboTasksBackend; + +use crate::util::create_tt; + +#[turbo_tasks::value(transparent)] +struct Generation(State); + +#[turbo_tasks::function(operation, root)] +fn create_generation() -> Vc { + Generation(State::new(0)).cell() +} + +/// A leaf keyed by (generation, index). Bumping the generation makes `wide_root` read an entirely +/// fresh set of these, disconnecting the whole previous generation. +#[turbo_tasks::function] +fn leaf(generation: u32, index: u32) -> Vc { + Vc::cell(generation.wrapping_mul(1000).wrapping_add(index)) +} + +/// A shared-dependency layer, so the disconnected garbage is a subtree (intermediate + leaf) and +/// the test exercises the cascade rather than single-node collection. +#[turbo_tasks::function] +async fn intermediate(generation: u32, index: u32) -> Result> { + Ok(Vc::cell(1 + *leaf(generation, index).await?)) +} + +const WIDTH: u32 = 24; + +/// Bumping the generation re-executes this and connects a fresh generation's worth of tasks, +/// disconnecting the entire previous generation (2*WIDTH tasks) as garbage. +#[turbo_tasks::function(operation, root)] +async fn wide_root(generation: ResolvedVc) -> Result> { + let generation = *generation.await?.get(); + let mut sum = 0u32; + for index in 0..WIDTH { + sum = sum.wrapping_add(*intermediate(generation, index).await?); + } + Ok(Vc::cell(sum)) +} + +/// Repeatedly swapping a wide set of common dependencies (as happens when a project is re-rooted or +/// its dependencies churn) must NOT grow the resident task set monotonically. Each round bumps the +/// generation (disconnecting the previous generation's 2*WIDTH tasks), then snapshots + GCs + +/// evicts, and the resident count must return to a flat baseline. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn gc_re_rooting_stays_flat() { + let (tt, _persistence_dir) = create_tt("gc_re_rooting_stays_flat"); + let tt2 = tt.clone(); + + const ROUNDS: u32 = 20; + + // Each round runs in its own `run_once` so the root's activeness is released before GC (a + // `run_once` root keeps everything it touched active until it returns). + async fn round(tt: &Arc>, gen_value: u32) -> (usize, usize) { + let tt_inner = tt.clone(); + turbo_tasks::run_once(tt.clone(), async move { + // `create_generation` is a cached root operation, so this returns the same task each + // round. + let generation_op = create_generation(); + let generation_vc = generation_op.resolve().strongly_consistent().await?; + if gen_value > 0 { + let generation = generation_op.read_strongly_consistent().await?; + generation.set(gen_value); + } + let output = wide_root(generation_vc); + output.read_strongly_consistent().await?; + let _ = &tt_inner; + anyhow::Ok(()) + }) + .await + .unwrap(); + + // GC runs BEFORE eviction (matching the production background cycle: snapshot -> GC -> + // evict), so the disconnected garbage is still resident when GC scans it. Running eviction + // first would drop the garbage to disk-only where the in-memory GC can't collect or + // tombstone it. + let collected = tt.backend().gc_for_testing(tt); + tt.backend().snapshot_and_evict_for_testing(tt); + // Measure the *persistent* resident count: GC only collects persistent tasks. Transient + // roots (each round's `run_once`/Once task) are never collected and accumulate + // independently of GC — including them would mask the real signal. + ( + collected, + tt.backend().resident_persistent_task_count_for_testing(), + ) + } + + // Warm up: build generation 0, then measure the steady-state baseline after one full cycle. + let (_, baseline) = round(&tt2, 0).await; + println!("baseline persistent resident after gen 0: {baseline}"); + + // Churn: swap the whole dependency set many times. + let mut max_resident = baseline; + let mut total_collected = 0usize; + for gen_value in 1..=ROUNDS { + let (collected, resident) = round(&tt2, gen_value).await; + println!("gen {gen_value}: collected={collected} persistent_resident={resident}"); + max_resident = max_resident.max(resident); + total_collected += collected; + } + + let no_gc_growth = baseline + (2 * WIDTH as usize) * (ROUNDS as usize); + println!( + "baseline={baseline} max_resident={max_resident} total_collected={total_collected} \ + no_gc_growth_would_be={no_gc_growth}" + ); + // Without GC the persistent resident set would climb by ~2*WIDTH per round (each generation's + // intermediates + leaves persisted forever), so a bound within a small constant of the baseline + // is only satisfiable if each generation's garbage subtree is actually collected. + assert!( + max_resident <= baseline + 2 * WIDTH as usize, + "persistent resident set grew too much across re-rooting (max={max_resident}, \ + baseline={baseline}); GC is not returning to a flat baseline" + ); + // Assert GC collected at least half of a full generation per round (slack for tasks that settle + // across passes), proving the flat baseline comes from GC collecting garbage — not merely from + // eviction hiding it on disk. + let expected_min_collected = (2 * WIDTH as usize) * (ROUNDS as usize) / 2; + assert!( + total_collected >= expected_min_collected, + "GC collected too little ({total_collected}); expected >= {expected_min_collected} across \ + {ROUNDS} re-rootings — GC is not doing the collection" + ); + + // The live graph must still compute correctly after all the churn. + let tt3 = tt.clone(); + let result = turbo_tasks::run_once(tt.clone(), async move { + let generation_op = create_generation(); + let generation_vc = generation_op.resolve().strongly_consistent().await?; + let output = wide_root(generation_vc); + // generation is ROUNDS; sum = WIDTH intermediates each = 1 + leaf(ROUNDS, i). + let expected: u32 = (0..WIDTH) + .map(|i| 1 + (ROUNDS.wrapping_mul(1000).wrapping_add(i))) + .fold(0u32, |a, b| a.wrapping_add(b)); + assert_eq!(*output.read_strongly_consistent().await?, expected); + let _ = &tt3; + anyhow::Ok(()) + }) + .await; + result.unwrap(); + + tt.stop_and_wait().await; +} diff --git a/turbopack/crates/turbo-tasks-backend/tests/util.rs b/turbopack/crates/turbo-tasks-backend/tests/util.rs index 8839262d0927..ae11e68fc375 100644 --- a/turbopack/crates/turbo-tasks-backend/tests/util.rs +++ b/turbopack/crates/turbo-tasks-backend/tests/util.rs @@ -27,6 +27,9 @@ fn open_tt_at(path: &Path, num_workers: usize) -> Arc); + + /// Removes a pin added by [`pin_task_for_gc`](Backend::pin_task_for_gc). + fn unpin_task_for_gc(&self, _task: TaskId, _turbo_tasks: &TurboTasks); + fn create_transient_task( &self, task_type: TransientTaskType, diff --git a/turbopack/crates/turbo-tasks/src/manager.rs b/turbopack/crates/turbo-tasks/src/manager.rs index 66a11c58368d..133b3cd46659 100644 --- a/turbopack/crates/turbo-tasks/src/manager.rs +++ b/turbopack/crates/turbo-tasks/src/manager.rs @@ -199,6 +199,13 @@ pub trait TurboTasksApi: TurboTasksCallApi + Sync + Send { ); fn mark_own_task_as_finished(&self, task: TaskId); + /// Pin a task against garbage collection. Delegates to + /// [`Backend::pin_task_for_gc`](crate::backend::Backend::pin_task_for_gc). + fn pin_task_for_gc(&self, task: TaskId); + + /// Removes a pin added by [`pin_task_for_gc`](TurboTasksApi::pin_task_for_gc). + fn unpin_task_for_gc(&self, task: TaskId); + fn connect_task(&self, task: TaskId); /// Wraps the given future in the current task. @@ -925,6 +932,18 @@ impl TurboTasks { self.backend.dispose_root_task(task_id, self); } + /// Pins a task against garbage collection (a transient, session-only reference). Balanced by + /// [`unpin_task_for_gc`](Self::unpin_task_for_gc). Used for references that escape the tracked + /// task graph — e.g. a `DetachedVc` holding an `OperationVc` across the NAPI boundary. + pub fn pin_task_for_gc(&self, task_id: TaskId) { + self.backend.pin_task_for_gc(task_id, self); + } + + /// Releases a pin added by [`pin_task_for_gc`](Self::pin_task_for_gc). + pub fn unpin_task_for_gc(&self, task_id: TaskId) { + self.backend.unpin_task_for_gc(task_id, self); + } + // TODO make sure that all dependencies settle before reading them /// Creates a new root task, that is only executed once. /// Dependencies will not invalidate the task. @@ -1898,6 +1917,14 @@ impl TurboTasksApi for TurboTasks { self.backend.mark_own_task_as_finished(task, self); } + fn pin_task_for_gc(&self, task: TaskId) { + self.backend.pin_task_for_gc(task, self); + } + + fn unpin_task_for_gc(&self, task: TaskId) { + self.backend.unpin_task_for_gc(task, self); + } + /// Creates a future that inherits the current task id and task state. The current global task /// will wait for this future to be dropped before exiting. fn spawn_detached_for_testing(&self, fut: Pin + Send + 'static>>) { @@ -2222,8 +2249,17 @@ pub fn unmark_top_level_task_may_leak_eventually_consistent_state() { } } +/// Pins the current task against garbage collection for the rest of the session, keeping it (and, +/// via the reachability it anchors, the values it produced) alive even if it becomes disconnected +/// from the live task graph. Use this when a value escapes the tracked graph — e.g. a `Vc` sent out +/// of a `spawn_detached` future across a channel, or handed across the NAPI boundary — so no +/// persistent parent lists it as a child and it would otherwise be collected. +/// +/// No-op outside a task context, and on backends without garbage collection. pub fn prevent_gc() { - // TODO implement garbage collection + if let Some(task) = current_task_if_available("prevent_gc") { + with_turbo_tasks(|tt| tt.pin_task_for_gc(task)); + } } pub fn emit(collectible: ResolvedVc) { diff --git a/turbopack/crates/turbo-tasks/src/task_dirty_cause.rs b/turbopack/crates/turbo-tasks/src/task_dirty_cause.rs index c17f4e8974ec..c9c49e5706f7 100644 --- a/turbopack/crates/turbo-tasks/src/task_dirty_cause.rs +++ b/turbopack/crates/turbo-tasks/src/task_dirty_cause.rs @@ -21,6 +21,9 @@ pub enum TaskDirtyCause { collectible_type: TraitTypeId, }, Invalidator, + /// Re-dirtied because a GC-soft-deleted task was resurrected by a new connection before its + /// hard-delete: its edges were scrubbed, so it must re-execute to rebuild them. + Resurrected, Unknown, } @@ -83,6 +86,7 @@ impl std::fmt::Display for TaskDirtyCause { ) } TaskDirtyCause::Invalidator => write!(f, "invalidator"), + TaskDirtyCause::Resurrected => write!(f, "resurrected"), TaskDirtyCause::Unknown => write!(f, "unknown"), } } diff --git a/turbopack/crates/turbo-tasks/src/vc/operation.rs b/turbopack/crates/turbo-tasks/src/vc/operation.rs index 76be6d47dfae..8b942413d969 100644 --- a/turbopack/crates/turbo-tasks/src/vc/operation.rs +++ b/turbopack/crates/turbo-tasks/src/vc/operation.rs @@ -192,6 +192,10 @@ impl OperationVc { { self.connect().into_trait_ref().strongly_consistent() } + + pub fn task_id(self) -> TaskId { + self.task + } } impl Copy for OperationVc where T: ?Sized {}