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 } => {