Async version?

Is an async version of this library planned? Have you used it in async applications? If yes, how? The embassy-net tcp socket only implements embedded_io_async traits.

No, currently there is no plans to support async IO.
Take a look at similar issue: #13

You can add RX and TX buffers and fill RX in your async code. Then in some dedicated "thread" you will take all bytes one by one from RX buffer, pass to cli. In cli pass writer that will add bytes to TX buffer. And in your async code take bytes from TX buffer and write to your IO backend. Of course you will have to handle overflows somehow.

Approach with RX/TX buffers could be supported by library one day but currently I don't have resources for that. But PRs are welcome

What writer do you use for writing to a buffer? Do you implement it for every async project? Is it in a crates.io crate?

I don't use it in async projects, so I don't have a writer. As I said, just create two buffers (RX and TX) behind a mutex. You can do something like this for TX (and for RX everything is similar). Mostly generated by chatgpt, but you get the idea:

use embedded_io::Write as EmbeddedWrite; // Import the embedded_io Write trait
use heapless::spsc::Queue;
use critical_section::Mutex;

pub struct BufferWriter<const BUF_SIZE: usize> {
    queue: Mutex<Queue<u8, BUF_SIZE>>, // Thread-safe queue for TX buffer
}

impl<const BUF_SIZE: usize> BufferWriter<BUF_SIZE> {
    pub const fn new() -> Self {
        Self {
            queue: Mutex::new(Queue::new()),
        }
    }

    /// Push a byte into the buffer. Returns `Err` if the buffer is full.
    pub fn push(&self, byte: u8) -> Result<(), ()> {
        critical_section::with(|cs| {
            let mut queue = self.queue.borrow_ref_mut(cs);
            queue.enqueue(byte).map_err(|_| ()) // Return error if the buffer is full
        })
    }

    /// Pop a byte from the buffer. Returns `None` if the buffer is empty.
    pub fn pop(&self) -> Option<u8> {
        critical_section::with(|cs| {
            let mut queue = self.queue.borrow_ref_mut(cs);
            queue.dequeue()
        })
    }

    /// Check if the buffer is empty.
    pub fn is_empty(&self) -> bool {
        critical_section::with(|cs| {
            let queue = self.queue.borrow_ref(cs);
            queue.is_empty()
        })
    }

    /// Check if the buffer is full.
    pub fn is_full(&self) -> bool {
        critical_section::with(|cs| {
            let queue = self.queue.borrow_ref(cs);
            queue.is_full()
        })
    }
}


/// Implement the embedded_io::Write trait
impl<const BUF_SIZE: usize> EmbeddedWrite for BufferWriter<BUF_SIZE> {
    type Error = (); // Define the error type (simple `()` for now)

    fn write(&mut self, data: &[u8]) -> Result<usize, Self::Error> {
        let mut written = 0;

        for &byte in data {
            if self.push(byte).is_err() {
                break; // Stop writing if the buffer is full
            }
            written += 1;
        }

        Ok(written) // Return the number of bytes successfully written
    }

    fn flush(&mut self) -> Result<(), Self::Error> {
        // For this implementation, flushing is a no-op
        Ok(())
    }
}

This writer is passed to cli. And somewhere in your code you should do writer.pop() and write received byte to your async IO.

So you just implement it every time, right?

No, as I said, I'm not using async in my embedded projects. And in sync projects writer is just a small wrapper around a buffer. So there's nothing to extract to separate crate.

Can you show me an example? I don't get how I'm supposed to flush the buffer out if I move the writer to the cli when I build it.

Take a look at api of heapless queue. There is a method for splitting:
https://docs.rs/heapless/latest/heapless/spsc/index.html

writer then will have consumer and you use producer to take out bytes that were written

There's also BBQueue which is a queue that can be filled/drained both sync and async, so you can use that as the output of your CLI and then drain it in an async worker.

The issue why async support would be useful is not that the read/write needs to be async. The biggest issue is that process_byte is not async. So we cannot actually perform async tasks from within shell commands.

Using Embassy, I've come up with a solution based on pipes that implements both the buffer-based console IO concept and allows commands to run in an async context. It depends on being able to run embassy executors at multiple priorities, (with the console itself running at lower priority than everything else) and ends up blocking all progress on the lower-priority executor while a command is executing.

For input to and output from the console, define two pipes with somewhat arbitrary capacity that are analogous to stdin and stdout:

use embassy_sync::pipe;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;

type ConsolePipe = pipe::Pipe<CriticalSectionRawMutex, 64>;
static CONSOLE_IN_PIPE: ConsolePipe = pipe::Pipe::new();
static CONSOLE_OUT_PIPE: ConsolePipe = pipe::Pipe::new();

I have two executors, one in thread mode and one in interrupt mode (I'm targeting RP2350, but most systems should have similar capability):

use embassy_executor::{Executor, InterruptExecutor, SendSpawner};
use embassy_rp::interrupt;
use static_cell::StaticCell;

static EXECUTOR_INT: InterrruptExecutor = InterruptExecutor::new();
static EXECUTOR_IRQ: interrupt::Interrupt = interrupt::SWI_IRQ_0;

#[interrupt]
fn SWI_IRQ_0() {
    unsafe {
        EXECUTOR_INT.on_interrupt();
    }
}

#[entry]
fn main() -> ! {
    let int_spawner = EXECUTOR_INT.start(EXECUTOR_IRQ);

    static EXECUTOR_THREAD: StaticCell<Executor> = StaticCell::new();
    EXECUTOR_THREAD
        .init_with(Executor::new)
        .run(move |thread_spawner| {
            // ...
        });
}

A task running on the interrupt executor shuttles bytes between external IO and the console pipes. In my application it's USB-CDC, but could be a regular UART or anything else.

#[embassy_executor::task]
async fn run_console_io() -> ! {
    use embassy_futures::select::select;

    loop {
        select(
            async {
                // Buffer size is arbitrary, can be adjusted depending on your needs
                let mut buf = [0u8; 64];  
                loop {
                    let n = todo!("read from hardware into buf");
                    CONSOLE_IN_PIPE.write_all(&buf[..n]).await;
                }
            },
            async {
                // Also arbitrary size
                let mut buf = [0u8; 64];
                loop {
                    let n = CONSOLE_OUT_PIPE.read(&mut buf).await;
                    todo!("write buf[..n] to hardware");
                }
            }
        ).await;
        // If hardware I/O is fallible, handle errors here or inside the inner loops
    }
}

A small adapter to convert the async pipe to blocking I/O as expected by the CLI:

use core::sync::atomic::{AtomicUsize, Ordering};

struct SynchronousPipeBridge<'a>(
    pub &'static ConsolePipe,
    &'a AtomicUsize,
);

