Split off wasmtime-wasi-io crate from wasmtime-wasi

.github/workflows/main.yml

@@ -541,7 +541,8 @@ jobs:
             cargo check -p wasmtime --no-default-features --features runtime,component-model &&
             cargo check -p wasmtime --no-default-features --features runtime,gc,component-model &&
             cargo check -p cranelift-control --no-default-features &&
-            cargo check -p pulley-interpreter --features encode,decode,disas,interp
+            cargo check -p pulley-interpreter --features encode,decode,disas,interp &&
+            cargo check -p wasmtime-wasi-io --no-default-features
         # Use `cross` for illumos to have a C compiler/linker available.
         - target: x86_64-unknown-illumos
           os: ubuntu-latest

Cargo.toml

@@ -230,7 +230,8 @@ wasmtime-fiber = { path = "crates/fiber", version = "=30.0.0" }
 wasmtime-jit-debug = { path = "crates/jit-debug", version = "=30.0.0" }
 wasmtime-wast = { path = "crates/wast", version = "=30.0.0" }
 wasmtime-wasi = { path = "crates/wasi", version = "30.0.0", default-features = false }
-wasmtime-wasi-http = { path = "crates/wasi-http", version = "=30.0.0", default-features = false }
+wasmtime-wasi-io = { path = "crates/wasi-io", version = "30.0.0", default-features = false }
+wasmtime-wasi-http = { path = "crates/wasi-http", version = "30.0.0", default-features = false }
 wasmtime-wasi-nn = { path = "crates/wasi-nn", version = "30.0.0" }
 wasmtime-wasi-config = { path = "crates/wasi-config", version = "30.0.0" }
 wasmtime-wasi-keyvalue = { path = "crates/wasi-keyvalue", version = "30.0.0" }
@@ -355,7 +356,7 @@ hyper = "1.0.1"
 http = "1.0.0"
 http-body = "1.0.0"
 http-body-util = "0.1.0"
-bytes = "1.4"
+bytes = { version = "1.4", default-features = false }
 futures = { version = "0.3.27", default-features = false }
 indexmap = { version = "2.0.0", default-features = false }
 pretty_env_logger = "0.5.0"

ci/vendor-wit.sh

@@ -36,6 +36,10 @@ make_vendor() {
 
 cache_dir=$(mktemp -d)
 
+make_vendor "wasi-io" "
+  io@v0.2.3
+"
+
 make_vendor "wasi" "
   cli@v0.2.3
   clocks@v0.2.3

crates/wasi-http/src/bindings.rs

@@ -59,9 +59,11 @@ pub mod sync {
             tracing: true,
             async: false,
             with: {
-                "wasi:http": crate::bindings::http, // http is in this crate
-                "wasi:io": wasmtime_wasi::bindings::sync::io, // io is sync
-                "wasi": wasmtime_wasi::bindings, // everything else
+                // http is in this crate
+                "wasi:http": crate::bindings::http,
+                // sync requires the wrapper in the wasmtime_wasi crate, in
+                // order to have in_tokio
+                "wasi:io": wasmtime_wasi::bindings::sync::io,
             },
             require_store_data_send: true,
         });

crates/wasi-http/src/body.rs

@@ -13,7 +13,7 @@ use std::{pin::Pin, sync::Arc, time::Duration};
 use tokio::sync::{mpsc, oneshot};
 use wasmtime_wasi::{
     runtime::{poll_noop, AbortOnDropJoinHandle},
-    HostInputStream, HostOutputStream, StreamError, Subscribe,
+    InputStream, OutputStream, Pollable, StreamError,
 };
 
 /// Common type for incoming bodies.
@@ -234,7 +234,7 @@ enum IncomingBodyStreamState {
 }
 
 #[async_trait::async_trait]
