perf(oxfmt): Walk and format at the same time

apps/oxfmt/src/format.rs

@@ -1,4 +1,4 @@
-use std::{env, io::Write, path::PathBuf, time::Instant};
+use std::{env, io::Write, path::PathBuf, sync::mpsc, time::Instant};
 
 use ignore::overrides::OverrideBuilder;
 
@@ -63,25 +63,14 @@ impl FormatRunner {
             })
             .flatten();
 
+        // TODO: Support ignoring files
         let walker = Walk::new(&target_paths, override_builder);
-        let entries = walker.collect_entries();
 
-        let files_to_format = entries
-            .into_iter()
-            // TODO: Support .pretttierignore
-            // TODO: Support default ignores like node_modules, .git, etc.
-            // .filter(|entry| !config_store.should_ignore(Path::new(&entry.path)))
-            .collect::<Vec<_>>();
+        // Get the receiver for streaming entries
+        let rx_entry = walker.stream_entries();
 
-        // It may be empty after filtering ignored files
-        if files_to_format.is_empty() {
-            if misc_options.no_error_on_unmatched_pattern {
-                return CliRunResult::None;
-            }
-
-            print_and_flush_stdout(stdout, "Expected at least one target file\n");
-            return CliRunResult::NoFilesFound;
-        }
+        // Count files for stats
+        let (tx_count, rx_count) = mpsc::channel::<()>();
 
         let (mut diagnostic_service, tx_error) =
             DiagnosticService::new(Box::new(DefaultReporter::default()));
@@ -90,21 +79,44 @@ impl FormatRunner {
             print_and_flush_stdout(stdout, "Checking formatting...\n");
         }
 
-        let target_files_count = files_to_format.len();
         let output_options_clone = output_options.clone();
-
-        // Spawn a thread to run formatting service and wait diagnostics
+        // Spawn a thread to run formatting service with streaming entries
         rayon::spawn(move || {
-            let mut format_service = FormatService::new(cwd, &output_options_clone);
-            format_service.with_entries(files_to_format);
-            format_service.run(&tx_error);
+            let format_service = FormatService::new(cwd, output_options_clone);
+            format_service.run_streaming(rx_entry, &tx_error, tx_count);
         });
-        // NOTE: This is a blocking
+
+        // NOTE: This is blocking - waits for all diagnostics
         let res = diagnostic_service.run(stdout);
 
+        // Count the processed files
+        let target_files_count = rx_count.iter().count();
+        let print_stats = |stdout| {
+            print_and_flush_stdout(
+                stdout,
+                &format!(
+                    "Finished in {}ms on {target_files_count} files using {} threads.\n",
+                    start_time.elapsed().as_millis(),
+                    rayon::current_num_threads()
+                ),
+            );
+        };
+
         // Add a new line between diagnostics and summary
         print_and_flush_stdout(stdout, "\n");
 
