diff --git a/Cargo.toml b/Cargo.toml index 2cda5b8..896d376 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,8 @@ name = "vm" [features] default = ["runtime", "cli", "cranelift-jit"] runtime = [] -http-client = ["runtime", "dep:reqwest", "dep:url", "dep:tokio", "dep:futures-util"] +async = ["runtime", "dep:reqwest", "dep:url", "dep:tokio", "dep:futures-util"] +http-client = ["async"] sqlite = ["runtime", "dep:rusqlite"] edge-abi = [ "dep:edge_abi", diff --git a/build.rs b/build.rs index 64979a9..f5c1deb 100644 --- a/build.rs +++ b/build.rs @@ -115,6 +115,7 @@ struct CallableDecl { wrapper: Option, host_binding_kind: HostBindingKind, host_execution: HostExecutionKind, + runtime_owned_pending: bool, } #[derive(Clone, Debug)] @@ -155,17 +156,19 @@ fn main() { module: "host".to_string(), category: SourceCategory::DefaultHost, }, - SourceSpec { - path: "src/builtins/runtime/http.rs".to_string(), - module: "http".to_string(), - category: SourceCategory::DefaultHost, - }, SourceSpec { path: "src/builtins/runtime/context_host.rs".to_string(), module: "context_host".to_string(), category: SourceCategory::DefaultHost, }, ]; + if env::var_os("CARGO_FEATURE_ASYNC").is_some() { + host_sources.push(SourceSpec { + path: "src/builtins/runtime/http.rs".to_string(), + module: "http".to_string(), + category: SourceCategory::DefaultHost, + }); + } if env::var_os("CARGO_FEATURE_SQLITE").is_some() { host_sources.push(SourceSpec { path: "src/builtins/runtime/sqlite.rs".to_string(), @@ -273,6 +276,9 @@ fn parse_sources( } pub(crate) fn classify_host_binding(function: &ItemFn) -> HostBindingKind { + if function.sig.asyncness.is_some() { + return HostBindingKind::StaticStack; + } if function.sig.inputs.iter().any(|input| match input { FnArg::Typed(pat_type) => is_vm_context_type(&pat_type.ty), _ => false, @@ -298,6 +304,9 @@ pub(crate) fn classify_host_binding(function: &ItemFn) -> HostBindingKind { } pub(crate) fn infer_host_execution(function: &ItemFn) -> HostExecutionKind { + if function.sig.asyncness.is_some() { + return HostExecutionKind::MaySuspend; + } let return_type = normalized_return_type(&function.sig.output); if contains_host_call_result(&return_type) { HostExecutionKind::MaySuspend @@ -444,6 +453,8 @@ fn parse_source_file(path: &Path, spec: &SourceSpec, _order_offset: usize) -> Ve wrapper, host_binding_kind: classify_host_binding(function), host_execution: infer_host_execution(function), + runtime_owned_pending: function.sig.asyncness.is_none() + && contains_host_call_result(&normalized_return_type(&function.sig.output)), }); } out @@ -1095,12 +1106,14 @@ fn render_builtin_runtime_dispatch( ) .unwrap(); } - writeln!( - &mut out, - " registry.mark_runtime_owned_pending({:?});", - callable.name - ) - .unwrap(); + if callable.runtime_owned_pending { + writeln!( + &mut out, + " registry.mark_runtime_owned_pending({:?});", + callable.name + ) + .unwrap(); + } } writeln!(&mut out, "}}").unwrap(); writeln!(&mut out).unwrap(); @@ -1117,12 +1130,14 @@ fn render_builtin_runtime_dispatch( .render_bind_static_call(&callable.name, &host_wrapper_adapter_name(callable)); writeln!(&mut out, " {:?} => {{", callable.name).unwrap(); writeln!(&mut out, " {bind_call}").unwrap(); - writeln!( - &mut out, - " vm.mark_runtime_owned_pending_binding({:?});", - callable.name - ) - .unwrap(); + if callable.runtime_owned_pending { + writeln!( + &mut out, + " vm.mark_runtime_owned_pending_binding({:?});", + callable.name + ) + .unwrap(); + } writeln!(&mut out, " true").unwrap(); writeln!(&mut out, " }}").unwrap(); } @@ -1891,6 +1906,9 @@ fn host_wrapper_adapter_name(callable: &CallableDecl) -> String { fn generated_wrapper_decl(function: &ItemFn) -> WrapperDecl { let mut params = Vec::new(); + if function.sig.asyncness.is_some() { + params.push(WrapperParamKind::Vm); + } for input in &function.sig.inputs { let FnArg::Typed(pat_type) = input else { panic!("methods are not supported in #[pd_host_function] declarations"); @@ -1917,6 +1935,13 @@ fn parse_callable_params(function: &ItemFn) -> Vec { let FnArg::Typed(pat_type) = input else { panic!("methods are not supported in #[pd_host_function] declarations"); }; + if pat_type + .attrs + .iter() + .any(|attr| attr.path().is_ident("pd_host_context")) + { + return None; + } if is_vm_context_type(&pat_type.ty) { return None; } diff --git a/pd-host-function/src/edge.rs b/pd-host-function/src/edge.rs deleted file mode 100644 index 4f6c268..0000000 --- a/pd-host-function/src/edge.rs +++ /dev/null @@ -1,1162 +0,0 @@ -use quote::{format_ident, quote}; -use syn::{ - Error, Expr, FnArg, Ident, ItemFn, LitStr, Meta, Pat, PatIdent, ReturnType, Token, Type, - punctuated::Punctuated, -}; - -pub(crate) fn expand_scoped_pd_host_function( - attr: Punctuated, - mut item: ItemFn, -) -> Result { - let edge_attr = parse_edge_host_attr(&attr)?; - let was_async = item.sig.asyncness.is_some(); - let docs = doc_string(&item.attrs); - if docs.trim().is_empty() { - return Err(Error::new_spanned( - &item.sig.ident, - "#[pd_host_function] requires /// doc comments", - )); - } - if edge_attr.scope.is_some() && !edge_attr.bind_params.is_empty() { - return Err(Error::new_spanned( - &item.sig.ident, - "scoped pd_host_function does not support bind(...); scoped registrations must be self-contained", - )); - } - - transform_async_edge_function(&mut item, &edge_attr)?; - validate_edge_bind_names(&item, &edge_attr.bind_params)?; - for input in &item.sig.inputs { - validate_edge_param(input, &edge_attr.bind_params)?; - } - validate_edge_return_type(&item.sig.output)?; - - let (wrapper_name, impl_name) = wrapper_and_impl_names(&item.sig.ident); - if item.sig.ident != impl_name { - item.sig.ident = impl_name.clone(); - } - let wrapper = generate_edge_host_binder(&item, &wrapper_name, &edge_attr)?; - let static_wrapper = - generate_scoped_edge_host_static_wrapper(&item, &wrapper_name, &edge_attr, was_async)?; - let registration = generate_edge_host_registration(&item, &wrapper_name, &edge_attr, &docs)?; - Ok(quote! { - #item - #wrapper - #static_wrapper - #registration - }) -} - -fn doc_string(attrs: &[syn::Attribute]) -> String { - attrs - .iter() - .filter_map(|attr| { - if !attr.path().is_ident("doc") { - return None; - } - match &attr.meta { - Meta::NameValue(name_value) => match &name_value.value { - syn::Expr::Lit(expr_lit) => match &expr_lit.lit { - syn::Lit::Str(value) => Some(value.value().trim().to_string()), - _ => None, - }, - _ => None, - }, - _ => None, - } - }) - .filter(|line| !line.is_empty()) - .collect::>() - .join("\n") -} - -struct EdgeHostAttr { - name: Expr, - scope: Option, - bind_params: Vec, -} - -#[derive(Clone, Copy)] -enum EdgeHostScopeAttr { - Runtime, - Http, - HttpExtension, - Io, - Transport, - Mqtt, - WebSocket, - WebRtc, - Proxy, - Console, -} - -fn parse_edge_host_attr(args: &Punctuated) -> Result { - let mut name = None; - let mut scope = None; - let mut bind_params = Vec::new(); - - for meta in args { - match meta { - Meta::NameValue(name_value) if name_value.path.is_ident("name") => { - if name.is_some() { - return Err(Error::new_spanned( - name_value, - "duplicate name argument in #[pd_host_function(...)]", - )); - } - name = Some(name_value.value.clone()); - } - Meta::NameValue(name_value) if name_value.path.is_ident("scope") => { - if scope.is_some() { - return Err(Error::new_spanned( - name_value, - "duplicate scope argument in #[pd_host_function(...)]", - )); - } - scope = Some(parse_edge_scope(&name_value.value)?); - } - Meta::List(list) if list.path.is_ident("bind") => { - let idents = - list.parse_args_with(Punctuated::::parse_terminated)?; - bind_params.extend(idents.into_iter()); - } - other => { - return Err(Error::new_spanned( - other, - "expected #[pd_host_function(name = ..., scope = ..., bind(...))]", - )); - } - } - } - - let Some(name) = name else { - return Err(Error::new( - proc_macro2::Span::call_site(), - "expected #[pd_host_function(name = ..., scope = ..., bind(...))]", - )); - }; - - Ok(EdgeHostAttr { - name, - scope, - bind_params, - }) -} - -fn parse_edge_scope(value: &Expr) -> Result { - let scope_name = match value { - Expr::Path(path) => { - let Some(segment) = path.path.segments.last() else { - return Err(Error::new_spanned( - value, - "scope must be one of runtime, http, http_extension, io, transport, mqtt, websocket, webrtc, proxy, or console", - )); - }; - if path.path.segments.len() != 1 { - return Err(Error::new_spanned( - value, - "scope must be one of runtime, http, http_extension, io, transport, mqtt, websocket, webrtc, proxy, or console", - )); - } - segment.ident.to_string() - } - Expr::Lit(expr_lit) => match &expr_lit.lit { - syn::Lit::Str(value) => value.value(), - _ => { - return Err(Error::new_spanned( - value, - "scope must be one of runtime, http, http_extension, io, transport, mqtt, websocket, webrtc, proxy, or console", - )); - } - }, - _ => { - return Err(Error::new_spanned( - value, - "scope must be one of runtime, http, http_extension, io, transport, mqtt, websocket, webrtc, proxy, or console", - )); - } - }; - - match scope_name.as_str() { - "runtime" => Ok(EdgeHostScopeAttr::Runtime), - "http" => Ok(EdgeHostScopeAttr::Http), - "http_extension" | "http_extensions" => Ok(EdgeHostScopeAttr::HttpExtension), - "io" | "io_override" | "io_overrides" => Ok(EdgeHostScopeAttr::Io), - "transport" => Ok(EdgeHostScopeAttr::Transport), - "mqtt" => Ok(EdgeHostScopeAttr::Mqtt), - "websocket" => Ok(EdgeHostScopeAttr::WebSocket), - "webrtc" => Ok(EdgeHostScopeAttr::WebRtc), - "proxy" => Ok(EdgeHostScopeAttr::Proxy), - "console" => Ok(EdgeHostScopeAttr::Console), - _ => Err(Error::new_spanned( - value, - "scope must be one of runtime, http, http_extension, io, transport, mqtt, websocket, webrtc, proxy, or console", - )), - } -} - -fn edge_scope_tokens(scope: EdgeHostScopeAttr) -> proc_macro2::TokenStream { - match scope { - EdgeHostScopeAttr::Runtime => { - quote!(crate::abi_impl::registry::EdgeHostScope::Runtime) - } - EdgeHostScopeAttr::Http => quote!(crate::abi_impl::registry::EdgeHostScope::Http), - EdgeHostScopeAttr::HttpExtension => { - quote!(crate::abi_impl::registry::EdgeHostScope::HttpExtension) - } - EdgeHostScopeAttr::Io => quote!(crate::abi_impl::registry::EdgeHostScope::Io), - EdgeHostScopeAttr::Transport => { - quote!(crate::abi_impl::registry::EdgeHostScope::Transport) - } - EdgeHostScopeAttr::Mqtt => quote!(crate::abi_impl::registry::EdgeHostScope::Mqtt), - EdgeHostScopeAttr::WebSocket => { - quote!(crate::abi_impl::registry::EdgeHostScope::WebSocket) - } - EdgeHostScopeAttr::WebRtc => { - quote!(crate::abi_impl::registry::EdgeHostScope::WebRtc) - } - EdgeHostScopeAttr::Proxy => { - quote!(crate::abi_impl::registry::EdgeHostScope::Proxy) - } - EdgeHostScopeAttr::Console => { - quote!(crate::abi_impl::registry::EdgeHostScope::Console) - } - } -} - -fn find_context_param_ident(item: &ItemFn) -> Option { - item.sig.inputs.iter().find_map(|input| { - let FnArg::Typed(pat_type) = input else { - return None; - }; - if !is_edge_context_type(&pat_type.ty) { - return None; - } - match pat_type.pat.as_ref() { - Pat::Ident(PatIdent { ident, .. }) => Some(ident.clone()), - _ => None, - } - }) -} - -fn async_scope_prepare_stmt( - item: &ItemFn, - attr: &EdgeHostAttr, -) -> Result { - let Some(scope) = attr.scope else { - return Ok(quote!()); - }; - let requires_prepare = matches!( - scope, - EdgeHostScopeAttr::Http | EdgeHostScopeAttr::HttpExtension - ); - if !requires_prepare { - return Ok(quote!()); - } - let Some(context_ident) = find_context_param_ident(item) else { - return Err(Error::new_spanned( - &item.sig.ident, - "async scoped http host functions must accept SharedProxyVmContext", - )); - }; - let scope_tokens = edge_scope_tokens(scope); - let name_expr = &attr.name; - Ok(quote! { - crate::abi_impl::prepare_scoped_host_call( - #context_ident.clone(), - #scope_tokens, - #name_expr, - ) - .await?; - }) -} - -fn transform_async_edge_function(item: &mut ItemFn, attr: &EdgeHostAttr) -> Result<(), Error> { - if item.sig.asyncness.is_none() { - return Ok(()); - } - - for input in &item.sig.inputs { - let FnArg::Typed(pat_type) = input else { - return Err(Error::new_spanned(input, "methods are not supported")); - }; - if is_value_slice_type(&pat_type.ty) { - return Err(Error::new_spanned( - &pat_type.ty, - "async edge host functions do not support raw args; use typed parameters instead", - )); - } - if edge_arg_decoder_kind(&pat_type.ty) - .ok() - .is_some_and(edge_arg_decoder_is_borrowed) - { - return Err(Error::new_spanned( - &pat_type.ty, - "async edge host functions do not support borrowed typed parameters; use owned String, Value, or VmMap inputs", - )); - } - } - - let Some(vm_ident) = find_vm_param_ident(item) else { - return Err(Error::new_spanned( - &item.sig.ident, - "async edge host functions must accept a Vm parameter so the macro can schedule the future", - )); - }; - - match edge_output_kind(&item.sig.output)? { - Some(EdgeOutputKind::ResultCallOutcome) => {} - Some(EdgeOutputKind::CallOutcome) => { - return Err(Error::new_spanned( - &item.sig.output, - "async edge host functions must return Result", - )); - } - None => { - return Err(Error::new_spanned( - &item.sig.output, - "edge host functions must return CallOutcome or Result", - )); - } - } - - let original_block = item.block.clone(); - let prepare_stmt = async_scope_prepare_stmt(item, attr)?; - item.sig.asyncness = None; - *item.block = syn::parse2(quote!({ - crate::abi_impl::schedule_current_future_call(#vm_ident, async move { - #prepare_stmt - let __pd_edge_outcome = (async move #original_block).await?; - match __pd_edge_outcome { - ::vm::CallOutcome::Return(values) => Ok(values), - ::vm::CallOutcome::Halt => Err(::vm::VmError::HostError( - "async edge host functions must not return Halt".to_string(), - )), - ::vm::CallOutcome::Yield => Err(::vm::VmError::HostError( - "async edge host functions must not return Yield".to_string(), - )), - ::vm::CallOutcome::Pending(_) => Err(::vm::VmError::HostError( - "async edge host functions must not return Pending".to_string(), - )), - } - }) - }))?; - Ok(()) -} - -fn validate_edge_bind_names(item: &ItemFn, bind_params: &[Ident]) -> Result<(), Error> { - let params = item - .sig - .inputs - .iter() - .filter_map(|input| match input { - FnArg::Typed(pat_type) => match pat_type.pat.as_ref() { - Pat::Ident(ident) => Some(ident.ident.to_string()), - _ => None, - }, - FnArg::Receiver(_) => None, - }) - .collect::>(); - - for bind in bind_params { - if !params.iter().any(|name| name == &bind.to_string()) { - return Err(Error::new_spanned( - bind, - format!( - "bind parameter '{}' does not match any function parameter", - bind - ), - )); - } - } - - Ok(()) -} - -fn validate_edge_param(arg: &FnArg, bind_params: &[Ident]) -> Result<(), Error> { - let FnArg::Typed(pat_type) = arg else { - return Err(Error::new_spanned(arg, "methods are not supported")); - }; - let Pat::Ident(PatIdent { ident, .. }) = pat_type.pat.as_ref() else { - return Err(Error::new_spanned( - &pat_type.pat, - "edge host parameters must use identifier patterns", - )); - }; - - if is_vm_context_type(&pat_type.ty) - || is_edge_async_ops_type(&pat_type.ty) - || is_edge_context_type(&pat_type.ty) - || is_value_slice_type(&pat_type.ty) - { - if bind_params.iter().any(|candidate| candidate == ident) { - return Err(Error::new_spanned( - ident, - "special edge host parameters must not be listed in bind(...)", - )); - } - return Ok(()); - } - - if bind_params.iter().any(|candidate| candidate == ident) { - return Ok(()); - } - - edge_arg_decoder_kind(&pat_type.ty).map(|_| ()) -} - -fn validate_edge_return_type(output: &ReturnType) -> Result<(), Error> { - match edge_output_kind(output)? { - Some(_) => Ok(()), - None => Err(Error::new_spanned( - output, - "edge host functions must return CallOutcome or Result", - )), - } -} - -fn generate_edge_host_binder( - item: &ItemFn, - wrapper_name: &syn::Ident, - attr: &EdgeHostAttr, -) -> Result { - let impl_name = &item.sig.ident; - let vis = &item.vis; - let name_expr = &attr.name; - let mut binder_params = Vec::::new(); - let mut binder_setup = Vec::::new(); - let mut closure_setup = Vec::::new(); - let mut call_args = Vec::::new(); - let mut extract_stmts = Vec::::new(); - let mut arg_index = 0usize; - let mut raw_args = false; - - binder_params.push(quote!(bind_vm: &mut ::vm::Vm)); - binder_params.push(quote!(bind_context: &crate::abi_impl::SharedProxyVmContext)); - binder_params.push(quote!(bind_async_ops: &crate::abi_impl::SharedVmAsyncOps)); - - for input in &item.sig.inputs { - let FnArg::Typed(pat_type) = input else { - return Err(Error::new_spanned(input, "methods are not supported")); - }; - let Pat::Ident(PatIdent { ident, .. }) = pat_type.pat.as_ref() else { - return Err(Error::new_spanned( - &pat_type.pat, - "edge host parameters must use identifier patterns", - )); - }; - let ty = &pat_type.ty; - - if is_vm_context_type(ty) { - call_args.push(quote!(vm)); - continue; - } - - if is_edge_async_ops_type(ty) { - binder_setup.push(quote!(let #ident = bind_async_ops.clone();)); - closure_setup.push(quote!(let #ident = #ident.clone();)); - call_args.push(quote!(#ident)); - continue; - } - - if is_edge_context_type(ty) { - binder_setup.push(quote!(let #ident = bind_context.clone();)); - closure_setup.push(quote!(let #ident = #ident.clone();)); - call_args.push(quote!(#ident)); - continue; - } - - if is_value_slice_type(ty) { - raw_args = true; - call_args.push(quote!(args)); - continue; - } - - if attr.bind_params.iter().any(|candidate| candidate == ident) { - binder_params.push(quote!(#ident: #ty)); - binder_setup.push(quote!(let #ident = #ident.clone();)); - closure_setup.push(quote!(let #ident = #ident.clone();)); - call_args.push(quote!(#ident)); - continue; - } - - let decoder = edge_arg_decoder_kind(ty)?; - extract_stmts.push(edge_extract_stmt(ident, decoder, arg_index, wrapper_name)); - call_args.push(quote!(#ident)); - arg_index += 1; - } - - let arity_check = if raw_args { - None - } else { - Some(quote! { - if args.len() != #arg_index { - return Err(::vm::VmError::HostError(format!( - "expected {} arguments, got {}", - #arg_index, - args.len() - ))); - } - }) - }; - - let call_expr = match edge_output_kind(&item.sig.output)? { - Some(EdgeOutputKind::ResultCallOutcome) => quote!(#impl_name(#(#call_args),*)), - Some(EdgeOutputKind::CallOutcome) => quote!(Ok(#impl_name(#(#call_args),*))), - None => { - return Err(Error::new_spanned( - &item.sig.output, - "edge host functions must return CallOutcome or Result", - )); - } - }; - - Ok(quote! { - #[allow(dead_code)] - #vis fn #wrapper_name(#(#binder_params),*) { - #(#binder_setup)* - crate::abi_impl::bind_async_host_handler(bind_vm, bind_async_ops, #name_expr, move |vm, args| { - #arity_check - #(#closure_setup)* - #(#extract_stmts)* - #call_expr - }); - } - }) -} - -fn generate_scoped_edge_host_static_wrapper( - item: &ItemFn, - wrapper_name: &syn::Ident, - attr: &EdgeHostAttr, - was_async: bool, -) -> Result { - let Some(scope) = attr.scope else { - return Ok(quote!()); - }; - let impl_name = &item.sig.ident; - let static_wrapper_name = format_ident!("__pd_edge_static_{}", wrapper_name); - let uses_vm = scoped_wrapper_uses_vm(item); - let scope_tokens = edge_scope_tokens(scope); - let scope_requires_prepare = matches!( - scope, - EdgeHostScopeAttr::Http | EdgeHostScopeAttr::HttpExtension - ); - let args_only_sync_fast_path = !uses_vm && !was_async; - let prepare_context_ident = format_ident!("__pd_edge_prepare_context"); - let mut setup_stmts = Vec::::new(); - let mut call_args = Vec::::new(); - let mut extract_stmts = Vec::::new(); - let mut arg_index = 0usize; - - for input in &item.sig.inputs { - let FnArg::Typed(pat_type) = input else { - return Err(Error::new_spanned(input, "methods are not supported")); - }; - let Pat::Ident(PatIdent { ident, .. }) = pat_type.pat.as_ref() else { - return Err(Error::new_spanned( - &pat_type.pat, - "edge host parameters must use identifier patterns", - )); - }; - let ty = &pat_type.ty; - - if is_vm_context_type(ty) { - call_args.push(quote!(vm)); - continue; - } - - if is_edge_async_ops_type(ty) { - setup_stmts.push(quote!(let #ident = crate::abi_impl::current_async_ops()?;)); - call_args.push(quote!(#ident)); - continue; - } - - if is_edge_context_type(ty) { - if args_only_sync_fast_path && scope_requires_prepare { - setup_stmts.push(quote!(let #ident = #prepare_context_ident.clone();)); - } else { - setup_stmts.push(quote!(let #ident = crate::abi_impl::current_vm_context()?;)); - } - call_args.push(quote!(#ident)); - continue; - } - - if is_value_slice_type(ty) { - return Err(Error::new_spanned( - ty, - "scoped pd_host_function does not support raw args", - )); - } - - if attr.bind_params.iter().any(|candidate| candidate == ident) { - return Err(Error::new_spanned( - ident, - "scoped pd_host_function does not support bind(...)", - )); - } - - let decoder = edge_arg_decoder_kind(ty)?; - extract_stmts.push(edge_extract_stmt(ident, decoder, arg_index, wrapper_name)); - call_args.push(quote!(#ident)); - arg_index += 1; - } - - let _ = u8::try_from(arg_index).map_err(|_| { - Error::new_spanned( - &item.sig.ident, - "edge host functions must have 255 arguments or fewer", - ) - })?; - let call_expr = match edge_output_kind(&item.sig.output)? { - Some(EdgeOutputKind::ResultCallOutcome) => quote!(#impl_name(#(#call_args),*)), - Some(EdgeOutputKind::CallOutcome) => quote!(Ok(#impl_name(#(#call_args),*))), - None => { - return Err(Error::new_spanned( - &item.sig.output, - "edge host functions must return CallOutcome or Result", - )); - } - }; - - if uses_vm { - Ok(quote! { - fn #static_wrapper_name( - vm: &mut ::vm::Vm, - args: &[::vm::Value], - ) -> Result<::vm::CallOutcome, ::vm::VmError> { - if args.len() != #arg_index { - return Err(::vm::VmError::HostError(format!( - "expected {} arguments, got {}", - #arg_index, - args.len() - ))); - } - #(#setup_stmts)* - let __pd_edge_outcome = { - #(#extract_stmts)* - #call_expr - }?; - Ok(__pd_edge_outcome) - } - }) - } else if args_only_sync_fast_path && scope_requires_prepare { - let name_expr = &attr.name; - Ok(quote! { - fn #static_wrapper_name( - args: &[::vm::Value], - ) -> Result<::vm::CallOutcome, ::vm::VmError> { - if args.len() != #arg_index { - return Err(::vm::VmError::HostError(format!( - "expected {} arguments, got {}", - #arg_index, - args.len() - ))); - } - let #prepare_context_ident = crate::abi_impl::current_vm_context()?; - if !crate::abi_impl::scoped_host_call_can_run_synchronously( - &#prepare_context_ident, - #scope_tokens, - #name_expr, - )? { - return Err(::vm::VmError::HostError(format!( - "synchronous scoped host function {} requires an async signature", - #name_expr, - ))); - } - #(#setup_stmts)* - #(#extract_stmts)* - let __pd_edge_outcome = #call_expr?; - Ok(__pd_edge_outcome) - } - }) - } else { - Ok(quote! { - fn #static_wrapper_name( - args: &[::vm::Value], - ) -> Result<::vm::CallOutcome, ::vm::VmError> { - if args.len() != #arg_index { - return Err(::vm::VmError::HostError(format!( - "expected {} arguments, got {}", - #arg_index, - args.len() - ))); - } - #(#setup_stmts)* - let __pd_edge_outcome = { - #(#extract_stmts)* - #call_expr - }?; - Ok(__pd_edge_outcome) - } - }) - } -} - -fn generate_edge_host_registration( - item: &ItemFn, - wrapper_name: &syn::Ident, - attr: &EdgeHostAttr, - docs: &str, -) -> Result { - let Some(scope) = attr.scope else { - return Ok(quote!()); - }; - - let entry_name = format_ident!("__pd_edge_registration_{}", wrapper_name); - let scope_tokens = edge_scope_tokens(scope); - let static_wrapper_name = format_ident!("__pd_edge_static_{}", wrapper_name); - let function_kind = if scoped_wrapper_uses_vm(item) { - quote!(crate::abi_impl::registry::EdgeHostRegistrationFunction::StackStatic(#static_wrapper_name)) - } else { - quote!(crate::abi_impl::registry::EdgeHostRegistrationFunction::ArgsStatic(#static_wrapper_name)) - }; - let mut arity = 0usize; - - for input in &item.sig.inputs { - let FnArg::Typed(pat_type) = input else { - return Err(Error::new_spanned(input, "methods are not supported")); - }; - if is_vm_context_type(&pat_type.ty) - || is_edge_async_ops_type(&pat_type.ty) - || is_edge_context_type(&pat_type.ty) - { - continue; - } - if is_value_slice_type(&pat_type.ty) { - return Err(Error::new_spanned( - &pat_type.ty, - "scoped pd_host_function does not support raw args", - )); - } - arity += 1; - } - - let arity = u8::try_from(arity).map_err(|_| { - Error::new_spanned( - &item.sig.ident, - "edge host functions must have 255 arguments or fewer", - ) - })?; - let name_expr = &attr.name; - let docs = docs.to_string(); - - Ok(quote! { - #[::linkme::distributed_slice(crate::abi_impl::registry::PD_EDGE_HOST_FUNCTIONS)] - #[allow(non_upper_case_globals)] - static #entry_name: crate::abi_impl::registry::EdgeHostRegistration = - crate::abi_impl::registry::EdgeHostRegistration { - scope: #scope_tokens, - name: #name_expr, - arity: #arity, - docs: #docs, - function: #function_kind, - }; - }) -} - -fn scoped_wrapper_uses_vm(item: &ItemFn) -> bool { - item.sig.inputs.iter().any(|input| match input { - FnArg::Typed(pat_type) => is_vm_context_type(&pat_type.ty), - FnArg::Receiver(_) => false, - }) -} - -fn find_vm_param_ident(item: &ItemFn) -> Option { - item.sig.inputs.iter().find_map(|input| { - let FnArg::Typed(pat_type) = input else { - return None; - }; - if !is_vm_context_type(&pat_type.ty) { - return None; - } - match pat_type.pat.as_ref() { - Pat::Ident(PatIdent { ident, .. }) => Some(ident.clone()), - _ => None, - } - }) -} - -fn wrapper_and_impl_names(name: &syn::Ident) -> (syn::Ident, syn::Ident) { - let original = name.to_string(); - match original.strip_suffix("_impl") { - Some(prefix) => ( - syn::Ident::new(prefix, name.span()), - syn::Ident::new(&original, name.span()), - ), - None => ( - syn::Ident::new(&original, name.span()), - syn::Ident::new(&format!("{original}_impl"), name.span()), - ), - } -} - -#[derive(Clone, Copy)] -enum EdgeArgDecoderKind { - String, - StringRef, - Int, - Bool, - Value, - ValueRef, - Map, - MapRef, -} - -#[derive(Clone, Copy)] -enum EdgeOutputKind { - CallOutcome, - ResultCallOutcome, -} - -fn edge_arg_decoder_kind(ty: &Type) -> Result { - match ty { - Type::Group(group) => edge_arg_decoder_kind(&group.elem), - Type::Paren(paren) => edge_arg_decoder_kind(&paren.elem), - Type::Reference(reference) => { - if reference.mutability.is_some() { - return Err(Error::new_spanned( - ty, - "mutable borrowed edge host argument types are not supported", - )); - } - let Some(inner) = type_path_terminal_ident(reference.elem.as_ref()) else { - return Err(Error::new_spanned( - ty, - "unsupported borrowed edge host argument type", - )); - }; - match inner.as_str() { - "str" => Ok(EdgeArgDecoderKind::StringRef), - "Value" => Ok(EdgeArgDecoderKind::ValueRef), - "VmMap" => Ok(EdgeArgDecoderKind::MapRef), - other => Err(Error::new_spanned( - ty, - format!("unsupported borrowed edge host argument type '&{other}'"), - )), - } - } - Type::Path(path) => { - let Some(segment) = path.path.segments.last() else { - return Err(Error::new_spanned( - ty, - "unsupported edge host argument type", - )); - }; - match segment.ident.to_string().as_str() { - "String" => Ok(EdgeArgDecoderKind::String), - "i8" | "i16" | "i32" | "i64" | "isize" | "u8" | "u16" | "u32" | "u64" | "usize" => { - Ok(EdgeArgDecoderKind::Int) - } - "bool" => Ok(EdgeArgDecoderKind::Bool), - "Value" => Ok(EdgeArgDecoderKind::Value), - "VmMap" => Ok(EdgeArgDecoderKind::Map), - other => Err(Error::new_spanned( - ty, - format!("unsupported edge host argument type '{other}'"), - )), - } - } - _ => Err(Error::new_spanned( - ty, - "unsupported edge host argument type", - )), - } -} - -fn edge_extract_stmt( - ident: &Ident, - decoder: EdgeArgDecoderKind, - arg_index: usize, - wrapper_name: &syn::Ident, -) -> proc_macro2::TokenStream { - let label = LitStr::new( - &format!("{} {}", wrapper_name, ident), - proc_macro2::Span::call_site(), - ); - let index = syn::Index::from(arg_index); - match decoder { - EdgeArgDecoderKind::String => quote! { - let #ident = match args.get(#index) { - Some(::vm::Value::String(value)) => value.to_string(), - Some(_) => return Err(::vm::VmError::TypeMismatch("string")), - None => { - return Err(::vm::VmError::HostError(format!( - "missing argument: {}", - #label - ))); - } - }; - }, - EdgeArgDecoderKind::StringRef => quote! { - let #ident = match args.get(#index) { - Some(::vm::Value::String(value)) => value.as_str(), - Some(_) => return Err(::vm::VmError::TypeMismatch("string")), - None => { - return Err(::vm::VmError::HostError(format!( - "missing argument: {}", - #label - ))); - } - }; - }, - EdgeArgDecoderKind::Int => quote! { - let #ident = match args.get(#index) { - Some(::vm::Value::Int(value)) => *value, - Some(_) => return Err(::vm::VmError::TypeMismatch("int")), - None => { - return Err(::vm::VmError::HostError(format!( - "missing argument: {}", - #label - ))); - } - }; - }, - EdgeArgDecoderKind::Bool => quote! { - let #ident = match args.get(#index) { - Some(::vm::Value::Bool(value)) => *value, - Some(_) => return Err(::vm::VmError::TypeMismatch("bool")), - None => { - return Err(::vm::VmError::HostError(format!( - "missing argument: {}", - #label - ))); - } - }; - }, - EdgeArgDecoderKind::Value => quote! { - let #ident = match args.get(#index) { - Some(value) => value.clone(), - None => { - return Err(::vm::VmError::HostError(format!( - "missing argument: {}", - #label - ))); - } - }; - }, - EdgeArgDecoderKind::ValueRef => quote! { - let #ident = match args.get(#index) { - Some(value) => value, - None => { - return Err(::vm::VmError::HostError(format!( - "missing argument: {}", - #label - ))); - } - }; - }, - EdgeArgDecoderKind::Map => quote! { - let #ident = match args.get(#index) { - Some(::vm::Value::Map(entries)) => entries.as_ref().clone(), - Some(_) => return Err(::vm::VmError::TypeMismatch("map")), - None => { - return Err(::vm::VmError::HostError(format!( - "missing argument: {}", - #label - ))); - } - }; - }, - EdgeArgDecoderKind::MapRef => quote! { - let #ident = match args.get(#index) { - Some(::vm::Value::Map(entries)) => entries.as_ref(), - Some(_) => return Err(::vm::VmError::TypeMismatch("map")), - None => { - return Err(::vm::VmError::HostError(format!( - "missing argument: {}", - #label - ))); - } - }; - }, - } -} - -fn edge_arg_decoder_is_borrowed(decoder: EdgeArgDecoderKind) -> bool { - matches!( - decoder, - EdgeArgDecoderKind::StringRef | EdgeArgDecoderKind::ValueRef | EdgeArgDecoderKind::MapRef - ) -} - -fn type_path_terminal_ident(ty: &Type) -> Option { - match ty { - Type::Group(group) => type_path_terminal_ident(&group.elem), - Type::Paren(paren) => type_path_terminal_ident(&paren.elem), - Type::Path(path) => path - .path - .segments - .last() - .map(|segment| segment.ident.to_string()), - _ => None, - } -} - -fn edge_output_kind(output: &ReturnType) -> Result, Error> { - match output { - ReturnType::Default => Ok(None), - ReturnType::Type(_, ty) => { - if is_call_outcome_type(ty) { - return Ok(Some(EdgeOutputKind::CallOutcome)); - } - if is_host_call_result_type(ty) { - return Ok(Some(EdgeOutputKind::ResultCallOutcome)); - } - if let Some(inner) = unwrap_result_type(ty)? - && is_call_outcome_type(&inner) - { - return Ok(Some(EdgeOutputKind::ResultCallOutcome)); - } - Ok(None) - } - } -} - -fn unwrap_result_type(ty: &Type) -> Result, Error> { - match ty { - Type::Group(group) => unwrap_result_type(&group.elem), - Type::Paren(paren) => unwrap_result_type(&paren.elem), - Type::Reference(reference) => unwrap_result_type(&reference.elem), - Type::Path(path) => { - let Some(segment) = path.path.segments.last() else { - return Ok(None); - }; - if segment.ident != "Result" { - return Ok(None); - } - let syn::PathArguments::AngleBracketed(args) = &segment.arguments else { - return Err(Error::new_spanned( - &segment.arguments, - "Result requires generic arguments", - )); - }; - let Some(syn::GenericArgument::Type(inner)) = args.args.first() else { - return Err(Error::new_spanned( - args, - "Result requires a return type argument", - )); - }; - Ok(Some(inner.clone())) - } - _ => Ok(None), - } -} - -fn is_call_outcome_type(ty: &Type) -> bool { - match ty { - Type::Group(group) => is_call_outcome_type(&group.elem), - Type::Paren(paren) => is_call_outcome_type(&paren.elem), - Type::Reference(reference) => is_call_outcome_type(&reference.elem), - Type::Path(path) => path - .path - .segments - .last() - .is_some_and(|segment| segment.ident == "CallOutcome"), - _ => false, - } -} - -fn is_host_call_result_type(ty: &Type) -> bool { - match ty { - Type::Group(group) => is_host_call_result_type(&group.elem), - Type::Paren(paren) => is_host_call_result_type(&paren.elem), - Type::Reference(reference) => is_host_call_result_type(&reference.elem), - Type::Path(path) => path - .path - .segments - .last() - .is_some_and(|segment| segment.ident == "HostCallResult"), - _ => false, - } -} - -fn is_vm_context_type(ty: &Type) -> bool { - match ty { - Type::Group(group) => is_vm_context_type(&group.elem), - Type::Paren(paren) => is_vm_context_type(&paren.elem), - Type::Reference(reference) => is_vm_context_type(&reference.elem), - Type::Path(path) => path - .path - .segments - .last() - .is_some_and(|segment| segment.ident == "Vm"), - _ => false, - } -} - -fn is_edge_async_ops_type(ty: &Type) -> bool { - match ty { - Type::Group(group) => is_edge_async_ops_type(&group.elem), - Type::Paren(paren) => is_edge_async_ops_type(&paren.elem), - Type::Reference(reference) => is_edge_async_ops_type(&reference.elem), - Type::Path(path) => path - .path - .segments - .last() - .is_some_and(|segment| segment.ident == "SharedVmAsyncOps"), - _ => false, - } -} - -fn is_edge_context_type(ty: &Type) -> bool { - match ty { - Type::Group(group) => is_edge_context_type(&group.elem), - Type::Paren(paren) => is_edge_context_type(&paren.elem), - Type::Reference(reference) => is_edge_context_type(&reference.elem), - Type::Path(path) => path - .path - .segments - .last() - .is_some_and(|segment| segment.ident == "SharedProxyVmContext"), - _ => false, - } -} - -fn is_value_slice_type(ty: &Type) -> bool { - match ty { - Type::Group(group) => is_value_slice_type(&group.elem), - Type::Paren(paren) => is_value_slice_type(&paren.elem), - Type::Reference(reference) => matches!( - reference.elem.as_ref(), - Type::Slice(slice) - if matches!( - slice.elem.as_ref(), - Type::Path(path) - if path - .path - .segments - .last() - .is_some_and(|segment| segment.ident == "Value") - ) - ), - _ => false, - } -} - -#[cfg(test)] -mod tests { - use super::expand_scoped_pd_host_function; - use syn::{ItemFn, Meta, Token, parse_quote, punctuated::Punctuated}; - - #[test] - fn preserves_async_scoped_edge_host_expansion() { - let attr: Punctuated = parse_quote!(name = "test::suspend", scope = http); - let item: ItemFn = parse_quote! { - /// Returns a value after a scoped host operation completes. - #[pd_host_function(name = "test::suspend", scope = http)] - async fn suspend( - _vm: &mut Vm, - _context: SharedProxyVmContext, - ) -> Result { - Ok(CallOutcome::Return(vm::CallReturn::none())) - } - }; - - let expanded = expand_scoped_pd_host_function(attr, item) - .expect("async scoped edge host functions should be accepted"); - let rendered = expanded.to_string(); - assert!(rendered.contains("schedule_current_future_call")); - assert!(rendered.contains("PD_EDGE_HOST_FUNCTIONS")); - } -} diff --git a/pd-host-function/src/lib.rs b/pd-host-function/src/lib.rs index 4a87884..a6b3514 100644 --- a/pd-host-function/src/lib.rs +++ b/pd-host-function/src/lib.rs @@ -5,43 +5,36 @@ use syn::{ punctuated::Punctuated, }; -mod edge; - #[proc_macro_attribute] pub fn pd_host_function(attr: TokenStream, item: TokenStream) -> TokenStream { let args = parse_macro_input!(attr with Punctuated::::parse_terminated); let item = parse_macro_input!(item as ItemFn); - let result = if uses_edge_host_contract(&args, &item) { - edge::expand_scoped_pd_host_function(args, item) - } else { - expand_pd_host_function(args, item) - }; + let result = expand_pd_host_function(args, item); match result { Ok(tokens) => tokens.into(), Err(err) => err.to_compile_error().into(), } } -fn uses_edge_host_contract(args: &Punctuated, item: &ItemFn) -> bool { - if item.sig.asyncness.is_some() { - return true; - } - - args.iter().any(|meta| match meta { - Meta::NameValue(name_value) if name_value.path.is_ident("scope") => true, - Meta::List(list) if list.path.is_ident("bind") => true, - _ => false, - }) -} - fn expand_pd_host_function( attr: Punctuated, mut item: ItemFn, ) -> Result { parse_name_arg(&attr)?; + let is_async = item.sig.asyncness.is_some(); let docs = doc_string(&item.attrs); for input in &item.sig.inputs { - validate_param(input)?; + if is_async { + validate_async_param(input)?; + } else if is_host_context_param(input) { + return Err(Error::new_spanned( + input, + "#[pd_host_context] is only valid on async host functions", + )); + } + if !is_host_context_param(input) { + validate_param(input)?; + } } validate_return_type(&item.sig.output)?; @@ -59,13 +52,85 @@ fn expand_pd_host_function( if item.sig.ident != impl_name { item.sig.ident = impl_name.clone(); } - let wrapper = generate_vm_wrapper(&item, &wrapper_name)?; + let wrapper = if is_async { + generate_async_vm_wrapper(&item, &wrapper_name)? + } else { + generate_vm_wrapper(&item, &wrapper_name)? + }; + for input in &mut item.sig.inputs { + if let FnArg::Typed(pat_type) = input { + pat_type + .attrs + .retain(|attr| !attr.path().is_ident("pd_host_context")); + } + } Ok(quote! { #item #wrapper }) } +fn validate_async_param(arg: &FnArg) -> Result<(), Error> { + let FnArg::Typed(pat_type) = arg else { + return Err(Error::new_spanned(arg, "methods are not supported")); + }; + if is_vm_context_type(&pat_type.ty) { + return Err(Error::new_spanned( + &pat_type.ty, + "async host functions cannot borrow Vm; capture owned host context before submission", + )); + } + if is_host_context_param(arg) { + return Ok(()); + } + if !is_async_owned_type(&pat_type.ty) { + return Err(Error::new_spanned( + &pat_type.ty, + "async host function parameters must be owned and 'static", + )); + } + Ok(()) +} + +fn is_host_context_param(arg: &FnArg) -> bool { + match arg { + FnArg::Typed(pat_type) => pat_type + .attrs + .iter() + .any(|attr| attr.path().is_ident("pd_host_context")), + FnArg::Receiver(_) => false, + } +} + +fn is_async_owned_type(ty: &Type) -> bool { + match ty { + Type::Group(group) => is_async_owned_type(&group.elem), + Type::Paren(paren) => is_async_owned_type(&paren.elem), + Type::Reference(_) | Type::Slice(_) => false, + Type::Tuple(tuple) => tuple.elems.iter().all(is_async_owned_type), + Type::Path(path) => { + let Some(segment) = path.path.segments.last() else { + return false; + }; + if matches!( + segment.ident.to_string().as_str(), + "str" | "VmStringRef" | "VmBytesRef" | "VmArrayRef" | "VmMapRef" | "VmValueRef" + ) { + return false; + } + match &segment.arguments { + syn::PathArguments::None => true, + syn::PathArguments::AngleBracketed(args) => args.args.iter().all(|arg| match arg { + syn::GenericArgument::Type(inner) => is_async_owned_type(inner), + _ => false, + }), + syn::PathArguments::Parenthesized(_) => false, + } + } + _ => false, + } +} + fn parse_name_arg(args: &Punctuated) -> Result { let Some(Meta::NameValue(name_value)) = args.first() else { return Err(Error::new( @@ -256,6 +321,88 @@ fn generate_vm_wrapper( }) } +fn generate_async_vm_wrapper( + item: &ItemFn, + wrapper_name: &syn::Ident, +) -> Result { + let impl_name = &item.sig.ident; + let mutable_wrapper_name = syn::Ident::new(&format!("{wrapper_name}_mut"), wrapper_name.span()); + let mut extract_stmts = Vec::::new(); + let mut call_args = Vec::::new(); + let mut arg_index = 0usize; + + for input in &item.sig.inputs { + let FnArg::Typed(pat_type) = input else { + return Err(Error::new_spanned(input, "methods are not supported")); + }; + let Pat::Ident(PatIdent { ident, .. }) = pat_type.pat.as_ref() else { + return Err(Error::new_spanned( + &pat_type.pat, + "callable parameters must use identifier patterns", + )); + }; + let ty = &pat_type.ty; + if is_host_context_param(input) { + extract_stmts.push(quote! { + let #ident = <#ty as super::CaptureAsyncHostContext>::capture(vm)?; + }); + call_args.push(quote!(#ident)); + continue; + } + let label = LitStr::new( + &format!("{} {}", wrapper_name, ident), + proc_macro2::Span::call_site(), + ); + let index = syn::Index::from(arg_index); + extract_stmts.push(quote! { + let #ident = super::borrow_arg::<#ty>(args, #index, #label)?; + }); + call_args.push(quote!(#ident)); + arg_index += 1; + } + + let await_value = if return_is_vm_result(&item.sig.output) { + quote!(#impl_name(#(#call_args),*).await?) + } else { + quote!(#impl_name(#(#call_args),*).await) + }; + let body = quote! { + #(#extract_stmts)* + vm.submit_host_future(Box::pin(async move { + let value = #await_value; + match super::IntoHostCallOutcome::into_host_call_outcome(value) { + super::CallOutcome::Return(values) => Ok(values), + super::CallOutcome::Pending(op_id) => Err(super::VmError::HostError( + format!("async host function returned nested pending operation {op_id}"), + )), + super::CallOutcome::Halt | super::CallOutcome::Yield => Err( + super::VmError::HostError( + "async host function returned a control-flow outcome".to_string(), + ), + ), + } + })) + }; + + Ok(quote! { + #[allow(dead_code)] + pub(super) fn #wrapper_name( + vm: &mut super::super::Vm, + args: &[super::super::Value], + ) -> super::super::VmResult { + #body + } + + #[allow(dead_code)] + pub(super) fn #mutable_wrapper_name( + vm: &mut super::super::Vm, + args: &mut [super::super::Value], + ) -> super::super::VmResult { + #body + } + }) +} + fn wrapper_and_impl_names(name: &syn::Ident) -> (syn::Ident, syn::Ident) { let original = name.to_string(); match original.strip_suffix("_impl") { @@ -502,7 +649,7 @@ fn uses_taken_extractor(ty: &Type) -> bool { #[cfg(test)] mod tests { - use super::{expand_pd_host_function, uses_edge_host_contract}; + use super::expand_pd_host_function; use syn::{ItemFn, Meta, Token, parse_quote, punctuated::Punctuated}; #[test] @@ -555,24 +702,41 @@ mod tests { } #[test] - fn native_async_signature_selects_edge_contract() { + fn ordinary_async_signature_generates_host_driven_future_submission() { let attr: Punctuated = parse_quote!(name = "test::async_call"); - let item: ItemFn = parse_quote! { - async fn async_call() -> VmResult> { - todo!() + let item: ItemFn = parse_quote!( + /// Returns an owned string asynchronously. + async fn async_call( + #[pd_host_context] context: TestContext, + value: String, + ) -> VmResult { + context.run(value).await } - }; - assert!(uses_edge_host_contract(&attr, &item)); + ); + let expanded = expand_pd_host_function(attr, item) + .expect("ordinary owned async function should use the generic async host contract") + .to_string(); + assert!(expanded.contains("submit_host_future")); + assert!(expanded.contains("async move")); + assert!(expanded.contains("borrow_arg")); + assert!(expanded.contains("CaptureAsyncHostContext")); + assert!(!expanded.contains("pd_host_context")); } #[test] - fn name_expression_alone_does_not_select_edge_contract() { - let attr: Punctuated = parse_quote!(name = NAME_PATH); + fn async_signature_rejects_borrowed_parameters() { + let attr: Punctuated = parse_quote!(name = "test::borrowed"); let item: ItemFn = parse_quote! { - fn sync_call() -> VmResult> { - todo!() + async fn borrowed(value: &str) -> VmResult { + Ok(value.to_string()) } }; - assert!(!uses_edge_host_contract(&attr, &item)); + + let error = expand_pd_host_function(attr, item).expect_err("borrow should be rejected"); + assert!( + error + .to_string() + .contains("parameters must be owned and 'static") + ); } } diff --git a/src/builtins/runtime/http.rs b/src/builtins/runtime/http.rs index c39ef87..9f11e89 100644 --- a/src/builtins/runtime/http.rs +++ b/src/builtins/runtime/http.rs @@ -1,24 +1,19 @@ -use std::task::{Context, Poll}; - -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] use futures_util::StreamExt; -#[cfg(feature = "http-client")] -use futures_util::future::{AbortHandle, Abortable}; +#[cfg(feature = "async")] use pd_host_function::pd_host_function; -use super::{HostCallResult, Vm, VmMap, VmResult}; -#[cfg(feature = "http-client")] -use crate::builtins::runtime::cancellation::{ - CancellationReason, CancellationToken, OperationId, OperationOwner, -}; -#[cfg(feature = "http-client")] -use crate::builtins::runtime::error::{RuntimeError, RuntimeErrorCode}; -#[cfg(feature = "http-client")] -use crate::builtins::runtime::resource::ResourceTypeId; -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] +use super::{Vm, VmMap, VmResult}; +#[cfg(feature = "async")] +use crate::builtins::runtime::cancellation::{CancellationReason, CancellationToken}; +#[cfg(feature = "async")] +use crate::vm::CaptureAsyncHostContext; +#[cfg(feature = "async")] use crate::vm::Value; -use crate::vm::{CallReturn, HostOpId, VmError}; +#[cfg(feature = "async")] +use crate::vm::VmError; #[derive(Clone, Debug, PartialEq, Eq)] pub struct HttpConfig { @@ -49,18 +44,8 @@ impl Default for HttpConfig { } } -#[cfg(feature = "http-client")] -struct HttpCompletion { - result: VmResult, -} - -#[cfg(feature = "http-client")] -struct HttpRequestResource { - receiver: futures_channel::oneshot::Receiver, -} - pub(crate) struct HttpState { - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] config: Option, pub(crate) max_in_flight: usize, } @@ -68,7 +53,7 @@ pub(crate) struct HttpState { impl Default for HttpState { fn default() -> Self { Self { - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] config: None, max_in_flight: crate::builtins::runtime::cancellation::DEFAULT_MAX_PENDING_OPERATIONS, } @@ -79,157 +64,42 @@ impl HttpState { pub(crate) fn reset_for_reuse(&mut self) {} pub(crate) fn configure(&mut self, config: HttpConfig) { - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] { self.config = Some(config); } - #[cfg(not(feature = "http-client"))] + #[cfg(not(feature = "async"))] let _ = config; } pub(crate) fn clear_configuration(&mut self) { - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] { self.config = None; } } - #[cfg(all(test, feature = "http-client"))] + #[cfg(all(test, feature = "async"))] pub(crate) fn configuration(&self) -> Option<&HttpConfig> { self.config.as_ref() } pub(crate) fn is_configured(&self) -> bool { - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] { self.config.is_some() } - #[cfg(not(feature = "http-client"))] + #[cfg(not(feature = "async"))] false } } -#[cfg(feature = "http-client")] -fn schedule_request(vm: &mut Vm, config: HttpConfig, request: HttpRequest) -> VmResult { - let max_in_flight = vm.host.http_state.max_in_flight; - if vm - .host - .runtime_operations - .operations_by_owner(OperationOwner::Http) - .len() - >= max_in_flight - { - return Err(VmError::HostError(format!( - "HTTP in-flight request limit of {} has been reached", - max_in_flight - ))); - } - - let deadline = std::time::Instant::now() + config.request_timeout; - let (sender, receiver) = futures_channel::oneshot::channel(); - let (abort_handle, abort_registration) = AbortHandle::new_pair(); - let operation = vm - .host - .runtime_operations - .start_owned( - OperationOwner::Http, - Some(&vm.run_ctx.cancellation), - Some(deadline), - Some(Box::new(move |_| { - abort_handle.abort(); - Ok(()) - })), - ) - .map_err(runtime_host_error)?; - let operation_id = operation.id(); - let op_id = operation_id.raw(); - let token = operation.token(); - let worker_operation = operation.clone(); - let resource = match vm.host.runtime_resources.insert( - ResourceTypeId::HTTP_REQUEST, - HttpRequestResource { receiver }, - ) { - Ok(resource) => resource, - Err(error) => { - let _ = vm - .host - .runtime_operations - .cancel(operation_id, CancellationReason::ResourceClosed); - return Err(runtime_host_error(error)); - } - }; - operation.set_payload(resource); - - let thread_name = format!("rustscript-http-{op_id}"); - if let Err(error) = std::thread::Builder::new() - .name(thread_name) - .spawn(move || { - let result = match tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - { - Ok(runtime) => runtime.block_on(async move { - match Abortable::new( - execute_request(&config, &request, &token, deadline), - abort_registration, - ) - .await - { - Ok(result) => result, - Err(_) => cancellation_error(&token), - } - }), - Err(error) => Err(VmError::HostError(format!( - "failed to create HTTP runtime: {error}" - ))), - }; - match &result { - Ok(_) => { - let _ = worker_operation.complete(); - } - Err(error) => { - let _ = worker_operation.fail( - RuntimeError::new( - RuntimeErrorCode::OperationFailed, - "http::request", - error.to_string(), - ) - .with_value(op_id), - ); - } - } - let _ = sender.send(HttpCompletion { result }); - }) - { - super::cancel_runtime_operation(vm, operation_id, CancellationReason::ResourceClosed); - return Err(VmError::HostError(format!( - "failed to start HTTP request: {error}" - ))); - } - - Ok(op_id) -} - -#[cfg(feature = "http-client")] -fn close_request_resource(vm: &mut Vm, op_id: HostOpId, reason: CancellationReason) { - let Ok(operation_id) = OperationId::from_raw(op_id) else { - return; - }; - let Ok(operation) = vm.host.runtime_operations.get(operation_id) else { - return; - }; - let Some(resource) = operation.payload() else { - return; - }; - let _ = super::close_runtime_resource(vm, resource, reason); -} - -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] fn runtime_host_error(error: impl std::fmt::Display) -> VmError { VmError::HostError(error.to_string()) } -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] fn cancellation_vm_error(token: &CancellationToken) -> VmError { token .check() @@ -237,97 +107,44 @@ fn cancellation_vm_error(token: &CancellationToken) -> VmError { .unwrap_or_else(runtime_host_error) } -#[cfg(feature = "http-client")] -fn cancellation_error(token: &CancellationToken) -> VmResult { - Err(cancellation_vm_error(token)) +#[cfg(feature = "async")] +pub(super) struct HttpRequestContext { + config: HttpConfig, + cancellation: CancellationToken, } -/// Starts an HTTP request under the VM's configured network policy. -/// -/// The request map accepts `method`, `url`, optional `headers`, and optional `body`. -/// The response map contains `status`, `headers`, `body`, and the final `url`. -#[pd_host_function(name = "http::client::request")] -pub(super) fn builtin_http_client_request( - vm: &mut Vm, - request: &VmMap, -) -> VmResult> { - #[cfg(not(feature = "http-client"))] - { - let _ = (vm, request); - Err(VmError::HostError( - "HTTP client support is disabled; enable the http-client feature".to_string(), - )) - } - - #[cfg(feature = "http-client")] - { +#[cfg(feature = "async")] +impl CaptureAsyncHostContext for HttpRequestContext { + fn capture(vm: &mut Vm) -> VmResult { let config = vm .host .http_state .config .clone() .ok_or_else(|| VmError::HostError("HTTP host is not configured".to_string()))?; - let request = parse_request(request, &config)?; - let op_id = schedule_request(vm, config, request)?; - Ok(HostCallResult::Pending(op_id)) + Ok(Self { + config, + cancellation: CancellationToken::root(), + }) } } -pub(super) fn poll_pending_op( - vm: &mut Vm, - op_id: HostOpId, - cx: &mut Context<'_>, -) -> Poll> { - #[cfg(feature = "http-client")] - { - use std::pin::Pin; - - let operation_id = match OperationId::from_raw(op_id) { - Ok(operation_id) => operation_id, - Err(error) => return Poll::Ready(Err(runtime_host_error(error))), - }; - let operation = match vm.host.runtime_operations.get(operation_id) { - Ok(operation) => operation, - Err(error) => return Poll::Ready(Err(runtime_host_error(error))), - }; - let Some(resource) = operation.payload() else { - return Poll::Ready(Err(VmError::HostError(format!( - "HTTP op {op_id} has no completion payload", - )))); - }; - let poll_result = match vm - .host - .runtime_resources - .get_mut::(resource, ResourceTypeId::HTTP_REQUEST) - { - Ok(request) => Pin::new(&mut request.receiver).poll(cx), - Err(error) => return Poll::Ready(Err(runtime_host_error(error))), - }; - match poll_result { - Poll::Pending => Poll::Pending, - Poll::Ready(Ok(completion)) => { - close_request_resource(vm, op_id, CancellationReason::ResourceClosed); - Poll::Ready(completion.result) - } - Poll::Ready(Err(_)) => { - close_request_resource(vm, op_id, CancellationReason::ResourceClosed); - Poll::Ready(Err(VmError::HostError(format!( - "HTTP op {op_id} was cancelled", - )))) - } - } - } - - #[cfg(not(feature = "http-client"))] - { - let _ = (vm, cx); - Poll::Ready(Err(VmError::HostError(format!( - "HTTP support is disabled for op {op_id}", - )))) - } +/// Starts an HTTP request under the VM's configured network policy. +/// +/// The request map accepts `method`, `url`, optional `headers`, and optional `body`. +/// The response map contains `status`, `headers`, `body`, and the final `url`. +#[cfg(feature = "async")] +#[pd_host_function(name = "http::client::request")] +pub(super) async fn builtin_http_client_request( + #[pd_host_context] context: HttpRequestContext, + request: VmMap, +) -> VmResult { + let request = parse_request(&request, &context.config)?; + let deadline = std::time::Instant::now() + context.config.request_timeout; + execute_request(&context.config, &request, &context.cancellation, deadline).await } -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] struct HttpRequest { method: reqwest::Method, url: url::Url, @@ -335,7 +152,7 @@ struct HttpRequest { body: Option>, } -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] fn parse_request(map: &VmMap, config: &HttpConfig) -> VmResult { let method = map_string(map, "method")?.to_ascii_uppercase(); if !matches!( @@ -413,7 +230,7 @@ fn parse_request(map: &VmMap, config: &HttpConfig) -> VmResult { }) } -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] fn map_string(map: &VmMap, key: &str) -> VmResult { match map.get(&Value::string(key)) { Some(Value::String(value)) => Ok(value.as_ref().clone()), @@ -424,7 +241,7 @@ fn map_string(map: &VmMap, key: &str) -> VmResult { } } -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] fn validate_url_policy<'a>(config: &HttpConfig, url: &'a url::Url) -> VmResult<(&'a str, u16)> { let scheme = url.scheme().to_ascii_lowercase(); if !config @@ -459,7 +276,7 @@ fn validate_url_policy<'a>(config: &HttpConfig, url: &'a url::Url) -> VmResult<( Ok((host, port)) } -#[cfg(all(feature = "http-client", test))] +#[cfg(all(feature = "async", test))] fn validate_url(config: &HttpConfig, url: &url::Url) -> VmResult> { let (host, port) = validate_url_policy(config, url)?; if config.allow_private_ips { @@ -480,7 +297,7 @@ fn validate_url(config: &HttpConfig, url: &url::Url) -> VmResult bool { match ip { std::net::IpAddr::V4(ip) => { @@ -575,13 +392,13 @@ fn is_restricted_ip(ip: std::net::IpAddr) -> bool { } } -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] async fn execute_request( config: &HttpConfig, request: &HttpRequest, token: &CancellationToken, deadline: std::time::Instant, -) -> VmResult { +) -> VmResult { token.check().map_err(runtime_host_error)?; let mut method = request.method.clone(); let mut url = request.url.clone(); @@ -716,9 +533,7 @@ async fn execute_request( (Value::string("body"), Value::bytes(bytes)), (Value::string("url"), Value::string(url.as_str())), ]); - return Ok(CallReturn::one(Value::Map(std::sync::Arc::new( - response_map, - )))); + return Ok(response_map); } Err(VmError::HostError( @@ -729,14 +544,15 @@ async fn execute_request( #[cfg(test)] mod tests { use super::HttpConfig; - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] use super::{ - CancellationReason, HttpRequest, HttpRequestResource, OperationOwner, ResourceTypeId, - execute_request, is_restricted_ip, schedule_request, validate_resolved_addresses, - validate_url, + CancellationReason, HttpRequest, VmMap, builtin_http_client_request, execute_request, + is_restricted_ip, validate_resolved_addresses, validate_url, + }; + #[cfg(feature = "async")] + use crate::vm::{ + CallOutcome, CallReturn, HostAsyncBridge, HostFuture, HostOpId, Value, VmResult, }; - #[cfg(feature = "http-client")] - use crate::builtins::runtime::cancellation::OperationId; #[test] fn default_http_policy_denies_all_hosts() { @@ -747,64 +563,57 @@ mod tests { assert!(!config.allow_private_ips); } - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] #[test] - fn request_uses_shared_operation_and_resource_lifecycle() { + fn request_submits_future_to_host_driver_without_runtime_operation() { + use std::sync::{Arc, Mutex}; + use std::task::{Context, Poll}; + + struct RecordingBridge { + submitted: Arc>>, + } + + impl HostAsyncBridge for RecordingBridge { + fn submit_op(&mut self, op_id: HostOpId, future: HostFuture) -> VmResult<()> { + *self.submitted.lock().expect("submission lock") = Some((op_id, future)); + Ok(()) + } + + fn poll_op( + &mut self, + _op_id: HostOpId, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Pending + } + } + + let submitted = Arc::new(Mutex::new(None)); let mut vm = crate::vm::Vm::new(crate::vm::Program::new(Vec::new(), Vec::new())); - vm.set_http_max_in_flight(1); - let config = HttpConfig { - allowed_schemes: vec!["http".to_string()], - allowed_hosts: vec!["127.0.0.1".to_string()], - allowed_ports: vec![1], - allow_private_ips: true, - ..HttpConfig::default() + vm.configure_http(HttpConfig::default()); + vm.set_async_bridge(Box::new(RecordingBridge { + submitted: Arc::clone(&submitted), + })); + let args = [Value::Map(Arc::new(VmMap::default()))]; + + let outcome = builtin_http_client_request(&mut vm, &args) + .expect("HTTP async host call should submit"); + let CallOutcome::Pending(op_id) = outcome else { + panic!("HTTP async host call should suspend"); }; - let request = HttpRequest { - method: reqwest::Method::GET, - url: "http://127.0.0.1:1/".parse().expect("valid URL"), - headers: Vec::new(), - body: None, - }; - - let op_id = schedule_request(&mut vm, config, request).expect("request should schedule"); - let operation_id = OperationId::from_raw(op_id).expect("operation id should be valid"); + assert_eq!(op_id, 1); assert_eq!( - vm.host - .runtime_operations - .get(operation_id) - .expect("operation should be registered") - .owner(), - OperationOwner::Http - ); - let operation = vm - .host - .runtime_operations - .get(operation_id) - .expect("request should remain registered"); - let resource = operation - .payload() - .expect("operation should reference the request resource"); - assert_eq!(resource.resource_type(), ResourceTypeId::HTTP_REQUEST); - assert!( - vm.host - .runtime_resources - .get::(resource, ResourceTypeId::HTTP_REQUEST) - .is_ok() - ); - - let token = operation.token(); - vm.clear_http_configuration(); - assert_eq!(token.reason(), Some(CancellationReason::Requested)); - assert!( - vm.host - .runtime_resources - .get::(resource, ResourceTypeId::HTTP_REQUEST) - .is_err() + submitted + .lock() + .expect("submission lock") + .as_ref() + .map(|(submitted_id, _)| *submitted_id), + Some(op_id) ); - assert!(vm.host.runtime_operations.get(operation_id).is_err()); + assert_eq!(vm.host.runtime_operations.active_count(), 0); } - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] #[test] fn production_request_timeout_sets_structured_deadline_reason() { use std::time::{Duration, Instant}; @@ -848,7 +657,7 @@ mod tests { server.join().expect("server should exit"); } - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] #[test] fn response_body_timeout_sets_structured_deadline_reason() { use std::io::{Read, Write}; @@ -901,7 +710,7 @@ mod tests { server.join().expect("server should exit"); } - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] #[test] fn empty_port_allowlist_rejects_explicit_and_default_ports() { let config = HttpConfig { @@ -915,7 +724,7 @@ mod tests { assert!(validate_url(&config, &default_port).is_err()); } - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] #[test] fn special_use_networks_and_mixed_dns_answers_are_restricted() { for address in [ @@ -959,7 +768,7 @@ mod tests { assert!(validate_resolved_addresses(&config, &addresses).is_err()); } - #[cfg(feature = "http-client")] + #[cfg(feature = "async")] #[test] fn ipv4_mapped_ipv6_loopback_is_restricted() { assert!(is_restricted_ip( diff --git a/src/builtins/runtime/mod.rs b/src/builtins/runtime/mod.rs index a79cb4f..40ead3e 100644 --- a/src/builtins/runtime/mod.rs +++ b/src/builtins/runtime/mod.rs @@ -4,6 +4,8 @@ use std::task::{Context, Poll}; use crate::builtins::BuiltinFunction; use crate::vm::{CallOutcome, CallReturn, HostOpId, Value, Vm, VmResult}; +#[cfg(feature = "async")] +use crate::vm::{CaptureAsyncHostContext, VmError}; use self::cancellation::{CancellationReason, OperationId, OperationOwner, OperationState}; use self::error::{RuntimeError, RuntimeErrorCode}; @@ -15,7 +17,6 @@ type RuntimeOperationPoller = fn(&mut Vm, HostOpId, &mut Context<'_>) -> Poll FromVmValue<'a> for &'a VmMap { } } +impl FromVmValue<'_> for VmMap { + fn from_vm_value(value: &Value, _label: &str) -> VmResult { + match value { + Value::Map(entries) => Ok(entries.as_ref().clone()), + _ => Err(VmError::TypeMismatch("map")), + } + } +} + impl FromVmValue<'_> for SharedArray { fn from_vm_value(value: &Value, _label: &str) -> VmResult { match value { diff --git a/src/vm/async_host/mod.rs b/src/vm/async_host/mod.rs new file mode 100644 index 0000000..b81c508 --- /dev/null +++ b/src/vm/async_host/mod.rs @@ -0,0 +1,266 @@ +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll, Wake, Waker}; + +use super::*; + +pub type HostFuture = Pin> + Send + 'static>>; + +pub trait CaptureAsyncHostContext: Send + 'static + Sized { + fn capture(vm: &mut Vm) -> VmResult; +} + +pub trait HostAsyncBridge: Send { + fn submit_op(&mut self, _op_id: HostOpId, _future: HostFuture) -> VmResult<()> { + Err(VmError::HostError( + "async host bridge does not accept submitted futures".to_string(), + )) + } + + fn poll_op(&mut self, op_id: HostOpId, cx: &mut Context<'_>) -> Poll>; + + fn cancel_op(&mut self, _op_id: HostOpId) {} + + fn cancel_op_with_reason(&mut self, op_id: HostOpId, _reason: CancellationReason) { + self.cancel_op(op_id); + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct WaitingHostOp { + pub(super) op_id: HostOpId, +} + +struct NoopWake; + +impl Wake for NoopWake { + fn wake(self: Arc) {} +} + +fn noop_waker() -> Waker { + Waker::from(Arc::new(NoopWake)) +} + +impl Vm { + pub fn set_async_bridge(&mut self, bridge: Box) { + self.cancel_waiting_host_op(); + self.host.async_bridge = Some(bridge); + } + + pub fn clear_async_bridge(&mut self) { + self.cancel_waiting_host_op(); + self.host.async_bridge = None; + } + + pub fn allocate_host_op_id(&mut self) -> HostOpId { + self.host + .runtime_operations + .allocate_id() + .expect("host operation id space should not be exhausted") + .raw() + } + + pub fn submit_host_future(&mut self, future: HostFuture) -> VmResult { + let op_id = self.allocate_host_op_id(); + let bridge = self.host.async_bridge.as_mut().ok_or_else(|| { + VmError::HostError("async host function requires a host async bridge".to_string()) + })?; + bridge.submit_op(op_id, future)?; + Ok(CallOutcome::Pending(op_id)) + } + + pub fn waiting_host_op_id(&self) -> Option { + self.instance.waiting_host_op.map(|op| op.op_id) + } + + pub fn cancel_waiting_host_op(&mut self) { + self.cancel_waiting_host_op_with_reason( + crate::builtins::runtime::cancellation::CancellationReason::Requested, + ); + } + + pub(crate) fn cancel_waiting_host_op_with_reason( + &mut self, + reason: crate::builtins::runtime::cancellation::CancellationReason, + ) { + let Some(waiting) = self.instance.waiting_host_op.take() else { + return; + }; + let Ok(operation_id) = + crate::builtins::runtime::cancellation::OperationId::from_raw(waiting.op_id) + else { + return; + }; + let owner = self + .host + .runtime_operations + .get(operation_id) + .ok() + .map(|operation| operation.owner()); + if owner == Some(crate::builtins::runtime::cancellation::OperationOwner::HostBridge) { + if let Some(bridge) = self.host.async_bridge.as_mut() { + bridge.cancel_op_with_reason(waiting.op_id, reason); + } + let _ = self.host.runtime_operations.cancel(operation_id, reason); + } else { + crate::builtins::runtime::cancel_builtin_io_op_with_reason(self, waiting.op_id, reason); + } + } + + pub fn complete_host_op( + &mut self, + op_id: HostOpId, + values: impl Into, + ) -> VmResult<()> { + let waiting = self.instance.waiting_host_op.ok_or_else(|| { + VmError::HostError(format!( + "host op {op_id} completed but vm is not waiting on any op", + )) + })?; + if waiting.op_id != op_id { + return Err(VmError::HostError(format!( + "host op {op_id} completed while vm waits on {}", + waiting.op_id + ))); + } + let operation_id = crate::builtins::runtime::cancellation::OperationId::from_raw(op_id) + .map_err(|error| VmError::HostError(error.to_string()))?; + let operation = self + .host + .runtime_operations + .get(operation_id) + .map_err(|error| VmError::HostError(error.to_string()))?; + if operation.owner() != crate::builtins::runtime::cancellation::OperationOwner::HostBridge { + return Err(VmError::HostError(format!( + "host bridge cannot complete runtime-owned operation {op_id}", + ))); + } + self.host + .runtime_operations + .complete(operation_id) + .map_err(|error| VmError::HostError(error.to_string()))?; + self.complete_waiting_host_op(op_id, values.into()) + } + + pub fn poll_waiting_host_op(&mut self, cx: &mut Context<'_>) -> Poll> { + let Some(waiting) = self.instance.waiting_host_op else { + return Poll::Ready(Ok(())); + }; + let operation_id = + match crate::builtins::runtime::cancellation::OperationId::from_raw(waiting.op_id) { + Ok(operation_id) => operation_id, + Err(error) => return Poll::Ready(Err(VmError::HostError(error.to_string()))), + }; + let operation = match self.host.runtime_operations.get(operation_id) { + Ok(operation) => operation, + Err(error) => return Poll::Ready(Err(VmError::HostError(error.to_string()))), + }; + let host_bridge_owned = + operation.owner() == crate::builtins::runtime::cancellation::OperationOwner::HostBridge; + + let poll_result = if host_bridge_owned { + let bridge_ptr = match self.host.async_bridge.as_mut() { + Some(bridge) => bridge.as_mut() as *mut dyn HostAsyncBridge, + None => { + return Poll::Ready(Err(VmError::HostError(format!( + "vm waiting on host op {} without an async bridge", + waiting.op_id + )))); + } + }; + unsafe { (&mut *bridge_ptr).poll_op(waiting.op_id, cx) } + } else { + crate::builtins::runtime::poll_builtin_io_op(self, waiting.op_id, cx) + }; + + match poll_result { + Poll::Pending => Poll::Pending, + Poll::Ready(Ok(values)) => { + if host_bridge_owned { + self.host + .runtime_operations + .complete(operation_id) + .map_err(|error| VmError::HostError(error.to_string()))?; + } + self.complete_waiting_host_op(waiting.op_id, values)?; + Poll::Ready(Ok(())) + } + Poll::Ready(Err(err)) => { + if host_bridge_owned { + let runtime_error = crate::builtins::runtime::error::RuntimeError::new( + crate::builtins::runtime::error::RuntimeErrorCode::OperationFailed, + "runtime::host_bridge", + err.to_string(), + ) + .with_value(waiting.op_id); + let _ = self + .host + .runtime_operations + .fail(operation_id, runtime_error); + } + self.instance.waiting_host_op = None; + Poll::Ready(Err(err)) + } + } + } + + pub async fn await_waiting_host_op(&mut self) -> VmResult<()> { + std::future::poll_fn(|cx| self.poll_waiting_host_op(cx)).await + } + + pub fn wait_for_host_op_blocking(&mut self) -> VmResult<()> { + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + loop { + match self.poll_waiting_host_op(&mut cx) { + Poll::Ready(result) => return result, + Poll::Pending => { + #[cfg(not(target_arch = "wasm32"))] + { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + #[cfg(target_arch = "wasm32")] + { + return Err(VmError::HostError( + "blocking host-op wait is unsupported on wasm32 runtime".to_string(), + )); + } + } + } + } + } + + pub fn wait_for_host_op_blocking_with_cancel(&mut self, mut should_cancel: F) -> VmResult<()> + where + F: FnMut() -> bool, + { + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + loop { + if should_cancel() { + let cancellation_result = self + .run_ctx + .cancel(crate::builtins::runtime::cancellation::CancellationReason::Requested); + self.cancel_waiting_host_op(); + cancellation_result?; + return Err(VmError::HostError("host operation cancelled".to_string())); + } + match self.poll_waiting_host_op(&mut cx) { + Poll::Ready(result) => return result, + Poll::Pending => { + #[cfg(not(target_arch = "wasm32"))] + { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + #[cfg(target_arch = "wasm32")] + { + return Err(VmError::HostError( + "blocking host-op wait is unsupported on wasm32 runtime".to_string(), + )); + } + } + } + } + } +} diff --git a/src/vm/host.rs b/src/vm/host.rs index 3af0379..2a5e55e 100644 --- a/src/vm/host.rs +++ b/src/vm/host.rs @@ -1,9 +1,9 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, OnceLock, RwLock}; -use std::task::{Context, Poll, Wake, Waker}; use crate::builtins::BuiltinFunction; +use super::async_host::WaitingHostOp; use super::*; pub type HostOpId = u64; @@ -86,16 +86,6 @@ pub trait HostArgsFunction: Send { fn call(&mut self, args: &[Value]) -> VmResult; } -pub trait HostAsyncBridge: Send { - fn poll_op(&mut self, op_id: HostOpId, cx: &mut Context<'_>) -> Poll>; - - fn cancel_op(&mut self, _op_id: HostOpId) {} - - fn cancel_op_with_reason(&mut self, op_id: HostOpId, _reason: CancellationReason) { - self.cancel_op(op_id); - } -} - pub type StaticHostFunction = fn(&mut Vm, &[Value]) -> VmResult; pub type StaticHostStackFunction = fn(&mut Vm, &[Value]) -> VmResult; pub type StaticHostArgsFunction = fn(&[Value]) -> VmResult; @@ -244,6 +234,7 @@ impl HostFunctionRegistry { self.plan_cache = Arc::new(RwLock::new(HashMap::new())); } + #[allow(dead_code)] pub(crate) fn mark_runtime_owned_pending(&mut self, name: &str) { let slot = self .by_name @@ -784,21 +775,6 @@ pub(crate) fn validate_non_yielding_host_value( Err(VmError::TypeMismatch(expected)) } -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(super) struct WaitingHostOp { - pub(super) op_id: HostOpId, -} - -struct NoopWake; - -impl Wake for NoopWake { - fn wake(self: Arc) {} -} - -fn noop_waker() -> Waker { - Waker::from(Arc::new(NoopWake)) -} - #[inline] fn builtin_for_binding_name(name: &str) -> Option { if !name.contains("::") { @@ -894,6 +870,7 @@ impl Vm { } } + #[allow(dead_code)] pub(crate) fn mark_runtime_owned_pending_binding(&mut self, name: &str) { let slot = builtin_for_binding_name(name) .and_then(|builtin| { @@ -1124,16 +1101,6 @@ impl Vm { .insert(builtin_call_index, host_slot); } - pub fn set_async_bridge(&mut self, bridge: Box) { - self.cancel_waiting_host_op(); - self.host.async_bridge = Some(bridge); - } - - pub fn clear_async_bridge(&mut self) { - self.cancel_waiting_host_op(); - self.host.async_bridge = None; - } - pub fn set_runtime_print_sink(&mut self, sink: F) where F: FnMut(String) + Send + 'static, @@ -1277,208 +1244,6 @@ impl Vm { Ok(()) } - pub fn allocate_host_op_id(&mut self) -> HostOpId { - self.host - .runtime_operations - .allocate_id() - .expect("host operation id space should not be exhausted") - .raw() - } - - pub fn waiting_host_op_id(&self) -> Option { - self.instance.waiting_host_op.map(|op| op.op_id) - } - - pub fn cancel_waiting_host_op(&mut self) { - self.cancel_waiting_host_op_with_reason( - crate::builtins::runtime::cancellation::CancellationReason::Requested, - ); - } - - pub(crate) fn cancel_waiting_host_op_with_reason( - &mut self, - reason: crate::builtins::runtime::cancellation::CancellationReason, - ) { - let Some(waiting) = self.instance.waiting_host_op.take() else { - return; - }; - let Ok(operation_id) = - crate::builtins::runtime::cancellation::OperationId::from_raw(waiting.op_id) - else { - return; - }; - let owner = self - .host - .runtime_operations - .get(operation_id) - .ok() - .map(|operation| operation.owner()); - if owner == Some(crate::builtins::runtime::cancellation::OperationOwner::HostBridge) { - if let Some(bridge) = self.host.async_bridge.as_mut() { - bridge.cancel_op_with_reason(waiting.op_id, reason); - } - let _ = self.host.runtime_operations.cancel(operation_id, reason); - } else { - crate::builtins::runtime::cancel_builtin_io_op_with_reason(self, waiting.op_id, reason); - } - } - - pub fn complete_host_op( - &mut self, - op_id: HostOpId, - values: impl Into, - ) -> VmResult<()> { - let waiting = self.instance.waiting_host_op.ok_or_else(|| { - VmError::HostError(format!( - "host op {op_id} completed but vm is not waiting on any op", - )) - })?; - if waiting.op_id != op_id { - return Err(VmError::HostError(format!( - "host op {op_id} completed while vm waits on {}", - waiting.op_id - ))); - } - let operation_id = crate::builtins::runtime::cancellation::OperationId::from_raw(op_id) - .map_err(|error| VmError::HostError(error.to_string()))?; - let operation = self - .host - .runtime_operations - .get(operation_id) - .map_err(|error| VmError::HostError(error.to_string()))?; - if operation.owner() != crate::builtins::runtime::cancellation::OperationOwner::HostBridge { - return Err(VmError::HostError(format!( - "host bridge cannot complete runtime-owned operation {op_id}", - ))); - } - self.host - .runtime_operations - .complete(operation_id) - .map_err(|error| VmError::HostError(error.to_string()))?; - self.complete_waiting_host_op(op_id, values.into()) - } - - pub fn poll_waiting_host_op(&mut self, cx: &mut Context<'_>) -> Poll> { - let Some(waiting) = self.instance.waiting_host_op else { - return Poll::Ready(Ok(())); - }; - let operation_id = - match crate::builtins::runtime::cancellation::OperationId::from_raw(waiting.op_id) { - Ok(operation_id) => operation_id, - Err(error) => return Poll::Ready(Err(VmError::HostError(error.to_string()))), - }; - let operation = match self.host.runtime_operations.get(operation_id) { - Ok(operation) => operation, - Err(error) => return Poll::Ready(Err(VmError::HostError(error.to_string()))), - }; - let host_bridge_owned = - operation.owner() == crate::builtins::runtime::cancellation::OperationOwner::HostBridge; - - let poll_result = if host_bridge_owned { - let bridge_ptr = match self.host.async_bridge.as_mut() { - Some(bridge) => bridge.as_mut() as *mut dyn HostAsyncBridge, - None => { - return Poll::Ready(Err(VmError::HostError(format!( - "vm waiting on host op {} without an async bridge", - waiting.op_id - )))); - } - }; - unsafe { (&mut *bridge_ptr).poll_op(waiting.op_id, cx) } - } else { - crate::builtins::runtime::poll_builtin_io_op(self, waiting.op_id, cx) - }; - - match poll_result { - Poll::Pending => Poll::Pending, - Poll::Ready(Ok(values)) => { - if host_bridge_owned { - self.host - .runtime_operations - .complete(operation_id) - .map_err(|error| VmError::HostError(error.to_string()))?; - } - self.complete_waiting_host_op(waiting.op_id, values)?; - Poll::Ready(Ok(())) - } - Poll::Ready(Err(err)) => { - if host_bridge_owned { - let runtime_error = crate::builtins::runtime::error::RuntimeError::new( - crate::builtins::runtime::error::RuntimeErrorCode::OperationFailed, - "runtime::host_bridge", - err.to_string(), - ) - .with_value(waiting.op_id); - let _ = self - .host - .runtime_operations - .fail(operation_id, runtime_error); - } - self.instance.waiting_host_op = None; - Poll::Ready(Err(err)) - } - } - } - - pub async fn await_waiting_host_op(&mut self) -> VmResult<()> { - std::future::poll_fn(|cx| self.poll_waiting_host_op(cx)).await - } - - pub fn wait_for_host_op_blocking(&mut self) -> VmResult<()> { - let waker = noop_waker(); - let mut cx = Context::from_waker(&waker); - loop { - match self.poll_waiting_host_op(&mut cx) { - Poll::Ready(result) => return result, - Poll::Pending => { - #[cfg(not(target_arch = "wasm32"))] - { - std::thread::sleep(std::time::Duration::from_millis(1)); - } - #[cfg(target_arch = "wasm32")] - { - return Err(VmError::HostError( - "blocking host-op wait is unsupported on wasm32 runtime".to_string(), - )); - } - } - } - } - } - - pub fn wait_for_host_op_blocking_with_cancel(&mut self, mut should_cancel: F) -> VmResult<()> - where - F: FnMut() -> bool, - { - let waker = noop_waker(); - let mut cx = Context::from_waker(&waker); - loop { - if should_cancel() { - let cancellation_result = self - .run_ctx - .cancel(crate::builtins::runtime::cancellation::CancellationReason::Requested); - self.cancel_waiting_host_op(); - cancellation_result?; - return Err(VmError::HostError("host operation cancelled".to_string())); - } - match self.poll_waiting_host_op(&mut cx) { - Poll::Ready(result) => return result, - Poll::Pending => { - #[cfg(not(target_arch = "wasm32"))] - { - std::thread::sleep(std::time::Duration::from_millis(1)); - } - #[cfg(target_arch = "wasm32")] - { - return Err(VmError::HostError( - "blocking host-op wait is unsupported on wasm32 runtime".to_string(), - )); - } - } - } - } - } - pub(super) fn execute_host_call( &mut self, index: u16, diff --git a/src/vm/host_runtime.rs b/src/vm/host_runtime.rs index e150396..34f2f76 100644 --- a/src/vm/host_runtime.rs +++ b/src/vm/host_runtime.rs @@ -22,7 +22,8 @@ use crate::builtins::runtime::resource::{DEFAULT_MAX_RESOURCES, ResourceArena}; use crate::vm::IoPolicy; #[cfg(feature = "sqlite")] use crate::vm::SqlitePolicy; -use crate::vm::host::{HostAsyncBridge, VmHostFunction}; +use crate::vm::async_host::HostAsyncBridge; +use crate::vm::host::VmHostFunction; /// Embedder-supplied print sink for `print`/`debug` output. pub(crate) type RuntimePrintSink = dyn FnMut(String) + Send; diff --git a/src/vm/instance.rs b/src/vm/instance.rs index baecdfb..475ff79 100644 --- a/src/vm/instance.rs +++ b/src/vm/instance.rs @@ -18,7 +18,7 @@ use std::sync::atomic::AtomicBool; use std::sync::{Arc, Weak}; use crate::bytecode::{CallableValue, Program, SharedCaptureCell, Value}; -use crate::vm::host::WaitingHostOp; +use crate::vm::async_host::WaitingHostOp; use crate::vm::map_iter::MapIteratorState; use crate::vm::{DEFAULT_MAX_SCRIPT_CALL_DEPTH, VmYieldReason}; diff --git a/src/vm/mod.rs b/src/vm/mod.rs index 3b9fd23..37cd643 100644 --- a/src/vm/mod.rs +++ b/src/vm/mod.rs @@ -4,6 +4,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; pub(crate) mod aot; +mod async_host; mod capability; pub mod diagnostics; mod engine; @@ -22,13 +23,15 @@ mod superinstructions; #[cfg(test)] mod tests; pub use self::aot::AotArtifactError; + +pub use self::async_host::{CaptureAsyncHostContext, HostAsyncBridge, HostFuture}; pub use self::capability::{CapabilityProfile, CapabilityProfileBuilder, IoPolicy}; use self::engine::Engine; pub use self::epoch::{EpochCheckpoint, EpochHandle}; pub use self::fuel::FuelCheckpoint; pub use self::host::{ - CallOutcome, CallReturn, HostArgsFunction, HostAsyncBridge, HostBindingPlan, HostFunction, - HostFunctionRegistry, HostOpId, HostStackFunction, StaticHostArgsFunction, StaticHostFunction, + CallOutcome, CallReturn, HostArgsFunction, HostBindingPlan, HostFunction, HostFunctionRegistry, + HostOpId, HostStackFunction, StaticHostArgsFunction, StaticHostFunction, StaticHostStackFunction, }; use self::host::{HostCallExecOutcome, VmHostFunction}; diff --git a/src/vm/tests.rs b/src/vm/tests.rs index 6003416..8726ac3 100644 --- a/src/vm/tests.rs +++ b/src/vm/tests.rs @@ -1,4 +1,4 @@ -use super::host::WaitingHostOp; +use super::async_host::WaitingHostOp; use super::*; use crate::builtins::BuiltinFunction; use crate::bytecode::TypeMap; @@ -27,6 +27,7 @@ fn failed_dynamic_builtin_override_preserves_runtime_owned_pending_binding() { vm.ensure_call_bindings() .expect("default fallback should bind runtime sleep"); let slot = vm.host.host_function_symbols["runtime::sleep"]; + vm.host.runtime_owned_pending_host_slots.insert(slot); assert!(vm.host.runtime_owned_pending_host_slots.contains(&slot)); vm.bind_builtin_override("runtime::sleep", Box::new(Dummy)) @@ -47,6 +48,7 @@ fn failed_static_builtin_override_preserves_runtime_owned_pending_binding() { vm.ensure_call_bindings() .expect("default fallback should bind runtime sleep"); let slot = vm.host.host_function_symbols["runtime::sleep"]; + vm.host.runtime_owned_pending_host_slots.insert(slot); assert!(vm.host.runtime_owned_pending_host_slots.contains(&slot)); vm.bind_builtin_static_override("runtime::sleep", dummy) @@ -81,6 +83,67 @@ fn reset_for_reuse_keeps_host_operation_ids_monotonic() { assert_eq!(vm.allocate_host_op_id(), 2); } +#[test] +fn async_host_future_is_submitted_to_the_host_bridge() { + use std::sync::{Arc, Mutex}; + + struct RecordingBridge { + submitted: Arc>>, + future: Arc>>, + } + + impl HostAsyncBridge for RecordingBridge { + fn submit_op(&mut self, op_id: HostOpId, future: HostFuture) -> VmResult<()> { + self.submitted.lock().expect("submitted lock").push(op_id); + *self.future.lock().expect("future lock") = Some(future); + Ok(()) + } + + fn poll_op( + &mut self, + _op_id: HostOpId, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Pending + } + } + + let submitted = Arc::new(Mutex::new(Vec::new())); + let future = Arc::new(Mutex::new(None)); + let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); + vm.set_async_bridge(Box::new(RecordingBridge { + submitted: Arc::clone(&submitted), + future: Arc::clone(&future), + })); + + let outcome = vm + .submit_host_future(Box::pin(async { Ok(CallReturn::one(Value::Int(42))) })) + .expect("host bridge should accept future"); + let CallOutcome::Pending(op_id) = outcome else { + panic!("async host submission should suspend"); + }; + + assert_eq!(*submitted.lock().expect("submitted lock"), vec![op_id]); + assert!(future.lock().expect("future lock").is_some()); + assert_eq!(vm.host.runtime_operations.active_count(), 0); +} + +#[test] +fn async_host_submission_without_driver_fails_and_retires_the_id() { + let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); + let error = vm + .submit_host_future(Box::pin(async { Ok(CallReturn::none()) })) + .expect_err("missing host async driver should fail"); + + assert!( + error + .to_string() + .contains("async host function requires a host async bridge") + ); + assert_eq!(vm.allocate_host_op_id(), 2); + assert_eq!(vm.host.runtime_operations.active_count(), 0); +} + #[test] fn unused_host_operation_ids_do_not_consume_registry_capacity() { let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); @@ -90,7 +153,7 @@ fn unused_host_operation_ids_do_not_consume_registry_capacity() { assert_eq!(vm.host.runtime_operations.active_count(), 0); } -#[cfg(feature = "http-client")] +#[cfg(feature = "async")] #[test] fn capability_profile_binding_installs_http_policy() { let policy = crate::builtins::runtime::HttpConfig { diff --git a/tests/host_binding_generation_tests.rs b/tests/host_binding_generation_tests.rs index ec15acd..b663f10 100644 --- a/tests/host_binding_generation_tests.rs +++ b/tests/host_binding_generation_tests.rs @@ -139,6 +139,18 @@ fn infers_host_suspension_from_the_return_signature() { ); } + let asynchronous = parse_quote!( + async fn host(value: String) -> VmResult {} + ); + assert_eq!( + infer_host_execution(&asynchronous), + HostExecutionKind::MaySuspend + ); + assert_eq!( + classify_host_binding(&asynchronous), + HostBindingKind::StaticStack + ); + let synchronous = parse_quote!( fn host() -> VmResult {} );