Add support for async/streams/futures

This adds support for loading, compiling, linking, and running components which use the [Async ABI](https://github.com/WebAssembly/component-model/blob/main/design/mvp/Async.md) along with the [`stream`, `future`, and `error-context`](WebAssembly/component-model#405) types. It also adds support for generating host bindings such that multiple host functions can be run concurrently with guest tasks -- without monopolizing the `Store`. See the [implementation RFC](bytecodealliance/rfcs#38) for details, as well as [this repo](https://github.com/dicej/component-async-demo) containing end-to-end smoke tests. Signed-off-by: Joel Dice <[email protected]> fix clippy warnings and bench/fuzzing errors Signed-off-by: Joel Dice <[email protected]> revert atomic.wit whitespace change Signed-off-by: Joel Dice <[email protected]> fix build when component-model disabled Signed-off-by: Joel Dice <[email protected]> bless component-macro expected output Signed-off-by: Joel Dice <[email protected]> fix no-std build error Signed-off-by: Joel Dice <[email protected]> fix build with --no-default-features --features runtime,component-model Signed-off-by: Joel Dice <[email protected]> partly fix no-std build It's still broken due to the use of `std::collections::HashMap` in crates/wasmtime/src/runtime/vm/component.rs. I'll address that as part of the work to avoid exposing global task/future/stream/error-context handles to guests. Signed-off-by: Joel Dice <[email protected]> maintain per-instance tables for futures, streams, and error-contexts Signed-off-by: Joel Dice <[email protected]> refactor task/stream/future handle lifting/lowering This addresses a couple of issues: - Previously, we were passing task/stream/future/error-context reps directly to instances while keeping track of which instance had access to which rep. That worked fine in that there was no way to forge access to inaccessible reps, but it leaked information about what other instances were doing. Now we maintain per-instance waitable and error-context tables which map the reps to and from the handles which the instance sees. - The `no_std` build was broken due to use of `HashMap` in `runtime::vm::component`, which is now fixed. Note that we use one single table per instance for all tasks, streams, and futures. This is partly necessary because, when async events are delivered to the guest, it wouldn't have enough context to know which stream or future we're talking about if each unique stream and future type had its own table. So at minimum, we need to use the same table for all streams (regardless of payload type), and likewise for futures. Also, per WebAssembly/component-model#395 (comment), the plan is to move towards a shared table for all resource types as well, so this moves us in that direction. Signed-off-by: Joel Dice <[email protected]> fix wave breakage due to new stream/future/error-context types Signed-off-by: Joel Dice <[email protected]> switch wasm-tools to v1.220.0-based branch Signed-off-by: Joel Dice <[email protected]> check `task.return` type at runtime We can't statically verify a given call to `task.return` corresponds to the expected core signature appropriate for the currently running task, so we must do so at runtime. In order to make that check efficient, we intern the types. My initial plan was to use `ModuleInternedTypeIndex` and/or `VMSharedTypeIndex` for interning, but that got hairy with WasmGC considerations, so instead I added new fields to `ComponentTypes` and `ComponentTypesBuilder`. Signed-off-by: Joel Dice <[email protected]> add `TypedFunc::call_concurrent` and refine stream/future APIs This implements what I proposed in https://github.com/dicej/rfcs/blob/component-async/accepted/component-model-async.md#wasmtime. Specifically, it adds: - A new `Promise` type, useful for working with concurrent operations that require access to a `Store` to make progress. - A new `PromisesUnordered` type for `await`ing multiple promises concurrently -`TypedFunc::call_concurrent` (which returns a `Promise`), allowing multiple host->guest calls to run concurrently on the same instance. - Updated `{Stream|Future}{Writer|Reader}` APIs which use `Promise` The upshot is that the embedder can now ergonomically manage arbitrary numbers of concurrent operations. Previously, this was a lot more difficult to do without accidentally starving some of the operations due to another one monopolizing the `Store`. Finally, this includes various refactorings and fixes for bugs exposed by the newer, more versatile APIs. Signed-off-by: Joel Dice <[email protected]> clean up verbosity in component/func.rs Signed-off-by: Joel Dice <[email protected]> snapshot Signed-off-by: Joel Dice <[email protected]> implement stream/future read/write cancellation This required a somewhat viral addition of `Send` and `Sync` bounds for async host function closure types, unfortunately. Signed-off-by: Joel Dice <[email protected]> add `Func::call_concurrent` and `LinkerInstance::func_new_concurrent` Signed-off-by: Joel Dice <[email protected]> dynamic API support for streams/futures/error-contexts Signed-off-by: Joel Dice <[email protected]> support callback-less (AKA stackful) async lifts Signed-off-by: Joel Dice <[email protected]>
dicej · Dec 11, 2024 · 22cb051 · 22cb051
1 parent 23e8dce
commit 22cb051
Show file tree

Hide file tree

Showing 164 changed files with 13,587 additions and 2,007 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -277,16 +277,16 @@ wit-bindgen = { version = "0.35.0", default-features = false }
 wit-bindgen-rust-macro = { version = "0.35.0", default-features = false }
 
 # wasm-tools family:
-wasmparser = { version = "0.221.2", default-features = false, features = ['simd'] }
-wat = "1.221.2"
-wast = "221.0.2"
-wasmprinter = "0.221.2"
-wasm-encoder = "0.221.2"
-wasm-smith = "0.221.2"
-wasm-mutate = "0.221.2"
-wit-parser = "0.221.2"
-wit-component = "0.221.2"
-wasm-wave = "0.221.2"
+wasmparser = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2", default-features = false, features = ['simd'] }
+wat = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2" }
+wast = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2" }
+wasmprinter = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2" }
+wasm-encoder = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2" }
+wasm-smith = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2" }
+wasm-mutate = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2" }
+wit-parser = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2" }
+wit-component = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2" }
+wasm-wave = { git = "https://github.com/dicej/wasm-tools", branch = "async-v1.221.2" }
 
 # Non-Bytecode Alliance maintained dependencies:
 # --------------------------

diff --git a/benches/call.rs b/benches/call.rs
@@ -628,7 +628,8 @@ mod component {
             + PartialEq
             + Debug
             + Send
-            + Sync,
+            + Sync
+            + 'static,
     {
         // Benchmark the "typed" version.
         c.bench_function(&format!("component - host-to-wasm - typed - {name}"), |b| {

diff --git a/crates/component-macro/Cargo.toml b/crates/component-macro/Cargo.toml
@@ -41,3 +41,4 @@ similar = { workspace = true }
 [features]
 async = []
 std = ['wasmtime-wit-bindgen/std']
+component-model-async = ['std', 'async', 'wasmtime-wit-bindgen/component-model-async']
diff --git a/crates/component-macro/src/bindgen.rs b/crates/component-macro/src/bindgen.rs
@@ -1,14 +1,15 @@
 use proc_macro2::{Span, TokenStream};
 use quote::ToTokens;
-use std::collections::HashMap;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::env;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
 use syn::parse::{Error, Parse, ParseStream, Result};
 use syn::punctuated::Punctuated;
 use syn::{braced, token, Token};
-use wasmtime_wit_bindgen::{AsyncConfig, Opts, Ownership, TrappableError, TrappableImports};
+use wasmtime_wit_bindgen::{
+    AsyncConfig, CallStyle, Opts, Ownership, TrappableError, TrappableImports,
+};
 use wit_parser::{PackageId, Resolve, UnresolvedPackageGroup, WorldId};
 
 pub struct Config {
@@ -20,13 +21,22 @@ pub struct Config {
 }
 
 pub fn expand(input: &Config) -> Result<TokenStream> {
-    if !cfg!(feature = "async") && input.opts.async_.maybe_async() {
+    if let (CallStyle::Async | CallStyle::Concurrent, false) =
+        (input.opts.call_style(), cfg!(feature = "async"))
+    {
         return Err(Error::new(
             Span::call_site(),
             "cannot enable async bindings unless `async` crate feature is active",
         ));
     }
 
+    if input.opts.concurrent_imports && !cfg!(feature = "component-model-async") {
+        return Err(Error::new(
+            Span::call_site(),
+            "cannot enable `concurrent_imports` option unless `component-model-async` crate feature is active",
+        ));
+    }
+
     let mut src = match input.opts.generate(&input.resolve, input.world) {
         Ok(s) => s,
         Err(e) => return Err(Error::new(Span::call_site(), e.to_string())),
@@ -40,7 +50,10 @@ pub fn expand(input: &Config) -> Result<TokenStream> {
     // place a formatted version of the expanded code into a file. This file
     // will then show up in rustc error messages for any codegen issues and can
     // be inspected manually.
-    if input.include_generated_code_from_file || std::env::var("WASMTIME_DEBUG_BINDGEN").is_ok() {
+    if input.include_generated_code_from_file
+        || input.opts.debug
+        || std::env::var("WASMTIME_DEBUG_BINDGEN").is_ok()
+    {
         static INVOCATION: AtomicUsize = AtomicUsize::new(0);
         let root = Path::new(env!("DEBUG_OUTPUT_DIR"));
         let world_name = &input.resolve.worlds[input.world].name;
@@ -107,13 +120,16 @@ impl Parse for Config {
                     }
                     Opt::Tracing(val) => opts.tracing = val,
                     Opt::VerboseTracing(val) => opts.verbose_tracing = val,
+                    Opt::Debug(val) => opts.debug = val,
                     Opt::Async(val, span) => {
                         if async_configured {
                             return Err(Error::new(span, "cannot specify second async config"));
                         }
                         async_configured = true;
                         opts.async_ = val;
                     }
+                    Opt::ConcurrentImports(val) => opts.concurrent_imports = val,
+                    Opt::ConcurrentExports(val) => opts.concurrent_exports = val,
                     Opt::TrappableErrorType(val) => opts.trappable_error_type = val,
                     Opt::TrappableImports(val) => opts.trappable_imports = val,
                     Opt::Ownership(val) => opts.ownership = val,
@@ -138,7 +154,7 @@ impl Parse for Config {
                                 "cannot specify a world with `interfaces`",
                             ));
                         }
-                        world = Some("interfaces".to_string());
+                        world = Some("wasmtime:component-macro-synthesized/interfaces".to_string());
 
                         opts.only_interfaces = true;
                     }
@@ -281,6 +297,9 @@ mod kw {
     syn::custom_keyword!(require_store_data_send);
     syn::custom_keyword!(wasmtime_crate);
     syn::custom_keyword!(include_generated_code_from_file);
+    syn::custom_keyword!(concurrent_imports);
+    syn::custom_keyword!(concurrent_exports);
+    syn::custom_keyword!(debug);
 }
 
 enum Opt {
@@ -301,12 +320,19 @@ enum Opt {
     RequireStoreDataSend(bool),
     WasmtimeCrate(syn::Path),
     IncludeGeneratedCodeFromFile(bool),
+    ConcurrentImports(bool),
+    ConcurrentExports(bool),
+    Debug(bool),
 }
 
 impl Parse for Opt {
     fn parse(input: ParseStream<'_>) -> Result<Self> {
         let l = input.lookahead1();
-        if l.peek(kw::path) {
+        if l.peek(kw::debug) {
+            input.parse::<kw::debug>()?;
+            input.parse::<Token![:]>()?;
+            Ok(Opt::Debug(input.parse::<syn::LitBool>()?.value))
+        } else if l.peek(kw::path) {
             input.parse::<kw::path>()?;
             input.parse::<Token![:]>()?;
 
@@ -380,6 +406,14 @@ impl Parse for Opt {
                     span,
                 ))
             }
+        } else if l.peek(kw::concurrent_imports) {
+            input.parse::<kw::concurrent_imports>()?;
+            input.parse::<Token![:]>()?;
+            Ok(Opt::ConcurrentImports(input.parse::<syn::LitBool>()?.value))
+        } else if l.peek(kw::concurrent_exports) {
+            input.parse::<kw::concurrent_exports>()?;
+            input.parse::<Token![:]>()?;
+            Ok(Opt::ConcurrentExports(input.parse::<syn::LitBool>()?.value))
         } else if l.peek(kw::ownership) {
             input.parse::<kw::ownership>()?;
             input.parse::<Token![:]>()?;

diff --git a/crates/component-macro/tests/expanded/char.rs b/crates/component-macro/tests/expanded/char.rs
@@ -18,7 +18,7 @@ impl<T> Clone for TheWorldPre<T> {
         }
     }
 }
-impl<_T> TheWorldPre<_T> {
+impl<_T: 'static> TheWorldPre<_T> {
     /// Creates a new copy of `TheWorldPre` bindings which can then
     /// be used to instantiate into a particular store.
     ///
@@ -152,7 +152,10 @@ const _: () = {
             mut store: impl wasmtime::AsContextMut<Data = _T>,
             component: &wasmtime::component::Component,
             linker: &wasmtime::component::Linker<_T>,
-        ) -> wasmtime::Result<TheWorld> {
+        ) -> wasmtime::Result<TheWorld>
+        where
+            _T: 'static,
+        {
             let pre = linker.instantiate_pre(component)?;
             TheWorldPre::new(pre)?.instantiate(store)
         }
@@ -194,19 +197,23 @@ pub mod foo {
             }
             pub trait GetHost<
                 T,
-            >: Fn(T) -> <Self as GetHost<T>>::Host + Send + Sync + Copy + 'static {
+                D,
+            >: Fn(T) -> <Self as GetHost<T, D>>::Host + Send + Sync + Copy + 'static {
                 type Host: Host;
             }
-            impl<F, T, O> GetHost<T> for F
+            impl<F, T, D, O> GetHost<T, D> for F
             where
                 F: Fn(T) -> O + Send + Sync + Copy + 'static,
                 O: Host,
             {
                 type Host = O;
             }
-            pub fn add_to_linker_get_host<T>(
+            pub fn add_to_linker_get_host<
+                T,
+                G: for<'a> GetHost<&'a mut T, T, Host: Host>,
+            >(
                 linker: &mut wasmtime::component::Linker<T>,
-                host_getter: impl for<'a> GetHost<&'a mut T>,
+                host_getter: G,
             ) -> wasmtime::Result<()> {
                 let mut inst = linker.instance("foo:foo/chars")?;
                 inst.func_wrap(
@@ -354,7 +361,10 @@ pub mod exports {
                         &self,
                         mut store: S,
                         arg0: char,
-                    ) -> wasmtime::Result<()> {
+                    ) -> wasmtime::Result<()>
+                    where
+                        <S as wasmtime::AsContext>::Data: Send + 'static,
+                    {
                         let callee = unsafe {
                             wasmtime::component::TypedFunc::<
                                 (char,),
@@ -369,7 +379,10 @@ pub mod exports {
                     pub fn call_return_char<S: wasmtime::AsContextMut>(
                         &self,
                         mut store: S,
-                    ) -> wasmtime::Result<char> {
+                    ) -> wasmtime::Result<char>
+                    where
+                        <S as wasmtime::AsContext>::Data: Send + 'static,
+                    {
                         let callee = unsafe {
                             wasmtime::component::TypedFunc::<
                                 (),