zephyr: Add sync::channel support

Provide an implementation of channels, inspired by crossbeam-channel.
Currently, unbounded channels that use allocation are implemented.

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

Author: d3zd3z

zephyr/src/sync.rs

@@ -15,6 +15,8 @@ use core::{
 use crate::time::Forever;
 use crate::sys::sync as sys;
 
+pub mod channel;
+
 pub mod atomic {
     //! Re-export portable atomic.
     //!

zephyr/src/sync/channel.rs

@@ -0,0 +1,183 @@
+//! Close-to-Zephyr channels
+//!
+//! This module attempts to provide a mechanism as close as possible to `crossbeam-channel` as we
+//! can get, directly using Zephyr primitives.
+//!
+//! The channels are built around `k_queue` in Zephyr.  As is the case with most Zephyr types,
+//! these are typically statically allocated.  Similar to the other close-to-zephyr primitives,
+//! this means that there is a constructor that can directly take one of these primitives.
+//!
+//! In other words, `zephyr::sys::Queue` is a Rust friendly implementation of `k_queue` in Zephyr.
+//! This module provides `Sender` and `Receiver`, which can be cloned and behave as if they had an
+//! internal `Arc` inside them, but without the overhead of an actual Arc.
+
+extern crate alloc;
+
+use alloc::boxed::Box;
+
+use core::ffi::c_void;
+use core::fmt;
+use core::marker::PhantomData;
+
+use crate::sys::queue::Queue;
+
+mod counter;
+
+// The zephyr queue does not allocate or manage the data of the messages, so we need to handle
+// allocation as such as well.  However, we don't need to manage anything, so it is sufficient to
+// simply Box the message, leak it out of the box, and give it to Zephyr, and then on receipt, wrap
+// it back into a Box, and give it to the recipient.
+
+/// Create a multi-producer multi-consumer channel of unbounded capacity.  The messages are
+/// allocated individually as "Box", and the queue is managed by the underlying Zephyr queue.
+pub fn unbounded_from<T>(queue: Queue) -> (Sender<T>, Receiver<T>) {
+    let (s, r) = counter::new(queue);
+    let s = Sender {
+        queue: s,
+        _phantom: PhantomData,
+    };
+    let r = Receiver {
+        queue: r,
+        _phantom: PhantomData,
+    };
+    (s, r)
+}
+
+#[repr(C)]
+struct Message<T> {
+    /// The private data used by the kernel to enqueue messages and such.
+    _private: usize,
+    /// The actual data being transported.
+    data: T,
+}
+
+impl<T> Message<T> {
+    fn new(data: T) -> Message<T> {
+        Message {
+            _private: 0,
+            data,
+        }
+    }
+}
+
+/// The sending side of a channel.
+pub struct Sender<T> {
+    queue: counter::Sender<Queue>,
+    _phantom: PhantomData<T>,
+}
+
+unsafe impl<T: Send> Send for Sender<T> {}
+unsafe impl<T: Send> Sync for Sender<T> {}
+
+impl<T> Sender<T> {
+    /// Sends a message over the given channel. This will perform an alloc of the message, which
+    /// will have an accompanied free on the recipient side.
+    pub fn send(&self, msg: T) -> Result<(), SendError<T>> {
+        let msg = Box::new(Message::new(msg));
+        let msg = Box::into_raw(msg);
+        unsafe {
+            self.queue.send(msg as *mut c_void);
+        }
+        Ok(())
+    }
+}
+
+impl<T> Drop for Sender<T> {
+    fn drop(&mut self) {
+        unsafe {
+            self.queue.release(|_| {
+                crate::printkln!("Release");
+                true
+            })
+        }
+    }
+}
+
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Self {
+        Sender {
+            queue: self.queue.acquire(),
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<T: fmt::Debug> fmt::Debug for Sender<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Sender {:?}", *self.queue)
+    }
+}
+
+/// The receiving side of a channel.
+pub struct Receiver<T> {
+    queue: counter::Receiver<Queue>,
+    _phantom: PhantomData<T>,
+}
+
+unsafe impl<T: Send> Send for Receiver<T> {}
+unsafe impl<T: Send> Sync for Receiver<T> {}
+
+impl<T> Receiver<T> {
+    /// Blocks the current thread until a message is received or the channel is empty and
+    /// disconnected.
+    ///
+    /// If the channel is empty and not disconnected, this call will block until the receive
+    /// operation can proceed.  If the channel is empty and becomes disconnected, this call will
+    /// wake up and return an error.
+    pub fn recv(&self) -> Result<T, RecvError> {
+        let msg = unsafe {
+            self.queue.recv()
+        };
+        let msg = msg as *mut Message<T>;
+        let msg = unsafe { Box::from_raw(msg) };
+        Ok(msg.data)
+    }
+}
+
+impl<T> Drop for Receiver<T> {
+    fn drop(&mut self) {
+        unsafe {
+            self.queue.release(|_| {
+                crate::printkln!("Release");
+                true
+            })
+        }
+    }
+}
+
+impl<T> Clone for Receiver<T> {
+    fn clone(&self) -> Self {
+        Receiver {
+            queue: self.queue.acquire(),
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<T> fmt::Debug for Receiver<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Sender {:?}", *self.queue)
+    }
+}
+
+// TODO: Move to err
+
+/// An error returned from the [`send`] method.
+///
+/// The message could not be sent because the channel is disconnected.
+///
+/// The error contains the message so it can be recovered.
+#[derive(PartialEq, Eq, Clone, Copy)]
+pub struct SendError<T>(pub T);
+
+impl<T> fmt::Debug for SendError<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        "SendError(..)".fmt(f)
+    }
+}
+
+/// An error returned from the [`recv`] method.
+///
+/// A message could not be received because the channel is empty and disconnected.
+#[derive(PartialEq, Eq, Clone, Copy, Debug)]
+pub struct RecvError;

zephyr/src/sync/channel/counter.rs

@@ -0,0 +1,151 @@
+// SPDX-License-Identifier: Apache-2.0
+//! Reference counter for channels.
+
+// This file is taken from crossbeam-channels, with modifications to be nostd.
+
+
+extern crate alloc;
+
+use alloc::boxed::Box;
+use core::ops;
+use core::ptr::NonNull;
+use crate::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+
+/// Reference counter internals.
+struct Counter<C> {
+    /// The number of senders associated with the channel.
+    senders: AtomicUsize,
+
+    /// The number of receivers associated with the channel.
+    receivers: AtomicUsize,
+
+    /// Set to `true` if the last sender or the last receiver reference deallocates the channel.
+    destroy: AtomicBool,
+
+    /// The internal channel.
+    chan: C,
+}
+
+/// Wraps a channel into the reference counter.
+pub(crate) fn new<C>(chan: C) -> (Sender<C>, Receiver<C>) {
+    let counter = NonNull::from(Box::leak(Box::new(Counter {
+        senders: AtomicUsize::new(1),
+        receivers: AtomicUsize::new(1),
+        destroy: AtomicBool::new(false),
+        chan,
+    })));
+    let s = Sender { counter };
+    let r = Receiver { counter };
+    (s, r)
+}
+
+/// The sending side.
+pub(crate) struct Sender<C> {
+    counter: NonNull<Counter<C>>,
+}
+
+impl<C> Sender<C> {
+    /// Returns the internal `Counter`.
+    fn counter(&self) -> &Counter<C> {
+        unsafe { self.counter.as_ref() }
+    }
+
+    /// Acquires another sender reference.
+    pub(crate) fn acquire(&self) -> Self {
+        let count = self.counter().senders.fetch_add(1, Ordering::Relaxed);
+
+        // Cloning senders and calling `mem::forget` on the clones could potentially overflow the
+        // counter. It's very difficult to recover sensibly from such degenerate scenarios so we
+        // just abort when the count becomes very large.
+        if count > isize::MAX as usize {
+            // TODO: We need some kind of equivalent here.
+            unimplemented!();
+        }
+
+        Self {
+            counter: self.counter,
+        }
+    }
+
+    /// Releases the sender reference.
+    ///
+    /// Function `disconnect` will be called if this is the last sender reference.
+    pub(crate) unsafe fn release<F: FnOnce(&C) -> bool>(&self, disconnect: F) {
+        if self.counter().senders.fetch_sub(1, Ordering::AcqRel) == 1 {
+            disconnect(&self.counter().chan);
+
+            if self.counter().destroy.swap(true, Ordering::AcqRel) {
+                drop(unsafe { Box::from_raw(self.counter.as_ptr()) });
+            }
+        }
+    }
+}
+
+impl<C> ops::Deref for Sender<C> {
+    type Target = C;
+
+    fn deref(&self) -> &C {
+        &self.counter().chan
+    }
+}
+
+impl<C> PartialEq for Sender<C> {
+    fn eq(&self, other: &Self) -> bool {
+        self.counter == other.counter
+    }
+}
+
+/// The receiving side.
+pub(crate) struct Receiver<C> {
+    counter: NonNull<Counter<C>>,
+}
+
+impl<C> Receiver<C> {
+    /// Returns the internal `Counter`.
+    fn counter(&self) -> &Counter<C> {
+        unsafe { self.counter.as_ref() }
+    }
+
+    /// Acquires another receiver reference.
+    pub(crate) fn acquire(&self) -> Self {
+        let count = self.counter().receivers.fetch_add(1, Ordering::Relaxed);
+
+        // Cloning receivers and calling `mem::forget` on the clones could potentially overflow the
+        // counter. It's very difficult to recover sensibly from such degenerate scenarios so we
+        // just abort when the count becomes very large.
+        if count > isize::MAX as usize {
+            unimplemented!();
+        }
+
+        Self {
+            counter: self.counter,
+        }
+    }
+
+    /// Releases the receiver reference.
+    ///
+    /// Function `disconnect` will be called if this is the last receiver reference.
+    pub(crate) unsafe fn release<F: FnOnce(&C) -> bool>(&self, disconnect: F) {
+        if self.counter().receivers.fetch_sub(1, Ordering::AcqRel) == 1 {
+            disconnect(&self.counter().chan);
+
+            if self.counter().destroy.swap(true, Ordering::AcqRel) {
+                drop(unsafe { Box::from_raw(self.counter.as_ptr()) });
+            }
+        }
+    }
+}
+
+impl<C> ops::Deref for Receiver<C> {
+    type Target = C;
+
+    fn deref(&self) -> &C {
+        &self.counter().chan
+    }
+}
+
+impl<C> PartialEq for Receiver<C> {
+    fn eq(&self, other: &Self) -> bool {
+        self.counter == other.counter
+    }
+}