23 files changed, 2924 insertions, 135 deletions

diff --git a/embassy-sync/src/blocking_mutex/mod.rs b/embassy-sync/src/blocking_mutex/mod.rs
index 8a4a4c642..62bfc26fb 100644
--- a/embassy-sync/src/blocking_mutex/mod.rs
+++ b/embassy-sync/src/blocking_mutex/mod.rs
@@ -22,6 +22,7 @@ use self::raw::RawMutex;
 ///
 /// 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.
+#[derive(Debug)]
 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`.
@@ -50,6 +51,23 @@ impl<R: RawMutex, T> Mutex<R, T> {
             f(inner)
         })
     }
+
+    /// Creates a critical section and grants temporary mutable access to the protected data.
+    ///
+    /// # Safety
+    ///
+    /// This method is marked unsafe because calling this method re-entrantly, i.e. within
+    /// another `lock_mut` or `lock` closure, violates Rust's aliasing rules. Calling this
+    /// method at the same time from different tasks is safe. For a safe alternative with
+    /// mutable access that never causes UB, use a `RefCell` in a `Mutex`.
+    pub unsafe fn lock_mut<U>(&self, f: impl FnOnce(&mut T) -> U) -> U {
+        self.raw.lock(|| {
+            let ptr = self.data.get() as *mut T;
+            // Safety: we have exclusive access to the data, as long as this mutex is not locked re-entrantly
+            let inner = unsafe { &mut *ptr };
+            f(inner)
+        })
+    }
 }
 
 impl<R, T> Mutex<R, T> {
@@ -104,6 +122,7 @@ impl<T> Mutex<raw::CriticalSectionRawMutex, T> {
 
 impl<T> Mutex<raw::NoopRawMutex, T> {
     /// Borrows the data
+    #[allow(clippy::should_implement_trait)]
     pub fn borrow(&self) -> &T {
         let ptr = self.data.get() as *const T;
         unsafe { &*ptr }
@@ -116,9 +135,9 @@ impl<T> Mutex<raw::NoopRawMutex, T> {
 // 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"))]
+#[cfg(any(cortex_m, doc, feature = "std"))]
 pub use thread_mode_mutex::*;
-#[cfg(any(cortex_m, feature = "std"))]
+#[cfg(any(cortex_m, doc, feature = "std"))]
 mod thread_mode_mutex {
     use super::*;
 
diff --git a/embassy-sync/src/blocking_mutex/raw.rs b/embassy-sync/src/blocking_mutex/raw.rs
index a8afcad34..fbb9ece15 100644
--- a/embassy-sync/src/blocking_mutex/raw.rs
+++ b/embassy-sync/src/blocking_mutex/raw.rs
@@ -37,6 +37,7 @@ pub unsafe trait RawMutex {
 /// # Safety
 ///
 /// This mutex is safe to share between different executors and interrupts.
+#[derive(Debug)]
 pub struct CriticalSectionRawMutex {
     _phantom: PhantomData<()>,
 }
@@ -65,6 +66,7 @@ unsafe impl RawMutex for CriticalSectionRawMutex {
 /// # Safety
 ///
 /// **This Mutex is only safe within a single executor.**
+#[derive(Debug)]
 pub struct NoopRawMutex {
     _phantom: PhantomData<*mut ()>,
 }
@@ -87,7 +89,7 @@ unsafe impl RawMutex for NoopRawMutex {
 
 // ================
 
-#[cfg(any(cortex_m, feature = "std"))]
+#[cfg(any(cortex_m, doc, feature = "std"))]
 mod thread_mode {
     use super::*;
 
@@ -145,5 +147,5 @@ mod thread_mode {
         return unsafe { (0xE000ED04 as *const u32).read_volatile() } & 0x1FF == 0;
     }
 }
-#[cfg(any(cortex_m, feature = "std"))]
+#[cfg(any(cortex_m, doc, feature = "std"))]
 pub use thread_mode::*;
diff --git a/embassy-sync/src/channel.rs b/embassy-sync/src/channel.rs
index 55ac5fb66..dbd24a6c7 100644
--- a/embassy-sync/src/channel.rs
+++ b/embassy-sync/src/channel.rs
@@ -17,6 +17,31 @@
 //! messages that it can store, and if this limit is reached, trying to send
 //! another message will result in an error being returned.
 //!
+//! # Example: Message passing between task and interrupt handler
+//!
+//! ```rust
+//! use embassy_sync::channel::Channel;
+//! use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
+//!
+//! static SHARED_CHANNEL: Channel<CriticalSectionRawMutex, u32, 8> = Channel::new();
+//!
+//! fn my_interrupt_handler() {
+//!     // Do some work..
+//!     // ...
+//!     if let Err(e) = SHARED_CHANNEL.sender().try_send(42) {
+//!         // Channel is full..
+//!     }
+//! }
+//!
+//! async fn my_async_task() {
+//!     // ...
+//!     let receiver = SHARED_CHANNEL.receiver();
+//!     loop {
+//!         let data_from_interrupt = receiver.receive().await;
+//!         // Do something with the data.
+//!     }
+//! }
+//! ```
 
 use core::cell::RefCell;
 use core::future::Future;
@@ -25,11 +50,12 @@ use core::task::{Context, Poll};
 
 use heapless::Deque;
 
-use crate::blocking_mutex::raw::RawMutex;
 use crate::blocking_mutex::Mutex;
+use crate::blocking_mutex::raw::RawMutex;
 use crate::waitqueue::WakerRegistration;
 
 /// Send-only access to a [`Channel`].
+#[derive(Debug)]
 pub struct Sender<'ch, M, T, const N: usize>
 where
     M: RawMutex,
@@ -72,6 +98,48 @@ where
     pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
         self.channel.poll_ready_to_send(cx)
     }
+
+    /// Returns the maximum number of elements the channel can hold.
+    ///
+    /// See [`Channel::capacity()`]
+    pub const fn capacity(&self) -> usize {
+        self.channel.capacity()
+    }
+
+    /// Returns the free capacity of the channel.
+    ///
+    /// See [`Channel::free_capacity()`]
+    pub fn free_capacity(&self) -> usize {
+        self.channel.free_capacity()
+    }
+
+    /// Clears all elements in the channel.
+    ///
+    /// See [`Channel::clear()`]
+    pub fn clear(&self) {
+        self.channel.clear();
+    }
+
+    /// Returns the number of elements currently in the channel.
+    ///
+    /// See [`Channel::len()`]
+    pub fn len(&self) -> usize {
+        self.channel.len()
+    }
+
+    /// Returns whether the channel is empty.
+    ///
+    /// See [`Channel::is_empty()`]
+    pub fn is_empty(&self) -> bool {
+        self.channel.is_empty()
+    }
+
+    /// Returns whether the channel is full.
+    ///
+    /// See [`Channel::is_full()`]
+    pub fn is_full(&self) -> bool {
+        self.channel.is_full()
+    }
 }
 
 /// Send-only access to a [`Channel`] without knowing channel size.
@@ -122,7 +190,59 @@ impl<'ch, T> DynamicSender<'ch, T> {
     }
 }
 
