14 files changed, 441 insertions, 211 deletions

diff --git a/embassy-sync/src/channel.rs b/embassy-sync/src/channel.rs
index 76f42d0e7..77352874d 100644
--- a/embassy-sync/src/channel.rs
+++ b/embassy-sync/src/channel.rs
@@ -181,6 +181,7 @@ where
 }
 
 /// Future returned by [`Channel::recv`] and  [`Receiver::recv`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct RecvFuture<'ch, M, T, const N: usize>
 where
     M: RawMutex,
@@ -203,6 +204,7 @@ where
 }
 
 /// Future returned by [`DynamicReceiver::recv`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct DynamicRecvFuture<'ch, T> {
     channel: &'ch dyn DynamicChannel<T>,
 }
@@ -219,6 +221,7 @@ impl<'ch, T> Future for DynamicRecvFuture<'ch, T> {
 }
 
 /// Future returned by [`Channel::send`] and  [`Sender::send`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct SendFuture<'ch, M, T, const N: usize>
 where
     M: RawMutex,
@@ -250,6 +253,7 @@ where
 impl<'ch, M, T, const N: usize> Unpin for SendFuture<'ch, M, T, N> where M: RawMutex {}
 
 /// Future returned by [`DynamicSender::send`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct DynamicSendFuture<'ch, T> {
     channel: &'ch dyn DynamicChannel<T>,
     message: Option<T>,
diff --git a/embassy-sync/src/fmt.rs b/embassy-sync/src/fmt.rs
index f8bb0a035..066970813 100644
--- a/embassy-sync/src/fmt.rs
+++ b/embassy-sync/src/fmt.rs
@@ -195,9 +195,6 @@ macro_rules! unwrap {
     }
 }
 
-#[cfg(feature = "defmt-timestamp-uptime")]
-defmt::timestamp! {"{=u64:us}", crate::time::Instant::now().as_micros() }
-
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
 pub struct NoneError;
 
diff --git a/embassy-sync/src/lib.rs b/embassy-sync/src/lib.rs
index 25150e8aa..53d95d081 100644
--- a/embassy-sync/src/lib.rs
+++ b/embassy-sync/src/lib.rs
@@ -1,5 +1,5 @@
 #![cfg_attr(not(any(feature = "std", feature = "wasm")), no_std)]
-#![cfg_attr(feature = "nightly", feature(generic_associated_types, type_alias_impl_trait))]
+#![cfg_attr(feature = "nightly", feature(async_fn_in_trait, impl_trait_projections))]
 #![allow(clippy::new_without_default)]
 #![doc = include_str!("../README.md")]
 #![warn(missing_docs)]
diff --git a/embassy-sync/src/mutex.rs b/embassy-sync/src/mutex.rs
index 75a6e8dd3..fcf056d36 100644
--- a/embassy-sync/src/mutex.rs
+++ b/embassy-sync/src/mutex.rs
@@ -2,11 +2,10 @@
 //!
 //! This module provides a mutex that can be used to synchronize data between asynchronous tasks.
 use core::cell::{RefCell, UnsafeCell};
+use core::future::poll_fn;
 use core::ops::{Deref, DerefMut};
 use core::task::Poll;
 
-use futures_util::future::poll_fn;
-
 use crate::blocking_mutex::raw::RawMutex;
 use crate::blocking_mutex::Mutex as BlockingMutex;
 use crate::waitqueue::WakerRegistration;
@@ -111,6 +110,22 @@ where
 
         Ok(MutexGuard { mutex: self })
     }
+
+    /// Consumes this mutex, returning the underlying data.
+    pub fn into_inner(self) -> T
+    where
+        T: Sized,
+    {
+        self.inner.into_inner()
+    }
+
+    /// Returns a mutable reference to the underlying data.
+    ///
+    /// Since this call borrows the Mutex mutably, no actual locking needs to
+    /// take place -- the mutable borrow statically guarantees no locks exist.
+    pub fn get_mut(&mut self) -> &mut T {
+        self.inner.get_mut()
+    }
 }
 
 /// Async mutex guard.
diff --git a/embassy-sync/src/pipe.rs b/embassy-sync/src/pipe.rs
index 7d64b648e..13bf4ef01 100644
--- a/embassy-sync/src/pipe.rs
+++ b/embassy-sync/src/pipe.rs
@@ -32,22 +32,23 @@ impl<'p, M, const N: usize> Writer<'p, M, N>
 where
     M: RawMutex,
 {
-    /// Writes a value.
+    /// Write some bytes to the pipe.
     ///
     /// See [`Pipe::write()`]
     pub fn write<'a>(&'a self, buf: &'a [u8]) -> WriteFuture<'a, M, N> {
         self.pipe.write(buf)
     }
 
-    /// Attempt to immediately write a message.
+    /// Attempt to immediately write some bytes to the pipe.
     ///
-    /// See [`Pipe::write()`]
+    /// See [`Pipe::try_write()`]
     pub fn try_write(&self, buf: &[u8]) -> Result<usize, TryWriteError> {
         self.pipe.try_write(buf)
     }
 }
 
 /// Future returned by [`Pipe::write`] and  [`Writer::write`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct WriteFuture<'p, M, const N: usize>
 where
     M: RawMutex,
