Turbopack: add parallel execution helpers

Author: sokra

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

@@ -37,6 +37,8 @@
 #![feature(never_type)]
 #![feature(downcast_unchecked)]
 #![feature(ptr_metadata)]
+#![feature(sync_unsafe_cell)]
+#![feature(vec_into_raw_parts)]
 
 pub mod backend;
 mod capture_future;
@@ -64,13 +66,14 @@ mod no_move_vec;
 mod once_map;
 mod output;
 pub mod panic_hooks;
+pub mod parallel;
 pub mod persisted_graph;
 pub mod primitives;
 mod raw_vc;
 mod read_options;
 mod read_ref;
 pub mod registry;
-mod scope;
+pub mod scope;
 mod serialization_invalidation;
 pub mod small_duration;
 mod spawn;
@@ -115,7 +118,6 @@ pub use raw_vc::{CellId, RawVc, ReadRawVcFuture, ResolveTypeError};
 pub use read_options::ReadCellOptions;
 pub use read_ref::ReadRef;
 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};

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]
@@ -1675,6 +1678,10 @@ pub fn turbo_tasks() -> Arc<dyn TurboTasksApi> {
     TURBO_TASKS.with(|arc| arc.clone())
 }
 
+pub fn try_turbo_tasks() -> Option<Arc<dyn TurboTasksApi>> {
+    TURBO_TASKS.try_with(|arc| arc.clone()).ok()
+}
+
 pub fn with_turbo_tasks<T>(func: impl FnOnce(&Arc<dyn TurboTasksApi>) -> T) -> T {
     TURBO_TASKS.with(|arc| func(arc))
 }