+/// Send-only access to a [`Channel`] without knowing channel size.
+/// This version can be sent between threads but can only be created if the underlying mutex is Sync.
+pub struct SendDynamicSender<'ch, T> {
+    pub(crate) channel: &'ch dyn DynamicChannel<T>,
+}
+
+impl<'ch, T> Clone for SendDynamicSender<'ch, T> {
+    fn clone(&self) -> Self {
+        *self
+    }
+}
+
+impl<'ch, T> Copy for SendDynamicSender<'ch, T> {}
+unsafe impl<'ch, T: Send> Send for SendDynamicSender<'ch, T> {}
+unsafe impl<'ch, T: Send> Sync for SendDynamicSender<'ch, T> {}
+
+impl<'ch, M, T, const N: usize> From<Sender<'ch, M, T, N>> for SendDynamicSender<'ch, T>
+where
+    M: RawMutex + Sync + Send,
+{
+    fn from(s: Sender<'ch, M, T, N>) -> Self {
+        Self { channel: s.channel }
+    }
+}
+
+impl<'ch, T> SendDynamicSender<'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`].
+#[derive(Debug)]
 pub struct Receiver<'ch, M, T, const N: usize>
 where
     M: RawMutex,
@@ -166,6 +286,16 @@ where
         self.channel.try_receive()
     }
 
+    /// Peek at the next value without removing it from the queue.
+    ///
+    /// See [`Channel::try_peek()`]
+    pub fn try_peek(&self) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        self.channel.try_peek()
+    }
+
     /// Allows a poll_fn to poll until the channel is ready to receive
     ///
     /// See [`Channel::poll_ready_to_receive()`]
@@ -179,6 +309,48 @@ where
     pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
         self.channel.poll_receive(cx)
     }
+
+    /// Returns the maximum number of elements the channel can hold.
+    ///
+    /// See [`Channel::capacity()`]
+    pub const fn capacity(&self) -> usize {
+        self.channel.capacity()
+    }
+
+    /// Returns the free capacity of the channel.
+    ///
+    /// See [`Channel::free_capacity()`]
+    pub fn free_capacity(&self) -> usize {
+        self.channel.free_capacity()
+    }
+
+    /// Clears all elements in the channel.
+    ///
+    /// See [`Channel::clear()`]
+    pub fn clear(&self) {
+        self.channel.clear();
+    }
+
+    /// Returns the number of elements currently in the channel.
+    ///
+    /// See [`Channel::len()`]
+    pub fn len(&self) -> usize {
+        self.channel.len()
+    }
+
+    /// Returns whether the channel is empty.
+    ///
+    /// See [`Channel::is_empty()`]
+    pub fn is_empty(&self) -> bool {
+        self.channel.is_empty()
+    }
+
+    /// Returns whether the channel is full.
+    ///
+    /// See [`Channel::is_full()`]
+    pub fn is_full(&self) -> bool {
+        self.channel.is_full()
+    }
 }
 
 /// Receive-only access to a [`Channel`] without knowing channel size.
@@ -209,6 +381,16 @@ impl<'ch, T> DynamicReceiver<'ch, T> {
         self.channel.try_receive_with_context(None)
     }
 
+    /// Peek at the next value without removing it from the queue.
+    ///
+    /// See [`Channel::try_peek()`]
+    pub fn try_peek(&self) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        self.channel.try_peek_with_context(None)
+    }
+
     /// Allows a poll_fn to poll until the channel is ready to receive
     ///
     /// See [`Channel::poll_ready_to_receive()`]
@@ -233,8 +415,80 @@ where
     }
 }
 
+/// Receive-only access to a [`Channel`] without knowing channel size.
+/// This version can be sent between threads but can only be created if the underlying mutex is Sync.
+pub struct SendDynamicReceiver<'ch, T> {
+    pub(crate) channel: &'ch dyn DynamicChannel<T>,
+}
+
+/// Receive-only access to a [`Channel`] without knowing channel size.
+/// This version can be sent between threads but can only be created if the underlying mutex is Sync.
+#[deprecated(since = "0.7.1", note = "please use `SendDynamicReceiver` instead")]
+pub type SendableDynamicReceiver<'ch, T> = SendDynamicReceiver<'ch, T>;
+
+impl<'ch, T> Clone for SendDynamicReceiver<'ch, T> {
+    fn clone(&self) -> Self {
+        *self
+    }
+}
+
+impl<'ch, T> Copy for SendDynamicReceiver<'ch, T> {}
+unsafe impl<'ch, T: Send> Send for SendDynamicReceiver<'ch, T> {}
+unsafe impl<'ch, T: Send> Sync for SendDynamicReceiver<'ch, T> {}
+
+impl<'ch, T> SendDynamicReceiver<'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 SendDynamicReceiver<'ch, T>
+where
+    M: RawMutex + Sync + Send,
+{
+    fn from(s: Receiver<'ch, M, T, N>) -> Self {
+        Self { channel: s.channel }
+    }
+}
+
+impl<'ch, M, T, const N: usize> futures_core::Stream for Receiver<'ch, M, T, N>
+where
+    M: RawMutex,
+{
+    type Item = T;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.channel.poll_receive(cx).map(Some)
+    }
+}
+
 /// Future returned by [`Channel::receive`] and  [`Receiver::receive`].
 #[must_use = "futures do nothing unless you `.await` or poll them"]
+#[derive(Debug)]
 pub struct ReceiveFuture<'ch, M, T, const N: usize>
 where
     M: RawMutex,
@@ -255,6 +509,7 @@ where
 
 /// Future returned by [`Channel::ready_to_receive`] and  [`Receiver::ready_to_receive`].
 #[must_use = "futures do nothing unless you `.await` or poll them"]
+#[derive(Debug)]
 pub struct ReceiveReadyFuture<'ch, M, T, const N: usize>
 where
     M: RawMutex,
@@ -298,6 +553,7 @@ impl<'ch, M: RawMutex, T, const N: usize> From<ReceiveFuture<'ch, M, T, N>> for
 
 /// Future returned by [`Channel::send`] and  [`Sender::send`].
 #[must_use = "futures do nothing unless you `.await` or poll them"]
+#[derive(Debug)]
 pub struct SendFuture<'ch, M, T, const N: usize>
 where
     M: RawMutex,
@@ -368,6 +624,10 @@ pub(crate) trait DynamicChannel<T> {
 
     fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>;
 
+    fn try_peek_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>
+    where
+        T: Clone;
+
     fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()>;
     fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()>;
 
@@ -391,6 +651,7 @@ pub enum TrySendError<T> {
     Full(T),
 }
 
+#[derive(Debug)]
 struct ChannelState<T, const N: usize> {
     queue: Deque<T, N>,
     receiver_waker: WakerRegistration,
@@ -410,6 +671,31 @@ impl<T, const N: usize> ChannelState<T, N> {
         self.try_receive_with_context(None)
     }
 
+    fn try_peek(&mut self) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        self.try_peek_with_context(None)
+    }
+
+    fn try_peek_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        if self.queue.is_full() {
+            self.senders_waker.wake();
+        }
+
+        if let Some(message) = self.queue.front() {
+            Ok(message.clone())
+        } else {
+            if let Some(cx) = cx {
+                self.receiver_waker.register(cx.waker());
+            }
+            Err(TryReceiveError::Empty)
+        }
+    }
+
     fn try_receive_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
         if self.queue.is_full() {
             self.senders_waker.wake();
@@ -478,6 +764,9 @@ impl<T, const N: usize> ChannelState<T, N> {
     }
 
     fn clear(&mut self) {
+        if self.queue.is_full() {
+            self.senders_waker.wake();
+        }
         self.queue.clear();
     }
 
@@ -502,6 +791,7 @@ impl<T, const N: usize> ChannelState<T, N> {
 /// received from the channel.
 ///
 /// All data sent will become available in the same order as it was sent.
+#[derive(Debug)]
 pub struct Channel<M, T, const N: usize>
 where
     M: RawMutex,
@@ -536,6 +826,13 @@ where
         self.lock(|c| c.try_receive_with_context(cx))
     }
 
+    fn try_peek_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        self.lock(|c| c.try_peek_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))
@@ -624,6 +921,17 @@ where
         self.lock(|c| c.try_receive())
     }
 
+    /// Peek at the next value without removing it from the queue.
+    ///
+    /// This method will either receive a copy of the message from the channel immediately or return
+    /// an error if the channel is empty.
+    pub fn try_peek(&self) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        self.lock(|c| c.try_peek())
+    }
+
     /// Returns the maximum number of elements the channel can hold.
     pub const fn capacity(&self) -> usize {
         N
@@ -671,6 +979,13 @@ where
         Channel::try_receive_with_context(self, cx)
     }
 
+    fn try_peek_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        Channel::try_peek_with_context(self, cx)
+    }
+
     fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
         Channel::poll_ready_to_send(self, cx)
     }
@@ -684,6 +999,17 @@ where
     }
 }
 
+impl<M, T, const N: usize> futures_core::Stream for Channel<M, T, N>
+where
+    M: RawMutex,
+{
+    type Item = T;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.poll_receive(cx).map(Some)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use core::time::Duration;
@@ -742,6 +1068,8 @@ mod tests {
     fn simple_send_and_receive() {
         let c = Channel::<NoopRawMutex, u32, 3>::new();
         assert!(c.try_send(1).is_ok());
+        assert_eq!(c.try_peek().unwrap(), 1);
+        assert_eq!(c.try_peek().unwrap(), 1);
         assert_eq!(c.try_receive().unwrap(), 1);
     }
 
@@ -772,6 +1100,8 @@ mod tests {
         let r = c.dyn_receiver();
 
         assert!(s.try_send(1).is_ok());
+        assert_eq!(r.try_peek().unwrap(), 1);
+        assert_eq!(r.try_peek().unwrap(), 1);
         assert_eq!(r.try_receive().unwrap(), 1);
     }
 
@@ -782,11 +1112,13 @@ mod tests {
         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!(
+            executor
+                .spawn(async move {
+                    assert!(c2.try_send(1).is_ok());
+                })
+                .is_ok()
+        );
         assert_eq!(c.receive().await, 1);
     }
 
@@ -813,13 +1145,15 @@ mod tests {
         // 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());
+        assert!(
+            executor
+                .spawn(async move {
+                    loop {
+                        c.receive().await;
+                    }
+                })
+                .is_ok()
+        );
         send_task_1.unwrap().await;
         send_task_2.unwrap().await;
     }
diff --git a/embassy-sync/src/fmt.rs b/embassy-sync/src/fmt.rs
index 2ac42c557..8ca61bc39 100644
--- a/embassy-sync/src/fmt.rs
+++ b/embassy-sync/src/fmt.rs
@@ -6,6 +6,7 @@ use core::fmt::{Debug, Display, LowerHex};
 #[cfg(all(feature = "defmt", feature = "log"))]
 compile_error!("You may not enable both `defmt` and `log` features.");
 
+#[collapse_debuginfo(yes)]
 macro_rules! assert {
     ($($x:tt)*) => {
         {
@@ -17,6 +18,7 @@ macro_rules! assert {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! assert_eq {
     ($($x:tt)*) => {
         {
@@ -28,6 +30,7 @@ macro_rules! assert_eq {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! assert_ne {
     ($($x:tt)*) => {
         {
@@ -39,6 +42,7 @@ macro_rules! assert_ne {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! debug_assert {
     ($($x:tt)*) => {
         {
@@ -50,6 +54,7 @@ macro_rules! debug_assert {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! debug_assert_eq {
     ($($x:tt)*) => {
         {
@@ -61,6 +66,7 @@ macro_rules! debug_assert_eq {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! debug_assert_ne {
     ($($x:tt)*) => {
         {
@@ -72,6 +78,7 @@ macro_rules! debug_assert_ne {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! todo {
     ($($x:tt)*) => {
         {
@@ -83,20 +90,19 @@ macro_rules! todo {
     };
 }
 
-#[cfg(not(feature = "defmt"))]
+#[collapse_debuginfo(yes)]
 macro_rules! unreachable {
     ($($x:tt)*) => {
-        ::core::unreachable!($($x)*)
-    };
-}
-
-#[cfg(feature = "defmt")]
-macro_rules! unreachable {
-    ($($x:tt)*) => {
-        ::defmt::unreachable!($($x)*)
+        {
+            #[cfg(not(feature = "defmt"))]
+            ::core::unreachable!($($x)*);
+            #[cfg(feature = "defmt")]
+            ::defmt::unreachable!($($x)*);
+        }
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! panic {
     ($($x:tt)*) => {
         {
@@ -108,6 +114,7 @@ macro_rules! panic {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! trace {
     ($s:literal $(, $x:expr)* $(,)?) => {
         {
@@ -121,6 +128,7 @@ macro_rules! trace {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! debug {
     ($s:literal $(, $x:expr)* $(,)?) => {
         {
@@ -134,6 +142,7 @@ macro_rules! debug {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! info {
     ($s:literal $(, $x:expr)* $(,)?) => {
         {
@@ -147,6 +156,7 @@ macro_rules! info {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! warn {
     ($s:literal $(, $x:expr)* $(,)?) => {
         {
@@ -160,6 +170,7 @@ macro_rules! warn {
     };
 }
 
+#[collapse_debuginfo(yes)]
 macro_rules! error {
     ($s:literal $(, $x:expr)* $(,)?) => {
         {
@@ -174,6 +185,7 @@ macro_rules! error {
 }
 
 #[cfg(feature = "defmt")]
+#[collapse_debuginfo(yes)]
 macro_rules! unwrap {
     ($($x:tt)*) => {
         ::defmt::unwrap!($($x)*)
@@ -181,6 +193,7 @@ macro_rules! unwrap {
 }
 
 #[cfg(not(feature = "defmt"))]
+#[collapse_debuginfo(yes)]
 macro_rules! unwrap {
     ($arg:expr) => {
         match $crate::fmt::Try::into_result($arg) {
diff --git a/embassy-sync/src/lazy_lock.rs b/embassy-sync/src/lazy_lock.rs
new file mode 100644
index 000000000..945560a80
--- /dev/null
+++ b/embassy-sync/src/lazy_lock.rs
@@ -0,0 +1,174 @@
+//! Synchronization primitive for initializing a value once, allowing others to get a reference to the value.
+
+use core::cell::UnsafeCell;
+use core::mem::ManuallyDrop;
+use core::sync::atomic::{AtomicBool, Ordering};
+
+/// The `LazyLock` is a synchronization primitive that allows for
+/// initializing a value once, and allowing others to obtain a
+/// reference to the value. This is useful for lazy initialization of
+/// a static value.
+///
+/// # Example
+/// ```
+/// use futures_executor::block_on;
+/// use embassy_sync::lazy_lock::LazyLock;
+///
+/// // Define a static value that will be lazily initialized
+/// // at runtime at the first access.
+/// static VALUE: LazyLock<u32> = LazyLock::new(|| 20);
+///
+/// let reference = VALUE.get();
+/// assert_eq!(reference, &20);
+/// ```
+#[derive(Debug)]
+pub struct LazyLock<T, F = fn() -> T> {
+    init: AtomicBool,
+    data: UnsafeCell<Data<T, F>>,
+}
+
+union Data<T, F> {
+    value: ManuallyDrop<T>,
+    f: ManuallyDrop<F>,
+}
+
+unsafe impl<T, F> Sync for LazyLock<T, F>
+where
+    T: Sync,
+    F: Sync,
+{
+}
+
+impl<T, F: FnOnce() -> T> LazyLock<T, F> {
+    /// Create a new uninitialized `StaticLock`.
+    pub const fn new(init_fn: F) -> Self {
+        Self {
+            init: AtomicBool::new(false),
+            data: UnsafeCell::new(Data {
+                f: ManuallyDrop::new(init_fn),
+            }),
+        }
+    }
+
+    /// Get a reference to the underlying value, initializing it if it
+    /// has not been done already.
+    #[inline]
+    pub fn get(&self) -> &T {
+        self.ensure_init_fast();
+        unsafe { &(*self.data.get()).value }
+    }
+
+    /// Get a mutable reference to the underlying value, initializing it if it
+    /// has not been done already.
+    #[inline]
+    pub fn get_mut(&mut self) -> &mut T {
+        self.ensure_init_fast();
+        unsafe { &mut (*self.data.get()).value }
+    }
+
+    /// Consume the `LazyLock`, returning the underlying value. The
+    /// initialization function will be called if it has not been
+    /// already.
+    #[inline]
+    pub fn into_inner(self) -> T {
+        self.ensure_init_fast();
+        let this = ManuallyDrop::new(self);
+        let data = unsafe { core::ptr::read(&this.data) }.into_inner();
+
+        ManuallyDrop::into_inner(unsafe { data.value })
+    }
+
+    /// Initialize the `LazyLock` if it has not been initialized yet.
+    /// This function is a fast track to [`Self::ensure_init`]
+    /// which does not require a critical section in most cases when
+    /// the value has been initialized already.
+    /// When this function returns, `self.data` is guaranteed to be
+    /// initialized and visible on the current core.
+    #[inline]
+    fn ensure_init_fast(&self) {
+        if !self.init.load(Ordering::Acquire) {
+            self.ensure_init();
+        }
+    }
+
+    /// Initialize the `LazyLock` if it has not been initialized yet.
+    /// When this function returns, `self.data` is guaranteed to be
+    /// initialized and visible on the current core.
+    fn ensure_init(&self) {
+        critical_section::with(|_| {
+            if !self.init.load(Ordering::Acquire) {
+                let data = unsafe { &mut *self.data.get() };
+                let f = unsafe { ManuallyDrop::take(&mut data.f) };
+                let value = f();
+                data.value = ManuallyDrop::new(value);
+
+                self.init.store(true, Ordering::Release);
+            }
+        });
+    }
+}
+
+impl<T, F> Drop for LazyLock<T, F> {
+    fn drop(&mut self) {
+        if self.init.load(Ordering::Acquire) {
+            unsafe { ManuallyDrop::drop(&mut self.data.get_mut().value) };
+        } else {
+            unsafe { ManuallyDrop::drop(&mut self.data.get_mut().f) };
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use core::sync::atomic::{AtomicU32, Ordering};
+
+    use super::*;
+
+    #[test]
+    fn test_lazy_lock() {
+        static VALUE: LazyLock<u32> = LazyLock::new(|| 20);
+        let reference = VALUE.get();
+        assert_eq!(reference, &20);
+    }
+    #[test]
+    fn test_lazy_lock_mutation() {
+        let mut value: LazyLock<u32> = LazyLock::new(|| 20);
+        *value.get_mut() = 21;
+        let reference = value.get();
+        assert_eq!(reference, &21);
+    }
+    #[test]
+    fn test_lazy_lock_into_inner() {
+        let lazy: LazyLock<u32> = LazyLock::new(|| 20);
+        let value = lazy.into_inner();
+        assert_eq!(value, 20);
+    }
+
+    static DROP_CHECKER: AtomicU32 = AtomicU32::new(0);
+    #[derive(Debug)]
+    struct DropCheck;
+
+    impl Drop for DropCheck {
+        fn drop(&mut self) {
+            DROP_CHECKER.fetch_add(1, Ordering::Acquire);
+        }
+    }
+
+    #[test]
+    fn test_lazy_drop() {
+        let lazy: LazyLock<DropCheck> = LazyLock::new(|| DropCheck);
+        assert_eq!(DROP_CHECKER.load(Ordering::Acquire), 0);
+        lazy.get();
+        drop(lazy);
+        assert_eq!(DROP_CHECKER.load(Ordering::Acquire), 1);
+
+        let dropper = DropCheck;
+        let lazy_fn: LazyLock<u32, _> = LazyLock::new(move || {
+            let _a = dropper;
+            20
+        });
+        assert_eq!(DROP_CHECKER.load(Ordering::Acquire), 1);
+        drop(lazy_fn);
+        assert_eq!(DROP_CHECKER.load(Ordering::Acquire), 2);
+    }
+}
diff --git a/embassy-sync/src/lib.rs b/embassy-sync/src/lib.rs
index a5eee8d02..1cfde8b10 100644
--- a/embassy-sync/src/lib.rs
+++ b/embassy-sync/src/lib.rs
@@ -1,6 +1,7 @@
 #![cfg_attr(not(feature = "std"), no_std)]
 #![allow(async_fn_in_trait)]
 #![allow(clippy::new_without_default)]
+#![allow(unsafe_op_in_unsafe_fn)]
 #![doc = include_str!("../README.md")]
 #![warn(missing_docs)]
 
@@ -12,12 +13,15 @@ mod ring_buffer;
 
 pub mod blocking_mutex;
 pub mod channel;
+pub mod lazy_lock;
 pub mod mutex;
 pub mod once_lock;
 pub mod pipe;
 pub mod priority_channel;
 pub mod pubsub;
+pub mod rwlock;
 pub mod semaphore;
 pub mod signal;
 pub mod waitqueue;
+pub mod watch;
 pub mod zerocopy_channel;
diff --git a/embassy-sync/src/mutex.rs b/embassy-sync/src/mutex.rs
index 8c3a3af9f..96b834f02 100644
--- a/embassy-sync/src/mutex.rs
+++ b/embassy-sync/src/mutex.rs
@@ -2,13 +2,13 @@
 //!
 //! 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::future::{Future, poll_fn};
 use core::ops::{Deref, DerefMut};
 use core::task::Poll;
 use core::{fmt, mem};
 
-use crate::blocking_mutex::raw::RawMutex;
 use crate::blocking_mutex::Mutex as BlockingMutex;
+use crate::blocking_mutex::raw::RawMutex;
 use crate::waitqueue::WakerRegistration;
 
 /// Error returned by [`Mutex::try_lock`]
@@ -16,6 +16,7 @@ use crate::waitqueue::WakerRegistration;
 #[cfg_attr(feature = "defmt", derive(defmt::Format))]
 pub struct TryLockError;
 
+#[derive(Debug)]
 struct State {
     locked: bool,
     waker: WakerRegistration,
@@ -23,7 +24,7 @@ struct State {
 
 /// Async mutex.
 ///
-/// The mutex is generic over a blocking [`RawMutex`](crate::blocking_mutex::raw::RawMutex).
+/// The mutex is generic over a blocking [`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.
@@ -73,7 +74,7 @@ where
     /// 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> {
+    pub fn lock(&self) -> impl Future<Output = MutexGuard<'_, M, T>> {
         poll_fn(|cx| {
             let ready = self.state.lock(|s| {
                 let mut s = s.borrow_mut();
@@ -92,7 +93,6 @@ where
                 Poll::Pending
             }
         })
-        .await
     }
 
     /// Attempt to immediately lock the mutex.
@@ -138,7 +138,7 @@ impl<M: RawMutex, T> From<T> for Mutex<M, T> {
 impl<M, T> Default for Mutex<M, T>
 where
     M: RawMutex,
-    T: ?Sized + Default,
+    T: Default,
 {
     fn default() -> Self {
         Self::new(Default::default())
@@ -172,6 +172,7 @@ where
 ///
 /// Dropping it unlocks the mutex.
 #[clippy::has_significant_drop]
+#[must_use = "if unused the Mutex will immediately unlock"]
 pub struct MutexGuard<'a, M, T>
 where
     M: RawMutex,
@@ -186,7 +187,7 @@ where
     T: ?Sized,
 {
     /// Returns a locked view over a portion of the locked data.
-    pub fn map<U>(this: Self, fun: impl FnOnce(&mut T) -> &mut U) -> MappedMutexGuard<'a, M, U> {
+    pub fn map<U: ?Sized>(this: Self, fun: impl FnOnce(&mut T) -> &mut U) -> MappedMutexGuard<'a, M, U> {
         let mutex = this.mutex;
         let value = fun(unsafe { &mut *this.mutex.inner.get() });
         // Don't run the `drop` method for MutexGuard. The ownership of the underlying
@@ -278,7 +279,7 @@ where
     T: ?Sized,
 {
     /// Returns a locked view over a portion of the locked data.
-    pub fn map<U>(this: Self, fun: impl FnOnce(&mut T) -> &mut U) -> MappedMutexGuard<'a, M, U> {
+    pub fn map<U: ?Sized>(this: Self, fun: impl FnOnce(&mut T) -> &mut U) -> MappedMutexGuard<'a, M, U> {
         let state = this.state;
         let value = fun(unsafe { &mut *this.value });
         // Don't run the `drop` method for MutexGuard. The ownership of the underlying
diff --git a/embassy-sync/src/once_lock.rs b/embassy-sync/src/once_lock.rs
index 55608ba32..2af19ca20 100644
--- a/embassy-sync/src/once_lock.rs
+++ b/embassy-sync/src/once_lock.rs
@@ -1,7 +1,8 @@
 //! Synchronization primitive for initializing a value once, allowing others to await a reference to the value.
 
 use core::cell::Cell;
-use core::future::poll_fn;
+use core::fmt::{Debug, Formatter};
+use core::future::{Future, poll_fn};
 use core::mem::MaybeUninit;
 use core::sync::atomic::{AtomicBool, Ordering};
 use core::task::Poll;
@@ -42,7 +43,16 @@ pub struct OnceLock<T> {
     data: Cell<MaybeUninit<T>>,
 }
 
-unsafe impl<T> Sync for OnceLock<T> {}
+impl<T> Debug for OnceLock<T> {
+    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
+        f.debug_struct("OnceLock")
+            .field("init", &self.init)
+            .field("data", &"Cell<MaybeUninit<{unprintable}>>")
+            .finish()
+    }
+}
+
+unsafe impl<T> Sync for OnceLock<T> where T: Sync {}
 
 impl<T> OnceLock<T> {
     /// Create a new uninitialized `OnceLock`.
@@ -55,7 +65,7 @@ impl<T> OnceLock<T> {
 
     /// Get a reference to the underlying value, waiting for it to be set.
     /// If the value is already set, this will return immediately.
-    pub async fn get(&self) -> &T {
+    pub fn get(&self) -> impl Future<Output = &T> {
         poll_fn(|cx| match self.try_get() {
             Some(data) => Poll::Ready(data),
             None => {
@@ -63,7 +73,6 @@ impl<T> OnceLock<T> {
                 Poll::Pending
             }
         })
-        .await
     }
 
     /// Try to get a reference to the underlying value if it exists.
diff --git a/embassy-sync/src/pipe.rs b/embassy-sync/src/pipe.rs
index cd5b8ed75..215a556d9 100644
--- a/embassy-sync/src/pipe.rs
+++ b/embassy-sync/src/pipe.rs
@@ -7,12 +7,13 @@ 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::blocking_mutex::raw::RawMutex;
 use crate::ring_buffer::RingBuffer;
 use crate::waitqueue::WakerRegistration;
 
 /// Write-only access to a [`Pipe`].
+#[derive(Debug)]
 pub struct Writer<'p, M, const N: usize>
 where
     M: RawMutex,
@@ -52,6 +53,7 @@ where
 
 /// Future returned by [`Pipe::write`] and  [`Writer::write`].
 #[must_use = "futures do nothing unless you `.await` or poll them"]
+#[derive(Debug)]
 pub struct WriteFuture<'p, M, const N: usize>
 where
     M: RawMutex,
@@ -77,6 +79,7 @@ where
 impl<'p, M, const N: usize> Unpin for WriteFuture<'p, M, N> where M: RawMutex {}
 
 /// Read-only access to a [`Pipe`].
+#[derive(Debug)]
 pub struct Reader<'p, M, const N: usize>
 where
     M: RawMutex,
@@ -128,6 +131,7 @@ where
 
 /// Future returned by [`Pipe::read`] and  [`Reader::read`].
 #[must_use = "futures do nothing unless you `.await` or poll them"]
+#[derive(Debug)]
 pub struct ReadFuture<'p, M, const N: usize>
 where
     M: RawMutex,
@@ -152,8 +156,9 @@ where
 
 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`].
+/// Future returned by [`Reader::fill_buf`].
 #[must_use = "futures do nothing unless you `.await` or poll them"]
+#[derive(Debug)]
 pub struct FillBufFuture<'p, M, const N: usize>
 where
     M: RawMutex,
@@ -199,6 +204,7 @@ pub enum TryWriteError {
     Full,
 }
 
+#[derive(Debug)]
 struct PipeState<const N: usize> {
     buffer: RingBuffer<N>,
     read_waker: WakerRegistration,
@@ -206,6 +212,7 @@ struct PipeState<const N: usize> {
 }
 
 #[repr(transparent)]
+#[derive(Debug)]
 struct Buffer<const N: usize>(UnsafeCell<[u8; N]>);
 
 impl<const N: usize> Buffer<N> {
@@ -230,6 +237,7 @@ unsafe impl<const N: usize> Sync for Buffer<N> {}
 /// 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.
+#[derive(Debug)]
 pub struct Pipe<M, const N: usize>
 where
     M: RawMutex,
@@ -532,6 +540,250 @@ impl<M: RawMutex, const N: usize> embedded_io_async::Write for Writer<'_, M, N>
     }
 }
 
+//
+// Type-erased variants
+//
+
+pub(crate) trait DynamicPipe {
+    fn write<'a>(&'a self, buf: &'a [u8]) -> DynamicWriteFuture<'a>;
+    fn read<'a>(&'a self, buf: &'a mut [u8]) -> DynamicReadFuture<'a>;
+
+    fn try_read(&self, buf: &mut [u8]) -> Result<usize, TryReadError>;
+    fn try_write(&self, buf: &[u8]) -> Result<usize, TryWriteError>;
+
+    fn try_write_with_context(&self, cx: Option<&mut Context<'_>>, buf: &[u8]) -> Result<usize, TryWriteError>;
+    fn try_read_with_context(&self, cx: Option<&mut Context<'_>>, buf: &mut [u8]) -> Result<usize, TryReadError>;
+
+    fn consume(&self, amt: usize);
+    unsafe fn try_fill_buf_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<&[u8], TryReadError>;
+}
+
+impl<M, const N: usize> DynamicPipe for Pipe<M, N>
+where
+    M: RawMutex,
+{
+    fn consume(&self, amt: usize) {
+        Pipe::consume(self, amt)
+    }
+
+    unsafe fn try_fill_buf_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<&[u8], TryReadError> {
+        Pipe::try_fill_buf_with_context(self, cx)
+    }
+
+    fn write<'a>(&'a self, buf: &'a [u8]) -> DynamicWriteFuture<'a> {
+        Pipe::write(self, buf).into()
+    }
+
+    fn read<'a>(&'a self, buf: &'a mut [u8]) -> DynamicReadFuture<'a> {
+        Pipe::read(self, buf).into()
+    }
+
+    fn try_read(&self, buf: &mut [u8]) -> Result<usize, TryReadError> {
+        Pipe::try_read(self, buf)
+    }
+
+    fn try_write(&self, buf: &[u8]) -> Result<usize, TryWriteError> {
+        Pipe::try_write(self, buf)
+    }
+
+    fn try_write_with_context(&self, cx: Option<&mut Context<'_>>, buf: &[u8]) -> Result<usize, TryWriteError> {
+        Pipe::try_write_with_context(self, cx, buf)
+    }
+
+    fn try_read_with_context(&self, cx: Option<&mut Context<'_>>, buf: &mut [u8]) -> Result<usize, TryReadError> {
+        Pipe::try_read_with_context(self, cx, buf)
+    }
+}
+
+/// Write-only access to the dynamic pipe.
+pub struct DynamicWriter<'p> {
+    pipe: &'p dyn DynamicPipe,
+}
+
+impl<'p> Clone for DynamicWriter<'p> {
+    fn clone(&self) -> Self {
+        *self
+    }
+}
+
+impl<'p> Copy for DynamicWriter<'p> {}
+
+impl<'p> DynamicWriter<'p> {
+    /// Write some bytes to the pipe.
+    ///
+    /// See [`Pipe::write()`]
+    pub fn write<'a>(&'a self, buf: &'a [u8]) -> DynamicWriteFuture<'a> {
+        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)
+    }
+}
+
+impl<'p, M, const N: usize> From<Writer<'p, M, N>> for DynamicWriter<'p>
+where
+    M: RawMutex,
+{
+    fn from(value: Writer<'p, M, N>) -> Self {
+        Self { pipe: value.pipe }
+    }
+}
+
+/// Future returned by [`DynamicWriter::write`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct DynamicWriteFuture<'p> {
+    pipe: &'p dyn DynamicPipe,
+    buf: &'p [u8],
+}
+
+impl<'p> Future for DynamicWriteFuture<'p> {
+    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> Unpin for DynamicWriteFuture<'p> {}
+
+impl<'p, M, const N: usize> From<WriteFuture<'p, M, N>> for DynamicWriteFuture<'p>
+where
+    M: RawMutex,
+{
+    fn from(value: WriteFuture<'p, M, N>) -> Self {
+        Self {
+            pipe: value.pipe,
+            buf: value.buf,
+        }
+    }
+}
+
+/// Read-only access to a dynamic pipe.
+pub struct DynamicReader<'p> {
+    pipe: &'p dyn DynamicPipe,
+}
+
+impl<'p> DynamicReader<'p> {
+    /// Read some bytes from the pipe.
+    ///
+    /// See [`Pipe::read()`]
+    pub fn read<'a>(&'a self, buf: &'a mut [u8]) -> DynamicReadFuture<'a> {
+        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) -> DynamicFillBufFuture<'_> {
+        DynamicFillBufFuture { 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)
+    }
+}
+
+impl<'p, M, const N: usize> From<Reader<'p, M, N>> for DynamicReader<'p>
+where
+    M: RawMutex,
+{
+    fn from(value: Reader<'p, M, N>) -> Self {
+        Self { pipe: value.pipe }
+    }
+}
+
+/// Future returned by [`Pipe::read`] and  [`Reader::read`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct DynamicReadFuture<'p> {
+    pipe: &'p dyn DynamicPipe,
+    buf: &'p mut [u8],
+}
+
+impl<'p> Future for DynamicReadFuture<'p> {
+    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> Unpin for DynamicReadFuture<'p> {}
+
+impl<'p, M, const N: usize> From<ReadFuture<'p, M, N>> for DynamicReadFuture<'p>
+where
+    M: RawMutex,
+{
+    fn from(value: ReadFuture<'p, M, N>) -> Self {
+        Self {
+            pipe: value.pipe,
+            buf: value.buf,
+        }
+    }
+}
+
+/// Future returned by [`DynamicReader::fill_buf`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct DynamicFillBufFuture<'p> {
+    pipe: Option<&'p dyn DynamicPipe>,
+}
+
+impl<'p> Future for DynamicFillBufFuture<'p> {
+    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> Unpin for DynamicFillBufFuture<'p> {}
+
+impl<'p, M, const N: usize> From<FillBufFuture<'p, M, N>> for DynamicFillBufFuture<'p>
+where
+    M: RawMutex,
+{
+    fn from(value: FillBufFuture<'p, M, N>) -> Self {
+        Self {
+            pipe: value.pipe.map(|p| p as &dyn DynamicPipe),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use futures_executor::ThreadPool;
@@ -619,6 +871,35 @@ mod tests {
         let _ = w.clone();
     }
 
+    #[test]
+    fn dynamic_dispatch_pipe() {
+        let mut c = Pipe::<NoopRawMutex, 3>::new();
+        let (r, w) = c.split();
+        let (mut r, w): (DynamicReader<'_>, DynamicWriter<'_>) = (r.into(), w.into());
+
+        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);
+    }
+
     #[futures_test::test]
     async fn receiver_receives_given_try_write_async() {
         let executor = ThreadPool::new().unwrap();
diff --git a/embassy-sync/src/priority_channel.rs b/embassy-sync/src/priority_channel.rs
index 24c6c5a7f..1af7d9221 100644
--- a/embassy-sync/src/priority_channel.rs
+++ b/embassy-sync/src/priority_channel.rs
@@ -1,18 +1,18 @@
 //! 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.
+//! Priority is determined by the `Ord` trait. Priority behavior is determined by the [`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;
+pub use heapless::binary_heap::{Kind, Max, Min};
 
-use crate::blocking_mutex::raw::RawMutex;
 use crate::blocking_mutex::Mutex;
+use crate::blocking_mutex::raw::RawMutex;
 use crate::channel::{DynamicChannel, DynamicReceiver, DynamicSender, TryReceiveError, TrySendError};
 use crate::waitqueue::WakerRegistration;
 
@@ -71,6 +71,48 @@ where
     pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
         self.channel.poll_ready_to_send(cx)
     }
+
+    /// Returns the maximum number of elements the channel can hold.
+    ///
+    /// See [`PriorityChannel::capacity()`]
+    pub const fn capacity(&self) -> usize {
+        self.channel.capacity()
+    }
+
+    /// Returns the free capacity of the channel.
+    ///
+    /// See [`PriorityChannel::free_capacity()`]
+    pub fn free_capacity(&self) -> usize {
+        self.channel.free_capacity()
+    }
+
+    /// Clears all elements in the channel.
+    ///
+    /// See [`PriorityChannel::clear()`]
+    pub fn clear(&self) {
+        self.channel.clear();
+    }
+
+    /// Returns the number of elements currently in the channel.
+    ///
+    /// See [`PriorityChannel::len()`]
+    pub fn len(&self) -> usize {
+        self.channel.len()
+    }
+
+    /// Returns whether the channel is empty.
+    ///
+    /// See [`PriorityChannel::is_empty()`]
+    pub fn is_empty(&self) -> bool {
+        self.channel.is_empty()
+    }
+
+    /// Returns whether the channel is full.
+    ///
+    /// See [`PriorityChannel::is_full()`]
+    pub fn is_full(&self) -> bool {
+        self.channel.is_full()
+    }
 }
 
 impl<'ch, M, T, K, const N: usize> From<Sender<'ch, M, T, K, N>> for DynamicSender<'ch, T>
@@ -133,6 +175,16 @@ where
         self.channel.try_receive()
     }
 
+    /// Peek at the next value without removing it from the queue.
+    ///
+    /// See [`PriorityChannel::try_peek()`]
+    pub fn try_peek(&self) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        self.channel.try_peek_with_context(None)
+    }
+
     /// Allows a poll_fn to poll until the channel is ready to receive
     ///
     /// See [`PriorityChannel::poll_ready_to_receive()`]
@@ -146,6 +198,59 @@ where
     pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
         self.channel.poll_receive(cx)
     }
+
+    /// Removes the elements from the channel that satisfy the predicate.
+    ///
+    /// See [`PriorityChannel::remove_if()`]
+    pub fn remove_if<F>(&self, predicate: F)
+    where
+        F: Fn(&T) -> bool,
+        T: Clone,
+    {
+        self.channel.remove_if(predicate)
+    }
+
+    /// Returns the maximum number of elements the channel can hold.
+    ///
+    /// See [`PriorityChannel::capacity()`]
+    pub const fn capacity(&self) -> usize {
+        self.channel.capacity()
+    }
+
+    /// Returns the free capacity of the channel.
+    ///
+    /// See [`PriorityChannel::free_capacity()`]
+    pub fn free_capacity(&self) -> usize {
+        self.channel.free_capacity()
+    }
+
+    /// Clears all elements in the channel.
+    ///
+    /// See [`PriorityChannel::clear()`]
+    pub fn clear(&self) {
+        self.channel.clear();
+    }
+
+    /// Returns the number of elements currently in the channel.
+    ///
+    /// See [`PriorityChannel::len()`]
+    pub fn len(&self) -> usize {
+        self.channel.len()
+    }
+
+    /// Returns whether the channel is empty.
+    ///
+    /// See [`PriorityChannel::is_empty()`]
+    pub fn is_empty(&self) -> bool {
+        self.channel.is_empty()
+    }
+
+    /// Returns whether the channel is full.
+    ///
+    /// See [`PriorityChannel::is_full()`]
+    pub fn is_full(&self) -> bool {
+        self.channel.is_full()
+    }
 }
 
 impl<'ch, M, T, K, const N: usize> From<Receiver<'ch, M, T, K, N>> for DynamicReceiver<'ch, T>
@@ -248,6 +353,31 @@ where
         self.try_receive_with_context(None)
     }
 
+    fn try_peek(&mut self) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        self.try_peek_with_context(None)
+    }
+
+    fn try_peek_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        if self.queue.len() == self.queue.capacity() {
+            self.senders_waker.wake();
+        }
+
+        if let Some(message) = self.queue.peek() {
+            Ok(message.clone())
+        } else {
+            if let Some(cx) = cx {
+                self.receiver_waker.register(cx.waker());
+            }
+            Err(TryReceiveError::Empty)
+        }
+    }
+
     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();
@@ -316,6 +446,9 @@ where
     }
 
     fn clear(&mut self) {
+        if self.queue.len() == self.queue.capacity() {
+            self.senders_waker.wake();
+        }
         self.queue.clear();
     }
 
@@ -340,7 +473,7 @@ where
 /// received from the channel.
 ///
 /// Sent data may be reordered based on their priority within the channel.
-/// For example, in a [`Max`](heapless::binary_heap::Max) [`PriorityChannel`]
+/// For example, in a [`Max`] [`PriorityChannel`]
 /// containing `u32`'s, data sent in the following order `[1, 2, 3]` will be received as `[3, 2, 1]`.
 pub struct PriorityChannel<M, T, K, const N: usize>
 where
@@ -380,6 +513,13 @@ where
         self.lock(|c| c.try_receive_with_context(cx))
     }
 
+    fn try_peek_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        self.lock(|c| c.try_peek_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))
@@ -450,6 +590,37 @@ where
         self.lock(|c| c.try_receive())
     }
 