@@ -94,22 +95,23 @@ impl<'p, M, const N: usize> Reader<'p, M, N>
 where
     M: RawMutex,
 {
-    /// Reads a value.
+    /// Read some bytes from the pipe.
     ///
     /// See [`Pipe::read()`]
     pub fn read<'a>(&'a self, buf: &'a mut [u8]) -> ReadFuture<'a, M, N> {
         self.pipe.read(buf)
     }
 
-    /// Attempt to immediately read a message.
+    /// Attempt to immediately read some bytes from the pipe.
     ///
-    /// See [`Pipe::read()`]
+    /// See [`Pipe::try_read()`]
     pub fn try_read(&self, buf: &mut [u8]) -> Result<usize, TryReadError> {
         self.pipe.try_read(buf)
     }
 }
 
 /// Future returned by [`Pipe::read`] and  [`Reader::read`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct ReadFuture<'p, M, const N: usize>
 where
     M: RawMutex,
@@ -219,12 +221,11 @@ impl<const N: usize> PipeState<N> {
     }
 }
 
-/// A bounded pipe for communicating between asynchronous tasks
+/// A bounded byte-oriented pipe for communicating between asynchronous tasks
 /// with backpressure.
 ///
-/// The pipe will buffer up to the provided number of messages.  Once the
-/// buffer is full, attempts to `write` new messages will wait until a message is
-/// read from the pipe.
+/// The pipe will buffer up to the provided number of bytes. Once the
+/// buffer is full, attempts to `write` new bytes will wait until buffer space is freed up.
 ///
 /// All data written will become available in the same order as it was written.
 pub struct Pipe<M, const N: usize>
@@ -275,40 +276,66 @@ where
         Reader { pipe: self }
     }
 
-    /// Write a value, waiting until there is capacity.
+    /// Write some bytes to the pipe.
+    ///
+    /// This method writes a nonzero amount of bytes from `buf` into the pipe, and
+    /// returns the amount of bytes written.
     ///
-    /// Writeing completes when the value has been pushed to the pipe's queue.
-    /// This doesn't mean the value has been read yet.
+    /// If it is not possible to write a nonzero amount of bytes because the pipe's buffer is full,
+    /// this method will wait until it isn't. See [`try_write`](Self::try_write) for a variant that
+    /// returns an error instead of waiting.
+    ///
+    /// It is not guaranteed that all bytes in the buffer are written, even if there's enough
+    /// free space in the pipe buffer for all. In other words, it is possible for `write` to return
+    /// without writing all of `buf` (returning a number less than `buf.len()`) and still leave
+    /// free space in the pipe buffer. You should always `write` in a loop, or use helpers like
+    /// `write_all` from the `embedded-io` crate.
     pub fn write<'a>(&'a self, buf: &'a [u8]) -> WriteFuture<'a, M, N> {
         WriteFuture { pipe: self, buf }
     }
 
-    /// Attempt to immediately write a message.
-    ///
-    /// This method differs from [`write`](Pipe::write) by returning immediately if the pipe's
-    /// buffer is full, instead of waiting.
+    /// Write all bytes to the pipe.
     ///
-    /// # Errors
+    /// This method writes all bytes from `buf` into the pipe
+    pub async fn write_all(&self, mut buf: &[u8]) {
+        while !buf.is_empty() {
+            let n = self.write(buf).await;
+            buf = &buf[n..];
+        }
+    }
+
+    /// Attempt to immediately write some bytes to the pipe.
     ///
-    /// If the pipe capacity has been reached, i.e., the pipe has `n`
-    /// buffered values where `n` is the argument passed to [`Pipe`], then an
-    /// error is returned.
+    /// This method will either write a nonzero amount of bytes to the pipe immediately,
+    /// or return an error if the pipe is empty. See [`write`](Self::write) for a variant
+    /// that waits instead of returning an error.
     pub fn try_write(&self, buf: &[u8]) -> Result<usize, TryWriteError> {
         self.lock(|c| c.try_write(buf))
     }
 
-    /// Receive the next value.
+    /// Read some bytes from the pipe.
+    ///
+    /// This method reads a nonzero amount of bytes from the pipe into `buf` and
+    /// returns the amount of bytes read.
+    ///
+    /// If it is not possible to read a nonzero amount of bytes because the pipe's buffer is empty,
+    /// this method will wait until it isn't. See [`try_read`](Self::try_read) for a variant that
+    /// returns an error instead of waiting.
     ///
