perf: spawn jsr futures on executor (#395)

Cargo.toml

@@ -19,16 +19,19 @@ name = "deno_graph"
 all-features = true
 
 [features]
-default = ["fast_check"]
+default = ["fast_check", "tokio_executor"]
 fast_check = ["symbols", "deno_ast/transpiling", "twox-hash"]
 symbols = ["deno_ast/transforms", "deno_ast/visit", "deno_ast/utils"]
+tokio_executor = ["deno_unsync"]
+wasm_executor = []
 
 [dependencies]
 anyhow = "1.0.43"
 async-trait = "0.1.68"
 data-url = "0.3.0"
 deno_ast = { version = "0.34.0", features = ["dep_analysis"] }
 deno_semver = "0.5.4"
+deno_unsync = { version = "0.3.2", optional = true }
 encoding_rs = "0.8.33"
 futures = "0.3.26"
 import_map = "0.18.0"

lib/Cargo.toml

@@ -16,7 +16,7 @@ crate-type = ["cdylib"]
 [dependencies]
 anyhow = "1.0.43"
 console_error_panic_hook = "0.1.7"
-deno_graph = { path = "../" }
+deno_graph = { path = "../", default-features = false, features = ["wasm_executor"] }
 getrandom = { version = "*", features = ["js"] }
 futures = "0.3.17"
 js-sys = "0.3.63"

lib/lib.rs

@@ -271,6 +271,7 @@ pub async fn js_create_graph(
         imports,
         reporter: None,
         workspace_members: &[],
+        executor: Default::default(),
       },
     )
     .await;

src/graph.rs

@@ -24,6 +24,9 @@ use crate::packages::resolve_version;
 use crate::packages::JsrPackageInfo;
 use crate::packages::JsrPackageVersionInfo;
 use crate::packages::PackageSpecifiers;
+use crate::rt::spawn;
+use crate::rt::Executor;
+use crate::rt::JoinHandle;
 use crate::source::*;
 
 use anyhow::anyhow;
@@ -1021,6 +1024,7 @@ pub struct BuildOptions<'a> {
   pub module_parser: Option<&'a dyn ModuleParser>,
   pub reporter: Option<&'a dyn Reporter>,
   pub workspace_members: &'a [WorkspaceMember],
+  pub executor: &'a dyn Executor,
 }
 
 #[derive(Debug, Copy, Clone)]
@@ -1461,6 +1465,7 @@ impl ModuleGraph {
       options.module_analyzer.unwrap_or(&default_module_parser),
       options.reporter,
       options.workspace_members,
+      options.executor,
     )
     .await
   }
@@ -2785,8 +2790,7 @@ struct PendingJsrPackageVersionInfoLoadItem {
   info: Arc<JsrPackageVersionInfo>,
 }
 
