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,29 +139,35 @@ 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| {
-                let env = env.clone();
-                async move {
-                    let mut path = PathBuf::from_str(&e.file_name().to_string_lossy()).unwrap();
-                    path.set_extension("");
+            .filter_map(|e| async move {
+                let mut path = PathBuf::from_str(&e.file_name().to_string_lossy()).unwrap();
+                path.set_extension("");
 
-                    let name = path.to_string_lossy().to_string();
+                let name = path.to_string_lossy().to_string();
 
-                    if name == "default" {
-                        return None;
-                    }
-
-                    Some(Cluster::get(&name, &env).await)
+                if name == "default" {
+                    return None;
                 }
+
+                Some(name)
             })
             .collect()
-            .await
+            .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
     }
 
     pub fn nodes(&self) -> &[Node] {

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 {
-        for node in cluster.nodes() {
-            run_command(node.talosctl_gen_config_command(&cluster)).await;
-        }
+    join_all(clusters.iter().map(|cluster| async move {
+        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,21 +29,20 @@ 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(|| {
-        Handle::current().block_on(async {
-            let res = client
-                .post("https://factory.talos.dev/schematics")
-                .body(content)
-                .send()
-                .await
+    // NOTE: This requires that the deserialization happens inside spawn_blocking
+    let response = Handle::current().block_on(async {
+        let res = client
+            .post("https://factory.talos.dev/schematics")
+            .body(content)
+            .send()
+            .await
+            .map_err(serde::de::Error::custom)?;
+
+        let response: Response =
+            serde_json::from_str(&res.text().await.map_err(serde::de::Error::custom)?)
                 .map_err(serde::de::Error::custom)?;
 
-            let response: Response =
-                serde_json::from_str(&res.text().await.map_err(serde::de::Error::custom)?)
-                    .map_err(serde::de::Error::custom)?;
-
-            Ok(response)
-        })
+        Ok(response)
     })?;
 
     Ok(response.id)