Merge #984 #1006

984: rp pico async i2c implementation r=Dirbaio a=jsgf

This implements an interrupt-driven async i2c master. It is based on https://github.com/embassy-rs/embassy/pull/914, a bit of https://github.com/embassy-rs/embassy/pull/978 and `@ithinuel's` https://github.com/ithinuel/rp2040-async-i2c.git

This is still work-in-progress, and is currently untested.

1006: Removes some of the code duplication for UarteWithIdle r=Dirbaio a=huntc

This PR removes some of the code duplications for `UarteWithIdle` at the slight expense of requiring a split when using idle processing. As the nRF example illustrates though given the LoC removed, this expense seems worth the benefit in terms of maintenance, and the avoidance of copying over methods. My main motivation for this PR was actually due to the `event_endtx` method not having been copied across to the idle-related code.

Tested the uart_idle example on my nRF52840-dk, and from within my app. Both appear to work fine.

Co-authored-by: Jeremy Fitzhardinge <jeremy@goop.org>
Co-authored-by: huntc <huntchr@gmail.com>

embassy-nrf/src/uarte.rs

@@ -173,6 +173,61 @@ impl<'d, T: Instance> Uarte<'d, T> {
         (self.tx, self.rx)
     }
 
+    /// Split the Uarte into a transmitter and receiver that will
+    /// return on idle, which is determined as the time it takes
+    /// for two bytes to be received.
+    pub fn split_with_idle<U: TimerInstance>(
+        self,
+        timer: impl Peripheral<P = U> + 'd,
+        ppi_ch1: impl Peripheral<P = impl ConfigurableChannel + 'd> + 'd,
+        ppi_ch2: impl Peripheral<P = impl ConfigurableChannel + 'd> + 'd,
+    ) -> (UarteTx<'d, T>, UarteRxWithIdle<'d, T, U>) {
+        let mut timer = Timer::new(timer);
+
+        into_ref!(ppi_ch1, ppi_ch2);
+
+        let r = T::regs();
+
+        // BAUDRATE register values are `baudrate * 2^32 / 16000000`
+        // source: https://devzone.nordicsemi.com/f/nordic-q-a/391/uart-baudrate-register-values
+        //
+        // We want to stop RX if line is idle for 2 bytes worth of time
+        // That is 20 bits (each byte is 1 start bit + 8 data bits + 1 stop bit)
+        // This gives us the amount of 16M ticks for 20 bits.
+        let baudrate = r.baudrate.read().baudrate().variant().unwrap();
+        let timeout = 0x8000_0000 / (baudrate as u32 / 40);
+
+        timer.set_frequency(Frequency::F16MHz);
+        timer.cc(0).write(timeout);
+        timer.cc(0).short_compare_clear();
+        timer.cc(0).short_compare_stop();
+
+        let mut ppi_ch1 = Ppi::new_one_to_two(
+            ppi_ch1.map_into(),
+            Event::from_reg(&r.events_rxdrdy),
+            timer.task_clear(),
+            timer.task_start(),
+        );
+        ppi_ch1.enable();
+
+        let mut ppi_ch2 = Ppi::new_one_to_one(
+            ppi_ch2.map_into(),
+            timer.cc(0).event_compare(),
+            Task::from_reg(&r.tasks_stoprx),
+        );
+        ppi_ch2.enable();
+
+        (
+            self.tx,
+            UarteRxWithIdle {
+                rx: self.rx,
+                timer,
+                ppi_ch1: ppi_ch1,
+                _ppi_ch2: ppi_ch2,
+            },
+        )
+    }
+
     /// Return the endtx event for use with PPI
     pub fn event_endtx(&self) -> Event {
         let r = T::regs();
@@ -597,6 +652,117 @@ impl<'a, T: Instance> Drop for UarteRx<'a, T> {
     }
 }
 
