feat: add display UI with splash screen, dashboard, and RSSI sparkline

Implement full SSD1306 OLED display support:
- Splash screen on boot (1.5s) with project name, version, board
- Dashboard showing radio state, config, counters, RSSI/SNR
- RSSI history sparkline (64-sample ring buffer, 2px-wide bars)
- DisplayOn/DisplayOff host commands routed via dedicated channel
- Pure render functions against generic DrawTarget (no hardware coupling)
- Display on/off control bypasses radio task entirely

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
swaitsandClaude Opus 4.6 committed 2026-03-29 20:42:54 -06:00
1 parent 7064110293
commit 3102fc747a
17624 files changed
+4564928 -38

No files matched your search

+189
View File
@@ -0,0 +1,189 @@
//! Blocking mutex.
//!
//! This module provides a blocking mutex that can be used to synchronize data.
pub mod raw;
use core::cell::UnsafeCell;
use self::raw::RawMutex;
/// Blocking mutex (not async)
///
/// Provides a blocking mutual exclusion primitive backed by an implementation of [`raw::RawMutex`].
///
/// Which implementation you select depends on the context in which you're using the mutex, and you can choose which kind
/// of interior mutability fits your use case.
///
/// Use [`CriticalSectionMutex`] when data can be shared between threads and interrupts.
///
/// Use [`NoopMutex`] when data is only shared between tasks running on the same executor.
///
/// Use [`ThreadModeMutex`] when data is shared between tasks running on the same executor but you want a global singleton.
///
/// In all cases, the blocking mutex is intended to be short lived and not held across await points.
/// Use the async [`Mutex`](crate::mutex::Mutex) if you need a lock that is held across await points.
pub struct Mutex<R, T: ?Sized> {
// NOTE: `raw` must be FIRST, so when using ThreadModeMutex the "can't drop in non-thread-mode" gets
// to run BEFORE dropping `data`.
raw: R,
data: UnsafeCell<T>,
}
unsafe impl<R: RawMutex + Send, T: ?Sized + Send> Send for Mutex<R, T> {}
unsafe impl<R: RawMutex + Sync, T: ?Sized + Send> Sync for Mutex<R, T> {}
impl<R: RawMutex, T> Mutex<R, T> {
/// Creates a new mutex in an unlocked state ready for use.
#[inline]
pub const fn new(val: T) -> Mutex<R, T> {
Mutex {
raw: R::INIT,
data: UnsafeCell::new(val),
}
}
/// Creates a critical section and grants temporary access to the protected data.
pub fn lock<U>(&self, f: impl FnOnce(&T) -> U) -> U {
self.raw.lock(|| {
let ptr = self.data.get() as *const T;
let inner = unsafe { &*ptr };
f(inner)
})
}
}
impl<R, T> Mutex<R, T> {
/// Creates a new mutex based on a pre-existing raw mutex.
///
/// This allows creating a mutex in a constant context on stable Rust.
#[inline]
pub const fn const_new(raw_mutex: R, val: T) -> Mutex<R, T> {
Mutex {
raw: raw_mutex,
data: UnsafeCell::new(val),
}
}
/// Consumes this mutex, returning the underlying data.
#[inline]
pub fn into_inner(self) -> T {
self.data.into_inner()
}
/// Returns a mutable reference to the underlying data.
///
/// Since this call borrows the `Mutex` mutably, no actual locking needs to
/// take place---the mutable borrow statically guarantees no locks exist.
#[inline]
pub fn get_mut(&mut self) -> &mut T {
unsafe { &mut *self.data.get() }
}
}
/// A mutex that allows borrowing data across executors and interrupts.
///
/// # Safety
///
/// This mutex is safe to share between different executors and interrupts.
pub type CriticalSectionMutex<T> = Mutex<raw::CriticalSectionRawMutex, T>;
/// A mutex that allows borrowing data in the context of a single executor.
///
/// # Safety
///
/// **This Mutex is only safe within a single executor.**
pub type NoopMutex<T> = Mutex<raw::NoopRawMutex, T>;
impl<T> Mutex<raw::CriticalSectionRawMutex, T> {
/// Borrows the data for the duration of the critical section
pub fn borrow<'cs>(&'cs self, _cs: critical_section::CriticalSection<'cs>) -> &'cs T {
let ptr = self.data.get() as *const T;
unsafe { &*ptr }
}
}
impl<T> Mutex<raw::NoopRawMutex, T> {
/// Borrows the data
pub fn borrow(&self) -> &T {
let ptr = self.data.get() as *const T;
unsafe { &*ptr }
}
}
// ThreadModeMutex does NOT use the generic mutex from above because it's special:
// it's Send+Sync even if T: !Send. There's no way to do that without specialization (I think?).
//
// There's still a ThreadModeRawMutex for use with the generic Mutex (handy with Channel, for example),
// but that will require T: Send even though it shouldn't be needed.
#[cfg(any(cortex_m, feature = "std"))]
pub use thread_mode_mutex::*;
#[cfg(any(cortex_m, feature = "std"))]
mod thread_mode_mutex {
use super::*;
/// A "mutex" that only allows borrowing from thread mode.
///
/// # Safety
///
/// **This Mutex is only safe on single-core systems.**
///
/// On multi-core systems, a `ThreadModeMutex` **is not sufficient** to ensure exclusive access.
pub struct ThreadModeMutex<T: ?Sized> {
inner: UnsafeCell<T>,
}
// NOTE: ThreadModeMutex only allows borrowing from one execution context ever: thread mode.
// Therefore it cannot be used to send non-sendable stuff between execution contexts, so it can
// be Send+Sync even if T is not Send (unlike CriticalSectionMutex)
unsafe impl<T: ?Sized> Sync for ThreadModeMutex<T> {}
unsafe impl<T: ?Sized> Send for ThreadModeMutex<T> {}
impl<T> ThreadModeMutex<T> {
/// Creates a new mutex
pub const fn new(value: T) -> Self {
ThreadModeMutex {
inner: UnsafeCell::new(value),
}
}
}
impl<T: ?Sized> ThreadModeMutex<T> {
/// Lock the `ThreadModeMutex`, granting access to the data.
///
/// # Panics
///
/// This will panic if not currently running in thread mode.
pub fn lock<R>(&self, f: impl FnOnce(&T) -> R) -> R {
f(self.borrow())
}
/// Borrows the data
///
/// # Panics
///
/// This will panic if not currently running in thread mode.
pub fn borrow(&self) -> &T {
assert!(
raw::in_thread_mode(),
"ThreadModeMutex can only be borrowed from thread mode."
);
unsafe { &*self.inner.get() }
}
}
impl<T: ?Sized> Drop for ThreadModeMutex<T> {
fn drop(&mut self) {
// Only allow dropping from thread mode. Dropping calls drop on the inner `T`, so
// `drop` needs the same guarantees as `lock`. `ThreadModeMutex<T>` is Send even if
// T isn't, so without this check a user could create a ThreadModeMutex in thread mode,
// send it to interrupt context and drop it there, which would "send" a T even if T is not Send.
assert!(
raw::in_thread_mode(),
"ThreadModeMutex can only be dropped from thread mode."
);
// Drop of the inner `T` happens after this.
}
}
}
+149
View File
@@ -0,0 +1,149 @@
//! Mutex primitives.
//!
//! This module provides a trait for mutexes that can be used in different contexts.
use core::marker::PhantomData;
/// Raw mutex trait.
///
/// This mutex is "raw", which means it does not actually contain the protected data, it
/// just implements the mutex mechanism. For most uses you should use [`super::Mutex`] instead,
/// which is generic over a RawMutex and contains the protected data.
///
/// Note that, unlike other mutexes, implementations only guarantee no
/// concurrent access from other threads: concurrent access from the current
/// thread is allowed. For example, it's possible to lock the same mutex multiple times reentrantly.
///
/// Therefore, locking a `RawMutex` is only enough to guarantee safe shared (`&`) access
/// to the data, it is not enough to guarantee exclusive (`&mut`) access.
///
/// # Safety
///
/// RawMutex implementations must ensure that, while locked, no other thread can lock
/// the RawMutex concurrently.
///
/// Unsafe code is allowed to rely on this fact, so incorrect implementations will cause undefined behavior.
pub unsafe trait RawMutex {
/// Create a new `RawMutex` instance.
///
/// This is a const instead of a method to allow creating instances in const context.
const INIT: Self;
/// Lock this `RawMutex`.
fn lock<R>(&self, f: impl FnOnce() -> R) -> R;
}
/// A mutex that allows borrowing data across executors and interrupts.
///
/// # Safety
///
/// This mutex is safe to share between different executors and interrupts.
pub struct CriticalSectionRawMutex {
_phantom: PhantomData<()>,
}
unsafe impl Send for CriticalSectionRawMutex {}
unsafe impl Sync for CriticalSectionRawMutex {}
impl CriticalSectionRawMutex {
/// Create a new `CriticalSectionRawMutex`.
pub const fn new() -> Self {
Self { _phantom: PhantomData }
}
}
unsafe impl RawMutex for CriticalSectionRawMutex {
const INIT: Self = Self::new();
fn lock<R>(&self, f: impl FnOnce() -> R) -> R {
critical_section::with(|_| f())
}
}
// ================
/// A mutex that allows borrowing data in the context of a single executor.
///
/// # Safety
///
/// **This Mutex is only safe within a single executor.**
pub struct NoopRawMutex {
_phantom: PhantomData<*mut ()>,
}
unsafe impl Send for NoopRawMutex {}
impl NoopRawMutex {
/// Create a new `NoopRawMutex`.
pub const fn new() -> Self {
Self { _phantom: PhantomData }
}
}
unsafe impl RawMutex for NoopRawMutex {
const INIT: Self = Self::new();
fn lock<R>(&self, f: impl FnOnce() -> R) -> R {
f()
}
}
// ================
#[cfg(any(cortex_m, feature = "std"))]
mod thread_mode {
use super::*;
/// A "mutex" that only allows borrowing from thread mode.
///
/// # Safety
///
/// **This Mutex is only safe on single-core systems.**
///
/// On multi-core systems, a `ThreadModeRawMutex` **is not sufficient** to ensure exclusive access.
pub struct ThreadModeRawMutex {
_phantom: PhantomData<()>,
}
unsafe impl Send for ThreadModeRawMutex {}
unsafe impl Sync for ThreadModeRawMutex {}
impl ThreadModeRawMutex {
/// Create a new `ThreadModeRawMutex`.
pub const fn new() -> Self {
Self { _phantom: PhantomData }
}
}
unsafe impl RawMutex for ThreadModeRawMutex {
const INIT: Self = Self::new();
fn lock<R>(&self, f: impl FnOnce() -> R) -> R {
assert!(in_thread_mode(), "ThreadModeMutex can only be locked from thread mode.");
f()
}
}
impl Drop for ThreadModeRawMutex {
fn drop(&mut self) {
// Only allow dropping from thread mode. Dropping calls drop on the inner `T`, so
// `drop` needs the same guarantees as `lock`. `ThreadModeMutex<T>` is Send even if
// T isn't, so without this check a user could create a ThreadModeMutex in thread mode,
// send it to interrupt context and drop it there, which would "send" a T even if T is not Send.
assert!(
in_thread_mode(),
"ThreadModeMutex can only be dropped from thread mode."
);
// Drop of the inner `T` happens after this.
}
}
pub(crate) fn in_thread_mode() -> bool {
#[cfg(feature = "std")]
return Some("main") == std::thread::current().name();
#[cfg(not(feature = "std"))]
// ICSR.VECTACTIVE == 0
return unsafe { (0xE000ED04 as *const u32).read_volatile() } & 0x1FF == 0;
}
}
#[cfg(any(cortex_m, feature = "std"))]
pub use thread_mode::*;
+708
View File
@@ -0,0 +1,708 @@
//! A queue for sending values between asynchronous tasks.
//!
//! It can be used concurrently by multiple producers (senders) and multiple
//! consumers (receivers), i.e. it is an "MPMC channel".
//!
//! Receivers are competing for messages. So a message that is received by
//! one receiver is not received by any other.
//!
//! This queue takes a Mutex type so that various
//! targets can be attained. For example, a ThreadModeMutex can be used
//! for single-core Cortex-M targets where messages are only passed
//! between tasks running in thread mode. Similarly, a CriticalSectionMutex
//! can also be used for single-core targets where messages are to be
//! passed from exception mode e.g. out of an interrupt handler.
//!
//! This module provides a bounded channel that has a limit on the number of
//! messages that it can store, and if this limit is reached, trying to send
//! another message will result in an error being returned.
//!
use core::cell::RefCell;
use core::future::Future;
use core::pin::Pin;
use core::task::{Context, Poll};
use heapless::Deque;
use crate::blocking_mutex::raw::RawMutex;
use crate::blocking_mutex::Mutex;
use crate::waitqueue::WakerRegistration;
/// Send-only access to a [`Channel`].
pub struct Sender<'ch, M, T, const N: usize>
where
M: RawMutex,
{
channel: &'ch Channel<M, T, N>,
}
impl<'ch, M, T, const N: usize> Clone for Sender<'ch, M, T, N>
where
M: RawMutex,
{
fn clone(&self) -> Self {
Sender { channel: self.channel }
}
}
impl<'ch, M, T, const N: usize> Copy for Sender<'ch, M, T, N> where M: RawMutex {}
impl<'ch, M, T, const N: usize> Sender<'ch, M, T, N>
where
M: RawMutex,
{
/// Sends a value.
///
/// See [`Channel::send()`]
pub fn send(&self, message: T) -> SendFuture<'ch, M, T, N> {
self.channel.send(message)
}
/// Attempt to immediately send a message.
///
/// See [`Channel::send()`]
pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
self.channel.try_send(message)
}
/// Allows a poll_fn to poll until the channel is ready to send
///
/// See [`Channel::poll_ready_to_send()`]
pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
self.channel.poll_ready_to_send(cx)
}
}
/// Send-only access to a [`Channel`] without knowing channel size.
pub struct DynamicSender<'ch, T> {
pub(crate) channel: &'ch dyn DynamicChannel<T>,
}
impl<'ch, T> Clone for DynamicSender<'ch, T> {
fn clone(&self) -> Self {
DynamicSender { channel: self.channel }
}
}
impl<'ch, T> Copy for DynamicSender<'ch, T> {}
impl<'ch, M, T, const N: usize> From<Sender<'ch, M, T, N>> for DynamicSender<'ch, T>
where
M: RawMutex,
{
fn from(s: Sender<'ch, M, T, N>) -> Self {
Self { channel: s.channel }
}
}
impl<'ch, T> DynamicSender<'ch, T> {
/// Sends a value.
///
/// See [`Channel::send()`]
pub fn send(&self, message: T) -> DynamicSendFuture<'ch, T> {
DynamicSendFuture {
channel: self.channel,
message: Some(message),
}
}
/// Attempt to immediately send a message.
///
/// See [`Channel::send()`]
pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
self.channel.try_send_with_context(message, None)
}
/// Allows a poll_fn to poll until the channel is ready to send
///
/// See [`Channel::poll_ready_to_send()`]
pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
self.channel.poll_ready_to_send(cx)
}
}
/// Receive-only access to a [`Channel`].
pub struct Receiver<'ch, M, T, const N: usize>
where
M: RawMutex,
{
channel: &'ch Channel<M, T, N>,
}
impl<'ch, M, T, const N: usize> Clone for Receiver<'ch, M, T, N>
where
M: RawMutex,
{
fn clone(&self) -> Self {
Receiver { channel: self.channel }
}
}
impl<'ch, M, T, const N: usize> Copy for Receiver<'ch, M, T, N> where M: RawMutex {}
impl<'ch, M, T, const N: usize> Receiver<'ch, M, T, N>
where
M: RawMutex,
{
/// Receive the next value.
///
/// See [`Channel::receive()`].
pub fn receive(&self) -> ReceiveFuture<'_, M, T, N> {
self.channel.receive()
}
/// Attempt to immediately receive the next value.
///
/// See [`Channel::try_receive()`]
pub fn try_receive(&self) -> Result<T, TryReceiveError> {
self.channel.try_receive()
}
/// Allows a poll_fn to poll until the channel is ready to receive
///
/// See [`Channel::poll_ready_to_receive()`]
pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
self.channel.poll_ready_to_receive(cx)
}
/// Poll the channel for the next item
///
/// See [`Channel::poll_receive()`]
pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
self.channel.poll_receive(cx)
}
}
/// Receive-only access to a [`Channel`] without knowing channel size.
pub struct DynamicReceiver<'ch, T> {
pub(crate) channel: &'ch dyn DynamicChannel<T>,
}
impl<'ch, T> Clone for DynamicReceiver<'ch, T> {
fn clone(&self) -> Self {
DynamicReceiver { channel: self.channel }
}
}
impl<'ch, T> Copy for DynamicReceiver<'ch, T> {}
impl<'ch, T> DynamicReceiver<'ch, T> {
/// Receive the next value.
///
/// See [`Channel::receive()`].
pub fn receive(&self) -> DynamicReceiveFuture<'_, T> {
DynamicReceiveFuture { channel: self.channel }
}
/// Attempt to immediately receive the next value.
///
/// See [`Channel::try_receive()`]
pub fn try_receive(&self) -> Result<T, TryReceiveError> {
self.channel.try_receive_with_context(None)
}
/// Allows a poll_fn to poll until the channel is ready to receive
///
/// See [`Channel::poll_ready_to_receive()`]
pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
self.channel.poll_ready_to_receive(cx)
}
/// Poll the channel for the next item
///
/// See [`Channel::poll_receive()`]
pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
self.channel.poll_receive(cx)
}
}
impl<'ch, M, T, const N: usize> From<Receiver<'ch, M, T, N>> for DynamicReceiver<'ch, T>
where
M: RawMutex,
{
fn from(s: Receiver<'ch, M, T, N>) -> Self {
Self { channel: s.channel }
}
}
/// Future returned by [`Channel::receive`] and [`Receiver::receive`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct ReceiveFuture<'ch, M, T, const N: usize>
where
M: RawMutex,
{
channel: &'ch Channel<M, T, N>,
}
impl<'ch, M, T, const N: usize> Future for ReceiveFuture<'ch, M, T, N>
where
M: RawMutex,
{
type Output = T;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
self.channel.poll_receive(cx)
}
}
/// Future returned by [`DynamicReceiver::receive`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct DynamicReceiveFuture<'ch, T> {
channel: &'ch dyn DynamicChannel<T>,
}
impl<'ch, T> Future for DynamicReceiveFuture<'ch, T> {
type Output = T;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
match self.channel.try_receive_with_context(Some(cx)) {
Ok(v) => Poll::Ready(v),
Err(TryReceiveError::Empty) => Poll::Pending,
}
}
}
/// Future returned by [`Channel::send`] and [`Sender::send`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct SendFuture<'ch, M, T, const N: usize>
where
M: RawMutex,
{
channel: &'ch Channel<M, T, N>,
message: Option<T>,
}
impl<'ch, M, T, const N: usize> Future for SendFuture<'ch, M, T, N>
where
M: RawMutex,
{
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.message.take() {
Some(m) => match self.channel.try_send_with_context(m, Some(cx)) {
Ok(..) => Poll::Ready(()),
Err(TrySendError::Full(m)) => {
self.message = Some(m);
Poll::Pending
}
},
None => panic!("Message cannot be None"),
}
}
}
impl<'ch, M, T, const N: usize> Unpin for SendFuture<'ch, M, T, N> where M: RawMutex {}
/// Future returned by [`DynamicSender::send`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct DynamicSendFuture<'ch, T> {
channel: &'ch dyn DynamicChannel<T>,
message: Option<T>,
}
impl<'ch, T> Future for DynamicSendFuture<'ch, T> {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.message.take() {
Some(m) => match self.channel.try_send_with_context(m, Some(cx)) {
Ok(..) => Poll::Ready(()),
Err(TrySendError::Full(m)) => {
self.message = Some(m);
Poll::Pending
}
},
None => panic!("Message cannot be None"),
}
}
}
impl<'ch, T> Unpin for DynamicSendFuture<'ch, T> {}
pub(crate) trait DynamicChannel<T> {
fn try_send_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>>;
fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>;
fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()>;
fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()>;
fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T>;
}
/// Error returned by [`try_receive`](Channel::try_receive).
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum TryReceiveError {
/// A message could not be received because the channel is empty.
Empty,
}
/// Error returned by [`try_send`](Channel::try_send).
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum TrySendError<T> {
/// The data could not be sent on the channel because the channel is
/// currently full and sending would require blocking.
Full(T),
}
struct ChannelState<T, const N: usize> {
queue: Deque<T, N>,
receiver_waker: WakerRegistration,
senders_waker: WakerRegistration,
}
impl<T, const N: usize> ChannelState<T, N> {
const fn new() -> Self {
ChannelState {
queue: Deque::new(),
receiver_waker: WakerRegistration::new(),
senders_waker: WakerRegistration::new(),
}
}
fn try_receive(&mut self) -> Result<T, TryReceiveError> {
self.try_receive_with_context(None)
}
fn try_receive_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
if self.queue.is_full() {
self.senders_waker.wake();
}
if let Some(message) = self.queue.pop_front() {
Ok(message)
} else {
if let Some(cx) = cx {
self.receiver_waker.register(cx.waker());
}
Err(TryReceiveError::Empty)
}
}
fn poll_receive(&mut self, cx: &mut Context<'_>) -> Poll<T> {
if self.queue.is_full() {
self.senders_waker.wake();
}
if let Some(message) = self.queue.pop_front() {
Poll::Ready(message)
} else {
self.receiver_waker.register(cx.waker());
Poll::Pending
}
}
fn poll_ready_to_receive(&mut self, cx: &mut Context<'_>) -> Poll<()> {
self.receiver_waker.register(cx.waker());
if !self.queue.is_empty() {
Poll::Ready(())
} else {
Poll::Pending
}
}
fn try_send(&mut self, message: T) -> Result<(), TrySendError<T>> {
self.try_send_with_context(message, None)
}
fn try_send_with_context(&mut self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
match self.queue.push_back(message) {
Ok(()) => {
self.receiver_waker.wake();
Ok(())
}
Err(message) => {
if let Some(cx) = cx {
self.senders_waker.register(cx.waker());
}
Err(TrySendError::Full(message))
}
}
}
fn poll_ready_to_send(&mut self, cx: &mut Context<'_>) -> Poll<()> {
self.senders_waker.register(cx.waker());
if !self.queue.is_full() {
Poll::Ready(())
} else {
Poll::Pending
}
}
}
/// A bounded channel for communicating between asynchronous tasks
/// with backpressure.
///
/// The channel will buffer up to the provided number of messages. Once the
/// buffer is full, attempts to `send` new messages will wait until a message is
/// received from the channel.
///
/// All data sent will become available in the same order as it was sent.
pub struct Channel<M, T, const N: usize>
where
M: RawMutex,
{
inner: Mutex<M, RefCell<ChannelState<T, N>>>,
}
impl<M, T, const N: usize> Channel<M, T, N>
where
M: RawMutex,
{
/// Establish a new bounded channel. For example, to create one with a NoopMutex:
///
/// ```
/// use embassy_sync::channel::Channel;
/// use embassy_sync::blocking_mutex::raw::NoopRawMutex;
///
/// // Declare a bounded channel of 3 u32s.
/// let mut channel = Channel::<NoopRawMutex, u32, 3>::new();
/// ```
pub const fn new() -> Self {
Self {
inner: Mutex::new(RefCell::new(ChannelState::new())),
}
}
fn lock<R>(&self, f: impl FnOnce(&mut ChannelState<T, N>) -> R) -> R {
self.inner.lock(|rc| f(&mut *unwrap!(rc.try_borrow_mut())))
}
fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
self.lock(|c| c.try_receive_with_context(cx))
}
/// Poll the channel for the next message
pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
self.lock(|c| c.poll_receive(cx))
}
fn try_send_with_context(&self, m: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
self.lock(|c| c.try_send_with_context(m, cx))
}
/// Allows a poll_fn to poll until the channel is ready to receive
pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
self.lock(|c| c.poll_ready_to_receive(cx))
}
/// Allows a poll_fn to poll until the channel is ready to send
pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
self.lock(|c| c.poll_ready_to_send(cx))
}
/// Get a sender for this channel.
pub fn sender(&self) -> Sender<'_, M, T, N> {
Sender { channel: self }
}
/// Get a receiver for this channel.
pub fn receiver(&self) -> Receiver<'_, M, T, N> {
Receiver { channel: self }
}
/// Send a value, waiting until there is capacity.
///
/// Sending completes when the value has been pushed to the channel's queue.
/// This doesn't mean the value has been received yet.
pub fn send(&self, message: T) -> SendFuture<'_, M, T, N> {
SendFuture {
channel: self,
message: Some(message),
}
}
/// Attempt to immediately send a message.
///
/// This method differs from [`send`](Channel::send) by returning immediately if the channel's
/// buffer is full, instead of waiting.
///
/// # Errors
///
/// If the channel capacity has been reached, i.e., the channel has `n`
/// buffered values where `n` is the argument passed to [`Channel`], then an
/// error is returned.
pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
self.lock(|c| c.try_send(message))
}
/// Receive the next value.
///
/// If there are no messages in the channel's buffer, this method will
/// wait until a message is sent.
pub fn receive(&self) -> ReceiveFuture<'_, M, T, N> {
ReceiveFuture { channel: self }
}
/// Attempt to immediately receive a message.
///
/// This method will either receive a message from the channel immediately or return an error
/// if the channel is empty.
pub fn try_receive(&self) -> Result<T, TryReceiveError> {
self.lock(|c| c.try_receive())
}
}
/// Implements the DynamicChannel to allow creating types that are unaware of the queue size with the
/// tradeoff cost of dynamic dispatch.
impl<M, T, const N: usize> DynamicChannel<T> for Channel<M, T, N>
where
M: RawMutex,
{
fn try_send_with_context(&self, m: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
Channel::try_send_with_context(self, m, cx)
}
fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
Channel::try_receive_with_context(self, cx)
}
fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
Channel::poll_ready_to_send(self, cx)
}
fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
Channel::poll_ready_to_receive(self, cx)
}
fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
Channel::poll_receive(self, cx)
}
}
#[cfg(test)]
mod tests {
use core::time::Duration;
use futures_executor::ThreadPool;
use futures_timer::Delay;
use futures_util::task::SpawnExt;
use static_cell::StaticCell;
use super::*;
use crate::blocking_mutex::raw::{CriticalSectionRawMutex, NoopRawMutex};
fn capacity<T, const N: usize>(c: &ChannelState<T, N>) -> usize {
c.queue.capacity() - c.queue.len()
}
#[test]
fn sending_once() {
let mut c = ChannelState::<u32, 3>::new();
assert!(c.try_send(1).is_ok());
assert_eq!(capacity(&c), 2);
}
#[test]
fn sending_when_full() {
let mut c = ChannelState::<u32, 3>::new();
let _ = c.try_send(1);
let _ = c.try_send(1);
let _ = c.try_send(1);
match c.try_send(2) {
Err(TrySendError::Full(2)) => assert!(true),
_ => assert!(false),
}
assert_eq!(capacity(&c), 0);
}
#[test]
fn receiving_once_with_one_send() {
let mut c = ChannelState::<u32, 3>::new();
assert!(c.try_send(1).is_ok());
assert_eq!(c.try_receive().unwrap(), 1);
assert_eq!(capacity(&c), 3);
}
#[test]
fn receiving_when_empty() {
let mut c = ChannelState::<u32, 3>::new();
match c.try_receive() {
Err(TryReceiveError::Empty) => assert!(true),
_ => assert!(false),
}
assert_eq!(capacity(&c), 3);
}
#[test]
fn simple_send_and_receive() {
let c = Channel::<NoopRawMutex, u32, 3>::new();
assert!(c.try_send(1).is_ok());
assert_eq!(c.try_receive().unwrap(), 1);
}
#[test]
fn cloning() {
let c = Channel::<NoopRawMutex, u32, 3>::new();
let r1 = c.receiver();
let s1 = c.sender();
let _ = r1.clone();
let _ = s1.clone();
}
#[test]
fn dynamic_dispatch() {
let c = Channel::<NoopRawMutex, u32, 3>::new();
let s: DynamicSender<'_, u32> = c.sender().into();
let r: DynamicReceiver<'_, u32> = c.receiver().into();
assert!(s.try_send(1).is_ok());
assert_eq!(r.try_receive().unwrap(), 1);
}
#[futures_test::test]
async fn receiver_receives_given_try_send_async() {
let executor = ThreadPool::new().unwrap();
static CHANNEL: StaticCell<Channel<CriticalSectionRawMutex, u32, 3>> = StaticCell::new();
let c = &*CHANNEL.init(Channel::new());
let c2 = c;
assert!(executor
.spawn(async move {
assert!(c2.try_send(1).is_ok());
})
.is_ok());
assert_eq!(c.receive().await, 1);
}
#[futures_test::test]
async fn sender_send_completes_if_capacity() {
let c = Channel::<CriticalSectionRawMutex, u32, 1>::new();
c.send(1).await;
assert_eq!(c.receive().await, 1);
}
#[futures_test::test]
async fn senders_sends_wait_until_capacity() {
let executor = ThreadPool::new().unwrap();
static CHANNEL: StaticCell<Channel<CriticalSectionRawMutex, u32, 1>> = StaticCell::new();
let c = &*CHANNEL.init(Channel::new());
assert!(c.try_send(1).is_ok());
let c2 = c;
let send_task_1 = executor.spawn_with_handle(async move { c2.send(2).await });
let c2 = c;
let send_task_2 = executor.spawn_with_handle(async move { c2.send(3).await });
// Wish I could think of a means of determining that the async send is waiting instead.
// However, I've used the debugger to observe that the send does indeed wait.
Delay::new(Duration::from_millis(500)).await;
assert_eq!(c.receive().await, 1);
assert!(executor
.spawn(async move {
loop {
c.receive().await;
}
})
.is_ok());
send_task_1.unwrap().await;
send_task_2.unwrap().await;
}
}
+258
View File
@@ -0,0 +1,258 @@
#![macro_use]
#![allow(unused_macros)]
use core::fmt::{Debug, Display, LowerHex};
#[cfg(all(feature = "defmt", feature = "log"))]
compile_error!("You may not enable both `defmt` and `log` features.");
macro_rules! assert {
($($x:tt)*) => {
{
#[cfg(not(feature = "defmt"))]
::core::assert!($($x)*);
#[cfg(feature = "defmt")]
::defmt::assert!($($x)*);
}
};
}
macro_rules! assert_eq {
($($x:tt)*) => {
{
#[cfg(not(feature = "defmt"))]
::core::assert_eq!($($x)*);
#[cfg(feature = "defmt")]
::defmt::assert_eq!($($x)*);
}
};
}
macro_rules! assert_ne {
($($x:tt)*) => {
{
#[cfg(not(feature = "defmt"))]
::core::assert_ne!($($x)*);
#[cfg(feature = "defmt")]
::defmt::assert_ne!($($x)*);
}
};
}
macro_rules! debug_assert {
($($x:tt)*) => {
{
#[cfg(not(feature = "defmt"))]
::core::debug_assert!($($x)*);
#[cfg(feature = "defmt")]
::defmt::debug_assert!($($x)*);
}
};
}
macro_rules! debug_assert_eq {
($($x:tt)*) => {
{
#[cfg(not(feature = "defmt"))]
::core::debug_assert_eq!($($x)*);
#[cfg(feature = "defmt")]
::defmt::debug_assert_eq!($($x)*);
}
};
}
macro_rules! debug_assert_ne {
($($x:tt)*) => {
{
#[cfg(not(feature = "defmt"))]
::core::debug_assert_ne!($($x)*);
#[cfg(feature = "defmt")]
::defmt::debug_assert_ne!($($x)*);
}
};
}
macro_rules! todo {
($($x:tt)*) => {
{
#[cfg(not(feature = "defmt"))]
::core::todo!($($x)*);
#[cfg(feature = "defmt")]
::defmt::todo!($($x)*);
}
};
}
#[cfg(not(feature = "defmt"))]
macro_rules! unreachable {
($($x:tt)*) => {
::core::unreachable!($($x)*)
};
}
#[cfg(feature = "defmt")]
macro_rules! unreachable {
($($x:tt)*) => {
::defmt::unreachable!($($x)*)
};
}
macro_rules! panic {
($($x:tt)*) => {
{
#[cfg(not(feature = "defmt"))]
::core::panic!($($x)*);
#[cfg(feature = "defmt")]
::defmt::panic!($($x)*);
}
};
}
macro_rules! trace {
($s:literal $(, $x:expr)* $(,)?) => {
{
#[cfg(feature = "log")]
::log::trace!($s $(, $x)*);
#[cfg(feature = "defmt")]
::defmt::trace!($s $(, $x)*);
#[cfg(not(any(feature = "log", feature="defmt")))]
let _ = ($( & $x ),*);
}
};
}
macro_rules! debug {
($s:literal $(, $x:expr)* $(,)?) => {
{
#[cfg(feature = "log")]
::log::debug!($s $(, $x)*);
#[cfg(feature = "defmt")]
::defmt::debug!($s $(, $x)*);
#[cfg(not(any(feature = "log", feature="defmt")))]
let _ = ($( & $x ),*);
}
};
}
macro_rules! info {
($s:literal $(, $x:expr)* $(,)?) => {
{
#[cfg(feature = "log")]
::log::info!($s $(, $x)*);
#[cfg(feature = "defmt")]
::defmt::info!($s $(, $x)*);
#[cfg(not(any(feature = "log", feature="defmt")))]
let _ = ($( & $x ),*);
}
};
}
macro_rules! warn {
($s:literal $(, $x:expr)* $(,)?) => {
{
#[cfg(feature = "log")]
::log::warn!($s $(, $x)*);
#[cfg(feature = "defmt")]
::defmt::warn!($s $(, $x)*);
#[cfg(not(any(feature = "log", feature="defmt")))]
let _ = ($( & $x ),*);
}
};
}
macro_rules! error {
($s:literal $(, $x:expr)* $(,)?) => {
{
#[cfg(feature = "log")]
::log::error!($s $(, $x)*);
#[cfg(feature = "defmt")]
::defmt::error!($s $(, $x)*);
#[cfg(not(any(feature = "log", feature="defmt")))]
let _ = ($( & $x ),*);
}
};
}
#[cfg(feature = "defmt")]
macro_rules! unwrap {
($($x:tt)*) => {
::defmt::unwrap!($($x)*)
};
}
#[cfg(not(feature = "defmt"))]
macro_rules! unwrap {
($arg:expr) => {
match $crate::fmt::Try::into_result($arg) {
::core::result::Result::Ok(t) => t,
::core::result::Result::Err(e) => {
::core::panic!("unwrap of `{}` failed: {:?}", ::core::stringify!($arg), e);
}
}
};
($arg:expr, $($msg:expr),+ $(,)? ) => {
match $crate::fmt::Try::into_result($arg) {
::core::result::Result::Ok(t) => t,
::core::result::Result::Err(e) => {
::core::panic!("unwrap of `{}` failed: {}: {:?}", ::core::stringify!($arg), ::core::format_args!($($msg,)*), e);
}
}
}
}
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
pub struct NoneError;
pub trait Try {
type Ok;
type Error;
fn into_result(self) -> Result<Self::Ok, Self::Error>;
}
impl<T> Try for Option<T> {
type Ok = T;
type Error = NoneError;
#[inline]
fn into_result(self) -> Result<T, NoneError> {
self.ok_or(NoneError)
}
}
impl<T, E> Try for Result<T, E> {
type Ok = T;
type Error = E;
#[inline]
fn into_result(self) -> Self {
self
}
}
#[allow(unused)]
pub(crate) struct Bytes<'a>(pub &'a [u8]);
impl<'a> Debug for Bytes<'a> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
write!(f, "{:#02x?}", self.0)
}
}
impl<'a> Display for Bytes<'a> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
write!(f, "{:#02x?}", self.0)
}
}
impl<'a> LowerHex for Bytes<'a> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
write!(f, "{:#02x?}", self.0)
}
}
#[cfg(feature = "defmt")]
impl<'a> defmt::Format for Bytes<'a> {
fn format(&self, fmt: defmt::Formatter) {
defmt::write!(fmt, "{:02x}", self.0)
}
}
+23
View File
@@ -0,0 +1,23 @@
#![cfg_attr(not(any(feature = "std", feature = "wasm")), no_std)]
#![cfg_attr(nightly, feature(async_fn_in_trait, impl_trait_projections))]
#![cfg_attr(nightly, allow(stable_features, unknown_lints))]
#![allow(async_fn_in_trait)]
#![allow(clippy::new_without_default)]
#![doc = include_str!("../README.md")]
#![warn(missing_docs)]
// This mod MUST go first, so that the others see its macros.
pub(crate) mod fmt;
// internal use
mod ring_buffer;
pub mod blocking_mutex;
pub mod channel;
pub mod mutex;
pub mod pipe;
pub mod priority_channel;
pub mod pubsub;
pub mod signal;
pub mod waitqueue;
pub mod zerocopy_channel;
+182
View File
@@ -0,0 +1,182 @@
//! Async mutex.
//!
//! This module provides a mutex that can be used to synchronize data between asynchronous tasks.
use core::cell::{RefCell, UnsafeCell};
use core::future::poll_fn;
use core::ops::{Deref, DerefMut};
use core::task::Poll;
use crate::blocking_mutex::raw::RawMutex;
use crate::blocking_mutex::Mutex as BlockingMutex;
use crate::waitqueue::WakerRegistration;
/// Error returned by [`Mutex::try_lock`]
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub struct TryLockError;
struct State {
locked: bool,
waker: WakerRegistration,
}
/// Async mutex.
///
/// The mutex is generic over a blocking [`RawMutex`](crate::blocking_mutex::raw::RawMutex).
/// The raw mutex is used to guard access to the internal "is locked" flag. It
/// is held for very short periods only, while locking and unlocking. It is *not* held
/// for the entire time the async Mutex is locked.
///
/// Which implementation you select depends on the context in which you're using the mutex.
///
/// Use [`CriticalSectionRawMutex`](crate::blocking_mutex::raw::CriticalSectionRawMutex) when data can be shared between threads and interrupts.
///
/// Use [`NoopRawMutex`](crate::blocking_mutex::raw::NoopRawMutex) when data is only shared between tasks running on the same executor.
///
/// Use [`ThreadModeRawMutex`](crate::blocking_mutex::raw::ThreadModeRawMutex) when data is shared between tasks running on the same executor but you want a singleton.
///
pub struct Mutex<M, T>
where
M: RawMutex,
T: ?Sized,
{
state: BlockingMutex<M, RefCell<State>>,
inner: UnsafeCell<T>,
}
unsafe impl<M: RawMutex + Send, T: ?Sized + Send> Send for Mutex<M, T> {}
unsafe impl<M: RawMutex + Sync, T: ?Sized + Send> Sync for Mutex<M, T> {}
/// Async mutex.
impl<M, T> Mutex<M, T>
where
M: RawMutex,
{
/// Create a new mutex with the given value.
pub const fn new(value: T) -> Self {
Self {
inner: UnsafeCell::new(value),
state: BlockingMutex::new(RefCell::new(State {
locked: false,
waker: WakerRegistration::new(),
})),
}
}
}
impl<M, T> Mutex<M, T>
where
M: RawMutex,
T: ?Sized,
{
/// Lock the mutex.
///
/// This will wait for the mutex to be unlocked if it's already locked.
pub async fn lock(&self) -> MutexGuard<'_, M, T> {
poll_fn(|cx| {
let ready = self.state.lock(|s| {
let mut s = s.borrow_mut();
if s.locked {
s.waker.register(cx.waker());
false
} else {
s.locked = true;
true
}
});
if ready {
Poll::Ready(MutexGuard { mutex: self })
} else {
Poll::Pending
}
})
.await
}
/// Attempt to immediately lock the mutex.
///
/// If the mutex is already locked, this will return an error instead of waiting.
pub fn try_lock(&self) -> Result<MutexGuard<'_, M, T>, TryLockError> {
self.state.lock(|s| {
let mut s = s.borrow_mut();
if s.locked {
Err(TryLockError)
} else {
s.locked = true;
Ok(())
}
})?;
Ok(MutexGuard { mutex: self })
}
/// Consumes this mutex, returning the underlying data.
pub fn into_inner(self) -> T
where
T: Sized,
{
self.inner.into_inner()
}
/// Returns a mutable reference to the underlying data.
///
/// Since this call borrows the Mutex mutably, no actual locking needs to
/// take place -- the mutable borrow statically guarantees no locks exist.
pub fn get_mut(&mut self) -> &mut T {
self.inner.get_mut()
}
}
/// Async mutex guard.
///
/// Owning an instance of this type indicates having
/// successfully locked the mutex, and grants access to the contents.
///
/// Dropping it unlocks the mutex.
pub struct MutexGuard<'a, M, T>
where
M: RawMutex,
T: ?Sized,
{
mutex: &'a Mutex<M, T>,
}
impl<'a, M, T> Drop for MutexGuard<'a, M, T>
where
M: RawMutex,
T: ?Sized,
{
fn drop(&mut self) {
self.mutex.state.lock(|s| {
let mut s = unwrap!(s.try_borrow_mut());
s.locked = false;
s.waker.wake();
})
}
}
impl<'a, M, T> Deref for MutexGuard<'a, M, T>
where
M: RawMutex,
T: ?Sized,
{
type Target = T;
fn deref(&self) -> &Self::Target {
// Safety: the MutexGuard represents exclusive access to the contents
// of the mutex, so it's OK to get it.
unsafe { &*(self.mutex.inner.get() as *const T) }
}
}
impl<'a, M, T> DerefMut for MutexGuard<'a, M, T>
where
M: RawMutex,
T: ?Sized,
{
fn deref_mut(&mut self) -> &mut Self::Target {
// Safety: the MutexGuard represents exclusive access to the contents
// of the mutex, so it's OK to get it.
unsafe { &mut *(self.mutex.inner.get()) }
}
}
+646
View File
@@ -0,0 +1,646 @@
//! Async byte stream pipe.
use core::cell::{RefCell, UnsafeCell};
use core::convert::Infallible;
use core::future::Future;
use core::ops::Range;
use core::pin::Pin;
use core::task::{Context, Poll};
use crate::blocking_mutex::raw::RawMutex;
use crate::blocking_mutex::Mutex;
use crate::ring_buffer::RingBuffer;
use crate::waitqueue::WakerRegistration;
/// Write-only access to a [`Pipe`].
pub struct Writer<'p, M, const N: usize>
where
M: RawMutex,
{
pipe: &'p Pipe<M, N>,
}
impl<'p, M, const N: usize> Clone for Writer<'p, M, N>
where
M: RawMutex,
{
fn clone(&self) -> Self {
Writer { pipe: self.pipe }
}
}
impl<'p, M, const N: usize> Copy for Writer<'p, M, N> where M: RawMutex {}
impl<'p, M, const N: usize> Writer<'p, M, N>
where
M: RawMutex,
{
/// Write some bytes to the pipe.
///
/// See [`Pipe::write()`]
pub fn write<'a>(&'a self, buf: &'a [u8]) -> WriteFuture<'a, M, N> {
self.pipe.write(buf)
}
/// Attempt to immediately write some bytes to the pipe.
///
/// See [`Pipe::try_write()`]
pub fn try_write(&self, buf: &[u8]) -> Result<usize, TryWriteError> {
self.pipe.try_write(buf)
}
}
/// Future returned by [`Pipe::write`] and [`Writer::write`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct WriteFuture<'p, M, const N: usize>
where
M: RawMutex,
{
pipe: &'p Pipe<M, N>,
buf: &'p [u8],
}
impl<'p, M, const N: usize> Future for WriteFuture<'p, M, N>
where
M: RawMutex,
{
type Output = usize;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.pipe.try_write_with_context(Some(cx), self.buf) {
Ok(n) => Poll::Ready(n),
Err(TryWriteError::Full) => Poll::Pending,
}
}
}
impl<'p, M, const N: usize> Unpin for WriteFuture<'p, M, N> where M: RawMutex {}
/// Read-only access to a [`Pipe`].
pub struct Reader<'p, M, const N: usize>
where
M: RawMutex,
{
pipe: &'p Pipe<M, N>,
}
impl<'p, M, const N: usize> Reader<'p, M, N>
where
M: RawMutex,
{
/// Read some bytes from the pipe.
///
/// See [`Pipe::read()`]
pub fn read<'a>(&'a self, buf: &'a mut [u8]) -> ReadFuture<'a, M, N> {
self.pipe.read(buf)
}
/// Attempt to immediately read some bytes from the pipe.
///
/// See [`Pipe::try_read()`]
pub fn try_read(&self, buf: &mut [u8]) -> Result<usize, TryReadError> {
self.pipe.try_read(buf)
}
/// Return the contents of the internal buffer, filling it with more data from the inner reader if it is empty.
///
/// If no bytes are currently available to read, this function waits until at least one byte is available.
///
/// If the reader is at end-of-file (EOF), an empty slice is returned.
pub fn fill_buf(&mut self) -> FillBufFuture<'_, M, N> {
FillBufFuture { pipe: Some(self.pipe) }
}
/// Try returning contents of the internal buffer.
///
/// If no bytes are currently available to read, this function returns `Err(TryReadError::Empty)`.
///
/// If the reader is at end-of-file (EOF), an empty slice is returned.
pub fn try_fill_buf(&mut self) -> Result<&[u8], TryReadError> {
unsafe { self.pipe.try_fill_buf_with_context(None) }
}
/// Tell this buffer that `amt` bytes have been consumed from the buffer, so they should no longer be returned in calls to `fill_buf`.
pub fn consume(&mut self, amt: usize) {
self.pipe.consume(amt)
}
}
/// Future returned by [`Pipe::read`] and [`Reader::read`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct ReadFuture<'p, M, const N: usize>
where
M: RawMutex,
{
pipe: &'p Pipe<M, N>,
buf: &'p mut [u8],
}
impl<'p, M, const N: usize> Future for ReadFuture<'p, M, N>
where
M: RawMutex,
{
type Output = usize;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.pipe.try_read_with_context(Some(cx), self.buf) {
Ok(n) => Poll::Ready(n),
Err(TryReadError::Empty) => Poll::Pending,
}
}
}
impl<'p, M, const N: usize> Unpin for ReadFuture<'p, M, N> where M: RawMutex {}
/// Future returned by [`Pipe::fill_buf`] and [`Reader::fill_buf`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct FillBufFuture<'p, M, const N: usize>
where
M: RawMutex,
{
pipe: Option<&'p Pipe<M, N>>,
}
impl<'p, M, const N: usize> Future for FillBufFuture<'p, M, N>
where
M: RawMutex,
{
type Output = &'p [u8];
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let pipe = self.pipe.take().unwrap();
match unsafe { pipe.try_fill_buf_with_context(Some(cx)) } {
Ok(buf) => Poll::Ready(buf),
Err(TryReadError::Empty) => {
self.pipe = Some(pipe);
Poll::Pending
}
}
}
}
impl<'p, M, const N: usize> Unpin for FillBufFuture<'p, M, N> where M: RawMutex {}
/// Error returned by [`try_read`](Pipe::try_read).
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum TryReadError {
/// No data could be read from the pipe because it is currently
/// empty, and reading would require blocking.
Empty,
}
/// Error returned by [`try_write`](Pipe::try_write).
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum TryWriteError {
/// No data could be written to the pipe because it is
/// currently full, and writing would require blocking.
Full,
}
struct PipeState<const N: usize> {
buffer: RingBuffer<N>,
read_waker: WakerRegistration,
write_waker: WakerRegistration,
}
#[repr(transparent)]
struct Buffer<const N: usize>(UnsafeCell<[u8; N]>);
impl<const N: usize> Buffer<N> {
unsafe fn get<'a>(&self, r: Range<usize>) -> &'a [u8] {
let p = self.0.get() as *const u8;
core::slice::from_raw_parts(p.add(r.start), r.end - r.start)
}
unsafe fn get_mut<'a>(&self, r: Range<usize>) -> &'a mut [u8] {
let p = self.0.get() as *mut u8;
core::slice::from_raw_parts_mut(p.add(r.start), r.end - r.start)
}
}
unsafe impl<const N: usize> Send for Buffer<N> {}
unsafe impl<const N: usize> Sync for Buffer<N> {}
/// A bounded byte-oriented pipe for communicating between asynchronous tasks
/// with backpressure.
///
/// The pipe will buffer up to the provided number of bytes. Once the
/// buffer is full, attempts to `write` new bytes will wait until buffer space is freed up.
///
/// All data written will become available in the same order as it was written.
pub struct Pipe<M, const N: usize>
where
M: RawMutex,
{
buf: Buffer<N>,
inner: Mutex<M, RefCell<PipeState<N>>>,
}
impl<M, const N: usize> Pipe<M, N>
where
M: RawMutex,
{
/// Establish a new bounded pipe. For example, to create one with a NoopMutex:
///
/// ```
/// use embassy_sync::pipe::Pipe;
/// use embassy_sync::blocking_mutex::raw::NoopRawMutex;
///
/// // Declare a bounded pipe, with a buffer of 256 bytes.
/// let mut pipe = Pipe::<NoopRawMutex, 256>::new();
/// ```
pub const fn new() -> Self {
Self {
buf: Buffer(UnsafeCell::new([0; N])),
inner: Mutex::new(RefCell::new(PipeState {
buffer: RingBuffer::new(),
read_waker: WakerRegistration::new(),
write_waker: WakerRegistration::new(),
})),
}
}
fn lock<R>(&self, f: impl FnOnce(&mut PipeState<N>) -> R) -> R {
self.inner.lock(|rc| f(&mut *rc.borrow_mut()))
}
fn try_read_with_context(&self, cx: Option<&mut Context<'_>>, buf: &mut [u8]) -> Result<usize, TryReadError> {
self.inner.lock(|rc: &RefCell<PipeState<N>>| {
let s = &mut *rc.borrow_mut();
if s.buffer.is_full() {
s.write_waker.wake();
}
let available = unsafe { self.buf.get(s.buffer.pop_buf()) };
if available.is_empty() {
if let Some(cx) = cx {
s.read_waker.register(cx.waker());
}
return Err(TryReadError::Empty);
}
let n = available.len().min(buf.len());
buf[..n].copy_from_slice(&available[..n]);
s.buffer.pop(n);
Ok(n)
})
}
// safety: While the returned slice is alive,
// no `read` or `consume` methods in the pipe must be called.
unsafe fn try_fill_buf_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<&[u8], TryReadError> {
self.inner.lock(|rc: &RefCell<PipeState<N>>| {
let s = &mut *rc.borrow_mut();
if s.buffer.is_full() {
s.write_waker.wake();
}
let available = unsafe { self.buf.get(s.buffer.pop_buf()) };
if available.is_empty() {
if let Some(cx) = cx {
s.read_waker.register(cx.waker());
}
return Err(TryReadError::Empty);
}
Ok(available)
})
}
fn consume(&self, amt: usize) {
self.inner.lock(|rc: &RefCell<PipeState<N>>| {
let s = &mut *rc.borrow_mut();
let available = s.buffer.pop_buf();
assert!(amt <= available.len());
s.buffer.pop(amt);
})
}
fn try_write_with_context(&self, cx: Option<&mut Context<'_>>, buf: &[u8]) -> Result<usize, TryWriteError> {
self.inner.lock(|rc: &RefCell<PipeState<N>>| {
let s = &mut *rc.borrow_mut();
if s.buffer.is_empty() {
s.read_waker.wake();
}
let available = unsafe { self.buf.get_mut(s.buffer.push_buf()) };
if available.is_empty() {
if let Some(cx) = cx {
s.write_waker.register(cx.waker());
}
return Err(TryWriteError::Full);
}
let n = available.len().min(buf.len());
available[..n].copy_from_slice(&buf[..n]);
s.buffer.push(n);
Ok(n)
})
}
/// Split this pipe into a BufRead-capable reader and a writer.
///
/// The reader and writer borrow the current pipe mutably, so it is not
/// possible to use it directly while they exist. This is needed because
/// implementing `BufRead` requires there is a single reader.
///
/// The writer is cloneable, the reader is not.
pub fn split(&mut self) -> (Reader<'_, M, N>, Writer<'_, M, N>) {
(Reader { pipe: self }, Writer { pipe: self })
}
/// Write some bytes to the pipe.
///
/// This method writes a nonzero amount of bytes from `buf` into the pipe, and
/// returns the amount of bytes written.
///
/// If it is not possible to write a nonzero amount of bytes because the pipe's buffer is full,
/// this method will wait until it isn't. See [`try_write`](Self::try_write) for a variant that
/// returns an error instead of waiting.
///
/// It is not guaranteed that all bytes in the buffer are written, even if there's enough
/// free space in the pipe buffer for all. In other words, it is possible for `write` to return
/// without writing all of `buf` (returning a number less than `buf.len()`) and still leave
/// free space in the pipe buffer. You should always `write` in a loop, or use helpers like
/// `write_all` from the `embedded-io` crate.
pub fn write<'a>(&'a self, buf: &'a [u8]) -> WriteFuture<'a, M, N> {
WriteFuture { pipe: self, buf }
}
/// Write all bytes to the pipe.
///
/// This method writes all bytes from `buf` into the pipe
pub async fn write_all(&self, mut buf: &[u8]) {
while !buf.is_empty() {
let n = self.write(buf).await;
buf = &buf[n..];
}
}
/// Attempt to immediately write some bytes to the pipe.
///
/// This method will either write a nonzero amount of bytes to the pipe immediately,
/// or return an error if the pipe is empty. See [`write`](Self::write) for a variant
/// that waits instead of returning an error.
pub fn try_write(&self, buf: &[u8]) -> Result<usize, TryWriteError> {
self.try_write_with_context(None, buf)
}
/// Read some bytes from the pipe.
///
/// This method reads a nonzero amount of bytes from the pipe into `buf` and
/// returns the amount of bytes read.
///
/// If it is not possible to read a nonzero amount of bytes because the pipe's buffer is empty,
/// this method will wait until it isn't. See [`try_read`](Self::try_read) for a variant that
/// returns an error instead of waiting.
///
/// It is not guaranteed that all bytes in the buffer are read, even if there's enough
/// space in `buf` for all. In other words, it is possible for `read` to return
/// without filling `buf` (returning a number less than `buf.len()`) and still leave bytes
/// in the pipe buffer. You should always `read` in a loop, or use helpers like
/// `read_exact` from the `embedded-io` crate.
pub fn read<'a>(&'a self, buf: &'a mut [u8]) -> ReadFuture<'a, M, N> {
ReadFuture { pipe: self, buf }
}
/// Attempt to immediately read some bytes from the pipe.
///
/// This method will either read a nonzero amount of bytes from the pipe immediately,
/// or return an error if the pipe is empty. See [`read`](Self::read) for a variant
/// that waits instead of returning an error.
pub fn try_read(&self, buf: &mut [u8]) -> Result<usize, TryReadError> {
self.try_read_with_context(None, buf)
}
/// Clear the data in the pipe's buffer.
pub fn clear(&self) {
self.inner.lock(|rc: &RefCell<PipeState<N>>| {
let s = &mut *rc.borrow_mut();
s.buffer.clear();
s.write_waker.wake();
})
}
/// Return whether the pipe is full (no free space in the buffer)
pub fn is_full(&self) -> bool {
self.len() == N
}
/// Return whether the pipe is empty (no data buffered)
pub fn is_empty(&self) -> bool {
self.len() == 0
}
/// Total byte capacity.
///
/// This is the same as the `N` generic param.
pub fn capacity(&self) -> usize {
N
}
/// Used byte capacity.
pub fn len(&self) -> usize {
self.lock(|c| c.buffer.len())
}
/// Free byte capacity.
///
/// This is equivalent to `capacity() - len()`
pub fn free_capacity(&self) -> usize {
N - self.len()
}
}
impl<M: RawMutex, const N: usize> embedded_io_async::ErrorType for Pipe<M, N> {
type Error = Infallible;
}
impl<M: RawMutex, const N: usize> embedded_io_async::Read for Pipe<M, N> {
async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Self::Error> {
Ok(Pipe::read(self, buf).await)
}
}
impl<M: RawMutex, const N: usize> embedded_io_async::Write for Pipe<M, N> {
async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
Ok(Pipe::write(self, buf).await)
}
async fn flush(&mut self) -> Result<(), Self::Error> {
Ok(())
}
}
impl<M: RawMutex, const N: usize> embedded_io_async::ErrorType for &Pipe<M, N> {
type Error = Infallible;
}
impl<M: RawMutex, const N: usize> embedded_io_async::Read for &Pipe<M, N> {
async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Self::Error> {
Ok(Pipe::read(self, buf).await)
}
}
impl<M: RawMutex, const N: usize> embedded_io_async::Write for &Pipe<M, N> {
async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
Ok(Pipe::write(self, buf).await)
}
async fn flush(&mut self) -> Result<(), Self::Error> {
Ok(())
}
}
impl<M: RawMutex, const N: usize> embedded_io_async::ErrorType for Reader<'_, M, N> {
type Error = Infallible;
}
impl<M: RawMutex, const N: usize> embedded_io_async::Read for Reader<'_, M, N> {
async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Self::Error> {
Ok(Reader::read(self, buf).await)
}
}
impl<M: RawMutex, const N: usize> embedded_io_async::BufRead for Reader<'_, M, N> {
async fn fill_buf(&mut self) -> Result<&[u8], Self::Error> {
Ok(Reader::fill_buf(self).await)
}
fn consume(&mut self, amt: usize) {
Reader::consume(self, amt)
}
}
impl<M: RawMutex, const N: usize> embedded_io_async::ErrorType for Writer<'_, M, N> {
type Error = Infallible;
}
impl<M: RawMutex, const N: usize> embedded_io_async::Write for Writer<'_, M, N> {
async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
Ok(Writer::write(self, buf).await)
}
async fn flush(&mut self) -> Result<(), Self::Error> {
Ok(())
}
}
#[cfg(test)]
mod tests {
use futures_executor::ThreadPool;
use futures_util::task::SpawnExt;
use static_cell::StaticCell;
use super::*;
use crate::blocking_mutex::raw::{CriticalSectionRawMutex, NoopRawMutex};
#[test]
fn writing_once() {
let c = Pipe::<NoopRawMutex, 3>::new();
assert!(c.try_write(&[1]).is_ok());
assert_eq!(c.free_capacity(), 2);
}
#[test]
fn writing_when_full() {
let c = Pipe::<NoopRawMutex, 3>::new();
assert_eq!(c.try_write(&[42]), Ok(1));
assert_eq!(c.try_write(&[43]), Ok(1));
assert_eq!(c.try_write(&[44]), Ok(1));
assert_eq!(c.try_write(&[45]), Err(TryWriteError::Full));
assert_eq!(c.free_capacity(), 0);
}
#[test]
fn receiving_once_with_one_send() {
let c = Pipe::<NoopRawMutex, 3>::new();
assert!(c.try_write(&[42]).is_ok());
let mut buf = [0; 16];
assert_eq!(c.try_read(&mut buf), Ok(1));
assert_eq!(buf[0], 42);
assert_eq!(c.free_capacity(), 3);
}
#[test]
fn receiving_when_empty() {
let c = Pipe::<NoopRawMutex, 3>::new();
let mut buf = [0; 16];
assert_eq!(c.try_read(&mut buf), Err(TryReadError::Empty));
assert_eq!(c.free_capacity(), 3);
}
#[test]
fn simple_send_and_receive() {
let c = Pipe::<NoopRawMutex, 3>::new();
assert!(c.try_write(&[42]).is_ok());
let mut buf = [0; 16];
assert_eq!(c.try_read(&mut buf), Ok(1));
assert_eq!(buf[0], 42);
}
#[test]
fn read_buf() {
let mut c = Pipe::<NoopRawMutex, 3>::new();
let (mut r, w) = c.split();
assert!(w.try_write(&[42, 43]).is_ok());
let buf = r.try_fill_buf().unwrap();
assert_eq!(buf, &[42, 43]);
let buf = r.try_fill_buf().unwrap();
assert_eq!(buf, &[42, 43]);
r.consume(1);
let buf = r.try_fill_buf().unwrap();
assert_eq!(buf, &[43]);
r.consume(1);
assert_eq!(r.try_fill_buf(), Err(TryReadError::Empty));
assert_eq!(w.try_write(&[44, 45, 46]), Ok(1));
assert_eq!(w.try_write(&[45, 46]), Ok(2));
let buf = r.try_fill_buf().unwrap();
assert_eq!(buf, &[44]); // only one byte due to wraparound.
r.consume(1);
let buf = r.try_fill_buf().unwrap();
assert_eq!(buf, &[45, 46]);
assert!(w.try_write(&[47]).is_ok());
let buf = r.try_fill_buf().unwrap();
assert_eq!(buf, &[45, 46, 47]);
r.consume(3);
}
#[test]
fn writer_is_cloneable() {
let mut c = Pipe::<NoopRawMutex, 3>::new();
let (_r, w) = c.split();
let _ = w.clone();
}
#[futures_test::test]
async fn receiver_receives_given_try_write_async() {
let executor = ThreadPool::new().unwrap();
static CHANNEL: StaticCell<Pipe<CriticalSectionRawMutex, 3>> = StaticCell::new();
let c = &*CHANNEL.init(Pipe::new());
let c2 = c;
let f = async move {
assert_eq!(c2.try_write(&[42]), Ok(1));
};
executor.spawn(f).unwrap();
let mut buf = [0; 16];
assert_eq!(c.read(&mut buf).await, 1);
assert_eq!(buf[0], 42);
}
#[futures_test::test]
async fn sender_send_completes_if_capacity() {
let c = Pipe::<CriticalSectionRawMutex, 1>::new();
c.write(&[42]).await;
let mut buf = [0; 16];
assert_eq!(c.read(&mut buf).await, 1);
assert_eq!(buf[0], 42);
}
}
+613
View File
@@ -0,0 +1,613 @@
//! A queue for sending values between asynchronous tasks.
//!
//! Similar to a [`Channel`](crate::channel::Channel), however [`PriorityChannel`] sifts higher priority items to the front of the queue.
//! Priority is determined by the `Ord` trait. Priority behavior is determined by the [`Kind`](heapless::binary_heap::Kind) parameter of the channel.
use core::cell::RefCell;
use core::future::Future;
use core::pin::Pin;
use core::task::{Context, Poll};
pub use heapless::binary_heap::{Kind, Max, Min};
use heapless::BinaryHeap;
use crate::blocking_mutex::raw::RawMutex;
use crate::blocking_mutex::Mutex;
use crate::channel::{DynamicChannel, DynamicReceiver, DynamicSender, TryReceiveError, TrySendError};
use crate::waitqueue::WakerRegistration;
/// Send-only access to a [`PriorityChannel`].
pub struct Sender<'ch, M, T, K, const N: usize>
where
T: Ord,
K: Kind,
M: RawMutex,
{
channel: &'ch PriorityChannel<M, T, K, N>,
}
impl<'ch, M, T, K, const N: usize> Clone for Sender<'ch, M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
fn clone(&self) -> Self {
Sender { channel: self.channel }
}
}
impl<'ch, M, T, K, const N: usize> Copy for Sender<'ch, M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
}
impl<'ch, M, T, K, const N: usize> Sender<'ch, M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
/// Sends a value.
///
/// See [`PriorityChannel::send()`]
pub fn send(&self, message: T) -> SendFuture<'ch, M, T, K, N> {
self.channel.send(message)
}
/// Attempt to immediately send a message.
///
/// See [`PriorityChannel::send()`]
pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
self.channel.try_send(message)
}
/// Allows a poll_fn to poll until the channel is ready to send
///
/// See [`PriorityChannel::poll_ready_to_send()`]
pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
self.channel.poll_ready_to_send(cx)
}
}
impl<'ch, M, T, K, const N: usize> From<Sender<'ch, M, T, K, N>> for DynamicSender<'ch, T>
where
T: Ord,
K: Kind,
M: RawMutex,
{
fn from(s: Sender<'ch, M, T, K, N>) -> Self {
Self { channel: s.channel }
}
}
/// Receive-only access to a [`PriorityChannel`].
pub struct Receiver<'ch, M, T, K, const N: usize>
where
T: Ord,
K: Kind,
M: RawMutex,
{
channel: &'ch PriorityChannel<M, T, K, N>,
}
impl<'ch, M, T, K, const N: usize> Clone for Receiver<'ch, M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
fn clone(&self) -> Self {
Receiver { channel: self.channel }
}
}
impl<'ch, M, T, K, const N: usize> Copy for Receiver<'ch, M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
}
impl<'ch, M, T, K, const N: usize> Receiver<'ch, M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
/// Receive the next value.
///
/// See [`PriorityChannel::receive()`].
pub fn receive(&self) -> ReceiveFuture<'_, M, T, K, N> {
self.channel.receive()
}
/// Attempt to immediately receive the next value.
///
/// See [`PriorityChannel::try_receive()`]
pub fn try_receive(&self) -> Result<T, TryReceiveError> {
self.channel.try_receive()
}
/// Allows a poll_fn to poll until the channel is ready to receive
///
/// See [`PriorityChannel::poll_ready_to_receive()`]
pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
self.channel.poll_ready_to_receive(cx)
}
/// Poll the channel for the next item
///
/// See [`PriorityChannel::poll_receive()`]
pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
self.channel.poll_receive(cx)
}
}
impl<'ch, M, T, K, const N: usize> From<Receiver<'ch, M, T, K, N>> for DynamicReceiver<'ch, T>
where
T: Ord,
K: Kind,
M: RawMutex,
{
fn from(s: Receiver<'ch, M, T, K, N>) -> Self {
Self { channel: s.channel }
}
}
/// Future returned by [`PriorityChannel::receive`] and [`Receiver::receive`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct ReceiveFuture<'ch, M, T, K, const N: usize>
where
T: Ord,
K: Kind,
M: RawMutex,
{
channel: &'ch PriorityChannel<M, T, K, N>,
}
impl<'ch, M, T, K, const N: usize> Future for ReceiveFuture<'ch, M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
type Output = T;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
self.channel.poll_receive(cx)
}
}
/// Future returned by [`PriorityChannel::send`] and [`Sender::send`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct SendFuture<'ch, M, T, K, const N: usize>
where
T: Ord,
K: Kind,
M: RawMutex,
{
channel: &'ch PriorityChannel<M, T, K, N>,
message: Option<T>,
}
impl<'ch, M, T, K, const N: usize> Future for SendFuture<'ch, M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.message.take() {
Some(m) => match self.channel.try_send_with_context(m, Some(cx)) {
Ok(..) => Poll::Ready(()),
Err(TrySendError::Full(m)) => {
self.message = Some(m);
Poll::Pending
}
},
None => panic!("Message cannot be None"),
}
}
}
impl<'ch, M, T, K, const N: usize> Unpin for SendFuture<'ch, M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
}
struct ChannelState<T, K, const N: usize> {
queue: BinaryHeap<T, K, N>,
receiver_waker: WakerRegistration,
senders_waker: WakerRegistration,
}
impl<T, K, const N: usize> ChannelState<T, K, N>
where
T: Ord,
K: Kind,
{
const fn new() -> Self {
ChannelState {
queue: BinaryHeap::new(),
receiver_waker: WakerRegistration::new(),
senders_waker: WakerRegistration::new(),
}
}
fn try_receive(&mut self) -> Result<T, TryReceiveError> {
self.try_receive_with_context(None)
}
fn try_receive_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
if self.queue.len() == self.queue.capacity() {
self.senders_waker.wake();
}
if let Some(message) = self.queue.pop() {
Ok(message)
} else {
if let Some(cx) = cx {
self.receiver_waker.register(cx.waker());
}
Err(TryReceiveError::Empty)
}
}
fn poll_receive(&mut self, cx: &mut Context<'_>) -> Poll<T> {
if self.queue.len() == self.queue.capacity() {
self.senders_waker.wake();
}
if let Some(message) = self.queue.pop() {
Poll::Ready(message)
} else {
self.receiver_waker.register(cx.waker());
Poll::Pending
}
}
fn poll_ready_to_receive(&mut self, cx: &mut Context<'_>) -> Poll<()> {
self.receiver_waker.register(cx.waker());
if !self.queue.is_empty() {
Poll::Ready(())
} else {
Poll::Pending
}
}
fn try_send(&mut self, message: T) -> Result<(), TrySendError<T>> {
self.try_send_with_context(message, None)
}
fn try_send_with_context(&mut self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
match self.queue.push(message) {
Ok(()) => {
self.receiver_waker.wake();
Ok(())
}
Err(message) => {
if let Some(cx) = cx {
self.senders_waker.register(cx.waker());
}
Err(TrySendError::Full(message))
}
}
}
fn poll_ready_to_send(&mut self, cx: &mut Context<'_>) -> Poll<()> {
self.senders_waker.register(cx.waker());
if !self.queue.len() == self.queue.capacity() {
Poll::Ready(())
} else {
Poll::Pending
}
}
}
/// A bounded channel for communicating between asynchronous tasks
/// with backpressure.
///
/// The channel will buffer up to the provided number of messages. Once the
/// buffer is full, attempts to `send` new messages will wait until a message is
/// received from the channel.
///
/// Sent data may be reordered based on their priorty within the channel.
/// For example, in a [`Max`](heapless::binary_heap::Max) [`PriorityChannel`]
/// containing `u32`'s, data sent in the following order `[1, 2, 3]` will be recieved as `[3, 2, 1]`.
pub struct PriorityChannel<M, T, K, const N: usize>
where
T: Ord,
K: Kind,
M: RawMutex,
{
inner: Mutex<M, RefCell<ChannelState<T, K, N>>>,
}
impl<M, T, K, const N: usize> PriorityChannel<M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
/// Establish a new bounded channel. For example, to create one with a NoopMutex:
///
/// ```
/// use embassy_sync::priority_channel::{PriorityChannel, Max};
/// use embassy_sync::blocking_mutex::raw::NoopRawMutex;
///
/// // Declare a bounded channel of 3 u32s.
/// let mut channel = PriorityChannel::<NoopRawMutex, u32, Max, 3>::new();
/// ```
pub const fn new() -> Self {
Self {
inner: Mutex::new(RefCell::new(ChannelState::new())),
}
}
fn lock<R>(&self, f: impl FnOnce(&mut ChannelState<T, K, N>) -> R) -> R {
self.inner.lock(|rc| f(&mut *unwrap!(rc.try_borrow_mut())))
}
fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
self.lock(|c| c.try_receive_with_context(cx))
}
/// Poll the channel for the next message
pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
self.lock(|c| c.poll_receive(cx))
}
fn try_send_with_context(&self, m: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
self.lock(|c| c.try_send_with_context(m, cx))
}
/// Allows a poll_fn to poll until the channel is ready to receive
pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
self.lock(|c| c.poll_ready_to_receive(cx))
}
/// Allows a poll_fn to poll until the channel is ready to send
pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
self.lock(|c| c.poll_ready_to_send(cx))
}
/// Get a sender for this channel.
pub fn sender(&self) -> Sender<'_, M, T, K, N> {
Sender { channel: self }
}
/// Get a receiver for this channel.
pub fn receiver(&self) -> Receiver<'_, M, T, K, N> {
Receiver { channel: self }
}
/// Send a value, waiting until there is capacity.
///
/// Sending completes when the value has been pushed to the channel's queue.
/// This doesn't mean the value has been received yet.
pub fn send(&self, message: T) -> SendFuture<'_, M, T, K, N> {
SendFuture {
channel: self,
message: Some(message),
}
}
/// Attempt to immediately send a message.
///
/// This method differs from [`send`](PriorityChannel::send) by returning immediately if the channel's
/// buffer is full, instead of waiting.
///
/// # Errors
///
/// If the channel capacity has been reached, i.e., the channel has `n`
/// buffered values where `n` is the argument passed to [`PriorityChannel`], then an
/// error is returned.
pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
self.lock(|c| c.try_send(message))
}
/// Receive the next value.
///
/// If there are no messages in the channel's buffer, this method will
/// wait until a message is sent.
pub fn receive(&self) -> ReceiveFuture<'_, M, T, K, N> {
ReceiveFuture { channel: self }
}
/// Attempt to immediately receive a message.
///
/// This method will either receive a message from the channel immediately or return an error
/// if the channel is empty.
pub fn try_receive(&self) -> Result<T, TryReceiveError> {
self.lock(|c| c.try_receive())
}
}
/// Implements the DynamicChannel to allow creating types that are unaware of the queue size with the
/// tradeoff cost of dynamic dispatch.
impl<M, T, K, const N: usize> DynamicChannel<T> for PriorityChannel<M, T, K, N>
where
T: Ord,
K: Kind,
M: RawMutex,
{
fn try_send_with_context(&self, m: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
PriorityChannel::try_send_with_context(self, m, cx)
}
fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
PriorityChannel::try_receive_with_context(self, cx)
}
fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
PriorityChannel::poll_ready_to_send(self, cx)
}
fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
PriorityChannel::poll_ready_to_receive(self, cx)
}
fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
PriorityChannel::poll_receive(self, cx)
}
}
#[cfg(test)]
mod tests {
use core::time::Duration;
use futures_executor::ThreadPool;
use futures_timer::Delay;
use futures_util::task::SpawnExt;
use heapless::binary_heap::{Kind, Max};
use static_cell::StaticCell;
use super::*;
use crate::blocking_mutex::raw::{CriticalSectionRawMutex, NoopRawMutex};
fn capacity<T, K, const N: usize>(c: &ChannelState<T, K, N>) -> usize
where
T: Ord,
K: Kind,
{
c.queue.capacity() - c.queue.len()
}
#[test]
fn sending_once() {
let mut c = ChannelState::<u32, Max, 3>::new();
assert!(c.try_send(1).is_ok());
assert_eq!(capacity(&c), 2);
}
#[test]
fn sending_when_full() {
let mut c = ChannelState::<u32, Max, 3>::new();
let _ = c.try_send(1);
let _ = c.try_send(1);
let _ = c.try_send(1);
match c.try_send(2) {
Err(TrySendError::Full(2)) => assert!(true),
_ => assert!(false),
}
assert_eq!(capacity(&c), 0);
}
#[test]
fn send_priority() {
// Prio channel with kind `Max` sifts larger numbers to the front of the queue
let mut c = ChannelState::<u32, Max, 3>::new();
assert!(c.try_send(1).is_ok());
assert!(c.try_send(2).is_ok());
assert!(c.try_send(3).is_ok());
assert_eq!(c.try_receive().unwrap(), 3);
assert_eq!(c.try_receive().unwrap(), 2);
assert_eq!(c.try_receive().unwrap(), 1);
}
#[test]
fn receiving_once_with_one_send() {
let mut c = ChannelState::<u32, Max, 3>::new();
assert!(c.try_send(1).is_ok());
assert_eq!(c.try_receive().unwrap(), 1);
assert_eq!(capacity(&c), 3);
}
#[test]
fn receiving_when_empty() {
let mut c = ChannelState::<u32, Max, 3>::new();
match c.try_receive() {
Err(TryReceiveError::Empty) => assert!(true),
_ => assert!(false),
}
assert_eq!(capacity(&c), 3);
}
#[test]
fn simple_send_and_receive() {
let c = PriorityChannel::<NoopRawMutex, u32, Max, 3>::new();
assert!(c.try_send(1).is_ok());
assert_eq!(c.try_receive().unwrap(), 1);
}
#[test]
fn cloning() {
let c = PriorityChannel::<NoopRawMutex, u32, Max, 3>::new();
let r1 = c.receiver();
let s1 = c.sender();
let _ = r1.clone();
let _ = s1.clone();
}
#[test]
fn dynamic_dispatch() {
let c = PriorityChannel::<NoopRawMutex, u32, Max, 3>::new();
let s: DynamicSender<'_, u32> = c.sender().into();
let r: DynamicReceiver<'_, u32> = c.receiver().into();
assert!(s.try_send(1).is_ok());
assert_eq!(r.try_receive().unwrap(), 1);
}
#[futures_test::test]
async fn receiver_receives_given_try_send_async() {
let executor = ThreadPool::new().unwrap();
static CHANNEL: StaticCell<PriorityChannel<CriticalSectionRawMutex, u32, Max, 3>> = StaticCell::new();
let c = &*CHANNEL.init(PriorityChannel::new());
let c2 = c;
assert!(executor
.spawn(async move {
assert!(c2.try_send(1).is_ok());
})
.is_ok());
assert_eq!(c.receive().await, 1);
}
#[futures_test::test]
async fn sender_send_completes_if_capacity() {
let c = PriorityChannel::<CriticalSectionRawMutex, u32, Max, 1>::new();
c.send(1).await;
assert_eq!(c.receive().await, 1);
}
#[futures_test::test]
async fn senders_sends_wait_until_capacity() {
let executor = ThreadPool::new().unwrap();
static CHANNEL: StaticCell<PriorityChannel<CriticalSectionRawMutex, u32, Max, 1>> = StaticCell::new();
let c = &*CHANNEL.init(PriorityChannel::new());
assert!(c.try_send(1).is_ok());
let c2 = c;
let send_task_1 = executor.spawn_with_handle(async move { c2.send(2).await });
let c2 = c;
let send_task_2 = executor.spawn_with_handle(async move { c2.send(3).await });
// Wish I could think of a means of determining that the async send is waiting instead.
// However, I've used the debugger to observe that the send does indeed wait.
Delay::new(Duration::from_millis(500)).await;
assert_eq!(c.receive().await, 1);
assert!(executor
.spawn(async move {
loop {
c.receive().await;
}
})
.is_ok());
send_task_1.unwrap().await;
send_task_2.unwrap().await;
}
}
+658
View File
@@ -0,0 +1,658 @@
//! Implementation of [PubSubChannel], a queue where published messages get received by all subscribers.
#![deny(missing_docs)]
use core::cell::RefCell;
use core::fmt::Debug;
use core::task::{Context, Poll};
use heapless::Deque;
use self::publisher::{ImmediatePub, Pub};
use self::subscriber::Sub;
use crate::blocking_mutex::raw::RawMutex;
use crate::blocking_mutex::Mutex;
use crate::waitqueue::MultiWakerRegistration;
pub mod publisher;
pub mod subscriber;
pub use publisher::{DynImmediatePublisher, DynPublisher, ImmediatePublisher, Publisher};
pub use subscriber::{DynSubscriber, Subscriber};
/// A broadcast channel implementation where multiple publishers can send messages to multiple subscribers
///
/// Any published message can be read by all subscribers.
/// A publisher can choose how it sends its message.
///
/// - With [Pub::publish()] the publisher has to wait until there is space in the internal message queue.
/// - With [Pub::publish_immediate()] the publisher doesn't await and instead lets the oldest message
/// in the queue drop if necessary. This will cause any [Subscriber] that missed the message to receive
/// an error to indicate that it has lagged.
///
/// ## Example
///
/// ```
/// # use embassy_sync::blocking_mutex::raw::NoopRawMutex;
/// # use embassy_sync::pubsub::WaitResult;
/// # use embassy_sync::pubsub::PubSubChannel;
/// # use futures_executor::block_on;
/// # let test = async {
/// // Create the channel. This can be static as well
/// let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
///
/// // This is a generic subscriber with a direct reference to the channel
/// let mut sub0 = channel.subscriber().unwrap();
/// // This is a dynamic subscriber with a dynamic (trait object) reference to the channel
/// let mut sub1 = channel.dyn_subscriber().unwrap();
///
/// let pub0 = channel.publisher().unwrap();
///
/// // Publish a message, but wait if the queue is full
/// pub0.publish(42).await;
///
/// // Publish a message, but if the queue is full, just kick out the oldest message.
/// // This may cause some subscribers to miss a message
/// pub0.publish_immediate(43);
///
/// // Wait for a new message. If the subscriber missed a message, the WaitResult will be a Lag result
/// assert_eq!(sub0.next_message().await, WaitResult::Message(42));
/// assert_eq!(sub1.next_message().await, WaitResult::Message(42));
///
/// // Wait again, but this time ignore any Lag results
/// assert_eq!(sub0.next_message_pure().await, 43);
/// assert_eq!(sub1.next_message_pure().await, 43);
///
/// // There's also a polling interface
/// assert_eq!(sub0.try_next_message(), None);
/// assert_eq!(sub1.try_next_message(), None);
/// # };
/// #
/// # block_on(test);
/// ```
///
pub struct PubSubChannel<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> {
inner: Mutex<M, RefCell<PubSubState<T, CAP, SUBS, PUBS>>>,
}
impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize>
PubSubChannel<M, T, CAP, SUBS, PUBS>
{
/// Create a new channel
pub const fn new() -> Self {
Self {
inner: Mutex::const_new(M::INIT, RefCell::new(PubSubState::new())),
}
}
/// Create a new subscriber. It will only receive messages that are published after its creation.
///
/// If there are no subscriber slots left, an error will be returned.
pub fn subscriber(&self) -> Result<Subscriber<M, T, CAP, SUBS, PUBS>, Error> {
self.inner.lock(|inner| {
let mut s = inner.borrow_mut();
if s.subscriber_count >= SUBS {
Err(Error::MaximumSubscribersReached)
} else {
s.subscriber_count += 1;
Ok(Subscriber(Sub::new(s.next_message_id, self)))
}
})
}
/// Create a new subscriber. It will only receive messages that are published after its creation.
///
/// If there are no subscriber slots left, an error will be returned.
pub fn dyn_subscriber(&self) -> Result<DynSubscriber<'_, T>, Error> {
self.inner.lock(|inner| {
let mut s = inner.borrow_mut();
if s.subscriber_count >= SUBS {
Err(Error::MaximumSubscribersReached)
} else {
s.subscriber_count += 1;
Ok(DynSubscriber(Sub::new(s.next_message_id, self)))
}
})
}
/// Create a new publisher
///
/// If there are no publisher slots left, an error will be returned.
pub fn publisher(&self) -> Result<Publisher<M, T, CAP, SUBS, PUBS>, Error> {
self.inner.lock(|inner| {
let mut s = inner.borrow_mut();
if s.publisher_count >= PUBS {
Err(Error::MaximumPublishersReached)
} else {
s.publisher_count += 1;
Ok(Publisher(Pub::new(self)))
}
})
}
/// Create a new publisher
///
/// If there are no publisher slots left, an error will be returned.
pub fn dyn_publisher(&self) -> Result<DynPublisher<'_, T>, Error> {
self.inner.lock(|inner| {
let mut s = inner.borrow_mut();
if s.publisher_count >= PUBS {
Err(Error::MaximumPublishersReached)
} else {
s.publisher_count += 1;
Ok(DynPublisher(Pub::new(self)))
}
})
}
/// Create a new publisher that can only send immediate messages.
/// This kind of publisher does not take up a publisher slot.
pub fn immediate_publisher(&self) -> ImmediatePublisher<M, T, CAP, SUBS, PUBS> {
ImmediatePublisher(ImmediatePub::new(self))
}
/// Create a new publisher that can only send immediate messages.
/// This kind of publisher does not take up a publisher slot.
pub fn dyn_immediate_publisher(&self) -> DynImmediatePublisher<T> {
DynImmediatePublisher(ImmediatePub::new(self))
}
}
impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> PubSubBehavior<T>
for PubSubChannel<M, T, CAP, SUBS, PUBS>
{
fn get_message_with_context(&self, next_message_id: &mut u64, cx: Option<&mut Context<'_>>) -> Poll<WaitResult<T>> {
self.inner.lock(|s| {
let mut s = s.borrow_mut();
// Check if we can read a message
match s.get_message(*next_message_id) {
// Yes, so we are done polling
Some(WaitResult::Message(message)) => {
*next_message_id += 1;
Poll::Ready(WaitResult::Message(message))
}
// No, so we need to reregister our waker and sleep again
None => {
if let Some(cx) = cx {
s.subscriber_wakers.register(cx.waker());
}
Poll::Pending
}
// We missed a couple of messages. We must do our internal bookkeeping and return that we lagged
Some(WaitResult::Lagged(amount)) => {
*next_message_id += amount;
Poll::Ready(WaitResult::Lagged(amount))
}
}
})
}
fn available(&self, next_message_id: u64) -> u64 {
self.inner.lock(|s| s.borrow().next_message_id - next_message_id)
}
fn publish_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), T> {
self.inner.lock(|s| {
let mut s = s.borrow_mut();
// Try to publish the message
match s.try_publish(message) {
// We did it, we are ready
Ok(()) => Ok(()),
// The queue is full, so we need to reregister our waker and go to sleep
Err(message) => {
if let Some(cx) = cx {
s.publisher_wakers.register(cx.waker());
}
Err(message)
}
}
})
}
fn publish_immediate(&self, message: T) {
self.inner.lock(|s| {
let mut s = s.borrow_mut();
s.publish_immediate(message)
})
}
fn space(&self) -> usize {
self.inner.lock(|s| {
let s = s.borrow();
s.queue.capacity() - s.queue.len()
})
}
fn unregister_subscriber(&self, subscriber_next_message_id: u64) {
self.inner.lock(|s| {
let mut s = s.borrow_mut();
s.unregister_subscriber(subscriber_next_message_id)
})
}
fn unregister_publisher(&self) {
self.inner.lock(|s| {
let mut s = s.borrow_mut();
s.unregister_publisher()
})
}
}
/// Internal state for the PubSub channel
struct PubSubState<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> {
/// The queue contains the last messages that have been published and a countdown of how many subscribers are yet to read it
queue: Deque<(T, usize), CAP>,
/// Every message has an id.
/// Don't worry, we won't run out.
/// If a million messages were published every second, then the ID's would run out in about 584942 years.
next_message_id: u64,
/// Collection of wakers for Subscribers that are waiting.
subscriber_wakers: MultiWakerRegistration<SUBS>,
/// Collection of wakers for Publishers that are waiting.
publisher_wakers: MultiWakerRegistration<PUBS>,
/// The amount of subscribers that are active
subscriber_count: usize,
/// The amount of publishers that are active
publisher_count: usize,
}
impl<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> PubSubState<T, CAP, SUBS, PUBS> {
/// Create a new internal channel state
const fn new() -> Self {
Self {
queue: Deque::new(),
next_message_id: 0,
subscriber_wakers: MultiWakerRegistration::new(),
publisher_wakers: MultiWakerRegistration::new(),
subscriber_count: 0,
publisher_count: 0,
}
}
fn try_publish(&mut self, message: T) -> Result<(), T> {
if self.subscriber_count == 0 {
// We don't need to publish anything because there is no one to receive it
return Ok(());
}
if self.queue.is_full() {
return Err(message);
}
// We just did a check for this
self.queue.push_back((message, self.subscriber_count)).ok().unwrap();
self.next_message_id += 1;
// Wake all of the subscribers
self.subscriber_wakers.wake();
Ok(())
}
fn publish_immediate(&mut self, message: T) {
// Make space in the queue if required
if self.queue.is_full() {
self.queue.pop_front();
}
// This will succeed because we made sure there is space
self.try_publish(message).ok().unwrap();
}
fn get_message(&mut self, message_id: u64) -> Option<WaitResult<T>> {
let start_id = self.next_message_id - self.queue.len() as u64;
if message_id < start_id {
return Some(WaitResult::Lagged(start_id - message_id));
}
let current_message_index = (message_id - start_id) as usize;
if current_message_index >= self.queue.len() {
return None;
}
// We've checked that the index is valid
let queue_item = self.queue.iter_mut().nth(current_message_index).unwrap();
// We're reading this item, so decrement the counter
queue_item.1 -= 1;
let message = if current_message_index == 0 && queue_item.1 == 0 {
let (message, _) = self.queue.pop_front().unwrap();
self.publisher_wakers.wake();
// Return pop'd message without clone
message
} else {
queue_item.0.clone()
};
Some(WaitResult::Message(message))
}
fn unregister_subscriber(&mut self, subscriber_next_message_id: u64) {
self.subscriber_count -= 1;
// All messages that haven't been read yet by this subscriber must have their counter decremented
let start_id = self.next_message_id - self.queue.len() as u64;
if subscriber_next_message_id >= start_id {
let current_message_index = (subscriber_next_message_id - start_id) as usize;
self.queue
.iter_mut()
.skip(current_message_index)
.for_each(|(_, counter)| *counter -= 1);
let mut wake_publishers = false;
while let Some((_, count)) = self.queue.front() {
if *count == 0 {
self.queue.pop_front().unwrap();
wake_publishers = true;
} else {
break;
}
}
if wake_publishers {
self.publisher_wakers.wake();
}
}
}
fn unregister_publisher(&mut self) {
self.publisher_count -= 1;
}
}
/// Error type for the [PubSubChannel]
#[derive(Debug, PartialEq, Eq, Clone)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum Error {
/// All subscriber slots are used. To add another subscriber, first another subscriber must be dropped or
/// the capacity of the channels must be increased.
MaximumSubscribersReached,
/// All publisher slots are used. To add another publisher, first another publisher must be dropped or
/// the capacity of the channels must be increased.
MaximumPublishersReached,
}
/// 'Middle level' behaviour of the pubsub channel.
/// This trait is used so that Sub and Pub can be generic over the channel.
pub trait PubSubBehavior<T> {
/// Try to get a message from the queue with the given message id.
///
/// If the message is not yet present and a context is given, then its waker is registered in the subsriber wakers.
fn get_message_with_context(&self, next_message_id: &mut u64, cx: Option<&mut Context<'_>>) -> Poll<WaitResult<T>>;
/// Get the amount of messages that are between the given the next_message_id and the most recent message.
/// This is not necessarily the amount of messages a subscriber can still received as it may have lagged.
fn available(&self, next_message_id: u64) -> u64;
/// Try to publish a message to the queue.
///
/// If the queue is full and a context is given, then its waker is registered in the publisher wakers.
fn publish_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), T>;
/// Publish a message immediately
fn publish_immediate(&self, message: T);
/// The amount of messages that can still be published without having to wait or without having to lag the subscribers
fn space(&self) -> usize;
/// Let the channel know that a subscriber has dropped
fn unregister_subscriber(&self, subscriber_next_message_id: u64);
/// Let the channel know that a publisher has dropped
fn unregister_publisher(&self);
}
/// The result of the subscriber wait procedure
#[derive(Debug, Clone, PartialEq, Eq)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum WaitResult<T> {
/// The subscriber did not receive all messages and lagged by the given amount of messages.
/// (This is the amount of messages that were missed)
Lagged(u64),
/// A message was received
Message(T),
}
#[cfg(test)]
mod tests {
use super::*;
use crate::blocking_mutex::raw::NoopRawMutex;
#[futures_test::test]
async fn dyn_pub_sub_works() {
let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
let mut sub0 = channel.dyn_subscriber().unwrap();
let mut sub1 = channel.dyn_subscriber().unwrap();
let pub0 = channel.dyn_publisher().unwrap();
pub0.publish(42).await;
assert_eq!(sub0.next_message().await, WaitResult::Message(42));
assert_eq!(sub1.next_message().await, WaitResult::Message(42));
assert_eq!(sub0.try_next_message(), None);
assert_eq!(sub1.try_next_message(), None);
}
#[futures_test::test]
async fn all_subscribers_receive() {
let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
let mut sub0 = channel.subscriber().unwrap();
let mut sub1 = channel.subscriber().unwrap();
let pub0 = channel.publisher().unwrap();
pub0.publish(42).await;
assert_eq!(sub0.next_message().await, WaitResult::Message(42));
assert_eq!(sub1.next_message().await, WaitResult::Message(42));
assert_eq!(sub0.try_next_message(), None);
assert_eq!(sub1.try_next_message(), None);
}
#[futures_test::test]
async fn lag_when_queue_full_on_immediate_publish() {
let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
let mut sub0 = channel.subscriber().unwrap();
let pub0 = channel.publisher().unwrap();
pub0.publish_immediate(42);
pub0.publish_immediate(43);
pub0.publish_immediate(44);
pub0.publish_immediate(45);
pub0.publish_immediate(46);
pub0.publish_immediate(47);
assert_eq!(sub0.try_next_message(), Some(WaitResult::Lagged(2)));
assert_eq!(sub0.next_message().await, WaitResult::Message(44));
assert_eq!(sub0.next_message().await, WaitResult::Message(45));
assert_eq!(sub0.next_message().await, WaitResult::Message(46));
assert_eq!(sub0.next_message().await, WaitResult::Message(47));
assert_eq!(sub0.try_next_message(), None);
}
#[test]
fn limited_subs_and_pubs() {
let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
let sub0 = channel.subscriber();
let sub1 = channel.subscriber();
let sub2 = channel.subscriber();
let sub3 = channel.subscriber();
let sub4 = channel.subscriber();
assert!(sub0.is_ok());
assert!(sub1.is_ok());
assert!(sub2.is_ok());
assert!(sub3.is_ok());
assert_eq!(sub4.err().unwrap(), Error::MaximumSubscribersReached);
drop(sub0);
let sub5 = channel.subscriber();
assert!(sub5.is_ok());
// publishers
let pub0 = channel.publisher();
let pub1 = channel.publisher();
let pub2 = channel.publisher();
let pub3 = channel.publisher();
let pub4 = channel.publisher();
assert!(pub0.is_ok());
assert!(pub1.is_ok());
assert!(pub2.is_ok());
assert!(pub3.is_ok());
assert_eq!(pub4.err().unwrap(), Error::MaximumPublishersReached);
drop(pub0);
let pub5 = channel.publisher();
assert!(pub5.is_ok());
}
#[test]
fn publisher_wait_on_full_queue() {
let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
let pub0 = channel.publisher().unwrap();
// There are no subscribers, so the queue will never be full
assert_eq!(pub0.try_publish(0), Ok(()));
assert_eq!(pub0.try_publish(0), Ok(()));
assert_eq!(pub0.try_publish(0), Ok(()));
assert_eq!(pub0.try_publish(0), Ok(()));
assert_eq!(pub0.try_publish(0), Ok(()));
let sub0 = channel.subscriber().unwrap();
assert_eq!(pub0.try_publish(0), Ok(()));
assert_eq!(pub0.try_publish(0), Ok(()));
assert_eq!(pub0.try_publish(0), Ok(()));
assert_eq!(pub0.try_publish(0), Ok(()));
assert_eq!(pub0.try_publish(0), Err(0));
drop(sub0);
}
#[futures_test::test]
async fn correct_available() {
let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
let sub0 = channel.subscriber().unwrap();
let mut sub1 = channel.subscriber().unwrap();
let pub0 = channel.publisher().unwrap();
assert_eq!(sub0.available(), 0);
assert_eq!(sub1.available(), 0);
pub0.publish(42).await;
assert_eq!(sub0.available(), 1);
assert_eq!(sub1.available(), 1);
sub1.next_message().await;
assert_eq!(sub1.available(), 0);
pub0.publish(42).await;
assert_eq!(sub0.available(), 2);
assert_eq!(sub1.available(), 1);
}
#[futures_test::test]
async fn correct_space() {
let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
let mut sub0 = channel.subscriber().unwrap();
let mut sub1 = channel.subscriber().unwrap();
let pub0 = channel.publisher().unwrap();
assert_eq!(pub0.space(), 4);
pub0.publish(42).await;
assert_eq!(pub0.space(), 3);
pub0.publish(42).await;
assert_eq!(pub0.space(), 2);
sub0.next_message().await;
sub0.next_message().await;
assert_eq!(pub0.space(), 2);
sub1.next_message().await;
assert_eq!(pub0.space(), 3);
sub1.next_message().await;
assert_eq!(pub0.space(), 4);
}
#[futures_test::test]
async fn empty_channel_when_last_subscriber_is_dropped() {
let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
let pub0 = channel.publisher().unwrap();
let mut sub0 = channel.subscriber().unwrap();
let mut sub1 = channel.subscriber().unwrap();
assert_eq!(4, pub0.space());
pub0.publish(1).await;
pub0.publish(2).await;
assert_eq!(2, channel.space());
assert_eq!(1, sub0.try_next_message_pure().unwrap());
assert_eq!(2, sub0.try_next_message_pure().unwrap());
assert_eq!(2, channel.space());
drop(sub0);
assert_eq!(2, channel.space());
assert_eq!(1, sub1.try_next_message_pure().unwrap());
assert_eq!(3, channel.space());
drop(sub1);
assert_eq!(4, channel.space());
}
struct CloneCallCounter(usize);
impl Clone for CloneCallCounter {
fn clone(&self) -> Self {
Self(self.0 + 1)
}
}
#[futures_test::test]
async fn skip_clone_for_last_message() {
let channel = PubSubChannel::<NoopRawMutex, CloneCallCounter, 1, 2, 1>::new();
let pub0 = channel.publisher().unwrap();
let mut sub0 = channel.subscriber().unwrap();
let mut sub1 = channel.subscriber().unwrap();
pub0.publish(CloneCallCounter(0)).await;
assert_eq!(1, sub0.try_next_message_pure().unwrap().0);
assert_eq!(0, sub1.try_next_message_pure().unwrap().0);
}
}
+199
View File
@@ -0,0 +1,199 @@
//! Implementation of anything directly publisher related
use core::future::Future;
use core::marker::PhantomData;
use core::ops::{Deref, DerefMut};
use core::pin::Pin;
use core::task::{Context, Poll};
use super::{PubSubBehavior, PubSubChannel};
use crate::blocking_mutex::raw::RawMutex;
/// A publisher to a channel
pub struct Pub<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
/// The channel we are a publisher for
channel: &'a PSB,
_phantom: PhantomData<T>,
}
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Pub<'a, PSB, T> {
pub(super) fn new(channel: &'a PSB) -> Self {
Self {
channel,
_phantom: Default::default(),
}
}
/// Publish a message right now even when the queue is full.
/// This may cause a subscriber to miss an older message.
pub fn publish_immediate(&self, message: T) {
self.channel.publish_immediate(message)
}
/// Publish a message. But if the message queue is full, wait for all subscribers to have read the last message
pub fn publish<'s>(&'s self, message: T) -> PublisherWaitFuture<'s, 'a, PSB, T> {
PublisherWaitFuture {
message: Some(message),
publisher: self,
}
}
/// Publish a message if there is space in the message queue
pub fn try_publish(&self, message: T) -> Result<(), T> {
self.channel.publish_with_context(message, None)
}
/// The amount of messages that can still be published without having to wait or without having to lag the subscribers
///
/// *Note: In the time between checking this and a publish action, other publishers may have had time to publish something.
/// So checking doesn't give any guarantees.*
pub fn space(&self) -> usize {
self.channel.space()
}
}
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Drop for Pub<'a, PSB, T> {
fn drop(&mut self) {
self.channel.unregister_publisher()
}
}
/// A publisher that holds a dynamic reference to the channel
pub struct DynPublisher<'a, T: Clone>(pub(super) Pub<'a, dyn PubSubBehavior<T> + 'a, T>);
impl<'a, T: Clone> Deref for DynPublisher<'a, T> {
type Target = Pub<'a, dyn PubSubBehavior<T> + 'a, T>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<'a, T: Clone> DerefMut for DynPublisher<'a, T> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
/// A publisher that holds a generic reference to the channel
pub struct Publisher<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize>(
pub(super) Pub<'a, PubSubChannel<M, T, CAP, SUBS, PUBS>, T>,
);
impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> Deref
for Publisher<'a, M, T, CAP, SUBS, PUBS>
{
type Target = Pub<'a, PubSubChannel<M, T, CAP, SUBS, PUBS>, T>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> DerefMut
for Publisher<'a, M, T, CAP, SUBS, PUBS>
{
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
/// A publisher that can only use the `publish_immediate` function, but it doesn't have to be registered with the channel.
/// (So an infinite amount is possible)
pub struct ImmediatePub<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
/// The channel we are a publisher for
channel: &'a PSB,
_phantom: PhantomData<T>,
}
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> ImmediatePub<'a, PSB, T> {
pub(super) fn new(channel: &'a PSB) -> Self {
Self {
channel,
_phantom: Default::default(),
}
}
/// Publish the message right now even when the queue is full.
/// This may cause a subscriber to miss an older message.
pub fn publish_immediate(&self, message: T) {
self.channel.publish_immediate(message)
}
/// Publish a message if there is space in the message queue
pub fn try_publish(&self, message: T) -> Result<(), T> {
self.channel.publish_with_context(message, None)
}
/// The amount of messages that can still be published without having to wait or without having to lag the subscribers
///
/// *Note: In the time between checking this and a publish action, other publishers may have had time to publish something.
/// So checking doesn't give any guarantees.*
pub fn space(&self) -> usize {
self.channel.space()
}
}
/// An immediate publisher that holds a dynamic reference to the channel
pub struct DynImmediatePublisher<'a, T: Clone>(pub(super) ImmediatePub<'a, dyn PubSubBehavior<T> + 'a, T>);
impl<'a, T: Clone> Deref for DynImmediatePublisher<'a, T> {
type Target = ImmediatePub<'a, dyn PubSubBehavior<T> + 'a, T>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<'a, T: Clone> DerefMut for DynImmediatePublisher<'a, T> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
/// An immediate publisher that holds a generic reference to the channel
pub struct ImmediatePublisher<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize>(
pub(super) ImmediatePub<'a, PubSubChannel<M, T, CAP, SUBS, PUBS>, T>,
);
impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> Deref
for ImmediatePublisher<'a, M, T, CAP, SUBS, PUBS>
{
type Target = ImmediatePub<'a, PubSubChannel<M, T, CAP, SUBS, PUBS>, T>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> DerefMut
for ImmediatePublisher<'a, M, T, CAP, SUBS, PUBS>
{
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
/// Future for the publisher wait action
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct PublisherWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
/// The message we need to publish
message: Option<T>,
publisher: &'s Pub<'a, PSB, T>,
}
impl<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Future for PublisherWaitFuture<'s, 'a, PSB, T> {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let message = self.message.take().unwrap();
match self.publisher.channel.publish_with_context(message, Some(cx)) {
Ok(()) => Poll::Ready(()),
Err(message) => {
self.message = Some(message);
Poll::Pending
}
}
}
}
impl<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Unpin for PublisherWaitFuture<'s, 'a, PSB, T> {}
+158
View File
@@ -0,0 +1,158 @@
//! Implementation of anything directly subscriber related
use core::future::Future;
use core::marker::PhantomData;
use core::ops::{Deref, DerefMut};
use core::pin::Pin;
use core::task::{Context, Poll};
use super::{PubSubBehavior, PubSubChannel, WaitResult};
use crate::blocking_mutex::raw::RawMutex;
/// A subscriber to a channel
pub struct Sub<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
/// The message id of the next message we are yet to receive
next_message_id: u64,
/// The channel we are a subscriber to
channel: &'a PSB,
_phantom: PhantomData<T>,
}
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Sub<'a, PSB, T> {
pub(super) fn new(next_message_id: u64, channel: &'a PSB) -> Self {
Self {
next_message_id,
channel,
_phantom: Default::default(),
}
}
/// Wait for a published message
pub fn next_message<'s>(&'s mut self) -> SubscriberWaitFuture<'s, 'a, PSB, T> {
SubscriberWaitFuture { subscriber: self }
}
/// Wait for a published message (ignoring lag results)
pub async fn next_message_pure(&mut self) -> T {
loop {
match self.next_message().await {
WaitResult::Lagged(_) => continue,
WaitResult::Message(message) => break message,
}
}
}
/// Try to see if there's a published message we haven't received yet.
///
/// This function does not peek. The message is received if there is one.
pub fn try_next_message(&mut self) -> Option<WaitResult<T>> {
match self.channel.get_message_with_context(&mut self.next_message_id, None) {
Poll::Ready(result) => Some(result),
Poll::Pending => None,
}
}
/// Try to see if there's a published message we haven't received yet (ignoring lag results).
///
/// This function does not peek. The message is received if there is one.
pub fn try_next_message_pure(&mut self) -> Option<T> {
loop {
match self.try_next_message() {
Some(WaitResult::Lagged(_)) => continue,
Some(WaitResult::Message(message)) => break Some(message),
None => break None,
}
}
}
/// The amount of messages this subscriber hasn't received yet
pub fn available(&self) -> u64 {
self.channel.available(self.next_message_id)
}
}
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Drop for Sub<'a, PSB, T> {
fn drop(&mut self) {
self.channel.unregister_subscriber(self.next_message_id)
}
}
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Unpin for Sub<'a, PSB, T> {}
/// Warning: The stream implementation ignores lag results and returns all messages.
/// This might miss some messages without you knowing it.
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> futures_util::Stream for Sub<'a, PSB, T> {
type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
match self
.channel
.get_message_with_context(&mut self.next_message_id, Some(cx))
{
Poll::Ready(WaitResult::Message(message)) => Poll::Ready(Some(message)),
Poll::Ready(WaitResult::Lagged(_)) => {
cx.waker().wake_by_ref();
Poll::Pending
}
Poll::Pending => Poll::Pending,
}
}
}
/// A subscriber that holds a dynamic reference to the channel
pub struct DynSubscriber<'a, T: Clone>(pub(super) Sub<'a, dyn PubSubBehavior<T> + 'a, T>);
impl<'a, T: Clone> Deref for DynSubscriber<'a, T> {
type Target = Sub<'a, dyn PubSubBehavior<T> + 'a, T>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<'a, T: Clone> DerefMut for DynSubscriber<'a, T> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
/// A subscriber that holds a generic reference to the channel
pub struct Subscriber<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize>(
pub(super) Sub<'a, PubSubChannel<M, T, CAP, SUBS, PUBS>, T>,
);
impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> Deref
for Subscriber<'a, M, T, CAP, SUBS, PUBS>
{
type Target = Sub<'a, PubSubChannel<M, T, CAP, SUBS, PUBS>, T>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> DerefMut
for Subscriber<'a, M, T, CAP, SUBS, PUBS>
{
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
/// Future for the subscriber wait action
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct SubscriberWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
subscriber: &'s mut Sub<'a, PSB, T>,
}
impl<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Future for SubscriberWaitFuture<'s, 'a, PSB, T> {
type Output = WaitResult<T>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.subscriber
.channel
.get_message_with_context(&mut self.subscriber.next_message_id, Some(cx))
}
}
impl<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Unpin for SubscriberWaitFuture<'s, 'a, PSB, T> {}
+138
View File
@@ -0,0 +1,138 @@
use core::ops::Range;
pub struct RingBuffer<const N: usize> {
start: usize,
end: usize,
empty: bool,
}
impl<const N: usize> RingBuffer<N> {
pub const fn new() -> Self {
Self {
start: 0,
end: 0,
empty: true,
}
}
pub fn push_buf(&mut self) -> Range<usize> {
if self.start == self.end && !self.empty {
trace!(" ringbuf: push_buf empty");
return 0..0;
}
let n = if self.start <= self.end {
N - self.end
} else {
self.start - self.end
};
trace!(" ringbuf: push_buf {:?}..{:?}", self.end, self.end + n);
self.end..self.end + n
}
pub fn push(&mut self, n: usize) {
trace!(" ringbuf: push {:?}", n);
if n == 0 {
return;
}
self.end = self.wrap(self.end + n);
self.empty = false;
}
pub fn pop_buf(&mut self) -> Range<usize> {
if self.empty {
trace!(" ringbuf: pop_buf empty");
return 0..0;
}
let n = if self.end <= self.start {
N - self.start
} else {
self.end - self.start
};
trace!(" ringbuf: pop_buf {:?}..{:?}", self.start, self.start + n);
self.start..self.start + n
}
pub fn pop(&mut self, n: usize) {
trace!(" ringbuf: pop {:?}", n);
if n == 0 {
return;
}
self.start = self.wrap(self.start + n);
self.empty = self.start == self.end;
}
pub fn is_full(&self) -> bool {
self.start == self.end && !self.empty
}
pub fn is_empty(&self) -> bool {
self.empty
}
#[allow(unused)]
pub fn len(&self) -> usize {
if self.empty {
0
} else if self.start < self.end {
self.end - self.start
} else {
N + self.end - self.start
}
}
pub fn clear(&mut self) {
self.start = 0;
self.end = 0;
self.empty = true;
}
fn wrap(&self, n: usize) -> usize {
assert!(n <= N);
if n == N {
0
} else {
n
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn push_pop() {
let mut rb: RingBuffer<4> = RingBuffer::new();
let buf = rb.push_buf();
assert_eq!(0..4, buf);
rb.push(4);
let buf = rb.pop_buf();
assert_eq!(0..4, buf);
rb.pop(1);
let buf = rb.pop_buf();
assert_eq!(1..4, buf);
rb.pop(1);
let buf = rb.pop_buf();
assert_eq!(2..4, buf);
rb.pop(1);
let buf = rb.pop_buf();
assert_eq!(3..4, buf);
rb.pop(1);
let buf = rb.pop_buf();
assert_eq!(0..0, buf);
let buf = rb.push_buf();
assert_eq!(0..4, buf);
}
}
+126
View File
@@ -0,0 +1,126 @@
//! A synchronization primitive for passing the latest value to a task.
use core::cell::Cell;
use core::future::{poll_fn, Future};
use core::task::{Context, Poll, Waker};
use crate::blocking_mutex::raw::RawMutex;
use crate::blocking_mutex::Mutex;
/// Single-slot signaling primitive.
///
/// This is similar to a [`Channel`](crate::channel::Channel) with a buffer size of 1, except
/// "sending" to it (calling [`Signal::signal`]) when full will overwrite the previous value instead
/// of waiting for the receiver to pop the previous value.
///
/// It is useful for sending data between tasks when the receiver only cares about
/// the latest data, and therefore it's fine to "lose" messages. This is often the case for "state"
/// updates.
///
/// For more advanced use cases, you might want to use [`Channel`](crate::channel::Channel) instead.
///
/// Signals are generally declared as `static`s and then borrowed as required.
///
/// ```
/// use embassy_sync::signal::Signal;
/// use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
///
/// enum SomeCommand {
/// On,
/// Off,
/// }
///
/// static SOME_SIGNAL: Signal<CriticalSectionRawMutex, SomeCommand> = Signal::new();
/// ```
pub struct Signal<M, T>
where
M: RawMutex,
{
state: Mutex<M, Cell<State<T>>>,
}
enum State<T> {
None,
Waiting(Waker),
Signaled(T),
}
impl<M, T> Signal<M, T>
where
M: RawMutex,
{
/// Create a new `Signal`.
pub const fn new() -> Self {
Self {
state: Mutex::new(Cell::new(State::None)),
}
}
}
impl<M, T> Default for Signal<M, T>
where
M: RawMutex,
{
fn default() -> Self {
Self::new()
}
}
impl<M, T: Send> Signal<M, T>
where
M: RawMutex,
{
/// Mark this Signal as signaled.
pub fn signal(&self, val: T) {
self.state.lock(|cell| {
let state = cell.replace(State::Signaled(val));
if let State::Waiting(waker) = state {
waker.wake();
}
})
}
/// Remove the queued value in this `Signal`, if any.
pub fn reset(&self) {
self.state.lock(|cell| cell.set(State::None));
}
fn poll_wait(&self, cx: &mut Context<'_>) -> Poll<T> {
self.state.lock(|cell| {
let state = cell.replace(State::None);
match state {
State::None => {
cell.set(State::Waiting(cx.waker().clone()));
Poll::Pending
}
State::Waiting(w) if w.will_wake(cx.waker()) => {
cell.set(State::Waiting(w));
Poll::Pending
}
State::Waiting(w) => {
cell.set(State::Waiting(cx.waker().clone()));
w.wake();
Poll::Pending
}
State::Signaled(res) => Poll::Ready(res),
}
})
}
/// Future that completes when this Signal has been signaled.
pub fn wait(&self) -> impl Future<Output = T> + '_ {
poll_fn(move |cx| self.poll_wait(cx))
}
/// non-blocking method to check whether this signal has been signaled.
pub fn signaled(&self) -> bool {
self.state.lock(|cell| {
let state = cell.replace(State::None);
let res = matches!(state, State::Signaled(_));
cell.set(state);
res
})
}
}
+41
View File
@@ -0,0 +1,41 @@
use core::cell::Cell;
use core::task::Waker;
use crate::blocking_mutex::raw::CriticalSectionRawMutex;
use crate::blocking_mutex::Mutex;
/// Utility struct to register and wake a waker.
pub struct AtomicWaker {
waker: Mutex<CriticalSectionRawMutex, Cell<Option<Waker>>>,
}
impl AtomicWaker {
/// Create a new `AtomicWaker`.
pub const fn new() -> Self {
Self {
waker: Mutex::const_new(CriticalSectionRawMutex::new(), Cell::new(None)),
}
}
/// Register a waker. Overwrites the previous waker, if any.
pub fn register(&self, w: &Waker) {
critical_section::with(|cs| {
let cell = self.waker.borrow(cs);
cell.set(match cell.replace(None) {
Some(w2) if (w2.will_wake(w)) => Some(w2),
_ => Some(w.clone()),
})
})
}
/// Wake the registered waker, if any.
pub fn wake(&self) {
critical_section::with(|cs| {
let cell = self.waker.borrow(cs);
if let Some(w) = cell.replace(None) {
w.wake_by_ref();
cell.set(Some(w));
}
})
}
}
@@ -0,0 +1,30 @@
use core::ptr;
use core::ptr::NonNull;
use core::sync::atomic::{AtomicPtr, Ordering};
use core::task::Waker;
/// Utility struct to register and wake a waker.
pub struct AtomicWaker {
waker: AtomicPtr<()>,
}
impl AtomicWaker {
/// Create a new `AtomicWaker`.
pub const fn new() -> Self {
Self {
waker: AtomicPtr::new(ptr::null_mut()),
}
}
/// Register a waker. Overwrites the previous waker, if any.
pub fn register(&self, w: &Waker) {
self.waker.store(w.as_turbo_ptr().as_ptr() as _, Ordering::Release);
}
/// Wake the registered waker, if any.
pub fn wake(&self) {
if let Some(ptr) = NonNull::new(self.waker.load(Ordering::Acquire)) {
unsafe { Waker::from_turbo_ptr(ptr) }.wake();
}
}
}
+11
View File
@@ -0,0 +1,11 @@
//! Async low-level wait queues
#[cfg_attr(feature = "turbowakers", path = "atomic_waker_turbo.rs")]
mod atomic_waker;
pub use atomic_waker::*;
mod waker_registration;
pub use waker_registration::*;
mod multi_waker;
pub use multi_waker::*;
+58
View File
@@ -0,0 +1,58 @@
use core::task::Waker;
use heapless::Vec;
/// Utility struct to register and wake multiple wakers.
pub struct MultiWakerRegistration<const N: usize> {
wakers: Vec<Waker, N>,
}
impl<const N: usize> MultiWakerRegistration<N> {
/// Create a new empty instance
pub const fn new() -> Self {
Self { wakers: Vec::new() }
}
/// Register a waker. If the buffer is full the function returns it in the error
pub fn register<'a>(&mut self, w: &'a Waker) {
// If we already have some waker that wakes the same task as `w`, do nothing.
// This avoids cloning wakers, and avoids unnecessary mass-wakes.
for w2 in &self.wakers {
if w.will_wake(w2) {
return;
}
}
if self.wakers.is_full() {
// All waker slots were full. It's a bit inefficient, but we can wake everything.
// Any future that is still active will simply reregister.
// This won't happen a lot, so it's ok.
self.wake();
}
if self.wakers.push(w.clone()).is_err() {
// This can't happen unless N=0
// (Either `wakers` wasn't full, or it was in which case `wake()` empied it)
panic!("tried to push a waker to a zero-length MultiWakerRegistration")
}
}
/// Wake all registered wakers. This clears the buffer
pub fn wake(&mut self) {
// heapless::Vec has no `drain()`, do it unsafely ourselves...
// First set length to 0, without dropping the contents.
// This is necessary for soundness: if wake() panics and we're using panic=unwind.
// Setting len=0 upfront ensures other code can't observe the vec in an inconsistent state.
// (it'll leak wakers, but that's not UB)
let len = self.wakers.len();
unsafe { self.wakers.set_len(0) }
for i in 0..len {
// Move a waker out of the vec.
let waker = unsafe { self.wakers.as_mut_ptr().add(i).read() };
// Wake it by value, which consumes (drops) it.
waker.wake();
}
}
}
@@ -0,0 +1,52 @@
use core::mem;
use core::task::Waker;
/// Utility struct to register and wake a waker.
#[derive(Debug, Default)]
pub struct WakerRegistration {
waker: Option<Waker>,
}
impl WakerRegistration {
/// Create a new `WakerRegistration`.
pub const fn new() -> Self {
Self { waker: None }
}
/// Register a waker. Overwrites the previous waker, if any.
pub fn register(&mut self, w: &Waker) {
match self.waker {
// Optimization: If both the old and new Wakers wake the same task, we can simply
// keep the old waker, skipping the clone. (In most executor implementations,
// cloning a waker is somewhat expensive, comparable to cloning an Arc).
Some(ref w2) if (w2.will_wake(w)) => {}
_ => {
// clone the new waker and store it
if let Some(old_waker) = mem::replace(&mut self.waker, Some(w.clone())) {
// We had a waker registered for another task. Wake it, so the other task can
// reregister itself if it's still interested.
//
// If two tasks are waiting on the same thing concurrently, this will cause them
// to wake each other in a loop fighting over this WakerRegistration. This wastes
// CPU but things will still work.
//
// If the user wants to have two tasks waiting on the same thing they should use
// a more appropriate primitive that can store multiple wakers.
old_waker.wake()
}
}
}
}
/// Wake the registered waker, if any.
pub fn wake(&mut self) {
if let Some(w) = self.waker.take() {
w.wake()
}
}
/// Returns true if a waker is currently registered
pub fn occupied(&self) -> bool {
self.waker.is_some()
}
}
+260
View File
@@ -0,0 +1,260 @@
//! A zero-copy queue for sending values between asynchronous tasks.
//!
//! It can be used concurrently by multiple producers (senders) and multiple
//! consumers (receivers), i.e. it is an "MPMC channel".
//!
//! Receivers are competing for messages. So a message that is received by
//! one receiver is not received by any other.
//!
//! This queue takes a Mutex type so that various
//! targets can be attained. For example, a ThreadModeMutex can be used
//! for single-core Cortex-M targets where messages are only passed
//! between tasks running in thread mode. Similarly, a CriticalSectionMutex
//! can also be used for single-core targets where messages are to be
//! passed from exception mode e.g. out of an interrupt handler.
//!
//! This module provides a bounded channel that has a limit on the number of
//! messages that it can store, and if this limit is reached, trying to send
//! another message will result in an error being returned.
use core::cell::RefCell;
use core::future::poll_fn;
use core::marker::PhantomData;
use core::task::{Context, Poll};
use crate::blocking_mutex::raw::RawMutex;
use crate::blocking_mutex::Mutex;
use crate::waitqueue::WakerRegistration;
/// A bounded zero-copy channel for communicating between asynchronous tasks
/// with backpressure.
///
/// The channel will buffer up to the provided number of messages. Once the
/// buffer is full, attempts to `send` new messages will wait until a message is
/// received from the channel.
///
/// All data sent will become available in the same order as it was sent.
///
/// The channel requires a buffer of recyclable elements. Writing to the channel is done through
/// an `&mut T`.
pub struct Channel<'a, M: RawMutex, T> {
buf: *mut T,
phantom: PhantomData<&'a mut T>,
state: Mutex<M, RefCell<State>>,
}
impl<'a, M: RawMutex, T> Channel<'a, M, T> {
/// Initialize a new [`Channel`].
///
/// The provided buffer will be used and reused by the channel's logic, and thus dictates the
/// channel's capacity.
pub fn new(buf: &'a mut [T]) -> Self {
let len = buf.len();
assert!(len != 0);
Self {
buf: buf.as_mut_ptr(),
phantom: PhantomData,
state: Mutex::new(RefCell::new(State {
len,
front: 0,
back: 0,
full: false,
send_waker: WakerRegistration::new(),
receive_waker: WakerRegistration::new(),
})),
}
}
/// Creates a [`Sender`] and [`Receiver`] from an existing channel.
///
/// Further Senders and Receivers can be created through [`Sender::borrow`] and
/// [`Receiver::borrow`] respectively.
pub fn split(&mut self) -> (Sender<'_, M, T>, Receiver<'_, M, T>) {
(Sender { channel: self }, Receiver { channel: self })
}
}
/// Send-only access to a [`Channel`].
pub struct Sender<'a, M: RawMutex, T> {
channel: &'a Channel<'a, M, T>,
}
impl<'a, M: RawMutex, T> Sender<'a, M, T> {
/// Creates one further [`Sender`] over the same channel.
pub fn borrow(&mut self) -> Sender<'_, M, T> {
Sender { channel: self.channel }
}
/// Attempts to send a value over the channel.
pub fn try_send(&mut self) -> Option<&mut T> {
self.channel.state.lock(|s| {
let s = &mut *s.borrow_mut();
match s.push_index() {
Some(i) => Some(unsafe { &mut *self.channel.buf.add(i) }),
None => None,
}
})
}
/// Attempts to send a value over the channel.
pub fn poll_send(&mut self, cx: &mut Context) -> Poll<&mut T> {
self.channel.state.lock(|s| {
let s = &mut *s.borrow_mut();
match s.push_index() {
Some(i) => Poll::Ready(unsafe { &mut *self.channel.buf.add(i) }),
None => {
s.receive_waker.register(cx.waker());
Poll::Pending
}
}
})
}
/// Asynchronously send a value over the channel.
pub async fn send(&mut self) -> &mut T {
let i = poll_fn(|cx| {
self.channel.state.lock(|s| {
let s = &mut *s.borrow_mut();
match s.push_index() {
Some(i) => Poll::Ready(i),
None => {
s.receive_waker.register(cx.waker());
Poll::Pending
}
}
})
})
.await;
unsafe { &mut *self.channel.buf.add(i) }
}
/// Notify the channel that the sending of the value has been finalized.
pub fn send_done(&mut self) {
self.channel.state.lock(|s| s.borrow_mut().push_done())
}
}
/// Receive-only access to a [`Channel`].
pub struct Receiver<'a, M: RawMutex, T> {
channel: &'a Channel<'a, M, T>,
}
impl<'a, M: RawMutex, T> Receiver<'a, M, T> {
/// Creates one further [`Sender`] over the same channel.
pub fn borrow(&mut self) -> Receiver<'_, M, T> {
Receiver { channel: self.channel }
}
/// Attempts to receive a value over the channel.
pub fn try_receive(&mut self) -> Option<&mut T> {
self.channel.state.lock(|s| {
let s = &mut *s.borrow_mut();
match s.pop_index() {
Some(i) => Some(unsafe { &mut *self.channel.buf.add(i) }),
None => None,
}
})
}
/// Attempts to asynchronously receive a value over the channel.
pub fn poll_receive(&mut self, cx: &mut Context) -> Poll<&mut T> {
self.channel.state.lock(|s| {
let s = &mut *s.borrow_mut();
match s.pop_index() {
Some(i) => Poll::Ready(unsafe { &mut *self.channel.buf.add(i) }),
None => {
s.send_waker.register(cx.waker());
Poll::Pending
}
}
})
}
/// Asynchronously receive a value over the channel.
pub async fn receive(&mut self) -> &mut T {
let i = poll_fn(|cx| {
self.channel.state.lock(|s| {
let s = &mut *s.borrow_mut();
match s.pop_index() {
Some(i) => Poll::Ready(i),
None => {
s.send_waker.register(cx.waker());
Poll::Pending
}
}
})
})
.await;
unsafe { &mut *self.channel.buf.add(i) }
}
/// Notify the channel that the receiving of the value has been finalized.
pub fn receive_done(&mut self) {
self.channel.state.lock(|s| s.borrow_mut().pop_done())
}
}
struct State {
len: usize,
/// Front index. Always 0..=(N-1)
front: usize,
/// Back index. Always 0..=(N-1).
back: usize,
/// Used to distinguish "empty" and "full" cases when `front == back`.
/// May only be `true` if `front == back`, always `false` otherwise.
full: bool,
send_waker: WakerRegistration,
receive_waker: WakerRegistration,
}
impl State {
fn increment(&self, i: usize) -> usize {
if i + 1 == self.len {
0
} else {
i + 1
}
}
fn is_full(&self) -> bool {
self.full
}
fn is_empty(&self) -> bool {
self.front == self.back && !self.full
}
fn push_index(&mut self) -> Option<usize> {
match self.is_full() {
true => None,
false => Some(self.back),
}
}
fn push_done(&mut self) {
assert!(!self.is_full());
self.back = self.increment(self.back);
if self.back == self.front {
self.full = true;
}
self.send_waker.wake();
}
fn pop_index(&mut self) -> Option<usize> {
match self.is_empty() {
true => None,
false => Some(self.front),
}
}
fn pop_done(&mut self) {
assert!(!self.is_empty());
self.front = self.increment(self.front);
self.full = false;
self.receive_waker.wake();
}
}