bare_sync/

watch.rs

//! A synchronization primitive for passing the latest value to **multiple** receivers.

use core::cell::{Cell, UnsafeCell};
use core::marker::PhantomData;
use core::mem::MaybeUninit;
use core::ops::{Deref, DerefMut};

use embassy_sync::blocking_mutex::raw::RawMutex;
use embassy_sync::blocking_mutex::Mutex;

/// The `Watch` is a single-slot signaling primitive that allows _multiple_ (`N`) receivers to get
/// changes to the value. Unlike a [`Signal`](crate::signal::Signal), `Watch` supports multiple receivers,
/// and unlike a [`PubSubChannel`](embassy_sync::pubsub::PubSubChannel), `Watch` immediately overwrites the previous
/// value when a new one is sent, without waiting for all receivers to read the previous value.
///
/// This makes `Watch` particularly useful when a single task updates a value or "state", and multiple other tasks
/// need to be notified about changes to this value asynchronously. Receivers may "lose" stale values, as they are
/// always provided with the latest value.
/// ```
///
/// use embedded_sync::watch::Watch;
/// use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
///
/// static WATCH: Watch<CriticalSectionRawMutex, u8> = Watch::new();
///
/// // Obtain receivers and sender
/// let mut rcv0 = WATCH.receiver();
/// let mut rcv1 = WATCH.receiver();
/// let mut snd = WATCH.sender();
///
/// snd.send(10);
///
/// // Receive the new value (async or try)
/// assert_eq!(rcv0.try_changed(), Some(10));
/// assert_eq!(rcv1.try_changed(), Some(10));
///
/// // No update
/// assert_eq!(rcv0.try_changed(), None);
/// assert_eq!(rcv1.try_changed(), None);
///
/// snd.send(20);
///
/// // Using `get` marks the value as seen
/// assert_eq!(rcv1.try_get(), Some(20));
/// assert_eq!(rcv1.try_changed(), None);
///
/// snd.send(20);
///
/// assert_eq!(rcv1.try_get(), Some(20));
/// assert_eq!(rcv1.try_get(), Some(20));
///
/// ```
#[derive(Debug)]
pub struct Watch<M: RawMutex, T: Clone> {
    mutex: Mutex<M, WatchState<T>>,
}

#[derive(Debug)]
struct WatchState<T: Clone> {
    data: UnsafeCell<MaybeUninit<T>>,
    current_id: Cell<u8>,
}

trait SealedWatchBehavior<T> {
    /// Tries to retrieve the value of the `Watch` if it has changed, marking it as seen.
    fn try_changed(&self, id: &mut u8) -> Option<T>;

    /// Clears the value of the `Watch`.
    fn clear(&self);

    /// Sends a new value to the `Watch`.
    fn send(&self, val: T);
}

/// A trait representing the 'inner' behavior of the `Watch`.
#[allow(private_bounds)]
pub trait WatchBehavior<T: Clone>: SealedWatchBehavior<T> {
    /// Tries to get the value of the `Watch`, marking it as seen, if an id is given.
    fn try_get(&self, id: Option<&mut u8>) -> Option<T>;

    /// Checks if the `Watch` is been initialized with a value.
    fn contains_value(&self) -> bool;
}

impl<M: RawMutex, T: Clone> SealedWatchBehavior<T> for Watch<M, T> {
    fn try_changed(&self, id: &mut u8) -> Option<T> {
        self.mutex.lock(|state| {
            let current_id = state.current_id.get();
            if current_id != *id {
                *id = current_id;
                let data = unsafe { state.data.get().read().assume_init() };
                Some(data)
            } else {
                None
            }
        })
    }

    fn clear(&self) {
        self.mutex.lock(|state| {
            state.current_id.set(0);
        })
    }

    fn send(&self, val: T) {
        self.mutex.lock(|state| {
            unsafe { state.data.get().write(MaybeUninit::new(val)) };
            let mut new_id = state.current_id.get().wrapping_add(1);
            if new_id == 0 {
                new_id = 1;
            }
            state.current_id.set(new_id);
        })
    }
}

impl<M: RawMutex, T: Clone> WatchBehavior<T> for Watch<M, T> {
    fn try_get(&self, id: Option<&mut u8>) -> Option<T> {
        self.mutex.lock(|state| {
            let current_id = state.current_id.get();
            if let Some(id) = id {
                *id = current_id;
            }
            if current_id == 0 {
                None
            } else {
                let data = unsafe { state.data.get().read().assume_init() };
                Some(data)
            }
        })
    }

    fn contains_value(&self) -> bool {
        self.mutex.lock(|state| state.current_id.get() != 0)
    }
}