+pub struct UarteRxWithIdle<'d, T: Instance, U: TimerInstance> {
+    rx: UarteRx<'d, T>,
+    timer: Timer<'d, U>,
+    ppi_ch1: Ppi<'d, AnyConfigurableChannel, 1, 2>,
+    _ppi_ch2: Ppi<'d, AnyConfigurableChannel, 1, 1>,
+}
+
+impl<'d, T: Instance, U: TimerInstance> UarteRxWithIdle<'d, T, U> {
+    pub async fn read(&mut self, buffer: &mut [u8]) -> Result<(), Error> {
+        self.ppi_ch1.disable();
+        self.rx.read(buffer).await
+    }
+
+    pub fn blocking_read(&mut self, buffer: &mut [u8]) -> Result<(), Error> {
+        self.ppi_ch1.disable();
+        self.rx.blocking_read(buffer)
+    }
+
+    pub async fn read_until_idle(&mut self, buffer: &mut [u8]) -> Result<usize, Error> {
+        if buffer.len() == 0 {
+            return Err(Error::BufferZeroLength);
+        }
+        if buffer.len() > EASY_DMA_SIZE {
+            return Err(Error::BufferTooLong);
+        }
+
+        let ptr = buffer.as_ptr();
+        let len = buffer.len();
+
+        let r = T::regs();
+        let s = T::state();
+
+        self.ppi_ch1.enable();
+
+        let drop = OnDrop::new(|| {
+            self.timer.stop();
+
+            r.intenclr.write(|w| w.endrx().clear());
+            r.events_rxto.reset();
+            r.tasks_stoprx.write(|w| unsafe { w.bits(1) });
+
+            while r.events_endrx.read().bits() == 0 {}
+        });
+
+        r.rxd.ptr.write(|w| unsafe { w.ptr().bits(ptr as u32) });
+        r.rxd.maxcnt.write(|w| unsafe { w.maxcnt().bits(len as _) });
+
+        r.events_endrx.reset();
+        r.intenset.write(|w| w.endrx().set());
+
+        compiler_fence(Ordering::SeqCst);
+
+        r.tasks_startrx.write(|w| unsafe { w.bits(1) });
+
+        poll_fn(|cx| {
+            s.endrx_waker.register(cx.waker());
+            if r.events_endrx.read().bits() != 0 {
+                return Poll::Ready(());
+            }
+            Poll::Pending
+        })
+        .await;
+
+        compiler_fence(Ordering::SeqCst);
+        let n = r.rxd.amount.read().amount().bits() as usize;
+
+        self.timer.stop();
+        r.events_rxstarted.reset();
+
+        drop.defuse();
+
+        Ok(n)
+    }
+
+    pub fn blocking_read_until_idle(&mut self, buffer: &mut [u8]) -> Result<usize, Error> {
+        if buffer.len() == 0 {
+            return Err(Error::BufferZeroLength);
+        }
+        if buffer.len() > EASY_DMA_SIZE {
+            return Err(Error::BufferTooLong);
+        }
+
+        let ptr = buffer.as_ptr();
+        let len = buffer.len();
+
+        let r = T::regs();
+
+        self.ppi_ch1.enable();
+
+        r.rxd.ptr.write(|w| unsafe { w.ptr().bits(ptr as u32) });
+        r.rxd.maxcnt.write(|w| unsafe { w.maxcnt().bits(len as _) });
+
+        r.events_endrx.reset();
+        r.intenclr.write(|w| w.endrx().clear());
+
+        compiler_fence(Ordering::SeqCst);
+
+        r.tasks_startrx.write(|w| unsafe { w.bits(1) });
+
+        while r.events_endrx.read().bits() == 0 {}
+
+        compiler_fence(Ordering::SeqCst);
+        let n = r.rxd.amount.read().amount().bits() as usize;
+
+        self.timer.stop();
+        r.events_rxstarted.reset();
+
+        Ok(n)
+    }
+}
+
 #[cfg(not(any(feature = "_nrf9160", feature = "nrf5340")))]
 pub(crate) fn apply_workaround_for_enable_anomaly(_r: &crate::pac::uarte0::RegisterBlock) {
     // Do nothing
@@ -665,270 +831,6 @@ pub(crate) fn drop_tx_rx(r: &pac::uarte0::RegisterBlock, s: &sealed::State) {
     }
 }
 
