Merge #730

730: Executor and task macro fixes. r=Dirbaio a=Dirbaio

See individual commits.

bors r+

Co-authored-by: Dario Nieuwenhuis <dirbaio@dirbaio.net>

embassy-macros/src/macros/task.rs

@@ -10,12 +10,10 @@ struct Args {
     #[darling(default)]
     pool_size: Option<usize>,
     #[darling(default)]
-    send: bool,
-    #[darling(default)]
     embassy_prefix: ModulePrefix,
 }
 
-pub fn run(args: syn::AttributeArgs, mut f: syn::ItemFn) -> Result<TokenStream, TokenStream> {
+pub fn run(args: syn::AttributeArgs, f: syn::ItemFn) -> Result<TokenStream, TokenStream> {
     let args = Args::from_list(&args).map_err(|e| e.write_errors())?;
 
     let embassy_prefix = args.embassy_prefix.append("embassy");
@@ -35,24 +33,27 @@ pub fn run(args: syn::AttributeArgs, mut f: syn::ItemFn) -> Result<TokenStream,
         ctxt.error_spanned_by(&f.sig, "pool_size must be 1 or greater");
     }
 
-    let mut arg_names: syn::punctuated::Punctuated<syn::Ident, syn::Token![,]> =
+    let mut arg_types = Vec::new();
-        syn::punctuated::Punctuated::new();
+    let mut arg_names = Vec::new();
+    let mut arg_indexes = Vec::new();
     let mut fargs = f.sig.inputs.clone();
 
-    for arg in fargs.iter_mut() {
+    for (i, arg) in fargs.iter_mut().enumerate() {
         match arg {
             syn::FnArg::Receiver(_) => {
                 ctxt.error_spanned_by(arg, "task functions must not have receiver arguments");
             }
             syn::FnArg::Typed(t) => match t.pat.as_mut() {
-                syn::Pat::Ident(i) => {
+                syn::Pat::Ident(id) => {
-                    arg_names.push(i.ident.clone());
+                    arg_names.push(id.ident.clone());
-                    i.mutability = None;
+                    arg_types.push(t.ty.clone());
+                    arg_indexes.push(syn::Index::from(i));
+                    id.mutability = None;
                 }
                 _ => {
                     ctxt.error_spanned_by(
                         arg,
-                        "pattern matching in task arguments is not yet supporteds",
+                        "pattern matching in task arguments is not yet supported",
                     );
                 }
             },
@@ -63,40 +64,38 @@ pub fn run(args: syn::AttributeArgs, mut f: syn::ItemFn) -> Result<TokenStream,
 
     let task_ident = f.sig.ident.clone();
     let task_inner_ident = format_ident!("__{}_task", task_ident);
-    let future_ident = format_ident!("__{}_Future", task_ident);
+    let mod_ident = format_ident!("__{}_mod", task_ident);
-    let pool_ident = format_ident!("__{}_POOL", task_ident);
+    let args_ident = format_ident!("__{}_args", task_ident);
-    let new_ts_ident = format_ident!("__{}_NEW_TASKSTORAGE", task_ident);
 
-    let visibility = &f.vis;
+    let mut task_inner = f;
-    f.sig.ident = task_inner_ident.clone();
+    let visibility = task_inner.vis.clone();
-    let impl_ty = if args.send {
+    task_inner.vis = syn::Visibility::Inherited;
-        quote!(impl ::core::future::Future + Send + 'static)
+    task_inner.sig.ident = task_inner_ident.clone();
-    } else {
-        quote!(impl ::core::future::Future + 'static)
-    };
-
-    let attrs = &f.attrs;
-
-    let spawn_token = quote!(#embassy_path::executor::SpawnToken);
-    let task_storage = quote!(#embassy_path::executor::raw::TaskStorage);
 
     let result = quote! {
+        #task_inner
 
         #[allow(non_camel_case_types)]
-        type #future_ident = #impl_ty;
+        type #args_ident = (#(#arg_types,)*);
 
-        #(#attrs)*
+        mod #mod_ident {
-        #visibility fn #task_ident(#fargs) -> #spawn_token<#future_ident> {
+            use #embassy_path::executor::SpawnToken;
-            #f
+            use #embassy_path::executor::raw::TaskStorage;
+
+            type Fut = impl ::core::future::Future + 'static;
 
-            #[allow(non_upper_case_globals)]
             #[allow(clippy::declare_interior_mutable_const)]
-            const #new_ts_ident: #task_storage<#future_ident> = #task_storage::new();
+            const NEW_TS: TaskStorage<Fut> = TaskStorage::new();
 
-            #[allow(non_upper_case_globals)]
+            static POOL: [TaskStorage<Fut>; #pool_size] = [NEW_TS; #pool_size];
-            static #pool_ident: [#task_storage<#future_ident>; #pool_size] = [#new_ts_ident; #pool_size];
 
-            unsafe { #task_storage::spawn_pool(&#pool_ident, move || #task_inner_ident(#arg_names)) }
+            pub(super) fn task(args: super::#args_ident) -> SpawnToken<Fut> {
+                unsafe { TaskStorage::spawn_pool(&POOL, move || super::#task_inner_ident(#(args.#arg_indexes),*)) }
+            }
+        }
+
+        #visibility fn #task_ident(#fargs) -> #embassy_path::executor::SpawnToken<impl ::core::future::Future + 'static> {
+            #mod_ident::task((#(#arg_names,)*))
         }
     };
 

embassy/src/executor/arch/cortex_m.rs

@@ -1,7 +1,7 @@
 use core::marker::PhantomData;
 use core::ptr;
 
-use super::{raw, Spawner};
+use super::{raw, SendSpawner, Spawner};
 use crate::interrupt::{Interrupt, InterruptExt};
 
 /// Thread mode executor, using WFE/SEV.
@@ -107,13 +107,13 @@ impl<I: Interrupt> InterruptExecutor<I> {
 
     /// Start the executor.
     ///
-    /// The `init` closure is called from interrupt mode, with a [`Spawner`] that spawns tasks on
+    /// This initializes the executor, configures and enables the interrupt, and returns.
-    /// this executor. Use it to spawn the initial task(s). After `init` returns,
+    /// The executor keeps running in the background through the interrupt.
-    /// the interrupt is configured so that the executor starts running the tasks.
-    /// Once the executor is started, `start` returns.
     ///
-    /// To spawn more tasks later, you may keep copies of the [`Spawner`] (it is `Copy`),
+    /// This returns a [`SendSpawner`] you can use to spawn tasks on it. A [`SendSpawner`]
-    /// for example by passing it as an argument to the initial tasks.
+    /// is returned instead of a [`Spawner`] because the executor effectively runs in a
+    /// different "thread" (the interrupt), so spawning tasks on it is effectively
+    /// sending them.
     ///
     /// This function requires `&'static mut self`. This means you have to store the
     /// Executor instance in a place where it'll live forever and grants you mutable
@@ -122,16 +122,16 @@ impl<I: Interrupt> InterruptExecutor<I> {
     /// - a [Forever](crate::util::Forever) (safe)
     /// - a `static mut` (unsafe)
     /// - a local variable in a function you know never returns (like `fn main() -> !`), upgrading its lifetime with `transmute`. (unsafe)
-    pub fn start(&'static mut self, init: impl FnOnce(Spawner) + Send) {
+    pub fn start(&'static mut self) -> SendSpawner {
         self.irq.disable();
 
-        init(self.inner.spawner());
-
         self.irq.set_handler(|ctx| unsafe {
             let executor = &*(ctx as *const raw::Executor);
             executor.poll();
         });
         self.irq.set_handler_context(&self.inner as *const _ as _);
         self.irq.enable();
+
+        self.inner.spawner().make_send()
     }
 }

embassy/src/executor/spawner.rs

@@ -105,7 +105,6 @@ impl Spawner {
     pub fn make_send(&self) -> SendSpawner {
         SendSpawner {
             executor: self.executor,
-            not_send: PhantomData,
         }
     }
 }
@@ -120,7 +119,6 @@ impl Spawner {
 #[derive(Copy, Clone)]
 pub struct SendSpawner {
     executor: &'static raw::Executor,
-    not_send: PhantomData<*mut ()>,
 }
 
 unsafe impl Send for SendSpawner {}

examples/nrf/src/bin/multiprio.rs

@@ -124,17 +124,15 @@ fn main() -> ! {
     let irq = interrupt::take!(SWI1_EGU1);
     irq.set_priority(interrupt::Priority::P6);
     let executor = EXECUTOR_HIGH.put(InterruptExecutor::new(irq));
-    executor.start(|spawner| {
+    let spawner = executor.start();
-        unwrap!(spawner.spawn(run_high()));
+    unwrap!(spawner.spawn(run_high()));
-    });
 
     // Medium-priority executor: SWI0_EGU0, priority level 7
     let irq = interrupt::take!(SWI0_EGU0);
     irq.set_priority(interrupt::Priority::P7);
     let executor = EXECUTOR_MED.put(InterruptExecutor::new(irq));
-    executor.start(|spawner| {
+    let spawner = executor.start();
-        unwrap!(spawner.spawn(run_med()));
+    unwrap!(spawner.spawn(run_med()));
-    });
 
     // Low priority executor: runs in thread mode, using WFE/SEV
     let executor = EXECUTOR_LOW.put(Executor::new());

examples/nrf/src/bin/self_spawn.rs

@@ -0,0 +1,24 @@
+#![no_std]
+#![no_main]
+#![feature(type_alias_impl_trait)]
+
+use defmt::{info, unwrap};
+use embassy::executor::Spawner;
+use embassy::time::{Duration, Timer};
+use embassy_nrf::Peripherals;
+
+use defmt_rtt as _; // global logger
+use panic_probe as _;
+
+#[embassy::task(pool_size = 2)]
+async fn my_task(spawner: Spawner, n: u32) {
+    Timer::after(Duration::from_secs(1)).await;
+    info!("Spawning self! {}", n);
+    unwrap!(spawner.spawn(my_task(spawner, n + 1)));
+}
+
+#[embassy::main]
+async fn main(spawner: Spawner, _p: Peripherals) {
+    info!("Hello World!");
+    unwrap!(spawner.spawn(my_task(spawner, 0)));
+}

examples/stm32f3/src/bin/multiprio.rs

@@ -124,17 +124,15 @@ fn main() -> ! {
     let irq = interrupt::take!(UART4);
     irq.set_priority(interrupt::Priority::P6);
     let executor = EXECUTOR_HIGH.put(InterruptExecutor::new(irq));
-    executor.start(|spawner| {
+    let spawner = executor.start();
-        unwrap!(spawner.spawn(run_high()));
+    unwrap!(spawner.spawn(run_high()));
-    });
 
     // Medium-priority executor: SWI0_EGU0, priority level 7
     let irq = interrupt::take!(UART5);
     irq.set_priority(interrupt::Priority::P7);
     let executor = EXECUTOR_MED.put(InterruptExecutor::new(irq));
-    executor.start(|spawner| {
+    let spawner = executor.start();
-        unwrap!(spawner.spawn(run_med()));
+    unwrap!(spawner.spawn(run_med()));
-    });
 
     // Low priority executor: runs in thread mode, using WFE/SEV
     let executor = EXECUTOR_LOW.put(Executor::new());

examples/stm32f4/src/bin/multiprio.rs

@@ -124,17 +124,15 @@ fn main() -> ! {
     let irq = interrupt::take!(UART4);
     irq.set_priority(interrupt::Priority::P6);
     let executor = EXECUTOR_HIGH.put(InterruptExecutor::new(irq));
-    executor.start(|spawner| {
+    let spawner = executor.start();
-        unwrap!(spawner.spawn(run_high()));
+    unwrap!(spawner.spawn(run_high()));
-    });
 
     // Medium-priority executor: SWI0_EGU0, priority level 7
     let irq = interrupt::take!(UART5);
     irq.set_priority(interrupt::Priority::P7);
     let executor = EXECUTOR_MED.put(InterruptExecutor::new(irq));
-    executor.start(|spawner| {
+    let spawner = executor.start();
-        unwrap!(spawner.spawn(run_med()));
+    unwrap!(spawner.spawn(run_med()));
-    });
 
     // Low priority executor: runs in thread mode, using WFE/SEV
     let executor = EXECUTOR_LOW.put(Executor::new());