Repository: vercel/next.js
Pull request: Turbopack: change AsyncModulesInfo to use keyed reads

### What?

Use keyed reads for the `AsyncModulesInfo` to reduce the number of tasks

Changed files:
filename: crates/next-api/src/server_actions.rs
patch: @@ -206,7 +206,9 @@ async fn build_manifest(
             &key,
             ActionManifestWorkerEntry {
                 module_id: loader_id.clone(),
-                is_async: *async_module_info.is_async(chunk_item.module()).await?,
+                is_async: async_module_info
+                    .is_async(chunk_item.module().to_resolved().await?)
+                    .await?,
                 exported_name: name.as_str(),
                 filename: filename.as_str(),
             },

filename: crates/next-core/src/next_manifests/client_reference_manifest.rs
patch: @@ -199,20 +199,21 @@ async fn build_manifest(
             .try_flat_join()
             .await?;
 
-        let async_modules = async_module_info
-            .is_async_multiple(Vc::cell(
-                client_references_ecmascript
-                    .iter()
-                    .flat_map(|(r, r_val)| {
-                        [
-                            ResolvedVc::upcast(*r),
-                            ResolvedVc::upcast(r_val.client_module),
-                            ResolvedVc::upcast(r_val.ssr_module),
-                        ]
+            let async_modules = client_references_ecmascript
+                .iter()
+                .flat_map(|(r, r_val)| {
+                    [
+                        ResolvedVc::upcast(*r),
+                        ResolvedVc::upcast(r_val.client_module),
+                        ResolvedVc::upcast(r_val.ssr_module),
+                    ]
+                }).map(async move |asset| {
+                    Ok(if async_module_info.is_async(asset).await? {
+                        Some(asset)
+                    } else {
+                        None
                     })
-                    .collect(),
-            ))
-            .await?;
+                }).try_flat_join().await?;
 
         async fn cached_chunk_paths(
             cache: &mut FxHashMap<ResolvedVc<Box<dyn OutputAsset>>, FileSystemPath>,

filename: turbopack/crates/turbopack-core/src/module_graph/async_module_info.rs
patch: @@ -3,7 +3,7 @@ use rustc_hash::FxHashSet;
 use turbo_tasks::{OperationVc, ResolvedVc, TryFlatJoinIterExt, Vc};
 
 use crate::{
-    module::{Module, Modules},
+    module::Module,
     module_graph::{GraphTraversalAction, ModuleGraph, ModuleGraphLayer},
 };
 
@@ -12,26 +12,12 @@ pub struct ModulesSet(FxHashSet<ResolvedVc<Box<dyn Module>>>);
 
 /// This lists all the modules that are async (self or transitively because they reference another
 /// module in this list).
-#[turbo_tasks::value(transparent)]
+#[turbo_tasks::value(transparent, cell = "keyed")]
 pub struct AsyncModulesInfo(FxHashSet<ResolvedVc<Box<dyn Module>>>);
 
-#[turbo_tasks::value_impl]
 impl AsyncModulesInfo {
-    #[turbo_tasks::function]
-    pub fn is_async(&self, module: ResolvedVc<Box<dyn Module>>) -> Vc<bool> {
-        Vc::cell(self.0.contains(&module))
-    }
-
-    #[turbo_tasks::function]
-    pub async fn is_async_multiple(&self, modules: ResolvedVc<Modules>) -> Result<Vc<ModulesSet>> {
-        Ok(Vc::cell(
-            modules
-                .await?
-                .iter()
-                .copied()
-                .filter(|m| self.0.contains(m))
-                .collect(),
-        ))
+    pub async fn is_async(self: Vc<Self>, module: ResolvedVc<Box<dyn Module>>) -> Result<bool> {
+        self.contains_key(&module).await
     }
 }