-/// Interface to an UARTE peripheral that uses an additional timer and two PPI channels,
-/// allowing it to implement the ReadUntilIdle trait.
-pub struct UarteWithIdle<'d, U: Instance, T: TimerInstance> {
-    tx: UarteTx<'d, U>,
-    rx: UarteRxWithIdle<'d, U, T>,
-}
-
-impl<'d, U: Instance, T: TimerInstance> UarteWithIdle<'d, U, T> {
-    /// Create a new UARTE without hardware flow control
-    pub fn new(
-        uarte: impl Peripheral<P = U> + 'd,
-        timer: impl Peripheral<P = T> + 'd,
-        ppi_ch1: impl Peripheral<P = impl ConfigurableChannel + 'd> + 'd,
-        ppi_ch2: impl Peripheral<P = impl ConfigurableChannel + 'd> + 'd,
-        irq: impl Peripheral<P = U::Interrupt> + 'd,
-        rxd: impl Peripheral<P = impl GpioPin> + 'd,
-        txd: impl Peripheral<P = impl GpioPin> + 'd,
-        config: Config,
-    ) -> Self {
-        into_ref!(rxd, txd);
-        Self::new_inner(
-            uarte,
-            timer,
-            ppi_ch1,
-            ppi_ch2,
-            irq,
-            rxd.map_into(),
-            txd.map_into(),
-            None,
-            None,
-            config,
-        )
-    }
-
-    /// Create a new UARTE with hardware flow control (RTS/CTS)
-    pub fn new_with_rtscts(
-        uarte: impl Peripheral<P = U> + 'd,
-        timer: impl Peripheral<P = T> + 'd,
-        ppi_ch1: impl Peripheral<P = impl ConfigurableChannel + 'd> + 'd,
-        ppi_ch2: impl Peripheral<P = impl ConfigurableChannel + 'd> + 'd,
-        irq: impl Peripheral<P = U::Interrupt> + 'd,
-        rxd: impl Peripheral<P = impl GpioPin> + 'd,
-        txd: impl Peripheral<P = impl GpioPin> + 'd,
-        cts: impl Peripheral<P = impl GpioPin> + 'd,
-        rts: impl Peripheral<P = impl GpioPin> + 'd,
-        config: Config,
-    ) -> Self {
-        into_ref!(rxd, txd, cts, rts);
-        Self::new_inner(
-            uarte,
-            timer,
-            ppi_ch1,
-            ppi_ch2,
-            irq,
-            rxd.map_into(),
-            txd.map_into(),
-            Some(cts.map_into()),
-            Some(rts.map_into()),
-            config,
-        )
-    }
-
-    fn new_inner(
-        uarte: impl Peripheral<P = U> + 'd,
-        timer: impl Peripheral<P = T> + 'd,
-        ppi_ch1: impl Peripheral<P = impl ConfigurableChannel + 'd> + 'd,
-        ppi_ch2: impl Peripheral<P = impl ConfigurableChannel + 'd> + 'd,
-        irq: impl Peripheral<P = U::Interrupt> + 'd,
-        rxd: PeripheralRef<'d, AnyPin>,
-        txd: PeripheralRef<'d, AnyPin>,
-        cts: Option<PeripheralRef<'d, AnyPin>>,
-        rts: Option<PeripheralRef<'d, AnyPin>>,
-        config: Config,
-    ) -> Self {
-        let baudrate = config.baudrate;
-        let (tx, rx) = Uarte::new_inner(uarte, irq, rxd, txd, cts, rts, config).split();
-
-        let mut timer = Timer::new(timer);
-
-        into_ref!(ppi_ch1, ppi_ch2);
-
-        let r = U::regs();
-
-        // BAUDRATE register values are `baudrate * 2^32 / 16000000`
-        // source: https://devzone.nordicsemi.com/f/nordic-q-a/391/uart-baudrate-register-values
-        //
-        // We want to stop RX if line is idle for 2 bytes worth of time
-        // That is 20 bits (each byte is 1 start bit + 8 data bits + 1 stop bit)
-        // This gives us the amount of 16M ticks for 20 bits.
-        let timeout = 0x8000_0000 / (baudrate as u32 / 40);
-
-        timer.set_frequency(Frequency::F16MHz);
-        timer.cc(0).write(timeout);
-        timer.cc(0).short_compare_clear();
-        timer.cc(0).short_compare_stop();
-
-        let mut ppi_ch1 = Ppi::new_one_to_two(
-            ppi_ch1.map_into(),
-            Event::from_reg(&r.events_rxdrdy),
-            timer.task_clear(),
-            timer.task_start(),
-        );
-        ppi_ch1.enable();
-
-        let mut ppi_ch2 = Ppi::new_one_to_one(
-            ppi_ch2.map_into(),
-            timer.cc(0).event_compare(),
-            Task::from_reg(&r.tasks_stoprx),
-        );
-        ppi_ch2.enable();
-
-        Self {
-            tx,
-            rx: UarteRxWithIdle {
-                rx,
-                timer,
-                ppi_ch1: ppi_ch1,
-                _ppi_ch2: ppi_ch2,
-            },
-        }
-    }
-
-    /// Split the Uarte into a transmitter and receiver, which is
-    /// particuarly useful when having two tasks correlating to
-    /// transmitting and receiving.
-    pub fn split(self) -> (UarteTx<'d, U>, UarteRxWithIdle<'d, U, T>) {
-        (self.tx, self.rx)
-    }
-
-    pub async fn read(&mut self, buffer: &mut [u8]) -> Result<(), Error> {
-        self.rx.read(buffer).await
-    }
-
-    pub async fn write(&mut self, buffer: &[u8]) -> Result<(), Error> {
-        self.tx.write(buffer).await
-    }
-
-    pub fn blocking_read(&mut self, buffer: &mut [u8]) -> Result<(), Error> {
-        self.rx.blocking_read(buffer)
-    }
-
-    pub fn blocking_write(&mut self, buffer: &[u8]) -> Result<(), Error> {
-        self.tx.blocking_write(buffer)
-    }
-
-    pub async fn read_until_idle(&mut self, buffer: &mut [u8]) -> Result<usize, Error> {
-        self.rx.read_until_idle(buffer).await
-    }
-
-    pub fn blocking_read_until_idle(&mut self, buffer: &mut [u8]) -> Result<usize, Error> {
-        self.rx.blocking_read_until_idle(buffer)
-    }
-}
-
-pub struct UarteRxWithIdle<'d, U: Instance, T: TimerInstance> {
-    rx: UarteRx<'d, U>,
-    timer: Timer<'d, T>,
-    ppi_ch1: Ppi<'d, AnyConfigurableChannel, 1, 2>,
-    _ppi_ch2: Ppi<'d, AnyConfigurableChannel, 1, 1>,
-}
-
-impl<'d, U: Instance, T: TimerInstance> UarteRxWithIdle<'d, U, T> {
-    pub async fn read(&mut self, buffer: &mut [u8]) -> Result<(), Error> {
-        self.ppi_ch1.disable();
-        self.rx.read(buffer).await
-    }
-
-    pub fn blocking_read(&mut self, buffer: &mut [u8]) -> Result<(), Error> {
-        self.ppi_ch1.disable();
-        self.rx.blocking_read(buffer)
-    }
-
-    pub async fn read_until_idle(&mut self, buffer: &mut [u8]) -> Result<usize, Error> {
-        if buffer.len() == 0 {
-            return Err(Error::BufferZeroLength);
-        }
-        if buffer.len() > EASY_DMA_SIZE {
-            return Err(Error::BufferTooLong);
-        }
-
-        let ptr = buffer.as_ptr();
-        let len = buffer.len();
-
-        let r = U::regs();
-        let s = U::state();
-
-        self.ppi_ch1.enable();
-
-        let drop = OnDrop::new(|| {
-            self.timer.stop();
-
-            r.intenclr.write(|w| w.endrx().clear());
-            r.events_rxto.reset();
-            r.tasks_stoprx.write(|w| unsafe { w.bits(1) });
-
-            while r.events_endrx.read().bits() == 0 {}
-        });
-
-        r.rxd.ptr.write(|w| unsafe { w.ptr().bits(ptr as u32) });
-        r.rxd.maxcnt.write(|w| unsafe { w.maxcnt().bits(len as _) });
-
-        r.events_endrx.reset();
-        r.intenset.write(|w| w.endrx().set());
-
-        compiler_fence(Ordering::SeqCst);
-
-        r.tasks_startrx.write(|w| unsafe { w.bits(1) });
-
-        poll_fn(|cx| {
-            s.endrx_waker.register(cx.waker());
-            if r.events_endrx.read().bits() != 0 {
-                return Poll::Ready(());
-            }
-            Poll::Pending
-        })
-        .await;
-
-        compiler_fence(Ordering::SeqCst);
-        let n = r.rxd.amount.read().amount().bits() as usize;
-
-        self.timer.stop();
-        r.events_rxstarted.reset();
-
-        drop.defuse();
-
-        Ok(n)
-    }
-
-    pub fn blocking_read_until_idle(&mut self, buffer: &mut [u8]) -> Result<usize, Error> {
-        if buffer.len() == 0 {
-            return Err(Error::BufferZeroLength);
-        }
-        if buffer.len() > EASY_DMA_SIZE {
-            return Err(Error::BufferTooLong);
-        }
-
-        let ptr = buffer.as_ptr();
-        let len = buffer.len();
-
-        let r = U::regs();
-
-        self.ppi_ch1.enable();
-
-        r.rxd.ptr.write(|w| unsafe { w.ptr().bits(ptr as u32) });
-        r.rxd.maxcnt.write(|w| unsafe { w.maxcnt().bits(len as _) });
-
-        r.events_endrx.reset();
-        r.intenclr.write(|w| w.endrx().clear());
-
-        compiler_fence(Ordering::SeqCst);
-
-        r.tasks_startrx.write(|w| unsafe { w.bits(1) });
-
-        while r.events_endrx.read().bits() == 0 {}
-
-        compiler_fence(Ordering::SeqCst);
-        let n = r.rxd.amount.read().amount().bits() as usize;
-
-        self.timer.stop();
-        r.events_rxstarted.reset();
-
-        Ok(n)
-    }
-}
 pub(crate) mod sealed {
     use core::sync::atomic::AtomicU8;
 
@@ -1006,18 +908,6 @@ mod eh02 {
             Ok(())
         }
     }
