feat: implement wasi:cli

Signed-off-by: Roman Volosatovs <rvolosatovs@riseup.net>

Author: rvolosatovs

crates/wasi/src/p3/bindings.rs

@@ -49,6 +49,9 @@
 //!     async: {
 //!         only_imports: [
 //!             "example:wasi/custom-host#my-custom-function",
+//!             "wasi:cli/stdin@0.3.0#get-stdin",
+//!             "wasi:cli/stdout@0.3.0#set-stdout",
+//!             "wasi:cli/stderr@0.3.0#set-stderr",
 //!             "wasi:clocks/monotonic-clock@0.3.0#wait-for",
 //!             "wasi:clocks/monotonic-clock@0.3.0#wait-until",
 //!             "wasi:filesystem/types@0.3.0#[method]descriptor.read-via-stream",
@@ -153,6 +156,9 @@ mod generated {
         concurrent_imports: true,
         async: {
             only_imports: [
+                "wasi:cli/stdin@0.3.0#get-stdin",
+                "wasi:cli/stdout@0.3.0#set-stdout",
+                "wasi:cli/stderr@0.3.0#set-stderr",
                 "wasi:clocks/monotonic-clock@0.3.0#wait-for",
                 "wasi:clocks/monotonic-clock@0.3.0#wait-until",
                 "wasi:filesystem/types@0.3.0#[method]descriptor.read-via-stream",
@@ -193,6 +199,8 @@ mod generated {
             ],
         },
         with: {
+            "wasi:cli/terminal-input/terminal-input": crate::p3::cli::TerminalInput,
+            "wasi:cli/terminal-output/terminal-output": crate::p3::cli::TerminalOutput,
             "wasi:filesystem/types/descriptor": crate::p3::filesystem::Descriptor,
             "wasi:sockets/types/tcp-socket": crate::p3::sockets::tcp::TcpSocket,
             "wasi:sockets/types/udp-socket": crate::p3::sockets::udp::UdpSocket,

crates/wasi/src/p3/cli/host.rs

@@ -1,16 +1,96 @@
-#![allow(unused)] // TODO: remove
+use anyhow::{anyhow, Context as _};
+use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};
+use wasmtime::component::{stream, Accessor, AccessorTask, Resource, StreamReader, StreamWriter};
 
-use anyhow::Context as _;
-use wasmtime::component::Resource;
-
-use crate::p3::bindings::cli::terminal_input::TerminalInput;
-use crate::p3::bindings::cli::terminal_output::TerminalOutput;
 use crate::p3::bindings::cli::{
     environment, exit, stderr, stdin, stdout, terminal_input, terminal_output, terminal_stderr,
     terminal_stdin, terminal_stdout,
 };
-use crate::p3::cli::{WasiCliImpl, WasiCliView};
-use crate::p3::ResourceView as _;
+use crate::p3::cli::{I32Exit, TerminalInput, TerminalOutput, WasiCliImpl, WasiCliView};
+use crate::p3::{next_item, ResourceView as _};
+
+struct InputTask<T> {
+    input: T,
+    tx: StreamWriter<u8>,
+}
+
+impl<T, U, V> AccessorTask<T, U, wasmtime::Result<()>> for InputTask<V>
+where
+    V: AsyncRead + Send + Sync + Unpin + 'static,
+{
+    async fn run(mut self, store: &mut Accessor<T, U>) -> wasmtime::Result<()> {
+        let mut tx = self.tx;
+        loop {
+            let mut buf = vec![0; 8096];
+            match self.input.read(&mut buf).await {
+                Ok(0) => {
+                    store.with(|mut view| tx.close(&mut view).context("failed to close stream"))?;
+                    return Ok(());
+                }
+                Ok(n) => {
+                    buf.truncate(n);
+                    let fut =
+                        store.with(|view| tx.write(view, buf).context("failed to send chunk"))?;
+                    let Some(tail) = fut.into_future().await else {
+                        break Ok(());
+                    };
+                    tx = tail;
+                }
+                Err(_err) => {
+                    // TODO: Close the stream with the real error context
+                    store.with(|mut view| {
+                        tx.close_with_error(&mut view, 0)
+                            .context("failed to close stream")
+                    })?;
+                    return Ok(());
+                }
+            }
+        }
+    }
+}
+
+struct OutputTask<T> {
+    output: T,
+    data: StreamReader<u8>,
+}
+
+impl<T, U, V> AccessorTask<T, U, wasmtime::Result<()>> for OutputTask<V>
+where
+    V: AsyncWrite + Send + Sync + Unpin + 'static,
+{
+    async fn run(mut self, store: &mut Accessor<T, U>) -> wasmtime::Result<()> {
+        let fut = store.with(|mut view| {
+            self.data
+                .read(&mut view)
+                .context("failed to read from stream")
+        })?;
+        let mut fut = fut.into_future();
+        'outer: loop {
+            let Some((tail, buf)) = fut.await else {
+                return Ok(());
+            };
+            let mut buf = buf.as_slice();
+            loop {
+                match self.output.write(&buf).await {
+                    Ok(n) => {
+                        if n == buf.len() {
+                            fut = next_item(store, tail)?;
+                            continue 'outer;
+                        } else {
+                            buf = &buf[n..];
+                        }
+                    }
+                    Err(_err) => {
+                        // TODO: Report the error to the guest
+                        return store.with(|mut view| {
+                            tail.close(&mut view).context("failed to close stream")
+                        });
+                    }
+                }
+            }
+        }
+    }
+}
 
 impl<T> terminal_input::Host for WasiCliImpl<T> where T: WasiCliView {}
 impl<T> terminal_output::Host for WasiCliImpl<T> where T: WasiCliView {}
@@ -44,7 +124,15 @@ where
     T: WasiCliView,
 {
     fn get_terminal_stdin(&mut self) -> wasmtime::Result<Option<Resource<TerminalInput>>> {
-        todo!()
+        if self.cli().stdin.is_terminal() {
+            let fd = self
+                .table()
+                .push(TerminalInput)
+                .context("failed to push terminal resource to table")?;
+            Ok(Some(fd))
+        } else {
+            Ok(None)
+        }
     }
 }
 
@@ -53,7 +141,15 @@ where
     T: WasiCliView,
 {
     fn get_terminal_stdout(&mut self) -> wasmtime::Result<Option<Resource<TerminalOutput>>> {
-        todo!()
+        if self.cli().stdout.is_terminal() {
+            let fd = self
+                .table()
+                .push(TerminalOutput)
+                .context("failed to push terminal resource to table")?;
+            Ok(Some(fd))
+        } else {
+            Ok(None)
+        }
     }
 }
 
@@ -62,34 +158,69 @@ where
     T: WasiCliView,
 {
     fn get_terminal_stderr(&mut self) -> wasmtime::Result<Option<Resource<TerminalOutput>>> {
-        todo!()
+        if self.cli().stderr.is_terminal() {
+            let fd = self
+                .table()
+                .push(TerminalOutput)
+                .context("failed to push terminal resource to table")?;
+            Ok(Some(fd))
+        } else {
+            Ok(None)
+        }
     }
 }
 
 impl<T> stdin::Host for WasiCliImpl<T>
 where
     T: WasiCliView,
 {
-    fn get_stdin(&mut self) -> wasmtime::Result<wasmtime::component::StreamReader<u8>> {
-        todo!()
+    async fn get_stdin<U: 'static>(
+        store: &mut Accessor<U, Self>,
+    ) -> wasmtime::Result<StreamReader<u8>> {
+        store.with(|mut view| {
+            let (tx, rx) = stream(&mut view).context("failed to create stream")?;
+            let stdin = view.cli().stdin.reader();
+            view.spawn(InputTask { input: stdin, tx });
+            Ok(rx)
+        })
     }
 }
 
 impl<T> stdout::Host for WasiCliImpl<T>
 where
     T: WasiCliView,
 {
-    fn set_stdout(&mut self, data: wasmtime::component::StreamReader<u8>) -> wasmtime::Result<()> {
-        todo!()
+    async fn set_stdout<U: 'static>(
+        store: &mut Accessor<U, Self>,
+        data: StreamReader<u8>,
+    ) -> wasmtime::Result<()> {
+        store.with(|mut view| {
+            let stdout = view.cli().stdout.writer();
+            view.spawn(OutputTask {
+                output: stdout,
+                data,
+            });
+            Ok(())
+        })
     }
 }
 
 impl<T> stderr::Host for WasiCliImpl<T>
 where
     T: WasiCliView,
 {
-    fn set_stderr(&mut self, data: wasmtime::component::StreamReader<u8>) -> wasmtime::Result<()> {
-        todo!()
+    async fn set_stderr<U: 'static>(
+        store: &mut Accessor<U, Self>,
+        data: StreamReader<u8>,
+    ) -> wasmtime::Result<()> {
+        store.with(|mut view| {
+            let stderr = view.cli().stderr.writer();
+            view.spawn(OutputTask {
+                output: stderr,
+                data,
+            });
+            Ok(())
+        })
     }
 }
 
@@ -115,10 +246,14 @@ where
     T: WasiCliView,
 {
     fn exit(&mut self, status: Result<(), ()>) -> wasmtime::Result<()> {
-        todo!()
+        let status = match status {
+            Ok(()) => 0,
+            Err(()) => 1,
+        };
+        Err(anyhow!(I32Exit(status)))
     }
 
     fn exit_with_code(&mut self, status_code: u8) -> wasmtime::Result<()> {
-        todo!()
+        Err(anyhow!(I32Exit(status_code.into())))
     }
 }

crates/wasi/src/p3/cli/mod.rs

@@ -1,5 +1,8 @@
 mod host;
 
+use core::fmt;
+
+use tokio::io::{empty, AsyncRead, AsyncWrite, Empty};
 use wasmtime::component::{Linker, ResourceTable};
 
 use crate::p3::ResourceView;
@@ -33,6 +36,9 @@ pub struct WasiCliCtx {
     pub environment: Vec<(String, String)>,
     pub arguments: Vec<String>,
     pub initial_cwd: Option<String>,
+    pub stdin: Box<dyn InputStream + Send>,
+    pub stdout: Box<dyn OutputStream + Send>,
+    pub stderr: Box<dyn OutputStream + Send>,
 }
 
 impl Default for WasiCliCtx {
@@ -41,6 +47,9 @@ impl Default for WasiCliCtx {
             environment: Vec::default(),
             arguments: Vec::default(),
             initial_cwd: None,
+            stdin: Box::new(empty()),
+            stdout: Box::new(empty()),
+            stderr: Box::new(empty()),
         }
     }
 }
