diff --git a/brane-cli/src/repl.rs b/brane-cli/src/repl.rs index 84bbc16a..d36326cd 100644 --- a/brane-cli/src/repl.rs +++ b/brane-cli/src/repl.rs @@ -31,6 +31,7 @@ use rustyline::history::DefaultHistory; use rustyline::validate::{self, MatchingBracketValidator, Validator}; use rustyline::{CompletionType, Config, Context, EditMode, Editor}; use rustyline_derive::Helper; +use specifications::profiling::ProfileScopeHandle; pub use crate::errors::ReplError as Error; use crate::instance::InstanceInfo; @@ -388,10 +389,12 @@ async fn local_repl( let snippet = Snippet { lines: line_count, workflow }; // Next, we run the VM (one snippet only ayway) - let res: FullValue = run_offline_vm(&mut state, snippet).await.map_err(|source| Error::RunError { what: "offline VM", source })?; + let res: FullValue = run_offline_vm(&mut state, snippet, ProfileScopeHandle::dummy()) + .await + .map_err(|source| Error::RunError { what: "offline VM", source })?; // Then, we collect and process the result - if let Err(source) = process_offline_result(res) { + if let Err(source) = process_offline_result(res, ProfileScopeHandle::dummy()) { error!("{}", Error::ProcessError { what: "offline VM", source }); continue; } diff --git a/brane-cli/src/run.rs b/brane-cli/src/run.rs index 5f3e8853..c18d250d 100644 --- a/brane-cli/src/run.rs +++ b/brane-cli/src/run.rs @@ -14,7 +14,7 @@ use std::borrow::Cow; use std::fs; -use std::io::{Read, Stderr, Stdout, Write}; +use std::io::{Read, Stderr, Stdout, Write, stdout}; use std::path::{Path, PathBuf}; use std::str::FromStr; use std::sync::Arc; @@ -33,6 +33,7 @@ use parking_lot::{Mutex, MutexGuard}; use specifications::data::{AccessKind, DataIndex, DataInfo}; use specifications::driving::{CreateSessionRequest, DriverServiceClient, ExecuteRequest}; use specifications::package::PackageIndex; +use specifications::profiling::{ProfileReport, ProfileScopeHandle}; use tempfile::{TempDir, tempdir}; use tonic::Code; @@ -558,17 +559,18 @@ pub async fn run_dummy_vm(state: &mut DummyVmState, what: impl AsRef, snipp /// /// # Arguments /// - `state`: The OfflineVmState that we use to run the local VM. -/// - `what`: The thing we're running. Either a filename, or something like stdin. /// - `snippet`: The snippet to compile and run. +/// - `scope`: Some [`ProfileScopeHandle`] to report profile timings in. Use a +/// [`ProfileScopeHandle::dummy()`] if you don't care. /// /// # Returns /// The FullValue that the workflow returned, if any. If there was no value, returns FullValue::Void instead. /// /// # Errors /// This function errors if we failed to compile or run the workflow somehow. -pub async fn run_offline_vm(state: &mut OfflineVmState, snippet: Snippet) -> Result { +pub async fn run_offline_vm(state: &mut OfflineVmState, snippet: Snippet, scope: ProfileScopeHandle) -> Result { // Run it in the local VM (which is a bit ugly do to the need to consume the VM itself) - let res: (OfflineVm, Result) = state.vm.take().unwrap().exec(snippet.workflow).await; + let res: (OfflineVm, Result) = state.vm.take().unwrap().exec(snippet.workflow, scope).await; state.vm = Some(res.0); let res: FullValue = match res.1 { Ok(res) => res, @@ -645,16 +647,17 @@ pub fn process_dummy_result(result: FullValue) { /// Processes the given result of an offline workflow execution. /// /// # Arguments -/// - `result_dir`: The directory where temporary results are stored. /// - `result`: The value to process. +/// - `scope`: Some [`ProfileScopeHandle`] to write profile timings in. /// /// # Returns /// Nothing, but does print any result to stdout. /// /// # Errors /// This function may error if we failed to get an up-to-date data index. -pub fn process_offline_result(result: FullValue) -> Result<(), Error> { +pub fn process_offline_result(result: FullValue, scope: ProfileScopeHandle) -> Result<(), Error> { // We only print + let guard = scope.time(format!("Processing of {result:?}")); if result != FullValue::Void { println!("\nWorkflow returned value {}", style(format!("'{result}'")).bold().cyan()); @@ -671,9 +674,12 @@ pub fn process_offline_result(result: FullValue) -> Result<(), Error> { let datasets_dir = ensure_datasets_dir(false).map_err(|source| Error::DatasetsDirError { source })?; // Fetch a new, local DataIndex to get up-to-date entries - let index: DataIndex = brane_tsk::local::get_data_index(datasets_dir).map_err(|source| Error::LocalDataIndexError { source })?; + let index: DataIndex = scope.time_func("Collecting local data index", || { + brane_tsk::local::get_data_index(datasets_dir).map_err(|source| Error::LocalDataIndexError { source }) + })?; // Fetch the method of its availability + let _guard = scope.time(format!("Finding dataset \"{name}\" in local data index")); let info: &DataInfo = index.get(&name).ok_or_else(|| Error::UnknownDataset { name: name.clone().into() })?; let access: &AccessKind = info .access @@ -690,6 +696,7 @@ pub fn process_offline_result(result: FullValue) -> Result<(), Error> { _ => {}, } } + drop(guard); // DOne Ok(()) @@ -782,7 +789,7 @@ pub async fn handle( // Run the thing remote_run(info, use_case, proxy_addr, options, source, source_code, profile).await } else { - local_run(options, docker_opts, source, source_code, keep_containers).await + local_run(options, docker_opts, source, source_code, keep_containers, profile).await } } else { dummy_run(options, source, source_code).await @@ -823,6 +830,7 @@ async fn dummy_run(options: ParserOptions, what: impl AsRef, source: impl A /// - `what`: A description of the source we're reading (e.g., the filename or stdin) /// - `source`: The source code to read. /// - `keep_containers`: Whether to keep containers after execution or not. +/// - `profile`: Whether to print profile timings (true) or not (false). /// /// # Returns /// Nothing, but does print results and such to stdout. Might also produce new datasets. @@ -832,6 +840,7 @@ async fn local_run( what: impl AsRef, source: impl AsRef, keep_containers: bool, + profile: bool, ) -> Result<(), Error> { let what: &str = what.as_ref(); let source: &str = source.as_ref(); @@ -843,11 +852,16 @@ async fn local_run( let snippet = Snippet::from_source(&mut state.state, &mut state.source, &state.pindex, &state.dindex, None, &state.options, what, source) .map_err(Error::CompileError)?; + // Get a report + let report: Option> = if profile { Some(ProfileReport::auto_reporting("brane_cli::local_run", stdout())) } else { None }; + // Next, we run the VM (one snippet only ayway) - let res: FullValue = run_offline_vm(&mut state, snippet).await?; + let res: FullValue = + run_offline_vm(&mut state, snippet, report.as_ref().map(|r| r.nest("run_offline_vm()")).unwrap_or_else(|| ProfileScopeHandle::dummy())) + .await?; // Then, we collect and process the result - process_offline_result(res)?; + process_offline_result(res, report.as_ref().map(|r| r.nest("process_offline_result()")).unwrap_or_else(|| ProfileScopeHandle::dummy()))?; // Done Ok(()) diff --git a/brane-cli/src/test.rs b/brane-cli/src/test.rs index 5a992429..63e19ae1 100644 --- a/brane-cli/src/test.rs +++ b/brane-cli/src/test.rs @@ -23,6 +23,7 @@ use brane_tsk::input::prompt_for_input; use console::style; use specifications::data::DataIndex; use specifications::package::PackageInfo; +use specifications::profiling::ProfileScopeHandle; use specifications::version::Version; use crate::errors::TestError; @@ -178,7 +179,8 @@ pub async fn test_generic( ) .map_err(|source| TestError::RunError { source: run::Error::CompileError(source) })?; - let result: FullValue = run_offline_vm(&mut state, snippet).await.map_err(|source| TestError::RunError { source })?; + let result: FullValue = + run_offline_vm(&mut state, snippet, ProfileScopeHandle::dummy()).await.map_err(|source| TestError::RunError { source })?; // Write the intermediate result if told to do so if let Some(file) = show_result { diff --git a/brane-cli/src/vm.rs b/brane-cli/src/vm.rs index 00c6da46..7fefccdd 100644 --- a/brane-cli/src/vm.rs +++ b/brane-cli/src/vm.rs @@ -68,7 +68,7 @@ impl VmPlugin for OfflinePlugin { _loc: Location, name: DataName, preprocess: PreprocessKind, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result { info!("Preprocessing data '{name}' for call at {pc} in an offline environment"); debug!("Method of preprocessing: {preprocess:?}"); @@ -84,7 +84,7 @@ impl VmPlugin for OfflinePlugin { global: &Arc>, _local: &Self::LocalState, info: TaskInfo<'_>, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result, Self::ExecuteError> { let mut info = info; info!("Calling task '{}' in an offline environment", info.name); @@ -112,7 +112,7 @@ impl VmPlugin for OfflinePlugin { // Resolve the input arguments, generating the folders we have to bind let binds: Vec = prof - .time_fut("argument preprocessing", docker::preprocess_args(&mut info.args, &info.input, info.result, None::, results_dir)) + .time_fut("Argument preprocessing", docker::preprocess_args(&mut info.args, &info.input, info.result, None::, results_dir)) .await?; let params: String = serde_json::to_string(&info.args).map_err(|source| ExecuteError::ArgsEncodeError { source })?; @@ -143,7 +143,7 @@ impl VmPlugin for OfflinePlugin { // We can now execute the task on the local Docker daemon debug!("Executing task '{}'...", info.name); let (code, stdout, stderr) = prof - .time_fut("execution", docker::run_and_wait(docker_opts, einfo, keep_container)) + .nest_fut("Execution", |scope| docker::run_and_wait(docker_opts, einfo, keep_container, scope)) .await .map_err(|source| ExecuteError::DockerError { name: info.name.into(), image: Box::new(image.clone()), source })?; debug!("Container return code: {}", code); @@ -171,7 +171,7 @@ impl VmPlugin for OfflinePlugin { _local: &Self::LocalState, text: &str, newline: bool, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result<(), Self::StdoutError> { info!("Writing '{}' to stdout (newline: {}) in an offline environment...", text, if newline { "yes" } else { "no" }); @@ -192,7 +192,7 @@ impl VmPlugin for OfflinePlugin { _loc: &Location, name: &str, path: &Path, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result<(), Self::CommitError> { info!("Publicizing intermediate result '{}' in an offline environment...", name); debug!("Physical file(s): {}", path.display()); @@ -210,7 +210,7 @@ impl VmPlugin for OfflinePlugin { name: &str, path: &Path, data_name: &str, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result<(), Self::CommitError> { info!("Committing intermediate result '{}' to '{}' in an offline environment...", name, data_name); debug!("Physical file(s): {}", path.display()); @@ -344,10 +344,11 @@ impl OfflineVm { /// /// # Arguments /// - `workflow`: The Workflow to execute. + /// - `scope`: Some [`ProfileScopeHandle`] to report any profile information in. /// /// # Returns /// The result of the workflow, if any. It also returns `self` again for subsequent runs. - pub async fn exec(self, workflow: Workflow) -> (Self, Result) { + pub async fn exec(self, workflow: Workflow, scope: ProfileScopeHandle) -> (Self, Result) { // Step 1: Plan let plan: Result = { let planner: OfflinePlanner = { @@ -372,7 +373,7 @@ impl OfflineVm { let this: Arc> = Arc::new(RwLock::new(self)); // Run the VM and get self back - let result: Result = Self::run::(this.clone(), plan, ProfileScopeHandle::dummy()).await; + let result: Result = Self::run::(this.clone(), plan, scope).await; let this: Self = match Arc::try_unwrap(this) { Ok(this) => this.into_inner().unwrap(), Err(_) => { diff --git a/brane-ctl/src/download.rs b/brane-ctl/src/download.rs index 26798ce8..362e65a6 100644 --- a/brane-ctl/src/download.rs +++ b/brane-ctl/src/download.rs @@ -25,6 +25,7 @@ use enum_debug::EnumDebug as _; use log::{debug, info, warn}; use specifications::arch::Arch; use specifications::container::Image; +use specifications::profiling::ProfileScopeHandle; use specifications::version::Version; use tempfile::TempDir; @@ -272,7 +273,7 @@ pub async fn services( // Make sure the image is pulled println!("Downloading auxillary image {}...", style(image).bold().green()); - ensure_image(&docker, Image::new(name, None::<&str>, None::<&str>), ImageSource::Registry(image.into())) + ensure_image(&docker, Image::new(name, None::<&str>, None::<&str>), ImageSource::Registry(image.into()), ProfileScopeHandle::dummy()) .await .map_err(|source| Error::PullError { name: name.into(), image: image.into(), source })?; diff --git a/brane-ctl/src/lifetime.rs b/brane-ctl/src/lifetime.rs index a1b4f15b..afc7e6d9 100644 --- a/brane-ctl/src/lifetime.rs +++ b/brane-ctl/src/lifetime.rs @@ -37,6 +37,7 @@ use rand::Rng; use rand::distr::Alphanumeric; use serde::{Deserialize, Serialize}; use specifications::container::Image; +use specifications::profiling::ProfileScopeHandle; use specifications::version::Version; pub use crate::errors::LifetimeError as Error; @@ -487,7 +488,7 @@ async fn load_images(docker: &Docker, images: HashMap, ImageSour }; // Simply rely on ensure_image - ensure_image(docker, &image, &image_source).await.map_err(|source| Error::ImageLoadError { + ensure_image(docker, &image, &image_source, ProfileScopeHandle::dummy()).await.map_err(|source| Error::ImageLoadError { image: Box::new(image), image_source: Box::new(image_source), source, diff --git a/brane-drv/src/planner.rs b/brane-drv/src/planner.rs index 67cf7106..2188b8c8 100644 --- a/brane-drv/src/planner.rs +++ b/brane-drv/src/planner.rs @@ -41,7 +41,7 @@ impl InstancePlanner { /// /// # Returns /// The same workflow as given, but now with all tasks and data transfers planned. - pub async fn plan(plr: &Address, app_id: AppId, workflow: Workflow, prof: ProfileScopeHandle<'_>) -> Result { + pub async fn plan(plr: &Address, app_id: AppId, workflow: Workflow, prof: ProfileScopeHandle) -> Result { // Generate the ID let task_id: String = format!("{}", TaskId::generate()); diff --git a/brane-drv/src/vm.rs b/brane-drv/src/vm.rs index 94ead93f..7910e250 100644 --- a/brane-drv/src/vm.rs +++ b/brane-drv/src/vm.rs @@ -81,7 +81,7 @@ impl VmPlugin for InstancePlugin { loc: Location, name: DataName, preprocess: PreprocessKind, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result { info!("Preprocessing {} '{}' on '{}' in a distributed environment...", name.variant(), name.name(), loc); debug!("Preprocessing to be done: {:?}", preprocess); @@ -159,7 +159,7 @@ impl VmPlugin for InstancePlugin { global: &Arc>, _local: &Self::LocalState, info: TaskInfo<'_>, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result, Self::ExecuteError> { info!("Executing task '{}' at '{}' in a distributed environment...", info.name, info.location); debug!("Package: '{}' v{}", info.package_name, info.package_version); @@ -376,7 +376,7 @@ impl VmPlugin for InstancePlugin { _local: &Self::LocalState, text: &str, newline: bool, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result<(), Self::StdoutError> { info!("Writing '{}' to stdout in a distributed environment...", text); debug!("Newline: {}", if newline { "yes" } else { "no" }); @@ -409,7 +409,7 @@ impl VmPlugin for InstancePlugin { loc: &Location, name: &str, path: &Path, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result<(), Self::CommitError> { info!("Publicizing intermediate result '{}' living at '{}' in a distributed environment...", name, loc); debug!("File: '{}'", path.display()); @@ -426,7 +426,7 @@ impl VmPlugin for InstancePlugin { name: &str, path: &Path, data_name: &str, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result<(), Self::CommitError> { info!("Committing intermediate result '{}' living at '{}' as '{}' in a distributed environment...", name, loc, data_name); debug!("File: '{}'", path.display()); @@ -519,7 +519,7 @@ impl InstanceVm { tx: Sender>, id: AppId, workflow: Workflow, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> (Self, Result) { // Step 0: Load files let plr_addr: Address = { diff --git a/brane-exe/src/dummy.rs b/brane-exe/src/dummy.rs index ba4dce66..f7a1e1c4 100644 --- a/brane-exe/src/dummy.rs +++ b/brane-exe/src/dummy.rs @@ -143,7 +143,7 @@ impl VmPlugin for DummyPlugin { _loc: Location, name: DataName, _preprocess: specifications::data::PreprocessKind, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result { info!("Processing dummy `DummyVm::preprocess()` call for intermediate result '{name}' in {pc}"); @@ -155,7 +155,7 @@ impl VmPlugin for DummyPlugin { global: &Arc>, _local: &Self::LocalState, info: TaskInfo<'_>, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result, Self::ExecuteError> { info!( "Processing dummy call to '{}'@'{}' with {} in {}[{}]...", @@ -179,7 +179,7 @@ impl VmPlugin for DummyPlugin { _local: &Self::LocalState, text: &str, newline: bool, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result<(), Self::StdoutError> { info!("Processing dummy stdout write (newline: {})...", if newline { "yes" } else { "no" },); @@ -198,7 +198,7 @@ impl VmPlugin for DummyPlugin { _loc: &Location, name: &str, path: &Path, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result<(), Self::CommitError> { info!("Processing dummy publicize for result '{}' @ '{:?}'...", name, path.display(),); @@ -213,7 +213,7 @@ impl VmPlugin for DummyPlugin { name: &str, path: &Path, data_name: &str, - _prof: ProfileScopeHandle<'_>, + _prof: ProfileScopeHandle, ) -> Result<(), Self::CommitError> { info!("Processing dummy commit for result '{}' @ '{:?}' to '{}'...", name, path.display(), data_name,); diff --git a/brane-exe/src/spec.rs b/brane-exe/src/spec.rs index 1eed1e15..28781ce2 100644 --- a/brane-exe/src/spec.rs +++ b/brane-exe/src/spec.rs @@ -99,7 +99,7 @@ pub trait VmPlugin: 'static + Send + Sync { loc: Location, name: DataName, preprocess: PreprocessKind, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result; @@ -124,7 +124,7 @@ pub trait VmPlugin: 'static + Send + Sync { global: &Arc>, local: &Self::LocalState, info: TaskInfo<'_>, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result, Self::ExecuteError>; @@ -150,7 +150,7 @@ pub trait VmPlugin: 'static + Send + Sync { local: &Self::LocalState, text: &str, newline: bool, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result<(), Self::StdoutError>; @@ -178,7 +178,7 @@ pub trait VmPlugin: 'static + Send + Sync { loc: &Location, name: &str, path: &Path, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result<(), Self::CommitError>; /// A function that commits the given intermediate result by promoting it a Data. @@ -206,7 +206,7 @@ pub trait VmPlugin: 'static + Send + Sync { name: &str, path: &Path, data_name: &str, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result<(), Self::CommitError>; } diff --git a/brane-exe/src/thread.rs b/brane-exe/src/thread.rs index 5ca1aacf..acf194f2 100644 --- a/brane-exe/src/thread.rs +++ b/brane-exe/src/thread.rs @@ -28,7 +28,7 @@ use enum_debug::EnumDebug as _; use futures::future::{BoxFuture, FutureExt}; use log::debug; use specifications::data::{AccessKind, AvailabilityKind, DataName}; -use specifications::profiling::{ProfileScopeHandle, ProfileScopeHandleOwned}; +use specifications::profiling::ProfileScopeHandle; use tokio::spawn; use tokio::task::JoinHandle; @@ -118,7 +118,7 @@ mod tests { results: Arc::new(Mutex::new(HashMap::new())), text: text.clone(), }); - match main.run::(ProfileScopeHandleOwned::dummy()).await { + match main.run::(ProfileScopeHandle::dummy()).await { Ok(value) => { println!("Workflow stdout:"); print!("{}", text.lock().unwrap()); @@ -178,7 +178,7 @@ enum EdgeResult { /// This function may error if the given `input` does not contain any of the data in the value _or_ if the referenced input is not yet planned. #[async_recursion] #[allow(clippy::too_many_arguments, clippy::multiple_bound_locations)] -async fn preprocess_value<'p: 'async_recursion, P: VmPlugin>( +async fn preprocess_value( global: &Arc>, local: &P::LocalState, pc: ProgramCounter, @@ -187,7 +187,7 @@ async fn preprocess_value<'p: 'async_recursion, P: VmPlugin>( value: &FullValue, input: &HashMap>, data: &mut HashMap>>, - prof: ProfileScopeHandle<'p>, + prof: ProfileScopeHandle, ) -> Result<(), Error> { // If it's a data or intermediate result, get it; skip it otherwise let name: DataName = match value { @@ -242,7 +242,6 @@ async fn preprocess_value<'p: 'async_recursion, P: VmPlugin>( // Ok(access) => access, // Err(err) => { return Err(Error::Custom{ pc, err: Box::new(err) }); } // } - let prof = ProfileScopeHandleOwned::from(prof); let global = global.clone(); let local = local.clone(); let at = at.clone(); @@ -1174,14 +1173,14 @@ impl Thread { /// # Arguments /// - `pc`: Points to the current edge to execute (as a [`ProgramCounter``]). /// - `plugins`: An object implementing various parts of task execution that are dependent on the actual setup (i.e., offline VS instance). - /// - `prof`: A ProfileScopeHandleOwned that is used to provide more details about the execution times of a single edge. Note that this is _not_ user-relevant, only debug/framework-relevant. + /// - `prof`: A [`ProfileScopeHandle`] that is used to provide more details about the execution times of a single edge. Note that this is _not_ user-relevant, only debug/framework-relevant. /// /// # Returns /// The next index to execute. Note that this is an _absolute_ index (so it will typically be `idx` + 1) /// /// # Errors /// This function may error if execution of the edge failed. This is typically due to incorrect runtime typing or due to failure to perform an external function call. - async fn exec_edge>(&mut self, pc: ProgramCounter, prof: ProfileScopeHandleOwned) -> EdgeResult { + async fn exec_edge>(&mut self, pc: ProgramCounter, prof: ProfileScopeHandle) -> EdgeResult { // We can early stop if the program counter is out-of-bounds if pc.func_id.is_main() { if pc.edge_idx >= self.graph.len() { @@ -1259,13 +1258,13 @@ impl Thread { // Next, fetch all the datasets required by calling the external transfer function; // The map created maps data names to ways of accessing them locally that may be passed to the container itself. - let prepr = prof.nest("argument preprocessing"); + let prepr = prof.nest(format!("Preprocessing of {} arguments", args.len())); let total = prepr.time("Total"); let mut handles: HashMap>> = HashMap::new(); for (i, value) in args.values().enumerate() { // Preprocess the given value if let Err(err) = prepr - .nest_fut(format!("argument {i}"), |scope| { + .nest_fut(format!("Argument {i}"), |scope| { preprocess_value::

(&self.global, &self.local, pc, task, at, value, input, &mut handles, scope) }) .await @@ -1427,7 +1426,7 @@ impl Thread { let prof = prof.clone(); // Schedule its running on the runtime (`spawn`) - self.threads.push((i, spawn(async move { prof.nest_fut(format!("branch {i}"), |scope| thread.run::

(scope.into())).await }))); + self.threads.push((i, spawn(async move { prof.nest_fut(format!("branch {i}"), |scope| thread.run::

(scope)).await }))); } // Mark those threads to wait for, and then move to the join @@ -1942,7 +1941,7 @@ impl Thread { /// /// # Arguments /// - `prof`: A ProfileScopeHandleOwned that is used to provide more details about the execution times of a workflow execution. Note that this is _not_ user-relevant, only debug/framework-relevant. - /// + /// /// The reason it is owned is due to the boxed return future. It's your responsibility to keep the parent into scope after the future returns; if you don't any collected profile results will likely not be printed. /// /// # Returns @@ -1950,14 +1949,14 @@ impl Thread { /// /// # Errors /// This function may error if execution of an edge or instruction failed. This is typically due to incorrect runtime typing. - pub fn run>(mut self, prof: ProfileScopeHandleOwned) -> BoxFuture<'static, Result> { + pub fn run>(mut self, prof: ProfileScopeHandle) -> BoxFuture<'static, Result> { async move { // Start executing edges from where we left off - let prof: ProfileScopeHandleOwned = prof; + let prof: ProfileScopeHandle = prof; loop { // Run the edge self.pc = match prof - .nest_fut(format!("{:?} ({})", self.get_edge(self.pc)?.variant(), self.pc), |scope| self.exec_edge::