-    /// If there are no messages in the pipe's buffer, this method will
-    /// wait until a message is written.
+    /// It is not guaranteed that all bytes in the buffer are read, even if there's enough
+    /// space in `buf` for all. In other words, it is possible for `read` to return
+    /// without filling `buf` (returning a number less than `buf.len()`) and still leave bytes
+    /// in the pipe buffer. You should always `read` in a loop, or use helpers like
+    /// `read_exact` from the `embedded-io` crate.
     pub fn read<'a>(&'a self, buf: &'a mut [u8]) -> ReadFuture<'a, M, N> {
         ReadFuture { pipe: self, buf }
     }
 
-    /// Attempt to immediately read a message.
+    /// Attempt to immediately read some bytes from the pipe.
     ///
-    /// This method will either read a message from the pipe immediately or return an error
-    /// if the pipe is empty.
+    /// This method will either read a nonzero amount of bytes from the pipe immediately,
+    /// or return an error if the pipe is empty. See [`read`](Self::read) for a variant
+    /// that waits instead of returning an error.
     pub fn try_read(&self, buf: &mut [u8]) -> Result<usize, TryReadError> {
         self.lock(|c| c.try_read(buf))
     }
@@ -352,8 +379,6 @@ where
 mod io_impls {
     use core::convert::Infallible;
 
-    use futures_util::FutureExt;
-
     use super::*;
 
     impl<M: RawMutex, const N: usize> embedded_io::Io for Pipe<M, N> {
@@ -361,30 +386,18 @@ mod io_impls {
     }
 
     impl<M: RawMutex, const N: usize> embedded_io::asynch::Read for Pipe<M, N> {
-        type ReadFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
-        where
-            Self: 'a;
-
-        fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Self::ReadFuture<'a> {
-            Pipe::read(self, buf).map(Ok)
+        async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Self::Error> {
+            Ok(Pipe::read(self, buf).await)
         }
     }
 
     impl<M: RawMutex, const N: usize> embedded_io::asynch::Write for Pipe<M, N> {
-        type WriteFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
-        where
-            Self: 'a;
-
-        fn write<'a>(&'a mut self, buf: &'a [u8]) -> Self::WriteFuture<'a> {
-            Pipe::write(self, buf).map(Ok)
+        async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
+            Ok(Pipe::write(self, buf).await)
         }
 
-        type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>>
-        where
-            Self: 'a;
-
-        fn flush<'a>(&'a mut self) -> Self::FlushFuture<'a> {
-            futures_util::future::ready(Ok(()))
+        async fn flush(&mut self) -> Result<(), Self::Error> {
+            Ok(())
         }
     }
 
@@ -393,30 +406,18 @@ mod io_impls {
     }
 
     impl<M: RawMutex, const N: usize> embedded_io::asynch::Read for &Pipe<M, N> {
-        type ReadFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
-        where
-            Self: 'a;
-
-        fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Self::ReadFuture<'a> {
-            Pipe::read(self, buf).map(Ok)
+        async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Self::Error> {
+            Ok(Pipe::read(self, buf).await)
         }
     }
 
     impl<M: RawMutex, const N: usize> embedded_io::asynch::Write for &Pipe<M, N> {
-        type WriteFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
-        where
-            Self: 'a;
-
-        fn write<'a>(&'a mut self, buf: &'a [u8]) -> Self::WriteFuture<'a> {
-            Pipe::write(self, buf).map(Ok)
+        async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
+            Ok(Pipe::write(self, buf).await)
         }
 
-        type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>>
-        where
-            Self: 'a;
-
-        fn flush<'a>(&'a mut self) -> Self::FlushFuture<'a> {
-            futures_util::future::ready(Ok(()))
+        async fn flush(&mut self) -> Result<(), Self::Error> {
+            Ok(())
         }
     }
 
@@ -425,12 +426,8 @@ mod io_impls {
     }
 
     impl<M: RawMutex, const N: usize> embedded_io::asynch::Read for Reader<'_, M, N> {
-        type ReadFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
-        where
-            Self: 'a;
-
-        fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Self::ReadFuture<'a> {
-            Reader::read(self, buf).map(Ok)
+        async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Self::Error> {
+            Ok(Reader::read(self, buf).await)
         }
     }
 
@@ -439,20 +436,12 @@ mod io_impls {
     }
 
     impl<M: RawMutex, const N: usize> embedded_io::asynch::Write for Writer<'_, M, N> {
-        type WriteFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
-        where
-            Self: 'a;
-
-        fn write<'a>(&'a mut self, buf: &'a [u8]) -> Self::WriteFuture<'a> {
-            Writer::write(self, buf).map(Ok)
+        async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
+            Ok(Writer::write(self, buf).await)
         }
 
-        type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>>
-        where
-            Self: 'a;
-
-        fn flush<'a>(&'a mut self) -> Self::FlushFuture<'a> {
-            futures_util::future::ready(Ok(()))
+        async fn flush(&mut self) -> Result<(), Self::Error> {
+            Ok(())
         }
     }
 }
diff --git a/embassy-sync/src/pubsub/mod.rs b/embassy-sync/src/pubsub/mod.rs
index 62a9e4763..6afd54af5 100644
--- a/embassy-sync/src/pubsub/mod.rs
+++ b/embassy-sync/src/pubsub/mod.rs
@@ -4,7 +4,7 @@
 
 use core::cell::RefCell;
 use core::fmt::Debug;
