ci: cache lockfiles

- Removes the slow "updating crates.io index, updating git" at start of the job.
- Avoids CI being randomly slower if a widely-used dep (like serde) gets bumped causing rebuilding everything.

.github/ci/build-stable.sh

@@ -12,9 +12,19 @@ export CARGO_TARGET_DIR=/ci/cache/target
 # used when pointing stm32-metapac to a CI-built one.
 export CARGO_NET_GIT_FETCH_WITH_CLI=true
 
+# Restore lockfiles
+if [ -f /ci/cache/lockfiles.tar ]; then
+    echo Restoring lockfiles...
+    tar xf /ci/cache/lockfiles.tar
+fi
+
 hashtime restore /ci/cache/filetime.json || true
 hashtime save /ci/cache/filetime.json
 
 sed -i 's/channel.*/channel = "stable"/g' rust-toolchain.toml
 
 ./ci_stable.sh
+
+# Save lockfiles
+echo Saving lockfiles...
+find . -type f -name Cargo.lock -exec tar -cf /ci/cache/lockfiles.tar '{}' \+

.github/ci/build.sh

@@ -18,7 +18,17 @@ fi
 # used when pointing stm32-metapac to a CI-built one.
 export CARGO_NET_GIT_FETCH_WITH_CLI=true
 
+# Restore lockfiles
+if [ -f /ci/cache/lockfiles.tar ]; then
+    echo Restoring lockfiles...
+    tar xf /ci/cache/lockfiles.tar
+fi
+
 hashtime restore /ci/cache/filetime.json || true
 hashtime save /ci/cache/filetime.json
 
 ./ci.sh
+
+# Save lockfiles
+echo Saving lockfiles...
+find . -type f -name Cargo.lock -exec tar -cf /ci/cache/lockfiles.tar '{}' \+

ci.sh

@@ -38,6 +38,7 @@ cargo batch  \
     --- build --release --manifest-path embassy-sync/Cargo.toml --target thumbv6m-none-eabi --features nightly,defmt \
     --- build --release --manifest-path embassy-time/Cargo.toml --target thumbv6m-none-eabi --features nightly,defmt,defmt-timestamp-uptime,tick-hz-32_768,generic-queue-8 \
     --- build --release --manifest-path embassy-net/Cargo.toml --target thumbv7em-none-eabi --features defmt,tcp,udp,dns,proto-ipv4,medium-ethernet \
+    --- build --release --manifest-path embassy-net/Cargo.toml --target thumbv7em-none-eabi --features defmt,tcp,udp,dns,proto-ipv4,igmp,medium-ethernet \
     --- build --release --manifest-path embassy-net/Cargo.toml --target thumbv7em-none-eabi --features defmt,tcp,udp,dns,dhcpv4,medium-ethernet \
     --- build --release --manifest-path embassy-net/Cargo.toml --target thumbv7em-none-eabi --features defmt,tcp,udp,dns,dhcpv4,medium-ethernet,nightly,dhcpv4-hostname \
     --- build --release --manifest-path embassy-net/Cargo.toml --target thumbv7em-none-eabi --features defmt,tcp,udp,dns,proto-ipv6,medium-ethernet \
@@ -110,6 +111,7 @@ cargo batch  \
     --- build --release --manifest-path embassy-stm32/Cargo.toml --target thumbv7em-none-eabi --features nightly,stm32h753zi,defmt,exti,time-driver-any,unstable-traits,time \
     --- build --release --manifest-path embassy-stm32/Cargo.toml --target thumbv7em-none-eabi --features nightly,stm32h735zg,defmt,exti,time-driver-any,unstable-traits,time \
     --- build --release --manifest-path embassy-stm32/Cargo.toml --target thumbv7em-none-eabi --features nightly,stm32h755zi-cm7,defmt,exti,time-driver-any,unstable-traits,time \
+    --- build --release --manifest-path embassy-stm32/Cargo.toml --target thumbv7em-none-eabi --features nightly,stm32h725re,defmt,exti,time-driver-any,unstable-traits,time \
     --- build --release --manifest-path embassy-stm32/Cargo.toml --target thumbv7em-none-eabi --features nightly,stm32h7b3ai,defmt,exti,time-driver-any,unstable-traits,time \
     --- build --release --manifest-path embassy-stm32/Cargo.toml --target thumbv7em-none-eabi --features nightly,stm32l476vg,defmt,exti,time-driver-any,unstable-traits,time \
     --- build --release --manifest-path embassy-stm32/Cargo.toml --target thumbv7em-none-eabi --features nightly,stm32l422cb,defmt,exti,time-driver-any,unstable-traits,time \

cyw43/src/control.rs

@@ -1,4 +1,5 @@
 use core::cmp::{max, min};
+use core::iter::zip;
 
 use embassy_net_driver_channel as ch;
 use embassy_net_driver_channel::driver::{HardwareAddress, LinkState};
@@ -16,6 +17,12 @@ pub struct Error {
     pub status: u32,
 }
 
+#[derive(Debug)]
+pub enum AddMulticastAddressError {
+    NotMulticast,
+    NoFreeSlots,
+}
+
 pub struct Control<'a> {
     state_ch: ch::StateRunner<'a>,
     events: &'a Events,
@@ -316,6 +323,54 @@ impl<'a> Control<'a> {
         self.set_iovar_u32x2("bss", 0, 1).await; // bss = BSS_UP
     }
 
