feat: Improve concurrency

src/cluster.rs

@@ -4,6 +4,7 @@ use std::str::FromStr;
 
 use async_walkdir::WalkDir;
 use futures::StreamExt;
+use futures::future::join_all;
 use futures::stream::FuturesUnordered;
 use minijinja::Environment;
 use optional_struct::{Applicable, optional_struct};
@@ -87,7 +88,11 @@ impl Cluster {
 
         let default: OptionalClusterDeserialize = if path.exists() {
             let content = tokio::fs::read_to_string(path).await.unwrap();
-            serde_yaml::from_str(&content).unwrap()
+
+            // The deserialization makes a blocking web request
+            tokio::task::spawn_blocking(move || serde_yaml::from_str(&content).unwrap())
+                .await
+                .unwrap()
         } else {
             Default::default()
         };
@@ -134,15 +139,13 @@ impl Cluster {
     pub async fn get_all() -> Vec<Cluster> {
         let env = PathEnvironment::new_patches();
 
-        WalkDir::new(get_talos_path().join("clusters"))
+        let names: Vec<_> = WalkDir::new(get_talos_path().join("clusters"))
             .filter_map(|e| async { e.ok() })
             .filter(|e| {
                 let e = e.clone();
                 async move { e.metadata().await.unwrap().is_file() }
             })
-            .filter_map(|e| {
+            .filter_map(|e| async move {
-                let env = env.clone();
-                async move {
                 let mut path = PathBuf::from_str(&e.file_name().to_string_lossy()).unwrap();
                 path.set_extension("");
 
@@ -152,10 +155,18 @@ impl Cluster {
                     return None;
                 }
 
-                    Some(Cluster::get(&name, &env).await)
+                Some(name)
-                }
             })
             .collect()
+            .await;
+
+        // Get the clusters concurrently
+        join_all(names.iter().map(|name| {
+            let env = env.clone();
+            let name = name.clone();
+
+            async move { Cluster::get(&name, &env).await }
+        }))
         .await
     }
 

src/main.rs

@@ -8,6 +8,7 @@ use crete::cluster::Cluster;
 use crete::environment::PathEnvironment;
 use crete::{get_configs_path, get_repo_path, set_repo_path};
 use futures::StreamExt;
+use futures::future::join_all;
 use minijinja::context;
 use thiserror::Error;
 use tokio::process::Command;
@@ -69,13 +70,25 @@ async fn generate(opts: &GlobalOpts) -> Result<(), Error> {
     tokio::fs::create_dir_all(&path).await.unwrap();
 
     // Generate config files
-    for cluster in clusters {
+    join_all(clusters.iter().map(|cluster| async move {
-        for node in cluster.nodes() {
+        join_all(cluster.nodes().iter().map(|node| {
+            let cluster = cluster.clone();
+            async move {
                 run_command(node.talosctl_gen_config_command(&cluster)).await;
             }
+        }))
+        .await;
 
+        // Run these commands in parallel with the the node commands
+        // NOTE: These do need to happen in the correct order
         run_command(cluster.talosctl_gen_config_command()).await;
         run_command(cluster.talosctl_add_endpoint_command()).await;
+    }))
+    .await;
+
+    // Merge all the cluster talosconfig files into a single file
+    // Since this all modifies the same output file this has to happen sequentially
+    for cluster in clusters {
         run_command(cluster.talosctl_merge_command()).await;
     }
 

src/schematic.rs

@@ -29,8 +29,8 @@ where
     // We can't use async directly inside a custom deserializer, and using the blocking client also
     // does not work as it will try to start tokio inside of tokio.
     // So we have do this to work around it.
-    let response = tokio::task::block_in_place(|| {
+    // NOTE: This requires that the deserialization happens inside spawn_blocking
-        Handle::current().block_on(async {
+    let response = Handle::current().block_on(async {
         let res = client
             .post("https://factory.talos.dev/schematics")
             .body(content)
@@ -43,7 +43,6 @@ where
                 .map_err(serde::de::Error::custom)?;
 
         Ok(response)
-        })
     })?;
 
     Ok(response.id)