impl<M: RawMutex, T: Clone> Watch<M, T> {
    /// Create a new `Watch` channel for `N` receivers.
    pub const fn new() -> Self {
        Self {
            mutex: Mutex::new(WatchState {
                data: UnsafeCell::new(MaybeUninit::zeroed()),
                current_id: Cell::new(0),
            }),
        }
    }

    /// Create a new `Watch` channel with default data.
    pub const fn new_with(data: T) -> Self {
        Self {
            mutex: Mutex::new(WatchState {
                data: UnsafeCell::new(MaybeUninit::new(data)),
                current_id: Cell::new(0),
            }),
        }
    }

    /// Create a new [`Sender`] for the `Watch`.
    pub fn sender(&self) -> Sender<'_, M, T> {
        Sender(Snd::new(self))
    }

    /// Try to create a new [`Receiver`] for the `Watch`. If the
    /// maximum number of receivers has been reached, `None` is returned.
    pub fn receiver(&self) -> Receiver<'_, M, T> {
        Receiver(Rcv::new(self))
    }

    /// Returns the message ID of the latest message sent to the `Watch`.
    ///
    /// This counter is monotonic, and is incremented every time a new message is sent.
    pub fn get_msg_id(&self) -> u8 {
        self.mutex.lock(|state| state.current_id.get())
    }

    /// Tries to get the value of the `Watch`.
    pub fn try_get(&self) -> Option<T> {
        WatchBehavior::try_get(self, None)
    }
}

/// A receiver can `.await` a change in the `Watch` value.
#[derive(Debug)]
pub struct Snd<'a, T: Clone, W: WatchBehavior<T> + ?Sized> {
    watch: &'a W,
    _phantom: PhantomData<T>,
}

impl<'a, T: Clone, W: WatchBehavior<T> + ?Sized> Clone for Snd<'a, T, W> {
    fn clone(&self) -> Self {
        Self {
            watch: self.watch,
            _phantom: PhantomData,
        }
    }
}

impl<'a, T: Clone, W: WatchBehavior<T> + ?Sized> Snd<'a, T, W> {
    /// Creates a new `Receiver` with a reference to the `Watch`.
    fn new(watch: &'a W) -> Self {
        Self {
            watch,
            _phantom: PhantomData,
        }
    }

    /// Sends a new value to the `Watch`.
    pub fn send(&self, val: T) {
        self.watch.send(val)
    }

    /// Clears the value of the `Watch`.
    /// This will cause calls to [`Rcv::get`] to be pending.
    pub fn clear(&self) {
        self.watch.clear()
    }

    /// Tries to retrieve the value of the `Watch`.
    pub fn try_get(&self) -> Option<T> {
        self.watch.try_get(None)
    }

    /// Returns true if the `Watch` contains a value.
    pub fn contains_value(&self) -> bool {
        self.watch.contains_value()
    }
}

/// A sender of a `Watch` channel.
///
/// For a simpler type definition, consider [`DynSender`] at the expense of
/// some runtime performance due to dynamic dispatch.
#[derive(Debug)]
pub struct Sender<'a, M: RawMutex, T: Clone>(Snd<'a, T, Watch<M, T>>);

impl<'a, M: RawMutex, T: Clone> Clone for Sender<'a, M, T> {
    fn clone(&self) -> Self {
        Self(self.0.clone())
    }
}

impl<'a, M: RawMutex, T: Clone> Deref for Sender<'a, M, T> {
    type Target = Snd<'a, T, Watch<M, T>>;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<'a, M: RawMutex, T: Clone> DerefMut for Sender<'a, M, T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}

/// A receiver can get a change in the `Watch` value.
pub struct Rcv<'a, T: Clone, W: WatchBehavior<T> + ?Sized> {
    watch: &'a W,
    at_id: u8,
    _phantom: PhantomData<T>,
}

impl<'a, T: Clone, W: WatchBehavior<T> + ?Sized> Rcv<'a, T, W> {
    /// Creates a new `Receiver` with a reference to the `Watch`.
    fn new(watch: &'a W) -> Self {
        Self {
            watch,
            at_id: 0,
            _phantom: PhantomData,
        }
    }

    /// Tries to get the current value of the `Watch` without waiting, marking it as seen.
    pub fn try_get(&mut self) -> Option<T> {
        self.watch.try_get(Some(&mut self.at_id))
    }

    /// Tries to get the new value of the watch without waiting, marking it as seen.
    pub fn try_changed(&mut self) -> Option<T> {
        self.watch.try_changed(&mut self.at_id)
    }

    /// Checks if the `Watch` contains a value. If this returns true,
    /// then awaiting [`Rcv::get`] will return immediately.
    pub fn contains_value(&self) -> bool {
        self.watch.contains_value()
    }
}

/// A receiver of a `Watch` channel.
pub struct Receiver<'a, M: RawMutex, T: Clone>(Rcv<'a, T, Watch<M, T>>);