+    /// Peek at the next value without removing it from the queue.
+    ///
+    /// This method will either receive a copy of the message from the channel immediately or return
+    /// an error if the channel is empty.
+    pub fn try_peek(&self) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        self.lock(|c| c.try_peek())
+    }
+
+    /// Removes elements from the channel based on the given predicate.
+    pub fn remove_if<F>(&self, predicate: F)
+    where
+        F: Fn(&T) -> bool,
+        T: Clone,
+    {
+        self.lock(|c| {
+            let mut new_heap = BinaryHeap::<T, K, N>::new();
+            for item in c.queue.iter() {
+                if !predicate(item) {
+                    match new_heap.push(item.clone()) {
+                        Ok(_) => (),
+                        Err(_) => panic!("Error pushing item to heap"),
+                    }
+                }
+            }
+            c.queue = new_heap;
+        });
+    }
+
     /// Returns the maximum number of elements the channel can hold.
     pub const fn capacity(&self) -> usize {
         N
@@ -499,6 +670,13 @@ where
         PriorityChannel::try_receive_with_context(self, cx)
     }
 
+    fn try_peek_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>
+    where
+        T: Clone,
+    {
+        PriorityChannel::try_peek_with_context(self, cx)
+    }
+
     fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
         PriorityChannel::poll_ready_to_send(self, cx)
     }