(self.pc, scope.into())) + .nest_fut(format!("{:?} ({})", self.get_edge(self.pc)?.variant(), self.pc), |scope| self.exec_edge::

(self.pc, scope)) .await { // Either quit or continue, noting down the time taken @@ -1982,7 +1981,7 @@ impl Thread { /// /// # Arguments /// - `prof`: A ProfileScopeHandleOwned that is used to provide more details about the execution times of a workflow execution. Note that this is _not_ user-relevant, only debug/framework-relevant. - /// + /// /// The reason it is owned is due to the boxed return future. It's your responsibility to keep the parent into scope after the future returns; if you don't any collected profile results will likely not be printed. /// /// # Returns @@ -1992,15 +1991,15 @@ impl Thread { /// This function may error if execution of an edge or instruction failed. This is typically due to incorrect runtime typing. pub fn run_snippet>( mut self, - prof: ProfileScopeHandleOwned, + prof: ProfileScopeHandle, ) -> BoxFuture<'static, Result<(Value, RunState), Error>> { async move { // Start executing edges from where we left off - let prof: ProfileScopeHandleOwned = prof; + let prof: ProfileScopeHandle = prof; loop { // Run the edge self.pc = match prof - .nest_fut(format!("{:?} ({})", self.get_edge(self.pc)?.variant(), self.pc), |scope| self.exec_edge::

(self.pc, scope.into())) + .nest_fut(format!("{:?} ({})", self.get_edge(self.pc)?.variant(), self.pc), |scope| self.exec_edge::

(self.pc, scope)) .await { // Either quit or continue, noting down the time taken diff --git a/brane-exe/src/vm.rs b/brane-exe/src/vm.rs index e99ddd94..b8f39534 100644 --- a/brane-exe/src/vm.rs +++ b/brane-exe/src/vm.rs @@ -94,7 +94,7 @@ pub trait Vm { async fn run>( this: Arc>, snippet: Workflow, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result where Self: Sync, diff --git a/brane-job/src/worker.rs b/brane-job/src/worker.rs index 98d1bfe0..9edd3f94 100644 --- a/brane-job/src/worker.rs +++ b/brane-job/src/worker.rs @@ -347,7 +347,7 @@ async fn preprocess_transfer_tar_local( workflow: Workflow, location: Location, dataname: DataName, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result { debug!("Preprocessing by executing a data transfer"); debug!("Downloading '{location}' from '{dataname}' to local machine"); @@ -493,7 +493,7 @@ async fn preprocess_transfer_tar_local( // /// // /// # Errors // /// This function can error for literally a million reasons - but they mostly relate to IO (file access, request success etc). -// async fn preprocess_transfer_tar_k8s(kinfo: K8sOptions, location: Location, address: impl AsRef, prof: ProfileScopeHandle<'_>) -> Result { +// async fn preprocess_transfer_tar_k8s(kinfo: K8sOptions, location: Location, address: impl AsRef, prof: ProfileScopeHandle) -> Result { // debug!("Preprocessing by executing a data transfer"); // let address: &str = address.as_ref(); // debug!("Downloading from {} ({}) to Kubernetes cluster", location, address); @@ -532,7 +532,7 @@ pub async fn preprocess_transfer_tar( workflow: Workflow, location: Location, dataname: DataName, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result { debug!("Preprocessing tar..."); @@ -892,7 +892,7 @@ async fn get_container( async fn get_container_ids( worker_cfg: &WorkerConfig, image_path: impl AsRef, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result<(String, Option), ExecuteError> { let image_path: &Path = image_path.as_ref(); debug!("Computing ID and hash for '{}'...", image_path.display()); @@ -980,7 +980,7 @@ async fn ensure_container( proxy: Arc, endpoint: impl AsRef, image: &Image, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result<(PathBuf, String, Option), ExecuteError> { // Download the file if we don't have it locally already let image_path: PathBuf = match prof.time_func("cache checking", || get_cached_container(worker_cfg, image)) { @@ -1021,7 +1021,7 @@ async fn execute_task_local( container_path: impl AsRef, tinfo: TaskInfo, keep_container: bool, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result { let container_path: &Path = container_path.as_ref(); let mut tinfo: TaskInfo = tinfo; @@ -1149,7 +1149,7 @@ async fn execute_task_local( // /// // /// # Errors // /// This function errors if the task fails for whatever reason or we didn't even manage to launch it. -// async fn execute_task_k8s(worker_cfg: &WorkerConfig, kinfo: K8sOptions, tx: &Sender>, container_path: impl AsRef, tinfo: TaskInfo, prof: ProfileScopeHandle<'_>) -> Result { +// async fn execute_task_k8s(worker_cfg: &WorkerConfig, kinfo: K8sOptions, tx: &Sender>, container_path: impl AsRef, tinfo: TaskInfo, prof: ProfileScopeHandle) -> Result { // let container_path : &Path = container_path.as_ref(); // let mut tinfo : TaskInfo = tinfo; // let image : Image = tinfo.image.clone().unwrap(); @@ -1275,7 +1275,7 @@ async fn execute_task( cinfo: CentralNodeInfo, tinfo: TaskInfo, keep_container: bool, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result<(), ExecuteError> { let mut tinfo = tinfo; @@ -1465,7 +1465,7 @@ async fn commit_result( worker_cfg: &WorkerConfig, name: impl AsRef, data_name: impl AsRef, - prof: ProfileScopeHandle<'_>, + prof: ProfileScopeHandle, ) -> Result<(), CommitError> { let name: &str = name.as_ref(); let data_name: &str = data_name.as_ref(); diff --git a/brane-tsk/src/docker.rs b/brane-tsk/src/docker.rs index d8c8a1de..ddce3915 100644 --- a/brane-tsk/src/docker.rs +++ b/brane-tsk/src/docker.rs @@ -36,6 +36,7 @@ use sha2::{Digest, Sha256}; use specifications::container::{Image, VolumeBind}; use specifications::data::{AccessKind, DataName}; use specifications::package::Capability; +use specifications::profiling::ProfileScopeHandle; use tokio::fs::{self as tfs, File as TFile}; use tokio::io::{self as tio, AsyncReadExt as _, AsyncWriteExt as _}; use tokio_tar::Archive; @@ -538,18 +539,20 @@ fn preprocess_arg( /// # Arguments /// - `docker`: The Docker instance to use for accessing the container. /// - `info`: The ExecuteInfo describing what to launch and how. +/// - `scope`: Some [`ProfileScopeHandle`] to report profile timings in. Use [`ProfileScopeHandle::dummy()`] if you're not interested in the results. /// /// # Returns /// The name of the container such that it can be waited on later. /// /// # Errors /// This function may error for many reasons, which usually means that the container failed to be created or started (wow!). -async fn create_and_start_container(docker: &Docker, info: &ExecuteInfo) -> Result { +async fn create_and_start_container(docker: &Docker, info: &ExecuteInfo, scope: ProfileScopeHandle) -> Result { // Generate unique (temporary) container name let container_name: String = format!("{}-{}", info.name, &uuid::Uuid::new_v4().to_string()[..6]); let create_options = CreateContainerOptions { name: &container_name, platform: None }; // Extract device requests from the capabilities + let guard = scope.time("Config initialization"); #[allow(clippy::unnecessary_filter_map)] let device_requests: Vec = info .capabilities @@ -588,16 +591,16 @@ async fn create_and_start_container(docker: &Docker, info: &ExecuteInfo) -> Resu // Create the container confic let create_config = Config { image: Some(info.image.name()), cmd: Some(info.command.clone()), host_config: Some(host_config), ..Default::default() }; + drop(guard); // Run it with that config debug!("Launching container with name '{}' (image: {})...", info.name, info.image.name()); - docker.create_container(Some(create_options), create_config).await.map_err(|source| Error::CreateContainerError { - name: info.name.clone(), - image: Box::new(info.image.clone()), - source, - })?; + scope + .time_fut("Container creation", docker.create_container(Some(create_options), create_config)) + .await + .map_err(|source| Error::CreateContainerError { name: info.name.clone(), image: Box::new(info.image.clone()), source })?; debug!(" > Container created"); - match docker.start_container(&container_name, None::>).await { + match scope.time_fut("Container launching", docker.start_container(&container_name, None::>)).await { Ok(_) => { debug!(" > Container '{}' started", container_name); Ok(container_name) @@ -613,26 +616,29 @@ async fn create_and_start_container(docker: &Docker, info: &ExecuteInfo) -> Resu /// - `name`: The name of the container to wait on. /// - `image`: The image that was run (used for debugging). /// - `keep_container`: Whether to keep the container around after it's finished or not. +/// - `scope`: Some [`ProfileScopeHandle`] to report profile timings in. Use [`ProfileScopeHandle::dummy()`] if you're not interested in the results. /// /// # Returns /// The return code of the docker container, its stdout and its stderr (in that order). /// /// # Errors /// This function may error for many reasons, which usually means that the container is unknown or the Docker engine is unreachable. -async fn join_container(docker: &Docker, name: &str, keep_container: bool) -> Result<(i32, String, String), Error> { +async fn join_container(docker: &Docker, name: &str, keep_container: bool, scope: ProfileScopeHandle) -> Result<(i32, String, String), Error> { // Wait for the container to complete - docker - .wait_container(name, None::>) - .try_collect::>() + scope + .time_fut("Container runtime", docker.wait_container(name, None::>).try_collect::>()) .await .map_err(|source| Error::WaitError { name: name.into(), source })?; // Get stdout and stderr logs from container let logs_options = Some(LogsOptions:: { stdout: true, stderr: true, ..Default::default() }); - let log_outputs = - docker.logs(name, logs_options).try_collect::>().await.map_err(|source| Error::LogsError { name: name.into(), source })?; + let log_outputs = scope + .time_fut("Log retrieval", docker.logs(name, logs_options).try_collect::>()) + .await + .map_err(|source| Error::LogsError { name: name.into(), source })?; // Collect them in one string per output channel + let guard = scope.time("Log postprocessing"); let mut stderr = String::new(); let mut stdout = String::new(); for log_output in log_outputs { @@ -644,13 +650,14 @@ async fn join_container(docker: &Docker, name: &str, keep_container: bool) -> Re }, } } + drop(guard); // Get the container's exit status by inspecting it - let code = returncode_container(docker, name).await?; + let code = scope.nest_fut("Retrieving container returncode", |scope| returncode_container(docker, name, scope)).await?; // Don't leave behind any waste: remove container (but only if told to do so!) if !keep_container { - remove_container(docker, name).await?; + scope.time_fut("Container removal", remove_container(docker, name)).await?; } // Return the return data of this container! @@ -662,17 +669,21 @@ async fn join_container(docker: &Docker, name: &str, keep_container: bool) -> Re /// # Arguments /// - `docker`: The Docker instance to use for accessing the container. /// - `name`: The container's name. +/// - `scope`: Some [`ProfileScopeHandle`] to report profile timings in. Use [`ProfileScopeHandle::dummy()`] if you're not interested in the results. /// /// # Returns /// The exit-/returncode that was returned by the container. /// /// # Errors /// This function errors if the Docker daemon could not be reached, such a container did not exist, could not be inspected or did not have a return code (yet). -async fn returncode_container(docker: &Docker, name: impl AsRef) -> Result { +async fn returncode_container(docker: &Docker, name: impl AsRef, scope: ProfileScopeHandle) -> Result { let name: &str = name.as_ref(); // Do the inspect call - let info = docker.inspect_container(name, None).await.map_err(|source| Error::InspectContainerError { name: name.into(), source })?; + let info = scope + .time_fut("Inspecting container", docker.inspect_container(name, None)) + .await + .map_err(|source| Error::InspectContainerError { name: name.into(), source })?; // Try to get the execution state from the container let state = match info.state { @@ -995,15 +1006,16 @@ pub async fn hash_container(container_path: impl AsRef) -> Result, source: impl Into) -> Result<(), Error> { +pub async fn ensure_image(docker: &Docker, image: impl Into, source: impl Into, scope: ProfileScopeHandle) -> Result<(), Error> { let image: Image = image.into(); let source: ImageSource = source.into(); // Abort if image is already loaded - match docker.inspect_image(&image.docker().to_string()).await { + match scope.time_fut("Checking image existance", docker.inspect_image(&image.docker().to_string())).await { Ok(_) => { debug!("Image '{}' already exists in Docker deamon.", image.docker()); return Ok(()); @@ -1020,12 +1032,12 @@ pub async fn ensure_image(docker: &Docker, image: impl Into, source: impl match source { ImageSource::Path(path) => { debug!(" > Importing file '{}'...", path.display()); - import_image(docker, image, path).await + scope.time_fut(format!("Importing image {:?}", path.display()), import_image(docker, image, path)).await }, ImageSource::Registry(source) => { debug!(" > Pulling image '{}'...", image); - pull_image(docker, image, source).await + scope.time_fut(format!("Pulling image {source:?}"), pull_image(docker, image, source)).await }, } } @@ -1107,10 +1119,10 @@ pub async fn launch(opts: impl AsRef, exec: ExecuteInfo) -> Resul let docker: Docker = connect_local(opts)?; // Either import or pull image, if not already present - ensure_image(&docker, &exec.image, &exec.image_source).await?; + ensure_image(&docker, &exec.image, &exec.image_source, ProfileScopeHandle::dummy()).await?; // Start container, return immediately (propagating any errors that occurred) - create_and_start_container(&docker, &exec).await + create_and_start_container(&docker, &exec, ProfileScopeHandle::dummy()).await } /// Joins the container with the given name, i.e., waits for it to complete and returns its results. @@ -1132,7 +1144,7 @@ pub async fn join(opts: impl AsRef, name: impl AsRef, keep_c let docker: Docker = connect_local(opts)?; // And now wait for it - join_container(&docker, name, keep_container).await + join_container(&docker, name, keep_container, ProfileScopeHandle::dummy()).await } /// Launches the given container and waits until its completed. @@ -1143,25 +1155,31 @@ pub async fn join(opts: impl AsRef, name: impl AsRef, keep_c /// - `opts`: The DockerOptions that contains information on how we can connect to the local daemon. /// - `exec`: The ExecuteInfo describing what to launch and how. /// - `keep_container`: If true, then will not remove the container after it has been launched. This is very useful for debugging. +/// - `scope`: Some [`ProfileScopeHandle`] to report profile timings in. Use [`ProfileScopeHandle::dummy()`] if you're not interested in the results. /// /// # Returns /// The return code of the docker container, its stdout and its stderr (in that order). /// /// # Errors /// This function errors for many reasons, some of which include not being able to connect to Docker or the container failing. -pub async fn run_and_wait(opts: impl AsRef, exec: ExecuteInfo, keep_container: bool) -> Result<(i32, String, String), Error> { +pub async fn run_and_wait( + opts: impl AsRef, + exec: ExecuteInfo, + keep_container: bool, + scope: ProfileScopeHandle, +) -> Result<(i32, String, String), Error> { // This next bit's basically launch but copied so that we have a docker connection of our own. // Connect to docker let docker: Docker = connect_local(opts)?; // Either import or pull image, if not already present - ensure_image(&docker, &exec.image, &exec.image_source).await?; + scope.nest_fut("Ensuring image", |scope| ensure_image(&docker, &exec.image, &exec.image_source, scope)).await?; // Start container, return immediately (propagating any errors that occurred) - let name: String = create_and_start_container(&docker, &exec).await?; + let name: String = scope.nest_fut("Launching container", |scope| create_and_start_container(&docker, &exec, scope)).await?; // And now wait for it - join_container(&docker, &name, keep_container).await + scope.nest_fut("Joining container", |scope| join_container(&docker, &name, keep_container, scope)).await } /// Tries to return the (IP-)address of the container with the given name. diff --git a/specifications/src/profiling.rs b/specifications/src/profiling.rs index c7ba8d6a..9957d143 100644 --- a/specifications/src/profiling.rs +++ b/specifications/src/profiling.rs @@ -12,7 +12,7 @@ //! A second version of the profiling library, with better support for //! generate dynamic yet pretty and (most of all) ordered profiling //! logs. -//! +//! //! Note that, while this library is not designed for Edge timings (i.e., //! user-relevant profiling), some parts of it can probably be re-used for //! that (especially the Timing struct). @@ -280,61 +280,54 @@ impl Drop for TimerGuard<'_> { -/// Provides a convenience wrapper around a reference to a [`ProfileScope`]. -#[derive(Clone, Debug)] -pub struct ProfileScopeHandle<'s> { - /// The actual scope itself. - scope: Arc, - /// A lifetime which allows us to assume the weak reference is valid. - _lifetime: PhantomData<&'s ()>, -} -impl ProfileScopeHandle<'static> { - /// Provides a dummy handle for if you are not interested in profiling, but need to use the functions. - #[inline] - pub fn dummy() -> Self { Self { scope: Arc::new(ProfileScope::new("<<>>")), _lifetime: PhantomData } } -} -impl ProfileScopeHandle<'_> { - /// Finishes a scope, by janking the handle wrapping it out-of-scope. - #[inline] - pub fn finish(self) {} -} -impl Deref for ProfileScopeHandle<'_> { - type Target = ProfileScope; - - #[inline] - fn deref(&self) -> &Self::Target { &self.scope } -} +// /// Provides a convenience wrapper around a reference to a [`ProfileScope`]. +// #[derive(Clone, Debug)] +// pub struct ProfileScopeHandle<'s> { +// /// The actual scope itself. +// scope: Arc, +// /// A lifetime which allows us to assume the weak reference is valid. +// _lifetime: PhantomData<&'s ()>, +// } +// impl ProfileScopeHandle<'static> { +// /// Provides a dummy handle for if you are not interested in profiling, but need to use the functions. +// #[inline] +// pub fn dummy() -> Self { Self { scope: Arc::new(ProfileScope::new("<<>>")), _lifetime: PhantomData } } +// } +// impl ProfileScopeHandle { +// /// Finishes a scope, by janking the handle wrapping it out-of-scope. +// #[inline] +// pub fn finish(self) {} +// } +// impl Deref for ProfileScopeHandle { +// type Target = ProfileScope; + +// #[inline] +// fn deref(&self) -> &Self::Target { &self.scope } +// } -/// Provides a convenience wrapper around a reference to a [`ProfileScope`] that ignores the lifetime mumbo. -/// -/// If this object outlives its parent scope, there won't be any errors; _however_, note that the profilings collected afterwards will not be printed. +/// Provides a convenience wrapper around a reference to a [`ProfileScope`]. #[derive(Clone, Debug)] -pub struct ProfileScopeHandleOwned { - /// The actual scope itself. +pub struct ProfileScopeHandle { + /// The scope we reference. scope: Arc, } -impl ProfileScopeHandleOwned { +impl ProfileScopeHandle { /// Provides a dummy handle for if you are not interested in profiling, but need to use the functions. #[inline] pub fn dummy() -> Self { Self { scope: Arc::new(ProfileScope::new("<<>>")) } } } -impl ProfileScopeHandleOwned { +impl ProfileScopeHandle { /// Finishes a scope, by janking the handle wrapping it out-of-scope. #[inline] pub fn finish(self) {} } -impl Deref for ProfileScopeHandleOwned { +impl Deref for ProfileScopeHandle { type Target = ProfileScope; #[inline] fn deref(&self) -> &Self::Target { &self.scope } } -impl<'s> From> for ProfileScopeHandleOwned { - #[inline] - fn from(value: ProfileScopeHandle<'s>) -> Self { Self { scope: value.scope } } -} - @@ -447,7 +440,7 @@ impl ProfileScope { /// /// # Returns /// A new `ProfileScope` instance. - pub fn new(name: impl Into) -> Self { Self { name: name.into(), timings: Mutex::new(vec![]) } } + pub fn new(name: impl Into) -> Self { Self { name: name.into(), timings: Mutex::new(Vec::new()) } } /// Returns a [`TimerGuard`], which takes a time exactly as long as it is in scope. /// @@ -533,14 +526,14 @@ impl ProfileScope { /// A new `ProfileScope` that can be used to take timings. pub fn nest(&self, name: impl Into) -> ProfileScopeHandle { // Create the new scope - let scope: Self = Self::new(name); + let scope: Arc = Arc::new(Self::new(name)); // Insert it internally let mut lock: MutexGuard> = self.timings.lock(); - lock.push(ProfileTiming::Scope(Arc::new(scope))); + lock.push(ProfileTiming::Scope(scope.clone())); // Return a weak reference to it - ProfileScopeHandle { scope: lock.last().unwrap().scope().clone(), _lifetime: PhantomData } + ProfileScopeHandle { scope: lock.last().unwrap().scope().clone() } } /// Profiles the given function, but provides it with extra profile options by giving it its own `ProfileScope` to populate. @@ -560,7 +553,7 @@ impl ProfileScope { // Add an entry for the scope let timing: Arc> = { let mut lock: MutexGuard> = scope.timings.lock(); - lock.push(ProfileTiming::Timing("total".into(), Arc::new(Mutex::new(Timing::none())))); + lock.push(ProfileTiming::Timing("Total".into(), Arc::new(Mutex::new(Timing::none())))); lock.last().unwrap().timing().clone() }; @@ -587,10 +580,10 @@ impl ProfileScope { /// /// # Returns /// A future that returns the same result as the given, but times its execution as a side-effect. - pub fn nest_fut<'s, F: Future>( + pub fn nest_fut<'s, 'f: 's, F: Future>( &'s self, name: impl Into, - fut: impl 's + FnOnce(ProfileScopeHandle<'s>) -> F, + fut: impl 'f + FnOnce(ProfileScopeHandle) -> F, ) -> impl 's + Future { let name: String = name.into(); @@ -600,7 +593,7 @@ impl ProfileScope { // Add an entry for the scope let timing: Arc> = { let mut lock: MutexGuard> = scope.timings.lock(); - lock.push(ProfileTiming::Timing("total".into(), Arc::new(Mutex::new(Timing::none())))); + lock.push(ProfileTiming::Timing("Total".into(), Arc::new(Mutex::new(Timing::none())))); lock.last().unwrap().timing().clone() };