Async uart

Author: tylerwhall

rust/sysroot-stage2/Cargo.lock

@@ -1,11 +1,11 @@
 extern crate alloc;
 extern crate zephyr_core;
 
-use core::cell::{UnsafeCell, RefCell};
+use alloc::sync::{Arc, Weak};
+use core::cell::{RefCell, UnsafeCell};
+use core::marker::PhantomData;
 use core::pin::Pin;
 use core::task::{Context, Poll, Waker};
-use core::marker::PhantomData;
-use alloc::sync::{Arc, Weak};
 
 use futures::future::Future;
 use futures::stream::Stream;
@@ -71,6 +71,14 @@ thread_local! {
     static REACTOR: RefCell<Option<Reactor>> = RefCell::new(None);
 }
 
+#[inline(never)]
+pub fn current_reactor_register(signal: &'static impl PollableKobj, context: &mut Context) {
+    match REACTOR.try_with(|r| r.borrow_mut().as_mut().map(|r| r.register(signal, context))) {
+        Ok(None) | Err(_) => panic!("register with no reactor"),
+        Ok(Some(_)) => (),
+    }
+}
+
 struct Task {
     future: UnsafeCell<Pin<Box<dyn Future<Output = ()>>>>,
     executor: ExecutorHandle,
@@ -132,7 +140,10 @@ impl ExecutorInner {
 // that calls poll being not Send or Sync. Since we're not requiring spawned futures to be Send or
 // Sync and Executor is the effective owner, add a PhantomData here as if we directly own a Future
 // that is not explicitly Send or Sync.
-pub struct Executor(Arc<Mutex<'static, ExecutorInner>>, PhantomData<Future<Output = ()>>);
+pub struct Executor(
+    Arc<Mutex<'static, ExecutorInner>>,
+    PhantomData<Future<Output = ()>>,
+);
 #[derive(Clone)]
 struct ExecutorHandle(Weak<Mutex<'static, ExecutorInner>>);
 

rust/zephyr-futures/src/lib.rs

@@ -5,3 +5,5 @@ authors = ["Tyler Hall <[email protected]>"]
 edition = "2018"
 
 [dependencies]
+futures-preview = "=0.3.0-alpha.17"
+zephyr-futures = { path = "../zephyr-futures" }

rust/zephyr-uart-buffered/Cargo.toml

@@ -0,0 +1,104 @@
+use core::pin::Pin;
+use core::task::{Context, Poll};
+
+use futures::io::{AsyncRead, AsyncWrite, Error, Initializer};
+
+use zephyr_core::context::Any as C;
+use zephyr_core::poll::Signal;
+use zephyr_futures::current_reactor_register;
+
+use super::{UartBufferedRx, UartBufferedTx};
+
+pub struct UartBufferedRxAsync {
+    uart: UartBufferedRx,
+}
+
+impl UartBufferedRxAsync {
+    pub fn new(uart: UartBufferedRx) -> Self {
+        UartBufferedRxAsync { uart }
+    }
+}
+
+impl AsyncRead for UartBufferedRxAsync {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut [u8],
+    ) -> Poll<Result<usize, Error>> {
+        let s = self.get_mut();
+        let uart = &mut s.uart;
+
+        if let Some(len) = uart.read_nb(buf) {
+            return Poll::Ready(Ok(len));
+        }
+
+        // Need to register for readiness on the signal. We wait to clear the
+        // signal until after the uart is not ready so that we don't make a
+        // redundant system call before each read attempt, e.g. if the client is
+        // reading one byte at a time and there are several buffered.
+        // Because the signal is edge triggered, resetting here could clear an
+        // event that happened since the poll above. So poll one more time.
+        let signal = uart.get_signal();
+        signal.reset::<C>();
+        current_reactor_register(signal, cx);
+
+        if let Some(len) = uart.read_nb(buf) {
+            return Poll::Ready(Ok(len));
+        }
+
+        Poll::Pending
+    }
+
+    unsafe fn initializer(&self) -> Initializer {
+        Initializer::nop()
+    }
+}
+
+pub struct UartBufferedTxAsync {
+    uart: UartBufferedTx,
+}
+
+impl UartBufferedTxAsync {
+    pub fn new(uart: UartBufferedTx) -> Self {
+        UartBufferedTxAsync { uart }
+    }
+}
+
+impl AsyncWrite for UartBufferedTxAsync {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &[u8],
+    ) -> Poll<Result<usize, Error>> {
+        let s = self.get_mut();
+        let uart = &mut s.uart;
+
+        if let Some(len) = uart.write_nb(buf) {
+            return Poll::Ready(Ok(len));
+        }
+
+        // Need to register for readiness on the signal. We wait to clear the
+        // signal until after the uart is not ready so that we don't make a
+        // redundant system call before each write attempt, e.g. if the client
+        // is writing one byte at a time and there are several buffered.
+        // Because the signal is edge triggered, resetting here could clear an
+        // event that happened since the poll above. So poll one more time.
+        let signal = uart.get_signal();
+        signal.reset::<C>();
+        current_reactor_register(signal, cx);
+
+        if let Some(len) = uart.write_nb(buf) {
+            return Poll::Ready(Ok(len));
+        }
+
+        Poll::Pending
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context) -> Poll<Result<(), Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_close(self: Pin<&mut Self>, _cx: &mut Context) -> Poll<Result<(), Error>> {
+        Poll::Ready(Ok(()))
+    }
+}

rust/zephyr-uart-buffered/src/futures.rs

@@ -5,11 +5,17 @@ use zephyr_core::poll::KPollSignal;
 use zephyr_core::NegErr;
 use zephyr_sys::raw::{uart_buffered_rx_handle, uart_buffered_tx_handle};
 
+mod futures;
+
+pub use crate::futures::{UartBufferedRxAsync, UartBufferedTxAsync};
+
 pub struct UartBufferedRx {
     handle: uart_buffered_rx_handle,
 }
 
 impl UartBufferedRx {
+    /// Unsafe because this is passed from C and caller must guarantee there is
+    /// only one instance created per handle.
     pub unsafe fn new(handle: uart_buffered_rx_handle) -> Self {
         UartBufferedRx { handle }
     }
@@ -44,13 +50,19 @@ impl UartBufferedRx {
     pub fn get_signal(&self) -> &'static KPollSignal {
         unsafe { &*self.handle.fifo.signal }
     }
+
+    pub fn into_async(self) -> UartBufferedRxAsync {
+        UartBufferedRxAsync::new(self)
+    }
 }
 
 pub struct UartBufferedTx {
     handle: uart_buffered_tx_handle,
 }
 
 impl UartBufferedTx {
+    /// Unsafe because this is passed from C and caller must guarantee there is
+    /// only one instance created per handle.
     pub unsafe fn new(handle: uart_buffered_tx_handle) -> Self {
         UartBufferedTx { handle }
     }
@@ -85,4 +97,8 @@ impl UartBufferedTx {
     pub fn get_signal(&self) -> &'static KPollSignal {
         unsafe { &*self.handle.fifo.signal }
     }
+
+    pub fn into_async(self) -> UartBufferedTxAsync {
+        UartBufferedTxAsync::new(self)
+    }
 }

rust/zephyr-uart-buffered/src/lib.rs

@@ -1,22 +1,41 @@
+#![feature(async_await)]
+
+extern crate futures;
 extern crate zephyr;
 extern crate zephyr_sys;
 extern crate zephyr_uart_buffered;
+extern crate zephyr_macros;
+extern crate zephyr_futures;
+
+use futures::{StreamExt, AsyncBufReadExt, AsyncWriteExt};
+use futures::io::BufReader;
 
 use zephyr_sys::raw::{uart_buffered_rx_handle, uart_buffered_tx_handle};
 use zephyr_uart_buffered::{UartBufferedRx, UartBufferedTx};
+use zephyr_futures::Executor;
+
+zephyr_macros::k_mutex_define!(EXECUTOR_MUTEX);
+
+async fn echo<R: AsyncBufReadExt + Unpin, W: AsyncWriteExt + Unpin>(rx: R, mut tx: W) {
+    let mut lines = rx.lines();
+    while let Some(line) = lines.next().await {
+        let line = line.unwrap();
+        println!("got line: {}", line);
+        let line = line.into_bytes();
+        tx.write_all(&line).await.unwrap();
+    }
+}
 
 #[no_mangle]
 pub extern "C" fn rust_main(rx: uart_buffered_rx_handle, tx: uart_buffered_tx_handle) {
-    let mut rx = unsafe { UartBufferedRx::new(rx) };
-    let mut tx = unsafe { UartBufferedTx::new(tx) };
+    use zephyr::context::Kernel as C;
 
-    loop {
-        const X: &str = "hello\n";
-        tx.write(X.as_bytes());
+    let rx = unsafe { UartBufferedRx::new(rx) };
+    let rx = BufReader::with_capacity(32, rx.into_async());
 
-        let mut buf = [0u8; 32];
-        let n = rx.read(&mut buf);
-        let mystr = std::str::from_utf8(&buf[..n]).unwrap();
-        println!("read: n={} mystr={}", n, mystr);
-    }
+    let tx = unsafe { UartBufferedTx::new(tx) }.into_async();
+
+    let mut executor = unsafe { Executor::new(&EXECUTOR_MUTEX) };
+    executor.spawn(C, echo(rx, tx));
+    executor.run::<C>();
 }