-use core::task::{Context, Poll, Waker};
+use core::task::{Context, Poll};
 
 use heapless::Deque;
 
@@ -179,7 +179,7 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
                 // No, so we need to reregister our waker and sleep again
                 None => {
                     if let Some(cx) = cx {
-                        s.register_subscriber_waker(cx.waker());
+                        s.subscriber_wakers.register(cx.waker());
                     }
                     Poll::Pending
                 }
@@ -192,6 +192,10 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
         })
     }
 
+    fn available(&self, next_message_id: u64) -> u64 {
+        self.inner.lock(|s| s.borrow().next_message_id - next_message_id)
+    }
+
     fn publish_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), T> {
         self.inner.lock(|s| {
             let mut s = s.borrow_mut();
@@ -202,7 +206,7 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
                 // The queue is full, so we need to reregister our waker and go to sleep
                 Err(message) => {
                     if let Some(cx) = cx {
-                        s.register_publisher_waker(cx.waker());
+                        s.publisher_wakers.register(cx.waker());
                     }
                     Err(message)
                 }
@@ -217,6 +221,13 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
         })
     }
 
+    fn space(&self) -> usize {
+        self.inner.lock(|s| {
+            let s = s.borrow();
+            s.queue.capacity() - s.queue.len()
+        })
+    }
+
     fn unregister_subscriber(&self, subscriber_next_message_id: u64) {
         self.inner.lock(|s| {
             let mut s = s.borrow_mut();
@@ -311,44 +322,19 @@ impl<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> PubSubSta
 
         // We're reading this item, so decrement the counter
         queue_item.1 -= 1;
-        let message = queue_item.0.clone();
 
-        if current_message_index == 0 && queue_item.1 == 0 {
-            self.queue.pop_front();
+        let message = if current_message_index == 0 && queue_item.1 == 0 {
+            let (message, _) = self.queue.pop_front().unwrap();
             self.publisher_wakers.wake();
-        }
+            // Return pop'd message without clone
+            message
+        } else {
+            queue_item.0.clone()
+        };
 
         Some(WaitResult::Message(message))
     }
 
-    fn register_subscriber_waker(&mut self, waker: &Waker) {
-        match self.subscriber_wakers.register(waker) {
-            Ok(()) => {}
-            Err(_) => {
-                // All waker slots were full. This can only happen when there was a subscriber that now has dropped.
-                // We need to throw it away. It's a bit inefficient, but we can wake everything.
-                // Any future that is still active will simply reregister.
-                // This won't happen a lot, so it's ok.
-                self.subscriber_wakers.wake();
-                self.subscriber_wakers.register(waker).unwrap();
-            }
-        }
-    }
-
-    fn register_publisher_waker(&mut self, waker: &Waker) {
-        match self.publisher_wakers.register(waker) {
-            Ok(()) => {}
-            Err(_) => {
-                // All waker slots were full. This can only happen when there was a publisher that now has dropped.
-                // We need to throw it away. It's a bit inefficient, but we can wake everything.
-                // Any future that is still active will simply reregister.
-                // This won't happen a lot, so it's ok.
-                self.publisher_wakers.wake();
-                self.publisher_wakers.register(waker).unwrap();
-            }
-        }
-    }
-
     fn unregister_subscriber(&mut self, subscriber_next_message_id: u64) {
         self.subscriber_count -= 1;
 
@@ -360,6 +346,20 @@ impl<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> PubSubSta
                 .iter_mut()
                 .skip(current_message_index)
                 .for_each(|(_, counter)| *counter -= 1);
+
+            let mut wake_publishers = false;
+            while let Some((_, count)) = self.queue.front() {
+                if *count == 0 {
+                    self.queue.pop_front().unwrap();
+                    wake_publishers = true;
+                } else {
+                    break;
+                }
+            }
+
+            if wake_publishers {
+                self.publisher_wakers.wake();
+            }
         }
     }
 
@@ -388,6 +388,10 @@ pub trait PubSubBehavior<T> {
     /// If the message is not yet present and a context is given, then its waker is registered in the subsriber wakers.
     fn get_message_with_context(&self, next_message_id: &mut u64, cx: Option<&mut Context<'_>>) -> Poll<WaitResult<T>>;
 
+    /// Get the amount of messages that are between the given the next_message_id and the most recent message.
+    /// This is not necessarily the amount of messages a subscriber can still received as it may have lagged.
+    fn available(&self, next_message_id: u64) -> u64;
+
     /// Try to publish a message to the queue.
     ///
     /// If the queue is full and a context is given, then its waker is registered in the publisher wakers.
@@ -396,6 +400,9 @@ pub trait PubSubBehavior<T> {
     /// Publish a message immediately
     fn publish_immediate(&self, message: T);
 
+    /// The amount of messages that can still be published without having to wait or without having to lag the subscribers
+    fn space(&self) -> usize;
+
     /// Let the channel know that a subscriber has dropped
     fn unregister_subscriber(&self, subscriber_next_message_id: u64);
 
@@ -539,4 +546,113 @@ mod tests {
 
         drop(sub0);
     }
+
+    #[futures_test::test]
+    async fn correct_available() {
+        let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
+
+        let sub0 = channel.subscriber().unwrap();
+        let mut sub1 = channel.subscriber().unwrap();
+        let pub0 = channel.publisher().unwrap();
+
+        assert_eq!(sub0.available(), 0);
+        assert_eq!(sub1.available(), 0);
+
+        pub0.publish(42).await;
+
+        assert_eq!(sub0.available(), 1);
+        assert_eq!(sub1.available(), 1);
+
+        sub1.next_message().await;
+
+        assert_eq!(sub1.available(), 0);
+
+        pub0.publish(42).await;
+
+        assert_eq!(sub0.available(), 2);
+        assert_eq!(sub1.available(), 1);
+    }
+
+    #[futures_test::test]
+    async fn correct_space() {
+        let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
+
+        let mut sub0 = channel.subscriber().unwrap();
+        let mut sub1 = channel.subscriber().unwrap();
+        let pub0 = channel.publisher().unwrap();
+
+        assert_eq!(pub0.space(), 4);
+
+        pub0.publish(42).await;
+
+        assert_eq!(pub0.space(), 3);
+
+        pub0.publish(42).await;
+
+        assert_eq!(pub0.space(), 2);
+
+        sub0.next_message().await;
+        sub0.next_message().await;
+
+        assert_eq!(pub0.space(), 2);
+
+        sub1.next_message().await;
+        assert_eq!(pub0.space(), 3);
+        sub1.next_message().await;
+        assert_eq!(pub0.space(), 4);
+    }
+
+    #[futures_test::test]
+    async fn empty_channel_when_last_subscriber_is_dropped() {
+        let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();
+
+        let pub0 = channel.publisher().unwrap();
+        let mut sub0 = channel.subscriber().unwrap();
+        let mut sub1 = channel.subscriber().unwrap();
+
+        assert_eq!(4, pub0.space());
+
+        pub0.publish(1).await;
+        pub0.publish(2).await;
+
+        assert_eq!(2, channel.space());
+
+        assert_eq!(1, sub0.try_next_message_pure().unwrap());
+        assert_eq!(2, sub0.try_next_message_pure().unwrap());
+
+        assert_eq!(2, channel.space());
+
+        drop(sub0);
+
+        assert_eq!(2, channel.space());
+
+        assert_eq!(1, sub1.try_next_message_pure().unwrap());
+
+        assert_eq!(3, channel.space());
+
+        drop(sub1);
+
+        assert_eq!(4, channel.space());
+    }
+
+    struct CloneCallCounter(usize);
+
+    impl Clone for CloneCallCounter {
+        fn clone(&self) -> Self {
+            Self(self.0 + 1)
+        }
+    }
+
+    #[futures_test::test]
+    async fn skip_clone_for_last_message() {
+        let channel = PubSubChannel::<NoopRawMutex, CloneCallCounter, 1, 2, 1>::new();
+        let pub0 = channel.publisher().unwrap();
+        let mut sub0 = channel.subscriber().unwrap();
+        let mut sub1 = channel.subscriber().unwrap();
+
+        pub0.publish(CloneCallCounter(0)).await;
+
+        assert_eq!(1, sub0.try_next_message_pure().unwrap().0);
+        assert_eq!(0, sub1.try_next_message_pure().unwrap().0);
+    }
 }
diff --git a/embassy-sync/src/pubsub/publisher.rs b/embassy-sync/src/pubsub/publisher.rs
index 705797f60..e1edc9eb9 100644
--- a/embassy-sync/src/pubsub/publisher.rs
+++ b/embassy-sync/src/pubsub/publisher.rs
@@ -42,6 +42,14 @@ impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Pub<'a, PSB, T> {
     pub fn try_publish(&self, message: T) -> Result<(), T> {
         self.channel.publish_with_context(message, None)
     }
+
+    /// The amount of messages that can still be published without having to wait or without having to lag the subscribers
+    ///
+    /// *Note: In the time between checking this and a publish action, other publishers may have had time to publish something.
+    /// So checking doesn't give any guarantees.*
+    pub fn space(&self) -> usize {
+        self.channel.space()
+    }
 }
 
 impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Drop for Pub<'a, PSB, T> {
@@ -115,6 +123,14 @@ impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> ImmediatePub<'a, PSB, T> {
     pub fn try_publish(&self, message: T) -> Result<(), T> {
         self.channel.publish_with_context(message, None)
     }
+
+    /// The amount of messages that can still be published without having to wait or without having to lag the subscribers
+    ///
+    /// *Note: In the time between checking this and a publish action, other publishers may have had time to publish something.
+    /// So checking doesn't give any guarantees.*
+    pub fn space(&self) -> usize {
+        self.channel.space()
+    }
 }
 
 /// An immediate publisher that holds a dynamic reference to the channel
@@ -158,6 +174,7 @@ impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS:
 }
 
 /// Future for the publisher wait action
