zephyr: Implement executor-zephyr to use with Embassy

This implements an executor for Embassy that runs a set of async tasks
on a single Zephyr thread.  The executor will suspend the thread when
there is no work to do, resuming when there is work.

It is permissible to run multiple executors, on different threads, to
support a hybrid async cooperative and priority-based scheduling.

Signed-off-by: David Brown <[email protected]>

Author: d3zd3z

zephyr/Cargo.toml

@@ -59,6 +59,10 @@ optional = true
 version = "0.6.2"
 optional = true
 
+[dependencies.embassy-executor]
+version = "0.7.0"
+optional = true
+
 # These are needed at build time.
 # Whether these need to be vendored is an open question.  They are not
 # used by the core Zephyr tree, but are needed by zephyr applications.
@@ -72,3 +76,8 @@ time-driver = [
   "dep:embassy-time-queue-utils",
   "dep:embassy-sync",
 ]
+
+# Provide an embassy-based executor that runs on individual Zephyr threads.
+executor-zephyr = [
+  "dep:embassy-executor",
+]

zephyr/src/embassy.rs

@@ -86,3 +86,8 @@
 
 #[cfg(feature = "time-driver")]
 mod time_driver;
+
+#[cfg(feature = "executor-zephyr")]
+pub use executor::Executor;
+#[cfg(feature = "executor-zephyr")]
+mod executor;

zephyr/src/embassy/executor.rs

@@ -0,0 +1,74 @@
+//! An embassy executor tailored for Zephyr
+
+use core::{marker::PhantomData, sync::atomic::Ordering};
+
+use embassy_executor::{raw, Spawner};
+use zephyr_sys::{k_current_get, k_thread_resume, k_thread_suspend, k_tid_t};
+
+use crate::{printkln, sync::atomic::AtomicBool};
+
+/// Zephyr-thread based executor.
+pub struct Executor {
+    inner: Option<raw::Executor>,
+    id: k_tid_t,
+    pend: AtomicBool,
+    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,
+            not_send: PhantomData,
+        }
+    }
+
+    /// Run the executor.
+    pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
+        let context = self as *mut _ as *mut ();
+        self.inner.replace(raw::Executor::new(context));
+        let inner = self.inner.as_mut().unwrap();
+        init(inner.spawner());
+
+        loop {
+            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 
+                inner.poll();
+                if !self.pend.swap(false, Ordering::SeqCst) {
+                    // printkln!("_suspend");
+                    k_thread_suspend(k_current_get());
+                }
+            }
+        }
+    }
+}
+
+#[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);
+    }
+}