Merge branch 'denoland:main' into main

1 files changed, 438 insertions, 193 deletions

diff --git a/cli/tools/task.rs b/cli/tools/task.rs
index 502b09d2c..682dbf814 100644
--- a/cli/tools/task.rs
+++ b/cli/tools/task.rs
@@ -1,14 +1,14 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
-use std::borrow::Cow;
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::num::NonZeroUsize;
 use std::path::Path;
 use std::path::PathBuf;
 use std::rc::Rc;
 use std::sync::Arc;
 
-use deno_config::deno_json::Task;
+use deno_config::workspace::TaskDefinition;
 use deno_config::workspace::TaskOrScript;
 use deno_config::workspace::WorkspaceDirectory;
 use deno_config::workspace::WorkspaceTasksConfig;
@@ -16,8 +16,15 @@ use deno_core::anyhow::anyhow;
 use deno_core::anyhow::bail;
 use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
+use deno_core::futures::future::LocalBoxFuture;
+use deno_core::futures::stream::futures_unordered;
+use deno_core::futures::FutureExt;
+use deno_core::futures::StreamExt;
+use deno_core::url::Url;
 use deno_path_util::normalize_path;
+use deno_runtime::deno_node::NodeResolver;
 use deno_task_shell::ShellCommand;
+use indexmap::IndexMap;
 
 use crate::args::CliOptions;
 use crate::args::Flags;
@@ -48,155 +55,376 @@ pub async fn execute_script(
         v == "1"
       })
       .unwrap_or(false);
-  let tasks_config = start_dir.to_tasks_config()?;
-  let tasks_config = if force_use_pkg_json {
-    tasks_config.with_only_pkg_json()
-  } else {
-    tasks_config
-  };
+  let mut tasks_config = start_dir.to_tasks_config()?;
+  if force_use_pkg_json {
+    tasks_config = tasks_config.with_only_pkg_json()
+  }
 
