zephyr: Add initial implementation of work queue and async support

This provides a first pass an an implementation of management of Zephyr
work queues, and an executor that schedules work using Zephyr's work
queues.

TODO: Clean up how Futures can use the Context to indicate scheduling.
TODO: Move a few things to make the used modules a bit cleaner.

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

Author: d3zd3z

zephyr-sys/build.rs

@@ -92,6 +92,7 @@ fn main() -> Result<()> {
         .allowlist_item("K_.*")
         .allowlist_item("ZR_.*")
         .allowlist_item("LOG_LEVEL_.*")
+        .allowlist_item("k_poll_modes")
         // Deprecated
         .blocklist_function("sys_clock_timeout_end_calc")
         .parse_callbacks(Box::new(bindgen::CargoCallbacks::new()))

zephyr-sys/src/lib.rs

@@ -49,3 +49,5 @@ macro_rules! derive_copy {
 
 derive_copy!(z_spinlock_key);
 derive_clone!(z_spinlock_key);
+derive_copy!(k_timeout_t);
+derive_clone!(k_timeout_t);

zephyr-sys/wrapper.h

@@ -52,3 +52,7 @@ extern int errno;
  */
 const uintptr_t ZR_STACK_ALIGN = Z_KERNEL_STACK_OBJ_ALIGN;
 const uintptr_t ZR_STACK_RESERVED = K_KERNEL_STACK_RESERVED;
+
+const uint32_t ZR_POLL_TYPE_SEM_AVAILABLE = K_POLL_TYPE_SEM_AVAILABLE;
+const uint32_t ZR_POLL_TYPE_SIGNAL = K_POLL_TYPE_SIGNAL;
+const uint32_t ZR_POLL_TYPE_DATA_AVAILABLE = K_POLL_TYPE_DATA_AVAILABLE;

zephyr/Cargo.toml

@@ -20,6 +20,7 @@ cfg-if = "1.0"
 
 # The log create is used if the user desires logging, and calls `set_logger()`.
 log = "0.4.22"
+arrayvec = { version = "0.7.6", default-features = false }
 
 [dependencies.fugit]
 version = "0.3.7"

zephyr/src/kio.rs

@@ -0,0 +1,96 @@
+//! Async IO for Zephyr
+//!
+//! This implements the basics of using Zephyr's work queues to implement async code on Zephyr.
+//!
+//! Most of the work happens in [`work`] and in [`futures`]
+//!
+//! [`work`]: crate::work
+//! [`futures`]: crate::work::futures
+
+use core::ffi::CStr;
+use core::task::Poll;
+use core::{future::Future, pin::Pin};
+
+use crate::time::NoWait;
+use crate::work::futures::WakeInfo;
+use crate::work::{futures::JoinHandle, futures::WorkBuilder, WorkQueue};
+
+pub mod sync;
+
+/// Run an async future on the given worker thread.
+///
+/// Arrange to have the given future run on the given worker thread.  The resulting `JoinHandle` has
+/// `join` and `join_async` methods that can be used to wait for the given thread.
+pub fn spawn<F>(future: F, worker: &WorkQueue, name: &'static CStr) -> JoinHandle<F>
+where
+    F: Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    WorkBuilder::new()
+        .set_worker(worker)
+        .set_name(name)
+        .start(future)
+}
+
+/// Run an async future on the current worker thread.
+///
+/// Arrange to have the given future run on the current worker thread.  The resulting `JoinHandle`
+/// has `join` and `join_async` methods that can be used to wait for the given thread.
+///
+/// The main use for this is to allow work threads to use `Rc` and `Rc<RefCell<T>>` within async
+/// tasks.  The main constraint is that references inside cannot be held across an `.await`.
+///
+/// # Panics
+/// If this is called other than from a worker task running on a work thread, it will panic.
+pub fn spawn_local<F>(future: F, name: &'static CStr) -> JoinHandle<F>
+where
+    F: Future + 'static,
+    F::Output: Send + 'static,
+{
+    WorkBuilder::new()
+        .set_name(name)
+        .start_local(future)
+}
+
+/// Yield the current thread, returning it to the work queue to be run after other work on that
+/// queue.  (This has to be called `yield_now` in Rust, because `yield` is a keyword.
+pub fn yield_now() -> impl Future<Output = ()> {
+    YieldNow { waited: false }
+}
+
+struct YieldNow {
+    waited: bool,
+}
+
+impl Future for YieldNow {
+    type Output = ();
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        cx: &mut core::task::Context<'_>,
+    ) -> core::task::Poll<Self::Output> {
+        if self.waited {
+            Poll::Ready(())
+        } else {
+            // Enqueue outselves with no wait and no events.
+            let info = unsafe { WakeInfo::from_context(cx) };
+
+            // Unsafely check if the work queue running us is empty.  We only check explicitly
+            // specified workers (TODO access the system work queue).  The check is racy, but should
+            // always fail indicating that the queue is not empty when it could be.  Checking this
+            // avoids re-scheduling the only worker back into the queue.
+            // SAFETY: The check is racy, but will fail with us yielding when we didn't need to.
+            if let Some(wq) = info.queue {
+                let wq = unsafe { wq.as_ref() };
+                if wq.pending.head == wq.pending.tail {
+                    return Poll::Ready(());
+                }
+            }
+
+            info.timeout = NoWait.into();
+            self.waited = true;
+
+            Poll::Pending
+        }
+    }
+}

zephyr/src/kio/sync.rs

@@ -0,0 +1,126 @@
+//! Synchronization mechanisms that work with async.
+//!
+//! Notably, Zephyr's `k_mutex` type isn't supported as a type that can be waited for
+//! asynchronously.
+//!
+//! The main problem with `k_mutex` (meaning [`crate::sync::Mutex`]) is that the `lock` operation
+//! can block, and since multiple tasks may be scheduled for the same work queue, the system can
+//! deadlock, as the scheduler may not run to allow the task that actually holds the mutex to run.
+//!
+//! As an initial stopgap. We provide a [`Mutex`] type that is usable within an async context.  We
+//! do not currently implement an associated `Condvar`.
+//!
+//! Note that using Semaphores for locking means that this mechanism doesn't handle priority
+//! inversion issues.  Be careful with workers that run at different priorities.
+
+// Use the same error types from the regular sync version.
+
+use core::{
+    cell::UnsafeCell,
+    fmt,
+    marker::PhantomData,
+    ops::{Deref, DerefMut},
+};
+
+use crate::{
+    sync::{LockResult, TryLockError, TryLockResult},
+    sys::sync::Semaphore,
+    time::{Forever, NoWait},
+};
+
+/// A mutual exclusion primitive useful for protecting shared data.  Async version.
+///
+/// This mutex will block a task waiting for the lock to become available.
+pub struct Mutex<T: ?Sized> {
+    /// The semaphore indicating ownership of the data.  When it is "0" the task that did the 'take'
+    /// on it owns the data, and will use `give` when it is unlocked.  This mechanism works for
+    /// simple Mutex that protects the data without needing a condition variable.
+    inner: Semaphore,
+    data: UnsafeCell<T>,
+}
+
+// SAFETY: The semaphore, with the semantics provided here, provide Send and Sync.
+unsafe impl<T: ?Sized + Send> Send for Mutex<T> {}
+unsafe impl<T: ?Sized + Send> Sync for Mutex<T> {}
+
+impl<T> fmt::Debug for Mutex<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Mutex {:?}", self.inner)
+    }
+}
+
+/// An RAII implementation of a held lock.
+pub struct MutexGuard<'a, T: ?Sized + 'a> {
+    lock: &'a Mutex<T>,
+    // Mark !Send explicitly until support is added to Rust for this.
+    _nosend: PhantomData<UnsafeCell<()>>,
+}
+
+// unsafe impl<T: ?Sized + Sync> Sync for MutexGuard<'_, T> {}
+unsafe impl<T: ?Sized + Sync> Sync for MutexGuard<'_, T> {}
+
+impl<T> Mutex<T> {
+    /// Construct a new Mutex.
+    pub fn new(t: T) -> Mutex<T> {
+        Mutex {
+            inner: Semaphore::new(1, 1).unwrap(),
+            data: UnsafeCell::new(t),
+        }
+    }
+}
+
+impl<T: ?Sized> Mutex<T> {
+    /// Acquire the mutex, blocking the current thread until it is able to do so.
+    ///
+    /// This is a sync version, and calling it from an async task will possibly block the async work
+    /// thread, potentially causing deadlock.
+    pub fn lock(&self) -> LockResult<MutexGuard<'_, T>> {
+        self.inner.take(Forever).unwrap();
+        unsafe { Ok(MutexGuard::new(self)) }
+    }
+
+    /// Aquire the mutex, async version.
+    pub async fn lock_async(&self) -> LockResult<MutexGuard<'_, T>> {
+        self.inner.take_async(Forever).await.unwrap();
+        unsafe { Ok(MutexGuard::new(self)) }
+    }
+
+    /// Attempt to aquire the lock.
+    pub fn try_lock(&self) -> TryLockResult<MutexGuard<'_, T>> {
+        match self.inner.take(NoWait) {
+            Ok(()) => unsafe { Ok(MutexGuard::new(self)) },
+            // TODO: Distinguish timeout from other errors.
+            Err(_) => Err(TryLockError::WouldBlock),
+        }
+    }
+}
+
+impl<'mutex, T: ?Sized> MutexGuard<'mutex, T> {
+    unsafe fn new(lock: &'mutex Mutex<T>) -> MutexGuard<'mutex, T> {
+        MutexGuard {
+            lock,
+            _nosend: PhantomData,
+        }
+    }
+}
+
+impl<T: ?Sized> Deref for MutexGuard<'_, T> {
+    type Target = T;
+
+    fn deref(&self) -> &T {
+        unsafe { &*self.lock.data.get() }
+    }
+}
+
+impl<T: ?Sized> DerefMut for MutexGuard<'_, T> {
+    fn deref_mut(&mut self) -> &mut T {
+        unsafe { &mut *self.lock.data.get() }
+    }
+}
+
+impl<T: ?Sized> Drop for MutexGuard<'_, T> {
+    #[inline]
+    fn drop(&mut self) {
+        self.lock.inner.give();
+    }
+}

zephyr/src/lib.rs

@@ -15,11 +15,16 @@ pub mod device;
 pub mod error;
 pub mod logging;
 pub mod object;
+pub mod simpletls;
 pub mod sync;
 pub mod sys;
 pub mod time;
 #[cfg(CONFIG_RUST_ALLOC)]
 pub mod timer;
+#[cfg(CONFIG_RUST_ALLOC)]
+pub mod work;
+#[cfg(CONFIG_RUST_ALLOC)]
+pub mod kio;
 
 pub use error::{Error, Result};
 
@@ -108,3 +113,11 @@ pub mod _export {
 // If allocation has been requested, provide the allocator.
 #[cfg(CONFIG_RUST_ALLOC)]
 pub mod alloc_impl;
+
+#[cfg(CONFIG_RUST_ALLOC)]
+pub mod task {
+    //! Provides the portable-atomic version of `alloc::task::Wake`, which uses the compatible
+    //! versionm of Arc.
+
+    pub use portable_atomic_util::task::Wake;
+}