-
-    impl<'d, U: Instance, T: TimerInstance> embedded_hal_02::blocking::serial::Write<u8> for UarteWithIdle<'d, U, T> {
-        type Error = Error;
-
-        fn bwrite_all(&mut self, buffer: &[u8]) -> Result<(), Self::Error> {
-            self.blocking_write(buffer)
-        }
-
-        fn bflush(&mut self) -> Result<(), Self::Error> {
-            Ok(())
-        }
-    }
 }
 
 #[cfg(feature = "unstable-traits")]
@@ -1067,10 +957,6 @@ mod eh1 {
     impl<'d, T: Instance> embedded_hal_1::serial::ErrorType for UarteRx<'d, T> {
         type Error = Error;
     }
-
-    impl<'d, U: Instance, T: TimerInstance> embedded_hal_1::serial::ErrorType for UarteWithIdle<'d, U, T> {
-        type Error = Error;
-    }
 }
 
 #[cfg(all(
@@ -1126,26 +1012,4 @@ mod eha {
             self.read(buffer)
         }
     }
-
-    impl<'d, U: Instance, T: TimerInstance> embedded_hal_async::serial::Read for UarteWithIdle<'d, U, T> {
-        type ReadFuture<'a> = impl Future<Output = Result<(), Self::Error>> + 'a where Self: 'a;
-
-        fn read<'a>(&'a mut self, buffer: &'a mut [u8]) -> Self::ReadFuture<'a> {
-            self.read(buffer)
-        }
-    }
-
-    impl<'d, U: Instance, T: TimerInstance> embedded_hal_async::serial::Write for UarteWithIdle<'d, U, T> {
-        type WriteFuture<'a> = impl Future<Output = Result<(), Self::Error>> + 'a where Self: 'a;
-
-        fn write<'a>(&'a mut self, buffer: &'a [u8]) -> Self::WriteFuture<'a> {
-            self.write(buffer)
-        }
-
-        type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>> + 'a where Self: 'a;
-
-        fn flush<'a>(&'a mut self) -> Self::FlushFuture<'a> {
-            async move { Ok(()) }
-        }
-    }
 }

embassy-rp/src/i2c.rs

@@ -1,9 +1,12 @@
+use core::future;
 use core::marker::PhantomData;
+use core::task::Poll;
 
+use embassy_cortex_m::interrupt::InterruptExt;
 use embassy_hal_common::{into_ref, PeripheralRef};
+use embassy_sync::waitqueue::AtomicWaker;
 use pac::i2c;
 
-use crate::dma::AnyChannel;
 use crate::gpio::sealed::Pin;
 use crate::gpio::AnyPin;
 use crate::{pac, peripherals, Peripheral};
