Parallelized deno fmt (#4823)

2 files changed, 113 insertions, 48 deletions

diff --git a/cli/fmt.rs b/cli/fmt.rs
index acdd816cf..03b7903fe 100644
--- a/cli/fmt.rs
+++ b/cli/fmt.rs
@@ -18,6 +18,8 @@ use std::io::Read;
 use std::io::Write;
 use std::path::Path;
 use std::path::PathBuf;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
 
 fn is_supported(path: &Path) -> bool {
   if let Some(ext) = path.extension() {
@@ -32,38 +34,47 @@ fn get_config() -> dprint::configuration::Configuration {
   ConfigurationBuilder::new().deno().build()
 }
 
-fn check_source_files(
+async fn check_source_files(
   config: dprint::configuration::Configuration,
   paths: Vec<PathBuf>,
 ) -> Result<(), ErrBox> {
-  let mut not_formatted_files = vec![];
-  let formatter = dprint::Formatter::new(config);
-
-  for file_path in paths {
-    let file_path_str = file_path.to_string_lossy();
-    let file_contents = fs::read_to_string(&file_path)?;
-    let r = formatter.format_text(&file_path_str, &file_contents);
-    match r {
-      Ok(formatted_text) => {
-        if formatted_text != file_contents {
-          not_formatted_files.push(file_path);
+  let not_formatted_files_count = Arc::new(AtomicUsize::new(0));
+  let formatter = Arc::new(dprint::Formatter::new(config));
+  let output_lock = Arc::new(Mutex::new(0)); // prevent threads outputting at the same time
+
+  run_parallelized(paths, {
+    let not_formatted_files_count = not_formatted_files_count.clone();
+    move |file_path| {
+      let file_path_str = file_path.to_string_lossy();
+      let file_contents = fs::read_to_string(&file_path)?;
+      let r = formatter.format_text(&file_path_str, &file_contents);
+      match r {
+        Ok(formatted_text) => {
+          if formatted_text != file_contents {
+            not_formatted_files_count.fetch_add(1, Ordering::SeqCst);
+          }
+        }
+        Err(e) => {
+          let _ = output_lock.lock().unwrap();
+          eprintln!("Error checking: {}", &file_path_str);
+          eprintln!("   {}", e);
         }
       }
-      Err(e) => {
-        eprintln!("Error checking: {}", &file_path_str);
-        eprintln!("   {}", e);
-      }
+      Ok(())
     }
-  }
+  })
+  .await?;
 
-  if not_formatted_files.is_empty() {
+  let not_formatted_files_count =
+    not_formatted_files_count.load(Ordering::SeqCst);
+  if not_formatted_files_count == 0 {
     Ok(())
   } else {
     Err(
       OpError::other(format!(
         "Found {} not formatted {}",
-        not_formatted_files.len(),
-        files_str(not_formatted_files.len()),
+        not_formatted_files_count,
+        files_str(not_formatted_files_count),
       ))
       .into(),
     )
@@ -78,35 +89,45 @@ fn files_str(len: usize) -> &'static str {
   }
 }
 
-fn format_source_files(
+async fn format_source_files(
   config: dprint::configuration::Configuration,
   paths: Vec<PathBuf>,
 ) -> Result<(), ErrBox> {
-  let mut not_formatted_files = vec![];
-  let formatter = dprint::Formatter::new(config);
-
-  for file_path in paths {
-    let file_path_str = file_path.to_string_lossy();
-    let file_contents = fs::read_to_string(&file_path)?;
-    let r = formatter.format_text(&file_path_str, &file_contents);
-    match r {
-      Ok(formatted_text) => {
-        if formatted_text != file_contents {
-          println!("{}", file_path_str);
-          fs::write(&file_path, formatted_text)?;
-          not_formatted_files.push(file_path);
+  let formatted_files_count = Arc::new(AtomicUsize::new(0));
+  let formatter = Arc::new(dprint::Formatter::new(config));
+  let output_lock = Arc::new(Mutex::new(0)); // prevent threads outputting at the same time
+
+  run_parallelized(paths, {
+    let formatted_files_count = formatted_files_count.clone();
+    move |file_path| {
+      let file_path_str = file_path.to_string_lossy();
+      let file_contents = fs::read_to_string(&file_path)?;
+      let r = formatter.format_text(&file_path_str, &file_contents);
+      match r {
+        Ok(formatted_text) => {
+          if formatted_text != file_contents {
+            fs::write(&file_path, formatted_text)?;
+            formatted_files_count.fetch_add(1, Ordering::SeqCst);
+            let _ = output_lock.lock().unwrap();
+            println!("{}", file_path_str);
+          }
+        }
+        Err(e) => {
+          let _ = output_lock.lock().unwrap();
+          eprintln!("Error formatting: {}", &file_path_str);
+          eprintln!("   {}", e);
         }
       }
-      Err(e) => {
-        eprintln!("Error formatting: {}", &file_path_str);
-        eprintln!("   {}", e);
-      }
+      Ok(())
     }
-  }
+  })
+  .await?;
+
+  let formatted_files_count = formatted_files_count.load(Ordering::SeqCst);
   debug!(
     "Formatted {} {}",
-    not_formatted_files.len(),
-    files_str(not_formatted_files.len()),
+    formatted_files_count,
+    files_str(formatted_files_count),
   );
   Ok(())
 }
@@ -115,7 +136,7 @@ fn format_source_files(
 ///
 /// First argument supports globs, and if it is `None`
 /// then the current directory is recursively walked.
-pub fn format(args: Vec<String>, check: bool) -> Result<(), ErrBox> {
+pub async fn format(args: Vec<String>, check: bool) -> Result<(), ErrBox> {
   if args.len() == 1 && args[0] == "-" {
     return format_stdin(check);
   }
@@ -139,9 +160,9 @@ pub fn format(args: Vec<String>, check: bool) -> Result<(), ErrBox> {
   }
   let config = get_config();
   if check {
-    check_source_files(config, target_files)?;
+    check_source_files(config, target_files).await?;
   } else {
-    format_source_files(config, target_files)?;
+    format_source_files(config, target_files).await?;
   }
   Ok(())
 }
@@ -173,6 +194,50 @@ fn format_stdin(check: bool) -> Result<(), ErrBox> {
   Ok(())
 }
 
+async fn run_parallelized<F>(
+  file_paths: Vec<PathBuf>,
+  f: F,
+) -> Result<(), ErrBox>
+where
+  F: FnOnce(PathBuf) -> Result<(), ErrBox> + Send + 'static + Clone,
+{
+  let handles = file_paths.iter().map(|file_path| {
+    let f = f.clone();
+    let file_path = file_path.clone();
+    tokio::task::spawn_blocking(move || f(file_path))
+  });
+  let join_results = futures::future::join_all(handles).await;
+
+  // find the tasks that panicked and let the user know which files
+  let panic_file_paths = join_results
+    .iter()
+    .enumerate()
+    .filter_map(|(i, join_result)| {
+      join_result
+        .as_ref()
+        .err()
+        .map(|_| file_paths[i].to_string_lossy())
+    })
+    .collect::<Vec<_>>();
+  if !panic_file_paths.is_empty() {
+    panic!("Panic formatting: {}", panic_file_paths.join(", "))
+  }
+
+  // check for any errors and if so return the first one
+  let mut errors = join_results.into_iter().filter_map(|join_result| {
+    join_result
+      .ok()
+      .map(|handle_result| handle_result.err())
+      .flatten()
+  });
+
+  if let Some(e) = errors.next() {
+    Err(e)
+  } else {
+    Ok(())
+  }
+}
+
 #[test]
 fn test_is_supported() {
   assert!(!is_supported(Path::new("tests/subdir/redirects")));
@@ -184,10 +249,10 @@ fn test_is_supported() {
   assert!(is_supported(Path::new("foo.tsx")));
 }
 
-#[test]
-fn check_tests_dir() {
+#[tokio::test]
+async fn check_tests_dir() {
   // Because of cli/tests/error_syntax.js the following should fail but not
   // crash.
-  let r = format(vec!["./tests".to_string()], true);
+  let r = format(vec!["./tests".to_string()], true).await;
   assert!(r.is_err());
 }
diff --git a/cli/lib.rs b/cli/lib.rs
index 178e36c99..2e52ace93 100644
--- a/cli/lib.rs
+++ b/cli/lib.rs
@@ -567,7 +567,7 @@ pub fn main() {
       cache_command(flags, files).boxed_local()
     }
     DenoSubcommand::Fmt { check, files } => {
-      async move { fmt::format(files, check) }.boxed_local()
+      fmt::format(files, check).boxed_local()
     }
     DenoSubcommand::Info { file } => info_command(flags, file).boxed_local(),
     DenoSubcommand::Install {