moss: Execute triggers in parallel batches

Use the batched_topo() method to return a vec of vecs and execute
each batch of triggers in parallel with rayon

Refactor so we only enter the isolation container once for each scope of
triggers executed (if neccessary)

Add progress callbacks so we can continue to present a smooth progress
bar to the user despite the batched execution

On a VM with 4 vCPUs on an NVMe drive formatted with ext4, total wall
time for installing a simple package is reduced from ~21secs to ~13secs.

moss: Postblit code cleanup

Add common entry to execute triggers and split off common functionality
into it's own function

moss: Fix percentage calculation, improved callbacks

Author: joebonrichie

crates/triggers/src/lib.rs

@@ -78,8 +78,8 @@ impl<'a> Collection<'a> {
         }
     }
 
-    /// Bake the trigger collection into a sane dependency order
-    pub fn bake(&mut self) -> Result<Vec<format::CompiledHandler>, Error> {
+    /// Bake the trigger collection into a sane dependency order, grouped by parallelizable stages.
+    pub fn bake_in_stages(&mut self) -> Result<Vec<Vec<format::CompiledHandler>>, Error> {
         let mut graph = dag::Dag::new();
 
         // ensure all keys are in place
@@ -117,11 +117,19 @@ impl<'a> Collection<'a> {
             }
         }
 
-        // Recollect in dependency order
-        let results = graph
-            .topo()
-            .filter_map(|i| self.hits.remove(i))
-            .flatten()
+        // Recollect in dependency order batches
+        let stages = graph.batched_topo();
+
+        let results = stages
+            .into_iter()
+            .map(|stage| {
+                stage
+                    .iter()
+                    .filter_map(|id| self.hits.get(id))
+                    .flatten()
+                    .cloned()
+                    .collect::<Vec<_>>()
+            })
             .collect::<Vec<_>>();
         Ok(results)
     }

moss/src/client/mod.rs

