Turbopack: show timing event for database compaction (#82369)

### What?

- show timing event for database compaction
- show timing event for persistent caching in dev too

Author: sokra

packages/next/src/server/dev/hot-reloader-turbopack.ts

@@ -258,7 +258,7 @@ export async function createHotReloaderTurbopack(
     }
   )
   backgroundLogCompilationEvents(project, {
-    eventTypes: ['StartupCacheInvalidationEvent'],
+    eventTypes: ['StartupCacheInvalidationEvent', 'TimingEvent'],
   })
   setBundlerFindSourceMapImplementation(
     getSourceMapFromTurbopack.bind(null, project, projectPath)

turbopack/crates/turbo-persistence/src/db.rs

@@ -650,7 +650,7 @@ impl TurboPersistence {
     /// files is above the given threshold. The coverage is the average number of SST files that
     /// need to be read to find a key. It also limits the maximum number of SST files that are
     /// merged at once, which is the main factor for the runtime of the compaction.
-    pub fn compact(&self, compact_config: &CompactConfig) -> Result<()> {
+    pub fn compact(&self, compact_config: &CompactConfig) -> Result<bool> {
         if self.read_only {
             bail!("Compaction is not allowed on a read only database");
         }
@@ -689,7 +689,8 @@ impl TurboPersistence {
             .context("Failed to compact database")?;
         }
 
-        if !new_meta_files.is_empty() {
+        let has_changes = !new_meta_files.is_empty();
+        if has_changes {
             self.commit(CommitOptions {
                 new_meta_files,
                 new_sst_files,
@@ -704,7 +705,7 @@ impl TurboPersistence {
 
         self.active_write_operation.store(false, Ordering::Release);
 
-        Ok(())
+        Ok(has_changes)
     }
 
     /// Internal function to perform a compaction.

turbopack/crates/turbo-tasks-backend/src/database/turbo.rs

@@ -1,15 +1,11 @@
-use std::{
-    cmp::max,
-    path::PathBuf,
-    sync::Arc,
-    thread::{JoinHandle, available_parallelism, spawn},
-};
+use std::{cmp::max, path::PathBuf, sync::Arc, thread::available_parallelism, time::Instant};
 
 use anyhow::{Ok, Result};
 use parking_lot::Mutex;
 use turbo_persistence::{
     ArcSlice, CompactConfig, KeyBase, StoreKey, TurboPersistence, ValueBuffer,
 };
+use turbo_tasks::{JoinHandle, message_queue::TimingEvent, spawn, turbo_tasks};
 
 use crate::database::{
     key_value_database::{KeySpace, KeyValueDatabase},
@@ -84,7 +80,7 @@ impl KeyValueDatabase for TurboKeyValueDatabase {
     ) -> Result<WriteBatch<'_, Self::SerialWriteBatch<'_>, Self::ConcurrentWriteBatch<'_>>> {
         // Wait for the compaction to finish
         if let Some(join_handle) = self.compact_join_handle.lock().take() {
-            join_handle.join().unwrap()?;
+            join_handle.join()?;
         }
         // Start a new write batch
         Ok(WriteBatch::concurrent(TurboWriteBatch {
@@ -100,23 +96,44 @@ impl KeyValueDatabase for TurboKeyValueDatabase {
     fn shutdown(&self) -> Result<()> {
         // Wait for the compaction to finish
         if let Some(join_handle) = self.compact_join_handle.lock().take() {
-            join_handle.join().unwrap()?;
+            join_handle.join()?;
         }
         // Compact the database on shutdown
-        self.db.compact(&CompactConfig {
-            max_merge_segment_count: if self.is_ci {
-                // Fully compact in CI to reduce cache size
-                usize::MAX
-            } else {
-                available_parallelism().map_or(4, |c| max(4, c.get()))
-            },
-            ..COMPACT_CONFIG
-        })?;
+        if self.is_ci {
+            // Fully compact in CI to reduce cache size
+            do_compact(&self.db, "Finished database compaction", usize::MAX)?;
+        } else {
+            // Compact with a reasonable limit in non-CI environments
+            do_compact(
+                &self.db,
+                "Finished database compaction",
+                available_parallelism().map_or(4, |c| max(4, c.get())),
+            )?;
+        }
         // Shutdown the database
         self.db.shutdown()
     }
 }
 
+fn do_compact(
+    db: &TurboPersistence,
+    message: &'static str,
+    max_merge_segment_count: usize,
+) -> Result<()> {
+    let start = Instant::now();
+    // Compact the database with the given max merge segment count
+    let ran = db.compact(&CompactConfig {
+        max_merge_segment_count,
+        ..COMPACT_CONFIG
+    })?;
+    if ran {
+        let elapsed = start.elapsed();
+        turbo_tasks()
+            .send_compilation_event(Arc::new(TimingEvent::new(message.to_string(), elapsed)));
+    }
+    Ok(())
+}
+
 pub struct TurboWriteBatch<'a> {
     batch: turbo_persistence::WriteBatch<WriteBuffer<'static>, 5>,
     db: &'a Arc<TurboPersistence>,
@@ -144,12 +161,12 @@ impl<'a> BaseWriteBatch<'a> for TurboWriteBatch<'a> {
         if let Some(compact_join_handle) = self.compact_join_handle {
             // Start a new compaction in the background
             let db = self.db.clone();
-            let handle = spawn(move || {
-                db.compact(&CompactConfig {
-                    max_merge_segment_count: available_parallelism()
-                        .map_or(4, |c| max(4, c.get() / 2)),
-                    ..COMPACT_CONFIG
-                })
+            let handle = spawn(async move {
+                do_compact(
+                    &db,
+                    "Finished database compaction",
+                    available_parallelism().map_or(4, |c| max(4, c.get() / 2)),
+                )
             });
             compact_join_handle.lock().replace(handle);
         }

turbopack/crates/turbo-tasks/src/lib.rs

@@ -118,7 +118,7 @@ use rustc_hash::FxHasher;
 pub use scope::scope;
 pub use serialization_invalidation::SerializationInvalidator;
 pub use shrink_to_fit::ShrinkToFit;
-pub use spawn::{JoinHandle, spawn, spawn_blocking, spawn_thread};
+pub use spawn::{JoinHandle, block_for_future, spawn, spawn_blocking, spawn_thread};
 pub use state::{State, TransientState};
 pub use task::{SharedReference, TypedSharedReference, task_input::TaskInput};
 pub use task_execution_reason::TaskExecutionReason;

turbopack/crates/turbo-tasks/src/manager.rs

@@ -1058,27 +1058,30 @@ impl<B: Backend + 'static> TurboTasks<B> {
     }
 
     pub async fn stop_and_wait(&self) {
-        self.backend.stopping(self);
-        self.stopped.store(true, Ordering::Release);
-        {
-            let listener = self
-                .event
-                .listen_with_note(|| || "wait for stop".to_string());
-            if self.currently_scheduled_tasks.load(Ordering::Acquire) != 0 {
-                listener.await;
+        turbo_tasks_future_scope(self.pin(), async move {
+            self.backend.stopping(self);
+            self.stopped.store(true, Ordering::Release);
+            {
+                let listener = self
+                    .event
+                    .listen_with_note(|| || "wait for stop".to_string());
+                if self.currently_scheduled_tasks.load(Ordering::Acquire) != 0 {
+                    listener.await;
+                }
             }
-        }
-        {
-            let listener = self.event_background.listen();
-            if self
-                .currently_scheduled_background_jobs
-                .load(Ordering::Acquire)
-                != 0
             {
-                listener.await;
+                let listener = self.event_background.listen();
+                if self
+                    .currently_scheduled_background_jobs
+                    .load(Ordering::Acquire)
+                    != 0
+                {
+                    listener.await;
+                }
             }
-        }
-        self.backend.stop(self);
+            self.backend.stop(self);
+        })
+        .await;
     }
 
     #[track_caller]

turbopack/crates/turbo-tasks/src/spawn.rs

@@ -8,7 +8,7 @@ use std::{
 
 use anyhow::Result;
 use futures::{FutureExt, ready};
-use tokio::runtime::Handle;
+use tokio::{runtime::Handle, task::block_in_place};
 use tracing::{Instrument, Span, info_span};
 use turbo_tasks_malloc::{AllocationInfo, TurboMalloc};
 
@@ -23,6 +23,12 @@ pub struct JoinHandle<T> {
     join_handle: tokio::task::JoinHandle<(Result<T, TurboTasksPanic>, Duration, AllocationInfo)>,
 }
 
+impl<T: Send + 'static> JoinHandle<T> {
+    pub fn join(self) -> T {
+        block_for_future(self)
+    }
+}
+
 impl<T> Future for JoinHandle<T> {
     type Output = T;
 
@@ -87,3 +93,14 @@ pub fn spawn_thread(func: impl FnOnce() + Send + 'static) {
         })
     });
 }
+
+/// Blocking waits for a future to complete. This blocks the current thread potentially staling
+/// other concurrent futures (but not other concurrent tasks). Try to avoid this method infavor of
+/// awaiting the future instead.
+pub fn block_for_future<T: Send>(future: impl Future<Output = T> + Send + 'static) -> T {
+    let handle = Handle::current();
+    block_in_place(|| {
+        let _span = info_span!("blocking").entered();
+        handle.block_on(future)
+    })
+}