diff --git a/Cargo.toml b/Cargo.toml index 5913ebdad42..22460a27c47 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,6 +76,7 @@ members = [ "benchmarks/cu_async_cl_io_bench", "benchmarks/cu_dorabench", "benchmarks/cu_zenoh_bridge_bench", + "examples/cu_anytime_task", "examples/cu_background_task", "examples/cu_baremetal_safety", "examples/cu_bridge_test", diff --git a/api/v1/cu29-runtime.txt b/api/v1/cu29-runtime.txt index 3e43eeff5e4..c7bc5759a06 100644 --- a/api/v1/cu29-runtime.txt +++ b/api/v1/cu29-runtime.txt @@ -525,8 +525,8 @@ pub type cu29_runtime::cuasynctask::CuAsyncSrcTask::Output<'m> = ::Resources<'r> = cu29_runtime::cuasynctask::CuAsyncSrcTaskResources<'r, T> pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::debug_state_type_path() -> &'static str where Self: cu29_runtime::reflect::TypePath + core::marker::Sized pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::new(config: core::option::Option<&cu29_runtime::config::ComponentConfig>, resources: Self::Resources) -> cu29_traits::CuResult where Self: core::marker::Sized -pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::postprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> -pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::preprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> +pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::postprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> +pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::preprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::process<'o>(&mut self, ctx: &cu29_runtime::context::CuContext, real_output: &mut Self::Output) -> cu29_traits::CuResult<()> pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::register_debug_state_types(registry: &mut cu29_runtime::reflect::TypeRegistry) where Self: cu29_runtime::reflect::GetTypeRegistration + core::marker::Sized pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::start(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> @@ -553,8 +553,8 @@ pub type cu29_runtime::cuasynctask::CuAsyncTask::Output<'m> = ::Resources<'r> = cu29_runtime::cuasynctask::CuAsyncTaskResources<'r, T> pub fn cu29_runtime::cuasynctask::CuAsyncTask::debug_state_type_path() -> &'static str where Self: cu29_runtime::reflect::TypePath + core::marker::Sized pub fn cu29_runtime::cuasynctask::CuAsyncTask::new(config: core::option::Option<&cu29_runtime::config::ComponentConfig>, resources: Self::Resources) -> cu29_traits::CuResult where Self: core::marker::Sized -pub fn cu29_runtime::cuasynctask::CuAsyncTask::postprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> -pub fn cu29_runtime::cuasynctask::CuAsyncTask::preprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> +pub fn cu29_runtime::cuasynctask::CuAsyncTask::postprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> +pub fn cu29_runtime::cuasynctask::CuAsyncTask::preprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> pub fn cu29_runtime::cuasynctask::CuAsyncTask::process<'i, 'o>(&mut self, ctx: &cu29_runtime::context::CuContext, input: &Self::Input, real_output: &mut Self::Output) -> cu29_traits::CuResult<()> pub fn cu29_runtime::cuasynctask::CuAsyncTask::register_debug_state_types(registry: &mut cu29_runtime::reflect::TypeRegistry) where Self: cu29_runtime::reflect::GetTypeRegistration + core::marker::Sized pub fn cu29_runtime::cuasynctask::CuAsyncTask::start(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> @@ -920,8 +920,8 @@ pub type cu29_runtime::cuasynctask::CuAsyncSrcTask::Output<'m> = ::Resources<'r> = cu29_runtime::cuasynctask::CuAsyncSrcTaskResources<'r, T> pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::debug_state_type_path() -> &'static str where Self: cu29_runtime::reflect::TypePath + core::marker::Sized pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::new(config: core::option::Option<&cu29_runtime::config::ComponentConfig>, resources: Self::Resources) -> cu29_traits::CuResult where Self: core::marker::Sized -pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::postprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> -pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::preprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> +pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::postprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> +pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::preprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::process<'o>(&mut self, ctx: &cu29_runtime::context::CuContext, real_output: &mut Self::Output) -> cu29_traits::CuResult<()> pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::register_debug_state_types(registry: &mut cu29_runtime::reflect::TypeRegistry) where Self: cu29_runtime::reflect::GetTypeRegistration + core::marker::Sized pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask::start(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> @@ -958,8 +958,8 @@ pub type cu29_runtime::cuasynctask::CuAsyncTask::Output<'m> = ::Resources<'r> = cu29_runtime::cuasynctask::CuAsyncTaskResources<'r, T> pub fn cu29_runtime::cuasynctask::CuAsyncTask::debug_state_type_path() -> &'static str where Self: cu29_runtime::reflect::TypePath + core::marker::Sized pub fn cu29_runtime::cuasynctask::CuAsyncTask::new(config: core::option::Option<&cu29_runtime::config::ComponentConfig>, resources: Self::Resources) -> cu29_traits::CuResult where Self: core::marker::Sized -pub fn cu29_runtime::cuasynctask::CuAsyncTask::postprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> -pub fn cu29_runtime::cuasynctask::CuAsyncTask::preprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> +pub fn cu29_runtime::cuasynctask::CuAsyncTask::postprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> +pub fn cu29_runtime::cuasynctask::CuAsyncTask::preprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> pub fn cu29_runtime::cuasynctask::CuAsyncTask::process<'i, 'o>(&mut self, ctx: &cu29_runtime::context::CuContext, input: &Self::Input, real_output: &mut Self::Output) -> cu29_traits::CuResult<()> pub fn cu29_runtime::cuasynctask::CuAsyncTask::register_debug_state_types(registry: &mut cu29_runtime::reflect::TypeRegistry) where Self: cu29_runtime::reflect::GetTypeRegistration + core::marker::Sized pub fn cu29_runtime::cuasynctask::CuAsyncTask::start(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()> diff --git a/core/cu29_derive/src/lib.rs b/core/cu29_derive/src/lib.rs index 21594a23af6..07dc64a295a 100644 --- a/core/cu29_derive/src/lib.rs +++ b/core/cu29_derive/src/lib.rs @@ -1726,15 +1726,6 @@ pub fn copper_runtime(args: TokenStream, input: TokenStream) -> TokenStream { Err(e) => return return_error(format!("Could not compute copperlist plan: {e}")), }; - // Anytime restrictions the runner imposes on top of config validation. - for (index, anytime) in task_specs.anytime_configs.iter().enumerate() { - if anytime.is_some() && task_specs.background_flags[index] { - return return_error(format!( - "Anytime task '{}' cannot use background: true yet: the background anytime runner is not implemented. Run it in the foreground for now.", - task_specs.ids[index] - )); - } - } // Single-input/single-output arity is validated at configuration time // (config.rs validate_anytime_graph), before the plan is built. @@ -2311,7 +2302,7 @@ pub fn copper_runtime(args: TokenStream, input: TokenStream) -> TokenStream { && !(sim_mode && task_specs.cutypes[index] == CuTaskType::Source && !task_specs.run_in_sim_flags[index]); - let inner_task_type = &task_specs.sim_task_types[index]; + let inner_task_type = &task_specs.inner_task_types[index]; match task_specs.cutypes[index] { CuTaskType::Source => { if background { @@ -2415,7 +2406,7 @@ pub fn copper_runtime(args: TokenStream, input: TokenStream) -> TokenStream { task_specs.type_names[index], index ); let mapping_ref = task_resource_mappings.refs[index].clone(); - let inner_task_type = &task_specs.sim_task_types[index]; + let inner_task_type = &task_specs.inner_task_types[index]; match task_specs.cutypes[index] { CuTaskType::Source => { if *background { @@ -6113,8 +6104,13 @@ fn task_trait_for_kind(task_kind: CuTaskType) -> proc_macro2::TokenStream { /// Like [`task_trait_for_kind`], but resolves anytime nodes to /// `CuAnytimeTask` (they stay `Regular` in the graph but implement the /// anytime trait instead of `CuTask`). +/// +/// Background comes first: a backgrounded node is driven through the +/// `CuAsyncTask` wrapper, which is a plain `CuTask` whatever it wraps. fn task_trait_for_specs(task_specs: &CuTaskSpecSet, index: usize) -> proc_macro2::TokenStream { - if task_specs.anytime_configs[index].is_some() { + let foreground_anytime = + task_specs.anytime_configs[index].is_some() && !task_specs.background_flags[index]; + if foreground_anytime { quote! { cu29::cutask_anytime::CuAnytimeTask } } else { task_trait_for_kind(task_specs.cutypes[index]) @@ -6166,6 +6162,10 @@ struct CuTaskSpecSet { /// are baked into a per-node `AnytimePolicy` ZST and the emitted /// base/refine steps. pub anytime_configs: Vec>, + /// The task type inside the optional async wrapper: the anytime runner + /// for a background anytime node, the declared task type for everything + /// else — including every foreground task, where nothing wraps it. + pub inner_task_types: Vec, pub logging_enabled: Vec, pub type_names: Vec, pub task_types: Vec, @@ -6186,7 +6186,7 @@ impl CuTaskSpecSet { .filter(|(_, node)| node.get_flavor() == Flavor::Task) .collect(); - let ids = all_id_nodes + let ids: Vec = all_id_nodes .iter() .map(|(_, node)| node.get_id().to_string()) .collect(); @@ -6252,21 +6252,39 @@ impl CuTaskSpecSet { }) .collect(); + // A background anytime node is handed to `CuAsyncTask` wrapped in the + // runner, which turns one whole job into a single `process` call. + let inner_task_types: Vec = parsed_task_types + .iter() + .zip(ids.iter()) + .zip(background_flags.iter()) + .zip(anytime_configs.iter()) + .map(|(((task_type, id), &background), anytime)| { + if background && anytime.is_some() { + let policy_ident = anytime_policy_ident(id.as_str()); + parse_quote!(cu29::cutask_anytime::CuAnytimeRunner<#task_type, #policy_ident>) + } else { + task_type.clone() + } + }) + .collect(); + let task_types = parsed_task_types .iter() .zip(type_names.iter()) .zip(cutypes.iter()) .zip(background_flags.iter()) .zip(output_types.iter()) - .map(|((((name_type, name), cutype), &background), output_type)| { + .zip(inner_task_types.iter()) + .map(|(((((name_type, name), cutype), &background), output_type), inner_type)| { if background { if let Some(output_type) = output_type { match cutype { CuTaskType::Source => { - parse_quote!(CuAsyncSrcTask<#name_type, #output_type>) + parse_quote!(CuAsyncSrcTask<#inner_type, #output_type>) } CuTaskType::Regular => { - parse_quote!(CuAsyncTask<#name_type, #output_type>) + parse_quote!(CuAsyncTask<#inner_type, #output_type>) } CuTaskType::Sink => { panic!("CuSinkTask {name} cannot be a background task, it should be a regular task."); @@ -6290,15 +6308,16 @@ impl CuTaskSpecSet { .zip(cutypes.iter()) .zip(background_flags.iter()) .zip(output_types.iter()) - .map(|((((name_type, name), cutype), &background), output_type)| { + .zip(inner_task_types.iter()) + .map(|(((((name_type, name), cutype), &background), output_type), inner_type)| { if background { if let Some(output_type) = output_type { match cutype { CuTaskType::Source => { - parse_quote!(CuAsyncSrcTask::<#name_type, #output_type>) + parse_quote!(CuAsyncSrcTask::<#inner_type, #output_type>) } CuTaskType::Regular => { - parse_quote!(CuAsyncTask::<#name_type, #output_type>) + parse_quote!(CuAsyncTask::<#inner_type, #output_type>) } CuTaskType::Sink => { panic!("CuSinkTask {name} cannot be a background task, it should be a regular task."); @@ -6334,6 +6353,7 @@ impl CuTaskSpecSet { background_flags, background_pools, anytime_configs, + inner_task_types, logging_enabled, type_names, task_types, @@ -7814,11 +7834,9 @@ fn build_task_resource_mappings( continue; } - let binding_task_type = if task_specs.background_flags[idx] { - &task_specs.sim_task_types[idx] - } else { - &task_specs.task_types[idx] - }; + // A backgrounded task binds the resources of what the wrapper wraps — + // the runner for an anytime node, the task itself otherwise. + let binding_task_type = &task_specs.inner_task_types[idx]; let binding_trait = task_trait_for_specs(task_specs, idx); @@ -8561,10 +8579,17 @@ fn build_anytime_policy_defs(task_specs: &CuTaskSpecSet) -> Vec quote! { Some(#stall) }, None => quote! { None }, }; + // Only the background runner reads MAX_REFINES: a foreground node + // carries the count as the number of refine steps in its plan. + let max_refines = match anytime.max_refines { + Some(refines) => quote! { Some(#refines) }, + None => quote! { None }, + }; let consts = quote! { const TIME_BUDGET: Option = #time_budget; const MAX_AGE: Option = #max_age; const MAX_STALL: Option = #max_stall; + const MAX_REFINES: Option = #max_refines; }; let has_quality_knob = anytime.quality_target.is_some() || anytime.quality_floor.is_some() @@ -8639,7 +8664,9 @@ fn build_anytime_job_locals(task_specs: &CuTaskSpecSet) -> Vec tov_time, - // A Range anchors on its earliest data: the entire input - // window must remain within max_age. - cu29::clock::Tov::Range(tov_range) => tov_range.start, - cu29::clock::Tov::None => __cu_any_now, - }; + let __cu_any_anchor = cu29::cutask_anytime::anchor_from_tov(cumsg_input.tov, __cu_any_now); if __cu_any_now >= __cu_any_anchor + cu29::clock::CuDuration(#max_age_nanos) { let __cu_any_outcome = cu29::cutask_anytime::skip_stale(cumsg_output); debug!(ctx, "Anytime task {}: input dead on arrival, job skipped.", #task_id); @@ -10023,7 +10044,8 @@ fn runtime_task_type_for_index( CuTaskType::Regular => { if background { if let Some(out_ty) = output_type { - parse_quote!(CuAsyncTask<#declared_task_type, #out_ty>) + let inner = &task_specs.inner_task_types[index]; + parse_quote!(CuAsyncTask<#inner, #out_ty>) } else { panic!("{task_id}: If a task is background, it has to have an output"); } diff --git a/core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task.rs b/core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task.rs new file mode 100644 index 00000000000..4f86c77f762 --- /dev/null +++ b/core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task.rs @@ -0,0 +1,205 @@ +//! Background placement of anytime nodes: `anytime:` + `background: true` +//! compiles to `CuAsyncTask, O>`, so the node keeps +//! one `Whole` step in the plan and runs its whole job on a worker thread. +//! +//! The config covers, in one application: a background node bounded by time +//! alone (no `max_refines`, which only a foreground plan needs), a background +//! node on a named pool that also binds resources, and a foreground anytime +//! node alongside them. +use cu29::cutask_anytime::{AnytimeStatus, CuAnytimeTask, Quality}; +use cu29::prelude::*; +use cu29::resource::{BundleContext, ResourceBundle, ResourceManager}; +use cu29::{bundle_resources, resources}; +use cu29_derive::copper_runtime; + +pub struct TestBundle; + +bundle_resources!(TestBundle: Scratch = "scratch"); + +impl ResourceBundle for TestBundle { + fn build( + bundle: BundleContext, + _config: Option<&ComponentConfig>, + manager: &mut ResourceManager, + ) -> CuResult<()> { + manager.add_owned(bundle.key(TestBundleId::Scratch), String::from("scratch"))?; + Ok(()) + } +} + +mod counter_resources { + use super::*; + + resources!({ + scratch => Owned, + }); +} + +type CounterResources = counter_resources::Resources; + +#[derive(Reflect)] +struct AnytimeSrc; + +impl Freezable for AnytimeSrc {} + +impl CuSrcTask for AnytimeSrc { + type Resources<'r> = (); + type Output<'m> = output_msg!(u32); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult { + Ok(Self) + } + + fn process(&mut self, _ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> { + output.set_payload(3); + Ok(()) + } +} + +/// Full policy surface: comparable quality, target/floor/stall knobs. +#[derive(Reflect)] +struct AnytimeRefiner { + target: u32, + acc: u32, +} + +impl Freezable for AnytimeRefiner {} + +impl CuAnytimeTask for AnytimeRefiner { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = (); + type Quality = Quality; + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult { + Ok(Self { target: 0, acc: 0 }) + } + + fn base( + &mut self, + _ctx: &CuContext, + input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult> { + self.target = input.payload().copied().unwrap_or(0); + self.acc = 0; + output.set_payload(self.acc); + Ok(AnytimeStatus::Improved(cu29::cutask_anytime::quality_from_f32(0.0))) + } + + fn refine( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'_>, + ) -> CuResult> { + if self.acc >= self.target { + return Ok(AnytimeStatus::Converged( + cu29::cutask_anytime::quality_from_f32(1.0), + )); + } + self.acc += 1; + output.set_payload(self.acc); + Ok(AnytimeStatus::Improved( + cu29::cutask_anytime::quality_from_f32(self.acc as f32 / self.target as f32), + )) + } +} + +/// Quality-less anytime task: only hard bounds are expressible. +#[derive(Reflect)] +struct AnytimeCounter; + +impl Freezable for AnytimeCounter {} + +impl CuAnytimeTask for AnytimeCounter { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = CounterResources; + type Quality = (); + + fn new(_config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult { + let _scratch = resources.scratch.0; + Ok(Self) + } + + fn base( + &mut self, + _ctx: &CuContext, + input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult> { + output.set_payload(input.payload().copied().unwrap_or(0)); + Ok(AnytimeStatus::Improved(())) + } + + fn refine( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'_>, + ) -> CuResult> { + let bumped = output.payload().copied().unwrap_or(0) + 1; + output.set_payload(bumped); + Ok(AnytimeStatus::Improved(())) + } +} + +/// Anytime node with a declared `kind: task` and no outgoing connection: its +/// output slot type is inferred through `CuAnytimeTask` (autogenerated nc). +#[derive(Reflect)] +struct AnytimeTail; + +impl Freezable for AnytimeTail {} + +impl CuAnytimeTask for AnytimeTail { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = (); + type Quality = (); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult { + Ok(Self) + } + + fn base( + &mut self, + _ctx: &CuContext, + input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult> { + output.set_payload(input.payload().copied().unwrap_or(0)); + Ok(AnytimeStatus::Improved(())) + } + + fn refine( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'_>, + ) -> CuResult> { + let bumped = output.payload().copied().unwrap_or(0) + 1; + output.set_payload(bumped); + Ok(AnytimeStatus::Improved(())) + } +} + +#[derive(Reflect)] +struct AnytimeSink; + +impl Freezable for AnytimeSink {} + +impl CuSinkTask for AnytimeSink { + type Resources<'r> = (); + type Input<'m> = input_msg!(u32); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult { + Ok(Self) + } + + fn process(&mut self, _ctx: &CuContext, _input: &Self::Input<'_>) -> CuResult<()> { + Ok(()) + } +} + +#[copper_runtime(config = "config/anytime_background_task_valid.ron")] +struct App {} + +fn main() {} diff --git a/core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task_sim.rs b/core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task_sim.rs new file mode 100644 index 00000000000..b0cb4d34030 --- /dev/null +++ b/core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task_sim.rs @@ -0,0 +1,201 @@ +//! Sim-mode expansion of a backgrounded anytime node: it is indistinguishable +//! from any other backgrounded task — one sim callback, one poll step. +//! +//! Same config as `anytime_background_task`, expanded with `sim_mode = true`. +use cu29::cutask_anytime::{AnytimeStatus, CuAnytimeTask, Quality}; +use cu29::prelude::*; +use cu29::resource::{BundleContext, ResourceBundle, ResourceManager}; +use cu29::{bundle_resources, resources}; +use cu29_derive::copper_runtime; + +pub struct TestBundle; + +bundle_resources!(TestBundle: Scratch = "scratch"); + +impl ResourceBundle for TestBundle { + fn build( + bundle: BundleContext, + _config: Option<&ComponentConfig>, + manager: &mut ResourceManager, + ) -> CuResult<()> { + manager.add_owned(bundle.key(TestBundleId::Scratch), String::from("scratch"))?; + Ok(()) + } +} + +mod counter_resources { + use super::*; + + resources!({ + scratch => Owned, + }); +} + +type CounterResources = counter_resources::Resources; + +#[derive(Reflect)] +struct AnytimeSrc; + +impl Freezable for AnytimeSrc {} + +impl CuSrcTask for AnytimeSrc { + type Resources<'r> = (); + type Output<'m> = output_msg!(u32); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult { + Ok(Self) + } + + fn process(&mut self, _ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> { + output.set_payload(3); + Ok(()) + } +} + +/// Full policy surface: comparable quality, target/floor/stall knobs. +#[derive(Reflect)] +struct AnytimeRefiner { + target: u32, + acc: u32, +} + +impl Freezable for AnytimeRefiner {} + +impl CuAnytimeTask for AnytimeRefiner { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = (); + type Quality = Quality; + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult { + Ok(Self { target: 0, acc: 0 }) + } + + fn base( + &mut self, + _ctx: &CuContext, + input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult> { + self.target = input.payload().copied().unwrap_or(0); + self.acc = 0; + output.set_payload(self.acc); + Ok(AnytimeStatus::Improved(cu29::cutask_anytime::quality_from_f32(0.0))) + } + + fn refine( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'_>, + ) -> CuResult> { + if self.acc >= self.target { + return Ok(AnytimeStatus::Converged( + cu29::cutask_anytime::quality_from_f32(1.0), + )); + } + self.acc += 1; + output.set_payload(self.acc); + Ok(AnytimeStatus::Improved( + cu29::cutask_anytime::quality_from_f32(self.acc as f32 / self.target as f32), + )) + } +} + +/// Quality-less anytime task: only hard bounds are expressible. +#[derive(Reflect)] +struct AnytimeCounter; + +impl Freezable for AnytimeCounter {} + +impl CuAnytimeTask for AnytimeCounter { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = CounterResources; + type Quality = (); + + fn new(_config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult { + let _scratch = resources.scratch.0; + Ok(Self) + } + + fn base( + &mut self, + _ctx: &CuContext, + input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult> { + output.set_payload(input.payload().copied().unwrap_or(0)); + Ok(AnytimeStatus::Improved(())) + } + + fn refine( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'_>, + ) -> CuResult> { + let bumped = output.payload().copied().unwrap_or(0) + 1; + output.set_payload(bumped); + Ok(AnytimeStatus::Improved(())) + } +} + +/// Anytime node with a declared `kind: task` and no outgoing connection: its +/// output slot type is inferred through `CuAnytimeTask` (autogenerated nc). +#[derive(Reflect)] +struct AnytimeTail; + +impl Freezable for AnytimeTail {} + +impl CuAnytimeTask for AnytimeTail { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = (); + type Quality = (); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult { + Ok(Self) + } + + fn base( + &mut self, + _ctx: &CuContext, + input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult> { + output.set_payload(input.payload().copied().unwrap_or(0)); + Ok(AnytimeStatus::Improved(())) + } + + fn refine( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'_>, + ) -> CuResult> { + let bumped = output.payload().copied().unwrap_or(0) + 1; + output.set_payload(bumped); + Ok(AnytimeStatus::Improved(())) + } +} + +#[derive(Reflect)] +struct AnytimeSink; + +impl Freezable for AnytimeSink {} + +impl CuSinkTask for AnytimeSink { + type Resources<'r> = (); + type Input<'m> = input_msg!(u32); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult { + Ok(Self) + } + + fn process(&mut self, _ctx: &CuContext, _input: &Self::Input<'_>) -> CuResult<()> { + Ok(()) + } +} + +#[copper_runtime(config = "config/anytime_background_task_valid.ron", sim_mode = true)] +struct App {} + +fn main() {} diff --git a/core/cu29_derive/tests/config/anytime_background_task_valid.ron b/core/cu29_derive/tests/config/anytime_background_task_valid.ron new file mode 100644 index 00000000000..429544f16c4 --- /dev/null +++ b/core/cu29_derive/tests/config/anytime_background_task_valid.ron @@ -0,0 +1,65 @@ +( + runtime: ( + thread_pools: [(id: "planner", threads: 1)], + ), + resources: [ + ( + id: "board", + provider: "TestBundle", + ), + ], + tasks: [ + (id: "src", type: "AnytimeSrc"), + ( + id: "refiner", + type: "AnytimeRefiner", + background: true, + anytime: ( + time_budget_ms: 5.0, + max_age_ms: 100.0, + quality_target: 0.9, + quality_floor: 0.1, + max_stall: 3, + ), + ), + ( + id: "counter", + type: "AnytimeCounter", + background: (pool: "planner"), + anytime: (max_refines: 2), + resources: {"scratch": "board.scratch"}, + ), + ( + id: "sink", + type: "AnytimeSink", + ), + ( + id: "tail", + type: "AnytimeTail", + kind: task, + anytime: (max_refines: 1), + ), + ], + cnx: [ + ( + src: "src", + dst: "refiner", + msg: "u32", + ), + ( + src: "refiner", + dst: "counter", + msg: "u32", + ), + ( + src: "counter", + dst: "sink", + msg: "u32", + ), + ( + src: "counter", + dst: "tail", + msg: "u32", + ), + ], +) \ No newline at end of file diff --git a/core/cu29_runtime/src/cuasynctask.rs b/core/cu29_runtime/src/cuasynctask.rs index 56fa11b39ff..444da6f3020 100644 --- a/core/cu29_runtime/src/cuasynctask.rs +++ b/core/cu29_runtime/src/cuasynctask.rs @@ -8,7 +8,7 @@ use bincode::error::{DecodeError, EncodeError}; use cu29_clock::CuTime; use cu29_traits::{CuError, CuResult}; use rayon::ThreadPool; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, TryLockError}; struct AsyncState { processing: bool, @@ -65,6 +65,21 @@ where Ok(output) } +/// Best-effort forwarding for `preprocess`/`postprocess`: the hook runs when +/// the wrapped task is idle and is skipped — never blocked on — while a +/// background run holds it. +fn try_forward( + task: &Mutex, + poisoned: &'static str, + hook: impl FnOnce(&mut T) -> CuResult<()>, +) -> CuResult<()> { + match task.try_lock() { + Ok(mut task) => hook(&mut task), + Err(TryLockError::WouldBlock) => Ok(()), + Err(TryLockError::Poisoned(_)) => Err(CuError::from(poisoned)), + } +} + fn record_async_error(state: &Mutex, error: CuError) { let mut guard = match state.lock() { Ok(guard) => guard, @@ -289,6 +304,22 @@ where task.start(ctx) } + fn preprocess(&mut self, ctx: &CuContext) -> CuResult<()> { + try_forward( + &self.task, + "Async task mutex poisoned during preprocess", + |t| t.preprocess(ctx), + ) + } + + fn postprocess(&mut self, ctx: &CuContext) -> CuResult<()> { + try_forward( + &self.task, + "Async task mutex poisoned during postprocess", + |t| t.postprocess(ctx), + ) + } + fn process<'i, 'o>( &mut self, ctx: &CuContext, @@ -491,6 +522,22 @@ where task.start(ctx) } + fn preprocess(&mut self, ctx: &CuContext) -> CuResult<()> { + try_forward( + &self.task, + "Async source mutex poisoned during preprocess", + |t| t.preprocess(ctx), + ) + } + + fn postprocess(&mut self, ctx: &CuContext) -> CuResult<()> { + try_forward( + &self.task, + "Async source mutex poisoned during postprocess", + |t| t.postprocess(ctx), + ) + } + fn process<'o>(&mut self, ctx: &CuContext, real_output: &mut Self::Output<'o>) -> CuResult<()> { if !begin_background_poll(ctx, &self.state, &self.output, real_output)? { return Ok(()); @@ -546,12 +593,17 @@ mod tests { use crate::config::ComponentConfig; use crate::cutask::CuMsg; use crate::cutask::Freezable; + use crate::cutask_anytime::{ + AnytimePolicy, AnytimeStatus, CuAnytimeRunner, CuAnytimeTask, Quality, quality_from_f32, + }; use crate::input_msg; use crate::output_msg; + use cu29_clock::CuDuration; use cu29_traits::CuResult; use rayon::ThreadPoolBuilder; use std::borrow::BorrowMut; use std::sync::OnceLock; + use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::mpsc; use std::time::Duration; @@ -783,6 +835,116 @@ mod tests { } } + #[derive(Clone)] + struct HookTaskResources { + entered: mpsc::Sender<()>, + release: Arc>>, + hooks: Arc, + } + + /// Counts pre/postprocess calls; `process` blocks holding the task mutex + /// until the test releases it. + #[derive(Reflect)] + #[reflect(no_field_bounds, from_reflect = false)] + struct HookTask { + #[reflect(ignore)] + entered: mpsc::Sender<()>, + #[reflect(ignore)] + release: Arc>>, + #[reflect(ignore)] + hooks: Arc, + } + + impl Freezable for HookTask {} + + impl CuTask for HookTask { + type Resources<'r> = HookTaskResources; + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + + fn new(_config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult + where + Self: Sized, + { + Ok(Self { + entered: resources.entered, + release: resources.release, + hooks: resources.hooks, + }) + } + + fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> { + self.hooks.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + + fn process( + &mut self, + _ctx: &CuContext, + _input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult<()> { + let _ = self.entered.send(()); + self.release + .lock() + .unwrap() + .recv_timeout(Duration::from_secs(1)) + .expect("timed out waiting for release"); + output.set_payload(1); + Ok(()) + } + + fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> { + self.hooks.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + } + + #[test] + fn background_hooks_forward_only_while_idle() { + let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap()); + let context = CuContext::new_with_clock(); + let (entered_tx, entered_rx) = mpsc::channel::<()>(); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let hooks = Arc::new(AtomicU32::new(0)); + let resources = HookTaskResources { + entered: entered_tx, + release: Arc::new(Mutex::new(release_rx)), + hooks: hooks.clone(), + }; + let mut task: CuAsyncTask = + CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap(); + + // Idle: both hooks reach the wrapped task. + task.preprocess(&context).unwrap(); + task.postprocess(&context).unwrap(); + assert_eq!(hooks.load(Ordering::SeqCst), 2); + + // Dispatch a job and wait for the worker to hold the task. + let input = CuMsg::new(Some(1u32)); + let mut output = CuMsg::new(None); + task.process(&context, &input, &mut output).unwrap(); + entered_rx + .recv_timeout(Duration::from_secs(1)) + .expect("worker never picked up the job"); + + // Busy: both hooks are skipped, not blocked on. + task.preprocess(&context).unwrap(); + task.postprocess(&context).unwrap(); + assert_eq!(hooks.load(Ordering::SeqCst), 2); + + release_tx.send(()).unwrap(); + // The worker releases the task shortly after process returns. + for _ in 0..1000 { + task.preprocess(&context).unwrap(); + if hooks.load(Ordering::SeqCst) == 3 { + return; + } + std::thread::sleep(Duration::from_millis(1)); + } + panic!("preprocess never reached the task after the job finished"); + } + #[derive(Clone)] struct ControlledSrcResources { ready_times: Arc>>, @@ -1061,6 +1223,85 @@ mod tests { let _ = done_rx.recv_timeout(Duration::from_secs(1)); } + /// Anytime task under the wrapper: one increment per quantum, quality + /// climbing toward the input target. + #[derive(Reflect)] + struct IncrementalPlanner { + target: u32, + acc: u32, + } + + impl Freezable for IncrementalPlanner {} + + impl CuAnytimeTask for IncrementalPlanner { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = (); + type Quality = Quality; + + fn new(_config: Option<&ComponentConfig>, _resources: ()) -> CuResult { + Ok(Self { target: 0, acc: 0 }) + } + + fn base( + &mut self, + _ctx: &CuContext, + input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult> { + self.target = input.payload().copied().ok_or("planner: no input")?; + self.acc = 0; + output.set_payload(self.acc); + Ok(AnytimeStatus::Improved(quality_from_f32(0.0))) + } + + fn refine( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'_>, + ) -> CuResult> { + self.acc += 1; + output.set_payload(self.acc); + Ok(AnytimeStatus::Improved(quality_from_f32( + self.acc as f32 / self.target as f32, + ))) + } + } + + /// Mirrors codegen for `anytime: (max_refines: 3)` on a background node. + struct ThreeQuantaPolicy; + impl AnytimePolicy for ThreeQuantaPolicy { + const TIME_BUDGET: Option = None; + const MAX_AGE: Option = None; + const MAX_STALL: Option = None; + const MAX_REFINES: Option = Some(3); + } + + #[test] + fn background_anytime_job_lands_with_its_status_stamp() { + let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap()); + let context = CuContext::new_with_clock(); + let mut task: CuAsyncTask, u32> = + CuAsyncTask::new(Some(&ComponentConfig::default()), (), tp).unwrap(); + + let input = CuMsg::new(Some(5u32)); + let mut output = CuMsg::new(None); + + // Poll until the worker's job comes back through the buffered output. + for _ in 0..1000 { + task.process(&context, &input, &mut output).unwrap(); + if output.payload().is_some() { + break; + } + std::thread::sleep(Duration::from_millis(1)); + } + + // Three quanta of a job needing five: stopped by the quanta bound, and + // the stamp the runner wrote survived the buffered-output copy. + assert_eq!(output.payload(), Some(&3)); + assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=0.60 max"); + } + #[test] fn background_source_respects_recorded_ready_time() { let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap()); diff --git a/core/cu29_runtime/src/cutask_anytime.rs b/core/cu29_runtime/src/cutask_anytime.rs index 2aac21f9fde..70d31adec4f 100644 --- a/core/cu29_runtime/src/cutask_anytime.rs +++ b/core/cu29_runtime/src/cutask_anytime.rs @@ -9,12 +9,15 @@ use crate::config::ComponentConfig; use crate::context::CuContext; -use crate::cutask::{CuMsg, CuMsgPack, CuMsgPayload, Freezable}; +use crate::cutask::{CuMsg, CuMsgPack, CuMsgPayload, CuTask, Freezable}; use crate::reflect::{GetTypeRegistration, Reflect, TypePath, TypeRegistry}; +use bincode::de::Decoder; +use bincode::enc::Encoder; +use bincode::error::{DecodeError, EncodeError}; use compact_str::format_compact; use core::fmt::{Debug, Formatter, Result as FmtResult}; use core::marker::PhantomData; -use cu29_clock::{CuDuration, CuTime}; +use cu29_clock::{CuDuration, CuTime, Tov}; use cu29_traits::{CuCompactString, CuResult}; use cu29_units::si::f32::Ratio; use cu29_units::si::ratio::ratio; @@ -207,8 +210,7 @@ impl AnytimeQuality for () {} /// /// Codegen emits one zero-sized impl per anytime node; `Q` is the task's /// [`CuAnytimeTask::Quality`]. An unset knob is `None` and its check in -/// [`AnytimeJob::check`] const-folds away. `max_refines` does not appear here: -/// it is consumed while emitting the execution plan and never read at run time. +/// [`AnytimeJob::check`] const-folds away. #[doc(hidden)] #[diagnostic::on_unimplemented( message = "the anytime policy `{Self}` is pinned to the shared quality scale, but this task's `Quality` is `{Q}`", @@ -222,6 +224,10 @@ pub trait AnytimePolicy { const MAX_AGE: Option; /// Stop after this many quanta without the best quality improving. const MAX_STALL: Option; + /// Hard quanta bound per job, read only by [`CuAnytimeRunner`]: a + /// foreground node encodes the count as the number of refine steps its + /// plan carries and never reads this. + const MAX_REFINES: Option; /// Codegen override: `q >= target` (never satisfied by NaN). Default false. #[inline(always)] @@ -436,6 +442,20 @@ fn stamp( }); } +/// Age anchor of one job: the input's time of validity, falling back to `now`. +/// +/// A range anchors on its earliest data: the entire input window must remain +/// within the age limit. +#[doc(hidden)] +#[inline(always)] +pub fn anchor_from_tov(tov: Tov, now: CuTime) -> CuTime { + match tov { + Tov::Time(time) => time, + Tov::Range(range) => range.start, + Tov::None => now, + } +} + /// Terminal outcome when the age limit passed before `base()`: the job is /// skipped and nothing is published. #[doc(hidden)] @@ -469,12 +489,194 @@ pub fn abort_at_base( } } +/// Runs one whole anytime job per `CuTask::process` call: the age check, +/// `base()`, then refine quanta under `P` until a stop cause fires. +/// +/// An `anytime:` node with `background: true` compiles to this runner wrapped +/// in `CuAsyncTask`. A worker thread has no copperlist steps to interleave +/// quanta with, so the refinement loop lives here instead of in the emitted +/// plan; a foreground node keeps its chunked steps and never uses this type. +/// +/// Every lifecycle call forwards to the task. `CuAsyncTask` forwards +/// `preprocess`/`postprocess` best-effort: they run when the worker is idle +/// and are skipped while a job is in flight. +#[doc(hidden)] +#[derive(Reflect)] +#[reflect(no_field_bounds, from_reflect = false, type_path = false)] +pub struct CuAnytimeRunner +where + T: Send + Sync + 'static, + P: Send + Sync + 'static, +{ + #[reflect(ignore)] + task: T, + #[reflect(ignore)] + _policy: PhantomData

, +} + +impl TypePath for CuAnytimeRunner +where + T: Send + Sync + 'static, + P: Send + Sync + 'static, +{ + fn type_path() -> &'static str { + "cu29_runtime::cutask_anytime::CuAnytimeRunner" + } + + fn short_type_path() -> &'static str { + "CuAnytimeRunner" + } + + fn type_ident() -> Option<&'static str> { + Some("CuAnytimeRunner") + } + + fn crate_name() -> Option<&'static str> { + Some("cu29_runtime") + } + + fn module_path() -> Option<&'static str> { + Some("cutask_anytime") + } +} + +impl Freezable for CuAnytimeRunner +where + T: Freezable + Send + Sync + 'static, + P: Send + Sync + 'static, +{ + fn freeze(&self, encoder: &mut E) -> Result<(), EncodeError> { + self.task.freeze(encoder) + } + + fn thaw(&mut self, decoder: &mut D) -> Result<(), DecodeError> { + self.task.thaw(decoder) + } +} + +impl CuTask for CuAnytimeRunner +where + T: for<'i, 'o> CuAnytimeTask = CuMsg, Output<'o> = CuMsg> + + Send + + Sync + + 'static, + I: CuMsgPayload, + O: CuMsgPayload, + P: AnytimePolicy + Send + Sync + 'static, +{ + type Resources<'r> = T::Resources<'r>; + type Input<'m> = T::Input<'m>; + type Output<'m> = T::Output<'m>; + + fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult + where + Self: Sized, + { + Ok(Self { + task: T::new(config, resources)?, + _policy: PhantomData, + }) + } + + fn start(&mut self, ctx: &CuContext) -> CuResult<()> { + self.task.start(ctx) + } + + fn preprocess(&mut self, ctx: &CuContext) -> CuResult<()> { + self.task.preprocess(ctx) + } + + fn process<'i, 'o>( + &mut self, + ctx: &CuContext, + input: &Self::Input<'i>, + output: &mut Self::Output<'o>, + ) -> CuResult<()> { + let start = ctx.now(); + let anchor = anchor_from_tov(input.tov, start); + if let Some(max_age) = P::MAX_AGE + && start >= anchor + max_age + { + skip_stale(output); + return Ok(()); + } + + // The job clock starts when this worker picks the job up, so queueing + // delay counts against the age limit above but not against the budget. + let mut job = match self.task.base(ctx, input, output)? { + AnytimeStatus::Improved(quality) => AnytimeJob::<_, P>::new(start, anchor, quality), + AnytimeStatus::Converged(quality) => { + AnytimeJob::<_, P>::new(start, anchor, quality).finish( + ctx.now(), + AnytimeStopCause::Converged, + 0, + output, + ); + return Ok(()); + } + AnytimeStatus::Aborted => { + abort_at_base(start, ctx.now(), output); + return Ok(()); + } + }; + + let mut ran = 0u32; + loop { + // One clock read per quantum, shared by check() and finish(); it is + // skipped entirely without a time knob, exactly as the foreground + // refine block does (CuTime subtraction saturates). + let now = if P::TIME_BUDGET.is_some() || P::MAX_AGE.is_some() { + ctx.now() + } else { + CuTime::default() + }; + if let Some(cause) = job.check(now) { + job.finish(now, cause, ran, output); + return Ok(()); + } + // An error surfaces at the next poll of the wrapper, like any other + // backgrounded task's. + let status = self.task.refine(ctx, output)?; + ran += 1; + match status { + AnytimeStatus::Improved(quality) => { + job.record(quality); + if let Some(max_refines) = P::MAX_REFINES + && ran >= max_refines + { + job.finish(now, AnytimeStopCause::MaxRefines, ran, output); + return Ok(()); + } + } + AnytimeStatus::Converged(quality) => { + job.record(quality); + job.finish(now, AnytimeStopCause::Converged, ran, output); + return Ok(()); + } + AnytimeStatus::Aborted => { + job.finish(now, AnytimeStopCause::Aborted, ran, output); + return Ok(()); + } + } + } + } + + fn postprocess(&mut self, ctx: &CuContext) -> CuResult<()> { + self.task.postprocess(ctx) + } + + fn stop(&mut self, ctx: &CuContext) -> CuResult<()> { + self.task.stop(ctx) + } +} + #[cfg(test)] mod tests { use super::*; use crate::cutask::CuMsg; use crate::input_msg; use crate::output_msg; + use cu29_clock::RobotClockMock; fn q(v: f32) -> Quality { quality_from_f32(v) @@ -568,6 +770,7 @@ mod tests { const TIME_BUDGET: Option = Some(CuDuration(1_000_000)); const MAX_AGE: Option = Some(CuDuration(2_000_000)); const MAX_STALL: Option = Some(2); + const MAX_REFINES: Option = Some(8); fn target_met(q: Quality) -> bool { q >= quality_from_f32(0.9) @@ -584,6 +787,7 @@ mod tests { const TIME_BUDGET: Option = None; const MAX_AGE: Option = None; const MAX_STALL: Option = None; + const MAX_REFINES: Option = None; } /// Mirrors a quality-less node (`Quality = ()`, no knobs set). @@ -592,6 +796,7 @@ mod tests { const TIME_BUDGET: Option = None; const MAX_AGE: Option = None; const MAX_STALL: Option = None; + const MAX_REFINES: Option = None; } #[test] @@ -767,4 +972,223 @@ mod tests { assert!(!outcome.published); assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it abort!"); } + + // --- background runner: one whole job per process() call --- + + /// Mirrors codegen for `anytime: (max_refines: 2)`. + struct MaxRefinesPolicy; + impl AnytimePolicy for MaxRefinesPolicy { + const TIME_BUDGET: Option = None; + const MAX_AGE: Option = None; + const MAX_STALL: Option = None; + const MAX_REFINES: Option = Some(2); + } + + /// Mirrors codegen for `anytime: (time_budget_ms: 1.0)`: no quanta bound, + /// so only the budget closes the loop. + struct BudgetOnlyPolicy; + impl AnytimePolicy for BudgetOnlyPolicy { + const TIME_BUDGET: Option = Some(CuDuration(1_000_000)); + const MAX_AGE: Option = None; + const MAX_STALL: Option = None; + const MAX_REFINES: Option = None; + } + + /// Advances the mock clock by one step per quantum, so a time-bounded + /// policy fires deterministically. + #[derive(Reflect)] + #[reflect(no_field_bounds, from_reflect = false)] + struct TickingTask { + #[reflect(ignore)] + clock: RobotClockMock, + step: CuDuration, + elapsed: CuDuration, + } + + impl TickingTask { + fn tick(&mut self) { + self.elapsed += self.step; + self.clock.set_value(self.elapsed.0); + } + } + + impl Freezable for TickingTask {} + + impl CuAnytimeTask for TickingTask { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = RobotClockMock; + type Quality = Quality; + + fn new(_config: Option<&ComponentConfig>, clock: RobotClockMock) -> CuResult { + Ok(Self { + clock, + step: CuDuration::from_millis(1), + elapsed: CuDuration::default(), + }) + } + + fn base<'i, 'o>( + &mut self, + _ctx: &CuContext, + _input: &Self::Input<'i>, + output: &mut Self::Output<'o>, + ) -> CuResult> { + self.tick(); + output.set_payload(0); + Ok(AnytimeStatus::Improved(q(0.5))) + } + + fn refine<'o>( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'o>, + ) -> CuResult> { + self.tick(); + output.set_payload(output.payload().copied().unwrap_or(0) + 1); + Ok(AnytimeStatus::Improved(q(0.5))) + } + } + + /// Gives up before producing anything and says so by clearing the payload. + #[derive(Reflect)] + struct AbortingTask; + + impl Freezable for AbortingTask {} + + impl CuAnytimeTask for AbortingTask { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = (); + type Quality = Quality; + + fn new(_config: Option<&ComponentConfig>, _resources: ()) -> CuResult { + Ok(Self) + } + + fn base<'i, 'o>( + &mut self, + _ctx: &CuContext, + _input: &Self::Input<'i>, + output: &mut Self::Output<'o>, + ) -> CuResult> { + output.clear_payload(); + Ok(AnytimeStatus::Aborted) + } + + fn refine<'o>( + &mut self, + _ctx: &CuContext, + _output: &mut Self::Output<'o>, + ) -> CuResult> { + unreachable!("refine after an abort at base") + } + } + + /// Drives one job and returns the output the runner published. + fn run_job(runner: &mut CuAnytimeRunner, ctx: &CuContext, tov: Tov) -> CuMsg + where + T: for<'i, 'o> CuAnytimeTask = CuMsg, Output<'o> = CuMsg> + + Send + + Sync + + 'static, + P: AnytimePolicy + Send + Sync + 'static, + { + let mut input = CuMsg::new(Some(3u32)); + input.tov = tov; + let mut output = CuMsg::new(None); + runner.process(ctx, &input, &mut output).unwrap(); + output + } + + #[test] + fn runner_stops_at_the_quanta_bound() { + let ctx = CuContext::new_mock_clock().0; + let mut runner: CuAnytimeRunner = + CuAnytimeRunner::new(None, ()).unwrap(); + + let output = run_job(&mut runner, &ctx, Tov::None); + // Two quanta of a job needing three: stopped by the bound, not by the task. + assert_eq!(output.payload(), Some(&2)); + assert_eq!(output.metadata.status_txt.0.as_str(), "any:2it q=0.67 max"); + } + + #[test] + fn runner_stops_when_the_task_converges() { + let ctx = CuContext::new_mock_clock().0; + let mut runner: CuAnytimeRunner = + CuAnytimeRunner::new(None, ()).unwrap(); + + // Three quanta reach the input, quality 1.0 >= the 0.9 target, so the + // check before the fourth quantum stops the job. + let output = run_job(&mut runner, &ctx, Tov::None); + assert_eq!(output.payload(), Some(&3)); + assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=1.00 tgt"); + } + + #[test] + fn runner_stops_when_the_budget_is_exhausted() { + let (ctx, clock) = CuContext::new_mock_clock(); + let mut runner: CuAnytimeRunner = + CuAnytimeRunner::new(None, clock).unwrap(); + + // base() alone burns the 1 ms budget, so no quantum runs. + let output = run_job(&mut runner, &ctx, Tov::None); + assert_eq!(output.payload(), Some(&0)); + assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it q=0.50 bdgt"); + } + + #[test] + fn runner_skips_a_dead_on_arrival_input() { + let (ctx, clock) = CuContext::new_mock_clock(); + clock.set_value(CuDuration::from_millis(5).0); + let mut runner: CuAnytimeRunner = + CuAnytimeRunner::new(None, ()).unwrap(); + + // The input is 5 ms old against a 2 ms horizon: base() never runs. + let output = run_job(&mut runner, &ctx, Tov::Time(CuTime::default())); + assert_eq!(output.payload(), None); + assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it stale!"); + } + + #[test] + fn runner_reports_an_abort_at_base() { + let ctx = CuContext::new_mock_clock().0; + let mut runner: CuAnytimeRunner = + CuAnytimeRunner::new(None, ()).unwrap(); + + let output = run_job(&mut runner, &ctx, Tov::None); + assert_eq!(output.payload(), None); + assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it abort!"); + } + + #[test] + fn runner_drops_a_result_below_the_quality_floor() { + let (ctx, clock) = CuContext::new_mock_clock(); + // FloorPolicy budgets 1 ms and floors at 0.8: base() alone burns the + // budget reporting 0.5, below the floor, so nothing is published. + let mut runner: CuAnytimeRunner = + CuAnytimeRunner::new(None, clock).unwrap(); + + let output = run_job(&mut runner, &ctx, Tov::None); + assert_eq!(output.payload(), None, "below the floor: nothing published"); + assert_eq!( + output.metadata.status_txt.0.as_str(), + "any:0it q=0.50 bdgt!" + ); + } + + /// Mirrors codegen for `anytime: (time_budget_ms: 1.0, quality_floor: 0.8)`. + struct FloorPolicy; + impl AnytimePolicy for FloorPolicy { + const TIME_BUDGET: Option = Some(CuDuration(1_000_000)); + const MAX_AGE: Option = None; + const MAX_STALL: Option = None; + const MAX_REFINES: Option = None; + + fn below_floor(q: Quality) -> bool { + q.partial_cmp(&quality_from_f32(0.8)) + .is_none_or(core::cmp::Ordering::is_lt) + } + } } diff --git a/examples/cu_anytime_task/Cargo.toml b/examples/cu_anytime_task/Cargo.toml new file mode 100644 index 00000000000..9be27c1c92e --- /dev/null +++ b/examples/cu_anytime_task/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "cu-anytime-task" +description = "This is an example for the Copper project to show how to set up anytime tasks (base computation plus bounded refinement quanta), in both foreground and background placements." +version.workspace = true +authors.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +keywords.workspace = true +categories.workspace = true +homepage.workspace = true +repository.workspace = true + +publish = false +[dependencies] +cu29 = { workspace = true } +serde = { workspace = true } + +[package.metadata.cargo-shear] +ignored = ["serde"] + +[build-dependencies] +cu29-build = { workspace = true } diff --git a/examples/cu_anytime_task/build.rs b/examples/cu_anytime_task/build.rs new file mode 100644 index 00000000000..7cbac12abe5 --- /dev/null +++ b/examples/cu_anytime_task/build.rs @@ -0,0 +1,3 @@ +fn main() { + cu29_build::setup(); +} diff --git a/examples/cu_anytime_task/copperconfig.ron b/examples/cu_anytime_task/copperconfig.ron new file mode 100644 index 00000000000..c7f76f7f61f --- /dev/null +++ b/examples/cu_anytime_task/copperconfig.ron @@ -0,0 +1,66 @@ +( + tasks: [ + ( + id: "camera", + type: "tasks::TargetSrc", + ), + ( + id: "planner", + type: "tasks::IncrementalPlanner", + anytime: ( + max_age_ms: 500.0, + quality_target: 0.9, + max_refines: 8, + ), + ), + ( + id: "smoother", + type: "tasks::CountingSmoother", + anytime: (max_refines: 2), + ), + ( + id: "recorder", + type: "tasks::RecordingSink", + ), + ( + id: "tracker", + type: "tasks::IncrementalPlanner", + background: true, + anytime: ( + time_budget_ms: 50.0, + quality_target: 0.9, + ), + ), + ( + id: "tracker_recorder", + type: "tasks::BackgroundSink", + ), + ], + cnx: [ + ( + src: "camera", + dst: "planner", + msg: "u32", + ), + ( + src: "planner", + dst: "smoother", + msg: "u32", + ), + ( + src: "smoother", + dst: "recorder", + msg: "u32", + ), + ( + src: "camera", + dst: "tracker", + msg: "u32", + ), + ( + src: "tracker", + dst: "tracker_recorder", + msg: "u32", + ), + ], +) \ No newline at end of file diff --git a/examples/cu_anytime_task/src/main.rs b/examples/cu_anytime_task/src/main.rs new file mode 100644 index 00000000000..35cc15acf0c --- /dev/null +++ b/examples/cu_anytime_task/src/main.rs @@ -0,0 +1,256 @@ +use cu29::prelude::*; +use std::fs; +use std::path::Path; + +pub mod tasks { + use cu29::cutask_anytime::{AnytimeStatus, CuAnytimeTask, Quality, quality_from_f32}; + use cu29::prelude::*; + use std::sync::Mutex; + + /// What the sink observed, for the assertions in `main`. + pub static RECORDED: Mutex> = Mutex::new(Vec::new()); + + #[derive(Reflect)] + pub struct TargetSrc; + + impl Freezable for TargetSrc {} + + impl CuSrcTask for TargetSrc { + type Resources<'r> = (); + type Output<'m> = output_msg!(u32); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult + where + Self: Sized, + { + Ok(Self) + } + + fn process(&mut self, ctx: &CuContext, new_msg: &mut Self::Output<'_>) -> CuResult<()> { + new_msg.set_payload(5); + // A fresh Tov: the planner's max_age anchors on it. + new_msg.tov = Tov::Time(ctx.clock.now()); + Ok(()) + } + } + + /// Anytime node with a measurable quality: `base()` publishes 0 and each + /// `refine()` commits one more increment toward the input target, so the + /// published quality climbs to 1.0 and the configured quality_target + /// stops refinement early. + #[derive(Reflect)] + pub struct IncrementalPlanner { + target: u32, + acc: u32, + } + + impl Freezable for IncrementalPlanner {} + + impl CuAnytimeTask for IncrementalPlanner { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = (); + type Quality = Quality; + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult + where + Self: Sized, + { + Ok(Self { target: 0, acc: 0 }) + } + + fn base( + &mut self, + _ctx: &CuContext, + input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult> { + self.target = input.payload().copied().ok_or("planner: no input")?; + self.acc = 0; + output.set_payload(self.acc); + Ok(AnytimeStatus::Improved(quality_from_f32(0.0))) + } + + fn refine( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'_>, + ) -> CuResult> { + if self.acc >= self.target { + return Ok(AnytimeStatus::Converged(quality_from_f32(1.0))); + } + self.acc += 1; + output.set_payload(self.acc); + Ok(AnytimeStatus::Improved(quality_from_f32( + self.acc as f32 / self.target as f32, + ))) + } + } + + /// Quality-less anytime node: every quantum bumps the output, so it runs + /// its whole emitted refine budget and stops by position (MaxRefines). + #[derive(Reflect)] + pub struct CountingSmoother; + + impl Freezable for CountingSmoother {} + + impl CuAnytimeTask for CountingSmoother { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = (); + type Quality = (); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult + where + Self: Sized, + { + Ok(Self) + } + + fn base( + &mut self, + _ctx: &CuContext, + input: &Self::Input<'_>, + output: &mut Self::Output<'_>, + ) -> CuResult> { + output.set_payload(input.payload().copied().ok_or("smoother: no input")?); + Ok(AnytimeStatus::Improved(())) + } + + fn refine( + &mut self, + _ctx: &CuContext, + output: &mut Self::Output<'_>, + ) -> CuResult> { + let bumped = output.payload().copied().unwrap_or(0) + 1; + output.set_payload(bumped); + Ok(AnytimeStatus::Improved(())) + } + } + + /// What the background chain published, when a job landed. + pub static BACKGROUND_RECORDED: Mutex> = Mutex::new(Vec::new()); + + #[derive(Reflect)] + pub struct RecordingSink; + + impl Freezable for RecordingSink {} + + impl CuSinkTask for RecordingSink { + type Resources<'r> = (); + type Input<'m> = input_msg!(u32); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult + where + Self: Sized, + { + Ok(Self) + } + + fn process(&mut self, _ctx: &CuContext, input: &Self::Input<'_>) -> CuResult<()> { + let payload = input.payload().copied().ok_or("recorder: no input")?; + let status = input.metadata.status_txt.0.to_string(); + RECORDED + .lock() + .expect("recorder poisoned") + .push((payload, status)); + Ok(()) + } + } + /// Sink of the `tracker` node, which runs the same anytime task as + /// `planner` but with `background: true`: most copperlists carry no payload + /// because the job is still running on its worker thread. + #[derive(Reflect)] + pub struct BackgroundSink; + + impl Freezable for BackgroundSink {} + + impl CuSinkTask for BackgroundSink { + type Resources<'r> = (); + type Input<'m> = input_msg!(u32); + + fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult + where + Self: Sized, + { + Ok(Self) + } + + fn process(&mut self, _ctx: &CuContext, input: &Self::Input<'_>) -> CuResult<()> { + if let Some(payload) = input.payload().copied() { + let status = input.metadata.status_txt.0.to_string(); + BACKGROUND_RECORDED + .lock() + .expect("recorder poisoned") + .push((payload, status)); + } + Ok(()) + } + } +} + +#[copper_runtime(config = "copperconfig.ron")] +struct App {} + +const SLAB_SIZE: Option = Some(16 * 1024 * 1024); + +fn main() { + let logger_path = "logs/anytime.copper"; + if let Some(parent) = Path::new(logger_path).parent() + && !parent.exists() + { + fs::create_dir_all(parent).expect("Failed to create logs directory"); + } + let mut application = App::builder() + .with_log_path(logger_path, SLAB_SIZE) + .expect("Failed to setup logger.") + .build() + .expect("Failed to create application."); + application + .start_all_tasks() + .expect("Failed to start application."); + // Three copperlists for the foreground chain; keep polling until the + // backgrounded node's first job comes back from its worker thread. + for iteration in 0..100 { + application + .run_one_iteration() + .expect("Failed to run application."); + let landed = !tasks::BACKGROUND_RECORDED + .lock() + .expect("recorder poisoned") + .is_empty(); + if iteration >= 2 && landed { + break; + } + } + application + .stop_all_tasks() + .expect("Failed to stop application."); + + let background = tasks::BACKGROUND_RECORDED + .lock() + .expect("recorder poisoned"); + // The worker ran the whole job: five quanta reach the target of 5 and the + // check before the sixth sees quality 1.0 >= quality_target. + assert_eq!( + background + .first() + .map(|(payload, status)| (*payload, status.as_str())), + Some((5, "any:5it q=1.00 tgt")), + "background anytime job did not land" + ); + println!("background anytime OK: {:?}", background.as_slice()); + drop(background); + + let recorded = tasks::RECORDED.lock().expect("recorder poisoned"); + assert!(recorded.len() >= 3, "one recorded value per copperlist"); + for (payload, status) in recorded.iter() { + // planner: base publishes 0, five quanta reach the target of 5, and + // quality 1.0 >= quality_target stops it before its 8-quantum budget. + // smoother: +1 per quantum for its whole 2-quantum budget. + assert_eq!(*payload, 7, "planner result (5) + smoother budget (2)"); + // The smoother's status stamp: 2 quanta, stopped by position. + assert_eq!(status, "any:2it max"); + } + println!("anytime example OK: {:?}", recorded.as_slice()); +}