Enable wasmtime async support

This is needed to allow for some future features such as cooperative
multi-tasking via fuel or epoch interruption.

Signed-off-by: Lann Martin <lann.martin@fermyon.com>

Author: lann

Cargo.lock

@@ -8,7 +8,12 @@ authors = [ "Fermyon Engineering <engineering@fermyon.com>" ]
 anyhow = "1.0"
 serde = { version = "1.0", features = [ "derive" ] }
 thiserror = "1"
-wit-bindgen-wasmtime = { git = "https://github.com/bytecodealliance/wit-bindgen", rev = "cb871cfa1ee460b51eb1d144b175b9aab9c50aba" }
+tokio = { version = "1", features = [ "rt-multi-thread" ] }
+
+[dependencies.wit-bindgen-wasmtime]
+git = "https://github.com/bytecodealliance/wit-bindgen"
+rev = "cb871cfa1ee460b51eb1d144b175b9aab9c50aba"
+features = ["async"]
 
 [dev-dependencies]
 toml = "0.5"

crates/config/Cargo.toml

@@ -3,9 +3,10 @@ use std::sync::Arc;
 use crate::{Error, Key, Resolver, TreePath};
 
 mod wit {
-    wit_bindgen_wasmtime::export!("../../wit/ephemeral/spin-config.wit");
+    wit_bindgen_wasmtime::export!({paths: ["../../wit/ephemeral/spin-config.wit"], async: *});
 }
 pub use wit::spin_config::add_to_linker;
+use wit_bindgen_wasmtime::async_trait;
 
 /// A component configuration interface implementation.
 pub struct ComponentConfig {
@@ -26,11 +27,13 @@ impl ComponentConfig {
     }
 }
 
