feat(task): dependencies

cli/schemas/config-file.v1.json

@@ -448,6 +448,13 @@
                   "type": "string",
                   "required": true,
                   "description": "The task to execute"
+                },
+                "dependencies": {
+                  "type": "array",
+                  "items": {
+                    "type": "string"
+                  },
+                  "description": "Tasks that should be executed before this task"
                 }
               }
             }

cli/tools/task.rs

@@ -2,6 +2,7 @@
 
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::num::NonZeroUsize;
 use std::path::Path;
 use std::path::PathBuf;
 use std::rc::Rc;
@@ -15,6 +16,11 @@ 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::parking_lot::Mutex;
 use deno_core::url::Url;
 use deno_path_util::normalize_path;
 use deno_runtime::deno_node::NodeResolver;
@@ -68,15 +74,55 @@ pub async fn execute_script(
   let node_resolver = factory.node_resolver().await?;
   let env_vars = task_runner::real_env_vars();
 
-  let task_runner = TaskRunner {
+  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 completed = Arc::new(Mutex::new(HashSet::with_capacity(8)));
+  let running = Arc::new(Mutex::new(HashSet::with_capacity(
+    no_of_concurrent_tasks.into(),
+  )));
+
+  let mut 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(),
+    completed,
+    running,
+    task_names: vec![],
   };
-  task_runner.run_task(task_name).await
+
+  match task_runner.sort_tasks_topo(task_name) {
+    Ok(sorted) => task_runner.run_tasks_in_parallel(sorted).await,
+    Err(err) => match err {
+      TaskError::NotFound(name) => {
+        if task_flags.is_run {
+          return Err(anyhow!("Task not found: {}", name));
+        }
+
+        log::error!("Task not found: {}", name);
+        if log::log_enabled!(log::Level::Error) {
+          print_available_tasks(
+            &mut std::io::stderr(),
+            &task_runner.cli_options.start_dir,
+            &task_runner.tasks_config,
+          )?;
+        }
+        Ok(1)
+      }
+      TaskError::TaskDepCycle(name) => {
+        // TODO(bartlomieju): this should be improved to show the cycle
+        log::error!("Task cycle detected: {}", name);
+        Ok(1)
+      }
+    },
+  }
 }
 
 struct RunSingleOptions<'a> {
@@ -86,37 +132,126 @@ struct RunSingleOptions<'a> {
   custom_commands: HashMap<String, Rc<dyn ShellCommand>>,
 }
 
+#[derive(Debug)]
+enum TaskError {
+  NotFound(String),
+  TaskDepCycle(String),
+}
+
 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,