+#[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct PublisherWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
     /// The message we need to publish
     message: Option<T>,
diff --git a/embassy-sync/src/pubsub/subscriber.rs b/embassy-sync/src/pubsub/subscriber.rs
index b9a2cbe18..f420a75f0 100644
--- a/embassy-sync/src/pubsub/subscriber.rs
+++ b/embassy-sync/src/pubsub/subscriber.rs
@@ -64,6 +64,11 @@ impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Sub<'a, PSB, T> {
             }
         }
     }
+
+    /// The amount of messages this subscriber hasn't received yet
+    pub fn available(&self) -> u64 {
+        self.channel.available(self.next_message_id)
+    }
 }
 
 impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Drop for Sub<'a, PSB, T> {
@@ -135,6 +140,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"]
 pub struct SubscriberWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
     subscriber: &'s mut Sub<'a, PSB, T>,
 }
diff --git a/embassy-sync/src/signal.rs b/embassy-sync/src/signal.rs
index f6ebeb9b9..bea67d8be 100644
--- a/embassy-sync/src/signal.rs
+++ b/embassy-sync/src/signal.rs
@@ -1,9 +1,11 @@
 //! A synchronization primitive for passing the latest value to a task.
-use core::cell::UnsafeCell;
-use core::future::Future;
-use core::mem;
+use core::cell::Cell;
+use core::future::{poll_fn, Future};
 use core::task::{Context, Poll, Waker};
 