@@ -587,6 +765,8 @@ mod tests {
     fn simple_send_and_receive() {
         let c = PriorityChannel::<NoopRawMutex, u32, Max, 3>::new();
         assert!(c.try_send(1).is_ok());
+        assert_eq!(c.try_peek().unwrap(), 1);
+        assert_eq!(c.try_peek().unwrap(), 1);
         assert_eq!(c.try_receive().unwrap(), 1);
     }
 
@@ -607,6 +787,8 @@ mod tests {
         let r: DynamicReceiver<'_, u32> = c.receiver().into();
 
         assert!(s.try_send(1).is_ok());
+        assert_eq!(r.try_peek().unwrap(), 1);
+        assert_eq!(r.try_peek().unwrap(), 1);
         assert_eq!(r.try_receive().unwrap(), 1);
     }
 
@@ -617,11 +799,13 @@ mod tests {
         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!(
+            executor
+                .spawn(async move {
+                    assert!(c2.try_send(1).is_ok());
+                })
+                .is_ok()
+        );
         assert_eq!(c.receive().await, 1);
     }
 
@@ -648,13 +832,15 @@ mod tests {
         // 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());
+        assert!(
+            executor
+                .spawn(async move {
+                    loop {
+                        c.receive().await;
+                    }
+                })
+                .is_ok()
+        );
         send_task_1.unwrap().await;
         send_task_2.unwrap().await;
     }
diff --git a/embassy-sync/src/pubsub/mod.rs b/embassy-sync/src/pubsub/mod.rs
index a97eb7d5b..127a208f1 100644
--- a/embassy-sync/src/pubsub/mod.rs
+++ b/embassy-sync/src/pubsub/mod.rs
@@ -10,8 +10,8 @@ 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::blocking_mutex::raw::RawMutex;
 use crate::waitqueue::MultiWakerRegistration;
 
 pub mod publisher;
@@ -27,8 +27,8 @@ pub use subscriber::{DynSubscriber, Subscriber};
 ///
 /// - 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.
+///   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
 ///
@@ -71,6 +71,7 @@ pub use subscriber::{DynSubscriber, Subscriber};
 /// # block_on(test);
 /// ```
 ///
+#[derive(Debug)]
 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>>>,
 }
@@ -88,7 +89,7 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
     /// 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> {
+    pub fn subscriber(&self) -> Result<Subscriber<'_, M, T, CAP, SUBS, PUBS>, Error> {
         self.inner.lock(|inner| {
             let mut s = inner.borrow_mut();
 
@@ -120,7 +121,7 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
     /// 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> {
+    pub fn publisher(&self) -> Result<Publisher<'_, M, T, CAP, SUBS, PUBS>, Error> {
         self.inner.lock(|inner| {
             let mut s = inner.borrow_mut();
 
@@ -151,13 +152,13 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
 
     /// 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> {
+    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> {
+    pub fn dyn_immediate_publisher(&self) -> DynImmediatePublisher<'_, T> {
         DynImmediatePublisher(ImmediatePub::new(self))
     }
 
@@ -194,6 +195,25 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
     }
 }
 
+impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> crate::pubsub::PubSubBehavior<T>
+    for PubSubChannel<M, T, CAP, SUBS, PUBS>
+{
+    fn publish_immediate(&self, message: T) {
+        self.inner.lock(|s| {
+            let mut s = s.borrow_mut();
+            s.publish_immediate(message)
+        })
+    }
+
+    fn capacity(&self) -> usize {
+        self.capacity()
+    }
+
+    fn is_full(&self) -> bool {
+        self.is_full()
+    }
+}
+
 impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> SealedPubSubBehavior<T>
     for PubSubChannel<M, T, CAP, SUBS, PUBS>
 {
@@ -246,13 +266,6 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
         })
     }
 
-    fn publish_immediate(&self, message: T) {
-        self.inner.lock(|s| {
-            let mut s = s.borrow_mut();
-            s.publish_immediate(message)
-        })
-    }
-
     fn unregister_subscriber(&self, subscriber_next_message_id: u64) {
         self.inner.lock(|s| {
             let mut s = s.borrow_mut();
@@ -267,10 +280,6 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
         })
     }
 
-    fn capacity(&self) -> usize {
-        self.capacity()
-    }
-
     fn free_capacity(&self) -> usize {
         self.free_capacity()
     }
@@ -286,13 +295,10 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
     fn is_empty(&self) -> bool {
         self.is_empty()
     }
-
-    fn is_full(&self) -> bool {
-        self.is_full()
-    }
 }
 
 /// Internal state for the PubSub channel
+#[derive(Debug)]
 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>,
@@ -417,6 +423,9 @@ impl<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> PubSubSta
     }
 
     fn clear(&mut self) {
+        if self.is_full() {
+            self.publisher_wakers.wake();
+        }
         self.queue.clear();
     }
 
@@ -445,8 +454,6 @@ pub enum Error {
     MaximumPublishersReached,
 }
 
-/// 'Middle level' behaviour of the pubsub channel.
-/// This trait is used so that Sub and Pub can be generic over the channel.
 trait SealedPubSubBehavior<T> {
     /// Try to get a message from the queue with the given message id.
     ///
@@ -462,12 +469,6 @@ trait SealedPubSubBehavior<T> {
     /// 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);
-
-    /// Returns the maximum number of elements the channel can hold.
-    fn capacity(&self) -> usize;
-
     /// Returns the free capacity of the channel.
     ///
     /// This is equivalent to `capacity() - len()`
@@ -482,9 +483,6 @@ trait SealedPubSubBehavior<T> {
     /// Returns whether the channel is empty.
     fn is_empty(&self) -> bool;
 
-    /// Returns whether the channel is full.
-    fn is_full(&self) -> bool;
-
     /// Let the channel know that a subscriber has dropped
     fn unregister_subscriber(&self, subscriber_next_message_id: u64);
 
@@ -495,9 +493,16 @@ trait SealedPubSubBehavior<T> {
 /// 'Middle level' behaviour of the pubsub channel.
 /// This trait is used so that Sub and Pub can be generic over the channel.
 #[allow(private_bounds)]
-pub trait PubSubBehavior<T>: SealedPubSubBehavior<T> {}
+pub trait PubSubBehavior<T>: SealedPubSubBehavior<T> {
+    /// Publish a message immediately
+    fn publish_immediate(&self, message: T);
 
-impl<T, C: SealedPubSubBehavior<T>> PubSubBehavior<T> for C {}
+    /// Returns the maximum number of elements the channel can hold.
+    fn capacity(&self) -> usize;
+
+    /// Returns whether the channel is full.
+    fn is_full(&self) -> bool;
+}
 
 /// The result of the subscriber wait procedure
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -755,4 +760,30 @@ mod tests {
         assert_eq!(1, sub0.try_next_message_pure().unwrap().0);
         assert_eq!(0, sub1.try_next_message_pure().unwrap().0);
     }
+
+    #[futures_test::test]
+    async fn publisher_sink() {
+        use futures_util::{SinkExt, StreamExt};
+
+        let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
+
+        let mut sub = channel.subscriber().unwrap();
+
+        let publ = channel.publisher().unwrap();
+        let mut sink = publ.sink();
+
+        sink.send(0).await.unwrap();
+        assert_eq!(0, sub.try_next_message_pure().unwrap());
+
+        sink.send(1).await.unwrap();
+        assert_eq!(1, sub.try_next_message_pure().unwrap());
+
+        sink.send_all(&mut futures_util::stream::iter(0..4).map(Ok))
+            .await
+            .unwrap();
+        assert_eq!(0, sub.try_next_message_pure().unwrap());
+        assert_eq!(1, sub.try_next_message_pure().unwrap());
+        assert_eq!(2, sub.try_next_message_pure().unwrap());
+        assert_eq!(3, sub.try_next_message_pure().unwrap());
+    }
 }
diff --git a/embassy-sync/src/pubsub/publisher.rs b/embassy-sync/src/pubsub/publisher.rs
index e66b3b1db..2a67a0002 100644
--- a/embassy-sync/src/pubsub/publisher.rs
+++ b/embassy-sync/src/pubsub/publisher.rs
@@ -10,6 +10,7 @@ use super::{PubSubBehavior, PubSubChannel};
 use crate::blocking_mutex::raw::RawMutex;
 
 /// A publisher to a channel
+#[derive(Debug)]
 pub struct Pub<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
     /// The channel we are a publisher for
     channel: &'a PSB,
@@ -74,6 +75,12 @@ impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Pub<'a, PSB, T> {
     pub fn is_full(&self) -> bool {
         self.channel.is_full()
     }
+
+    /// Create a [`futures_sink::Sink`] adapter for this publisher.
+    #[inline]
+    pub const fn sink(&self) -> PubSink<'a, '_, PSB, T> {
+        PubSink { publ: self, fut: None }
+    }
 }
 
 impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Drop for Pub<'a, PSB, T> {
@@ -100,6 +107,7 @@ impl<'a, T: Clone> DerefMut for DynPublisher<'a, T> {
 }
 
 /// A publisher that holds a generic reference to the channel
+#[derive(Debug)]
 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>,
 );
@@ -124,6 +132,7 @@ impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS:
 
 /// 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)
+#[derive(Debug)]
 pub struct ImmediatePub<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
     /// The channel we are a publisher for
     channel: &'a PSB,
@@ -199,6 +208,7 @@ impl<'a, T: Clone> DerefMut for DynImmediatePublisher<'a, T> {
 }
 
 /// An immediate publisher that holds a generic reference to the channel
+#[derive(Debug)]
 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>,
 );
@@ -221,8 +231,71 @@ impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS:
     }
 }
 
+#[must_use = "Sinks do nothing unless polled"]
+/// [`futures_sink::Sink`] adapter for [`Pub`].
+#[derive(Debug)]
+pub struct PubSink<'a, 'p, PSB, T>
+where
+    T: Clone,
+    PSB: PubSubBehavior<T> + ?Sized,
+{
+    publ: &'p Pub<'a, PSB, T>,
+    fut: Option<PublisherWaitFuture<'p, 'a, PSB, T>>,
+}
+
+impl<'a, 'p, PSB, T> PubSink<'a, 'p, PSB, T>
+where
+    PSB: PubSubBehavior<T> + ?Sized,
+    T: Clone,
+{
+    /// Try to make progress on the pending future if we have one.
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+        let Some(mut fut) = self.fut.take() else {
+            return Poll::Ready(());
+        };
+
+        if Pin::new(&mut fut).poll(cx).is_pending() {
+            self.fut = Some(fut);
+            return Poll::Pending;
+        }
+
+        Poll::Ready(())
+    }
+}
+
+impl<'a, 'p, PSB, T> futures_sink::Sink<T> for PubSink<'a, 'p, PSB, T>
+where
+    PSB: PubSubBehavior<T> + ?Sized,
+    T: Clone,
+{
+    type Error = core::convert::Infallible;
+
+    #[inline]
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.poll(cx).map(Ok)
+    }
+
+    #[inline]
+    fn start_send(mut self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
+        self.fut = Some(self.publ.publish(item));
+
+        Ok(())
+    }
+
+    #[inline]
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.poll(cx).map(Ok)
+    }
+
+    #[inline]
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.poll(cx).map(Ok)
+    }
+}
+
 /// Future for the publisher wait action
 #[must_use = "futures do nothing unless you `.await` or poll them"]
