From 74d66b1deb5be4a2c73fdcc9e0e12a05bdb43e38 Mon Sep 17 00:00:00 2001 From: Roman Volosatovs Date: Mon, 24 Aug 2026 17:29:06 +0200 Subject: [PATCH 1/2] [TMP] build: switch wasm-tools to git dep Pick up `wasmparser`/`wast`/`wasm-encoder` support for the `stream.forward` and `future.forward` canonical built-ins from https://github.com/bytecodealliance/wasm-tools/pull/2614 until that lands upstream. --- Cargo.lock | 39 +++++++++++++-------------------------- Cargo.toml | 15 +++++++++++++++ 2 files changed, 28 insertions(+), 26 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2202f6137d35..a209c0fd8ced 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2248,8 +2248,7 @@ dependencies = [ [[package]] name = "json-from-wast" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c74cef75a485ea9c18b6f7c1928bda2ef3a5727227fe7ebab4260626bc75f98e" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "anyhow", "serde", @@ -4395,8 +4394,7 @@ dependencies = [ [[package]] name = "wasm-compose" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7dd08e5283f98238f2f16f5f978188a33af9901b81186bb549faad306a7b5388" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "anyhow", "heck", @@ -4422,8 +4420,7 @@ dependencies = [ [[package]] name = "wasm-encoder" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e974fe6821a8cf64575d51ea2194e2c8f77e7b66e9afe7419ce8a97f9ee0d251" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "leb128fmt", "wasmparser 0.258.0", @@ -4444,8 +4441,7 @@ dependencies = [ [[package]] name = "wasm-metadata" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "18a11585adb92fe9b55ad1d760e8d8fb5d87e0d2e303cb8eed57f078d54293a2" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "anyhow", "indexmap 2.14.0", @@ -4456,8 +4452,7 @@ dependencies = [ [[package]] name = "wasm-mutate" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ae9cac04ab7c539efa6b4288726ced0d423c95f1bb63659d74506142ff0499b" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "egg", "log", @@ -4470,8 +4465,7 @@ dependencies = [ [[package]] name = "wasm-smith" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a84b31d645ea084a94e1a7c410e30219c165a0960b336f535f99162caf2cfd0c" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "anyhow", "arbitrary", @@ -4493,8 +4487,7 @@ dependencies = [ [[package]] name = "wasm-wave" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce1d8cbc89bc22dac012f5fd56fca419330cb88d8d2c8f6ccb313f08ec4a6836" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "anyhow", "logos", @@ -4567,8 +4560,7 @@ dependencies = [ [[package]] name = "wasmparser" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d9a61719f93a87b16d325921e251800c4833f8fab50fa21c7de73aed50086313" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "bitflags 2.11.1", "hashbrown 0.17.0", @@ -4580,8 +4572,7 @@ dependencies = [ [[package]] name = "wasmprinter" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a65e30fb2cd3cc5cb7761a6064787b2abf017e6566d8d0dedf2bb7f55ad5383" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "anyhow", "termcolor", @@ -5334,8 +5325,7 @@ dependencies = [ [[package]] name = "wast" version = "258.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97f7defc7ecca8b19ac7f824598eadd0c53985ee00c74060d65051e9da5b58a1" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "bumpalo", "gimli 0.32.3", @@ -5348,8 +5338,7 @@ dependencies = [ [[package]] name = "wat" version = "1.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7555c008cca87f2ac58d9f83ccda7e7b44611093ce28eb28f052e7c78024b9bf" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "wast 258.0.0", ] @@ -5785,8 +5774,7 @@ dependencies = [ [[package]] name = "wit-component" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "481b5c47b2ecce0389b5e08a05557d6a190c9cd761773b8880a8017ee04dc7ef" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "anyhow", "bitflags 2.11.1", @@ -5822,8 +5810,7 @@ dependencies = [ [[package]] name = "wit-parser" version = "0.258.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff4daaa3cd97ae49ecd0a99dc009d453f93e0f083dd3be38c0f24a83a93e37ac" +source = "git+https://github.com/bytecodealliance/wasm-tools?rev=a7a7254c359e82c7a55b9827ddca30a42eecb1e9#a7a7254c359e82c7a55b9827ddca30a42eecb1e9" dependencies = [ "anyhow", "hashbrown 0.17.0", diff --git a/Cargo.toml b/Cargo.toml index 8af2632f615b..33a385de73d4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -791,3 +791,18 @@ fpr = "fpr" [workspace.metadata.typos.files] extend-exclude = [ "docs/js/mermaid*.js", "crates/wasi-nn/**/*.txt", "*.isle" ] + +[patch.crates-io] +wasmparser = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wat = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wast = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wasmprinter = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wasm-encoder = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wasm-smith = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wasm-mutate = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wit-parser = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wit-component = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wasm-wave = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wasm-compose = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +wasm-metadata = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } +json-from-wast = { git = "https://github.com/bytecodealliance/wasm-tools", rev = "a7a7254c359e82c7a55b9827ddca30a42eecb1e9" } From 1b4f01f6cd804257e900ea93462aaaaf4a75389a Mon Sep 17 00:00:00 2001 From: Roman Volosatovs Date: Tue, 25 Aug 2026 18:55:45 +0200 Subject: [PATCH 2/2] implement `{future,stream}.forward` Signed-off-by: Roman Volosatovs --- crates/cranelift/src/compiler/component.rs | 24 + crates/environ/src/component.rs | 4 + crates/environ/src/component/dfg.rs | 16 + crates/environ/src/component/info.rs | 22 + crates/environ/src/component/translate.rs | 28 + .../environ/src/component/translate/inline.rs | 30 + .../src/runtime/component/concurrent.rs | 48 + .../concurrent/futures_and_streams.rs | 616 +++++++- .../src/runtime/vm/component/libcalls.rs | 34 + tests/all/component_model.rs | 1 + tests/all/component_model/forward.rs | 1405 +++++++++++++++++ 11 files changed, 2206 insertions(+), 22 deletions(-) create mode 100644 tests/all/component_model/forward.rs diff --git a/crates/cranelift/src/compiler/component.rs b/crates/cranelift/src/compiler/component.rs index 1a77491e7887..d94e4206153b 100644 --- a/crates/cranelift/src/compiler/component.rs +++ b/crates/cranelift/src/compiler/component.rs @@ -401,6 +401,17 @@ impl<'a> TrampolineCompiler<'a> { ); } } + Trampoline::StreamForward { instance, ty } => { + self.translate_libcall( + host::stream_forward, + TrapSentinel::Falsy, + WasmArgs::InRegisters, + |me, params| { + params.push(me.index_value(*instance)); + params.push(me.index_value(*ty)); + }, + ); + } Trampoline::StreamCancelRead { instance, ty, @@ -498,6 +509,17 @@ impl<'a> TrampolineCompiler<'a> { }, ); } + Trampoline::FutureForward { instance, ty } => { + self.translate_libcall( + host::future_forward, + TrapSentinel::Falsy, + WasmArgs::InRegisters, + |me, params| { + params.push(me.index_value(*instance)); + params.push(me.index_value(*ty)); + }, + ); + } Trampoline::FutureCancelRead { instance, ty, @@ -1551,6 +1573,7 @@ impl<'a> TrampolineCompiler<'a> { | Trampoline::StreamNew { instance, .. } | Trampoline::StreamRead { instance, .. } | Trampoline::StreamWrite { instance, .. } + | Trampoline::StreamForward { instance, .. } | Trampoline::StreamCancelRead { instance, .. } | Trampoline::StreamCancelWrite { instance, .. } | Trampoline::StreamDropReadable { instance, .. } @@ -1558,6 +1581,7 @@ impl<'a> TrampolineCompiler<'a> { | Trampoline::FutureNew { instance, .. } | Trampoline::FutureRead { instance, .. } | Trampoline::FutureWrite { instance, .. } + | Trampoline::FutureForward { instance, .. } | Trampoline::FutureCancelRead { instance, .. } | Trampoline::FutureCancelWrite { instance, .. } | Trampoline::FutureDropReadable { instance, .. } diff --git a/crates/environ/src/component.rs b/crates/environ/src/component.rs index 9e3233209d66..404904a47657 100644 --- a/crates/environ/src/component.rs +++ b/crates/environ/src/component.rs @@ -146,6 +146,8 @@ macro_rules! foreach_builtin_component_function { #[cfg(feature = "component-model-async")] future_read(vmctx: vmctx, caller_instance: u32, ty: u32, options: u32, future: u32, address: u32) -> u64; #[cfg(feature = "component-model-async")] + future_forward(vmctx: vmctx, caller_instance: u32, ty: u32, reader: u32, writer: u32) -> bool; + #[cfg(feature = "component-model-async")] future_cancel_write(vmctx: vmctx, caller_instance: u32, ty: u32, async_: u8, writer: u32) -> u64; #[cfg(feature = "component-model-async")] future_cancel_read(vmctx: vmctx, caller_instance: u32, ty: u32, async_: u8, reader: u32) -> u64; @@ -160,6 +162,8 @@ macro_rules! foreach_builtin_component_function { #[cfg(feature = "component-model-async")] stream_read(vmctx: vmctx, caller_instance: u32, ty: u32, options: u32, stream: u32, address: u32, count: u32) -> u64; #[cfg(feature = "component-model-async")] + stream_forward(vmctx: vmctx, caller_instance: u32, ty: u32, reader: u32, writer: u32) -> bool; + #[cfg(feature = "component-model-async")] stream_cancel_write(vmctx: vmctx, caller_instance: u32, ty: u32, async_: u8, writer: u32) -> u64; #[cfg(feature = "component-model-async")] stream_cancel_read(vmctx: vmctx, caller_instance: u32, ty: u32, async_: u8, reader: u32) -> u64; diff --git a/crates/environ/src/component/dfg.rs b/crates/environ/src/component/dfg.rs index fad632be02ef..f54aae1e5270 100644 --- a/crates/environ/src/component/dfg.rs +++ b/crates/environ/src/component/dfg.rs @@ -390,6 +390,10 @@ pub enum Trampoline { ty: TypeStreamTableIndex, options: OptionsId, }, + StreamForward { + instance: RuntimeComponentInstanceIndex, + ty: TypeStreamTableIndex, + }, StreamCancelRead { instance: RuntimeComponentInstanceIndex, ty: TypeStreamTableIndex, @@ -422,6 +426,10 @@ pub enum Trampoline { ty: TypeFutureTableIndex, options: OptionsId, }, + FutureForward { + instance: RuntimeComponentInstanceIndex, + ty: TypeFutureTableIndex, + }, FutureCancelRead { instance: RuntimeComponentInstanceIndex, ty: TypeFutureTableIndex, @@ -1033,6 +1041,10 @@ impl LinearizeDfg<'_> { ty: *ty, options: self.options(*options), }, + Trampoline::StreamForward { instance, ty } => info::Trampoline::StreamForward { + instance: *instance, + ty: *ty, + }, Trampoline::StreamCancelRead { instance, ty, @@ -1085,6 +1097,10 @@ impl LinearizeDfg<'_> { ty: *ty, options: self.options(*options), }, + Trampoline::FutureForward { instance, ty } => info::Trampoline::FutureForward { + instance: *instance, + ty: *ty, + }, Trampoline::FutureCancelRead { instance, ty, diff --git a/crates/environ/src/component/info.rs b/crates/environ/src/component/info.rs index 679564a33a6c..5ffd4fac847a 100644 --- a/crates/environ/src/component/info.rs +++ b/crates/environ/src/component/info.rs @@ -887,6 +887,16 @@ pub enum Trampoline { options: OptionsIndex, }, + /// A `stream.forward` intrinsic to forward all remaining elements from the + /// readable end of one `stream` into the writable end of another `stream` + /// of the specified type. + StreamForward { + /// The specific component instance which is calling the intrinsic. + instance: RuntimeComponentInstanceIndex, + /// The table index for the specific `stream` type and caller instance. + ty: TypeStreamTableIndex, + }, + /// A `stream.cancel-read` intrinsic to cancel an in-progress read from a /// `stream` of the specified type. StreamCancelRead { @@ -960,6 +970,16 @@ pub enum Trampoline { options: OptionsIndex, }, + /// A `future.forward` intrinsic to forward the value of the `future` with + /// the specified readable end into the `future` with the specified + /// writable end. + FutureForward { + /// The specific component instance which is calling the intrinsic. + instance: RuntimeComponentInstanceIndex, + /// The table index for the specific `future` type and caller instance. + ty: TypeFutureTableIndex, + }, + /// A `future.cancel-read` intrinsic to cancel an in-progress read from a /// `future` of the specified type. FutureCancelRead { @@ -1225,6 +1245,7 @@ impl Trampoline { StreamNew { .. } => format!("stream-new"), StreamRead { .. } => format!("stream-read"), StreamWrite { .. } => format!("stream-write"), + StreamForward { .. } => format!("stream-forward"), StreamCancelRead { .. } => format!("stream-cancel-read"), StreamCancelWrite { .. } => format!("stream-cancel-write"), StreamDropReadable { .. } => format!("stream-drop-readable"), @@ -1232,6 +1253,7 @@ impl Trampoline { FutureNew { .. } => format!("future-new"), FutureRead { .. } => format!("future-read"), FutureWrite { .. } => format!("future-write"), + FutureForward { .. } => format!("future-forward"), FutureCancelRead { .. } => format!("future-cancel-read"), FutureCancelWrite { .. } => format!("future-cancel-write"), FutureDropReadable { .. } => format!("future-drop-readable"), diff --git a/crates/environ/src/component/translate.rs b/crates/environ/src/component/translate.rs index 4db699a8d5b4..0c6d3ff78085 100644 --- a/crates/environ/src/component/translate.rs +++ b/crates/environ/src/component/translate.rs @@ -242,6 +242,10 @@ enum LocalInitializer<'data> { ty: ComponentDefinedTypeId, options: LocalCanonicalOptions, }, + StreamForward { + ty: ComponentDefinedTypeId, + func: ModuleInternedTypeIndex, + }, StreamCancelRead { ty: ComponentDefinedTypeId, func: ModuleInternedTypeIndex, @@ -272,6 +276,10 @@ enum LocalInitializer<'data> { ty: ComponentDefinedTypeId, options: LocalCanonicalOptions, }, + FutureForward { + ty: ComponentDefinedTypeId, + func: ModuleInternedTypeIndex, + }, FutureCancelRead { ty: ComponentDefinedTypeId, func: ModuleInternedTypeIndex, @@ -1145,6 +1153,16 @@ impl<'a, 'data> Translator<'a, 'data> { core_func_index += 1; LocalInitializer::StreamWrite { ty, options } } + wasmparser::CanonicalFunction::StreamForward { ty } => { + let ty = self + .validator + .types(0) + .unwrap() + .component_defined_type_at(ty); + let func = self.core_func_signature(core_func_index)?; + core_func_index += 1; + LocalInitializer::StreamForward { ty, func } + } wasmparser::CanonicalFunction::StreamCancelRead { ty, async_ } => { let ty = self .validator @@ -1215,6 +1233,16 @@ impl<'a, 'data> Translator<'a, 'data> { core_func_index += 1; LocalInitializer::FutureWrite { ty, options } } + wasmparser::CanonicalFunction::FutureForward { ty } => { + let ty = self + .validator + .types(0) + .unwrap() + .component_defined_type_at(ty); + let func = self.core_func_signature(core_func_index)?; + core_func_index += 1; + LocalInitializer::FutureForward { ty, func } + } wasmparser::CanonicalFunction::FutureCancelRead { ty, async_ } => { let ty = self .validator diff --git a/crates/environ/src/component/translate/inline.rs b/crates/environ/src/component/translate/inline.rs index d5e8c41379e1..8a7751a6a74b 100644 --- a/crates/environ/src/component/translate/inline.rs +++ b/crates/environ/src/component/translate/inline.rs @@ -849,6 +849,21 @@ impl<'a> Inliner<'a> { )); frame.funcs.push((func, dfg::CoreDef::Trampoline(index))); } + StreamForward { ty, func } => { + let InterfaceType::Stream(ty) = + types.defined_type(frame.translation.types_ref(), *ty)? + else { + unreachable!() + }; + let index = self.result.trampolines.push(( + *func, + dfg::Trampoline::StreamForward { + instance: frame.instance, + ty, + }, + )); + frame.funcs.push((*func, dfg::CoreDef::Trampoline(index))); + } StreamCancelRead { ty, func, async_ } => { let InterfaceType::Stream(ty) = types.defined_type(frame.translation.types_ref(), *ty)? @@ -966,6 +981,21 @@ impl<'a> Inliner<'a> { )); frame.funcs.push((func, dfg::CoreDef::Trampoline(index))); } + FutureForward { ty, func } => { + let InterfaceType::Future(ty) = + types.defined_type(frame.translation.types_ref(), *ty)? + else { + unreachable!() + }; + let index = self.result.trampolines.push(( + *func, + dfg::Trampoline::FutureForward { + instance: frame.instance, + ty, + }, + )); + frame.funcs.push((*func, dfg::CoreDef::Trampoline(index))); + } FutureCancelRead { ty, func, async_ } => { let InterfaceType::Future(ty) = types.defined_type(frame.translation.types_ref(), *ty)? diff --git a/crates/wasmtime/src/runtime/component/concurrent.rs b/crates/wasmtime/src/runtime/component/concurrent.rs index 2bae1c0ccb8d..59266adbcfed 100644 --- a/crates/wasmtime/src/runtime/component/concurrent.rs +++ b/crates/wasmtime/src/runtime/component/concurrent.rs @@ -4205,6 +4205,15 @@ pub trait VMComponentAsyncStore { address: u32, ) -> Result; + /// The `future.forward` intrinsic. + fn future_forward( + &mut self, + instance: Instance, + ty: TypeFutureTableIndex, + reader: u32, + writer: u32, + ) -> Result<()>; + /// The `future.drop-writable` intrinsic. fn future_drop_writable( &mut self, @@ -4267,6 +4276,15 @@ pub trait VMComponentAsyncStore { count: u32, ) -> Result; + /// The `stream.forward` intrinsic. + fn stream_forward( + &mut self, + instance: Instance, + ty: TypeStreamTableIndex, + reader: u32, + writer: u32, + ) -> Result<()>; + /// The `stream.drop-writable` intrinsic. fn stream_drop_writable( &mut self, @@ -4494,6 +4512,21 @@ impl VMComponentAsyncStore for StoreInner { .map(|result| result.encode()) } + fn future_forward( + &mut self, + instance: Instance, + ty: TypeFutureTableIndex, + reader: u32, + writer: u32, + ) -> Result<()> { + instance.guest_forward( + StoreContextMut(self), + TransmitIndex::Future(ty), + reader, + writer, + ) + } + fn future_drop_writable( &mut self, instance: Instance, @@ -4561,6 +4594,21 @@ impl VMComponentAsyncStore for StoreInner { .map(|result| result.encode()) } + fn stream_forward( + &mut self, + instance: Instance, + ty: TypeStreamTableIndex, + reader: u32, + writer: u32, + ) -> Result<()> { + instance.guest_forward( + StoreContextMut(self), + TransmitIndex::Stream(ty), + reader, + writer, + ) + } + fn stream_drop_writable( &mut self, instance: Instance, diff --git a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs index 4daa4392766a..8d9fb11b2c7f 100644 --- a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs +++ b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs @@ -2203,6 +2203,14 @@ type PollStream = Box< dyn Fn() -> Pin> + Send + 'static>> + Send + Sync, >; +type ConsumeFn = Box< + dyn for<'a> Fn( + Option<&'a mut UntypedWriteBuffer<'a>>, + ) -> Pin> + Send + 'a>> + + Send + + Sync, +>; + type TryInto = Box Option> + Send + Sync>; /// Represents the state of the write end of a stream or future. @@ -2259,9 +2267,11 @@ enum ReadState { count: ItemCount, handle: u32, }, - /// The read end is owned by a host task, and it is ready to consume items. + /// The read end is owned by a host task, and it is ready to consume items, + /// either from the write end of the same stream or future (when passed + /// `None`) or from a host-owned buffer (when passed `Some`). HostReady { - consume: PollStream, + consume: ConsumeFn, guest_offset: ItemCount, cancel: bool, cancel_waker: Option, @@ -2707,6 +2717,7 @@ impl StoreContextMut<'_, T> { let (count, host_offset) = match &transmit.read { &ReadState::GuestReady { count, .. } => (count.as_u32(), 0), &ReadState::HostToHost { limit, .. } => (1, limit), + ReadState::Open => (0, 0), _ => bail_bug!("invalid read state"), }; let guest_offset = match &transmit.write { @@ -2894,11 +2905,15 @@ impl StoreContextMut<'_, T> { Ok(result) } }; - let consume = { + let consume: ConsumeFn = { let consume = consume_with_buffer.clone(); - Box::new(move || { + Box::new(move |input| { let consume = consume.clone(); - async move { consume(None).await }.boxed() + if let Some(input) = input { + async move { consume(Some(input.get_mut::())).await }.boxed() + } else { + async move { consume(None).await }.boxed() + } }) }; @@ -2912,7 +2927,7 @@ impl StoreContextMut<'_, T> { }; } &WriteState::GuestReady { .. } => { - let future = consume(); + let future = consume(None); transmit.read = ReadState::HostReady { consume, guest_offset: ItemCount::ZERO, @@ -3145,6 +3160,8 @@ async fn write Ok(()), + _ => bail_bug!("unexpected read state"), } } @@ -3157,11 +3174,11 @@ impl Instance { store: &mut dyn VMStore, kind: TransmitKind, transmit_id: TableId, - consume: PollStream, + consume: ConsumeFn, guest_offset: ItemCount, cancel: bool, ) -> Result { - let mut future = consume(); + let mut future = consume(None); store.concurrent_state_mut()?.get_mut(transmit_id)?.read = ReadState::HostReady { consume, guest_offset, @@ -3775,9 +3792,71 @@ impl Instance { *state = TransmitLocalState::Busy; let transmit_handle = TableId::::new(rep); let caller_thread = store.0.current_guest_thread()?; + let transmit_id = store + .0 + .concurrent_state_mut()? + .get_mut(transmit_handle)? + .state; + + let mut result = self.perform_guest_read( + store.as_context_mut(), + transmit_id, + ty, + options, + flat_abi, + handle, + address, + count, + caller_instance, + caller_thread, + )?; + + if result == ReturnCode::Blocked && !self.options(store.0, options).async_ { + result = self.wait_for_read(store.0, transmit_handle)?; + } + + if result != ReturnCode::Blocked { + *self.id().get_mut(store.0).get_mut_by_index(ty, handle)?.1 = + TransmitLocalState::Read { + done: matches!( + (result, ty), + (ReturnCode::Dropped(_), TransmitIndex::Stream(_)) + ), + }; + } + + log::trace!( + "guest_read result for {transmit_handle:?} (handle {handle}; state {transmit_id:?}): {result:?}", + ); + + Ok(result) + } + + /// Match a guest read with the current write state of the specified stream + /// or future, either completing it immediately or leaving it pending in + /// `ReadState::GuestReady`. + /// + /// This is shared between `guest_read`, which issues a fresh read on + /// behalf of the calling guest, and `guest_forward`, which transfers a + /// read already pending on the destination of a forward to the fused + /// stream or future. + fn perform_guest_read( + self, + mut store: StoreContextMut, + transmit_id: TableId, + ty: TransmitIndex, + options: OptionsIndex, + flat_abi: Option, + handle: u32, + address: usize, + count: ItemCount, + caller_instance: RuntimeComponentInstanceIndex, + caller_thread: QualifiedThreadId, + ) -> Result { let concurrent_state = store.0.concurrent_state_mut()?; - let transmit_id = concurrent_state.get_mut(transmit_handle)?.state; let transmit = concurrent_state.get_mut(transmit_id)?; + let transmit_handle = transmit.read_handle; + let rep = transmit_handle.rep(); log::trace!( "guest_read {count} from {transmit_handle:?} (handle {handle}; state {transmit_id:?}); {:?}", transmit.write @@ -3812,7 +3891,7 @@ impl Instance { Ok::<_, crate::Error>(()) }; - let mut result = match mem::replace(&mut transmit.write, new_state) { + let result = match mem::replace(&mut transmit.write, new_state) { WriteState::GuestReady { instance: write_instance, ty: write_ty, @@ -3954,25 +4033,518 @@ impl Instance { WriteState::Dropped => ReturnCode::Dropped(ItemCount::ZERO), }; - if result == ReturnCode::Blocked && !self.options(store.0, options).async_ { - result = self.wait_for_read(store.0, transmit_handle)?; + Ok(result) + } + + /// Implements the `{stream,future}.forward` intrinsics. + /// + /// This fuses the stream or future whose readable end is `src` (the + /// source) with the stream or future whose writable end is `dst` (the + /// destination), transferring both ends out of the calling component + /// instance. The consumer of the destination's readable end and the + /// producer of the source's writable end are connected directly, as if the + /// source's readable end had been transferred to the consumer in place of + /// the destination's readable end. + /// + /// Rather than recording a delegation from the destination to the source + /// as the Canonical ABI specification does, the two `TransmitState`s are + /// eagerly merged into one, which also means chains of forwards always + /// collapse to a single stream or future. Which of the two states + /// survives depends on which ends are host-owned, since + /// `StreamProducer`/`StreamConsumer` closures capture the + /// `TableId` they were registered with: + /// + /// * By default the source's state survives and the consumer's readable + /// end is re-pointed at it. + /// + /// * If the consumer is host-owned (`ReadState::HostReady`), the + /// destination's state survives and the producer's writable end is + /// re-pointed at it instead. + /// + /// * If the producer and consumer are both host-owned, neither state can + /// be discarded; the source's producer is piped directly to the + /// consumer, keeping the destination alive as a detached stub for the + /// consumer's closures. + pub(super) fn guest_forward( + self, + mut store: StoreContextMut, + ty: TransmitIndex, + src: u32, + dst: u32, + ) -> Result<()> { + let kind = ty.kind(); + let desc = match kind { + TransmitKind::Stream => "stream", + TransmitKind::Future => "future", + }; + + let (src_rep, src_state) = self.id().get_mut(store.0).get_mut_by_index(ty, src)?; + let TransmitLocalState::Read { done: src_done } = *src_state else { + bail!(Trap::ConcurrentFutureStreamOp); + }; + if src_done { + bail!("cannot forward from {desc} after being notified that the writable end dropped"); } + let src_transmit_handle = TableId::::new(src_rep); - if result != ReturnCode::Blocked { - *self.id().get_mut(store.0).get_mut_by_index(ty, handle)?.1 = - TransmitLocalState::Read { - done: matches!( - (result, ty), - (ReturnCode::Dropped(_), TransmitIndex::Stream(_)) - ), - }; + let (dst_rep, dst_state) = self.id().get_mut(store.0).get_mut_by_index(ty, dst)?; + let TransmitLocalState::Write { done: dst_done } = *dst_state else { + bail!(Trap::ConcurrentFutureStreamOp); + }; + if dst_done { + bail!("cannot forward to {desc} after being notified that the readable end dropped"); + } + let dst_transmit_handle = TableId::::new(dst_rep); + + let concurrent_state = store.0.concurrent_state_mut()?; + + let src_transmit_id = concurrent_state.get_mut(src_transmit_handle)?.state; + let dst_transmit_id = concurrent_state.get_mut(dst_transmit_handle)?.state; + + // Since forwards eagerly fuse the source and destination, a forward + // which would make a stream or future its own (transitive) source is + // exactly a forward within a single already-fused stream or future. + if src_transmit_id == dst_transmit_id { + bail!("cannot forward a stream or future into itself"); + } + + if concurrent_state + .get_mut(src_transmit_handle)? + .common + .set + .is_some() + || concurrent_state + .get_mut(dst_transmit_handle)? + .common + .set + .is_some() + { + bail!("cannot forward while either end is in a waitable set"); + } + + let src_transmit = concurrent_state.get_mut(src_transmit_id)?; + if src_transmit.done { + bail!("cannot forward from future after previous read succeeded"); } + if !matches!(src_transmit.read, ReadState::Open) { + bail_bug!("expected `ReadState::Open`; got `{:?}`", src_transmit.read); + } + let src_write_handle = src_transmit.write_handle; + + let dst_transmit = concurrent_state.get_mut(dst_transmit_id)?; + if dst_transmit.done { + bail!("cannot forward to future after previous write succeeded"); + } + if !matches!(dst_transmit.write, WriteState::Open) { + bail_bug!( + "expected `WriteState::Open`; got `{:?}`", + dst_transmit.write + ); + } + let dst_read_handle = dst_transmit.read_handle; + + let table = self.id().get_mut(store.0).table_for_transmit(ty); + match ty { + TransmitIndex::Stream(ty) => { + table.stream_remove_readable(ty, src)?; + table.stream_remove_writable(ty, dst)?; + } + TransmitIndex::Future(ty) => { + table.future_remove_readable(ty, src)?; + table.future_remove_writable(ty, dst)?; + } + } + let concurrent_state = store.0.concurrent_state_mut()?; + concurrent_state.get_mut(src_transmit_handle)?.common.handle = None; + concurrent_state.get_mut(dst_transmit_handle)?.common.handle = None; log::trace!( - "guest_read result for {transmit_handle:?} (handle {handle}; state {transmit_id:?}): {result:?}", + "guest_forward from {src_transmit_handle:?} to {dst_transmit_handle:?} \ + (src handle {src} src state {src_transmit_id:?}; \ + dst handle {dst} dst state {dst_transmit_id:?})", ); - Ok(result) + // If the destination's readable end has already been dropped there is + // nothing left to forward into: discard the destination and drop the + // source's readable end, notifying the source's producer. + if let ReadState::Dropped = concurrent_state.get_mut(dst_transmit_id)?.read { + store.0.host_drop_writer(dst_transmit_handle, None)?; + return store.0.host_drop_reader(src_transmit_handle, kind); + } + + // A pending read or write which has already made partial progress has + // an undelivered completion event; complete it at its current + // progress (as the Canonical ABI specifies for reads, and which the + // eager merge extends to writes) rather than transfer it to the fused + // stream. If the opposite end is already dropped, the drop + // notification is merged into the pending event further down, so the + // consumer or producer observes the single `DROPPED` event it would + // have seen without the forward. + if concurrent_state + .get_mut(dst_read_handle)? + .common + .event + .is_some() + { + let dst_transmit = concurrent_state.get_mut(dst_transmit_id)?; + if let ReadState::GuestReady { .. } = &dst_transmit.read { + dst_transmit.read = ReadState::Open; + } + } + if concurrent_state + .get_mut(src_write_handle)? + .common + .event + .is_some() + { + let src_transmit = concurrent_state.get_mut(src_transmit_id)?; + if let WriteState::GuestReady { .. } = &src_transmit.write { + src_transmit.write = WriteState::Open; + } + } + + // If the producer and consumer are both host-owned, connect them + // directly, mirroring the `WriteState::HostReady` arm of + // `set_consumer`. Neither state can be discarded: the source hosts + // the pipe itself while the destination stays alive as a detached + // stub for the consumer's closures, and both are deleted once piping + // finishes. + if matches!( + concurrent_state.get_mut(src_transmit_id)?.write, + WriteState::HostReady { .. } + ) && matches!( + concurrent_state.get_mut(dst_transmit_id)?.read, + ReadState::HostReady { .. } + ) { + let dst_transmit = concurrent_state.get_mut(dst_transmit_id)?; + let ReadState::HostReady { + consume, + guest_offset, + cancel, + cancel_waker, + } = mem::replace(&mut dst_transmit.read, ReadState::Open) + else { + unreachable!() + }; + if cancel_waker.is_some() { + bail_bug!("expected cancel_waker to be none"); + } + if cancel { + bail_bug!("expected cancel to be false"); + } + if guest_offset != 0 { + bail_bug!("expected guest_offset to be 0"); + } + // The consumer's closures use the destination for bookkeeping, + // expecting the write state a `Source` fed from a host buffer may + // have; give it a placeholder. + dst_transmit.write = WriteState::HostReady { + produce: Box::new(|| { + Box::pin(async { bail_bug!("unexpected invocation of `produce`") }) + }), + try_into: Box::new(|_| None), + guest_offset: ItemCount::ZERO, + cancel: false, + cancel_waker: None, + }; + + let src_transmit = concurrent_state.get_mut(src_transmit_id)?; + let WriteState::HostReady { produce, .. } = mem::replace( + &mut src_transmit.write, + WriteState::HostReady { + produce: Box::new(|| { + Box::pin(async { bail_bug!("unexpected invocation of `produce`") }) + }), + try_into: Box::new(|_| None), + guest_offset: ItemCount::ZERO, + cancel: false, + cancel_waker: None, + }, + ) else { + unreachable!() + }; + + src_transmit.read = ReadState::HostToHost { + accept: Box::new(move |input| consume(Some(input))), + buffer: Vec::new(), + limit: 0, + }; + + let future = async move { + loop { + if tls::get(|store| { + crate::error::Ok(matches!( + store.concurrent_state_mut()?.get_mut(src_transmit_id)?.read, + ReadState::Dropped + )) + })? { + break Ok(()); + } + + match produce().await? { + StreamResult::Completed | StreamResult::Cancelled => {} + StreamResult::Dropped => break Ok(()), + } + + if let TransmitKind::Future = kind { + break Ok(()); + } + } + } + .map(move |result| { + tls::get(|store| { + let state = store.concurrent_state_mut()?; + state.delete_transmit(src_transmit_id)?; + state.delete_transmit(dst_transmit_id)?; + crate::error::Ok(()) + })?; + result + }); + + concurrent_state.push_future(Box::pin(future)); + return Ok(()); + } + + if let ReadState::HostReady { .. } = concurrent_state.get_mut(dst_transmit_id)?.read { + // Keep the destination: discard the source's state after + // migrating its write state and re-pointing the producer's + // writable end. + let src_transmit = concurrent_state.get_mut(src_transmit_id)?; + let origin = src_transmit.origin; + let src_write = mem::replace(&mut src_transmit.write, WriteState::Dropped); + concurrent_state.delete(src_transmit_id)?; + concurrent_state.delete(src_transmit_handle)?; + + match src_write { + WriteState::Open => { + concurrent_state.get_mut(src_write_handle)?.state = dst_transmit_id; + let dst_transmit = concurrent_state.get_mut(dst_transmit_id)?; + dst_transmit.write_handle = src_write_handle; + dst_transmit.origin = origin; + concurrent_state.delete(dst_transmit_handle)?; + Ok(()) + } + write @ WriteState::GuestReady { .. } => { + concurrent_state.get_mut(src_write_handle)?.state = dst_transmit_id; + let dst_transmit = concurrent_state.get_mut(dst_transmit_id)?; + dst_transmit.write_handle = src_write_handle; + dst_transmit.origin = origin; + + // Mirror the `ReadState::HostReady` arm of + // `guest_write`, except that the pending write's + // completion is delivered as an event. + let ReadState::HostReady { + consume, + guest_offset, + cancel, + cancel_waker, + } = mem::replace(&mut dst_transmit.read, ReadState::Open) + else { + unreachable!() + }; + if cancel_waker.is_some() { + bail_bug!("expected cancel_waker to be none"); + } + if cancel { + bail_bug!("expected cancel to be false"); + } + if guest_offset != 0 { + bail_bug!("expected guest_offset to be 0"); + } + if let TransmitIndex::Future(_) = ty { + dst_transmit.done = true; + } + let (write_ty, write_index) = match &write { + WriteState::GuestReady { ty, handle, .. } => (*ty, *handle), + _ => unreachable!(), + }; + dst_transmit.write = write; + concurrent_state.delete(dst_transmit_handle)?; + + let code = self.consume( + store.0, + kind, + dst_transmit_id, + consume, + ItemCount::ZERO, + false, + )?; + if code != ReturnCode::Blocked { + store.0.concurrent_state_mut()?.send_write_result( + write_ty, + dst_transmit_id, + write_index, + code, + )?; + } + Ok(()) + } + WriteState::Dropped => { + // The source's producer is already gone; deleting the + // destination is what notifies the host-owned + // consumer, mirroring `host_drop_writer`. + concurrent_state.delete(src_write_handle)?; + concurrent_state.delete_transmit(dst_transmit_id) + } + WriteState::HostReady { .. } => unreachable!(), + } + } else { + // Keep the source: discard the destination's state after migrating + // its read state and re-pointing the consumer's readable end. + let dst_transmit = concurrent_state.get_mut(dst_transmit_id)?; + let dst_read = mem::replace(&mut dst_transmit.read, ReadState::Dropped); + concurrent_state.delete(dst_transmit_id)?; + concurrent_state.delete(dst_transmit_handle)?; + concurrent_state.get_mut(dst_read_handle)?.state = src_transmit_id; + concurrent_state.get_mut(src_transmit_id)?.read_handle = dst_read_handle; + concurrent_state.delete(src_transmit_handle)?; + + match dst_read { + ReadState::Open => { + // If the consumer has an undelivered partial-progress + // event and the source's producer is host-owned, poll the + // producer once so that a source which has already ended + // is observed as `WriteState::Dropped` below. This + // mirrors the `src.dropped` check in the Canonical ABI's + // `forward`, which merges the end of the stream into the + // partially-copied pending read's completion, delivering + // a single `DROPPED` event. A producer which is not yet + // ready is left alone; its end of stream (if any) will be + // observed by a later read. + if let TransmitKind::Stream = kind { + let state = store.0.concurrent_state_mut()?; + if matches!( + state.get_mut(dst_read_handle)?.common.event, + Some(Event::StreamRead { + code: ReturnCode::Completed(_), + .. + }) + ) && matches!( + state.get_mut(src_transmit_id)?.write, + WriteState::HostReady { .. } + ) { + let WriteState::HostReady { + produce, + try_into, + guest_offset, + cancel, + cancel_waker, + } = mem::replace( + &mut state.get_mut(src_transmit_id)?.write, + WriteState::Open, + ) + else { + unreachable!() + }; + if cancel_waker.is_some() { + bail_bug!("expected cancel_waker to be none"); + } + if cancel { + bail_bug!("expected cancel to be false"); + } + if guest_offset != 0 { + bail_bug!("expected guest_offset to be 0"); + } + + let mut future = produce(); + state.get_mut(src_transmit_id)?.write = WriteState::HostReady { + produce, + try_into, + guest_offset: ItemCount::ZERO, + cancel: false, + cancel_waker: None, + }; + let poll = tls::set(store.0, || { + future + .as_mut() + .poll(&mut Context::from_waker(&Waker::noop())) + }); + match poll { + Poll::Ready(result) => { + let transmit = + store.0.concurrent_state_mut()?.get_mut(src_transmit_id)?; + settle_host_write(transmit, kind, result?)?; + } + Poll::Pending => { + // Dropping the future returns the + // producer to its lock; clear the waker + // it stored, which will never be woken. + drop(future); + if let WriteState::HostReady { cancel_waker, .. } = &mut store + .0 + .concurrent_state_mut()? + .get_mut(src_transmit_id)? + .write + { + *cancel_waker = None; + } + } + } + } + } + + // Mirror the `ReadState::Open` arm of `host_drop_writer`: + // a consumer which would have learned of the drop from the + // destination must learn of it from the fused stream or + // future. + let concurrent_state = store.0.concurrent_state_mut()?; + if let WriteState::Dropped = concurrent_state.get_mut(src_transmit_id)?.write { + concurrent_state.update_event( + dst_read_handle.rep(), + match kind { + TransmitKind::Future => Event::FutureRead { + code: ReturnCode::Dropped(ItemCount::ZERO), + pending: None, + }, + TransmitKind::Stream => Event::StreamRead { + code: ReturnCode::Dropped(ItemCount::ZERO), + pending: None, + }, + }, + )?; + } + Ok(()) + } + ReadState::GuestReady { + ty: read_ty, + flat_abi, + options, + address, + count, + handle, + instance, + caller_instance, + caller_thread, + } => { + // Transfer the pending read to the fused stream or future, + // delivering its completion (if any) as an event. + let code = instance.perform_guest_read( + store.as_context_mut(), + src_transmit_id, + read_ty, + options, + flat_abi, + handle, + address, + count, + caller_instance, + caller_thread, + )?; + if code != ReturnCode::Blocked { + store.0.concurrent_state_mut()?.send_read_result( + read_ty, + src_transmit_id, + handle, + code, + )?; + } + Ok(()) + } + ReadState::HostReady { .. } | ReadState::HostToHost { .. } | ReadState::Dropped => { + bail_bug!("unexpected read state") + } + } + } } fn wait_for_write( diff --git a/crates/wasmtime/src/runtime/vm/component/libcalls.rs b/crates/wasmtime/src/runtime/vm/component/libcalls.rs index 11ec01108492..ad2a025a2e7d 100644 --- a/crates/wasmtime/src/runtime/vm/component/libcalls.rs +++ b/crates/wasmtime/src/runtime/vm/component/libcalls.rs @@ -1013,6 +1013,23 @@ fn future_read( ) } +#[cfg(feature = "component-model-async")] +fn future_forward( + store: &mut dyn VMStore, + instance: Instance, + _caller_instance: u32, + ty: u32, + reader: u32, + writer: u32, +) -> Result<()> { + store.component_async_store().future_forward( + instance, + TypeFutureTableIndex::from_u32(ty), + reader, + writer, + ) +} + #[cfg(feature = "component-model-async")] fn future_cancel_write( store: &mut dyn VMStore, @@ -1127,6 +1144,23 @@ fn stream_read( ) } +#[cfg(feature = "component-model-async")] +fn stream_forward( + store: &mut dyn VMStore, + instance: Instance, + _caller_instance: u32, + ty: u32, + reader: u32, + writer: u32, +) -> Result<()> { + store.component_async_store().stream_forward( + instance, + TypeStreamTableIndex::from_u32(ty), + reader, + writer, + ) +} + #[cfg(feature = "component-model-async")] fn stream_cancel_write( store: &mut dyn VMStore, diff --git a/tests/all/component_model.rs b/tests/all/component_model.rs index 43b6a9b194f6..de6e71543068 100644 --- a/tests/all/component_model.rs +++ b/tests/all/component_model.rs @@ -14,6 +14,7 @@ mod bindgen; mod call_hook; mod dynamic; mod fixed_length_list; +mod forward; mod func; mod import; mod instance; diff --git a/tests/all/component_model/forward.rs b/tests/all/component_model/forward.rs new file mode 100644 index 000000000000..026798f99466 --- /dev/null +++ b/tests/all/component_model/forward.rs @@ -0,0 +1,1405 @@ +//! Tests for the `stream.forward` and `future.forward` built-ins with +//! host-owned producers and/or consumers. Guest-to-guest forwarding is +//! covered by `tests/component-model/test/async/forward-{stream,future}.wast`. + +use std::collections::VecDeque; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::task::{Context, Poll, Waker}; +use wasmtime::component::{ + Component, Destination, FutureConsumer, FutureReader, Linker, Source, StreamConsumer, + StreamProducer, StreamReader, StreamResult, +}; +use wasmtime::{Config, Engine, Result, Store, StoreContextMut}; + +const BLOCKED: u32 = 0xffff_ffff; +const COMPLETED: u32 = 0; +const DROPPED: u32 = 1; +const COMPLETED_THREE_ITEMS: u32 = 3 << 4; +const DROPPED_THREE_ITEMS: u32 = (3 << 4) | 1; +const CANCELLED_THREE_ITEMS: u32 = (3 << 4) | 2; +const DROPPED_ONE_ITEM: u32 = (1 << 4) | 1; +const CANCELLED_ONE_ITEM: u32 = (1 << 4) | 2; + +/// A `StreamConsumer` which collects everything it receives and flags its own +/// destruction (which is how the host is notified that the stream ended). +struct CollectConsumer { + data: Arc>>, + dropped: Arc, +} + +impl Drop for CollectConsumer { + fn drop(&mut self) { + self.dropped.store(true, Ordering::SeqCst); + } +} + +impl StreamConsumer<()> for CollectConsumer { + type Item = u8; + + fn poll_consume( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + store: StoreContextMut<()>, + mut source: Source, + _finish: bool, + ) -> Poll> { + let mut buffer = Vec::with_capacity(64); + source.read(store, &mut buffer)?; + self.data.lock().unwrap().extend(buffer); + Poll::Ready(Ok(StreamResult::Completed)) + } +} + +/// State shared between a `ChunkProducer` and the test driving it. +#[derive(Default)] +struct ChunkState { + chunks: VecDeque, + ended: bool, + waker: Option, +} + +impl ChunkState { + fn push(state: &Mutex, chunk: &'static [u8]) { + let mut state = state.lock().unwrap(); + state.chunks.push_back(bytes::Bytes::from_static(chunk)); + if let Some(waker) = state.waker.take() { + waker.wake(); + } + } + + fn end(state: &Mutex) { + let mut state = state.lock().unwrap(); + state.ended = true; + if let Some(waker) = state.waker.take() { + waker.wake(); + } + } +} + +/// A `StreamProducer` which stays live until the test explicitly ends it, +/// producing whatever chunks the test has pushed in the meantime, and which +/// flags its own destruction (which is how the test knows the runtime has +/// finished with it). +struct ChunkProducer { + state: Arc>, + dropped: Arc, +} + +impl Drop for ChunkProducer { + fn drop(&mut self) { + self.dropped.store(true, Ordering::SeqCst); + } +} + +impl StreamProducer<()> for ChunkProducer { + type Item = u8; + type Buffer = bytes::Bytes; + + fn poll_produce<'a>( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + _store: StoreContextMut<'a, ()>, + mut dst: Destination<'a, Self::Item, Self::Buffer>, + finish: bool, + ) -> Poll> { + let mut state = self.state.lock().unwrap(); + if let Some(chunk) = state.chunks.pop_front() { + dst.set_buffer(chunk); + Poll::Ready(Ok(StreamResult::Completed)) + } else if state.ended { + Poll::Ready(Ok(StreamResult::Dropped)) + } else if finish { + Poll::Ready(Ok(StreamResult::Cancelled)) + } else { + state.waker = Some(cx.waker().clone()); + Poll::Pending + } + } +} + +/// A guest component which can create streams, forward into them, and +/// read/write on either side of a forward. +/// +/// Memory layout: bytes read land at address 0; bytes written come from +/// address 16. +const FORWARDER: &str = r#" +(component + (core module $libc (memory (export "m") 1)) + (core instance $libc (instantiate $libc)) + + (type $s (stream u8)) + (core func $stream.new (canon stream.new $s)) + (core func $stream.read (canon stream.read $s async (memory (core memory $libc "m")))) + (core func $stream.write (canon stream.write $s async (memory (core memory $libc "m")))) + (core func $stream.forward (canon stream.forward $s)) + (core func $stream.cancel-read (canon stream.cancel-read $s async)) + (core func $stream.cancel-write (canon stream.cancel-write $s async)) + (core func $stream.drop-writable (canon stream.drop-writable $s)) + + (core module $m + (import "" "m" (memory 1)) + (import "" "stream.new" (func $stream.new (result i64))) + (import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32))) + (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32))) + (import "" "stream.forward" (func $stream.forward (param i32 i32))) + (import "" "stream.cancel-read" (func $stream.cancel-read (param i32) (result i32))) + (import "" "stream.cancel-write" (func $stream.cancel-write (param i32) (result i32))) + (import "" "stream.drop-writable" (func $stream.drop-writable (param i32))) + + ;; The writable end of the stream returned by `mk`. + (global $dst-w (mut i32) (i32.const 0)) + ;; The ends of the stream created by `mk-src`. + (global $src-r (mut i32) (i32.const 0)) + (global $src-w (mut i32) (i32.const 0)) + + ;; Create a new stream, saving its writable end and returning its + ;; readable end (which the host will consume). + (func (export "mk") (result i32) + (local $tmp i64) + (local.set $tmp (call $stream.new)) + (global.set $dst-w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + (i32.wrap_i64 (local.get $tmp)) + ) + + ;; Create a new stream, saving both ends (which this component will + ;; produce into). + (func (export "mk-src") + (local $tmp i64) + (local.set $tmp (call $stream.new)) + (global.set $src-r (i32.wrap_i64 (local.get $tmp))) + (global.set $src-w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + ) + + ;; Forward the stream created by `mk-src` into the stream created by `mk`. + (func (export "fwd-src") + (call $stream.forward (global.get $src-r) (global.get $dst-w)) + ) + + ;; Forward the given stream into the stream created by `mk`. + (func (export "fwd") (param $r i32) + (call $stream.forward (local.get $r) (global.get $dst-w)) + ) + + ;; Write "xyz" to the stream created by `mk-src`, returning the packed + ;; result code. + (func (export "write") (result i32) + (i32.store8 (i32.const 16) (i32.const 120)) + (i32.store8 (i32.const 17) (i32.const 121)) + (i32.store8 (i32.const 18) (i32.const 122)) + (call $stream.write (global.get $src-w) (i32.const 16) (i32.const 3)) + ) + + ;; Retrieve the completion of a pending write on the stream created by + ;; `mk-src` by cancelling it, returning the packed result code. + (func (export "check-write") (result i32) + (call $stream.cancel-write (global.get $src-w)) + ) + + ;; Drop the writable end of the stream created by `mk-src`. + (func (export "drop-w") + (call $stream.drop-writable (global.get $src-w)) + ) + + ;; Forward the given stream into a fresh stream, then read three bytes + ;; from the latter, checking that "abc" arrived. + (func (export "run") (param $r i32) (result i32) + (local $tmp i64) (local $r2 i32) (local $w2 i32) (local $code i32) + (local.set $tmp (call $stream.new)) + (local.set $r2 (i32.wrap_i64 (local.get $tmp))) + (local.set $w2 (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + + (call $stream.forward (local.get $r) (local.get $w2)) + + (local.set $code (call $stream.read (local.get $r2) (i32.const 0) (i32.const 3))) + (call $check-abc) + (local.get $code) + ) + + ;; Like `run`, except the read is issued (and blocks) before the forward, + ;; in which case its completion is delivered as an event, retrieved here + ;; via `stream.cancel-read`. + (func (export "run-pending") (param $r i32) (result i32) + (local $tmp i64) (local $r2 i32) (local $w2 i32) (local $code i32) + (local.set $tmp (call $stream.new)) + (local.set $r2 (i32.wrap_i64 (local.get $tmp))) + (local.set $w2 (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + + (call $stream.read (local.get $r2) (i32.const 0) (i32.const 3)) + i32.const -1 ;; BLOCKED + i32.ne + if unreachable end + + (call $stream.forward (local.get $r) (local.get $w2)) + + (local.set $code (call $stream.cancel-read (local.get $r2))) + (call $check-abc) + (local.get $code) + ) + + ;; Read eight bytes from a fresh stream (which blocks), write "abc" into + ;; it (leaving an undelivered completion with three bytes copied), then + ;; forward the given stream into it and retrieve the completion via + ;; `stream.cancel-read`, returning the packed result code. + (func (export "run-partial") (param $r i32) (result i32) + (local $tmp i64) (local $r2 i32) (local $w2 i32) (local $code i32) + (local.set $tmp (call $stream.new)) + (local.set $r2 (i32.wrap_i64 (local.get $tmp))) + (local.set $w2 (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + + (call $stream.read (local.get $r2) (i32.const 0) (i32.const 8)) + i32.const -1 ;; BLOCKED + i32.ne + if unreachable end + + (i32.store8 (i32.const 16) (i32.const 97)) + (i32.store8 (i32.const 17) (i32.const 98)) + (i32.store8 (i32.const 18) (i32.const 99)) + (call $stream.write (local.get $w2) (i32.const 16) (i32.const 3)) + i32.const 48 ;; COMPLETED, three items + i32.ne + if unreachable end + + (call $stream.forward (local.get $r) (local.get $w2)) + + (local.set $code (call $stream.cancel-read (local.get $r2))) + (call $check-abc) + (local.get $code) + ) + + (func $check-abc + (i32.ne (i32.load8_u (i32.const 0)) (i32.const 97)) + if unreachable end + (i32.ne (i32.load8_u (i32.const 1)) (i32.const 98)) + if unreachable end + (i32.ne (i32.load8_u (i32.const 2)) (i32.const 99)) + if unreachable end + ) + ) + + (core instance $i (instantiate $m + (with "" (instance + (export "m" (memory $libc "m")) + (export "stream.new" (func $stream.new)) + (export "stream.read" (func $stream.read)) + (export "stream.write" (func $stream.write)) + (export "stream.forward" (func $stream.forward)) + (export "stream.cancel-read" (func $stream.cancel-read)) + (export "stream.cancel-write" (func $stream.cancel-write)) + (export "stream.drop-writable" (func $stream.drop-writable)) + )) + )) + + (func (export "mk") (result (stream u8)) (canon lift (core func $i "mk"))) + (func (export "mk-src") (canon lift (core func $i "mk-src"))) + (func (export "fwd-src") (canon lift (core func $i "fwd-src"))) + (func (export "fwd") (param "s" (stream u8)) (canon lift (core func $i "fwd"))) + (func (export "write") (result u32) (canon lift (core func $i "write"))) + (func (export "check-write") (result u32) (canon lift (core func $i "check-write"))) + (func (export "drop-w") (canon lift (core func $i "drop-w"))) + (func (export "run") (param "s" (stream u8)) (result u32) (canon lift (core func $i "run"))) + (func (export "run-pending") (param "s" (stream u8)) (result u32) + (canon lift (core func $i "run-pending"))) + (func (export "run-partial") (param "s" (stream u8)) (result u32) + (canon lift (core func $i "run-partial"))) +) +"#; + +async fn instantiate( + store: &mut Store<()>, + engine: &Engine, + wat: &str, +) -> Result { + let component = Component::new(engine, wat)?; + Linker::new(engine) + .instantiate_async(store, &component) + .await +} + +fn engine() -> Result { + let mut config = Config::new(); + config.wasm_component_model_async(true); + config.wasm_component_model_async_stackful(true); + config.wasm_component_model_more_async_builtins(true); + Engine::new(&config) +} + +/// Host producer, forwarded by the guest into a stream the guest then reads. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_producer_to_guest() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FORWARDER).await?; + let run = instance.get_typed_func::<(StreamReader,), (u32,)>(&mut store, "run")?; + + let reader = StreamReader::new(&mut store, bytes::Bytes::from_static(b"abc"))?; + assert_eq!( + run.call_async(&mut store, (reader,)).await?, + (DROPPED_THREE_ITEMS,) + ); + + Ok(()) +} + +/// Host producer, forwarded by the guest into a stream the guest was already +/// blocked reading from. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_producer_to_pending_guest_read() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FORWARDER).await?; + let run = instance.get_typed_func::<(StreamReader,), (u32,)>(&mut store, "run-pending")?; + + let reader = StreamReader::new(&mut store, bytes::Bytes::from_static(b"abc"))?; + assert_eq!( + run.call_async(&mut store, (reader,)).await?, + (DROPPED_THREE_ITEMS,) + ); + + Ok(()) +} + +/// Host producer which has already ended, forwarded by the guest into a +/// stream with a partially-copied pending read: the read's progress and the +/// end of the stream must be merged into a single `DROPPED` event. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_ended_host_producer_to_partial_guest_read() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FORWARDER).await?; + let run = instance.get_typed_func::<(StreamReader,), (u32,)>(&mut store, "run-partial")?; + + let state = Arc::new(Mutex::new(ChunkState { + ended: true, + ..ChunkState::default() + })); + let dropped = Arc::new(AtomicBool::new(false)); + let reader = StreamReader::new( + &mut store, + ChunkProducer { + state, + dropped: dropped.clone(), + }, + )?; + assert_eq!( + run.call_async(&mut store, (reader,)).await?, + (DROPPED_THREE_ITEMS,) + ); + assert!(dropped.load(Ordering::SeqCst)); + + Ok(()) +} + +/// Guest producer, forwarded by the guest into a stream consumed by the host. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_guest_producer_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FORWARDER).await?; + let mk = instance.get_typed_func::<(), (StreamReader,)>(&mut store, "mk")?; + let mk_src = instance.get_typed_func::<(), ()>(&mut store, "mk-src")?; + let fwd_src = instance.get_typed_func::<(), ()>(&mut store, "fwd-src")?; + let write = instance.get_typed_func::<(), (u32,)>(&mut store, "write")?; + let drop_w = instance.get_typed_func::<(), ()>(&mut store, "drop-w")?; + + let data = Arc::new(Mutex::new(Vec::new())); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + CollectConsumer { + data: data.clone(), + dropped: dropped.clone(), + }, + )?; + + mk_src.call_async(&mut store, ()).await?; + fwd_src.call_async(&mut store, ()).await?; + assert_eq!( + write.call_async(&mut store, ()).await?, + (COMPLETED_THREE_ITEMS,) + ); + + assert_eq!(*data.lock().unwrap(), b"xyz"); + + assert!(!dropped.load(Ordering::SeqCst)); + drop_w.call_async(&mut store, ()).await?; + store + .run_concurrent(async |_| { + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await?; + + Ok(()) +} + +/// Guest producer with a write already pending when the guest forwards into a +/// stream consumed by the host. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_pending_guest_write_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FORWARDER).await?; + let mk = instance.get_typed_func::<(), (StreamReader,)>(&mut store, "mk")?; + let mk_src = instance.get_typed_func::<(), ()>(&mut store, "mk-src")?; + let fwd_src = instance.get_typed_func::<(), ()>(&mut store, "fwd-src")?; + let write = instance.get_typed_func::<(), (u32,)>(&mut store, "write")?; + let check_write = instance.get_typed_func::<(), (u32,)>(&mut store, "check-write")?; + let drop_w = instance.get_typed_func::<(), ()>(&mut store, "drop-w")?; + + let data = Arc::new(Mutex::new(Vec::new())); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + CollectConsumer { + data: data.clone(), + dropped: dropped.clone(), + }, + )?; + + mk_src.call_async(&mut store, ()).await?; + assert_eq!(write.call_async(&mut store, ()).await?, (BLOCKED,)); + fwd_src.call_async(&mut store, ()).await?; + + assert_eq!(*data.lock().unwrap(), b"xyz"); + assert_eq!( + check_write.call_async(&mut store, ()).await?, + (CANCELLED_THREE_ITEMS,) + ); + + assert!(!dropped.load(Ordering::SeqCst)); + drop_w.call_async(&mut store, ()).await?; + store + .run_concurrent(async |_| { + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await?; + + Ok(()) +} + +/// Host producer forwarded by the guest into a stream already being consumed +/// by the host. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_producer_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FORWARDER).await?; + let mk = instance.get_typed_func::<(), (StreamReader,)>(&mut store, "mk")?; + let fwd = instance.get_typed_func::<(StreamReader,), ()>(&mut store, "fwd")?; + + let data = Arc::new(Mutex::new(Vec::new())); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + CollectConsumer { + data: data.clone(), + dropped: dropped.clone(), + }, + )?; + + let producer = StreamReader::new(&mut store, bytes::Bytes::from_static(b"abc"))?; + fwd.call_async(&mut store, (producer,)).await?; + + store + .run_concurrent(async |_| { + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await?; + + assert_eq!(*data.lock().unwrap(), b"abc"); + + Ok(()) +} + +/// Host producer forwarded by the guest into a stream whose readable end the +/// host holds but only starts consuming after the forward. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_producer_to_late_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FORWARDER).await?; + let mk = instance.get_typed_func::<(), (StreamReader,)>(&mut store, "mk")?; + let fwd = instance.get_typed_func::<(StreamReader,), ()>(&mut store, "fwd")?; + + let data = Arc::new(Mutex::new(Vec::new())); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + + let producer = StreamReader::new(&mut store, bytes::Bytes::from_static(b"abc"))?; + fwd.call_async(&mut store, (producer,)).await?; + + reader.pipe( + &mut store, + CollectConsumer { + data: data.clone(), + dropped: dropped.clone(), + }, + )?; + + store + .run_concurrent(async |_| { + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await?; + + assert_eq!(*data.lock().unwrap(), b"abc"); + + Ok(()) +} + +/// Host producer which stays live across the forward, forwarded by the guest +/// into a stream consumed by the host: data produced after the forward +/// reaches the consumer, and the end of the stream is observed separately +/// once the producer ends. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_live_host_producer_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FORWARDER).await?; + let mk = instance.get_typed_func::<(), (StreamReader,)>(&mut store, "mk")?; + let fwd = instance.get_typed_func::<(StreamReader,), ()>(&mut store, "fwd")?; + + let data = Arc::new(Mutex::new(Vec::new())); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + CollectConsumer { + data: data.clone(), + dropped: dropped.clone(), + }, + )?; + + let state = Arc::new(Mutex::new(ChunkState::default())); + ChunkState::push(&state, b"abcd"); + let producer = StreamReader::new( + &mut store, + ChunkProducer { + state: state.clone(), + dropped: Arc::new(AtomicBool::new(false)), + }, + )?; + fwd.call_async(&mut store, (producer,)).await?; + + store + .run_concurrent({ + let data = data.clone(); + let dropped = dropped.clone(); + let state = state.clone(); + async move |_| { + while data.lock().unwrap().len() < 4 { + tokio::task::yield_now().await; + } + ChunkState::push(&state, b"efgh"); + while data.lock().unwrap().len() < 8 { + tokio::task::yield_now().await; + } + assert!(!dropped.load(Ordering::SeqCst)); + ChunkState::end(&state); + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + } + }) + .await?; + + assert_eq!(*data.lock().unwrap(), b"abcdefgh"); + + Ok(()) +} + +/// Guest producer whose writable end was dropped before the guest forwards +/// into a stream consumed by the host: the consumer must observe the end of +/// the stream with no items delivered. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_dropped_guest_writer_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FORWARDER).await?; + let mk = instance.get_typed_func::<(), (StreamReader,)>(&mut store, "mk")?; + let mk_src = instance.get_typed_func::<(), ()>(&mut store, "mk-src")?; + let fwd_src = instance.get_typed_func::<(), ()>(&mut store, "fwd-src")?; + let drop_w = instance.get_typed_func::<(), ()>(&mut store, "drop-w")?; + + let data = Arc::new(Mutex::new(Vec::new())); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + CollectConsumer { + data: data.clone(), + dropped: dropped.clone(), + }, + )?; + + mk_src.call_async(&mut store, ()).await?; + drop_w.call_async(&mut store, ()).await?; + assert!(!dropped.load(Ordering::SeqCst)); + fwd_src.call_async(&mut store, ()).await?; + + assert!(dropped.load(Ordering::SeqCst)); + assert!(data.lock().unwrap().is_empty()); + + Ok(()) +} + +/// A `StreamConsumer` which collects the strings it receives and flags its +/// own destruction (which is how the host is notified that the stream ended). +struct CollectStringsConsumer { + data: Arc>>, + dropped: Arc, +} + +impl Drop for CollectStringsConsumer { + fn drop(&mut self) { + self.dropped.store(true, Ordering::SeqCst); + } +} + +impl StreamConsumer<()> for CollectStringsConsumer { + type Item = String; + + fn poll_consume( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + store: StoreContextMut<()>, + mut source: Source, + _finish: bool, + ) -> Poll> { + let mut buffer = Vec::with_capacity(8); + source.read(store, &mut buffer)?; + self.data.lock().unwrap().extend(buffer); + Poll::Ready(Ok(StreamResult::Completed)) + } +} + +/// Like `FORWARDER`, but for `stream`, whose payload is not "flat" +/// and is therefore copied via lift/lower (calling the reader's `realloc`) +/// rather than `memcpy`. +/// +/// Memory layout: the (ptr, len) pair read lands at address 0x10 with the +/// string bytes themselves landing at 0x200 (via `realloc`); the pair +/// written points at "hello" at address 0x100. `run` and `intra` block, so +/// they are lifted `async` (stackful) while everything else is lifted sync. +const STRING_FORWARDER: &str = r#" +(component + (core module $libc + (memory (export "m") 1) + (data (i32.const 0x100) "hello") + (func (export "realloc") (param i32 i32 i32 i32) (result i32) (i32.const 0x200)) + ) + (core instance $libc (instantiate $libc)) + + (type $s (stream string)) + (core func $stream.new (canon stream.new $s)) + (core func $stream.read (canon stream.read $s + (memory (core memory $libc "m")) + (realloc (core func $libc "realloc")))) + (core func $stream.write (canon stream.write $s async (memory (core memory $libc "m")))) + (core func $stream.forward (canon stream.forward $s)) + (core func $stream.cancel-write (canon stream.cancel-write $s async)) + (core func $stream.drop-writable (canon stream.drop-writable $s)) + + (core module $m + (import "" "m" (memory 1)) + (import "" "stream.new" (func $stream.new (result i64))) + (import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32))) + (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32))) + (import "" "stream.forward" (func $stream.forward (param i32 i32))) + (import "" "stream.cancel-write" (func $stream.cancel-write (param i32) (result i32))) + (import "" "stream.drop-writable" (func $stream.drop-writable (param i32))) + + ;; The writable end of the stream returned by `mk`. + (global $dst-w (mut i32) (i32.const 0)) + ;; The ends of the stream created by `mk-src`. + (global $src-r (mut i32) (i32.const 0)) + (global $src-w (mut i32) (i32.const 0)) + + ;; Create a new stream, saving its writable end and returning its + ;; readable end (which the host will consume). + (func (export "mk") (result i32) + (local $tmp i64) + (local.set $tmp (call $stream.new)) + (global.set $dst-w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + (i32.wrap_i64 (local.get $tmp)) + ) + + ;; Create a new stream, saving both ends (which this component will + ;; produce into). + (func (export "mk-src") + (local $tmp i64) + (local.set $tmp (call $stream.new)) + (global.set $src-r (i32.wrap_i64 (local.get $tmp))) + (global.set $src-w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + ) + + ;; Forward the stream created by `mk-src` into the stream created by `mk`. + (func (export "fwd-src") + (call $stream.forward (global.get $src-r) (global.get $dst-w)) + ) + + ;; Forward the given stream into the stream created by `mk`. + (func (export "fwd") (param $r i32) + (call $stream.forward (local.get $r) (global.get $dst-w)) + ) + + ;; Write "hello" to the stream created by `mk-src`, returning the packed + ;; result code. + (func (export "write") (result i32) + (i32.store (i32.const 0x10) (i32.const 0x100)) + (i32.store (i32.const 0x14) (i32.const 5)) + (call $stream.write (global.get $src-w) (i32.const 0x10) (i32.const 1)) + ) + + ;; Retrieve the completion of a pending write on the stream created by + ;; `mk-src` by cancelling it, returning the packed result code. + (func (export "check-write") (result i32) + (call $stream.cancel-write (global.get $src-w)) + ) + + ;; Drop the writable end of the stream created by `mk-src`. + (func (export "drop-w") + (call $stream.drop-writable (global.get $src-w)) + ) + + ;; Forward the given stream into a fresh stream, then read one string + ;; from the latter with a blocking read, checking that "hello" arrived. + (func (export "run") (param $r i32) (result i32) + (local $tmp i64) (local $r2 i32) (local $w2 i32) (local $code i32) + (local.set $tmp (call $stream.new)) + (local.set $r2 (i32.wrap_i64 (local.get $tmp))) + (local.set $w2 (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + + (call $stream.forward (local.get $r) (local.get $w2)) + + (local.set $code (call $stream.read (local.get $r2) (i32.const 0x10) (i32.const 1))) + (call $check-hello) + (local.get $code) + ) + + ;; Forward between two streams whose outer ends both belong to this + ;; instance, then attempt to send a string through the fused stream, + ;; which must trap: intra-component copies are restricted to numeric + ;; payloads. + (func (export "intra") + (local $tmp i64) (local $r2 i32) (local $w2 i32) + (call $mk-src-impl) + (local.set $tmp (call $stream.new)) + (local.set $r2 (i32.wrap_i64 (local.get $tmp))) + (local.set $w2 (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + + (call $stream.forward (global.get $src-r) (local.get $w2)) + + (i32.store (i32.const 0x10) (i32.const 0x100)) + (i32.store (i32.const 0x14) (i32.const 5)) + (call $stream.write (global.get $src-w) (i32.const 0x10) (i32.const 1)) + i32.const -1 ;; BLOCKED + i32.ne + if unreachable end + + ;; boom + (call $stream.read (local.get $r2) (i32.const 0x18) (i32.const 1)) + drop + ) + + (func $mk-src-impl + (local $tmp i64) + (local.set $tmp (call $stream.new)) + (global.set $src-r (i32.wrap_i64 (local.get $tmp))) + (global.set $src-w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + ) + + (func $check-hello + (local $ptr i32) + (i32.ne (i32.load (i32.const 0x14)) (i32.const 5)) + if unreachable end + (local.set $ptr (i32.load (i32.const 0x10))) + (i32.ne (i32.load8_u (local.get $ptr)) (i32.const 104)) + if unreachable end + (i32.ne (i32.load8_u (i32.add (local.get $ptr) (i32.const 1))) (i32.const 101)) + if unreachable end + (i32.ne (i32.load8_u (i32.add (local.get $ptr) (i32.const 2))) (i32.const 108)) + if unreachable end + (i32.ne (i32.load8_u (i32.add (local.get $ptr) (i32.const 3))) (i32.const 108)) + if unreachable end + (i32.ne (i32.load8_u (i32.add (local.get $ptr) (i32.const 4))) (i32.const 111)) + if unreachable end + ) + ) + + (core instance $i (instantiate $m + (with "" (instance + (export "m" (memory $libc "m")) + (export "stream.new" (func $stream.new)) + (export "stream.read" (func $stream.read)) + (export "stream.write" (func $stream.write)) + (export "stream.forward" (func $stream.forward)) + (export "stream.cancel-write" (func $stream.cancel-write)) + (export "stream.drop-writable" (func $stream.drop-writable)) + )) + )) + + (func (export "mk") (result (stream string)) (canon lift (core func $i "mk"))) + (func (export "mk-src") (canon lift (core func $i "mk-src"))) + (func (export "fwd-src") (canon lift (core func $i "fwd-src"))) + (func (export "fwd") (param "s" (stream string)) (canon lift (core func $i "fwd"))) + (func (export "write") (result u32) (canon lift (core func $i "write"))) + (func (export "check-write") (result u32) (canon lift (core func $i "check-write"))) + (func (export "drop-w") (canon lift (core func $i "drop-w"))) + (func (export "run") async (param "s" (stream string)) (result u32) + (canon lift (core func $i "run"))) + (func (export "intra") async (canon lift (core func $i "intra"))) +) +"#; + +/// Host string producer, forwarded by the guest into a stream the guest then +/// reads with a blocking read: the string must be copied via `realloc` into +/// the guest's memory through the forward. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_string_producer_to_guest() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, STRING_FORWARDER).await?; + let run = instance.get_typed_func::<(StreamReader,), (u32,)>(&mut store, "run")?; + + let reader = StreamReader::new(&mut store, vec!["hello".to_string()])?; + assert_eq!( + run.call_async(&mut store, (reader,)).await?, + (DROPPED_ONE_ITEM,) + ); + + Ok(()) +} + +/// Guest string producer with a write already pending when the guest +/// forwards into a stream consumed by the host: the string must be lifted +/// from the writer's memory through the forward. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_pending_guest_string_write_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, STRING_FORWARDER).await?; + let mk = instance.get_typed_func::<(), (StreamReader,)>(&mut store, "mk")?; + let mk_src = instance.get_typed_func::<(), ()>(&mut store, "mk-src")?; + let fwd_src = instance.get_typed_func::<(), ()>(&mut store, "fwd-src")?; + let write = instance.get_typed_func::<(), (u32,)>(&mut store, "write")?; + let check_write = instance.get_typed_func::<(), (u32,)>(&mut store, "check-write")?; + let drop_w = instance.get_typed_func::<(), ()>(&mut store, "drop-w")?; + + let data = Arc::new(Mutex::new(Vec::new())); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + CollectStringsConsumer { + data: data.clone(), + dropped: dropped.clone(), + }, + )?; + + mk_src.call_async(&mut store, ()).await?; + assert_eq!(write.call_async(&mut store, ()).await?, (BLOCKED,)); + fwd_src.call_async(&mut store, ()).await?; + + assert_eq!(*data.lock().unwrap(), ["hello"]); + assert_eq!( + check_write.call_async(&mut store, ()).await?, + (CANCELLED_ONE_ITEM,) + ); + + assert!(!dropped.load(Ordering::SeqCst)); + drop_w.call_async(&mut store, ()).await?; + store + .run_concurrent(async |_| { + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await?; + + Ok(()) +} + +/// Host string producer forwarded by the guest into a stream already being +/// consumed by the host: the strings must move host-to-host through the +/// forward without any guest memory involved. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_string_producer_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, STRING_FORWARDER).await?; + let mk = instance.get_typed_func::<(), (StreamReader,)>(&mut store, "mk")?; + let fwd = instance.get_typed_func::<(StreamReader,), ()>(&mut store, "fwd")?; + + let data = Arc::new(Mutex::new(Vec::new())); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + CollectStringsConsumer { + data: data.clone(), + dropped: dropped.clone(), + }, + )?; + + let producer = StreamReader::new(&mut store, vec!["hello".to_string(), "world".to_string()])?; + fwd.call_async(&mut store, (producer,)).await?; + + store + .run_concurrent(async |_| { + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await?; + + assert_eq!(*data.lock().unwrap(), ["hello", "world"]); + + Ok(()) +} + +/// A forward fusing two streams whose outer ends both belong to the same +/// instance does not lift the intra-component restriction: sending a +/// non-numeric payload through the fused stream still traps. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_intra_component_string() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, STRING_FORWARDER).await?; + let intra = instance.get_typed_func::<(), ()>(&mut store, "intra")?; + + let message = format!("{:?}", intra.call_async(&mut store, ()).await.unwrap_err()); + assert!( + message.contains("cannot read from and write to intra-component"), + "unexpected error: {message}" + ); + + Ok(()) +} + +/// A `FutureConsumer` which stores the value it receives and flags its own +/// destruction (which is how the host is notified when the writer end is +/// dropped without producing a value). +struct TakeConsumer { + value: Arc>>, + dropped: Arc, +} + +impl Drop for TakeConsumer { + fn drop(&mut self) { + self.dropped.store(true, Ordering::SeqCst); + } +} + +impl FutureConsumer<()> for TakeConsumer { + type Item = u8; + + fn poll_consume( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + store: StoreContextMut<()>, + mut source: Source, + _finish: bool, + ) -> Poll> { + let mut buffer = Vec::with_capacity(1); + source.read(store, &mut buffer)?; + *self.value.lock().unwrap() = Some(buffer[0]); + Poll::Ready(Ok(())) + } +} + +/// A guest component which can create futures, forward into them, and +/// read/write on either side of a forward. +/// +/// Memory layout: the value read lands at address 0; the value written comes +/// from address 16. The host writes 42; the guest writes 0xab. +const FUTURE_FORWARDER: &str = r#" +(component + (core module $libc (memory (export "m") 1)) + (core instance $libc (instantiate $libc)) + + (type $f (future u8)) + (core func $future.new (canon future.new $f)) + (core func $future.read (canon future.read $f async (memory (core memory $libc "m")))) + (core func $future.write (canon future.write $f async (memory (core memory $libc "m")))) + (core func $future.forward (canon future.forward $f)) + (core func $future.cancel-read (canon future.cancel-read $f async)) + (core func $future.cancel-write (canon future.cancel-write $f async)) + (core func $future.drop-writable (canon future.drop-writable $f)) + + (core module $m + (import "" "m" (memory 1)) + (import "" "future.new" (func $future.new (result i64))) + (import "" "future.read" (func $future.read (param i32 i32) (result i32))) + (import "" "future.write" (func $future.write (param i32 i32) (result i32))) + (import "" "future.forward" (func $future.forward (param i32 i32))) + (import "" "future.cancel-read" (func $future.cancel-read (param i32) (result i32))) + (import "" "future.cancel-write" (func $future.cancel-write (param i32) (result i32))) + (import "" "future.drop-writable" (func $future.drop-writable (param i32))) + + ;; The writable end of the future returned by `mk`. + (global $dst-w (mut i32) (i32.const 0)) + ;; The ends of the future created by `mk-src`. + (global $src-r (mut i32) (i32.const 0)) + (global $src-w (mut i32) (i32.const 0)) + + ;; Create a new future, saving its writable end and returning its + ;; readable end (which the host will consume). + (func (export "mk") (result i32) + (local $tmp i64) + (local.set $tmp (call $future.new)) + (global.set $dst-w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + (i32.wrap_i64 (local.get $tmp)) + ) + + ;; Create a new future, saving both ends (which this component will + ;; produce into). + (func (export "mk-src") + (local $tmp i64) + (local.set $tmp (call $future.new)) + (global.set $src-r (i32.wrap_i64 (local.get $tmp))) + (global.set $src-w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + ) + + ;; Forward the future created by `mk-src` into the future created by `mk`. + (func (export "fwd-src") + (call $future.forward (global.get $src-r) (global.get $dst-w)) + ) + + ;; Forward the given future into the future created by `mk`. + (func (export "fwd") (param $r i32) + (call $future.forward (local.get $r) (global.get $dst-w)) + ) + + ;; Write 0xab to the future created by `mk-src`, returning the result + ;; code. + (func (export "write") (result i32) + (i32.store8 (i32.const 16) (i32.const 0xab)) + (call $future.write (global.get $src-w) (i32.const 16)) + ) + + ;; Retrieve the completion of a pending write on the future created by + ;; `mk-src` by cancelling it, returning the result code. + (func (export "check-write") (result i32) + (call $future.cancel-write (global.get $src-w)) + ) + + ;; Drop the writable end of the future created by `mk-src` without + ;; writing a value. + (func (export "drop-w") + (call $future.drop-writable (global.get $src-w)) + ) + + ;; Forward the given future into a fresh future, then read from the + ;; latter, checking that 42 arrived. + (func (export "run") (param $r i32) (result i32) + (local $tmp i64) (local $r2 i32) (local $w2 i32) (local $code i32) + (local.set $tmp (call $future.new)) + (local.set $r2 (i32.wrap_i64 (local.get $tmp))) + (local.set $w2 (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + + (call $future.forward (local.get $r) (local.get $w2)) + + (local.set $code (call $future.read (local.get $r2) (i32.const 0))) + (call $check-42) + (local.get $code) + ) + + ;; Like `run`, except the read is issued (and blocks) before the forward, + ;; in which case its completion is delivered as an event, retrieved here + ;; via `future.cancel-read`. + (func (export "run-pending") (param $r i32) (result i32) + (local $tmp i64) (local $r2 i32) (local $w2 i32) (local $code i32) + (local.set $tmp (call $future.new)) + (local.set $r2 (i32.wrap_i64 (local.get $tmp))) + (local.set $w2 (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32)))) + + (call $future.read (local.get $r2) (i32.const 0)) + i32.const -1 ;; BLOCKED + i32.ne + if unreachable end + + (call $future.forward (local.get $r) (local.get $w2)) + + (local.set $code (call $future.cancel-read (local.get $r2))) + (call $check-42) + (local.get $code) + ) + + (func $check-42 + (i32.ne (i32.load8_u (i32.const 0)) (i32.const 42)) + if unreachable end + ) + ) + + (core instance $i (instantiate $m + (with "" (instance + (export "m" (memory $libc "m")) + (export "future.new" (func $future.new)) + (export "future.read" (func $future.read)) + (export "future.write" (func $future.write)) + (export "future.forward" (func $future.forward)) + (export "future.cancel-read" (func $future.cancel-read)) + (export "future.cancel-write" (func $future.cancel-write)) + (export "future.drop-writable" (func $future.drop-writable)) + )) + )) + + (func (export "mk") (result (future u8)) (canon lift (core func $i "mk"))) + (func (export "mk-src") (canon lift (core func $i "mk-src"))) + (func (export "fwd-src") (canon lift (core func $i "fwd-src"))) + (func (export "fwd") (param "f" (future u8)) (canon lift (core func $i "fwd"))) + (func (export "write") (result u32) (canon lift (core func $i "write"))) + (func (export "check-write") (result u32) (canon lift (core func $i "check-write"))) + (func (export "drop-w") (canon lift (core func $i "drop-w"))) + (func (export "run") (param "f" (future u8)) (result u32) (canon lift (core func $i "run"))) + (func (export "run-pending") (param "f" (future u8)) (result u32) + (canon lift (core func $i "run-pending"))) +) +"#; + +/// Host producer, forwarded by the guest into a future the guest then reads. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_future_producer_to_guest() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FUTURE_FORWARDER).await?; + let run = instance.get_typed_func::<(FutureReader,), (u32,)>(&mut store, "run")?; + + let reader = FutureReader::new(&mut store, async { Ok::<_, wasmtime::Error>(42u8) })?; + assert_eq!(run.call_async(&mut store, (reader,)).await?, (COMPLETED,)); + + Ok(()) +} + +/// Host producer, forwarded by the guest into a future the guest was already +/// blocked reading from. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_future_producer_to_pending_guest_read() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FUTURE_FORWARDER).await?; + let run = instance.get_typed_func::<(FutureReader,), (u32,)>(&mut store, "run-pending")?; + + let reader = FutureReader::new(&mut store, async { Ok::<_, wasmtime::Error>(42u8) })?; + assert_eq!(run.call_async(&mut store, (reader,)).await?, (COMPLETED,)); + + Ok(()) +} + +/// Guest producer, forwarded by the guest into a future consumed by the host. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_guest_future_producer_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FUTURE_FORWARDER).await?; + let mk = instance.get_typed_func::<(), (FutureReader,)>(&mut store, "mk")?; + let mk_src = instance.get_typed_func::<(), ()>(&mut store, "mk-src")?; + let fwd_src = instance.get_typed_func::<(), ()>(&mut store, "fwd-src")?; + let write = instance.get_typed_func::<(), (u32,)>(&mut store, "write")?; + let drop_w = instance.get_typed_func::<(), ()>(&mut store, "drop-w")?; + + let value = Arc::new(Mutex::new(None)); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + TakeConsumer { + value: value.clone(), + dropped: dropped.clone(), + }, + )?; + + mk_src.call_async(&mut store, ()).await?; + fwd_src.call_async(&mut store, ()).await?; + assert_eq!(write.call_async(&mut store, ()).await?, (COMPLETED,)); + + assert_eq!(*value.lock().unwrap(), Some(0xab)); + + drop_w.call_async(&mut store, ()).await?; + store + .run_concurrent(async |_| { + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await?; + + Ok(()) +} + +/// Guest producer with a write already pending when the guest forwards into a +/// future consumed by the host. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_pending_guest_future_write_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FUTURE_FORWARDER).await?; + let mk = instance.get_typed_func::<(), (FutureReader,)>(&mut store, "mk")?; + let mk_src = instance.get_typed_func::<(), ()>(&mut store, "mk-src")?; + let fwd_src = instance.get_typed_func::<(), ()>(&mut store, "fwd-src")?; + let write = instance.get_typed_func::<(), (u32,)>(&mut store, "write")?; + let check_write = instance.get_typed_func::<(), (u32,)>(&mut store, "check-write")?; + + let value = Arc::new(Mutex::new(None)); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + TakeConsumer { + value: value.clone(), + dropped: dropped.clone(), + }, + )?; + + mk_src.call_async(&mut store, ()).await?; + assert_eq!(write.call_async(&mut store, ()).await?, (BLOCKED,)); + fwd_src.call_async(&mut store, ()).await?; + + assert_eq!(*value.lock().unwrap(), Some(0xab)); + assert_eq!(check_write.call_async(&mut store, ()).await?, (COMPLETED,)); + + Ok(()) +} + +/// Host producer forwarded by the guest into a future already being consumed +/// by the host. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_future_producer_to_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FUTURE_FORWARDER).await?; + let mk = instance.get_typed_func::<(), (FutureReader,)>(&mut store, "mk")?; + let fwd = instance.get_typed_func::<(FutureReader,), ()>(&mut store, "fwd")?; + + let value = Arc::new(Mutex::new(None)); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + reader.pipe( + &mut store, + TakeConsumer { + value: value.clone(), + dropped: dropped.clone(), + }, + )?; + + let producer = FutureReader::new(&mut store, async { Ok::<_, wasmtime::Error>(42u8) })?; + fwd.call_async(&mut store, (producer,)).await?; + + store + .run_concurrent(async |_| { + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await?; + + assert_eq!(*value.lock().unwrap(), Some(42)); + + Ok(()) +} + +/// Host producer forwarded by the guest into a future whose readable end the +/// host holds but only starts consuming after the forward. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_host_future_producer_to_late_host_consumer() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FUTURE_FORWARDER).await?; + let mk = instance.get_typed_func::<(), (FutureReader,)>(&mut store, "mk")?; + let fwd = instance.get_typed_func::<(FutureReader,), ()>(&mut store, "fwd")?; + + let value = Arc::new(Mutex::new(None)); + let dropped = Arc::new(AtomicBool::new(false)); + + let (reader,) = mk.call_async(&mut store, ()).await?; + + let producer = FutureReader::new(&mut store, async { Ok::<_, wasmtime::Error>(42u8) })?; + fwd.call_async(&mut store, (producer,)).await?; + + reader.pipe( + &mut store, + TakeConsumer { + value: value.clone(), + dropped: dropped.clone(), + }, + )?; + + store + .run_concurrent(async |_| { + while !dropped.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await?; + + assert_eq!(*value.lock().unwrap(), Some(42)); + + Ok(()) +} + +/// Host reader closed without consuming after the guest forwards into it: the +/// drop must propagate back through the forward to the guest's write. +#[tokio::test] +#[cfg_attr(miri, ignore)] +async fn forward_guest_future_write_to_dropped_host_reader() -> Result<()> { + let engine = engine()?; + let mut store = Store::new(&engine, ()); + let instance = instantiate(&mut store, &engine, FUTURE_FORWARDER).await?; + let mk = instance.get_typed_func::<(), (FutureReader,)>(&mut store, "mk")?; + let mk_src = instance.get_typed_func::<(), ()>(&mut store, "mk-src")?; + let fwd_src = instance.get_typed_func::<(), ()>(&mut store, "fwd-src")?; + let write = instance.get_typed_func::<(), (u32,)>(&mut store, "write")?; + + let (mut reader,) = mk.call_async(&mut store, ()).await?; + mk_src.call_async(&mut store, ()).await?; + fwd_src.call_async(&mut store, ()).await?; + reader.close(&mut store)?; + + assert_eq!(write.call_async(&mut store, ()).await?, (DROPPED,)); + + Ok(()) +}