feat(cli): run test modules in parallel

cli/main.rs

@@ -1,6 +1,6 @@
 // Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
 
-#![deny(warnings)]
+// #![deny(warnings)]
 
 #[macro_use]
 extern crate lazy_static;
@@ -36,6 +36,7 @@ mod program_state;
 mod source_maps;
 mod specifier_handler;
 mod standalone;
+mod test_dispatcher;
 mod text_encoding;
 mod tokio_util;
 mod tools;
@@ -54,9 +55,12 @@ use crate::program_state::exit_unstable;
 use crate::program_state::ProgramState;
 use crate::source_maps::apply_source_map;
 use crate::specifier_handler::FetchHandler;
+use crate::test_dispatcher::TestMessage;
+use crate::test_dispatcher::TestResult;
 use crate::tools::installer::infer_name_from_url;
 use deno_core::error::generic_error;
 use deno_core::error::AnyError;
+use deno_core::futures::future;
 use deno_core::futures::future::FutureExt;
 use deno_core::futures::Future;
 use deno_core::resolve_url_or_path;
@@ -79,6 +83,8 @@ use std::iter::once;
 use std::path::PathBuf;
 use std::pin::Pin;
 use std::rc::Rc;
+use std::sync::mpsc::channel;
+use std::sync::mpsc::Sender;
 use std::sync::Arc;
 use std::sync::Mutex;
 
@@ -218,6 +224,7 @@ pub fn create_main_worker(
     // above
     ops::errors::init(js_runtime);
     ops::runtime_compiler::init(js_runtime);
+    ops::test_dispatcher::init(js_runtime);
   }
   worker.bootstrap(&options);
 
@@ -896,6 +903,57 @@ async fn coverage_command(
   .await
 }
 
