add concurrency limit to effects

Author: sokra

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

@@ -2,6 +2,7 @@ use std::{borrow::Cow, future::Future, mem::replace, panic, pin::Pin};
 
 use anyhow::{anyhow, Result};
 use auto_hash_map::AutoSet;
+use futures::StreamExt;
 use parking_lot::Mutex;
 use rustc_hash::FxHashSet;
 use tracing::{Instrument, Span};
@@ -17,6 +18,8 @@ use crate::{
     CollectiblesSource, NonLocalValue, ReadRef, ResolvedVc, TryJoinIterExt,
 };
 
+const APPLY_EFFECTS_CONCURRENCY_LIMIT: usize = 1024;
+
 /// A trait to emit a task effect as collectible. This trait only has one
 /// implementation, `EffectInstance` and no other implementation is allowed.
 /// The trait is private to this module so that no other implementation can be
@@ -168,16 +171,23 @@ pub async fn apply_effects(source: impl CollectiblesSource) -> Result<()> {
     }
     let span = tracing::info_span!("apply effects", count = effects.len());
     async move {
-        effects
+        let effects = effects
             .into_iter()
             .map(async |effect| {
                 let Some(effect) = ResolvedVc::try_downcast_type::<EffectInstance>(effect) else {
                     panic!("Effect must only be implemented by EffectInstance");
                 };
                 effect.await?.apply().await
             })
-            .try_join()
-            .await?;
+            // TODO remove this collect(), but rust was not happy with it...
+            .collect::<Vec<_>>();
+        // Limit the concurrency of effects,
+        // run them all even if an error occurs,
+        // report the first error.
+        let mut results = futures::stream::iter(effects).buffered(APPLY_EFFECTS_CONCURRENCY_LIMIT);
+        while let Some(result) = results.next().await {
+            result?;
+        }
         Ok(())
     }
     .instrument(span)
@@ -251,11 +261,20 @@ impl Effects {
     pub async fn apply(&self) -> Result<()> {
         let span = tracing::info_span!("apply effects", count = self.effects.len());
         async move {
-            self.effects
+            let effects = self
+                .effects
                 .iter()
                 .map(async |effect| effect.apply().await)
-                .try_join()
-                .await?;
+                // TODO remove this collect(), but rust was not happy with it...
+                .collect::<Vec<_>>();
+            // Limit the concurrency of effects,
+            // run them all even if an error occurs,
+            // report the first error.
+            let mut results =
+                futures::stream::iter(effects).buffered(APPLY_EFFECTS_CONCURRENCY_LIMIT);
+            while let Some(result) = results.next().await {
+                result?;
+            }
             Ok(())
         }
         .instrument(span)