diff --git a/Cargo.toml b/Cargo.toml index 0c2f571..da4c63f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,6 +30,10 @@ rust-version.workspace=true name="wasm-rust" test=false +[[example]] +name="host_coro" +required-features=["async"] + [dev-dependencies] wat={workspace=true} eyre={workspace=true} diff --git a/crates/tinywasm/Cargo.toml b/crates/tinywasm/Cargo.toml index 5dc0447..d9d34ea 100644 --- a/crates/tinywasm/Cargo.toml +++ b/crates/tinywasm/Cargo.toml @@ -37,6 +37,8 @@ logging=["log", "tinywasm-parser?/logging", "tinywasm-types/logging"] std=["tinywasm-parser?/std", "tinywasm-types/std"] parser=["dep:tinywasm-parser"] archive=["tinywasm-types/archive"] +async=[] +test_async=["async"] #feels weird putting it here [[test]] name="test-wasm-1" diff --git a/crates/tinywasm/src/coro.rs b/crates/tinywasm/src/coro.rs new file mode 100644 index 0000000..64313b8 --- /dev/null +++ b/crates/tinywasm/src/coro.rs @@ -0,0 +1,260 @@ +#![cfg_attr(not(feature = "async"), allow(unused))] +#![cfg_attr(not(feature = "async"), allow(unreachable_pub))] + +mod module { + use crate::Result; + use core::fmt::Debug; + pub(crate) use tinywasm_types::{ResumeArgument, YieldedValue}; + + ///"coroutine statse", "coroutine instance", "resumable". Stores info to continue a function that was paused + pub trait CoroState: Debug { + #[cfg(feature = "async")] + /// resumes the execution of the coroutine + fn resume(&mut self, ctx: ResumeContext, arg: ResumeArgument) -> Result>; + } + + /// explains why did execution suspend, and carries payload if needed + #[derive(Debug)] + #[non_exhaustive] // some variants are feature-gated + #[cfg(feature = "async")] + pub enum SuspendReason { + /// host function yielded + /// some host functions might expect resume argument when calling resume + Yield(YieldedValue), + + /// time to suspend has come, + /// host shouldn't provide resume argument when calling resume + #[cfg(feature = "std")] + SuspendedEpoch, + + /// user's should-suspend-callback returned Break, + /// host shouldn't provide resume argument when calling resume + SuspendedCallback, + + /// async should_suspend flag was set + /// host shouldn't provide resume argument when calling resume + SuspendedFlag, + // possible others: delimited continuations proposal, debugger breakpoint, out of fuel + } + + #[cfg(not(feature = "async"))] + pub type SuspendReason = core::convert::Infallible; + + /// result of a function that might pause in the middle and yield + /// to be resumed later + #[derive(Debug)] + pub enum PotentialCoroCallResult +//where for + // State: CoroState, // can't in stable rust + { + /// function returns normally + Return(R), + /// interpreter will be suspended and execution will return to host along with SuspendReason + Suspended(SuspendReason, State), + } + + /// result of resuming coroutine state. Unlike [`PotentialCoroCallResult`] + /// doesn't need to have state, since it's contained in self + #[derive(Debug)] + pub enum CoroStateResumeResult { + /// CoroState has finished + /// after this CoroState::resume can't be called again on that CoroState + Return(R), + + /// host function yielded + /// execution returns to host along with yielded value + Suspended(SuspendReason), + } + + impl PotentialCoroCallResult { + /// in case you expect function only to return + /// you can make Suspend into [crate::Error::UnexpectedSuspend] error + pub fn suspend_to_err(self) -> Result { + match self { + PotentialCoroCallResult::Return(r) => Ok(r), + #[cfg(feature = "async")] + PotentialCoroCallResult::Suspended(r, _) => Err(crate::Error::UnexpectedSuspend(r.into())), + } + } + + /// true if coro is finished + pub fn finished(&self) -> bool { + matches!(self, Self::Return(_)) + } + /// separates state from PotentialCoroCallResult, leaving CoroStateResumeResult (one without state) + pub fn split_state(self) -> (CoroStateResumeResult, Option) { + match self { + Self::Return(val) => (CoroStateResumeResult::Return(val), None), + Self::Suspended(suspend, state) => (CoroStateResumeResult::Suspended(suspend), Some(state)), + } + } + /// separates result from PotentialCoroCallResult, leaving unit type in it's place + pub fn split_result(self) -> (PotentialCoroCallResult<(), State>, Option) { + match self { + Self::Return(result) => (PotentialCoroCallResult::Return(()), Some(result)), + Self::Suspended(suspend, state) => (PotentialCoroCallResult::Suspended(suspend, state), None), + } + } + + /// transforms state + pub fn map_state(self, mapper: impl FnOnce(State) -> OutS) -> PotentialCoroCallResult { + match self { + Self::Return(val) => PotentialCoroCallResult::Return(val), + Self::Suspended(suspend, state) => PotentialCoroCallResult::Suspended(suspend, mapper(state)), + } + } + /// transform result with mapper if there is none - calls "otherwise". + /// user_val passed to whichever is called and is guaranteed to be used + pub fn map( + self, + user_val: Usr, + res_mapper: impl FnOnce(R, Usr) -> OutR, + state_mapper: impl FnOnce(State, Usr) -> OutS, + ) -> PotentialCoroCallResult { + match self { + Self::Return(res) => PotentialCoroCallResult::Return(res_mapper(res, user_val)), + Self::Suspended(suspend, state) => { + PotentialCoroCallResult::Suspended(suspend, state_mapper(state, user_val)) + } + } + } + /// transforms result + pub fn map_result(self, mapper: impl FnOnce(R) -> OutR) -> PotentialCoroCallResult { + self.map((), |val, _| mapper(val), |s, _| s) + } + } + + impl PotentialCoroCallResult, State> { + /// turns Self, S> into Resulf, S> + pub fn propagate_err_result(self) -> core::result::Result, E> { + Ok(match self { + PotentialCoroCallResult::Return(res) => PotentialCoroCallResult::::Return(res?), + PotentialCoroCallResult::Suspended(why, state) => { + PotentialCoroCallResult::::Suspended(why, state) + } + }) + } + } + impl PotentialCoroCallResult> { + /// turns Self> into Resulf> + pub fn propagate_err_state(self) -> core::result::Result, E> { + Ok(match self { + PotentialCoroCallResult::Return(res) => PotentialCoroCallResult::::Return(res), + PotentialCoroCallResult::Suspended(why, state) => { + PotentialCoroCallResult::::Suspended(why, state?) + } + }) + } + } + + impl CoroStateResumeResult { + /// in case you expect function only to return + /// you can make Suspend into [crate::Error::UnexpectedSuspend] error + pub fn suspend_to_err(self) -> Result { + match self { + Self::Return(r) => Ok(r), + #[cfg(feature = "async")] + Self::Suspended(r) => Err(crate::Error::UnexpectedSuspend(r.into())), + } + } + + /// true if coro is finished + pub fn finished(&self) -> bool { + matches!(self, Self::Return(_)) + } + /// separates result from CoroStateResumeResult, leaving unit type in it's place + pub fn split_result(self) -> (CoroStateResumeResult<()>, Option) { + let (a, r) = PotentialCoroCallResult::::from(self).split_result(); + (a.into(), r) + } + /// transforms result + pub fn map_result(self, mapper: impl FnOnce(R) -> OutR) -> CoroStateResumeResult { + PotentialCoroCallResult::::from(self).map_result(mapper).into() + } + /// transform result with mapper. If there is none - calls "otherwise" + /// user_val passed to whichever is called and is guaranteed to be used + pub fn map( + self, + user_val: Usr, + mapper: impl FnOnce(R, Usr) -> OutR, + otherwise: impl FnOnce(Usr), + ) -> CoroStateResumeResult { + PotentialCoroCallResult::::from(self).map(user_val, mapper, |(), usr| otherwise(usr)).into() + } + } + + impl CoroStateResumeResult> { + /// turns Self> into Resulf> + pub fn propagate_err(self) -> core::result::Result, E> { + Ok(PotentialCoroCallResult::, ()>::from(self).propagate_err_result()?.into()) + } + } + + // convert between PotentialCoroCallResult and CoroStateResumeResult + impl From> for CoroStateResumeResult + where + DstR: From, + { + fn from(value: PotentialCoroCallResult) -> Self { + match value { + PotentialCoroCallResult::Return(val) => Self::Return(val.into()), + PotentialCoroCallResult::Suspended(suspend, ()) => Self::Suspended(suspend), + } + } + } + impl From> for PotentialCoroCallResult { + fn from(value: CoroStateResumeResult) -> Self { + match value { + CoroStateResumeResult::Return(val) => PotentialCoroCallResult::Return(val), + CoroStateResumeResult::Suspended(suspend) => PotentialCoroCallResult::Suspended(suspend, ()), + } + } + } + + #[cfg(feature = "async")] + impl SuspendReason { + /// shotrhand to package val into a Box in a [SuspendReason::Yield] variant + /// you'll need to specify type explicitly, because you'll need to use exact same type when downcasting + pub fn make_yield(val: impl Into + core::any::Any) -> Self { + Self::Yield(Some(alloc::boxed::Box::new(val) as alloc::boxed::Box)) + } + } + + /// for use in error [`crate::Error::UnexpectedSuspend`] + /// same as [SuspendReason], but without [tinywasm_types::YieldedValue], since we can't debug-print it + /// and including it would either require YieldedValue to be Send+Sync or disable that for Error + #[derive(Debug)] + pub enum UnexpectedSuspendError { + /// host function yielded + Yield, + + /// timeout, + #[cfg(feature = "std")] + SuspendedEpoch, + + /// user's should-suspend-callback returned Break, + SuspendedCallback, + + /// async should_suspend flag was set + SuspendedFlag, + } + + #[cfg(feature = "async")] + impl From for UnexpectedSuspendError { + fn from(value: SuspendReason) -> Self { + match value { + SuspendReason::Yield(_) => Self::Yield, + #[cfg(feature = "std")] + SuspendReason::SuspendedEpoch => Self::SuspendedEpoch, + SuspendReason::SuspendedCallback => Self::SuspendedCallback, + SuspendReason::SuspendedFlag => Self::SuspendedFlag, + } + } + } +} + +#[cfg(feature = "async")] +pub use module::*; + +#[cfg(not(feature = "async"))] +pub(crate) use module::*; diff --git a/crates/tinywasm/src/error.rs b/crates/tinywasm/src/error.rs index e0510b2..0750412 100644 --- a/crates/tinywasm/src/error.rs +++ b/crates/tinywasm/src/error.rs @@ -6,6 +6,11 @@ use tinywasm_types::FuncType; #[cfg(feature = "parser")] pub use tinywasm_parser::ParseError; +#[cfg(feature = "async")] +use crate::coro::UnexpectedSuspendError; + +use crate::interpreter; + /// Errors that can occur for `TinyWasm` operations #[derive(Debug)] pub enum Error { @@ -35,6 +40,17 @@ pub enum Error { /// The store is not the one that the module instance was instantiated in InvalidStore, + /// ResumeArgument of wrong type was provided + InvalidResumeArgument, + + /// Tried to resume on runtime when it's not suspended + InvalidResume, + + /// Function unexpectedly yielded instead of returning + /// (for backwards compatibility with old api) + #[cfg(feature = "async")] + UnexpectedSuspend(UnexpectedSuspendError), + #[cfg(feature = "std")] /// An I/O error occurred Io(crate::std::io::Error), @@ -184,6 +200,9 @@ impl Display for Error { #[cfg(feature = "std")] Self::Io(err) => write!(f, "I/O error: {err}"), + #[cfg(feature = "async")] + Self::UnexpectedSuspend(_) => write!(f, "funtion yielded instead of returning"), + Self::Trap(trap) => write!(f, "trap: {trap}"), Self::Linker(err) => write!(f, "linking error: {err}"), Self::InvalidLabelType => write!(f, "invalid label type"), @@ -193,6 +212,8 @@ impl Display for Error { write!(f, "invalid host function return: expected={expected:?}, actual={actual:?}") } Self::InvalidStore => write!(f, "invalid store"), + Self::InvalidResumeArgument => write!(f, "invalid resume argument supplied to suspended function"), + Self::InvalidResume => write!(f, "attempt to resume coroutine that has already finished"), } } } @@ -246,14 +267,14 @@ impl From for Error { pub type Result = crate::std::result::Result; pub(crate) trait Controlify { - fn to_cf(self) -> ControlFlow, T>; + fn to_cf(self) -> ControlFlow; } impl Controlify for Result { - fn to_cf(self) -> ControlFlow, T> { + fn to_cf(self) -> ControlFlow { match self { Ok(value) => ControlFlow::Continue(value), - Err(err) => ControlFlow::Break(Some(err)), + Err(err) => ControlFlow::Break(interpreter::executor::ReasonToBreak::Errored(err)), } } } diff --git a/crates/tinywasm/src/func.rs b/crates/tinywasm/src/func.rs index f5a9873..930ff4a 100644 --- a/crates/tinywasm/src/func.rs +++ b/crates/tinywasm/src/func.rs @@ -1,3 +1,8 @@ +#[cfg(feature = "async")] +use {crate::coro::CoroState, tinywasm_types::ResumeArgument}; + +use crate::interpreter; +use crate::interpreter::executor::SuspendedHostCoroState; use crate::interpreter::stack::{CallFrame, Stack}; use crate::{log, unlikely, Function}; use crate::{Error, FuncContext, Result, Store}; @@ -19,8 +24,15 @@ impl FuncHandle { /// Call a function (Invocation) /// /// See + /// #[inline] pub fn call(&self, store: &mut Store, params: &[WasmValue]) -> Result> { + self.call_coro(store, params)?.suspend_to_err() + } + + /// Call a function (Invocation) and anticipate possible yield instead as well as return + #[inline] + pub fn call_coro(&self, store: &mut Store, params: &[WasmValue]) -> Result { // Comments are ordered by the steps in the spec // In this implementation, some steps are combined and ordered differently for performance reasons @@ -53,7 +65,14 @@ impl FuncHandle { Function::Host(host_func) => { let host_func = host_func.clone(); let ctx = FuncContext { store, module_addr: self.module_addr }; - return host_func.call(ctx, params); + return Ok(host_func.call(ctx, params)?.map_state(|state| SuspendedFunc { + func: SuspendedFuncInner::Host(SuspendedHostCoroState { + coro_state: state, + coro_orig_function: self.addr, + }), + module_addr: self.module_addr, + store_id: store.id(), + })); } Function::Wasm(wasm_func) => wasm_func, }; @@ -63,23 +82,33 @@ impl FuncHandle { // 7. Push the frame f to the call stack // & 8. Push the values to the stack (Not needed since the call frame owns the values) - let mut stack = Stack::new(call_frame); + let stack = Stack::new(call_frame); // 9. Invoke the function instance let runtime = store.runtime(); - runtime.exec(store, &mut stack)?; - - // Once the function returns: - // let result_m = func_ty.results.len(); - - // 1. Assert: m values are on the top of the stack (Ensured by validation) - // assert!(stack.values.len() >= result_m); - - // 2. Pop m values from the stack - let res = stack.values.pop_results(&func_ty.results); - - // The values are returned as the results of the invocation. - Ok(res) + let exec_outcome = runtime.exec(store, stack)?; + Ok(exec_outcome + .map_result(|mut stack| -> Vec { + // Once the function returns: + // let result_m = func_ty.results.len(); + + // 1. Assert: m values are on the top of the stack (Ensured by validation) + // assert!(stack.values.len() >= result_m); + + // 2. Pop m values from the stack + stack.values.pop_results(&func_ty.results) + // The values are returned as the results of the invocation. + }) + .map_state(|coro_state| -> SuspendedFunc { + SuspendedFunc { + func: SuspendedFuncInner::Wasm(SuspendedWasmFunc { + runtime: coro_state, + result_types: func_ty.results.clone(), + }), + module_addr: self.module_addr, + store_id: store.id(), + } + })) } } @@ -113,6 +142,102 @@ impl FuncHandleTyped { // Convert the Vec back to R R::from_wasm_value_tuple(&result) } + + /// call a typed function, anticipating possible suspension of execution + pub fn call_coro(&self, store: &mut Store, params: P) -> Result> { + // Convert params into Vec + let wasm_values = params.into_wasm_value_tuple(); + + // Call the underlying WASM function + let result = self.func.call_coro(store, &wasm_values)?; + + // Convert the Vec back to R + result + .map_result(|vals| R::from_wasm_value_tuple(&vals)) + .map_state(|state| SuspendedFuncTyped:: { func: state, _marker: Default::default() }) + .propagate_err_result() + } +} + +pub(crate) type FuncHandleCallOutcome = crate::coro::PotentialCoroCallResult, SuspendedFunc>; +pub(crate) type TypedFuncHandleCallOutcome = crate::coro::PotentialCoroCallResult>; + +#[derive(Debug)] +#[cfg_attr(not(feature = "async"), allow(unused))] +struct SuspendedWasmFunc { + runtime: interpreter::SuspendedRuntime, + result_types: Box<[ValType]>, +} +impl SuspendedWasmFunc { + #[cfg(feature = "async")] + fn resume( + &mut self, + ctx: FuncContext<'_>, + arg: ResumeArgument, + ) -> Result>> { + Ok(self.runtime.resume(ctx, arg)?.map_result(|mut stack| stack.values.pop_results(&self.result_types))) + } +} + +#[derive(Debug)] +#[cfg_attr(not(feature = "async"), allow(unused))] +#[allow(clippy::large_enum_variant)] // Wasm is bigger, but also much more common variant +enum SuspendedFuncInner { + Wasm(SuspendedWasmFunc), + Host(SuspendedHostCoroState), +} + +/// handle to function that was suspended and can be resumed +#[derive(Debug)] +#[cfg_attr(not(feature = "async"), allow(unused))] +pub struct SuspendedFunc { + func: SuspendedFuncInner, + module_addr: ModuleInstanceAddr, + store_id: usize, +} + +impl crate::coro::CoroState, &mut Store> for SuspendedFunc { + #[cfg(feature = "async")] + fn resume( + &mut self, + store: &mut Store, + arg: ResumeArgument, + ) -> Result>> { + if store.id() != self.store_id { + return Err(Error::InvalidStore); + } + + let ctx = FuncContext { store, module_addr: self.module_addr }; + match &mut self.func { + SuspendedFuncInner::Wasm(wasm) => wasm.resume(ctx, arg), + SuspendedFuncInner::Host(host) => Ok(host.coro_state.resume(ctx, arg)?), + } + } +} + +/// A typed suspended function. +/// Only returned value(s) are typed, yielded value and resume argument types are impossible to know +#[cfg_attr(not(feature = "async"), allow(unused))] +pub struct SuspendedFuncTyped { + /// The underlying untyped suspended function + pub func: SuspendedFunc, + pub(crate) _marker: core::marker::PhantomData, +} + +impl core::fmt::Debug for SuspendedFuncTyped { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.debug_struct("SuspendedFuncTyped").field("func", &self.func).finish() + } +} + +impl crate::coro::CoroState for SuspendedFuncTyped +where + R: FromWasmValueTuple, +{ + #[cfg(feature = "async")] + fn resume(&mut self, ctx: &mut Store, arg: ResumeArgument) -> Result> { + self.func.resume(ctx, arg)?.map_result(|vals| R::from_wasm_value_tuple(&vals)).propagate_err() + } } macro_rules! impl_into_wasm_value_tuple { diff --git a/crates/tinywasm/src/imports.rs b/crates/tinywasm/src/imports.rs index 0797470..08ce049 100644 --- a/crates/tinywasm/src/imports.rs +++ b/crates/tinywasm/src/imports.rs @@ -5,8 +5,9 @@ use alloc::string::{String, ToString}; use alloc::vec::Vec; use core::fmt::Debug; +use crate::coro::CoroState; use crate::func::{FromWasmValueTuple, IntoWasmValueTuple, ValTypesFromTuple}; -use crate::{log, LinkingError, MemoryRef, MemoryRefMut, Result}; +use crate::{coro, log, LinkingError, MemoryRef, MemoryRefMut, PotentialCoroCallResult, Result}; use tinywasm_types::*; /// The internal representation of a function @@ -28,6 +29,11 @@ impl Function { } } +/// A "resumable" function. If a host function need to suspend wasm execution +/// it can return [`coro::PotentialCoroCallResult::Suspended`] with an object that implements this trait +pub trait HostCoroState: for<'a> CoroState, FuncContext<'a>> + core::fmt::Debug + Send {} +impl CoroState, FuncContext<'a>>> HostCoroState for T {} + /// A host function pub struct HostFunction { pub(crate) ty: tinywasm_types::FuncType, @@ -41,12 +47,14 @@ impl HostFunction { } /// Call the function - pub fn call(&self, ctx: FuncContext<'_>, args: &[WasmValue]) -> Result> { + pub fn call(&self, ctx: FuncContext<'_>, args: &[WasmValue]) -> Result { (self.func)(ctx, args) } } -pub(crate) type HostFuncInner = Box, &[WasmValue]) -> Result>>; +pub(crate) type InnerHostFunCallOutcome = coro::PotentialCoroCallResult, Box>; + +pub(crate) type HostFuncInner = Box, &[WasmValue]) -> Result>; /// The context of a host-function call #[derive(Debug)] @@ -134,13 +142,22 @@ impl Extern { Self::Memory { ty } } + /// Create a new function import + pub fn func_coro( + ty: &tinywasm_types::FuncType, + func: impl Fn(FuncContext<'_>, &[WasmValue]) -> Result + 'static, + ) -> Self { + Self::Function(Function::Host(Rc::new(HostFunction { func: Box::new(func), ty: ty.clone() }))) + } + /// Create a new function import pub fn func( ty: &tinywasm_types::FuncType, func: impl Fn(FuncContext<'_>, &[WasmValue]) -> Result> + 'static, ) -> Self { let _ty = ty.clone(); - let inner_func = move |ctx: FuncContext<'_>, args: &[WasmValue]| -> Result> { + let inner_func = move |ctx: FuncContext<'_>, args: &[WasmValue]| + -> Result { let _ty = _ty.clone(); let result = func(ctx, args)?; @@ -155,10 +172,30 @@ impl Extern { Ok(()) })?; - Ok(result) + Ok(PotentialCoroCallResult::Return(result)) }; Self::Function(Function::Host(Rc::new(HostFunction { func: Box::new(inner_func), ty: ty.clone() }))) + + } + + /// Create a new typed function import + pub fn typed_func_coro( + func: impl Fn(FuncContext<'_>, P) -> Result>> + + 'static, + ) -> Self + where + P: FromWasmValueTuple + ValTypesFromTuple, + R: IntoWasmValueTuple + ValTypesFromTuple + Debug, + { + let inner_func = move |ctx: FuncContext<'_>, args: &[WasmValue]| -> Result { + let args = P::from_wasm_value_tuple(args)?; + let result = func(ctx, args)?; + Ok(result.map_result(|vals|{vals.into_wasm_value_tuple().to_vec()})) + }; + + let ty = tinywasm_types::FuncType { params: P::val_types(), results: R::val_types() }; + Self::Function(Function::Host(Rc::new(HostFunction { func: Box::new(inner_func), ty }))) } /// Create a new typed function import @@ -167,10 +204,10 @@ impl Extern { P: FromWasmValueTuple + ValTypesFromTuple, R: IntoWasmValueTuple + ValTypesFromTuple + Debug, { - let inner_func = move |ctx: FuncContext<'_>, args: &[WasmValue]| -> Result> { + let inner_func = move |ctx: FuncContext<'_>, args: &[WasmValue]| -> Result { let args = P::from_wasm_value_tuple(args)?; let result = func(ctx, args)?; - Ok(result.into_wasm_value_tuple().to_vec()) + Ok(InnerHostFunCallOutcome::Return(result.into_wasm_value_tuple().to_vec())) }; let ty = tinywasm_types::FuncType { params: P::val_types(), results: R::val_types() }; diff --git a/crates/tinywasm/src/instance.rs b/crates/tinywasm/src/instance.rs index eebf5ff..4eaa6db 100644 --- a/crates/tinywasm/src/instance.rs +++ b/crates/tinywasm/src/instance.rs @@ -3,6 +3,8 @@ use tinywasm_types::*; use crate::func::{FromWasmValueTuple, IntoWasmValueTuple}; use crate::{Error, FuncHandle, FuncHandleTyped, Imports, MemoryRef, MemoryRefMut, Module, Result, Store}; +#[cfg(feature = "async")] +use crate::{PotentialCoroCallResult, SuspendedFunc}; /// An instanciated WebAssembly module /// @@ -189,7 +191,7 @@ impl ModuleInstance { } /// Get an exported memory by name - pub fn exported_memory<'a>(&self, store: &'a mut Store, name: &str) -> Result> { + pub fn exported_memory<'a>(&self, store: &'a Store, name: &str) -> Result> { let export = self.export_addr(name).ok_or_else(|| Error::Other(format!("Export not found: {name}")))?; let ExternVal::Memory(mem_addr) = export else { return Err(Error::Other(format!("Export is not a memory: {}", name))); @@ -263,4 +265,19 @@ impl ModuleInstance { let _ = func.call(store, &[])?; Ok(Some(())) } + + /// Invoke the start function of the module + /// + /// Returns None if the module has no start function + /// If start function suspends, returns SuspendedFunc. + /// Only when it finishes can this module instance be considered instantiated + #[cfg(feature = "async")] + pub fn start_coro(&self, store: &mut Store) -> Result>> { + let Some(func) = self.start_func(store)? else { + return Ok(None); + }; + + let res = func.call_coro(store, &[])?; + Ok(Some(res.map_result(|_| {}))) + } } diff --git a/crates/tinywasm/src/interpreter/executor.rs b/crates/tinywasm/src/interpreter/executor.rs index f05ae30..b660f9e 100644 --- a/crates/tinywasm/src/interpreter/executor.rs +++ b/crates/tinywasm/src/interpreter/executor.rs @@ -2,45 +2,155 @@ #[allow(unused_imports)] use super::no_std_floats::NoStdFloatExt; +use alloc::boxed::Box; use alloc::{format, rc::Rc, string::ToString}; use core::ops::ControlFlow; use interpreter::simd::exec_next_simd; use interpreter::stack::CallFrame; use tinywasm_types::*; +#[cfg(feature = "async")] +use coro::SuspendReason; + use super::num_helpers::*; use super::stack::{BlockFrame, BlockType, Stack}; use super::values::*; use crate::*; +pub(crate) enum ReasonToBreak { + Errored(Error), + #[cfg_attr(not(feature = "async"), allow(unused))] + Suspended(SuspendReason), + Finished, +} + +impl From for ControlFlow { + fn from(value: ReasonToBreak) -> Self { + ControlFlow::Break(value) + } +} + +#[derive(Debug)] +#[cfg_attr(not(feature = "async"), allow(unused))] +pub(crate) struct SuspendedHostCoroState { + pub(crate) coro_state: Box, + // plug into used in store.get_func to get original function + // can be used for checking returned types + #[allow(dead_code)] // knowing context is useful for debug and other possible future uses + pub(crate) coro_orig_function: u32, +} + +#[derive(Debug)] pub(crate) struct Executor<'store, 'stack> { pub(crate) cf: CallFrame, pub(crate) module: ModuleInstance, + #[cfg(feature = "async")] + pub(crate) suspended_host_coro: Option, pub(crate) store: &'store mut Store, pub(crate) stack: &'stack mut Stack, } +pub(crate) type ExecOutcome = coro::CoroStateResumeResult<()>; + impl<'store, 'stack> Executor<'store, 'stack> { pub(crate) fn new(store: &'store mut Store, stack: &'stack mut Stack) -> Result { let current_frame = stack.call_stack.pop().expect("no call frame, this is a bug"); let current_module = store.get_module_instance_raw(current_frame.module_addr()); - Ok(Self { cf: current_frame, module: current_module, stack, store }) + Ok(Self { + cf: current_frame, + module: current_module, + #[cfg(feature = "async")] + suspended_host_coro: None, + stack, + store, + }) } #[inline(always)] - pub(crate) fn run_to_completion(&mut self) -> Result<()> { + pub(crate) fn run_to_suspension(&mut self) -> Result { loop { if let ControlFlow::Break(res) = self.exec_next() { return match res { - Some(e) => Err(e), - None => Ok(()), + ReasonToBreak::Errored(e) => Err(e), + ReasonToBreak::Suspended(suspend_reason) => Ok(ExecOutcome::Suspended(suspend_reason)), + ReasonToBreak::Finished => Ok(ExecOutcome::Return(())), }; } } } + #[cfg(feature = "async")] #[inline(always)] - fn exec_next(&mut self) -> ControlFlow> { + pub(crate) fn resume(&mut self, res_arg: ResumeArgument) -> Result { + if let Some(coro_state) = self.suspended_host_coro.as_mut() { + let ctx = FuncContext { store: self.store, module_addr: self.module.id() }; + let host_res = coro_state.coro_state.resume(ctx, res_arg)?; + let res = match host_res { + CoroStateResumeResult::Return(res) => res, + CoroStateResumeResult::Suspended(suspend_reason) => { + return Ok(ExecOutcome::Suspended(suspend_reason)); + } + }; + self.stack.values.extend_from_wasmvalues(&res); + self.suspended_host_coro = None; + + // we don't know how much time we spent in host function + if let ControlFlow::Break(ReasonToBreak::Suspended(reason)) = self.check_should_suspend() { + return Ok(ExecOutcome::Suspended(reason)); + } + } + + loop { + if let ControlFlow::Break(res) = self.exec_next() { + return match res { + ReasonToBreak::Errored(e) => Err(e), + ReasonToBreak::Suspended(suspend_reason) => Ok(ExecOutcome::Suspended(suspend_reason)), + ReasonToBreak::Finished => Ok(ExecOutcome::Return(())), + }; + } + } + } + + /// for controlling how long execution spends in wasm + /// called when execution loops back, because that might happen indefinite amount of times + /// and before and after function calls, because even without loops or infinite recursion, wasm function calls + /// can mutliply time spent in execution + /// execution may not be suspended in the middle of execution the funcion: + /// so only do it as the last thing or first thing in the intsruction execution + #[must_use = "If this returns ControlFlow::Break, the caller should propagate it"] + #[cfg(feature = "async")] + fn check_should_suspend(&mut self) -> ControlFlow { + if let Some(flag) = &self.store.suspend_cond.suspend_flag { + if flag.load(core::sync::atomic::Ordering::Acquire) { + return ReasonToBreak::Suspended(SuspendReason::SuspendedFlag).into(); + } + } + + #[cfg(feature = "std")] + if let Some(when) = &self.store.suspend_cond.timeout_instant { + if crate::std::time::Instant::now() >= *when { + return ReasonToBreak::Suspended(SuspendReason::SuspendedEpoch).into(); + } + } + + if let Some(mut cb) = self.store.suspend_cond.suspend_cb.take() { + let should_suspend = matches!(cb(self.store), ControlFlow::Break(())); + self.store.suspend_cond.suspend_cb = Some(cb); // put it back + if should_suspend { + return ReasonToBreak::Suspended(SuspendReason::SuspendedCallback).into(); + } + } + + ControlFlow::Continue(()) + } + + #[cfg(not(feature = "async"))] + fn check_should_suspend(&mut self) -> ControlFlow { + ControlFlow::Continue(()) + } + + #[inline(always)] + fn exec_next(&mut self) -> ControlFlow { use tinywasm_types::Instruction::*; match self.cf.fetch_instr() { Nop | BrLabel(_) | I32ReinterpretF32 | I64ReinterpretF64 | F32ReinterpretI32 | F64ReinterpretI64 => {} @@ -311,11 +421,11 @@ impl<'store, 'stack> Executor<'store, 'stack> { } #[cold] - fn exec_unreachable(&self) -> ControlFlow> { - ControlFlow::Break(Some(Trap::Unreachable.into())) + fn exec_unreachable(&self) -> ControlFlow { + ReasonToBreak::Errored(Trap::Unreachable.into()).into() } - fn exec_call(&mut self, wasm_func: Rc, owner: ModuleInstanceAddr) -> ControlFlow> { + fn exec_call(&mut self, wasm_func: Rc, owner: ModuleInstanceAddr) -> ControlFlow { let locals = self.stack.values.pop_locals(wasm_func.params, wasm_func.locals); let new_call_frame = CallFrame::new_raw(wasm_func, owner, locals, self.stack.blocks.len() as u32); self.cf.incr_instr_ptr(); // skip the call instruction @@ -323,25 +433,41 @@ impl<'store, 'stack> Executor<'store, 'stack> { self.module.swap_with(self.cf.module_addr(), self.store); ControlFlow::Continue(()) } - fn exec_call_direct(&mut self, v: u32) -> ControlFlow> { - let func_inst = self.store.get_func(self.module.resolve_func_addr(v)); + fn exec_call_host(&mut self, host_func: Rc, _func_ref: u32) -> ControlFlow { + let params = self.stack.values.pop_params(&host_func.ty.params); + let res = host_func.call(FuncContext { store: self.store, module_addr: self.module.id() }, ¶ms).to_cf()?; + match res { + PotentialCoroCallResult::Return(res) => { + self.stack.values.extend_from_wasmvalues(&res); + self.cf.incr_instr_ptr(); + self.check_should_suspend()?; // who knows how long we've spent in host function + ControlFlow::Continue(()) + } + #[cfg(feature = "async")] + PotentialCoroCallResult::Suspended(suspend_reason, state) => { + self.suspended_host_coro = + Some(SuspendedHostCoroState { coro_state: state, coro_orig_function: _func_ref }); + self.cf.incr_instr_ptr(); + ReasonToBreak::Suspended(suspend_reason).into() + } + } + } + fn exec_call_direct(&mut self, v: u32) -> ControlFlow { + self.check_should_suspend()?; // don't commit to function if we should be stopping now + let func_ref = self.module.resolve_func_addr(v); + let func_inst = self.store.get_func(func_ref); let wasm_func = match &func_inst.func { crate::Function::Wasm(wasm_func) => wasm_func, crate::Function::Host(host_func) => { - let func = &host_func.clone(); - let params = self.stack.values.pop_params(&host_func.ty.params); - let res = - func.call(FuncContext { store: self.store, module_addr: self.module.id() }, ¶ms).to_cf()?; - self.stack.values.extend_from_wasmvalues(&res); - self.cf.incr_instr_ptr(); - return ControlFlow::Continue(()); + return self.exec_call_host(host_func.clone(), func_ref); } }; self.exec_call(wasm_func.clone(), func_inst.owner) } - fn exec_call_indirect(&mut self, type_addr: u32, table_addr: u32) -> ControlFlow> { - // verify that the table is of the right type, this should be validated by the parser already + fn exec_call_indirect(&mut self, type_addr: u32, table_addr: u32) -> ControlFlow { + self.check_should_suspend()?; // check if we should suspend now before commiting to function + // verify that the table is of the right type, this should be validated by the parser already let func_ref = { let table = self.store.get_table(self.module.resolve_table_addr(table_addr)); let table_idx: u32 = self.stack.values.pop::() as u32; @@ -361,30 +487,21 @@ impl<'store, 'stack> Executor<'store, 'stack> { crate::Function::Wasm(f) => f, crate::Function::Host(host_func) => { if unlikely(host_func.ty != *call_ty) { - return ControlFlow::Break(Some( + return ReasonToBreak::Errored( Trap::IndirectCallTypeMismatch { actual: host_func.ty.clone(), expected: call_ty.clone() } .into(), - )); + ) + .into(); } - - let host_func = host_func.clone(); - let params = self.stack.values.pop_params(&host_func.ty.params); - let res = - match host_func.call(FuncContext { store: self.store, module_addr: self.module.id() }, ¶ms) { - Ok(res) => res, - Err(e) => return ControlFlow::Break(Some(e)), - }; - - self.stack.values.extend_from_wasmvalues(&res); - self.cf.incr_instr_ptr(); - return ControlFlow::Continue(()); + return self.exec_call_host(host_func.clone(), func_ref); } }; if unlikely(wasm_func.ty != *call_ty) { - return ControlFlow::Break(Some( + return ReasonToBreak::Errored( Trap::IndirectCallTypeMismatch { actual: wasm_func.ty.clone(), expected: call_ty.clone() }.into(), - )); + ) + .into(); } self.exec_call(wasm_func.clone(), func_inst.owner) @@ -424,52 +541,74 @@ impl<'store, 'stack> Executor<'store, 'stack> { ty, }); } - fn exec_br(&mut self, to: u32) -> ControlFlow> { - if self.cf.break_to(to, &mut self.stack.values, &mut self.stack.blocks).is_none() { + fn exec_br(&mut self, to: u32) -> ControlFlow { + let block_ty = self.cf.break_to(to, &mut self.stack.values, &mut self.stack.blocks); + if block_ty.is_none() { return self.exec_return(); } self.cf.incr_instr_ptr(); + + if matches!(block_ty, Some(BlockType::Loop)) { + self.check_should_suspend()?; + } ControlFlow::Continue(()) } - fn exec_br_if(&mut self, to: u32) -> ControlFlow> { - if self.stack.values.pop::() != 0 - && self.cf.break_to(to, &mut self.stack.values, &mut self.stack.blocks).is_none() - { - return self.exec_return(); - } + fn exec_br_if(&mut self, to: u32) -> ControlFlow { + let should_check_suspend = if self.stack.values.pop::() != 0 { + // condition says we should break + let block_ty = self.cf.break_to(to, &mut self.stack.values, &mut self.stack.blocks); + if block_ty.is_none() { + return self.exec_return(); + } + matches!(block_ty, Some(BlockType::Loop)) + } else { + // condition says we shouldn't break + false + }; + self.cf.incr_instr_ptr(); + + if should_check_suspend { + self.check_should_suspend()?; + } ControlFlow::Continue(()) } - fn exec_brtable(&mut self, default: u32, len: u32) -> ControlFlow> { + fn exec_brtable(&mut self, default: u32, len: u32) -> ControlFlow { let start = self.cf.instr_ptr() + 1; let end = start + len as usize; if end > self.cf.instructions().len() { - return ControlFlow::Break(Some(Error::Other(format!( + return ReasonToBreak::Errored(Error::Other(format!( "br_table out of bounds: {} >= {}", end, self.cf.instructions().len() - )))); + ))) + .into(); } let idx = self.stack.values.pop::(); let to = match self.cf.instructions()[start..end].get(idx as usize) { None => default, Some(Instruction::BrLabel(to)) => *to, - _ => return ControlFlow::Break(Some(Error::Other("br_table out of bounds".to_string()))), + _ => return ReasonToBreak::Errored(Error::Other("br_table out of bounds".to_string())).into(), }; - if self.cf.break_to(to, &mut self.stack.values, &mut self.stack.blocks).is_none() { + let block_ty = self.cf.break_to(to, &mut self.stack.values, &mut self.stack.blocks); + if block_ty.is_none() { return self.exec_return(); } self.cf.incr_instr_ptr(); + + if matches!(block_ty, Some(BlockType::Loop)) { + self.check_should_suspend()?; + } ControlFlow::Continue(()) } - fn exec_return(&mut self) -> ControlFlow> { + fn exec_return(&mut self) -> ControlFlow { let old = self.cf.block_ptr(); match self.stack.call_stack.pop() { - None => return ControlFlow::Break(None), + None => return ReasonToBreak::Finished.into(), Some(cf) => self.cf = cf, } @@ -478,6 +617,8 @@ impl<'store, 'stack> Executor<'store, 'stack> { } self.module.swap_with(self.cf.module_addr(), self.store); + + self.check_should_suspend()?; ControlFlow::Continue(()) } fn exec_end_block(&mut self) { @@ -615,16 +756,17 @@ impl<'store, 'stack> Executor<'store, 'stack> { mem_addr: tinywasm_types::MemAddr, offset: u64, cast: fn(LOAD) -> TARGET, - ) -> ControlFlow> { + ) -> ControlFlow { let mem = self.store.get_mem(self.module.resolve_mem_addr(mem_addr)); let val = self.stack.values.pop::() as u64; let Some(Ok(addr)) = offset.checked_add(val).map(TryInto::try_into) else { cold(); - return ControlFlow::Break(Some(Error::Trap(Trap::MemoryOutOfBounds { + return ReasonToBreak::Errored(Error::Trap(Trap::MemoryOutOfBounds { offset: val as usize, len: LOAD_SIZE, max: 0, - }))); + })) + .into(); }; let val = mem.load_as::(addr).to_cf()?; self.stack.values.push(cast(val)); @@ -635,13 +777,13 @@ impl<'store, 'stack> Executor<'store, 'stack> { mem_addr: tinywasm_types::MemAddr, offset: u64, cast: fn(T) -> U, - ) -> ControlFlow> { + ) -> ControlFlow { let mem = self.store.get_mem_mut(self.module.resolve_mem_addr(mem_addr)); let val = self.stack.values.pop::(); let val = (cast(val)).to_mem_bytes(); let addr = self.stack.values.pop::() as u64; if let Err(e) = mem.store((offset + addr) as usize, val.len(), &val) { - return ControlFlow::Break(Some(e)); + return ReasonToBreak::Errored(e).into(); } ControlFlow::Continue(()) } diff --git a/crates/tinywasm/src/interpreter/mod.rs b/crates/tinywasm/src/interpreter/mod.rs index e5c1081..65b2fbb 100644 --- a/crates/tinywasm/src/interpreter/mod.rs +++ b/crates/tinywasm/src/interpreter/mod.rs @@ -6,8 +6,13 @@ mod values; #[cfg(not(feature = "std"))] mod no_std_floats; +#[cfg(feature = "async")] +use {executor::Executor, tinywasm_types::ResumeArgument}; -use crate::{Result, Store}; +use crate::coro; +use crate::{FuncContext, ModuleInstance, Result, Store}; +use executor::SuspendedHostCoroState; +use stack::{CallFrame, Stack}; pub use values::*; /// The main `TinyWasm` runtime. @@ -16,8 +21,78 @@ pub use values::*; #[derive(Debug, Default)] pub struct InterpreterRuntime {} +#[derive(Debug)] +#[cfg_attr(not(feature = "async"), allow(unused))] +pub(crate) struct SuspendedRuntimeBody { + pub(crate) suspended_host_coro: Option, + pub(crate) module: ModuleInstance, + pub(crate) frame: CallFrame, +} + +#[derive(Debug)] +pub(crate) struct SuspendedRuntime { + #[cfg_attr(not(feature = "async"), allow(unused))] + pub(crate) body: Option<(SuspendedRuntimeBody, Stack)>, +} +#[cfg(feature = "async")] +impl SuspendedRuntime { + fn make_exec<'store, 'stack>( + body: SuspendedRuntimeBody, + stack: &'stack mut Stack, + store: &'store mut Store, + ) -> Executor<'store, 'stack> { + Executor { cf: body.frame, suspended_host_coro: body.suspended_host_coro, module: body.module, store, stack } + } + fn unmake_exec(exec: Executor<'_, '_>) -> SuspendedRuntimeBody { + SuspendedRuntimeBody { suspended_host_coro: exec.suspended_host_coro, module: exec.module, frame: exec.cf } + } +} + +impl coro::CoroState> for SuspendedRuntime { + #[cfg(feature = "async")] + fn resume( + &mut self, + ctx: FuncContext<'_>, + arg: ResumeArgument, + ) -> Result> { + // should be put back into self.body unless we're finished + let (body, mut stack) = if let Some(body_) = self.body.take() { + body_ + } else { + return Err(crate::error::Error::InvalidResume); + }; + + let mut exec = Self::make_exec(body, &mut stack, ctx.store); + let resumed = match exec.resume(arg) { + Ok(resumed) => resumed, + Err(err) => { + self.body = Some((Self::unmake_exec(exec), stack)); + return Err(err); + } + }; + match resumed { + executor::ExecOutcome::Return(()) => Ok(coro::CoroStateResumeResult::Return(stack)), + executor::ExecOutcome::Suspended(suspend) => { + self.body = Some((Self::unmake_exec(exec), stack)); + Ok(coro::CoroStateResumeResult::Suspended(suspend)) + } + } + } +} + +pub(crate) type RuntimeExecOutcome = coro::PotentialCoroCallResult; + impl InterpreterRuntime { - pub(crate) fn exec(&self, store: &mut Store, stack: &mut stack::Stack) -> Result<()> { - executor::Executor::new(store, stack)?.run_to_completion() + pub(crate) fn exec(&self, store: &mut Store, stack: stack::Stack) -> Result { + let mut stack = stack; + let mut executor = executor::Executor::new(store, &mut stack)?; + match executor.run_to_suspension()? { + coro::CoroStateResumeResult::Return(()) => Ok(RuntimeExecOutcome::Return(stack)), + #[cfg(feature = "async")] + coro::CoroStateResumeResult::Suspended(suspend) => Ok(RuntimeExecOutcome::Suspended( + suspend, + SuspendedRuntime { body: Some((SuspendedRuntime::unmake_exec(executor), stack)) }, + )), + } } } diff --git a/crates/tinywasm/src/interpreter/stack/block_stack.rs b/crates/tinywasm/src/interpreter/stack/block_stack.rs index 2267194..98e7e2f 100644 --- a/crates/tinywasm/src/interpreter/stack/block_stack.rs +++ b/crates/tinywasm/src/interpreter/stack/block_stack.rs @@ -60,7 +60,7 @@ pub(crate) struct BlockFrame { pub(crate) ty: BlockType, } -#[derive(Debug)] +#[derive(Debug, Clone, Copy)] pub(crate) enum BlockType { Loop, If, diff --git a/crates/tinywasm/src/interpreter/stack/call_stack.rs b/crates/tinywasm/src/interpreter/stack/call_stack.rs index f0e8d18..0ea5306 100644 --- a/crates/tinywasm/src/interpreter/stack/call_stack.rs +++ b/crates/tinywasm/src/interpreter/stack/call_stack.rs @@ -1,9 +1,10 @@ use core::ops::ControlFlow; use super::BlockType; +use crate::interpreter::executor::ReasonToBreak; use crate::interpreter::values::*; +use crate::unlikely; use crate::Trap; -use crate::{unlikely, Error}; use alloc::boxed::Box; use alloc::{rc::Rc, vec, vec::Vec}; @@ -28,9 +29,9 @@ impl CallStack { } #[inline] - pub(crate) fn push(&mut self, call_frame: CallFrame) -> ControlFlow> { + pub(crate) fn push(&mut self, call_frame: CallFrame) -> ControlFlow { if unlikely((self.stack.len() + 1) >= MAX_CALL_STACK_SIZE) { - return ControlFlow::Break(Some(Trap::CallStackOverflow.into())); + return ControlFlow::Break(ReasonToBreak::Errored(Trap::CallStackOverflow.into())); } self.stack.push(call_frame); ControlFlow::Continue(()) @@ -100,18 +101,25 @@ impl CallFrame { /// Break to a block at the given index (relative to the current frame) /// Returns `None` if there is no block at the given index (e.g. if we need to return, this is handled by the caller) + /// otherwise returns type if block it broke to + ///
+ /// if it returned Some (broke to block), + /// it expects caller to increment instruction pointer after calling it: + /// otherwise caller might exit block that's already exited or inter block caller's already in + ///
#[inline] pub(crate) fn break_to( &mut self, break_to_relative: u32, values: &mut super::ValueStack, blocks: &mut super::BlockStack, - ) -> Option<()> { + ) -> Option { let break_to = blocks.get_relative_to(break_to_relative, self.block_ptr)?; + let block_ty = break_to.ty; // instr_ptr points to the label instruction, but the next step // will increment it by 1 since we're changing the "current" instr_ptr - match break_to.ty { + match block_ty { BlockType::Loop => { // this is a loop, so we want to jump back to the start of the loop self.instr_ptr = break_to.instr_ptr; @@ -123,7 +131,7 @@ impl CallFrame { if break_to_relative != 0 { // we also want to trim the label stack to the loop (but not including the loop) blocks.truncate(blocks.len() as u32 - break_to_relative); - return Some(()); + return Some(BlockType::Loop); } } @@ -140,7 +148,7 @@ impl CallFrame { } } - Some(()) + Some(block_ty) } #[inline] diff --git a/crates/tinywasm/src/lib.rs b/crates/tinywasm/src/lib.rs index d3431e2..06dd495 100644 --- a/crates/tinywasm/src/lib.rs +++ b/crates/tinywasm/src/lib.rs @@ -91,6 +91,19 @@ pub(crate) mod log { } mod error; +#[cfg(not(feature = "async"))] +#[allow(unused)] +use { + coro::{CoroState, CoroStateResumeResult, PotentialCoroCallResult, SuspendReason}, + func::{SuspendedFunc, SuspendedFuncTyped}, +}; +#[cfg(feature = "async")] +pub use { + coro::{CoroState, CoroStateResumeResult, PotentialCoroCallResult, SuspendReason}, + func::{SuspendedFunc, SuspendedFuncTyped}, + module::IncompleteModule, +}; + pub use error::*; pub use func::{FuncHandle, FuncHandleTyped}; pub use imports::*; @@ -99,6 +112,7 @@ pub use module::Module; pub use reference::*; pub use store::*; +mod coro; mod func; mod imports; mod instance; diff --git a/crates/tinywasm/src/module.rs b/crates/tinywasm/src/module.rs index 26cea9c..c46fc8f 100644 --- a/crates/tinywasm/src/module.rs +++ b/crates/tinywasm/src/module.rs @@ -1,3 +1,8 @@ +#[cfg(feature = "async")] +use crate::{CoroState, PotentialCoroCallResult, SuspendedFunc}; +#[cfg(feature = "async")] +use tinywasm_types::ResumeArgument; + use crate::{Imports, ModuleInstance, Result, Store}; use tinywasm_types::TinyWasmModule; @@ -56,4 +61,61 @@ impl Module { let _ = instance.start(store)?; Ok(instance) } + + /// same as [Self::instantiate] but accounts for possibility of start function suspending, in which case it returns + /// [PotentialCoroCallResult::Suspended]. You can call [CoroState::resume] on it at any time to resume instantiation + #[cfg(feature = "async")] + pub fn instantiate_coro( + self, + store: &mut Store, + imports: Option, + ) -> Result> { + let instance = ModuleInstance::instantiate(store, self, imports)?; + let core_res = match instance.start_coro(store)? { + Some(res) => res, + None => return Ok(PotentialCoroCallResult::Return(instance)), + }; + Ok(match core_res { + crate::PotentialCoroCallResult::Return(_) => PotentialCoroCallResult::Return(instance), + crate::PotentialCoroCallResult::Suspended(suspend_reason, state) => { + PotentialCoroCallResult::Suspended(suspend_reason, IncompleteModule(Some(HitTheFloor(instance, state)))) + } + }) + } +} + +/// a corostate that results in [ModuleInstance] when finished +#[derive(Debug)] +#[cfg(feature = "async")] +pub struct IncompleteModule(Option); + +#[derive(Debug)] +#[cfg(feature = "async")] +struct HitTheFloor(ModuleInstance, SuspendedFunc); + +#[cfg(feature = "async")] +impl CoroState for IncompleteModule { + fn resume(&mut self, ctx: &mut Store, arg: ResumeArgument) -> Result> { + let mut body: HitTheFloor = match self.0.take() { + Some(body) => body, + None => return Err(crate::Error::InvalidResume), + }; + let coro_res = match body.1.resume(ctx, arg) { + Ok(res) => res, + Err(e) => { + self.0 = Some(body); + return Err(e); + } + }; + match coro_res { + crate::CoroStateResumeResult::Return(_) => { + let res = body.0; + Ok(crate::CoroStateResumeResult::Return(res)) + } + crate::CoroStateResumeResult::Suspended(suspend_reason) => { + self.0 = Some(body); // ...once told me + Ok(crate::CoroStateResumeResult::Suspended(suspend_reason)) + } + } + } } diff --git a/crates/tinywasm/src/store/mod.rs b/crates/tinywasm/src/store/mod.rs index 86cbb5d..4ff7f85 100644 --- a/crates/tinywasm/src/store/mod.rs +++ b/crates/tinywasm/src/store/mod.rs @@ -11,8 +11,12 @@ mod element; mod function; mod global; mod memory; +mod suspend_conditions; mod table; +#[cfg(feature = "async")] +pub use suspend_conditions::*; + pub(crate) use {data::*, element::*, function::*, global::*, memory::*, table::*}; // global store id counter @@ -33,6 +37,13 @@ pub struct Store { pub(crate) data: StoreData, pub(crate) runtime: Runtime, + + // idk where really to put it, but it should be accessible to host environment (obviously) + // and (less obviously) to host functions called from store - for calling wasm callbacks and propagating this config to them + // (or just complying with suspend conditions themselves) + // alternatively it could be passed to function handles and passend into function context + #[cfg(feature = "async")] + pub(crate) suspend_cond: SuspendConditions, } impl Debug for Store { @@ -83,7 +94,14 @@ impl PartialEq for Store { impl Default for Store { fn default() -> Self { let id = STORE_ID.fetch_add(1, Ordering::Relaxed); - Self { id, module_instances: Vec::new(), data: StoreData::default(), runtime: Runtime::Default } + Self { + id, + module_instances: Vec::new(), + data: StoreData::default(), + runtime: Runtime::Default, + #[cfg(feature = "async")] + suspend_cond: SuspendConditions::default(), + } } } @@ -476,3 +494,21 @@ fn get_pair_mut(slice: &mut [T], i: usize, j: usize) -> Option<(&mut T, &mut let pair = if i < j { (&mut x[0], &mut y[0]) } else { (&mut y[0], &mut x[0]) }; Some(pair) } + +// suspend_conditions-related functions +#[cfg(feature = "async")] +impl Store { + /// sets suspend conditions for store + pub fn set_suspend_conditions(&mut self, val: SuspendConditions) { + self.suspend_cond = val; + } + /// gets suspend conditions of store + pub fn get_suspend_conditions(&self) -> &SuspendConditions { + &self.suspend_cond + } + /// transforms suspend conditions for store using user-provided function + pub fn update_suspend_conditions(&mut self, mapper: impl FnOnce(SuspendConditions) -> SuspendConditions) { + let temp = core::mem::take(&mut self.suspend_cond); + self.suspend_cond = mapper(temp); + } +} diff --git a/crates/tinywasm/src/store/suspend_conditions.rs b/crates/tinywasm/src/store/suspend_conditions.rs new file mode 100644 index 0000000..5c6ccdf --- /dev/null +++ b/crates/tinywasm/src/store/suspend_conditions.rs @@ -0,0 +1,96 @@ +#![cfg(feature = "async")] + +use crate::store::Store; +use alloc::boxed::Box; +use core::fmt::Debug; +use core::ops::ControlFlow; + +/// user callback for use in [SuspendConditions::suspend_cb] +pub type ShouldSuspendCb = Box ControlFlow<(), ()>>; + +/// used to limit execution time wasm code takes +#[derive(Default)] +#[non_exhaustive] // some fields are feature-gated, use with*-methods to construct +pub struct SuspendConditions { + /// atomic flag. when set to true it means execution should suspend + /// can be used to tell executor to stop from another thread + pub suspend_flag: Option>, + + /// instant at which execution should suspend + /// can be used to control how much time will be spent in wasm without requiring other threads + /// such as for time-slice multitasking + /// uses rust standard library for checking time - so not available in no-std + #[cfg(feature = "std")] + pub timeout_instant: Option, + + /// callback that returns [`ControlFlow::Break`]` when execution should suspend + /// can be used when above ways are insufficient or + /// instead of [`timeout_instant`] in no-std builds, with your own clock function + pub suspend_cb: Option, +} + +impl Debug for SuspendConditions { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + let stop_cb_text = if self.suspend_cb.is_some() { "" } else { "" }; + let mut f = f.debug_struct("SuspendConditions"); + f.field("stop_flag", &self.suspend_flag); + #[cfg(feature = "std")] + { + f.field("timeout_instant", &self.timeout_instant); + } + f.field("stop_cb", &stop_cb_text).finish() + } +} + +impl SuspendConditions { + /// creates suspend_conditions with every condition unset + pub fn new() -> Self { + Default::default() + } + + /// sets timeout_instant to `how_long` from now + #[cfg(feature = "std")] + pub fn set_timeout_in(&mut self, how_long: crate::std::time::Duration) -> &mut Self { + self.timeout_instant = Some(crate::std::time::Instant::now() + how_long); + self + } + /// adds timeout at specified instant + #[cfg(feature = "std")] + pub fn with_timeout_at(self, when: crate::std::time::Instant) -> Self { + Self { timeout_instant: Some(when), ..self } + } + /// adds timeout in specified duration + #[cfg(feature = "std")] + pub fn with_timeout_in(self, how_long: crate::std::time::Duration) -> Self { + Self { timeout_instant: Some(crate::std::time::Instant::now() + how_long), ..self } + } + /// removes timeout + pub fn without_timeout(self) -> Self { + #[cfg(feature = "std")] + { + Self { timeout_instant: None, ..self } + } + #[cfg(not(feature = "std"))] + { + self + } + } + + /// adds susped flag + pub fn with_suspend_flag(self, should_suspend: alloc::sync::Arc) -> Self { + Self { suspend_flag: Some(should_suspend), ..self } + } + /// removes susped flag + pub fn without_suspend_flag(self) -> Self { + Self { suspend_flag: None, ..self } + } + + /// adds suspend callback + pub fn with_suspend_callback(self, cb: ShouldSuspendCb) -> Self { + Self { suspend_cb: Some(cb), ..self } + } + /// removes suspend callback + pub fn without_suspend_callback(self) -> Self { + Self { suspend_cb: None, ..self } + } +} diff --git a/crates/tinywasm/tests/testsuite/util.rs b/crates/tinywasm/tests/testsuite/util.rs index b555153..35155c9 100644 --- a/crates/tinywasm/tests/testsuite/util.rs +++ b/crates/tinywasm/tests/testsuite/util.rs @@ -1,6 +1,9 @@ +use std::hash::Hasher; use std::panic::{self, AssertUnwindSafe}; use eyre::{bail, eyre, Result}; +#[cfg(feature = "test_async")] +use tinywasm::{CoroState, SuspendConditions, SuspendReason}; use tinywasm_types::{ExternRef, FuncRef, ModuleInstanceAddr, TinyWasmModule, ValType, WasmValue}; use wasm_testsuite::wast; use wasm_testsuite::wast::{core::AbstractHeapType, QuoteWat}; @@ -12,6 +15,25 @@ pub fn try_downcast_panic(panic: Box) -> String { info.unwrap_or(info_str.unwrap_or(&info_string.unwrap_or("unknown panic".to_owned())).to_string()) } +// due to imprecision it's not exact +#[cfg(feature = "test_async")] +fn make_sometimes_breaking_cb(probability: f64) -> impl FnMut(&tinywasm::Store) -> std::ops::ControlFlow<(), ()> { + let mut counter = 0 as u64; + let mut hasher = std::hash::DefaultHasher::new(); + let threshhold = (probability * (u64::MAX as f64)) as u64; // 2 lossy conversions + + move |_| { + hasher.write_u64(counter); + counter += 1; + if hasher.finish() < threshhold { + std::ops::ControlFlow::Break(()) + } else { + std::ops::ControlFlow::Continue(()) + } + } +} + +#[cfg(not(feature = "test_async"))] pub fn exec_fn_instance( instance: Option<&ModuleInstanceAddr>, store: &mut tinywasm::Store, @@ -30,6 +52,50 @@ pub fn exec_fn_instance( func.call(store, args) } +#[cfg(feature = "test_async")] +pub fn exec_fn_instance( + instance: Option<&ModuleInstanceAddr>, + store: &mut tinywasm::Store, + name: &str, + args: &[tinywasm_types::WasmValue], +) -> Result, tinywasm::Error> { + let Some(instance) = instance else { + return Err(tinywasm::Error::Other("no instance found".to_string())); + }; + + let mut prev_reason = None; + store.update_suspend_conditions(|old_cond| { + prev_reason = Some(old_cond); + SuspendConditions::new().with_suspend_callback(Box::new(make_sometimes_breaking_cb(2.0 / 3.0))) + }); + let res = || -> Result, tinywasm::Error> { + let Some(instance) = store.get_module_instance(*instance) else { + return Err(tinywasm::Error::Other("no instance found".to_string())); + }; + + let func = instance.exported_func_untyped(store, name)?; + let mut state = match func.call_coro(store, args)? { + tinywasm::PotentialCoroCallResult::Return(val) => return Ok(val), + tinywasm::PotentialCoroCallResult::Suspended(suspend_reason, state) => { + assert!(matches!(suspend_reason, SuspendReason::SuspendedCallback)); + state + } + }; + loop { + match state.resume(store, None)? { + tinywasm::CoroStateResumeResult::Return(val) => return Ok(val), + tinywasm::CoroStateResumeResult::Suspended(suspend_reason) => { + assert!(matches!(suspend_reason, SuspendReason::SuspendedCallback)) + } + } + } + }(); + // restore store suspend conditions before returning error or success + store.set_suspend_conditions(prev_reason.unwrap()); + res +} + +#[cfg(not(feature = "test_async"))] pub fn exec_fn( module: Option<&TinyWasmModule>, name: &str, @@ -39,13 +105,60 @@ pub fn exec_fn( let Some(module) = module else { return Err(tinywasm::Error::Other("no module found".to_string())); }; - let mut store = tinywasm::Store::new(); let module = tinywasm::Module::from(module); let instance = module.instantiate(&mut store, imports)?; instance.exported_func_untyped(&store, name)?.call(&mut store, args) } +#[cfg(feature = "test_async")] +pub fn exec_fn( + module: Option<&TinyWasmModule>, + name: &str, + args: &[tinywasm_types::WasmValue], + imports: Option, +) -> Result, tinywasm::Error> { + let Some(module) = module else { + return Err(tinywasm::Error::Other("no module found".to_string())); + }; + + let mut store = tinywasm::Store::new(); + + store.set_suspend_conditions( + SuspendConditions::new().with_suspend_callback(Box::new(make_sometimes_breaking_cb(2.0 / 3.0))), + ); + + let module = tinywasm::Module::from(module); + let instance = match module.instantiate_coro(&mut store, imports)? { + tinywasm::PotentialCoroCallResult::Return(res) => res, + tinywasm::PotentialCoroCallResult::Suspended(suspend_reason, mut state) => loop { + assert!(matches!(suspend_reason, SuspendReason::SuspendedCallback)); + match state.resume(&mut store, None)? { + tinywasm::CoroStateResumeResult::Return(res) => break res, + tinywasm::CoroStateResumeResult::Suspended(suspend_reason) => { + assert!(matches!(suspend_reason, SuspendReason::SuspendedCallback)); + } + } + }, + }; + + let mut state = match instance.exported_func_untyped(&store, name)?.call_coro(&mut store, args)? { + tinywasm::PotentialCoroCallResult::Return(r) => return Ok(r), + tinywasm::PotentialCoroCallResult::Suspended(suspend_reason, state) => { + assert!(matches!(suspend_reason, SuspendReason::SuspendedCallback)); + state + } + }; + loop { + match state.resume(&mut store, None)? { + tinywasm::CoroStateResumeResult::Return(res) => return Ok(res), + tinywasm::CoroStateResumeResult::Suspended(suspend_reason) => { + assert!(matches!(suspend_reason, SuspendReason::SuspendedCallback)) + } + } + } +} + pub fn catch_unwind_silent(f: impl FnOnce() -> R) -> std::thread::Result { let prev_hook = panic::take_hook(); panic::set_hook(Box::new(|_| {})); diff --git a/crates/tinywasm/tests/wasm_resume.rs b/crates/tinywasm/tests/wasm_resume.rs new file mode 100644 index 0000000..156ed95 --- /dev/null +++ b/crates/tinywasm/tests/wasm_resume.rs @@ -0,0 +1,374 @@ +#![cfg(feature = "async")] + +use core::panic; +use eyre; +use std::sync; +use std::{ops::ControlFlow, time::Duration}; +use tinywasm::{ + CoroState, CoroStateResumeResult, Extern, Imports, Module, ModuleInstance, PotentialCoroCallResult, Store, + SuspendConditions, SuspendReason, +}; +use wat; + +#[test] +fn main() -> std::result::Result<(), eyre::Report> { + println!("\n# testing with callback"); + let mut cb_cond = |store: &mut Store| { + let callback = make_suspend_in_time_cb(30); + store.set_suspend_conditions(SuspendConditions::new().with_suspend_callback(Box::new(callback))); + }; + suspend_with_pure_loop(&mut cb_cond, SuspendReason::SuspendedCallback)?; + suspend_with_wasm_fn(&mut cb_cond, SuspendReason::SuspendedCallback)?; + suspend_with_host_fn(&mut cb_cond, SuspendReason::SuspendedCallback)?; + + println!("\n# testing with epoch"); + let mut time_cond = |store: &mut Store| { + store.set_suspend_conditions(SuspendConditions::new().with_timeout_in(Duration::from_millis(10))) + }; + suspend_with_pure_loop(&mut time_cond, SuspendReason::SuspendedEpoch)?; + suspend_with_wasm_fn(&mut time_cond, SuspendReason::SuspendedEpoch)?; + suspend_with_host_fn(&mut time_cond, SuspendReason::SuspendedEpoch)?; + + println!("\n# testing atomic bool"); + let mut cb_thead = |store: &mut Store| { + let arc = sync::Arc::::new(sync::atomic::AtomicBool::new(false)); + store.set_suspend_conditions(SuspendConditions::new().with_suspend_flag(arc.clone())); + let handle = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(10)); + arc.store(true, sync::atomic::Ordering::Release); + }); + drop(handle); + }; + suspend_with_pure_loop(&mut cb_thead, SuspendReason::SuspendedFlag)?; + suspend_with_wasm_fn(&mut cb_thead, SuspendReason::SuspendedFlag)?; + suspend_with_host_fn(&mut cb_thead, SuspendReason::SuspendedFlag)?; + + Ok(()) +} + +fn make_suspend_in_time_cb(milis: u64) -> impl FnMut(&Store) -> ControlFlow<(), ()> { + let mut counter = 0 as u64; + move |_| -> ControlFlow<(), ()> { + counter += 1; + if counter > milis { + counter = 0; + ControlFlow::Break(()) + } else { + ControlFlow::Continue(()) + } + } +} + +fn try_compare(lhs: &SuspendReason, rhs: &SuspendReason) -> eyre::Result { + Ok(match lhs { + SuspendReason::Yield(_) => eyre::bail!("Can't compare yields"), + SuspendReason::SuspendedEpoch => matches!(rhs, SuspendReason::SuspendedEpoch), + SuspendReason::SuspendedCallback => matches!(rhs, SuspendReason::SuspendedCallback), + SuspendReason::SuspendedFlag => matches!(rhs, SuspendReason::SuspendedFlag), + _ => eyre::bail!("unimplemented new variant"), + }) +} + +// check if you can suspend while looping +fn suspend_with_pure_loop( + set_cond: &mut impl FnMut(&mut Store) -> (), + expected_reason: SuspendReason, +) -> eyre::Result<()> { + println!("## test suspend in loop"); + + let wasm: String = { + let detect_overflow = overflow_detect_snippet("$res"); + format!( + r#"(module + (memory $mem 1) + (export "memory" (memory $mem)) ;; first 8 bytes - counter, next 4 - overflow flag + + (func (export "start_counter") + (local $res i64) + (loop $lp + (i32.const 0) ;;where to store + (i64.load $mem (i32.const 0)) + (i64.const 1) + (i64.add) + (local.set $res) + (local.get $res) + (i64.store $mem) + {detect_overflow} + (br $lp) + ) + ) + )"# + ) + .into() + }; + + let mut tested = { + let wasm = wat::parse_str(wasm)?; + let module = Module::parse_bytes(&wasm)?; + let mut store = Store::default(); + let instance = module.instantiate(&mut store, None)?; + TestedModule { store, instance: instance, resumable: None } + }; + + let increases = run_loops_look_at_numbers(&mut tested, set_cond, expected_reason, 16)?; + assert!(increases > 2, "code doesn't enough: either suspend condition is too tight or something is broken"); + Ok(()) +} + +// check if you can suspend when calling wasm function +fn suspend_with_wasm_fn( + set_cond: &mut impl FnMut(&mut Store) -> (), + expected_reason: SuspendReason, +) -> eyre::Result<()> { + println!("## test suspend wasm fn"); + + let wasm: String = { + let detect_overflow = overflow_detect_snippet("$res"); + format!( + r#"(module + (memory $mem 1) + (export "memory" (memory $mem)) ;; first 8 bytes - counter, next 8 - overflow counter + + (func $wasm_nop + nop + ) + + (func $wasm_adder (param i64 i64) (result i64) + (local.get 0) + (local.get 1) + (i64.add) + ) + + (func $overflow_detect (param $res i64) + {detect_overflow} + ) + + (func (export "start_counter") + (local $res i64) + (loop $lp + (call $wasm_nop) + (i32.const 0) ;;where to store + (i64.load $mem (i32.const 0)) + (i64.const 1) + (call $wasm_adder) + (local.set $res) + (call $wasm_nop) + (local.get $res) + (i64.store $mem) + (local.get $res) + (call $overflow_detect) + (call $wasm_nop) + (br $lp) + ) + ) + )"# + ) + .into() + }; + + let mut tested = { + let wasm = wat::parse_str(wasm)?; + let module = Module::parse_bytes(&wasm)?; + let mut store = Store::default(); + let instance = module.instantiate(&mut store, None)?; + TestedModule { store, instance: instance, resumable: None } + }; + + let increases = run_loops_look_at_numbers(&mut tested, set_cond, expected_reason, 16)?; + assert!(increases > 2, "code doesn't enough: either suspend condition is too tight or something is broken"); + + Ok(()) +} + +// check if you can suspend when calling host function +fn suspend_with_host_fn( + set_cond: &mut impl FnMut(&mut Store) -> (), + expected_reason: SuspendReason, +) -> eyre::Result<()> { + println!("## test suspend host fn"); + + let wasm: String = { + format!( + r#"(module + (import "host" "adder" (func $host_adder (param i64 i64)(result i64))) + (import "host" "nop" (func $host_nop)) + (import "host" "detect" (func $overflow_detect (param $res i64))) + (memory $mem 1) + (export "memory" (memory $mem)) ;; first 8 bytes - counter, next 8 - overflow counter + + (func (export "start_counter") + (local $res i64) + (loop $lp + (call $host_nop) + (i32.const 0) ;;where to store + (i64.load $mem (i32.const 0)) + (i64.const 1) + (call $host_adder) + (local.set $res) + (call $host_nop) + (local.get $res) + (i64.store $mem) + (local.get $res) + (call $overflow_detect) + (call $host_nop) + (br $lp) + ) + ) + )"# + ) + .into() + }; + + let mut tested = { + let wasm = wat::parse_str(wasm)?; + let module = Module::parse_bytes(&wasm)?; + let mut store = Store::default(); + let mut imports = Imports::new(); + imports.define( + "host", + "adder", + Extern::typed_func(|_, args: (i64, i64)| -> tinywasm::Result { Ok(args.0 + args.1) }), + )?; + imports.define( + "host", + "nop", + Extern::typed_func(|_, ()| -> tinywasm::Result<()> { + std::thread::sleep(Duration::from_micros(1)); + Ok(()) + }), + )?; + imports.define( + "host", + "detect", + Extern::typed_func(|mut ctx, arg: i64| -> tinywasm::Result<()> { + if arg != 0 { + return Ok(()); + } + let mut mem = ctx.module().exported_memory_mut(ctx.store_mut(), "memory").expect("where's memory"); + let mut buf = [0 as u8; 8]; + buf.copy_from_slice(mem.load(8, 8)?); + let counter = i64::from_be_bytes(buf); + mem.store(8, 8, &i64::to_be_bytes(counter + 1))?; + Ok(()) + }), + )?; + + let instance = module.instantiate(&mut store, Some(imports))?; + TestedModule { store, instance: instance, resumable: None } + }; + + let increases = run_loops_look_at_numbers(&mut tested, set_cond, expected_reason, 16)?; + assert!(increases > 2, "code doesn't enough: either suspend condition is too tight or something is broken"); + Ok(()) +} + +fn run_loops_look_at_numbers( + tested: &mut TestedModule, + set_cond: &mut impl FnMut(&mut Store) -> (), + expected_reason: SuspendReason, + times: u32, +) -> eyre::Result { + set_cond(&mut tested.store); + let suspend = tested.start_counter_incrementing_loop("start_counter")?; + assert!(try_compare(&suspend, &expected_reason).expect("unexpected yield")); + + let mut prev_counter = tested.get_counter(); + let mut times_increased = 0 as u32; + + { + let (big, small) = prev_counter; + println!("after start {big} {small}"); + } + + assert!(prev_counter >= (0, 0)); + + for _ in 0..times - 1 { + set_cond(&mut tested.store); + assert!(try_compare(&tested.continue_counter_incrementing_loop()?, &expected_reason)?); + let new_counter = tested.get_counter(); + // save for scheduling weirdness, loop should run for a bunch of times in 3ms + assert!(new_counter >= prev_counter); + { + let (big, small) = new_counter; + println!("after continue {big} {small}"); + } + if new_counter > prev_counter { + times_increased += 1; + } + prev_counter = new_counter; + } + Ok(times_increased) +} + +fn overflow_detect_snippet(var: &str) -> String { + format!( + r#"(i64.eq (i64.const 0) (local.get {var})) + (if + (then + ;; we wrapped around back to 0 - set flag + (i32.const 8) ;;where to store + (i32.const 8) ;;where to load + (i64.load) + (i64.const 1) + (i64.add) + (i64.store $mem) + ) + (else + nop + ) + ) + "# + ) + .into() +} + +// should have exported memory "memory" and +struct TestedModule { + store: Store, + instance: ModuleInstance, + resumable: Option, +} + +impl TestedModule { + fn start_counter_incrementing_loop(&mut self, fn_name: &str) -> tinywasm::Result { + let starter = self.instance.exported_func_untyped(&self.store, fn_name)?; + if let PotentialCoroCallResult::Suspended(res, coro) = starter.call_coro(&mut self.store, &[])? { + self.resumable = Some(coro); + return Ok(res); + } else { + panic!("that should never return"); + } + } + + fn continue_counter_incrementing_loop(&mut self) -> tinywasm::Result { + let paused = if let Some(val) = self.resumable.as_mut() { + val + } else { + panic!("nothing to continue"); + }; + let resume_res = (*paused).resume(&mut self.store, None)?; + match resume_res { + CoroStateResumeResult::Suspended(res) => Ok(res), + CoroStateResumeResult::Return(_) => panic!("should never return"), + } + } + + // (counter, overflow flag) + fn get_counter(&self) -> (u64, u64) { + let counter_now = { + let mem = self.instance.exported_memory(&self.store, "memory").expect("where's memory"); + let mut buff: [u8; 8] = [0; 8]; + let in_mem = mem.load(0, 8).expect("where's memory"); + buff.clone_from_slice(in_mem); + u64::from_le_bytes(buff) + }; + let overflow_times = { + let mem = self.instance.exported_memory(&self.store, "memory").expect("where's memory"); + let mut buff: [u8; 8] = [0; 8]; + let in_mem = mem.load(8, 8).expect("where's memory"); + buff.clone_from_slice(in_mem); + u64::from_le_bytes(buff) + }; + (overflow_times, counter_now) + } +} diff --git a/crates/types/src/lib.rs b/crates/types/src/lib.rs index 58120fe..a34e26c 100644 --- a/crates/types/src/lib.rs +++ b/crates/types/src/lib.rs @@ -10,7 +10,7 @@ extern crate alloc; use alloc::boxed::Box; -use core::{fmt::Debug, ops::Range}; +use core::{any::Any, fmt::Debug, ops::Range}; // Memory defaults const MEM_PAGE_SIZE: u64 = 65536; @@ -408,3 +408,6 @@ pub enum ElementItem { Func(FuncAddr), Expr(ConstInstruction), } + +pub type YieldedValue = Option>; +pub type ResumeArgument = Option>; diff --git a/examples/host_coro.rs b/examples/host_coro.rs new file mode 100644 index 0000000..8cce656 --- /dev/null +++ b/examples/host_coro.rs @@ -0,0 +1,156 @@ +use eyre::{self, bail}; +use tinywasm::{ + types::{FuncType, ValType, WasmValue}, + CoroState, CoroStateResumeResult, Extern, FuncContext, HostCoroState, Imports, Module, PotentialCoroCallResult, + Store, SuspendReason, +}; +use wat; + +fn main() -> eyre::Result<()> { + untyped()?; + typed()?; + Ok(()) +} + +const WASM: &str = r#"(module + (import "host" "hello" (func $host_hello (param i32))) + (import "host" "wait" (func $host_suspend (param i32)(result i32))) + + (func (export "call_hello") (result f32) + (call $host_hello (i32.const -3)) + (call $host_suspend (i32.const 10)) + (call $host_hello) + (f32.const 6.28) + ) +) +"#; + +#[derive(Debug)] +struct MyUserData { + magic: u16, +} + +#[derive(Debug)] +struct MySuspendedState { + base: i32, +} +impl<'_> CoroState, FuncContext<'_>> for MySuspendedState { + fn resume( + &mut self, + _: FuncContext<'_>, + arg: tinywasm::types::ResumeArgument, + ) -> tinywasm::Result>> { + let val = arg.expect("you din't send").downcast::().expect("you sent wrong"); + return Ok(CoroStateResumeResult::Return(vec![WasmValue::I32(*val + self.base)])); + } +} + +fn untyped() -> eyre::Result<()> { + let wasm = wat::parse_str(WASM).expect("failed to parse wat"); + let module = Module::parse_bytes(&wasm)?; + let mut store = Store::default(); + + let mut imports = Imports::new(); + imports.define( + "host", + "hello", + Extern::func(&FuncType { params: Box::new([ValType::I32]), results: Box::new([]) }, |_: FuncContext<'_>, x| { + x.first().map(|x| println!("{:?}", x)); + Ok(vec![]) + }), + )?; + let my_coro_starter = |_ctx: FuncContext<'_>, + vals: &[WasmValue]| + -> tinywasm::Result, Box>> { + let base = if let WasmValue::I32(v) = vals.first().expect("wrong args") { v } else { panic!("wrong arg") }; + let coro = Box::new(MySuspendedState { base: *base }); + return Ok(PotentialCoroCallResult::Suspended( + SuspendReason::make_yield::(MyUserData { magic: 42 }), + coro, + )); + }; + imports.define( + "host", + "wait", + Extern::func_coro( + &FuncType { params: Box::new([ValType::I32]), results: Box::new([ValType::I32]) }, + my_coro_starter, + ), + )?; + + let instance = module.instantiate(&mut store, Some(imports))?; + + let greeter = instance.exported_func_untyped(&store, "call_hello")?; + let call_res = greeter.call_coro(&mut store, &[])?; + let mut resumable = match call_res { + tinywasm::PotentialCoroCallResult::Return(..) => bail!("it's not supposed to return yet"), + tinywasm::PotentialCoroCallResult::Suspended(SuspendReason::Yield(Some(val)), resumable) => { + match val.downcast::() { + Ok(val) => assert_eq!(val.magic, 42), + Err(_) => bail!("invalid yielded val"), + } + resumable + } + tinywasm::PotentialCoroCallResult::Suspended(..) => bail!("wrong suspend"), + }; + + let final_res = resumable.resume(&mut store, Some(Box::::new(7)))?; + if let CoroStateResumeResult::Return(vals) = final_res { + println!("{:?}", vals.first().unwrap()); + } else { + panic!("should have finished"); + } + + Ok(()) +} + +fn typed() -> eyre::Result<()> { + let wasm = wat::parse_str(WASM).expect("failed to parse wat"); + let module = Module::parse_bytes(&wasm)?; + let mut store = Store::default(); + + let mut imports = Imports::new(); + imports.define( + "host", + "hello", + Extern::typed_func(|_: FuncContext<'_>, x: i32| { + println!("{x}"); + Ok(()) + }), + )?; + let my_coro_starter = + |_ctx: FuncContext<'_>, base: i32| -> tinywasm::Result>> { + let coro = Box::new(MySuspendedState { base: base }); + return Ok(PotentialCoroCallResult::Suspended( + SuspendReason::make_yield::(MyUserData { magic: 42 }), + coro, + )); + }; + imports.define("host", "wait", Extern::typed_func_coro(my_coro_starter))?; + + let instance = module.instantiate(&mut store, Some(imports))?; + + let greeter = instance.exported_func::<(), f32>(&store, "call_hello")?; + let call_res = greeter.call_coro(&mut store, ())?; + let mut resumable = match call_res { + tinywasm::PotentialCoroCallResult::Return(..) => bail!("it's not supposed to return yet"), + tinywasm::PotentialCoroCallResult::Suspended(SuspendReason::Yield(Some(val)), resumable) => { + match val.downcast::() { + Ok(val) => assert_eq!(val.magic, 42), + Err(_) => bail!("invalid yielded val"), + } + resumable + } + tinywasm::PotentialCoroCallResult::Suspended(..) => bail!("wrong suspend"), + }; + + let final_res = resumable.resume(&mut store, Some(Box::::new(7)))?; + + if let CoroStateResumeResult::Return(res) = final_res { + println!("{res}"); + } else { + panic!("should have returned"); + } + + Ok(()) +}