From 5455283e62eeee7010809cae0ac68cf896038eba Mon Sep 17 00:00:00 2001 From: Scofield626 Date: Thu, 30 Jul 2026 15:22:04 +0000 Subject: [PATCH 1/8] feat: anytime background runner (one whole job per process call) --- core/cu29_runtime/src/cuasynctask.rs | 83 +++++ core/cu29_runtime/src/cutask_anytime.rs | 429 +++++++++++++++++++++++- 2 files changed, 508 insertions(+), 4 deletions(-) diff --git a/core/cu29_runtime/src/cuasynctask.rs b/core/cu29_runtime/src/cuasynctask.rs index 56fa11b39ff..a33298061a6 100644 --- a/core/cu29_runtime/src/cuasynctask.rs +++ b/core/cu29_runtime/src/cuasynctask.rs @@ -546,8 +546,12 @@ 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; @@ -1061,6 +1065,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..3fbbad633c5 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,21 @@ fn stamp( }); } +/// Age anchor of one job: the input's time of validity, falling back to `now`. +/// +/// A range anchors on its newest data (`end`); anchoring on `start` would +/// declare any input whose span exceeds the age limit (a full lidar sweep, say) +/// dead on arrival forever. +#[doc(hidden)] +#[inline(always)] +pub fn anchor_from_tov(tov: Tov, now: CuTime) -> CuTime { + match tov { + Tov::Time(time) => time, + Tov::Range(range) => range.end, + Tov::None => now, + } +} + /// Terminal outcome when the age limit passed before `base()`: the job is /// skipped and nothing is published. #[doc(hidden)] @@ -469,12 +490,190 @@ 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. +#[doc(hidden)] +#[derive(Reflect)] +#[reflect(no_field_bounds, from_reflect = false, type_path = false)] +pub struct CuAnytimeRunner +where + T: Reflect + Send + Sync + 'static, + P: Send + Sync + 'static, +{ + #[reflect(ignore)] + task: T, + #[reflect(ignore)] + _policy: PhantomData

