perf: spawn jsr futures on executor

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.33.1", 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 }
+}