From d7eac73e4d8548bcde3f1fc42e0647c9c7d46ba5 Mon Sep 17 00:00:00 2001 From: Joey Riches Date: Wed, 18 Feb 2026 13:08:25 +0000 Subject: [PATCH] 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. --- crates/triggers/src/lib.rs | 22 +++-- moss/src/client/mod.rs | 35 ++++---- moss/src/client/postblit.rs | 168 ++++++++++++++++++++++++++---------- 3 files changed, 153 insertions(+), 72 deletions(-) diff --git a/crates/triggers/src/lib.rs b/crates/triggers/src/lib.rs index 2dc919963..0ccdde91f 100644 --- a/crates/triggers/src/lib.rs +++ b/crates/triggers/src/lib.rs @@ -77,8 +77,8 @@ impl<'a> Collection<'a> { } } - /// Bake the trigger collection into a sane dependency order - pub fn bake(&mut self) -> Result, Error> { + /// Bake the trigger collection into a sane dependency order, grouped by parallelizable stages. + pub fn bake_in_stages(&mut self) -> Result>, Error> { let mut graph = dag::Dag::new(); // ensure all keys are in place @@ -116,11 +116,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::>() + }) .collect::>(); Ok(results) } diff --git a/moss/src/client/mod.rs b/moss/src/client/mod.rs index 1c3b014fb..fca83781c 100644 --- a/moss/src/client/mod.rs +++ b/moss/src/client/mod.rs @@ -413,7 +413,9 @@ impl Client { fn apply_triggers(scope: TriggerScope<'_>, fstree: &vfs::Tree) -> 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("■≡=- "), @@ -421,11 +423,11 @@ impl Client { 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" } }; @@ -434,37 +436,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(()) } diff --git a/moss/src/client/postblit.rs b/moss/src/client/postblit.rs index 68cf1585a..d05af76d3 100644 --- a/moss/src/client/postblit.rs +++ b/moss/src/client/postblit.rs @@ -10,11 +10,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}; @@ -99,13 +101,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 /// @@ -114,7 +122,7 @@ pub(super) struct TriggerRunner<'a> { pub(super) fn triggers<'a>( scope: TriggerScope<'a>, fstree: &vfs::tree::Tree, -) -> Result>, Error> { +) -> Result>, Error> { // Pre-calculate trigger root path once let trigger_root = { let mut path = PathBuf::with_capacity(50); @@ -144,57 +152,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], + 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

( + install: &Installation, + scope: TriggerScope<'_>, + triggers: &[Vec], + 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

( + install: &Installation, + scope: TriggerScope<'_>, + triggers: &[Vec], + 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

(triggers: &[Vec], 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 } => {