Merge pull request #44 from akiles/executor-v2

Executor API V2.

.vscode/settings.json

@@ -1,4 +1,5 @@
 {
+  "rust-analyzer.assist.importMergeBehavior": "last",
   "editor.formatOnSave": true,
   "rust-analyzer.cargo.allFeatures": false,
   "rust-analyzer.checkOnSave.allFeatures": false,
@@ -8,4 +9,4 @@
     "**/.git/subtree-cache/**": true,
     "**/target/**": true
   }
-}
+}

embassy-macros/src/lib.rs

@@ -11,21 +11,23 @@ use syn::spanned::Spanned;
 struct MacroArgs {
     #[darling(default)]
     pool_size: Option<usize>,
+    #[darling(default)]
+    send: bool,
 }
 
 #[proc_macro_attribute]
 pub fn task(args: TokenStream, item: TokenStream) -> TokenStream {
-    let args = syn::parse_macro_input!(args as syn::AttributeArgs);
+    let macro_args = syn::parse_macro_input!(args as syn::AttributeArgs);
     let mut task_fn = syn::parse_macro_input!(item as syn::ItemFn);
 
-    let args = match MacroArgs::from_list(&args) {
+    let macro_args = match MacroArgs::from_list(&macro_args) {
         Ok(v) => v,
         Err(e) => {
             return TokenStream::from(e.write_errors());
         }
     };
 
-    let pool_size: usize = args.pool_size.unwrap_or(1);
+    let pool_size: usize = macro_args.pool_size.unwrap_or(1);
 
     let mut fail = false;
     if task_fn.sig.asyncness.is_none() {
@@ -90,11 +92,16 @@ pub fn task(args: TokenStream, item: TokenStream) -> TokenStream {
 
     let visibility = &task_fn.vis;
     task_fn.sig.ident = format_ident!("task");
+    let impl_ty = if macro_args.send {
+        quote!(impl ::core::future::Future + Send + 'static)
+    } else {
+        quote!(impl ::core::future::Future + 'static)
+    };
 
     let result = quote! {
-        #visibility fn #name(#args) -> ::embassy::executor::SpawnToken {
+        #visibility fn #name(#args) -> ::embassy::executor::SpawnToken<#impl_ty> {
             #task_fn
-            type F = impl ::core::future::Future + 'static;
+            type F = #impl_ty;
             static POOL: [::embassy::executor::Task<F>; #pool_size] = [::embassy::executor::Task::new(); #pool_size];
             unsafe { ::embassy::executor::Task::spawn(&POOL, move || task(#arg_names)) }
         }
@@ -119,6 +126,9 @@ pub fn interrupt_declare(item: TokenStream) -> TokenStream {
                 let irq = Interrupt::#name;
                 irq.nr() as u8
             }
+            unsafe fn steal() -> Self {
+                Self(())
+            }
             unsafe fn __handler(&self) -> &'static ::embassy::interrupt::Handler {
                 #[export_name = #name_handler]
                 static HANDLER: ::embassy::interrupt::Handler = ::embassy::interrupt::Handler::new();

embassy-nrf-examples/src/bin/buffered_uart.rs

@@ -83,11 +83,8 @@ static EXECUTOR: Forever<Executor> = Forever::new();
 fn main() -> ! {
     info!("Hello World!");
 
-    let executor = EXECUTOR.put(Executor::new(cortex_m::asm::sev));
+    let executor = EXECUTOR.put(Executor::new());
-    unwrap!(executor.spawn(run()));
+    executor.run(|spawner| {
-
+        unwrap!(spawner.spawn(run()));
-    loop {
+    });
-        executor.run();
-        cortex_m::asm::wfe();
-    }
 }

embassy-nrf-examples/src/bin/executor_fairness_test.rs

@@ -61,14 +61,11 @@ fn main() -> ! {
     unsafe { embassy::time::set_clock(rtc) };
 
     let alarm = ALARM.put(rtc.alarm0());
-    let executor = EXECUTOR.put(Executor::new_with_alarm(alarm, cortex_m::asm::sev));
+    let executor = EXECUTOR.put(Executor::new());
-
+    executor.set_alarm(alarm);
-    unwrap!(executor.spawn(run1()));
+    executor.run(|spawner| {
-    unwrap!(executor.spawn(run2()));
+        unwrap!(spawner.spawn(run1()));
-    unwrap!(executor.spawn(run3()));
+        unwrap!(spawner.spawn(run2()));
-
+        unwrap!(spawner.spawn(run3()));
-    loop {
+    });
-        executor.run();
-        cortex_m::asm::wfe();
-    }
 }

embassy-nrf-examples/src/bin/gpiote.rs

@@ -73,11 +73,8 @@ static EXECUTOR: Forever<Executor> = Forever::new();
 fn main() -> ! {
     info!("Hello World!");
 
-    let executor = EXECUTOR.put(Executor::new(cortex_m::asm::sev));
+    let executor = EXECUTOR.put(Executor::new());
-    unwrap!(executor.spawn(run()));
+    executor.run(|spawner| {
-
+        unwrap!(spawner.spawn(run()));
-    loop {
+    });
-        executor.run();
-        cortex_m::asm::wfe();
-    }
 }

embassy-nrf-examples/src/bin/gpiote_port.rs

@@ -52,11 +52,8 @@ static EXECUTOR: Forever<Executor> = Forever::new();
 fn main() -> ! {
     info!("Hello World!");
 
-    let executor = EXECUTOR.put(Executor::new(cortex_m::asm::sev));
+    let executor = EXECUTOR.put(Executor::new());
-    unwrap!(executor.spawn(run()));
+    executor.run(|spawner| {
-
+        unwrap!(spawner.spawn(run()));
-    loop {
+    });
-        executor.run();
-        cortex_m::asm::wfe();
-    }
 }

embassy-nrf-examples/src/bin/multiprio.rs

@@ -66,9 +66,10 @@ use cortex_m_rt::entry;
 use defmt::panic;
 use nrf52840_hal::clocks;
 
-use embassy::executor::{task, Executor};
+use embassy::executor::{task, Executor, IrqExecutor};
 use embassy::time::{Duration, Instant, Timer};
 use embassy::util::Forever;
+use embassy_nrf::interrupt::OwnedInterrupt;
 use embassy_nrf::{interrupt, pac, rtc};
 
 #[task]
@@ -114,12 +115,12 @@ async fn run_low() {
 }
 
 static RTC: Forever<rtc::RTC<pac::RTC1>> = Forever::new();
+static ALARM_HIGH: Forever<rtc::Alarm<pac::RTC1>> = Forever::new();
+static EXECUTOR_HIGH: Forever<IrqExecutor<interrupt::SWI1_EGU1Interrupt>> = Forever::new();
+static ALARM_MED: Forever<rtc::Alarm<pac::RTC1>> = Forever::new();
+static EXECUTOR_MED: Forever<IrqExecutor<interrupt::SWI0_EGU0Interrupt>> = Forever::new();
 static ALARM_LOW: Forever<rtc::Alarm<pac::RTC1>> = Forever::new();
 static EXECUTOR_LOW: Forever<Executor> = Forever::new();
-static ALARM_MED: Forever<rtc::Alarm<pac::RTC1>> = Forever::new();
-static EXECUTOR_MED: Forever<Executor> = Forever::new();
-static ALARM_HIGH: Forever<rtc::Alarm<pac::RTC1>> = Forever::new();
-static EXECUTOR_HIGH: Forever<Executor> = Forever::new();
 
 #[entry]
 fn main() -> ! {
@@ -136,41 +137,31 @@ fn main() -> ! {
     rtc.start();
     unsafe { embassy::time::set_clock(rtc) };
 
-    let alarm_low = ALARM_LOW.put(rtc.alarm0());
+    // High-priority executor: SWI1_EGU1, priority level 6
-    let executor_low = EXECUTOR_LOW.put(Executor::new_with_alarm(alarm_low, cortex_m::asm::sev));
+    let irq = interrupt::take!(SWI1_EGU1);
-    let alarm_med = ALARM_MED.put(rtc.alarm1());
+    irq.set_priority(interrupt::Priority::Level6);
-    let executor_med = EXECUTOR_MED.put(Executor::new_with_alarm(alarm_med, || {
+    let alarm = ALARM_HIGH.put(rtc.alarm2());
-        NVIC::pend(interrupt::SWI0_EGU0)
+    let executor = EXECUTOR_HIGH.put(IrqExecutor::new(irq));
-    }));
+    executor.set_alarm(alarm);
-    let alarm_high = ALARM_HIGH.put(rtc.alarm2());
+    executor.start(|spawner| {
-    let executor_high = EXECUTOR_HIGH.put(Executor::new_with_alarm(alarm_high, || {
+        unwrap!(spawner.spawn(run_high()));
-        NVIC::pend(interrupt::SWI1_EGU1)
+    });
-    }));
 
-    unsafe {
+    // Medium-priority executor: SWI0_EGU0, priority level 7
-        let mut nvic: NVIC = core::mem::transmute(());
+    let irq = interrupt::take!(SWI0_EGU0);
-        nvic.set_priority(interrupt::SWI0_EGU0, 7 << 5);
+    irq.set_priority(interrupt::Priority::Level7);
-        nvic.set_priority(interrupt::SWI1_EGU1, 6 << 5);
+    let alarm = ALARM_MED.put(rtc.alarm1());
-        NVIC::unmask(interrupt::SWI0_EGU0);
+    let executor = EXECUTOR_MED.put(IrqExecutor::new(irq));
-        NVIC::unmask(interrupt::SWI1_EGU1);
+    executor.set_alarm(alarm);
-    }
+    executor.start(|spawner| {
+        unwrap!(spawner.spawn(run_med()));
+    });
 
-    unwrap!(executor_low.spawn(run_low()));
+    // Low priority executor: runs in thread mode, using WFE/SEV
-    unwrap!(executor_med.spawn(run_med()));
+    let alarm = ALARM_LOW.put(rtc.alarm0());
-    unwrap!(executor_high.spawn(run_high()));
+    let executor = EXECUTOR_LOW.put(Executor::new());
-
+    executor.set_alarm(alarm);
-    loop {
+    executor.run(|spawner| {
-        executor_low.run();
+        unwrap!(spawner.spawn(run_low()));
-        cortex_m::asm::wfe();
+    });
-    }
-}
-
-#[interrupt]
-unsafe fn SWI0_EGU0() {
-    EXECUTOR_MED.steal().run()
-}
-
-#[interrupt]
-unsafe fn SWI1_EGU1() {
-    EXECUTOR_HIGH.steal().run()
 }

embassy-nrf-examples/src/bin/qspi.rs

@@ -124,11 +124,8 @@ static EXECUTOR: Forever<Executor> = Forever::new();
 fn main() -> ! {
     info!("Hello World!");
 
-    let executor = EXECUTOR.put(Executor::new(cortex_m::asm::sev));
+    let executor = EXECUTOR.put(Executor::new());
-    unwrap!(executor.spawn(run()));
+    executor.run(|spawner| {
-
+        unwrap!(spawner.spawn(run()));
-    loop {
+    });
-        executor.run();
-        cortex_m::asm::wfe();
-    }
 }

embassy-nrf-examples/src/bin/rtc_async.rs

@@ -53,13 +53,10 @@ fn main() -> ! {
     unsafe { embassy::time::set_clock(rtc) };
 
     let alarm = ALARM.put(rtc.alarm0());
-    let executor = EXECUTOR.put(Executor::new_with_alarm(alarm, cortex_m::asm::sev));
+    let executor = EXECUTOR.put(Executor::new());
-
+    executor.set_alarm(alarm);
-    unwrap!(executor.spawn(run1()));
+    executor.run(|spawner| {
-    unwrap!(executor.spawn(run2()));
+        unwrap!(spawner.spawn(run1()));
-
+        unwrap!(spawner.spawn(run2()));
-    loop {
+    });
-        executor.run();
-        cortex_m::asm::wfe();
-    }
 }

embassy-nrf-examples/src/bin/rtc_raw.rs

@@ -38,7 +38,7 @@ fn main() -> ! {
 
     rtc.start();
 
-    alarm.set_callback(|| info!("ALARM TRIGGERED"));
+    alarm.set_callback(|_| info!("ALARM TRIGGERED"), core::ptr::null_mut());
     alarm.set(53719);
 
     info!("initialized!");

embassy-nrf-examples/src/bin/uart.rs

@@ -18,7 +18,31 @@ use nrf52840_hal::clocks;
 use nrf52840_hal::gpio;
 
 #[task]
-async fn run(mut uart: uarte::Uarte<pac::UARTE0>) {
+async fn run(uart: pac::UARTE0, port: pac::P0) {
+    // Init UART
+    let port0 = gpio::p0::Parts::new(port);
+
+    let pins = uarte::Pins {
+        rxd: port0.p0_08.into_floating_input().degrade(),
+        txd: port0
+            .p0_06
+            .into_push_pull_output(gpio::Level::Low)
+            .degrade(),
+        cts: None,
+        rts: None,
+    };
+
+    // NOTE(unsafe): Safe becasue we do not use `mem::forget` anywhere.
+    let mut uart = unsafe {
+        uarte::Uarte::new(
+            uart,
+            interrupt::take!(UARTE0_UART0),
+            pins,
+            uarte::Parity::EXCLUDED,
+            uarte::Baudrate::BAUD115200,
+        )
+    };
+
     info!("uarte initialized!");
 
     // Message must be in SRAM
@@ -81,36 +105,12 @@ fn main() -> ! {
     unsafe { embassy::time::set_clock(rtc) };
 
     let alarm = ALARM.put(rtc.alarm0());
-    let executor = EXECUTOR.put(Executor::new_with_alarm(alarm, cortex_m::asm::sev));
+    let executor = EXECUTOR.put(Executor::new());
+    executor.set_alarm(alarm);
 
-    // Init UART
+    let uarte0 = p.UARTE0;
-    let port0 = gpio::p0::Parts::new(p.P0);
+    let p0 = p.P0;
-
+    executor.run(|spawner| {
-    let pins = uarte::Pins {
+        unwrap!(spawner.spawn(run(uarte0, p0)));
-        rxd: port0.p0_08.into_floating_input().degrade(),
+    });
-        txd: port0
-            .p0_06
-            .into_push_pull_output(gpio::Level::Low)
-            .degrade(),
-        cts: None,
-        rts: None,
-    };
-
-    // NOTE(unsafe): Safe becasue we do not use `mem::forget` anywhere.
-    let uart = unsafe {
-        uarte::Uarte::new(
-            p.UARTE0,
-            interrupt::take!(UARTE0_UART0),
-            pins,
-            uarte::Parity::EXCLUDED,
-            uarte::Baudrate::BAUD115200,
-        )
-    };
-
-    unwrap!(executor.spawn(run(uart)));
-
-    loop {
-        executor.run();
-        cortex_m::asm::wfe();
-    }
 }

embassy-nrf/src/rtc.rs

@@ -40,7 +40,7 @@ mod test {
 
 struct AlarmState {
     timestamp: Cell<u64>,
-    callback: Cell<Option<fn()>>,
+    callback: Cell<Option<(fn(*mut ()), *mut ())>>,
 }
 
 impl AlarmState {
@@ -159,13 +159,13 @@ impl<T: Instance> RTC<T> {
         alarm.timestamp.set(u64::MAX);
 
         // Call after clearing alarm, so the callback can set another alarm.
-        alarm.callback.get().map(|f| f());
+        alarm.callback.get().map(|(f, ctx)| f(ctx));
     }
 
-    fn set_alarm_callback(&self, n: usize, callback: fn()) {
+    fn set_alarm_callback(&self, n: usize, callback: fn(*mut ()), ctx: *mut ()) {
         interrupt::free(|cs| {
             let alarm = &self.alarms.borrow(cs)[n];
-            alarm.callback.set(Some(callback));
+            alarm.callback.set(Some((callback, ctx)));
         })
     }
 
@@ -220,8 +220,8 @@ pub struct Alarm<T: Instance> {
 }
 
 impl<T: Instance> embassy::time::Alarm for Alarm<T> {
-    fn set_callback(&self, callback: fn()) {
+    fn set_callback(&self, callback: fn(*mut ()), ctx: *mut ()) {
-        self.rtc.set_alarm_callback(self.n, callback);
+        self.rtc.set_alarm_callback(self.n, callback, ctx);
     }
 
     fn set(&self, timestamp: u64) {

embassy-stm32f4-examples/src/bin/exti.rs

@@ -49,11 +49,8 @@ fn main() -> ! {
     let dp = stm32::Peripherals::take().unwrap();
     let cp = cortex_m::peripheral::Peripherals::take().unwrap();
 
-    let executor = EXECUTOR.put(Executor::new(cortex_m::asm::sev));
+    let executor = EXECUTOR.put(Executor::new());
-    executor.spawn(run(dp, cp)).unwrap();
+    executor.run(|spawner| {
-
+        unwrap!(spawner.spawn(run(dp, cp)));
-    loop {
+    });
-        executor.run();
-        //cortex_m::asm::wfe(); // wfe causes RTT to stop working on stm32
-    }
 }

embassy-stm32f4-examples/src/bin/serial.rs

@@ -59,11 +59,8 @@ fn main() -> ! {
     let dp = stm32::Peripherals::take().unwrap();
     let cp = cortex_m::peripheral::Peripherals::take().unwrap();
 
-    let executor = EXECUTOR.put(Executor::new(cortex_m::asm::sev));
+    let executor = EXECUTOR.put(Executor::new());
-    executor.spawn(run(dp, cp)).unwrap();
+    executor.run(|spawner| {
-
+        unwrap!(spawner.spawn(run(dp, cp)));
-    loop {
+    });
-        executor.run();
-        //cortex_m::asm::wfe(); // wfe causes RTT to stop working on stm32
-    }
 }

embassy/src/executor/mod.rs

@@ -2,129 +2,68 @@ pub use embassy_macros::task;
 
 use core::future::Future;
 use core::marker::PhantomData;
-use core::mem;
 use core::pin::Pin;
-use core::ptr;
 use core::ptr::NonNull;
-use core::sync::atomic::{AtomicU32, Ordering};
+use core::sync::atomic::Ordering;
-use core::task::{Context, Poll, Waker};
+use core::task::{Context, Poll};
-use core::{
+use core::{mem, ptr};
-    cell::{Cell, UnsafeCell},
-    cmp::min,
-};
 
+pub mod raw;
 mod run_queue;
 pub(crate) mod timer;
 mod timer_queue;
 mod util;
 mod waker;
 
-use self::run_queue::{RunQueue, RunQueueItem};
-use self::timer_queue::{TimerQueue, TimerQueueItem};
 use self::util::UninitCell;
-use crate::{
+use crate::fmt::{panic, *};
-    fmt::{panic, *},
+use crate::interrupt::OwnedInterrupt;
-    time::{Alarm, Instant},
+use crate::time::Alarm;
-};
 
-/// Task is spawned (has a future)
+// repr(C) is needed to guarantee that the raw::Task is located at offset 0
-pub(crate) const STATE_SPAWNED: u32 = 1 << 0;
+// This makes it safe to cast between raw::Task and Task pointers.
-/// Task is in the executor run queue
-pub(crate) const STATE_RUN_QUEUED: u32 = 1 << 1;
-/// Task is in the executor timer queue
-pub(crate) const STATE_TIMER_QUEUED: u32 = 1 << 2;
-
-pub(crate) struct TaskHeader {
-    state: AtomicU32,
-    run_queue_item: RunQueueItem,
-    expires_at: Cell<Instant>,
-    timer_queue_item: TimerQueueItem,
-    executor: Cell<*const Executor>, // Valid if state != 0
-    poll_fn: UninitCell<unsafe fn(*mut TaskHeader)>, // Valid if STATE_SPAWNED
-}
-
-impl TaskHeader {
-    const fn new() -> Self {
-        Self {
-            state: AtomicU32::new(0),
-            expires_at: Cell::new(Instant::from_ticks(0)),
-            run_queue_item: RunQueueItem::new(),
-            timer_queue_item: TimerQueueItem::new(),
-            executor: Cell::new(ptr::null()),
-            poll_fn: UninitCell::uninit(),
-        }
-    }
-
-    pub(crate) unsafe fn enqueue(&self) {
-        let mut current = self.state.load(Ordering::Acquire);
-        loop {
-            // If already scheduled, or if not started,
-            if (current & STATE_RUN_QUEUED != 0) || (current & STATE_SPAWNED == 0) {
-                return;
-            }
-
-            // Mark it as scheduled
-            let new = current | STATE_RUN_QUEUED;
-
-            match self.state.compare_exchange_weak(
-                current,
-                new,
-                Ordering::AcqRel,
-                Ordering::Acquire,
-            ) {
-                Ok(_) => break,
-                Err(next_current) => current = next_current,
-            }
-        }
-
-        // We have just marked the task as scheduled, so enqueue it.
-        let executor = &*self.executor.get();
-        executor.enqueue(self as *const TaskHeader as *mut TaskHeader);
-    }
-}
-
-// repr(C) is needed to guarantee that header is located at offset 0
-// This makes it safe to cast between Header and Task pointers.
 #[repr(C)]
 pub struct Task<F: Future + 'static> {
-    header: TaskHeader,
+    raw: raw::Task,
     future: UninitCell<F>, // Valid if STATE_SPAWNED
 }
 
 impl<F: Future + 'static> Task<F> {
     pub const fn new() -> Self {
         Self {
-            header: TaskHeader::new(),
+            raw: raw::Task::new(),
             future: UninitCell::uninit(),
         }
     }
 
-    pub unsafe fn spawn(pool: &'static [Self], future: impl FnOnce() -> F) -> SpawnToken {
+    pub unsafe fn spawn(pool: &'static [Self], future: impl FnOnce() -> F) -> SpawnToken<F> {
         for task in pool {
-            let state = STATE_SPAWNED | STATE_RUN_QUEUED;
+            let state = raw::STATE_SPAWNED | raw::STATE_RUN_QUEUED;
             if task
-                .header
+                .raw
                 .state
                 .compare_exchange(0, state, Ordering::AcqRel, Ordering::Acquire)
                 .is_ok()
             {
                 // Initialize the task
-                task.header.poll_fn.write(Self::poll);
+                task.raw.poll_fn.write(Self::poll);
                 task.future.write(future());
 
                 return SpawnToken {
-                    header: Some(NonNull::new_unchecked(
+                    raw_task: Some(NonNull::new_unchecked(&task.raw as *const raw::Task as _)),
-                        &task.header as *const TaskHeader as _,
+                    phantom: PhantomData,
-                    )),
                 };
             }
         }
 
-        return SpawnToken { header: None };
+        return SpawnToken {
+            raw_task: None,
+            phantom: PhantomData,
+        };
     }
 
-    unsafe fn poll(p: *mut TaskHeader) {
+    unsafe fn poll(p: NonNull<raw::Task>) {
-        let this = &*(p as *const Task<F>);
+        let this = &*(p.as_ptr() as *const Task<F>);
 
         let future = Pin::new_unchecked(this.future.as_mut());
         let waker = waker::from_task(p);
@@ -132,9 +71,9 @@ impl<F: Future + 'static> Task<F> {
         match future.poll(&mut cx) {
             Poll::Ready(_) => {
                 this.future.drop_in_place();
-                this.header
+                this.raw
                     .state
-                    .fetch_and(!STATE_SPAWNED, Ordering::AcqRel);
+                    .fetch_and(!raw::STATE_SPAWNED, Ordering::AcqRel);
             }
             Poll::Pending => {}
         }
@@ -144,11 +83,12 @@ impl<F: Future + 'static> Task<F> {
 unsafe impl<F: Future + 'static> Sync for Task<F> {}
 
 #[must_use = "Calling a task function does nothing on its own. You must pass the returned SpawnToken to Executor::spawn()"]
-pub struct SpawnToken {
+pub struct SpawnToken<F> {
-    header: Option<NonNull<TaskHeader>>,
+    raw_task: Option<NonNull<raw::Task>>,
+    phantom: PhantomData<*mut F>,
 }
 
-impl Drop for SpawnToken {
+impl<F> Drop for SpawnToken<F> {
     fn drop(&mut self) {
         // TODO deallocate the task instead.
         panic!("SpawnToken instances may not be dropped. You must pass them to Executor::spawn()")
@@ -161,116 +101,167 @@ pub enum SpawnError {
     Busy,
 }
 
-pub struct Executor {
+/// Handle to spawn tasks into an executor.
-    alarm: Option<&'static dyn Alarm>,
+///
-    run_queue: RunQueue,
+/// This Spawner can spawn any task (Send and non-Send ones), but it can
-    timer_queue: TimerQueue,
+/// only be used in the executor thread (it is not Send itself).
-    signal_fn: fn(),
+///
+/// If you want to spawn tasks from another thread, use [SendSpawner].
+pub struct Spawner {
+    executor: &'static raw::Executor,
     not_send: PhantomData<*mut ()>,
 }
 
-impl Executor {
+impl Spawner {
-    pub const fn new(signal_fn: fn()) -> Self {
+    fn new(executor: &'static raw::Executor) -> Self {
         Self {
-            alarm: None,
+            executor,
-            run_queue: RunQueue::new(),
-            timer_queue: TimerQueue::new(),
-            signal_fn: signal_fn,
-            not_send: PhantomData,
-        }
-    }
-    pub const fn new_with_alarm(alarm: &'static dyn Alarm, signal_fn: fn()) -> Self {
-        Self {
-            alarm: Some(alarm),
-            run_queue: RunQueue::new(),
-            timer_queue: TimerQueue::new(),
-            signal_fn: signal_fn,
             not_send: PhantomData,
         }
     }
 
-    unsafe fn enqueue(&self, item: *mut TaskHeader) {
+    pub fn spawn<F>(&self, token: SpawnToken<F>) -> Result<(), SpawnError> {
-        if self.run_queue.enqueue(item) {
+        let task = token.raw_task;
-            (self.signal_fn)()
-        }
-    }
-
-    /// Spawn a future on this executor.
-    pub fn spawn(&'static self, token: SpawnToken) -> Result<(), SpawnError> {
-        let header = token.header;
         mem::forget(token);
 
-        match header {
+        match task {
-            Some(header) => unsafe {
+            Some(task) => {
-                let header = header.as_ref();
+                unsafe { self.executor.spawn(task) };
-                header.executor.set(self);
-                self.enqueue(header as *const _ as _);
                 Ok(())
-            },
+            }
             None => Err(SpawnError::Busy),
         }
     }
 
-    /// Runs the executor until the queue is empty.
+    /// Convert this Spawner to a SendSpawner. This allows you to send the
-    pub fn run(&self) {
+    /// spawner to other threads, but the spawner loses the ability to spawn
-        unsafe {
+    /// non-Send tasks.
-            if self.alarm.is_some() {
+    pub fn make_send(&self) -> SendSpawner {
-                self.timer_queue.dequeue_expired(Instant::now(), |p| {
+        SendSpawner {
-                    let header = &*p;
+            executor: self.executor,
-                    header.enqueue();
+            not_send: PhantomData,
-                });
-            }
-
-            self.run_queue.dequeue_all(|p| {
-                let header = &*p;
-                header.expires_at.set(Instant::MAX);
-
-                let state = header.state.fetch_and(!STATE_RUN_QUEUED, Ordering::AcqRel);
-                if state & STATE_SPAWNED == 0 {
-                    // If task is not running, ignore it. This can happen in the following scenario:
-                    //   - Task gets dequeued, poll starts
-                    //   - While task is being polled, it gets woken. It gets placed in the queue.
-                    //   - Task poll finishes, returning done=true
-                    //   - RUNNING bit is cleared, but the task is already in the queue.
-                    return;
-                }
-
-                // Run the task
-                header.poll_fn.read()(p as _);
-
-                // Enqueue or update into timer_queue
-                self.timer_queue.update(p);
-            });
-
-            // If this is in the past, set_alarm will immediately trigger the alarm,
-            // which will make the wfe immediately return so we do another loop iteration.
-            if let Some(alarm) = self.alarm {
-                let next_expiration = self.timer_queue.next_expiration();
-                alarm.set_callback(self.signal_fn);
-                alarm.set(next_expiration.as_ticks());
-            }
         }
     }
 }
 
-pub(crate) unsafe fn register_timer(at: Instant, waker: &Waker) {
+/// Handle to spawn tasks into an executor from any thread.
-    let p = waker::task_from_waker(waker);
+///
-    let header = &*p;
+/// This Spawner can be used from any thread (it implements Send and Sync, so after  any task (Send and non-Send ones), but it can
-    let expires_at = header.expires_at.get();
+/// only be used in the executor thread (it is not Send itself).
-    header.expires_at.set(min(expires_at, at));
+///
+/// If you want to spawn tasks from another thread, use [SendSpawner].
+pub struct SendSpawner {
+    executor: &'static raw::Executor,
+    not_send: PhantomData<*mut ()>,
 }
 
-pub mod raw {
+unsafe impl Send for SendSpawner {}
-    use super::waker;
+unsafe impl Sync for SendSpawner {}
-    use core::ptr::NonNull;
-    use core::task::Waker;
 
-    pub fn task_from_waker(waker: &Waker) -> NonNull<()> {
+/// Handle to spawn tasks to an executor.
-        unsafe { NonNull::new_unchecked(waker::task_from_waker(waker) as *mut ()) }
+///
+/// This Spawner can spawn any task (Send and non-Send ones), but it can
+/// only be used in the executor thread (it is not Send itself).
+///
+/// If you want to spawn tasks from another thread, use [SendSpawner].
+impl SendSpawner {
+    fn new(executor: &'static raw::Executor) -> Self {
+        Self {
+            executor,
+            not_send: PhantomData,
+        }
     }
 
-    pub unsafe fn wake_task(task: NonNull<()>) {
+    pub fn spawn<F: Send>(&self, token: SpawnToken<F>) -> Result<(), SpawnError> {
-        let header = &*waker::task_from_ptr(task.as_ptr());
+        let header = token.raw_task;
-        header.enqueue();
+        mem::forget(token);
+
+        match header {
+            Some(header) => {
+                unsafe { self.executor.spawn(header) };
+                Ok(())
+            }
+            None => Err(SpawnError::Busy),
+        }
+    }
+}
+
+pub struct Executor {
+    inner: raw::Executor,
+    not_send: PhantomData<*mut ()>,
+}
+
+impl Executor {
+    pub const fn new() -> Self {
+        Self {
+            inner: raw::Executor::new(|_| cortex_m::asm::sev(), ptr::null_mut()),
+            not_send: PhantomData,
+        }
+    }
+
+    pub fn set_alarm(&mut self, alarm: &'static dyn Alarm) {
+        self.inner.set_alarm(alarm);
+    }
+
+    /// Runs the executor.
+    ///
+    /// This function never returns.
+    pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
+        init(Spawner::new(&self.inner));
+
+        loop {
+            unsafe { self.inner.run_queued() };
+            cortex_m::asm::wfe();
+        }
+    }
+}
+
+fn pend_by_number(n: u8) {
+    struct N(u8);
+    unsafe impl cortex_m::interrupt::Nr for N {
+        fn nr(&self) -> u8 {
+            self.0
+        }
+    }
+    cortex_m::peripheral::NVIC::pend(N(n))
+}
+
+pub struct IrqExecutor<I: OwnedInterrupt> {
+    irq: I,
+    inner: raw::Executor,
+    not_send: PhantomData<*mut ()>,
+}
+
+impl<I: OwnedInterrupt> IrqExecutor<I> {
+    pub fn new(irq: I) -> Self {
+        let ctx = irq.number() as *mut ();
+        Self {
+            irq,
+            inner: raw::Executor::new(|ctx| pend_by_number(ctx as u8), ctx),
+            not_send: PhantomData,
+        }
+    }
+
+    pub fn set_alarm(&mut self, alarm: &'static dyn Alarm) {
+        self.inner.set_alarm(alarm);
+    }
+
+    /// Start the executor.
+    ///
+    /// `init` is called in the interrupt context, then the interrupt is
+    /// configured to run the executor.
+    pub fn start(&'static mut self, init: impl FnOnce(Spawner) + Send) {
+        self.irq.disable();
+
+        init(Spawner::new(&self.inner));
+
+        self.irq.set_handler(
+            |ctx| unsafe {
+                let executor = &*(ctx as *const raw::Executor);
+                executor.run_queued();
+            },
+            &self.inner as *const _ as _,
+        );
+        self.irq.enable();
     }
 }

embassy/src/executor/raw.rs

@@ -0,0 +1,154 @@
+use core::cell::Cell;
+use core::cmp::min;
+use core::ptr;
+use core::ptr::NonNull;
+use core::sync::atomic::{AtomicU32, Ordering};
+use core::task::Waker;
+
+use super::run_queue::{RunQueue, RunQueueItem};
+use super::timer_queue::{TimerQueue, TimerQueueItem};
+use super::util::UninitCell;
+use super::waker;
+use crate::time::{Alarm, Instant};
+
+/// Task is spawned (has a future)
+pub(crate) const STATE_SPAWNED: u32 = 1 << 0;
+/// Task is in the executor run queue
+pub(crate) const STATE_RUN_QUEUED: u32 = 1 << 1;
+/// Task is in the executor timer queue
+pub(crate) const STATE_TIMER_QUEUED: u32 = 1 << 2;
+
+pub struct Task {
+    pub(crate) state: AtomicU32,
+    pub(crate) run_queue_item: RunQueueItem,
+    pub(crate) expires_at: Cell<Instant>,
+    pub(crate) timer_queue_item: TimerQueueItem,
+    pub(crate) executor: Cell<*const Executor>, // Valid if state != 0
+    pub(crate) poll_fn: UninitCell<unsafe fn(NonNull<Task>)>, // Valid if STATE_SPAWNED
+}
+
+impl Task {
+    pub(crate) const fn new() -> Self {
+        Self {
+            state: AtomicU32::new(0),
+            expires_at: Cell::new(Instant::from_ticks(0)),
+            run_queue_item: RunQueueItem::new(),
+            timer_queue_item: TimerQueueItem::new(),
+            executor: Cell::new(ptr::null()),
+            poll_fn: UninitCell::uninit(),
+        }
+    }
+
+    pub(crate) unsafe fn enqueue(&self) {
+        let mut current = self.state.load(Ordering::Acquire);
+        loop {
+            // If already scheduled, or if not started,
+            if (current & STATE_RUN_QUEUED != 0) || (current & STATE_SPAWNED == 0) {
+                return;
+            }
+
+            // Mark it as scheduled
+            let new = current | STATE_RUN_QUEUED;
+
+            match self.state.compare_exchange_weak(
+                current,
+                new,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => break,
+                Err(next_current) => current = next_current,
+            }
+        }
+
+        // We have just marked the task as scheduled, so enqueue it.
+        let executor = &*self.executor.get();
+        executor.enqueue(self as *const Task as *mut Task);
+    }
+}
+
+pub(crate) struct Executor {
+    run_queue: RunQueue,
+    timer_queue: TimerQueue,
+    signal_fn: fn(*mut ()),
+    signal_ctx: *mut (),
+    alarm: Option<&'static dyn Alarm>,
+}
+
+impl Executor {
+    pub(crate) const fn new(signal_fn: fn(*mut ()), signal_ctx: *mut ()) -> Self {
+        Self {
+            run_queue: RunQueue::new(),
+            timer_queue: TimerQueue::new(),
+            signal_fn,
+            signal_ctx,
+            alarm: None,
+        }
+    }
+
+    pub(crate) fn set_alarm(&mut self, alarm: &'static dyn Alarm) {
+        self.alarm = Some(alarm);
+    }
+
+    unsafe fn enqueue(&self, item: *mut Task) {
+        if self.run_queue.enqueue(item) {
+            (self.signal_fn)(self.signal_ctx)
+        }
+    }
+
+    pub(crate) unsafe fn spawn(&'static self, task: NonNull<Task>) {
+        let task = task.as_ref();
+        task.executor.set(self);
+        self.enqueue(task as *const _ as _);
+    }
+
+    pub(crate) unsafe fn run_queued(&self) {
+        if self.alarm.is_some() {
+            self.timer_queue.dequeue_expired(Instant::now(), |p| {
+                p.as_ref().enqueue();
+            });
+        }
+
+        self.run_queue.dequeue_all(|p| {
+            let task = p.as_ref();
+            task.expires_at.set(Instant::MAX);
+
+            let state = task.state.fetch_and(!STATE_RUN_QUEUED, Ordering::AcqRel);
+            if state & STATE_SPAWNED == 0 {
+                // If task is not running, ignore it. This can happen in the following scenario:
+                //   - Task gets dequeued, poll starts
+                //   - While task is being polled, it gets woken. It gets placed in the queue.
+                //   - Task poll finishes, returning done=true
+                //   - RUNNING bit is cleared, but the task is already in the queue.
+                return;
+            }
+
+            // Run the task
+            task.poll_fn.read()(p as _);
+
+            // Enqueue or update into timer_queue
+            self.timer_queue.update(p);
+        });
+
+        // If this is in the past, set_alarm will immediately trigger the alarm,
+        // which will make the wfe immediately return so we do another loop iteration.
+        if let Some(alarm) = self.alarm {
+            let next_expiration = self.timer_queue.next_expiration();
+            alarm.set_callback(self.signal_fn, self.signal_ctx);
+            alarm.set(next_expiration.as_ticks());
+        }
+    }
+}
+
+pub use super::waker::task_from_waker;
+
+pub unsafe fn wake_task(task: NonNull<Task>) {
+    task.as_ref().enqueue();
+}
+
+pub(crate) unsafe fn register_timer(at: Instant, waker: &Waker) {
+    let task = waker::task_from_waker(waker);
+    let task = task.as_ref();
+    let expires_at = task.expires_at.get();
+    task.expires_at.set(min(expires_at, at));
+}

embassy/src/executor/run_queue.rs

@@ -1,10 +1,11 @@
 use core::ptr;
+use core::ptr::NonNull;
 use core::sync::atomic::{AtomicPtr, Ordering};
 
-use super::TaskHeader;
+use super::raw::Task;
 
 pub(crate) struct RunQueueItem {
-    next: AtomicPtr<TaskHeader>,
+    next: AtomicPtr<Task>,
 }
 
 impl RunQueueItem {
@@ -27,7 +28,7 @@ impl RunQueueItem {
 /// current batch is completely processed, so even if a task enqueues itself instantly (for example
 /// by waking its own waker) can't prevent other tasks from running.
 pub(crate) struct RunQueue {
-    head: AtomicPtr<TaskHeader>,
+    head: AtomicPtr<Task>,
 }
 
 impl RunQueue {
@@ -38,7 +39,7 @@ impl RunQueue {
     }
 
     /// Enqueues an item. Returns true if the queue was empty.
-    pub(crate) unsafe fn enqueue(&self, item: *mut TaskHeader) -> bool {
+    pub(crate) unsafe fn enqueue(&self, item: *mut Task) -> bool {
         let mut prev = self.head.load(Ordering::Acquire);
         loop {
             (*item).run_queue_item.next.store(prev, Ordering::Relaxed);
@@ -54,7 +55,7 @@ impl RunQueue {
         prev.is_null()
     }
 
-    pub(crate) unsafe fn dequeue_all(&self, on_task: impl Fn(*mut TaskHeader)) {
+    pub(crate) unsafe fn dequeue_all(&self, on_task: impl Fn(NonNull<Task>)) {
         let mut task = self.head.swap(ptr::null_mut(), Ordering::AcqRel);
 
         while !task.is_null() {
@@ -62,7 +63,7 @@ impl RunQueue {
             // Therefore, first read the next pointer, and only then process the task.
             let next = (*task).run_queue_item.next.load(Ordering::Relaxed);
 
-            on_task(task);
+            on_task(NonNull::new_unchecked(task));
 
             task = next
         }

embassy/src/executor/timer.rs

@@ -3,6 +3,7 @@ use core::pin::Pin;
 use core::task::{Context, Poll};
 use futures::Stream;
 
+use super::raw;
 use crate::time::{Duration, Instant};
 
 pub struct Timer {
@@ -34,7 +35,7 @@ impl Future for Timer {
         if self.yielded_once && self.expires_at <= Instant::now() {
             Poll::Ready(())
         } else {
-            unsafe { super::register_timer(self.expires_at, cx.waker()) };
+            unsafe { raw::register_timer(self.expires_at, cx.waker()) };
             self.yielded_once = true;
             Poll::Pending
         }
@@ -66,7 +67,7 @@ impl Stream for Ticker {
             self.expires_at += dur;
             Poll::Ready(Some(()))
         } else {
-            unsafe { super::register_timer(self.expires_at, cx.waker()) };
+            unsafe { raw::register_timer(self.expires_at, cx.waker()) };
             Poll::Pending
         }
     }

embassy/src/executor/timer_queue.rs

@@ -1,13 +1,14 @@
 use core::cell::Cell;
+use core::cmp::min;
+use core::ptr;
+use core::ptr::NonNull;
 use core::sync::atomic::{AtomicPtr, Ordering};
-use core::{cmp::min, ptr};
 
+use super::raw::{Task, STATE_TIMER_QUEUED};
 use crate::time::Instant;
 
-use super::{TaskHeader, STATE_TIMER_QUEUED};
-
 pub(crate) struct TimerQueueItem {
-    next: Cell<*mut TaskHeader>,
+    next: Cell<*mut Task>,
 }
 
 impl TimerQueueItem {
@@ -19,7 +20,7 @@ impl TimerQueueItem {
 }
 
 pub(crate) struct TimerQueue {
-    head: Cell<*mut TaskHeader>,
+    head: Cell<*mut Task>,
 }
 
 impl TimerQueue {
@@ -29,15 +30,15 @@ impl TimerQueue {
         }
     }
 
-    pub(crate) unsafe fn update(&self, p: *mut TaskHeader) {
+    pub(crate) unsafe fn update(&self, p: NonNull<Task>) {
-        let header = &*p;
+        let task = p.as_ref();
-        if header.expires_at.get() != Instant::MAX {
+        if task.expires_at.get() != Instant::MAX {
-            let old_state = header.state.fetch_or(STATE_TIMER_QUEUED, Ordering::AcqRel);
+            let old_state = task.state.fetch_or(STATE_TIMER_QUEUED, Ordering::AcqRel);
             let is_new = old_state & STATE_TIMER_QUEUED == 0;
 
             if is_new {
-                header.timer_queue_item.next.set(self.head.get());
+                task.timer_queue_item.next.set(self.head.get());
-                self.head.set(p);
+                self.head.set(p.as_ptr());
             }
         }
     }
@@ -45,18 +46,18 @@ impl TimerQueue {
     pub(crate) unsafe fn next_expiration(&self) -> Instant {
         let mut res = Instant::MAX;
         self.retain(|p| {
-            let header = &*p;
+            let task = p.as_ref();
-            let expires = header.expires_at.get();
+            let expires = task.expires_at.get();
             res = min(res, expires);
             expires != Instant::MAX
         });
         res
     }
 
-    pub(crate) unsafe fn dequeue_expired(&self, now: Instant, on_task: impl Fn(*mut TaskHeader)) {
+    pub(crate) unsafe fn dequeue_expired(&self, now: Instant, on_task: impl Fn(NonNull<Task>)) {
         self.retain(|p| {
-            let header = &*p;
+            let task = p.as_ref();
-            if header.expires_at.get() <= now {
+            if task.expires_at.get() <= now {
                 on_task(p);
                 false
             } else {
@@ -65,20 +66,18 @@ impl TimerQueue {
         });
     }
 
-    pub(crate) unsafe fn retain(&self, mut f: impl FnMut(*mut TaskHeader) -> bool) {
+    pub(crate) unsafe fn retain(&self, mut f: impl FnMut(NonNull<Task>) -> bool) {
         let mut prev = &self.head;
         while !prev.get().is_null() {
-            let p = prev.get();
+            let p = NonNull::new_unchecked(prev.get());
-            let header = &*p;
+            let task = &*p.as_ptr();
             if f(p) {
                 // Skip to next
-                prev = &header.timer_queue_item.next;
+                prev = &task.timer_queue_item.next;
             } else {
                 // Remove it
-                prev.set(header.timer_queue_item.next.get());
+                prev.set(task.timer_queue_item.next.get());
-                header
+                task.state.fetch_and(!STATE_TIMER_QUEUED, Ordering::AcqRel);
-                    .state
-                    .fetch_and(!STATE_TIMER_QUEUED, Ordering::AcqRel);
             }
         }
     }

embassy/src/executor/waker.rs

@@ -1,7 +1,8 @@
 use core::mem;
+use core::ptr::NonNull;
 use core::task::{RawWaker, RawWakerVTable, Waker};
 
-use super::TaskHeader;
+use super::raw::Task;
 
 const VTABLE: RawWakerVTable = RawWakerVTable::new(clone, wake, wake, drop);
 
@@ -10,26 +11,21 @@ unsafe fn clone(p: *const ()) -> RawWaker {
 }
 
 unsafe fn wake(p: *const ()) {
-    let header = &*task_from_ptr(p);
+    (*(p as *mut Task)).enqueue()
-    header.enqueue();
 }
 
 unsafe fn drop(_: *const ()) {
     // nop
 }
 
-pub(crate) unsafe fn from_task(p: *mut TaskHeader) -> Waker {
+pub(crate) unsafe fn from_task(p: NonNull<Task>) -> Waker {
-    Waker::from_raw(RawWaker::new(p as _, &VTABLE))
+    Waker::from_raw(RawWaker::new(p.as_ptr() as _, &VTABLE))
 }
 
-pub(crate) unsafe fn task_from_ptr(p: *const ()) -> *mut TaskHeader {
+pub unsafe fn task_from_waker(waker: &Waker) -> NonNull<Task> {
-    p as *mut TaskHeader
+    let hack: &WakerHack = mem::transmute(waker);
-}
+    assert_eq!(hack.vtable, &VTABLE);
-
+    NonNull::new_unchecked(hack.data as *mut Task)
-pub(crate) unsafe fn task_from_waker(w: &Waker) -> *mut TaskHeader {
-    let w: &WakerHack = mem::transmute(w);
-    assert_eq!(w.vtable, &VTABLE);
-    task_from_ptr(w.data)
 }
 
 struct WakerHack {

embassy/src/interrupt.rs

@@ -32,6 +32,7 @@ unsafe impl cortex_m::interrupt::Nr for NrWrap {
 pub unsafe trait OwnedInterrupt {
     type Priority: From<u8> + Into<u8> + Copy;
     fn number(&self) -> u8;
+    unsafe fn steal() -> Self;
 
     /// Implementation detail, do not use outside embassy crates.
     #[doc(hidden)]

embassy/src/lib.rs

@@ -2,7 +2,6 @@
 #![feature(generic_associated_types)]
 #![feature(const_fn)]
 #![feature(const_fn_fn_ptr_basics)]
-#![feature(const_in_array_repeat_expressions)]
 #![feature(const_option)]
 
 // This mod MUST go first, so that the others see its macros.

embassy/src/time/traits.rs

@@ -16,7 +16,7 @@ impl<T: Clock + ?Sized> Clock for &T {
 pub trait Alarm {
     /// Sets the callback function to be called when the alarm triggers.
     /// The callback may be called from any context (interrupt or thread mode).
-    fn set_callback(&self, callback: fn());
+    fn set_callback(&self, callback: fn(*mut ()), ctx: *mut ());
 
     /// Sets an alarm at the given timestamp. When the clock reaches that
     /// timestamp, the provided callback funcion will be called.
@@ -32,8 +32,8 @@ pub trait Alarm {
 }
 
 impl<T: Alarm + ?Sized> Alarm for &T {
-    fn set_callback(&self, callback: fn()) {
+    fn set_callback(&self, callback: fn(*mut ()), ctx: *mut ()) {
-        T::set_callback(self, callback);
+        T::set_callback(self, callback, ctx);
     }
     fn set(&self, timestamp: u64) {
         T::set(self, timestamp);

embassy/src/util/signal.rs

@@ -110,7 +110,7 @@ impl<'a, I: OwnedInterrupt> InterruptFuture<'a, I> {
         };
 
         if ctx as *const _ != ptr::null() {
-            executor::raw::wake_task(ptr::NonNull::new_unchecked(ctx));
+            executor::raw::wake_task(ptr::NonNull::new_unchecked(ctx as _));
         }
 
         NVIC::mask(NrWrap(irq));
@@ -124,10 +124,8 @@ impl<'a, I: OwnedInterrupt> Future for InterruptFuture<'a, I> {
 
     fn poll(self: core::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
         let s = unsafe { self.get_unchecked_mut() };
-        s.interrupt.set_handler(
+        let ctx = unsafe { executor::raw::task_from_waker(&cx.waker()).cast().as_ptr() };
-            Self::interrupt_handler,
+        s.interrupt.set_handler(Self::interrupt_handler, ctx);
-            executor::raw::task_from_waker(&cx.waker()).cast().as_ptr(),
-        );
         if s.interrupt.is_enabled() {
             Poll::Pending
         } else {

test-build.sh

@@ -3,7 +3,7 @@
 set -euxo pipefail
 
 # embassy std
-(cd embassy; cargo build --features log,std)
+#(cd embassy; cargo build --features log,std)
 
 # embassy embedded
 (cd embassy; cargo build --target thumbv7em-none-eabi)