From bcfa15d0ed595084763651d673c2097da7e7a26b Mon Sep 17 00:00:00 2001 From: Luke Sandberg Date: Tue, 1 Sep 2026 22:27:43 -0700 Subject: [PATCH 1/2] Remove the worker thread for turbopack builds (#96862) Avoid spawning a worker thread to run turbopack The typical arguments for worker threads are that they isolate the JS heap and enable parallelism, but turbopack does both these things by default (native heap is isolated, the js deps are minor and have a lot of overlap with existing next server deps). There is also no substantial parallelism opportunity from worker threads that aren't immediately delivered by turbopacks execution model ## Benchmarks Measured with a paired A/B harness that alternates `head` and `base` builds so thermal drift and background load cancel out, discarding a warmup build per variant and clearing `.next` before each run. Both variants are the same commit except for this change. 35 paired iterations on a default `create-next-app` (App Router, TS), Node 20.11.1, Apple silicon. | Metric | this PR | canary | Diff | 95% CI | Wins | | --- | --- | --- | --- | --- | --- | | `next-build` total | 2478 ms | 2595 ms | **-117 ms (4.5%)** | [50, 183] | 30/35 | | `run-turbopack` | 1228 ms | 1335 ms | -107 ms (8.0%) | [62, 152] | 31/35 | | overhead (total - turbopack) | 1250 ms | 1260 ms | -10 ms (0.8%) | [-27, 46] | 20/35 | | wall clock | 2775 ms | 2888 ms | -112 ms (3.9%) | [45, 180] | 30/35 | | peak RSS | 1175 MiB | 1195 MiB | **-19 MiB (1.6%)** | [3.4, 35.3] | 22/35 | Significant by both a paired t-test and a Wilcoxon signed-rank test (p = 0.0001 for total build time, p = 0.025 for RSS). The Wilcoxon is included because a couple of slow builds per run would otherwise dominate the variance. Nearly all of the saving lands inside the `run-turbopack` span rather than around it, which is expected: that span wraps the worker spawn, the IPC to marshal the build across, and the teardown, so deleting the worker removes work from inside it. The win is proportionally smaller on larger apps, where compilation dominates. On `bench/basic-app` (~400 source modules, ~19s builds) the same ~115 ms is well inside the noise -- 12 paired iterations there showed no significant difference on any metric. So this is a real but small improvement that is only user-visible on small builds: | Metric | this PR | canary | Diff | 95% CI | | --- | --- | --- | --- | --- | | `next-build` total | 19313 ms | 19436 ms | -123 ms (0.6%) | [-256, 502] | | overhead (total - turbopack) | 1424 ms | 1415 ms | +9 ms (0.6% slower) | [-52, 70] | Caveats: single machine and two fixture shapes. Peak RSS is the build process's own footprint (`/usr/bin/time -l`), which includes the worker thread's isolate but not the separately spawned static-generation workers. A CI runner with fewer cores would plausibly show a larger effect. --- packages/next/src/build/index.ts | 12 +- .../next/src/build/turbopack-build/impl.ts | 115 +++--------------- .../next/src/build/turbopack-build/index.ts | 93 +------------- .../lib/turbopack/compilation-events.ts | 15 +-- .../next/src/trace/report/to-json-build.ts | 1 - .../trace-build-file/trace-build-file.test.ts | 4 +- .../cpu-profiling/cpu-profiling-build.test.ts | 8 +- .../prerender-worker-threads.test.ts | 31 ++--- 8 files changed, 50 insertions(+), 229 deletions(-) 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) }) }) From 8ea76d64ca3931c1beccceb15d32df5d770f4957 Mon Sep 17 00:00:00 2001 From: Luke Sandberg Date: Tue, 1 Sep 2026 22:30:32 -0700 Subject: [PATCH 2/2] turbo-tasks-backend: parent_count-driven garbage collection (#97282) Adds garbage collection, built on the `parent_count` reference counting ### How collection works * we scan all storage shards for collectible tasks to seed the sweep * For each collectible task we 1. mark it as deleted 2. remove all outgoing edges 3. queue new tasks for deletion if the edge removals triggered it For removing edges we rely on the existing `CleanupOldEdges` operation, though it is enhanced to collect the tasks that become collectible as edges are removed Deletion markers are transformed into tombstones by persistence and then dropped from memory by eviction. ### Coordinator changes GC mutates the graph, so we exclude other tasks from running while it works. When gc completes we hand off to persistence which switches to the existing copy on write mode. The subtle piece is `GcPhase::into_snapshot`: GC has to hand its exclusion directly to the snapshot that tombstones what it collected. Swapping the flags under one lock leaves no window where an operation could start and resurrect a just-collected task. ### Enabling it Off by default; `TURBO_ENGINE_GC` turns it on, also requires eviction to be enabled (otherwise deleted tasks persist in RAM) `prevent_gc()` becomes real, pinning the calling task through `transient_ref_count`. --- .../crates/turbo-tasks-backend/benches/gc.rs | 161 ++++++ .../crates/turbo-tasks-backend/benches/mod.rs | 3 +- .../turbo-tasks-backend/src/backend/gc.rs | 191 ++++++++ .../turbo-tasks-backend/src/backend/mod.rs | 156 ++++-- .../backend/operation/aggregation_update.rs | 85 +++- .../backend/operation/cleanup_old_edges.rs | 67 ++- .../src/backend/operation/connect_child.rs | 55 ++- .../src/backend/operation/connect_children.rs | 4 +- .../src/backend/operation/invalidate.rs | 25 +- .../src/backend/operation/mod.rs | 157 +++++- .../src/backend/operation/update_cell.rs | 5 +- .../src/backend/snapshot_coordinator.rs | 117 +++-- .../src/backend/storage.rs | 125 ++++- .../src/backend/storage_schema.rs | 51 +- .../src/backing_storage.rs | 3 - .../crates/turbo-tasks-backend/tests/debug.rs | 2 + .../tests/gc_collection.rs | 202 ++++++++ .../tests/gc_resurrection.rs | 462 ++++++++++++++++++ .../turbo-tasks-backend/tests/gc_stress.rs | 153 ++++++ .../crates/turbo-tasks-backend/tests/util.rs | 3 + turbopack/crates/turbo-tasks/src/backend.rs | 8 + turbopack/crates/turbo-tasks/src/manager.rs | 38 +- .../turbo-tasks/src/task_dirty_cause.rs | 4 + .../crates/turbo-tasks/src/vc/operation.rs | 4 + 24 files changed, 1938 insertions(+), 143 deletions(-) create mode 100644 turbopack/crates/turbo-tasks-backend/benches/gc.rs create mode 100644 turbopack/crates/turbo-tasks-backend/src/backend/gc.rs create mode 100644 turbopack/crates/turbo-tasks-backend/tests/gc_collection.rs create mode 100644 turbopack/crates/turbo-tasks-backend/tests/gc_resurrection.rs create mode 100644 turbopack/crates/turbo-tasks-backend/tests/gc_stress.rs 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 {}