zephyr: sync: channel: Implement Async operations

Add async operations to the channels, where it makes sense.

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

Author: d3zd3z

zephyr-sys/wrapper.h

@@ -48,3 +48,4 @@ 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/src/sync/channel.rs

@@ -45,12 +45,15 @@ use alloc::boxed::Box;
 use core::cell::UnsafeCell;
 use core::ffi::c_void;
 use core::fmt;
+use core::future::Future;
 use core::marker::PhantomData;
 use core::mem::MaybeUninit;
 use core::pin::Pin;
+use core::task::Poll;
 
 use crate::sys::queue::Queue;
 use crate::time::{Forever, NoWait, Timeout};
+use crate::work::futures::WorkWaker;
 
 mod counter;
 
@@ -205,6 +208,88 @@ impl<T> Sender<T> {
     }
 }
 
+// A little note about the Unpin constraint here.  Because Futures are pinned in Rust Async code,
+// and the future stores the messages, we can only send and receive messages that aren't pinned.
+impl<T: Unpin> Sender<T> {
+    /// Waits for a message to be sent into the channel, but only for a limited time.  Async
+    /// version.
+    ///
+    /// This has the same behavior as [`send_timeout`], but as an Async function.
+    pub fn send_timeout_async<'a>(&'a self, msg: T, timeout: impl Into<Timeout>)
+        -> impl Future<Output = Result<(), SendError<T>>> + 'a
+    {
+        SendFuture {
+            sender: self,
+            msg: Some(msg),
+            timeout: timeout.into(),
+            waited: false,
+        }
+    }
+
+    /// Sends a message over the given channel, waiting if necessary. Async version.
+    pub async fn send_async(&self, msg: T) -> Result<(), SendError<T>> {
+        self.send_timeout_async(msg, Forever).await
+    }
+
+    // Note that there is no async version of `try_send`.
+}
+
+/// The implementation of Future for Sender::send_timeout_async.
+struct SendFuture<'a, T: Unpin> {
+    sender: &'a Sender<T>,
+    msg: Option<T>,
+    timeout: Timeout,
+    waited: bool,
+}
+
+impl<'a, T: Unpin> Future for SendFuture<'a, T> {
+    type Output = Result<(), SendError<T>>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut core::task::Context<'_>) -> core::task::Poll<Self::Output> {
+        /*
+        let this = unsafe {
+            Pin::get_unchecked_mut(self)
+        };
+        */
+        let this = Pin::get_mut(self);
+
+        // Take the message out in preparation to try sending it.  It is a logic error if the unwrap
+        // fails.
+        let msg = this.msg.take().unwrap();
+
+        // Try sending the message, with no timeout.
+        let msg = match this.sender.try_send(msg) {
+            Ok(()) => return Poll::Ready(Ok(())),
+            Err(SendError(msg)) => msg,
+        };
+
+        if this.waited {
+            // We already waited, and no message, so give the messagre back, indiciating a timeout.
+            return Poll::Ready(Err(SendError(msg)));
+        }
+
+        // Send didn't happen, put the message back to have for the next call.
+        this.msg = Some(msg);
+
+        // Otherwise, schedule to wake up on receipt or timeout.
+        let work = unsafe { WorkWaker::from_waker(cx.waker()) };
+        let mut lock = work.lock().unwrap();
+        match &this.sender.flavor {
+            SenderFlavor::Unbounded { .. } => {
+                panic!("Implementation error: unbounded queues should never fail");
+            }
+            SenderFlavor::Bounded(chan) => {
+                unsafe {
+                    lock.add_queue(&chan.free);
+                }
+            }
+        }
+        lock.timeout = this.timeout;
+
+        Poll::Pending
+    }
+}
+
 impl<T> Drop for Sender<T> {
     fn drop(&mut self) {
         match &self.flavor {
@@ -341,6 +426,42 @@ impl<T> Receiver<T> {
     }
 }
 
+// Note that receive doesn't need the Unpin constraint, as we aren't storing any message.
+impl<T> Receiver<T> {
+    /// Waits for a message to be received from the channel, but only for a limited time.
+    /// Async version.
+    ///
+    /// If the channel is empty and not disconnected, this call will block until the receive
+    /// operation can proceed or the operation times out.
+    /// wake up and return an error.
+    pub fn recv_timeout_async<'a>(&'a self, timeout: impl Into<Timeout>)
+        -> impl Future<Output = Result<T, RecvError>> + 'a
+    {
+        RecvFuture {
+            receiver: self,
+            timeout: timeout.into(),
+            waited: false,
+        }
+    }
+
+    /// Blocks the current thread until a message is received or the channel is empty and
+    /// disconnected.  Async version.
+    ///
+    /// If the channel is empty and not disconnected, this call will block until the receive
+    /// operation can proceed.
+    pub async fn recv_async(&self) -> Result<T, RecvError> {
+        self.recv_timeout_async(Forever).await
+    }
+
+    /// Return a reference to the inner queue.
+    fn as_queue(&self) -> &Queue {
+        match &self.flavor {
+            ReceiverFlavor::Unbounded { queue, .. } => queue,
+            ReceiverFlavor::Bounded(chan) => &chan.chan,
+        }
+    }
+}
+
 impl<T> Drop for Receiver<T> {
     fn drop(&mut self) {
         match &self.flavor {
@@ -390,6 +511,38 @@ impl<T> fmt::Debug for Receiver<T> {
     }
 }
 
+struct RecvFuture<'a, T> {
+    receiver: &'a Receiver<T>,
+    timeout: Timeout,
+    waited: bool,
+}
+
+impl<'a, T> Future for RecvFuture<'a, T> {
+    type Output = Result<T, RecvError>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut core::task::Context<'_>) -> Poll<Self::Output> {
+        // Try to receive a message.
+        if let Ok(msg) = self.receiver.try_recv() {
+            return Poll::Ready(Ok(msg));
+        }
+
+        if self.waited {
+            // Wait already happened, so this is a timeout.
+            return Poll::Ready(Err(RecvError));
+        }
+
+        // Otherwise, schedule to wakeup on receipt or timeout.
+        let work = unsafe { WorkWaker::from_waker(cx.waker()) };
+        let mut lock = work.lock().unwrap();
+        unsafe {
+            lock.add_queue(self.receiver.as_queue());
+        }
+        lock.timeout = self.timeout;
+
+        Poll::Pending
+    }
+}
+
 /// The "flavor" of a receiver.  This maps to the type of the channel.
 enum ReceiverFlavor<T> {
     /// An unbounded queue.  Messages were allocated with Box, and will be freed upon receipt.

zephyr/src/sys/queue.rs

@@ -23,7 +23,7 @@ use crate::time::Timeout;
 
 /// A wrapper around a Zephyr `k_queue` object.
 pub struct Queue {
-    item: Fixed<k_queue>,
+    pub(crate) item: Fixed<k_queue>,
 }
 
 unsafe impl Sync for StaticKernelObject<k_queue> { }

zephyr/src/sys/sync/semaphore.rs

@@ -18,7 +18,8 @@ use core::fmt;
 #[cfg(CONFIG_RUST_ALLOC)]
 use core::mem;
 
-use crate::printkln;
+use zephyr_sys::ETIMEDOUT;
+
 use crate::time::NoWait;
 use crate::work::futures::WorkWaker;
 use crate::{
@@ -130,13 +131,14 @@ impl<'a> Future for SemTake<'a> {
     type Output = Result<()>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        // Always check if data is available.
+        if let Ok(()) = self.sem.take(NoWait) {
+            return Poll::Ready(Ok(()));
+        }
+
         if self.ran {
-            // The worker ran, we don't actually care about the reason we were awoken, as we can
-            // just do the take on the semaphore with NoWait, and use that as our result.  This
-            // eliminates the race where we wake due to timeout, but the semaphore is given before
-            // we are actually run.
-            printkln!("Semaphore kpoll woke and we're reading");
-            return Poll::Ready(self.sem.take(NoWait));
+            // If we ran once, and still don't have any data, indicate this as a timeout.
+            return Poll::Ready(Err(crate::Error(ETIMEDOUT)));
 
         }
 
@@ -148,7 +150,7 @@ impl<'a> Future for SemTake<'a> {
             // Future won't outlive the semaphore.
             lock.add_semaphore(self.sem);
         }
-        self.timeout = self.timeout.into();
+        lock.timeout = self.timeout;
         self.ran = true;
 
         Poll::Pending

zephyr/src/work.rs

@@ -321,7 +321,7 @@ impl<'a> Future for SignalWait<'a> {
             // Future can't outlive the Signal.
             lock.add_signal(self.signal);
         }
-        self.timeout = self.timeout.into();
+        lock.timeout = self.timeout;
         self.ran = true;
 
         Poll::Pending

zephyr/src/work/futures.rs

@@ -16,9 +16,9 @@ extern crate alloc;
 use core::{cell::UnsafeCell, ffi::{c_int, c_void}, future::Future, mem, pin::Pin, ptr::{self, NonNull}, task::{Context, Poll, Waker}};
 
 use arrayvec::ArrayVec;
-use zephyr_sys::{k_poll_event, k_poll_event_init, k_poll_modes_K_POLL_MODE_NOTIFY_ONLY, k_work, k_work_poll, k_work_poll_init, k_work_poll_submit, k_work_poll_submit_to_queue, k_work_q, ZR_POLL_TYPE_SEM_AVAILABLE, ZR_POLL_TYPE_SIGNAL};
+use zephyr_sys::{k_poll_event, k_poll_event_init, k_poll_modes_K_POLL_MODE_NOTIFY_ONLY, k_work, k_work_poll, k_work_poll_init, k_work_poll_submit, k_work_poll_submit_to_queue, k_work_q, ZR_POLL_TYPE_DATA_AVAILABLE, ZR_POLL_TYPE_SEM_AVAILABLE, ZR_POLL_TYPE_SIGNAL};
 
-use crate::{printkln, sync::{Arc, Mutex}, sys::sync::Semaphore, task::Wake, time::{Duration, Forever, NoWait, Tick, Timeout}};
+use crate::{printkln, sync::{Arc, Mutex}, sys::{queue::Queue, sync::Semaphore}, task::Wake, time::{Duration, Forever, NoWait, Tick, Timeout}};
 
 use super::{Signal, SubmitResult, WorkQueue};
 
@@ -385,6 +385,22 @@ impl WorkWaker {
             );
         }
     }
+
+    /// Add an event that represents waiting for a queue to have a message.
+    pub unsafe fn add_queue<'a>(&'a mut self, queue: &'a Queue) {
+        // SAFETY: Fill with zeroed memory, initializatuon happens in the init function next.
+        self.events.push(unsafe { mem::zeroed() });
+        let ev = self.events.last().unwrap();
+
+        unsafe {
+            k_poll_event_init(
+                ev.get(),
+                ZR_POLL_TYPE_DATA_AVAILABLE,
+                k_poll_modes_K_POLL_MODE_NOTIFY_ONLY as i32,
+                queue.item.get() as *mut c_void,
+            );
+        }
+    }
 }
 
 /// To avoid having to parameterize everything, we limit the size of the ArrayVec of events to