feat: Improve concurrency

Cargo.lock

@@ -52,6 +52,57 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "async-channel"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2"
+dependencies = [
+ "concurrent-queue",
+ "event-listener-strategy",
+ "futures-core",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-fs"
+version = "2.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8034a681df4aed8b8edbd7fbe472401ecf009251c8b40556b304567052e294c5"
+dependencies = [
+ "async-lock",
+ "blocking",
+ "futures-lite",
+]
+
+[[package]]
+name = "async-lock"
+version = "3.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311"
+dependencies = [
+ "event-listener",
+ "event-listener-strategy",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-task"
+version = "4.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de"
+
+[[package]]
+name = "async-walkdir"
+version = "2.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "37672978ae0febce7516ae0a85b53e6185159a9a28787391eb63fc44ec36037d"
+dependencies = [
+ "async-fs",
+ "futures-lite",
+ "thiserror",
+]
+
 [[package]]
 name = "atomic-waker"
 version = "1.1.2"
@@ -85,6 +136,19 @@ dependencies = [
  "generic-array",
 ]
 
+[[package]]
+name = "blocking"
+version = "1.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a70e4329df6cb94385eed412ec92375c3cdd8a6e502493d1229b6414e4036dfa"
+dependencies = [
+ "async-channel",
+ "async-task",
+ "futures-io",
+ "futures-lite",
+ "piper",
+]
+
 [[package]]
 name = "bstr"
 version = "1.13.1"
@@ -187,6 +251,15 @@ version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
 
+[[package]]
+name = "concurrent-queue"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "core-foundation"
 version = "0.10.1"
@@ -216,9 +289,11 @@ dependencies = [
 name = "crete"
 version = "0.1.3"
 dependencies = [
+ "async-walkdir",
  "byteorder",
  "clap",
  "clap_complete",
+ "futures",
  "gix-discover",
  "minijinja",
  "netascii",
@@ -232,9 +307,14 @@ dependencies = [
  "serde_yaml",
  "thiserror",
  "tokio",
- "walkdir",
 ]
 
+[[package]]
+name = "crossbeam-utils"
+version = "0.8.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6"
+
 [[package]]
 name = "crypto-common"
 version = "0.1.7"
@@ -325,6 +405,26 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "event-listener"
+version = "5.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2"
+dependencies = [
+ "parking",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "event-listener-strategy"
+version = "0.5.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93"
+dependencies = [
+ "event-listener",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "faster-hex"
 version = "0.10.0"
@@ -371,6 +471,21 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "futures"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
 [[package]]
 name = "futures-channel"
 version = "0.3.34"
@@ -387,12 +502,36 @@ version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"
 
+[[package]]
+name = "futures-executor"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
 [[package]]
 name = "futures-io"
 version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed"
 
+[[package]]
+name = "futures-lite"
+version = "2.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad"
+dependencies = [
+ "fastrand",
+ "futures-core",
+ "futures-io",
+ "parking",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "futures-macro"
 version = "0.3.34"
@@ -422,6 +561,7 @@ version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc"
 dependencies = [
+ "futures-channel",
  "futures-core",
  "futures-io",
  "futures-macro",
@@ -1218,6 +1358,12 @@ dependencies = [
  "syn 2.0.119",
 ]
 
+[[package]]
+name = "parking"
+version = "2.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
+
 [[package]]
 name = "parking_lot"
 version = "0.12.5"
@@ -1253,6 +1399,17 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
 
+[[package]]
+name = "piper"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1"
+dependencies = [
+ "atomic-waker",
+ "fastrand",
+ "futures-io",
+]
+
 [[package]]
 name = "pkg-config"
 version = "0.3.34"

Cargo.toml

@@ -14,9 +14,11 @@ schemars = { version = "1.2.1", features = ["semver1"] }
 serde_json = "1.0.149"
 
 [dependencies]
+async-walkdir = "2.1.0"
 byteorder = "1.5.0"
 clap = { version = "4.6.0", features = ["derive"] }
 clap_complete = "4.6.0"
+futures = "0.3.34"
 gix-discover = { version = "0.55.0", features = ["sha1"] }
 minijinja = { version = "2.18.0", features = ["json", "loader"] }
 netascii = "0.1.0"
@@ -34,7 +36,6 @@ serde_json = { workspace = true }
 serde_yaml = "0.9.34"
 thiserror = "2.0.18"
 tokio = { version = "1.52.1", features = ["full"] }
-walkdir = "2.5.0"
 
 [lib]
 test = false

src/cluster.rs

@@ -1,13 +1,16 @@
 use std::net::Ipv4Addr;
 use std::path::PathBuf;
-use std::process::Command;
 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};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
-use walkdir::WalkDir;
+use tokio::process::Command;
 
 use crate::environment::PathEnvironment;
 use crate::node::{Node, OptionalNodeDeserialize};
@@ -68,7 +71,7 @@ struct ClusterDeserialize {
     base: Base,
 }
 
-#[derive(Debug, Serialize)]
+#[derive(Debug, Serialize, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct Cluster {
     pub(crate) name: String,
@@ -80,19 +83,23 @@ pub struct Cluster {
 }
 
 impl Cluster {
-    pub(crate) fn get(cluster_name: &str, env: &Environment) -> Self {
+    pub(crate) async fn get(cluster_name: &str, env: &Environment<'static>) -> Self {
         let path = get_talos_path().join("clusters").join("default.yaml");
 
         let default: OptionalClusterDeserialize = if path.exists() {
-            let content = std::fs::read_to_string(path).unwrap();
-            serde_yaml::from_str(&content).unwrap()
+            let content = tokio::fs::read_to_string(path).await.unwrap();
+
+            // The deserialization makes a blocking web request
+            tokio::task::spawn_blocking(move || serde_yaml::from_str(&content).unwrap())
+                .await
+                .unwrap()
         } else {
             Default::default()
         };
 
         let mut path = get_talos_path().join("clusters").join(cluster_name);
         path.add_extension("yaml");
-        let content = std::fs::read_to_string(path).unwrap();
+        let content = tokio::fs::read_to_string(path).await.unwrap();
 
         let mut cluster: OptionalClusterDeserialize = serde_yaml::from_str(&content).unwrap();
 
@@ -122,20 +129,23 @@ impl Cluster {
 
         let nodes = nodes
             .iter()
-            .map(|name| Node::get(name, env, &cluster, &default, &base))
-            .collect();
+            .map(|name| async { Node::get(name, env, &cluster, &default, &base).await });
+
+        let nodes = FuturesUnordered::from_iter(nodes).collect().await;
 
         Self { nodes, ..cluster }
     }
 
-    pub fn get_all() -> Vec<Cluster> {
+    pub async fn get_all() -> Vec<Cluster> {
         let env = PathEnvironment::new_patches();
 
-        WalkDir::new(get_talos_path().join("clusters"))
-            .into_iter()
-            .filter_map(|e| e.ok())
-            .filter(|e| e.metadata().unwrap().is_file())
-            .filter_map(|e| {
+        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| async move {
                 let mut path = PathBuf::from_str(&e.file_name().to_string_lossy()).unwrap();
                 path.set_extension("");
 
@@ -145,9 +155,19 @@ impl Cluster {
                     return None;
                 }
 
-                Some(Cluster::get(&name, &env))
+                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
     }
 
     pub fn nodes(&self) -> &[Node] {

src/environment.rs

@@ -2,15 +2,17 @@ use std::net::Ipv4Addr;
 use std::ops::Deref;
 use std::path::{Path, PathBuf};
 
+use async_walkdir::WalkDir;
+use futures::{Stream, StreamExt};
 use minijinja::value::ViaDeserialize;
 use minijinja::{AutoEscape, Environment, path_loader};
-use walkdir::WalkDir;
 
 use crate::{get_configs_path, get_talos_path};
 
 /// Transparent wrapper around minijinja::Environment that loads templates from a path and
 /// configures better defaults. It also implements IntoIter, making it possible to iterate over all
 /// the templates.
+#[derive(Clone)]
 pub struct PathEnvironment<'a> {
     env: Environment<'a>,
     path: PathBuf,
@@ -76,6 +78,23 @@ impl<'a> PathEnvironment<'a> {
     pub(crate) fn new_patches() -> Self {
         Self::new(&get_talos_path().join("patches"))
     }
+
+    pub fn files(&self) -> impl Stream<Item = String> {
+        // Find all templates in path
+        WalkDir::new(&self.path)
+            .filter_map(|e| async { e.ok() })
+            .filter(|e| {
+                let e = e.clone();
+                async move { e.metadata().await.unwrap().is_file() }
+            })
+            .map(move |e| {
+                e.path()
+                    .strip_prefix(&self.path)
+                    .expect("All paths should have prefix")
+                    .to_string_lossy()
+                    .to_string()
+            })
+    }
 }
 
 // Make PathEnvironment act like a normal Environment transparently
@@ -86,25 +105,3 @@ impl<'a> Deref for PathEnvironment<'a> {
         &self.env
     }
 }
-
-// Iterate over all the files in the path
-impl<'a, 'b> IntoIterator for &'b PathEnvironment<'a> {
-    type Item = String;
-
-    type IntoIter = impl Iterator<Item = Self::Item>;
-
-    fn into_iter(self) -> Self::IntoIter {
-        // Find all templates in path
-        WalkDir::new(&self.path)
-            .into_iter()
-            .filter_map(|e| e.ok())
-            .filter(|e| e.metadata().unwrap().is_file())
-            .map(|e| {
-                e.path()
-                    .strip_prefix(&self.path)
-                    .expect("All paths should have prefix")
-                    .to_string_lossy()
-                    .to_string()
-            })
-    }
-}

src/lib.rs

@@ -1,4 +1,4 @@
-#![feature(impl_trait_in_assoc_type, path_absolute_method)]
+#![feature(path_absolute_method)]
 use std::path::{Path, PathBuf};
 use std::sync::OnceLock;
 

src/main.rs

@@ -1,14 +1,17 @@
 mod cli;
 
-use std::process::Command;
+use std::pin::pin;
 
 use clap::{CommandFactory, Parser};
 use clap_complete::{Shell, generate as generate_complete};
 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;
 
 use crate::cli::{Cli, Commands, GlobalOpts};
 
@@ -18,8 +21,8 @@ enum Error {
     NoClustersFound,
 }
 
-fn run_command(mut command: Command) {
-    let output = command.output().unwrap();
+async fn run_command(mut command: Command) {
+    let output = command.output().await.unwrap();
     if !output.stdout.is_empty() {
         println!("{}", String::from_utf8_lossy(&output.stdout).trim_end());
     }
@@ -28,24 +31,25 @@ fn run_command(mut command: Command) {
     }
 }
 
-fn generate(opts: &GlobalOpts) -> Result<(), Error> {
+async fn generate(opts: &GlobalOpts) -> Result<(), Error> {
     set_repo_path(&opts.repo);
 
-    let clusters = Cluster::get_all();
+    let clusters = Cluster::get_all().await;
 
     if clusters.is_empty() {
         return Err(Error::NoClustersFound);
     }
 
     let path = opts.repo.join("rendered");
-    if path.exists() {
-        std::fs::remove_dir_all(&path).unwrap();
-    }
-    std::fs::create_dir(&path).unwrap();
+    // We don't really care if this fails, as that likely mean that the directory did not exist
+    tokio::fs::remove_dir_all(&path).await.ok();
+    tokio::fs::create_dir_all(&path).await.unwrap();
 
     // Render templates
     let template_env = PathEnvironment::new(&get_repo_path().join("templates"));
-    for template_name in &template_env {
+
+    let mut stream = pin!(template_env.files());
+    while let Some(template_name) = stream.next().await {
         let template = template_env.get_template(&template_name).unwrap();
 
         let content = template
@@ -55,35 +59,48 @@ fn generate(opts: &GlobalOpts) -> Result<(), Error> {
             })
             .unwrap();
 
-        std::fs::write(path.join(template_name), content).unwrap();
+        tokio::fs::write(path.join(template_name), content)
+            .await
+            .unwrap();
     }
 
     // Remove existing config files and create output directory
     let path = get_configs_path();
-    if path.exists() {
-        std::fs::remove_dir_all(&path).unwrap();
-    }
-    std::fs::create_dir(&path).unwrap();
+    tokio::fs::remove_dir_all(&path).await.ok();
+    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));
-        }
+    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_command(cluster.talosctl_gen_config_command());
-        run_command(cluster.talosctl_add_endpoint_command());
-        run_command(cluster.talosctl_merge_command());
+        // 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;
     }
 
     Ok(())
 }
 
-fn main() -> Result<(), Error> {
+#[tokio::main]
+async fn main() -> Result<(), Error> {
     let cli = Cli::parse();
 
     match cli.command {
-        Commands::Generate => generate(&cli.global_opts)?,
+        Commands::Generate => generate(&cli.global_opts).await?,
         Commands::ShellCompletions => generate_complete(
             Shell::from_env().unwrap_or(Shell::Bash),
             &mut Cli::command(),

src/node.rs

@@ -1,10 +1,10 @@
 use std::net::Ipv4Addr;
-use std::process::Command;
 
 use minijinja::Environment;
 use optional_struct::{Applicable, optional_struct};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use tokio::process::Command;
 
 use crate::cluster::{Base, Cluster};
 use crate::patch::{OptionalPatches, OptionalPatchesString, Patches};
@@ -90,7 +90,7 @@ pub(crate) struct NodeDeserialize {
     sops: Secret,
 }
 
-#[derive(Debug, Serialize)]
+#[derive(Debug, Serialize, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct Node {
     hostname: String,
@@ -106,9 +106,9 @@ pub struct Node {
 }
 
 impl Node {
-    pub(crate) fn get(
+    pub(crate) async fn get(
         node_name: &str,
-        env: &Environment,
+        env: &Environment<'static>,
         cluster: &Cluster,
         default: &OptionalNodeDeserialize,
         base: &Base,
@@ -121,7 +121,7 @@ impl Node {
             .to_string();
 
         path.add_extension("yaml");
-        let content = std::fs::read_to_string(path).unwrap();
+        let content = tokio::fs::read_to_string(path).await.unwrap();
 
         let node: OptionalNodeDeserialize = serde_yaml::from_str(&content).unwrap();
 
@@ -172,7 +172,7 @@ impl Node {
         };
 
         // Render patches
-        let patches = patches.render(env, cluster, &node);
+        let patches = patches.render(env, cluster, &node).await;
 
         Node { patches, ..node }
     }

src/patch.rs

@@ -16,27 +16,36 @@ pub(crate) struct Patches<T> {
 
 pub(crate) type OptionalPatchesString = OptionalPatches<String>;
 
-fn render(
+async fn render(
     patches: Vec<String>,
-    env: &Environment,
+    env: &Environment<'static>,
     cluster: &Cluster,
     node: &Node,
 ) -> Vec<serde_yaml::Value> {
-    patches
-        .into_iter()
-        .map(|name| {
-            let content = env
-                .get_template(&name)
-                .unwrap()
-                .render(context! {
-                    node,
-                    cluster
-                })
-                .unwrap();
+    let env = env.clone();
+    let cluster = cluster.clone();
+    let node = node.clone();
 
-            serde_yaml::from_str(&content).unwrap()
-        })
-        .collect()
+    // Minijinja will still have blocking I/O internally
+    tokio::task::spawn_blocking(move || {
+        patches
+            .into_iter()
+            .map(move |name| {
+                let content = env
+                    .get_template(&name)
+                    .unwrap()
+                    .render(context! {
+                        node,
+                        cluster
+                    })
+                    .unwrap();
+
+                serde_yaml::from_str(&content).unwrap()
+            })
+            .collect()
+    })
+    .await
+    .unwrap()
 }
 
 impl Patches<String> {
@@ -50,15 +59,15 @@ impl Patches<String> {
         }
     }
 
-    pub(crate) fn render(
+    pub(crate) async fn render(
         self,
-        env: &Environment,
+        env: &Environment<'static>,
         cluster: &Cluster,
         node: &Node,
     ) -> Patches<serde_yaml::Value> {
         Patches {
-            all: render(self.all.clone(), env, cluster, node),
-            control_plane: render(self.control_plane.clone(), env, cluster, node),
+            all: render(self.all.clone(), env, cluster, node).await,
+            control_plane: render(self.control_plane.clone(), env, cluster, node).await,
         }
     }
 }

src/schematic.rs

@@ -1,5 +1,6 @@
 use schemars::JsonSchema;
 use serde::{Deserialize, Deserializer, Serialize};
+use tokio::runtime::Handle;
 
 use crate::get_talos_path;
 
@@ -11,7 +12,7 @@ where
     let path = get_talos_path().join("schematics").join(name);
     let content = std::fs::read_to_string(path).unwrap().trim().to_owned();
 
-    let client = reqwest::blocking::ClientBuilder::new()
+    let client = reqwest::ClientBuilder::new()
         .user_agent(format!(
             "{}/{}",
             std::env!("CARGO_PKG_NAME"),
@@ -20,19 +21,29 @@ where
         .build()
         .unwrap();
 
-    let res = client
-        .post("https://factory.talos.dev/schematics")
-        .body(content)
-        .send()
-        .map_err(serde::de::Error::custom)?;
-
     #[derive(Debug, Deserialize)]
     struct Response {
         id: String,
     }
 
-    let response: Response = serde_json::from_str(&res.text().map_err(serde::de::Error::custom)?)
-        .map_err(serde::de::Error::custom)?;
+    // 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.
+    // 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)?;
+
+        Ok(response)
+    })?;
 
     Ok(response.id)
 }