kata-agent: Mount multi-layer EROFS partitions concurrently

This commit is just a enhancement without any functionality changes.

Replace the sequential loop in handle_multi_layer_erofs_group with
join_all-based concurrent mounting. Base device paths and mount
directories are pre-resolved before spawning futures to avoid lock
contention. On partial failure, successfully mounted layers are
unmounted and dm-verity devices cleaned up before propagating the
error.

Signed-off-by: Alex Lyn <alex.lyn@antgroup.com>

src/agent/src/storage/multi_layer_erofs.rs

@@ -356,16 +356,14 @@ pub async fn handle_multi_layer_erofs_group(
 
     let mut lower_mounts = Vec::new();
     let mut verity_devices = Vec::new();
-    let mut base_device_cache: HashMap<String, String> = HashMap::new();
 
-    for (index, erofs) in erofs_storages.iter().enumerate() {
-        let lower_mount = temp_base.join(format!("lower-{}", index));
-        fs::create_dir_all(&lower_mount).context(format!(
-            "failed to create lower mount dir {}",
-            lower_mount.display()
-        ))?;
-
-        let base_dev_path = if is_gpt_partitioned(erofs) {
+    // Pre-resolve all base device paths outside the parallel block to avoid
+    // contention on the sandbox lock and the HashMap.
+    let erofs_count = erofs_storages.len();
+    let mut resolved_base_devs: Vec<Option<String>> = Vec::with_capacity(erofs_count);
+    let mut base_device_cache: HashMap<String, String> = HashMap::with_capacity(erofs_count);
+    for erofs in &erofs_storages {
+        let base_dev = if is_gpt_partitioned(erofs) {
             Some(
                 base_device_cache
                     .entry(erofs.source.clone())
@@ -375,12 +373,97 @@ pub async fn handle_multi_layer_erofs_group(
         } else {
             None
         };
+        resolved_base_devs.push(base_dev);
+    }
 
-        let mount_info =
-            wait_and_mount_layer(erofs, &lower_mount, sandbox, &logger, base_dev_path).await?;
+    // Pre-create all lower mount directories so they are ready before
+    // concurrent mounting begins.
+    let lower_mount_paths: Vec<PathBuf> = (0..erofs_count)
+        .map(|index| temp_base.join(format!("lower-{}", index)))
+        .collect();
+    for lm in &lower_mount_paths {
+        fs::create_dir_all(lm)
+            .context(format!("failed to create lower mount dir {}", lm.display()))?;
+    }
+
+    // Concurrently create dm-verity devices and mount layers.
+    // Each layer has independent device nodes and mount points, so there is no
+    // ordering dependency during creation — only the final lowerdir list must
+    // preserve the original sort order.
+    let futures: Vec<_> = erofs_storages
+        .iter()
+        .enumerate()
+        .zip(resolved_base_devs.into_iter())
+        .map(|((index, erofs), base_dev_path)| {
+            let lower_mount = lower_mount_paths[index].clone();
+            let sandbox = Arc::clone(sandbox);
+            let logger = logger.clone();
+            async move {
+                wait_and_mount_layer(erofs, &lower_mount, &sandbox, &logger, base_dev_path)
+                    .await
+                    .map(|mount_info| (index, lower_mount, mount_info))
+            }
+        })
+        .collect();
+
+    info!(
+        logger,
+        "Starting concurrent mounting of EROFS layers";
+        "layer-count" => erofs_count,
+    );
+
+    let mut results: Vec<(usize, PathBuf, LayerMountInfo)> = Vec::with_capacity(erofs_count);
+    let mut failed_verity_devices: Vec<String> = Vec::new();
+    let mut failed_mount_points: Vec<PathBuf> = Vec::new();
+    let mut first_err: Option<anyhow::Error> = None;
+
+    for result in futures::future::join_all(futures).await {
+        match result {
+            Ok(t) => results.push(t),
+            Err(e) => {
+                if first_err.is_none() {
+                    first_err = Some(e);
+                }
+            }
+        }
+    }
+
+    if let Some(e) = first_err {
+        // Collect resources from successfully created layers
+        let mut sort_results: Vec<_> = results;
+        sort_results.sort_by_key(|(index, _, _)| *index);
+        for (_, lower_mount, mount_info) in sort_results {
+            if let Some(verity_dev) = mount_info.verity_device {
+                failed_verity_devices.push(verity_dev);
+            }
+            failed_mount_points.push(lower_mount);
+        }
+
+        // Unmount layers before destroying dm-verity devices.
+        for mp in failed_mount_points.iter().rev() {
+            if let Err(umount_err) = nix::mount::umount(mp) {
+                warn!(
+                    logger,
+                    "failed to unmount layer during concurrent-mount cleanup";
+                    "mount-point" => mp.display(),
+                    "error" => format!("{:#}", umount_err),
+                );
+            }
+        }
+
+        if !failed_verity_devices.is_empty() {
+            cleanup_dmverity_devices(&failed_verity_devices, &logger);
+        }
+
+        return Err(e.context("failed to mount one or more EROFS layers concurrently"));
+    }
+
+    // Restore the original sort order so that lowerdir precedence
+    // matches the host-supplied partition ordering.
+    results.sort_by_key(|(index, _, _)| *index);
+
+    for (_, lower_mount, mount_info) in results {
         lower_mounts.push(lower_mount);
-
-        // Collect dm-verity device for cleanup
         if let Some(verity_dev) = mount_info.verity_device {
             verity_devices.push(verity_dev);
         }