, +} + +impl TypePath for CuAnytimeRunner +where + T: Reflect + 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: Reflect + 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 +767,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 +784,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 +793,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 +969,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(); + // FullPolicy floors at 0.3 and budgets 1 ms; the ticking task reports + // 0.5 but the DOA-free job stops on the budget with that best quality. + 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) + } + } } From d7d918099e9bce99b69fcd23b7e3c51a4eee975e Mon Sep 17 00:00:00 2001 From: Scofield626 Date: Thu, 30 Jul 2026 15:22:04 +0000 Subject: [PATCH 2/8] feat: codegen for anytime nodes with background: true --- core/cu29_derive/src/lib.rs | 100 ++++++--- .../copper_runtime/anytime_background_task.rs | 200 ++++++++++++++++++ .../anytime_background_task_sim.rs | 199 +++++++++++++++++ .../config/anytime_background_task_valid.ron | 71 +++++++ core/cu29_runtime/src/cutask_anytime.rs | 7 +- 5 files changed, 547 insertions(+), 30 deletions(-) create mode 100644 core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task.rs create mode 100644 core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task_sim.rs create mode 100644 core/cu29_derive/tests/config/anytime_background_task_valid.ron diff --git a/core/cu29_derive/src/lib.rs b/core/cu29_derive/src/lib.rs index 21594a23af6..d58eea3a386 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,13 @@ 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]; + // What the wrapper wraps: the anytime runner for an anytime + // node, the declared task otherwise. + let inner_task_type = &background_inner_type( + &task_specs.sim_task_types[index], + task_specs.ids[index].as_str(), + task_specs.anytime_configs[index].is_some(), + ); match task_specs.cutypes[index] { CuTaskType::Source => { if background { @@ -2415,7 +2412,11 @@ 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 = &background_inner_type( + &task_specs.sim_task_types[index], + task_specs.ids[index].as_str(), + task_specs.anytime_configs[index].is_some(), + ); match task_specs.cutypes[index] { CuTaskType::Source => { if *background { @@ -6113,14 +6114,33 @@ 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() { + if task_specs.background_flags[index] { + task_trait_for_kind(task_specs.cutypes[index]) + } else if task_specs.anytime_configs[index].is_some() { quote! { cu29::cutask_anytime::CuAnytimeTask } } else { task_trait_for_kind(task_specs.cutypes[index]) } } +/// The task a backgrounded node hands to `CuAsyncTask`: an anytime node goes +/// through [`CuAnytimeRunner`], which turns one whole job — age check, `base()`, +/// refine quanta under the policy — into a single `CuTask::process` call. +/// +/// `policy` names the node's `AnytimePolicy` ZST, emitted in the mission module +/// alongside every type built here. +fn background_inner_type(task_type: &Type, task_id: &str, is_anytime: bool) -> Type { + if !is_anytime { + return task_type.clone(); + } + let policy_ident = anytime_policy_ident(task_id); + parse_quote!(cu29::cutask_anytime::CuAnytimeRunner<#task_type, #policy_ident>) +} + fn task_output_payload_type( graph: &CuGraph, node: &Node, @@ -6186,7 +6206,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(); @@ -6258,8 +6278,16 @@ impl CuTaskSpecSet { .zip(cutypes.iter()) .zip(background_flags.iter()) .zip(output_types.iter()) - .map(|((((name_type, name), cutype), &background), output_type)| { + .enumerate() + .map(|(index, ((((name_type, name), cutype), &background), output_type))| { if background { + // A foreground anytime node keeps its raw type in the tuple; + // a backgrounded one is driven through the anytime runner. + let name_type = &background_inner_type( + name_type, + ids[index].as_str(), + anytime_configs[index].is_some(), + ); if let Some(output_type) = output_type { match cutype { CuTaskType::Source => { @@ -6290,8 +6318,14 @@ impl CuTaskSpecSet { .zip(cutypes.iter()) .zip(background_flags.iter()) .zip(output_types.iter()) - .map(|((((name_type, name), cutype), &background), output_type)| { + .enumerate() + .map(|(index, ((((name_type, name), cutype), &background), output_type))| { if background { + let name_type = &background_inner_type( + name_type, + ids[index].as_str(), + anytime_configs[index].is_some(), + ); if let Some(output_type) = output_type { match cutype { CuTaskType::Source => { @@ -7814,10 +7848,16 @@ fn build_task_resource_mappings( continue; } + // 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 = if task_specs.background_flags[idx] { - &task_specs.sim_task_types[idx] + background_inner_type( + &task_specs.sim_task_types[idx], + task_specs.ids[idx].as_str(), + task_specs.anytime_configs[idx].is_some(), + ) } else { - &task_specs.task_types[idx] + task_specs.task_types[idx].clone() }; let binding_trait = task_trait_for_specs(task_specs, idx); @@ -8561,10 +8601,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 +8686,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 +10066,12 @@ 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 = background_inner_type( + declared_task_type, + task_id, + task_specs.anytime_configs[index].is_some(), + ); + 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..6f492e53d76 --- /dev/null +++ b/core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task.rs @@ -0,0 +1,200 @@ +//! 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. +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..db213fc6e65 --- /dev/null +++ b/core/cu29_derive/tests/compile_pass/copper_runtime/anytime_background_task_sim.rs @@ -0,0 +1,199 @@ +//! Sim-mode expansion of a backgrounded anytime node: it is indistinguishable +//! from any other backgrounded task — one sim callback, one poll step. +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..1b7515922cb --- /dev/null +++ b/core/cu29_derive/tests/config/anytime_background_task_valid.ron @@ -0,0 +1,71 @@ +( + runtime: ( + thread_pools: [ + (id: "planner", threads: 1), + ], + ), + resources: [ + ( + id: "board", + provider: "TestBundle", + ), + ], + tasks: [ + (id: "src", type: "AnytimeSrc"), + // Background placement: a time bound alone is enough, the quanta count + // is not needed to build the plan. + ( + 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, + ), + ), + // Background on a named pool, with resources and a quanta bound. + ( + id: "counter", + type: "AnytimeCounter", + background: (pool: "planner"), + anytime: (max_refines: 2), + resources: {"scratch": "board.scratch"}, + ), + ( + id: "sink", + type: "AnytimeSink", + ), + // Foreground anytime in the same application: both placements coexist. + ( + 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", + ), + ], +) diff --git a/core/cu29_runtime/src/cutask_anytime.rs b/core/cu29_runtime/src/cutask_anytime.rs index 3fbbad633c5..692bf58625c 100644 --- a/core/cu29_runtime/src/cutask_anytime.rs +++ b/core/cu29_runtime/src/cutask_anytime.rs @@ -444,15 +444,14 @@ fn stamp( /// Age anchor of one job: the input's time of validity, falling back to `now`. /// -/// A range anchors on its newest data (`end`); anchoring on `start` would -/// declare any input whose span exceeds the age limit (a full lidar sweep, say) -/// dead on arrival forever. +/// 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.end, + Tov::Range(range) => range.start, Tov::None => now, } } From f788b86b400cb6d0d6457bdfe3f436478434ca5b Mon Sep 17 00:00:00 2001 From: Scofield626 Date: Thu, 30 Jul 2026 15:22:09 +0000 Subject: [PATCH 3/8] example: background anytime node in cu_anytime_task --- Cargo.toml | 1 + examples/cu_anytime_task/Cargo.toml | 23 ++ examples/cu_anytime_task/build.rs | 3 + examples/cu_anytime_task/copperconfig.ron | 69 ++++++ examples/cu_anytime_task/src/main.rs | 255 ++++++++++++++++++++++ 5 files changed, 351 insertions(+) create mode 100644 examples/cu_anytime_task/Cargo.toml create mode 100644 examples/cu_anytime_task/build.rs create mode 100644 examples/cu_anytime_task/copperconfig.ron create mode 100644 examples/cu_anytime_task/src/main.rs 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/examples/cu_anytime_task/Cargo.toml b/examples/cu_anytime_task/Cargo.toml new file mode 100644 index 00000000000..1f229acf566 --- /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 a foreground anytime task (base computation plus bounded refinement quanta)." +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..6a42018dee1 --- /dev/null +++ b/examples/cu_anytime_task/copperconfig.ron @@ -0,0 +1,69 @@ +( + 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", + ), + // Background placement: the same task type, run to completion on a + // worker thread. A time bound alone is enough here - the copperlist + // plan does not need the quanta count. + ( + 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..804d3cc9dee --- /dev/null +++ b/examples/cu_anytime_task/src/main.rs @@ -0,0 +1,255 @@ +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 a backgrounded node: 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()); +} From 5bfcd616bd92167529c09573a04192b1a644ab65 Mon Sep 17 00:00:00 2001 From: Scofield626 Date: Thu, 30 Jul 2026 15:28:12 +0000 Subject: [PATCH 4/8] fix: review findings on the background anytime runner --- core/cu29_derive/src/lib.rs | 10 +++++----- core/cu29_runtime/src/cutask_anytime.rs | 4 ++++ 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/core/cu29_derive/src/lib.rs b/core/cu29_derive/src/lib.rs index d58eea3a386..0a681d14020 100644 --- a/core/cu29_derive/src/lib.rs +++ b/core/cu29_derive/src/lib.rs @@ -2307,7 +2307,7 @@ pub fn copper_runtime(args: TokenStream, input: TokenStream) -> TokenStream { let inner_task_type = &background_inner_type( &task_specs.sim_task_types[index], task_specs.ids[index].as_str(), - task_specs.anytime_configs[index].is_some(), + background && task_specs.anytime_configs[index].is_some(), ); match task_specs.cutypes[index] { CuTaskType::Source => { @@ -2415,7 +2415,7 @@ pub fn copper_runtime(args: TokenStream, input: TokenStream) -> TokenStream { let inner_task_type = &background_inner_type( &task_specs.sim_task_types[index], task_specs.ids[index].as_str(), - task_specs.anytime_configs[index].is_some(), + *background && task_specs.anytime_configs[index].is_some(), ); match task_specs.cutypes[index] { CuTaskType::Source => { @@ -6118,9 +6118,9 @@ fn task_trait_for_kind(task_kind: CuTaskType) -> proc_macro2::TokenStream { /// 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.background_flags[index] { - task_trait_for_kind(task_specs.cutypes[index]) - } else 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]) diff --git a/core/cu29_runtime/src/cutask_anytime.rs b/core/cu29_runtime/src/cutask_anytime.rs index 692bf58625c..97b4311c9f5 100644 --- a/core/cu29_runtime/src/cutask_anytime.rs +++ b/core/cu29_runtime/src/cutask_anytime.rs @@ -496,6 +496,10 @@ pub fn abort_at_base( /// 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, but `CuAsyncTask` currently calls +/// neither `preprocess` nor `postprocess` on what it wraps, so a backgrounded +/// anytime task does not see them either. #[doc(hidden)] #[derive(Reflect)] #[reflect(no_field_bounds, from_reflect = false, type_path = false)] From 041757320e3681c6dae169bc0b3b225758fd6c11 Mon Sep 17 00:00:00 2001 From: Scofield626 Date: Thu, 30 Jul 2026 15:30:03 +0000 Subject: [PATCH 5/8] chore: fmt (ron) --- .../copper_runtime/anytime_background_task.rs | 5 +++++ .../copper_runtime/anytime_background_task_sim.rs | 2 ++ .../tests/config/anytime_background_task_valid.ron | 10 ++-------- examples/cu_anytime_task/copperconfig.ron | 3 --- examples/cu_anytime_task/src/main.rs | 5 +++-- 5 files changed, 12 insertions(+), 13 deletions(-) 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 index 6f492e53d76..4f86c77f762 100644 --- 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 @@ -1,6 +1,11 @@ //! 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}; 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 index db213fc6e65..b0cb4d34030 100644 --- 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 @@ -1,5 +1,7 @@ //! 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}; diff --git a/core/cu29_derive/tests/config/anytime_background_task_valid.ron b/core/cu29_derive/tests/config/anytime_background_task_valid.ron index 1b7515922cb..429544f16c4 100644 --- a/core/cu29_derive/tests/config/anytime_background_task_valid.ron +++ b/core/cu29_derive/tests/config/anytime_background_task_valid.ron @@ -1,8 +1,6 @@ ( runtime: ( - thread_pools: [ - (id: "planner", threads: 1), - ], + thread_pools: [(id: "planner", threads: 1)], ), resources: [ ( @@ -12,8 +10,6 @@ ], tasks: [ (id: "src", type: "AnytimeSrc"), - // Background placement: a time bound alone is enough, the quanta count - // is not needed to build the plan. ( id: "refiner", type: "AnytimeRefiner", @@ -26,7 +22,6 @@ max_stall: 3, ), ), - // Background on a named pool, with resources and a quanta bound. ( id: "counter", type: "AnytimeCounter", @@ -38,7 +33,6 @@ id: "sink", type: "AnytimeSink", ), - // Foreground anytime in the same application: both placements coexist. ( id: "tail", type: "AnytimeTail", @@ -68,4 +62,4 @@ msg: "u32", ), ], -) +) \ No newline at end of file diff --git a/examples/cu_anytime_task/copperconfig.ron b/examples/cu_anytime_task/copperconfig.ron index 6a42018dee1..c7f76f7f61f 100644 --- a/examples/cu_anytime_task/copperconfig.ron +++ b/examples/cu_anytime_task/copperconfig.ron @@ -22,9 +22,6 @@ id: "recorder", type: "tasks::RecordingSink", ), - // Background placement: the same task type, run to completion on a - // worker thread. A time bound alone is enough here - the copperlist - // plan does not need the quanta count. ( id: "tracker", type: "tasks::IncrementalPlanner", diff --git a/examples/cu_anytime_task/src/main.rs b/examples/cu_anytime_task/src/main.rs index 804d3cc9dee..35cc15acf0c 100644 --- a/examples/cu_anytime_task/src/main.rs +++ b/examples/cu_anytime_task/src/main.rs @@ -157,8 +157,9 @@ pub mod tasks { Ok(()) } } - /// Sink of a backgrounded node: most copperlists carry no payload because - /// the job is still running on its worker thread. + /// 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; From 57eee2fdd102881eba178b3c7466d003bd9922a9 Mon Sep 17 00:00:00 2001 From: Scofield626 Date: Mon, 3 Aug 2026 10:18:20 +0000 Subject: [PATCH 6/8] fix: drop unneeded T: Reflect bound on CuAnytimeRunner Also fix a stale test comment and the example crate description. --- core/cu29_runtime/src/cutask_anytime.rs | 10 +++++----- examples/cu_anytime_task/Cargo.toml | 2 +- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/core/cu29_runtime/src/cutask_anytime.rs b/core/cu29_runtime/src/cutask_anytime.rs index 97b4311c9f5..a7d48e09d2e 100644 --- a/core/cu29_runtime/src/cutask_anytime.rs +++ b/core/cu29_runtime/src/cutask_anytime.rs @@ -505,7 +505,7 @@ pub fn abort_at_base( #[reflect(no_field_bounds, from_reflect = false, type_path = false)] pub struct CuAnytimeRunner where - T: Reflect + Send + Sync + 'static, + T: Send + Sync + 'static, P: Send + Sync + 'static, { #[reflect(ignore)] @@ -516,7 +516,7 @@ where impl TypePath for CuAnytimeRunner where - T: Reflect + Send + Sync + 'static, + T: Send + Sync + 'static, P: Send + Sync + 'static, { fn type_path() -> &'static str { @@ -542,7 +542,7 @@ where impl Freezable for CuAnytimeRunner where - T: Reflect + Freezable + Send + Sync + 'static, + T: Freezable + Send + Sync + 'static, P: Send + Sync + 'static, { fn freeze(&self, encoder: &mut E) -> Result<(), EncodeError> { @@ -1165,8 +1165,8 @@ mod tests { #[test] fn runner_drops_a_result_below_the_quality_floor() { let (ctx, clock) = CuContext::new_mock_clock(); - // FullPolicy floors at 0.3 and budgets 1 ms; the ticking task reports - // 0.5 but the DOA-free job stops on the budget with that best quality. + // 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(); diff --git a/examples/cu_anytime_task/Cargo.toml b/examples/cu_anytime_task/Cargo.toml index 1f229acf566..9be27c1c92e 100644 --- a/examples/cu_anytime_task/Cargo.toml +++ b/examples/cu_anytime_task/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cu-anytime-task" -description = "This is an example for the Copper project to show how to set up a foreground anytime task (base computation plus bounded refinement quanta)." +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 From 7c36efe0a8d6c900c49250e4126e007b4c4cc46d Mon Sep 17 00:00:00 2001 From: Scofield626 Date: Tue, 4 Aug 2026 13:13:17 +0000 Subject: [PATCH 7/8] cu29-derive: resolve the inner task type once in CuTaskSpecSet One precomputed vec (inner_task_types: the type inside the optional async wrapper) replaces the per-site background_inner_type calls and the index threading through the type-building closures. --- core/cu29_derive/src/lib.rs | 94 ++++++++++++++----------------------- 1 file changed, 34 insertions(+), 60 deletions(-) diff --git a/core/cu29_derive/src/lib.rs b/core/cu29_derive/src/lib.rs index 0a681d14020..07dc64a295a 100644 --- a/core/cu29_derive/src/lib.rs +++ b/core/cu29_derive/src/lib.rs @@ -2302,13 +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]); - // What the wrapper wraps: the anytime runner for an anytime - // node, the declared task otherwise. - let inner_task_type = &background_inner_type( - &task_specs.sim_task_types[index], - task_specs.ids[index].as_str(), - background && task_specs.anytime_configs[index].is_some(), - ); + let inner_task_type = &task_specs.inner_task_types[index]; match task_specs.cutypes[index] { CuTaskType::Source => { if background { @@ -2412,11 +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 = &background_inner_type( - &task_specs.sim_task_types[index], - task_specs.ids[index].as_str(), - *background && task_specs.anytime_configs[index].is_some(), - ); + let inner_task_type = &task_specs.inner_task_types[index]; match task_specs.cutypes[index] { CuTaskType::Source => { if *background { @@ -6127,20 +6117,6 @@ fn task_trait_for_specs(task_specs: &CuTaskSpecSet, index: usize) -> proc_macro2 } } -/// The task a backgrounded node hands to `CuAsyncTask`: an anytime node goes -/// through [`CuAnytimeRunner`], which turns one whole job — age check, `base()`, -/// refine quanta under the policy — into a single `CuTask::process` call. -/// -/// `policy` names the node's `AnytimePolicy` ZST, emitted in the mission module -/// alongside every type built here. -fn background_inner_type(task_type: &Type, task_id: &str, is_anytime: bool) -> Type { - if !is_anytime { - return task_type.clone(); - } - let policy_ident = anytime_policy_ident(task_id); - parse_quote!(cu29::cutask_anytime::CuAnytimeRunner<#task_type, #policy_ident>) -} - fn task_output_payload_type( graph: &CuGraph, node: &Node, @@ -6186,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, @@ -6272,29 +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()) - .enumerate() - .map(|(index, ((((name_type, name), cutype), &background), output_type))| { + .zip(inner_task_types.iter()) + .map(|(((((name_type, name), cutype), &background), output_type), inner_type)| { if background { - // A foreground anytime node keeps its raw type in the tuple; - // a backgrounded one is driven through the anytime runner. - let name_type = &background_inner_type( - name_type, - ids[index].as_str(), - anytime_configs[index].is_some(), - ); 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."); @@ -6318,21 +6308,16 @@ impl CuTaskSpecSet { .zip(cutypes.iter()) .zip(background_flags.iter()) .zip(output_types.iter()) - .enumerate() - .map(|(index, ((((name_type, name), cutype), &background), output_type))| { + .zip(inner_task_types.iter()) + .map(|(((((name_type, name), cutype), &background), output_type), inner_type)| { if background { - let name_type = &background_inner_type( - name_type, - ids[index].as_str(), - anytime_configs[index].is_some(), - ); 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."); @@ -6368,6 +6353,7 @@ impl CuTaskSpecSet { background_flags, background_pools, anytime_configs, + inner_task_types, logging_enabled, type_names, task_types, @@ -7850,15 +7836,7 @@ fn build_task_resource_mappings( // 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 = if task_specs.background_flags[idx] { - background_inner_type( - &task_specs.sim_task_types[idx], - task_specs.ids[idx].as_str(), - task_specs.anytime_configs[idx].is_some(), - ) - } else { - task_specs.task_types[idx].clone() - }; + let binding_task_type = &task_specs.inner_task_types[idx]; let binding_trait = task_trait_for_specs(task_specs, idx); @@ -10066,11 +10044,7 @@ fn runtime_task_type_for_index( CuTaskType::Regular => { if background { if let Some(out_ty) = output_type { - let inner = background_inner_type( - declared_task_type, - task_id, - task_specs.anytime_configs[index].is_some(), - ); + 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"); From 43cfbe16f6120fea5040313cb8c486df1a08d603 Mon Sep 17 00:00:00 2001 From: Scofield626 Date: Tue, 4 Aug 2026 20:17:59 +0000 Subject: [PATCH 8/8] fix: drive the per-job hooks from the background runner The runner brackets every job on the worker with the task's preprocess/postprocess, in the order CuAnytimeTask documents; its own CuTask hook slots stay no-ops so a forwarding wrapper could never double-call them. --- core/cu29_runtime/src/cutask_anytime.rs | 272 +++++++++++++++++------- 1 file changed, 192 insertions(+), 80 deletions(-) diff --git a/core/cu29_runtime/src/cutask_anytime.rs b/core/cu29_runtime/src/cutask_anytime.rs index a7d48e09d2e..2877303c399 100644 --- a/core/cu29_runtime/src/cutask_anytime.rs +++ b/core/cu29_runtime/src/cutask_anytime.rs @@ -497,9 +497,10 @@ pub fn abort_at_base( /// 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, but `CuAsyncTask` currently calls -/// neither `preprocess` nor `postprocess` on what it wraps, so a backgrounded -/// anytime task does not see them either. +/// The runner drives the per-job hooks documented on [`CuAnytimeTask`] itself: +/// `preprocess` right before the job and `postprocess` once it settles, both on +/// the worker. Its own `CuTask` hook slots stay no-ops so a wrapper that one +/// day forwards per-cycle hooks cannot double-call the task. #[doc(hidden)] #[derive(Reflect)] #[reflect(no_field_bounds, from_reflect = false, type_path = false)] @@ -582,92 +583,108 @@ where 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); + // The per-job bracket CuAnytimeTask documents; the worker has no + // copperlist bracket to hang the hooks on, so the runner drives them. + // Both run outside the job clock, as in the foreground placement. + self.task.preprocess(ctx)?; + let job = run_job::(&mut self.task, ctx, input, output); + let post = self.task.postprocess(ctx); + job.and(post) + } + + fn stop(&mut self, ctx: &CuContext) -> CuResult<()> { + self.task.stop(ctx) + } +} + +/// One whole job: the age check, `base()`, then refine quanta under `P` until +/// a stop cause fires. Split out of `process` so the per-job hooks can bracket +/// every exit path. +fn run_job( + task: &mut T, + ctx: &CuContext, + input: &CuMsg, + output: &mut CuMsg, +) -> CuResult<()> +where + T: for<'i, 'o> CuAnytimeTask = CuMsg, Output<'o> = CuMsg>, + I: CuMsgPayload, + O: CuMsgPayload, + P: AnytimePolicy, +{ + 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 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(()); } - - // 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::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 = 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) => { - AnytimeJob::<_, P>::new(start, anchor, quality).finish( - ctx.now(), - AnytimeStopCause::Converged, - 0, - output, - ); + job.record(quality); + job.finish(now, AnytimeStopCause::Converged, ran, 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); + job.finish(now, AnytimeStopCause::Aborted, 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)] @@ -676,6 +693,8 @@ mod tests { use crate::cutask::CuMsg; use crate::input_msg; use crate::output_msg; + use alloc::sync::Arc; + use core::sync::atomic::{AtomicU32, Ordering}; use cu29_clock::RobotClockMock; fn q(v: f32) -> Quality { @@ -1085,8 +1104,101 @@ mod tests { } } + /// Appends a digit per lifecycle call so per-job hook order reads back as + /// one number: 1 preprocess, 2 base, 3 refine, 4 postprocess. + #[derive(Reflect)] + #[reflect(no_field_bounds, from_reflect = false)] + struct HookOrderTask { + #[reflect(ignore)] + seq: Arc, + } + + impl HookOrderTask { + fn tag(&self, digit: u32) { + let seq = self.seq.load(Ordering::SeqCst); + self.seq.store(seq * 10 + digit, Ordering::SeqCst); + } + } + + impl Freezable for HookOrderTask {} + + impl CuAnytimeTask for HookOrderTask { + type Input<'m> = input_msg!(u32); + type Output<'m> = output_msg!(u32); + type Resources<'r> = Arc; + type Quality = Quality; + + fn new(_config: Option<&ComponentConfig>, seq: Arc) -> CuResult { + Ok(Self { seq }) + } + + fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> { + self.tag(1); + Ok(()) + } + + fn base<'i, 'o>( + &mut self, + _ctx: &CuContext, + _input: &Self::Input<'i>, + output: &mut Self::Output<'o>, + ) -> CuResult> { + self.tag(2); + output.set_payload(0); + Ok(AnytimeStatus::Improved(q(0.5))) + } + + fn refine<'o>( + &mut self, + _ctx: &CuContext, + _output: &mut Self::Output<'o>, + ) -> CuResult> { + self.tag(3); + Ok(AnytimeStatus::Converged(q(1.0))) + } + + fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> { + self.tag(4); + Ok(()) + } + } + + #[test] + fn runner_drives_the_per_job_hooks_in_order() { + let ctx = CuContext::new_mock_clock().0; + let seq = Arc::new(AtomicU32::new(0)); + let mut runner: CuAnytimeRunner = + CuAnytimeRunner::new(None, seq.clone()).unwrap(); + + process_job(&mut runner, &ctx, Tov::None); + assert_eq!(seq.load(Ordering::SeqCst), 1234, "pre, base, refine, post"); + + // The bracket repeats per job, not per run. + process_job(&mut runner, &ctx, Tov::None); + assert_eq!(seq.load(Ordering::SeqCst), 12_341_234); + } + + #[test] + fn per_job_hooks_bracket_even_a_skipped_job() { + let (ctx, clock) = CuContext::new_mock_clock(); + clock.set_value(CuDuration::from_millis(5).0); + let seq = Arc::new(AtomicU32::new(0)); + let mut runner: CuAnytimeRunner = + CuAnytimeRunner::new(None, seq.clone()).unwrap(); + + // A 5 ms old input against a 2 ms horizon: no job runs, but the hooks + // still bracket it — the foreground per-cycle pair is unconditional too. + let output = process_job(&mut runner, &ctx, Tov::Time(CuTime::default())); + assert_eq!(output.payload(), None); + assert_eq!(seq.load(Ordering::SeqCst), 14, "pre, post only"); + } + /// Drives one job and returns the output the runner published. - fn run_job(runner: &mut CuAnytimeRunner, ctx: &CuContext, tov: Tov) -> CuMsg + fn process_job( + runner: &mut CuAnytimeRunner, + ctx: &CuContext, + tov: Tov, + ) -> CuMsg where T: for<'i, 'o> CuAnytimeTask = CuMsg, Output<'o> = CuMsg> + Send @@ -1107,7 +1219,7 @@ mod tests { let mut runner: CuAnytimeRunner = CuAnytimeRunner::new(None, ()).unwrap(); - let output = run_job(&mut runner, &ctx, Tov::None); + let output = process_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"); @@ -1121,7 +1233,7 @@ mod tests { // 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); + let output = process_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"); } @@ -1133,7 +1245,7 @@ mod tests { 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); + let output = process_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"); } @@ -1146,7 +1258,7 @@ mod tests { 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())); + let output = process_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!"); } @@ -1157,7 +1269,7 @@ mod tests { let mut runner: CuAnytimeRunner = CuAnytimeRunner::new(None, ()).unwrap(); - let output = run_job(&mut runner, &ctx, Tov::None); + let output = process_job(&mut runner, &ctx, Tov::None); assert_eq!(output.payload(), None); assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it abort!"); } @@ -1170,7 +1282,7 @@ mod tests { let mut runner: CuAnytimeRunner = CuAnytimeRunner::new(None, clock).unwrap(); - let output = run_job(&mut runner, &ctx, Tov::None); + let output = process_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(),