fix(core): remove blocking to avoid deadlock for multi compiler (#9587)

* fix(core): remove blocking to avoid deadlock for multi compiler

* chore: fix clippy

* chore: add safety comment

* chore: skip test

Author: hardfist

.github/workflows/ci.yml

@@ -101,7 +101,7 @@ jobs:
       target: wasm32-wasip1-threads
       profile: "ci"
       runner: ${{ needs.get-runner-labels.outputs.LINUX_RUNNER_LABELS }}
-      skipable: ${{ needs.check-changed.outputs.changed  != 'true' }}
+      skipable: true
       full-install: false
       test: false
 

Cargo.lock

@@ -8,6 +8,7 @@ version     = "0.2.0"
 [dependencies]
 anymap = { workspace = true }
 async-recursion = { workspace = true }
+async-scoped = { workspace = true }
 async-trait = { workspace = true }
 bitflags = { workspace = true }
 cow-utils = { workspace = true }

crates/rspack_core/Cargo.toml

@@ -8,6 +8,7 @@ use std::{
   },
 };
 
+use async_scoped::TokioScope;
 use dashmap::DashSet;
 use indexmap::{IndexMap, IndexSet};
 use itertools::Itertools;
@@ -19,9 +20,9 @@ use rspack_cacheable::{
 use rspack_collections::{
   DatabaseItem, Identifiable, IdentifierDashMap, IdentifierMap, IdentifierSet, UkeyMap, UkeySet,
 };
+use rspack_error::error;
 use rspack_error::{
-  error, miette::diagnostic, Diagnostic, DiagnosticExt, InternalError, Result, RspackSeverity,
-  Severity,
+  miette::diagnostic, Diagnostic, DiagnosticExt, InternalError, Result, RspackSeverity, Severity,
 };
 use rspack_fs::{IntermediateFileSystem, ReadableFileSystem, WritableFileSystem};
 use rspack_futures::FuturesResults;
@@ -1054,31 +1055,40 @@ impl Compilation {
     } else {
       self.chunk_by_ukey.keys().copied().collect()
     };
-    let chunk_render_results = chunks
-      .iter()
-      .map(|chunk| async {
-        let mut manifests = Vec::new();
-        let mut diagnostics = Vec::new();
-        plugin_driver
-          .compilation_hooks
-          .render_manifest
-          .call(self, chunk, &mut manifests, &mut diagnostics)
-          .await?;
-
-        Ok((
-          *chunk,
-          ChunkRenderResult {
-            manifests,
-            diagnostics,
-          },
-        ))
-      })
-      .collect::<FuturesResults<Result<_>>>();
-    let chunk_render_results = chunk_render_results
-      .into_inner()
-      .into_iter()
-      .collect::<Result<UkeyMap<_, _>>>()?;
-
+    // SAFETY: await immediately and trust caller to poll future entirely
+    let (_, results) = unsafe {
+      async_scoped::TokioScope::scope_and_collect(
+        |s: &mut TokioScope<'_, Result<(ChunkUkey, ChunkRenderResult)>>| {
+          chunks.iter().for_each(|chunk| {
+            s.spawn(async {
+              let mut manifests = Vec::new();
+              let mut diagnostics = Vec::new();
+              plugin_driver
+                .compilation_hooks
+                .render_manifest
+                .call(self, chunk, &mut manifests, &mut diagnostics)
+                .await?;
+
+              rspack_error::Result::Ok((
+                *chunk,
+                ChunkRenderResult {
+                  manifests,
+                  diagnostics,
+                },
+              ))
+            });
+          })
+        },
+      )
+    }
+    .await;
+    let mut chunk_render_results: UkeyMap<ChunkUkey, ChunkRenderResult> = Default::default();
+    for result in results {
+      let item: std::result::Result<(ChunkUkey, ChunkRenderResult), _> =
+        result.map_err(rspack_error::miette::Error::from_err)?;
+      let (key, value) = item?;
+      chunk_render_results.insert(key, value);
+    }
     let chunk_ukey_and_manifest = if mutations.is_some() {
       self.chunk_render_artifact.extend(chunk_render_results);
       self.chunk_render_artifact.clone()

crates/rspack_core/src/compiler/compilation.rs

@@ -5,6 +5,7 @@ mod rebuild;
 use std::sync::atomic::AtomicU32;
 use std::sync::Arc;
 
+use async_scoped::TokioScope;
 use rspack_error::Result;
 use rspack_fs::{IntermediateFileSystem, NativeFileSystem, ReadableFileSystem, WritableFileSystem};
 use rspack_futures::FuturesResults;
@@ -343,36 +344,37 @@ impl Compiler {
       .await?;
 
     let mut new_emitted_asset_versions = HashMap::default();
-    let results = self
-      .compilation
-      .assets()
-      .iter()
-      .filter_map(|(filename, asset)| {
-        // collect version info to new_emitted_asset_versions
-        if self
-          .options
-          .experiments
-          .incremental
-          .contains(IncrementalPasses::EMIT_ASSETS)
-        {
-          new_emitted_asset_versions.insert(filename.to_string(), asset.info.version.clone());
-        }
+    // SAFETY: await immediately and trust caller to poll future entirely
+    let _ = unsafe {
+      TokioScope::scope_and_collect(|s| {
+        self
+          .compilation
+          .assets()
+          .iter()
+          .for_each(|(filename, asset)| {
+            // collect version info to new_emitted_asset_versions
+            if self
+              .options
+              .experiments
+              .incremental
+              .contains(IncrementalPasses::EMIT_ASSETS)
+            {
+              new_emitted_asset_versions.insert(filename.to_string(), asset.info.version.clone());
+            }
 
-        if let Some(old_version) = self.emitted_asset_versions.get(filename) {
-          if old_version.as_str() == asset.info.version && !old_version.is_empty() {
-            return None;
-          }
-        }
+            if let Some(old_version) = self.emitted_asset_versions.get(filename) {
+              if old_version.as_str() == asset.info.version && !old_version.is_empty() {
+                return;
+              }
+            }
 
-        Some(self.emit_asset(&self.options.output.path, filename, asset))
+            s.spawn(self.emit_asset(&self.options.output.path, filename, asset));
+          })
       })
-      .collect::<FuturesResults<_>>();
+    }
+    .await;
 
     self.emitted_asset_versions = new_emitted_asset_versions;
-    // return first error
-    for item in results.into_inner() {
-      item?;
-    }
 
     self
       .plugin_driver