Repository: denoland/deno_core
Pull request #110 (2023-08-01 14:30:57)

feat(ops): Add async(lazy) to avoid eager polling
`#[op2(async(lazy))]` prevents eager polling and will eventually allow for fast async ops. 

This is distinctly different than `deferred` which polls the op immediately and then defers that result.


Patch:
diff --git a/core/runtime/ops.rs b/core/runtime/ops.rs
index 5f8e2be..04b448d 100644
--- a/core/runtime/ops.rs
+++ b/core/runtime/ops.rs
@@ -153,6 +153,7 @@ pub fn queue_async_op<'s>(
 #[inline]
 pub fn map_async_op_infallible<R: 'static>(
   ctx: &OpCtx,
+  lazy: bool,
   promise_id: i32,
   op: impl Future<Output = R> + 'static,
   rv_map: for<'r> fn(
@@ -163,6 +164,22 @@ pub fn map_async_op_infallible<R: 'static>(
   let id = ctx.id;
   ctx.state.borrow().tracker.track_async(id);
 
+  if lazy {
+    let mapper = move |r| {
+      (
+        promise_id,
+        id,
+        OpResult::Op2Temp(Box::new(move |scope| rv_map(scope, r))),
+      )
+    };
+    ctx
+      .context_state
+      .borrow_mut()
+      .pending_ops
+      .spawn(op.map(mapper));
+    return None;
+  }
+
   // TODO(mmastrac): We have to poll every future here because that assumption is baked into a large number
   // of ops. If we can figure out a way around this, we can remove this call to boxed_local and save a malloc per future.
   let mut pinned = op.map(move |res| (promise_id, id, res)).boxed_local();