+#[derive(Debug)]
 pub struct PublisherWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
     /// The message we need to publish
     message: Option<T>,
diff --git a/embassy-sync/src/pubsub/subscriber.rs b/embassy-sync/src/pubsub/subscriber.rs
index 6ad660cb3..356de23f6 100644
--- a/embassy-sync/src/pubsub/subscriber.rs
+++ b/embassy-sync/src/pubsub/subscriber.rs
@@ -10,6 +10,7 @@ use super::{PubSubBehavior, PubSubChannel, WaitResult};
 use crate::blocking_mutex::raw::RawMutex;
 
 /// A subscriber to a channel
+#[derive(Debug)]
 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,
@@ -115,7 +116,7 @@ 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> {
+impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> futures_core::Stream for Sub<'a, PSB, T> {
     type Item = T;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
@@ -151,6 +152,7 @@ impl<'a, T: Clone> DerefMut for DynSubscriber<'a, T> {
 }
 
 /// A subscriber that holds a generic reference to the channel
+#[derive(Debug)]
 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>,
 );
@@ -175,6 +177,7 @@ impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS:
 
 /// Future for the subscriber wait action
 #[must_use = "futures do nothing unless you `.await` or poll them"]
+#[derive(Debug)]
 pub struct SubscriberWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
     subscriber: &'s mut Sub<'a, PSB, T>,
 }
diff --git a/embassy-sync/src/ring_buffer.rs b/embassy-sync/src/ring_buffer.rs
index 81e60c42b..608447cd6 100644
--- a/embassy-sync/src/ring_buffer.rs
+++ b/embassy-sync/src/ring_buffer.rs
@@ -1,5 +1,6 @@
 use core::ops::Range;
 
+#[derive(Debug)]
 pub struct RingBuffer<const N: usize> {
     start: usize,
     end: usize,
@@ -94,11 +95,7 @@ impl<const N: usize> RingBuffer<N> {
 
     fn wrap(&self, n: usize) -> usize {
         assert!(n <= N);
-        if n == N {
-            0
-        } else {
-            n
-        }
+        if n == N { 0 } else { n }
     }
 }
 
diff --git a/embassy-sync/src/rwlock.rs b/embassy-sync/src/rwlock.rs
new file mode 100644
index 000000000..918a6aa41
--- /dev/null
+++ b/embassy-sync/src/rwlock.rs
@@ -0,0 +1,386 @@
+//! Async read-write lock.
+//!
+//! This module provides a read-write lock that can be used to synchronize data between asynchronous tasks.
+use core::cell::{RefCell, UnsafeCell};
+use core::fmt;
+use core::future::{Future, poll_fn};
+use core::ops::{Deref, DerefMut};
+use core::task::Poll;
+
+use crate::blocking_mutex::Mutex as BlockingMutex;
+use crate::blocking_mutex::raw::RawMutex;
+use crate::waitqueue::WakerRegistration;
+
+/// Error returned by [`RwLock::try_read`] and [`RwLock::try_write`] when the lock is already held.
+#[derive(PartialEq, Eq, Clone, Copy, Debug)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
+pub struct TryLockError;
+
+#[derive(Debug)]
+struct State {
+    readers: usize,
+    writer: bool,
+    waker: WakerRegistration,
+}
+
+/// Async read-write lock.
+///
+/// The read-write lock is generic over the raw mutex implementation `M` and the data `T` it protects.
+/// The raw read-write lock is used to guard access to the internal state. It
+/// is held for very short periods only, while locking and unlocking. It is *not* held
+/// for the entire time the async RwLock is locked.
+///
+/// Which implementation you select depends on the context in which you're using the read-write lock.
+///
+/// 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 RwLock<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 RwLock<M, T> {}
+unsafe impl<M: RawMutex + Sync, T: ?Sized + Send> Sync for RwLock<M, T> {}
+
+/// Async read-write lock.
+impl<M, T> RwLock<M, T>
+where
+    M: RawMutex,
+{
+    /// Create a new read-write lock with the given value.
+    pub const fn new(value: T) -> Self {
+        Self {
+            inner: UnsafeCell::new(value),
+            state: BlockingMutex::new(RefCell::new(State {
+                readers: 0,
+                writer: false,
+                waker: WakerRegistration::new(),
+            })),
+        }
+    }
+}
+
+impl<M, T> RwLock<M, T>
+where
+    M: RawMutex,
+    T: ?Sized,
+{
+    /// Lock the read-write lock for reading.
+    ///
+    /// This will wait for the lock to be available if it's already locked for writing.
+    pub fn read(&self) -> impl Future<Output = RwLockReadGuard<'_, M, T>> {
+        poll_fn(|cx| {
+            let ready = self.state.lock(|s| {
+                let mut s = s.borrow_mut();
+                if s.writer {
+                    s.waker.register(cx.waker());
+                    false
+                } else {
+                    s.readers += 1;
+                    true
+                }
+            });
+
+            if ready {
+                Poll::Ready(RwLockReadGuard { rwlock: self })
+            } else {
+                Poll::Pending
+            }
+        })
+    }
+
+    /// Lock the read-write lock for writing.
+    ///
+    /// This will wait for the lock to be available if it's already locked for reading or writing.
+    pub fn write(&self) -> impl Future<Output = RwLockWriteGuard<'_, M, T>> {
+        poll_fn(|cx| {
+            let ready = self.state.lock(|s| {
+                let mut s = s.borrow_mut();
+                if s.writer || s.readers > 0 {
+                    s.waker.register(cx.waker());
+                    false
+                } else {
+                    s.writer = true;
+                    true
+                }
+            });
+
+            if ready {
+                Poll::Ready(RwLockWriteGuard { rwlock: self })
+            } else {
+                Poll::Pending
+            }
+        })
+    }
+
+    /// Attempt to immediately lock the rwlock.
+    ///
+    /// If the rwlock is already locked, this will return an error instead of waiting.
+    pub fn try_read(&self) -> Result<RwLockReadGuard<'_, M, T>, TryLockError> {
+        self.state
+            .lock(|s| {
+                let mut s = s.borrow_mut();
+                if s.writer {
+                    return Err(());
+                }
+                s.readers += 1;
+                Ok(())
+            })
+            .map_err(|_| TryLockError)?;
+
+        Ok(RwLockReadGuard { rwlock: self })
+    }
+
+    /// Attempt to immediately lock the rwlock.
+    ///
+    /// If the rwlock is already locked, this will return an error instead of waiting.
+    pub fn try_write(&self) -> Result<RwLockWriteGuard<'_, M, T>, TryLockError> {
+        self.state
+            .lock(|s| {
+                let mut s = s.borrow_mut();
+                if s.writer || s.readers > 0 {
+                    return Err(());
+                }
+                s.writer = true;
+                Ok(())
+            })
+            .map_err(|_| TryLockError)?;
+
+        Ok(RwLockWriteGuard { rwlock: self })
+    }
+
+    /// Consumes this read-write lock, 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 RwLock 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()
+    }
+}
+
+impl<M: RawMutex, T> From<T> for RwLock<M, T> {
+    fn from(from: T) -> Self {
+        Self::new(from)
+    }
+}
+
+impl<M, T> Default for RwLock<M, T>
+where
+    M: RawMutex,
+    T: Default,
+{
+    fn default() -> Self {
+        Self::new(Default::default())
+    }
+}
+
+impl<M, T> fmt::Debug for RwLock<M, T>
+where
+    M: RawMutex,
+    T: ?Sized + fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut d = f.debug_struct("RwLock");
+        match self.try_read() {
+            Ok(guard) => d.field("inner", &&*guard),
+            Err(TryLockError) => d.field("inner", &"Locked"),
+        }
+        .finish_non_exhaustive()
+    }
+}
+
+/// Async read lock guard.
+///
+/// Owning an instance of this type indicates having
+/// successfully locked the read-write lock for reading, and grants access to the contents.
+///
+/// Dropping it unlocks the read-write lock.
+#[clippy::has_significant_drop]
+#[must_use = "if unused the RwLock will immediately unlock"]
+pub struct RwLockReadGuard<'a, R, T>
+where
+    R: RawMutex,
+    T: ?Sized,
+{
+    rwlock: &'a RwLock<R, T>,
+}
+
+impl<'a, M, T> Drop for RwLockReadGuard<'a, M, T>
+where
+    M: RawMutex,
+    T: ?Sized,
+{
+    fn drop(&mut self) {
+        self.rwlock.state.lock(|s| {
+            let mut s = unwrap!(s.try_borrow_mut());
+            s.readers -= 1;
+            if s.readers == 0 {
+                s.waker.wake();
+            }
+        })
+    }
+}
+
+impl<'a, M, T> Deref for RwLockReadGuard<'a, M, T>
+where
+    M: RawMutex,
+    T: ?Sized,
+{
+    type Target = T;
+    fn deref(&self) -> &Self::Target {
+        // Safety: the RwLockReadGuard represents shared access to the contents
+        // of the read-write lock, so it's OK to get it.
+        unsafe { &*(self.rwlock.inner.get() as *const T) }
+    }
+}
+
+impl<'a, M, T> fmt::Debug for RwLockReadGuard<'a, M, T>
+where
+    M: RawMutex,
+    T: ?Sized + fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Debug::fmt(&**self, f)
+    }
+}
+
+impl<'a, M, T> fmt::Display for RwLockReadGuard<'a, M, T>
+where
+    M: RawMutex,
+    T: ?Sized + fmt::Display,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(&**self, f)
+    }
+}
+
+/// Async write lock guard.
+///
+/// Owning an instance of this type indicates having
+/// successfully locked the read-write lock for writing, and grants access to the contents.
+///
+/// Dropping it unlocks the read-write lock.
+#[clippy::has_significant_drop]
+#[must_use = "if unused the RwLock will immediately unlock"]
+pub struct RwLockWriteGuard<'a, R, T>
+where
+    R: RawMutex,
+    T: ?Sized,
+{
+    rwlock: &'a RwLock<R, T>,
+}
+
+impl<'a, R, T> Drop for RwLockWriteGuard<'a, R, T>
+where
+    R: RawMutex,
+    T: ?Sized,
+{
+    fn drop(&mut self) {
+        self.rwlock.state.lock(|s| {
+            let mut s = unwrap!(s.try_borrow_mut());
+            s.writer = false;
+            s.waker.wake();
+        })
+    }
+}
+
+impl<'a, R, T> Deref for RwLockWriteGuard<'a, R, T>
+where
+    R: RawMutex,
+    T: ?Sized,
+{
+    type Target = T;
+    fn deref(&self) -> &Self::Target {
+        // Safety: the RwLockWriteGuard represents exclusive access to the contents
+        // of the read-write lock, so it's OK to get it.
+        unsafe { &*(self.rwlock.inner.get() as *mut T) }
+    }
+}
+
+impl<'a, R, T> DerefMut for RwLockWriteGuard<'a, R, T>
+where
+    R: RawMutex,
+    T: ?Sized,
+{
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        // Safety: the RwLockWriteGuard represents exclusive access to the contents
+        // of the read-write lock, so it's OK to get it.
+        unsafe { &mut *(self.rwlock.inner.get()) }
+    }
+}
+
+impl<'a, R, T> fmt::Debug for RwLockWriteGuard<'a, R, T>
+where
+    R: RawMutex,
+    T: ?Sized + fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Debug::fmt(&**self, f)
+    }
+}
+
+impl<'a, R, T> fmt::Display for RwLockWriteGuard<'a, R, T>
+where
+    R: RawMutex,
+    T: ?Sized + fmt::Display,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(&**self, f)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::blocking_mutex::raw::NoopRawMutex;
+    use crate::rwlock::RwLock;
+
+    #[futures_test::test]
+    async fn read_guard_releases_lock_when_dropped() {
+        let rwlock: RwLock<NoopRawMutex, [i32; 2]> = RwLock::new([0, 1]);
+
+        {
+            let guard = rwlock.read().await;
+            assert_eq!(*guard, [0, 1]);
+        }
+
+        {
+            let guard = rwlock.read().await;
+            assert_eq!(*guard, [0, 1]);
+        }
+
+        assert_eq!(*rwlock.read().await, [0, 1]);
+    }
+
+    #[futures_test::test]
+    async fn write_guard_releases_lock_when_dropped() {
+        let rwlock: RwLock<NoopRawMutex, [i32; 2]> = RwLock::new([0, 1]);
+
+        {
+            let mut guard = rwlock.write().await;
+            assert_eq!(*guard, [0, 1]);
+            guard[1] = 2;
+        }
+
+        {
+            let guard = rwlock.read().await;
+            assert_eq!(*guard, [0, 2]);
+        }
+
+        assert_eq!(*rwlock.read().await, [0, 2]);
+    }
+}
diff --git a/embassy-sync/src/semaphore.rs b/embassy-sync/src/semaphore.rs
index d30eee30b..8d2413931 100644
--- a/embassy-sync/src/semaphore.rs
+++ b/embassy-sync/src/semaphore.rs
@@ -1,13 +1,13 @@
 //! A synchronization primitive for controlling access to a pool of resources.
 use core::cell::{Cell, RefCell};
 use core::convert::Infallible;
-use core::future::{poll_fn, Future};
+use core::future::{Future, poll_fn};
 use core::task::{Poll, Waker};
 
 use heapless::Deque;
 
-use crate::blocking_mutex::raw::RawMutex;
 use crate::blocking_mutex::Mutex;
+use crate::blocking_mutex::raw::RawMutex;
 use crate::waitqueue::WakerRegistration;
 
 /// An asynchronous semaphore.
@@ -46,6 +46,7 @@ pub trait Semaphore: Sized {
 /// A representation of a number of acquired permits.
 ///
 /// The acquired permits will be released back to the [`Semaphore`] when this is dropped.
+#[derive(Debug)]
 pub struct SemaphoreReleaser<'a, S: Semaphore> {
     semaphore: &'a S,
     permits: usize,
@@ -181,6 +182,7 @@ impl<M: RawMutex> Semaphore for GreedySemaphore<M> {
     }
 }
 
+#[derive(Debug)]
 struct SemaphoreState {
     permits: usize,
     waker: WakerRegistration,
@@ -221,6 +223,7 @@ impl SemaphoreState {
 ///
 /// Up to `N` tasks may attempt to acquire permits concurrently. If additional
 /// tasks attempt to acquire a permit, a [`WaitQueueFull`] error will be returned.
+#[derive(Debug)]
 pub struct FairSemaphore<M, const N: usize>
 where
     M: RawMutex,
@@ -341,6 +344,7 @@ impl<M: RawMutex, const N: usize> Semaphore for FairSemaphore<M, N> {
     }
 }
 
+#[derive(Debug)]
 struct FairAcquire<'a, M: RawMutex, const N: usize> {
     sema: &'a FairSemaphore<M, N>,
     permits: usize,
@@ -364,6 +368,7 @@ impl<'a, M: RawMutex, const N: usize> core::future::Future for FairAcquire<'a, M
     }
 }
 