use core::convert::Infallible;
use embedded_io::{ErrorType as BlockingErrorType, Write as BlockingWrite};

impl<'a> BlockingErrorType for SynchronousPipeBridge<'a> {
    type Error = Infallible;
}

impl<'a> BlockingWrite for SynchronousPipeBridge<'a> {
    fn write(&mut self, buf: &[u8]) -> Result<usize, Infallible> {
        match self.0.try_write(buf) {
            Ok(n) => Ok(n),
            Err(_) => {
                // No space available in the pipe. Silently drop the write but record how many bytes have been
                // lost, so we can report the overrun.
                self.1.fetch_add(buf.len(), Ordering::AcqRel);
                Ok(buf.len())
            }
        }
    }

    fn flush(&mut self) -> Result<(), Infallible> { Ok(()) }
}

In a task that runs in thread mode, drive the console:

#[embassy_executor::task]
async fn console_task(spawner: SendSpawner) {
    let r = &CONSOLE_IN_PIPE;
    let dropped_count = AtomicUsize::new(0);
    let w = SynchronousPipeBridge::new(&CONSOLE_OUT_PIPE, &dropped_count);

    let mut cli = CliBuilder::default()
        .writer(w)
        .build()
        .unwrap();

    loop {
        let dropped = dropped_count.swap(0, Ordering::AcqRel);
        if dropped > 0 {
            // Write a warning to logs or something
        }

        let mut c = 0u8;
        r.read_exact(slice::from_mut(&c)).await.expect("Pipe read is infallible");

        let mut processor = Command::processor(|_cli, cmd| {
            // See below
            handle_command(&cmd, spawner);
            Ok(())
        }
        cli.process_byte::<Command<'_>, _>(c, &mut processor).expect("pipe writes are infallible");
    }
}

The final piece is bridging from the synchronous command processor into an async task, with a little bit of unsafe to avoid needing to copy things into the task. In the interest of efficiency I'm using wfe/sev because I'm not aiming for great portability beyond Cortex-M and they allow the system to idle in a lower-power mode, but they can be omitted without harm.

use core::sync::atomic::AtomicBool;
use cortex_m::{
    asm::{sev, wfe},
    peripheral::{SCB, scb}
};

fn handle_command(command: &Command<'_>, spawner: SendSpawner) {
    static DONE: AtomicBool = AtomicBool::new(false);
    fn mark_done() {
        DONE.store(true, Ordering::Release);
        sev();
    }
    // Before each call, the task is not done
    DONE.store(false, Ordering::Release);

    // Launder parameter lifetime. This is safe as long as the task never refers to its
    // reference parameters after calling mark_done().
    let command = unsafe {
        core::mem::transmute::<&'_ Command<'_>, &'static Command<'static>>(command)
    };

    spawner.spawn(
        handle_command_task(&CONSOLE_IN_PIPE, command, mark_done)
            .expect("handle_command is not reentrant")
    );
    
    // Wait for task to complete. SendSpawner always executes tasks on an interrupt executor,
    // so verifying we're currently in thread mode ensures the task is running at higher priority
    // and this won't deadlock.
    assert_eq!(SCB::vect_active(), scb::VectActive::ThreadMode);
    while !DONE.load(Ordering::Acquire) {
        wfe();
    }
}

#[embassy_executor::task]
async fn handle_command_task(
    writer: &'static ConsolePipe,
    command: &'static Command<'static>,
    mark_done: fn(),
) {
    scopeguard::defer! {
        mark_done();
    }

    todo!("process command, write output to writer");
}

The portable version of this polls DONE in a tight loop, but you could insert some other useful work or delay there as well.

async closures have been stable for a while, could these be used instead in the processor function?

@tommasoclini

@tommasoclini then the processor function itself would have to become async, which would be quite a rewrite

Sooner or later I think async is the way to go anyway

I made a draft PR using async closures, changed like 20 lines of code in total, desktop example works, other existing blocking ones obviously don't.
#28

@tari

@tari you can take a look at #30 if you need async functionality.

Navigation Menu

Repository navigation

Async version? #16

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Async version? #16

Description

Activity

funbiscuit commented on Dec 9, 2024

tommasoclini commented on Dec 9, 2024

funbiscuit commented on Dec 10, 2024

tommasoclini commented on Dec 10, 2024

funbiscuit commented on Dec 10, 2024

tommasoclini commented on Dec 10, 2024

funbiscuit commented on Dec 10, 2024

Finomnis commented on May 14, 2026

Finomnis commented on May 19, 2026

tari commented on Jun 10, 2026

tommasoclini commented on Jun 10, 2026

Finomnis commented on Jun 10, 2026

tommasoclini commented on Jun 10, 2026

tommasoclini commented on Jun 10, 2026

tommasoclini commented on Jul 31, 2026

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions