feat(io): split blocking and host-driven async backends

Author: fffonion

Cargo.lock

@@ -67,7 +67,7 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"
 rusqlite = { version = "0.32", default-features = false, features = ["bundled", "hooks", "limits"], optional = true }
 url = { version = "2", optional = true }
 futures-util = { version = "0.3", optional = true }
-tokio = { version = "1", features = ["rt-multi-thread", "net", "time", "sync"], optional = true }
+tokio = { version = "1", features = ["rt-multi-thread", "net", "time", "sync", "fs", "io-util", "process"], optional = true }
 edge_abi = { package = "pd-edge-abi", version = "0.1.1", default-features = false, optional = true }
 futures-channel = "0.3"
 paste = "1"

Cargo.toml

@@ -251,10 +251,21 @@ fn write_generated_file(path: &Path, contents: &str) {
 fn builtin_source_specs(namespaces: &[NamespaceDecl]) -> Vec<SourceSpec> {
     namespaces
         .iter()
-        .map(|namespace| SourceSpec {
-            path: format!("src/builtins/runtime/{}.rs", namespace.module),
-            module: namespace.module.clone(),
-            category: SourceCategory::NamespacedBuiltin,
+        .map(|namespace| {
+            let path = if namespace.module == "io" {
+                if cfg!(feature = "async") {
+                    "src/builtins/runtime/io/async_io.rs".to_string()
+                } else {
+                    "src/builtins/runtime/io/blocking.rs".to_string()
+                }
+            } else {
+                format!("src/builtins/runtime/{}.rs", namespace.module)
+            };
+            SourceSpec {
+                path,
+                module: namespace.module.clone(),
+                category: SourceCategory::NamespacedBuiltin,
+            }
         })
         .collect()
 }
@@ -2113,7 +2124,8 @@ fn type_label(ty: &Type) -> String {
                     };
                     format!("{} | null", type_label(inner))
                 }
-                "VmResult" | "BuiltinResult" | "HostResult" | "HostCallResult" => {
+                "VmResult" | "BuiltinResult" | "HostResult" | "HostCallResult"
+                | "HostFutureOutput" => {
                     let syn::PathArguments::AngleBracketed(args) = &segment.arguments else {
                         panic!("{ident}<T> requires one generic argument");
                     };

build.rs

@@ -308,13 +308,13 @@ fn generate_vm_wrapper(
 
     Ok(quote! {
         #[allow(dead_code)]
-        pub(super) fn #wrapper_name(#(#imm_wrapper_params),*) -> #wrapper_output {
+        pub(crate) fn #wrapper_name(#(#imm_wrapper_params),*) -> #wrapper_output {
             #(#imm_extract_stmts)*
             #call_expr
         }
 
         #[allow(dead_code)]
-        pub(super) fn #mutable_wrapper_name(#(#mut_wrapper_params),*) -> #wrapper_output {
+        pub(crate) fn #mutable_wrapper_name(#(#mut_wrapper_params),*) -> #wrapper_output {
             #(#mut_extract_stmts)*
             #call_expr
         }
@@ -344,7 +344,7 @@ fn generate_async_vm_wrapper(
         let ty = &pat_type.ty;
         if is_host_context_param(input) {
             extract_stmts.push(quote! {
-                let #ident = <#ty as super::CaptureAsyncHostContext>::capture(vm)?;
+                let #ident = <#ty as super::CaptureAsyncHostContext>::capture_with_args(vm, args)?;
             });
             call_args.push(quote!(#ident));
             continue;
@@ -366,12 +366,14 @@ fn generate_async_vm_wrapper(
     } else {
         quote!(#impl_name(#(#call_args),*).await)
     };
-    let body = quote! {
-        #(#extract_stmts)*
-        vm.submit_host_future(Box::pin(async move {
-            let value = #await_value;
+    let future_result = if return_is_host_future_output(&item.sig.output) {
+        quote!(Ok(value.map(super::return_one)))
+    } else {
+        quote! {
             match super::IntoHostCallOutcome::into_host_call_outcome(value) {
-                super::CallOutcome::Return(values) => Ok(values),
+                super::CallOutcome::Return(values) => {
+                    Ok(super::HostFutureOutput::returning(values))
+                }
                 super::CallOutcome::Pending(op_id) => Err(super::VmError::HostError(
                     format!("async host function returned nested pending operation {op_id}"),
                 )),
@@ -381,20 +383,27 @@ fn generate_async_vm_wrapper(
                     ),
                 ),
             }
+        }
+    };
+    let body = quote! {
+        #(#extract_stmts)*
+        vm.submit_host_future(Box::pin(async move {
+            let value = #await_value;
+            #future_result
         }))
     };
 
     Ok(quote! {
         #[allow(dead_code)]
-        pub(super) fn #wrapper_name(
+        pub(crate) fn #wrapper_name(
             vm: &mut super::super::Vm,
             args: &[super::super::Value],
         ) -> super::super::VmResult<super::CallOutcome> {
             #body
         }
 
         #[allow(dead_code)]
-        pub(super) fn #mutable_wrapper_name(
+        pub(crate) fn #mutable_wrapper_name(
             vm: &mut super::super::Vm,
             args: &mut [super::super::Value],
         ) -> super::super::VmResult<super::CallOutcome> {
@@ -468,6 +477,20 @@ fn unwrap_vm_result_type(ty: &Type) -> Result<Option<Type>, Error> {
     }
 }
 
+fn return_is_host_future_output(output: &ReturnType) -> bool {
+    vm_result_inner_type(output)
+        .expect("pd_host_function return type should already be validated")
+        .and_then(|ty| match ty {
+            Type::Path(path) => path
+                .path
+                .segments
+                .last()
+                .map(|segment| segment.ident.clone()),
+            _ => None,
+        })
+        .is_some_and(|ident| ident == "HostFutureOutput")
+}
+
 fn return_is_vm_result(output: &ReturnType) -> bool {
     vm_result_inner_type(output)
         .expect("pd_host_function return type should already be validated")
@@ -529,7 +552,8 @@ fn type_label(ty: &Type) -> Result<String, Error> {
                     let inner_label = type_label(inner)?;
                     Ok(format!("{inner_label} | null"))
                 }
-                "VmResult" | "BuiltinResult" | "HostResult" | "HostCallResult" => {
+                "VmResult" | "BuiltinResult" | "HostResult" | "HostCallResult"
+                | "HostFutureOutput" => {
                     let syn::PathArguments::AngleBracketed(args) = &segment.arguments else {
                         return Err(Error::new_spanned(
                             &segment.arguments,
@@ -690,9 +714,26 @@ mod tests {
         assert!(expanded.contains("async move"));
         assert!(expanded.contains("borrow_arg"));
         assert!(expanded.contains("CaptureAsyncHostContext"));
+        assert!(expanded.contains("capture_with_args"));
         assert!(!expanded.contains("pd_host_context"));
     }
 
+    #[test]
+    fn async_host_future_output_maps_its_inner_value_to_call_return() {
+        let attr: Punctuated<Meta, Token![,]> = parse_quote!(name = "test::completion");
+        let item: ItemFn = parse_quote! {
+            /// Completes after mutating VM-owned state.
+            async fn completion() -> VmResult<HostFutureOutput<i64>> {
+                todo!()
+            }
+        };
+
+        let expanded = expand_pd_host_function(attr, item)
+            .expect("host future output should be accepted")
+            .to_string();
+        assert!(expanded.contains("value . map (super :: return_one)"));
+    }
+
     #[test]
     fn async_signature_rejects_borrowed_parameters() {
         let attr: Punctuated<Meta, Token![,]> = parse_quote!(name = "test::borrowed");

pd-host-function/src/lib.rs

@@ -410,6 +410,7 @@ impl OperationState {
         self.core.status()
     }
 
+    #[cfg_attr(feature = "async", allow(dead_code))]
     pub fn set_payload(&self, payload: ResourceHandle) {
         self.core
             .inner
@@ -453,6 +454,7 @@ impl OperationState {
             .payload
     }
 
+    #[cfg_attr(feature = "async", allow(dead_code))]
     pub fn set_resource(&self, resource: ResourceHandle) {
         self.core
             .inner
@@ -560,6 +562,7 @@ impl OperationRegistry {
         Ok(id)
     }
 
+    #[cfg_attr(feature = "async", allow(dead_code))]
     pub fn start_owned(
         &mut self,
         owner: OperationOwner,