Skip to content

feat: background placement for anytime tasks #1248

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Scofield626 wants to merge 8 commits into
base: master
Choose a base branch
from
Draft

feat: background placement for anytime tasks #1248

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
5455283
feat: anytime background runner (one whole job per process call)
Scofield626 Jul 30, 2026
d7d9180
feat: codegen for anytime nodes with background: true
Scofield626 Jul 30, 2026
f788b86
example: background anytime node in cu_anytime_task
Scofield626 Jul 30, 2026
5bfcd61
fix: review findings on the background anytime runner
Scofield626 Jul 30, 2026
0417573
chore: fmt (ron)
Scofield626 Jul 30, 2026
57eee2f
fix: drop unneeded T: Reflect bound on CuAnytimeRunner
Scofield626 Aug 3, 2026
be9339b
cuasynctask: forward preprocess/postprocess best-effort via try_lock
Scofield626 Aug 4, 2026
6b741d4
cu29-derive: resolve the background inner type once in CuTaskSpecSet
Scofield626 Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
16 changes: 8 additions & 8 deletions api/v1/cu29-runtime.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -525,8 +525,8 @@ pub type cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::Output<'m> = <T as cu2
pub type cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::Resources<'r> = cu29_runtime::cuasynctask::CuAsyncSrcTaskResources<'r, T>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::debug_state_type_path() -> &'static str where Self: cu29_runtime::reflect::TypePath + core::marker::Sized
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::new(config: core::option::Option<&cu29_runtime::config::ComponentConfig>, resources: Self::Resources) -> cu29_traits::CuResult<Self> where Self: core::marker::Sized
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::postprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::preprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::postprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::preprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::process<'o>(&mut self, ctx: &cu29_runtime::context::CuContext, real_output: &mut Self::Output) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::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<T, O>::start(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
Expand All @@ -553,8 +553,8 @@ pub type cu29_runtime::cuasynctask::CuAsyncTask<T, O>::Output<'m> = <T as cu29_r
pub type cu29_runtime::cuasynctask::CuAsyncTask<T, O>::Resources<'r> = cu29_runtime::cuasynctask::CuAsyncTaskResources<'r, T>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::debug_state_type_path() -> &'static str where Self: cu29_runtime::reflect::TypePath + core::marker::Sized
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::new(config: core::option::Option<&cu29_runtime::config::ComponentConfig>, resources: Self::Resources) -> cu29_traits::CuResult<Self> where Self: core::marker::Sized
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::postprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::preprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::postprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::preprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::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<T, O>::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<T, O>::start(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
Expand Down Expand Up @@ -920,8 +920,8 @@ pub type cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::Output<'m> = <T as cu2
pub type cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::Resources<'r> = cu29_runtime::cuasynctask::CuAsyncSrcTaskResources<'r, T>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::debug_state_type_path() -> &'static str where Self: cu29_runtime::reflect::TypePath + core::marker::Sized
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::new(config: core::option::Option<&cu29_runtime::config::ComponentConfig>, resources: Self::Resources) -> cu29_traits::CuResult<Self> where Self: core::marker::Sized
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::postprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::preprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::postprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::preprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::process<'o>(&mut self, ctx: &cu29_runtime::context::CuContext, real_output: &mut Self::Output) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncSrcTask<T, O>::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<T, O>::start(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
Expand Down Expand Up @@ -958,8 +958,8 @@ pub type cu29_runtime::cuasynctask::CuAsyncTask<T, O>::Output<'m> = <T as cu29_r
pub type cu29_runtime::cuasynctask::CuAsyncTask<T, O>::Resources<'r> = cu29_runtime::cuasynctask::CuAsyncTaskResources<'r, T>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::debug_state_type_path() -> &'static str where Self: cu29_runtime::reflect::TypePath + core::marker::Sized
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::new(config: core::option::Option<&cu29_runtime::config::ComponentConfig>, resources: Self::Resources) -> cu29_traits::CuResult<Self> where Self: core::marker::Sized
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::postprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::preprocess(&mut self, _ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::postprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::preprocess(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
pub fn cu29_runtime::cuasynctask::CuAsyncTask<T, O>::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<T, O>::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<T, O>::start(&mut self, ctx: &cu29_runtime::context::CuContext) -> cu29_traits::CuResult<()>
Expand Down
88 changes: 55 additions & 33 deletions core/cu29_derive/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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.background_inner_types[index];
match task_specs.cutypes[index] {
CuTaskType::Source => {
if background {
Expand Down Expand Up @@ -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.background_inner_types[index];
match task_specs.cutypes[index] {
CuTaskType::Source => {
if *background {
Expand Down Expand Up @@ -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])
Expand Down Expand Up @@ -6166,6 +6162,10 @@ struct CuTaskSpecSet {
/// are baked into a per-node `AnytimePolicy` ZST and the emitted
/// base/refine steps.
pub anytime_configs: Vec<Option<AnytimeConfig>>,
/// What the background wrapper wraps, per task: the anytime runner for a
/// background anytime node, the declared task type otherwise (including
/// every foreground task, where nothing wraps it).
pub background_inner_types: Vec<Type>,
pub logging_enabled: Vec<bool>,
pub type_names: Vec<String>,
pub task_types: Vec<Type>,
Expand All @@ -6186,7 +6186,7 @@ impl CuTaskSpecSet {
.filter(|(_, node)| node.get_flavor() == Flavor::Task)
.collect();

let ids = all_id_nodes
let ids: Vec<String> = all_id_nodes
.iter()
.map(|(_, node)| node.get_id().to_string())
.collect();
Expand Down Expand Up @@ -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 background_inner_types: Vec<Type> = 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(background_inner_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.");
Expand All @@ -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(background_inner_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.");
Expand Down Expand Up @@ -6334,6 +6353,7 @@ impl CuTaskSpecSet {
background_flags,
background_pools,
anytime_configs,
background_inner_types,
logging_enabled,
type_names,
task_types,
Expand Down Expand Up @@ -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.background_inner_types[idx];

let binding_trait = task_trait_for_specs(task_specs, idx);

Expand Down Expand Up @@ -8561,10 +8579,17 @@ fn build_anytime_policy_defs(task_specs: &CuTaskSpecSet) -> Vec<proc_macro2::Tok
Some(stall) => 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<cu29::clock::CuDuration> = #time_budget;
const MAX_AGE: Option<cu29::clock::CuDuration> = #max_age;
const MAX_STALL: Option<u32> = #max_stall;
const MAX_REFINES: Option<u32> = #max_refines;
};
let has_quality_knob = anytime.quality_target.is_some()
|| anytime.quality_floor.is_some()
Expand Down Expand Up @@ -8639,7 +8664,9 @@ fn build_anytime_job_locals(task_specs: &CuTaskSpecSet) -> Vec<proc_macro2::Toke
.anytime_configs
.iter()
.enumerate()
.filter(|(_, anytime)| anytime.is_some())
// A background anytime node keeps its single `Whole` step and runs its
// job inside the runner, so it has no plan-level job to carry.
.filter(|(index, anytime)| anytime.is_some() && !task_specs.background_flags[*index])
.map(|(index, _)| anytime_job_local_tokens(task_specs, index))
.collect()
}
Expand Down Expand Up @@ -8885,13 +8912,7 @@ fn generate_anytime_base_block(
let base_dispatch = if let Some(max_age_ms) = anytime.max_age_ms {
let max_age_nanos = anytime_ms_to_nanos(max_age_ms);
quote! {
let __cu_any_anchor: cu29::clock::CuTime = match cumsg_input.tov {
cu29::clock::Tov::Time(tov_time) => 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);
Expand Down Expand Up @@ -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.background_inner_types[index];
parse_quote!(CuAsyncTask<#inner, #out_ty>)
} else {
panic!("{task_id}: If a task is background, it has to have an output");
}
Expand Down
Loading
Loading