+#[async_trait]
 impl wit::spin_config::SpinConfig for ComponentConfig {
-    fn get_config(&mut self, key: &str) -> Result<String, wit::spin_config::Error> {
+    async fn get_config(&mut self, key: &str) -> Result<String, wit::spin_config::Error> {
         let key = Key::new(key)?;
         let path = &self.component_root + key;
-        Ok(self.resolver.resolve(&path)?)
+        // TODO(lann): Make resolve async
+        tokio::task::block_in_place(|| Ok(self.resolver.resolve(&path)?))
     }
 }
 

crates/config/src/host_component.rs

@@ -17,9 +17,11 @@ tracing = { version = "0.1", features = [ "log" ] }
 tracing-futures = "0.2"
 wasi-cap-std-sync = "0.39.1"
 wasi-common = "0.39.1"
-wasmtime = "0.39.1"
+wasmtime = { version = "0.39.1", features = [ "async" ] }
 wasmtime-wasi = "0.39.1"
 cap-std = "0.24.1"
 
-[dev-dependencies]
-wit-bindgen-wasmtime = { git = "https://github.com/bytecodealliance/wit-bindgen", rev = "cb871cfa1ee460b51eb1d144b175b9aab9c50aba" }
+[dev-dependencies.wit-bindgen-wasmtime]
+git = "https://github.com/bytecodealliance/wit-bindgen"
+rev = "cb871cfa1ee460b51eb1d144b175b9aab9c50aba"
+features = [ "async" ]

crates/engine/Cargo.toml

@@ -12,7 +12,7 @@ pub trait HostComponent: Send + Sync {
     type State: Any + Send;
 
     /// Add this component to the given Linker, using the given runtime state-getting handle.
-    fn add_to_linker<T>(
+    fn add_to_linker<T: Send>(
         linker: &mut Linker<RuntimeContext<T>>,
         state_handle: HostComponentsStateHandle<Self::State>,
     ) -> Result<()>;
@@ -30,7 +30,7 @@ pub(crate) struct HostComponents {
 }
 
 impl HostComponents {
-    pub(crate) fn insert<T: 'static, Component: HostComponent + 'static>(
+    pub(crate) fn insert<T: Send + 'static, Component: HostComponent + 'static>(
         &mut self,
         linker: &mut Linker<RuntimeContext<T>>,
         host_component: Component,

crates/engine/src/host_component.rs

@@ -59,7 +59,8 @@ pub struct Engine(wasmtime::Engine);
 
 impl Engine {
     /// Create a new engine and initialize it with the given config.
-    pub fn new(config: wasmtime::Config) -> Result<Self> {
+    pub fn new(mut config: wasmtime::Config) -> Result<Self> {
+        config.async_support(true);
         Ok(Self(wasmtime::Engine::new(&config)?))
     }
 
@@ -80,7 +81,7 @@ pub struct Builder<T: Default> {
     host_components: HostComponents,
 }
 
-impl<T: Default + 'static> Builder<T> {
+impl<T: Default + Send + 'static> Builder<T> {
     /// Creates a new instance of the execution builder.
     pub fn new(config: ExecutionContextConfiguration) -> Result<Builder<T>> {
         Self::with_engine(config, Engine::new(Default::default())?)
@@ -216,10 +217,10 @@ pub struct ExecutionContext<T: Default> {
     host_components: Arc<HostComponents>,
 }
 
-impl<T: Default> ExecutionContext<T> {
+impl<T: Default + Send> ExecutionContext<T> {
     /// Creates a store for a given component given its configuration and runtime data.
     #[instrument(skip(self, data, io))]
-    pub fn prepare_component(
+    pub async fn prepare_component(
         &self,
         component: &str,
         data: Option<T>,
@@ -234,7 +235,7 @@ impl<T: Default> ExecutionContext<T> {
         };
 
         let mut store = self.store(component, data, io, env, args)?;
-        let instance = component.pre.instantiate(&mut store)?;
+        let instance = component.pre.instantiate_async(&mut store).await?;
 
         Ok((store, instance))
     }

crates/engine/src/lib.rs

@@ -39,9 +39,13 @@ tracing-subscriber = { version = "0.3.7", features = ["env-filter"] }
 url = "2.2"
 wasi-cap-std-sync = "0.39.1"
 wasi-common = "0.39.1"
-wasmtime = "0.39.1"
+wasmtime = { version = "0.39.1", features = ["async"] }
 wasmtime-wasi = "0.39.1"
-wit-bindgen-wasmtime = { git = "https://github.com/bytecodealliance/wit-bindgen", rev = "cb871cfa1ee460b51eb1d144b175b9aab9c50aba" }
+
+[dependencies.wit-bindgen-wasmtime]
+git = "https://github.com/bytecodealliance/wit-bindgen"
+rev = "cb871cfa1ee460b51eb1d144b175b9aab9c50aba"
+features = ["async"]
 
 [dev-dependencies]
 criterion = { version = "0.3.5", features = ["async_tokio"] }

crates/http/Cargo.toml

@@ -33,7 +33,7 @@ use crate::{
     wagi::WagiHttpExecutor,
 };
 
-wit_bindgen_wasmtime::import!("../../wit/ephemeral/spin-http.wit");
+wit_bindgen_wasmtime::import!({paths: ["../../wit/ephemeral/spin-http.wit"], async: *});
 
 type ExecutionContext = spin_engine::ExecutionContext<SpinHttpData>;
 type RuntimeContext = spin_engine::RuntimeContext<SpinHttpData>;

crates/http/src/lib.rs

@@ -7,7 +7,6 @@ use async_trait::async_trait;
 use hyper::{Body, Request, Response};
 use spin_engine::io::ModuleIoRedirects;
 use std::{net::SocketAddr, str, str::FromStr};
-use tokio::task::spawn_blocking;
 use tracing::log;
 use wasmtime::{Instance, Store};
 
@@ -33,8 +32,9 @@ impl HttpExecutor for SpinHttpExecutor {
 
         let mior = ModuleIoRedirects::new(follow);
 
-        let (store, instance) =
-            engine.prepare_component(component, None, Some(mior.pipes), None, None)?;
+        let (store, instance) = engine
+            .prepare_component(component, None, Some(mior.pipes), None, None)
+            .await?;
 
         let resp_result = Self::execute_impl(store, instance, base, raw_route, req)
             .await
@@ -76,50 +76,47 @@ impl SpinHttpExecutor {
         let (parts, bytes) = req.into_parts();
         let bytes = hyper::body::to_bytes(bytes).await?.to_vec();
 
-        let res = spawn_blocking(move || -> Result<crate::spin_http::Response> {
-            let method = Self::method(&parts.method);
-
-            let headers: Vec<(&str, &str)> = headers
-                .iter()
-                .map(|(k, v)| (k.as_str(), v.as_str()))
-                .collect();
-
-            // Preparing to remove the params field. We are leaving it in place for now
-            // to avoid breaking the ABI, but no longer pass or accept values in it.
-            // https://github.com/fermyon/spin/issues/663
-            let params = vec![];
-
-            let body = Some(&bytes[..]);
-            let uri = match parts.uri.path_and_query() {
-                Some(u) => u.to_string(),
-                None => parts.uri.to_string(),
-            };
-
-            let req = crate::spin_http::Request {
-                method,
-                uri: &uri,
-                headers: &headers,
-                params: &params,
-                body,
-            };
-
-            Ok(engine.handle_http_request(&mut store, req)?)
-        })
-        .await??;
-
-        if res.status < 100 || res.status > 600 {
+        let method = Self::method(&parts.method);
+
+        let headers: Vec<(&str, &str)> = headers
+            .iter()
+            .map(|(k, v)| (k.as_str(), v.as_str()))
+            .collect();
+
+        // Preparing to remove the params field. We are leaving it in place for now
+        // to avoid breaking the ABI, but no longer pass or accept values in it.
+        // https://github.com/fermyon/spin/issues/663
+        let params = vec![];
+
+        let body = Some(&bytes[..]);
+        let uri = match parts.uri.path_and_query() {
+            Some(u) => u.to_string(),
+            None => parts.uri.to_string(),
+        };
+
+        let req = crate::spin_http::Request {
+            method,
+            uri: &uri,
+            headers: &headers,
+            params: &params,
+            body,
+        };
+
+        let resp = engine.handle_http_request(&mut store, req).await?;
+
+        if resp.status < 100 || resp.status > 600 {
             log::error!("malformed HTTP status code");
             return Ok(Response::builder()
                 .status(http::StatusCode::INTERNAL_SERVER_ERROR)
                 .body(Body::empty())?);
         };
 
-        let mut response = http::Response::builder().status(res.status);
+        let mut response = http::Response::builder().status(resp.status);
         if let Some(headers) = response.headers_mut() {
-            Self::append_headers(headers, res.headers)?;
+            Self::append_headers(headers, resp.headers)?;
         }
 
-        let body = match res.body {
+        let body = match resp.body {
             Some(b) => Body::from(b),
             None => Body::empty(),
         };

crates/http/src/spin.rs

@@ -12,7 +12,6 @@ use std::{
     net::SocketAddr,
     sync::{Arc, RwLock, RwLockReadGuard},
 };
-use tokio::task::spawn_blocking;
 use tracing::log;
 use wasi_common::pipe::{ReadPipe, WritePipe};
 
@@ -84,13 +83,15 @@ impl HttpExecutor for WagiHttpExecutor {
             headers.insert(keys[1].to_string(), val);
         }
 
-        let (mut store, instance) = engine.prepare_component(
-            component,
-            None,
-            Some(redirects),
-            Some(headers),
-            Some(argv.split(' ').map(|s| s.to_owned()).collect()),
-        )?;
+        let (mut store, instance) = engine
+            .prepare_component(
+                component,
+                None,
+                Some(redirects),
+                Some(headers),
+                Some(argv.split(' ').map(|s| s.to_owned()).collect()),
+            )
+            .await?;
 
         let start = instance
             .get_func(&mut store, &self.wagi_config.entrypoint)
@@ -102,7 +103,7 @@ impl HttpExecutor for WagiHttpExecutor {
                 )
             })?;
         tracing::trace!("Calling Wasm entry point");
-        let guest_result = spawn_blocking(move || start.call(&mut store, &[], &mut [])).await;
+        let guest_result = start.call_async(&mut store, &[], &mut []).await;
         tracing::info!("Module execution complete");
 
         let log_result = engine.save_output_to_logs(outputs.read(), component, false, true);
@@ -111,7 +112,7 @@ impl HttpExecutor for WagiHttpExecutor {
         // even if the guest code fails. (And when checking, check the guest
         // result first, so that guest failures are returned in preference to
         // log failures.)
-        guest_result?.or_else(ignore_successful_proc_exit_trap)?;
+        guest_result.or_else(ignore_successful_proc_exit_trap)?;
         log_result?;
 
         let stdout = outputs.stdout.read().unwrap();