+#[derive(Debug)]
 struct FairAcquireAll<'a, M: RawMutex, const N: usize> {
     sema: &'a FairSemaphore<M, N>,
     min: usize,
@@ -387,6 +392,7 @@ impl<'a, M: RawMutex, const N: usize> core::future::Future for FairAcquireAll<'a
     }
 }
 
+#[derive(Debug)]
 struct FairSemaphoreState<const N: usize> {
     permits: usize,
     next_ticket: usize,
diff --git a/embassy-sync/src/signal.rs b/embassy-sync/src/signal.rs
index a0f4b5a74..cc02228cf 100644
--- a/embassy-sync/src/signal.rs
+++ b/embassy-sync/src/signal.rs
@@ -1,12 +1,12 @@
 //! A synchronization primitive for passing the latest value to a task.
 use core::cell::Cell;
-use core::future::{poll_fn, Future};
+use core::future::{Future, poll_fn};
 use core::task::{Context, Poll, Waker};
 
-use crate::blocking_mutex::raw::RawMutex;
 use crate::blocking_mutex::Mutex;
+use crate::blocking_mutex::raw::RawMutex;
 
-/// Single-slot signaling primitive.
+/// Single-slot signaling primitive for a _single_ consumer.
 ///
 /// 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
@@ -17,6 +17,7 @@ use crate::blocking_mutex::Mutex;
 /// updates.
 ///
 /// For more advanced use cases, you might want to use [`Channel`](crate::channel::Channel) instead.
+/// For multiple consumers, use [`Watch`](crate::watch::Watch) instead.
 ///
 /// Signals are generally declared as `static`s and then borrowed as required.
 ///
@@ -38,6 +39,7 @@ where
     state: Mutex<M, Cell<State<T>>>,
 }
 
+#[derive(Debug)]
 enum State<T> {
     None,
     Waiting(Waker),
@@ -81,7 +83,7 @@ where
 
     /// Remove the queued value in this `Signal`, if any.
     pub fn reset(&self) {
-        self.state.lock(|cell| cell.set(State::None));
+        self.try_take();
     }
 
     fn poll_wait(&self, cx: &mut Context<'_>) -> Poll<T> {
@@ -106,7 +108,7 @@ where
         })
     }
 
-    /// Future that completes when this Signal has been signaled.
+    /// Future that completes when this Signal has been signaled, taking the value out of the signal.
     pub fn wait(&self) -> impl Future<Output = T> + '_ {
         poll_fn(move |cx| self.poll_wait(cx))
     }
diff --git a/embassy-sync/src/waitqueue/atomic_waker.rs b/embassy-sync/src/waitqueue/atomic_waker.rs
index 63fe04a6e..d2bf890e5 100644
--- a/embassy-sync/src/waitqueue/atomic_waker.rs
+++ b/embassy-sync/src/waitqueue/atomic_waker.rs
@@ -1,26 +1,28 @@
 use core::cell::Cell;
 use core::task::Waker;
 
-use crate::blocking_mutex::raw::CriticalSectionRawMutex;
 use crate::blocking_mutex::Mutex;
+use crate::blocking_mutex::raw::{CriticalSectionRawMutex, RawMutex};
 
 /// Utility struct to register and wake a waker.
