esp_rtos/embassy/
mod.rs

1//! OS-aware embassy executors.
2
3use core::{cell::UnsafeCell, mem::MaybeUninit, ptr::NonNull, sync::atomic::Ordering};
4
5use embassy_executor::{SendSpawner, Spawner, raw};
6use esp_hal::{
7    interrupt::{
8        InterruptHandler,
9        Priority,
10        software::{Instance, SoftwareInterrupt},
11    },
12    peripherals::{FROM_CPU_INTR0, FROM_CPU_INTR1, FROM_CPU_INTR2, FROM_CPU_INTR3},
13    system::Cpu,
14    time::{Duration, Instant},
15};
16use macros::ram;
17use portable_atomic::AtomicPtr;
18
19use crate::{
20    SCHEDULER,
21    scheduler::SchedulerState,
22    task::{TaskPtr, read_thread_pointer},
23};
24
25/// A zero-overhead lock that allows mutable access to the contained value through the scheduler.
26struct SchedulerLocked<T> {
27    inner: UnsafeCell<T>,
28}
29
30unsafe impl<T: Send> Sync for SchedulerLocked<T> {}
31unsafe impl<T: Send> Send for SchedulerLocked<T> {}
32
33impl<T> SchedulerLocked<T> {
34    fn new(inner: T) -> Self {
35        Self {
36            inner: UnsafeCell::new(inner),
37        }
38    }
39
40    fn with<'s>(&'s self, _scheduler: &'s mut SchedulerState) -> &'s mut T {
41        // Safety: The `_scheduler` parameter proves the caller holds the scheduler lock,
42        // so exclusive access to the contained value is safe.
43        unsafe { &mut *self.inner.get() }
44    }
45}
46
47pub(crate) struct FlagsInner {
48    owner: TaskPtr,
49    waiting: Option<TaskPtr>,
50    set: bool,
51}
52impl FlagsInner {
53    fn take(&mut self) -> bool {
54        if self.set {
55            // The flag was set while we weren't looking.
56            self.set = false;
57            true
58        } else {
59            // `waiting` signals that the owner should be resumed when the flag is set. Copying
60            // the task pointer is an optimization that allows clearing the
61            // waiting state without computing the address of a separate field.
62            self.waiting = Some(self.owner);
63
64            false
65        }
66    }
67}
68
69/// A single event bit, optimized for the thread-mode embassy executor.
70///
71/// This takes shortcuts, which make it unsuitable for general purpose use (such as no wait
72/// queue, no timeout, assumes a single thread waits for the flag, there is only a single bit of
73/// flag information).
74struct ThreadFlag {
75    inner: SchedulerLocked<FlagsInner>,
76}
77
78impl ThreadFlag {
79    fn new() -> Self {
80        let owner = SCHEDULER.with(|scheduler| {
81            if let Some(current_task) = NonNull::new(read_thread_pointer()) {
82                current_task
83            } else {
84                // We're cheating, the task hasn't been initialized yet.
85                let current_cpu = Cpu::current();
86                NonNull::from(&scheduler.per_cpu[current_cpu as usize].main_task)
87            }
88        });
89        Self {
90            inner: SchedulerLocked::new(FlagsInner {
91                owner,
92                waiting: None,
93                set: false,
94            }),
95        }
96    }
97
98    fn with<R>(&self, scheduler: &mut SchedulerState, f: impl FnOnce(&mut FlagsInner) -> R) -> R {
99        let inner = self.inner.with(scheduler);
100        f(inner)
101    }
102
103    fn set(&self) {
104        SCHEDULER.with(|scheduler| {
105            let to_resume = self.with(scheduler, |inner| {
106                let to_resume = inner.waiting.take();
107
108                if to_resume.is_none() {
109                    // The task isn't waiting, set the flag.
110                    inner.set = true;
111                }
112
113                to_resume
114            });
115
116            if let Some(waiting) = to_resume {
117                // The task is waiting, there is no need to set the flag - resuming the thread
118                // is all the signal we need.
119                scheduler.resume_task(waiting);
120            }
121        });
122    }
123
124    fn get(&self) -> bool {
125        SCHEDULER.with(|scheduler| self.with(scheduler, |inner| inner.set))
126    }
127
128    fn wait(&self) {
129        // SCHEDULER.sleep_until, but we know the current task's ID, and we know there
130        // is no timeout.
131        SCHEDULER.with(|scheduler| {
132            let owner_to_suspend = self.with(scheduler, |inner| {
133                if !inner.take() {
134                    Some(inner.owner)
135                } else {
136                    None
137                }
138            });
139
140            if let Some(owner) = owner_to_suspend {
141                scheduler.sleep_task_until(owner, Instant::EPOCH + Duration::MAX);
142                crate::task::yield_task();
143            }
144        });
145    }
146}
147
148#[unsafe(export_name = "__pender")]
149#[ram]
150fn __pender(context: *mut ()) {
151    match context as usize {
152        0 => SoftwareInterrupt::new(unsafe { FROM_CPU_INTR0::steal() }).raise(),
153        1 => SoftwareInterrupt::new(unsafe { FROM_CPU_INTR1::steal() }).raise(),
154        2 => SoftwareInterrupt::new(unsafe { FROM_CPU_INTR2::steal() }).raise(),
155        3 => SoftwareInterrupt::new(unsafe { FROM_CPU_INTR3::steal() }).raise(),
156        _ => {
157            // This forces us to keep the embassy timer queue separate, otherwise we'd need to
158            // reentrantly lock SCHEDULER.
159            let flags = unwrap!(unsafe { context.cast::<ThreadFlag>().as_ref() });
160            flags.set();
161        }
162    }
163}
164
165/// Callbacks to run code before/after polling the task queue.
166pub trait Callbacks {
167    /// Called just before polling the executor.
168    fn before_poll(&mut self);
169
170    /// Called after the executor is polled, if there is no work scheduled.
171    ///
172    /// Note that tasks can become ready at any point during the execution
173    /// of this function.
174    fn on_idle(&mut self);
175}
176
177/// Thread-mode executor.
178///
179/// This executor runs in an OS thread, meaning the scheduler needs to be started before using any
180/// async operations. If you wish to write async code without the scheduler running, consider
181/// using the [`InterruptExecutor`].
182#[cfg_attr(
183    multi_core,
184    doc = r"
185
186If you want to start the executor on the second core, you will need to start the second core using
187[`crate::start_second_core`], or [`crate::start_on_second_core_only`] if the first core should stay
188bare-metal.
189If you are looking for a way to run code on the second core without the scheduler, use the [`InterruptExecutor`].
190"
191)]
192pub struct Executor {
193    executor: UnsafeCell<MaybeUninit<raw::Executor>>,
194}
195
196impl Executor {
197    /// Create a new thread-mode executor.
198    pub const fn new() -> Self {
199        Self {
200            executor: UnsafeCell::new(MaybeUninit::uninit()),
201        }
202    }
203
204    /// Run the executor.
205    ///
206    /// The `init` closure is called with a [`Spawner`] that spawns tasks on
207    /// this executor. Use it to spawn the initial task(s). After `init`
208    /// returns, the executor starts running the tasks.
209    ///
210    /// To spawn more tasks later, you may keep copies of the [`Spawner`] (it is
211    /// `Copy`), for example by passing it as an argument to the initial
212    /// tasks.
213    ///
214    /// This function requires `&'static mut self`. This means you have to store
215    /// the Executor instance in a place where it'll live forever and grants
216    /// you mutable access. There's a few ways to do this:
217    ///
218    /// - a [StaticCell](https://docs.rs/static_cell/latest/static_cell/) (safe)
219    /// - a `static mut` (unsafe, not recommended)
220    /// - a local variable in a function you know never returns (like `fn main() -> !`), upgrading
221    ///   its lifetime with `transmute`. (unsafe)
222    ///
223    /// This function never returns.
224    pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
225        let flags = ThreadFlag::new();
226        struct NoHooks;
227
228        impl Callbacks for NoHooks {
229            fn before_poll(&mut self) {}
230
231            fn on_idle(&mut self) {}
232        }
233
234        self.run_inner(init, &flags, NoHooks)
235    }
236
237    /// Run the executor with callbacks.
238    ///
239    /// See [Callbacks] on when the callbacks are called.
240    ///
241    /// See [Self::run] for more information about running the executor.
242    ///
243    /// This function never returns.
244    pub fn run_with_callbacks(
245        &'static mut self,
246        init: impl FnOnce(Spawner),
247        callbacks: impl Callbacks,
248    ) -> ! {
249        let flags = ThreadFlag::new();
250        struct Hooks<'a, CB: Callbacks>(CB, &'a ThreadFlag);
251
252        impl<CB: Callbacks> Callbacks for Hooks<'_, CB> {
253            fn before_poll(&mut self) {
254                self.0.before_poll()
255            }
256
257            fn on_idle(&mut self) {
258                // Make sure we only call on_idle if the executor would otherwise go to sleep.
259                if !self.1.get() {
260                    self.0.on_idle();
261                }
262            }
263        }
264
265        self.run_inner(init, &flags, Hooks(callbacks, &flags))
266    }
267
268    fn run_inner(
269        &'static self,
270        init: impl FnOnce(Spawner),
271        flags: &ThreadFlag,
272        mut hooks: impl Callbacks,
273    ) -> ! {
274        let executor = unsafe {
275            (&mut *self.executor.get()).write(raw::Executor::new(
276                (flags as *const ThreadFlag).cast::<()>().cast_mut(),
277            ))
278        };
279
280        // The main task may start the scheduler from inside this executor, so we cannot require the
281        // scheduler to run already. We can, however, refuse to run on a CPU that the scheduler
282        // never runs on.
283        #[cfg(multi_core)]
284        if crate::SCHEDULER.with(|scheduler| !scheduler.active_cores.contains(Cpu::current())) {
285            panic!("Executor cannot be started: the scheduler is not running on the current CPU.");
286        }
287
288        init(executor.spawner());
289
290        loop {
291            hooks.before_poll();
292
293            unsafe { executor.poll() };
294
295            hooks.on_idle();
296
297            // Wait for work to become available.
298            flags.wait();
299        }
300    }
301}
302
303impl Default for Executor {
304    fn default() -> Self {
305        Self::new()
306    }
307}
308
309/// Interrupt mode executor.
310///
311/// This executor runs tasks in interrupt mode. The interrupt handler is set up
312/// to poll tasks, and when a task is woken the interrupt is pended from
313/// software.
314///
315/// Interrupt executors have potentially lower latency than thread-mode executors, but only a
316/// limited number can be created.
317pub struct InterruptExecutor<const SWI: u8> {
318    executor: UnsafeCell<MaybeUninit<raw::Executor>>,
319    interrupt: SoftwareInterrupt<'static, SWI>,
320}
321
322const COUNT: usize = 4;
323static INTERRUPT_EXECUTORS: [InterruptExecutorStorage; COUNT] =
324    [const { InterruptExecutorStorage::new() }; COUNT];
325
326unsafe impl<const SWI: u8> Send for InterruptExecutor<SWI> {}
327unsafe impl<const SWI: u8> Sync for InterruptExecutor<SWI> {}
328
329struct InterruptExecutorStorage {
330    raw_executor: AtomicPtr<raw::Executor>,
331}
332
333impl InterruptExecutorStorage {
334    const fn new() -> Self {
335        Self {
336            raw_executor: AtomicPtr::new(core::ptr::null_mut()),
337        }
338    }
339
340    /// # Safety:
341    ///
342    /// The caller must ensure `set` has been called before.
343    #[inline(always)]
344    unsafe fn get(&self) -> &raw::Executor {
345        unsafe { &*self.raw_executor.load(Ordering::Relaxed) }
346    }
347
348    fn set(&self, executor: *mut raw::Executor) {
349        self.raw_executor.store(executor, Ordering::Relaxed);
350    }
351}
352
353extern "C" fn handle_interrupt<const NUM: u8>() {
354    match NUM {
355        0 => SoftwareInterrupt::new(unsafe { FROM_CPU_INTR0::steal() }).reset(),
356        1 => SoftwareInterrupt::new(unsafe { FROM_CPU_INTR1::steal() }).reset(),
357        2 => SoftwareInterrupt::new(unsafe { FROM_CPU_INTR2::steal() }).reset(),
358        3 => SoftwareInterrupt::new(unsafe { FROM_CPU_INTR3::steal() }).reset(),
359        _ => unreachable!(),
360    };
361
362    unsafe {
363        // SAFETY: The executor is always initialized before the interrupt is enabled.
364        let executor = INTERRUPT_EXECUTORS[NUM as usize].get();
365        executor.poll();
366    }
367}
368
369impl<const SWI: u8> InterruptExecutor<SWI> {
370    /// Create a new `InterruptExecutor`.
371    /// This takes the software interrupt to be used internally.
372    #[inline]
373    pub const fn new(interrupt: impl Instance<SWI> + 'static) -> Self {
374        Self {
375            executor: UnsafeCell::new(MaybeUninit::uninit()),
376            interrupt: SoftwareInterrupt::new(interrupt),
377        }
378    }
379
380    /// Start the executor at the given priority level.
381    ///
382    /// This initializes the executor, enables the interrupt, and returns.
383    /// The executor keeps running in the background through the interrupt.
384    ///
385    /// This returns a [`SendSpawner`] you can use to spawn tasks on it. A
386    /// [`SendSpawner`] is returned instead of a [`Spawner`] because the
387    /// executor effectively runs in a different "thread" (the interrupt),
388    /// so spawning tasks on it is effectively sending them.
389    ///
390    /// To obtain a [`Spawner`] for this executor, use [`Spawner::for_current_executor`]
391    /// from a task running in it.
392    pub fn start(&'static mut self, priority: Priority) -> SendSpawner {
393        unsafe {
394            (*self.executor.get()).write(raw::Executor::new((SWI as usize) as *mut ()));
395
396            INTERRUPT_EXECUTORS[SWI as usize].set((*self.executor.get()).as_mut_ptr());
397        }
398
399        let swi_handler = match SWI {
400            0 => handle_interrupt::<0>,
401            1 => handle_interrupt::<1>,
402            2 => handle_interrupt::<2>,
403            3 => handle_interrupt::<3>,
404            _ => unreachable!(),
405        };
406
407        self.interrupt
408            .set_interrupt_handler(InterruptHandler::new(swi_handler, priority));
409
410        let executor = unsafe { (*self.executor.get()).assume_init_ref() };
411        executor.spawner().make_send()
412    }
413
414    /// Get a SendSpawner for this executor
415    ///
416    /// This returns a [`SendSpawner`] you can use to spawn tasks on this
417    /// executor.
418    ///
419    /// This MUST only be called on an executor that has already been started.
420    /// The function will panic otherwise.
421    pub fn spawner(&'static self) -> SendSpawner {
422        if INTERRUPT_EXECUTORS[SWI as usize]
423            .raw_executor
424            .load(Ordering::Acquire)
425            .is_null()
426        {
427            panic!("InterruptExecutor::spawner() called on uninitialized executor.");
428        }
429        let executor = unsafe { (*self.executor.get()).assume_init_ref() };
430        executor.spawner().make_send()
431    }
432}
esp_rtos/embassy/mod.rs

esp_rtos/embassy/
mod.rs
