From 668a075168e3a664f6ce01fd4ee0fac1858ad8e9 Mon Sep 17 00:00:00 2001
From: Lann <lann.martin@fermyon.com>
Date: Tue, 10 May 2022 17:58:43 -0400
Subject: [PATCH] Limit parallelism copying local assets (#465)

Signed-off-by: Lann Martin <lann.martin@fermyon.com>
Signed-off-by: danbugs <danilochiarlone@gmail.com>
---
 crates/loader/src/bindle/assets.rs |  5 +----
 crates/loader/src/lib.rs           |  3 +++
 crates/loader/src/local/assets.rs  | 23 +++++++++++++----------
 3 files changed, 17 insertions(+), 14 deletions(-)

diff --git a/crates/loader/src/bindle/assets.rs b/crates/loader/src/bindle/assets.rs
index 14101f415b..1ccb0d83ef 100644
--- a/crates/loader/src/bindle/assets.rs
+++ b/crates/loader/src/bindle/assets.rs
@@ -12,9 +12,6 @@ use std::path::Path;
 use tokio::{fs, io::AsyncWriteExt};
 use tracing::log;
 
-/// Maximum number of assets to download in parallel
-const MAX_PARALLEL_COPIES: usize = 16;
-
 pub(crate) async fn prepare_component(
     reader: &BindleReader,
     bindle_id: &Id,
@@ -56,7 +53,7 @@ impl Copier {
 
     async fn copy_all(&self, parcels: &[Label], dir: impl AsRef<Path>) -> Result<()> {
         match stream::iter(parcels.iter().map(|p| self.copy(p, &dir)))
-            .buffer_unordered(MAX_PARALLEL_COPIES)
+            .buffer_unordered(crate::MAX_PARALLEL_ASSET_PROCESSING)
             .filter_map(|r| future::ready(r.err()))
             .map(|e| log::error!("{:?}", e))
             .count()
diff --git a/crates/loader/src/lib.rs b/crates/loader/src/lib.rs
index 66b8c6b0f0..12fbb77d8a 100644
--- a/crates/loader/src/lib.rs
+++ b/crates/loader/src/lib.rs
@@ -19,3 +19,6 @@ pub use local::from_file;
 
 /// Load a Spin application configuration from Bindle.
 pub use crate::bindle::from_bindle;
+
+/// Maximum number of assets to process in parallel
+pub(crate) const MAX_PARALLEL_ASSET_PROCESSING: usize = 16;
diff --git a/crates/loader/src/local/assets.rs b/crates/loader/src/local/assets.rs
index 304da5ae13..938b7b1f2c 100644
--- a/crates/loader/src/local/assets.rs
+++ b/crates/loader/src/local/assets.rs
@@ -1,8 +1,8 @@
 #![deny(missing_docs)]
 
 use crate::assets::{create_dir, ensure_all_under, ensure_under, to_relative};
-use anyhow::{anyhow, bail, Context, Result};
-use futures::future;
+use anyhow::{anyhow, bail, ensure, Context, Result};
+use futures::{future, stream, StreamExt};
 use spin_manifest::DirectoryMount;
 use std::path::{Path, PathBuf};
 use tracing::log;
@@ -178,16 +178,19 @@ fn collect_pattern(pattern: &str, rel: impl AsRef<Path>) -> Result<Vec<FileMount
 
 /// Copy all files to the mount directory.
 async fn copy_all(files: &[FileMount], dir: impl AsRef<Path>) -> Result<()> {
-    let res = future::join_all(files.iter().map(|f| copy(f, &dir))).await;
-    match res
-        .into_iter()
-        .filter_map(|r| r.err())
+    let copy_futures = files.iter().map(|f| copy(f, &dir));
+    let errors = stream::iter(copy_futures)
+        .buffer_unordered(crate::MAX_PARALLEL_ASSET_PROCESSING)
+        .filter_map(|r| future::ready(r.err()))
         .map(|e| log::error!("{:?}", e))
         .count()
-    {
-        0 => Ok(()),
-        n => bail!("Error copying assets: {} file(s) not copied", n),
-    }
+        .await;
+    ensure!(
+        errors == 0,
+        "Error copying assets: {} file(s) not copied",
+        errors
+    );
+    Ok(())
 }
 
 /// Copy a single file to the mount directory, setting it as read-only.