@@ -137,3 +146,70 @@ where
 {
     val
 }
+
+/// An error returned from the `proc_exit` host syscall.
+///
+/// Embedders can test if an error returned from wasm is this error, in which
+/// case it may signal a non-fatal trap.
+#[derive(Debug)]
+pub struct I32Exit(pub i32);
+
+impl fmt::Display for I32Exit {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Exited with i32 exit status {}", self.0)
+    }
+}
+
+impl std::error::Error for I32Exit {}
+
+pub struct TerminalInput;
+pub struct TerminalOutput;
+
+pub trait IsTerminal {
+    /// Returns whether this stream is backed by a TTY.
+    fn is_terminal(&self) -> bool;
+}
+
+impl IsTerminal for Empty {
+    fn is_terminal(&self) -> bool {
+        false
+    }
+}
+
+impl InputStream for Empty {
+    fn reader(&self) -> Box<dyn AsyncRead + Send + Sync + Unpin> {
+        Box::new(empty())
+    }
+}
+
+impl OutputStream for Empty {
+    fn writer(&self) -> Box<dyn AsyncWrite + Send + Sync + Unpin> {
+        Box::new(empty())
+    }
+}
+
+impl IsTerminal for std::io::Empty {
+    fn is_terminal(&self) -> bool {
+        false
+    }
+}
+
+impl InputStream for std::io::Empty {
+    fn reader(&self) -> Box<dyn AsyncRead + Send + Sync + Unpin> {
+        Box::new(empty())
+    }
+}
+
+impl OutputStream for std::io::Empty {
+    fn writer(&self) -> Box<dyn AsyncWrite + Send + Sync + Unpin> {
+        Box::new(empty())
+    }
+}
+
+pub trait InputStream: IsTerminal {
+    fn reader(&self) -> Box<dyn AsyncRead + Send + Sync + Unpin>;
+}
+
+pub trait OutputStream: IsTerminal {
+    fn writer(&self) -> Box<dyn AsyncWrite + Send + Sync + Unpin>;
+}