diff --git a/nativelink-config/src/stores.rs b/nativelink-config/src/stores.rs index 81b6179f92..f1f1b36fdc 100644 --- a/nativelink-config/src/stores.rs +++ b/nativelink-config/src/stores.rs @@ -844,6 +844,32 @@ pub struct FilesystemSpec { #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")] pub max_concurrent_writes: usize, + /// Maximum number of concurrent background file-deletion operations. + /// When an entry leaves the store, its backing file is unlinked by a + /// background task. Without a bound, an eviction storm (cache wipe or a + /// flood of overwrites) can spawn these without limit: memory grows with the + /// queued task frames and disk/inode reclamation lags behind what the async + /// runtime can schedule. Bounding it applies backpressure the same way + /// `max_concurrent_writes` does — once this many deletes are in flight, a + /// further delete runs synchronously inline in the evicting path instead of + /// spawning, which throttles eviction to delete throughput and never leaks + /// a file. + /// + /// Bounded by default, unlike `max_concurrent_writes`: an unbounded delete + /// backlog is a latent memory/inode blowup, and leaving the protection off + /// by default just moves that risk from a high-context code review to a + /// 4 AM config change nobody remembers to make. The default is generous + /// enough that normal steady-state eviction never reaches it (so the inline + /// path never fires in ordinary operation); it only engages under a genuine + /// storm, which is exactly when the backpressure is wanted. + /// A value of 0 is the explicit escape hatch for unlimited. + /// Default: 1024. + #[serde( + default = "default_max_concurrent_deletes", + deserialize_with = "convert_numeric_with_shellexpand" + )] + pub max_concurrent_deletes: usize, + /// When true, advise the kernel to drop the page cache for each blob after /// it is written or read (`posix_fadvise` with `POSIX_FADV_DONTNEED`). On /// real filesystems this takes a globally serialized, all-CPU kernel path @@ -1797,6 +1823,14 @@ const fn default_read_batching_dispatch_slots() -> usize { 4 } +/// Generous enough that normal eviction never reaches it (so the synchronous +/// inline delete path never fires in ordinary operation), low enough that the +/// in-flight backlog stays bounded to a few MB under a storm. Tune per +/// deployment; 0 disables the bound. See `FilesystemSpec::max_concurrent_deletes`. +const fn default_max_concurrent_deletes() -> usize { + 1024 +} + const fn default_read_batching_max_queued_bytes() -> u64 { 32 * 1024 * 1024 // 32 MiB. } diff --git a/nativelink-store/src/filesystem_store.rs b/nativelink-store/src/filesystem_store.rs index a7925e18ce..9ad1780f9f 100644 --- a/nativelink-store/src/filesystem_store.rs +++ b/nativelink-store/src/filesystem_store.rs @@ -102,6 +102,17 @@ pub struct SharedContext { // deleted (similar to how it is done in tests). #[metric(help = "Number of active drop spawns")] pub active_drop_spawns: AtomicU64, + /// Count of deletes that ran synchronously inline because the delete + /// semaphore was saturated. A nonzero, growing value means eviction is + /// outrunning background delete throughput — the backpressure signal. + #[metric(help = "Number of deletes performed inline due to delete backpressure")] + pub inline_deletes: AtomicU64, + /// Bounds the number of in-flight background file deletions. `None` means + /// unlimited (the historical behavior). When the permits are exhausted, + /// a drop deletes its file synchronously inline instead of spawning, so + /// an eviction storm applies backpressure rather than spawning without + /// limit. See `FilesystemSpec::max_concurrent_deletes`. + delete_semaphore: Option>, #[metric(help = "Path to the configured temp path")] temp_path: String, #[metric(help = "Path to the configured content path")] @@ -187,6 +198,40 @@ impl Drop for EncodedFilePath { let file_path = self.get_file_path().to_os_string(); let shared_context = self.shared_context.clone(); + + // Bound the number of in-flight background deletes. `drop` is sync, so + // we cannot `.await` a permit the way the write path does; instead we + // `try_acquire_owned` (never blocks) and, when the permits are + // exhausted, delete synchronously inline. That inline path is the + // backpressure: the evicting thread pays the unlink cost directly + // instead of queueing an unbounded number of tasks, and the file is + // always removed (never leaked). `None` keeps the historical + // unbounded-spawn behavior. + let delete_permit = match &shared_context.delete_semaphore { + Some(sem) => { + if let Ok(permit) = Arc::clone(sem).try_acquire_owned() { + Some(permit) + } else { + // Saturated: delete synchronously inline, which is the + // backpressure, then bail out (no spawn). + shared_context + .inline_deletes + .fetch_add(1, Ordering::Relaxed); + match std::fs::remove_file(&file_path) { + Ok(()) => { + debug!(?file_path, "File deleted inline (delete backpressure)"); + } + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + debug!(?file_path, "File already gone, nothing to delete"); + } + Err(err) => error!(?file_path, ?err, "Failed to delete file inline"), + } + return; + } + } + None => None, + }; + // .fetch_add returns previous value, so we add one to get approximate current value let current_active_drop_spawns = shared_context .active_drop_spawns @@ -198,6 +243,9 @@ impl Drop for EncodedFilePath { "Spawned a filesystem_delete_file" ); background_spawn!("filesystem_delete_file", async move { + // Hold the permit until the delete actually completes so the bound + // reflects deletes in flight, not merely scheduled. + let _delete_permit = delete_permit; match fs::remove_file(&file_path).await { Ok(()) => debug!(?file_path, "File deleted"), // The file already being gone is the desired end state of a @@ -1433,8 +1481,15 @@ impl FilesystemStore { enabled }; + let delete_semaphore = if spec.max_concurrent_deletes > 0 { + Some(Arc::new(Semaphore::new(spec.max_concurrent_deletes))) + } else { + None + }; let shared_context = Arc::new(SharedContext { active_drop_spawns: AtomicU64::new(0), + inline_deletes: AtomicU64::new(0), + delete_semaphore, temp_path: spec.temp_path.clone(), content_path: spec.content_path.clone(), }); diff --git a/nativelink-store/tests/filesystem_store_test.rs b/nativelink-store/tests/filesystem_store_test.rs index 7f659a1612..d92383df92 100644 --- a/nativelink-store/tests/filesystem_store_test.rs +++ b/nativelink-store/tests/filesystem_store_test.rs @@ -711,6 +711,62 @@ async fn file_gets_cleans_up_on_cache_eviction() -> Result<(), Error> { check_storage_dir_empty(&temp_path).await } +// Bounding background deletes with `max_concurrent_deletes` must not change the +// correctness of eviction: an evicted entry's backing file is still removed. +// This exercises the bounded (semaphore) delete path — permit acquired, held +// for the lifetime of the delete, then released. The default unbounded path is +// covered by the rest of this suite; the inline-overflow fallback (which fires +// only under saturation) is exercised under real load by the furnace via the +// `inline_deletes` metric. +#[nativelink_test] +async fn bounded_delete_semaphore_still_removes_evicted_files() -> Result<(), Error> { + static DELETES_FINISHED: AtomicU32 = AtomicU32::new(0); + struct LocalHooks {} + impl FileEntryHooks for LocalHooks { + fn on_drop(_file_entry: &Fe) { + DELETES_FINISHED.fetch_add(1, Ordering::Relaxed); + } + } + + let digest1 = DigestInfo::try_new(HASH1, VALUE1.len())?; + let digest2 = DigestInfo::try_new(HASH2, VALUE2.len())?; + let content_path = make_temp_path("content_path"); + let temp_path = make_temp_path("temp_path"); + + let store = Arc::new( + FilesystemStore::>::new(&FilesystemSpec { + content_path: content_path.clone(), + temp_path: temp_path.clone(), + eviction_policy: Some(EvictionPolicy { + max_count: 1, + ..Default::default() + }), + // The bound under test. With it set, the delete routes through the + // semaphore path rather than the unconditional spawn. + max_concurrent_deletes: 1, + block_size: 1, + read_buffer_size: 1, + ..Default::default() + }) + .await?, + ); + + // Insert, then evict digest1 by inserting digest2 (max_count == 1). + store.update_oneshot(digest1, VALUE1.into()).await?; + store.update_oneshot(digest2, VALUE2.into()).await?; + + // Wait for the eviction's background delete to finish. + loop { + if DELETES_FINISHED.load(Ordering::Relaxed) == 1 { + break; + } + tokio::task::yield_now().await; + } + + // The evicted file must be gone from the temp staging area. + check_storage_dir_empty(&temp_path).await +} + // Test to ensure that if we are holding a reference to `FileEntry` and the contents are // replaced, the `FileEntry` continues to use the old data. // `FileEntry` file contents should be immutable for the lifetime of the object. diff --git a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx b/web/apps/docs/content/docs/reference/nativelink-config/main.mdx index 5d97a6a58b..cb58419344 100644 --- a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx +++ b/web/apps/docs/content/docs/reference/nativelink-config/main.mdx @@ -5,7 +5,7 @@ full: true --- {/* AUTOGENERATED — do not edit by hand. - Source: nativelink-config @ main (6d6266cf) + Source: nativelink-config @ main (80678933f8) Regenerate from web/: bun --filter @nativelink/docs gen:config-reference */} @@ -830,6 +830,7 @@ See [`experimental_cloud_object_store`](#experimental_cloud_object_store) for de | `eviction_policy` | [EvictionPolicy](#evictionpolicy) | — | — | Policy used to evict items out of the store. Failure to set this value will cause items to never be removed from the store causing infinite memory usage. | | `block_size` | integer (uint64) | — | 4kb | The block size of the filesystem for the running machine value is used to determine an entry's actual size on disk consumed For a 4KB block size filesystem, a 1B file actually consumes 4KB | | `max_concurrent_writes` | integer (uint) | — | unlimited | Maximum number of concurrent write operations allowed. Each write involves streaming data to a temp file and calling `sync_all()`, which can saturate disk I/O when many writes happen simultaneously. Limiting concurrency prevents disk saturation from blocking the async runtime. A value of 0 means unlimited (no concurrency limit). | +| `max_concurrent_deletes` | integer (uint) | — | 1024 | Maximum number of concurrent background file-deletion operations. When an entry leaves the store, its backing file is unlinked by a background task. Without a bound, an eviction storm (cache wipe or a flood of overwrites) can spawn these without limit: memory grows with the queued task frames and disk/inode reclamation lags behind what the async runtime can schedule. Bounding it applies backpressure the same way `max_concurrent_writes` does — once this many deletes are in flight, a further delete runs synchronously inline in the evicting path instead of spawning, which throttles eviction to delete throughput and never leaks a file. | | `evict_page_cache` | boolean | — | false | When true, advise the kernel to drop the page cache for each blob after it is written or read (`posix_fadvise` with `POSIX_FADV_DONTNEED`). On real filesystems this takes a globally serialized, all-CPU kernel path that stalls on many-core hosts, and it evicts the page-cache tier that fast-disk deployments rely on. Leave off unless you specifically want to keep this store's I/O out of the page cache. | ## RefSpec