-  let task_name = match &task_flags.task {
-    Some(task) => task,
-    None => {
-      print_available_tasks(
-        &mut std::io::stdout(),
-        &cli_options.start_dir,
-        &tasks_config,
-      )?;
-      return Ok(0);
-    }
+  let Some(task_name) = &task_flags.task else {
+    print_available_tasks(
+      &mut std::io::stdout(),
+      &cli_options.start_dir,
+      &tasks_config,
+    )?;
+    return Ok(0);
   };
 
   let npm_resolver = factory.npm_resolver().await?;
   let node_resolver = factory.node_resolver().await?;
   let env_vars = task_runner::real_env_vars();
 
-  match tasks_config.task(task_name) {
-    Some((dir_url, task_or_script)) => match task_or_script {
-      TaskOrScript::Task(_tasks, script) => {
-        let cwd = match task_flags.cwd {
-          Some(path) => canonicalize_path(&PathBuf::from(path))
-            .context("failed canonicalizing --cwd")?,
-          None => normalize_path(dir_url.to_file_path().unwrap()),
-        };
-
-        let custom_commands = task_runner::resolve_custom_commands(
-          npm_resolver.as_ref(),
-          node_resolver,
-        )?;
-        run_task(RunTaskOptions {
-          task_name,
-          script,
-          cwd: &cwd,
-          env_vars,
-          custom_commands,
-          npm_resolver: npm_resolver.as_ref(),
-          cli_options,
-        })
-        .await
-      }
-      TaskOrScript::Script(scripts, _script) => {
-        // ensure the npm packages are installed if using a managed resolver
-        if let Some(npm_resolver) = npm_resolver.as_managed() {
-          npm_resolver.ensure_top_level_package_json_install().await?;
+  let no_of_concurrent_tasks = if let Ok(value) = std::env::var("DENO_JOBS") {
+    value.parse::<NonZeroUsize>().ok()
+  } else {
+    std::thread::available_parallelism().ok()
+  }
+  .unwrap_or_else(|| NonZeroUsize::new(2).unwrap());
+
+  let task_runner = TaskRunner {
+    tasks_config,
+    task_flags: &task_flags,
+    npm_resolver: npm_resolver.as_ref(),
+    node_resolver: node_resolver.as_ref(),
+    env_vars,
+    cli_options,
+    concurrency: no_of_concurrent_tasks.into(),
+  };
+
+  task_runner.run_task(task_name).await
+}
+
+struct RunSingleOptions<'a> {
+  task_name: &'a str,
+  script: &'a str,
+  cwd: &'a Path,
+  custom_commands: HashMap<String, Rc<dyn ShellCommand>>,
+}
+
+struct TaskRunner<'a> {
+  tasks_config: WorkspaceTasksConfig,
+  task_flags: &'a TaskFlags,
+  npm_resolver: &'a dyn CliNpmResolver,
+  node_resolver: &'a NodeResolver,
+  env_vars: HashMap<String, String>,
+  cli_options: &'a CliOptions,
+  concurrency: usize,
+}
+
+impl<'a> TaskRunner<'a> {
+  pub async fn run_task(
+    &self,
+    task_name: &str,
+  ) -> Result<i32, deno_core::anyhow::Error> {
+    match sort_tasks_topo(task_name, &self.tasks_config) {
+      Ok(sorted) => self.run_tasks_in_parallel(sorted).await,
+      Err(err) => match err {
+        TaskError::NotFound(name) => {
+          if self.task_flags.is_run {
+            return Err(anyhow!("Task not found: {}", name));
+          }
+
+          log::error!("Task not found: {}", name);
+          if log::log_enabled!(log::Level::Error) {
+            self.print_available_tasks()?;
+          }
+          Ok(1)
+        }
+        TaskError::TaskDepCycle { path } => {
+          log::error!("Task cycle detected: {}", path.join(" -> "));
+          Ok(1)
         }
+      },
+    }
+  }
+
+  pub fn print_available_tasks(&self) -> Result<(), std::io::Error> {
+    print_available_tasks(
+      &mut std::io::stderr(),
+      &self.cli_options.start_dir,
+      &self.tasks_config,
+    )
+  }
 
-        let cwd = match task_flags.cwd {
-          Some(path) => canonicalize_path(&PathBuf::from(path))?,
-          None => normalize_path(dir_url.to_file_path().unwrap()),
-        };
-
-        // At this point we already checked if the task name exists in package.json.
-        // We can therefore check for "pre" and "post" scripts too, since we're only
-        // dealing with package.json here and not deno.json
-        let task_names = vec![
-          format!("pre{}", task_name),
-          task_name.clone(),
-          format!("post{}", task_name),
-        ];
-        let custom_commands = task_runner::resolve_custom_commands(
-          npm_resolver.as_ref(),
-          node_resolver,
-        )?;
-        for task_name in &task_names {
-          if let Some(script) = scripts.get(task_name) {
-            let exit_code = run_task(RunTaskOptions {
-              task_name,
-              script,
-              cwd: &cwd,
-              env_vars: env_vars.clone(),
-              custom_commands: custom_commands.clone(),
-              npm_resolver: npm_resolver.as_ref(),
-              cli_options,
-            })
-            .await?;
-            if exit_code > 0 {
-              return Ok(exit_code);
+  async fn run_tasks_in_parallel(
+    &self,
+    task_names: Vec<String>,
+  ) -> Result<i32, deno_core::anyhow::Error> {
+    struct PendingTasksContext {
+      completed: HashSet<String>,
+      running: HashSet<String>,
+      task_names: Vec<String>,
+    }
+
+    impl PendingTasksContext {
+      fn has_remaining_tasks(&self) -> bool {
+        self.completed.len() < self.task_names.len()
+      }
+
+      fn mark_complete(&mut self, task_name: String) {
+        self.running.remove(&task_name);
+        self.completed.insert(task_name);
+      }
+
+      fn get_next_task<'a>(
+        &mut self,
+        runner: &'a TaskRunner<'a>,
+      ) -> Option<LocalBoxFuture<'a, Result<(i32, String), AnyError>>> {
+        for name in &self.task_names {
+          if self.completed.contains(name) || self.running.contains(name) {
+            continue;
+          }
+
+          let should_run = if let Ok((_, def)) = runner.get_task(name) {
+            match def {
+              TaskOrScript::Task(_, def) => def
+                .dependencies
+                .iter()
+                .all(|dep| self.completed.contains(dep)),
+              TaskOrScript::Script(_, _) => true,
             }
+          } else {
+            false
+          };
+
+          if !should_run {
+            continue;
           }
+
+          self.running.insert(name.clone());
+          let name = name.clone();
+          return Some(
+            async move {
+              runner
+                .run_task_no_dependencies(&name)
+                .await
+                .map(|exit_code| (exit_code, name))
+            }
+            .boxed_local(),
+          );
         }
+        None
+      }
+    }
 
-        Ok(0)
+    let mut context = PendingTasksContext {
+      completed: HashSet::with_capacity(task_names.len()),
+      running: HashSet::with_capacity(self.concurrency),
+      task_names,
+    };
+
+    let mut queue = futures_unordered::FuturesUnordered::new();
+
+    while context.has_remaining_tasks() {
+      while queue.len() < self.concurrency {
+        if let Some(task) = context.get_next_task(self) {
+          queue.push(task);
+        } else {
+          break;
+        }
       }
-    },
-    None => {
-      if task_flags.is_run {
-        return Err(anyhow!("Task not found: {}", task_name));
+
+      // If queue is empty at this point, then there are no more tasks in the queue.
+      let Some(result) = queue.next().await else {
+        debug_assert_eq!(context.task_names.len(), 0);
+        break;
+      };
+
+      let (exit_code, name) = result?;
+      if exit_code > 0 {
+        return Ok(exit_code);
       }
-      log::error!("Task not found: {}", task_name);
-      if log::log_enabled!(log::Level::Error) {
-        print_available_tasks(
-          &mut std::io::stderr(),
-          &cli_options.start_dir,
-          &tasks_config,
-        )?;
+
+      context.mark_complete(name);
+    }
+
+    Ok(0)
+  }
+
+  fn get_task(
+    &self,
+    task_name: &str,
+  ) -> Result<(&Url, TaskOrScript), TaskError> {
+    let Some(result) = self.tasks_config.task(task_name) else {
+      return Err(TaskError::NotFound(task_name.to_string()));
+    };
+
+    Ok(result)
+  }
+
+  async fn run_task_no_dependencies(
+    &self,
+    task_name: &String,
+  ) -> Result<i32, deno_core::anyhow::Error> {
+    let (dir_url, task_or_script) = self.get_task(task_name.as_str()).unwrap();
+
+    match task_or_script {
+      TaskOrScript::Task(_tasks, definition) => {
+        self.run_deno_task(dir_url, task_name, definition).await
+      }
+      TaskOrScript::Script(scripts, _script) => {
+        self.run_npm_script(dir_url, task_name, scripts).await
       }
-      Ok(1)
     }
   }
-}
 
-struct RunTaskOptions<'a> {
-  task_name: &'a str,
-  script: &'a str,
-  cwd: &'a Path,
-  env_vars: HashMap<String, String>,
-  custom_commands: HashMap<String, Rc<dyn ShellCommand>>,
-  npm_resolver: &'a dyn CliNpmResolver,
-  cli_options: &'a CliOptions,
-}
+  async fn run_deno_task(
+    &self,
+    dir_url: &Url,
+    task_name: &String,
+    definition: &TaskDefinition,
+  ) -> Result<i32, deno_core::anyhow::Error> {
+    let cwd = match &self.task_flags.cwd {
+      Some(path) => canonicalize_path(&PathBuf::from(path))
+        .context("failed canonicalizing --cwd")?,
+      None => normalize_path(dir_url.to_file_path().unwrap()),
+    };
 
-async fn run_task(opts: RunTaskOptions<'_>) -> Result<i32, AnyError> {
-  let RunTaskOptions {
-    task_name,
-    script,
-    cwd,
-    env_vars,
-    custom_commands,
-    npm_resolver,
-    cli_options,
-  } = opts;
+    let custom_commands = task_runner::resolve_custom_commands(
+      self.npm_resolver,
+      self.node_resolver,
+    )?;
+    self
+      .run_single(RunSingleOptions {
+        task_name,
+        script: &definition.command,
+        cwd: &cwd,
+        custom_commands,
+      })
+      .await
+  }
 
-  output_task(
-    opts.task_name,
-    &task_runner::get_script_with_args(script, cli_options.argv()),
-  );
+  async fn run_npm_script(
+    &self,
+    dir_url: &Url,
+    task_name: &String,
+    scripts: &IndexMap<String, String>,
+  ) -> Result<i32, deno_core::anyhow::Error> {
+    // ensure the npm packages are installed if using a managed resolver
+    if let Some(npm_resolver) = self.npm_resolver.as_managed() {
+      npm_resolver.ensure_top_level_package_json_install().await?;
+    }
+
+    let cwd = match &self.task_flags.cwd {
+      Some(path) => canonicalize_path(&PathBuf::from(path))?,
+      None => normalize_path(dir_url.to_file_path().unwrap()),
+    };
 
-  Ok(
-    task_runner::run_task(task_runner::RunTaskOptions {
+    // At this point we already checked if the task name exists in package.json.
+    // We can therefore check for "pre" and "post" scripts too, since we're only
+    // dealing with package.json here and not deno.json
+    let task_names = vec![
+      format!("pre{}", task_name),
+      task_name.clone(),
+      format!("post{}", task_name),
+    ];
+    let custom_commands = task_runner::resolve_custom_commands(
+      self.npm_resolver,
+      self.node_resolver,
+    )?;
+    for task_name in &task_names {
+      if let Some(script) = scripts.get(task_name) {
+        let exit_code = self
+          .run_single(RunSingleOptions {
+            task_name,
+            script,
+            cwd: &cwd,
+            custom_commands: custom_commands.clone(),
+          })
+          .await?;
+        if exit_code > 0 {
+          return Ok(exit_code);
+        }
+      }
+    }
+
+    Ok(0)
+  }
+
+  async fn run_single(
+    &self,
+    opts: RunSingleOptions<'_>,
+  ) -> Result<i32, AnyError> {
+    let RunSingleOptions {
       task_name,
       script,
       cwd,
-      env_vars,
       custom_commands,
-      init_cwd: opts.cli_options.initial_cwd(),
-      argv: cli_options.argv(),
-      root_node_modules_dir: npm_resolver.root_node_modules_path(),
-      stdio: None,
-    })
-    .await?
-    .exit_code,
-  )
+    } = opts;
+
+    output_task(
+      opts.task_name,
+      &task_runner::get_script_with_args(script, self.cli_options.argv()),
+    );
+
+    Ok(
+      task_runner::run_task(task_runner::RunTaskOptions {
+        task_name,
+        script,
+        cwd,
+        env_vars: self.env_vars.clone(),
+        custom_commands,
+        init_cwd: self.cli_options.initial_cwd(),
+        argv: self.cli_options.argv(),
+        root_node_modules_dir: self.npm_resolver.root_node_modules_path(),
+        stdio: None,
+      })
+      .await?
+      .exit_code,
+    )
+  }
+}
+
+#[derive(Debug)]
+enum TaskError {
+  NotFound(String),
+  TaskDepCycle { path: Vec<String> },
+}
+
+fn sort_tasks_topo(
+  name: &str,
+  task_config: &WorkspaceTasksConfig,
+) -> Result<Vec<String>, TaskError> {
+  fn sort_visit<'a>(
+    name: &'a str,
+    sorted: &mut Vec<String>,
+    mut path: Vec<&'a str>,
+    tasks_config: &'a WorkspaceTasksConfig,
+  ) -> Result<(), TaskError> {
+    // Already sorted
+    if sorted.iter().any(|sorted_name| sorted_name == name) {
+      return Ok(());
+    }
+
+    // Graph has a cycle
+    if path.contains(&name) {
+      path.push(name);
+      return Err(TaskError::TaskDepCycle {
+        path: path.iter().map(|s| s.to_string()).collect(),
+      });
+    }
+
+    let Some(def) = tasks_config.task(name) else {
+      return Err(TaskError::NotFound(name.to_string()));
+    };
+
+    if let TaskOrScript::Task(_, actual_def) = def.1 {
+      for dep in &actual_def.dependencies {
+        let mut path = path.clone();
+        path.push(name);
+        sort_visit(dep, sorted, path, tasks_config)?
+      }
+    }
+
+    sorted.push(name.to_string());
+
+    Ok(())
+  }
+
+  let mut sorted: Vec<String> = vec![];
+
+  sort_visit(name, &mut sorted, Vec::new(), task_config)?;
+
+  Ok(sorted)
 }
 
 fn output_task(task_name: &str, script: &str) {
@@ -222,80 +450,97 @@ fn print_available_tasks(
       "  {}",
       colors::red("No tasks found in configuration file")
     )?;
-  } else {
-    let mut seen_task_names =
-      HashSet::with_capacity(tasks_config.tasks_count());
-    for maybe_config in [&tasks_config.member, &tasks_config.root] {
-      let Some(config) = maybe_config else {
-        continue;
-      };
-      for (is_root, is_deno, (key, task)) in config
-        .deno_json
-        .as_ref()
-        .map(|config| {
-          let is_root = !is_cwd_root_dir
-            && config.folder_url
-              == *workspace_dir.workspace.root_dir().as_ref();
-          config
-            .tasks
-            .iter()
-            .map(move |(k, t)| (is_root, true, (k, Cow::Borrowed(t))))
-        })
-        .into_iter()
-        .flatten()
-        .chain(
-          config
-            .package_json
-            .as_ref()
-            .map(|config| {
-              let is_root = !is_cwd_root_dir
-                && config.folder_url
-                  == *workspace_dir.workspace.root_dir().as_ref();
-              config.tasks.iter().map(move |(k, v)| {
-                (is_root, false, (k, Cow::Owned(Task::Definition(v.clone()))))
-              })
-            })
-            .into_iter()
-            .flatten(),
-        )
-      {
-        if !seen_task_names.insert(key) {
+    return Ok(());
+  }
+
+  struct AvailableTaskDescription {
+    is_root: bool,
+    is_deno: bool,
+    name: String,
+    task: TaskDefinition,
+  }
+  let mut seen_task_names = HashSet::with_capacity(tasks_config.tasks_count());
+  let mut task_descriptions = Vec::with_capacity(tasks_config.tasks_count());
+
+  for maybe_config in [&tasks_config.member, &tasks_config.root] {
+    let Some(config) = maybe_config else {
+      continue;
+    };
+
+    if let Some(config) = config.deno_json.as_ref() {
+      let is_root = !is_cwd_root_dir
+        && config.folder_url == *workspace_dir.workspace.root_dir().as_ref();
+
+      for (name, definition) in &config.tasks {
+        if !seen_task_names.insert(name) {
           continue; // already seen
         }
-        writeln!(
-          writer,
-          "- {}{}",
-          colors::cyan(key),
-          if is_root {
-            if is_deno {
-              format!(" {}", colors::italic_gray("(workspace)"))
-            } else {
-              format!(" {}", colors::italic_gray("(workspace package.json)"))
-            }
-          } else if is_deno {
-            "".to_string()
-          } else {
-            format!(" {}", colors::italic_gray("(package.json)"))
-          }
-        )?;
-        let definition = match task.as_ref() {
-          Task::Definition(definition) => definition,
-          Task::Commented { definition, .. } => definition,
-        };
-        if let Task::Commented { comments, .. } = task.as_ref() {
-          let slash_slash = colors::italic_gray("//");
-          for comment in comments {
-            writeln!(
-              writer,
-              "    {slash_slash} {}",
-              colors::italic_gray(comment)
-            )?;
-          }
+        task_descriptions.push(AvailableTaskDescription {
+          is_root,
+          is_deno: true,
+          name: name.to_string(),
+          task: definition.clone(),
+        });
+      }
+    }
+
+    if let Some(config) = config.package_json.as_ref() {
+      let is_root = !is_cwd_root_dir
+        && config.folder_url == *workspace_dir.workspace.root_dir().as_ref();
+      for (name, script) in &config.tasks {
+        if !seen_task_names.insert(name) {
+          continue; // already seen
         }
-        writeln!(writer, "    {definition}")?;
+
+        task_descriptions.push(AvailableTaskDescription {
+          is_root,
+          is_deno: false,
+          name: name.to_string(),
+          task: deno_config::deno_json::TaskDefinition {
+            command: script.to_string(),
+            dependencies: vec![],
+            description: None,
+          },
+        });
       }
     }
   }
 
+  for desc in task_descriptions {
+    writeln!(
+      writer,
+      "- {}{}",
+      colors::cyan(desc.name),
+      if desc.is_root {
+        if desc.is_deno {
+          format!(" {}", colors::italic_gray("(workspace)"))
+        } else {
+          format!(" {}", colors::italic_gray("(workspace package.json)"))
+        }
+      } else if desc.is_deno {
+        "".to_string()
+      } else {
+        format!(" {}", colors::italic_gray("(package.json)"))
+      }
+    )?;
+    if let Some(description) = &desc.task.description {
+      let slash_slash = colors::italic_gray("//");
+      writeln!(
+        writer,
+        "    {slash_slash} {}",
+        colors::italic_gray(description)
+      )?;
+    }
+    writeln!(writer, "    {}", desc.task.command)?;
+    if !desc.task.dependencies.is_empty() {
+      writeln!(
+        writer,
+        "    {} {}",
+        colors::gray("depends on:"),
+        colors::cyan(desc.task.dependencies.join(", "))
+      )?;
+    }
+  }
+
   Ok(())
 }