+async fn run_test(
+  program_state: Arc<ProgramState>,
+  main_module: ModuleSpecifier,
+  permissions: Permissions,
+  channel: Sender<TestMessage>,
+  filter: Option<String>,
+) -> Result<(), AnyError> {
+  let mut worker =
+    create_main_worker(&program_state, main_module.clone(), permissions);
+
+  {
+    let js_runtime = &mut worker.js_runtime;
+    js_runtime
+      .op_state()
+      .borrow_mut()
+      .put::<Sender<TestMessage>>(channel.clone());
+  }
+
+  let mut maybe_coverage_collector =
+    if let Some(ref coverage_dir) = program_state.coverage_dir {
+      let session = worker.create_inspector_session();
+      let coverage_dir = PathBuf::from(coverage_dir);
+      let mut coverage_collector =
+        tools::coverage::CoverageCollector::new(coverage_dir, session);
+      coverage_collector.start_collecting().await?;
+
+      Some(coverage_collector)
+    } else {
+      None
+    };
+
+  let options = json!({
+    "filter": filter,
+  });
+
+  let execute_result = worker.execute_module(&main_module).await;
+  execute_result?;
+  worker.execute("window.dispatchEvent(new Event('load'))")?;
+  worker.execute(&format!("Deno[Deno.internal].runTests({})", options))?;
+
+  worker.run_event_loop().await?;
+  worker.execute("window.dispatchEvent(new Event('unload'))")?;
+  worker.run_event_loop().await?;
+
+  if let Some(coverage_collector) = maybe_coverage_collector.as_mut() {
+    coverage_collector.stop_collecting().await?;
+  }
+
+  Ok(())
+}
+
 async fn test_command(
   flags: Flags,
   include: Option<Vec<String>>,
@@ -910,7 +968,7 @@ async fn test_command(
   let cwd = std::env::current_dir().expect("No current directory");
   let include = include.unwrap_or_else(|| vec![".".to_string()]);
   let test_modules =
-    tools::test_runner::prepare_test_modules_urls(include, &cwd)?;
+    tools::test_runner::collect_test_module_specifiers(include, &cwd)?;
 
   if test_modules.is_empty() {
     println!("No matching test modules found");
@@ -919,71 +977,113 @@ async fn test_command(
     }
     return Ok(());
   }
-  let main_module = deno_core::resolve_path("$deno$test.ts")?;
-  // Create a dummy source file.
-  let source_file = File {
-    local: main_module.to_file_path().unwrap(),
-    maybe_types: None,
-    media_type: MediaType::TypeScript,
-    source: tools::test_runner::render_test_file(
-      test_modules.clone(),
-      fail_fast,
-      quiet,
-      filter,
-    ),
-    specifier: main_module.clone(),
+
+  let lib = if flags.unstable {
+    module_graph::TypeLib::UnstableDenoWindow
+  } else {
+    module_graph::TypeLib::DenoWindow
   };
-  // Save our fake file into file fetcher cache
-  // to allow module access by TS compiler
-  program_state.file_fetcher.insert_cached(source_file);
+
+  program_state
+    .prepare_module_graph(
+      test_modules.clone(),
+      lib.clone(),
+      permissions.clone(),
+      program_state.maybe_import_map.clone(),
+    )
+    .await?;
 
   if no_run {
-    let lib = if flags.unstable {
-      module_graph::TypeLib::UnstableDenoWindow
-    } else {
-      module_graph::TypeLib::DenoWindow
-    };
-    program_state
-      .prepare_module_load(
-        main_module.clone(),
-        lib,
-        Permissions::allow_all(),
-        false,
-        program_state.maybe_import_map.clone(),
-      )
-      .await?;
     return Ok(());
   }
 
-  let mut worker =
-    create_main_worker(&program_state, main_module.clone(), permissions);
-
   if let Some(ref coverage_dir) = flags.coverage_dir {
     env::set_var("DENO_UNSTABLE_COVERAGE_DIR", coverage_dir);
   }
 
-  let mut maybe_coverage_collector =
-    if let Some(ref coverage_dir) = program_state.coverage_dir {
-      let session = worker.create_inspector_session();
-      let coverage_dir = PathBuf::from(coverage_dir);
-      let mut coverage_collector =
-        tools::coverage::CoverageCollector::new(coverage_dir, session);
-      coverage_collector.start_collecting().await?;
+  let (sender, receiver) = channel::<TestMessage>();
+
+  let join_handles = test_modules.iter().map(|module_specifier| {
+    let program_state = program_state.clone();
+    let module_specifier = module_specifier.clone();
+    let permissions = permissions.clone();
+    let sender = sender.clone();
+    let filter = filter.clone();
+
+    tokio::task::spawn_blocking(move || {
+      let join_handle = std::thread::spawn(move || {
+        let future = run_test(
+          program_state.clone(),
+          module_specifier.clone(),
+          permissions.clone(),
+          sender.clone(),
+          filter.clone(),
+        );
 
-      Some(coverage_collector)
-    } else {
-      None
-    };
+        tokio_util::run_basic(future)
+      });
 
-  let execute_result = worker.execute_module(&main_module).await;
-  execute_result?;
-  worker.execute("window.dispatchEvent(new Event('load'))")?;
-  worker.run_event_loop().await?;
-  worker.execute("window.dispatchEvent(new Event('unload'))")?;
-  worker.run_event_loop().await?;
+      join_handle.join()
+    })
+  });
 
-  if let Some(coverage_collector) = maybe_coverage_collector.as_mut() {
-    coverage_collector.stop_collecting().await?;
+  let join_results = future::join_all(join_handles);
+
+  let handler = {
+    let test_modules = test_modules.clone();
+
+    tokio::task::spawn_blocking(move || {
+      let mut has_error = false;
+
+      for message in receiver.iter() {
+        match message {
+          TestMessage::Result {
+            name,
+            duration,
+            result,
+          } => match result {
+            TestResult::Ok => {
+              println!(
+                "test {} ... {} {}",
+                name,
+                colors::green("ok"),
+                colors::gray(format!("({}ms)", duration))
+              );
+            }
+            TestResult::Ignored => {
+              println!(
+                "test {} ... {} {}",
+                name,
+                colors::yellow("ignored"),
+                colors::gray(format!("({}ms)", duration))
+              );
+            }
+            TestResult::Failed(_) => {
+              println!(
+                "test {} ... {} {}",
+                name,
+                colors::red("FAILED"),
+                colors::gray(format!("({}ms)", duration))
+              );
+
+              has_error = true;
+            }
+          },
+          _ => {}
+        }
+
+        if has_error && fail_fast {
+          break;
+        }
+      }
+
+      has_error
+    })
+  };
+
+  let has_error = handler.await?;
+  if has_error {
+    std::process::exit(1);
   }
 
   Ok(())

cli/ops/mod.rs

@@ -2,6 +2,7 @@
 
 pub mod errors;
 pub mod runtime_compiler;
+pub mod test_dispatcher;
 
 use deno_core::error::AnyError;
 use deno_core::json_op_async;

cli/ops/test_dispatcher.rs

@@ -0,0 +1,40 @@
+use crate::test_dispatcher::TestMessage;
+use deno_core::error::AnyError;
+use deno_core::json_op_async;
+use deno_core::json_op_sync;
+use deno_core::serde_json;
+use deno_core::serde_json::json;
+use deno_core::serde_json::Value;
+use deno_core::BufVec;
+use deno_core::JsRuntime;
+use deno_core::OpState;
+use deno_core::ZeroCopyBuf;
+use serde::Deserialize;
+use std::cell::RefCell;
+use std::rc::Rc;
+use std::sync::mpsc::Sender;
+use std::sync::Arc;
+use std::sync::Mutex;
+
+pub fn init(rt: &mut deno_core::JsRuntime) {
+  super::reg_json_sync(rt, "op_send_test_message", op_send_test_message);
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct SendTestMessageArgs {
+  message: TestMessage,
+}
+
+fn op_send_test_message(
+  state: &mut OpState,
+  args: Value,
+  _zero_copy: &mut [ZeroCopyBuf],
+) -> Result<Value, AnyError> {
+  let args: SendTestMessageArgs = serde_json::from_value(args)?;
+  let sender = state.borrow::<Sender<TestMessage>>().clone();
+
+  sender.send(args.message);
+
+  Ok(json!({}))
+}