+        // Check if no files were found
+        if target_files_count == 0 {
+            if misc_options.no_error_on_unmatched_pattern {
+                print_and_flush_stdout(stdout, "No files found matching the given patterns.\n");
+                print_stats(stdout);
+                return CliRunResult::None;
+            }
+
+            print_and_flush_stdout(stdout, "Expected at least one target file\n");
+            return CliRunResult::NoFilesFound;
+        }
+
         if 0 < res.errors_count() {
             // Each error is already printed in reporter
             print_and_flush_stdout(
@@ -114,17 +126,6 @@ impl FormatRunner {
             return CliRunResult::FormatFailed;
         }
 
-        let print_stats = |stdout| {
-            print_and_flush_stdout(
-                stdout,
-                &format!(
-                    "Finished in {}ms on {target_files_count} files using {} threads.\n",
-                    start_time.elapsed().as_millis(),
-                    rayon::current_num_threads()
-                ),
-            );
-        };
-
         match (&output_options, res.warnings_count()) {
             // `--list-different` outputs nothing here, mismatched paths are already printed in reporter
             (OutputOptions::ListDifferent, 0) => CliRunResult::FormatSucceeded,

apps/oxfmt/src/service.rs

@@ -1,4 +1,4 @@
-use std::{fs, path::Path, time::Instant};
+use std::{fs, path::Path, sync::mpsc, time::Instant};
 
 use cow_utils::CowUtils;
 use rayon::prelude::*;
@@ -13,96 +13,104 @@ use crate::{command::OutputOptions, walk::WalkEntry};
 pub struct FormatService {
     cwd: Box<Path>,
     output_options: OutputOptions,
-    entries: Vec<WalkEntry>,
 }
 
 impl FormatService {
-    pub fn new<T>(cwd: T, output_options: &OutputOptions) -> Self
+    pub fn new<T>(cwd: T, output_options: OutputOptions) -> Self
     where
         T: Into<Box<Path>>,
     {
-        Self { cwd: cwd.into(), output_options: output_options.clone(), entries: Vec::new() }
+        Self { cwd: cwd.into(), output_options }
     }
 
-    pub fn with_entries(&mut self, entries: Vec<WalkEntry>) -> &mut Self {
-        self.entries = entries;
-        self
+    /// Process entries as they are received from the channel
+    #[expect(clippy::needless_pass_by_value)]
+    pub fn run_streaming(
+        &self,
+        rx_entry: mpsc::Receiver<WalkEntry>,
+        tx_error: &DiagnosticSender,
+        // Take ownership to close the channel when done
+        tx_count: mpsc::Sender<()>,
+    ) {
+        rx_entry.into_iter().par_bridge().for_each(|entry| {
+            self.process_entry(&entry, tx_error);
+            // Signal that we processed one file (ignore send errors if receiver dropped)
+            let _ = tx_count.send(());
+        });
     }
 
-    pub fn run(&self, tx_error: &DiagnosticSender) {
-        self.entries.iter().par_bridge().for_each(|entry| {
-            let start_time = Instant::now();
+    /// Process a single entry
+    fn process_entry(&self, entry: &WalkEntry, tx_error: &DiagnosticSender) {
+        let start_time = Instant::now();
 
-            let path = Path::new(&entry.path);
-            let source_type = entry.source_type;
+        let path = Path::new(&entry.path);
+        let source_type = entry.source_type;
 
-            // TODO: read_to_arena_str()?
-            let source_text = fs::read_to_string(path).expect("Failed to read file");
+        // TODO: Use `read_to_arena_str()` like `oxlint`?
+        let source_text = fs::read_to_string(path).expect("Failed to read file");
+        // TODO: Use `AllocatorPool.get()` like `oxlint`?
+        let allocator = Allocator::new();
 
-            // TODO: AllocatorPool.get()?
-            let allocator = Allocator::new();
+        let ret = Parser::new(&allocator, &source_text, source_type)
+            .with_options(ParseOptions {
+                parse_regular_expression: false,
+                // Enable all syntax features
+                allow_v8_intrinsics: true,
+                allow_return_outside_function: true,
+                // `oxc_formatter` expects this to be false
+                preserve_parens: false,
+            })
+            .parse();
+        if !ret.errors.is_empty() {
+            let diagnostics = DiagnosticService::wrap_diagnostics(
+                self.cwd.clone(),
+                path,
+                &source_text,
+                ret.errors,
+            );
+            tx_error.send((path.to_path_buf(), diagnostics)).unwrap();
+            return;
+        }
 
-            let ret = Parser::new(&allocator, &source_text, source_type)
-                .with_options(ParseOptions {
-                    parse_regular_expression: false,
-                    // Enable all syntax features
-                    allow_v8_intrinsics: true,
-                    allow_return_outside_function: true,
-                    // `oxc_formatter` expects this to be false
-                    preserve_parens: false,
-                })
-                .parse();
-            if !ret.errors.is_empty() {
-                let diagnostics = DiagnosticService::wrap_diagnostics(
-                    self.cwd.clone(),
-                    path,
-                    &source_text,
-                    ret.errors,
-                );
-                tx_error.send((path.to_path_buf(), diagnostics)).unwrap();
-                return;
-            }
+        let options = FormatOptions {
+            // semicolons: "always".parse().unwrap(),
+            semicolons: "as-needed".parse().unwrap(),
+            ..FormatOptions::default()
+        };
+        let code = Formatter::new(&allocator, options).build(&ret.program);
 
-            let options = FormatOptions {
-                // semicolons: "always".parse().unwrap(),
-                semicolons: "as-needed".parse().unwrap(),
-                ..FormatOptions::default()
-            };
-            let code = Formatter::new(&allocator, options).build(&ret.program);
+        let elapsed = start_time.elapsed();
+        let is_changed = source_text != code;
 
-            let elapsed = start_time.elapsed();
-            let is_changed = source_text != code;
+        // Write back if needed
+        if matches!(self.output_options, OutputOptions::DefaultWrite) && is_changed {
+            fs::write(path, code)
+                .map_err(|_| format!("Failed to write to '{}'", path.to_string_lossy()))
+                .unwrap();
+        }
 
-            // Write back if needed (default behavior is to write)
-            if matches!(self.output_options, OutputOptions::DefaultWrite) && is_changed {
-                fs::write(path, code)
-                    .map_err(|_| format!("Failed to write to '{}'", path.to_string_lossy()))
-                    .unwrap();
+        // Notify if needed
+        let display_path = path
+            // Show path relative to `cwd` for cleaner output
+            .strip_prefix(&self.cwd)
+            .unwrap_or(path)
+            .to_string_lossy()
+            // Normalize path separators for consistent output across platforms
+            .cow_replace('\\', "/")
+            .to_string();
+        let elapsed = elapsed.as_millis();
+        if let Some(diagnostic) = match (&self.output_options, is_changed) {
+            (OutputOptions::Check, true) => {
+                Some(OxcDiagnostic::warn(format!("{display_path} ({elapsed}ms)")))
             }
-
-            // Notify if needed
-            let display_path = path
-                // Show path relative to `cwd` for cleaner output
-                .strip_prefix(&self.cwd)
-                .unwrap_or(path)
-                .to_string_lossy()
-                // Normalize path separators for consistent output across platforms
-                .cow_replace('\\', "/")
-                .to_string();
-            let elapsed = elapsed.as_millis();
-            if let Some(diagnostic) = match (&self.output_options, is_changed) {
-                (OutputOptions::Check, true) => {
-                    Some(OxcDiagnostic::warn(format!("{display_path} ({elapsed}ms)")))
-                }
-                (OutputOptions::ListDifferent, true) => Some(OxcDiagnostic::warn(display_path)),
-                (OutputOptions::DefaultWrite, _) => Some(OxcDiagnostic::warn(format!(
-                    "{display_path} {elapsed}ms{}",
-                    if is_changed { "" } else { " (unchanged)" }
-                ))),
-                _ => None,
-            } {
-                tx_error.send((path.to_path_buf(), vec![diagnostic.into()])).unwrap();
-            }
-        });
+            (OutputOptions::ListDifferent, true) => Some(OxcDiagnostic::warn(display_path)),
+            (OutputOptions::DefaultWrite, _) => Some(OxcDiagnostic::warn(format!(
+                "{display_path} {elapsed}ms{}",
+                if is_changed { "" } else { " (unchanged)" }
+            ))),
+            _ => None,
+        } {
+            tx_error.send((path.to_path_buf(), vec![diagnostic.into()])).unwrap();
+        }
     }
 }

apps/oxfmt/src/walk.rs

@@ -15,39 +15,34 @@ pub struct WalkEntry {
 }
 
 struct WalkBuilder {
-    sender: mpsc::Sender<Vec<WalkEntry>>,
+    sender: mpsc::Sender<WalkEntry>,
 }
 
 impl<'s> ignore::ParallelVisitorBuilder<'s> for WalkBuilder {
     fn build(&mut self) -> Box<dyn ignore::ParallelVisitor + 's> {
-        Box::new(WalkCollector { entries: vec![], sender: self.sender.clone() })
+        Box::new(WalkVisitor { sender: self.sender.clone() })
     }
 }
 
-struct WalkCollector {
-    entries: Vec<WalkEntry>,
-    sender: mpsc::Sender<Vec<WalkEntry>>,
+struct WalkVisitor {
+    sender: mpsc::Sender<WalkEntry>,
 }
 
-impl Drop for WalkCollector {
-    fn drop(&mut self) {
-        let entries = std::mem::take(&mut self.entries);
-        self.sender.send(entries).unwrap();
-    }
-}
-
-impl ignore::ParallelVisitor for WalkCollector {
+impl ignore::ParallelVisitor for WalkVisitor {
     fn visit(&mut self, entry: Result<ignore::DirEntry, ignore::Error>) -> ignore::WalkState {
         match entry {
             Ok(entry) => {
                 // Skip if we can't get file type or if it's a directory
                 if let Some(file_type) = entry.file_type() {
                     if !file_type.is_dir() {
                         if let Some(source_type) = get_supported_source_type(entry.path()) {
-                            self.entries.push(WalkEntry {
-                                path: entry.path().as_os_str().into(),
-                                source_type,
-                            });
+                            let walk_entry =
+                                WalkEntry { path: entry.path().as_os_str().into(), source_type };
+                            // Send each entry immediately through the channel
+                            // If send fails, the receiver has been dropped, so stop walking
+                            if self.sender.send(walk_entry).is_err() {
+                                return ignore::WalkState::Quit;
+                            }
                         }
                     }
                 }
@@ -57,6 +52,7 @@ impl ignore::ParallelVisitor for WalkCollector {
         }
     }
 }
+
 impl Walk {
     /// Will not canonicalize paths.
     /// # Panics
@@ -86,11 +82,17 @@ impl Walk {
         Self { inner }
     }
 
-    pub fn collect_entries(self) -> Vec<WalkEntry> {
-        let (sender, receiver) = mpsc::channel::<Vec<WalkEntry>>();
-        let mut builder = WalkBuilder { sender };
-        self.inner.visit(&mut builder);
-        drop(builder);
-        receiver.into_iter().flatten().collect()
+    /// Stream entries through a channel as they are discovered
+    pub fn stream_entries(self) -> mpsc::Receiver<WalkEntry> {
+        let (sender, receiver) = mpsc::channel::<WalkEntry>();
+
+        // Spawn the walk operation in a separate thread
+        rayon::spawn(move || {
+            let mut builder = WalkBuilder { sender };
+            self.inner.visit(&mut builder);
+            // Channel will be closed when builder is dropped
+        });
+
+        receiver
     }
 }

apps/oxfmt/tests/snapshots/_--check --no-error-on-unmatched-pattern __non__existent__file.js --check __non__existent__file.js@oxfmt.snap

@@ -5,6 +5,10 @@ source: apps/oxfmt/tests/tester.rs
 arguments: --check --no-error-on-unmatched-pattern __non__existent__file.js
 working directory: 
 ----------
+Checking formatting...
+
+No files found matching the given patterns.
+Finished in <variable>ms on 0 files using 1 threads.
 ----------
 CLI result: None
 ----------
@@ -13,6 +17,8 @@ CLI result: None
 arguments: --check __non__existent__file.js
 working directory: 
 ----------
+Checking formatting...
+
 Expected at least one target file
 ----------
 CLI result: NoFilesFound