+    /// Add specified address to the list of hardware addresses the device
+    /// listens on. The address must be a Group address (I/G bit set). Up
+    /// to 10 addresses are supported by the firmware. Returns the number of
+    /// address slots filled after adding, or an error.
+    pub async fn add_multicast_address(&mut self, address: [u8; 6]) -> Result<usize, AddMulticastAddressError> {
+        // The firmware seems to ignore non-multicast addresses, so let's
+        // prevent the user from adding them and wasting space.
+        if address[0] & 0x01 != 1 {
+            return Err(AddMulticastAddressError::NotMulticast);
+        }
+
+        let mut buf = [0; 64];
+        self.get_iovar("mcast_list", &mut buf).await;
+
+        let n = u32::from_le_bytes(buf[..4].try_into().unwrap()) as usize;
+        let (used, free) = buf[4..].split_at_mut(n * 6);
+
+        if used.chunks(6).any(|a| a == address) {
+            return Ok(n);
+        }
+
+        if free.len() < 6 {
+            return Err(AddMulticastAddressError::NoFreeSlots);
+        }
+
+        free[..6].copy_from_slice(&address);
+        let n = n + 1;
+        buf[..4].copy_from_slice(&(n as u32).to_le_bytes());
+
+        self.set_iovar_v::<80>("mcast_list", &buf).await;
+        Ok(n)
+    }
+
+    /// Retrieve the list of configured multicast hardware addresses.
+    pub async fn list_mulistcast_addresses(&mut self, result: &mut [[u8; 6]; 10]) -> usize {
+        let mut buf = [0; 64];
+        self.get_iovar("mcast_list", &mut buf).await;
+
+        let n = u32::from_le_bytes(buf[..4].try_into().unwrap()) as usize;
+        let used = &buf[4..][..n * 6];
+
+        for (addr, output) in zip(used.chunks(6), result.iter_mut()) {
+            output.copy_from_slice(addr)
+        }
+
+        n
+    }
+
     async fn set_iovar_u32x2(&mut self, name: &str, val1: u32, val2: u32) {
         let mut buf = [0; 8];
         buf[0..4].copy_from_slice(&val1.to_le_bytes());

cyw43/src/lib.rs

@@ -27,7 +27,7 @@ use ioctl::IoctlState;
 
 use crate::bus::Bus;
 pub use crate::bus::SpiBusCyw43;
-pub use crate::control::{Control, Error as ControlError, Scanner};
+pub use crate::control::{AddMulticastAddressError, Control, Error as ControlError, Scanner};
 pub use crate::runner::Runner;
 pub use crate::structs::BssInfo;
 

docs/modules/ROOT/nav.adoc

@@ -10,3 +10,4 @@
 * xref:examples.adoc[Examples]
 * xref:developer.adoc[Developer]
 ** xref:developer_stm32.adoc[Developer: STM32]
+* xref:faq.adoc[Frequently Asked Questions]

docs/modules/ROOT/pages/faq.adoc

@@ -0,0 +1,25 @@
+= Frequently Asked Questions
+
+These are a list of unsorted, commonly asked questions and answers.
+
+Please feel free to add items to link:https://github.com/embassy-rs/embassy/edit/main/docs/modules/ROOT/pages/faq.adoc[this page], especially if someone in the chat answered a question for you!
+
+== Missing main macro
+
+If you see an error like this:
+
+[source,rust]
+----
+#[embassy_executor::main]
+|                   ^^^^ could not find `main` in `embassy_executor`
+----
+
+You are likely missing some features of the `embassy-executor` crate.
+
+For Cortex-M targets, consider making sure that ALL of the following features are active in your `Cargo.toml` for the `embassy-executor` crate:
+
+* `arch-cortex-m`
+* `executor-thread`
+* `nightly`
+
+For Xtensa ESP32, consider using the executors and `#[main]` macro provided by your appropriate link:https://crates.io/crates/esp-hal-common[HAL crate].

embassy-executor/CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## 0.3.3 - 2023-11-15
+
+- Add `main` macro reexport for Xtensa arch.
+- Remove use of `atomic-polyfill`. The executor now has multiple implementations of its internal data structures for cases where the target supports atomics or doesn't.
+
 ## 0.3.2 - 2023-11-06
 
 - Use `atomic-polyfill` for `riscv32`

embassy-executor/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "embassy-executor"
-version = "0.3.2"
+version = "0.3.3"
 edition = "2021"
 license = "MIT OR Apache-2.0"
 description = "async/await executor designed for embedded usage"
@@ -34,7 +34,7 @@ _arch = [] # some arch was picked
 arch-std = ["_arch", "critical-section/std"]
 arch-cortex-m = ["_arch", "dep:cortex-m"]
 arch-xtensa = ["_arch"]
-arch-riscv32 = ["_arch"]
+arch-riscv32 = ["_arch", "dep:portable-atomic"]
 arch-wasm = ["_arch", "dep:wasm-bindgen", "dep:js-sys"]
 
 # Enable the thread-mode executor (using WFE/SEV in Cortex-M, WFI in other embedded archs)
@@ -59,9 +59,12 @@ rtos-trace = { version = "0.1.2", optional = true }
 
 embassy-macros = { version = "0.2.1", path = "../embassy-macros" }
 embassy-time = { version = "0.1.5", path = "../embassy-time", optional = true}
-atomic-polyfill = "1.0.1"
 critical-section = "1.1"
 
+# needed for riscv
+# remove when https://github.com/rust-lang/rust/pull/114499 is merged
+portable-atomic = { version = "1.5", optional = true }
+
 # arch-cortex-m dependencies
 cortex-m = { version = "0.7.6", optional = true }
 

embassy-executor/src/arch/cortex_m.rs

@@ -115,12 +115,12 @@ mod thread {
 pub use interrupt::*;
 #[cfg(feature = "executor-interrupt")]
 mod interrupt {
-    use core::cell::UnsafeCell;
+    use core::cell::{Cell, UnsafeCell};
     use core::mem::MaybeUninit;
 
-    use atomic_polyfill::{AtomicBool, Ordering};
     use cortex_m::interrupt::InterruptNumber;
     use cortex_m::peripheral::NVIC;
+    use critical_section::Mutex;
 
     use crate::raw;
 
@@ -146,7 +146,7 @@ mod interrupt {
     /// It is somewhat more complex to use, it's recommended to use the thread-mode
     /// [`Executor`] instead, if it works for your use case.
     pub struct InterruptExecutor {
-        started: AtomicBool,
+        started: Mutex<Cell<bool>>,
         executor: UnsafeCell<MaybeUninit<raw::Executor>>,
     }
 
@@ -158,7 +158,7 @@ mod interrupt {
         #[inline]
         pub const fn new() -> Self {
             Self {
-                started: AtomicBool::new(false),
+                started: Mutex::new(Cell::new(false)),
                 executor: UnsafeCell::new(MaybeUninit::uninit()),
             }
         }
@@ -167,7 +167,8 @@ mod interrupt {
         ///
         /// # Safety
         ///
-        /// You MUST call this from the interrupt handler, and from nowhere else.
+        /// - You MUST call this from the interrupt handler, and from nowhere else.
+        /// - You must not call this before calling `start()`.
         pub unsafe fn on_interrupt(&'static self) {
             let executor = unsafe { (&*self.executor.get()).assume_init_ref() };
             executor.poll();
@@ -196,11 +197,7 @@ mod interrupt {
         /// do it after.
         ///
         pub fn start(&'static self, irq: impl InterruptNumber) -> crate::SendSpawner {
-            if self
-                .started
-                .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
-                .is_err()
-            {
+            if critical_section::with(|cs| self.started.borrow(cs).replace(true)) {
                 panic!("InterruptExecutor::start() called multiple times on the same executor.");
             }
 
@@ -222,10 +219,10 @@ mod interrupt {
         /// This returns a [`SendSpawner`] you can use to spawn tasks on this
         /// executor.
         ///
-        /// This MUST only be called on an executor that has already been spawned.
+        /// This MUST only be called on an executor that has already been started.
         /// The function will panic otherwise.
         pub fn spawner(&'static self) -> crate::SendSpawner {
-            if !self.started.load(Ordering::Acquire) {
+            if !critical_section::with(|cs| self.started.borrow(cs).get()) {
                 panic!("InterruptExecutor::spawner() called on uninitialized executor.");
             }
             let executor = unsafe { (&*self.executor.get()).assume_init_ref() };

embassy-executor/src/arch/riscv32.rs

@@ -7,9 +7,9 @@ pub use thread::*;
 mod thread {
     use core::marker::PhantomData;
 
-    use atomic_polyfill::{AtomicBool, Ordering};
     #[cfg(feature = "nightly")]
     pub use embassy_macros::main_riscv as main;
+    use portable_atomic::{AtomicBool, Ordering};
 
     use crate::{raw, Spawner};
 

embassy-executor/src/arch/xtensa.rs

@@ -8,6 +8,9 @@ mod thread {
     use core::marker::PhantomData;
     use core::sync::atomic::{AtomicBool, Ordering};
 
+    #[cfg(feature = "nightly")]
+    pub use embassy_macros::main_riscv as main;
+
     use crate::{raw, Spawner};
 
     /// global atomic used to keep track of whether there is work to do since sev() is not available on Xtensa

embassy-executor/src/raw/mod.rs

@@ -7,7 +7,15 @@
 //! Using this module requires respecting subtle safety contracts. If you can, prefer using the safe
 //! [executor wrappers](crate::Executor) and the [`embassy_executor::task`](embassy_macros::task) macro, which are fully safe.
 
+#[cfg_attr(target_has_atomic = "ptr", path = "run_queue_atomics.rs")]
+#[cfg_attr(not(target_has_atomic = "ptr"), path = "run_queue_critical_section.rs")]
 mod run_queue;
+
+#[cfg_attr(all(cortex_m, target_has_atomic = "8"), path = "state_atomics_arm.rs")]
+#[cfg_attr(all(not(cortex_m), target_has_atomic = "8"), path = "state_atomics.rs")]
+#[cfg_attr(not(target_has_atomic = "8"), path = "state_critical_section.rs")]
+mod state;
+
 #[cfg(feature = "integrated-timers")]
 mod timer_queue;
 pub(crate) mod util;
@@ -21,7 +29,6 @@ use core::pin::Pin;
 use core::ptr::NonNull;
 use core::task::{Context, Poll};
 
-use atomic_polyfill::{AtomicU32, Ordering};
 #[cfg(feature = "integrated-timers")]
 use embassy_time::driver::{self, AlarmHandle};
 #[cfg(feature = "integrated-timers")]
@@ -30,21 +37,14 @@ use embassy_time::Instant;
 use rtos_trace::trace;
 
 use self::run_queue::{RunQueue, RunQueueItem};
+use self::state::State;
 use self::util::{SyncUnsafeCell, UninitCell};
 pub use self::waker::task_from_waker;
 use super::SpawnToken;
 
-/// Task is spawned (has a future)
-pub(crate) const STATE_SPAWNED: u32 = 1 << 0;
-/// Task is in the executor run queue
-pub(crate) const STATE_RUN_QUEUED: u32 = 1 << 1;
-/// Task is in the executor timer queue
-#[cfg(feature = "integrated-timers")]
-pub(crate) const STATE_TIMER_QUEUED: u32 = 1 << 2;
-
 /// Raw task header for use in task pointers.
 pub(crate) struct TaskHeader {
-    pub(crate) state: AtomicU32,
+    pub(crate) state: State,
     pub(crate) run_queue_item: RunQueueItem,
     pub(crate) executor: SyncUnsafeCell<Option<&'static SyncExecutor>>,
     poll_fn: SyncUnsafeCell<Option<unsafe fn(TaskRef)>>,
@@ -116,7 +116,7 @@ impl<F: Future + 'static> TaskStorage<F> {
     pub const fn new() -> Self {
         Self {
             raw: TaskHeader {
-                state: AtomicU32::new(0),
+                state: State::new(),
                 run_queue_item: RunQueueItem::new(),
                 executor: SyncUnsafeCell::new(None),
                 // Note: this is lazily initialized so that a static `TaskStorage` will go in `.bss`
@@ -161,7 +161,7 @@ impl<F: Future + 'static> TaskStorage<F> {
         match future.poll(&mut cx) {
             Poll::Ready(_) => {
                 this.future.drop_in_place();
-                this.raw.state.fetch_and(!STATE_SPAWNED, Ordering::AcqRel);
+                this.raw.state.despawn();
 
                 #[cfg(feature = "integrated-timers")]
                 this.raw.expires_at.set(Instant::MAX);
@@ -193,11 +193,7 @@ impl<F: Future + 'static> AvailableTask<F> {
     ///
     /// This function returns `None` if a task has already been spawned and has not finished running.
     pub fn claim(task: &'static TaskStorage<F>) -> Option<Self> {
-        task.raw
-            .state
-            .compare_exchange(0, STATE_SPAWNED | STATE_RUN_QUEUED, Ordering::AcqRel, Ordering::Acquire)
-            .ok()
-            .map(|_| Self { task })
+        task.raw.state.spawn().then(|| Self { task })
     }
 
     fn initialize_impl<S>(self, future: impl FnOnce() -> F) -> SpawnToken<S> {
@@ -394,8 +390,7 @@ impl SyncExecutor {
                 #[cfg(feature = "integrated-timers")]
                 task.expires_at.set(Instant::MAX);
 
-                let state = task.state.fetch_and(!STATE_RUN_QUEUED, Ordering::AcqRel);
-                if state & STATE_SPAWNED == 0 {
+                if !task.state.run_dequeue() {
                     // If task is not running, ignore it. This can happen in the following scenario:
                     //   - Task gets dequeued, poll starts
                     //   - While task is being polled, it gets woken. It gets placed in the queue.
@@ -546,18 +541,7 @@ impl Executor {
 /// You can obtain a `TaskRef` from a `Waker` using [`task_from_waker`].
 pub fn wake_task(task: TaskRef) {
     let header = task.header();
-
-    let res = header.state.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |state| {
-        // If already scheduled, or if not started,
-        if (state & STATE_RUN_QUEUED != 0) || (state & STATE_SPAWNED == 0) {
-            None
-        } else {
-            // Mark it as scheduled
-            Some(state | STATE_RUN_QUEUED)
-        }
-    });
-
-    if res.is_ok() {
+    if header.state.run_enqueue() {
         // We have just marked the task as scheduled, so enqueue it.
         unsafe {
             let executor = header.executor.get().unwrap_unchecked();
@@ -571,18 +555,7 @@ pub fn wake_task(task: TaskRef) {
 /// You can obtain a `TaskRef` from a `Waker` using [`task_from_waker`].
 pub fn wake_task_no_pend(task: TaskRef) {
     let header = task.header();
-
-    let res = header.state.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |state| {
-        // If already scheduled, or if not started,
-        if (state & STATE_RUN_QUEUED != 0) || (state & STATE_SPAWNED == 0) {
-            None
-        } else {
-            // Mark it as scheduled
-            Some(state | STATE_RUN_QUEUED)
-        }
-    });
-
-    if res.is_ok() {
+    if header.state.run_enqueue() {
         // We have just marked the task as scheduled, so enqueue it.
         unsafe {
             let executor = header.executor.get().unwrap_unchecked();

embassy-executor/src/raw/run_queue_atomics.rs

@@ -1,7 +1,6 @@
 use core::ptr;
 use core::ptr::NonNull;
-
-use atomic_polyfill::{AtomicPtr, Ordering};
+use core::sync::atomic::{AtomicPtr, Ordering};
 
 use super::{TaskHeader, TaskRef};
 use crate::raw::util::SyncUnsafeCell;

embassy-executor/src/raw/run_queue_critical_section.rs

@@ -0,0 +1,75 @@
+use core::cell::Cell;
+
+use critical_section::{CriticalSection, Mutex};
+
+use super::TaskRef;
+
+pub(crate) struct RunQueueItem {
+    next: Mutex<Cell<Option<TaskRef>>>,
+}
+
+impl RunQueueItem {
+    pub const fn new() -> Self {
+        Self {
+            next: Mutex::new(Cell::new(None)),
+        }
+    }
+}
+
+/// Atomic task queue using a very, very simple lock-free linked-list queue:
+///
+/// To enqueue a task, task.next is set to the old head, and head is atomically set to task.
+///
+/// Dequeuing is done in batches: the queue is emptied by atomically replacing head with
+/// null. Then the batch is iterated following the next pointers until null is reached.
+///
+/// Note that batches will be iterated in the reverse order as they were enqueued. This is OK
+/// for our purposes: it can't create fairness problems since the next batch won't run until the
+/// current batch is completely processed, so even if a task enqueues itself instantly (for example
+/// by waking its own waker) can't prevent other tasks from running.
+pub(crate) struct RunQueue {
+    head: Mutex<Cell<Option<TaskRef>>>,
+}
+
+impl RunQueue {
+    pub const fn new() -> Self {
+        Self {
+            head: Mutex::new(Cell::new(None)),
+        }
+    }
+
+    /// Enqueues an item. Returns true if the queue was empty.
+    ///
+    /// # Safety
+    ///
+    /// `item` must NOT be already enqueued in any queue.
+    #[inline(always)]
+    pub(crate) unsafe fn enqueue(&self, task: TaskRef) -> bool {
+        critical_section::with(|cs| {
+            let prev = self.head.borrow(cs).replace(Some(task));
+            task.header().run_queue_item.next.borrow(cs).set(prev);
+
+            prev.is_none()
+        })
+    }
+
+    /// Empty the queue, then call `on_task` for each task that was in the queue.
+    /// NOTE: It is OK for `on_task` to enqueue more tasks. In this case they're left in the queue
+    /// and will be processed by the *next* call to `dequeue_all`, *not* the current one.
+    pub(crate) fn dequeue_all(&self, on_task: impl Fn(TaskRef)) {
+        // Atomically empty the queue.
+        let mut next = critical_section::with(|cs| self.head.borrow(cs).take());
+
+        // Iterate the linked list of tasks that were previously in the queue.
+        while let Some(task) = next {
+            // If the task re-enqueues itself, the `next` pointer will get overwritten.
+            // Therefore, first read the next pointer, and only then process the task.
+
+            // safety: we know if the task is enqueued, no one else will touch the `next` pointer.
+            let cs = unsafe { CriticalSection::new() };
+            next = task.header().run_queue_item.next.borrow(cs).get();
+
+            on_task(task);
+        }
+    }
+}

embassy-executor/src/raw/state_atomics.rs

@@ -0,0 +1,73 @@
+use core::sync::atomic::{AtomicU32, Ordering};
+
+/// Task is spawned (has a future)
+pub(crate) const STATE_SPAWNED: u32 = 1 << 0;
+/// Task is in the executor run queue
+pub(crate) const STATE_RUN_QUEUED: u32 = 1 << 1;
+/// Task is in the executor timer queue
+#[cfg(feature = "integrated-timers")]
+pub(crate) const STATE_TIMER_QUEUED: u32 = 1 << 2;
+
+pub(crate) struct State {
+    state: AtomicU32,
+}
+
+impl State {
+    pub const fn new() -> State {
+        Self {
+            state: AtomicU32::new(0),
+        }
+    }
+
+    /// If task is idle, mark it as spawned + run_queued and return true.
+    #[inline(always)]
+    pub fn spawn(&self) -> bool {
+        self.state
+            .compare_exchange(0, STATE_SPAWNED | STATE_RUN_QUEUED, Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+    }
+
+    /// Unmark the task as spawned.
+    #[inline(always)]
+    pub fn despawn(&self) {
+        self.state.fetch_and(!STATE_SPAWNED, Ordering::AcqRel);
+    }
+
+    /// Mark the task as run-queued if it's spawned and isn't already run-queued. Return true on success.
+    #[inline(always)]
+    pub fn run_enqueue(&self) -> bool {
+        self.state
+            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |state| {
+                // If already scheduled, or if not started,
+                if (state & STATE_RUN_QUEUED != 0) || (state & STATE_SPAWNED == 0) {
+                    None
+                } else {
+                    // Mark it as scheduled
+                    Some(state | STATE_RUN_QUEUED)
+                }
+            })
+            .is_ok()
+    }
+
+    /// Unmark the task as run-queued. Return whether the task is spawned.
+    #[inline(always)]
+    pub fn run_dequeue(&self) -> bool {
+        let state = self.state.fetch_and(!STATE_RUN_QUEUED, Ordering::AcqRel);
+        state & STATE_SPAWNED != 0
+    }
+
+    /// Mark the task as timer-queued. Return whether it was newly queued (i.e. not queued before)
+    #[cfg(feature = "integrated-timers")]
+    #[inline(always)]
+    pub fn timer_enqueue(&self) -> bool {
+        let old_state = self.state.fetch_or(STATE_TIMER_QUEUED, Ordering::AcqRel);
+        old_state & STATE_TIMER_QUEUED == 0
+    }
+
+    /// Unmark the task as timer-queued.
+    #[cfg(feature = "integrated-timers")]
+    #[inline(always)]
+    pub fn timer_dequeue(&self) {
+        self.state.fetch_and(!STATE_TIMER_QUEUED, Ordering::AcqRel);
+    }
+}

embassy-executor/src/raw/state_atomics_arm.rs

@@ -0,0 +1,103 @@
+use core::arch::asm;
+use core::sync::atomic::{compiler_fence, AtomicBool, AtomicU32, Ordering};
+
+// Must be kept in sync with the layout of `State`!
+pub(crate) const STATE_SPAWNED: u32 = 1 << 0;
+pub(crate) const STATE_RUN_QUEUED: u32 = 1 << 8;
+
+#[repr(C, align(4))]
+pub(crate) struct State {
+    /// Task is spawned (has a future)
+    spawned: AtomicBool,
+    /// Task is in the executor run queue
+    run_queued: AtomicBool,
+    /// Task is in the executor timer queue
+    timer_queued: AtomicBool,
+    pad: AtomicBool,
+}
+
+impl State {
+    pub const fn new() -> State {
+        Self {
+            spawned: AtomicBool::new(false),
+            run_queued: AtomicBool::new(false),
+            timer_queued: AtomicBool::new(false),
+            pad: AtomicBool::new(false),
+        }
+    }
+
+    fn as_u32(&self) -> &AtomicU32 {
+        unsafe { &*(self as *const _ as *const AtomicU32) }
+    }
+
+    /// If task is idle, mark it as spawned + run_queued and return true.
+    #[inline(always)]
+    pub fn spawn(&self) -> bool {
+        compiler_fence(Ordering::Release);
+        let r = self
+            .as_u32()
+            .compare_exchange(
+                0,
+                STATE_SPAWNED | STATE_RUN_QUEUED,
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+            )
+            .is_ok();
+        compiler_fence(Ordering::Acquire);
+        r
+    }
+
+    /// Unmark the task as spawned.
+    #[inline(always)]
+    pub fn despawn(&self) {
+        compiler_fence(Ordering::Release);
+        self.spawned.store(false, Ordering::Relaxed);
+    }
+
+    /// Mark the task as run-queued if it's spawned and isn't already run-queued. Return true on success.
+    #[inline(always)]
+    pub fn run_enqueue(&self) -> bool {
+        unsafe {
+            loop {
+                let state: u32;
+                asm!("ldrex {}, [{}]", out(reg) state, in(reg) self, options(nostack));
+
+                if (state & STATE_RUN_QUEUED != 0) || (state & STATE_SPAWNED == 0) {
+                    asm!("clrex", options(nomem, nostack));
+                    return false;
+                }
+
+                let outcome: usize;
+                let new_state = state | STATE_RUN_QUEUED;
+                asm!("strex {}, {}, [{}]", out(reg) outcome, in(reg) new_state, in(reg) self, options(nostack));
+                if outcome == 0 {
+                    return true;
+                }
+            }
+        }
+    }
+
+    /// Unmark the task as run-queued. Return whether the task is spawned.
+    #[inline(always)]
+    pub fn run_dequeue(&self) -> bool {
+        compiler_fence(Ordering::Release);
+
+        let r = self.spawned.load(Ordering::Relaxed);
+        self.run_queued.store(false, Ordering::Relaxed);
+        r
+    }
+
+    /// Mark the task as timer-queued. Return whether it was newly queued (i.e. not queued before)
+    #[cfg(feature = "integrated-timers")]
+    #[inline(always)]
+    pub fn timer_enqueue(&self) -> bool {
+        !self.timer_queued.swap(true, Ordering::Relaxed)
+    }
+
+    /// Unmark the task as timer-queued.
+    #[cfg(feature = "integrated-timers")]
+    #[inline(always)]
+    pub fn timer_dequeue(&self) {
+        self.timer_queued.store(false, Ordering::Relaxed);
+    }
+}

embassy-executor/src/raw/state_critical_section.rs

@@ -0,0 +1,93 @@
+use core::cell::Cell;
+
+use critical_section::Mutex;
+
+/// Task is spawned (has a future)
+pub(crate) const STATE_SPAWNED: u32 = 1 << 0;
+/// Task is in the executor run queue
+pub(crate) const STATE_RUN_QUEUED: u32 = 1 << 1;
+/// Task is in the executor timer queue
+#[cfg(feature = "integrated-timers")]
+pub(crate) const STATE_TIMER_QUEUED: u32 = 1 << 2;
+
+pub(crate) struct State {
+    state: Mutex<Cell<u32>>,
+}
+
+impl State {
+    pub const fn new() -> State {
+        Self {
+            state: Mutex::new(Cell::new(0)),
+        }
+    }
+
+    fn update<R>(&self, f: impl FnOnce(&mut u32) -> R) -> R {
+        critical_section::with(|cs| {
+            let s = self.state.borrow(cs);
+            let mut val = s.get();
+            let r = f(&mut val);
+            s.set(val);
+            r
+        })
+    }
+
+    /// If task is idle, mark it as spawned + run_queued and return true.
+    #[inline(always)]
+    pub fn spawn(&self) -> bool {
+        self.update(|s| {
+            if *s == 0 {
+                *s = STATE_SPAWNED | STATE_RUN_QUEUED;
+                true
+            } else {
+                false
+            }
+        })
+    }
+
+    /// Unmark the task as spawned.
+    #[inline(always)]
+    pub fn despawn(&self) {
+        self.update(|s| *s &= !STATE_SPAWNED);
+    }
+
+    /// Mark the task as run-queued if it's spawned and isn't already run-queued. Return true on success.
+    #[inline(always)]
+    pub fn run_enqueue(&self) -> bool {
+        self.update(|s| {
+            if (*s & STATE_RUN_QUEUED != 0) || (*s & STATE_SPAWNED == 0) {
+                false
+            } else {
+                *s |= STATE_RUN_QUEUED;
+                true
+            }
+        })
+    }
+
+    /// Unmark the task as run-queued. Return whether the task is spawned.
+    #[inline(always)]
+    pub fn run_dequeue(&self) -> bool {
+        self.update(|s| {
+            let ok = *s & STATE_SPAWNED != 0;
+            *s &= !STATE_RUN_QUEUED;
+            ok
+        })
+    }
+
+    /// Mark the task as timer-queued. Return whether it was newly queued (i.e. not queued before)
+    #[cfg(feature = "integrated-timers")]
+    #[inline(always)]
+    pub fn timer_enqueue(&self) -> bool {
+        self.update(|s| {
+            let ok = *s & STATE_TIMER_QUEUED == 0;
+            *s |= STATE_TIMER_QUEUED;
+            ok
+        })
+    }
+
+    /// Unmark the task as timer-queued.
+    #[cfg(feature = "integrated-timers")]
+    #[inline(always)]
+    pub fn timer_dequeue(&self) {
+        self.update(|s| *s &= !STATE_TIMER_QUEUED);
+    }
+}

embassy-executor/src/raw/timer_queue.rs

@@ -1,9 +1,8 @@
 use core::cmp::min;
 
-use atomic_polyfill::Ordering;
 use embassy_time::Instant;
 
-use super::{TaskRef, STATE_TIMER_QUEUED};
+use super::TaskRef;
 use crate::raw::util::SyncUnsafeCell;
 
 pub(crate) struct TimerQueueItem {
@@ -32,10 +31,7 @@ impl TimerQueue {
     pub(crate) unsafe fn update(&self, p: TaskRef) {
         let task = p.header();
         if task.expires_at.get() != Instant::MAX {
-            let old_state = task.state.fetch_or(STATE_TIMER_QUEUED, Ordering::AcqRel);
-            let is_new = old_state & STATE_TIMER_QUEUED == 0;
-
-            if is_new {
+            if task.state.timer_enqueue() {
                 task.timer_queue_item.next.set(self.head.get());
                 self.head.set(Some(p));
             }
@@ -75,7 +71,7 @@ impl TimerQueue {
             } else {
                 // Remove it
                 prev.set(task.timer_queue_item.next.get());
-                task.state.fetch_and(!STATE_TIMER_QUEUED, Ordering::AcqRel);
+                task.state.timer_dequeue();
             }
         }
     }

embassy-net/CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Unreleased
 
 - Avoid never resolving `TcpIo::read` when the output buffer is empty.
+- Update to `smoltcp` git.
+- Forward constants from `smoltcp` in DNS query results so changing DNS result size in `smoltcp` properly propagates.
 
 ## 0.2.1 - 2023-10-31
 

embassy-net/Cargo.toml

@@ -25,7 +25,7 @@ features = ["nightly", "defmt", "tcp", "udp", "dns", "dhcpv4", "proto-ipv6", "me
 default = []
 std = []
 
-defmt = ["dep:defmt", "smoltcp/defmt", "embassy-net-driver/defmt"]
+defmt = ["dep:defmt", "smoltcp/defmt", "embassy-net-driver/defmt", "heapless/defmt-03"]
 
 nightly = ["dep:embedded-io-async", "dep:embedded-nal-async"]
 
@@ -46,7 +46,7 @@ igmp = ["smoltcp/proto-igmp"]
 defmt = { version = "0.3", optional = true }
 log = { version = "0.4.14", optional = true }
 
-smoltcp = { git = "https://github.com/smoltcp-rs/smoltcp", rev = "9b791ae3057e10f7afcb70c67deb5daf714293a9", default-features = false, features = [
+smoltcp = { git = "https://github.com/smoltcp-rs/smoltcp.git", rev = "b57e2f9e70e82a13f31d5ea17e55232c11cc2b2d", default-features = false, features = [
   "socket",
   "async",
 ] }

embassy-net/src/dns.rs

@@ -63,7 +63,11 @@ where
     }
 
     /// Make a query for a given name and return the corresponding IP addresses.
-    pub async fn query(&self, name: &str, qtype: DnsQueryType) -> Result<Vec<IpAddress, 1>, Error> {
+    pub async fn query(
+        &self,
+        name: &str,
+        qtype: DnsQueryType,
+    ) -> Result<Vec<IpAddress, { smoltcp::config::DNS_MAX_RESULT_COUNT }>, Error> {
         self.stack.dns_query(name, qtype).await
     }
 }

embassy-net/src/lib.rs

@@ -494,7 +494,11 @@ impl<D: Driver> Stack<D> {
 
     /// Make a query for a given name and return the corresponding IP addresses.
     #[cfg(feature = "dns")]
-    pub async fn dns_query(&self, name: &str, qtype: dns::DnsQueryType) -> Result<Vec<IpAddress, 1>, dns::Error> {
+    pub async fn dns_query(
+        &self,
+        name: &str,
+        qtype: dns::DnsQueryType,
+    ) -> Result<Vec<IpAddress, { smoltcp::config::DNS_MAX_RESULT_COUNT }>, dns::Error> {
         // For A and AAAA queries we try detect whether `name` is just an IP address
         match qtype {
             #[cfg(feature = "proto-ipv4")]
@@ -612,9 +616,11 @@ impl<D: Driver> Stack<D> {
         let addr = addr.into();
 
         self.with_mut(|s, i| {
+            let (_hardware_addr, medium) = to_smoltcp_hardware_address(i.device.hardware_address());
             let mut smoldev = DriverAdapter {
                 cx: Some(cx),
                 inner: &mut i.device,
+                medium,
             };
 
             match s
@@ -649,9 +655,11 @@ impl<D: Driver> Stack<D> {
         let addr = addr.into();
 
         self.with_mut(|s, i| {
+            let (_hardware_addr, medium) = to_smoltcp_hardware_address(i.device.hardware_address());
             let mut smoldev = DriverAdapter {
                 cx: Some(cx),
                 inner: &mut i.device,
+                medium,
             };
 
             match s

embassy-nrf/src/buffered_uarte.rs

@@ -12,7 +12,7 @@ use core::cmp::min;
 use core::future::poll_fn;
 use core::marker::PhantomData;
 use core::slice;
-use core::sync::atomic::{compiler_fence, AtomicU8, AtomicUsize, Ordering};
+use core::sync::atomic::{compiler_fence, AtomicBool, AtomicU8, AtomicUsize, Ordering};
 use core::task::Poll;
 
 use embassy_hal_internal::atomic_ring_buffer::RingBuffer;
@@ -41,7 +41,9 @@ mod sealed {
 
         pub rx_waker: AtomicWaker,
         pub rx_buf: RingBuffer,
-        pub rx_bufs: AtomicU8,
+        pub rx_started: AtomicBool,
+        pub rx_started_count: AtomicU8,
+        pub rx_ended_count: AtomicU8,
         pub rx_ppi_ch: AtomicU8,
     }
 }
@@ -65,7 +67,9 @@ impl State {
 
             rx_waker: AtomicWaker::new(),
             rx_buf: RingBuffer::new(),
-            rx_bufs: AtomicU8::new(0),
+            rx_started: AtomicBool::new(false),
+            rx_started_count: AtomicU8::new(0),
+            rx_ended_count: AtomicU8::new(0),
             rx_ppi_ch: AtomicU8::new(0),
         }
     }
@@ -104,28 +108,20 @@ impl<U: UarteInstance> interrupt::typelevel::Handler<U::Interrupt> for Interrupt
             s.rx_waker.wake();
         }
 
-        // If not RXing, start.
-        if s.rx_bufs.load(Ordering::Relaxed) == 0 {
-            let (ptr, len) = rx.push_buf();
-            if len >= half_len {
-                //trace!("  irq_rx: starting {:?}", half_len);
-                s.rx_bufs.store(1, Ordering::Relaxed);
+        if r.events_endrx.read().bits() != 0 {
+            //trace!("  irq_rx: endrx");
+            r.events_endrx.reset();
 
-                // Set up the DMA read
-                r.rxd.ptr.write(|w| unsafe { w.ptr().bits(ptr as u32) });
-                r.rxd.maxcnt.write(|w| unsafe { w.maxcnt().bits(half_len as _) });
-
-                // Start UARTE Receive transaction
-                r.tasks_startrx.write(|w| unsafe { w.bits(1) });
-                rx.push_done(half_len);
-                r.intenset.write(|w| w.rxstarted().set());
-            }
+            let val = s.rx_ended_count.load(Ordering::Relaxed);
+            s.rx_ended_count.store(val.wrapping_add(1), Ordering::Relaxed);
         }
 
-        if r.events_rxstarted.read().bits() != 0 {
+        if r.events_rxstarted.read().bits() != 0 || !s.rx_started.load(Ordering::Relaxed) {
             //trace!("  irq_rx: rxstarted");
             let (ptr, len) = rx.push_buf();
             if len >= half_len {
+                r.events_rxstarted.reset();
+
                 //trace!("  irq_rx: starting second {:?}", half_len);
 
                 // Set up the DMA read
@@ -134,11 +130,50 @@ impl<U: UarteInstance> interrupt::typelevel::Handler<U::Interrupt> for Interrupt
 
                 let chn = s.rx_ppi_ch.load(Ordering::Relaxed);
 
+                // Enable endrx -> startrx PPI channel.
+                // From this point on, if endrx happens, startrx is automatically fired.
                 ppi::regs().chenset.write(|w| unsafe { w.bits(1 << chn) });
 
+                // It is possible that endrx happened BEFORE enabling the PPI. In this case
+                // the PPI channel doesn't trigger, and we'd hang. We have to detect this
+                // and manually start.
+
+                // check again in case endrx has happened between the last check and now.
+                if r.events_endrx.read().bits() != 0 {
+                    //trace!("  irq_rx: endrx");
+                    r.events_endrx.reset();
+
+                    let val = s.rx_ended_count.load(Ordering::Relaxed);
+                    s.rx_ended_count.store(val.wrapping_add(1), Ordering::Relaxed);
+                }
+
+                let rx_ended = s.rx_ended_count.load(Ordering::Relaxed);
+                let rx_started = s.rx_started_count.load(Ordering::Relaxed);
+
+                // If we started the same amount of transfers as ended, the last rxend has
+                // already occured.
+                let rxend_happened = rx_started == rx_ended;
+
+                // Check if the PPI channel is still enabled. The PPI channel disables itself
+                // when it fires, so if it's still enabled it hasn't fired.
+                let ppi_ch_enabled = ppi::regs().chen.read().bits() & (1 << chn) != 0;
+
+                // if rxend happened, and the ppi channel hasn't fired yet, the rxend got missed.
+                // this condition also naturally matches if `!started`, needed to kickstart the DMA.
+                if rxend_happened && ppi_ch_enabled {
+                    //trace!("manually starting.");
+
+                    // disable the ppi ch, it's of no use anymore.
+                    ppi::regs().chenclr.write(|w| unsafe { w.bits(1 << chn) });
+
+                    // manually start
+                    r.tasks_startrx.write(|w| unsafe { w.bits(1) });
+                }
+
                 rx.push_done(half_len);
 
-                r.events_rxstarted.reset();
+                s.rx_started_count.store(rx_started.wrapping_add(1), Ordering::Relaxed);
+                s.rx_started.store(true, Ordering::Relaxed);
             } else {
                 //trace!("  irq_rx: rxstarted no buf");
                 r.intenclr.write(|w| w.rxstarted().clear());
@@ -282,6 +317,8 @@ impl<'d, U: UarteInstance, T: TimerInstance> BufferedUarte<'d, U, T> {
 
         let r = U::regs();
 
+        let hwfc = cts.is_some();
+
         rxd.conf().write(|w| w.input().connect().drive().h0h1());
         r.psel.rxd.write(|w| unsafe { w.bits(rxd.psel_bits()) });
 
@@ -303,7 +340,8 @@ impl<'d, U: UarteInstance, T: TimerInstance> BufferedUarte<'d, U, T> {
         // Initialize state
         let s = U::buffered_state();
         s.tx_count.store(0, Ordering::Relaxed);
-        s.rx_bufs.store(0, Ordering::Relaxed);
+        s.rx_started_count.store(0, Ordering::Relaxed);
+        s.rx_ended_count.store(0, Ordering::Relaxed);
         let len = tx_buffer.len();
         unsafe { s.tx_buf.init(tx_buffer.as_mut_ptr(), len) };
         let len = rx_buffer.len();
@@ -311,7 +349,7 @@ impl<'d, U: UarteInstance, T: TimerInstance> BufferedUarte<'d, U, T> {
 
         // Configure
         r.config.write(|w| {
-            w.hwfc().bit(false);
+            w.hwfc().bit(hwfc);
             w.parity().variant(config.parity);
             w
         });
@@ -333,6 +371,7 @@ impl<'d, U: UarteInstance, T: TimerInstance> BufferedUarte<'d, U, T> {
             w.endtx().set();
             w.rxstarted().set();
             w.error().set();
+            w.endrx().set();
             w
         });
 

embassy-rp/Cargo.toml

@@ -94,5 +94,5 @@ pio = {version= "0.2.1" }
 rp2040-boot2 = "0.3"
 
 [dev-dependencies]
-embassy-executor = { version = "0.3.1", path = "../embassy-executor", features = ["nightly", "arch-std", "executor-thread"] }
+embassy-executor = { version = "0.3.3", path = "../embassy-executor", features = ["nightly", "arch-std", "executor-thread"] }
 static_cell = { version = "2" }

embassy-stm32/Cargo.toml

@@ -39,7 +39,7 @@ embassy-hal-internal = {version = "0.1.0", path = "../embassy-hal-internal", fea
 embassy-embedded-hal = {version = "0.1.0", path = "../embassy-embedded-hal" }
 embassy-net-driver = { version = "0.2.0", path = "../embassy-net-driver" }
 embassy-usb-driver = {version = "0.1.0", path = "../embassy-usb-driver", optional = true }
-embassy-executor = { version = "0.3.1", path = "../embassy-executor", optional = true }
+embassy-executor = { version = "0.3.3", path = "../embassy-executor", optional = true }
 
 embedded-hal-02 = { package = "embedded-hal", version = "0.2.6", features = ["unproven"] }
 embedded-hal-1 = { package = "embedded-hal", version = "=1.0.0-rc.1", optional = true}

embassy-stm32/build.rs

@@ -61,6 +61,7 @@ fn main() {
     let mut singletons: Vec<String> = Vec::new();
     for p in METADATA.peripherals {
         if let Some(r) = &p.registers {
+            println!("cargo:rustc-cfg=peri_{}", p.name.to_ascii_lowercase());
             match r.kind {
                 // Generate singletons per pin, not per port
                 "gpio" => {

embassy-stm32/src/can/bxcan.rs

@@ -1,4 +1,3 @@
-use core::cell::{RefCell, RefMut};
 use core::future::poll_fn;
 use core::marker::PhantomData;
 use core::ops::{Deref, DerefMut};
@@ -84,7 +83,7 @@ impl<T: Instance> interrupt::typelevel::Handler<T::SCEInterrupt> for SceInterrup
 }
 
 pub struct Can<'d, T: Instance> {
-    pub can: RefCell<bxcan::Can<BxcanInstance<'d, T>>>,
+    pub can: bxcan::Can<BxcanInstance<'d, T>>,
 }
 
 #[derive(Debug)]
@@ -175,17 +174,12 @@ impl<'d, T: Instance> Can<'d, T> {
         tx.set_as_af(tx.af_num(), AFType::OutputPushPull);
 
         let can = bxcan::Can::builder(BxcanInstance(peri)).leave_disabled();
-        let can_ref_cell = RefCell::new(can);
-        Self { can: can_ref_cell }
+        Self { can }
     }
 
     pub fn set_bitrate(&mut self, bitrate: u32) {
         let bit_timing = Self::calc_bxcan_timings(T::frequency(), bitrate).unwrap();
-        self.can
-            .borrow_mut()
-            .modify_config()
-            .set_bit_timing(bit_timing)
-            .leave_disabled();
+        self.can.modify_config().set_bit_timing(bit_timing).leave_disabled();
     }
 
     /// Enables the peripheral and synchronizes with the bus.
@@ -193,7 +187,7 @@ impl<'d, T: Instance> Can<'d, T> {
     /// This will wait for 11 consecutive recessive bits (bus idle state).
     /// Contrary to enable method from bxcan library, this will not freeze the executor while waiting.
     pub async fn enable(&mut self) {
-        while self.borrow_mut().enable_non_blocking().is_err() {
+        while self.enable_non_blocking().is_err() {
             // SCE interrupt is only generated for entering sleep mode, but not leaving.
             // Yield to allow other tasks to execute while can bus is initializing.
             embassy_futures::yield_now().await;
@@ -202,46 +196,46 @@ impl<'d, T: Instance> Can<'d, T> {
 
     /// Queues the message to be sent but exerts backpressure
     pub async fn write(&mut self, frame: &Frame) -> bxcan::TransmitStatus {
-        CanTx { can: &self.can }.write(frame).await
+        self.split().0.write(frame).await
     }
 
     /// Attempts to transmit a frame without blocking.
     ///
     /// Returns [Err(TryWriteError::Full)] if all transmit mailboxes are full.
     pub fn try_write(&mut self, frame: &Frame) -> Result<bxcan::TransmitStatus, TryWriteError> {
-        CanTx { can: &self.can }.try_write(frame)
+        self.split().0.try_write(frame)
     }
 
     /// Waits for a specific transmit mailbox to become empty
     pub async fn flush(&self, mb: bxcan::Mailbox) {
-        CanTx { can: &self.can }.flush(mb).await
+        CanTx::<T>::flush_inner(mb).await
     }
 
     /// Waits until any of the transmit mailboxes become empty
     pub async fn flush_any(&self) {
-        CanTx { can: &self.can }.flush_any().await
+        CanTx::<T>::flush_any_inner().await
     }
 
     /// Waits until all of the transmit mailboxes become empty
     pub async fn flush_all(&self) {
-        CanTx { can: &self.can }.flush_all().await
+        CanTx::<T>::flush_all_inner().await
     }
 
     /// Returns a tuple of the time the message was received and the message frame
     pub async fn read(&mut self) -> Result<Envelope, BusError> {
-        CanRx { can: &self.can }.read().await
+        self.split().1.read().await
     }
 
     /// Attempts to read a can frame without blocking.
     ///
     /// Returns [Err(TryReadError::Empty)] if there are no frames in the rx queue.
     pub fn try_read(&mut self) -> Result<Envelope, TryReadError> {
-        CanRx { can: &self.can }.try_read()
+        self.split().1.try_read()
     }
 
     /// Waits while receive queue is empty.
     pub async fn wait_not_empty(&mut self) {
-        CanRx { can: &self.can }.wait_not_empty().await
+        self.split().1.wait_not_empty().await
     }
 
     unsafe fn receive_fifo(fifo: RxFifo) {
@@ -385,24 +379,25 @@ impl<'d, T: Instance> Can<'d, T> {
         Some((sjw - 1) << 24 | (bs1 as u32 - 1) << 16 | (bs2 as u32 - 1) << 20 | (prescaler - 1))
     }
 
-    pub fn split<'c>(&'c self) -> (CanTx<'c, 'd, T>, CanRx<'c, 'd, T>) {
-        (CanTx { can: &self.can }, CanRx { can: &self.can })
+    pub fn split<'c>(&'c mut self) -> (CanTx<'c, 'd, T>, CanRx<'c, 'd, T>) {
+        let (tx, rx0, rx1) = self.can.split_by_ref();
+        (CanTx { tx }, CanRx { rx0, rx1 })
     }
 
-    pub fn as_mut(&self) -> RefMut<'_, bxcan::Can<BxcanInstance<'d, T>>> {
-        self.can.borrow_mut()
+    pub fn as_mut(&mut self) -> &mut bxcan::Can<BxcanInstance<'d, T>> {
+        &mut self.can
     }
 }
 
 pub struct CanTx<'c, 'd, T: Instance> {
-    can: &'c RefCell<bxcan::Can<BxcanInstance<'d, T>>>,
+    tx: &'c mut bxcan::Tx<BxcanInstance<'d, T>>,
 }
 
 impl<'c, 'd, T: Instance> CanTx<'c, 'd, T> {
     pub async fn write(&mut self, frame: &Frame) -> bxcan::TransmitStatus {
         poll_fn(|cx| {
             T::state().tx_waker.register(cx.waker());
-            if let Ok(status) = self.can.borrow_mut().transmit(frame) {
+            if let Ok(status) = self.tx.transmit(frame) {
                 return Poll::Ready(status);
             }
 
@@ -415,11 +410,10 @@ impl<'c, 'd, T: Instance> CanTx<'c, 'd, T> {
     ///
     /// Returns [Err(TryWriteError::Full)] if all transmit mailboxes are full.
     pub fn try_write(&mut self, frame: &Frame) -> Result<bxcan::TransmitStatus, TryWriteError> {
-        self.can.borrow_mut().transmit(frame).map_err(|_| TryWriteError::Full)
+        self.tx.transmit(frame).map_err(|_| TryWriteError::Full)
     }
 
-    /// Waits for a specific transmit mailbox to become empty
-    pub async fn flush(&self, mb: bxcan::Mailbox) {
+    async fn flush_inner(mb: bxcan::Mailbox) {
         poll_fn(|cx| {
             T::state().tx_waker.register(cx.waker());
             if T::regs().tsr().read().tme(mb.index()) {
@@ -431,8 +425,12 @@ impl<'c, 'd, T: Instance> CanTx<'c, 'd, T> {
         .await;
     }
 
-    /// Waits until any of the transmit mailboxes become empty
-    pub async fn flush_any(&self) {
+    /// Waits for a specific transmit mailbox to become empty
+    pub async fn flush(&self, mb: bxcan::Mailbox) {
+        Self::flush_inner(mb).await
+    }
+
+    async fn flush_any_inner() {
         poll_fn(|cx| {
             T::state().tx_waker.register(cx.waker());
 
@@ -449,8 +447,12 @@ impl<'c, 'd, T: Instance> CanTx<'c, 'd, T> {
         .await;
     }
 
-    /// Waits until all of the transmit mailboxes become empty
-    pub async fn flush_all(&self) {
+    /// Waits until any of the transmit mailboxes become empty
+    pub async fn flush_any(&self) {
+        Self::flush_any_inner().await
+    }
+
+    async fn flush_all_inner() {
         poll_fn(|cx| {
             T::state().tx_waker.register(cx.waker());
 
@@ -466,11 +468,17 @@ impl<'c, 'd, T: Instance> CanTx<'c, 'd, T> {
         })
         .await;
     }
+
+    /// Waits until all of the transmit mailboxes become empty
+    pub async fn flush_all(&self) {
+        Self::flush_all_inner().await
+    }
 }
 
 #[allow(dead_code)]
 pub struct CanRx<'c, 'd, T: Instance> {
-    can: &'c RefCell<bxcan::Can<BxcanInstance<'d, T>>>,
+    rx0: &'c mut bxcan::Rx0<BxcanInstance<'d, T>>,
+    rx1: &'c mut bxcan::Rx1<BxcanInstance<'d, T>>,
 }
 
 impl<'c, 'd, T: Instance> CanRx<'c, 'd, T> {
@@ -538,7 +546,7 @@ impl<'d, T: Instance> Drop for Can<'d, T> {
 }
 
 impl<'d, T: Instance> Deref for Can<'d, T> {
-    type Target = RefCell<bxcan::Can<BxcanInstance<'d, T>>>;
+    type Target = bxcan::Can<BxcanInstance<'d, T>>;
 
     fn deref(&self) -> &Self::Target {
         &self.can

embassy-stm32/src/dma/gpdma.rs

@@ -299,19 +299,15 @@ impl<'a, C: Channel> Transfer<'a, C> {
 
     pub fn request_stop(&mut self) {
         let ch = self.channel.regs().ch(self.channel.num());
-
-        // Disable the channel. Keep the IEs enabled so the irqs still fire.
-        ch.cr().write(|w| {
-            w.set_tcie(true);
-            w.set_useie(true);
-            w.set_dteie(true);
-            w.set_suspie(true);
+        ch.cr().modify(|w| {
+            w.set_susp(true);
         })
     }
 
     pub fn is_running(&mut self) -> bool {
         let ch = self.channel.regs().ch(self.channel.num());
-        !ch.sr().read().tcf()
+        let sr = ch.sr().read();
+        !sr.tcf() && !sr.suspf()
     }
 
     /// Gets the total remaining transfers for the channel

embassy-stm32/src/i2c/v2.rs

@@ -1,21 +1,16 @@
 use core::cmp;
-#[cfg(feature = "time")]
 use core::future::poll_fn;
 use core::marker::PhantomData;
-#[cfg(feature = "time")]
 use core::task::Poll;
 
 use embassy_embedded_hal::SetConfig;
-#[cfg(feature = "time")]
 use embassy_hal_internal::drop::OnDrop;
 use embassy_hal_internal::{into_ref, PeripheralRef};
 use embassy_sync::waitqueue::AtomicWaker;
 #[cfg(feature = "time")]
 use embassy_time::{Duration, Instant};
 
-use crate::dma::NoDma;
-#[cfg(feature = "time")]
-use crate::dma::Transfer;
+use crate::dma::{NoDma, Transfer};
 use crate::gpio::sealed::AFType;
 use crate::gpio::Pull;
 use crate::i2c::{Error, Instance, SclPin, SdaPin};
@@ -24,6 +19,23 @@ use crate::pac::i2c;
 use crate::time::Hertz;
 use crate::{interrupt, Peripheral};
 
+#[cfg(feature = "time")]
+fn timeout_fn(timeout: Duration) -> impl Fn() -> Result<(), Error> {
+    let deadline = Instant::now() + timeout;
+    move || {
+        if Instant::now() > deadline {
+            Err(Error::Timeout)
+        } else {
+            Ok(())
+        }
+    }
+}
+
+#[cfg(not(feature = "time"))]
+pub fn no_timeout_fn() -> impl Fn() -> Result<(), Error> {
+    move || Ok(())
+}
+
 /// Interrupt handler.
 pub struct InterruptHandler<T: Instance> {
     _phantom: PhantomData<T>,
@@ -260,21 +272,12 @@ impl<'d, T: Instance, TXDMA, RXDMA> I2c<'d, T, TXDMA, RXDMA> {
     }
 
     fn flush_txdr(&self) {
-        //if $i2c.isr.read().txis().bit_is_set() {
-        //$i2c.txdr.write(|w| w.txdata().bits(0));
-        //}
-
         if T::regs().isr().read().txis() {
             T::regs().txdr().write(|w| w.set_txdata(0));
         }
         if !T::regs().isr().read().txe() {
             T::regs().isr().modify(|w| w.set_txe(true))
         }
-
-        // If TXDR is not flagged as empty, write 1 to flush it
-        //if $i2c.isr.read().txe().is_not_empty() {
-        //$i2c.isr.write(|w| w.txe().set_bit());
-        //}
     }
 
     fn wait_txe(&self, check_timeout: impl Fn() -> Result<(), Error>) -> Result<(), Error> {
@@ -437,7 +440,6 @@ impl<'d, T: Instance, TXDMA, RXDMA> I2c<'d, T, TXDMA, RXDMA> {
         result
     }
 
-    #[cfg(feature = "time")]
     async fn write_dma_internal(
         &mut self,
         address: u8,
@@ -528,7 +530,6 @@ impl<'d, T: Instance, TXDMA, RXDMA> I2c<'d, T, TXDMA, RXDMA> {
         Ok(())
     }
 
-    #[cfg(feature = "time")]
     async fn read_dma_internal(
         &mut self,
         address: u8,
@@ -610,42 +611,38 @@ impl<'d, T: Instance, TXDMA, RXDMA> I2c<'d, T, TXDMA, RXDMA> {
 
     // =========================
     //  Async public API
-
     #[cfg(feature = "time")]
     pub async fn write(&mut self, address: u8, write: &[u8]) -> Result<(), Error>
-    where
-        TXDMA: crate::i2c::TxDma<T>,
-    {
-        self.write_timeout(address, write, self.timeout).await
-    }
-
-    #[cfg(feature = "time")]
-    pub async fn write_timeout(&mut self, address: u8, write: &[u8], timeout: Duration) -> Result<(), Error>
     where
         TXDMA: crate::i2c::TxDma<T>,
     {
         if write.is_empty() {
-            self.write_internal(address, write, true, timeout_fn(timeout))
+            self.write_internal(address, write, true, timeout_fn(self.timeout))
         } else {
             embassy_time::with_timeout(
-                timeout,
-                self.write_dma_internal(address, write, true, true, timeout_fn(timeout)),
+                self.timeout,
+                self.write_dma_internal(address, write, true, true, timeout_fn(self.timeout)),
             )
             .await
             .unwrap_or(Err(Error::Timeout))
         }
     }
 
-    #[cfg(feature = "time")]
-    pub async fn write_vectored(&mut self, address: u8, write: &[&[u8]]) -> Result<(), Error>
+    #[cfg(not(feature = "time"))]
+    pub async fn write(&mut self, address: u8, write: &[u8]) -> Result<(), Error>
     where
         TXDMA: crate::i2c::TxDma<T>,
     {
-        self.write_vectored_timeout(address, write, self.timeout).await
+        if write.is_empty() {
+            self.write_internal(address, write, true, no_timeout_fn())
+        } else {
+            self.write_dma_internal(address, write, true, true, no_timeout_fn())
+                .await
+        }
     }
 
     #[cfg(feature = "time")]
-    pub async fn write_vectored_timeout(&mut self, address: u8, write: &[&[u8]], timeout: Duration) -> Result<(), Error>
+    pub async fn write_vectored(&mut self, address: u8, write: &[&[u8]]) -> Result<(), Error>
     where
         TXDMA: crate::i2c::TxDma<T>,
     {
@@ -661,8 +658,8 @@ impl<'d, T: Instance, TXDMA, RXDMA> I2c<'d, T, TXDMA, RXDMA> {
             let is_last = next.is_none();
 
             embassy_time::with_timeout(
-                timeout,
-                self.write_dma_internal(address, c, first, is_last, timeout_fn(timeout)),
+                self.timeout,
+                self.write_dma_internal(address, c, first, is_last, timeout_fn(self.timeout)),
             )
             .await
             .unwrap_or(Err(Error::Timeout))?;
@@ -672,66 +669,79 @@ impl<'d, T: Instance, TXDMA, RXDMA> I2c<'d, T, TXDMA, RXDMA> {
         Ok(())
     }
 
+    #[cfg(not(feature = "time"))]
+    pub async fn write_vectored(&mut self, address: u8, write: &[&[u8]]) -> Result<(), Error>
+    where
+        TXDMA: crate::i2c::TxDma<T>,
+    {
+        if write.is_empty() {
+            return Err(Error::ZeroLengthTransfer);
+        }
+        let mut iter = write.iter();
+
+        let mut first = true;
+        let mut current = iter.next();
+        while let Some(c) = current {
+            let next = iter.next();
+            let is_last = next.is_none();
+
+            self.write_dma_internal(address, c, first, is_last, no_timeout_fn())
+                .await?;
+            first = false;
+            current = next;
+        }
+        Ok(())
+    }
+
     #[cfg(feature = "time")]
     pub async fn read(&mut self, address: u8, buffer: &mut [u8]) -> Result<(), Error>
     where
         RXDMA: crate::i2c::RxDma<T>,
     {
-        self.read_timeout(address, buffer, self.timeout).await
+        if buffer.is_empty() {
+            self.read_internal(address, buffer, false, timeout_fn(self.timeout))
+        } else {
+            embassy_time::with_timeout(
+                self.timeout,
+                self.read_dma_internal(address, buffer, false, timeout_fn(self.timeout)),
+            )
+            .await
+            .unwrap_or(Err(Error::Timeout))
+        }
     }
 
-    #[cfg(feature = "time")]
-    pub async fn read_timeout(&mut self, address: u8, buffer: &mut [u8], timeout: Duration) -> Result<(), Error>
+    #[cfg(not(feature = "time"))]
+    pub async fn read(&mut self, address: u8, buffer: &mut [u8]) -> Result<(), Error>
     where
         RXDMA: crate::i2c::RxDma<T>,
     {
         if buffer.is_empty() {
-            self.read_internal(address, buffer, false, timeout_fn(timeout))
+            self.read_internal(address, buffer, false, no_timeout_fn())
         } else {
-            embassy_time::with_timeout(
-                timeout,
-                self.read_dma_internal(address, buffer, false, timeout_fn(timeout)),
-            )
-            .await
-            .unwrap_or(Err(Error::Timeout))
+            self.read_dma_internal(address, buffer, false, no_timeout_fn()).await
         }
     }
 
     #[cfg(feature = "time")]
     pub async fn write_read(&mut self, address: u8, write: &[u8], read: &mut [u8]) -> Result<(), Error>
-    where
-        TXDMA: super::TxDma<T>,
-        RXDMA: super::RxDma<T>,
-    {
-        self.write_read_timeout(address, write, read, self.timeout).await
-    }
-
-    #[cfg(feature = "time")]
-    pub async fn write_read_timeout(
-        &mut self,
-        address: u8,
-        write: &[u8],
-        read: &mut [u8],
-        timeout: Duration,
-    ) -> Result<(), Error>
     where
         TXDMA: super::TxDma<T>,
         RXDMA: super::RxDma<T>,
     {
         let start_instant = Instant::now();
-        let check_timeout = timeout_fn(timeout);
+        let check_timeout = timeout_fn(self.timeout);
         if write.is_empty() {
             self.write_internal(address, write, false, &check_timeout)?;
         } else {
             embassy_time::with_timeout(
-                timeout,
+                self.timeout,
                 self.write_dma_internal(address, write, true, true, &check_timeout),
             )
             .await
             .unwrap_or(Err(Error::Timeout))?;
         }
 
-        let time_left_until_timeout = timeout - Instant::now().duration_since(start_instant);
+        let time_left_until_timeout = self.timeout - Instant::now().duration_since(start_instant);
 
         if read.is_empty() {
             self.read_internal(address, read, true, &check_timeout)?;
@@ -747,6 +757,28 @@ impl<'d, T: Instance, TXDMA, RXDMA> I2c<'d, T, TXDMA, RXDMA> {
         Ok(())
     }
 
+    #[cfg(not(feature = "time"))]
+    pub async fn write_read(&mut self, address: u8, write: &[u8], read: &mut [u8]) -> Result<(), Error>
+    where
+        TXDMA: super::TxDma<T>,
+        RXDMA: super::RxDma<T>,
+    {
+        let no_timeout = no_timeout_fn();
+        if write.is_empty() {
+            self.write_internal(address, write, false, &no_timeout)?;
+        } else {
+            self.write_dma_internal(address, write, true, true, &no_timeout).await?;
+        }
+
+        if read.is_empty() {
+            self.read_internal(address, read, true, &no_timeout)?;
+        } else {
+            self.read_dma_internal(address, read, true, &no_timeout).await?;
+        }
+
+        Ok(())
+    }
+
     // =========================
     //  Blocking public API
 
@@ -1201,15 +1233,3 @@ impl<'d, T: Instance> SetConfig for I2c<'d, T> {
         Ok(())
     }
 }
-
-#[cfg(feature = "time")]
-fn timeout_fn(timeout: Duration) -> impl Fn() -> Result<(), Error> {
-    let deadline = Instant::now() + timeout;
-    move || {
-        if Instant::now() > deadline {
-            Err(Error::Timeout)
-        } else {
-            Ok(())
-        }
-    }
-}

embassy-stm32/src/opamp.rs

@@ -31,12 +31,9 @@ impl From<OpAmpSpeed> for crate::pac::opamp::vals::OpampCsrOpahsm {
 
 /// OpAmp external outputs, wired to a GPIO pad.
 ///
-/// The GPIO output pad is held by this struct to ensure it cannot be used elsewhere.
-///
 /// This struct can also be used as an ADC input.
-pub struct OpAmpOutput<'d, 'p, T: Instance, P: OutputPin<T>> {
+pub struct OpAmpOutput<'d, T: Instance> {
     _inner: &'d OpAmp<'d, T>,
-    _output: &'p mut P,
 }
 
 /// OpAmp internal outputs, wired directly to ADC inputs.
@@ -54,19 +51,12 @@ pub struct OpAmp<'d, T: Instance> {
 impl<'d, T: Instance> OpAmp<'d, T> {
     /// Create a new driver instance.
     ///
-    /// Enables the OpAmp and configures the speed, but
-    /// does not set any other configuration.
+    /// Does not enable the opamp, but does set the speed mode on some families.
     pub fn new(opamp: impl Peripheral<P = T> + 'd, #[cfg(opamp_g4)] speed: OpAmpSpeed) -> Self {
         into_ref!(opamp);
 
-        #[cfg(opamp_f3)]
-        T::regs().opampcsr().modify(|w| {
-            w.set_opampen(true);
-        });
-
         #[cfg(opamp_g4)]
         T::regs().opamp_csr().modify(|w| {
-            w.set_opaen(true);
             w.set_opahsm(speed.into());
         });
 
@@ -74,24 +64,23 @@ impl<'d, T: Instance> OpAmp<'d, T> {
     }
 
     /// Configure the OpAmp as a buffer for the provided input pin,
-    /// outputting to the provided output pin.
+    /// outputting to the provided output pin, and enable the opamp.
     ///
     /// The input pin is configured for analogue mode but not consumed,
     /// so it may subsequently be used for ADC or comparator inputs.
     ///
     /// The output pin is held within the returned [`OpAmpOutput`] struct,
     /// preventing it being used elsewhere. The `OpAmpOutput` can then be
-    /// directly used as an ADC input.
-    pub fn buffer_ext<'a, 'b, IP, OP>(
-        &'a mut self,
-        in_pin: &IP,
-        out_pin: &'b mut OP,
+    /// directly used as an ADC input. The opamp will be disabled when the
+    /// [`OpAmpOutput`] is dropped.
+    pub fn buffer_ext(
+        &'d mut self,
+        in_pin: impl Peripheral<P = impl NonInvertingPin<T> + crate::gpio::sealed::Pin>,
+        out_pin: impl Peripheral<P = impl OutputPin<T> + crate::gpio::sealed::Pin> + 'd,
         gain: OpAmpGain,
-    ) -> OpAmpOutput<'a, 'b, T, OP>
-    where
-        IP: NonInvertingPin<T> + crate::gpio::sealed::Pin,
-        OP: OutputPin<T> + crate::gpio::sealed::Pin,
-    {
+    ) -> OpAmpOutput<'d, T> {
+        into_ref!(in_pin);
+        into_ref!(out_pin);
         in_pin.set_as_analog();
         out_pin.set_as_analog();
 
@@ -122,24 +111,24 @@ impl<'d, T: Instance> OpAmp<'d, T> {
             w.set_opaen(true);
         });
 
-        OpAmpOutput {
-            _inner: self,
-            _output: out_pin,
-        }
+        OpAmpOutput { _inner: self }
     }
 
     /// Configure the OpAmp as a buffer for the provided input pin,
-    /// with the output only used internally.
+    /// with the output only used internally, and enable the opamp.
     ///
     /// The input pin is configured for analogue mode but not consumed,
     /// so it may be subsequently used for ADC or comparator inputs.
     ///
     /// The returned `OpAmpInternalOutput` struct may be used as an ADC input.
+    /// The opamp output will be disabled when it is dropped.
     #[cfg(opamp_g4)]
-    pub fn buffer_int<'a, P>(&'a mut self, pin: &P, gain: OpAmpGain) -> OpAmpInternalOutput<'a, T>
-    where
-        P: NonInvertingPin<T> + crate::gpio::sealed::Pin,
-    {
+    pub fn buffer_int(
+        &'d mut self,
+        pin: impl Peripheral<P = impl NonInvertingPin<T> + crate::gpio::sealed::Pin>,
+        gain: OpAmpGain,
+    ) -> OpAmpInternalOutput<'d, T> {
+        into_ref!(pin);
         pin.set_as_analog();
 
         let (vm_sel, pga_gain) = match gain {
@@ -163,7 +152,21 @@ impl<'d, T: Instance> OpAmp<'d, T> {
     }
 }
 
-impl<'d, T: Instance> Drop for OpAmp<'d, T> {
+impl<'d, T: Instance> Drop for OpAmpOutput<'d, T> {
+    fn drop(&mut self) {
+        #[cfg(opamp_f3)]
+        T::regs().opampcsr().modify(|w| {
+            w.set_opampen(false);
+        });
+
+        #[cfg(opamp_g4)]
+        T::regs().opamp_csr().modify(|w| {
+            w.set_opaen(false);
+        });
+    }
+}
+
+impl<'d, T: Instance> Drop for OpAmpInternalOutput<'d, T> {
     fn drop(&mut self) {
         #[cfg(opamp_f3)]
         T::regs().opampcsr().modify(|w| {
@@ -203,16 +206,16 @@ macro_rules! impl_opamp_external_output {
     ($inst:ident, $adc:ident, $ch:expr) => {
         foreach_adc!(
             ($adc, $common_inst:ident, $adc_clock:ident) => {
-                impl<'d, 'p, P: OutputPin<crate::peripherals::$inst>> crate::adc::sealed::AdcPin<crate::peripherals::$adc>
-                    for OpAmpOutput<'d, 'p, crate::peripherals::$inst, P>
+                impl<'d> crate::adc::sealed::AdcPin<crate::peripherals::$adc>
+                    for OpAmpOutput<'d, crate::peripherals::$inst>
                 {
                     fn channel(&self) -> u8 {
                         $ch
                     }
                 }
 
-                impl<'d, 'p, P: OutputPin<crate::peripherals::$inst>> crate::adc::AdcPin<crate::peripherals::$adc>
-                    for OpAmpOutput<'d, 'p, crate::peripherals::$inst, P>
+                impl<'d> crate::adc::AdcPin<crate::peripherals::$adc>
+                    for OpAmpOutput<'d, crate::peripherals::$inst>
                 {
                 }
             };

embassy-stm32/src/rcc/h.rs

@@ -168,7 +168,12 @@ impl Default for Config {
             apb4_pre: APBPrescaler::DIV1,
 
             per_clock_source: PerClockSource::HSI,
-            adc_clock_source: AdcClockSource::from_bits(0), // PLL2_P on H7, HCLK on H5
+
+            #[cfg(stm32h5)]
+            adc_clock_source: AdcClockSource::HCLK1,
+            #[cfg(stm32h7)]
+            adc_clock_source: AdcClockSource::PER,
+
             timer_prescaler: TimerPrescaler::DefaultX2,
             voltage_scale: VoltageScale::Scale0,
             ls: Default::default(),

embassy-stm32/src/sai/mod.rs

@@ -207,27 +207,40 @@ impl Protocol {
 }
 
 #[derive(Copy, Clone, PartialEq)]
-pub enum SyncEnable {
-    Asynchronous,
+pub enum SyncInput {
+    /// Not synced to any other SAI unit.
+    None,
     /// Syncs with the other A/B sub-block within the SAI unit
     Internal,
-    /// Syncs with a sub-block in the other SAI unit - use set_sync_output() and set_sync_input()
-    #[cfg(any(sai_v4))]
-    External,
+    /// Syncs with a sub-block in the other SAI unit
+    #[cfg(sai_v4)]
+    External(SyncInputInstance),
 }
 
-impl SyncEnable {
-    #[cfg(any(sai_v1, sai_v2, sai_v3, sai_v4))]
+impl SyncInput {
     pub const fn syncen(&self) -> vals::Syncen {
         match self {
-            SyncEnable::Asynchronous => vals::Syncen::ASYNCHRONOUS,
-            SyncEnable::Internal => vals::Syncen::INTERNAL,
+            SyncInput::None => vals::Syncen::ASYNCHRONOUS,
+            SyncInput::Internal => vals::Syncen::INTERNAL,
             #[cfg(any(sai_v4))]
-            SyncEnable::External => vals::Syncen::EXTERNAL,
+            SyncInput::External(_) => vals::Syncen::EXTERNAL,
         }
     }
 }
 
+#[cfg(sai_v4)]
+#[derive(Copy, Clone, PartialEq)]
+pub enum SyncInputInstance {
+    #[cfg(peri_sai1)]
+    Sai1 = 0,
+    #[cfg(peri_sai2)]
+    Sai2 = 1,
+    #[cfg(peri_sai3)]
+    Sai3 = 2,
+    #[cfg(peri_sai4)]
+    Sai4 = 3,
+}
+
 #[derive(Copy, Clone, PartialEq)]
 pub enum StereoMono {
     Stereo,
@@ -428,8 +441,8 @@ impl MasterClockDivider {
 pub struct Config {
     pub mode: Mode,
     pub tx_rx: TxRx,
-    pub sync_enable: SyncEnable,
-    pub is_sync_output: bool,
+    pub sync_input: SyncInput,
+    pub sync_output: bool,
     pub protocol: Protocol,
     pub slot_size: SlotSize,
     pub slot_count: word::U4,
@@ -459,8 +472,8 @@ impl Default for Config {
         Self {
             mode: Mode::Master,
             tx_rx: TxRx::Transmitter,
-            is_sync_output: false,
-            sync_enable: SyncEnable::Asynchronous,
+            sync_output: false,
+            sync_input: SyncInput::None,
             protocol: Protocol::Free,
             slot_size: SlotSize::DataSize,
             slot_count: word::U4(2),
@@ -608,18 +621,18 @@ impl<'d, T: Instance> Sai<'d, T> {
 
 fn update_synchronous_config(config: &mut Config) {
     config.mode = Mode::Slave;
-    config.is_sync_output = false;
+    config.sync_output = false;
 
     #[cfg(any(sai_v1, sai_v2, sai_v3))]
     {
-        config.sync_enable = SyncEnable::Internal;
+        config.sync_input = SyncInput::Internal;
     }
 
     #[cfg(any(sai_v4))]
     {
         //this must either be Internal or External
-        //The asynchronous sub-block on the same SAI needs to enable is_sync_output
-        assert!(config.sync_enable != SyncEnable::Asynchronous);
+        //The asynchronous sub-block on the same SAI needs to enable sync_output
+        assert!(config.sync_input != SyncInput::None);
     }
 }
 
@@ -866,20 +879,13 @@ impl<'d, T: Instance, C: Channel, W: word::Word> SubBlock<'d, T, C, W> {
 
         #[cfg(any(sai_v4))]
         {
-            // Not totally clear from the datasheet if this is right
-            // This is only used if using SyncEnable::External on the other SAI unit
-            // Syncing from SAIX subblock A to subblock B does not require this
-            // Only syncing from SAI1 subblock A/B to SAI2 subblock A/B
-            let value: u8 = if T::REGS.as_ptr() == stm32_metapac::SAI1.as_ptr() {
-                1 //this is SAI1, so sync with SAI2
-            } else {
-                0 //this is SAI2, so sync with SAI1
-            };
-            T::REGS.gcr().modify(|w| {
-                w.set_syncin(value);
-            });
+            if let SyncInput::External(i) = config.sync_input {
+                T::REGS.gcr().modify(|w| {
+                    w.set_syncin(i as u8);
+                });
+            }
 
-            if config.is_sync_output {
+            if config.sync_output {
                 let syncout: u8 = match sub_block {
                     WhichSubBlock::A => 0b01,
                     WhichSubBlock::B => 0b10,
@@ -903,7 +909,7 @@ impl<'d, T: Instance, C: Channel, W: word::Word> SubBlock<'d, T, C, W> {
                 w.set_ds(config.data_size.ds());
                 w.set_lsbfirst(config.bit_order.lsbfirst());
                 w.set_ckstr(config.clock_strobe.ckstr());
-                w.set_syncen(config.sync_enable.syncen());
+                w.set_syncen(config.sync_input.syncen());
                 w.set_mono(config.stereo_mono.mono());
                 w.set_outdriv(config.output_drive.outdriv());
                 w.set_mckdiv(config.master_clock_divider.mckdiv());

embassy-sync/README.md

@@ -5,6 +5,7 @@ An [Embassy](https://embassy.dev) project.
 Synchronization primitives and data structures with async support:
 
 - [`Channel`](channel::Channel) - A Multiple Producer Multiple Consumer (MPMC) channel. Each message is only received by a single consumer.
+- [`PriorityChannel`](channel::priority::PriorityChannel) - A Multiple Producer Multiple Consumer (MPMC) channel. Each message is only received by a single consumer. Higher priority items are sifted to the front of the channel.
 - [`PubSubChannel`](pubsub::PubSubChannel) - A broadcast channel (publish-subscribe) channel. Each message is received by all consumers.
 - [`Signal`](signal::Signal) - Signalling latest value to a single consumer.
 - [`Mutex`](mutex::Mutex) - Mutex for synchronizing state between asynchronous tasks.

embassy-sync/src/channel.rs

@@ -76,7 +76,7 @@ where
 
 /// Send-only access to a [`Channel`] without knowing channel size.
 pub struct DynamicSender<'ch, T> {
-    channel: &'ch dyn DynamicChannel<T>,
+    pub(crate) channel: &'ch dyn DynamicChannel<T>,
 }
 
 impl<'ch, T> Clone for DynamicSender<'ch, T> {
@@ -176,7 +176,7 @@ where
 
 /// Receive-only access to a [`Channel`] without knowing channel size.
 pub struct DynamicReceiver<'ch, T> {
-    channel: &'ch dyn DynamicChannel<T>,
+    pub(crate) channel: &'ch dyn DynamicChannel<T>,
 }
 
 impl<'ch, T> Clone for DynamicReceiver<'ch, T> {
@@ -321,7 +321,7 @@ impl<'ch, T> Future for DynamicSendFuture<'ch, T> {
 
 impl<'ch, T> Unpin for DynamicSendFuture<'ch, T> {}
 
-trait DynamicChannel<T> {
+pub(crate) trait DynamicChannel<T> {
     fn try_send_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>>;
 
     fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError>;

embassy-sync/src/lib.rs

@@ -15,6 +15,7 @@ pub mod blocking_mutex;
 pub mod channel;
 pub mod mutex;
 pub mod pipe;
+pub mod priority_channel;
 pub mod pubsub;
 pub mod signal;
 pub mod waitqueue;

embassy-sync/src/priority_channel.rs

@@ -0,0 +1,613 @@
+//! A queue for sending values between asynchronous tasks.
+//!
+//! Similar to a [`Channel`](crate::channel::Channel), however [`PriorityChannel`] sifts higher priority items to the front of the queue.
+//! Priority is determined by the `Ord` trait. Priority behavior is determined by the [`Kind`](heapless::binary_heap::Kind) parameter of the channel.
+
+use core::cell::RefCell;
+use core::future::Future;
+use core::pin::Pin;
+use core::task::{Context, Poll};
+
+pub use heapless::binary_heap::{Kind, Max, Min};
+use heapless::BinaryHeap;
+
+use crate::blocking_mutex::raw::RawMutex;
+use crate::blocking_mutex::Mutex;
+use crate::channel::{DynamicChannel, DynamicReceiver, DynamicSender, TryReceiveError, TrySendError};
+use crate::waitqueue::WakerRegistration;
+
+/// Send-only access to a [`PriorityChannel`].
+pub struct Sender<'ch, M, T, K, const N: usize>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    channel: &'ch PriorityChannel<M, T, K, N>,
+}
+
+impl<'ch, M, T, K, const N: usize> Clone for Sender<'ch, M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    fn clone(&self) -> Self {
+        Sender { channel: self.channel }
+    }
+}
+
+impl<'ch, M, T, K, const N: usize> Copy for Sender<'ch, M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+}
+
+impl<'ch, M, T, K, const N: usize> Sender<'ch, M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    /// Sends a value.
+    ///
+    /// See [`PriorityChannel::send()`]
+    pub fn send(&self, message: T) -> SendFuture<'ch, M, T, K, N> {
+        self.channel.send(message)
+    }
+
+    /// Attempt to immediately send a message.
+    ///
+    /// See [`PriorityChannel::send()`]
+    pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
+        self.channel.try_send(message)
+    }
+
+    /// Allows a poll_fn to poll until the channel is ready to send
+    ///
+    /// See [`PriorityChannel::poll_ready_to_send()`]
+    pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
+        self.channel.poll_ready_to_send(cx)
+    }
+}
+
+impl<'ch, M, T, K, const N: usize> From<Sender<'ch, M, T, K, N>> for DynamicSender<'ch, T>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    fn from(s: Sender<'ch, M, T, K, N>) -> Self {
+        Self { channel: s.channel }
+    }
+}
+
+/// Receive-only access to a [`PriorityChannel`].
+pub struct Receiver<'ch, M, T, K, const N: usize>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    channel: &'ch PriorityChannel<M, T, K, N>,
+}
+
+impl<'ch, M, T, K, const N: usize> Clone for Receiver<'ch, M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    fn clone(&self) -> Self {
+        Receiver { channel: self.channel }
+    }
+}
+
+impl<'ch, M, T, K, const N: usize> Copy for Receiver<'ch, M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+}
+
+impl<'ch, M, T, K, const N: usize> Receiver<'ch, M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    /// Receive the next value.
+    ///
+    /// See [`PriorityChannel::receive()`].
+    pub fn receive(&self) -> ReceiveFuture<'_, M, T, K, N> {
+        self.channel.receive()
+    }
+
+    /// Attempt to immediately receive the next value.
+    ///
+    /// See [`PriorityChannel::try_receive()`]
+    pub fn try_receive(&self) -> Result<T, TryReceiveError> {
+        self.channel.try_receive()
+    }
+
+    /// Allows a poll_fn to poll until the channel is ready to receive
+    ///
+    /// See [`PriorityChannel::poll_ready_to_receive()`]
+    pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
+        self.channel.poll_ready_to_receive(cx)
+    }
+
+    /// Poll the channel for the next item
+    ///
+    /// See [`PriorityChannel::poll_receive()`]
+    pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
+        self.channel.poll_receive(cx)
+    }
+}
+
+impl<'ch, M, T, K, const N: usize> From<Receiver<'ch, M, T, K, N>> for DynamicReceiver<'ch, T>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    fn from(s: Receiver<'ch, M, T, K, N>) -> Self {
+        Self { channel: s.channel }
+    }
+}
+
+/// Future returned by [`PriorityChannel::receive`] and  [`Receiver::receive`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReceiveFuture<'ch, M, T, K, const N: usize>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    channel: &'ch PriorityChannel<M, T, K, N>,
+}
+
+impl<'ch, M, T, K, const N: usize> Future for ReceiveFuture<'ch, M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    type Output = T;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
+        self.channel.poll_receive(cx)
+    }
+}
+
+/// Future returned by [`PriorityChannel::send`] and  [`Sender::send`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct SendFuture<'ch, M, T, K, const N: usize>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    channel: &'ch PriorityChannel<M, T, K, N>,
+    message: Option<T>,
+}
+
+impl<'ch, M, T, K, const N: usize> Future for SendFuture<'ch, M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match self.message.take() {
+            Some(m) => match self.channel.try_send_with_context(m, Some(cx)) {
+                Ok(..) => Poll::Ready(()),
+                Err(TrySendError::Full(m)) => {
+                    self.message = Some(m);
+                    Poll::Pending
+                }
+            },
+            None => panic!("Message cannot be None"),
+        }
+    }
+}
+
+impl<'ch, M, T, K, const N: usize> Unpin for SendFuture<'ch, M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+}
+
+struct ChannelState<T, K, const N: usize> {
+    queue: BinaryHeap<T, K, N>,
+    receiver_waker: WakerRegistration,
+    senders_waker: WakerRegistration,
+}
+
+impl<T, K, const N: usize> ChannelState<T, K, N>
+where
+    T: Ord,
+    K: Kind,
+{
+    const fn new() -> Self {
+        ChannelState {
+            queue: BinaryHeap::new(),
+            receiver_waker: WakerRegistration::new(),
+            senders_waker: WakerRegistration::new(),
+        }
+    }
+
+    fn try_receive(&mut self) -> Result<T, TryReceiveError> {
+        self.try_receive_with_context(None)
+    }
+
+    fn try_receive_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
+        if self.queue.len() == self.queue.capacity() {
+            self.senders_waker.wake();
+        }
+
+        if let Some(message) = self.queue.pop() {
+            Ok(message)
+        } else {
+            if let Some(cx) = cx {
+                self.receiver_waker.register(cx.waker());
+            }
+            Err(TryReceiveError::Empty)
+        }
+    }
+
+    fn poll_receive(&mut self, cx: &mut Context<'_>) -> Poll<T> {
+        if self.queue.len() == self.queue.capacity() {
+            self.senders_waker.wake();
+        }
+
+        if let Some(message) = self.queue.pop() {
+            Poll::Ready(message)
+        } else {
+            self.receiver_waker.register(cx.waker());
+            Poll::Pending
+        }
+    }
+
+    fn poll_ready_to_receive(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        self.receiver_waker.register(cx.waker());
+
+        if !self.queue.is_empty() {
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
+    }
+
+    fn try_send(&mut self, message: T) -> Result<(), TrySendError<T>> {
+        self.try_send_with_context(message, None)
+    }
+
+    fn try_send_with_context(&mut self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
+        match self.queue.push(message) {
+            Ok(()) => {
+                self.receiver_waker.wake();
+                Ok(())
+            }
+            Err(message) => {
+                if let Some(cx) = cx {
+                    self.senders_waker.register(cx.waker());
+                }
+                Err(TrySendError::Full(message))
+            }
+        }
+    }
+
+    fn poll_ready_to_send(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        self.senders_waker.register(cx.waker());
+
+        if !self.queue.len() == self.queue.capacity() {
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
+    }
+}
+
+/// A bounded channel for communicating between asynchronous tasks
+/// with backpressure.
+///
+/// The channel will buffer up to the provided number of messages.  Once the
+/// buffer is full, attempts to `send` new messages will wait until a message is
+/// received from the channel.
+///
+/// Sent data may be reordered based on their priorty within the channel.
+/// For example, in a [`Max`](heapless::binary_heap::Max) [`PriorityChannel`]
+/// containing `u32`'s, data sent in the following order `[1, 2, 3]` will be recieved as `[3, 2, 1]`.
+pub struct PriorityChannel<M, T, K, const N: usize>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    inner: Mutex<M, RefCell<ChannelState<T, K, N>>>,
+}
+
+impl<M, T, K, const N: usize> PriorityChannel<M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    /// Establish a new bounded channel. For example, to create one with a NoopMutex:
+    ///
+    /// ```
+    /// use embassy_sync::priority_channel::{PriorityChannel, Max};
+    /// use embassy_sync::blocking_mutex::raw::NoopRawMutex;
+    ///
+    /// // Declare a bounded channel of 3 u32s.
+    /// let mut channel = PriorityChannel::<NoopRawMutex, u32, Max, 3>::new();
+    /// ```
+    pub const fn new() -> Self {
+        Self {
+            inner: Mutex::new(RefCell::new(ChannelState::new())),
+        }
+    }
+
+    fn lock<R>(&self, f: impl FnOnce(&mut ChannelState<T, K, N>) -> R) -> R {
+        self.inner.lock(|rc| f(&mut *unwrap!(rc.try_borrow_mut())))
+    }
+
+    fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
+        self.lock(|c| c.try_receive_with_context(cx))
+    }
+
+    /// Poll the channel for the next message
+    pub fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
+        self.lock(|c| c.poll_receive(cx))
+    }
+
+    fn try_send_with_context(&self, m: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
+        self.lock(|c| c.try_send_with_context(m, cx))
+    }
+
+    /// Allows a poll_fn to poll until the channel is ready to receive
+    pub fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
+        self.lock(|c| c.poll_ready_to_receive(cx))
+    }
+
+    /// Allows a poll_fn to poll until the channel is ready to send
+    pub fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
+        self.lock(|c| c.poll_ready_to_send(cx))
+    }
+
+    /// Get a sender for this channel.
+    pub fn sender(&self) -> Sender<'_, M, T, K, N> {
+        Sender { channel: self }
+    }
+
+    /// Get a receiver for this channel.
+    pub fn receiver(&self) -> Receiver<'_, M, T, K, N> {
+        Receiver { channel: self }
+    }
+
+    /// Send a value, waiting until there is capacity.
+    ///
+    /// Sending completes when the value has been pushed to the channel's queue.
+    /// This doesn't mean the value has been received yet.
+    pub fn send(&self, message: T) -> SendFuture<'_, M, T, K, N> {
+        SendFuture {
+            channel: self,
+            message: Some(message),
+        }
+    }
+
+    /// Attempt to immediately send a message.
+    ///
+    /// This method differs from [`send`](PriorityChannel::send) by returning immediately if the channel's
+    /// buffer is full, instead of waiting.
+    ///
+    /// # Errors
+    ///
+    /// If the channel capacity has been reached, i.e., the channel has `n`
+    /// buffered values where `n` is the argument passed to [`PriorityChannel`], then an
+    /// error is returned.
+    pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
+        self.lock(|c| c.try_send(message))
+    }
+
+    /// Receive the next value.
+    ///
+    /// If there are no messages in the channel's buffer, this method will
+    /// wait until a message is sent.
+    pub fn receive(&self) -> ReceiveFuture<'_, M, T, K, N> {
+        ReceiveFuture { channel: self }
+    }
+
+    /// Attempt to immediately receive a message.
+    ///
+    /// This method will either receive a message from the channel immediately or return an error
+    /// if the channel is empty.
+    pub fn try_receive(&self) -> Result<T, TryReceiveError> {
+        self.lock(|c| c.try_receive())
+    }
+}
+
+/// Implements the DynamicChannel to allow creating types that are unaware of the queue size with the
+/// tradeoff cost of dynamic dispatch.
+impl<M, T, K, const N: usize> DynamicChannel<T> for PriorityChannel<M, T, K, N>
+where
+    T: Ord,
+    K: Kind,
+    M: RawMutex,
+{
+    fn try_send_with_context(&self, m: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
+        PriorityChannel::try_send_with_context(self, m, cx)
+    }
+
+    fn try_receive_with_context(&self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
+        PriorityChannel::try_receive_with_context(self, cx)
+    }
+
+    fn poll_ready_to_send(&self, cx: &mut Context<'_>) -> Poll<()> {
+        PriorityChannel::poll_ready_to_send(self, cx)
+    }
+
+    fn poll_ready_to_receive(&self, cx: &mut Context<'_>) -> Poll<()> {
+        PriorityChannel::poll_ready_to_receive(self, cx)
+    }
+
+    fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<T> {
+        PriorityChannel::poll_receive(self, cx)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use core::time::Duration;
+
+    use futures_executor::ThreadPool;
+    use futures_timer::Delay;
+    use futures_util::task::SpawnExt;
+    use heapless::binary_heap::{Kind, Max};
+    use static_cell::StaticCell;
+
+    use super::*;
+    use crate::blocking_mutex::raw::{CriticalSectionRawMutex, NoopRawMutex};
+
+    fn capacity<T, K, const N: usize>(c: &ChannelState<T, K, N>) -> usize
+    where
+        T: Ord,
+        K: Kind,
+    {
+        c.queue.capacity() - c.queue.len()
+    }
+
+    #[test]
+    fn sending_once() {
+        let mut c = ChannelState::<u32, Max, 3>::new();
+        assert!(c.try_send(1).is_ok());
+        assert_eq!(capacity(&c), 2);
+    }
+
+    #[test]
+    fn sending_when_full() {
+        let mut c = ChannelState::<u32, Max, 3>::new();
+        let _ = c.try_send(1);
+        let _ = c.try_send(1);
+        let _ = c.try_send(1);
+        match c.try_send(2) {
+            Err(TrySendError::Full(2)) => assert!(true),
+            _ => assert!(false),
+        }
+        assert_eq!(capacity(&c), 0);
+    }
+
+    #[test]
+    fn send_priority() {
+        // Prio channel with kind `Max` sifts larger numbers to the front of the queue
+        let mut c = ChannelState::<u32, Max, 3>::new();
+        assert!(c.try_send(1).is_ok());
+        assert!(c.try_send(2).is_ok());
+        assert!(c.try_send(3).is_ok());
+        assert_eq!(c.try_receive().unwrap(), 3);
+        assert_eq!(c.try_receive().unwrap(), 2);
+        assert_eq!(c.try_receive().unwrap(), 1);
+    }
+
+    #[test]
+    fn receiving_once_with_one_send() {
+        let mut c = ChannelState::<u32, Max, 3>::new();
+        assert!(c.try_send(1).is_ok());
+        assert_eq!(c.try_receive().unwrap(), 1);
+        assert_eq!(capacity(&c), 3);
+    }
+
+    #[test]
+    fn receiving_when_empty() {
+        let mut c = ChannelState::<u32, Max, 3>::new();
+        match c.try_receive() {
+            Err(TryReceiveError::Empty) => assert!(true),
+            _ => assert!(false),
+        }
+        assert_eq!(capacity(&c), 3);
+    }
+
+    #[test]
+    fn simple_send_and_receive() {
+        let c = PriorityChannel::<NoopRawMutex, u32, Max, 3>::new();
+        assert!(c.try_send(1).is_ok());
+        assert_eq!(c.try_receive().unwrap(), 1);
+    }
+
+    #[test]
+    fn cloning() {
+        let c = PriorityChannel::<NoopRawMutex, u32, Max, 3>::new();
+        let r1 = c.receiver();
+        let s1 = c.sender();
+
+        let _ = r1.clone();
+        let _ = s1.clone();
+    }
+
+    #[test]
+    fn dynamic_dispatch() {
+        let c = PriorityChannel::<NoopRawMutex, u32, Max, 3>::new();
+        let s: DynamicSender<'_, u32> = c.sender().into();
+        let r: DynamicReceiver<'_, u32> = c.receiver().into();
+
+        assert!(s.try_send(1).is_ok());
+        assert_eq!(r.try_receive().unwrap(), 1);
+    }
+
+    #[futures_test::test]
+    async fn receiver_receives_given_try_send_async() {
+        let executor = ThreadPool::new().unwrap();
+
+        static CHANNEL: StaticCell<PriorityChannel<CriticalSectionRawMutex, u32, Max, 3>> = StaticCell::new();
+        let c = &*CHANNEL.init(PriorityChannel::new());
+        let c2 = c;
+        assert!(executor
+            .spawn(async move {
+                assert!(c2.try_send(1).is_ok());
+            })
+            .is_ok());
+        assert_eq!(c.receive().await, 1);
+    }
+
+    #[futures_test::test]
+    async fn sender_send_completes_if_capacity() {
+        let c = PriorityChannel::<CriticalSectionRawMutex, u32, Max, 1>::new();
+        c.send(1).await;
+        assert_eq!(c.receive().await, 1);
+    }
+
+    #[futures_test::test]
+    async fn senders_sends_wait_until_capacity() {
+        let executor = ThreadPool::new().unwrap();
+
+        static CHANNEL: StaticCell<PriorityChannel<CriticalSectionRawMutex, u32, Max, 1>> = StaticCell::new();
+        let c = &*CHANNEL.init(PriorityChannel::new());
+        assert!(c.try_send(1).is_ok());
+
+        let c2 = c;
+        let send_task_1 = executor.spawn_with_handle(async move { c2.send(2).await });
+        let c2 = c;
+        let send_task_2 = executor.spawn_with_handle(async move { c2.send(3).await });
+        // Wish I could think of a means of determining that the async send is waiting instead.
+        // However, I've used the debugger to observe that the send does indeed wait.
+        Delay::new(Duration::from_millis(500)).await;
+        assert_eq!(c.receive().await, 1);
+        assert!(executor
+            .spawn(async move {
+                loop {
+                    c.receive().await;
+                }
+            })
+            .is_ok());
+        send_task_1.unwrap().await;
+        send_task_2.unwrap().await;
+    }
+}

embassy-time/Cargo.toml

@@ -258,4 +258,4 @@ wasm-timer = { version = "0.2.5", optional = true }
 [dev-dependencies]
 serial_test = "0.9"
 critical-section = { version = "1.1", features = ["std"] }
-embassy-executor = { version = "0.3.1", path = "../embassy-executor", features = ["nightly"] }
+embassy-executor = { version = "0.3.3", path = "../embassy-executor", features = ["nightly"] }

examples/boot/application/nrf/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers", "arch-cortex-m", "executor-thread"] }
+embassy-executor = { version = "0.3.3", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers", "arch-cortex-m", "executor-thread"] }
 embassy-time = { version = "0.1.5", path = "../../../../embassy-time", features = ["nightly"] }
 embassy-nrf = { version = "0.1.0", path = "../../../../embassy-nrf", features = ["time-driver-rtc1", "gpiote", "nightly"] }
 embassy-boot = { version = "0.1.0", path = "../../../../embassy-boot/boot", features = ["nightly"] }

examples/boot/application/rp/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers", "arch-cortex-m", "executor-thread"] }
+embassy-executor = { version = "0.3.3", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers", "arch-cortex-m", "executor-thread"] }
 embassy-time = { version = "0.1.5", path = "../../../../embassy-time", features = ["nightly"] }
 embassy-rp = { version = "0.1.0", path = "../../../../embassy-rp", features = ["time-driver", "unstable-traits", "nightly"] }
 embassy-boot-rp = { version = "0.1.0", path = "../../../../embassy-boot/rp", features = ["nightly"] }

examples/boot/application/rp/memory.x

@@ -5,7 +5,19 @@ MEMORY
   BOOTLOADER_STATE                  : ORIGIN = 0x10006000, LENGTH = 4K
   FLASH                             : ORIGIN = 0x10007000, LENGTH = 512K
   DFU                               : ORIGIN = 0x10087000, LENGTH = 516K
-  RAM                               : ORIGIN = 0x20000000, LENGTH = 256K
+
+  /* Pick one of the two options for RAM layout     */
+
+  /* OPTION A: Use all RAM banks as one big block   */
+  /* Reasonable, unless you are doing something     */
+  /* really particular with DMA or other concurrent */
+  /* access that would benefit from striping        */
+  RAM   : ORIGIN = 0x20000000, LENGTH = 264K
+
+  /* OPTION B: Keep the unstriped sections separate */
+  /* RAM: ORIGIN = 0x20000000, LENGTH = 256K        */
+  /* SCRATCH_A: ORIGIN = 0x20040000, LENGTH = 4K    */
+  /* SCRATCH_B: ORIGIN = 0x20041000, LENGTH = 4K    */
 }
 
 __bootloader_state_start = ORIGIN(BOOTLOADER_STATE) - ORIGIN(BOOT2);

examples/boot/application/stm32f3/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../../../embassy-time", features = ["nightly", "tick-hz-32_768"] }
 embassy-stm32 = { version = "0.1.0", path = "../../../../embassy-stm32", features = ["unstable-traits", "nightly", "stm32f303re", "time-driver-any", "exti"]  }
 embassy-boot-stm32 = { version = "0.1.0", path = "../../../../embassy-boot/stm32", features = ["nightly"] }

examples/boot/application/stm32f7/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../../../embassy-time", features = ["nightly", "tick-hz-32_768"] }
 embassy-stm32 = { version = "0.1.0", path = "../../../../embassy-stm32", features = ["unstable-traits", "nightly", "stm32f767zi", "time-driver-any", "exti"]  }
 embassy-boot-stm32 = { version = "0.1.0", path = "../../../../embassy-boot/stm32", features = ["nightly"] }

examples/boot/application/stm32h7/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../../../embassy-time", features = ["nightly", "tick-hz-32_768"] }
 embassy-stm32 = { version = "0.1.0", path = "../../../../embassy-stm32", features = ["unstable-traits", "nightly", "stm32h743zi", "time-driver-any", "exti"]  }
 embassy-boot-stm32 = { version = "0.1.0", path = "../../../../embassy-boot/stm32", features = ["nightly"] }

examples/boot/application/stm32l0/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../../../embassy-time", features = ["nightly", "tick-hz-32_768"] }
 embassy-stm32 = { version = "0.1.0", path = "../../../../embassy-stm32", features = ["unstable-traits", "nightly", "stm32l072cz", "time-driver-any", "exti", "memory-x"]  }
 embassy-boot-stm32 = { version = "0.1.0", path = "../../../../embassy-boot/stm32", features = ["nightly"] }

examples/boot/application/stm32l1/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../../../embassy-time", features = ["nightly", "tick-hz-32_768"] }
 embassy-stm32 = { version = "0.1.0", path = "../../../../embassy-stm32", features = ["unstable-traits", "nightly", "stm32l151cb-a", "time-driver-any", "exti"]  }
 embassy-boot-stm32 = { version = "0.1.0", path = "../../../../embassy-boot/stm32", features = ["nightly"] }

examples/boot/application/stm32l4/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../../../embassy-time", features = ["nightly", "tick-hz-32_768"] }
 embassy-stm32 = { version = "0.1.0", path = "../../../../embassy-stm32", features = ["unstable-traits", "nightly", "stm32l475vg", "time-driver-any", "exti"]  }
 embassy-boot-stm32 = { version = "0.1.0", path = "../../../../embassy-boot/stm32", features = ["nightly"] }

examples/boot/application/stm32wl/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../../../embassy-time", features = ["nightly", "tick-hz-32_768"] }
 embassy-stm32 = { version = "0.1.0", path = "../../../../embassy-stm32", features = ["unstable-traits", "nightly", "stm32wl55jc-cm4", "time-driver-any", "exti"]  }
 embassy-boot-stm32 = { version = "0.1.0", path = "../../../../embassy-boot/stm32", features = ["nightly"] }

examples/boot/bootloader/rp/memory.x

@@ -6,7 +6,19 @@ MEMORY
   BOOTLOADER_STATE                  : ORIGIN = 0x10006000, LENGTH = 4K
   ACTIVE                            : ORIGIN = 0x10007000, LENGTH = 512K
   DFU                               : ORIGIN = 0x10087000, LENGTH = 516K
-  RAM                               : ORIGIN = 0x20000000, LENGTH = 256K
+
+  /* Pick one of the two options for RAM layout     */
+
+  /* OPTION A: Use all RAM banks as one big block   */
+  /* Reasonable, unless you are doing something     */
+  /* really particular with DMA or other concurrent */
+  /* access that would benefit from striping        */
+  RAM   : ORIGIN = 0x20000000, LENGTH = 264K
+
+  /* OPTION B: Keep the unstriped sections separate */
+  /* RAM: ORIGIN = 0x20000000, LENGTH = 256K        */
+  /* SCRATCH_A: ORIGIN = 0x20040000, LENGTH = 4K    */
+  /* SCRATCH_B: ORIGIN = 0x20041000, LENGTH = 4K    */
 }
 
 __bootloader_state_start = ORIGIN(BOOTLOADER_STATE) - ORIGIN(BOOT2);

examples/nrf-rtos-trace/Cargo.toml

@@ -17,7 +17,7 @@ log = [
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "rtos-trace", "rtos-trace-interrupt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "rtos-trace", "rtos-trace-interrupt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time" }
 embassy-nrf = { version = "0.1.0", path = "../../embassy-nrf", features = ["nrf52840", "time-driver-rtc1", "gpiote", "unstable-pac"] }
 

examples/nrf52840/Cargo.toml

@@ -30,7 +30,7 @@ nightly = [
 [dependencies]
 embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime"] }
 embassy-nrf = { version = "0.1.0", path = "../../embassy-nrf", features = ["defmt", "nrf52840", "time-driver-rtc1", "gpiote", "unstable-pac", "time"] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features = ["defmt", "tcp", "dhcpv4", "medium-ethernet"], optional = true }

examples/nrf5340/Cargo.toml

@@ -9,7 +9,7 @@ embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = [
     "defmt",
 ] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", 
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", 
     "nightly",
     "defmt",
     "integrated-timers",

examples/rp/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 [dependencies]
 embassy-embedded-hal = { version = "0.1.0", path = "../../embassy-embedded-hal", features = ["defmt"] }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["nightly", "unstable-traits", "defmt", "defmt-timestamp-uptime"] }
 embassy-rp = { version = "0.1.0", path = "../../embassy-rp", features = ["defmt", "unstable-traits", "nightly", "unstable-pac", "time-driver", "critical-section-impl"] }
 embassy-usb = { version = "0.1.0", path = "../../embassy-usb", features = ["defmt"] }

examples/rp/memory.x

@@ -1,5 +1,17 @@
 MEMORY {
     BOOT2 : ORIGIN = 0x10000000, LENGTH = 0x100
     FLASH : ORIGIN = 0x10000100, LENGTH = 2048K - 0x100
-    RAM   : ORIGIN = 0x20000000, LENGTH = 256K
-}
+
+    /* Pick one of the two options for RAM layout     */
+
+    /* OPTION A: Use all RAM banks as one big block   */
+    /* Reasonable, unless you are doing something     */
+    /* really particular with DMA or other concurrent */
+    /* access that would benefit from striping        */
+    RAM   : ORIGIN = 0x20000000, LENGTH = 264K
+
+    /* OPTION B: Keep the unstriped sections separate */
+    /* RAM: ORIGIN = 0x20000000, LENGTH = 256K        */
+    /* SCRATCH_A: ORIGIN = 0x20040000, LENGTH = 4K    */
+    /* SCRATCH_B: ORIGIN = 0x20041000, LENGTH = 4K    */
+}

examples/std/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["log"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["arch-std", "executor-thread", "log", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["arch-std", "executor-thread", "log", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["log", "std", "nightly"] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features=[ "std", "nightly", "log", "medium-ethernet", "medium-ip", "tcp", "udp", "dns", "dhcpv4", "proto-ipv6"] }
 embassy-net-tuntap = { version = "0.1.0", path = "../../embassy-net-tuntap" }

examples/stm32c0/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32c031c6 to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "time-driver-any", "stm32c031c6", "memory-x", "unstable-pac", "exti"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 
 defmt = "0.3"

examples/stm32f0/Cargo.toml

@@ -15,7 +15,7 @@ defmt = "0.3"
 defmt-rtt = "0.4"
 panic-probe = "0.3"
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 static_cell = { version = "2", features = ["nightly"]}
 portable-atomic = { version = "1.5", features = ["unsafe-assume-single-core"] }

examples/stm32f1/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32f103c8 to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32f103c8", "unstable-pac", "memory-x", "time-driver-any", "unstable-traits" ]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-usb = { version = "0.1.0", path = "../../embassy-usb", features = ["defmt"] }
 embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }

examples/stm32f2/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32f207zg to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32f207zg", "unstable-pac", "memory-x", "time-driver-any", "exti"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 
 defmt = "0.3"

examples/stm32f3/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32f303ze to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32f303ze", "unstable-pac", "memory-x", "time-driver-any", "exti"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-usb = { version = "0.1.0", path = "../../embassy-usb", features = ["defmt"] }
 embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }

examples/stm32f334/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.0", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32f334r8", "unstable-pac", "memory-x", "time-driver-any", "exti"]  }
 embassy-usb = { version = "0.1.0", path = "../../embassy-usb", features = ["defmt"] }

examples/stm32f334/src/bin/opamp.rs

@@ -39,7 +39,7 @@ async fn main(_spawner: Spawner) -> ! {
 
     let mut vrefint = adc.enable_vref(&mut Delay);
     let mut temperature = adc.enable_temperature();
-    let mut buffer = opamp.buffer_ext(&p.PA7, &mut p.PA6, OpAmpGain::Mul1);
+    let mut buffer = opamp.buffer_ext(&mut p.PA7, &mut p.PA6, OpAmpGain::Mul1);
 
     loop {
         let vref = adc.read(&mut vrefint).await;

examples/stm32f4/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32f429zi to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "unstable-traits", "defmt", "stm32f429zi", "unstable-pac", "memory-x", "time-driver-any", "exti", "embedded-sdmmc", "chrono"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "executor-interrupt", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "unstable-traits", "tick-hz-32_768"] }
 embassy-usb = { version = "0.1.0", path = "../../embassy-usb", features = ["defmt" ] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features = ["defmt", "tcp", "dhcpv4", "medium-ethernet", "nightly"] }

examples/stm32f7/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32f767zi to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32f767zi", "memory-x", "unstable-pac", "time-driver-any", "exti"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features = ["defmt", "nightly", "tcp", "dhcpv4", "medium-ethernet"] }
 embedded-io-async = { version = "0.6.0" }

examples/stm32g0/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32g071rb to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "time-driver-any", "stm32g071rb", "memory-x", "unstable-pac", "exti"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 
 defmt = "0.3"

examples/stm32g4/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32g491re to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "time-driver-any", "stm32g491re", "memory-x", "unstable-pac", "exti"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-usb = { version = "0.1.0", path = "../../embassy-usb", features = ["defmt"] }
 embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }

examples/stm32h5/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32h563zi to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32h563zi", "memory-x", "time-driver-any", "exti", "unstable-pac", "unstable-traits"] }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "unstable-traits", "tick-hz-32_768"] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features = ["defmt", "nightly", "tcp", "dhcpv4", "medium-ethernet", "proto-ipv6"] }
 embedded-io-async = { version = "0.6.0" }

examples/stm32h7/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32h743bi to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32h743bi", "time-driver-any", "exti", "memory-x", "unstable-pac", "unstable-traits", "chrono"] }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "unstable-traits", "tick-hz-32_768"] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features = ["defmt", "nightly", "tcp", "dhcpv4", "medium-ethernet", "proto-ipv6", "dns"] }
 embedded-io-async = { version = "0.6.0" }

examples/stm32l0/Cargo.toml

@@ -13,7 +13,7 @@ nightly = ["embassy-stm32/nightly", "embassy-time/nightly", "embassy-time/unstab
 # Change stm32l072cz to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["defmt", "stm32l072cz", "time-driver-any", "exti", "unstable-traits", "memory-x"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-lora = { version = "0.1.0", path = "../../embassy-lora", features = ["time", "defmt"], optional = true }
 lora-phy = { version = "2", optional = true }

examples/stm32l1/Cargo.toml

@@ -6,7 +6,7 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32l151cb-a", "time-driver-any", "memory-x"]  }
 

examples/stm32l4/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32l4s5vi to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "unstable-pac", "stm32l4s5qi", "memory-x", "time-driver-any", "exti", "unstable-traits", "chrono"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768", "unstable-traits", "nightly"] }
 embassy-embedded-hal = { version = "0.1.0", path = "../../embassy-embedded-hal" }
 embassy-usb = { version = "0.1.0", path = "../../embassy-usb", features = ["defmt"] }

examples/stm32l5/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32l552ze to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "unstable-pac", "stm32l552ze", "time-driver-any", "exti", "unstable-traits", "memory-x"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-usb = { version = "0.1.0", path = "../../embassy-usb", features = ["defmt"] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features = ["defmt", "nightly", "tcp", "dhcpv4", "medium-ethernet"] }

examples/stm32u5/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32u585ai to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "unstable-pac", "stm32u585ai", "time-driver-any", "memory-x" ]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-usb = { version = "0.1.0", path = "../../embassy-usb", features = ["defmt"] }
 

examples/stm32wb/Cargo.toml

@@ -9,7 +9,7 @@ license = "MIT OR Apache-2.0"
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32wb55rg", "time-driver-any", "memory-x", "exti"]  }
 embassy-stm32-wpan = { version = "0.1.0", path = "../../embassy-stm32-wpan", features = ["defmt", "stm32wb55rg"] }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features = ["defmt", "udp", "proto-ipv6", "medium-ieee802154", "nightly"], optional=true }
 

examples/stm32wba/Cargo.toml

@@ -7,7 +7,7 @@ license = "MIT OR Apache-2.0"
 [dependencies]
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32wba52cg", "time-driver-any", "memory-x", "exti"]  }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features = ["defmt", "udp", "proto-ipv6", "medium-ieee802154", "nightly"], optional=true }
 

examples/stm32wl/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 # Change stm32wl55jc-cm4 to your chip name, if necessary.
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "unstable-traits", "defmt", "stm32wl55jc-cm4", "time-driver-any", "memory-x", "unstable-pac", "exti", "chrono"] }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["nightly", "unstable-traits", "defmt", "defmt-timestamp-uptime", "tick-hz-32_768"] }
 embassy-embedded-hal = { version = "0.1.0", path = "../../embassy-embedded-hal" }
 embassy-lora = { version = "0.1.0", path = "../../embassy-lora", features = ["stm32wl", "time", "defmt"] }

examples/wasm/Cargo.toml

@@ -9,7 +9,7 @@ crate-type = ["cdylib"]
 
 [dependencies]
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["log"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["arch-wasm", "executor-thread", "log", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["arch-wasm", "executor-thread", "log", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["log", "wasm", "nightly"] }
 
 wasm-logger = "0.2.0"

tests/nrf/Cargo.toml

@@ -9,10 +9,10 @@ teleprobe-meta = "1"
 
 embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt", "nightly"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "defmt", "nightly", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["arch-cortex-m", "executor-thread", "defmt", "nightly", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "nightly", "unstable-traits", "defmt-timestamp-uptime"] }
 embassy-nrf = { version = "0.1.0", path = "../../embassy-nrf", features = ["defmt", "nightly", "unstable-traits", "nrf52840", "time-driver-rtc1", "gpiote", "unstable-pac"] }
-embedded-io-async = { version = "0.6.0" }
+embedded-io-async = { version = "0.6.0", features = ["defmt-03"] }
 embassy-net = { version = "0.2.0", path = "../../embassy-net", features = ["defmt", "tcp", "dhcpv4", "medium-ethernet", "nightly"] }
 embassy-net-esp-hosted = { version = "0.1.0", path = "../../embassy-net-esp-hosted", features = ["defmt"] }
 embassy-net-enc28j60 = { version = "0.1.0", path = "../../embassy-net-enc28j60", features = ["defmt"] }

tests/nrf/src/bin/buffered_uart_full.rs

@@ -0,0 +1,72 @@
+#![no_std]
+#![no_main]
+#![feature(type_alias_impl_trait)]
+teleprobe_meta::target!(b"nrf52840-dk");
+
+use defmt::{assert_eq, *};
+use embassy_executor::Spawner;
+use embassy_nrf::buffered_uarte::{self, BufferedUarte};
+use embassy_nrf::{bind_interrupts, peripherals, uarte};
+use embedded_io_async::{Read, Write};
+use {defmt_rtt as _, panic_probe as _};
+
+bind_interrupts!(struct Irqs {
+    UARTE0_UART0 => buffered_uarte::InterruptHandler<peripherals::UARTE0>;
+});
+
+#[embassy_executor::main]
+async fn main(_spawner: Spawner) {
+    let p = embassy_nrf::init(Default::default());
+    let mut config = uarte::Config::default();
+    config.parity = uarte::Parity::EXCLUDED;
+    config.baudrate = uarte::Baudrate::BAUD1M;
+
+    let mut tx_buffer = [0u8; 1024];
+    let mut rx_buffer = [0u8; 1024];
+
+    let mut u = BufferedUarte::new(
+        p.UARTE0,
+        p.TIMER0,
+        p.PPI_CH0,
+        p.PPI_CH1,
+        p.PPI_GROUP0,
+        Irqs,
+        p.P1_03,
+        p.P1_02,
+        config.clone(),
+        &mut rx_buffer,
+        &mut tx_buffer,
+    );
+
+    info!("uarte initialized!");
+
+    let (mut rx, mut tx) = u.split();
+
+    let mut buf = [0; 1024];
+    for (j, b) in buf.iter_mut().enumerate() {
+        *b = j as u8;
+    }
+
+    // Write 1024b. This causes the rx buffer to get exactly full.
+    unwrap!(tx.write_all(&buf).await);
+    unwrap!(tx.flush().await);
+
+    // Read those 1024b.
+    unwrap!(rx.read_exact(&mut buf).await);
+    for (j, b) in buf.iter().enumerate() {
+        assert_eq!(*b, j as u8);
+    }
+
+    // The buffer should now be unclogged. Write 1024b again.
+    unwrap!(tx.write_all(&buf).await);
+    unwrap!(tx.flush().await);
+
+    // Read should work again.
+    unwrap!(rx.read_exact(&mut buf).await);
+    for (j, b) in buf.iter().enumerate() {
+        assert_eq!(*b, j as u8);
+    }
+
+    info!("Test OK");
+    cortex_m::asm::bkpt();
+}

tests/riscv32/Cargo.toml

@@ -7,7 +7,7 @@ license = "MIT OR Apache-2.0"
 [dependencies]
 critical-section = { version = "1.1.1", features = ["restore-state-bool"] }
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync" }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["arch-riscv32", "nightly", "executor-thread"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["arch-riscv32", "nightly", "executor-thread"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time" }
 embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }
 

tests/rp/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 teleprobe-meta = "1.1"
 
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "nightly", "unstable-traits"] }
 embassy-rp = { version = "0.1.0", path = "../../embassy-rp", features = ["nightly", "defmt", "unstable-pac", "unstable-traits", "time-driver", "critical-section-impl", "intrinsics", "rom-v2-intrinsics", "run-from-ram"]  }
 embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }

tests/stm32/Cargo.toml

@@ -48,7 +48,7 @@ cm0 = ["portable-atomic/unsafe-assume-single-core"]
 teleprobe-meta = "1"
 
 embassy-sync = { version = "0.4.0", path = "../../embassy-sync", features = ["defmt"] }
-embassy-executor = { version = "0.3.1", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
+embassy-executor = { version = "0.3.3", path = "../../embassy-executor", features = ["nightly", "arch-cortex-m", "executor-thread", "defmt", "integrated-timers"] }
 embassy-time = { version = "0.1.5", path = "../../embassy-time", features = ["defmt", "tick-hz-131_072", "defmt-timestamp-uptime"] }
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "unstable-pac", "memory-x", "time-driver-any"]  }
 embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }