wasmtime_wasi/
cli.rs

1use crate::p2;
2use std::pin::Pin;
3use std::sync::Arc;
4use tokio::io::{AsyncRead, AsyncWrite, empty};
5use wasmtime::component::{HasData, ResourceTable};
6use wasmtime_wasi_io::streams::{InputStream, OutputStream, StreamError};
7
8mod empty;
9mod file;
10mod locked_async;
11mod mem;
12mod stdout;
13mod worker_thread_stdin;
14
15pub use self::file::{InputFile, OutputFile};
16pub use self::locked_async::{AsyncStdinStream, AsyncStdoutStream};
17
18/// Convert a host `io::Error` into a `StreamError`, matching the error-code
19/// recovery that wasip1 performs via `filesystem::ErrorCode::from`.
20///
21/// * `BrokenPipe` is mapped to `StreamError::Closed` so that downstream
22///   consumers (e.g. wasi-libc) can recover `EPIPE` rather than falling back
23///   to a generic `EIO`.
24///
25/// * All other errors (including `IsADirectory`, permission errors, etc.) are
26///   preserved as `LastOperationFailed` with the original `std::io::Error`
27///   intact. This allows guests to recover the specific error code via the
28///   `wasi:filesystem/types#filesystem-error-code` function, which downcasts
29///   the error back to `std::io::Error` and maps it through
30///   `ErrorCode::from`.
31fn stream_error_from(e: std::io::Error) -> StreamError {
32    if e.kind() == std::io::ErrorKind::BrokenPipe {
33        StreamError::Closed
34    } else {
35        StreamError::LastOperationFailed(e.into())
36    }
37}
38
39// Convenience reexport for stdio types so tokio doesn't have to be imported
40// itself.
41#[doc(no_inline)]
42pub use tokio::io::{Stderr, Stdin, Stdout, stderr, stdin, stdout};
43
44/// A helper struct which implements [`HasData`] for the `wasi:cli` APIs.
45///
46/// This can be useful when directly calling `add_to_linker` functions directly,
47/// such as [`wasmtime_wasi::p2::bindings::cli::environment::add_to_linker`] as
48/// the `D` type parameter. See [`HasData`] for more information about the type
49/// parameter's purpose.
50///
51/// When using this type you can skip the [`WasiCliView`] trait, for
52/// example.
53///
54/// [`wasmtime_wasi::p2::bindings::cli::environment::add_to_linker`]: crate::p2::bindings::cli::environment::add_to_linker
55///
56/// # Examples
57///
58/// ```
59/// use wasmtime::component::{Linker, ResourceTable};
60/// use wasmtime::{Engine, Result};
61/// use wasmtime_wasi::cli::*;
62///
63/// struct MyStoreState {
64///     table: ResourceTable,
65///     cli: WasiCliCtx,
66/// }
67///
68/// fn main() -> Result<()> {
69///     let engine = Engine::default();
70///     let mut linker = Linker::new(&engine);
71///
72///     wasmtime_wasi::p2::bindings::cli::environment::add_to_linker::<MyStoreState, WasiCli>(
73///         &mut linker,
74///         |state| WasiCliCtxView {
75///             table: &mut state.table,
76///             ctx: &mut state.cli,
77///         },
78///     )?;
79///     Ok(())
80/// }
81/// ```
82pub struct WasiCli;
83
84impl HasData for WasiCli {
85    type Data<'a> = WasiCliCtxView<'a>;
86}
87
88/// Provides a "view" of `wasi:cli`-related context used to implement host
89/// traits.
90pub trait WasiCliView: Send {
91    fn cli(&mut self) -> WasiCliCtxView<'_>;
92}
93
94pub struct WasiCliCtxView<'a> {
95    pub ctx: &'a mut WasiCliCtx,
96    pub table: &'a mut ResourceTable,
97}
98
99pub struct WasiCliCtx {
100    pub(crate) environment: Vec<(String, String)>,
101    pub(crate) arguments: Vec<String>,
102    pub(crate) initial_cwd: Option<String>,
103    pub(crate) stdin: Box<dyn StdinStream>,
104    pub(crate) stdout: Box<dyn StdoutStream>,
105    pub(crate) stderr: Box<dyn StdoutStream>,
106}
107
108impl Default for WasiCliCtx {
109    fn default() -> WasiCliCtx {
110        WasiCliCtx {
111            environment: Vec::new(),
112            arguments: Vec::new(),
113            initial_cwd: None,
114            stdin: Box::new(empty()),
115            stdout: Box::new(empty()),
116            stderr: Box::new(empty()),
117        }
118    }
119}
120
121pub trait IsTerminal {
122    /// Returns whether this stream is backed by a TTY.
123    fn is_terminal(&self) -> bool;
124}
125
126/// A trait used to represent the standard input to a guest program.
127///
128/// Note that there are many built-in implementations of this trait for various
129/// types such as [`tokio::io::Stdin`], [`tokio::io::Empty`], and
130/// [`p2::pipe::MemoryInputPipe`].
131pub trait StdinStream: IsTerminal + Send {
132    /// Creates a fresh stream which is reading stdin.
133    ///
134    /// Note that the returned stream must share state with all other streams
135    /// previously created. Guests may create multiple handles to the same stdin
136    /// and they should all be synchronized in their progress through the
137    /// program's input.
138    ///
139    /// Note that this means that if one handle becomes ready for reading they
140    /// all become ready for reading. Subsequently if one is read from it may
141    /// mean that all the others are no longer ready for reading. This is
142    /// basically a consequence of the way the WIT APIs are designed today.
143    fn async_stream(&self) -> Box<dyn AsyncRead + Send + Sync>;
144
145    /// Same as [`Self::async_stream`] except that a WASIp2 [`InputStream`] is
146    /// returned.
147    ///
148    /// Note that this has a default implementation which uses
149    /// [`p2::pipe::AsyncReadStream`] as an adapter, but this can be overridden
150    /// if there's a more specialized implementation available.
151    fn p2_stream(&self) -> Box<dyn InputStream> {
152        Box::new(p2::pipe::AsyncReadStream::new(Pin::from(
153            self.async_stream(),
154        )))
155    }
156}
157
158/// Similar to [`StdinStream`], except for output.
159///
160/// This is used both for a guest stdin and a guest stdout.
161///
162/// Note that there are many built-in implementations of this trait for various
163/// types such as [`tokio::io::Stdout`], [`tokio::io::Empty`], and
164/// [`p2::pipe::MemoryOutputPipe`].
165pub trait StdoutStream: IsTerminal + Send {
166    /// Returns a fresh new stream which can write to this output stream.
167    ///
168    /// Note that all output streams should output to the same logical source.
169    /// This means that it's possible for each independent stream to acquire a
170    /// separate "permit" to write and then act on that permit. Note that
171    /// additionally at this time once a permit is "acquired" there's no way to
172    /// release it, for example you can wait for readiness and then never
173    /// actually write in WASI. This means that acquisition of a permit for one
174    /// stream cannot discount the size of a permit another stream could
175    /// obtain.
176    ///
177    /// Implementations must be able to handle this
178    fn async_stream(&self) -> Box<dyn AsyncWrite + Send + Sync>;
179
180    /// Same as [`Self::async_stream`] except that a WASIp2 [`OutputStream`] is
181    /// returned.
182    ///
183    /// Note that this has a default implementation which uses
184    /// [`p2::pipe::AsyncWriteStream`] as an adapter, but this can be overridden
185    /// if there's a more specialized implementation available.
186    fn p2_stream(&self) -> Box<dyn OutputStream> {
187        Box::new(p2::pipe::AsyncWriteStream::new(
188            8192, // FIXME: extract this to a constant.
189            Pin::from(self.async_stream()),
190        ))
191    }
192}
193
194// Forward `&T => T`
195impl<T: ?Sized + IsTerminal> IsTerminal for &T {
196    fn is_terminal(&self) -> bool {
197        T::is_terminal(self)
198    }
199}
200impl<T: ?Sized + StdinStream + Sync> StdinStream for &T {
201    fn p2_stream(&self) -> Box<dyn InputStream> {
202        T::p2_stream(self)
203    }
204    fn async_stream(&self) -> Box<dyn AsyncRead + Send + Sync> {
205        T::async_stream(self)
206    }
207}
208impl<T: ?Sized + StdoutStream + Sync> StdoutStream for &T {
209    fn p2_stream(&self) -> Box<dyn OutputStream> {
210        T::p2_stream(self)
211    }
212    fn async_stream(&self) -> Box<dyn AsyncWrite + Send + Sync> {
213        T::async_stream(self)
214    }
215}
216
217// Forward `&mut T => T`
218impl<T: ?Sized + IsTerminal> IsTerminal for &mut T {
219    fn is_terminal(&self) -> bool {
220        T::is_terminal(self)
221    }
222}
223impl<T: ?Sized + StdinStream + Sync> StdinStream for &mut T {
224    fn p2_stream(&self) -> Box<dyn InputStream> {
225        T::p2_stream(self)
226    }
227    fn async_stream(&self) -> Box<dyn AsyncRead + Send + Sync> {
228        T::async_stream(self)
229    }
230}
231impl<T: ?Sized + StdoutStream + Sync> StdoutStream for &mut T {
232    fn p2_stream(&self) -> Box<dyn OutputStream> {
233        T::p2_stream(self)
234    }
235    fn async_stream(&self) -> Box<dyn AsyncWrite + Send + Sync> {
236        T::async_stream(self)
237    }
238}
239
240// Forward `Box<T> => T`
241impl<T: ?Sized + IsTerminal> IsTerminal for Box<T> {
242    fn is_terminal(&self) -> bool {
243        T::is_terminal(self)
244    }
245}
246impl<T: ?Sized + StdinStream + Sync> StdinStream for Box<T> {
247    fn p2_stream(&self) -> Box<dyn InputStream> {
248        T::p2_stream(self)
249    }
250    fn async_stream(&self) -> Box<dyn AsyncRead + Send + Sync> {
251        T::async_stream(self)
252    }
253}
254impl<T: ?Sized + StdoutStream + Sync> StdoutStream for Box<T> {
255    fn p2_stream(&self) -> Box<dyn OutputStream> {
256        T::p2_stream(self)
257    }
258    fn async_stream(&self) -> Box<dyn AsyncWrite + Send + Sync> {
259        T::async_stream(self)
260    }
261}
262
263// Forward `Arc<T> => T`
264impl<T: ?Sized + IsTerminal> IsTerminal for Arc<T> {
265    fn is_terminal(&self) -> bool {
266        T::is_terminal(self)
267    }
268}
269impl<T: ?Sized + StdinStream + Sync> StdinStream for Arc<T> {
270    fn p2_stream(&self) -> Box<dyn InputStream> {
271        T::p2_stream(self)
272    }
273    fn async_stream(&self) -> Box<dyn AsyncRead + Send + Sync> {
274        T::async_stream(self)
275    }
276}
277impl<T: ?Sized + StdoutStream + Sync> StdoutStream for Arc<T> {
278    fn p2_stream(&self) -> Box<dyn OutputStream> {
279        T::p2_stream(self)
280    }
281    fn async_stream(&self) -> Box<dyn AsyncWrite + Send + Sync> {
282        T::async_stream(self)
283    }
284}
285
286#[cfg(test)]
287mod test {
288    use crate::cli::{AsyncStdoutStream, StdinStream, StdoutStream};
289    use crate::p2::{self, OutputStream};
290    use bytes::Bytes;
291    use tokio::io::AsyncReadExt;
292    use wasmtime::Result;
293
294    #[test]
295    fn memory_stdin_stream() {
296        // A StdinStream has the property that there are multiple
297        // InputStreams created, using the stream() method which are each
298        // views on the same shared state underneath. Consuming input on one
299        // stream results in consuming that input on all streams.
300        //
301        // The simplest way to measure this is to check if the MemoryInputPipe
302        // impl of StdinStream follows this property.
303
304        let pipe =
305            p2::pipe::MemoryInputPipe::new("the quick brown fox jumped over the three lazy dogs");
306
307        let mut view1 = pipe.p2_stream();
308        let mut view2 = pipe.p2_stream();
309
310        let read1 = view1.read(10).expect("read first 10 bytes");
311        assert_eq!(read1, "the quick ".as_bytes(), "first 10 bytes");
312        let read2 = view2.read(10).expect("read second 10 bytes");
313        assert_eq!(read2, "brown fox ".as_bytes(), "second 10 bytes");
314        let read3 = view1.read(10).expect("read third 10 bytes");
315        assert_eq!(read3, "jumped ove".as_bytes(), "third 10 bytes");
316        let read4 = view2.read(10).expect("read fourth 10 bytes");
317        assert_eq!(read4, "r the thre".as_bytes(), "fourth 10 bytes");
318    }
319
320    #[tokio::test]
321    async fn async_stdin_stream() {
322        // A StdinStream has the property that there are multiple
323        // InputStreams created, using the stream() method which are each
324        // views on the same shared state underneath. Consuming input on one
325        // stream results in consuming that input on all streams.
326        //
327        // AsyncStdinStream is a slightly more complex impl of StdinStream
328        // than the MemoryInputPipe above. We can create an AsyncReadStream
329        // from a file on the disk, and an AsyncStdinStream from that common
330        // stream, then check that the same property holds as above.
331
332        let dir = tempfile::tempdir().unwrap();
333        let mut path = std::path::PathBuf::from(dir.path());
334        path.push("file");
335        std::fs::write(&path, "the quick brown fox jumped over the three lazy dogs").unwrap();
336
337        let file = tokio::fs::File::open(&path)
338            .await
339            .expect("open created file");
340        let stdin_stream = super::AsyncStdinStream::new(file);
341
342        use super::StdinStream;
343
344        let mut view1 = stdin_stream.p2_stream();
345        let mut view2 = stdin_stream.p2_stream();
346
347        view1.ready().await;
348
349        let read1 = view1.read(10).expect("read first 10 bytes");
350        assert_eq!(read1, "the quick ".as_bytes(), "first 10 bytes");
351        let read2 = view2.read(10).expect("read second 10 bytes");
352        assert_eq!(read2, "brown fox ".as_bytes(), "second 10 bytes");
353        let read3 = view1.read(10).expect("read third 10 bytes");
354        assert_eq!(read3, "jumped ove".as_bytes(), "third 10 bytes");
355        let read4 = view2.read(10).expect("read fourth 10 bytes");
356        assert_eq!(read4, "r the thre".as_bytes(), "fourth 10 bytes");
357    }
358
359    #[tokio::test]
360    async fn async_stdout_stream_unblocks() {
361        let (mut read, write) = tokio::io::duplex(32);
362        let stdout = AsyncStdoutStream::new(32, write);
363
364        let task = tokio::task::spawn(async move {
365            let mut stream = stdout.p2_stream();
366            blocking_write_and_flush(&mut *stream, "x".into())
367                .await
368                .unwrap();
369        });
370
371        let mut buf = [0; 100];
372        let n = read.read(&mut buf).await.unwrap();
373        assert_eq!(&buf[..n], b"x");
374
375        task.await.unwrap();
376    }
377
378    async fn blocking_write_and_flush(s: &mut dyn OutputStream, mut bytes: Bytes) -> Result<()> {
379        while !bytes.is_empty() {
380            let permit = s.write_ready().await?;
381            let len = bytes.len().min(permit);
382            let chunk = bytes.split_to(len);
383            s.write(chunk)?;
384        }
385
386        s.flush()?;
387        s.write_ready().await?;
388        Ok(())
389    }
390
391    // Verify that the stdio OutputStream implementation reports a usable
392    // write permit and can successfully write + flush (exercises the full
393    // trait impl including the error conversion path).
394    #[test]
395    fn stdio_output_stream_write_flush() {
396        let mut stream: Box<dyn wasmtime_wasi_io::streams::OutputStream> =
397            StdoutStream::p2_stream(&std::io::stderr());
398
399        let permit = stream.check_write().expect("check_write");
400        assert!(permit > 0, "permit should be nonzero");
401
402        // Writing empty bytes must succeed.
403        stream
404            .write(Bytes::new())
405            .expect("writing empty bytes should succeed");
406
407        // Flushing must succeed.
408        stream.flush().expect("flush should succeed");
409    }
410
411    #[test]
412    fn stream_error_from_broken_pipe_maps_to_closed() {
413        use std::io;
414        use wasmtime_wasi_io::streams::StreamError;
415
416        let err = super::stream_error_from(io::Error::from(io::ErrorKind::BrokenPipe));
417        assert!(matches!(err, StreamError::Closed));
418    }
419
420    #[test]
421    fn stream_error_from_preserves_io_error() {
422        use std::io;
423        use wasmtime_wasi_io::streams::StreamError;
424
425        let err = super::stream_error_from(io::Error::from(io::ErrorKind::IsADirectory));
426        match err {
427            StreamError::LastOperationFailed(e) => {
428                let io_err = e.downcast::<io::Error>().expect("should downcast");
429                assert_eq!(io_err.kind(), io::ErrorKind::IsADirectory);
430            }
431            other => panic!("expected LastOperationFailed, got: {other:?}"),
432        }
433    }
434
435    #[cfg(unix)]
436    #[test]
437    fn stream_error_from_raw_os_eisdir() {
438        use rustix::io::Errno;
439        use std::io;
440        use wasmtime_wasi_io::streams::StreamError;
441
442        let err =
443            super::stream_error_from(io::Error::from_raw_os_error(Errno::ISDIR.raw_os_error()));
444        match err {
445            StreamError::LastOperationFailed(e) => {
446                let io_err = e.downcast::<io::Error>().expect("should downcast");
447                assert_eq!(io_err.raw_os_error(), Some(Errno::ISDIR.raw_os_error()));
448            }
449            other => panic!("expected LastOperationFailed, got: {other:?}"),
450        }
451    }
452}
wasmtime_wasi/cli.rs

wasmtime_wasi/
cli.rs