impl<'a, M: RawMutex, T: Clone> Deref for Receiver<'a, M, T> {
    type Target = Rcv<'a, T, Watch<M, T>>;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<'a, M: RawMutex, T: Clone> DerefMut for Receiver<'a, M, T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}

#[cfg(test)]
mod tests {
    use super::Watch;
    use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;

    #[test]
    fn multiple_sends() {
        static WATCH: Watch<CriticalSectionRawMutex, u8> = Watch::new();

        // Obtain receiver and sender
        let mut rcv = WATCH.receiver();
        let snd = WATCH.sender();

        // Not initialized
        assert_eq!(rcv.try_changed(), None);

        // Receive another value
        snd.send(20);
        assert_eq!(rcv.try_changed(), Some(20));

        // No update
        assert_eq!(rcv.try_changed(), None);
    }

    #[test]
    fn all_try_get() {
        static WATCH: Watch<CriticalSectionRawMutex, u8> = Watch::new();

        // Obtain receiver and sender
        let mut rcv = WATCH.receiver();
        let snd = WATCH.sender();

        // Not initialized
        assert_eq!(WATCH.try_get(), None);
        assert_eq!(rcv.try_get(), None);
        assert_eq!(snd.try_get(), None);

        // Receive the new value
        snd.send(10);
        assert_eq!(WATCH.try_get(), Some(10));
        assert_eq!(rcv.try_get(), Some(10));
        assert_eq!(snd.try_get(), Some(10));
    }

    #[test]
    fn once_lock_like() {
        static CONFIG0: u8 = 10;
        static CONFIG1: u8 = 20;

        static WATCH: Watch<CriticalSectionRawMutex, &'static u8> = Watch::new();

        // Obtain receiver and sender
        let mut rcv = WATCH.receiver();
        let snd = WATCH.sender();

        // Not initialized
        assert_eq!(rcv.try_changed(), None);

        // Receive the new value
        snd.send(&CONFIG0);
        let rcv0 = rcv.try_changed().unwrap();
        assert_eq!(rcv0, &10);

        // Receive another value
        snd.send(&CONFIG1);
        let rcv1 = rcv.try_changed();
        assert_eq!(rcv1, Some(&20));

        // No update
        assert_eq!(rcv.try_changed(), None);

        // Ensure similarity with original static
        assert_eq!(rcv0, &CONFIG0);
        assert_eq!(rcv1, Some(&CONFIG1));
    }

    #[test]
    fn sender_modify() {
        static WATCH: Watch<CriticalSectionRawMutex, u8> = Watch::new();

        // Obtain receiver and sender
        let mut rcv = WATCH.receiver();
        let snd = WATCH.sender();

        // Receive the new value
        snd.send(10);
        assert_eq!(rcv.try_changed(), Some(10));
    }

    #[test]
    fn receive_after_create() {
        static WATCH: Watch<CriticalSectionRawMutex, u8> = Watch::new();

        // Obtain sender and send value
        let snd = WATCH.sender();
        snd.send(10);

        // Obtain receiver and receive value
        let mut rcv = WATCH.receiver();
        assert_eq!(rcv.try_changed(), Some(10));
    }

    #[test]
    fn multiple_receivers() {
        static WATCH: Watch<CriticalSectionRawMutex, u8> = Watch::new();

        // Obtain receivers and sender
        let mut rcv0 = WATCH.receiver();
        let snd = WATCH.sender();

        // No update for both
        assert_eq!(rcv0.try_changed(), None);

        // Send a new value
        snd.send(0);

        // Both receivers receive the new value
        assert_eq!(rcv0.try_changed(), Some(0));
    }

    #[test]
    fn clone_senders() {
        // Obtain different ways to send
        static WATCH: Watch<CriticalSectionRawMutex, u8> = Watch::new();
        let snd0 = WATCH.sender();
        let snd1 = snd0.clone();

        // Obtain Receiver
        let mut rcv = WATCH.receiver();

        // Send a value from first sender
        snd0.send(10);
        assert_eq!(rcv.try_changed(), Some(10));

        // Send a value from second sender
        snd1.send(20);
        assert_eq!(rcv.try_changed(), Some(20));
    }

    #[test]
    fn contains_value() {
        static WATCH: Watch<CriticalSectionRawMutex, u8> = Watch::new();

        // Obtain receiver and sender
        let rcv = WATCH.receiver();
        let snd = WATCH.sender();

        // check if the watch contains a value
        assert_eq!(rcv.contains_value(), false);
        assert_eq!(snd.contains_value(), false);

        // Send a value
        snd.send(10);

        // check if the watch contains a value
        assert_eq!(rcv.contains_value(), true);
        assert_eq!(snd.contains_value(), true);
    }
}