@@ -192,6 +209,7 @@ pub fn map_async_op_infallible<R: 'static>(
 #[inline]
 pub fn map_async_op_fallible<R: 'static, E: Into<Error> + 'static>(
   ctx: &OpCtx,
+  lazy: bool,
   promise_id: i32,
   op: impl Future<Output = Result<R, E>> + 'static,
   rv_map: for<'r> fn(
@@ -202,6 +220,25 @@ pub fn map_async_op_fallible<R: 'static, E: Into<Error> + 'static>(
   let id = ctx.id;
   ctx.state.borrow().tracker.track_async(id);
 
+  if lazy {
+    let get_class = RefCell::borrow(&ctx.state).get_error_class_fn;
+    ctx.context_state.borrow_mut().pending_ops.spawn(op.map(
+      move |r| match r {
+        Ok(v) => (
+          promise_id,
+          id,
+          OpResult::Op2Temp(Box::new(move |scope| rv_map(scope, v))),
+        ),
+        Err(err) => (
+          promise_id,
+          id,
+          OpResult::Err(OpError::new(get_class, err.into())),
+        ),
+      },
+    ));
+    return None;
+  }
+
   // TODO(mmastrac): We have to poll every future here because that assumption is baked into a large number
   // of ops. If we can figure out a way around this, we can remove this call to boxed_local and save a malloc per future.
   let mut pinned = op.map(move |res| (promise_id, id, res)).boxed_local();
@@ -645,6 +682,7 @@ mod tests {
       op_async_sleep,
       op_async_sleep_impl,
       op_async_sleep_error,
+      op_async_lazy_error,
       op_async_result_impl,
       op_async_state_rc,
       op_async_buffer,
@@ -1599,6 +1637,23 @@ mod tests {
     Ok(())
   }
 
+  #[op2(async(lazy), core)]
+  pub async fn op_async_lazy_error() -> Result<(), Error> {
+    bail!("whoops")
+  }
+
+  #[tokio::test]
+  pub async fn test_op_async_lazy_error(
+  ) -> Result<(), Box<dyn std::error::Error>> {
+    run_async_test(
+      5,
+      "op_async_lazy_error",
+      "try { await op_async_lazy_error(); assert(false) } catch (e) {{ assertErrorContains(e, 'whoops') }}",
+    )
+    .await?;
+    Ok(())
+  }
+
   /// Test exits from the three possible routes -- before future, future immediate,
   /// future polled failed, future polled success.
   #[op2(async, core)]
diff --git a/ops/op2/config.rs b/ops/op2/config.rs
new file mode 100644
index 0000000..6d0a966
--- /dev/null
+++ b/ops/op2/config.rs
@@ -0,0 +1,194 @@
+use deno_proc_macro_rules::rules;
+use proc_macro2::TokenStream;
+use proc_macro2::TokenTree;
+use quote::ToTokens;
+
+use crate::op2::Op2Error;
+
+#[derive(Debug, Default, Eq, PartialEq)]
+pub(crate) struct MacroConfig {
+  /// Use `(core)` for ops that live in `deno_core`.
+  pub core: bool,
+  /// Generate a fastcall method (must be fastcall compatible).
+  pub fast: bool,
+  /// Do not generate a fastcall method (must be fastcall compatible).
+  pub nofast: bool,
+  /// Marks an async function (either `async fn` or `fn -> impl Future`)
+  pub r#async: bool,
+  /// Marks an lazy async function (async must also be true)
+  pub async_lazy: bool,
+}
+
+impl MacroConfig {
+  fn from_token_trees(
+    flags: Vec<TokenTree>,
+    args: Vec<Option<Vec<TokenTree>>>,
+  ) -> Result<Self, Op2Error> {
+    let flags = flags.into_iter().zip(args.into_iter()).map(|(flag, args)| {
+      if let Some(args) = args {
+        let args = args
+          .into_iter()
+          .map(|arg| arg.to_string())
+          .collect::<Vec<_>>()
+          .join(",");
+        format!("{}({args})", flag.into_token_stream())
+      } else {
+        flag.into_token_stream().to_string()
+      }
+    });
+    Self::from_flags(flags)
+  }
+
+  /// Parses a list of string flags.
+  pub fn from_flags(
+    flags: impl IntoIterator<Item = String>,
+  ) -> Result<Self, Op2Error> {
+    let mut config: MacroConfig = Self::default();
+    let flags = flags.into_iter().collect::<Vec<_>>();
+
+    // Ensure that the flags are sorted in alphabetical order for consistency and searchability
+    let mut flags_sorted = flags.clone();
+    flags_sorted.sort();
+    if flags != flags_sorted {
+      return Err(Op2Error::ImproperlySortedAttribute(flags_sorted.join(", ")));
+    }
+
+    for flag in flags {
+      if flag == "core" {
+        config.core = true;
+      } else if flag == "fast" {
+        config.fast = true;
+      } else if flag == "nofast" {
+        config.nofast = true;
+      } else if flag == "async" {
+        config.r#async = true;
+      } else if flag == "async(lazy)" {
+        config.r#async = true;
+        config.async_lazy = true;
+      } else {
+        return Err(Op2Error::InvalidAttribute(flag));
+      }
+    }
+
+    // Test for invalid attribute combinations
+    if config.fast && config.nofast {
+      return Err(Op2Error::InvalidAttributeCombination("fast", "nofast"));
+    }
+    if config.fast && (config.r#async && !config.async_lazy) {
+      return Err(Op2Error::InvalidAttributeCombination("fast", "async"));
+    }
+    if config.nofast && (config.r#async && !config.async_lazy) {
+      return Err(Op2Error::InvalidAttributeCombination("nofast", "async"));
+    }
+
+    Ok(config)
+  }
+
+  /// Parses the attribute parameters that `proc_macro` passes to the macro. For `#[op2(arg1, arg2)]`, the
+  /// tokens will be `arg1, arg2`.
+  pub fn from_tokens(tokens: TokenStream) -> Result<Self, Op2Error> {
+    let attr_string = tokens.to_string();
+
+    let config = std::panic::catch_unwind(|| {
+      rules!(tokens => {
+        () => {
+          Ok(MacroConfig::default())
+        }
+        ( $($flags:tt $( ( $( $args:tt ),* ) )? ),+ ) => {
+          Self::from_token_trees(flags, args)
+        }
+      })
+    })
+    .map_err(|_| Op2Error::PatternMatchFailed("attribute", attr_string))??;
+    Ok(config)
+  }
+
+  #[cfg(test)]
+  fn from_token_tree(tokens: TokenTree) -> Result<Self, Op2Error> {
+    let attr_string = tokens.to_string();
+
+    let config = std::panic::catch_unwind(|| {
+      rules!(tokens.into_token_stream() => {
+        ( ( $($flags:tt $( ( $( $args:tt ),* ) )? ),+ ) ) => {
+          Self::from_token_trees(flags, args)
+        }
+      })
+    })
+    .map_err(|_| Op2Error::PatternMatchFailed("attribute", attr_string))??;
+    Ok(config)
+  }
+
+  /// If the attribute matches #[op2(...)] or #[op2], returns `Some(MacroConfig)`, otherwise returns `None`.
+  #[cfg(test)]
+  pub fn from_maybe_attribute_tokens(
+    tokens: TokenStream,
+  ) -> Result<Option<Self>, Op2Error> {
+    use syn2 as syn;
+    Ok(rules!(tokens => {
+      (#[op2]) => {
+        Some(MacroConfig::default())
+      }
+      (#[op2 $flags:tt ]) => {
+        Some(MacroConfig::from_token_tree(flags)?)
+      }
+      (#[$_attr:meta]) => {
+        None
+      }
+    }))
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use syn2::{ItemFn, Meta};
+
+  fn test_parse(s: &str, expected: MacroConfig) {
+    let item_fn = syn2::parse_str::<ItemFn>(&format!("#[op2{s}] fn x() {{ }}"))
+      .expect("Failed to parse function");
+    let attr = item_fn.attrs.get(0).unwrap();
+    let config =
+      MacroConfig::from_maybe_attribute_tokens(attr.into_token_stream())
+        .expect("Failed to parse attribute")
+        .expect("Attribute was None");
+    assert_eq!(expected, config);
+    if let Meta::List(list) = &attr.meta {
+      let config =
+        MacroConfig::from_tokens(list.tokens.clone().into_token_stream())
+          .expect("Failed to parse attribute");
+      assert_eq!(expected, config);
+    } else if let Meta::Path(..) = &attr.meta {
+      // Ignored
+    } else {
+      panic!("Not a list or path");
+    }
+  }
+
+  #[test]
+  fn test_macro_parse() {
+    test_parse("", MacroConfig::default());
+    test_parse(
+      "(async)",
+      MacroConfig {
+        r#async: true,
+        ..Default::default()
+      },
+    );
+    test_parse(
+      "(async(lazy))",
+      MacroConfig {
+        r#async: true,
+        async_lazy: true,
+        ..Default::default()
+      },
+    );
+    test_parse(
+      "(async, core)",
+      MacroConfig {
+        r#async: true,
+        core: true,
+        ..Default::default()
+      },
+    );
+  }
+}
diff --git a/ops/op2/dispatch_async.rs b/ops/op2/dispatch_async.rs
index dc202e4..c790729 100644
--- a/ops/op2/dispatch_async.rs
+++ b/ops/op2/dispatch_async.rs
@@ -1,4 +1,5 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+use super::config::MacroConfig;
 use super::dispatch_slow::call;
 use super::dispatch_slow::extract_arg;
 use super::dispatch_slow::from_arg;
@@ -20,6 +21,7 @@ use proc_macro2::TokenStream;
 use quote::quote;
 
 pub(crate) fn generate_dispatch_async(
+  config: &MacroConfig,
   generator_state: &mut GeneratorState,
   signature: &ParsedSignature,
 ) -> Result<TokenStream, V8MappingError> {
@@ -99,9 +101,10 @@ pub(crate) fn generate_dispatch_async(
     }));
   }
 
+  let lazy = config.async_lazy;
   output.extend(gs_quote!(generator_state(promise_id, fn_args, result, opctx, scope, deno_core) => {
     let #promise_id = #deno_core::_ops::to_i32_option(&#fn_args.get(0)).unwrap_or_default();
-    if let Some(#result) = #deno_core::_ops::#mapper(#opctx, #promise_id, #result, |#scope, #result| {
+    if let Some(#result) = #deno_core::_ops::#mapper(#opctx, #lazy, #promise_id, #result, |#scope, #result| {
       #return_value
     }) {
       // Eager poll returned a value
@@ -139,21 +142,16 @@ pub(crate) fn generate_dispatch_async(
     quote!()
   };
 
-  let GeneratorState {
-    deno_core,
-    info,
-    slow_function,
-    ..
-  } = &generator_state;
-
-  Ok(quote! {
-    extern "C" fn #slow_function(#info: *const #deno_core::v8::FunctionCallbackInfo) {
-    #with_scope
-    #with_retval
-    #with_args
-    #with_opctx
-    #with_opstate
-
-    #output
-  }})
+  Ok(
+    gs_quote!(generator_state(deno_core, info, slow_function) => {
+      extern "C" fn #slow_function(#info: *const #deno_core::v8::FunctionCallbackInfo) {
+      #with_scope
+      #with_retval
+      #with_args
+      #with_opctx
+      #with_opstate
+
+      #output
+    }}),
+  )
 }
diff --git a/ops/op2/dispatch_shared.rs b/ops/op2/dispatch_shared.rs
index 0177a2f..d77c82a 100644
--- a/ops/op2/dispatch_shared.rs
+++ b/ops/op2/dispatch_shared.rs
@@ -8,7 +8,7 @@ use proc_macro2::TokenStream;
 use quote::format_ident;
 use quote::quote;
 
-/// Given an [`Arg`] containing a V8 value, converts this value to itself final form.
+/// Given an [`Arg`] containing a V8 value, converts this value to its final argument form.
 pub fn v8_intermediate_to_arg(i: &Ident, arg: &Arg) -> TokenStream {
   let arg = match arg {
     Arg::V8Ref(RefType::Ref, _) => quote!(&#i),
@@ -26,7 +26,7 @@ pub fn v8_intermediate_to_arg(i: &Ident, arg: &Arg) -> TokenStream {
   quote!(let #i = #arg;)
 }
 
-/// Given an [`Arg`] containing a V8 value, converts this value to itself final form.
+/// Given an [`Arg`] containing a V8 value, converts this value to its final argument form.
 pub fn v8_intermediate_to_global_arg(
   deno_core: &TokenStream,
   isolate: &Ident,
diff --git a/ops/op2/dispatch_slow.rs b/ops/op2/dispatch_slow.rs
index 2d274bc..83367e4 100644
--- a/ops/op2/dispatch_slow.rs
+++ b/ops/op2/dispatch_slow.rs
@@ -1,4 +1,5 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+use super::config::MacroConfig;
 use super::dispatch_shared::v8_intermediate_to_arg;
 use super::dispatch_shared::v8_intermediate_to_global_arg;
 use super::dispatch_shared::v8_to_arg;
@@ -10,7 +11,6 @@ use super::signature::ParsedSignature;
 use super::signature::RefType;
 use super::signature::RetVal;
 use super::signature::Special;
-use super::MacroConfig;
 use super::V8MappingError;
 use crate::op2::generator_state::gs_extract;
 use crate::op2::generator_state::gs_quote;
diff --git a/ops/op2/mod.rs b/ops/op2/mod.rs
index 137d5a7..3494770 100644
--- a/ops/op2/mod.rs
+++ b/ops/op2/mod.rs
@@ -1,9 +1,7 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
-use deno_proc_macro_rules::rules;
 use proc_macro2::Ident;
 use proc_macro2::Span;
 use proc_macro2::TokenStream;
-use proc_macro2::TokenTree;
 use quote::format_ident;
 use quote::quote;
 use quote::ToTokens;
@@ -18,6 +16,7 @@ use syn2::LifetimeParam;
 use syn2::Path;
 use thiserror::Error;
 
+use self::config::MacroConfig;
 use self::dispatch_async::generate_dispatch_async;
 use self::dispatch_fast::generate_dispatch_fast;
 use self::dispatch_slow::generate_dispatch_slow;
@@ -27,6 +26,7 @@ use self::signature::parse_signature;
 use self::signature::Arg;
 use self::signature::SignatureError;
 
+pub mod config;
 pub mod dispatch_async;
 pub mod dispatch_fast;
 pub mod dispatch_shared;
@@ -67,81 +67,6 @@ pub enum V8MappingError {
   NoMapping(&'static str, Arg),
 }
 
-#[derive(Default)]
-pub(crate) struct MacroConfig {
-  /// Use `(core)` for ops that live in `deno_core`.
-  pub core: bool,
-  /// Generate a fastcall method (must be fastcall compatible).
-  pub fast: bool,
-  /// Do not generate a fastcall method (must be fastcall compatible).
-  pub nofast: bool,
-  /// Marks an async function (either `async fn` or `fn -> impl Future`)
-  pub r#async: bool,
-}
-
-impl MacroConfig {
-  pub fn from_token_trees(flags: Vec<TokenTree>) -> Result<Self, Op2Error> {
-    Self::from_flags(flags.into_iter().map(|tt| tt.to_string()))
-  }
-
-  pub fn from_flags(
-    flags: impl IntoIterator<Item = String>,
-  ) -> Result<Self, Op2Error> {
-    let mut config: MacroConfig = Self::default();
-    let flags = flags.into_iter().collect::<Vec<_>>();
-
-    // Ensure that the flags are sorted in alphabetical order for consistency and searchability
-    let mut flags_sorted = flags.clone();
-    flags_sorted.sort();
-    if flags != flags_sorted {
-      return Err(Op2Error::ImproperlySortedAttribute(flags_sorted.join(", ")));
-    }
-
-    for flag in flags {
-      if flag == "core" {
-        config.core = true;
-      } else if flag == "fast" {
-        config.fast = true;
-      } else if flag == "nofast" {
-        config.nofast = true;
-      } else if flag == "async" {
-        config.r#async = true;
-      } else {
-        return Err(Op2Error::InvalidAttribute(flag));
-      }
-    }
-
-    // Test for invalid attribute combinations
-    if config.fast && config.nofast {
-      return Err(Op2Error::InvalidAttributeCombination("fast", "nofast"));
-    }
-    if config.fast && config.r#async {
-      return Err(Op2Error::InvalidAttributeCombination("fast", "async"));
-    }
-    if config.nofast && config.r#async {
-      return Err(Op2Error::InvalidAttributeCombination("nofast", "async"));
-    }
-
-    Ok(config)
-  }
-
-  pub fn from_tokens(tokens: TokenStream) -> Result<Self, Op2Error> {
-    let attr_string = tokens.to_string();
-    let config = std::panic::catch_unwind(|| {
-      rules!(tokens => {
-        () => {
-          Ok(MacroConfig::default())
-        }
-        ($($flags:tt),+) => {
-          Self::from_token_trees(flags)
-        }
-      })
-    })
-    .map_err(|_| Op2Error::PatternMatchFailed("attribute", attr_string))??;
-    Ok(config)
-  }
-}
-
 pub fn op2(
   attr: TokenStream,
   item: TokenStream,
@@ -246,7 +171,7 @@ fn generate_op2(
   let name = func.sig.ident;
 
   let slow_fn = if signature.ret_val.is_async() {
-    generate_dispatch_async(&mut generator_state, &signature)?
+    generate_dispatch_async(&config, &mut generator_state, &signature)?
   } else {
     generate_dispatch_slow(&config, &mut generator_state, &signature)?
   };
@@ -384,18 +309,10 @@ deno_ops_compile_test_runner::prelude!();";
           let tokens = attr.into_token_stream();
           let attr_string = attr.clone().into_token_stream().to_string();
           println!("{}", attr_string);
-          use syn2 as syn;
-          if let Some(new_config) = rules!(tokens => {
-            (#[op2]) => {
-              Some(MacroConfig::default())
-            }
-            (#[op2( $($x:tt),* )]) => {
-              Some(MacroConfig::from_token_trees(x).expect("Failed to parse attribute"))
-            }
-            (#[$_attr:meta]) => {
-              None
-            }
-          }) {
+          if let Some(new_config) =
+            MacroConfig::from_maybe_attribute_tokens(tokens)
+              .expect("Failed to parse attribute")
+          {
             config = Some(new_config);
             false
           } else {


Test patch:
diff --git a/ops/op2/test_cases/async/async_arg_return.out b/ops/op2/test_cases/async/async_arg_return.out
index 3f01cfe..5ba4e56 100644
--- a/ops/op2/test_cases/async/async_arg_return.out
+++ b/ops/op2/test_cases/async/async_arg_return.out
@@ -56,6 +56,7 @@ impl op_async {
         if let Some(result)
             = deno_core::_ops::map_async_op_infallible(
                 opctx,
+                false,
                 promise_id,
                 result,
                 |scope, result| {
diff --git a/ops/op2/test_cases/async/async_arg_return_result.out b/ops/op2/test_cases/async/async_arg_return_result.out
index caf693c..264bf67 100644
--- a/ops/op2/test_cases/async/async_arg_return_result.out
+++ b/ops/op2/test_cases/async/async_arg_return_result.out
@@ -56,6 +56,7 @@ impl op_async {
         if let Some(result)
             = deno_core::_ops::map_async_op_fallible(
                 opctx,
+                false,
                 promise_id,
                 result,
                 |scope, result| {
diff --git a/ops/op2/test_cases/async/async_jsbuffer.out b/ops/op2/test_cases/async/async_jsbuffer.out
index 0ebcd37..28e1638 100644
--- a/ops/op2/test_cases/async/async_jsbuffer.out
+++ b/ops/op2/test_cases/async/async_jsbuffer.out
@@ -63,6 +63,7 @@ impl op_async_v8_buffer {
         if let Some(result)
             = deno_core::_ops::map_async_op_infallible(
                 opctx,
+                false,
                 promise_id,
                 result,
                 |scope, result| {
diff --git a/ops/op2/test_cases/async/async_opstate.out b/ops/op2/test_cases/async/async_opstate.out
index 4513d8b..881af3e 100644
--- a/ops/op2/test_cases/async/async_opstate.out
+++ b/ops/op2/test_cases/async/async_opstate.out
@@ -44,6 +44,7 @@ impl op_async_opstate {
         if let Some(result)
             = deno_core::_ops::map_async_op_fallible(
                 opctx,
+                false,
                 promise_id,
                 result,
                 |scope, result| {
diff --git a/ops/op2/test_cases/async/async_result.out b/ops/op2/test_cases/async/async_result.out
index d0838e3..1308061 100644
--- a/ops/op2/test_cases/async/async_result.out
+++ b/ops/op2/test_cases/async/async_result.out
@@ -40,6 +40,7 @@ impl op_async {
         if let Some(result)
             = deno_core::_ops::map_async_op_fallible(
                 opctx,
+                false,
                 promise_id,
                 result,
                 |scope, result| {
diff --git a/ops/op2/test_cases/async/async_result_impl.out b/ops/op2/test_cases/async/async_result_impl.out
index 43bc209..64e8eac 100644
--- a/ops/op2/test_cases/async/async_result_impl.out
+++ b/ops/op2/test_cases/async/async_result_impl.out
@@ -70,6 +70,7 @@ impl op_async_result_impl {
         if let Some(result)
             = deno_core::_ops::map_async_op_fallible(
                 opctx,
+                false,
                 promise_id,
                 result,
                 |scope, result| {
diff --git a/ops/op2/test_cases/async/async_result_smi.out b/ops/op2/test_cases/async/async_result_smi.out
index fb55b89..b49df43 100644
--- a/ops/op2/test_cases/async/async_result_smi.out
+++ b/ops/op2/test_cases/async/async_result_smi.out
@@ -56,6 +56,7 @@ impl op_async {
         if let Some(result)
             = deno_core::_ops::map_async_op_fallible(
                 opctx,
+                false,
                 promise_id,
                 result,
                 |scope, result| {
diff --git a/ops/op2/test_cases/async/async_v8_global.out b/ops/op2/test_cases/async/async_v8_global.out
index f694015..dbc9667 100644
--- a/ops/op2/test_cases/async/async_v8_global.out
+++ b/ops/op2/test_cases/async/async_v8_global.out
@@ -58,6 +58,7 @@ impl op_async_v8_global {
         if let Some(result)
             = deno_core::_ops::map_async_op_infallible(
                 opctx,
+                false,
                 promise_id,
                 result,
                 |scope, result| { Ok(deno_core::v8::null(scope).into()) },
diff --git a/ops/op2/test_cases/async/async_void.out b/ops/op2/test_cases/async/async_void.out
index 670042e..4a8e332 100644
--- a/ops/op2/test_cases/async/async_void.out
+++ b/ops/op2/test_cases/async/async_void.out
@@ -40,6 +40,7 @@ impl op_async {
         if let Some(result)
             = deno_core::_ops::map_async_op_infallible(
                 opctx,
+                false,
                 promise_id,
                 result,
                 |scope, result| { Ok(deno_core::v8::null(scope).into()) },