-pub struct AtomicWaker {
-    waker: Mutex<CriticalSectionRawMutex, Cell<Option<Waker>>>,
+/// If a waker is registered, registering another waker will replace the previous one without waking it.
+/// Intended to wake a task from an interrupt. Therefore, it is generally not expected,
+/// that multiple tasks register try to register a waker simultaneously.
+pub struct GenericAtomicWaker<M: RawMutex> {
+    waker: Mutex<M, Cell<Option<Waker>>>,
 }
 
-impl AtomicWaker {
+impl<M: RawMutex> GenericAtomicWaker<M> {
     /// Create a new `AtomicWaker`.
-    pub const fn new() -> Self {
+    pub const fn new(mutex: M) -> Self {
         Self {
-            waker: Mutex::const_new(CriticalSectionRawMutex::new(), Cell::new(None)),
+            waker: Mutex::const_new(mutex, 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);
+        self.waker.lock(|cell| {
             cell.set(match cell.replace(None) {
                 Some(w2) if (w2.will_wake(w)) => Some(w2),
                 _ => Some(w.clone()),
@@ -30,8 +32,7 @@ impl AtomicWaker {
 
     /// Wake the registered waker, if any.
     pub fn wake(&self) {
-        critical_section::with(|cs| {
-            let cell = self.waker.borrow(cs);
+        self.waker.lock(|cell| {
             if let Some(w) = cell.replace(None) {
                 w.wake_by_ref();
                 cell.set(Some(w));
@@ -39,3 +40,27 @@ impl AtomicWaker {
         })
     }
 }
+
+/// Utility struct to register and wake a waker.
+pub struct AtomicWaker {
+    waker: GenericAtomicWaker<CriticalSectionRawMutex>,
+}
+
+impl AtomicWaker {
+    /// Create a new `AtomicWaker`.
+    pub const fn new() -> Self {
+        Self {
+            waker: GenericAtomicWaker::new(CriticalSectionRawMutex::new()),
+        }
+    }
+
+    /// Register a waker. Overwrites the previous waker, if any.
+    pub fn register(&self, w: &Waker) {
+        self.waker.register(w);
+    }
+
+    /// Wake the registered waker, if any.
+    pub fn wake(&self) {
+        self.waker.wake();
+    }
+}
diff --git a/embassy-sync/src/waitqueue/atomic_waker_turbo.rs b/embassy-sync/src/waitqueue/atomic_waker_turbo.rs
index 5c6a96ec8..a45adeab8 100644
--- a/embassy-sync/src/waitqueue/atomic_waker_turbo.rs
+++ b/embassy-sync/src/waitqueue/atomic_waker_turbo.rs
@@ -4,6 +4,10 @@ use core::sync::atomic::{AtomicPtr, Ordering};
 use core::task::Waker;
 
 /// Utility struct to register and wake a waker.
+/// If a waker is registered, registering another waker will replace the previous one without waking it.
+/// The intended use case is to wake tasks from interrupts. Therefore, it is generally not expected,
+/// that multiple tasks register try to register a waker simultaneously.
+#[derive(Debug)]
 pub struct AtomicWaker {
     waker: AtomicPtr<()>,
 }
diff --git a/embassy-sync/src/waitqueue/multi_waker.rs b/embassy-sync/src/waitqueue/multi_waker.rs
index 0e520bf40..56c0cd1b2 100644
--- a/embassy-sync/src/waitqueue/multi_waker.rs
+++ b/embassy-sync/src/waitqueue/multi_waker.rs
@@ -3,6 +3,9 @@ use core::task::Waker;
 use heapless::Vec;
 
 /// Utility struct to register and wake multiple wakers.
+/// Queue of wakers with a maximum length of `N`.
+/// Intended for waking multiple tasks.
+#[derive(Debug)]
 pub struct MultiWakerRegistration<const N: usize> {
     wakers: Vec<Waker, N>,
 }
@@ -13,7 +16,9 @@ impl<const N: usize> MultiWakerRegistration<N> {
         Self { wakers: Vec::new() }
     }
 
-    /// Register a waker. If the buffer is full the function returns it in the error
+    /// Register a waker.
+    ///
+    /// If the buffer is full, [wakes all the wakers](Self::wake), clears its buffer and registers the waker.
     pub fn register(&mut self, w: &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.
diff --git a/embassy-sync/src/waitqueue/waker_registration.rs b/embassy-sync/src/waitqueue/waker_registration.rs
index 9b666e7c4..7f24f8fb6 100644
--- a/embassy-sync/src/waitqueue/waker_registration.rs
+++ b/embassy-sync/src/waitqueue/waker_registration.rs
@@ -2,6 +2,10 @@ use core::mem;
 use core::task::Waker;
 
 /// Utility struct to register and wake a waker.
+/// If a waker is registered, registering another waker will replace the previous one.
+/// The previous waker will be woken in this case, giving it a chance to reregister itself.
+/// Although it is possible to wake multiple tasks this way,
+/// this will cause them to wake each other in a loop registering themselves.
 #[derive(Debug, Default)]
 pub struct WakerRegistration {
     waker: Option<Waker>,
diff --git a/embassy-sync/src/watch.rs b/embassy-sync/src/watch.rs
new file mode 100644
index 000000000..0f8a8d679
--- /dev/null
+++ b/embassy-sync/src/watch.rs
@@ -0,0 +1,1127 @@
+//! A synchronization primitive for passing the latest value to **multiple** receivers.
+
+use core::cell::RefCell;
+use core::future::{Future, poll_fn};
+use core::marker::PhantomData;
+use core::ops::{Deref, DerefMut};
+use core::task::{Context, Poll};
+
+use crate::blocking_mutex::Mutex;
+use crate::blocking_mutex::raw::RawMutex;
+use crate::waitqueue::MultiWakerRegistration;
+
+/// The `Watch` is a single-slot signaling primitive that allows _multiple_ (`N`) receivers to concurrently await
+/// changes to the value. Unlike a [`Signal`](crate::signal::Signal), `Watch` supports multiple receivers,
+/// and unlike a [`PubSubChannel`](crate::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.
+///
+/// Typically, `Watch` instances are declared as `static`, and a [`Sender`] and [`Receiver`]
+/// (or [`DynSender`] and/or [`DynReceiver`]) are obtained where relevant. An [`AnonReceiver`]
+/// and [`DynAnonReceiver`] are also available, which do not increase the receiver count for the
+/// channel, and unwrapping is therefore not required, but it is not possible to `.await` the channel.
+/// ```
+///
+/// use futures_executor::block_on;
+/// use embassy_sync::watch::Watch;
+/// use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
+///
+/// let f = async {
+///
+/// static WATCH: Watch<CriticalSectionRawMutex, u8, 2> = Watch::new();
+///
+/// // Obtain receivers and sender
+/// let mut rcv0 = WATCH.receiver().unwrap();
+/// let mut rcv1 = WATCH.dyn_receiver().unwrap();
+/// let mut snd = WATCH.sender();
+///
+/// // No more receivers, and no update
+/// assert!(WATCH.receiver().is_none());
+/// assert_eq!(rcv1.try_changed(), None);
+///
+/// snd.send(10);
+///
+/// // Receive the new value (async or try)
+/// assert_eq!(rcv0.changed().await, 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.get().await, 20);
+/// assert_eq!(rcv1.try_changed(), None);
+///
+/// // But `get` also returns when unchanged
+/// assert_eq!(rcv1.get().await, 20);
+/// assert_eq!(rcv1.get().await, 20);
+///
+/// };
+/// block_on(f);
+/// ```
+#[derive(Debug)]
+pub struct Watch<M: RawMutex, T: Clone, const N: usize> {
+    mutex: Mutex<M, RefCell<WatchState<T, N>>>,
+}
+
+#[derive(Debug)]
+struct WatchState<T: Clone, const N: usize> {
+    data: Option<T>,
+    current_id: u64,
+    wakers: MultiWakerRegistration<N>,
+    receiver_count: usize,
+}
+
+trait SealedWatchBehavior<T> {
+    /// Poll the `Watch` for the current value, making it as seen.
+    fn poll_get(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll<T>;
+
+    /// Poll the `Watch` for the value if it matches the predicate function
+    /// `f`, making it as seen.
+    fn poll_get_and(&self, id: &mut u64, f: &mut dyn Fn(&T) -> bool, cx: &mut Context<'_>) -> Poll<T>;
+
+    /// Poll the `Watch` for a changed value, marking it as seen, if an id is given.
+    fn poll_changed(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll<T>;
+
+    /// Tries to retrieve the value of the `Watch` if it has changed, marking it as seen.
+    fn try_changed(&self, id: &mut u64) -> Option<T>;
+
+    /// Poll the `Watch` for a changed value that matches the predicate function
+    /// `f`, marking it as seen.
+    fn poll_changed_and(&self, id: &mut u64, f: &mut dyn Fn(&T) -> bool, cx: &mut Context<'_>) -> Poll<T>;
+
+    /// Tries to retrieve the value of the `Watch` if it has changed and matches the
+    /// predicate function `f`, marking it as seen.
+    fn try_changed_and(&self, id: &mut u64, f: &mut dyn Fn(&T) -> bool) -> Option<T>;
+
+    /// Used when a receiver is dropped to decrement the receiver count.
+    ///
+    /// ## This method should not be called by the user.
+    fn drop_receiver(&self);
+
+    /// Clears the value of the `Watch`.
+    fn clear(&self);
+
+    /// Sends a new value to the `Watch`.
+    fn send(&self, val: T);
+
+    /// Modify the value of the `Watch` using a closure. Returns `false` if the
+    /// `Watch` does not already contain a value.
+    fn send_modify(&self, f: &mut dyn Fn(&mut Option<T>));
+
+    /// Modify the value of the `Watch` using a closure. Returns `false` if the
+    /// `Watch` does not already contain a value.
+    fn send_if_modified(&self, f: &mut dyn Fn(&mut Option<T>) -> bool);
+}
+
+/// 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 u64>) -> Option<T>;
+
+    /// Tries to get the value of the `Watch` if it matches the predicate function
+    /// `f`, marking it as seen.
+    fn try_get_and(&self, id: Option<&mut u64>, f: &mut dyn Fn(&T) -> bool) -> Option<T>;
+
+    /// Checks if the `Watch` is been initialized with a value.
+    fn contains_value(&self) -> bool;
+}
+
+impl<M: RawMutex, T: Clone, const N: usize> SealedWatchBehavior<T> for Watch<M, T, N> {
+    fn poll_get(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll<T> {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            match &s.data {
+                Some(data) => {
+                    *id = s.current_id;
+                    Poll::Ready(data.clone())
+                }
+                None => {
+                    s.wakers.register(cx.waker());
+                    Poll::Pending
+                }
+            }
+        })
+    }
+
+    fn poll_get_and(&self, id: &mut u64, f: &mut dyn Fn(&T) -> bool, cx: &mut Context<'_>) -> Poll<T> {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            match s.data {
+                Some(ref data) if f(data) => {
+                    *id = s.current_id;
+                    Poll::Ready(data.clone())
+                }
+                _ => {
+                    s.wakers.register(cx.waker());
+                    Poll::Pending
+                }
+            }
+        })
+    }
+
+    fn poll_changed(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll<T> {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            match (&s.data, s.current_id > *id) {
+                (Some(data), true) => {
+                    *id = s.current_id;
+                    Poll::Ready(data.clone())
+                }
+                _ => {
+                    s.wakers.register(cx.waker());
+                    Poll::Pending
+                }
+            }
+        })
+    }
+
+    fn try_changed(&self, id: &mut u64) -> Option<T> {
+        self.mutex.lock(|state| {
+            let s = state.borrow();
+            match s.current_id > *id {
+                true => {
+                    *id = s.current_id;
+                    s.data.clone()
+                }
+                false => None,
+            }
+        })
+    }
+
+    fn poll_changed_and(&self, id: &mut u64, f: &mut dyn Fn(&T) -> bool, cx: &mut Context<'_>) -> Poll<T> {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            match (&s.data, s.current_id > *id) {
+                (Some(data), true) if f(data) => {
+                    *id = s.current_id;
+                    Poll::Ready(data.clone())
+                }
+                _ => {
+                    s.wakers.register(cx.waker());
+                    Poll::Pending
+                }
+            }
+        })
+    }
+
+    fn try_changed_and(&self, id: &mut u64, f: &mut dyn Fn(&T) -> bool) -> Option<T> {
+        self.mutex.lock(|state| {
+            let s = state.borrow();
+            match (&s.data, s.current_id > *id) {
+                (Some(data), true) if f(data) => {
+                    *id = s.current_id;
+                    s.data.clone()
+                }
+                _ => None,
+            }
+        })
+    }
+
+    fn drop_receiver(&self) {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            s.receiver_count -= 1;
+        })
+    }
+
+    fn clear(&self) {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            s.data = None;
+        })
+    }
+
+    fn send(&self, val: T) {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            s.data = Some(val);
+            s.current_id += 1;
+            s.wakers.wake();
+        })
+    }
+
+    fn send_modify(&self, f: &mut dyn Fn(&mut Option<T>)) {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            f(&mut s.data);
+            s.current_id += 1;
+            s.wakers.wake();
+        })
+    }
+
+    fn send_if_modified(&self, f: &mut dyn Fn(&mut Option<T>) -> bool) {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            if f(&mut s.data) {
+                s.current_id += 1;
+                s.wakers.wake();
+            }
+        })
+    }
+}
+
+impl<M: RawMutex, T: Clone, const N: usize> WatchBehavior<T> for Watch<M, T, N> {
+    fn try_get(&self, id: Option<&mut u64>) -> Option<T> {
+        self.mutex.lock(|state| {
+            let s = state.borrow();
+            if let Some(id) = id {
+                *id = s.current_id;
+            }
+            s.data.clone()
+        })
+    }
+
+    fn try_get_and(&self, id: Option<&mut u64>, f: &mut dyn Fn(&T) -> bool) -> Option<T> {
+        self.mutex.lock(|state| {
+            let s = state.borrow();
+            match s.data {
+                Some(ref data) if f(data) => {
+                    if let Some(id) = id {
+                        *id = s.current_id;
+                    }
+                    Some(data.clone())
+                }
+                _ => None,
+            }
+        })
+    }
+
+    fn contains_value(&self) -> bool {
+        self.mutex.lock(|state| state.borrow().data.is_some())
+    }
+}
+
+impl<M: RawMutex, T: Clone, const N: usize> Watch<M, T, N> {
+    /// Create a new `Watch` channel for `N` receivers.
+    pub const fn new() -> Self {
+        Self {
+            mutex: Mutex::new(RefCell::new(WatchState {
+                data: None,
+                current_id: 0,
+                wakers: MultiWakerRegistration::new(),
+                receiver_count: 0,
+            })),
+        }
+    }
+
+    /// Create a new `Watch` channel with default data.
+    pub const fn new_with(data: T) -> Self {
+        Self {
+            mutex: Mutex::new(RefCell::new(WatchState {
+                data: Some(data),
+                current_id: 0,
+                wakers: MultiWakerRegistration::new(),
+                receiver_count: 0,
+            })),
+        }
+    }
+
+    /// Create a new [`Sender`] for the `Watch`.
+    pub fn sender(&self) -> Sender<'_, M, T, N> {
+        Sender(Snd::new(self))
+    }
+
+    /// Create a new [`DynSender`] for the `Watch`.
+    pub fn dyn_sender(&self) -> DynSender<'_, T> {
+        DynSender(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) -> Option<Receiver<'_, M, T, N>> {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            if s.receiver_count < N {
+                s.receiver_count += 1;
+                Some(Receiver(Rcv::new(self, 0)))
+            } else {
+                None
+            }
+        })
+    }
+
+    /// Try to create a new [`DynReceiver`] for the `Watch`. If the
+    /// maximum number of receivers has been reached, `None` is returned.
+    pub fn dyn_receiver(&self) -> Option<DynReceiver<'_, T>> {
+        self.mutex.lock(|state| {
+            let mut s = state.borrow_mut();
+            if s.receiver_count < N {
+                s.receiver_count += 1;
+                Some(DynReceiver(Rcv::new(self, 0)))
+            } else {
+                None
+            }
+        })
+    }
+
+    /// Try to create a new [`AnonReceiver`] for the `Watch`.
+    pub fn anon_receiver(&self) -> AnonReceiver<'_, M, T, N> {
+        AnonReceiver(AnonRcv::new(self, 0))
+    }
+
+    /// Try to create a new [`DynAnonReceiver`] for the `Watch`.
+    pub fn dyn_anon_receiver(&self) -> DynAnonReceiver<'_, T> {
+        DynAnonReceiver(AnonRcv::new(self, 0))
+    }
+
+    /// 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) -> u64 {
+        self.mutex.lock(|state| state.borrow().current_id)
+    }
+
+    /// Tries to get the value of the `Watch`.
+    pub fn try_get(&self) -> Option<T> {
+        WatchBehavior::try_get(self, None)
+    }
+
+    /// Tries to get the value of the `Watch` if it matches the predicate function `f`.
+    pub fn try_get_and<F>(&self, mut f: F) -> Option<T>
+    where
+        F: Fn(&T) -> bool,
+    {
+        WatchBehavior::try_get_and(self, None, &mut f)
+    }
+}
+
+/// 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)
+    }
+
+    /// Tries to peek the current value of the `Watch` if it matches the predicate
+    /// function `f`.
+    pub fn try_get_and<F>(&self, mut f: F) -> Option<T>
+    where
+        F: Fn(&T) -> bool,
+    {
+        self.watch.try_get_and(None, &mut f)
+    }
+
+    /// Returns true if the `Watch` contains a value.
+    pub fn contains_value(&self) -> bool {
+        self.watch.contains_value()
+    }
+
+    /// Modify the value of the `Watch` using a closure.
+    pub fn send_modify<F>(&self, mut f: F)
+    where
+        F: Fn(&mut Option<T>),
+    {
+        self.watch.send_modify(&mut f)
+    }
+
+    /// Modify the value of the `Watch` using a closure. The closure must return
+    /// `true` if the value was modified, which notifies all receivers.
+    pub fn send_if_modified<F>(&self, mut f: F)
+    where
+        F: Fn(&mut Option<T>) -> bool,
+    {
+        self.watch.send_if_modified(&mut f)
+    }
+}
+
+/// 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, const N: usize>(Snd<'a, T, Watch<M, T, N>>);
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> Clone for Sender<'a, M, T, N> {
+    fn clone(&self) -> Self {
+        Self(self.0.clone())
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> Sender<'a, M, T, N> {
+    /// Converts the `Sender` into a [`DynSender`].
+    pub fn as_dyn(self) -> DynSender<'a, T> {
+        DynSender(Snd::new(self.watch))
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> Into<DynSender<'a, T>> for Sender<'a, M, T, N> {
+    fn into(self) -> DynSender<'a, T> {
+        self.as_dyn()
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> Deref for Sender<'a, M, T, N> {
+    type Target = Snd<'a, T, Watch<M, T, N>>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> DerefMut for Sender<'a, M, T, N> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+/// A sender which holds a **dynamic** reference to a `Watch` channel.
+///
+/// This is an alternative to [`Sender`] with a simpler type definition,
+pub struct DynSender<'a, T: Clone>(Snd<'a, T, dyn WatchBehavior<T> + 'a>);
+
+impl<'a, T: Clone> Clone for DynSender<'a, T> {
+    fn clone(&self) -> Self {
+        Self(self.0.clone())
+    }
+}
+
+impl<'a, T: Clone> Deref for DynSender<'a, T> {
+    type Target = Snd<'a, T, dyn WatchBehavior<T> + 'a>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<'a, T: Clone> DerefMut for DynSender<'a, T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+/// A receiver can `.await` a change in the `Watch` value.
+pub struct Rcv<'a, T: Clone, W: WatchBehavior<T> + ?Sized> {
+    watch: &'a W,
+    at_id: u64,
+    _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, at_id: u64) -> Self {
+        Self {
+            watch,
+            at_id,
+            _phantom: PhantomData,
+        }
+    }
+
+    /// Returns the current value of the `Watch` once it is initialized, marking it as seen.
+    ///
+    /// **Note**: Futures do nothing unless you `.await` or poll them.
+    pub fn get(&mut self) -> impl Future<Output = T> + '_ {
+        poll_fn(|cx| self.watch.poll_get(&mut self.at_id, cx))
+    }
+
+    /// 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))
+    }
+
+    /// Returns the value of the `Watch` if it matches the predicate function `f`,
+    /// or waits for it to match, marking it as seen.
+    ///
+    /// **Note**: Futures do nothing unless you `.await` or poll them.
+    pub async fn get_and<F>(&mut self, mut f: F) -> T
+    where
+        F: Fn(&T) -> bool,
+    {
+        poll_fn(|cx| self.watch.poll_get_and(&mut self.at_id, &mut f, cx)).await
+    }
+
+    /// Tries to get the current value of the `Watch` if it matches the predicate
+    /// function `f` without waiting, marking it as seen.
+    pub fn try_get_and<F>(&mut self, mut f: F) -> Option<T>
+    where
+        F: Fn(&T) -> bool,
+    {
+        self.watch.try_get_and(Some(&mut self.at_id), &mut f)
+    }
+
+    /// Waits for the `Watch` to change and returns the new value, marking it as seen.
+    ///
+    /// **Note**: Futures do nothing unless you `.await` or poll them.
+    pub async fn changed(&mut self) -> T {
+        poll_fn(|cx| self.watch.poll_changed(&mut self.at_id, cx)).await
+    }
+
+    /// 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)
+    }
+
+    /// Waits for the `Watch` to change to a value which satisfies the predicate
+    /// function `f` and returns the new value, marking it as seen.
+    ///
+    /// **Note**: Futures do nothing unless you `.await` or poll them.
+    pub async fn changed_and<F>(&mut self, mut f: F) -> T
+    where
+        F: Fn(&T) -> bool,
+    {
+        poll_fn(|cx| self.watch.poll_changed_and(&mut self.at_id, &mut f, cx)).await
+    }
+
+    /// Tries to get the new value of the watch which satisfies the predicate
+    /// function `f` and returns the new value without waiting, marking it as seen.
+    pub fn try_changed_and<F>(&mut self, mut f: F) -> Option<T>
+    where
+        F: Fn(&T) -> bool,
+    {
+        self.watch.try_changed_and(&mut self.at_id, &mut f)
+    }
+
+    /// 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()
+    }
+}
+
+impl<'a, T: Clone, W: WatchBehavior<T> + ?Sized> Drop for Rcv<'a, T, W> {
+    fn drop(&mut self) {
+        self.watch.drop_receiver();
+    }
+}
+
+/// A anonymous receiver can NOT `.await` a change in the `Watch` value.
+#[derive(Debug)]
+pub struct AnonRcv<'a, T: Clone, W: WatchBehavior<T> + ?Sized> {
+    watch: &'a W,
+    at_id: u64,
+    _phantom: PhantomData<T>,
+}
+
+impl<'a, T: Clone, W: WatchBehavior<T> + ?Sized> AnonRcv<'a, T, W> {
+    /// Creates a new `Receiver` with a reference to the `Watch`.
+    fn new(watch: &'a W, at_id: u64) -> Self {
+        Self {
+            watch,
+            at_id,
+            _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 current value of the `Watch` if it matches the predicate
+    /// function `f` without waiting, marking it as seen.
+    pub fn try_get_and<F>(&mut self, mut f: F) -> Option<T>
+    where
+        F: Fn(&T) -> bool,
+    {
+        self.watch.try_get_and(Some(&mut self.at_id), &mut f)
+    }
+
+    /// 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)
+    }
+
+    /// Tries to get the new value of the watch which satisfies the predicate
+    /// function `f` and returns the new value without waiting, marking it as seen.
+    pub fn try_changed_and<F>(&mut self, mut f: F) -> Option<T>
+    where
+        F: Fn(&T) -> bool,
+    {
+        self.watch.try_changed_and(&mut self.at_id, &mut f)
+    }
+
+    /// 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, const N: usize>(Rcv<'a, T, Watch<M, T, N>>);
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> Receiver<'a, M, T, N> {
+    /// Converts the `Receiver` into a [`DynReceiver`].
+    pub fn as_dyn(self) -> DynReceiver<'a, T> {
+        let rcv = DynReceiver(Rcv::new(self.0.watch, self.at_id));
+        core::mem::forget(self); // Ensures the destructor is not called
+        rcv
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> Into<DynReceiver<'a, T>> for Receiver<'a, M, T, N> {
+    fn into(self) -> DynReceiver<'a, T> {
+        self.as_dyn()
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> Deref for Receiver<'a, M, T, N> {
+    type Target = Rcv<'a, T, Watch<M, T, N>>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> DerefMut for Receiver<'a, M, T, N> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+/// A receiver which holds a **dynamic** reference to a `Watch` channel.
+///
+/// This is an alternative to [`Receiver`] with a simpler type definition, at the expense of
+/// some runtime performance due to dynamic dispatch.
+pub struct DynReceiver<'a, T: Clone>(Rcv<'a, T, dyn WatchBehavior<T> + 'a>);
+
+impl<'a, T: Clone> Deref for DynReceiver<'a, T> {
+    type Target = Rcv<'a, T, dyn WatchBehavior<T> + 'a>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<'a, T: Clone> DerefMut for DynReceiver<'a, T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+/// A receiver of a `Watch` channel that cannot `.await` values.
+#[derive(Debug)]
+pub struct AnonReceiver<'a, M: RawMutex, T: Clone, const N: usize>(AnonRcv<'a, T, Watch<M, T, N>>);
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> AnonReceiver<'a, M, T, N> {
+    /// Converts the `Receiver` into a [`DynReceiver`].
+    pub fn as_dyn(self) -> DynAnonReceiver<'a, T> {
+        let rcv = DynAnonReceiver(AnonRcv::new(self.0.watch, self.at_id));
+        core::mem::forget(self); // Ensures the destructor is not called
+        rcv
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> Into<DynAnonReceiver<'a, T>> for AnonReceiver<'a, M, T, N> {
+    fn into(self) -> DynAnonReceiver<'a, T> {
+        self.as_dyn()
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> Deref for AnonReceiver<'a, M, T, N> {
+    type Target = AnonRcv<'a, T, Watch<M, T, N>>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<'a, M: RawMutex, T: Clone, const N: usize> DerefMut for AnonReceiver<'a, M, T, N> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+/// A receiver that cannot `.await` value, which holds a **dynamic** reference to a `Watch` channel.
+///
+/// This is an alternative to [`AnonReceiver`] with a simpler type definition, at the expense of
+/// some runtime performance due to dynamic dispatch.
+pub struct DynAnonReceiver<'a, T: Clone>(AnonRcv<'a, T, dyn WatchBehavior<T> + 'a>);
+
+impl<'a, T: Clone> Deref for DynAnonReceiver<'a, T> {
+    type Target = AnonRcv<'a, T, dyn WatchBehavior<T> + 'a>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<'a, T: Clone> DerefMut for DynAnonReceiver<'a, T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use futures_executor::block_on;
+
+    use super::Watch;
+    use crate::blocking_mutex::raw::CriticalSectionRawMutex;
+
+    #[test]
+    fn multiple_sends() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 1> = Watch::new();
+
+            // Obtain receiver and sender
+            let mut rcv = WATCH.receiver().unwrap();
+            let snd = WATCH.sender();
+
+            // Not initialized
+            assert_eq!(rcv.try_changed(), None);
+
+            // Receive the new value
+            snd.send(10);
+            assert_eq!(rcv.changed().await, 10);
+
+            // Receive another value
+            snd.send(20);
+            assert_eq!(rcv.try_changed(), Some(20));
+
+            // No update
+            assert_eq!(rcv.try_changed(), None);
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn all_try_get() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 1> = Watch::new();
+
+            // Obtain receiver and sender
+            let mut rcv = WATCH.receiver().unwrap();
+            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));
+
+            assert_eq!(WATCH.try_get_and(|x| x > &5), Some(10));
+            assert_eq!(rcv.try_get_and(|x| x > &5), Some(10));
+            assert_eq!(snd.try_get_and(|x| x > &5), Some(10));
+
+            assert_eq!(WATCH.try_get_and(|x| x < &5), None);
+            assert_eq!(rcv.try_get_and(|x| x < &5), None);
+            assert_eq!(snd.try_get_and(|x| x < &5), None);
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn once_lock_like() {
+        let f = async {
+            static CONFIG0: u8 = 10;
+            static CONFIG1: u8 = 20;
+
+            static WATCH: Watch<CriticalSectionRawMutex, &'static u8, 1> = Watch::new();
+
+            // Obtain receiver and sender
+            let mut rcv = WATCH.receiver().unwrap();
+            let snd = WATCH.sender();
+
+            // Not initialized
+            assert_eq!(rcv.try_changed(), None);
+
+            // Receive the new value
+            snd.send(&CONFIG0);
+            let rcv0 = rcv.changed().await;
+            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));
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn sender_modify() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 1> = Watch::new();
+
+            // Obtain receiver and sender
+            let mut rcv = WATCH.receiver().unwrap();
+            let snd = WATCH.sender();
+
+            // Receive the new value
+            snd.send(10);
+            assert_eq!(rcv.try_changed(), Some(10));
+
+            // Modify the value inplace
+            snd.send_modify(|opt| {
+                if let Some(inner) = opt {
+                    *inner += 5;
+                }
+            });
+
+            // Get the modified value
+            assert_eq!(rcv.try_changed(), Some(15));
+            assert_eq!(rcv.try_changed(), None);
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn predicate_fn() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 1> = Watch::new();
+
+            // Obtain receiver and sender
+            let mut rcv = WATCH.receiver().unwrap();
+            let snd = WATCH.sender();
+
+            snd.send(15);
+            assert_eq!(rcv.try_get_and(|x| x > &5), Some(15));
+            assert_eq!(rcv.try_get_and(|x| x < &5), None);
+            assert!(rcv.try_changed().is_none());
+
+            snd.send(20);
+            assert_eq!(rcv.try_changed_and(|x| x > &5), Some(20));
+            assert_eq!(rcv.try_changed_and(|x| x > &5), None);
+
+            snd.send(25);
+            assert_eq!(rcv.try_changed_and(|x| x < &5), None);
+            assert_eq!(rcv.try_changed(), Some(25));
+
+            snd.send(30);
+            assert_eq!(rcv.changed_and(|x| x > &5).await, 30);
+            assert_eq!(rcv.get_and(|x| x > &5).await, 30);
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn receive_after_create() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 1> = Watch::new();
+
+            // Obtain sender and send value
+            let snd = WATCH.sender();
+            snd.send(10);
+
+            // Obtain receiver and receive value
+            let mut rcv = WATCH.receiver().unwrap();
+            assert_eq!(rcv.try_changed(), Some(10));
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn max_receivers_drop() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 2> = Watch::new();
+
+            // Try to create 3 receivers (only 2 can exist at once)
+            let rcv0 = WATCH.receiver();
+            let rcv1 = WATCH.receiver();
+            let rcv2 = WATCH.receiver();
+
+            // Ensure the first two are successful and the third is not
+            assert!(rcv0.is_some());
+            assert!(rcv1.is_some());
+            assert!(rcv2.is_none());
+
+            // Drop the first receiver
+            drop(rcv0);
+
+            // Create another receiver and ensure it is successful
+            let rcv3 = WATCH.receiver();
+            assert!(rcv3.is_some());
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn multiple_receivers() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 2> = Watch::new();
+
+            // Obtain receivers and sender
+            let mut rcv0 = WATCH.receiver().unwrap();
+            let mut rcv1 = WATCH.anon_receiver();
+            let snd = WATCH.sender();
+
+            // No update for both
+            assert_eq!(rcv0.try_changed(), None);
+            assert_eq!(rcv1.try_changed(), None);
+
+            // Send a new value
+            snd.send(0);
+
+            // Both receivers receive the new value
+            assert_eq!(rcv0.try_changed(), Some(0));
+            assert_eq!(rcv1.try_changed(), Some(0));
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn clone_senders() {
+        let f = async {
+            // Obtain different ways to send
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 1> = Watch::new();
+            let snd0 = WATCH.sender();
+            let snd1 = snd0.clone();
+
+            // Obtain Receiver
+            let mut rcv = WATCH.receiver().unwrap().as_dyn();
+
+            // 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));
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn use_dynamics() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 2> = Watch::new();
+
+            // Obtain receiver and sender
+            let mut anon_rcv = WATCH.dyn_anon_receiver();
+            let mut dyn_rcv = WATCH.dyn_receiver().unwrap();
+            let dyn_snd = WATCH.dyn_sender();
+
+            // Send a value
+            dyn_snd.send(10);
+
+            // Ensure the dynamic receiver receives the value
+            assert_eq!(anon_rcv.try_changed(), Some(10));
+            assert_eq!(dyn_rcv.try_changed(), Some(10));
+            assert_eq!(dyn_rcv.try_changed(), None);
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn convert_to_dyn() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 2> = Watch::new();
+
+            // Obtain receiver and sender
+            let anon_rcv = WATCH.anon_receiver();
+            let rcv = WATCH.receiver().unwrap();
+            let snd = WATCH.sender();
+
+            // Convert to dynamic
+            let mut dyn_anon_rcv = anon_rcv.as_dyn();
+            let mut dyn_rcv = rcv.as_dyn();
+            let dyn_snd = snd.as_dyn();
+
+            // Send a value
+            dyn_snd.send(10);
+
+            // Ensure the dynamic receiver receives the value
+            assert_eq!(dyn_anon_rcv.try_changed(), Some(10));
+            assert_eq!(dyn_rcv.try_changed(), Some(10));
+            assert_eq!(dyn_rcv.try_changed(), None);
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn dynamic_receiver_count() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 2> = Watch::new();
+
+            // Obtain receiver and sender
+            let rcv0 = WATCH.receiver();
+            let rcv1 = WATCH.receiver();
+            let rcv2 = WATCH.receiver();
+
+            // Ensure the first two are successful and the third is not
+            assert!(rcv0.is_some());
+            assert!(rcv1.is_some());
+            assert!(rcv2.is_none());
+
+            // Convert to dynamic
+            let dyn_rcv0 = rcv0.unwrap().as_dyn();
+
+            // Drop the (now dynamic) receiver
+            drop(dyn_rcv0);
+
+            // Create another receiver and ensure it is successful
+            let rcv3 = WATCH.receiver();
+            let rcv4 = WATCH.receiver();
+            assert!(rcv3.is_some());
+            assert!(rcv4.is_none());
+        };
+        block_on(f);
+    }
+
+    #[test]
+    fn contains_value() {
+        let f = async {
+            static WATCH: Watch<CriticalSectionRawMutex, u8, 2> = Watch::new();
+
+            // Obtain receiver and sender
+            let rcv = WATCH.receiver().unwrap();
+            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);
+        };
+        block_on(f);
+    }
+}
diff --git a/embassy-sync/src/zerocopy_channel.rs b/embassy-sync/src/zerocopy_channel.rs
index cfce9a571..c572592b8 100644
--- a/embassy-sync/src/zerocopy_channel.rs
+++ b/embassy-sync/src/zerocopy_channel.rs
@@ -15,12 +15,12 @@
 //! another message will result in an error being returned.
 
 use core::cell::RefCell;
-use core::future::poll_fn;
+use core::future::{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::blocking_mutex::raw::RawMutex;
 use crate::waitqueue::WakerRegistration;
 
 /// A bounded zero-copy channel for communicating between asynchronous tasks
@@ -34,8 +34,9 @@ use crate::waitqueue::WakerRegistration;
 ///
 /// The channel requires a buffer of recyclable elements.  Writing to the channel is done through
 /// an `&mut T`.
+#[derive(Debug)]
 pub struct Channel<'a, M: RawMutex, T> {
-    buf: *mut T,
+    buf: BufferPtr<T>,
     phantom: PhantomData<&'a mut T>,
     state: Mutex<M, RefCell<State>>,
 }
@@ -50,10 +51,10 @@ impl<'a, M: RawMutex, T> Channel<'a, M, T> {
         assert!(len != 0);
 
         Self {
-            buf: buf.as_mut_ptr(),
+            buf: BufferPtr(buf.as_mut_ptr()),
             phantom: PhantomData,
             state: Mutex::new(RefCell::new(State {
-                len,
+                capacity: len,
                 front: 0,
                 back: 0,
                 full: false,
@@ -70,9 +71,45 @@ impl<'a, M: RawMutex, T> Channel<'a, M, T> {
     pub fn split(&mut self) -> (Sender<'_, M, T>, Receiver<'_, M, T>) {
         (Sender { channel: self }, Receiver { channel: self })
     }
+
+    /// Clears all elements in the channel.
+    pub fn clear(&mut self) {
+        self.state.lock(|s| {
+            s.borrow_mut().clear();
+        });
+    }
+
+    /// Returns the number of elements currently in the channel.
+    pub fn len(&self) -> usize {
+        self.state.lock(|s| s.borrow().len())
+    }
+
+    /// Returns whether the channel is empty.
+    pub fn is_empty(&self) -> bool {
+        self.state.lock(|s| s.borrow().is_empty())
+    }
+
+    /// Returns whether the channel is full.
+    pub fn is_full(&self) -> bool {
+        self.state.lock(|s| s.borrow().is_full())
+    }
+}
+
+#[repr(transparent)]
+#[derive(Debug)]
+struct BufferPtr<T>(*mut T);
+
+impl<T> BufferPtr<T> {
+    unsafe fn add(&self, count: usize) -> *mut T {
+        self.0.add(count)
+    }
 }
 
+unsafe impl<T> Send for BufferPtr<T> {}
+unsafe impl<T> Sync for BufferPtr<T> {}
+
 /// Send-only access to a [`Channel`].
+#[derive(Debug)]
 pub struct Sender<'a, M: RawMutex, T> {
     channel: &'a Channel<'a, M, T>,
 }
@@ -109,12 +146,15 @@ impl<'a, M: RawMutex, T> Sender<'a, M, T> {
     }
 
     /// Asynchronously send a value over the channel.
-    pub async fn send(&mut self) -> &mut T {
-        let i = poll_fn(|cx| {
+    pub fn send(&mut self) -> impl Future<Output = &mut T> {
+        poll_fn(|cx| {
             self.channel.state.lock(|s| {
                 let s = &mut *s.borrow_mut();
                 match s.push_index() {
-                    Some(i) => Poll::Ready(i),
+                    Some(i) => {
+                        let r = unsafe { &mut *self.channel.buf.add(i) };
+                        Poll::Ready(r)
+                    }
                     None => {
                         s.receive_waker.register(cx.waker());
                         Poll::Pending
@@ -122,23 +162,44 @@ impl<'a, M: RawMutex, T> Sender<'a, M, T> {
                 }
             })
         })
-        .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())
     }
+
+    /// Clears all elements in the channel.
+    pub fn clear(&mut self) {
+        self.channel.state.lock(|s| {
+            s.borrow_mut().clear();
+        });
+    }
+
+    /// Returns the number of elements currently in the channel.
+    pub fn len(&self) -> usize {
+        self.channel.state.lock(|s| s.borrow().len())
+    }
+
+    /// Returns whether the channel is empty.
+    pub fn is_empty(&self) -> bool {
+        self.channel.state.lock(|s| s.borrow().is_empty())
+    }
+
+    /// Returns whether the channel is full.
+    pub fn is_full(&self) -> bool {
+        self.channel.state.lock(|s| s.borrow().is_full())
+    }
 }
 
 /// Receive-only access to a [`Channel`].
+#[derive(Debug)]
 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.
+    /// Creates one further [`Receiver`] over the same channel.
     pub fn borrow(&mut self) -> Receiver<'_, M, T> {
         Receiver { channel: self.channel }
     }
@@ -169,12 +230,15 @@ impl<'a, M: RawMutex, T> Receiver<'a, M, T> {
     }
 
     /// Asynchronously receive a value over the channel.
-    pub async fn receive(&mut self) -> &mut T {
-        let i = poll_fn(|cx| {
+    pub fn receive(&mut self) -> impl Future<Output = &mut T> {
+        poll_fn(|cx| {
             self.channel.state.lock(|s| {
                 let s = &mut *s.borrow_mut();
                 match s.pop_index() {
-                    Some(i) => Poll::Ready(i),
+                    Some(i) => {
+                        let r = unsafe { &mut *self.channel.buf.add(i) };
+                        Poll::Ready(r)
+                    }
                     None => {
                         s.send_waker.register(cx.waker());
                         Poll::Pending
@@ -182,18 +246,40 @@ impl<'a, M: RawMutex, T> Receiver<'a, M, T> {
                 }
             })
         })
-        .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())
     }
+
+    /// Clears all elements in the channel.
+    pub fn clear(&mut self) {
+        self.channel.state.lock(|s| {
+            s.borrow_mut().clear();
+        });
+    }
+
+    /// Returns the number of elements currently in the channel.
+    pub fn len(&self) -> usize {
+        self.channel.state.lock(|s| s.borrow().len())
+    }
+
+    /// Returns whether the channel is empty.
+    pub fn is_empty(&self) -> bool {
+        self.channel.state.lock(|s| s.borrow().is_empty())
+    }
+
+    /// Returns whether the channel is full.
+    pub fn is_full(&self) -> bool {
+        self.channel.state.lock(|s| s.borrow().is_full())
+    }
 }
 
+#[derive(Debug)]
 struct State {
-    len: usize,
+    /// Maximum number of elements the channel can hold.
+    capacity: usize,
 
     /// Front index. Always 0..=(N-1)
     front: usize,
@@ -210,10 +296,27 @@ struct State {
 
 impl State {
     fn increment(&self, i: usize) -> usize {
-        if i + 1 == self.len {
-            0
+        if i + 1 == self.capacity { 0 } else { i + 1 }
+    }
+
+    fn clear(&mut self) {
+        if self.full {
+            self.receive_waker.wake();
+        }
+        self.front = 0;
+        self.back = 0;
+        self.full = false;
+    }
+
+    fn len(&self) -> usize {
+        if !self.full {
+            if self.back >= self.front {
+                self.back - self.front
+            } else {
+                self.capacity + self.back - self.front
+            }
         } else {
-            i + 1
+            self.capacity
         }
     }