@@ -414,19 +414,21 @@ impl Client {
     fn apply_triggers(scope: TriggerScope<'_>, fstree: &vfs::Tree<PendingFile>) -> Result<(), postblit::Error> {
         let triggers = postblit::triggers(scope, fstree)?;
 
-        let progress = ProgressBar::new(triggers.len() as u64).with_style(
+        let total_items: u64 = triggers.iter().map(|batch| batch.len() as u64).sum();
+
+        let progress_bar = ProgressBar::new(total_items).with_style(
             ProgressStyle::with_template("\n|{bar:20.green/blue}| {pos}/{len} {msg}")
                 .unwrap()
                 .progress_chars("■≡=- "),
         );
 
         let phase_name = match &scope {
             TriggerScope::Transaction(_, _) => {
-                progress.set_message("Running transaction-scope triggers");
+                progress_bar.set_message("Running transaction-scope triggers");
                 "transaction-scope-triggers"
             }
             TriggerScope::System(_, _) => {
-                progress.set_message("Running system-scope triggers");
+                progress_bar.set_message("Running system-scope triggers");
                 "system-scope-triggers"
             }
         };
@@ -435,37 +437,32 @@ impl Client {
 
         info!(
             phase = phase_name,
-            total_items = triggers.len(),
+            total_items = total_items,
             progress = 0.0,
             event_type = "progress_start",
         );
 
-        for (i, trigger) in progress.wrap_iter(triggers.iter()).enumerate() {
-            trigger.execute()?;
-
-            let trigger_command = match trigger.handler() {
-                triggers::format::Handler::Run { run, .. } => run.clone(),
-                triggers::format::Handler::Delete { .. } => "delete operation".to_owned(),
-            };
+        postblit::execute_triggers(scope, &triggers, |progress| {
+            progress_bar.set_position(progress.completed);
             info!(
-                progress = (i + 1) as f32 / triggers.len() as f32,
-                current = i + 1,
-                total = triggers.len(),
+                progress = progress.completed as f32 / total_items as f32,
+                current = progress.completed,
+                total = total_items,
                 event_type = "progress_update",
-                "Executing {}",
-                trigger_command
+                "Executing {:?}",
+                progress.item
             );
-        }
+        })?;
 
         info!(
             phase = phase_name,
             duration_ms = timer.elapsed().as_millis(),
-            items_processed = triggers.len(),
+            items_processed = total_items,
             progress = 1.0,
             event_type = "progress_completed",
         );
 
-        progress.finish_and_clear();
+        progress_bar.finish_and_clear();
 
         Ok(())
     }

moss/src/client/postblit.rs

@@ -11,11 +11,13 @@
 use std::{
     path::{Path, PathBuf},
     process,
+    sync::atomic::{AtomicUsize, Ordering},
 };
 
 use crate::Installation;
 use container::Container;
 use itertools::Itertools;
+use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 use serde::Deserialize;
 use thiserror::Error;
 use tracing::{error, warn};
@@ -100,13 +102,19 @@ impl TriggerScope<'_> {
 }
 
 /// Condensed type for loaded triggers with scope and executor
-#[derive(Debug)]
-pub(super) struct TriggerRunner<'a> {
-    scope: TriggerScope<'a>,
+pub(super) struct TriggerRunner {
     trigger: CompiledHandler,
 }
 
-/// Load all triggers matching the given scope and staging filesystem
+/// Progress callback handler
+#[derive(Debug, Clone)]
+pub struct Progress<'a> {
+    pub completed: u64,
+    pub item: &'a str,
+}
+
+/// Load all triggers matching the given scope and staging filesystem, return in batches
+/// suitable for concurrent/parallel processing.
 ///
 /// # Arguments
 ///
@@ -115,7 +123,7 @@ pub(super) struct TriggerRunner<'a> {
 pub(super) fn triggers<'a>(
     scope: TriggerScope<'a>,
     fstree: &vfs::tree::Tree<PendingFile>,
-) -> Result<Vec<TriggerRunner<'a>>, Error> {
+) -> Result<Vec<Vec<TriggerRunner>>, Error> {
     // Pre-calculate trigger root path once
     let trigger_root = {
         let mut path = PathBuf::with_capacity(50);
@@ -145,57 +153,125 @@ pub(super) fn triggers<'a>(
     // Load trigger collection, process all the paths, convert to scoped TriggerRunner vec
     let mut collection = triggers::Collection::new(triggers.iter())?;
     collection.process_paths(fstree.iter().map(|m| m.to_string()));
-    let computed_commands = collection
-        .bake()?
+    let batches = collection
+        .bake_in_stages()?
         .into_iter()
-        .map(|trigger| TriggerRunner { scope, trigger })
+        .map(|batch| batch.into_iter().map(|trigger| TriggerRunner { trigger }).collect_vec())
         .collect_vec();
-    Ok(computed_commands)
+    Ok(batches)
 }
 
-impl TriggerRunner<'_> {
-    pub fn handler(&self) -> &Handler {
-        self.trigger.handler()
+/// Execute triggers based on TriggerScope
+///
+/// Execute either transaction or system scope triggers using container sandboxing as necessary
+pub fn execute_triggers(
+    scope: TriggerScope<'_>,
+    triggers: &[Vec<TriggerRunner>],
+    on_progress: impl Fn(Progress<'_>) + Send + Sync,
+) -> Result<(), Error> {
+    match scope {
+        scope @ TriggerScope::Transaction(install, _) => {
+            execute_transaction_triggers(install, scope, triggers, &on_progress)?;
+        }
+        scope @ TriggerScope::System(install, _) => {
+            execute_system_triggers(install, scope, triggers, &on_progress)?;
+        }
+    };
+
+    Ok(())
+}
+
+/// Execute transaction triggers
+///
+/// Transaction triggers are run via sandboxing ([`container::Container`]) to limit their
+/// system view, and limit write access. Each batch of triggers are executed in parallel
+/// to speed up execution time.
+fn execute_transaction_triggers<P>(
+    install: &Installation,
+    scope: TriggerScope<'_>,
+    triggers: &[Vec<TriggerRunner>],
+    on_progress: P,
+) -> Result<(), Error>
+where
+    P: Fn(Progress<'_>) + Send + Sync,
+{
+    // TODO: Add caching support via /var/
+    let isolation = Container::new(install.isolation_dir())
+        .networking(false)
+        .bind_ro(scope.host_path("etc"), "/etc")
+        .bind_rw(scope.guest_path("usr"), "/usr")
+        .work_dir("/");
+
+    isolation.run(|| execute_triggers_directly(triggers, &on_progress))?;
+
+    Ok(())
+}
+
+/// Execute system triggers
+///
+/// System triggers will execute without any sandboxing when moss is used directly against the
+/// live root filesystem, and will force sandboxing when using a non-`/` root (such as using the
+/// `-D argument with `moss install`). Each batch of triggers is executed in parallel to speed up
+/// execution time.
+fn execute_system_triggers<P>(
+    install: &Installation,
+    scope: TriggerScope<'_>,
+    triggers: &[Vec<TriggerRunner>],
+    on_progress: P,
+) -> Result<(), Error>
+where
+    P: Fn(Progress<'_>) + Send + Sync,
+{
+    // OK, if the root == `/` then we can run directly, otherwise we need to containerise with RW.
+    if install.root.to_string_lossy() == "/" {
+        execute_triggers_directly(triggers, on_progress)?;
+    } else {
+        let isolation = Container::new(install.isolation_dir())
+            .networking(false)
+            .bind_rw(scope.host_path("etc"), "/etc")
+            .bind_rw(scope.guest_path("usr"), "/usr")
+            .work_dir("/");
+
+        isolation.run(|| execute_triggers_directly(triggers, &on_progress))?;
     }
+    Ok(())
+}
 
-    /// Execute a trigger, taking care to account for the transaction scope and client scope
-    ///
-    /// All transaction triggers are run via sandboxing ([`container::Container`]) to limit their
-    /// system view, and limit write access.
-    /// System triggers will execute without any sandboxing when moss is used directly against the
-    /// live root filesystem, and will force sandboxing when using a non-`/` root (such as using the
-    /// `-D argument with `moss install`)
-    pub fn execute(&self) -> Result<(), Error> {
-        match self.scope {
-            TriggerScope::Transaction(install, _) => {
-                // TODO: Add caching support via /var/
-                let isolation = Container::new(install.isolation_dir())
-                    .networking(false)
-                    .bind_ro(self.scope.host_path("etc"), "/etc")
-                    .bind_rw(self.scope.guest_path("usr"), "/usr")
-                    .work_dir("/");
-
-                Ok(isolation.run(|| execute_trigger_directly(&self.trigger))?)
-            }
-            TriggerScope::System(install, _) => {
-                // OK, if the root == `/` then we can run directly, otherwise we need to containerise with RW.
-                if install.root.to_string_lossy() == "/" {
-                    Ok(execute_trigger_directly(&self.trigger)?)
-                } else {
-                    let isolation = Container::new(install.isolation_dir())
-                        .networking(false)
-                        .bind_rw(self.scope.host_path("etc"), "/etc")
-                        .bind_rw(self.scope.guest_path("usr"), "/usr")
-                        .work_dir("/");
-
-                    Ok(isolation.run(|| execute_trigger_directly(&self.trigger))?)
-                }
-            }
-        }
+impl TriggerRunner {
+    pub fn handler(&self) -> &Handler {
+        self.trigger.handler()
     }
 }
 
 /// Internal executor for triggers.
+fn execute_triggers_directly<P>(triggers: &[Vec<TriggerRunner>], on_progress: P) -> Result<(), Error>
+where
+    P: Fn(Progress<'_>) + Send + Sync,
+{
+    let rayon_runtime = rayon::ThreadPoolBuilder::new().build().expect("rayon runtime");
+
+    let counter = AtomicUsize::new(0);
+
+    rayon_runtime.install(|| {
+        triggers.iter().try_for_each(|batch| {
+            batch.par_iter().try_for_each(|trigger| {
+                let res = execute_trigger_directly(&trigger.trigger);
+                let completed = counter.fetch_add(1, Ordering::Relaxed);
+                (on_progress)(Progress {
+                    completed: completed as u64,
+                    item: match trigger.handler() {
+                        Handler::Run { run, .. } => run,
+                        Handler::Delete { .. } => "delete operation",
+                    },
+                });
+                res
+            })
+        })
+    })?;
+    Ok(())
+}
+
+/// Internal executor for individual triggers.
 fn execute_trigger_directly(trigger: &CompiledHandler) -> Result<(), Error> {
     match trigger.handler() {
         Handler::Run { run, args } => {