zephyr: embassy: Use a semaphore for the executor

Fixes a potential race condition in the suspend/resume sequence.

Signed-off-by: Aaron Olowin <[email protected]>

Author: aolowin

zephyr/src/embassy.rs

@@ -48,10 +48,9 @@
 //!
 //! The following features in the `zephyr` crate configure what is supported:
 //!
-//! - **`executor-zephyr`**: This implements an executor that uses Zephyr's thread primitives
-//!   (`k_thread_suspend` and `k_thread_resume`) to suspend the executor thread when there is no work
-//!   to perform.  This feature is incompatible with either `embassy-thread` or `embassy-interrupt`
-//!   in the `embassy-executor` crate.
+//! - **`executor-zephyr`**: This implements an executor that uses a Zephyr semaphore to suspend the
+//!   executor thread when there is no work to perform. This feature is incompatible with either
+//!   `embassy-thread` or `embassy-interrupt` in the `embassy-executor` crate.
 //! - **`embassy-time-driver`**: This feature causes the `zephyr` crate to provide a time driver to
 //!   Embassy.  This driver uses a single `k_timer` in Zephyr to wake async operations that are
 //!   dependent on time.  This enables the `embassy-time` crate's functionality to be used freely
@@ -67,7 +66,7 @@
 //!   because there are no features to enable this, this functions will still be accessible.  Be
 //!   careful.  You should enable `no-kio` in the zephyr crate to hide these functions.
 //! - This executor does not coordinate with the scheduler on Zephyr, but uses an
-//!   architecture-specific mechanmism when there is no work. On Cortex-M, this is the 'wfe'
+//!   architecture-specific mechanism when there is no work. On Cortex-M, this is the 'wfe'
 //!   instruction, on riscv32, the 'wfi' instruction.  This means that no tasks of lower priority
 //!   will ever run, so this should only be started from the lowest priority task on the system.
 //! - Because the 'idle' thread in Zephyr will never run, some platforms will not enter low power

zephyr/src/embassy/executor.rs

@@ -1,29 +1,24 @@
 //! An embassy executor tailored for Zephyr
 
-use core::{marker::PhantomData, sync::atomic::Ordering};
+use core::marker::PhantomData;
 
+use crate::sys::sync::Semaphore;
+use crate::time::Forever;
 use embassy_executor::{raw, Spawner};
-use zephyr_sys::{k_current_get, k_thread_resume, k_thread_suspend, k_tid_t};
-
-use crate::sync::atomic::AtomicBool;
 
 /// Zephyr-thread based executor.
 pub struct Executor {
     inner: Option<raw::Executor>,
-    id: k_tid_t,
-    pend: AtomicBool,
+    poll_needed: Semaphore,
     not_send: PhantomData<*mut ()>,
 }
 
 impl Executor {
     /// Create a new Executor.
     pub fn new() -> Self {
-        let id = unsafe { k_current_get() };
-
         Self {
             inner: None,
-            pend: AtomicBool::new(false),
-            id,
+            poll_needed: Semaphore::new(0, 1),
             not_send: PhantomData,
         }
     }
@@ -36,17 +31,13 @@ impl Executor {
         init(inner.spawner());
 
         loop {
+            let _ = self.poll_needed.take(Forever);
             unsafe {
                 // The raw executor's poll only runs things that were queued _before_ this poll
                 // itself is actually run. This means, specifically, that if the polled execution
                 // causes this, or other threads to enqueue, this will return without running them.
-                // `__pender` _will_ be called, but it isn't "sticky" like `wfe/sev` are.  To
-                // simulate this, we will use the 'pend' atomic to count
+                // `__pender` _will_ be called, so the next time around the semaphore will be taken.
                 inner.poll();
-                if !self.pend.swap(false, Ordering::SeqCst) {
-                    // printkln!("_suspend");
-                    k_thread_suspend(k_current_get());
-                }
             }
         }
     }
@@ -61,20 +52,7 @@ impl Default for Executor {
 #[export_name = "__pender"]
 fn __pender(context: *mut ()) {
     unsafe {
-        let myself = k_current_get();
-
         let this = context as *const Executor;
-        let other = (*this).id;
-
-        // If the other is a different thread, resume it.
-        if other != myself {
-            // printkln!("_resume");
-            k_thread_resume(other);
-        }
-        // Otherwise, we need to make sure our own next suspend doesn't happen.
-        // We need to also prevent a suspend from happening in the case where the only running
-        // thread causes other work to become pending.  The resume below will do nothing, as we
-        // are just running.
-        (*this).pend.store(true, Ordering::SeqCst);
+        (*this).poll_needed.give();
     }
 }