5 files changed, 620 insertions, 290 deletions

diff --git a/embassy-nrf/src/uarte.rs b/embassy-nrf/src/uarte.rs
index d99599112..636d6c7a3 100644
--- a/embassy-nrf/src/uarte.rs
+++ b/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,236 +652,14 @@ impl<'a, T: Instance> Drop for UarteRx<'a, T> {
     }
 }
 
-#[cfg(not(any(feature = "_nrf9160", feature = "nrf5340")))]
-pub(crate) fn apply_workaround_for_enable_anomaly(_r: &crate::pac::uarte0::RegisterBlock) {
-    // Do nothing
-}
-
-#[cfg(any(feature = "_nrf9160", feature = "nrf5340"))]
-pub(crate) fn apply_workaround_for_enable_anomaly(r: &crate::pac::uarte0::RegisterBlock) {
-    use core::ops::Deref;
-
-    // Apply workaround for anomalies:
-    // - nRF9160 - anomaly 23
-    // - nRF5340 - anomaly 44
-    let rxenable_reg: *const u32 = ((r.deref() as *const _ as usize) + 0x564) as *const u32;
-    let txenable_reg: *const u32 = ((r.deref() as *const _ as usize) + 0x568) as *const u32;
-
-    // NB Safety: This is taken from Nordic's driver -
-    // https://github.com/NordicSemiconductor/nrfx/blob/master/drivers/src/nrfx_uarte.c#L197
-    if unsafe { core::ptr::read_volatile(txenable_reg) } == 1 {
-        r.tasks_stoptx.write(|w| unsafe { w.bits(1) });
-    }
-
-    // NB Safety: This is taken from Nordic's driver -
-    // https://github.com/NordicSemiconductor/nrfx/blob/master/drivers/src/nrfx_uarte.c#L197
-    if unsafe { core::ptr::read_volatile(rxenable_reg) } == 1 {
-        r.enable.write(|w| w.enable().enabled());
-        r.tasks_stoprx.write(|w| unsafe { w.bits(1) });
-
-        let mut workaround_succeded = false;
-        // The UARTE is able to receive up to four bytes after the STOPRX task has been triggered.
-        // On lowest supported baud rate (1200 baud), with parity bit and two stop bits configured
-        // (resulting in 12 bits per data byte sent), this may take up to 40 ms.
-        for _ in 0..40000 {
-            // NB Safety: This is taken from Nordic's driver -
-            // https://github.com/NordicSemiconductor/nrfx/blob/master/drivers/src/nrfx_uarte.c#L197
-            if unsafe { core::ptr::read_volatile(rxenable_reg) } == 0 {
-                workaround_succeded = true;
-                break;
-            } else {
-                // Need to sleep for 1us here
-            }
-        }
-
-        if !workaround_succeded {
-            panic!("Failed to apply workaround for UART");
-        }
-
-        let errors = r.errorsrc.read().bits();
-        // NB Safety: safe to write back the bits we just read to clear them
-        r.errorsrc.write(|w| unsafe { w.bits(errors) });
-        r.enable.write(|w| w.enable().disabled());
-    }
-}
-
-pub(crate) fn drop_tx_rx(r: &pac::uarte0::RegisterBlock, s: &sealed::State) {
-    if s.tx_rx_refcount.fetch_sub(1, Ordering::Relaxed) == 1 {
-        // Finally we can disable, and we do so for the peripheral
-        // i.e. not just rx concerns.
-        r.enable.write(|w| w.enable().disabled());
-
-        gpio::deconfigure_pin(r.psel.rxd.read().bits());
-        gpio::deconfigure_pin(r.psel.txd.read().bits());
-        gpio::deconfigure_pin(r.psel.rts.read().bits());
-        gpio::deconfigure_pin(r.psel.cts.read().bits());
-
-        trace!("uarte tx and rx drop: done");
-    }
-}
-
-/// 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>,
+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, U: Instance, T: TimerInstance> UarteRxWithIdle<'d, U, T> {
+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
@@ -848,8 +681,8 @@ impl<'d, U: Instance, T: TimerInstance> UarteRxWithIdle<'d, U, T> {
         let ptr = buffer.as_ptr();
         let len = buffer.len();
 
-        let r = U::regs();
-        let s = U::state();
+        let r = T::regs();
+        let s = T::state();
 
         self.ppi_ch1.enable();
 
@@ -904,7 +737,7 @@ impl<'d, U: Instance, T: TimerInstance> UarteRxWithIdle<'d, U, T> {
         let ptr = buffer.as_ptr();
         let len = buffer.len();
 
-        let r = U::regs();
+        let r = T::regs();
 
         self.ppi_ch1.enable();
 
@@ -929,6 +762,75 @@ impl<'d, U: Instance, T: TimerInstance> UarteRxWithIdle<'d, U, T> {
         Ok(n)
     }
 }
+
+#[cfg(not(any(feature = "_nrf9160", feature = "nrf5340")))]
+pub(crate) fn apply_workaround_for_enable_anomaly(_r: &crate::pac::uarte0::RegisterBlock) {
+    // Do nothing
+}
+
+#[cfg(any(feature = "_nrf9160", feature = "nrf5340"))]
+pub(crate) fn apply_workaround_for_enable_anomaly(r: &crate::pac::uarte0::RegisterBlock) {
+    use core::ops::Deref;
+
+    // Apply workaround for anomalies:
+    // - nRF9160 - anomaly 23
+    // - nRF5340 - anomaly 44
+    let rxenable_reg: *const u32 = ((r.deref() as *const _ as usize) + 0x564) as *const u32;
+    let txenable_reg: *const u32 = ((r.deref() as *const _ as usize) + 0x568) as *const u32;
+
+    // NB Safety: This is taken from Nordic's driver -
+    // https://github.com/NordicSemiconductor/nrfx/blob/master/drivers/src/nrfx_uarte.c#L197
+    if unsafe { core::ptr::read_volatile(txenable_reg) } == 1 {
+        r.tasks_stoptx.write(|w| unsafe { w.bits(1) });
+    }
+
+    // NB Safety: This is taken from Nordic's driver -
+    // https://github.com/NordicSemiconductor/nrfx/blob/master/drivers/src/nrfx_uarte.c#L197
+    if unsafe { core::ptr::read_volatile(rxenable_reg) } == 1 {
+        r.enable.write(|w| w.enable().enabled());
+        r.tasks_stoprx.write(|w| unsafe { w.bits(1) });
+
+        let mut workaround_succeded = false;
+        // The UARTE is able to receive up to four bytes after the STOPRX task has been triggered.
+        // On lowest supported baud rate (1200 baud), with parity bit and two stop bits configured
+        // (resulting in 12 bits per data byte sent), this may take up to 40 ms.
+        for _ in 0..40000 {
+            // NB Safety: This is taken from Nordic's driver -
+            // https://github.com/NordicSemiconductor/nrfx/blob/master/drivers/src/nrfx_uarte.c#L197
+            if unsafe { core::ptr::read_volatile(rxenable_reg) } == 0 {
+                workaround_succeded = true;
+                break;
+            } else {
+                // Need to sleep for 1us here
+            }
+        }
+
+        if !workaround_succeded {
+            panic!("Failed to apply workaround for UART");
+        }
+
+        let errors = r.errorsrc.read().bits();
+        // NB Safety: safe to write back the bits we just read to clear them
+        r.errorsrc.write(|w| unsafe { w.bits(errors) });
+        r.enable.write(|w| w.enable().disabled());
+    }
+}
+
+pub(crate) fn drop_tx_rx(r: &pac::uarte0::RegisterBlock, s: &sealed::State) {
+    if s.tx_rx_refcount.fetch_sub(1, Ordering::Relaxed) == 1 {
+        // Finally we can disable, and we do so for the peripheral
+        // i.e. not just rx concerns.
+        r.enable.write(|w| w.enable().disabled());
+
+        gpio::deconfigure_pin(r.psel.rxd.read().bits());
+        gpio::deconfigure_pin(r.psel.txd.read().bits());
+        gpio::deconfigure_pin(r.psel.rts.read().bits());
+        gpio::deconfigure_pin(r.psel.cts.read().bits());
+
+        trace!("uarte tx and rx drop: done");
+    }
+}
+
 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(()) }
-        }
-    }
 }
diff --git a/embassy-rp/src/i2c.rs b/embassy-rp/src/i2c.rs
index 52f910cef..68cfc653e 100644
--- a/embassy-rp/src/i2c.rs
+++ b/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 {
-            _tx_dma,
-            _rx_dma,
-            _dma_buf: [0; 256],
-            phantom: PhantomData,
-        }
+        Self { 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!(I2C1, I2C1_IRQ, 34, 35);
+impl_instance!(I2C0, I2C0_IRQ, set_i2c0, 32, 33);
+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 {}
diff --git a/examples/nrf/src/bin/uart_idle.rs b/examples/nrf/src/bin/uart_idle.rs
index 09ec624c0..6af4f7097 100644
--- a/examples/nrf/src/bin/uart_idle.rs
+++ b/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);
     }
 }
diff --git a/examples/rp/Cargo.toml b/examples/rp/Cargo.toml
index a5af8b2f0..747dde515 100644
--- a/examples/rp/Cargo.toml
+++ b/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
diff --git a/examples/rp/src/bin/i2c_async.rs b/examples/rp/src/bin/i2c_async.rs
new file mode 100644
index 000000000..d1a2e3cd7
--- /dev/null
+++ b/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;
+    }
+}