mirror/rtic
1
0
Fork 0
mirror of https://github.com/rtic-rs/rtic.git synced 2025-12-19 06:15:45 +01:00
Code Issues Projects Releases Packages Wiki Activity

Allow spawning non Send/Sync tasks from the same prio using local spawner

Browse source
This commit is contained in:
Albin Hedman 2025-11-13 01:46:49 +01:00
parent bbc37ca3fe
commit 578bcbff12
6 changed files with 163 additions and 81 deletions
Download patch file Download diff file Expand all files Collapse all files

214
rtic-macros/src/codegen/module.rs
View file

@ -1,5 +1,6 @@
use crate::syntax::{ast::App, Context};
use crate::{analyze::Analysis, codegen::bindings::interrupt_mod, codegen::util};
use proc_macro2::TokenStream as TokenStream2;
use quote::quote;
@ -112,6 +113,144 @@ pub fn codegen(ctxt: Context, app: &App, analysis: &Analysis) -> TokenStream2 {
let internal_context_name = util::internal_task_ident(name, "Context");
let exec_name = util::internal_task_ident(name, "EXEC");
if let Context::SoftwareTask(t) = ctxt {
let spawnee = &app.software_tasks[name];
let priority = spawnee.args.priority;
let cfgs = &spawnee.cfgs;
// Store a copy of the task cfgs
task_cfgs.clone_from(cfgs);
let pend_interrupt = if priority > 0 {
let int_mod = interrupt_mod(app);
let interrupt = &analysis.interrupts.get(&priority).expect("UREACHABLE").0;
quote!(rtic::export::pend(#int_mod::#interrupt);)
} else {
quote!()
};
let internal_spawn_ident = util::internal_task_ident(name, "spawn");
let internal_spawn_helper_ident = util::internal_task_ident(name, "spawn_helper");
let internal_waker_ident = util::internal_task_ident(name, "waker");
let from_ptr_n_args = util::from_ptr_n_args_ident(spawnee.inputs.len());
let (generic_input_args, input_args, input_tupled, input_untupled, input_ty) =
util::regroup_inputs(&spawnee.inputs);
// Spawn caller
items.push(quote!(
#(#cfgs)*
/// Spawns the task without checking if the spawner and spawnee are the same priority
///
/// SAFETY: The caller needs to check that the spawner and spawnee are the same priority
#[allow(non_snake_case)]
#[doc(hidden)]
pub unsafe fn #internal_spawn_helper_ident(#(#input_args,)*) -> ::core::result::Result<(), #input_ty> {
// SAFETY: If `try_allocate` succeeds one must call `spawn`, which we do.
unsafe {
let exec = rtic::export::executor::AsyncTaskExecutor::#from_ptr_n_args(#name, &#exec_name);
if exec.try_allocate() {
exec.spawn(#name(unsafe { #name::Context::new() } #(,#input_untupled)*));
#pend_interrupt
Ok(())
} else {
Err(#input_tupled)
}
}
}
/// Spawns the task directly
#[allow(non_snake_case)]
#[doc(hidden)]
pub fn #internal_spawn_ident(#(#generic_input_args,)*) -> ::core::result::Result<(), #input_ty> {
// SAFETY: The generic args require Send + Sync
unsafe { #internal_spawn_helper_ident(#(#input_untupled.to()),*) }
}
));
// Waker
items.push(quote!(
#(#cfgs)*
/// Gives waker to the task
#[allow(non_snake_case)]
#[doc(hidden)]
pub fn #internal_waker_ident() -> ::core::task::Waker {
// SAFETY: #exec_name is a valid pointer to an executor.
unsafe {
let exec = rtic::export::executor::AsyncTaskExecutor::#from_ptr_n_args(#name, &#exec_name);
exec.waker(|| {
let exec = rtic::export::executor::AsyncTaskExecutor::#from_ptr_n_args(#name, &#exec_name);
exec.set_pending();
#pend_interrupt
})
}
}
));
module_items.push(quote!{
#(#cfgs)*
#[doc(inline)]
pub use super::#internal_spawn_ident as spawn;
});
let tasks_on_same_executor: Vec<_> = app
.software_tasks
.iter()
.filter(|(_, t)| t.args.priority == priority)
.collect();
if !tasks_on_same_executor.is_empty() {
let local_spawner = util::internal_task_ident(t, "LocalSpawner");
fields.push(quote! {
/// Used to spawn tasks on the same executor
///
/// This is useful for tasks that take args which are !Send/!Sync.
pub local_spawner: #local_spawner
});
let tasks = tasks_on_same_executor
.iter()
.map(|(ident, task)| {
// Copied mostly from software_tasks.rs
let internal_spawn_ident = util::internal_task_ident(ident, "spawn_helper");
let attrs = &task.attrs;
let cfgs = &task.cfgs;
let generics = if task.is_bottom {
quote!()
} else {
quote!(<'a>)
};
let (_generic_input_args, input_args, _input_tupled, input_untupled, input_ty) = util::regroup_inputs(&task.inputs);
quote! {
#(#attrs)*
#(#cfgs)*
#[allow(non_snake_case)]
pub(super) fn #ident #generics(&self #(,#input_args)*) -> ::core::result::Result<(), #input_ty> {
// SAFETY: This is safe to call since this can only be called
// from the same executor
unsafe { #internal_spawn_ident(#(#input_untupled,)*) }
}
}
})
.collect::<Vec<_>>();
values.push(quote!(local_spawner: #local_spawner { _p: core::marker::PhantomData }));
items.push(quote! {
struct #local_spawner {
_p: core::marker::PhantomData<*mut ()>,
}
impl #local_spawner {
#(#tasks)*
}
});
}
module_items.push(quote!(
#(#cfgs)*
#[doc(inline)]
pub use super::#internal_waker_ident as waker;
));
}
items.push(quote!(
#(#cfgs)*
/// Execution context
@ -142,81 +281,6 @@ pub fn codegen(ctxt: Context, app: &App, analysis: &Analysis) -> TokenStream2 {
pub use super::#internal_context_name as Context;
));
if let Context::SoftwareTask(..) = ctxt {
let spawnee = &app.software_tasks[name];
let priority = spawnee.args.priority;
let cfgs = &spawnee.cfgs;
// Store a copy of the task cfgs
task_cfgs.clone_from(cfgs);
let pend_interrupt = if priority > 0 {
let int_mod = interrupt_mod(app);
let interrupt = &analysis.interrupts.get(&priority).expect("UREACHABLE").0;
quote!(rtic::export::pend(#int_mod::#interrupt);)
} else {
quote!()
};
let internal_spawn_ident = util::internal_task_ident(name, "spawn");
let internal_waker_ident = util::internal_task_ident(name, "waker");
let from_ptr_n_args = util::from_ptr_n_args_ident(spawnee.inputs.len());
let (input_args, input_tupled, input_untupled, input_ty) =
util::regroup_inputs(&spawnee.inputs);
// Spawn caller
items.push(quote!(
#(#cfgs)*
/// Spawns the task directly
#[allow(non_snake_case)]
#[doc(hidden)]
pub fn #internal_spawn_ident(#(#input_args,)*) -> ::core::result::Result<(), #input_ty> {
// SAFETY: If `try_allocate` succeeds one must call `spawn`, which we do.
unsafe {
let exec = rtic::export::executor::AsyncTaskExecutor::#from_ptr_n_args(#name, &#exec_name);
if exec.try_allocate() {
exec.spawn(#name(unsafe { #name::Context::new() } #(,#input_untupled)*));
#pend_interrupt
Ok(())
} else {
Err(#input_tupled)
}
}
}
));
// Waker
items.push(quote!(
#(#cfgs)*
/// Gives waker to the task
#[allow(non_snake_case)]
#[doc(hidden)]
pub fn #internal_waker_ident() -> ::core::task::Waker {
// SAFETY: #exec_name is a valid pointer to an executor.
unsafe {
let exec = rtic::export::executor::AsyncTaskExecutor::#from_ptr_n_args(#name, &#exec_name);
exec.waker(|| {
let exec = rtic::export::executor::AsyncTaskExecutor::#from_ptr_n_args(#name, &#exec_name);
exec.set_pending();
#pend_interrupt
})
}
}
));
module_items.push(quote!(
#(#cfgs)*
#[doc(inline)]
pub use super::#internal_spawn_ident as spawn;
));
module_items.push(quote!(
#(#cfgs)*
#[doc(inline)]
pub use super::#internal_waker_ident as waker;
));
}
if items.is_empty() {
quote!()
} else {

8
rtic-macros/src/codegen/util.rs
View file

@ -35,6 +35,8 @@ pub fn link_section_uninit() -> TokenStream2 {
pub fn regroup_inputs(
inputs: &[PatType],
) -> (
// Generic args e.g. &[`_0: impl Dummy<T=i32>`, `_1: impl Dummy<T=i64>`]
Vec<TokenStream2>,
// args e.g. &[`_0`], &[`_0: i32`, `_1: i64`]
Vec<TokenStream2>,
// tupled e.g. `_0`, `(_0, _1)`
@ -48,12 +50,14 @@ pub fn regroup_inputs(
let ty = &inputs[0].ty;
(
vec![quote!(_0: impl rtic::export::dummy::Dummy<T=#ty> + Send + Sync)],
vec![quote!(_0: #ty)],
quote!(_0),
vec![quote!(_0)],
quote!(#ty),
)
} else {
let mut generic_args = vec![];
let mut args = vec![];
let mut pats = vec![];
let mut tys = vec![];
@ -62,6 +66,8 @@ pub fn regroup_inputs(
let i = Ident::new(&format!("_{i}"), Span::call_site());
let ty = &input.ty;
generic_args.push(quote!(#i: impl rtic::export::dummy::Dummy<T=#ty> + Send + Sync));
args.push(quote!(#i: #ty));
pats.push(quote!(#i));
@ -74,7 +80,7 @@ pub fn regroup_inputs(
quote!((#(#pats,)*))
};
let ty = quote!((#(#tys,)*));
(args, tupled, pats, ty)
(generic_args, args, tupled, pats, ty)
}
}

5
rtic-macros/src/syntax/analyze.rs
View file

@ -287,11 +287,6 @@ pub(crate) fn app(app: &App) -> Result<Analysis, syn::Error> {
let channel = channels.entry(spawnee_prio).or_default();
channel.tasks.insert(name.clone());
// All inputs are send as we do not know from where they may be spawned.
spawnee.inputs.iter().for_each(|input| {
send_types.insert(input.ty.clone());
});
}
// No channel should ever be empty