Repository: vercel/next.js
Pull request: [turbopack] Move empty snapshot filtering into take_snapshot

When snapshotting we create a number of iterators to encode every `shard` of the `modified` dashmap.

Some of these shards end up being empty and rarely they might all be empty.   To avoid the overhead of opening a write transaction we detect this prior to persisting.  Unfortunately this is done in a single threaded pass and can be expensive since there are often thousands of shards.

To improve this we move the filtering into shard production so it can happen concurrently.

This is a small improvment, from vercel-site builds timing the `Finished writing to filesystem cache` event we get

Before: 6.6s 7.8s 7.1s
After: 6.6s 6.5s 6.4s

So a small progression

Changed files:
filename: .claude/worktrees/keen-bassi
patch: @@ -0,0 +1 @@
+Subproject commit 0d1968d350efcb3890dda9f429053027013de536

filename: .claude/worktrees/modest-stonebraker
patch: @@ -0,0 +1 @@
+Subproject commit b9757fc8d466ee39d6b4437ea6d75c598f7f0cc3

filename: turbopack/crates/turbo-tasks-backend/src/backend/mod.rs
patch: @@ -1208,17 +1208,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                 )
             };
 
-        let snapshot = self.storage.take_snapshot(&process, &process_snapshot);
-
-        let task_snapshots = snapshot
-            .into_iter()
-            .filter_map(|iter| {
-                let mut iter = iter
-                    .filter_map(|item: SnapshotItem| (!item.is_empty()).then_some(item))
-                    .peekable();
-                iter.peek().is_some().then_some(iter)
-            })
-            .collect::<Vec<_>>();
+        // take_snapshot already filters empty items and empty shards in parallel
+        let task_snapshots = self.storage.take_snapshot(&process, &process_snapshot);
 
         swap_retain(&mut persisted_task_cache_log, |shard| !shard.is_empty());
 

filename: turbopack/crates/turbo-tasks-backend/src/backend/storage.rs
patch: @@ -1,5 +1,6 @@
 use std::{
     hash::Hash,
+    iter::Peekable,
     ops::{Deref, DerefMut},
     sync::{Arc, atomic::AtomicBool},
 };
@@ -10,6 +11,7 @@ use turbo_tasks::{FxDashMap, TaskId, parallel};
 
 use crate::{
     backend::storage_schema::TaskStorage,
+    backing_storage::SnapshotItem,
     database::key_value_database::KeySpace,
     utils::{
         dash_map_drop_contents::drop_contents,
@@ -111,16 +113,16 @@ impl Storage {
     /// receives an owned Box<TaskStorage> snapshot.
     /// Both callbacks receive a mutable scratch buffer that can be reused across iterations
     /// to avoid repeated allocations.
+    /// The returned iterators are guaranteed to be non-empty and only yield non-empty items.
     pub fn take_snapshot<
         'l,
-        R,
-        P: for<'a> Fn(TaskId, &'a TaskStorage, &mut TurboBincodeBuffer) -> R + Sync,
-        PS: Fn(TaskId, Box<TaskStorage>, &mut TurboBincodeBuffer) -> R + Sync,
+        P: for<'a> Fn(TaskId, &'a TaskStorage, &mut TurboBincodeBuffer) -> SnapshotItem + Sync,
+        PS: Fn(TaskId, Box<TaskStorage>, &mut TurboBincodeBuffer) -> SnapshotItem + Sync,
     >(
         &'l self,
         process: &'l P,
         process_snapshot: &'l PS,
-    ) -> Vec<SnapshotShard<'l, P, PS>> {
+    ) -> Vec<Peekable<SnapshotShard<'l, P, PS>>> {
         if !self.snapshot_mode() {
             self.start_snapshot();
         }
@@ -134,9 +136,9 @@ impl Storage {
             let mut modified: SmallVec<[TaskId; 4]> = SmallVec::new();
             {
                 // Take the snapshots from the modified map
-                let guard = shard.write();
-                // Safety: guard must outlive the iterator.
-                for bucket in unsafe { guard.iter() } {
+                let shard_guard = shard.write();
+                // Safety: shard_guard must outlive the iterator.
+                for bucket in unsafe { shard_guard.iter() } {
                     // Safety: the guard guarantees that the bucket is not removed and the ptr
                     // is valid.
                     let (key, shared_value) = unsafe { bucket.as_mut() };
@@ -152,22 +154,35 @@ impl Storage {
                         }
                     }
                 }
-                // Safety: guard must outlive the iterator.
-                drop(guard);
+                // Safety: shard_guard must outlive the iterator.
+                drop(shard_guard);
             }
+
+            // Early return for shards with no entries at all
+            if direct_snapshots.is_empty() && modified.is_empty() {
+                return None;
+            }
+
             /// How big of a buffer to allocate initially.  Based on metrics from a large
             /// application this should cover about 98% of values with no resizes
             const SCRATCH_BUFFER_SIZE: usize = 4096;
-            SnapshotShard {
+            let shard = SnapshotShard {
                 direct_snapshots,
                 modified,
                 storage: self,
                 guard: Some(guard.clone()),
                 process,
                 process_snapshot,
                 scratch_buffer: TurboBincodeBuffer::with_capacity(SCRATCH_BUFFER_SIZE),
-            }
+            };
+
+            // Peek to filter out shards that only produce empty items
+            let mut iter = shard.peekable();
+            iter.peek().is_some().then_some(iter)
         })
+        .into_iter()
+        .flatten()
+        .collect()
     }
 
     /// Start snapshot mode.
@@ -388,14 +403,12 @@ pub struct SnapshotShard<'l, P, PS> {
     scratch_buffer: TurboBincodeBuffer,
 }
 
-impl<'l, R, P, PS> Iterator for SnapshotShard<'l, P, PS>
+impl<'l, P, PS> SnapshotShard<'l, P, PS>
 where
-    P: for<'a> Fn(TaskId, &'a TaskStorage, &mut TurboBincodeBuffer) -> R + Sync,
-    PS: Fn(TaskId, Box<TaskStorage>, &mut TurboBincodeBuffer) -> R + Sync,
+    P: Fn(TaskId, &TaskStorage, &mut TurboBincodeBuffer) -> SnapshotItem + Sync,
+    PS: Fn(TaskId, Box<TaskStorage>, &mut TurboBincodeBuffer) -> SnapshotItem + Sync,
 {
-    type Item = R;
-
-    fn next(&mut self) -> Option<Self::Item> {
+    fn next_item(&mut self) -> Option<SnapshotItem> {
         if let Some((task_id, snapshot)) = self.direct_snapshots.pop() {
             return Some((self.process_snapshot)(
                 task_id,
@@ -425,6 +438,23 @@ where
                 }
             }
         }
+        None
+    }
+}
+
+impl<'l, P, PS> Iterator for SnapshotShard<'l, P, PS>
+where
+    P: Fn(TaskId, &TaskStorage, &mut TurboBincodeBuffer) -> SnapshotItem + Sync,
+    PS: Fn(TaskId, Box<TaskStorage>, &mut TurboBincodeBuffer) -> SnapshotItem + Sync,
+{
+    type Item = SnapshotItem;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        while let Some(item) = self.next_item() {
+            if !item.is_empty() {
+                return Some(item);
+            }
+        }
         self.guard = None;
         None
     }