-impl HostInputStream for HostIncomingBodyStream {
+impl InputStream for HostIncomingBodyStream {
     fn read(&mut self, size: usize) -> Result<Bytes, StreamError> {
         loop {
             // Handle buffered data/errors if any
@@ -271,7 +271,7 @@ impl HostInputStream for HostIncomingBodyStream {
 }
 
 #[async_trait::async_trait]
-impl Subscribe for HostIncomingBodyStream {
+impl Pollable for HostIncomingBodyStream {
     async fn ready(&mut self) {
         if !self.buffer.is_empty() || self.error.is_some() {
             return;
@@ -327,7 +327,7 @@ pub enum HostFutureTrailers {
 }
 
 #[async_trait::async_trait]
-impl Subscribe for HostFutureTrailers {
+impl Pollable for HostFutureTrailers {
     async fn ready(&mut self) {
         let body = match self {
             HostFutureTrailers::Waiting(body) => body,
@@ -415,7 +415,7 @@ impl WrittenState {
 /// The concrete type behind a `wasi:http/types/outgoing-body` resource.
 pub struct HostOutgoingBody {
     /// The output stream that the body is written to.
-    body_output_stream: Option<Box<dyn HostOutputStream>>,
+    body_output_stream: Option<Box<dyn OutputStream>>,
     context: StreamContext,
     written: Option<WrittenState>,
     finish_sender: Option<tokio::sync::oneshot::Sender<FinishMessage>>,
@@ -499,7 +499,7 @@ impl HostOutgoingBody {
     }
 
     /// Take the output stream, if it's available.
-    pub fn take_output_stream(&mut self) -> Option<Box<dyn HostOutputStream>> {
+    pub fn take_output_stream(&mut self) -> Option<Box<dyn OutputStream>> {
         self.body_output_stream.take()
     }
 
@@ -605,7 +605,7 @@ impl BodyWriteStream {
 }
 
 #[async_trait::async_trait]
-impl HostOutputStream for BodyWriteStream {
+impl OutputStream for BodyWriteStream {
     fn write(&mut self, bytes: Bytes) -> Result<(), StreamError> {
         let len = bytes.len();
         match self.writer.try_send(bytes) {
@@ -665,7 +665,7 @@ impl HostOutputStream for BodyWriteStream {
 }
 
 #[async_trait::async_trait]
-impl Subscribe for BodyWriteStream {
+impl Pollable for BodyWriteStream {
     async fn ready(&mut self) {
         // Attempt to perform a reservation for a send. If there's capacity in
         // the channel or it's already closed then this will return immediately.

crates/wasi-http/src/error.rs

@@ -1,7 +1,7 @@
 use crate::bindings::http::types::ErrorCode;
 use std::error::Error;
 use std::fmt;
-use wasmtime_wasi::ResourceTableError;
+use wasmtime::component::ResourceTableError;
 
 /// A [`Result`] type where the error type defaults to [`HttpError`].
 pub type HttpResult<T, E = HttpError> = Result<T, E>;

crates/wasi-http/src/lib.rs

@@ -285,12 +285,13 @@ where
     T: WasiHttpView + wasmtime_wasi::WasiView,
 {
     let io_closure = type_annotate_io::<T, _>(|t| wasmtime_wasi::IoImpl(t));
-    let closure = type_annotate_wasi::<T, _>(|t| wasmtime_wasi::WasiImpl(wasmtime_wasi::IoImpl(t)));
-    wasmtime_wasi::bindings::clocks::wall_clock::add_to_linker_get_host(l, closure)?;
-    wasmtime_wasi::bindings::clocks::monotonic_clock::add_to_linker_get_host(l, closure)?;
     wasmtime_wasi::bindings::io::poll::add_to_linker_get_host(l, io_closure)?;
     wasmtime_wasi::bindings::io::error::add_to_linker_get_host(l, io_closure)?;
     wasmtime_wasi::bindings::io::streams::add_to_linker_get_host(l, io_closure)?;
+
+    let closure = type_annotate_wasi::<T, _>(|t| wasmtime_wasi::WasiImpl(wasmtime_wasi::IoImpl(t)));
+    wasmtime_wasi::bindings::clocks::wall_clock::add_to_linker_get_host(l, closure)?;
+    wasmtime_wasi::bindings::clocks::monotonic_clock::add_to_linker_get_host(l, closure)?;
     wasmtime_wasi::bindings::cli::stdin::add_to_linker_get_host(l, closure)?;
     wasmtime_wasi::bindings::cli::stdout::add_to_linker_get_host(l, closure)?;
     wasmtime_wasi::bindings::cli::stderr::add_to_linker_get_host(l, closure)?;
@@ -383,13 +384,17 @@ where
     T: WasiHttpView + wasmtime_wasi::WasiView,
 {
     let io_closure = type_annotate_io::<T, _>(|t| wasmtime_wasi::IoImpl(t));
+    // For the sync linker, use the definitions of poll and streams from the
+    // wasmtime_wasi::bindings::sync space because those are defined using in_tokio.
+    wasmtime_wasi::bindings::sync::io::poll::add_to_linker_get_host(l, io_closure)?;
+    wasmtime_wasi::bindings::sync::io::streams::add_to_linker_get_host(l, io_closure)?;
+    // The error interface in the wasmtime_wasi is synchronous
+    wasmtime_wasi::bindings::io::error::add_to_linker_get_host(l, io_closure)?;
+
     let closure = type_annotate_wasi::<T, _>(|t| wasmtime_wasi::WasiImpl(wasmtime_wasi::IoImpl(t)));
 
     wasmtime_wasi::bindings::clocks::wall_clock::add_to_linker_get_host(l, closure)?;
     wasmtime_wasi::bindings::clocks::monotonic_clock::add_to_linker_get_host(l, closure)?;
-    wasmtime_wasi::bindings::sync::io::poll::add_to_linker_get_host(l, io_closure)?;
-    wasmtime_wasi::bindings::sync::io::streams::add_to_linker_get_host(l, io_closure)?;
-    wasmtime_wasi::bindings::io::error::add_to_linker_get_host(l, io_closure)?;
     wasmtime_wasi::bindings::cli::stdin::add_to_linker_get_host(l, closure)?;
     wasmtime_wasi::bindings::cli::stdout::add_to_linker_get_host(l, closure)?;
     wasmtime_wasi::bindings::cli::stderr::add_to_linker_get_host(l, closure)?;

crates/wasi-http/src/types.rs

@@ -18,7 +18,7 @@ use std::time::Duration;
 use tokio::net::TcpStream;
 use tokio::time::timeout;
 use wasmtime::component::{Resource, ResourceTable};
-use wasmtime_wasi::{runtime::AbortOnDropJoinHandle, IoImpl, IoView, Subscribe};
+use wasmtime_wasi::{runtime::AbortOnDropJoinHandle, IoImpl, IoView, Pollable};
 
 /// Capture the state necessary for use in the wasi-http API implementation.
 #[derive(Debug)]
@@ -715,7 +715,7 @@ impl HostFutureIncomingResponse {
 }
 
 #[async_trait::async_trait]
-impl Subscribe for HostFutureIncomingResponse {
+impl Pollable for HostFutureIncomingResponse {
     async fn ready(&mut self) {
         if let Self::Pending(handle) = self {
             *self = Self::Ready(handle.await);

crates/wasi-http/src/types_impl.rs

@@ -13,11 +13,8 @@ use crate::{
 use anyhow::Context;
 use std::any::Any;
 use std::str::FromStr;
-use wasmtime::component::{Resource, ResourceTable};
-use wasmtime_wasi::{
-    bindings::io::streams::{InputStream, OutputStream},
-    IoView, Pollable, ResourceTableError,
-};
+use wasmtime::component::{Resource, ResourceTable, ResourceTableError};
+use wasmtime_wasi::{DynInputStream, DynOutputStream, DynPollable, IoView};
 
 impl<T> crate::bindings::http::types::Host for WasiHttpImpl<T>
 where
@@ -662,7 +659,7 @@ where
     fn subscribe(
         &mut self,
         index: Resource<HostFutureTrailers>,
-    ) -> wasmtime::Result<Resource<Pollable>> {
+    ) -> wasmtime::Result<Resource<DynPollable>> {
         wasmtime_wasi::subscribe(self.table(), index)
     }
 
@@ -704,11 +701,11 @@ where
     fn stream(
         &mut self,
         id: Resource<HostIncomingBody>,
-    ) -> wasmtime::Result<Result<Resource<InputStream>, ()>> {
+    ) -> wasmtime::Result<Result<Resource<DynInputStream>, ()>> {
         let body = self.table().get_mut(&id)?;
 
         if let Some(stream) = body.take_stream() {
-            let stream: InputStream = Box::new(stream);
+            let stream: DynInputStream = Box::new(stream);
             let stream = self.table().push_child(stream, &id)?;
             return Ok(Ok(stream));
         }
@@ -883,7 +880,7 @@ where
     fn subscribe(
         &mut self,
         id: Resource<HostFutureIncomingResponse>,
-    ) -> wasmtime::Result<Resource<Pollable>> {
+    ) -> wasmtime::Result<Resource<DynPollable>> {
         wasmtime_wasi::subscribe(self.table(), id)
     }
 }
@@ -895,7 +892,7 @@ where
     fn write(
         &mut self,
         id: Resource<HostOutgoingBody>,
-    ) -> wasmtime::Result<Result<Resource<OutputStream>, ()>> {
+    ) -> wasmtime::Result<Result<Resource<DynOutputStream>, ()>> {
         let body = self.table().get_mut(&id)?;
         if let Some(stream) = body.take_output_stream() {
             let id = self.table().push_child(stream, &id)?;

crates/wasi-io/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "wasmtime-wasi-io"
+version.workspace = true
+authors.workspace = true
+description = "wasi-io common traits to be shared among other wasi implementations"
+license = "Apache-2.0 WITH LLVM-exception"
+categories = ["wasm"]
+keywords = ["webassembly", "wasm"]
+repository = "https://github.com/bytecodealliance/wasmtime"
+edition.workspace = true
+rust-version.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+wasmtime = { workspace = true, features = ["component-model", "async", "runtime"] }
+anyhow = { workspace = true }
+bytes = { workspace = true }
+async-trait = { workspace = true }
+futures = { workspace = true }
+
+[features]
+default = [ "std" ]
+std = [
+    "bytes/std",
+    "anyhow/std",
+    "wasmtime/std",
+]
+

crates/wasi-io/src/bindings.rs

@@ -0,0 +1,29 @@
+wasmtime::component::bindgen!({
+    path: "wit",
+    trappable_imports: true,
+    with: {
+        "wasi:io/poll/pollable": crate::poll::DynPollable,
+        "wasi:io/streams/input-stream": crate::streams::DynInputStream,
+        "wasi:io/streams/output-stream": crate::streams::DynOutputStream,
+        "wasi:io/error/error": crate::streams::Error,
+    },
+    async: {
+        only_imports: [
+            "poll",
+            "[method]pollable.block",
+            "[method]pollable.ready",
+            "[method]input-stream.blocking-read",
+            "[method]input-stream.blocking-skip",
+            "[drop]input-stream",
+            "[method]output-stream.blocking-splice",
+            "[method]output-stream.blocking-flush",
+            "[method]output-stream.blocking-write",
+            "[method]output-stream.blocking-write-and-flush",
+            "[method]output-stream.blocking-write-zeroes-and-flush",
+            "[drop]output-stream",
+        ]
+    },
+    trappable_error_type: {
+        "wasi:io/streams/stream-error" => crate::streams::StreamError,
+    }
+});