+use crate::blocking_mutex::raw::RawMutex;
+use crate::blocking_mutex::Mutex;
+
 /// Single-slot signaling primitive.
 ///
 /// This is similar to a [`Channel`](crate::channel::Channel) with a buffer size of 1, except
@@ -20,16 +22,20 @@ use core::task::{Context, Poll, Waker};
 ///
 /// ```
 /// use embassy_sync::signal::Signal;
+/// use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
 ///
 /// enum SomeCommand {
 ///   On,
 ///   Off,
 /// }
 ///
-/// static SOME_SIGNAL: Signal<SomeCommand> = Signal::new();
+/// static SOME_SIGNAL: Signal<CriticalSectionRawMutex, SomeCommand> = Signal::new();
 /// ```
-pub struct Signal<T> {
-    state: UnsafeCell<State<T>>,
+pub struct Signal<M, T>
+where
+    M: RawMutex,
+{
+    state: Mutex<M, Cell<State<T>>>,
 }
 
 enum State<T> {
@@ -38,24 +44,36 @@ enum State<T> {
     Signaled(T),
 }
 
-unsafe impl<T: Send> Send for Signal<T> {}
-unsafe impl<T: Send> Sync for Signal<T> {}
-
-impl<T> Signal<T> {
+impl<M, T> Signal<M, T>
+where
+    M: RawMutex,
+{
     /// Create a new `Signal`.
     pub const fn new() -> Self {
         Self {
-            state: UnsafeCell::new(State::None),
+            state: Mutex::new(Cell::new(State::None)),
         }
     }
 }
 
-impl<T: Send> Signal<T> {
+impl<M, T> Default for Signal<M, T>
+where
+    M: RawMutex,
+{
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<M, T: Send> Signal<M, T>
+where
+    M: RawMutex,
+{
     /// Mark this Signal as signaled.
     pub fn signal(&self, val: T) {
-        critical_section::with(|_| unsafe {
-            let state = &mut *self.state.get();
-            if let State::Waiting(waker) = mem::replace(state, State::Signaled(val)) {
+        self.state.lock(|cell| {
+            let state = cell.replace(State::Signaled(val));
+            if let State::Waiting(waker) = state {
                 waker.wake();
             }
         })
@@ -63,38 +81,46 @@ impl<T: Send> Signal<T> {
 
     /// Remove the queued value in this `Signal`, if any.
     pub fn reset(&self) {
-        critical_section::with(|_| unsafe {
-            let state = &mut *self.state.get();
-            *state = State::None
-        })
+        self.state.lock(|cell| cell.set(State::None));
     }
 
-    /// Manually poll the Signal future.
-    pub fn poll_wait(&self, cx: &mut Context<'_>) -> Poll<T> {
-        critical_section::with(|_| unsafe {
-            let state = &mut *self.state.get();
+    fn poll_wait(&self, cx: &mut Context<'_>) -> Poll<T> {
+        self.state.lock(|cell| {
+            let state = cell.replace(State::None);
             match state {
                 State::None => {
-                    *state = State::Waiting(cx.waker().clone());
+                    cell.set(State::Waiting(cx.waker().clone()));
+                    Poll::Pending
+                }
+                State::Waiting(w) if w.will_wake(cx.waker()) => {
+                    cell.set(State::Waiting(w));
+                    Poll::Pending
+                }
+                State::Waiting(w) => {
+                    cell.set(State::Waiting(cx.waker().clone()));
+                    w.wake();
                     Poll::Pending
                 }
-                State::Waiting(w) if w.will_wake(cx.waker()) => Poll::Pending,
-                State::Waiting(_) => panic!("waker overflow"),
-                State::Signaled(_) => match mem::replace(state, State::None) {
-                    State::Signaled(res) => Poll::Ready(res),
-                    _ => unreachable!(),
-                },
+                State::Signaled(res) => Poll::Ready(res),
             }
         })
     }
 
     /// Future that completes when this Signal has been signaled.
     pub fn wait(&self) -> impl Future<Output = T> + '_ {
-        futures_util::future::poll_fn(move |cx| self.poll_wait(cx))
+        poll_fn(move |cx| self.poll_wait(cx))
     }
 
     /// non-blocking method to check whether this signal has been signaled.
     pub fn signaled(&self) -> bool {
-        critical_section::with(|_| matches!(unsafe { &*self.state.get() }, State::Signaled(_)))
+        self.state.lock(|cell| {
+            let state = cell.replace(State::None);
+
+            let res = matches!(state, State::Signaled(_));
+
+            cell.set(state);
+
+            res
+        })
     }
 }
diff --git a/embassy-sync/src/waitqueue/atomic_waker.rs b/embassy-sync/src/waitqueue/atomic_waker.rs
new file mode 100644
index 000000000..63fe04a6e
--- /dev/null
+++ b/embassy-sync/src/waitqueue/atomic_waker.rs
@@ -0,0 +1,41 @@
+use core::cell::Cell;
+use core::task::Waker;
+
+use crate::blocking_mutex::raw::CriticalSectionRawMutex;
+use crate::blocking_mutex::Mutex;
+
+/// Utility struct to register and wake a waker.
+pub struct AtomicWaker {
+    waker: Mutex<CriticalSectionRawMutex, Cell<Option<Waker>>>,
+}
+
+impl AtomicWaker {
+    /// Create a new `AtomicWaker`.
+    pub const fn new() -> Self {
+        Self {
+            waker: Mutex::const_new(CriticalSectionRawMutex::new(), Cell::new(None)),
+        }
+    }
+
+    /// Register a waker. Overwrites the previous waker, if any.
+    pub fn register(&self, w: &Waker) {
+        critical_section::with(|cs| {
+            let cell = self.waker.borrow(cs);
+            cell.set(match cell.replace(None) {
+                Some(w2) if (w2.will_wake(w)) => Some(w2),
+                _ => Some(w.clone()),
+            })
+        })
+    }
+
+    /// Wake the registered waker, if any.
+    pub fn wake(&self) {
+        critical_section::with(|cs| {
+            let cell = self.waker.borrow(cs);
+            if let Some(w) = cell.replace(None) {
+                w.wake_by_ref();
+                cell.set(Some(w));
+            }
+        })
+    }
+}
diff --git a/embassy-sync/src/waitqueue/atomic_waker_turbo.rs b/embassy-sync/src/waitqueue/atomic_waker_turbo.rs
new file mode 100644
index 000000000..5c6a96ec8
--- /dev/null
+++ b/embassy-sync/src/waitqueue/atomic_waker_turbo.rs
@@ -0,0 +1,30 @@
+use core::ptr;
+use core::ptr::NonNull;
+use core::sync::atomic::{AtomicPtr, Ordering};
+use core::task::Waker;
+
+/// Utility struct to register and wake a waker.
+pub struct AtomicWaker {
+    waker: AtomicPtr<()>,
+}
+
+impl AtomicWaker {
+    /// Create a new `AtomicWaker`.
+    pub const fn new() -> Self {
+        Self {
+            waker: AtomicPtr::new(ptr::null_mut()),
+        }
+    }
+
+    /// Register a waker. Overwrites the previous waker, if any.
+    pub fn register(&self, w: &Waker) {
+        self.waker.store(w.as_turbo_ptr().as_ptr() as _, Ordering::Release);
+    }
+
+    /// Wake the registered waker, if any.
+    pub fn wake(&self) {
+        if let Some(ptr) = NonNull::new(self.waker.load(Ordering::Acquire)) {
+            unsafe { Waker::from_turbo_ptr(ptr) }.wake();
+        }
+    }
+}
diff --git a/embassy-sync/src/waitqueue/mod.rs b/embassy-sync/src/waitqueue/mod.rs
index 6661a6b61..6b0b0c64e 100644
--- a/embassy-sync/src/waitqueue/mod.rs
+++ b/embassy-sync/src/waitqueue/mod.rs
@@ -1,7 +1,11 @@
 //! Async low-level wait queues
 
-mod waker;
-pub use waker::*;
+#[cfg_attr(feature = "turbowakers", path = "atomic_waker_turbo.rs")]
+mod atomic_waker;
+pub use atomic_waker::*;
+
+mod waker_registration;
+pub use waker_registration::*;
 
 mod multi_waker;
 pub use multi_waker::*;
diff --git a/embassy-sync/src/waitqueue/multi_waker.rs b/embassy-sync/src/waitqueue/multi_waker.rs
index 325d2cb3a..824d192da 100644
--- a/embassy-sync/src/waitqueue/multi_waker.rs
+++ b/embassy-sync/src/waitqueue/multi_waker.rs
@@ -1,33 +1,58 @@
 use core::task::Waker;
 
-use super::WakerRegistration;
+use heapless::Vec;
 
 /// Utility struct to register and wake multiple wakers.
 pub struct MultiWakerRegistration<const N: usize> {
-    wakers: [WakerRegistration; N],
+    wakers: Vec<Waker, N>,
 }
 
 impl<const N: usize> MultiWakerRegistration<N> {
     /// Create a new empty instance
     pub const fn new() -> Self {
-        const WAKER: WakerRegistration = WakerRegistration::new();
-        Self { wakers: [WAKER; N] }
+        Self { wakers: Vec::new() }
     }
 
     /// Register a waker. If the buffer is full the function returns it in the error
-    pub fn register<'a>(&mut self, w: &'a Waker) -> Result<(), &'a Waker> {
-        if let Some(waker_slot) = self.wakers.iter_mut().find(|waker_slot| !waker_slot.occupied()) {
-            waker_slot.register(w);
-            Ok(())
-        } else {
-            Err(w)
+    pub fn register<'a>(&mut self, w: &'a Waker) {
+        // If we already have some waker that wakes the same task as `w`, do nothing.
+        // This avoids cloning wakers, and avoids unnecessary mass-wakes.
+        for w2 in &self.wakers {
+            if w.will_wake(w2) {
+                return;
+            }
+        }
+
+        if self.wakers.is_full() {
+            // All waker slots were full. It's a bit inefficient, but we can wake everything.
+            // Any future that is still active will simply reregister.
+            // This won't happen a lot, so it's ok.
+            self.wake();
+        }
+
+        if self.wakers.push(w.clone()).is_err() {
+            // This can't happen unless N=0
+            // (Either `wakers` wasn't full, or it was in which case `wake()` empied it)
+            panic!("tried to push a waker to a zero-length MultiWakerRegistration")
         }
     }
 
     /// Wake all registered wakers. This clears the buffer
     pub fn wake(&mut self) {
-        for waker_slot in self.wakers.iter_mut() {
-            waker_slot.wake()
+        // heapless::Vec has no `drain()`, do it unsafely ourselves...
+
+        // First set length to 0, without dropping the contents.
+        // This is necessary for soundness: if wake() panics and we're using panic=unwind.
+        // Setting len=0 upfront ensures other code can't observe the vec in an inconsistent state.
+        // (it'll leak wakers, but that's not UB)
+        let len = self.wakers.len();
+        unsafe { self.wakers.set_len(0) }
+
+        for i in 0..len {
+            // Move a waker out of the vec.
+            let waker = unsafe { self.wakers.as_mut_ptr().add(i).read() };
+            // Wake it by value, which consumes (drops) it.
+            waker.wake();
         }
     }
 }
diff --git a/embassy-sync/src/waitqueue/waker.rs b/embassy-sync/src/waitqueue/waker_registration.rs
index 64e300eb8..9b666e7c4 100644
--- a/embassy-sync/src/waitqueue/waker.rs
+++ b/embassy-sync/src/waitqueue/waker_registration.rs
@@ -1,12 +1,8 @@
-use core::cell::Cell;
 use core::mem;
 use core::task::Waker;
 
-use crate::blocking_mutex::raw::CriticalSectionRawMutex;
-use crate::blocking_mutex::Mutex;
-
 /// Utility struct to register and wake a waker.
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub struct WakerRegistration {
     waker: Option<Waker>,
 }
@@ -54,39 +50,3 @@ impl WakerRegistration {
         self.waker.is_some()
     }
 }
-
-/// Utility struct to register and wake a waker.
-pub struct AtomicWaker {
-    waker: Mutex<CriticalSectionRawMutex, Cell<Option<Waker>>>,
-}
-
-impl AtomicWaker {
-    /// Create a new `AtomicWaker`.
-    pub const fn new() -> Self {
-        Self {
-            waker: Mutex::const_new(CriticalSectionRawMutex::new(), Cell::new(None)),
-        }
-    }
-
-    /// Register a waker. Overwrites the previous waker, if any.
-    pub fn register(&self, w: &Waker) {
-        critical_section::with(|cs| {
-            let cell = self.waker.borrow(cs);
-            cell.set(match cell.replace(None) {
-                Some(w2) if (w2.will_wake(w)) => Some(w2),
-                _ => Some(w.clone()),
-            })
-        })
-    }
-
-    /// Wake the registered waker, if any.
-    pub fn wake(&self) {
-        critical_section::with(|cs| {
-            let cell = self.waker.borrow(cs);
-            if let Some(w) = cell.replace(None) {
-                w.wake_by_ref();
-                cell.set(Some(w));
-            }
-        })
-    }
-}