tokio + rayon

Author: kdy1

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

@@ -1887,7 +1887,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                     }
 
                     let this = self.clone();
-                    let snapshot = turbo_tasks::spawn_blocking(move || this.snapshot()).await;
+                    let snapshot = turbo_tasks::spawn_blocking_rayon(move || this.snapshot()).await;
                     if let Some((snapshot_start, new_data)) = snapshot {
                         last_snapshot = snapshot_start;
                         if new_data {

turbopack/crates/turbo-tasks-fs/src/retry.rs

@@ -29,7 +29,7 @@ where
 {
     let path = path.as_ref().to_owned();
 
-    turbo_tasks::spawn_blocking(move || {
+    turbo_tasks::spawn_blocking_tokio(move || {
         let mut attempt = 1;
 
         loop {

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

@@ -109,8 +109,8 @@ pub use manager::{
     CurrentCellRef, ReadConsistency, TaskPersistence, TurboTasks, TurboTasksApi,
     TurboTasksBackendApi, TurboTasksBackendApiExt, TurboTasksCallApi, Unused, UpdateInfo,
     dynamic_call, emit, mark_finished, mark_root, mark_session_dependent, mark_stateful,
-    prevent_gc, run_once, run_once_with_reason, spawn_blocking, spawn_thread, trait_call,
-    turbo_tasks, turbo_tasks_scope,
+    prevent_gc, run_once, run_once_with_reason, spawn_blocking_rayon, spawn_blocking_tokio,
+    spawn_thread, trait_call, turbo_tasks, turbo_tasks_scope,
 };
 pub use output::OutputContent;
 pub use raw_vc::{CellId, RawVc, ReadRawVcFuture, ResolveTypeError};

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

@@ -1784,7 +1784,9 @@ pub fn emit<T: VcValueTrait + ?Sized>(collectible: ResolvedVc<T>) {
     })
 }
 
-pub async fn spawn_blocking<T: Send + 'static>(func: impl FnOnce() -> T + Send + 'static) -> T {
+pub async fn spawn_blocking_tokio<T: Send + 'static>(
+    func: impl FnOnce() -> T + Send + 'static,
+) -> T {
     let turbo_tasks = turbo_tasks();
     let span = Span::current();
     let (result, duration, alloc_info) = tokio::task::spawn_blocking(|| {
@@ -1801,6 +1803,27 @@ pub async fn spawn_blocking<T: Send + 'static>(func: impl FnOnce() -> T + Send +
     result
 }
 
+pub async fn spawn_blocking_rayon<T: Send + 'static>(
+    func: impl FnOnce() -> T + Send + 'static,
+) -> T {
+    let (tx, rx) = tokio::sync::oneshot::channel();
+
+    let turbo_tasks = turbo_tasks();
+    let span = Span::current();
+    rayon::spawn(|| {
+        let _guard = span.entered();
+        let start = Instant::now();
+        let start_allocations = TurboMalloc::allocation_counters();
+        let r = turbo_tasks_scope(turbo_tasks, func);
+        let _ = tx.send((r, start.elapsed(), start_allocations.until_now()));
+    });
+
+    let (result, duration, alloc_info) = rx.await.unwrap();
+    capture_future::add_duration(duration);
+    capture_future::add_allocation_info(alloc_info);
+    result
+}
+
 pub fn spawn_thread(func: impl FnOnce() + Send + 'static) {
     let handle = Handle::current();
     let span = info_span!("thread").or_current();