+  completed: Arc<Mutex<HashSet<String>>>,
+  running: Arc<Mutex<HashSet<String>>>,
+  task_names: Vec<String>,
 }
 
 impl<'a> TaskRunner<'a> {
-  async fn run_task(
-    &self,
-    task_name: &String,
+  async fn run_tasks_in_parallel(
+    &mut self,
+    task_names: Vec<String>,
   ) -> Result<i32, deno_core::anyhow::Error> {
-    let Some((dir_url, task_or_script)) = self.tasks_config.task(task_name)
-    else {
-      if self.task_flags.is_run {
-        return Err(anyhow!("Task not found: {}", task_name));
+    self.task_names = task_names;
+    let mut queue = futures_unordered::FuturesUnordered::new();
+
+    loop {
+      if !self.has_remaining_tasks() {
+        break;
       }
 
-      log::error!("Task not found: {}", task_name);
-      if log::log_enabled!(log::Level::Error) {
-        print_available_tasks(
-          &mut std::io::stderr(),
-          &self.cli_options.start_dir,
-          &self.tasks_config,
-        )?;
+      while queue.len() < self.concurrency {
+        if let Some(task) = self.get_next_task() {
+          queue.push(task);
+        } else {
+          break;
+        }
       }
-      return Ok(1);
+
+      // If queue is empty at this point, then there are no more tasks in the queue.
+      let Some(result) = queue.next().await else {
+        break;
+      };
+
+      let (exit_code, name) = result?;
+      if exit_code > 0 {
+        return Ok(exit_code);
+      }
+
+      self.running.lock().remove(&name);
+      self.completed.lock().insert(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)
+  }
+
+  fn has_remaining_tasks(&self) -> bool {
+    self.completed.lock().len() < self.task_names.len()
+  }
+
+  fn get_next_task(
+    &'a self,
+  ) -> Option<LocalBoxFuture<'a, Result<(i32, String), AnyError>>> {
+    for name in &self.task_names {
+      if self.completed.lock().contains(name)
+        || self.running.lock().contains(name)
+      {
+        continue;
+      }
+
+      let should_run = if let Ok((_, def)) = self.get_task(name) {
+        match def {
+          TaskOrScript::Task(_, def) => def
+            .dependencies
+            .iter()
+            .all(|dep| self.completed.lock().contains(dep)),
+          TaskOrScript::Script(_, _) => true,
+        }
+      } else {
+        false
+      };
+
+      // TODO(bartlomieju): is this correct? Shouldn't it be `return None;`?
+      if !should_run {
+        continue;
+      }
+
+      self.running.lock().insert(name.clone());
+      let name = name.clone();
+      return Some(
+        async move {
+          self
+            .run_task(&name)
+            .await
+            .map(|exit_code| (exit_code, name.clone()))
+        }
+        .boxed_local(),
+      );
+    }
+    None
+  }
+
+  async fn run_task(
+    &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
@@ -232,6 +367,48 @@ impl<'a> TaskRunner<'a> {
       .exit_code,
     )
   }
+
+  fn sort_tasks_topo(&self, name: &str) -> Result<Vec<String>, TaskError> {
+    let mut marked: Vec<String> = vec![];
+    let mut sorted: Vec<String> = vec![];
+
+    self.sort_visit(name, &mut marked, &mut sorted)?;
+
+    Ok(sorted)
+  }
+
+  fn sort_visit(
+    &self,
+    name: &str,
+    marked: &mut Vec<String>,
+    sorted: &mut Vec<String>,
+  ) -> Result<(), TaskError> {
+    // Already sorted
+    if sorted.contains(&name.to_string()) {
+      return Ok(());
+    }
+
+    // Graph has a cycle
+    if marked.contains(&name.to_string()) {
+      return Err(TaskError::TaskDepCycle(name.to_string()));
+    }
+
+    marked.push(name.to_string());
+
+    let Some(def) = self.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 {
+        self.sort_visit(dep, marked, sorted)?
+      }
+    }
+
+    sorted.push(name.to_string());
+
+    Ok(())
+  }
 }
 
 fn output_task(task_name: &str, script: &str) {
@@ -339,6 +516,14 @@ fn print_available_tasks(
       )?;
     }
     writeln!(writer, "    {}", desc.task.command)?;
+    if !desc.task.dependencies.is_empty() {
+      writeln!(
+        writer,
+        "    {} {}",
+        colors::gray("depends on:"),
+        colors::cyan(desc.task.dependencies.join(", "))
+      )?;
+    }
   }
 
   Ok(())

tests/specs/task/dependencies/__test__.jsonc

@@ -0,0 +1,61 @@
+{
+  "tests": {
+    "basic1": {
+      "cwd": "basic1",
+      "tempDir": true,
+      "args": "task run",
+      "output": "./basic1.out"
+    },
+    "basic2": {
+      "cwd": "basic2",
+      "tempDir": true,
+      "args": "task run",
+      "output": "./basic2.out"
+    },
+    "cross_package": {
+      "cwd": "cross_package/package1",
+      "tempDir": true,
+      "args": "task run",
+      "output": "./cross_package.out",
+      "exitCode": 1
+    },
+    "diamond": {
+      "cwd": "diamond",
+      "tempDir": true,
+      "args": "task a",
+      "output": "./diamond.out"
+    },
+    "diamond_list": {
+      "cwd": "diamond",
+      "tempDir": true,
+      "args": "task",
+      "output": "./diamond_list.out"
+    },
+    "diamond_big": {
+      "cwd": "diamond_big",
+      "tempDir": true,
+      "args": "task a",
+      "output": "./diamond_big.out"
+    },
+    "diamond_big_list": {
+      "cwd": "diamond_big",
+      "tempDir": true,
+      "args": "task",
+      "output": "./diamond_big_list.out"
+    },
+    "cycle": {
+      "cwd": "cycle",
+      "tempDir": true,
+      "output": "./cycle.out",
+      "args": "task a",
+      "exitCode": 1
+    },
+    "cycle_2": {
+      "cwd": "cycle_2",
+      "tempDir": true,
+      "args": "task a",
+      "output": "./cycle_2.out",
+      "exitCode": 1
+    }
+  }
+}

tests/specs/task/dependencies/basic1.out

@@ -0,0 +1,12 @@
+Task build1 deno run ../build1.js
+Task build2 deno run ../build2.js
+[UNORDERED_START]
+Starting build1
+build1 performing more work...
+build1 finished
+Starting build2
+build2 performing more work...
+build2 finished
+[UNORDERED_END]
+Task run deno run ../run.js
+run finished

tests/specs/task/dependencies/basic1/deno.json

@@ -0,0 +1,10 @@
+{
+  "tasks": {
+    "build1": "deno run ../build1.js",
+    "build2": "deno run ../build2.js",
+    "run": {
+      "command": "deno run ../run.js",
+      "dependencies": ["build1", "build2"]
+    }
+  }
+}

tests/specs/task/dependencies/basic2.out

@@ -0,0 +1,10 @@
+Task build1 deno run ../build1.js
+Starting build1
+build1 performing more work...
+build1 finished
+Task build2 deno run ../build2.js
+Starting build2
+build2 performing more work...
+build2 finished
+Task run deno run ../run.js
+run finished