-type PendingResult<T> =
-  Shared<LocalBoxFuture<'static, Result<T, Arc<anyhow::Error>>>>;
+type PendingResult<T> = Shared<JoinHandle<Result<T, Arc<anyhow::Error>>>>;
 
 #[derive(Default)]
 struct PendingJsrState {
@@ -2893,6 +2897,7 @@ struct Builder<'a, 'graph> {
   fill_pass_mode: FillPassMode,
   workspace_members: &'a [WorkspaceMember],
   diagnostics: Vec<BuildDiagnostic>,
+  executor: &'a dyn Executor,
 }
 
 impl<'a, 'graph> Builder<'a, 'graph> {
@@ -2910,6 +2915,7 @@ impl<'a, 'graph> Builder<'a, 'graph> {
     module_analyzer: &'a dyn ModuleAnalyzer,
     reporter: Option<&'a dyn Reporter>,
     workspace_members: &'a [WorkspaceMember],
+    executor: &'a dyn Executor,
   ) -> Vec<BuildDiagnostic> {
     let fill_pass_mode = match graph.roots.is_empty() {
       true => FillPassMode::AllowRestart,
@@ -2929,6 +2935,7 @@ impl<'a, 'graph> Builder<'a, 'graph> {
       fill_pass_mode,
       workspace_members,
       diagnostics: Vec::new(),
+      executor,
     };
     builder.fill(roots, imports).await;
     builder.diagnostics
@@ -3864,18 +3871,21 @@ impl<'a, 'graph> Builder<'a, 'graph> {
         maybe_checksum: None,
       },
     );
-    let fut = async move {
-      let data = fut.await.map_err(Arc::new)?;
-      match data {
-        Some(LoadResponse::Module { content, .. }) => {
-          let package_info: JsrPackageInfo =
-            serde_json::from_slice(&content).map_err(|e| Arc::new(e.into()))?;
-          Ok(Some(Arc::new(package_info)))
+    let fut = spawn(
+      self.executor,
+      async move {
+        let data = fut.await.map_err(Arc::new)?;
+        match data {
+          Some(LoadResponse::Module { content, .. }) => {
+            let package_info: JsrPackageInfo = serde_json::from_slice(&content)
+              .map_err(|e| Arc::new(e.into()))?;
+            Ok(Some(Arc::new(package_info)))
+          }
+          _ => Ok(None),
         }
-        _ => Ok(None),
       }
-    }
-    .boxed_local();
+      .boxed_local(),
+    );
     self
       .state
       .jsr
@@ -3916,25 +3926,29 @@ impl<'a, 'graph> Builder<'a, 'graph> {
         maybe_checksum: maybe_expected_checksum.clone(),
       },
     );
-    let fut = async move {
-      let data = fut.await.map_err(Arc::new)?;
-      match data {
-        Some(LoadResponse::Module { content, .. }) => {
-          // if we have the expected checksum, then we can re-use that here
-          let checksum = maybe_expected_checksum
-            .map(|c| c.into_string())
-            .unwrap_or_else(|| LoaderChecksum::gen(&content));
-          let version_info: JsrPackageVersionInfo =
-            serde_json::from_slice(&content).map_err(|e| Arc::new(e.into()))?;
-          Ok(PendingJsrPackageVersionInfoLoadItem {
-            checksum,
-            info: Arc::new(version_info),
-          })
+    let fut = spawn(
+      self.executor,
+      async move {
+        let data = fut.await.map_err(Arc::new)?;
+        match data {
+          Some(LoadResponse::Module { content, .. }) => {
+            // if we have the expected checksum, then we can re-use that here
+            let checksum = maybe_expected_checksum
+              .map(|c| c.into_string())
+              .unwrap_or_else(|| LoaderChecksum::gen(&content));
+            let version_info: JsrPackageVersionInfo =
+              serde_json::from_slice(&content)
+                .map_err(|e| Arc::new(e.into()))?;
+            Ok(PendingJsrPackageVersionInfoLoadItem {
+              checksum,
+              info: Arc::new(version_info),
+            })
+          }
+          _ => Err(Arc::new(anyhow!("Not found: {}", specifier))),
         }
-        _ => Err(Arc::new(anyhow!("Not found: {}", specifier))),
       }
-    }
-    .boxed_local();
+      .boxed_local(),
+    );
     self
       .state
       .jsr

src/lib.rs

@@ -7,6 +7,7 @@ mod analyzer;
 mod ast;
 mod graph;
 mod module_specifier;
+mod rt;
 
 #[cfg(feature = "symbols")]
 pub mod symbols;
@@ -85,6 +86,7 @@ pub use graph::WorkspaceMember;
 pub use module_specifier::resolve_import;
 pub use module_specifier::ModuleSpecifier;
 pub use module_specifier::SpecifierError;
+pub use rt::Executor;
 pub use source::NpmPackageReqResolution;
 
 pub use deno_ast::dep::DependencyKind;

src/rt.rs

@@ -0,0 +1,74 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+use futures::channel::oneshot;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+pub type BoxedFuture = Pin<Box<dyn Future<Output = ()> + 'static>>;
+
+/// An executor for futures.
+///
+/// This trait allows deno_graph to run background tasks on
+/// the async executor.
+pub trait Executor {
+  /// Spawns a future to run on this executor.
+  fn execute(&self, fut: BoxedFuture) -> BoxedFuture;
+}
+
+impl<'a> Default for &'a dyn Executor {
+  fn default() -> &'a dyn Executor {
+    {
+      struct DefaultExecutor;
+
+      impl Executor for DefaultExecutor {
+        fn execute(&self, future: BoxedFuture) -> BoxedFuture {
+          #[cfg(not(feature = "tokio_executor"))]
+          return future;
+
+          #[cfg(feature = "tokio_executor")]
+          Box::pin(async { deno_unsync::spawn(future).await.unwrap() })
+        }
+      }
+
+      &DefaultExecutor
+    }
+  }
+}
+
+pub(crate) struct JoinHandle<T> {
+  rx: oneshot::Receiver<T>,
+  fut: BoxedFuture,
+}
+
+impl<T> Future for JoinHandle<T> {
+  type Output = T;
+
+  fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+    if let Poll::Ready(()) = Pin::new(&mut self.fut).poll(cx) {
+      if let Poll::Ready(Ok(res)) = Pin::new(&mut self.rx).poll(cx) {
+        Poll::Ready(res)
+      } else {
+        panic!("task panic");
+      }
+    } else {
+      Poll::Pending
+    }
+  }
+}
+
+pub(crate) fn spawn<F, T: 'static>(
+  executor: &dyn Executor,
+  f: F,
+) -> JoinHandle<T>
+where
+  F: Future<Output = T> + 'static,
+{
+  let (tx, rx) = oneshot::channel();
+  let fut = executor.execute(Box::pin(async move {
+    tx.send(f.await).ok();
+  }));
+
+  JoinHandle { rx, fut }
+}