@@ -52,31 +55,278 @@ impl Default for Config {
 const FIFO_SIZE: u8 = 16;
 
 pub struct I2c<'d, T: Instance, M: Mode> {
-    _tx_dma: Option<PeripheralRef<'d, AnyChannel>>,
-    _rx_dma: Option<PeripheralRef<'d, AnyChannel>>,
-    _dma_buf: [u16; 256],
     phantom: PhantomData<(&'d mut T, M)>,
 }
 
 impl<'d, T: Instance> I2c<'d, T, Blocking> {
     pub fn new_blocking(
-        _peri: impl Peripheral<P = T> + 'd,
+        peri: impl Peripheral<P = T> + 'd,
         scl: impl Peripheral<P = impl SclPin<T>> + 'd,
         sda: impl Peripheral<P = impl SdaPin<T>> + 'd,
         config: Config,
     ) -> Self {
         into_ref!(scl, sda);
-        Self::new_inner(_peri, scl.map_into(), sda.map_into(), None, None, config)
+        Self::new_inner(peri, scl.map_into(), sda.map_into(), config)
     }
 }
 
-impl<'d, T: Instance, M: Mode> I2c<'d, T, M> {
+static I2C_WAKER: AtomicWaker = AtomicWaker::new();
+
+impl<'d, T: Instance> I2c<'d, T, Async> {
+    pub fn new_async(
+        peri: impl Peripheral<P = T> + 'd,
+        scl: impl Peripheral<P = impl SclPin<T>> + 'd,
+        sda: impl Peripheral<P = impl SdaPin<T>> + 'd,
+        irq: impl Peripheral<P = T::Interrupt> + 'd,
+        config: Config,
+    ) -> Self {
+        into_ref!(scl, sda, irq);
+
+        let i2c = Self::new_inner(peri, scl.map_into(), sda.map_into(), config);
+
+        irq.set_handler(Self::on_interrupt);
+        unsafe {
+            let i2c = T::regs();
+
+            // mask everything initially
+            i2c.ic_intr_mask().write_value(i2c::regs::IcIntrMask(0));
+        }
+        irq.unpend();
+        debug_assert!(!irq.is_pending());
+        irq.enable();
+
+        i2c
+    }
+
+    /// Calls `f` to check if we are ready or not.
+    /// If not, `g` is called once the waker is set (to eg enable the required interrupts).
+    async fn wait_on<F, U, G>(&mut self, mut f: F, mut g: G) -> U
+    where
+        F: FnMut(&mut Self) -> Poll<U>,
+        G: FnMut(&mut Self),
+    {
+        future::poll_fn(|cx| {
+            let r = f(self);
+
+            if r.is_pending() {
+                I2C_WAKER.register(cx.waker());
+                g(self);
+            }
+            r
+        })
+        .await
+    }
+
+    // Mask interrupts and wake any task waiting for this interrupt
+    unsafe fn on_interrupt(_: *mut ()) {
+        let i2c = T::regs();
+        i2c.ic_intr_mask().write_value(pac::i2c::regs::IcIntrMask::default());
+
+        I2C_WAKER.wake();
+    }
+
+    async fn read_async_internal(&mut self, buffer: &mut [u8], restart: bool, send_stop: bool) -> Result<(), Error> {
+        if buffer.is_empty() {
+            return Err(Error::InvalidReadBufferLength);
+        }
+
+        let p = T::regs();
+
+        let mut remaining = buffer.len();
+        let mut remaining_queue = buffer.len();
+
+        let mut abort_reason = Ok(());
+
+        while remaining > 0 {
+            // Waggle SCK - basically the same as write
+            let tx_fifo_space = Self::tx_fifo_capacity();
+            let mut batch = 0;
+
+            debug_assert!(remaining_queue > 0);
+
+            for _ in 0..remaining_queue.min(tx_fifo_space as usize) {
+                remaining_queue -= 1;
+                let last = remaining_queue == 0;
+                batch += 1;
+
+                unsafe {
+                    p.ic_data_cmd().write(|w| {
+                        w.set_restart(restart && remaining_queue == buffer.len() - 1);
+                        w.set_stop(last && send_stop);
+                        w.set_cmd(true);
+                    });
+                }
+            }
+
+            // We've either run out of txfifo or just plain finished setting up
+            // the clocks for the message - either way we need to wait for rx
+            // data.
+
+            debug_assert!(batch > 0);
+            let res = self
+                .wait_on(
+                    |me| {
+                        let rxfifo = Self::rx_fifo_len();
+                        if let Err(abort_reason) = me.read_and_clear_abort_reason() {
+                            Poll::Ready(Err(abort_reason))
+                        } else if rxfifo >= batch {
+                            Poll::Ready(Ok(rxfifo))
+                        } else {
+                            Poll::Pending
+                        }
+                    },
+                    |_me| unsafe {
+                        // Set the read threshold to the number of bytes we're
+                        // expecting so we don't get spurious interrupts.
+                        p.ic_rx_tl().write(|w| w.set_rx_tl(batch - 1));
+
+                        p.ic_intr_mask().modify(|w| {
+                            w.set_m_rx_full(true);
+                            w.set_m_tx_abrt(true);
+                        });
+                    },
+                )
+                .await;
+
+            match res {
+                Err(reason) => {
+                    abort_reason = Err(reason);
+                    break;
+                }
+                Ok(rxfifo) => {
+                    // Fetch things from rx fifo. We're assuming we're the only
+                    // rxfifo reader, so nothing else can take things from it.
+                    let rxbytes = (rxfifo as usize).min(remaining);
+                    let received = buffer.len() - remaining;
+                    for b in &mut buffer[received..received + rxbytes] {
+                        *b = unsafe { p.ic_data_cmd().read().dat() };
+                    }
+                    remaining -= rxbytes;
+                }
+            };
+        }
+
+        self.wait_stop_det(abort_reason, send_stop).await
+    }
+
+    async fn write_async_internal(
+        &mut self,
+        bytes: impl IntoIterator<Item = u8>,
+        send_stop: bool,
+    ) -> Result<(), Error> {
+        let p = T::regs();
+
+        let mut bytes = bytes.into_iter().peekable();
+
+        let res = 'xmit: loop {
+            let tx_fifo_space = Self::tx_fifo_capacity();
+
+            for _ in 0..tx_fifo_space {
+                if let Some(byte) = bytes.next() {
+                    let last = bytes.peek().is_none();
+
+                    unsafe {
+                        p.ic_data_cmd().write(|w| {
+                            w.set_stop(last && send_stop);
+                            w.set_cmd(false);
+                            w.set_dat(byte);
+                        });
+                    }
+                } else {
+                    break 'xmit Ok(());
+                }
+            }
+
+            let res = self
+                .wait_on(
+                    |me| {
+                        if let abort_reason @ Err(_) = me.read_and_clear_abort_reason() {
+                            Poll::Ready(abort_reason)
+                        } else if !Self::tx_fifo_full() {
+                            // resume if there's any space free in the tx fifo
+                            Poll::Ready(Ok(()))
+                        } else {
+                            Poll::Pending
+                        }
+                    },
+                    |_me| unsafe {
+                        // Set tx "free" threshold a little high so that we get
+                        // woken before the fifo completely drains to minimize
+                        // transfer stalls.
+                        p.ic_tx_tl().write(|w| w.set_tx_tl(1));
+
+                        p.ic_intr_mask().modify(|w| {
+                            w.set_m_tx_empty(true);
+                            w.set_m_tx_abrt(true);
+                        })
+                    },
+                )
+                .await;
+            if res.is_err() {
+                break res;
+            }
+        };
+
+        self.wait_stop_det(res, send_stop).await
+    }
+
+    /// Helper to wait for a stop bit, for both tx and rx. If we had an abort,
+    /// then we'll get a hardware-generated stop, otherwise wait for a stop if
+    /// we're expecting it.
+    ///
+    /// Also handles an abort which arises while processing the tx fifo.
+    async fn wait_stop_det(&mut self, had_abort: Result<(), Error>, do_stop: bool) -> Result<(), Error> {
+        if had_abort.is_err() || do_stop {
+            let p = T::regs();
+
+            let had_abort2 = self
+                .wait_on(
+                    |me| unsafe {
+                        // We could see an abort while processing fifo backlog,
+                        // so handle it here.
+                        let abort = me.read_and_clear_abort_reason();
+                        if had_abort.is_ok() && abort.is_err() {
+                            Poll::Ready(abort)
+                        } else if p.ic_raw_intr_stat().read().stop_det() {
+                            Poll::Ready(Ok(()))
+                        } else {
+                            Poll::Pending
+                        }
+                    },
+                    |_me| unsafe {
+                        p.ic_intr_mask().modify(|w| {
+                            w.set_m_stop_det(true);
+                            w.set_m_tx_abrt(true);
+                        });
+                    },
+                )
+                .await;
+            unsafe {
+                p.ic_clr_stop_det().read();
+            }
+
+            had_abort.and(had_abort2)
+        } else {
+            had_abort
+        }
+    }
+
+    pub async fn read_async(&mut self, addr: u16, buffer: &mut [u8]) -> Result<(), Error> {
+        Self::setup(addr)?;
+        self.read_async_internal(buffer, false, true).await
+    }
+
+    pub async fn write_async(&mut self, addr: u16, bytes: impl IntoIterator<Item = u8>) -> Result<(), Error> {
+        Self::setup(addr)?;
+        self.write_async_internal(bytes, true).await
+    }
+}
+
+impl<'d, T: Instance + 'd, M: Mode> I2c<'d, T, M> {
     fn new_inner(
         _peri: impl Peripheral<P = T> + 'd,
         scl: PeripheralRef<'d, AnyPin>,
         sda: PeripheralRef<'d, AnyPin>,
-        _tx_dma: Option<PeripheralRef<'d, AnyChannel>>,
-        _rx_dma: Option<PeripheralRef<'d, AnyChannel>>,
         config: Config,
     ) -> Self {
         into_ref!(_peri);
@@ -87,6 +337,10 @@ impl<'d, T: Instance, M: Mode> I2c<'d, T, M> {
         let p = T::regs();
 
         unsafe {
+            let reset = T::reset();
+            crate::reset::reset(reset);
+            crate::reset::unreset_wait(reset);
+
             p.ic_enable().write(|w| w.set_enable(false));
 
             // Select controller mode & speed
@@ -172,12 +426,7 @@ impl<'d, T: Instance, M: Mode> I2c<'d, T, M> {
             p.ic_enable().write(|w| w.set_enable(true));
         }
 
-        Self {
+        Self { phantom: PhantomData }
-            _tx_dma,
-            _rx_dma,
-            _dma_buf: [0; 256],
-            phantom: PhantomData,
-        }
     }
 
     fn setup(addr: u16) -> Result<(), Error> {
@@ -198,6 +447,23 @@ impl<'d, T: Instance, M: Mode> I2c<'d, T, M> {
         Ok(())
     }
 
+    #[inline]
+    fn tx_fifo_full() -> bool {
+        Self::tx_fifo_capacity() == 0
+    }
+
+    #[inline]
+    fn tx_fifo_capacity() -> u8 {
+        let p = T::regs();
+        unsafe { FIFO_SIZE - p.ic_txflr().read().txflr() }
+    }
+
+    #[inline]
+    fn rx_fifo_len() -> u8 {
+        let p = T::regs();
+        unsafe { p.ic_rxflr().read().rxflr() }
+    }
+
     fn read_and_clear_abort_reason(&mut self) -> Result<(), Error> {
         let p = T::regs();
         unsafe {
@@ -240,7 +506,7 @@ impl<'d, T: Instance, M: Mode> I2c<'d, T, M> {
             // NOTE(unsafe) We have &mut self
             unsafe {
                 // wait until there is space in the FIFO to write the next byte
-                while p.ic_txflr().read().txflr() == FIFO_SIZE {}
+                while Self::tx_fifo_full() {}
 
                 p.ic_data_cmd().write(|w| {
                     w.set_restart(restart && first);
@@ -249,7 +515,7 @@ impl<'d, T: Instance, M: Mode> I2c<'d, T, M> {
                     w.set_cmd(true);
                 });
 
-                while p.ic_rxflr().read().rxflr() == 0 {
+                while Self::rx_fifo_len() == 0 {
                     self.read_and_clear_abort_reason()?;
                 }
 
@@ -451,6 +717,91 @@ mod eh1 {
         }
     }
 }
+#[cfg(all(feature = "unstable-traits", feature = "nightly"))]
+mod nightly {
+    use core::future::Future;
+
+    use embedded_hal_1::i2c::Operation;
+    use embedded_hal_async::i2c::AddressMode;
+
+    use super::*;
+
+    impl<'d, A, T> embedded_hal_async::i2c::I2c<A> for I2c<'d, T, Async>
+    where
+        A: AddressMode + Into<u16> + 'static,
+        T: Instance + 'd,
+    {
+        type ReadFuture<'a> = impl Future<Output = Result<(), Self::Error>> + 'a
+            where Self: 'a;
+        type WriteFuture<'a> = impl Future<Output = Result<(), Self::Error>> + 'a
+            where Self: 'a;
+        type WriteReadFuture<'a> = impl Future<Output = Result<(), Self::Error>> + 'a
+            where Self: 'a;
+        type TransactionFuture<'a, 'b> = impl Future<Output = Result<(), Error>> + 'a
+            where Self: 'a, 'b: 'a;
+
+        fn read<'a>(&'a mut self, address: A, buffer: &'a mut [u8]) -> Self::ReadFuture<'a> {
+            let addr: u16 = address.into();
+
+            async move {
+                Self::setup(addr)?;
+                self.read_async_internal(buffer, false, true).await
+            }
+        }
+
+        fn write<'a>(&'a mut self, address: A, write: &'a [u8]) -> Self::WriteFuture<'a> {
+            let addr: u16 = address.into();
+
+            async move {
+                Self::setup(addr)?;
+                self.write_async_internal(write.iter().copied(), true).await
+            }
+        }
+
+        fn write_read<'a>(
+            &'a mut self,
+            address: A,
+            bytes: &'a [u8],
+            buffer: &'a mut [u8],
+        ) -> Self::WriteReadFuture<'a> {
+            let addr: u16 = address.into();
+
+            async move {
+                Self::setup(addr)?;
+                self.write_async_internal(bytes.iter().cloned(), false).await?;
+                self.read_async_internal(buffer, false, true).await
+            }
+        }
+
+        fn transaction<'a, 'b>(
+            &'a mut self,
+            address: A,
+            operations: &'a mut [Operation<'b>],
+        ) -> Self::TransactionFuture<'a, 'b> {
+            let addr: u16 = address.into();
+
+            async move {
+                let mut iterator = operations.iter_mut();
+
+                while let Some(op) = iterator.next() {
+                    let last = iterator.len() == 0;
+
+                    match op {
+                        Operation::Read(buffer) => {
+                            Self::setup(addr)?;
+                            self.read_async_internal(buffer, false, last).await?;
+                        }
+                        Operation::Write(buffer) => {
+                            Self::setup(addr)?;
+                            self.write_async_internal(buffer.into_iter().cloned(), last).await?;
+                        }
+                    }
+                }
+                Ok(())
+            }
+        }
+    }
+}
 
 fn i2c_reserved_addr(addr: u16) -> bool {
     (addr & 0x78) == 0 || (addr & 0x78) == 0x78
@@ -466,6 +817,7 @@ mod sealed {
         type Interrupt: Interrupt;
 
         fn regs() -> crate::pac::i2c::I2c;
+        fn reset() -> crate::pac::resets::regs::Peripherals;
     }
 
     pub trait Mode {}
@@ -492,23 +844,31 @@ impl_mode!(Async);
 pub trait Instance: sealed::Instance {}
 
 macro_rules! impl_instance {
-    ($type:ident, $irq:ident, $tx_dreq:expr, $rx_dreq:expr) => {
+    ($type:ident, $irq:ident, $reset:ident, $tx_dreq:expr, $rx_dreq:expr) => {
         impl sealed::Instance for peripherals::$type {
             const TX_DREQ: u8 = $tx_dreq;
             const RX_DREQ: u8 = $rx_dreq;
 
             type Interrupt = crate::interrupt::$irq;
 
+            #[inline]
             fn regs() -> pac::i2c::I2c {
                 pac::$type
             }
+
+            #[inline]
+            fn reset() -> pac::resets::regs::Peripherals {
+                let mut ret = pac::resets::regs::Peripherals::default();
+                ret.$reset(true);
+                ret
+            }
         }
         impl Instance for peripherals::$type {}
     };
 }
 
-impl_instance!(I2C0, I2C0_IRQ, 32, 33);
+impl_instance!(I2C0, I2C0_IRQ, set_i2c0, 32, 33);
-impl_instance!(I2C1, I2C1_IRQ, 34, 35);
+impl_instance!(I2C1, I2C1_IRQ, set_i2c1, 34, 35);
 
 pub trait SdaPin<T: Instance>: sealed::SdaPin<T> + crate::gpio::Pin {}
 pub trait SclPin<T: Instance>: sealed::SclPin<T> + crate::gpio::Pin {}

examples/nrf/src/bin/uart_idle.rs

@@ -15,7 +15,8 @@ async fn main(_spawner: Spawner) {
     config.baudrate = uarte::Baudrate::BAUD115200;
 
     let irq = interrupt::take!(UARTE0_UART0);
-    let mut uart = uarte::UarteWithIdle::new(p.UARTE0, p.TIMER0, p.PPI_CH0, p.PPI_CH1, irq, p.P0_08, p.P0_06, config);
+    let uart = uarte::Uarte::new(p.UARTE0, irq, p.P0_08, p.P0_06, config);
+    let (mut tx, mut rx) = uart.split_with_idle(p.TIMER0, p.PPI_CH0, p.PPI_CH1);
 
     info!("uarte initialized!");
 
@@ -23,12 +24,12 @@ async fn main(_spawner: Spawner) {
     let mut buf = [0; 8];
     buf.copy_from_slice(b"Hello!\r\n");
 
-    unwrap!(uart.write(&buf).await);
+    unwrap!(tx.write(&buf).await);
     info!("wrote hello in uart!");
 
     loop {
         info!("reading...");
-        let n = unwrap!(uart.read_until_idle(&mut buf).await);
+        let n = unwrap!(rx.read_until_idle(&mut buf).await);
         info!("got {} bytes", n);
     }
 }

examples/rp/Cargo.toml

@@ -31,3 +31,6 @@ embedded-hal-1 = { package = "embedded-hal", version = "=1.0.0-alpha.9" }
 embedded-hal-async = { version = "0.1.0-alpha.1" }
 embedded-io = { version = "0.3.0", features = ["async", "defmt"] }
 static_cell = "1.0.0"
+
+[profile.release]
+debug = true

examples/rp/src/bin/i2c_async.rs

@@ -0,0 +1,102 @@
+#![no_std]
+#![no_main]
+#![feature(type_alias_impl_trait)]
+
+use defmt::*;
+use embassy_executor::Spawner;
+use embassy_rp::i2c::{self, Config};
+use embassy_rp::interrupt;
+use embassy_time::{Duration, Timer};
+use embedded_hal_async::i2c::I2c;
+use {defmt_rtt as _, panic_probe as _};
+
+#[allow(dead_code)]
+mod mcp23017 {
+    pub const ADDR: u8 = 0x20; // default addr
+
+    macro_rules! mcpregs {
+        ($($name:ident : $val:expr),* $(,)?) => {
+            $(
+                pub const $name: u8 = $val;
+            )*
+
+            pub fn regname(reg: u8) -> &'static str {
+                match reg {
+                    $(
+                        $val => stringify!($name),
+                    )*
+                    _ => panic!("bad reg"),
+                }
+            }
+        }
+    }
+
+    // These are correct for IOCON.BANK=0
+    mcpregs! {
+        IODIRA: 0x00,
+        IPOLA: 0x02,
+        GPINTENA: 0x04,
+        DEFVALA: 0x06,
+        INTCONA: 0x08,
+        IOCONA: 0x0A,
+        GPPUA: 0x0C,
+        INTFA: 0x0E,
+        INTCAPA: 0x10,
+        GPIOA: 0x12,
+        OLATA: 0x14,
+        IODIRB: 0x01,
+        IPOLB: 0x03,
+        GPINTENB: 0x05,
+        DEFVALB: 0x07,
+        INTCONB: 0x09,
+        IOCONB: 0x0B,
+        GPPUB: 0x0D,
+        INTFB: 0x0F,
+        INTCAPB: 0x11,
+        GPIOB: 0x13,
+        OLATB: 0x15,
+    }
+}
+
+#[embassy_executor::main]
+async fn main(_spawner: Spawner) {
+    let p = embassy_rp::init(Default::default());
+
+    let sda = p.PIN_14;
+    let scl = p.PIN_15;
+    let irq = interrupt::take!(I2C1_IRQ);
+
+    info!("set up i2c ");
+    let mut i2c = i2c::I2c::new_async(p.I2C1, scl, sda, irq, Config::default());
+
+    use mcp23017::*;
+
+    info!("init mcp23017 config for IxpandO");
+    // init - a outputs, b inputs
+    i2c.write(ADDR, &[IODIRA, 0x00]).await.unwrap();
+    i2c.write(ADDR, &[IODIRB, 0xff]).await.unwrap();
+    i2c.write(ADDR, &[GPPUB, 0xff]).await.unwrap(); // pullups
+
+    let mut val = 1;
+    loop {
+        let mut portb = [0];
+
+        i2c.write_read(mcp23017::ADDR, &[GPIOB], &mut portb).await.unwrap();
+        info!("portb = {:02x}", portb[0]);
+        i2c.write(mcp23017::ADDR, &[GPIOA, val | portb[0]]).await.unwrap();
+        val = val.rotate_left(1);
+
+        // get a register dump
+        info!("getting register dump");
+        let mut regs = [0; 22];
+        i2c.write_read(ADDR, &[0], &mut regs).await.unwrap();
+        // always get the regdump but only display it if portb'0 is set
+        if portb[0] & 1 != 0 {
+            for (idx, reg) in regs.into_iter().enumerate() {
+                info!("{} => {:02x}", regname(idx as u8), reg);
+            }
+        }
+
+        Timer::after(Duration::from_millis(100)).await;
+    }
+}